diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 6bc9a67c..49f79dc5 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -153,6 +153,10 @@ jobs:
               run: |
                   cargo clippy -p warp-core --features native_rule_bootstrap,trusted_runtime --test edict_projected_cas_tests -- -D warnings -D missing_docs
                   cargo clippy -p warp-core --features native_rule_bootstrap,trusted_runtime,host_test --test edict_projected_cas_tests -- -D warnings -D missing_docs
+            - name: cargo clippy (trusted host loop and WAL fault injection)
+              run: |
+                  cargo clippy -p warp-core --features native_rule_bootstrap,trusted_runtime,host_test --test trusted_runtime_host_loop_tests -- -D warnings -D missing_docs
+                  cargo clippy -p warp-core --features host_test --test causal_wal_hardening_tests -- -D warnings -D missing_docs
 
             - name: cargo clippy (compiler-produced pure evaluation)
               run: cargo clippy -p warp-core --features trusted_runtime --lib --test edict_pure_evaluation_tests --test edict_pure_unsigned_subtraction_tests --test edict_pure_byte_length_tests --test edict_byte_equality_tests --test edict_byte_slice_tests --test edict_node_read_tests --test edict_byte_concat_tests --test edict_source_functions_tests -- -D warnings -D missing_docs
@@ -328,6 +332,10 @@ jobs:
               run: |
                   cargo test -p warp-core --features native_rule_bootstrap,trusted_runtime --test edict_projected_cas_tests
                   cargo test -p warp-core --features native_rule_bootstrap,trusted_runtime,host_test --test edict_projected_cas_tests
+            - name: cargo test (trusted host loop and WAL fault injection)
+              run: |
+                  cargo test -p warp-core --features native_rule_bootstrap,trusted_runtime,host_test --test trusted_runtime_host_loop_tests
+                  cargo test -p warp-core --features host_test --test causal_wal_hardening_tests
             - name: cargo test (compiler-produced pure evaluation)
               run: |
                   cargo test -p warp-core --features trusted_runtime --test edict_pure_evaluation_tests --test edict_pure_unsigned_subtraction_tests --test edict_pure_byte_length_tests --test edict_byte_equality_tests --test edict_byte_slice_tests --test edict_node_read_tests --test edict_byte_concat_tests --test edict_source_functions_tests
diff --git a/CHANGELOG.md b/CHANGELOG.md
index d6c6a10f..af1e4e9b 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -7,6 +7,8 @@
 
 ### Added
 
+- `submit_intents_with_runtime_wal_ack` accepts several intents and returns after one storage sync has made all of their acceptance transactions durable. Each intent writes the same acceptance transaction as a single submission, so recovery is unchanged. No acceptance of a batch counts as durable, or can be scheduled, before that sync succeeds. A failed batch acknowledges nothing. If the runtime refuses an intent before anything is written for it, the intents before it stay accepted and are synced. If an append or the sync fails, the call's intake is rolled back and the WAL refuses further work with `StorageFailed` until it is reopened.
+
 - `SlotWriteIndex` resolves the value of one node, edge or attachment slot at a tick cursor from tick patches and the initial boundary, without materializing a graph or replaying unrelated ticks. It indexes the operations that wrote each slot, including values cleared by node, edge and instance deletion, and truncates in proportion to the ticks removed. A reader is bound to the exact patches the index was built from by patch digest, and truncation refuses patches that are not the indexed suffix. It is a derived reading aid; the runtime does not use it yet.
 
 - Native projected compare-and-set invocation encoding retains canonical application input and statically binds the expected attachment-value digest. The native lowerer and independent verifier reconstruct CAS-specific profiles and reject rebound digest paths or under-budget configurations. Generated schema, reproducible components and a retained public Edict compiler build establish package production. The retained compiler package now has scheduler execution, pending/committed Action recovery, stale-digest noncommit recovery and typed refusal witnesses. `EchoOperationInvocationV1::anchored_node_attachment_compare_and_set_with_application_input` and `ExecutableOperationPackageV1::with_compare_and_set_application_result_projection` construct the projected forms without byte editing.
@@ -31,6 +33,8 @@
 
 ### Fixed
 
+- A filesystem WAL writer that takes over an existing log syncs its segment files before reading them as recovered history. A record that an earlier writer wrote and never synced was readable after reopen and was reported as a durable acceptance; a later power loss could still lose it.
+
 - A scheduler pass that commits executable-operation Actions hashes the worldline's whole reachable state twice instead of three times. The root before the commit is the root the pass already computed for the frontier basis; the commit no longer computes it again.
 
 - A scheduler pass resolves a writer head's frontier basis once instead of once per admitted executable-operation Action. Resolving it hashes the whole reachable state, so a tick's cost grew with Actions times graph size. The bases produced are unchanged.
diff --git a/crates/warp-core/src/causal_wal.rs b/crates/warp-core/src/causal_wal.rs
index e3b4b41f..95772b28 100644
--- a/crates/warp-core/src/causal_wal.rs
+++ b/crates/warp-core/src/causal_wal.rs
@@ -5517,6 +5517,12 @@ pub enum FilesystemWalFaultTarget {
     CommitMarkerSynced,
     /// Fail the next manifest publish before writing manifest material.
     PublishManifest,
+    /// Fail the next storage sync that would make deferred commit records
+    /// durable.
+    DeferredCommitSync,
+    /// Fail the next sync of existing segments by a writer that takes over a
+    /// log.
+    RecoveredSegmentSync,
 }
 
 /// Host-test-only filesystem WAL fault plan.
@@ -5527,10 +5533,22 @@ pub struct FilesystemWalFaultPlan {
     flush_commit: u32,
     commit_marker_synced: u32,
     publish_manifest: u32,
+    deferred_commit_sync: u32,
+    recovered_segment_sync: u32,
+    /// Operations of the failing target to let through first.
+    skip: u32,
 }
 
 #[cfg(any(test, feature = "host_test"))]
 impl FilesystemWalFaultPlan {
+    /// Lets `skipped` operations of the failing target succeed before the
+    /// failure.
+    #[must_use]
+    pub const fn after(mut self, skipped: u32) -> Self {
+        self.skip = skipped;
+        self
+    }
+
     /// Builds a plan that fails the next operation for `target`.
     #[must_use]
     pub const fn fail_next(target: FilesystemWalFaultTarget) -> Self {
@@ -5540,24 +5558,54 @@ impl FilesystemWalFaultPlan {
                 flush_commit: 0,
                 commit_marker_synced: 0,
                 publish_manifest: 0,
+                deferred_commit_sync: 0,
+                recovered_segment_sync: 0,
+                skip: 0,
             },
             FilesystemWalFaultTarget::FlushCommit => Self {
                 append_frame: 0,
                 flush_commit: 1,
                 commit_marker_synced: 0,
                 publish_manifest: 0,
+                deferred_commit_sync: 0,
+                recovered_segment_sync: 0,
+                skip: 0,
             },
             FilesystemWalFaultTarget::CommitMarkerSynced => Self {
                 append_frame: 0,
                 flush_commit: 0,
                 commit_marker_synced: 1,
                 publish_manifest: 0,
+                deferred_commit_sync: 0,
+                recovered_segment_sync: 0,
+                skip: 0,
             },
             FilesystemWalFaultTarget::PublishManifest => Self {
                 append_frame: 0,
                 flush_commit: 0,
                 commit_marker_synced: 0,
                 publish_manifest: 1,
+                deferred_commit_sync: 0,
+                recovered_segment_sync: 0,
+                skip: 0,
+            },
+            FilesystemWalFaultTarget::DeferredCommitSync => Self {
+                append_frame: 0,
+                flush_commit: 0,
+                commit_marker_synced: 0,
+                publish_manifest: 0,
+                deferred_commit_sync: 1,
+                recovered_segment_sync: 0,
+                skip: 0,
+            },
+            FilesystemWalFaultTarget::RecoveredSegmentSync => Self {
+                append_frame: 0,
+                flush_commit: 0,
+                commit_marker_synced: 0,
+                publish_manifest: 0,
+                deferred_commit_sync: 0,
+                recovered_segment_sync: 1,
+                skip: 0,
             },
         }
     }
@@ -5568,10 +5616,16 @@ impl FilesystemWalFaultPlan {
             FilesystemWalFaultTarget::FlushCommit => &mut self.flush_commit,
             FilesystemWalFaultTarget::CommitMarkerSynced => &mut self.commit_marker_synced,
             FilesystemWalFaultTarget::PublishManifest => &mut self.publish_manifest,
+            FilesystemWalFaultTarget::DeferredCommitSync => &mut self.deferred_commit_sync,
+            FilesystemWalFaultTarget::RecoveredSegmentSync => &mut self.recovered_segment_sync,
         };
         if *remaining == 0 {
             return false;
         }
+        if self.skip > 0 {
+            self.skip -= 1;
+            return false;
+        }
         *remaining -= 1;
         true
     }
@@ -5588,6 +5642,16 @@ pub struct FilesystemWalStore {
     writer_lock: Option<WriterEpochLock>,
     manifests: Vec<WalManifest>,
     sync_evidence: Vec<FilesystemSyncEvidence>,
+    /// Transactions whose commit records are appended and not yet synced.
+    deferred_commits: Vec<WalTransactionId>,
+    /// Set when a sync of deferred commit records fails. A later sync cannot
+    /// vouch for records an earlier sync failed to write, so the store then
+    /// refuses to append, sync, rotate or close until it is reopened.
+    deferred_commit_sync_failed: bool,
+    /// Number of storage syncs that made commit records durable.
+    #[cfg(any(test, feature = "host_test"))]
+    commit_sync_count: u64,
+    recovered_segment_sync_count: u64,
     #[cfg(any(test, feature = "host_test"))]
     fault_plan: FilesystemWalFaultPlan,
 }
@@ -5638,6 +5702,11 @@ impl FilesystemWalStore {
             writer_lock: None,
             manifests: Vec::new(),
             sync_evidence,
+            deferred_commits: Vec::new(),
+            deferred_commit_sync_failed: false,
+            #[cfg(any(test, feature = "host_test"))]
+            commit_sync_count: 0,
+            recovered_segment_sync_count: 0,
             #[cfg(any(test, feature = "host_test"))]
             fault_plan: FilesystemWalFaultPlan::default(),
         })
