diff --git a/desktop/src-tauri/src/commands/personas/pending.rs b/desktop/src-tauri/src/commands/personas/pending.rs index 171a9d6766..b46ac8e799 100644 --- a/desktop/src-tauri/src/commands/personas/pending.rs +++ b/desktop/src-tauri/src/commands/personas/pending.rs @@ -12,6 +12,7 @@ use crate::managed_agents::{ pub(super) struct PreparedPersonaPublication { pub scope: RetentionScope, + #[allow(dead_code)] // event is retained in the DB; the flush loop publishes it pub event: nostr::Event, pub retained: RetainedEvent, pub persona: AgentDefinition, @@ -153,7 +154,9 @@ pub(super) fn prepare_persona_publication_at( ) -> Result<(nostr::Event, RetainedEvent, AgentDefinition), String> { use crate::managed_agents::{ persona_events::{build_persona_event, monotonic_created_at, persona_d_tag}, - retention::{get_retained_event, open_retention_db, retain_event, RetainedEvent}, + retention::{ + get_retained_event, open_retention_db, retain_user_intent_event, RetainedEvent, + }, }; use buzz_core_pkg::kind::KIND_PERSONA; @@ -171,7 +174,9 @@ pub(super) fn prepare_persona_publication_at( .sign_with_keys(keys) .map_err(|e| format!("failed to sign persona event: {e}"))?; let retained = RetainedEvent::pending(KIND_PERSONA, pubkey, d_tag, &event); - retain_event(&conn, &retained)?; + // User intent: must clear publish_blocked so an edit to a parked/suppressed + // coordinate takes effect in the current session, not just the next boot. + retain_user_intent_event(&conn, &retained)?; Ok((event, retained, scoped_persona)) } @@ -195,8 +200,8 @@ pub(in crate::commands) fn tombstone_persona_pending( use crate::managed_agents::{ persona_events::build_persona_delete, retention::{ - delete_retained_event, open_retention_db, retain_event, tombstone_retention_d_tag, - RetainedEvent, + delete_retained_event, open_retention_db, retain_user_intent_event, + tombstone_retention_d_tag, RetainedEvent, }, }; use buzz_core_pkg::kind::KIND_PERSONA; @@ -211,9 +216,13 @@ pub(in crate::commands) fn tombstone_persona_pending( .map_err(|e| format!("failed to sign persona tombstone: {e}"))?; let conn = open_retention_db(&scope.db_path)?; // Purge the persona row first so an unpublished edit can never resurrect - // it after the tombstone publishes. + // it after the tombstone publishes. The persona_baselines entry is NOT + // deleted — it must survive so the next boot sees baseline=Present + queued + // deletion → Gate C arbitration, not baseline=None → park-forever. delete_retained_event(&conn, KIND_PERSONA, &pubkey, d_tag)?; - retain_event( + // Tombstone is user intent: clear publish_blocked so an offline delete that + // gets parked on a prior boot can propagate in the current session. + retain_user_intent_event( &conn, &RetainedEvent::pending( KIND_DELETE, diff --git a/desktop/src-tauri/src/commands/personas/sharing.rs b/desktop/src-tauri/src/commands/personas/sharing.rs index 914c56252d..b7392d2949 100644 --- a/desktop/src-tauri/src/commands/personas/sharing.rs +++ b/desktop/src-tauri/src/commands/personas/sharing.rs @@ -4,7 +4,7 @@ use crate::{ app_state::AppState, managed_agents::{ load_personas, - retention::{mark_synced, open_retention_db}, + retention::{get_retained_event, open_retention_db}, AgentDefinition, }, }; @@ -60,7 +60,7 @@ pub async fn set_persona_shared( .map_err(|e| format!("spawn_blocking failed: {e}"))??; let state = app.state::(); - publish_prepared_persona(&state, prepared).await + publish_prepared_persona_via_flush(&state, prepared).await } /// Save a persona edit AND publish its catalog head, returning the same @@ -69,8 +69,8 @@ pub async fn set_persona_shared( /// The "save and publish" affordance in the edit dialog promises the change /// reaches the catalog on save. Plain `update_persona` only enqueues /// best-effort, so the UI could not report whether the relay accepted it. This -/// takes the identical input and reuses the strict preparation path, then awaits -/// the relay exactly like the share toggle does — a rejection or an unreachable +/// takes the identical input and reuses the strict preparation path, then +/// routes through the single gated publisher — a rejection or an unreachable /// relay stays durably queued for the flush loop and is reported as `queued`. #[tauri::command] pub async fn update_persona_and_publish( @@ -87,44 +87,73 @@ pub async fn update_persona_and_publish( .await?; let state = app.state::(); - publish_prepared_persona(&state, prepared).await + publish_prepared_persona_via_flush(&state, prepared).await } -async fn publish_prepared_persona( +/// Retain the prepared event (already done by `prepare_persona_publication` +/// as user intent, gate cleared) then trigger a flush pass and await the +/// relay outcome. +/// +/// This is the single publish path — it goes through the gated flush loop, +/// which checks `publish_blocked` and the readiness latch. Routing here +/// instead of submitting directly ensures: +/// +/// 1. A parked/suppressed coordinate cannot bypass its own gate. +/// 2. Baseline provenance is stamped on relay acceptance (via +/// `mark_synced_and_stamp_baseline` inside the flush loop). +/// 3. The `Queued` report means the row is retained and will flush on the +/// next sweep — not "could not even enqueue." +async fn publish_prepared_persona_via_flush( state: &AppState, prepared: PreparedPersonaPublication, ) -> Result { - let api_base_url = crate::relay::relay_http_base_url(&prepared.scope.relay_url); - let publish_result = crate::relay::submit_signed_event_at_with_keys( - &prepared.event, + use crate::managed_agents::persona_events::flush_pending_events_at; + + // Run a targeted flush for this scope. The event is already retained as + // user intent (publish_blocked = 0), so it will be in the pending queue. + let flushed = flush_pending_events_at( + &prepared.scope.db_path, state, - &api_base_url, + &prepared.scope.relay_url, &prepared.scope.owner_keys, ) - .await; + .await + .unwrap_or(0); - match publish_result { - Ok(_) => { - let conn = open_retention_db(&prepared.scope.db_path)?; - mark_synced( - &conn, - prepared.retained.kind, - &prepared.retained.pubkey, - &prepared.retained.d_tag, - prepared.retained.created_at, - &prepared.retained.content, - )?; - Ok(SetPersonaSharedResult { - persona: prepared.persona, - publication_status: PersonaSharePublicationStatus::Published, - relay_message: None, - }) - } - Err(error) => Ok(SetPersonaSharedResult { + // Determine whether the specific row published: re-read pending_sync. + // If flushed > 0 and the row is no longer pending, it published. + // If the relay rejected/was unreachable, pending_sync stays 1. + let conn = open_retention_db(&prepared.scope.db_path)?; + let row = get_retained_event( + &conn, + prepared.retained.kind, + &prepared.retained.pubkey, + &prepared.retained.d_tag, + )?; + + let published = row.as_ref().is_some_and(|r| { + // The row published if pending_sync is now 0 AND it still carries the + // same content (a newer edit would also have pending_sync=0 only if it + // too published, which is fine — we just report what actually happened). + !r.pending_sync + && r.created_at == prepared.retained.created_at + && r.content == prepared.retained.content + }) || flushed > 0 && row.as_ref().is_none_or(|r| !r.pending_sync); + + if published { + Ok(SetPersonaSharedResult { + persona: prepared.persona, + publication_status: PersonaSharePublicationStatus::Published, + relay_message: None, + }) + } else { + Ok(SetPersonaSharedResult { persona: prepared.persona, publication_status: PersonaSharePublicationStatus::Queued, - relay_message: Some(error), - }), + relay_message: Some( + "event queued for publication; relay unreachable or not yet ready".to_string(), + ), + }) } } @@ -218,7 +247,9 @@ mod tests { let prepared = prepared(&db_path, spawn_relay(false).await, keys, Some(true)); let state = build_app_state(); - let result = publish_prepared_persona(&state, prepared).await.unwrap(); + let result = publish_prepared_persona_via_flush(&state, prepared) + .await + .unwrap(); assert_eq!( result.publication_status, @@ -227,7 +258,7 @@ mod tests { assert!(result .relay_message .as_deref() - .is_some_and(|message| message.contains("relay rejected event"))); + .is_some_and(|message| message.contains("queued"))); assert!( get_retained_event( &open_retention_db(&db_path).unwrap(), @@ -253,7 +284,9 @@ mod tests { let prepared = prepared(&db_path, relay_url, keys, Some(true)); let state = build_app_state(); - let result = publish_prepared_persona(&state, prepared).await.unwrap(); + let result = publish_prepared_persona_via_flush(&state, prepared) + .await + .unwrap(); assert_eq!( result.publication_status, @@ -262,7 +295,7 @@ mod tests { assert!(result .relay_message .as_deref() - .is_some_and(|message| message.starts_with("relay unreachable:"))); + .is_some_and(|message| message.contains("queued"))); assert!( get_retained_event( &open_retention_db(&db_path).unwrap(), @@ -285,7 +318,9 @@ mod tests { let prepared = prepared(&db_path, spawn_relay(true).await, keys, Some(true)); let state = build_app_state(); - let result = publish_prepared_persona(&state, prepared).await.unwrap(); + let result = publish_prepared_persona_via_flush(&state, prepared) + .await + .unwrap(); assert_eq!( result.publication_status, @@ -318,7 +353,9 @@ mod tests { let prepared = prepared(&db_path, spawn_relay(true).await, keys, None); let state = build_app_state(); - let result = publish_prepared_persona(&state, prepared).await.unwrap(); + let result = publish_prepared_persona_via_flush(&state, prepared) + .await + .unwrap(); assert_eq!( result.publication_status, @@ -351,7 +388,9 @@ mod tests { let prepared = prepared(&db_path, spawn_relay(false).await, keys, None); let state = build_app_state(); - let result = publish_prepared_persona(&state, prepared).await.unwrap(); + let result = publish_prepared_persona_via_flush(&state, prepared) + .await + .unwrap(); assert_eq!( result.publication_status, @@ -360,7 +399,7 @@ mod tests { assert!(result .relay_message .as_deref() - .is_some_and(|message| message.contains("relay rejected event"))); + .is_some_and(|message| message.contains("queued"))); assert!( get_retained_event( &open_retention_db(&db_path).unwrap(), diff --git a/desktop/src-tauri/src/managed_agents/config_barrier.rs b/desktop/src-tauri/src/managed_agents/config_barrier.rs index 2295b4ec27..ae86978e17 100644 --- a/desktop/src-tauri/src/managed_agents/config_barrier.rs +++ b/desktop/src-tauri/src/managed_agents/config_barrier.rs @@ -279,6 +279,13 @@ async fn run_boot_barrier_for_scope( /// generation is set to `InProgress` and its own barrier — or a full flush /// retry — supplies the authoritative gate writes; this barrier returns /// `Abandoned` without writing. +/// +/// Returns `Abandoned` (dropping the claim) if any gate was driven by +/// incomplete evidence (lookup or scan failure). The 30s flush retry will +/// re-run the full transition once the relay is reachable again. This is the +/// fail-closed path for a wifi-race boot: all coordinates defer, but the scope +/// does NOT advance to `Ready`, so the retry machinery fires instead of the +/// session staying muted forever. pub(crate) fn enforce_decision_pass( claim: &super::config_sync_readiness::ReadinessClaim, conn: &rusqlite::Connection, @@ -293,7 +300,19 @@ pub(crate) fn enforce_decision_pass( return Ok(BarrierOutcome::Abandoned); } - run_decision_pass(conn, owner_pubkey, states)?; + let (_plan, has_uncertain_gates) = run_decision_pass(conn, owner_pubkey, states)?; + + if has_uncertain_gates { + tracing::warn!( + target: "buzz::config_sync", + "boot barrier: some coordinates deferred due to lookup/scan failures; \ + scope stays closed for retry (wifi-race or offline boot)" + ); + // Return Abandoned so the caller drops the claim → latch resets to + // Unready → the 30s flush retry re-runs the full transition. + return Ok(BarrierOutcome::Abandoned); + } + Ok(BarrierOutcome::Success) } diff --git a/desktop/src-tauri/src/managed_agents/config_sync.rs b/desktop/src-tauri/src/managed_agents/config_sync.rs index 3c0c526828..51388b1709 100644 --- a/desktop/src-tauri/src/managed_agents/config_sync.rs +++ b/desktop/src-tauri/src/managed_agents/config_sync.rs @@ -25,7 +25,7 @@ use super::{ canonical_projection::CanonicalProjection, decision::{Decision, Head, HeadState, Observation, Resolution, TombstoneEvidence}, head_lookup::Coordinate, - retention::{get_baseline, get_retained_event, RetainedEvent}, + retention::{get_baseline_full, get_retained_event, RetainedEvent}, }; use rusqlite::Connection; @@ -68,7 +68,7 @@ pub fn local_state( disk, queued: queued_projection(row.as_ref()), head, - baseline: baseline_projection(conn, owner_pubkey, coordinate, row.as_ref())?, + baseline: baseline_projection(conn, owner_pubkey, coordinate)?, queued_deletion_at: tombstone_row .as_ref() .filter(|row| row.pending_sync) @@ -95,38 +95,22 @@ fn queued_projection(row: Option<&RetainedEvent>) -> Option /// 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. +/// Read from the `persona_baselines` table, which stores the full canonical +/// projection including `shared`. The baseline survives live-row deletion, +/// so an offline delete can still present a baseline on the next boot and +/// reach gate C's timestamp arbitration rather than parking forever. 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)? + let Some((_, content, shared)) = + get_baseline_full(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, - })) + Ok(Some(CanonicalProjection { content, shared })) } /// Resolve every gathered coordinate, pairing each resolution with its target. @@ -284,16 +268,43 @@ fn stamp_baseline_from_head( coordinate: &Coordinate, head: &Head, ) -> Result<(), String> { - super::retention::set_baseline( + super::retention::set_baseline_full( conn, coordinate.kind, owner_pubkey, &coordinate.d_tag, &head.event_id, &head.projection.content, + head.projection.shared, ) } +/// Whether a decision was driven by incomplete evidence (lookup or scan failure) +/// rather than by a successful read. +/// +/// A pass that contains any uncertain decision must not certify the scope +/// `Ready`: the coordinate that deferred due to failure would stay muted for +/// the entire session if the latch advanced to `Ready`. Dropping the claim +/// instead lets the 30s flush retry re-run the full transition once the +/// network recovers. +/// +/// Succesful parks (`LookupFailed`-free) are NOT uncertain — a completed +/// lookup that returned `Absent` or a completed scan that returned `NotFound` +/// produced a genuine decision. Only a decision driven by a lookup or scan +/// that did not complete is uncertain. +fn decision_is_uncertain(resolution: &Resolution, observation: &Observation) -> bool { + use crate::managed_agents::decision::TombstoneEvidence; + match &resolution.decision { + // Defer is the only decision the table emits when a lookup or scan + // failed to complete. Any other decision came from completed reads. + Decision::Defer => { + matches!(&observation.head, HeadState::LookupFailed) + || observation.tombstone == TombstoneEvidence::ScanIncomplete + } + _ => false, + } +} + /// 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 @@ -308,14 +319,18 @@ fn stamp_baseline_from_head( /// 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. +/// Returns the plan it enforced and whether any gate was driven by uncertain +/// evidence (lookup/scan failure). The caller uses the uncertainty flag to +/// decide whether the barrier should certify the scope `Ready` — a pass with +/// uncertain gates must not certify, so a later retry can see the relay once +/// it becomes reachable. pub fn run_decision_pass( conn: &Connection, owner_pubkey: &str, states: &[CoordinateState], -) -> Result, String> { +) -> Result<(Vec<(Coordinate, Resolution)>, bool), String> { let plan = plan(states); + let mut has_uncertain_gates = false; for (state, (coordinate, resolution)) in states.iter().zip(plan.iter()) { apply_gate( @@ -326,6 +341,10 @@ pub fn run_decision_pass( resolution, )?; + if decision_is_uncertain(resolution, &state.observation) { + has_uncertain_gates = true; + } + // 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. @@ -363,11 +382,14 @@ pub fn run_decision_pass( coordinates = plan.len(), blocked, cleared, + uncertain = has_uncertain_gates, "config-sync decision pass complete" ); - Ok(plan) + Ok((plan, has_uncertain_gates)) } +#[cfg(test)] +mod regression_tests; #[cfg(test)] mod tests; diff --git a/desktop/src-tauri/src/managed_agents/config_sync/regression_tests.rs b/desktop/src-tauri/src/managed_agents/config_sync/regression_tests.rs new file mode 100644 index 0000000000..49c7ab1bc6 --- /dev/null +++ b/desktop/src-tauri/src/managed_agents/config_sync/regression_tests.rs @@ -0,0 +1,389 @@ +//! Regression tests for am#1, peon-P1#1/P1#2, am#2/peon-P2#1. +//! +//! Kept separate from `tests.rs` to respect the 1000-line ceiling on new files. + +use std::path::Path; + +use super::*; +use crate::managed_agents::{ + decision::{Decision, HeadState, TombstoneEvidence}, + persona_events::build_persona_event, + retention::{ + delete_retained_event, get_baseline, get_retained_event, is_publish_blocked, + mark_synced_and_stamp_baseline, open_retention_db, retain_event, retain_user_intent_event, + set_baseline, set_publish_blocked, tombstone_retention_d_tag, RetainedEvent, + }, +}; +use buzz_core_pkg::kind::{KIND_DELETION, KIND_PERSONA}; +use rusqlite::Connection; + +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(), + } +} + +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() +} + +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()), + publish_blocked: false, + }, + ) + .unwrap(); + event +} + +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(); +} + +fn run_pass(conn: &Connection, observation: &Observation) -> (Decision, usize) { + use crate::managed_agents::retention::get_pending_sync; + 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(), + ) +} + +// ── (a) am#1 / peon-P1#1: publish-confirm stamps baseline so next boot +// suppresses revert. +// +// Mutation-sensitive: removing `mark_synced_and_stamp_baseline`'s baseline +// write causes queued != baseline and the test fails on the ApplyHead assert. +#[test] +fn test_publish_confirm_stamps_baseline_so_next_boot_suppresses_revert() { + let conn = test_db(); + const D_TAG: &str = "test-persona"; + + // Retain v2 as a queued edit (simulates what reconcile produced at boot). + let v2_event = retain_prompt(&conn, KIND_PERSONA, D_TAG, true, "v2-prompt"); + let v2_id = v2_event.id.to_hex(); + let v2_content = v2_event.content.to_string(); + + // Simulate the flush loop accepting v2: mark synced + stamp baseline. + mark_synced_and_stamp_baseline( + &conn, + KIND_PERSONA, + OWNER, + D_TAG, + v2_event.created_at.as_secs() as i64, + &v2_content, + &v2_id, + false, + ) + .unwrap(); + + let baseline = get_baseline(&conn, KIND_PERSONA, OWNER, D_TAG) + .unwrap() + .expect("baseline must be stamped after publish-confirm"); + assert_eq!( + baseline.0, v2_id, + "baseline event_id must be the published event" + ); + assert_eq!( + baseline.1, v2_content, + "baseline content must be the published content" + ); + + // Second boot: reconcile re-queued v2 (disk still has it). + set_pending(&conn, KIND_PERSONA, D_TAG); + + // The relay now has v3 (a newer edit from the other device). + let v3_head = Head { + event_id: "id-v3".to_string(), + created_at: v2_event.created_at.as_secs() as i64 + 1, + projection: CanonicalProjection { + content: "v3-content-json".to_string(), + shared: false, + }, + }; + // Build the observation directly from the real DB so queued == baseline. + let observation = local_state( + &conn, + OWNER, + &Coordinate { + kind: KIND_PERSONA, + d_tag: D_TAG.to_string(), + }, + Some(CanonicalProjection { + content: v2_content.clone(), + shared: false, + }), + HeadState::Present(v3_head), + TombstoneEvidence::NotFound, + ) + .unwrap(); + + let q = observation.queued.as_ref().expect("queued must be Some"); + let b = observation + .baseline + .as_ref() + .expect("baseline must be Some"); + assert_eq!(q, b, "queued == baseline → vacuous ApplyHead"); + + let (decision, publishable) = run_pass(&conn, &observation); + assert_eq!( + decision, + Decision::ApplyHead, + "must adopt v3, not publish stale v2" + ); + assert_eq!(publishable, 0, "no pending rows after vacuous clear"); +} + +// ── (b) peon-P1#2: offline delete preserves baseline so next boot reaches +// Gate C (timestamp arbitration), not park-forever. +// +// Mutation-sensitive: if `delete_retained_event` also deleted the +// `persona_baselines` row, observation.baseline would be None and decide() +// would return Park(NoBaselineWithHead) instead of PublishDeletion. +#[test] +fn test_offline_delete_preserves_baseline_so_next_boot_reaches_gate_c() { + let conn = test_db(); + const D_TAG: &str = "test-persona"; + + let live_event = retain_prompt(&conn, KIND_PERSONA, D_TAG, false, "original-prompt"); + let live_id = live_event.id.to_hex(); + let live_content = live_event.content.to_string(); + set_baseline(&conn, KIND_PERSONA, OWNER, D_TAG, &live_id, &live_content).unwrap(); + + // Simulate `tombstone_persona_pending`: purge live row, retain tombstone. + delete_retained_event(&conn, KIND_PERSONA, OWNER, D_TAG).unwrap(); + let tombstone_d = tombstone_retention_d_tag(KIND_PERSONA, D_TAG); + let tombstone_event = signed_persona("tombstone"); + retain_user_intent_event( + &conn, + &RetainedEvent { + kind: KIND_DELETION, + pubkey: OWNER.to_string(), + d_tag: tombstone_d.clone(), + content: tombstone_event.content.to_string(), + created_at: tombstone_event.created_at.as_secs() as i64 + 1, + raw_event: { + use nostr::JsonUtil; + tombstone_event.as_json() + }, + event_id: Some(tombstone_event.id.to_hex()), + pending_sync: true, + publish_blocked: false, + }, + ) + .unwrap(); + + // Baseline must survive the live-row deletion. + assert!( + get_baseline(&conn, KIND_PERSONA, OWNER, D_TAG) + .unwrap() + .is_some(), + "baseline must survive delete_retained_event (lives in persona_baselines)" + ); + assert!( + get_retained_event(&conn, KIND_PERSONA, OWNER, D_TAG) + .unwrap() + .is_none(), + "live row must be purged" + ); + + // Next boot: relay still shows the live head (delete didn't reach it). + let observation = local_state( + &conn, + OWNER, + &coordinate(), + Some(CanonicalProjection { + content: live_content.clone(), + shared: false, + }), + HeadState::Present(Head { + event_id: live_id.clone(), + created_at: live_event.created_at.as_secs() as i64, + projection: CanonicalProjection { + content: live_content.clone(), + shared: false, + }, + }), + TombstoneEvidence::NotFound, + ) + .unwrap(); + + assert!( + observation.baseline.is_some(), + "baseline present → avoids park cell" + ); + assert!( + observation.queued_deletion_at.is_some(), + "tombstone visible" + ); + let (decision, _) = run_pass(&conn, &observation); + assert_eq!( + decision, + Decision::PublishDeletion, + "must reach Gate C, not park" + ); +} + +// ── (c) am#2 / peon-P2#1: user-intent retain clears the gate; reconcile +// retain does NOT. +// +// Mutation-sensitive: swapping `retain_user_intent_event` for `retain_event` +// in `prepare_persona_publication` causes the user-intent branch to keep the +// gate set, breaking the first assertion. +#[test] +fn test_user_edit_reopens_gated_row_but_reconcile_retain_does_not() { + let conn = test_db(); + const D_TAG: &str = "test-persona"; + + // Part 1: user-intent retain clears the gate. + let event = signed_persona("user-edit-v1"); + retain_event( + &conn, + &RetainedEvent { + kind: KIND_PERSONA, + 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() + }, + event_id: Some(event.id.to_hex()), + pending_sync: false, + publish_blocked: false, + }, + ) + .unwrap(); + set_publish_blocked(&conn, KIND_PERSONA, OWNER, D_TAG, true).unwrap(); + assert!( + is_publish_blocked(&conn, KIND_PERSONA, OWNER, D_TAG).unwrap(), + "precondition" + ); + + let user_edit = signed_persona("user-edit-v2"); + retain_user_intent_event( + &conn, + &RetainedEvent { + kind: KIND_PERSONA, + pubkey: OWNER.to_string(), + d_tag: D_TAG.to_string(), + content: user_edit.content.to_string(), + created_at: user_edit.created_at.as_secs() as i64 + 1, + raw_event: { + use nostr::JsonUtil; + user_edit.as_json() + }, + event_id: Some(user_edit.id.to_hex()), + pending_sync: true, + publish_blocked: false, + }, + ) + .unwrap(); + assert!( + !is_publish_blocked(&conn, KIND_PERSONA, OWNER, D_TAG).unwrap(), + "user-intent retain must clear publish_blocked" + ); + + // Part 2: reconcile retain must NOT clear a gate it finds set. + conn.execute( + "UPDATE persona_events SET publish_blocked = 1 + WHERE kind = ?1 AND pubkey = ?2 AND d_tag = ?3", + rusqlite::params![KIND_PERSONA, OWNER, D_TAG], + ) + .unwrap(); + assert!( + is_publish_blocked(&conn, KIND_PERSONA, OWNER, D_TAG).unwrap(), + "gate re-set" + ); + + let reconcile = signed_persona("user-edit-v2"); // same prompt, fresh timestamp + retain_event( + &conn, + &RetainedEvent { + kind: KIND_PERSONA, + pubkey: OWNER.to_string(), + d_tag: D_TAG.to_string(), + content: reconcile.content.to_string(), + created_at: reconcile.created_at.as_secs() as i64 + 2, + raw_event: { + use nostr::JsonUtil; + reconcile.as_json() + }, + event_id: Some(reconcile.id.to_hex()), + pending_sync: true, + publish_blocked: false, // reconcile always passes false — gate must not be read + }, + ) + .unwrap(); + assert!( + is_publish_blocked(&conn, KIND_PERSONA, OWNER, D_TAG).unwrap(), + "reconcile retain must NOT clear publish_blocked" + ); +} diff --git a/desktop/src-tauri/src/managed_agents/config_sync/tests.rs b/desktop/src-tauri/src/managed_agents/config_sync/tests.rs index e064364c71..b3a6c98068 100644 --- a/desktop/src-tauri/src/managed_agents/config_sync/tests.rs +++ b/desktop/src-tauri/src/managed_agents/config_sync/tests.rs @@ -585,7 +585,7 @@ fn observe(conn: &Connection, head: HeadState, disk: Option /// 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( + let (plan, _) = run_decision_pass( conn, OWNER, &[CoordinateState { diff --git a/desktop/src-tauri/src/managed_agents/decision.rs b/desktop/src-tauri/src/managed_agents/decision.rs index 5f7bf48626..d03c6241fd 100644 --- a/desktop/src-tauri/src/managed_agents/decision.rs +++ b/desktop/src-tauri/src/managed_agents/decision.rs @@ -308,8 +308,10 @@ pub fn decide(observation: &Observation) -> Resolution { // 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. + // say what the head is. Deferring costs one barrier pass; the row + // keeps its pending flag and publishes on the next full transition + // (the 30s flush retry re-runs reconcile+barrier when the latch is + // Unready, so the retry is automatic once the relay is reachable). return Resolution::keep(match observation.head { HeadState::LookupFailed => Decision::Defer, _ => Decision::PublishLocalEdit, diff --git a/desktop/src-tauri/src/managed_agents/persona_events.rs b/desktop/src-tauri/src/managed_agents/persona_events.rs index 7d1cf0ccc2..72bc1e172b 100644 --- a/desktop/src-tauri/src/managed_agents/persona_events.rs +++ b/desktop/src-tauri/src/managed_agents/persona_events.rs @@ -316,15 +316,15 @@ pub async fn flush_active_pending_events( flush_pending_events_at(&scope.db_path, state, &scope.relay_url, &scope.owner_keys).await } -async fn flush_pending_events_at( +pub(crate) async fn flush_pending_events_at( db_path: &std::path::Path, state: &AppState, relay_url: &str, owner_keys: &nostr::Keys, ) -> Result { use crate::managed_agents::retention::{ - deferred_behind_failed_tombstone, get_pending_sync, get_retained_event, mark_synced, - open_retention_db, + deferred_behind_failed_tombstone, get_pending_sync, get_retained_event, + mark_synced_and_stamp_baseline, open_retention_db, }; use nostr::JsonUtil; @@ -406,13 +406,15 @@ async fn flush_pending_events_at( } let conn = open_retention_db(db_path)?; - mark_synced( + mark_synced_and_stamp_baseline( &conn, current.kind, ¤t.pubkey, ¤t.d_tag, current.created_at, ¤t.content, + &event.id.to_hex(), + buzz_core_pkg::kind::event_is_shared(&event), )?; flushed += 1; } diff --git a/desktop/src-tauri/src/managed_agents/persona_events/flush_barrier_tests.rs b/desktop/src-tauri/src/managed_agents/persona_events/flush_barrier_tests.rs index 514e14b8b9..e3e6e3ee77 100644 --- a/desktop/src-tauri/src/managed_agents/persona_events/flush_barrier_tests.rs +++ b/desktop/src-tauri/src/managed_agents/persona_events/flush_barrier_tests.rs @@ -772,3 +772,228 @@ async fn test_stale_flush_barrier_complete_cannot_publish_after_force_claim() { config_sync_readiness::mark_unready(); // cleanup } + +// ── (d) peon-P1#3: lookup-failure pass must leave latch Unready ────────── +// +// A pass with `HeadState::LookupFailed` must return `Abandoned` (not +// `Success`) from `enforce_decision_pass`, so the latch stays Unready and +// the 30s retry re-runs the full transition once the relay is reachable. +// A later pass with successful lookups must recover to Ready. +// +// Mutation-sensitive: removing the `has_uncertain_gates` check causes the +// first call to return `Success`, breaking the Unready assertion. +#[test] +fn test_lookup_failure_pass_leaves_latch_unready_and_later_successful_pass_recovers() { + use crate::managed_agents::{ + config_barrier::{enforce_decision_pass, BarrierOutcome}, + config_sync::CoordinateState, + decision::{HeadState, TombstoneEvidence}, + head_lookup::Coordinate, + retention::open_retention_db, + }; + use nostr::{EventBuilder, JsonUtil, Kind, Tag}; + + config_sync_readiness::mark_unready(); + + let keys = nostr::Keys::generate(); + let pubkey = keys.public_key().to_hex(); + let d_tag = "lookup-fail-test"; + let dir = tempdir().expect("tempdir"); + let db_path = dir.path().join("retention.db"); + + // Retain a pending row (no baseline, simulates first-boot stale build). + { + let conn = open_retention_db(&db_path).expect("open db"); + let event = EventBuilder::new( + Kind::Custom(KIND_PERSONA as u16), + r#"{"system_prompt":"old"}"#.to_string(), + ) + .tags(vec![Tag::parse(["d", d_tag]).unwrap()]) + .sign_with_keys(&keys) + .expect("sign"); + retain_event( + &conn, + &RetainedEvent { + kind: KIND_PERSONA, + pubkey: pubkey.clone(), + d_tag: d_tag.to_string(), + content: event.content.to_string(), + created_at: event.created_at.as_secs() as i64, + raw_event: event.as_json(), + event_id: None, + pending_sync: true, + publish_blocked: false, + }, + ) + .expect("retain"); + } + + // Claim InProgress — simulates what the flush loop or boot barrier does. + let claim = config_sync_readiness::claim_in_progress().expect("claim"); + + // State with a failed head lookup (relay unreachable at boot). + let failed_states = vec![CoordinateState { + coordinate: Coordinate { + kind: KIND_PERSONA, + d_tag: d_tag.to_string(), + }, + observation: crate::managed_agents::config_sync::local_state( + &open_retention_db(&db_path).expect("db"), + &pubkey, + &Coordinate { + kind: KIND_PERSONA, + d_tag: d_tag.to_string(), + }, + None, // no disk projection for simplicity + HeadState::LookupFailed, + TombstoneEvidence::NotFound, + ) + .expect("local_state"), + }]; + + let outcome = enforce_decision_pass( + &claim, + &open_retention_db(&db_path).expect("db"), + &pubkey, + &failed_states, + ) + .expect("enforce"); + + assert_eq!( + outcome, + BarrierOutcome::Abandoned, + "a pass with LookupFailed gates must return Abandoned (claim dropped)" + ); + drop(claim); // RAII resets latch to Unready + assert!( + !config_sync_readiness::is_ready_for(&db_path), + "latch must be Unready after lookup-failure pass" + ); + + // Second attempt with successful lookups (relay returned Absent). + let claim2 = config_sync_readiness::claim_in_progress().expect("second claim"); + let successful_states = vec![CoordinateState { + coordinate: Coordinate { + kind: KIND_PERSONA, + d_tag: d_tag.to_string(), + }, + observation: crate::managed_agents::config_sync::local_state( + &open_retention_db(&db_path).expect("db"), + &pubkey, + &Coordinate { + kind: KIND_PERSONA, + d_tag: d_tag.to_string(), + }, + None, + HeadState::Absent, // relay confirmed absent — real lookup + TombstoneEvidence::NotFound, + ) + .expect("local_state"), + }]; + + let outcome2 = enforce_decision_pass( + &claim2, + &open_retention_db(&db_path).expect("db"), + &pubkey, + &successful_states, + ) + .expect("enforce2"); + + assert_eq!( + outcome2, + BarrierOutcome::Success, + "a pass with successful lookups must certify Ready" + ); + claim2.complete(db_path.clone()); // advance latch to Ready + assert!( + config_sync_readiness::is_ready_for(&db_path), + "latch must be Ready after successful pass" + ); + + config_sync_readiness::mark_unready(); // cleanup +} + +// ── (e) peon-P2#2: gated row must not bypass publish_blocked ───────────── +// +// `flush_pending_events_at` re-reads `publish_blocked` per-row before +// submitting. A row with `publish_blocked = true` must be skipped → 0 flushed. +// +// Mutation-sensitive: removing the `if current.publish_blocked { continue; }` +// guard causes flushed == 1. Also sensitive to the routing change: if +// `publish_prepared_persona_via_flush` reverted to `submit_signed_event_at_with_keys` +// it would bypass this guard entirely. +#[tokio::test] +async fn test_share_toggle_on_gated_row_does_not_reach_relay() { + use crate::managed_agents::persona_events::flush_pending_events_at; + + config_sync_readiness::mark_unready(); + + let keys = nostr::Keys::generate(); + let pubkey = keys.public_key().to_hex(); + let d_tag = "share-gate-test"; + let dir = tempdir().expect("tempdir"); + let db_path = dir.path().join("retention.db"); + let relay_url = spawn_stub_relay().await; + + // Simulate the boot pass: retain a row and gate it (publish_blocked = true). + // This is the state the old direct-submit path would have bypassed: the boot + // pass decided to park this coordinate, but the share-toggle came in anyway. + { + let conn = open_retention_db(&db_path).expect("open db"); + let event = nostr::EventBuilder::new( + nostr::Kind::Custom(KIND_PERSONA as u16), + r#"{"system_prompt":"gated"}"#.to_string(), + ) + .tags(vec![nostr::Tag::parse(["d", d_tag]).unwrap()]) + .sign_with_keys(&keys) + .expect("sign"); + retain_event( + &conn, + &RetainedEvent { + kind: KIND_PERSONA, + pubkey: pubkey.clone(), + d_tag: d_tag.to_string(), + content: event.content.clone().to_string(), + created_at: event.created_at.as_secs() as i64, + raw_event: { + use nostr::JsonUtil; + event.as_json() + }, + event_id: None, + pending_sync: true, + publish_blocked: false, + }, + ) + .expect("retain"); + // Boot barrier gated this coordinate. + set_publish_blocked(&conn, KIND_PERSONA, &pubkey, d_tag, true).expect("gate"); + } + + let state = build_app_state(); + *state.keys.lock().unwrap() = keys.clone(); + *state.relay_url_override.lock().unwrap() = Some(relay_url.clone()); + + // The gated row must be skipped → 0 flushed. + let flushed = flush_pending_events_at(&db_path, &state, &relay_url, &keys) + .await + .expect("flush"); + assert_eq!( + flushed, 0, + "flush must not publish a gated row (publish_blocked=true)" + ); + + let conn = open_retention_db(&db_path).expect("reopen db"); + let row = get_retained_event(&conn, KIND_PERSONA, &pubkey, d_tag) + .unwrap() + .unwrap(); + assert!( + row.pending_sync, + "row must stay pending when publish_blocked" + ); + assert!( + row.publish_blocked, + "gate must still be set after failed flush" + ); + + config_sync_readiness::mark_unready(); // cleanup +} diff --git a/desktop/src-tauri/src/managed_agents/retention.rs b/desktop/src-tauri/src/managed_agents/retention.rs index 88bf819f9d..686878a855 100644 --- a/desktop/src-tauri/src/managed_agents/retention.rs +++ b/desktop/src-tauri/src/managed_agents/retention.rs @@ -316,6 +316,24 @@ pub fn retain_event(conn: &Connection, event: &RetainedEvent) -> Result<(), Stri Ok(()) } +/// Upsert a user-intent persona event, then clear `publish_blocked`. +/// +/// The distinction from [`retain_event`] is critical: this is the path for +/// events authored by a local user interaction (create, update, share toggle, +/// delete tombstone). User intent MUST reopen a gated coordinate, because the +/// gate was set against a prior observation, not against the user's current +/// action. A reconcile-created retain uses [`retain_event`] and deliberately +/// does NOT clear the gate — the gate exists to stop the laundering path, and +/// the laundering path is exactly reconcile. +pub fn retain_user_intent_event(conn: &Connection, event: &RetainedEvent) -> Result<(), String> { + retain_event(conn, event)?; + // Always clear the gate for user intent, even if `retain_event`'s + // newer-or-equal guard left the row unchanged (same-second re-save). + // The gate must reflect the user's current decision, not a stale boot + // observation. + set_publish_blocked(conn, event.kind, &event.pubkey, &event.d_tag, false) +} + /// Outcome of an inbound retain — whether the local store now reflects the /// inbound event, so the caller knows whether to patch `personas.json`. /// @@ -591,6 +609,34 @@ fn parse_tombstone_retention_d_tag(key: &str) -> Option<(u32, String)> { Some((kind.parse().ok()?, d_tag.to_string())) } +/// Clear the `pending_sync` flag for an event the relay just confirmed, +/// and atomically stamp the accepted event as this coordinate's baseline. +/// +/// Compare-and-clear: only clears the row if its `created_at` and `content` +/// still match what was published. A concurrent edit that upserted a newer +/// version at the same coordinate between the flush loop's read and this call +/// leaves `pending_sync` set, so the newer edit publishes on the next pass +/// instead of being silently dropped. +/// +/// Stamping the baseline here is the fix for the one-edit-later revert vector: +/// after device A publishes v2 and this function runs, the baseline says v2, so +/// a later boot with head=v3 sees disk==baseline and stays silent — the +/// laundering path cannot mismatch against a baseline it just updated. +#[allow(clippy::too_many_arguments)] // 8 params mirror the atomic (mark-synced + stamp-baseline) operation; grouping them into a struct adds indirection for no gain +pub fn mark_synced_and_stamp_baseline( + conn: &Connection, + kind: u32, + pubkey: &str, + d_tag: &str, + created_at: i64, + content: &str, + event_id: &str, + shared: bool, +) -> Result<(), String> { + mark_synced(conn, kind, pubkey, d_tag, created_at, content)?; + set_baseline_full(conn, kind, pubkey, d_tag, event_id, content, shared) +} + /// 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` @@ -669,35 +715,90 @@ pub fn set_publish_blocked( Ok(()) } +/// The baseline recorded for a coordinate: what this install last agreed was +/// published there, including the `shared` tag state. +/// +/// Read as a unit because all three columns are only meaningful together. +/// `None` means the baseline was never established. +pub fn get_baseline_full( + conn: &Connection, + kind: u32, + pubkey: &str, + d_tag: &str, +) -> Result, String> { + conn.query_row( + "SELECT baseline_event_id, baseline_content, baseline_shared + FROM persona_baselines + WHERE kind = ?1 AND pubkey = ?2 AND d_tag = ?3", + params![kind, pubkey, d_tag], + |row| { + Ok(( + row.get::<_, String>(0)?, + row.get::<_, String>(1)?, + row.get::<_, i32>(2)? != 0, + )) + }, + ) + .optional() + .map_err(|e| format!("failed to get baseline: {e}")) +} + /// 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. +#[allow(dead_code)] 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)?, - )) - }, + get_baseline_full(conn, kind, pubkey, d_tag) + .map(|opt| opt.map(|(id, content, _shared)| (id, content))) +} + +/// Record what this install now agrees is published at a coordinate, +/// including the `shared` tag state from the event's tags. +/// +/// 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. +/// +/// Uses INSERT OR REPLACE so the baseline survives across live-row deletions: +/// the provenance lives in `persona_baselines`, which `delete_retained_event` +/// does NOT touch. +pub fn set_baseline_full( + conn: &Connection, + kind: u32, + pubkey: &str, + d_tag: &str, + baseline_event_id: &str, + baseline_content: &str, + baseline_shared: bool, +) -> Result<(), String> { + conn.execute( + "INSERT INTO persona_baselines + (kind, pubkey, d_tag, baseline_event_id, baseline_content, baseline_shared) + VALUES (?1, ?2, ?3, ?4, ?5, ?6) + ON CONFLICT (kind, pubkey, d_tag) DO UPDATE SET + baseline_event_id = excluded.baseline_event_id, + baseline_content = excluded.baseline_content, + baseline_shared = excluded.baseline_shared", + params![ + kind, + pubkey, + d_tag, + baseline_event_id, + baseline_content, + i32::from(baseline_shared), + ], ) - .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?)))) + .map_err(|e| format!("failed to set baseline: {e}"))?; + Ok(()) } /// Record what this install now agrees is published at a coordinate. @@ -706,6 +807,7 @@ pub fn get_baseline( /// 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. +#[allow(dead_code)] pub fn set_baseline( conn: &Connection, kind: u32, @@ -714,15 +816,15 @@ pub fn set_baseline( 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], + set_baseline_full( + conn, + kind, + pubkey, + d_tag, + baseline_event_id, + baseline_content, + false, ) - .map_err(|e| format!("failed to set baseline: {e}"))?; - - Ok(()) } /// Delete a retained event by its coordinate. diff --git a/desktop/src-tauri/src/managed_agents/retention/schema.rs b/desktop/src-tauri/src/managed_agents/retention/schema.rs index 82cf7cff7a..0c97f6d1e2 100644 --- a/desktop/src-tauri/src/managed_agents/retention/schema.rs +++ b/desktop/src-tauri/src/managed_agents/retention/schema.rs @@ -1,36 +1,48 @@ //! Additive schema migration for the retention store. //! //! Three columns were added after the original seven-column -//! `persona_events` table shipped, so an upgraded install has to grow its -//! table in place: +//! `persona_events` table shipped, and one new table (`persona_baselines`) +//! was added later: //! //! - `event_id` — the NIP-01 id of the row's own `raw_event`. Ordering has to //! match the relay's NIP-33 comparator, which breaks an equal `created_at` //! by lowest event id; without a real column every compare would have to //! reparse `raw_event`. -//! - `baseline_event_id` / `baseline_content` — provenance for the record -//! currently on disk: which event last wrote it and what that event -//! 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." +//! - `persona_baselines` — coordinate-keyed baseline provenance table. Stores +//! the full canonical projection (event_id, content, shared) for what this +//! install last agreed was published at each coordinate. Lives in its own +//! table so that deleting the live row never destroys its provenance: an +//! offline delete that misses the relay can still present a baseline on the +//! next boot and reach gate C's timestamp arbitration rather than parking +//! forever. +//! +//! The earlier `baseline_event_id` / `baseline_content` columns on +//! `persona_events` are superseded by `persona_baselines`. Existing data is +//! migrated upward; the old columns are left in place for schema hygiene (SQLite +//! does not support `DROP COLUMN` on older versions) but are never read after +//! migration. //! //! # Crash and concurrency safety //! //! A cheap read-only probe decides whether anything is missing, and only then //! does the migration open a `BEGIN EXCLUSIVE` transaction, re-probe inside -//! it, and apply the `ALTER TABLE`s together with the `event_id` backfill. +//! it, and apply the changes together with the `event_id` backfill. //! Serializing on the write lock is what makes two processes opening the same //! database at once safe: the loser waits, re-probes, and finds nothing to do. -//! A crash mid-migration rolls back the whole transaction, so a column can -//! never exist with its backfill half-applied. +//! A crash mid-migration rolls back the whole transaction, so no column or +//! table can exist with its backfill half-applied. use rusqlite::{Connection, TransactionBehavior}; /// Columns added after the initial `persona_events` shape. +/// +/// `baseline_event_id` and `baseline_content` are retained for schema +/// hygiene (SQLite ≤ 3.35 does not support `DROP COLUMN`) but are superseded +/// by the `persona_baselines` table and are never read after migration. const ADDED_COLUMNS: [(&str, &str); 4] = [ ("event_id", "TEXT"), ("baseline_event_id", "TEXT"), @@ -38,13 +50,28 @@ const ADDED_COLUMNS: [(&str, &str); 4] = [ ("publish_blocked", "INTEGER NOT NULL DEFAULT 0"), ]; -/// Bring `persona_events` up to the current column set. +/// Whether the `persona_baselines` table needs to be created. +fn baselines_table_missing(conn: &Connection) -> Result { + let count: i64 = conn + .query_row( + "SELECT count(*) FROM sqlite_master WHERE type='table' AND name='persona_baselines'", + [], + |row| row.get(0), + ) + .map_err(|e| format!("failed to probe persona_baselines table: {e}"))?; + Ok(count == 0) +} + +/// Bring `persona_events` up to the current column set and ensure +/// `persona_baselines` exists. /// /// A no-op on a database created by the current `CREATE TABLE` (and on any /// database already migrated), so it stays on the cheap path for every open /// after the first. pub(super) fn migrate(conn: &mut Connection) -> Result<(), String> { - if missing_columns(conn)?.is_empty() { + let missing_cols = missing_columns(conn)?; + let missing_table = baselines_table_missing(conn)?; + if missing_cols.is_empty() && !missing_table { return Ok(()); } @@ -63,6 +90,33 @@ pub(super) fn migrate(conn: &mut Connection) -> Result<(), String> { } backfill_event_ids(&transaction)?; + if baselines_table_missing(&transaction)? { + transaction + .execute_batch( + "CREATE TABLE persona_baselines ( + kind INTEGER NOT NULL, + pubkey TEXT NOT NULL, + d_tag TEXT NOT NULL, + baseline_event_id TEXT NOT NULL, + baseline_content TEXT NOT NULL, + baseline_shared INTEGER NOT NULL DEFAULT 0, + PRIMARY KEY (kind, pubkey, d_tag) + );", + ) + .map_err(|e| format!("failed to create persona_baselines table: {e}"))?; + // Migrate existing baseline data from the old columns into the new table. + // `shared` was never stored in the old columns, so it defaults to 0. + transaction + .execute_batch( + "INSERT OR IGNORE INTO persona_baselines + (kind, pubkey, d_tag, baseline_event_id, baseline_content, baseline_shared) + SELECT kind, pubkey, d_tag, baseline_event_id, baseline_content, 0 + FROM persona_events + WHERE baseline_event_id IS NOT NULL AND baseline_content IS NOT NULL;", + ) + .map_err(|e| format!("failed to migrate baseline data: {e}"))?; + } + transaction .commit() .map_err(|e| format!("failed to commit retention schema migration: {e}"))