Route ACP observer traffic over relay (#421)

Signed-off-by: Tyler Longwell <tlongwell@squareup.com>
Co-authored-by: tlongwell-block <109685178+tlongwell-block@users.noreply.github.com>
Co-authored-by: Tyler Longwell <tlongwell@squareup.com>

Cargo.lock

@@ -3632,7 +3632,6 @@ name = "sprout-acp"
 version = "0.1.0"
 dependencies = [
  "anyhow",
- "axum",
  "chrono",
  "clap",
  "evalexpr",

crates/sprout-acp/Cargo.toml

@@ -25,7 +25,6 @@ tokio = { workspace = true }
 
 # WebSocket
 tokio-tungstenite = { workspace = true }
-axum = { workspace = true }
 
 # Codec (bounded line reads)
 tokio-util = { workspace = true }

crates/sprout-acp/src/config.rs

@@ -359,16 +359,9 @@ pub struct CliArgs {
     #[arg(long, env = "SPROUT_ACP_PERSONA_NAME")]
     pub persona_name: Option<String>,
 
-    /// Local bind address for the ACP observer API.
-    ///
-    /// Intended for the desktop app. The feed stays on localhost and is not
-    /// part of relay-visible Sprout history.
-    #[arg(long, env = "SPROUT_ACP_OBSERVER_ADDR")]
-    pub observer_addr: Option<String>,
-
-    /// Optional token required by the observer API.
-    #[arg(long, env = "SPROUT_ACP_OBSERVER_TOKEN")]
-    pub observer_token: Option<String>,
+    /// Publish encrypted ACP observer frames over the relay.
+    #[arg(long, env = "SPROUT_ACP_RELAY_OBSERVER", default_value_t = false)]
+    pub relay_observer: bool,
 }
 
 // ── Merged NIP-01 filter ──────────────────────────────────────────────────────
@@ -423,10 +416,8 @@ pub struct Config {
     /// Per-persona env vars to inject at agent spawn time (e.g., GOOSE_PROVIDER, GOOSE_MODEL).
     /// Populated from persona pack resolution. Empty when no pack is configured.
     pub persona_env_vars: Vec<(String, String)>,
-    /// Local observer bind address, when enabled.
-    pub observer_addr: Option<String>,
-    /// Local observer token, when configured.
-    pub observer_token: Option<String>,
+    /// Whether to publish encrypted observer frames through the relay.
+    pub relay_observer: bool,
 }
 
 /// Validate and deduplicate allowlist entries: each must be exactly 64 hex chars.
@@ -762,8 +753,7 @@ impl Config {
             respond_to: args.respond_to,
             respond_to_allowlist,
             persona_env_vars,
-            observer_addr: args.observer_addr,
-            observer_token: args.observer_token,
+            relay_observer: args.relay_observer,
         };
 
         Ok(config)
@@ -1124,8 +1114,7 @@ mod tests {
             respond_to: RespondTo::Anyone,
             respond_to_allowlist: HashSet::new(),
             persona_env_vars: vec![],
-            observer_addr: None,
-            observer_token: None,
+            relay_observer: false,
         }
     }
 

crates/sprout-acp/src/main.rs

@@ -18,17 +18,20 @@ use clap::Parser;
 use config::{Config, DedupMode, ModelsArgs, MultipleEventHandling, RespondTo, SubscribeMode};
 use filter::SubscriptionRule;
 use futures_util::FutureExt;
-use nostr::ToBech32;
+use nostr::{PublicKey, ToBech32};
 use pool::{
     AgentPool, CancelMode, OwnedAgent, PromptContext, PromptOutcome, PromptResult, PromptSource,
     SessionState,
 };
 use queue::{EventQueue, QueuedEvent, ThreadTags};
-use relay::HarnessRelay;
+use relay::{HarnessRelay, RelayEventPublisher};
 use sprout_core::kind::{
     KIND_MEMBER_ADDED_NOTIFICATION, KIND_MEMBER_REMOVED_NOTIFICATION, KIND_STREAM_MESSAGE,
     KIND_STREAM_REMINDER, KIND_WORKFLOW_APPROVAL_REQUESTED,
 };
+use sprout_core::observer::{
+    decrypt_observer_payload, encrypt_observer_payload, OBSERVER_FRAME_TELEMETRY,
+};
 use tokio::sync::{mpsc, watch};
 use tracing_subscriber::EnvFilter;
 use uuid::Uuid;
@@ -293,6 +296,305 @@ async fn is_owner_or_sibling(
     }
 }
 
