@@ -7,6 +7,8 @@ pub use error::CandidateError;
 mod private_input;
 #[cfg(target_os = "linux")]
 #[path = "../../runtime-core/src/provider/storage_root_witness.rs"]
+// The host also uses request encoding/reply decoding from this shared source.
+#[allow(dead_code)]
 mod storage_root_witness;
 // This shared module also supplies host-only authority/effect APIs. Keep them
 // compiled and checked by runtime-core without requiring guest-side callers.
 
@@ -1526,9 +1526,33 @@ created only during the original enrollment with `XATTR_CREATE` on a retained
 nofollow root descriptor, synchronized, and independently reread with exact
 descriptor/path/root checks. Retained startup only reads; missing/changed witness
 refuses without repair. Whole-root/xattr copying remains outside the claimed
-guarantee. This native helper/transport is not implemented or qualified here.
+guarantee. The metadata-only checkpoint does not implement or qualify that helper
+boundary; the inactive witnessed continuation below adds its source.
 
 These runtime/SQL gates and replacement safety are not yet qualified by this source implementation. Stock
 PostgreSQL parity remains a separate NC05 gate requiring authored ownership/user
 or a specifically qualified capability policy. This slice guesses no UID/GID,
 changes no volume permissions, and adds no capability to make a stock image work.
