refactor(fs): one fsatomic module, and publish the state plane at 0600

docs/vrs/08-harness-context/requirements.md

@@ -103,7 +103,13 @@ record.
   other purposes, buys nothing a named include entry does not — bounded by
   st2-side test that pins the names it expects (HC-R05), and by the fact that no
   correctness property here depends on the transport at all: everything works
-  with no replication, and remote visibility is what is lost.
+  with no replication, and remote visibility is what is lost. Both records are
+  published mode `0600` — as is every record st2's shared publication primitive
+  writes, though not the bus message files, which stay at the writing process's
+  umask — so a transport reading the catalog as a different unprivileged uid
+  needs that decision taken deliberately rather than inherited from a default.
+  No such transport runs today, which is also what leaves the wire-cost half of
+  `DQ-C1` unmeasured.
 
 ## Requirements
 

src/agent_publish.rs

@@ -676,7 +676,7 @@ fn copy_filtered_catalog(
                     name_text.as_ref(),
                     "resources" | "archive" | "inbox" | "status"
                 )
-                || declaration_parent && name_text.starts_with(".status.tmp-"))
+                || declaration_parent && name_text.starts_with(crate::status::TMP_STAGING_PREFIX))
         {
             continue;
         }

src/catalog.rs

@@ -641,7 +641,7 @@ pub(crate) fn validate_catalog_relative_profile_module_path(relative: &Path) ->
             matches!(
                 *name,
                 ".workspace" | "resources" | "archive" | "inbox" | "status"
-            ) || name.starts_with(".status.tmp-")
+            ) || name.starts_with(crate::status::TMP_STAGING_PREFIX)
         });
     let reserved_template_subtree = first == "_templates"
         && components.iter().skip(1).any(|name| {
@@ -657,7 +657,7 @@ pub(crate) fn validate_catalog_relative_profile_module_path(relative: &Path) ->
                     | "archive"
                     | "inbox"
                     | "status"
-            ) || name.starts_with(".status.tmp-")
+            ) || name.starts_with(crate::status::TMP_STAGING_PREFIX)
         });
     anyhow::ensure!(
         !(reserved_control || reserved_root || reserved_agent_state || reserved_template_subtree),

src/catalog_transaction.rs

@@ -2033,7 +2033,7 @@ fn collect_bundle_files(
                 | ".harness-state.seq"
                 | ".harness-state.lock"
                 | ".harness-context.lock"
-        ) || name_text.starts_with(".status.tmp-");
+        ) || name_text.starts_with(crate::status::TMP_STAGING_PREFIX);
         if first.is_some() && relative_to_bundle.components().count() == 1 && state {
             if source == ProjectionSource::Prepared {
                 anyhow::bail!(
@@ -2244,7 +2244,7 @@ fn reject_state_children(identity_path: &Path) -> Result<()> {
         let name = entry.file_name();
         let name = name.to_str().context("identity path is not UTF-8")?;
         if matches!(name, "resources" | "archive" | "inbox" | "status")
-            || name.starts_with(".status.tmp-")
+            || name.starts_with(crate::status::TMP_STAGING_PREFIX)
         {
             anyhow::bail!(
                 "prepared catalog contains state-plane path: {}",
@@ -2934,7 +2934,7 @@ fn validate_declaration_leaf_path(path: &str) -> Result<()> {
             !matches!(
                 components[3],
                 ".workspace" | "resources" | "archive" | "inbox" | "status"
-            ) && !components[3].starts_with(".status.tmp-"),
+            ) && !components[3].starts_with(crate::status::TMP_STAGING_PREFIX),
             "catalog apply marker contains a workspace or state-plane path"
         );
     }

src/codex_app_server.rs

@@ -2975,6 +2975,15 @@ fn acquire_owner_lock(state_dir: &Path) -> Result<File> {
     Ok(file)
 }
 
+/// Stage-and-rename this runtime's own state files, deliberately NOT through the shared
+/// `fsatomic` primitive.
+///
+/// Two reasons, and neither is the durability: [`secure_dir`] re-establishes `0700` on the state
+/// directory on EVERY write, because this directory holds the Codex socket and its owner lock and
+/// a mode drifting open there is a takeover surface rather than a readability question; and the
+/// bytes are `to_writer_pretty`, because these files are read by humans debugging a live runtime.
+/// The shared primitive owns neither, and giving it a "chmod the parent" mode would hand every
+/// caller a directory-permissions policy it has no business having.
 fn atomic_json(path: &Path, value: &impl Serialize) -> Result<()> {
     let parent = path.parent().context("state file has no parent")?;
     secure_dir(parent)?;

src/context.rs

@@ -181,20 +181,19 @@ fn trim_trailing_period(s: &str) -> &str {
     s.strip_suffix('.').unwrap_or(s)
 }
 
-/// Atomic write: tmp sibling + rename.
+/// Atomic write: staged sibling + rename.
+///
+/// The staging name used to end in `now_ms()`, so two writers in the same millisecond shared one
+/// staging path and the second truncated the first's staged bytes before renaming it. The shared
+/// primitive names the sibling with a counter and creates it exclusively, which turns that race
+/// into an impossible `AlreadyExists`.
 fn write_atomic(path: &Path, content: &str) -> anyhow::Result<()> {
-    let dir = path.parent().unwrap_or(Path::new("."));
-    fs::create_dir_all(dir)?;
-    let tmp = dir.join(format!(
-        ".ctx.tmp-{}-{}",
-        std::process::id(),
-        message::now_ms()
-    ));
-    fs::write(&tmp, content)?;
-    if let Err(e) = fs::rename(&tmp, path) {
-        let _ = fs::remove_file(&tmp);
-        return Err(e.into());
-    }
+    crate::fsatomic::replace(
+        path,
+        content.as_bytes(),
+        crate::fsatomic::Staging::new(".ctx"),
+        crate::fsatomic::Durability::Rename,
+    )?;
     Ok(())
 }
 
@@ -222,6 +221,37 @@ fn iso_utc_now() -> String {
 mod tests {
     use super::*;
 
+    /// [`write_atomic`]'s publication contract: the target ends up carrying the complete new
+    /// bytes, no staged sibling survives a successful write, and the record is owner-only.
+    ///
+    /// The mode is the deliberate change of the fold onto `fsatomic` — this record used to be
+    /// published at whatever an ordinary write produces (`0644` under the fleet's umask), and the
+    /// shared primitive stages exclusively at `0600`. An agent's context is its own working
+    /// notes; nothing but st2 and that agent has ever read it.
+    #[test]
+    fn a_context_write_replaces_the_target_and_leaves_no_staged_sibling() {
+        use std::os::unix::fs::PermissionsExt as _;
+
+        let tmp = tempfile::tempdir().unwrap();
+        let dir = context_dir(tmp.path());
+        let path = now_file(&dir);
+        write_atomic(&path, "first\n").unwrap();
+        write_atomic(&path, "second\n").unwrap();
+        assert_eq!(fs::read_to_string(&path).unwrap(), "second\n");
+        assert_eq!(
+            fs::metadata(&path).unwrap().permissions().mode() & 0o777,
+            0o600,
+            "the context record is published owner-only"
+        );
+
+        let staged = fs::read_dir(&dir)
+            .unwrap()
+            .map(|entry| entry.unwrap().file_name().to_string_lossy().into_owned())
+            .filter(|name| name.starts_with(".ctx"))
+            .collect::<Vec<_>>();
+        assert!(staged.is_empty(), "staging residue left behind: {staged:?}");
+    }
+
     #[test]
     fn missing_context_reads_empty() {
         let tmp = tempfile::tempdir().unwrap();

src/delivery_ledger.rs

@@ -12,7 +12,6 @@
 //! The translation itself lives outside this module, behind the one seam in [`Ledger::open`].
 
 use std::fs;
-use std::io::Write as _;
 use std::path::{Path, PathBuf};
 
 use anyhow::{Context as _, Result};
@@ -608,37 +607,18 @@ pub(crate) fn asserted_entries(path: &Path) -> Result<usize> {
 
 /// Durable replacement: file bytes reach disk before rename, then the directory entry is synced.
 ///
-/// The temp file is created exclusively at `0600` under a name unique to this process and write,
-/// so a stale or adversarial path cannot be followed or truncated and two writes cannot collide.
+/// The directory sync is now STRICT — a parent that cannot be opened for it fails
+/// [`Ledger::persist`], where it used to be swallowed. A ledger whose directory entry may not
+/// survive a crash is exactly the state the ledger exists to prevent being invisible, and a
+/// failure edge nothing can observe is a guarantee nothing can review.
 fn atomic_json(path: &Path, value: &impl Serialize) -> Result<()> {
-    use std::os::unix::fs::OpenOptionsExt as _;
-    use std::sync::atomic::{AtomicU64, Ordering};
-
-    static WRITE: AtomicU64 = AtomicU64::new(0);
-
     let bytes = serde_json::to_vec(value)?;
-    let parent = path.parent().context("ledger file has no parent")?;
-    fs::create_dir_all(parent)?;
-    let temp = parent.join(format!(
-        ".delivery-ledger.{}.{}.tmp",
-        std::process::id(),
-        WRITE.fetch_add(1, Ordering::Relaxed)
-    ));
-    let mut file = fs::OpenOptions::new()
-        .write(true)
-        .create_new(true)
-        .mode(0o600)
-        .open(&temp)?;
-    file.write_all(&bytes)?;
-    file.sync_all()?;
-    drop(file);
-    if let Err(error) = fs::rename(&temp, path) {
-        let _ = fs::remove_file(&temp);
-        return Err(error.into());
-    }
-    if let Ok(dir) = fs::File::open(parent) {
-        let _ = dir.sync_all();
-    }
+    crate::fsatomic::replace(
+        path,
+        &bytes,
+        crate::fsatomic::Staging::new(".delivery-ledger"),
+        crate::fsatomic::Durability::FsyncFileAndDir,
+    )?;
     Ok(())
 }
 
@@ -786,11 +766,54 @@ mod tests {
         let residue = fs::read_dir(tmp.path())
             .unwrap()
             .map(|entry| entry.unwrap().file_name().to_string_lossy().into_owned())
-            .filter(|name| name.ends_with(".tmp"))
+            // By staging-name PREFIX, not by a `.tmp` suffix: the suffix is this helper's own
+            // spelling, and a filter that only matches its current spelling stops testing the
+            // moment the spelling changes.
+            .filter(|name| name.starts_with(".delivery-ledger"))
             .collect::<Vec<_>>();
         assert!(residue.is_empty(), "temp residue left behind: {residue:?}");
     }
 
+    /// The directory sync is strict since the fold onto `fsatomic`: a parent that cannot be
+    /// opened for it fails the publication, where it used to be swallowed. This is the deliberate
+    /// behaviour change of that fold on this caller — `Ledger::persist` can now fail on an edge it
+    /// previously reported success for.
+    ///
+    /// Real only for a non-root uid; the hermetic gate runs as the sandbox's unprivileged build
+    /// user, and a local root run skips the edge instead of asserting what root cannot observe.
+    #[test]
+    fn a_directory_that_cannot_be_synced_fails_the_publication() {
+        use std::os::unix::fs::PermissionsExt as _;
+
+        if unsafe { libc::geteuid() } == 0 {
+            return;
+        }
+        let tmp = tempfile::tempdir().unwrap();
+        let dir = tmp.path().join("agent");
+        fs::create_dir_all(&dir).unwrap();
+        let path = dir.join(LEDGER_FILE);
+        let record = Record {
+            schema: LEDGER_SCHEMA.to_owned(),
+            harness: "codex".to_owned(),
+            agent: "h.worker".to_owned(),
+            runtime_id: "runtime".to_owned(),
+            entries: Vec::new(),
+        };
+
+        // Write and traverse, but not read: staging and renaming still work, opening the
+        // directory to sync it does not.
+        fs::set_permissions(&dir, fs::Permissions::from_mode(0o300)).unwrap();
+        let published = atomic_json(&path, &record);
+        fs::set_permissions(&dir, fs::Permissions::from_mode(0o700)).unwrap();
+        assert!(
+            published.is_err(),
+            "the ledger's directory sync is strict: {published:?}"
+        );
+        // The bytes did land — the rename happens before the sync — so the failure is a report
+        // about durability, not about the record's contents.
+        assert!(path.exists(), "the record still landed");
+    }
+
     #[test]
     fn positive_evidence_never_downgrades() {
         let tmp = tempfile::tempdir().unwrap();