feat(acp): post owner control command outcomes back into the channel

crates/buzz-acp/README.md

@@ -159,6 +159,10 @@ Use `!cancel` to stop only the current turn; it is a no-op when the channel is i
 
 Owner control commands must be kind:9 stream messages from the owner, must mention this agent with a `p` tag, and are consumed by the harness instead of being forwarded to the agent.
 
+Because the command never reaches the agent, the harness publishes the outcome back into the channel as a kind:40099 system message — the grey row clients render beside "X joined the channel", since a cancelled turn or a rotated session is a lifecycle event rather than the agent speaking. It lands in the thread the command was typed in, and it is published for the no-op outcomes too (`!cancel` with no turn in flight, `!rotate` with no cached session), so a command that changed nothing stays distinguishable from one that was never received.
+
+The payload is `{"type": …, "actor": <owner>, "target": <agent>}`. Clients switch on `type` and silently drop an unknown one, so a client older than this contract shows no row at all.
+
 > **Note:** The default mode is `owner-only`. Agents without a registered `agent_owner_pubkey` will not respond to any events until the owner is resolved. Set `--respond-to anyone` to disable the gate entirely.
 
 **Examples:**

crates/buzz-acp/src/lib.rs

@@ -2044,6 +2044,20 @@ async fn tokio_main() -> Result<()> {
                                             sender = %buzz_event.event.pubkey.to_hex(),
                                             "shutdown command from owner — exiting gracefully"
                                         );
+                                        // Awaited, not spawned: the harness is
+                                        // about to exit, and a spawned task
+                                        // would race the shutdown.
+                                        pool::post_system_notice(
+                                            &ctx.rest_client,
+                                            buzz_event.channel_id,
+                                            &queue::parse_thread_tags(&buzz_event.event),
+                                            &control_notice_payload(
+                                                NOTICE_SHUTDOWN,
+                                                owner,
+                                                &pubkey_hex,
+                                            ),
+                                        )
+                                        .await;
                                         let _ = shutdown_tx.send(());
                                         continue;
                                     }
@@ -2080,6 +2094,18 @@ async fn tokio_main() -> Result<()> {
                                                 "!cancel received but no in-flight task — no-op"
                                             );
                                         }
+                                        spawn_control_notice(
+                                            &ctx.rest_client,
+                                            &buzz_event.event,
+                                            buzz_event.channel_id,
+                                            if fired {
+                                                NOTICE_TURN_CANCELLED
+                                            } else {
+                                                NOTICE_TURN_CANCEL_NOOP
+                                            },
+                                            owner,
+                                            &pubkey_hex,
+                                        );
                                         continue; // consume event — do NOT push to queue
                                     }
                                 }
@@ -2112,19 +2138,29 @@ async fn tokio_main() -> Result<()> {
                                             buzz_event.channel_id,
                                             ControlSignal::Rotate,
                                         );
-                                        if fired {
+                                        let notice = if fired {
                                             tracing::info!(
                                                 channel_id = %buzz_event.channel_id,
                                                 "!rotate received — cancelling in-flight turn and rotating session"
                                             );
+                                            NOTICE_SESSION_ROTATED_IN_FLIGHT
                                         } else {
                                             let invalidated = pool.invalidate_channel_sessions(buzz_event.channel_id);
                                             tracing::info!(
                                                 channel_id = %buzz_event.channel_id,
                                                 invalidated,
                                                 "!rotate received — invalidated idle channel session(s)"
                                             );
-                                        }
+                                            rotate_idle_notice_type(invalidated)
+                                        };
+                                        spawn_control_notice(
+                                            &ctx.rest_client,
+                                            &buzz_event.event,
+                                            buzz_event.channel_id,
+                                            notice,
+                                            owner,
+                                            &pubkey_hex,
+                                        );
                                         continue; // consume event — do NOT push to queue
                                     }
                                 }
@@ -3051,6 +3087,73 @@ fn spawn_failure_notice(
     }
 }
 
