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

fix(desktop): gate agent config publishes behind a boot resolution pass

Browse Source 
Boot reconcile re-signs whatever is on disk at `monotonic_created_at =
max(now, head + 1)`, so a merely-stale store mints events strictly newer
than the edits they revert. Last-write-wins then adopts the revert. A
second install with a fresh retention database is exactly this state,
which is why a dev build silently reverts config edited elsewhere.

`created_at` cannot arbitrate it: the value the stale store carries is
manufactured by the reconcile, not observed. The barrier arbitrates on
PROVENANCE instead. A baseline records what this install last agreed was
published at a coordinate; with no baseline, the install cannot show it
ever agreed to anything there, so neither its content nor its timestamps
are evidence of intent about the relay's state.

`decide` runs four ordered gates: positive tombstone evidence, then the
no-baseline park, then the queued-deletion timestamp compare, then the
queued edit against the baseline. The ordering is load-bearing — the
timestamp compare is sound only because the no-baseline gate already
excluded every store that would win it spuriously.

Both publication gates are enforced, because a coordinate's record and
its queued tombstone occupy different primary keys and the flush loop
reads them independently: gating one row computes a suppression and
enforces half of it. The live gate is default-deny over the decision
set, so a variant added later is withheld until deliberately admitted.

The scan is per-owner and unfiltered by `#a` because the relay
post-filters `#a` after its `LIMIT`, so absence is claimed only from a
page shorter than the limit it asked for.

Nothing is applied to disk here. `ApplyHead`, `RestoreFromRelay`, and
`DeleteLocal` are decided, logged, and gated; resolving them is the
user-driven step.

On relay deployments with a replica-routed read pool, the head lookups
and tombstone scan read best-available rather than writer-consistent;
staleness within replication lag is a documented known limitation (see
`head_lookup` and `tombstone_scan` module docs). The barrier's
per-decision tracing line is the detection mechanism if it fires.

Co-authored-by: Will Pfleger <pfleger.will@gmail.com>
Signed-off-by: Will Pfleger <pfleger.will@gmail.com>
This commit is contained in:
npub1mn7jgtj4w2pd0g0zeuhxsa6jy6p0rewxz4kujt98my82ahfmp72sxjexk7
2026-07-30 18:02:30 -04:00
parent e10f613a59
commit cf30ff47d3
16 changed files with 3999 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
desktop/src-tauri/src/event_sync.rs
View File

@@ -25,19 +25,35 @@ pub fn run_event_sync(app: &tauri::AppHandle, owner_keys: &nostr::Keys, db_path:
/// `AppState::keys` mutex. The reconcile itself is still synchronous JSON,
/// SQLite, and signing work, so it runs on the blocking pool rather than an
/// async worker.
///
/// # Ordering: reconcile, then barrier, and both before the flush publishes
///
/// The reconcile marks changed coordinates `pending_sync = 1`; the boot barrier
/// ([`crate::managed_agents::config_barrier::run_boot_barrier`]) then decides
/// which of those may actually go out and gates the rest. Running the barrier
/// after the reconcile is what lets it see the rows the reconcile just queued —
/// the stale-store republish it exists to suppress is created right here. The
/// flush loop's first tick is 30s of relay lookups away, so in practice the gate
/// is set before anything publishes; if a flush did win the race, the barrier
/// still gates every later sweep.
pub fn spawn_event_sync(
app: tauri::AppHandle,
owner_keys: nostr::Keys,
db_path: std::path::PathBuf,
) {
tauri::async_runtime::spawn(async move {
let barrier_app = app.clone();
if let Err(e) = tauri::async_runtime::spawn_blocking(move || {
run_event_sync(&app, &owner_keys, &db_path);
})
.await
{
eprintln!("buzz-desktop: event-sync: spawn_blocking failed: {e}");
// The barrier still runs: a failed reconcile leaves whatever rows
// a previous boot queued, and gating those is exactly the case the
// barrier exists for.
}
crate::managed_agents::config_barrier::run_boot_barrier(&barrier_app).await;
});
}

98
desktop/src-tauri/src/managed_agents/canonical_projection.rs Normal file
View File

@@ -0,0 +1,98 @@
//! The canonical published projection of a config record (V3.1).
//!
//! Every comparison the boot decision pass makes — disk against baseline, disk
//! against the relay head, baseline against what a publish would emit — runs on
//! this projection, never on whole JSON and never on content alone.
//!
//! Two reasons it cannot be either of those:
//!
//! - **Whole JSON is too wide.** Records carry install-local fields the relay
//! never sees (`source_dir`, `is_symlink`, `version`, timestamps, and for
//! managed agents `private_key_nsec`, `auth_tag`, `env_vars`). Comparing them
//! would report a difference for two records that publish identically, and
//! the pass would "repair" a divergence that does not exist.
//! - **Content alone is too narrow.** A persona's `shared` state rides in a
//! relay-significant *tag*, not in the content body. Two personas with equal
//! content but different `shared` tags publish as different events, so a
//! content-only compare would call them equal and skip a real publish.
//!
//! The projection is therefore exactly what publishing emits: the event content
//! plus the relay-significant tags, in a form that compares by value.
use buzz_core_pkg::kind::{KIND_MANAGED_AGENT, KIND_PERSONA, KIND_TEAM};
use super::{
agent_events::agent_event_content, persona_events::persona_event_content,
team_events::team_event_content, AgentDefinition, ManagedAgentRecord, TeamRecord,
};
/// What one coordinate publishes: the event content and the relay-significant
/// tags that travel with it.
///
/// Equality is the decision-table's notion of "same": two projections are equal
/// exactly when publishing both would put the same thing on the relay.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CanonicalProjection {
/// The serialized event content, byte-identical to what the builder emits.
pub content: String,
/// Whether the persona carries `["shared", "true"]`. Always `false` for
/// teams and managed agents, whose builders emit no `shared` tag.
pub shared: bool,
}
impl CanonicalProjection {
/// The projection a persona (kind:30175) would publish.
///
/// `shared` is read from the record rather than inferred, because it is the
/// field the builder turns into the tag.
pub fn for_persona(record: &AgentDefinition) -> Result<Self, String> {
Ok(Self {
content: serde_json::to_string(&persona_event_content(record))
.map_err(|e| format!("failed to serialize persona projection: {e}"))?,
shared: record.shared,
})
}
/// The projection a team (kind:30176) would publish.
pub fn for_team(record: &TeamRecord) -> Result<Self, String> {
Ok(Self {
content: serde_json::to_string(&team_event_content(record))
.map_err(|e| format!("failed to serialize team projection: {e}"))?,
shared: false,
})
}
/// The projection a managed agent (kind:30177) would publish.
///
/// `agent_event_content` is the opt-IN no-secrets allowlist, so this can
/// never carry a private key, auth tag, or env var.
pub fn for_managed_agent(record: &ManagedAgentRecord) -> Result<Self, String> {
Ok(Self {
content: serde_json::to_string(&agent_event_content(record))
.map_err(|e| format!("failed to serialize managed-agent projection: {e}"))?,
shared: false,
})
}
/// The projection an already-signed event carries.
///
/// This is the head side of a disk-vs-head compare. `shared` is read with
/// the same fail-closed helper the relay's read-authorization uses, so a
/// malformed tag shape is treated as not-shared on both sides of the
/// comparison rather than only one.
pub fn from_event(event: &nostr::Event) -> Self {
let kind = event.kind.as_u16() as u32;
Self {
content: event.content.to_string(),
shared: kind == KIND_PERSONA && buzz_core_pkg::kind::event_is_shared(event),
}
}
/// Whether `kind` is a config kind this projection knows how to build.
pub fn is_config_kind(kind: u32) -> bool {
matches!(kind, KIND_PERSONA | KIND_TEAM | KIND_MANAGED_AGENT)
}
}
#[cfg(test)]
mod tests;

199
desktop/src-tauri/src/managed_agents/canonical_projection/tests.rs Normal file
View File

@@ -0,0 +1,199 @@
use std::collections::BTreeMap;
use std::path::PathBuf;
use super::*;
use crate::managed_agents::persona_events::build_persona_event;
fn sample_persona() -> AgentDefinition {
AgentDefinition {
id: "test-persona".to_string(),
display_name: "Test Persona".to_string(),
avatar_url: Some("https://example.com/avatar.png".to_string()),
system_prompt: "You are a test assistant.".to_string(),
runtime: Some("goose".to_string()),
model: Some("claude-opus-4".to_string()),
provider: Some("anthropic".to_string()),
name_pool: vec!["Alpha".to_string()],
is_builtin: false,
is_active: true,
shared: false,
source_team: None,
source_team_persona_slug: Some("test-slug".to_string()),
catalog_source: None,
env_vars: BTreeMap::from([("KEY".to_string(), "value".to_string())]),
respond_to: None,
respond_to_allowlist: Vec::new(),
parallelism: None,
created_at: "2025-01-01T00:00:00Z".to_string(),
updated_at: "2025-01-01T00:00:00Z".to_string(),
}
}
fn sample_team() -> TeamRecord {
TeamRecord {
id: "team-123".to_string(),
name: "Test Team".to_string(),
description: Some("A test team".to_string()),
instructions: Some("Coordinate carefully.".to_string()),
persona_ids: vec!["p1".to_string()],
is_builtin: false,
source_dir: Some(PathBuf::from("/local/only/path")),
is_symlink: true,
symlink_target: Some("/somewhere".to_string()),
version: Some("1.0".to_string()),
created_at: "2025-01-01T00:00:00Z".to_string(),
updated_at: "2025-01-01T00:00:00Z".to_string(),
}
}
/// The whole point of the projection: a field the relay never sees must not
/// register as a difference, or the decision pass "repairs" a divergence that
/// does not exist and republishes on every boot.
#[test]
fn local_only_persona_fields_do_not_change_the_projection() {
let base = sample_persona();
let local_edit = AgentDefinition {
// None of these are in the published content.
is_active: !base.is_active,
source_team: Some("some-team".to_string()),
env_vars: BTreeMap::from([("OTHER".to_string(), "different".to_string())]),
created_at: "2099-01-01T00:00:00Z".to_string(),
updated_at: "2099-01-01T00:00:00Z".to_string(),
..base.clone()
};
assert_eq!(
CanonicalProjection::for_persona(&base).unwrap(),
CanonicalProjection::for_persona(&local_edit).unwrap()
);
}
/// A published field must register, or a real edit is silently never published.
#[test]
fn published_persona_fields_change_the_projection() {
let base = sample_persona();
for edited in [
AgentDefinition {
system_prompt: "Different prompt.".to_string(),
..base.clone()
},
AgentDefinition {
model: Some("other-model".to_string()),
..base.clone()
},
AgentDefinition {
provider: Some("other-provider".to_string()),
..base.clone()
},
AgentDefinition {
display_name: "Renamed".to_string(),
..base.clone()
},
] {
assert_ne!(
CanonicalProjection::for_persona(&base).unwrap(),
CanonicalProjection::for_persona(&edited).unwrap(),
"a published field edit must be visible to the decision pass"
);
}
}
/// `shared` rides in a relay-significant TAG, not the content body. A
/// content-only compare would call these two equal and skip a real publish —
/// this is the case that makes "content alone" wrong.
#[test]
fn shared_tag_difference_is_visible_even_when_content_matches() {
let private = sample_persona();
let shared = AgentDefinition {
shared: true,
..private.clone()
};
let private_projection = CanonicalProjection::for_persona(&private).unwrap();
let shared_projection = CanonicalProjection::for_persona(&shared).unwrap();
assert_eq!(
private_projection.content, shared_projection.content,
"precondition: the content bodies are identical"
);
assert_ne!(
private_projection, shared_projection,
"the shared tag must still make the projections differ"
);
}
/// Disk-vs-head is only meaningful if both sides project identically for a
/// record that was published unchanged.
#[test]
fn disk_projection_round_trips_through_a_signed_event() {
let keys = nostr::Keys::generate();
for shared in [false, true] {
let record = AgentDefinition {
shared,
..sample_persona()
};
let event = build_persona_event(&record)
.unwrap()
.sign_with_keys(&keys)
.unwrap();
assert_eq!(
CanonicalProjection::for_persona(&record).unwrap(),
CanonicalProjection::from_event(&event),
"a freshly published record must project equal to its own event"
);
}
}
/// Teams and managed agents emit no `shared` tag, so their projections must
/// never claim one — otherwise a team would compare unequal to its own event.
#[test]
fn non_persona_kinds_are_never_shared() {
assert!(
!CanonicalProjection::for_team(&sample_team())
.unwrap()
.shared
);
let keys = nostr::Keys::generate();
let event = crate::managed_agents::team_events::build_team_event(&sample_team())
.unwrap()
.sign_with_keys(&keys)
.unwrap();
assert_eq!(
CanonicalProjection::for_team(&sample_team()).unwrap(),
CanonicalProjection::from_event(&event)
);
}
/// A team's install-local fields (`source_dir`, `is_symlink`, `version`) are
/// exactly the ones V3.10 cites as unrecoverable from a relay projection, so
/// they must not drive a publish decision either.
#[test]
fn local_only_team_fields_do_not_change_the_projection() {
let base = sample_team();
let local_edit = TeamRecord {
source_dir: Some(PathBuf::from("/a/completely/different/path")),
is_symlink: !base.is_symlink,
symlink_target: None,
version: Some("9.9".to_string()),
updated_at: "2099-01-01T00:00:00Z".to_string(),
..base.clone()
};
assert_eq!(
CanonicalProjection::for_team(&base).unwrap(),
CanonicalProjection::for_team(&local_edit).unwrap()
);
}
#[test]
fn config_kinds_are_recognized_and_others_are_not() {
for kind in [KIND_PERSONA, KIND_TEAM, KIND_MANAGED_AGENT] {
assert!(CanonicalProjection::is_config_kind(kind));
}
// kind:5 deletions and chat messages are not config coordinates.
for kind in [1u32, 5, 30178] {
assert!(!CanonicalProjection::is_config_kind(kind));
}
}

284
desktop/src-tauri/src/managed_agents/config_barrier.rs Normal file
View File

