diff --git a/crates/cyber-core/src/binary/editor.rs b/crates/cyber-core/src/binary/editor.rs
index 82b18838..397af716 100644
--- a/crates/cyber-core/src/binary/editor.rs
+++ b/crates/cyber-core/src/binary/editor.rs
@@ -95,6 +95,23 @@ pub struct Editor {
draft: Draft,
record: Record,
}
+/// Borrows the actual exclusive document store lock, not a boolean supplied by
+/// a caller. This proves filesystem-only execution ownership, never a file outcome.
+pub struct RecoveryProof<'a> {
+ editor: &'a Editor,
+}
+impl RecoveryProof<'_> {
+ pub fn document_id(&self) -> &str {
+ self.editor.store.document_id()
+ }
+ pub fn owner(&self) -> &Owner {
+ self.editor.owner()
+ }
+ pub fn validate(&self, storage: &super::storage::StorageRoot) -> Result<(), Error> {
+ self.editor.store.belongs_to(storage)
+ }
+}
+
pub struct PreparedSave {
intent: Intent,
after: Draft,
@@ -133,10 +150,13 @@ impl Write for Counter {
Ok(())
}
}
-fn persist(store: &mut Store, record: &Record) -> Result<(), Error> {
+fn encoded_size(value: &impl Serialize) -> Result {
let mut count = Counter(0);
- serde_json::to_writer(&mut count, record).map_err(|_| Error::Io)?;
- store.record("document.json", count.0, |writer| {
+ serde_json::to_writer(&mut count, value).map_err(|_| Error::Io)?;
+ Ok(count.0)
+}
+fn persist(store: &mut Store, record: &Record) -> Result<(), Error> {
+ store.record("document.json", encoded_size(record)?, |writer| {
serde_json::to_writer(writer, record).map_err(|_| Error::Io)
})
}
@@ -223,6 +243,10 @@ impl Editor {
record,
})
}
+ pub fn recovery_proof(&self) -> Result, Error> {
+ self.store.validate()?;
+ Ok(RecoveryProof { editor: self })
+ }
pub fn draft(&self) -> &Draft {
&self.draft
}
@@ -562,6 +586,29 @@ impl Editor {
if cancelled() {
return Err(Error::Cancelled);
}
+ // Only the pending JSON value changes. Count the prospective publication
+ // by streaming serialization, without cloning the bounded draft history.
+ let pending_size = encoded_size(&Some(&prepared.intent))?;
+ let prior_pending_size = encoded_size(&self.record.pending)?;
+ let intent_bytes = encoded_size(&self.record)?
+ .checked_sub(prior_pending_size)
+ .and_then(|size| size.checked_add(pending_size))
+ .ok_or(Error::Limit)?;
+ // The installed observation is always a 64-character SHA-256 token. Its
+ // value is captured after exchange; its encoded size is known beforehand.
+ let mut installed = prepared.intent.after.clone();
+ installed.observed = "0".repeat(64);
+ let (outcome, _) = self.acknowledged_record(&prepared.intent, installed, &prepared.after);
+ self.store.reserve_exchange(
+ intent_bytes,
+ encoded_size(&outcome)?,
+ prepared
+ .intent
+ .prior
+ .length
+ .0
+ .saturating_sub(prepared.intent.after.length.0),
+ )?;
self.record.pending = Some(prepared.intent);
persist(&mut self.store, &self.record)?;
let intent = self.record.pending.as_ref().ok_or(Error::Invalid)?;
@@ -618,17 +665,31 @@ impl Editor {
intent.after_draft.clone(),
)?,
};
+ // Preserve the prior intent if publishing this outcome fails.
+ let (next, operation) = self.acknowledged_record(intent, installed_target, &draft);
+ if persist(&mut self.store, &next).is_err() {
+ return Ok(unknown);
+ }
+ self.record = next;
+ self.draft = draft;
+ self._pins._file = Some(installed_pin);
+ Ok(operation)
+ }
+ fn acknowledged_record(
+ &self,
+ intent: &Intent,
+ target: Target,
+ draft: &Draft,
+ ) -> (Record, Operation) {
let mut operation = intent.operation.clone();
operation.outcome = Outcome::Acknowledged;
operation.file_effect = FileEffect::Changed;
operation.reason = None;
operation.notification_pending = true;
- // Build the next durable record without discarding the prior intent if a
- // record write fails after the file was installed.
- let next = Record {
+ let record = Record {
version: 1,
owner: self.record.owner.clone(),
- target: installed_target,
+ target,
base: intent.new_base.clone(),
draft: draft.snapshot(),
operations: self
@@ -645,13 +706,7 @@ impl Editor {
operation: operation.id.clone(),
}),
};
- if persist(&mut self.store, &next).is_err() {
- return Ok(unknown);
- }
- self.record = next;
- self.draft = draft;
- self._pins._file = Some(installed_pin);
- Ok(operation)
+ (record, operation)
}
/// Called only after the runtime durably publishes the acknowledged observation.
/// The operation ID is the producer's deduplication identity, not just a path.
@@ -816,6 +871,97 @@ mod tests {
assert!(editor.validate_read_resource().is_err());
}
+ #[test]
+ fn restore_reserves_the_displaced_file_quota_before_any_native_effect() {
+ let (root, mut editor, path, _owner) = setup();
+ editor
+ .patch(
+ "grow_for_restore",
+ &PatchBatch {
+ revision: editor.draft().revision(),
+ splices: vec![crate::binary::draft::Splice {
+ offset: Offset(0),
+ delete_length: Offset(8),
+ expected_hex: crate::binary::draft::encode_hex(b"original"),
+ insert_hex: "ff".repeat(48 * 1024),
+ }],
+ },
+ )
+ .unwrap();
+ let save = ready(&mut editor, "save_grown");
+ assert_eq!(
+ editor.commit_save(*save).unwrap().outcome,
+ Outcome::Acknowledged
+ );
+ let review = editor.restore_review(|| false).unwrap();
+ for outcome_space_missing in [false, true] {
+ let SavePreparation::Ready(prepared) = editor
+ .prepare_restore("restore_quota", &review.fingerprint, || false)
+ .unwrap()
+ else {
+ panic!("restore preparation");
+ };
+ let intent_bytes = serde_json::to_vec(&prepared.intent).unwrap().len() as u64;
+ let displaced_extra = prepared.intent.prior.length.0 - prepared.intent.after.length.0;
+ let allowance = if outcome_space_missing {
+ let mut installed = prepared.intent.after.clone();
+ installed.observed = "0".repeat(64);
+ let (outcome, _) =
+ editor.acknowledged_record(&prepared.intent, installed, &prepared.after);
+ let publication = encoded_size(&editor.record).unwrap() + intent_bytes
+ - encoded_size(&editor.record.pending).unwrap();
+ let growth = publication - encoded_size(&editor.record).unwrap();
+ let allowance = displaced_extra + growth + encoded_size(&outcome).unwrap() / 2;
+ assert!(allowance > publication, "intent publication must still fit");
+ allowance
+ } else {
+ let allowance = intent_bytes + 4096;
+ assert!(
+ allowance < displaced_extra,
+ "displacement must exceed publication allowance"
+ );
+ allowance
+ };
+ let directory = root.path().join("data/binary/bin_one");
+ let used = std::fs::read_dir(&directory)
+ .unwrap()
+ .map(|entry| entry.unwrap().metadata().unwrap().len())
+ .sum::();
+ let filler_path = directory.join("quota_filler");
+ let filler = std::fs::OpenOptions::new()
+ .write(true)
+ .create_new(true)
+ .open(&filler_path)
+ .unwrap();
+ // Sparse allocation: exercise logical document quota without allocating 2 GiB.
+ filler
+ .set_len(super::super::storage::MAX_STORAGE - used - allowance)
+ .unwrap();
+ let result = editor.commit_save(*prepared);
+ assert!(
+ matches!(result, Err(Error::Limit)),
+ "{result:?}; source length after refusal: {}",
+ std::fs::metadata(&path).unwrap().len()
+ );
+ assert!(!editor.unresolved());
+ assert!(editor.lookup_operation("restore_quota").unwrap().is_none());
+ assert_eq!(std::fs::read(&path).unwrap(), vec![255; 48 * 1024]);
+ drop(filler);
+ std::fs::remove_file(filler_path).unwrap();
+ }
+ let SavePreparation::Ready(prepared) = editor
+ .prepare_restore("restore_quota", &review.fingerprint, || false)
+ .unwrap()
+ else {
+ panic!("restore after capacity release");
+ };
+ assert_eq!(
+ editor.commit_save(*prepared).unwrap().outcome,
+ Outcome::Acknowledged
+ );
+ assert_eq!(std::fs::read(path).unwrap(), b"original");
+ }
+
#[test]
fn changed_replacement_bytes_are_refused_before_intent_or_source_effects() {
let (root, mut editor, path, _owner) = setup();
@@ -1306,6 +1452,18 @@ mod tests {
&owner,
)
.unwrap();
+ let storage =
+ super::super::storage::StorageRoot::prepare(&root.path().join("data")).unwrap();
+ resumed
+ .recovery_proof()
+ .unwrap()
+ .validate(&storage)
+ .unwrap();
+ assert_eq!(
+ resumed.pending_operation().unwrap().outcome,
+ Outcome::Unknown,
+ "exclusive execution proof must not acknowledge an uncertain file effect"
+ );
assert_eq!(resumed.reconcile().unwrap().outcome, Outcome::Unknown);
assert!(resumed.unresolved());
assert_eq!(std::fs::read(path).unwrap(), b"original");
diff --git a/crates/cyber-core/src/binary/storage.rs b/crates/cyber-core/src/binary/storage.rs
index 079a9534..82b02465 100644
--- a/crates/cyber-core/src/binary/storage.rs
+++ b/crates/cyber-core/src/binary/storage.rs
@@ -384,7 +384,36 @@ pub(super) fn flush_dir(dir: &Dir) -> Result<(), Error> {
}
}
+#[cfg(not(windows))]
+fn existing_lock(dir: &Dir) -> Result {
+ let mut options = OpenOptions::new();
+ options.read(true).write(true).follow(FollowSymlinks::No);
+ #[cfg(unix)]
+ {
+ use cap_std::fs::OpenOptionsExt;
+ options.custom_flags(libc::O_NONBLOCK);
+ }
+ let file = dir
+ .open_with("lock", &options)
+ .map_err(|_| Error::File)?
+ .into_std();
+ private(dir, &file, false)?;
+ Ok(file)
+}
+
impl Store {
+ #[cfg(any(target_os = "macos", target_os = "linux"))]
+ pub(super) fn document_id(&self) -> &str {
+ &self.id
+ }
+ #[cfg(any(target_os = "macos", target_os = "linux"))]
+ pub(super) fn belongs_to(&self, root: &StorageRoot) -> Result<(), Error> {
+ self.validate()?;
+ if self.data_path != root.data || identity(&self.root)? != identity(root.directory()?)? {
+ return Err(Error::Changed);
+ }
+ Ok(())
+ }
#[cfg(any(target_os = "macos", target_os = "linux"))]
pub(super) fn directory(&self) -> &Dir {
&self.dir
@@ -413,31 +442,31 @@ impl Store {
Self::locked(data_path, data, root, dir, id)
}
fn locked(data_path: PathBuf, data: Dir, root: Dir, dir: Dir, id: &str) -> Result {
+ // Only an empty newly allocated namespace may create its first lock.
+ // A missing lock beside existing recovery data is corruption, not repair.
+ let may_create_lock = dir.entries().map_err(|_| Error::Io)?.next().is_none();
#[cfg(windows)]
let lock = {
let owner = dir.try_clone().map_err(|_| Error::Io)?.into_std_file();
- match crate::memory::windows::create_private_file(&owner, "lock") {
- Ok(created) => drop(created),
- Err(crate::memory::MemoryStorageError::Io(e))
- if e.kind() == std::io::ErrorKind::AlreadyExists => {}
- Err(_) => return Err(Error::File),
+ if may_create_lock {
+ match crate::memory::windows::create_private_file(&owner, "lock") {
+ Ok(created) => drop(created),
+ Err(crate::memory::MemoryStorageError::Io(e))
+ if e.kind() == std::io::ErrorKind::AlreadyExists => {}
+ Err(_) => return Err(Error::File),
+ }
}
crate::memory::windows::open_pinned_private_file(&owner, "lock")
.map_err(|_| Error::File)?
};
#[cfg(not(windows))]
- let lock = match create(&dir, "lock") {
- Ok(f) => f,
- Err(_) => {
- let mut options = OpenOptions::new();
- options.read(true).write(true).follow(FollowSymlinks::No);
- let file = dir
- .open_with("lock", &options)
- .map_err(|_| Error::File)?
- .into_std();
- private(&dir, &file, false)?;
- file
+ let lock = if may_create_lock {
+ match create(&dir, "lock") {
+ Ok(file) => file,
+ Err(_) => existing_lock(&dir)?,
}
+ } else {
+ existing_lock(&dir)?
};
lock.try_lock().map_err(|_| Error::Limit)?;
flush_dir(&dir)?;
@@ -487,9 +516,49 @@ impl Store {
&dir.try_clone().map_err(|_| Error::Io)?.into_std_file(),
true,
)?;
+ #[cfg(any(target_os = "macos", target_os = "linux"))]
+ if super::native::id(&existing_lock(&dir)?)? != super::native::id(&self._lock)? {
+ return Err(Error::Changed);
+ }
Ok(())
}
fn capacity(&self, additional: u64) -> Result<(), Error> {
+ self.reserve(additional, additional)
+ }
+ /// Atomic exchange moves the old source into the private stage name. A
+ /// shrinking replacement can therefore increase private usage after exchange.
+ /// Reserve both the temporary intent publication and the final displaced file.
+ #[cfg(any(target_os = "macos", target_os = "linux"))]
+ pub(super) fn reserve_exchange(
+ &self,
+ intent_bytes: u64,
+ outcome_bytes: u64,
+ displaced_extra: u64,
+ ) -> Result<(), Error> {
+ self.validate()?;
+ if intent_bytes > MAX_RECORD || outcome_bytes > MAX_RECORD {
+ return Err(Error::Limit);
+ }
+ let existing = self
+ .dir
+ .symlink_metadata("document.json")
+ .map_err(|_| Error::Io)?
+ .len();
+ let after_publication = intent_bytes.saturating_sub(existing);
+ let outcome_publication = after_publication
+ .checked_add(outcome_bytes)
+ .ok_or(Error::Limit)?;
+ let future_private = displaced_extra
+ .checked_add(outcome_publication)
+ .ok_or(Error::Limit)?;
+ // Moving the existing source allocates no payload blocks. Both record
+ // publication peaks, including the acknowledgement, reserve quota/space.
+ self.reserve(
+ intent_bytes.max(future_private),
+ intent_bytes.max(outcome_publication),
+ )
+ }
+ fn reserve(&self, private_extra: u64, allocate: u64) -> Result<(), Error> {
let mut used = 0u64;
for (count, entry) in self.dir.entries().map_err(|_| Error::Io)?.enumerate() {
if count >= 4096 {
@@ -505,14 +574,19 @@ impl Store {
}
used = used.checked_add(metadata.len()).ok_or(Error::Limit)?;
}
- if used.checked_add(additional).is_none_or(|n| n > MAX_STORAGE) {
+ if used
+ .checked_add(private_extra)
+ .is_none_or(|n| n > MAX_STORAGE)
+ {
return Err(Error::Limit);
}
+ #[cfg(not(unix))]
+ let _ = allocate;
#[cfg(unix)]
{
let stat = rustix::fs::fstatvfs(self.dir.as_ref()).map_err(|_| Error::Io)?;
let available = stat.f_bavail.saturating_mul(stat.f_frsize);
- if available < additional.saturating_add(1024 * 1024) {
+ if available < allocate.saturating_add(1024 * 1024) {
return Err(Error::Limit);
}
}
@@ -860,6 +934,41 @@ mod tests {
}
}
+ #[cfg(any(target_os = "macos", target_os = "linux"))]
+ #[test]
+ fn missing_document_locks_are_not_repaired_and_live_owners_reject_rebinding() {
+ let data = tempfile::tempdir().unwrap();
+ let storage = StorageRoot::prepare(data.path()).unwrap();
+ let mut store = storage.open("bin_lock").unwrap();
+ store
+ .record("state.json", 2, |writer| {
+ writer.write_all(b"{}").map_err(|_| Error::Io)
+ })
+ .unwrap();
+ let lock = storage.path().join("bin_lock/lock");
+ std::fs::remove_file(&lock).unwrap();
+ assert!(store.validate().is_err());
+ assert!(storage.open_existing("bin_lock").is_err());
+ assert!(!lock.exists(), "recovery repaired a missing live lock");
+ assert!(storage.open("bin_lock").is_err());
+ assert!(
+ !lock.exists(),
+ "ordinary open repaired a missing recovery lock"
+ );
+ use std::os::unix::fs::OpenOptionsExt;
+ let replacement = std::fs::OpenOptions::new()
+ .write(true)
+ .create_new(true)
+ .mode(0o600)
+ .open(&lock)
+ .unwrap();
+ assert!(
+ store.validate().is_err(),
+ "old owner accepted a replacement lock inode"
+ );
+ drop(replacement);
+ }
+
#[test]
fn private_base_survives_store_reopen_and_feeds_a_recoverable_draft() {
let data = tempfile::tempdir().unwrap();
diff --git a/crates/cyber-server/src/http/binary_files/recovery.rs b/crates/cyber-server/src/http/binary_files/recovery.rs
index e870febb..b22cb74f 100644
--- a/crates/cyber-server/src/http/binary_files/recovery.rs
+++ b/crates/cyber-server/src/http/binary_files/recovery.rs
@@ -134,6 +134,9 @@ pub(super) async fn recover(
let _lifecycle = lifecycle;
let editor = Editor::resume(storage.open_existing(&id)?, &owner)?;
editor.validate_read_resource()?;
+ registry
+ .recover_binary_activity(editor.recovery_proof()?)
+ .map_err(|_| Error::Changed)?;
registry
.binary_documents()
.insert_recovered(id, owner, editor, input.read_only)
diff --git a/crates/cyber-server/src/runtime/binary_documents_tests.rs b/crates/cyber-server/src/runtime/binary_documents_tests.rs
index 10282e8f..565dd3ba 100644
--- a/crates/cyber-server/src/runtime/binary_documents_tests.rs
+++ b/crates/cyber-server/src/runtime/binary_documents_tests.rs
@@ -213,3 +213,137 @@ async fn session_deletion_does_not_infer_settlement_from_a_missing_local_binary_
drop(activity);
runtime.shutdown().await;
}
+
+#[tokio::test]
+async fn recovery_requires_the_configured_exclusive_document_and_leaves_generic_activity_alone() {
+ use cyber_core::binary::{
+ Document, Offset,
+ draft::{PatchBatch, Splice},
+ editor::{DraftScope, Editor, Owner},
+ storage::StorageRoot,
+ };
+ let (_temp, runtime, session, resource, _paths) = 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 = resource.path().parent().unwrap().to_owned();
+ storage.admit_project(&root).unwrap();
+ let owner = Owner {
+ session_id: session.clone(),
+ directory: root.clone(),
+ root,
+ scope: DraftScope::Manual,
+ };
+ let mut source = Document::open(&owner.root, "bytes").unwrap();
+ let mut editor = Editor::create(
+ storage.open("bin_recovery").unwrap(),
+ owner.clone(),
+ &mut source,
+ || false,
+ )
+ .unwrap();
+ let patch = editor
+ .patch(
+ "retained_patch",
+ &PatchBatch {
+ revision: editor.draft().revision(),
+ splices: vec![Splice {
+ offset: Offset(0),
+ delete_length: Offset(1),
+ expected_hex: "00".into(),
+ insert_hex: "aa".into(),
+ }],
+ },
+ )
+ .unwrap();
+ let revision = editor.draft().revision();
+ let handle = runtime.inner.handle(&session).await.unwrap();
+ let mut binary =
+ super::activity::Scope::reserve_binary(&runtime.inner, &handle, "bin_recovery".into())
+ .await
+ .unwrap();
+ binary.launch().unwrap();
+ let generic = super::activity::Scope::reserve(&runtime.inner, &handle)
+ .await
+ .unwrap();
+ let latest = |id: &str| {
+ let id = id.to_owned();
+ runtime
+ .inner
+ .store
+ .read(move |db| {
+ db.query_row(
+ "SELECT data FROM event WHERE aggregate_id=?1 ORDER BY seq DESC LIMIT 1",
+ [id],
+ |row| row.get::<_, String>(0),
+ )
+ .map_err(Into::into)
+ })
+ .unwrap()
+ };
+ assert!(
+ runtime
+ .recover_binary_activity(editor.recovery_proof().unwrap())
+ .is_err(),
+ "live local worker was marked finished"
+ );
+ drop(editor);
+ // The exclusive native lock must be reacquired after the old document closes.
+ let editor = Editor::resume(storage.open_existing("bin_recovery").unwrap(), &owner).unwrap();
+ runtime.inner.activities.lock().unwrap().remove(binary.id());
+ let other_data = tempfile::tempdir().unwrap();
+ let wrong_storage = StorageRoot::prepare(other_data.path()).unwrap();
+ let wrong = Editor::create(
+ wrong_storage.open("bin_recovery").unwrap(),
+ owner.clone(),
+ &mut source,
+ || false,
+ )
+ .unwrap();
+ assert!(
+ runtime
+ .recover_binary_activity(wrong.recovery_proof().unwrap())
+ .is_err(),
+ "another private root forged document settlement"
+ );
+ assert_eq!(
+ serde_json::from_str::(&latest(binary.id())).unwrap()["status"],
+ "pending"
+ );
+ assert_eq!(
+ runtime
+ .recover_binary_activity(editor.recovery_proof().unwrap())
+ .unwrap(),
+ 1
+ );
+ assert_eq!(
+ runtime
+ .recover_binary_activity(editor.recovery_proof().unwrap())
+ .unwrap(),
+ 0,
+ "recovery replay appended another disposition"
+ );
+ let disposition: serde_json::Value = serde_json::from_str(&latest(binary.id())).unwrap();
+ assert_eq!(disposition["status"], "settled");
+ assert_eq!(disposition["phase"], "filesystem_recovered_document_locked");
+ assert_eq!(
+ serde_json::from_str::(&latest(generic.id())).unwrap()["status"],
+ "pending"
+ );
+ assert_eq!(editor.draft().revision(), revision);
+ assert_eq!(
+ editor
+ .lookup_operation("retained_patch")
+ .unwrap()
+ .unwrap()
+ .payload_sha256,
+ patch.payload_sha256
+ );
+ assert!(
+ editor.lookup_operation("unproven_save").unwrap().is_none(),
+ "execution recovery fabricated a file receipt"
+ );
+ assert_eq!(std::fs::read(owner.root.join("bytes")).unwrap(), [0, 255]);
+ drop((generic, binary));
+ runtime.shutdown().await;
+}
diff --git a/crates/cyber-server/src/runtime/binary_recovery.rs b/crates/cyber-server/src/runtime/binary_recovery.rs
new file mode 100644
index 00000000..b89fab1d
--- /dev/null
+++ b/crates/cyber-server/src/runtime/binary_recovery.rs
@@ -0,0 +1,76 @@
+//! A document's actual exclusive lock proves that a prior filesystem-only worker
+//! cannot still install bytes through that document. Generic native processes need
+//! their own acknowledgement and are never settled by this proof.
+use super::{Runtime, RuntimeError, activity};
+use cyber_core::binary::editor::RecoveryProof;
+use cyber_store::{Expected, NewEvent};
+
+impl Runtime {
+ pub(crate) fn recover_binary_activity(
+ &self,
+ proof: RecoveryProof<'_>,
+ ) -> Result {
+ let storage = self
+ .binary_documents()
+ .storage()
+ .map_err(|_| RuntimeError::Conflict("Binary recovery storage unavailable".into()))?;
+ proof
+ .validate(&storage)
+ .map_err(|_| RuntimeError::Conflict("Binary recovery namespace changed".into()))?;
+ let session = proof.owner().session_id.clone();
+ let document = proof.document_id().to_owned();
+ if self
+ .inner
+ .activities
+ .lock()
+ .map_err(|_| RuntimeError::Conflict("Binary activity registry unavailable".into()))?
+ .values()
+ .any(|control| {
+ control.source == session && control.binary_document.as_deref() == Some(&document)
+ })
+ {
+ return Err(RuntimeError::Conflict(
+ "The binary filesystem worker is still active".into(),
+ ));
+ }
+ let rows = self.inner.store.read(move |db| {
+ let mut query = db.prepare("SELECT aggregate_id,seq,data,(SELECT data FROM event origin WHERE origin.aggregate_id=e.aggregate_id ORDER BY seq LIMIT 1) FROM event e WHERE type='native.activity.changed.1' AND json_extract(data,'$.session_id')=?1 AND json_extract(data,'$.binary_document')=?2 AND json_extract(data,'$.status')<>'settled' AND NOT EXISTS(SELECT 1 FROM event n WHERE n.aggregate_id=e.aggregate_id AND n.seq>e.seq) LIMIT 4097")?;
+ Ok(query.query_map([session, document], |row| Ok((row.get::<_,String>(0)?, row.get::<_,i64>(1)?, row.get::<_,String>(2)?, row.get::<_,String>(3)?)))?.collect::,_>>()?)
+ })?;
+ if rows.len() > 4096 {
+ return Err(RuntimeError::Conflict(
+ "Binary recovery activity limit exceeded".into(),
+ ));
+ }
+ let count = rows.len();
+ for (id, sequence, data, origin) in rows {
+ let mut record: activity::Record = serde_json::from_str(&data)
+ .map_err(|error| RuntimeError::Corrupt(error.to_string()))?;
+ let origin: activity::Record = serde_json::from_str(&origin)
+ .map_err(|error| RuntimeError::Corrupt(error.to_string()))?;
+ if origin.session_id != proof.owner().session_id
+ || origin.binary_document.as_deref() != Some(proof.document_id())
+ {
+ return Err(RuntimeError::Corrupt(
+ "Binary activity origin differs from its document".into(),
+ ));
+ }
+ proof
+ .validate(&storage)
+ .map_err(|_| RuntimeError::Conflict("Binary recovery namespace changed".into()))?;
+ record.status = "settled".into();
+ record.phase = "filesystem_recovered_document_locked".into();
+ record.admission_bindings = None;
+ self.inner.store.append(
+ &id,
+ Expected::Seq(sequence),
+ vec![NewEvent::new(
+ activity::CHANGED,
+ serde_json::to_value(record)
+ .map_err(|error| RuntimeError::Corrupt(error.to_string()))?,
+ )],
+ )?;
+ }
+ Ok(count)
+ }
+}
diff --git a/crates/cyber-server/src/runtime/mod.rs b/crates/cyber-server/src/runtime/mod.rs
index 16d2c56d..4eeaca6d 100644
--- a/crates/cyber-server/src/runtime/mod.rs
+++ b/crates/cyber-server/src/runtime/mod.rs
@@ -52,6 +52,8 @@ mod hooks;
mod installation;
pub use installation::{InstallationNotice, UpdateStatus};
pub(crate) mod binary_documents;
+#[cfg(any(target_os = "macos", target_os = "linux"))]
+mod binary_recovery;
mod file_observation_hooks;
mod file_observation_roots;
mod file_observation_scopes;
diff --git a/docs/binary-editor.md b/docs/binary-editor.md
index 41b8ca37..79649f82 100644
--- a/docs/binary-editor.md
+++ b/docs/binary-editor.md
@@ -131,5 +131,7 @@ Closing releases the in-memory handle, retaining private commands and receipts.
Use discovery in the same Session Location, recover the selected ID, then prepare
again before changing its draft. Recovery does not install bytes or replay a save.
A replaced/aliased parent refuses access; an ordinary changed source remains separate
-from the private draft. Production save/restore and crash reconciliation of native
-activity scopes are not yet exposed by these routes.
+from the private draft. Recovery uses its exclusive private document lock to settle
+matching filesystem-only activity records left by a prior process. It refuses active
+local workers and another private root; generic native processes and uncertain file
+effects remain unresolved. Production source save/restore is not yet exposed.
diff --git a/openspec/changes/add-binary-hex-editor/implementation.md b/openspec/changes/add-binary-hex-editor/implementation.md
index 4209718a..1163b1b2 100644
--- a/openspec/changes/add-binary-hex-editor/implementation.md
+++ b/openspec/changes/add-binary-hex-editor/implementation.md
@@ -375,3 +375,30 @@ as well as deserialization, with a cross-platform round-trip regression that che
an offset above 2^53. A Windows production-route regression explicitly checks native
edit refusal alongside successful live viewing and unchanged source bytes; that
Windows-only test remains subject to native CI. The local wire regression passed.
+
+## Save integration follow-up work
+
+The shrinking-restore quota regression reproduced a native replacement followed
+by an unknown receipt when the displaced old source exceeded private quota.
+Preflight now reserves displaced logical size and both intent/outcome publication
+peaks, preserving the source/draft on refusal. The regression also reserves enough
+for intent and displacement but too little for the acknowledgement, then verifies
+successful restore after releasing capacity. Sparse filler exercises logical quota
+without allocating 2 GiB of payload.
+
+Explicit owner recovery now connects an actual borrowed document-lock proof to
+matching tagged SQL filesystem activity. The proof must belong to the runtime's
+configured private root. Active local workers, another private root and generic
+native activity are not settled. This records execution completion only and never
+acknowledges an uncertain file effect. Runtime proof/isolation regressions passed;
+platform checks and the remaining [save activation work](owner-save-integration.md)
+still apply. Source installation remains unavailable through production routes.
+
+The follow-up's local binary suite passed 60 tests (one child helper ignored by the
+ordinary selector and exercised by its parent). Four runtime ownership/proof tests,
+all 23 production file-route tests, affected core/server library/test Clippy,
+formatting/generated checks and all 114 strict OpenSpec items passed. Review checked
+the actual acknowledgement record shape in the capacity forecast, lock-file binding
+and refusal to recreate missing recovery locks. Linux/macOS/Windows CI must still
+validate this follow-up. PR #97 merged at `d6a9b1ae` after all 24 current-head checks
+passed, including its Windows-specific live-view/unsupported-edit regression.
diff --git a/openspec/changes/add-binary-hex-editor/owner-save-integration.md b/openspec/changes/add-binary-hex-editor/owner-save-integration.md
new file mode 100644
index 00000000..67c113ae
--- /dev/null
+++ b/openspec/changes/add-binary-hex-editor/owner-save-integration.md
@@ -0,0 +1,57 @@
+# Owner save integration work
+
+This implements the already approved binary save/recovery, snapshot and FileChanged
+requirements. Source installation remains gated until the integrations and their
+regressions are present. This document records implementation decisions, not completed
+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;
+the private document journal records acknowledged/conflict/unknown file outcomes.
+
+Explicit document recovery obtains the real exclusive private document lock. Its
+borrowed proof is checked against the runtime's configured, pinned private root;
+a same-named document under another root is insufficient. The proof may settle
+only matching filesystem activity records with matching original provenance and
+no live local control. Generic native processes retain their own acknowledgement
+requirements. Recovery never exchanges files or manufactures a file receipt.
+A still-unknown native document intent remains unknown after execution recovery.
+Missing locks beside recovery data are not recreated; live stores recheck the lock's
+native identity. The actual lock descriptor stays owned while validation probes the
+binding ([Rust locking contract](https://doc.rust-lang.org/std/fs/struct.File.html#method.try_lock)).
+
+## Capacity before native effects
+
+Atomic exchange moves the old source into the private stage name. A shrinking save
+or restore can increase private logical usage even though it allocates no source
+payload blocks. Preflight reserves both record publication peaks, the displaced
+file delta and the outcome record before publishing native intent. Record sizes
+are counted by streaming serialization, including the actual acknowledgement
+shape, without cloning the draft command history. Available filesystem space and
+the 2 GiB logical document ceiling remain distinct checks. Failure retains the
+draft and source unchanged. External filesystem failure after intent still requires
+honest unknown-outcome handling and retained evidence.
+
+## Remaining save activation work
+
+- Refresh current read/edit/Mode/sandbox and captured ancestor authority at the
+ native commit boundary after expensive preparation. No permission hook or model
+ classifier may run while holding mutation ownership; a stricter final decision
+ refuses and requires explicit readmission.
+- Retain protected preimage and byte-range evidence for every acknowledged save,
+ including non-git and ordinary-snapshot-excluded files. Shared version discovery
+ exposes only acknowledged source versions, never another owner's unsaved commands.
+ Restore retains exact current-identity/content checks and separate byte review.
+- Durably queue shared version/diff and FileChanged work by document/operation
+ identity after native acknowledgement. A durable terminal disposition prevents
+ duplicate publication after lost responses. Unknown/failed operations produce
+ no fabricated success event, and expired hook provenance is never replayed as a
+ new user input. Integrate with the existing native observation producer.
+- Add owner save/restore/reconciliation routes, SDK helpers and terminal states
+ only after those invariants are tested. Model patch uses invocation-scoped
+ drafts and the same guarded save path, with plugin filesystem intersections.
+- Complete explicit retention/cleanup, native race/process-death tests, real
+ terminal captures, performance/accessibility evidence and final delivery review.