fix(resource): preserve publish ABI and refresh fence

agent-identity: dev3.direct.omp.2cshu64q
agent-persona: generalist
agent-supervisor: unavailable
agent-tool: OMP
agent-tool-version: 18.0.9
agent-runtime: OMP 18.0.9
tooling-profile: dotfiles@b607597

Author: schickling-assistant

crates/st2-resource-protocol/src/lib.rs

@@ -644,12 +644,11 @@ impl<'de> Deserialize<'de> for ObservationResult {
     }
 }
 
-#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[derive(Debug, Clone, PartialEq, Serialize)]
 #[serde(
     tag = "type",
     rename_all = "camelCase",
-    rename_all_fields = "camelCase",
-    deny_unknown_fields
+    rename_all_fields = "camelCase"
 )]
 pub enum RuntimeMessage {
     Publish {
@@ -678,6 +677,91 @@ pub enum RuntimeMessage {
     },
 }
 
+#[derive(Deserialize)]
+#[serde(
+    tag = "type",
+    rename_all = "camelCase",
+    rename_all_fields = "camelCase",
+    deny_unknown_fields
+)]
+enum RuntimeMessageWire {
+    Publish {
+        owner: RuntimeOwner,
+        binding_id: BindingId,
+        registration: RegistrationToken,
+        /// ABI-3 runtimes may still send this retired producer timestamp. The host never trusted it,
+        /// so decode it only to preserve that ABI and discard it at this boundary.
+        observed_at: Option<String>,
+        #[serde(flatten)]
+        publication: Publication,
+    },
+    Health {
+        owner: RuntimeOwner,
+        binding_id: Option<BindingId>,
+        registration: Option<RegistrationToken>,
+        state: RuntimeHealthState,
+        detail: Option<String>,
+    },
+    ObservationResult {
+        owner: RuntimeOwner,
+        binding_id: BindingId,
+        registration: RegistrationToken,
+        demand_watermark: u64,
+        result: ObservationResult,
+    },
+}
+
+impl<'de> Deserialize<'de> for RuntimeMessage {
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    where
+        D: Deserializer<'de>,
+    {
+        Ok(match RuntimeMessageWire::deserialize(deserializer)? {
+            RuntimeMessageWire::Publish {
+                owner,
+                binding_id,
+                registration,
+                observed_at,
+                publication,
+            } => {
+                drop(observed_at);
+                Self::Publish {
+                    owner,
+                    binding_id,
+                    registration,
+                    publication,
+                }
+            },
+            RuntimeMessageWire::Health {
+                owner,
+                binding_id,
+                registration,
+                state,
+                detail,
+            } => Self::Health {
+                owner,
+                binding_id,
+                registration,
+                state,
+                detail,
+            },
+            RuntimeMessageWire::ObservationResult {
+                owner,
+                binding_id,
+                registration,
+                demand_watermark,
+                result,
+            } => Self::ObservationResult {
+                owner,
+                binding_id,
+                registration,
+                demand_watermark,
+                result,
+            },
+        })
+    }
+}
+
 #[derive(Debug)]
 pub enum ProtocolError {
     MissingNewline,
@@ -968,6 +1052,26 @@ mod tests {
         );
     }
 
