diff --git a/crates/buzz-acp/README.md b/crates/buzz-acp/README.md index e6164b02dd..2b04706fcd 100644 --- a/crates/buzz-acp/README.md +++ b/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": , "target": }`. 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:** diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index 403512a322..7f94bc99de 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/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,11 +2138,12 @@ 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!( @@ -2124,7 +2151,16 @@ async fn tokio_main() -> Result<()> { 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); diff --git a/crates/buzz-acp/src/pool.rs b/crates/buzz-acp/src/pool.rs index d1e005cbcc..67c393f775 100644 --- a/crates/buzz-acp/src/pool.rs +++ b/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 { + 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, diff --git a/crates/buzz-core/src/kind.rs b/crates/buzz-core/src/kind.rs index afec52305a..d09706db42 100644 --- a/crates/buzz-core/src/kind.rs +++ 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) diff --git a/crates/buzz-sdk/src/builders.rs b/crates/buzz-sdk/src/builders.rs index 8cc9c8650a..9e3970fd22 100644 --- a/crates/buzz-sdk/src/builders.rs +++ b/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`. //! 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 { + 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 diff --git a/desktop/src/features/messages/ui/SystemMessageRow.tsx b/desktop/src/features/messages/ui/SystemMessageRow.tsx index 4410964dd9..1bbfb2f335 100644 --- a/desktop/src/features/messages/ui/SystemMessageRow.tsx +++ b/desktop/src/features/messages/ui/SystemMessageRow.tsx @@ -574,6 +574,53 @@ function describeSystemEvent( ), }; } + // Owner control commands consumed by the ACP harness (`!shutdown`, + // `!cancel`, `!rotate`). The command never reaches the agent, so this row + // is the only signal it took effect — including when it changed nothing, + // which is why the no-op outcomes get their own wording instead of being + // dropped. Published by the harness, not the relay; see `buzz-acp`. + case "agent_shutdown": + if (!payload.actor || !payload.target) return null; + return { + title: actorName, + action: <>shut down {targetName}, + }; + case "agent_turn_cancelled": + if (!payload.actor || !payload.target) return null; + return { + title: actorName, + action: <>cancelled {targetName}’s current turn, + }; + case "agent_turn_cancel_noop": + if (!payload.actor || !payload.target) return null; + return { + title: actorName, + action: ( + <>tried to cancel {targetName}’s turn — nothing was running + ), + }; + case "agent_session_rotated_in_flight": + if (!payload.actor || !payload.target) return null; + return { + title: actorName, + action: ( + <>cancelled {targetName}’s turn and rotated its session + ), + }; + case "agent_session_rotated": + if (!payload.actor || !payload.target) return null; + return { + title: actorName, + action: <>rotated {targetName}’s session, + }; + case "agent_session_rotate_noop": + if (!payload.actor || !payload.target) return null; + return { + title: actorName, + action: ( + <>rotated {targetName}’s session — it was already fresh + ), + }; case "member_left": return { title: actorName, diff --git a/mobile/lib/features/channels/timeline_message.dart b/mobile/lib/features/channels/timeline_message.dart index 253c949703..223448286b 100644 --- a/mobile/lib/features/channels/timeline_message.dart +++ b/mobile/lib/features/channels/timeline_message.dart @@ -17,6 +17,16 @@ enum SystemEventType { channelUnarchived, huddleStarted, huddleEnded, + + /// Owner control commands consumed by the ACP harness. The command never + /// reaches the agent, so the row is the only signal it took effect — + /// including the no-op outcomes, which is why they are distinct values. + agentShutdown, + agentTurnCancelled, + agentTurnCancelNoop, + agentSessionRotatedInFlight, + agentSessionRotated, + agentSessionRotateNoop, } @immutable @@ -58,6 +68,13 @@ class SystemEvent { 'channel_created' => SystemEventType.channelCreated, 'channel_archived' => SystemEventType.channelArchived, 'channel_unarchived' => SystemEventType.channelUnarchived, + 'agent_shutdown' => SystemEventType.agentShutdown, + 'agent_turn_cancelled' => SystemEventType.agentTurnCancelled, + 'agent_turn_cancel_noop' => SystemEventType.agentTurnCancelNoop, + 'agent_session_rotated_in_flight' => + SystemEventType.agentSessionRotatedInFlight, + 'agent_session_rotated' => SystemEventType.agentSessionRotated, + 'agent_session_rotate_noop' => SystemEventType.agentSessionRotateNoop, _ => null, }; @@ -109,6 +126,18 @@ class SystemEvent { SystemEventType.channelUnarchived => '$actor unarchived this channel', SystemEventType.huddleStarted => '$actor started a huddle', SystemEventType.huddleEnded => '$actor ended the huddle', + SystemEventType.agentShutdown => + '$actor shut down ${resolveLabel(targetPubkey)}', + SystemEventType.agentTurnCancelled => + "$actor cancelled ${resolveLabel(targetPubkey)}'s current turn", + SystemEventType.agentTurnCancelNoop => + "$actor tried to cancel ${resolveLabel(targetPubkey)}'s turn — nothing was running", + SystemEventType.agentSessionRotatedInFlight => + "$actor cancelled ${resolveLabel(targetPubkey)}'s turn and rotated its session", + SystemEventType.agentSessionRotated => + "$actor rotated ${resolveLabel(targetPubkey)}'s session", + SystemEventType.agentSessionRotateNoop => + "$actor rotated ${resolveLabel(targetPubkey)}'s session — it was already fresh", }; } } diff --git a/mobile/test/features/channels/timeline_message_test.dart b/mobile/test/features/channels/timeline_message_test.dart index 87f39d1c65..746cc64ea8 100644 --- a/mobile/test/features/channels/timeline_message_test.dart +++ b/mobile/test/features/channels/timeline_message_test.dart @@ -172,6 +172,13 @@ void main() { 'channel_created': SystemEventType.channelCreated, 'channel_archived': SystemEventType.channelArchived, 'channel_unarchived': SystemEventType.channelUnarchived, + 'agent_shutdown': SystemEventType.agentShutdown, + 'agent_turn_cancelled': SystemEventType.agentTurnCancelled, + 'agent_turn_cancel_noop': SystemEventType.agentTurnCancelNoop, + 'agent_session_rotated_in_flight': + SystemEventType.agentSessionRotatedInFlight, + 'agent_session_rotated': SystemEventType.agentSessionRotated, + 'agent_session_rotate_noop': SystemEventType.agentSessionRotateNoop, }; for (final entry in types.entries) { @@ -254,6 +261,33 @@ void main() { expect(event.describe(resolve), 'Alice left the channel'); }); + // Commandes owner consommées par le harness ACP. `actor` est l'owner qui + // lance la commande, `target` l'agent visé : les intervertir attribuerait + // la commande à la mauvaise partie dans l'historique du canal. + test('agent control outcomes name the owner then the agent', () { + final cases = { + SystemEventType.agentShutdown: 'Alice shut down Bob', + SystemEventType.agentTurnCancelled: + "Alice cancelled Bob's current turn", + SystemEventType.agentTurnCancelNoop: + "Alice tried to cancel Bob's turn — nothing was running", + SystemEventType.agentSessionRotatedInFlight: + "Alice cancelled Bob's turn and rotated its session", + SystemEventType.agentSessionRotated: "Alice rotated Bob's session", + SystemEventType.agentSessionRotateNoop: + "Alice rotated Bob's session — it was already fresh", + }; + + for (final entry in cases.entries) { + final event = SystemEvent( + type: entry.key, + actorPubkey: 'pk1', + targetPubkey: 'pk2', + ); + expect(event.describe(resolve), entry.value, reason: '${entry.key}'); + } + }); + test('member_removed', () { final event = SystemEvent( type: SystemEventType.memberRemoved,