fix(stream): close review edge cases

Author: schickling

docs/vrs/04-stream/spec.md

@@ -155,9 +155,11 @@ archive-wide event-identity lookup.
 
 ## Supersession (STREAM-R07)
 
-`--supersede` archives the most recent retained predecessor for the same
-`(stream, key)` before publishing the successor; without `--key`, it archives
-the stream-wide head — log-compaction semantics. The archive move uses the ordinary archive path, so
+`--supersede` archives the most recent retained unread predecessor for the
+same `(stream, key)` before publishing the successor; without `--key`, it
+archives the latest retained unread stream-wide predecessor — log-compaction
+semantics. Already archived receipts are skipped rather than masking an older
+unread predecessor. The archive move uses the ordinary archive path, so
 a DING-staged predecessor resolves through the existing archive-receipt rule:
 pasted at most once ever, never re-pasted, successor delivers next. Proven:
 24 supersedes in 146 ms produced one fresh poke, zero staged retries, one

src/agent_author.rs

@@ -241,11 +241,14 @@ fn author_stream(
         ));
     }
     let target = resolve_target(&found.specs, selector, this_host)?;
+    let actor = actor
+        .map(|actor| resolve_target(&found.specs, actor, this_host).map(|target| target.identity))
+        .transpose()?;
     authorize_actor(
         &found.specs,
         &target.identity,
         this_host,
-        actor,
+        actor.as_deref(),
         "stream-not-authorized",
     )?;
     let result = edit_stream_declaration(
@@ -631,6 +634,8 @@ fn edit_stream_declaration(
         catalog,
         path,
         &replacement,
+        expected_identity,
+        expected_host,
         expected_agent,
         name,
         launch,
@@ -744,6 +749,8 @@ fn verify_stream_candidate(
     catalog: &Path,
     path: &Path,
     candidate: &str,
+    expected_identity: &str,
+    expected_host: &str,
     expected_agent: &str,
     name: &str,
     launch: Option<&StreamLaunch>,
@@ -778,7 +785,9 @@ fn verify_stream_candidate(
         .map_err(|error| AuthorError::new("invalid-stream", error.to_string()))?;
     let spec = specs
         .iter()
-        .find(|spec| spec.identity == expected_agent)
+        .find(|spec| {
+            spec.identity == expected_agent && spec.bus_id(expected_host) == expected_identity
+        })
         .ok_or_else(|| {
             AuthorError::new(
                 "unsafe-source-edit",
@@ -1891,6 +1900,45 @@ mod tests {
         assert_eq!(fs::read_to_string(path).unwrap(), original);
     }
 
+    #[test]
+    fn stream_candidate_verification_matches_the_exact_host_agent() {
+        let temporary = tempfile::tempdir().unwrap();
+        let root = temporary.path();
+        let path = write(
+            root,
+            "agents.kdl",
+            "agent \"worker\" { host \"alpha\"; command \"sleep 60\"; stream \"existing\" {} }\nagent \"worker\" { host \"beta\"; command \"sleep 60\"; stream \"existing\" {} }\n",
+        );
+
+        assert_eq!(
+            add_stream(
+                root,
+                "beta.worker",
+                "beta",
+                Some("beta.worker"),
+                "webhook",
+                None,
+            )
+            .unwrap()
+            .result,
+            AuthorOutcome::Changed
+        );
+        assert_eq!(
+            remove_stream(root, "beta.worker", "beta", Some("beta.worker"), "existing",)
+                .unwrap()
+                .result,
+            AuthorOutcome::Changed
+        );
+
+        let authored = fs::read_to_string(path).unwrap();
+        let document = KdlDocument::parse(&authored).unwrap();
+        let agents = document.nodes();
+        assert!(agents[0].to_string().contains("stream \"existing\""));
+        assert!(!agents[0].to_string().contains("stream \"webhook\""));
+        assert!(!agents[1].to_string().contains("stream \"existing\""));
+        assert!(agents[1].to_string().contains("stream \"webhook\""));
+    }
+
     #[test]
     fn stream_authoring_enforces_authority_nix_ownership_and_canonical_validation() {
         let temporary = tempfile::tempdir().unwrap();

src/event.rs

@@ -261,25 +261,31 @@ pub fn emit(
                 write_record(&record_path, &record)?;
             }
 
-            let predecessor = if supersede {
+            let predecessor_candidates = if supersede {
                 record
                     .recent
                     .iter()
-                    .find(|entry| {
+                    .filter(|entry| {
                         entry.filename != filename
                             && key.is_none_or(|key| entry.key.as_deref() == Some(key))
                     })
                     .map(|entry| entry.filename.clone())
+                    .collect::<Vec<_>>()
             } else {
-                None
+                Vec::new()
             };
+            let mut predecessor = None;
             let created = message::with_resolved_message_boxes(
                 root,
                 &canonical_recipient,
                 this_host,
                 |inbox, archive| {
-                    if let Some(predecessor) = &predecessor {
-                        message::archive_msg(inbox, archive, predecessor)?;
+                    if let Some(unread) = predecessor_candidates
+                        .iter()
+                        .find(|candidate| inbox.join(candidate).is_file())
+                    {
+                        message::archive_msg(inbox, archive, unread)?;
+                        predecessor = Some(unread.clone());
                     }
                     // An archive filename is the bus's authoritative durable receipt. A crash after
                     // materializing an external archive, but before advancing this state, must not restore

tests/event_e2e.rs

@@ -347,6 +347,28 @@ fn supersede_collapses_only_the_matching_key_and_preserves_archive_receipts() {
     assert!(!inbox.join(&pr_1_old.filename).exists());
 }
 
+#[test]
+fn supersede_skips_an_archived_head_and_retires_the_latest_unread_predecessor() {
+    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 archived_head = emit(catalog.path(), "pr1-running", Some("pr-1"), false);
+    let inbox = message::inbox_dir(&agent);
+    let archive = message::archive_dir(&agent);
+    message::archive_msg(&inbox, &archive, &archived_head.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(&older.filename).exists());
+    assert!(archive.join(&older.filename).exists());
+    assert!(archive.join(&archived_head.filename).exists());
+    assert!(inbox.join(&successor.filename).exists());
+}
+
 #[test]
 fn keyless_supersede_replaces_the_stream_wide_head() {
     let catalog = tempfile::tempdir().unwrap();

tests/stream_authoring_cli.rs

@@ -164,3 +164,33 @@ fn a_direct_adapter_launch_executes_the_exact_event_cli_contract() {
         1
     );
 }
+
+#[test]
+fn a_bare_actor_can_self_author_on_the_selected_host() {
+    let catalog = tempfile::tempdir().unwrap();
+    write_agent(catalog.path());
+
+    let add = st2(
+        catalog.path(),
+        &[
+            "stream", "add", "webhook", "--as", "worker", "--host", "hetz",
+        ],
+    );
+    assert!(
+        add.status.success(),
+        "{}",
+        String::from_utf8_lossy(&add.stderr)
+    );
+
+    let remove = st2(
+        catalog.path(),
+        &[
+            "stream", "rm", "webhook", "--as", "worker", "--host", "hetz",
+        ],
+    );
+    assert!(
+        remove.status.success(),
+        "{}",
+        String::from_utf8_lossy(&remove.stderr)
+    );
+}