diff --git a/.ban-nondeterminism-allowlist b/.ban-nondeterminism-allowlist index bf26dbfa..4f6187ca 100644 --- a/.ban-nondeterminism-allowlist +++ b/.ban-nondeterminism-allowlist @@ -12,6 +12,9 @@ std-fs crates/warp-core/src/wsc/store.rs native WSC filesystem store adapter pen std-fs crates/warp-core/src/wsc/view.rs native WSC file-open helper pending extraction to a boundary adapter. std-fs crates/warp-core/src/causal_wal_tests.rs WAL filesystem fixture I/O only. std-fs crates/warp-core/tests/causal_wal_hardening_tests.rs WAL filesystem fixture I/O only. +std-env crates/warp-core/tests/external_action_protocol_tests.rs external-action filesystem WAL fixture temp directory selection only. +std-fs crates/warp-core/tests/external_action_protocol_tests.rs external-action filesystem WAL fixture I/O only. +std-process crates/warp-core/tests/external_action_protocol_tests.rs external-action filesystem WAL fixture temp directory disambiguation only. std-fs crates/warp-core/tests/external_consumer_contract_fixture_tests.rs installed-contract restart WAL fixture I/O only. std-fs crates/warp-core/tests/executable_operation_pipeline_tests.rs executable-operation restart WAL fixture I/O only. std-fs crates/warp-core/tests/provider_contract_admission_tests.rs provider invocation restart WAL fixture I/O only. diff --git a/CHANGELOG.md b/CHANGELOG.md index e6a138a7..1eb58a1e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,30 @@ ### Added +- Echo now admits domain-neutral external actions through separate + request-before-effect, bounded claim, and settlement-before-resumption WAL + transactions (ADR 0026). Canonical requests bind worldline, operation, + schemas, authority scope, basis, single-claim and retained-byte budgets, input + digest, and reconciliation law. Runtime-owner adapter registration attenuates + operation and scope policy into an exact request-, basis-, and + registry-policy-bound authorization without granting Edict or the provider + seam external authority. `Succeeded`, `Rejected`, `Failed`, and + `OutcomeUnknown` settlements bind the exact request, attempt, adapter, basis, + schema, canonical result bytes, admission evidence, and nonzero external + evidence. Echo derives each lifecycle frontier from a canonical + request-id-keyed sparse Merkle index; insertion order cannot move its root, + one planned mutation advances its bounded path without replaying prior WAL + payloads, and recovery rejects substituted roots. Raw WAL builders and + commit flushes cannot mint the coordinator's opaque authority; causal + transaction coordinates come from one checked local continuation. Arbitrary + recovery reports are observation-only. A coordinator recovered from a + fallible local-store snapshot reconstructs interrupted request tokens, claim + grants, and resumable settlements; storage corruption cannot masquerade as + genesis. Recovery reconstructs requested, claimed, and settled posture from + committed WAL records, including strict filesystem reopen; duplicate, + conflicting, stale, unauthorized, malformed, and over-budget evidence fails + closed. Replay consumes retained settlement bytes and never invokes an + adapter. - The generic Edict-operation runner now exposes complete fresh-host and WAL-recovered application-result records beside the applied result. Report construction fails closed unless all three schema-neutral projection diff --git a/crates/warp-core/src/causal_wal.rs b/crates/warp-core/src/causal_wal.rs index e8b57349..f5a24081 100644 --- a/crates/warp-core/src/causal_wal.rs +++ b/crates/warp-core/src/causal_wal.rs @@ -308,6 +308,8 @@ pub enum WalAppendAuthority { AdmissionKernel, /// Echo executable-operation interpreter and commit authority. ExecutionKernel, + /// Echo external-action request and settlement coordinator authority. + ExternalActionCoordinator, /// Recovery authority. Recovery, } @@ -333,6 +335,12 @@ pub enum WalTransactionKind { ExecutableOperationInstallation, /// Execution-kernel-owned commit of one executable operation consequence. ExecutableOperationTick, + /// Echo-owned admission of one external-action request before execution. + ExternalActionRequest, + /// Echo-owned claim of one admitted external-action request. + ExternalActionClaim, + /// Echo-owned admission of one external-action settlement. + ExternalActionSettlement, } impl WalTransactionKind { @@ -349,6 +357,9 @@ impl WalTransactionKind { Self::CausalAnchorAdmission => 7, Self::ExecutableOperationInstallation => 8, Self::ExecutableOperationTick => 9, + Self::ExternalActionRequest => 10, + Self::ExternalActionClaim => 11, + Self::ExternalActionSettlement => 12, } } @@ -367,10 +378,22 @@ impl WalTransactionKind { } Self::CausalAnchorAdmission => WalAppendAuthority::AdmissionKernel, Self::ExecutableOperationTick => WalAppendAuthority::ExecutionKernel, + Self::ExternalActionRequest + | Self::ExternalActionClaim + | Self::ExternalActionSettlement => WalAppendAuthority::ExternalActionCoordinator, Self::Checkpoint => WalAppendAuthority::Recovery, } } + pub(crate) const fn external_action_record_kind(self) -> Option { + match self { + Self::ExternalActionRequest => Some(WalRecordKind::ExternalActionRequestRecorded), + Self::ExternalActionClaim => Some(WalRecordKind::ExternalActionClaimRecorded), + Self::ExternalActionSettlement => Some(WalRecordKind::ExternalActionSettlementRecorded), + _ => None, + } + } + fn from_code(code: u8) -> Result { match code { 1 => Ok(Self::SubmissionIntake), @@ -382,6 +405,9 @@ impl WalTransactionKind { 7 => Ok(Self::CausalAnchorAdmission), 8 => Ok(Self::ExecutableOperationInstallation), 9 => Ok(Self::ExecutableOperationTick), + 10 => Ok(Self::ExternalActionRequest), + 11 => Ok(Self::ExternalActionClaim), + 12 => Ok(Self::ExternalActionSettlement), _ => Err(WalDecodeError::UnknownEnumCode { enum_name: "WalTransactionKind", code, @@ -450,6 +476,12 @@ pub enum WalRecordKind { /// Trusted scheduler retained one typed executable-operation Action /// outcome inside a scheduler-owned Tick. ExecutableOperationActionOutcomeRecorded, + /// Echo recorded a canonical external-action request before execution. + ExternalActionRequestRecorded, + /// Echo recorded one bounded external-action claim. + ExternalActionClaimRecorded, + /// Echo admitted one schema-bound external-action settlement. + ExternalActionSettlementRecorded, } impl WalRecordKind { @@ -486,6 +518,9 @@ impl WalRecordKind { Self::ExecutableOperationActionOutcomeRecorded => { "ExecutableOperationActionOutcomeRecorded" } + Self::ExternalActionRequestRecorded => "ExternalActionRequestRecorded", + Self::ExternalActionClaimRecorded => "ExternalActionClaimRecorded", + Self::ExternalActionSettlementRecorded => "ExternalActionSettlementRecorded", } } @@ -520,6 +555,11 @@ impl WalRecordKind { Self::CausalAnchorFactRecorded | Self::CausalAnchorAdmissionReceiptRecorded => { WalAppendAuthority::AdmissionKernel } + Self::ExternalActionRequestRecorded + | Self::ExternalActionClaimRecorded + | Self::ExternalActionSettlementRecorded => { + WalAppendAuthority::ExternalActionCoordinator + } Self::CheckpointPublicationRecorded | Self::RecoveryPostureRecorded => { WalAppendAuthority::Recovery } @@ -567,6 +607,9 @@ impl WalRecordKind { Self::ExecutableOperationExecutionRecorded => 26, Self::ExecutableOperationStateDeltaRecorded => 27, Self::ExecutableOperationActionOutcomeRecorded => 28, + Self::ExternalActionRequestRecorded => 29, + Self::ExternalActionClaimRecorded => 30, + Self::ExternalActionSettlementRecorded => 31, } } @@ -600,6 +643,9 @@ impl WalRecordKind { 26 => Ok(Self::ExecutableOperationExecutionRecorded), 27 => Ok(Self::ExecutableOperationStateDeltaRecorded), 28 => Ok(Self::ExecutableOperationActionOutcomeRecorded), + 29 => Ok(Self::ExternalActionRequestRecorded), + 30 => Ok(Self::ExternalActionClaimRecorded), + 31 => Ok(Self::ExternalActionSettlementRecorded), _ => Err(WalDecodeError::UnknownEnumCode { enum_name: "WalRecordKind", code, @@ -731,6 +777,8 @@ pub enum AffectedFrontierKind { ExecutableOperationCatalog, /// Typed executable-operation receipt frontier. ExecutableOperationReceiptIndex, + /// Durable external-action lifecycle frontier. + ExternalActionIndex, } impl AffectedFrontierKind { @@ -748,6 +796,7 @@ impl AffectedFrontierKind { Self::CausalAnchorIndex => 8, Self::ExecutableOperationCatalog => 9, Self::ExecutableOperationReceiptIndex => 10, + Self::ExternalActionIndex => 11, } } @@ -1028,6 +1077,7 @@ pub struct WalCommittedTransaction { /// Commit marker. pub commit: WalTransactionCommit, admission_kernel_capability: Option, + external_action_coordinator_capability: Option, } impl WalCommittedTransaction { @@ -1037,6 +1087,10 @@ impl WalCommittedTransaction { self.commit.transaction_kind, self.admission_kernel_capability, )?; + validate_external_action_coordinator_capability( + self.commit.transaction_kind, + self.external_action_coordinator_capability, + )?; validate_transaction_frames(&self.frames, &self.commit)?; validate_transaction_semantics(&self.frames, self.commit.transaction_kind)?; validate_transaction_frontiers(&self.affected_frontiers, self.commit.transaction_kind)?; @@ -1049,11 +1103,31 @@ impl WalCommittedTransaction { } Ok(()) } + + pub(crate) const fn external_action_coordinator_capability( + &self, + ) -> Option { + self.external_action_coordinator_capability + } } #[derive(Clone, Copy, Debug, PartialEq, Eq)] struct AdmissionKernelCapability; +/// Opaque authority required to flush an external-action lifecycle commit. +/// +/// Only Echo's external-action coordinator can construct this value. WAL +/// stores receive it solely to distinguish coordinator-owned commits from raw +/// caller-authored commit markers. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct ExternalActionCoordinatorCapability(()); + +impl ExternalActionCoordinatorCapability { + const fn new() -> Self { + Self(()) + } +} + fn validate_admission_kernel_capability( transaction_kind: WalTransactionKind, capability: Option, @@ -1064,6 +1138,16 @@ fn validate_admission_kernel_capability( Ok(()) } +fn validate_external_action_coordinator_capability( + transaction_kind: WalTransactionKind, + capability: Option, +) -> Result<(), WalValidationError> { + if transaction_kind.external_action_record_kind().is_some() && capability.is_none() { + return Err(WalValidationError::ExternalActionCoordinatorCapabilityRequired); + } + Ok(()) +} + /// Builder for a contiguous WAL transaction. #[derive(Clone, Debug)] pub struct WalTransactionBuilder { @@ -1085,6 +1169,7 @@ pub struct WalTransactionBuilder { frames: Vec, closed: bool, admission_kernel_capability: Option, + external_action_coordinator_capability: Option, } impl WalTransactionBuilder { @@ -1106,7 +1191,7 @@ impl WalTransactionBuilder { canonical_encoding_version: u16, digest_domain: Hash, ) -> Self { - Self::new_with_admission_kernel_capability( + Self::new_with_capabilities( writer_epoch, segment_id, transaction_id, @@ -1122,6 +1207,7 @@ impl WalTransactionBuilder { canonical_encoding_version, digest_domain, None, + None, ) } @@ -1145,7 +1231,7 @@ impl WalTransactionBuilder { canonical_encoding_version: u16, digest_domain: Hash, ) -> Self { - Self::new_with_admission_kernel_capability( + Self::new_with_capabilities( writer_epoch, segment_id, transaction_id, @@ -1161,11 +1247,12 @@ impl WalTransactionBuilder { canonical_encoding_version, digest_domain, Some(AdmissionKernelCapability), + None, ) } #[allow(clippy::too_many_arguments)] - fn new_with_admission_kernel_capability( + fn new_with_capabilities( writer_epoch: WriterEpochId, segment_id: WalSegmentId, transaction_id: WalTransactionId, @@ -1181,6 +1268,7 @@ impl WalTransactionBuilder { canonical_encoding_version: u16, digest_domain: Hash, admission_kernel_capability: Option, + external_action_coordinator_capability: Option, ) -> Self { Self { writer_epoch, @@ -1201,9 +1289,47 @@ impl WalTransactionBuilder { frames: Vec::new(), closed: false, admission_kernel_capability, + external_action_coordinator_capability, } } + #[allow(clippy::too_many_arguments)] + pub(crate) fn new_external_action( + writer_epoch: WriterEpochId, + segment_id: WalSegmentId, + transaction_id: WalTransactionId, + transaction_kind: WalTransactionKind, + first_lsn: Lsn, + previous_frame_digest: Hash, + previous_committed_transaction_digest: Hash, + durability_mode: WalDurabilityMode, + payload_codec_id: PayloadCodecId, + payload_schema_id: PayloadSchemaId, + payload_schema_version: u16, + canonical_encoding_version: u16, + digest_domain: Hash, + ) -> Self { + debug_assert!(transaction_kind.external_action_record_kind().is_some()); + Self::new_with_capabilities( + writer_epoch, + segment_id, + transaction_id, + transaction_kind, + WalAppendAuthority::ExternalActionCoordinator, + first_lsn, + previous_frame_digest, + previous_committed_transaction_digest, + durability_mode, + payload_codec_id, + payload_schema_id, + payload_schema_version, + canonical_encoding_version, + digest_domain, + None, + Some(ExternalActionCoordinatorCapability::new()), + ) + } + /// Appends a record to the transaction. pub fn push_record( &mut self, @@ -1218,6 +1344,11 @@ impl WalTransactionBuilder { { return Err(WalBuildError::AdmissionKernelCapabilityRequired); } + if kind.required_authority() == WalAppendAuthority::ExternalActionCoordinator + && self.external_action_coordinator_capability.is_none() + { + return Err(WalBuildError::ExternalActionCoordinatorCapabilityRequired); + } if kind.required_authority() != self.authority { return Err(WalBuildError::WrongAppendAuthority { record_kind: kind, @@ -1301,6 +1432,7 @@ impl WalTransactionBuilder { affected_frontiers, commit, admission_kernel_capability: self.admission_kernel_capability, + external_action_coordinator_capability: self.external_action_coordinator_capability, }; transaction.validate()?; Ok(transaction) @@ -1333,12 +1465,31 @@ pub trait WalStorePort { commit: WalTransactionCommit, ) -> Result<(), WalStoreError>; + /// Flushes one Echo-coordinator-owned external-action commit marker. + /// + /// The opaque capability cannot be constructed by callers of the raw WAL + /// surface. + fn flush_external_action_commit( + &mut self, + epoch_id: WriterEpochId, + commit: WalTransactionCommit, + capability: ExternalActionCoordinatorCapability, + ) -> Result<(), WalStoreError>; + /// Reads the recorded frames. fn read_frames(&self) -> Vec; /// Reads the flushed commit markers. fn read_commits(&self) -> Vec; + /// Reads one fallible, causally coherent WAL snapshot. + fn read_snapshot(&self) -> Result { + Ok(WalStoreSnapshot { + frames: self.read_frames(), + commits: self.read_commits(), + }) + } + /// Seals a segment. fn seal_segment( &mut self, @@ -1360,6 +1511,15 @@ pub trait WalStorePort { fn close_epoch(&mut self, epoch_id: WriterEpochId) -> Result<(), WalStoreError>; } +/// One fallible WAL storage snapshot used for trusted local recovery. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct WalStoreSnapshot { + /// Recorded frames in storage order. + pub frames: Vec, + /// Flushed commit markers in storage order. + pub commits: Vec, +} + /// Writer epoch acquisition request. #[derive(Clone, Debug, PartialEq, Eq)] pub struct WriterEpochRequest { @@ -1765,18 +1925,26 @@ impl InMemoryWalStore { ) -> Result<(), WalStoreError> { transaction.validate()?; let admission_kernel_capability = transaction.admission_kernel_capability; + let external_action_coordinator_capability = + transaction.external_action_coordinator_capability; let epoch_id = transaction.commit.writer_epoch; for frame in transaction.frames { self.append_frame(epoch_id, frame)?; } - self.flush_commit_with_capability(epoch_id, transaction.commit, admission_kernel_capability) + self.flush_commit_with_capabilities( + epoch_id, + transaction.commit, + admission_kernel_capability, + external_action_coordinator_capability, + ) } - fn flush_commit_with_capability( + fn flush_commit_with_capabilities( &mut self, epoch_id: WriterEpochId, commit: WalTransactionCommit, admission_kernel_capability: Option, + external_action_coordinator_capability: Option, ) -> Result<(), WalStoreError> { let active_epoch = self .active_epoch @@ -1786,6 +1954,10 @@ impl InMemoryWalStore { return Err(WalStoreError::WriterEpochMismatch); } validate_admission_kernel_capability(commit.transaction_kind, admission_kernel_capability)?; + validate_external_action_coordinator_capability( + commit.transaction_kind, + external_action_coordinator_capability, + )?; self.epoch_closures.insert( epoch_id, WriterEpochClosure { @@ -1810,6 +1982,12 @@ impl InMemoryWalStore { pub fn manifests(&self) -> &[WalManifest] { &self.manifests } + + /// Returns the number of durably flushed commit markers. + #[must_use] + pub const fn commit_count(&self) -> usize { + self.commits.len() + } } impl WalStorePort for InMemoryWalStore { @@ -1849,7 +2027,16 @@ impl WalStorePort for InMemoryWalStore { epoch_id: WriterEpochId, commit: WalTransactionCommit, ) -> Result<(), WalStoreError> { - self.flush_commit_with_capability(epoch_id, commit, None) + self.flush_commit_with_capabilities(epoch_id, commit, None, None) + } + + fn flush_external_action_commit( + &mut self, + epoch_id: WriterEpochId, + commit: WalTransactionCommit, + capability: ExternalActionCoordinatorCapability, + ) -> Result<(), WalStoreError> { + self.flush_commit_with_capabilities(epoch_id, commit, None, Some(capability)) } fn read_frames(&self) -> Vec { @@ -5471,18 +5658,26 @@ impl FilesystemWalStore { ) -> Result<(), WalStoreError> { transaction.validate()?; let admission_kernel_capability = transaction.admission_kernel_capability; + let external_action_coordinator_capability = + transaction.external_action_coordinator_capability; let epoch_id = transaction.commit.writer_epoch; for frame in transaction.frames { self.append_frame(epoch_id, frame)?; } - self.flush_commit_with_capability(epoch_id, transaction.commit, admission_kernel_capability) + self.flush_commit_with_capabilities( + epoch_id, + transaction.commit, + admission_kernel_capability, + external_action_coordinator_capability, + ) } - fn flush_commit_with_capability( + fn flush_commit_with_capabilities( &mut self, epoch_id: WriterEpochId, commit: WalTransactionCommit, admission_kernel_capability: Option, + external_action_coordinator_capability: Option, ) -> Result<(), WalStoreError> { let active_epoch = self .active_epoch @@ -5492,6 +5687,10 @@ impl FilesystemWalStore { return Err(WalStoreError::WriterEpochMismatch); } validate_admission_kernel_capability(commit.transaction_kind, admission_kernel_capability)?; + validate_external_action_coordinator_capability( + commit.transaction_kind, + external_action_coordinator_capability, + )?; #[cfg(any(test, feature = "host_test"))] if self .fault_plan @@ -5622,7 +5821,16 @@ impl WalStorePort for FilesystemWalStore { epoch_id: WriterEpochId, commit: WalTransactionCommit, ) -> Result<(), WalStoreError> { - self.flush_commit_with_capability(epoch_id, commit, None) + self.flush_commit_with_capabilities(epoch_id, commit, None, None) + } + + fn flush_external_action_commit( + &mut self, + epoch_id: WriterEpochId, + commit: WalTransactionCommit, + capability: ExternalActionCoordinatorCapability, + ) -> Result<(), WalStoreError> { + self.flush_commit_with_capabilities(epoch_id, commit, None, Some(capability)) } fn read_frames(&self) -> Vec { @@ -5639,6 +5847,11 @@ impl WalStorePort for FilesystemWalStore { } } + fn read_snapshot(&self) -> Result { + let (frames, commits, _) = read_filesystem_segments(&self.root)?; + Ok(WalStoreSnapshot { frames, commits }) + } + fn seal_segment( &mut self, epoch_id: WriterEpochId, @@ -8983,6 +9196,9 @@ pub enum WalBuildError { /// Public builders do not carry Echo's causal-anchor admission capability. #[error("Echo admission-kernel capability is required")] AdmissionKernelCapabilityRequired, + /// Public builders do not carry Echo's external-action coordinator capability. + #[error("Echo external-action coordinator capability is required")] + ExternalActionCoordinatorCapabilityRequired, /// Record requires a different append authority. #[error( "wrong append authority for {record_kind:?}: required {required:?}, actual {actual:?}" @@ -9039,6 +9255,9 @@ pub enum WalValidationError { /// A causal-anchor transaction lacks Echo's private admission capability. #[error("WAL causal-anchor transaction lacks Echo admission-kernel capability")] AdmissionKernelCapabilityRequired, + /// An external-action transaction lacks Echo's private coordinator capability. + #[error("WAL external-action transaction lacks Echo coordinator capability")] + ExternalActionCoordinatorCapabilityRequired, /// Frame payload kind does not match header kind. #[error("WAL frame record kind mismatch")] RecordKindMismatch, @@ -9063,12 +9282,18 @@ pub enum WalValidationError { /// Executable-operation commit does not contain one receipt followed by one state delta. #[error("WAL executable-operation tick frame shape is invalid")] ExecutableOperationTickFrameShapeMismatch, + /// External-action transitions do not contain their one canonical record. + #[error("WAL external-action frame shape is invalid")] + ExternalActionFrameShapeMismatch, /// Executable-operation installation does not advance exactly its catalog frontier. #[error("WAL executable-operation installation frontier shape is invalid")] ExecutableOperationInstallationFrontierShapeMismatch, /// Executable-operation commit does not advance its receipt and runtime frontiers in order. #[error("WAL executable-operation tick frontier shape is invalid")] ExecutableOperationTickFrontierShapeMismatch, + /// External-action transitions do not advance exactly their lifecycle frontier. + #[error("WAL external-action frontier shape is invalid")] + ExternalActionFrontierShapeMismatch, /// Transaction contains no frames. #[error("WAL transaction contains no frames")] EmptyTransaction, @@ -9577,6 +9802,12 @@ fn validate_transaction_semantics( { return Err(WalValidationError::ExecutableOperationTickFrameShapeMismatch); } + if transaction_kind + .external_action_record_kind() + .is_some_and(|record_kind| frames.len() != 1 || frames[0].header.record_kind != record_kind) + { + return Err(WalValidationError::ExternalActionFrameShapeMismatch); + } Ok(()) } @@ -9597,6 +9828,15 @@ fn validate_transaction_frontiers( { return Err(WalValidationError::ExecutableOperationTickFrontierShapeMismatch); } + if matches!( + transaction_kind, + WalTransactionKind::ExternalActionRequest + | WalTransactionKind::ExternalActionClaim + | WalTransactionKind::ExternalActionSettlement + ) && (frontiers.len() != 1 || frontiers[0].kind != AffectedFrontierKind::ExternalActionIndex) + { + return Err(WalValidationError::ExternalActionFrontierShapeMismatch); + } for frontier in frontiers { if !frontier_kind_allowed_for_transaction(transaction_kind, frontier.kind) { return Err(WalValidationError::FrontierTransitionKindMismatch); @@ -9643,6 +9883,11 @@ fn frontier_kind_allowed_for_transaction( AffectedFrontierKind::RuntimeState | AffectedFrontierKind::ExecutableOperationReceiptIndex ), + WalTransactionKind::ExternalActionRequest + | WalTransactionKind::ExternalActionClaim + | WalTransactionKind::ExternalActionSettlement => { + matches!(frontier_kind, AffectedFrontierKind::ExternalActionIndex) + } } } diff --git a/crates/warp-core/src/evidence.rs b/crates/warp-core/src/evidence.rs index f3c816eb..8cc05f8c 100644 --- a/crates/warp-core/src/evidence.rs +++ b/crates/warp-core/src/evidence.rs @@ -371,6 +371,21 @@ mod tests { WalRecordKind::ExecutableOperationExecutionRecorded, AffectedFrontierKind::ExecutableOperationReceiptIndex, ), + WalTransactionKind::ExternalActionRequest => ( + WalAppendAuthority::ExternalActionCoordinator, + WalRecordKind::ExternalActionRequestRecorded, + AffectedFrontierKind::ExternalActionIndex, + ), + WalTransactionKind::ExternalActionClaim => ( + WalAppendAuthority::ExternalActionCoordinator, + WalRecordKind::ExternalActionClaimRecorded, + AffectedFrontierKind::ExternalActionIndex, + ), + WalTransactionKind::ExternalActionSettlement => ( + WalAppendAuthority::ExternalActionCoordinator, + WalRecordKind::ExternalActionSettlementRecorded, + AffectedFrontierKind::ExternalActionIndex, + ), } } diff --git a/crates/warp-core/src/external_action.rs b/crates/warp-core/src/external_action.rs new file mode 100644 index 00000000..cf60e92a --- /dev/null +++ b/crates/warp-core/src/external_action.rs @@ -0,0 +1,1838 @@ +// SPDX-License-Identifier: Apache-2.0 +// © James Ross Ω FLYING•ROBOTS +//! Durable, domain-neutral external-action request and settlement protocol. +//! +//! Edict-authored programs construct request values. Echo records each request +//! before an adapter may act, records one bounded claim, and admits one +//! schema-bound settlement before deterministic resumption. This module does +//! not execute external effects and grants no filesystem, process, network, or +//! model authority. + +use std::collections::{BTreeMap, BTreeSet}; +use std::sync::OnceLock; + +use thiserror::Error; + +use crate::causal_wal::{ + affected_frontiers_root, recover_from_frames_and_commits, AffectedFrontier, + AffectedFrontierKind, Lsn, PayloadCodecId, PayloadSchemaId, RecoveryAccessMode, + RecoveryScanReport, RecoveryTailPosture, WalBuildError, WalCommittedTransaction, + WalDecodeError, WalDurabilityMode, WalRecordKind, WalRecoveryError, WalSegmentId, + WalStoreError, WalStorePort, WalTransactionBuilder, WalTransactionId, WalTransactionKind, + WriterEpochId, +}; +use crate::{Hash, WorldlineId}; + +const REQUEST_ID_DOMAIN: &[u8] = b"echo:external-action:request-id:v1\0"; +const ATTEMPT_ID_DOMAIN: &[u8] = b"echo:external-action:attempt-id:v1\0"; +const IDEMPOTENCY_KEY_DOMAIN: &[u8] = b"echo:external-action:idempotency-key:v1\0"; +const ADAPTER_REGISTRY_ID_DOMAIN: &[u8] = b"echo:external-action:adapter-registry-id:v1\0"; +const INDEX_EMPTY_LEAF_DOMAIN: &[u8] = b"echo:external-action:index-empty-leaf:v1\0"; +const INDEX_LEAF_DOMAIN: &[u8] = b"echo:external-action:index-leaf:v1\0"; +const INDEX_NODE_DOMAIN: &[u8] = b"echo:external-action:index-node:v1\0"; +const REQUEST_PAYLOAD_MAGIC: &[u8; 4] = b"EAR1"; +const CLAIM_PAYLOAD_MAGIC: &[u8; 4] = b"EAC1"; +const SETTLEMENT_PAYLOAD_MAGIC: &[u8; 4] = b"EAS1"; + +/// Absolute v1 ceiling for settlement bytes retained directly in the WAL. +pub const MAX_EXTERNAL_ACTION_SETTLEMENT_BYTES_V1: u64 = 1_048_576; + +/// Non-causal transaction metadata supplied to the external-action coordinator. +/// +/// LSN and predecessor coordinates are intentionally absent: the coordinator +/// derives them from its checked local WAL continuation. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct ExternalActionTransactionContextV1 { + /// Active writer epoch. + pub writer_epoch: WriterEpochId, + /// Active WAL segment. + pub segment_id: WalSegmentId, + /// Identity of this lifecycle transaction. + pub transaction_id: WalTransactionId, + /// Required durability mode. + pub durability_mode: WalDurabilityMode, + /// Canonical payload codec. + pub payload_codec_id: PayloadCodecId, + /// Canonical payload schema. + pub payload_schema_id: PayloadSchemaId, + /// Payload schema version. + pub payload_schema_version: u16, + /// Canonical encoding version. + pub canonical_encoding_version: u16, + /// Digest domain for WAL framing. + pub digest_domain: Hash, +} + +/// Stable identity of one external operation family. +#[repr(transparent)] +#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)] +pub struct ExternalActionOperationIdV1(Hash); + +impl ExternalActionOperationIdV1 { + /// Reconstructs an operation identity from its canonical digest. + #[must_use] + pub const fn from_hash(hash: Hash) -> Self { + Self(hash) + } + + /// Returns the canonical digest. + #[must_use] + pub const fn as_hash(self) -> Hash { + self.0 + } +} + +/// Stable identity of one authorized external adapter. +#[repr(transparent)] +#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)] +pub struct ExternalActionAdapterIdV1(Hash); + +impl ExternalActionAdapterIdV1 { + /// Reconstructs an adapter identity from its canonical digest. + #[must_use] + pub const fn from_hash(hash: Hash) -> Self { + Self(hash) + } + + /// Returns the canonical digest. + #[must_use] + pub const fn as_hash(self) -> Hash { + self.0 + } +} + +/// Stable identity of one canonical external-action request. +#[repr(transparent)] +#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)] +pub struct ExternalActionRequestIdV1(Hash); + +impl ExternalActionRequestIdV1 { + /// Returns the canonical digest. + #[must_use] + pub const fn as_hash(self) -> Hash { + self.0 + } + + const fn from_hash(hash: Hash) -> Self { + Self(hash) + } +} + +/// Stable identity of one bounded adapter attempt. +#[repr(transparent)] +#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)] +pub struct ExternalActionAttemptIdV1(Hash); + +impl ExternalActionAttemptIdV1 { + /// Reconstructs an attempt identity from its canonical digest. + #[must_use] + pub const fn from_hash(hash: Hash) -> Self { + Self(hash) + } + + /// Returns the canonical digest. + #[must_use] + pub const fn as_hash(self) -> Hash { + self.0 + } +} + +/// Bounds delegated to one external-action request. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct ExternalActionBudgetV1 { + /// Maximum canonical settlement bytes retained in the WAL. + pub max_settlement_bytes: u64, + /// Maximum number of adapter attempts authorized for this request. + pub max_attempts: u32, +} + +/// Deterministic request emitted by an Edict-authored program. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct ExternalActionRequestV1 { + request_id: ExternalActionRequestIdV1, + /// Worldline whose admitted history produced the request. + pub worldline_id: WorldlineId, + /// Declared external operation family. + pub operation_id: ExternalActionOperationIdV1, + /// Schema digest for the canonical request input. + pub input_schema_digest: Hash, + /// Schema digest required for the canonical settlement. + pub settlement_schema_digest: Hash, + /// Requested authority scope. + pub authority_scope_digest: Hash, + /// Exact current-world basis. + pub basis_digest: Hash, + /// Delegated execution and settlement bounds. + pub budget: ExternalActionBudgetV1, + /// Digest of the canonical operation input. + pub input_digest: Hash, + /// Named reconciliation law for ambiguous outcomes. + pub reconciliation_law_digest: Hash, +} + +impl ExternalActionRequestV1 { + /// Constructs a canonical request and derives its identity. + #[allow(clippy::too_many_arguments)] + pub fn new( + worldline_id: WorldlineId, + operation_id: ExternalActionOperationIdV1, + input_schema_digest: Hash, + settlement_schema_digest: Hash, + authority_scope_digest: Hash, + basis_digest: Hash, + budget: ExternalActionBudgetV1, + input_digest: Hash, + reconciliation_law_digest: Hash, + ) -> Result { + if budget.max_settlement_bytes == 0 || budget.max_attempts == 0 { + return Err(ExternalActionProtocolErrorV1::EmptyBudget); + } + if budget.max_attempts != 1 { + return Err(ExternalActionProtocolErrorV1::UnsupportedAttemptBudget); + } + if budget.max_settlement_bytes > MAX_EXTERNAL_ACTION_SETTLEMENT_BYTES_V1 { + return Err(ExternalActionProtocolErrorV1::RequestBudgetLimitExceeded); + } + let mut request = Self { + request_id: ExternalActionRequestIdV1::from_hash([0; 32]), + worldline_id, + operation_id, + input_schema_digest, + settlement_schema_digest, + authority_scope_digest, + basis_digest, + budget, + input_digest, + reconciliation_law_digest, + }; + request.request_id = + ExternalActionRequestIdV1::from_hash(request.expected_request_id_digest()); + Ok(request) + } + + /// Returns the canonical request identity. + #[must_use] + pub const fn request_id(&self) -> ExternalActionRequestIdV1 { + self.request_id + } + + fn expected_request_id_digest(&self) -> Hash { + let mut hasher = blake3::Hasher::new(); + hasher.update(REQUEST_ID_DOMAIN); + hasher.update(self.worldline_id.as_bytes()); + hasher.update(&self.operation_id.as_hash()); + hasher.update(&self.input_schema_digest); + hasher.update(&self.settlement_schema_digest); + hasher.update(&self.authority_scope_digest); + hasher.update(&self.basis_digest); + hasher.update(&self.budget.max_settlement_bytes.to_le_bytes()); + hasher.update(&self.budget.max_attempts.to_le_bytes()); + hasher.update(&self.input_digest); + hasher.update(&self.reconciliation_law_digest); + hasher.finalize().into() + } + + fn validate_identity(&self) -> Result<(), ExternalActionProtocolErrorV1> { + if self.request_id.as_hash() != self.expected_request_id_digest() { + return Err(ExternalActionProtocolErrorV1::RequestIdentityMismatch); + } + if self.budget.max_settlement_bytes == 0 || self.budget.max_attempts == 0 { + return Err(ExternalActionProtocolErrorV1::EmptyBudget); + } + if self.budget.max_attempts != 1 { + return Err(ExternalActionProtocolErrorV1::UnsupportedAttemptBudget); + } + if self.budget.max_settlement_bytes > MAX_EXTERNAL_ACTION_SETTLEMENT_BYTES_V1 { + return Err(ExternalActionProtocolErrorV1::RequestBudgetLimitExceeded); + } + Ok(()) + } + + fn to_payload_bytes(self) -> Vec { + let mut out = Vec::with_capacity(4 + (9 * 32) + 12); + out.extend_from_slice(REQUEST_PAYLOAD_MAGIC); + out.extend_from_slice(&self.request_id.as_hash()); + out.extend_from_slice(self.worldline_id.as_bytes()); + out.extend_from_slice(&self.operation_id.as_hash()); + out.extend_from_slice(&self.input_schema_digest); + out.extend_from_slice(&self.settlement_schema_digest); + out.extend_from_slice(&self.authority_scope_digest); + out.extend_from_slice(&self.basis_digest); + out.extend_from_slice(&self.budget.max_settlement_bytes.to_le_bytes()); + out.extend_from_slice(&self.budget.max_attempts.to_le_bytes()); + out.extend_from_slice(&self.input_digest); + out.extend_from_slice(&self.reconciliation_law_digest); + out + } + + fn from_payload_bytes(bytes: &[u8]) -> Result { + let mut cursor = ExternalActionPayloadCursor::new(bytes); + cursor.expect_magic(REQUEST_PAYLOAD_MAGIC, "ExternalActionRequestV1")?; + let request = Self { + request_id: ExternalActionRequestIdV1::from_hash(cursor.read_hash()?), + worldline_id: WorldlineId::from_bytes(cursor.read_hash()?), + operation_id: ExternalActionOperationIdV1::from_hash(cursor.read_hash()?), + input_schema_digest: cursor.read_hash()?, + settlement_schema_digest: cursor.read_hash()?, + authority_scope_digest: cursor.read_hash()?, + basis_digest: cursor.read_hash()?, + budget: ExternalActionBudgetV1 { + max_settlement_bytes: cursor.read_u64()?, + max_attempts: cursor.read_u32()?, + }, + input_digest: cursor.read_hash()?, + reconciliation_law_digest: cursor.read_hash()?, + }; + cursor.finish()?; + request.validate_identity()?; + Ok(request) + } +} + +/// Runtime-owner binding that permits one adapter for one operation and scope. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct ExternalActionAdapterBindingV1 { + /// Adapter admitted by the runtime owner. + pub adapter_id: ExternalActionAdapterIdV1, + /// Operation family the adapter may perform. + pub operation_id: ExternalActionOperationIdV1, + /// Maximum authority scope admitted for the adapter. + pub authority_scope_digest: Hash, +} + +/// Runtime-owned adapter registry. +#[derive(Clone, Debug, Default, PartialEq, Eq)] +pub struct ExternalActionAdapterRegistryV1 { + bindings: BTreeMap<(ExternalActionOperationIdV1, Hash), BTreeSet>, +} + +impl ExternalActionAdapterRegistryV1 { + /// Builds a runtime-owned registry from explicit domain-specific bindings. + #[must_use] + pub fn new(bindings: impl IntoIterator) -> Self { + let mut registry = Self::default(); + for binding in bindings { + registry + .bindings + .entry((binding.operation_id, binding.authority_scope_digest)) + .or_default() + .insert(binding.adapter_id); + } + registry + } + + /// Attenuates runtime-owner policy into one request-specific authorization. + pub fn authorize( + &self, + request: &ExternalActionRequestV1, + adapter_id: ExternalActionAdapterIdV1, + ) -> Result { + let admitted = self + .bindings + .get(&(request.operation_id, request.authority_scope_digest)) + .is_some_and(|adapters| adapters.contains(&adapter_id)); + if !admitted { + return Err(ExternalActionProtocolErrorV1::UnauthorizedAdapter); + } + Ok(ExternalActionAdapterAuthorizationV1 { + adapter_id, + operation_id: request.operation_id, + authority_scope_digest: request.authority_scope_digest, + request_id: request.request_id, + basis_digest: request.basis_digest, + registry_policy_digest: self.identity_digest(), + }) + } + + /// Returns the canonical identity of the complete runtime-owned registry. + #[must_use] + pub fn identity_digest(&self) -> Hash { + let binding_count = self + .bindings + .values() + .map(BTreeSet::len) + .fold(0_u64, |count, len| { + count.saturating_add(u64::try_from(len).unwrap_or(u64::MAX)) + }); + let mut hasher = blake3::Hasher::new(); + hasher.update(ADAPTER_REGISTRY_ID_DOMAIN); + hasher.update(&binding_count.to_le_bytes()); + for ((operation_id, authority_scope_digest), adapters) in &self.bindings { + for adapter_id in adapters { + hasher.update(&operation_id.as_hash()); + hasher.update(authority_scope_digest); + hasher.update(&adapter_id.as_hash()); + } + } + hasher.finalize().into() + } +} + +/// Request-specific authorization attenuated from the runtime-owned registry. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct ExternalActionAdapterAuthorizationV1 { + adapter_id: ExternalActionAdapterIdV1, + operation_id: ExternalActionOperationIdV1, + authority_scope_digest: Hash, + request_id: ExternalActionRequestIdV1, + basis_digest: Hash, + registry_policy_digest: Hash, +} + +/// One durably recorded adapter claim. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct ExternalActionClaimV1 { + /// Request being claimed. + pub request_id: ExternalActionRequestIdV1, + /// Stable attempt identity. + pub attempt_id: ExternalActionAttemptIdV1, + /// Zero-based attempt ordinal. + pub attempt_ordinal: u32, + /// Authorized adapter. + pub adapter_id: ExternalActionAdapterIdV1, + /// Lease or fencing evidence. + pub lease_evidence_digest: Hash, + /// Request-stable idempotency key. + pub idempotency_key: Hash, + /// Named reconciliation law copied from the request. + pub reconciliation_law_digest: Hash, + /// Exact basis copied from the request. + pub basis_digest: Hash, + /// Runtime-owned registry policy that admitted this exact request. + pub authorization_policy_digest: Hash, +} + +impl ExternalActionClaimV1 { + fn for_request( + request: &ExternalActionRequestV1, + adapter_id: ExternalActionAdapterIdV1, + attempt_ordinal: u32, + lease_evidence_digest: Hash, + authorization_policy_digest: Hash, + ) -> Self { + let idempotency_key = external_action_idempotency_key(request); + let attempt_id = external_action_attempt_id( + request.request_id, + attempt_ordinal, + adapter_id, + lease_evidence_digest, + authorization_policy_digest, + ); + Self { + request_id: request.request_id, + attempt_id, + attempt_ordinal, + adapter_id, + lease_evidence_digest, + idempotency_key, + reconciliation_law_digest: request.reconciliation_law_digest, + basis_digest: request.basis_digest, + authorization_policy_digest, + } + } + + fn to_payload_bytes(self) -> Vec { + let mut out = Vec::with_capacity(4 + (8 * 32) + 4); + out.extend_from_slice(CLAIM_PAYLOAD_MAGIC); + out.extend_from_slice(&self.request_id.as_hash()); + out.extend_from_slice(&self.attempt_id.as_hash()); + out.extend_from_slice(&self.attempt_ordinal.to_le_bytes()); + out.extend_from_slice(&self.adapter_id.as_hash()); + out.extend_from_slice(&self.lease_evidence_digest); + out.extend_from_slice(&self.idempotency_key); + out.extend_from_slice(&self.reconciliation_law_digest); + out.extend_from_slice(&self.basis_digest); + out.extend_from_slice(&self.authorization_policy_digest); + out + } + + fn from_payload_bytes(bytes: &[u8]) -> Result { + let mut cursor = ExternalActionPayloadCursor::new(bytes); + cursor.expect_magic(CLAIM_PAYLOAD_MAGIC, "ExternalActionClaimV1")?; + let claim = Self { + request_id: ExternalActionRequestIdV1::from_hash(cursor.read_hash()?), + attempt_id: ExternalActionAttemptIdV1::from_hash(cursor.read_hash()?), + attempt_ordinal: cursor.read_u32()?, + adapter_id: ExternalActionAdapterIdV1::from_hash(cursor.read_hash()?), + lease_evidence_digest: cursor.read_hash()?, + idempotency_key: cursor.read_hash()?, + reconciliation_law_digest: cursor.read_hash()?, + basis_digest: cursor.read_hash()?, + authorization_policy_digest: cursor.read_hash()?, + }; + cursor.finish()?; + Ok(claim) + } +} + +/// Typed terminal observation supplied by an external adapter. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum ExternalActionSettlementKindV1 { + /// The requested external postcondition was established. + Succeeded, + /// The external system rejected the request as a typed refusal. + Rejected, + /// The adapter established that execution failed. + Failed, + /// The adapter cannot establish whether the external effect occurred. + OutcomeUnknown, +} + +impl ExternalActionSettlementKindV1 { + /// Returns the stable code retained in the WAL payload. + #[must_use] + pub const fn stable_code(self) -> u8 { + match self { + Self::Succeeded => 1, + Self::Rejected => 2, + Self::Failed => 3, + Self::OutcomeUnknown => 4, + } + } + + fn from_stable_code(code: u8) -> Result { + match code { + 1 => Ok(Self::Succeeded), + 2 => Ok(Self::Rejected), + 3 => Ok(Self::Failed), + 4 => Ok(Self::OutcomeUnknown), + _ => Err(WalDecodeError::UnknownEnumCode { + enum_name: "ExternalActionSettlementKindV1", + code, + }), + } + } +} + +/// Untrusted settlement candidate submitted for Echo admission. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct ExternalActionSettlementCandidateV1 { + /// Request being settled. + pub request_id: ExternalActionRequestIdV1, + /// Attempt being settled. + pub attempt_id: ExternalActionAttemptIdV1, + /// Adapter claiming the result. + pub adapter_id: ExternalActionAdapterIdV1, + /// Typed terminal outcome. + pub kind: ExternalActionSettlementKindV1, + /// Claimed settlement schema. + pub settlement_schema_digest: Hash, + /// Exact request basis. + pub basis_digest: Hash, + /// Canonical result bytes retained for replay. + pub canonical_result_bytes: Vec, + /// Claimed digest of the canonical result bytes. + pub declared_result_digest: Hash, + /// Evidence for schema admission. + pub schema_admission_evidence_digest: Hash, + /// Adapter-supplied external evidence digest. + pub external_evidence_digest: Hash, +} + +impl ExternalActionSettlementCandidateV1 { + /// Builds a candidate with a correctly declared result digest. + #[allow(clippy::too_many_arguments)] + #[must_use] + pub fn new( + request_id: ExternalActionRequestIdV1, + attempt_id: ExternalActionAttemptIdV1, + adapter_id: ExternalActionAdapterIdV1, + kind: ExternalActionSettlementKindV1, + settlement_schema_digest: Hash, + basis_digest: Hash, + canonical_result_bytes: Vec, + schema_admission_evidence_digest: Hash, + external_evidence_digest: Hash, + ) -> Self { + let declared_result_digest = blake3::hash(&canonical_result_bytes).into(); + Self { + request_id, + attempt_id, + adapter_id, + kind, + settlement_schema_digest, + basis_digest, + canonical_result_bytes, + declared_result_digest, + schema_admission_evidence_digest, + external_evidence_digest, + } + } +} + +/// Admitted settlement retained in Echo history. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct ExternalActionSettlementV1 { + /// Request being settled. + pub request_id: ExternalActionRequestIdV1, + /// Attempt being settled. + pub attempt_id: ExternalActionAttemptIdV1, + /// Authorized adapter identity. + pub adapter_id: ExternalActionAdapterIdV1, + /// Typed terminal outcome. + pub kind: ExternalActionSettlementKindV1, + /// Admitted settlement schema. + pub settlement_schema_digest: Hash, + /// Exact request basis. + pub basis_digest: Hash, + /// Canonical result bytes retained for replay. + pub canonical_result_bytes: Vec, + /// Digest of the canonical result bytes. + pub result_digest: Hash, + /// Evidence for schema admission. + pub schema_admission_evidence_digest: Hash, + /// Adapter-supplied external evidence digest. + pub external_evidence_digest: Hash, +} + +impl ExternalActionSettlementV1 { + fn from_candidate(candidate: ExternalActionSettlementCandidateV1) -> Self { + Self { + request_id: candidate.request_id, + attempt_id: candidate.attempt_id, + adapter_id: candidate.adapter_id, + kind: candidate.kind, + settlement_schema_digest: candidate.settlement_schema_digest, + basis_digest: candidate.basis_digest, + canonical_result_bytes: candidate.canonical_result_bytes, + result_digest: candidate.declared_result_digest, + schema_admission_evidence_digest: candidate.schema_admission_evidence_digest, + external_evidence_digest: candidate.external_evidence_digest, + } + } + + fn to_payload_bytes(&self) -> Vec { + let mut out = Vec::with_capacity(4 + 1 + (8 * 32) + 8 + self.canonical_result_bytes.len()); + out.extend_from_slice(SETTLEMENT_PAYLOAD_MAGIC); + out.extend_from_slice(&self.request_id.as_hash()); + out.extend_from_slice(&self.attempt_id.as_hash()); + out.extend_from_slice(&self.adapter_id.as_hash()); + out.push(self.kind.stable_code()); + out.extend_from_slice(&self.settlement_schema_digest); + out.extend_from_slice(&self.basis_digest); + out.extend_from_slice( + &u64::try_from(self.canonical_result_bytes.len()) + .unwrap_or(u64::MAX) + .to_le_bytes(), + ); + out.extend_from_slice(&self.canonical_result_bytes); + out.extend_from_slice(&self.result_digest); + out.extend_from_slice(&self.schema_admission_evidence_digest); + out.extend_from_slice(&self.external_evidence_digest); + out + } + + fn from_payload_bytes(bytes: &[u8]) -> Result { + let mut cursor = ExternalActionPayloadCursor::new(bytes); + cursor.expect_magic(SETTLEMENT_PAYLOAD_MAGIC, "ExternalActionSettlementV1")?; + let request_id = ExternalActionRequestIdV1::from_hash(cursor.read_hash()?); + let attempt_id = ExternalActionAttemptIdV1::from_hash(cursor.read_hash()?); + let adapter_id = ExternalActionAdapterIdV1::from_hash(cursor.read_hash()?); + let kind = ExternalActionSettlementKindV1::from_stable_code(cursor.read_u8()?)?; + let settlement_schema_digest = cursor.read_hash()?; + let basis_digest = cursor.read_hash()?; + let result_len = cursor.read_u64()?; + if result_len > MAX_EXTERNAL_ACTION_SETTLEMENT_BYTES_V1 { + return Err(ExternalActionProtocolErrorV1::SettlementBudgetExceeded); + } + let result_len = usize::try_from(result_len) + .map_err(|_| ExternalActionProtocolErrorV1::SettlementBudgetExceeded)?; + let canonical_result_bytes = cursor.read_bytes(result_len)?.to_vec(); + let settlement = Self { + request_id, + attempt_id, + adapter_id, + kind, + settlement_schema_digest, + basis_digest, + canonical_result_bytes, + result_digest: cursor.read_hash()?, + schema_admission_evidence_digest: cursor.read_hash()?, + external_evidence_digest: cursor.read_hash()?, + }; + cursor.finish()?; + if Hash::from(blake3::hash(&settlement.canonical_result_bytes)) != settlement.result_digest + { + return Err(ExternalActionProtocolErrorV1::SettlementResultDigestMismatch); + } + Ok(settlement) + } +} + +/// Proof that a request was committed before adapter execution became reachable. +#[derive(Debug, PartialEq, Eq)] +pub struct DurablyRecordedExternalActionRequestV1 { + request: ExternalActionRequestV1, + request_commit_digest: Hash, +} + +impl DurablyRecordedExternalActionRequestV1 { + /// Returns the recorded request. + #[must_use] + pub const fn request(&self) -> ExternalActionRequestV1 { + self.request + } + + /// Returns the WAL commit that made the request durable. + #[must_use] + pub const fn request_commit_digest(&self) -> Hash { + self.request_commit_digest + } +} + +/// Adapter work grant returned only after the claim transaction is durable. +#[derive(Debug, PartialEq, Eq)] +pub struct ExternalActionClaimGrantV1 { + request: ExternalActionRequestV1, + claim: ExternalActionClaimV1, + claim_commit_digest: Hash, +} + +impl ExternalActionClaimGrantV1 { + /// Returns the exact request an adapter may attempt. + #[must_use] + pub const fn request(&self) -> ExternalActionRequestV1 { + self.request + } + + /// Returns the exact attempt claim. + #[must_use] + pub const fn claim(&self) -> ExternalActionClaimV1 { + self.claim + } + + /// Returns the WAL commit that made the claim durable. + #[must_use] + pub const fn claim_commit_digest(&self) -> Hash { + self.claim_commit_digest + } +} + +/// Settlement value exposed to deterministic execution only after WAL commit. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct AdmittedExternalActionSettlementV1 { + settlement: ExternalActionSettlementV1, + settlement_commit_digest: Hash, +} + +impl AdmittedExternalActionSettlementV1 { + /// Returns the admitted settlement fact. + #[must_use] + pub const fn settlement(&self) -> &ExternalActionSettlementV1 { + &self.settlement + } + + /// Returns the WAL commit that made resumption lawful. + #[must_use] + pub const fn settlement_commit_digest(&self) -> Hash { + self.settlement_commit_digest + } +} + +/// Recovered lifecycle posture for one external request. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum RecoveredExternalActionPostureV1 { + /// Request is durable and has not been claimed. + Requested, + /// Claim is durable; recovery must reconcile rather than reissue. + Claimed, + /// Settlement is durable and replayable. + Settled(ExternalActionSettlementKindV1), +} + +/// Observation-only lifecycle reconstructed from a supplied recovery report. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct RecoveredExternalActionV1 { + /// Canonical request. + pub request: ExternalActionRequestV1, + /// Commit that durably admitted the request. + pub request_commit_digest: Hash, + /// Recorded claim, when present. + pub claim: Option, + /// Commit that durably admitted the claim, when present. + pub claim_commit_digest: Option, + /// Admitted settlement, when present. + pub settlement: Option, + /// Commit that durably admitted the settlement, when present. + pub settlement_commit_digest: Option, + /// Lifecycle posture. + pub posture: RecoveredExternalActionPostureV1, +} + +/// Observation-only external-action lifecycle index. +#[derive(Clone, Debug, Default, PartialEq, Eq)] +pub struct RecoveredExternalActionIndexV1 { + entries: BTreeMap, + merkle_nodes: BTreeMap, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)] +struct ExternalActionIndexNodeKeyV1 { + depth: u16, + prefix: Hash, +} + +/// Trusted local coordinator state recovered from one fallible WAL snapshot. +/// +/// Arbitrary recovery reports expose observation-only lifecycle values. +/// Transition grants and resumable settlement facts can be reconstructed only +/// through this locally recovered coordinator. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct ExternalActionCoordinatorV1 { + index: RecoveredExternalActionIndexV1, + next_lsn: Lsn, + previous_frame_digest: Hash, + previous_commit_digest: Hash, + ready: bool, +} + +impl ExternalActionCoordinatorV1 { + /// Recovers coordinator authority from one checked local-store snapshot. + pub fn recover(store: &impl WalStorePort) -> Result { + let snapshot = store.read_snapshot()?; + let report = recover_from_frames_and_commits( + &snapshot.frames, + &snapshot.commits, + RecoveryAccessMode::ReadOnly, + )?; + if report.tail_posture != RecoveryTailPosture::Clean { + return Err(ExternalActionProtocolErrorV1::WalTailNotClean); + } + let index = observe_external_actions(&report)?; + let (next_lsn, previous_frame_digest, previous_commit_digest) = + external_action_wal_continuation(&report)?; + Ok(Self { + index, + next_lsn, + previous_frame_digest, + previous_commit_digest, + ready: true, + }) + } + + /// Returns the observation-only lifecycle index. + #[must_use] + pub const fn observed_index(&self) -> &RecoveredExternalActionIndexV1 { + &self.index + } + + /// Reconstructs request-transition authority after a request-commit crash. + pub fn recorded_request( + &self, + request_id: ExternalActionRequestIdV1, + ) -> Result { + self.ensure_ready()?; + let entry = self + .index + .get(request_id) + .ok_or(ExternalActionProtocolErrorV1::MissingRequest)?; + if entry.claim.is_some() { + return Err(ExternalActionProtocolErrorV1::DuplicateClaim); + } + Ok(DurablyRecordedExternalActionRequestV1 { + request: entry.request, + request_commit_digest: entry.request_commit_digest, + }) + } + + /// Reconstructs adapter settlement authority after a claim-commit crash. + pub fn claim_grant( + &self, + request_id: ExternalActionRequestIdV1, + ) -> Result { + self.ensure_ready()?; + let entry = self + .index + .get(request_id) + .ok_or(ExternalActionProtocolErrorV1::MissingRequest)?; + let claim = entry + .claim + .ok_or(ExternalActionProtocolErrorV1::MissingClaim)?; + if entry.settlement.is_some() { + return Err(ExternalActionProtocolErrorV1::DuplicateSettlement); + } + Ok(ExternalActionClaimGrantV1 { + request: entry.request, + claim, + claim_commit_digest: entry + .claim_commit_digest + .ok_or(ExternalActionProtocolErrorV1::MissingClaim)?, + }) + } + + /// Reconstructs the deterministic resumption fact after settlement commit. + pub fn admitted_settlement( + &self, + request_id: ExternalActionRequestIdV1, + ) -> Result { + self.ensure_ready()?; + let entry = self + .index + .get(request_id) + .ok_or(ExternalActionProtocolErrorV1::MissingRequest)?; + Ok(AdmittedExternalActionSettlementV1 { + settlement: entry + .settlement + .clone() + .ok_or(ExternalActionProtocolErrorV1::MissingSettlement)?, + settlement_commit_digest: entry + .settlement_commit_digest + .ok_or(ExternalActionProtocolErrorV1::MissingSettlement)?, + }) + } + + fn ensure_ready(&self) -> Result<(), ExternalActionProtocolErrorV1> { + if self.ready { + Ok(()) + } else { + Err(ExternalActionProtocolErrorV1::CoordinatorRecoveryRequired) + } + } + + fn transaction_builder( + &self, + context: ExternalActionTransactionContextV1, + expected_kind: WalTransactionKind, + ) -> Result { + self.ensure_ready()?; + Ok(WalTransactionBuilder::new_external_action( + context.writer_epoch, + context.segment_id, + context.transaction_id, + expected_kind, + self.next_lsn, + self.previous_frame_digest, + self.previous_commit_digest, + context.durability_mode, + context.payload_codec_id, + context.payload_schema_id, + context.payload_schema_version, + context.canonical_encoding_version, + context.digest_domain, + )) + } + + fn append_transaction( + &mut self, + store: &mut impl WalStorePort, + transaction: WalCommittedTransaction, + ) -> Result { + transaction.validate().map_err(WalBuildError::Validation)?; + let capability = transaction + .external_action_coordinator_capability() + .ok_or(WalBuildError::ExternalActionCoordinatorCapabilityRequired)?; + let epoch_id = transaction.commit.writer_epoch; + let commit = transaction.commit; + let last_lsn = commit.last_lsn; + let last_frame_digest = transaction + .frames + .last() + .map(crate::causal_wal::WalFrame::digest) + .ok_or(WalBuildError::EmptyTransaction)?; + self.ready = false; + for frame in transaction.frames { + store.append_frame(epoch_id, frame)?; + } + store.flush_external_action_commit(epoch_id, commit.clone(), capability)?; + self.next_lsn = last_lsn.checked_next().ok_or(WalBuildError::LsnOverflow)?; + self.previous_frame_digest = last_frame_digest; + self.previous_commit_digest = commit.commit_digest; + self.ready = true; + Ok(commit.commit_digest) + } +} + +impl RecoveredExternalActionIndexV1 { + /// Returns one recovered request. + #[must_use] + pub fn get(&self, request_id: ExternalActionRequestIdV1) -> Option<&RecoveredExternalActionV1> { + self.entries.get(&request_id) + } + + /// Returns the number of recovered requests. + #[must_use] + pub fn len(&self) -> usize { + self.entries.len() + } + + /// Returns whether the index contains no requests. + #[must_use] + pub fn is_empty(&self) -> bool { + self.entries.is_empty() + } + + /// Commits the complete authoritative external-action lifecycle index. + #[must_use] + pub fn root_digest(&self) -> Hash { + self.merkle_nodes + .get(&ExternalActionIndexNodeKeyV1 { + depth: 0, + prefix: [0; 32], + }) + .copied() + .unwrap_or_else(|| external_action_empty_hashes()[0]) + } + + fn plan_entry(&self, entry: RecoveredExternalActionV1) -> ExternalActionIndexMutationV1 { + let request_hash = entry.request.request_id.as_hash(); + let mut child_hash = external_action_index_leaf(&entry); + let mut node_updates = Vec::with_capacity(257); + node_updates.push(( + ExternalActionIndexNodeKeyV1 { + depth: 256, + prefix: request_hash, + }, + child_hash, + )); + for depth in (0_u16..256).rev() { + let child_depth = depth + 1; + let mut sibling_prefix = external_action_index_prefix(request_hash, child_depth); + external_action_toggle_index_bit(&mut sibling_prefix, depth); + let sibling_hash = self + .merkle_nodes + .get(&ExternalActionIndexNodeKeyV1 { + depth: child_depth, + prefix: sibling_prefix, + }) + .copied() + .unwrap_or_else(|| external_action_empty_hashes()[usize::from(child_depth)]); + child_hash = if external_action_index_bit(request_hash, depth) { + external_action_index_node_hash(depth, sibling_hash, child_hash) + } else { + external_action_index_node_hash(depth, child_hash, sibling_hash) + }; + node_updates.push(( + ExternalActionIndexNodeKeyV1 { + depth, + prefix: external_action_index_prefix(request_hash, depth), + }, + child_hash, + )); + } + ExternalActionIndexMutationV1 { + entry, + node_updates, + root_digest: child_hash, + } + } + + fn insert_entry(&mut self, entry: RecoveredExternalActionV1) -> bool { + let request_id = entry.request.request_id; + if self.entries.contains_key(&request_id) { + return false; + } + let mutation = self.plan_entry(entry); + self.apply_mutation(mutation); + true + } + + fn replace_entry(&mut self, entry: RecoveredExternalActionV1) { + let request_id = entry.request.request_id; + debug_assert!(self.entries.contains_key(&request_id)); + let mutation = self.plan_entry(entry); + self.apply_mutation(mutation); + } + + fn apply_mutation(&mut self, mutation: ExternalActionIndexMutationV1) { + self.entries + .insert(mutation.entry.request.request_id, mutation.entry); + for (key, digest) in mutation.node_updates { + self.merkle_nodes.insert(key, digest); + } + } +} + +#[derive(Clone, Debug, PartialEq, Eq)] +struct ExternalActionIndexMutationV1 { + entry: RecoveredExternalActionV1, + node_updates: Vec<(ExternalActionIndexNodeKeyV1, Hash)>, + root_digest: Hash, +} + +/// Fail-closed protocol and admission errors. +#[derive(Debug, Error, PartialEq, Eq)] +pub enum ExternalActionProtocolErrorV1 { + /// A request delegated no usable result or attempt budget. + #[error("external-action request budget must be non-zero")] + EmptyBudget, + /// A request exceeded Echo's absolute retained-settlement ceiling. + #[error("external-action request settlement budget exceeds the v1 limit")] + RequestBudgetLimitExceeded, + /// Protocol v1 permits exactly one claim per request. + #[error("external-action protocol v1 requires exactly one attempt per request")] + UnsupportedAttemptBudget, + /// The request identity did not match its canonical fields. + #[error("external-action request identity mismatch")] + RequestIdentityMismatch, + /// The adapter did not own the requested operation and scope. + #[error("external-action adapter is unauthorized")] + UnauthorizedAdapter, + /// The adapter authorization named a different request, basis, or policy. + #[error("external-action adapter authorization binding mismatch")] + AuthorizationBindingMismatch, + /// The adapter claim omitted lease or fencing evidence. + #[error("external-action claim omitted lease evidence")] + MissingLeaseEvidence, + /// The adapter claim omitted runtime registry policy evidence. + #[error("external-action claim omitted authorization policy evidence")] + MissingAuthorizationPolicyEvidence, + /// The current basis differed from the request basis. + #[error("external-action request basis is stale")] + StaleBasis, + /// The attempt exceeded the delegated request budget. + #[error("external-action attempt budget exhausted")] + AttemptBudgetExhausted, + /// A recovered claim did not match its exact request. + #[error("external-action claim binding mismatch")] + ClaimBindingMismatch, + /// The settlement named a schema other than the request schema. + #[error("external-action settlement schema mismatch")] + SettlementSchemaMismatch, + /// The settlement result digest did not match its canonical bytes. + #[error("external-action settlement result digest mismatch")] + SettlementResultDigestMismatch, + /// The settlement exceeded the delegated byte budget. + #[error("external-action settlement byte budget exceeded")] + SettlementBudgetExceeded, + /// The settlement did not name the claimed request, attempt, adapter, or basis. + #[error("external-action settlement claim binding mismatch")] + SettlementClaimMismatch, + /// A request was recorded more than once. + #[error("duplicate external-action request")] + DuplicateRequest, + /// A request was claimed more than once. + #[error("duplicate external-action claim")] + DuplicateClaim, + /// A request was settled more than once. + #[error("duplicate external-action settlement")] + DuplicateSettlement, + /// Distinct settlements claimed the same request. + #[error("conflicting external-action settlement")] + ConflictingSettlement, + /// A claim appeared without its request. + #[error("external-action claim is missing its request")] + MissingRequest, + /// A settlement appeared without its claim. + #[error("external-action settlement is missing its claim")] + MissingClaim, + /// Deterministic resumption was requested before settlement admission. + #[error("external-action request is missing its settlement")] + MissingSettlement, + /// A prior append failed after mutation may have begun; local recovery is required. + #[error("external-action coordinator requires trusted local recovery")] + CoordinatorRecoveryRequired, + /// WAL recovery found an uncommitted tail; lifecycle admission must stop. + #[error("external-action admission requires a clean committed WAL tail")] + WalTailNotClean, + /// Schema admission evidence was absent. + #[error("external-action settlement omitted schema admission evidence")] + MissingSchemaAdmissionEvidence, + /// External observation or reconciliation evidence was absent. + #[error("external-action settlement omitted external evidence")] + MissingExternalEvidence, + /// The WAL frontier did not commit the Echo-derived lifecycle index roots. + #[error("external-action frontier mismatch")] + ExternalActionFrontierMismatch { + /// Frontier root required by the reconstructed lifecycle transition. + expected: Hash, + /// Frontier root retained by the WAL commit. + actual: Hash, + }, + /// Canonical payload decoding failed. + #[error(transparent)] + Decode(#[from] WalDecodeError), + /// WAL transaction construction failed. + #[error(transparent)] + WalBuild(#[from] WalBuildError), + /// Durable WAL append failed. + #[error(transparent)] + WalStore(#[from] WalStoreError), + /// Reading current committed WAL posture failed. + #[error(transparent)] + WalRecovery(#[from] WalRecoveryError), +} + +fn build_external_action_request_transaction( + mut builder: WalTransactionBuilder, + request: &ExternalActionRequestV1, + affected_frontiers: Vec, +) -> Result { + request.validate_identity()?; + builder.push_record( + WalRecordKind::ExternalActionRequestRecorded, + request.to_payload_bytes(), + )?; + Ok(builder.commit(affected_frontiers)?) +} + +fn build_external_action_claim_transaction( + mut builder: WalTransactionBuilder, + claim: &ExternalActionClaimV1, + affected_frontiers: Vec, +) -> Result { + builder.push_record( + WalRecordKind::ExternalActionClaimRecorded, + claim.to_payload_bytes(), + )?; + Ok(builder.commit(affected_frontiers)?) +} + +fn build_external_action_settlement_transaction( + mut builder: WalTransactionBuilder, + settlement: &ExternalActionSettlementV1, + affected_frontiers: Vec, +) -> Result { + builder.push_record( + WalRecordKind::ExternalActionSettlementRecorded, + settlement.to_payload_bytes(), + )?; + Ok(builder.commit(affected_frontiers)?) +} + +/// Commits a request before returning the only value accepted by claim admission. +pub fn record_external_action_request( + store: &mut impl WalStorePort, + coordinator: &mut ExternalActionCoordinatorV1, + context: ExternalActionTransactionContextV1, + request: ExternalActionRequestV1, +) -> Result { + coordinator.ensure_ready()?; + if coordinator.index.get(request.request_id).is_some() { + return Err(ExternalActionProtocolErrorV1::DuplicateRequest); + } + let next_entry = RecoveredExternalActionV1 { + request, + request_commit_digest: [0; 32], + claim: None, + claim_commit_digest: None, + settlement: None, + settlement_commit_digest: None, + posture: RecoveredExternalActionPostureV1::Requested, + }; + let mut mutation = coordinator.index.plan_entry(next_entry); + let builder = + coordinator.transaction_builder(context, WalTransactionKind::ExternalActionRequest)?; + let transaction = build_external_action_request_transaction( + builder, + &request, + external_action_index_frontier(coordinator.index.root_digest(), mutation.root_digest), + )?; + let request_commit_digest = coordinator.append_transaction(store, transaction)?; + mutation.entry.request_commit_digest = request_commit_digest; + coordinator.index.apply_mutation(mutation); + Ok(DurablyRecordedExternalActionRequestV1 { + request, + request_commit_digest, + }) +} + +/// Commits a bounded claim before returning adapter work authority. +#[allow(clippy::too_many_arguments)] +pub fn claim_external_action( + store: &mut impl WalStorePort, + coordinator: &mut ExternalActionCoordinatorV1, + context: ExternalActionTransactionContextV1, + recorded_request: DurablyRecordedExternalActionRequestV1, + authorization: ExternalActionAdapterAuthorizationV1, + current_basis_digest: Hash, + attempt_ordinal: u32, + lease_evidence_digest: Hash, +) -> Result { + coordinator.ensure_ready()?; + let request = recorded_request.request; + request.validate_identity()?; + let recovered = coordinator + .index + .get(request.request_id) + .cloned() + .ok_or(ExternalActionProtocolErrorV1::MissingRequest)?; + if recovered.request != request { + return Err(ExternalActionProtocolErrorV1::RequestIdentityMismatch); + } + if recovered.claim.is_some() { + return Err(ExternalActionProtocolErrorV1::DuplicateClaim); + } + if authorization.operation_id != request.operation_id + || authorization.authority_scope_digest != request.authority_scope_digest + { + return Err(ExternalActionProtocolErrorV1::UnauthorizedAdapter); + } + if authorization.request_id != request.request_id + || authorization.basis_digest != request.basis_digest + || authorization.registry_policy_digest == [0; 32] + { + return Err(ExternalActionProtocolErrorV1::AuthorizationBindingMismatch); + } + if current_basis_digest != request.basis_digest { + return Err(ExternalActionProtocolErrorV1::StaleBasis); + } + if attempt_ordinal >= request.budget.max_attempts { + return Err(ExternalActionProtocolErrorV1::AttemptBudgetExhausted); + } + if lease_evidence_digest == [0; 32] { + return Err(ExternalActionProtocolErrorV1::MissingLeaseEvidence); + } + let claim = ExternalActionClaimV1::for_request( + &request, + authorization.adapter_id, + attempt_ordinal, + lease_evidence_digest, + authorization.registry_policy_digest, + ); + let mut next_entry = recovered; + next_entry.claim = Some(claim); + next_entry.claim_commit_digest = None; + next_entry.posture = RecoveredExternalActionPostureV1::Claimed; + let mut mutation = coordinator.index.plan_entry(next_entry); + let builder = + coordinator.transaction_builder(context, WalTransactionKind::ExternalActionClaim)?; + let transaction = build_external_action_claim_transaction( + builder, + &claim, + external_action_index_frontier(coordinator.index.root_digest(), mutation.root_digest), + )?; + let claim_commit_digest = coordinator.append_transaction(store, transaction)?; + mutation.entry.claim_commit_digest = Some(claim_commit_digest); + coordinator.index.apply_mutation(mutation); + Ok(ExternalActionClaimGrantV1 { + request, + claim, + claim_commit_digest, + }) +} + +/// Validates and commits a settlement before returning a resumable fact. +pub fn admit_external_action_settlement( + store: &mut impl WalStorePort, + coordinator: &mut ExternalActionCoordinatorV1, + context: ExternalActionTransactionContextV1, + claim_grant: ExternalActionClaimGrantV1, + candidate: ExternalActionSettlementCandidateV1, +) -> Result { + coordinator.ensure_ready()?; + let recovered = coordinator + .index + .get(claim_grant.request.request_id) + .cloned() + .ok_or(ExternalActionProtocolErrorV1::MissingRequest)?; + let recovered_claim = recovered + .claim + .ok_or(ExternalActionProtocolErrorV1::MissingClaim)?; + if recovered.request != claim_grant.request || recovered_claim != claim_grant.claim { + return Err(ExternalActionProtocolErrorV1::SettlementClaimMismatch); + } + if recovered.settlement.is_some() { + return Err(ExternalActionProtocolErrorV1::DuplicateSettlement); + } + validate_settlement_candidate(&claim_grant.request, &claim_grant.claim, &candidate)?; + let settlement = ExternalActionSettlementV1::from_candidate(candidate); + let mut next_entry = recovered; + next_entry.posture = RecoveredExternalActionPostureV1::Settled(settlement.kind); + next_entry.settlement = Some(settlement.clone()); + next_entry.settlement_commit_digest = None; + let mut mutation = coordinator.index.plan_entry(next_entry); + let builder = + coordinator.transaction_builder(context, WalTransactionKind::ExternalActionSettlement)?; + let transaction = build_external_action_settlement_transaction( + builder, + &settlement, + external_action_index_frontier(coordinator.index.root_digest(), mutation.root_digest), + )?; + let settlement_commit_digest = coordinator.append_transaction(store, transaction)?; + mutation.entry.settlement_commit_digest = Some(settlement_commit_digest); + coordinator.index.apply_mutation(mutation); + Ok(AdmittedExternalActionSettlementV1 { + settlement, + settlement_commit_digest, + }) +} + +/// Observes request, claim, and settlement posture in an arbitrary recovery report. +/// +/// This projection carries no transition or replay authority. Use +/// [`ExternalActionCoordinatorV1::recover`] to reconstruct trusted local +/// transition grants and resumable settlements. +pub fn observe_external_actions( + report: &RecoveryScanReport, +) -> Result { + let mut index = RecoveredExternalActionIndexV1::default(); + for transaction in &report.transactions { + let Some(frame) = + external_action_frame(transaction.commit.transaction_kind, &transaction.frames)? + else { + continue; + }; + let before_root = index.root_digest(); + match frame.header.record_kind { + WalRecordKind::ExternalActionRequestRecorded => { + let request = + ExternalActionRequestV1::from_payload_bytes(&frame.payload.canonical_bytes)?; + if !index.insert_entry(RecoveredExternalActionV1 { + request, + request_commit_digest: transaction.commit.commit_digest, + claim: None, + claim_commit_digest: None, + settlement: None, + settlement_commit_digest: None, + posture: RecoveredExternalActionPostureV1::Requested, + }) { + return Err(ExternalActionProtocolErrorV1::DuplicateRequest); + } + } + WalRecordKind::ExternalActionClaimRecorded => { + let claim = + ExternalActionClaimV1::from_payload_bytes(&frame.payload.canonical_bytes)?; + let mut entry = index + .get(claim.request_id) + .cloned() + .ok_or(ExternalActionProtocolErrorV1::MissingRequest)?; + if entry.claim.is_some() { + return Err(ExternalActionProtocolErrorV1::DuplicateClaim); + } + validate_claim(&entry.request, &claim)?; + entry.claim = Some(claim); + entry.claim_commit_digest = Some(transaction.commit.commit_digest); + entry.posture = RecoveredExternalActionPostureV1::Claimed; + index.replace_entry(entry); + } + WalRecordKind::ExternalActionSettlementRecorded => { + let settlement = + ExternalActionSettlementV1::from_payload_bytes(&frame.payload.canonical_bytes)?; + apply_recovered_settlement( + &mut index, + settlement, + transaction.commit.commit_digest, + )?; + } + _ => unreachable!("external_action_frame filters record kinds"), + } + let after_root = index.root_digest(); + let expected_frontier_root = affected_frontiers_root(&[AffectedFrontier { + kind: AffectedFrontierKind::ExternalActionIndex, + before_digest: before_root, + after_digest: after_root, + }]); + if transaction.commit.affected_frontiers_root != expected_frontier_root { + return Err( + ExternalActionProtocolErrorV1::ExternalActionFrontierMismatch { + expected: expected_frontier_root, + actual: transaction.commit.affected_frontiers_root, + }, + ); + } + } + Ok(index) +} + +fn apply_recovered_settlement( + index: &mut RecoveredExternalActionIndexV1, + settlement: ExternalActionSettlementV1, + commit_digest: Hash, +) -> Result<(), ExternalActionProtocolErrorV1> { + let mut entry = index + .get(settlement.request_id) + .cloned() + .ok_or(ExternalActionProtocolErrorV1::MissingRequest)?; + let claim = entry + .claim + .ok_or(ExternalActionProtocolErrorV1::MissingClaim)?; + validate_settlement(&entry.request, &claim, &settlement)?; + if let Some(existing) = &entry.settlement { + return if existing == &settlement { + Err(ExternalActionProtocolErrorV1::DuplicateSettlement) + } else { + Err(ExternalActionProtocolErrorV1::ConflictingSettlement) + }; + } + entry.posture = RecoveredExternalActionPostureV1::Settled(settlement.kind); + entry.settlement = Some(settlement); + entry.settlement_commit_digest = Some(commit_digest); + index.replace_entry(entry); + Ok(()) +} + +fn external_action_index_frontier( + before_digest: Hash, + after_digest: Hash, +) -> Vec { + vec![AffectedFrontier { + kind: AffectedFrontierKind::ExternalActionIndex, + before_digest, + after_digest, + }] +} + +fn hash_len_prefixed(hasher: &mut blake3::Hasher, bytes: &[u8]) { + hasher.update(&u64::try_from(bytes.len()).unwrap_or(u64::MAX).to_le_bytes()); + hasher.update(bytes); +} + +fn external_action_index_leaf(entry: &RecoveredExternalActionV1) -> Hash { + let mut hasher = blake3::Hasher::new(); + hasher.update(INDEX_LEAF_DOMAIN); + hasher.update(&entry.request.request_id.as_hash()); + hash_len_prefixed(&mut hasher, &entry.request.to_payload_bytes()); + match entry.claim { + Some(claim) => { + hasher.update(&[1]); + hash_len_prefixed(&mut hasher, &claim.to_payload_bytes()); + } + None => { + hasher.update(&[0]); + } + } + match &entry.settlement { + Some(settlement) => { + hasher.update(&[1]); + hash_len_prefixed(&mut hasher, &settlement.to_payload_bytes()); + } + None => { + hasher.update(&[0]); + } + } + hasher.finalize().into() +} + +fn external_action_index_node_hash(depth: u16, left: Hash, right: Hash) -> Hash { + let mut hasher = blake3::Hasher::new(); + hasher.update(INDEX_NODE_DOMAIN); + hasher.update(&depth.to_le_bytes()); + hasher.update(&left); + hasher.update(&right); + hasher.finalize().into() +} + +fn external_action_empty_hashes() -> &'static [Hash; 257] { + static EMPTY_HASHES: OnceLock<[Hash; 257]> = OnceLock::new(); + EMPTY_HASHES.get_or_init(|| { + let mut hashes = [[0; 32]; 257]; + hashes[256] = blake3::hash(INDEX_EMPTY_LEAF_DOMAIN).into(); + for depth in (0_u16..256).rev() { + let child = hashes[usize::from(depth + 1)]; + hashes[usize::from(depth)] = external_action_index_node_hash(depth, child, child); + } + hashes + }) +} + +fn external_action_index_prefix(mut request_id: Hash, depth: u16) -> Hash { + if depth == 256 { + return request_id; + } + let byte_index = usize::from(depth / 8); + let retained_bits = depth % 8; + if retained_bits == 0 { + request_id[byte_index..].fill(0); + } else { + request_id[byte_index] &= u8::MAX << (8 - retained_bits); + request_id[(byte_index + 1)..].fill(0); + } + request_id +} + +fn external_action_index_bit(request_id: Hash, depth: u16) -> bool { + let byte_index = usize::from(depth / 8); + let bit_index = 7 - (depth % 8); + request_id[byte_index] & (1_u8 << bit_index) != 0 +} + +fn external_action_toggle_index_bit(prefix: &mut Hash, depth: u16) { + let byte_index = usize::from(depth / 8); + let bit_index = 7 - (depth % 8); + prefix[byte_index] ^= 1_u8 << bit_index; +} + +fn external_action_idempotency_key(request: &ExternalActionRequestV1) -> Hash { + let mut hasher = blake3::Hasher::new(); + hasher.update(IDEMPOTENCY_KEY_DOMAIN); + hasher.update(&request.request_id.as_hash()); + hasher.update(&request.reconciliation_law_digest); + hasher.finalize().into() +} + +fn external_action_attempt_id( + request_id: ExternalActionRequestIdV1, + attempt_ordinal: u32, + adapter_id: ExternalActionAdapterIdV1, + lease_evidence_digest: Hash, + authorization_policy_digest: Hash, +) -> ExternalActionAttemptIdV1 { + let mut hasher = blake3::Hasher::new(); + hasher.update(ATTEMPT_ID_DOMAIN); + hasher.update(&request_id.as_hash()); + hasher.update(&attempt_ordinal.to_le_bytes()); + hasher.update(&adapter_id.as_hash()); + hasher.update(&lease_evidence_digest); + hasher.update(&authorization_policy_digest); + ExternalActionAttemptIdV1::from_hash(hasher.finalize().into()) +} + +fn validate_claim( + request: &ExternalActionRequestV1, + claim: &ExternalActionClaimV1, +) -> Result<(), ExternalActionProtocolErrorV1> { + let expected = ExternalActionClaimV1::for_request( + request, + claim.adapter_id, + claim.attempt_ordinal, + claim.lease_evidence_digest, + claim.authorization_policy_digest, + ); + if *claim != expected { + return Err(ExternalActionProtocolErrorV1::ClaimBindingMismatch); + } + if claim.attempt_ordinal >= request.budget.max_attempts { + return Err(ExternalActionProtocolErrorV1::AttemptBudgetExhausted); + } + if claim.lease_evidence_digest == [0; 32] { + return Err(ExternalActionProtocolErrorV1::MissingLeaseEvidence); + } + if claim.authorization_policy_digest == [0; 32] { + return Err(ExternalActionProtocolErrorV1::MissingAuthorizationPolicyEvidence); + } + Ok(()) +} + +fn validate_settlement_candidate( + request: &ExternalActionRequestV1, + claim: &ExternalActionClaimV1, + candidate: &ExternalActionSettlementCandidateV1, +) -> Result<(), ExternalActionProtocolErrorV1> { + if candidate.request_id != request.request_id + || candidate.attempt_id != claim.attempt_id + || candidate.adapter_id != claim.adapter_id + || candidate.basis_digest != request.basis_digest + { + return Err(ExternalActionProtocolErrorV1::SettlementClaimMismatch); + } + if candidate.settlement_schema_digest != request.settlement_schema_digest { + return Err(ExternalActionProtocolErrorV1::SettlementSchemaMismatch); + } + if candidate.schema_admission_evidence_digest == [0; 32] { + return Err(ExternalActionProtocolErrorV1::MissingSchemaAdmissionEvidence); + } + if candidate.external_evidence_digest == [0; 32] { + return Err(ExternalActionProtocolErrorV1::MissingExternalEvidence); + } + if u64::try_from(candidate.canonical_result_bytes.len()).unwrap_or(u64::MAX) + > request.budget.max_settlement_bytes + { + return Err(ExternalActionProtocolErrorV1::SettlementBudgetExceeded); + } + if Hash::from(blake3::hash(&candidate.canonical_result_bytes)) + != candidate.declared_result_digest + { + return Err(ExternalActionProtocolErrorV1::SettlementResultDigestMismatch); + } + Ok(()) +} + +fn validate_settlement( + request: &ExternalActionRequestV1, + claim: &ExternalActionClaimV1, + settlement: &ExternalActionSettlementV1, +) -> Result<(), ExternalActionProtocolErrorV1> { + validate_settlement_candidate( + request, + claim, + &ExternalActionSettlementCandidateV1 { + request_id: settlement.request_id, + attempt_id: settlement.attempt_id, + adapter_id: settlement.adapter_id, + kind: settlement.kind, + settlement_schema_digest: settlement.settlement_schema_digest, + basis_digest: settlement.basis_digest, + canonical_result_bytes: settlement.canonical_result_bytes.clone(), + declared_result_digest: settlement.result_digest, + schema_admission_evidence_digest: settlement.schema_admission_evidence_digest, + external_evidence_digest: settlement.external_evidence_digest, + }, + ) +} + +fn external_action_wal_continuation( + report: &RecoveryScanReport, +) -> Result<(Lsn, Hash, Hash), ExternalActionProtocolErrorV1> { + let Some(last_transaction) = report.transactions.last() else { + return Ok((Lsn::from_raw(0), [0; 32], [0; 32])); + }; + let next_lsn = last_transaction + .commit + .last_lsn + .checked_next() + .ok_or(WalBuildError::LsnOverflow)?; + let previous_frame_digest = last_transaction + .frames + .last() + .map(crate::causal_wal::WalFrame::digest) + .ok_or(WalBuildError::EmptyTransaction)?; + Ok(( + next_lsn, + previous_frame_digest, + last_transaction.commit.commit_digest, + )) +} + +fn external_action_frame( + transaction_kind: WalTransactionKind, + frames: &[crate::causal_wal::WalFrame], +) -> Result, ExternalActionProtocolErrorV1> { + let expected = transaction_kind.external_action_record_kind(); + let Some(expected) = expected else { + return Ok(None); + }; + if frames.len() != 1 || frames[0].header.record_kind != expected { + return Err(WalBuildError::Validation( + crate::causal_wal::WalValidationError::ExternalActionFrameShapeMismatch, + ) + .into()); + } + Ok(frames.first()) +} + +struct ExternalActionPayloadCursor<'a> { + bytes: &'a [u8], + offset: usize, +} + +impl<'a> ExternalActionPayloadCursor<'a> { + const fn new(bytes: &'a [u8]) -> Self { + Self { bytes, offset: 0 } + } + + fn read_bytes(&mut self, len: usize) -> Result<&'a [u8], WalDecodeError> { + let end = self + .offset + .checked_add(len) + .ok_or(WalDecodeError::UnexpectedEof)?; + let bytes = self + .bytes + .get(self.offset..end) + .ok_or(WalDecodeError::UnexpectedEof)?; + self.offset = end; + Ok(bytes) + } + + fn read_u8(&mut self) -> Result { + Ok(self.read_bytes(1)?[0]) + } + + fn read_u32(&mut self) -> Result { + let mut bytes = [0; 4]; + bytes.copy_from_slice(self.read_bytes(4)?); + Ok(u32::from_le_bytes(bytes)) + } + + fn read_u64(&mut self) -> Result { + let mut bytes = [0; 8]; + bytes.copy_from_slice(self.read_bytes(8)?); + Ok(u64::from_le_bytes(bytes)) + } + + fn read_hash(&mut self) -> Result { + let mut bytes = [0; 32]; + bytes.copy_from_slice(self.read_bytes(32)?); + Ok(bytes) + } + + fn expect_magic( + &mut self, + magic: &[u8], + record_kind: &'static str, + ) -> Result<(), WalDecodeError> { + if self.read_bytes(magic.len())? != magic { + return Err(WalDecodeError::InvalidRecordMagic { record_kind }); + } + Ok(()) + } + + fn finish(self) -> Result<(), WalDecodeError> { + if self.offset != self.bytes.len() { + return Err(WalDecodeError::TrailingBytes); + } + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn digest(label: &str) -> Hash { + blake3::hash(label.as_bytes()).into() + } + + fn request() -> ExternalActionRequestV1 { + match ExternalActionRequestV1::new( + WorldlineId::from_bytes([41; 32]), + ExternalActionOperationIdV1::from_hash(digest("test.operation@1")), + digest("test.operation@1.input"), + digest("test.operation@1.settlement"), + digest("test.scope"), + digest("test.basis"), + ExternalActionBudgetV1 { + max_settlement_bytes: 64, + max_attempts: 1, + }, + digest("test.input"), + digest("test.reconciliation"), + ) { + Ok(request) => request, + Err(error) => panic!("request fixture failed: {error:?}"), + } + } + + #[test] + fn conflicting_recovered_settlement_is_obstructed() { + let request = request(); + let claim = ExternalActionClaimV1::for_request( + &request, + ExternalActionAdapterIdV1::from_hash(digest("test.adapter")), + 0, + digest("test.lease"), + digest("test.policy"), + ); + let mut index = RecoveredExternalActionIndexV1::default(); + assert!(index.insert_entry(RecoveredExternalActionV1 { + request, + request_commit_digest: digest("request.commit"), + claim: Some(claim), + claim_commit_digest: Some(digest("claim.commit")), + settlement: None, + settlement_commit_digest: None, + posture: RecoveredExternalActionPostureV1::Claimed, + })); + let first = + ExternalActionSettlementV1::from_candidate(ExternalActionSettlementCandidateV1::new( + request.request_id, + claim.attempt_id, + claim.adapter_id, + ExternalActionSettlementKindV1::Succeeded, + request.settlement_schema_digest, + request.basis_digest, + b"first".to_vec(), + digest("test.schema-evidence"), + digest("test.external-evidence"), + )); + assert_eq!( + apply_recovered_settlement(&mut index, first, digest("settlement.commit")), + Ok(()) + ); + let conflicting = + ExternalActionSettlementV1::from_candidate(ExternalActionSettlementCandidateV1::new( + request.request_id, + claim.attempt_id, + claim.adapter_id, + ExternalActionSettlementKindV1::Succeeded, + request.settlement_schema_digest, + request.basis_digest, + b"second".to_vec(), + digest("test.schema-evidence"), + digest("test.external-evidence"), + )); + assert_eq!( + apply_recovered_settlement(&mut index, conflicting, digest("conflict.commit")), + Err(ExternalActionProtocolErrorV1::ConflictingSettlement) + ); + } +} diff --git a/crates/warp-core/src/lib.rs b/crates/warp-core/src/lib.rs index 4b79ff44..cc76da61 100644 --- a/crates/warp-core/src/lib.rs +++ b/crates/warp-core/src/lib.rs @@ -69,6 +69,7 @@ mod echo_operation; mod edict_target_ir; mod engine_impl; pub mod evidence; +pub mod external_action; mod footprint; /// Footprint enforcement guard for parallel execution. /// diff --git a/crates/warp-core/src/trusted_runtime_host.rs b/crates/warp-core/src/trusted_runtime_host.rs index 0f7c4471..0d06bfa4 100644 --- a/crates/warp-core/src/trusted_runtime_host.rs +++ b/crates/warp-core/src/trusted_runtime_host.rs @@ -29,15 +29,16 @@ use crate::{ recover_from_frames_and_commits, recover_receipt_index, recover_submission_index, recovered_submission_receipt_index_root, tick_receipt_payload_is_batch, trusted_runtime_wal_digest, validate_recovered_causal_anchor_history, AffectedFrontier, - AffectedFrontierKind, FilesystemWalStore, InMemoryWalStore, Lsn, PayloadCodecId, - PayloadSchemaId, RecoveredCausalAnchorAdmission, RecoveredReceiptIndex, - RecoveredSubmissionIndex, RecoveryAccessMode, RecoveryCertificate, RecoveryScanReport, - SubmissionAcceptanceRecord, TickReceiptRecord, WalAppendAuthority, WalBuildError, - WalCommittedTransaction, WalDecodeError, WalDurabilityMode, WalReceiptCorrelationRecord, - WalRecordKind, WalRecoveryError, WalRecoveryIndexError, WalRuntimeStateDeltaRecord, - WalSegmentId, WalStoreError, WalStorePort, WalSubmissionEnvelopeRecord, WalTickDecision, - WalTransactionBuilder, WalTransactionCommit, WalTransactionId, WalTransactionKind, - WriterEpochId, WriterEpochRequest, TRUSTED_RUNTIME_WAL_DOMAIN, + AffectedFrontierKind, ExternalActionCoordinatorCapability, FilesystemWalStore, + InMemoryWalStore, Lsn, PayloadCodecId, PayloadSchemaId, RecoveredCausalAnchorAdmission, + RecoveredReceiptIndex, RecoveredSubmissionIndex, RecoveryAccessMode, RecoveryCertificate, + RecoveryScanReport, SubmissionAcceptanceRecord, TickReceiptRecord, WalAppendAuthority, + WalBuildError, WalCommittedTransaction, WalDecodeError, WalDurabilityMode, + WalReceiptCorrelationRecord, WalRecordKind, WalRecoveryError, WalRecoveryIndexError, + WalRuntimeStateDeltaRecord, WalSegmentId, WalStoreError, WalStorePort, WalStoreSnapshot, + WalSubmissionEnvelopeRecord, WalTickDecision, WalTransactionBuilder, WalTransactionCommit, + WalTransactionId, WalTransactionKind, WriterEpochId, WriterEpochRequest, + TRUSTED_RUNTIME_WAL_DOMAIN, }, contract_host::{decode_canonical_eint, encode_canonical_eint}, echo_operation::{ @@ -3449,6 +3450,22 @@ impl WalStorePort for TrustedRuntimeWalStore { } } + fn flush_external_action_commit( + &mut self, + epoch_id: WriterEpochId, + commit: WalTransactionCommit, + capability: ExternalActionCoordinatorCapability, + ) -> Result<(), WalStoreError> { + match self { + Self::InMemory(store) => { + store.flush_external_action_commit(epoch_id, commit, capability) + } + Self::Filesystem(store) => { + store.flush_external_action_commit(epoch_id, commit, capability) + } + } + } + fn read_frames(&self) -> Vec { match self { Self::InMemory(store) => store.read_frames(), @@ -3463,6 +3480,13 @@ impl WalStorePort for TrustedRuntimeWalStore { } } + fn read_snapshot(&self) -> Result { + match self { + Self::InMemory(store) => store.read_snapshot(), + Self::Filesystem(store) => store.read_snapshot(), + } + } + fn seal_segment( &mut self, epoch_id: WriterEpochId, diff --git a/crates/warp-core/tests/external_action_protocol_tests.rs b/crates/warp-core/tests/external_action_protocol_tests.rs new file mode 100644 index 00000000..fd3c5d9e --- /dev/null +++ b/crates/warp-core/tests/external_action_protocol_tests.rs @@ -0,0 +1,1432 @@ +// SPDX-License-Identifier: Apache-2.0 +// © James Ross Ω FLYING•ROBOTS +//! Durable external-action request and settlement protocol tests. + +#![allow(clippy::panic)] + +use std::cell::Cell; +use std::path::PathBuf; +use std::sync::atomic::{AtomicU64, Ordering}; + +use warp_core::causal_wal::{ + recover_filesystem_store, recover_from_frames_and_commits, recover_in_memory_store, + AffectedFrontierKind, ExternalActionCoordinatorCapability, FilesystemWalStore, + InMemoryWalStore, Lsn, PayloadCodecId, PayloadSchemaId, RecoveryAccessMode, + RecoveryTailPosture, WalAppendAuthority, WalBuildError, WalDurabilityMode, WalFrame, + WalManifest, WalRecordKind, WalSegmentId, WalSegmentSeal, WalStoreError, WalStorePort, + WalStoreSnapshot, WalTransactionBuilder, WalTransactionCommit, WalTransactionId, + WalTransactionKind, WriterEpoch, WriterEpochId, WriterEpochRequest, +}; +use warp_core::external_action::{ + admit_external_action_settlement, claim_external_action, observe_external_actions, + record_external_action_request, ExternalActionAdapterAuthorizationV1, + ExternalActionAdapterBindingV1, ExternalActionAdapterIdV1, ExternalActionAdapterRegistryV1, + ExternalActionBudgetV1, ExternalActionClaimGrantV1, ExternalActionCoordinatorV1, + ExternalActionOperationIdV1, ExternalActionProtocolErrorV1, ExternalActionRequestV1, + ExternalActionSettlementCandidateV1, ExternalActionSettlementKindV1, + ExternalActionTransactionContextV1, RecoveredExternalActionPostureV1, +}; +use warp_core::{Hash, WorldlineId}; + +fn digest(label: &str) -> Hash { + blake3::hash(label.as_bytes()).into() +} + +fn must_ok(result: Result) -> T { + match result { + Ok(value) => value, + Err(error) => panic!("expected Ok(..), got {error:?}"), + } +} + +fn epoch_id() -> WriterEpochId { + WriterEpochId::from_hash(digest("external-action:epoch")) +} + +fn store() -> InMemoryWalStore { + let mut store = InMemoryWalStore::new(); + must_ok(store.acquire_writer_epoch(WriterEpochRequest { + epoch_id: epoch_id(), + storage_fencing_token: digest("external-action:fencing"), + process_identity: digest("external-action:process"), + host_identity: digest("external-action:host"), + started_at_lsn: Lsn::from_raw(0), + previous_epoch_id: None, + previous_epoch_final_commit_digest: None, + lease_or_lock_evidence: digest("external-action:lease"), + })); + store +} + +fn coordinator(store: &impl WalStorePort) -> ExternalActionCoordinatorV1 { + must_ok(ExternalActionCoordinatorV1::recover(store)) +} + +fn raw_builder(label: &str, first_lsn: u64, kind: WalTransactionKind) -> WalTransactionBuilder { + WalTransactionBuilder::new( + epoch_id(), + WalSegmentId::from_raw(1), + WalTransactionId::from_hash(digest(label)), + kind, + WalAppendAuthority::ExternalActionCoordinator, + Lsn::from_raw(first_lsn), + [0; 32], + [0; 32], + WalDurabilityMode::Buffered, + PayloadCodecId::from_hash(digest("external-action:codec")), + PayloadSchemaId::from_hash(digest("external-action:schema")), + 1, + 1, + digest("external-action:domain"), + ) +} + +fn context(label: &str) -> ExternalActionTransactionContextV1 { + context_with_durability(label, WalDurabilityMode::Buffered) +} + +fn context_with_durability( + label: &str, + durability_mode: WalDurabilityMode, +) -> ExternalActionTransactionContextV1 { + ExternalActionTransactionContextV1 { + writer_epoch: epoch_id(), + segment_id: WalSegmentId::from_raw(1), + transaction_id: WalTransactionId::from_hash(digest(label)), + durability_mode, + payload_codec_id: PayloadCodecId::from_hash(digest("external-action:codec")), + payload_schema_id: PayloadSchemaId::from_hash(digest("external-action:schema")), + payload_schema_version: 1, + canonical_encoding_version: 1, + digest_domain: digest("external-action:domain"), + } +} + +static TEMP_WAL_COUNTER: AtomicU64 = AtomicU64::new(0); + +struct TempWalDir(PathBuf); + +impl TempWalDir { + fn new(label: &str) -> Self { + let counter = TEMP_WAL_COUNTER.fetch_add(1, Ordering::Relaxed); + let path = std::env::temp_dir().join(format!( + "echo-external-action-{}-{counter}-{label}", + std::process::id() + )); + if path.exists() { + must_ok(std::fs::remove_dir_all(&path)); + } + must_ok(std::fs::create_dir_all(&path)); + Self(path) + } +} + +impl Drop for TempWalDir { + fn drop(&mut self) { + let _ = std::fs::remove_dir_all(&self.0); + } +} + +fn request_with( + label: &str, + worldline_byte: u8, + max_settlement_bytes: u64, +) -> ExternalActionRequestV1 { + must_ok(ExternalActionRequestV1::new( + WorldlineId::from_bytes([worldline_byte; 32]), + ExternalActionOperationIdV1::from_hash(digest("workspace.observe@1")), + digest("workspace.observe@1.input"), + digest("workspace.observe@1.settlement"), + digest("workspace:/bounded"), + digest(&format!("basis:{label}")), + ExternalActionBudgetV1 { + max_settlement_bytes, + max_attempts: 1, + }, + digest(&format!("input:{label}")), + digest("workspace.observe@1.reconcile"), + )) +} + +fn adapter_id() -> ExternalActionAdapterIdV1 { + ExternalActionAdapterIdV1::from_hash(digest("adapter:workspace-observer")) +} + +fn adapter_registry() -> ExternalActionAdapterRegistryV1 { + ExternalActionAdapterRegistryV1::new([ExternalActionAdapterBindingV1 { + adapter_id: adapter_id(), + operation_id: ExternalActionOperationIdV1::from_hash(digest("workspace.observe@1")), + authority_scope_digest: digest("workspace:/bounded"), + }]) +} + +fn authorization(request: &ExternalActionRequestV1) -> ExternalActionAdapterAuthorizationV1 { + must_ok(adapter_registry().authorize(request, adapter_id())) +} + +#[allow(clippy::large_types_passed_by_value)] +fn record( + store: &mut impl WalStorePort, + coordinator: &mut ExternalActionCoordinatorV1, + request: ExternalActionRequestV1, + label: &str, +) -> warp_core::external_action::DurablyRecordedExternalActionRequestV1 { + must_ok(record_external_action_request( + store, + coordinator, + context(label), + request, + )) +} + +#[allow(clippy::large_types_passed_by_value)] +fn claim( + store: &mut impl WalStorePort, + coordinator: &mut ExternalActionCoordinatorV1, + recorded: warp_core::external_action::DurablyRecordedExternalActionRequestV1, + label: &str, +) -> ExternalActionClaimGrantV1 { + let basis = recorded.request().basis_digest; + let authorization = authorization(&recorded.request()); + must_ok(claim_external_action( + store, + coordinator, + context(label), + recorded, + authorization, + basis, + 0, + digest(&format!("{label}:lease")), + )) +} + +fn candidate( + grant: &ExternalActionClaimGrantV1, + kind: ExternalActionSettlementKindV1, + bytes: Vec, +) -> ExternalActionSettlementCandidateV1 { + let request = grant.request(); + let claim = grant.claim(); + ExternalActionSettlementCandidateV1::new( + request.request_id(), + claim.attempt_id, + claim.adapter_id, + kind, + request.settlement_schema_digest, + request.basis_digest, + bytes, + digest("settlement:schema-admission"), + digest("settlement:external-evidence"), + ) +} + +#[test] +fn request_and_settlement_are_committed_before_authority_crosses_the_boundary() { + let mut store = store(); + let mut coordinator = coordinator(&store); + let request = request_with("golden", 7, 128); + let recorded = record(&mut store, &mut coordinator, request, "request:golden"); + assert_eq!(store.read_commits().len(), 1); + assert_eq!( + store.read_commits()[0].commit_digest, + recorded.request_commit_digest() + ); + + let grant = claim(&mut store, &mut coordinator, recorded, "claim:golden"); + assert_eq!(store.read_commits().len(), 2); + assert_eq!( + store.read_commits()[1].commit_digest, + grant.claim_commit_digest() + ); + let result_bytes = b"observed workspace bytes".to_vec(); + let candidate = candidate( + &grant, + ExternalActionSettlementKindV1::Succeeded, + result_bytes.clone(), + ); + let admitted = must_ok(admit_external_action_settlement( + &mut store, + &mut coordinator, + context("settlement:golden"), + grant, + candidate, + )); + assert_eq!(store.read_commits().len(), 3); + assert_eq!( + store.read_commits()[2].commit_digest, + admitted.settlement_commit_digest() + ); + assert_eq!(admitted.settlement().canonical_result_bytes, result_bytes); +} + +#[test] +fn unauthorized_adapter_and_stale_basis_obstruct_before_claim_commit() { + let mut store = store(); + let mut coordinator = coordinator(&store); + let request = request_with("claim-obstructions", 8, 64); + let recorded = record( + &mut store, + &mut coordinator, + request, + "request:claim-obstructions", + ); + let commits_before = store.read_commits().len(); + assert_eq!( + adapter_registry().authorize( + &request, + ExternalActionAdapterIdV1::from_hash(digest("adapter:unauthorized")), + ), + Err(ExternalActionProtocolErrorV1::UnauthorizedAdapter) + ); + assert_eq!(store.read_commits().len(), commits_before); + + assert_eq!( + claim_external_action( + &mut store, + &mut coordinator, + context("claim:stale"), + recorded, + authorization(&request), + digest("basis:changed"), + 0, + digest("claim:stale:lease"), + ), + Err(ExternalActionProtocolErrorV1::StaleBasis) + ); + assert_eq!(store.read_commits().len(), commits_before); +} + +#[test] +fn adapter_authorization_is_bound_to_the_exact_request() { + let mut store = store(); + let mut coordinator = coordinator(&store); + let authorized_request = request_with("authorization-source", 8, 64); + let claimed_request = request_with("authorization-target", 8, 64); + let recorded = record( + &mut store, + &mut coordinator, + claimed_request, + "request:authorization-target", + ); + let commits_before = store.read_commits().len(); + + assert_eq!( + claim_external_action( + &mut store, + &mut coordinator, + context("claim:authorization-target"), + recorded, + authorization(&authorized_request), + claimed_request.basis_digest, + 0, + digest("claim:authorization-target:lease"), + ), + Err(ExternalActionProtocolErrorV1::AuthorizationBindingMismatch) + ); + assert_eq!(store.read_commits().len(), commits_before); +} + +#[test] +fn claims_and_settlements_require_nonzero_external_evidence() { + let mut claim_store = store(); + let mut claim_coordinator = coordinator(&claim_store); + let claim_request = request_with("missing-lease-evidence", 8, 64); + let claim_recorded = record( + &mut claim_store, + &mut claim_coordinator, + claim_request, + "request:missing-lease-evidence", + ); + let claim_commits_before = claim_store.read_commits().len(); + assert_eq!( + claim_external_action( + &mut claim_store, + &mut claim_coordinator, + context("claim:missing-lease-evidence"), + claim_recorded, + authorization(&claim_request), + claim_request.basis_digest, + 0, + [0; 32], + ), + Err(ExternalActionProtocolErrorV1::MissingLeaseEvidence) + ); + assert_eq!(claim_store.read_commits().len(), claim_commits_before); + + let mut settlement_store = store(); + let mut settlement_coordinator = coordinator(&settlement_store); + let settlement_request = request_with("missing-external-evidence", 8, 64); + let settlement_recorded = record( + &mut settlement_store, + &mut settlement_coordinator, + settlement_request, + "request:missing-external-evidence", + ); + let grant = claim( + &mut settlement_store, + &mut settlement_coordinator, + settlement_recorded, + "claim:missing-external-evidence", + ); + let mut candidate = candidate( + &grant, + ExternalActionSettlementKindV1::Succeeded, + b"observed".to_vec(), + ); + candidate.external_evidence_digest = [0; 32]; + let settlement_commits_before = settlement_store.read_commits().len(); + assert_eq!( + admit_external_action_settlement( + &mut settlement_store, + &mut settlement_coordinator, + context("settlement:missing-external-evidence"), + grant, + candidate, + ), + Err(ExternalActionProtocolErrorV1::MissingExternalEvidence) + ); + assert_eq!( + settlement_store.read_commits().len(), + settlement_commits_before + ); +} + +#[test] +fn recovery_rejects_forged_external_action_frontier_evidence() { + let mut store = store(); + let mut coordinator = coordinator(&store); + let request = request_with("forged-frontier", 8, 64); + record( + &mut store, + &mut coordinator, + request, + "request:forged-frontier", + ); + let mut report = must_ok(recover_in_memory_store( + &mut store, + RecoveryAccessMode::ReadOnly, + )); + match report.transactions.first_mut() { + Some(transaction) => { + transaction.commit.affected_frontiers_root = digest("forged-frontier-root"); + } + None => panic!("request transaction was not recovered"), + } + assert!(matches!( + observe_external_actions(&report), + Err(ExternalActionProtocolErrorV1::ExternalActionFrontierMismatch { .. }) + )); +} + +#[test] +fn malformed_schema_digest_and_oversized_settlements_fail_closed() { + for (label, mutation, expected) in [ + ( + "schema", + 1_u8, + ExternalActionProtocolErrorV1::SettlementSchemaMismatch, + ), + ( + "digest", + 2_u8, + ExternalActionProtocolErrorV1::SettlementResultDigestMismatch, + ), + ( + "budget", + 3_u8, + ExternalActionProtocolErrorV1::SettlementBudgetExceeded, + ), + ( + "schema-evidence", + 4_u8, + ExternalActionProtocolErrorV1::MissingSchemaAdmissionEvidence, + ), + ] { + let mut store = store(); + let mut coordinator = coordinator(&store); + let request = request_with(label, 9, 4); + let recorded = record( + &mut store, + &mut coordinator, + request, + &format!("request:{label}"), + ); + let grant = claim( + &mut store, + &mut coordinator, + recorded, + &format!("claim:{label}"), + ); + let mut candidate = candidate( + &grant, + ExternalActionSettlementKindV1::Succeeded, + b"four".to_vec(), + ); + match mutation { + 1 => candidate.settlement_schema_digest = digest("wrong-schema"), + 2 => candidate.declared_result_digest = digest("wrong-result"), + 3 => candidate.canonical_result_bytes.push(b'!'), + 4 => candidate.schema_admission_evidence_digest = [0; 32], + _ => unreachable!(), + } + let commits_before = store.read_commits().len(); + assert_eq!( + admit_external_action_settlement( + &mut store, + &mut coordinator, + context(&format!("settlement:{label}")), + grant, + candidate, + ), + Err(expected) + ); + assert_eq!(store.read_commits().len(), commits_before); + } +} + +#[test] +fn request_and_attempt_budget_boundaries_obstruct_before_commit() { + assert_eq!( + ExternalActionRequestV1::new( + WorldlineId::from_bytes([19; 32]), + ExternalActionOperationIdV1::from_hash(digest("workspace.observe@1")), + digest("workspace.observe@1.input"), + digest("workspace.observe@1.settlement"), + digest("workspace:/bounded"), + digest("basis:empty-budget"), + ExternalActionBudgetV1 { + max_settlement_bytes: 0, + max_attempts: 1, + }, + digest("input:empty-budget"), + digest("workspace.observe@1.reconcile"), + ), + Err(ExternalActionProtocolErrorV1::EmptyBudget) + ); + assert_eq!( + ExternalActionRequestV1::new( + WorldlineId::from_bytes([19; 32]), + ExternalActionOperationIdV1::from_hash(digest("workspace.observe@1")), + digest("workspace.observe@1.input"), + digest("workspace.observe@1.settlement"), + digest("workspace:/bounded"), + digest("basis:oversized-budget"), + ExternalActionBudgetV1 { + max_settlement_bytes: + warp_core::external_action::MAX_EXTERNAL_ACTION_SETTLEMENT_BYTES_V1 + 1, + max_attempts: 1, + }, + digest("input:oversized-budget"), + digest("workspace.observe@1.reconcile"), + ), + Err(ExternalActionProtocolErrorV1::RequestBudgetLimitExceeded) + ); + assert_eq!( + ExternalActionRequestV1::new( + WorldlineId::from_bytes([19; 32]), + ExternalActionOperationIdV1::from_hash(digest("workspace.observe@1")), + digest("workspace.observe@1.input"), + digest("workspace.observe@1.settlement"), + digest("workspace:/bounded"), + digest("basis:multi-attempt-budget"), + ExternalActionBudgetV1 { + max_settlement_bytes: 64, + max_attempts: 2, + }, + digest("input:multi-attempt-budget"), + digest("workspace.observe@1.reconcile"), + ), + Err(ExternalActionProtocolErrorV1::UnsupportedAttemptBudget) + ); + + let mut store = store(); + let mut coordinator = coordinator(&store); + let request = request_with("attempt-budget", 19, 64); + let recorded = record( + &mut store, + &mut coordinator, + request, + "request:attempt-budget", + ); + let commits_before = store.read_commits().len(); + assert_eq!( + claim_external_action( + &mut store, + &mut coordinator, + context("claim:attempt-budget"), + recorded, + authorization(&request), + request.basis_digest, + 1, + digest("claim:attempt-budget:lease"), + ), + Err(ExternalActionProtocolErrorV1::AttemptBudgetExhausted) + ); + assert_eq!(store.read_commits().len(), commits_before); +} + +#[test] +fn second_claim_for_one_request_is_obstructed_without_commit() { + let mut store = store(); + let mut coordinator = coordinator(&store); + let request = request_with("duplicate-claim", 19, 64); + let recorded = record( + &mut store, + &mut coordinator, + request, + "request:duplicate-claim", + ); + let duplicate_token = must_ok(coordinator.recorded_request(request.request_id())); + let _grant = claim( + &mut store, + &mut coordinator, + recorded, + "claim:duplicate-claim:first", + ); + let commits_before = store.commit_count(); + assert_eq!( + claim_external_action( + &mut store, + &mut coordinator, + context("claim:duplicate-claim:second"), + duplicate_token, + authorization(&request), + request.basis_digest, + 0, + digest("claim:duplicate-claim:second:lease"), + ), + Err(ExternalActionProtocolErrorV1::DuplicateClaim) + ); + assert_eq!(store.commit_count(), commits_before); +} + +#[test] +fn recovery_distinguishes_unclaimed_claimed_settled_and_ambiguous_requests() { + let mut store = store(); + let mut coordinator = coordinator(&store); + + let requested = request_with("requested", 10, 64); + record(&mut store, &mut coordinator, requested, "request:requested"); + + let claimed = request_with("claimed", 10, 64); + let claimed_recorded = record(&mut store, &mut coordinator, claimed, "request:claimed"); + claim( + &mut store, + &mut coordinator, + claimed_recorded, + "claim:claimed", + ); + + let settled = request_with("settled", 10, 64); + let settled_recorded = record(&mut store, &mut coordinator, settled, "request:settled"); + let settled_grant = claim( + &mut store, + &mut coordinator, + settled_recorded, + "claim:settled", + ); + let settled_candidate = candidate( + &settled_grant, + ExternalActionSettlementKindV1::Succeeded, + b"settled".to_vec(), + ); + must_ok(admit_external_action_settlement( + &mut store, + &mut coordinator, + context("settlement:settled"), + settled_grant, + settled_candidate, + )); + + let ambiguous = request_with("ambiguous", 10, 64); + let ambiguous_recorded = record(&mut store, &mut coordinator, ambiguous, "request:ambiguous"); + let ambiguous_grant = claim( + &mut store, + &mut coordinator, + ambiguous_recorded, + "claim:ambiguous", + ); + let ambiguous_candidate = candidate( + &ambiguous_grant, + ExternalActionSettlementKindV1::OutcomeUnknown, + b"connection-lost".to_vec(), + ); + must_ok(admit_external_action_settlement( + &mut store, + &mut coordinator, + context("settlement:ambiguous"), + ambiguous_grant, + ambiguous_candidate, + )); + + let report = must_ok(recover_in_memory_store( + &mut store, + RecoveryAccessMode::ReadOnly, + )); + let index = must_ok(observe_external_actions(&report)); + assert_eq!(index.len(), 4); + assert_eq!( + index.get(requested.request_id()).map(|entry| entry.posture), + Some(RecoveredExternalActionPostureV1::Requested) + ); + assert_eq!( + index.get(claimed.request_id()).map(|entry| entry.posture), + Some(RecoveredExternalActionPostureV1::Claimed) + ); + assert_eq!( + index.get(settled.request_id()).map(|entry| entry.posture), + Some(RecoveredExternalActionPostureV1::Settled( + ExternalActionSettlementKindV1::Succeeded + )) + ); + assert_eq!( + index.get(ambiguous.request_id()).map(|entry| entry.posture), + Some(RecoveredExternalActionPostureV1::Settled( + ExternalActionSettlementKindV1::OutcomeUnknown + )) + ); +} + +#[test] +fn replay_returns_admitted_bytes_without_reissuing_an_effect() { + let mut store = store(); + let mut coordinator = coordinator(&store); + let request = request_with("replay", 11, 64); + let recorded = record(&mut store, &mut coordinator, request, "request:replay"); + let grant = claim(&mut store, &mut coordinator, recorded, "claim:replay"); + let candidate = candidate( + &grant, + ExternalActionSettlementKindV1::Succeeded, + b"recorded-result".to_vec(), + ); + must_ok(admit_external_action_settlement( + &mut store, + &mut coordinator, + context("settlement:replay"), + grant, + candidate, + )); + + let report = must_ok(recover_in_memory_store( + &mut store, + RecoveryAccessMode::ReadOnly, + )); + let recovered_coordinator = must_ok(ExternalActionCoordinatorV1::recover(&store)); + let admitted = must_ok(recovered_coordinator.admitted_settlement(request.request_id())); + assert_eq!( + admitted.settlement().canonical_result_bytes.as_slice(), + b"recorded-result" + ); + let index = must_ok(observe_external_actions(&report)); + let recovered = match index.get(request.request_id()) { + Some(recovered) => recovered, + None => panic!("request was not recovered"), + }; + assert_eq!( + recovered + .settlement + .as_ref() + .map(|value| value.canonical_result_bytes.as_slice()), + Some(b"recorded-result".as_slice()) + ); +} + +#[test] +fn trusted_recovery_reconstructs_each_interrupted_transition_grant() { + let mut store = store(); + let request = request_with("recover-grants", 11, 64); + + let request_commit_digest = { + let mut request_coordinator = coordinator(&store); + let recorded = record( + &mut store, + &mut request_coordinator, + request, + "request:recover-grants", + ); + recorded.request_commit_digest() + }; + + let claim_commit_digest = { + let mut claim_coordinator = must_ok(ExternalActionCoordinatorV1::recover(&store)); + let recovered_request = must_ok(claim_coordinator.recorded_request(request.request_id())); + assert_eq!( + recovered_request.request_commit_digest(), + request_commit_digest + ); + let grant = claim( + &mut store, + &mut claim_coordinator, + recovered_request, + "claim:recover-grants", + ); + grant.claim_commit_digest() + }; + + let settlement_commit_digest = { + let mut settlement_coordinator = must_ok(ExternalActionCoordinatorV1::recover(&store)); + let recovered_grant = must_ok(settlement_coordinator.claim_grant(request.request_id())); + assert_eq!(recovered_grant.claim_commit_digest(), claim_commit_digest); + let candidate = candidate( + &recovered_grant, + ExternalActionSettlementKindV1::Succeeded, + b"recovered-result".to_vec(), + ); + let admitted = must_ok(admit_external_action_settlement( + &mut store, + &mut settlement_coordinator, + context("settlement:recover-grants"), + recovered_grant, + candidate, + )); + admitted.settlement_commit_digest() + }; + + let recovered_coordinator = must_ok(ExternalActionCoordinatorV1::recover(&store)); + let resumed = must_ok(recovered_coordinator.admitted_settlement(request.request_id())); + assert_eq!(resumed.settlement_commit_digest(), settlement_commit_digest); + assert_eq!( + resumed.settlement().canonical_result_bytes.as_slice(), + b"recovered-result" + ); +} + +#[test] +fn filesystem_reopen_recovers_settlement_without_adapter_reexecution() { + let wal_dir = TempWalDir::new("reopen"); + let request = request_with("filesystem-reopen", 21, 128); + let result_bytes = b"durable observed bytes".to_vec(); + { + let mut store = must_ok(FilesystemWalStore::open( + &wal_dir.0, + WalSegmentId::from_raw(1), + )); + must_ok(store.acquire_writer_epoch(WriterEpochRequest { + epoch_id: epoch_id(), + storage_fencing_token: digest("external-action:fencing"), + process_identity: digest("external-action:process"), + host_identity: digest("external-action:host"), + started_at_lsn: Lsn::from_raw(0), + previous_epoch_id: None, + previous_epoch_final_commit_digest: None, + lease_or_lock_evidence: digest("external-action:lease"), + })); + let mut coordinator = coordinator(&store); + let recorded = must_ok(record_external_action_request( + &mut store, + &mut coordinator, + context_with_durability( + "request:filesystem-reopen", + WalDurabilityMode::StrictFilesystem, + ), + request, + )); + let grant = must_ok(claim_external_action( + &mut store, + &mut coordinator, + context_with_durability( + "claim:filesystem-reopen", + WalDurabilityMode::StrictFilesystem, + ), + recorded, + authorization(&request), + request.basis_digest, + 0, + digest("claim:filesystem-reopen:lease"), + )); + let candidate = candidate( + &grant, + ExternalActionSettlementKindV1::Succeeded, + result_bytes.clone(), + ); + must_ok(admit_external_action_settlement( + &mut store, + &mut coordinator, + context_with_durability( + "settlement:filesystem-reopen", + WalDurabilityMode::StrictFilesystem, + ), + grant, + candidate, + )); + } + + let report = must_ok(recover_filesystem_store( + &wal_dir.0, + RecoveryAccessMode::ReadOnly, + )); + assert_eq!(report.tail_posture, RecoveryTailPosture::Clean); + let index = must_ok(observe_external_actions(&report)); + let recovered = match index.get(request.request_id()) { + Some(recovered) => recovered, + None => panic!("filesystem settlement was not recovered"), + }; + assert_eq!( + recovered.posture, + RecoveredExternalActionPostureV1::Settled(ExternalActionSettlementKindV1::Succeeded) + ); + assert_eq!( + recovered + .settlement + .as_ref() + .map(|settlement| settlement.canonical_result_bytes.as_slice()), + Some(result_bytes.as_slice()) + ); +} + +#[test] +fn filesystem_scan_failure_cannot_be_admitted_as_genesis() { + let wal_dir = TempWalDir::new("corrupt-snapshot"); + let store = must_ok(FilesystemWalStore::open( + &wal_dir.0, + WalSegmentId::from_raw(1), + )); + must_ok(std::fs::write(store.segment_path(), b"not-a-wal-segment")); + + assert!(matches!( + ExternalActionCoordinatorV1::recover(&store), + Err(ExternalActionProtocolErrorV1::WalStore(_)) + )); +} + +#[test] +fn request_identity_is_deterministic_and_worldline_scoped() { + let first = request_with("identity", 12, 64); + let same = request_with("identity", 12, 64); + let fork = request_with("identity", 13, 64); + assert_eq!(first.request_id(), same.request_id()); + assert_ne!(first.request_id(), fork.request_id()); +} + +#[test] +fn raw_wal_builder_cannot_mint_external_action_authority() { + let mut builder = raw_builder( + "raw-external-action", + 0, + WalTransactionKind::ExternalActionRequest, + ); + assert_eq!( + builder.push_record(WalRecordKind::ExternalActionRequestRecorded, Vec::new()), + Err(WalBuildError::ExternalActionCoordinatorCapabilityRequired) + ); +} + +#[test] +fn lifecycle_index_root_is_independent_of_request_insertion_order() { + let first = request_with("index-order:first", 22, 64); + let second = request_with("index-order:second", 22, 64); + let mut left = store(); + let mut left_coordinator = coordinator(&left); + record( + &mut left, + &mut left_coordinator, + first, + "request:index-order:left:first", + ); + record( + &mut left, + &mut left_coordinator, + second, + "request:index-order:left:second", + ); + let mut right = store(); + let mut right_coordinator = coordinator(&right); + record( + &mut right, + &mut right_coordinator, + second, + "request:index-order:right:second", + ); + record( + &mut right, + &mut right_coordinator, + first, + "request:index-order:right:first", + ); + + let left_report = must_ok(recover_in_memory_store( + &mut left, + RecoveryAccessMode::ReadOnly, + )); + let right_report = must_ok(recover_in_memory_store( + &mut right, + RecoveryAccessMode::ReadOnly, + )); + assert_eq!( + must_ok(observe_external_actions(&left_report)).root_digest(), + must_ok(observe_external_actions(&right_report)).root_digest() + ); +} + +#[test] +fn fixed_seed_request_property_round_trips_unique_identities() { + const SEED: u64 = 0x5eed_cafe_f00d_beef; + const CASES: usize = 32; + let mut state = SEED; + let mut store = store(); + let mut coordinator = coordinator(&store); + let mut request_ids = std::collections::BTreeSet::new(); + for index in 0..CASES { + state = state + .wrapping_mul(6_364_136_223_846_793_005) + .wrapping_add(1_442_695_040_888_963_407); + let request = request_with(&format!("property:{state:016x}"), 14, 64); + assert!(request_ids.insert(request.request_id())); + record( + &mut store, + &mut coordinator, + request, + &format!("request:property:{index}"), + ); + } + let report = must_ok(recover_in_memory_store( + &mut store, + RecoveryAccessMode::ReadOnly, + )); + let index = must_ok(observe_external_actions(&report)); + assert_eq!(index.len(), CASES); +} + +#[test] +fn bounded_stress_recovers_all_requests_without_adapter_execution() { + const REQUESTS: usize = 64; + let mut store = store(); + let mut coordinator = coordinator(&store); + for index in 0..REQUESTS { + let request = request_with(&format!("stress:{index}"), 15, 64); + record( + &mut store, + &mut coordinator, + request, + &format!("request:stress:{index}"), + ); + } + let report = must_ok(recover_in_memory_store( + &mut store, + RecoveryAccessMode::ReadOnly, + )); + let index = must_ok(observe_external_actions(&report)); + assert_eq!(index.len(), REQUESTS); +} + +#[test] +fn duplicate_settlement_is_a_recovery_obstruction() { + let mut store = store(); + let mut coordinator = coordinator(&store); + let request = request_with("duplicate", 16, 64); + let recorded = record(&mut store, &mut coordinator, request, "request:duplicate"); + let grant = claim(&mut store, &mut coordinator, recorded, "claim:duplicate"); + let first = candidate( + &grant, + ExternalActionSettlementKindV1::Succeeded, + b"first".to_vec(), + ); + must_ok(admit_external_action_settlement( + &mut store, + &mut coordinator, + context("settlement:duplicate"), + grant, + first, + )); + let mut report = must_ok(recover_in_memory_store( + &mut store, + RecoveryAccessMode::ReadOnly, + )); + let duplicate = match report.transactions.last().cloned() { + Some(transaction) => transaction, + None => panic!("settlement transaction was not recovered"), + }; + report.transactions.push(duplicate); + assert_eq!( + observe_external_actions(&report), + Err(ExternalActionProtocolErrorV1::DuplicateSettlement) + ); +} + +#[derive(Debug)] +struct SnapshotCountingStore { + inner: InMemoryWalStore, + snapshot_reads: Cell, +} + +impl WalStorePort for SnapshotCountingStore { + fn acquire_writer_epoch( + &mut self, + request: WriterEpochRequest, + ) -> Result { + self.inner.acquire_writer_epoch(request) + } + + fn append_frame( + &mut self, + epoch_id: WriterEpochId, + frame: WalFrame, + ) -> Result<(), WalStoreError> { + self.inner.append_frame(epoch_id, frame) + } + + fn flush_commit( + &mut self, + epoch_id: WriterEpochId, + commit: WalTransactionCommit, + ) -> Result<(), WalStoreError> { + self.inner.flush_commit(epoch_id, commit) + } + + fn flush_external_action_commit( + &mut self, + epoch_id: WriterEpochId, + commit: WalTransactionCommit, + capability: ExternalActionCoordinatorCapability, + ) -> Result<(), WalStoreError> { + self.inner + .flush_external_action_commit(epoch_id, commit, capability) + } + + fn read_frames(&self) -> Vec { + self.inner.read_frames() + } + + fn read_commits(&self) -> Vec { + self.inner.read_commits() + } + + fn read_snapshot(&self) -> Result { + self.snapshot_reads + .set(self.snapshot_reads.get().saturating_add(1)); + self.inner.read_snapshot() + } + + fn seal_segment( + &mut self, + epoch_id: WriterEpochId, + segment_id: WalSegmentId, + ) -> Result { + self.inner.seal_segment(epoch_id, segment_id) + } + + fn truncate_tail_after(&mut self, after_lsn: Lsn) -> Result<(), WalStoreError> { + self.inner.truncate_tail_after(after_lsn) + } + + fn publish_manifest( + &mut self, + epoch_id: WriterEpochId, + manifest: WalManifest, + ) -> Result<(), WalStoreError> { + self.inner.publish_manifest(epoch_id, manifest) + } + + fn close_epoch(&mut self, epoch_id: WriterEpochId) -> Result<(), WalStoreError> { + self.inner.close_epoch(epoch_id) + } +} + +#[test] +fn hot_path_advances_the_recovered_index_without_replaying_the_wal() { + let mut store = SnapshotCountingStore { + inner: store(), + snapshot_reads: Cell::new(0), + }; + let mut coordinator = coordinator(&store); + assert_eq!(store.snapshot_reads.get(), 1); + + let request = request_with("incremental-index", 16, 64); + let recorded = record( + &mut store, + &mut coordinator, + request, + "request:incremental-index", + ); + let grant = claim( + &mut store, + &mut coordinator, + recorded, + "claim:incremental-index", + ); + let candidate = candidate( + &grant, + ExternalActionSettlementKindV1::Succeeded, + b"incremental".to_vec(), + ); + must_ok(admit_external_action_settlement( + &mut store, + &mut coordinator, + context("settlement:incremental-index"), + grant, + candidate, + )); + + assert_eq!(store.snapshot_reads.get(), 1); + assert_eq!(coordinator.observed_index().len(), 1); +} + +#[derive(Debug)] +struct CommitFailingStore { + inner: InMemoryWalStore, + fail_on_commit_ordinal: usize, +} + +impl WalStorePort for CommitFailingStore { + fn acquire_writer_epoch( + &mut self, + request: WriterEpochRequest, + ) -> Result { + self.inner.acquire_writer_epoch(request) + } + + fn append_frame( + &mut self, + epoch_id: WriterEpochId, + frame: WalFrame, + ) -> Result<(), WalStoreError> { + self.inner.append_frame(epoch_id, frame) + } + + fn flush_commit( + &mut self, + epoch_id: WriterEpochId, + commit: WalTransactionCommit, + ) -> Result<(), WalStoreError> { + if self.inner.commit_count() == self.fail_on_commit_ordinal { + Err(WalStoreError::Io( + "injected external-action commit failure".to_owned(), + )) + } else { + self.inner.flush_commit(epoch_id, commit) + } + } + + fn flush_external_action_commit( + &mut self, + epoch_id: WriterEpochId, + commit: WalTransactionCommit, + capability: ExternalActionCoordinatorCapability, + ) -> Result<(), WalStoreError> { + if self.inner.commit_count() == self.fail_on_commit_ordinal { + Err(WalStoreError::Io( + "injected external-action commit failure".to_owned(), + )) + } else { + self.inner + .flush_external_action_commit(epoch_id, commit, capability) + } + } + + fn read_frames(&self) -> Vec { + self.inner.read_frames() + } + + fn read_commits(&self) -> Vec { + self.inner.read_commits() + } + + fn seal_segment( + &mut self, + epoch_id: WriterEpochId, + segment_id: WalSegmentId, + ) -> Result { + self.inner.seal_segment(epoch_id, segment_id) + } + + fn truncate_tail_after(&mut self, after_lsn: Lsn) -> Result<(), WalStoreError> { + self.inner.truncate_tail_after(after_lsn) + } + + fn publish_manifest( + &mut self, + epoch_id: WriterEpochId, + manifest: WalManifest, + ) -> Result<(), WalStoreError> { + self.inner.publish_manifest(epoch_id, manifest) + } + + fn close_epoch(&mut self, epoch_id: WriterEpochId) -> Result<(), WalStoreError> { + self.inner.close_epoch(epoch_id) + } +} + +#[test] +fn failed_request_commit_exposes_no_adapter_reachable_token() { + let mut store = CommitFailingStore { + inner: store(), + fail_on_commit_ordinal: 0, + }; + let mut coordinator = coordinator(&store); + let request = request_with("commit-failure", 17, 64); + assert_eq!( + record_external_action_request( + &mut store, + &mut coordinator, + context("request:commit-failure"), + request, + ), + Err(ExternalActionProtocolErrorV1::WalStore(WalStoreError::Io( + "injected external-action commit failure".to_owned() + ))) + ); + assert_eq!(store.read_commits().len(), 0); + assert_eq!(store.read_frames().len(), 1); + let report = must_ok(recover_from_frames_and_commits( + &store.read_frames(), + &store.read_commits(), + RecoveryAccessMode::ReadOnly, + )); + assert_eq!(report.tail_posture, RecoveryTailPosture::WouldTruncateAll); + assert!(must_ok(observe_external_actions(&report)).is_empty()); + assert_eq!( + coordinator.recorded_request(request.request_id()), + Err(ExternalActionProtocolErrorV1::CoordinatorRecoveryRequired) + ); +} + +#[test] +fn failed_claim_commit_exposes_no_adapter_work_grant() { + let mut store = CommitFailingStore { + inner: store(), + fail_on_commit_ordinal: 1, + }; + let mut coordinator = coordinator(&store); + let request = request_with("claim-commit-failure", 17, 64); + let recorded = must_ok(record_external_action_request( + &mut store, + &mut coordinator, + context("request:claim-commit-failure"), + request, + )); + assert_eq!( + claim_external_action( + &mut store, + &mut coordinator, + context("claim:commit-failure"), + recorded, + authorization(&request), + request.basis_digest, + 0, + digest("claim:commit-failure:lease"), + ), + Err(ExternalActionProtocolErrorV1::WalStore(WalStoreError::Io( + "injected external-action commit failure".to_owned() + ))) + ); + assert_eq!(store.read_commits().len(), 1); + assert_eq!(store.read_frames().len(), 2); +} + +#[test] +fn failed_settlement_commit_exposes_no_resumable_fact() { + let mut store = CommitFailingStore { + inner: store(), + fail_on_commit_ordinal: 2, + }; + let mut coordinator = coordinator(&store); + let request = request_with("settlement-commit-failure", 17, 64); + let recorded = must_ok(record_external_action_request( + &mut store, + &mut coordinator, + context("request:settlement-commit-failure"), + request, + )); + let grant = must_ok(claim_external_action( + &mut store, + &mut coordinator, + context("claim:settlement-commit-failure"), + recorded, + authorization(&request), + request.basis_digest, + 0, + digest("claim:settlement-commit-failure:lease"), + )); + let candidate = candidate( + &grant, + ExternalActionSettlementKindV1::Succeeded, + b"uncommitted-result".to_vec(), + ); + assert_eq!( + admit_external_action_settlement( + &mut store, + &mut coordinator, + context("settlement:commit-failure"), + grant, + candidate, + ), + Err(ExternalActionProtocolErrorV1::WalStore(WalStoreError::Io( + "injected external-action commit failure".to_owned() + ))) + ); + assert_eq!(store.read_commits().len(), 2); + assert_eq!(store.read_frames().len(), 3); +} + +#[test] +fn malformed_committed_settlement_payload_is_rejected() { + let mut store = store(); + let mut coordinator = coordinator(&store); + let request = request_with("malformed-payload", 18, 64); + let recorded = record( + &mut store, + &mut coordinator, + request, + "request:malformed-payload", + ); + let grant = claim( + &mut store, + &mut coordinator, + recorded, + "claim:malformed-payload", + ); + let candidate = candidate( + &grant, + ExternalActionSettlementKindV1::Succeeded, + b"retained".to_vec(), + ); + must_ok(admit_external_action_settlement( + &mut store, + &mut coordinator, + context("settlement:malformed-payload"), + grant, + candidate, + )); + let mut report = must_ok(recover_in_memory_store( + &mut store, + RecoveryAccessMode::ReadOnly, + )); + let settlement_frame = match report + .transactions + .last_mut() + .and_then(|transaction| transaction.frames.first_mut()) + { + Some(frame) => frame, + None => panic!("settlement frame was not recovered"), + }; + settlement_frame.payload.canonical_bytes.truncate(7); + assert!(matches!( + observe_external_actions(&report), + Err(ExternalActionProtocolErrorV1::Decode( + warp_core::causal_wal::WalDecodeError::UnexpectedEof + )) + )); +} + +#[test] +fn external_action_wal_codes_and_frontier_are_stable() { + assert_eq!(WalTransactionKind::ExternalActionRequest.stable_code(), 10); + assert_eq!(WalTransactionKind::ExternalActionClaim.stable_code(), 11); + assert_eq!( + WalTransactionKind::ExternalActionSettlement.stable_code(), + 12 + ); + assert_eq!( + warp_core::causal_wal::WalRecordKind::ExternalActionRequestRecorded.stable_code(), + 29 + ); + assert_eq!( + warp_core::causal_wal::WalRecordKind::ExternalActionClaimRecorded.stable_code(), + 30 + ); + assert_eq!( + warp_core::causal_wal::WalRecordKind::ExternalActionSettlementRecorded.stable_code(), + 31 + ); + assert_eq!(AffectedFrontierKind::ExternalActionIndex.stable_code(), 11); +} diff --git a/docs/README.md b/docs/README.md index 4562084c..db5e3516 100644 --- a/docs/README.md +++ b/docs/README.md @@ -56,6 +56,7 @@ causal history. Git history is the archive; GitHub owns live work and status. - [Admitted executable-operation packages](adr/0023-admitted-executable-operation-packages.md) - [Anchored-node creation from absence](adr/0024-anchored-node-creation-from-absence.md) - [Scheduler-owned executable-operation Actions](adr/0025-scheduler-owned-executable-operation-actions.md) +- [Durable external-action settlement](adr/0026-durable-external-action-settlement.md) ## Normative Contracts diff --git a/docs/adr/0026-durable-external-action-settlement.md b/docs/adr/0026-durable-external-action-settlement.md new file mode 100644 index 00000000..be9694dd --- /dev/null +++ b/docs/adr/0026-durable-external-action-settlement.md @@ -0,0 +1,223 @@ + + + +# ADR 0026: Durable External-Action Settlement + +- **Status:** Accepted +- **Date:** 2026-07-29 + +## Context + +Deterministic application law, external interaction, and model judgment are +different authority categories. A filesystem read, process invocation, network +request, timer, or model call is mechanical but not deterministic over Echo +history. Granting those operations directly to Edict or to the +compiler/provider seam would add ambient authority and make replay consult a +different external world. + +Witnessing only the inbound result is insufficient. A crash after an external +system accepts an operation but before Echo records it leaves no durable answer +to whether the operation happened. The outbound request must enter causal +history before an adapter can act. + +## Decision + +Echo owns a domain-neutral external-action protocol: + +```text +Edict decision + -> canonical request + -> REQUESTED WAL commit + -> bounded CLAIMED WAL commit + -> operation-specific adapter + -> schema-bound settlement candidate + -> SETTLED WAL commit + -> deterministic resumption +``` + +Edict constructs request values. It does not execute external effects. Echo +records and schedules the boundary crossing. An operation-specific adapter +alone possesses the external credential or host access. The settlement returns +as witnessed ingress. + +### Request identity + +`ExternalActionRequestV1` commits: + +1. the originating `WorldlineId`; +2. a declared operation-family identity; +3. input and settlement schema digests; +4. the requested authority-scope digest; +5. the exact current-world basis digest; +6. a maximum retained-settlement byte budget; +7. a v1 attempt budget fixed to exactly one claim; +8. the canonical input digest; and +9. a named reconciliation-law digest. + +The request id is a domain-separated BLAKE3 commitment over those fields. +Changing the worldline, basis, operation, schema, scope, budget, input, or +reconciliation law creates a different request. + +The 1 MiB v1 retained-settlement ceiling is an absolute Echo limit, not an +application default. Operation profiles should delegate smaller bounds. + +### Claim and authority + +The runtime owner installs an `ExternalActionAdapterRegistryV1` containing +operation-, scope-, and adapter-specific bindings. Registry lookup attenuates +that policy into an authorization bound to the exact request id, request basis, +and canonical registry-policy digest. An authorization for another request +cannot be replayed even when operation and scope match. Application code and +Edict receive neither the registry nor the adapter's external credential. + +A claim commits the request id, attempt id, zero-based attempt ordinal, adapter +identity, lease or fencing evidence, request-stable idempotency key, +reconciliation law, basis, and registry-policy digest. The attempt identity +binds the registry policy and nonzero lease evidence. The adapter work grant +becomes constructible only after the claim transaction commits. + +Protocol v1 admits exactly one claim per request. Recovery of `CLAIMED` is a +reconciliation obligation, not permission to repeat the operation. A retry is +a new deterministic decision and request. This keeps blind re-execution +unrepresentable while later versions establish operation-specific retry laws. + +### Settlement + +The four terminal settlement kinds are: + +- `Succeeded`; +- `Rejected`; +- `Failed`; and +- `OutcomeUnknown`. + +`OutcomeUnknown` is not collapsed into failure. It states that the adapter +cannot establish whether the effect occurred. The named reconciliation law and +request-stable idempotency key remain available for a later explicit decision. + +A settlement binds the exact request, attempt, adapter, basis, settlement +schema, canonical result bytes, result digest, schema-admission evidence, and +external evidence. Echo rejects mismatched claims, stale bases, wrong schemas, +missing schema-admission or external evidence, digest substitution, oversize +results, duplicate settlements, conflicting settlements, malformed payloads, +and unknown outcome codes. + +The schema-admission evidence is a protocol binding, not a general schema +engine. Each operation profile must define which validator produces that +evidence. The bounded workspace-observation profile will supply the first +concrete validator. + +### WAL ownership + +Three transaction kinds own the transitions: + +| Transition | Transaction code | Record code | Frontier | +| -------------------- | ---------------: | ----------: | --------------------- | +| request admission | 10 | 29 | `ExternalActionIndex` | +| claim admission | 11 | 30 | `ExternalActionIndex` | +| settlement admission | 12 | 31 | `ExternalActionIndex` | + +All three require `ExternalActionCoordinator` append authority. Each +transaction contains exactly one matching record and advances exactly one +external-action frontier. A frame without its transaction commit is not a +request, claim, or settlement. + +Raw WAL builders and raw commit flushes do not carry the coordinator's opaque +capability. A high-level caller supplies only non-causal transaction metadata. +The coordinator derives the next LSN, previous-frame digest, and +previous-commit digest from its checked local WAL continuation. A caller +therefore cannot manufacture coordinator authority or select transaction +coordinates that make a successful append unrecoverable. + +The coordinator derives both frontier roots from its canonical lifecycle +index. The root is a domain-separated sparse Merkle commitment keyed by request +id, so insertion order cannot move the reading and one lifecycle update touches +one bounded 256-bit path. One planned mutation computes and retains that path, +then advances the in-memory index only after commit. The request, claim, and +settlement hot paths do not replay prior WAL payloads. Full reconstruction is +reserved for initial or crash recovery. Callers cannot select frontier roots. +Recovery rebuilds the index around every transition and rejects a WAL commit +whose frontier commitment differs. + +The high-level APIs return a durable request token, adapter work grant, or +resumable settlement only after the corresponding commit marker flushes. A +commit failure returns no token and poisons that coordinator instance; trusted +local recovery is required before another transition. An uncommitted tail +obstructs further external-action admission until ordinary WAL recovery +resolves it. + +### Recovery and replay + +Recovery consumes only committed external-action records and reconstructs one +of: + +```text +REQUESTED +CLAIMED +SETTLED(SUCCEEDED | REJECTED | FAILED | OUTCOME_UNKNOWN) +``` + +Missing predecessors, repeated requests or claims, and duplicate or conflicting +settlements obstruct recovery. Settled canonical result bytes are replay input. +Replay does not invoke an adapter. Consulting the current external world again +requires a new program transition and a new request; changing worldline or +basis changes request identity. + +An arbitrary `RecoveryScanReport` produces observation-only lifecycle values. +It cannot mint request-transition tokens, adapter work grants, or resumable +settlement facts. `ExternalActionCoordinatorV1::recover` alone reads one +fallible, coherent local-store snapshot, validates its clean committed history, +and reconstructs those authorities. A crash after request or claim commit can +therefore resume from the durable lifecycle without an ephemeral native return +value. + +Filesystem snapshot read or decode failure is an obstruction, never an empty +genesis history. The filesystem recovery witness drops the live store, reopens +its strict filesystem WAL, and recovers the exact settled bytes without +invoking adapter execution. + +## Consequences + +- The compiler/provider seam remains deterministic and capability-denied. +- Edict may request an operation family without receiving the authority to + perform it. +- Adapter credentials and host capabilities stay outside Edict and the model. +- Request-before-effect and settlement-before-resumption are type-visible API + boundaries backed by WAL commits. +- Arbitrary recovered reports remain observation-only; trusted local recovery + owns transition and replay authority. +- Crash ambiguity has an explicit causal representation. +- Operation-specific idempotency and reconciliation laws remain mandatory; + Echo does not claim general exactly-once external execution. +- The current protocol proves the durable lifecycle without implementing a + filesystem, process, network, Git, GitHub, timer, or model adapter. + +## Rejected Alternatives + +### Callable imports in the provider seam + +Rejected. Lowering and independent verification would acquire execution-host +authority and replay could perform new I/O. + +### Generic shell, filesystem, or network capabilities + +Rejected. These are ambient authority with type names. Adapters must expose +domain-specific operations and validation laws. + +### Native stack suspension + +Rejected. Waiting is explicit durable protocol state keyed by request identity, +not a serialized host stack or hidden callback continuation. + +### Model-owned mutation capability + +Rejected. Model output is untrusted data. A model may propose a typed artifact; +deterministic validation and a separately authorized adapter govern mutation. + +## Evidence + +- `crates/warp-core/src/external_action.rs` +- `crates/warp-core/src/causal_wal.rs` +- `crates/warp-core/tests/external_action_protocol_tests.rs` +- [WAL](../topics/WAL.md) +- [Runtime Authority](../topics/RuntimeAuthority.md) +- [Security And Authority Boundaries](../topics/security/AuthorityBoundaries.md) diff --git a/docs/adr/README.md b/docs/adr/README.md index 4ca1995d..a971974c 100644 --- a/docs/adr/README.md +++ b/docs/adr/README.md @@ -47,6 +47,7 @@ track work, progress, priority, or release readiness. | [0023](0023-admitted-executable-operation-packages.md) | Accepted | Admitted executable operation packages | | [0024](0024-anchored-node-creation-from-absence.md) | Accepted | Anchored-node creation as a separate executable program | | [0025](0025-scheduler-owned-executable-operation-actions.md) | Accepted | Scheduler-owned executable-operation Actions | +| [0026](0026-durable-external-action-settlement.md) | Accepted | Request-before-effect and settlement-before-resumption | ADR 0006 predates this index contract and did not declare a status. Its superseded tombstone preserves that fact without silently ratifying the old diff --git a/docs/architecture/application-contract-hosting.md b/docs/architecture/application-contract-hosting.md index 83374dcb..1d321544 100644 --- a/docs/architecture/application-contract-hosting.md +++ b/docs/architecture/application-contract-hosting.md @@ -1122,3 +1122,23 @@ jedit renders buffers, cursors, diffs, diagnostics, history, and UI. Echo gives applications deterministic causal substrate, witnessed ingress, reading envelopes, product-shaped optic DTOs, registry identity, and retention hooks. It should not become the application. + +World-touching operations cross a distinct durable external-action boundary. +An Edict-authored decision produces a typed request value; Echo commits that +request before an operation-specific adapter can receive work authority. The +adapter alone holds the external credential or host access. Its result returns +as a schema-bound settlement candidate, and Echo commits the accepted +settlement before deterministic execution resumes. Replay consumes the +committed settlement and never repeats the effect. This protocol is outside the +compiler/provider seam: lowering and independent verification gain no +filesystem, process, network, timer, model, Git, or GitHub imports. + +Adapter authorization is a runtime-owned registry decision bound to the exact +request, basis, and registry-policy digest. Echo derives the lifecycle frontier +for every request, claim, and settlement transition; applications and adapters +cannot supply it. The locally recovered coordinator also derives WAL +continuation coordinates and carries the opaque authority required to flush +those commits. Arbitrary recovery reports are observation-only; they cannot +reconstruct adapter work grants or resumable settlement facts. Checked local +recovery validates the frontier commitment and storage snapshot before +restoring those authorities. diff --git a/docs/topics/RuntimeAuthority.md b/docs/topics/RuntimeAuthority.md index 46e1bfc4..6c47c6bc 100644 --- a/docs/topics/RuntimeAuthority.md +++ b/docs/topics/RuntimeAuthority.md @@ -34,6 +34,31 @@ read-only observers. They do not tick the runtime or invoke mutation handlers. Reading envelopes identify their basis, aperture, observer plan, budget, and evidence posture. +## External Actions + +- Edict may construct a typed request value; it receives no authority to + perform the operation. +- Echo commits `REQUESTED` before it can expose claimable work. +- Runtime-owner adapter registration attenuates operation and scope policy into + one request-specific authorization. +- Echo commits `CLAIMED` before it can expose an adapter work grant. +- Only an operation-specific adapter possesses filesystem, process, network, + timer, model, or other world-touching authority. +- Adapter output is untrusted settlement input until Echo validates its exact + request, attempt, adapter, basis, schema, digest, evidence, and budget + bindings. +- Raw WAL callers cannot mint `ExternalActionCoordinator` authority. The + coordinator owns an opaque commit capability and derives causal transaction + coordinates from one checked local continuation. +- Echo commits `SETTLED` before deterministic program resumption. +- Recovery of a claimed request requires reconciliation. It does not authorize + blind re-execution. +- Arbitrary recovery reports expose observation-only lifecycle values. Trusted + local coordinator recovery reconstructs interrupted transition grants. +- Replay consumes the admitted settlement bytes and never invokes the adapter. +- `OutcomeUnknown` is a first-class terminal observation, not an alias for + failure. + ## Host Boundary Trusted hosts may install verified generated packages, stage ticketed ingress, @@ -45,6 +70,7 @@ adapters. ## Evidence Anchors - [Registry/provider/host boundary](../adr/0015-registry-provider-host-boundary.md) +- [Durable external-action settlement](../adr/0026-durable-external-action-settlement.md) - `docs/architecture/application-contract-hosting.md` - `crates/warp-core/src/trusted_runtime_host.rs` - `crates/warp-core/src/engine_impl.rs` diff --git a/docs/topics/WAL.md b/docs/topics/WAL.md index dbde296e..1d84ca36 100644 --- a/docs/topics/WAL.md +++ b/docs/topics/WAL.md @@ -17,7 +17,7 @@ Echo may only claim what its WAL can recover. ## What We Found -The current runtime WAL evidence says ten concrete things. +The current runtime WAL evidence says eleven concrete things. First, accepted-submission evidence is not just an in-memory editor event. The WAL-backed ACK path, `submit_intent_with_runtime_wal_ack(...)`, returns only @@ -148,6 +148,44 @@ envelope, invocation identity, installed operation, scope, Tick entry, disposition, blocker set, composite patch, and final state root. Swapped or misclassified outcome evidence therefore fails closed. +Eleventh, external actions use a separate request, claim, and settlement +protocol under Echo-owned `ExternalActionCoordinator` authority. The canonical +request commits its originating worldline, operation family, input and +settlement schemas, authority scope, exact basis, retained-result and +single-claim budgets, input digest, and reconciliation law. Echo flushes the +request transaction before it can return a claimable token, and flushes the +claim transaction before it can return adapter work authority. The compiler, +provider seam, Edict program, and application receive no filesystem, process, +network, or model authority from those values. + +The runtime-owned registry decision is bound to the exact request, basis, and +canonical registry-policy digest. Every transition advances an +`ExternalActionIndex` frontier whose before and after roots Echo derives from +the complete lifecycle index through a request-id-keyed sparse Merkle +commitment. Recovery recomputes those roots and rejects caller-selected or +substituted frontier evidence. The live coordinator advances one planned +256-bit Merkle path after each successful commit; it does not replay the full +WAL on every transition. Full reconstruction is reserved for initial and crash +recovery. + +An admitted settlement binds the exact request, attempt, adapter, basis, +settlement schema, canonical result bytes, result digest, schema-admission +evidence, and nonzero external evidence. It has one of four explicit outcomes: +`Succeeded`, `Rejected`, `Failed`, or `OutcomeUnknown`. Echo flushes the +settlement before exposing the result to deterministic resumption. Recovery +reconstructs `REQUESTED`, `CLAIMED`, or the exact settled outcome using only +committed records. It rejects missing predecessors, duplicate requests or +claims, conflicting settlements, malformed payloads, wrong schemas, stale +bases, digest substitution, and budget overruns. A recovered claim is a +reconciliation obligation rather than permission to reissue an effect. +Settled replay, including strict filesystem WAL reopen, consumes retained +result bytes without consulting an adapter. Arbitrary recovery reports are +observation-only. Only a coordinator recovered from one fallible local-store +snapshot can reconstruct request tokens, claim grants, or resumable settlement +facts. Raw WAL builders and commit flushes lack its opaque capability, and the +coordinator derives LSN and predecessor coordinates from the validated local +continuation. + ## Boundaries The WAL belongs to the trusted runtime host. Application-facing code can submit @@ -173,6 +211,17 @@ The useful postures are: | `obstructed` | Recovery found accepted or decided evidence, but required material or consistency checks obstruct restoring the work. | | `recovery_faulted` | Required committed WAL evidence or retained material is missing or corrupt. | +External-action recovery has its own narrower lifecycle: + +| Posture | Meaning | +| -------------------------- | ------------------------------------------------------------------------------------- | +| `REQUESTED` | The request commit exists; no adapter claim is committed. | +| `CLAIMED` | One bounded claim exists; recovery must reconcile and must not blindly execute again. | +| `SETTLED(SUCCEEDED)` | The external postcondition was witnessed and admitted. | +| `SETTLED(REJECTED)` | The external system's typed refusal was witnessed and admitted. | +| `SETTLED(FAILED)` | A definite adapter failure was witnessed and admitted. | +| `SETTLED(OUTCOME_UNKNOWN)` | The external outcome is ambiguous and remains explicit causal evidence. | + An app such as `jedit` maps these generic postures into product language outside Echo. Echo should not grow editor, file, buffer, or dirty-state nouns in order to explain them. diff --git a/docs/topics/security/AuthorityBoundaries.md b/docs/topics/security/AuthorityBoundaries.md index ab0f5961..7d054fb0 100644 --- a/docs/topics/security/AuthorityBoundaries.md +++ b/docs/topics/security/AuthorityBoundaries.md @@ -68,30 +68,35 @@ support and durable evidence can be validated. ## Proposition Ledger -| Artifact or mechanism | Proposition it can support | What it does not prove | -| --------------------- | -------------------------- | ---------------------- | -| Canonical bytes | The value has one deterministic encoding under the named schema and version. | Validity under current policy, admission, authenticity, authorization, confidentiality, or availability. | -| Domain-separated BLAKE3 digest | The canonical bytes match the named digest domain, assuming collision resistance. | Who created the bytes, whether they are lawful, whether they are secret, or whether they are current. | -| Header or frame checksum | Accidental corruption is detectable relative to the checksum algorithm. | Adversarial authenticity or tamper resistance. | -| Content-addressed CAS id | Retrieved bytes match the addressed content hash. | Semantic coordinate, causal authority, retention, reveal permission, or future availability. | -| WSC payload | A deterministic physical representation can be decoded and validated under its profile. | Causal admission, WAL commit, recovery authority, or application meaning. | -| Storage locator or file path | A host suggests where bytes may be found. | Identity, integrity, durability, authority, or causal meaning. | -| Application proposal | A caller requested a specific operation against supplied coordinates. | Echo acceptance, eligibility, execution, or success. | -| Trusted admission support chain | A receipt-backed trusted-host result or recovery-validated record proves Echo admitted the exact claim under the bound basis and policy. | Application execution, domain success, reveal permission, or material availability. A copied public ticket or fact is not sufficient support. | -| Trusted-host or recovery-validated tick receipt | Echo committed the exact scheduler-owned outcome and evidence named by the receipt. | That a copied receipt value has trusted support, or that unrelated operations are authorized. | -| Causal parent reference | The child cites a specific admitted parent event. | Undo, redo, compensation, or any other domain semantics not supplied by the application contract. | -| Causal-anchor claim | A canonical caller claim binds a subject, supplied basis, roots, purpose, and schema. | Echo admission, root existence, root authority, retention pinning, or domain checkpoint validity. | -| Trusted-host or recovery-validated causal-anchor admission | Echo committed the exact claim at its current logical basis under the bound support-policy digest. | Physical retention, materialization availability, application-domain validity, or text mutation. A standalone copied fact or anchor id is not sufficient support. | -| Capability presentation | A caller cites a grant identity in a presentation shape. | That the grant exists, is authentic, is unexpired, covers the artifact/operation/scope, or authorizes the current request. | -| Validated capability grant | The named grant covers the validated identity requirements under the evaluated policy posture. | Runtime support, scheduler work, law execution, unlimited aperture, or automatic reveal of retained material. | -| Reading envelope | Echo reports the basis, observer plan, aperture, budget, rights, residual, and evidence posture for a reading. | Causal mutation authority or permission to widen the reading. | -| Proof opening | The named commitment proposition verified for the named coordinates under the named proof system. | Capability, admission, scheduler, WAL, recovery, or reveal authority. | -| Graph or hologram materialization | A derived view was computed from a named basis and support set. | Substrate truth, freshness at another basis, or validity for another observer. | -| Projection-cache hit | A cached artifact matches the complete cache key and coverage contract. | That an incomplete key is safe, that the observer is equivalent, or that source authority remains available. | -| WAL frame | A record claims a typed payload and coordinate with integrity metadata. | A committed transaction when viewed alone. | -| WAL commit marker | The complete transaction validated and crossed the adapter's commit boundary. | Confidentiality, remote freshness, application authorization, or protection from a compromised adapter. | -| Recovery certificate | Echo derived a named recovery posture and index root from a committed replay range. | That omitted external retained material is available or that the store is not a valid old prefix. | -| Obstruction fact | Echo could not lawfully produce the requested result under the named posture and evidence. | Permanent impossibility, authorization for retry, or application success. | +| Artifact or mechanism | Proposition it can support | What it does not prove | +| ---------------------------------------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| Canonical bytes | The value has one deterministic encoding under the named schema and version. | Validity under current policy, admission, authenticity, authorization, confidentiality, or availability. | +| Domain-separated BLAKE3 digest | The canonical bytes match the named digest domain, assuming collision resistance. | Who created the bytes, whether they are lawful, whether they are secret, or whether they are current. | +| Header or frame checksum | Accidental corruption is detectable relative to the checksum algorithm. | Adversarial authenticity or tamper resistance. | +| Content-addressed CAS id | Retrieved bytes match the addressed content hash. | Semantic coordinate, causal authority, retention, reveal permission, or future availability. | +| WSC payload | A deterministic physical representation can be decoded and validated under its profile. | Causal admission, WAL commit, recovery authority, or application meaning. | +| Storage locator or file path | A host suggests where bytes may be found. | Identity, integrity, durability, authority, or causal meaning. | +| Application proposal | A caller requested a specific operation against supplied coordinates. | Echo acceptance, eligibility, execution, or success. | +| External-action request | An Edict decision named one operation family, scope, basis, budget, input digest, settlement schema, and reconciliation law. | That Echo committed the request, that an adapter is authorized, or that any external effect occurred. | +| External-action recovery observation | A supplied recovery report contains a structurally valid requested, claimed, or settled lifecycle. | That the report came from trusted local storage, or that it can mint a transition grant or resumable result. | +| Recovered external-action coordinator | One fallible local-store snapshot validated a clean committed continuation and reconstructed its exact lifecycle index. | External truth, permission for application code to write WAL records, or permission to repeat a claimed effect. | +| Committed external-action claim | Echo durably selected one adapter attempt under the exact request, basis, runtime registry-policy digest, nonzero lease evidence, idempotency key, and reconciliation law. | That the adapter began or completed the effect, or that a recovered claim may be blindly reissued. | +| Admitted external-action settlement | Echo durably admitted one schema-, request-, attempt-, adapter-, basis-, digest-, nonzero-evidence-, budget-, and lifecycle-frontier-bound external observation. | That an external claim is objectively true, that another request has the same outcome, or that `OutcomeUnknown` means failure. | +| Trusted admission support chain | A receipt-backed trusted-host result or recovery-validated record proves Echo admitted the exact claim under the bound basis and policy. | Application execution, domain success, reveal permission, or material availability. A copied public ticket or fact is not sufficient support. | +| Trusted-host or recovery-validated tick receipt | Echo committed the exact scheduler-owned outcome and evidence named by the receipt. | That a copied receipt value has trusted support, or that unrelated operations are authorized. | +| Causal parent reference | The child cites a specific admitted parent event. | Undo, redo, compensation, or any other domain semantics not supplied by the application contract. | +| Causal-anchor claim | A canonical caller claim binds a subject, supplied basis, roots, purpose, and schema. | Echo admission, root existence, root authority, retention pinning, or domain checkpoint validity. | +| Trusted-host or recovery-validated causal-anchor admission | Echo committed the exact claim at its current logical basis under the bound support-policy digest. | Physical retention, materialization availability, application-domain validity, or text mutation. A standalone copied fact or anchor id is not sufficient support. | +| Capability presentation | A caller cites a grant identity in a presentation shape. | That the grant exists, is authentic, is unexpired, covers the artifact/operation/scope, or authorizes the current request. | +| Validated capability grant | The named grant covers the validated identity requirements under the evaluated policy posture. | Runtime support, scheduler work, law execution, unlimited aperture, or automatic reveal of retained material. | +| Reading envelope | Echo reports the basis, observer plan, aperture, budget, rights, residual, and evidence posture for a reading. | Causal mutation authority or permission to widen the reading. | +| Proof opening | The named commitment proposition verified for the named coordinates under the named proof system. | Capability, admission, scheduler, WAL, recovery, or reveal authority. | +| Graph or hologram materialization | A derived view was computed from a named basis and support set. | Substrate truth, freshness at another basis, or validity for another observer. | +| Projection-cache hit | A cached artifact matches the complete cache key and coverage contract. | That an incomplete key is safe, that the observer is equivalent, or that source authority remains available. | +| WAL frame | A record claims a typed payload and coordinate with integrity metadata. | A committed transaction when viewed alone. | +| WAL commit marker | The complete transaction validated and crossed the adapter's commit boundary. | Confidentiality, remote freshness, application authorization, or protection from a compromised adapter. | +| Recovery certificate | Echo derived a named recovery posture and index root from a committed replay range. | That omitted external retained material is available or that the store is not a valid old prefix. | +| Obstruction fact | Echo could not lawfully produce the requested result under the named posture and evidence. | Permanent impossibility, authorization for retry, or application success. | ## Caller Claim Versus Echo Evidence @@ -229,6 +234,13 @@ that check. The lower-level raw observation path and current high-level bridge must not be advertised as a complete product authorization boundary until trusted capability and law binding is end to end. +External-action adapter registration follows the same separation. The runtime +owner installs operation-, scope-, and adapter-specific bindings. Lookup +attenuates that policy into a request-specific authorization; the request +itself is not a credential. The adapter's external credential remains outside +Echo history, Edict, and model output. A committed claim records which adapter +was authorized without turning that public evidence into the credential. + ## Confidentiality, Revelation, And Erasure Confidentiality requires more than a content hash or redaction flag. A complete @@ -303,15 +315,15 @@ host threat model. Security-sensitive failures must preserve their category: -| Posture | Meaning | Caller response | -| ------- | ------- | --------------- | -| Rejected | Named law evaluated the candidate and lawfully declined it. | Change the proposal or accept the decision. | -| Obstructed | Required basis, capability, support, material, policy, or budget is unavailable. | Repair the named condition or make a new explicit request. | -| Conflict | Causal expectations or declared footprints are incompatible. | Preserve plurality or invoke an explicit resolution law. | -| Corrupt | Evidence fails canonical, integrity, or cross-evidence validation. | Quarantine and recover from trusted evidence; never normalize into success. | -| Internal fault | Echo failed its own invariant. | Isolate or quarantine the affected runtime scope and use trusted recovery. | -| Missing | Referenced material is absent. | Restore material or return an availability obstruction. | -| Redacted | Policy intentionally withholds material. | Do not infer absence or reveal through another cache. | +| Posture | Meaning | Caller response | +| -------------- | -------------------------------------------------------------------------------- | --------------------------------------------------------------------------- | +| Rejected | Named law evaluated the candidate and lawfully declined it. | Change the proposal or accept the decision. | +| Obstructed | Required basis, capability, support, material, policy, or budget is unavailable. | Repair the named condition or make a new explicit request. | +| Conflict | Causal expectations or declared footprints are incompatible. | Preserve plurality or invoke an explicit resolution law. | +| Corrupt | Evidence fails canonical, integrity, or cross-evidence validation. | Quarantine and recover from trusted evidence; never normalize into success. | +| Internal fault | Echo failed its own invariant. | Isolate or quarantine the affected runtime scope and use trusted recovery. | +| Missing | Referenced material is absent. | Restore material or return an availability obstruction. | +| Redacted | Policy intentionally withholds material. | Do not infer absence or reveal through another cache. | Retry is a new causal act unless the API explicitly defines exact idempotent recovery of the same claim. Hidden retry loops can duplicate effects and erase diff --git a/tests/docs/test_adr_namespace.sh b/tests/docs/test_adr_namespace.sh index d0b07989..4253337f 100755 --- a/tests/docs/test_adr_namespace.sh +++ b/tests/docs/test_adr_namespace.sh @@ -36,9 +36,10 @@ readonly current_adrs=( "0023-admitted-executable-operation-packages.md" "0024-anchored-node-creation-from-absence.md" "0025-scheduler-owned-executable-operation-actions.md" + "0026-durable-external-action-settlement.md" ) -readonly current_adr_last=25 +readonly current_adr_last=26 readonly superseded_legacy_adrs=( "ADR-0003-Materialization-Bus.md"