commit 13ac1f2135199702a07ba6aa14ab22101f17fcb0
parent 9119bb31c59cd84a4a05c0a75d65d708b49cdd8e
Author: triesap <tyson@radroots.org>
Date: Thu, 10 Sep 2026 02:13:59 +0000
storage: Select verified heads before payload visibility
- Retain opted-in signed contract failures without granting visibility
- Prevent stale fallback while preserving canonical author and tie rules
- Verify same-count admission changes across memory and SQLite
- Pass additive API review coverage workspace and release-preflight gates
Diffstat:
11 files changed, 441 insertions(+), 22 deletions(-)
diff --git a/contracts/api_baselines/radroots_sync.txt b/contracts/api_baselines/radroots_sync.txt
@@ -4,6 +4,9 @@ pub enum radroots_sync::ingest::AdmissionDecision
pub radroots_sync::ingest::AdmissionDecision::Reject
pub radroots_sync::ingest::AdmissionDecision::Verified
pub radroots_sync::ingest::AdmissionDecision::Visible
+pub enum radroots_sync::ingest::ContractFailureDecision
+pub radroots_sync::ingest::ContractFailureDecision::Reject
+pub radroots_sync::ingest::ContractFailureDecision::Verified
pub struct radroots_sync::ingest::IngestBatchReceipt
impl radroots_sync::ingest::IngestBatchReceipt
pub fn radroots_sync::ingest::IngestBatchReceipt::accepted(&self) -> usize
@@ -20,14 +23,17 @@ impl radroots_sync::ingest::RegistryPolicy
pub const fn radroots_sync::ingest::RegistryPolicy::verified() -> Self
pub const fn radroots_sync::ingest::RegistryPolicy::visible() -> Self
impl radroots_sync::ingest::AdmissionPolicy for radroots_sync::ingest::RegistryPolicy
+pub fn radroots_sync::ingest::RegistryPolicy::contract_failure(&self, &radroots_event::verification::SignatureVerifiedEvent) -> radroots_sync::ingest::ContractFailureDecision
pub fn radroots_sync::ingest::RegistryPolicy::decide(&self, &radroots_event::verification::ContractValidatedEvent) -> radroots_sync::ingest::AdmissionDecision
pub fn radroots_sync::ingest::RegistryPolicy::policy_id(&self) -> &'static str
pub fn radroots_sync::ingest::RegistryPolicy::select_contract(&self, &radroots_event::verification::SignatureVerifiedEvent) -> core::option::Option<&'static str>
pub trait radroots_sync::ingest::AdmissionPolicy: core::marker::Send + core::marker::Sync
+pub fn radroots_sync::ingest::AdmissionPolicy::contract_failure(&self, &radroots_event::verification::SignatureVerifiedEvent) -> radroots_sync::ingest::ContractFailureDecision
pub fn radroots_sync::ingest::AdmissionPolicy::decide(&self, &radroots_event::verification::ContractValidatedEvent) -> radroots_sync::ingest::AdmissionDecision
pub fn radroots_sync::ingest::AdmissionPolicy::policy_id(&self) -> &'static str
pub fn radroots_sync::ingest::AdmissionPolicy::select_contract(&self, &radroots_event::verification::SignatureVerifiedEvent) -> core::option::Option<&'static str>
impl radroots_sync::ingest::AdmissionPolicy for radroots_sync::ingest::RegistryPolicy
+pub fn radroots_sync::ingest::RegistryPolicy::contract_failure(&self, &radroots_event::verification::SignatureVerifiedEvent) -> radroots_sync::ingest::ContractFailureDecision
pub fn radroots_sync::ingest::RegistryPolicy::decide(&self, &radroots_event::verification::ContractValidatedEvent) -> radroots_sync::ingest::AdmissionDecision
pub fn radroots_sync::ingest::RegistryPolicy::policy_id(&self) -> &'static str
pub fn radroots_sync::ingest::RegistryPolicy::select_contract(&self, &radroots_event::verification::SignatureVerifiedEvent) -> core::option::Option<&'static str>
diff --git a/contracts/architecture/decisions/verified_event_heads.v1.json b/contracts/architecture/decisions/verified_event_heads.v1.json
@@ -0,0 +1,12 @@
+{
+ "schema": "radroots.verified-event-heads.v1",
+ "status": "implemented",
+ "owners": ["radroots_sync", "radroots_storage", "radroots_storage_sqlite"],
+ "ingest": "Perform actual canonical event-ID and NIP-01 signature verification before any policy decision. Contract validation failure defaults to rejection. A host may explicitly choose signature-verified retention through a decision with only Reject and Verified variants; this path cannot grant visibility. Preserve atomic ingestion, provenance and idempotency.",
+ "head_selection": "Select current replaceable and addressable heads from signature-verified and visible admissions using existing canonical coordinate, timestamp and event-ID rules before payload visibility. Raw records have no head authority. A verified-only winner supersedes an older visible record without itself becoming visible, and deletion of a winner never revives a predecessor.",
+ "coordinates": "Use the existing total NIP-01 selector unchanged: addressable events with missing or empty first d values select the empty address coordinate. Do not add application-specific coordinate validation. Preserve the existing defensive corruption rejection if the selector ever reports a malformed candidate.",
+ "deletion": "Only contract-valid visible deletion requests participate in author-authorized suppression. Retain deletion-before-target evidence and existing address cutoff semantics; verified-only or raw deletion requests have no suppression authority.",
+ "invalidation": "The deterministic shared visibility snapshot digest represents selected heads, admitted deletion requests and resulting visibility. Admission advancement can change this digest without changing the raw event count. Consumers must not use raw count alone as visibility freshness evidence.",
+ "compatibility": "The opt-in sync policy method has a default that preserves existing contract-failure rejection. Existing valid-event policy, wire schemas, package identities, database schema and source boundaries remain unchanged. Retained verified heads intentionally correct stale visible fallback in the shared reducer.",
+ "non_goals": ["application payload policy", "application database ownership", "new protocol or transport", "unverified observation authority", "release publication or deferred platform activation"]
+}
diff --git a/contracts/architecture/deviations.toml b/contracts/architecture/deviations.toml
@@ -2,6 +2,35 @@ schema_version = 1
architecture_id = "radroots.crates.release.v1"
[[deviation]]
+id = "RCRV1-DEV-017"
+date = "2026-09-10"
+status = "closed"
+approval = "Explicit user authorization covers necessary shared-owner prerequisites, verified commits and non-force source publication."
+affected_steps = ["157", "174", "204"]
+spec_anchors = [
+ "contracts/crates/release_v1/radroots_crates_release_v1.toml#package.radroots_storage",
+ "contracts/crates/release_v1/radroots_crates_release_v1.toml#package.radroots_sync",
+]
+source_evidence = [
+ "Shared sync rejects contract-invalid signed observations before the host can choose verified-only retention.",
+ "Shared visibility selects replacement heads only from Visible records, allowing an older visible payload to survive a newer retained signature-verified head.",
+]
+replacement_action = "Implement contracts/architecture/decisions/verified_event_heads.v1.json: explicit default-reject verified retention and canonical head selection before payload visibility, without adding product policy to shared storage."
+verification = [
+ "Reproduce stale fallback and verify raw, malformed, signature, tie, author, deletion-before-target, order, stage advancement and memory/SQLite parity cases.",
+ "Review additive public API and pass affected owner and SDK tests, unchanged coverage thresholds, workspace contracts and release preflight.",
+]
+unresolved_risk = "Owner, backend, SDK, additive API, unchanged coverage and workspace qualification pass. Verified heads intentionally suppress older visible payloads; only opt-in policies retain contract-invalid signed observations. No external release or deferred platform qualification is claimed."
+normative_architecture_change = false
+adr_required = false
+closure_evidence = [
+ "The original stale-fallback regression now passes with raw-to-verified advancement, canonical ties, empty coordinates, reversed arrival order and author-authorized deletion-before-target.",
+ "Real signed sync ingestion preserves default rejection and rejects bad signatures before host retention; opt-in failures stay Verified and valid later admission advances at the same raw count.",
+ "Memory and reopened SQLite return matching admission receipts, visibility snapshots and bounded queries. All isolated owner and SDK lanes pass; only the defaulted retention method and closed decision are additive API changes.",
+ "Fresh storage, SQLite, sync and SDK coverage and all 45 required reports/aggregate pass with unchanged thresholds, together with complete workspace, portable and release-preflight gates.",
+]
+
+[[deviation]]
id = "RCRV1-DEV-016"
date = "2026-09-10"
status = "closed"
diff --git a/crates/storage/README.md b/crates/storage/README.md
@@ -92,6 +92,14 @@ source generation, monotonically increasing position, and every unique
transport provenance observation. Queries are bounded and generation-aware;
backends fail closed on corrupt rows or source changes.
+Current replacement heads are selected from verified and visible admissions
+before interpreting application payloads. A verified-only winner supersedes an
+older visible record without itself becoming visible. Raw records cannot select
+heads, and only visible contract-valid author-authorized deletion requests
+suppress events. Deleted winners never revive predecessors. Consumers use the
+shared visibility snapshot digest to detect admission changes even when the raw
+event count remains unchanged.
+
The journal records a command lifecycle under a validated idempotency key and
optimistic revision. The outbox persists explicit multi-target delivery plans,
leases, attempts, normalized receipts, partial success, and satisfaction
diff --git a/crates/storage/src/event/visibility.rs b/crates/storage/src/event/visibility.rs
@@ -92,11 +92,11 @@ pub fn evaluate_visibility<'a>(
if input.position.generation() != generation {
return Err(Error::CorruptStoredEvent);
}
- if input.stage != AdmissionStage::Visible {
+ if input.stage == AdmissionStage::Raw {
continue;
}
let envelope = input.event.envelope();
- if envelope.kind_u32() == KIND_DELETION_REQUEST {
+ if input.stage == AdmissionStage::Visible && envelope.kind_u32() == KIND_DELETION_REQUEST {
deletion_requests.push(parse_deletion_request(input.event)?);
}
let (coordinate, ephemeral) = match event_head_candidate_for_nip01_event(envelope) {
@@ -119,6 +119,9 @@ pub fn evaluate_visibility<'a>(
EventHeadCandidateResult::NotPersisted => (None, true),
EventHeadCandidateResult::Malformed(_) => return Err(Error::CorruptStoredEvent),
};
+ if input.stage != AdmissionStage::Visible {
+ continue;
+ }
candidates.push(Candidate {
position: input.position,
event: input.event,
@@ -362,6 +365,94 @@ mod tests {
}
#[test]
+ fn verified_head_supersedes_visible_payload_without_becoming_visible() {
+ for (kind, tags) in [(0, vec![]), (30_023, vec![vec!["d", "same"]])] {
+ let old = signed_event(AUTHOR, 10, kind, tags.clone(), r#"{"name":"old"}"#);
+ let invalid = signed_event(AUTHOR, 20, kind, tags, "malformed payload");
+ let before = evaluate([
+ (&old, AdmissionStage::Visible),
+ (&invalid, AdmissionStage::Raw),
+ ]);
+ assert!(before.is_visible(old.id()));
+ let after = evaluate([
+ (&old, AdmissionStage::Visible),
+ (&invalid, AdmissionStage::Verified),
+ ]);
+ assert!(!after.is_visible(old.id()));
+ assert!(!after.is_visible(invalid.id()));
+ assert_eq!(after.snapshot().current_heads()[0].event_id, *invalid.id());
+ assert_eq!(after.snapshot().superseded_event_ids(), &[*old.id()]);
+ assert_ne!(before.snapshot().digest(), after.snapshot().digest());
+ let reverse = evaluate([
+ (&invalid, AdmissionStage::Verified),
+ (&old, AdmissionStage::Visible),
+ ]);
+ assert_eq!(after.snapshot(), reverse.snapshot());
+ }
+ }
+
+ #[test]
+ fn verified_heads_keep_canonical_ties_and_empty_address_coordinates() {
+ let a = signed_event(AUTHOR, 10, 30_023, vec![], "a");
+ let b = signed_event(AUTHOR, 10, 30_023, vec![vec!["d", ""]], "b");
+ let (winner, loser) = if a.id() < b.id() { (&a, &b) } else { (&b, &a) };
+ let result = evaluate([
+ (loser, AdmissionStage::Visible),
+ (winner, AdmissionStage::Verified),
+ ]);
+ assert_eq!(result.snapshot().current_heads().len(), 1);
+ assert_eq!(result.snapshot().current_heads()[0].event_id, *winner.id());
+ assert!(result.snapshot().visible_event_ids().is_empty());
+ let reverse = evaluate([
+ (winner, AdmissionStage::Verified),
+ (loser, AdmissionStage::Visible),
+ ]);
+ assert_eq!(result.snapshot(), reverse.snapshot());
+ let visible_winner = evaluate([
+ (winner, AdmissionStage::Visible),
+ (loser, AdmissionStage::Verified),
+ ]);
+ assert!(visible_winner.is_visible(winner.id()));
+ assert!(!visible_winner.is_visible(loser.id()));
+ }
+
+ #[test]
+ fn verified_deletion_has_no_authority_and_valid_deletion_precedes_target() {
+ let target = signed_event(AUTHOR, 10, 0, vec![], "profile");
+ let deletion = signed_event(
+ AUTHOR,
+ 20,
+ 5,
+ vec![vec!["e", target.id().to_hex().as_str()]],
+ "",
+ );
+ let forged = signed_event(
+ OTHER_AUTHOR,
+ 20,
+ 5,
+ vec![vec!["e", target.id().to_hex().as_str()]],
+ "",
+ );
+ let malformed = signed_event(AUTHOR, 30, 5, vec![], "");
+ for stage in [AdmissionStage::Raw, AdmissionStage::Verified] {
+ let result = evaluate([
+ (&deletion, stage),
+ (&malformed, stage),
+ (&forged, AdmissionStage::Visible),
+ (&target, AdmissionStage::Visible),
+ ]);
+ assert!(result.is_visible(target.id()));
+ assert_eq!(result.snapshot().deletion_request_ids(), &[*forged.id()]);
+ }
+ let result = evaluate([
+ (&deletion, AdmissionStage::Visible),
+ (&target, AdmissionStage::Visible),
+ ]);
+ assert!(!result.is_visible(target.id()));
+ assert_eq!(result.snapshot().suppressed_event_ids(), &[*target.id()]);
+ }
+
+ #[test]
fn selected_head_is_stable_and_a_deleted_head_does_not_resurrect() {
let old = signed_event(AUTHOR, 10, 0, vec![], r#"{"name":"old"}"#);
let current = signed_event(AUTHOR, 20, 0, vec![], r#"{"name":"current"}"#);
diff --git a/crates/storage/tests/memory.rs b/crates/storage/tests/memory.rs
@@ -151,6 +151,51 @@ fn visible_admission(event: SignedEvent, at: u64) -> EventAdmission {
}
#[test]
+fn verified_replacement_changes_visibility_without_increasing_raw_count() {
+ let store = MemoryStorage::new(SourceGeneration::new([19; 32]).unwrap());
+ let old = signed_event_with(10, 0, vec![], r#"{"display_name":"Old Farm","bot":false}"#);
+ let newer = signed_event_with(20, 0, vec![], "malformed profile");
+ block_on(store.admit(visible_admission(old.clone(), 10))).unwrap();
+ let raw = admission(newer.clone(), 20);
+ block_on(store.admit(raw.clone())).unwrap();
+ let before = block_on(store.rebuild_visibility()).unwrap();
+ assert_eq!(before.visible_event_ids(), &[*old.id()]);
+ let verified = RawEvent::new(newer.envelope().clone())
+ .verify_id()
+ .unwrap()
+ .verify_signature(&Allow)
+ .unwrap();
+ let advanced = block_on(
+ store.admit(
+ EventAdmission::verified(
+ ObservedEvent::new(newer.clone(), raw.provenance().clone()),
+ verified,
+ )
+ .unwrap(),
+ ),
+ )
+ .unwrap();
+ assert_eq!(
+ advanced.disposition(),
+ radroots_storage::event::AdmissionDisposition::Advanced
+ );
+ let after = block_on(store.rebuild_visibility()).unwrap();
+ assert_ne!(before.digest(), after.digest());
+ assert_eq!(after.current_heads()[0].event_id, *newer.id());
+ assert!(after.visible_event_ids().is_empty());
+ assert_eq!(
+ block_on(EventStore::status(&store)).unwrap().raw_events(),
+ 2
+ );
+ assert!(
+ block_on(store.query_visible(EventQuery::all(EventQueryBounds::first(10).unwrap())))
+ .unwrap()
+ .items()
+ .is_empty()
+ );
+}
+
+#[test]
fn memory_visibility_rebuild_is_current_delete_aware_and_atomic_parity_safe() {
let generation = SourceGeneration::new([17; 32]).expect("generation");
let direct = MemoryStorage::new(generation);
diff --git a/crates/storage_sqlite/Cargo.toml b/crates/storage_sqlite/Cargo.toml
@@ -54,6 +54,7 @@ sqlx = { workspace = true, features = ["runtime-tokio", "sqlite-bundled"] }
[dev-dependencies]
hex = { workspace = true }
+radroots_storage = { workspace = true, default-features = false, features = ["memory", "serde"] }
tempfile = { workspace = true }
tokio = { workspace = true, features = ["macros", "rt"] }
toml = { workspace = true }
diff --git a/crates/storage_sqlite/src/event/mod.rs b/crates/storage_sqlite/src/event/mod.rs
@@ -946,6 +946,51 @@ mod tests {
}
#[tokio::test]
+ async fn verified_replacement_and_same_count_advance_match_memory_after_reopen() {
+ let generation = SourceGeneration::new([19; 32]).unwrap();
+ let store = store(generation).await;
+ let memory = radroots_storage::memory::MemoryStorage::new(generation);
+ let old = signed_event_with(
+ r#"{"display_name":"Old Farm","bot":false}"#,
+ false,
+ 10,
+ 0,
+ vec![],
+ );
+ let newer = signed_event_with("malformed profile", false, 20, 0, vec![]);
+ let admissions = [
+ EventAdmission::visible(observed(old.clone(), 10, None), visible(&old)).unwrap(),
+ EventAdmission::raw(observed(newer.clone(), 20, None)),
+ EventAdmission::verified(observed(newer.clone(), 20, None), verified(&newer)).unwrap(),
+ ];
+ let mut previous_digest = None;
+ for (index, admission) in admissions.into_iter().enumerate() {
+ assert_eq!(
+ store.admit(admission.clone()).await.unwrap(),
+ memory.admit(admission).await.unwrap()
+ );
+ let reopened =
+ SqliteStorage::new(store.pool.clone(), generation, EventStoreMode::ReadWrite);
+ let snapshot = reopened.rebuild_visibility().await.unwrap();
+ assert_eq!(snapshot, memory.rebuild_visibility().await.unwrap());
+ let query = EventQuery::all(EventQueryBounds::first(10).unwrap());
+ assert_eq!(
+ reopened.query_visible(query.clone()).await.unwrap(),
+ memory.query_visible(query).await.unwrap()
+ );
+ if index == 2 {
+ assert_ne!(previous_digest, Some(snapshot.digest()));
+ assert!(snapshot.visible_event_ids().is_empty());
+ assert_eq!(snapshot.current_heads()[0].event_id, *newer.id());
+ assert_eq!(reopened.status().await.unwrap().raw_events(), 2);
+ } else {
+ assert_eq!(snapshot.visible_event_ids(), &[*old.id()]);
+ }
+ previous_digest = Some(snapshot.digest());
+ }
+ }
+
+ #[tokio::test]
async fn visibility_rebuild_survives_reopen_and_matches_current_head_deletion_queries() {
let generation = SourceGeneration::new([18; 32]).expect("generation");
let store = store(generation).await;
diff --git a/crates/sync/README.md b/crates/sync/README.md
@@ -6,6 +6,12 @@ The package owns the shared ingest, pull, projection, push, policy, and status
boundaries. It does not create an executor, spawn workers, install timers, own
process lifecycle, store UI state, or branch on concrete transport adapters.
+Ingest performs real event-ID and signature verification before host policy.
+Contract failure defaults to rejection. A host can explicitly retain such a
+signed observation through `AdmissionPolicy::contract_failure` for canonical
+replacement evidence; its closed decision permits only rejection or verified
+retention, never visibility. Invalid IDs and signatures cannot reach this policy.
+
Pull receipts retain the last available outcome for each target and bounded
cumulative target summaries across returned pages. A later complete outcome
does not erase an earlier incomplete or missing outcome. Summaries preserve
diff --git a/crates/sync/src/ingest.rs b/crates/sync/src/ingest.rs
@@ -34,6 +34,20 @@ pub enum AdmissionDecision {
Visible,
}
+/// Host retention decision for an authentically signed, contract-invalid event.
+///
+/// This decision cannot authorize visibility. It may preserve canonical
+/// replacement evidence before an application can interpret the payload.
+#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
+#[cfg_attr(feature = "serde", serde(rename_all = "snake_case"))]
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub enum ContractFailureDecision {
+ /// Reject without mutating canonical storage.
+ Reject,
+ /// Retain signature-verified evidence without granting visibility.
+ Verified,
+}
+
/// Deterministic host policy for canonical admission and visibility.
///
/// Implementations must be side-effect free. The engine may evaluate a policy
@@ -49,6 +63,12 @@ pub trait AdmissionPolicy: Send + Sync {
None
}
+ /// Decides retention after contract failure and real cryptographic verification.
+ /// Existing policies reject by default; invalid IDs or signatures never reach it.
+ fn contract_failure(&self, _event: &SignatureVerifiedEvent) -> ContractFailureDecision {
+ ContractFailureDecision::Reject
+ }
+
/// Decides whether a contract-valid event is rejected, verified-only, or visible.
fn decide(&self, event: &ContractValidatedEvent) -> AdmissionDecision;
}
@@ -155,28 +175,37 @@ impl Engine {
.clone()
.validate_contract_for_admission(contract_id),
None => verify::contract(verified.clone()),
- }
- .map_err(|_| Error::VerificationFailed)?;
- let decision = policy.decide(&validated);
- if decision == AdmissionDecision::Reject {
- return Err(Error::PolicyRejected);
- }
-
- let admission = match decision {
- AdmissionDecision::Verified => EventAdmission::verified(observed.clone(), verified),
- AdmissionDecision::Visible => {
- let evidence = DecisionEvidence {
- policy_id: policy.policy_id(),
+ };
+ let (decision, admission) = match validated {
+ Ok(validated) => {
+ let decision = policy.decide(&validated);
+ let admission = match decision {
+ AdmissionDecision::Verified => {
+ EventAdmission::verified(observed.clone(), verified)
+ }
+ AdmissionDecision::Visible => {
+ let evidence = DecisionEvidence {
+ policy_id: policy.policy_id(),
+ };
+ let visible = validated
+ .admit_with(&evidence)
+ .and_then(|event| event.make_visible_with(&evidence))
+ .map_err(|never| match never {})?;
+ EventAdmission::visible(observed.clone(), visible)
+ }
+ AdmissionDecision::Reject => return Err(Error::PolicyRejected),
};
- let visible = validated
- .admit_with(&evidence)
- .and_then(|event| event.make_visible_with(&evidence))
- .map_err(|never| match never {})?;
- EventAdmission::visible(observed.clone(), visible)
+ (decision, admission)
}
- AdmissionDecision::Reject => unreachable!("rejection returned before admission"),
- }
- .map_err(map_storage_error)?;
+ Err(_) => match policy.contract_failure(&verified) {
+ ContractFailureDecision::Reject => return Err(Error::VerificationFailed),
+ ContractFailureDecision::Verified => (
+ AdmissionDecision::Verified,
+ EventAdmission::verified(observed.clone(), verified),
+ ),
+ },
+ };
+ let admission = admission.map_err(map_storage_error)?;
let sync_id = self.ids.next_id(OperationKind::Ingest)?;
let requested_at_unix_ms = self.clock.now_unix_ms()?;
diff --git a/crates/sync/tests/ingest.rs b/crates/sync/tests/ingest.rs
@@ -257,6 +257,153 @@ fn valid_visible_ingest_is_atomic_and_preserves_provenance() {
);
}
+struct RetainInvalid {
+ calls: AtomicU8,
+ explicit_contract: bool,
+}
+
+impl AdmissionPolicy for RetainInvalid {
+ fn policy_id(&self) -> &'static str {
+ "test.retain-invalid.v1"
+ }
+ fn select_contract(
+ &self,
+ _: &radroots_event::admission::SignatureVerifiedEvent,
+ ) -> Option<&'static str> {
+ self.explicit_contract
+ .then_some("radroots.food.availability.v1")
+ }
+ fn contract_failure(
+ &self,
+ _: &radroots_event::admission::SignatureVerifiedEvent,
+ ) -> radroots_sync::ingest::ContractFailureDecision {
+ self.calls.fetch_add(1, Ordering::Relaxed);
+ radroots_sync::ingest::ContractFailureDecision::Verified
+ }
+ fn decide(&self, _: &radroots_event::admission::ContractValidatedEvent) -> AdmissionDecision {
+ AdmissionDecision::Visible
+ }
+}
+
+fn observed_profile(created_at: u64, content: &str, valid_signature: bool) -> ObservedEvent {
+ let pair = Keypair::from_secret_key(
+ &Secp256k1::new(),
+ &SecretKey::from_slice(&[1; 32]).expect("fixture key"),
+ );
+ let mut wire = radroots_event::wire::Nip01EventWire {
+ id: "0".repeat(64),
+ pubkey: pair.x_only_public_key().0.to_string(),
+ created_at,
+ kind: 0,
+ tags: vec![],
+ content: content.to_owned(),
+ sig: "42".repeat(64),
+ extra: Default::default(),
+ };
+ let id = wire.computed_event_id().expect("id");
+ wire.id = id.to_hex();
+ if valid_signature {
+ wire.sig = Secp256k1::new()
+ .sign_schnorr_no_aux_rand(&Message::from_digest(*id.as_bytes()), &pair)
+ .to_string();
+ }
+ let raw = serde_json::json!({"id":wire.id,"pubkey":wire.pubkey,"created_at":wire.created_at,"kind":wire.kind,"tags":wire.tags,"content":wire.content,"sig":wire.sig}).to_string();
+ let event = SignedEvent::from_wire_verified_id(wire, raw).expect("ID-valid signed profile");
+ let target = Target::new(TransportId::NOSTR, "wss://relay.example").expect("target");
+ ObservedEvent::new(
+ event,
+ EventProvenance::new(
+ TransportId::NOSTR,
+ target.fingerprint().clone(),
+ created_at * 1000,
+ )
+ .expect("provenance"),
+ )
+}
+
+#[test]
+fn contract_failure_defaults_to_reject_and_opt_in_retains_only_verified_heads() {
+ let (engine, storage) = setup_engine(1);
+ let old = observed_profile(10, r#"{"display_name":"Old Farm","bot":false}"#, true);
+ let newer = observed_profile(20, "not JSON", true);
+ let forged = observed_profile(30, "not JSON", false);
+ block_on(engine.ingest(old.clone(), &RegistryPolicy::visible())).expect("old visible");
+ assert_eq!(
+ block_on(engine.ingest(newer.clone(), &RegistryPolicy::visible())).unwrap_err(),
+ Error::VerificationFailed
+ );
+ assert_eq!(block_on(storage.status()).expect("status").raw_events(), 1);
+ let policy = RetainInvalid {
+ calls: AtomicU8::new(0),
+ explicit_contract: false,
+ };
+ let batch = block_on(engine.ingest_batch(vec![forged, newer.clone(), old], &policy));
+ assert_eq!(batch.accepted(), 2);
+ assert_eq!(batch.rejected(), 1);
+ assert_eq!(batch.outcomes()[0], Err(Error::VerificationFailed));
+ assert_eq!(
+ batch.outcomes()[1].as_ref().unwrap().admission().stage(),
+ AdmissionStage::Verified
+ );
+ assert_eq!(policy.calls.load(Ordering::Relaxed), 1);
+ let status = block_on(storage.status()).expect("status");
+ assert_eq!(status.raw_events(), 2);
+ assert_eq!(status.verified_events(), 2);
+ assert_eq!(status.visible_events(), 0);
+ assert!(
+ block_on(storage.query_visible(EventQuery::all(EventQueryBounds::first(10).unwrap())))
+ .unwrap()
+ .items()
+ .is_empty()
+ );
+ let snapshot = block_on(storage.rebuild_visibility()).unwrap();
+ assert_eq!(snapshot.current_heads()[0].event_id, *newer.event().id());
+}
+
+#[test]
+fn explicit_contract_failure_retention_advances_to_visible_without_new_raw_record() {
+ let (engine, storage) = setup_engine(1);
+ let profile = observed_profile(10, r#"{"display_name":"Farm","bot":false}"#, true);
+ let policy = RetainInvalid {
+ calls: AtomicU8::new(0),
+ explicit_contract: true,
+ };
+ let retained = block_on(engine.ingest(profile.clone(), &policy)).expect("retained");
+ assert_eq!(retained.admission().stage(), AdmissionStage::Verified);
+ let before = block_on(storage.rebuild_visibility()).unwrap();
+ assert!(before.visible_event_ids().is_empty());
+ let advanced = block_on(engine.ingest(profile, &RegistryPolicy::visible())).expect("advanced");
+ assert_eq!(
+ advanced.admission().disposition(),
+ AdmissionDisposition::Advanced
+ );
+ assert_eq!(block_on(storage.status()).unwrap().raw_events(), 1);
+ let after = block_on(storage.rebuild_visibility()).unwrap();
+ assert_ne!(before.digest(), after.digest());
+ assert_eq!(
+ after.visible_event_ids(),
+ &[*retained.admission().event_id()]
+ );
+ assert_eq!(policy.calls.load(Ordering::Relaxed), 1);
+}
+
+#[cfg(feature = "serde")]
+#[test]
+fn contract_failure_wire_decision_cannot_authorize_visibility() {
+ use radroots_sync::ingest::ContractFailureDecision;
+ for value in [
+ ContractFailureDecision::Reject,
+ ContractFailureDecision::Verified,
+ ] {
+ let encoded = serde_json::to_string(&value).unwrap();
+ assert_eq!(
+ serde_json::from_str::<ContractFailureDecision>(&encoded).unwrap(),
+ value
+ );
+ }
+ assert!(serde_json::from_str::<ContractFailureDecision>(r#""visible""#).is_err());
+}
+
#[test]
fn admission_policy_selects_and_fully_validates_admission_only_wire_profiles() {
let (engine, storage) = setup_engine(1);