@@ -0,0 +1,284 @@
//! The boot config barrier: gather every coordinate, decide, enforce.
//!
//! This is the entry point that makes [`super::decision::decide`] mean
//! something. It runs once per boot, before the flush loop can publish, and
//! wires the three layers together in the order V3.4 requires:
//!
//! 1. enumerate the coordinates this device knows about, from disk;
//! 2. run every relay lookup with NO store lock held;
//! 3. take `managed_agents_store_lock` once, re-check the scope, gather the
//! local half, decide, and enforce — all under that single guard.
//!
//! # Why the barrier exists
//!
//! Without it, boot re-signs local disk content at
//! `monotonic_created_at = max(now, head + 1)` and queues it for publish. A
//! store that is merely stale therefore mints events that are strictly newer
//! than the edits they revert, and the relay's last-write-wins resolution
//! adopts the revert. The barrier's job is to run BEFORE that publish and
//! decide, per coordinate, whether the local content is an edit worth
//! publishing or a stale copy that must be suppressed.
//!
//! # What it enforces, and what it does not
//!
//! It enforces exactly one thing: the publication gate
//! ([`super::config_sync::apply_gate`]). A coordinate whose decision blocks
//! publication is withheld from [`super::retention::get_pending_sync`] and
//! cannot reach the relay on this boot. Decisions that PATCH disk
//! (`ApplyHead`, `RestoreFromRelay`, `DeleteLocal`) are logged and gated but
//! not applied here — applying them is the resolution path (V3.6), which is
//! user-driven by design. Suppression is the half that stops the data loss;
//! adopting remote content is the half that needs a human.
use std::collections::BTreeSet;
use tauri::Manager;
use super::{
canonical_projection::CanonicalProjection,
config_sync::{local_state, run_decision_pass, CoordinateState},
decision::HeadState,
head_lookup::{fetch_head, Coordinate},
retention::{active_retention_scope, open_retention_db, pending_coordinates, RetentionScope},
tombstone_scan::scan_tombstones,
};
use crate::app_state::AppState;
use buzz_core_pkg::kind::{KIND_MANAGED_AGENT, KIND_PERSONA, KIND_TEAM};
/// Run the boot barrier for the active scope, best-effort.
///
/// Failures are logged and swallowed: a barrier that cannot run must not block
/// boot, and its absence is not itself dangerous — it leaves the pre-barrier
/// behaviour, which is what every prior release did.
pub async fn run_boot_barrier(app: &tauri::AppHandle) {
let state = app.state::<AppState>();
let scope = match active_retention_scope(app, &state) {
Ok(scope) => scope,
Err(error) => {
tracing::warn!(target: "buzz::config_sync", %error, "boot barrier skipped: no active scope");
return;
}
};
if let Err(error) = run_boot_barrier_for_scope(app, &state, &scope).await {
tracing::warn!(target: "buzz::config_sync", %error, "boot barrier failed");
}
}
/// Withhold every queued publish that has no baseline, before any network I/O.
///
/// This is V3.9's legacy-pending quarantine, and it is also what makes the
/// barrier safe against its own latency. The barrier's decisions need a
/// per-coordinate relay lookup plus a full tombstone scan; the flush loop needs
/// only a timer. On a slow network the flush can therefore win the race and
/// publish exactly the rows the barrier was about to gate.
///
/// Quarantining first inverts that: the gate is closed synchronously, before
/// the first round trip, and the decision pass re-opens it for the coordinates
/// that earn it. A row is either provably publishable or it is not going out.
///
/// Scoped to rows with no baseline because those are precisely the unprovable
/// ones: a legacy row queued before the baseline column existed, and a row
/// queued by a second store whose retention database has never agreed to
/// anything at that coordinate. A row WITH a baseline is left alone — its
/// arbitration is decidable, and closing its gate here would stall an ordinary
/// edit behind a scan it does not need.
///
/// Returns how many rows it withheld.
fn quarantine_unprovable_pending(
conn: &rusqlite::Connection,
owner_pubkey: &str,
) -> Result<usize, String> {
conn.execute(
"UPDATE persona_events SET publish_blocked = 1
WHERE pending_sync = 1 AND pubkey = ?1 AND baseline_event_id IS NULL",
rusqlite::params![owner_pubkey],
)
.map_err(|error| format!("failed to quarantine unprovable pending rows: {error}"))
}
async fn run_boot_barrier_for_scope(
app: &tauri::AppHandle,
state: &AppState,
scope: &RetentionScope,
) -> Result<(), String> {
let owner_pubkey = scope.owner_keys.public_key().to_hex();
// Enumerate BEFORE the lookups: the set of coordinates to decide is the
// union of what disk holds and what is already queued for publish. A
// pending row can outlive its record (deleted persona, removed agent, or a
// tombstone), and those rows publish with nothing on disk to enumerate
// them — exactly the ones that most need gating. The gate must see
// everything the publisher can see, and the publisher reads the table.
let mut disk = disk_projections(app)?;
{
let conn = open_retention_db(&scope.db_path)?;
let quarantined = quarantine_unprovable_pending(&conn, &owner_pubkey)?;
if quarantined > 0 {
tracing::info!(
target: "buzz::config_sync",
quarantined,
"withheld pending rows with no baseline pending the decision pass"
);
}
let known: BTreeSet<(u32, String)> = disk
.iter()
.map(|(coordinate, _)| (coordinate.kind, coordinate.d_tag.clone()))
.collect();
for (kind, d_tag) in pending_coordinates(&conn, &owner_pubkey)? {
if !known.contains(&(kind, d_tag.clone())) {
// No disk record: the projection is `None`, which is what the
// table reads as "absent locally".
disk.push((Coordinate { kind, d_tag }, None));
}
}
}
// Phase 1 — relay lookups, no lock held. Each head read is an exact
// per-coordinate query; a failure becomes `LookupFailed`
// rather than an absence, so a flaky network can never be read as a
// deletion. The tombstone scan runs once for the whole owner.
let api_base_url = crate::relay::relay_http_base_url(&scope.relay_url);
let mut heads: Vec<(Coordinate, HeadState)> = Vec::with_capacity(disk.len());
for (coordinate, _) in &disk {
heads.push((
coordinate.clone(),
fetch_head(state, scope, &api_base_url, coordinate).await,
));
}
let tombstones = scan_tombstones(state, scope, &api_base_url).await;
// Phase 2 — one guard for the whole decision pass.
let _store_guard = state
.managed_agents_store_lock
.lock()
.map_err(|error| error.to_string())?;
// Phase 3 — re-resolve the scope under the lock. `apply_workspace` can swap
// the relay and keys mid-flight without taking this lock, so without this
// check a completed pull from community A could patch community B's store.
let current = active_retention_scope(app, state)?;
if current.db_path != scope.db_path {
tracing::info!(
target: "buzz::config_sync",
"boot barrier abandoned: workspace switched during relay lookups"
);
return Ok(());
}
let conn = open_retention_db(&scope.db_path)?;
let mut states: Vec<CoordinateState> = Vec::with_capacity(disk.len());
for ((coordinate, projection), (_, head)) in disk.into_iter().zip(heads) {
let tombstone = tombstones.evidence(&coordinate);
let observation = local_state(
&conn,
&owner_pubkey,
&coordinate,
projection,
head,
tombstone,
)?;
states.push(CoordinateState {
coordinate,
observation,
});
}
run_decision_pass(&conn, &owner_pubkey, &states)?;
Ok(())
}
/// Read the three record stores and enumerate their coordinates.
fn disk_projections(
app: &tauri::AppHandle,
) -> Result<Vec<(Coordinate, Option<CanonicalProjection>)>, String> {
use super::{load_managed_agents, load_teams, personas::load_personas};
Ok(project_records(
&load_personas(app)?,
&load_teams(app)?,
&load_managed_agents(app)?,
))
}
/// Every config coordinate the given records occupy, with what each would
/// publish.
///
/// Uses the same d-tag derivations the publish paths use, so a coordinate can
/// never be enumerated under a key that differs from the one it would publish
/// under. Built-in records are excluded: they come from code, are never
/// published, and have no relay head to compare against.
///
/// A record whose projection cannot be built is skipped with a warning rather
/// than failing the pass — one malformed record must not stop the barrier from
/// protecting every other coordinate.
fn project_records(
personas: &[super::AgentDefinition],
teams: &[super::TeamRecord],
agents: &[super::ManagedAgentRecord],
) -> Vec<(Coordinate, Option<CanonicalProjection>)> {
let mut out: Vec<(Coordinate, Option<CanonicalProjection>)> = Vec::new();
let mut seen: BTreeSet<(u32, String)> = BTreeSet::new();
let mut push = |kind: u32, d_tag: String, projection: Result<CanonicalProjection, String>| {
match projection {
Ok(projection) => {
// Two records can normalize onto one d-tag. Deciding the
// coordinate twice would gate it twice from the same inputs,
// so the first wins and the collision is visible in the log.
if seen.insert((kind, d_tag.clone())) {
out.push((Coordinate { kind, d_tag }, Some(projection)));
} else {
tracing::warn!(
target: "buzz::config_sync",
kind,
d_tag = %d_tag,
"duplicate coordinate on disk; deciding the first record only"
);
}
}
Err(error) => {
tracing::warn!(
target: "buzz::config_sync",
kind,
d_tag = %d_tag,
%error,
"skipping coordinate with an unbuildable projection"
);
}
}
};
for record in personas.iter().filter(|record| !record.is_builtin) {
push(
KIND_PERSONA,
super::persona_events::persona_d_tag(record),
CanonicalProjection::for_persona(record),
);
}
for record in teams.iter().filter(|record| !record.is_builtin) {
push(
KIND_TEAM,
record.id.clone(),
CanonicalProjection::for_team(record),
);
}
for record in agents
.iter()
// A key-less agent has no coordinate yet: its d-tag IS its pubkey, and
// it mints one on first start.
.filter(|record| !record.pubkey.is_empty())
{
push(
KIND_MANAGED_AGENT,
record.pubkey.clone(),
CanonicalProjection::for_managed_agent(record),
);
}
out
}
#[cfg(test)]
mod tests;

199
desktop/src-tauri/src/managed_agents/config_barrier/tests.rs Normal file
View File

@@ -0,0 +1,199 @@
use std::collections::BTreeMap;
use std::path::PathBuf;
use super::*;
use crate::managed_agents::{AgentDefinition, ManagedAgentRecord, TeamRecord};
fn persona(id: &str, slug: Option<&str>, builtin: bool) -> AgentDefinition {
AgentDefinition {
id: id.to_string(),
display_name: "Test".to_string(),
avatar_url: None,
system_prompt: "prompt".to_string(),
runtime: None,
model: None,
provider: None,
name_pool: Vec::new(),
is_builtin: builtin,
is_active: true,
shared: false,
source_team: None,
source_team_persona_slug: slug.map(str::to_string),
catalog_source: None,
env_vars: BTreeMap::new(),
respond_to: None,
respond_to_allowlist: Vec::new(),
parallelism: None,
created_at: "2026-01-01T00:00:00Z".to_string(),
updated_at: "2026-01-01T00:00:00Z".to_string(),
}
}
fn team(id: &str, builtin: bool) -> TeamRecord {
TeamRecord {
id: id.to_string(),
name: "Team".to_string(),
description: None,
instructions: None,
persona_ids: Vec::new(),
is_builtin: builtin,
source_dir: Some(PathBuf::from("/local/only")),
is_symlink: false,
symlink_target: None,
version: None,
created_at: "2026-01-01T00:00:00Z".to_string(),
updated_at: "2026-01-01T00:00:00Z".to_string(),
}
}
fn agent(pubkey: &str) -> ManagedAgentRecord {
serde_json::from_str(&format!(
r#"{{
"pubkey": "{pubkey}",
"name": "Agent",
"relay_url": "wss://localhost:3000",
"acp_command": "buzz-acp",
"agent_command": "goose",
"agent_args": [],
"mcp_command": "",
"turn_timeout_seconds": 320,
"system_prompt": "You are a test agent.",
"created_at": "2026-01-01T00:00:00Z",
"updated_at": "2026-01-01T00:00:00Z",
"last_started_at": null,
"last_stopped_at": null,
"last_exit_code": null,
"last_error": null
}}"#
))
.unwrap()
}
fn coordinates(records: &[(Coordinate, Option<CanonicalProjection>)]) -> Vec<(u32, String)> {
records
.iter()
.map(|(coordinate, _)| (coordinate.kind, coordinate.d_tag.clone()))
.collect()
}
/// All three config kinds must be enumerated. A kind the barrier does not
/// enumerate is a kind it can never suppress, so the revert bug would survive
/// for exactly those records.
#[test]
fn test_all_three_config_kinds_are_enumerated() {
let projected = project_records(
&[persona("p1", None, false)],
&[team("t1", false)],
&[agent("agentpubkey")],
);
assert_eq!(
coordinates(&projected),
vec![
(KIND_PERSONA, "p1".to_string()),
(KIND_TEAM, "t1".to_string()),
(KIND_MANAGED_AGENT, "agentpubkey".to_string()),
]
);
}
/// Built-ins ship in the binary and are never published, so a head lookup for
/// one would always report `Absent` and gate a coordinate that has nothing to
/// gate.
#[test]
fn test_builtin_records_are_not_enumerated() {
let projected = project_records(
&[persona("builtin", None, true), persona("real", None, false)],
&[team("builtin-team", true), team("real-team", false)],
&[],
);
assert_eq!(
coordinates(&projected),
vec![
(KIND_PERSONA, "real".to_string()),
(KIND_TEAM, "real-team".to_string()),
]
);
}
/// A key-less agent has no coordinate at all — its d-tag IS its pubkey, minted
/// on first start. Enumerating it would look up `{"#d": [""]}`.
#[test]
fn test_keyless_agents_are_not_enumerated() {
let projected = project_records(&[], &[], &[agent(""), agent("haskey")]);
assert_eq!(
coordinates(&projected),
vec![(KIND_MANAGED_AGENT, "haskey".to_string())]
);
}
/// The enumerated d-tag must be the one the publish path would use, or the
/// barrier looks up a head for a coordinate this device never writes to and
/// reads a spurious absence. `persona_d_tag` prefers the team slug and
/// normalizes it.
#[test]
fn test_persona_coordinate_uses_the_published_d_tag_derivation() {
let projected = project_records(&[persona("uuid-id", Some("CodeReviewer"), false)], &[], &[]);
assert_eq!(
coordinates(&projected),
vec![(KIND_PERSONA, "codereviewer".to_string())]
);
}
/// Two personas can normalize onto one d-tag. Deciding it twice would gate the
/// same coordinate from the same inputs twice; the second pass is noise.
#[test]
fn test_duplicate_coordinates_are_decided_once() {
let projected = project_records(
&[
persona("first", Some("Shared"), false),
persona("second", Some("shared"), false),
],
&[],
&[],
);
assert_eq!(
coordinates(&projected),
vec![(KIND_PERSONA, "shared".to_string())]
);
}
/// A d-tag collision ACROSS kinds is not a duplicate: `30175:x` and `30176:x`
/// are different coordinates with different heads, and collapsing them would
/// let a team's head answer a persona's lookup.
#[test]
fn test_same_d_tag_on_different_kinds_are_separate_coordinates() {
let projected = project_records(&[persona("same", None, false)], &[team("same", false)], &[]);
assert_eq!(
coordinates(&projected),
vec![
(KIND_PERSONA, "same".to_string()),
(KIND_TEAM, "same".to_string()),
]
);
}
/// The projection carried into the decision must be what publishing emits, or
/// every disk-vs-head compare is against the wrong value.
#[test]
fn test_projection_matches_what_publishing_would_emit() {
let record = persona("p1", None, false);
let projected = project_records(std::slice::from_ref(&record), &[], &[]);
assert_eq!(
projected[0].1,
Some(CanonicalProjection::for_persona(&record).unwrap())
);
}
/// Empty stores are the fresh-install case: no coordinates, no lookups, no
/// gate writes.
#[test]
fn test_no_records_yields_no_coordinates() {
assert!(project_records(&[], &[], &[]).is_empty());
}

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

