From cf30ff47d395b5f85ebe90388a5a7f71bf670f8e Mon Sep 17 00:00:00 2001 From: npub1mn7jgtj4w2pd0g0zeuhxsa6jy6p0rewxz4kujt98my82ahfmp72sxjexk7 Date: Thu, 30 Jul 2026 18:02:30 -0400 Subject: [PATCH] fix(desktop): gate agent config publishes behind a boot resolution pass MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Boot reconcile re-signs whatever is on disk at `monotonic_created_at = max(now, head + 1)`, so a merely-stale store mints events strictly newer than the edits they revert. Last-write-wins then adopts the revert. A second install with a fresh retention database is exactly this state, which is why a dev build silently reverts config edited elsewhere. `created_at` cannot arbitrate it: the value the stale store carries is manufactured by the reconcile, not observed. The barrier arbitrates on PROVENANCE instead. A baseline records what this install last agreed was published at a coordinate; with no baseline, the install cannot show it ever agreed to anything there, so neither its content nor its timestamps are evidence of intent about the relay's state. `decide` runs four ordered gates: positive tombstone evidence, then the no-baseline park, then the queued-deletion timestamp compare, then the queued edit against the baseline. The ordering is load-bearing — the timestamp compare is sound only because the no-baseline gate already excluded every store that would win it spuriously. Both publication gates are enforced, because a coordinate's record and its queued tombstone occupy different primary keys and the flush loop reads them independently: gating one row computes a suppression and enforces half of it. The live gate is default-deny over the decision set, so a variant added later is withheld until deliberately admitted. The scan is per-owner and unfiltered by `#a` because the relay post-filters `#a` after its `LIMIT`, so absence is claimed only from a page shorter than the limit it asked for. Nothing is applied to disk here. `ApplyHead`, `RestoreFromRelay`, and `DeleteLocal` are decided, logged, and gated; resolving them is the user-driven step. On relay deployments with a replica-routed read pool, the head lookups and tombstone scan read best-available rather than writer-consistent; staleness within replication lag is a documented known limitation (see `head_lookup` and `tombstone_scan` module docs). The barrier's per-decision tracing line is the detection mechanism if it fires. Co-authored-by: Will Pfleger Signed-off-by: Will Pfleger --- desktop/src-tauri/src/event_sync.rs | 16 + .../managed_agents/canonical_projection.rs | 98 ++ .../canonical_projection/tests.rs | 199 ++++ .../src/managed_agents/config_barrier.rs | 284 ++++++ .../managed_agents/config_barrier/tests.rs | 199 ++++ .../src/managed_agents/config_sync.rs | 372 +++++++ .../src/managed_agents/config_sync/tests.rs | 950 ++++++++++++++++++ .../src-tauri/src/managed_agents/decision.rs | 377 +++++++ .../src/managed_agents/decision/tests.rs | 554 ++++++++++ .../src/managed_agents/head_lookup.rs | 140 +++ .../src/managed_agents/head_lookup/tests.rs | 144 +++ desktop/src-tauri/src/managed_agents/mod.rs | 6 + .../src-tauri/src/managed_agents/retention.rs | 197 +++- .../src/managed_agents/retention/schema.rs | 7 +- .../src/managed_agents/tombstone_scan.rs | 234 +++++ .../managed_agents/tombstone_scan/tests.rs | 224 +++++ 16 files changed, 3999 insertions(+), 2 deletions(-) create mode 100644 desktop/src-tauri/src/managed_agents/canonical_projection.rs create mode 100644 desktop/src-tauri/src/managed_agents/canonical_projection/tests.rs create mode 100644 desktop/src-tauri/src/managed_agents/config_barrier.rs create mode 100644 desktop/src-tauri/src/managed_agents/config_barrier/tests.rs create mode 100644 desktop/src-tauri/src/managed_agents/config_sync.rs create mode 100644 desktop/src-tauri/src/managed_agents/config_sync/tests.rs create mode 100644 desktop/src-tauri/src/managed_agents/decision.rs create mode 100644 desktop/src-tauri/src/managed_agents/decision/tests.rs create mode 100644 desktop/src-tauri/src/managed_agents/head_lookup.rs create mode 100644 desktop/src-tauri/src/managed_agents/head_lookup/tests.rs create mode 100644 desktop/src-tauri/src/managed_agents/tombstone_scan.rs create mode 100644 desktop/src-tauri/src/managed_agents/tombstone_scan/tests.rs diff --git a/desktop/src-tauri/src/event_sync.rs b/desktop/src-tauri/src/event_sync.rs index 6526fc7d70..d05701b1c3 100644 --- a/desktop/src-tauri/src/event_sync.rs +++ b/desktop/src-tauri/src/event_sync.rs @@ -25,19 +25,35 @@ pub fn run_event_sync(app: &tauri::AppHandle, owner_keys: &nostr::Keys, db_path: /// `AppState::keys` mutex. The reconcile itself is still synchronous JSON, /// SQLite, and signing work, so it runs on the blocking pool rather than an /// async worker. +/// +/// # Ordering: reconcile, then barrier, and both before the flush publishes +/// +/// The reconcile marks changed coordinates `pending_sync = 1`; the boot barrier +/// ([`crate::managed_agents::config_barrier::run_boot_barrier`]) then decides +/// which of those may actually go out and gates the rest. Running the barrier +/// after the reconcile is what lets it see the rows the reconcile just queued — +/// the stale-store republish it exists to suppress is created right here. The +/// flush loop's first tick is 30s of relay lookups away, so in practice the gate +/// is set before anything publishes; if a flush did win the race, the barrier +/// still gates every later sweep. pub fn spawn_event_sync( app: tauri::AppHandle, owner_keys: nostr::Keys, db_path: std::path::PathBuf, ) { tauri::async_runtime::spawn(async move { + let barrier_app = app.clone(); if let Err(e) = tauri::async_runtime::spawn_blocking(move || { run_event_sync(&app, &owner_keys, &db_path); }) .await { eprintln!("buzz-desktop: event-sync: spawn_blocking failed: {e}"); + // The barrier still runs: a failed reconcile leaves whatever rows + // a previous boot queued, and gating those is exactly the case the + // barrier exists for. } + crate::managed_agents::config_barrier::run_boot_barrier(&barrier_app).await; }); } diff --git a/desktop/src-tauri/src/managed_agents/canonical_projection.rs b/desktop/src-tauri/src/managed_agents/canonical_projection.rs new file mode 100644 index 0000000000..45eeecbf9b --- /dev/null +++ b/desktop/src-tauri/src/managed_agents/canonical_projection.rs @@ -0,0 +1,98 @@ +//! The canonical published projection of a config record (V3.1). +//! +//! Every comparison the boot decision pass makes — disk against baseline, disk +//! against the relay head, baseline against what a publish would emit — runs on +//! this projection, never on whole JSON and never on content alone. +//! +//! Two reasons it cannot be either of those: +//! +//! - **Whole JSON is too wide.** Records carry install-local fields the relay +//! never sees (`source_dir`, `is_symlink`, `version`, timestamps, and for +//! managed agents `private_key_nsec`, `auth_tag`, `env_vars`). Comparing them +//! would report a difference for two records that publish identically, and +//! the pass would "repair" a divergence that does not exist. +//! - **Content alone is too narrow.** A persona's `shared` state rides in a +//! relay-significant *tag*, not in the content body. Two personas with equal +//! content but different `shared` tags publish as different events, so a +//! content-only compare would call them equal and skip a real publish. +//! +//! The projection is therefore exactly what publishing emits: the event content +//! plus the relay-significant tags, in a form that compares by value. + +use buzz_core_pkg::kind::{KIND_MANAGED_AGENT, KIND_PERSONA, KIND_TEAM}; + +use super::{ + agent_events::agent_event_content, persona_events::persona_event_content, + team_events::team_event_content, AgentDefinition, ManagedAgentRecord, TeamRecord, +}; + +/// What one coordinate publishes: the event content and the relay-significant +/// tags that travel with it. +/// +/// Equality is the decision-table's notion of "same": two projections are equal +/// exactly when publishing both would put the same thing on the relay. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct CanonicalProjection { + /// The serialized event content, byte-identical to what the builder emits. + pub content: String, + /// Whether the persona carries `["shared", "true"]`. Always `false` for + /// teams and managed agents, whose builders emit no `shared` tag. + pub shared: bool, +} + +impl CanonicalProjection { + /// The projection a persona (kind:30175) would publish. + /// + /// `shared` is read from the record rather than inferred, because it is the + /// field the builder turns into the tag. + pub fn for_persona(record: &AgentDefinition) -> Result { + Ok(Self { + content: serde_json::to_string(&persona_event_content(record)) + .map_err(|e| format!("failed to serialize persona projection: {e}"))?, + shared: record.shared, + }) + } + + /// The projection a team (kind:30176) would publish. + pub fn for_team(record: &TeamRecord) -> Result { + Ok(Self { + content: serde_json::to_string(&team_event_content(record)) + .map_err(|e| format!("failed to serialize team projection: {e}"))?, + shared: false, + }) + } + + /// The projection a managed agent (kind:30177) would publish. + /// + /// `agent_event_content` is the opt-IN no-secrets allowlist, so this can + /// never carry a private key, auth tag, or env var. + pub fn for_managed_agent(record: &ManagedAgentRecord) -> Result { + Ok(Self { + content: serde_json::to_string(&agent_event_content(record)) + .map_err(|e| format!("failed to serialize managed-agent projection: {e}"))?, + shared: false, + }) + } + + /// The projection an already-signed event carries. + /// + /// This is the head side of a disk-vs-head compare. `shared` is read with + /// the same fail-closed helper the relay's read-authorization uses, so a + /// malformed tag shape is treated as not-shared on both sides of the + /// comparison rather than only one. + pub fn from_event(event: &nostr::Event) -> Self { + let kind = event.kind.as_u16() as u32; + Self { + content: event.content.to_string(), + shared: kind == KIND_PERSONA && buzz_core_pkg::kind::event_is_shared(event), + } + } + + /// Whether `kind` is a config kind this projection knows how to build. + pub fn is_config_kind(kind: u32) -> bool { + matches!(kind, KIND_PERSONA | KIND_TEAM | KIND_MANAGED_AGENT) + } +} + +#[cfg(test)] +mod tests; diff --git a/desktop/src-tauri/src/managed_agents/canonical_projection/tests.rs b/desktop/src-tauri/src/managed_agents/canonical_projection/tests.rs new file mode 100644 index 0000000000..2e6af9b033 --- /dev/null +++ b/desktop/src-tauri/src/managed_agents/canonical_projection/tests.rs @@ -0,0 +1,199 @@ +use std::collections::BTreeMap; +use std::path::PathBuf; + +use super::*; +use crate::managed_agents::persona_events::build_persona_event; + +fn sample_persona() -> AgentDefinition { + AgentDefinition { + id: "test-persona".to_string(), + display_name: "Test Persona".to_string(), + avatar_url: Some("https://example.com/avatar.png".to_string()), + system_prompt: "You are a test assistant.".to_string(), + runtime: Some("goose".to_string()), + model: Some("claude-opus-4".to_string()), + provider: Some("anthropic".to_string()), + name_pool: vec!["Alpha".to_string()], + is_builtin: false, + is_active: true, + shared: false, + source_team: None, + source_team_persona_slug: Some("test-slug".to_string()), + catalog_source: None, + env_vars: BTreeMap::from([("KEY".to_string(), "value".to_string())]), + respond_to: None, + respond_to_allowlist: Vec::new(), + parallelism: None, + created_at: "2025-01-01T00:00:00Z".to_string(), + updated_at: "2025-01-01T00:00:00Z".to_string(), + } +} + +fn sample_team() -> TeamRecord { + TeamRecord { + id: "team-123".to_string(), + name: "Test Team".to_string(), + description: Some("A test team".to_string()), + instructions: Some("Coordinate carefully.".to_string()), + persona_ids: vec!["p1".to_string()], + is_builtin: false, + source_dir: Some(PathBuf::from("/local/only/path")), + is_symlink: true, + symlink_target: Some("/somewhere".to_string()), + version: Some("1.0".to_string()), + created_at: "2025-01-01T00:00:00Z".to_string(), + updated_at: "2025-01-01T00:00:00Z".to_string(), + } +} + +/// The whole point of the projection: a field the relay never sees must not +/// register as a difference, or the decision pass "repairs" a divergence that +/// does not exist and republishes on every boot. +#[test] +fn local_only_persona_fields_do_not_change_the_projection() { + let base = sample_persona(); + let local_edit = AgentDefinition { + // None of these are in the published content. + is_active: !base.is_active, + source_team: Some("some-team".to_string()), + env_vars: BTreeMap::from([("OTHER".to_string(), "different".to_string())]), + created_at: "2099-01-01T00:00:00Z".to_string(), + updated_at: "2099-01-01T00:00:00Z".to_string(), + ..base.clone() + }; + + assert_eq!( + CanonicalProjection::for_persona(&base).unwrap(), + CanonicalProjection::for_persona(&local_edit).unwrap() + ); +} + +/// A published field must register, or a real edit is silently never published. +#[test] +fn published_persona_fields_change_the_projection() { + let base = sample_persona(); + for edited in [ + AgentDefinition { + system_prompt: "Different prompt.".to_string(), + ..base.clone() + }, + AgentDefinition { + model: Some("other-model".to_string()), + ..base.clone() + }, + AgentDefinition { + provider: Some("other-provider".to_string()), + ..base.clone() + }, + AgentDefinition { + display_name: "Renamed".to_string(), + ..base.clone() + }, + ] { + assert_ne!( + CanonicalProjection::for_persona(&base).unwrap(), + CanonicalProjection::for_persona(&edited).unwrap(), + "a published field edit must be visible to the decision pass" + ); + } +} + +/// `shared` rides in a relay-significant TAG, not the content body. A +/// content-only compare would call these two equal and skip a real publish — +/// this is the case that makes "content alone" wrong. +#[test] +fn shared_tag_difference_is_visible_even_when_content_matches() { + let private = sample_persona(); + let shared = AgentDefinition { + shared: true, + ..private.clone() + }; + + let private_projection = CanonicalProjection::for_persona(&private).unwrap(); + let shared_projection = CanonicalProjection::for_persona(&shared).unwrap(); + + assert_eq!( + private_projection.content, shared_projection.content, + "precondition: the content bodies are identical" + ); + assert_ne!( + private_projection, shared_projection, + "the shared tag must still make the projections differ" + ); +} + +/// Disk-vs-head is only meaningful if both sides project identically for a +/// record that was published unchanged. +#[test] +fn disk_projection_round_trips_through_a_signed_event() { + let keys = nostr::Keys::generate(); + for shared in [false, true] { + let record = AgentDefinition { + shared, + ..sample_persona() + }; + let event = build_persona_event(&record) + .unwrap() + .sign_with_keys(&keys) + .unwrap(); + + assert_eq!( + CanonicalProjection::for_persona(&record).unwrap(), + CanonicalProjection::from_event(&event), + "a freshly published record must project equal to its own event" + ); + } +} + +/// Teams and managed agents emit no `shared` tag, so their projections must +/// never claim one — otherwise a team would compare unequal to its own event. +#[test] +fn non_persona_kinds_are_never_shared() { + assert!( + !CanonicalProjection::for_team(&sample_team()) + .unwrap() + .shared + ); + + let keys = nostr::Keys::generate(); + let event = crate::managed_agents::team_events::build_team_event(&sample_team()) + .unwrap() + .sign_with_keys(&keys) + .unwrap(); + assert_eq!( + CanonicalProjection::for_team(&sample_team()).unwrap(), + CanonicalProjection::from_event(&event) + ); +} + +/// A team's install-local fields (`source_dir`, `is_symlink`, `version`) are +/// exactly the ones V3.10 cites as unrecoverable from a relay projection, so +/// they must not drive a publish decision either. +#[test] +fn local_only_team_fields_do_not_change_the_projection() { + let base = sample_team(); + let local_edit = TeamRecord { + source_dir: Some(PathBuf::from("/a/completely/different/path")), + is_symlink: !base.is_symlink, + symlink_target: None, + version: Some("9.9".to_string()), + updated_at: "2099-01-01T00:00:00Z".to_string(), + ..base.clone() + }; + + assert_eq!( + CanonicalProjection::for_team(&base).unwrap(), + CanonicalProjection::for_team(&local_edit).unwrap() + ); +} + +#[test] +fn config_kinds_are_recognized_and_others_are_not() { + for kind in [KIND_PERSONA, KIND_TEAM, KIND_MANAGED_AGENT] { + assert!(CanonicalProjection::is_config_kind(kind)); + } + // kind:5 deletions and chat messages are not config coordinates. + for kind in [1u32, 5, 30178] { + assert!(!CanonicalProjection::is_config_kind(kind)); + } +} diff --git a/desktop/src-tauri/src/managed_agents/config_barrier.rs b/desktop/src-tauri/src/managed_agents/config_barrier.rs new file mode 100644 index 0000000000..7c07bfdde2 --- /dev/null +++ b/desktop/src-tauri/src/managed_agents/config_barrier.rs @@ -0,0 +1,284 @@ +//! The boot config barrier: gather every coordinate, decide, enforce. +//! +//! This is the entry point that makes [`super::decision::decide`] mean +//! something. It runs once per boot, before the flush loop can publish, and +//! wires the three layers together in the order V3.4 requires: +//! +//! 1. enumerate the coordinates this device knows about, from disk; +//! 2. run every relay lookup with NO store lock held; +//! 3. take `managed_agents_store_lock` once, re-check the scope, gather the +//! local half, decide, and enforce — all under that single guard. +//! +//! # Why the barrier exists +//! +//! Without it, boot re-signs local disk content at +//! `monotonic_created_at = max(now, head + 1)` and queues it for publish. A +//! store that is merely stale therefore mints events that are strictly newer +//! than the edits they revert, and the relay's last-write-wins resolution +//! adopts the revert. The barrier's job is to run BEFORE that publish and +//! decide, per coordinate, whether the local content is an edit worth +//! publishing or a stale copy that must be suppressed. +//! +//! # What it enforces, and what it does not +//! +//! It enforces exactly one thing: the publication gate +//! ([`super::config_sync::apply_gate`]). A coordinate whose decision blocks +//! publication is withheld from [`super::retention::get_pending_sync`] and +//! cannot reach the relay on this boot. Decisions that PATCH disk +//! (`ApplyHead`, `RestoreFromRelay`, `DeleteLocal`) are logged and gated but +//! not applied here — applying them is the resolution path (V3.6), which is +//! user-driven by design. Suppression is the half that stops the data loss; +//! adopting remote content is the half that needs a human. + +use std::collections::BTreeSet; + +use tauri::Manager; + +use super::{ + canonical_projection::CanonicalProjection, + config_sync::{local_state, run_decision_pass, CoordinateState}, + decision::HeadState, + head_lookup::{fetch_head, Coordinate}, + retention::{active_retention_scope, open_retention_db, pending_coordinates, RetentionScope}, + tombstone_scan::scan_tombstones, +}; +use crate::app_state::AppState; +use buzz_core_pkg::kind::{KIND_MANAGED_AGENT, KIND_PERSONA, KIND_TEAM}; + +/// Run the boot barrier for the active scope, best-effort. +/// +/// Failures are logged and swallowed: a barrier that cannot run must not block +/// boot, and its absence is not itself dangerous — it leaves the pre-barrier +/// behaviour, which is what every prior release did. +pub async fn run_boot_barrier(app: &tauri::AppHandle) { + let state = app.state::(); + let scope = match active_retention_scope(app, &state) { + Ok(scope) => scope, + Err(error) => { + tracing::warn!(target: "buzz::config_sync", %error, "boot barrier skipped: no active scope"); + return; + } + }; + + if let Err(error) = run_boot_barrier_for_scope(app, &state, &scope).await { + tracing::warn!(target: "buzz::config_sync", %error, "boot barrier failed"); + } +} + +/// Withhold every queued publish that has no baseline, before any network I/O. +/// +/// This is V3.9's legacy-pending quarantine, and it is also what makes the +/// barrier safe against its own latency. The barrier's decisions need a +/// per-coordinate relay lookup plus a full tombstone scan; the flush loop needs +/// only a timer. On a slow network the flush can therefore win the race and +/// publish exactly the rows the barrier was about to gate. +/// +/// Quarantining first inverts that: the gate is closed synchronously, before +/// the first round trip, and the decision pass re-opens it for the coordinates +/// that earn it. A row is either provably publishable or it is not going out. +/// +/// Scoped to rows with no baseline because those are precisely the unprovable +/// ones: a legacy row queued before the baseline column existed, and a row +/// queued by a second store whose retention database has never agreed to +/// anything at that coordinate. A row WITH a baseline is left alone — its +/// arbitration is decidable, and closing its gate here would stall an ordinary +/// edit behind a scan it does not need. +/// +/// Returns how many rows it withheld. +fn quarantine_unprovable_pending( + conn: &rusqlite::Connection, + owner_pubkey: &str, +) -> Result { + conn.execute( + "UPDATE persona_events SET publish_blocked = 1 + WHERE pending_sync = 1 AND pubkey = ?1 AND baseline_event_id IS NULL", + rusqlite::params![owner_pubkey], + ) + .map_err(|error| format!("failed to quarantine unprovable pending rows: {error}")) +} + +async fn run_boot_barrier_for_scope( + app: &tauri::AppHandle, + state: &AppState, + scope: &RetentionScope, +) -> Result<(), String> { + let owner_pubkey = scope.owner_keys.public_key().to_hex(); + + // Enumerate BEFORE the lookups: the set of coordinates to decide is the + // union of what disk holds and what is already queued for publish. A + // pending row can outlive its record (deleted persona, removed agent, or a + // tombstone), and those rows publish with nothing on disk to enumerate + // them — exactly the ones that most need gating. The gate must see + // everything the publisher can see, and the publisher reads the table. + let mut disk = disk_projections(app)?; + { + let conn = open_retention_db(&scope.db_path)?; + let quarantined = quarantine_unprovable_pending(&conn, &owner_pubkey)?; + if quarantined > 0 { + tracing::info!( + target: "buzz::config_sync", + quarantined, + "withheld pending rows with no baseline pending the decision pass" + ); + } + + let known: BTreeSet<(u32, String)> = disk + .iter() + .map(|(coordinate, _)| (coordinate.kind, coordinate.d_tag.clone())) + .collect(); + for (kind, d_tag) in pending_coordinates(&conn, &owner_pubkey)? { + if !known.contains(&(kind, d_tag.clone())) { + // No disk record: the projection is `None`, which is what the + // table reads as "absent locally". + disk.push((Coordinate { kind, d_tag }, None)); + } + } + } + + // Phase 1 — relay lookups, no lock held. Each head read is an exact + // per-coordinate query; a failure becomes `LookupFailed` + // rather than an absence, so a flaky network can never be read as a + // deletion. The tombstone scan runs once for the whole owner. + let api_base_url = crate::relay::relay_http_base_url(&scope.relay_url); + let mut heads: Vec<(Coordinate, HeadState)> = Vec::with_capacity(disk.len()); + for (coordinate, _) in &disk { + heads.push(( + coordinate.clone(), + fetch_head(state, scope, &api_base_url, coordinate).await, + )); + } + let tombstones = scan_tombstones(state, scope, &api_base_url).await; + + // Phase 2 — one guard for the whole decision pass. + let _store_guard = state + .managed_agents_store_lock + .lock() + .map_err(|error| error.to_string())?; + + // Phase 3 — re-resolve the scope under the lock. `apply_workspace` can swap + // the relay and keys mid-flight without taking this lock, so without this + // check a completed pull from community A could patch community B's store. + let current = active_retention_scope(app, state)?; + if current.db_path != scope.db_path { + tracing::info!( + target: "buzz::config_sync", + "boot barrier abandoned: workspace switched during relay lookups" + ); + return Ok(()); + } + + let conn = open_retention_db(&scope.db_path)?; + let mut states: Vec = Vec::with_capacity(disk.len()); + for ((coordinate, projection), (_, head)) in disk.into_iter().zip(heads) { + let tombstone = tombstones.evidence(&coordinate); + let observation = local_state( + &conn, + &owner_pubkey, + &coordinate, + projection, + head, + tombstone, + )?; + states.push(CoordinateState { + coordinate, + observation, + }); + } + + run_decision_pass(&conn, &owner_pubkey, &states)?; + Ok(()) +} + +/// Read the three record stores and enumerate their coordinates. +fn disk_projections( + app: &tauri::AppHandle, +) -> Result)>, String> { + use super::{load_managed_agents, load_teams, personas::load_personas}; + + Ok(project_records( + &load_personas(app)?, + &load_teams(app)?, + &load_managed_agents(app)?, + )) +} + +/// Every config coordinate the given records occupy, with what each would +/// publish. +/// +/// Uses the same d-tag derivations the publish paths use, so a coordinate can +/// never be enumerated under a key that differs from the one it would publish +/// under. Built-in records are excluded: they come from code, are never +/// published, and have no relay head to compare against. +/// +/// A record whose projection cannot be built is skipped with a warning rather +/// than failing the pass — one malformed record must not stop the barrier from +/// protecting every other coordinate. +fn project_records( + personas: &[super::AgentDefinition], + teams: &[super::TeamRecord], + agents: &[super::ManagedAgentRecord], +) -> Vec<(Coordinate, Option)> { + let mut out: Vec<(Coordinate, Option)> = Vec::new(); + let mut seen: BTreeSet<(u32, String)> = BTreeSet::new(); + + let mut push = |kind: u32, d_tag: String, projection: Result| { + match projection { + Ok(projection) => { + // Two records can normalize onto one d-tag. Deciding the + // coordinate twice would gate it twice from the same inputs, + // so the first wins and the collision is visible in the log. + if seen.insert((kind, d_tag.clone())) { + out.push((Coordinate { kind, d_tag }, Some(projection))); + } else { + tracing::warn!( + target: "buzz::config_sync", + kind, + d_tag = %d_tag, + "duplicate coordinate on disk; deciding the first record only" + ); + } + } + Err(error) => { + tracing::warn!( + target: "buzz::config_sync", + kind, + d_tag = %d_tag, + %error, + "skipping coordinate with an unbuildable projection" + ); + } + } + }; + + for record in personas.iter().filter(|record| !record.is_builtin) { + push( + KIND_PERSONA, + super::persona_events::persona_d_tag(record), + CanonicalProjection::for_persona(record), + ); + } + for record in teams.iter().filter(|record| !record.is_builtin) { + push( + KIND_TEAM, + record.id.clone(), + CanonicalProjection::for_team(record), + ); + } + for record in agents + .iter() + // A key-less agent has no coordinate yet: its d-tag IS its pubkey, and + // it mints one on first start. + .filter(|record| !record.pubkey.is_empty()) + { + push( + KIND_MANAGED_AGENT, + record.pubkey.clone(), + CanonicalProjection::for_managed_agent(record), + ); + } + + out +} + +#[cfg(test)] +mod tests; diff --git a/desktop/src-tauri/src/managed_agents/config_barrier/tests.rs b/desktop/src-tauri/src/managed_agents/config_barrier/tests.rs new file mode 100644 index 0000000000..da49441dfa --- /dev/null +++ b/desktop/src-tauri/src/managed_agents/config_barrier/tests.rs @@ -0,0 +1,199 @@ +use std::collections::BTreeMap; +use std::path::PathBuf; + +use super::*; +use crate::managed_agents::{AgentDefinition, ManagedAgentRecord, TeamRecord}; + +fn persona(id: &str, slug: Option<&str>, builtin: bool) -> AgentDefinition { + AgentDefinition { + id: id.to_string(), + display_name: "Test".to_string(), + avatar_url: None, + system_prompt: "prompt".to_string(), + runtime: None, + model: None, + provider: None, + name_pool: Vec::new(), + is_builtin: builtin, + is_active: true, + shared: false, + source_team: None, + source_team_persona_slug: slug.map(str::to_string), + catalog_source: None, + env_vars: BTreeMap::new(), + respond_to: None, + respond_to_allowlist: Vec::new(), + parallelism: None, + created_at: "2026-01-01T00:00:00Z".to_string(), + updated_at: "2026-01-01T00:00:00Z".to_string(), + } +} + +fn team(id: &str, builtin: bool) -> TeamRecord { + TeamRecord { + id: id.to_string(), + name: "Team".to_string(), + description: None, + instructions: None, + persona_ids: Vec::new(), + is_builtin: builtin, + source_dir: Some(PathBuf::from("/local/only")), + is_symlink: false, + symlink_target: None, + version: None, + created_at: "2026-01-01T00:00:00Z".to_string(), + updated_at: "2026-01-01T00:00:00Z".to_string(), + } +} + +fn agent(pubkey: &str) -> ManagedAgentRecord { + serde_json::from_str(&format!( + r#"{{ + "pubkey": "{pubkey}", + "name": "Agent", + "relay_url": "wss://localhost:3000", + "acp_command": "buzz-acp", + "agent_command": "goose", + "agent_args": [], + "mcp_command": "", + "turn_timeout_seconds": 320, + "system_prompt": "You are a test agent.", + "created_at": "2026-01-01T00:00:00Z", + "updated_at": "2026-01-01T00:00:00Z", + "last_started_at": null, + "last_stopped_at": null, + "last_exit_code": null, + "last_error": null + }}"# + )) + .unwrap() +} + +fn coordinates(records: &[(Coordinate, Option)]) -> Vec<(u32, String)> { + records + .iter() + .map(|(coordinate, _)| (coordinate.kind, coordinate.d_tag.clone())) + .collect() +} + +/// All three config kinds must be enumerated. A kind the barrier does not +/// enumerate is a kind it can never suppress, so the revert bug would survive +/// for exactly those records. +#[test] +fn test_all_three_config_kinds_are_enumerated() { + let projected = project_records( + &[persona("p1", None, false)], + &[team("t1", false)], + &[agent("agentpubkey")], + ); + + assert_eq!( + coordinates(&projected), + vec![ + (KIND_PERSONA, "p1".to_string()), + (KIND_TEAM, "t1".to_string()), + (KIND_MANAGED_AGENT, "agentpubkey".to_string()), + ] + ); +} + +/// Built-ins ship in the binary and are never published, so a head lookup for +/// one would always report `Absent` and gate a coordinate that has nothing to +/// gate. +#[test] +fn test_builtin_records_are_not_enumerated() { + let projected = project_records( + &[persona("builtin", None, true), persona("real", None, false)], + &[team("builtin-team", true), team("real-team", false)], + &[], + ); + + assert_eq!( + coordinates(&projected), + vec![ + (KIND_PERSONA, "real".to_string()), + (KIND_TEAM, "real-team".to_string()), + ] + ); +} + +/// A key-less agent has no coordinate at all — its d-tag IS its pubkey, minted +/// on first start. Enumerating it would look up `{"#d": [""]}`. +#[test] +fn test_keyless_agents_are_not_enumerated() { + let projected = project_records(&[], &[], &[agent(""), agent("haskey")]); + + assert_eq!( + coordinates(&projected), + vec![(KIND_MANAGED_AGENT, "haskey".to_string())] + ); +} + +/// The enumerated d-tag must be the one the publish path would use, or the +/// barrier looks up a head for a coordinate this device never writes to and +/// reads a spurious absence. `persona_d_tag` prefers the team slug and +/// normalizes it. +#[test] +fn test_persona_coordinate_uses_the_published_d_tag_derivation() { + let projected = project_records(&[persona("uuid-id", Some("CodeReviewer"), false)], &[], &[]); + + assert_eq!( + coordinates(&projected), + vec![(KIND_PERSONA, "codereviewer".to_string())] + ); +} + +/// Two personas can normalize onto one d-tag. Deciding it twice would gate the +/// same coordinate from the same inputs twice; the second pass is noise. +#[test] +fn test_duplicate_coordinates_are_decided_once() { + let projected = project_records( + &[ + persona("first", Some("Shared"), false), + persona("second", Some("shared"), false), + ], + &[], + &[], + ); + + assert_eq!( + coordinates(&projected), + vec![(KIND_PERSONA, "shared".to_string())] + ); +} + +/// A d-tag collision ACROSS kinds is not a duplicate: `30175:x` and `30176:x` +/// are different coordinates with different heads, and collapsing them would +/// let a team's head answer a persona's lookup. +#[test] +fn test_same_d_tag_on_different_kinds_are_separate_coordinates() { + let projected = project_records(&[persona("same", None, false)], &[team("same", false)], &[]); + + assert_eq!( + coordinates(&projected), + vec![ + (KIND_PERSONA, "same".to_string()), + (KIND_TEAM, "same".to_string()), + ] + ); +} + +/// The projection carried into the decision must be what publishing emits, or +/// every disk-vs-head compare is against the wrong value. +#[test] +fn test_projection_matches_what_publishing_would_emit() { + let record = persona("p1", None, false); + let projected = project_records(std::slice::from_ref(&record), &[], &[]); + + assert_eq!( + projected[0].1, + Some(CanonicalProjection::for_persona(&record).unwrap()) + ); +} + +/// Empty stores are the fresh-install case: no coordinates, no lookups, no +/// gate writes. +#[test] +fn test_no_records_yields_no_coordinates() { + assert!(project_records(&[], &[], &[]).is_empty()); +} diff --git a/desktop/src-tauri/src/managed_agents/config_sync.rs b/desktop/src-tauri/src/managed_agents/config_sync.rs new file mode 100644 index 0000000000..49876609b8 --- /dev/null +++ b/desktop/src-tauri/src/managed_agents/config_sync.rs @@ -0,0 +1,372 @@ +//! Boot config sync: gather each coordinate's observable state, decide, enforce. +//! +//! This is the layer between the pure decision table +//! ([`super::decision::decide`]) and the stores it reads. It owns two things +//! the table deliberately does not: WHERE each input comes from, and the +//! ordering guarantees around reading them. +//! +//! # Ordering (V3.4) +//! +//! No network I/O may happen while the store lock is held, and no decision may +//! be made from inputs gathered under a different workspace: +//! +//! 1. capture `(relay, owner, db path)` as a scope snapshot; +//! 2. run the relay lookups with NO lock held; +//! 3. take `managed_agents_store_lock` for the whole decision pass; +//! 4. re-resolve the active scope and abandon the pass if it moved; +//! 5. evaluate the table and write, still under that one guard. +//! +//! Step 4 is load-bearing rather than defensive: `apply_workspace` can swap the +//! relay and keys mid-flight and does not take this lock, so without the +//! recheck a completed pull from community A could patch community B's shared +//! disk store. + +use super::{ + canonical_projection::CanonicalProjection, + decision::{Decision, Head, HeadState, Observation, Resolution, TombstoneEvidence}, + head_lookup::Coordinate, + retention::{get_baseline, get_retained_event, RetainedEvent}, +}; +use rusqlite::Connection; + +/// Everything the decision table needs about one coordinate, plus the identity +/// of the coordinate itself. +#[derive(Debug, Clone)] +pub struct CoordinateState { + pub coordinate: Coordinate, + pub observation: Observation, +} + +/// Read the locally-known half of a coordinate's state: the queued publish, any +/// queued deletion, and the baseline. +/// +/// The head and the tombstone evidence come from the relay and are supplied by +/// the caller, which has already run those lookups outside the lock. +/// +/// The coordinate's own row and its paired tombstone row occupy different +/// primary keys and are reported separately, because the table resolves them on +/// different evidence — a deletion by timestamp against the head, an edit +/// against the baseline. Collapsing them into one flag was the cross-coordinate +/// hole called out in review; reporting only the live row is the same hole. +pub fn local_state( + conn: &Connection, + owner_pubkey: &str, + coordinate: &Coordinate, + disk: Option, + head: HeadState, + tombstone: TombstoneEvidence, +) -> Result { + let row = get_retained_event(conn, coordinate.kind, owner_pubkey, &coordinate.d_tag)?; + let tombstone_row = get_retained_event( + conn, + buzz_core_pkg::kind::KIND_DELETION, + owner_pubkey, + &super::retention::tombstone_retention_d_tag(coordinate.kind, &coordinate.d_tag), + )?; + + Ok(Observation { + disk, + queued: queued_projection(row.as_ref()), + head, + baseline: baseline_projection(conn, owner_pubkey, coordinate, row.as_ref())?, + queued_deletion_at: tombstone_row + .as_ref() + .filter(|row| row.pending_sync) + .map(|row| row.created_at), + tombstone, + }) +} + +/// What a coordinate's queued row would publish, or `None` when nothing is +/// queued there. +/// +/// Built from the retained row's own `raw_event` rather than from disk: the +/// flush loop publishes exactly those bytes, so those are the bytes the +/// baseline arbitration has to be about. A row whose stored event will not +/// parse reports `None` — it cannot publish either, since the flush loop parses +/// the same field and fails the row. +fn queued_projection(row: Option<&RetainedEvent>) -> Option { + use nostr::JsonUtil; + let row = row.filter(|row| row.pending_sync)?; + let event = nostr::Event::from_json(&row.raw_event).ok()?; + Some(CanonicalProjection::from_event(&event)) +} + +/// The baseline projection for a coordinate, or `None` when this install has +/// no provenance for it. +/// +/// The stored baseline is the CONTENT that was published; the `shared` half of +/// the projection is recovered from the baseline event's own tags via the +/// retained row, so a baseline comparison sees the same two fields a head +/// comparison does. +fn baseline_projection( + conn: &Connection, + owner_pubkey: &str, + coordinate: &Coordinate, + row: Option<&RetainedEvent>, +) -> Result, String> { + let Some((baseline_event_id, baseline_content)) = + get_baseline(conn, coordinate.kind, owner_pubkey, &coordinate.d_tag)? + else { + return Ok(None); + }; + + // Recover `shared` from the retained event when the row still holds the + // baseline event; otherwise the tag state is unknown and the baseline is + // only usable for kinds that never carry one. + let shared = row + .filter(|row| row.event_id.as_deref() == Some(baseline_event_id.as_str())) + .and_then(|row| { + use nostr::JsonUtil; + nostr::Event::from_json(&row.raw_event).ok() + }) + .map(|event| CanonicalProjection::from_event(&event).shared) + .unwrap_or(false); + + Ok(Some(CanonicalProjection { + content: baseline_content, + shared, + })) +} + +/// Resolve every gathered coordinate, pairing each resolution with its target. +/// +/// Kept separate from execution so a caller can log or inspect the full plan +/// before anything is written — and so the mapping is testable end to end +/// without a relay or a disk store. +pub fn plan(states: &[CoordinateState]) -> Vec<(Coordinate, Resolution)> { + states + .iter() + .map(|state| { + ( + state.coordinate.clone(), + super::decision::decide(&state.observation), + ) + }) + .collect() +} + +/// Whether a resolution must withhold the coordinate's LIVE row from the flush +/// loop. +/// +/// The gate is DEFAULT-DENY over the decision set: it lists what may publish, +/// not what may not. Every publishing decision names itself here, so a variant +/// added later is withheld until someone deliberately admits it — the opposite +/// default would let a new decision reach the relay by omission. +/// +/// Clearing the gate is as load-bearing as setting it: a coordinate suppressed +/// on one boot must resume publishing once the condition that suppressed it is +/// gone, or the gate latches shut permanently and the record strands. +pub fn blocks_publication(resolution: &Resolution) -> bool { + match resolution.decision { + // The one decision that publishes local content. + Decision::PublishLocalEdit => false, + // The deletion publishes from the TOMBSTONE row, not this one. The + // live row is the thing being deleted, so it must stay withheld: + // publishing it would race its own tombstone at the same coordinate. + Decision::PublishDeletion => true, + // Adopts the relay's version onto disk, so nothing local is going out. + // Only reachable with a baseline present and nothing queued (gate D + // takes a queued row first), so releasing strands nothing. + Decision::ApplyHead => false, + // Only reachable with NO local record and nothing queued, so there is + // nothing to withhold — and nothing to strand by withholding. Blocked + // rather than released because it is the one adopt-the-relay cell + // reachable with no baseline, and a coordinate with no provenance + // against a live head must not leave an open gate behind for a row + // queued after the pass. + Decision::RestoreFromRelay => true, + // `StampBaseline` is easy to miss: disk already equals the head, so + // publishing would push byte-identical content and burn a + // `monotonic_created_at` bump for nothing. + Decision::StampBaseline + | Decision::SuppressPublish + | Decision::Park(_) + | Decision::DeleteLocal + | Decision::Defer => true, + } +} + +/// Whether a resolution must withhold the coordinate's queued TOMBSTONE row. +/// +/// A queued deletion does not live at the coordinate it deletes: it is retained +/// at `(5, owner, ":")`, a different primary key. +/// Gating only [`blocks_publication`]'s row would therefore compute a +/// suppression and enforce it on the wrong row, leaving the tombstone free to +/// publish — the exact "decorative gate" failure this pass exists to avoid, and +/// the worse direction of it: a deletion that escapes the gate destroys a head +/// rather than merely reverting it. +/// +/// Exactly one decision releases it. Every other outcome — including the +/// timestamp arbitration falling through to the live cells — means the queued +/// deletion did not win, so it stays withheld. +pub fn blocks_tombstone_publication(resolution: &Resolution) -> bool { + resolution.decision != Decision::PublishDeletion +} + +/// Log one resolution with the inputs that produced it, then enforce BOTH +/// publication gates. +/// +/// Two rows, not one: the coordinate's live row and its queued tombstone row +/// occupy different primary keys, and the flush loop reads them independently. +/// A single `set_publish_blocked` call therefore enforces half a decision. +/// +/// The log line is a hard requirement, not debug output: this pass silently +/// changes what a device publishes, and the boot log is the only evidence of +/// why a coordinate did or did not go out. Every field the table branched on is +/// recorded, so a surprising decision can be re-derived from the log alone +/// without reproducing the machine's store. +pub fn apply_gate( + conn: &Connection, + owner_pubkey: &str, + coordinate: &Coordinate, + observation: &Observation, + resolution: &Resolution, +) -> Result<(), String> { + tracing::info!( + target: "buzz::config_sync", + kind = coordinate.kind, + d_tag = %coordinate.d_tag, + decision = ?resolution.decision, + blocked = blocks_publication(resolution), + tombstone_blocked = blocks_tombstone_publication(resolution), + cleared_queued_publish = resolution.clear_queued_publish, + head = match &observation.head { + HeadState::Present(_) => "present", + HeadState::Absent => "absent", + HeadState::LookupFailed => "lookup_failed", + }, + disk_present = observation.disk.is_some(), + disk_matches_head = matches!( + (&observation.disk, &observation.head), + (Some(disk), HeadState::Present(head)) if head.projection == *disk + ), + has_baseline = observation.baseline.is_some(), + queued = observation.queued.is_some(), + queued_matches_baseline = matches!( + (&observation.queued, &observation.baseline), + (Some(queued), Some(baseline)) if queued == baseline + ), + queued_deletion = observation.queued_deletion_at.is_some(), + tombstone = ?observation.tombstone, + "config-sync decision" + ); + + super::retention::set_publish_blocked( + conn, + coordinate.kind, + owner_pubkey, + &coordinate.d_tag, + blocks_publication(resolution), + )?; + + super::retention::set_publish_blocked( + conn, + buzz_core_pkg::kind::KIND_DELETION, + owner_pubkey, + &super::retention::tombstone_retention_d_tag(coordinate.kind, &coordinate.d_tag), + blocks_tombstone_publication(resolution), + ) +} + +/// Record the relay head as this install's baseline for a coordinate. +/// +/// Only ever called with a head that a completed, writer-consistent lookup +/// actually returned, which is what makes the baseline a claim about the RELAY +/// rather than about local disk. Stamping from disk content would defeat the +/// entire model: the baseline's job is to distinguish "the user edited this" +/// from "this store never learned about the newer head", and content that was +/// never published cannot answer that question. +fn stamp_baseline_from_head( + conn: &Connection, + owner_pubkey: &str, + coordinate: &Coordinate, + head: &Head, +) -> Result<(), String> { + super::retention::set_baseline( + conn, + coordinate.kind, + owner_pubkey, + &coordinate.d_tag, + &head.event_id, + &head.projection.content, + ) +} + +/// Run the boot decision pass for a set of coordinates and enforce the result. +/// +/// This is the barrier's entry point. Ordering follows V3.4 strictly: the +/// caller performs every relay lookup BEFORE calling this, and this function +/// does only local work, so no network I/O ever happens while the store lock is +/// held. +/// +/// Three outcomes are executed here, and only three: the publication gate, the +/// baseline stamp, and dropping a vacuous queued publish. All three are +/// provenance — none touches the user's records. The decisions that PATCH disk +/// (`ApplyHead`, `RestoreFromRelay`, `DeleteLocal`) are logged and gated but +/// deliberately not applied, because adopting remote content over a local +/// record is the resolution path (V3.6) and is user-driven by design. +/// +/// Returns the plan it enforced, so the caller can act on the non-gate half of +/// each resolution and log a summary. +pub fn run_decision_pass( + conn: &Connection, + owner_pubkey: &str, + states: &[CoordinateState], +) -> Result, String> { + let plan = plan(states); + + for (state, (coordinate, resolution)) in states.iter().zip(plan.iter()) { + apply_gate( + conn, + owner_pubkey, + coordinate, + &state.observation, + resolution, + )?; + + // A vacuous queued publish is dropped rather than gated. Gating would + // hold it forever: it can never become publishable, because it says + // nothing the baseline does not already say. + if resolution.clear_queued_publish { + super::retention::clear_pending_sync( + conn, + coordinate.kind, + owner_pubkey, + &coordinate.d_tag, + )?; + } + + // `StampBaseline` means disk and head already agree, so the head is a + // sound baseline and recording it is what gives every later boot the + // provenance to tell an edit from a stale copy. Without this the + // baseline stays `None` forever and every baseline-dependent cell is + // unreachable. + if resolution.decision == Decision::StampBaseline { + if let HeadState::Present(head) = &state.observation.head { + stamp_baseline_from_head(conn, owner_pubkey, coordinate, head)?; + } + } + } + + let blocked = plan + .iter() + .filter(|(_, resolution)| blocks_publication(resolution)) + .count(); + let cleared = plan + .iter() + .filter(|(_, resolution)| resolution.clear_queued_publish) + .count(); + tracing::info!( + target: "buzz::config_sync", + coordinates = plan.len(), + blocked, + cleared, + "config-sync decision pass complete" + ); + + Ok(plan) +} + +#[cfg(test)] +mod tests; diff --git a/desktop/src-tauri/src/managed_agents/config_sync/tests.rs b/desktop/src-tauri/src/managed_agents/config_sync/tests.rs new file mode 100644 index 0000000000..c6d4766e7d --- /dev/null +++ b/desktop/src-tauri/src/managed_agents/config_sync/tests.rs @@ -0,0 +1,950 @@ +use std::path::Path; + +use super::*; +use crate::managed_agents::{ + decision::{decide, ParkReason}, + persona_events::build_persona_event, + retention::{ + get_baseline, get_pending_sync, get_retained_event, is_publish_blocked, open_retention_db, + retain_event, set_baseline, tombstone_retention_d_tag, RetainedEvent, + }, +}; +use buzz_core_pkg::kind::{KIND_DELETION, KIND_PERSONA}; + +const OWNER: &str = "ownerpubkeyhex"; + +fn test_db() -> Connection { + open_retention_db(Path::new(":memory:")).unwrap() +} + +fn coordinate() -> Coordinate { + Coordinate { + kind: KIND_PERSONA, + d_tag: "test-persona".to_string(), + } +} + +/// The retention key the coordinate's queued tombstone occupies — a DIFFERENT +/// primary key from the coordinate itself, which is the whole reason the gate +/// has two halves. +fn tombstone_key() -> String { + tombstone_retention_d_tag(KIND_PERSONA, "test-persona") +} + +fn projection(content: &str) -> CanonicalProjection { + CanonicalProjection { + content: content.to_string(), + shared: false, + } +} + +/// A relay head carrying `content` at `created_at`, with an id derived from it. +fn head(content: &str, created_at: i64) -> Head { + Head { + event_id: format!("id-{content}"), + created_at, + projection: projection(content), + } +} + +fn signed_persona(system_prompt: &str) -> nostr::Event { + use std::collections::BTreeMap; + let record = crate::managed_agents::AgentDefinition { + id: "test-persona".to_string(), + display_name: "Test".to_string(), + avatar_url: None, + system_prompt: system_prompt.to_string(), + runtime: None, + model: None, + provider: None, + name_pool: Vec::new(), + is_builtin: false, + is_active: true, + shared: false, + source_team: None, + source_team_persona_slug: None, + catalog_source: None, + env_vars: BTreeMap::new(), + respond_to: None, + respond_to_allowlist: Vec::new(), + parallelism: None, + created_at: "2025-01-01T00:00:00Z".to_string(), + updated_at: "2025-01-01T00:00:00Z".to_string(), + }; + build_persona_event(&record) + .unwrap() + .sign_with_keys(&nostr::Keys::generate()) + .unwrap() +} + +/// Retain a row at the coordinate, optionally pending. Returns the signed +/// event so a caller can stamp a baseline naming it. +fn retain_row(conn: &Connection, kind: u32, d_tag: &str, pending: bool) -> nostr::Event { + retain_prompt(conn, kind, d_tag, pending, "prompt") +} + +/// The same, with control over the persona's system prompt — which is what +/// [`CanonicalProjection`] reads, so it is how a test makes the queued row's +/// content differ from the baseline's. +fn retain_prompt( + conn: &Connection, + kind: u32, + d_tag: &str, + pending: bool, + system_prompt: &str, +) -> nostr::Event { + let event = signed_persona(system_prompt); + retain_event( + conn, + &RetainedEvent { + kind, + pubkey: OWNER.to_string(), + d_tag: d_tag.to_string(), + content: event.content.to_string(), + created_at: event.created_at.as_secs() as i64, + raw_event: { + use nostr::JsonUtil; + event.as_json() + }, + pending_sync: pending, + event_id: Some(event.id.to_hex()), + }, + ) + .unwrap(); + event +} + +/// Re-queue an already-retained row, the way an edit or a reconcile would. +fn set_pending(conn: &Connection, kind: u32, d_tag: &str) { + conn.execute( + "UPDATE persona_events SET pending_sync = 1 + WHERE kind = ?1 AND pubkey = ?2 AND d_tag = ?3", + rusqlite::params![kind, OWNER, d_tag], + ) + .unwrap(); +} + +// ── local_state: reading the observation off a real store ────────────────── + +#[test] +fn test_no_local_row_reports_nothing_queued_and_no_baseline() { + let conn = test_db(); + let observation = local_state( + &conn, + OWNER, + &coordinate(), + Some(projection("v1")), + HeadState::Present(head("v1", 100)), + TombstoneEvidence::NotFound, + ) + .unwrap(); + + assert!(observation.queued.is_none()); + assert!(observation.queued_deletion_at.is_none()); + assert!(observation.baseline.is_none()); +} + +/// The queued projection is read from the row's own `raw_event`, because those +/// are the bytes the flush loop publishes. +#[test] +fn test_pending_live_row_reports_its_own_queued_projection() { + let conn = test_db(); + retain_prompt(&conn, KIND_PERSONA, "test-persona", true, "queued-prompt"); + + let observation = local_state( + &conn, + OWNER, + &coordinate(), + Some(projection("whatever-is-on-disk")), + HeadState::Present(head("v2", 200)), + TombstoneEvidence::NotFound, + ) + .unwrap(); + + let queued = observation.queued.expect("the pending row must be queued"); + assert!( + queued.content.contains("queued-prompt"), + "the queued projection must come from the row's raw_event, got {queued:?}" + ); + assert!( + observation.queued_deletion_at.is_none(), + "a queued edit is not a queued deletion" + ); +} + +/// A queued DELETION lives at a different primary key than the record itself. +/// Reporting only the live row was the cross-coordinate hole from review. +#[test] +fn test_pending_tombstone_row_is_reported_for_its_target_coordinate() { + let conn = test_db(); + // The live row is NOT pending; only the paired tombstone is. + retain_row(&conn, KIND_PERSONA, "test-persona", false); + let tombstone = retain_row(&conn, KIND_DELETION, &tombstone_key(), true); + + let observation = local_state( + &conn, + OWNER, + &coordinate(), + Some(projection("v1")), + HeadState::Present(head("v2", 200)), + TombstoneEvidence::NotFound, + ) + .unwrap(); + + assert_eq!( + observation.queued_deletion_at, + Some(tombstone.created_at.as_secs() as i64), + "a pending tombstone must be reported against its TARGET coordinate" + ); + assert!( + observation.queued.is_none(), + "the live row is settled; only the tombstone is queued" + ); +} + +/// A settled tombstone row must not pin the coordinate as queued forever. +#[test] +fn test_non_pending_tombstone_row_is_not_reported_as_queued() { + let conn = test_db(); + retain_row(&conn, KIND_PERSONA, "test-persona", false); + retain_row(&conn, KIND_DELETION, &tombstone_key(), false); + + let observation = local_state( + &conn, + OWNER, + &coordinate(), + Some(projection("v1")), + HeadState::Present(head("v1", 100)), + TombstoneEvidence::NotFound, + ) + .unwrap(); + + assert!(observation.queued.is_none()); + assert!(observation.queued_deletion_at.is_none()); +} + +#[test] +fn test_stamped_baseline_is_read_back() { + let conn = test_db(); + let event = retain_row(&conn, KIND_PERSONA, "test-persona", false); + set_baseline( + &conn, + KIND_PERSONA, + OWNER, + "test-persona", + &event.id.to_hex(), + "baseline-content", + ) + .unwrap(); + + let observation = local_state( + &conn, + OWNER, + &coordinate(), + Some(projection("baseline-content")), + HeadState::Present(head("newer", 200)), + TombstoneEvidence::NotFound, + ) + .unwrap(); + + assert_eq!( + observation.baseline.as_ref().map(|b| b.content.as_str()), + Some("baseline-content") + ); + // Disk still equals the baseline, so the differing head wins — the exact + // stale-store case the fix exists for. + assert_eq!(decide(&observation).decision, Decision::ApplyHead); +} + +/// A half-written baseline (id without content, or vice versa) is not a usable +/// provenance record and must read as "no baseline". +#[test] +fn test_partial_baseline_reads_as_absent() { + let conn = test_db(); + retain_row(&conn, KIND_PERSONA, "test-persona", false); + conn.execute( + "UPDATE persona_events SET baseline_event_id = 'someid', baseline_content = NULL + WHERE kind = ?1 AND pubkey = ?2 AND d_tag = ?3", + rusqlite::params![KIND_PERSONA, OWNER, "test-persona"], + ) + .unwrap(); + + let observation = local_state( + &conn, + OWNER, + &coordinate(), + Some(projection("local")), + HeadState::Present(head("remote", 200)), + TombstoneEvidence::NotFound, + ) + .unwrap(); + + assert!(observation.baseline.is_none()); + assert_eq!(decide(&observation).decision, Decision::Defer); +} + +#[test] +fn test_plan_pairs_every_coordinate_with_its_decision() { + let base = Observation { + disk: Some(projection("v1")), + queued: None, + head: HeadState::Present(head("v2", 200)), + baseline: Some(projection("v1")), + queued_deletion_at: None, + tombstone: TombstoneEvidence::NotFound, + }; + let states = vec![ + CoordinateState { + coordinate: Coordinate { + kind: KIND_PERSONA, + d_tag: "adopts".to_string(), + }, + observation: base.clone(), + }, + CoordinateState { + coordinate: Coordinate { + kind: KIND_PERSONA, + d_tag: "parks".to_string(), + }, + observation: Observation { + disk: Some(projection("edited")), + head: HeadState::Absent, + ..base + }, + }, + ]; + + let plan = plan(&states); + + assert_eq!(plan.len(), 2); + assert_eq!(plan[0].0.d_tag, "adopts"); + assert_eq!(plan[0].1.decision, Decision::ApplyHead); + assert_eq!(plan[1].0.d_tag, "parks"); + assert_eq!( + plan[1].1.decision, + Decision::Park(ParkReason::LocalEditWithNoHead) + ); +} + +// ── The publication gates (V3.9) ─────────────────────────────────────────── + +fn resolution(decision: Decision) -> Resolution { + Resolution { + decision, + clear_queued_publish: false, + } +} + +/// The live-row gate is deny-by-default over the decision set, so a variant +/// added later is withheld until someone deliberately admits it. +#[test] +fn test_exactly_the_publishing_decisions_release_the_live_gate() { + for decision in [ + Decision::SuppressPublish, + Decision::Park(ParkReason::LocalEditWithNoHead), + Decision::Park(ParkReason::NoBaselineWithHead), + Decision::StampBaseline, + Decision::DeleteLocal, + Decision::Defer, + // The deletion publishes from the tombstone row; the live row it + // deletes must not race it. + Decision::PublishDeletion, + // The one adopt-the-relay cell reachable with NO baseline, so it must + // not leave an open gate behind for a row queued after the pass. + Decision::RestoreFromRelay, + ] { + assert!( + blocks_publication(&resolution(decision.clone())), + "{decision:?} must withhold the live row" + ); + } + + for decision in [ + Decision::PublishLocalEdit, + // Only reachable with a baseline present and nothing queued, so + // gating it would strand a later edit and suppress nothing. + Decision::ApplyHead, + ] { + assert!( + !blocks_publication(&resolution(decision.clone())), + "{decision:?} must not withhold the live row" + ); + } +} + +/// Exactly one decision releases the tombstone row. Anything else — including +/// the timestamp arbitration falling through to the live cells — means the +/// queued deletion did not win. +#[test] +fn test_only_publish_deletion_releases_the_tombstone_gate() { + for decision in [ + Decision::PublishLocalEdit, + Decision::ApplyHead, + Decision::RestoreFromRelay, + Decision::StampBaseline, + Decision::SuppressPublish, + Decision::DeleteLocal, + Decision::Defer, + Decision::Park(ParkReason::NoBaselineWithHead), + Decision::Park(ParkReason::LocalEditWithNoHead), + ] { + assert!( + blocks_tombstone_publication(&resolution(decision.clone())), + "{decision:?} must withhold the queued tombstone" + ); + } + assert!(!blocks_tombstone_publication(&resolution( + Decision::PublishDeletion + ))); +} + +/// The gate's whole purpose: a suppressed coordinate must actually vanish from +/// the flush loop's work list. +#[test] +fn test_suppressed_coordinate_is_withheld_from_the_flush_loop() { + let conn = test_db(); + retain_row(&conn, KIND_PERSONA, "test-persona", true); + + assert_eq!( + get_pending_sync(&conn).unwrap().len(), + 1, + "precondition: the pending row is publishable" + ); + + let observation = Observation { + disk: Some(projection("v1")), + queued: None, + head: HeadState::Absent, + baseline: Some(projection("v1")), + queued_deletion_at: None, + tombstone: TombstoneEvidence::NotFound, + }; + apply_gate( + &conn, + OWNER, + &coordinate(), + &observation, + &resolution(Decision::SuppressPublish), + ) + .unwrap(); + + assert!( + get_pending_sync(&conn).unwrap().is_empty(), + "a suppressed coordinate must not reach the publisher" + ); + // The content and its pending flag survive — suppression withholds, it + // does not discard what this device believes. + let row = get_retained_event(&conn, KIND_PERSONA, OWNER, "test-persona") + .unwrap() + .unwrap(); + assert!(row.pending_sync); +} + +/// Suppression must be reversible: once the head is visible again the same +/// coordinate resumes publishing, or the gate latches shut permanently. +#[test] +fn test_gate_reopens_when_a_later_decision_publishes() { + let conn = test_db(); + retain_row(&conn, KIND_PERSONA, "test-persona", true); + + let suppressed = Observation { + disk: Some(projection("v1")), + queued: None, + head: HeadState::Absent, + baseline: Some(projection("v1")), + queued_deletion_at: None, + tombstone: TombstoneEvidence::NotFound, + }; + apply_gate( + &conn, + OWNER, + &coordinate(), + &suppressed, + &resolution(Decision::SuppressPublish), + ) + .unwrap(); + assert!(is_publish_blocked(&conn, KIND_PERSONA, OWNER, "test-persona").unwrap()); + + // Next boot: the head is visible and disk carries a real edit. + let recovered = Observation { + disk: Some(projection("edited")), + head: HeadState::Present(head("v1", 100)), + ..suppressed + }; + apply_gate( + &conn, + OWNER, + &coordinate(), + &recovered, + &resolution(Decision::PublishLocalEdit), + ) + .unwrap(); + + assert!(!is_publish_blocked(&conn, KIND_PERSONA, OWNER, "test-persona").unwrap()); + assert_eq!( + get_pending_sync(&conn).unwrap().len(), + 1, + "the coordinate must be publishable again" + ); +} + +/// The gate is stored per coordinate: withholding one persona must not +/// withhold another, or one unreachable head would silence the whole store. +#[test] +fn test_gate_is_scoped_to_its_own_coordinate() { + let conn = test_db(); + retain_row(&conn, KIND_PERSONA, "test-persona", true); + retain_row(&conn, KIND_PERSONA, "other-persona", true); + + let observation = Observation { + disk: Some(projection("v1")), + queued: None, + head: HeadState::Absent, + baseline: Some(projection("v1")), + queued_deletion_at: None, + tombstone: TombstoneEvidence::NotFound, + }; + apply_gate( + &conn, + OWNER, + &coordinate(), + &observation, + &resolution(Decision::SuppressPublish), + ) + .unwrap(); + + let pending = get_pending_sync(&conn).unwrap(); + assert_eq!(pending.len(), 1); + assert_eq!(pending[0].d_tag, "other-persona"); +} + +/// The gate has two halves because the rows have two primary keys. A pass that +/// wrote only the live half would compute a deletion suppression and enforce +/// it on the wrong row, leaving the tombstone free to publish. +#[test] +fn test_the_gate_enforces_on_the_tombstone_row_too() { + let conn = test_db(); + retain_row(&conn, KIND_DELETION, &tombstone_key(), true); + + assert_eq!( + get_pending_sync(&conn).unwrap().len(), + 1, + "precondition: the queued tombstone is publishable" + ); + + let observation = Observation { + disk: None, + queued: None, + head: HeadState::Present(head("still-there", 200)), + baseline: None, + queued_deletion_at: Some(100), + tombstone: TombstoneEvidence::NotFound, + }; + apply_gate( + &conn, + OWNER, + &coordinate(), + &observation, + &resolution(Decision::Park(ParkReason::NoBaselineWithHead)), + ) + .unwrap(); + + assert!( + get_pending_sync(&conn).unwrap().is_empty(), + "a withheld deletion must not reach the publisher" + ); + assert!(is_publish_blocked(&conn, KIND_DELETION, OWNER, &tombstone_key()).unwrap()); +} + +// ── Composition: the whole pass, not one layer ───────────────────────────── +// +// Every test above exercises one layer in isolation, which is exactly how a +// gate that can never fire passed a full suite: `decide` was tested with +// hand-built observations and `apply_gate` with hand-picked decisions, so +// nothing asserted that the decision a REAL pending row produces is one the +// gate actually blocks. These run the composition — retained row → +// `local_state` → `decide` → `apply_gate` → `get_pending_sync` — because that +// path is the product behaviour, and the seam between layers is where the bug +// lived. + +/// Build the observation for a coordinate the way the barrier does, from rows +/// that are actually in the store. +fn observe(conn: &Connection, head: HeadState, disk: Option) -> Observation { + local_state( + conn, + OWNER, + &coordinate(), + disk, + head, + TombstoneEvidence::NotFound, + ) + .unwrap() +} + +/// Run the real pass over one coordinate and report what the flush loop would +/// then publish. +fn run_pass(conn: &Connection, observation: &Observation) -> (Decision, usize) { + let plan = run_decision_pass( + conn, + OWNER, + &[CoordinateState { + coordinate: coordinate(), + observation: observation.clone(), + }], + ) + .unwrap(); + ( + plan[0].1.decision.clone(), + get_pending_sync(conn).unwrap().len(), + ) +} + +/// **Will's bug, end to end, and the regression test for this whole PR.** +/// +/// A second install holds the OLD system prompt on disk with a FRESH retention +/// database — so no baseline, which is what a dev build against the same relay +/// always looks like. The good edit was made on the other install and is the +/// relay head. Boot's reconcile has already re-signed the stale disk content +/// at `monotonic_created_at = max(now, head + 1)` and queued it, so it would +/// win the relay's last-write-wins compare on `created_at` every time. +/// +/// Composed through the real path rather than asserted on `decide` alone: 43 +/// layer-isolated tests passed while the composition was a no-op, and that is +/// the gap class this closes. Zero rows publishable is the only acceptable +/// outcome. +#[test] +fn test_stale_store_with_no_baseline_cannot_revert_a_newer_head() { + let conn = test_db(); + // The reconcile's queued re-sign of the stale disk content. No baseline is + // stamped: this install has never agreed to anything at this coordinate. + retain_prompt(&conn, KIND_PERSONA, "test-persona", true, "old-prompt"); + assert_eq!( + get_pending_sync(&conn).unwrap().len(), + 1, + "precondition: without the barrier this row publishes" + ); + assert!( + get_baseline(&conn, KIND_PERSONA, OWNER, "test-persona") + .unwrap() + .is_none(), + "precondition: a fresh retention database has no baseline" + ); + + let observation = observe( + &conn, + // The good edit, on the relay, at an OLDER wall-clock time than the + // re-sign — which is precisely why `created_at` cannot arbitrate this. + HeadState::Present(head("good-edit", 1)), + Some(projection("old-prompt")), + ); + let (decision, publishable) = run_pass(&conn, &observation); + + assert_eq!( + decision, + Decision::Park(ParkReason::NoBaselineWithHead), + "the no-baseline-with-head cell must park" + ); + assert_eq!( + publishable, 0, + "the stale revert reached the publisher (decision: {decision:?})" + ); + // Withheld, not destroyed — the device still holds its copy, and the row + // keeps its pending flag so a later resolution can act on it. + let row = get_retained_event(&conn, KIND_PERSONA, OWNER, "test-persona") + .unwrap() + .expect("the record must survive suppression"); + assert!(row.pending_sync); +} + +/// The same bug in its second shape: a queued DELETION from a no-baseline +/// store against a live head. Worse than the revert, because an a-tag delete +/// destroys content this install has never seen rather than reverting it. +#[test] +fn test_stale_store_with_no_baseline_cannot_delete_a_live_head() { + let conn = test_db(); + retain_row(&conn, KIND_DELETION, &tombstone_key(), true); + + let observation = observe(&conn, HeadState::Present(head("still-there", 1)), None); + let (decision, publishable) = run_pass(&conn, &observation); + + assert_eq!(decision, Decision::Park(ParkReason::NoBaselineWithHead)); + assert_eq!( + publishable, 0, + "an unprovable deletion reached the publisher (decision: {decision:?})" + ); +} + +/// **The tombstone escape, composed over BOTH rows.** +/// +/// A coordinate's record and its queued tombstone occupy different primary +/// keys — `(kind, owner, d_tag)` versus `(5, owner, ":")` — and +/// the flush loop reads them independently. A pass that computes the right +/// decision but enforces it on one key is indistinguishable from no gate at +/// all for the other, so the no-baseline invariant is only meaningful stated +/// over both. +/// +/// The deletion here is deliberately NEWER than the head, which is the exact +/// input `deletion_at >= head.created_at` would resolve to `PublishDeletion`. +/// It must not get there: the no-baseline gate runs first, so an install that +/// cannot show it ever agreed to anything at this coordinate never destroys a +/// head it has not seen. Asserted through `get_pending_sync` — what the flush +/// loop actually selects — not through the decision alone. +#[test] +fn test_no_baseline_store_withholds_both_the_record_and_its_tombstone() { + let conn = test_db(); + // Both halves queued at once: the reconcile's re-sign of stale disk + // content, and a tombstone targeting the same coordinate. + retain_prompt(&conn, KIND_PERSONA, "test-persona", true, "old-prompt"); + let tombstone = retain_row(&conn, KIND_DELETION, &tombstone_key(), true); + assert_eq!( + get_pending_sync(&conn).unwrap().len(), + 2, + "precondition: without the barrier both rows publish" + ); + assert!( + get_baseline(&conn, KIND_PERSONA, OWNER, "test-persona") + .unwrap() + .is_none(), + "precondition: a fresh retention database has no baseline" + ); + + // The head predates the tombstone, so cell 4's timestamp arbitration would + // hand this to `PublishDeletion` if the baseline gate did not run first. + let head_created_at = tombstone.created_at.as_secs() as i64 - 1; + let observation = observe( + &conn, + HeadState::Present(head("good-edit", head_created_at)), + Some(projection("old-prompt")), + ); + assert!( + observation.queued_deletion_at.unwrap() >= head_created_at, + "precondition: the queued deletion must be the newer of the two" + ); + + let (decision, publishable) = run_pass(&conn, &observation); + + assert_eq!( + decision, + Decision::Park(ParkReason::NoBaselineWithHead), + "the no-baseline gate must take this before the timestamp compare" + ); + assert_eq!( + publishable, 0, + "a row escaped the flush-selection gate (decision: {decision:?})" + ); + assert!( + is_publish_blocked(&conn, KIND_PERSONA, OWNER, "test-persona").unwrap(), + "the record's own row must be gated" + ); + assert!( + is_publish_blocked(&conn, KIND_DELETION, OWNER, &tombstone_key()).unwrap(), + "the tombstone row must be gated at its own primary key" + ); +} + +/// The mirror case, and the reason suppression cannot simply block everything +/// pending: a genuine local edit against a head this install has already +/// agreed on must still publish, or the fix freezes every agent config. +#[test] +fn test_genuine_local_edit_still_publishes() { + let conn = test_db(); + let event = retain_prompt(&conn, KIND_PERSONA, "test-persona", true, "my-new-edit"); + // The baseline is what this install last agreed was published; the queued + // row has since moved past it. + set_baseline( + &conn, + KIND_PERSONA, + OWNER, + "test-persona", + &event.id.to_hex(), + "published-v1", + ) + .unwrap(); + + let observation = observe( + &conn, + HeadState::Present(head("published-v1", 100)), + Some(projection("my-new-edit")), + ); + let (decision, publishable) = run_pass(&conn, &observation); + + assert_eq!( + decision, + Decision::PublishLocalEdit, + "a post-baseline edit must be recognized as an edit" + ); + assert_eq!(publishable, 1, "the user's edit must still reach the relay"); +} + +/// A vacuous re-sign — the queued row publishes exactly what the baseline +/// already says — is DROPPED, not gated. Gating would hold it forever, since +/// it can never become publishable. +#[test] +fn test_vacuous_re_sign_is_cleared_rather_than_held() { + let conn = test_db(); + let event = retain_prompt(&conn, KIND_PERSONA, "test-persona", true, "agreed"); + // Baseline names this exact event, so the queued projection equals it. + set_baseline( + &conn, + KIND_PERSONA, + OWNER, + "test-persona", + &event.id.to_hex(), + &event.content, + ) + .unwrap(); + + let observation = observe( + &conn, + HeadState::Present(head("newer-elsewhere", 200)), + Some(projection(&event.content)), + ); + let (_, publishable) = run_pass(&conn, &observation); + + assert_eq!(publishable, 0); + let row = get_retained_event(&conn, KIND_PERSONA, OWNER, "test-persona") + .unwrap() + .unwrap(); + assert!( + !row.pending_sync, + "a re-sign carrying no new intent must be dropped, not held pending forever" + ); +} + +/// A pending row whose disk content already equals the head is the +/// crash-convergence case: stamp the baseline and stop, rather than burn a +/// `created_at` bump republishing identical content. +#[test] +fn test_pending_row_matching_the_head_stamps_and_stops() { + let conn = test_db(); + retain_row(&conn, KIND_PERSONA, "test-persona", true); + + let observation = observe( + &conn, + HeadState::Present(head("agreed", 100)), + Some(projection("agreed")), + ); + let (decision, publishable) = run_pass(&conn, &observation); + + assert_eq!(decision, Decision::Park(ParkReason::NoBaselineWithHead)); + assert_eq!(publishable, 0, "identical content must not be republished"); +} + +/// `StampBaseline` has to actually write the baseline. It is the input every +/// other cell depends on, so a decision computed and dropped leaves the table +/// permanently in its no-provenance state. +#[test] +fn test_stamp_baseline_records_the_head_as_provenance() { + let conn = test_db(); + retain_row(&conn, KIND_PERSONA, "test-persona", false); + + let observation = observe( + &conn, + HeadState::Present(head("agreed", 100)), + Some(projection("agreed")), + ); + run_pass(&conn, &observation); + + assert_eq!( + get_baseline(&conn, KIND_PERSONA, OWNER, "test-persona").unwrap(), + Some(("id-agreed".to_string(), "agreed".to_string())), + "the baseline must be stamped from the head, content and id together" + ); +} + +/// The two-boot sequence that makes the fix durable rather than a one-shot: +/// boot 1 converges and stamps the baseline, boot 2 uses that provenance to +/// recognize the stale copy the reconcile re-queued. +#[test] +fn test_stamped_baseline_lets_the_next_boot_suppress_a_revert() { + let conn = test_db(); + let event = retain_prompt(&conn, KIND_PERSONA, "test-persona", false, "v1"); + // The head IS the event this store already holds — the converged state + // boot 1 is supposed to recognize. + let converged = Head { + event_id: event.id.to_hex(), + created_at: event.created_at.as_secs() as i64, + projection: CanonicalProjection::from_event(&event), + }; + let disk = converged.projection.clone(); + + // Boot 1: disk equals the head, so the baseline is stamped from it. + let settled = observe( + &conn, + HeadState::Present(converged.clone()), + Some(disk.clone()), + ); + let (decision, _) = run_pass(&conn, &settled); + assert_eq!(decision, Decision::StampBaseline); + assert_eq!( + get_baseline(&conn, KIND_PERSONA, OWNER, "test-persona").unwrap(), + Some((converged.event_id.clone(), converged.projection.content)), + "boot 1 must establish provenance" + ); + + // Boot 2: the other install published a newer edit; the reconcile on THIS + // install re-queued its unchanged copy at a manufactured timestamp. + set_pending(&conn, KIND_PERSONA, "test-persona"); + let stale = observe( + &conn, + HeadState::Present(head("good-edit", 1)), + Some(disk.clone()), + ); + let (decision, publishable) = run_pass(&conn, &stale); + + assert_eq!( + publishable, 0, + "with a baseline, the stale copy must be recognized and withheld \ + (decision: {decision:?})" + ); + // Recognized as vacuous rather than merely gated: the queued row says + // exactly what the baseline already says, so it is dropped and the newer + // head is what this coordinate should adopt. + assert_eq!(decision, Decision::ApplyHead); + assert!( + !get_retained_event(&conn, KIND_PERSONA, OWNER, "test-persona") + .unwrap() + .unwrap() + .pending_sync + ); +} + +/// A queued deletion from a store that CAN prove its provenance still reaches +/// the relay. Suppressing every tombstone would resurrect records the user +/// deliberately removed. +#[test] +fn test_a_provable_queued_tombstone_still_publishes() { + let conn = test_db(); + let tombstone = retain_row(&conn, KIND_DELETION, &tombstone_key(), true); + // Provenance at the target coordinate: this install agreed to what is + // published there, so its deletion is arbitrable (cell 4). + retain_prompt(&conn, KIND_PERSONA, "test-persona", false, "agreed"); + let event = get_retained_event(&conn, KIND_PERSONA, OWNER, "test-persona") + .unwrap() + .unwrap(); + set_baseline( + &conn, + KIND_PERSONA, + OWNER, + "test-persona", + event.event_id.as_deref().unwrap(), + &event.content, + ) + .unwrap(); + + // The head is older than the tombstone, so the deletion wins. + let observation = observe( + &conn, + HeadState::Present(head("still-there", 1)), + Some(projection(&event.content)), + ); + assert_eq!( + observation.queued_deletion_at, + Some(tombstone.created_at.as_secs() as i64) + ); + let (decision, publishable) = run_pass(&conn, &observation); + + assert_eq!(decision, Decision::PublishDeletion); + assert_eq!(publishable, 1, "the queued deletion must reach the relay"); + assert_eq!(get_pending_sync(&conn).unwrap()[0].kind, KIND_DELETION); +} diff --git a/desktop/src-tauri/src/managed_agents/decision.rs b/desktop/src-tauri/src/managed_agents/decision.rs new file mode 100644 index 0000000000..b0d8374790 --- /dev/null +++ b/desktop/src-tauri/src/managed_agents/decision.rs @@ -0,0 +1,377 @@ +//! The boot decision table (V3.2, as amended by V3.10 and V3.11). +//! +//! One pure function, [`decide`], maps the observable state of a single config +//! coordinate onto the action the boot pass should take. It performs no I/O so +//! that every cell is testable directly, and so the ordering between them is +//! visible in one place rather than spread across the caller. +//! +//! # Why a table instead of a content diff +//! +//! Agent configs sync as NIP-33 replaceable events resolved last-write-wins on +//! `created_at`, and the boot reconcile re-signs local disk content at +//! `monotonic_created_at = max(now, head + 1)`. A store that is merely *stale* +//! therefore mints events that are always strictly newer than the edits they +//! revert. Content equality alone cannot tell "the user changed this" from +//! "this store never learned about the newer head", so the pass needs +//! provenance — the baseline — and not just a diff. +//! +//! # Why `created_at` decides almost nothing here (V3.11) +//! +//! V3.2's original row 1 resolved a queued local edit against the relay head by +//! "greater `created_at` wins". Against this failure mode that is not +//! arbitration at all: the queued event's timestamp is manufactured by the very +//! path being stopped, so the stale store wins that comparison every single +//! time. Row 1 as written ratified the laundering vector. +//! +//! V3.11 splits it by evidence class. A queued **edit** is resolved against the +//! baseline, never against a timestamp. A queued **deletion** keeps timestamp +//! arbitration, but only downstream of the baseline gates — and it is sound +//! there precisely because every timestamp that reaches it was minted by an +//! interactive save or by the relay, never by a reconcile re-sign. +//! +//! # The suppression/deletion split (V3.10) +//! +//! Head absence and local deletion are decided on different evidence because +//! the cost of being wrong is wildly asymmetric. Missing head → stop +//! publishing, and nothing else; if that read was wrong the cost is one boot of +//! deferral. Removing a local record requires positive tombstone evidence, +//! because for a managed agent the removal is irreversible: the relay +//! projection is a strict subset of the record, so `private_key_nsec`, +//! `auth_tag`, `env_vars`, and the runtime config cannot be restored from it. +//! The retention scope key is only `owner || relay_url`, so any relay-side +//! history loss under the same URL presents as *every* coordinate being +//! head-absent at once — under a delete-on-absence rule that wipes the device. + +use super::canonical_projection::CanonicalProjection; + +/// What the boot pass observes about one coordinate, after the pulls have run +/// and before anything is written. +#[derive(Debug, Clone)] +pub struct Observation { + /// The canonical projection of the record currently on disk, or `None` + /// when the record is absent from the local store. + pub disk: Option, + /// The projection of the event QUEUED for publish at this coordinate. + /// + /// This, not [`Self::disk`], is what the flush loop would put on the relay, + /// so it is what the baseline arbitration compares. The two agree in both + /// minting paths today — the reconcile signs disk content, an interactive + /// save writes both — but the table must not assume that silently: it is + /// the queued bytes that publish, and a divergence would mean deciding + /// about content that is not the content going out. + pub queued: Option, + /// The relay head's state. `Absent` and `LookupFailed` are distinct: only a + /// completed, writer-consistent lookup can report absence. + pub head: HeadState, + /// The projection this install last agreed was published for this + /// coordinate. `None` before the baseline is established. + pub baseline: Option, + /// `created_at` of a queued deletion targeting this coordinate, if one is + /// queued. + /// + /// The only timestamp the table consults, and it is honest: a tombstone is + /// only ever minted by an interactive delete — a real user action at a real + /// time — never by the reconcile re-sign that manufactures timestamps. + pub queued_deletion_at: Option, + /// Positive evidence that the owner deleted this coordinate: a verified + /// owner-authored kind:5 naming it, from a scan that ran to completion. + pub tombstone: TombstoneEvidence, +} + +/// A relay head: what it publishes, when it was published, and the event id +/// that published it. +/// +/// The id travels with the projection because a baseline is a claim about a +/// specific event on the relay, not about content in the abstract. Stamping a +/// baseline from a projection alone would record "this is what's published" +/// with no way to tell later whether the row still holds that same event. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct Head { + pub event_id: String, + pub created_at: i64, + pub projection: CanonicalProjection, +} + +/// The result of looking up a coordinate's head on the relay. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum HeadState { + /// A head exists, with this projection. + Present(Head), + /// A completed lookup found no head. + /// + /// Only reachable from an exact per-coordinate query + /// (`{kinds, authors, #d}`, `limit: 2`); never inferred from a combined + /// inventory, which cannot bound absence due to server-side page caps. + Absent, + /// The lookup did not complete. Never treated as absence. + LookupFailed, +} + +/// Whether the owner positively deleted this coordinate. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum TombstoneEvidence { + /// A verified owner-authored kind:5 names this exact coordinate. + Found, + /// The scan ran to completion and found no tombstone for it. + NotFound, + /// The scan did not complete. Never treated as `NotFound`. + ScanIncomplete, +} + +/// What the boot pass should do with one coordinate. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum Decision { + /// Record that disk and head already agree. Stamps or repairs the + /// baseline; publishes nothing. + StampBaseline, + /// Adopt the relay's version onto disk and stamp the baseline from it. + ApplyHead, + /// Recreate a locally missing record from the relay head. + RestoreFromRelay, + /// A genuine post-baseline local edit: publish the queued event. + PublishLocalEdit, + /// The queued deletion wins: publish the tombstone. + PublishDeletion, + /// Remove the local record — only ever returned with positive tombstone + /// evidence. + DeleteLocal, + /// Stop publishing this coordinate but change nothing. The safe answer + /// whenever the head is absent without deletion evidence. + SuppressPublish, + /// Needs an explicit human decision; never auto-resolves on a later boot. + Park(ParkReason), + /// Not enough information yet. Nothing is written and nothing publishes. + Defer, +} + +/// Why a coordinate was parked, so the resolution surface can explain it. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ParkReason { + /// Disk differs from the baseline and no head exists — there is no + /// trustworthy timestamp for the local edit, and file mtime must not be + /// used to manufacture one. + LocalEditWithNoHead, + /// Something is queued at a coordinate this install has no baseline for, + /// and the relay holds a head. V3.11 cell 2, and the field failure this + /// whole change exists to stop: a second store with its own fresh + /// retention database is ALWAYS in this state, so publishing here is + /// exactly how a stale copy reverts a good edit. + /// + /// Covers a queued DELETION as well as a queued edit, and the deletion is + /// the worse half. An edit that loses this race can be re-made; a deletion + /// published against a head this install never agreed to destroys content + /// it has never seen, and the a-tag delete then propagates that removal to + /// every install that CAN see it. + NoBaselineWithHead, +} + +/// The boot pass's resolution for one coordinate. +/// +/// The decision and the fate of the queued publish travel together in one value +/// deliberately. They are not independent: `ApplyHead` and `PublishLocalEdit` +/// both permit publication, so a caller that took the decision and forgot to +/// drop a vacuous queued row would publish the stale copy those decisions were +/// chosen to avoid. One value cannot be half-used. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct Resolution { + pub decision: Decision, + /// Drop the queued publish: it re-publishes content this install already + /// agreed was published, so it is a reconcile re-sign with nothing to say. + pub clear_queued_publish: bool, +} + +impl Resolution { + fn keep(decision: Decision) -> Self { + Self { + decision, + clear_queued_publish: false, + } + } +} + +/// Decide what to do with one coordinate. +/// +/// Gates are evaluated in order and the first match wins. The ordering is the +/// contract: durable local intent is never silently discarded, every +/// destructive outcome is gated behind positive evidence rather than an +/// inference, and no cell publishes on the strength of a lookup that did not +/// complete. +pub fn decide(observation: &Observation) -> Resolution { + // Gate A — positive deletion evidence, checked first so a coordinate the + // owner deleted elsewhere is never "restored" by the absent-on-disk cell + // and never republished by the cells below. + if observation.tombstone == TombstoneEvidence::Found { + match &observation.head { + // A live head overrides the tombstone: a later upsert legitimately + // re-creates a coordinate, and the head is current relay truth. + HeadState::Present(_) => {} + // Deletion is proven and no head contradicts it. + HeadState::Absent => { + return Resolution::keep(if observation.disk.is_some() { + Decision::DeleteLocal + } else { + // Already gone locally; the tombstone confirms it. Nothing + // to do, but it must not be republished. + Decision::SuppressPublish + }); + } + // The owner deleted this coordinate, but a re-creation reads + // exactly like this and only a completed head read tells them + // apart. Nothing publishes and nothing is removed. + HeadState::LookupFailed => return Resolution::keep(Decision::Defer), + } + } + + // Gate B — nothing queued here is provable (V3.11 cells 2 and 3). + // + // Ordered ahead of BOTH arbitrations below, and that ordering is what V3.11 + // turns on. With no baseline this install cannot show it ever agreed to + // anything at the coordinate, so neither its content nor its timestamps are + // evidence of intent about the relay's state — and a fresh retention + // database is exactly this state. Gate C's timestamp compare is sound only + // downstream of here. + let queued_deletion = observation.queued_deletion_at.is_some(); + if observation.baseline.is_none() && (observation.queued.is_some() || queued_deletion) { + return Resolution::keep(match &observation.head { + // Cell 2 — Will's bug, as a cell. A head exists and this install + // cannot show it ever agreed to anything here, so it cannot claim + // its queued copy is newer intent. Parked, never auto-resolved. + HeadState::Present(_) => Decision::Park(ParkReason::NoBaselineWithHead), + HeadState::Absent => match observation.tombstone { + // Cell 3 — a create against a provably empty coordinate: a + // completed writer-consistent lookup says no head and a + // completed scan says no tombstone. Nothing to revert and + // nothing to resurrect, so the genuinely-new record publishes. + // A queued deletion here removes a purely local coordinate. + TombstoneEvidence::NotFound => { + if queued_deletion { + Decision::PublishDeletion + } else { + Decision::PublishLocalEdit + } + } + // `Found` already returned at gate A; it is grouped here so the + // resurrection vector stays closed if that ever changes. + // `ScanIncomplete` cannot support the absence cell 3 needs. + TombstoneEvidence::Found | TombstoneEvidence::ScanIncomplete => Decision::Defer, + }, + HeadState::LookupFailed => Decision::Defer, + }); + } + + // Gate C — a queued deletion arbitrated against the head by `created_at` + // (V3.11 cell 4). + // + // Sound HERE where V3.2's row 1 was not, and only because of what gate B + // already excluded: both sides are honest. The tombstone was minted by an + // interactive delete at a real time and the head's timestamp is the relay's + // own; neither is the reconcile re-sign that manufactures + // `max(now, head + 1)`. A store with no baseline never reaches this + // compare, which is what stops a stale copy from deleting a head it has + // never seen. + if let Some(deletion_at) = observation.queued_deletion_at { + match &observation.head { + // Deciding a deletion from a lookup that did not complete could + // wipe a head this device simply failed to see. + HeadState::LookupFailed => return Resolution::keep(Decision::Defer), + // Nothing on the relay to contradict the deletion. + HeadState::Absent => return Resolution::keep(Decision::PublishDeletion), + // Deletion wins equality: at the same second the removal is the + // more conservative reading of intent. + HeadState::Present(head) if deletion_at >= head.created_at => { + return Resolution::keep(Decision::PublishDeletion) + } + // The head is strictly newer than the tombstone, so the coordinate + // was legitimately re-created after the delete. The tombstone is + // stale; fall through and decide the live record on its own terms. + HeadState::Present(_) => {} + } + } + + // Gate D — a queued edit resolved against the baseline (V3.11 cells 1a and + // 1b), NEVER against `created_at`. Gate B returned for every no-baseline + // queue, so a baseline is present here whenever anything is queued. + if let (Some(queued), Some(baseline)) = + (observation.queued.as_ref(), observation.baseline.as_ref()) + { + if baseline == queued { + // Cell 1a — the queued event publishes exactly what this install + // already agreed was published. A reconcile re-sign carrying no new + // intent, so it is dropped and the coordinate decided as if nothing + // were queued. + return Resolution { + decision: decide_unqueued(observation).decision, + clear_queued_publish: true, + }; + } + // Cell 1b — the queued event differs from the last agreed publish: + // genuine post-baseline intent, and the only cell that publishes an + // edit. It still needs a completed lookup, because publishing means + // re-signing PAST the head and a lookup that did not complete cannot + // say what the head is. Deferring costs one boot; the row keeps its + // pending flag and publishes on the next pass that can see the relay. + return Resolution::keep(match observation.head { + HeadState::LookupFailed => Decision::Defer, + _ => Decision::PublishLocalEdit, + }); + } + + decide_unqueued(observation) +} + +/// Decide a coordinate with nothing queued for publish. +/// +/// Reached directly when no publish is queued, and from cell 1a once a vacuous +/// queued event has been dropped — the two states are identical from here down, +/// which is why the fall-through is a call rather than a copy of these cells. +fn decide_unqueued(observation: &Observation) -> Resolution { + let Some(disk) = observation.disk.as_ref() else { + // Absent on disk. Absence is never deletion intent; that requires a + // tombstone, handled by gate B. + return Resolution::keep(match &observation.head { + HeadState::Present(_) => Decision::RestoreFromRelay, + // No local record and no head: nothing to reconcile. + HeadState::Absent => Decision::SuppressPublish, + HeadState::LookupFailed => Decision::Defer, + }); + }; + + Resolution::keep(match &observation.head { + // Disk already equals the head. This is the crash-convergence cell: + // inbound retains the head before writing JSON, so a crash between + // those two steps lands here and must repair the baseline rather than + // be misread as a hand edit and republished. + HeadState::Present(head) if head.projection == *disk => Decision::StampBaseline, + + HeadState::Present(_) => match observation.baseline.as_ref() { + // Disk is untouched since the baseline, so the head is strictly + // newer information. + Some(baseline) if baseline == disk => Decision::ApplyHead, + // Disk differs from both: a genuine post-baseline edit. Nothing is + // queued for it yet; the next reconcile queues it and cell 1b + // publishes it. + Some(_) => Decision::PublishLocalEdit, + // No baseline: this install has no provenance for the record, so it + // cannot claim the difference is an edit. + None => Decision::Defer, + }, + + // V3.10 — head absent. Suppress, never delete: a wrong absence read + // costs one boot of deferral, a wrong deletion is unrecoverable. + HeadState::Absent => match observation.baseline.as_ref() { + // Disk is untouched and the head is gone. Under V3.2 this deleted + // locally; V3.10 downgrades it to suppression absent a tombstone. + Some(baseline) if baseline == disk => Decision::SuppressPublish, + // A local edit with no head to sign past: no trustworthy edit + // timestamp exists, so this needs an explicit decision. + Some(_) => Decision::Park(ParkReason::LocalEditWithNoHead), + None => Decision::Defer, + }, + + // Nothing is decided from a lookup that did not complete. + HeadState::LookupFailed => Decision::Defer, + }) +} + +#[cfg(test)] +mod tests; diff --git a/desktop/src-tauri/src/managed_agents/decision/tests.rs b/desktop/src-tauri/src/managed_agents/decision/tests.rs new file mode 100644 index 0000000000..7f821389e2 --- /dev/null +++ b/desktop/src-tauri/src/managed_agents/decision/tests.rs @@ -0,0 +1,554 @@ +use super::*; +use crate::managed_agents::config_sync::{blocks_publication, blocks_tombstone_publication}; + +fn projection(content: &str) -> CanonicalProjection { + CanonicalProjection { + content: content.to_string(), + shared: false, + } +} + +/// A relay head carrying `content` at `created_at`. The event id is derived +/// from the content so two heads with different content never collide on id. +fn head(content: &str, created_at: i64) -> Head { + Head { + event_id: format!("id-{content}"), + created_at, + projection: projection(content), + } +} + +/// A coordinate whose disk, baseline, and head all agree, with nothing queued — +/// the steady state. Individual tests mutate one axis at a time from here. +fn settled() -> Observation { + Observation { + disk: Some(projection("v1")), + queued: None, + head: HeadState::Present(head("v1", 100)), + baseline: Some(projection("v1")), + queued_deletion_at: None, + tombstone: TombstoneEvidence::NotFound, + } +} + +/// Every state the table can be asked about, as a cross product of each axis. +/// +/// The sweeps below are the only proofs that hold over states nobody thought +/// to write down, which is where this table's failures have actually come +/// from — the reverted-config bug was a cell no row-by-row test covered. +/// Enumerating is cheap (under a thousand states) and the coverage is total. +fn all_observations() -> impl Iterator { + let disks = [None, Some(projection("v1")), Some(projection("edited"))]; + let queues = [None, Some(projection("v1")), Some(projection("edited"))]; + let heads = [ + HeadState::Present(head("v1", 100)), + HeadState::Present(head("v2", 200)), + HeadState::Absent, + HeadState::LookupFailed, + ]; + let baselines = [None, Some(projection("v1")), Some(projection("old"))]; + // One tombstone older than every head and one newer, so the cell-4 + // arbitration is exercised in both directions wherever it is reachable. + let deletions = [None, Some(50), Some(500)]; + let tombstones = [ + TombstoneEvidence::Found, + TombstoneEvidence::NotFound, + TombstoneEvidence::ScanIncomplete, + ]; + + disks.into_iter().flat_map(move |disk| { + let queues = queues.clone(); + let heads = heads.clone(); + let baselines = baselines.clone(); + queues.into_iter().flat_map(move |queued| { + let heads = heads.clone(); + let baselines = baselines.clone(); + let disk = disk.clone(); + heads.into_iter().flat_map(move |head| { + let baselines = baselines.clone(); + let disk = disk.clone(); + let queued = queued.clone(); + baselines.into_iter().flat_map(move |baseline| { + let disk = disk.clone(); + let queued = queued.clone(); + let head = head.clone(); + deletions.into_iter().flat_map(move |queued_deletion_at| { + let disk = disk.clone(); + let queued = queued.clone(); + let head = head.clone(); + let baseline = baseline.clone(); + tombstones.into_iter().map(move |tombstone| Observation { + disk: disk.clone(), + queued: queued.clone(), + head: head.clone(), + baseline: baseline.clone(), + queued_deletion_at, + tombstone, + }) + }) + }) + }) + }) + }) +} + +// ── The invariants ───────────────────────────────────────────────────────── + +/// **Will's bug, stated as a property.** This is the whole point of V3.11. +/// +/// A store with no baseline for a coordinate cannot show it ever agreed to +/// anything published there. When the relay nonetheless holds a head, that +/// store's queued content is by definition uninformed about the head — which +/// is exactly the state of a second install with a fresh retention database. +/// Its `created_at` is manufactured by `monotonic_created_at = max(now, +/// head + 1)`, so any comparison decided on it is one the stale store always +/// wins. Nothing may reach the relay from this state, for any combination of +/// the remaining axes. +/// +/// Asserted against the real publication gates rather than against the +/// `Decision` enum, because the gate is what the flush loop obeys: a decision +/// that suppresses but is enforced on the wrong row is indistinguishable from +/// no gate at all. +/// +/// The statement is over BOTH rows and is unconditional on what is queued. A +/// coordinate's live row and its tombstone row have different primary keys, so +/// a single-row invariant is provably insufficient — and the assertion holds +/// even where nothing is queued right now, because the gate this pass leaves +/// behind is what governs a row queued after it. +#[test] +fn test_no_baseline_with_a_live_head_releases_neither_row() { + let mut covered = 0; + for observation in all_observations() { + if observation.baseline.is_some() || !matches!(observation.head, HeadState::Present(_)) { + continue; + } + let resolution = decide(&observation); + + assert!( + blocks_publication(&resolution), + "the live row escaped the gate with no baseline against a live head: \ + {observation:?} -> {resolution:?}" + ); + assert!( + blocks_tombstone_publication(&resolution), + "the tombstone row escaped the gate with no baseline against a live head: \ + {observation:?} -> {resolution:?}" + ); + covered += 1; + } + assert!( + covered > 0, + "the sweep must actually reach the guarded states" + ); +} + +/// The cell-2 decision behind that property, named so a future reader can see +/// which cell the sweep is protecting. +#[test] +fn test_no_baseline_with_a_live_head_parks() { + for queued in [Some(projection("stale")), None] { + for queued_deletion_at in [None, Some(500)] { + if queued.is_none() && queued_deletion_at.is_none() { + continue; // nothing queued: not cell 2 + } + let observation = Observation { + disk: Some(projection("stale")), + queued: queued.clone(), + head: HeadState::Present(head("good-edit", 200)), + baseline: None, + queued_deletion_at, + tombstone: TombstoneEvidence::NotFound, + }; + assert_eq!( + decide(&observation).decision, + Decision::Park(ParkReason::NoBaselineWithHead), + "cell 2 must park, not auto-resolve: {observation:?}" + ); + } + } +} + +/// `created_at` is never consulted for a live queued row against a head. +/// +/// The head's timestamp is varied across the whole range around the queued +/// row's while every other axis is held fixed. If any decision moved, some +/// cell would be arbitrating on a value the reconcile manufactures, and the +/// stale store would win it. +/// +/// Deliberately excludes queued deletions: cell 4 arbitrates those on +/// `created_at` by design, and soundly, because a tombstone is only ever +/// minted by an interactive delete. +#[test] +fn test_head_created_at_never_moves_a_live_decision() { + for observation in all_observations() { + if observation.queued_deletion_at.is_some() { + continue; // cell 4 consults timestamps on purpose + } + let HeadState::Present(present) = &observation.head else { + continue; + }; + let baseline = decide(&observation); + for created_at in [1, 99, 100, 101, 1_000_000] { + let shifted = Observation { + head: HeadState::Present(Head { + created_at, + ..present.clone() + }), + ..observation.clone() + }; + assert_eq!( + decide(&shifted), + baseline, + "the head's created_at moved a live decision: {observation:?} at {created_at}" + ); + } + } +} + +/// No decision path may destroy local state without positive tombstone +/// evidence. The safety property of the whole table, over the full state space +/// rather than row by row: for a managed agent the removal is irreversible, +/// since the relay projection is a strict subset of the local record. +#[test] +fn test_deletion_is_unreachable_without_tombstone_evidence() { + for observation in all_observations() { + if observation.tombstone == TombstoneEvidence::Found { + continue; + } + assert_ne!( + decide(&observation).decision, + Decision::DeleteLocal, + "reached DeleteLocal without evidence: {observation:?}" + ); + } +} + +/// Nothing is decided from a lookup that did not complete. A flaky network +/// must never present as a deletion or as an absence worth publishing into. +/// +/// Scoped to the DECISION. Dropping a vacuous queued publish is deliberately +/// still allowed here: "the queued row says exactly what the baseline says" is +/// a purely local fact that needs no relay evidence, and the row could never +/// become publishable anyway. `test_a_queued_publish_is_only_dropped_when_it_\ +/// equals_the_baseline` is what bounds that drop. +#[test] +fn test_a_failed_head_lookup_decides_nothing() { + for observation in all_observations() { + if observation.head != HeadState::LookupFailed { + continue; + } + assert_eq!( + decide(&observation).decision, + Decision::Defer, + "a failed lookup must decide nothing: {observation:?}" + ); + } +} + +/// A queued publish is only ever DROPPED as vacuous, never as a judgement +/// about content. Dropping it anywhere else would silently discard durable +/// local intent, which is the one thing row 1 has always guaranteed against. +#[test] +fn test_a_queued_publish_is_only_dropped_when_it_equals_the_baseline() { + for observation in all_observations() { + if !decide(&observation).clear_queued_publish { + continue; + } + assert_eq!( + observation.queued, observation.baseline, + "dropped a queued publish that carried new intent: {observation:?}" + ); + assert!(observation.queued.is_some()); + } +} + +// ── Cell 1: a queued row against its baseline ────────────────────────────── + +/// Cell 1a. The queued event publishes exactly what this install already +/// agreed was published, so it is a reconcile re-sign carrying nothing. It is +/// dropped rather than gated: it can never become publishable, so gating would +/// hold it forever. +#[test] +fn test_vacuous_re_sign_is_dropped_and_falls_through() { + let observation = Observation { + disk: Some(projection("v1")), + queued: Some(projection("v1")), + head: HeadState::Present(head("v2", 200)), + baseline: Some(projection("v1")), + queued_deletion_at: None, + tombstone: TombstoneEvidence::NotFound, + }; + let resolution = decide(&observation); + assert!(resolution.clear_queued_publish); + // Fell through to the unqueued cells, which see disk untouched since the + // baseline and adopt the newer head. + assert_eq!(resolution.decision, Decision::ApplyHead); +} + +/// Cell 1b. An edit made after the last agreed publish is real user intent and +/// must still reach the relay — suppression cannot simply block anything +/// queued. +#[test] +fn test_queued_edit_past_the_baseline_publishes() { + let observation = Observation { + disk: Some(projection("my-edit")), + queued: Some(projection("my-edit")), + head: HeadState::Present(head("v1", 100)), + baseline: Some(projection("v1")), + queued_deletion_at: None, + tombstone: TombstoneEvidence::NotFound, + }; + let resolution = decide(&observation); + assert_eq!(resolution.decision, Decision::PublishLocalEdit); + assert!(!blocks_publication(&resolution)); +} + +/// The comparand is the QUEUED event, not disk. The flush loop publishes the +/// queued row's bytes, so a divergence between disk and the queued row must be +/// decided on what actually goes out. Here disk still equals the baseline +/// while the queued row does not — the queued row wins the comparison. +#[test] +fn test_arbitration_reads_the_queued_row_not_disk() { + let observation = Observation { + disk: Some(projection("v1")), + queued: Some(projection("diverged")), + head: HeadState::Present(head("v1", 100)), + baseline: Some(projection("v1")), + queued_deletion_at: None, + tombstone: TombstoneEvidence::NotFound, + }; + assert_eq!(decide(&observation).decision, Decision::PublishLocalEdit); +} + +// ── Cell 3: the evidence-gated carve-out ─────────────────────────────────── + +/// A create against a provably empty coordinate: a completed writer-consistent +/// lookup says no head and a completed scan says no tombstone. Nothing to +/// revert and nothing to resurrect, so a genuinely new agent publishes rather +/// than parking forever. +#[test] +fn test_new_coordinate_with_both_evidence_gates_publishes() { + let observation = Observation { + disk: Some(projection("brand-new")), + queued: Some(projection("brand-new")), + head: HeadState::Absent, + baseline: None, + queued_deletion_at: None, + tombstone: TombstoneEvidence::NotFound, + }; + let resolution = decide(&observation); + assert_eq!(resolution.decision, Decision::PublishLocalEdit); + assert!(!blocks_publication(&resolution)); +} + +/// The resurrection vector. A deleted coordinate reads head-absent exactly +/// like a new one, and the relay keeps no watermark — so without the tombstone +/// gate, cell 3 would republish every record the owner ever deleted. +#[test] +fn test_a_deleted_coordinate_is_never_republished_as_new() { + let observation = Observation { + disk: Some(projection("deleted-agent")), + queued: Some(projection("deleted-agent")), + head: HeadState::Absent, + baseline: None, + queued_deletion_at: None, + tombstone: TombstoneEvidence::Found, + }; + let resolution = decide(&observation); + assert_eq!(resolution.decision, Decision::DeleteLocal); + assert!(blocks_publication(&resolution)); +} + +/// An incomplete scan cannot support the absence cell 3 needs, so the queued +/// row is held rather than published on unproven evidence. +#[test] +fn test_new_coordinate_defers_on_an_incomplete_scan() { + let observation = Observation { + disk: Some(projection("brand-new")), + queued: Some(projection("brand-new")), + head: HeadState::Absent, + baseline: None, + queued_deletion_at: None, + tombstone: TombstoneEvidence::ScanIncomplete, + }; + assert_eq!(decide(&observation).decision, Decision::Defer); +} + +// ── Cell 4: tombstone-vs-live arbitration ────────────────────────────────── + +/// Downstream of the baseline gates, both timestamps are honest — the +/// tombstone's was minted by an interactive delete, the head's by the relay — +/// so arbitrating on them is sound. Deletion wins equality: at the same second +/// the removal is the more conservative reading of intent. +#[test] +fn test_a_tombstone_at_or_after_the_head_publishes_the_deletion() { + for deletion_at in [100, 200] { + let observation = Observation { + queued_deletion_at: Some(deletion_at), + ..settled() + }; + let resolution = decide(&observation); + assert_eq!(resolution.decision, Decision::PublishDeletion); + // The tombstone row is released and the live row it deletes is + // withheld, so the record cannot race its own deletion. + assert!(!blocks_tombstone_publication(&resolution)); + assert!(blocks_publication(&resolution)); + } +} + +/// A head strictly newer than the tombstone means the coordinate was +/// legitimately re-created after the delete. The tombstone is stale and the +/// live record is decided on its own terms. +#[test] +fn test_a_head_newer_than_the_tombstone_survives_it() { + let observation = Observation { + queued_deletion_at: Some(50), + ..settled() + }; + let resolution = decide(&observation); + assert_eq!(resolution.decision, Decision::StampBaseline); + assert!(blocks_tombstone_publication(&resolution)); +} + +/// Deciding a deletion from a lookup that did not complete could wipe a head +/// this device simply failed to see. +#[test] +fn test_a_deletion_behind_a_failed_lookup_defers() { + let observation = Observation { + head: HeadState::LookupFailed, + queued_deletion_at: Some(500), + ..settled() + }; + assert_eq!(decide(&observation).decision, Decision::Defer); +} + +// ── V3.10: suppression is not deletion ───────────────────────────────────── + +/// The core V3.10 rule. Relay-side history loss under one URL presents as +/// every coordinate being head-absent at once; under a delete-on-absence rule +/// that wipes the device. +#[test] +fn test_head_absent_with_clean_disk_suppresses_instead_of_deleting() { + let observation = Observation { + head: HeadState::Absent, + ..settled() + }; + let resolution = decide(&observation); + assert_eq!(resolution.decision, Decision::SuppressPublish); + assert!(blocks_publication(&resolution)); +} + +/// A live head is current relay truth: a later upsert legitimately re-creates +/// a coordinate, so an older tombstone must not delete it. +#[test] +fn test_live_head_overrides_an_older_tombstone() { + let observation = Observation { + tombstone: TombstoneEvidence::Found, + ..settled() + }; + assert_eq!(decide(&observation).decision, Decision::StampBaseline); +} + +/// Proven deletion plus a head read that never completed is ambiguous: a +/// re-creation reads the same way. Neither removing nor republishing is safe. +#[test] +fn test_tombstone_with_a_failed_lookup_defers() { + let observation = Observation { + head: HeadState::LookupFailed, + tombstone: TombstoneEvidence::Found, + ..settled() + }; + assert_eq!(decide(&observation).decision, Decision::Defer); +} + +// ── Crash convergence and the unqueued cells ─────────────────────────────── + +/// Inbound retains the head BEFORE writing JSON, so a crash in between leaves +/// disk equal to the head but unequal to the old baseline. That must repair +/// the baseline, not be misread as a hand edit and republished. +#[test] +fn test_disk_equal_to_head_stamps_baseline_even_when_baseline_is_stale() { + let observation = Observation { + disk: Some(projection("v2")), + head: HeadState::Present(head("v2", 200)), + baseline: Some(projection("v1")), + ..settled() + }; + assert_eq!(decide(&observation).decision, Decision::StampBaseline); +} + +/// The same cell with no baseline at all. This is the bootstrap: without it +/// the baseline stays `None` forever and every baseline-dependent cell is +/// unreachable. +#[test] +fn test_disk_equal_to_head_stamps_baseline_when_baseline_is_missing() { + let observation = Observation { + disk: Some(projection("v2")), + head: HeadState::Present(head("v2", 200)), + baseline: None, + ..settled() + }; + assert_eq!(decide(&observation).decision, Decision::StampBaseline); +} + +/// Disk is untouched since the baseline, so the differing head is strictly +/// newer information and wins. +#[test] +fn test_untouched_disk_adopts_a_differing_head() { + let observation = Observation { + disk: Some(projection("v1")), + head: HeadState::Present(head("v2", 200)), + baseline: Some(projection("v1")), + ..settled() + }; + assert_eq!(decide(&observation).decision, Decision::ApplyHead); +} + +/// Without a baseline the difference cannot be attributed to a local edit, so +/// nothing is adopted and nothing is published. +#[test] +fn test_no_baseline_never_acts_on_a_difference() { + let observation = Observation { + disk: Some(projection("stale")), + head: HeadState::Present(head("current", 200)), + baseline: None, + ..settled() + }; + assert_eq!(decide(&observation).decision, Decision::Defer); +} + +/// Absence on disk is never deletion intent; that requires a tombstone. +#[test] +fn test_missing_disk_record_with_live_head_restores() { + let observation = Observation { + disk: None, + ..settled() + }; + assert_eq!(decide(&observation).decision, Decision::RestoreFromRelay); +} + +#[test] +fn test_missing_disk_record_and_missing_head_is_a_no_op() { + let observation = Observation { + disk: None, + head: HeadState::Absent, + ..settled() + }; + assert_eq!(decide(&observation).decision, Decision::SuppressPublish); +} + +/// A local edit with no head has no trustworthy timestamp to sign past, and +/// file mtime must not be used to manufacture one — so it needs a human. +#[test] +fn test_local_edit_with_no_head_parks() { + let observation = Observation { + disk: Some(projection("edited")), + head: HeadState::Absent, + baseline: Some(projection("v1")), + ..settled() + }; + assert_eq!( + decide(&observation).decision, + Decision::Park(ParkReason::LocalEditWithNoHead) + ); +} diff --git a/desktop/src-tauri/src/managed_agents/head_lookup.rs b/desktop/src-tauri/src/managed_agents/head_lookup.rs new file mode 100644 index 0000000000..dbeccf7cd1 --- /dev/null +++ b/desktop/src-tauri/src/managed_agents/head_lookup.rs @@ -0,0 +1,140 @@ +//! Exact per-coordinate head lookups for the boot decision pass (V3.3). +//! +//! The decision table may only treat a head as ABSENT on the strength of a +//! lookup that (a) named the exact coordinate and (b) completed. Anything +//! weaker is reported as [`HeadState::LookupFailed`], which decides nothing. +//! +//! # Why absence needs its own query shape +//! +//! It is tempting to pull one inventory of all `30175/30176/30177` events for +//! the owner and treat "not in the list" as absent. That is unsound: the +//! effective server-side cap on a query is 1000 rows +//! (`crates/buzz-db/src/event.rs`), so a truncated page is indistinguishable +//! from a short one — absence would be inferred from a cut-off list. The +//! inventory may discover records, but only an exact `{kinds, authors, #d}` +//! lookup may deny them. +//! +//! # Known limitation +//! +//! On relay deployments that route historical reads through a read-replica +//! pool, a replica that has not yet replayed an event returns the same empty +//! page as a genuine deletion, with no staleness budget bounding the +//! difference. The barrier's tracing line is the detection mechanism if this +//! ever fires in the wild; re-adding a `sync_authoritative` opt-in to the +//! protocol is a small standalone change reversible on evidence. + +use serde_json::json; + +use super::{ + canonical_projection::CanonicalProjection, + decision::{Head, HeadState}, + retention::RetentionScope, +}; +use crate::app_state::AppState; + +/// One coordinate to look up: kind plus `d` tag, under the scope's owner. +#[derive(Debug, Clone, PartialEq, Eq, Hash)] +pub struct Coordinate { + pub kind: u32, + pub d_tag: String, +} + +/// Build the exact filter for one coordinate's head. +/// +/// `limit: 2` rather than 1 is deliberate: NIP-33 guarantees at most one head +/// per coordinate, so a second row means the relay disagrees with that +/// assumption and the result must not be trusted as authoritative. +pub fn head_filter(owner_pubkey_hex: &str, coordinate: &Coordinate) -> serde_json::Value { + json!({ + "kinds": [coordinate.kind], + "authors": [owner_pubkey_hex], + "#d": [coordinate.d_tag], + "limit": 2, + }) +} + +/// Interpret the events a head lookup returned for one coordinate. +/// +/// Split from the network call so the interpretation rules are testable +/// without a relay. The caller passes only events it actually received; a +/// transport error must become [`HeadState::LookupFailed`] instead of an empty +/// slice, or a network failure would masquerade as a deletion. +pub fn head_state_from_events(coordinate: &Coordinate, events: &[nostr::Event]) -> HeadState { + // Defense in depth: the relay filtered by kind and `#d`, but a mismatched + // row here means the answer is about a different coordinate, so it cannot + // be used to deny this one. + let matching: Vec<&nostr::Event> = events + .iter() + .filter(|event| { + event.kind.as_u16() as u32 == coordinate.kind + && event_d_tag(event).as_deref() == Some(coordinate.d_tag.as_str()) + }) + .collect(); + + match matching.len() { + 0 => HeadState::Absent, + 1 => HeadState::Present(Head { + event_id: matching[0].id.to_hex(), + // Safety: nostr timestamps are seconds and stay below i64::MAX + // until year 2262 — the same cast the retention store uses. + created_at: matching[0].created_at.as_secs() as i64, + projection: CanonicalProjection::from_event(matching[0]), + }), + // More than one head for a replaceable coordinate contradicts NIP-33. + // Refuse to decide rather than pick one. + _ => HeadState::LookupFailed, + } +} + +/// The `d` tag of an event, if it has exactly one. +fn event_d_tag(event: &nostr::Event) -> Option { + let mut found: Option = None; + for tag in event.tags.iter() { + let parts = tag.as_slice(); + if parts.len() >= 2 && parts[0].as_str() == "d" { + if found.is_some() { + // Two `d` tags: the coordinate is ambiguous. + return None; + } + found = Some(parts[1].to_string()); + } + } + found +} + +/// Look up one coordinate's head on the relay. +/// +/// Every failure mode — transport error, malformed response, contradictory +/// result — collapses to [`HeadState::LookupFailed`] so that no caller can +/// mistake a failed lookup for an authoritative absence. +pub async fn fetch_head( + state: &AppState, + scope: &RetentionScope, + api_base_url: &str, + coordinate: &Coordinate, +) -> HeadState { + let owner_hex = scope.owner_keys.public_key().to_hex(); + let filter = head_filter(&owner_hex, coordinate); + + match crate::relay::query_relay_at_with_keys( + state, + api_base_url, + std::slice::from_ref(&filter), + &scope.owner_keys, + None, + ) + .await + { + Ok(events) => head_state_from_events(coordinate, &events), + Err(error) => { + eprintln!( + "buzz-desktop: config-sync: head lookup failed for {}:{} — {error}", + coordinate.kind, coordinate.d_tag + ); + HeadState::LookupFailed + } + } +} + +#[cfg(test)] +mod tests; diff --git a/desktop/src-tauri/src/managed_agents/head_lookup/tests.rs b/desktop/src-tauri/src/managed_agents/head_lookup/tests.rs new file mode 100644 index 0000000000..033cddeb13 --- /dev/null +++ b/desktop/src-tauri/src/managed_agents/head_lookup/tests.rs @@ -0,0 +1,144 @@ +use super::*; +use crate::managed_agents::persona_events::build_persona_event; +use buzz_core_pkg::kind::{KIND_PERSONA, KIND_TEAM}; + +fn coordinate() -> Coordinate { + Coordinate { + kind: KIND_PERSONA, + d_tag: "test-persona".to_string(), + } +} + +fn signed_persona(keys: &nostr::Keys, d_tag: &str, shared: bool) -> nostr::Event { + use std::collections::BTreeMap; + let record = crate::managed_agents::AgentDefinition { + id: d_tag.to_string(), + display_name: "Test".to_string(), + avatar_url: None, + system_prompt: "prompt".to_string(), + runtime: None, + model: None, + provider: None, + name_pool: Vec::new(), + is_builtin: false, + is_active: true, + shared, + source_team: None, + source_team_persona_slug: None, + catalog_source: None, + env_vars: BTreeMap::new(), + respond_to: None, + respond_to_allowlist: Vec::new(), + parallelism: None, + created_at: "2025-01-01T00:00:00Z".to_string(), + updated_at: "2025-01-01T00:00:00Z".to_string(), + }; + build_persona_event(&record) + .unwrap() + .sign_with_keys(keys) + .unwrap() +} + +/// The filter names the exact coordinate (kinds + authors + #d) and requests +/// limit:2 so a NIP-33 violation is detectable. This shape is what makes the +/// absence claim sound — an inventory query cannot bound absence. +#[test] +fn test_head_filter_is_exact() { + let filter = head_filter("ownerhex", &coordinate()); + + assert_eq!(filter["kinds"], serde_json::json!([KIND_PERSONA])); + assert_eq!(filter["authors"], serde_json::json!(["ownerhex"])); + assert_eq!(filter["#d"], serde_json::json!(["test-persona"])); +} + +/// A cap of 1 could not tell "exactly one head" from "the relay truncated the +/// page", so the lookup asks for one more than NIP-33 permits. +#[test] +fn test_head_filter_can_observe_a_nip33_violation() { + assert_eq!(head_filter("ownerhex", &coordinate())["limit"], 2); +} + +#[test] +fn test_empty_result_is_absent() { + assert_eq!( + head_state_from_events(&coordinate(), &[]), + HeadState::Absent + ); +} + +#[test] +fn test_single_matching_event_is_present_with_its_projection() { + let keys = nostr::Keys::generate(); + let event = signed_persona(&keys, "test-persona", false); + + assert_eq!( + head_state_from_events(&coordinate(), std::slice::from_ref(&event)), + HeadState::Present(Head { + event_id: event.id.to_hex(), + created_at: event.created_at.as_secs() as i64, + projection: CanonicalProjection::from_event(&event), + }) + ); +} + +/// The `shared` tag has to survive the lookup, or a disk-vs-head compare would +/// report a spurious difference for a shared persona and republish it forever. +#[test] +fn test_present_head_preserves_the_shared_tag() { + let keys = nostr::Keys::generate(); + let event = signed_persona(&keys, "test-persona", true); + + match head_state_from_events(&coordinate(), std::slice::from_ref(&event)) { + HeadState::Present(head) => assert!(head.projection.shared), + other => panic!("expected Present, got {other:?}"), + } +} + +/// Two heads for one replaceable coordinate contradicts NIP-33. Picking one +/// would make the decision depend on relay row order, so the lookup refuses. +#[test] +fn test_multiple_heads_fail_the_lookup_rather_than_guessing() { + let keys = nostr::Keys::generate(); + let events = [ + signed_persona(&keys, "test-persona", false), + signed_persona(&keys, "test-persona", true), + ]; + + assert_eq!( + head_state_from_events(&coordinate(), &events), + HeadState::LookupFailed + ); +} + +/// A row for a different coordinate cannot be used to answer this one. If it +/// were counted as a match, an unrelated event would mask a real absence; if +/// it were merely ignored without filtering, a mismatched relay response would +/// still read as "present". +#[test] +fn test_events_for_other_coordinates_are_ignored() { + let keys = nostr::Keys::generate(); + let other = signed_persona(&keys, "a-different-persona", false); + + assert_eq!( + head_state_from_events(&coordinate(), std::slice::from_ref(&other)), + HeadState::Absent + ); +} + +/// A kind mismatch is the cross-kind d-tag collision case: d-tags are not +/// unique across kinds, so a team with the same d-tag must not answer a +/// persona lookup. +#[test] +fn test_events_of_another_kind_do_not_answer_the_lookup() { + let keys = nostr::Keys::generate(); + let persona = signed_persona(&keys, "test-persona", false); + + let team_coordinate = Coordinate { + kind: KIND_TEAM, + d_tag: "test-persona".to_string(), + }; + assert_eq!( + head_state_from_events(&team_coordinate, std::slice::from_ref(&persona)), + HeadState::Absent + ); +} diff --git a/desktop/src-tauri/src/managed_agents/mod.rs b/desktop/src-tauri/src/managed_agents/mod.rs index 772d707f27..8ae92874b5 100644 --- a/desktop/src-tauri/src/managed_agents/mod.rs +++ b/desktop/src-tauri/src/managed_agents/mod.rs @@ -2,7 +2,13 @@ mod agent_env; pub(crate) mod agent_events; pub(crate) mod agent_snapshot; pub(crate) mod agent_snapshot_envelope; +pub(crate) mod canonical_projection; +pub(crate) mod config_barrier; +pub(crate) mod config_sync; +pub(crate) mod decision; +pub(crate) mod head_lookup; pub(crate) mod team_snapshot; +pub(crate) mod tombstone_scan; pub(crate) use agent_env::{ baked_build_env, build_buzz_agent_provider_defaults, discovery_env_with_baked_floor, }; diff --git a/desktop/src-tauri/src/managed_agents/retention.rs b/desktop/src-tauri/src/managed_agents/retention.rs index 9476b3a468..0c24cac6f4 100644 --- a/desktop/src-tauri/src/managed_agents/retention.rs +++ b/desktop/src-tauri/src/managed_agents/retention.rs @@ -208,6 +208,7 @@ pub fn open_retention_db(path: &Path) -> Result { event_id TEXT, baseline_event_id TEXT, baseline_content TEXT, + publish_blocked INTEGER NOT NULL DEFAULT 0, PRIMARY KEY (kind, pubkey, d_tag) );", ) @@ -480,12 +481,23 @@ pub fn get_retained_personas( /// deletion soft-deletes every live row at the coordinate with no timestamp /// comparison, so a tombstone published AFTER the replacement would wipe it. /// Within each group, oldest first for the same reason. +/// +/// # The publication gate (V3.9) +/// +/// `publish_blocked = 1` withholds a row from this list while leaving it +/// retained. The boot pass sets it when it decides a coordinate must stop +/// publishing — a head it cannot see, or a conflict parked for the user — +/// and clearing `pending_sync` would be wrong for both: the local content is +/// still the thing this device believes, and dropping the flag would lose that. +/// Suppression therefore needs a state of its own, and this query is the single +/// place it takes effect. Anything that bypasses this function bypasses the +/// gate. pub fn get_pending_sync(conn: &Connection) -> Result, String> { let mut stmt = conn .prepare( "SELECT kind, pubkey, d_tag, content, created_at, raw_event, pending_sync, event_id FROM persona_events - WHERE pending_sync = 1 + WHERE pending_sync = 1 AND publish_blocked = 0 ORDER BY (kind != 5), created_at ASC", ) .map_err(|e| format!("failed to prepare pending sync query: {e}"))?; @@ -509,6 +521,57 @@ pub fn get_pending_sync(conn: &Connection) -> Result, String> .map_err(|e| format!("failed to read pending sync row: {e}")) } +/// Every coordinate this owner has queued for publish, gated or not. +/// +/// The boot barrier enumerates from disk, but a pending row can outlive its +/// record: a deleted persona, a removed agent, or a kind-5 tombstone all leave +/// a queued row with nothing on disk to enumerate. Those are precisely the rows +/// that publish unsupervised, so the barrier unions this with its disk walk. +/// +/// Tombstone rows are mapped back to the coordinate they target, since that is +/// the coordinate the decision is about. Rows whose gate is already set are +/// still returned: a gate must be re-decided each boot so suppression can lift +/// once the condition that caused it is gone. +pub fn pending_coordinates(conn: &Connection, pubkey: &str) -> Result, String> { + let mut stmt = conn + .prepare( + "SELECT kind, d_tag FROM persona_events + WHERE pending_sync = 1 AND pubkey = ?1", + ) + .map_err(|e| format!("failed to prepare pending coordinate query: {e}"))?; + + let rows = stmt + .query_map(params![pubkey], |row| { + Ok((row.get::<_, u32>(0)?, row.get::<_, String>(1)?)) + }) + .map_err(|e| format!("failed to query pending coordinates: {e}"))?; + + let mut out = Vec::new(); + for row in rows { + let (kind, d_tag) = row.map_err(|e| format!("failed to read pending coordinate: {e}"))?; + out.push(if kind == buzz_core_pkg::kind::KIND_DELETION { + match parse_tombstone_retention_d_tag(&d_tag) { + Some(target) => target, + // An unparseable tombstone key names no coordinate, so there is + // nothing to decide about. Skipped rather than guessed at. + None => continue, + } + } else { + (kind, d_tag) + }); + } + Ok(out) +} + +/// Recover the target coordinate from a tombstone row's retention key. +/// +/// Inverse of [`tombstone_retention_d_tag`]. Splits once from the left because +/// only the kind prefix is structural — a target d-tag may itself contain `:`. +fn parse_tombstone_retention_d_tag(key: &str) -> Option<(u32, String)> { + let (kind, d_tag) = key.split_once(':')?; + Some((kind.parse().ok()?, d_tag.to_string())) +} + /// Clear the `pending_sync` flag for an event the relay just confirmed. /// /// Compare-and-clear: only clears the row if its `created_at` and `content` @@ -535,6 +598,114 @@ pub fn mark_synced( Ok(()) } +/// Drop a queued publish without touching the row's retained content. +/// +/// Used by the boot pass for a queued event that publishes exactly what the +/// baseline already says is published — a reconcile re-sign carrying no new +/// intent. Such a row can never become publishable, so gating it would hold it +/// forever; clearing the flag is the honest outcome. +/// +/// Distinct from [`mark_synced`], which clears the flag on evidence the RELAY +/// accepted those exact bytes. This clears it on evidence the relay already +/// had them. +pub fn clear_pending_sync( + conn: &Connection, + kind: u32, + pubkey: &str, + d_tag: &str, +) -> Result<(), String> { + conn.execute( + "UPDATE persona_events SET pending_sync = 0 + WHERE kind = ?1 AND pubkey = ?2 AND d_tag = ?3", + params![kind, pubkey, d_tag], + ) + .map_err(|e| format!("failed to clear pending sync: {e}"))?; + + Ok(()) +} + +/// Withhold a coordinate from the flush loop without discarding its content. +/// +/// Set when the boot pass decides a coordinate must stop publishing: a head it +/// could not see (V3.10 suppression) or a conflict parked for the user (V3.6). +/// The row keeps `pending_sync` and its content — this device still believes +/// that content — but [`get_pending_sync`] will not hand it to the publisher. +/// +/// Idempotent, and a no-op on a coordinate with no row: there is nothing to +/// withhold until something is retained there. +pub fn set_publish_blocked( + conn: &Connection, + kind: u32, + pubkey: &str, + d_tag: &str, + blocked: bool, +) -> Result<(), String> { + conn.execute( + "UPDATE persona_events SET publish_blocked = ?4 + WHERE kind = ?1 AND pubkey = ?2 AND d_tag = ?3", + params![kind, pubkey, d_tag, i32::from(blocked)], + ) + .map_err(|e| format!("failed to set publish gate: {e}"))?; + + Ok(()) +} + +/// The baseline recorded for a coordinate: what this install last agreed was +/// published there. +/// +/// Read as a unit because the two columns are only meaningful together — an +/// id without content cannot be compared, and content without an id has no +/// provenance. `None` means the baseline was never established. +pub fn get_baseline( + conn: &Connection, + kind: u32, + pubkey: &str, + d_tag: &str, +) -> Result, String> { + conn.query_row( + "SELECT baseline_event_id, baseline_content + FROM persona_events + WHERE kind = ?1 AND pubkey = ?2 AND d_tag = ?3", + params![kind, pubkey, d_tag], + |row| { + Ok(( + row.get::<_, Option>(0)?, + row.get::<_, Option>(1)?, + )) + }, + ) + .optional() + .map_err(|e| format!("failed to get baseline: {e}")) + // Both halves must be present: a row carrying only one is not a usable + // baseline and is treated as if none was recorded. + .map(|row| row.and_then(|(id, content)| Some((id?, content?)))) +} + +/// Record what this install now agrees is published at a coordinate. +/// +/// Only the boot decision pass and the publish-confirmation path call this: a +/// baseline is a claim about the RELAY's state, so it may only be stamped from +/// an event that is actually on the relay — never from local disk content that +/// has not been published. +pub fn set_baseline( + conn: &Connection, + kind: u32, + pubkey: &str, + d_tag: &str, + baseline_event_id: &str, + baseline_content: &str, +) -> Result<(), String> { + conn.execute( + "UPDATE persona_events + SET baseline_event_id = ?4, baseline_content = ?5 + WHERE kind = ?1 AND pubkey = ?2 AND d_tag = ?3", + params![kind, pubkey, d_tag, baseline_event_id, baseline_content], + ) + .map_err(|e| format!("failed to set baseline: {e}"))?; + + Ok(()) +} + /// Delete a retained event by its coordinate. /// /// Called from the synchronous, lock-held delete-persona command body so the @@ -557,6 +728,30 @@ pub fn delete_retained_event( Ok(()) } +/// Whether a coordinate's row is currently withheld from the flush loop. +/// +/// Test-only: production code never asks, because +/// [`get_pending_sync`] applies the gate in SQL rather than filtering after +/// the fact. Tests need to distinguish "withheld" from "no row at all", which +/// a `get_pending_sync` count alone cannot. +#[cfg(test)] +pub fn is_publish_blocked( + conn: &Connection, + kind: u32, + pubkey: &str, + d_tag: &str, +) -> Result { + conn.query_row( + "SELECT publish_blocked FROM persona_events + WHERE kind = ?1 AND pubkey = ?2 AND d_tag = ?3", + params![kind, pubkey, d_tag], + |row| row.get::<_, i32>(0), + ) + .optional() + .map_err(|e| format!("failed to read publish gate: {e}")) + .map(|blocked| blocked == Some(1)) +} + /// Check if the retention store has any persona events for the given pubkey. #[cfg(test)] pub fn has_retained_personas(conn: &Connection, pubkey: &str) -> Result { diff --git a/desktop/src-tauri/src/managed_agents/retention/schema.rs b/desktop/src-tauri/src/managed_agents/retention/schema.rs index c0d548aabd..82cf7cff7a 100644 --- a/desktop/src-tauri/src/managed_agents/retention/schema.rs +++ b/desktop/src-tauri/src/managed_agents/retention/schema.rs @@ -13,6 +13,10 @@ //! published. This is what lets the boot pass tell "the user edited this //! locally" from "this store never learned about the newer relay head", //! instead of inferring intent from a content diff. +//! - `publish_blocked` — the publication gate. A pending row is normally +//! published by the flush loop; a row the boot pass suppressed must stay +//! retained but unpublished, and there is no other state that expresses +//! "durable, but must not go out." //! //! # Crash and concurrency safety //! @@ -27,10 +31,11 @@ use rusqlite::{Connection, TransactionBehavior}; /// Columns added after the initial `persona_events` shape. -const ADDED_COLUMNS: [(&str, &str); 3] = [ +const ADDED_COLUMNS: [(&str, &str); 4] = [ ("event_id", "TEXT"), ("baseline_event_id", "TEXT"), ("baseline_content", "TEXT"), + ("publish_blocked", "INTEGER NOT NULL DEFAULT 0"), ]; /// Bring `persona_events` up to the current column set. diff --git a/desktop/src-tauri/src/managed_agents/tombstone_scan.rs b/desktop/src-tauri/src/managed_agents/tombstone_scan.rs new file mode 100644 index 0000000000..d342ed3e38 --- /dev/null +++ b/desktop/src-tauri/src/managed_agents/tombstone_scan.rs @@ -0,0 +1,234 @@ +//! The exhaustive owner kind:5 scan (V3.3), and the evidence it yields. +//! +//! [`TombstoneEvidence::NotFound`] is a claim that the owner never deleted a +//! coordinate. Under V3.11 that claim is what lets a genuinely new record +//! publish against a provably empty coordinate, and under V3.10 its opposite is +//! the only thing that authorizes deleting a local record. Both directions are +//! consequential, so the claim may only come from a scan that ran to +//! completion — anything less reports [`TombstoneEvidence::ScanIncomplete`], +//! which decides nothing. +//! +//! # Why the scan is exhaustive rather than filtered +//! +//! The obvious query is one `#a`-filtered lookup per coordinate. The relay +//! cannot serve it soundly: `#a` is not a pushed-down constraint +//! (`filter_fully_pushable` in `crates/buzz-relay/src/handlers/req.rs` lists +//! `#a` among the post-filtered keys), so the server applies it AFTER the SQL +//! `LIMIT`. A page that fills with 1000 unrelated kind:5 events returns empty +//! for the coordinate asked about — indistinguishable from no tombstone +//! existing. So the scan pulls the owner's kind:5 history by pages, filters +//! `a`-tags locally, and only reports absence once a page comes back short of +//! its own limit. +//! +//! # Why one scan for every coordinate +//! +//! The scan is per-OWNER, not per-coordinate: one paginated walk yields the +//! evidence for every coordinate the boot pass is about to decide. A +//! per-coordinate scan would multiply the same pages by the number of records +//! on the device for an identical answer. + +use std::collections::BTreeSet; + +use serde_json::json; + +use super::{decision::TombstoneEvidence, head_lookup::Coordinate, retention::RetentionScope}; +use crate::app_state::AppState; + +/// Page size for the scan. The relay clamps any request to 1000 +/// (`crates/buzz-db/src/event.rs`), so asking for more would silently get 1000 +/// and make a full page look short. +const SCAN_PAGE_LIMIT: usize = 1000; + +/// Hard bound on pages walked in one boot, so a pathological history cannot +/// stall startup indefinitely. Hitting it yields an INCOMPLETE scan, never a +/// premature `NotFound`. +const MAX_SCAN_PAGES: usize = 20; + +/// Which coordinates the owner has tombstoned, from a scan that either ran to +/// completion or did not. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct TombstonedCoordinates { + tombstoned: BTreeSet<(u32, String)>, + complete: bool, +} + +impl TombstonedCoordinates { + /// The result of a scan that did not finish. Every lookup against it + /// reports [`TombstoneEvidence::ScanIncomplete`], even for coordinates the + /// partial results did name — a partial scan cannot support `NotFound`, and + /// reporting `Found` from it would make the evidence's meaning depend on + /// which page a tombstone happened to land on. + pub fn incomplete() -> Self { + Self { + tombstoned: BTreeSet::new(), + complete: false, + } + } + + /// The evidence this scan provides about one coordinate. + pub fn evidence(&self, coordinate: &Coordinate) -> TombstoneEvidence { + if !self.complete { + return TombstoneEvidence::ScanIncomplete; + } + if self + .tombstoned + .contains(&(coordinate.kind, coordinate.d_tag.clone())) + { + TombstoneEvidence::Found + } else { + TombstoneEvidence::NotFound + } + } +} + +/// One page of the owner's kind:5 history, oldest-bound by a keyset cursor. +/// +/// # Known limitation +/// +/// On relay deployments that route historical reads through a read-replica +/// pool, a replica that has not yet replayed a tombstone answers "not found" +/// in the dangerous direction. The barrier's tracing line surfaces this if +/// it ever fires; re-adding `sync_authoritative` to the protocol is a small +/// standalone change reversible on observed evidence. +pub fn scan_page_filter(owner_pubkey_hex: &str, cursor: Option<&PageCursor>) -> serde_json::Value { + let mut filter = json!({ + "kinds": [buzz_core_pkg::kind::KIND_DELETION], + "authors": [owner_pubkey_hex], + "limit": SCAN_PAGE_LIMIT, + }); + if let Some(cursor) = cursor { + // Composite keyset cursor: the relay requires both halves or neither, + // and rejects `before_id` without `until`. A timestamp-only cursor + // would drop every event tied at the page boundary's second. + filter["until"] = json!(cursor.until); + filter["before_id"] = json!(cursor.before_id); + } + filter +} + +/// The `(created_at, id)` pair that bounds the next page. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct PageCursor { + pub until: i64, + pub before_id: String, +} + +/// The cursor for the page after `events`, or `None` when this page ends the +/// walk. +/// +/// A page shorter than the limit it asked for is the ONLY proof of exhaustion. +/// `rows == limit` never proves there is more, but it never proves there isn't +/// either, so the walk continues until a page comes back short. +pub fn next_cursor(events: &[nostr::Event]) -> Option { + if events.len() < SCAN_PAGE_LIMIT { + return None; + } + let last = events.last()?; + Some(PageCursor { + until: last.created_at.as_secs() as i64, + before_id: last.id.to_hex(), + }) +} + +/// The config coordinates a page of kind:5 events tombstones. +/// +/// `a`-tags are parsed locally with the same owner-scoping rule the inbound +/// tombstone path uses: only the coordinate's own author may tombstone it, so +/// a validly signed kind:5 naming another owner's coordinate names nothing +/// here. Coordinates outside the three config kinds are ignored — the owner's +/// kind:5 history includes deletions this pass has no opinion about. +pub fn tombstoned_in_page(events: &[nostr::Event]) -> BTreeSet<(u32, String)> { + use super::canonical_projection::CanonicalProjection; + + events + .iter() + .filter_map(|event| { + let (kind, d_tag) = deletion_coordinate(event)?; + CanonicalProjection::is_config_kind(kind).then_some((kind, d_tag)) + }) + .collect() +} + +/// Parse a NIP-09 `a`-tag coordinate `::` into its +/// target kind and d-tag, rejecting a coordinate the event's author does not +/// own. +fn deletion_coordinate(event: &nostr::Event) -> Option<(u32, String)> { + event.tags.iter().find_map(|tag| { + let values = tag.as_slice(); + if values.first().map(|v| v.as_str()) != Some("a") { + return None; + } + // `::` — the d-tag may itself contain ':', so split + // at most twice and keep the remainder whole. + let mut parts = values.get(1)?.splitn(3, ':'); + let kind: u32 = parts.next()?.parse().ok()?; + if parts.next()? != event.pubkey.to_hex() { + return None; + } + Some((kind, parts.next()?.to_string())) + }) +} + +/// Walk the owner's kind:5 history to exhaustion. +/// +/// Every failure mode — a transport error, a page that never ends — yields +/// [`TombstonedCoordinates::incomplete`], so a scan that could not prove +/// absence never claims it. +pub async fn scan_tombstones( + state: &AppState, + scope: &RetentionScope, + api_base_url: &str, +) -> TombstonedCoordinates { + let owner_hex = scope.owner_keys.public_key().to_hex(); + let mut tombstoned = BTreeSet::new(); + let mut cursor: Option = None; + + for _ in 0..MAX_SCAN_PAGES { + let filter = scan_page_filter(&owner_hex, cursor.as_ref()); + let events = match crate::relay::query_relay_at_with_keys( + state, + api_base_url, + std::slice::from_ref(&filter), + &scope.owner_keys, + None, + ) + .await + { + Ok(events) => events, + Err(error) => { + tracing::warn!( + target: "buzz::config_sync", + %error, + "tombstone scan page failed; evidence is incomplete" + ); + return TombstonedCoordinates::incomplete(); + } + }; + + tombstoned.extend(tombstoned_in_page(&events)); + match next_cursor(&events) { + Some(next) => cursor = Some(next), + None => { + tracing::info!( + target: "buzz::config_sync", + tombstoned = tombstoned.len(), + "tombstone scan complete" + ); + return TombstonedCoordinates { + tombstoned, + complete: true, + }; + } + } + } + + tracing::warn!( + target: "buzz::config_sync", + max_pages = MAX_SCAN_PAGES, + "tombstone scan hit its page bound; evidence is incomplete" + ); + TombstonedCoordinates::incomplete() +} + +#[cfg(test)] +mod tests; diff --git a/desktop/src-tauri/src/managed_agents/tombstone_scan/tests.rs b/desktop/src-tauri/src/managed_agents/tombstone_scan/tests.rs new file mode 100644 index 0000000000..7209022cd9 --- /dev/null +++ b/desktop/src-tauri/src/managed_agents/tombstone_scan/tests.rs @@ -0,0 +1,224 @@ +use super::*; +use buzz_core_pkg::kind::{KIND_DELETION, KIND_MANAGED_AGENT, KIND_PERSONA, KIND_TEAM}; +use nostr::{EventBuilder, Keys, Kind, Tag}; + +/// A signed kind:5 carrying one `a`-tag, with control over whose coordinate it +/// names — the owner-scoping rule is what makes a foreign tombstone inert. +fn deletion_of(keys: &Keys, coordinate_owner_hex: &str, kind: u32, d_tag: &str) -> nostr::Event { + let coord = format!("{kind}:{coordinate_owner_hex}:{d_tag}"); + EventBuilder::new(Kind::Custom(KIND_DELETION as u16), "") + .tags(vec![Tag::parse(["a", coord.as_str()]).unwrap()]) + .sign_with_keys(keys) + .unwrap() +} + +/// A kind:5 deleting by event id instead of coordinate. It names no +/// parameterized coordinate at all. +fn deletion_by_event_id(keys: &Keys) -> nostr::Event { + EventBuilder::new(Kind::Custom(KIND_DELETION as u16), "") + .tags(vec![Tag::parse(["e", &"ab".repeat(32)]).unwrap()]) + .sign_with_keys(keys) + .unwrap() +} + +fn coordinate(kind: u32, d_tag: &str) -> Coordinate { + Coordinate { + kind, + d_tag: d_tag.to_string(), + } +} + +// ── The evidence contract ────────────────────────────────────────────────── + +/// An incomplete scan decides nothing. Reporting `Found` from partial results +/// would make the evidence's meaning depend on which page a tombstone landed +/// on, and reporting `NotFound` would be the unsound direction outright. +#[test] +fn test_an_incomplete_scan_reports_no_evidence_for_any_coordinate() { + let scan = TombstonedCoordinates::incomplete(); + + assert_eq!( + scan.evidence(&coordinate(KIND_PERSONA, "anything")), + TombstoneEvidence::ScanIncomplete + ); +} + +#[test] +fn test_a_complete_scan_distinguishes_tombstoned_from_untouched() { + let keys = Keys::generate(); + let owner = keys.public_key().to_hex(); + let events = [deletion_of(&keys, &owner, KIND_PERSONA, "deleted")]; + + let scan = TombstonedCoordinates { + tombstoned: tombstoned_in_page(&events), + complete: true, + }; + + assert_eq!( + scan.evidence(&coordinate(KIND_PERSONA, "deleted")), + TombstoneEvidence::Found + ); + assert_eq!( + scan.evidence(&coordinate(KIND_PERSONA, "never-deleted")), + TombstoneEvidence::NotFound + ); + // Same d-tag under a different kind is a different coordinate. + assert_eq!( + scan.evidence(&coordinate(KIND_TEAM, "deleted")), + TombstoneEvidence::NotFound + ); +} + +// ── Parsing a page into coordinates ──────────────────────────────────────── + +/// Only the coordinate's own author may tombstone it. A validly signed kind:5 +/// naming someone else's coordinate is inert, or one owner could suppress +/// another's config by publishing a deletion for it. +#[test] +fn test_a_tombstone_for_another_owners_coordinate_names_nothing() { + let keys = Keys::generate(); + let stranger = Keys::generate().public_key().to_hex(); + + let tombstoned = tombstoned_in_page(&[deletion_of(&keys, &stranger, KIND_PERSONA, "victim")]); + + assert!( + tombstoned.is_empty(), + "a foreign coordinate must not be treated as deleted: {tombstoned:?}" + ); +} + +/// The owner's kind:5 history covers deletions this pass has no opinion about. +#[test] +fn test_deletions_outside_the_config_kinds_are_ignored() { + let keys = Keys::generate(); + let owner = keys.public_key().to_hex(); + + let tombstoned = tombstoned_in_page(&[ + deletion_of(&keys, &owner, 30023, "some-article"), + deletion_of(&keys, &owner, KIND_MANAGED_AGENT, "agent"), + ]); + + assert_eq!( + tombstoned, + BTreeSet::from([(KIND_MANAGED_AGENT, "agent".to_string())]) + ); +} + +/// A d-tag may itself contain `:`, so the coordinate splits at most twice and +/// keeps the remainder whole. Truncating here would report evidence about a +/// coordinate that does not exist while missing the one that does. +#[test] +fn test_a_d_tag_containing_colons_survives_parsing() { + let keys = Keys::generate(); + let owner = keys.public_key().to_hex(); + + let tombstoned = tombstoned_in_page(&[deletion_of(&keys, &owner, KIND_PERSONA, "a:b:c")]); + + assert_eq!( + tombstoned, + BTreeSet::from([(KIND_PERSONA, "a:b:c".to_string())]) + ); +} + +#[test] +fn test_an_event_id_deletion_names_no_coordinate() { + let keys = Keys::generate(); + + assert!(tombstoned_in_page(&[deletion_by_event_id(&keys)]).is_empty()); +} + +// ── Pagination ───────────────────────────────────────────────────────────── + +/// A short page is the ONLY proof the walk is exhausted. +#[test] +fn test_a_short_page_ends_the_walk() { + let keys = Keys::generate(); + let owner = keys.public_key().to_hex(); + + assert_eq!( + next_cursor(&[deletion_of(&keys, &owner, KIND_PERSONA, "one")]), + None + ); + assert_eq!(next_cursor(&[]), None); +} + +/// A page that exactly fills its limit proves nothing about exhaustion, so the +/// walk continues from the oldest event on it. +#[test] +fn test_a_full_page_continues_from_its_oldest_event() { + let keys = Keys::generate(); + let owner = keys.public_key().to_hex(); + let filler = deletion_of(&keys, &owner, KIND_PERSONA, "filler"); + let oldest = deletion_of(&keys, &owner, KIND_PERSONA, "oldest"); + + let mut page = vec![filler; SCAN_PAGE_LIMIT - 1]; + page.push(oldest.clone()); + + assert_eq!( + next_cursor(&page), + Some(PageCursor { + until: oldest.created_at.as_secs() as i64, + before_id: oldest.id.to_hex(), + }), + "the cursor must be the composite keyset of the page's last event" + ); +} + +// ── The page filter ──────────────────────────────────────────────────────── + +/// The page filter names the owner's kind:5 events with the SCAN_PAGE_LIMIT +/// cap. The scan is per-owner (not per-coordinate) so one walk covers every +/// coordinate the barrier is deciding on. +#[test] +fn test_page_filter_covers_owner_deletions() { + let with_cursor = scan_page_filter( + "ownerhex", + Some(&PageCursor { + until: 100, + before_id: "abc".to_string(), + }), + ); + + for filter in [scan_page_filter("ownerhex", None), with_cursor] { + assert_eq!(filter["kinds"], serde_json::json!([KIND_DELETION])); + assert_eq!(filter["authors"], serde_json::json!(["ownerhex"])); + } +} + +/// The relay rejects `before_id` without `until` and drops events tied at the +/// boundary second given `until` alone, so the cursor is all-or-nothing. +#[test] +fn test_the_cursor_halves_are_both_present_or_both_absent() { + let first = scan_page_filter("ownerhex", None); + assert!(first.get("until").is_none()); + assert!(first.get("before_id").is_none()); + + let next = scan_page_filter( + "ownerhex", + Some(&PageCursor { + until: 100, + before_id: "abc".to_string(), + }), + ); + assert_eq!(next["until"], serde_json::json!(100)); + assert_eq!(next["before_id"], serde_json::json!("abc")); +} + +/// Asking for more than the relay's clamp would silently get the clamp back +/// and make a full page look short — which is exactly the state that ends the +/// walk and licenses a `NotFound`. +#[test] +fn test_the_page_limit_does_not_exceed_the_relay_clamp() { + /// The relay's own clamp (`crates/buzz-db/src/event.rs`). + const RELAY_CLAMP: u64 = 1000; + + let limit = scan_page_filter("ownerhex", None)["limit"] + .as_u64() + .expect("the filter must carry a numeric limit"); + + assert_eq!(limit, SCAN_PAGE_LIMIT as u64); + assert!( + limit <= RELAY_CLAMP, + "asking for {limit} would silently get {RELAY_CLAMP} back and make a full page look short" + ); +}