+    #[test]
+    fn abi_3_publish_decodes_with_or_without_deprecated_observed_at_only() {
+        let without_observed_at = b"{\"type\":\"publish\",\"owner\":{\"incarnation\":\"incarnation\",\"claim\":\"claim\"},\"bindingId\":\"binding\",\"registration\":\"registration\",\"schemaId\":\"schema.v1\",\"mediaType\":\"application/json\",\"bytes\":\"b25lIGJ5dGU=\",\"topics\":[\"selected\"]}\n";
+        let with_observed_at = b"{\"type\":\"publish\",\"owner\":{\"incarnation\":\"incarnation\",\"claim\":\"claim\"},\"bindingId\":\"binding\",\"registration\":\"registration\",\"schemaId\":\"schema.v1\",\"mediaType\":\"application/json\",\"bytes\":\"b25lIGJ5dGU=\",\"topics\":[\"selected\"],\"observedAt\":\"2026-08-30T00:00:00Z\"}\n";
+        let unknown_field = b"{\"type\":\"publish\",\"owner\":{\"incarnation\":\"incarnation\",\"claim\":\"claim\"},\"bindingId\":\"binding\",\"registration\":\"registration\",\"schemaId\":\"schema.v1\",\"mediaType\":\"application/json\",\"bytes\":\"b25lIGJ5dGU=\",\"topics\":[\"selected\"],\"extra\":true}\n";
+
+        assert_eq!(
+            decode_runtime_line(without_observed_at).unwrap(),
+            publish(b"one byte")
+        );
+        assert_eq!(
+            decode_runtime_line(with_observed_at).unwrap(),
+            publish(b"one byte")
+        );
+        assert!(matches!(
+            decode_runtime_line(unknown_field),
+            Err(ProtocolError::Json(_))
+        ));
+    }
+
     #[test]
     fn observation_results_have_one_atomic_tagged_wire_shape() {
         let unchanged = RuntimeMessage::ObservationResult {
@@ -1104,11 +1208,7 @@ mod tests {
             decode_runtime_line(unknown_publication_field),
             Err(ProtocolError::Json(_))
         ));
-        let obsolete_observed_at = b"{\"type\":\"publish\",\"owner\":{\"incarnation\":\"i\",\"claim\":\"c\"},\"bindingId\":\"b\",\"registration\":\"r\",\"schemaId\":\"s\",\"mediaType\":\"m\",\"bytes\":\"\",\"topics\":[],\"observedAt\":\"2026-08-30T00:00:00Z\"}\n";
-        assert!(matches!(
-            decode_runtime_line(obsolete_observed_at),
-            Err(ProtocolError::Json(_))
-        ));
+
         let obsolete_settlement = b"{\"type\":\"observationSettled\",\"owner\":{\"incarnation\":\"i\",\"claim\":\"c\"},\"bindingId\":\"b\",\"registration\":\"r\",\"demandWatermark\":1,\"outcome\":\"unchanged\"}\n";
         assert!(matches!(
             decode_runtime_line(obsolete_settlement),

src/main.rs

@@ -3379,6 +3379,19 @@ fn resource_bindings(
     selector: &str,
     host: &str,
 ) -> Result<(String, Vec<st2::Resource>)> {
+    with_resource_bindings_snapshot(root, selector, host, |identity, bindings| {
+        Ok((identity, bindings))
+    })
+}
+
+/// Resolve one binding projection and let the caller finish consuming that exact catalog snapshot
+/// before its shared authoring fence is released.
+fn with_resource_bindings_snapshot<T>(
+    root: &Path,
+    selector: &str,
+    host: &str,
+    consume: impl FnOnce(String, Vec<st2::Resource>) -> Result<T>,
+) -> Result<T> {
     let _catalog_lock = st2::CatalogLock::shared(root)
         .context("acquire shared catalog-authoring lock for Resource bindings")?;
     let found = st2::discover_strict(root);
@@ -3403,9 +3416,9 @@ fn resource_bindings(
     } else {
         exact
     };
-    match matches.as_slice() {
+    let (identity, bindings) = match matches.as_slice() {
         [] => anyhow::bail!("no agent '{selector}' found in catalog {}", root.display()),
-        [spec] => Ok((spec.bus_id(host), spec.resources.clone())),
+        [spec] => (spec.bus_id(host), spec.resources.clone()),
         many => {
             let mut candidates = many
                 .iter()
@@ -3417,9 +3430,27 @@ fn resource_bindings(
                 candidates.join(", ")
             )
         }
+    };
+    consume(identity, bindings)
+}
+
+#[cfg(debug_assertions)]
+fn resource_refresh_snapshot_checkpoint() {
+    let (Ok(ready), Ok(release)) = (
+        std::env::var("ST2_TEST_RESOURCE_REFRESH_SNAPSHOT_READY"),
+        std::env::var("ST2_TEST_RESOURCE_REFRESH_SNAPSHOT_RELEASE"),
+    ) else {
+        return;
+    };
+    let _ = std::fs::write(ready, b"ready");
+    while !Path::new(&release).exists() {
+        std::thread::yield_now();
     }
 }
 
+#[cfg(not(debug_assertions))]
+fn resource_refresh_snapshot_checkpoint() {}
+
 fn resource_cmd(cmd: ResourceCmd) -> Result<()> {
     match cmd {
         ResourceCmd::Ls {
@@ -3502,17 +3533,24 @@ fn resource_cmd(cmd: ResourceCmd) -> Result<()> {
                     (selector, first)
                 }
             };
-            let (identity, bindings) = resource_bindings(&root, &selector, &host)?;
-            bindings
-                .iter()
-                .find(|binding| binding.name() == name)
-                .with_context(|| format!("no resource binding '{name}' declared by {identity}"))?;
-            let request = st2::resource_observe::ObserveRequest::new(
-                identity.clone(),
-                name.clone(),
-                st2::resource_observe::catalog_generation(&root)?,
-                None,
-            )?;
+            let (identity, request) =
+                with_resource_bindings_snapshot(&root, &selector, &host, |identity, bindings| {
+                    bindings
+                        .iter()
+                        .find(|binding| binding.name() == name)
+                        .with_context(|| {
+                            format!("no resource binding '{name}' declared by {identity}")
+                        })?;
+                    resource_refresh_snapshot_checkpoint();
+                    let generation = st2::resource_observe::catalog_generation(&root)?;
+                    let request = st2::resource_observe::ObserveRequest::new(
+                        identity.clone(),
+                        name.clone(),
+                        generation,
+                        None,
+                    )?;
+                    Ok((identity, request))
+                })?;
             let request_id = request.request_id.clone();
             let client = st2::resource_observe::submit_request(&root, &host, &request)?;
             let waited = client.wait_for_terminal(Duration::from_secs(wait))?;

tests/agent_resource.rs

@@ -507,6 +507,135 @@ fn a_binding_with_a_trailing_line_comment_is_removable_and_updatable() {
     );
 }
 
+#[test]
+fn refresh_binding_validation_and_generation_share_one_catalog_snapshot() {
+    let _guard = OBSERVE_ENV_LOCK
+        .lock()
+        .unwrap_or_else(std::sync::PoisonError::into_inner);
+    let temporary = tempfile::tempdir().unwrap();
+    let root = temporary.path().join("catalog");
+    let state = temporary.path().join("state");
+    write(
+        &root,
+        "h/worker/agent.kdl",
+        &declaration(
+            "worker",
+            "catalog",
+            "  resource \"work\" reason=\"Old binding.\" uri=\"https://example.test/old\"\n",
+        ),
+    );
+    ok(
+        &root,
+        &[
+            "resource",
+            "add",
+            "anchor",
+            "--agent",
+            "worker",
+            "--uri",
+            "https://example.test/anchor",
+            "--reason",
+            "Initialize the catalog generation.",
+        ],
+    );
+    let old_generation = st2::resource_observe::catalog_generation(&root)
+        .unwrap()
+        .expect("mediated edit initialized catalog generation");
+
+    let previous_state = std::env::var_os("XDG_STATE_HOME");
+    unsafe { std::env::set_var("XDG_STATE_HOME", &state) };
+    st2::event::publish_owner_binding_for_test(&root, "h").unwrap();
+    let scope = st2::park::SupervisorScope::current(&root, "h").unwrap();
+    let request_dir = scope
+        .park_dir()
+        .parent()
+        .unwrap()
+        .join("observe-requests");
+    match previous_state {
+        Some(value) => unsafe { std::env::set_var("XDG_STATE_HOME", value) },
+        None => unsafe { std::env::remove_var("XDG_STATE_HOME") },
+    }
+
+    let snapshot_ready = temporary.path().join("snapshot-ready");
+    let snapshot_release = temporary.path().join("snapshot-release");
+    let replacement_attempt = temporary.path().join("replacement-attempt");
+    let refresh = Command::new(env!("CARGO_BIN_EXE_st2"))
+        .args(["--catalog", root.to_str().unwrap()])
+        .args([
+            "resource", "refresh", "worker", "work", "--wait", "0", "--host", "h",
+        ])
+        .env("XDG_STATE_HOME", &state)
+        .env("ST2_TEST_RESOURCE_REFRESH_SNAPSHOT_READY", &snapshot_ready)
+        .env(
+            "ST2_TEST_RESOURCE_REFRESH_SNAPSHOT_RELEASE",
+            &snapshot_release,
+        )
+        .env_remove("ST_AGENT")
+        .stdout(Stdio::piped())
+        .stderr(Stdio::piped())
+        .spawn()
+        .unwrap();
+    wait_for_file(&snapshot_ready);
+
+    let mut replacement = Command::new(env!("CARGO_BIN_EXE_st2"))
+        .args(["--catalog", root.to_str().unwrap()])
+        .args([
+            "resource",
+            "add",
+            "work",
+            "--agent",
+            "worker",
+            "--uri",
+            "https://example.test/replacement",
+            "--reason",
+            "Replacement binding.",
+        ])
+        .env("ST2_TEST_CATALOG_LOCK_ATTEMPT", &replacement_attempt)
+        .env_remove("ST_AGENT")
+        .stdout(Stdio::piped())
+        .stderr(Stdio::piped())
+        .spawn()
+        .unwrap();
+    wait_for_file(&replacement_attempt);
+    assert!(
+        replacement.try_wait().unwrap().is_none(),
+        "replacement crossed the shared snapshot lock before generation capture"
+    );
+    assert!(
+        spec(&root).contains("https://example.test/old"),
+        "replacement landed while refresh held its snapshot"
+    );
+
+    fs::write(&snapshot_release, b"release").unwrap();
+    let refresh = refresh.wait_with_output().unwrap();
+    assert!(
+        !refresh.status.success(),
+        "zero client wait should leave the request queued"
+    );
+    let replacement = replacement.wait_with_output().unwrap();
+    assert!(
+        replacement.status.success(),
+        "stdout: {}\nstderr: {}",
+        stdout(&replacement),
+        stderr(&replacement)
+    );
+
+    let request = wait_for_refresh_request(&request_dir);
+    assert_eq!(
+        request["expectedCatalogGeneration"],
+        serde_json::json!(old_generation),
+        "request generation must come from the validated old binding snapshot"
+    );
+    let replacement_generation = st2::resource_observe::catalog_generation(&root)
+        .unwrap()
+        .unwrap();
+    assert!(
+        replacement_generation > old_generation,
+        "replacement did not advance catalog generation"
+    );
+    assert!(spec(&root).contains("https://example.test/replacement"));
+}
+
 #[test]
 fn refresh_cli_reports_exact_receipts_and_wait_expiry_keeps_the_request() {
     let _guard = OBSERVE_ENV_LOCK