@@ -0,0 +1,372 @@
//! Boot config sync: gather each coordinate's observable state, decide, enforce.
//!
//! This is the layer between the pure decision table
//! ([`super::decision::decide`]) and the stores it reads. It owns two things
//! the table deliberately does not: WHERE each input comes from, and the
//! ordering guarantees around reading them.
//!
//! # Ordering (V3.4)
//!
//! No network I/O may happen while the store lock is held, and no decision may
//! be made from inputs gathered under a different workspace:
//!
//! 1. capture `(relay, owner, db path)` as a scope snapshot;
//! 2. run the relay lookups with NO lock held;
//! 3. take `managed_agents_store_lock` for the whole decision pass;
//! 4. re-resolve the active scope and abandon the pass if it moved;
//! 5. evaluate the table and write, still under that one guard.
//!
//! Step 4 is load-bearing rather than defensive: `apply_workspace` can swap the
//! relay and keys mid-flight and does not take this lock, so without the
//! recheck a completed pull from community A could patch community B's shared
//! disk store.
use super::{
canonical_projection::CanonicalProjection,
decision::{Decision, Head, HeadState, Observation, Resolution, TombstoneEvidence},
head_lookup::Coordinate,
retention::{get_baseline, get_retained_event, RetainedEvent},
};
use rusqlite::Connection;
/// Everything the decision table needs about one coordinate, plus the identity
/// of the coordinate itself.
#[derive(Debug, Clone)]
pub struct CoordinateState {
pub coordinate: Coordinate,
pub observation: Observation,
}
/// Read the locally-known half of a coordinate's state: the queued publish, any
/// queued deletion, and the baseline.
///
/// The head and the tombstone evidence come from the relay and are supplied by
/// the caller, which has already run those lookups outside the lock.
///
/// The coordinate's own row and its paired tombstone row occupy different
/// primary keys and are reported separately, because the table resolves them on
/// different evidence — a deletion by timestamp against the head, an edit
/// against the baseline. Collapsing them into one flag was the cross-coordinate
/// hole called out in review; reporting only the live row is the same hole.
pub fn local_state(
conn: &Connection,
owner_pubkey: &str,
coordinate: &Coordinate,
disk: Option<CanonicalProjection>,
head: HeadState,
tombstone: TombstoneEvidence,
) -> Result<Observation, String> {
let row = get_retained_event(conn, coordinate.kind, owner_pubkey, &coordinate.d_tag)?;
let tombstone_row = get_retained_event(
conn,
buzz_core_pkg::kind::KIND_DELETION,
owner_pubkey,
&super::retention::tombstone_retention_d_tag(coordinate.kind, &coordinate.d_tag),
)?;
Ok(Observation {
disk,
queued: queued_projection(row.as_ref()),
head,
baseline: baseline_projection(conn, owner_pubkey, coordinate, row.as_ref())?,
queued_deletion_at: tombstone_row
.as_ref()
.filter(|row| row.pending_sync)
.map(|row| row.created_at),
tombstone,
})
}
/// What a coordinate's queued row would publish, or `None` when nothing is
/// queued there.
///
/// Built from the retained row's own `raw_event` rather than from disk: the
/// flush loop publishes exactly those bytes, so those are the bytes the
/// baseline arbitration has to be about. A row whose stored event will not
/// parse reports `None` — it cannot publish either, since the flush loop parses
/// the same field and fails the row.
fn queued_projection(row: Option<&RetainedEvent>) -> Option<CanonicalProjection> {
use nostr::JsonUtil;
let row = row.filter(|row| row.pending_sync)?;
let event = nostr::Event::from_json(&row.raw_event).ok()?;
Some(CanonicalProjection::from_event(&event))
}
/// The baseline projection for a coordinate, or `None` when this install has
/// no provenance for it.
///
/// The stored baseline is the CONTENT that was published; the `shared` half of
/// the projection is recovered from the baseline event's own tags via the
/// retained row, so a baseline comparison sees the same two fields a head
/// comparison does.
fn baseline_projection(
conn: &Connection,
owner_pubkey: &str,
coordinate: &Coordinate,
row: Option<&RetainedEvent>,
) -> Result<Option<CanonicalProjection>, String> {
let Some((baseline_event_id, baseline_content)) =
get_baseline(conn, coordinate.kind, owner_pubkey, &coordinate.d_tag)?
else {
return Ok(None);
};
// Recover `shared` from the retained event when the row still holds the
// baseline event; otherwise the tag state is unknown and the baseline is
// only usable for kinds that never carry one.
let shared = row
.filter(|row| row.event_id.as_deref() == Some(baseline_event_id.as_str()))
.and_then(|row| {
use nostr::JsonUtil;
nostr::Event::from_json(&row.raw_event).ok()
})
.map(|event| CanonicalProjection::from_event(&event).shared)
.unwrap_or(false);
Ok(Some(CanonicalProjection {
content: baseline_content,
shared,
}))
}
/// Resolve every gathered coordinate, pairing each resolution with its target.
///
/// Kept separate from execution so a caller can log or inspect the full plan
/// before anything is written — and so the mapping is testable end to end
/// without a relay or a disk store.
pub fn plan(states: &[CoordinateState]) -> Vec<(Coordinate, Resolution)> {
states
.iter()
.map(|state| {
(
state.coordinate.clone(),
super::decision::decide(&state.observation),
)
})
.collect()
}
/// Whether a resolution must withhold the coordinate's LIVE row from the flush
/// loop.
///
/// The gate is DEFAULT-DENY over the decision set: it lists what may publish,
/// not what may not. Every publishing decision names itself here, so a variant
/// added later is withheld until someone deliberately admits it — the opposite
/// default would let a new decision reach the relay by omission.
///
/// Clearing the gate is as load-bearing as setting it: a coordinate suppressed
/// on one boot must resume publishing once the condition that suppressed it is
/// gone, or the gate latches shut permanently and the record strands.
pub fn blocks_publication(resolution: &Resolution) -> bool {
match resolution.decision {
// The one decision that publishes local content.
Decision::PublishLocalEdit => false,
// The deletion publishes from the TOMBSTONE row, not this one. The
// live row is the thing being deleted, so it must stay withheld:
// publishing it would race its own tombstone at the same coordinate.
Decision::PublishDeletion => true,
// Adopts the relay's version onto disk, so nothing local is going out.
// Only reachable with a baseline present and nothing queued (gate D
// takes a queued row first), so releasing strands nothing.
Decision::ApplyHead => false,
// Only reachable with NO local record and nothing queued, so there is
// nothing to withhold — and nothing to strand by withholding. Blocked
// rather than released because it is the one adopt-the-relay cell
// reachable with no baseline, and a coordinate with no provenance
// against a live head must not leave an open gate behind for a row
// queued after the pass.
Decision::RestoreFromRelay => true,
// `StampBaseline` is easy to miss: disk already equals the head, so
// publishing would push byte-identical content and burn a
// `monotonic_created_at` bump for nothing.
Decision::StampBaseline
| Decision::SuppressPublish
| Decision::Park(_)
| Decision::DeleteLocal
| Decision::Defer => true,
}
}
/// Whether a resolution must withhold the coordinate's queued TOMBSTONE row.
///
/// A queued deletion does not live at the coordinate it deletes: it is retained
/// at `(5, owner, "<target_kind>:<target_d_tag>")`, a different primary key.
/// Gating only [`blocks_publication`]'s row would therefore compute a
/// suppression and enforce it on the wrong row, leaving the tombstone free to
/// publish — the exact "decorative gate" failure this pass exists to avoid, and
/// the worse direction of it: a deletion that escapes the gate destroys a head
/// rather than merely reverting it.
///
/// Exactly one decision releases it. Every other outcome — including the
/// timestamp arbitration falling through to the live cells — means the queued
/// deletion did not win, so it stays withheld.
pub fn blocks_tombstone_publication(resolution: &Resolution) -> bool {
resolution.decision != Decision::PublishDeletion
}
/// Log one resolution with the inputs that produced it, then enforce BOTH
/// publication gates.
///
/// Two rows, not one: the coordinate's live row and its queued tombstone row
/// occupy different primary keys, and the flush loop reads them independently.
/// A single `set_publish_blocked` call therefore enforces half a decision.
///
/// The log line is a hard requirement, not debug output: this pass silently
/// changes what a device publishes, and the boot log is the only evidence of
/// why a coordinate did or did not go out. Every field the table branched on is
/// recorded, so a surprising decision can be re-derived from the log alone
/// without reproducing the machine's store.
pub fn apply_gate(
conn: &Connection,
owner_pubkey: &str,
coordinate: &Coordinate,
observation: &Observation,
resolution: &Resolution,
) -> Result<(), String> {
tracing::info!(
target: "buzz::config_sync",
kind = coordinate.kind,
d_tag = %coordinate.d_tag,
decision = ?resolution.decision,
blocked = blocks_publication(resolution),
tombstone_blocked = blocks_tombstone_publication(resolution),
cleared_queued_publish = resolution.clear_queued_publish,
head = match &observation.head {
HeadState::Present(_) => "present",
HeadState::Absent => "absent",
HeadState::LookupFailed => "lookup_failed",
},
disk_present = observation.disk.is_some(),
disk_matches_head = matches!(
(&observation.disk, &observation.head),
(Some(disk), HeadState::Present(head)) if head.projection == *disk
),
has_baseline = observation.baseline.is_some(),
queued = observation.queued.is_some(),
queued_matches_baseline = matches!(
(&observation.queued, &observation.baseline),
(Some(queued), Some(baseline)) if queued == baseline
),
queued_deletion = observation.queued_deletion_at.is_some(),
tombstone = ?observation.tombstone,
"config-sync decision"
);
super::retention::set_publish_blocked(
conn,
coordinate.kind,
owner_pubkey,
&coordinate.d_tag,
blocks_publication(resolution),
)?;
super::retention::set_publish_blocked(
conn,
buzz_core_pkg::kind::KIND_DELETION,
owner_pubkey,
&super::retention::tombstone_retention_d_tag(coordinate.kind, &coordinate.d_tag),
blocks_tombstone_publication(resolution),
)
}
/// Record the relay head as this install's baseline for a coordinate.
///
/// Only ever called with a head that a completed, writer-consistent lookup
/// actually returned, which is what makes the baseline a claim about the RELAY
/// rather than about local disk. Stamping from disk content would defeat the
/// entire model: the baseline's job is to distinguish "the user edited this"
/// from "this store never learned about the newer head", and content that was
/// never published cannot answer that question.
fn stamp_baseline_from_head(
conn: &Connection,
owner_pubkey: &str,
coordinate: &Coordinate,
head: &Head,
) -> Result<(), String> {
super::retention::set_baseline(
conn,
coordinate.kind,
owner_pubkey,
&coordinate.d_tag,
&head.event_id,
&head.projection.content,
)
}
/// Run the boot decision pass for a set of coordinates and enforce the result.
///
/// This is the barrier's entry point. Ordering follows V3.4 strictly: the
/// caller performs every relay lookup BEFORE calling this, and this function
/// does only local work, so no network I/O ever happens while the store lock is
/// held.
///
/// Three outcomes are executed here, and only three: the publication gate, the
/// baseline stamp, and dropping a vacuous queued publish. All three are
/// provenance — none touches the user's records. The decisions that PATCH disk
/// (`ApplyHead`, `RestoreFromRelay`, `DeleteLocal`) are logged and gated but
/// deliberately not applied, because adopting remote content over a local
/// record is the resolution path (V3.6) and is user-driven by design.
///
/// Returns the plan it enforced, so the caller can act on the non-gate half of
/// each resolution and log a summary.
pub fn run_decision_pass(
conn: &Connection,
owner_pubkey: &str,
states: &[CoordinateState],
) -> Result<Vec<(Coordinate, Resolution)>, String> {
let plan = plan(states);
for (state, (coordinate, resolution)) in states.iter().zip(plan.iter()) {
apply_gate(
conn,
owner_pubkey,
coordinate,
&state.observation,
resolution,
)?;
// A vacuous queued publish is dropped rather than gated. Gating would
// hold it forever: it can never become publishable, because it says
// nothing the baseline does not already say.
if resolution.clear_queued_publish {
super::retention::clear_pending_sync(
conn,
coordinate.kind,
owner_pubkey,
&coordinate.d_tag,
)?;
}
// `StampBaseline` means disk and head already agree, so the head is a
// sound baseline and recording it is what gives every later boot the
// provenance to tell an edit from a stale copy. Without this the
// baseline stays `None` forever and every baseline-dependent cell is
// unreachable.
if resolution.decision == Decision::StampBaseline {
if let HeadState::Present(head) = &state.observation.head {
stamp_baseline_from_head(conn, owner_pubkey, coordinate, head)?;
}
}
}
let blocked = plan
.iter()
.filter(|(_, resolution)| blocks_publication(resolution))
.count();
let cleared = plan
.iter()
.filter(|(_, resolution)| resolution.clear_queued_publish)
.count();
tracing::info!(
target: "buzz::config_sync",
coordinates = plan.len(),
blocked,
cleared,
"config-sync decision pass complete"
);
Ok(plan)
}
#[cfg(test)]
mod tests;

950
desktop/src-tauri/src/managed_agents/config_sync/tests.rs Normal file
View File

