fix(fs): narrow create_once's duplicate arm and single-source the staging prefixes

Four corrections from independent review of the fold, each a real gap rather than
a wording fix:

1. `create_once` evaluated `written.and_then(|()| hard_link(…))`, so its
   `Err(_) if path.is_file() => Ok(false)` arm also caught failures from the
   staged WRITE. An `ENOSPC`/`EIO` on the staged bytes was then reported as "this
   key was already published" whenever the target happened to exist — bytes that
   were never written are nobody's publication. Only the hardlink may answer that
   question now. The absorbed callers propagated a failed write unconditionally,
   so this restores their contract. No test can force the edge in-process (there
   is no write-fault injection in this tree), so it is review-pinned.
2. `replace` removed the staged file on a REFUSED creation, which meant st2
   unlinked whatever an agent had planted at the staging path instead of merely
   refusing to follow it — and asymmetric with `create_once`, which returns
   before its cleanup. The creation now returns before the cleanup scope in both.
3. Folding the staging name into a module made each caller's prefix a bare string
   argument, and the assertion that pinned `.status.tmp-` went with `tmp_name`.
   The walkers that skip staged files by prefix now read the writer's own const —
   `status::TMP_STAGING_PREFIX` for the six in `catalog`, `catalog_transaction`
   and `agent_publish`, `message::TMP_STAGING_PREFIX` for the four in `message`,
   and `harness_context::is_legacy_staging_name` derives its prefix from the
   `TMP_PREFIX` the writer passes — so a walker can no longer drift from the
   writer. Each module also asserts its const's VALUE, which is the part a shared
   const cannot protect: renaming both sides together would leave every
   already-staged file on the fleet unrecognized.
4. `reserved_message_temporary_symlinks_are_never_followed` predicted staging
   names from one probe of a counter that is now process-global, so a sibling
   test advancing it past the planted window would have turned the test
   intermittent. It now plants, re-probes, and only proceeds once the very next
   name is one it has blocked.

Also corrects the HC-T08 sentence added by the fold: bus message files are
state-plane records too and stay at the writing process's umask, so "like every
other state-plane record" was wrong.

agent-identity: dev3.direct.omp.43sz6ujq
agent-persona: generalist
agent-supervisor: unavailable
agent-tool: OMP
agent-tool-version: 18.1.7
agent-runtime: OMP 18.1.7
tooling-profile: dotfiles@39a19af

Author: schickling-assistant

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

@@ -103,11 +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. The pair is
-  published mode `0600`, like every other state-plane record, so a transport
-  that reads 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 what leaves the wire-cost half of `DQ-C1` unmeasured.
+  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/fsatomic.rs

