Merge: digest status groups and tags, and module events that target one flow are admitted only to that flow

# Conflicts:
#	crates/basal-core/src/manifest.rs
#	mutations.toml

Author: ualtinok

crates/basal-core/src/authorize.rs

@@ -268,6 +268,32 @@ pub fn check(
             }
             Primitive::SinkDigest => {
                 let agent = agent_arg(args, *p)?;
+                if let Some(tag) = args.get("options").and_then(|o| o.get("status_tag")) {
+                    if !tag
+                        .as_str()
+                        .is_some_and(crate::manifest::valid_status_label)
+                    {
+                        return Err(Refusal::new(
+                            "status_tag_invalid",
+                            "status_tag must match [A-Z]{1,4}",
+                        ));
+                    }
+                    let target = if manifest.audience.is_some() {
+                        crate::manifest::AUDIENCE_TARGET
+                    } else {
+                        agent
+                    };
+                    if !manifest
+                        .sinks
+                        .iter()
+                        .any(|s| s.agent == target && s.status_group.is_some())
+                    {
+                        return Err(Refusal::new(
+                            "status_tag_without_group",
+                            "a status_tag requires the sink grant's status_group",
+                        ));
+                    }
+                }
                 if manifest.audience.is_some() {
                     return if manifest.sinks.is_empty() {
                         Err(Refusal::denied("the manifest grants no digest sink"))

crates/basal-core/src/events.rs

@@ -31,7 +31,25 @@ pub struct Notice {
     pub published_at_ms: i64,
 }
 
+// EventDeclaration currently has no flow-targeting field. Keep the known
+// flow-targeted event contracts here so ingress, fan-out and body replay use
+// the same rule without inventing a declaration wire field.
+const FLOW_TARGETED_EVENTS: &[(&str, &str)] = &[("prefrontal-core", "digest_delivered")];
+
 impl Notice {
+    pub fn target_flow(&self) -> Result<Option<&str>> {
+        let (module, name, _) = self.identity()?;
+        if !FLOW_TARGETED_EVENTS.contains(&(module, name)) {
+            return Ok(None);
+        }
+        self.headers
+            .get("flow_id")
+            .map(String::as_str)
+            .filter(|flow| !flow.is_empty())
+            .map(Some)
+            .ok_or_else(|| CoreError::Invalid("flow-targeted event lacks flow_id header".into()))
+    }
+
     pub fn identity(&self) -> Result<(&str, &str, u32)> {
         let parts: Vec<_> = self.subject.split('.').collect();
         if parts.len() != 6
@@ -77,7 +95,13 @@ pub struct Report {
     pub overflow: usize,
 }
 
-fn matches(manifest: &str, notice: &Notice) -> Result<bool> {
+fn matches(manifest: &str, notice: &Notice, flow_id: &str) -> Result<bool> {
+    if notice
+        .target_flow()?
+        .is_some_and(|target| target != flow_id)
+    {
+        return Ok(false);
+    }
     let (module, name, version) = notice.identity()?;
     let manifest = Manifest::parse(manifest).map_err(|e| CoreError::Corrupt(e.to_string()))?;
     Ok(manifest.trigger.events.as_ref().is_some_and(|events| {
@@ -183,10 +207,10 @@ impl Runtime {
                 .collect::<Result<Vec<_>>>()?;
             let mut report = Report::default();
             for notice in notices {
-                notice.identity()?;
+                notice.target_flow()?;
                 for (flow, first_approved, approved) in &approved {
                     if let Some(approved) = approved
-                        && matches(&approved.manifest, notice)?
+                        && matches(&approved.manifest, notice, flow)?
                     {
                         // Upgrades retain the first approval point, so events
                         // during an upgrade remain eligible for the current code.
@@ -222,7 +246,7 @@ impl Runtime {
                     continue;
                 }
                 let approved = install::approved(tx, &flow)?;
-                if !approved.as_ref().map(|a| matches(&a.manifest, &notice)).transpose()?.unwrap_or(false) {
+                if !approved.as_ref().map(|a| matches(&a.manifest, &notice, &flow)).transpose()?.unwrap_or(false) {
                     tx.execute("UPDATE event_receipts SET state='refused' WHERE rowid=?1", [rowid])?;
                     continue;
                 }
@@ -319,6 +343,18 @@ fn script_trigger(notice: &Notice, body: &[u8]) -> Result<JsonText> {
     .map_err(|e| CoreError::Invalid(e.to_string()))
 }
 
+fn body_targets_run(notice: &Notice, body: &[u8], flow_id: &str) -> bool {
+    match notice.target_flow() {
+        Ok(None) => true,
+        Ok(Some(target)) => {
+            target == flow_id
+                && serde_json::from_slice::<serde_json::Value>(body)
+                    .is_ok_and(|body| body["flow_id"].as_str() == Some(flow_id))
+        }
+        Err(_) => false,
+    }
+}
+
 impl Runtime {
     pub(crate) fn event_preamble(&self, run: &crate::Run, manifest: &Manifest) -> Result<Preamble> {
         use basal_host::{CallClass, CallRequest, Dispatched, TransportError};
@@ -350,6 +386,9 @@ impl Runtime {
             .optional()?)
         })?;
         if let Some((Some(body), _, _)) = &saved {
+            if !body_targets_run(&notice, body, &run.flow_id) {
+                return Ok(mismatch("flow-targeted body does not name the run's flow"));
+            }
             return Ok(Preamble::Ready(script_trigger(&notice, body)?));
         }
         if saved
@@ -424,6 +463,9 @@ impl Runtime {
                     Ok(body) => body,
                     Err(kind) => return Ok(mismatch(kind)),
                 };
+                if !body_targets_run(&notice, &body, &run.flow_id) {
+                    return Ok(mismatch("flow-targeted body does not name the run's flow"));
+                }
                 let trigger = script_trigger(&notice, &body)?;
                 self.store().write(|tx| {
                     tx.execute("UPDATE event_body_journal SET body=?2,digest=?3,retry_at=NULL WHERE run_id=?1", params![run.run_id, body, notice.digest])?;

crates/basal-core/src/events_tests.rs

@@ -643,3 +643,72 @@ fn queued_history_before_approval_is_refused_without_a_run() {
             .unwrap();
     assert_eq!(runs, 0);
 }
+
+#[test]
+fn flow_targeted_notice_missing_header_is_refused_without_flows() {
+    let f = Fixture::new(60);
+    let mut notice = notice(0);
+    notice.subject = "ck.account.event.prefrontal-core.digest_delivered.v1".into();
+    let error = f.rt.admit_events(&[notice]).unwrap_err();
+    assert!(error.to_string().contains("flow_id"), "{error}");
+}
+
+#[test]
+fn journaled_flow_targeted_body_is_checked_against_run() {
+    let f = Fixture::new(60);
+    f.catalog.set_flow_capable("prefrontal-core", true);
+    f.catalog.set_event(
+        "prefrontal-core",
+        "digest_delivered",
+        1,
+        basal_host::EventDecl {
+            origin: basal_host::EventOrigin::External,
+            body: basal_host::EventBody::Resolved {
+                resolve_op: basal_host::catalog::EVENT_BODY_OP.into(),
+            },
+        },
+    );
+    f.catalog.set_op(
+        "prefrontal-core",
+        basal_host::catalog::EVENT_BODY_OP,
+        OpDecl {
+            kind: Some(OpKind::Query),
+            cause_echo: false,
+            shell_capable: false,
+        },
+    );
+    let request = InstallRequest {
+        script: "return trigger.event.body.flow_id;".into(), author: "operator".into(), loop_override: false,
+        manifest: json!({"id":"flow","version":1,"purpose":"Replay delivered events",
+            "trigger":{"events":[{"module":"prefrontal-core","name":"digest_delivered","version":1}]},
+            "ops":[{"module":"prefrontal-core","op":basal_host::catalog::EVENT_BODY_OP}]}).to_string(),
+    };
+    let installed = f.rt.install(&request).unwrap();
+    f.rt.approve("flow", 1, &installed.code_hash, "test")
+        .unwrap();
+    let mut notice = notice(0);
+    notice.subject = "ck.account.event.prefrontal-core.digest_delivered.v1".into();
+    notice.headers.insert("flow_id".into(), "flow".into());
+    f.rt.admit_events(&[notice]).unwrap();
+    let (id, _) = f.rt.startable().unwrap().remove(0);
+    let run = f.rt.run(&id).unwrap();
+    let manifest = Manifest::parse(&request.manifest).unwrap();
+    for flow in ["other", "flow"] {
+        let body = json!({"flow_id":flow}).to_string().into_bytes();
+        f.rt.store().write(|tx| {
+            tx.execute("INSERT INTO event_body_journal(run_id,attempt,body) VALUES (?1,1,?2) ON CONFLICT(run_id) DO UPDATE SET body=excluded.body", params![id, body])?;
+            Ok(())
+        }).unwrap();
+        match f.rt.event_preamble(&run, &manifest).unwrap() {
+            Preamble::Failed { kind, .. } if flow == "other" => {
+                assert_eq!(kind, "event_body_mismatch")
+            }
+            Preamble::Ready(_) if flow == "flow" => {}
+            _ => panic!("unexpected preamble for {flow}"),
+        }
+    }
+    assert!(
+        f.source.calls.lock().unwrap().is_empty(),
+        "saved bodies require no live fetch"
+    );
+}

crates/basal-core/src/manifest.rs

@@ -144,6 +144,9 @@ impl DigestAction {
 pub struct SinkGrant {
     pub agent: String,
     pub digest_max: DigestAction,
+    /// Groups undelivered tagged items in the agent's status segment.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub status_group: Option<String>,
     /// Shown on the card as requested; only the operator grants it, through
     /// policy.
     #[serde(default)]
@@ -465,6 +468,10 @@ fn agent_byte(b: u8) -> bool {
     b.is_ascii_alphanumeric() || matches!(b, b'_' | b'-')
 }
 
+pub(crate) fn valid_status_label(value: &str) -> bool {
+    (1..=4).contains(&value.len()) && value.bytes().all(|b| b.is_ascii_uppercase())
+}
+
 fn check_list<T: Ord + Clone + std::fmt::Debug>(
     field: &str,
     items: impl Iterator<Item = T>,
@@ -594,6 +601,11 @@ impl Manifest {
         };
         for s in &self.sinks {
             check_sink("sinks.agent", &s.agent)?;
+            if let Some(group) = &s.status_group
+                && !valid_status_label(group)
+            {
+                return Err(invalid("sinks.status_group", "must match [A-Z]{1,4}"));
+            }
         }
         check_list("sinks", self.sinks.iter().map(|s| s.agent.clone()))?;
         for a in &self.status {
@@ -896,6 +908,33 @@ mod tests {
         }
     }
 
+    #[test]
+    fn status_group_is_one_to_four_uppercase_ascii_letters() {
+        let mut value = serde_json::json!({
+            "id":"counts", "version":1, "purpose":"Report counts",
+            "trigger":{"schedule":{"interval":"1h"}},
+            "sinks":[{"agent":"SYNAPSE","digest_max":"silent"}]
+        });
+        assert!(Manifest::parse(&value.to_string()).is_ok());
+        for group in ["A", "GH", "ABCD"] {
+            value["sinks"][0]["status_group"] = group.into();
+            let manifest = Manifest::parse(&value.to_string()).unwrap();
+            assert_eq!(manifest.sinks[0].status_group.as_deref(), Some(group));
+            assert_eq!(
+                serde_json::to_value(manifest).unwrap()["sinks"][0]["status_group"],
+                group
+            );
+        }
+        for group in ["", "ABCDE", "gh", "Gh", "G1", "G_", "É", " GH", "GH\n"] {
+            value["sinks"][0]["status_group"] = group.into();
+            assert!(
+                matches!(Manifest::parse(&value.to_string()),
+                Err(ManifestError::Invalid { field, .. }) if field == "sinks.status_group"),
+                "{group:?}"
+            );
+        }
+    }
+
     /// The platform's own grammar is the one a manifest is parsed with.
     #[test]
     fn manifest_roots_follow_this_platforms_spelling() {

crates/basal-host/src/core_host.rs

@@ -32,9 +32,13 @@ pub fn intent(kind: Primitive, args: &Value, ctx: IntentContext<'_>) -> Value {
     } = ctx;
     match kind {
         Primitive::SinkDigest => {
-            json!({"flow_id":flow_id,"flow_version":version,"run_id":run_id,"call_position":position,
+            let mut request = json!({"flow_id":flow_id,"flow_version":version,"run_id":run_id,"call_position":position,
             "agent":args["agent"],"action":args["action"],
-            "due_at":due_at,"created_at":created_at,"item":args["item"],"claim":null})
+            "due_at":due_at,"created_at":created_at,"item":args["item"],"claim":null});
+            if let Some(tag) = args.get("options").and_then(|o| o.get("status_tag")) {
+                request["status_tag"] = tag.clone();
+            }
+            request
         }
         Primitive::SinkStatus => {
             json!({"flow_id":flow_id,"flow_version":version,"agent":args["agent"],

crates/basal-module/src/events.rs

@@ -284,7 +284,7 @@ pub fn decode(message: &jetstream::Message) -> Result<Notice, String> {
         headers: labels,
         published_at_ms,
     };
-    notice.identity().map_err(|e| e.to_string())?;
+    notice.target_flow().map_err(|e| e.to_string())?;
     Ok(notice)
 }
 

crates/basal-module/tests/dry_run.rs

@@ -24,6 +24,41 @@ fn calls(run: &Value) -> Vec<Value> {
     run["calls"].as_array().cloned().unwrap_or_default()
 }
 
+#[test]
+fn status_group_is_validated_at_install_and_dry_run() {
+    use basal_module::dryrun::{DryRunRequest, DryTrigger, Mode};
+    let f = fixture("status-group-validation", Options::default());
+    for group in ["", "ABCDE", "gh", "G1", "É"] {
+        let mut manifest = schedule_manifest("flow-group", json!({"interval":"1h"}));
+        manifest["sinks"][0]["status_group"] = json!(group);
+        let manifest = manifest.to_string();
+        let error = f
+            .module
+            .handle(
+                &Caller::Operator,
+                "flow.install",
+                json!({"script":"return 1;","manifest":manifest}),
+            )
+            .unwrap_err();
+        assert!(error.message.contains("sinks.status_group"), "{error:?}");
+        let error = f
+            .module
+            .dry
+            .run(&DryRunRequest {
+                script: "return 1;".into(),
+                manifest,
+                author: "operator".into(),
+                loop_override: false,
+                mode: Mode::Capture,
+                trigger: DryTrigger::Synthetic(json!({})),
+                now_ms: common::T0,
+            })
+            .unwrap_err();
+        assert!(error.to_string().contains("sinks.status_group"), "{error}");
+    }
+    assert_eq!(f.mock.total_sends(), 0);
+}
+
 #[test]
 fn capture_mode_executes_no_host_call_and_replays_the_schedule_window() {
     let f = fixture("dry-capture", Options::default());