@@ -0,0 +1,950 @@
use std::path::Path;
use super::*;
use crate::managed_agents::{
decision::{decide, ParkReason},
persona_events::build_persona_event,
retention::{
get_baseline, get_pending_sync, get_retained_event, is_publish_blocked, open_retention_db,
retain_event, set_baseline, tombstone_retention_d_tag, RetainedEvent,
},
};
use buzz_core_pkg::kind::{KIND_DELETION, KIND_PERSONA};
const OWNER: &str = "ownerpubkeyhex";
fn test_db() -> Connection {
open_retention_db(Path::new(":memory:")).unwrap()
}
fn coordinate() -> Coordinate {
Coordinate {
kind: KIND_PERSONA,
d_tag: "test-persona".to_string(),
}
}
/// The retention key the coordinate's queued tombstone occupies — a DIFFERENT
/// primary key from the coordinate itself, which is the whole reason the gate
/// has two halves.
fn tombstone_key() -> String {
tombstone_retention_d_tag(KIND_PERSONA, "test-persona")
}
fn projection(content: &str) -> CanonicalProjection {
CanonicalProjection {
content: content.to_string(),
shared: false,
}
}
/// A relay head carrying `content` at `created_at`, with an id derived from it.
fn head(content: &str, created_at: i64) -> Head {
Head {
event_id: format!("id-{content}"),
created_at,
projection: projection(content),
}
}
fn signed_persona(system_prompt: &str) -> nostr::Event {
use std::collections::BTreeMap;
let record = crate::managed_agents::AgentDefinition {
id: "test-persona".to_string(),
display_name: "Test".to_string(),
avatar_url: None,
system_prompt: system_prompt.to_string(),
runtime: None,
model: None,
provider: None,
name_pool: Vec::new(),
is_builtin: false,
is_active: true,
shared: false,
source_team: None,
source_team_persona_slug: None,
catalog_source: None,
env_vars: BTreeMap::new(),
respond_to: None,
respond_to_allowlist: Vec::new(),
parallelism: None,
created_at: "2025-01-01T00:00:00Z".to_string(),
updated_at: "2025-01-01T00:00:00Z".to_string(),
};
build_persona_event(&record)
.unwrap()
.sign_with_keys(&nostr::Keys::generate())
.unwrap()
}
/// Retain a row at the coordinate, optionally pending. Returns the signed
/// event so a caller can stamp a baseline naming it.
fn retain_row(conn: &Connection, kind: u32, d_tag: &str, pending: bool) -> nostr::Event {
retain_prompt(conn, kind, d_tag, pending, "prompt")
}
/// The same, with control over the persona's system prompt — which is what
/// [`CanonicalProjection`] reads, so it is how a test makes the queued row's
/// content differ from the baseline's.
fn retain_prompt(
conn: &Connection,
kind: u32,
d_tag: &str,
pending: bool,
system_prompt: &str,
) -> nostr::Event {
let event = signed_persona(system_prompt);
retain_event(
conn,
&RetainedEvent {
kind,
pubkey: OWNER.to_string(),
d_tag: d_tag.to_string(),
content: event.content.to_string(),
created_at: event.created_at.as_secs() as i64,
raw_event: {
use nostr::JsonUtil;
event.as_json()
},
pending_sync: pending,
event_id: Some(event.id.to_hex()),
},
)
.unwrap();
event
}
/// Re-queue an already-retained row, the way an edit or a reconcile would.
fn set_pending(conn: &Connection, kind: u32, d_tag: &str) {
conn.execute(
"UPDATE persona_events SET pending_sync = 1
WHERE kind = ?1 AND pubkey = ?2 AND d_tag = ?3",
rusqlite::params![kind, OWNER, d_tag],
)
.unwrap();
}
// ── local_state: reading the observation off a real store ──────────────────
#[test]
fn test_no_local_row_reports_nothing_queued_and_no_baseline() {
let conn = test_db();
let observation = local_state(
&conn,
OWNER,
&coordinate(),
Some(projection("v1")),
HeadState::Present(head("v1", 100)),
TombstoneEvidence::NotFound,
)
.unwrap();
assert!(observation.queued.is_none());
assert!(observation.queued_deletion_at.is_none());
assert!(observation.baseline.is_none());
}
/// The queued projection is read from the row's own `raw_event`, because those
/// are the bytes the flush loop publishes.
#[test]
fn test_pending_live_row_reports_its_own_queued_projection() {
let conn = test_db();
retain_prompt(&conn, KIND_PERSONA, "test-persona", true, "queued-prompt");
let observation = local_state(
&conn,
OWNER,
&coordinate(),
Some(projection("whatever-is-on-disk")),
HeadState::Present(head("v2", 200)),
TombstoneEvidence::NotFound,
)
.unwrap();
let queued = observation.queued.expect("the pending row must be queued");
assert!(
queued.content.contains("queued-prompt"),
"the queued projection must come from the row's raw_event, got {queued:?}"
);
assert!(
observation.queued_deletion_at.is_none(),
"a queued edit is not a queued deletion"
);
}
/// A queued DELETION lives at a different primary key than the record itself.
/// Reporting only the live row was the cross-coordinate hole from review.
#[test]
fn test_pending_tombstone_row_is_reported_for_its_target_coordinate() {
let conn = test_db();
// The live row is NOT pending; only the paired tombstone is.
retain_row(&conn, KIND_PERSONA, "test-persona", false);
let tombstone = retain_row(&conn, KIND_DELETION, &tombstone_key(), true);
let observation = local_state(
&conn,
OWNER,
&coordinate(),
Some(projection("v1")),
HeadState::Present(head("v2", 200)),
TombstoneEvidence::NotFound,
)
.unwrap();
assert_eq!(
observation.queued_deletion_at,
Some(tombstone.created_at.as_secs() as i64),
"a pending tombstone must be reported against its TARGET coordinate"
);
assert!(
observation.queued.is_none(),
"the live row is settled; only the tombstone is queued"
);
}
/// A settled tombstone row must not pin the coordinate as queued forever.
#[test]
fn test_non_pending_tombstone_row_is_not_reported_as_queued() {
let conn = test_db();
retain_row(&conn, KIND_PERSONA, "test-persona", false);
retain_row(&conn, KIND_DELETION, &tombstone_key(), false);
let observation = local_state(
&conn,
OWNER,
&coordinate(),
Some(projection("v1")),
HeadState::Present(head("v1", 100)),
TombstoneEvidence::NotFound,
)
.unwrap();
assert!(observation.queued.is_none());
assert!(observation.queued_deletion_at.is_none());
}
#[test]
fn test_stamped_baseline_is_read_back() {
let conn = test_db();
let event = retain_row(&conn, KIND_PERSONA, "test-persona", false);
set_baseline(
&conn,
KIND_PERSONA,
OWNER,
"test-persona",
&event.id.to_hex(),
"baseline-content",
)
.unwrap();
let observation = local_state(
&conn,
OWNER,
&coordinate(),
Some(projection("baseline-content")),
HeadState::Present(head("newer", 200)),
TombstoneEvidence::NotFound,
)
.unwrap();
assert_eq!(
observation.baseline.as_ref().map(|b| b.content.as_str()),
Some("baseline-content")
);
// Disk still equals the baseline, so the differing head wins — the exact
// stale-store case the fix exists for.
assert_eq!(decide(&observation).decision, Decision::ApplyHead);
}
/// A half-written baseline (id without content, or vice versa) is not a usable
/// provenance record and must read as "no baseline".
#[test]
fn test_partial_baseline_reads_as_absent() {
let conn = test_db();
retain_row(&conn, KIND_PERSONA, "test-persona", false);
conn.execute(
"UPDATE persona_events SET baseline_event_id = 'someid', baseline_content = NULL
WHERE kind = ?1 AND pubkey = ?2 AND d_tag = ?3",
rusqlite::params![KIND_PERSONA, OWNER, "test-persona"],
)
.unwrap();
let observation = local_state(
&conn,
OWNER,
&coordinate(),
Some(projection("local")),
HeadState::Present(head("remote", 200)),
TombstoneEvidence::NotFound,
)
.unwrap();
assert!(observation.baseline.is_none());
assert_eq!(decide(&observation).decision, Decision::Defer);
}
#[test]
fn test_plan_pairs_every_coordinate_with_its_decision() {
let base = Observation {
disk: Some(projection("v1")),
queued: None,
head: HeadState::Present(head("v2", 200)),
baseline: Some(projection("v1")),
queued_deletion_at: None,
tombstone: TombstoneEvidence::NotFound,
};
let states = vec![
CoordinateState {
coordinate: Coordinate {
kind: KIND_PERSONA,
d_tag: "adopts".to_string(),
},
observation: base.clone(),
},
CoordinateState {
coordinate: Coordinate {
kind: KIND_PERSONA,
d_tag: "parks".to_string(),
},
observation: Observation {
disk: Some(projection("edited")),
head: HeadState::Absent,
..base
},
},
];
let plan = plan(&states);
assert_eq!(plan.len(), 2);
assert_eq!(plan[0].0.d_tag, "adopts");
assert_eq!(plan[0].1.decision, Decision::ApplyHead);
assert_eq!(plan[1].0.d_tag, "parks");
assert_eq!(
plan[1].1.decision,
Decision::Park(ParkReason::LocalEditWithNoHead)
);
}
// ── The publication gates (V3.9) ───────────────────────────────────────────
fn resolution(decision: Decision) -> Resolution {
Resolution {
decision,
clear_queued_publish: false,
}
}
/// The live-row gate is deny-by-default over the decision set, so a variant
/// added later is withheld until someone deliberately admits it.
#[test]
fn test_exactly_the_publishing_decisions_release_the_live_gate() {
for decision in [
Decision::SuppressPublish,
Decision::Park(ParkReason::LocalEditWithNoHead),
Decision::Park(ParkReason::NoBaselineWithHead),
Decision::StampBaseline,
Decision::DeleteLocal,
Decision::Defer,
// The deletion publishes from the tombstone row; the live row it
// deletes must not race it.
Decision::PublishDeletion,
// The one adopt-the-relay cell reachable with NO baseline, so it must
// not leave an open gate behind for a row queued after the pass.
Decision::RestoreFromRelay,
] {
assert!(
blocks_publication(&resolution(decision.clone())),
"{decision:?} must withhold the live row"
);
}
for decision in [
Decision::PublishLocalEdit,
// Only reachable with a baseline present and nothing queued, so
// gating it would strand a later edit and suppress nothing.
Decision::ApplyHead,
] {
assert!(
!blocks_publication(&resolution(decision.clone())),
"{decision:?} must not withhold the live row"
);
}
}
/// Exactly one decision releases the tombstone row. Anything else — including
/// the timestamp arbitration falling through to the live cells — means the
/// queued deletion did not win.
#[test]
fn test_only_publish_deletion_releases_the_tombstone_gate() {
for decision in [
Decision::PublishLocalEdit,
Decision::ApplyHead,
Decision::RestoreFromRelay,
Decision::StampBaseline,
Decision::SuppressPublish,
Decision::DeleteLocal,
Decision::Defer,
Decision::Park(ParkReason::NoBaselineWithHead),
Decision::Park(ParkReason::LocalEditWithNoHead),
] {
assert!(
blocks_tombstone_publication(&resolution(decision.clone())),
"{decision:?} must withhold the queued tombstone"
);
}
assert!(!blocks_tombstone_publication(&resolution(
Decision::PublishDeletion
)));
}
/// The gate's whole purpose: a suppressed coordinate must actually vanish from
/// the flush loop's work list.
#[test]
fn test_suppressed_coordinate_is_withheld_from_the_flush_loop() {
let conn = test_db();
retain_row(&conn, KIND_PERSONA, "test-persona", true);
assert_eq!(
get_pending_sync(&conn).unwrap().len(),
1,
"precondition: the pending row is publishable"
);
let observation = Observation {
disk: Some(projection("v1")),
queued: None,
head: HeadState::Absent,
baseline: Some(projection("v1")),
queued_deletion_at: None,
tombstone: TombstoneEvidence::NotFound,
};
apply_gate(
&conn,
OWNER,
&coordinate(),
&observation,
&resolution(Decision::SuppressPublish),
)
.unwrap();
assert!(
get_pending_sync(&conn).unwrap().is_empty(),
"a suppressed coordinate must not reach the publisher"
);
// The content and its pending flag survive — suppression withholds, it
// does not discard what this device believes.
let row = get_retained_event(&conn, KIND_PERSONA, OWNER, "test-persona")
.unwrap()
.unwrap();
assert!(row.pending_sync);
}
/// Suppression must be reversible: once the head is visible again the same
/// coordinate resumes publishing, or the gate latches shut permanently.
#[test]
fn test_gate_reopens_when_a_later_decision_publishes() {
let conn = test_db();
retain_row(&conn, KIND_PERSONA, "test-persona", true);
let suppressed = Observation {
disk: Some(projection("v1")),
queued: None,
head: HeadState::Absent,
baseline: Some(projection("v1")),
queued_deletion_at: None,
tombstone: TombstoneEvidence::NotFound,
};
apply_gate(
&conn,
OWNER,
&coordinate(),
&suppressed,
&resolution(Decision::SuppressPublish),
)
.unwrap();
assert!(is_publish_blocked(&conn, KIND_PERSONA, OWNER, "test-persona").unwrap());
// Next boot: the head is visible and disk carries a real edit.
let recovered = Observation {
disk: Some(projection("edited")),
head: HeadState::Present(head("v1", 100)),
..suppressed
};
apply_gate(
&conn,
OWNER,
&coordinate(),
&recovered,
&resolution(Decision::PublishLocalEdit),
)
.unwrap();
assert!(!is_publish_blocked(&conn, KIND_PERSONA, OWNER, "test-persona").unwrap());
assert_eq!(
get_pending_sync(&conn).unwrap().len(),
1,
"the coordinate must be publishable again"
);
}
/// The gate is stored per coordinate: withholding one persona must not
/// withhold another, or one unreachable head would silence the whole store.
#[test]
fn test_gate_is_scoped_to_its_own_coordinate() {
let conn = test_db();
retain_row(&conn, KIND_PERSONA, "test-persona", true);
retain_row(&conn, KIND_PERSONA, "other-persona", true);
let observation = Observation {
disk: Some(projection("v1")),
queued: None,
head: HeadState::Absent,
baseline: Some(projection("v1")),
queued_deletion_at: None,
tombstone: TombstoneEvidence::NotFound,
};
apply_gate(
&conn,
OWNER,
&coordinate(),
&observation,
&resolution(Decision::SuppressPublish),
)
.unwrap();
let pending = get_pending_sync(&conn).unwrap();
assert_eq!(pending.len(), 1);
assert_eq!(pending[0].d_tag, "other-persona");
}
/// The gate has two halves because the rows have two primary keys. A pass that
/// wrote only the live half would compute a deletion suppression and enforce
/// it on the wrong row, leaving the tombstone free to publish.
#[test]
fn test_the_gate_enforces_on_the_tombstone_row_too() {
let conn = test_db();
retain_row(&conn, KIND_DELETION, &tombstone_key(), true);
assert_eq!(
get_pending_sync(&conn).unwrap().len(),
1,
"precondition: the queued tombstone is publishable"
);
let observation = Observation {
disk: None,
queued: None,
head: HeadState::Present(head("still-there", 200)),
baseline: None,
queued_deletion_at: Some(100),
tombstone: TombstoneEvidence::NotFound,
};
apply_gate(
&conn,
OWNER,
&coordinate(),
&observation,
&resolution(Decision::Park(ParkReason::NoBaselineWithHead)),
)
.unwrap();
assert!(
get_pending_sync(&conn).unwrap().is_empty(),
"a withheld deletion must not reach the publisher"
);
assert!(is_publish_blocked(&conn, KIND_DELETION, OWNER, &tombstone_key()).unwrap());
}
// ── Composition: the whole pass, not one layer ─────────────────────────────
//
// Every test above exercises one layer in isolation, which is exactly how a
// gate that can never fire passed a full suite: `decide` was tested with
// hand-built observations and `apply_gate` with hand-picked decisions, so
// nothing asserted that the decision a REAL pending row produces is one the
// gate actually blocks. These run the composition — retained row →
// `local_state` → `decide` → `apply_gate` → `get_pending_sync` — because that
// path is the product behaviour, and the seam between layers is where the bug
// lived.
/// Build the observation for a coordinate the way the barrier does, from rows
/// that are actually in the store.
fn observe(conn: &Connection, head: HeadState, disk: Option<CanonicalProjection>) -> Observation {
local_state(
conn,
OWNER,
&coordinate(),
disk,
head,
TombstoneEvidence::NotFound,
)
.unwrap()
}
/// Run the real pass over one coordinate and report what the flush loop would
/// then publish.
fn run_pass(conn: &Connection, observation: &Observation) -> (Decision, usize) {
let plan = run_decision_pass(
conn,
OWNER,
&[CoordinateState {
coordinate: coordinate(),
observation: observation.clone(),
}],
)
.unwrap();
(
plan[0].1.decision.clone(),
get_pending_sync(conn).unwrap().len(),
)
}
/// **Will's bug, end to end, and the regression test for this whole PR.**
///
/// A second install holds the OLD system prompt on disk with a FRESH retention
/// database — so no baseline, which is what a dev build against the same relay
/// always looks like. The good edit was made on the other install and is the
/// relay head. Boot's reconcile has already re-signed the stale disk content
/// at `monotonic_created_at = max(now, head + 1)` and queued it, so it would
/// win the relay's last-write-wins compare on `created_at` every time.
///
/// Composed through the real path rather than asserted on `decide` alone: 43
/// layer-isolated tests passed while the composition was a no-op, and that is
/// the gap class this closes. Zero rows publishable is the only acceptable
/// outcome.
#[test]
fn test_stale_store_with_no_baseline_cannot_revert_a_newer_head() {
let conn = test_db();
// The reconcile's queued re-sign of the stale disk content. No baseline is
// stamped: this install has never agreed to anything at this coordinate.
retain_prompt(&conn, KIND_PERSONA, "test-persona", true, "old-prompt");
assert_eq!(
get_pending_sync(&conn).unwrap().len(),
1,
"precondition: without the barrier this row publishes"
);
assert!(
get_baseline(&conn, KIND_PERSONA, OWNER, "test-persona")
.unwrap()
.is_none(),
"precondition: a fresh retention database has no baseline"
);
let observation = observe(
&conn,
// The good edit, on the relay, at an OLDER wall-clock time than the
// re-sign — which is precisely why `created_at` cannot arbitrate this.
HeadState::Present(head("good-edit", 1)),
Some(projection("old-prompt")),
);
let (decision, publishable) = run_pass(&conn, &observation);
assert_eq!(
decision,
Decision::Park(ParkReason::NoBaselineWithHead),
"the no-baseline-with-head cell must park"
);
assert_eq!(
publishable, 0,
"the stale revert reached the publisher (decision: {decision:?})"
);
// Withheld, not destroyed — the device still holds its copy, and the row
// keeps its pending flag so a later resolution can act on it.
let row = get_retained_event(&conn, KIND_PERSONA, OWNER, "test-persona")
.unwrap()
.expect("the record must survive suppression");
assert!(row.pending_sync);
}
/// The same bug in its second shape: a queued DELETION from a no-baseline
/// store against a live head. Worse than the revert, because an a-tag delete
/// destroys content this install has never seen rather than reverting it.
#[test]
fn test_stale_store_with_no_baseline_cannot_delete_a_live_head() {
let conn = test_db();
retain_row(&conn, KIND_DELETION, &tombstone_key(), true);
let observation = observe(&conn, HeadState::Present(head("still-there", 1)), None);
let (decision, publishable) = run_pass(&conn, &observation);
assert_eq!(decision, Decision::Park(ParkReason::NoBaselineWithHead));
assert_eq!(
publishable, 0,
"an unprovable deletion reached the publisher (decision: {decision:?})"
);
}
/// **The tombstone escape, composed over BOTH rows.**
///
/// A coordinate's record and its queued tombstone occupy different primary
/// keys — `(kind, owner, d_tag)` versus `(5, owner, "<kind>:<d_tag>")` — and
/// the flush loop reads them independently. A pass that computes the right
/// decision but enforces it on one key is indistinguishable from no gate at
/// all for the other, so the no-baseline invariant is only meaningful stated
/// over both.
///
/// The deletion here is deliberately NEWER than the head, which is the exact
/// input `deletion_at >= head.created_at` would resolve to `PublishDeletion`.
/// It must not get there: the no-baseline gate runs first, so an install that
/// cannot show it ever agreed to anything at this coordinate never destroys a
/// head it has not seen. Asserted through `get_pending_sync` — what the flush
/// loop actually selects — not through the decision alone.
#[test]
fn test_no_baseline_store_withholds_both_the_record_and_its_tombstone() {
let conn = test_db();
// Both halves queued at once: the reconcile's re-sign of stale disk
// content, and a tombstone targeting the same coordinate.
retain_prompt(&conn, KIND_PERSONA, "test-persona", true, "old-prompt");
let tombstone = retain_row(&conn, KIND_DELETION, &tombstone_key(), true);
assert_eq!(
get_pending_sync(&conn).unwrap().len(),
2,
"precondition: without the barrier both rows publish"
);
assert!(
get_baseline(&conn, KIND_PERSONA, OWNER, "test-persona")
.unwrap()
.is_none(),
"precondition: a fresh retention database has no baseline"
);
// The head predates the tombstone, so cell 4's timestamp arbitration would
// hand this to `PublishDeletion` if the baseline gate did not run first.
let head_created_at = tombstone.created_at.as_secs() as i64 - 1;
let observation = observe(
&conn,
HeadState::Present(head("good-edit", head_created_at)),
Some(projection("old-prompt")),
);
assert!(
observation.queued_deletion_at.unwrap() >= head_created_at,
"precondition: the queued deletion must be the newer of the two"
);
let (decision, publishable) = run_pass(&conn, &observation);
assert_eq!(
decision,
Decision::Park(ParkReason::NoBaselineWithHead),
"the no-baseline gate must take this before the timestamp compare"
);
assert_eq!(
publishable, 0,
"a row escaped the flush-selection gate (decision: {decision:?})"
);
assert!(
is_publish_blocked(&conn, KIND_PERSONA, OWNER, "test-persona").unwrap(),
"the record's own row must be gated"
);
assert!(
is_publish_blocked(&conn, KIND_DELETION, OWNER, &tombstone_key()).unwrap(),
"the tombstone row must be gated at its own primary key"
);
}
/// The mirror case, and the reason suppression cannot simply block everything
/// pending: a genuine local edit against a head this install has already
/// agreed on must still publish, or the fix freezes every agent config.
#[test]
fn test_genuine_local_edit_still_publishes() {
let conn = test_db();
let event = retain_prompt(&conn, KIND_PERSONA, "test-persona", true, "my-new-edit");
// The baseline is what this install last agreed was published; the queued
// row has since moved past it.
set_baseline(
&conn,
KIND_PERSONA,
OWNER,
"test-persona",
&event.id.to_hex(),
"published-v1",
)
.unwrap();
let observation = observe(
&conn,
HeadState::Present(head("published-v1", 100)),
Some(projection("my-new-edit")),
);
let (decision, publishable) = run_pass(&conn, &observation);
assert_eq!(
decision,
Decision::PublishLocalEdit,
"a post-baseline edit must be recognized as an edit"
);
assert_eq!(publishable, 1, "the user's edit must still reach the relay");
}
/// A vacuous re-sign — the queued row publishes exactly what the baseline
/// already says — is DROPPED, not gated. Gating would hold it forever, since
/// it can never become publishable.
#[test]
fn test_vacuous_re_sign_is_cleared_rather_than_held() {
let conn = test_db();
let event = retain_prompt(&conn, KIND_PERSONA, "test-persona", true, "agreed");
// Baseline names this exact event, so the queued projection equals it.
set_baseline(
&conn,
KIND_PERSONA,
OWNER,
"test-persona",
&event.id.to_hex(),
&event.content,
)
.unwrap();
let observation = observe(
&conn,
HeadState::Present(head("newer-elsewhere", 200)),
Some(projection(&event.content)),
);
let (_, publishable) = run_pass(&conn, &observation);
assert_eq!(publishable, 0);
let row = get_retained_event(&conn, KIND_PERSONA, OWNER, "test-persona")
.unwrap()
.unwrap();
assert!(
!row.pending_sync,
"a re-sign carrying no new intent must be dropped, not held pending forever"
);
}
/// A pending row whose disk content already equals the head is the
/// crash-convergence case: stamp the baseline and stop, rather than burn a
/// `created_at` bump republishing identical content.
#[test]
fn test_pending_row_matching_the_head_stamps_and_stops() {
let conn = test_db();
retain_row(&conn, KIND_PERSONA, "test-persona", true);
let observation = observe(
&conn,
HeadState::Present(head("agreed", 100)),
Some(projection("agreed")),
);
let (decision, publishable) = run_pass(&conn, &observation);
assert_eq!(decision, Decision::Park(ParkReason::NoBaselineWithHead));
assert_eq!(publishable, 0, "identical content must not be republished");
}
/// `StampBaseline` has to actually write the baseline. It is the input every
/// other cell depends on, so a decision computed and dropped leaves the table
/// permanently in its no-provenance state.
#[test]
fn test_stamp_baseline_records_the_head_as_provenance() {
let conn = test_db();
retain_row(&conn, KIND_PERSONA, "test-persona", false);
let observation = observe(
&conn,
HeadState::Present(head("agreed", 100)),
Some(projection("agreed")),
);
run_pass(&conn, &observation);
assert_eq!(
get_baseline(&conn, KIND_PERSONA, OWNER, "test-persona").unwrap(),
Some(("id-agreed".to_string(), "agreed".to_string())),
"the baseline must be stamped from the head, content and id together"
);
}
/// The two-boot sequence that makes the fix durable rather than a one-shot:
/// boot 1 converges and stamps the baseline, boot 2 uses that provenance to
/// recognize the stale copy the reconcile re-queued.
#[test]
fn test_stamped_baseline_lets_the_next_boot_suppress_a_revert() {
let conn = test_db();
let event = retain_prompt(&conn, KIND_PERSONA, "test-persona", false, "v1");
// The head IS the event this store already holds — the converged state
// boot 1 is supposed to recognize.
let converged = Head {
event_id: event.id.to_hex(),
created_at: event.created_at.as_secs() as i64,
projection: CanonicalProjection::from_event(&event),
};
let disk = converged.projection.clone();
// Boot 1: disk equals the head, so the baseline is stamped from it.
let settled = observe(
&conn,
HeadState::Present(converged.clone()),
Some(disk.clone()),
);
let (decision, _) = run_pass(&conn, &settled);
assert_eq!(decision, Decision::StampBaseline);
assert_eq!(
get_baseline(&conn, KIND_PERSONA, OWNER, "test-persona").unwrap(),
Some((converged.event_id.clone(), converged.projection.content)),
"boot 1 must establish provenance"
);
// Boot 2: the other install published a newer edit; the reconcile on THIS
// install re-queued its unchanged copy at a manufactured timestamp.
set_pending(&conn, KIND_PERSONA, "test-persona");
let stale = observe(
&conn,
HeadState::Present(head("good-edit", 1)),
Some(disk.clone()),
);
let (decision, publishable) = run_pass(&conn, &stale);
assert_eq!(
publishable, 0,
"with a baseline, the stale copy must be recognized and withheld \
(decision: {decision:?})"
);
// Recognized as vacuous rather than merely gated: the queued row says
// exactly what the baseline already says, so it is dropped and the newer
// head is what this coordinate should adopt.
assert_eq!(decision, Decision::ApplyHead);
assert!(
!get_retained_event(&conn, KIND_PERSONA, OWNER, "test-persona")
.unwrap()
.unwrap()
.pending_sync
);
}
/// A queued deletion from a store that CAN prove its provenance still reaches
/// the relay. Suppressing every tombstone would resurrect records the user
/// deliberately removed.
#[test]
fn test_a_provable_queued_tombstone_still_publishes() {
let conn = test_db();
let tombstone = retain_row(&conn, KIND_DELETION, &tombstone_key(), true);
// Provenance at the target coordinate: this install agreed to what is
// published there, so its deletion is arbitrable (cell 4).
retain_prompt(&conn, KIND_PERSONA, "test-persona", false, "agreed");
let event = get_retained_event(&conn, KIND_PERSONA, OWNER, "test-persona")
.unwrap()
.unwrap();
set_baseline(
&conn,
KIND_PERSONA,
OWNER,
"test-persona",
event.event_id.as_deref().unwrap(),
&event.content,
)
.unwrap();
// The head is older than the tombstone, so the deletion wins.
let observation = observe(
&conn,
HeadState::Present(head("still-there", 1)),
Some(projection(&event.content)),
);
assert_eq!(
observation.queued_deletion_at,
Some(tombstone.created_at.as_secs() as i64)
);
let (decision, publishable) = run_pass(&conn, &observation);
assert_eq!(decision, Decision::PublishDeletion);
assert_eq!(publishable, 1, "the queued deletion must reach the relay");
assert_eq!(get_pending_sync(&conn).unwrap()[0].kind, KIND_DELETION);
}