+/// `type` values of the kind:40099 system message reporting the outcome of a
+/// consumed owner control command.
+///
+/// The renderers switch on these strings and drop an unknown one silently
+/// (`describeSystemEvent` in `SystemMessageRow.tsx`, `SystemEvent.fromContent`
+/// in `timeline_message.dart`), so adding a variant here means adding a case in
+/// both.
+const NOTICE_SHUTDOWN: &str = "agent_shutdown";
+const NOTICE_TURN_CANCELLED: &str = "agent_turn_cancelled";
+const NOTICE_TURN_CANCEL_NOOP: &str = "agent_turn_cancel_noop";
+const NOTICE_SESSION_ROTATED_IN_FLIGHT: &str = "agent_session_rotated_in_flight";
+const NOTICE_SESSION_ROTATED: &str = "agent_session_rotated";
+const NOTICE_SESSION_ROTATE_NOOP: &str = "agent_session_rotate_noop";
+
+/// Build the system-message payload for a consumed owner control command.
+///
+/// `actor` is the owner who issued the command, `target` the agent it acted on
+/// — the same two fields every other system message uses, so the clients
+/// resolve both to profile names and prefetch their profiles.
+fn control_notice_payload(
+    notice_type: &str,
+    owner_hex: &str,
+    agent_hex: &str,
+) -> serde_json::Value {
+    serde_json::json!({
+        "type": notice_type,
+        "actor": owner_hex,
+        "target": agent_hex,
+    })
+}
+
+/// Spawn a task that publishes the outcome of a consumed owner control command
+/// into the channel it came from.
+///
+/// Control commands are consumed by the harness and never reach the agent, so
+/// this notice is the only signal the owner gets. Without it a successful
+/// command, a no-op, and an event the harness never saw all look the same from
+/// the chat: your own message, then silence.
+fn spawn_control_notice(
+    rest_client: &relay::RestClient,
+    event: &nostr::Event,
+    channel_id: Uuid,
+    notice_type: &'static str,
+    owner_hex: &str,
+    agent_hex: &str,
+) {
+    let thread_tags = queue::parse_thread_tags(event);
+    let payload = control_notice_payload(notice_type, owner_hex, agent_hex);
+    let rest = rest_client.clone();
+    tokio::spawn(async move {
+        pool::post_system_notice(&rest, channel_id, &thread_tags, &payload).await;
+    });
+}
+
+/// Notice type for a `!rotate` that found no turn in flight.
+///
+/// `invalidated` is how many cached sessions were dropped. Zero means there was
+/// nothing to rotate — the next turn was already going to start fresh. Saying
+/// so explicitly is the point: a silent no-op reads as a broken command.
+fn rotate_idle_notice_type(invalidated: usize) -> &'static str {
+    if invalidated > 0 {
+        NOTICE_SESSION_ROTATED
+    } else {
+        NOTICE_SESSION_ROTATE_NOOP
+    }
+}
+
 #[allow(clippy::too_many_arguments)]
 fn handle_prompt_result(
     pool: &mut AgentPool,
@@ -4314,6 +4417,76 @@ mod owner_control_command_tests {
         ));
     }
 
