From 5e74b42f8bcd79edba04b0863c684fa47dfee3b5 Mon Sep 17 00:00:00 2001 From: cyberzero000 Date: Sun, 23 Aug 2026 17:46:37 -0700 Subject: [PATCH] fix(cli): stop set-add-policy erasing the agent profile MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `cmd_set_add_policy` published a kind:10100 whose content was only `{"channel_add_policy": ...}`. The kind is replaceable, and the relay projects just that one field into a column (`crates/buzz-relay/src/handlers/side_effects.rs`) — `channel_ids`, `name`, `display_name` and `respond_to` live solely in the event body clients read. One policy change wiped the rest of the profile, which took the agent out of every channel's @mention picker until an operator republished by hand. Read the current profile, merge the field, republish. Three things the merge has to get right, each covered by a test: - A result this command cannot fully read is an error rather than an empty profile. A body that is not a JSON object is refused, and so is a result row missing a readable `content` or `created_at`. Treating either as absent would republish a single-field profile and cause the exact wipe this fixes. - The write out-bids the stored copy's `created_at` instead of tying it. `Db::replace_addressable_event` breaks a same-second tie by lowest event id, so publishing at `now` is a coin flip against a peer republishing the profile concurrently, and a deterministic loss against a copy stamped later by a skewed peer. - The lead carries headroom under the relay's 900s tolerance. That window is measured against the relay's clock and this budget against ours, so a copy accepted at `S` when relay time was `R` proves only `S - R <= 900`, and out-bidding it at `S + 1` can sit up to 901s from relay time. Read-modify-write on a replaceable event has no compare-and-set, so this re-reads after publishing. Comparing the stored head's id to ours is not enough on its own: kind:10100 is global, so the filter carries neither a channel pin nor an `until`, which makes it `RoutePredicate::Bounded` and lets the relay serve it from a read replica whenever `BUZZ_REPLICA_READ_MAX_AGE_MS` is set. That budget bounds how recently the replica proved its replay position, not whether a write from a moment ago is visible, so a read issued straight after the publish can return the pre-write event. The relay's own replace rule separates that lag from a real loss: a stored copy that loses the `created_at`/lowest-id comparison to the event we just published cannot have replaced it. So the read-back is classified three ways: - Landed — our event is the stored head. Success. - Replaced — a copy that beats ours is stored. Re-merge onto it and publish again, and return exit 5 if every attempt is replaced. - Unconfirmed — the read cannot see our write, or it failed. The relay accepted the event and nothing that could have replaced it is stored, so this is a success we could not confirm: exit 0 with the documented `{event_id, accepted, message}` plus an additive `warning`. Reporting a conflict here would tell a caller to retry a change that had already landed, and the retry would rebuild the same body at the same stamp, come back `duplicate:`, and look like a conflict again. A confirmation read that fails outright is Unconfirmed rather than the command's error: the mutation has already happened by then, so borrowing the read's error reported a stored policy change as a network failure. Profile-read failures also keep their `CliError` variant instead of flattening to `Other`. `exit_code` and `is_retryable_error` classify by variant, so flattening reported a retryable 503 as exit 4 with `retryable: false`, and an expired `BUZZ_AUTH_TAG` as exit 4 instead of 3 — telling an agent to abandon both as permanent. The window is narrowed, not closed: a peer publishing between our read and our write still loses its change, because a replaceable write carries the whole body and nothing records what we replaced. Signed-off-by: cyberzero000 --- crates/buzz-cli/src/commands/channels.rs | 570 ++++++++++++++++++++++- 1 file changed, 557 insertions(+), 13 deletions(-) diff --git a/crates/buzz-cli/src/commands/channels.rs b/crates/buzz-cli/src/commands/channels.rs index 7ad051ef9fc..f171b807b76 100644 --- a/crates/buzz-cli/src/commands/channels.rs +++ b/crates/buzz-cli/src/commands/channels.rs @@ -10,6 +10,7 @@ use crate::client::{ }; use crate::commands::agents::fetch_archived_snapshot; use crate::commands::channel_templates::{self, ChannelTemplateRecord, TemplateAgentRoster}; +use crate::commands::parse_write_response; use crate::error::CliError; use crate::validate::{parse_uuid, read_or_stdin, validate_hex64, validate_uuid}; @@ -1018,6 +1019,165 @@ pub async fn cmd_remove_channel_member( Ok(()) } +/// The kind:10100 profile the relay currently stores for us. +#[derive(Debug)] +struct StoredAgentProfile { + /// Parsed event body, or empty when no profile is stored. + body: serde_json::Map, + /// `created_at` of the stored event, or 0 when none is stored. A replaceable + /// write must out-bid this to land. + created_at: u64, +} + +/// Fetch this identity's current kind:10100 profile body as a JSON object. +/// +/// Returns an empty object only when the identity genuinely has no profile +/// yet. Every other outcome is an error: the caller writes a replaceable event +/// built from this map, so treating a failed lookup as "no profile" would +/// republish a single-field profile and erase `name`, `display_name`, +/// `channel_ids` and `respond_to` — the exact wipe this merge exists to +/// prevent. Failing loudly leaves the stored profile intact. +async fn fetch_own_agent_profile(client: &BuzzClient) -> Result { + let filter = serde_json::json!({ + "kinds": [buzz_sdk::kind::KIND_AGENT_PROFILE], + "authors": [client.keys().public_key().to_hex()], + "limit": 1, + }); + let raw = client.query(&filter).await.map_err(profile_read_error)?; + let events: Vec = serde_json::from_str(&raw).map_err(|e| { + CliError::Other(format!( + "relay returned an unreadable profile query result: {e}" + )) + })?; + parse_stored_profile(&events) +} + +/// Explain a failed profile read without flattening its category. +/// +/// `exit_code` and `is_retryable_error` classify by variant, and those codes are +/// the CLI's agent-facing contract (2 = network/relay, 3 = auth, 4 = other), so +/// mapping a 503 or a connect timeout to `Other` would tell a caller to abandon +/// a transient outage as permanent, and an expired `BUZZ_AUTH_TAG` to treat +/// re-authenticating as pointless. Context is added where the variant has room +/// for it. +fn profile_read_error(e: CliError) -> CliError { + const CONTEXT: &str = "could not read the current agent profile, so the policy change \ + was not published (publishing without it would erase the profile)"; + match e { + // `reqwest::Error` has nowhere to put added context, and the variant — + // exit 2, retryable — is what a caller acts on. + CliError::Network(_) => e, + CliError::Relay { status, body } => CliError::Relay { + status, + body: format!("{CONTEXT}: {body}"), + }, + CliError::Auth(message) => CliError::Auth(format!("{CONTEXT}: {message}")), + CliError::Key(message) => CliError::Key(format!("{CONTEXT}: {message}")), + other => CliError::Other(format!("{CONTEXT}: {other}")), + } +} + +/// The stored profile a query result describes, or an error rather than a guess. +/// +/// Separate from the query so the decision that matters — when it is safe to +/// treat a result as "no profile" — is testable without a relay. Getting it +/// wrong republishes a single-field profile and erases the rest. +fn parse_stored_profile(events: &[serde_json::Value]) -> Result { + let Some(event) = events.first() else { + return Ok(StoredAgentProfile { + body: serde_json::Map::new(), + created_at: 0, + }); + }; + // A result row we cannot fully read is not an absent profile, and both + // fields are checked together because both defaults were unsafe: + // defaulting `content` to `""` takes the empty-body path below and + // republishes a single-field profile — the wipe this merge exists to + // prevent — and defaulting `created_at` to 0 stamps the write at a plain + // `now`, restoring the same-second event-id coin flip + // `profile_publish_timestamp` exists to avoid. Only a genuinely absent + // event yields an empty body. + let (Some(created_at), Some(content)) = ( + event.get("created_at").and_then(serde_json::Value::as_u64), + event.get("content").and_then(serde_json::Value::as_str), + ) else { + return Err(CliError::Other( + "the relay returned an agent profile event without a readable \ + `created_at` and `content`; refusing to replace the stored profile \ + from a result this command cannot interpret." + .to_string(), + )); + }; + if content.trim().is_empty() { + // A profile event with no body carries nothing to preserve, so this is + // the "no profile" case rather than a lookup we failed to read. + return Ok(StoredAgentProfile { + body: serde_json::Map::new(), + created_at, + }); + } + match serde_json::from_str::(content) { + Ok(serde_json::Value::Object(body)) => Ok(StoredAgentProfile { body, created_at }), + _ => Err(CliError::Other( + "the stored agent profile is not a JSON object; refusing to replace it \ + with a single-field profile. This command only merges into an object \ + body and is the only in-repo publisher of kind:10100, so recover by \ + signing a corrected kind:10100 event and submitting it to the relay's \ + `POST /events`, then re-run." + .to_string(), + )), + } +} + +/// How far ahead of this host's clock a merge may stamp its `created_at`. +/// +/// The relay accepts a ±900s window measured against *its own* clock +/// (`MAX_TIMESTAMP_DRIFT_SECS`, `crates/buzz-relay/src/handlers/ingest.rs`); +/// this budget is measured against ours, so the two are not directly +/// comparable. A stored copy accepted at `S` when relay time was `R` proves +/// only `S - R <= 900`, so out-bidding it at `S + 1` can sit up to 901s from +/// relay time. The headroom below 900 absorbs that overshoot, so a refusal here +/// means the stored profile's timestamp is genuinely broken rather than merely +/// written by a host whose clock leads ours. +const MAX_PROFILE_PUBLISH_LEAD_SECS: u64 = 600; + +fn unix_secs_now() -> u64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_secs()) + .unwrap_or(0) +} + +/// The `created_at` a merge must publish to replace `stored_created_at`. +/// +/// Out-bids the stored copy by timestamp rather than leaving it to the event-id +/// tiebreak. `Db::replace_addressable_event` resolves a same-second write by +/// lowest id, so a plain `now` write is a coin flip against a peer republishing +/// the profile in the same second — and a deterministic *loss* against a stored +/// copy stamped later by a skewed peer. Retrying at `now` cannot win either, so +/// retries alone do not fix it. +/// +/// Bounded against the *relay's* tolerance, not against a tight local-clock +/// assumption. Stamping `stored_created_at + 1` is only ever one second ahead of +/// whichever writer produced the stored copy, so the lead measured against our +/// own clock is just this host's skew from that writer's — and refusing on a few +/// seconds of it made the command unusable on any deployment where the harness +/// host's clock runs ahead of the operator's. What needs guarding is a stored +/// copy so far in the future that out-bidding it would fall outside the relay's +/// ±900s window. +fn profile_publish_timestamp(now: u64, stored_created_at: u64) -> Result { + let created_at = now.max(stored_created_at + 1); + let lead = created_at.saturating_sub(now); + if lead > MAX_PROFILE_PUBLISH_LEAD_SECS { + return Err(CliError::Other(format!( + "the stored agent profile is timestamped {lead}s ahead of this host's clock; \ + out-bidding it would fall outside the relay's timestamp tolerance. \ + Check for clock skew, or republish a correctly stamped kind:10100 profile." + ))); + } + Ok(created_at) +} + /// Set the channel addition policy — sign and submit a kind:10100 (agent profile) event. pub async fn cmd_set_add_policy(client: &BuzzClient, policy: &str) -> Result<(), CliError> { match policy { @@ -1049,18 +1209,219 @@ pub async fn cmd_set_add_policy(client: &BuzzClient, policy: &str) -> Result<(), } } - let content = serde_json::json!({ "channel_add_policy": policy }).to_string(); + // kind:10100 is replaceable, and the relay projects only + // `channel_add_policy` into a column — `channel_ids`, `name`, + // `display_name` and `respond_to` live solely in the event body clients + // read. Publishing this one field alone therefore erased the rest of the + // profile, taking the agent out of every channel's @mention picker. + // + // This is a read-modify-write on a replaceable event with no + // compare-and-set, so it races any other client publishing kind:10100 for + // this identity under the same key. No in-repo writer other than this + // command publishes that kind, so the peer is an out-of-repo one — an agent + // harness that republishes the profile as its channel membership changes. + // Re-read after publishing and redo the merge if someone else's copy is now + // stored: without that, an operator running this command while an agent is + // being invited to a channel silently drops the invite from `channel_ids`. + // + // The window is narrowed, not closed. A peer that publishes between our + // read and our write still loses its change, because a replaceable write + // carries the whole body and nothing records what we replaced. + const MAX_MERGE_ATTEMPTS: usize = 3; + const CONFLICT_MSG: &str = "another writer replaced the agent profile while this policy \ + change was being published; the change did not stick. Retry."; use nostr::{EventBuilder, Kind}; - let builder = EventBuilder::new( - Kind::Custom(buzz_sdk::kind::KIND_AGENT_PROFILE as u16), - &content, - ) - .tags([]); - let event = client.sign_event(builder)?; + // No sleep between attempts: unlike a harness that republishes repeatedly + // and would ratchet its own stamps upward, this runs at most three times + // per invocation. Sleeping here only delayed the command. + for _ in 1..=MAX_MERGE_ATTEMPTS { + let stored = fetch_own_agent_profile(client).await?; + let mut profile = stored.body; + profile.insert( + "channel_add_policy".to_string(), + serde_json::Value::String(policy.to_string()), + ); + let content = serde_json::Value::Object(profile).to_string(); + + // Out-bid the stored copy by timestamp rather than leaving it to the + // event-id tiebreak. `Db::replace_addressable_event` resolves a + // same-second write by lowest id, so a default `created_at = now` write + // is a coin flip against a peer republishing the profile in the same + // second — and a deterministic *loss* against a stored copy stamped + // later by a skewed peer. Retrying at `now` cannot win either, so retries alone do + // not fix it. + let created_at = profile_publish_timestamp(unix_secs_now(), stored.created_at)?; + + let builder = EventBuilder::new( + Kind::Custom(buzz_sdk::kind::KIND_AGENT_PROFILE as u16), + &content, + ) + .tags([]) + .custom_created_at(nostr::Timestamp::from_secs(created_at)); + let event = client.sign_event(builder)?; + let event_id = event.id.to_hex(); + + let resp = client.submit_event(event).await?; + + // `duplicate:` means the relay took the write and rolled it back — the + // loser of a replaceable compare. It arrives as `accepted: true`, so + // reading only `accepted` would report success for a discarded write. + match parse_write_response(&resp, CONFLICT_MSG) { + // Accepted is still not landed: a peer may have published straight + // after us and replaced our copy. + Ok(rendered) => match confirm_stored_profile(client, &event_id, created_at).await { + Confirmation::Landed => { + println!("{rendered}"); + return Ok(()); + } + Confirmation::Replaced => {} + // The relay accepted the event and nothing that could have + // replaced it is stored, so this is a write we could not + // confirm rather than one that failed. Reporting a conflict + // here told a caller to retry a change that had already + // landed — and the retry rebuilt the same body at the same + // stamp, so it came back `duplicate:` and looked like a + // conflict again, all the way to exit 5. + Confirmation::Unconfirmed(reason) => { + println!("{}", unconfirmed_write_response(&rendered, &reason)); + return Ok(()); + } + }, + Err(CliError::Conflict(_)) => {} + Err(other) => return Err(other), + } + } + // Never report success for a write we could not confirm landed. Exit 5 is + // the CLI's write-conflict code, so a caller can distinguish "retry me" + // from a usage or network error. + Err(CliError::Conflict(CONFLICT_MSG.to_string())) +} - let resp = client.submit_event(event).await?; - println!("{}", normalize_write_response(&resp)); - Ok(()) +/// What a post-publish read proves about the event we just published. +#[derive(Debug, PartialEq, Eq)] +enum Confirmation { + /// Our event is the stored copy. + Landed, + /// A copy that beats ours is stored, so ours was replaced: merge onto the + /// new one and publish again. + Replaced, + /// The read proves nothing — it cannot see our write yet, or it failed. + Unconfirmed(String), +} + +/// Read the stored kind:10100 head back and classify our write against it. +/// +/// A failed read is `Unconfirmed`, never an error: the mutation has already +/// happened by the time this runs, so borrowing the read's error would report a +/// successful policy change as a network or relay failure and send exit 2 to a +/// caller whose write landed. +async fn confirm_stored_profile( + client: &BuzzClient, + published_id: &str, + published_created_at: u64, +) -> Confirmation { + match stored_profile_head(client).await { + Ok(stored) => classify_confirmation( + published_id, + published_created_at, + stored + .as_ref() + .map(|(id, created_at)| (id.as_str(), *created_at)), + ), + Err(e) => Confirmation::Unconfirmed(format!( + "the relay accepted the event, but reading the profile back to confirm \ + it did not succeed: {e}" + )), + } +} + +/// Classify a read-back head against the event we published. +/// +/// The read cannot be pinned to the primary: kind:10100 is global, so the +/// filter carries neither a channel pin nor an `until`, which makes it +/// `RoutePredicate::Bounded` and lets the relay serve it from a read replica +/// whenever `BUZZ_REPLICA_READ_MAX_AGE_MS` is set. That budget bounds how +/// recently the replica proved its replay position, not whether a write from a +/// moment ago is visible, so a read issued straight after the publish can +/// legitimately return the pre-write event. +/// +/// The relay's own replace rule separates that lag from a real loss: +/// `Db::replace_addressable_event` keeps the higher `created_at` and breaks a +/// tie by lowest event id. A stored copy that loses that comparison to the +/// event we just published therefore cannot have replaced it — the read is +/// simply behind. Treating it as a conflict republished a byte-identical event, +/// drew `duplicate:`, and returned exit 5 for a policy change that was stored. +fn classify_confirmation( + published_id: &str, + published_created_at: u64, + stored: Option<(&str, u64)>, +) -> Confirmation { + let Some((stored_id, stored_created_at)) = stored else { + // The relay accepted our event, so an empty read is one that cannot see + // it yet: a replaceable write is never removed by a peer's. + return Confirmation::Unconfirmed( + "the relay accepted the event, but the profile read back empty".to_string(), + ); + }; + if stored_id == published_id { + return Confirmation::Landed; + } + let stored_beats_ours = match stored_created_at.cmp(&published_created_at) { + std::cmp::Ordering::Greater => true, + std::cmp::Ordering::Equal => stored_id < published_id, + std::cmp::Ordering::Less => false, + }; + if stored_beats_ours { + Confirmation::Replaced + } else { + Confirmation::Unconfirmed(format!( + "the relay accepted the event, but the profile read back a copy \ + ({stored_id} at {stored_created_at}) that cannot have replaced it" + )) + } +} + +/// Event id and `created_at` of the kind:10100 profile the relay currently +/// stores for us. +async fn stored_profile_head(client: &BuzzClient) -> Result, CliError> { + let filter = serde_json::json!({ + "kinds": [buzz_sdk::kind::KIND_AGENT_PROFILE], + "authors": [client.keys().public_key().to_hex()], + "limit": 1, + }); + let raw = client.query(&filter).await?; + let events: Vec = serde_json::from_str(&raw) + .map_err(|e| CliError::Other(format!("unreadable profile query result: {e}")))?; + let Some(event) = events.first() else { + return Ok(None); + }; + let (Some(id), Some(created_at)) = ( + event.get("id").and_then(serde_json::Value::as_str), + event.get("created_at").and_then(serde_json::Value::as_u64), + ) else { + return Err(CliError::Other( + "the relay returned a profile event without a readable id and `created_at`".to_string(), + )); + }; + Ok(Some((id.to_string(), created_at))) +} + +/// Attach a confirmation caveat to an accepted write response. +/// +/// The relay accepted the event, so this is a success: stdout keeps the +/// documented `{event_id, accepted, message}` shape, and the extra `warning` +/// key is additive the same way the create commands inject the new entity id. +fn unconfirmed_write_response(rendered: &str, reason: &str) -> String { + match serde_json::from_str::(rendered) { + Ok(serde_json::Value::Object(mut response)) => { + response.insert( + "warning".to_string(), + serde_json::Value::String(reason.to_string()), + ); + serde_json::Value::Object(response).to_string() + } + _ => rendered.to_string(), + } } pub async fn cmd_set_canvas( @@ -1196,10 +1557,11 @@ pub async fn dispatch_canvas(cmd: crate::CanvasCmd, client: &BuzzClient) -> Resu #[cfg(test)] mod tests { use super::{ - apply_cardinality_rule, build_template_report, cmd_set_add_policy, - finalize_roster_resolution, name_matches, resolve_roster_with_archive_filter, + apply_cardinality_rule, build_template_report, classify_confirmation, cmd_set_add_policy, + finalize_roster_resolution, name_matches, parse_stored_profile, profile_publish_timestamp, + profile_read_error, resolve_roster_with_archive_filter, unconfirmed_write_response, validate_ttl_seconds, validate_update_channel_fields, ArchivedExclusion, ChannelSummary, - ResolvedAgent, RosterResolution, SkippedSlug, + Confirmation, ResolvedAgent, RosterResolution, SkippedSlug, MAX_PROFILE_PUBLISH_LEAD_SECS, }; use crate::client::BuzzClient; use crate::CliError; @@ -1747,4 +2109,186 @@ mod tests { "no warning key expected: {report}" ); } + + // ---- kind:10100 profile merge ----------------------------------------- + + fn profile_event(created_at: u64, content: &str) -> serde_json::Value { + json!({ "created_at": created_at, "content": content }) + } + + #[test] + fn a_stored_profile_keeps_every_field_the_policy_does_not_own() { + // The defect: publishing `{"channel_add_policy": ...}` alone replaced + // the whole event body, taking the agent out of every @mention picker. + let stored = parse_stored_profile(&[profile_event( + 100, + r#"{"name":"probe","channel_ids":["a","b"],"respond_to":["owner"]}"#, + )]) + .expect("a JSON object body parses"); + let mut merged = stored.body; + merged.insert("channel_add_policy".into(), json!("open")); + + assert_eq!(merged.get("name"), Some(&json!("probe"))); + assert_eq!(merged.get("channel_ids"), Some(&json!(["a", "b"]))); + assert_eq!(merged.get("respond_to"), Some(&json!(["owner"]))); + assert_eq!(merged.get("channel_add_policy"), Some(&json!("open"))); + } + + #[test] + fn no_stored_profile_is_the_only_case_that_yields_an_empty_body() { + let none = parse_stored_profile(&[]).expect("no events is not an error"); + assert!(none.body.is_empty()); + assert_eq!(none.created_at, 0); + + let blank = parse_stored_profile(&[profile_event(42, " ")]) + .expect("an empty body carries nothing to preserve"); + assert!(blank.body.is_empty()); + // Still out-bid the stored copy: the event exists, only its body is empty. + assert_eq!(blank.created_at, 42); + } + + #[test] + fn an_unreadable_profile_is_refused_rather_than_replaced() { + // Treating this as "no profile" would republish a single-field profile + // and erase the rest, which is the wipe this merge exists to prevent. + for content in [r#"["not","an","object"]"#, "not json at all", r#""string""#] { + let err = parse_stored_profile(&[profile_event(1, content)]) + .expect_err("a non-object body must not be treated as absent"); + assert!( + matches!(err, CliError::Other(ref m) if m.contains("refusing to replace")), + "unexpected error for {content:?}: {err:?}" + ); + } + } + + #[test] + fn a_merge_out_bids_the_stored_copy_instead_of_tying_it() { + // A same-second write is resolved by lowest event id, so publishing at + // `now` is a coin flip against the harness republishing concurrently. + assert_eq!(profile_publish_timestamp(1_000, 999).unwrap(), 1_000); + assert_eq!(profile_publish_timestamp(1_000, 1_000).unwrap(), 1_001); + assert_eq!(profile_publish_timestamp(1_000, 1_050).unwrap(), 1_051); + } + + #[test] + fn ordinary_clock_skew_does_not_block_a_policy_change() { + // Refusing on a few seconds of skew made the command unusable wherever + // the harness host's clock led the operator's. + let stored = 1_000 + MAX_PROFILE_PUBLISH_LEAD_SECS - 1; + assert!(profile_publish_timestamp(1_000, stored).is_ok()); + } + + #[test] + fn a_stored_stamp_past_the_relay_window_is_refused_not_out_bid() { + let stored = 1_000 + MAX_PROFILE_PUBLISH_LEAD_SECS + 1; + let err = profile_publish_timestamp(1_000, stored) + .expect_err("out-bidding this would fall outside the relay's tolerance"); + assert!( + matches!(err, CliError::Other(ref m) if m.contains("clock skew")), + "unexpected error: {err:?}" + ); + } + + #[test] + fn a_result_row_we_cannot_read_is_refused_rather_than_treated_as_absent() { + // Defaulting `content` to "" republished a single-field profile — the + // wipe this merge exists to prevent — and defaulting `created_at` to 0 + // dropped the write back to a plain `now` stamp, restoring the + // same-second event-id coin flip. + for event in [ + json!({ "created_at": 100 }), + json!({ "created_at": 100, "content": 42 }), + json!({ "content": "{}" }), + ] { + let err = parse_stored_profile(std::slice::from_ref(&event)) + .expect_err("an unreadable row must not look like an absent profile"); + assert!( + matches!(err, CliError::Other(ref m) if m.contains("cannot interpret")), + "unexpected error for {event}: {err:?}" + ); + } + } + + #[test] + fn a_transient_profile_read_failure_keeps_its_exit_code() { + // Flattening these to `Other` reports exit 4 and `retryable: false` for + // a relay outage or an expired auth tag, against the exit-code contract + // agents drive the CLI by. + let relay = profile_read_error(CliError::Relay { + status: 503, + body: "unavailable".into(), + }); + assert!( + matches!(relay, CliError::Relay { status: 503, .. }), + "{relay:?}" + ); + assert!(crate::error::is_retryable_error(&relay)); + assert_eq!(crate::error::exit_code(&relay), 2); + + let auth = profile_read_error(CliError::Auth("token expired".into())); + assert_eq!(crate::error::exit_code(&auth), 3); + + let other = profile_read_error(CliError::Other("unreadable body".into())); + assert!(matches!(other, CliError::Other(_)), "{other:?}"); + assert_eq!(crate::error::exit_code(&other), 4); + } + + // ---- post-publish confirmation ---------------------------------------- + + #[test] + fn our_own_event_reading_back_is_the_only_landed_case() { + assert_eq!( + classify_confirmation("aa", 1_000, Some(("aa", 1_000))), + Confirmation::Landed + ); + } + + #[test] + fn a_read_that_cannot_see_our_write_is_not_a_conflict() { + // The confirmation filter is neither channel-pinned nor `until`-bounded, + // so the relay may serve it from a lagging read replica and return the + // pre-write event. A stored copy that loses the relay's replace + // comparison cannot have replaced ours, so the read is behind — and + // calling that a conflict republished a byte-identical event, drew + // `duplicate:`, and exited 5 on a policy change that was stored. + let older = classify_confirmation("bb", 1_001, Some(("aa", 1_000))); + assert!(matches!(older, Confirmation::Unconfirmed(_)), "{older:?}"); + + // Same second, higher stored id: the lowest-id tiebreak went to us, so + // this copy cannot be the stored head either. + let tie = classify_confirmation("aa", 1_000, Some(("bb", 1_000))); + assert!(matches!(tie, Confirmation::Unconfirmed(_)), "{tie:?}"); + + // Our write was accepted, so nothing can have removed it. + let empty = classify_confirmation("aa", 1_000, None); + assert!(matches!(empty, Confirmation::Unconfirmed(_)), "{empty:?}"); + } + + #[test] + fn a_copy_that_beats_ours_is_a_replacement_to_merge_onto() { + assert_eq!( + classify_confirmation("aa", 1_000, Some(("bb", 1_001))), + Confirmation::Replaced + ); + // Same second, lower stored id: the lowest-id tiebreak went to the peer. + assert_eq!( + classify_confirmation("bb", 1_000, Some(("aa", 1_000))), + Confirmation::Replaced + ); + } + + #[test] + fn an_unconfirmed_write_reports_success_with_a_warning() { + let rendered = r#"{"event_id":"aa","accepted":true,"message":"ok"}"#; + let out = unconfirmed_write_response(rendered, "could not confirm the write"); + let parsed: serde_json::Value = serde_json::from_str(&out).expect("json response"); + // The documented write-response shape survives; the caveat is additive. + assert_eq!(parsed.get("event_id"), Some(&json!("aa"))); + assert_eq!(parsed.get("accepted"), Some(&json!(true))); + assert_eq!(parsed.get("message"), Some(&json!("ok"))); + assert_eq!( + parsed.get("warning"), + Some(&json!("could not confirm the write")) + ); + } }