diff --git a/CHANGELOG.md b/CHANGELOG.md index 5d0842fc..ace6ea72 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -66,6 +66,11 @@ Applied, Rejected, Obstructed}` with receipt evidence and typed contract pipeline replay tests, reference trusted host loop test, and serious external consumer fixture without requiring developers to run the full DIND suite for normal local iteration. +- `xtask test-slice` now includes `runtime-wal-ack`, a narrow runtime WAL + ACK witness. The slice runs WAL-backed app-facing submission acceptance, + scheduler tick receipt commit-before-publish, recovered runtime indexes, + CLI submission-posture JSON, stale-claim guard, and generated man-page + checks. - The docs now include an executable local contract-host quickstart and a v0.1.0 authority-boundary audit. The quickstart points developers at `cargo xtask test-slice contract-path-release`, names the app-facing and @@ -528,6 +533,15 @@ Applied, Rejected, Obstructed}` with receipt evidence and typed contract ### Changed +- `echo-cli wal submission-posture` now exposes generic read-only recovery JSON + for one submission id and canonical envelope digest. The output reports retry + posture, recovered submission posture, receipt digest, and ticket digest + without importing any application nouns into Echo. +- Runtime WAL-backed scheduler ticks now roll back all tick WAL evidence from a + failed multi-head scheduler pass, treat missing or mismatched receipt + correlation evidence as an invariant error instead of a normal obstruction, + and bind runtime recovery certificates to rebuilt submission and receipt + indexes. - `warp-wasm` no longer contains the legacy Stack Witness 0001 `createBuffer`/`replaceRange`/`textWindow` shortcut. QueryView requests now route through the generic installed contract observer boundary only; without diff --git a/crates/warp-cli/src/cli.rs b/crates/warp-cli/src/cli.rs index af6edb4f..bf259023 100644 --- a/crates/warp-cli/src/cli.rs +++ b/crates/warp-cli/src/cli.rs @@ -89,6 +89,17 @@ pub enum WalCommands { #[arg(default_value = ".")] root: PathBuf, }, + /// Report recovered posture for one submission id/envelope pair. + SubmissionPosture { + /// Filesystem WAL root to inspect. + root: PathBuf, + /// 64-character hex submission id. + #[arg(long)] + submission_id: String, + /// 64-character hex canonical envelope digest. + #[arg(long)] + canonical_envelope_digest: String, + }, } /// Output format selector. @@ -239,6 +250,42 @@ mod tests { } } + #[test] + fn parse_wal_submission_posture() { + let cli = Cli::try_parse_from([ + "echo-cli", + "wal", + "submission-posture", + "runtime.wal", + "--submission-id", + "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "--canonical-envelope-digest", + "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", + ]) + .unwrap(); + match cli.command { + Commands::Wal { + command: + WalCommands::SubmissionPosture { + ref root, + ref submission_id, + ref canonical_envelope_digest, + }, + } => { + assert_eq!(root, &PathBuf::from("runtime.wal")); + assert_eq!( + submission_id, + "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa" + ); + assert_eq!( + canonical_envelope_digest, + "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb" + ); + } + _ => panic!("expected Wal submission-posture command"), + } + } + #[test] fn unknown_subcommand_is_error() { let result = Cli::try_parse_from(["echo-cli", "bogus"]); diff --git a/crates/warp-cli/src/main.rs b/crates/warp-cli/src/main.rs index a0b9091f..bdb273c7 100644 --- a/crates/warp-cli/src/main.rs +++ b/crates/warp-cli/src/main.rs @@ -48,5 +48,13 @@ fn main() -> Result<()> { Commands::Wal { command: WalCommands::Doctor { ref root }, } => wal::doctor(root, &cli.format), + Commands::Wal { + command: + WalCommands::SubmissionPosture { + ref root, + ref submission_id, + ref canonical_envelope_digest, + }, + } => wal::submission_posture(root, submission_id, canonical_envelope_digest, &cli.format), } } diff --git a/crates/warp-cli/src/wal.rs b/crates/warp-cli/src/wal.rs index 85b24c98..c85fd26b 100644 --- a/crates/warp-cli/src/wal.rs +++ b/crates/warp-cli/src/wal.rs @@ -4,12 +4,16 @@ use std::path::Path; -use anyhow::Result; +use anyhow::{bail, Result}; use serde::Serialize; -use warp_core::causal_wal::{doctor_filesystem_store, RecoveryTailPosture}; +use warp_core::causal_wal::{ + doctor_filesystem_store, recover_filesystem_store, recover_receipt_index, + recover_submission_index, RecoveredSubmissionPosture, RecoveryAccessMode, RecoveryTailPosture, + SubmissionRetryPosture, +}; use crate::cli::OutputFormat; -use crate::output::emit; +use crate::output::{emit, hex_hash}; /// Read-only WAL doctor JSON/text report. #[derive(Debug, Serialize)] @@ -20,6 +24,18 @@ pub(crate) struct WalDoctorOutput { pub(crate) obstruction_count: u64, } +/// Read-only recovered posture for one submission id/envelope pair. +#[derive(Debug, Serialize)] +pub(crate) struct WalSubmissionPostureOutput { + pub(crate) root: String, + pub(crate) submission_id: String, + pub(crate) canonical_envelope_digest: String, + pub(crate) retry_posture: &'static str, + pub(crate) recovered_posture: Option<&'static str>, + pub(crate) receipt_digest: Option, + pub(crate) ticket_digest: Option, +} + /// Runs `echo-cli wal doctor`. pub(crate) fn doctor(root: &Path, format: &OutputFormat) -> Result<()> { let report = doctor_filesystem_store(root)?; @@ -43,6 +59,49 @@ pub(crate) fn doctor(root: &Path, format: &OutputFormat) -> Result<()> { emit(format, &text, &json) } +/// Runs `echo-cli wal submission-posture`. +pub(crate) fn submission_posture( + root: &Path, + submission_id: &str, + canonical_envelope_digest: &str, + format: &OutputFormat, +) -> Result<()> { + let submission_id = parse_hash_hex(submission_id)?; + let canonical_envelope_digest = parse_hash_hex(canonical_envelope_digest)?; + let recovery = recover_filesystem_store(root, RecoveryAccessMode::ReadOnly)?; + let submissions = recover_submission_index(&recovery)?; + let receipts = recover_receipt_index(&recovery)?; + let matching_entry = submissions + .get(&submission_id) + .filter(|entry| entry.acceptance.canonical_envelope_digest == canonical_envelope_digest); + let output = WalSubmissionPostureOutput { + root: root.display().to_string(), + submission_id: hex_hash(&submission_id), + canonical_envelope_digest: hex_hash(&canonical_envelope_digest), + retry_posture: retry_posture_label( + submissions.retry_posture(submission_id, canonical_envelope_digest), + ), + recovered_posture: matching_entry.map(|entry| recovered_posture_label(entry.posture)), + receipt_digest: matching_entry + .and_then(|entry| entry.receipt_digest.map(|digest| hex_hash(&digest))), + ticket_digest: matching_entry + .and_then(|_| receipts.ticket_by_submission.get(&submission_id)) + .map(hex_hash), + }; + let text = format!( + "echo-cli wal submission-posture\nRoot: {}\nSubmission: {}\nCanonical envelope: {}\nRetry posture: {}\nRecovered posture: {}\nReceipt: {}\nTicket: {}\n", + output.root, + output.submission_id, + output.canonical_envelope_digest, + output.retry_posture, + output.recovered_posture.unwrap_or("None"), + output.receipt_digest.as_deref().unwrap_or("None"), + output.ticket_digest.as_deref().unwrap_or("None") + ); + let json = serde_json::to_value(&output)?; + emit(format, &text, &json) +} + fn tail_posture_label(posture: RecoveryTailPosture) -> &'static str { match posture { RecoveryTailPosture::Clean => "Clean", @@ -52,3 +111,42 @@ fn tail_posture_label(posture: RecoveryTailPosture) -> &'static str { RecoveryTailPosture::WouldTruncateAfter(_) => "WouldTruncateAfter", } } + +fn recovered_posture_label(posture: RecoveredSubmissionPosture) -> &'static str { + match posture { + RecoveredSubmissionPosture::AcceptedPending => "AcceptedPending", + RecoveredSubmissionPosture::DecidedApplied => "DecidedApplied", + RecoveredSubmissionPosture::DecidedRejected => "DecidedRejected", + RecoveredSubmissionPosture::Obstructed => "Obstructed", + RecoveredSubmissionPosture::RecoveryFaulted => "RecoveryFaulted", + } +} + +fn retry_posture_label(posture: SubmissionRetryPosture) -> &'static str { + match posture { + SubmissionRetryPosture::NotAccepted => "NotAccepted", + SubmissionRetryPosture::AlreadyAcceptedPending => "AlreadyAcceptedPending", + SubmissionRetryPosture::AlreadyDecidedApplied => "AlreadyDecidedApplied", + SubmissionRetryPosture::AlreadyDecidedRejected => "AlreadyDecidedRejected", + SubmissionRetryPosture::AlreadyObstructed => "AlreadyObstructed", + SubmissionRetryPosture::ConflictSameIdDifferentEnvelope => { + "ConflictSameIdDifferentEnvelope" + } + SubmissionRetryPosture::NewSubmissionWithoutPolicyDedupe => { + "NewSubmissionWithoutPolicyDedupe" + } + } +} + +fn parse_hash_hex(input: &str) -> Result { + let bytes = hex::decode(input)?; + if bytes.len() != 32 { + bail!( + "expected a 64-character hex hash, got {} bytes", + bytes.len() + ); + } + let mut hash = [0_u8; 32]; + hash.copy_from_slice(&bytes); + Ok(hash) +} diff --git a/crates/warp-cli/tests/cli_integration.rs b/crates/warp-cli/tests/cli_integration.rs index 3ab45154..227b1b86 100644 --- a/crates/warp-cli/tests/cli_integration.rs +++ b/crates/warp-cli/tests/cli_integration.rs @@ -14,10 +14,11 @@ use assert_cmd::cargo::cargo_bin; use predicates::prelude::*; use tempfile::TempDir; use warp_core::causal_wal::{ - build_submission_acceptance_transaction, AffectedFrontier, AffectedFrontierKind, - FilesystemWalStore, Lsn, PayloadCodecId, PayloadSchemaId, SubmissionAcceptanceRecord, - WalAppendAuthority, WalDurabilityMode, WalSegmentId, WalStorePort, WalTransactionBuilder, - WalTransactionId, WalTransactionKind, WriterEpochId, WriterEpochRequest, + build_submission_acceptance_transaction, build_tick_transaction, AffectedFrontier, + AffectedFrontierKind, FilesystemWalStore, Lsn, PayloadCodecId, PayloadSchemaId, + SubmissionAcceptanceRecord, TickReceiptRecord, WalAppendAuthority, WalDurabilityMode, + WalReceiptCorrelationRecord, WalSegmentId, WalStorePort, WalTickDecision, + WalTransactionBuilder, WalTransactionId, WalTransactionKind, WriterEpochId, WriterEpochRequest, }; use warp_core::wsc::{build_one_warp_input, write_wsc_one_warp}; use warp_core::{ @@ -142,6 +143,89 @@ fn filesystem_wal_with_committed_submission() -> TestResult { Ok(temp) } +fn filesystem_wal_with_decided_submission() -> TestResult { + let temp = TempDir::new()?; + let epoch = writer_epoch_request(); + let mut store = FilesystemWalStore::open(temp.path(), WalSegmentId::from_raw(1))?; + store.acquire_writer_epoch(epoch.clone())?; + let acceptance = build_submission_acceptance_transaction( + WalTransactionBuilder::new( + epoch.epoch_id, + WalSegmentId::from_raw(1), + WalTransactionId::from_hash(digest("transaction:accepted")), + WalTransactionKind::SubmissionIntake, + WalAppendAuthority::SubmissionIntake, + Lsn::from_raw(0), + digest("previous-frame"), + digest("previous-commit"), + WalDurabilityMode::Buffered, + PayloadCodecId::from_hash(digest("codec")), + PayloadSchemaId::from_hash(digest("schema")), + 1, + 1, + digest("domain"), + ), + SubmissionAcceptanceRecord { + submission_id: digest("submission:decided"), + canonical_envelope_digest: digest("envelope:decided"), + idempotency_key_digest: None, + acceptance_evidence_digest: digest("evidence:decided"), + }, + vec![AffectedFrontier { + kind: AffectedFrontierKind::SubmissionQueue, + before_digest: digest("submission-frontier-before"), + after_digest: digest("submission-frontier-after"), + }], + )?; + store.append_transaction(acceptance)?; + let receipt = TickReceiptRecord { + submission_id: digest("submission:decided"), + ticket_digest: digest("ticket:decided"), + receipt_digest: digest("receipt:decided"), + decision: WalTickDecision::Applied, + }; + let correlation = WalReceiptCorrelationRecord { + submission_id: receipt.submission_id, + ticket_digest: receipt.ticket_digest, + receipt_digest: receipt.receipt_digest, + }; + let tick = build_tick_transaction( + WalTransactionBuilder::new( + epoch.epoch_id, + WalSegmentId::from_raw(1), + WalTransactionId::from_hash(digest("transaction:ticked")), + WalTransactionKind::SchedulerTick, + WalAppendAuthority::TrustedScheduler, + Lsn::from_raw(2), + digest("tick-previous-frame"), + digest("tick-previous-commit"), + WalDurabilityMode::Buffered, + PayloadCodecId::from_hash(digest("codec")), + PayloadSchemaId::from_hash(digest("schema")), + 1, + 1, + digest("domain"), + ), + receipt, + correlation, + digest("state-delta:decided"), + vec![ + AffectedFrontier { + kind: AffectedFrontierKind::ReceiptIndex, + before_digest: digest("receipt-frontier-before"), + after_digest: digest("receipt-frontier-after"), + }, + AffectedFrontier { + kind: AffectedFrontierKind::RuntimeState, + before_digest: digest("runtime-frontier-before"), + after_digest: digest("runtime-frontier-after"), + }, + ], + )?; + store.append_transaction(tick)?; + Ok(temp) +} + #[test] fn help_shows_all_subcommands() { echo_cli() @@ -232,7 +316,8 @@ fn wal_doctor_help_lists_read_only_doctor() { .args(["wal", "--help"]) .assert() .success() - .stdout(predicate::str::contains("doctor")); + .stdout(predicate::str::contains("doctor")) + .stdout(predicate::str::contains("submission-posture")); } #[test] @@ -272,6 +357,109 @@ fn wal_doctor_json_reports_committed_filesystem_wal() -> TestResult { Ok(()) } +#[test] +fn wal_submission_posture_json_reports_generic_recovered_status() -> TestResult { + let temp = filesystem_wal_with_decided_submission()?; + let assert = echo_cli() + .args([ + "--format", + "json", + "wal", + "submission-posture", + temp.path().to_str().ok_or("temp path is not UTF-8")?, + "--submission-id", + &hex::encode(digest("submission:decided")), + "--canonical-envelope-digest", + &hex::encode(digest("envelope:decided")), + ]) + .assert() + .success(); + let json: serde_json::Value = serde_json::from_slice(&assert.get_output().stdout)?; + + assert_eq!(json["retry_posture"], "AlreadyDecidedApplied"); + assert_eq!(json["recovered_posture"], "DecidedApplied"); + assert_eq!( + json["receipt_digest"], + hex::encode(digest("receipt:decided")) + ); + assert_eq!(json["ticket_digest"], hex::encode(digest("ticket:decided"))); + Ok(()) +} + +#[test] +fn wal_submission_posture_text_reports_canonical_envelope_digest() -> TestResult { + let temp = filesystem_wal_with_decided_submission()?; + let envelope_digest = hex::encode(digest("envelope:decided")); + let assert = echo_cli() + .args([ + "wal", + "submission-posture", + temp.path().to_str().ok_or("temp path is not UTF-8")?, + "--submission-id", + &hex::encode(digest("submission:decided")), + "--canonical-envelope-digest", + &envelope_digest, + ]) + .assert() + .success(); + let stdout = String::from_utf8_lossy(&assert.get_output().stdout); + + assert!(stdout.contains(&format!("Canonical envelope: {envelope_digest}"))); + Ok(()) +} + +#[test] +fn wal_submission_posture_json_reports_not_accepted_without_app_nouns() -> TestResult { + let temp = filesystem_wal_with_committed_submission()?; + let assert = echo_cli() + .args([ + "--format", + "json", + "wal", + "submission-posture", + temp.path().to_str().ok_or("temp path is not UTF-8")?, + "--submission-id", + &hex::encode(digest("submission:missing")), + "--canonical-envelope-digest", + &hex::encode(digest("envelope:missing")), + ]) + .assert() + .success(); + let json: serde_json::Value = serde_json::from_slice(&assert.get_output().stdout)?; + + assert_eq!(json["retry_posture"], "NotAccepted"); + assert!(json["recovered_posture"].is_null()); + assert!(json["receipt_digest"].is_null()); + assert!(json["ticket_digest"].is_null()); + Ok(()) +} + +#[test] +fn wal_submission_posture_json_suppresses_recovered_fields_for_envelope_conflict() -> TestResult { + let temp = filesystem_wal_with_decided_submission()?; + let assert = echo_cli() + .args([ + "--format", + "json", + "wal", + "submission-posture", + temp.path().to_str().ok_or("temp path is not UTF-8")?, + "--submission-id", + &hex::encode(digest("submission:decided")), + "--canonical-envelope-digest", + &hex::encode(digest("envelope:conflicting")), + ]) + .assert() + .success(); + let json: serde_json::Value = serde_json::from_slice(&assert.get_output().stdout)?; + + assert_eq!(json["retry_posture"], "ConflictSameIdDifferentEnvelope"); + assert!(json["recovered_posture"].is_null()); + assert!(json["receipt_digest"].is_null()); + assert!(json["ticket_digest"].is_null()); + Ok(()) +} + #[test] fn inspect_text_reports_metadata_stats_and_tree() -> TestResult { let temp = write_demo_snapshot()?; diff --git a/crates/warp-core/src/causal_wal.rs b/crates/warp-core/src/causal_wal.rs index 750c4b84..6ad80e1b 100644 --- a/crates/warp-core/src/causal_wal.rs +++ b/crates/warp-core/src/causal_wal.rs @@ -27,6 +27,7 @@ const WAL_PAYLOAD_DOMAIN: &[u8] = b"echo:causal_wal:payload:v1\0"; const WAL_RECORDS_ROOT_DOMAIN: &[u8] = b"echo:causal_wal:records_root:v1\0"; const WAL_FRONTIERS_ROOT_DOMAIN: &[u8] = b"echo:causal_wal:frontiers_root:v1\0"; const WAL_COMMIT_DOMAIN: &[u8] = b"echo:causal_wal:commit:v1\0"; +const WAL_RECOVERED_INDEX_ROOT_DOMAIN: &[u8] = b"echo:causal_wal:recovered_index_root:v1\0"; const WAL_HEADER_CHECKSUM_DOMAIN: &[u8] = b"echo:causal_wal:header_checksum:v1\0"; const WAL_FRAME_CHECKSUM_DOMAIN: &[u8] = b"echo:causal_wal:frame_checksum:v1\0"; const WAL_DISK_RECORD_DOMAIN: &[u8] = b"echo:causal_wal:disk_record:v1\0"; @@ -2136,6 +2137,87 @@ impl RecoveredSubmissionIndex { } } +/// Builds a stable root over recovered submission and receipt indexes. +#[must_use] +pub fn recovered_submission_receipt_index_root( + submissions: &RecoveredSubmissionIndex, + receipts: &RecoveredReceiptIndex, +) -> Hash { + let mut hasher = blake3::Hasher::new(); + hasher.update(WAL_RECOVERED_INDEX_ROOT_DOMAIN); + hasher.update(&len_u64(submissions.submissions.len()).to_le_bytes()); + for (submission_id, entry) in &submissions.submissions { + hasher.update(b"submission"); + hasher.update(submission_id); + hasher.update(&entry.acceptance.submission_id); + hasher.update(&entry.acceptance.canonical_envelope_digest); + match entry.acceptance.idempotency_key_digest { + Some(digest) => { + hasher.update(&[1]); + hasher.update(&digest); + } + None => { + hasher.update(&[0]); + } + } + hasher.update(&entry.acceptance.acceptance_evidence_digest); + hasher.update(&[recovered_submission_posture_code(entry.posture)]); + match entry.receipt_digest { + Some(digest) => { + hasher.update(&[1]); + hasher.update(&digest); + } + None => { + hasher.update(&[0]); + } + } + } + hasher.update(&len_u64(submissions.envelope_to_submissions.len()).to_le_bytes()); + for (envelope_digest, submission_ids) in &submissions.envelope_to_submissions { + hasher.update(b"envelope"); + hasher.update(envelope_digest); + hasher.update(&len_u64(submission_ids.len()).to_le_bytes()); + for submission_id in submission_ids { + hasher.update(submission_id); + } + } + hasher.update(&len_u64(receipts.receipt_by_submission.len()).to_le_bytes()); + for (submission_id, receipt_digest) in &receipts.receipt_by_submission { + hasher.update(b"receipt-by-submission"); + hasher.update(submission_id); + hasher.update(receipt_digest); + } + hasher.update(&len_u64(receipts.receipt_by_ticket.len()).to_le_bytes()); + for (ticket_digest, receipt_digest) in &receipts.receipt_by_ticket { + hasher.update(b"receipt-by-ticket"); + hasher.update(ticket_digest); + hasher.update(receipt_digest); + } + hasher.update(&len_u64(receipts.ticket_by_submission.len()).to_le_bytes()); + for (submission_id, ticket_digest) in &receipts.ticket_by_submission { + hasher.update(b"ticket-by-submission"); + hasher.update(submission_id); + hasher.update(ticket_digest); + } + hasher.update(&len_u64(receipts.decisions_by_receipt.len()).to_le_bytes()); + for (receipt_digest, decision) in &receipts.decisions_by_receipt { + hasher.update(b"decision-by-receipt"); + hasher.update(receipt_digest); + hasher.update(&[decision.code()]); + } + hasher.finalize().into() +} + +fn recovered_submission_posture_code(posture: RecoveredSubmissionPosture) -> u8 { + match posture { + RecoveredSubmissionPosture::AcceptedPending => 1, + RecoveredSubmissionPosture::DecidedApplied => 2, + RecoveredSubmissionPosture::DecidedRejected => 3, + RecoveredSubmissionPosture::Obstructed => 4, + RecoveredSubmissionPosture::RecoveryFaulted => 5, + } +} + /// Recovered receipt correlation index. #[derive(Clone, Debug, Default, PartialEq, Eq)] pub struct RecoveredReceiptIndex { diff --git a/crates/warp-core/src/trusted_runtime_host.rs b/crates/warp-core/src/trusted_runtime_host.rs index 926a19a7..c47148ec 100644 --- a/crates/warp-core/src/trusted_runtime_host.rs +++ b/crates/warp-core/src/trusted_runtime_host.rs @@ -7,21 +7,28 @@ //! daemon, does not make wall-clock cadence semantic, and does not give //! application code tick authority. +use std::collections::BTreeSet; + use thiserror::Error; use crate::{ causal_wal::{ - build_submission_acceptance_transaction, AffectedFrontier, AffectedFrontierKind, - InMemoryWalStore, Lsn, PayloadCodecId, PayloadSchemaId, SubmissionAcceptanceRecord, + build_recovery_certificate, build_submission_acceptance_transaction, + build_tick_transaction, recover_in_memory_store, recover_receipt_index, + recover_submission_index, recovered_submission_receipt_index_root, AffectedFrontier, + AffectedFrontierKind, InMemoryWalStore, Lsn, PayloadCodecId, PayloadSchemaId, + RecoveredReceiptIndex, RecoveredSubmissionIndex, RecoveryAccessMode, RecoveryCertificate, + RecoveryScanReport, SubmissionAcceptanceRecord, SubmissionRetryPosture, TickReceiptRecord, WalAppendAuthority, WalBuildError, WalCommittedTransaction, WalDurabilityMode, - WalRecordKind, WalSegmentId, WalStoreError, WalStorePort, WalTransactionBuilder, - WalTransactionCommit, WalTransactionId, WalTransactionKind, WriterEpochId, - WriterEpochRequest, + WalReceiptCorrelationRecord, WalRecoveryError, WalSegmentId, WalStoreError, WalStorePort, + WalTickDecision, WalTransactionBuilder, WalTransactionCommit, WalTransactionId, + WalTransactionKind, WriterEpochId, WriterEpochRequest, }, Engine, IngressEnvelope, InstalledContractPackage, InstalledContractPackageError, - InstalledContractPackageRecord, IntentOutcome, IntentSubmissionHandle, ObservationArtifact, - ObservationError, ObservationRequest, ObservationService, OpticAdmissionTicket, - ProvenanceService, RuntimeError, SchedulerCoordinator, StepRecord, + InstalledContractPackageRecord, IntentOutcome, IntentOutcomeDecision, IntentOutcomeObservation, + IntentSubmissionHandle, ObservationArtifact, ObservationError, ObservationRequest, + ObservationService, OpticAdmissionTicket, ProvenanceService, ReceiptCorrelationRecord, + RuntimeError, SchedulerCoordinator, StepRecord, TickReceiptRejection, TicketedRuntimeIngressAuthority, TicketedRuntimeIngressDisposition, WorldlineRuntime, }; use crate::{Hash, HistoryError}; @@ -66,6 +73,29 @@ pub enum TrustedRuntimeWalError { /// WAL storage failed before durable acknowledgement. #[error("trusted runtime WAL store error: {0}")] Store(#[from] WalStoreError), + /// WAL recovery failed while rebuilding runtime evidence. + #[error("trusted runtime WAL recovery error: {0}")] + Recovery(#[from] WalRecoveryError), + /// Runtime outcome evidence could not be matched to the receipt correlation. + #[error( + "trusted runtime WAL tick outcome unavailable for submission {submission_id:?} receipt {receipt_digest:?}" + )] + TickOutcomeUnavailable { + /// Submission whose scheduler tick outcome should have been decided. + submission_id: Hash, + /// Receipt digest expected by the new correlation. + receipt_digest: Hash, + }, + /// Runtime outcome evidence did not match the receipt correlation. + #[error( + "trusted runtime WAL receipt digest mismatch: expected {expected_receipt_digest:?}, observed {observed_receipt_digest:?}" + )] + TickReceiptDigestMismatch { + /// Receipt digest expected by the new correlation. + expected_receipt_digest: Hash, + /// Receipt digest observed through the outcome surface. + observed_receipt_digest: Hash, + }, } /// Summary returned after a trusted host runs the scheduler until idle. @@ -77,6 +107,17 @@ pub struct TrustedRuntimeHostRunReport { pub committed_steps: usize, } +/// Read-only runtime WAL recovery report. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct TrustedRuntimeWalRecovery { + /// Recovery certificate summarizing committed history replay. + pub certificate: RecoveryCertificate, + /// Rebuilt submission posture index. + pub submissions: RecoveredSubmissionIndex, + /// Rebuilt receipt/correlation index. + pub receipts: RecoveredReceiptIndex, +} + /// Local trusted runtime host for the app-safe contract-host path. /// /// Application code should receive [`TrustedRuntimeApp`], not this type. This @@ -225,8 +266,59 @@ impl TrustedRuntimeHost { /// /// Returns a runtime error if the scheduler pass fails. pub fn tick_once(&mut self) -> Result, TrustedRuntimeHostError> { - SchedulerCoordinator::super_tick(&mut self.runtime, &mut self.provenance, &mut self.engine) - .map_err(Into::into) + let existing_correlations = self + .runtime + .receipt_correlations() + .map(|correlation| correlation.ticketed_ingress_id) + .collect::>(); + let runtime_before = self.runtime.clone(); + let provenance_before = self.provenance.clone(); + let records = SchedulerCoordinator::super_tick( + &mut self.runtime, + &mut self.provenance, + &mut self.engine, + )?; + let mut tick_wal_records = Vec::new(); + if self.runtime_wal.is_some() { + let new_correlations = self + .runtime + .receipt_correlations() + .filter(|correlation| { + !existing_correlations.contains(&correlation.ticketed_ingress_id) + }) + .cloned() + .collect::>(); + for correlation in new_correlations { + let decision = match wal_tick_decision_from_observation( + self.runtime + .observe_intent_outcome(&correlation.submission_id), + correlation.tick_receipt_digest, + ) { + Ok(decision) => decision, + Err(error) => { + self.runtime = runtime_before; + self.provenance = provenance_before; + return Err(error.into()); + } + }; + let state_delta_digest = tick_state_delta_digest(&correlation); + tick_wal_records.push((correlation, decision, state_delta_digest)); + } + } + if let Some(runtime_wal) = self.runtime_wal.as_mut() { + let runtime_wal_before = runtime_wal.clone(); + for (correlation, decision, state_delta_digest) in &tick_wal_records { + if let Err(error) = + runtime_wal.record_tick_receipt(correlation, *decision, *state_delta_digest) + { + *runtime_wal = runtime_wal_before; + self.runtime = runtime_before; + self.provenance = provenance_before; + return Err(error.into()); + } + } + } + Ok(records) } /// Runs scheduler-owned passes until an idle pass occurs. @@ -276,6 +368,8 @@ pub struct TrustedRuntimeWal { payload_schema_id: PayloadSchemaId, digest_domain: Hash, submission_frontier_digest: Hash, + receipt_frontier_digest: Hash, + runtime_state_frontier_digest: Hash, } impl TrustedRuntimeWal { @@ -315,6 +409,8 @@ impl TrustedRuntimeWal { )), digest_domain: trusted_runtime_wal_digest("digest-domain"), submission_frontier_digest: trusted_runtime_wal_digest("submission-frontier:genesis"), + receipt_frontier_digest: trusted_runtime_wal_digest("receipt-frontier:genesis"), + runtime_state_frontier_digest: trusted_runtime_wal_digest("runtime-frontier:genesis"), }) } @@ -324,7 +420,7 @@ impl TrustedRuntimeWal { self.store.read_commits() } - /// Returns committed WAL frames recorded by the adapter. + /// Returns WAL frames recorded by the adapter. #[must_use] pub fn frames(&self) -> Vec { self.store.read_frames() @@ -336,6 +432,20 @@ impl TrustedRuntimeWal { self.store.clone() } + /// Recovers submission and receipt indexes from committed WAL transactions + /// without scheduler callbacks. + pub fn recover_read_only(&self) -> Result { + let mut store = self.cloned_store(); + let report = recover_in_memory_store(&mut store, RecoveryAccessMode::ReadOnly)?; + let submissions = recover_submission_index(&report).map_err(WalRecoveryError::from)?; + let receipts = recover_receipt_index(&report).map_err(WalRecoveryError::from)?; + Ok(TrustedRuntimeWalRecovery { + certificate: runtime_wal_recovery_certificate(&report, &submissions, &receipts), + submissions, + receipts, + }) + } + /// Returns the number of committed submission-intake transactions. #[must_use] pub fn submission_acceptance_count(&self) -> usize { @@ -346,21 +456,31 @@ impl TrustedRuntimeWal { .count() } + /// Returns the number of committed scheduler-tick transactions. + #[must_use] + pub fn scheduler_tick_count(&self) -> usize { + self.store + .read_commits() + .into_iter() + .filter(|commit| commit.transaction_kind == WalTransactionKind::SchedulerTick) + .count() + } + fn has_submission_acceptance( &self, submission_id: Hash, canonical_envelope_digest: Hash, - ) -> bool { - self.store.read_frames().into_iter().any(|frame| { - if frame.header.record_kind != WalRecordKind::SubmissionAcceptedRecorded { - return false; - } - SubmissionAcceptanceRecord::from_payload_bytes(&frame.payload.canonical_bytes) - .is_ok_and(|record| { - record.submission_id == submission_id - && record.canonical_envelope_digest == canonical_envelope_digest - }) - }) + ) -> Result { + let recovery = self.recover_read_only()?; + Ok(matches!( + recovery + .submissions + .retry_posture(submission_id, canonical_envelope_digest), + SubmissionRetryPosture::AlreadyAcceptedPending + | SubmissionRetryPosture::AlreadyDecidedApplied + | SubmissionRetryPosture::AlreadyDecidedRejected + | SubmissionRetryPosture::AlreadyObstructed + )) } fn record_submission_acceptance( @@ -394,6 +514,62 @@ impl TrustedRuntimeWal { Ok(commit) } + fn record_tick_receipt( + &mut self, + correlation: &ReceiptCorrelationRecord, + decision: WalTickDecision, + state_delta_digest: Hash, + ) -> Result { + let receipt = TickReceiptRecord { + submission_id: correlation.submission_id, + ticket_digest: correlation.ticket_digest, + receipt_digest: correlation.tick_receipt_digest, + decision, + }; + let wal_correlation = WalReceiptCorrelationRecord { + submission_id: correlation.submission_id, + ticket_digest: correlation.ticket_digest, + receipt_digest: correlation.tick_receipt_digest, + }; + let next_receipt_frontier = + receipt_frontier_digest(self.receipt_frontier_digest, receipt, wal_correlation); + let next_runtime_frontier = runtime_state_frontier_digest( + self.runtime_state_frontier_digest, + correlation, + state_delta_digest, + ); + let transaction = build_tick_transaction( + self.builder( + WalTransactionKind::SchedulerTick, + WalAppendAuthority::TrustedScheduler, + WalTransactionId::from_hash(tick_transaction_digest( + correlation, + decision, + state_delta_digest, + )), + ), + receipt, + wal_correlation, + state_delta_digest, + vec![ + AffectedFrontier { + kind: AffectedFrontierKind::ReceiptIndex, + before_digest: self.receipt_frontier_digest, + after_digest: next_receipt_frontier, + }, + AffectedFrontier { + kind: AffectedFrontierKind::RuntimeState, + before_digest: self.runtime_state_frontier_digest, + after_digest: next_runtime_frontier, + }, + ], + )?; + let commit = self.append_transaction(transaction)?; + self.receipt_frontier_digest = next_receipt_frontier; + self.runtime_state_frontier_digest = next_runtime_frontier; + Ok(commit) + } + fn append_transaction( &mut self, transaction: WalCommittedTransaction, @@ -446,6 +622,51 @@ impl TrustedRuntimeWal { pub fn new_in_memory_at_lsn_for_test(next_lsn: Lsn) -> Result { Self::new_in_memory_at_lsn(next_lsn) } + + /// Appends submission acceptance frames without a transaction commit marker. + pub fn append_uncommitted_submission_acceptance_for_test( + &mut self, + envelope: &IngressEnvelope, + handle: IntentSubmissionHandle, + ) -> Result<(), TrustedRuntimeWalError> { + let record = SubmissionAcceptanceRecord { + submission_id: handle.submission_id, + canonical_envelope_digest: envelope.ingress_id(), + idempotency_key_digest: None, + acceptance_evidence_digest: acceptance_evidence_digest(handle), + }; + let next_submission_frontier = + submission_frontier_digest(self.submission_frontier_digest, record); + let transaction = build_submission_acceptance_transaction( + self.builder( + WalTransactionKind::SubmissionIntake, + WalAppendAuthority::SubmissionIntake, + WalTransactionId::from_hash(submission_transaction_digest(handle, record)), + ), + record, + vec![AffectedFrontier { + kind: AffectedFrontierKind::SubmissionQueue, + before_digest: self.submission_frontier_digest, + after_digest: next_submission_frontier, + }], + )?; + let last_frame_digest = transaction.frames.last().map_or( + self.previous_frame_digest, + crate::causal_wal::WalFrame::digest, + ); + let next_lsn = transaction + .commit + .last_lsn + .checked_next() + .ok_or(WalBuildError::LsnOverflow)?; + for frame in transaction.frames { + self.store + .append_uncommitted_frame(self.writer_epoch, frame)?; + } + self.next_lsn = next_lsn; + self.previous_frame_digest = last_frame_digest; + Ok(()) + } } /// App-facing handle for a trusted local runtime host. @@ -496,10 +717,16 @@ impl TrustedRuntimeApp<'_> { self.host.runtime = before_runtime; return Err(TrustedRuntimeHostError::RuntimeWalUnavailable); }; - if handle.duplicate - && runtime_wal.has_submission_acceptance(handle.submission_id, envelope.ingress_id()) - { - return Ok(handle); + if handle.duplicate { + match runtime_wal.has_submission_acceptance(handle.submission_id, envelope.ingress_id()) + { + Ok(true) => return Ok(handle), + Ok(false) => {} + Err(error) => { + self.host.runtime = before_runtime; + return Err(error.into()); + } + } } if let Err(error) = runtime_wal.record_submission_acceptance(&envelope, handle) { self.host.runtime = before_runtime; @@ -588,3 +815,247 @@ fn submission_frontier_digest(previous: Hash, record: SubmissionAcceptanceRecord hasher.update(&record.acceptance_evidence_digest); hasher.finalize().into() } + +fn wal_tick_decision_from_observation( + observation: IntentOutcomeObservation, + expected_receipt_digest: Hash, +) -> Result { + let (correlation, decision) = match observation { + IntentOutcomeObservation::Decided { + correlation, + decision, + } => (correlation, decision), + IntentOutcomeObservation::UnknownSubmission { submission_id } + | IntentOutcomeObservation::Pending { submission_id, .. } => { + return Err(TrustedRuntimeWalError::TickOutcomeUnavailable { + submission_id, + receipt_digest: expected_receipt_digest, + }); + } + }; + if correlation.tick_receipt_digest != expected_receipt_digest { + return Err(TrustedRuntimeWalError::TickReceiptDigestMismatch { + expected_receipt_digest, + observed_receipt_digest: correlation.tick_receipt_digest, + }); + } + Ok(match decision { + IntentOutcomeDecision::Applied { .. } => WalTickDecision::Applied, + IntentOutcomeDecision::Rejected { + reason: TickReceiptRejection::FootprintConflict, + .. + } => WalTickDecision::RejectedFootprintConflict, + IntentOutcomeDecision::NoMatchingReceiptEntry { .. } => { + return Err(TrustedRuntimeWalError::TickOutcomeUnavailable { + submission_id: correlation.submission_id, + receipt_digest: expected_receipt_digest, + }); + } + }) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::{ + GlobalTick, IngressSubmissionGeneration, WorldlineId, WorldlineTick, WriterHeadKey, + }; + + fn test_head_key() -> WriterHeadKey { + WriterHeadKey { + worldline_id: WorldlineId::from_bytes([9; 32]), + head_id: crate::make_head_id("runtime-wal-test"), + } + } + + fn test_correlation(receipt_digest: Hash) -> ReceiptCorrelationRecord { + ReceiptCorrelationRecord { + ticketed_ingress_id: [1; 32], + submission_id: [2; 32], + ticket_digest: [3; 32], + ingress_id: [4; 32], + head_key: test_head_key(), + contract: None, + commit_global_tick: GlobalTick::from_raw(1), + worldline_tick_after: WorldlineTick::from_raw(1), + tick_receipt_digest: receipt_digest, + commit_hash: [5; 32], + } + } + + #[test] + fn runtime_wal_tick_decision_rejects_pending_observation_as_invariant() { + let err = wal_tick_decision_from_observation( + IntentOutcomeObservation::Pending { + submission_id: [2; 32], + submission_generation: IngressSubmissionGeneration::from_raw(1), + ticketed_ingress_id: Some([6; 32]), + }, + [7; 32], + ) + .expect_err("pending outcome cannot produce scheduler tick WAL evidence"); + + assert!(matches!( + err, + TrustedRuntimeWalError::TickOutcomeUnavailable { + submission_id, + receipt_digest, + } if submission_id == [2; 32] && receipt_digest == [7; 32] + )); + } + + #[test] + fn runtime_wal_tick_decision_rejects_receipt_digest_mismatch_as_invariant() { + let err = wal_tick_decision_from_observation( + IntentOutcomeObservation::Decided { + correlation: Box::new(test_correlation([8; 32])), + decision: IntentOutcomeDecision::Applied { + receipt_entry_index: 0, + rule_id: [9; 32], + }, + }, + [7; 32], + ) + .expect_err("mismatched receipt digest cannot produce scheduler tick WAL evidence"); + + assert!(matches!( + err, + TrustedRuntimeWalError::TickReceiptDigestMismatch { + expected_receipt_digest, + observed_receipt_digest, + } if expected_receipt_digest == [7; 32] && observed_receipt_digest == [8; 32] + )); + } + + #[test] + fn runtime_wal_tick_decision_rejects_missing_receipt_entry_as_invariant() { + let err = wal_tick_decision_from_observation( + IntentOutcomeObservation::Decided { + correlation: Box::new(test_correlation([7; 32])), + decision: IntentOutcomeDecision::NoMatchingReceiptEntry { + tick_receipt_digest: [7; 32], + }, + }, + [7; 32], + ) + .expect_err("missing receipt entries cannot produce scheduler tick WAL evidence"); + + assert!(matches!( + err, + TrustedRuntimeWalError::TickOutcomeUnavailable { + submission_id, + receipt_digest, + } if submission_id == [2; 32] && receipt_digest == [7; 32] + )); + } + + #[test] + fn runtime_wal_tick_decision_maps_matching_outcome() { + let decision = wal_tick_decision_from_observation( + IntentOutcomeObservation::Decided { + correlation: Box::new(test_correlation([7; 32])), + decision: IntentOutcomeDecision::Applied { + receipt_entry_index: 0, + rule_id: [9; 32], + }, + }, + [7; 32], + ) + .expect("matching outcome should map to a WAL tick decision"); + + assert_eq!(decision, WalTickDecision::Applied); + } +} + +fn tick_transaction_digest( + correlation: &ReceiptCorrelationRecord, + decision: WalTickDecision, + state_delta_digest: Hash, +) -> Hash { + let mut hasher = blake3::Hasher::new(); + hasher.update(TRUSTED_RUNTIME_WAL_DOMAIN); + hasher.update(b"tick-transaction"); + hasher.update(&correlation.ticketed_ingress_id); + hasher.update(&correlation.submission_id); + hasher.update(&correlation.ticket_digest); + hasher.update(&correlation.ingress_id); + hasher.update(&correlation.tick_receipt_digest); + hasher.update(&correlation.commit_hash); + hasher.update(&correlation.commit_global_tick.as_u64().to_le_bytes()); + hasher.update(&correlation.worldline_tick_after.as_u64().to_le_bytes()); + hasher.update(&[wal_tick_decision_code(decision)]); + hasher.update(&state_delta_digest); + hasher.finalize().into() +} + +fn tick_state_delta_digest(correlation: &ReceiptCorrelationRecord) -> Hash { + let mut hasher = blake3::Hasher::new(); + hasher.update(TRUSTED_RUNTIME_WAL_DOMAIN); + hasher.update(b"tick-state-delta"); + hasher.update(&correlation.commit_hash); + hasher.update(correlation.head_key.worldline_id.as_bytes()); + hasher.update(correlation.head_key.head_id.as_bytes()); + hasher.update(&correlation.commit_global_tick.as_u64().to_le_bytes()); + hasher.update(&correlation.worldline_tick_after.as_u64().to_le_bytes()); + hasher.finalize().into() +} + +fn receipt_frontier_digest( + previous: Hash, + receipt: TickReceiptRecord, + correlation: WalReceiptCorrelationRecord, +) -> Hash { + let mut hasher = blake3::Hasher::new(); + hasher.update(TRUSTED_RUNTIME_WAL_DOMAIN); + hasher.update(b"receipt-frontier"); + hasher.update(&previous); + hasher.update(&receipt.submission_id); + hasher.update(&receipt.ticket_digest); + hasher.update(&receipt.receipt_digest); + hasher.update(&[wal_tick_decision_code(receipt.decision)]); + hasher.update(&correlation.submission_id); + hasher.update(&correlation.ticket_digest); + hasher.update(&correlation.receipt_digest); + hasher.finalize().into() +} + +fn wal_tick_decision_code(decision: WalTickDecision) -> u8 { + match decision { + WalTickDecision::Applied => 1, + WalTickDecision::RejectedFootprintConflict => 2, + WalTickDecision::Obstructed => 3, + } +} + +fn runtime_state_frontier_digest( + previous: Hash, + correlation: &ReceiptCorrelationRecord, + state_delta_digest: Hash, +) -> Hash { + let mut hasher = blake3::Hasher::new(); + hasher.update(TRUSTED_RUNTIME_WAL_DOMAIN); + hasher.update(b"runtime-state-frontier"); + hasher.update(&previous); + hasher.update(&correlation.commit_hash); + hasher.update(&state_delta_digest); + hasher.update(&correlation.commit_global_tick.as_u64().to_le_bytes()); + hasher.update(&correlation.worldline_tick_after.as_u64().to_le_bytes()); + hasher.finalize().into() +} + +fn runtime_wal_recovery_certificate( + report: &RecoveryScanReport, + submissions: &RecoveredSubmissionIndex, + receipts: &RecoveredReceiptIndex, +) -> RecoveryCertificate { + let recovered_frontier_root = report + .last_commit_digest() + .unwrap_or_else(|| trusted_runtime_wal_digest("recovery-frontier:empty")); + build_recovery_certificate( + report, + None, + 0, + recovered_frontier_root, + recovered_submission_receipt_index_root(submissions, receipts), + ) +} diff --git a/crates/warp-core/tests/trusted_runtime_host_loop_tests.rs b/crates/warp-core/tests/trusted_runtime_host_loop_tests.rs index 5e500153..667a677d 100644 --- a/crates/warp-core/tests/trusted_runtime_host_loop_tests.rs +++ b/crates/warp-core/tests/trusted_runtime_host_loop_tests.rs @@ -10,8 +10,9 @@ use echo_registry_api::{ }; use warp_core::{ causal_wal::{ - recover_in_memory_store, recover_submission_index, Lsn, RecoveredSubmissionPosture, - RecoveryAccessMode, WalBuildError, + recover_in_memory_store, recover_receipt_index, recover_submission_index, + recovered_submission_receipt_index_root, Lsn, RecoveredSubmissionPosture, + RecoveryAccessMode, WalBuildError, WalTransactionKind, }, make_head_id, make_intent_kind, make_node_id, make_type_id, AuthoredObserverPlan, ContractMutationHandler, ContractOperationKind, ContractPackageIdentity, ContractQueryObserver, @@ -239,6 +240,30 @@ fn runtime() -> (WorldlineRuntime, WorldlineId) { (runtime, worldline_id) } +fn runtime_pair() -> (WorldlineRuntime, WorldlineId, WorldlineId) { + let mut runtime = WorldlineRuntime::new(); + let first = WorldlineId::from_bytes([1; 32]); + let second = WorldlineId::from_bytes([2; 32]); + for (worldline_id, head_label) in [(first, "default-a"), (second, "default-b")] { + runtime + .register_worldline(worldline_id, WorldlineState::empty()) + .expect("worldline should register"); + runtime + .register_writer_head(WriterHead::with_routing( + WriterHeadKey { + worldline_id, + head_id: make_head_id(head_label), + }, + PlaybackMode::Play, + InboxPolicy::AcceptAll, + None, + true, + )) + .expect("writer head should register"); + } + (runtime, first, second) +} + fn eint_envelope(worldline_id: WorldlineId) -> IngressEnvelope { IngressEnvelope::local_intent( IngressTarget::DefaultWriter { worldline_id }, @@ -442,6 +467,7 @@ fn runtime_wal_ack_duplicate_without_prior_wal_backfills_acceptance() { .expect("runtime WAL should initialize"); let envelope = eint_envelope(worldline_id); + let envelope_digest = envelope.ingress_id(); let first = { let mut app = host.app(); app.submit_intent(envelope.clone()) @@ -460,6 +486,59 @@ fn runtime_wal_ack_duplicate_without_prior_wal_backfills_acceptance() { .expect("WAL ACK duplicate should backfill acceptance evidence") }; + assert!(duplicate.duplicate); + assert_eq!(duplicate.submission_id, first.submission_id); + let runtime_wal = host + .runtime_wal() + .expect("runtime WAL should stay configured"); + assert_eq!(runtime_wal.submission_acceptance_count(), 1); + let mut store = runtime_wal.cloned_store(); + let report = recover_in_memory_store(&mut store, RecoveryAccessMode::ReadOnly) + .expect("backfilled acceptance should recover"); + let recovered = recover_submission_index(&report) + .expect("backfilled acceptance should rebuild submission index"); + assert_eq!( + recovered + .get(&first.submission_id) + .expect("submission should recover from committed WAL") + .acceptance + .canonical_envelope_digest, + envelope_digest + ); +} + +#[test] +fn runtime_wal_ack_duplicate_ignores_uncommitted_acceptance_frames() { + let (runtime, worldline_id) = runtime(); + let mut host = + TrustedRuntimeHost::new(runtime, empty_engine()).expect("trusted host should initialize"); + host.enable_in_memory_runtime_wal() + .expect("runtime WAL should initialize"); + + let envelope = eint_envelope(worldline_id); + let first = { + let mut app = host.app(); + app.submit_intent(envelope.clone()) + .expect("legacy non-WAL submission should be accepted") + }; + let mut raw_wal = TrustedRuntimeWal::new_in_memory().expect("test WAL should initialize"); + raw_wal + .append_uncommitted_submission_acceptance_for_test(&envelope, first) + .expect("test fixture should append raw acceptance frames"); + host.replace_runtime_wal_for_test(raw_wal); + assert_eq!( + host.runtime_wal() + .expect("runtime WAL should stay configured") + .submission_acceptance_count(), + 0 + ); + + let duplicate = { + let mut app = host.app(); + app.submit_intent_with_runtime_wal_ack(envelope) + .expect("WAL ACK duplicate should commit recoverable acceptance evidence") + }; + assert!(duplicate.duplicate); assert_eq!(duplicate.submission_id, first.submission_id); assert_eq!( @@ -516,3 +595,272 @@ fn runtime_wal_ack_failure_rolls_back_intake_mutation() { 0 ); } + +#[test] +fn runtime_wal_ack_tick_commits_receipt_transaction_before_outcome_is_observed() { + let (runtime, worldline_id) = runtime(); + let mut host = + TrustedRuntimeHost::new(runtime, empty_engine()).expect("trusted host should initialize"); + host.enable_in_memory_runtime_wal() + .expect("runtime WAL should initialize"); + host.install_contract_package(package()) + .expect("host should install package"); + + let envelope = eint_envelope(worldline_id); + let submission = { + let mut app = host.app(); + app.submit_intent_with_runtime_wal_ack(envelope) + .expect("runtime WAL ACK submit should return accepted evidence") + }; + let ticket = admission_ticket(91); + host.stage_installed_contract_submission(submission.submission_id, &ticket) + .expect("trusted host should stage package-supported ticketed ingress"); + + let report = host + .run_until_idle(4) + .expect("trusted host should tick until idle"); + assert_eq!(report.committed_steps, 1); + + let outcome = { + let app = host.app(); + app.observe_intent_outcome(&submission.submission_id) + }; + let IntentOutcome::Applied { receipt, .. } = outcome else { + panic!("expected applied outcome"); + }; + let tick_receipt_digest = receipt.tick_receipt_digest; + + let runtime_wal = host + .runtime_wal() + .expect("runtime WAL should stay configured"); + assert_eq!( + runtime_wal + .commits() + .into_iter() + .filter(|commit| commit.transaction_kind == WalTransactionKind::SchedulerTick) + .count(), + 1 + ); + + let mut store = runtime_wal.cloned_store(); + let recovery = recover_in_memory_store(&mut store, RecoveryAccessMode::ReadOnly) + .expect("committed tick receipt should recover"); + let submissions = recover_submission_index(&recovery) + .expect("recovered tick receipt should update submission posture"); + assert_eq!( + submissions + .get(&submission.submission_id) + .expect("submission should recover") + .posture, + RecoveredSubmissionPosture::DecidedApplied + ); + + let receipts = + recover_receipt_index(&recovery).expect("recovered tick receipt should rebuild index"); + assert_eq!( + receipts + .receipt_by_submission + .get(&submission.submission_id), + Some(&tick_receipt_digest) + ); + assert_eq!( + receipts.ticket_by_submission.get(&submission.submission_id), + Some(&ticket.ticket_digest) + ); +} + +#[test] +fn runtime_wal_ack_tick_failure_rolls_back_visible_outcome() { + let (runtime, worldline_id) = runtime(); + let mut host = + TrustedRuntimeHost::new(runtime, empty_engine()).expect("trusted host should initialize"); + host.enable_in_memory_runtime_wal() + .expect("runtime WAL should initialize"); + host.install_contract_package(package()) + .expect("host should install package"); + + let submission = { + let mut app = host.app(); + app.submit_intent_with_runtime_wal_ack(eint_envelope(worldline_id)) + .expect("submission acceptance should commit before ACK") + }; + host.stage_installed_contract_submission(submission.submission_id, &admission_ticket(92)) + .expect("trusted host should stage package-supported ticketed ingress"); + let overflowing_wal = TrustedRuntimeWal::new_in_memory_at_lsn_for_test(Lsn::from_raw(u64::MAX)) + .expect("overflow fixture WAL should initialize"); + host.replace_runtime_wal_for_test(overflowing_wal); + + let err = host + .run_until_idle(4) + .expect_err("tick WAL overflow should reject publication"); + assert!(matches!( + err, + TrustedRuntimeHostError::Wal(TrustedRuntimeWalError::Build(WalBuildError::LsnOverflow)) + )); + assert_eq!(host.runtime().receipt_correlation_count(), 0); + assert_eq!( + host.runtime_wal() + .expect("runtime WAL should stay configured") + .scheduler_tick_count(), + 0 + ); + + let outcome = { + let app = host.app(); + app.observe_intent_outcome(&submission.submission_id) + }; + assert!(matches!( + outcome, + IntentOutcome::Pending { + submission_id, + ticketed_ingress_id: Some(_), + .. + } if submission_id == submission.submission_id + )); +} + +#[test] +fn runtime_wal_ack_multi_head_tick_failure_rolls_back_all_tick_records() { + let (runtime, worldline_a, worldline_b) = runtime_pair(); + let mut host = + TrustedRuntimeHost::new(runtime, empty_engine()).expect("trusted host should initialize"); + host.enable_in_memory_runtime_wal() + .expect("runtime WAL should initialize"); + host.install_contract_package(package()) + .expect("host should install package"); + + let submission_a = { + let mut app = host.app(); + app.submit_intent_with_runtime_wal_ack(eint_envelope(worldline_a)) + .expect("first submission acceptance should commit before ACK") + }; + let submission_b = { + let mut app = host.app(); + app.submit_intent_with_runtime_wal_ack(eint_envelope(worldline_b)) + .expect("second submission acceptance should commit before ACK") + }; + host.stage_installed_contract_submission(submission_a.submission_id, &admission_ticket(95)) + .expect("trusted host should stage first ticketed ingress"); + host.stage_installed_contract_submission(submission_b.submission_id, &admission_ticket(96)) + .expect("trusted host should stage second ticketed ingress"); + + let overflowing_wal = + TrustedRuntimeWal::new_in_memory_at_lsn_for_test(Lsn::from_raw(u64::MAX - 5)) + .expect("overflow fixture WAL should initialize"); + host.replace_runtime_wal_for_test(overflowing_wal); + + let err = host + .run_until_idle(4) + .expect_err("second tick WAL transaction should fail after first would have committed"); + assert!(matches!( + err, + TrustedRuntimeHostError::Wal(TrustedRuntimeWalError::Build(WalBuildError::LsnOverflow)) + )); + assert_eq!( + host.runtime().receipt_correlation_count(), + 0, + "failed scheduler pass must not leave receipt correlations visible" + ); + assert_eq!( + host.runtime_wal() + .expect("runtime WAL should stay configured") + .scheduler_tick_count(), + 0, + "failed scheduler pass must roll back every tick WAL record from the attempt" + ); + + for submission_id in [submission_a.submission_id, submission_b.submission_id] { + let outcome = { + let app = host.app(); + app.observe_intent_outcome(&submission_id) + }; + assert!(matches!( + outcome, + IntentOutcome::Pending { + submission_id: observed_submission_id, + ticketed_ingress_id: Some(_), + .. + } if observed_submission_id == submission_id + )); + } +} + +#[test] +fn runtime_wal_ack_recover_read_only_rebuilds_submission_and_receipt_indexes() { + let (runtime, worldline_id) = runtime(); + let mut host = + TrustedRuntimeHost::new(runtime, empty_engine()).expect("trusted host should initialize"); + host.enable_in_memory_runtime_wal() + .expect("runtime WAL should initialize"); + host.install_contract_package(package()) + .expect("host should install package"); + + let submission = { + let mut app = host.app(); + app.submit_intent_with_runtime_wal_ack(eint_envelope(worldline_id)) + .expect("submission acceptance should commit before ACK") + }; + let ticket = admission_ticket(93); + host.stage_installed_contract_submission(submission.submission_id, &ticket) + .expect("trusted host should stage package-supported ticketed ingress"); + host.run_until_idle(4) + .expect("trusted host should tick until idle"); + + let recovery = host + .runtime_wal() + .expect("runtime WAL should stay configured") + .recover_read_only() + .expect("runtime WAL recovery should rebuild indexes"); + + assert_eq!( + recovery + .submissions + .get(&submission.submission_id) + .expect("submission should recover") + .posture, + RecoveredSubmissionPosture::DecidedApplied + ); + assert_eq!( + recovery + .receipts + .ticket_by_submission + .get(&submission.submission_id), + Some(&ticket.ticket_digest) + ); + assert_eq!( + recovery.certificate.recovered_indexes_root, + recovered_submission_receipt_index_root(&recovery.submissions, &recovery.receipts) + ); +} + +#[test] +fn runtime_wal_ack_recover_read_only_exposes_recovery_certificate() { + let (runtime, worldline_id) = runtime(); + let mut host = + TrustedRuntimeHost::new(runtime, empty_engine()).expect("trusted host should initialize"); + host.enable_in_memory_runtime_wal() + .expect("runtime WAL should initialize"); + host.install_contract_package(package()) + .expect("host should install package"); + + let submission = { + let mut app = host.app(); + app.submit_intent_with_runtime_wal_ack(eint_envelope(worldline_id)) + .expect("submission acceptance should commit before ACK") + }; + host.stage_installed_contract_submission(submission.submission_id, &admission_ticket(94)) + .expect("trusted host should stage package-supported ticketed ingress"); + host.run_until_idle(4) + .expect("trusted host should tick until idle"); + + let recovery = host + .runtime_wal() + .expect("runtime WAL should stay configured") + .recover_read_only() + .expect("runtime WAL recovery should expose certificate"); + + assert_eq!(recovery.certificate.committed_transactions_replayed, 2); + assert_eq!(recovery.certificate.obstruction_count, 0); + assert!(recovery.certificate.first_lsn.is_some()); + assert!(recovery.certificate.last_lsn.is_some()); +} diff --git a/docs/BEARING.md b/docs/BEARING.md index 5bc6a5bf..a450f33d 100644 --- a/docs/BEARING.md +++ b/docs/BEARING.md @@ -3,7 +3,7 @@ # BEARING -Last updated: 2026-05-24. +Last updated: 2026-05-25. This signpost summarizes current direction. It does not create commitments or replace backlog items, design docs, retros, or CLI status. If it disagrees with @@ -959,7 +959,7 @@ completed slice. - Test plan: `runtime_wal_ack_failure_rolls_back_intake_mutation` and `runtime_wal_ack_path_requires_configured_runtime_wal`. -- [ ] **Slice 90: Tick receipt transaction wiring** +- [x] **Slice 90: Tick receipt transaction wiring** - User story: As Echo, visible tick receipts should eventually be backed by committed scheduler-tick WAL transactions. - Acceptance criteria: host-owned scheduler runs record receipt and @@ -967,21 +967,21 @@ completed slice. - Test plan: trusted-host applied-intent fixture plus recovered receipt index witness. -- [ ] **Slice 91: Tick commit-before-publish rollback guard** +- [x] **Slice 91: Tick commit-before-publish rollback guard** - User story: As Echo, a tick WAL failure must not leave a half-visible receipt/outcome. - Acceptance criteria: tick WAL failure either restores runtime/provenance state or blocks receipt publication under a typed runtime fault posture. - Test plan: injected tick-WAL failure fixture. -- [ ] **Slice 92: Runtime index rebuild contract** +- [x] **Slice 92: Runtime index rebuild contract** - User story: As recovery, WAL-backed submission and receipt indexes should rebuild without scheduler callbacks. - Acceptance criteria: recovered indexes answer pending/applied/rejected posture from committed WAL transactions only. - Test plan: pure in-memory recovery fixture for submit plus tick records. -- [ ] **Slice 93: WAL-backed recovery certificate in runtime** +- [x] **Slice 93: WAL-backed recovery certificate in runtime** - User story: As an operator, restart should produce inspectable evidence about what committed history was replayed. - Acceptance criteria: recovery certificate covers checkpoint, LSN range, @@ -989,20 +989,23 @@ completed slice. - Test plan: recovery certificate fixture over committed and truncated-tail WAL shapes. -- [ ] **Slice 94: `jedit` recovery fixture contract** - - User story: As a real app consumer, `jedit` should be able to distinguish - not-accepted, accepted-pending, decided, rejected, and obstructed edits - from Echo recovery evidence. +- [x] **Slice 94: Echo recovery posture contract** + - User story: As a real app consumer, sibling applications should be able + to distinguish not-accepted, accepted-pending, decided, rejected, and + obstructed work from Echo recovery evidence without Echo importing + application nouns. - Acceptance criteria: Echo exposes only generic submission/receipt posture; - `jedit` maps that posture to editor terms outside Echo. - - Test plan: sibling `jedit` fixture consumes generic Echo recovery JSON. + sibling applications map that posture to product terms outside Echo. + - Test plan: Echo CLI fixture emits generic recovery JSON; the sibling + `jedit` fixture remains the next cross-repo consumer witness. -- [ ] **Slice 95: Runtime ACK drift gate** +- [x] **Slice 95: Runtime ACK drift gate** - User story: As a maintainer, docs and tests should fail if Echo claims durable ACK semantics without a WAL-backed witness. - Acceptance criteria: release readiness names runtime ACK coverage as a - distinct gate. - - Test plan: runtime ACK readiness gate fixture plus stale-claim grep. + distinct gate through `cargo xtask test-slice runtime-wal-ack`. + - Test plan: runtime ACK readiness gate fixture plus stale-claim grep: + `cargo xtask test-slice runtime-wal-ack`. ## Recently Completed Slice Batch diff --git a/docs/design/v0.1.0-jedit-release-gate.md b/docs/design/v0.1.0-jedit-release-gate.md index ebb5a850..d6fe0c83 100644 --- a/docs/design/v0.1.0-jedit-release-gate.md +++ b/docs/design/v0.1.0-jedit-release-gate.md @@ -124,6 +124,16 @@ product-shaped text result. An MCP wrapper can come later, but it should wrap the same CLI once retained evidence and replay output are strong enough to expose through a protocol. +Echo's generic WAL recovery posture surface is `echo-cli wal +submission-posture --format json`. The current Echo-side contract reports only +submission ids, canonical envelope digests, ticket digests, receipt digests, and +recovered posture labels. `crates/warp-cli/src/wal.rs` is the authoritative CLI +label source for this release-gate fixture. The next sibling jedit fixture +should call this surface through its Echo adapter to classify a submitted edit +as not accepted, accepted pending, decided applied, decided rejected, +obstructed, or recovery faulted. jedit owns the mapping from those generic +labels to editor language. + jedit keeps sibling Echo checkout paths and process execution behind its witness-runner port/adapter. The CLI is not the authority boundary; it delegates to a Node adapter that owns filesystem and process details while the witness diff --git a/docs/man/echo-cli-wal-submission-posture.1 b/docs/man/echo-cli-wal-submission-posture.1 new file mode 100644 index 00000000..5d8135db --- /dev/null +++ b/docs/man/echo-cli-wal-submission-posture.1 @@ -0,0 +1,22 @@ +.ie \n(.g .ds Aq \(aq +.el .ds Aq ' +.TH echo-cli-wal-submission-posture 1 "echo-cli-wal-submission-posture " +.SH NAME +echo\-cli\-wal\-submission\-posture \- Report recovered posture for one submission id/envelope pair +.SH SYNOPSIS +\fBecho\-cli\-wal\-submission\-posture\fR <\fB\-\-submission\-id\fR> <\fB\-\-canonical\-envelope\-digest\fR> [\fB\-h\fR|\fB\-\-help\fR] <\fIROOT\fR> +.SH DESCRIPTION +Report recovered posture for one submission id/envelope pair +.SH OPTIONS +.TP +\fB\-\-submission\-id\fR \fI\fR +64\-character hex submission id +.TP +\fB\-\-canonical\-envelope\-digest\fR \fI\fR +64\-character hex canonical envelope digest +.TP +\fB\-h\fR, \fB\-\-help\fR +Print help +.TP +<\fIROOT\fR> +Filesystem WAL root to inspect diff --git a/docs/man/echo-cli-wal.1 b/docs/man/echo-cli-wal.1 index 94fef728..aa499ac3 100644 --- a/docs/man/echo-cli-wal.1 +++ b/docs/man/echo-cli-wal.1 @@ -16,5 +16,8 @@ Print help echo\-cli\-wal\-doctor(1) Report read\-only WAL recovery posture .TP +echo\-cli\-wal\-submission\-posture(1) +Report recovered posture for one submission id/envelope pair +.TP echo\-cli\-wal\-help(1) Print this message or the help of the given subcommand(s) diff --git a/docs/workflows.md b/docs/workflows.md index d745156d..8e4b7d1f 100644 --- a/docs/workflows.md +++ b/docs/workflows.md @@ -142,6 +142,7 @@ The repo also exposes maintenance commands via `cargo xtask …`: - `cargo xtask test-slice neighborhood` runs only `warp-core` neighborhood module unit tests via `--lib`. - `cargo xtask test-slice warp-core-smoke` runs the `warp-core` lib tests plus the strand integration target without launching every integration-test binary. - `cargo xtask test-slice contract-path-release` runs the v0.1 local contract-host release witness: installed contract pipeline replay, reference trusted host loop, and the serious external consumer fixture. +- `cargo xtask test-slice runtime-wal-ack` runs the runtime WAL-backed ACK witness: app-facing acceptance rollback, scheduler tick receipt invariant checks, scheduler tick commit-before-publish, recovered indexes, CLI submission posture JSON, stale-claim guard, and generated man-page check. - `cargo xtask pr-preflight` runs the default changed-scope pre-PR gate against `origin/main`. - `cargo xtask pr-preflight --full` runs the broader explicit full pre-PR gate. - `cargo xtask dind` runs the DIND (Deterministic Ironclad Nightmare Drills) harness locally. diff --git a/xtask/src/main.rs b/xtask/src/main.rs index 0f9b48d6..26491e17 100644 --- a/xtask/src/main.rs +++ b/xtask/src/main.rs @@ -91,6 +91,8 @@ enum TestSlice { WarpCoreSmoke, /// v0.1 local contract-host replay/release witness. ContractPathRelease, + /// Runtime WAL-backed ACK and recovery posture witness. + RuntimeWalAck, } #[derive(Args)] @@ -605,6 +607,37 @@ fn build_test_slice_commands(slice: TestSlice) -> Vec { "external_consumer_contract_fixture_tests", ]), ], + TestSlice::RuntimeWalAck => vec![ + cargo_command([ + "test", + "-p", + "warp-core", + "--features", + "native_rule_bootstrap trusted_runtime host_test", + "--lib", + "trusted_runtime_host::tests::runtime_wal_tick_decision", + ]), + cargo_command([ + "test", + "-p", + "warp-core", + "--features", + "native_rule_bootstrap trusted_runtime host_test", + "--test", + "trusted_runtime_host_loop_tests", + "runtime_wal_ack", + ]), + cargo_command([ + "test", + "-p", + "warp-cli", + "--test", + "cli_integration", + "wal_submission_posture", + ]), + cargo_command(["test", "-p", "xtask", "runtime_wal_ack_stale_claims"]), + cargo_command(["xtask", "man-pages", "--check"]), + ], } } @@ -6379,6 +6412,111 @@ mod tests { ); } + #[test] + fn test_slice_runtime_wal_ack_stays_explicit() { + let commands = build_test_slice_commands(TestSlice::RuntimeWalAck); + assert_eq!(commands.len(), 5); + + let (program, args) = command_program_and_args(&commands[0]); + assert_eq!(program, "cargo"); + assert_eq!( + args, + vec![ + "test", + "-p", + "warp-core", + "--features", + "native_rule_bootstrap trusted_runtime host_test", + "--lib", + "trusted_runtime_host::tests::runtime_wal_tick_decision", + ] + ); + + let (program, args) = command_program_and_args(&commands[1]); + assert_eq!(program, "cargo"); + assert_eq!( + args, + vec![ + "test", + "-p", + "warp-core", + "--features", + "native_rule_bootstrap trusted_runtime host_test", + "--test", + "trusted_runtime_host_loop_tests", + "runtime_wal_ack", + ] + ); + + let (program, args) = command_program_and_args(&commands[2]); + assert_eq!(program, "cargo"); + assert_eq!( + args, + vec![ + "test", + "-p", + "warp-cli", + "--test", + "cli_integration", + "wal_submission_posture", + ] + ); + + let (program, args) = command_program_and_args(&commands[3]); + assert_eq!(program, "cargo"); + assert_eq!( + args, + vec!["test", "-p", "xtask", "runtime_wal_ack_stale_claims"] + ); + + let (program, args) = command_program_and_args(&commands[4]); + assert_eq!(program, "cargo"); + assert_eq!(args, vec!["xtask", "man-pages", "--check"]); + } + + #[test] + fn runtime_wal_ack_stale_claims_stay_current() { + let repo_root = Path::new(env!("CARGO_MANIFEST_DIR")) + .parent() + .expect("xtask crate should live under repository root"); + let checked_docs = [ + "docs/BEARING.md", + "docs/design/v0.1.0-jedit-release-gate.md", + "docs/design/causal-wal-hardening-matrix.md", + "docs/workflows.md", + ]; + let stale_claims = [ + "durable ACK semantics are unimplemented", + "runtime ACK coverage remains ungated", + "tick receipts are visible before WAL commit", + "accepted submission evidence can be returned before WAL commit", + "jedit recovery fixture contract is still missing", + ]; + + for relative_path in checked_docs { + let path = repo_root.join(relative_path); + let content = fs::read_to_string(&path) + .unwrap_or_else(|err| panic!("failed to read {}: {err}", path.display())); + for stale_claim in stale_claims { + assert!( + !content.contains(stale_claim), + "{} contains stale runtime WAL ACK claim `{}`", + relative_path, + stale_claim + ); + } + } + + let bearing = fs::read_to_string(repo_root.join("docs/BEARING.md")) + .expect("BEARING should be readable"); + assert!(bearing.contains("Runtime ACK drift gate")); + assert!(bearing.contains("cargo xtask test-slice runtime-wal-ack")); + + let workflows = fs::read_to_string(repo_root.join("docs/workflows.md")) + .expect("workflow docs should be readable"); + assert!(workflows.contains("cargo xtask test-slice runtime-wal-ack")); + } + #[test] fn wesley_sync_extracts_rust_and_typescript_schema_hashes() { let hash = "d55d6000b43562e7be04702cdd4335452d1eb6df1f0fbea924e4c6434fff2871";