+    #[test]
+    fn control_notice_lands_in_the_thread_the_command_was_typed_in() {
+        // A command typed in a thread must be answered in that thread. Posting
+        // the notice at channel level instead would surface an unrelated root
+        // message, which is how the owner loses track of what it answers.
+        let keys = Keys::generate();
+        let root = "11".repeat(32);
+        let reply = "22".repeat(32);
+        let threaded = EventBuilder::new(Kind::Custom(KIND_STREAM_MESSAGE as u16), "!rotate")
+            .tags([
+                Tag::parse(["e", &root, "", "root"]).expect("root tag"),
+                Tag::parse(["e", &reply, "", "reply"]).expect("reply tag"),
+            ])
+            .sign_with_keys(&keys)
+            .unwrap();
+
+        let thread_ref = pool::thread_ref_from_tags(&queue::parse_thread_tags(&threaded))
+            .expect("a threaded command yields a thread ref");
+        assert_eq!(thread_ref.root_event_id.to_hex(), root);
+        assert_eq!(thread_ref.parent_event_id.to_hex(), reply);
+
+        // Conversely, a command typed at channel level must NOT be threaded, or
+        // the notice would attach itself to whatever root it inherited.
+        let bare = EventBuilder::new(Kind::Custom(KIND_STREAM_MESSAGE as u16), "!rotate")
+            .sign_with_keys(&keys)
+            .unwrap();
+        assert!(pool::thread_ref_from_tags(&queue::parse_thread_tags(&bare)).is_none());
+    }
+
+    #[test]
+    fn rotate_idle_notice_distinguishes_a_rotation_from_a_no_op() {
+        // The no-op is the case worth distinguishing: nothing was cached, so
+        // the command changed nothing, and the row has to say that rather than
+        // claim a rotation that did not happen.
+        assert_eq!(rotate_idle_notice_type(0), NOTICE_SESSION_ROTATE_NOOP);
+
+        let rotated = rotate_idle_notice_type(1);
+        assert_eq!(rotated, NOTICE_SESSION_ROTATED);
+        assert_eq!(rotate_idle_notice_type(3), rotated);
+    }
+
+    #[test]
+    fn control_notice_payload_names_the_owner_and_the_agent() {
+        // `actor` and `target` are the fields every other system message uses,
+        // and both renderers resolve them to profile names — swapping them
+        // would credit the wrong party for the command in the channel history.
+        let owner = "ab".repeat(32);
+        let agent = "cd".repeat(32);
+        let payload = control_notice_payload(NOTICE_TURN_CANCELLED, &owner, &agent);
+
+        assert_eq!(payload["type"], NOTICE_TURN_CANCELLED);
+        assert_eq!(payload["actor"], owner);
+        assert_eq!(payload["target"], agent);
+    }
+
+    #[test]
+    fn control_notice_types_are_distinct() {
+        // A duplicate would make two different outcomes render the same row.
+        let all = [
+            NOTICE_SHUTDOWN,
+            NOTICE_TURN_CANCELLED,
+            NOTICE_TURN_CANCEL_NOOP,
+            NOTICE_SESSION_ROTATED_IN_FLIGHT,
+            NOTICE_SESSION_ROTATED,
+            NOTICE_SESSION_ROTATE_NOOP,
+        ];
+        let unique: std::collections::HashSet<&str> = all.iter().copied().collect();
+        assert_eq!(unique.len(), all.len());
+    }
+
     #[test]
     fn mode_gate_signal_maps_handling_to_control_signal() {
         let owner = "a".repeat(64);

crates/buzz-acp/src/pool.rs

@@ -3615,6 +3615,66 @@ pub(crate) async fn reaction_add(rest: &crate::relay::RestClient, event_id: &str
     }
 }
 
+/// Resolve the NIP-10 thread a notice must land in, from the tags of the event
+/// that triggered it.
+///
+/// `None` when the trigger carried no thread tags — the notice then posts at
+/// channel level. This is what keeps a notice in the thread the owner was
+/// typing in instead of surfacing as an unrelated root message.
+pub(crate) fn thread_ref_from_tags(thread_tags: &ThreadTags) -> Option<buzz_sdk::ThreadRef> {
+    let root = thread_tags.root_event_id.as_deref()?;
+    let root_id = nostr::EventId::from_hex(root).ok()?;
+    let parent_id = thread_tags
+        .parent_event_id
+        .as_deref()
+        .and_then(|p| nostr::EventId::from_hex(p).ok())
+        .unwrap_or(root_id);
+    Some(buzz_sdk::ThreadRef {
+        root_event_id: root_id,
+        parent_event_id: parent_id,
+    })
+}
+
+/// Best-effort: publish a kind:40099 system message into `channel_id`, signed
+/// with the agent's own key.
+///
+/// Carries the outcome of an owner control command the harness consumed. A
+/// cancelled turn or a rotated session is a lifecycle event, not the agent
+/// speaking, so it renders as a system row beside "X joined the channel"
+/// instead of as a message in the conversation. It still lands in the thread
+/// the command was typed in — thread placement is derived from NIP-10 `e` tags
+/// and is kind-agnostic on the client.
+///
+/// Errors are logged and swallowed: the notice must never take down the main
+/// loop, and must never block the command it reports on.
+pub(crate) async fn post_system_notice(
+    rest: &crate::relay::RestClient,
+    channel_id: Uuid,
+    thread_tags: &ThreadTags,
+    payload: &serde_json::Value,
+) {
+    let thread_ref = thread_ref_from_tags(thread_tags);
+    let builder = match buzz_sdk::build_system_message(channel_id, payload, thread_ref.as_ref()) {
+        Ok(b) => b,
+        Err(e) => {
+            tracing::warn!(channel = %channel_id, "system notice: build failed: {e}");
+            return;
+        }
+    };
+    let event = match builder.sign_with_keys(&rest.keys) {
+        Ok(e) => e,
+        Err(e) => {
+            tracing::warn!(channel = %channel_id, "system notice: sign failed: {e}");
+            return;
+        }
+    };
+    match tokio::time::timeout(Duration::from_secs(5), rest.submit_event(&event)).await {
+        Ok(Ok(_)) => {}
+        Ok(Err(e)) => tracing::warn!(channel = %channel_id, "system notice failed: {e}"),
+        Err(_) => tracing::warn!(channel = %channel_id, "system notice timed out"),
+    }
+}
+
 /// Best-effort: post a visible failure notice (kind:9) to a channel after a
 /// batch is dead-lettered. Replies into the thread of `thread_tags` when the
 /// triggering event was threaded. Errors are logged and swallowed — the