@@ -95,18 +95,20 @@ pub(crate) fn replace(
 ) -> io::Result<()> {
     let parent = parent_of(path)?;
     let staged = prepare(parent, &staging)?;
-    let landed = (|| -> io::Result<()> {
-        let mut file = create_staging(&staged)?;
+    // A refused creation returns BEFORE the cleanup scope: the thing occupying a
+    // `{prefix}.tmp-{pid}-{counter}` path is not ours, and refusing to follow it must not turn
+    // into unlinking it.
+    let mut file = create_staging(&staged)?;
+    let staged_bytes = (|| -> io::Result<()> {
         file.write_all(bytes)?;
         if durability == Durability::FsyncFileAndDir {
             file.sync_all()?;
         }
-        drop(file);
-        fs::rename(&staged, path)
+        Ok(())
     })();
-    if let Err(error) = landed {
-        // Best-effort: the staging name is unique per write, so a leftover is inert rather than a
-        // path a later write could collide with.
+    drop(file);
+    if let Err(error) = staged_bytes.and_then(|()| fs::rename(&staged, path)) {
+        // Best-effort, and only for the file this call created.
         let _ = fs::remove_file(&staged);
         return Err(error);
     }
@@ -127,15 +129,18 @@ pub(crate) fn create_once(path: &Path, bytes: &[u8], staging: Staging<'_>) -> io
     let parent = parent_of(path)?;
     let staged = prepare(parent, &staging)?;
     let mut file = create_staging(&staged)?;
-    let written = file.write_all(bytes);
+    let staged_bytes = file.write_all(bytes);
     drop(file);
-    let created = match written.and_then(|()| fs::hard_link(&staged, path)) {
+    // A failed staged write is never a duplicate: `Ok(false)` claims somebody else published
+    // these bytes, and bytes that were never written are nobody's publication. Only the hardlink
+    // may answer that question.
+    let created = staged_bytes.and_then(|()| match fs::hard_link(&staged, path) {
         Ok(()) => Ok(true),
         // `hard_link` reports `AlreadyExists` for a taken name, but a target that is already a
         // regular file is the same answer whatever the error says.
         Err(_) if path.is_file() => Ok(false),
         Err(error) => Err(error),
-    };
+    });
     let _ = fs::remove_file(&staged);
     created
 }

src/harness_context.rs

@@ -392,8 +392,14 @@ pub(crate) fn is_legacy_harness_context_staging_file(
     Ok(true)
 }
 
+/// The matcher and the writer must agree on the prefix, so both read [`TMP_PREFIX`]: this name is
+/// the one INVARIANTS row 29 says current-catalog identity walkers overlook, and a writer that
+/// drifted from this matcher would leave a staged file the walkers no longer recognize.
 fn is_legacy_staging_name(name: &str) -> bool {
-    let Some(suffix) = name.strip_prefix(".harness-context.tmp-") else {
+    let Some(suffix) = name
+        .strip_prefix(TMP_PREFIX)
+        .and_then(|rest| rest.strip_prefix(".tmp-"))
+    else {
         return false;
     };
     let Some((pid, counter)) = suffix.split_once('-') else {

src/message.rs

@@ -399,10 +399,14 @@ pub fn materialize_message_once(
     result
 }
 
-/// The staging-name prefix for everything this module stages. Four sent-record and inbox walkers
-/// match `.message.tmp-` by prefix, so the name is a contract with them, not a local detail.
+/// The staging-name prefix for everything this module stages.
 const TMP_PREFIX: &str = ".message";
 
+/// The full staged-name prefix four inbox and sent-record walkers skip by prefix, spelled ONCE so
+/// a walker cannot drift from the writer. Tied to [`TMP_PREFIX`] by
+/// `the_staging_prefix_is_the_one_the_inbox_walkers_skip`.
+const TMP_STAGING_PREFIX: &str = ".message.tmp-";
+
 fn tmp_name() -> String {
     crate::fsatomic::staging_name(TMP_PREFIX)
 }
@@ -711,7 +715,7 @@ fn read_sent_records(directory: &Path) -> anyhow::Result<Vec<SentRecord>> {
         let Some(name) = name.to_str() else {
             anyhow::bail!("sent record filename is not UTF-8");
         };
-        if name.starts_with(".message.tmp-") {
+        if name.starts_with(TMP_STAGING_PREFIX) {
             continue;
         }
         anyhow::ensure!(name.ends_with(".json"), "unexpected sent record entry");
@@ -744,7 +748,7 @@ fn read_pending_records(directory: &Path) -> anyhow::Result<Vec<SentRecord>> {
         let Some(name) = name.to_str() else {
             anyhow::bail!("pending sent record filename is not UTF-8");
         };
-        if name.starts_with(".message.tmp-") {
+        if name.starts_with(TMP_STAGING_PREFIX) {
             continue;
         }
         let digest = name
@@ -804,7 +808,7 @@ fn read_sent_commits(directory: &Path) -> anyhow::Result<BTreeMap<String, SentCo
         let Some(name) = name.to_str() else {
             anyhow::bail!("sent commit filename is not UTF-8");
         };
-        if name.starts_with(".message.tmp-") {
+        if name.starts_with(TMP_STAGING_PREFIX) {
             continue;
         }
         let digest = name
@@ -845,7 +849,7 @@ fn read_sent_keys(directory: &Path) -> anyhow::Result<BTreeMap<String, SentKey>>
         let Some(name) = name.to_str() else {
             anyhow::bail!("sent key filename is not UTF-8");
         };
-        if name.starts_with(".message.tmp-") {
+        if name.starts_with(TMP_STAGING_PREFIX) {
             continue;
         }
         let digest = name
@@ -2561,6 +2565,18 @@ fn remove_inbox_duplicate(source: &Path, filename: &str) -> anyhow::Result<()> {
 mod tests {
     use super::*;
 
+    /// The staged name is a contract with four walkers that skip it by prefix (`inbox_dir` and the
+    /// three sent-record scans), all of which now match [`TMP_STAGING_PREFIX`] so none can drift
+    /// from the writer. What a shared const cannot catch is both sides being renamed together,
+    /// which would leave every already-staged file on the fleet unrecognized — hence the value
+    /// assertion.
+    #[test]
+    fn the_staging_prefix_is_the_one_the_inbox_walkers_skip() {
+        assert_eq!(TMP_PREFIX, ".message");
+        assert_eq!(TMP_STAGING_PREFIX, format!("{TMP_PREFIX}.tmp-"));
+        assert!(!is_message_filename(&tmp_name()), "a staged name must never look like a message");
+    }
+
     /// [`atomic_create_file`]'s create-once contract. It is a hardlink, not a rename, and that is
     /// the whole point: the first publication wins, a second reports `false` instead of replacing
     /// the winner's bytes, and neither leaves a staged sibling behind for the four `.message.tmp-`
@@ -2588,7 +2604,7 @@ mod tests {
         let residue = fs::read_dir(path.parent().unwrap())
             .unwrap()
             .map(|entry| entry.unwrap().file_name().to_string_lossy().into_owned())
-            .filter(|name| name.starts_with(".message.tmp-"))
+            .filter(|name| name.starts_with(TMP_STAGING_PREFIX))
             .collect::<Vec<_>>();
         assert!(residue.is_empty(), "staging residue left behind: {residue:?}");
     }
@@ -2606,7 +2622,7 @@ mod tests {
         let residue = fs::read_dir(tmp.path())
             .unwrap()
             .map(|entry| entry.unwrap().file_name().to_string_lossy().into_owned())
-            .filter(|name| name.starts_with(".message.tmp-"))
+            .filter(|name| name.starts_with(TMP_STAGING_PREFIX))
             .collect::<Vec<_>>();
         assert!(residue.is_empty(), "staging residue left behind: {residue:?}");
     }
@@ -2764,23 +2780,29 @@ mod tests {
         fs::create_dir_all(&inbox).unwrap();
         let victim = tmp.path().join("victim");
         fs::write(&victim, "must remain unchanged").unwrap();
-        // The counter now lives in `fsatomic`, so the next names are predicted from a probe
-        // rather than read off a module-local static: one call consumes `start`, so the writes
-        // this test blocks are the 4096 after it.
-        let probe = tmp_name();
-        let start = probe
-            .rsplit_once('-')
-            .and_then(|(_, counter)| counter.parse::<u64>().ok())
-            .expect("the staging grammar ends in the counter");
-        for counter in start + 1..start + 1 + 4096 {
-            symlink(
-                &victim,
-                inbox.join(format!(
-                    "{TMP_PREFIX}.tmp-{}-{counter}",
-                    std::process::id()
-                )),
-            )
-            .unwrap();
+        // The counter lives in `fsatomic` and is shared with every other staging site, so a
+        // sibling test running in parallel advances it too. Plant a window, then PROBE again: the
+        // loop only exits once the very next name is one this test has already blocked, so the
+        // assertion below cannot become "the call happened to pick a free name".
+        let counter_of = |name: &str| {
+            name.rsplit_once('-')
+                .and_then(|(_, counter)| counter.parse::<u64>().ok())
+                .expect("the staging grammar ends in the counter")
+        };
+        let mut planted_through = 0;
+        loop {
+            let probe = counter_of(&tmp_name());
+            if probe < planted_through {
+                break;
+            }
+            for counter in probe + 1..=probe + 512 {
+                symlink(
+                    &victim,
+                    inbox.join(format!("{TMP_PREFIX}.tmp-{}-{counter}", std::process::id())),
+                )
+                .unwrap();
+            }
+            planted_through = probe + 512;
         }
 
         let error = materialize_message_once(&inbox, "1784649988123-symlnk.md", "must not escape")

src/status.rs

@@ -284,6 +284,14 @@ fn write_record(path: &Path, state: State, written_at_ms: u64) -> anyhow::Result
     )
 }
 
+/// The staging-name prefix this module hands [`crate::fsatomic`].
+const TMP_PREFIX: &str = ".status";
+
+/// The full staged-name prefix six catalog and publication walkers match, spelled ONCE here so a
+/// walker can never drift from the writer. Tied to [`TMP_PREFIX`] by
+/// `the_staging_prefix_is_the_one_the_catalog_walkers_match`.
+pub(crate) const TMP_STAGING_PREFIX: &str = ".status.tmp-";
+
 /// Atomic write: a staged sibling + rename, so a concurrent reader sees either the old bytes or
 /// the new bytes, never a partial file.
 ///
@@ -294,7 +302,7 @@ fn write_atomic(path: &Path, content: &str) -> anyhow::Result<()> {
     crate::fsatomic::replace(
         path,
         content.as_bytes(),
-        crate::fsatomic::Staging::new(".status"),
+        crate::fsatomic::Staging::new(TMP_PREFIX),
         crate::fsatomic::Durability::Rename,
     )?;
     Ok(())
@@ -305,6 +313,18 @@ mod tests {
     use super::*;
     use std::time::{Duration as Dur, SystemTime};
 
+    /// The staged name is a contract with six walkers that skip it by prefix, in three other
+    /// modules: `catalog.rs`, `catalog_transaction.rs` and `agent_publish.rs` all match
+    /// [`TMP_STAGING_PREFIX`], so a changed writer prefix cannot desynchronize them. What a const
+    /// cannot catch is the prefix being renamed on BOTH sides at once — a staged status file would
+    /// then still be skipped locally, but the fleet's existing records and any other reader of the
+    /// old name would not. That is what this assertion is for.
+    #[test]
+    fn the_staging_prefix_is_the_one_the_catalog_walkers_match() {
+        assert_eq!(TMP_PREFIX, ".status");
+        assert_eq!(TMP_STAGING_PREFIX, format!("{TMP_PREFIX}.tmp-"));
+    }
+
     /// [`write_atomic`]'s publication contract: the target ends up carrying the complete new
     /// bytes, no staged sibling survives, and the record is owner-only.
     ///