jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-11-01 06:51:17 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(agent-config): am/peon feedback round — baseline table, user-intent gate, uncertain-lookup barrier, single publisher

Browse Source 
Addresses all five blocking findings from am's and peon's reviews:

am#1 / peon-P1#1: Move baseline provenance to a dedicated persona_baselines
table keyed by coordinate. The old in-row columns were wiped on delete, causing
an offline delete to park forever. The new table survives delete_retained_event,
so the next boot sees baseline=Present + queued deletion → Gate C arbitration.
stamp_baseline is now written at both boot (StampBaseline) and publish-confirm
(mark_synced_and_stamp_baseline in flush_pending_events_at), killing the
one-edit-later revert vector: after publishing v2, the baseline records v2, so
a later boot with head=v3 sees queued==baseline and stays silent.

am#2 / peon-P2#1: Separate user-intent writes from reconcile writes at the
retention API. retain_user_intent_event clears publish_blocked unconditionally;
retain_event (reconcile path) never touches the gate. All user-facing save and
tombstone paths now call retain_user_intent_event, so an edit to a parked
coordinate takes effect in-session without waiting for the next boot.

peon-P1#3: A decision pass driven by LookupFailed or ScanIncomplete must not
certify the scope Ready. run_decision_pass now returns a has_uncertain_gates
flag; enforce_decision_pass returns Abandoned when it is set, dropping the
claim and leaving the latch Unready for the 30s flush retry. A wifi-race boot
costs seconds, not a whole muted session. The false 'next pass' comment in
decision.rs is corrected.

peon-P2#2: set_persona_shared and update_persona_and_publish no longer submit
directly to the relay. Both route through publish_prepared_persona_via_flush,
which calls retain_user_intent_event (gate cleared for user intent) then
flush_pending_events_at (per-row publish_blocked re-read). The duplicated
publish/mark-synced logic in sharing.rs is deleted; retention.rs:508's
invariant is now total.

P3 (non-blocking): Deferred paragraph in PR body updated to reflect the
sharpened stakes of parked rows.

Regression tests (a)-(e): one test per finding, each traversing the production
seam it protects.

Co-authored-by: Will Pfleger <pfleger.will@gmail.com>
Signed-off-by: Will Pfleger <pfleger.will@gmail.com>
This commit is contained in:
npub1mn7jgtj4w2pd0g0zeuhxsa6jy6p0rewxz4kujt98my82ahfmp72sxjexk7
2026-08-01 14:08:50 -04:00
parent 3710265823
commit 4c2c84527f
11 changed files with 982 additions and 119 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
desktop/src-tauri/src/commands/personas/pending.rs
View File

@@ -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,

117
desktop/src-tauri/src/commands/personas/sharing.rs
View File

@@ -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::<AppState>();
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::<AppState>();
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<SetPersonaSharedResult, String> {
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(),

21
desktop/src-tauri/src/managed_agents/config_barrier.rs
View File

@@ -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)
}

82
desktop/src-tauri/src/managed_agents/config_sync.rs
View File

@@ -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<CanonicalProjection>
/// 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<Option<CanonicalProjection>, 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<Vec<(Coordinate, Resolution)>, 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;

389
desktop/src-tauri/src/managed_agents/config_sync/regression_tests.rs Normal file
View File

@@ -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"
);
}

2
desktop/src-tauri/src/managed_agents/config_sync/tests.rs
View File

@@ -585,7 +585,7 @@ fn observe(conn: &Connection, head: HeadState, disk: Option<CanonicalProjection>
/// 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 {

6
desktop/src-tauri/src/managed_agents/decision.rs
View File

@@ -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,

10
desktop/src-tauri/src/managed_agents/persona_events.rs
View File

@@ -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<u32, String> {
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,
&current.pubkey,
&current.d_tag,
current.created_at,
&current.content,
&event.id.to_hex(),
buzz_core_pkg::kind::event_is_shared(&event),
)?;
flushed += 1;
}

225
desktop/src-tauri/src/managed_agents/persona_events/flush_barrier_tests.rs
View File

@@ -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
}

150
desktop/src-tauri/src/managed_agents/retention.rs
View File

@@ -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<Option<(String, String, bool)>, 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<Option<(String, String)>, 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<String>>(0)?,
row.get::<_, Option<String>>(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.

78
desktop/src-tauri/src/managed_agents/retention/schema.rs
View File

@@ -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<bool, String> {
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}"))