fix(resync): retry publication and tighten carriers

Author: schickling

crates/agent-spec/src/spec.rs

@@ -240,8 +240,18 @@ struct ResourceDescriptor {
 impl Resource {
     /// Construct a descriptor after enforcing the same invariants as catalog parsing.
     pub fn new(name: String, uri: String, reason: String) -> Result<Self, String> {
-        if name.is_empty() {
-            return Err("resource binding name cannot be empty".into());
+        if name.is_empty()
+            || name.len() > 200
+            || name.trim() != name
+            || name.chars().any(char::is_control)
+        {
+            return Err(
+                "resource binding name must be 1..=200 bytes without surrounding whitespace or controls"
+                    .into(),
+            );
+        }
+        if name == "declaration" {
+            return Err("resource binding name 'declaration' is reserved by resync".into());
         }
         validate_resource_uri(&uri).map_err(|reason| {
             format!("resource binding '{name}' `uri` must be an exact absolute URI or a catalog-relative path: {reason}")
@@ -752,22 +762,16 @@ fn validate_resource_explanation(name: &str, field: &str, value: &str) -> Result
 /// scheme at all, resolved by the consumer against the declaration directory. Relative carriers
 /// are an st2 extension pending canonical Agent Spec adoption (see 06-resync).
 fn validate_resource_uri(uri: &str) -> Result<(), &'static str> {
-    match uri.split_once(':') {
-        Some((scheme, _))
-            if !scheme.contains('/')
-                && scheme
-                    .chars()
-                    .all(|c| c.is_ascii_alphanumeric() || matches!(c, '+' | '-' | '.')) =>
-        {
-            validate_absolute_uri(uri)
-        }
-        _ => {
-            if uri.is_empty() || uri.starts_with('/') || uri.contains("..") {
-                return Err("catalog-relative uri must be a non-empty relative path without `..`");
-            }
-            Ok(())
+    if let Some(colon) = uri.find(':') {
+        let first_separator = uri.find(|character| matches!(character, '/' | '\\'));
+        if first_separator.is_none_or(|separator| colon < separator) {
+            return validate_absolute_uri(uri);
         }
     }
+    if uri.is_empty() || uri.starts_with('/') || uri.contains("..") {
+        return Err("catalog-relative uri must be a non-empty relative path without `..`");
+    }
+    Ok(())
 }
 
 fn validate_absolute_uri(uri: &str) -> Result<(), &'static str> {

crates/agent-spec/tests/discovery.rs

@@ -1134,6 +1134,30 @@ fn resource_explanation_byte_bounds_are_enforced() {
     );
 }
 
+#[test]
+fn resource_binding_names_and_scheme_candidates_fail_loudly() {
+    for name in [
+        " declaration".to_owned(),
+        "declaration".to_owned(),
+        "line\nbreak".to_owned(),
+        "x".repeat(201),
+    ] {
+        assert!(
+            Resource::new(name.clone(), "issue://one".into(), "Task.".into()).is_err(),
+            "invalid binding name was accepted: {name:?}"
+        );
+    }
+    assert!(
+        Resource::new(
+            "work".into(),
+            "_github://org/repo".into(),
+            "Task.".into(),
+        )
+        .is_err(),
+        "a malformed scheme prefix must not become a catalog-relative path"
+    );
+}
+
 #[test]
 fn malformed_resource_envelopes_are_rejected_without_defining_downstream_types() {
     let tmp = tempfile::tempdir().unwrap();

docs/vrs/06-resync/spec.md

@@ -71,6 +71,8 @@ gets its own event so per-binding supersession stays meaningful.
   bus id, label, and class so every subscriber retains its digest and pending
   dirty state even when multiple bindings share one path; only new
   subscriptions seed silently.
+- A previously blind path is digest-diffed before its recovered parent directory is added to the
+  registered-watch set, covering changes made during the blind interval.
 - Installation failure is diagnosed once and degrades to timer-based digest
   polling over the watch set (bounded by the number of bindings), never to
   silence about the mechanism.
@@ -82,11 +84,15 @@ agent without a declaration check; every other rule (host ownership,
 desired-state running, owner-binding validation, ring transaction) is the
 implemented [`STREAM-R03..R05`](../04-stream/spec.md) path untouched.
 Declaring a stream named `resync` in an Agent Spec is a validation error:
-the reservation must not be shadowable.
-
-Digest state lives with the supervisor process (seeded at start, updated on
-each observed change); the durable dedup horizon remains the stream receipt
-ring, exactly as for external producers.
+the reservation must not be shadowable. Resource binding names use the event-key
+grammar (1..=200 bytes, no surrounding whitespace or controls), and
+`declaration` is reserved for the synthetic declaration carrier so supersession
+keys cannot collide.
+
+Digest state lives with the supervisor process (seeded at start) and advances
+only after the unchanged ingress returns a successful receipt. Failed
+publication retains the old digest and dirty state and schedules the identical
+transition for retry; the durable dedup horizon remains the stream receipt ring.
 
 ## What this does not do
 

src/resync.rs

@@ -325,10 +325,10 @@ impl Worker {
             self.diff_emit(&mut next);
         }
         self.carriers = next;
-        self.refresh_watches();
-        // Directories that could not be registered stay blind to events; their carriers fall
-        // back to digest polling at refresh cadence (`RESYNC-R01`).
+        // Diff paths that were blind before registering newly recovered parents; otherwise the
+        // new watch suppresses polling of mutations that happened during the blind interval.
         self.poll_unwatched();
+        self.refresh_watches();
     }
 
     /// Digest-diff every entry in `next`, emitting observed transitions. Seeding stays silent
@@ -348,17 +348,18 @@ impl Worker {
                 if entry.digest.as_deref() == Some(new_digest.as_str()) {
                     continue;
                 }
-                let old = entry.digest.take();
-                emit_resync(
+                let old = entry.digest.as_deref();
+                if emit_resync(
                     &root,
                     &this_host,
                     &entry.bus_id,
                     &entry.label,
                     &path,
-                    old.as_deref(),
+                    old,
                     &new_digest,
-                );
-                entry.digest = Some(new_digest);
+                ) {
+                    entry.digest = Some(new_digest);
+                }
             }
         }
     }
@@ -487,33 +488,45 @@ impl Worker {
         }
     }
 
-    /// Flush every subscriber of one path: clear dirty flags, diff digests, emit transitions.
+    /// Flush every subscriber of one path: diff digests, emit transitions, and retain failed
+    /// publications for retry with the same event identity.
     fn flush_path(&mut self, path: &Path) {
         let Some(entries) = self.carriers.get_mut(path) else {
             return;
         };
+        let mut retries = Vec::new();
         for entry in entries.iter_mut() {
             entry.dirty = false;
             let Some(new_digest) = read_digest(path) else {
                 // Unreadable right now (deleted mid-window): stay quiet and keep the previous
-                // digest, so a later reappearance still counts as a change. The next mutation
-                // re-dirties.
+                // digest, so a later reappearance still counts as a change.
                 continue;
             };
             if entry.digest.as_deref() == Some(new_digest.as_str()) {
                 continue;
             }
-            let old = entry.digest.take();
-            emit_resync(
+            if emit_resync(
                 &self.root,
                 &self.this_host,
                 &entry.bus_id,
                 &entry.label,
                 path,
-                old.as_deref(),
+                entry.digest.as_deref(),
                 &new_digest,
-            );
-            entry.digest = Some(new_digest);
+            ) {
+                entry.digest = Some(new_digest);
+            } else {
+                entry.dirty = true;
+                retries.push(entry.class);
+            }
+        }
+        let now = Instant::now();
+        for class in retries {
+            let deadline = now + class.window();
+            self.deadlines
+                .entry(class)
+                .and_modify(|existing| *existing = (*existing).min(deadline))
+                .or_insert(deadline);
         }
     }
 }
@@ -533,11 +546,11 @@ fn emit_resync(
     path: &Path,
     old: Option<&str>,
     new_digest: &str,
-) {
+) -> bool {
     let subject = format!("resource {label} changed");
     let body = render_body(label, path, old, new_digest);
     let event_id = transition_identity(&body);
-    let receipt = crate::event::emit(
+    match crate::event::emit(
         root,
         this_host,
         bus_id,
@@ -547,12 +560,15 @@ fn emit_resync(
         Some(subject.as_str()),
         &body,
         true,
-    );
-    if let Err(error) = receipt {
-        eprintln!(
-            "st2: resync emit for '{}' failed: {error:#}",
-            path.display()
-        );
+    ) {
+        Ok(_) => true,
+        Err(error) => {
+            eprintln!(
+                "st2: resync emit for '{}' failed: {error:#}",
+                path.display()
+            );
+            false
+        }
     }
 }
 
@@ -678,6 +694,36 @@ mod tests {
         }
     }
 
+    #[test]
+    fn failed_emit_retains_digest_and_schedules_the_same_transition_for_retry() {
+        let root = tempfile::tempdir().unwrap();
+        let carrier = root.path().join("carrier.md");
+        std::fs::write(&carrier, "new bytes").unwrap();
+        let mut worker = Worker {
+            root: root.path().to_path_buf(),
+            this_host: "host".to_owned(),
+            carriers: BTreeMap::from([(
+                carrier.clone(),
+                vec![Entry {
+                    bus_id: "host.missing".to_owned(),
+                    label: "goal".to_owned(),
+                    class: CarrierClass::Immediate,
+                    digest: Some("old-digest".to_owned()),
+                    dirty: true,
+                }],
+            )]),
+            deadlines: BTreeMap::new(),
+            watched: BTreeMap::new(),
+            watcher: None,
+        };
+
+        worker.flush_path(&carrier);
+        let entry = &worker.carriers[&carrier][0];
+        assert_eq!(entry.digest.as_deref(), Some("old-digest"));
+        assert!(entry.dirty);
+        assert!(worker.deadlines.contains_key(&CarrierClass::Immediate));
+    }
+
     #[test]
     fn transition_identity_covers_every_rendered_transition_dimension() {
         let baseline = render_body(