377
desktop/src-tauri/src/managed_agents/decision.rs Normal file
View File

@@ -0,0 +1,377 @@
//! The boot decision table (V3.2, as amended by V3.10 and V3.11).
//!
//! One pure function, [`decide`], maps the observable state of a single config
//! coordinate onto the action the boot pass should take. It performs no I/O so
//! that every cell is testable directly, and so the ordering between them is
//! visible in one place rather than spread across the caller.
//!
//! # Why a table instead of a content diff
//!
//! Agent configs sync as NIP-33 replaceable events resolved last-write-wins on
//! `created_at`, and the boot reconcile re-signs local disk content at
//! `monotonic_created_at = max(now, head + 1)`. A store that is merely *stale*
//! therefore mints events that are always strictly newer than the edits they
//! revert. Content equality alone cannot tell "the user changed this" from
//! "this store never learned about the newer head", so the pass needs
//! provenance — the baseline — and not just a diff.
//!
//! # Why `created_at` decides almost nothing here (V3.11)
//!
//! V3.2's original row 1 resolved a queued local edit against the relay head by
//! "greater `created_at` wins". Against this failure mode that is not
//! arbitration at all: the queued event's timestamp is manufactured by the very
//! path being stopped, so the stale store wins that comparison every single
//! time. Row 1 as written ratified the laundering vector.
//!
//! V3.11 splits it by evidence class. A queued **edit** is resolved against the
//! baseline, never against a timestamp. A queued **deletion** keeps timestamp
//! arbitration, but only downstream of the baseline gates — and it is sound
//! there precisely because every timestamp that reaches it was minted by an
//! interactive save or by the relay, never by a reconcile re-sign.
//!
//! # The suppression/deletion split (V3.10)
//!
//! Head absence and local deletion are decided on different evidence because
//! the cost of being wrong is wildly asymmetric. Missing head → stop
//! publishing, and nothing else; if that read was wrong the cost is one boot of
//! deferral. Removing a local record requires positive tombstone evidence,
//! because for a managed agent the removal is irreversible: the relay
//! projection is a strict subset of the record, so `private_key_nsec`,
//! `auth_tag`, `env_vars`, and the runtime config cannot be restored from it.
//! The retention scope key is only `owner || relay_url`, so any relay-side
//! history loss under the same URL presents as *every* coordinate being
//! head-absent at once — under a delete-on-absence rule that wipes the device.
use super::canonical_projection::CanonicalProjection;
/// What the boot pass observes about one coordinate, after the pulls have run
/// and before anything is written.
#[derive(Debug, Clone)]
pub struct Observation {
/// The canonical projection of the record currently on disk, or `None`
/// when the record is absent from the local store.
pub disk: Option<CanonicalProjection>,
/// The projection of the event QUEUED for publish at this coordinate.
///
/// This, not [`Self::disk`], is what the flush loop would put on the relay,
/// so it is what the baseline arbitration compares. The two agree in both
/// minting paths today — the reconcile signs disk content, an interactive
/// save writes both — but the table must not assume that silently: it is
/// the queued bytes that publish, and a divergence would mean deciding
/// about content that is not the content going out.
pub queued: Option<CanonicalProjection>,
/// The relay head's state. `Absent` and `LookupFailed` are distinct: only a
/// completed, writer-consistent lookup can report absence.
pub head: HeadState,
/// The projection this install last agreed was published for this
/// coordinate. `None` before the baseline is established.
pub baseline: Option<CanonicalProjection>,
/// `created_at` of a queued deletion targeting this coordinate, if one is
/// queued.
///
/// The only timestamp the table consults, and it is honest: a tombstone is
/// only ever minted by an interactive delete — a real user action at a real
/// time — never by the reconcile re-sign that manufactures timestamps.
pub queued_deletion_at: Option<i64>,
/// Positive evidence that the owner deleted this coordinate: a verified
/// owner-authored kind:5 naming it, from a scan that ran to completion.
pub tombstone: TombstoneEvidence,
}
/// A relay head: what it publishes, when it was published, and the event id
/// that published it.
///
/// The id travels with the projection because a baseline is a claim about a
/// specific event on the relay, not about content in the abstract. Stamping a
/// baseline from a projection alone would record "this is what's published"
/// with no way to tell later whether the row still holds that same event.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Head {
pub event_id: String,
pub created_at: i64,
pub projection: CanonicalProjection,
}
/// The result of looking up a coordinate's head on the relay.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum HeadState {
/// A head exists, with this projection.
Present(Head),
/// A completed lookup found no head.
///
/// Only reachable from an exact per-coordinate query
/// (`{kinds, authors, #d}`, `limit: 2`); never inferred from a combined
/// inventory, which cannot bound absence due to server-side page caps.
Absent,
/// The lookup did not complete. Never treated as absence.
LookupFailed,
}
/// Whether the owner positively deleted this coordinate.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum TombstoneEvidence {
/// A verified owner-authored kind:5 names this exact coordinate.
Found,
/// The scan ran to completion and found no tombstone for it.
NotFound,
/// The scan did not complete. Never treated as `NotFound`.
ScanIncomplete,
}
/// What the boot pass should do with one coordinate.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Decision {
/// Record that disk and head already agree. Stamps or repairs the
/// baseline; publishes nothing.
StampBaseline,
/// Adopt the relay's version onto disk and stamp the baseline from it.
ApplyHead,
/// Recreate a locally missing record from the relay head.
RestoreFromRelay,
/// A genuine post-baseline local edit: publish the queued event.
PublishLocalEdit,
/// The queued deletion wins: publish the tombstone.
PublishDeletion,
/// Remove the local record — only ever returned with positive tombstone
/// evidence.
DeleteLocal,
/// Stop publishing this coordinate but change nothing. The safe answer
/// whenever the head is absent without deletion evidence.
SuppressPublish,
/// Needs an explicit human decision; never auto-resolves on a later boot.
Park(ParkReason),
/// Not enough information yet. Nothing is written and nothing publishes.
Defer,
}
/// Why a coordinate was parked, so the resolution surface can explain it.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ParkReason {
/// Disk differs from the baseline and no head exists — there is no
/// trustworthy timestamp for the local edit, and file mtime must not be
/// used to manufacture one.
LocalEditWithNoHead,
/// Something is queued at a coordinate this install has no baseline for,
/// and the relay holds a head. V3.11 cell 2, and the field failure this
/// whole change exists to stop: a second store with its own fresh
/// retention database is ALWAYS in this state, so publishing here is
/// exactly how a stale copy reverts a good edit.
///
/// Covers a queued DELETION as well as a queued edit, and the deletion is
/// the worse half. An edit that loses this race can be re-made; a deletion
/// published against a head this install never agreed to destroys content
/// it has never seen, and the a-tag delete then propagates that removal to
/// every install that CAN see it.
NoBaselineWithHead,
}
/// The boot pass's resolution for one coordinate.
///
/// The decision and the fate of the queued publish travel together in one value
/// deliberately. They are not independent: `ApplyHead` and `PublishLocalEdit`
/// both permit publication, so a caller that took the decision and forgot to
/// drop a vacuous queued row would publish the stale copy those decisions were
/// chosen to avoid. One value cannot be half-used.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Resolution {
pub decision: Decision,
/// Drop the queued publish: it re-publishes content this install already
/// agreed was published, so it is a reconcile re-sign with nothing to say.
pub clear_queued_publish: bool,
}
impl Resolution {
fn keep(decision: Decision) -> Self {
Self {
decision,
clear_queued_publish: false,
}
}
}
/// Decide what to do with one coordinate.
///
/// Gates are evaluated in order and the first match wins. The ordering is the
/// contract: durable local intent is never silently discarded, every
/// destructive outcome is gated behind positive evidence rather than an
/// inference, and no cell publishes on the strength of a lookup that did not
/// complete.
pub fn decide(observation: &Observation) -> Resolution {
// Gate A — positive deletion evidence, checked first so a coordinate the
// owner deleted elsewhere is never "restored" by the absent-on-disk cell
// and never republished by the cells below.
if observation.tombstone == TombstoneEvidence::Found {
match &observation.head {
// A live head overrides the tombstone: a later upsert legitimately
// re-creates a coordinate, and the head is current relay truth.
HeadState::Present(_) => {}
// Deletion is proven and no head contradicts it.
HeadState::Absent => {
return Resolution::keep(if observation.disk.is_some() {
Decision::DeleteLocal
} else {
// Already gone locally; the tombstone confirms it. Nothing
// to do, but it must not be republished.
Decision::SuppressPublish
});
}
// The owner deleted this coordinate, but a re-creation reads
// exactly like this and only a completed head read tells them
// apart. Nothing publishes and nothing is removed.
HeadState::LookupFailed => return Resolution::keep(Decision::Defer),
}
}
// Gate B — nothing queued here is provable (V3.11 cells 2 and 3).
//
// Ordered ahead of BOTH arbitrations below, and that ordering is what V3.11
// turns on. With no baseline this install cannot show it ever agreed to
// anything at the coordinate, so neither its content nor its timestamps are
// evidence of intent about the relay's state — and a fresh retention
// database is exactly this state. Gate C's timestamp compare is sound only
// downstream of here.
let queued_deletion = observation.queued_deletion_at.is_some();
if observation.baseline.is_none() && (observation.queued.is_some() || queued_deletion) {
return Resolution::keep(match &observation.head {
// Cell 2 — Will's bug, as a cell. A head exists and this install
// cannot show it ever agreed to anything here, so it cannot claim
// its queued copy is newer intent. Parked, never auto-resolved.
HeadState::Present(_) => Decision::Park(ParkReason::NoBaselineWithHead),
HeadState::Absent => match observation.tombstone {
// Cell 3 — a create against a provably empty coordinate: a
// completed writer-consistent lookup says no head and a
// completed scan says no tombstone. Nothing to revert and
// nothing to resurrect, so the genuinely-new record publishes.
// A queued deletion here removes a purely local coordinate.
TombstoneEvidence::NotFound => {
if queued_deletion {
Decision::PublishDeletion
} else {
Decision::PublishLocalEdit
}
}
// `Found` already returned at gate A; it is grouped here so the
// resurrection vector stays closed if that ever changes.
// `ScanIncomplete` cannot support the absence cell 3 needs.
TombstoneEvidence::Found | TombstoneEvidence::ScanIncomplete => Decision::Defer,
},
HeadState::LookupFailed => Decision::Defer,
});
}
// Gate C — a queued deletion arbitrated against the head by `created_at`
// (V3.11 cell 4).
//
// Sound HERE where V3.2's row 1 was not, and only because of what gate B
// already excluded: both sides are honest. The tombstone was minted by an
// interactive delete at a real time and the head's timestamp is the relay's
// own; neither is the reconcile re-sign that manufactures
// `max(now, head + 1)`. A store with no baseline never reaches this
// compare, which is what stops a stale copy from deleting a head it has
// never seen.
if let Some(deletion_at) = observation.queued_deletion_at {
match &observation.head {
// Deciding a deletion from a lookup that did not complete could
// wipe a head this device simply failed to see.
HeadState::LookupFailed => return Resolution::keep(Decision::Defer),
// Nothing on the relay to contradict the deletion.
HeadState::Absent => return Resolution::keep(Decision::PublishDeletion),
// Deletion wins equality: at the same second the removal is the
// more conservative reading of intent.
HeadState::Present(head) if deletion_at >= head.created_at => {
return Resolution::keep(Decision::PublishDeletion)
}
// The head is strictly newer than the tombstone, so the coordinate
// was legitimately re-created after the delete. The tombstone is
// stale; fall through and decide the live record on its own terms.
HeadState::Present(_) => {}
}
}
// Gate D — a queued edit resolved against the baseline (V3.11 cells 1a and
// 1b), NEVER against `created_at`. Gate B returned for every no-baseline
// queue, so a baseline is present here whenever anything is queued.
if let (Some(queued), Some(baseline)) =
(observation.queued.as_ref(), observation.baseline.as_ref())
{
if baseline == queued {
// Cell 1a — the queued event publishes exactly what this install
// already agreed was published. A reconcile re-sign carrying no new
// intent, so it is dropped and the coordinate decided as if nothing
// were queued.
return Resolution {
decision: decide_unqueued(observation).decision,
clear_queued_publish: true,
};
}
// Cell 1b — the queued event differs from the last agreed publish:
// genuine post-baseline intent, and the only cell that publishes an
// edit. It still needs a completed lookup, because publishing means
// re-signing PAST the head and a lookup that did not complete cannot
// say what the head is. Deferring costs one boot; the row keeps its
// pending flag and publishes on the next pass that can see the relay.
return Resolution::keep(match observation.head {
HeadState::LookupFailed => Decision::Defer,
_ => Decision::PublishLocalEdit,
});
}
decide_unqueued(observation)
}
/// Decide a coordinate with nothing queued for publish.
///
/// Reached directly when no publish is queued, and from cell 1a once a vacuous
/// queued event has been dropped — the two states are identical from here down,
/// which is why the fall-through is a call rather than a copy of these cells.
fn decide_unqueued(observation: &Observation) -> Resolution {
let Some(disk) = observation.disk.as_ref() else {
// Absent on disk. Absence is never deletion intent; that requires a
// tombstone, handled by gate B.
return Resolution::keep(match &observation.head {
HeadState::Present(_) => Decision::RestoreFromRelay,
// No local record and no head: nothing to reconcile.
HeadState::Absent => Decision::SuppressPublish,
HeadState::LookupFailed => Decision::Defer,
});
};
Resolution::keep(match &observation.head {
// Disk already equals the head. This is the crash-convergence cell:
// inbound retains the head before writing JSON, so a crash between
// those two steps lands here and must repair the baseline rather than
// be misread as a hand edit and republished.
HeadState::Present(head) if head.projection == *disk => Decision::StampBaseline,
HeadState::Present(_) => match observation.baseline.as_ref() {
// Disk is untouched since the baseline, so the head is strictly
// newer information.
Some(baseline) if baseline == disk => Decision::ApplyHead,
// Disk differs from both: a genuine post-baseline edit. Nothing is
// queued for it yet; the next reconcile queues it and cell 1b
// publishes it.
Some(_) => Decision::PublishLocalEdit,
// No baseline: this install has no provenance for the record, so it
// cannot claim the difference is an edit.
None => Decision::Defer,
},
// V3.10 — head absent. Suppress, never delete: a wrong absence read
// costs one boot of deferral, a wrong deletion is unrecoverable.
HeadState::Absent => match observation.baseline.as_ref() {
// Disk is untouched and the head is gone. Under V3.2 this deleted
// locally; V3.10 downgrades it to suppression absent a tombstone.
Some(baseline) if baseline == disk => Decision::SuppressPublish,
// A local edit with no head to sign past: no trustworthy edit
// timestamp exists, so this needs an explicit decision.
Some(_) => Decision::Park(ParkReason::LocalEditWithNoHead),
None => Decision::Defer,
},
// Nothing is decided from a lookup that did not complete.
HeadState::LookupFailed => Decision::Defer,
})
}
#[cfg(test)]
mod tests;

554
desktop/src-tauri/src/managed_agents/decision/tests.rs Normal file
View File

