Journal binary save notifications through the native producer

crates/cyber-app/src/host.rs

@@ -391,6 +391,13 @@ impl ToolHost for AppHost {
         self.builtin
             .authorize_binary_read(turn, resource, asker, cancel)
     }
+    #[cfg(any(target_os = "macos", target_os = "linux"))]
+    fn binary_file_changed(
+        &self,
+        change: Arc<cyber_core::binary::editor::AcknowledgedChange>,
+    ) -> BoxFuture<'_, Result<(), String>> {
+        self.builtin.binary_file_changed(change)
+    }
     fn authorize_binary_edit<'a>(
         &'a self,
         turn: &'a TurnContext,

crates/cyber-core/src/binary/editor.rs

@@ -17,11 +17,14 @@ use std::path::PathBuf;
 
 #[path = "editor/catalog.rs"]
 mod catalog;
+#[path = "editor/changes.rs"]
+mod changes;
 #[path = "editor/restore.rs"]
 mod restore;
 #[path = "editor/versions.rs"]
 mod versions;
 pub use catalog::{StoredDocument, StoredDocuments};
+pub use changes::AcknowledgedChange;
 pub use versions::VersionView;
 
 const MAX_OPERATIONS: usize = 4096;
@@ -775,7 +778,7 @@ mod tests {
         storage::Store,
     };
     use std::path::Path;
