test(stream): prove recovery receipts and exact argv

Author: schickling

src/main.rs

@@ -860,7 +860,7 @@ enum StreamCmd {
         /// Adapter command run under `sh -c`; omit both launch forms for external ingress.
         #[arg(long, conflicts_with = "adapter_argv")]
         command: Option<String>,
-        /// Direct adapter argv. Element 0 is the program.
+        /// Direct adapter argv after `--`. Element 0 is the program; values are preserved exactly.
         #[arg(trailing_var_arg = true, allow_hyphen_values = true)]
         adapter_argv: Vec<String>,
         #[arg(long)]

tests/event_e2e.rs

@@ -451,10 +451,10 @@ fn initial_supersession_authenticates_predecessor_immediately_before_archive() {
         "hetz",
         "hetz.worker",
         "gh-ci",
-        "passed",
+        "initial-passed",
         Some("pr-1"),
-        Some("passed"),
-        "passed",
+        Some("initial passed"),
+        "initial passed",
         true,
     )
     .unwrap_err();
@@ -470,7 +470,7 @@ fn initial_supersession_authenticates_predecessor_immediately_before_archive() {
         message::list_inbox(&inbox)
             .unwrap()
             .iter()
-            .any(|message| { message.event_id.as_deref() == Some("passed") })
+            .any(|message| { message.event_id.as_deref() == Some("initial-passed") })
     );
 }
 
@@ -869,6 +869,92 @@ fn pending_supersession_authenticates_its_predecessor_before_archive() {
     );
 }
 
+#[test]
+fn pending_supersession_accepts_an_authenticated_archive_only_predecessor() {
+    let _fail_env = EVENT_FAIL_ENV.lock().unwrap();
+    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);
+    unsafe { std::env::set_var("ST2_TEST_EVENT_FAIL_AT", "passed:materialized") };
+    let _ = event::emit(
+        catalog.path(),
+        "hetz",
+        "hetz.worker",
+        "gh-ci",
+        "passed",
+        Some("pr-1"),
+        Some("passed"),
+        "passed",
+        true,
+    )
+    .unwrap_err();
+    unsafe { std::env::remove_var("ST2_TEST_EVENT_FAIL_AT") };
+    let inbox = message::inbox_dir(&agent);
+    let archive = message::archive_dir(&agent);
+    message::archive_msg(&inbox, &archive, &predecessor.filename).unwrap();
+
+    let next = emit(catalog.path(), "unrelated", Some("pr-2"), false);
+
+    assert_eq!(next.status, EventReceiptStatus::Created);
+    assert!(archive.join(&predecessor.filename).is_file());
+    assert!(
+        message::list_inbox(&inbox)
+            .unwrap()
+            .iter()
+            .any(|message| { message.event_id.as_deref() == Some("passed") })
+    );
+}
+
+#[test]
+fn pending_supersession_fails_closed_when_predecessor_has_no_receipt() {
+    let _fail_env = EVENT_FAIL_ENV.lock().unwrap();
+    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);
+    unsafe { std::env::set_var("ST2_TEST_EVENT_FAIL_AT", "passed:materialized") };
+    let _ = event::emit(
+        catalog.path(),
+        "hetz",
+        "hetz.worker",
+        "gh-ci",
+        "passed",
+        Some("pr-1"),
+        Some("passed"),
+        "passed",
+        true,
+    )
+    .unwrap_err();
+    unsafe { std::env::remove_var("ST2_TEST_EVENT_FAIL_AT") };
+    let inbox = message::inbox_dir(&agent);
+    fs::remove_file(inbox.join(&predecessor.filename)).unwrap();
+
+    let error = event::emit(
+        catalog.path(),
+        "hetz",
+        "hetz.worker",
+        "gh-ci",
+        "unrelated",
+        None,
+        None,
+        "unrelated",
+        false,
+    )
+    .unwrap_err();
+
+    assert!(
+        error
+            .to_string()
+            .contains("has no inbox file or archive receipt"),
+        "{error:#}"
+    );
+    assert!(
+        message::list_inbox(&inbox)
+            .unwrap()
+            .iter()
+            .all(|message| { message.event_id.as_deref() != Some("unrelated") })
+    );
+}
+
 #[test]
 fn keyless_supersede_replaces_the_stream_wide_head() {
     let catalog = tempfile::tempdir().unwrap();

tests/stream_authoring_cli.rs

@@ -165,6 +165,83 @@ fn a_direct_adapter_launch_executes_the_exact_event_cli_contract() {
     );
 }
 
+#[test]
+fn direct_adapter_argv_preserves_spaces_and_metacharacters_exactly() {
+    let catalog = tempfile::tempdir().unwrap();
+    write_agent(catalog.path());
+    let expected = [
+        "/bin/example adapter",
+        "argument with spaces",
+        "$HOME",
+        "$(never-executed)",
+        "semi;colon",
+        "quote\"and\\slash",
+        "--looks-like-a-flag",
+    ];
+    let mut args = vec![
+        "stream",
+        "add",
+        "exact-argv",
+        "--agent",
+        "hetz.worker",
+        "--host",
+        "hetz",
+        "--",
+    ];
+    args.extend(expected);
+
+    let add = st2(catalog.path(), &args);
+
+    assert!(
+        add.status.success(),
+        "{}",
+        String::from_utf8_lossy(&add.stderr)
+    );
+    let spec = st2::discover(catalog.path()).specs.remove(0);
+    let stream = spec
+        .streams
+        .iter()
+        .find(|stream| stream.name == "exact-argv")
+        .unwrap();
+    assert_eq!(
+        stream.launch,
+        Some(st2::spec::StreamLaunch::Argv(
+            expected.iter().map(|value| (*value).to_owned()).collect()
+        ))
+    );
+}
+
+#[test]
+fn command_and_direct_argv_are_mutually_exclusive() {
+    let catalog = tempfile::tempdir().unwrap();
+    write_agent(catalog.path());
+
+    let add = st2(
+        catalog.path(),
+        &[
+            "stream",
+            "add",
+            "ambiguous",
+            "--agent",
+            "hetz.worker",
+            "--host",
+            "hetz",
+            "--command",
+            "echo shell",
+            "--",
+            "/bin/echo",
+            "direct",
+        ],
+    );
+
+    assert!(!add.status.success());
+    assert!(
+        String::from_utf8_lossy(&add.stderr).contains("cannot be used with"),
+        "{}",
+        String::from_utf8_lossy(&add.stderr)
+    );
+}
+
 #[test]
 fn a_bare_actor_can_self_author_on_the_selected_host() {
     let catalog = tempfile::tempdir().unwrap();