@@ -0,0 +1,554 @@
use super::*;
use crate::managed_agents::config_sync::{blocks_publication, blocks_tombstone_publication};
fn projection(content: &str) -> CanonicalProjection {
CanonicalProjection {
content: content.to_string(),
shared: false,
}
}
/// A relay head carrying `content` at `created_at`. The event id is derived
/// from the content so two heads with different content never collide on id.
fn head(content: &str, created_at: i64) -> Head {
Head {
event_id: format!("id-{content}"),
created_at,
projection: projection(content),
}
}
/// A coordinate whose disk, baseline, and head all agree, with nothing queued —
/// the steady state. Individual tests mutate one axis at a time from here.
fn settled() -> Observation {
Observation {
disk: Some(projection("v1")),
queued: None,
head: HeadState::Present(head("v1", 100)),
baseline: Some(projection("v1")),
queued_deletion_at: None,
tombstone: TombstoneEvidence::NotFound,
}
}
/// Every state the table can be asked about, as a cross product of each axis.
///
/// The sweeps below are the only proofs that hold over states nobody thought
/// to write down, which is where this table's failures have actually come
/// from — the reverted-config bug was a cell no row-by-row test covered.
/// Enumerating is cheap (under a thousand states) and the coverage is total.
fn all_observations() -> impl Iterator<Item = Observation> {
let disks = [None, Some(projection("v1")), Some(projection("edited"))];
let queues = [None, Some(projection("v1")), Some(projection("edited"))];
let heads = [
HeadState::Present(head("v1", 100)),
HeadState::Present(head("v2", 200)),
HeadState::Absent,
HeadState::LookupFailed,
];
let baselines = [None, Some(projection("v1")), Some(projection("old"))];
// One tombstone older than every head and one newer, so the cell-4
// arbitration is exercised in both directions wherever it is reachable.
let deletions = [None, Some(50), Some(500)];
let tombstones = [
TombstoneEvidence::Found,
TombstoneEvidence::NotFound,
TombstoneEvidence::ScanIncomplete,
];
disks.into_iter().flat_map(move |disk| {
let queues = queues.clone();
let heads = heads.clone();
let baselines = baselines.clone();
queues.into_iter().flat_map(move |queued| {
let heads = heads.clone();
let baselines = baselines.clone();
let disk = disk.clone();
heads.into_iter().flat_map(move |head| {
let baselines = baselines.clone();
let disk = disk.clone();
let queued = queued.clone();
baselines.into_iter().flat_map(move |baseline| {
let disk = disk.clone();
let queued = queued.clone();
let head = head.clone();
deletions.into_iter().flat_map(move |queued_deletion_at| {
let disk = disk.clone();
let queued = queued.clone();
let head = head.clone();
let baseline = baseline.clone();
tombstones.into_iter().map(move |tombstone| Observation {
disk: disk.clone(),
queued: queued.clone(),
head: head.clone(),
baseline: baseline.clone(),
queued_deletion_at,
tombstone,
})
})
})
})
})
})
}
// ── The invariants ─────────────────────────────────────────────────────────
/// **Will's bug, stated as a property.** This is the whole point of V3.11.
///
/// A store with no baseline for a coordinate cannot show it ever agreed to
/// anything published there. When the relay nonetheless holds a head, that
/// store's queued content is by definition uninformed about the head — which
/// is exactly the state of a second install with a fresh retention database.
/// Its `created_at` is manufactured by `monotonic_created_at = max(now,
/// head + 1)`, so any comparison decided on it is one the stale store always
/// wins. Nothing may reach the relay from this state, for any combination of
/// the remaining axes.
///
/// Asserted against the real publication gates rather than against the
/// `Decision` enum, because the gate is what the flush loop obeys: a decision
/// that suppresses but is enforced on the wrong row is indistinguishable from
/// no gate at all.
///
/// The statement is over BOTH rows and is unconditional on what is queued. A
/// coordinate's live row and its tombstone row have different primary keys, so
/// a single-row invariant is provably insufficient — and the assertion holds
/// even where nothing is queued right now, because the gate this pass leaves
/// behind is what governs a row queued after it.
#[test]
fn test_no_baseline_with_a_live_head_releases_neither_row() {
let mut covered = 0;
for observation in all_observations() {
if observation.baseline.is_some() || !matches!(observation.head, HeadState::Present(_)) {
continue;
}
let resolution = decide(&observation);
assert!(
blocks_publication(&resolution),
"the live row escaped the gate with no baseline against a live head: \
{observation:?} -> {resolution:?}"
);
assert!(
blocks_tombstone_publication(&resolution),
"the tombstone row escaped the gate with no baseline against a live head: \
{observation:?} -> {resolution:?}"
);
covered += 1;
}
assert!(
covered > 0,
"the sweep must actually reach the guarded states"
);
}
/// The cell-2 decision behind that property, named so a future reader can see
/// which cell the sweep is protecting.
#[test]
fn test_no_baseline_with_a_live_head_parks() {
for queued in [Some(projection("stale")), None] {
for queued_deletion_at in [None, Some(500)] {
if queued.is_none() && queued_deletion_at.is_none() {
continue; // nothing queued: not cell 2
}
let observation = Observation {
disk: Some(projection("stale")),
queued: queued.clone(),
head: HeadState::Present(head("good-edit", 200)),
baseline: None,
queued_deletion_at,
tombstone: TombstoneEvidence::NotFound,
};
assert_eq!(
decide(&observation).decision,
Decision::Park(ParkReason::NoBaselineWithHead),
"cell 2 must park, not auto-resolve: {observation:?}"
);
}
}
}
/// `created_at` is never consulted for a live queued row against a head.
///
/// The head's timestamp is varied across the whole range around the queued
/// row's while every other axis is held fixed. If any decision moved, some
/// cell would be arbitrating on a value the reconcile manufactures, and the
/// stale store would win it.
///
/// Deliberately excludes queued deletions: cell 4 arbitrates those on
/// `created_at` by design, and soundly, because a tombstone is only ever
/// minted by an interactive delete.
#[test]
fn test_head_created_at_never_moves_a_live_decision() {
for observation in all_observations() {
if observation.queued_deletion_at.is_some() {
continue; // cell 4 consults timestamps on purpose
}
let HeadState::Present(present) = &observation.head else {
continue;
};
let baseline = decide(&observation);
for created_at in [1, 99, 100, 101, 1_000_000] {
let shifted = Observation {
head: HeadState::Present(Head {
created_at,
..present.clone()
}),
..observation.clone()
};
assert_eq!(
decide(&shifted),
baseline,
"the head's created_at moved a live decision: {observation:?} at {created_at}"
);
}
}
}
/// No decision path may destroy local state without positive tombstone
/// evidence. The safety property of the whole table, over the full state space
/// rather than row by row: for a managed agent the removal is irreversible,
/// since the relay projection is a strict subset of the local record.
#[test]
fn test_deletion_is_unreachable_without_tombstone_evidence() {
for observation in all_observations() {
if observation.tombstone == TombstoneEvidence::Found {
continue;
}
assert_ne!(
decide(&observation).decision,
Decision::DeleteLocal,
"reached DeleteLocal without evidence: {observation:?}"
);
}
}
/// Nothing is decided from a lookup that did not complete. A flaky network
/// must never present as a deletion or as an absence worth publishing into.
///
/// Scoped to the DECISION. Dropping a vacuous queued publish is deliberately
/// still allowed here: "the queued row says exactly what the baseline says" is
/// a purely local fact that needs no relay evidence, and the row could never
/// become publishable anyway. `test_a_queued_publish_is_only_dropped_when_it_\
/// equals_the_baseline` is what bounds that drop.
#[test]
fn test_a_failed_head_lookup_decides_nothing() {
for observation in all_observations() {
if observation.head != HeadState::LookupFailed {
continue;
}
assert_eq!(
decide(&observation).decision,
Decision::Defer,
"a failed lookup must decide nothing: {observation:?}"
);
}
}
/// A queued publish is only ever DROPPED as vacuous, never as a judgement
/// about content. Dropping it anywhere else would silently discard durable
/// local intent, which is the one thing row 1 has always guaranteed against.
#[test]
fn test_a_queued_publish_is_only_dropped_when_it_equals_the_baseline() {
for observation in all_observations() {
if !decide(&observation).clear_queued_publish {
continue;
}
assert_eq!(
observation.queued, observation.baseline,
"dropped a queued publish that carried new intent: {observation:?}"
);
assert!(observation.queued.is_some());
}
}
// ── Cell 1: a queued row against its baseline ──────────────────────────────
/// Cell 1a. The queued event publishes exactly what this install already
/// agreed was published, so it is a reconcile re-sign carrying nothing. It is
/// dropped rather than gated: it can never become publishable, so gating would
/// hold it forever.
#[test]
fn test_vacuous_re_sign_is_dropped_and_falls_through() {
let observation = Observation {
disk: Some(projection("v1")),
queued: Some(projection("v1")),
head: HeadState::Present(head("v2", 200)),
baseline: Some(projection("v1")),
queued_deletion_at: None,
tombstone: TombstoneEvidence::NotFound,
};
let resolution = decide(&observation);
assert!(resolution.clear_queued_publish);
// Fell through to the unqueued cells, which see disk untouched since the
// baseline and adopt the newer head.
assert_eq!(resolution.decision, Decision::ApplyHead);
}
/// Cell 1b. An edit made after the last agreed publish is real user intent and
/// must still reach the relay — suppression cannot simply block anything
/// queued.
#[test]
fn test_queued_edit_past_the_baseline_publishes() {
let observation = Observation {
disk: Some(projection("my-edit")),
queued: Some(projection("my-edit")),
head: HeadState::Present(head("v1", 100)),
baseline: Some(projection("v1")),
queued_deletion_at: None,
tombstone: TombstoneEvidence::NotFound,
};
let resolution = decide(&observation);
assert_eq!(resolution.decision, Decision::PublishLocalEdit);
assert!(!blocks_publication(&resolution));
}
/// The comparand is the QUEUED event, not disk. The flush loop publishes the
/// queued row's bytes, so a divergence between disk and the queued row must be
/// decided on what actually goes out. Here disk still equals the baseline
/// while the queued row does not — the queued row wins the comparison.
#[test]
fn test_arbitration_reads_the_queued_row_not_disk() {
let observation = Observation {
disk: Some(projection("v1")),
queued: Some(projection("diverged")),
head: HeadState::Present(head("v1", 100)),
baseline: Some(projection("v1")),
queued_deletion_at: None,
tombstone: TombstoneEvidence::NotFound,
};
assert_eq!(decide(&observation).decision, Decision::PublishLocalEdit);
}
// ── Cell 3: the evidence-gated carve-out ───────────────────────────────────
/// A create against a provably empty coordinate: a completed writer-consistent
/// lookup says no head and a completed scan says no tombstone. Nothing to
/// revert and nothing to resurrect, so a genuinely new agent publishes rather
/// than parking forever.
#[test]
fn test_new_coordinate_with_both_evidence_gates_publishes() {
let observation = Observation {
disk: Some(projection("brand-new")),
queued: Some(projection("brand-new")),
head: HeadState::Absent,
baseline: None,
queued_deletion_at: None,
tombstone: TombstoneEvidence::NotFound,
};
let resolution = decide(&observation);
assert_eq!(resolution.decision, Decision::PublishLocalEdit);
assert!(!blocks_publication(&resolution));
}
/// The resurrection vector. A deleted coordinate reads head-absent exactly
/// like a new one, and the relay keeps no watermark — so without the tombstone
/// gate, cell 3 would republish every record the owner ever deleted.
#[test]
fn test_a_deleted_coordinate_is_never_republished_as_new() {
let observation = Observation {
disk: Some(projection("deleted-agent")),
queued: Some(projection("deleted-agent")),
head: HeadState::Absent,
baseline: None,
queued_deletion_at: None,
tombstone: TombstoneEvidence::Found,
};
let resolution = decide(&observation);
assert_eq!(resolution.decision, Decision::DeleteLocal);
assert!(blocks_publication(&resolution));
}
/// An incomplete scan cannot support the absence cell 3 needs, so the queued
/// row is held rather than published on unproven evidence.
#[test]
fn test_new_coordinate_defers_on_an_incomplete_scan() {
let observation = Observation {
disk: Some(projection("brand-new")),
queued: Some(projection("brand-new")),
head: HeadState::Absent,
baseline: None,
queued_deletion_at: None,
tombstone: TombstoneEvidence::ScanIncomplete,
};
assert_eq!(decide(&observation).decision, Decision::Defer);
}
// ── Cell 4: tombstone-vs-live arbitration ──────────────────────────────────
/// Downstream of the baseline gates, both timestamps are honest — the
/// tombstone's was minted by an interactive delete, the head's by the relay —
/// so arbitrating on them is sound. Deletion wins equality: at the same second
/// the removal is the more conservative reading of intent.
#[test]
fn test_a_tombstone_at_or_after_the_head_publishes_the_deletion() {
for deletion_at in [100, 200] {
let observation = Observation {
queued_deletion_at: Some(deletion_at),
..settled()
};
let resolution = decide(&observation);
assert_eq!(resolution.decision, Decision::PublishDeletion);
// The tombstone row is released and the live row it deletes is
// withheld, so the record cannot race its own deletion.
assert!(!blocks_tombstone_publication(&resolution));
assert!(blocks_publication(&resolution));
}
}
/// A head strictly newer than the tombstone means the coordinate was
/// legitimately re-created after the delete. The tombstone is stale and the
/// live record is decided on its own terms.
#[test]
fn test_a_head_newer_than_the_tombstone_survives_it() {
let observation = Observation {
queued_deletion_at: Some(50),
..settled()
};
let resolution = decide(&observation);
assert_eq!(resolution.decision, Decision::StampBaseline);
assert!(blocks_tombstone_publication(&resolution));
}
/// Deciding a deletion from a lookup that did not complete could wipe a head
/// this device simply failed to see.
#[test]
fn test_a_deletion_behind_a_failed_lookup_defers() {
let observation = Observation {
head: HeadState::LookupFailed,
queued_deletion_at: Some(500),
..settled()
};
assert_eq!(decide(&observation).decision, Decision::Defer);
}
// ── V3.10: suppression is not deletion ─────────────────────────────────────
/// The core V3.10 rule. Relay-side history loss under one URL presents as
/// every coordinate being head-absent at once; under a delete-on-absence rule
/// that wipes the device.
#[test]
fn test_head_absent_with_clean_disk_suppresses_instead_of_deleting() {
let observation = Observation {
head: HeadState::Absent,
..settled()
};
let resolution = decide(&observation);
assert_eq!(resolution.decision, Decision::SuppressPublish);
assert!(blocks_publication(&resolution));
}
/// A live head is current relay truth: a later upsert legitimately re-creates
/// a coordinate, so an older tombstone must not delete it.
#[test]
fn test_live_head_overrides_an_older_tombstone() {
let observation = Observation {
tombstone: TombstoneEvidence::Found,
..settled()
};
assert_eq!(decide(&observation).decision, Decision::StampBaseline);
}
/// Proven deletion plus a head read that never completed is ambiguous: a
/// re-creation reads the same way. Neither removing nor republishing is safe.
#[test]
fn test_tombstone_with_a_failed_lookup_defers() {
let observation = Observation {
head: HeadState::LookupFailed,
tombstone: TombstoneEvidence::Found,
..settled()
};
assert_eq!(decide(&observation).decision, Decision::Defer);
}
// ── Crash convergence and the unqueued cells ───────────────────────────────
/// Inbound retains the head BEFORE writing JSON, so a crash in between leaves
/// disk equal to the head but unequal to the old baseline. That must repair
/// the baseline, not be misread as a hand edit and republished.
#[test]
fn test_disk_equal_to_head_stamps_baseline_even_when_baseline_is_stale() {
let observation = Observation {
disk: Some(projection("v2")),
head: HeadState::Present(head("v2", 200)),
baseline: Some(projection("v1")),
..settled()
};
assert_eq!(decide(&observation).decision, Decision::StampBaseline);
}
/// The same cell with no baseline at all. This is the bootstrap: without it
/// the baseline stays `None` forever and every baseline-dependent cell is
/// unreachable.
#[test]
fn test_disk_equal_to_head_stamps_baseline_when_baseline_is_missing() {
let observation = Observation {
disk: Some(projection("v2")),
head: HeadState::Present(head("v2", 200)),
baseline: None,
..settled()
};
assert_eq!(decide(&observation).decision, Decision::StampBaseline);
}
/// Disk is untouched since the baseline, so the differing head is strictly
/// newer information and wins.
#[test]
fn test_untouched_disk_adopts_a_differing_head() {
let observation = Observation {
disk: Some(projection("v1")),
head: HeadState::Present(head("v2", 200)),
baseline: Some(projection("v1")),
..settled()
};
assert_eq!(decide(&observation).decision, Decision::ApplyHead);
}
/// Without a baseline the difference cannot be attributed to a local edit, so
/// nothing is adopted and nothing is published.
#[test]
fn test_no_baseline_never_acts_on_a_difference() {
let observation = Observation {
disk: Some(projection("stale")),
head: HeadState::Present(head("current", 200)),
baseline: None,
..settled()
};
assert_eq!(decide(&observation).decision, Decision::Defer);
}
/// Absence on disk is never deletion intent; that requires a tombstone.
#[test]
fn test_missing_disk_record_with_live_head_restores() {
let observation = Observation {
disk: None,
..settled()
};
assert_eq!(decide(&observation).decision, Decision::RestoreFromRelay);
}
#[test]
fn test_missing_disk_record_and_missing_head_is_a_no_op() {
let observation = Observation {
disk: None,
head: HeadState::Absent,
..settled()
};
assert_eq!(decide(&observation).decision, Decision::SuppressPublish);
}
/// A local edit with no head has no trustworthy timestamp to sign past, and
/// file mtime must not be used to manufacture one — so it needs a human.
#[test]
fn test_local_edit_with_no_head_parks() {
let observation = Observation {
disk: Some(projection("edited")),
head: HeadState::Absent,
baseline: Some(projection("v1")),
..settled()
};
assert_eq!(
decide(&observation).decision,
Decision::Park(ParkReason::LocalEditWithNoHead)
);
}

140
desktop/src-tauri/src/managed_agents/head_lookup.rs Normal file
View File

