fix(stream): serialize admission with catalog edits

Author: schickling

src/catalog_transaction.rs

@@ -600,7 +600,7 @@ fn agent_semantic_deltas(
 }
 
 fn normalize_agent(spec: &agent_spec::AgentSpec) -> Result<BTreeMap<String, SemanticAtom>> {
-    use agent_spec::{Restart, RestartMode, TaskKind, TaskLifecycle};
+    use agent_spec::{Restart, RestartMode, StreamLaunch, TaskKind, TaskLifecycle};
 
     let host = spec
         .host
@@ -682,6 +682,48 @@ fn normalize_agent(spec: &agent_spec::AgentSpec) -> Result<BTreeMap<String, Sema
         spec.delivery.map(|delivery| delivery.as_str()),
     );
 
+    for stream in &spec.streams {
+        let stream_base = format!("{base}/streams/{}", pointer_segment(&stream.name));
+        match &stream.launch {
+            None => insert_value(
+                &mut fields,
+                &format!("{stream_base}/launch"),
+                SemanticType::String,
+                "external",
+            ),
+            Some(StreamLaunch::Command(command)) => {
+                insert_value(
+                    &mut fields,
+                    &format!("{stream_base}/launch"),
+                    SemanticType::String,
+                    "command",
+                );
+                insert_value(
+                    &mut fields,
+                    &format!("{stream_base}/command"),
+                    SemanticType::String,
+                    command,
+                );
+            }
+            Some(StreamLaunch::Argv(argv)) => {
+                insert_value(
+                    &mut fields,
+                    &format!("{stream_base}/launch"),
+                    SemanticType::String,
+                    "argv",
+                );
+                for (index, argument) in argv.iter().enumerate() {
+                    insert_value(
+                        &mut fields,
+                        &format!("{stream_base}/argv/{index}"),
+                        SemanticType::String,
+                        argument,
+                    );
+                }
+            }
+        }
+    }
+
     let restart = spec.restart_policy();
     let default_restart = Restart::default();
     insert_default_value(
@@ -3421,6 +3463,27 @@ fn test_forced_cross_device(_point: &str) -> std::io::Result<()> {
 mod tests {
     use super::*;
 
+    #[test]
+    fn external_streams_participate_in_semantic_catalog_projection() {
+        let root = tempfile::tempdir().unwrap();
+        let agent = root.path().join("agents/host/worker");
+        std::fs::create_dir_all(&agent).unwrap();
+        std::fs::write(
+            agent.join("agent.kdl"),
+            "agent \"worker\" {\n  host \"host\"\n  command \"worker\"\n  stream \"webhook\" {}\n}\n",
+        )
+        .unwrap();
+        let discovered = agent_spec::discover_strict(root.path());
+        assert!(discovered.errors.is_empty(), "{:?}", discovered.errors);
+
+        let fields = normalize_agent(&discovered.specs[0]).unwrap();
+
+        let atom = fields
+            .get("/agents/host/worker/streams/webhook/launch")
+            .expect("external stream must affect semantic projection");
+        assert_eq!(atom, &present_atom(SemanticType::String, "external"));
+    }
+
     #[test]
     fn workspace_directory_facts_are_typed_into_the_projection_hash() {
         let files = BTreeMap::new();

src/event.rs

@@ -194,7 +194,7 @@ pub fn emit(
     }
     // Serialize the eligibility observation with self-authoring and desired-state changes. Once a
     // suspension edit owns this lock, no later emit can publish from a stale running observation.
-    let _catalog_lock = crate::catalog_lock::CatalogLock::exclusive(root)?;
+    let _catalog_lock = crate::catalog_lock::CatalogLock::shared(root)?;
     let resolved = resolve_stream(root, this_host, recipient, stream)?;
     let canonical_recipient = resolved.recipient;
     let from = format!("{canonical_recipient}/{stream}");
@@ -305,11 +305,13 @@ pub fn emit(
                             root,
                             &canonical_recipient,
                             this_host,
-                            |inbox, _archive| {
+                            |inbox, archive| {
                                 for entry in record.recent.iter().filter(|entry| {
                                     key.is_none_or(|key| entry.key.as_deref() == Some(key))
                                 }) {
-                                    if read_message_entry(inbox, &entry.filename)?.is_some() {
+                                    if read_message_entry(inbox, &entry.filename)?.is_some()
+                                        && read_message_entry(archive, &entry.filename)?.is_none()
+                                    {
                                         return Ok(Some(entry.clone()));
                                     }
                                 }
@@ -353,7 +355,17 @@ pub fn emit(
                             );
                             false
                         }
-                        None => message::materialize_message_once(inbox, &filename, &rendered)?,
+                        None => match read_message_entry(inbox, &filename)? {
+                            Some(bytes) => {
+                                validate_pending_message(
+                                    stream,
+                                    record.pending.as_ref().expect("pending reservation exists"),
+                                    &bytes,
+                                )?;
+                                false
+                            }
+                            None => message::materialize_message_once(inbox, &filename, &rendered)?,
+                        },
                     };
                     test_event_checkpoint(event_id, "materialized")?;
                     if let Some(predecessor) = predecessor.as_ref() {

tests/event_e2e.rs

@@ -1,7 +1,8 @@
 use std::fs;
 use std::path::{Path, PathBuf};
 use std::process::{Command, Stdio};
-use std::sync::{Arc, Barrier, Mutex};
+use std::sync::{Arc, Barrier, Mutex, mpsc};
+use std::time::Duration;
 
 use sha2::{Digest as _, Sha256};
 use st2::event::{self, EventReceiptStatus, RING_CAPACITY};
@@ -412,7 +413,155 @@ fn supersede_skips_an_archived_head_and_retires_the_latest_unread_predecessor()
 }
 
 #[test]
-fn failed_predecessor_archive_leaves_the_successor_unread_and_replay_completes() {
+fn supersede_skips_an_archive_shadowed_inbox_candidate() {
+    let catalog = tempfile::tempdir().unwrap();
+    let agent = declare_agent(catalog.path(), "\"running\"", "  stream \"gh-ci\" {}\n");
+    let older = emit(catalog.path(), "pr1-queued", Some("pr-1"), false);
+    let shadowed = emit(catalog.path(), "pr1-running", Some("pr-1"), false);
+    let inbox = message::inbox_dir(&agent);
+    let archive = message::archive_dir(&agent);
+    fs::create_dir_all(&archive).unwrap();
+    fs::copy(
+        inbox.join(&shadowed.filename),
+        archive.join(&shadowed.filename),
+    )
+    .unwrap();
+
+    let successor = emit(catalog.path(), "pr1-pass", Some("pr-1"), true);
+
+    assert_eq!(
+        successor.superseded.as_deref(),
+        Some(older.filename.as_str())
+    );
+    assert!(inbox.join(&shadowed.filename).is_file());
+    assert!(archive.join(&shadowed.filename).is_file());
+    assert!(archive.join(&older.filename).is_file());
+}
+
+#[test]
+fn initial_supersession_authenticates_predecessor_immediately_before_archive() {
+    let catalog = tempfile::tempdir().unwrap();
+    let agent = declare_agent(catalog.path(), "\"running\"", "  stream \"gh-ci\" {}\n");
+    let predecessor = emit(catalog.path(), "running", Some("pr-1"), false);
+    let inbox = message::inbox_dir(&agent);
+    fs::write(inbox.join(&predecessor.filename), "forged predecessor").unwrap();
+
+    let error = event::emit(
+        catalog.path(),
+        "hetz",
+        "hetz.worker",
+        "gh-ci",
+        "passed",
+        Some("pr-1"),
+        Some("passed"),
+        "passed",
+        true,
+    )
+    .unwrap_err();
+
+    assert!(error.to_string().contains("different bytes"), "{error:#}");
+    assert!(inbox.join(&predecessor.filename).is_file());
+    assert!(
+        !message::archive_dir(&agent)
+            .join(&predecessor.filename)
+            .exists()
+    );
+    assert!(
+        message::list_inbox(&inbox)
+            .unwrap()
+            .iter()
+            .any(|message| { message.event_id.as_deref() == Some("passed") })
+    );
+}
+
+#[test]
+fn catalog_authoring_and_emit_linearize_without_deadlock() {
+    let catalog = tempfile::tempdir().unwrap();
+    declare_agent(catalog.path(), "\"running\"", "  stream \"gh-ci\" {}\n");
+    let admission = st2::CatalogLock::shared(catalog.path()).unwrap();
+    let root = catalog.path().to_path_buf();
+    let (done_tx, done_rx) = mpsc::channel();
+    let author = std::thread::spawn(move || {
+        let result = st2::agent_author::remove_stream(&root, "hetz.worker", "hetz", None, "gh-ci");
+        done_tx.send(result).unwrap();
+    });
+    assert!(done_rx.recv_timeout(Duration::from_millis(100)).is_err());
+
+    let receipt = emit(catalog.path(), "before-remove", None, false);
+    assert_eq!(receipt.status, EventReceiptStatus::Created);
+    drop(admission);
+    done_rx
+        .recv_timeout(Duration::from_secs(5))
+        .unwrap()
+        .unwrap();
+    author.join().unwrap();
+
+    let error = event::emit(
+        catalog.path(),
+        "hetz",
+        "hetz.worker",
+        "gh-ci",
+        "after-remove",
+        None,
+        None,
+        "after",
+        false,
+    )
+    .unwrap_err();
+    assert!(
+        error.to_string().contains("does not declare stream"),
+        "{error:#}"
+    );
+}
+
+#[test]
+fn desired_state_authoring_and_emit_linearize_without_deadlock() {
+    let catalog = tempfile::tempdir().unwrap();
+    declare_agent(catalog.path(), "\"running\"", "  stream \"gh-ci\" {}\n");
+    let admission = st2::CatalogLock::shared(catalog.path()).unwrap();
+    let root = catalog.path().to_path_buf();
+    let (done_tx, done_rx) = mpsc::channel();
+    let author = std::thread::spawn(move || {
+        let result = st2::agent_author::set_desired_state(
+            &root,
+            "hetz.worker",
+            "hetz",
+            None,
+            st2::agent_author::DesiredStateValue::Suspended,
+            Some("maintenance"),
+        );
+        done_tx.send(result).unwrap();
+    });
+    assert!(done_rx.recv_timeout(Duration::from_millis(100)).is_err());
+
+    assert_eq!(
+        emit(catalog.path(), "before-suspend", None, false).status,
+        EventReceiptStatus::Created
+    );
+    drop(admission);
+    done_rx
+        .recv_timeout(Duration::from_secs(5))
+        .unwrap()
+        .unwrap();
+    author.join().unwrap();
+
+    let error = event::emit(
+        catalog.path(),
+        "hetz",
+        "hetz.worker",
+        "gh-ci",
+        "after-suspend",
+        None,
+        None,
+        "after",
+        false,
+    )
+    .unwrap_err();
+    assert!(error.to_string().contains("is suspended"), "{error:#}");
+}
+
+#[test]
+fn invalid_archive_receipt_blocks_supersession_before_successor_publication() {
     let catalog = tempfile::tempdir().unwrap();
     let agent = declare_agent(catalog.path(), "\"running\"", "  stream \"gh-ci\" {}\n");
     let predecessor = emit(catalog.path(), "pr1-running", Some("pr-1"), false);
@@ -438,17 +587,17 @@ fn failed_predecessor_archive_leaves_the_successor_unread_and_replay_completes()
         "{error:#}"
     );
     let unread = message::list_inbox(&inbox).unwrap();
-    assert_eq!(unread.len(), 2);
+    assert_eq!(unread.len(), 1);
     assert!(
         unread
             .iter()
-            .any(|message| message.event_id.as_deref() == Some("pr1-pass"))
+            .all(|message| message.event_id.as_deref() != Some("pr1-pass"))
     );
     assert!(inbox.join(&predecessor.filename).exists());
 
     fs::remove_dir(archive.join(&predecessor.filename)).unwrap();
     let replay = emit(catalog.path(), "pr1-pass", Some("pr-1"), true);
-    assert_eq!(replay.status, EventReceiptStatus::Deduplicated);
+    assert_eq!(replay.status, EventReceiptStatus::Created);
     assert_eq!(
         replay.superseded.as_deref(),
         Some(predecessor.filename.as_str())