From 7ee258b8ab5d0205c7498a9b6bf7de006b16fb6a Mon Sep 17 00:00:00 2001 From: James Ross Date: Thu, 15 Oct 2026 19:39:39 -0700 Subject: [PATCH 1/7] Strands: inherit Continuum occurrences decided in the fork basis (#886) A strand inherits a causal past, and an act already present in that past is not a new act because the worldline changed. When a strand forks, live or while recovery replays the fork, Echo records the submissions its copied ticks attribute. Admission on a strand with no binding of its own walks the fork ancestry: an ancestor's binding counts only when its submission is in that set, so occurrences undecided at the basis, or decided after it, are not inherited, and nested forks inherit through every level. Redelivering an inherited occurrence returns the new non-error disposition AlreadyPresentInInheritedHistory from submit_intent and ingest, naming the original submission, its head, the fork basis and its receipt when ticketed. Nothing is admitted and the host writes no acceptance; the app handle reports a duplicate with inherited_through set. Reusing an inherited key for other material is an occurrence conflict. Unknown coverage on an ancestor obstructs admission. IntentSubmissionDisposition and IngressDisposition are no longer Copy: the inherited evidence is boxed to keep the variants small. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 9 + crates/warp-core/src/coordinator.rs | 412 +++++++++++++++++- crates/warp-core/src/lib.rs | 16 +- crates/warp-core/src/trusted_runtime_host.rs | 9 +- crates/warp-core/tests/inbox.rs | 21 +- ...nstalled_contract_intent_pipeline_tests.rs | 27 +- .../tests/trusted_runtime_host_loop_tests.rs | 94 ++++ crates/warp-wasm/src/warp_kernel.rs | 4 + docs/spec/canonical-inbox-sequencing.md | 22 + 9 files changed, 570 insertions(+), 44 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 449b7f9ac..08245252f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -105,6 +105,15 @@ ### Added +- A strand inherits the Continuum occurrences decided in its causal past + (#886). Redelivering such an occurrence to the strand returns + `AlreadyPresentInInheritedHistory` (from `submit_intent` and `ingest`) + naming the original submission, its head, the fork basis and its receipt + when ticketed; nothing is admitted and no acceptance is written. + `IntentSubmissionHandle` reports it as a duplicate with `inherited_through` + set. Reusing an inherited occurrence key for other material is an + occurrence conflict. Occurrences undecided at the fork basis, or decided + after it, are not inherited. Nested forks and restart preserve this. - Restoring committed history without the witnessed submissions it accounts for no longer reopens admission (#887). `restore_causal_runtime_history` keeps the restored worldline readable, but `submit_intent`, `ingest` and diff --git a/crates/warp-core/src/coordinator.rs b/crates/warp-core/src/coordinator.rs index 58b833f08..8b817a1e9 100644 --- a/crates/warp-core/src/coordinator.rs +++ b/crates/warp-core/src/coordinator.rs @@ -628,7 +628,7 @@ impl SchedulerFaultRecoveryAuthority { } /// Result of accepting an intent into witnessed ingress history. -#[derive(Clone, Copy, Debug, PartialEq, Eq)] +#[derive(Clone, Debug, PartialEq, Eq)] pub enum IntentSubmissionDisposition { /// Echo recorded a new witnessed submission without entering runtime ingress. Accepted { @@ -653,6 +653,36 @@ pub enum IntentSubmissionDisposition { /// identity can be derived from replayed committed state. submission_generation: IngressSubmissionGeneration, }, + /// The Continuum occurrence is already present in the causal past this + /// worldline inherited at its fork basis. + /// + /// This is an idempotency result, not an admission: no submission is + /// created on this worldline and nothing is written. `inherited` names the + /// original submission and the evidence for where the act already exists. + AlreadyPresentInInheritedHistory { + /// Content-addressed ingress id that was presented. + ingress_id: Hash, + /// Resolved writer head on the inheriting worldline. + head_key: WriterHeadKey, + /// The original submission and its inheritance evidence. + inherited: Box, + }, +} + +/// An occurrence's original submission, found in a worldline's inherited past. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct InheritedSubmission { + /// Original witnessed submission that admitted the occurrence. + pub submission_id: Hash, + /// Writer head the original submission belongs to. + pub origin_head_key: WriterHeadKey, + /// Generation of the original submission. + pub submission_generation: IngressSubmissionGeneration, + /// Fork basis of the inheriting worldline, through which the original + /// decision is part of its causal past. + pub fork_basis_ref: crate::ForkBasisRef, + /// Receipt of the original decision, when it was ticketed. + pub causal_receipt_ref: Option, } /// App-facing submission handle returned after Echo accepts or recognizes an @@ -670,8 +700,35 @@ pub struct IntentSubmissionHandle { pub submission_id: Hash, /// Echo-owned intake/correlation generation. pub submission_generation: IngressSubmissionGeneration, - /// `true` when Echo had already witnessed this semantic submission. + /// `true` when Echo had already witnessed this semantic submission, + /// including when it is present in inherited history. pub duplicate: bool, + /// Fork basis through which the submission is inherited, when the + /// occurrence was found in the worldline's inherited past. `head_key` and + /// `submission_id` then name the original submission. + pub inherited_through: Option, +} + +/// Compact identity of the fork basis an inherited submission is reached +/// through: the source worldline, the last copied tick and its commit hash. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct InheritedThrough { + /// Source worldline of the fork. + pub source_lane_id: WorldlineId, + /// Last tick of the copied prefix. + pub fork_tick: WorldlineTick, + /// Commit hash at `fork_tick`. + pub commit_hash: Hash, +} + +impl From for InheritedThrough { + fn from(basis: crate::ForkBasisRef) -> Self { + Self { + source_lane_id: basis.source_lane_id, + fork_tick: basis.fork_tick, + commit_hash: basis.commit_hash, + } + } } impl From for IntentSubmissionHandle { @@ -688,6 +745,7 @@ impl From for IntentSubmissionHandle { submission_id, submission_generation, duplicate: false, + inherited_through: None, }, IntentSubmissionDisposition::Duplicate { ingress_id, @@ -700,13 +758,26 @@ impl From for IntentSubmissionHandle { submission_id, submission_generation, duplicate: true, + inherited_through: None, + }, + IntentSubmissionDisposition::AlreadyPresentInInheritedHistory { + ingress_id, + inherited, + .. + } => Self { + ingress_id, + head_key: inherited.origin_head_key, + submission_id: inherited.submission_id, + submission_generation: inherited.submission_generation, + duplicate: true, + inherited_through: Some(inherited.fork_basis_ref.into()), }, } } } /// Result of ingesting an envelope into the runtime. -#[derive(Clone, Copy, Debug, PartialEq, Eq)] +#[derive(Clone, Debug, PartialEq, Eq)] pub enum IngressDisposition { /// The envelope was admitted to the resolved head inbox. Accepted { @@ -731,6 +802,20 @@ pub enum IngressDisposition { /// identity can be derived from replayed committed state. submission_generation: IngressSubmissionGeneration, }, + /// The Continuum occurrence is already present in the causal past this + /// worldline inherited at its fork basis. + /// + /// This is an idempotency result, not an admission: no submission is + /// created on this worldline and nothing is written. `inherited` names the + /// original submission and the evidence for where the act already exists. + AlreadyPresentInInheritedHistory { + /// Content-addressed ingress id that was presented. + ingress_id: Hash, + /// Resolved writer head on the inheriting worldline. + head_key: WriterHeadKey, + /// The original submission and its inheritance evidence. + inherited: Box, + }, } /// Runtime ingress staged from a witnessed submission and admission ticket. @@ -1313,6 +1398,12 @@ pub struct WorldlineRuntime { /// Witnessed submissions, by worldline, whose envelope material is absent, /// so the occurrence mapping cannot be known complete on that worldline. submissions_without_envelope: BTreeSet<(WorldlineId, Hash)>, + /// Submissions that the inherited ticks of each strand child worldline + /// attribute: the decided part of its causal past at its fork basis. + /// + /// Computed from the child's provenance when the strand forks, live or + /// during recovery. Undecided submissions at the basis are not included. + inherited_decided_submissions: BTreeMap>, /// Worldlines whose committed history was restored before admission /// coverage was established through the restored basis. admission_coverage_pending: BTreeMap, @@ -2545,6 +2636,22 @@ impl WorldlineRuntime { request.fork_tick, request.child_worldline_id, )?; + // The child's causal past is exactly its copied prefix; its + // decided submissions are those that prefix attributes. + let mut inherited = BTreeSet::new(); + for tick in 0..provenance.len(request.child_worldline_id)? { + let entry = provenance + .entry(request.child_worldline_id, WorldlineTick::from_raw(tick))?; + if let Some(attribution) = entry + .tick_receipt + .as_ref() + .and_then(crate::TickReceipt::attribution) + { + inherited.extend(attribution.iter().map(|record| record.submission_id)); + } + } + self.inherited_decided_submissions + .insert(request.child_worldline_id, inherited); let child_target_tick = request.fork_tick.checked_increment().ok_or( RuntimeError::FrontierTickOverflow(request.child_worldline_id), @@ -3233,6 +3340,15 @@ impl WorldlineRuntime { let head_key = self.resolve_target(envelope.target())?; self.check_admission_coverage(head_key.worldline_id)?; self.check_continuum_occurrence_binding(head_key, &envelope)?; + if let Some(inherited) = self.inherited_continuum_occurrence(head_key, &envelope)? { + return Ok( + IntentSubmissionDisposition::AlreadyPresentInInheritedHistory { + ingress_id, + head_key, + inherited, + }, + ); + } if self .worldlines @@ -3522,6 +3638,13 @@ impl WorldlineRuntime { let head_key = self.resolve_target(envelope.target())?; self.check_admission_coverage(head_key.worldline_id)?; self.check_continuum_occurrence_binding(head_key, &envelope)?; + if let Some(inherited) = self.inherited_continuum_occurrence(head_key, &envelope)? { + return Ok(IngressDisposition::AlreadyPresentInInheritedHistory { + ingress_id, + head_key, + inherited, + }); + } if self .worldlines @@ -3627,6 +3750,97 @@ impl WorldlineRuntime { } } + /// Finds an occurrence in the causal past a strand child inherited. + /// + /// Only consulted when this worldline has no binding of its own. Walks the + /// fork ancestry: an ancestor's binding counts only when its submission + /// was decided within the child's inherited prefix, so an occurrence + /// pending at the fork basis, or admitted after it, is not inherited. + /// Missing knowledge about an ancestor obstructs instead of admitting. + fn inherited_continuum_occurrence( + &self, + head_key: WriterHeadKey, + envelope: &IngressEnvelope, + ) -> Result>, RuntimeError> { + let Some(occurrence) = envelope.occurrence() else { + return Ok(None); + }; + let key = occurrence.occurrence_key_digest(); + if self + .continuum_occurrence_bindings + .contains_key(&(head_key.worldline_id, key)) + { + return Ok(None); + } + let (Some(strand), Some(inherited)) = ( + self.strands.find_by_child_worldline(&head_key.worldline_id), + self.inherited_decided_submissions + .get(&head_key.worldline_id), + ) else { + return Ok(None); + }; + let fork_basis_ref = strand.fork_basis_ref; + let mut ancestor = Some(fork_basis_ref.source_lane_id); + // Strands form a tree; the bound only guards against a malformed one. + for _ in 0..=self.strands.len() { + let Some(ancestor_id) = ancestor else { + break; + }; + self.check_admission_coverage(ancestor_id)?; + match self + .continuum_occurrence_bindings + .get(&(ancestor_id, key)) + .copied() + { + Some(submission_id) if inherited.contains(&submission_id) => { + let same_binding = self + .witnessed_submissions + .get(&submission_id) + .zip(self.witnessed_submission_envelopes.get(&submission_id)) + .filter(|(record, existing)| { + record.ingress_id == envelope.ingress_id() + && existing.occurrence() == Some(occurrence) + }) + .map(|(record, _)| record.clone()); + let Some(record) = same_binding else { + return Err(RuntimeError::ContinuumOccurrenceConflict { + worldline_id: head_key.worldline_id, + occurrence_key_digest: key, + existing_submission_id: submission_id, + }); + }; + return Ok(Some(Box::new(InheritedSubmission { + submission_id, + origin_head_key: record.head_key, + submission_generation: record.submission_generation, + fork_basis_ref, + causal_receipt_ref: self + .receipt_correlation_for_submission(&submission_id) + .map(|correlation| correlation.causal_receipt_ref), + }))); + } + Some(_) => {} + None => { + if let Some((_, submission_id)) = self + .submissions_without_envelope + .range((ancestor_id, [0; 32])..=(ancestor_id, [0xff; 32])) + .next() + { + return Err(RuntimeError::ContinuumOccurrenceCoverageUnavailable { + worldline_id: ancestor_id, + submission_id: *submission_id, + }); + } + } + } + ancestor = self + .strands + .find_by_child_worldline(&ancestor_id) + .map(|strand| strand.fork_basis_ref.source_lane_id); + } + Ok(None) + } + /// Applies Continuum's occurrence binding law at admission. /// /// An envelope without an occurrence, or with an occurrence this worldline @@ -5675,7 +5889,8 @@ mod tests { let ingress_id = envelope.ingress_id(); let submission_id = match runtime.ingest(envelope).unwrap() { IngressDisposition::Accepted { submission_id, .. } => submission_id, - IngressDisposition::Duplicate { .. } => { + IngressDisposition::Duplicate { .. } + | IngressDisposition::AlreadyPresentInInheritedHistory { .. } => { panic!("test envelope must be a new submission") } }; @@ -6053,7 +6268,8 @@ mod tests { fn accepted_submission(disposition: IngressDisposition) -> Hash { match disposition { IngressDisposition::Accepted { submission_id, .. } => submission_id, - IngressDisposition::Duplicate { .. } => { + IngressDisposition::Duplicate { .. } + | IngressDisposition::AlreadyPresentInInheritedHistory { .. } => { panic!("expected a new submission, got a duplicate") } } @@ -6168,7 +6384,10 @@ mod tests { IngressDisposition::Duplicate { submission_id, .. } => { assert_eq!(submission_id, first_submission); } - IngressDisposition::Accepted { .. } => panic!("a retry must be a duplicate"), + IngressDisposition::Accepted { .. } + | IngressDisposition::AlreadyPresentInInheritedHistory { .. } => { + panic!("a retry must be a duplicate") + } } let retry_steps = SchedulerCoordinator::super_tick(&mut runtime, &mut provenance, &mut engine).unwrap(); @@ -6272,7 +6491,8 @@ mod tests { fn accepted_intent(disposition: IntentSubmissionDisposition) -> Hash { match disposition { IntentSubmissionDisposition::Accepted { submission_id, .. } => submission_id, - IntentSubmissionDisposition::Duplicate { .. } => { + IntentSubmissionDisposition::Duplicate { .. } + | IntentSubmissionDisposition::AlreadyPresentInInheritedHistory { .. } => { panic!("expected a new submission, got a duplicate") } } @@ -6283,7 +6503,10 @@ mod tests { IntentSubmissionDisposition::Duplicate { submission_id, .. } => { assert_eq!(submission_id, original, "redelivery names the submission"); } - IntentSubmissionDisposition::Accepted { .. } => panic!("redelivery must be a retry"), + IntentSubmissionDisposition::Accepted { .. } + | IntentSubmissionDisposition::AlreadyPresentInInheritedHistory { .. } => { + panic!("redelivery must be a retry") + } } } @@ -6460,6 +6683,142 @@ mod tests { )); } + fn fork_child( + runtime: &mut WorldlineRuntime, + provenance: &mut ProvenanceService, + source_lane_id: WorldlineId, + fork_tick: u64, + label: &str, + seed: u8, + ) -> (WorldlineId, crate::ForkBasisRef) { + let child_worldline_id = wl(seed); + let receipt = runtime + .fork_strand( + provenance, + ForkStrandRequest { + strand_id: make_strand_id(label), + source_lane_id, + fork_tick: wt(fork_tick), + child_worldline_id, + writer_heads: vec![WriterHead::with_routing( + WriterHeadKey { + worldline_id: child_worldline_id, + head_id: make_head_id(label), + }, + PlaybackMode::Play, + InboxPolicy::AcceptAll, + None, + true, + )], + retention_posture: test_retention_posture(10), + }, + ) + .unwrap(); + (child_worldline_id, receipt.fork_basis_ref) + } + + /// Issue #886: a strand inherits the occurrences decided in its causal + /// past at the fork basis. Redelivering one is an idempotency result that + /// admits nothing; reusing its key for other material is a conflict; an + /// occurrence pending at the basis or decided after it is admitted on the + /// child as usual. Nested forks inherit through every level. + #[test] + fn a_strand_inherits_the_occurrences_of_its_fork_basis() { + let (mut runtime, mut engine, source) = toy_contract_runtime(); + let occurred = |worldline_id: WorldlineId, key: u8, bytes: &[u8]| { + IngressEnvelope::continuum_occurred_intent( + IngressTarget::DefaultWriter { worldline_id }, + make_intent_kind("test/occurrence"), + bytes.to_vec(), + Vec::new(), + crate::ContinuumOccurrence::new([key; 32], [0xa1; 32]), + ) + }; + let mut provenance = mirrored_provenance(&runtime); + let before_basis = + accepted_submission(runtime.ingest(occurred(source, 1, b"one")).unwrap()); + SchedulerCoordinator::super_tick(&mut runtime, &mut provenance, &mut engine).unwrap(); + accepted_submission(runtime.ingest(occurred(source, 2, b"two")).unwrap()); + SchedulerCoordinator::super_tick(&mut runtime, &mut provenance, &mut engine).unwrap(); + accepted_submission(runtime.ingest(occurred(source, 3, b"pending")).unwrap()); + + // Fork at tick 0: only the first occurrence is in the child's past. + let (child, child_basis) = fork_child(&mut runtime, &mut provenance, source, 0, "child", 2); + let assert_inherited = + |result: IngressDisposition, original: Hash, basis: crate::ForkBasisRef| match result { + IngressDisposition::AlreadyPresentInInheritedHistory { inherited, .. } => { + assert_eq!(inherited.submission_id, original); + assert_eq!(inherited.fork_basis_ref, basis); + } + other => panic!("expected an inherited occurrence, got {other:?}"), + }; + let witnessed = runtime.witnessed_submission_count(); + assert_inherited( + runtime.ingest(occurred(child, 1, b"one")).unwrap(), + before_basis, + child_basis, + ); + assert!(matches!( + runtime.submit_intent(occurred(child, 1, b"one")).unwrap(), + IntentSubmissionDisposition::AlreadyPresentInInheritedHistory { inherited, .. } + if inherited.submission_id == before_basis + )); + assert_eq!( + runtime.witnessed_submission_count(), + witnessed, + "an inherited occurrence admits nothing on the child" + ); + assert!(matches!( + runtime.ingest(occurred(child, 1, b"other")), + Err(RuntimeError::ContinuumOccurrenceConflict { + worldline_id, + existing_submission_id, + .. + }) if worldline_id == child && existing_submission_id == before_basis + )); + // Decided after the basis, or pending at it: not inherited. + accepted_submission(runtime.ingest(occurred(child, 2, b"two")).unwrap()); + accepted_submission(runtime.ingest(occurred(child, 3, b"pending")).unwrap()); + let child_own = accepted_submission(runtime.ingest(occurred(child, 4, b"four")).unwrap()); + SchedulerCoordinator::super_tick(&mut runtime, &mut provenance, &mut engine).unwrap(); + + // A grandchild forked after the child's tick inherits through both levels. + let (grandchild, grandchild_basis) = + fork_child(&mut runtime, &mut provenance, child, 1, "grandchild", 3); + assert_inherited( + runtime.ingest(occurred(grandchild, 1, b"one")).unwrap(), + before_basis, + grandchild_basis, + ); + assert_inherited( + runtime.ingest(occurred(grandchild, 4, b"four")).unwrap(), + child_own, + grandchild_basis, + ); + assert!(matches!( + runtime.ingest(occurred(grandchild, 4, b"changed")), + Err(RuntimeError::ContinuumOccurrenceConflict { + existing_submission_id, + .. + }) if existing_submission_id == child_own + )); + // The grandchild forked from the child's tick 0 sees only the source's. + let (early, early_basis) = fork_child( + &mut runtime, + &mut provenance, + child, + 0, + "early-grandchild", + 4, + ); + assert_inherited( + runtime.ingest(occurred(early, 1, b"one")).unwrap(), + before_basis, + early_basis, + ); + accepted_submission(runtime.ingest(occurred(early, 4, b"four")).unwrap()); + } + /// Issue #887: restoring committed history alone keeps inspection working /// but cannot tell a resent request from a new one, so every admission /// entry point is obstructed until a complete witnessed restore covers the @@ -6567,7 +6926,10 @@ mod tests { .unwrap(); match restored.ingest(committed).unwrap() { IngressDisposition::Duplicate { submission_id, .. } => assert_eq!(submission_id, first), - IngressDisposition::Accepted { .. } => panic!("a committed request must be a retry"), + IngressDisposition::Accepted { .. } + | IngressDisposition::AlreadyPresentInInheritedHistory { .. } => { + panic!("a committed request must be a retry") + } } accepted_submission(restored.ingest(fresh).unwrap()); } @@ -6654,7 +7016,10 @@ mod tests { IngressDisposition::Duplicate { submission_id, .. } => { assert_eq!(submission_id, first, "redelivery returns the submission"); } - IngressDisposition::Accepted { .. } => panic!("redelivery must be a retry"), + IngressDisposition::Accepted { .. } + | IngressDisposition::AlreadyPresentInInheritedHistory { .. } => { + panic!("redelivery must be a retry") + } } let mut other_bytes = bytes.clone(); other_bytes.extend_from_slice(b";other"); @@ -7811,7 +8176,8 @@ mod tests { submission_id, submission_generation, } => Some((ingress_id, head_key, submission_id, submission_generation)), - IngressDisposition::Duplicate { .. } => None, + IngressDisposition::Duplicate { .. } + | IngressDisposition::AlreadyPresentInInheritedHistory { .. } => None, }; assert!( default_accepted.is_some(), @@ -7830,7 +8196,8 @@ mod tests { submission_id, submission_generation, } => Some((ingress_id, head_key, submission_id, submission_generation)), - IngressDisposition::Duplicate { .. } => None, + IngressDisposition::Duplicate { .. } + | IngressDisposition::AlreadyPresentInInheritedHistory { .. } => None, }; assert!( named_accepted.is_some(), @@ -7908,7 +8275,8 @@ mod tests { submission_generation, .. } => Some((submission_id, submission_generation)), - IngressDisposition::Duplicate { .. } => None, + IngressDisposition::Duplicate { .. } + | IngressDisposition::AlreadyPresentInInheritedHistory { .. } => None, }; assert!( first_accepted.is_some(), @@ -7924,7 +8292,8 @@ mod tests { submission_generation, .. } => Some((submission_id, submission_generation)), - IngressDisposition::Accepted { .. } => None, + IngressDisposition::Accepted { .. } + | IngressDisposition::AlreadyPresentInInheritedHistory { .. } => None, }; assert!( duplicate_posture.is_some(), @@ -8011,7 +8380,8 @@ mod tests { submission_generation, .. } => (submission_id, submission_generation), - IntentSubmissionDisposition::Duplicate { .. } => { + IntentSubmissionDisposition::Duplicate { .. } + | IntentSubmissionDisposition::AlreadyPresentInInheritedHistory { .. } => { panic!("first submission should be accepted") } }; @@ -8096,7 +8466,8 @@ mod tests { submission_generation, .. } => (submission_id, submission_generation), - IntentSubmissionDisposition::Duplicate { .. } => { + IntentSubmissionDisposition::Duplicate { .. } + | IntentSubmissionDisposition::AlreadyPresentInInheritedHistory { .. } => { panic!("first submission should be accepted") } }; @@ -8151,7 +8522,8 @@ mod tests { ); let submission_id = match runtime.submit_intent(env).unwrap() { IntentSubmissionDisposition::Accepted { submission_id, .. } => submission_id, - IntentSubmissionDisposition::Duplicate { .. } => { + IntentSubmissionDisposition::Duplicate { .. } + | IntentSubmissionDisposition::AlreadyPresentInInheritedHistory { .. } => { panic!("first submission should be accepted") } }; @@ -8416,7 +8788,8 @@ mod tests { ); let unrelated_submission_id = match runtime.submit_intent(unrelated_envelope).unwrap() { IntentSubmissionDisposition::Accepted { submission_id, .. } => submission_id, - IntentSubmissionDisposition::Duplicate { .. } => { + IntentSubmissionDisposition::Duplicate { .. } + | IntentSubmissionDisposition::AlreadyPresentInInheritedHistory { .. } => { panic!("unrelated submission must be newly witnessed") } }; @@ -9853,7 +10226,8 @@ mod tests { let ingress_id = env_a.ingress_id(); let submission_id = match runtime.ingest(env_a).unwrap() { IngressDisposition::Accepted { submission_id, .. } => submission_id, - IngressDisposition::Duplicate { .. } => { + IngressDisposition::Duplicate { .. } + | IngressDisposition::AlreadyPresentInInheritedHistory { .. } => { unreachable!("first runtime ingress must be accepted") } }; diff --git a/crates/warp-core/src/lib.rs b/crates/warp-core/src/lib.rs index 4153e1443..301d8fb20 100644 --- a/crates/warp-core/src/lib.rs +++ b/crates/warp-core/src/lib.rs @@ -534,14 +534,14 @@ pub use worldline::{ /// Prefer this coordinator/runtime API for new stepping and routing code. pub use coordinator::{ ForkStrandReceipt, ForkStrandRequest, IngressDisposition, IngressSubmissionGeneration, - IntentOutcome, IntentOutcomeDecision, IntentOutcomeObservation, IntentOutcomeReceipt, - IntentSubmissionDisposition, IntentSubmissionHandle, IntentSubmissionRecord, - ReceiptCorrelationPersistenceRecord, ReceiptCorrelationRecord, RuntimeError, - SchedulerCoordinator, SchedulerFaultGeneration, SchedulerFaultId, SchedulerFaultRecord, - SchedulerFaultRecoveryAuthority, SchedulerFaultScope, SchedulerFaultStatus, SchedulerRunId, - StepRecord, TicketedRuntimeIngressAuthority, TicketedRuntimeIngressDisposition, - TicketedRuntimeIngressRecord, WitnessedSubmissionPersistenceRecord, - WitnessedSubmissionPersistenceSnapshot, WorldlineRuntime, + InheritedSubmission, IntentOutcome, IntentOutcomeDecision, IntentOutcomeObservation, + IntentOutcomeReceipt, IntentSubmissionDisposition, IntentSubmissionHandle, + IntentSubmissionRecord, ReceiptCorrelationPersistenceRecord, ReceiptCorrelationRecord, + RuntimeError, SchedulerCoordinator, SchedulerFaultGeneration, SchedulerFaultId, + SchedulerFaultRecord, SchedulerFaultRecoveryAuthority, SchedulerFaultScope, + SchedulerFaultStatus, SchedulerRunId, StepRecord, TicketedRuntimeIngressAuthority, + TicketedRuntimeIngressDisposition, TicketedRuntimeIngressRecord, + WitnessedSubmissionPersistenceRecord, WitnessedSubmissionPersistenceSnapshot, WorldlineRuntime, }; /// Writer-head registry and routing primitives used by the runtime-owned ingress path. pub use head::{ diff --git a/crates/warp-core/src/trusted_runtime_host.rs b/crates/warp-core/src/trusted_runtime_host.rs index 76692729a..ab947e73d 100644 --- a/crates/warp-core/src/trusted_runtime_host.rs +++ b/crates/warp-core/src/trusted_runtime_host.rs @@ -6737,7 +6737,7 @@ impl TrustedRuntimeApp<'_> { return Err(RuntimeError::EchoOperationActionRequiresRuntimeWalAck); } let handle = self.host.runtime.submit_app_intent(envelope)?; - if self.host.runtime_wal.is_none() { + if self.host.runtime_wal.is_none() && handle.inherited_through.is_none() { self.host.track_pending_echo_operation_action_v1( handle.submission_id, is_echo_operation_action, @@ -6884,6 +6884,11 @@ impl TrustedRuntimeApp<'_> { .rollback_intent_submission(submission_rollback, &handle); return Err(TrustedRuntimeHostError::RuntimeWalUnavailable); }; + // An occurrence already in inherited history admits nothing here, so + // there is no acceptance to record. + if handle.inherited_through.is_some() { + return Ok(handle); + } if handle.duplicate && runtime_wal .has_recorded_submission_acceptance(handle.submission_id, envelope.ingress_id()) @@ -8252,6 +8257,7 @@ mod tests { submission_id: [12; 32], submission_generation: IngressSubmissionGeneration::from_raw(1), duplicate: false, + inherited_through: None, }; let record = SubmissionAcceptanceRecord { submission_id: handle.submission_id, @@ -8365,6 +8371,7 @@ mod tests { submission_id: [21; 32], submission_generation: IngressSubmissionGeneration::from_raw(1), duplicate: false, + inherited_through: None, }; let acceptance = SubmissionAcceptanceRecord { submission_id: handle.submission_id, diff --git a/crates/warp-core/tests/inbox.rs b/crates/warp-core/tests/inbox.rs index ca7f61244..18bb033b3 100644 --- a/crates/warp-core/tests/inbox.rs +++ b/crates/warp-core/tests/inbox.rs @@ -249,7 +249,8 @@ fn ticketed_invocation_ingests_runtime_envelope_without_ticking() { ); let accepted = match runtime.submit_intent(envelope.clone()).unwrap() { IntentSubmissionDisposition::Accepted { submission_id, .. } => Some(submission_id), - IntentSubmissionDisposition::Duplicate { .. } => None, + IntentSubmissionDisposition::Duplicate { .. } + | IntentSubmissionDisposition::AlreadyPresentInInheritedHistory { .. } => None, }; assert!(accepted.is_some(), "first submission should be accepted"); let Some(submission) = accepted else { @@ -327,7 +328,8 @@ fn ticketed_ingress_preserves_submission_and_ticket_identity() { ); let accepted = match runtime.submit_intent(envelope.clone()).unwrap() { IntentSubmissionDisposition::Accepted { submission_id, .. } => Some(submission_id), - IntentSubmissionDisposition::Duplicate { .. } => None, + IntentSubmissionDisposition::Duplicate { .. } + | IntentSubmissionDisposition::AlreadyPresentInInheritedHistory { .. } => None, }; assert!(accepted.is_some(), "first submission should be accepted"); let Some(submission) = accepted else { @@ -395,7 +397,8 @@ fn receipt_correlation_does_not_exist_before_scheduler_tick() { ); let accepted = match runtime.submit_intent(envelope.clone()).unwrap() { IntentSubmissionDisposition::Accepted { submission_id, .. } => Some(submission_id), - IntentSubmissionDisposition::Duplicate { .. } => None, + IntentSubmissionDisposition::Duplicate { .. } + | IntentSubmissionDisposition::AlreadyPresentInInheritedHistory { .. } => None, }; assert!(accepted.is_some(), "first submission should be accepted"); let Some(submission) = accepted else { @@ -463,7 +466,8 @@ fn witnessed_submission_outcome_observation_is_pending_without_tick() { submission_generation, .. } => Some((submission_id, submission_generation)), - IntentSubmissionDisposition::Duplicate { .. } => None, + IntentSubmissionDisposition::Duplicate { .. } + | IntentSubmissionDisposition::AlreadyPresentInInheritedHistory { .. } => None, }; assert!(accepted.is_some(), "first submission should be accepted"); let Some((submission, generation)) = accepted else { @@ -497,7 +501,8 @@ fn ticketed_ingress_correlates_tick_receipt_after_scheduler_owned_tick() { ); let accepted = match runtime.submit_intent(envelope.clone()).unwrap() { IntentSubmissionDisposition::Accepted { submission_id, .. } => Some(submission_id), - IntentSubmissionDisposition::Duplicate { .. } => None, + IntentSubmissionDisposition::Duplicate { .. } + | IntentSubmissionDisposition::AlreadyPresentInInheritedHistory { .. } => None, }; assert!(accepted.is_some(), "first submission should be accepted"); let Some(submission) = accepted else { @@ -579,7 +584,8 @@ fn ticketed_submission_outcome_observation_is_decided_after_scheduler_tick() { ); let accepted = match runtime.submit_intent(envelope.clone()).unwrap() { IntentSubmissionDisposition::Accepted { submission_id, .. } => Some(submission_id), - IntentSubmissionDisposition::Duplicate { .. } => None, + IntentSubmissionDisposition::Duplicate { .. } + | IntentSubmissionDisposition::AlreadyPresentInInheritedHistory { .. } => None, }; assert!(accepted.is_some(), "first submission should be accepted"); let Some(submission) = accepted else { @@ -659,7 +665,8 @@ fn ticketed_ingress_rejects_legacy_pending_duplicate_without_correlation() { let accepted = match runtime.ingest(envelope.clone()).unwrap() { IngressDisposition::Accepted { submission_id, .. } => Some(submission_id), - IngressDisposition::Duplicate { .. } => None, + IngressDisposition::Duplicate { .. } + | IngressDisposition::AlreadyPresentInInheritedHistory { .. } => None, }; assert!( accepted.is_some(), diff --git a/crates/warp-core/tests/installed_contract_intent_pipeline_tests.rs b/crates/warp-core/tests/installed_contract_intent_pipeline_tests.rs index d0a0f31c5..8b71cf3af 100644 --- a/crates/warp-core/tests/installed_contract_intent_pipeline_tests.rs +++ b/crates/warp-core/tests/installed_contract_intent_pipeline_tests.rs @@ -480,7 +480,8 @@ fn installed_contract_mutation_dispatches_only_through_ticketed_scheduler_tick() .expect("submission should be witnessed") { IntentSubmissionDisposition::Accepted { submission_id, .. } => submission_id, - IntentSubmissionDisposition::Duplicate { .. } => { + IntentSubmissionDisposition::Duplicate { .. } + | IntentSubmissionDisposition::AlreadyPresentInInheritedHistory { .. } => { panic!("first submission should not be duplicate") } }; @@ -570,7 +571,8 @@ fn unsupported_installed_contract_mutation_cannot_enter_ticketed_runtime_ingress .expect("unsupported submission should still be witnessed") { IntentSubmissionDisposition::Accepted { submission_id, .. } => submission_id, - IntentSubmissionDisposition::Duplicate { .. } => { + IntentSubmissionDisposition::Duplicate { .. } + | IntentSubmissionDisposition::AlreadyPresentInInheritedHistory { .. } => { panic!("first submission should not be duplicate") } }; @@ -620,7 +622,8 @@ fn relabeled_eint_cannot_enter_installed_contract_runtime_ingress() { .expect("relabeled submission is witnessed before contract admission") { IntentSubmissionDisposition::Accepted { submission_id, .. } => submission_id, - IntentSubmissionDisposition::Duplicate { .. } => { + IntentSubmissionDisposition::Duplicate { .. } + | IntentSubmissionDisposition::AlreadyPresentInInheritedHistory { .. } => { panic!("first relabeled submission should not be duplicate") } }; @@ -663,7 +666,8 @@ fn footprint_conflict_is_final_without_hidden_retry() { .expect("first submission should be witnessed") { IntentSubmissionDisposition::Accepted { submission_id, .. } => submission_id, - IntentSubmissionDisposition::Duplicate { .. } => { + IntentSubmissionDisposition::Duplicate { .. } + | IntentSubmissionDisposition::AlreadyPresentInInheritedHistory { .. } => { panic!("first submission must not be duplicate") } }; @@ -672,7 +676,8 @@ fn footprint_conflict_is_final_without_hidden_retry() { .expect("second submission should be witnessed") { IntentSubmissionDisposition::Accepted { submission_id, .. } => submission_id, - IntentSubmissionDisposition::Duplicate { .. } => { + IntentSubmissionDisposition::Duplicate { .. } + | IntentSubmissionDisposition::AlreadyPresentInInheritedHistory { .. } => { panic!("second submission must not be duplicate") } }; @@ -864,7 +869,8 @@ fn witnessed_submission_replay_restores_pending_history_without_runtime_ingress( .expect("first submission should be witnessed") { IntentSubmissionDisposition::Accepted { submission_id, .. } => submission_id, - IntentSubmissionDisposition::Duplicate { .. } => { + IntentSubmissionDisposition::Duplicate { .. } + | IntentSubmissionDisposition::AlreadyPresentInInheritedHistory { .. } => { panic!("first submission must not be duplicate") } }; @@ -873,7 +879,8 @@ fn witnessed_submission_replay_restores_pending_history_without_runtime_ingress( .expect("second submission should be witnessed") { IntentSubmissionDisposition::Accepted { submission_id, .. } => submission_id, - IntentSubmissionDisposition::Duplicate { .. } => { + IntentSubmissionDisposition::Duplicate { .. } + | IntentSubmissionDisposition::AlreadyPresentInInheritedHistory { .. } => { panic!("second submission must not be duplicate") } }; @@ -964,7 +971,8 @@ fn external_contract_fixture_proves_mutation_query_retention_and_replay() { .expect("external fixture submission should be witnessed") { IntentSubmissionDisposition::Accepted { submission_id, .. } => submission_id, - IntentSubmissionDisposition::Duplicate { .. } => { + IntentSubmissionDisposition::Duplicate { .. } + | IntentSubmissionDisposition::AlreadyPresentInInheritedHistory { .. } => { panic!("first external fixture submission must not be duplicate") } }; @@ -1156,7 +1164,8 @@ fn installed_contract_pipeline_replays_to_same_receipt_and_outcome() { .expect("submission should be witnessed") { IntentSubmissionDisposition::Accepted { submission_id, .. } => submission_id, - IntentSubmissionDisposition::Duplicate { .. } => { + IntentSubmissionDisposition::Duplicate { .. } + | IntentSubmissionDisposition::AlreadyPresentInInheritedHistory { .. } => { panic!("first submission must not be duplicate") } }; diff --git a/crates/warp-core/tests/trusted_runtime_host_loop_tests.rs b/crates/warp-core/tests/trusted_runtime_host_loop_tests.rs index 6048be9f5..c0fc6a69f 100644 --- a/crates/warp-core/tests/trusted_runtime_host_loop_tests.rs +++ b/crates/warp-core/tests/trusted_runtime_host_loop_tests.rs @@ -4117,3 +4117,97 @@ fn continuum_occurrence_identity_survives_host_restart_and_redelivery() { drop(restarted); fs::remove_dir_all(&wal_root).expect("WAL fixture cleanup succeeds"); } + +/// Issue #886: after restart, a strand still inherits the occurrences of its +/// fork basis. Redelivering one to the child names the original submission and +/// writes no acceptance; reusing its key for other material is a conflict. +#[test] +fn a_strand_inherits_its_fork_basis_occurrences_across_restart() { + let root = temp_runtime_wal_dir("strand-occurrence-inheritance"); + let (rt, lane) = runtime(); + let head = *rt.heads().iter().next().expect("head").0; + let mut host = TrustedRuntimeHost::new(rt, empty_engine()).expect("host"); + host.enable_runtime_wal(TrustedRuntimeWalConfig::filesystem(&root)) + .expect("wal"); + host.register_contract_package(package()).expect("package"); + let applied = [0xa1; 32]; + let original = host + .app() + .submit_intent_with_runtime_wal_ack(occurred_eint_envelope(lane, 1, applied, MUTATION_VARS)) + .expect("submit"); + host.stage_installed_contract_submission(original.submission_id, &admission_ticket(31)) + .expect("stage"); + host.run_until_idle(4).expect("commit"); + let child = host + .fork_local_operation_strand_v1(head, "occurrence-child") + .expect("native fork"); + drop(host); + + let (rt, _) = runtime(); + let mut restored = TrustedRuntimeHost::new(rt, empty_engine()).expect("host"); + restored + .enable_runtime_wal(TrustedRuntimeWalConfig::filesystem(&root)) + .expect("recovery"); + restored + .register_contract_package(package()) + .expect("package"); + let to_child = |key: u8, vars: &[u8]| { + IngressEnvelope::continuum_occurred_intent( + IngressTarget::ExactHead { key: child }, + make_intent_kind("echo.intent/eint-v1"), + echo_wasm_abi::pack_intent_v1(MUTATION_OP_ID, vars).expect("EINT should pack"), + Vec::new(), + ContinuumOccurrence::new([key; 32], applied), + ) + }; + let acceptances = restored + .runtime_wal() + .expect("wal") + .submission_acceptance_count(); + let witnessed = restored.runtime().witnessed_submission_count(); + let inherited = restored + .app() + .submit_intent_with_runtime_wal_ack(to_child(1, MUTATION_VARS)) + .expect("an inherited occurrence is an idempotency result"); + assert!(inherited.duplicate); + assert_eq!(inherited.submission_id, original.submission_id); + assert_eq!(inherited.head_key, original.head_key); + let fork = restored + .runtime() + .strands() + .find_by_child_worldline(&child.worldline_id) + .expect("retained strand") + .fork_basis_ref(); + assert_eq!(inherited.inherited_through, Some(fork.into())); + assert_eq!( + restored + .runtime_wal() + .expect("wal") + .submission_acceptance_count(), + acceptances, + "an inherited occurrence writes no acceptance" + ); + assert_eq!(restored.runtime().witnessed_submission_count(), witnessed); + + let conflict = restored + .app() + .submit_intent_with_runtime_wal_ack(to_child(1, b"amount=8")) + .expect_err("reusing an inherited occurrence key for other content"); + assert!(matches!( + conflict, + TrustedRuntimeHostError::Runtime(error) if matches!( + *error, + RuntimeError::ContinuumOccurrenceConflict { existing_submission_id, .. } + if existing_submission_id == original.submission_id + ) + )); + + let fresh = restored + .app() + .submit_intent_with_runtime_wal_ack(to_child(2, MUTATION_VARS)) + .expect("an occurrence absent from the inherited past is admitted"); + assert!(!fresh.duplicate && fresh.inherited_through.is_none()); + assert_eq!(fresh.head_key, child); + drop(restored); + fs::remove_dir_all(&root).expect("WAL fixture cleanup succeeds"); +} diff --git a/crates/warp-wasm/src/warp_kernel.rs b/crates/warp-wasm/src/warp_kernel.rs index 4c38a8fea..1c607657c 100644 --- a/crates/warp-wasm/src/warp_kernel.rs +++ b/crates/warp-wasm/src/warp_kernel.rs @@ -1014,6 +1014,10 @@ impl KernelPort for WarpKernel { Some(submission_id.to_vec()), Some(submission_generation.as_u64()), ), + IngressDisposition::AlreadyPresentInInheritedHistory { inherited, .. } => ( + Some(inherited.submission_id.to_vec()), + Some(inherited.submission_generation.as_u64()), + ), }; self.refresh_scheduler_status(); Ok(DispatchResponse { diff --git a/docs/spec/canonical-inbox-sequencing.md b/docs/spec/canonical-inbox-sequencing.md index c2aaa685f..df8823613 100644 --- a/docs/spec/canonical-inbox-sequencing.md +++ b/docs/spec/canonical-inbox-sequencing.md @@ -109,4 +109,26 @@ envelopes are restored, admitting any occurrence on that worldline fails with `RuntimeError::ContinuumOccurrenceCoverageUnavailable`: unknown coverage is not an unused key. Ingress without an occurrence is unaffected. +**Strands inherit occurrences ([#886](https://github.com/flyingrobots/echo/issues/886)).** +A strand's causal past is the prefix it copied at its fork basis +(`ForkBasisRef`: source worldline and fork tick). An act already present in +that past does not become a new act because the worldline changed, so the +occurrence lineage of a strand includes its inherited prefix: + +| Presented on the strand | Echo's answer | +| --- | --- | +| Occurrence decided in the inherited prefix, same binding | `AlreadyPresentInInheritedHistory`, naming the original submission, its head, the fork basis and its receipt when ticketed; nothing is admitted or written | +| Same inherited occurrence key, different binding | `RuntimeError::ContinuumOccurrenceConflict` | +| Occurrence absent from the inherited prefix | Ordinary admission on the strand | +| Occurrence decided on the source after the fork basis | Not inherited | +| Occurrence accepted on the source but undecided at the fork basis | Not inherited: acceptance before the fork is not part of its causal past | + +When a strand forks, live or while recovery replays the fork, Echo records the +submissions its copied ticks attribute. Admission on a strand with no binding of +its own walks the fork ancestry, and an ancestor's binding counts only when its +submission is in that set, so nested forks inherit through every level with +each level's fork tick as the bound. Unknown coverage on an ancestor obstructs +admission. Reconciling occurrences admitted independently on divergent +branches is a separate merge question. + Decision record: [#880](https://github.com/flyingrobots/echo/issues/880). From bd660749e48e879ca1a6cb3d77b3deeb287828cc Mon Sep 17 00:00:00 2001 From: James Ross Date: Thu, 15 Oct 2026 19:52:17 -0700 Subject: [PATCH 2/7] Tests: cover the inherited disposition in the merged coverage test Co-Authored-By: Claude Opus 5.5 --- crates/warp-core/src/coordinator.rs | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/crates/warp-core/src/coordinator.rs b/crates/warp-core/src/coordinator.rs index 426abff01..4a07c7144 100644 --- a/crates/warp-core/src/coordinator.rs +++ b/crates/warp-core/src/coordinator.rs @@ -7048,7 +7048,10 @@ mod tests { restored.attest_complete_witnessed_submissions(); match restored.ingest(committed).unwrap() { IngressDisposition::Duplicate { submission_id, .. } => assert_eq!(submission_id, first), - IngressDisposition::Accepted { .. } => panic!("a committed request must be a retry"), + IngressDisposition::Accepted { .. } + | IngressDisposition::AlreadyPresentInInheritedHistory { .. } => { + panic!("a committed request must be a retry") + } } } From 078aab04ce58bbf921a234d1d7c34effc59a55ba Mon Sep 17 00:00:00 2001 From: James Ross Date: Thu, 15 Oct 2026 22:04:03 -0700 Subject: [PATCH 3/7] Strands: fail closed on unproven prefixes; test the owner's conditions Review of #890 and the owner's conditions for it: - A fork basis older than attribution added nothing to the inherited set, so a child could admit an occurrence its legacy prefix decided. The prefix now proves a decision through the runtime ingress event node its tick created (matched by head id, because forking rewrites copied heads onto the child); a bound request with no event node in the copied state was not decided there; one with a node but no tick evidence obstructs. - A strand registered through register_strand has no recorded prefix and was read as an empty past. Occurrence admission on it now obstructs. - InheritedThrough is re-exported with InheritedSubmission. - New test for the owner's conditions: an inherited decision stays the same however the parent advances, a parent decision made after the fork (even for a request pending at the basis) does not leak, and redelivery changes no retained receipt on either lane. Co-Authored-By: Claude Opus 5.5 --- crates/warp-core/src/coordinator.rs | 284 +++++++++++++++++++++++- crates/warp-core/src/lib.rs | 10 +- docs/spec/canonical-inbox-sequencing.md | 9 +- 3 files changed, 285 insertions(+), 18 deletions(-) diff --git a/crates/warp-core/src/coordinator.rs b/crates/warp-core/src/coordinator.rs index 1f1390f49..0403dc0c6 100644 --- a/crates/warp-core/src/coordinator.rs +++ b/crates/warp-core/src/coordinator.rs @@ -669,6 +669,37 @@ pub enum IntentSubmissionDisposition { }, } +/// Decision evidence in a strand's copied prefix. +#[derive(Clone, Debug, Default, PartialEq, Eq)] +struct InheritedPrefix { + /// Submissions the prefix's attributed ticks decided. + attributed: BTreeSet, + /// Runtime ingress events the prefix's unattributed ticks created, with + /// the writer head id of each tick. Forking rewrites a copied entry's head + /// onto the child worldline, so only the head id survives from the + /// original tick. + legacy_ingress_events: BTreeSet<(crate::head::HeadId, Hash)>, + /// Whether the prefix has ticks older than attribution, so it cannot + /// enumerate everything it decided. + unattributed_ticks: bool, +} + +impl InheritedPrefix { + /// Whether the prefix proves `submission_id` was decided in it: attributed + /// by one of its ticks, or created as a runtime ingress event under the + /// submission's own head by one of its older ticks. + fn proves_decided(&self, runtime: &WorldlineRuntime, submission_id: Hash) -> bool { + self.attributed.contains(&submission_id) + || runtime + .witnessed_submissions + .get(&submission_id) + .is_some_and(|record| { + self.legacy_ingress_events + .contains(&(record.head_key.head_id, record.ingress_id)) + }) + } +} + /// An occurrence's original submission, found in a worldline's inherited past. #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub struct InheritedSubmission { @@ -1405,12 +1436,12 @@ pub struct WorldlineRuntime { /// Witnessed submissions, by worldline, whose envelope material is absent, /// so the occurrence mapping cannot be known complete on that worldline. submissions_without_envelope: BTreeSet<(WorldlineId, Hash)>, - /// Submissions that the inherited ticks of each strand child worldline - /// attribute: the decided part of its causal past at its fork basis. + /// What each strand child worldline's copied prefix proves was decided in + /// its causal past at its fork basis. /// /// Computed from the child's provenance when the strand forks, live or /// during recovery. Undecided submissions at the basis are not included. - inherited_decided_submissions: BTreeMap>, + inherited_prefixes: BTreeMap, /// Worldlines whose committed history was restored before admission /// coverage was established through the restored basis. admission_coverage_pending: BTreeMap, @@ -2639,7 +2670,8 @@ impl WorldlineRuntime { )?; // The child's causal past is exactly its copied prefix; its // decided submissions are those that prefix attributes. - let mut inherited = BTreeSet::new(); + let mut inherited = InheritedPrefix::default(); + let event_type = crate::engine_impl::runtime_ingress_event_type(); for tick in 0..provenance.len(request.child_worldline_id)? { let entry = provenance .entry(request.child_worldline_id, WorldlineTick::from_raw(tick))?; @@ -2648,10 +2680,25 @@ impl WorldlineRuntime { .as_ref() .and_then(crate::TickReceipt::attribution) { - inherited.extend(attribution.iter().map(|record| record.submission_id)); + inherited + .attributed + .extend(attribution.iter().map(|record| record.submission_id)); + } else { + inherited.unattributed_ticks = true; + if let (Some(head_key), Some(patch)) = (entry.head_key, &entry.patch) { + for op in &patch.ops { + if let crate::WarpOp::UpsertNode { node, record } = op { + if record.ty == event_type { + inherited + .legacy_ingress_events + .insert((head_key.head_id, node.local_id.0)); + } + } + } + } } } - self.inherited_decided_submissions + self.inherited_prefixes .insert(request.child_worldline_id, inherited); let child_target_tick = request.fork_tick.checked_increment().ok_or( @@ -3856,14 +3903,21 @@ impl WorldlineRuntime { { return Ok(None); } - let (Some(strand), Some(inherited)) = ( - self.strands.find_by_child_worldline(&head_key.worldline_id), - self.inherited_decided_submissions - .get(&head_key.worldline_id), - ) else { + let Some(strand) = self.strands.find_by_child_worldline(&head_key.worldline_id) else { return Ok(None); }; let fork_basis_ref = strand.fork_basis_ref; + // A strand registered without forking has no recorded prefix + // evidence; an unknown causal past is not an empty one. + let Some(inherited) = self.inherited_prefixes.get(&head_key.worldline_id) else { + return Err(RuntimeError::AdmissionCoverageUnestablished { + worldline_id: head_key.worldline_id, + basis_tick: fork_basis_ref + .fork_tick + .checked_increment() + .unwrap_or(fork_basis_ref.fork_tick), + }); + }; let mut ancestor = Some(fork_basis_ref.source_lane_id); // Strands form a tree; the bound only guards against a malformed one. for _ in 0..=self.strands.len() { @@ -3876,7 +3930,7 @@ impl WorldlineRuntime { .get(&(ancestor_id, key)) .copied() { - Some(submission_id) if inherited.contains(&submission_id) => { + Some(submission_id) if inherited.proves_decided(self, submission_id) => { let same_binding = self .witnessed_submissions .get(&submission_id) @@ -3903,6 +3957,23 @@ impl WorldlineRuntime { .map(|correlation| correlation.causal_receipt_ref), }))); } + // Bound, not proven decided, and the prefix cannot enumerate its + // decisions. Deciding an ingress creates its event node, so if + // the copied state has no such node the request was not decided + // in the prefix. If the node exists without tick evidence, Echo + // cannot tell, and obstructs. + Some(submission_id) + if inherited.unattributed_ticks + && self.copied_state_holds_ingress_node( + head_key.worldline_id, + submission_id, + ) => + { + return Err(RuntimeError::ContinuumOccurrenceCoverageUnavailable { + worldline_id: ancestor_id, + submission_id, + }); + } Some(_) => {} None => { if let Some((_, submission_id)) = self @@ -3925,6 +3996,25 @@ impl WorldlineRuntime { Ok(None) } + /// Whether the worldline's state holds a node keyed by the submission's + /// ingress id (or the submission is unknown, which is treated the same). + fn copied_state_holds_ingress_node( + &self, + worldline_id: WorldlineId, + submission_id: Hash, + ) -> bool { + let Some(record) = self.witnessed_submissions.get(&submission_id) else { + return true; + }; + self.worldlines.get(&worldline_id).is_none_or(|frontier| { + let state = frontier.state(); + state + .warp_state() + .store(&state.root().warp_id) + .is_none_or(|store| store.node(&crate::NodeId(record.ingress_id)).is_some()) + }) + } + /// Applies Continuum's occurrence binding law at admission. /// /// An envelope without an occurrence, or with an occurrence this worldline @@ -6903,6 +6993,176 @@ mod tests { accepted_submission(runtime.ingest(occurred(early, 4, b"four")).unwrap()); } + /// Owner conditions for #890: an inherited occurrence keeps resolving to + /// its decision at the fork basis however the parent advances; a parent + /// decision made after the fork never leaks into the strand, even for a + /// request already pending at the basis; and redelivery changes no + /// retained receipt on either lane. + #[test] + fn inherited_decisions_are_stable_and_post_fork_decisions_do_not_leak() { + let (mut runtime, mut engine, source) = toy_contract_runtime(); + let occurred = |worldline_id: WorldlineId, key: u8, bytes: &[u8]| { + IngressEnvelope::continuum_occurred_intent( + IngressTarget::DefaultWriter { worldline_id }, + make_intent_kind("test/occurrence"), + bytes.to_vec(), + Vec::new(), + crate::ContinuumOccurrence::new([key; 32], [0xa1; 32]), + ) + }; + let mut provenance = mirrored_provenance(&runtime); + let decided = accepted_submission(runtime.ingest(occurred(source, 1, b"one")).unwrap()); + SchedulerCoordinator::super_tick(&mut runtime, &mut provenance, &mut engine).unwrap(); + accepted_submission(runtime.ingest(occurred(source, 5, b"pending")).unwrap()); + let (child, _) = fork_child(&mut runtime, &mut provenance, source, 0, "child", 2); + let inherited_now = |runtime: &mut WorldlineRuntime| match runtime + .ingest(occurred(child, 1, b"one")) + .unwrap() + { + IngressDisposition::AlreadyPresentInInheritedHistory { inherited, .. } => *inherited, + other => panic!("expected an inherited occurrence, got {other:?}"), + }; + let at_fork = inherited_now(&mut runtime); + assert_eq!(at_fork.submission_id, decided); + + // The parent advances: it decides the request pending at the basis + // and a new one. + accepted_submission(runtime.ingest(occurred(source, 6, b"later")).unwrap()); + SchedulerCoordinator::super_tick(&mut runtime, &mut provenance, &mut engine).unwrap(); + assert_eq!(provenance.len(source).unwrap(), 2); + + let entries = |provenance: &ProvenanceService, worldline_id| { + (0..provenance.len(worldline_id).unwrap()) + .map(|tick| provenance.entry(worldline_id, wt(tick)).unwrap()) + .collect::>() + }; + let (parent_before, child_before) = + (entries(&provenance, source), entries(&provenance, child)); + assert_eq!( + inherited_now(&mut runtime), + at_fork, + "stable at the fork basis" + ); + assert_eq!(entries(&provenance, source), parent_before); + assert_eq!(entries(&provenance, child), child_before); + + // Decided on the parent after the fork: ordinary admission on the child. + accepted_submission(runtime.ingest(occurred(child, 5, b"pending")).unwrap()); + accepted_submission(runtime.ingest(occurred(child, 6, b"later")).unwrap()); + } + + /// PR #890 review: a strand registered without forking has no recorded + /// prefix evidence. An unknown causal past is not an empty one, so an + /// occurrence admitted by its source obstructs instead of being admitted + /// again. + #[test] + fn a_strand_registered_without_forking_obstructs_occurrence_admission() { + let (mut runtime, mut engine, source) = toy_contract_runtime(); + let occurred = |worldline_id: WorldlineId| { + IngressEnvelope::continuum_occurred_intent( + IngressTarget::DefaultWriter { worldline_id }, + make_intent_kind("test/occurrence"), + b"one".to_vec(), + Vec::new(), + crate::ContinuumOccurrence::new([1; 32], [0xa1; 32]), + ) + }; + let mut provenance = mirrored_provenance(&runtime); + accepted_submission(runtime.ingest(occurred(source)).unwrap()); + SchedulerCoordinator::super_tick(&mut runtime, &mut provenance, &mut engine).unwrap(); + let (_, fork_basis_ref) = fork_child(&mut runtime, &mut provenance, source, 0, "forked", 2); + + let registered = wl(3); + runtime + .register_worldline(registered, WorldlineState::empty()) + .unwrap(); + let head = register_head( + &mut runtime, + registered, + "registered", + None, + true, + InboxPolicy::AcceptAll, + ); + runtime + .register_strand( + Strand::new( + make_strand_id("registered"), + fork_basis_ref, + registered, + vec![head], + Vec::new(), + test_retention_posture(11), + ) + .unwrap(), + ) + .unwrap(); + assert!(matches!( + runtime.ingest(occurred(registered)), + Err(RuntimeError::AdmissionCoverageUnestablished { worldline_id, .. }) + if worldline_id == registered + )); + } + + /// PR #890 review: a fork basis older than attribution cannot enumerate + /// what it decided. An occurrence whose event its legacy tick created is + /// still inherited; one whose request left no event node in the copied + /// state was not decided there and is admitted as usual. (A request with + /// a node but no tick evidence keeps the parent itself obstructed, which + /// the inheritance walk honours first.) + #[test] + fn a_legacy_fork_basis_inherits_what_its_ticks_prove() { + let (mut source, mut engine, worldline_id) = toy_contract_runtime(); + let occurred = |worldline_id: WorldlineId, key: u8, bytes: &[u8]| { + IngressEnvelope::continuum_occurred_intent( + IngressTarget::DefaultWriter { worldline_id }, + make_intent_kind("test/occurrence"), + bytes.to_vec(), + Vec::new(), + crate::ContinuumOccurrence::new([key; 32], [0xa1; 32]), + ) + }; + let decided = + accepted_submission(source.ingest(occurred(worldline_id, 1, b"one")).unwrap()); + let mut provenance = mirrored_provenance(&source); + SchedulerCoordinator::super_tick(&mut source, &mut provenance, &mut engine).unwrap(); + accepted_submission( + source + .ingest(occurred(worldline_id, 2, b"pending")) + .unwrap(), + ); + let legacy_entry = without_attribution(provenance.entry(worldline_id, wt(0)).unwrap()); + let snapshot = source.witnessed_submission_persistence_snapshot().unwrap(); + + let (mut restored, _engine, _) = toy_contract_runtime(); + let mut legacy_provenance = mirrored_provenance(&restored); + legacy_provenance + .append_local_commit(legacy_entry.clone()) + .unwrap(); + restored + .restore_causal_runtime_history(&legacy_provenance, &[legacy_entry], &[]) + .unwrap(); + restored + .restore_witnessed_submission_persistence(snapshot) + .unwrap(); + restored.attest_complete_witnessed_submissions(); + let (child, _) = fork_child( + &mut restored, + &mut legacy_provenance, + worldline_id, + 0, + "legacy-child", + 2, + ); + + assert!(matches!( + restored.ingest(occurred(child, 1, b"one")).unwrap(), + IngressDisposition::AlreadyPresentInInheritedHistory { inherited, .. } + if inherited.submission_id == decided + )); + accepted_submission(restored.ingest(occurred(child, 2, b"pending")).unwrap()); + } + /// Issue #887: restoring committed history alone keeps inspection working /// but cannot tell a resent request from a new one, so every admission /// entry point is obstructed until a complete witnessed restore covers the diff --git a/crates/warp-core/src/lib.rs b/crates/warp-core/src/lib.rs index 301d8fb20..38cfeee41 100644 --- a/crates/warp-core/src/lib.rs +++ b/crates/warp-core/src/lib.rs @@ -534,11 +534,11 @@ pub use worldline::{ /// Prefer this coordinator/runtime API for new stepping and routing code. pub use coordinator::{ ForkStrandReceipt, ForkStrandRequest, IngressDisposition, IngressSubmissionGeneration, - InheritedSubmission, IntentOutcome, IntentOutcomeDecision, IntentOutcomeObservation, - IntentOutcomeReceipt, IntentSubmissionDisposition, IntentSubmissionHandle, - IntentSubmissionRecord, ReceiptCorrelationPersistenceRecord, ReceiptCorrelationRecord, - RuntimeError, SchedulerCoordinator, SchedulerFaultGeneration, SchedulerFaultId, - SchedulerFaultRecord, SchedulerFaultRecoveryAuthority, SchedulerFaultScope, + InheritedSubmission, InheritedThrough, IntentOutcome, IntentOutcomeDecision, + IntentOutcomeObservation, IntentOutcomeReceipt, IntentSubmissionDisposition, + IntentSubmissionHandle, IntentSubmissionRecord, ReceiptCorrelationPersistenceRecord, + ReceiptCorrelationRecord, RuntimeError, SchedulerCoordinator, SchedulerFaultGeneration, + SchedulerFaultId, SchedulerFaultRecord, SchedulerFaultRecoveryAuthority, SchedulerFaultScope, SchedulerFaultStatus, SchedulerRunId, StepRecord, TicketedRuntimeIngressAuthority, TicketedRuntimeIngressDisposition, TicketedRuntimeIngressRecord, WitnessedSubmissionPersistenceRecord, WitnessedSubmissionPersistenceSnapshot, WorldlineRuntime, diff --git a/docs/spec/canonical-inbox-sequencing.md b/docs/spec/canonical-inbox-sequencing.md index df8823613..547396178 100644 --- a/docs/spec/canonical-inbox-sequencing.md +++ b/docs/spec/canonical-inbox-sequencing.md @@ -128,7 +128,14 @@ submissions its copied ticks attribute. Admission on a strand with no binding of its own walks the fork ancestry, and an ancestor's binding counts only when its submission is in that set, so nested forks inherit through every level with each level's fork tick as the bound. Unknown coverage on an ancestor obstructs -admission. Reconciling occurrences admitted independently on divergent +admission. A prefix tick older than attribution proves a decision only through +the runtime ingress event node it created (matched by head id, since forking +rewrites copied heads onto the child); a bound request with no event node in the +copied state was not decided there, and one with a node but no tick evidence +obstructs. A strand registered without forking has no recorded prefix and +obstructs occurrence admission (`AdmissionCoverageUnestablished`): an unknown +causal past is not an empty one. Redelivery returns the same original decision +however the parent advances, and changes no retained receipt on either lane. Reconciling occurrences admitted independently on divergent branches is a separate merge question. Decision record: [#880](https://github.com/flyingrobots/echo/issues/880). From 0d702c51ab78947b617231d470b99785b7798058 Mon Sep 17 00:00:00 2001 From: James Ross Date: Thu, 15 Oct 2026 22:22:35 -0700 Subject: [PATCH 4/7] Strands: legacy prefixes use the attested decision log, not graph state Following the #888 rework, a fork basis older than attribution no longer infers decisions from runtime ingress event nodes. A bound submission counts as inherited from such a prefix only when its receipt correlation (the decision log the trusted host attests complete) lies on the ancestor at or before the last tick the child inherits; the cap is the minimum fork tick along the ancestor walk. Without a correlation it was not decided. Witness a_legacy_prefix_counts_only_correlated_decisions_within_the_cap fails with correlations ignored and passes now. Co-Authored-By: Claude Opus 5.5 --- crates/warp-core/src/coordinator.rs | 207 ++++++++++-------------- docs/spec/canonical-inbox-sequencing.md | 13 +- 2 files changed, 96 insertions(+), 124 deletions(-) diff --git a/crates/warp-core/src/coordinator.rs b/crates/warp-core/src/coordinator.rs index 0f516d81a..157f2b3a6 100644 --- a/crates/warp-core/src/coordinator.rs +++ b/crates/warp-core/src/coordinator.rs @@ -674,28 +674,38 @@ pub enum IntentSubmissionDisposition { struct InheritedPrefix { /// Submissions the prefix's attributed ticks decided. attributed: BTreeSet, - /// Runtime ingress events the prefix's unattributed ticks created, with - /// the writer head id of each tick. Forking rewrites a copied entry's head - /// onto the child worldline, so only the head id survives from the - /// original tick. - legacy_ingress_events: BTreeSet<(crate::head::HeadId, Hash)>, /// Whether the prefix has ticks older than attribution, so it cannot /// enumerate everything it decided. unattributed_ticks: bool, } impl InheritedPrefix { - /// Whether the prefix proves `submission_id` was decided in it: attributed - /// by one of its ticks, or created as a runtime ingress event under the - /// submission's own head by one of its older ticks. - fn proves_decided(&self, runtime: &WorldlineRuntime, submission_id: Hash) -> bool { - self.attributed.contains(&submission_id) - || runtime - .witnessed_submissions - .get(&submission_id) - .is_some_and(|record| { - self.legacy_ingress_events - .contains(&(record.head_key.head_id, record.ingress_id)) + /// Whether the prefix proves `submission_id` was decided in it. + /// + /// An attributed tick names its decisions. A tick older than attribution + /// does not, so for those the evidence is the submission's receipt + /// correlation, whose coordinate must lie on `ancestor_id` at or before + /// `cap`, the last tick of that ancestor the child inherits. The ancestor's + /// own coverage gate guarantees its decision log is complete, so a + /// submission without a correlation was not decided. Final graph state is + /// never treated as evidence. + fn proves_decided( + &self, + runtime: &WorldlineRuntime, + submission_id: Hash, + ancestor_id: WorldlineId, + cap: WorldlineTick, + ) -> bool { + if self.attributed.contains(&submission_id) { + return true; + } + self.unattributed_ticks + && runtime + .receipt_correlation_for_submission(&submission_id) + .is_some_and(|correlation| { + correlation.causal_receipt_ref.worldline_id == ancestor_id + && correlation.causal_receipt_ref.worldline_tick_after.as_u64() + <= cap.as_u64().saturating_add(1) }) } } @@ -2664,7 +2674,6 @@ impl WorldlineRuntime { // The child's causal past is exactly its copied prefix; its // decided submissions are those that prefix attributes. let mut inherited = InheritedPrefix::default(); - let event_type = crate::engine_impl::runtime_ingress_event_type(); for tick in 0..provenance.len(request.child_worldline_id)? { let entry = provenance .entry(request.child_worldline_id, WorldlineTick::from_raw(tick))?; @@ -2678,17 +2687,6 @@ impl WorldlineRuntime { .extend(attribution.iter().map(|record| record.submission_id)); } else { inherited.unattributed_ticks = true; - if let (Some(head_key), Some(patch)) = (entry.head_key, &entry.patch) { - for op in &patch.ops { - if let crate::WarpOp::UpsertNode { node, record } = op { - if record.ty == event_type { - inherited - .legacy_ingress_events - .insert((head_key.head_id, node.local_id.0)); - } - } - } - } } } self.inherited_prefixes @@ -3852,6 +3850,8 @@ impl WorldlineRuntime { }); }; let mut ancestor = Some(fork_basis_ref.source_lane_id); + // The last tick of the current ancestor that the child inherits. + let mut cap = fork_basis_ref.fork_tick; // Strands form a tree; the bound only guards against a malformed one. for _ in 0..=self.strands.len() { let Some(ancestor_id) = ancestor else { @@ -3863,7 +3863,9 @@ impl WorldlineRuntime { .get(&(ancestor_id, key)) .copied() { - Some(submission_id) if inherited.proves_decided(self, submission_id) => { + Some(submission_id) + if inherited.proves_decided(self, submission_id, ancestor_id, cap) => + { let same_binding = self .witnessed_submissions .get(&submission_id) @@ -3890,23 +3892,6 @@ impl WorldlineRuntime { .map(|correlation| correlation.causal_receipt_ref), }))); } - // Bound, not proven decided, and the prefix cannot enumerate its - // decisions. Deciding an ingress creates its event node, so if - // the copied state has no such node the request was not decided - // in the prefix. If the node exists without tick evidence, Echo - // cannot tell, and obstructs. - Some(submission_id) - if inherited.unattributed_ticks - && self.copied_state_holds_ingress_node( - head_key.worldline_id, - submission_id, - ) => - { - return Err(RuntimeError::ContinuumOccurrenceCoverageUnavailable { - worldline_id: ancestor_id, - submission_id, - }); - } Some(_) => {} None => { if let Some((_, submission_id)) = self @@ -3921,33 +3906,17 @@ impl WorldlineRuntime { } } } - ancestor = self - .strands - .find_by_child_worldline(&ancestor_id) - .map(|strand| strand.fork_basis_ref.source_lane_id); + let next = self.strands.find_by_child_worldline(&ancestor_id); + if let Some(strand) = next { + if strand.fork_basis_ref.fork_tick.as_u64() < cap.as_u64() { + cap = strand.fork_basis_ref.fork_tick; + } + } + ancestor = next.map(|strand| strand.fork_basis_ref.source_lane_id); } Ok(None) } - /// Whether the worldline's state holds a node keyed by the submission's - /// ingress id (or the submission is unknown, which is treated the same). - fn copied_state_holds_ingress_node( - &self, - worldline_id: WorldlineId, - submission_id: Hash, - ) -> bool { - let Some(record) = self.witnessed_submissions.get(&submission_id) else { - return true; - }; - self.worldlines.get(&worldline_id).is_none_or(|frontier| { - let state = frontier.state(); - state - .warp_state() - .store(&state.root().warp_id) - .is_none_or(|store| store.node(&crate::NodeId(record.ingress_id)).is_some()) - }) - } - /// Applies Continuum's occurrence binding law at admission. /// /// An envelope without an occurrence, or with an occurrence this worldline @@ -7037,63 +7006,63 @@ mod tests { )); } - /// PR #890 review: a fork basis older than attribution cannot enumerate - /// what it decided. An occurrence whose event its legacy tick created is - /// still inherited; one whose request left no event node in the copied - /// state was not decided there and is admitted as usual. (A request with - /// a node but no tick evidence keeps the parent itself obstructed, which - /// the inheritance walk honours first.) + /// PR #890 review: a prefix tick older than attribution cannot say what it + /// decided. The evidence is the submission's receipt correlation (the + /// attested decision log): it counts only when it lies on the ancestor at + /// or before the last tick the child inherits. Without one the submission + /// was not decided; final graph state is never consulted. #[test] - fn a_legacy_fork_basis_inherits_what_its_ticks_prove() { - let (mut source, mut engine, worldline_id) = toy_contract_runtime(); - let occurred = |worldline_id: WorldlineId, key: u8, bytes: &[u8]| { - IngressEnvelope::continuum_occurred_intent( + fn a_legacy_prefix_counts_only_correlated_decisions_within_the_cap() { + let (mut runtime, mut engine, worldline_id) = toy_contract_runtime(); + let head_key = runtime + .resolve_target(&IngressTarget::DefaultWriter { worldline_id }) + .unwrap(); + let mut provenance = mirrored_provenance(&runtime); + let decision = |bytes: &[u8]| { + IngressEnvelope::local_intent( IngressTarget::DefaultWriter { worldline_id }, - make_intent_kind("test/occurrence"), - bytes.to_vec(), - Vec::new(), - crate::ContinuumOccurrence::new([key; 32], [0xa1; 32]), + make_intent_kind("echo.intent/eint-v1"), + toy_increment_intent(bytes), ) }; - let decided = - accepted_submission(source.ingest(occurred(worldline_id, 1, b"one")).unwrap()); - let mut provenance = mirrored_provenance(&source); - SchedulerCoordinator::super_tick(&mut source, &mut provenance, &mut engine).unwrap(); - accepted_submission( - source - .ingest(occurred(worldline_id, 2, b"pending")) - .unwrap(), + let first = commit_ticketed_envelope( + &mut runtime, + &mut provenance, + &mut engine, + head_key, + decision(b"amount=1"), + hash(91), ); - let legacy_entry = without_attribution(provenance.entry(worldline_id, wt(0)).unwrap()); - let snapshot = source.witnessed_submission_persistence_snapshot().unwrap(); - - let (mut restored, _engine, _) = toy_contract_runtime(); - let mut legacy_provenance = mirrored_provenance(&restored); - legacy_provenance - .append_local_commit(legacy_entry.clone()) - .unwrap(); - restored - .restore_causal_runtime_history(&legacy_provenance, &[legacy_entry], &[]) - .unwrap(); - restored - .restore_witnessed_submission_persistence(snapshot) - .unwrap(); - restored.attest_complete_witnessed_submissions(); - let (child, _) = fork_child( - &mut restored, - &mut legacy_provenance, - worldline_id, - 0, - "legacy-child", - 2, + let second = commit_ticketed_envelope( + &mut runtime, + &mut provenance, + &mut engine, + head_key, + decision(b"amount=2"), + hash(92), ); - - assert!(matches!( - restored.ingest(occurred(child, 1, b"one")).unwrap(), - IngressDisposition::AlreadyPresentInInheritedHistory { inherited, .. } - if inherited.submission_id == decided + let legacy = InheritedPrefix { + attributed: BTreeSet::new(), + unattributed_ticks: true, + }; + // Decided at tick 0 and tick 1 of the ancestor. + assert!(legacy.proves_decided(&runtime, first.submission_id, worldline_id, wt(0))); + assert!(!legacy.proves_decided(&runtime, second.submission_id, worldline_id, wt(0))); + assert!(legacy.proves_decided(&runtime, second.submission_id, worldline_id, wt(1))); + // On another ancestor, or with no correlation, nothing is proven. + assert!(!legacy.proves_decided(&runtime, first.submission_id, wl(9), wt(5))); + assert!(!legacy.proves_decided(&runtime, hash(93), worldline_id, wt(5))); + // A fully attributed prefix relies on its attribution alone. + let attributed_only = InheritedPrefix { + attributed: BTreeSet::new(), + unattributed_ticks: false, + }; + assert!(!attributed_only.proves_decided( + &runtime, + first.submission_id, + worldline_id, + wt(5) )); - accepted_submission(restored.ingest(occurred(child, 2, b"pending")).unwrap()); } /// Issue #887: restoring committed history alone keeps inspection working diff --git a/docs/spec/canonical-inbox-sequencing.md b/docs/spec/canonical-inbox-sequencing.md index 547396178..b54dd6d0b 100644 --- a/docs/spec/canonical-inbox-sequencing.md +++ b/docs/spec/canonical-inbox-sequencing.md @@ -128,11 +128,14 @@ submissions its copied ticks attribute. Admission on a strand with no binding of its own walks the fork ancestry, and an ancestor's binding counts only when its submission is in that set, so nested forks inherit through every level with each level's fork tick as the bound. Unknown coverage on an ancestor obstructs -admission. A prefix tick older than attribution proves a decision only through -the runtime ingress event node it created (matched by head id, since forking -rewrites copied heads onto the child); a bound request with no event node in the -copied state was not decided there, and one with a node but no tick evidence -obstructs. A strand registered without forking has no recorded prefix and +admission. A prefix tick older than attribution does not say what it decided, so +there the evidence is the submission's receipt correlation, from the decision +log the trusted host attests complete (#887): it counts only when it lies on the +ancestor at or before the last tick the child inherits, and a bound submission +without one was not decided. Final graph state is never treated as evidence. (An +occurrence cannot have been decided by such a tick in practice: occurrence +envelopes exist only since #880, and every writer since #873 attributes its +ticks.) A strand registered without forking has no recorded prefix and obstructs occurrence admission (`AdmissionCoverageUnestablished`): an unknown causal past is not an empty one. Redelivery returns the same original decision however the parent advances, and changes no retained receipt on either lane. Reconciling occurrences admitted independently on divergent From aecf87ee012fa87651d4c168ea4ed097559e44a4 Mon Sep 17 00:00:00 2001 From: James Ross Date: Thu, 15 Oct 2026 22:35:01 -0700 Subject: [PATCH 5/7] Strands: one strand relation per child worldline Review of #890: register_strand accepted a second strand id and fork basis for an existing child worldline, and lookups then chose one relation by strand id order. The chosen basis could disagree with the child's recorded prefix and let an inherited occurrence be admitted again. The registry now refuses it with StrandError::ChildWorldlineAlreadyRelated. Witness a_child_worldline_cannot_belong_to_two_strands fails without the check and passes now. Co-Authored-By: Claude Opus 5.5 --- crates/warp-core/src/coordinator.rs | 50 +++++++++++++++++++++++++++++ crates/warp-core/src/strand.rs | 16 +++++++++ 2 files changed, 66 insertions(+) diff --git a/crates/warp-core/src/coordinator.rs b/crates/warp-core/src/coordinator.rs index 157f2b3a6..d11c12324 100644 --- a/crates/warp-core/src/coordinator.rs +++ b/crates/warp-core/src/coordinator.rs @@ -4862,6 +4862,11 @@ fn hash_strand_error(hasher: &mut blake3::Hasher, err: &StrandError) { hasher.update(b"already-exists"); hasher.update(strand_id.as_bytes()); } + StrandError::ChildWorldlineAlreadyRelated { child, existing } => { + hasher.update(b"child-worldline-already-related"); + hash_worldline_id(hasher, child); + hasher.update(existing.as_bytes()); + } StrandError::NotFound(strand_id) => { hasher.update(b"not-found"); hasher.update(strand_id.as_bytes()); @@ -7006,6 +7011,51 @@ mod tests { )); } + /// PR #890 review: a child worldline has exactly one causal past, so a + /// second strand relation for it is refused instead of letting lookups + /// pick one of two fork bases. + #[test] + fn a_child_worldline_cannot_belong_to_two_strands() { + let (mut runtime, mut engine, source) = toy_contract_runtime(); + let mut provenance = mirrored_provenance(&runtime); + accepted_submission( + runtime + .ingest(IngressEnvelope::local_intent( + IngressTarget::DefaultWriter { + worldline_id: source, + }, + make_intent_kind("echo.intent/eint-v1"), + toy_increment_intent(TOY_INCREMENT_VARS), + )) + .unwrap(), + ); + SchedulerCoordinator::super_tick(&mut runtime, &mut provenance, &mut engine).unwrap(); + let (child, fork_basis_ref) = + fork_child(&mut runtime, &mut provenance, source, 0, "child", 2); + let head = runtime + .resolve_target(&IngressTarget::DefaultWriter { + worldline_id: child, + }) + .unwrap(); + assert!(matches!( + runtime.register_strand( + Strand::new( + make_strand_id("second-relation"), + fork_basis_ref, + child, + vec![head], + Vec::new(), + test_retention_posture(12), + ) + .unwrap(), + ), + Err(RuntimeError::Strand(StrandError::ChildWorldlineAlreadyRelated { + child: refused, + .. + })) if refused == child + )); + } + /// PR #890 review: a prefix tick older than attribution cannot say what it /// decided. The evidence is the submission's receipt correlation (the /// attested decision log): it counts only when it lies on the ancestor at diff --git a/crates/warp-core/src/strand.rs b/crates/warp-core/src/strand.rs index 9b567a027..595e7dc51 100644 --- a/crates/warp-core/src/strand.rs +++ b/crates/warp-core/src/strand.rs @@ -658,6 +658,15 @@ pub enum StrandError { #[error("strand already exists: {0:?}")] AlreadyExists(StrandId), + /// The child worldline already belongs to another strand. + #[error("worldline {child:?} is already the child of strand {existing:?}")] + ChildWorldlineAlreadyRelated { + /// Child worldline named by the rejected strand. + child: WorldlineId, + /// Strand the worldline already belongs to. + existing: StrandId, + }, + /// The strand was not found in the registry. #[error("strand not found: {0:?}")] NotFound(StrandId), @@ -838,6 +847,13 @@ impl StrandRegistry { if self.strands.contains_key(&strand.strand_id) { return Err(StrandError::AlreadyExists(strand.strand_id)); } + // A child worldline has exactly one causal past: one strand relation. + if let Some(existing) = self.find_by_child_worldline(&strand.child_worldline_id) { + return Err(StrandError::ChildWorldlineAlreadyRelated { + child: strand.child_worldline_id, + existing: existing.strand_id, + }); + } validate_local_strand_invariants::( strand.strand_id, &strand.fork_basis_ref, From bb9436cfed59206dd8f283d15d955c3119283df5 Mon Sep 17 00:00:00 2001 From: James Ross Date: Thu, 15 Oct 2026 22:48:50 -0700 Subject: [PATCH 6/7] Strands: do not follow an unproven intermediate relation Review of #890: a strand forked from a registered-only strand has a recorded prefix, so the missing-prefix guard passed, but the ancestry walk then followed the registered-only relation upward and could admit again an occurrence decided further up. The walk now requires every intermediate child it follows to have recorded prefix evidence, and otherwise obstructs with AdmissionCoverageUnestablished naming that worldline. Witness an_unproven_intermediate_relation_obstructs_inheritance_through_it fails without the check and passes now. Co-Authored-By: Claude Opus 5.5 --- crates/warp-core/src/coordinator.rs | 93 +++++++++++++++++++++++++++++ 1 file changed, 93 insertions(+) diff --git a/crates/warp-core/src/coordinator.rs b/crates/warp-core/src/coordinator.rs index d11c12324..d74105992 100644 --- a/crates/warp-core/src/coordinator.rs +++ b/crates/warp-core/src/coordinator.rs @@ -3907,6 +3907,21 @@ impl WorldlineRuntime { } } let next = self.strands.find_by_child_worldline(&ancestor_id); + // Following a strand relation upward is only sound when that child + // was forked with recorded prefix evidence; a registered-only + // relation is an unproven claim about its causal past. + if let Some(strand) = next { + if !self.inherited_prefixes.contains_key(&ancestor_id) { + return Err(RuntimeError::AdmissionCoverageUnestablished { + worldline_id: ancestor_id, + basis_tick: strand + .fork_basis_ref + .fork_tick + .checked_increment() + .unwrap_or(strand.fork_basis_ref.fork_tick), + }); + } + } if let Some(strand) = next { if strand.fork_basis_ref.fork_tick.as_u64() < cap.as_u64() { cap = strand.fork_basis_ref.fork_tick; @@ -7011,6 +7026,84 @@ mod tests { )); } + /// PR #890 review: a strand forked from a registered-only strand has a + /// recorded prefix, but its parent's relation upward is unproven. The + /// ancestry walk must not follow it. + #[test] + fn an_unproven_intermediate_relation_obstructs_inheritance_through_it() { + let (mut runtime, mut engine, source) = toy_contract_runtime(); + let occurred = |worldline_id: WorldlineId| { + IngressEnvelope::continuum_occurred_intent( + IngressTarget::DefaultWriter { worldline_id }, + make_intent_kind("test/occurrence"), + b"one".to_vec(), + Vec::new(), + crate::ContinuumOccurrence::new([1; 32], [0xa1; 32]), + ) + }; + let mut provenance = mirrored_provenance(&runtime); + accepted_submission(runtime.ingest(occurred(source)).unwrap()); + SchedulerCoordinator::super_tick(&mut runtime, &mut provenance, &mut engine).unwrap(); + let (_, fork_basis_ref) = fork_child(&mut runtime, &mut provenance, source, 0, "forked", 2); + + // B claims to descend from the source, by registration only. + let registered = wl(3); + runtime + .register_worldline(registered, WorldlineState::empty()) + .unwrap(); + provenance + .register_worldline(registered, &WorldlineState::empty()) + .unwrap(); + let head = register_head( + &mut runtime, + registered, + "registered", + None, + true, + InboxPolicy::AcceptAll, + ); + runtime + .register_strand( + Strand::new( + make_strand_id("registered"), + fork_basis_ref, + registered, + vec![head], + Vec::new(), + test_retention_posture(11), + ) + .unwrap(), + ) + .unwrap(); + accepted_submission( + runtime + .ingest(IngressEnvelope::local_intent( + IngressTarget::DefaultWriter { + worldline_id: registered, + }, + make_intent_kind("echo.intent/eint-v1"), + toy_increment_intent(TOY_INCREMENT_VARS), + )) + .unwrap(), + ); + SchedulerCoordinator::super_tick(&mut runtime, &mut provenance, &mut engine).unwrap(); + + // C is forked properly from B, so it has a recorded prefix. + let (grandchild, _) = fork_child( + &mut runtime, + &mut provenance, + registered, + 0, + "grandchild", + 4, + ); + assert!(matches!( + runtime.ingest(occurred(grandchild)), + Err(RuntimeError::AdmissionCoverageUnestablished { worldline_id, .. }) + if worldline_id == registered + )); + } + /// PR #890 review: a child worldline has exactly one causal past, so a /// second strand relation for it is refused instead of letting lookups /// pick one of two fork bases. From 35350678ae72673c2a9bb23033687f912fee30e6 Mon Sep 17 00:00:00 2001 From: James Ross Date: Thu, 15 Oct 2026 23:02:41 -0700 Subject: [PATCH 7/7] Strands: validate the whole lineage before trusting any binding Review of #890: when a registered-only intermediate itself held the occurrence binding, the walk returned its inherited result before reaching the intermediate guard. Every strand relation in the ancestry is now checked for recorded prefix evidence before any ancestor binding is read or followed; the per-step guard is replaced by this pre-pass. Witnesses an_unproven_intermediate_binding_is_not_trusted and an_unproven_intermediate_relation_obstructs_inheritance_through_it both fail without the pre-pass and pass now. #901 records the underlying question of register_strand asserting lineage without fork evidence. Co-Authored-By: Claude Opus 5.5 --- crates/warp-core/src/coordinator.rs | 110 ++++++++++++++++++++++++---- 1 file changed, 95 insertions(+), 15 deletions(-) diff --git a/crates/warp-core/src/coordinator.rs b/crates/warp-core/src/coordinator.rs index d74105992..1544b4a10 100644 --- a/crates/warp-core/src/coordinator.rs +++ b/crates/warp-core/src/coordinator.rs @@ -3849,6 +3849,26 @@ impl WorldlineRuntime { .unwrap_or(fork_basis_ref.fork_tick), }); }; + // Every strand relation in the ancestry must be backed by recorded + // prefix evidence (a fork, not a bare registration) before any + // ancestor's binding is trusted or followed. + let mut lineage = fork_basis_ref.source_lane_id; + for _ in 0..=self.strands.len() { + let Some(strand) = self.strands.find_by_child_worldline(&lineage) else { + break; + }; + if !self.inherited_prefixes.contains_key(&lineage) { + return Err(RuntimeError::AdmissionCoverageUnestablished { + worldline_id: lineage, + basis_tick: strand + .fork_basis_ref + .fork_tick + .checked_increment() + .unwrap_or(strand.fork_basis_ref.fork_tick), + }); + } + lineage = strand.fork_basis_ref.source_lane_id; + } let mut ancestor = Some(fork_basis_ref.source_lane_id); // The last tick of the current ancestor that the child inherits. let mut cap = fork_basis_ref.fork_tick; @@ -3907,21 +3927,6 @@ impl WorldlineRuntime { } } let next = self.strands.find_by_child_worldline(&ancestor_id); - // Following a strand relation upward is only sound when that child - // was forked with recorded prefix evidence; a registered-only - // relation is an unproven claim about its causal past. - if let Some(strand) = next { - if !self.inherited_prefixes.contains_key(&ancestor_id) { - return Err(RuntimeError::AdmissionCoverageUnestablished { - worldline_id: ancestor_id, - basis_tick: strand - .fork_basis_ref - .fork_tick - .checked_increment() - .unwrap_or(strand.fork_basis_ref.fork_tick), - }); - } - } if let Some(strand) = next { if strand.fork_basis_ref.fork_tick.as_u64() < cap.as_u64() { cap = strand.fork_basis_ref.fork_tick; @@ -7104,6 +7109,81 @@ mod tests { )); } + /// PR #890 review: the unproven intermediate may itself hold the binding; + /// its lineage must be validated before its binding is trusted. + #[test] + fn an_unproven_intermediate_binding_is_not_trusted() { + let (mut runtime, mut engine, source) = toy_contract_runtime(); + let occurred = |worldline_id: WorldlineId| { + IngressEnvelope::continuum_occurred_intent( + IngressTarget::DefaultWriter { worldline_id }, + make_intent_kind("test/occurrence"), + b"one".to_vec(), + Vec::new(), + crate::ContinuumOccurrence::new([1; 32], [0xa1; 32]), + ) + }; + let mut provenance = mirrored_provenance(&runtime); + accepted_submission( + runtime + .ingest(IngressEnvelope::local_intent( + IngressTarget::DefaultWriter { + worldline_id: source, + }, + make_intent_kind("echo.intent/eint-v1"), + toy_increment_intent(TOY_INCREMENT_VARS), + )) + .unwrap(), + ); + SchedulerCoordinator::super_tick(&mut runtime, &mut provenance, &mut engine).unwrap(); + let (_, fork_basis_ref) = fork_child(&mut runtime, &mut provenance, source, 0, "forked", 2); + + // B decides the occurrence itself, then claims descent by registration. + let registered = wl(3); + runtime + .register_worldline(registered, WorldlineState::empty()) + .unwrap(); + provenance + .register_worldline(registered, &WorldlineState::empty()) + .unwrap(); + let head = register_head( + &mut runtime, + registered, + "registered", + None, + true, + InboxPolicy::AcceptAll, + ); + accepted_submission(runtime.ingest(occurred(registered)).unwrap()); + SchedulerCoordinator::super_tick(&mut runtime, &mut provenance, &mut engine).unwrap(); + runtime + .register_strand( + Strand::new( + make_strand_id("registered"), + fork_basis_ref, + registered, + vec![head], + Vec::new(), + test_retention_posture(11), + ) + .unwrap(), + ) + .unwrap(); + let (grandchild, _) = fork_child( + &mut runtime, + &mut provenance, + registered, + 0, + "grandchild", + 4, + ); + assert!(matches!( + runtime.ingest(occurred(grandchild)), + Err(RuntimeError::AdmissionCoverageUnestablished { worldline_id, .. }) + if worldline_id == registered + )); + } + /// PR #890 review: a child worldline has exactly one causal past, so a /// second strand relation for it is refused instead of letting lookups /// pick one of two fork bases.