diff --git a/crates/agent-spec/src/profile.rs b/crates/agent-spec/src/profile.rs
index a21affc1..be30bc4e 100644
--- a/crates/agent-spec/src/profile.rs
+++ b/crates/agent-spec/src/profile.rs
@@ -36,7 +36,7 @@ pub const AGENT_GOAL_SCHEME: &str = "dev.schickling.agent-goal";
 /// Maximum resolver module bytes admitted by both catalog transactions and the wasm runtime.
 pub const DEFAULT_MODULE_LIMIT_BYTES: usize = 16 * 1024 * 1024;
 /// Descriptor ABI implemented by this host.
-pub const PROFILE_DESCRIPTOR_ABI_VERSION: u32 = 2;
+pub const PROFILE_DESCRIPTOR_ABI_VERSION: u32 = 3;
 /// Maximum canonical compact JSON bytes accepted for one binding selector.
 pub const DEFAULT_SELECTOR_LIMIT_BYTES: usize = 16 * 1024;
 
@@ -1091,7 +1091,7 @@ mod tests {
     use super::*;
 
     const VALID_DESCRIPTOR_JSON: &str = r#"{
-        "abiVersion": 2,
+        "abiVersion": 3,
         "capabilities": ["resolve", "read", "observe"],
         "selectorSchema": {
             "type": "object",
@@ -1157,7 +1157,7 @@ mod tests {
     }
 
     #[test]
-    fn valid_v2_descriptor_and_nested_selector_validate() {
+    fn valid_v3_descriptor_and_nested_selector_validate() {
         let descriptor = valid_descriptor();
         assert_eq!(descriptor.abi_version, PROFILE_DESCRIPTOR_ABI_VERSION);
         assert_eq!(descriptor.runtime.topology, RuntimeTopology::Shared);
@@ -1171,7 +1171,7 @@ mod tests {
 
     #[test]
     fn descriptor_rejects_unknown_abi_capability_and_fields() {
-        let unknown_abi = VALID_DESCRIPTOR_JSON.replace("\"abiVersion\": 2", "\"abiVersion\": 9");
+        let unknown_abi = VALID_DESCRIPTOR_JSON.replace("\"abiVersion\": 3", "\"abiVersion\": 9");
         assert!(
             ProfileDescriptor::from_json(unknown_abi.as_bytes())
                 .unwrap_err()
@@ -1187,7 +1187,7 @@ mod tests {
         );
 
         let unknown_field =
-            VALID_DESCRIPTOR_JSON.replace("\"abiVersion\": 2,", "\"abiVersion\": 2, \"extra\": true,");
+            VALID_DESCRIPTOR_JSON.replace("\"abiVersion\": 3,", "\"abiVersion\": 3, \"extra\": true,");
         assert!(
             ProfileDescriptor::from_json(unknown_field.as_bytes())
                 .unwrap_err()
diff --git a/crates/agent-spec/src/spec.rs b/crates/agent-spec/src/spec.rs
index a0bbd549..58fc7884 100644
--- a/crates/agent-spec/src/spec.rs
+++ b/crates/agent-spec/src/spec.rs
@@ -363,7 +363,6 @@ impl AgentSpec {
             .or_else(|| self.driver.as_ref().map(Driver::session_driver))
     }
 }
-
 fn deserialize_optional_selector<'de, D>(
     deserializer: D,
 ) -> Result<Option<serde_json::Value>, D::Error>
diff --git a/crates/agent-spec/tests/profile_wasm.rs b/crates/agent-spec/tests/profile_wasm.rs
index 61aef4e1..5335c1bf 100644
--- a/crates/agent-spec/tests/profile_wasm.rs
+++ b/crates/agent-spec/tests/profile_wasm.rs
@@ -74,7 +74,7 @@ fn descriptor_module(payload: &[u8], reported_len: usize) -> WasmResolver {
 }
 
 const VALID_DESCRIPTOR: &str = r#"{
-  "abiVersion": 2,
+  "abiVersion": 3,
   "capabilities": ["resolve", "read", "observe"],
   "selectorSchema": {
     "type": "object",
@@ -112,12 +112,12 @@ fn current_rss_bytes() -> u64 {
 }
 
 #[test]
-fn valid_v2_descriptor_executes_and_resolve_only_module_stays_passive() {
+fn valid_v3_descriptor_executes_and_resolve_only_module_stays_passive() {
     let descriptor = descriptor_module(VALID_DESCRIPTOR.as_bytes(), VALID_DESCRIPTOR.len())
         .describe_once()
         .expect("describe call succeeds")
         .expect("descriptor is present");
-    assert_eq!(descriptor.abi_version, 2);
+    assert_eq!(descriptor.abi_version, 3);
 
     assert!(
         WasmResolver::load(Path::new(DEMO_WASM_PATH))
diff --git a/crates/st2-resource-protocol/src/lib.rs b/crates/st2-resource-protocol/src/lib.rs
index c4dd0fe2..f765a61b 100644
--- a/crates/st2-resource-protocol/src/lib.rs
+++ b/crates/st2-resource-protocol/src/lib.rs
@@ -15,6 +15,199 @@ pub const MAX_SELECTOR_BYTES: usize = 16 * 1024;
 pub const MAX_HEALTH_DETAIL_BYTES: usize = 16 * 1024;
 const MAX_OPAQUE_ID_BYTES: usize = 16 * 1024;
 
+/// Fact bounds are deliberately small relative to the 2 MiB frame: even 32 facts whose strings
+/// all require JSON escaping leave ample room beside a maximal base64-encoded 1 MiB snapshot.
+pub const MAX_FACTS: usize = 32;
+pub const MAX_FACT_KEY_BYTES: usize = 128;
+pub const MAX_FACT_VALUE_BYTES: usize = 1024;
+
+#[derive(Debug, Clone, PartialEq, Eq, Default)]
+pub enum FactValue {
+    #[default]
+    Omitted,
+    Null,
+    Value(String),
+}
+
+impl FactValue {
+    pub fn value(value: impl Into<String>) -> Self {
+        Self::Value(value.into())
+    }
+
+    pub fn as_option(&self) -> Option<Option<&str>> {
+        match self {
+            Self::Omitted => None,
+            Self::Null => Some(None),
+            Self::Value(value) => Some(Some(value)),
+        }
+    }
+
+    fn is_omitted(&self) -> bool {
+        matches!(self, Self::Omitted)
+    }
+}
+
+impl Serialize for FactValue {
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: Serializer,
+    {
+        match self {
+            Self::Omitted => serializer.serialize_unit(),
+            Self::Null => serializer.serialize_none(),
+            Self::Value(value) => serializer.serialize_str(value),
+        }
+    }
+}
+
+impl<'de> Deserialize<'de> for FactValue {
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    where
+        D: Deserializer<'de>,
+    {
+        Option::<String>::deserialize(deserializer)
+            .map(|value| value.map_or(Self::Null, Self::Value))
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase", deny_unknown_fields)]
+pub struct ResourceFact {
+    key: String,
+    #[serde(default, skip_serializing_if = "FactValue::is_omitted")]
+    before: FactValue,
+    #[serde(default, skip_serializing_if = "FactValue::is_omitted")]
+    after: FactValue,
+}
+
+impl ResourceFact {
+    pub fn new(
+        key: impl Into<String>,
+        before: FactValue,
+        after: FactValue,
+    ) -> Result<Self, FactError> {
+        let fact = Self {
+            key: key.into(),
+            before,
+            after,
+        };
+        fact.validate()?;
+        Ok(fact)
+    }
+
+    pub fn current(
+        key: impl Into<String>,
+        value: impl Into<String>,
+    ) -> Result<Self, FactError> {
+        Self::new(key, FactValue::Omitted, FactValue::value(value))
+    }
+
+    pub fn transition(
+        key: impl Into<String>,
+        before: Option<impl Into<String>>,
+        after: Option<impl Into<String>>,
+    ) -> Result<Self, FactError> {
+        Self::new(
+            key,
+            before.map_or(FactValue::Null, |value| FactValue::value(value)),
+            after.map_or(FactValue::Null, |value| FactValue::value(value)),
+        )
+    }
+
+    pub fn key(&self) -> &str {
+        &self.key
+    }
+
+    pub fn before(&self) -> Option<Option<&str>> {
+        self.before.as_option()
+    }
+
+    pub fn after(&self) -> Option<Option<&str>> {
+        self.after.as_option()
+    }
+
+    pub fn validate(&self) -> Result<(), FactError> {
+        validate_fact_string("key", &self.key, MAX_FACT_KEY_BYTES, true)?;
+        if self.before.is_omitted() && self.after.is_omitted() {
+            return Err(FactError::MissingValue);
+        }
+        for (field, value) in [("before", &self.before), ("after", &self.after)] {
+            if let FactValue::Value(value) = value {
+                validate_fact_string(field, value, MAX_FACT_VALUE_BYTES, false)?;
+            }
+        }
+        Ok(())
+    }
+}
+
+fn validate_fact_string(
+    field: &'static str,
+    value: &str,
+    maximum: usize,
+    nonempty: bool,
+) -> Result<(), FactError> {
+    if nonempty && value.is_empty() {
+        return Err(FactError::Empty { field });
+    }
+    if value.len() > maximum {
+        return Err(FactError::TooLarge {
+            field,
+            actual: value.len(),
+            maximum,
+        });
+    }
+    if value
+        .chars()
+        .any(|character| character.is_control() || matches!(character, '\u{2028}' | '\u{2029}'))
+    {
+        return Err(FactError::NotPrintable { field });
+    }
+    Ok(())
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum FactError {
+    TooMany { actual: usize },
+    Empty {
+        field: &'static str,
+    },
+    TooLarge {
+        field: &'static str,
+        actual: usize,
+        maximum: usize,
+    },
+    NotPrintable {
+        field: &'static str,
+    },
+    MissingValue,
+}
+
+impl fmt::Display for FactError {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            Self::TooMany { actual } => {
+                write!(formatter, "fact list has {actual} entries; maximum is {MAX_FACTS}")
+            }
+            Self::Empty { field } => write!(formatter, "fact {field} must not be empty"),
+            Self::TooLarge {
+                field,
+                actual,
+                maximum,
+            } => write!(
+                formatter,
+                "fact {field} is {actual} bytes; maximum is {maximum}"
+            ),
+            Self::NotPrintable { field } => {
+                write!(formatter, "fact {field} must be one printable line")
+            }
+            Self::MissingValue => {
+                formatter.write_str("fact must include before, after, or both")
+            }
+        }
+    }
+}
+
+impl std::error::Error for FactError {}
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct OpaqueIdError {
     kind: &'static str,
@@ -400,6 +593,8 @@ pub enum RuntimeMessage {
         bytes: SnapshotBytes,
         topics: Vec<String>,
         #[serde(skip_serializing_if = "Option::is_none")]
+        facts: Option<Vec<ResourceFact>>,
+        #[serde(skip_serializing_if = "Option::is_none")]
         observed_at: Option<String>,
     },
     Health {
@@ -423,6 +618,7 @@ pub enum ProtocolError {
     SelectorTooLarge { actual: usize },
     HealthDetailTooLarge { actual: usize },
     InvalidTopics(&'static str),
+    InvalidFacts(FactError),
     InvalidHealthScope,
     Json(serde_json::Error),
 }
@@ -446,6 +642,7 @@ impl fmt::Display for ProtocolError {
                 "health detail is {actual} bytes; maximum is {MAX_HEALTH_DETAIL_BYTES}"
             ),
             Self::InvalidTopics(reason) => write!(formatter, "invalid topics: {reason}"),
+            Self::InvalidFacts(error) => write!(formatter, "invalid facts: {error}"),
             Self::InvalidHealthScope => formatter
                 .write_str("binding-scoped health must carry both bindingId and registration"),
             Self::Json(error) => write!(formatter, "invalid protocol JSON: {error}"),
@@ -457,6 +654,7 @@ impl std::error::Error for ProtocolError {
     fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
         match self {
             Self::Json(error) => Some(error),
+            Self::InvalidFacts(error) => Some(error),
             _ => None,
         }
     }
@@ -525,7 +723,10 @@ fn validate_host_message(message: &HostMessage) -> Result<(), ProtocolError> {
 
 fn validate_runtime_message(message: &RuntimeMessage) -> Result<(), ProtocolError> {
     match message {
-        RuntimeMessage::Publish { topics, .. } => validate_topics(topics),
+        RuntimeMessage::Publish { topics, facts, .. } => {
+            validate_topics(topics)?;
+            validate_facts(facts.as_deref().unwrap_or_default())
+        }
         RuntimeMessage::Health {
             binding_id,
             registration,
@@ -546,6 +747,17 @@ fn validate_runtime_message(message: &RuntimeMessage) -> Result<(), ProtocolErro
         }
     }
 }
+fn validate_facts(facts: &[ResourceFact]) -> Result<(), ProtocolError> {
+    if facts.len() > MAX_FACTS {
+        return Err(ProtocolError::InvalidFacts(FactError::TooMany {
+            actual: facts.len(),
+        }));
+    }
+    facts
+        .iter()
+        .try_for_each(ResourceFact::validate)
+        .map_err(ProtocolError::InvalidFacts)
+}
 
 fn validate_topics(topics: &[String]) -> Result<(), ProtocolError> {
     let mut unique = BTreeSet::new();
@@ -583,6 +795,7 @@ mod tests {
             media_type: "application/json".to_owned(),
             bytes: SnapshotBytes::new(bytes.to_vec()).unwrap(),
             topics: vec!["selected".to_owned()],
+            facts: None,
             observed_at: Some("2026-08-30T00:00:00Z".to_owned()),
         }
     }
@@ -636,6 +849,57 @@ mod tests {
         );
     }
 
+    #[test]
+    fn fact_wire_shape_distinguishes_omission_from_explicit_null() {
+        let mut message = publish(b"fact");
+        let RuntimeMessage::Publish { facts, .. } = &mut message else {
+            unreachable!();
+        };
+        *facts = Some(vec![
+            ResourceFact::current("state", "ready").unwrap(),
+            ResourceFact::transition("label", None::<String>, Some("added")).unwrap(),
+            ResourceFact::transition("removed", Some("old"), None::<String>).unwrap(),
+        ]);
+        let encoded = encode_runtime_line(&message).unwrap();
+        let json: Value = serde_json::from_slice(encoded.strip_suffix(b"\n").unwrap()).unwrap();
+        assert_eq!(
+            json["facts"],
+            json!([
+                {"key": "state", "after": "ready"},
+                {"key": "label", "before": null, "after": "added"},
+                {"key": "removed", "before": "old", "after": null}
+            ])
+        );
+        assert_eq!(decode_runtime_line(&encoded).unwrap(), message);
+    }
+
+    #[test]
+    fn invalid_fact_shapes_and_bounds_are_rejected() {
+        let missing_values = b"{\"type\":\"publish\",\"owner\":{\"incarnation\":\"i\",\"claim\":\"c\"},\"bindingId\":\"b\",\"registration\":\"r\",\"schemaId\":\"s\",\"mediaType\":\"m\",\"bytes\":\"\",\"topics\":[],\"facts\":[{\"key\":\"state\"}]}\n";
+        assert!(matches!(
+            decode_runtime_line(missing_values),
+            Err(ProtocolError::InvalidFacts(FactError::MissingValue))
+        ));
+        assert!(ResourceFact::current("", "value").is_err());
+        assert!(ResourceFact::current("state", "two\nlines").is_err());
+        assert!(ResourceFact::current("x".repeat(MAX_FACT_KEY_BYTES + 1), "value").is_err());
+        assert!(ResourceFact::current("state", "x".repeat(MAX_FACT_VALUE_BYTES + 1)).is_err());
+
+        let mut message = publish(b"facts");
+        let RuntimeMessage::Publish { facts, .. } = &mut message else {
+            unreachable!();
+        };
+        *facts = Some(
+            (0..=MAX_FACTS)
+                .map(|index| ResourceFact::current(format!("key-{index}"), "value").unwrap())
+                .collect(),
+        );
+        assert!(matches!(
+            encode_runtime_line(&message),
+            Err(ProtocolError::InvalidFacts(FactError::TooMany { .. }))
+        ));
+    }
+
     #[test]
     fn decoding_is_strict_about_fields_ids_and_digest_encoding() {
         let unknown_message_field = b"{\"type\":\"health\",\"owner\":{\"incarnation\":\"i\",\"claim\":\"c\"},\"state\":\"ready\",\"extra\":true}\n";
diff --git a/docs/vrs/07-resource-profile/requirements.md b/docs/vrs/07-resource-profile/requirements.md
index 7d1bbe37..e5dfbbe9 100644
--- a/docs/vrs/07-resource-profile/requirements.md
+++ b/docs/vrs/07-resource-profile/requirements.md
@@ -207,18 +207,26 @@ one latest-state catch-up. The direction is recorded in
 - **PROFILE-R16A Finite protocol and publication bounds:** A selector's
   canonical compact JSON is at most 16 KiB. One encoded runtime-protocol line is
   at most 2 MiB including its newline. Decoded snapshot bytes are at most 1 MiB.
-  Health detail is at most 16 KiB of UTF-8. st2 rejects an oversized value
-  without truncation and contains the failure to the affected runtime or
+  Health detail is at most 16 KiB of UTF-8. One publication carries at most 32
+  ordered facts; each fact key is at most 128 bytes and each before/after value
+  is at most 1 KiB of printable single-line UTF-8. st2 rejects an oversized
+  value without truncation and contains the failure to the affected runtime or
   binding.
 
 ### Must bound attention and catch up to current state
 
-- **PROFILE-R17 Semantic invalidation:** When snapshot bytes change, including
-  on the first successful publication, the profile classifies the change with
-  zero or more descriptor-published semantic topics. st2 applies the binding
-  selector before delivery. A selected change emits one thin invalidation
-  carrying binding identity, current snapshot digest, and selected topics. It
-  does not copy snapshot bytes or a profile-rendered summary into the event.
+- **PROFILE-R17 Semantic invalidation:** Every Resource invalidation carries the
+  same bounded ordered fact envelope in its durable body and renders at most
+  three whole facts into a subject of at most 96 Unicode scalars. Observable
+  profiles may publish facts and semantic topics beside changed snapshot bytes;
+  st2 validates the facts, applies the binding selector to topics, and retains
+  both through catch-up. Passive carrier changes publish one `content` topic
+  and a short digest transition fact. Agent Spec declaration changes publish
+  ordered binding-label facts for added, removed, and semantically changed
+  Resource declarations without exposing URIs or reasons; unavailable
+  declaration parsing falls back to a digest transition fact rather than
+  dropping the invalidation. Snapshot bytes and provider payloads remain out of
+  the event.
 - **PROFILE-R18 Built-in superseding delivery:** Smart Resource invalidations
   reuse one built-in per-agent delivery stream and the existing inbox, DING,
   deduplication, and producer-side supersession machinery. The binding name is
diff --git a/docs/vrs/07-resource-profile/spec.md b/docs/vrs/07-resource-profile/spec.md
index 8e0fa7fc..fcb367a0 100644
--- a/docs/vrs/07-resource-profile/spec.md
+++ b/docs/vrs/07-resource-profile/spec.md
@@ -50,12 +50,12 @@ closed wasm describe() -> capabilities + selector schema/default + topology
 catalog-trusted host runtime argv ----+
        |
        v provider-native observation
-publish(binding-id, bytes, topics)
+publish(binding-id, bytes, topics, facts)
        |
        v host validation + contained atomic replacement
 canonical snapshot + current digest
        |
-       v selector + pending-relevance reducer
+       v selector + pending-relevance reducer retaining topics + facts
 built-in resync event (key=binding, supersede=true)
        |
        v existing inbox + DING
@@ -309,10 +309,19 @@ For each active Resource binding, resync applies this precedence:
 3. A schemeless path uses the existing agent-directory-relative rule.
 4. Every other unregistered scheme remains opaque and unwatchable.
 
-For a passive resolved carrier, Resource Profiles add no event semantics.
-Parent-directory observation, rename replacement, digest seeding, equal-byte
-deduplication, deterministic transition identity, bounded windows, and built-in
-`resync` delivery remain the [`06-resync`](../06-resync/spec.md) pipeline.
+Passive resolved carriers use the same Resource fact envelope as observable
+publications. A content transition emits topic `content` and one
+`digest=<old-short>→<new-short>` fact. Declaration subscriptions retain a
+bounded summary keyed by binding label whose values digest URI, reason,
+inactive state, and selector. On a declaration flush, one bounded catalog parse
+derives the current summary: added labels transition absent→`declared`, removed
+labels transition `declared`→absent, and changed summary digests publish
+`label=changed`, ordered by label. URIs and reasons never enter the event. A
+parse failure, unchanged Resource summary, or summary outside fact bounds falls
+back to the declaration carrier's digest transition. Parent-directory
+observation, rename replacement, digest seeding, equal-byte deduplication,
+deterministic transition identity, and bounded windows otherwise remain the
+[`06-resync`](../06-resync/spec.md) pipeline.
 
 `notify-chain #true` extends only subscription selection. For each active
 binding through that profile, resync validates the bound agent's supervisor
@@ -341,7 +350,7 @@ fresh-instance policy, fuel budget, and no-import rule as `resolve`:
 
 ```json
 {
-  "abiVersion": 2,
+  "abiVersion": 3,
   "capabilities": ["resolve", "read", "observe"],
   "selectorSchema": {
     "type": "object",
@@ -439,7 +448,7 @@ host -> unregister {
 runtime -> publish {
   owner: { incarnation, claim },
   bindingId, registration,
-  schemaId, mediaType, bytes, topics, observedAt?
+  schemaId, mediaType, bytes, topics, facts?, observedAt?
 }
 runtime -> health {
   owner: { incarnation, claim },
@@ -465,9 +474,12 @@ Each encoded protocol line is at most 2 MiB, including the newline. `publish`
 encodes `bytes` as a padded RFC 4648 base64 string; the decoded snapshot is
 opaque. Selectors are at most 16 KiB when encoded as canonical compact JSON.
 Decoded snapshot bytes are at most 1 MiB. Health `detail` is at most 16 KiB of
-UTF-8. These bounds are checked before allocation or decoding where the
-transport permits and fail only the affected binding or runtime. st2 never
-truncates canonical snapshot bytes or health text to satisfy a bound.
+UTF-8. A publication has at most 32 ordered facts; keys are at most 128 bytes
+and values at most 1 KiB of printable single-line UTF-8. A fact carries
+`key` plus `before`, `after`, or both; explicit JSON null denotes absence.
+These bounds are checked before allocation or decoding where the transport
+permits and fail only the affected binding or runtime. st2 never truncates
+canonical snapshot bytes, facts, or health text to satisfy a bound.
 
 The host rejects unknown bindings, stale owners or registrations, mismatched
 schema or media type, unpublished topics, invalid messages, output after
@@ -484,7 +496,7 @@ The resolver's contained carrier path is the observable snapshot path. A
 successful `publish` follows one host-owned transaction:
 
 ```text
-validate binding + schema + topics + size
+validate binding + schema + topics + facts + size
         |
         v
 write new bytes to contained sibling temporary file
@@ -495,8 +507,7 @@ write new bytes to contained sibling temporary file
 compute/record current sha256 digest and freshness
         |
         `-> equal digest: no invalidation
-            changed digest: apply binding selector
-```
+            changed digest: apply binding selector and retain selected topics + facts
 
 The runtime never writes the carrier directly. Descriptor-relative no-follow
 containment from the existing resolver contract applies to the temporary file,
@@ -516,14 +527,17 @@ manifest, profile event log, or host retention policy.
 ## Semantic invalidation and catch-up (PROFILE-R17..R20)
 
 For every changed digest, including the first successful publication, the host
-intersects `publish.topics` with the normalized binding selector. An empty
-intersection updates the canonical snapshot and freshness without scheduling
-delivery. A non-empty intersection updates this bounded per-binding state:
+validates and preserves the runtime's ordered facts and intersects
+`publish.topics` with the normalized binding selector. An empty intersection
+updates the canonical snapshot and freshness without scheduling delivery. A
+non-empty intersection updates this bounded per-binding state:
 
 ```text
 current_snapshot_digest: Digest?
 last_delivered_digest: Digest?
 pending_relevant_change: bool
+pending_selected_topics: Topic[]
+pending_facts: ResourceFact[]
 deliverable: bool
 ```
 
@@ -534,23 +548,26 @@ If delivery is available, st2 emits one event on the existing built-in
 stream = resync
 key = binding name
 supersede = true
-subject = resource <binding> changed
-body = { binding, snapshotDigest, topics }
+subject = <binding> · <up to three whole facts> [<topics>]
+body = { binding, snapshotDigest, topics, facts }
 ```
 
-The body is a thin invalidation. It contains no snapshot bytes, provider
-payload, rendered summary, credential, or provider cursor. Existing event
+Subjects are at most 96 Unicode scalars. Facts retain publication order and are
+included only whole; topic space is reserved before facts are admitted. If no
+fact fits, a compatible bounded fallback remains. The durable body always
+retains the complete bounded fact list. It contains no snapshot bytes, provider
+payload, credential, URI, reason, or provider cursor. Existing event
 deduplication, inbox storage, DING rendering, and supersession apply unchanged.
-Multiple topics for one atomic publication produce one invalidation, not one
-stream or record per topic.
+Multiple topics for one atomic publication produce one invalidation.
 
-If delivery is unavailable, a relevant publication sets
+If delivery is unavailable, a relevant publication replaces the pending
+selected topics and facts with the latest relevant publication and sets
 `pending_relevant_change = true`. Later irrelevant publications may advance
-`current_snapshot_digest` but do not clear the bit. When delivery becomes
-available, st2 emits at most one invalidation for the then-current digest and
-clears the bit only after event ingress accepts the record. No pending digest
-or transition backlog exists. This is level-triggered current-state catch-up,
-not event replay.
+`current_snapshot_digest` but do not clear that state. When delivery becomes
+available, st2 emits at most one invalidation for the then-current digest with
+the retained latest relevant fact envelope, and clears it only after event
+ingress accepts the record. No transition backlog exists. This is
+level-triggered current-state catch-up, not event replay.
 
 Health has separate descriptor, selector, runtime, observation, publication,
 and delivery stages. Every stage reports affected scheme and binding without
diff --git a/src/resource_profile.rs b/src/resource_profile.rs
index a27a892c..81fdf000 100644
--- a/src/resource_profile.rs
+++ b/src/resource_profile.rs
@@ -17,14 +17,16 @@ use std::sync::atomic::{AtomicU64, Ordering};
 use serde::{Deserialize, Serialize};
 
 pub use st2_resource_protocol::{
-    BindingId, HostMessage, OpaqueIdError, OwnerClaim, ProtocolError, RegistrationToken,
-    RuntimeHealthState, RuntimeIncarnation, RuntimeMessage, RuntimeOwner, SnapshotBytes,
-    SnapshotDigest, SnapshotSizeError, MAX_HEALTH_DETAIL_BYTES, MAX_PROTOCOL_LINE_BYTES,
-    MAX_SELECTOR_BYTES, MAX_SNAPSHOT_BYTES, decode_host_line, decode_runtime_line,
-    encode_host_line, encode_runtime_line,
+    BindingId, FactError, FactValue, HostMessage, OpaqueIdError, OwnerClaim, ProtocolError,
+    RegistrationToken, ResourceFact, RuntimeHealthState, RuntimeIncarnation, RuntimeMessage,
+    RuntimeOwner, SnapshotBytes, SnapshotDigest, SnapshotSizeError, MAX_FACTS, MAX_FACT_KEY_BYTES,
+    MAX_FACT_VALUE_BYTES, MAX_HEALTH_DETAIL_BYTES, MAX_PROTOCOL_LINE_BYTES, MAX_SELECTOR_BYTES,
+    MAX_SNAPSHOT_BYTES, decode_host_line, decode_runtime_line, encode_host_line,
+    encode_runtime_line,
 };
 
-const MAX_CATCH_UP_FILE_BYTES: usize = 16 * 1024;
+// Covers the prior state envelope plus 32 maximally sized facts after worst-case JSON escaping.
+const MAX_CATCH_UP_FILE_BYTES: usize = 256 * 1024;
 const CATCH_UP_FILE: &str = "resource-profile-catch-up.json";
 const PUBLICATION_INTENT_FILE: &str = "resource-profile-publication-intent.json";
 static TEMPORARY_SEQUENCE: AtomicU64 = AtomicU64::new(0);
@@ -368,6 +370,7 @@ impl RuntimeLifecycle {
                 media_type,
                 bytes,
                 topics,
+                facts,
                 observed_at,
             } => {
                 let binding = self.require_registration(owner, binding_id, registration)?;
@@ -390,6 +393,7 @@ impl RuntimeLifecycle {
                     target: &binding.target,
                     bytes,
                     selected_topics: binding.contract.selection.select(topics),
+                    facts: facts.as_deref().unwrap_or_default(),
                     observed_at: observed_at.as_deref(),
                 }))
             }
@@ -497,6 +501,7 @@ pub struct AcceptedPublication<'a> {
     target: &'a SnapshotTarget,
     bytes: &'a SnapshotBytes,
     selected_topics: Vec<String>,
+    facts: &'a [ResourceFact],
     observed_at: Option<&'a str>,
 }
 
@@ -509,12 +514,20 @@ impl<'a> AcceptedPublication<'a> {
         &self.selected_topics
     }
 
+    pub fn facts(&self) -> &[ResourceFact] {
+        self.facts
+    }
     pub fn observed_at(&self) -> Option<&str> {
         self.observed_at
     }
 
     fn prepare(self) -> Result<PreparedPublication<'a>, PublicationError> {
-        prepare_snapshot(self.target, self.bytes.as_slice(), self.selected_topics)
+        prepare_snapshot(
+            self.target,
+            self.bytes.as_slice(),
+            self.selected_topics,
+            self.facts.to_vec(),
+        )
     }
 }
 
@@ -551,6 +564,7 @@ pub struct PublicationOutcome {
     digest: SnapshotDigest,
     change: SnapshotChange,
     selected_topics: Vec<String>,
+    facts: Vec<ResourceFact>,
 }
 
 impl PublicationOutcome {
@@ -566,6 +580,9 @@ impl PublicationOutcome {
         &self.selected_topics
     }
 
+    pub fn facts(&self) -> &[ResourceFact] {
+        &self.facts
+    }
     pub fn invalidating(&self) -> bool {
         self.change != SnapshotChange::Equal && !self.selected_topics.is_empty()
     }
@@ -637,6 +654,7 @@ fn prepare_snapshot<'a>(
     target: &'a SnapshotTarget,
     bytes: &'a [u8],
     selected_topics: Vec<String>,
+    facts: Vec<ResourceFact>,
 ) -> Result<PreparedPublication<'a>, PublicationError> {
     if bytes.len() > MAX_SNAPSHOT_BYTES {
         return Err(PublicationError::SnapshotTooLarge { actual: bytes.len() });
@@ -657,6 +675,7 @@ fn prepare_snapshot<'a>(
             digest,
             change,
             selected_topics,
+            facts,
         },
     })
 }
@@ -665,8 +684,9 @@ fn publish_snapshot(
     target: &SnapshotTarget,
     bytes: &[u8],
     selected_topics: Vec<String>,
+    facts: Vec<ResourceFact>,
 ) -> Result<PublicationOutcome, PublicationError> {
-    let prepared = prepare_snapshot(target, bytes, selected_topics)?;
+    let prepared = prepare_snapshot(target, bytes, selected_topics, facts)?;
     prepared.commit()?;
     Ok(prepared.outcome)
 }
@@ -679,6 +699,8 @@ pub struct CatchUpState {
     pending_relevant_change: bool,
     #[serde(default)]
     pending_selected_topics: Vec<String>,
+    #[serde(default)]
+    pending_facts: Vec<ResourceFact>,
     deliverable: bool,
 }
 
@@ -699,6 +721,9 @@ impl CatchUpState {
         &self.pending_selected_topics
     }
 
+    pub fn pending_facts(&self) -> &[ResourceFact] {
+        &self.pending_facts
+    }
     pub fn deliverable(&self) -> bool {
         self.deliverable
     }
@@ -715,6 +740,12 @@ impl CatchUpState {
             ));
         }
         validate_persisted_topics(&self.pending_selected_topics)?;
+        validate_persisted_facts(&self.pending_facts)?;
+        if !self.pending_relevant_change && !self.pending_facts.is_empty() {
+            return Err(CatchUpError::InvalidState(
+                "pending facts require a pending relevant change",
+            ));
+        }
         Ok(())
     }
 }
@@ -723,6 +754,7 @@ impl CatchUpState {
 pub struct DeliveryRequest {
     digest: SnapshotDigest,
     selected_topics: Vec<String>,
+    facts: Vec<ResourceFact>,
 }
 
 impl DeliveryRequest {
@@ -733,6 +765,10 @@ impl DeliveryRequest {
     pub fn selected_topics(&self) -> &[String] {
         &self.selected_topics
     }
+
+    pub fn facts(&self) -> &[ResourceFact] {
+        &self.facts
+    }
 }
 
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
@@ -740,6 +776,8 @@ impl DeliveryRequest {
 struct PublicationIntent {
     digest: SnapshotDigest,
     selected_topics: Vec<String>,
+    #[serde(default)]
+    facts: Vec<ResourceFact>,
 }
 
 impl PublicationIntent {
@@ -747,11 +785,13 @@ impl PublicationIntent {
         Self {
             digest: outcome.digest,
             selected_topics: outcome.selected_topics.clone(),
+            facts: outcome.facts.clone(),
         }
     }
 
     fn validate(&self) -> Result<(), CatchUpError> {
-        validate_persisted_topics(&self.selected_topics)
+        validate_persisted_topics(&self.selected_topics)?;
+        validate_persisted_facts(&self.facts)
     }
 }
 
@@ -766,6 +806,12 @@ fn validate_persisted_topics(topics: &[String]) -> Result<(), CatchUpError> {
     }
     Ok(())
 }
+fn validate_persisted_facts(facts: &[ResourceFact]) -> Result<(), CatchUpError> {
+    if facts.len() > MAX_FACTS || facts.iter().any(|fact| fact.validate().is_err()) {
+        return Err(CatchUpError::InvalidState("persisted facts are invalid"));
+    }
+    Ok(())
+}
 
 #[derive(Debug)]
 pub struct CatchUp {
@@ -850,6 +896,7 @@ impl CatchUp {
                 if !intent.selected_topics.is_empty() {
                     next.pending_relevant_change = true;
                     next.pending_selected_topics = intent.selected_topics.clone();
+                    next.pending_facts = intent.facts.clone();
                 }
             }
             Some(_) | None => {
@@ -890,6 +937,7 @@ impl CatchUp {
         Some(DeliveryRequest {
             digest: self.state.current_snapshot_digest?,
             selected_topics: self.state.pending_selected_topics.clone(),
+            facts: self.state.pending_facts.clone(),
         })
     }
 
@@ -906,6 +954,7 @@ impl CatchUp {
         next.last_delivered_digest = Some(digest);
         next.pending_relevant_change = false;
         next.pending_selected_topics.clear();
+        next.pending_facts.clear();
         self.commit(next)?;
         Ok(true)
     }
@@ -919,6 +968,7 @@ impl CatchUp {
         if outcome.invalidating() {
             next.pending_relevant_change = true;
             next.pending_selected_topics = outcome.selected_topics.clone();
+            next.pending_facts = outcome.facts.clone();
         }
         self.commit(next)?;
         Ok(self.pending_delivery())
@@ -1316,6 +1366,7 @@ mod tests {
             media_type: "application/json".to_owned(),
             bytes: SnapshotBytes::new(bytes.to_vec()).unwrap(),
             topics: topics.iter().map(|topic| (*topic).to_owned()).collect(),
+            facts: None,
             observed_at: None,
         }
     }
@@ -1441,7 +1492,8 @@ mod tests {
         let target = SnapshotTarget::new(&root, "resources/github-pr/owner/repo/389.json").unwrap();
         assert_eq!(target.current_digest().unwrap(), None);
 
-        let outcome = publish_snapshot(&target, b"bytes", vec!["selected".to_owned()]).unwrap();
+        let outcome =
+            publish_snapshot(&target, b"bytes", vec!["selected".to_owned()], Vec::new()).unwrap();
 
         assert_eq!(outcome.change(), SnapshotChange::First);
         assert_eq!(
@@ -1491,6 +1543,7 @@ mod tests {
             digest: SnapshotDigest::of(b"relevant"),
             change: SnapshotChange::First,
             selected_topics: vec!["selected".to_owned()],
+            facts: vec![ResourceFact::current("state", "ready").unwrap()],
         };
         let irrelevant = PublicationOutcome {
             digest: SnapshotDigest::of(b"later but irrelevant"),
@@ -1498,6 +1551,7 @@ mod tests {
                 previous: relevant.digest(),
             },
             selected_topics: Vec::new(),
+            facts: Vec::new(),
         };
         let mut catch_up = CatchUp::open(&state_directory).unwrap();
         assert_eq!(catch_up.record_publication(&relevant).unwrap(), None);
@@ -1507,6 +1561,7 @@ mod tests {
         let request = catch_up.set_deliverable(true).unwrap().unwrap();
         assert_eq!(request.digest(), irrelevant.digest());
         assert_eq!(request.selected_topics(), ["selected"]);
+        assert_eq!(request.facts(), relevant.facts());
         assert!(!catch_up.acknowledge_delivery(relevant.digest()).unwrap());
         assert!(catch_up.state().pending_relevant_change());
         assert!(catch_up.acknowledge_delivery(irrelevant.digest()).unwrap());
@@ -1525,6 +1580,7 @@ mod tests {
             digest: SnapshotDigest::of(b"snapshot"),
             change: SnapshotChange::First,
             selected_topics: vec!["selected".to_owned()],
+            facts: vec![ResourceFact::current("state", "ready").unwrap()],
         };
         {
             let mut catch_up = CatchUp::open(&state_directory).unwrap();
@@ -1546,6 +1602,36 @@ mod tests {
             catch_up.state().pending_selected_topics(),
             ["selected"]
         );
+        assert_eq!(catch_up.state().pending_facts(), outcome.facts());
+    }
+
+    #[test]
+    fn catch_up_persists_a_maximal_worst_case_escaped_fact_envelope() {
+        let directory = tempfile::tempdir().unwrap();
+        let state_directory = fs::canonicalize(directory.path()).unwrap();
+        let facts = (0..MAX_FACTS)
+            .map(|_| {
+                ResourceFact::new(
+                    "\"".repeat(MAX_FACT_KEY_BYTES),
+                    FactValue::value("\"".repeat(MAX_FACT_VALUE_BYTES)),
+                    FactValue::value("\\".repeat(MAX_FACT_VALUE_BYTES)),
+                )
+                .unwrap()
+            })
+            .collect::<Vec<_>>();
+        let outcome = PublicationOutcome {
+            digest: SnapshotDigest::of(b"snapshot"),
+            change: SnapshotChange::First,
+            selected_topics: vec!["selected".to_owned()],
+            facts,
+        };
+
+        {
+            let mut catch_up = CatchUp::open(&state_directory).unwrap();
+            catch_up.record_publication(&outcome).unwrap();
+        }
+        let catch_up = CatchUp::open(&state_directory).unwrap();
+        assert_eq!(catch_up.state().pending_facts(), outcome.facts());
     }
 
     #[test]
@@ -1606,7 +1692,7 @@ mod tests {
         symlink(outside.path(), root.join("linked-parent")).unwrap();
         let target = SnapshotTarget::new(&root, "linked-parent/snapshot.json").unwrap();
         assert!(matches!(
-            publish_snapshot(&target, b"bytes", vec!["selected".to_owned()]),
+            publish_snapshot(&target, b"bytes", vec!["selected".to_owned()], Vec::new()),
             Err(PublicationError::Io(_))
         ));
 
diff --git a/src/resource_profile_supervisor.rs b/src/resource_profile_supervisor.rs
index 18537562..84beb250 100644
--- a/src/resource_profile_supervisor.rs
+++ b/src/resource_profile_supervisor.rs
@@ -24,7 +24,7 @@ use sha2::{Digest as _, Sha256};
 use crate::catalog::CatalogConfig;
 use crate::resource_profile::{
     AcceptedOutput, BindingId, BindingRegistration, CatchUp, HostMessage, OwnerClaim,
-    PublicationContract, RegistrationToken, RuntimeHealthState, RuntimeIncarnation,
+    PublicationContract, RegistrationToken, ResourceFact, RuntimeHealthState, RuntimeIncarnation,
     RuntimeLifecycle, RuntimeMessage, RuntimeOwner, SnapshotDigest, SnapshotTarget, TopicSelection,
     MAX_PROTOCOL_LINE_BYTES, decode_runtime_line, encode_host_line,
 };
@@ -982,20 +982,28 @@ fn emit_pending_at(
         binding: &'a str,
         snapshot_digest: String,
         topics: &'a [String],
+        facts: &'a [ResourceFact],
     }
     let digest = delivery.digest();
     let topics = delivery.selected_topics().to_vec();
+    let facts = delivery.facts();
     let body = serde_json::to_string(&Body {
         binding: &active.desired.binding_name,
         snapshot_digest: digest.to_string(),
         topics: &topics,
+        facts,
     })?;
     let event_id = publication_event_id(
         &active.desired.recipient,
         &active.desired.binding_name,
         digest,
     );
-    let subject = publication_subject(&active.desired.binding_name, &topics);
+    let subject = resource_change_subject(
+        &active.desired.binding_name,
+        facts,
+        &topics,
+        "snapshot updated",
+    );
     crate::event::emit_builtin_resync(
         catalog_root,
         this_host,
@@ -1014,11 +1022,68 @@ 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")
+pub(crate) fn resource_change_subject(
+    binding: &str,
+    facts: &[ResourceFact],
+    topics: &[String],
+    fallback: &str,
+) -> String {
+    const SUBJECT_MAX_SCALARS: usize = 96;
+    const MAX_RENDERED_FACTS: usize = 3;
+
+    let topic_suffix = if topics.is_empty() {
+        String::new()
+    } else {
+        format!(" [{}]", topics.join(", "))
+    };
+    let base = format!("{binding} · ");
+    let mut rendered = Vec::new();
+    for fact in facts.iter().take(MAX_RENDERED_FACTS) {
+        let fact = render_fact(fact);
+        let candidate = format!("{base}{}{topic_suffix}", {
+            let mut candidate_facts = rendered.clone();
+            candidate_facts.push(fact.clone());
+            candidate_facts.join("; ")
+        });
+        if candidate.chars().count() > SUBJECT_MAX_SCALARS {
+            break;
+        }
+        rendered.push(fact);
+    }
+
+    let detail = if rendered.is_empty() {
+        fallback.to_owned()
+    } else {
+        rendered.join("; ")
+    };
+    let subject = format!("{base}{detail}{topic_suffix}");
+    if subject.chars().count() <= SUBJECT_MAX_SCALARS {
+        return subject;
+    }
+
+    // Facts and topic names are never clipped. A pathological oversized binding or topic suffix
+    // falls back to the binding and bounded generic detail; the durable body remains complete.
+    let fallback = format!("{binding} · {fallback}");
+    if fallback.chars().count() <= SUBJECT_MAX_SCALARS {
+        fallback
     } else {
-        format!("resource {binding} changed: {}", topics.join(", "))
+        fallback.chars().take(SUBJECT_MAX_SCALARS).collect()
+    }
+}
+
+fn render_fact(fact: &ResourceFact) -> String {
+    match (fact.before(), fact.after()) {
+        (None, Some(Some(after))) => format!("{}={after}", fact.key()),
+        (None, Some(None)) => format!("{}=removed", fact.key()),
+        (Some(None), Some(Some(after))) => format!("{}=+{after}", fact.key()),
+        (Some(Some(before)), Some(None)) => format!("{}=-{before}", fact.key()),
+        (Some(Some(before)), Some(Some(after))) => {
+            format!("{}={before}→{after}", fact.key())
+        }
+        (Some(None), Some(None)) => format!("{}=absent", fact.key()),
+        (Some(Some(before)), None) => format!("{} was {before}", fact.key()),
+        (Some(None), None) => format!("{} was absent", fact.key()),
+        (None, None) => unreachable!("validated facts always carry a value"),
     }
 }
 
@@ -1139,17 +1204,45 @@ mod tests {
     }
 
     #[test]
-    fn publication_subject_names_the_semantic_topics_visible_in_ding() {
+    fn publication_subject_renders_ordered_facts_and_reserves_topics() {
+        let facts = vec![
+            ResourceFact::current("state", "ready").unwrap(),
+            ResourceFact::transition("label", None::<String>, Some("bug")).unwrap(),
+            ResourceFact::transition("owner", Some("alice"), Some("bob")).unwrap(),
+            ResourceFact::current("fourth", "omitted").unwrap(),
+        ];
         assert_eq!(
-            publication_subject(
-                "st2-resource-profiles-pr",
-                &["ci.failure".to_owned(), "mergeability.conflict".to_owned()]
+            resource_change_subject(
+                "review",
+                &facts,
+                &["ci.failure".to_owned()],
+                "snapshot updated"
             ),
-            "resource st2-resource-profiles-pr changed: ci.failure, mergeability.conflict"
+            "review · state=ready; label=+bug; owner=alice→bob [ci.failure]"
         );
+    }
+
+    #[test]
+    fn publication_subject_drops_low_priority_facts_atomically_within_96_scalars() {
+        let facts = vec![
+            ResourceFact::current("priority", "x".repeat(80)).unwrap(),
+            ResourceFact::current("lower", "must-not-leapfrog").unwrap(),
+        ];
+        let subject = resource_change_subject(
+            "review",
+            &facts,
+            &["selected.topic".to_owned()],
+            "snapshot updated",
+        );
+        assert_eq!(subject, "review · snapshot updated [selected.topic]");
+        assert!(subject.chars().count() <= 96);
+    }
+
+    #[test]
+    fn publication_subject_without_facts_has_a_useful_compatible_fallback() {
         assert_eq!(
-            publication_subject("review", &[]),
-            "resource review changed: snapshot updated"
+            resource_change_subject("review", &[], &[], "snapshot updated"),
+            "review · snapshot updated"
         );
     }
 
diff --git a/src/resync.rs b/src/resync.rs
index 9c51687c..79afd5eb 100644
--- a/src/resync.rs
+++ b/src/resync.rs
@@ -16,11 +16,14 @@ use std::thread::JoinHandle;
 use std::time::{Duration, Instant};
 
 use notify::Watcher as _;
+use serde::Serialize;
 use sha2::{Digest as _, Sha256};
 use agent_spec::profile::{
     ProfileClass, ResourceProfileRefresh, ResourceProfileRegistry,
 };
-use agent_spec::spec::{AgentSpec, decode_percent_path};
+use agent_spec::spec::{AgentSpec, Resource, decode_percent_path};
+
+use crate::resource_profile::{MAX_FACTS, MAX_FACT_KEY_BYTES, ResourceFact};
 
 /// The reserved stream used only by the supervisor's crate-internal resync publisher.
 pub const RESYNC_STREAM: &str = "resync";
@@ -85,6 +88,57 @@ pub struct WatchableCarrier {
     pub class: CarrierClass,
     pub containment_root: Option<PathBuf>,
 }
+#[derive(Debug, Clone, PartialEq, Eq)]
+struct DeclarationSummary {
+    bindings: BTreeMap<String, String>,
+    complete: bool,
+}
+
+fn declaration_summary(spec: &AgentSpec) -> DeclarationSummary {
+    let mut bindings = spec
+        .resources
+        .iter()
+        .map(|resource| {
+            (
+                resource.name().to_owned(),
+                declaration_resource_digest(resource),
+            )
+        })
+        .collect::<Vec<_>>();
+    bindings.sort_by(|left, right| left.0.cmp(&right.0));
+    let complete = bindings.len() <= MAX_FACTS
+        && bindings
+            .iter()
+            .all(|(label, _)| label.len() <= MAX_FACT_KEY_BYTES);
+    DeclarationSummary {
+        bindings: bindings.into_iter().take(MAX_FACTS).collect(),
+        complete,
+    }
+}
+
+fn declaration_resource_digest(resource: &Resource) -> String {
+    let mut digest = Sha256::new();
+    digest.update(b"st2.resync.resource-declaration.v1\0");
+    update_digest_field(&mut digest, resource.uri().as_bytes());
+    update_digest_field(&mut digest, resource.reason().as_bytes());
+    match resource.inactive_reason() {
+        Some(reason) => {
+            digest.update([1]);
+            update_digest_field(&mut digest, reason.as_bytes());
+        }
+        None => digest.update([0]),
+    }
+    let selector = serde_json::to_vec(&resource.selector())
+        .expect("a parsed JSON selector always serializes");
+    update_digest_field(&mut digest, &selector);
+    format!("{:x}", digest.finalize())
+}
+
+fn update_digest_field(digest: &mut Sha256, value: &[u8]) {
+    let length = u64::try_from(value.len()).expect("declaration fields fit in u64");
+    digest.update(length.to_be_bytes());
+    digest.update(value);
+}
 
 /// The watchable carriers of one agent, keyed by its declaration path with current routing IDs.
 #[derive(Debug, Clone, PartialEq, Eq)]
@@ -93,6 +147,7 @@ pub struct AgentWatchSet {
     pub bus_id: String,
     pub seat_id: Option<String>,
     pub carriers: Vec<WatchableCarrier>,
+    declaration_summary: Option<DeclarationSummary>,
 }
 
 /// Resolve one spec's watch set: the declaration file plus every active resource binding whose
@@ -207,6 +262,7 @@ fn resolve_watch_set(
                     .unwrap_or_else(|| format!("{}.{}", spec.bus_id(this_host), task.name))
             }),
             carriers,
+            declaration_summary: Some(declaration_summary(spec)),
         },
         diagnostics,
     )
@@ -660,13 +716,72 @@ enum CarrierState {
 }
 
 impl CarrierState {
-    fn render(&self) -> &str {
+    fn fact_value(&self) -> String {
         match self {
-            Self::Present(digest) => digest,
-            Self::Missing => "missing",
+            Self::Present(digest) => digest.chars().take(12).collect(),
+            Self::Missing => "missing".to_owned(),
         }
     }
 }
+fn digest_transition_fact(old: &CarrierState, new: &CarrierState) -> Vec<ResourceFact> {
+    vec![
+        ResourceFact::transition("digest", Some(old.fact_value()), Some(new.fact_value()))
+            .expect("short carrier digests are valid facts"),
+    ]
+}
+
+fn declaration_transition_facts(
+    old: Option<&DeclarationSummary>,
+    new: Option<&DeclarationSummary>,
+    old_state: &CarrierState,
+    new_state: &CarrierState,
+) -> Vec<ResourceFact> {
+    let (Some(old), Some(new)) = (old, new) else {
+        return digest_transition_fact(old_state, new_state);
+    };
+    if !old.complete || !new.complete {
+        return digest_transition_fact(old_state, new_state);
+    }
+
+    let labels = old
+        .bindings
+        .keys()
+        .chain(new.bindings.keys())
+        .cloned()
+        .collect::<BTreeSet<_>>();
+    let facts = labels
+        .into_iter()
+        .filter_map(|label| match (old.bindings.get(&label), new.bindings.get(&label)) {
+            (None, Some(_)) => Some(ResourceFact::transition(
+                label,
+                None::<String>,
+                Some("declared".to_owned()),
+            )),
+            (Some(_), None) => Some(ResourceFact::transition(
+                label,
+                Some("declared".to_owned()),
+                None::<String>,
+            )),
+            (Some(before), Some(after)) if before != after => {
+                Some(ResourceFact::current(label, "changed"))
+            }
+            _ => None,
+        })
+        .collect::<Result<Vec<_>, _>>();
+    match facts {
+        Ok(facts) if !facts.is_empty() => facts,
+        Ok(_) | Err(_) => digest_transition_fact(old_state, new_state),
+    }
+}
+
+fn current_declaration_summary(root: &Path, path: &Path) -> Option<DeclarationSummary> {
+    crate::discover_strict(root)
+        .specs
+        .into_iter()
+        .find(|spec| lexical_clean(&spec.path) == path)
+        .map(|spec| declaration_summary(&spec))
+}
+
 
 #[derive(Debug, Clone, PartialEq, Eq)]
 struct PendingTransition {
@@ -674,8 +789,11 @@ struct PendingTransition {
     path: PathBuf,
     old_state: CarrierState,
     new_state: CarrierState,
+    facts: Vec<ResourceFact>,
+    topics: Vec<String>,
     body: String,
     event_id: String,
+    new_declaration_summary: Option<DeclarationSummary>,
 }
 
 impl PendingTransition {
@@ -686,17 +804,73 @@ impl PendingTransition {
         new_state: &CarrierState,
         incarnation: crate::event::StreamOwnerIncarnation,
         sequence: u64,
+    ) -> Self {
+        Self::capture(
+            binding,
+            path,
+            old_state,
+            new_state,
+            digest_transition_fact(old_state, new_state),
+            vec!["content".to_owned()],
+            None,
+            incarnation,
+            sequence,
+        )
+    }
+
+    fn declaration(
+        path: &Path,
+        old_state: &CarrierState,
+        new_state: &CarrierState,
+        old_summary: Option<&DeclarationSummary>,
+        new_summary: Option<DeclarationSummary>,
+        incarnation: crate::event::StreamOwnerIncarnation,
+        sequence: u64,
+    ) -> Self {
+        let facts = declaration_transition_facts(
+            old_summary,
+            new_summary.as_ref(),
+            old_state,
+            new_state,
+        );
+        Self::capture(
+            "declaration",
+            path,
+            old_state,
+            new_state,
+            facts,
+            vec!["declaration".to_owned()],
+            new_summary,
+            incarnation,
+            sequence,
+        )
+    }
+
+    #[allow(clippy::too_many_arguments)]
+    fn capture(
+        binding: &str,
+        path: &Path,
+        old_state: &CarrierState,
+        new_state: &CarrierState,
+        facts: Vec<ResourceFact>,
+        topics: Vec<String>,
+        new_declaration_summary: Option<DeclarationSummary>,
+        incarnation: crate::event::StreamOwnerIncarnation,
+        sequence: u64,
     ) -> Self {
         let occurrence = incarnation.occurrence_token(sequence);
-        let body = render_body(binding, path, old_state, new_state, &occurrence);
+        let body = render_body(binding, &topics, &facts, &occurrence);
         let event_id = transition_identity(&body);
         Self {
             binding: binding.to_owned(),
             path: path.to_path_buf(),
             old_state: old_state.clone(),
             new_state: new_state.clone(),
+            facts,
+            topics,
             body,
             event_id,
+            new_declaration_summary,
         }
     }
 }
@@ -708,6 +882,7 @@ struct Entry {
     class: CarrierClass,
     containment_root: Option<PathBuf>,
     state: Option<CarrierState>,
+    declaration_summary: Option<DeclarationSummary>,
     /// Last occurrence sequence reserved by this retained subscription. Sequence zero is the
     /// silent seeded state; only capturing a new immutable transition advances it.
     occurrence_sequence: u64,
@@ -843,6 +1018,7 @@ fn rebuild_carriers(
 ) -> BTreeMap<PathBuf, Vec<Entry>> {
     let mut next: BTreeMap<PathBuf, Vec<Entry>> = BTreeMap::new();
     for set in refresh.sets {
+        let seeded_declaration_summary = set.declaration_summary.clone();
         for carrier in set.carriers {
             // The canonical recipient and binding label identify one subscription across
             // declaration and carrier relocation. Rebuild every retained entry from the current
@@ -852,32 +1028,37 @@ fn rebuild_carriers(
             // recipient-scoped namespace.
             let identity = (set.bus_id.clone(), carrier.label.clone());
             let retained = take_retained_entry(&mut previous, &set.bus_id, &carrier.label);
-            let (state, occurrence_sequence, pending_transition, dirty) = retained.map_or_else(
-                || {
-                    let (state, dirty) =
-                        match read_state(&carrier.path, carrier.containment_root.as_deref()) {
-                            Ok(state) => (Some(state), false),
-                            Err(error) => {
-                                diagnose_read_error(&carrier.path, &error);
-                                (None, true)
-                            }
-                        };
-                    (
-                        state,
-                        subscription_sequences.get(&identity).copied().unwrap_or(0),
-                        None,
-                        dirty,
-                    )
-                },
-                |entry| {
-                    (
-                        entry.state,
-                        entry.occurrence_sequence,
-                        entry.pending_transition,
-                        entry.dirty,
-                    )
-                },
-            );
+            let (state, declaration_summary, occurrence_sequence, pending_transition, dirty) =
+                retained.map_or_else(
+                    || {
+                        let (state, dirty) =
+                            match read_state(&carrier.path, carrier.containment_root.as_deref()) {
+                                Ok(state) => (Some(state), false),
+                                Err(error) => {
+                                    diagnose_read_error(&carrier.path, &error);
+                                    (None, true)
+                                }
+                            };
+                        (
+                            state,
+                            (carrier.label == "declaration")
+                                .then(|| seeded_declaration_summary.clone())
+                                .flatten(),
+                            subscription_sequences.get(&identity).copied().unwrap_or(0),
+                            None,
+                            dirty,
+                        )
+                    },
+                    |entry| {
+                        (
+                            entry.state,
+                            entry.declaration_summary,
+                            entry.occurrence_sequence,
+                            entry.pending_transition,
+                            entry.dirty,
+                        )
+                    },
+                );
             let entry = Entry {
                 bus_id: set.bus_id.clone(),
                 seat_id: set.seat_id.clone(),
@@ -885,6 +1066,7 @@ fn rebuild_carriers(
                 class: carrier.class,
                 containment_root: carrier.containment_root.clone(),
                 state,
+                declaration_summary,
                 occurrence_sequence,
                 pending_transition,
                 dirty,
@@ -1235,6 +1417,11 @@ impl Worker {
     fn flush_path(&mut self, path: &Path, due_class: Option<CarrierClass>) {
         let occurrence_incarnation =
             crate::event::current_stream_owner_incarnation(&self.root, &self.this_host).ok();
+        let observed_declaration_summary = self
+            .carriers
+            .get(path)
+            .is_some_and(|entries| entries.iter().any(|entry| entry.label == "declaration"))
+            .then(|| current_declaration_summary(&self.root, path));
         let Some(entries) = self.carriers.get_mut(path) else {
             return;
         };
@@ -1253,6 +1440,9 @@ impl Worker {
                         .take()
                         .expect("pending transition was just observed");
                     entry.state = Some(completed.new_state);
+                    if completed.binding == "declaration" {
+                        entry.declaration_summary = completed.new_declaration_summary;
+                    }
                     // The current carrier may have advanced or rebound while the immutable
                     // transition was pending. Complete it first, then schedule current state.
                     match observed {
@@ -1305,17 +1495,32 @@ impl Worker {
                 retries.push(entry.class);
                 continue;
             };
-            let transition = PendingTransition::new(
-                &entry.label,
-                path,
-                old_state,
-                &target_state,
-                incarnation,
-                sequence,
-            );
+            let transition = if entry.label == "declaration" {
+                PendingTransition::declaration(
+                    path,
+                    old_state,
+                    &target_state,
+                    entry.declaration_summary.as_ref(),
+                    observed_declaration_summary.clone().flatten(),
+                    incarnation,
+                    sequence,
+                )
+            } else {
+                PendingTransition::new(
+                    &entry.label,
+                    path,
+                    old_state,
+                    &target_state,
+                    incarnation,
+                    sequence,
+                )
+            };
             entry.occurrence_sequence = sequence;
             if emit_resync(&self.root, &self.this_host, &entry.bus_id, &transition) {
                 entry.state = Some(target_state);
+                if transition.binding == "declaration" {
+                    entry.declaration_summary = transition.new_declaration_summary;
+                }
             } else {
                 entry.pending_transition = Some(transition);
                 entry.dirty = true;
@@ -1346,7 +1551,12 @@ fn emit_resync(
     bus_id: &str,
     transition: &PendingTransition,
 ) -> bool {
-    let subject = resync_subject(&transition.binding);
+    let subject = crate::resource_profile_supervisor::resource_change_subject(
+        &transition.binding,
+        &transition.facts,
+        &transition.topics,
+        "content changed",
+    );
     match crate::event::emit_builtin_resync(
         root,
         this_host,
@@ -1368,13 +1578,7 @@ 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 {
@@ -1547,19 +1751,28 @@ fn read_confined(_path: &Path, _root: &Path) -> std::io::Result<CarrierState> {
     ))
 }
 
+#[derive(Serialize)]
+#[serde(rename_all = "camelCase")]
+struct ResyncBody<'a> {
+    binding: &'a str,
+    topics: &'a [String],
+    facts: &'a [ResourceFact],
+    occurrence: &'a str,
+}
+
 fn render_body(
-    label: &str,
-    path: &Path,
-    old: &CarrierState,
-    new: &CarrierState,
+    binding: &str,
+    topics: &[String],
+    facts: &[ResourceFact],
     occurrence: &str,
 ) -> String {
-    format!(
-        "resource `{label}` changed\n\nbinding: {label}\npath: {}\nold: {}\nnew: {}\noccurrence: {occurrence}\n",
-        path.display(),
-        old.render(),
-        new.render(),
-    )
+    serde_json::to_string(&ResyncBody {
+        binding,
+        topics,
+        facts,
+        occurrence,
+    })
+    .expect("validated resync facts always serialize")
 }
 #[cfg(test)]
 mod tests {
@@ -1574,15 +1787,25 @@ 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"
+    fn resync_subject_uses_the_shared_three_fact_and_96_scalar_renderer() {
+        let facts = vec![
+            ResourceFact::transition("alpha", None::<String>, Some("declared")).unwrap(),
+            ResourceFact::current("beta", "changed").unwrap(),
+            ResourceFact::transition("charlie", Some("declared"), None::<String>).unwrap(),
+            ResourceFact::current("delta", "omitted").unwrap(),
+        ];
+        let subject = crate::resource_profile_supervisor::resource_change_subject(
+            "declaration",
+            &facts,
+            &["declaration".to_owned()],
+            "content changed",
         );
         assert_eq!(
-            resync_subject("goal"),
-            "resource goal changed: re-read carrier"
+            subject,
+            "declaration · alpha=+declared; beta=changed; charlie=-declared [declaration]"
         );
+        assert!(subject.chars().count() <= 96);
+        assert!(!subject.contains("delta"));
     }
 
     fn owner_incarnation(seed: u64) -> crate::event::StreamOwnerIncarnation {
@@ -1612,13 +1835,135 @@ mod tests {
             .collect()
     }
 
+    fn event_body(event: &str) -> serde_json::Value {
+        let body = event
+            .lines()
+            .rev()
+            .find(|line| line.starts_with('{'))
+            .expect("JSON resync body");
+        serde_json::from_str(body).expect("valid JSON resync body")
+    }
+
     fn event_field(event: &str, field: &str) -> String {
-        event
+        if let Some(value) = event
             .lines()
             .find_map(|line| line.strip_prefix(&format!("{field}: ")))
+        {
+            return value.to_owned();
+        }
+        let body = event_body(event);
+        if matches!(field, "old" | "new") {
+            let digest = body["facts"]
+                .as_array()
+                .and_then(|facts| facts.iter().find(|fact| fact["key"] == "digest"))
+                .expect("digest transition fact");
+            let value = if field == "old" {
+                &digest["before"]
+            } else {
+                &digest["after"]
+            };
+            return value.as_str().expect("digest fact value").to_owned();
+        }
+        body[field]
+            .as_str()
             .unwrap_or_else(|| panic!("missing {field} in event"))
             .to_owned()
     }
+    #[test]
+    fn declaration_facts_are_ordered_added_removed_and_semantically_changed_labels() {
+        let root = tempfile::tempdir().unwrap();
+        let declaration = root.path().join("agent.kdl");
+        std::fs::write(
+            &declaration,
+            r#"agent "worker" {
+  host "host"
+  command "true"
+  resource "inactive" uri="file:///inactive" reason="kept"
+  resource "reason" uri="file:///reason" reason="before"
+  resource "removed" uri="file:///removed" reason="gone"
+  resource "uri" uri="file:///before" reason="same"
+}"#,
+        )
+        .unwrap();
+        let before = declaration_summary(&discover(root.path()));
+        std::fs::write(
+            &declaration,
+            r#"agent "worker" {
+  host "host"
+  command "true"
+  resource "added" uri="file:///added" reason="new"
+  resource "inactive" uri="file:///inactive" reason="kept" inactive-reason="paused"
+  resource "reason" uri="file:///reason" reason="after"
+  resource "uri" uri="file:///after" reason="same"
+}"#,
+        )
+        .unwrap();
+        let after = declaration_summary(&discover(root.path()));
+        let facts = declaration_transition_facts(
+            Some(&before),
+            Some(&after),
+            &CarrierState::Present("before-digest".to_owned()),
+            &CarrierState::Present("after-digest".to_owned()),
+        );
+        assert_eq!(
+            facts.iter().map(ResourceFact::key).collect::<Vec<_>>(),
+            vec!["added", "inactive", "reason", "removed", "uri"]
+        );
+        assert_eq!(facts[0].before(), Some(None));
+        assert_eq!(facts[0].after(), Some(Some("declared")));
+        for index in [1, 2, 4] {
+            assert_eq!(facts[index].before(), None);
+            assert_eq!(facts[index].after(), Some(Some("changed")));
+        }
+        assert_eq!(facts[3].before(), Some(Some("declared")));
+        assert_eq!(facts[3].after(), Some(None));
+    }
+
+    #[test]
+    fn declaration_parse_failure_retains_a_digest_fact_for_later_delivery() {
+        let root = tempfile::tempdir().unwrap();
+        let agent_dir = root.path().join("agents/host/worker");
+        std::fs::create_dir_all(&agent_dir).unwrap();
+        let declaration = agent_dir.join("agent.kdl");
+        let valid = r#"agent "worker" {
+  host "host"
+  command "true"
+  resource "goal" uri="resources/goal.md" reason="Mission."
+}"#;
+        std::fs::write(&declaration, valid).unwrap();
+        crate::event::publish_owner_binding_for_test(root.path(), "host").unwrap();
+        let set = watch_set_for(&discover(root.path()), "host", &Default::default());
+        let mut worker = Worker {
+            root: root.path().to_path_buf(),
+            this_host: "host".to_owned(),
+            carriers: BTreeMap::new(),
+            subscription_sequences: BTreeMap::new(),
+            deadlines: BTreeMap::new(),
+            watched: BTreeMap::new(),
+            watcher: None,
+        };
+        worker.apply_watch_sets(refresh_for(vec![set]));
+        std::fs::write(&declaration, "not an Agent Spec").unwrap();
+        worker.flush_path(&declaration, None);
+        let pending = worker.carriers[&declaration][0]
+            .pending_transition
+            .as_ref()
+            .expect("malformed declaration keeps its digest fallback");
+        assert_eq!(pending.facts.len(), 1);
+        assert_eq!(pending.facts[0].key(), "digest");
+        assert_eq!(pending.topics, ["declaration"]);
+        assert_eq!(event_body(&pending.body)["facts"].as_array().unwrap().len(), 1);
+        std::fs::write(&declaration, valid).unwrap();
+        worker.flush_path(&declaration, None);
+        let delivered = resync_inbox_event(&agent_dir);
+        let body = event_body(&delivered);
+        assert_eq!(body["binding"], "declaration");
+        assert_eq!(body["topics"], serde_json::json!(["declaration"]));
+        assert_eq!(body["facts"][0]["key"], "digest");
+        assert!(!delivered.contains("file:///"));
+        assert!(!delivered.contains("Mission."));
+    }
+
 
     #[test]
     fn watch_set_covers_declaration_and_local_bindings_only() {
@@ -1755,6 +2100,7 @@ mod tests {
                     class: CarrierClass::Immediate,
                     containment_root: None,
                     state: Some(CarrierState::Present("alpha-before".to_owned())),
+                    declaration_summary: None,
                     occurrence_sequence: 4,
                     pending_transition: None,
                     dirty: true,
@@ -1766,6 +2112,7 @@ mod tests {
                     class: CarrierClass::Coalesced,
                     containment_root: None,
                     state: Some(CarrierState::Present("beta-before".to_owned())),
+                    declaration_summary: None,
                     occurrence_sequence: 9,
                     dirty: true,
                     pending_transition: None,
@@ -1783,6 +2130,7 @@ mod tests {
                     class: CarrierClass::Immediate,
                     containment_root: None,
                 }],
+                declaration_summary: None,
             },
             AgentWatchSet {
                 declaration_path: PathBuf::from("/catalog/beta/agent.kdl"),
@@ -1794,6 +2142,7 @@ mod tests {
                     class: CarrierClass::Coalesced,
                     containment_root: None,
                 }],
+                declaration_summary: None,
             },
         ];
 
@@ -1850,6 +2199,7 @@ mod tests {
                 class: CarrierClass::Coalesced,
                 containment_root: None,
                 state: Some(CarrierState::Present("old-digest".to_owned())),
+                declaration_summary: None,
                 occurrence_sequence: 3,
                 pending_transition: None,
                 dirty: true,
@@ -1919,6 +2269,7 @@ mod tests {
                     class: CarrierClass::Immediate,
                     containment_root: None,
                     state: Some(CarrierState::Present("old-digest".to_owned())),
+                    declaration_summary: None,
                     occurrence_sequence: 0,
                     pending_transition: None,
                     dirty: true,
@@ -1957,11 +2308,11 @@ mod tests {
             .expect("the retry routes through the refreshed recipient");
         assert!(event.contains(&format!("event-id: {}", pending.event_id)), "{event}");
         assert!(event.contains(&pending.body), "{event}");
-        assert!(pending.body.contains(&format!("path: {}", old_path.display())));
+        let pending_body: serde_json::Value = serde_json::from_str(&pending.body).unwrap();
+        assert_eq!(pending_body["binding"], "goal");
+        assert_eq!(pending_body["facts"][0]["before"], "old-digest");
         assert!(
-            !pending
-                .body
-                .contains(&format!("path: {}", current_path.display())),
+            !pending.body.contains(&current_path.display().to_string()),
             "rebinding must not rewrite bytes reserved under the pending event identity"
         );
         let entry = &worker.carriers[&current_path][0];
@@ -1995,6 +2346,7 @@ mod tests {
                     class: CarrierClass::Immediate,
                     containment_root: None,
                     state: read_state(&old_path, None).ok(),
+                    declaration_summary: None,
                     occurrence_sequence: 0,
                     pending_transition: None,
                     dirty: false,
@@ -2016,6 +2368,7 @@ mod tests {
                 class: CarrierClass::Immediate,
                 containment_root: None,
             }],
+            declaration_summary: None,
         }]));
 
         assert!(!worker.carriers.contains_key(&old_path));
@@ -2045,6 +2398,7 @@ mod tests {
                     class: CarrierClass::Immediate,
                     containment_root: None,
                     state: read_state(&carrier, None).ok(),
+                    declaration_summary: None,
                     occurrence_sequence: 0,
                     pending_transition: None,
                     dirty: true,
@@ -2066,6 +2420,7 @@ mod tests {
                     class,
                     containment_root: None,
                 }],
+                declaration_summary: None,
             }])
         };
 
@@ -2099,6 +2454,7 @@ mod tests {
                     class: CarrierClass::Immediate,
                     containment_root: None,
                     state: Some(CarrierState::Present("before".to_owned())),
+                    declaration_summary: None,
                     occurrence_sequence: 1,
                     pending_transition: Some(PendingTransition::new(
                         "declaration",
@@ -2183,6 +2539,7 @@ mod tests {
                     class: CarrierClass::Immediate,
                     containment_root: None,
                     state: Some(CarrierState::Present("old-digest".to_owned())),
+                    declaration_summary: None,
                     occurrence_sequence: 1,
                     pending_transition: Some(PendingTransition::new(
                         "goal",
@@ -2213,8 +2570,8 @@ mod tests {
             .map(|entry| std::fs::read_to_string(entry.unwrap().path()).unwrap())
             .find(|body| body.contains("stream: resync"))
             .expect("pending transition is replayed");
-        assert!(event.contains("old: old-digest"), "{event}");
-        assert!(event.contains("new: pending-target"), "{event}");
+        assert_eq!(event_field(&event, "old"), "old-digest");
+        assert_eq!(event_field(&event, "new"), "pending-targ");
         let entry = &worker.carriers[&carrier][0];
         assert_eq!(
             entry.state,
@@ -2246,6 +2603,7 @@ mod tests {
                     class: CarrierClass::Coalesced,
                     containment_root: None,
                     state: baseline.clone(),
+                    declaration_summary: None,
                     occurrence_sequence: 0,
                     pending_transition: None,
                     dirty: false,
@@ -2268,6 +2626,7 @@ mod tests {
                 class: CarrierClass::Coalesced,
                 containment_root: None,
             }],
+            declaration_summary: None,
         }]));
 
         worker.flush_due(now + IMMEDIATE_WINDOW + Duration::from_secs(1));
@@ -2332,7 +2691,7 @@ mod tests {
             .collect::<Vec<_>>();
         assert_eq!(events.len(), 1);
         assert!(events[0].contains("stream: resync"));
-        assert!(events[0].contains("binding: goal"));
+        assert!(events[0].contains(r#""binding":"goal""#));
     }
 
     #[test]
@@ -2453,6 +2812,7 @@ mod tests {
                 class,
                 containment_root: None,
                 state: state.clone(),
+                declaration_summary: None,
                 occurrence_sequence: 0,
                 pending_transition: None,
                 dirty: true,
@@ -2568,7 +2928,10 @@ mod tests {
         worker.flush_path(&carrier, None);
         let deletion = resync_inbox_events(&agent_dir);
         assert_eq!(deletion.len(), 1);
-        assert_eq!(event_field(&deletion[0], "old"), original_digest);
+        assert_eq!(
+            event_field(&deletion[0], "old"),
+            original_digest.chars().take(12).collect::<String>()
+        );
         assert_eq!(event_field(&deletion[0], "new"), "missing");
         assert!(event_field(&deletion[0], "occurrence").ends_with(":1"));
         assert_eq!(
@@ -2884,6 +3247,7 @@ mod tests {
                 class: CarrierClass::Immediate,
                 containment_root: None,
                 state: Some(CarrierState::Present("old-digest".to_owned())),
+                declaration_summary: None,
                 occurrence_sequence: 0,
                 pending_transition: None,
                 dirty: true,
@@ -2952,6 +3316,7 @@ mod tests {
                     class: CarrierClass::Immediate,
                     containment_root: None,
                     state: Some(CarrierState::Present("old-digest".to_owned())),
+                    declaration_summary: None,
                     occurrence_sequence: 0,
                     pending_transition: None,
                     dirty: true,
@@ -2970,8 +3335,8 @@ mod tests {
             .clone()
             .expect("failed transition snapshot is retained");
         assert_eq!(pending_transition.new_state, CarrierState::Missing);
-        assert!(pending_transition.body.contains("old: old-digest"));
-        assert!(pending_transition.body.contains("new: missing"));
+        assert_eq!(event_field(&pending_transition.body, "old"), "old-digest");
+        assert_eq!(event_field(&pending_transition.body, "new"), "missing");
         assert_eq!(worker.carriers[&carrier][0].occurrence_sequence, 1);
         std::fs::write(&carrier, "old bytes").unwrap();
         worker.flush_path(&carrier, None);
@@ -2992,91 +3357,34 @@ mod tests {
 
     #[test]
     fn transition_identity_covers_every_rendered_transition_dimension() {
-        let old = CarrierState::Present("old-digest".to_owned());
-        let new = CarrierState::Present("new-digest".to_owned());
-        let baseline = render_body(
-            "goal",
-            Path::new("/agent/goal.md"),
-            &old,
-            &new,
-            "v1:1:2:42:3:1",
-        );
+        let topics = vec!["content".to_owned()];
+        let facts =
+            vec![ResourceFact::transition("digest", Some("old"), Some("new")).unwrap()];
+        let baseline = render_body("goal", &topics, &facts, "v1:1:2:42:3:1");
         assert_eq!(
             transition_identity(&baseline),
             transition_identity(&baseline),
             "replaying one canonical body must reproduce its identity"
         );
 
+        let changed_facts =
+            vec![ResourceFact::transition("digest", Some("old"), Some("other")).unwrap()];
         for (dimension, changed) in [
             (
                 "binding",
-                render_body(
-                    "spec",
-                    Path::new("/agent/goal.md"),
-                    &old,
-                    &new,
-                    "v1:1:2:42:3:1",
-                ),
-            ),
-            (
-                "path",
-                render_body(
-                    "goal",
-                    Path::new("/other/goal.md"),
-                    &old,
-                    &new,
-                    "v1:1:2:42:3:1",
-                ),
+                render_body("spec", &topics, &facts, "v1:1:2:42:3:1"),
             ),
             (
-                "old state",
-                render_body(
-                    "goal",
-                    Path::new("/agent/goal.md"),
-                    &CarrierState::Present("other-old".to_owned()),
-                    &new,
-                    "v1:1:2:42:3:1",
-                ),
+                "topic",
+                render_body("goal", &["other".to_owned()], &facts, "v1:1:2:42:3:1"),
             ),
             (
-                "missing old state",
-                render_body(
-                    "goal",
-                    Path::new("/agent/goal.md"),
-                    &CarrierState::Missing,
-                    &new,
-                    "v1:1:2:42:3:1",
-                ),
-            ),
-            (
-                "new state",
-                render_body(
-                    "goal",
-                    Path::new("/agent/goal.md"),
-                    &old,
-                    &CarrierState::Present("other-new".to_owned()),
-                    "v1:1:2:42:3:1",
-                ),
-            ),
-            (
-                "missing new state",
-                render_body(
-                    "goal",
-                    Path::new("/agent/goal.md"),
-                    &old,
-                    &CarrierState::Missing,
-                    "v1:1:2:42:3:1",
-                ),
+                "fact",
+                render_body("goal", &topics, &changed_facts, "v1:1:2:42:3:1"),
             ),
             (
                 "occurrence",
-                render_body(
-                    "goal",
-                    Path::new("/agent/goal.md"),
-                    &old,
-                    &new,
-                    "v1:1:2:42:3:2",
-                ),
+                render_body("goal", &topics, &facts, "v1:1:2:42:3:2"),
             ),
         ] {
             assert_ne!(
diff --git a/src/run.rs b/src/run.rs
index f1704311..afa2e775 100644
--- a/src/run.rs
+++ b/src/run.rs
@@ -4504,7 +4504,7 @@ mod tests {
         std::fs::write(&live_goal, "changed while compile failed\n").unwrap();
         let first_event = wait_for_resync_event(&live_dir)
             .expect("the already-live valid seat must stay watched across the compile error");
-        assert!(first_event.contains("binding: goal"), "{first_event}");
+        assert!(first_event.contains(r#""binding":"goal""#), "{first_event}");
 
         std::fs::write(&broken_goal, "invalid seat changed\n").unwrap();
         std::thread::sleep(Duration::from_millis(750));
@@ -4547,7 +4547,7 @@ mod tests {
         );
         let corrected_event = wait_for_resync_event_change(&live_dir, &first_event)
             .expect("correcting another declaration must not reseed and hide the live transition");
-        assert!(corrected_event.contains("binding: goal"), "{corrected_event}");
+        assert!(corrected_event.contains(r#""binding":"goal""#), "{corrected_event}");
     }
 
     #[test]
@@ -4616,7 +4616,7 @@ mod tests {
         std::fs::write(&dormant_goal, "unwatched while materialization failed\n").unwrap();
         let first_event = wait_for_resync_event(&live_dir)
             .expect("the observed live seat must remain watched through materialization failure");
-        assert!(first_event.contains("binding: goal"), "{first_event}");
+        assert!(first_event.contains(r#""binding":"goal""#), "{first_event}");
         std::thread::sleep(Duration::from_millis(750));
         assert!(
             current_resync_event(&dormant_dir).is_none(),
@@ -4645,7 +4645,7 @@ mod tests {
         let recovered_event = wait_for_resync_event_change(&live_dir, &first_event)
             .expect("recovery must preserve the pending transition instead of silently reseeding");
         assert!(
-            recovered_event.contains("binding: goal"),
+            recovered_event.contains(r#""binding":"goal""#),
             "{recovered_event}"
         );
     }
@@ -4798,7 +4798,7 @@ mod tests {
 
         let event = wait_for_resync_event(&first_dir)
             .expect("the first seat must observe a carrier transition during the later launch");
-        assert!(event.contains("binding: goal"), "{event}");
+        assert!(event.contains(r#""binding":"goal""#), "{event}");
     }
 
     #[test]
@@ -4917,7 +4917,7 @@ mod tests {
         std::fs::write(&goal, "changed after replacement launch\n").unwrap();
         let event = wait_for_resync_event(&agent_dir)
             .expect("the successful replacement must receive a fresh silent baseline");
-        assert!(event.contains("binding: goal"), "{event}");
+        assert!(event.contains(r#""binding":"goal""#), "{event}");
     }
 
     #[test]
@@ -4990,7 +4990,7 @@ mod tests {
         );
         let event = wait_for_resync_event(&agent_dir)
             .expect("the companion launch and final refresh must preserve the canonical baseline");
-        assert!(event.contains("binding: goal"), "{event}");
+        assert!(event.contains(r#""binding":"goal""#), "{event}");
     }
 
     #[test]
diff --git a/tests/resource_profile_supervisor_e2e.rs b/tests/resource_profile_supervisor_e2e.rs
index 1f976854..8a980597 100755
--- a/tests/resource_profile_supervisor_e2e.rs
+++ b/tests/resource_profile_supervisor_e2e.rs
@@ -9,8 +9,8 @@ use std::path::{Path, PathBuf};
 use std::time::{Duration, Instant};
 
 use st2::resource_profile::{
-    BindingId, HostMessage, RegistrationToken, RuntimeHealthState, RuntimeMessage, RuntimeOwner,
-    SnapshotBytes, decode_host_line, encode_runtime_line,
+    BindingId, HostMessage, RegistrationToken, ResourceFact, RuntimeHealthState, RuntimeMessage,
+    RuntimeOwner, SnapshotBytes, decode_host_line, encode_runtime_line,
 };
 use st2::resource_profile_supervisor::ResourceProfileSupervisor;
 
@@ -67,6 +67,9 @@ impl RuntimeControl {
             media_type: MEDIA_TYPE.to_owned(),
             bytes: SnapshotBytes::new(bytes.to_vec()).unwrap(),
             topics: topics.iter().map(|topic| (*topic).to_owned()).collect(),
+            facts: Some(vec![
+                ResourceFact::current("revision", health_marker).unwrap(),
+            ]),
             observed_at: None,
         };
         let health = RuntimeMessage::Health {
@@ -220,6 +223,16 @@ fn observable_publication_reaches_builtin_resync_with_filter_catch_up_and_scope_
         1,
         "the first selected publication must create one built-in resync record"
     );
+    assert!(
+        first_inbox[0].contains("subject: observed · revision=primary-first [selected]"),
+        "{}",
+        first_inbox[0]
+    );
+    assert!(
+        first_inbox[0].contains(r#""facts":[{"key":"revision","after":"primary-first"}]"#),
+        "{}",
+        first_inbox[0]
+    );
 
     primary
         .runtime
@@ -274,6 +287,11 @@ fn observable_publication_reaches_builtin_resync_with_filter_catch_up_and_scope_
         caught_up_inbox, first_inbox,
         "restoring delivery must replace the old head with the pending digest"
     );
+    assert!(
+        caught_up_inbox[0].contains("revision=delivery-unavailable"),
+        "{}",
+        caught_up_inbox[0]
+    );
     let caught_up_projection = file_tree(&primary.agent_dir.join("resources"));
     primary.refresh(&primary_supervisor);
     assert_eq!(
@@ -391,7 +409,7 @@ fn file_tree(root: &Path) -> Vec<(PathBuf, Vec<u8>)> {
 }
 
 fn observable_resolver_wasm() -> Vec<u8> {
-    const DESCRIPTOR: &[u8] = br#"{"abiVersion":2,"capabilities":["resolve","read","observe"],"selectorSchema":{"type":"object","properties":{"topics":{"type":"array","items":{"type":"string"},"uniqueItems":true}},"required":["topics"],"additionalProperties":false},"defaultSelector":{"topics":["selected"]},"topics":[{"name":"selected"},{"name":"ignored"}],"runtime":{"topology":"shared"},"snapshot":{"mediaType":"application/json","schemaId":"dev.example.observable.snapshot.v1"}}"#;
+    const DESCRIPTOR: &[u8] = br#"{"abiVersion":3,"capabilities":["resolve","read","observe"],"selectorSchema":{"type":"object","properties":{"topics":{"type":"array","items":{"type":"string"},"uniqueItems":true}},"required":["topics"],"additionalProperties":false},"defaultSelector":{"topics":["selected"]},"topics":[{"name":"selected"},{"name":"ignored"}],"runtime":{"topology":"shared"},"snapshot":{"mediaType":"application/json","schemaId":"dev.example.observable.snapshot.v1"}}"#;
     const RESOLUTION: &[u8] = br#"{"path":"resources/snapshot.json","class":"observable"}"#;
     const DESCRIPTOR_PTR: i64 = 1024;
     const RESOLUTION_PTR: i64 = 4096;
diff --git a/tests/resync.rs b/tests/resync.rs
index 6cde6bb7..81a27e3d 100644
--- a/tests/resync.rs
+++ b/tests/resync.rs
@@ -95,8 +95,8 @@ fn carrier_change_emits_one_superseded_resync_event_and_silent_stores_stay_quiet
         "goal change must produce exactly one resync event within the immediate window"
     );
     let body = &resync_events(&agent_dir)[0];
-    assert!(body.contains("resource goal changed"), "{body}");
-    assert!(body.contains("binding: goal"), "{body}");
+    assert!(body.contains("subject: goal · digest="), "{body}");
+    assert!(body.contains(r#""binding":"goal""#), "{body}");
     let legitimate_event_id = body
         .lines()
         .find(|line| line.starts_with("event-id:"))
@@ -153,7 +153,7 @@ fn carrier_change_emits_one_superseded_resync_event_and_silent_stores_stay_quiet
         || {
             resync_events(&agent_dir)
                 .iter()
-                .filter(|b| !b.contains(&first_event_id) && b.contains("binding: goal"))
+                .filter(|b| !b.contains(&first_event_id) && b.contains(r#""binding":"goal""#))
                 .count()
         },
         1
@@ -189,7 +189,7 @@ fn whole_file_declaration_replacement_by_rename_notifies_immediately() {
             || {
                 resync_events(&agent_dir)
                     .iter()
-                    .filter(|b| b.contains("binding: declaration"))
+                    .filter(|b| b.contains(r#""binding":"declaration""#))
                     .count()
             },
             1
@@ -283,7 +283,7 @@ fn declared_wasm_profile_resolves_a_scheme_uri_goal_binding_and_fires_on_change(
         resync_events(&agent_dir)
     );
     let body = &resync_events(&agent_dir)[0];
-    assert!(body.contains("resource goal changed"), "{body}");
+    assert!(body.contains("subject: goal · digest="), "{body}");
 }
 
 #[test]