@@ -5745,6 +5814,9 @@ impl FilesystemWalStore {
         }
         let writer_lock = acquire_writer_epoch_lock(&self.root)?;
         self.reload_writer_epoch_ledger()?;
+        // Make what the earlier writer left readable durable before it is
+        // read as recovered history.
+        self.sync_existing_segments()?;
         let recovery = recover_filesystem_store(&self.root, RecoveryAccessMode::ReadOnly).map_err(
             |error| match error {
                 WalRecoveryError::Store(error) => error,
@@ -5865,6 +5937,31 @@ impl FilesystemWalStore {
         &mut self,
         transaction: WalCommittedTransaction,
     ) -> Result<(), WalStoreError> {
+        self.append_transaction_with_commit_sync(transaction, true)
+    }
+
+    /// Appends a transaction without syncing its commit record.
+    ///
+    /// The transaction is not durable until [`Self::sync_deferred_commits`]
+    /// or a later synced commit on the same segment succeeds. The caller must
+    /// not report it as durable before then.
+    ///
+    /// # Errors
+    ///
+    /// Returns the same errors as [`Self::append_transaction`].
+    pub fn append_transaction_deferring_commit_sync(
+        &mut self,
+        transaction: WalCommittedTransaction,
+    ) -> Result<(), WalStoreError> {
+        self.append_transaction_with_commit_sync(transaction, false)
+    }
+
+    fn append_transaction_with_commit_sync(
+        &mut self,
+        transaction: WalCommittedTransaction,
+        sync_now: bool,
+    ) -> Result<(), WalStoreError> {
+        self.ensure_deferred_commit_sync_has_not_failed()?;
         transaction.validate()?;
         let admission_kernel_capability = transaction.admission_kernel_capability;
         let external_action_coordinator_capability =
@@ -5878,6 +5975,7 @@ impl FilesystemWalStore {
             transaction.commit,
             admission_kernel_capability,
             external_action_coordinator_capability,
+            sync_now,
         )
     }
 
@@ -5887,7 +5985,9 @@ impl FilesystemWalStore {
         commit: WalTransactionCommit,
         admission_kernel_capability: Option<AdmissionKernelCapability>,
         external_action_coordinator_capability: Option<ExternalActionCoordinatorCapability>,
+        sync_now: bool,
     ) -> Result<(), WalStoreError> {
+        self.ensure_deferred_commit_sync_has_not_failed()?;
         self.ensure_writer_lock()?;
         let active_epoch = self
             .active_epoch
@@ -5902,16 +6002,33 @@ impl FilesystemWalStore {
             external_action_coordinator_capability,
         )?;
         #[cfg(any(test, feature = "host_test"))]
-        if self
+        let injected = self
             .fault_plan
-            .should_fail(FilesystemWalFaultTarget::FlushCommit)
-        {
-            return Err(WalStoreError::Io(
+            .should_fail(FilesystemWalFaultTarget::FlushCommit);
+        #[cfg(not(any(test, feature = "host_test")))]
+        let injected = false;
+        let transaction_id = commit.transaction_id;
+        let flushed = if injected {
+            Err(WalStoreError::Io(
                 "injected filesystem WAL flush_commit failure".to_owned(),
-            ));
+            ))
+        } else {
+            append_segment_record(
+                &self.segment_path(),
+                DiskWalRecord::Commit(&commit),
+                sync_now,
+            )
+        };
+        if let Err(error) = flushed {
+            if sync_now && !self.deferred_commits.is_empty() {
+                // This commit's sync would also have covered the deferred
+                // records. It failed, so they are in the same position as
+                // after a failed deferred sync.
+                self.deferred_commit_sync_failed = true;
+                self.deferred_commits.clear();
+            }
+            return Err(error);
         }
-        let transaction_id = commit.transaction_id;
-        append_segment_record(&self.segment_path(), DiskWalRecord::Commit(&commit), true)?;
         self.epoch_closures.insert(
             epoch_id,
             WriterEpochClosure {
@@ -5919,22 +6036,164 @@ impl FilesystemWalStore {
                 final_commit_digest: Some(commit.commit_digest),
             },
         );
-        self.sync_evidence.push(FilesystemSyncEvidence::transaction(
-            FilesystemSyncBoundary::CommitFileSynced,
-            transaction_id,
-        ));
+        self.deferred_commits.push(transaction_id);
+        if sync_now {
+            // One sync of the segment file covers this commit record and any
+            // earlier record that was appended without its own sync.
+            self.record_commit_records_synced();
+        }
+        // The marker-synced fault models a failure after this commit's own
+        // sync. A deferred commit has no sync of its own to fail after.
+        #[cfg(any(test, feature = "host_test"))]
+        if sync_now
+            && self
+                .fault_plan
+                .should_fail(FilesystemWalFaultTarget::CommitMarkerSynced)
+        {
+            return Err(WalStoreError::Io(
+                "injected filesystem WAL commit_marker_synced failure".to_owned(),
+            ));
+        }
+        Ok(())
+    }
+
+    fn record_commit_records_synced(&mut self) {
+        #[cfg(any(test, feature = "host_test"))]
+        {
+            self.commit_sync_count += 1;
+        }
+        for transaction_id in self.deferred_commits.drain(..) {
+            self.sync_evidence.push(FilesystemSyncEvidence::transaction(
+                FilesystemSyncBoundary::CommitFileSynced,
+                transaction_id,
+            ));
+        }
+    }
+
+    /// Makes every commit record that was appended without its own sync
+    /// durable with one storage sync.
+    ///
+    /// # Errors
+    ///
+    /// Returns a store error if the segment cannot be synced. Nothing appended
+    /// without a sync may be reported as durable after such an error.
+    ///
+    /// If the sync fails, the deferred records are never evidenced as synced
+    /// and the store refuses further writes until it is reopened.
+    pub fn sync_deferred_commits(&mut self) -> Result<(), WalStoreError> {
+        self.ensure_deferred_commit_sync_has_not_failed()?;
+        if self.deferred_commits.is_empty() {
+            return Ok(());
+        }
+        #[cfg(any(test, feature = "host_test"))]
+        let injected = self
+            .fault_plan
+            .should_fail(FilesystemWalFaultTarget::DeferredCommitSync);
+        #[cfg(not(any(test, feature = "host_test")))]
+        let injected = false;
+        let synced = if injected {
+            Err(WalStoreError::Io(
+                "injected filesystem WAL deferred_commit_sync failure".to_owned(),
+            ))
+        } else {
+            self.sync_segment()
+        };
+        if let Err(error) = synced {
+            self.deferred_commit_sync_failed = true;
+            self.deferred_commits.clear();
+            return Err(error);
+        }
+        self.record_commit_records_synced();
+        Ok(())
+    }
+
+    /// Syncs every segment file under the store root, and the directories
+    /// that name them.
+    ///
+    /// An earlier writer may have stopped, or failed a sync, after writing
+    /// records. They are readable and not yet durable. Call this before
+    /// reading an existing log as recovered history; taking the writer lock
+    /// does it as well.
+    ///
+    /// # Errors
+    ///
+    /// Returns the storage error if a segment or its directory cannot be
+    /// synced. The log must then not be treated as recovered.
+    pub fn sync_existing_segments(&mut self) -> Result<(), WalStoreError> {
         #[cfg(any(test, feature = "host_test"))]
         if self
             .fault_plan
-            .should_fail(FilesystemWalFaultTarget::CommitMarkerSynced)
+            .should_fail(FilesystemWalFaultTarget::RecoveredSegmentSync)
         {
             return Err(WalStoreError::Io(
-                "injected filesystem WAL commit_marker_synced failure".to_owned(),
+                "injected filesystem WAL recovered_segment_sync failure".to_owned(),
             ));
         }
+        let mut synced = 0_u64;
+        for scan_root in segment_scan_roots(&self.root) {
+            if !scan_root.exists() {
+                continue;
+            }
+            let mut paths = Vec::new();
+            for entry in fs::read_dir(&scan_root)? {
+                let path = entry?.path();
+                if path
+                    .file_stem()
+                    .and_then(|name| name.to_str())
+                    .is_some_and(|name| name.starts_with("segment-"))
+                    && path
+                        .extension()
+                        .is_some_and(|extension| extension.eq_ignore_ascii_case("ecwal"))
+                {
+                    paths.push(path);
+                }
+            }
+            if paths.is_empty() {
+                continue;
+            }
+            paths.sort();
+            for path in paths {
+                File::options().append(true).open(path)?.sync_all()?;
+                synced += 1;
+            }
+            sync_directory_store(&scan_root)?;
+        }
+        self.recovered_segment_sync_count += synced;
+        Ok(())
+    }
+
+    /// Returns how many existing segment files this store synced when it took
+    /// over the log.
+    #[cfg(any(test, feature = "host_test"))]
+    #[must_use]
+    pub fn recovered_segment_sync_count_for_test(&self) -> u64 {
+        self.recovered_segment_sync_count
+    }
+
+    fn sync_segment(&self) -> Result<(), WalStoreError> {
+        File::options()
+            .append(true)
+            .open(self.segment_path())?
+            .sync_all()?;
         Ok(())
     }
 
+    fn ensure_deferred_commit_sync_has_not_failed(&self) -> Result<(), WalStoreError> {
+        if self.deferred_commit_sync_failed {
+            return Err(WalStoreError::Io(
+                "a deferred commit sync failed; reopen the WAL store".to_owned(),
+            ));
+        }
+        Ok(())
+    }
+
+    /// Returns how many storage syncs have made commit records durable.
+    #[cfg(any(test, feature = "host_test"))]
+    #[must_use]
+    pub fn commit_sync_count_for_test(&self) -> u64 {
+        self.commit_sync_count
+    }
+
     /// Appends an uncommitted frame to simulate a torn transaction tail.
     pub fn append_uncommitted_frame(
         &mut self,
@@ -5950,6 +6209,8 @@ impl FilesystemWalStore {
         epoch_id: WriterEpochId,
     ) -> Result<WalSegmentSeal, WalStoreError> {
         self.ensure_writer_lock()?;
+        // A sealed segment must not hold a commit record that was never synced.
+        self.sync_deferred_commits()?;
         let active_epoch = self
             .active_epoch
             .as_ref()
@@ -5993,6 +6254,10 @@ impl WalStorePort for FilesystemWalStore {
         };
         if writer_lock.is_some() {
             self.reload_writer_epoch_ledger()?;
+            // An earlier writer may have stopped, or failed a sync, after
+            // writing records. They are readable. Make them durable before
+            // this writer treats the readable log as recovered history.
+            self.sync_existing_segments()?;
         }
         let epoch = validate_writer_epoch_request(
             self.active_epoch.as_ref(),
@@ -6016,6 +6281,7 @@ impl WalStorePort for FilesystemWalStore {
         epoch_id: WriterEpochId,
         frame: WalFrame,
     ) -> Result<(), WalStoreError> {
+        self.ensure_deferred_commit_sync_has_not_failed()?;
         self.ensure_writer_lock()?;
         let active_epoch = self
             .active_epoch
@@ -6048,7 +6314,7 @@ impl WalStorePort for FilesystemWalStore {
         epoch_id: WriterEpochId,
         commit: WalTransactionCommit,
     ) -> Result<(), WalStoreError> {
-        self.flush_commit_with_capabilities(epoch_id, commit, None, None)
+        self.flush_commit_with_capabilities(epoch_id, commit, None, None, true)
     }
 
     fn flush_external_action_commit(
@@ -6057,7 +6323,7 @@ impl WalStorePort for FilesystemWalStore {
         commit: WalTransactionCommit,
         capability: ExternalActionCoordinatorCapability,
     ) -> Result<(), WalStoreError> {
-        self.flush_commit_with_capabilities(epoch_id, commit, None, Some(capability))
+        self.flush_commit_with_capabilities(epoch_id, commit, None, Some(capability), true)
     }
 
     fn read_frames(&self) -> Vec<WalFrame> {
@@ -6091,6 +6357,9 @@ impl WalStorePort for FilesystemWalStore {
         if active_epoch.epoch_id != epoch_id {
             return Err(WalStoreError::WriterEpochMismatch);
         }
+        // A seal names committed transactions. It must not exist before
+        // their commit records are durable.
+        self.sync_deferred_commits()?;
         let (frames, _, _) = read_segment_file(&segment_path(&self.root, segment_id))?;
         let frame_refs = frames.iter().collect::<Vec<_>>();
         let seal = WalSegmentSeal {
@@ -6102,6 +6371,15 @@ impl WalStorePort for FilesystemWalStore {
     }
 
     fn truncate_tail_after(&mut self, after_lsn: Lsn) -> Result<(), WalStoreError> {
+        self.ensure_deferred_commit_sync_has_not_failed()?;
+        if !self.deferred_commits.is_empty() {
+            // Truncation could remove a record whose sync is still owed, and
+            // a later sync would then evidence a record the log no longer
+            // holds.
+            return Err(WalStoreError::Io(
+                "deferred commit records must be synced before truncation".to_owned(),
+            ));
+        }
         rewrite_filesystem_segments_after_truncation(&self.root, after_lsn)
     }
 
@@ -6118,6 +6396,9 @@ impl WalStorePort for FilesystemWalStore {
         if active_epoch.epoch_id != epoch_id {
             return Err(WalStoreError::WriterEpochMismatch);
         }
+        // A manifest names committed records. It must not become durable
+        // before they are.
+        self.sync_deferred_commits()?;
         #[cfg(any(test, feature = "host_test"))]
         if self
             .fault_plan
@@ -6140,6 +6421,8 @@ impl WalStorePort for FilesystemWalStore {
 
     fn close_epoch(&mut self, epoch_id: WriterEpochId) -> Result<(), WalStoreError> {
         self.ensure_writer_lock()?;
+        // An epoch must not close over a commit record that was never synced.
+        self.sync_deferred_commits()?;
         let previous_ledger = self.writer_epoch_ledger();
         let epoch = self
             .active_epoch
diff --git a/crates/warp-core/src/trusted_runtime_host.rs b/crates/warp-core/src/trusted_runtime_host.rs
index fb94814e..7991ff70 100644
--- a/crates/warp-core/src/trusted_runtime_host.rs
+++ b/crates/warp-core/src/trusted_runtime_host.rs
@@ -174,6 +174,12 @@ pub enum TrustedRuntimeWalError {
     /// An earlier append must be reconciled before another can reach storage.
     #[error("trusted runtime WAL writer prefix requires reconciliation")]
     UncertainWriterPrefix,
+    /// Storage failed while commit syncs were deferred. Records written before
+    /// the failure are not known to be durable, so this adapter refuses
+    /// further work until the host reopens the WAL and recovers from what
+    /// storage actually holds.
+    #[error("trusted runtime WAL storage failed; reopen the WAL to recover")]
+    StorageFailed,
     /// WAL transaction construction failed before storage append.
     #[error("trusted runtime WAL transaction build error: {0}")]
     Build(#[from] WalBuildError),
@@ -1419,6 +1425,20 @@ impl TrustedRuntimeHost {
         self.runtime_wal = Some(runtime_wal);
     }
 
+    /// Refuses one runtime WAL append before it reaches storage, after letting
+    /// `appends_let_through` appends succeed.
+    #[cfg(any(test, feature = "host_test"))]
+    pub fn refuse_runtime_wal_append_before_storage_for_test(
+        &mut self,
+        appends_let_through: u32,
+    ) -> Result<(), TrustedRuntimeHostError> {
+        self.runtime_wal
+            .as_mut()
+            .ok_or(TrustedRuntimeHostError::RuntimeWalUnavailable)?
+            .refuse_append_before_storage_for_test(appends_let_through);
+        Ok(())
+    }
+
     /// Replaces the filesystem WAL fault plan for targeted host tests.
     #[cfg(any(test, feature = "host_test"))]
     pub fn inject_runtime_wal_filesystem_fault_for_test(
@@ -2520,6 +2540,10 @@ pub struct TrustedRuntimeWal {
     evidence_catalog_posture: EvidenceCatalogPosture,
     #[cfg(any(test, feature = "host_test"))]
     fail_next_evidence_catalog_update: bool,
+    /// Refuses one append before it reaches storage, after letting this many
+    /// appends through.
+    #[cfg(any(test, feature = "host_test"))]
+    refuse_append_before_storage_after: Option<u32>,
     #[cfg(any(test, feature = "host_test"))]
     recover_read_only_call_count: std::cell::Cell<usize>,
     writer_epoch: WriterEpochId,
@@ -2540,6 +2564,24 @@ pub struct TrustedRuntimeWal {
     causal_history_frontier_digest: Hash,
     causal_anchor_claim_projection: CausalAnchorClaimProjection,
     durable_submission_acceptances: BTreeMap<Hash, Hash>,
+    /// Acceptances whose commit records are appended and not yet synced. They
+    /// stop a second append of the same submission and are never treated as
+    /// durable.
+    unsynced_submission_acceptances: BTreeMap<Hash, Hash>,
+    /// Set when storage fails while commit syncs are deferred; see
+    /// [`TrustedRuntimeWalError::StorageFailed`].
+    storage_failed: bool,
+    /// Digest of the last commit that a successful sync is known to cover.
+    last_synced_commit_digest: Hash,
+}
+
+/// Whether an appended commit record is synced before the append returns.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+enum CommitSync {
+    /// Sync the commit record now. The transaction is durable on return.
+    Now,
+    /// Leave the commit record unsynced for a later shared sync.
+    Deferred,
 }
 
 impl TrustedRuntimeWal {
@@ -2609,11 +2651,18 @@ impl TrustedRuntimeWal {
             causal_history_frontier_digest: recovered_cursor.causal_history_frontier_digest,
             causal_anchor_claim_projection: recovered_cursor.causal_anchor_claim_projection,
             durable_submission_acceptances,
+            unsynced_submission_acceptances: BTreeMap::new(),
+            storage_failed: false,
+            // Opening the store synced the log, so the recovered prefix is
+            // durable.
+            last_synced_commit_digest: recovered_cursor.previous_committed_transaction_digest,
             evidence_catalog: Some(evidence_catalog),
             evidence_catalog_posture: EvidenceCatalogPosture::Fresh,
             #[cfg(any(test, feature = "host_test"))]
             fail_next_evidence_catalog_update: false,
             #[cfg(any(test, feature = "host_test"))]
+            refuse_append_before_storage_after: None,
+            #[cfg(any(test, feature = "host_test"))]
             recover_read_only_call_count: std::cell::Cell::new(0),
         })
     }
@@ -2625,6 +2674,9 @@ impl TrustedRuntimeWal {
     }
 
     /// Returns committed WAL markers recorded by the adapter.
+    ///
+    /// This is what the log holds, not what is durable: it includes a commit
+    /// record whose sync is still deferred or has failed.
     #[must_use]
     pub fn commits(&self) -> Vec<WalTransactionCommit> {
         self.store.read_commits()
@@ -2652,6 +2704,8 @@ impl TrustedRuntimeWal {
             #[cfg(any(test, feature = "host_test"))]
             fail_next_evidence_catalog_update: self.fail_next_evidence_catalog_update,
             #[cfg(any(test, feature = "host_test"))]
+            refuse_append_before_storage_after: self.refuse_append_before_storage_after,
+            #[cfg(any(test, feature = "host_test"))]
             recover_read_only_call_count: std::cell::Cell::new(
                 self.recover_read_only_call_count.get(),
             ),
@@ -2675,6 +2729,9 @@ impl TrustedRuntimeWal {
             causal_history_frontier_digest: self.causal_history_frontier_digest,
             causal_anchor_claim_projection: self.causal_anchor_claim_projection.clone(),
             durable_submission_acceptances: self.durable_submission_acceptances.clone(),
+            unsynced_submission_acceptances: self.unsynced_submission_acceptances.clone(),
+            storage_failed: self.storage_failed,
+            last_synced_commit_digest: self.last_synced_commit_digest,
         })
     }
 
@@ -2708,6 +2765,11 @@ impl TrustedRuntimeWal {
     /// Recovers submission and receipt indexes from committed WAL transactions
     /// without scheduler callbacks.
     pub fn recover_read_only(&self) -> Result<TrustedRuntimeWalRecovery, TrustedRuntimeWalError> {
+        // After a storage failure the log may hold records that are readable
+        // and not durable. This adapter does not report them as recovered.
+        if self.storage_failed {
+            return Err(TrustedRuntimeWalError::StorageFailed);
+        }
         #[cfg(any(test, feature = "host_test"))]
         self.recover_read_only_call_count
             .set(self.recover_read_only_call_count.get() + 1);
@@ -2792,6 +2854,9 @@ impl TrustedRuntimeWal {
     pub fn recover_evidence_catalog_read_only(
         &self,
     ) -> Result<crate::evidence::CausalSegmentCatalog, TrustedRuntimeWalError> {
+        if self.storage_failed {
+            return Err(TrustedRuntimeWalError::StorageFailed);
+        }
         let report = self.store.recover_read_only()?;
         crate::evidence::CausalSegmentCatalog::from_recovery_scan(&report)
             .map_err(TrustedRuntimeWalError::EvidenceCatalog)
@@ -2875,6 +2940,80 @@ impl TrustedRuntimeWal {
         self.store.is_filesystem()
     }
 
+    /// Returns `true` when an acceptance for this submission is in the log,
+    /// whether or not its commit record has been synced. Use this only to
+    /// avoid appending the same acceptance twice.
+    fn has_recorded_submission_acceptance(
+        &self,
+        submission_id: Hash,
+        canonical_envelope_digest: Hash,
+    ) -> bool {
+        self.has_submission_acceptance(submission_id, canonical_envelope_digest)
+            || self.unsynced_submission_acceptances.get(&submission_id)
+                == Some(&canonical_envelope_digest)
+    }
+
+    /// Makes every transaction appended with [`CommitSync::Deferred`] durable
+    /// with one storage sync, and only then counts its acceptances as durable.
+    fn sync_deferred_commits(&mut self) -> Result<(), TrustedRuntimeWalError> {
+        if self.storage_failed {
+            return Err(TrustedRuntimeWalError::StorageFailed);
+        }
+        if let TrustedRuntimeWalStore::Filesystem(store) = &mut self.store {
+            if let Err(error) = store.sync_deferred_commits() {
+                self.fail_storage();
+                return Err(error.into());
+            }
+        }
+        self.promote_synced_submission_acceptances();
+        Ok(())
+    }
+
+    /// Records that storage failed while commit syncs were deferred.
+    ///
+    /// No unsynced acceptance can become durable after this, and the adapter
+    /// does not reconcile itself from the log: a record that is readable is
+    /// not thereby durable.
+    fn fail_storage(&mut self) {
+        self.storage_failed = true;
+        self.unsynced_submission_acceptances.clear();
+        // The live catalog may describe records that are not durable. It is
+        // dropped, and the posture names the last commit known to be synced.
+        self.evidence_catalog = None;
+        self.evidence_catalog_posture = EvidenceCatalogPosture::NeedsRebuild {
+            reason: *blake3::hash(b"storage_failed").as_bytes(),
+            last_good_commit: self.last_synced_commit_digest,
+        };
+    }
+
+    fn storage_failed(&self) -> bool {
+        self.storage_failed
+    }
+
+    /// Rebuilds the writer cursor after an append that was refused before it
+    /// reached storage. Unsynced acceptances stay unsynced.
+    fn restore_cursor_after_refused_append(&mut self) -> Result<(), TrustedRuntimeWalError> {
+        self.refresh_cursor_from_store_for_writer()
+    }
+
+    /// Records that a successful sync covers every record appended so far.
+    fn promote_synced_submission_acceptances(&mut self) {
+        let synced = std::mem::take(&mut self.unsynced_submission_acceptances);
+        self.durable_submission_acceptances.extend(synced);
+        self.last_synced_commit_digest = self.previous_committed_transaction_digest;
+    }
+
+    /// Returns how many storage syncs have made commit records durable, when
+    /// the adapter is backed by the filesystem store.
+    #[cfg(any(test, feature = "host_test"))]
+    #[must_use]
+    pub fn filesystem_commit_sync_count_for_test(&self) -> Option<u64> {
+        match &self.store {
+            TrustedRuntimeWalStore::Filesystem(store) => Some(store.commit_sync_count_for_test()),
+            TrustedRuntimeWalStore::InMemory(_) => None,
+        }
+    }
+
     #[cfg(any(test, feature = "host_test"))]
     fn replace_filesystem_fault_plan_for_test(
         &mut self,
@@ -2886,6 +3025,12 @@ impl TrustedRuntimeWal {
     }
 
     fn refresh_cursor_from_store_for_writer(&mut self) -> Result<(), TrustedRuntimeWalError> {
+        // After a failed storage sync a readable record is not a durable one,
+        // so the live adapter must not reconcile itself from the log. Only
+        // reopening the WAL recovers.
+        if self.storage_failed {
+            return Err(TrustedRuntimeWalError::StorageFailed);
+        }
         self.operation_context_index = None;
         let report = self.store.recover_for_writer()?;
         let contexts = observed_context::Contexts::from_recovery(&report)?;
@@ -2923,6 +3068,11 @@ impl TrustedRuntimeWal {
         self.causal_history_frontier_digest = cursor.causal_history_frontier_digest;
         self.causal_anchor_claim_projection = cursor.causal_anchor_claim_projection;
         self.durable_submission_acceptances = durable_submission_acceptances;
+        // The log is read as written, not as synced. An acceptance whose sync
+        // is still owed stays unsynced however readable its record is.
+        for submission_id in self.unsynced_submission_acceptances.keys() {
+            self.durable_submission_acceptances.remove(submission_id);
+        }
         self.operation_context_index = Some((self.previous_committed_transaction_digest, contexts));
         Ok(())
     }
@@ -3017,6 +3167,7 @@ impl TrustedRuntimeWal {
         &mut self,
         envelope: &IngressEnvelope,
         handle: IntentSubmissionHandle,
+        sync: CommitSync,
     ) -> Result<WalTransactionCommit, TrustedRuntimeWalError> {
         let record = SubmissionAcceptanceRecord {
             submission_id: handle.submission_id,
@@ -3040,10 +3191,18 @@ impl TrustedRuntimeWal {
                 after_digest: next_submission_frontier,
             }],
         )?;
-        let commit = self.append_transaction(transaction)?;
+        let commit = self.append_transaction_with(transaction, sync)?;
         self.submission_frontier_digest = next_submission_frontier;
-        self.durable_submission_acceptances
-            .insert(record.submission_id, record.canonical_envelope_digest);
+        match sync {
+            CommitSync::Now => {
+                self.durable_submission_acceptances
+                    .insert(record.submission_id, record.canonical_envelope_digest);
+            }
+            CommitSync::Deferred => {
+                self.unsynced_submission_acceptances
+                    .insert(record.submission_id, record.canonical_envelope_digest);
+            }
+        }
         Ok(commit)
     }
 
@@ -3348,6 +3507,17 @@ impl TrustedRuntimeWal {
         &mut self,
         transaction: WalCommittedTransaction,
     ) -> Result<WalTransactionCommit, TrustedRuntimeWalError> {
+        self.append_transaction_with(transaction, CommitSync::Now)
+    }
+
+    fn append_transaction_with(
+        &mut self,
+        transaction: WalCommittedTransaction,
+        sync: CommitSync,
+    ) -> Result<WalTransactionCommit, TrustedRuntimeWalError> {
+        if self.storage_failed {
+            return Err(TrustedRuntimeWalError::StorageFailed);
+        }
         let next_causal_history_frontier = logical_causal_history_frontier_digest(
             self.causal_history_frontier_digest,
             transaction.commit.transaction_kind,
@@ -3375,13 +3545,31 @@ impl TrustedRuntimeWal {
         if prefix != self.previous_committed_transaction_digest {
             return Err(TrustedRuntimeWalError::UncertainWriterPrefix);
         }
+        #[cfg(any(test, feature = "host_test"))]
+        match self.refuse_append_before_storage_after {
+            Some(0) => {
+                self.refuse_append_before_storage_after = None;
+                return Err(TrustedRuntimeWalError::UncertainWriterPrefix);
+            }
+            Some(remaining) => self.refuse_append_before_storage_after = Some(remaining - 1),
+            None => {}
+        }
         contexts.apply_frames(&frames)?;
-        self.store.append_transaction(transaction)?;
+        match sync {
+            CommitSync::Now => self.store.append_transaction(transaction)?,
+            CommitSync::Deferred => self
+                .store
+                .append_transaction_deferring_commit_sync(transaction)?,
+        }
         self.operation_context_index = Some((commit.commit_digest, contexts));
         self.next_lsn = next_lsn;
         self.previous_frame_digest = last_frame_digest;
         self.previous_committed_transaction_digest = commit.commit_digest;
         self.causal_history_frontier_digest = next_causal_history_frontier;
+        if matches!(sync, CommitSync::Now) {
+            // A synced commit covers every earlier record in the segment.
+            self.promote_synced_submission_acceptances();
+        }
         self.try_update_evidence_catalog_after_commit(&commit, &frames, last_good_commit);
         Ok(commit)
     }
@@ -3480,6 +3668,10 @@ impl TrustedRuntimeWalStore {
                 };
                 #[cfg(not(any(test, feature = "host_test")))]
                 let store = FilesystemWalStore::open(root, segment_id)?;
+                let mut store = store;
+                // Make what an earlier writer left readable durable before
+                // any of it is read as recovered history.
+                store.sync_existing_segments()?;
                 Ok(Self::Filesystem(store))
             }
         }
@@ -3566,6 +3758,18 @@ impl TrustedRuntimeWalStore {
         }
     }
 
+    /// Appends without syncing the commit record. The in-memory store has no
+    /// sync, so it appends as usual.
+    fn append_transaction_deferring_commit_sync(
+        &mut self,
+        transaction: WalCommittedTransaction,
+    ) -> Result<(), WalStoreError> {
+        match self {
+            Self::InMemory(store) => store.append_transaction(transaction),
+            Self::Filesystem(store) => store.append_transaction_deferring_commit_sync(transaction),
+        }
+    }
+
     #[cfg(any(test, feature = "host_test"))]
     fn append_uncommitted_frame(
         &mut self,
@@ -6016,6 +6220,12 @@ impl TrustedRuntimeWal {
     pub fn fail_next_evidence_catalog_update_for_test(&mut self) {
         self.fail_next_evidence_catalog_update = true;
     }
+
+    /// Refuses one append before it reaches storage, after letting
+    /// `appends_let_through` appends succeed.
+    pub fn refuse_append_before_storage_for_test(&mut self, appends_let_through: u32) {
+        self.refuse_append_before_storage_after = Some(appends_let_through);
+    }
 }
 
 /// App-facing handle for a trusted local runtime host.
@@ -6049,6 +6259,11 @@ impl TrustedRuntimeApp<'_> {
             .runtime_wal
             .as_ref()
             .ok_or(TrustedRuntimeHostError::RuntimeWalUnavailable)?;
+        // After a storage failure the cached frontier may name records that
+        // are not durable. It is not a basis.
+        if runtime_wal.storage_failed() {
+            return Err(TrustedRuntimeWalError::StorageFailed.into());
+        }
         Ok(runtime_wal.current_causal_anchor_basis())
     }
 
@@ -6199,13 +6414,104 @@ impl TrustedRuntimeApp<'_> {
         &mut self,
         envelope: IngressEnvelope,
     ) -> Result<IntentSubmissionHandle, TrustedRuntimeHostError> {
-        self.submit_intent_with_runtime_wal_ack_inner(envelope, AppIntentAdmission::Ordinary)
+        self.submit_intent_with_runtime_wal_ack_inner(
+            envelope,
+            AppIntentAdmission::Ordinary,
+            CommitSync::Now,
+        )
+    }
+
+    /// Submits several canonical intents and returns only after one storage
+    /// sync has made all of their acceptance transactions durable.
+    ///
+    /// Each envelope is accepted exactly as
+    /// [`Self::submit_intent_with_runtime_wal_ack`] accepts it, in order, and
+    /// writes the same acceptance transaction. Only the sync is shared. No
+    /// acceptance of the batch counts as durable, and none can be scheduled,
+    /// before that sync succeeds.
+    ///
+    /// # Errors
+    ///
+    /// No handle is returned on error, so no submission of the batch is
+    /// acknowledged.
+    ///
+    /// - If the runtime refuses an envelope before anything is written for it,
+    ///   the envelopes before it are still synced. They stay accepted and
+    ///   durable, as they would after a process loss between commit and
+    ///   acknowledgement; submitting the same envelope again returns its
+    ///   existing submission. The error of the refused envelope is returned.
+    ///   The same holds if the WAL refuses an append before it reaches
+    ///   storage.
+    /// - If storage fails during an append or the sync, the in-memory intake
+    ///   of every submission this call accepted is rolled back, the storage
+    ///   error is returned, and the WAL refuses further work with
+    ///   [`TrustedRuntimeWalError::StorageFailed`] until the host reopens
+    ///   it. Storage may still hold some of the records; reopening recovers
+    ///   whatever it holds.
+    pub fn submit_intents_with_runtime_wal_ack(
+        &mut self,
+        envelopes: Vec<IngressEnvelope>,
+    ) -> Result<Vec<IntentSubmissionHandle>, TrustedRuntimeHostError> {
+        if self.host.runtime_wal.is_none() {
+            return Err(TrustedRuntimeHostError::RuntimeWalUnavailable);
+        }
+        let mut handles = Vec::with_capacity(envelopes.len());
+        let mut accepted = Vec::new();
+        let mut failure = None;
+        for envelope in envelopes {
+            let marker = self.host.runtime.intent_submission_rollback_marker();
+            match self.submit_intent_with_runtime_wal_ack_inner(
+                envelope,
+                AppIntentAdmission::Ordinary,
+                CommitSync::Deferred,
+            ) {
+                Ok(handle) => {
+                    if !handle.duplicate {
+                        accepted.push((marker, handle));
+                    }
+                    handles.push(handle);
+                }
+                Err(error) => {
+                    failure = Some(error);
+                    break;
+                }
+            }
+        }
+        let Some(runtime_wal) = self.host.runtime_wal.as_mut() else {
+            return Err(TrustedRuntimeHostError::RuntimeWalUnavailable);
+        };
+        let storage_error = if runtime_wal.storage_failed() {
+            // An append failed. Its error is the one to report.
+            failure.take()
+        } else {
+            // Sync whatever was appended: those records are in the log
+            // whether or not the batch is acknowledged.
+            runtime_wal.sync_deferred_commits().err().map(Into::into)
+        };
+        if let Some(error) = storage_error {
+            // Nothing this call appended is known to be durable. Do not leave
+            // it visible as accepted intake.
+            for (marker, handle) in accepted.into_iter().rev() {
+                self.host
+                    .pending_echo_operation_actions
+                    .remove(&handle.submission_id);
+                self.host
+                    .runtime
+                    .rollback_intent_submission(marker, &handle);
+            }
+            return Err(error);
+        }
+        if let Some(error) = failure {
+            return Err(error);
+        }
+        Ok(handles)
     }
 
     fn submit_intent_with_runtime_wal_ack_inner(
         &mut self,
         envelope: IngressEnvelope,
         admission: AppIntentAdmission,
+        sync: CommitSync,
     ) -> Result<IntentSubmissionHandle, TrustedRuntimeHostError> {
         if self.host.runtime_wal.is_none() {
             return Err(TrustedRuntimeHostError::RuntimeWalUnavailable);
@@ -6230,7 +6536,8 @@ impl TrustedRuntimeApp<'_> {
             return Err(TrustedRuntimeHostError::RuntimeWalUnavailable);
         };
         if handle.duplicate
-            && runtime_wal.has_submission_acceptance(handle.submission_id, envelope.ingress_id())
+            && runtime_wal
+                .has_recorded_submission_acceptance(handle.submission_id, envelope.ingress_id())
         {
             self.host.track_pending_echo_operation_action_v1(
                 handle.submission_id,
@@ -6238,7 +6545,23 @@ impl TrustedRuntimeApp<'_> {
             );
             return Ok(handle);
         }
-        if let Err(error) = runtime_wal.record_submission_acceptance(&envelope, handle) {
+        if let Err(error) = runtime_wal.record_submission_acceptance(&envelope, handle, sync) {
+            if matches!(sync, CommitSync::Deferred) {
+                // Earlier records of this batch are appended and not synced.
+                // A storage error may have left part of this one in the log
+                // as well, and nothing readable there is durable: fail
+                // closed. An append refused before it reached storage wrote
+                // nothing, so only the writer cursor needs rebuilding.
+                if matches!(error, TrustedRuntimeWalError::Store(_))
+                    || runtime_wal.restore_cursor_after_refused_append().is_err()
+                {
+                    runtime_wal.fail_storage();
+                }
+                self.host
+                    .runtime
+                    .rollback_intent_submission(submission_rollback, &handle);
+                return Err(error.into());
+            }
             if runtime_wal.recover_filesystem_submission_acceptance_after_error(
                 handle.submission_id,
                 envelope.ingress_id(),
@@ -6281,7 +6604,11 @@ impl TrustedRuntimeApp<'_> {
             return Err(TrustedRuntimeHostError::RuntimeWalUnavailable);
         }
         let envelope = self.host.resolve_contract_inverse_envelope(&request)?;
-        self.submit_intent_with_runtime_wal_ack_inner(envelope, AppIntentAdmission::ContractInverse)
+        self.submit_intent_with_runtime_wal_ack_inner(
+            envelope,
+            AppIntentAdmission::ContractInverse,
+            CommitSync::Now,
+        )
     }
 
     /// Recovers the typed causal derivation for one admitted inverse receipt.
diff --git a/crates/warp-core/tests/causal_wal_hardening_tests.rs b/crates/warp-core/tests/causal_wal_hardening_tests.rs
index a2b4c5ee..4f757756 100644
--- a/crates/warp-core/tests/causal_wal_hardening_tests.rs
+++ b/crates/warp-core/tests/causal_wal_hardening_tests.rs
@@ -1075,6 +1075,298 @@ fn filesystem_takeover_after_process_loss_derives_final_commit_from_the_segment(
     must_ok(fs::remove_dir_all(root));
 }
 
+fn commit_sync_evidence_count(store: &FilesystemWalStore) -> usize {
+    store
+        .sync_evidence()
+        .iter()
+        .filter(|entry| entry.boundary == FilesystemSyncBoundary::CommitFileSynced)
+        .count()
+}
+
+#[test]
+fn filesystem_deferred_commits_are_not_evidenced_as_synced_until_one_sync() {
+    let root = temp_wal_root("deferred-commit-sync");
+    let mut store = must_ok(FilesystemWalStore::open(&root, WalSegmentId::from_raw(1)));
+    must_ok(store.acquire_writer_epoch(writer_epoch_request()));
+
+    must_ok(
+        store.append_transaction_deferring_commit_sync(submission_transaction(
+            "deferred-first",
+            Lsn::from_raw(0),
+        )),
+    );
+    must_ok(
+        store.append_transaction_deferring_commit_sync(submission_transaction(
+            "deferred-second",
+            Lsn::from_raw(2),
+        )),
+    );
+    assert_eq!(
+        commit_sync_evidence_count(&store),
+        0,
+        "an unsynced commit record must not be evidenced as synced"
+    );
+
+    must_ok(store.sync_deferred_commits());
+    assert_eq!(commit_sync_evidence_count(&store), 2);
+    must_ok(store.sync_deferred_commits());
+    assert_eq!(commit_sync_evidence_count(&store), 2);
+    drop(store);
+    must_ok(fs::remove_dir_all(root));
+}
+
+#[cfg(feature = "host_test")]
+#[test]
+fn filesystem_failed_deferred_sync_is_never_evidenced_and_closes_the_store() {
+    use warp_core::causal_wal::{FilesystemWalFaultPlan, FilesystemWalFaultTarget};
+
+    let root = temp_wal_root("deferred-commit-sync-failed");
+    let mut store = must_ok(FilesystemWalStore::open(&root, WalSegmentId::from_raw(1)));
+    must_ok(store.acquire_writer_epoch(writer_epoch_request()));
+    must_ok(
+        store.append_transaction_deferring_commit_sync(submission_transaction(
+            "failed-first",
+            Lsn::from_raw(0),
+        )),
+    );
+    store.replace_fault_plan_for_test(FilesystemWalFaultPlan::fail_next(
+        FilesystemWalFaultTarget::DeferredCommitSync,
+    ));
+
+    assert!(store.sync_deferred_commits().is_err());
+    // A later sync cannot vouch for what an earlier sync failed to write.
+    assert!(store.sync_deferred_commits().is_err());
+    assert!(store
+        .append_transaction(submission_transaction("failed-second", Lsn::from_raw(2)))
+        .is_err());
+    assert!(store.close_epoch(epoch_id()).is_err());
+    // The lower-level port a coordinator writes through is refused as well.
+    let later = submission_transaction("failed-third", Lsn::from_raw(2));
+    let later_commit = later.commit.clone();
+    assert!(later
+        .frames
+        .into_iter()
+        .all(|frame| store.append_frame(epoch_id(), frame).is_err()));
+    assert!(store.flush_commit(epoch_id(), later_commit).is_err());
+    assert!(store.truncate_tail_after(Lsn::from_raw(0)).is_err());
+    assert!(store
+        .publish_manifest(
+            epoch_id(),
+            WalManifest {
+                manifest_digest: digest("hardening:manifest:after-failed-sync"),
+                last_committed_lsn: None,
+                last_commit_digest: None,
+                sealed_segment_count: 0,
+            },
+        )
+        .is_err());
+    assert_eq!(commit_sync_evidence_count(&store), 0);
+    drop(store);
+    must_ok(fs::remove_dir_all(root));
+}
+
+#[cfg(feature = "host_test")]
+#[test]
+fn filesystem_writer_syncs_an_existing_log_before_trusting_it() {
+    use warp_core::causal_wal::{FilesystemWalFaultPlan, FilesystemWalFaultTarget};
+    let root = temp_wal_root("recovered-segment-sync");
+    let mut store = must_ok(FilesystemWalStore::open(&root, WalSegmentId::from_raw(1)));
+    must_ok(store.acquire_writer_epoch(writer_epoch_request()));
+    // The record is written and its sync never happens, as when a writer
+    // stops between the two.
+    must_ok(
+        store.append_transaction_deferring_commit_sync(submission_transaction(
+            "left-unsynced",
+            Lsn::from_raw(0),
+        )),
+    );
+    drop(store);
+
+    let mut refused = must_ok(FilesystemWalStore::open_with_fault_plan_for_test(
+        &root,
+        WalSegmentId::from_raw(1),
+        FilesystemWalFaultPlan::fail_next(FilesystemWalFaultTarget::RecoveredSegmentSync),
+    ));
+    let error = must_err(
+        refused.acquire_fresh_writer_epoch(Lsn::from_raw(2)),
+        "a writer that cannot sync the existing log must not take it over",
+    );
+    assert!(matches!(
+        error,
+        WalStoreError::Io(message) if message.contains("recovered_segment_sync")
+    ));
+    assert_eq!(refused.recovered_segment_sync_count_for_test(), 0);
+    drop(refused);
+
+    let mut successor = must_ok(FilesystemWalStore::open(&root, WalSegmentId::from_raw(1)));
+    must_ok(successor.acquire_fresh_writer_epoch(Lsn::from_raw(2)));
+    assert_eq!(
+        successor.recovered_segment_sync_count_for_test(),
+        1,
+        "the unsynced record is made durable before the log is read as history"
+    );
+    drop(successor);
+    must_ok(fs::remove_dir_all(root));
+}
+
+#[cfg(feature = "host_test")]
+#[test]
+fn filesystem_failed_synced_commit_never_evidences_pending_deferred_commits() {
+    use warp_core::causal_wal::{FilesystemWalFaultPlan, FilesystemWalFaultTarget};
+    let root = temp_wal_root("deferred-commit-covered-by-failed-sync");
+    let mut store = must_ok(FilesystemWalStore::open(&root, WalSegmentId::from_raw(1)));
+    must_ok(store.acquire_writer_epoch(writer_epoch_request()));
+    must_ok(
+        store.append_transaction_deferring_commit_sync(submission_transaction(
+            "deferred-then-failed",
+            Lsn::from_raw(0),
+        )),
+    );
+    store.replace_fault_plan_for_test(FilesystemWalFaultPlan::fail_next(
+        FilesystemWalFaultTarget::FlushCommit,
+    ));
+
+    // The synced commit would have covered the deferred record as well.
+    assert!(store
+        .append_transaction(submission_transaction("synced-fails", Lsn::from_raw(2)))
+        .is_err());
+    assert!(store.sync_deferred_commits().is_err());
+    assert_eq!(commit_sync_evidence_count(&store), 0);
+    drop(store);
+    must_ok(fs::remove_dir_all(root));
+}
+
+#[test]
+fn filesystem_truncation_is_refused_while_a_commit_sync_is_owed() {
+    let mut fixture = WalHardeningFixture::new("deferred-commit-truncation");
+    must_ok(
+        fixture
+            .store
+            .append_transaction_deferring_commit_sync(submission_transaction(
+                "before-truncation",
+                Lsn::from_raw(0),
+            )),
+    );
+
+    assert!(
+        fixture.store.truncate_tail_after(Lsn::from_raw(0)).is_err(),
+        "truncation must not remove a record whose sync is still owed"
+    );
+    assert_eq!(commit_sync_evidence_count(&fixture.store), 0);
+
+    must_ok(fixture.store.sync_deferred_commits());
+    assert_eq!(commit_sync_evidence_count(&fixture.store), 1);
+    must_ok(fixture.store.truncate_tail_after(Lsn::from_raw(1)));
+}
+
+#[test]
+fn filesystem_segment_seal_syncs_deferred_commits_first() {
+    let mut fixture = WalHardeningFixture::new("deferred-commit-seal");
+    must_ok(
+        fixture
+            .store
+            .append_transaction_deferring_commit_sync(submission_transaction(
+                "before-seal",
+                Lsn::from_raw(0),
+            )),
+    );
+    assert_eq!(commit_sync_evidence_count(&fixture.store), 0);
+
+    must_ok(
+        fixture
+            .store
+            .seal_segment(epoch_id(), WalSegmentId::from_raw(1)),
+    );
+
+    assert_eq!(
+        commit_sync_evidence_count(&fixture.store),
+        1,
+        "a seal must not exist before the commits it covers are durable"
+    );
+}
+
+#[test]
+fn filesystem_manifest_publication_syncs_deferred_commits_first() {
+    let mut fixture = WalHardeningFixture::new("deferred-commit-manifest");
+    let transaction = submission_transaction("before-manifest", Lsn::from_raw(0));
+    let commit = transaction.commit.clone();
+    must_ok(
+        fixture
+            .store
+            .append_transaction_deferring_commit_sync(transaction),
+    );
+    assert_eq!(commit_sync_evidence_count(&fixture.store), 0);
+
+    must_ok(fixture.store.publish_manifest(
+        epoch_id(),
+        WalManifest {
+            manifest_digest: digest("hardening:manifest:deferred"),
+            last_committed_lsn: Some(commit.last_lsn),
+            last_commit_digest: Some(commit.commit_digest),
+            sealed_segment_count: 0,
+        },
+    ));
+
+    let boundaries = fixture
+        .store
+        .sync_evidence()
+        .iter()
+        .map(|entry| entry.boundary)
+        .collect::<Vec<_>>();
+    let commit_synced = boundaries
+        .iter()
+        .position(|boundary| *boundary == FilesystemSyncBoundary::CommitFileSynced);
+    let manifest_synced = boundaries
+        .iter()
+        .position(|boundary| *boundary == FilesystemSyncBoundary::ManifestTempFileSynced);
+    assert!(
+        matches!((commit_synced, manifest_synced), (Some(commit), Some(manifest)) if commit < manifest),
+        "a manifest must not become durable before the commit it names: {boundaries:?}"
+    );
+}
+
+#[test]
+fn filesystem_synced_commit_and_epoch_close_cover_deferred_commits() {
+    let root = temp_wal_root("deferred-commit-covered");
+    let mut store = must_ok(FilesystemWalStore::open(&root, WalSegmentId::from_raw(1)));
+    must_ok(store.acquire_writer_epoch(writer_epoch_request()));
+
+    must_ok(
+        store.append_transaction_deferring_commit_sync(submission_transaction(
+            "covered-first",
+            Lsn::from_raw(0),
+        )),
+    );
+    must_ok(store.append_transaction(submission_transaction("covered-second", Lsn::from_raw(2))));
+    assert_eq!(
+        commit_sync_evidence_count(&store),
+        2,
+        "a synced commit covers every earlier record in the segment"
+    );
+
+    must_ok(
+        store.append_transaction_deferring_commit_sync(submission_transaction(
+            "covered-third",
+            Lsn::from_raw(4),
+        )),
+    );
+    assert_eq!(commit_sync_evidence_count(&store), 2);
+    must_ok(store.close_epoch(epoch_id()));
+    assert_eq!(
+        commit_sync_evidence_count(&store),
+        3,
+        "an epoch must not close over an unsynced commit record"
+    );
+    drop(store);
+
+    let report = must_ok(recover_filesystem_store(
+        &root,
+        RecoveryAccessMode::ReadOnly,
+    ));
+    assert_eq!(report.transactions.len(), 3);
+    must_ok(fs::remove_dir_all(root));
+}
+
 #[test]
 fn filesystem_takeover_refuses_unreconciled_tail_before_reusing_its_lsn() {
     let mut fixture = WalHardeningFixture::new("takeover-uncommitted");
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 6a180d50..5c73c1a6 100644
--- a/crates/warp-core/tests/trusted_runtime_host_loop_tests.rs
+++ b/crates/warp-core/tests/trusted_runtime_host_loop_tests.rs
@@ -1909,6 +1909,425 @@ fn filesystem_runtime_wal_failure_submission_restores_exact_runtime() {
     }
 }
 
+fn distinct_intents(worldline_id: WorldlineId, count: u8) -> Vec<IngressEnvelope> {
+    (0..count)
+        .map(|index| {
+            IngressEnvelope::local_intent(
+                IngressTarget::DefaultWriter { worldline_id },
+                make_intent_kind("echo.intent/eint-v1"),
+                vec![index; 4],
+            )
+        })
+        .collect()
+}
+
+#[test]
+fn batched_submission_makes_every_acceptance_durable_with_one_sync() {
+    let wal_root = temp_runtime_wal_dir("batched-submission-one-sync");
+    let (runtime, worldline_id) = runtime();
+    let mut host =
+        TrustedRuntimeHost::new(runtime, empty_engine()).expect("trusted host should initialize");
+    host.enable_runtime_wal(TrustedRuntimeWalConfig::filesystem(&wal_root))
+        .expect("host should configure filesystem runtime WAL adapter");
+    let syncs = |host: &TrustedRuntimeHost| {
+        host.runtime_wal()
+            .expect("runtime WAL should stay configured")
+            .filesystem_commit_sync_count_for_test()
+            .expect("filesystem store")
+    };
+
+    let single_before = syncs(&host);
+    let mut envelopes = distinct_intents(worldline_id, 8);
+    let batch = envelopes.split_off(3);
+    for envelope in envelopes {
+        host.app()
+            .submit_intent_with_runtime_wal_ack(envelope)
+            .expect("single submission should be acknowledged");
+    }
+    assert_eq!(
+        syncs(&host) - single_before,
+        3,
+        "each single submission waits for its own sync"
+    );
+
+    let batch_before = syncs(&host);
+    let handles = host
+        .app()
+        .submit_intents_with_runtime_wal_ack(batch)
+        .expect("batched submission should be acknowledged");
+    assert_eq!(handles.len(), 5);
+    assert!(handles.iter().all(|handle| !handle.duplicate));
+    assert_eq!(
+        syncs(&host) - batch_before,
+        1,
+        "a batch waits for one sync, not one per submission"
+    );
+
+    let submission_ids = handles
+        .iter()
+        .map(|handle| handle.submission_id)
+        .collect::<Vec<_>>();
+    drop(host);
+    let (runtime, _) = crate::runtime();
+    let mut reopened =
+        TrustedRuntimeHost::new(runtime, empty_engine()).expect("trusted host should initialize");
+    reopened
+        .enable_runtime_wal(TrustedRuntimeWalConfig::filesystem(&wal_root))
+        .expect("a fresh host should recover the batch");
+    let recovery = reopened
+        .runtime_wal()
+        .expect("runtime WAL should stay configured")
+        .recover_read_only()
+        .expect("batched acceptances should recover");
+    assert_eq!(recovery.certificate.committed_transactions_replayed, 8);
+    for submission_id in submission_ids {
+        assert_eq!(
+            recovery
+                .submissions
+                .get(&submission_id)
+                .expect("every batched submission should recover")
+                .posture,
+            RecoveredSubmissionPosture::AcceptedPending
+        );
+    }
+    drop(reopened);
+    fs::remove_dir_all(&wal_root).expect("WAL fixture should be removable");
+}
+
+#[test]
+fn batched_submission_append_failure_fails_closed_until_the_wal_is_reopened() {
+    let wal_root = temp_runtime_wal_dir("batched-submission-failure");
+    let (runtime, worldline_id) = runtime();
+    let mut host =
+        TrustedRuntimeHost::new(runtime, empty_engine()).expect("trusted host should initialize");
+    host.enable_runtime_wal(TrustedRuntimeWalConfig::filesystem(&wal_root))
+        .expect("host should configure filesystem runtime WAL adapter");
+    let runtime_before = format!("{:?}", host.runtime());
+    // Let the first acceptance commit, unsynced, and fail the second.
+    host.inject_runtime_wal_filesystem_fault_for_test(
+        FilesystemWalFaultPlan::fail_next(FilesystemWalFaultTarget::FlushCommit).after(1),
+    )
+    .expect("host should inject a filesystem failure");
+
+    let error = host
+        .app()
+        .submit_intents_with_runtime_wal_ack(distinct_intents(worldline_id, 4))
+        .expect_err("a batch whose append fails must not be acknowledged");
+    assert!(matches!(
+        error,
+        TrustedRuntimeHostError::Wal(TrustedRuntimeWalError::Store(WalStoreError::Io(
+            message
+        ))) if message.contains("injected filesystem WAL flush_commit failure")
+    ));
+    assert_eq!(
+        format!("{:?}", host.runtime()),
+        runtime_before,
+        "the acceptance appended before the failure must not stay visible"
+    );
+    // The first acceptance is readable in the log and was never synced. The
+    // host must not treat it as an existing durable submission.
+    let refused = host
+        .app()
+        .submit_intent_with_runtime_wal_ack(distinct_intents(worldline_id, 1).remove(0))
+        .expect_err("the WAL must refuse work after a failed append in a batch");
+    assert!(matches!(
+        refused,
+        TrustedRuntimeHostError::Wal(TrustedRuntimeWalError::StorageFailed)
+    ));
+    assert_eq!(format!("{:?}", host.runtime()), runtime_before);
+
+    drop(host);
+    let (runtime, _) = crate::runtime();
+    let mut reopened =
+        TrustedRuntimeHost::new(runtime, empty_engine()).expect("trusted host should initialize");
+    reopened
+        .enable_runtime_wal(TrustedRuntimeWalConfig::filesystem(&wal_root))
+        .expect("a fresh host should recover what storage holds");
+    let single = reopened
+        .app()
+        .submit_intent_with_runtime_wal_ack(distinct_intents(worldline_id, 1).remove(0))
+        .expect("a reopened WAL accepts work again");
+    assert!(
+        single.duplicate,
+        "reopening syncs the log before trusting it, so the acceptance it holds is durable"
+    );
+    drop(reopened);
+    fs::remove_dir_all(&wal_root).expect("WAL fixture should be removable");
+}
+
+#[test]
+fn batched_submission_refused_before_storage_keeps_earlier_intents_and_the_wal() {
+    let wal_root = temp_runtime_wal_dir("batched-submission-refused");
+    let (runtime, worldline_id) = runtime();
+    let mut host =
+        TrustedRuntimeHost::new(runtime, empty_engine()).expect("trusted host should initialize");
+    host.enable_runtime_wal(TrustedRuntimeWalConfig::filesystem(&wal_root))
+        .expect("host should configure filesystem runtime WAL adapter");
+    let syncs_before = host
+        .runtime_wal()
+        .expect("runtime WAL should stay configured")
+        .filesystem_commit_sync_count_for_test()
+        .expect("filesystem store");
+    // The first acceptance commits, unsynced. The second is refused before
+    // anything is written for it.
+    host.refuse_runtime_wal_append_before_storage_for_test(1)
+        .expect("host should arm the refusal");
+
+    let error = host
+        .app()
+        .submit_intents_with_runtime_wal_ack(distinct_intents(worldline_id, 3))
+        .expect_err("a batch with a refused intent returns no handles");
+    assert!(matches!(
+        error,
+        TrustedRuntimeHostError::Wal(TrustedRuntimeWalError::UncertainWriterPrefix)
+    ));
+    assert_eq!(
+        host.runtime().witnessed_submission_count(),
+        1,
+        "the intent before the refusal stays accepted"
+    );
+    let runtime_wal = host
+        .runtime_wal()
+        .expect("runtime WAL should stay configured");
+    assert_eq!(
+        runtime_wal
+            .filesystem_commit_sync_count_for_test()
+            .expect("filesystem store")
+            - syncs_before,
+        1,
+        "the intent before the refusal is synced"
+    );
+    assert_eq!(runtime_wal.submission_acceptance_count(), 1);
+
+    // Nothing failed in storage, so the WAL keeps working.
+    let mut intents = distinct_intents(worldline_id, 3);
+    let first = host
+        .app()
+        .submit_intent_with_runtime_wal_ack(intents.remove(0))
+        .expect("the accepted intent is returned as its existing submission");
+    assert!(first.duplicate);
+    let rest = host
+        .app()
+        .submit_intents_with_runtime_wal_ack(intents)
+        .expect("the refused intent and its successor are accepted on retry");
+    assert_eq!(rest.len(), 2);
+    assert!(rest.iter().all(|handle| !handle.duplicate));
+    assert_eq!(host.runtime().witnessed_submission_count(), 3);
+    drop(host);
+    fs::remove_dir_all(&wal_root).expect("WAL fixture should be removable");
+}
+
+#[test]
+fn batched_submission_refused_before_storage_keeps_an_in_memory_wal_usable() {
+    let (runtime, worldline_id) = runtime();
+    let mut host =
+        TrustedRuntimeHost::new(runtime, empty_engine()).expect("trusted host should initialize");
+    host.enable_runtime_wal(TrustedRuntimeWalConfig::in_memory())
+        .expect("host should configure in-memory runtime WAL adapter");
+    host.refuse_runtime_wal_append_before_storage_for_test(1)
+        .expect("host should arm the refusal");
+
+    host.app()
+        .submit_intents_with_runtime_wal_ack(distinct_intents(worldline_id, 3))
+        .expect_err("a batch with a refused intent returns no handles");
+    assert_eq!(host.runtime().witnessed_submission_count(), 1);
+
+    let mut intents = distinct_intents(worldline_id, 3);
+    let first = host
+        .app()
+        .submit_intent_with_runtime_wal_ack(intents.remove(0))
+        .expect("the accepted intent is returned as its existing submission");
+    assert!(first.duplicate);
+    let rest = host
+        .app()
+        .submit_intents_with_runtime_wal_ack(intents)
+        .expect("the writer cursor is rebuilt, so later appends are accepted");
+    assert_eq!(rest.len(), 2);
+    assert!(rest.iter().all(|handle| !handle.duplicate));
+}
+
+#[test]
+fn a_host_does_not_open_a_log_it_cannot_sync() {
+    let wal_root = temp_runtime_wal_dir("open-sync-failure");
+    let (runtime, worldline_id) = runtime();
+    let mut host =
+        TrustedRuntimeHost::new(runtime, empty_engine()).expect("trusted host should initialize");
+    host.enable_runtime_wal(TrustedRuntimeWalConfig::filesystem(&wal_root))
+        .expect("host should configure filesystem runtime WAL adapter");
+    host.app()
+        .submit_intent_with_runtime_wal_ack(distinct_intents(worldline_id, 1).remove(0))
+        .expect("the first host accepts a submission");
+    drop(host);
+
+    let (runtime, _) = crate::runtime();
+    let mut reopened =
+        TrustedRuntimeHost::new(runtime, empty_engine()).expect("trusted host should initialize");
+    let error = reopened
+        .enable_runtime_wal(
+            TrustedRuntimeWalConfig::filesystem_with_fault_plan_for_test(
+                &wal_root,
+                FilesystemWalFaultPlan::fail_next(FilesystemWalFaultTarget::RecoveredSegmentSync),
+            ),
+        )
+        .expect_err("a log that cannot be synced must not be read as recovered history");
+    assert!(matches!(
+        error,
+        TrustedRuntimeHostError::Wal(TrustedRuntimeWalError::Store(WalStoreError::Io(
+            message
+        ))) if message.contains("recovered_segment_sync")
+    ));
+    assert!(reopened.runtime_wal().is_none());
+    assert_eq!(reopened.runtime().witnessed_submission_count(), 0);
+    drop(reopened);
+    fs::remove_dir_all(&wal_root).expect("WAL fixture should be removable");
+}
+
+#[test]
+fn batched_submission_sync_failure_rolls_back_intake_and_fails_closed() {
+    let wal_root = temp_runtime_wal_dir("batched-submission-sync-failure");
+    let (runtime, worldline_id) = runtime();
+    let mut host =
+        TrustedRuntimeHost::new(runtime, empty_engine()).expect("trusted host should initialize");
+    host.enable_runtime_wal(TrustedRuntimeWalConfig::filesystem(&wal_root))
+        .expect("host should configure filesystem runtime WAL adapter");
+    let runtime_before = format!("{:?}", host.runtime());
+    host.inject_runtime_wal_filesystem_fault_for_test(FilesystemWalFaultPlan::fail_next(
+        FilesystemWalFaultTarget::DeferredCommitSync,
+    ))
+    .expect("host should inject a sync failure");
+
+    let error = host
+        .app()
+        .submit_intents_with_runtime_wal_ack(distinct_intents(worldline_id, 4))
+        .expect_err("a batch whose sync fails must not be acknowledged");
+    assert!(matches!(
+        error,
+        TrustedRuntimeHostError::Wal(TrustedRuntimeWalError::Store(WalStoreError::Io(
+            message
+        ))) if message.contains("injected filesystem WAL deferred_commit_sync failure")
+    ));
+    assert_eq!(
+        format!("{:?}", host.runtime()),
+        runtime_before,
+        "a failed sync must not leave the batch visible as accepted intake"
+    );
+    let refused = host
+        .app()
+        .submit_intent_with_runtime_wal_ack(distinct_intents(worldline_id, 1).remove(0))
+        .expect_err("the WAL must refuse work after a failed sync");
+    assert!(matches!(
+        refused,
+        TrustedRuntimeHostError::Wal(TrustedRuntimeWalError::StorageFailed)
+    ));
+    assert_eq!(
+        format!("{:?}", host.runtime()),
+        runtime_before,
+        "a refused submission must not stay visible either"
+    );
+    assert!(
+        matches!(host.tick_once(), Ok(steps) if steps.is_empty()),
+        "nothing from the failed batch may be scheduled"
+    );
+    let failed_wal = host
+        .runtime_wal()
+        .expect("runtime WAL should stay configured");
+    assert!(
+        matches!(
+            failed_wal.recover_read_only(),
+            Err(TrustedRuntimeWalError::StorageFailed)
+        ),
+        "the running host must not report readable records as recovered"
+    );
+    assert!(matches!(
+        failed_wal.recover_evidence_catalog_read_only(),
+        Err(TrustedRuntimeWalError::StorageFailed)
+    ));
+    assert!(
+        failed_wal.evidence_catalog().is_none(),
+        "the live catalog must not describe records that are not durable"
+    );
+    assert!(matches!(
+        failed_wal.evidence_catalog_posture(),
+        warp_core::EvidenceCatalogPosture::NeedsRebuild { .. }
+    ));
+    assert!(
+        matches!(
+            host.app().current_causal_anchor_basis(),
+            Err(TrustedRuntimeHostError::Wal(
+                TrustedRuntimeWalError::StorageFailed
+            ))
+        ),
+        "a frontier over records that are not durable is not a basis"
+    );
+
+    // The records were written; only their sync failed. Reopening recovers
+    // what storage holds, as it would after a process loss.
+    drop(host);
+    let (runtime, _) = crate::runtime();
+    let mut reopened =
+        TrustedRuntimeHost::new(runtime, empty_engine()).expect("trusted host should initialize");
+    reopened
+        .enable_runtime_wal(TrustedRuntimeWalConfig::filesystem(&wal_root))
+        .expect("a fresh host should recover what storage holds");
+    let recovery = reopened
+        .runtime_wal()
+        .expect("runtime WAL should stay configured")
+        .recover_read_only()
+        .expect("the written prefix should recover");
+    assert_eq!(recovery.certificate.committed_transactions_replayed, 4);
+    assert!(recovery.receipts.receipt_by_submission.is_empty());
+    drop(reopened);
+    fs::remove_dir_all(&wal_root).expect("WAL fixture should be removable");
+}
+
+#[test]
+fn batched_submission_handles_duplicates_and_an_empty_batch() {
+    let wal_root = temp_runtime_wal_dir("batched-submission-duplicates");
+    let (runtime, worldline_id) = runtime();
+    let mut host =
+        TrustedRuntimeHost::new(runtime, empty_engine()).expect("trusted host should initialize");
+    host.enable_runtime_wal(TrustedRuntimeWalConfig::filesystem(&wal_root))
+        .expect("host should configure filesystem runtime WAL adapter");
+    let syncs = |host: &TrustedRuntimeHost| {
+        host.runtime_wal()
+            .expect("runtime WAL should stay configured")
+            .filesystem_commit_sync_count_for_test()
+            .expect("filesystem store")
+    };
+
+    let before = syncs(&host);
+    let empty = host
+        .app()
+        .submit_intents_with_runtime_wal_ack(Vec::new())
+        .expect("an empty batch is acknowledged without work");
+    assert!(empty.is_empty());
+    assert_eq!(syncs(&host), before, "an empty batch forces no sync");
+
+    let mut envelopes = distinct_intents(worldline_id, 2);
+    envelopes.insert(1, envelopes[0].clone());
+    let handles = host
+        .app()
+        .submit_intents_with_runtime_wal_ack(envelopes)
+        .expect("a batch that repeats an envelope should be acknowledged");
+    assert_eq!(handles.len(), 3);
+    assert!(!handles[0].duplicate);
+    assert!(handles[1].duplicate);
+    assert_eq!(handles[1].submission_id, handles[0].submission_id);
+    assert!(!handles[2].duplicate);
+    assert_eq!(syncs(&host) - before, 1);
+    let runtime_wal = host
+        .runtime_wal()
+        .expect("runtime WAL should stay configured");
+    assert_eq!(
+        runtime_wal.submission_acceptance_count(),
+        2,
+        "a repeated envelope must not append a second acceptance"
+    );
+    assert_eq!(runtime_wal.commits().len(), 2);
+    drop(host);
+    fs::remove_dir_all(&wal_root).expect("WAL fixture should be removable");
+}
+
 #[test]
 fn filesystem_runtime_wal_failure_manifest_publish_reports_store_error() {
     let wal_root = temp_runtime_wal_dir("failure-manifest-publish");
diff --git a/docs/topics/WAL.md b/docs/topics/WAL.md
index d019498d..3a5d4fcf 100644
--- a/docs/topics/WAL.md
+++ b/docs/topics/WAL.md
@@ -25,6 +25,30 @@ after the runtime WAL has committed the submission-acceptance transaction. That
 acceptance fact recovers as an `AcceptedPending` submission if the scheduler has
 not yet produced a decided outcome.
 
+`submit_intents_with_runtime_wal_ack(...)` accepts several intents in order
+and returns after one storage sync covers all of their acceptance
+transactions. Each intent still writes its own acceptance transaction, so
+recovery reads a batch exactly as it reads single submissions. Until that sync
+succeeds, no acceptance of the batch counts as durable and none can be
+scheduled. A batch is not atomic across a crash: a prefix can survive, none of
+it acknowledged, and each survivor recovers as `AcceptedPending`.
+
+If the runtime or the WAL refuses an intent of a batch before anything is
+written for it, the intents before it are still synced and stay accepted; the
+call returns the refusal and no handles.
+
+If storage fails during an append or the shared sync, Echo rolls back the
+in-memory intake of every submission that call accepted, returns the storage
+error, and refuses further WAL work with `StorageFailed` until the host
+reopens the WAL. A record that is readable after a storage failure is not
+treated as durable by the running host.
+
+A host that opens an existing log, and a writer that takes it over, sync its
+segment files before reading them as recovered history. An earlier writer may
+have stopped, or failed a sync, after writing records; those records are
+readable and not yet durable. If that sync fails, the host does not open the
+log and the writer does not take it over.
+
 Second, failed WAL acceptance is not allowed to leave half-visible runtime
 state. If no runtime WAL is configured, the ACK path returns an explicit
 unavailable error before mutating intake state. If WAL commit fails, Echo rolls