fix: Thufir pass-1 findings — readiness enforcement, user-intent gate, atomic mark-synced, baseline join

Four correctness fixes on top of the am/peon feedback round:

1. publish_prepared_persona_via_flush now checks the readiness latch
   (ReadinessState::Ready for exactly the prepared scope's db_path) before
   calling flush_pending_events_at. Unready or InProgress → row stays
   pending, returns Queued. Avoids publishing during boot/workspace
   reconcile window before scope-wide arbitration completes.
   New test: test_publish_via_flush_blocked_when_latch_not_ready uses an
   accepting relay so a bypass would flip the assertion.

2. retain_agent_record gains a user_intent: bool parameter. Interactive
   saves pass true (clears publish_blocked even on unchanged-content
   early-return); boot reconcile passes false. tombstone_managed_agent_pending
   and tombstone_team_pending route through retain_user_intent_event.
   snapshot import callers (team_snapshot, personas/snapshot/import) use
   retain_user_intent_event. All three config kinds now reopen gated rows
   on explicit user action.
   New tests: test_retain_agent_record_user_intent_clears_gate_but_reconcile_does_not
   (via retain_agent_record production seam) and
   test_non_persona_tombstone_user_intent_clears_gate (kind-5 tombstone path).

3. mark_synced_and_stamp_baseline wraps both SQL statements in a single
   SQLite transaction (unchecked_transaction). The baseline is only stamped
   when the compare-and-clear affected the intended row (rows_changed == 1).
   A crash or error between the two statements can no longer produce
   pending_sync=0 without a baseline stamp.
   New test: test_mark_synced_and_stamp_baseline_atomic_both_written_on_success
   verifies both pending_sync cleared AND baseline stamped in one call.

4. quarantine_unprovable_pending replaces the stale
   baseline_event_id IS NULL column check with a NOT EXISTS subquery
   against persona_baselines (the table baselines now live in).
   publish_prepared_persona_via_flush replaces the aggregate-flushed-count
   + row-absence published heuristic with a per-coordinate check: baseline
   stamped with the expected event_id (primary) OR row pending_sync cleared
   with matching content (fallback).

Co-authored-by: Will Pfleger <pfleger.will@gmail.com>
Signed-off-by: Will Pfleger <pfleger.will@gmail.com>
This commit is contained in:
npub1mn7jgtj4w2pd0g0zeuhxsa6jy6p0rewxz4kujt98my82ahfmp72sxjexk7
2026-08-01 15:23:24 -04:00
parent 4c2c84527f
commit 7e3823a802
10 changed files with 511 additions and 60 deletions

View File

@@ -51,8 +51,11 @@ pub(super) fn retain_managed_agent_pending(
let conn = open_retention_db(&scope.db_path)?;
// Shared engine with the boot-time reconcile: projection content diff
// (no republish for runtime-only churn) + monotonic created_at bump
// past the retained head (NIP-AP step 3).
retain_agent_record(&conn, &scope.owner_keys, record).map(|_| ())
// past the retained head (NIP-AP step 3). User-intent=true clears any
// stale publish_blocked gate so an edit to a parked agent takes
// effect in-session (the gate was set against a prior boot observation,
// not the user's current action).
retain_agent_record(&conn, &scope.owner_keys, record, true).map(|_| ())
})();
if let Err(e) = result {
eprintln!("buzz-desktop: agent-retain: {e}");
@@ -78,8 +81,8 @@ pub(super) fn tombstone_managed_agent_pending(
use crate::managed_agents::{
agent_events::build_agent_delete,
retention::{
delete_retained_event, open_retention_db, retain_event, tombstone_retention_d_tag,
RetainedEvent,
delete_retained_event, open_retention_db, retain_user_intent_event,
tombstone_retention_d_tag, RetainedEvent,
},
};
use buzz_core_pkg::kind::KIND_MANAGED_AGENT;
@@ -94,7 +97,7 @@ pub(super) fn tombstone_managed_agent_pending(
.map_err(|e| format!("failed to sign managed-agent tombstone: {e}"))?;
let conn = open_retention_db(&scope.db_path)?;
delete_retained_event(&conn, KIND_MANAGED_AGENT, &owner_pubkey, agent_pubkey)?;
retain_event(
retain_user_intent_event(
&conn,
&RetainedEvent::pending(
KIND_DELETE,

View File

@@ -4,7 +4,7 @@ use crate::{
app_state::AppState,
managed_agents::{
load_personas,
retention::{get_retained_event, open_retention_db},
retention::{get_baseline, get_retained_event, open_retention_db},
AgentDefinition,
},
};
@@ -92,26 +92,51 @@ pub async fn update_persona_and_publish(
/// Retain the prepared event (already done by `prepare_persona_publication`
/// as user intent, gate cleared) then trigger a flush pass and await the
/// relay outcome.
/// relay outcome for this specific scope.
///
/// This is the single publish path — it goes through the gated flush loop,
/// which checks `publish_blocked` and the readiness latch. Routing here
/// instead of submitting directly ensures:
/// Readiness is checked BEFORE flushing: if the scope is `Unready` or
/// `InProgress`, the row stays pending and `Queued` is returned — the boot
/// reconcile+barrier transition will complete readiness and the flush loop
/// will publish the row once the scope is `Ready`. This must NOT be
/// short-circuited by re-resolving the active scope: the command already
/// captured a coherent `RetentionScope` and a workspace switch mid-flight
/// must not redirect the publish to a different database.
///
/// 1. A parked/suppressed coordinate cannot bypass its own gate.
/// 2. Baseline provenance is stamped on relay acceptance (via
/// `mark_synced_and_stamp_baseline` inside the flush loop).
/// 3. The `Queued` report means the row is retained and will flush on the
/// next sweep — not "could not even enqueue."
/// When the scope IS ready, a targeted `flush_pending_events_at` is run
/// for exactly this scope's `db_path`. The row was retained as user intent
/// (publish_blocked = 0), so it will be in the pending queue.
async fn publish_prepared_persona_via_flush(
state: &AppState,
prepared: PreparedPersonaPublication,
) -> Result<SetPersonaSharedResult, String> {
use crate::managed_agents::persona_events::flush_pending_events_at;
use crate::managed_agents::{
config_sync_readiness::{self, ReadinessState},
persona_events::flush_pending_events_at,
};
// Check readiness for the PREPARED scope's database, not the currently
// active scope (a workspace switch mid-flight must not redirect the
// publish to a different db_path).
let ready = matches!(
config_sync_readiness::readiness_state(),
Some(ReadinessState::Ready(ref p)) if *p == prepared.scope.db_path
);
if !ready {
// Not yet ready — row is durably retained as user intent and will
// flush once the boot barrier completes. Report Queued.
return Ok(SetPersonaSharedResult {
persona: prepared.persona,
publication_status: PersonaSharePublicationStatus::Queued,
relay_message: Some(
"event queued for publication; boot barrier not yet complete for this scope"
.to_string(),
),
});
}
// Run a targeted flush for this scope. The event is already retained as
// user intent (publish_blocked = 0), so it will be in the pending queue.
let flushed = flush_pending_events_at(
flush_pending_events_at(
&prepared.scope.db_path,
state,
&prepared.scope.relay_url,
@@ -120,10 +145,18 @@ async fn publish_prepared_persona_via_flush(
.await
.unwrap_or(0);
// Determine whether the specific row published: re-read pending_sync.
// If flushed > 0 and the row is no longer pending, it published.
// If the relay rejected/was unreachable, pending_sync stays 1.
// Determine whether the specific row published: re-read the retained row.
// The row published if its pending_sync is now 0 AND the baseline at this
// coordinate was stamped with the expected event_id (set atomically by
// mark_synced_and_stamp_baseline on relay acceptance). This is
// per-coordinate — another row flushing cannot produce a false-positive.
let conn = open_retention_db(&prepared.scope.db_path)?;
let baseline = get_baseline(
&conn,
prepared.retained.kind,
&prepared.retained.pubkey,
&prepared.retained.d_tag,
)?;
let row = get_retained_event(
&conn,
prepared.retained.kind,
@@ -131,14 +164,13 @@ async fn publish_prepared_persona_via_flush(
&prepared.retained.d_tag,
)?;
let published = row.as_ref().is_some_and(|r| {
// The row published if pending_sync is now 0 AND it still carries the
// same content (a newer edit would also have pending_sync=0 only if it
// too published, which is fine — we just report what actually happened).
let published = baseline.as_ref().is_some_and(|(baseline_event_id, _)| {
*baseline_event_id == prepared.retained.event_id.as_deref().unwrap_or("")
}) || row.as_ref().is_some_and(|r| {
!r.pending_sync
&& r.created_at == prepared.retained.created_at
&& r.content == prepared.retained.content
}) || flushed > 0 && row.as_ref().is_none_or(|r| !r.pending_sync);
});
if published {
Ok(SetPersonaSharedResult {
@@ -240,6 +272,7 @@ mod tests {
#[tokio::test]
async fn relay_rejection_stays_durably_queued() {
use crate::managed_agents::config_sync_readiness;
let dir = tempfile::tempdir().unwrap();
let db_path = dir.path().join("retention.db");
let keys = nostr::Keys::generate();
@@ -247,6 +280,10 @@ mod tests {
let prepared = prepared(&db_path, spawn_relay(false).await, keys, Some(true));
let state = build_app_state();
// Mark this scope Ready so Queued is driven by relay rejection, not latch.
let claim = config_sync_readiness::claim_in_progress().unwrap();
claim.complete(db_path.clone());
let result = publish_prepared_persona_via_flush(&state, prepared)
.await
.unwrap();
@@ -274,6 +311,7 @@ mod tests {
#[tokio::test]
async fn unavailable_relay_stays_durably_queued() {
use crate::managed_agents::config_sync_readiness;
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let relay_url = format!("http://{}", listener.local_addr().unwrap());
drop(listener);
@@ -284,6 +322,10 @@ mod tests {
let prepared = prepared(&db_path, relay_url, keys, Some(true));
let state = build_app_state();
// Mark this scope Ready so Queued is driven by relay unavailability, not latch.
let claim = config_sync_readiness::claim_in_progress().unwrap();
claim.complete(db_path.clone());
let result = publish_prepared_persona_via_flush(&state, prepared)
.await
.unwrap();
@@ -311,6 +353,7 @@ mod tests {
#[tokio::test]
async fn relay_acceptance_marks_the_scoped_head_synced() {
use crate::managed_agents::config_sync_readiness;
let dir = tempfile::tempdir().unwrap();
let db_path = dir.path().join("retention.db");
let keys = nostr::Keys::generate();
@@ -318,6 +361,10 @@ mod tests {
let prepared = prepared(&db_path, spawn_relay(true).await, keys, Some(true));
let state = build_app_state();
// Mark this scope Ready so the readiness gate permits the flush.
let claim = config_sync_readiness::claim_in_progress().unwrap();
claim.complete(db_path.clone());
let result = publish_prepared_persona_via_flush(&state, prepared)
.await
.unwrap();
@@ -344,6 +391,7 @@ mod tests {
/// head already says, and it reports the relay outcome to the caller.
#[tokio::test]
async fn test_update_and_publish_acceptance_publishes_the_edit_at_the_current_share_state() {
use crate::managed_agents::config_sync_readiness;
let dir = tempfile::tempdir().unwrap();
let db_path = dir.path().join("retention.db");
let keys = nostr::Keys::generate();
@@ -353,6 +401,10 @@ mod tests {
let prepared = prepared(&db_path, spawn_relay(true).await, keys, None);
let state = build_app_state();
// Mark this scope Ready so the readiness gate permits the flush.
let claim = config_sync_readiness::claim_in_progress().unwrap();
claim.complete(db_path.clone());
let result = publish_prepared_persona_via_flush(&state, prepared)
.await
.unwrap();
@@ -380,6 +432,7 @@ mod tests {
#[tokio::test]
async fn test_update_and_publish_relay_rejection_reports_queued_not_failure() {
use crate::managed_agents::config_sync_readiness;
let dir = tempfile::tempdir().unwrap();
let db_path = dir.path().join("retention.db");
let keys = nostr::Keys::generate();
@@ -388,6 +441,10 @@ mod tests {
let prepared = prepared(&db_path, spawn_relay(false).await, keys, None);
let state = build_app_state();
// Mark this scope Ready so Queued is driven by relay rejection, not latch.
let claim = config_sync_readiness::claim_in_progress().unwrap();
claim.complete(db_path.clone());
let result = publish_prepared_persona_via_flush(&state, prepared)
.await
.unwrap();
@@ -427,4 +484,89 @@ mod tests {
assert!(error.contains("failed to open retention db"));
}
/// `publish_prepared_persona_via_flush` must enforce the readiness latch:
/// when the scope is `Unready` or `InProgress`, the row stays pending and
/// the caller receives `Queued` — no relay request is made. This covers the
/// Thufir pass-1 IMPORTANT finding: the old helper skipped the latch entirely
/// (it called `flush_pending_events_at` directly).
///
/// Mutation-sensitive: removing the readiness check from
/// `publish_prepared_persona_via_flush` lets `flush_pending_events_at` run
/// regardless of latch state, which — when an accepting relay is present —
/// would return `Published` and leave `pending_sync = 0`. Both the
/// `Queued` assertion and the `pending_sync` assertion would then fail,
/// catching the regression.
#[tokio::test]
async fn test_publish_via_flush_blocked_when_latch_not_ready() {
use crate::managed_agents::config_sync_readiness;
// Thread-isolated latch (test builds use thread_local!).
config_sync_readiness::mark_unready();
let dir = tempfile::tempdir().unwrap();
let db_path = dir.path().join("retention.db");
let keys = nostr::Keys::generate();
let owner = keys.public_key().to_hex();
// Use an accepting relay so the test would catch a bypass.
let relay_url = spawn_relay(true).await;
let state = build_app_state();
// ── Unready ─────────────────────────────────────────────────────────
let result = publish_prepared_persona_via_flush(
&state,
prepared(&db_path, relay_url.clone(), keys.clone(), Some(true)),
)
.await
.unwrap();
assert_eq!(
result.publication_status,
PersonaSharePublicationStatus::Queued,
"must report Queued when latch is Unready"
);
assert!(
get_retained_event(
&open_retention_db(&db_path).unwrap(),
buzz_core_pkg::kind::KIND_PERSONA,
&owner,
"catalog-reviewer"
)
.unwrap()
.unwrap()
.pending_sync,
"row must stay pending when latch is Unready"
);
// ── InProgress ───────────────────────────────────────────────────────
// Simulate the boot barrier claiming the latch (InProgress).
let _claim =
config_sync_readiness::claim_in_progress().expect("claim must succeed from Unready");
let result2 = publish_prepared_persona_via_flush(
&state,
prepared(&db_path, relay_url, keys, Some(true)),
)
.await
.unwrap();
assert_eq!(
result2.publication_status,
PersonaSharePublicationStatus::Queued,
"must report Queued when latch is InProgress"
);
assert!(
get_retained_event(
&open_retention_db(&db_path).unwrap(),
buzz_core_pkg::kind::KIND_PERSONA,
&owner,
"catalog-reviewer"
)
.unwrap()
.unwrap()
.pending_sync,
"row must stay pending when latch is InProgress"
);
// Cleanup: drop the claim so the latch resets to Unready.
config_sync_readiness::mark_unready();
}
}

View File

@@ -756,7 +756,9 @@ fn retain_agent_pending(app: &AppHandle, state: &AppState, record: &ManagedAgent
use crate::managed_agents::{
agent_events::{agent_event_content, build_agent_event},
persona_events::monotonic_created_at,
retention::{get_retained_event, open_retention_db, retain_event, RetainedEvent},
retention::{
get_retained_event, open_retention_db, retain_user_intent_event, RetainedEvent,
},
};
use buzz_core_pkg::kind::KIND_MANAGED_AGENT;
@@ -779,7 +781,10 @@ fn retain_agent_pending(app: &AppHandle, state: &AppState, record: &ManagedAgent
.map_err(|e| format!("failed to sign agent event: {e}"))?;
(owner_pubkey, event)
};
retain_event(
// User-intent: a snapshot import is an explicit user action — clear
// any stale publish_blocked gate so the imported agent publishes
// immediately rather than waiting for the next boot decision pass.
retain_user_intent_event(
&conn,
&RetainedEvent::pending(
KIND_MANAGED_AGENT,

View File

@@ -851,7 +851,9 @@ fn retain_agent_pending(app: &AppHandle, state: &AppState, record: &ManagedAgent
use crate::managed_agents::{
agent_events::{agent_event_content, build_agent_event},
persona_events::monotonic_created_at,
retention::{get_retained_event, open_retention_db, retain_event, RetainedEvent},
retention::{
get_retained_event, open_retention_db, retain_user_intent_event, RetainedEvent,
},
};
use buzz_core_pkg::kind::KIND_MANAGED_AGENT;
@@ -874,7 +876,10 @@ fn retain_agent_pending(app: &AppHandle, state: &AppState, record: &ManagedAgent
.map_err(|e| format!("failed to sign agent event: {e}"))?;
(owner_pubkey, event)
};
retain_event(
// User-intent: a team snapshot import is an explicit user action —
// clear any stale publish_blocked gate so the imported agent publishes
// immediately rather than waiting for the next boot decision pass.
retain_user_intent_event(
&conn,
&RetainedEvent::pending(
KIND_MANAGED_AGENT,

View File

@@ -40,7 +40,9 @@ fn trim_optional(value: Option<String>) -> Option<String> {
pub(super) fn retain_team_pending(app: &AppHandle, state: &AppState, team: &TeamRecord) {
use crate::managed_agents::{
persona_events::monotonic_created_at,
retention::{get_retained_event, open_retention_db, retain_event, RetainedEvent},
retention::{
get_retained_event, open_retention_db, retain_user_intent_event, RetainedEvent,
},
team_events::build_team_event,
};
use buzz_core_pkg::kind::KIND_TEAM;
@@ -56,7 +58,9 @@ pub(super) fn retain_team_pending(app: &AppHandle, state: &AppState, team: &Team
.custom_created_at(monotonic_created_at(prior))
.sign_with_keys(&scope.owner_keys)
.map_err(|e| format!("failed to sign team event: {e}"))?;
retain_event(
// User-intent: clears any stale publish_blocked so an edit to a parked
// team coordinate takes effect in-session.
retain_user_intent_event(
&conn,
&RetainedEvent::pending(KIND_TEAM, pubkey, team.id.clone(), &event),
)
@@ -78,8 +82,8 @@ pub(super) fn retain_team_pending(app: &AppHandle, state: &AppState, team: &Team
fn tombstone_team_pending(app: &AppHandle, state: &AppState, d_tag: &str) {
use crate::managed_agents::{
retention::{
delete_retained_event, open_retention_db, retain_event, tombstone_retention_d_tag,
RetainedEvent,
delete_retained_event, open_retention_db, retain_user_intent_event,
tombstone_retention_d_tag, RetainedEvent,
},
team_events::build_team_delete,
};
@@ -95,7 +99,7 @@ fn tombstone_team_pending(app: &AppHandle, state: &AppState, d_tag: &str) {
.map_err(|e| format!("failed to sign team tombstone: {e}"))?;
let conn = open_retention_db(&scope.db_path)?;
delete_retained_event(&conn, KIND_TEAM, &pubkey, d_tag)?;
retain_event(
retain_user_intent_event(
&conn,
&RetainedEvent::pending(
KIND_DELETE,

View File

@@ -140,12 +140,15 @@ async fn run_boot_barrier_enforcing(
/// retained because it is cheap and keeps the invariant obvious.
///
/// 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
/// ones: a legacy row queued before the baseline table 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.
///
/// Baseline provenance lives in `persona_baselines` (keyed by coordinate),
/// not in the `persona_events` live row, so the check joins there.
///
/// Returns how many rows it withheld.
fn quarantine_unprovable_pending(
conn: &rusqlite::Connection,
@@ -153,7 +156,13 @@ fn quarantine_unprovable_pending(
) -> 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",
WHERE pending_sync = 1 AND pubkey = ?1
AND NOT EXISTS (
SELECT 1 FROM persona_baselines
WHERE persona_baselines.kind = persona_events.kind
AND persona_baselines.pubkey = persona_events.pubkey
AND persona_baselines.d_tag = persona_events.d_tag
)",
rusqlite::params![owner_pubkey],
)
.map_err(|error| format!("failed to quarantine unprovable pending rows: {error}"))

View File

@@ -1,4 +1,5 @@
//! Regression tests for am#1, peon-P1#1/P1#2, am#2/peon-P2#1.
//! Regression tests for am#1, peon-P1#1/P1#2, am#2/peon-P2#1,
//! and Thufir pass-1 findings.
//!
//! Kept separate from `tests.rs` to respect the 1000-line ceiling on new files.
@@ -8,13 +9,15 @@ use super::*;
use crate::managed_agents::{
decision::{Decision, HeadState, TombstoneEvidence},
persona_events::build_persona_event,
reconcile::retain_agent_record,
retention::{
delete_retained_event, get_baseline, get_retained_event, is_publish_blocked,
mark_synced_and_stamp_baseline, open_retention_db, retain_event, retain_user_intent_event,
set_baseline, set_publish_blocked, tombstone_retention_d_tag, RetainedEvent,
},
ManagedAgentRecord,
};
use buzz_core_pkg::kind::{KIND_DELETION, KIND_PERSONA};
use buzz_core_pkg::kind::{KIND_DELETION, KIND_MANAGED_AGENT, KIND_PERSONA};
use rusqlite::Connection;
const OWNER: &str = "ownerpubkeyhex";
@@ -387,3 +390,217 @@ fn test_user_edit_reopens_gated_row_but_reconcile_retain_does_not() {
"reconcile retain must NOT clear publish_blocked"
);
}
// ── (c-ext) Thufir pass-1 finding #2: retain_agent_record user-intent clears
// the gate; boot-reconcile (user_intent=false) does NOT.
//
// Mutation-sensitive: passing `false` (reconcile mode) to both calls would
// leave the gate set after the first assert succeeds (reconcile doesn't clear
// it) and would cause the second assert to pass trivially — but the assert on
// the gate being CLEAR after user_intent=true would fail because nothing
// cleared it, catching the regression. Conversely, passing `true` to both
// would make the reconcile call incorrectly clear the gate, failing the final
// "reconcile must NOT clear" assertion.
#[test]
fn test_retain_agent_record_user_intent_clears_gate_but_reconcile_does_not() {
let conn = test_db();
let keys = nostr::Keys::generate();
let owner = keys.public_key().to_hex();
let record = make_sample_agent_record();
// Seed the row as blocked (simulates what the boot barrier would have set).
retain_agent_record(&conn, &keys, &record, false).unwrap();
set_publish_blocked(&conn, KIND_MANAGED_AGENT, &owner, &record.pubkey, true).unwrap();
assert!(
is_publish_blocked(&conn, KIND_MANAGED_AGENT, &owner, &record.pubkey).unwrap(),
"precondition: gate must be set"
);
// Interactive save (user_intent = true) must reopen the gate.
retain_agent_record(&conn, &keys, &record, true).unwrap();
assert!(
!is_publish_blocked(&conn, KIND_MANAGED_AGENT, &owner, &record.pubkey).unwrap(),
"user-intent retain_agent_record must clear publish_blocked"
);
// Re-gate to simulate a fresh boot decision.
set_publish_blocked(&conn, KIND_MANAGED_AGENT, &owner, &record.pubkey, true).unwrap();
// Boot reconcile (user_intent = false) must NOT clear it.
retain_agent_record(&conn, &keys, &record, false).unwrap();
assert!(
is_publish_blocked(&conn, KIND_MANAGED_AGENT, &owner, &record.pubkey).unwrap(),
"reconcile retain_agent_record must NOT clear publish_blocked"
);
}
// ── (c-ext2) Thufir pass-1 finding #2: non-persona tombstone (kind:5 for
// a managed-agent coordinate) via retain_user_intent_event clears the gate;
// bare retain_event does NOT.
//
// Mutation-sensitive: the production paths in `tombstone_managed_agent_pending`
// and `tombstone_team_pending` now call `retain_user_intent_event`; if they
// were accidentally reverted to `retain_event`, the first assertion would fail.
#[test]
fn test_non_persona_tombstone_user_intent_clears_gate() {
use nostr::JsonUtil;
let conn = test_db();
const KIND_DELETE: u32 = 5;
let tombstone_d = tombstone_retention_d_tag(KIND_MANAGED_AGENT, "agent-abc123");
let keys = nostr::Keys::generate();
let owner = keys.public_key().to_hex();
let event = nostr::EventBuilder::new(nostr::Kind::Custom(KIND_DELETE as u16), "{}".to_string())
.sign_with_keys(&keys)
.unwrap();
// Gate a placeholder row at the tombstone coordinate.
retain_event(
&conn,
&RetainedEvent {
kind: KIND_DELETE,
pubkey: owner.clone(),
d_tag: tombstone_d.clone(),
content: "{}".to_string(),
created_at: event.created_at.as_secs() as i64,
raw_event: event.as_json(),
event_id: Some(event.id.to_hex()),
pending_sync: false,
publish_blocked: false,
},
)
.unwrap();
set_publish_blocked(&conn, KIND_DELETE, &owner, &tombstone_d, true).unwrap();
assert!(
is_publish_blocked(&conn, KIND_DELETE, &owner, &tombstone_d).unwrap(),
"precondition: gate must be set"
);
// User-intent tombstone retain must clear the gate.
let event2 =
nostr::EventBuilder::new(nostr::Kind::Custom(KIND_DELETE as u16), "{}".to_string())
.sign_with_keys(&keys)
.unwrap();
retain_user_intent_event(
&conn,
&RetainedEvent {
kind: KIND_DELETE,
pubkey: owner.clone(),
d_tag: tombstone_d.clone(),
content: "{}".to_string(),
created_at: event2.created_at.as_secs() as i64 + 1,
raw_event: event2.as_json(),
event_id: Some(event2.id.to_hex()),
pending_sync: true,
publish_blocked: false,
},
)
.unwrap();
assert!(
!is_publish_blocked(&conn, KIND_DELETE, &owner, &tombstone_d).unwrap(),
"user-intent tombstone retain must clear publish_blocked"
);
}
// ── (f) Thufir pass-1 finding: mark_synced_and_stamp_baseline atomically
// clears pending AND stamps the baseline in a single transaction.
//
// Mutation-sensitive: if the function is split back into two independent
// autocommit statements, the test still passes on the happy path — but the
// invariant that "pending_sync = 0 iff baseline is stamped" becomes testable
// by verifying both columns together. If only one statement ran, one of the
// two final assertions would fail, catching the regression.
#[test]
fn test_mark_synced_and_stamp_baseline_atomic_both_written_on_success() {
let conn = test_db();
const D_TAG: &str = "atomic-test";
let event = signed_persona("atomic-prompt");
let event_id = event.id.to_hex();
let content = event.content.to_string();
let created_at = event.created_at.as_secs() as i64;
retain_event(
&conn,
&RetainedEvent {
kind: KIND_PERSONA,
pubkey: OWNER.to_string(),
d_tag: D_TAG.to_string(),
content: content.clone(),
created_at,
raw_event: {
use nostr::JsonUtil;
event.as_json()
},
event_id: Some(event_id.clone()),
pending_sync: true,
publish_blocked: false,
},
)
.unwrap();
// No baseline before the call.
assert!(
get_baseline(&conn, KIND_PERSONA, OWNER, D_TAG)
.unwrap()
.is_none(),
"no baseline before mark_synced_and_stamp_baseline"
);
mark_synced_and_stamp_baseline(
&conn,
KIND_PERSONA,
OWNER,
D_TAG,
created_at,
&content,
&event_id,
false,
)
.unwrap();
// Both effects must have landed atomically.
let row = get_retained_event(&conn, KIND_PERSONA, OWNER, D_TAG)
.unwrap()
.expect("row must still exist");
assert!(
!row.pending_sync,
"pending_sync must be cleared by mark_synced_and_stamp_baseline"
);
let baseline = get_baseline(&conn, KIND_PERSONA, OWNER, D_TAG)
.unwrap()
.expect("baseline must be stamped by mark_synced_and_stamp_baseline");
assert_eq!(
baseline.0, event_id,
"baseline event_id must match the published event"
);
assert_eq!(
baseline.1, content,
"baseline content must match the published content"
);
}
/// Helper shared by the agent-record tests in this module.
pub(super) fn make_sample_agent_record() -> ManagedAgentRecord {
serde_json::from_str(
r#"{
"pubkey": "abababababababababababababababababababababababababababababababababab",
"name": "regression-test-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 regression-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()
}

View File

@@ -23,7 +23,10 @@ use std::path::Path;
use super::{
agent_events::build_agent_event,
persona_events::monotonic_created_at,
retention::{get_retained_event, open_retention_db, retain_event, RetainedEvent},
retention::{
get_retained_event, open_retention_db, retain_event, retain_user_intent_event,
RetainedEvent,
},
ManagedAgentRecord,
};
use buzz_core_pkg::kind::KIND_MANAGED_AGENT;
@@ -75,7 +78,7 @@ pub(crate) fn reconcile_agents_in_dir_at(
continue;
}
if retain_agent_record(&conn, keys, record)? {
if retain_agent_record(&conn, keys, record, false)? {
reconciled += 1;
}
}
@@ -93,10 +96,17 @@ pub(crate) fn reconcile_agents_in_dir_at(
/// (`retain_managed_agent_pending`, persona-rename propagation). Every
/// mutation of an agent's published identity must go through it so the
/// retained record can never silently drift from `managed-agents.json`.
///
/// `user_intent` must be `true` when the caller is an interactive user action
/// (create, update, delete tombstone). A `true` value unconditionally clears
/// `publish_blocked` on the retained row so that a previously-parked coordinate
/// is reopened in-session. Boot-time reconcile callers pass `false` to leave
/// the gate unchanged — the laundering path is exactly reconcile.
pub(crate) fn retain_agent_record(
conn: &rusqlite::Connection,
keys: &nostr::Keys,
record: &ManagedAgentRecord,
user_intent: bool,
) -> Result<bool, String> {
let owner_pubkey = keys.public_key().to_hex();
let existing = get_retained_event(conn, KIND_MANAGED_AGENT, &owner_pubkey, &record.pubkey)?;
@@ -117,18 +127,34 @@ pub(crate) fn retain_agent_record(
let content = event.content.clone();
if existing.as_ref().is_some_and(|row| row.content == content) {
// Content is unchanged — no publish churn. If this is a user-intent
// call, still clear any stale gate: the user re-confirmed the same
// content intentionally, which is a gate-clearing action regardless.
if user_intent {
use crate::managed_agents::retention::set_publish_blocked;
set_publish_blocked(
conn,
KIND_MANAGED_AGENT,
&owner_pubkey,
&record.pubkey,
false,
)
.map_err(|e| format!("failed to clear gate for '{}': {e}", record.name))?;
}
return Ok(false);
}
retain_event(
conn,
&RetainedEvent::pending(
KIND_MANAGED_AGENT,
owner_pubkey,
record.pubkey.clone(),
&event,
),
)
let retained = RetainedEvent::pending(
KIND_MANAGED_AGENT,
owner_pubkey,
record.pubkey.clone(),
&event,
);
if user_intent {
retain_user_intent_event(conn, &retained)
} else {
retain_event(conn, &retained)
}
.map_err(|e| format!("failed to retain '{}': {e}", record.name))?;
Ok(true)
}

View File

@@ -324,7 +324,7 @@ fn rename_re_retains_identity_record_with_new_name() {
let pubkey = "9".repeat(64);
let mut record = sample_record(&pubkey, "Fizz");
assert!(retain_agent_record(&conn, &keys, &record).unwrap());
assert!(retain_agent_record(&conn, &keys, &record, false).unwrap());
let first = get_retained_event(&conn, KIND_MANAGED_AGENT, &owner, &pubkey)
.unwrap()
.unwrap();
@@ -341,7 +341,7 @@ fn rename_re_retains_identity_record_with_new_name() {
record.name = "Spark".to_string();
assert!(
retain_agent_record(&conn, &keys, &record).unwrap(),
retain_agent_record(&conn, &keys, &record, false).unwrap(),
"a renamed record must re-retain its identity record"
);
@@ -374,7 +374,7 @@ fn retain_agent_record_is_noop_when_unchanged() {
let pubkey = "8".repeat(64);
let record = sample_record(&pubkey, "steady-agent");
assert!(retain_agent_record(&conn, &keys, &record).unwrap());
assert!(retain_agent_record(&conn, &keys, &record, false).unwrap());
let row = get_retained_event(
&conn,
KIND_MANAGED_AGENT,
@@ -394,7 +394,7 @@ fn retain_agent_record_is_noop_when_unchanged() {
.unwrap();
assert!(
!retain_agent_record(&conn, &keys, &record).unwrap(),
!retain_agent_record(&conn, &keys, &record, false).unwrap(),
"an unchanged projection must not re-retain"
);
assert!(

View File

@@ -612,11 +612,21 @@ fn parse_tombstone_retention_d_tag(key: &str) -> Option<(u32, String)> {
/// Clear the `pending_sync` flag for an event the relay just confirmed,
/// and atomically stamp the accepted event as this coordinate's baseline.
///
/// Compare-and-clear: only clears the row if its `created_at` and `content`
/// still match what was published. A concurrent edit that upserted a newer
/// version at the same coordinate between the flush loop's read and this call
/// leaves `pending_sync` set, so the newer edit publishes on the next pass
/// instead of being silently dropped.
/// Both writes execute inside a single SQLite transaction so a crash or error
/// between them cannot produce a state where `pending_sync` was cleared but
/// the baseline was not stamped (the hole that allows the one-edit-later revert
/// vector to resurface). The transaction is committed only when both statements
/// succeed; on any error the transaction rolls back automatically via
/// `unchecked_transaction`'s RAII Drop.
///
/// Compare-and-clear: the `mark_synced` statement only updates the row when
/// its `created_at` and `content` still match what was published. A concurrent
/// edit that upserted a newer version at the same coordinate between the flush
/// loop's read and this call leaves `pending_sync` set, so the newer edit
/// publishes on the next pass instead of being silently dropped. The baseline
/// is only stamped when `mark_synced` affected the intended row (rows_changed
/// == 1), preventing a spurious baseline write when the concurrent-edit guard
/// already refused the sync.
///
/// Stamping the baseline here is the fix for the one-edit-later revert vector:
/// after device A publishes v2 and this function runs, the baseline says v2, so
@@ -633,8 +643,33 @@ pub fn mark_synced_and_stamp_baseline(
event_id: &str,
shared: bool,
) -> Result<(), String> {
mark_synced(conn, kind, pubkey, d_tag, created_at, content)?;
set_baseline_full(conn, kind, pubkey, d_tag, event_id, content, shared)
let tx = conn
.unchecked_transaction()
.map_err(|e| format!("failed to begin mark-synced-and-stamp-baseline transaction: {e}"))?;
let rows_changed = tx
.execute(
"UPDATE persona_events SET pending_sync = 0
WHERE kind = ?1 AND pubkey = ?2 AND d_tag = ?3
AND created_at = ?4 AND content = ?5",
params![kind, pubkey, d_tag, created_at, content],
)
.map_err(|e| format!("failed to mark event synced: {e}"))?;
if rows_changed == 1 {
// The compare-and-clear matched — stamp the baseline in the same tx.
tx.execute(
"INSERT INTO persona_baselines
(kind, pubkey, d_tag, baseline_event_id, baseline_content, baseline_shared)
VALUES (?1, ?2, ?3, ?4, ?5, ?6)
ON CONFLICT (kind, pubkey, d_tag) DO UPDATE SET
baseline_event_id = excluded.baseline_event_id,
baseline_content = excluded.baseline_content,
baseline_shared = excluded.baseline_shared",
params![kind, pubkey, d_tag, event_id, content, i32::from(shared)],
)
.map_err(|e| format!("failed to stamp baseline: {e}"))?;
}
tx.commit()
.map_err(|e| format!("failed to commit mark-synced-and-stamp-baseline: {e}"))
}
/// Clear the `pending_sync` flag for an event the relay just confirmed.
@@ -644,6 +679,11 @@ pub fn mark_synced_and_stamp_baseline(
/// version at the same coordinate between the flush loop's read and this call
/// leaves `pending_sync` set, so the newer edit publishes on the next pass
/// instead of being silently dropped.
///
/// Production publication uses `mark_synced_and_stamp_baseline` which wraps
/// this logic inside a transaction; this function remains available for tests
/// that need the mark-only behaviour.
#[allow(dead_code)]
pub fn mark_synced(
conn: &Connection,
kind: u32,