feat(resource): add demand observation protocol

agent-identity: dev3.direct.omp.536sbpvb
agent-persona: generalist
agent-supervisor: unavailable
agent-tool: OMP
agent-tool-version: 18.0.11
agent-runtime: OMP 18.0.11
tooling-profile: dotfiles@000f2b3

Author: schickling-assistant

INVARIANTS.md

@@ -34,4 +34,5 @@ materialization, messaging, DING, or presence must preserve them.
 | **Parked tasks are visible and individually recoverable** | A parked task is reported alongside an unmodified runtime observation as a complete known fault; only an unbelievable marker fails closed. Park markers and unpark requests share the exact canonical catalog-folder plus host ownership scope, and the projected recovery argv carries both axes, so same-host supervisors cannot see, delete, consume, or advertise recovery into each other's channels even for the same task ID. A projected park whose supervisor generation is gone is positively not parked. An explicit per-task unpark clears that task's park and spent budget so it is launchable again and stays recovered past `interval`, releases no other parked task, restarts no healthy peer, and restores the agent's derived DING. | `src/flapping.rs::unpark_restores_a_launchable_task_not_just_a_cleared_flag`; `src/flapping.rs::unpark_is_per_task_and_reports_whether_it_changed_anything`; `src/park.rs::same_host_supervisors_isolate_markers_and_requests_by_catalog`; `src/park.rs::a_marker_from_a_dead_supervisor_reads_as_not_parked`; `src/park.rs::published_parks_are_readable_and_clear_when_the_task_recovers`; `src/park.rs::an_unbelievable_marker_is_indeterminate_not_absent`; `src/park.rs::a_request_is_consumed_exactly_once`; `src/task_inventory.rs::a_parked_task_reports_its_fault_alongside_a_truthful_runtime_state`; `src/task_inventory.rs::an_unbelievable_park_marker_makes_the_envelope_incomplete`; `tests/task_inventory_cli.rs::projected_recovery_targets_its_exact_catalog_and_host_despite_ambient_defaults`; `tests/run.rs::an_operator_recovers_one_parked_task_without_disturbing_a_healthy_peer`; `tests/run.rs::an_unpark_request_for_a_task_that_is_not_parked_says_so` |
 | **Tracked workspaces fail closed** | Materialization simulates content operations before writing and refuses a real change to any Git-tracked target. Byte-identical tracked, untracked, and non-Git targets retain useful behavior. | `tests/materialize.rs::every_content_directive_refuses_to_change_a_tracked_target_before_any_write`; `tests/materialize.rs::byte_identical_tracked_target_is_allowed_without_modification`; `tests/materialize.rs::untracked_and_non_git_targets_remain_materializable` |
 | **Native flat root** | Without an authored override, catalog tasks, eval messaging, shell helpers, and DING all use the catalog itself as `ST_ROOT`; no nested bus directory is synthesized. | `src/eval_run.rs::bus_root_expands_st_root_else_defaults`; `tests/eval_run_e2e.rs::st2_eval_runs_a_benign_folder_to_a_pass_verdict`; `tests/pty.rs` |
+| **Demand observation is declared, fenced, settled, and clock-free** | `st2 resource refresh` reaches only the resident Resource Profile runtime and only when its catalog runtime declares `capability "demand"`. Every observe dispatch and settlement carries the exact owner, binding ID, and registration; one outstanding dispatch plus one latest trailing watermark coalesces bursts without losing in-flight arrivals. Only exact unchanged, published, failed, stale-generation, or provider-unavailable evidence closes accepted work. Writer backpressure retains queued demand, restart rescans after installing the watch, client disconnect does not cancel it, and the CLI wait bound never participates in correctness or provider scheduling. | `crates/st2-resource-protocol/src/lib.rs::host_frames_have_exact_json_shape_and_newline`; `crates/st2-resource-protocol/src/lib.rs::runtime_frames_have_exact_json_shape_and_padded_base64`; `src/resource_profile_supervisor.rs::demand_batches_coalesce_bursts_and_keep_one_trailing_watermark`; `src/resource_profile_supervisor.rs::full_writer_queue_is_reported_without_consuming_the_frame`; `tests/resource_profile_supervisor_e2e.rs::demand_observation_is_watermark_settled_coalesced_and_restart_safe`; `tests/agent_resource.rs::refresh_cli_reports_exact_receipts_and_wait_expiry_keeps_the_request` |
 | **Proof references resolve** | Every qualified test named in this table exists in its named source file, so stale invariant claims fail the suite instead of silently surviving a refactor. | `tests/invariants.rs::qualified_proof_references_resolve` |

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

