feat(resource): make resync DING subjects actionable

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

src/resource_profile_supervisor.rs

@@ -995,7 +995,7 @@ fn emit_pending_at(
         &active.desired.binding_name,
         digest,
     );
-    let subject = format!("resource {} changed", active.desired.binding_name);
+    let subject = publication_subject(&active.desired.binding_name, &topics);
     crate::event::emit_builtin_resync(
         catalog_root,
         this_host,
@@ -1014,6 +1014,14 @@ fn publication_event_id(recipient: &str, binding: &str, digest: SnapshotDigest)
     hash_text(&format!("resource-profile\0{recipient}\0{binding}\0{digest}"))
 }
 
+fn publication_subject(binding: &str, topics: &[String]) -> String {
+    if topics.is_empty() {
+        format!("resource {binding} changed: snapshot updated")
+    } else {
+        format!("resource {binding} changed: {}", topics.join(", "))
+    }
+}
+
 fn binding_state_directory(desired: &DesiredBinding) -> anyhow::Result<PathBuf> {
     let state = lexical_absolute(&crate::run::state_root())?;
     Ok(state
@@ -1130,6 +1138,21 @@ mod tests {
         assert!(!owner_matches(None, &old));
     }
 
+    #[test]
+    fn publication_subject_names_the_semantic_topics_visible_in_ding() {
+        assert_eq!(
+            publication_subject(
+                "st2-resource-profiles-pr",
+                &["ci.failure".to_owned(), "mergeability.conflict".to_owned()]
+            ),
+            "resource st2-resource-profiles-pr changed: ci.failure, mergeability.conflict"
+        );
+        assert_eq!(
+            publication_subject("review", &[]),
+            "resource review changed: snapshot updated"
+        );
+    }
+
     #[test]
     fn stale_protocol_failure_does_not_remove_the_replacement_process() {
         let old = RuntimeOwner::new(

src/resync.rs

@@ -1346,7 +1346,7 @@ fn emit_resync(
     bus_id: &str,
     transition: &PendingTransition,
 ) -> bool {
-    let subject = format!("resource {} changed", transition.binding);
+    let subject = resync_subject(&transition.binding);
     match crate::event::emit_builtin_resync(
         root,
         this_host,
@@ -1368,6 +1368,14 @@ fn emit_resync(
     }
 }
 
+fn resync_subject(binding: &str) -> String {
+    if binding == "declaration" {
+        "resource bindings changed: re-read agent spec".to_owned()
+    } else {
+        format!("resource {binding} changed: re-read carrier")
+    }
+}
+
 fn read_state(path: &Path, containment_root: Option<&Path>) -> std::io::Result<CarrierState> {
     match containment_root {
         Some(root) => read_confined(path, root),
@@ -1565,6 +1573,18 @@ mod tests {
         }
     }
 
+    #[test]
+    fn resync_subject_gives_the_agent_a_semantic_next_action() {
+        assert_eq!(
+            resync_subject("declaration"),
+            "resource bindings changed: re-read agent spec"
+        );
+        assert_eq!(
+            resync_subject("goal"),
+            "resource goal changed: re-read carrier"
+        );
+    }
+
     fn owner_incarnation(seed: u64) -> crate::event::StreamOwnerIncarnation {
         crate::event::StreamOwnerIncarnation::for_test(seed, seed + 1, 42, seed + 2)
     }