diff --git a/crates/buzz-db/src/event.rs b/crates/buzz-db/src/event.rs index 334fb1bfe1..fc61d68a52 100644 --- a/crates/buzz-db/src/event.rs +++ b/crates/buzz-db/src/event.rs @@ -895,7 +895,7 @@ pub async fn soft_delete_by_coordinate( kind as u32, buzz_core::kind::KIND_PERSONA | buzz_core::kind::KIND_MANAGED_AGENT ) && crate::managed_agent::projection_coordinate_is_authoritative_on( - &mut *tx, + &mut tx, community_id, kind as u32, pubkey, @@ -963,7 +963,7 @@ pub async fn soft_delete_event_and_update_thread( .execute(&mut *tx) .await?; if crate::managed_agent::projection_coordinate_is_authoritative_on( - &mut *tx, + &mut tx, community_id, kind as u32, &pubkey, diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index fb593abd36..998336c3d9 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -4879,7 +4879,7 @@ impl Db { kind_i32 as u32, buzz_core::kind::KIND_PERSONA | buzz_core::kind::KIND_MANAGED_AGENT ) && managed_agent::projection_coordinate_is_authoritative_on( - &mut *tx, + &mut tx, community_id, kind_i32 as u32, pubkey_bytes.as_slice(), diff --git a/desktop/src-tauri/src/commands/agent_models.rs b/desktop/src-tauri/src/commands/agent_models.rs index 8d9456f95d..7e3780b194 100644 --- a/desktop/src-tauri/src/commands/agent_models.rs +++ b/desktop/src-tauri/src/commands/agent_models.rs @@ -19,9 +19,17 @@ use crate::{ managed_agents::{ build_managed_agent_summary, current_instance_id, discovery_env_with_baked_floor, find_managed_agent_mut, known_acp_runtime, load_global_agent_config, load_managed_agents, - load_personas, managed_agent_avatar_url, missing_command_message, normalize_agent_args, - resolve_command, save_managed_agents, sync_managed_agent_processes, try_regenerate_nest, - AgentModelInfo, AgentModelsResponse, UpdateManagedAgentRequest, UpdateManagedAgentResponse, + load_personas, managed_agent_avatar_url, + migration::activation::{ + confirm_authoritative_edit, enqueue_authoritative_edit, submit_authoritative_edit, + }, + missing_command_message, normalize_agent_args, resolve_command, + retention::{ + active_retention_scope, get_retained_managed_agent_aggregate, open_retention_db, + retire_managed_agent_aggregate, + }, + save_managed_agents, sync_managed_agent_processes, try_regenerate_nest, AgentModelInfo, + AgentModelsResponse, UpdateManagedAgentRequest, UpdateManagedAgentResponse, DEFAULT_ACP_COMMAND, }, relay::{relay_ws_url_with_override, sync_managed_agent_profile}, @@ -856,9 +864,7 @@ pub async fn update_managed_agent( let authoritative_edit = record.relay_authority.is_relay_authoritative(); if authoritative_edit { - crate::managed_agents::migration::activation::enqueue_authoritative_edit( - &app, &state, record, - )?; + enqueue_authoritative_edit(&app, &state, record)?; } save_managed_agents(&app, &records)?; @@ -916,57 +922,51 @@ pub async fn update_managed_agent( try_regenerate_nest(&app); if authoritative_edit { - let scope = crate::managed_agents::retention::active_retention_scope(&app, &state)?; + let scope = active_retention_scope(&app, &state)?; let relay_api = crate::relay::relay_http_base_url(&scope.relay_url); - let verified = - match crate::managed_agents::migration::activation::submit_authoritative_edit( - &state.http_client, - &relay_api, - &scope.owner_keys, - &scope.db_path, - &summary.pubkey, - ) - .await - { - Ok(verified) => verified, - Err(sync_error) => { - if sync_error.starts_with("conflict:") { - let pending = crate::managed_agents::retention::get_retained_managed_agent_aggregate( - &crate::managed_agents::retention::open_retention_db(&scope.db_path)?, - &scope.owner_keys.public_key().to_hex(), - &summary.pubkey, - )? - .ok_or_else(|| "conflicted edit lost its retained attempt".to_string())?; - let conn = crate::managed_agents::retention::open_retention_db( - &scope.db_path, - )?; - if !crate::managed_agents::retention::retire_managed_agent_aggregate( - &conn, - &pending.owner_pubkey, - &pending.agent_pubkey, - pending.generation, - &pending.private_event_id, - )? { - return Err("conflicted edit could not retire its retained attempt".to_string()); - } - let rollback = rollback.as_ref().ok_or_else(|| { - "missing local rollback state after authoritative conflict".to_string() - })?; - rollback_failed_agent_update( - &app, - &state, - &summary.pubkey, - rollback.clone(), - )?; - return Err(format!( + let verified = match submit_authoritative_edit( + &state.http_client, + &relay_api, + &scope.owner_keys, + &scope.db_path, + &summary.pubkey, + ) + .await + { + Ok(verified) => verified, + Err(sync_error) => { + if sync_error.starts_with("conflict:") { + let pending = get_retained_managed_agent_aggregate( + &open_retention_db(&scope.db_path)?, + &scope.owner_keys.public_key().to_hex(), + &summary.pubkey, + )? + .ok_or_else(|| "conflicted edit lost its retained attempt".to_string())?; + let conn = open_retention_db(&scope.db_path)?; + if !retire_managed_agent_aggregate( + &conn, + &pending.owner_pubkey, + &pending.agent_pubkey, + pending.generation, + &pending.private_event_id, + )? { + return Err( + "conflicted edit could not retire its retained attempt".to_string() + ); + } + let rollback = rollback.as_ref().ok_or_else(|| { + "missing local rollback state after authoritative conflict".to_string() + })?; + rollback_failed_agent_update(&app, &state, &summary.pubkey, rollback.clone())?; + return Err(format!( "Agent edit conflicted with a newer relay head. No local changes were kept: {sync_error}" )); - } - return Err(format!( + } + return Err(format!( "Agent edit is retained for relay retry; local changes remain pending confirmation: {sync_error}" )); - } - }; + } + }; { let _guard = state .managed_agents_store_lock @@ -982,10 +982,7 @@ pub async fn update_managed_agent( ); save_managed_agents(&app, &records)?; } - crate::managed_agents::migration::activation::confirm_authoritative_edit( - &scope.db_path, - &verified, - )?; + confirm_authoritative_edit(&scope.db_path, &verified)?; } // Phase 2: relay profile sync (async, outside lock). A rename is committed @@ -1018,87 +1015,8 @@ pub async fn update_managed_agent( }) } -// ── Model normalization ─────────────────────────────────────────────────────── - -/// Normalize raw `buzz-acp models --json` output into a typed DTO for the frontend. -/// -/// Merges models from both ACP paths (stable configOptions + unstable SessionModelState), -/// deduplicates by ID (stable takes precedence), and returns a unified list. -pub(super) fn normalize_agent_models( - raw: &serde_json::Value, - persisted_model: Option, -) -> AgentModelsResponse { - let agent_name = raw["agent"]["name"] - .as_str() - .unwrap_or("unknown") - .to_string(); - let agent_version = raw["agent"]["version"] - .as_str() - .unwrap_or("unknown") - .to_string(); - - let mut models: Vec = Vec::new(); - let mut seen_ids: HashSet = HashSet::new(); - - // 1. Stable configOptions (preferred). Only entries with category "model" - // are model options — the CLI pre-filters, but we're defensive here. - if let Some(config_options) = raw["stable"]["configOptions"].as_array() { - for opt in config_options { - if opt.get("category").and_then(|c| c.as_str()) != Some("model") { - continue; - } - if let Some(options) = opt.get("options").and_then(|v| v.as_array()) { - for o in options { - if let Some(value) = o.get("value").and_then(|v| v.as_str()) { - if seen_ids.insert(value.to_string()) { - models.push(AgentModelInfo { - id: value.to_string(), - name: o - .get("displayName") - .and_then(|v| v.as_str()) - .map(str::to_string), - description: None, - }); - } - } - } - } - } - } - - // 2. Unstable availableModels (fallback — skip duplicates from stable). - let mut agent_default_model: Option = None; - if let Some(unstable) = raw.get("unstable") { - agent_default_model = unstable["currentModelId"].as_str().map(str::to_string); - if let Some(available) = unstable["availableModels"].as_array() { - for m in available { - if let Some(id) = m.get("modelId").and_then(|v| v.as_str()) { - if seen_ids.insert(id.to_string()) { - models.push(AgentModelInfo { - id: id.to_string(), - name: m.get("name").and_then(|v| v.as_str()).map(str::to_string), - description: m - .get("description") - .and_then(|v| v.as_str()) - .map(str::to_string), - }); - } - } - } - } - } - - let supports_switching = !models.is_empty(); - - AgentModelsResponse { - agent_name, - agent_version, - models, - agent_default_model, - selected_model: persisted_model, - supports_switching, - } -} +mod normalization; +pub(crate) use normalization::normalize_agent_models; #[cfg(test)] #[path = "agent_models_tests.rs"] diff --git a/desktop/src-tauri/src/commands/agent_models/normalization.rs b/desktop/src-tauri/src/commands/agent_models/normalization.rs new file mode 100644 index 0000000000..c77b54c7b7 --- /dev/null +++ b/desktop/src-tauri/src/commands/agent_models/normalization.rs @@ -0,0 +1,85 @@ +use std::collections::HashSet; + +use crate::managed_agents::{AgentModelInfo, AgentModelsResponse}; + +// ── Model normalization ─────────────────────────────────────────────────────── + +/// Normalize raw `buzz-acp models --json` output into a typed DTO for the frontend. +/// +/// Merges models from both ACP paths (stable configOptions + unstable SessionModelState), +/// deduplicates by ID (stable takes precedence), and returns a unified list. +pub(crate) fn normalize_agent_models( + raw: &serde_json::Value, + persisted_model: Option, +) -> AgentModelsResponse { + let agent_name = raw["agent"]["name"] + .as_str() + .unwrap_or("unknown") + .to_string(); + let agent_version = raw["agent"]["version"] + .as_str() + .unwrap_or("unknown") + .to_string(); + + let mut models: Vec = Vec::new(); + let mut seen_ids: HashSet = HashSet::new(); + + // 1. Stable configOptions (preferred). Only entries with category "model" + // are model options — the CLI pre-filters, but we're defensive here. + if let Some(config_options) = raw["stable"]["configOptions"].as_array() { + for opt in config_options { + if opt.get("category").and_then(|c| c.as_str()) != Some("model") { + continue; + } + if let Some(options) = opt.get("options").and_then(|v| v.as_array()) { + for o in options { + if let Some(value) = o.get("value").and_then(|v| v.as_str()) { + if seen_ids.insert(value.to_string()) { + models.push(AgentModelInfo { + id: value.to_string(), + name: o + .get("displayName") + .and_then(|v| v.as_str()) + .map(str::to_string), + description: None, + }); + } + } + } + } + } + } + + // 2. Unstable availableModels (fallback — skip duplicates from stable). + let mut agent_default_model: Option = None; + if let Some(unstable) = raw.get("unstable") { + agent_default_model = unstable["currentModelId"].as_str().map(str::to_string); + if let Some(available) = unstable["availableModels"].as_array() { + for m in available { + if let Some(id) = m.get("modelId").and_then(|v| v.as_str()) { + if seen_ids.insert(id.to_string()) { + models.push(AgentModelInfo { + id: id.to_string(), + name: m.get("name").and_then(|v| v.as_str()).map(str::to_string), + description: m + .get("description") + .and_then(|v| v.as_str()) + .map(str::to_string), + }); + } + } + } + } + } + + let supports_switching = !models.is_empty(); + + AgentModelsResponse { + agent_name, + agent_version, + models, + agent_default_model, + selected_model: persisted_model, + supports_switching, + } +} diff --git a/desktop/src-tauri/src/commands/agents.rs b/desktop/src-tauri/src/commands/agents.rs index 9fa25781d2..a38dddd877 100644 --- a/desktop/src-tauri/src/commands/agents.rs +++ b/desktop/src-tauri/src/commands/agents.rs @@ -65,279 +65,10 @@ pub(super) fn retain_managed_agent_pending( } } -/// Decision for a delete retry that finds (or does not find) a pending -/// tombstone already retained at the agent's coordinate. Keeps the immutable -/// tombstone from ever being rebuilt at a fresh event id/timestamp. -#[derive(Debug, PartialEq, Eq)] -enum TombstoneRetry { - /// The exact pending tombstone for THIS deletion already exists; resume it - /// (return `DeferErase`) without touching retention. - Resume, - /// No pending tombstone exists and the record is still authoritative — mint - /// the tombstone now (the normal first-delete path, and the cascade - /// enqueue-before-flip path). - Build, - /// Refuse to proceed: either a mid-deletion (`Deleting`) record has no - /// pending tombstone to resume (inconsistent — its tombstone should exist), - /// or a pending row exists but does not match this agent's verified head - /// (foreign/drifted). Rebuilding would violate immutable retention. - FailClosed, -} - -/// Pure retry classifier for [`tombstone_managed_agent_pending`]. -/// -/// * `pending_row_present` — a `state="deleted"`, `pending_sync` row exists at -/// the coordinate. -/// * `matches_this_deletion` — that row's generation is `evidence.gen + 1` and -/// its decrypted predecessor is the record's verified head. -/// * `is_deleting` — the record's authority is `Deleting` (vs -/// `RelayAuthoritative`). -fn classify_tombstone_retry( - pending_row_present: bool, - matches_this_deletion: bool, - is_deleting: bool, -) -> TombstoneRetry { - match (pending_row_present, matches_this_deletion, is_deleting) { - // Exact pending tombstone for this deletion: idempotent resume. - (true, true, _) => TombstoneRetry::Resume, - // A row exists but is foreign/drifted: never rebuild over it. - (true, false, _) => TombstoneRetry::FailClosed, - // No row, still authoritative: first delete (or cascade pre-flip) builds. - (false, _, false) => TombstoneRetry::Build, - // No row, but already Deleting: inconsistent — its tombstone should exist. - (false, _, true) => TombstoneRetry::FailClosed, - } -} - -/// What the caller must do to the local record after -/// [`tombstone_managed_agent_pending`] has durably enqueued the tombstone. -/// -/// The variant enforces enqueue-before-erase: a `RelayAuthoritative` agent's -/// authoritative tombstone must be verified at the relay before the local -/// record/key are destroyed, so its record stays on disk in `Deleting` state -/// and erase is deferred to the boot deletion flush. A `LegacyOnly` agent has -/// no async relay confirmation, so once its kind:5 tombstone is durably -/// enqueued the caller erases in the same lock. -pub(super) enum TombstoneDisposition { - /// Legacy kind:5 path: the tombstone is durably enqueued; erase the record - /// and key now, in the same lock. - EraseNow, - /// Relay-canonical path: the authoritative deleted aggregate is durably - /// retained (`pending_sync = 1`). The caller MUST NOT erase yet — flip the - /// record to [`RelayAuthority::Deleting`] with this evidence and keep it on - /// disk. The deletion flush erases only after verified relay confirmation. - DeferErase { - evidence: crate::managed_agents::RelayAuthorityEvidence, - }, -} - -/// Durably enqueue a managed agent's authoritative tombstone, returning what -/// the caller must do to the local record — WITHOUT erasing anything itself. -/// -/// This is the crash-safe enqueue-before-erase seam. It runs inside the -/// `managed_agents_store_lock`-held delete body and NEVER across an `.await`. -/// -/// * **RelayAuthoritative**: builds the next-generation deleted kind:30179 -/// aggregate from the record's verified head evidence and retains it with -/// `pending_sync = 1`. This durable row is the retry input; it MUST land -/// before the record/key are destroyed, so a crash between enqueue and the -/// (deferred, verified) erase retries cleanly and the evidence needed to -/// re-sign the tombstone survives. Returns [`TombstoneDisposition::DeferErase`] -/// — the caller keeps the record on disk in [`RelayAuthority::Deleting`]. -/// A failure here is FATAL and propagates: the record is left fully intact -/// ([`RelayAuthority::RelayAuthoritative`]) and the delete is retryable, never -/// erased against a missing durable tombstone. -/// * **LegacyOnly**: retains the kind:5 NIP-09 tombstone at its own coordinate. -/// Returns [`TombstoneDisposition::EraseNow`]. A retention hiccup here is -/// logged and swallowed (legacy has no live authoritative aggregate to leak -/// and no async confirmation), so a disk-authoritative delete is never blocked. -/// -/// Mirrors `commands::personas::tombstone_persona_pending`: the agent row at -/// `(30177, owner, agent_pubkey)` is purged first so an unpublished edit can -/// never resurrect it after the tombstone publishes. -pub(super) fn tombstone_managed_agent_pending( - app: &AppHandle, - state: &AppState, - agent_pubkey: &str, - relay_authority: &crate::managed_agents::RelayAuthority, -) -> Result { - use crate::managed_agents::{ - agent_events::build_agent_delete, - migration::build_tombstone_event, - retention::{ - delete_retained_event, open_retention_db, retain_event, retain_managed_agent_aggregate, - tombstone_retention_d_tag, RetainedEvent, RetainedManagedAgentAggregate, - }, - }; - use buzz_core_pkg::kind::KIND_MANAGED_AGENT; - use nostr::JsonUtil; - - const KIND_DELETE: u32 = 5; - - // Relay-canonical agents durably enqueue the authoritative tombstone before - // ANY local destruction. A failure propagates so the caller never erases - // against a missing retry — the record stays RelayAuthoritative and the - // delete is retryable. - if let Some(evidence) = relay_authority.evidence() { - let scope = crate::managed_agents::retention::active_retention_scope(app, state)?; - let owner_pubkey = scope.owner_keys.public_key().to_hex(); - - // Idempotent retry recognition. A repeat delete on this coordinate may - // find its immutable tombstone ALREADY enqueued — either because the - // record is now `Deleting` (its authority flip/save landed) or because it - // is still `RelayAuthoritative` (a cascade crash after enqueue N but - // before the batch authority save, per the self-heal contract). In BOTH - // cases rebuilding would mint a new event id / timestamp at the same - // coordinate and hit immutable-retention drift. So recognize the exact - // pending tombstone bound to THIS verified head and return DeferErase - // without touching retention; the deletion flush self-heals the authority - // state. If a pending tombstone exists but does NOT match its predecessor, - // fail closed — never rebuild over a foreign row. - { - let conn = open_retention_db(&scope.db_path)?; - let pending_deleted = - crate::managed_agents::retention::get_retained_managed_agent_aggregate( - &conn, - &owner_pubkey, - agent_pubkey, - )? - .filter(|retained| retained.state == "deleted" && retained.pending_sync); - - // Decrypt the pending row (if any) and reduce it to a single fact: - // does it name THIS exact deletion (generation = evidence.gen+1 and - // decrypted predecessor = the record's verified head)? Decryption - // stays here; the branch decision is a pure classifier below. - let matches_this_deletion = match &pending_deleted { - Some(retained) => { - let generation_ok = evidence - .generation - .checked_add(1) - .is_some_and(|next| retained.generation == next); - generation_ok && { - let parsed: serde_json::Value = - serde_json::from_str(&retained.request_json).map_err(|e| { - format!("retained deleted request json invalid: {e}") - })?; - let private_event: nostr::Event = serde_json::from_value( - parsed.get("private_event").cloned().ok_or_else(|| { - "retained deleted request missing private_event".to_string() - })?, - ) - .map_err(|e| format!("retained deleted private_event invalid: {e}"))?; - let (_, payload) = - buzz_core_pkg::private_managed_agent::validate_and_decrypt( - &private_event, - &scope.owner_keys, - ) - .map_err(|e| format!("retained deletion head does not decrypt: {e}"))?; - payload.state == buzz_core_pkg::private_managed_agent::State::Deleted - && payload.owner_pubkey == owner_pubkey - && payload.agent_pubkey == agent_pubkey - && payload.previous_event_id.as_deref() - == Some(&evidence.private_event_id) - } - } - None => false, - }; - - match classify_tombstone_retry( - pending_deleted.is_some(), - matches_this_deletion, - relay_authority.is_deleting(), - ) { - TombstoneRetry::Resume => { - return Ok(TombstoneDisposition::DeferErase { - evidence: evidence.clone(), - }) - } - TombstoneRetry::FailClosed => { - return Err( - "delete retry cannot rebuild an immutable tombstone: a pending row is \ - missing or does not match this agent's verified head (fail closed)" - .to_string(), - ) - } - TombstoneRetry::Build => {} - } - } - - let mut conn = open_retention_db(&scope.db_path)?; - // Build path: no pending tombstone yet exists for this coordinate. - // Purge the agent's 30177 row first so an unpublished edit can never - // resurrect it after the tombstone publishes. - delete_retained_event(&conn, KIND_MANAGED_AGENT, &owner_pubkey, agent_pubkey)?; - - let timestamp = crate::util::now_iso(); - let event = build_tombstone_event( - &scope.owner_keys, - agent_pubkey, - evidence.generation, - &evidence.private_event_id, - ×tamp, - nostr::Timestamp::now().as_secs(), - ) - .map_err(|error| format!("failed to build managed-agent aggregate tombstone: {error:?}"))?; - let generation = evidence - .generation - .checked_add(1) - .ok_or_else(|| "managed-agent aggregate generation overflow".to_string())?; - let request_json = serde_json::to_string(&serde_json::json!({ - "private_event": event, - "definition_event": null, - "instance_event": null, - "expected_definition_revision": null - })) - .map_err(|error| format!("failed to serialize managed-agent tombstone: {error}"))?; - retain_managed_agent_aggregate( - &mut conn, - &RetainedManagedAgentAggregate { - owner_pubkey, - agent_pubkey: agent_pubkey.to_string(), - generation, - private_event_id: event.id.to_hex(), - state: "deleted".to_string(), - request_json, - pending_sync: true, - last_error: None, - local_authority_applied: false, - }, - )?; - return Ok(TombstoneDisposition::DeferErase { - evidence: evidence.clone(), - }); - } - - // LegacyOnly: enqueue the kind:5 tombstone before erase, but best-effort — - // a retention hiccup never blocks the disk-authoritative delete. - let result = (|| -> Result<(), String> { - let scope = crate::managed_agents::retention::active_retention_scope(app, state)?; - let owner_pubkey = scope.owner_keys.public_key().to_hex(); - let conn = open_retention_db(&scope.db_path)?; - delete_retained_event(&conn, KIND_MANAGED_AGENT, &owner_pubkey, agent_pubkey)?; - - let event = build_agent_delete(agent_pubkey, &owner_pubkey)? - .sign_with_keys(&scope.owner_keys) - .map_err(|e| format!("failed to sign managed-agent tombstone: {e}"))?; - retain_event( - &conn, - &RetainedEvent { - kind: KIND_DELETE, - pubkey: owner_pubkey, - // Key by the target coordinate so cross-kind d-tag tombstones - // occupy distinct rows (F2c). - d_tag: tombstone_retention_d_tag(KIND_MANAGED_AGENT, agent_pubkey), - content: event.content.to_string(), - created_at: event.created_at.as_secs() as i64, - raw_event: event.as_json(), - pending_sync: true, - }, - ) - })(); - if let Err(e) = result { - eprintln!("buzz-desktop: agent-tombstone: {e}"); - } - Ok(TombstoneDisposition::EraseNow) -} +mod tombstone; +#[cfg(test)] +use tombstone::{classify_tombstone_retry, TombstoneRetry}; +pub(super) use tombstone::{tombstone_managed_agent_pending, TombstoneDisposition}; /// Build and sign the NIP-IA `kind:9035` archive request enqueued when an /// agent is deleted. Pure given the keys — unit-testable without an diff --git a/desktop/src-tauri/src/commands/agents/tombstone.rs b/desktop/src-tauri/src/commands/agents/tombstone.rs new file mode 100644 index 0000000000..3f8fd0bc6b --- /dev/null +++ b/desktop/src-tauri/src/commands/agents/tombstone.rs @@ -0,0 +1,275 @@ +use super::*; + +/// Decision for a delete retry that finds (or does not find) a pending +/// tombstone already retained at the agent's coordinate. Keeps the immutable +/// tombstone from ever being rebuilt at a fresh event id/timestamp. +#[derive(Debug, PartialEq, Eq)] +pub(crate) enum TombstoneRetry { + /// The exact pending tombstone for THIS deletion already exists; resume it + /// (return `DeferErase`) without touching retention. + Resume, + /// No pending tombstone exists and the record is still authoritative — mint + /// the tombstone now (the normal first-delete path, and the cascade + /// enqueue-before-flip path). + Build, + /// Refuse to proceed: either a mid-deletion (`Deleting`) record has no + /// pending tombstone to resume (inconsistent — its tombstone should exist), + /// or a pending row exists but does not match this agent's verified head + /// (foreign/drifted). Rebuilding would violate immutable retention. + FailClosed, +} + +/// Pure retry classifier for [`tombstone_managed_agent_pending`]. +/// +/// * `pending_row_present` — a `state="deleted"`, `pending_sync` row exists at +/// the coordinate. +/// * `matches_this_deletion` — that row's generation is `evidence.gen + 1` and +/// its decrypted predecessor is the record's verified head. +/// * `is_deleting` — the record's authority is `Deleting` (vs +/// `RelayAuthoritative`). +pub(crate) fn classify_tombstone_retry( + pending_row_present: bool, + matches_this_deletion: bool, + is_deleting: bool, +) -> TombstoneRetry { + match (pending_row_present, matches_this_deletion, is_deleting) { + // Exact pending tombstone for this deletion: idempotent resume. + (true, true, _) => TombstoneRetry::Resume, + // A row exists but is foreign/drifted: never rebuild over it. + (true, false, _) => TombstoneRetry::FailClosed, + // No row, still authoritative: first delete (or cascade pre-flip) builds. + (false, _, false) => TombstoneRetry::Build, + // No row, but already Deleting: inconsistent — its tombstone should exist. + (false, _, true) => TombstoneRetry::FailClosed, + } +} + +/// What the caller must do to the local record after +/// [`tombstone_managed_agent_pending`] has durably enqueued the tombstone. +/// +/// The variant enforces enqueue-before-erase: a `RelayAuthoritative` agent's +/// authoritative tombstone must be verified at the relay before the local +/// record/key are destroyed, so its record stays on disk in `Deleting` state +/// and erase is deferred to the boot deletion flush. A `LegacyOnly` agent has +/// no async relay confirmation, so once its kind:5 tombstone is durably +/// enqueued the caller erases in the same lock. +pub(crate) enum TombstoneDisposition { + /// Legacy kind:5 path: the tombstone is durably enqueued; erase the record + /// and key now, in the same lock. + EraseNow, + /// Relay-canonical path: the authoritative deleted aggregate is durably + /// retained (`pending_sync = 1`). The caller MUST NOT erase yet — flip the + /// record to [`RelayAuthority::Deleting`] with this evidence and keep it on + /// disk. The deletion flush erases only after verified relay confirmation. + DeferErase { + evidence: crate::managed_agents::RelayAuthorityEvidence, + }, +} + +/// Durably enqueue a managed agent's authoritative tombstone, returning what +/// the caller must do to the local record — WITHOUT erasing anything itself. +/// +/// This is the crash-safe enqueue-before-erase seam. It runs inside the +/// `managed_agents_store_lock`-held delete body and NEVER across an `.await`. +/// +/// * **RelayAuthoritative**: builds the next-generation deleted kind:30179 +/// aggregate from the record's verified head evidence and retains it with +/// `pending_sync = 1`. This durable row is the retry input; it MUST land +/// before the record/key are destroyed, so a crash between enqueue and the +/// (deferred, verified) erase retries cleanly and the evidence needed to +/// re-sign the tombstone survives. Returns [`TombstoneDisposition::DeferErase`] +/// — the caller keeps the record on disk in [`RelayAuthority::Deleting`]. +/// A failure here is FATAL and propagates: the record is left fully intact +/// ([`RelayAuthority::RelayAuthoritative`]) and the delete is retryable, never +/// erased against a missing durable tombstone. +/// * **LegacyOnly**: retains the kind:5 NIP-09 tombstone at its own coordinate. +/// Returns [`TombstoneDisposition::EraseNow`]. A retention hiccup here is +/// logged and swallowed (legacy has no live authoritative aggregate to leak +/// and no async confirmation), so a disk-authoritative delete is never blocked. +/// +/// Mirrors `commands::personas::tombstone_persona_pending`: the agent row at +/// `(30177, owner, agent_pubkey)` is purged first so an unpublished edit can +/// never resurrect it after the tombstone publishes. +pub(crate) fn tombstone_managed_agent_pending( + app: &AppHandle, + state: &AppState, + agent_pubkey: &str, + relay_authority: &crate::managed_agents::RelayAuthority, +) -> Result { + use crate::managed_agents::{ + agent_events::build_agent_delete, + migration::build_tombstone_event, + retention::{ + delete_retained_event, open_retention_db, retain_event, retain_managed_agent_aggregate, + tombstone_retention_d_tag, RetainedEvent, RetainedManagedAgentAggregate, + }, + }; + use buzz_core_pkg::kind::KIND_MANAGED_AGENT; + use nostr::JsonUtil; + + const KIND_DELETE: u32 = 5; + + // Relay-canonical agents durably enqueue the authoritative tombstone before + // ANY local destruction. A failure propagates so the caller never erases + // against a missing retry — the record stays RelayAuthoritative and the + // delete is retryable. + if let Some(evidence) = relay_authority.evidence() { + let scope = crate::managed_agents::retention::active_retention_scope(app, state)?; + let owner_pubkey = scope.owner_keys.public_key().to_hex(); + + // Idempotent retry recognition. A repeat delete on this coordinate may + // find its immutable tombstone ALREADY enqueued — either because the + // record is now `Deleting` (its authority flip/save landed) or because it + // is still `RelayAuthoritative` (a cascade crash after enqueue N but + // before the batch authority save, per the self-heal contract). In BOTH + // cases rebuilding would mint a new event id / timestamp at the same + // coordinate and hit immutable-retention drift. So recognize the exact + // pending tombstone bound to THIS verified head and return DeferErase + // without touching retention; the deletion flush self-heals the authority + // state. If a pending tombstone exists but does NOT match its predecessor, + // fail closed — never rebuild over a foreign row. + { + let conn = open_retention_db(&scope.db_path)?; + let pending_deleted = + crate::managed_agents::retention::get_retained_managed_agent_aggregate( + &conn, + &owner_pubkey, + agent_pubkey, + )? + .filter(|retained| retained.state == "deleted" && retained.pending_sync); + + // Decrypt the pending row (if any) and reduce it to a single fact: + // does it name THIS exact deletion (generation = evidence.gen+1 and + // decrypted predecessor = the record's verified head)? Decryption + // stays here; the branch decision is a pure classifier below. + let matches_this_deletion = match &pending_deleted { + Some(retained) => { + let generation_ok = evidence + .generation + .checked_add(1) + .is_some_and(|next| retained.generation == next); + generation_ok && { + let parsed: serde_json::Value = + serde_json::from_str(&retained.request_json).map_err(|e| { + format!("retained deleted request json invalid: {e}") + })?; + let private_event: nostr::Event = serde_json::from_value( + parsed.get("private_event").cloned().ok_or_else(|| { + "retained deleted request missing private_event".to_string() + })?, + ) + .map_err(|e| format!("retained deleted private_event invalid: {e}"))?; + let (_, payload) = + buzz_core_pkg::private_managed_agent::validate_and_decrypt( + &private_event, + &scope.owner_keys, + ) + .map_err(|e| format!("retained deletion head does not decrypt: {e}"))?; + payload.state == buzz_core_pkg::private_managed_agent::State::Deleted + && payload.owner_pubkey == owner_pubkey + && payload.agent_pubkey == agent_pubkey + && payload.previous_event_id.as_deref() + == Some(&evidence.private_event_id) + } + } + None => false, + }; + + match classify_tombstone_retry( + pending_deleted.is_some(), + matches_this_deletion, + relay_authority.is_deleting(), + ) { + TombstoneRetry::Resume => { + return Ok(TombstoneDisposition::DeferErase { + evidence: evidence.clone(), + }) + } + TombstoneRetry::FailClosed => { + return Err( + "delete retry cannot rebuild an immutable tombstone: a pending row is \ + missing or does not match this agent's verified head (fail closed)" + .to_string(), + ) + } + TombstoneRetry::Build => {} + } + } + + let mut conn = open_retention_db(&scope.db_path)?; + // Build path: no pending tombstone yet exists for this coordinate. + // Purge the agent's 30177 row first so an unpublished edit can never + // resurrect it after the tombstone publishes. + delete_retained_event(&conn, KIND_MANAGED_AGENT, &owner_pubkey, agent_pubkey)?; + + let timestamp = crate::util::now_iso(); + let event = build_tombstone_event( + &scope.owner_keys, + agent_pubkey, + evidence.generation, + &evidence.private_event_id, + ×tamp, + nostr::Timestamp::now().as_secs(), + ) + .map_err(|error| format!("failed to build managed-agent aggregate tombstone: {error:?}"))?; + let generation = evidence + .generation + .checked_add(1) + .ok_or_else(|| "managed-agent aggregate generation overflow".to_string())?; + let request_json = serde_json::to_string(&serde_json::json!({ + "private_event": event, + "definition_event": null, + "instance_event": null, + "expected_definition_revision": null + })) + .map_err(|error| format!("failed to serialize managed-agent tombstone: {error}"))?; + retain_managed_agent_aggregate( + &mut conn, + &RetainedManagedAgentAggregate { + owner_pubkey, + agent_pubkey: agent_pubkey.to_string(), + generation, + private_event_id: event.id.to_hex(), + state: "deleted".to_string(), + request_json, + pending_sync: true, + last_error: None, + local_authority_applied: false, + }, + )?; + return Ok(TombstoneDisposition::DeferErase { + evidence: evidence.clone(), + }); + } + + // LegacyOnly: enqueue the kind:5 tombstone before erase, but best-effort — + // a retention hiccup never blocks the disk-authoritative delete. + let result = (|| -> Result<(), String> { + let scope = crate::managed_agents::retention::active_retention_scope(app, state)?; + let owner_pubkey = scope.owner_keys.public_key().to_hex(); + let conn = open_retention_db(&scope.db_path)?; + delete_retained_event(&conn, KIND_MANAGED_AGENT, &owner_pubkey, agent_pubkey)?; + + let event = build_agent_delete(agent_pubkey, &owner_pubkey)? + .sign_with_keys(&scope.owner_keys) + .map_err(|e| format!("failed to sign managed-agent tombstone: {e}"))?; + retain_event( + &conn, + &RetainedEvent { + kind: KIND_DELETE, + pubkey: owner_pubkey, + // Key by the target coordinate so cross-kind d-tag tombstones + // occupy distinct rows (F2c). + d_tag: tombstone_retention_d_tag(KIND_MANAGED_AGENT, agent_pubkey), + content: event.content.to_string(), + created_at: event.created_at.as_secs() as i64, + raw_event: event.as_json(), + pending_sync: true, + }, + ) + })(); + if let Err(e) = result { + eprintln!("buzz-desktop: agent-tombstone: {e}"); + } + Ok(TombstoneDisposition::EraseNow) +} diff --git a/desktop/src-tauri/src/commands/personas/inbound.rs b/desktop/src-tauri/src/commands/personas/inbound.rs index 2eb059e195..25e427dbcc 100644 --- a/desktop/src-tauri/src/commands/personas/inbound.rs +++ b/desktop/src-tauri/src/commands/personas/inbound.rs @@ -287,10 +287,7 @@ fn reconcile_inbound_private_managed_agent( Ok(()) } -fn merge_reconstructed_managed_agent( - local: &mut ManagedAgentRecord, - inbound: ManagedAgentRecord, -) { +fn merge_reconstructed_managed_agent(local: &mut ManagedAgentRecord, inbound: ManagedAgentRecord) { let start_on_app_launch = local.start_on_app_launch; let auto_restart = local.auto_restart_on_config_change; let provider_binary_path = local.provider_binary_path.clone(); diff --git a/desktop/src-tauri/src/managed_agents/discovery/tests.rs b/desktop/src-tauri/src/managed_agents/discovery/tests.rs index 50b3928075..c1236c43fd 100644 --- a/desktop/src-tauri/src/managed_agents/discovery/tests.rs +++ b/desktop/src-tauri/src/managed_agents/discovery/tests.rs @@ -1,5 +1,3 @@ -use std::path::PathBuf; - use super::overrides::{divergent_agent_command_override, update_time_agent_command_override}; use super::{ apply_agent_command_update, classify_runtime, codex_adapter_availability, @@ -11,6 +9,7 @@ use super::{ GOOSE_AVATAR_URL, }; use crate::managed_agents::AcpAvailabilityStatus; +use std::path::PathBuf; #[test] fn resolves_known_avatar_for_bare_command() { diff --git a/desktop/src-tauri/src/managed_agents/migration/activation.rs b/desktop/src-tauri/src/managed_agents/migration/activation.rs index f63f23dee6..f80771b052 100644 --- a/desktop/src-tauri/src/managed_agents/migration/activation.rs +++ b/desktop/src-tauri/src/managed_agents/migration/activation.rs @@ -64,13 +64,18 @@ pub(crate) fn enqueue_authoritative_edit( let retained = get_retained_managed_agent_aggregate(&conn, &owner_pubkey, &record.pubkey)? .ok_or_else(|| "authoritative edit is missing its retained relay head".to_string())?; if retained.pending_sync { - return Err("managed-agent relay synchronization is still pending; retry the edit shortly".to_string()); + return Err( + "managed-agent relay synchronization is still pending; retry the edit shortly" + .to_string(), + ); } if retained.state != "active" || retained.generation != evidence.generation || retained.private_event_id != evidence.private_event_id { - return Err("local relay evidence does not match the retained authoritative head".to_string()); + return Err( + "local relay evidence does not match the retained authoritative head".to_string(), + ); } let stored: StoredAggregateRequest = serde_json::from_str(&retained.request_json) .map_err(|error| format!("retained authoritative request is invalid: {error}"))?; diff --git a/desktop/src-tauri/src/managed_agents/readiness.rs b/desktop/src-tauri/src/managed_agents/readiness.rs index 24ab051e4e..a2dc38629c 100644 --- a/desktop/src-tauri/src/managed_agents/readiness.rs +++ b/desktop/src-tauri/src/managed_agents/readiness.rs @@ -37,7 +37,6 @@ //! separately because it is not part of the process env — the harness reads //! it at startup. We do not evaluate it here; it is exposed for future //! UI display only. - use serde::{Deserialize, Serialize}; use std::collections::BTreeMap; diff --git a/desktop/src-tauri/src/managed_agents/retention.rs b/desktop/src-tauri/src/managed_agents/retention.rs index 1fcf7b1f4b..49bb9752c6 100644 --- a/desktop/src-tauri/src/managed_agents/retention.rs +++ b/desktop/src-tauri/src/managed_agents/retention.rs @@ -106,334 +106,8 @@ pub fn arrival_retention_scope( )) } -/// One durable kind:30179 aggregate request retained for offline retry. -/// -/// The exact serialized request is immutable for a generation: retrying must -/// submit the same signed events and CAS predecessor, never rebuild them with a -/// fresh timestamp after local state has moved. -#[derive(Debug, Clone, PartialEq, Eq)] -#[allow(dead_code)] // Consumed by the PMA aggregate submit/retry driver in the next slice. -pub struct RetainedManagedAgentAggregate { - pub owner_pubkey: String, - pub agent_pubkey: String, - pub generation: u64, - pub private_event_id: String, - pub state: String, - pub request_json: String, - pub pending_sync: bool, - pub last_error: Option, - /// Durable proof that a verified deletion has already applied the local - /// record/key erase for this exact generation. Only ever set on a - /// `state = "deleted"` row, immediately before `pending_sync` is cleared, - /// so crash-replay can distinguish "our verified deletion erased the - /// record" from an unrelated/manual local deletion. Always `false` on a - /// freshly retained row. - pub local_authority_applied: bool, -} - -pub fn retire_managed_agent_aggregate( - conn: &Connection, - owner_pubkey: &str, - agent_pubkey: &str, - generation: u64, - private_event_id: &str, -) -> Result { - let changed = conn - .execute( - "DELETE FROM managed_agent_aggregates - WHERE owner_pubkey = ?1 AND agent_pubkey = ?2 - AND generation = ?3 AND private_event_id = ?4 AND pending_sync = 1", - params![owner_pubkey, agent_pubkey, generation as i64, private_event_id], - ) - .map_err(|error| format!("failed to retire managed-agent aggregate: {error}"))?; - Ok(changed == 1) -} - -pub fn seed_confirmed_managed_agent_aggregate( - conn: &Connection, - aggregate: &RetainedManagedAgentAggregate, -) -> Result<(), String> { - if aggregate.pending_sync || aggregate.state != "active" { - return Err("confirmed managed-agent seed must be a synced active head".to_string()); - } - conn.execute( - "INSERT OR IGNORE INTO managed_agent_aggregates - (owner_pubkey, agent_pubkey, generation, private_event_id, state, - request_json, pending_sync, last_error, local_authority_applied) - VALUES (?1, ?2, ?3, ?4, ?5, ?6, 0, NULL, 0)", - params![ - aggregate.owner_pubkey, - aggregate.agent_pubkey, - aggregate.generation as i64, - aggregate.private_event_id, - aggregate.state, - aggregate.request_json, - ], - ) - .map_err(|error| format!("failed to seed confirmed managed-agent aggregate: {error}"))?; - Ok(()) -} - -/// Insert or idempotently refresh a retained aggregate generation. -/// -/// Generation may advance by exactly one. A byte-identical rewrite of the -/// current generation is accepted for crash recovery; divergent same-generation -/// content and skipped/stale generations are rejected before touching disk. -#[allow(dead_code)] // Consumed by the PMA aggregate submit/retry driver in the next slice. -pub fn retain_managed_agent_aggregate( - conn: &mut Connection, - aggregate: &RetainedManagedAgentAggregate, -) -> Result<(), String> { - if aggregate.generation == 0 || aggregate.generation > i64::MAX as u64 { - return Err("managed-agent aggregate generation is out of range".to_string()); - } - if !matches!(aggregate.state.as_str(), "active" | "deleted") { - return Err("managed-agent aggregate state must be active or deleted".to_string()); - } - - if !aggregate.pending_sync || aggregate.last_error.is_some() { - return Err("new managed-agent aggregate must start pending without an error".to_string()); - } - if aggregate.local_authority_applied { - return Err( - "new managed-agent aggregate must not start with local authority applied".to_string(), - ); - } - - let tx = conn - .transaction() - .map_err(|e| format!("failed to retain managed-agent aggregate: {e}"))?; - let current = tx - .query_row( - "SELECT generation, private_event_id, state, request_json - FROM managed_agent_aggregates - WHERE owner_pubkey = ?1 AND agent_pubkey = ?2 - ORDER BY generation DESC - LIMIT 1", - params![aggregate.owner_pubkey, aggregate.agent_pubkey], - |row| { - Ok(( - row.get::<_, i64>(0)?, - row.get::<_, String>(1)?, - row.get::<_, String>(2)?, - row.get::<_, String>(3)?, - )) - }, - ) - .optional() - .map_err(|e| format!("failed to read retained managed-agent aggregate: {e}"))?; - - if let Some((generation, event_id, state, request_json)) = current { - let generation = u64::try_from(generation) - .map_err(|_| "retained managed-agent aggregate generation is invalid".to_string())?; - if aggregate.generation == generation { - if aggregate.private_event_id != event_id - || aggregate.state != state - || aggregate.request_json != request_json - { - return Err( - "managed-agent aggregate conflicts with retained generation".to_string() - ); - } - // An exact retry is a true no-op. Do not re-arm an already - // confirmed row or erase its persisted diagnostic on startup. - return Ok(()); - } - - let next_generation = generation.checked_add(1).ok_or_else(|| { - "retained managed-agent aggregate generation cannot advance".to_string() - })?; - if aggregate.generation != next_generation { - return Err(format!( - "managed-agent aggregate generation must advance from {generation} to {next_generation}" - )); - } - } else if aggregate.generation != 1 { - return Err("first retained managed-agent aggregate must be generation 1".to_string()); - } - - tx.execute( - "INSERT INTO managed_agent_aggregates - (owner_pubkey, agent_pubkey, generation, private_event_id, state, - request_json, pending_sync, last_error) - VALUES (?1, ?2, ?3, ?4, ?5, ?6, 1, NULL)", - params![ - aggregate.owner_pubkey, - aggregate.agent_pubkey, - aggregate.generation as i64, - aggregate.private_event_id, - aggregate.state, - aggregate.request_json, - ], - ) - .map_err(|e| format!("failed to write retained managed-agent aggregate: {e}"))?; - tx.commit() - .map_err(|e| format!("failed to commit retained managed-agent aggregate: {e}")) -} - -#[allow(dead_code)] // Consumed by the PMA aggregate submit/retry driver in the next slice. -pub fn get_retained_managed_agent_aggregate( - conn: &Connection, - owner_pubkey: &str, - agent_pubkey: &str, -) -> Result, String> { - conn.query_row( - "SELECT owner_pubkey, agent_pubkey, generation, private_event_id, state, - request_json, pending_sync, last_error, local_authority_applied - FROM managed_agent_aggregates - WHERE owner_pubkey = ?1 AND agent_pubkey = ?2 - ORDER BY generation DESC - LIMIT 1", - params![owner_pubkey, agent_pubkey], - aggregate_from_row, - ) - .optional() - .map_err(|e| format!("failed to get retained managed-agent aggregate: {e}")) -} - -/// Snapshot every pending aggregate attempt for one captured owner scope. -/// -/// At most one generation per agent may be pending during normal operation, -/// but selecting the latest pending generation defensively avoids replaying a -/// superseded row after a crash between generation advance and acknowledgement. -pub fn get_pending_managed_agent_aggregates( - conn: &Connection, - owner_pubkey: &str, -) -> Result, String> { - let mut stmt = conn - .prepare( - "SELECT a.owner_pubkey, a.agent_pubkey, a.generation, - a.private_event_id, a.state, a.request_json, - a.pending_sync, a.last_error, a.local_authority_applied - FROM managed_agent_aggregates a - WHERE a.owner_pubkey = ?1 AND a.pending_sync = 1 - AND a.generation = ( - SELECT MAX(latest.generation) - FROM managed_agent_aggregates latest - WHERE latest.owner_pubkey = a.owner_pubkey - AND latest.agent_pubkey = a.agent_pubkey - ) - ORDER BY a.agent_pubkey", - ) - .map_err(|e| format!("failed to prepare pending managed-agent aggregates: {e}"))?; - let rows = stmt - .query_map(params![owner_pubkey], aggregate_from_row) - .map_err(|e| format!("failed to query pending managed-agent aggregates: {e}"))?; - rows.collect::, _>>() - .map_err(|e| format!("failed to read pending managed-agent aggregate: {e}")) -} - -fn aggregate_from_row(row: &rusqlite::Row<'_>) -> rusqlite::Result { - let generation = row.get::<_, i64>(2)?; - let generation = u64::try_from(generation).map_err(|error| { - rusqlite::Error::FromSqlConversionFailure( - 2, - rusqlite::types::Type::Integer, - Box::new(error), - ) - })?; - Ok(RetainedManagedAgentAggregate { - owner_pubkey: row.get(0)?, - agent_pubkey: row.get(1)?, - generation, - private_event_id: row.get(3)?, - state: row.get(4)?, - request_json: row.get(5)?, - pending_sync: row.get::<_, i32>(6)? != 0, - last_error: row.get(7)?, - local_authority_applied: row.get::<_, i32>(8)? != 0, - }) -} - -/// Mark one exact retained aggregate request as confirmed by relay read-back. -#[allow(dead_code)] // Consumed by the PMA aggregate submit/retry driver in the next slice. -pub fn mark_managed_agent_aggregate_synced( - conn: &Connection, - owner_pubkey: &str, - agent_pubkey: &str, - generation: u64, - private_event_id: &str, -) -> Result { - let changed = conn - .execute( - "UPDATE managed_agent_aggregates - SET pending_sync = 0, last_error = NULL - WHERE owner_pubkey = ?1 AND agent_pubkey = ?2 - AND generation = ?3 AND private_event_id = ?4", - params![ - owner_pubkey, - agent_pubkey, - generation as i64, - private_event_id - ], - ) - .map_err(|e| format!("failed to mark managed-agent aggregate synced: {e}"))?; - Ok(changed == 1) -} - -/// Persist a diagnostic for one exact retained attempt without clearing retry. -#[allow(dead_code)] // Consumed by the PMA aggregate submit/retry driver in the next slice. -pub fn record_managed_agent_aggregate_error( - conn: &Connection, - owner_pubkey: &str, - agent_pubkey: &str, - generation: u64, - private_event_id: &str, - error: &str, -) -> Result { - let changed = conn - .execute( - "UPDATE managed_agent_aggregates - SET last_error = ?5, pending_sync = 1 - WHERE owner_pubkey = ?1 AND agent_pubkey = ?2 - AND generation = ?3 AND private_event_id = ?4", - params![ - owner_pubkey, - agent_pubkey, - generation as i64, - private_event_id, - error - ], - ) - .map_err(|e| format!("failed to record managed-agent aggregate error: {e}"))?; - Ok(changed == 1) -} - -/// Durably record that a verified deletion has applied the local record/key -/// erase for one exact retained tombstone generation. -/// -/// Written on the `state = "deleted"` row immediately AFTER the local erase and -/// BEFORE [`mark_managed_agent_aggregate_synced`] clears the retry. Together -/// they order the crash-safe deletion seam: erase → mark-applied → clear. On -/// replay, a set marker (never mere record absence) proves the exact deletion -/// reached local authority, so the retry may be cleared without re-erasing an -/// unrelated agent. Returns `true` iff the exact `(owner, agent, generation, -/// event)` deleted row was updated. -#[allow(dead_code)] // Consumed by the deletion flush lane. -pub fn mark_managed_agent_deletion_local_authority_applied( - conn: &Connection, - owner_pubkey: &str, - agent_pubkey: &str, - generation: u64, - private_event_id: &str, -) -> Result { - let changed = conn - .execute( - "UPDATE managed_agent_aggregates - SET local_authority_applied = 1 - WHERE owner_pubkey = ?1 AND agent_pubkey = ?2 - AND generation = ?3 AND private_event_id = ?4 - AND state = 'deleted'", - params![ - owner_pubkey, - agent_pubkey, - generation as i64, - private_event_id - ], - ) - .map_err(|e| format!("failed to mark managed-agent deletion authority applied: {e}"))?; - Ok(changed == 1) -} +mod managed_agent_aggregates; +pub use managed_agent_aggregates::*; /// A retained persona event row. #[derive(Debug, Clone)] @@ -841,713 +515,5 @@ pub fn get_retained_event( } #[cfg(test)] -mod tests { - use super::*; - - #[test] - fn retention_scope_is_stable_and_separates_relay_and_owner() { - let base = Path::new("/tmp/buzz-retention-test"); - let owner_a = "a".repeat(64); - let owner_b = "b".repeat(64); - let community_a = scoped_retention_db_path(base, "wss://a.example/", &owner_a); - assert_eq!( - community_a, - scoped_retention_db_path(base, "wss://a.example", &owner_a) - ); - assert_ne!( - community_a, - scoped_retention_db_path(base, "wss://b.example", &owner_a) - ); - assert_ne!( - community_a, - scoped_retention_db_path(base, "wss://a.example", &owner_b) - ); - } - - #[test] - fn test_arrival_relay_matching_agrees_with_database_identity() { - let base = Path::new("/tmp/buzz-retention-test"); - let keys = nostr::Keys::generate(); - let owner = keys.public_key().to_hex(); - let scope = |relay: &str| RetentionScope { - db_path: scoped_retention_db_path(base, relay, &owner), - relay_url: relay.to_string(), - owner_keys: keys.clone(), - }; - let community_a = scoped_retention_db_path(base, "wss://a.example", &owner); - - // "Same relay" and "same database" must never disagree: every URL the - // match accepts has to hash to the scope's own db path, and every URL it - // rejects has to hash somewhere else. - for equivalent in ["wss://a.example", "wss://a.example/", " wss://a.example "] { - assert_eq!( - scope_for_arrival(scope("wss://a.example"), equivalent).map(|scope| scope.db_path), - Some(community_a.clone()), - "{equivalent}" - ); - assert_eq!( - scoped_retention_db_path(base, equivalent, &owner), - community_a, - "{equivalent}" - ); - } - - assert!( - scope_for_arrival(scope("wss://b.example"), "wss://a.example").is_none(), - "an event from community A must not be filed while community B is active" - ); - assert_ne!( - scoped_retention_db_path(base, "wss://b.example", &owner), - community_a - ); - } - - #[test] - fn concurrent_open_waits_for_initialization_lock() { - let dir = tempfile::tempdir().unwrap(); - let path = dir.path().join("retention.db"); - let first = open_retention_db(&path).unwrap(); - first.execute_batch("BEGIN EXCLUSIVE").unwrap(); - - let second_path = path.clone(); - let second = std::thread::spawn(move || open_retention_db(&second_path)); - std::thread::sleep(std::time::Duration::from_millis(100)); - first.execute_batch("COMMIT").unwrap(); - - assert!(second.join().unwrap().is_ok()); - } - - fn test_db() -> Connection { - open_retention_db(Path::new(":memory:")).unwrap() - } - - fn sample_event() -> RetainedEvent { - RetainedEvent { - kind: 30175, - pubkey: "abc123".to_string(), - d_tag: "test-persona".to_string(), - content: r#"{"display_name":"Test"}"#.to_string(), - created_at: 1000, - raw_event: r#"{"id":"..."}"#.to_string(), - pending_sync: true, - } - } - - fn aggregate( - generation: u64, - event_id: &str, - state: &str, - request_json: &str, - ) -> RetainedManagedAgentAggregate { - RetainedManagedAgentAggregate { - owner_pubkey: "owner".to_string(), - agent_pubkey: "agent".to_string(), - generation, - private_event_id: event_id.to_string(), - state: state.to_string(), - request_json: request_json.to_string(), - pending_sync: true, - last_error: None, - local_authority_applied: false, - } - } - - #[test] - fn aggregate_retention_enforces_contiguous_immutable_generations() { - let mut conn = test_db(); - let first = aggregate(1, "event-1", "active", r#"{"generation":1}"#); - retain_managed_agent_aggregate(&mut conn, &first).unwrap(); - retain_managed_agent_aggregate(&mut conn, &first).unwrap(); - - let conflicting = aggregate(1, "other", "active", r#"{"generation":1}"#); - assert!(retain_managed_agent_aggregate(&mut conn, &conflicting) - .unwrap_err() - .contains("conflicts")); - let skipped = aggregate(3, "event-3", "deleted", r#"{"generation":3}"#); - assert!(retain_managed_agent_aggregate(&mut conn, &skipped) - .unwrap_err() - .contains("advance from 1 to 2")); - - let tombstone = aggregate(2, "event-2", "deleted", r#"{"generation":2}"#); - retain_managed_agent_aggregate(&mut conn, &tombstone).unwrap(); - assert_eq!( - get_retained_managed_agent_aggregate(&conn, "owner", "agent").unwrap(), - Some(tombstone) - ); - assert_eq!( - conn.query_row( - "SELECT COUNT(*) FROM managed_agent_aggregates - WHERE owner_pubkey = 'owner' AND agent_pubkey = 'agent'", - [], - |row| row.get::<_, i64>(0) - ) - .unwrap(), - 2 - ); - } - - #[test] - fn pending_aggregate_snapshot_is_owner_scoped_and_latest_only() { - let mut conn = test_db(); - retain_managed_agent_aggregate( - &mut conn, - &aggregate(1, "event-a1", "active", r#"{"generation":1}"#), - ) - .unwrap(); - retain_managed_agent_aggregate( - &mut conn, - &aggregate(2, "event-a2", "deleted", r#"{"generation":2}"#), - ) - .unwrap(); - let mut other = aggregate(1, "event-b1", "active", r#"{"generation":1}"#); - other.owner_pubkey = "other-owner".into(); - other.agent_pubkey = "other-agent".into(); - retain_managed_agent_aggregate(&mut conn, &other).unwrap(); - - let pending = get_pending_managed_agent_aggregates(&conn, "owner").unwrap(); - assert_eq!(pending.len(), 1); - assert_eq!(pending[0].generation, 2); - assert_eq!(pending[0].private_event_id, "event-a2"); - } - - #[test] - fn aggregate_compare_and_clear_cannot_ack_a_newer_attempt() { - let mut conn = test_db(); - retain_managed_agent_aggregate( - &mut conn, - &aggregate(1, "event-1", "active", r#"{"generation":1}"#), - ) - .unwrap(); - retain_managed_agent_aggregate( - &mut conn, - &aggregate(2, "event-2", "deleted", r#"{"generation":2}"#), - ) - .unwrap(); - - assert!( - mark_managed_agent_aggregate_synced(&conn, "owner", "agent", 1, "event-1").unwrap() - ); - // A stale success can clear its own retained attempt, but cannot clear - // or otherwise acknowledge the newer tombstone. - assert!( - get_retained_managed_agent_aggregate(&conn, "owner", "agent") - .unwrap() - .unwrap() - .pending_sync - ); - assert!( - mark_managed_agent_aggregate_synced(&conn, "owner", "agent", 2, "event-2").unwrap() - ); - assert!( - !get_retained_managed_agent_aggregate(&conn, "owner", "agent") - .unwrap() - .unwrap() - .pending_sync - ); - - // Rebuilding the exact same generation during startup must not turn a - // confirmed row back into pending work. - retain_managed_agent_aggregate( - &mut conn, - &aggregate(2, "event-2", "deleted", r#"{"generation":2}"#), - ) - .unwrap(); - assert!( - !get_retained_managed_agent_aggregate(&conn, "owner", "agent") - .unwrap() - .unwrap() - .pending_sync - ); - } - - #[test] - fn aggregate_error_is_scoped_to_the_exact_attempt() { - let mut conn = test_db(); - retain_managed_agent_aggregate( - &mut conn, - &aggregate(1, "event-1", "active", r#"{"generation":1}"#), - ) - .unwrap(); - assert!( - !record_managed_agent_aggregate_error(&conn, "owner", "agent", 1, "wrong", "nope") - .unwrap() - ); - assert!(record_managed_agent_aggregate_error( - &conn, - "owner", - "agent", - 1, - "event-1", - "relay unavailable" - ) - .unwrap()); - let row = get_retained_managed_agent_aggregate(&conn, "owner", "agent") - .unwrap() - .unwrap(); - assert!(row.pending_sync); - assert_eq!(row.last_error.as_deref(), Some("relay unavailable")); - } - - #[test] - fn retain_and_retrieve() { - let conn = test_db(); - let event = sample_event(); - retain_event(&conn, &event).unwrap(); - - let results = get_retained_personas(&conn, "abc123").unwrap(); - assert_eq!(results.len(), 1); - assert_eq!(results[0].d_tag, "test-persona"); - assert_eq!(results[0].created_at, 1000); - assert!(results[0].pending_sync); - } - - #[test] - fn tombstone_retention_keys_are_distinct_across_kinds() { - // A persona slug, team id, and agent pubkey that all happen to equal - // "shared" must occupy DISTINCT kind:5 rows so one tombstone's pending - // publish never clobbers another's (F2c). - let conn = test_db(); - for target_kind in [30175u32, 30176, 30177] { - retain_event( - &conn, - &RetainedEvent { - kind: 5, - pubkey: "owner".to_string(), - d_tag: tombstone_retention_d_tag(target_kind, "shared"), - content: String::new(), - created_at: 1000, - raw_event: format!("{{\"k\":{target_kind}}}"), - pending_sync: true, - }, - ) - .unwrap(); - } - // Three distinct rows survive — no PK collision clobbered any of them. - for target_kind in [30175u32, 30176, 30177] { - let row = get_retained_event( - &conn, - 5, - "owner", - &tombstone_retention_d_tag(target_kind, "shared"), - ) - .unwrap(); - assert!( - row.is_some(), - "tombstone for kind {target_kind} was clobbered" - ); - } - } - - #[test] - fn upsert_replaces_newer() { - let conn = test_db(); - let mut event = sample_event(); - retain_event(&conn, &event).unwrap(); - - event.content = r#"{"display_name":"Updated"}"#.to_string(); - event.created_at = 2000; - retain_event(&conn, &event).unwrap(); - - let results = get_retained_personas(&conn, "abc123").unwrap(); - assert_eq!(results.len(), 1); - assert_eq!(results[0].created_at, 2000); - assert!(results[0].content.contains("Updated")); - } - - #[test] - fn upsert_ignores_older() { - let conn = test_db(); - let mut event = sample_event(); - event.created_at = 2000; - retain_event(&conn, &event).unwrap(); - - event.content = r#"{"display_name":"Old"}"#.to_string(); - event.created_at = 1000; - retain_event(&conn, &event).unwrap(); - - let results = get_retained_personas(&conn, "abc123").unwrap(); - assert_eq!(results.len(), 1); - assert_eq!(results[0].created_at, 2000); - assert!(!results[0].content.contains("Old")); - } - - #[test] - fn pending_sync_query() { - let conn = test_db(); - let mut event = sample_event(); - event.pending_sync = true; - retain_event(&conn, &event).unwrap(); - - let mut event2 = sample_event(); - event2.d_tag = "other".to_string(); - event2.pending_sync = false; - retain_event(&conn, &event2).unwrap(); - - let pending = get_pending_sync(&conn).unwrap(); - assert_eq!(pending.len(), 1); - assert_eq!(pending[0].d_tag, "test-persona"); - } - - #[test] - fn test_mark_synced_matching_row_clears_flag() { - let conn = test_db(); - let event = sample_event(); - retain_event(&conn, &event).unwrap(); - - mark_synced(&conn, 30175, "abc123", "test-persona", 1000, &event.content).unwrap(); - - let pending = get_pending_sync(&conn).unwrap(); - assert!(pending.is_empty()); - - let results = get_retained_personas(&conn, "abc123").unwrap(); - assert_eq!(results.len(), 1); - assert!(!results[0].pending_sync); - } - - #[test] - fn test_mark_synced_stale_version_leaves_flag_set() { - let conn = test_db(); - let published = sample_event(); - retain_event(&conn, &published).unwrap(); - - // A newer edit lands at the same coordinate before the flush loop - // clears the version it published. - let mut newer = sample_event(); - newer.content = r#"{"display_name":"Edited"}"#.to_string(); - newer.created_at = 2000; - retain_event(&conn, &newer).unwrap(); - - // Clearing against the OLD version must not touch the newer pending row. - mark_synced( - &conn, - 30175, - "abc123", - "test-persona", - 1000, - &published.content, - ) - .unwrap(); - - let pending = get_pending_sync(&conn).unwrap(); - assert_eq!(pending.len(), 1); - assert_eq!(pending[0].created_at, 2000); - } - - #[test] - fn test_delete_retained_event_removes_row() { - let conn = test_db(); - retain_event(&conn, &sample_event()).unwrap(); - - delete_retained_event(&conn, 30175, "abc123", "test-persona").unwrap(); - - assert!(get_retained_event(&conn, 30175, "abc123", "test-persona") - .unwrap() - .is_none()); - } - - #[test] - fn test_delete_retained_event_missing_row_is_noop() { - let conn = test_db(); - delete_retained_event(&conn, 30175, "abc123", "nonexistent").unwrap(); - } - - #[test] - fn has_retained_personas_works() { - let conn = test_db(); - assert!(!has_retained_personas(&conn, "abc123").unwrap()); - - let event = sample_event(); - retain_event(&conn, &event).unwrap(); - - assert!(has_retained_personas(&conn, "abc123").unwrap()); - assert!(!has_retained_personas(&conn, "other").unwrap()); - } - - #[test] - fn get_retained_event_by_coordinate() { - let conn = test_db(); - let event = sample_event(); - retain_event(&conn, &event).unwrap(); - - let found = get_retained_event(&conn, 30175, "abc123", "test-persona").unwrap(); - assert!(found.is_some()); - assert_eq!(found.unwrap().d_tag, "test-persona"); - - let not_found = get_retained_event(&conn, 30175, "abc123", "nonexistent").unwrap(); - assert!(not_found.is_none()); - } - - #[test] - fn idempotent_retain_same_timestamp() { - let conn = test_db(); - let event = sample_event(); - retain_event(&conn, &event).unwrap(); - retain_event(&conn, &event).unwrap(); - - let results = get_retained_personas(&conn, "abc123").unwrap(); - assert_eq!(results.len(), 1); - } - - #[test] - fn inbound_no_local_row_applies() { - let conn = test_db(); - let mut event = sample_event(); - event.pending_sync = false; - - assert_eq!( - retain_inbound_event(&conn, &event).unwrap(), - InboundOutcome::Applied - ); - - let row = get_retained_event(&conn, 30175, "abc123", "test-persona") - .unwrap() - .unwrap(); - assert_eq!(row.created_at, 1000); - assert!(!row.pending_sync); - } - - #[test] - fn inbound_equal_second_skips_and_preserves_pending() { - let conn = test_db(); - // Pending local edit at t=1000. - let local = sample_event(); - retain_event(&conn, &local).unwrap(); - - // Inbound at the SAME second with different content. - let inbound = RetainedEvent { - content: r#"{"display_name":"Remote"}"#.to_string(), - pending_sync: false, - ..sample_event() - }; - assert_eq!( - retain_inbound_event(&conn, &inbound).unwrap(), - InboundOutcome::Skipped - ); - - // Local pending row is untouched: flag preserved, content unchanged so - // the flush republishes and the relay resolves last-writer-wins. - let row = get_retained_event(&conn, 30175, "abc123", "test-persona") - .unwrap() - .unwrap(); - assert!(row.pending_sync); - assert!(row.content.contains("Test")); - } - - #[test] - fn inbound_strictly_newer_applies_and_clears_pending() { - let conn = test_db(); - // Pending local edit at t=1000. - let local = sample_event(); - retain_event(&conn, &local).unwrap(); - - // Inbound strictly newer with different content. - let inbound = RetainedEvent { - content: r#"{"display_name":"Remote"}"#.to_string(), - created_at: 2000, - pending_sync: false, - ..sample_event() - }; - assert_eq!( - retain_inbound_event(&conn, &inbound).unwrap(), - InboundOutcome::Applied - ); - - // Inbound wins: content replaced and pending cleared, so the stale - // local edit stops republishing instead of looping. - let row = get_retained_event(&conn, 30175, "abc123", "test-persona") - .unwrap() - .unwrap(); - assert_eq!(row.created_at, 2000); - assert!(!row.pending_sync); - assert!(row.content.contains("Remote")); - } - - #[test] - fn inbound_older_skips() { - let conn = test_db(); - let mut local = sample_event(); - local.created_at = 2000; - retain_event(&conn, &local).unwrap(); - - let inbound = RetainedEvent { - content: r#"{"display_name":"Stale"}"#.to_string(), - created_at: 1000, - pending_sync: false, - ..sample_event() - }; - assert_eq!( - retain_inbound_event(&conn, &inbound).unwrap(), - InboundOutcome::Skipped - ); - - let row = get_retained_event(&conn, 30175, "abc123", "test-persona") - .unwrap() - .unwrap(); - assert_eq!(row.created_at, 2000); - assert!(!row.content.contains("Stale")); - } - - #[test] - fn pending_sync_publishes_tombstones_before_replacements() { - // B5 resurrection race: a kind:5 retained in session N and the same - // coordinate's replacement 30175 retained on the next boot can sit - // pending together. The relay's a-tag deletion ignores timestamps, - // so the tombstone MUST publish first or it wipes the replacement. - let conn = test_db(); - let replacement = RetainedEvent { - kind: 30175, - created_at: 2000, - pending_sync: true, - ..sample_event() - }; - retain_event(&conn, &replacement).unwrap(); - let tombstone = RetainedEvent { - kind: 5, - d_tag: tombstone_retention_d_tag(30175, "test-persona"), - content: String::new(), - created_at: 1000, - pending_sync: true, - ..sample_event() - }; - retain_event(&conn, &tombstone).unwrap(); - - let pending = get_pending_sync(&conn).unwrap(); - assert_eq!(pending.len(), 2); - assert_eq!(pending[0].kind, 5, "tombstone first"); - assert_eq!(pending[1].kind, 30175, "replacement second"); - } - - #[test] - fn deferral_predicate_is_kind_and_pubkey_qualified() { - // Mid-sweep barrier semantics: a failed tombstone defers ONLY the - // replacement at its exact coordinate — same target kind, same pubkey. - use std::collections::HashSet; - - let failed: HashSet<(String, String)> = HashSet::from([( - "abc123".to_string(), - tombstone_retention_d_tag(30175, "test-persona"), - )]); - - // The covered replacement defers. - assert!(deferred_behind_failed_tombstone( - 30175, - "abc123", - "test-persona", - &failed - )); - // Kind-qualified: a coinciding slug under a DIFFERENT kind is a - // distinct coordinate (the cross-kind collision the retention d-tag - // encoding exists to prevent) — never deferred. - assert!(!deferred_behind_failed_tombstone( - 30177, - "abc123", - "test-persona", - &failed - )); - // Never crosses pubkeys. - assert!(!deferred_behind_failed_tombstone( - 30175, - "other-key", - "test-persona", - &failed - )); - // Never defers kind:5 rows, even at a "matching" retention key. - assert!(!deferred_behind_failed_tombstone( - 5, - "abc123", - "test-persona", - &failed - )); - // Unrelated d-tags publish normally. - assert!(!deferred_behind_failed_tombstone( - 30175, - "abc123", - "other-persona", - &failed - )); - } - - #[test] - fn deletion_marker_is_the_terminal_crash_replay_proof() { - // Carl's correction: bare record absence must NOT license clearing the - // retry. The durable `local_authority_applied` marker is the proof that - // THIS exact verified deletion reached local authority. - let mut conn = test_db(); - retain_managed_agent_aggregate( - &mut conn, - &aggregate(1, "event-1", "active", r#"{"generation":1}"#), - ) - .unwrap(); - retain_managed_agent_aggregate( - &mut conn, - &aggregate(2, "event-2", "deleted", r#"{"generation":2}"#), - ) - .unwrap(); - - // Freshly retained: marker is unset — absence of the marker means the - // flush has not proven erase yet. - let row = get_retained_managed_agent_aggregate(&conn, "owner", "agent") - .unwrap() - .unwrap(); - assert!(!row.local_authority_applied); - - // A new aggregate can never be born with the marker set. - let mut premature = aggregate(3, "event-3", "deleted", r#"{"generation":3}"#); - premature.local_authority_applied = true; - assert!(retain_managed_agent_aggregate(&mut conn, &premature) - .unwrap_err() - .contains("must not start with local authority applied")); - - // The marker sets only the EXACT deleted row and only once effectively. - assert!(mark_managed_agent_deletion_local_authority_applied( - &conn, "owner", "agent", 2, "event-2" - ) - .unwrap()); - let marked = get_retained_managed_agent_aggregate(&conn, "owner", "agent") - .unwrap() - .unwrap(); - assert!( - marked.local_authority_applied, - "the marker is durable and read back" - ); - - // A wrong coordinate (generation/event/owner/agent) never sets it. - assert!(!mark_managed_agent_deletion_local_authority_applied( - &conn, - "owner", - "agent", - 2, - "wrong-event" - ) - .unwrap()); - assert!(!mark_managed_agent_deletion_local_authority_applied( - &conn, - "owner", - "other-agent", - 2, - "event-2" - ) - .unwrap()); - } - - #[test] - fn deletion_marker_refuses_a_non_deleted_row() { - // The marker is a deletion-terminal proof; it must never flip an active - // (promotion) row. `mark_managed_agent_deletion_local_authority_applied` - // is scoped to state = 'deleted'. - let mut conn = test_db(); - retain_managed_agent_aggregate( - &mut conn, - &aggregate(1, "event-1", "active", r#"{"generation":1}"#), - ) - .unwrap(); - assert!(!mark_managed_agent_deletion_local_authority_applied( - &conn, "owner", "agent", 1, "event-1" - ) - .unwrap()); - let row = get_retained_managed_agent_aggregate(&conn, "owner", "agent") - .unwrap() - .unwrap(); - assert!(!row.local_authority_applied); - } -} +#[path = "retention/tests.rs"] +mod tests; diff --git a/desktop/src-tauri/src/managed_agents/retention/managed_agent_aggregates.rs b/desktop/src-tauri/src/managed_agents/retention/managed_agent_aggregates.rs new file mode 100644 index 0000000000..e05b9ee643 --- /dev/null +++ b/desktop/src-tauri/src/managed_agents/retention/managed_agent_aggregates.rs @@ -0,0 +1,335 @@ +use rusqlite::{params, Connection, OptionalExtension}; + +/// One durable kind:30179 aggregate request retained for offline retry. +/// +/// The exact serialized request is immutable for a generation: retrying must +/// submit the same signed events and CAS predecessor, never rebuild them with a +/// fresh timestamp after local state has moved. +#[derive(Debug, Clone, PartialEq, Eq)] +#[allow(dead_code)] // Consumed by the PMA aggregate submit/retry driver in the next slice. +pub struct RetainedManagedAgentAggregate { + pub owner_pubkey: String, + pub agent_pubkey: String, + pub generation: u64, + pub private_event_id: String, + pub state: String, + pub request_json: String, + pub pending_sync: bool, + pub last_error: Option, + /// Durable proof that a verified deletion has already applied the local + /// record/key erase for this exact generation. Only ever set on a + /// `state = "deleted"` row, immediately before `pending_sync` is cleared, + /// so crash-replay can distinguish "our verified deletion erased the + /// record" from an unrelated/manual local deletion. Always `false` on a + /// freshly retained row. + pub local_authority_applied: bool, +} + +pub fn retire_managed_agent_aggregate( + conn: &Connection, + owner_pubkey: &str, + agent_pubkey: &str, + generation: u64, + private_event_id: &str, +) -> Result { + let changed = conn + .execute( + "DELETE FROM managed_agent_aggregates + WHERE owner_pubkey = ?1 AND agent_pubkey = ?2 + AND generation = ?3 AND private_event_id = ?4 AND pending_sync = 1", + params![ + owner_pubkey, + agent_pubkey, + generation as i64, + private_event_id + ], + ) + .map_err(|error| format!("failed to retire managed-agent aggregate: {error}"))?; + Ok(changed == 1) +} + +pub fn seed_confirmed_managed_agent_aggregate( + conn: &Connection, + aggregate: &RetainedManagedAgentAggregate, +) -> Result<(), String> { + if aggregate.pending_sync || aggregate.state != "active" { + return Err("confirmed managed-agent seed must be a synced active head".to_string()); + } + conn.execute( + "INSERT OR IGNORE INTO managed_agent_aggregates + (owner_pubkey, agent_pubkey, generation, private_event_id, state, + request_json, pending_sync, last_error, local_authority_applied) + VALUES (?1, ?2, ?3, ?4, ?5, ?6, 0, NULL, 0)", + params![ + aggregate.owner_pubkey, + aggregate.agent_pubkey, + aggregate.generation as i64, + aggregate.private_event_id, + aggregate.state, + aggregate.request_json, + ], + ) + .map_err(|error| format!("failed to seed confirmed managed-agent aggregate: {error}"))?; + Ok(()) +} + +/// Insert or idempotently refresh a retained aggregate generation. +/// +/// Generation may advance by exactly one. A byte-identical rewrite of the +/// current generation is accepted for crash recovery; divergent same-generation +/// content and skipped/stale generations are rejected before touching disk. +#[allow(dead_code)] // Consumed by the PMA aggregate submit/retry driver in the next slice. +pub fn retain_managed_agent_aggregate( + conn: &mut Connection, + aggregate: &RetainedManagedAgentAggregate, +) -> Result<(), String> { + if aggregate.generation == 0 || aggregate.generation > i64::MAX as u64 { + return Err("managed-agent aggregate generation is out of range".to_string()); + } + if !matches!(aggregate.state.as_str(), "active" | "deleted") { + return Err("managed-agent aggregate state must be active or deleted".to_string()); + } + + if !aggregate.pending_sync || aggregate.last_error.is_some() { + return Err("new managed-agent aggregate must start pending without an error".to_string()); + } + if aggregate.local_authority_applied { + return Err( + "new managed-agent aggregate must not start with local authority applied".to_string(), + ); + } + + let tx = conn + .transaction() + .map_err(|e| format!("failed to retain managed-agent aggregate: {e}"))?; + let current = tx + .query_row( + "SELECT generation, private_event_id, state, request_json + FROM managed_agent_aggregates + WHERE owner_pubkey = ?1 AND agent_pubkey = ?2 + ORDER BY generation DESC + LIMIT 1", + params![aggregate.owner_pubkey, aggregate.agent_pubkey], + |row| { + Ok(( + row.get::<_, i64>(0)?, + row.get::<_, String>(1)?, + row.get::<_, String>(2)?, + row.get::<_, String>(3)?, + )) + }, + ) + .optional() + .map_err(|e| format!("failed to read retained managed-agent aggregate: {e}"))?; + + if let Some((generation, event_id, state, request_json)) = current { + let generation = u64::try_from(generation) + .map_err(|_| "retained managed-agent aggregate generation is invalid".to_string())?; + if aggregate.generation == generation { + if aggregate.private_event_id != event_id + || aggregate.state != state + || aggregate.request_json != request_json + { + return Err( + "managed-agent aggregate conflicts with retained generation".to_string() + ); + } + // An exact retry is a true no-op. Do not re-arm an already + // confirmed row or erase its persisted diagnostic on startup. + return Ok(()); + } + + let next_generation = generation.checked_add(1).ok_or_else(|| { + "retained managed-agent aggregate generation cannot advance".to_string() + })?; + if aggregate.generation != next_generation { + return Err(format!( + "managed-agent aggregate generation must advance from {generation} to {next_generation}" + )); + } + } else if aggregate.generation != 1 { + return Err("first retained managed-agent aggregate must be generation 1".to_string()); + } + + tx.execute( + "INSERT INTO managed_agent_aggregates + (owner_pubkey, agent_pubkey, generation, private_event_id, state, + request_json, pending_sync, last_error) + VALUES (?1, ?2, ?3, ?4, ?5, ?6, 1, NULL)", + params![ + aggregate.owner_pubkey, + aggregate.agent_pubkey, + aggregate.generation as i64, + aggregate.private_event_id, + aggregate.state, + aggregate.request_json, + ], + ) + .map_err(|e| format!("failed to write retained managed-agent aggregate: {e}"))?; + tx.commit() + .map_err(|e| format!("failed to commit retained managed-agent aggregate: {e}")) +} + +#[allow(dead_code)] // Consumed by the PMA aggregate submit/retry driver in the next slice. +pub fn get_retained_managed_agent_aggregate( + conn: &Connection, + owner_pubkey: &str, + agent_pubkey: &str, +) -> Result, String> { + conn.query_row( + "SELECT owner_pubkey, agent_pubkey, generation, private_event_id, state, + request_json, pending_sync, last_error, local_authority_applied + FROM managed_agent_aggregates + WHERE owner_pubkey = ?1 AND agent_pubkey = ?2 + ORDER BY generation DESC + LIMIT 1", + params![owner_pubkey, agent_pubkey], + aggregate_from_row, + ) + .optional() + .map_err(|e| format!("failed to get retained managed-agent aggregate: {e}")) +} + +/// Snapshot every pending aggregate attempt for one captured owner scope. +/// +/// At most one generation per agent may be pending during normal operation, +/// but selecting the latest pending generation defensively avoids replaying a +/// superseded row after a crash between generation advance and acknowledgement. +pub fn get_pending_managed_agent_aggregates( + conn: &Connection, + owner_pubkey: &str, +) -> Result, String> { + let mut stmt = conn + .prepare( + "SELECT a.owner_pubkey, a.agent_pubkey, a.generation, + a.private_event_id, a.state, a.request_json, + a.pending_sync, a.last_error, a.local_authority_applied + FROM managed_agent_aggregates a + WHERE a.owner_pubkey = ?1 AND a.pending_sync = 1 + AND a.generation = ( + SELECT MAX(latest.generation) + FROM managed_agent_aggregates latest + WHERE latest.owner_pubkey = a.owner_pubkey + AND latest.agent_pubkey = a.agent_pubkey + ) + ORDER BY a.agent_pubkey", + ) + .map_err(|e| format!("failed to prepare pending managed-agent aggregates: {e}"))?; + let rows = stmt + .query_map(params![owner_pubkey], aggregate_from_row) + .map_err(|e| format!("failed to query pending managed-agent aggregates: {e}"))?; + rows.collect::, _>>() + .map_err(|e| format!("failed to read pending managed-agent aggregate: {e}")) +} + +fn aggregate_from_row(row: &rusqlite::Row<'_>) -> rusqlite::Result { + let generation = row.get::<_, i64>(2)?; + let generation = u64::try_from(generation).map_err(|error| { + rusqlite::Error::FromSqlConversionFailure( + 2, + rusqlite::types::Type::Integer, + Box::new(error), + ) + })?; + Ok(RetainedManagedAgentAggregate { + owner_pubkey: row.get(0)?, + agent_pubkey: row.get(1)?, + generation, + private_event_id: row.get(3)?, + state: row.get(4)?, + request_json: row.get(5)?, + pending_sync: row.get::<_, i32>(6)? != 0, + last_error: row.get(7)?, + local_authority_applied: row.get::<_, i32>(8)? != 0, + }) +} + +/// Mark one exact retained aggregate request as confirmed by relay read-back. +#[allow(dead_code)] // Consumed by the PMA aggregate submit/retry driver in the next slice. +pub fn mark_managed_agent_aggregate_synced( + conn: &Connection, + owner_pubkey: &str, + agent_pubkey: &str, + generation: u64, + private_event_id: &str, +) -> Result { + let changed = conn + .execute( + "UPDATE managed_agent_aggregates + SET pending_sync = 0, last_error = NULL + WHERE owner_pubkey = ?1 AND agent_pubkey = ?2 + AND generation = ?3 AND private_event_id = ?4", + params![ + owner_pubkey, + agent_pubkey, + generation as i64, + private_event_id + ], + ) + .map_err(|e| format!("failed to mark managed-agent aggregate synced: {e}"))?; + Ok(changed == 1) +} + +/// Persist a diagnostic for one exact retained attempt without clearing retry. +#[allow(dead_code)] // Consumed by the PMA aggregate submit/retry driver in the next slice. +pub fn record_managed_agent_aggregate_error( + conn: &Connection, + owner_pubkey: &str, + agent_pubkey: &str, + generation: u64, + private_event_id: &str, + error: &str, +) -> Result { + let changed = conn + .execute( + "UPDATE managed_agent_aggregates + SET last_error = ?5, pending_sync = 1 + WHERE owner_pubkey = ?1 AND agent_pubkey = ?2 + AND generation = ?3 AND private_event_id = ?4", + params![ + owner_pubkey, + agent_pubkey, + generation as i64, + private_event_id, + error + ], + ) + .map_err(|e| format!("failed to record managed-agent aggregate error: {e}"))?; + Ok(changed == 1) +} + +/// Durably record that a verified deletion has applied the local record/key +/// erase for one exact retained tombstone generation. +/// +/// Written on the `state = "deleted"` row immediately AFTER the local erase and +/// BEFORE [`mark_managed_agent_aggregate_synced`] clears the retry. Together +/// they order the crash-safe deletion seam: erase → mark-applied → clear. On +/// replay, a set marker (never mere record absence) proves the exact deletion +/// reached local authority, so the retry may be cleared without re-erasing an +/// unrelated agent. Returns `true` iff the exact `(owner, agent, generation, +/// event)` deleted row was updated. +#[allow(dead_code)] // Consumed by the deletion flush lane. +pub fn mark_managed_agent_deletion_local_authority_applied( + conn: &Connection, + owner_pubkey: &str, + agent_pubkey: &str, + generation: u64, + private_event_id: &str, +) -> Result { + let changed = conn + .execute( + "UPDATE managed_agent_aggregates + SET local_authority_applied = 1 + WHERE owner_pubkey = ?1 AND agent_pubkey = ?2 + AND generation = ?3 AND private_event_id = ?4 + AND state = 'deleted'", + params![ + owner_pubkey, + agent_pubkey, + generation as i64, + private_event_id + ], + ) + .map_err(|e| format!("failed to mark managed-agent deletion authority applied: {e}"))?; + Ok(changed == 1) +} diff --git a/desktop/src-tauri/src/managed_agents/retention/tests.rs b/desktop/src-tauri/src/managed_agents/retention/tests.rs new file mode 100644 index 0000000000..70b157c6b3 --- /dev/null +++ b/desktop/src-tauri/src/managed_agents/retention/tests.rs @@ -0,0 +1,703 @@ +use super::*; + +#[test] +fn retention_scope_is_stable_and_separates_relay_and_owner() { + let base = Path::new("/tmp/buzz-retention-test"); + let owner_a = "a".repeat(64); + let owner_b = "b".repeat(64); + let community_a = scoped_retention_db_path(base, "wss://a.example/", &owner_a); + assert_eq!( + community_a, + scoped_retention_db_path(base, "wss://a.example", &owner_a) + ); + assert_ne!( + community_a, + scoped_retention_db_path(base, "wss://b.example", &owner_a) + ); + assert_ne!( + community_a, + scoped_retention_db_path(base, "wss://a.example", &owner_b) + ); +} + +#[test] +fn test_arrival_relay_matching_agrees_with_database_identity() { + let base = Path::new("/tmp/buzz-retention-test"); + let keys = nostr::Keys::generate(); + let owner = keys.public_key().to_hex(); + let scope = |relay: &str| RetentionScope { + db_path: scoped_retention_db_path(base, relay, &owner), + relay_url: relay.to_string(), + owner_keys: keys.clone(), + }; + let community_a = scoped_retention_db_path(base, "wss://a.example", &owner); + + // "Same relay" and "same database" must never disagree: every URL the + // match accepts has to hash to the scope's own db path, and every URL it + // rejects has to hash somewhere else. + for equivalent in ["wss://a.example", "wss://a.example/", " wss://a.example "] { + assert_eq!( + scope_for_arrival(scope("wss://a.example"), equivalent).map(|scope| scope.db_path), + Some(community_a.clone()), + "{equivalent}" + ); + assert_eq!( + scoped_retention_db_path(base, equivalent, &owner), + community_a, + "{equivalent}" + ); + } + + assert!( + scope_for_arrival(scope("wss://b.example"), "wss://a.example").is_none(), + "an event from community A must not be filed while community B is active" + ); + assert_ne!( + scoped_retention_db_path(base, "wss://b.example", &owner), + community_a + ); +} + +#[test] +fn concurrent_open_waits_for_initialization_lock() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("retention.db"); + let first = open_retention_db(&path).unwrap(); + first.execute_batch("BEGIN EXCLUSIVE").unwrap(); + + let second_path = path.clone(); + let second = std::thread::spawn(move || open_retention_db(&second_path)); + std::thread::sleep(std::time::Duration::from_millis(100)); + first.execute_batch("COMMIT").unwrap(); + + assert!(second.join().unwrap().is_ok()); +} + +fn test_db() -> Connection { + open_retention_db(Path::new(":memory:")).unwrap() +} + +fn sample_event() -> RetainedEvent { + RetainedEvent { + kind: 30175, + pubkey: "abc123".to_string(), + d_tag: "test-persona".to_string(), + content: r#"{"display_name":"Test"}"#.to_string(), + created_at: 1000, + raw_event: r#"{"id":"..."}"#.to_string(), + pending_sync: true, + } +} + +fn aggregate( + generation: u64, + event_id: &str, + state: &str, + request_json: &str, +) -> RetainedManagedAgentAggregate { + RetainedManagedAgentAggregate { + owner_pubkey: "owner".to_string(), + agent_pubkey: "agent".to_string(), + generation, + private_event_id: event_id.to_string(), + state: state.to_string(), + request_json: request_json.to_string(), + pending_sync: true, + last_error: None, + local_authority_applied: false, + } +} + +#[test] +fn aggregate_retention_enforces_contiguous_immutable_generations() { + let mut conn = test_db(); + let first = aggregate(1, "event-1", "active", r#"{"generation":1}"#); + retain_managed_agent_aggregate(&mut conn, &first).unwrap(); + retain_managed_agent_aggregate(&mut conn, &first).unwrap(); + + let conflicting = aggregate(1, "other", "active", r#"{"generation":1}"#); + assert!(retain_managed_agent_aggregate(&mut conn, &conflicting) + .unwrap_err() + .contains("conflicts")); + let skipped = aggregate(3, "event-3", "deleted", r#"{"generation":3}"#); + assert!(retain_managed_agent_aggregate(&mut conn, &skipped) + .unwrap_err() + .contains("advance from 1 to 2")); + + let tombstone = aggregate(2, "event-2", "deleted", r#"{"generation":2}"#); + retain_managed_agent_aggregate(&mut conn, &tombstone).unwrap(); + assert_eq!( + get_retained_managed_agent_aggregate(&conn, "owner", "agent").unwrap(), + Some(tombstone) + ); + assert_eq!( + conn.query_row( + "SELECT COUNT(*) FROM managed_agent_aggregates + WHERE owner_pubkey = 'owner' AND agent_pubkey = 'agent'", + [], + |row| row.get::<_, i64>(0) + ) + .unwrap(), + 2 + ); +} + +#[test] +fn pending_aggregate_snapshot_is_owner_scoped_and_latest_only() { + let mut conn = test_db(); + retain_managed_agent_aggregate( + &mut conn, + &aggregate(1, "event-a1", "active", r#"{"generation":1}"#), + ) + .unwrap(); + retain_managed_agent_aggregate( + &mut conn, + &aggregate(2, "event-a2", "deleted", r#"{"generation":2}"#), + ) + .unwrap(); + let mut other = aggregate(1, "event-b1", "active", r#"{"generation":1}"#); + other.owner_pubkey = "other-owner".into(); + other.agent_pubkey = "other-agent".into(); + retain_managed_agent_aggregate(&mut conn, &other).unwrap(); + + let pending = get_pending_managed_agent_aggregates(&conn, "owner").unwrap(); + assert_eq!(pending.len(), 1); + assert_eq!(pending[0].generation, 2); + assert_eq!(pending[0].private_event_id, "event-a2"); +} + +#[test] +fn aggregate_compare_and_clear_cannot_ack_a_newer_attempt() { + let mut conn = test_db(); + retain_managed_agent_aggregate( + &mut conn, + &aggregate(1, "event-1", "active", r#"{"generation":1}"#), + ) + .unwrap(); + retain_managed_agent_aggregate( + &mut conn, + &aggregate(2, "event-2", "deleted", r#"{"generation":2}"#), + ) + .unwrap(); + + assert!(mark_managed_agent_aggregate_synced(&conn, "owner", "agent", 1, "event-1").unwrap()); + // A stale success can clear its own retained attempt, but cannot clear + // or otherwise acknowledge the newer tombstone. + assert!( + get_retained_managed_agent_aggregate(&conn, "owner", "agent") + .unwrap() + .unwrap() + .pending_sync + ); + assert!(mark_managed_agent_aggregate_synced(&conn, "owner", "agent", 2, "event-2").unwrap()); + assert!( + !get_retained_managed_agent_aggregate(&conn, "owner", "agent") + .unwrap() + .unwrap() + .pending_sync + ); + + // Rebuilding the exact same generation during startup must not turn a + // confirmed row back into pending work. + retain_managed_agent_aggregate( + &mut conn, + &aggregate(2, "event-2", "deleted", r#"{"generation":2}"#), + ) + .unwrap(); + assert!( + !get_retained_managed_agent_aggregate(&conn, "owner", "agent") + .unwrap() + .unwrap() + .pending_sync + ); +} + +#[test] +fn aggregate_error_is_scoped_to_the_exact_attempt() { + let mut conn = test_db(); + retain_managed_agent_aggregate( + &mut conn, + &aggregate(1, "event-1", "active", r#"{"generation":1}"#), + ) + .unwrap(); + assert!( + !record_managed_agent_aggregate_error(&conn, "owner", "agent", 1, "wrong", "nope").unwrap() + ); + assert!(record_managed_agent_aggregate_error( + &conn, + "owner", + "agent", + 1, + "event-1", + "relay unavailable" + ) + .unwrap()); + let row = get_retained_managed_agent_aggregate(&conn, "owner", "agent") + .unwrap() + .unwrap(); + assert!(row.pending_sync); + assert_eq!(row.last_error.as_deref(), Some("relay unavailable")); +} + +#[test] +fn retain_and_retrieve() { + let conn = test_db(); + let event = sample_event(); + retain_event(&conn, &event).unwrap(); + + let results = get_retained_personas(&conn, "abc123").unwrap(); + assert_eq!(results.len(), 1); + assert_eq!(results[0].d_tag, "test-persona"); + assert_eq!(results[0].created_at, 1000); + assert!(results[0].pending_sync); +} + +#[test] +fn tombstone_retention_keys_are_distinct_across_kinds() { + // A persona slug, team id, and agent pubkey that all happen to equal + // "shared" must occupy DISTINCT kind:5 rows so one tombstone's pending + // publish never clobbers another's (F2c). + let conn = test_db(); + for target_kind in [30175u32, 30176, 30177] { + retain_event( + &conn, + &RetainedEvent { + kind: 5, + pubkey: "owner".to_string(), + d_tag: tombstone_retention_d_tag(target_kind, "shared"), + content: String::new(), + created_at: 1000, + raw_event: format!("{{\"k\":{target_kind}}}"), + pending_sync: true, + }, + ) + .unwrap(); + } + // Three distinct rows survive — no PK collision clobbered any of them. + for target_kind in [30175u32, 30176, 30177] { + let row = get_retained_event( + &conn, + 5, + "owner", + &tombstone_retention_d_tag(target_kind, "shared"), + ) + .unwrap(); + assert!( + row.is_some(), + "tombstone for kind {target_kind} was clobbered" + ); + } +} + +#[test] +fn upsert_replaces_newer() { + let conn = test_db(); + let mut event = sample_event(); + retain_event(&conn, &event).unwrap(); + + event.content = r#"{"display_name":"Updated"}"#.to_string(); + event.created_at = 2000; + retain_event(&conn, &event).unwrap(); + + let results = get_retained_personas(&conn, "abc123").unwrap(); + assert_eq!(results.len(), 1); + assert_eq!(results[0].created_at, 2000); + assert!(results[0].content.contains("Updated")); +} + +#[test] +fn upsert_ignores_older() { + let conn = test_db(); + let mut event = sample_event(); + event.created_at = 2000; + retain_event(&conn, &event).unwrap(); + + event.content = r#"{"display_name":"Old"}"#.to_string(); + event.created_at = 1000; + retain_event(&conn, &event).unwrap(); + + let results = get_retained_personas(&conn, "abc123").unwrap(); + assert_eq!(results.len(), 1); + assert_eq!(results[0].created_at, 2000); + assert!(!results[0].content.contains("Old")); +} + +#[test] +fn pending_sync_query() { + let conn = test_db(); + let mut event = sample_event(); + event.pending_sync = true; + retain_event(&conn, &event).unwrap(); + + let mut event2 = sample_event(); + event2.d_tag = "other".to_string(); + event2.pending_sync = false; + retain_event(&conn, &event2).unwrap(); + + let pending = get_pending_sync(&conn).unwrap(); + assert_eq!(pending.len(), 1); + assert_eq!(pending[0].d_tag, "test-persona"); +} + +#[test] +fn test_mark_synced_matching_row_clears_flag() { + let conn = test_db(); + let event = sample_event(); + retain_event(&conn, &event).unwrap(); + + mark_synced(&conn, 30175, "abc123", "test-persona", 1000, &event.content).unwrap(); + + let pending = get_pending_sync(&conn).unwrap(); + assert!(pending.is_empty()); + + let results = get_retained_personas(&conn, "abc123").unwrap(); + assert_eq!(results.len(), 1); + assert!(!results[0].pending_sync); +} + +#[test] +fn test_mark_synced_stale_version_leaves_flag_set() { + let conn = test_db(); + let published = sample_event(); + retain_event(&conn, &published).unwrap(); + + // A newer edit lands at the same coordinate before the flush loop + // clears the version it published. + let mut newer = sample_event(); + newer.content = r#"{"display_name":"Edited"}"#.to_string(); + newer.created_at = 2000; + retain_event(&conn, &newer).unwrap(); + + // Clearing against the OLD version must not touch the newer pending row. + mark_synced( + &conn, + 30175, + "abc123", + "test-persona", + 1000, + &published.content, + ) + .unwrap(); + + let pending = get_pending_sync(&conn).unwrap(); + assert_eq!(pending.len(), 1); + assert_eq!(pending[0].created_at, 2000); +} + +#[test] +fn test_delete_retained_event_removes_row() { + let conn = test_db(); + retain_event(&conn, &sample_event()).unwrap(); + + delete_retained_event(&conn, 30175, "abc123", "test-persona").unwrap(); + + assert!(get_retained_event(&conn, 30175, "abc123", "test-persona") + .unwrap() + .is_none()); +} + +#[test] +fn test_delete_retained_event_missing_row_is_noop() { + let conn = test_db(); + delete_retained_event(&conn, 30175, "abc123", "nonexistent").unwrap(); +} + +#[test] +fn has_retained_personas_works() { + let conn = test_db(); + assert!(!has_retained_personas(&conn, "abc123").unwrap()); + + let event = sample_event(); + retain_event(&conn, &event).unwrap(); + + assert!(has_retained_personas(&conn, "abc123").unwrap()); + assert!(!has_retained_personas(&conn, "other").unwrap()); +} + +#[test] +fn get_retained_event_by_coordinate() { + let conn = test_db(); + let event = sample_event(); + retain_event(&conn, &event).unwrap(); + + let found = get_retained_event(&conn, 30175, "abc123", "test-persona").unwrap(); + assert!(found.is_some()); + assert_eq!(found.unwrap().d_tag, "test-persona"); + + let not_found = get_retained_event(&conn, 30175, "abc123", "nonexistent").unwrap(); + assert!(not_found.is_none()); +} + +#[test] +fn idempotent_retain_same_timestamp() { + let conn = test_db(); + let event = sample_event(); + retain_event(&conn, &event).unwrap(); + retain_event(&conn, &event).unwrap(); + + let results = get_retained_personas(&conn, "abc123").unwrap(); + assert_eq!(results.len(), 1); +} + +#[test] +fn inbound_no_local_row_applies() { + let conn = test_db(); + let mut event = sample_event(); + event.pending_sync = false; + + assert_eq!( + retain_inbound_event(&conn, &event).unwrap(), + InboundOutcome::Applied + ); + + let row = get_retained_event(&conn, 30175, "abc123", "test-persona") + .unwrap() + .unwrap(); + assert_eq!(row.created_at, 1000); + assert!(!row.pending_sync); +} + +#[test] +fn inbound_equal_second_skips_and_preserves_pending() { + let conn = test_db(); + // Pending local edit at t=1000. + let local = sample_event(); + retain_event(&conn, &local).unwrap(); + + // Inbound at the SAME second with different content. + let inbound = RetainedEvent { + content: r#"{"display_name":"Remote"}"#.to_string(), + pending_sync: false, + ..sample_event() + }; + assert_eq!( + retain_inbound_event(&conn, &inbound).unwrap(), + InboundOutcome::Skipped + ); + + // Local pending row is untouched: flag preserved, content unchanged so + // the flush republishes and the relay resolves last-writer-wins. + let row = get_retained_event(&conn, 30175, "abc123", "test-persona") + .unwrap() + .unwrap(); + assert!(row.pending_sync); + assert!(row.content.contains("Test")); +} + +#[test] +fn inbound_strictly_newer_applies_and_clears_pending() { + let conn = test_db(); + // Pending local edit at t=1000. + let local = sample_event(); + retain_event(&conn, &local).unwrap(); + + // Inbound strictly newer with different content. + let inbound = RetainedEvent { + content: r#"{"display_name":"Remote"}"#.to_string(), + created_at: 2000, + pending_sync: false, + ..sample_event() + }; + assert_eq!( + retain_inbound_event(&conn, &inbound).unwrap(), + InboundOutcome::Applied + ); + + // Inbound wins: content replaced and pending cleared, so the stale + // local edit stops republishing instead of looping. + let row = get_retained_event(&conn, 30175, "abc123", "test-persona") + .unwrap() + .unwrap(); + assert_eq!(row.created_at, 2000); + assert!(!row.pending_sync); + assert!(row.content.contains("Remote")); +} + +#[test] +fn inbound_older_skips() { + let conn = test_db(); + let mut local = sample_event(); + local.created_at = 2000; + retain_event(&conn, &local).unwrap(); + + let inbound = RetainedEvent { + content: r#"{"display_name":"Stale"}"#.to_string(), + created_at: 1000, + pending_sync: false, + ..sample_event() + }; + assert_eq!( + retain_inbound_event(&conn, &inbound).unwrap(), + InboundOutcome::Skipped + ); + + let row = get_retained_event(&conn, 30175, "abc123", "test-persona") + .unwrap() + .unwrap(); + assert_eq!(row.created_at, 2000); + assert!(!row.content.contains("Stale")); +} + +#[test] +fn pending_sync_publishes_tombstones_before_replacements() { + // B5 resurrection race: a kind:5 retained in session N and the same + // coordinate's replacement 30175 retained on the next boot can sit + // pending together. The relay's a-tag deletion ignores timestamps, + // so the tombstone MUST publish first or it wipes the replacement. + let conn = test_db(); + let replacement = RetainedEvent { + kind: 30175, + created_at: 2000, + pending_sync: true, + ..sample_event() + }; + retain_event(&conn, &replacement).unwrap(); + let tombstone = RetainedEvent { + kind: 5, + d_tag: tombstone_retention_d_tag(30175, "test-persona"), + content: String::new(), + created_at: 1000, + pending_sync: true, + ..sample_event() + }; + retain_event(&conn, &tombstone).unwrap(); + + let pending = get_pending_sync(&conn).unwrap(); + assert_eq!(pending.len(), 2); + assert_eq!(pending[0].kind, 5, "tombstone first"); + assert_eq!(pending[1].kind, 30175, "replacement second"); +} + +#[test] +fn deferral_predicate_is_kind_and_pubkey_qualified() { + // Mid-sweep barrier semantics: a failed tombstone defers ONLY the + // replacement at its exact coordinate — same target kind, same pubkey. + use std::collections::HashSet; + + let failed: HashSet<(String, String)> = HashSet::from([( + "abc123".to_string(), + tombstone_retention_d_tag(30175, "test-persona"), + )]); + + // The covered replacement defers. + assert!(deferred_behind_failed_tombstone( + 30175, + "abc123", + "test-persona", + &failed + )); + // Kind-qualified: a coinciding slug under a DIFFERENT kind is a + // distinct coordinate (the cross-kind collision the retention d-tag + // encoding exists to prevent) — never deferred. + assert!(!deferred_behind_failed_tombstone( + 30177, + "abc123", + "test-persona", + &failed + )); + // Never crosses pubkeys. + assert!(!deferred_behind_failed_tombstone( + 30175, + "other-key", + "test-persona", + &failed + )); + // Never defers kind:5 rows, even at a "matching" retention key. + assert!(!deferred_behind_failed_tombstone( + 5, + "abc123", + "test-persona", + &failed + )); + // Unrelated d-tags publish normally. + assert!(!deferred_behind_failed_tombstone( + 30175, + "abc123", + "other-persona", + &failed + )); +} + +#[test] +fn deletion_marker_is_the_terminal_crash_replay_proof() { + // Carl's correction: bare record absence must NOT license clearing the + // retry. The durable `local_authority_applied` marker is the proof that + // THIS exact verified deletion reached local authority. + let mut conn = test_db(); + retain_managed_agent_aggregate( + &mut conn, + &aggregate(1, "event-1", "active", r#"{"generation":1}"#), + ) + .unwrap(); + retain_managed_agent_aggregate( + &mut conn, + &aggregate(2, "event-2", "deleted", r#"{"generation":2}"#), + ) + .unwrap(); + + // Freshly retained: marker is unset — absence of the marker means the + // flush has not proven erase yet. + let row = get_retained_managed_agent_aggregate(&conn, "owner", "agent") + .unwrap() + .unwrap(); + assert!(!row.local_authority_applied); + + // A new aggregate can never be born with the marker set. + let mut premature = aggregate(3, "event-3", "deleted", r#"{"generation":3}"#); + premature.local_authority_applied = true; + assert!(retain_managed_agent_aggregate(&mut conn, &premature) + .unwrap_err() + .contains("must not start with local authority applied")); + + // The marker sets only the EXACT deleted row and only once effectively. + assert!(mark_managed_agent_deletion_local_authority_applied( + &conn, "owner", "agent", 2, "event-2" + ) + .unwrap()); + let marked = get_retained_managed_agent_aggregate(&conn, "owner", "agent") + .unwrap() + .unwrap(); + assert!( + marked.local_authority_applied, + "the marker is durable and read back" + ); + + // A wrong coordinate (generation/event/owner/agent) never sets it. + assert!(!mark_managed_agent_deletion_local_authority_applied( + &conn, + "owner", + "agent", + 2, + "wrong-event" + ) + .unwrap()); + assert!(!mark_managed_agent_deletion_local_authority_applied( + &conn, + "owner", + "other-agent", + 2, + "event-2" + ) + .unwrap()); +} + +#[test] +fn deletion_marker_refuses_a_non_deleted_row() { + // The marker is a deletion-terminal proof; it must never flip an active + // (promotion) row. `mark_managed_agent_deletion_local_authority_applied` + // is scoped to state = 'deleted'. + let mut conn = test_db(); + retain_managed_agent_aggregate( + &mut conn, + &aggregate(1, "event-1", "active", r#"{"generation":1}"#), + ) + .unwrap(); + assert!(!mark_managed_agent_deletion_local_authority_applied( + &conn, "owner", "agent", 1, "event-1" + ) + .unwrap()); + let row = get_retained_managed_agent_aggregate(&conn, "owner", "agent") + .unwrap() + .unwrap(); + assert!(!row.local_authority_applied); +} diff --git a/desktop/src-tauri/src/managed_agents/runtime/tests.rs b/desktop/src-tauri/src/managed_agents/runtime/tests.rs index 66662d31b1..ffc81d93e9 100644 --- a/desktop/src-tauri/src/managed_agents/runtime/tests.rs +++ b/desktop/src-tauri/src/managed_agents/runtime/tests.rs @@ -1,5 +1,4 @@ use crate::managed_agents::known_acp_runtime; - // ── desktop binary name tests ─────────────────────────────────────────── #[test] diff --git a/desktop/src-tauri/src/managed_agents/types.rs b/desktop/src-tauri/src/managed_agents/types.rs index d4cc06d50a..422a71d34c 100644 --- a/desktop/src-tauri/src/managed_agents/types.rs +++ b/desktop/src-tauri/src/managed_agents/types.rs @@ -897,109 +897,8 @@ impl RespondTo { /// - Each entry is exactly 64 hex chars (any case in, lowercase out). /// - Duplicates removed, insertion order preserved. /// -/// Empty input is allowed here — the boundary check (allowlist mode requires -/// at least one entry) is the caller's job, because an `UpdateManagedAgentRequest` -/// may want to validate a list without yet knowing the final mode. -pub fn validate_respond_to_allowlist(input: &[String]) -> Result, String> { - let mut seen = std::collections::HashSet::new(); - let mut out = Vec::with_capacity(input.len()); - for entry in input { - let trimmed = entry.trim(); - if trimmed.len() != 64 || !trimmed.chars().all(|c| c.is_ascii_hexdigit()) { - return Err(format!( - "invalid pubkey in respond-to allowlist: '{trimmed}' (must be 64 hex chars)" - )); - } - let lower = trimmed.to_ascii_lowercase(); - if seen.insert(lower.clone()) { - out.push(lower); - } - } - Ok(out) -} - -/// The behavioral fields resolved for a new instance at mint time. -#[derive(Debug, PartialEq, Eq)] -pub struct MintBehavioralDefaults { - pub respond_to: RespondTo, - pub respond_to_allowlist: Vec, - /// Validated (1..=32) when present; caller applies its own default. - pub parallelism: Option, -} - -/// Resolve the NIP-AP behavioral quad for a new instance: explicit input -/// wins, then the linked definition's defaults, then client defaults. -/// -/// This is the ONLY place definition behavioral strings are parsed — an -/// unrecognized `respond_to` mode or out-of-range `parallelism` on a -/// definition fails the mint loudly instead of silently substituting a -/// default the definition author did not choose. The empty-allowlist guard -/// fires here too, because inbound definitions bypass the dialog entirely. -/// -/// `input_allowlist` must already be normalized via -/// [`validate_respond_to_allowlist`]; the definition's allowlist is -/// validated here since it arrives from the wire. -pub fn resolve_mint_behavioral_defaults( - input_respond_to: Option, - input_allowlist: Vec, - input_parallelism: Option, - definition: Option<&AgentDefinition>, -) -> Result { - let (respond_to, respond_to_allowlist) = match input_respond_to { - // Explicit instance-level choice: the definition default is ignored - // wholesale (mode AND list travel together). - Some(mode) => (mode, input_allowlist), - None => match definition.and_then(|d| d.respond_to.as_deref()) { - Some(wire) => { - let mode = RespondTo::parse_wire(wire)?; - let list = if input_allowlist.is_empty() { - validate_respond_to_allowlist( - definition - .map(|d| d.respond_to_allowlist.as_slice()) - .unwrap_or(&[]), - ) - .map_err(|e| format!("definition respond-to allowlist is invalid: {e}"))? - } else { - input_allowlist - }; - (mode, list) - } - None => (RespondTo::default(), input_allowlist), - }, - }; - if respond_to == RespondTo::Allowlist && respond_to_allowlist.is_empty() { - return Err( - "respond-to mode 'allowlist' requires at least one pubkey in the allowlist".to_string(), - ); - } - - let parallelism = match input_parallelism { - // Explicit input is validated here too (not just at the command - // call sites) so the "validated when present" contract on - // `MintBehavioralDefaults.parallelism` is unskippable. - Some(count) if (1..=32).contains(&count) => Some(count), - Some(count) => { - return Err(format!( - "parallelism {count} is out of range (must be between 1 and 32)" - )) - } - None => match definition.and_then(|d| d.parallelism) { - Some(count) if (1..=32).contains(&count) => Some(count), - Some(count) => { - return Err(format!( - "parallelism {count} on the linked agent definition is out of range (must be between 1 and 32)" - )) - } - None => None, - }, - }; - - Ok(MintBehavioralDefaults { - respond_to, - respond_to_allowlist, - parallelism, - }) -} +mod behavioral_defaults; +pub use behavioral_defaults::*; mod catalog_source; pub use catalog_source::CatalogSource; diff --git a/desktop/src-tauri/src/managed_agents/types/behavioral_defaults.rs b/desktop/src-tauri/src/managed_agents/types/behavioral_defaults.rs new file mode 100644 index 0000000000..3cc40d471d --- /dev/null +++ b/desktop/src-tauri/src/managed_agents/types/behavioral_defaults.rs @@ -0,0 +1,105 @@ +use super::{AgentDefinition, RespondTo}; + +/// Empty input is allowed here — the boundary check (allowlist mode requires +/// at least one entry) is the caller's job, because an `UpdateManagedAgentRequest` +/// may want to validate a list without yet knowing the final mode. +pub fn validate_respond_to_allowlist(input: &[String]) -> Result, String> { + let mut seen = std::collections::HashSet::new(); + let mut out = Vec::with_capacity(input.len()); + for entry in input { + let trimmed = entry.trim(); + if trimmed.len() != 64 || !trimmed.chars().all(|c| c.is_ascii_hexdigit()) { + return Err(format!( + "invalid pubkey in respond-to allowlist: '{trimmed}' (must be 64 hex chars)" + )); + } + let lower = trimmed.to_ascii_lowercase(); + if seen.insert(lower.clone()) { + out.push(lower); + } + } + Ok(out) +} + +/// The behavioral fields resolved for a new instance at mint time. +#[derive(Debug, PartialEq, Eq)] +pub struct MintBehavioralDefaults { + pub respond_to: RespondTo, + pub respond_to_allowlist: Vec, + /// Validated (1..=32) when present; caller applies its own default. + pub parallelism: Option, +} + +/// Resolve the NIP-AP behavioral quad for a new instance: explicit input +/// wins, then the linked definition's defaults, then client defaults. +/// +/// This is the ONLY place definition behavioral strings are parsed — an +/// unrecognized `respond_to` mode or out-of-range `parallelism` on a +/// definition fails the mint loudly instead of silently substituting a +/// default the definition author did not choose. The empty-allowlist guard +/// fires here too, because inbound definitions bypass the dialog entirely. +/// +/// `input_allowlist` must already be normalized via +/// [`validate_respond_to_allowlist`]; the definition's allowlist is +/// validated here since it arrives from the wire. +pub fn resolve_mint_behavioral_defaults( + input_respond_to: Option, + input_allowlist: Vec, + input_parallelism: Option, + definition: Option<&AgentDefinition>, +) -> Result { + let (respond_to, respond_to_allowlist) = match input_respond_to { + // Explicit instance-level choice: the definition default is ignored + // wholesale (mode AND list travel together). + Some(mode) => (mode, input_allowlist), + None => match definition.and_then(|d| d.respond_to.as_deref()) { + Some(wire) => { + let mode = RespondTo::parse_wire(wire)?; + let list = if input_allowlist.is_empty() { + validate_respond_to_allowlist( + definition + .map(|d| d.respond_to_allowlist.as_slice()) + .unwrap_or(&[]), + ) + .map_err(|e| format!("definition respond-to allowlist is invalid: {e}"))? + } else { + input_allowlist + }; + (mode, list) + } + None => (RespondTo::default(), input_allowlist), + }, + }; + if respond_to == RespondTo::Allowlist && respond_to_allowlist.is_empty() { + return Err( + "respond-to mode 'allowlist' requires at least one pubkey in the allowlist".to_string(), + ); + } + + let parallelism = match input_parallelism { + // Explicit input is validated here too (not just at the command + // call sites) so the "validated when present" contract on + // `MintBehavioralDefaults.parallelism` is unskippable. + Some(count) if (1..=32).contains(&count) => Some(count), + Some(count) => { + return Err(format!( + "parallelism {count} is out of range (must be between 1 and 32)" + )) + } + None => match definition.and_then(|d| d.parallelism) { + Some(count) if (1..=32).contains(&count) => Some(count), + Some(count) => { + return Err(format!( + "parallelism {count} on the linked agent definition is out of range (must be between 1 and 32)" + )) + } + None => None, + }, + }; + + Ok(MintBehavioralDefaults { + respond_to, + respond_to_allowlist, + parallelism, + }) +} diff --git a/desktop/src/features/agents/lib/personaCatalogRelay.test.mjs b/desktop/src/features/agents/lib/personaCatalogRelay.test.mjs index fbaf1f5274..f3727598c0 100644 --- a/desktop/src/features/agents/lib/personaCatalogRelay.test.mjs +++ b/desktop/src/features/agents/lib/personaCatalogRelay.test.mjs @@ -356,7 +356,7 @@ test("test_foreign_entry_with_no_local_copy_stays_unselected", () => { BOB, ); - assert.equal(personas[0].id, "catalog:" + ALICE + ":reviewer"); + assert.equal(personas[0].id, `catalog:${ALICE}:reviewer`); assert.equal(personas[0].isActive, false); }); @@ -377,7 +377,7 @@ test("test_catalog_source_match_is_scoped_to_the_publishing_owner", () => { ALICE, ); - assert.equal(personas[0].id, "catalog:" + BOB + ":reviewer"); + assert.equal(personas[0].id, `catalog:${BOB}:reviewer`); assert.equal(personas[0].isActive, false); });