@@ -13,6 +13,7 @@ pub const MAX_PROTOCOL_LINE_BYTES: usize = 2 * 1024 * 1024;
 pub const MAX_SNAPSHOT_BYTES: usize = 1024 * 1024;
 pub const MAX_SELECTOR_BYTES: usize = 16 * 1024;
 pub const MAX_HEALTH_DETAIL_BYTES: usize = 16 * 1024;
+pub const MAX_OBSERVATION_DIAGNOSTIC_BYTES: usize = 16 * 1024;
 const MAX_OPAQUE_ID_BYTES: usize = 16 * 1024;
 
 #[derive(Debug, Clone, PartialEq, Eq)]
@@ -213,7 +214,7 @@ fn encode_base64(bytes: &[u8]) -> String {
 }
 
 fn decode_base64(encoded: &str) -> Result<Vec<u8>, Base64Error> {
-    if encoded.len() % 4 != 0 {
+    if !encoded.len().is_multiple_of(4) {
         return Err(Base64Error("base64 length is not a multiple of four"));
     }
     let maximum_encoded = MAX_SNAPSHOT_BYTES.div_ceil(3) * 4;
@@ -372,6 +373,11 @@ pub enum HostMessage {
         binding_id: BindingId,
         registration: RegistrationToken,
     },
+    Observe {
+        owner: RuntimeOwner,
+        binding_id: BindingId,
+        registration: RegistrationToken,
+    },
 }
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
@@ -383,6 +389,14 @@ pub enum RuntimeHealthState {
     Failed,
 }
 
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub enum ObservationOutcome {
+    Unchanged,
+    Published,
+    Failed,
+}
+
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
 #[serde(
     tag = "type",
@@ -412,6 +426,17 @@ pub enum RuntimeMessage {
         #[serde(skip_serializing_if = "Option::is_none")]
         detail: Option<String>,
     },
+    ObservationSettled {
+        owner: RuntimeOwner,
+        binding_id: BindingId,
+        registration: RegistrationToken,
+        demand_watermark: u64,
+        outcome: ObservationOutcome,
+        #[serde(skip_serializing_if = "Option::is_none")]
+        digest: Option<SnapshotDigest>,
+        #[serde(skip_serializing_if = "Option::is_none")]
+        diagnostic: Option<String>,
+    },
 }
 
 #[derive(Debug)]
@@ -422,6 +447,8 @@ pub enum ProtocolError {
     LineTooLarge { actual: usize },
     SelectorTooLarge { actual: usize },
     HealthDetailTooLarge { actual: usize },
+    ObservationDiagnosticTooLarge { actual: usize },
+    InvalidDemandWatermark,
     InvalidTopics(&'static str),
     InvalidHealthScope,
     Json(serde_json::Error),
@@ -445,6 +472,13 @@ impl fmt::Display for ProtocolError {
                 formatter,
                 "health detail is {actual} bytes; maximum is {MAX_HEALTH_DETAIL_BYTES}"
             ),
+            Self::ObservationDiagnosticTooLarge { actual } => write!(
+                formatter,
+                "observation diagnostic is {actual} bytes; maximum is {MAX_OBSERVATION_DIAGNOSTIC_BYTES}"
+            ),
+            Self::InvalidDemandWatermark => {
+                formatter.write_str("observation demand watermark must be positive")
+            }
             Self::InvalidTopics(reason) => write!(formatter, "invalid topics: {reason}"),
             Self::InvalidHealthScope => formatter
                 .write_str("binding-scoped health must carry both bindingId and registration"),
@@ -544,6 +578,23 @@ fn validate_runtime_message(message: &RuntimeMessage) -> Result<(), ProtocolErro
             }
             Ok(())
         }
+        RuntimeMessage::ObservationSettled {
+            demand_watermark,
+            diagnostic,
+            ..
+        } => {
+            if *demand_watermark == 0 {
+                return Err(ProtocolError::InvalidDemandWatermark);
+            }
+            if let Some(diagnostic) = diagnostic
+                && diagnostic.len() > MAX_OBSERVATION_DIAGNOSTIC_BYTES
+            {
+                return Err(ProtocolError::ObservationDiagnosticTooLarge {
+                    actual: diagnostic.len(),
+                });
+            }
+            Ok(())
+        }
     }
 }
 
@@ -611,6 +662,16 @@ mod tests {
             encode_host_line(&unregister).unwrap(),
             b"{\"type\":\"unregister\",\"owner\":{\"incarnation\":\"incarnation\",\"claim\":\"claim\"},\"bindingId\":\"binding\",\"registration\":\"registration\"}\n"
         );
