fix(resync): close review lifecycle gaps

Author: schickling

src/agents.rs

@@ -31,6 +31,8 @@ pub struct AgentRow {
     pub desired_state_reason: Option<String>,
     /// Typed Resource bindings declared directly by the agent.
     pub resources: Vec<Resource>,
+    /// Resync coverage derived from the declaration directory and each Resource URI.
+    pub resource_resync: Vec<crate::resync::ResyncCoverage>,
     /// Newest activity time across inbox, archive, and status. Version 1 status uses its embedded
     /// writer timestamp; message files and legacy status use local mtime. `--enrich` only.
     pub last_activity_ms: Option<f64>,
@@ -74,6 +76,11 @@ pub fn roster_from_discovered(
                 desired_state: s.desired_state.as_str().to_owned(),
                 desired_state_reason: s.desired_state.reason().map(str::to_owned),
                 resources: s.resources.clone(),
+                resource_resync: s
+                    .resources
+                    .iter()
+                    .map(|resource| crate::resync::resource_coverage(agent_dir, resource))
+                    .collect(),
                 last_activity_ms: newest_activity_ms(agent_dir),
                 inbox: inbox_count(agent_dir),
                 observed: observed_state(s, agent_dir, &pty_root, this_host),
@@ -202,6 +209,30 @@ impl<'a> DriverDiagnosticJson<'a> {
     }
 }
 
+#[derive(Serialize)]
+struct ResourceJson<'a> {
+    name: &'a str,
+    uri: &'a str,
+    reason: &'a str,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    inactive_reason: Option<&'a str>,
+    resync: &'static str,
+}
+
+fn resource_json(row: &AgentRow) -> Vec<ResourceJson<'_>> {
+    row.resources
+        .iter()
+        .zip(&row.resource_resync)
+        .map(|(resource, coverage)| ResourceJson {
+            name: resource.name(),
+            uri: resource.uri(),
+            reason: resource.reason(),
+            inactive_reason: resource.inactive_reason(),
+            resync: coverage.as_str(),
+        })
+        .collect()
+}
+
 /// `st2 agents --json` row. Field order and names are the stable wire contract.
 #[derive(Serialize)]
 struct SummaryJson<'a> {
@@ -210,7 +241,7 @@ struct SummaryJson<'a> {
     name: Option<&'a str>,
     description: Option<&'a str>,
     retired: bool,
-    resources: &'a [Resource],
+    resources: Vec<ResourceJson<'a>>,
     #[serde(rename = "desiredState")]
     desired_state: &'a str,
     #[serde(rename = "desiredStateReason")]
@@ -229,7 +260,7 @@ struct EnrichedJson<'a> {
     name: Option<&'a str>,
     description: Option<&'a str>,
     retired: bool,
-    resources: &'a [Resource],
+    resources: Vec<ResourceJson<'a>>,
     #[serde(rename = "lastActivity")]
     last_activity: Option<f64>,
     inbox: usize,