-    fn setup() -> (tempfile::TempDir, Editor, PathBuf, Owner) {
+    pub(super) fn setup() -> (tempfile::TempDir, Editor, PathBuf, Owner) {
         let root = tempfile::tempdir().unwrap();
         let data = root.path().join("data");
         let project = root.path().join("project");
@@ -799,7 +802,7 @@ mod tests {
         .unwrap();
         (root, editor, path, owner)
     }
-    fn patch(editor: &mut Editor, id: &str) -> Operation {
+    pub(super) fn patch(editor: &mut Editor, id: &str) -> Operation {
         let batch = PatchBatch {
             revision: editor.draft().revision(),
             splices: vec![Splice {
@@ -811,7 +814,7 @@ mod tests {
         };
         editor.patch(id, &batch).unwrap()
     }
-    fn ready(editor: &mut Editor, id: &str) -> Box<PreparedSave> {
+    pub(super) fn ready(editor: &mut Editor, id: &str) -> Box<PreparedSave> {
         let review = editor.review(0, 128, || false).unwrap();
         match editor
             .prepare_save(id, &review.fingerprint, || false)

crates/cyber-core/src/binary/editor/changes.rs

@@ -0,0 +1,156 @@
+//! Native acknowledgement evidence for the shared observation producer. This
+//! type cannot be deserialized or manufactured from an HTTP/plugin payload.
+use super::*;
+use crate::hooks::file_observations::Stamp;
+
+pub struct AcknowledgedChange {
+    key: String,
+    owner: Owner,
+    resource: super::super::Resource,
+    version: crate::binary::receipts::Version,
+    target: Target,
+    storage: crate::binary::storage::StorageRoot,
+    _pins: native::Pins,
+}
+impl AcknowledgedChange {
+    pub fn belongs_to(&self, storage: &crate::binary::storage::StorageRoot) -> Result<(), Error> {
+        self.storage.same_namespace(storage)
+    }
+    pub fn key(&self) -> &str {
+        &self.key
+    }
+    pub fn owner(&self) -> &Owner {
+        &self.owner
+    }
+    pub fn resource(&self) -> &super::super::Resource {
+        &self.resource
+    }
+    pub fn version(&self) -> &crate::binary::receipts::Version {
+        &self.version
+    }
+    /// Metadata-only confirmation for the existing native scanner. This is not
+    /// writer attribution or permission to execute a hook. A superseded source
+    /// refuses confirmation rather than fabricating an observation of old bytes.
+    pub fn observed_stamp(&self) -> Result<Stamp, Error> {
+        use std::os::unix::fs::MetadataExt;
+        let (_, source) = self.target.validate()?;
+        let metadata = source.metadata().map_err(|_| Error::Io)?;
+        Ok(Stamp {
+            length: metadata.len(),
+            modified_ns: metadata
+                .modified()
+                .ok()
+                .and_then(|time| time.duration_since(std::time::UNIX_EPOCH).ok())
+                .and_then(|time| time.as_nanos().try_into().ok()),
+            identity: Some((metadata.dev(), metadata.ino())),
+        })
+    }
+}
+impl Editor {
+    /// A known earlier save can finish metadata publication after explicit
+    /// recovery. Unsaved edits and unknown native intents are never candidates.
+    pub fn pending_notification_change(&mut self) -> Result<Option<AcknowledgedChange>, Error> {
+        let pending = self
+            .record
+            .operations
+            .iter()
+            .rev()
+            .find(|operation| {
+                operation.notification_pending
+                    && operation.outcome == Outcome::Acknowledged
+                    && operation.file_effect == FileEffect::Changed
+            })
+            .map(|operation| operation.id.clone());
+        match pending {
+            Some(id) => self.acknowledged_change(&id),
+            None => Ok(None),
+        }
+    }
+    /// Retain durable source-version evidence before handing acknowledgement to
+    /// another producer. Unknown/conflict/no-effect operations never mint proof.
+    pub fn acknowledged_change(&mut self, id: &str) -> Result<Option<AcknowledgedChange>, Error> {
+        self.publish_last_version()?;
+        let Some(operation) = self.lookup_operation(id)? else {
+            return Ok(None);
+        };
+        if operation.outcome != Outcome::Acknowledged
+            || operation.file_effect != FileEffect::Changed
+        {
+            return Ok(None);
+        }
+        let version = format!(
+            "ver_{}",
+            signature(&(
+                "binary-version-v1",
+                self.store.document_id(),
+                &operation.id,
+                &operation.payload_sha256
+            ))?
+        );
+        let evidence =
+            versions::read_evidence(self.store.directory(), self.store.document_id(), &version)?;
+        if evidence.owner != self.record.owner
+            || evidence.operation.payload_sha256 != operation.payload_sha256
+        {
+            return Err(Error::Changed);
+        }
+        let pins = evidence.target.pins()?;
+        Ok(Some(AcknowledgedChange {
+            key: version,
+            owner: self.record.owner.clone(),
+            resource: self.resource(),
+            version: evidence.info,
+            target: evidence.target,
+            storage: self.store.storage_root()?,
+            _pins: pins,
+        }))
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    #[test]
+    fn only_durable_changed_acknowledgements_mint_source_change_evidence() {
+        let (_temp, mut editor, path, owner) = super::super::tests::setup();
+        assert!(editor.pending_notification_change().unwrap().is_none());
+        assert!(editor.acknowledged_change("absent").unwrap().is_none());
+        super::super::tests::patch(&mut editor, "patch_one");
+        assert!(editor.pending_notification_change().unwrap().is_none());
+        assert!(editor.acknowledged_change("patch_one").unwrap().is_none());
+        let prepared = super::super::tests::ready(&mut editor, "saved");
+        assert!(editor.acknowledged_change("saved").unwrap().is_none());
+        editor.commit_save(*prepared).unwrap();
+        let change = editor.acknowledged_change("saved").unwrap().unwrap();
+        assert_eq!(
+            editor.pending_notification_change().unwrap().unwrap().key(),
+            change.key()
+        );
+        assert_eq!(change.owner(), &owner);
+        assert_eq!(change.resource().path(), path.canonicalize().unwrap());
+        assert_eq!(change.version().after_length, Offset(4));
+        assert_eq!(change.key(), change.version().version_id);
+        assert_eq!(change.observed_stamp().unwrap().length, 4);
+        assert_eq!(
+            change.key(),
+            editor.acknowledged_change("saved").unwrap().unwrap().key()
+        );
+        std::fs::write(&path, [9, 8, 7, 6]).unwrap();
+        assert!(change.observed_stamp().is_err());
+        assert_eq!(std::fs::read(path).unwrap(), [9, 8, 7, 6]);
+    }
+    #[test]
+    fn acknowledgement_proof_belongs_only_to_its_configured_private_namespace() {
+        let (temp, mut editor, _, _) = super::super::tests::setup();
+        super::super::tests::patch(&mut editor, "patch_one");
+        let prepared = super::super::tests::ready(&mut editor, "saved");
+        editor.commit_save(*prepared).unwrap();
+        let change = editor.acknowledged_change("saved").unwrap().unwrap();
+        let storage =
+            crate::binary::storage::StorageRoot::prepare(&temp.path().join("data")).unwrap();
+        change.belongs_to(&storage).unwrap();
+        let other = tempfile::tempdir().unwrap();
+        let other = crate::binary::storage::StorageRoot::prepare(other.path()).unwrap();
+        assert!(change.belongs_to(&other).is_err());
+    }
+}

crates/cyber-core/src/binary/editor/versions.rs

@@ -9,7 +9,7 @@ pub(super) struct Evidence {
     version: u32,
     pub info: Version,
     pub operation: Operation,
-    owner: Owner,
+    pub(super) owner: Owner,
     before: BaseRef,
     after: BaseRef,
     pub target: Target,
@@ -409,7 +409,11 @@ fn valid_version(version: &str) -> bool {
         && version.starts_with("ver_")
         && version[4..].bytes().all(|b| b.is_ascii_hexdigit())
 }
-fn read_evidence(dir: &cap_std::fs::Dir, document: &str, version: &str) -> Result<Evidence, Error> {
+pub(super) fn read_evidence(
+    dir: &cap_std::fs::Dir,
+    document: &str,
+    version: &str,
+) -> Result<Evidence, Error> {
     let bytes =
         crate::binary::storage::small_record(dir, &format!("{version}.json"), 16 * 1024 * 1024)?;
     let evidence: Evidence = serde_json::from_slice(&bytes).map_err(|_| Error::Invalid)?;

crates/cyber-core/src/binary/storage.rs

@@ -90,6 +90,14 @@ pub struct StorageRoot {
     root: Dir,
 }
 impl StorageRoot {
+    #[cfg(any(target_os = "macos", target_os = "linux"))]
+    pub(super) fn same_namespace(&self, other: &Self) -> Result<(), Error> {
+        if self.data != other.data || identity(self.directory()?)? != identity(other.directory()?)?
+        {
+            return Err(Error::Changed);
+        }
+        Ok(())
+    }
     pub fn prepare(data: &Path) -> Result<Self, Error> {
         let (data, _parent, root) = open_root(data)?;
         Ok(Self { data, root })
@@ -432,6 +440,14 @@ fn existing_lock(dir: &Dir) -> Result<File, Error> {
 }
 
 impl Store {
+    #[cfg(any(target_os = "macos", target_os = "linux"))]
+    pub(super) fn storage_root(&self) -> Result<StorageRoot, Error> {
+        self.validate()?;
+        Ok(StorageRoot {
+            data: self.data_path.clone(),
+            root: self.root.try_clone().map_err(|_| Error::Io)?,
+        })
+    }
     #[cfg(any(target_os = "macos", target_os = "linux"))]
     pub(super) fn document_id(&self) -> &str {
         &self.id

crates/cyber-core/src/hooks/file_observations.rs

@@ -188,6 +188,11 @@ fn generation(value: &str) -> Result<(), String> {
         .map_err(|_| "invalid file watcher generation".into())
 }
 
+/// Validate a retained generation token without minting observation authority.
+pub fn validate_generation(value: &str) -> Result<(), String> {
+    generation(value)
+}
+
 fn bounded<T: Serialize>(value: &T) -> Result<(), String> {
     let bytes = serde_json::to_vec(value).map_err(|error| error.to_string())?;
     if bytes.len() > MAX_RECORD_BYTES {

crates/cyber-server/src/http/binary_files/editing.rs

@@ -246,6 +246,9 @@ pub(super) async fn prepare(
                         entry.editing.store(true, Ordering::Release);
                     }
                     *entry.edit_admission.lock().map_err(|_| Error::Io)? = Some(admission);
+                    if let Content::Draft(editor) = &mut *content {
+                        cancel.resume_notifications(editor);
+                    }
                     Ok(describe(&entry, &content, true, grant))
                 },
             )
@@ -347,6 +350,7 @@ pub(super) async fn patch(
                             splices: input.splices,
                         },
                     )?;
+                    cancel.resume_notifications(editor);
                     Ok(Edited {
                         operation,
                         status: describe(&entry, &content, true, grant),
@@ -413,6 +417,7 @@ pub(super) async fn history(
                             editor.discard(&input.operation_id, &input.revision)?
                         }
                     };
+                    cancel.resume_notifications(editor);
                     Ok(Edited {
                         operation,
                         status: describe(&entry, &content, true, grant),