@@ -0,0 +1,140 @@
//! Exact per-coordinate head lookups for the boot decision pass (V3.3).
//!
//! The decision table may only treat a head as ABSENT on the strength of a
//! lookup that (a) named the exact coordinate and (b) completed. Anything
//! weaker is reported as [`HeadState::LookupFailed`], which decides nothing.
//!
//! # Why absence needs its own query shape
//!
//! It is tempting to pull one inventory of all `30175/30176/30177` events for
//! the owner and treat "not in the list" as absent. That is unsound: the
//! effective server-side cap on a query is 1000 rows
//! (`crates/buzz-db/src/event.rs`), so a truncated page is indistinguishable
//! from a short one — absence would be inferred from a cut-off list. The
//! inventory may discover records, but only an exact `{kinds, authors, #d}`
//! lookup may deny them.
//!
//! # Known limitation
//!
//! On relay deployments that route historical reads through a read-replica
//! pool, a replica that has not yet replayed an event returns the same empty
//! page as a genuine deletion, with no staleness budget bounding the
//! difference. The barrier's tracing line is the detection mechanism if this
//! ever fires in the wild; re-adding a `sync_authoritative` opt-in to the
//! protocol is a small standalone change reversible on evidence.
use serde_json::json;
use super::{
canonical_projection::CanonicalProjection,
decision::{Head, HeadState},
retention::RetentionScope,
};
use crate::app_state::AppState;
/// One coordinate to look up: kind plus `d` tag, under the scope's owner.
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub struct Coordinate {
pub kind: u32,
pub d_tag: String,
}
/// Build the exact filter for one coordinate's head.
///
/// `limit: 2` rather than 1 is deliberate: NIP-33 guarantees at most one head
/// per coordinate, so a second row means the relay disagrees with that
/// assumption and the result must not be trusted as authoritative.
pub fn head_filter(owner_pubkey_hex: &str, coordinate: &Coordinate) -> serde_json::Value {
json!({
"kinds": [coordinate.kind],
"authors": [owner_pubkey_hex],
"#d": [coordinate.d_tag],
"limit": 2,
})
}
/// Interpret the events a head lookup returned for one coordinate.
///
/// Split from the network call so the interpretation rules are testable
/// without a relay. The caller passes only events it actually received; a
/// transport error must become [`HeadState::LookupFailed`] instead of an empty
/// slice, or a network failure would masquerade as a deletion.
pub fn head_state_from_events(coordinate: &Coordinate, events: &[nostr::Event]) -> HeadState {
// Defense in depth: the relay filtered by kind and `#d`, but a mismatched
// row here means the answer is about a different coordinate, so it cannot
// be used to deny this one.
let matching: Vec<&nostr::Event> = events
.iter()
.filter(|event| {
event.kind.as_u16() as u32 == coordinate.kind
&& event_d_tag(event).as_deref() == Some(coordinate.d_tag.as_str())
})
.collect();
match matching.len() {
0 => HeadState::Absent,
1 => HeadState::Present(Head {
event_id: matching[0].id.to_hex(),
// Safety: nostr timestamps are seconds and stay below i64::MAX
// until year 2262 — the same cast the retention store uses.
created_at: matching[0].created_at.as_secs() as i64,
projection: CanonicalProjection::from_event(matching[0]),
}),
// More than one head for a replaceable coordinate contradicts NIP-33.
// Refuse to decide rather than pick one.
_ => HeadState::LookupFailed,
}
}
/// The `d` tag of an event, if it has exactly one.
fn event_d_tag(event: &nostr::Event) -> Option<String> {
let mut found: Option<String> = None;
for tag in event.tags.iter() {
let parts = tag.as_slice();
if parts.len() >= 2 && parts[0].as_str() == "d" {
if found.is_some() {
// Two `d` tags: the coordinate is ambiguous.
return None;
}
found = Some(parts[1].to_string());
}
}
found
}
/// Look up one coordinate's head on the relay.
///
/// Every failure mode — transport error, malformed response, contradictory
/// result — collapses to [`HeadState::LookupFailed`] so that no caller can
/// mistake a failed lookup for an authoritative absence.
pub async fn fetch_head(
state: &AppState,
scope: &RetentionScope,
api_base_url: &str,
coordinate: &Coordinate,
) -> HeadState {
let owner_hex = scope.owner_keys.public_key().to_hex();
let filter = head_filter(&owner_hex, coordinate);
match crate::relay::query_relay_at_with_keys(
state,
api_base_url,
std::slice::from_ref(&filter),
&scope.owner_keys,
None,
)
.await
{
Ok(events) => head_state_from_events(coordinate, &events),
Err(error) => {
eprintln!(
"buzz-desktop: config-sync: head lookup failed for {}:{} — {error}",
coordinate.kind, coordinate.d_tag
);
HeadState::LookupFailed
}
}
}
#[cfg(test)]
mod tests;

144
desktop/src-tauri/src/managed_agents/head_lookup/tests.rs Normal file
View File

@@ -0,0 +1,144 @@
use super::*;
use crate::managed_agents::persona_events::build_persona_event;
use buzz_core_pkg::kind::{KIND_PERSONA, KIND_TEAM};
fn coordinate() -> Coordinate {
Coordinate {
kind: KIND_PERSONA,
d_tag: "test-persona".to_string(),
}
}
fn signed_persona(keys: &nostr::Keys, d_tag: &str, shared: bool) -> nostr::Event {
use std::collections::BTreeMap;
let record = crate::managed_agents::AgentDefinition {
id: d_tag.to_string(),
display_name: "Test".to_string(),
avatar_url: None,
system_prompt: "prompt".to_string(),
runtime: None,
model: None,
provider: None,
name_pool: Vec::new(),
is_builtin: false,
is_active: true,
shared,
source_team: None,
source_team_persona_slug: None,
catalog_source: None,
env_vars: BTreeMap::new(),
respond_to: None,
respond_to_allowlist: Vec::new(),
parallelism: None,
created_at: "2025-01-01T00:00:00Z".to_string(),
updated_at: "2025-01-01T00:00:00Z".to_string(),
};
build_persona_event(&record)
.unwrap()
.sign_with_keys(keys)
.unwrap()
}
/// The filter names the exact coordinate (kinds + authors + #d) and requests
/// limit:2 so a NIP-33 violation is detectable. This shape is what makes the
/// absence claim sound — an inventory query cannot bound absence.
#[test]
fn test_head_filter_is_exact() {
let filter = head_filter("ownerhex", &coordinate());
assert_eq!(filter["kinds"], serde_json::json!([KIND_PERSONA]));
assert_eq!(filter["authors"], serde_json::json!(["ownerhex"]));
assert_eq!(filter["#d"], serde_json::json!(["test-persona"]));
}
/// A cap of 1 could not tell "exactly one head" from "the relay truncated the
/// page", so the lookup asks for one more than NIP-33 permits.
#[test]
fn test_head_filter_can_observe_a_nip33_violation() {
assert_eq!(head_filter("ownerhex", &coordinate())["limit"], 2);
}
#[test]
fn test_empty_result_is_absent() {
assert_eq!(
head_state_from_events(&coordinate(), &[]),
HeadState::Absent
);
}
#[test]
fn test_single_matching_event_is_present_with_its_projection() {
let keys = nostr::Keys::generate();
let event = signed_persona(&keys, "test-persona", false);
assert_eq!(
head_state_from_events(&coordinate(), std::slice::from_ref(&event)),
HeadState::Present(Head {
event_id: event.id.to_hex(),
created_at: event.created_at.as_secs() as i64,
projection: CanonicalProjection::from_event(&event),
})
);
}
/// The `shared` tag has to survive the lookup, or a disk-vs-head compare would
/// report a spurious difference for a shared persona and republish it forever.
#[test]
fn test_present_head_preserves_the_shared_tag() {
let keys = nostr::Keys::generate();
let event = signed_persona(&keys, "test-persona", true);
match head_state_from_events(&coordinate(), std::slice::from_ref(&event)) {
HeadState::Present(head) => assert!(head.projection.shared),
other => panic!("expected Present, got {other:?}"),
}
}
/// Two heads for one replaceable coordinate contradicts NIP-33. Picking one
/// would make the decision depend on relay row order, so the lookup refuses.
#[test]
fn test_multiple_heads_fail_the_lookup_rather_than_guessing() {
let keys = nostr::Keys::generate();
let events = [
signed_persona(&keys, "test-persona", false),
signed_persona(&keys, "test-persona", true),
];
assert_eq!(
head_state_from_events(&coordinate(), &events),
HeadState::LookupFailed
);
}
/// A row for a different coordinate cannot be used to answer this one. If it
/// were counted as a match, an unrelated event would mask a real absence; if
/// it were merely ignored without filtering, a mismatched relay response would
/// still read as "present".
#[test]
fn test_events_for_other_coordinates_are_ignored() {
let keys = nostr::Keys::generate();
let other = signed_persona(&keys, "a-different-persona", false);
assert_eq!(
head_state_from_events(&coordinate(), std::slice::from_ref(&other)),
HeadState::Absent
);
}
/// A kind mismatch is the cross-kind d-tag collision case: d-tags are not
/// unique across kinds, so a team with the same d-tag must not answer a
/// persona lookup.
#[test]
fn test_events_of_another_kind_do_not_answer_the_lookup() {
let keys = nostr::Keys::generate();
let persona = signed_persona(&keys, "test-persona", false);
let team_coordinate = Coordinate {
kind: KIND_TEAM,
d_tag: "test-persona".to_string(),
};
assert_eq!(
head_state_from_events(&team_coordinate, std::slice::from_ref(&persona)),
HeadState::Absent
);
}

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

@@ -2,7 +2,13 @@ mod agent_env;
pub(crate) mod agent_events;
pub(crate) mod agent_snapshot;
pub(crate) mod agent_snapshot_envelope;
pub(crate) mod canonical_projection;
pub(crate) mod config_barrier;
pub(crate) mod config_sync;
pub(crate) mod decision;
pub(crate) mod head_lookup;
pub(crate) mod team_snapshot;
pub(crate) mod tombstone_scan;
pub(crate) use agent_env::{
baked_build_env, build_buzz_agent_provider_defaults, discovery_env_with_baked_floor,
};

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

