Retain observation-bound executable requests across host loss

CHANGELOG.md

@@ -7,6 +7,13 @@
 
 ### Added
 
+- Bounded executable-operation host sessions retain immutable observations and
+  logical-request bindings in the native WAL. Echo evaluates supplied node/atom
+  preconditions inside operation preparation, includes their reads in scheduler
+  footprints, and retains typed refusal or commitment outcomes across reopening.
+  The generic `run-edict-operation --serve` driver supports repeated operations,
+  historical outcome lookup, and derived observation-change discovery.
+
 - A disabled-by-default experimental Keep ReferenceStore adapter implements the complete-object CAS port with explicit limits, private bindings, sanitized backend errors and atomic output promotion. It preserves existing CAS defaults and claims no restart durability.
 
 - An isolated experimental Echo–Keep identity bridge checks both independent hash laws and exact byte length from one bounded stream. It leaves the default CAS dependency graph unchanged.
@@ -15,6 +22,8 @@
 
 ### Fixed
 
+- Observation slots preflight their canonical byte bound and charge execution reads before copying Atom payloads.
+
 - Private Edict runtime decoding resolves named integer types through their declared width, matching source-function provider acceptance.
 
 - Filesystem writer takeover preserves unused LSNs after empty epochs and refuses unreconciled tails before changing the epoch ledger.