+# Witnessed enrollment continuation (inactive)
+
+The separate persistent owner version2 stores its expected root xattr in the private
+pending intent before the single original volume create. It captures the original
+volume and root UID/GID/device/inode, seeds with `XATTR_CREATE`, independently verifies
+the root, and only then publishes enrollment under the original private lock. A missing
+or changed witness refuses retained reads without repair. Version1 metadata records
+are not upgraded or treated as continuity proof. Ambiguous create/seed retains the
+intent; publication after rename can be uncertain even when an enrolled file is visible.
+
+Native receipt4 may describe a separate pinned guest tool installation, including the
+original directory and helper identities. This control path shares the existing relay
+artifact/upload boundary and uses private descriptor input under the original operation
+deadline. Dependency-free legacy graphs continue to install no guest artifact. Tool
+receipt parsing grants neither installation nor storage authority, and this continuation
+does not enable ordinary persistent startup. The Linux private-filesystem controls,
+installed helper/guest transport, and real Bun/SQLite retention still require qualification.
+The shared artifact reader now requires a current-UID regular file with one link,
+no group/world write permission, and matching held-FD/path identity and bytes. Legacy
+relay graphs with dependencies use that stricter reader and its fixed
+`guest_tool_artifact` refusal; dependency-free graphs do not read an artifact.
+Whole-root copies retaining the exact xattr remain outside this witness guarantee.
+Stock PostgreSQL remains a separate ownership/capability acceptance gate.
@@ -136,6 +136,19 @@ pub(super) fn exec_until(
     script: &str,
     arguments: &[&str],
     deadline: Instant,
+) -> Result<String, CandidateError> {
+    exec_input_until(path, script, arguments, None, deadline)
+}
+
+/// Private input follows the same absolute deadline as connect, dispatch and reply.
+/// A timed-out exec is uncertain; no late successful reply restores authority.
+#[cfg(feature = "native-config-plan")]
+pub(super) fn exec_input_until(
+    path: &Path,
+    script: &str,
+    arguments: &[&str],
+    input: Option<&str>,
+    deadline: Instant,
 ) -> Result<String, CandidateError> {
     let mut stream = connect_until(path, deadline)?;
     let remaining = deadline
@@ -147,7 +160,7 @@ pub(super) fn exec_until(
     }
     let mut command = vec!["/bin/sh", "-c", script, "hack-local"];
     command.extend_from_slice(arguments);
-    let body = json!({"method":"vm_exec","command":command,"env":[["PATH","/opt/hack-engine:/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin"]],"workdir":"/","timeout_ms":milliseconds,"interactive":false,"tty":false,"background":false,"stdin_data":null});
+    let body = json!({"method":"vm_exec","command":command,"env":[["PATH","/opt/hack-engine:/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin"]],"workdir":"/","timeout_ms":milliseconds,"interactive":false,"tty":false,"background":false,"stdin_data":input});
     let bytes = encode(&body)?;
     let timeout = deadline
         .checked_duration_since(Instant::now())
 
@@ -69,6 +69,8 @@ pub struct Receipt {
     #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
     pub(super) data_mounts: BTreeMap<String, Vec<crate::project::native::StorageMount>>,
     #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub(super) data_tool: Option<persistent_data::tool::Reference>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
     failure: Option<Failure>,
     #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
     pub(super) terminal: BTreeMap<String, super::super::shutdown::Terminal>,
@@ -91,6 +93,8 @@ struct ReceiptWire {
     data: Option<BTreeMap<String, persistent_data::engine::Reference>>,
     #[serde(default, deserialize_with = "present_map")]
     data_mounts: Option<BTreeMap<String, Vec<crate::project::native::StorageMount>>>,
+    #[serde(default, deserialize_with = "present_map")]
+    data_tool: Option<persistent_data::tool::Reference>,
     #[serde(default)]
     failure: Option<Failure>,
     #[serde(default)]
@@ -104,7 +108,8 @@ fn present_map<'de, D: serde::Deserializer<'de>, T: Deserialize<'de>>(
 impl TryFrom<ReceiptWire> for Receipt {
     type Error = &'static str;
     fn try_from(wire: ReceiptWire) -> Result<Self, Self::Error> {
-        if (wire.version == 2 && (wire.data.is_some() || wire.data_mounts.is_some()))
+        if (wire.version == 2
+            && (wire.data.is_some() || wire.data_mounts.is_some() || wire.data_tool.is_some()))
             || (wire.version == 4 && (wire.data.is_none() || wire.data_mounts.is_none()))
         {
             return Err("Native receipt storage fields do not match its wire version.");
@@ -120,6 +125,7 @@ impl TryFrom<ReceiptWire> for Receipt {
             resources: wire.resources,
             data: wire.data.unwrap_or_default(),
             data_mounts: wire.data_mounts.unwrap_or_default(),
+            data_tool: wire.data_tool,
             failure: wire.failure,
             terminal: wire.terminal,
         })
@@ -134,6 +140,7 @@ impl Receipt {
             || self.readiness != expected.readiness
             || self.data != expected.data
             || self.data_mounts != expected.data_mounts
+            || self.data_tool != expected.data_tool
             || self.resources.keys().ne(expected.resources.keys())
             || self.resources.iter().any(|(key, resource)| {
                 let prior = &expected.resources[key];
@@ -150,6 +157,9 @@ impl Receipt {
         }
         Ok(())
     }
+    pub(in crate::provider::graph::native) fn persistent(&self) -> bool {
+        self.version == 4
+    }
     pub fn phase(&self) -> &Phase {
         &self.phase
     }
@@ -202,6 +212,7 @@ impl Receipt {
             resources: config.resources.clone(),
             data: config.data.clone(),
             data_mounts: config.data_mounts.clone(),
+            data_tool: None,
             failure: None,
             terminal: BTreeMap::new(),
         };
@@ -348,7 +359,10 @@ impl Receipt {
 
     fn validate_data(&self) -> Result<(), CandidateError> {
         if self.version == 2 {
-            return if self.data.is_empty() && self.data_mounts.is_empty() {
+            return if self.data.is_empty()
+                && self.data_mounts.is_empty()
+                && self.data_tool.is_none()
+            {
                 Ok(())
             } else {
                 Err(refused())
@@ -357,6 +371,18 @@ impl Receipt {
         if self.data.is_empty() || self.data_mounts.is_empty() {
             return Err(refused());
         }
+        if let Some(tool) = &self.data_tool {
+            tool.validate()?;
+            if (self.phase == Phase::ReadyObserved
+                || self
+                    .resources
+                    .values()
+                    .any(|resource| resource.kind == Kind::Container && resource.id.is_some()))
+                && tool.helper.is_none()
+            {
+                return Err(refused());
+            }
+        }
         let mut used = BTreeSet::new();
         for (service, mounts) in &self.data_mounts {
             if !self.readiness.contains_key(service) || mounts.is_empty() {
 
@@ -261,3 +261,8 @@ mod tests;
 
 pub(super) mod engine;
 pub mod enrollment;
+pub mod witnessed;
+
+// Inactive APIs are deliberately not selected by ordinary startup until qualification.
+#[allow(dead_code)]
+pub(super) mod tool;
@@ -9,6 +9,10 @@ use reqwest::Method;
 use serde_json::{Value, json};
 use std::{collections::BTreeMap, io::Read, sync::atomic::AtomicBool, time::Instant};
 
+// This installed transport is intentionally not called by the gated runtime yet.
+#[allow(dead_code)]
+mod witnessed;
+
 #[derive(Clone, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(tag = "status", rename_all = "snake_case", deny_unknown_fields)]
 enum State {
 
@@ -0,0 +1,125 @@
+//! Installed tool + original Engine lease transport. No ordinary runtime caller is
+//! enabled. A helper observation never grants durable promotion by itself.
+use super::*;
+use crate::provider::graph::native::persistent_data::{
+    tool::Installed,
+    witnessed::{ExpectedWitness, RootIdentity},
+};
+use crate::provider::storage_root_witness::{Observation as Reply, Request};
+use enrollment::Transport as _;
+
+pub(in crate::provider::graph::native) struct WitnessAdapter<'a, 'guest> {
+    base: Adapter<'a, 'guest>,
+    tool: &'a Installed,
+}
+impl<'a, 'guest> WitnessAdapter<'a, 'guest> {
+    pub(in crate::provider::graph::native) fn new(
+        engine: &'a Engine<'guest>,
+        tool: &'a Installed,
+        fresh: &'a dyn Fn() -> Result<(), CandidateError>,
+    ) -> Self {
+        Self {
+            base: Adapter::new(engine, fresh),
+            tool,
+        }
+    }
+    fn exact(&mut self, captured: &Observation, deadline: Instant) -> Result<(), CandidateError> {
+        self.base.verify(&captured.binding, deadline)?;
+        if self.base.inspect(&captured.volume.name, deadline)?.as_ref() != Some(captured) {
+            return Err(refused());
+        }
+        self.base.check(deadline)
+    }
+    fn invoke(&self, request: Request, deadline: Instant) -> Result<Reply, CandidateError> {
+        self.tool
+            .invoke(self.base.engine, request, deadline, self.base.fresh)
+    }
+}
+impl enrollment::sealed::Transport for WitnessAdapter<'_, '_> {}
+impl enrollment::Transport for WitnessAdapter<'_, '_> {
+    fn verify(&mut self, expected: &Binding, deadline: Instant) -> Result<(), CandidateError> {
+        self.base.verify(expected, deadline)
+    }
+    fn inspect(
+        &mut self,
+        name: &str,
+        deadline: Instant,
+    ) -> Result<Option<Observation>, CandidateError> {
+        self.base.inspect(name, deadline)
+    }
+    fn create_new(
+        &mut self,
+        request: &enrollment::CreateRequest,
+        deadline: Instant,
+    ) -> Result<Observation, CandidateError> {
+        self.base.create_new(request, deadline)
+    }
+}
+impl enrollment::witnessed::Transport for WitnessAdapter<'_, '_> {
+    fn root(
+        &mut self,
+        volume: &Observation,
+        deadline: Instant,
+    ) -> Result<RootIdentity, CandidateError> {
+        self.exact(volume, deadline)?;
+        let Reply::Root(root) = self.invoke(Request::root(&volume.volume.name)?, deadline)? else {
+            return Err(refused());
+        };
+        if (root.device, root.inode)
+            != (
+                volume.volume.directory.device,
+                volume.volume.directory.inode,
+            )
+        {
+            return Err(refused());
+        }
+        self.exact(volume, deadline)?;
+        Ok(root.into())
+    }
+    fn seed(
+        &mut self,
+        request: &enrollment::witnessed::SeedRequest,
+        deadline: Instant,
+    ) -> Result<(), CandidateError> {
+        let observed = request.captured();
+        self.exact(observed, deadline)?;
+        let witness = request.witness();
+        let reply = self.invoke(
+            Request::bound(
+                &observed.volume.name,
+                true,
+                request.root().into(),
+                &witness.name,
+                &witness.value,
+            )?,
+            deadline,
+        )?;
+        if !matches!(reply, Reply::Seeded) {
+            return Err(refused());
+        }
+        self.exact(observed, deadline)
+    }
+    fn verify_witness(
+        &mut self,
+        volume: &Observation,
+        root: RootIdentity,
+        witness: &ExpectedWitness,
+        deadline: Instant,
+    ) -> Result<(), CandidateError> {
+        self.exact(volume, deadline)?;
+        let reply = self.invoke(
+            Request::bound(
+                &volume.volume.name,
+                false,
+                root.into(),
+                &witness.name,
+                &witness.value,
+            )?,
+            deadline,
+        )?;
+        if !matches!(reply, Reply::Verified) {
+            return Err(refused());
+        }
+        self.exact(volume, deadline)
+    }
+}
@@ -141,7 +141,7 @@ fn snapshot(binding: &Binding) -> Result<Binding, CandidateError> {
     }
     Ok(binding.clone())
 }
-fn refused() -> CandidateError {
+pub(super) fn refused() -> CandidateError {
     CandidateError::new(
         "native_persistent_data_enrollment",
         "Persistent data enrollment is incomplete, ambiguous or changed; retained data was not adopted or deleted.",
@@ -505,6 +505,17 @@ impl RecordPin {
     ) -> Result<Self, CandidateError> {
         let bytes = serde_json::to_vec(owner).map_err(|_| refused())?;
         super::decode(&bytes)?;
+        Self::create_bytes(path, bytes, sync, step)
+    }
+    fn create_bytes<S: Sync>(
+        path: &Path,
+        bytes: Vec<u8>,
+        sync: &mut S,
+        step: Step,
+    ) -> Result<Self, CandidateError> {
+        if bytes.is_empty() || bytes.len() > super::LIMIT {
+            return Err(refused());
+        }
         let mut file = OpenOptions::new()
             .read(true)
             .write(true)
@@ -524,13 +535,19 @@ impl RecordPin {
         Ok(pin)
     }
     fn read(path: &Path) -> Result<Self, CandidateError> {
+        Self::read_validated(path, |bytes| super::decode(bytes).map(|_| ()))
+    }
+    fn read_validated(
+        path: &Path,
+        validate: impl FnOnce(&[u8]) -> Result<(), CandidateError>,
+    ) -> Result<Self, CandidateError> {
         let file = OpenOptions::new()
             .read(true)
             .custom_flags(libc::O_NOFOLLOW | libc::O_NONBLOCK)
             .open(path)
             .map_err(|_| refused())?;
         let bytes = read_bytes(&file)?;
-        super::decode(&bytes)?;
+        validate(&bytes)?;
         let pin = Self {
             path: path.into(),
             file,
@@ -588,3 +605,5 @@ impl Sync for SystemSync {
 
 #[cfg(test)]
 mod tests;
+
+pub mod witnessed;