@@ -208,6 +208,7 @@ pub fn open_retention_db(path: &Path) -> Result<Connection, String> {
event_id TEXT,
baseline_event_id TEXT,
baseline_content TEXT,
publish_blocked INTEGER NOT NULL DEFAULT 0,
PRIMARY KEY (kind, pubkey, d_tag)
);",
)
@@ -480,12 +481,23 @@ pub fn get_retained_personas(
/// deletion soft-deletes every live row at the coordinate with no timestamp
/// comparison, so a tombstone published AFTER the replacement would wipe it.
/// Within each group, oldest first for the same reason.
///
/// # The publication gate (V3.9)
///
/// `publish_blocked = 1` withholds a row from this list while leaving it
/// retained. The boot pass sets it when it decides a coordinate must stop
/// publishing — a head it cannot see, or a conflict parked for the user —
/// and clearing `pending_sync` would be wrong for both: the local content is
/// still the thing this device believes, and dropping the flag would lose that.
/// Suppression therefore needs a state of its own, and this query is the single
/// place it takes effect. Anything that bypasses this function bypasses the
/// gate.
pub fn get_pending_sync(conn: &Connection) -> Result<Vec<RetainedEvent>, String> {
let mut stmt = conn
.prepare(
"SELECT kind, pubkey, d_tag, content, created_at, raw_event, pending_sync, event_id
FROM persona_events
WHERE pending_sync = 1
WHERE pending_sync = 1 AND publish_blocked = 0
ORDER BY (kind != 5), created_at ASC",
)
.map_err(|e| format!("failed to prepare pending sync query: {e}"))?;
@@ -509,6 +521,57 @@ pub fn get_pending_sync(conn: &Connection) -> Result<Vec<RetainedEvent>, String>
.map_err(|e| format!("failed to read pending sync row: {e}"))
}
/// Every coordinate this owner has queued for publish, gated or not.
///
/// The boot barrier enumerates from disk, but a pending row can outlive its
/// record: a deleted persona, a removed agent, or a kind-5 tombstone all leave
/// a queued row with nothing on disk to enumerate. Those are precisely the rows
/// that publish unsupervised, so the barrier unions this with its disk walk.
///
/// Tombstone rows are mapped back to the coordinate they target, since that is
/// the coordinate the decision is about. Rows whose gate is already set are
/// still returned: a gate must be re-decided each boot so suppression can lift
/// once the condition that caused it is gone.
pub fn pending_coordinates(conn: &Connection, pubkey: &str) -> Result<Vec<(u32, String)>, String> {
let mut stmt = conn
.prepare(
"SELECT kind, d_tag FROM persona_events
WHERE pending_sync = 1 AND pubkey = ?1",
)
.map_err(|e| format!("failed to prepare pending coordinate query: {e}"))?;
let rows = stmt
.query_map(params![pubkey], |row| {
Ok((row.get::<_, u32>(0)?, row.get::<_, String>(1)?))
})
.map_err(|e| format!("failed to query pending coordinates: {e}"))?;
let mut out = Vec::new();
for row in rows {
let (kind, d_tag) = row.map_err(|e| format!("failed to read pending coordinate: {e}"))?;
out.push(if kind == buzz_core_pkg::kind::KIND_DELETION {
match parse_tombstone_retention_d_tag(&d_tag) {
Some(target) => target,
// An unparseable tombstone key names no coordinate, so there is
// nothing to decide about. Skipped rather than guessed at.
None => continue,
}
} else {
(kind, d_tag)
});
}
Ok(out)
}
/// Recover the target coordinate from a tombstone row's retention key.
///
/// Inverse of [`tombstone_retention_d_tag`]. Splits once from the left because
/// only the kind prefix is structural — a target d-tag may itself contain `:`.
fn parse_tombstone_retention_d_tag(key: &str) -> Option<(u32, String)> {
let (kind, d_tag) = key.split_once(':')?;
Some((kind.parse().ok()?, d_tag.to_string()))
}
/// Clear the `pending_sync` flag for an event the relay just confirmed.
///
/// Compare-and-clear: only clears the row if its `created_at` and `content`
@@ -535,6 +598,114 @@ pub fn mark_synced(
Ok(())
}
/// Drop a queued publish without touching the row's retained content.
///
/// Used by the boot pass for a queued event that publishes exactly what the
/// baseline already says is published — a reconcile re-sign carrying no new
/// intent. Such a row can never become publishable, so gating it would hold it
/// forever; clearing the flag is the honest outcome.
///
/// Distinct from [`mark_synced`], which clears the flag on evidence the RELAY
/// accepted those exact bytes. This clears it on evidence the relay already
/// had them.
pub fn clear_pending_sync(
conn: &Connection,
kind: u32,
pubkey: &str,
d_tag: &str,
) -> Result<(), String> {
conn.execute(
"UPDATE persona_events SET pending_sync = 0
WHERE kind = ?1 AND pubkey = ?2 AND d_tag = ?3",
params![kind, pubkey, d_tag],
)
.map_err(|e| format!("failed to clear pending sync: {e}"))?;
Ok(())
}
/// Withhold a coordinate from the flush loop without discarding its content.
///
/// Set when the boot pass decides a coordinate must stop publishing: a head it
/// could not see (V3.10 suppression) or a conflict parked for the user (V3.6).
/// The row keeps `pending_sync` and its content — this device still believes
/// that content — but [`get_pending_sync`] will not hand it to the publisher.
///
/// Idempotent, and a no-op on a coordinate with no row: there is nothing to
/// withhold until something is retained there.
pub fn set_publish_blocked(
conn: &Connection,
kind: u32,
pubkey: &str,
d_tag: &str,
blocked: bool,
) -> Result<(), String> {
conn.execute(
"UPDATE persona_events SET publish_blocked = ?4
WHERE kind = ?1 AND pubkey = ?2 AND d_tag = ?3",
params![kind, pubkey, d_tag, i32::from(blocked)],
)
.map_err(|e| format!("failed to set publish gate: {e}"))?;
Ok(())
}
/// The baseline recorded for a coordinate: what this install last agreed was
/// published there.
///
/// Read as a unit because the two columns are only meaningful together — an
/// id without content cannot be compared, and content without an id has no
/// provenance. `None` means the baseline was never established.
pub fn get_baseline(
conn: &Connection,
kind: u32,
pubkey: &str,
d_tag: &str,
) -> Result<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)?,
))
},
)
.optional()
.map_err(|e| format!("failed to get baseline: {e}"))
// Both halves must be present: a row carrying only one is not a usable
// baseline and is treated as if none was recorded.
.map(|row| row.and_then(|(id, content)| Some((id?, content?))))
}
/// Record what this install now agrees is published at a coordinate.
///
/// Only the boot decision pass and the publish-confirmation path call this: a
/// baseline is a claim about the RELAY's state, so it may only be stamped from
/// an event that is actually on the relay — never from local disk content that
/// has not been published.
pub fn set_baseline(
conn: &Connection,
kind: u32,
pubkey: &str,
d_tag: &str,
baseline_event_id: &str,
baseline_content: &str,
) -> Result<(), String> {
conn.execute(
"UPDATE persona_events
SET baseline_event_id = ?4, baseline_content = ?5
WHERE kind = ?1 AND pubkey = ?2 AND d_tag = ?3",
params![kind, pubkey, d_tag, baseline_event_id, baseline_content],
)
.map_err(|e| format!("failed to set baseline: {e}"))?;
Ok(())
}
/// Delete a retained event by its coordinate.
///
/// Called from the synchronous, lock-held delete-persona command body so the
@@ -557,6 +728,30 @@ pub fn delete_retained_event(
Ok(())
}
/// Whether a coordinate's row is currently withheld from the flush loop.
///
/// Test-only: production code never asks, because
/// [`get_pending_sync`] applies the gate in SQL rather than filtering after
/// the fact. Tests need to distinguish "withheld" from "no row at all", which
/// a `get_pending_sync` count alone cannot.
#[cfg(test)]
pub fn is_publish_blocked(
conn: &Connection,
kind: u32,
pubkey: &str,
d_tag: &str,
) -> Result<bool, String> {
conn.query_row(
"SELECT publish_blocked FROM persona_events
WHERE kind = ?1 AND pubkey = ?2 AND d_tag = ?3",
params![kind, pubkey, d_tag],
|row| row.get::<_, i32>(0),
)
.optional()
.map_err(|e| format!("failed to read publish gate: {e}"))
.map(|blocked| blocked == Some(1))
}
/// Check if the retention store has any persona events for the given pubkey.
#[cfg(test)]
pub fn has_retained_personas(conn: &Connection, pubkey: &str) -> Result<bool, String> {

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

@@ -13,6 +13,10 @@
//! published. This is what lets the boot pass tell "the user edited this
//! locally" from "this store never learned about the newer relay head",
//! instead of inferring intent from a content diff.
//! - `publish_blocked` — the publication gate. A pending row is normally
//! published by the flush loop; a row the boot pass suppressed must stay
//! retained but unpublished, and there is no other state that expresses
//! "durable, but must not go out."
//!
//! # Crash and concurrency safety
//!
@@ -27,10 +31,11 @@
use rusqlite::{Connection, TransactionBehavior};
/// Columns added after the initial `persona_events` shape.
const ADDED_COLUMNS: [(&str, &str); 3] = [
const ADDED_COLUMNS: [(&str, &str); 4] = [
("event_id", "TEXT"),
("baseline_event_id", "TEXT"),
("baseline_content", "TEXT"),
("publish_blocked", "INTEGER NOT NULL DEFAULT 0"),
];
/// Bring `persona_events` up to the current column set.

234
desktop/src-tauri/src/managed_agents/tombstone_scan.rs Normal file
View File

@@ -0,0 +1,234 @@
//! The exhaustive owner kind:5 scan (V3.3), and the evidence it yields.
//!
//! [`TombstoneEvidence::NotFound`] is a claim that the owner never deleted a
//! coordinate. Under V3.11 that claim is what lets a genuinely new record
//! publish against a provably empty coordinate, and under V3.10 its opposite is
//! the only thing that authorizes deleting a local record. Both directions are
//! consequential, so the claim may only come from a scan that ran to
//! completion — anything less reports [`TombstoneEvidence::ScanIncomplete`],
//! which decides nothing.
//!
//! # Why the scan is exhaustive rather than filtered
//!
//! The obvious query is one `#a`-filtered lookup per coordinate. The relay
//! cannot serve it soundly: `#a` is not a pushed-down constraint
//! (`filter_fully_pushable` in `crates/buzz-relay/src/handlers/req.rs` lists
//! `#a` among the post-filtered keys), so the server applies it AFTER the SQL
//! `LIMIT`. A page that fills with 1000 unrelated kind:5 events returns empty
//! for the coordinate asked about — indistinguishable from no tombstone
//! existing. So the scan pulls the owner's kind:5 history by pages, filters
//! `a`-tags locally, and only reports absence once a page comes back short of
//! its own limit.
//!
//! # Why one scan for every coordinate
//!
//! The scan is per-OWNER, not per-coordinate: one paginated walk yields the
//! evidence for every coordinate the boot pass is about to decide. A
//! per-coordinate scan would multiply the same pages by the number of records
//! on the device for an identical answer.
use std::collections::BTreeSet;
use serde_json::json;
use super::{decision::TombstoneEvidence, head_lookup::Coordinate, retention::RetentionScope};
use crate::app_state::AppState;
/// Page size for the scan. The relay clamps any request to 1000
/// (`crates/buzz-db/src/event.rs`), so asking for more would silently get 1000
/// and make a full page look short.
const SCAN_PAGE_LIMIT: usize = 1000;
/// Hard bound on pages walked in one boot, so a pathological history cannot
/// stall startup indefinitely. Hitting it yields an INCOMPLETE scan, never a
/// premature `NotFound`.
const MAX_SCAN_PAGES: usize = 20;
/// Which coordinates the owner has tombstoned, from a scan that either ran to
/// completion or did not.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct TombstonedCoordinates {
tombstoned: BTreeSet<(u32, String)>,
complete: bool,
}
impl TombstonedCoordinates {
/// The result of a scan that did not finish. Every lookup against it
/// reports [`TombstoneEvidence::ScanIncomplete`], even for coordinates the
/// partial results did name — a partial scan cannot support `NotFound`, and
/// reporting `Found` from it would make the evidence's meaning depend on
/// which page a tombstone happened to land on.
pub fn incomplete() -> Self {
Self {
tombstoned: BTreeSet::new(),
complete: false,
}
}
/// The evidence this scan provides about one coordinate.
pub fn evidence(&self, coordinate: &Coordinate) -> TombstoneEvidence {
if !self.complete {
return TombstoneEvidence::ScanIncomplete;
}
if self
.tombstoned
.contains(&(coordinate.kind, coordinate.d_tag.clone()))
{
TombstoneEvidence::Found
} else {
TombstoneEvidence::NotFound
}
}
}
/// One page of the owner's kind:5 history, oldest-bound by a keyset cursor.
///
/// # Known limitation
///
/// On relay deployments that route historical reads through a read-replica
/// pool, a replica that has not yet replayed a tombstone answers "not found"
/// in the dangerous direction. The barrier's tracing line surfaces this if
/// it ever fires; re-adding `sync_authoritative` to the protocol is a small
/// standalone change reversible on observed evidence.
pub fn scan_page_filter(owner_pubkey_hex: &str, cursor: Option<&PageCursor>) -> serde_json::Value {
let mut filter = json!({
"kinds": [buzz_core_pkg::kind::KIND_DELETION],
"authors": [owner_pubkey_hex],
"limit": SCAN_PAGE_LIMIT,
});
if let Some(cursor) = cursor {
// Composite keyset cursor: the relay requires both halves or neither,
// and rejects `before_id` without `until`. A timestamp-only cursor
// would drop every event tied at the page boundary's second.
filter["until"] = json!(cursor.until);
filter["before_id"] = json!(cursor.before_id);
}
filter
}
/// The `(created_at, id)` pair that bounds the next page.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PageCursor {
pub until: i64,
pub before_id: String,
}
/// The cursor for the page after `events`, or `None` when this page ends the
/// walk.
///
/// A page shorter than the limit it asked for is the ONLY proof of exhaustion.
/// `rows == limit` never proves there is more, but it never proves there isn't
/// either, so the walk continues until a page comes back short.
pub fn next_cursor(events: &[nostr::Event]) -> Option<PageCursor> {
if events.len() < SCAN_PAGE_LIMIT {
return None;
}
let last = events.last()?;
Some(PageCursor {
until: last.created_at.as_secs() as i64,
before_id: last.id.to_hex(),
})
}
/// The config coordinates a page of kind:5 events tombstones.
///
/// `a`-tags are parsed locally with the same owner-scoping rule the inbound
/// tombstone path uses: only the coordinate's own author may tombstone it, so
/// a validly signed kind:5 naming another owner's coordinate names nothing
/// here. Coordinates outside the three config kinds are ignored — the owner's
/// kind:5 history includes deletions this pass has no opinion about.
pub fn tombstoned_in_page(events: &[nostr::Event]) -> BTreeSet<(u32, String)> {
use super::canonical_projection::CanonicalProjection;
events
.iter()
.filter_map(|event| {
let (kind, d_tag) = deletion_coordinate(event)?;
CanonicalProjection::is_config_kind(kind).then_some((kind, d_tag))
})
.collect()
}
/// Parse a NIP-09 `a`-tag coordinate `<kind>:<owner_pubkey>:<d_tag>` into its
/// target kind and d-tag, rejecting a coordinate the event's author does not
/// own.
fn deletion_coordinate(event: &nostr::Event) -> Option<(u32, String)> {
event.tags.iter().find_map(|tag| {
let values = tag.as_slice();
if values.first().map(|v| v.as_str()) != Some("a") {
return None;
}
// `<kind>:<owner>:<d_tag>` — the d-tag may itself contain ':', so split
// at most twice and keep the remainder whole.
let mut parts = values.get(1)?.splitn(3, ':');
let kind: u32 = parts.next()?.parse().ok()?;
if parts.next()? != event.pubkey.to_hex() {
return None;
}
Some((kind, parts.next()?.to_string()))
})
}
/// Walk the owner's kind:5 history to exhaustion.
///
/// Every failure mode — a transport error, a page that never ends — yields
/// [`TombstonedCoordinates::incomplete`], so a scan that could not prove
/// absence never claims it.
pub async fn scan_tombstones(
state: &AppState,
scope: &RetentionScope,
api_base_url: &str,
) -> TombstonedCoordinates {
let owner_hex = scope.owner_keys.public_key().to_hex();
let mut tombstoned = BTreeSet::new();
let mut cursor: Option<PageCursor> = None;
for _ in 0..MAX_SCAN_PAGES {
let filter = scan_page_filter(&owner_hex, cursor.as_ref());
let events = match crate::relay::query_relay_at_with_keys(
state,
api_base_url,
std::slice::from_ref(&filter),
&scope.owner_keys,
None,
)
.await
{
Ok(events) => events,
Err(error) => {
tracing::warn!(
target: "buzz::config_sync",
%error,
"tombstone scan page failed; evidence is incomplete"
);
return TombstonedCoordinates::incomplete();
}
};
tombstoned.extend(tombstoned_in_page(&events));
match next_cursor(&events) {
Some(next) => cursor = Some(next),
None => {
tracing::info!(
target: "buzz::config_sync",
tombstoned = tombstoned.len(),
"tombstone scan complete"
);
return TombstonedCoordinates {
tombstoned,
complete: true,
};
}
}
}
tracing::warn!(
target: "buzz::config_sync",
max_pages = MAX_SCAN_PAGES,
"tombstone scan hit its page bound; evidence is incomplete"
);
TombstonedCoordinates::incomplete()
}
#[cfg(test)]
mod tests;

224
desktop/src-tauri/src/managed_agents/tombstone_scan/tests.rs Normal file
View File

@@ -0,0 +1,224 @@
use super::*;
use buzz_core_pkg::kind::{KIND_DELETION, KIND_MANAGED_AGENT, KIND_PERSONA, KIND_TEAM};
use nostr::{EventBuilder, Keys, Kind, Tag};
/// A signed kind:5 carrying one `a`-tag, with control over whose coordinate it
/// names — the owner-scoping rule is what makes a foreign tombstone inert.
fn deletion_of(keys: &Keys, coordinate_owner_hex: &str, kind: u32, d_tag: &str) -> nostr::Event {
let coord = format!("{kind}:{coordinate_owner_hex}:{d_tag}");
EventBuilder::new(Kind::Custom(KIND_DELETION as u16), "")
.tags(vec![Tag::parse(["a", coord.as_str()]).unwrap()])
.sign_with_keys(keys)
.unwrap()
}
/// A kind:5 deleting by event id instead of coordinate. It names no
/// parameterized coordinate at all.
fn deletion_by_event_id(keys: &Keys) -> nostr::Event {
EventBuilder::new(Kind::Custom(KIND_DELETION as u16), "")
.tags(vec![Tag::parse(["e", &"ab".repeat(32)]).unwrap()])
.sign_with_keys(keys)
.unwrap()
}
fn coordinate(kind: u32, d_tag: &str) -> Coordinate {
Coordinate {
kind,
d_tag: d_tag.to_string(),
}
}
// ── The evidence contract ──────────────────────────────────────────────────
/// An incomplete scan decides nothing. Reporting `Found` from partial results
/// would make the evidence's meaning depend on which page a tombstone landed
/// on, and reporting `NotFound` would be the unsound direction outright.
#[test]
fn test_an_incomplete_scan_reports_no_evidence_for_any_coordinate() {
let scan = TombstonedCoordinates::incomplete();
assert_eq!(
scan.evidence(&coordinate(KIND_PERSONA, "anything")),
TombstoneEvidence::ScanIncomplete
);
}
#[test]
fn test_a_complete_scan_distinguishes_tombstoned_from_untouched() {
let keys = Keys::generate();
let owner = keys.public_key().to_hex();
let events = [deletion_of(&keys, &owner, KIND_PERSONA, "deleted")];
let scan = TombstonedCoordinates {
tombstoned: tombstoned_in_page(&events),
complete: true,
};
assert_eq!(
scan.evidence(&coordinate(KIND_PERSONA, "deleted")),
TombstoneEvidence::Found
);
assert_eq!(
scan.evidence(&coordinate(KIND_PERSONA, "never-deleted")),
TombstoneEvidence::NotFound
);
// Same d-tag under a different kind is a different coordinate.
assert_eq!(
scan.evidence(&coordinate(KIND_TEAM, "deleted")),
TombstoneEvidence::NotFound
);
}
// ── Parsing a page into coordinates ────────────────────────────────────────
/// Only the coordinate's own author may tombstone it. A validly signed kind:5
/// naming someone else's coordinate is inert, or one owner could suppress
/// another's config by publishing a deletion for it.
#[test]
fn test_a_tombstone_for_another_owners_coordinate_names_nothing() {
let keys = Keys::generate();
let stranger = Keys::generate().public_key().to_hex();
let tombstoned = tombstoned_in_page(&[deletion_of(&keys, &stranger, KIND_PERSONA, "victim")]);
assert!(
tombstoned.is_empty(),
"a foreign coordinate must not be treated as deleted: {tombstoned:?}"
);
}
/// The owner's kind:5 history covers deletions this pass has no opinion about.
#[test]
fn test_deletions_outside_the_config_kinds_are_ignored() {
let keys = Keys::generate();
let owner = keys.public_key().to_hex();
let tombstoned = tombstoned_in_page(&[
deletion_of(&keys, &owner, 30023, "some-article"),
deletion_of(&keys, &owner, KIND_MANAGED_AGENT, "agent"),
]);
assert_eq!(
tombstoned,
BTreeSet::from([(KIND_MANAGED_AGENT, "agent".to_string())])
);
}
/// A d-tag may itself contain `:`, so the coordinate splits at most twice and
/// keeps the remainder whole. Truncating here would report evidence about a
/// coordinate that does not exist while missing the one that does.
#[test]
fn test_a_d_tag_containing_colons_survives_parsing() {
let keys = Keys::generate();
let owner = keys.public_key().to_hex();
let tombstoned = tombstoned_in_page(&[deletion_of(&keys, &owner, KIND_PERSONA, "a:b:c")]);
assert_eq!(
tombstoned,
BTreeSet::from([(KIND_PERSONA, "a:b:c".to_string())])
);
}
#[test]
fn test_an_event_id_deletion_names_no_coordinate() {
let keys = Keys::generate();
assert!(tombstoned_in_page(&[deletion_by_event_id(&keys)]).is_empty());
}
// ── Pagination ─────────────────────────────────────────────────────────────
/// A short page is the ONLY proof the walk is exhausted.
#[test]
fn test_a_short_page_ends_the_walk() {
let keys = Keys::generate();
let owner = keys.public_key().to_hex();
assert_eq!(
next_cursor(&[deletion_of(&keys, &owner, KIND_PERSONA, "one")]),
None
);
assert_eq!(next_cursor(&[]), None);
}
/// A page that exactly fills its limit proves nothing about exhaustion, so the
/// walk continues from the oldest event on it.
#[test]
fn test_a_full_page_continues_from_its_oldest_event() {
let keys = Keys::generate();
let owner = keys.public_key().to_hex();
let filler = deletion_of(&keys, &owner, KIND_PERSONA, "filler");
let oldest = deletion_of(&keys, &owner, KIND_PERSONA, "oldest");
let mut page = vec![filler; SCAN_PAGE_LIMIT - 1];
page.push(oldest.clone());
assert_eq!(
next_cursor(&page),
Some(PageCursor {
until: oldest.created_at.as_secs() as i64,
before_id: oldest.id.to_hex(),
}),
"the cursor must be the composite keyset of the page's last event"
);
}
// ── The page filter ────────────────────────────────────────────────────────
/// The page filter names the owner's kind:5 events with the SCAN_PAGE_LIMIT
/// cap. The scan is per-owner (not per-coordinate) so one walk covers every
/// coordinate the barrier is deciding on.
#[test]
fn test_page_filter_covers_owner_deletions() {
let with_cursor = scan_page_filter(
"ownerhex",
Some(&PageCursor {
until: 100,
before_id: "abc".to_string(),
}),
);
for filter in [scan_page_filter("ownerhex", None), with_cursor] {
assert_eq!(filter["kinds"], serde_json::json!([KIND_DELETION]));
assert_eq!(filter["authors"], serde_json::json!(["ownerhex"]));
}
}
/// The relay rejects `before_id` without `until` and drops events tied at the
/// boundary second given `until` alone, so the cursor is all-or-nothing.
#[test]
fn test_the_cursor_halves_are_both_present_or_both_absent() {
let first = scan_page_filter("ownerhex", None);
assert!(first.get("until").is_none());
assert!(first.get("before_id").is_none());
let next = scan_page_filter(
"ownerhex",
Some(&PageCursor {
until: 100,
before_id: "abc".to_string(),
}),
);
assert_eq!(next["until"], serde_json::json!(100));
assert_eq!(next["before_id"], serde_json::json!("abc"));
}
/// Asking for more than the relay's clamp would silently get the clamp back
/// and make a full page look short — which is exactly the state that ends the
/// walk and licenses a `NotFound`.
#[test]
fn test_the_page_limit_does_not_exceed_the_relay_clamp() {
/// The relay's own clamp (`crates/buzz-db/src/event.rs`).
const RELAY_CLAMP: u64 = 1000;
let limit = scan_page_filter("ownerhex", None)["limit"]
.as_u64()
.expect("the filter must carry a numeric limit");
assert_eq!(limit, SCAN_PAGE_LIMIT as u64);
assert!(
limit <= RELAY_CLAMP,
"asking for {limit} would silently get {RELAY_CLAMP} back and make a full page look short"
);
}