Continue retained native operation strands across host recovery

CHANGELOG.md

@@ -7,6 +7,11 @@
 
 ### Added
 
+- The trusted executable-operation host can retain two native alternative strands,
+  continue a losing strand under new observations, and recover its fork ancestry
+  and operation outcomes from the WAL. Application selection records do not
+  imply merge or promotion.
+
 - Bounded executable-operation host sessions retain immutable observations and
   logical-request bindings in the native WAL. Echo evaluates supplied node/atom
   preconditions inside operation preparation, includes their reads in scheduler
@@ -22,6 +27,8 @@
 
 ### Fixed
 
+- Retained local fork retries reconcile durable topology after uncertain appends, preserving the original basis and preventing duplicate fork records.
+
 - Observation slots preflight their canonical byte bound and charge execution reads before copying Atom payloads.
 
 - Private Edict runtime decoding resolves named integer types through their declared width, matching source-function provider acceptance.

crates/warp-core/src/trusted_runtime_host.rs

@@ -15,6 +15,7 @@ use std::{
 use thiserror::Error;
 
 mod observed_context;
+mod retained_strands;
 pub use observed_context::EchoOperationContextErrorV1;
 
 use crate::causal_anchor::prepare_causal_anchor_admission;
@@ -1341,24 +1342,30 @@ impl TrustedRuntimeHost {
         config: TrustedRuntimeWalConfig,
     ) -> Result<(), TrustedRuntimeHostError> {
         let runtime_wal = TrustedRuntimeWal::from_config(config)?;
-        let recovery = runtime_wal.recover_read_only()?;
+        let mut recovery = runtime_wal.recover_read_only()?;
         if let Some(submission_id) = recovery.missing_submission_envelopes.first().copied() {
             return Err(TrustedRuntimeWalError::SubmissionEnvelopeMissing { submission_id }.into());
         }
         if let Some(receipt_digest) = recovery.missing_runtime_state_deltas.first().copied() {
             return Err(TrustedRuntimeWalError::RuntimeStateDeltaMissing { receipt_digest }.into());
         }
-        ensure_runtime_authority_is_durable(
+        let (forked_runtime, forked_provenance) = retained_strands::restore(
             &self.runtime,
             &self.provenance,
+            &runtime_wal,
+            &mut recovery,
+        )?;
+        ensure_runtime_authority_is_durable(
+            &forked_runtime,
+            &forked_provenance,
             &self.engine,
             &recovery,
         )?;
         self.engine
             .preflight_recovered_echo_operation_packages_v1(&recovery.installed_echo_operations)?;
 
-        let mut restored_runtime = self.runtime.clone();
-        let mut restored_provenance = self.provenance.clone();
+        let mut restored_runtime = forked_runtime;
+        let mut restored_provenance = forked_provenance;
         restored_runtime
             .restore_witnessed_submission_persistence(recovery.witnessed_submissions)?;
         restore_provenance_entries(&mut restored_provenance, &recovery.provenance_entries)?;
@@ -4951,7 +4958,9 @@ fn evaluation_basis_matches_recovered_coordinate(
             crate::WorldlineTick::from_raw(parent_tick),
         ))
         .is_some_and(|parent| {
-            parent.head_key == Some(basis.writer_head())
+            // A native fork preserves the source writer's head in its copied
+            // prefix. The next writer need not have authored the parent commit.
+            parent.worldline_id == basis.writer_head().worldline_id
                 && parent.expected.state_root == basis.state_root()
                 && parent.expected.commit_hash == basis.commit_id()
                 && basis.commit_global_tick() == Some(parent.commit_global_tick)
@@ -7145,6 +7154,15 @@ mod tests {
         ));
 
         let mut wrong_root = parent.clone();
+        let mut other_writer = parent.clone();
+        other_writer.head_key = Some(crate::WriterHeadKey {
+            worldline_id: parent.worldline_id,
+            head_id: crate::make_head_id("original-fork-writer"),
+        });
+        assert!(evaluation_basis_matches_recovered_coordinate(
+            basis,
+            &BTreeMap::from([(parent_coordinate, &other_writer)])
+        ));
         wrong_root.expected.state_root = [39; 32];
         let provenance = BTreeMap::from([(parent_coordinate, &wrong_root)]);
         assert!(validate_operation_receipt_parent_material(basis, &child, &provenance).is_err());

crates/warp-core/src/trusted_runtime_host/retained_strands.rs

@@ -0,0 +1,281 @@
+// SPDX-License-Identifier: Apache-2.0
+// © James Ross Ω FLYING•ROBOTS <https://github.com/flyingrobots>
+//! Bounded trusted-local native strand retention and recovery.
+use super::{
+    restore_provenance_entries, Hash, ProvenanceService, ProvenanceStore, RuntimeWalActivationGap,
+    TrustedRuntimeHost, TrustedRuntimeHostError, TrustedRuntimeWal, TrustedRuntimeWalError,
+    TrustedRuntimeWalRecovery, WalAppendAuthority, WalTransactionId, WalTransactionKind,
+    WorldlineRuntime,
+};
+use crate::causal_wal::{StrandForkRecord, WalRecordKind};
+use crate::playback::SessionId;
+use crate::{
+    ActorId, AuthorityBinding, AuthorityDomainId, AuthorityDomainRef, CausalPosture,
+    ForkStrandRequest, InboxPolicy, OriginId, PlaybackMode, RetentionContractId, SealStrength,
+    SessionContext, StrandId, WorldlineTick, WriterHead, WriterHeadKey,
+};
+
+fn invalid() -> TrustedRuntimeHostError {
+    TrustedRuntimeWalError::RuntimeAuthorityNotDurable {
+        gap: RuntimeWalActivationGap::Provenance,
+    }
+    .into()
+}
+fn digest(domain: &[u8], identity: &[u8]) -> Hash {
+    let mut hash = blake3::Hasher::new();
+    hash.update(domain);
+    hash.update(identity);
+    *hash.finalize().as_bytes()
+}
+fn request(record: &StrandForkRecord) -> Result<ForkStrandRequest, TrustedRuntimeHostError> {
+    let identity = *record.strand_id.as_bytes();
+    let origin = OriginId::from_bytes(identity);
+    let session = SessionContext::new(
+        SessionId(identity),
+        origin,
+        ActorId::from_bytes(identity),
+        AuthorityDomainRef::new(origin, AuthorityDomainId::from_bytes(identity)),
+        AuthorityBinding::LocalUnbound { origin },
+        SealStrength::Advisory,
+        CausalPosture::AuthorOnly,
+        None,
+        RetentionContractId::from_bytes(identity),
+    )
+    .map_err(|_| invalid())?;
+    if record.writer_heads.len() != 1
+        || record.writer_heads[0].worldline_id != record.child_worldline_id
+        || record.child_worldline_id.as_bytes() != &digest(b"echo:local-strand-child:v1", &identity)
+        || record.writer_heads[0].head_id.as_bytes()
+            != &digest(b"echo:local-strand-head:v1", &identity)
+        || record.topology_intent_id != identity
+        || record.idempotency_key_digest != Some(identity)
+        || record.retention_posture_digest != digest(b"echo:local-author-only-strand:v1", &identity)
+        || record.issuer_evidence_digest != identity
+    {
+        return Err(invalid());
+    }
+    ForkStrandRequest::from_session_default(
+        record.strand_id,
+        record.source_worldline_id,
+        record.fork_tick,
+        record.child_worldline_id,
+        vec![WriterHead::with_routing(
+            record.writer_heads[0],
+            PlaybackMode::Play,
+            InboxPolicy::AcceptAll,
+            None,
+            true,
+        )],
+        &session,
+    )
+    .map_err(|_| invalid())
+}
+
+fn apply_record(
+    runtime: &mut WorldlineRuntime,
+    provenance: &mut ProvenanceService,
+    record: &StrandForkRecord,
+) -> Result<WriterHeadKey, TrustedRuntimeHostError> {
+    let receipt = runtime.fork_strand(provenance, request(record)?)?;
+    if receipt.fork_basis_ref.commit_hash != record.source_commit_hash
+        || receipt.fork_basis_ref.boundary_hash != record.source_boundary_hash
+    {
+        return Err(invalid());
+    }
+    record.writer_heads.first().copied().ok_or_else(invalid)
+}
+
+// Recover committed topology before counting capacity or acknowledging retries.
+// All candidate changes stay private until every retained record validates.
+fn reconcile_pending_forks(
+    host: &mut TrustedRuntimeHost,
+) -> Result<std::collections::BTreeSet<StrandId>, TrustedRuntimeHostError> {
+    let wal = host.runtime_wal.as_mut().ok_or_else(invalid)?;
+    wal.refresh_cursor_from_store_for_writer()?;
+    let scan = wal
+        .store
+        .recover_read_only()
+        .map_err(TrustedRuntimeWalError::from)?;
+    let mut seen = std::collections::BTreeSet::new();
+    let mut pending = Vec::new();
+    for transaction in scan.transactions {
+        for frame in transaction.frames {
+            if frame.header.record_kind != WalRecordKind::TopologyStrandForkRecorded {
+                continue;
+            }
+            let record = StrandForkRecord::from_payload_bytes(&frame.payload.canonical_bytes)
+                .map_err(|_| invalid())?;
+            if !seen.insert(record.strand_id) || seen.len() > 64 {
+                return Err(invalid());
+            }
+            request(&record)?;
+            if let Some(strand) = host.runtime.strands().get(&record.strand_id) {
+                let basis = strand.fork_basis_ref();
+                let entry = host
+                    .provenance
+                    .entry(record.source_worldline_id, record.fork_tick)?;
+                // ForkBasisRef uses legacy lane terminology for this worldline.
+                let expected_source = record.source_worldline_id;
+                if basis.source_lane_id != expected_source
+                    || basis.fork_tick != record.fork_tick
+                    || basis.commit_hash != record.source_commit_hash
+                    || basis.boundary_hash != record.source_boundary_hash
+                    || strand.child_worldline_id() != record.child_worldline_id
+                    || strand.writer_heads() != record.writer_heads.as_slice()
+                    || entry.expected.commit_hash != record.source_commit_hash
+                    || entry.expected.state_root != record.source_boundary_hash
+                {
+                    return Err(invalid());
+                }
+            } else {
+                pending.push(record);
+            }
+        }
+    }
+    if !pending.is_empty() {
+        let mut runtime = host.runtime.clone();
+        let mut provenance = host.provenance.clone();
+        for record in pending {
+            apply_record(&mut runtime, &mut provenance, &record)?;
+        }
+        host.runtime = runtime;
+        host.provenance = provenance;
+    }
+    Ok(seen)
+}
+
+impl TrustedRuntimeHost {
+    /// Forks a retained AuthorOnly strand under the trusted local host's fixed
+    /// advisory authority profile. This is not an authenticated admission API.
+    /// Reusing an existing strand identity returns its original writer head.
+    ///
+    /// # Errors
+    /// Refuses missing WAL/source history, invalid labels, or more than 64 strands.
+    pub fn fork_local_operation_strand_v1(
+        &mut self,
+        source: WriterHeadKey,
+        label: &str,
+    ) -> Result<WriterHeadKey, TrustedRuntimeHostError> {
+        if label.is_empty() || label.len() > 64 || self.runtime.heads().get(&source).is_none() {
+            return Err(invalid());
+        }
+        let identity = digest(source.worldline_id.as_bytes(), label.as_bytes());
+        let strand_id = StrandId::from_bytes(identity);
+        let retained = reconcile_pending_forks(self)?;
+        if let Some(strand) = self.runtime.strands().get(&strand_id) {
+            if !retained.contains(&strand_id)
+                || strand.fork_basis_ref().source_lane_id != source.worldline_id
+            {
+                return Err(invalid());
+            }
+            return strand.writer_heads().first().copied().ok_or_else(invalid);
+        }
+        if self.runtime.strands().len() >= 64 {
+            return Err(invalid());
+        }
+        let tip = self
+            .provenance
+            .tip_ref(source.worldline_id)?
+            .ok_or_else(invalid)?;
+        let entry = self
+            .provenance
+            .entry(source.worldline_id, tip.worldline_tick)?;
+        let head = WriterHeadKey {
+            worldline_id: crate::WorldlineId::from_bytes(digest(
+                b"echo:local-strand-child:v1",
+                &identity,
+            )),
+            head_id: crate::head::HeadId::from_bytes(digest(
+                b"echo:local-strand-head:v1",
+                &identity,
+            )),
+        };
+        let record = StrandForkRecord {
+            topology_intent_id: identity,
+            strand_id,
+            source_worldline_id: source.worldline_id,
+            fork_tick: tip.worldline_tick,
+            source_commit_hash: tip.commit_hash,
+            source_boundary_hash: entry.expected.state_root,
+            child_worldline_id: head.worldline_id,
+            writer_heads: vec![head],
+            retention_posture_digest: digest(b"echo:local-author-only-strand:v1", &identity),
+            issuer_evidence_digest: identity,
+            idempotency_key_digest: Some(identity),
+        };
+        let mut runtime = self.runtime.clone();
+        let mut provenance = self.provenance.clone();
+        apply_record(&mut runtime, &mut provenance, &record)?;
+        let wal = self.runtime_wal.as_mut().ok_or_else(invalid)?;
+        let mut builder = wal.builder(
+            WalTransactionKind::TopologyIntent,
+            WalAppendAuthority::TrustedScheduler,
+            WalTransactionId::from_hash(identity),
+        );
+        builder
+            .push_record(
+                WalRecordKind::TopologyStrandForkRecorded,
+                record.to_payload_bytes(),
+            )
+            .map_err(TrustedRuntimeWalError::from)?;
+        wal.append_transaction(
+            builder
+                .commit(Vec::new())
+                .map_err(TrustedRuntimeWalError::from)?,
+        )?;
+        self.runtime = runtime;
+        self.provenance = provenance;
+        Ok(head)
+    }
+}
+
+pub(super) fn restore(
+    runtime: &WorldlineRuntime,
+    provenance: &ProvenanceService,
+    wal: &TrustedRuntimeWal,
+    recovery: &mut TrustedRuntimeWalRecovery,
+) -> Result<(WorldlineRuntime, ProvenanceService), TrustedRuntimeHostError> {
+    let mut runtime = runtime.clone();
+    let mut provenance = provenance.clone();
+    let scan = wal
+        .store
+        .recover_read_only()
+        .map_err(TrustedRuntimeWalError::from)?;
+    let mut count = 0;
+    for transaction in scan.transactions {
+        for frame in transaction.frames {
+            if frame.header.record_kind != WalRecordKind::TopologyStrandForkRecorded {
+                continue;
+            }
+            count += 1;
+            if count > 64 {
+                return Err(invalid());
+            }
+            let record = StrandForkRecord::from_payload_bytes(&frame.payload.canonical_bytes)
+                .map_err(|_| invalid())?;
+            if runtime.strands().contains(&record.strand_id) {
+                return Err(invalid());
+            }
+            let source_entries = recovery
+                .provenance_entries
+                .iter()
+                .filter(|entry| entry.worldline_id == record.source_worldline_id)
+                .cloned()
+                .collect::<Vec<_>>();
+            restore_provenance_entries(&mut provenance, &source_entries)?;
+            runtime.restore_causal_runtime_history(&provenance, &source_entries, &[])?;
+            apply_record(&mut runtime, &mut provenance, &record)?;
+            // The copied prefix is derived from the validated native fork and
+            // retained source entries, not reconstructed from application bytes.
+            for tick in 0..=record.fork_tick.as_u64() {
+                recovery.provenance_entries.push(
+                    provenance.entry(record.child_worldline_id, WorldlineTick::from_raw(tick))?,
+                );
+            }
+        }
+    }
+    recovery
+        .provenance_entries
+        .sort_by_key(|entry| (entry.worldline_id, entry.worldline_tick));
+    Ok((runtime, provenance))
+}