+
+        let observe = HostMessage::Observe {
+            owner: owner(),
+            binding_id: BindingId::new("binding").unwrap(),
+            registration: RegistrationToken::new("registration").unwrap(),
+        };
+        assert_eq!(
+            encode_host_line(&observe).unwrap(),
+            b"{\"type\":\"observe\",\"owner\":{\"incarnation\":\"incarnation\",\"claim\":\"claim\"},\"bindingId\":\"binding\",\"registration\":\"registration\"}\n"
+        );
     }
 
     #[test]
@@ -630,6 +691,19 @@ mod tests {
             encode_runtime_line(&health).unwrap(),
             b"{\"type\":\"health\",\"owner\":{\"incarnation\":\"incarnation\",\"claim\":\"claim\"},\"state\":\"ready\"}\n"
         );
+        let settled = RuntimeMessage::ObservationSettled {
+            owner: owner(),
+            binding_id: BindingId::new("binding").unwrap(),
+            registration: RegistrationToken::new("registration").unwrap(),
+            demand_watermark: 7,
+            outcome: ObservationOutcome::Published,
+            digest: Some(SnapshotDigest::of(b"one byte")),
+            diagnostic: None,
+        };
+        assert_eq!(
+            encode_runtime_line(&settled).unwrap(),
+            b"{\"type\":\"observationSettled\",\"owner\":{\"incarnation\":\"incarnation\",\"claim\":\"claim\"},\"bindingId\":\"binding\",\"registration\":\"registration\",\"demandWatermark\":7,\"outcome\":\"published\",\"digest\":\"23c70185c68c8fa6326887db7feb4bb951c1cad2a7b2a9acf70c3d2e466aced0\"}\n"
+        );
         assert_eq!(
             decode_runtime_line(&encode_runtime_line(&publish(b"one byte")).unwrap()).unwrap(),
             publish(b"one byte")
@@ -658,6 +732,16 @@ mod tests {
             decode_host_line(uppercase_digest),
             Err(ProtocolError::Json(_))
         ));
+        let unknown_settlement_field = b"{\"type\":\"observationSettled\",\"owner\":{\"incarnation\":\"i\",\"claim\":\"c\"},\"bindingId\":\"b\",\"registration\":\"r\",\"demandWatermark\":1,\"outcome\":\"unchanged\",\"extra\":true}\n";
+        assert!(matches!(
+            decode_runtime_line(unknown_settlement_field),
+            Err(ProtocolError::Json(_))
+        ));
+        let zero_watermark = b"{\"type\":\"observationSettled\",\"owner\":{\"incarnation\":\"i\",\"claim\":\"c\"},\"bindingId\":\"b\",\"registration\":\"r\",\"demandWatermark\":0,\"outcome\":\"failed\"}\n";
+        assert!(matches!(
+            decode_runtime_line(zero_watermark),
+            Err(ProtocolError::InvalidDemandWatermark)
+        ));
     }
 
     #[test]
@@ -758,6 +842,19 @@ mod tests {
             decode_runtime_line(&health_line),
             Err(ProtocolError::HealthDetailTooLarge { .. })
         ));
+        let settlement = RuntimeMessage::ObservationSettled {
+            owner: owner(),
+            binding_id: BindingId::new("binding").unwrap(),
+            registration: RegistrationToken::new("registration").unwrap(),
+            demand_watermark: 1,
+            outcome: ObservationOutcome::Failed,
+            digest: None,
+            diagnostic: Some("x".repeat(MAX_OBSERVATION_DIAGNOSTIC_BYTES + 1)),
+        };
+        assert!(matches!(
+            encode_runtime_line(&settlement),
+            Err(ProtocolError::ObservationDiagnosticTooLarge { .. })
+        ));
         assert!(RuntimeIncarnation::new("x".repeat(MAX_OPAQUE_ID_BYTES + 1)).is_err());
     }
 

docs/vrs/.decisions/0014-resource-profiles-are-state-first-read-and-observe-capabilities.md

@@ -22,6 +22,11 @@ Two prototypes bounded the design. A synthetic 16-observation pull-request seque
 6. The descriptor declares shared or per-binding runtime topology. Both use one normalized host protocol and the same per-binding delivery state.
 7. When delivery is unavailable, st2 retains only `pending_relevant_change` beside current and last-delivered digests. Resume emits at most one invalidation for current state.
 8. The initial capability set stops at read and observe. Provider mutations, actions, approvals, and a canonical event log require separate research and design.