@@ -3625,18 +3685,7 @@ pub(crate) async fn post_failure_notice(
     thread_tags: &ThreadTags,
     content: &str,
 ) {
-    let thread_ref = thread_tags.root_event_id.as_deref().and_then(|root| {
-        let root_id = nostr::EventId::from_hex(root).ok()?;
-        let parent_id = thread_tags
-            .parent_event_id
-            .as_deref()
-            .and_then(|p| nostr::EventId::from_hex(p).ok())
-            .unwrap_or(root_id);
-        Some(buzz_sdk::ThreadRef {
-            root_event_id: root_id,
-            parent_event_id: parent_id,
-        })
-    });
+    let thread_ref = thread_ref_from_tags(thread_tags);
     let builder =
         match buzz_sdk::build_message(channel_id, content, thread_ref.as_ref(), &[], false, &[]) {
             Ok(b) => b,

crates/buzz-core/src/kind.rs

@@ -433,7 +433,9 @@ pub const KIND_STREAM_REMINDER: u32 = 40007;
 pub const KIND_STREAM_MESSAGE_DIFF: u32 = 40008;
 /// Canvas (shared document) for a channel.
 pub const KIND_CANVAS: u32 = 40100;
-/// System message for channel state changes (join, leave, rename, etc.).
+/// System message for channel state changes (join, leave, rename, etc.),
+/// and for the outcome of an owner control command consumed by the ACP
+/// harness. Relay-signed for channel state; agent-signed for the latter.
 pub const KIND_SYSTEM_MESSAGE: u32 = 40099;
 
 // Relay-only sidecar kinds (never client-submitted)

crates/buzz-sdk/src/builders.rs

@@ -1,4 +1,4 @@
-//! Typed event builder functions (38 builders).
+//! Typed event builder functions (39 builders).
 //!
 //! All functions return `Result<nostr::EventBuilder, SdkError>`.
 //! The caller signs: `builder.sign_with_keys(&keys)?`.
@@ -237,6 +237,35 @@ pub fn build_message(
     Ok(EventBuilder::new(Kind::Custom(9), content).tags(tags))
 }
 
+/// Build a system message (kind 40099) for a channel.
+///
+/// System messages carry a JSON payload whose `type` field clients switch on to
+/// render a system row — a channel lifecycle event such as a membership change,
+/// or the outcome of an owner control command consumed by the harness — instead
+/// of a message in the conversation. Unknown `type` values are dropped by the
+/// clients, so a new one needs matching cases in the renderers.
+///
+/// `thread_ref` places the row in the thread that triggered it; `None` posts it
+/// at channel level. Thread placement is derived from NIP-10 `e` tags and is
+/// kind-agnostic on the client, so this threads exactly like [`build_message`].
+pub fn build_system_message(
+    channel_id: Uuid,
+    payload: &serde_json::Value,
+    thread_ref: Option<&ThreadRef>,
+) -> Result<EventBuilder, SdkError> {
+    let content = payload.to_string();
+    check_content(&content, 64 * 1024)?;
+    let mut tags = vec![tag(&["h", &channel_id.to_string()])?];
+    if let Some(tr) = thread_ref {
+        thread_tags(tr, &mut tags)?;
+    }
+    Ok(EventBuilder::new(
+        Kind::Custom(buzz_core::kind::KIND_SYSTEM_MESSAGE as u16),
+        content,
+    )
+    .tags(tags))
+}
+
 /// Build an encrypted agent observer frame (kind 24200).
 ///
 /// `recipient_pubkey` is the cleartext `p` tag used by the relay for owner-only