@@ -254,7 +285,7 @@ pub fn to_json(rows: &[AgentRow], enrich: bool) -> String {
                 name: r.name.as_deref(),
                 description: r.description.as_deref(),
                 retired: r.retired,
-                resources: &r.resources,
+                resources: resource_json(r),
                 desired_state: &r.desired_state,
                 desired_state_reason: r.desired_state_reason.as_deref(),
                 last_activity: r.last_activity_ms,
@@ -273,7 +304,7 @@ pub fn to_json(rows: &[AgentRow], enrich: bool) -> String {
                 name: r.name.as_deref(),
                 description: r.description.as_deref(),
                 retired: r.retired,
-                resources: &r.resources,
+                resources: resource_json(r),
                 desired_state: &r.desired_state,
                 desired_state_reason: r.desired_state_reason.as_deref(),
                 observed_state: ObservedJson::from_row(r.observed.as_ref()),
@@ -340,6 +371,7 @@ mod tests {
             desired_state: if retired { "retired" } else { "running" }.to_owned(),
             desired_state_reason: None,
             resources: Vec::new(),
+            resource_resync: Vec::new(),
             last_activity_ms: last,
             inbox,
             observed: None,
@@ -385,10 +417,13 @@ mod tests {
             )
             .unwrap(),
         );
+        resource_row
+            .resource_resync
+            .push(crate::resync::ResyncCoverage::Unsupported);
 
         assert_eq!(
             to_json(&[resource_row], false),
-            r#"[{"identity":"hetz.worker","status":"available","name":null,"description":null,"retired":false,"resources":[{"name":"work","uri":"vendor+thing://authority/exact%20identity","reason":"Current implementation task."}],"desiredState":"running","desiredStateReason":null,"observedState":null,"driverDiagnostic":{"status":"absent","driver":null,"stage":null,"reason":null,"source":null,"producerVersion":null,"support":"unknown","observedAt":null,"evidenceAgeMs":null,"recovery":"publishFailureOrClearOnStageRecovery"}}]"#
+            r#"[{"identity":"hetz.worker","status":"available","name":null,"description":null,"retired":false,"resources":[{"name":"work","uri":"vendor+thing://authority/exact%20identity","reason":"Current implementation task.","resync":"unsupported"}],"desiredState":"running","desiredStateReason":null,"observedState":null,"driverDiagnostic":{"status":"absent","driver":null,"stage":null,"reason":null,"source":null,"producerVersion":null,"support":"unknown","observedAt":null,"evidenceAgeMs":null,"recovery":"publishFailureOrClearOnStageRecovery"}}]"#
         );
     }
 

src/resync.rs

@@ -43,6 +43,36 @@ impl CarrierClass {
         }
     }
 }
+/// Catalog-observable resync coverage for one declared Resource binding.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum ResyncCoverage {
+    Immediate,
+    Coalesced,
+    Silent,
+    Unsupported,
+    Inactive,
+}
+
+impl ResyncCoverage {
+    pub const fn as_str(self) -> &'static str {
+        match self {
+            Self::Immediate => "immediate",
+            Self::Coalesced => "coalesced",
+            Self::Silent => "silent",
+            Self::Unsupported => "unsupported",
+            Self::Inactive => "inactive",
+        }
+    }
+
+    fn carrier_class(self) -> Option<CarrierClass> {
+        match self {
+            Self::Immediate => Some(CarrierClass::Immediate),
+            Self::Coalesced => Some(CarrierClass::Coalesced),
+            Self::Silent | Self::Unsupported | Self::Inactive => None,
+        }
+    }
+}
+
 
 /// One watchable local carrier: binding label, absolute path, notification class.
 #[derive(Debug, Clone, PartialEq, Eq)]