+9. st2 may carry a generic declared, fenced demand-observation scheduling hint
+   to the resident profile runtime. The profile still owns polling, push,
+   subscription, provider mechanism, cursor, rate limits, cache, and backoff.
+   Demand is neither a provider-specific reconcile command nor a provider
+   write, and exact settlement evidence rather than a host clock closes it.
 
 ## Options
 
@@ -71,7 +76,40 @@ claim, each binding registration receives a token, and host acceptance requires
 both to match. A new claim fences all prior output and clears registrations.
 Shared and per-binding topologies use the same reducer.
 
-The normalized wire protocol contains only `register`, `unregister`, `publish`,
-and `health`. EOF and existing supervisor lifecycle replace `shutdown`;
-implementation-owned observation replaces host `reconcile`. Evidence:
+The initial normalized wire protocol contained only `register`, `unregister`,
+`publish`, and `health`. EOF and existing supervisor lifecycle replace
+`shutdown`; implementation-owned observation replaces provider-specific host
+`reconcile`. Evidence:
 [selector and runtime protocol prototype](../07-resource-profile/.experiments/2026-08-29-selector-and-runtime-protocol-prototype.md).
+
+## Amendment 2: declared demand observation
+
+Generic demand observation extends the same state-first read boundary without
+transferring observation ownership to st2. A catalog runtime opts into one
+atomic `demand` capability. Only then may the host send fenced `observe`; the
+runtime answers a demand-initiated cycle with fenced `observationSettled`
+carrying its positive demand watermark and exact `unchanged`, `published`, or
+`failed` outcome. The gate protects older strict runtimes from both halves of
+the vocabulary.
+
+The resident Resource Profile supervisor remains the sole provider-process
+owner. It resolves the exact active owner, binding ID, and registration,
+retains one outstanding dispatch and one latest trailing batch, and closes work
+only from settlement, re-registration, or provider-process failure evidence.
+There is no correctness timer. A client wait bound does not cancel, retract, or
+duplicate accepted demand.
+
+The scheduling state is level-triggered: demand arriving during an observation
+survives completion and collapses into one later cycle. The runtime remains
+free to defer that cycle under its own backoff and retains every provider
+cursor, conditional cache, rate-limit decision, and normal observation
+schedule. Thus `observe` is not the rejected host `reconcile`; it cannot name a
+provider action, reset provider state, or mutate the Resource.
+
+The local client channel uses bounded strict atomic request and receipt records
+inside the existing private `(catalog, host)` supervisor scope. The watch is
+installed before startup scan and only final basenames are consumed. Records
+are intentionally not fsynced: power loss may yield missing evidence and a
+retry, never false success. Exact wire, coalescing, trailing demand,
+re-registration, restart, disconnect, capability-gate, and CLI outcome proofs
+live with the implementation.

docs/vrs/07-resource-profile/requirements.md

@@ -190,12 +190,17 @@ one latest-state catch-up. The direction is recorded in
   authority after missed, duplicated, reordered, or coalesced provider
   observations.
 - **PROFILE-R15 Implementation-owned observation:** A profile implementation
-  chooses polling, push, native subscription, or a hybrid and may retain its own
-  provider cursor. st2 standardizes registration, atomic snapshot publication,
-  backpressure, cancellation, and health outcomes but does not send
-  observation-specific reconcile commands or prescribe the provider mechanism.
-  Provider payloads never bypass snapshot publication to become canonical
-  delivery records.
+  chooses polling, push, native subscription, or a hybrid and retains its own
+  provider mechanism, cursor, rate-limit state, cache, backoff, and repair
+  policy. A catalog runtime may separately declare the generic fenced demand
+  capability. When declared, st2 may issue a coalesced demand-observation
+  scheduling hint and receive exact unchanged, published, or failed settlement
+  evidence; the hint never prescribes how the observation occurs, resets
+  provider state, or becomes a provider-specific reconcile command. st2
+  standardizes registration, atomic snapshot publication, backpressure,
+  cancellation, demand fencing and settlement, and health outcomes. Provider
+  payloads never bypass snapshot publication to become canonical delivery
+  records, and demand observation never authorizes a provider write.
 - **PROFILE-R16 Declared runtime topology:** The descriptor declares either one
   shared runtime per catalog and exact scheme or one runtime per active binding.
   Both modes use one host protocol and per-binding lifecycle state. Each runtime
@@ -207,9 +212,10 @@ 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
-  binding.
+  Health detail and demand-settlement diagnostic are each at most 16 KiB of
+  UTF-8. Local demand request and receipt records are bounded, strict schemas
+  with plain final basenames. 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