From 0fdc0fc3f93587727cdd68a57640cc7684181838 Mon Sep 17 00:00:00 2001 From: Edvane Cristina Santos Daltro Date: Mon, 2 Nov 2026 00:52:00 -0400 Subject: [PATCH] Journal binary source handoff through the native observation producer --- crates/cyber-app/src/host.rs | 7 + crates/cyber-core/src/binary/editor.rs | 9 +- .../cyber-core/src/binary/editor/changes.rs | 156 ++++ .../cyber-core/src/binary/editor/versions.rs | 8 +- crates/cyber-core/src/binary/storage.rs | 16 + .../cyber-core/src/hooks/file_observations.rs | 5 + .../src/http/binary_files/editing.rs | 5 + .../src/runtime/binary_documents.rs | 71 +- .../src/runtime/binary_documents_tests.rs | 136 +++- .../src/runtime/binary_effects.rs | 743 ++++++++++++++++++ crates/cyber-server/src/runtime/events.rs | 2 + .../src/runtime/file_observations.rs | 28 + crates/cyber-server/src/runtime/host.rs | 10 + crates/cyber-server/src/runtime/mod.rs | 1 + crates/cyber-tools/src/file_changes/owner.rs | 16 +- .../src/file_changes/owner/binary.rs | 153 ++++ crates/cyber-tools/src/host.rs | 7 + .../tests/file_observation_dispatch.rs | 98 +++ .../add-binary-hex-editor/implementation.md | 49 ++ .../owner-save-integration.md | 7 +- 20 files changed, 1515 insertions(+), 12 deletions(-) create mode 100644 crates/cyber-core/src/binary/editor/changes.rs create mode 100644 crates/cyber-server/src/runtime/binary_effects.rs create mode 100644 crates/cyber-tools/src/file_changes/owner/binary.rs diff --git a/crates/cyber-app/src/host.rs b/crates/cyber-app/src/host.rs index 7175c8ac7..41c6a4757 100644 --- a/crates/cyber-app/src/host.rs +++ b/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, + ) -> BoxFuture<'_, Result<(), String>> { + self.builtin.binary_file_changed(change) + } fn authorize_binary_edit<'a>( &'a self, turn: &'a TurnContext, diff --git a/crates/cyber-core/src/binary/editor.rs b/crates/cyber-core/src/binary/editor.rs index d6267b952..c9030d37d 100644 --- a/crates/cyber-core/src/binary/editor.rs +++ b/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 { + pub(super) fn ready(editor: &mut Editor, id: &str) -> Box { let review = editor.review(0, 128, || false).unwrap(); match editor .prepare_save(id, &review.fingerprint, || false) diff --git a/crates/cyber-core/src/binary/editor/changes.rs b/crates/cyber-core/src/binary/editor/changes.rs new file mode 100644 index 000000000..b3512acdb --- /dev/null +++ b/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 { + 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, 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, 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()); + } +} diff --git a/crates/cyber-core/src/binary/editor/versions.rs b/crates/cyber-core/src/binary/editor/versions.rs index a02cb1999..84e8501a7 100644 --- a/crates/cyber-core/src/binary/editor/versions.rs +++ b/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 { +pub(super) fn read_evidence( + dir: &cap_std::fs::Dir, + document: &str, + version: &str, +) -> Result { 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)?; diff --git a/crates/cyber-core/src/binary/storage.rs b/crates/cyber-core/src/binary/storage.rs index 561fd725e..e550b78bc 100644 --- a/crates/cyber-core/src/binary/storage.rs +++ b/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 { let (data, _parent, root) = open_root(data)?; Ok(Self { data, root }) @@ -432,6 +440,14 @@ fn existing_lock(dir: &Dir) -> Result { } impl Store { + #[cfg(any(target_os = "macos", target_os = "linux"))] + pub(super) fn storage_root(&self) -> Result { + 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 diff --git a/crates/cyber-core/src/hooks/file_observations.rs b/crates/cyber-core/src/hooks/file_observations.rs index 9cd5ea23d..402564911 100644 --- a/crates/cyber-core/src/hooks/file_observations.rs +++ b/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(value: &T) -> Result<(), String> { let bytes = serde_json::to_vec(value).map_err(|error| error.to_string())?; if bytes.len() > MAX_RECORD_BYTES { diff --git a/crates/cyber-server/src/http/binary_files/editing.rs b/crates/cyber-server/src/http/binary_files/editing.rs index b933d2465..dacd0d3df 100644 --- a/crates/cyber-server/src/http/binary_files/editing.rs +++ b/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), diff --git a/crates/cyber-server/src/runtime/binary_documents.rs b/crates/cyber-server/src/runtime/binary_documents.rs index 449942d71..a8c0756a8 100644 --- a/crates/cyber-server/src/runtime/binary_documents.rs +++ b/crates/cyber-server/src/runtime/binary_documents.rs @@ -75,9 +75,52 @@ pub(crate) struct BinaryWork { grants: (super::FileDecision, super::FileDecision), authority: super::AdmissionAuthority, cancel: tokio_util::sync::CancellationToken, + post: Arc>>>, } #[cfg(any(target_os = "macos", target_os = "linux"))] impl BinaryWork { + /// Metadata recovery of a previously acknowledged source change. Failure + /// leaves its notification pending and never changes a staged edit's receipt. + pub fn resume_notifications(&self, editor: &mut cyber_core::binary::editor::Editor) { + let result = editor.pending_notification_change().and_then(|change| { + change.map_or(Ok(()), |change| self.acknowledge_source(Arc::new(change))) + }); + if result.is_err() { + self.runtime.hook_notice( + &self.turn.session_id, + "FileChanged", + "A previously acknowledged binary save still has pending notification metadata", + ); + } + } + /// Retain known source acknowledgement for publication after path/checkout + /// ownership is released, including when the HTTP future was disposed. + pub fn acknowledge_source( + &self, + change: Arc, + ) -> Result<(), Error> { + if change.owner().session_id != self.turn.session_id + || change.resource().path() != self.resource.path() + { + return Err(Error::Changed); + } + let storage = self.runtime.binary_documents().storage()?; + change.belongs_to(&storage)?; + { + let mut post = self.post.lock().map_err(|_| Error::Io)?; + if post + .as_ref() + .is_some_and(|prior| prior.key() != change.key()) + { + return Err(Error::Limit); + } + *post = Some(change.clone()); + } + self.runtime + .queue_binary_change(&change) + .map_err(|_| Error::Io)?; + Ok(()) + } pub fn is_cancelled(&self) -> bool { self.cancel.is_cancelled() } @@ -351,7 +394,7 @@ impl super::Runtime { )); } // This scope owns no subprocess. The retained proof keeps the native - // checkout lock through every synchronous filesystem call, including crash. + // checkout lock through source mutation and journal settlement, including crash. let location = location .settle_retained() .map_err(super::RuntimeError::Invalid)?; @@ -376,6 +419,9 @@ impl super::Runtime { .binary_documents() .worker() .map_err(|error| super::RuntimeError::Invalid(error.to_string()))?; + let post = Arc::new(Mutex::new(None)); + let executor = tokio::runtime::Handle::current(); + let runtime = self.clone(); let context = BinaryWork { runtime: self.clone(), executor: tokio::runtime::Handle::current(), @@ -384,6 +430,7 @@ impl super::Runtime { grants, authority, cancel, + post: post.clone(), }; tokio::task::spawn_blocking(move || { let (_lifecycle, _location, _paths, _permit, _edit_permit) = @@ -392,7 +439,27 @@ impl super::Runtime { let result = work(context); // A returned unknown file outcome is not a live/unowned native process. // Its durable intent remains unresolved in the document journal. - activity.settle()?; + let settlement = activity.settle(); + // Keep lifecycle/capacity ownership until the post-commit handoff + // finishes, but release every filesystem mutation owner first. + drop((_location, _paths)); + let change = post + .lock() + .map_err(|_| { + super::RuntimeError::Invalid("Binary handoff registry unavailable".into()) + })? + .take(); + if let Some(change) = change { + let session = change.owner().session_id.clone(); + if let Err(error) = executor.block_on(runtime.publish_binary_change(change)) { + runtime.hook_notice( + &session, + "FileChanged", + &format!("Binary save notification remains pending: {error}"), + ); + } + } + settlement?; Ok(result) }) .await diff --git a/crates/cyber-server/src/runtime/binary_documents_tests.rs b/crates/cyber-server/src/runtime/binary_documents_tests.rs index c8d9a111f..2be545b79 100644 --- a/crates/cyber-server/src/runtime/binary_documents_tests.rs +++ b/crates/cyber-server/src/runtime/binary_documents_tests.rs @@ -16,8 +16,23 @@ impl ModelResolver for Models { struct Host( Arc, Arc, + Arc, ); impl ToolHost for Host { + fn binary_file_changed( + &self, + _: Arc, + ) -> BoxFuture<'_, Result<(), String>> { + Box::pin(async { + assert_eq!( + self.0.available_permits(), + 1, + "notification ran under the path mutation owner" + ); + self.2.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + Ok(()) + }) + } fn definitions(&self, _: &TurnContext) -> Vec { vec![] } @@ -51,7 +66,7 @@ impl ToolHost for Host { }) } } -async fn fixture() -> ( +pub(super) async fn fixture() -> ( tempfile::TempDir, Runtime, String, @@ -68,6 +83,18 @@ async fn fixture_with_policy( String, cyber_core::binary::Resource, Arc, +) { + fixture_with_observer(denied, Arc::new(std::sync::atomic::AtomicUsize::new(0))).await +} +async fn fixture_with_observer( + denied: Arc, + observed: Arc, +) -> ( + tempfile::TempDir, + Runtime, + String, + cyber_core::binary::Resource, + Arc, ) { let temp = tempfile::tempdir().unwrap(); let root = temp.path().canonicalize().unwrap(); @@ -85,7 +112,7 @@ async fn fixture_with_policy( .unwrap(), ), resolver: Arc::new(Models), - tools: Arc::new(Host(paths.clone(), denied)), + tools: Arc::new(Host(paths.clone(), denied, observed)), global_config_dir: root.join("config"), shell: "test".into(), claude_compat: false, @@ -221,6 +248,111 @@ async fn disposed_binary_request_retains_worker_ownership_until_actual_exit() { assert!(runtime.inner.activities.lock().unwrap().is_empty()); } +#[tokio::test] +async fn disposed_saved_binary_worker_hands_off_only_after_releasing_mutation_owners() { + use cyber_core::binary::{ + Document, Offset, + draft::{PatchBatch, Splice}, + editor::{DraftScope, Editor, Owner, SavePreparation}, + storage::StorageRoot, + }; + let observed = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let (temp, runtime, session, resource, paths) = fixture_with_observer( + Arc::new(std::sync::atomic::AtomicBool::new(false)), + observed.clone(), + ) + .await; + let data = tempfile::tempdir().unwrap(); + let storage = Arc::new(StorageRoot::prepare(data.path()).unwrap()); + runtime.configure_binary_storage(storage.clone()).unwrap(); + let root = temp.path().canonicalize().unwrap(); + let mut source = Document::open(&root, "bytes").unwrap(); + let mut editor = Editor::create( + storage.open("bin_post_commit").unwrap(), + Owner { + session_id: session.clone(), + directory: root.clone(), + root, + scope: DraftScope::Manual, + }, + &mut source, + || false, + ) + .unwrap(); + let turn = runtime + .file_turn(Some(&session), temp.path()) + .await + .unwrap(); + let (started, running) = tokio::sync::oneshot::channel(); + let (finish, wait) = std::sync::mpsc::channel(); + let task = { + let runtime = runtime.clone(); + tokio::spawn(async move { + runtime + .binary_file_work( + turn, + "bin_post_commit".into(), + resource, + (FileDecision::Allow, FileDecision::Allow), + move |work| { + editor + .patch( + "patch_one", + &PatchBatch { + revision: editor.draft().revision(), + splices: vec![Splice { + offset: Offset(0), + delete_length: Offset(1), + expected_hex: "00".into(), + insert_hex: "01".into(), + }], + }, + ) + .unwrap(); + let review = editor.review(0, 128, || false).unwrap(); + let SavePreparation::Ready(prepared) = editor + .prepare_save("save_one", &review.fingerprint, || false) + .unwrap() + else { + panic!("save"); + }; + editor + .commit_save_checked( + *prepared, + || work.is_cancelled(), + || work.verify(), + ) + .unwrap(); + work.acknowledge_source(Arc::new( + editor.acknowledged_change("save_one").unwrap().unwrap(), + )) + .unwrap(); + started.send(()).unwrap(); + wait.recv().unwrap(); + }, + ) + .await + }) + }; + running.await.unwrap(); + task.abort(); + assert!(task.await.unwrap_err().is_cancelled()); + assert_eq!(paths.available_permits(), 0); + assert_eq!(observed.load(std::sync::atomic::Ordering::SeqCst), 0); + finish.send(()).unwrap(); + tokio::time::timeout(std::time::Duration::from_secs(5), async { + while observed.load(std::sync::atomic::Ordering::SeqCst) == 0 { + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); + assert_eq!(paths.available_permits(), 1); + assert_eq!(observed.load(std::sync::atomic::Ordering::SeqCst), 1); + assert_eq!(std::fs::read(temp.path().join("bytes")).unwrap(), [1, 255]); + runtime.shutdown().await; +} + #[tokio::test] async fn unwound_binary_worker_does_not_claim_file_success_or_poison_native_process_recovery() { let (temp, runtime, session, resource, paths) = fixture().await; diff --git a/crates/cyber-server/src/runtime/binary_effects.rs b/crates/cyber-server/src/runtime/binary_effects.rs new file mode 100644 index 000000000..b34b0190e --- /dev/null +++ b/crates/cyber-server/src/runtime/binary_effects.rs @@ -0,0 +1,743 @@ +//! Durable binary-save handoff. Native observations remain owned by FileWatch; +//! these records neither mint UserInput nor replay hook callbacks after restart. +use super::{Runtime, RuntimeError}; +use cyber_core::{ + binary::receipts::Version, + hooks::file_observations::{self as observations, Stamp}, +}; +use cyber_store::{EventRegistry, Expected, NewEvent, StoredEvent}; +use rusqlite::{OptionalExtension, Transaction}; +use serde::{Deserialize, Serialize}; + +const CHANGED: &str = "binary.source.handoff.1"; + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct Handoff { + pub version: Version, + pub session_id: String, + pub directory: String, + pub resource: String, + pub state: State, +} +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)] +pub enum State { + Queued, + Claimed { + aggregate: String, + generation: String, + path: String, + stamp: Stamp, + }, + Observed { + observation_id: String, + }, + Suppressed { + reason: String, + }, +} +fn aggregate(key: &str) -> String { + format!("binary-source:{key}") +} +fn invalid(message: impl Into) -> RuntimeError { + RuntimeError::Invalid(message.into()) +} +fn decode(data: &str) -> Result { + serde_json::from_str(data).map_err(|error| error.to_string()) +} +fn valid_key(key: &str) -> bool { + key.len() == 68 && key.starts_with("ver_") && key[4..].bytes().all(|b| b.is_ascii_hexdigit()) +} +fn valid_source_path(path: &str) -> bool { + !path.is_empty() + && path.len() <= 4096 + && !path.contains('\0') + && std::path::Path::new(path) + .components() + .all(|part| matches!(part, std::path::Component::Normal(_))) +} +fn validate(record: &Handoff) -> Result<(), String> { + if !valid_key(&record.version.version_id) + || !cyber_core::ids::has_prefix(&record.session_id, "ses") + || !valid_source_path(&record.version.path) + || [&record.directory, &record.resource] + .into_iter() + .any(|path| { + path.len() > 4096 + || path.contains('\0') + || !std::path::Path::new(path).is_absolute() + }) + || serde_json::to_vec(record).map_err(|e| e.to_string())?.len() > 32 * 1024 + { + return Err("invalid binary source handoff".into()); + } + match &record.state { + State::Claimed { + aggregate, + generation, + path, + .. + } if !aggregate.starts_with("file-watch:") + || observations::validate_generation(generation).is_err() + || !observations::relative_name(path) => + { + Err("invalid native observation binding".into()) + } + State::Observed { observation_id } + if !cyber_core::ids::has_prefix(observation_id, "evt") => + { + Err("invalid observation reference".into()) + } + State::Suppressed { reason } + if !matches!( + reason.as_str(), + "no_observer" + | "source_superseded" + | "origin_expired" + | "outside_observed_location" + | "handoff_interrupted" + | "observation_unavailable" + | "unobserved_path" + ) => + { + Err("invalid handoff disposition".into()) + } + _ => Ok(()), + } +} +fn same_source(a: &Handoff, b: &Handoff) -> bool { + a.session_id == b.session_id + && a.directory == b.directory + && a.resource == b.resource + && serde_json::to_value(&a.version).ok() == serde_json::to_value(&b.version).ok() +} +pub(super) fn register(registry: &mut EventRegistry) { + registry + .register_with(CHANGED, |data| { + let record: Handoff = + serde_json::from_value(data.clone()).map_err(|e| e.to_string())?; + validate(&record) + }) + .expect("valid binary source handoff event"); +} +pub(super) fn project(tx: &Transaction<'_>, event: &StoredEvent) -> Result<(), String> { + if event.kind != CHANGED { + return Ok(()); + } + let next: Handoff = serde_json::from_value(event.data.clone()).map_err(|e| e.to_string())?; + if event.aggregate_id != aggregate(&next.version.version_id) { + return Err("binary handoff aggregate mismatch".into()); + } + let prior: Option = tx + .query_row( + "SELECT data FROM event WHERE aggregate_id=?1 AND seq Ok(()), + None => Err("binary handoff lacks queued source evidence".into()), + Some(prior) => { + let prior = decode(&prior)?; + if !same_source(&prior, &next) { + return Err("binary source evidence changed".into()); + } + match (&prior.state, &next.state) { + (State::Queued, State::Claimed { .. }) => verify_claim(tx, &next), + (State::Queued, State::Suppressed { .. }) => Ok(()), + (State::Claimed { .. }, State::Suppressed { .. }) => Ok(()), + (State::Claimed { .. }, State::Observed { observation_id }) => { + verify_observation(tx, &prior.state, observation_id) + } + _ => Err("binary handoff cannot replay or change its terminal disposition".into()), + } + } + } +} +fn verify_claim(tx: &Transaction<'_>, record: &Handoff) -> Result<(), String> { + let State::Claimed { + aggregate, + generation, + path, + stamp, + } = &record.state + else { + return Err("missing native claim".into()); + }; + let data: Option = tx + .query_row( + "SELECT data FROM event WHERE aggregate_id=?1 AND type=?2 ORDER BY seq DESC LIMIT 1", + (aggregate, observations::CHECKPOINT), + |row| row.get(0), + ) + .optional() + .map_err(|error| error.to_string())?; + let checkpoint: observations::Checkpoint = + serde_json::from_str(&data.ok_or("native checkpoint is absent")?) + .map_err(|error| error.to_string())?; + if !checkpoint.requires_native_lease + || checkpoint.directory != record.directory + || checkpoint.generation != *generation + || stamp.length != record.version.after_length.0 + || stamp.identity.is_none() + || std::path::Path::new(&checkpoint.directory).join(path) + != std::path::Path::new(&record.resource) + { + return Err("binary source differs from its native observation claim".into()); + } + Ok(()) +} +fn matches_observation( + kind: &str, + data: &str, + generation: &str, + path: &str, + stamp: &Stamp, +) -> bool { + match kind { + observations::CHANGED => serde_json::from_str::(data) + .ok() + .is_some_and(|delta| { + delta.generation == generation + && delta + .updates + .iter() + .any(|update| update.path == path && update.stamp.as_ref() == Some(stamp)) + }), + observations::CHECKPOINT => serde_json::from_str::(data) + .ok() + .is_some_and(|checkpoint| { + checkpoint.generation == generation && checkpoint.files.get(path) == Some(stamp) + }), + _ => false, + } +} +fn verify_observation(tx: &Transaction<'_>, claim: &State, id: &str) -> Result<(), String> { + let State::Claimed { + aggregate, + generation, + path, + stamp, + } = claim + else { + return Err("missing native claim".into()); + }; + let observation: Option<(String, String, String)> = tx + .query_row( + "SELECT aggregate_id,type,data FROM event WHERE id=?1", + [id], + |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)), + ) + .optional() + .map_err(|e| e.to_string())?; + match observation { + Some((owner, kind, data)) + if owner == *aggregate + && matches_observation(&kind, &data, generation, path, stamp) => + { + Ok(()) + } + _ => Err("binary handoff reference does not prove the native source observation".into()), + } +} +impl Runtime { + /// Invoke after the actual filesystem worker releases checkout/path owners. + /// A repeated receipt may reconcile durable evidence, never re-enqueue an + /// already claimed native observation or renew expired hook provenance. + #[cfg(any(target_os = "macos", target_os = "linux"))] + pub async fn publish_binary_change( + &self, + change: std::sync::Arc, + ) -> Result { + let record = self.queue_binary_change(&change)?; + if matches!(record.state, State::Queued) { + self.inner + .tools + .binary_file_changed(change.clone()) + .await + .map_err(invalid)?; + } + self.reconcile_binary_handoff(change.key()) + } + #[cfg(any(target_os = "macos", target_os = "linux"))] + pub fn claim_binary_observation( + &self, + watch: &super::file_observations::FileWatch, + change: &cyber_core::binary::editor::AcknowledgedChange, + ) -> Result { + self.queue_binary_change(change)?; + let (sequence, mut record) = self + .binary_handoff(change.key())? + .ok_or_else(|| invalid("Binary handoff is absent"))?; + if !matches!(record.state, State::Queued) { + return Ok(record); + } + let (aggregate, _, state) = watch.binary_source_binding(self)?; + if state.directory != record.directory { + return self.suppress_binary_change(change.key(), "outside_observed_location"); + } + let relative = std::path::Path::new(&record.resource) + .strip_prefix(&state.directory) + .ok() + .map(cyber_core::vcs::slash_path); + let Some(path) = relative else { + return self.suppress_binary_change(change.key(), "outside_observed_location"); + }; + if !observations::relative_name(&path) { + return self.suppress_binary_change(change.key(), "unobserved_path"); + } + let stamp = match change.observed_stamp() { + Ok(stamp) => stamp, + Err(_) => return self.suppress_binary_change(change.key(), "source_superseded"), + }; + watch.binary_source_binding(self)?; + record.state = State::Claimed { + aggregate: aggregate.into(), + generation: state.generation.clone(), + path, + stamp, + }; + self.append_binary_handoff(sequence, &record)?; + Ok(record) + } + /// Link existing durable native evidence only. This never records another + /// file change, renews UserInput or dispatches a hook, including on restart. + pub fn reconcile_binary_handoff(&self, key: &str) -> Result { + let (sequence, mut record) = self + .binary_handoff(key)? + .ok_or_else(|| invalid("Binary handoff is absent"))?; + let State::Claimed { + aggregate, + generation, + path, + stamp, + } = &record.state + else { + return Ok(record); + }; + let (aggregate, generation, path, stamp) = ( + aggregate.clone(), + generation.clone(), + path.clone(), + stamp.clone(), + ); + let retained_generation = generation.clone(); + let observed = self.inner.store.read(move |db| { + let mut query = db.prepare( + "SELECT id,type,data FROM event WHERE aggregate_id=?1 ORDER BY seq DESC LIMIT ?2", + )?; + let rows = + query.query_map((&aggregate, (observations::MAX_REPLAY + 2) as i64), |row| { + Ok(( + row.get::<_, String>(0)?, + row.get::<_, String>(1)?, + row.get::<_, String>(2)?, + )) + })?; + for row in rows { + let (id, kind, data) = row?; + if matches_observation(&kind, &data, &generation, &path, &stamp) { + return Ok(Some(id)); + } + } + Ok(None) + })?; + if let Some(observation_id) = observed { + record.state = State::Observed { observation_id }; + self.append_binary_handoff(sequence, &record)?; + } else if self + .inner + .file_observation_roots + .get(&retained_generation) + .is_none_or(|root| !root.matches(std::path::Path::new(&record.directory))) + { + return self.suppress_binary_change(key, "handoff_interrupted"); + } + Ok(record) + } + /// Metadata only. HTTP callers must separately enforce owner/read scope. + pub fn binary_handoff(&self, key: &str) -> Result, RuntimeError> { + if !valid_key(key) { + return Err(invalid("Invalid binary source key")); + } + let id = aggregate(key); + self.inner + .store + .read(move |db| { + db.query_row( + "SELECT seq,data FROM event WHERE aggregate_id=?1 ORDER BY seq DESC LIMIT 1", + [id], + |row| Ok((row.get::<_, i64>(0)?, row.get::<_, String>(1)?)), + ) + .optional() + .map_err(Into::into) + })? + .map(|(seq, data)| { + decode(&data) + .map(|record| (seq, record)) + .map_err(RuntimeError::Corrupt) + }) + .transpose() + } + #[cfg(any(target_os = "macos", target_os = "linux"))] + pub fn queue_binary_change( + &self, + change: &cyber_core::binary::editor::AcknowledgedChange, + ) -> Result { + let storage = self + .binary_documents() + .storage() + .map_err(|e| invalid(e.to_string()))?; + change + .belongs_to(&storage) + .map_err(|e| invalid(e.to_string()))?; + let queued = Handoff { + version: change.version().clone(), + session_id: change.owner().session_id.clone(), + directory: change + .owner() + .directory + .to_str() + .ok_or_else(|| invalid("Invalid source Location"))? + .into(), + resource: change + .resource() + .path() + .to_str() + .ok_or_else(|| invalid("Invalid source resource"))? + .into(), + state: State::Queued, + }; + if let Some((_, prior)) = self.binary_handoff(change.key())? { + if !same_source(&prior, &queued) { + return Err(invalid("Binary source key differs from retained evidence")); + } + return Ok(prior); + } + self.inner.store.append( + &aggregate(change.key()), + Expected::Seq(-1), + vec![NewEvent::new( + CHANGED, + serde_json::to_value(&queued).map_err(|e| invalid(e.to_string()))?, + )], + )?; + Ok(queued) + } + pub fn suppress_binary_change(&self, key: &str, reason: &str) -> Result { + let (seq, mut record) = self + .binary_handoff(key)? + .ok_or_else(|| invalid("Binary handoff is absent"))?; + if matches!( + record.state, + State::Observed { .. } | State::Suppressed { .. } + ) { + return Ok(record); + } + record.state = State::Suppressed { + reason: reason.into(), + }; + self.append_binary_handoff(seq, &record)?; + Ok(record) + } + fn append_binary_handoff(&self, sequence: i64, record: &Handoff) -> Result<(), RuntimeError> { + self.inner.store.append( + &aggregate(&record.version.version_id), + Expected::Seq(sequence), + vec![NewEvent::new( + CHANGED, + serde_json::to_value(record).map_err(|e| invalid(e.to_string()))?, + )], + )?; + Ok(()) + } +} + +#[cfg(all(test, any(target_os = "macos", target_os = "linux")))] +mod tests { + use super::*; + use cyber_core::binary::{ + Document, Offset, + draft::{PatchBatch, Splice}, + editor::{DraftScope, Editor, Owner, SavePreparation}, + storage::StorageRoot, + }; + use std::{collections::BTreeMap, sync::Arc}; + + async fn saved() -> ( + tempfile::TempDir, + tempfile::TempDir, + Runtime, + Editor, + super::super::file_observations::FileWatch, + ) { + let (project, runtime, session, _, _) = + super::super::binary_documents_tests::fixture().await; + let data = tempfile::tempdir().unwrap(); + let storage = Arc::new(StorageRoot::prepare(data.path()).unwrap()); + runtime.configure_binary_storage(storage.clone()).unwrap(); + let root = project.path().canonicalize().unwrap(); + let watch = runtime + .claim_pinned_file_watch( + root.clone(), + std::fs::File::open(&root).unwrap(), + BTreeMap::new(), + ) + .await + .unwrap(); + let owner = Owner { + session_id: session, + directory: root.clone(), + root: root.clone(), + scope: DraftScope::Manual, + }; + let mut source = Document::open(&root, "bytes").unwrap(); + let mut editor = Editor::create( + storage.open("bin_handoff").unwrap(), + owner, + &mut source, + || false, + ) + .unwrap(); + editor + .patch( + "patch_one", + &PatchBatch { + revision: editor.draft().revision(), + splices: vec![Splice { + offset: Offset(0), + delete_length: Offset(1), + expected_hex: "00".into(), + insert_hex: "01".into(), + }], + }, + ) + .unwrap(); + let review = editor.review(0, 128, || false).unwrap(); + let SavePreparation::Ready(prepared) = editor + .prepare_save("saved", &review.fingerprint, || false) + .unwrap() + else { + panic!("native save"); + }; + editor.commit_save(*prepared).unwrap(); + (project, data, runtime, editor, watch) + } + #[tokio::test] + async fn handoff_links_real_durable_observations_once_without_running_or_replaying_hooks() { + let (_project, _data, runtime, mut editor, mut watch) = saved().await; + let proof = editor.acknowledged_change("saved").unwrap().unwrap(); + let queued = runtime.queue_binary_change(&proof).unwrap(); + assert!(matches!(queued.state, State::Queued)); + assert!(matches!( + runtime.queue_binary_change(&proof).unwrap().state, + State::Queued + )); + assert!( + runtime + .append_binary_handoff( + 0, + &Handoff { + state: State::Observed { + observation_id: "evt_fake".into() + }, + ..queued + } + ) + .is_err() + ); + let claimed = runtime.claim_binary_observation(&watch, &proof).unwrap(); + assert!(matches!(claimed.state, State::Claimed { .. })); + assert!(matches!( + runtime.reconcile_binary_handoff(proof.key()).unwrap().state, + State::Claimed { .. } + )); + let observation = watch + .record(observations::Delta { + generation: watch.state().generation.clone(), + updates: vec![observations::Update { + path: "bytes".into(), + stamp: Some(proof.observed_stamp().unwrap()), + }], + remaining: watch.state().remaining, + dispatch: false, + renewal: None, + }) + .await + .unwrap(); + let linked = runtime.reconcile_binary_handoff(proof.key()).unwrap(); + assert!( + matches!(linked.state,State::Observed {ref observation_id} if observation_id == &observation.id) + ); + let before = runtime + .inner + .store + .aggregate_seq(&aggregate(proof.key())) + .unwrap(); + drop(watch); // Reconciliation survives loss of the native generation and launches no callbacks. + assert!(matches!( + runtime.reconcile_binary_handoff(proof.key()).unwrap().state, + State::Observed { .. } + )); + assert!(matches!( + runtime.queue_binary_change(&proof).unwrap().state, + State::Observed { .. } + )); + assert_eq!( + runtime + .inner + .store + .aggregate_seq(&aggregate(proof.key())) + .unwrap(), + before + ); + assert!( + runtime + .append_binary_handoff( + before.unwrap(), + &Handoff { + state: State::Queued, + ..linked + } + ) + .is_err() + ); + runtime.shutdown().await; + } + #[tokio::test] + async fn superseded_sources_are_suppressed_and_fake_observation_references_are_rejected() { + let (project, _data, runtime, mut editor, mut watch) = saved().await; + let proof = editor.acknowledged_change("saved").unwrap().unwrap(); + runtime.claim_binary_observation(&watch, &proof).unwrap(); + let other = watch + .record(observations::Delta { + generation: watch.state().generation.clone(), + updates: vec![observations::Update { + path: "other".into(), + stamp: Some(proof.observed_stamp().unwrap()), + }], + remaining: watch.state().remaining, + dispatch: false, + renewal: None, + }) + .await + .unwrap(); + let (sequence, claimed) = runtime.binary_handoff(proof.key()).unwrap().unwrap(); + assert!( + runtime + .append_binary_handoff( + sequence, + &Handoff { + state: State::Observed { + observation_id: other.id + }, + ..claimed + } + ) + .is_err() + ); + runtime + .suppress_binary_change(proof.key(), "handoff_interrupted") + .unwrap(); + assert!(matches!( + runtime.reconcile_binary_handoff(proof.key()).unwrap().state, + State::Suppressed { .. } + )); + std::fs::write(project.path().join("bytes"), [9, 8]).unwrap(); + assert!(proof.observed_stamp().is_err()); + assert!(matches!( + runtime.queue_binary_change(&proof).unwrap().state, + State::Suppressed { .. } + )); + assert_eq!(std::fs::read(project.path().join("bytes")).unwrap(), [9, 8]); + runtime.shutdown().await; + } + #[tokio::test] + async fn a_source_changed_before_handoff_creates_no_successful_observation() { + let (project, _data, runtime, mut editor, watch) = saved().await; + let proof = editor.acknowledged_change("saved").unwrap().unwrap(); + std::fs::write(project.path().join("bytes"), [9, 8]).unwrap(); + let disposition = runtime.claim_binary_observation(&watch, &proof).unwrap(); + assert!( + matches!(disposition.state,State::Suppressed {ref reason} if reason == "source_superseded") + ); + let (_, sequence, _) = watch.binary_source_binding(&runtime).unwrap(); + assert_eq!( + sequence, 0, + "superseded handoff manufactured a native change" + ); + assert_eq!(std::fs::read(project.path().join("bytes")).unwrap(), [9, 8]); + runtime.shutdown().await; + } + #[tokio::test] + async fn losing_the_original_watcher_cannot_replay_an_unobserved_claim() { + let (project, _data, runtime, mut editor, watch) = saved().await; + let change = Arc::new(editor.acknowledged_change("saved").unwrap().unwrap()); + runtime.claim_binary_observation(&watch, &change).unwrap(); + drop(watch); + let disposition = runtime.publish_binary_change(change.clone()).await.unwrap(); + assert!( + matches!(disposition.state,State::Suppressed {ref reason} if reason == "handoff_interrupted") + ); + assert!(matches!( + runtime.publish_binary_change(change).await.unwrap().state, + State::Suppressed { .. } + )); + assert_eq!( + std::fs::read(project.path().join("bytes")).unwrap(), + [1, 255] + ); + runtime.shutdown().await; + } + #[tokio::test] + async fn native_names_outside_the_watch_path_contract_have_an_explicit_disposition() { + let (project, _data, runtime, editor, watch) = saved().await; + let owner = editor.owner().clone(); + std::fs::write(project.path().join("outside:watch"), [0, 255]).unwrap(); + let storage = runtime.binary_documents().storage().unwrap(); + let mut source = Document::open(&owner.root, "outside:watch").unwrap(); + let mut other = Editor::create( + storage.open("bin_unobserved").unwrap(), + owner, + &mut source, + || false, + ) + .unwrap(); + other + .patch( + "patch_one", + &PatchBatch { + revision: other.draft().revision(), + splices: vec![Splice { + offset: Offset(0), + delete_length: Offset(1), + expected_hex: "00".into(), + insert_hex: "01".into(), + }], + }, + ) + .unwrap(); + let review = other.review(0, 128, || false).unwrap(); + let SavePreparation::Ready(prepared) = other + .prepare_save("saved", &review.fingerprint, || false) + .unwrap() + else { + panic!("save"); + }; + other.commit_save(*prepared).unwrap(); + let change = other.acknowledged_change("saved").unwrap().unwrap(); + let disposition = runtime.claim_binary_observation(&watch, &change).unwrap(); + assert!( + matches!(disposition.state,State::Suppressed {ref reason} if reason == "unobserved_path") + ); + assert_eq!(watch.binary_source_binding(&runtime).unwrap().1, 0); + assert_eq!( + std::fs::read(project.path().join("outside:watch")).unwrap(), + [1, 255] + ); + runtime.shutdown().await; + } +} diff --git a/crates/cyber-server/src/runtime/events.rs b/crates/cyber-server/src/runtime/events.rs index 885e62d26..6ae69b7c1 100644 --- a/crates/cyber-server/src/runtime/events.rs +++ b/crates/cyber-server/src/runtime/events.rs @@ -548,6 +548,7 @@ pub fn registry() -> EventRegistry { super::reservations::register(&mut registry); super::hooks::register(&mut registry); super::file_observations::register(&mut registry); + super::binary_effects::register(&mut registry); super::jobs::register(&mut registry); super::delegations::register(&mut registry); super::mcp::register(&mut registry); @@ -573,6 +574,7 @@ fn project(tx: &Transaction<'_>, e: &StoredEvent) -> Result<(), String> { super::admission_authority::project(tx, e)?; super::hooks::project(tx, e)?; super::file_observations::project(tx, e)?; + super::binary_effects::project(tx, e)?; super::auto_statistics::project(tx, e)?; super::auto_override::project(tx, e)?; project_event(tx, e).map_err(|err| err.to_string())?; diff --git a/crates/cyber-server/src/runtime/file_observations.rs b/crates/cyber-server/src/runtime/file_observations.rs index aee85396d..119f82ed6 100644 --- a/crates/cyber-server/src/runtime/file_observations.rs +++ b/crates/cyber-server/src/runtime/file_observations.rs @@ -325,6 +325,34 @@ impl Runtime { } impl FileWatch { + #[cfg(any(target_os = "macos", target_os = "linux"))] + pub(super) fn binary_source_binding( + &self, + owner: &Runtime, + ) -> Result<(&str, i64, &Checkpoint), RuntimeError> { + let runtime = self.runtime()?; + if !Arc::ptr_eq(&runtime.inner, &owner.inner) { + return Err(RuntimeError::Invalid( + "Binary observation has a different runtime owner".into(), + )); + } + let root = self.pinned_root.as_ref().ok_or_else(|| { + RuntimeError::Invalid("Binary source requires a native observation owner".into()) + })?; + if !self.current()? + || !runtime + .inner + .file_observation_roots + .get(&self.state.generation) + .is_some_and(|current| Arc::ptr_eq(¤t, root)) + || !root.matches(std::path::Path::new(&self.state.directory)) + { + return Err(RuntimeError::Conflict( + "Binary observation generation changed".into(), + )); + } + Ok((&self.aggregate, self.sequence, &self.state)) + } pub(super) fn validate_native_scope(&self, directory: &str) -> Result<(), RuntimeError> { if let Some(root) = &self.pinned_root { if !root.matches(std::path::Path::new(directory)) { diff --git a/crates/cyber-server/src/runtime/host.rs b/crates/cyber-server/src/runtime/host.rs index 3f21a477a..fbf250310 100644 --- a/crates/cyber-server/src/runtime/host.rs +++ b/crates/cyber-server/src/runtime/host.rs @@ -1220,6 +1220,16 @@ pub trait ToolHost: Send + Sync { ) -> BoxFuture<'a, FileDecision> { Box::pin(async { FileDecision::Deny }) } + /// Queue acknowledged source evidence with the existing native producer. + /// Call only after filesystem mutation owners are released. This must never + /// manufacture UserInput, directly execute a hook or replay an old callback. + #[cfg(any(target_os = "macos", target_os = "linux"))] + fn binary_file_changed( + &self, + _change: Arc, + ) -> BoxFuture<'_, Result<(), String>> { + Box::pin(async { Err("Binary observation producer is unavailable".into()) }) + } fn authorize_binary_read<'a>( &'a self, _turn: &'a TurnContext, diff --git a/crates/cyber-server/src/runtime/mod.rs b/crates/cyber-server/src/runtime/mod.rs index 4eeaca6dc..4fe3c3805 100644 --- a/crates/cyber-server/src/runtime/mod.rs +++ b/crates/cyber-server/src/runtime/mod.rs @@ -52,6 +52,7 @@ mod hooks; mod installation; pub use installation::{InstallationNotice, UpdateStatus}; pub(crate) mod binary_documents; +pub mod binary_effects; #[cfg(any(target_os = "macos", target_os = "linux"))] mod binary_recovery; mod file_observation_hooks; diff --git a/crates/cyber-tools/src/file_changes/owner.rs b/crates/cyber-tools/src/file_changes/owner.rs index b80ba91cd..7f8631ec8 100644 --- a/crates/cyber-tools/src/file_changes/owner.rs +++ b/crates/cyber-tools/src/file_changes/owner.rs @@ -25,6 +25,9 @@ const MAX_LOCATIONS: usize = 32; const MAX_SESSIONS: usize = 256; const TICK: Duration = Duration::from_millis(150); const RECONCILE: Duration = Duration::from_secs(3); +#[cfg(any(target_os = "macos", target_os = "linux"))] +#[path = "owner/binary.rs"] +mod binary; #[derive(Default)] struct Membership { sessions: BTreeMap, @@ -36,6 +39,8 @@ struct Membership { renewals: BTreeMap, scopes: BTreeSet, explicit: Hints, + #[cfg(any(target_os = "macos", target_os = "linux"))] + binary: BTreeMap, } type Members = Arc>; impl Membership { @@ -144,6 +149,8 @@ impl Observers { renewals: BTreeMap::new(), scopes: BTreeSet::new(), explicit: Hints::default(), + #[cfg(any(target_os = "macos", target_os = "linux"))] + binary: BTreeMap::new(), })); let weak = host.weak.clone(); let root = path.clone(); @@ -777,6 +784,8 @@ impl RunState { return false; } if !host.observes_files(directory, &sessions).await { + #[cfg(any(target_os = "macos", target_os = "linux"))] + binary::suppress_pending(host, membership); self.observer = None; return true; } @@ -790,7 +799,12 @@ impl RunState { }; current.refresh_scopes(host, membership, cancel).await; current.renew(host, membership, cancel).await; - if let Err(error) = current.reconcile(host, &sessions, membership, cancel).await { + #[cfg(any(target_os = "macos", target_os = "linux"))] + let binary = current.claim_binary(host, membership); + let result = current.reconcile(host, &sessions, membership, cancel).await; + #[cfg(any(target_os = "macos", target_os = "linux"))] + current.finish_binary(host, &binary); + if let Err(error) = result { if cancel.is_cancelled() { return false; } diff --git a/crates/cyber-tools/src/file_changes/owner/binary.rs b/crates/cyber-tools/src/file_changes/owner/binary.rs new file mode 100644 index 000000000..9205912dc --- /dev/null +++ b/crates/cyber-tools/src/file_changes/owner/binary.rs @@ -0,0 +1,153 @@ +//! Binary acknowledgements are hints to the existing native producer. They do +//! not arm a new observer, renew input provenance or directly run callbacks. +use super::*; +use cyber_core::binary::editor::AcknowledgedChange; +use cyber_server::runtime::binary_effects::State as HandoffState; + +pub(super) struct Pending { + change: Arc, + member: CancellationToken, +} +impl Observers { + pub(crate) fn binary_changed( + &self, + host: &BuiltinHost, + change: Arc, + ) -> Result<(), String> { + let runtime = host.runtime().ok_or("runtime unavailable")?; + let queued = runtime + .queue_binary_change(&change) + .map_err(|error| error.to_string())?; + if !matches!(queued.state, HandoffState::Queued) { + runtime + .reconcile_binary_handoff(change.key()) + .map_err(|error| error.to_string())?; + return Ok(()); + } + // Registry/membership locks protect only this bounded queue. No journal + // IO, native metadata check or callback takes place under these locks. + let accepted = { + let registry = self.0.lock().unwrap_or_else(PoisonError::into_inner); + registry + .entries + .get(&change.owner().directory) + .filter(|entry| !entry.cancel.is_cancelled()) + .map(|entry| { + let mut members = entry.members.lock().unwrap_or_else(PoisonError::into_inner); + let member = members + .sessions + .get(&change.owner().session_id) + .filter(|cancel| !cancel.is_cancelled()) + .cloned(); + match member { + Some(_) if members.binary.contains_key(change.key()) => Ok(true), + Some(member) if members.binary.len() < 8 => { + members.binary.insert( + change.key().into(), + Pending { + change: change.clone(), + member, + }, + ); + Ok(true) + } + Some(_) => Err("Binary observation queue limit reached"), + None => Ok(false), + } + }) + .unwrap_or(Ok(false)) + }?; + if !accepted { + runtime + .suppress_binary_change(change.key(), "no_observer") + .map_err(|error| error.to_string())?; + } + Ok(()) + } +} +fn take(membership: &Members) -> BTreeMap { + std::mem::take( + &mut membership + .lock() + .unwrap_or_else(PoisonError::into_inner) + .binary, + ) +} +pub(super) fn suppress_pending(host: &BuiltinHost, membership: &Members) { + let pending = take(membership); + let Some(runtime) = host.runtime() else { + return; + }; + for (key, pending) in pending { + if let Err(error) = runtime.suppress_binary_change(&key, "no_observer") { + notice( + host, + &pending.change.owner().session_id, + &format!("Binary observation disposition remains pending: {error}"), + ); + } + } +} +impl Observation { + pub(super) fn claim_binary( + &self, + host: &BuiltinHost, + membership: &Members, + ) -> Vec> { + let pending = take(membership); + let Some(runtime) = host.runtime() else { + return vec![]; + }; + let mut claimed = vec![]; + for (key, pending) in pending { + let result = if pending.member.is_cancelled() { + runtime.suppress_binary_change(&key, "origin_expired") + } else { + runtime.claim_binary_observation(&self.journal, &pending.change) + }; + match result { + Ok(record) => { + if let HandoffState::Claimed { path, .. } = record.state { + membership + .lock() + .unwrap_or_else(PoisonError::into_inner) + .explicit + .record(Path::new(&path)); + // Keep every native pin through reconciliation and its + // durable observation handoff, including after HTTP close. + claimed.push(pending.change); + } + } + Err(error) => notice( + host, + &pending.change.owner().session_id, + &format!("Binary file observation remains pending: {error}"), + ), + } + } + claimed + } + pub(super) fn finish_binary(&self, host: &BuiltinHost, changes: &[Arc]) { + let Some(runtime) = host.runtime() else { + return; + }; + for change in changes { + let key = change.key(); + let session = &change.owner().session_id; + let result = runtime.reconcile_binary_handoff(key).and_then(|record| { + if matches!(record.state, HandoffState::Claimed { .. }) { + runtime.suppress_binary_change(key, "observation_unavailable") + } else { + Ok(record) + } + }); + if let Err(error) = result { + notice( + host, + session, + &format!("Binary native observation reference remains pending: {error}"), + ); + } + } + } +} diff --git a/crates/cyber-tools/src/host.rs b/crates/cyber-tools/src/host.rs index 889deefc7..232a8784a 100644 --- a/crates/cyber-tools/src/host.rs +++ b/crates/cyber-tools/src/host.rs @@ -964,6 +964,13 @@ impl ToolHost for BuiltinHost { serde_json::json!({"requires_confirmation": true}), )) } + #[cfg(any(target_os = "macos", target_os = "linux"))] + fn binary_file_changed( + &self, + change: Arc, + ) -> BoxFuture<'_, Result<(), String>> { + Box::pin(async move { self.file_observers.binary_changed(self, change) }) + } fn authorize_binary_edit<'a>( &'a self, turn: &'a TurnContext, diff --git a/crates/cyber-tools/tests/file_observation_dispatch.rs b/crates/cyber-tools/tests/file_observation_dispatch.rs index f928654c4..80f8d37ab 100644 --- a/crates/cyber-tools/tests/file_observation_dispatch.rs +++ b/crates/cyber-tools/tests/file_observation_dispatch.rs @@ -182,6 +182,104 @@ fn automatic_hooks(flow: &support::flow::Flow, recorder: &Recorder) { }]}]})); } +#[cfg(any(target_os = "macos", target_os = "linux"))] +#[tokio::test] +async fn acknowledged_binary_save_uses_the_native_producer_once_without_inference() { + use cyber_core::binary::{ + Document, Offset, + draft::{PatchBatch, Splice}, + editor::{DraftScope, Editor, Owner, SavePreparation}, + storage::StorageRoot, + }; + use cyber_server::runtime::binary_effects::State; + let flow = support::flow::Flow::new(vec![], false); + let private = tempfile::tempdir().unwrap(); + let storage = Arc::new(StorageRoot::prepare(private.path()).unwrap()); + flow.runtime + .configure_binary_storage(storage.clone()) + .unwrap(); + let root = flow.f.repo.canonicalize().unwrap(); + std::fs::write(root.join("binary.fixture"), [0, 255, 128, 27]).unwrap(); + let recorder = Recorder::start().await; + automatic_hooks(&flow, &recorder); + let session = flow.session("default").await; + await_condition("binary source baseline", || { + journal_count(&flow, records::CHECKPOINT) == 1 + }) + .await; + let mut source = Document::open(&root, "binary.fixture").unwrap(); + let mut editor = Editor::create( + storage.open("bin_native_save").unwrap(), + Owner { + session_id: session, + directory: root.clone(), + root: root.clone(), + scope: DraftScope::Manual, + }, + &mut source, + || false, + ) + .unwrap(); + editor + .patch( + "patch_one", + &PatchBatch { + revision: editor.draft().revision(), + splices: vec![Splice { + offset: Offset(0), + delete_length: Offset(2), + expected_hex: "00ff".into(), + insert_hex: "7f00".into(), + }], + }, + ) + .unwrap(); + let review = editor.review(0, 128, || false).unwrap(); + let SavePreparation::Ready(prepared) = editor + .prepare_save("native_save", &review.fingerprint, || false) + .unwrap() + else { + panic!("native save"); + }; + editor.commit_save(*prepared).unwrap(); + let change = Arc::new(editor.acknowledged_change("native_save").unwrap().unwrap()); + flow.runtime + .publish_binary_change(change.clone()) + .await + .unwrap(); + await_condition("acknowledged binary native observation",|| { + matches!(flow.runtime.binary_handoff(change.key()).unwrap(),Some((_,ref handoff)) if matches!(handoff.state,State::Observed {..})) + }).await; + await_condition("binary FileChanged callback", || { + recorder + .seen + .lock() + .unwrap() + .iter() + .any(|event| event["file_path"] == "binary.fixture") + }) + .await; + let observations = journal_count(&flow, records::CHANGED); + let callbacks = recorder.seen.lock().unwrap().len(); + for _ in 0..3 { + let handoff = flow + .runtime + .publish_binary_change(change.clone()) + .await + .unwrap(); + assert!(matches!(handoff.state, State::Observed { .. })); + } + assert_eq!(journal_count(&flow, records::CHANGED), observations); + assert_eq!(recorder.seen.lock().unwrap().len(), callbacks); + assert_eq!(callbacks, 1); + assert_eq!( + std::fs::read(root.join("binary.fixture")).unwrap(), + [127, 0, 128, 27] + ); + assert!(flow.requests("test/main").is_empty()); + flow.runtime.shutdown().await; +} + #[tokio::test] async fn native_owner_delivers_external_edits_rename_removal_to_shared_sessions_and_joins() { let flow = support::flow::Flow::new(vec![], false); diff --git a/openspec/changes/add-binary-hex-editor/implementation.md b/openspec/changes/add-binary-hex-editor/implementation.md index 5e25f6be8..9bcf4f004 100644 --- a/openspec/changes/add-binary-hex-editor/implementation.md +++ b/openspec/changes/add-binary-hex-editor/implementation.md @@ -448,3 +448,52 @@ the final check and an explicitly authorized successful follow-up. Local results 66 binary core tests, five runtime worker/proof tests and all 23 production file route tests passed; affected core/server library/test Clippy passed. This does not activate public save or complete the outstanding history/notification integrations. + +## Native observation handoff + +Only a durable acknowledged changed source operation can mint an opaque handoff +proof. It retains native pins and the configured private-root identity, publishes +its immutable saved-version evidence first, and confirms the installed source's +metadata before the producer claims it. Draft-only/no-effect/unknown operations +cannot manufacture this proof; another configured private namespace is refused. + +The SQL handoff journal records queued, claimed and terminal observed/suppressed +states. Claims bind the existing native watch generation, Location, relative path +and exact stamp. An observed disposition references a real durable matching native +delta/checkpoint; unrelated paths and fabricated references are rejected. Repeated +publication/recovery links retained evidence without dispatching another callback. +An unobserved claim whose original native generation is gone is suppressed as +interrupted, with no new UserInput or callback replay. + +The BuiltinHost queues bounded metadata hints on its existing observation actor. +Its normal privacy scan, committed delta, feedback budget, membership incarnation +and hook admission checks remain the producer. No configured observer, expired +membership, a superseded source and incomplete observation have explicit pending +or suppressed dispositions. Production callers must publish after the filesystem +worker releases mutation ownership. Owner save routes and core notification receipt +settlement are not yet connected. + +Actual workers durably queue known acknowledgements and retain their proof through +a post-commit handoff even if the HTTP future is dropped. They release native +Location/path mutation owners before enqueueing with the observation actor; the +lifecycle/capacity owners remain until enqueue finishes. A worker regression verifies +both HTTP disposal and that the host receives the handoff only after path release. +Explicit private staging can resume pending metadata for a previously acknowledged +source save; draft-only operations never mint a source change. Native pins remain +retained through the actor's reconciliation and durable reference settlement. + +The real BuiltinHost/HTTP-hook integration regression passes: an actual native +binary replacement produces one callback, three receipt retries produce no extra +observations/callbacks, the bytes remain exact and inference is unused. Runtime +regressions cover durable reference linking, terminal-state fencing, wrong-path +references and superseded sources. Further recovery/negative cases and final scoped +checks are being completed; this does not claim full editor acceptance or retention. + +Current local checks pass 68 binary core tests (one parent-invoked child helper is +ignored by the ordinary selector), 11 runtime worker/handoff tests, all 23 production +file-route tests and the real native producer/HTTP-hook integration. Affected four- +crate library/test Clippy, formatting/diff checks and all 114 strict OpenSpec items +passed after the post-commit lifetime changes. The independent namespace regression +was split from the larger source-proof test to keep the complexity gate enabled. +PR #98 merged at `69372bab` after all 22 checks passed. PR #100 carries the independent +saved-version/commit-authority predecessor; the producer follow-up is not merged. diff --git a/openspec/changes/add-binary-hex-editor/owner-save-integration.md b/openspec/changes/add-binary-hex-editor/owner-save-integration.md index 67c113aec..b5ea4bb5f 100644 --- a/openspec/changes/add-binary-hex-editor/owner-save-integration.md +++ b/openspec/changes/add-binary-hex-editor/owner-save-integration.md @@ -8,8 +8,11 @@ acceptance or an expansion of the model/plugin scope. ## Execution and file outcome are separate A filesystem-only worker retains the native Location, workspace/path, lifecycle and -capacity owners until the actual blocking worker finishes. No subprocess or hook may -escape that worker. The worker's tagged SQL activity records execution ownership; +capacity owners through the actual source work and journal settlement. A following +metadata-only handoff releases native Location/path mutation owners first, retaining +lifecycle/capacity ownership until enqueue finishes, including after HTTP disposal. +It cannot launch or wait for a hook callback. No subprocess escapes source work. +The worker's tagged SQL activity records execution ownership; the private document journal records acknowledged/conflict/unknown file outcomes. Explicit document recovery obtains the real exclusive private document lock. Its