+fn spawn_relay_observer_publisher(
+    observer: observer::ObserverHandle,
+    publisher: RelayEventPublisher,
+    keys: nostr::Keys,
+    agent_pubkey_hex: String,
+    owner_pubkey_hex: String,
+    owner_pubkey: PublicKey,
+) -> tokio::task::JoinHandle<()> {
+    tokio::spawn(async move {
+        let mut coalescer = ObserverChunkCoalescer::default();
+        for event in observer.snapshot() {
+            for event in coalescer.ingest(event) {
+                publish_relay_observer_event(
+                    &publisher,
+                    &keys,
+                    &agent_pubkey_hex,
+                    &owner_pubkey_hex,
+                    &owner_pubkey,
+                    event,
+                )
+                .await;
+            }
+        }
+
+        let mut rx = observer.subscribe();
+        let mut flush_interval = tokio::time::interval(std::time::Duration::from_millis(500));
+        flush_interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
+        loop {
+            tokio::select! {
+                result = rx.recv() => {
+                    match result {
+                        Ok(event) => {
+                            for event in coalescer.ingest(event) {
+                                publish_relay_observer_event(
+                                    &publisher, &keys, &agent_pubkey_hex,
+                                    &owner_pubkey_hex, &owner_pubkey, event,
+                                ).await;
+                            }
+                        }
+                        Err(tokio::sync::broadcast::error::RecvError::Lagged(count)) => {
+                            for event in coalescer.flush() {
+                                publish_relay_observer_event(
+                                    &publisher, &keys, &agent_pubkey_hex,
+                                    &owner_pubkey_hex, &owner_pubkey, event,
+                                ).await;
+                            }
+                            tracing::warn!(dropped = count, "relay observer publisher lagged");
+                        }
+                        Err(tokio::sync::broadcast::error::RecvError::Closed) => {
+                            for event in coalescer.flush() {
+                                publish_relay_observer_event(
+                                    &publisher, &keys, &agent_pubkey_hex,
+                                    &owner_pubkey_hex, &owner_pubkey, event,
+                                ).await;
+                            }
+                            break;
+                        }
+                    }
+                }
+                _ = flush_interval.tick() => {
+                    // Periodic flush ensures live streaming even during continuous chunk delivery.
+                    for event in coalescer.flush() {
+                        publish_relay_observer_event(
+                            &publisher, &keys, &agent_pubkey_hex,
+                            &owner_pubkey_hex, &owner_pubkey, event,
+                        ).await;
+                    }
+                }
+            }
+        }
+    })
+}
+
+#[derive(Default)]
+struct ObserverChunkCoalescer {
+    pending: Vec<PendingObserverChunk>,
+}
+
+struct PendingObserverChunk {
+    key: ObserverChunkKey,
+    event: observer::ObserverEvent,
+    text: String,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+struct ObserverChunkKey {
+    update_type: String,
+    message_id: Option<String>,
+    channel_id: Option<String>,
+    session_id: Option<String>,
+    turn_id: Option<String>,
+    agent_index: Option<usize>,
+}
+
+/// Flush coalesced chunks before they exceed the NIP-44 plaintext limit (65,535 bytes).
+/// Leave headroom for the JSON envelope wrapping the text.
+const OBSERVER_CHUNK_MAX_TEXT_BYTES: usize = 60_000;
+
+impl ObserverChunkCoalescer {
+    fn ingest(&mut self, event: observer::ObserverEvent) -> Vec<observer::ObserverEvent> {
+        let Some((key, text)) = observer_chunk_key_and_text(&event) else {
+            let mut events = self.flush();
+            events.push(event);
+            return events;
+        };
+
+        if let Some(pending) = self.pending.iter_mut().find(|pending| pending.key == key) {
+            // Flush before appending if this would exceed the plaintext size limit.
+            if pending.text.len() + text.len() >= OBSERVER_CHUNK_MAX_TEXT_BYTES {
+                let events = self.flush();
+                // Start a new pending entry with the current chunk.
+                self.pending.push(PendingObserverChunk { key, event, text });
+                return events;
+            }
+            pending.text.push_str(&text);
+            pending.event.seq = event.seq;
+            pending.event.timestamp = event.timestamp;
+            return Vec::new();
+        }
+
+        self.pending.push(PendingObserverChunk { key, event, text });
+        Vec::new()
+    }
+
+    fn flush(&mut self) -> Vec<observer::ObserverEvent> {
+        self.pending
+            .drain(..)
+            .map(|mut pending| {
+                set_observer_chunk_text(&mut pending.event.payload, pending.text);
+                pending.event
+            })
+            .collect()
+    }
+}
+
+fn observer_chunk_key_and_text(
+    event: &observer::ObserverEvent,
+) -> Option<(ObserverChunkKey, String)> {
+    let update = event.payload.get("params")?.get("update")?;
+    let update_type = update.get("sessionUpdate")?.as_str()?;
+    if !matches!(
+        update_type,
+        "agent_message_chunk" | "user_message_chunk" | "agent_thought_chunk"
+    ) {
+        return None;
+    }
+
+    let text = update.get("content")?.get("text")?.as_str()?.to_string();
+    let message_id = update
+        .get("messageId")
+        .and_then(|value| value.as_str())
+        .map(ToOwned::to_owned);
+
+    Some((
+        ObserverChunkKey {
+            update_type: update_type.to_string(),
+            message_id,
+            channel_id: event.channel_id.clone(),
+            session_id: event.session_id.clone(),
+            turn_id: event.turn_id.clone(),
+            agent_index: event.agent_index,
+        },
+        text,
+    ))
+}
+
+fn set_observer_chunk_text(payload: &mut serde_json::Value, text: String) {
+    let Some(content) = payload
+        .get_mut("params")
+        .and_then(|params| params.get_mut("update"))
+        .and_then(|update| update.get_mut("content"))
+    else {
+        return;
+    };
+
+    if let Some(content_object) = content.as_object_mut() {
+        content_object.insert("text".to_string(), serde_json::Value::String(text));
+    }
+}
+
+async fn publish_relay_observer_event(
+    publisher: &RelayEventPublisher,
+    keys: &nostr::Keys,
+    agent_pubkey_hex: &str,
+    owner_pubkey_hex: &str,
+    owner_pubkey: &PublicKey,
+    event: observer::ObserverEvent,
+) {
+    let encrypted = match encrypt_observer_payload(keys, owner_pubkey, &event) {
+        Ok(encrypted) => encrypted,
+        Err(error) => {
+            tracing::warn!("failed to encrypt relay observer event: {error}");
+            return;
+        }
+    };
+    let builder = match sprout_sdk::build_agent_observer_frame(
+        owner_pubkey_hex,
+        agent_pubkey_hex,
+        OBSERVER_FRAME_TELEMETRY,
+        &encrypted,
+    ) {
+        Ok(builder) => builder,
+        Err(error) => {
+            tracing::warn!("failed to build relay observer event: {error}");
+            return;
+        }
+    };
+    let signed = match builder.sign_with_keys(keys) {
+        Ok(event) => event,
+        Err(error) => {
+            tracing::warn!("failed to sign relay observer event: {error}");
+            return;
+        }
+    };
+    if let Err(error) = publisher.publish_event(signed).await {
+        tracing::debug!("relay observer event dropped: {error}");
+    }
+}
+
+/// Maximum age (seconds) for an observer control frame to be considered fresh.
+const OBSERVER_CONTROL_FRESHNESS_SECS: i64 = 300;
+
+fn handle_relay_observer_control_event(
+    keys: &nostr::Keys,
+    event: nostr::Event,
+    pool: &mut AgentPool,
+    observer: Option<&observer::ObserverHandle>,
+    owner_pubkey_hex: &str,
+) {
+    // Defense-in-depth: verify signature even though the relay already checked.
+    if let Err(e) = sprout_core::verify_event(&event) {
+        tracing::warn!(error = %e, "observer control frame failed signature verification");
+        return;
+    }
+
+    // Defense-in-depth: verify the sender is the resolved owner.
+    if event.pubkey.to_hex() != owner_pubkey_hex {
+        tracing::warn!(
+            sender = %event.pubkey,
+            expected = %owner_pubkey_hex,
+            "observer control frame from non-owner — dropping"
+        );
+        return;
+    }
+
+    // Freshness: reject stale/replayed frames outside ±5 minute window.
+    let now = chrono::Utc::now().timestamp();
+    let event_ts = event.created_at.as_u64() as i64;
+    if (event_ts - now).unsigned_abs() > OBSERVER_CONTROL_FRESHNESS_SECS as u64 {
+        tracing::warn!(
+            event_ts,
+            now,
+            "observer control frame outside freshness window — dropping"
+        );
+        return;
+    }
+
+    let payload = match decrypt_observer_payload::<serde_json::Value>(keys, &event) {
+        Ok(payload) => payload,
+        Err(error) => {
+            tracing::warn!("failed to decrypt observer control frame: {error}");
+            return;
+        }
+    };
+
+    let command_type = payload.get("type").and_then(|value| value.as_str());
+    if command_type != Some("cancel_turn") {
+        tracing::debug!(payload = %payload, "ignoring unknown observer control frame");
+        return;
+    }
+
+    let Some(channel_id) = payload
+        .get("channelId")
+        .and_then(|value| value.as_str())
+        .and_then(|value| value.parse::<Uuid>().ok())
+    else {
+        tracing::warn!("observer cancel_turn control frame missing valid channelId");
+        return;
+    };
+
+    let fired = cancel_in_flight_task(pool, channel_id, CancelMode::Stop);
+    let status = if fired { "sent" } else { "no_active_turn" };
+    if let Some(observer) = observer {
+        observer.emit(
+            "control_result",
+            None,
+            &observer::ObserverContext {
+                channel_id: Some(channel_id.to_string()),
+                session_id: None,
+                turn_id: None,
+            },
+            serde_json::json!({
+                "type": "cancel_turn",
+                "status": status,
+            }),
+        );
+    }
+}
+
 /// Maximum crashes in a 60-second window before a slot's circuit opens.
 const CIRCUIT_BREAKER_THRESHOLD: usize = 3;
 /// Window for circuit-breaker crash counting.
@@ -531,44 +833,23 @@ async fn tokio_main() -> Result<()> {
     let config = Config::from_cli().map_err(|e| anyhow::anyhow!("configuration error: {e}"))?;
     tracing::info!("sprout-acp starting: {}", config.summary());
 
-    let (observer_control_tx, mut observer_control_rx) =
-        mpsc::channel::<observer::ObserverControlCommand>(32);
-
-    let observer = match config.observer_addr.as_deref() {
-        Some(addr) => {
-            match observer::spawn_observer_server(
-                addr,
-                config.observer_token.clone(),
-                Some(observer_control_tx.clone()),
-            )
-            .await
-            {
-                Ok(handle) => {
-                    tracing::info!(target: "observer", "ACP observer listening on http://{}", handle.addr());
-                    handle.emit(
-                        "harness_started",
-                        None,
-                        &observer::ObserverContext::default(),
-                        serde_json::json!({
-                            "relayUrl": config.relay_url,
-                            "agentCommand": config.agent_command,
-                            "agentArgs": config.agent_args,
-                            "parallelism": config.agents,
-                        }),
-                    );
-                    Some(handle)
-                }
-                Err(error) => {
-                    tracing::warn!(
-                        target: "observer",
-                        "failed to start local ACP observer at {addr}: {error}"
-                    );
-                    None
-                }
-            }
-        }
-        None => None,
-    };
+    let observer = config
+        .relay_observer
+        .then(observer::ObserverHandle::in_process);
+    if let Some(handle) = &observer {
+        handle.emit(
+            "harness_started",
+            None,
+            &observer::ObserverContext::default(),
+            serde_json::json!({
+                "relayUrl": config.relay_url,
+                "agentCommand": config.agent_command,
+                "agentArgs": config.agent_args,
+                "parallelism": config.agents,
+                "relayObserver": config.relay_observer,
+            }),
+        );
+    }
 
     // ── Step 1: Spawn N ACP agent subprocesses and initialize ─────────────────
     //
@@ -739,6 +1020,41 @@ async fn tokio_main() -> Result<()> {
     let mut owner_cache = OwnerCache::new(startup_owner);
     let mut sibling_cache = SiblingCache::new();
 
+    let mut relay_observer_control_rx = None;
+    let mut relay_observer_publisher_task = None;
+    if config.relay_observer {
+        if let (Some(observer), Some(owner_pubkey_hex)) =
+            (observer.clone(), owner_cache.pubkey.clone())
+        {
+            match PublicKey::from_hex(&owner_pubkey_hex) {
+                Ok(owner_pubkey) => {
+                    relay_observer_publisher_task = Some(spawn_relay_observer_publisher(
+                        observer,
+                        relay.event_publisher(),
+                        config.keys.clone(),
+                        pubkey_hex.clone(),
+                        owner_pubkey_hex,
+                        owner_pubkey,
+                    ));
+                    relay
+                        .subscribe_observer_controls()
+                        .await
+                        .map_err(|e| anyhow::anyhow!("observer control subscribe error: {e}"))?;
+                    relay_observer_control_rx = relay.take_observer_control_rx();
+                    tracing::info!("relay observer enabled");
+                }
+                Err(error) => {
+                    tracing::warn!("relay observer disabled: invalid owner pubkey: {error}");
+                }
+            }
+        } else {
+            tracing::warn!(
+                "relay observer requested but no agent owner was resolved at startup; \
+                 observer frames will not be published"
+            );
+        }
+    }
+
     // ── Step 3: Discover channels and build subscription rules ────────────────
     let channel_info_map = relay
         .discover_channels()
@@ -1039,26 +1355,25 @@ async fn tokio_main() -> Result<()> {
                 Some(Err(e)) = join_set.join_next(), if !join_set.is_empty() => {
                     Some(PoolEvent::Panic(e))
                 }
-                control = observer_control_rx.recv() => {
+                control_event = async {
+                    match relay_observer_control_rx.as_mut() {
+                        Some(rx) => rx.recv().await,
+                        None => std::future::pending().await,
+                    }
+                } => {
                     let _ = result_rx;
-                    if let Some(command) = control {
-                        match command {
-                            observer::ObserverControlCommand::CancelTurn { channel_id, respond_to } => {
-                                let fired = cancel_in_flight_task(&mut pool, channel_id, CancelMode::Stop);
-                                if !fired {
-                                    tracing::warn!(
-                                        channel_id = %channel_id,
-                                        "observer cancel requested but no in-flight task — no-op"
-                                    );
-                                }
-                                let status = if fired {
-                                    observer::CancelTurnStatus::Sent
-                                } else {
-                                    observer::CancelTurnStatus::NoActiveTurn
-                                };
-                                let _ = respond_to.send(observer::CancelTurnResponse { status });
+                    match control_event {
+                        Some(event) => {
+                            if let Some(ref owner_hex) = owner_cache.pubkey {
+                                handle_relay_observer_control_event(&config.keys, event, &mut pool, observer.as_ref(), owner_hex);
+                            } else {
+                                tracing::warn!("observer control frame received but no owner resolved — dropping");
                             }
                         }
+                        None => {
+                            relay_observer_control_rx = None;
+                            tracing::warn!("relay observer control channel closed");
+                        }
                     }
                     None
                 }
@@ -1593,6 +1908,10 @@ async fn tokio_main() -> Result<()> {
         }
     }
 
+    if let Some(handle) = relay_observer_publisher_task.take() {
+        handle.abort();
+    }
+
     // Graceful relay shutdown — sends WebSocket close frame and waits up to 5s
     // for the background task to finish, rather than aborting immediately (#40).
     relay.shutdown().await;
@@ -2435,6 +2754,102 @@ mod owner_cache_tests {
     }
 }
 
+#[cfg(test)]
+mod observer_chunk_coalescer_tests {
+    use super::*;
+
+    fn chunk_event(
+        seq: u64,
+        update_type: &str,
+        message_id: &str,
+        text: &str,
+    ) -> observer::ObserverEvent {
+        observer::ObserverEvent {
+            seq,
+            timestamp: format!("2026-04-29T04:00:0{seq}Z"),
+            kind: "acp_read".to_string(),
+            agent_index: Some(0),
+            channel_id: Some("channel-1".to_string()),
+            session_id: Some("session-1".to_string()),
+            turn_id: Some("turn-1".to_string()),
+            payload: serde_json::json!({
+                "jsonrpc": "2.0",
+                "method": "session/update",
+                "params": {
+                    "sessionId": "session-1",
+                    "update": {
+                        "sessionUpdate": update_type,
+                        "messageId": message_id,
+                        "content": {
+                            "type": "text",
+                            "text": text,
+                        },
+                    },
+                },
+            }),
+        }
+    }
+
+    fn non_chunk_event(seq: u64) -> observer::ObserverEvent {
+        observer::ObserverEvent {
+            seq,
+            timestamp: format!("2026-04-29T04:00:0{seq}Z"),
+            kind: "turn_started".to_string(),
+            agent_index: Some(0),
+            channel_id: Some("channel-1".to_string()),
+            session_id: Some("session-1".to_string()),
+            turn_id: Some("turn-1".to_string()),
+            payload: serde_json::json!({ "type": "turn_started" }),
+        }
+    }
+
+    fn chunk_text(event: &observer::ObserverEvent) -> &str {
+        event.payload["params"]["update"]["content"]["text"]
+            .as_str()
+            .expect("chunk text")
+    }
+
+    #[test]
+    fn coalesces_chunks_until_non_chunk_event() {
+        let mut coalescer = ObserverChunkCoalescer::default();
+
+        assert!(coalescer
+            .ingest(chunk_event(1, "agent_message_chunk", "message-1", "hello "))
+            .is_empty());
+        assert!(coalescer
+            .ingest(chunk_event(2, "agent_message_chunk", "message-1", "world"))
+            .is_empty());
+
+        let events = coalescer.ingest(non_chunk_event(3));
+        assert_eq!(events.len(), 2);
+        assert_eq!(events[0].seq, 2);
+        assert_eq!(chunk_text(&events[0]), "hello world");
+        assert_eq!(events[1].kind, "turn_started");
+    }
+
+    #[test]
+    fn keeps_independent_chunk_streams_separate() {
+        let mut coalescer = ObserverChunkCoalescer::default();
+
+        assert!(coalescer
+            .ingest(chunk_event(1, "agent_message_chunk", "message-1", "answer"))
+            .is_empty());
+        assert!(coalescer
+            .ingest(chunk_event(
+                2,
+                "agent_thought_chunk",
+                "thought-1",
+                "thinking"
+            ))
+            .is_empty());
+
+        let events = coalescer.flush();
+        assert_eq!(events.len(), 2);
+        assert_eq!(chunk_text(&events[0]), "answer");
+        assert_eq!(chunk_text(&events[1]), "thinking");
+    }
+}
+
 #[cfg(test)]
 mod sibling_cache_tests {
     use super::*;
@@ -2647,8 +3062,7 @@ mod build_mcp_servers_tests {
             respond_to: config::RespondTo::Anyone,
             respond_to_allowlist: std::collections::HashSet::new(),
             persona_env_vars: vec![],
-            observer_addr: None,
-            observer_token: None,
+            relay_observer: false,
         }
     }
 

crates/sprout-acp/src/observer.rs

@@ -1,32 +1,19 @@
-//! Local observer and control endpoint for ACP session activity.
+//! In-process observer bus for ACP session activity.
 //!
-//! This is intentionally process-local infrastructure: it lets the desktop app
-//! watch the raw ACP JSON-RPC stream and send tightly-scoped control commands
-//! without sending private execution detail through the Sprout relay.
+//! This is intentionally process-local infrastructure: it lets the harness
+//! collect raw ACP JSON-RPC activity and publish owner-scoped encrypted relay
+//! frames without exposing a local HTTP port.
 
 use std::{
     collections::VecDeque,
-    convert::Infallible,
-    net::SocketAddr,
     sync::{
         atomic::{AtomicU64, Ordering},
         Arc, Mutex,
     },
 };
 
-use axum::{
-    extract::{Query, State},
-    http::{header, HeaderMap, HeaderValue, StatusCode},
-    response::{
-        sse::{Event, KeepAlive, Sse},
-        IntoResponse, Response,
-    },
-    routing::{get, post},
-    Json, Router,
-};
-use futures_util::{stream, StreamExt};
-use serde::{Deserialize, Serialize};
-use tokio::sync::{broadcast, mpsc, oneshot};
+use serde::Serialize;
+use tokio::sync::broadcast;
 
 const OBSERVER_BUFFER_CAP: usize = 1_000;
 
@@ -51,10 +38,20 @@ struct ObserverInner {
     tx: broadcast::Sender<ObserverEvent>,
     buffer: Mutex<VecDeque<ObserverEvent>>,
     seq: AtomicU64,
-    addr: SocketAddr,
 }
 
-/// Event delivered over the local observer SSE stream.
+fn new_observer_handle() -> ObserverHandle {
+    let (tx, _) = broadcast::channel(OBSERVER_BUFFER_CAP);
+    ObserverHandle {
+        inner: Arc::new(ObserverInner {
+            tx,
+            buffer: Mutex::new(VecDeque::with_capacity(OBSERVER_BUFFER_CAP)),
+            seq: AtomicU64::new(1),
+        }),
+    }
+}
+
+/// Event delivered through the in-process observer bus.
 #[derive(Clone, Serialize)]
 #[serde(rename_all = "camelCase")]
 pub struct ObserverEvent {
@@ -76,97 +73,10 @@ pub struct ObserverEvent {
     pub payload: serde_json::Value,
 }
 
-/// Commands accepted by the observer control loop.
-#[derive(Debug)]
-pub enum ObserverControlCommand {
-    /// Stop the active turn for a channel, if one exists.
-    CancelTurn {
-        channel_id: uuid::Uuid,
-        respond_to: oneshot::Sender<CancelTurnResponse>,
-    },
-}
-
-/// Response returned by observer control commands.
-#[derive(Debug, Clone, Serialize)]
-#[serde(rename_all = "camelCase")]
-pub struct CancelTurnResponse {
-    /// Result of attempting to send the cancel signal.
-    pub status: CancelTurnStatus,
-}
-
-/// Status for a cancel-turn request.
-#[derive(Debug, Clone, Serialize)]
-#[serde(rename_all = "snake_case")]
-pub enum CancelTurnStatus {
-    /// A cancel signal was sent to an in-flight channel turn.
-    Sent,
-    /// The channel had no active turn at the time of the request.
-    NoActiveTurn,
-}
-
-#[derive(Clone)]
-struct ObserverServerState {
-    observer: ObserverHandle,
-    token: Option<String>,
-    control_tx: Option<mpsc::Sender<ObserverControlCommand>>,
-}
-
-#[derive(Deserialize)]
-struct EventQuery {
-    token: Option<String>,
-}
-
-#[derive(Deserialize)]
-#[serde(rename_all = "camelCase")]
-struct CancelTurnRequest {
-    channel_id: uuid::Uuid,
-}
-
-/// Start the loopback observer HTTP server.
-pub async fn spawn_observer_server(
-    bind_addr: &str,
-    token: Option<String>,
-    control_tx: Option<mpsc::Sender<ObserverControlCommand>>,
-) -> anyhow::Result<ObserverHandle> {
-    let listener = tokio::net::TcpListener::bind(bind_addr).await?;
-    let addr = listener.local_addr()?;
-    if !addr.ip().is_loopback() {
-        anyhow::bail!("observer bind address must be loopback, got {addr}");
-    }
-
-    let (tx, _) = broadcast::channel(OBSERVER_BUFFER_CAP);
-    let observer = ObserverHandle {
-        inner: Arc::new(ObserverInner {
-            tx,
-            buffer: Mutex::new(VecDeque::with_capacity(OBSERVER_BUFFER_CAP)),
-            seq: AtomicU64::new(1),
-            addr,
-        }),
-    };
-    let state = ObserverServerState {
-        observer: observer.clone(),
-        token,
-        control_tx,
-    };
-    let app = Router::new()
-        .route("/events", get(events_handler))
-        .route("/health", get(health_handler))
-        .route("/control/cancel", post(cancel_turn_handler))
-        .with_state(state);
-
-    tokio::spawn(async move {
-        if let Err(error) = axum::serve(listener, app).await {
-            tracing::warn!(target: "observer", "observer server stopped: {error}");
-        }
-    });
-
-    Ok(observer)
-}
-
 impl ObserverHandle {
-    /// Return the bound loopback address.
-    pub fn addr(&self) -> SocketAddr {
-        self.inner.addr
+    /// Create an in-process observer feed.
+    pub fn in_process() -> Self {
+        new_observer_handle()
     }
 
     /// Subscribe to live observer events.
@@ -220,146 +130,6 @@ impl ObserverHandle {
     }
 }
 
-async fn health_handler(State(state): State<ObserverServerState>) -> Response {
-    serde_json::json!({
-        "ok": true,
-        "addr": state.observer.addr().to_string(),
-        "control": state.control_tx.is_some(),
-    })
-    .to_string()
-    .into_response()
-}
-
-async fn events_handler(
-    State(state): State<ObserverServerState>,
-    Query(query): Query<EventQuery>,
-    headers: HeaderMap,
-) -> Response {
-    if !authorized(state.token.as_deref(), query.token.as_deref(), &headers) {
-        return (StatusCode::UNAUTHORIZED, "invalid observer token").into_response();
-    }
-
-    let origin = headers.get(header::ORIGIN).cloned();
-    if !origin_allowed(origin.as_ref()) {
-        return (StatusCode::FORBIDDEN, "forbidden: invalid origin").into_response();
-    }
-
-    let replay = state.observer.snapshot();
-    let replay_stream = stream::iter(replay.into_iter().map(event_to_sse));
-    let live_rx = state.observer.subscribe();
-    let live_stream = stream::unfold(live_rx, |mut rx| async move {
-        loop {
-            match rx.recv().await {
-                Ok(event) => return Some((event_to_sse(event), rx)),
-                Err(broadcast::error::RecvError::Lagged(_)) => continue,
-                Err(broadcast::error::RecvError::Closed) => return None,
-            }
-        }
-    });
-
-    let stream = replay_stream.chain(live_stream);
-    let sse = Sse::new(stream).keep_alive(KeepAlive::default());
-    let mut response = sse.into_response();
-    if let Some(origin) = origin {
-        response
-            .headers_mut()
-            .insert(header::ACCESS_CONTROL_ALLOW_ORIGIN, origin);
-    }
-    response.headers_mut().insert(
-        header::CACHE_CONTROL,
-        HeaderValue::from_static("no-store, no-cache"),
-    );
-    response
-}
-
-async fn cancel_turn_handler(
-    State(state): State<ObserverServerState>,
-    Query(query): Query<EventQuery>,
-    headers: HeaderMap,
-    Json(request): Json<CancelTurnRequest>,
-) -> Response {
-    if !authorized(state.token.as_deref(), query.token.as_deref(), &headers) {
-        return (StatusCode::UNAUTHORIZED, "invalid observer token").into_response();
-    }
-
-    if !origin_allowed(headers.get(header::ORIGIN)) {
-        return (StatusCode::FORBIDDEN, "forbidden: invalid origin").into_response();
-    }
-
-    let Some(control_tx) = state.control_tx else {
-        return (
-            StatusCode::SERVICE_UNAVAILABLE,
-            "observer control is not available",
-        )
-            .into_response();
-    };
-
-    let (respond_to, response_rx) = oneshot::channel();
-    let command = ObserverControlCommand::CancelTurn {
-        channel_id: request.channel_id,
-        respond_to,
-    };
-
-    if control_tx.send(command).await.is_err() {
-        return (
-            StatusCode::SERVICE_UNAVAILABLE,
-            "observer control loop is not available",
-        )
-            .into_response();
-    }
-
-    match tokio::time::timeout(std::time::Duration::from_secs(2), response_rx).await {
-        Ok(Ok(response)) => Json(response).into_response(),
-        Ok(Err(_)) => (
-            StatusCode::SERVICE_UNAVAILABLE,
-            "observer control response was dropped",
-        )
-            .into_response(),
-        Err(_) => (StatusCode::GATEWAY_TIMEOUT, "observer control timed out").into_response(),
-    }
-}
-
-fn authorized(expected: Option<&str>, query_token: Option<&str>, headers: &HeaderMap) -> bool {
-    token_matches(expected, query_token) || token_matches(expected, bearer_token(headers))
-}
-
-fn bearer_token(headers: &HeaderMap) -> Option<&str> {
-    let value = headers.get(header::AUTHORIZATION)?.to_str().ok()?;
-    value.strip_prefix("Bearer ")
-}
-
-fn token_matches(expected: Option<&str>, actual: Option<&str>) -> bool {
-    match expected {
-        Some(expected) => actual == Some(expected),
-        None => true,
-    }
-}
-
-fn origin_allowed(origin: Option<&HeaderValue>) -> bool {
-    let Some(origin) = origin.and_then(|value| value.to_str().ok()) else {
-        return true;
-    };
-
-    origin == "tauri://localhost"
-        || origin == "http://tauri.localhost"
-        || origin == "https://tauri.localhost"
-        || origin.starts_with("http://localhost:")
-        || origin.starts_with("http://127.0.0.1:")
-}
-
-fn event_to_sse(event: ObserverEvent) -> Result<Event, Infallible> {
-    let data = serde_json::to_string(&event).unwrap_or_else(|error| {
-        serde_json::json!({
-            "seq": event.seq,
-            "timestamp": event.timestamp,
-            "kind": "observer_serialize_error",
-            "payload": {"error": error.to_string()},
-        })
-        .to_string()
-    });
-    Ok(Event::default().id(event.seq.to_string()).data(data))
-}
-
 /// Build observer context values from optional channel/session/turn IDs.
 pub fn context_for(
     channel_id: Option<uuid::Uuid>,
@@ -372,47 +142,3 @@ pub fn context_for(
         turn_id,
     }
 }
-
-#[cfg(test)]
-mod tests {
-    use axum::http::{HeaderMap, HeaderValue};
-
-    use super::{authorized, origin_allowed, token_matches};
-
-    #[test]
-    fn observer_token_requires_exact_match_when_configured() {
-        assert!(token_matches(Some("secret"), Some("secret")));
-        assert!(!token_matches(Some("secret"), Some("wrong")));
-        assert!(!token_matches(Some("secret"), None));
-        assert!(token_matches(None, None));
-        assert!(token_matches(None, Some("anything")));
-    }
-
-    #[test]
-    fn observer_authorization_accepts_bearer_token() {
-        let mut headers = HeaderMap::new();
-        headers.insert(
-            axum::http::header::AUTHORIZATION,
-            HeaderValue::from_static("Bearer secret"),
-        );
-        assert!(authorized(Some("secret"), None, &headers));
-        assert!(!authorized(Some("other"), None, &headers));
-    }
-
-    #[test]
-    fn observer_origin_allows_tauri_and_loopback_only() {
-        assert!(origin_allowed(None));
-        assert!(origin_allowed(Some(&HeaderValue::from_static(
-            "tauri://localhost"
-        ))));
-        assert!(origin_allowed(Some(&HeaderValue::from_static(
-            "http://localhost:1420"
-        ))));
-        assert!(origin_allowed(Some(&HeaderValue::from_static(
-            "http://127.0.0.1:1420"
-        ))));
-        assert!(!origin_allowed(Some(&HeaderValue::from_static(
-            "https://example.com"
-        ))));
-    }
-}

crates/sprout-acp/src/relay.rs

@@ -65,7 +65,8 @@ use futures_util::{SinkExt, StreamExt};
 use nostr::{Event, EventBuilder, Keys, Kind, Tag, Url as NostrUrl};
 use serde_json::{json, Value};
 use sprout_core::kind::{
-    KIND_MEMBER_ADDED_NOTIFICATION, KIND_MEMBER_REMOVED_NOTIFICATION, KIND_TYPING_INDICATOR,
+    KIND_AGENT_OBSERVER_FRAME, KIND_MEMBER_ADDED_NOTIFICATION, KIND_MEMBER_REMOVED_NOTIFICATION,
+    KIND_TYPING_INDICATOR,
 };
 use tokio::sync::mpsc;
 use tokio::time::timeout;
@@ -325,6 +326,8 @@ enum RelayMessage {
 
 /// Subscription ID for the global membership notification subscription.
 const MEMBERSHIP_NOTIF_SUB_ID: &str = "membership-notif";
+/// Subscription ID for encrypted owner-to-agent observer control frames.
+const OBSERVER_CONTROL_SUB_ID: &str = "agent-observer-control";
 
 /// Commands sent from `HarnessRelay` to the background WebSocket task.
 enum RelayCommand {
@@ -342,6 +345,8 @@ enum RelayCommand {
     Shutdown,
     /// Subscribe to global membership notifications.
     SubscribeMembership,
+    /// Subscribe to encrypted observer control frames addressed to this agent.
+    SubscribeObserverControls,
     /// Publish a signed event to the relay (for typing indicators, etc.).
     PublishEvent { event: Box<Event> },
     /// Set the startup watermark timestamp for Finding #22.
@@ -366,6 +371,8 @@ type WsStream = WebSocketStream<MaybeTlsStream<tokio::net::TcpStream>>;
 pub struct HarnessRelay {
     /// Receiver for events forwarded by the background task.
     event_rx: mpsc::Receiver<Option<SproutEvent>>,
+    /// Receiver for encrypted observer control events addressed to this agent.
+    observer_control_rx: Option<mpsc::Receiver<Event>>,
     /// Sender for commands to the background task.
     cmd_tx: mpsc::Sender<RelayCommand>,
     /// HTTP client for REST API calls.
@@ -385,6 +392,24 @@ pub struct HarnessRelay {
     bg_handle: Option<tokio::task::JoinHandle<()>>,
 }
 
+/// Cloneable publisher handle for signed events on the relay background socket.
+#[derive(Clone)]
+pub struct RelayEventPublisher {
+    cmd_tx: mpsc::Sender<RelayCommand>,
+}
+
+impl RelayEventPublisher {
+    /// Publish a signed event through the relay background task.
+    pub async fn publish_event(&self, event: Event) -> Result<(), RelayError> {
+        self.cmd_tx
+            .send(RelayCommand::PublishEvent {
+                event: Box::new(event),
+            })
+            .await
+            .map_err(|_| RelayError::ConnectionClosed)
+    }
+}
+
 impl HarnessRelay {
     // ── Public API ────────────────────────────────────────────────────────────
 
@@ -401,6 +426,8 @@ impl HarnessRelay {
         let (ws, handshake_buffer) = do_connect(relay_url, keys, api_token).await?;
 
         let (event_tx, event_rx) = mpsc::channel::<Option<SproutEvent>>(event_channel_capacity());
+        let (observer_control_tx, observer_control_rx) =
+            mpsc::channel::<Event>(event_channel_capacity());
         let (cmd_tx, cmd_rx) = mpsc::channel::<RelayCommand>(CMD_CHANNEL_CAPACITY);
 
         let bg_keys = keys.clone();
@@ -413,6 +440,7 @@ impl HarnessRelay {
                 ws,
                 handshake_buffer,
                 event_tx,
+                observer_control_tx,
                 cmd_rx,
                 bg_keys,
                 bg_relay_url,
@@ -424,6 +452,7 @@ impl HarnessRelay {
 
         Ok(Self {
             event_rx,
+            observer_control_rx: Some(observer_control_rx),
             cmd_tx,
             http: reqwest::Client::builder()
                 .timeout(std::time::Duration::from_secs(10))
@@ -517,6 +546,27 @@ impl HarnessRelay {
         Ok(())
     }
 
+    /// Subscribe to encrypted observer control frames addressed to this agent.
+    pub async fn subscribe_observer_controls(&mut self) -> Result<(), RelayError> {
+        self.cmd_tx
+            .send(RelayCommand::SubscribeObserverControls)
+            .await
+            .map_err(|_| RelayError::ConnectionClosed)?;
+        Ok(())
+    }
+
+    /// Take the observer-control receiver for polling outside this relay object.
+    pub fn take_observer_control_rx(&mut self) -> Option<mpsc::Receiver<Event>> {
+        self.observer_control_rx.take()
+    }
+
+    /// Return a cloneable publisher handle for signed relay events.
+    pub fn event_publisher(&self) -> RelayEventPublisher {
+        RelayEventPublisher {
+            cmd_tx: self.cmd_tx.clone(),
+        }
+    }
+
     /// Unsubscribe from a channel.
     #[allow(dead_code)]
     pub async fn unsubscribe_channel(&mut self, channel_id: Uuid) -> Result<(), RelayError> {
@@ -717,6 +767,8 @@ struct BgState {
     membership_last_seen: Option<u64>,
     /// Whether the membership notification subscription is active.
     membership_sub_active: bool,
+    /// Whether the observer control subscription is active.
+    observer_control_sub_active: bool,
     /// Oldest dropped channel-event timestamp per channel, keyed by channel_id.
     /// Mirrors `membership_dropped_since` but for ordinary channel events.
     /// On reconnect resubscribe, `since` = min(last_seen, channel_dropped_since).
@@ -749,6 +801,7 @@ impl BgState {
             membership_dropped_since: None,
             membership_last_seen: None,
             membership_sub_active: false,
+            observer_control_sub_active: false,
             channel_dropped_since: HashMap::new(),
             proactive_resubscribe_needed: false,
             startup_watermark: None,
@@ -841,6 +894,9 @@ fn apply_command_to_state(state: &mut BgState, cmd: RelayCommand) {
         RelayCommand::SubscribeMembership => {
             state.membership_sub_active = true;
         }
+        RelayCommand::SubscribeObserverControls => {
+            state.observer_control_sub_active = true;
+        }
         RelayCommand::SetStartupWatermark { ts } => {
             state.startup_watermark = Some(ts);
             if state.membership_last_seen.is_none() {
@@ -942,6 +998,17 @@ async fn execute_connected_command(
                 false
             }
         }
+        RelayCommand::SubscribeObserverControls => {
+            let sent = send_observer_control_subscribe(ws, agent_pubkey_hex).await;
+            if sent {
+                state.observer_control_sub_active = true;
+                true
+            } else {
+                warn!("observer control subscribe REQ failed — recording intent for reconnect");
+                state.observer_control_sub_active = true;
+                false
+            }
+        }
         RelayCommand::PublishEvent { event } => {
             let msg = json!(["EVENT", event]);
             if let Ok(text) = serde_json::to_string(&msg) {
@@ -984,6 +1051,7 @@ async fn run_background_task(
     mut ws: WsStream,
     initial_handshake_buffer: std::collections::VecDeque<RelayMessage>,
     event_tx: mpsc::Sender<Option<SproutEvent>>,
+    observer_control_tx: mpsc::Sender<Event>,
     mut cmd_rx: mpsc::Receiver<RelayCommand>,
     keys: Keys,
     relay_url: String,
@@ -998,6 +1066,7 @@ async fn run_background_task(
         &mut ws,
         initial_handshake_buffer,
         &event_tx,
+        &observer_control_tx,
         &mut state,
         &keys,
         &relay_url,
@@ -1019,6 +1088,7 @@ async fn run_background_task(
             api_token.as_deref(),
             &agent_pubkey_hex,
             &event_tx,
+            &observer_control_tx,
         )
         .await
         {
@@ -1042,6 +1112,7 @@ async fn run_background_task(
                         api_token.as_deref(),
                         &agent_pubkey_hex,
                         &event_tx,
+                        &observer_control_tx,
                         true,
                     )
                     .await,
@@ -1082,6 +1153,7 @@ async fn run_background_task(
                     api_token.as_deref(),
                     &agent_pubkey_hex,
                     &event_tx,
+                    &observer_control_tx,
                 )
                 .await
                 {
@@ -1111,6 +1183,7 @@ async fn run_background_task(
                                 api_token.as_deref(),
                                 &agent_pubkey_hex,
                                 &event_tx,
+                                &observer_control_tx,
                                 true,
                             )
                             .await,
@@ -1143,6 +1216,7 @@ async fn run_background_task(
                                 msg,
                                 &mut ws,
                                 &event_tx,
+                                &observer_control_tx,
                                 &mut state,
                                 &keys,
                                 &relay_url,
@@ -1176,6 +1250,7 @@ async fn run_background_task(
                         api_token.as_deref(),
                         &agent_pubkey_hex,
                         &event_tx,
+                    &observer_control_tx,
                     )
                     .await;
                     match outcome {
@@ -1195,7 +1270,7 @@ async fn run_background_task(
                         if matches!(
                             wait_for_reconnect(
                                 &mut ws, &mut cmd_rx, &mut state, &keys, &relay_url,
-                                api_token.as_deref(), &agent_pubkey_hex, &event_tx, true,
+                                api_token.as_deref(), &agent_pubkey_hex, &event_tx, &observer_control_tx, true,
                             ).await,
                             ReconnectOutcome::Shutdown
                         ) { return; }
@@ -1215,7 +1290,7 @@ async fn run_background_task(
                         if matches!(
                             wait_for_reconnect(
                                 &mut ws, &mut cmd_rx, &mut state, &keys, &relay_url,
-                                api_token.as_deref(), &agent_pubkey_hex, &event_tx, true,
+                                api_token.as_deref(), &agent_pubkey_hex, &event_tx, &observer_control_tx, true,
                             ).await,
                             ReconnectOutcome::Shutdown
                         ) { return; }
@@ -1249,6 +1324,7 @@ async fn run_background_task(
                             match try_autonomous_reconnect(
                                 &mut ws, &mut cmd_rx, &mut state, &keys, &relay_url,
                                 api_token.as_deref(), &agent_pubkey_hex, &event_tx,
+                            &observer_control_tx,
                             ).await {
                                 ReconnectOutcome::Shutdown => return,
                                 ReconnectOutcome::Ok => {
@@ -1261,7 +1337,7 @@ async fn run_background_task(
                                     if matches!(
                                         wait_for_reconnect(
                                             &mut ws, &mut cmd_rx, &mut state, &keys, &relay_url,
-                                            api_token.as_deref(), &agent_pubkey_hex, &event_tx, true,
+                                            api_token.as_deref(), &agent_pubkey_hex, &event_tx, &observer_control_tx, true,
                                         ).await,
                                         ReconnectOutcome::Shutdown
                                     ) { return; }
@@ -1286,6 +1362,7 @@ async fn run_background_task(
                     match try_autonomous_reconnect(
                         &mut ws, &mut cmd_rx, &mut state, &keys, &relay_url,
                         api_token.as_deref(), &agent_pubkey_hex, &event_tx,
+                    &observer_control_tx,
                     ).await {
                         ReconnectOutcome::Shutdown => return,
                         ReconnectOutcome::Ok => {
@@ -1298,7 +1375,7 @@ async fn run_background_task(
                             if matches!(
                                 wait_for_reconnect(
                                     &mut ws, &mut cmd_rx, &mut state, &keys, &relay_url,
-                                    api_token.as_deref(), &agent_pubkey_hex, &event_tx, true,
+                                    api_token.as_deref(), &agent_pubkey_hex, &event_tx, &observer_control_tx, true,
                                 ).await,
                                 ReconnectOutcome::Shutdown
                             ) { return; }
@@ -1316,6 +1393,7 @@ async fn run_background_task(
                         match try_autonomous_reconnect(
                             &mut ws, &mut cmd_rx, &mut state, &keys, &relay_url,
                             api_token.as_deref(), &agent_pubkey_hex, &event_tx,
+                        &observer_control_tx,
                         ).await {
                             ReconnectOutcome::Shutdown => return,
                             ReconnectOutcome::Ok => {
@@ -1328,7 +1406,7 @@ async fn run_background_task(
                                 if matches!(
                                     wait_for_reconnect(
                                         &mut ws, &mut cmd_rx, &mut state, &keys, &relay_url,
-                                        api_token.as_deref(), &agent_pubkey_hex, &event_tx, true,
+                                        api_token.as_deref(), &agent_pubkey_hex, &event_tx, &observer_control_tx, true,
                                     ).await,
                                     ReconnectOutcome::Shutdown
                                 ) { return; }
@@ -1365,6 +1443,7 @@ async fn handle_ws_message(
     msg: Message,
     ws: &mut WsStream,
     event_tx: &mpsc::Sender<Option<SproutEvent>>,
+    observer_control_tx: &mpsc::Sender<Event>,
     state: &mut BgState,
     keys: &Keys,
     relay_url: &str,
@@ -1386,7 +1465,15 @@ async fn handle_ws_message(
                     subscription_id,
                     event,
                 } => {
-                    if subscription_id == MEMBERSHIP_NOTIF_SUB_ID {
+                    if subscription_id == OBSERVER_CONTROL_SUB_ID {
+                        match observer_control_tx.try_send(*event) {
+                            Ok(()) => {}
+                            Err(mpsc::error::TrySendError::Full(_)) => {
+                                warn!("observer control event dropped because control channel is full");
+                            }
+                            Err(mpsc::error::TrySendError::Closed(_)) => return false,
+                        }
+                    } else if subscription_id == MEMBERSHIP_NOTIF_SUB_ID {
                         // Membership notification — extract channel UUID from h tag.
                         let channel_uuid = match extract_h_tag_uuid(&event) {
                             Some(uuid) => uuid,
@@ -1535,7 +1622,15 @@ async fn handle_ws_message(
                     // the attempt — if the send fails and triggers reconnect,
                     // resubscribe_after_reconnect() needs the subscription to
                     // still be in state so it can restore it.
-                    if subscription_id == MEMBERSHIP_NOTIF_SUB_ID {
+                    if subscription_id == OBSERVER_CONTROL_SUB_ID {
+                        let sent = send_observer_control_subscribe(ws, agent_pubkey_hex).await;
+                        if sent {
+                            state.observer_control_sub_active = true;
+                        } else {
+                            warn!("observer control resubscribe failed after CLOSED — triggering reconnect");
+                            return false;
+                        }
+                    } else if subscription_id == MEMBERSHIP_NOTIF_SUB_ID {
                         let since =
                             match (state.membership_dropped_since, state.membership_last_seen) {
                                 (Some(d), Some(l)) => Some(d.min(l)),
@@ -1652,6 +1747,7 @@ async fn process_handshake_buffer(
     ws: &mut WsStream,
     buffer: std::collections::VecDeque<RelayMessage>,
     event_tx: &mpsc::Sender<Option<SproutEvent>>,
+    observer_control_tx: &mpsc::Sender<Event>,
     state: &mut BgState,
     keys: &Keys,
     relay_url: &str,
@@ -1694,6 +1790,7 @@ async fn process_handshake_buffer(
                 Message::Text(text.into()),
                 ws,
                 event_tx,
+                observer_control_tx,
                 state,
                 keys,
                 relay_url,
@@ -1769,6 +1866,13 @@ async fn resubscribe_after_reconnect(
         }
     }
 
+    if state.observer_control_sub_active
+        && !send_observer_control_subscribe(ws, agent_pubkey_hex).await
+    {
+        warn!("failed to resubscribe observer controls after reconnect");
+        all_ok = false;
+    }
+
     all_ok
 }
 
@@ -1855,6 +1959,7 @@ async fn try_autonomous_reconnect(
     api_token: Option<&str>,
     agent_pubkey_hex: &str,
     event_tx: &mpsc::Sender<Option<SproutEvent>>,
+    observer_control_tx: &mpsc::Sender<Event>,
 ) -> ReconnectOutcome {
     // Finding #42: 5 attempts, up to 16s base backoff.
     let backoffs = [
@@ -1880,6 +1985,7 @@ async fn try_autonomous_reconnect(
                     ws,
                     handshake_buffer,
                     event_tx,
+                    observer_control_tx,
                     state,
                     keys,
                     relay_url,
@@ -1956,6 +2062,7 @@ async fn wait_for_reconnect(
     api_token: Option<&str>,
     agent_pubkey_hex: &str,
     event_tx: &mpsc::Sender<Option<SproutEvent>>,
+    observer_control_tx: &mpsc::Sender<Event>,
     skip_drain: bool,
 ) -> ReconnectOutcome {
     if !skip_drain {
@@ -1993,6 +2100,7 @@ async fn wait_for_reconnect(
                     ws,
                     handshake_buffer,
                     event_tx,
+                    observer_control_tx,
                     state,
                     keys,
                     relay_url,
@@ -2170,6 +2278,41 @@ async fn send_membership_subscribe(
     }
 }
 
+/// Send a NIP-01 REQ for owner-to-agent observer control frames.
+async fn send_observer_control_subscribe(ws: &mut WsStream, agent_pubkey_hex: &str) -> bool {
+    let req = json!([
+        "REQ",
+        OBSERVER_CONTROL_SUB_ID,
+        {
+            "kinds": [KIND_AGENT_OBSERVER_FRAME],
+            "#p": [agent_pubkey_hex],
+            "since": std::time::SystemTime::now()
+                .duration_since(std::time::UNIX_EPOCH)
+                .unwrap_or_default()
+                .as_secs(),
+        }
+    ]);
+
+    match serde_json::to_string(&req) {
+        Ok(text) => {
+            match ws_send_timeout(ws, Message::Text(text.into()), WS_SEND_TIMEOUT_SECS).await {
+                Ok(()) => {
+                    debug!("subscribed to observer control frames");
+                    true
+                }
+                Err(e) => {
+                    warn!("failed to send observer control REQ: {e}");
+                    false
+                }
+            }
+        }
+        Err(e) => {
+            warn!("failed to serialize observer control REQ: {e}");
+            false
+        }
+    }
+}
+
 /// Send a WebSocket message with a hard timeout.
 ///
 /// All `ws.send()` calls go through here so a stalled TCP socket can't wedge

crates/sprout-core/src/kind.rs

@@ -88,6 +88,8 @@ pub const KIND_PRESENCE_UPDATE: u32 = 20001;
 pub const KIND_PAIRING: u32 = 24134;
 /// Ephemeral: typing indicator for a channel.
 pub const KIND_TYPING_INDICATOR: u32 = 20002;
+/// Ephemeral: owner-scoped encrypted agent observer telemetry and control frame.
+pub const KIND_AGENT_OBSERVER_FRAME: u32 = 24200;
 
 // Stream messaging
 /// NIP-29 group chat message kind. V1 used kind:10001 (replaceable range — wrong), then 40001.
@@ -284,6 +286,7 @@ pub const ALL_KINDS: &[u32] = &[
     KIND_PRESENCE_UPDATE,
     KIND_TYPING_INDICATOR,
     KIND_PAIRING,
+    KIND_AGENT_OBSERVER_FRAME,
     KIND_STREAM_MESSAGE,
     KIND_STREAM_MESSAGE_V2,
     KIND_STREAM_MESSAGE_EDIT,

crates/sprout-core/src/lib.rs

@@ -17,6 +17,8 @@ pub mod filter;
 pub mod kind;
 /// Network utilities — SSRF-safe IP classification.
 pub mod network;
+/// Agent observer frame helpers.
+pub mod observer;
 /// NIP-AB device pairing — crypto primitives, message types, and errors.
 pub mod pairing;
 /// Presence status types shared across crates.

crates/sprout-core/src/observer.rs

@@ -0,0 +1,156 @@
+//! Agent observer frame helpers.
+//!
+//! Observer frames are transient, owner-scoped agent telemetry/control messages.
+//! They use a Sprout ephemeral event kind and carry NIP-44 encrypted JSON in the
+//! event content so relays can route frames without reading ACP internals.
+
+use nostr::{nips::nip44, Event, Keys, PublicKey};
+use serde::{de::DeserializeOwned, Serialize};
+use thiserror::Error;
+use zeroize::Zeroize;
+
+/// Tag name that identifies the agent pubkey the observer frame belongs to.
+pub const OBSERVER_AGENT_TAG: &str = "agent";
+/// Tag name that identifies the cleartext frame direction.
+pub const OBSERVER_FRAME_TAG: &str = "frame";
+/// Frame value for agent-to-owner observer telemetry.
+pub const OBSERVER_FRAME_TELEMETRY: &str = "telemetry";
+/// Frame value for owner-to-agent observer control commands.
+pub const OBSERVER_FRAME_CONTROL: &str = "control";
+/// Minimum plausible NIP-44 v2 ciphertext length.
+pub const NIP44_MIN_CONTENT_LEN: usize = 132;
+/// Maximum NIP-44 v2 ciphertext length.
+pub const NIP44_MAX_CONTENT_LEN: usize = 87_472;
+/// Maximum observer plaintext JSON size accepted by helpers.
+pub const OBSERVER_MAX_PLAINTEXT_LEN: usize = 65_535;
+
+/// Errors returned by observer payload encryption/decryption helpers.
+#[derive(Debug, Error)]
+pub enum ObserverPayloadError {
+    /// NIP-44 encryption or decryption failed.
+    #[error("NIP-44 error: {0}")]
+    Nip44(#[from] nip44::Error),
+    /// JSON serialization or deserialization failed.
+    #[error("JSON error: {0}")]
+    Json(#[from] serde_json::Error),
+    /// Ciphertext did not fit the expected NIP-44 v2 length envelope.
+    #[error("invalid NIP-44 ciphertext length: {0}")]
+    InvalidCiphertextLength(usize),
+    /// Decrypted JSON exceeded the observer plaintext size limit.
+    #[error("observer plaintext exceeds {max} bytes (got {got})")]
+    PlaintextTooLarge {
+        /// Maximum accepted plaintext bytes.
+        max: usize,
+        /// Actual plaintext byte count.
+        got: usize,
+    },
+}
+
+/// Returns true when `content` fits the NIP-44 v2 ciphertext length envelope.
+pub fn content_looks_like_nip44(content: &str) -> bool {
+    (NIP44_MIN_CONTENT_LEN..=NIP44_MAX_CONTENT_LEN).contains(&content.len())
+}
+
+/// Serialize and NIP-44 encrypt an observer payload for `recipient`.
+pub fn encrypt_observer_payload<T: Serialize>(
+    sender_keys: &Keys,
+    recipient: &PublicKey,
+    payload: &T,
+) -> Result<String, ObserverPayloadError> {
+    let mut plaintext = serde_json::to_string(payload)?;
+    if plaintext.len() > OBSERVER_MAX_PLAINTEXT_LEN {
+        let got = plaintext.len();
+        plaintext.zeroize();
+        return Err(ObserverPayloadError::PlaintextTooLarge {
+            max: OBSERVER_MAX_PLAINTEXT_LEN,
+            got,
+        });
+    }
+
+    let encrypted = nip44::encrypt(
+        sender_keys.secret_key(),
+        recipient,
+        &plaintext,
+        nip44::Version::V2,
+    )?;
+    plaintext.zeroize();
+    Ok(encrypted)
+}
+
+/// NIP-44 decrypt and deserialize an observer payload from `event`.
+pub fn decrypt_observer_payload<T: DeserializeOwned>(
+    recipient_keys: &Keys,
+    event: &Event,
+) -> Result<T, ObserverPayloadError> {
+    if !content_looks_like_nip44(&event.content) {
+        return Err(ObserverPayloadError::InvalidCiphertextLength(
+            event.content.len(),
+        ));
+    }
+
+    let mut plaintext = nip44::decrypt(
+        recipient_keys.secret_key(),
+        &event.pubkey,
+        event.content.as_str(),
+    )?;
+    if plaintext.len() > OBSERVER_MAX_PLAINTEXT_LEN {
+        let got = plaintext.len();
+        plaintext.zeroize();
+        return Err(ObserverPayloadError::PlaintextTooLarge {
+            max: OBSERVER_MAX_PLAINTEXT_LEN,
+            got,
+        });
+    }
+
+    let result = serde_json::from_str(&plaintext);
+    plaintext.zeroize();
+    Ok(result?)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use nostr::{EventBuilder, Kind, Tag};
+
+    #[test]
+    fn observer_payload_round_trips_with_nip44() {
+        let sender = Keys::generate();
+        let recipient = Keys::generate();
+        let payload = serde_json::json!({
+            "type": "turn_started",
+            "turnId": "turn-1"
+        });
+        let encrypted = encrypt_observer_payload(&sender, &recipient.public_key(), &payload)
+            .expect("encrypt payload");
+        assert!(content_looks_like_nip44(&encrypted));
+
+        let event = EventBuilder::new(
+            Kind::Custom(crate::kind::KIND_AGENT_OBSERVER_FRAME as u16),
+            encrypted,
+            [Tag::public_key(recipient.public_key())],
+        )
+        .sign_with_keys(&sender)
+        .expect("sign event");
+        let decrypted: serde_json::Value =
+            decrypt_observer_payload(&recipient, &event).expect("decrypt payload");
+        assert_eq!(decrypted, payload);
+    }
+
+    #[test]
+    fn observer_payload_rejects_short_ciphertext() {
+        let sender = Keys::generate();
+        let recipient = Keys::generate();
+        let event = EventBuilder::new(
+            Kind::Custom(crate::kind::KIND_AGENT_OBSERVER_FRAME as u16),
+            "not encrypted",
+            [Tag::public_key(recipient.public_key())],
+        )
+        .sign_with_keys(&sender)
+        .expect("sign event");
+
+        assert!(matches!(
+            decrypt_observer_payload::<serde_json::Value>(&recipient, &event),
+            Err(ObserverPayloadError::InvalidCiphertextLength(_))
+        ));
+    }
+}

crates/sprout-relay/src/handlers/event.rs

@@ -4,9 +4,15 @@ use std::sync::Arc;
 
 use tracing::{debug, error, info, warn};
 
-use nostr::Event;
+use nostr::{Event, PublicKey};
 use sprout_core::event::StoredEvent;
-use sprout_core::kind::{event_kind_u32, is_ephemeral, KIND_GIFT_WRAP, KIND_PRESENCE_UPDATE};
+use sprout_core::kind::{
+    event_kind_u32, is_ephemeral, KIND_AGENT_OBSERVER_FRAME, KIND_GIFT_WRAP, KIND_PRESENCE_UPDATE,
+};
+use sprout_core::observer::{
+    content_looks_like_nip44, OBSERVER_AGENT_TAG, OBSERVER_FRAME_CONTROL, OBSERVER_FRAME_TAG,
+    OBSERVER_FRAME_TELEMETRY,
+};
 use sprout_core::verification::verify_event;
 
 use crate::connection::{AuthState, ConnectionState};
@@ -207,6 +213,24 @@ pub async fn handle_event(event: Event, conn: Arc<ConnectionState>, state: Arc<A
         return;
     }
 
+    // ── Agent observer frames are owner-scoped, encrypted, and never stored ──
+    if kind_u32 == KIND_AGENT_OBSERVER_FRAME {
+        if !scopes.is_empty()
+            && !scopes.contains(&sprout_auth::Scope::MessagesWrite)
+            && !has_proxy_scope
+        {
+            reject("scope");
+            conn.send(RelayMessage::ok(
+                &event_id_hex,
+                false,
+                "restricted: insufficient scope for agent observer frames",
+            ));
+            return;
+        }
+        handle_agent_observer_event(event, conn_id, &event_id_hex, conn, state).await;
+        return;
+    }
+
     // ── Ephemeral events are WS-only (never stored) ──────────────────────
     // Scope enforcement for ephemeral kinds: require MessagesWrite or
     // ProxySubmit. Persistent events skip this gate and rely on
@@ -435,11 +459,262 @@ async fn handle_ephemeral_event(
     conn.send(RelayMessage::ok(event_id_hex, true, ""));
 }
 
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum AgentObserverDirection {
+    Telemetry,
+    Control,
+}
+
+#[derive(Debug, Clone, Copy)]
+struct AgentObserverRoute {
+    agent: PublicKey,
+    owner: PublicKey,
+    direction: AgentObserverDirection,
+}
+
+/// Handle encrypted agent observer frames (kind 24200).
+///
+/// These frames bypass storage and are routed as global ephemeral events. The
+/// relay gates publication by the existing `agent_owner_pubkey` mapping and
+/// gates subscription in the REQ handler via the cleartext `p` tag.
+async fn handle_agent_observer_event(
+    event: Event,
+    conn_id: uuid::Uuid,
+    event_id_hex: &str,
+    conn: Arc<ConnectionState>,
+    state: Arc<AppState>,
+) {
+    let event_clone = event.clone();
+    let verify_result = tokio::task::spawn_blocking(move || verify_event(&event_clone)).await;
+    match verify_result {
+        Ok(Ok(())) => {}
+        Ok(Err(e)) => {
+            conn.send(RelayMessage::ok(
+                event_id_hex,
+                false,
+                &format!("invalid: {e}"),
+            ));
+            return;
+        }
+        Err(_) => {
+            conn.send(RelayMessage::ok(
+                event_id_hex,
+                false,
+                "error: internal error",
+            ));
+            return;
+        }
+    }
+
+    // Freshness check: reject observer frames with stale/future timestamps
+    let now = chrono::Utc::now().timestamp();
+    let event_ts = event.created_at.as_u64() as i64;
+    if (event_ts - now).unsigned_abs() > 300 {
+        conn.send(RelayMessage::ok(
+            event_id_hex,
+            false,
+            "invalid: observer frame timestamp outside ±5 minute freshness window",
+        ));
+        return;
+    }
+
+    let route = match agent_observer_route(&event) {
+        Ok(Some(route)) => route,
+        Ok(None) => {
+            // Unknown frame value — silently drop, no error to publisher.
+            conn.send(RelayMessage::ok(event_id_hex, true, ""));
+            return;
+        }
+        Err(message) => {
+            reject("invalid");
+            conn.send(RelayMessage::ok(event_id_hex, false, &message));
+            return;
+        }
+    };
+
+    let agent_bytes = route.agent.serialize().to_vec();
+    let owner_bytes = route.owner.serialize().to_vec();
+    let cache_key = (agent_bytes.clone(), owner_bytes.clone());
+    let is_owner = match state.observer_owner_cache.get(&cache_key) {
+        Some(cached) => cached,
+        None => {
+            let result = state.db.is_agent_owner(&agent_bytes, &owner_bytes).await;
+            match result {
+                Ok(v) => {
+                    state.observer_owner_cache.insert(cache_key, v);
+                    v
+                }
+                Err(e) => {
+                    warn!(conn_id = %conn_id, event_id = %event_id_hex, "agent observer owner check failed: {e}");
+                    conn.send(RelayMessage::ok(
+                        event_id_hex,
+                        false,
+                        "error: internal server error",
+                    ));
+                    return;
+                }
+            }
+        }
+    };
+    if !is_owner {
+        reject("auth");
+        conn.send(RelayMessage::ok(
+            event_id_hex,
+            false,
+            "restricted: observer frame is not authorized for this agent owner",
+        ));
+        return;
+    }
+
+    // Rate limit telemetry frames only (100/sec per agent).
+    // Control frames (owner → agent) bypass the limiter — they are rare and must not
+    // be starved by bursty telemetry from the agent.
+    if matches!(route.direction, AgentObserverDirection::Telemetry) {
+        let agent_key: [u8; 32] = agent_bytes.as_slice().try_into().unwrap_or([0u8; 32]);
+        let now = std::time::Instant::now();
+        let mut entry = state
+            .observer_rate_limiter
+            .entry(agent_key)
+            .or_insert((0, now));
+        let (count, window_start) = entry.value_mut();
+        if now.duration_since(*window_start).as_secs() >= 1 {
+            *count = 1;
+            *window_start = now;
+        } else {
+            *count += 1;
+            if *count > 100 {
+                conn.send(RelayMessage::ok(
+                    event_id_hex,
+                    false,
+                    "rate-limited: observer frame rate exceeded (100/sec per agent)",
+                ));
+                return;
+            }
+        }
+    }
+
+    let event_json = match serde_json::to_string(&event) {
+        Ok(json) => json,
+        Err(e) => {
+            error!(event_id = %event_id_hex, "Failed to serialize agent observer event: {e}");
+            conn.send(RelayMessage::ok(
+                event_id_hex,
+                false,
+                "error: internal server error",
+            ));
+            return;
+        }
+    };
+
+    state.mark_local_event(&event.id);
+    if let Err(e) = state.pubsub.publish_event(uuid::Uuid::nil(), &event).await {
+        state.local_event_ids.invalidate(&event.id.to_bytes());
+        warn!(conn_id = %conn_id, event_id = %event_id_hex, "Agent observer publish failed: {e}");
+    }
+
+    let stored_event = StoredEvent::new(event.clone(), None);
+    let matches = state.sub_registry.fan_out(&stored_event);
+    metrics::histogram!("sprout_fanout_recipients").record(matches.len() as f64);
+    debug!(
+        event_id = %event_id_hex,
+        agent = %route.agent.to_hex(),
+        owner = %route.owner.to_hex(),
+        direction = ?route.direction,
+        match_count = matches.len(),
+        "Agent observer fan-out"
+    );
+
+    let mut drop_count = 0u32;
+    for (target_conn_id, sub_id) in &matches {
+        let msg = format!(r#"["EVENT","{}",{}]"#, sub_id, event_json);
+        if !state.conn_manager.send_to(*target_conn_id, msg) {
+            drop_count += 1;
+        }
+    }
+    if drop_count > 0 {
+        tracing::warn!(
+            event_id = %event_id_hex,
+            drop_count,
+            "agent observer fan-out: {drop_count} connection(s) cancelled due to full/closed buffers"
+        );
+    }
+
+    conn.send(RelayMessage::ok(event_id_hex, true, ""));
+}
+
+fn agent_observer_route(event: &Event) -> Result<Option<AgentObserverRoute>, String> {
+    if !content_looks_like_nip44(&event.content) {
+        return Err("invalid: observer content must be NIP-44 encrypted".into());
+    }
+
+    let recipient = parse_single_pubkey_tag(event, "p")?;
+    let agent = parse_single_pubkey_tag(event, OBSERVER_AGENT_TAG)?;
+    let frame = single_tag_content(event, OBSERVER_FRAME_TAG)?;
+
+    let (owner, direction, expected_frame) = if event.pubkey == agent && recipient != agent {
+        (
+            recipient,
+            AgentObserverDirection::Telemetry,
+            OBSERVER_FRAME_TELEMETRY,
+        )
+    } else if recipient == agent && event.pubkey != agent {
+        (
+            event.pubkey,
+            AgentObserverDirection::Control,
+            OBSERVER_FRAME_CONTROL,
+        )
+    } else {
+        return Err(
+            "invalid: observer frame must be agent-to-owner telemetry or owner-to-agent control"
+                .into(),
+        );
+    };
+
+    if frame != expected_frame {
+        // Unknown frame value — silently drop without notifying the publisher.
+        return Ok(None);
+    }
+
+    Ok(Some(AgentObserverRoute {
+        agent,
+        owner,
+        direction,
+    }))
+}
+
+fn parse_single_pubkey_tag(event: &Event, tag_name: &str) -> Result<PublicKey, String> {
+    let value = single_tag_content(event, tag_name)?;
+    PublicKey::from_hex(value)
+        .map_err(|_| format!("invalid: observer {tag_name} tag must be a hex pubkey"))
+}
+
+fn single_tag_content<'a>(event: &'a Event, tag_name: &str) -> Result<&'a str, String> {
+    let mut values = event
+        .tags
+        .iter()
+        .filter(|tag| tag.kind().to_string() == tag_name)
+        .filter_map(|tag| tag.content());
+    let Some(value) = values.next() else {
+        return Err(format!("invalid: observer frame missing {tag_name} tag"));
+    };
+    if values.next().is_some() {
+        return Err(format!(
+            "invalid: observer frame has multiple {tag_name} tags"
+        ));
+    }
+    Ok(value)
+}
+
 #[cfg(test)]
 mod tests {
+    use nostr::{EventBuilder, Keys, Kind, Tag};
     use sprout_core::kind::{
-        KIND_CANVAS, KIND_FORUM_COMMENT, KIND_FORUM_POST, KIND_FORUM_VOTE, KIND_PRESENCE_UPDATE,
-        KIND_STREAM_MESSAGE, KIND_STREAM_MESSAGE_DIFF,
+        KIND_AGENT_OBSERVER_FRAME, KIND_CANVAS, KIND_FORUM_COMMENT, KIND_FORUM_POST,
+        KIND_FORUM_VOTE, KIND_PRESENCE_UPDATE, KIND_STREAM_MESSAGE, KIND_STREAM_MESSAGE_DIFF,
+    };
+    use sprout_core::observer::{
+        encrypt_observer_payload, OBSERVER_AGENT_TAG, OBSERVER_FRAME_CONTROL, OBSERVER_FRAME_TAG,
+        OBSERVER_FRAME_TELEMETRY,
     };
 
     #[test]
@@ -470,4 +745,84 @@ mod tests {
             "presence updates are global/ephemeral"
         );
     }
+
+    #[test]
+    fn agent_observer_route_accepts_agent_to_owner_telemetry() {
+        let agent = Keys::generate();
+        let owner = Keys::generate();
+        let encrypted = encrypt_observer_payload(
+            &agent,
+            &owner.public_key(),
+            &serde_json::json!({"type": "acp_read"}),
+        )
+        .expect("encrypt observer payload");
+        let event = EventBuilder::new(
+            Kind::Custom(KIND_AGENT_OBSERVER_FRAME as u16),
+            encrypted,
+            [
+                Tag::parse(&["p", &owner.public_key().to_hex()]).expect("p tag"),
+                Tag::parse(&[OBSERVER_AGENT_TAG, &agent.public_key().to_hex()]).expect("agent tag"),
+                Tag::parse(&[OBSERVER_FRAME_TAG, OBSERVER_FRAME_TELEMETRY]).expect("frame tag"),
+            ],
+        )
+        .sign_with_keys(&agent)
+        .expect("sign event");
+
+        let route = super::agent_observer_route(&event)
+            .expect("observer route")
+            .expect("route should be Some");
+        assert_eq!(route.agent, agent.public_key());
+        assert_eq!(route.owner, owner.public_key());
+        assert_eq!(route.direction, super::AgentObserverDirection::Telemetry);
+    }
+
+    #[test]
+    fn agent_observer_route_accepts_owner_to_agent_control() {
+        let agent = Keys::generate();
+        let owner = Keys::generate();
+        let encrypted = encrypt_observer_payload(
+            &owner,
+            &agent.public_key(),
+            &serde_json::json!({"type": "cancel_turn"}),
+        )
+        .expect("encrypt observer payload");
+        let event = EventBuilder::new(
+            Kind::Custom(KIND_AGENT_OBSERVER_FRAME as u16),
+            encrypted,
+            [
+                Tag::parse(&["p", &agent.public_key().to_hex()]).expect("p tag"),
+                Tag::parse(&[OBSERVER_AGENT_TAG, &agent.public_key().to_hex()]).expect("agent tag"),
+                Tag::parse(&[OBSERVER_FRAME_TAG, OBSERVER_FRAME_CONTROL]).expect("frame tag"),
+            ],
+        )
+        .sign_with_keys(&owner)
+        .expect("sign event");
+
+        let route = super::agent_observer_route(&event)
+            .expect("observer route")
+            .expect("route should be Some");
+        assert_eq!(route.agent, agent.public_key());
+        assert_eq!(route.owner, owner.public_key());
+        assert_eq!(route.direction, super::AgentObserverDirection::Control);
+    }
+
+    #[test]
+    fn agent_observer_route_rejects_plaintext_content() {
+        let agent = Keys::generate();
+        let owner = Keys::generate();
+        let event = EventBuilder::new(
+            Kind::Custom(KIND_AGENT_OBSERVER_FRAME as u16),
+            "not encrypted",
+            [
+                Tag::parse(&["p", &owner.public_key().to_hex()]).expect("p tag"),
+                Tag::parse(&[OBSERVER_AGENT_TAG, &agent.public_key().to_hex()]).expect("agent tag"),
+                Tag::parse(&[OBSERVER_FRAME_TAG, OBSERVER_FRAME_TELEMETRY]).expect("frame tag"),
+            ],
+        )
+        .sign_with_keys(&agent)
+        .expect("sign event");
+
+        let err = super::agent_observer_route(&event).expect_err("route should reject plaintext");
+        assert!(err.contains("NIP-44"));
+    }
 }

crates/sprout-relay/src/handlers/req.rs

@@ -9,7 +9,8 @@ use hex;
 use nostr::Filter;
 use sprout_core::filter::filters_match;
 use sprout_core::kind::{
-    KIND_GIFT_WRAP, KIND_MEMBER_ADDED_NOTIFICATION, KIND_MEMBER_REMOVED_NOTIFICATION,
+    KIND_AGENT_OBSERVER_FRAME, KIND_GIFT_WRAP, KIND_MEMBER_ADDED_NOTIFICATION,
+    KIND_MEMBER_REMOVED_NOTIFICATION,
 };
 use sprout_db::EventQuery;
 
@@ -21,6 +22,12 @@ use crate::state::AppState;
 
 const MAX_HISTORICAL_LIMIT: i64 = 500;
 const MAX_SUBSCRIPTIONS: usize = 1024;
+const P_GATED_KINDS: [u32; 4] = [
+    KIND_AGENT_OBSERVER_FRAME,
+    KIND_MEMBER_ADDED_NOTIFICATION,
+    KIND_MEMBER_REMOVED_NOTIFICATION,
+    KIND_GIFT_WRAP,
+];
 
 /// Handle a REQ message: register the subscription, deliver historical events, then send EOSE.
 pub async fn handle_req(
@@ -113,33 +120,14 @@ pub async fn handle_req(
     // Only applies to GLOBAL subscriptions (channel_id = None). Channel-scoped
     // subscriptions can never receive globally-stored events — the fan_out()
     // invariant in subscription.rs prevents it.
-    const P_GATED_KINDS: [u32; 3] = [
-        KIND_MEMBER_ADDED_NOTIFICATION,
-        KIND_MEMBER_REMOVED_NOTIFICATION,
-        KIND_GIFT_WRAP,
-    ];
-
     if channel_id.is_none() {
         let authed_pubkey_hex = hex::encode(&pubkey_bytes);
-        let p_tag = nostr::SingleLetterTag::lowercase(nostr::Alphabet::P);
-
-        for filter in &filters {
-            let can_match_p_gated = filter.kinds.as_ref().is_none_or(|ks| {
-                ks.iter()
-                    .any(|k| P_GATED_KINDS.contains(&(k.as_u16() as u32)))
-            });
-            if can_match_p_gated {
-                let has_matching_p = filter.generic_tags.get(&p_tag).is_some_and(|values| {
-                    !values.is_empty() && values.iter().all(|v| *v == authed_pubkey_hex)
-                });
-                if !has_matching_p {
-                    conn.send(RelayMessage::closed(
-                        &sub_id,
-                        "restricted: p-gated events require #p matching your pubkey",
-                    ));
-                    return;
-                }
-            }
+        if !p_gated_filters_authorized(&filters, &authed_pubkey_hex) {
+            conn.send(RelayMessage::closed(
+                &sub_id,
+                "restricted: p-gated events require #p matching your pubkey",
+            ));
+            return;
         }
     }
 
@@ -581,6 +569,23 @@ fn extract_channel_id_from_filters(filters: &[Filter]) -> Option<uuid::Uuid> {
     found_id
 }
 
+fn p_gated_filters_authorized(filters: &[Filter], authed_pubkey_hex: &str) -> bool {
+    let p_tag = nostr::SingleLetterTag::lowercase(nostr::Alphabet::P);
+    filters.iter().all(|filter| {
+        let can_match_p_gated = filter.kinds.as_ref().is_none_or(|ks| {
+            ks.iter()
+                .any(|kind| P_GATED_KINDS.contains(&(kind.as_u16() as u32)))
+        });
+        if !can_match_p_gated {
+            return true;
+        }
+
+        filter.generic_tags.get(&p_tag).is_some_and(|values| {
+            !values.is_empty() && values.iter().all(|value| value == authed_pubkey_hex)
+        })
+    })
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -663,6 +668,32 @@ mod tests {
         assert!(!has_non_search, "all-search filters should not be mixed");
     }
 
+    #[test]
+    fn agent_observer_subscription_requires_matching_p_tag() {
+        let p_tag = SingleLetterTag::lowercase(Alphabet::P);
+        let authed = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa";
+        let other = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb";
+
+        let missing_p = Filter::new().kind(nostr::Kind::Custom(
+            sprout_core::kind::KIND_AGENT_OBSERVER_FRAME as u16,
+        ));
+        assert!(!p_gated_filters_authorized(&[missing_p], authed));
+
+        let wrong_p = Filter::new()
+            .kind(nostr::Kind::Custom(
+                sprout_core::kind::KIND_AGENT_OBSERVER_FRAME as u16,
+            ))
+            .custom_tag(p_tag, [other]);
+        assert!(!p_gated_filters_authorized(&[wrong_p], authed));
+
+        let matching_p = Filter::new()
+            .kind(nostr::Kind::Custom(
+                sprout_core::kind::KIND_AGENT_OBSERVER_FRAME as u16,
+            ))
+            .custom_tag(p_tag, [authed]);
+        assert!(p_gated_filters_authorized(&[matching_p], authed));
+    }
+
     #[test]
     fn d_tag_pushdown_only_for_nip33_kinds() {
         let d_tag = SingleLetterTag::lowercase(Alphabet::D);

crates/sprout-relay/src/state.rs

@@ -225,6 +225,16 @@ pub struct AppState {
     pub shutting_down: Arc<AtomicBool>,
     /// Process start time — used by `/_status` endpoint.
     pub started_at: Instant,
+    /// Per-agent sliding-window rate limiter for observer frames (kind 24200).
+    /// Key: agent pubkey bytes (32). Value: (count, window_start).
+    /// 100 events/sec per agent — prevents relay/DB pressure from bursty telemetry.
+    pub observer_rate_limiter: Arc<DashMap<[u8; 32], (u32, Instant)>>,
+    /// Cache for observer agent-owner authorization (kind 24200).
+    /// Key: (agent_pubkey_bytes, owner_pubkey_bytes). Value: is_owner.
+    /// agent_owner_pubkey is immutable so a long TTL (5 min) is safe.
+    /// Prevents repeated DB lookups from bursty observer traffic.
+    #[allow(clippy::type_complexity)]
+    pub observer_owner_cache: Arc<moka::sync::Cache<(Vec<u8>, Vec<u8>), bool>>,
 }
 
 impl AppState {
@@ -351,6 +361,13 @@ impl AppState {
             audio_rooms: Arc::new(AudioRoomManager::new()),
             shutting_down: Arc::new(AtomicBool::new(false)),
             started_at: Instant::now(),
+            observer_rate_limiter: Arc::new(DashMap::new()),
+            observer_owner_cache: Arc::new(
+                moka::sync::Cache::builder()
+                    .max_capacity(1_000)
+                    .time_to_live(std::time::Duration::from_secs(300))
+                    .build(),
+            ),
         };
         (
             state,

crates/sprout-relay/src/subscription.rs

@@ -1,9 +1,9 @@
-//! Subscription registry with (channel, kind) index for O(1) fan-out.
+//! Subscription registry with active WebSocket indexes for targeted fan-out.
 
-use std::collections::HashMap;
+use std::collections::{HashMap, HashSet};
 
 use dashmap::DashMap;
-use nostr::{Filter, Kind};
+use nostr::{Alphabet, Filter, Kind, SingleLetterTag};
 use uuid::Uuid;
 
 use sprout_core::{filter::filters_match, StoredEvent};
@@ -24,7 +24,13 @@ pub struct IndexKey {
     pub kind: Kind,
 }
 
-/// Thread-safe registry of active subscriptions with a (channel, kind) index for O(1) fan-out.
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+struct GlobalPKindIndexKey {
+    kind: Kind,
+    p: String,
+}
+
+/// Thread-safe registry of active subscriptions with targeted in-memory fan-out indexes.
 #[derive(Debug, Default)]
 pub struct SubscriptionRegistry {
     /// Maps conn_id → sub_id → (filters, channel_id).
@@ -35,6 +41,8 @@ pub struct SubscriptionRegistry {
     channel_wildcard_index: DashMap<Uuid, Vec<(ConnId, SubId)>>,
     /// Global subscriptions indexed by kind — avoids O(all_subs) scan for global events.
     global_kind_index: DashMap<Kind, Vec<(ConnId, SubId)>>,
+    /// Global subscriptions indexed by both kind and `#p` recipient.
+    global_p_kind_index: DashMap<GlobalPKindIndexKey, Vec<(ConnId, SubId)>>,
     /// Global subscriptions with no kind filter — wildcard, receives all global events.
     global_wildcard_index: DashMap<(), Vec<(ConnId, SubId)>>,
 }
@@ -91,22 +99,33 @@ impl SubscriptionRegistry {
                 }
             }
         } else {
-            // Global subscription — index by kind for sub-linear fan-out.
-            match extract_kinds_from_filters(&filters) {
-                None => {
-                    self.global_wildcard_index
-                        .entry(())
+            // Global subscription. Fully p-constrained filters can use the
+            // narrower (kind, #p) index; broader filters stay on the generic
+            // kind/wildcard indexes.
+            if let Some(keys) = extract_global_p_kind_index_keys(&filters) {
+                for key in keys {
+                    self.global_p_kind_index
+                        .entry(key)
                         .or_default()
                         .push((conn_id, sub_id.clone()));
                 }
-                Some(kinds) if kinds.is_empty() => {}
-                Some(kinds) => {
-                    for kind in kinds {
-                        self.global_kind_index
-                            .entry(kind)
+            } else {
+                match extract_kinds_from_filters(&filters) {
+                    None => {
+                        self.global_wildcard_index
+                            .entry(())
                             .or_default()
                             .push((conn_id, sub_id.clone()));
                     }
+                    Some(kinds) if kinds.is_empty() => {}
+                    Some(kinds) => {
+                        for kind in kinds {
+                            self.global_kind_index
+                                .entry(kind)
+                                .or_default()
+                                .push((conn_id, sub_id.clone()));
+                        }
+                    }
                 }
             }
         }
@@ -158,6 +177,7 @@ impl SubscriptionRegistry {
     /// Return all (conn_id, sub_id) pairs whose filters match the given event.
     pub fn fan_out(&self, event: &StoredEvent) -> Vec<(ConnId, SubId)> {
         let mut results = Vec::new();
+        let mut seen = HashSet::new();
 
         if let Some(channel_id) = event.channel_id {
             let key = IndexKey {
@@ -166,52 +186,39 @@ impl SubscriptionRegistry {
             };
             if let Some(candidates) = self.channel_kind_index.get(&key) {
                 for (conn_id, sub_id) in candidates.iter() {
-                    if let Some(conn_subs) = self.subs.get(conn_id) {
-                        if let Some((filters, _)) = conn_subs.get(sub_id.as_str()) {
-                            if filters_match(filters, event) {
-                                results.push((*conn_id, sub_id.clone()));
-                            }
-                        }
-                    }
+                    self.push_match(*conn_id, sub_id, event, &mut results, &mut seen);
                 }
             }
             // Also check wildcard (channel-only, kindless) index
             if let Some(wildcards) = self.channel_wildcard_index.get(&channel_id) {
                 for (conn_id, sub_id) in wildcards.iter() {
-                    if let Some(conn_subs) = self.subs.get(conn_id) {
-                        if let Some((filters, _)) = conn_subs.get(sub_id.as_str()) {
-                            if filters_match(filters, event) {
-                                results.push((*conn_id, sub_id.clone()));
-                            }
-                        }
-                    }
+                    self.push_match(*conn_id, sub_id, event, &mut results, &mut seen);
                 }
             }
         } else {
             // Global event (channel_id = None) — use global indexes for sub-linear fan-out.
             // Channel-scoped subscriptions are never in these indexes, preserving the
             // scoping invariant without an explicit skip check.
+            for p in event_p_tag_values(event) {
+                let key = GlobalPKindIndexKey {
+                    kind: event.event.kind,
+                    p,
+                };
+                if let Some(candidates) = self.global_p_kind_index.get(&key) {
+                    for (conn_id, sub_id) in candidates.iter() {
+                        self.push_match(*conn_id, sub_id, event, &mut results, &mut seen);
+                    }
+                }
+            }
             if let Some(candidates) = self.global_kind_index.get(&event.event.kind) {
                 for (conn_id, sub_id) in candidates.iter() {
-                    if let Some(conn_subs) = self.subs.get(conn_id) {
-                        if let Some((filters, _)) = conn_subs.get(sub_id.as_str()) {
-                            if filters_match(filters, event) {
-                                results.push((*conn_id, sub_id.clone()));
-                            }
-                        }
-                    }
+                    self.push_match(*conn_id, sub_id, event, &mut results, &mut seen);
                 }
             }
             // Also check global wildcard (kindless global subs).
             if let Some(wildcards) = self.global_wildcard_index.get(&()) {
                 for (conn_id, sub_id) in wildcards.iter() {
-                    if let Some(conn_subs) = self.subs.get(conn_id) {
-                        if let Some((filters, _)) = conn_subs.get(sub_id.as_str()) {
-                            if filters_match(filters, event) {
-                                results.push((*conn_id, sub_id.clone()));
-                            }
-                        }
-                    }
+                    self.push_match(*conn_id, sub_id, event, &mut results, &mut seen);
                 }
             }
         }
@@ -243,6 +250,26 @@ impl SubscriptionRegistry {
         self.subs.len()
     }
 
+    fn push_match(
+        &self,
+        conn_id: ConnId,
+        sub_id: &str,
+        event: &StoredEvent,
+        results: &mut Vec<(ConnId, SubId)>,
+        seen: &mut HashSet<(ConnId, SubId)>,
+    ) {
+        if let Some(conn_subs) = self.subs.get(&conn_id) {
+            if let Some((filters, _)) = conn_subs.get(sub_id) {
+                if filters_match(filters, event) {
+                    let entry = (conn_id, sub_id.to_string());
+                    if seen.insert(entry.clone()) {
+                        results.push(entry);
+                    }
+                }
+            }
+        }
+    }
+
     /// Removes a subscription from the channel_kind_index (or channel_wildcard_index) using
     /// targeted O(k) lookup where k = number of kinds in the filters, instead of O(n) full-scan.
     ///
@@ -290,25 +317,37 @@ impl SubscriptionRegistry {
                 }
             }
         } else {
-            // Global subscription — remove from global indexes.
-            match extract_kinds_from_filters(filters) {
-                None => {
-                    if let Some(mut entries) = self.global_wildcard_index.get_mut(&()) {
+            // Global subscription — remove from the same global index chosen at registration.
+            if let Some(keys) = extract_global_p_kind_index_keys(filters) {
+                for key in keys {
+                    if let Some(mut entries) = self.global_p_kind_index.get_mut(&key) {
                         entries.retain(|(cid, sid)| !(*cid == conn_id && sid == sub_id));
                         if entries.is_empty() {
                             drop(entries);
-                            self.global_wildcard_index.remove(&());
+                            self.global_p_kind_index.remove(&key);
                         }
                     }
                 }
-                Some(kinds) if kinds.is_empty() => {}
-                Some(kinds) => {
-                    for kind in kinds {
-                        if let Some(mut entries) = self.global_kind_index.get_mut(&kind) {
+            } else {
+                match extract_kinds_from_filters(filters) {
+                    None => {
+                        if let Some(mut entries) = self.global_wildcard_index.get_mut(&()) {
                             entries.retain(|(cid, sid)| !(*cid == conn_id && sid == sub_id));
                             if entries.is_empty() {
                                 drop(entries);
-                                self.global_kind_index.remove(&kind);
+                                self.global_wildcard_index.remove(&());
+                            }
+                        }
+                    }
+                    Some(kinds) if kinds.is_empty() => {}
+                    Some(kinds) => {
+                        for kind in kinds {
+                            if let Some(mut entries) = self.global_kind_index.get_mut(&kind) {
+                                entries.retain(|(cid, sid)| !(*cid == conn_id && sid == sub_id));
+                                if entries.is_empty() {
+                                    drop(entries);
+                                    self.global_kind_index.remove(&kind);
+                                }
                             }
                         }
                     }
@@ -318,6 +357,59 @@ impl SubscriptionRegistry {
     }
 }
 
+fn p_tag() -> SingleLetterTag {
+    SingleLetterTag::lowercase(Alphabet::P)
+}
+
+fn extract_global_p_kind_index_keys(filters: &[Filter]) -> Option<Vec<GlobalPKindIndexKey>> {
+    let mut seen = HashSet::new();
+    let mut keys = Vec::new();
+    let p_tag = p_tag();
+
+    for filter in filters {
+        let kinds = filter.kinds.as_ref()?;
+        if kinds.is_empty() {
+            continue;
+        }
+
+        let p_values = filter.generic_tags.get(&p_tag)?;
+        if p_values.is_empty() {
+            return None;
+        }
+
+        for kind in kinds {
+            for p in p_values {
+                let key = GlobalPKindIndexKey {
+                    kind: *kind,
+                    p: p.clone(),
+                };
+                if seen.insert(key.clone()) {
+                    keys.push(key);
+                }
+            }
+        }
+    }
+
+    Some(keys)
+}
+
+fn event_p_tag_values(event: &StoredEvent) -> Vec<String> {
+    let mut seen = HashSet::new();
+    let mut values = Vec::new();
+    for tag in event.event.tags.iter() {
+        if tag.kind().to_string() != "p" {
+            continue;
+        }
+        if let Some(value) = tag.content() {
+            let value = value.to_string();
+            if seen.insert(value.clone()) {
+                values.push(value);
+            }
+        }
+    }
+    values
+}
+
 /// Returns the union of all `kinds` across filters, or `None` if any filter
 /// lacks a `kinds` array (meaning that filter matches all kinds — wildcard).
 ///
@@ -351,7 +443,7 @@ fn extract_kinds_from_filters(filters: &[Filter]) -> Option<Vec<Kind>> {
 mod tests {
     use super::*;
     use chrono::Utc;
-    use nostr::{EventBuilder, Keys, Kind};
+    use nostr::{EventBuilder, Keys, Kind, Tag};
     use sprout_core::StoredEvent;
 
     fn make_stored_event(kind: Kind, channel_id: Option<Uuid>) -> StoredEvent {
@@ -362,6 +454,14 @@ mod tests {
         StoredEvent::with_received_at(event, Utc::now(), channel_id, true)
     }
 
+    fn make_stored_event_with_p(kind: Kind, p: &str, channel_id: Option<Uuid>) -> StoredEvent {
+        let keys = Keys::generate();
+        let event = EventBuilder::new(kind, "test", [Tag::parse(&["p", p]).expect("valid p tag")])
+            .sign_with_keys(&keys)
+            .expect("sign");
+        StoredEvent::with_received_at(event, Utc::now(), channel_id, true)
+    }
+
     #[test]
     fn test_subscription_registry_register_and_fan_out() {
         let registry = SubscriptionRegistry::new();
@@ -862,6 +962,60 @@ mod tests {
         assert!(registry.fan_out(&event_custom).is_empty());
     }
 
+    #[test]
+    fn test_global_p_kind_index_fan_out_targets_matching_p() {
+        let registry = SubscriptionRegistry::new();
+        let conn_a = Uuid::new_v4();
+        let conn_b = Uuid::new_v4();
+        let kind = Kind::Custom(sprout_core::kind::KIND_AGENT_OBSERVER_FRAME as u16);
+        let p_a = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa";
+        let p_b = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb";
+
+        registry.register(
+            conn_a,
+            "observer_a".to_string(),
+            vec![Filter::new().kind(kind).custom_tag(p_tag(), [p_a])],
+            None,
+        );
+        registry.register(
+            conn_b,
+            "observer_b".to_string(),
+            vec![Filter::new().kind(kind).custom_tag(p_tag(), [p_b])],
+            None,
+        );
+
+        assert!(
+            registry.global_kind_index.get(&kind).is_none(),
+            "fully p-constrained global subscriptions should use the p-kind index"
+        );
+
+        let event = make_stored_event_with_p(kind, p_a, None);
+        let matches = registry.fan_out(&event);
+        assert_eq!(matches, vec![(conn_a, "observer_a".to_string())]);
+    }
+
+    #[test]
+    fn test_global_p_kind_index_removal_cleanup() {
+        let registry = SubscriptionRegistry::new();
+        let conn_id = Uuid::new_v4();
+        let kind = Kind::Custom(sprout_core::kind::KIND_AGENT_OBSERVER_FRAME as u16);
+        let p = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa";
+        let filter = Filter::new().kind(kind).custom_tag(p_tag(), [p]);
+        let key = GlobalPKindIndexKey {
+            kind,
+            p: p.to_string(),
+        };
+
+        registry.register(conn_id, "observer".to_string(), vec![filter], None);
+        assert!(registry.global_p_kind_index.get(&key).is_some());
+
+        registry.remove_subscription(conn_id, "observer");
+        assert!(registry.global_p_kind_index.get(&key).is_none());
+
+        let event = make_stored_event_with_p(kind, p, None);
+        assert!(registry.fan_out(&event).is_empty());
+    }
+
     #[test]
     fn test_global_wildcard_index_fan_out() {
         // A global subscription with no kind filter should receive all global events.

crates/sprout-sdk/src/builders.rs

@@ -4,6 +4,13 @@
 //! The caller signs: `builder.sign_with_keys(&keys)?`.
 
 use nostr::{EventBuilder, Kind, Tag};
+use sprout_core::{
+    kind::KIND_AGENT_OBSERVER_FRAME,
+    observer::{
+        content_looks_like_nip44, OBSERVER_AGENT_TAG, OBSERVER_FRAME_CONTROL, OBSERVER_FRAME_TAG,
+        OBSERVER_FRAME_TELEMETRY,
+    },
+};
 use uuid::Uuid;
 
 use crate::{ChannelKind, DiffMeta, MemberRole, SdkError, ThreadRef, Visibility, VoteDirection};
@@ -35,6 +42,15 @@ fn check_hex_len(s: &str, min_len: usize, field: &str) -> Result<(), SdkError> {
     Ok(())
 }
 
+fn check_pubkey_hex(s: &str, field: &str) -> Result<String, SdkError> {
+    if s.len() != 64 || !s.chars().all(|c| c.is_ascii_hexdigit()) {
+        return Err(SdkError::InvalidInput(format!(
+            "{field} must be a 64-character hex pubkey"
+        )));
+    }
+    Ok(s.to_ascii_lowercase())
+}
+
 /// Emit NIP-10 e-tags for a `ThreadRef`.
 fn thread_tags(thread_ref: &ThreadRef, tags: &mut Vec<Tag>) -> Result<(), SdkError> {
     let root = thread_ref.root_event_id.to_hex();
@@ -105,6 +121,45 @@ pub fn build_message(
     Ok(EventBuilder::new(Kind::Custom(9), content, tags))
 }
 
+// ── Builder: build_agent_observer_frame ─────────────────────────────────────
+
+/// Build an encrypted agent observer frame (kind 24200).
+///
+/// `recipient_pubkey` is the cleartext `p` tag used by the relay for owner-only
+/// routing. `agent_pubkey` identifies the managed agent whose observer stream
+/// this frame belongs to. `encrypted_content` must be NIP-44 v2 ciphertext.
+pub fn build_agent_observer_frame(
+    recipient_pubkey: &str,
+    agent_pubkey: &str,
+    frame: &str,
+    encrypted_content: &str,
+) -> Result<EventBuilder, SdkError> {
+    if frame != OBSERVER_FRAME_TELEMETRY && frame != OBSERVER_FRAME_CONTROL {
+        return Err(SdkError::InvalidInput(format!(
+            "observer frame must be {OBSERVER_FRAME_TELEMETRY:?} or {OBSERVER_FRAME_CONTROL:?}"
+        )));
+    }
+    if !content_looks_like_nip44(encrypted_content) {
+        return Err(SdkError::InvalidInput(
+            "observer frame content must be NIP-44 v2 ciphertext".into(),
+        ));
+    }
+
+    let recipient_pubkey = check_pubkey_hex(recipient_pubkey, "recipient_pubkey")?;
+    let agent_pubkey = check_pubkey_hex(agent_pubkey, "agent_pubkey")?;
+    let tags = vec![
+        tag(&["p", &recipient_pubkey])?,
+        tag(&[OBSERVER_AGENT_TAG, &agent_pubkey])?,
+        tag(&[OBSERVER_FRAME_TAG, frame])?,
+    ];
+
+    Ok(EventBuilder::new(
+        Kind::Custom(KIND_AGENT_OBSERVER_FRAME as u16),
+        encrypted_content,
+        tags,
+    ))
+}
+
 // ── Builder 2: build_forum_post ───────────────────────────────────────────────
 
 /// Build a forum post thread root (kind 45001).
@@ -749,6 +804,50 @@ mod tests {
         assert!(has_tag(&ev, "h", &cid.to_string()));
     }
 
+    #[test]
+    fn agent_observer_frame_happy_path() {
+        let sender = keys();
+        let recipient = keys();
+        let agent = keys();
+        let encrypted = sprout_core::observer::encrypt_observer_payload(
+            &sender,
+            &recipient.public_key(),
+            &serde_json::json!({"type": "acp_read"}),
+        )
+        .unwrap();
+        let ev = sign(
+            build_agent_observer_frame(
+                &recipient.public_key().to_hex(),
+                &agent.public_key().to_hex(),
+                OBSERVER_FRAME_TELEMETRY,
+                &encrypted,
+            )
+            .unwrap(),
+        );
+
+        assert_eq!(ev.kind.as_u16(), KIND_AGENT_OBSERVER_FRAME as u16);
+        assert_eq!(ev.content, encrypted);
+        assert!(has_tag(&ev, "p", &recipient.public_key().to_hex()));
+        assert!(has_tag(
+            &ev,
+            OBSERVER_AGENT_TAG,
+            &agent.public_key().to_hex()
+        ));
+        assert!(has_tag(&ev, OBSERVER_FRAME_TAG, OBSERVER_FRAME_TELEMETRY));
+    }
+
+    #[test]
+    fn agent_observer_frame_rejects_plaintext_content() {
+        let err = build_agent_observer_frame(
+            &"a".repeat(64),
+            &"b".repeat(64),
+            OBSERVER_FRAME_TELEMETRY,
+            "not encrypted",
+        )
+        .unwrap_err();
+        assert!(matches!(err, SdkError::InvalidInput(_)));
+    }
+
     #[test]
     fn message_direct_reply() {
         let cid = uuid();

desktop/src-tauri/src/commands/agent_control.rs

@@ -1,90 +0,0 @@
-use serde::{Deserialize, Serialize};
-use tauri::State;
-
-use crate::{app_state::AppState, relay::relay_error_message};
-
-#[derive(Debug, Serialize)]
-#[serde(rename_all = "camelCase")]
-pub struct CancelManagedAgentTurnResponse {
-    pub status: String,
-}
-
-#[derive(Debug, Deserialize)]
-#[serde(rename_all = "camelCase")]
-struct ObserverCancelTurnResponse {
-    status: String,
-}
-
-#[tauri::command]
-pub async fn cancel_managed_agent_turn(
-    pubkey: String,
-    channel_id: String,
-    state: State<'_, AppState>,
-) -> Result<CancelManagedAgentTurnResponse, String> {
-    let observer_url = {
-        let runtimes = state
-            .managed_agent_processes
-            .lock()
-            .map_err(|error| error.to_string())?;
-        let runtime = runtimes
-            .get(&pubkey)
-            .ok_or_else(|| format!("agent {pubkey} is not running locally"))?;
-        runtime
-            .observer_url
-            .clone()
-            .ok_or_else(|| format!("agent {pubkey} does not expose an observer control endpoint"))?
-    };
-
-    let (control_url, token) = observer_control_url(&observer_url)?;
-    let request = state
-        .http_client
-        .post(control_url)
-        .bearer_auth(token)
-        .json(&serde_json::json!({ "channelId": channel_id }));
-
-    let response = request
-        .send()
-        .await
-        .map_err(|error| format!("observer cancel request failed: {error}"))?;
-
-    if !response.status().is_success() {
-        return Err(relay_error_message(response).await);
-    }
-
-    let body = response
-        .json::<ObserverCancelTurnResponse>()
-        .await
-        .map_err(|error| format!("observer cancel response parse failed: {error}"))?;
-
-    Ok(CancelManagedAgentTurnResponse {
-        status: body.status,
-    })
-}
-
-fn observer_control_url(observer_url: &str) -> Result<(String, String), String> {
-    let mut url = url::Url::parse(observer_url)
-        .map_err(|error| format!("invalid observer URL for agent: {error}"))?;
-    let token = url
-        .query_pairs()
-        .find_map(|(key, value)| (key == "token").then(|| value.into_owned()))
-        .filter(|value| !value.is_empty())
-        .ok_or_else(|| "observer URL is missing its control token".to_string())?;
-
-    url.set_path("/control/cancel");
-    url.set_query(None);
-
-    Ok((url.to_string(), token))
-}
-
-#[cfg(test)]
-mod tests {
-    use super::observer_control_url;
-
-    #[test]
-    fn derives_control_url_and_token_from_events_url() {
-        let (url, token) =
-            observer_control_url("http://127.0.0.1:1234/events?token=abc").expect("control url");
-        assert_eq!(url, "http://127.0.0.1:1234/control/cancel");
-        assert_eq!(token, "abc");
-    }
-}

desktop/src-tauri/src/commands/identity.rs

@@ -1,4 +1,8 @@
 use nostr::{nips::nip44, EventBuilder, JsonUtil, Kind, Tag, Timestamp, ToBech32};
+use nostr_compat::{
+    Event as CompatEvent, JsonUtil as CompatJsonUtil, Keys as CompatKeys,
+    PublicKey as CompatPublicKey,
+};
 use tauri::State;
 
 use crate::{
@@ -76,6 +80,64 @@ pub fn sign_event(
     Ok(event.as_json())
 }
 
+#[tauri::command]
+pub fn decrypt_observer_event(
+    event_json: String,
+    state: State<'_, AppState>,
+) -> Result<serde_json::Value, String> {
+    let nsec = {
+        let keys = state.keys.lock().map_err(|error| error.to_string())?;
+        keys.secret_key()
+            .to_bech32()
+            .map_err(|error| format!("encode nsec: {error}"))?
+    };
+    let keys = CompatKeys::parse(&nsec).map_err(|error| format!("parse nsec: {error}"))?;
+    let event =
+        CompatEvent::from_json(event_json).map_err(|error| format!("invalid event: {error}"))?;
+
+    // Defense-in-depth: verify event ID and signature before decrypting.
+    if !event.verify_id() {
+        return Err("observer event has invalid ID".into());
+    }
+    if !event.verify_signature() {
+        return Err("observer event has invalid signature".into());
+    }
+
+    sprout_core::observer::decrypt_observer_payload(&keys, &event)
+        .map_err(|error| format!("decrypt observer event failed: {error}"))
+}
+
+#[tauri::command]
+pub fn build_observer_control_event(
+    agent_pubkey: String,
+    payload: serde_json::Value,
+    state: State<'_, AppState>,
+) -> Result<String, String> {
+    let nsec = {
+        let keys = state.keys.lock().map_err(|error| error.to_string())?;
+        keys.secret_key()
+            .to_bech32()
+            .map_err(|error| format!("encode nsec: {error}"))?
+    };
+    let keys = CompatKeys::parse(&nsec).map_err(|error| format!("parse nsec: {error}"))?;
+    let agent_pubkey = CompatPublicKey::from_hex(agent_pubkey.trim())
+        .map_err(|error| format!("invalid agent pubkey: {error}"))?;
+    let agent_pubkey_hex = agent_pubkey.to_hex();
+    let encrypted = sprout_core::observer::encrypt_observer_payload(&keys, &agent_pubkey, &payload)
+        .map_err(|error| format!("encrypt observer control failed: {error}"))?;
+    let builder = sprout_sdk::build_agent_observer_frame(
+        &agent_pubkey_hex,
+        &agent_pubkey_hex,
+        sprout_core::observer::OBSERVER_FRAME_CONTROL,
+        &encrypted,
+    )
+    .map_err(|error| format!("build observer control failed: {error}"))?;
+    let event = builder
+        .sign_with_keys(&keys)
+        .map_err(|error| format!("sign observer control failed: {error}"))?;
+    Ok(event.as_json())
+}
+
 #[tauri::command]
 pub fn get_nsec(state: State<'_, AppState>) -> Result<String, String> {
     let keys = state.keys.lock().map_err(|error| error.to_string())?;

desktop/src-tauri/src/commands/mod.rs

@@ -1,4 +1,3 @@
-mod agent_control;
 mod agent_discovery;
 mod agent_models;
 mod agent_settings;
@@ -19,7 +18,6 @@ pub mod tokens;
 mod workflows;
 mod workspace;
 
-pub use agent_control::*;
 pub use agent_discovery::*;
 pub use agent_models::*;
 pub use agent_settings::*;

desktop/src-tauri/src/lib.rs

@@ -372,6 +372,8 @@ pub fn run() {
             discover_acp_providers,
             discover_managed_agent_prereqs,
             sign_event,
+            decrypt_observer_event,
+            build_observer_control_event,
             create_auth_event,
             nip44_encrypt_to_self,
             nip44_decrypt_from_self,
@@ -412,7 +414,6 @@ pub fn run() {
             revoke_token,
             revoke_all_tokens,
             list_relay_agents,
-            cancel_managed_agent_turn,
             list_managed_agents,
             create_managed_agent,
             start_managed_agent,

desktop/src-tauri/src/managed_agents/mod.rs

@@ -1,7 +1,6 @@
 mod backend;
 mod discovery;
 mod nest;
-mod observer;
 mod persona_avatars;
 mod persona_card;
 mod personas;
@@ -14,7 +13,6 @@ mod types;
 pub use backend::*;
 pub use discovery::*;
 pub use nest::*;
-pub use observer::*;
 pub use persona_card::*;
 pub use personas::*;
 pub use restore::*;

desktop/src-tauri/src/managed_agents/observer.rs

@@ -1,25 +0,0 @@
-/// Loopback observer endpoint assigned to one local ACP harness process.
-pub struct ObserverEndpoint {
-    /// Host:port bind address passed to `sprout-acp`.
-    pub addr: String,
-    /// Token required by observer requests.
-    pub token: String,
-    /// Event-stream URL including the observer token.
-    pub url: String,
-}
-
-/// Allocate an unused loopback port and random observer token.
-pub fn allocate_observer_endpoint() -> Result<ObserverEndpoint, String> {
-    let listener = std::net::TcpListener::bind("127.0.0.1:0")
-        .map_err(|error| format!("failed to allocate observer port: {error}"))?;
-    let port = listener
-        .local_addr()
-        .map_err(|error| format!("failed to inspect observer port: {error}"))?
-        .port();
-    drop(listener);
-
-    let token = uuid::Uuid::new_v4().to_string();
-    let addr = format!("127.0.0.1:{port}");
-    let url = format!("http://127.0.0.1:{port}/events?token={token}");
-    Ok(ObserverEndpoint { addr, token, url })
-}

desktop/src-tauri/src/managed_agents/restore.rs

@@ -86,7 +86,7 @@ pub fn restore_managed_agents_on_launch(
     // ── Phase B (no locks): resolve commands and spawn processes in parallel ──
     let spawn_results: Vec<(
         String,
-        Result<(std::process::Child, std::path::PathBuf, Option<String>), String>,
+        Result<(std::process::Child, std::path::PathBuf), String>,
     )> = std::thread::scope(|scope| {
         let handles: Vec<_> = agents_to_start
             .iter()
@@ -125,7 +125,7 @@ pub fn restore_managed_agents_on_launch(
             Err(_) => continue,
         };
         match result {
-            Ok((child, log_path, observer_url)) => {
+            Ok((child, log_path)) => {
                 let now = util::now_iso();
                 record.updated_at = now.clone();
                 record.runtime_pid = Some(child.id());
@@ -133,14 +133,7 @@ pub fn restore_managed_agents_on_launch(
                 record.last_stopped_at = None;
                 record.last_exit_code = None;
                 record.last_error = None;
-                runtimes.insert(
-                    pubkey,
-                    ManagedAgentProcess {
-                        child,
-                        log_path,
-                        observer_url,
-                    },
-                );
+                runtimes.insert(pubkey, ManagedAgentProcess { child, log_path });
             }
             Err(error) => {
                 record.updated_at = util::now_iso();

desktop/src-tauri/src/managed_agents/runtime.rs

@@ -4,9 +4,9 @@ use tauri::AppHandle;
 
 use crate::{
     managed_agents::{
-        allocate_observer_endpoint, append_log_marker, login_shell_path, managed_agent_log_path,
-        missing_command_message, normalize_agent_args, open_log_file, resolve_command,
-        ManagedAgentProcess, ManagedAgentRecord, ManagedAgentSummary,
+        append_log_marker, login_shell_path, managed_agent_log_path, missing_command_message,
+        normalize_agent_args, open_log_file, resolve_command, ManagedAgentProcess,
+        ManagedAgentRecord, ManagedAgentSummary,
     },
     util::now_iso,
 };
@@ -343,7 +343,7 @@ pub fn build_managed_agent_summary(
 ) -> Result<ManagedAgentSummary, String> {
     use crate::managed_agents::BackendKind;
 
-    let (status, pid, log_path, observer_url) = if record.backend != BackendKind::Local {
+    let (status, pid, log_path) = if record.backend != BackendKind::Local {
         // Two-axis status model for remote agents:
         //
         //   Control-plane (this field): "deployed" = provider has been invoked and
@@ -364,7 +364,7 @@ pub fn build_managed_agent_summary(
         } else {
             "not_deployed".to_string()
         };
-        (status, None, String::new(), None)
+        (status, None, String::new())
     } else {
         let persisted_pid = record.runtime_pid.filter(|pid| process_is_running(*pid));
         if let Some(runtime) = runtimes.get(&record.pubkey) {
@@ -372,7 +372,6 @@ pub fn build_managed_agent_summary(
                 "running".to_string(),
                 Some(runtime.child.id()),
                 runtime.log_path.display().to_string(),
-                runtime.observer_url.clone(),
             )
         } else if let Some(pid) = persisted_pid {
             (
@@ -381,7 +380,6 @@ pub fn build_managed_agent_summary(
                 managed_agent_log_path(app, &record.pubkey)?
                     .display()
                     .to_string(),
-                None,
             )
         } else {
             (
@@ -390,7 +388,6 @@ pub fn build_managed_agent_summary(
                 managed_agent_log_path(app, &record.pubkey)?
                     .display()
                     .to_string(),
-                None,
             )
         }
     };
@@ -424,7 +421,6 @@ pub fn build_managed_agent_summary(
         last_error: record.last_error.clone(),
         start_on_app_launch: record.start_on_app_launch,
         log_path,
-        observer_url,
     })
 }
 
@@ -444,7 +440,7 @@ pub fn find_managed_agent_mut<'a>(
 pub fn spawn_agent_child(
     app: &AppHandle,
     record: &ManagedAgentRecord,
-) -> Result<(std::process::Child, std::path::PathBuf, Option<String>), String> {
+) -> Result<(std::process::Child, std::path::PathBuf), String> {
     let log_path = managed_agent_log_path(app, &record.pubkey)?;
     append_log_marker(
         &log_path,
@@ -483,6 +479,7 @@ pub fn spawn_agent_child(
     if let Some(ref path) = augmented_path {
         command.env("PATH", path);
     }
+    command.env("RUST_LOG", child_rust_log_filter());
     command.env("SPROUT_PRIVATE_KEY", &record.private_key_nsec);
     command.env("SPROUT_RELAY_URL", &record.relay_url);
     command.env("SPROUT_ACP_AGENT_COMMAND", &resolved_agent_command);
@@ -571,9 +568,7 @@ pub fn spawn_agent_child(
         command.env_remove("SPROUT_API_TOKEN");
     }
 
-    let observer = allocate_observer_endpoint()?;
-    command.env("SPROUT_ACP_OBSERVER_ADDR", &observer.addr);
-    command.env("SPROUT_ACP_OBSERVER_TOKEN", &observer.token);
+    command.env("SPROUT_ACP_RELAY_OBSERVER", "true");
 
     // Spawn the harness in its own process group so we can kill the entire
     // tree (harness + MCP servers + agent subprocesses) on shutdown.
@@ -593,7 +588,15 @@ pub fn spawn_agent_child(
 
     let _ = super::write_agent_pid_file(app, &record.pubkey, child.id());
 
-    Ok((child, log_path, Some(observer.url)))
+    Ok((child, log_path))
+}
+
+fn child_rust_log_filter() -> String {
+    match std::env::var("RUST_LOG") {
+        Ok(existing) if existing.contains("sprout_acp") => existing,
+        Ok(existing) if !existing.trim().is_empty() => format!("{existing},sprout_acp=info"),
+        _ => "sprout_acp=info".to_string(),
+    }
 }
 
 pub fn start_managed_agent_process(
@@ -624,7 +627,7 @@ pub fn start_managed_agent_process(
         record.runtime_pid = None;
     }
 
-    let (child, log_path, observer_url) = spawn_agent_child(app, record)?;
+    let (child, log_path) = spawn_agent_child(app, record)?;
 
     let now = now_iso();
     record.updated_at = now.clone();
@@ -636,11 +639,7 @@ pub fn start_managed_agent_process(
 
     runtimes.insert(
         record.pubkey.clone(),
-        ManagedAgentProcess {
-            child,
-            log_path,
-            observer_url,
-        },
+        ManagedAgentProcess { child, log_path },
     );
     Ok(())
 }

desktop/src-tauri/src/managed_agents/types.rs

@@ -127,7 +127,6 @@ pub struct ManagedAgentRecord {
 pub struct ManagedAgentProcess {
     pub child: Child,
     pub log_path: PathBuf,
-    pub observer_url: Option<String>,
 }
 
 #[derive(Debug, Clone, Serialize)]
@@ -160,7 +159,6 @@ pub struct ManagedAgentSummary {
     pub last_error: Option<String>,
     pub start_on_app_launch: bool,
     pub log_path: String,
-    pub observer_url: Option<String>,
 }
 
 #[derive(Debug, Deserialize)]

desktop/src/features/agents/observerRelayStore.ts

@@ -0,0 +1,237 @@
+import * as React from "react";
+
+import { subscribeToAgentObserverFrames } from "@/shared/api/observerRelay";
+import type { RelayEvent, ManagedAgent } from "@/shared/api/types";
+import { getIdentity } from "@/shared/api/tauri";
+import { decryptObserverEvent } from "@/shared/api/tauriObserver";
+import { normalizePubkey } from "@/shared/lib/pubkey";
+import type { ConnectionState, ObserverEvent } from "./ui/agentSessionTypes";
+
+const MAX_OBSERVER_EVENTS = 800;
+
+type ObserverSnapshot = {
+  connectionState: ConnectionState;
+  errorMessage: string | null;
+  events: ObserverEvent[];
+};
+
+const listeners = new Set<() => void>();
+const eventsByAgent = new Map<string, ObserverEvent[]>();
+
+// Normalized pubkeys of agents we are actively managing. Only events whose
+// "agent" tag matches an entry here will be decrypted (defense-in-depth).
+const knownAgentPubkeys = new Set<string>();
+
+let connectionState: ConnectionState = "idle";
+let errorMessage: string | null = null;
+let unsubscribeRelay: (() => Promise<void>) | null = null;
+let startPromise: Promise<void> | null = null;
+let eventProcessingQueue: Promise<void> = Promise.resolve();
+let generation = 0;
+
+function notifyListeners() {
+  for (const listener of listeners) {
+    listener();
+  }
+}
+
+function setConnectionState(
+  nextState: ConnectionState,
+  nextErrorMessage: string | null = errorMessage,
+) {
+  connectionState = nextState;
+  errorMessage = nextErrorMessage;
+  notifyListeners();
+}
+
+function observerTag(event: RelayEvent, tagName: string) {
+  return event.tags.find((tag) => tag[0] === tagName)?.[1] ?? null;
+}
+
+function appendAgentEvent(agentPubkey: string, event: ObserverEvent) {
+  const key = normalizePubkey(agentPubkey);
+  const current = eventsByAgent.get(key) ?? [];
+  if (
+    current.some(
+      (existing) =>
+        existing.seq === event.seq && existing.timestamp === event.timestamp,
+    )
+  ) {
+    return;
+  }
+
+  const next = [...current, event].sort(compareObserverEvents);
+  eventsByAgent.set(
+    key,
+    next.length > MAX_OBSERVER_EVENTS
+      ? next.slice(next.length - MAX_OBSERVER_EVENTS)
+      : next,
+  );
+  notifyListeners();
+}
+
+function compareObserverEvents(left: ObserverEvent, right: ObserverEvent) {
+  const leftTime = Date.parse(left.timestamp);
+  const rightTime = Date.parse(right.timestamp);
+  if (Number.isFinite(leftTime) && Number.isFinite(rightTime)) {
+    const timeDiff = leftTime - rightTime;
+    if (timeDiff !== 0) {
+      return timeDiff;
+    }
+  }
+
+  return left.seq - right.seq;
+}
+
+async function handleRelayObserverEvent(
+  event: RelayEvent,
+  activeGeneration: number,
+) {
+  const agentPubkey = observerTag(event, "agent");
+  const frame = observerTag(event, "frame");
+  if (!agentPubkey || frame !== "telemetry") {
+    return;
+  }
+
+  // Verify agent is known/trusted before decrypting.
+  // Silently drop events from agents we are not managing.
+  if (!knownAgentPubkeys.has(normalizePubkey(agentPubkey))) {
+    return;
+  }
+
+  // Defense-in-depth: verify the event sender matches the claimed agent pubkey.
+  // The relay gates on is_agent_owner, but a compromised relay could misroute.
+  if (normalizePubkey(event.pubkey) !== normalizePubkey(agentPubkey)) {
+    return;
+  }
+
+  try {
+    const parsed = (await decryptObserverEvent(event)) as ObserverEvent;
+    if (activeGeneration !== generation) {
+      return;
+    }
+    appendAgentEvent(agentPubkey, parsed);
+  } catch (error) {
+    if (activeGeneration !== generation) {
+      return;
+    }
+    setConnectionState(
+      "error",
+      error instanceof Error
+        ? `Observer event decrypt failed: ${error.message}`
+        : "Observer event decrypt failed.",
+    );
+  }
+}
+
+export function ensureRelayObserverSubscription() {
+  if (unsubscribeRelay) {
+    return Promise.resolve();
+  }
+  if (startPromise) {
+    return startPromise;
+  }
+
+  const activeGeneration = generation;
+  setConnectionState("connecting", null);
+  startPromise = (async () => {
+    const identity = await getIdentity();
+    const unsubscribe = await subscribeToAgentObserverFrames(
+      identity.pubkey,
+      (event) => {
+        eventProcessingQueue = eventProcessingQueue
+          .then(() => handleRelayObserverEvent(event, activeGeneration))
+          .catch((error) => {
+            if (activeGeneration !== generation) {
+              return;
+            }
+            setConnectionState(
+              "error",
+              error instanceof Error
+                ? `Observer event handling failed: ${error.message}`
+                : "Observer event handling failed.",
+            );
+          });
+      },
+    );
+    if (activeGeneration !== generation) {
+      await unsubscribe();
+      return;
+    }
+    unsubscribeRelay = unsubscribe;
+    setConnectionState("open", null);
+  })()
+    .catch((error) => {
+      if (activeGeneration === generation) {
+        setConnectionState(
+          "error",
+          error instanceof Error
+            ? error.message
+            : "Observer relay subscription failed.",
+        );
+      }
+    })
+    .finally(() => {
+      if (activeGeneration === generation) {
+        startPromise = null;
+      }
+    });
+
+  return startPromise;
+}
+
+export function subscribeAgentObserverStore(listener: () => void) {
+  listeners.add(listener);
+  return () => {
+    listeners.delete(listener);
+  };
+}
+
+export function getAgentObserverSnapshot(
+  agentPubkey: string,
+): ObserverSnapshot {
+  return {
+    connectionState,
+    errorMessage,
+    events: eventsByAgent.get(normalizePubkey(agentPubkey)) ?? [],
+  };
+}
+
+export function useManagedAgentObserverBridge(agents: readonly ManagedAgent[]) {
+  const hasActiveAgent = React.useMemo(
+    () =>
+      agents.some(
+        (agent) => agent.status === "running" || agent.status === "deployed",
+      ),
+    [agents],
+  );
+
+  // Keep the trusted-pubkey set in sync with the current managed agent list.
+  React.useEffect(() => {
+    knownAgentPubkeys.clear();
+    for (const agent of agents) {
+      knownAgentPubkeys.add(normalizePubkey(agent.pubkey));
+    }
+  }, [agents]);
+
+  React.useEffect(() => {
+    if (!hasActiveAgent) {
+      return;
+    }
+    void ensureRelayObserverSubscription();
+  }, [hasActiveAgent]);
+}
+
+export function resetAgentObserverStore() {
+  generation += 1;
+  const unsubscribe = unsubscribeRelay;
+  unsubscribeRelay = null;
+  startPromise = null;
+  eventProcessingQueue = Promise.resolve();
+  eventsByAgent.clear();
+  knownAgentPubkeys.clear();
+  connectionState = "idle";
+  errorMessage = null;
+  notifyListeners();
+  void unsubscribe?.();
+}

desktop/src/features/agents/ui/ManagedAgentSessionPanel.tsx

@@ -41,8 +41,8 @@ export function ManagedAgentSessionPanel({
   showRaw = true,
 }: ManagedAgentSessionPanelProps) {
   const { connectionState, errorMessage, events } = useObserverEvents(
-    agent.observerUrl,
     agent.status === "running",
+    agent.pubkey,
   );
   const scopedEvents = React.useMemo(
     () =>
@@ -72,7 +72,7 @@ export function ManagedAgentSessionPanel({
         <SessionHeader
           connectionState={connectionState}
           eventCount={scopedEvents.length}
-          hasObserver={Boolean(agent.observerUrl)}
+          hasObserver={agent.status === "running"}
           latestSessionId={latestSessionId}
         />
       ) : null}
@@ -83,7 +83,7 @@ export function ManagedAgentSessionPanel({
         emptyDescription={emptyDescription}
         errorMessage={errorMessage}
         events={scopedEvents}
-        hasObserver={Boolean(agent.observerUrl)}
+        hasObserver={agent.status === "running"}
         showRaw={showRaw}
         transcript={transcript}
       />

desktop/src/features/agents/ui/useManagedAgentActions.ts

@@ -13,6 +13,7 @@ import {
 } from "@/features/agents/hooks";
 import { useChannelsQuery } from "@/features/channels/hooks";
 import { usePresenceQuery } from "@/features/presence/hooks";
+import { useManagedAgentObserverBridge } from "@/features/agents/observerRelayStore";
 import type {
   Channel,
   CreateManagedAgentResponse,
@@ -69,6 +70,7 @@ export function useManagedAgentActions() {
       }),
     [managedAgentsQuery.data],
   );
+  useManagedAgentObserverBridge(managedAgents);
 
   const managedPubkeys = React.useMemo(
     () => new Set(managedAgents.map((agent) => agent.pubkey)),

desktop/src/features/agents/ui/useObserverEvents.ts

@@ -1,12 +1,15 @@
 import * as React from "react";
 
+import {
+  ensureRelayObserverSubscription,
+  getAgentObserverSnapshot,
+  subscribeAgentObserverStore,
+} from "@/features/agents/observerRelayStore";
 import type { ConnectionState, ObserverEvent } from "./agentSessionTypes";
 
-const MAX_OBSERVER_EVENTS = 800;
-
 export function useObserverEvents(
-  observerUrl: string | null,
   enabled: boolean,
+  agentPubkey?: string | null,
 ) {
   const [events, setEvents] = React.useState<ObserverEvent[]>([]);
   const [connectionState, setConnectionState] =
@@ -17,52 +20,31 @@ export function useObserverEvents(
     setEvents([]);
     setErrorMessage(null);
 
-    if (!observerUrl || !enabled) {
+    if (!enabled) {
       setConnectionState("idle");
       return;
     }
 
-    setConnectionState("connecting");
-    const source = new EventSource(observerUrl);
+    if (!agentPubkey) {
+      setConnectionState("idle");
+      return;
+    }
 
-    source.onopen = () => {
-      setConnectionState("open");
-      setErrorMessage(null);
+    const syncSnapshot = () => {
+      const snapshot = getAgentObserverSnapshot(agentPubkey);
+      setConnectionState(snapshot.connectionState);
+      setErrorMessage(snapshot.errorMessage);
+      setEvents(snapshot.events);
     };
 
-    source.onmessage = (event) => {
-      try {
-        const parsed = JSON.parse(event.data) as ObserverEvent;
-        setEvents((current) => {
-          if (current.some((existing) => existing.seq === parsed.seq)) {
-            return current;
-          }
-          const next = [...current, parsed];
-          return next.length > MAX_OBSERVER_EVENTS
-            ? next.slice(next.length - MAX_OBSERVER_EVENTS)
-            : next;
-        });
-      } catch (error) {
-        setErrorMessage(
-          error instanceof Error
-            ? `Observer event parse failed: ${error.message}`
-            : "Observer event parse failed.",
-        );
-      }
-    };
-
-    source.onerror = () => {
-      setConnectionState((current) =>
-        current === "open" ? "closed" : "error",
-      );
-      setErrorMessage("Observer stream is not available right now.");
-    };
+    syncSnapshot();
+    const unsubscribe = subscribeAgentObserverStore(syncSnapshot);
+    void ensureRelayObserverSubscription();
 
     return () => {
-      source.close();
-      setConnectionState("closed");
+      unsubscribe();
     };
-  }, [enabled, observerUrl]);
+  }, [agentPubkey, enabled]);
 
   return { connectionState, errorMessage, events };
 }

desktop/src/features/channels/ui/AgentSessionThreadPanel.tsx

@@ -39,7 +39,7 @@ export function AgentSessionThreadPanel({
   onResizeStart,
   widthPx,
 }: AgentSessionThreadPanelProps) {
-  const isLive = agent.status === "running" && Boolean(agent.observerUrl);
+  const isLive = agent.status === "running";
   const isOverlay = useIsThreadPanelOverlay();
   useEscapeKey(onClose, isOverlay);
 
@@ -47,12 +47,10 @@ export function AgentSessionThreadPanel({
 
   async function handleInterruptTurn() {
     try {
-      const result = await cancelManagedAgentTurn(agent.pubkey, channel.id);
-      if (result.status === "sent") {
-        toast.success(`Stop signal sent to ${agent.name}.`);
-      } else {
-        toast.info(`${agent.name} has no active turn in this channel.`);
-      }
+      await cancelManagedAgentTurn(agent.pubkey, channel.id);
+      toast.success(
+        `Stop signal sent to ${agent.name}. It may take a moment to respond.`,
+      );
     } catch (error) {
       toast.error(
         error instanceof Error

desktop/src/features/channels/ui/ChannelScreen.tsx

@@ -17,6 +17,7 @@ import {
   useManagedAgentsQuery,
   usePersonasQuery,
 } from "@/features/agents/hooks";
+import { useManagedAgentObserverBridge } from "@/features/agents/observerRelayStore";
 import {
   mergeMessages,
   useChannelMessagesQuery,
@@ -46,7 +47,6 @@ import { useChannelFind } from "@/features/search/useChannelFind";
 import { ViewLoadingFallback } from "@/shared/ui/ViewLoadingFallback";
 import { AgentSessionProvider } from "@/shared/context/AgentSessionContext";
 import { useChannelAgentSessions } from "./useChannelAgentSessions";
-
 type ChannelScreenProps = {
   activeChannel: Channel | null;
   currentIdentity?: Identity;
@@ -87,7 +87,6 @@ export function ChannelScreen({
   const [editTargetId, setEditTargetId] = React.useState<string | null>(null);
   const currentPubkey = currentIdentity?.pubkey;
   const activeChannelId = activeChannel?.id ?? null;
-
   const messagesQuery = useChannelMessagesQuery(activeChannel);
   useChannelSubscription(activeChannel);
   const { fetchOlder, hasOlderMessages, isFetchingOlder } =
@@ -165,6 +164,7 @@ export function ChannelScreen({
     enabled: messageProfilePubkeys.length > 0,
   });
   const managedAgentsQuery = useManagedAgentsQuery();
+  useManagedAgentObserverBridge(managedAgentsQuery.data ?? []);
   const { humanTypingPubkeys, botTypingPubkeys } = React.useMemo(() => {
     const localAgentSet = new Set(
       (managedAgentsQuery.data ?? [])

desktop/src/features/workspaces/useWorkspaceInit.ts

@@ -5,6 +5,7 @@ import { applyWorkspace, getDefaultRelayUrl } from "@/shared/api/tauri";
 import { resetMediaCaches } from "@/shared/lib/mediaUrl";
 import { clearSearchHitEventCache } from "@/app/navigation/searchHitEventCache";
 import { clearAllDrafts } from "@/features/messages/lib/useDrafts";
+import { resetAgentObserverStore } from "@/features/agents/observerRelayStore";
 
 import type { Workspace } from "./types";
 
@@ -16,6 +17,7 @@ import type { Workspace } from "./types";
  */
 function resetWorkspaceState(): void {
   relayClient.disconnect();
+  resetAgentObserverStore();
   resetMediaCaches();
   clearSearchHitEventCache();
   clearAllDrafts();

desktop/src/shared/api/agentControl.ts

@@ -1,12 +1,13 @@
-import { invokeTauri } from "@/shared/api/tauri";
+import { sendAgentObserverControl } from "@/shared/api/observerRelay";
 import type { CancelManagedAgentTurnResult } from "@/shared/api/types";
 
 export async function cancelManagedAgentTurn(
   pubkey: string,
   channelId: string,
 ): Promise<CancelManagedAgentTurnResult> {
-  return invokeTauri<CancelManagedAgentTurnResult>(
-    "cancel_managed_agent_turn",
-    { pubkey, channelId },
-  );
+  await sendAgentObserverControl(pubkey, {
+    type: "cancel_turn",
+    channelId,
+  });
+  return { status: "sent" };
 }

desktop/src/shared/api/observerRelay.ts

@@ -0,0 +1,32 @@
+import { buildObserverControlEvent } from "@/shared/api/tauriObserver";
+import type { RelayEvent } from "@/shared/api/types";
+import { KIND_AGENT_OBSERVER_FRAME } from "@/shared/constants/kinds";
+import { relayClient } from "./relayClient";
+
+export function subscribeToAgentObserverFrames(
+  ownerPubkey: string,
+  onEvent: (event: RelayEvent) => void,
+) {
+  return relayClient.subscribeLive(
+    {
+      kinds: [KIND_AGENT_OBSERVER_FRAME],
+      "#p": [ownerPubkey],
+      limit: 0,
+      since: Math.floor(Date.now() / 1_000),
+    },
+    onEvent,
+  );
+}
+
+export async function sendAgentObserverControl(
+  agentPubkey: string,
+  payload: unknown,
+) {
+  await relayClient.preconnect();
+  const event = await buildObserverControlEvent({ agentPubkey, payload });
+  await relayClient.publishEvent(
+    event,
+    "Timed out while sending the agent control command.",
+    "Failed to send the agent control command.",
+  );
+}

desktop/src/shared/api/tauri.ts

@@ -252,7 +252,6 @@ export type RawManagedAgent = {
   last_exit_code: number | null;
   last_error: string | null;
   log_path: string;
-  observer_url: string | null;
   start_on_app_launch: boolean;
   backend: ManagedAgentBackend;
   backend_agent_id: string | null;
@@ -863,7 +862,6 @@ export function fromRawManagedAgent(agent: RawManagedAgent): ManagedAgent {
     lastExitCode: agent.last_exit_code,
     lastError: agent.last_error,
     logPath: agent.log_path,
-    observerUrl: agent.observer_url,
     startOnAppLaunch: agent.start_on_app_launch,
     backend: agent.backend,
     backendAgentId: agent.backend_agent_id,

desktop/src/shared/api/tauriObserver.ts

@@ -0,0 +1,21 @@
+import type { RelayEvent } from "@/shared/api/types";
+import { invokeTauri } from "./tauri";
+
+export async function decryptObserverEvent(
+  event: RelayEvent,
+): Promise<unknown> {
+  return invokeTauri<unknown>("decrypt_observer_event", {
+    eventJson: JSON.stringify(event),
+  });
+}
+
+export async function buildObserverControlEvent(input: {
+  agentPubkey: string;
+  payload: unknown;
+}): Promise<RelayEvent> {
+  const eventJson = await invokeTauri<string>("build_observer_control_event", {
+    agentPubkey: input.agentPubkey,
+    payload: input.payload,
+  });
+  return JSON.parse(eventJson) as RelayEvent;
+}

desktop/src/shared/api/types.ts

@@ -307,7 +307,6 @@ export type ManagedAgent = {
   lastExitCode: number | null;
   lastError: string | null;
   logPath: string;
-  observerUrl: string | null;
   startOnAppLaunch: boolean;
   backend: ManagedAgentBackend;
   backendAgentId: string | null;

desktop/src/shared/constants/kinds.ts

@@ -17,6 +17,7 @@ export const KIND_FORUM_COMMENT = 45003;
 export const KIND_APPROVAL_REQUEST = 46010;
 export const KIND_TYPING_INDICATOR = 20002;
 export const KIND_READ_STATE = 30078;
+export const KIND_AGENT_OBSERVER_FRAME = 24200;
 
 const CHANNEL_MESSAGE_EVENT_KINDS = [
   KIND_STREAM_MESSAGE,

desktop/src/testing/e2eBridge.ts

@@ -331,7 +331,6 @@ type RawManagedAgent = {
   last_exit_code: number | null;
   last_error: string | null;
   log_path: string;
-  observer_url: string | null;
   start_on_app_launch: boolean;
   backend:
     | { type: "local" }
@@ -682,7 +681,6 @@ function cloneManagedAgent(agent: MockManagedAgent): RawManagedAgent {
     last_exit_code: agent.last_exit_code,
     last_error: agent.last_error,
     log_path: agent.log_path,
-    observer_url: agent.observer_url ?? null,
     start_on_app_launch: agent.start_on_app_launch,
     backend: agent.backend ?? { type: "local" as const },
     backend_agent_id: agent.backend_agent_id ?? null,
@@ -3646,9 +3644,6 @@ async function handleCreateManagedAgent(args: {
     last_exit_code: null,
     last_error: null,
     log_path: `/tmp/mock-agent-${pubkey}.log`,
-    observer_url: args.input.spawnAfterCreate
-      ? `http://127.0.0.1:42000/events?token=mock-${pubkey.slice(0, 8)}`
-      : null,
     start_on_app_launch: args.input.startOnAppLaunch ?? true,
     backend: args.input.backend ?? { type: "local" as const },
     backend_agent_id: null,
@@ -3695,9 +3690,6 @@ async function handleStartManagedAgent(args: {
   const now = new Date().toISOString();
   agent.status = "running";
   agent.pid = agent.pid ?? 42000 + mockManagedAgents.indexOf(agent);
-  agent.observer_url =
-    agent.observer_url ??
-    `http://127.0.0.1:${agent.pid}/events?token=mock-${agent.pubkey.slice(0, 8)}`;
   agent.updated_at = now;
   agent.last_started_at = now;
   agent.last_error = null;
@@ -3713,7 +3705,6 @@ async function handleStopManagedAgent(args: {
   const now = new Date().toISOString();
   agent.status = "stopped";
   agent.pid = null;
-  agent.observer_url = null;
   agent.updated_at = now;
   agent.last_stopped_at = now;
   agent.log_lines.push(`stopped mock harness at ${now}`);
@@ -4498,8 +4489,6 @@ export function maybeInstallE2eTauriMocks() {
         return handleStopManagedAgent(
           payload as Parameters<typeof handleStopManagedAgent>[0],
         );
-      case "cancel_managed_agent_turn":
-        return { status: "sent" };
       case "set_managed_agent_start_on_app_launch":
         return handleSetManagedAgentStartOnAppLaunch(
           payload as Parameters<

docs/nips/NIP-AO.md

@@ -0,0 +1,300 @@
+NIP-AO
+======
+
+Agent Observability
+-------------------
+
+`draft` `optional`
+
+This NIP defines ephemeral, encrypted event kinds for streaming internal session telemetry between AI agent processes and their owners' desktop clients via Nostr relays.
+
+## Motivation
+
+AI agent harnesses execute long-running sessions that invoke tools, send protocol
+frames to models, and emit intermediate reasoning. Owners need real-time visibility
+into this activity for debugging, auditing, and control — without that telemetry
+being stored on any relay or visible to third parties.
+
+Kind 24200 provides a dedicated, encrypted, ephemeral channel for this purpose.
+It is strictly scoped to the agent↔owner relationship and carries no durable state.
+
+## Definitions
+
+- **Agent**: An AI process with its own Nostr keypair, executing a session on behalf of an owner.
+- **Owner**: The human (or system) whose pubkey the agent was provisioned under.
+- **Observer Frame**: A single kind 24200 event carrying one unit of telemetry or control.
+- **Session**: A bounded agent execution correlated by a shared `sessionId`.
+
+## Event Kinds
+
+| Kind  | Name                  | Direction         |
+|-------|-----------------------|-------------------|
+| 24200 | Agent Observer Frame  | agent↔owner (both)|
+
+Kind 24200 falls in the ephemeral range (20000–29999) defined by NIP-01. Relays
+MUST NOT persist it.
+
+## Event Structure
+
+```json
+{
+  "kind": 24200,
+  "pubkey": "<sender_pubkey>",
+  "created_at": <unix_timestamp>,
+  "content": "<NIP-44 v2 ciphertext>",
+  "tags": [
+    ["p",     "<recipient_pubkey>"],
+    ["agent", "<agent_pubkey>"],
+    ["frame", "telemetry" | "control"]
+  ]
+}
+```
+
+Events MUST have exactly one `p` tag, exactly one `agent` tag, and exactly one
+`frame` tag.
+
+**Telemetry** (agent → owner): `pubkey`=agent, `p`=owner, `agent`=agent.
+**Control** (owner → agent): `pubkey`=owner, `p`=agent, `agent`=agent (target).
+
+`frame` MUST be `"telemetry"` or `"control"`. Relays SHOULD silently drop events
+with unrecognized `frame` values (returning OK to the publisher for forward
+compatibility). Clients MUST ignore events with unrecognized `frame` values. An `h` tag MAY be included when the session runs within a NIP-29 group
+context.
+
+## Encryption
+
+All `content` fields MUST be encrypted with NIP-44 v2 (XChaCha20-Poly1305 over a
+secp256k1 ECDH shared secret).
+
+- **Telemetry**: encrypted with `(agent_privkey, owner_pubkey)`
+- **Control**: encrypted with `(owner_privkey, agent_pubkey)`
+
+Plaintext SHOULD be zeroized from memory immediately after encrypt/decrypt.
+Decrypted payload MUST NOT exceed 65,535 bytes.
+
+## Decrypted Payload
+
+### Telemetry (`frame=telemetry`)
+
+The `content` field decrypts to an `ObserverEvent` JSON object:
+
+```json
+{
+  "seq":         <monotonic_integer>,
+  "timestamp":   "<rfc3339_string>",
+  "kind":        "<frame_kind>",
+  "agentIndex":  <integer> | null,
+  "channelId":   "<channel_uuid>" | null,
+  "sessionId":   "<session_id>" | null,
+  "turnId":      "<turn_id>" | null,
+  "payload":     { ... }
+}
+```
+
+`seq`, `timestamp`, `kind`, and `payload` are REQUIRED. `agentIndex`, `channelId`, `sessionId`,
+and `turnId` are OPTIONAL — they MAY be `null` when the value is not yet known
+(e.g., `sessionId` before session establishment). Clients MUST handle `null` values
+gracefully.
+
+`seq` is monotonically increasing per session (drop detection). `timestamp` is an
+RFC 3339 datetime string with sub-second precision (e.g., `"2026-04-29T12:00:41.500Z"`).
+`agentIndex` identifies the agent in multi-agent scenarios. `sessionId`/`turnId`
+correlate frames across a session and turn. `payload` is kind-specific (MAY be `{}`).
+Unknown `kind` values MUST be ignored.
+
+### Frame Kinds
+
+| `kind`             | Description                                              |
+|--------------------|----------------------------------------------------------|
+| `acp_read`         | Inbound ACP protocol frame (model → harness)             |
+| `acp_write`        | Outbound ACP protocol frame (harness → model)            |
+| `turn_started`     | A new agent turn has begun                               |
+| `session_resolved` | Session completed or terminated                          |
+
+### Control (`frame=control`)
+
+The `content` field decrypts to:
+
+```json
+{
+  "type":      "cancel_turn",
+  "channelId": "<channel_uuid>"
+}
+```
+
+The only defined control type is `cancel_turn`. Implementations MUST ignore
+events with unrecognized `type` values.
+
+## Ephemerality Contract
+
+- Relays MUST NOT persist kind 24200 events to any durable storage.
+- Relays MUST NOT include kind 24200 events in search indexes.
+- Relays MUST NOT include kind 24200 events in audit logs.
+- Relays SHOULD fan out kind 24200 events only via in-memory pub/sub,
+  never via a database write path.
+- Clients SHOULD subscribe with `since=<now>`; historical replay is not supported.
+- Clients SHOULD buffer received events in a bounded in-memory ring buffer.
+
+## Authorization
+
+**Telemetry** (agent → owner):
+- `event.pubkey` MUST equal the agent pubkey.
+- `p` tag MUST equal the owner pubkey.
+- Relay MUST verify `is_agent_owner(agent, owner)` via authenticated ownership lookup.
+
+**Control** (owner → agent):
+- `event.pubkey` MUST equal the owner pubkey.
+- `p` tag MUST equal the agent pubkey.
+- Relay MUST verify `is_agent_owner(agent, owner)` where agent is resolved from the
+  `agent` tag.
+
+Both directions require relay confirmation of the agent-owner relationship via
+database lookup. `#p` tag matching alone is insufficient. Unauthorized publish or
+subscribe attempts MUST be rejected with `AUTH required`.
+
+## Relay Behavior
+
+On receiving a kind 24200 event, a relay MUST:
+
+1. Validate the event signature per NIP-01.
+2. Verify authorization per the rules above.
+3. Fan out to matching subscribers via in-memory pub/sub.
+4. NOT invoke the normal event ingestion or persistence path.
+
+Relays SHOULD enforce a rate limit of 100 events/second per agent pubkey.
+Relays are RECOMMENDED to reject events whose `created_at` falls outside a ±5-minute
+freshness window to prevent replay of captured events.
+
+## Client Behavior
+
+Clients subscribe with:
+
+```json
+{"kinds": [24200], "#p": ["<own_pubkey>"], "since": <now>}
+```
+
+On receiving an event, a client MUST:
+
+1. Verify the event signature.
+2. Decrypt `content` using own secret key and `event.pubkey`.
+3. Parse the decrypted payload and dispatch on `kind` (telemetry) or `type` (control).
+4. Ignore unknown `kind`/`type` values.
+
+Clients SHOULD verify that the `agent` tag matches a known/trusted agent pubkey
+before decrypting.
+
+Clients SHOULD buffer events in a bounded ring buffer (RECOMMENDED maximum: 800 events).
+Clients MUST NOT request historical kind 24200 events (no `since` in the past, no
+`until`, no `ids` queries).
+
+## Security Considerations
+
+**Metadata leakage.** Routing tags (`p`, `agent`, `frame`, `created_at`) are
+cleartext. A relay operator can observe that agent X is streaming to owner Y at what
+rate. For maximum metadata privacy, implementors MAY wrap events in NIP-59 gift wrap.
+
+**No forward secrecy.** NIP-44 does not provide forward secrecy; compromise of the
+agent's private key allows decryption of any captured ciphertext.
+
+**Replay attacks.** A captured, signed event could be replayed without a freshness
+check. Relays are RECOMMENDED to enforce a `created_at` freshness window.
+
+**Rogue relays.** The ephemerality contract is relay policy, not cryptography.
+NIP-44 encryption ensures stored events remain opaque to the relay operator absent
+key compromise.
+
+**Best-effort delivery.** Control frames can be dropped during reconnect or queue
+overflow. Control commands SHOULD be treated as advisory with idempotent semantics.
+Agents MUST NOT rely on guaranteed delivery of control frames.
+
+**Operational persistence vectors.** Telemetry may transiently exist in process
+memory, crash dumps, and application logs. Implementations SHOULD minimize logging
+of decrypted payloads and MUST NOT log it at INFO level or above.
+
+## Relationship to Other NIPs
+
+- **NIP-01**: Kind 24200 is in the ephemeral range (20000–29999); standard event
+  structure and signature rules apply.
+- **NIP-42**: Recommended for relay-side authentication gating.
+- **NIP-44**: Required encryption algorithm for all `content` fields.
+- **NIP-29**: An `h` tag MAY be included when the agent session is scoped to a
+  NIP-29 group.
+- **NIP-XX (PR #2226)**: NIP-XX defines the agent *output* plane; this NIP defines
+  the *observability* plane (internal agent activity). They are complementary and
+  non-overlapping.
+
+## Examples
+
+### 1. Telemetry Event — `acp_write` frame
+
+**Wire event (encrypted):**
+
+```json
+{
+  "id":         "a1b2c3d4...",
+  "kind":       24200,
+  "pubkey":     "agent_pubkey_hex",
+  "created_at": 1777464041,
+  "content":    "<NIP-44 v2 ciphertext>",
+  "tags": [
+    ["p",     "owner_pubkey_hex"],
+    ["agent", "agent_pubkey_hex"],
+    ["frame", "telemetry"]
+  ],
+  "sig": "..."
+}
+```
+
+**Decrypted payload:**
+
+```json
+{
+  "seq":        42,
+  "timestamp":  "2026-04-29T12:00:41.500Z",
+  "kind":       "acp_write",
+  "agentIndex": 0,
+  "channelId":  "52a85618-0f8f-4542-94ec-599e6e1c6f2e",
+  "sessionId":  "a1b2c3d4",
+  "turnId":     "e5f6g7h8",
+  "payload": {
+    "jsonrpc": "2.0",
+    "method":  "tools/call",
+    "params":  { "name": "shell", "arguments": { "command": "ls -la" } }
+  }
+}
+```
+
+---
+
+### 2. Control Event — `cancel_turn` frame
+
+**Wire event (encrypted):**
+
+```json
+{
+  "id":         "e5f6a7b8...",
+  "kind":       24200,
+  "pubkey":     "owner_pubkey_hex",
+  "created_at": 1777464042,
+  "content":    "<NIP-44 v2 ciphertext>",
+  "tags": [
+    ["p",     "agent_pubkey_hex"],
+    ["agent", "agent_pubkey_hex"],
+    ["frame", "control"]
+  ],
+  "sig": "..."
+}
+```
+
+**Decrypted payload:**
+
+```json
+{
+  "type":      "cancel_turn",
+  "channelId": "52a85618-0f8f-4542-94ec-599e6e1c6f2e"
+}
+```
+
+## Reference Implementation
+
+[block/sprout PR #421](https://github.com/block/sprout/pull/421)