@@ -73,15 +103,11 @@ pub fn watch_set_for(spec: &AgentSpec, this_host: &str) -> AgentWatchSet {
         class: CarrierClass::Immediate,
     }];
     for resource in &spec.resources {
-        if resource.inactive_reason().is_some() {
-            continue;
-        }
-        let Some(path) = resolve_local_path(agent_dir, resource.uri()) else {
-            continue;
-        };
-        let Some(class) = classify(agent_dir, resource.name(), &path) else {
+        let Some(class) = resource_coverage(agent_dir, resource).carrier_class() else {
             continue;
         };
+        let path = resolve_local_path(agent_dir, resource.uri())
+            .expect("watchable coverage must have a local path");
         carriers.push(WatchableCarrier {
             label: resource.name().to_owned(),
             path,
@@ -133,6 +159,21 @@ fn resolve_local_path(agent_dir: &Path, uri: &str) -> Option<PathBuf> {
     let path = PathBuf::from(decode_percent_path(uri).ok()?);
     Some(lexical_clean(&agent_dir.join(path)))
 }
+/// Resolve the externally visible resync coverage for one Resource binding.
+pub fn resource_coverage(agent_dir: &Path, resource: &agent_spec::spec::Resource) -> ResyncCoverage {
+    if resource.inactive_reason().is_some() {
+        return ResyncCoverage::Inactive;
+    }
+    let Some(path) = resolve_local_path(agent_dir, resource.uri()) else {
+        return ResyncCoverage::Unsupported;
+    };
+    match classify(agent_dir, resource.name(), &path) {
+        Some(CarrierClass::Immediate) => ResyncCoverage::Immediate,
+        Some(CarrierClass::Coalesced) => ResyncCoverage::Coalesced,
+        None => ResyncCoverage::Silent,
+    }
+}
+
 
 /// Remove `.` and `..` components lexically. This deliberately does not inspect the filesystem:
 /// classification follows the authored path structure without resolving symlinks.
@@ -190,6 +231,7 @@ struct WatchRefresh {
 enum Msg {
     WatchSet(WatchRefresh),
     Install(AgentWatchSet, Sender<()>),
+    Deactivate(String, Sender<()>),
     Mutations(Vec<PathBuf>),
     Rescan,
     /// Explicit stop: the worker's own watcher holds the last `Sender`, so `Disconnected`
@@ -267,6 +309,19 @@ impl ResyncSupervisor {
             let _ = ack_rx.recv();
         }
     }
+
+    /// Synchronously remove a canonical seat's active subscriptions before relaunch work begins.
+    /// Sequence floors remain retained so a later successful install cannot reuse an occurrence.
+    pub fn deactivate(&self, spec: &AgentSpec, this_host: &str) {
+        let (ack_tx, ack_rx) = channel();
+        if self
+            .tx
+            .as_ref()
+            .is_some_and(|tx| tx.send(Msg::Deactivate(spec.bus_id(this_host), ack_tx)).is_ok())
+        {
+            let _ = ack_rx.recv();
+        }
+    }
 }
 
 impl Drop for ResyncSupervisor {
@@ -422,6 +477,10 @@ fn worker_loop(root: PathBuf, this_host: String, rx: Receiver<Msg>, forward: Sen
                 worker.install_watch_set(set);
                 let _ = ack.send(());
             }
+            Ok(Msg::Deactivate(bus_id, ack)) => {
+                worker.deactivate_watch_set(&bus_id);
+                let _ = ack.send(());
+            }
             Ok(Msg::Mutations(paths)) => worker.mark_mutated(paths),
             Ok(Msg::Rescan) => worker.rescan_all(),
             Ok(Msg::Shutdown) => break,
@@ -620,6 +679,22 @@ impl Worker {
         self.finish_carrier_update(&previous_paths);
     }
 
+    fn deactivate_watch_set(&mut self, bus_id: &str) {
+        let previous = std::mem::take(&mut self.carriers);
+        let previous_paths = self.prepare_carrier_update(&previous);
+        self.carriers = previous
+            .into_iter()
+            .filter_map(|(path, entries)| {
+                let retained = entries
+                    .into_iter()
+                    .filter(|entry| entry.bus_id != bus_id)
+                    .collect::<Vec<_>>();
+                (!retained.is_empty()).then_some((path, retained))
+            })
+            .collect();
+        self.finish_carrier_update(&previous_paths);
+    }
+
     fn reconcile_dirty_deadlines(&mut self, now: Instant) {
         let dirty_classes = self
             .carriers
@@ -1033,18 +1108,30 @@ mod tests {
   host "hetz"
   command "true"
   resource "goal" uri="resources/goal.md" reason="Mission."
+  resource "journal" uri="resources/context/journal.md" reason="Memory."
   resource "issue" uri="github-issue://org/repo/41" reason="Task."
+  resource "old" uri="resources/old.md" reason="History." inactive-reason="No longer used."
 }"#,
         )
         .unwrap();
-        let set = watch_set_for(&discover(tmp.path()), "hetz");
+        let spec = discover(tmp.path());
+        let set = watch_set_for(&spec, "hetz");
         assert_eq!(set.bus_id, "hetz.worker");
         let mut labels: Vec<&str> = set.carriers.iter().map(|c| c.label.as_str()).collect();
         labels.sort();
         assert_eq!(labels, vec!["declaration", "goal"]);
         let goal = set.carriers.iter().find(|c| c.label == "goal").unwrap();
         assert_eq!(goal.class, CarrierClass::Immediate);
         assert_eq!(goal.path, dir.join("resources/goal.md"));
+        let coverage = spec
+            .resources
+            .iter()
+            .map(|resource| (resource.name(), resource_coverage(&dir, resource)))
+            .collect::<BTreeMap<_, _>>();
+        assert_eq!(coverage["goal"], ResyncCoverage::Immediate);
+        assert_eq!(coverage["journal"], ResyncCoverage::Silent);
+        assert_eq!(coverage["issue"], ResyncCoverage::Unsupported);
+        assert_eq!(coverage["old"], ResyncCoverage::Inactive);
     }
 
     #[test]