@@ -1835,6 +1844,35 @@ Applied, Rejected, Obstructed}` with receipt evidence and typed contract
 
 ### Fixed
 
+- Filesystem writer leases explicitly unlock when their owner leaves scope,
+  so a descriptor briefly inherited by a concurrent child process cannot keep
+  the departed writer's lease alive and spuriously refuse its successor.
+
+- The observed-session driver refuses an obviously insufficient compiled
+  budget before durable setup and explains that observation reads require an
+  authored, verified allowance rather than silently increasing the grant.
+
+- Filesystem writer takeover refuses an unreconciled WAL tail before reusing
+  an empty epoch's log position, preventing duplicate physical LSNs.
+
+- Observation change discovery preserves intervening writes after values are
+  restored, including after reopening. The session driver resolves keys and
+  commit evidence together from native provenance instead of repeatedly
+  reconstructing WAL history.
+
+- Retained observations bind the anchor's actual occupancy instead of claiming
+  that every observed anchor was absent.
+
+- Observation reads now populate the footprint partition mask as well as the
+  exact read sets, keeping retained footprint evidence consistent.
+
+- Observed operations reject observations from another writer head even when
+  both heads belong to the same worldline.
+
+- Reopening a filesystem WAL through an empty writer epoch no longer consumes an
+  unwritten log position. A second reopen followed by append previously left an
+  LSN gap and made subsequent recovery fail.
+
 - Generic executable-operation lowering and independent verification now
   resolve source-local obstruction constructor aliases through the exact
   digest-locked lawpack import before encoding or comparing the package.

README.md

@@ -278,7 +278,11 @@ mutation. A standalone application-owned Edict create-if-absent operation now
 crosses this route as an exact compiler-produced package and independently
 accepted verification report through `cargo xtask run-edict-operation`. That
 witness proves its package-declared typed obstruction and one Action in one
-Tick. Its duplicate report exposes equal before/after application-state roots
+Tick. Its `--serve` mode continues one worldline through multiple operations,
+retains observation-bound requests, and reopens native history; the bounded
+trusted-host scope is described in
+[application contract hosting](docs/architecture/application-contract-hosting.md#observation-bound-native-host-sessions).
+Its duplicate report exposes equal before/after application-state roots
 and typed target-value digests; the roots commit the reachable graph state, not
 WAL, history, Receipt, or commit metadata. It does not claim external
 multi-Action Tick composition, a product-ready application runner, a Jedit

crates/warp-core/src/causal_wal.rs

@@ -486,6 +486,8 @@ pub enum WalRecordKind {
     ExternalActionClaimRecorded,
     /// Echo admitted one schema-bound external-action settlement.
     ExternalActionSettlementRecorded,
+    /// Runtime retained an immutable operation observation or request binding.
+    ExecutableOperationContextRetained,
 }
 
 impl WalRecordKind {
@@ -525,6 +527,7 @@ impl WalRecordKind {
             Self::ExternalActionRequestRecorded => "ExternalActionRequestRecorded",
             Self::ExternalActionClaimRecorded => "ExternalActionClaimRecorded",
             Self::ExternalActionSettlementRecorded => "ExternalActionSettlementRecorded",
+            Self::ExecutableOperationContextRetained => "ExecutableOperationContextRetained",
         }
     }
 
@@ -553,7 +556,8 @@ impl WalRecordKind {
             Self::SchedulerFaultQuarantined | Self::TrustedRuntimeControlRecorded => {
                 WalAppendAuthority::RuntimeControl
             }
-            Self::ExecutableOperationPackageInstalled => WalAppendAuthority::RuntimeControl,
+            Self::ExecutableOperationPackageInstalled
+            | Self::ExecutableOperationContextRetained => WalAppendAuthority::RuntimeControl,
             Self::ExecutableOperationExecutionRecorded
             | Self::ExecutableOperationStateDeltaRecorded => WalAppendAuthority::ExecutionKernel,
             Self::CausalAnchorFactRecorded | Self::CausalAnchorAdmissionReceiptRecorded => {
@@ -614,6 +618,7 @@ impl WalRecordKind {
             Self::ExternalActionRequestRecorded => 29,
             Self::ExternalActionClaimRecorded => 30,
             Self::ExternalActionSettlementRecorded => 31,
+            Self::ExecutableOperationContextRetained => 32,
         }
     }
 
@@ -650,6 +655,7 @@ impl WalRecordKind {
             29 => Ok(Self::ExternalActionRequestRecorded),
             30 => Ok(Self::ExternalActionClaimRecorded),
             31 => Ok(Self::ExternalActionSettlementRecorded),
+            32 => Ok(Self::ExecutableOperationContextRetained),
             _ => Err(WalDecodeError::UnknownEnumCode {
                 enum_name: "WalRecordKind",
                 code,

crates/warp-core/src/echo_operation.rs

@@ -29,6 +29,9 @@ use echo_edict_canonical::{
 use sha2::{Digest as _, Sha256};
 use thiserror::Error;
 
+mod observed;
+pub use observed::EchoOperationObservationV1;
+
 use crate::{
     attachment::{AtomPayload, AttachmentKey, AttachmentValue},
     clock::{GlobalTick, WorldlineTick},
@@ -2779,6 +2782,7 @@ enum AnchoredNodeOperationModeV1 {
 /// Canonical basis-bearing invocation emitted by a generated client/helper.
 #[derive(Clone, Debug, PartialEq, Eq)]
 pub struct EchoOperationInvocationV1 {
+    observation: Option<EchoOperationObservationV1>,
     package_id: EchoOperationPackageIdV1,
     operation_coordinate: String,
     evaluation_basis: EchoOperationEvaluationBasisV1,
@@ -2816,6 +2820,7 @@ impl EchoOperationInvocationV1 {
             },
             replacement_bytes,
             application_input_bytes: None,
+            observation: None,
         }
     }
 
@@ -2841,6 +2846,7 @@ impl EchoOperationInvocationV1 {
             kind: EchoOperationInvocationKindV1::AnchoredNodeAttachmentCreateIfAbsent,
             replacement_bytes,
             application_input_bytes: None,
+            observation: None,
         }
     }
 
@@ -2868,6 +2874,7 @@ impl EchoOperationInvocationV1 {
             kind: EchoOperationInvocationKindV1::AnchoredNodeAttachmentCreateIfAbsent,
             replacement_bytes,
             application_input_bytes: Some(canonical_application_input_bytes),
+            observation: None,
         }
     }
 
@@ -2893,6 +2900,19 @@ impl EchoOperationInvocationV1 {
                 "invocation delegated step budget must be nonzero",
             ));
         }
+        if let Some(observation) = &self.observation {
+            let mut inner = self.clone();
+            inner.observation = None;
+            return encode_canonical_cbor_v1(&map_value([
+                ("schema", text_value(observed::OBSERVED_SCHEMA)),
+                (
+                    "invocation",
+                    CanonicalValueV1::Bytes(inner.to_canonical_bytes()?),
+                ),
+                ("observation", observation.to_value()),
+            ]))
+            .map_err(canonical_error);
+        }
         let common = |schema| {
             [
                 (
@@ -2971,7 +2991,7 @@ impl EchoOperationInvocationV1 {
         encode_canonical_cbor_v1(&value).map_err(canonical_error)
     }
 
-    fn from_canonical_bytes(bytes: &[u8]) -> Result<Self, EchoOperationArtifactErrorV1> {
+    pub(crate) fn from_canonical_bytes(bytes: &[u8]) -> Result<Self, EchoOperationArtifactErrorV1> {
         let value = decode_canonical_cbor_v1(bytes).map_err(canonical_error)?;
         let schema = match &value {
             CanonicalValueV1::Map(entries) => entries
@@ -2986,6 +3006,27 @@ impl EchoOperationInvocationV1 {
                 .ok_or_else(|| invalid_structure("invocation schema must be text"))?,
             _ => return Err(invalid_structure("artifact root must be a map")),
         };
+        if schema == observed::OBSERVED_SCHEMA {
+            let mut fields = exact_text_map(value, &["schema", "invocation", "observation"])?;
+            let inner_bytes = take_bytes(&mut fields, "invocation")?;
+            let inner = decode_canonical_cbor_v1(&inner_bytes).map_err(canonical_error)?;
+            if let CanonicalValueV1::Map(entries) = &inner {
+                if entries.iter().any(|(key, value)| {
+                    key == &text_value("schema") && value == &text_value(observed::OBSERVED_SCHEMA)
+                }) {
+                    return Err(invalid_structure("nested observed invocation is forbidden"));
+                }
+            }
+            let mut invocation = Self::from_canonical_bytes(&inner_bytes)?;
+            invocation.observation = Some(EchoOperationObservationV1::from_value(take_field(
+                &mut fields,
+                "observation",
+            )?)?);
+            if invocation.to_canonical_bytes()? != bytes {
+                return Err(invalid_structure("observed invocation is not canonical"));
+            }
+            return Ok(invocation);
+        }
         let (expected_fields, create_if_absent, projected) = match schema {
             INVOCATION_SCHEMA => (
                 &[
@@ -3063,6 +3104,7 @@ impl EchoOperationInvocationV1 {
             }
         };
         let invocation = Self {
+            observation: None,
             package_id: EchoOperationPackageIdV1(take_hash(&mut fields, "package_id")?),
             operation_coordinate: take_text(&mut fields, "operation_coordinate")?,
             evaluation_basis: EchoOperationEvaluationBasisV1::from_value(take_field(
@@ -3709,6 +3751,10 @@ pub enum EchoOperationObstructionKindV1 {
     /// The compiler-owned result projection could not produce one bounded
     /// canonical result from the exact admitted application input.
     ResultProjectionInvalid,
+    /// A supplied observation changed despite a current submission basis.
+    ObservationChanged,
+    /// An observed resource is no longer available under the bounded profile.
+    ObservationUnavailable,
 }
 
 /// Retained runtime policy evidence needed to reproduce one obstruction.
@@ -4381,6 +4427,8 @@ fn obstruction_kind_from_code(
         11 => Ok(EchoOperationObstructionKindV1::ReplacementTooLarge),
         12 => Ok(EchoOperationObstructionKindV1::EvaluationAuthorityMismatch),
         13 => Ok(EchoOperationObstructionKindV1::ResultProjectionInvalid),
+        14 => Ok(EchoOperationObstructionKindV1::ObservationChanged),
+        15 => Ok(EchoOperationObstructionKindV1::ObservationUnavailable),
         _ => Err(invalid_structure(
             "unknown executable-operation obstruction kind",
         )),
@@ -4615,6 +4663,17 @@ pub(crate) fn prepare_operation_v1(
     let node = admitted.invocation.node;
     let mut actual_footprint = Footprint::default();
     let mut budget_meter = EchoOperationBudgetMeterV1::new(admitted.invocation.delegated_budget);
+    if let Some(observation) = &admitted.invocation.observation {
+        if let Err(kind) = observation.validate_at_execution(
+            state,
+            current_basis,
+            &mut actual_footprint,
+            &mut budget_meter,
+        ) {
+            return obstruction(kind);
+        }
+    }
+    let observed_footprint = actual_footprint.clone();
     let descent_stack =
         match operation_descent_stack_with_portal_reads(state, node.warp_id, |portal| {
             if !budget_meter.charge(1, 32, 0) {
@@ -4629,14 +4688,15 @@ pub(crate) fn prepare_operation_v1(
     let Some(store) = state.store(&node.warp_id) else {
         return obstruction(EchoOperationObstructionKindV1::NodeMissing);
     };
-    let declared_footprint = match mode {
+    let mut declared_footprint = match mode {
         AnchoredNodeOperationModeV1::CompareAndSet { .. } => {
             anchored_node_compare_and_set_footprint(node, &descent_stack)
         }
         AnchoredNodeOperationModeV1::CreateIfAbsent => {
             anchored_node_create_if_absent_footprint(node, &descent_stack)
         }
     };
+    declared_footprint.union_assign(&observed_footprint);
     if !budget_meter.charge(1, 32, 0) {
         return obstruction(EchoOperationObstructionKindV1::BudgetExceeded);
     }
@@ -4739,6 +4799,14 @@ pub(crate) fn prepare_operation_v1(
     }
     let mut in_slots = vec![SlotId::Node(node), SlotId::Attachment(slot)];
     in_slots.extend(descent_stack.iter().copied().map(SlotId::Attachment));
+    in_slots.extend(observed_footprint.n_read.iter().copied().map(SlotId::Node));
+    in_slots.extend(
+        observed_footprint
+            .a_read
+            .iter()
+            .copied()
+            .map(SlotId::Attachment),
+    );
     let patch = WarpTickPatchV1::new(
         policy_id,
         installed.installed_operation_id.as_hash(),
@@ -6767,6 +6835,8 @@ fn obstruction_kind_code(kind: EchoOperationObstructionKindV1) -> u8 {
         EchoOperationObstructionKindV1::ReplacementTooLarge => 11,
         EchoOperationObstructionKindV1::EvaluationAuthorityMismatch => 12,
         EchoOperationObstructionKindV1::ResultProjectionInvalid => 13,
+        EchoOperationObstructionKindV1::ObservationChanged => 14,
+        EchoOperationObstructionKindV1::ObservationUnavailable => 15,
     }
 }
 
@@ -7311,7 +7381,7 @@ mod tests {
         }
     }
 
-    fn projected_create_fixture(
+    pub(super) fn projected_create_fixture(
         max_output_bytes: u64,
     ) -> (
         InstalledEchoOperationV1,