diff --git a/Cargo.lock b/Cargo.lock index 2cdc92a051..bb1a0e760e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3621,6 +3621,7 @@ name = "sprout-acp" version = "0.1.0" dependencies = [ "anyhow", + "axum", "chrono", "clap", "evalexpr", diff --git a/bin/.flutter-3.41.7.pkg b/bin/.flutter-3.41.7.pkg new file mode 120000 index 0000000000..383f4511d4 --- /dev/null +++ b/bin/.flutter-3.41.7.pkg @@ -0,0 +1 @@ +hermit \ No newline at end of file diff --git a/bin/dart b/bin/dart new file mode 120000 index 0000000000..781b79bc07 --- /dev/null +++ b/bin/dart @@ -0,0 +1 @@ +.flutter-3.41.7.pkg \ No newline at end of file diff --git a/bin/flutter b/bin/flutter new file mode 120000 index 0000000000..781b79bc07 --- /dev/null +++ b/bin/flutter @@ -0,0 +1 @@ +.flutter-3.41.7.pkg \ No newline at end of file diff --git a/crates/sprout-acp/Cargo.toml b/crates/sprout-acp/Cargo.toml index d03079f1d1..a3de291cbe 100644 --- a/crates/sprout-acp/Cargo.toml +++ b/crates/sprout-acp/Cargo.toml @@ -25,6 +25,7 @@ tokio = { workspace = true } # WebSocket tokio-tungstenite = { workspace = true } +axum = { workspace = true } # Codec (bounded line reads) tokio-util = { workspace = true } diff --git a/crates/sprout-acp/src/acp.rs b/crates/sprout-acp/src/acp.rs index e9b2999400..bfd8ccbb46 100644 --- a/crates/sprout-acp/src/acp.rs +++ b/crates/sprout-acp/src/acp.rs @@ -13,6 +13,8 @@ use tokio::io::AsyncWriteExt; use tokio::process::{Child, ChildStdin, ChildStdout}; use tokio_util::codec::{FramedRead, LinesCodec, LinesCodecError}; +use crate::observer::{ObserverContext, ObserverHandle}; + /// Maximum allowed size of a single NDJSON line from the agent's stdout. /// Lines exceeding this limit are rejected to prevent OOM from rogue agents. const MAX_LINE_SIZE: usize = 10_000_000; // 10 MB @@ -137,6 +139,12 @@ pub struct AcpClient { /// Inherited by `cancel_with_cleanup` so the drain loop shares the same budget /// rather than starting a fresh timer (prevents double-jeopardy). current_hard_deadline: Option, + /// Optional local observer feed used by the desktop app. + observer: Option, + /// Pool slot index for this agent process. + observer_agent_index: Option, + /// Best-effort context attached to raw ACP wire events. + observer_context: ObserverContext, } impl AcpClient { @@ -225,9 +233,35 @@ impl AcpClient { permission_responded: false, last_prompt_id: None, current_hard_deadline: None, + observer: None, + observer_agent_index: None, + observer_context: ObserverContext::default(), }) } + /// Attach a local observer feed to this ACP client. + pub fn set_observer(&mut self, observer: Option, agent_index: usize) { + self.observer = observer; + self.observer_agent_index = Some(agent_index); + } + + /// Update metadata that will be attached to subsequent raw wire events. + pub fn set_observer_context(&mut self, context: ObserverContext) { + self.observer_context = context; + } + + /// Emit a semantic event to the local observer feed, if enabled. + pub fn observe(&self, kind: impl Into, payload: serde_json::Value) { + if let Some(observer) = &self.observer { + observer.emit( + kind, + self.observer_agent_index, + &self.observer_context, + payload, + ); + } + } + /// Send the `initialize` request and return the agent's response result value. /// /// Must be called exactly once, before any other ACP method. @@ -484,7 +518,9 @@ impl AcpClient { }) .await .map_err(|_| AcpError::WriteTimeout(WRITE_TIMEOUT))? - .map_err(AcpError::Io) + .map_err(AcpError::Io)?; + self.observe("acp_write", value.clone()); + Ok(()) } /// Default timeout for non-prompt RPCs (initialize, session/new, etc.). @@ -626,6 +662,13 @@ impl AcpClient { let msg: serde_json::Value = match serde_json::from_str(trimmed) { Ok(v) => v, Err(e) => { + self.observe( + "acp_parse_error", + serde_json::json!({ + "line": trimmed, + "error": e.to_string(), + }), + ); tracing::warn!( target: "acp::wire", "failed to parse line as JSON: {e} — skipping" @@ -633,6 +676,7 @@ impl AcpClient { continue; } }; + self.observe("acp_read", msg.clone()); // Check if this is a response to our expected request (has matching id // AND no `method` field — a `method` field means it's an agent-initiated @@ -729,6 +773,13 @@ impl AcpClient { let msg: serde_json::Value = match serde_json::from_str(trimmed) { Ok(v) => v, Err(e) => { + self.observe( + "acp_parse_error", + serde_json::json!({ + "line": trimmed, + "error": e.to_string(), + }), + ); tracing::warn!( target: "acp::wire", "failed to parse line as JSON: {e} — skipping" @@ -736,6 +787,7 @@ impl AcpClient { continue; } }; + self.observe("acp_read", msg.clone()); // Only reset the idle clock on lines that parse as valid JSON. // Malformed lines (skipped above) don't count as real agent activity. diff --git a/crates/sprout-acp/src/config.rs b/crates/sprout-acp/src/config.rs index 368bc99a0b..78a8ca9163 100644 --- a/crates/sprout-acp/src/config.rs +++ b/crates/sprout-acp/src/config.rs @@ -358,6 +358,17 @@ pub struct CliArgs { /// Name of the persona within the pack to use. Required when --persona-pack is set. #[arg(long, env = "SPROUT_ACP_PERSONA_NAME")] pub persona_name: Option, + + /// Local bind address for the read-only ACP observer feed. + /// + /// 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, + + /// Optional query token required by the observer /events endpoint. + #[arg(long, env = "SPROUT_ACP_OBSERVER_TOKEN")] + pub observer_token: Option, } // ── Merged NIP-01 filter ────────────────────────────────────────────────────── @@ -412,6 +423,10 @@ 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, + /// Local observer token, when configured. + pub observer_token: Option, } /// Validate and deduplicate allowlist entries: each must be exactly 64 hex chars. @@ -747,6 +762,8 @@ impl Config { respond_to: args.respond_to, respond_to_allowlist, persona_env_vars, + observer_addr: args.observer_addr, + observer_token: args.observer_token, }; Ok(config) @@ -1107,6 +1124,8 @@ mod tests { respond_to: RespondTo::Anyone, respond_to_allowlist: HashSet::new(), persona_env_vars: vec![], + observer_addr: None, + observer_token: None, } } diff --git a/crates/sprout-acp/src/main.rs b/crates/sprout-acp/src/main.rs index aed16cd0af..c4a9e343e6 100644 --- a/crates/sprout-acp/src/main.rs +++ b/crates/sprout-acp/src/main.rs @@ -3,6 +3,7 @@ mod acp; mod config; mod filter; +mod observer; mod pool; mod queue; mod relay; @@ -529,6 +530,36 @@ 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 = match config.observer_addr.as_deref() { + Some(addr) => { + match observer::spawn_observer_server(addr, config.observer_token.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, + }; + // ── Step 1: Spawn N ACP agent subprocesses and initialize ───────────────── // // Finding #10: one agent failing to start must not kill the whole pool. @@ -549,9 +580,17 @@ async fn tokio_main() -> Result<()> { .await; match spawn_result { Ok(mut acp) => { + acp.set_observer(observer.clone(), i); match tokio::time::timeout(Duration::from_secs(60), acp.initialize()).await { Ok(Ok(init_result)) => { tracing::info!(agent = i, "agent initialized: {init_result}"); + acp.observe( + "agent_initialized", + serde_json::json!({ + "agentIndex": i, + "initializeResult": init_result, + }), + ); agent_slots.push(Some(OwnedAgent { index: i, acp, @@ -918,9 +957,10 @@ async fn tokio_main() -> Result<()> { let cmd = config.agent_command.clone(); let args = config.agent_args.clone(); let env = config.persona_env_vars.clone(); + let observer = observer.clone(); let guard = RespawnGuard::new(idx, respawn_tx.clone()); respawn_tasks.spawn(async move { - let result = spawn_and_init(&cmd, &args, &env).await; + let result = spawn_and_init(&cmd, &args, &env, idx, observer).await; guard.send(result); }); } @@ -1379,6 +1419,7 @@ async fn tokio_main() -> Result<()> { &mut crash_history, &respawn_tx, &mut respawn_tasks, + observer.clone(), ) == LoopAction::Exit { break; @@ -1393,6 +1434,7 @@ async fn tokio_main() -> Result<()> { &mut crash_history, &respawn_tx, &mut respawn_tasks, + observer.clone(), ) == LoopAction::Exit { break; @@ -1414,6 +1456,7 @@ async fn tokio_main() -> Result<()> { &mut crash_history, &respawn_tx, &mut respawn_tasks, + observer.clone(), ); if pool.live_count() == 0 && !any_respawn_in_flight(&crash_history) { tracing::error!("all agents dead — exiting"); @@ -1643,6 +1686,7 @@ fn handle_prompt_result( crash_history: &mut [SlotCircuit], respawn_tx: &mpsc::Sender, respawn_tasks: &mut tokio::task::JoinSet<()>, + observer: Option, ) -> LoopAction { let before = pool.task_map().len(); let agent_index = result.agent.index; @@ -1722,6 +1766,7 @@ fn handle_prompt_result( slot_history, respawn_tx, respawn_tasks, + observer.clone(), ) { // Circuit open — slot stays empty until maintenance refill. if pool.live_count() == 0 && !any_respawn_in_flight(crash_history) { @@ -1775,6 +1820,7 @@ fn handle_prompt_result( slot_history, respawn_tx, respawn_tasks, + observer, ) && pool.live_count() == 0 && !any_respawn_in_flight(crash_history) { @@ -1808,6 +1854,7 @@ fn recover_panicked_agent( crash_history: &mut [SlotCircuit], respawn_tx: &mpsc::Sender, respawn_tasks: &mut tokio::task::JoinSet<()>, + observer: Option, ) { let task_id = join_error.id(); let Some(meta) = pool.task_map_mut().remove(&task_id) else { @@ -1874,7 +1921,7 @@ fn recover_panicked_agent( if !delay.is_zero() { tokio::time::sleep(delay).await; } - let result = spawn_and_init(&cmd, &args, &env).await; + let result = spawn_and_init(&cmd, &args, &env, i, observer).await; guard.send(result); }); } @@ -1892,6 +1939,7 @@ fn drain_ready_join_results( crash_history: &mut [SlotCircuit], respawn_tx: &mpsc::Sender, respawn_tasks: &mut tokio::task::JoinSet<()>, + observer: Option, ) -> LoopAction { while let Some(Some(join_result)) = pool.join_set.join_next().now_or_never() { if let Err(join_error) = join_result { @@ -1907,6 +1955,7 @@ fn drain_ready_join_results( crash_history, respawn_tx, respawn_tasks, + observer.clone(), ); if pool.live_count() == 0 && !any_respawn_in_flight(crash_history) { return LoopAction::Exit; @@ -1991,6 +2040,7 @@ fn spawn_respawn_task( slot: &mut SlotCircuit, respawn_tx: &mpsc::Sender, respawn_tasks: &mut tokio::task::JoinSet<()>, + observer: Option, ) -> bool { let index = old_agent.index; @@ -2027,7 +2077,7 @@ fn spawn_respawn_task( tokio::time::sleep(delay).await; } - let result = spawn_and_init(&cmd, &args, &env).await; + let result = spawn_and_init(&cmd, &args, &env, index, observer).await; guard.send(result); }); @@ -2044,14 +2094,24 @@ async fn spawn_and_init( command: &str, args: &[String], extra_env: &[(String, String)], + agent_index: usize, + observer: Option, ) -> Result { let mut acp = AcpClient::spawn(command, args, extra_env) .await .map_err(|e| anyhow::anyhow!("failed to spawn agent: {e}"))?; + acp.set_observer(observer, agent_index); match acp.initialize().await { Ok(init_result) => { tracing::info!("agent initialized: {init_result}"); + acp.observe( + "agent_initialized", + serde_json::json!({ + "agentIndex": agent_index, + "initializeResult": init_result, + }), + ); Ok(acp) } Err(e) => { diff --git a/crates/sprout-acp/src/observer.rs b/crates/sprout-acp/src/observer.rs new file mode 100644 index 0000000000..f2652d44e9 --- /dev/null +++ b/crates/sprout-acp/src/observer.rs @@ -0,0 +1,292 @@ +//! Local read-only observer feed for ACP session activity. +//! +//! This is intentionally process-local infrastructure: it lets the desktop app +//! watch the raw ACP JSON-RPC stream without sending private execution detail +//! through the Sprout relay. + +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, + Router, +}; +use futures_util::{stream, StreamExt}; +use serde::Serialize; +use tokio::sync::broadcast; + +const OBSERVER_BUFFER_CAP: usize = 1_000; + +#[derive(Clone, Debug, Default)] +pub struct ObserverContext { + pub channel_id: Option, + pub session_id: Option, + pub turn_id: Option, +} + +#[derive(Clone)] +pub struct ObserverHandle { + inner: Arc, +} + +struct ObserverInner { + tx: broadcast::Sender, + buffer: Mutex>, + seq: AtomicU64, + addr: SocketAddr, +} + +#[derive(Clone, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct ObserverEvent { + pub seq: u64, + pub timestamp: String, + pub kind: String, + pub agent_index: Option, + pub channel_id: Option, + pub session_id: Option, + pub turn_id: Option, + pub payload: serde_json::Value, +} + +#[derive(Clone)] +struct ObserverServerState { + observer: ObserverHandle, + token: Option, +} + +#[derive(serde::Deserialize)] +struct EventQuery { + token: Option, +} + +pub async fn spawn_observer_server( + bind_addr: &str, + token: Option, +) -> anyhow::Result { + 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, + }; + let app = Router::new() + .route("/events", get(events_handler)) + .route("/health", get(health_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 { + pub fn addr(&self) -> SocketAddr { + self.inner.addr + } + + pub fn subscribe(&self) -> broadcast::Receiver { + self.inner.tx.subscribe() + } + + pub fn snapshot(&self) -> Vec { + match self.inner.buffer.lock() { + Ok(buffer) => buffer.iter().cloned().collect(), + Err(error) => { + tracing::warn!(target: "observer", "observer replay buffer lock poisoned: {error}"); + Vec::new() + } + } + } + + pub fn emit( + &self, + kind: impl Into, + agent_index: Option, + context: &ObserverContext, + payload: serde_json::Value, + ) { + let event = ObserverEvent { + seq: self.inner.seq.fetch_add(1, Ordering::Relaxed), + timestamp: chrono::Utc::now().to_rfc3339(), + kind: kind.into(), + agent_index, + channel_id: context.channel_id.clone(), + session_id: context.session_id.clone(), + turn_id: context.turn_id.clone(), + payload, + }; + + match self.inner.buffer.lock() { + Ok(mut buffer) => { + if buffer.len() >= OBSERVER_BUFFER_CAP { + buffer.pop_front(); + } + buffer.push_back(event.clone()); + } + Err(error) => { + tracing::warn!(target: "observer", "observer replay buffer lock poisoned: {error}"); + } + } + + let _ = self.inner.tx.send(event); + } +} + +async fn health_handler(State(state): State) -> Response { + serde_json::json!({ + "ok": true, + "addr": state.observer.addr().to_string(), + }) + .to_string() + .into_response() +} + +async fn events_handler( + State(state): State, + Query(query): Query, + headers: HeaderMap, +) -> Response { + if !token_matches(state.token.as_deref(), query.token.as_deref()) { + 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 +} + +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 { + 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)) +} + +pub fn context_for( + channel_id: Option, + session_id: Option, + turn_id: Option, +) -> ObserverContext { + ObserverContext { + channel_id: channel_id.map(|id| id.to_string()), + session_id, + turn_id, + } +} + +#[cfg(test)] +mod tests { + use axum::http::HeaderValue; + + use super::{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_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" + )))); + } +} diff --git a/crates/sprout-acp/src/pool.rs b/crates/sprout-acp/src/pool.rs index da8e251114..8105c5d1b9 100644 --- a/crates/sprout-acp/src/pool.rs +++ b/crates/sprout-acp/src/pool.rs @@ -33,6 +33,7 @@ use crate::acp::{ AcpError, McpServer, ModelSwitchMethod, StopReason, }; use crate::config::{DedupMode, PermissionMode}; +use crate::observer; use crate::queue::{ ContextMessage, ConversationContext, FlushBatch, PromptChannelInfo, PromptProfile, PromptProfileLookup, @@ -605,6 +606,30 @@ pub async fn run_prompt_task( Some(b) => PromptSource::Channel(b.channel_id), None => PromptSource::Heartbeat, }; + let turn_id = uuid::Uuid::new_v4().to_string(); + let observer_channel_id = match &source { + PromptSource::Channel(channel_id) => Some(*channel_id), + PromptSource::Heartbeat => None, + }; + agent.acp.set_observer_context(observer::context_for( + observer_channel_id, + None, + Some(turn_id.clone()), + )); + let triggering_event_ids: Vec = batch + .as_ref() + .map(|b| b.events.iter().map(|be| be.event.id.to_hex()).collect()) + .unwrap_or_default(); + agent.acp.observe( + "turn_started", + serde_json::json!({ + "source": match &source { + PromptSource::Channel(_) => "channel", + PromptSource::Heartbeat => "heartbeat", + }, + "triggeringEventIds": triggering_event_ids, + }), + ); // ── Reaction cleanup guard ──────────────────────────────────────────── // Collects event IDs up front. On drop (any exit path — normal, early @@ -690,6 +715,18 @@ pub async fn run_prompt_task( } } }; + agent.acp.set_observer_context(observer::context_for( + observer_channel_id, + Some(session_id.clone()), + Some(turn_id.clone()), + )); + agent.acp.observe( + "session_resolved", + serde_json::json!({ + "sessionId": session_id, + "isNewSession": is_new_session, + }), + ); // ── Send initial_message on new channel sessions ────────────────────── diff --git a/desktop/scripts/check-file-sizes.mjs b/desktop/scripts/check-file-sizes.mjs index 7e4c101841..0c1894aea3 100644 --- a/desktop/scripts/check-file-sizes.mjs +++ b/desktop/scripts/check-file-sizes.mjs @@ -67,7 +67,7 @@ const overrides = new Map([ ["src-tauri/src/huddle/tts.rs", 1030], // TTS pipeline + session warmup + cancel/shutdown handling + apply_fades + 18 unit tests for remote interrupt mechanism ["src-tauri/src/commands/pairing.rs", 550], // NIP-AB pairing actor: 3 Tauri commands + background WS task + NIP-42 auth + event parsing helpers ["src-tauri/src/lib.rs", 715], // +4 lines for PairingHandle managed state + 3 pairing command registrations - ["src/shared/api/tauri.ts", 1110], // +14 lines for 3 NIP-AB pairing command wrappers + ["src/shared/api/tauri.ts", 1112], // +14 lines for 3 NIP-AB pairing command wrappers + observer_url field ]); async function walkFiles(directory) { diff --git a/desktop/src-tauri/src/managed_agents/mod.rs b/desktop/src-tauri/src/managed_agents/mod.rs index 5ce89dddc4..ce42d630ab 100644 --- a/desktop/src-tauri/src/managed_agents/mod.rs +++ b/desktop/src-tauri/src/managed_agents/mod.rs @@ -1,6 +1,7 @@ mod backend; mod discovery; mod nest; +mod observer; mod persona_avatars; mod persona_card; mod personas; @@ -13,6 +14,7 @@ mod types; pub use backend::*; pub use discovery::*; pub use nest::*; +pub use observer::*; pub use persona_card::*; pub use personas::*; pub use restore::*; diff --git a/desktop/src-tauri/src/managed_agents/observer.rs b/desktop/src-tauri/src/managed_agents/observer.rs new file mode 100644 index 0000000000..b57aa72868 --- /dev/null +++ b/desktop/src-tauri/src/managed_agents/observer.rs @@ -0,0 +1,20 @@ +pub struct ObserverEndpoint { + pub addr: String, + pub token: String, + pub url: String, +} + +pub fn allocate_observer_endpoint() -> Result { + 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 }) +} diff --git a/desktop/src-tauri/src/managed_agents/restore.rs b/desktop/src-tauri/src/managed_agents/restore.rs index a390c28bb2..aae7ccfcad 100644 --- a/desktop/src-tauri/src/managed_agents/restore.rs +++ b/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), String>, + Result<(std::process::Child, std::path::PathBuf, Option), 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)) => { + Ok((child, log_path, observer_url)) => { let now = util::now_iso(); record.updated_at = now.clone(); record.runtime_pid = Some(child.id()); @@ -133,7 +133,14 @@ 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 }); + runtimes.insert( + pubkey, + ManagedAgentProcess { + child, + log_path, + observer_url, + }, + ); } Err(error) => { record.updated_at = util::now_iso(); diff --git a/desktop/src-tauri/src/managed_agents/runtime.rs b/desktop/src-tauri/src/managed_agents/runtime.rs index 0b4df7e570..0f10641835 100644 --- a/desktop/src-tauri/src/managed_agents/runtime.rs +++ b/desktop/src-tauri/src/managed_agents/runtime.rs @@ -4,9 +4,9 @@ use tauri::AppHandle; use crate::{ managed_agents::{ - append_log_marker, login_shell_path, managed_agent_log_path, missing_command_message, - normalize_agent_args, open_log_file, resolve_command, ManagedAgentProcess, - ManagedAgentRecord, ManagedAgentSummary, + 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, }, util::now_iso, }; @@ -343,7 +343,7 @@ pub fn build_managed_agent_summary( ) -> Result { use crate::managed_agents::BackendKind; - let (status, pid, log_path) = if record.backend != BackendKind::Local { + let (status, pid, log_path, observer_url) = 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()) + (status, None, String::new(), None) } else { let persisted_pid = record.runtime_pid.filter(|pid| process_is_running(*pid)); if let Some(runtime) = runtimes.get(&record.pubkey) { @@ -372,6 +372,7 @@ 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 { ( @@ -380,6 +381,7 @@ pub fn build_managed_agent_summary( managed_agent_log_path(app, &record.pubkey)? .display() .to_string(), + None, ) } else { ( @@ -388,6 +390,7 @@ pub fn build_managed_agent_summary( managed_agent_log_path(app, &record.pubkey)? .display() .to_string(), + None, ) } }; @@ -421,6 +424,7 @@ 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, }) } @@ -440,7 +444,7 @@ pub fn find_managed_agent_mut<'a>( pub fn spawn_agent_child( app: &AppHandle, record: &ManagedAgentRecord, -) -> Result<(std::process::Child, std::path::PathBuf), String> { +) -> Result<(std::process::Child, std::path::PathBuf, Option), String> { let log_path = managed_agent_log_path(app, &record.pubkey)?; append_log_marker( &log_path, @@ -561,6 +565,10 @@ 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); + // Spawn the harness in its own process group so we can kill the entire // tree (harness + MCP servers + agent subprocesses) on shutdown. #[cfg(unix)] @@ -579,7 +587,7 @@ pub fn spawn_agent_child( let _ = super::write_agent_pid_file(app, &record.pubkey, child.id()); - Ok((child, log_path)) + Ok((child, log_path, Some(observer.url))) } pub fn start_managed_agent_process( @@ -610,7 +618,7 @@ pub fn start_managed_agent_process( record.runtime_pid = None; } - let (child, log_path) = spawn_agent_child(app, record)?; + let (child, log_path, observer_url) = spawn_agent_child(app, record)?; let now = now_iso(); record.updated_at = now.clone(); @@ -622,7 +630,11 @@ pub fn start_managed_agent_process( runtimes.insert( record.pubkey.clone(), - ManagedAgentProcess { child, log_path }, + ManagedAgentProcess { + child, + log_path, + observer_url, + }, ); Ok(()) } diff --git a/desktop/src-tauri/src/managed_agents/types.rs b/desktop/src-tauri/src/managed_agents/types.rs index af73e8a066..5f9d528b43 100644 --- a/desktop/src-tauri/src/managed_agents/types.rs +++ b/desktop/src-tauri/src/managed_agents/types.rs @@ -120,6 +120,7 @@ pub struct ManagedAgentRecord { pub struct ManagedAgentProcess { pub child: Child, pub log_path: PathBuf, + pub observer_url: Option, } #[derive(Debug, Clone, Serialize)] @@ -152,6 +153,7 @@ pub struct ManagedAgentSummary { pub last_error: Option, pub start_on_app_launch: bool, pub log_path: String, + pub observer_url: Option, } #[derive(Debug, Deserialize)] diff --git a/desktop/src/features/agents/ui/AgentSessionToolItem.tsx b/desktop/src/features/agents/ui/AgentSessionToolItem.tsx new file mode 100644 index 0000000000..6d4205ea15 --- /dev/null +++ b/desktop/src/features/agents/ui/AgentSessionToolItem.tsx @@ -0,0 +1,352 @@ +import * as React from "react"; +import { ArrowUpRight, ChevronDown, Wrench } from "lucide-react"; + +import { useAppNavigation } from "@/app/navigation/useAppNavigation"; +import type { Channel } from "@/shared/api/types"; +import { useChannelNavigation } from "@/shared/context/ChannelNavigationContext"; +import { cn } from "@/shared/lib/cn"; +import type { TranscriptItem } from "./agentSessionTypes"; +import { + formatToolTitle, + getSproutToolInfo, + getToolStatusDisplay, +} from "./agentSessionToolCatalog"; +import { + asRecord, + formatCodeValue, + getResultArray, + getToolString, + getToolStringList, + shortenMiddle, +} from "./agentSessionUtils"; + +export function ToolItem({ + item, +}: { + item: Extract; +}) { + const [isExpanded, setIsExpanded] = React.useState(false); + const status = getToolStatusDisplay(item.status, item.isError); + const hasArgs = Object.keys(item.args).length > 0; + const hasResult = item.result.trim().length > 0; + const canonicalToolName = item.sproutToolName ?? item.toolName; + const sproutTool = getSproutToolInfo(canonicalToolName); + const ToolIcon = sproutTool?.icon ?? Wrench; + const showStatus = status.state !== "output-available"; + const toolTitle = formatToolTitle(canonicalToolName, item.title); + const handleToggle = React.useCallback( + (event: React.SyntheticEvent) => { + setIsExpanded(event.currentTarget.open); + }, + [], + ); + + return ( +
+
+ + {ToolIcon ? ( + + ) : null} + + {toolTitle} + + {sproutTool ? ( + + ) : null} + {showStatus ? ( + + + {status.label} + + ) : null} + + + + +
+
+ ); +} + +function ToolDetailBlocks({ + args, + description, + hasArgs, + hasResult, + isError, + result, +}: { + args: Record; + description?: string; + hasArgs: boolean; + hasResult: boolean; + isError: boolean; + result: string; +}) { + return ( +
+ {description ? ( +

+ {description} +

+ ) : null} + {hasArgs ? ( + + ) : null} + {hasResult ? ( + + ) : null} + {!hasArgs && !hasResult ? ( +

+ Waiting for tool details. +

+ ) : null} +
+ ); +} + +function ToolCodeBlock({ + label, + tone, + value, +}: { + label: string; + tone: "muted" | "error"; + value: string; +}) { + return ( +
+

+ {label} +

+ 
+        {formatCodeValue(value)}
+
+
+ ); +} + +function SproutToolInlineAction({ + args, + result, +}: { + args: Record; + result: string; +}) { + const { channels } = useChannelNavigation(); + const { goChannel } = useAppNavigation(); + const resultValue = React.useMemo( + () => parseToolResultValue(result), + [result], + ); + const resultRecord = asRecord(resultValue); + const channelId = + getToolString(args, ["channel_id", "channelId"]) ?? + getToolString(resultRecord, ["channel_id", "channelId"]); + const openChannel = React.useCallback( + (messageId?: string) => { + if (!channelId) return; + void goChannel(channelId, messageId ? { messageId } : undefined); + }, + [channelId, goChannel], + ); + const action = React.useMemo( + () => + getSproutToolInlineAction({ + args, + channelId, + channels, + openChannel, + resultValue, + }), + [args, channelId, channels, openChannel, resultValue], + ); + + if (!action) { + return null; + } + + if (action.onClick) { + return ( + + ); + } + + return ( + + {action.label} + {action.value} + + ); +} + +type SproutToolInlineActionModel = { + label: string; + value: string; + title: string; + onClick?: () => void; +}; + +function getSproutToolInlineAction({ + args, + channelId, + channels, + openChannel, + resultValue, +}: { + args: Record; + channelId: string | null; + channels: Channel[]; + openChannel: (messageId?: string) => void; + resultValue: unknown; +}): SproutToolInlineActionModel | null { + const resultRecord = asRecord(resultValue); + const eventId = + getToolString(args, ["event_id", "eventId"]) ?? + getToolString(resultRecord, ["event_id", "eventId", "id"]); + + if (eventId && channelId) { + return { + label: resultRecord.accepted === true ? "posted" : "event", + onClick: () => openChannel(eventId), + title: eventId, + value: getChannelChipLabel(channels, channelId), + }; + } + + const messages = getResultArray(resultValue, resultRecord, "messages"); + if (messages) { + return { + label: "read", + onClick: channelId ? () => openChannel() : undefined, + title: `${messages.length} messages`, + value: `${messages.length} message${messages.length === 1 ? "" : "s"}`, + }; + } + + if (channelId) { + return { + label: "channel", + onClick: () => openChannel(), + title: channelId, + value: getChannelChipLabel(channels, channelId), + }; + } + + const workflowId = + getToolString(args, ["workflow_id", "workflowId"]) ?? + getToolString(resultRecord, ["workflow_id", "workflowId"]); + if (workflowId) { + return { + label: "workflow", + title: workflowId, + value: shortenMiddle(workflowId, 26), + }; + } + + const pubkeys = getToolStringList(args, ["pubkeys", "pubkey"]); + if (pubkeys.length > 0) { + return { + label: pubkeys.length === 1 ? "pubkey" : "users", + title: pubkeys.join(", "), + value: + pubkeys.length === 1 + ? shortenMiddle(pubkeys[0], 24) + : `${pubkeys.length} pubkeys`, + }; + } + + const query = getToolString(args, ["query"]); + if (query) { + return { + label: "query", + title: query, + value: shortenMiddle(query, 30), + }; + } + + if (typeof resultRecord.accepted === "boolean") { + return { + label: "relay", + title: resultRecord.accepted ? "accepted" : "rejected", + value: resultRecord.accepted ? "accepted" : "rejected", + }; + } + + return null; +} + +function parseToolResultValue(result: string): unknown { + const trimmed = result.trim(); + if (!trimmed) return null; + + try { + const parsed = JSON.parse(trimmed); + if (typeof parsed !== "string") return parsed; + try { + return JSON.parse(parsed); + } catch { + return parsed; + } + } catch { + return null; + } +} + +function getChannelChipLabel(channels: Channel[], channelId: string) { + const channel = channels.find((candidate) => candidate.id === channelId); + return channel ? `#${channel.name}` : `#${shortenMiddle(channelId, 22)}`; +} diff --git a/desktop/src/features/agents/ui/AgentSessionTranscriptList.tsx b/desktop/src/features/agents/ui/AgentSessionTranscriptList.tsx new file mode 100644 index 0000000000..a776e47f07 --- /dev/null +++ b/desktop/src/features/agents/ui/AgentSessionTranscriptList.tsx @@ -0,0 +1,184 @@ +import { Bot, Brain, ChevronDown, Radio, TerminalSquare } from "lucide-react"; + +import { cn } from "@/shared/lib/cn"; +import { Markdown } from "@/shared/ui/markdown"; +import type { TranscriptItem } from "./agentSessionTypes"; +import { ToolItem } from "./AgentSessionToolItem"; + +export function AgentSessionTranscriptList({ + agentName, + emptyDescription, + items, +}: { + agentName: string; + emptyDescription: string; + items: TranscriptItem[]; +}) { + if (items.length === 0) { + return ( +
+ +

No ACP activity yet

+

{emptyDescription}

+
+ ); + } + + return ( +
+ {items.map((item) => ( +
+ +
+ ))} +
+ ); +} + +function TranscriptItemView({ + agentName, + item, +}: { + agentName: string; + item: TranscriptItem; +}) { + if (item.type === "message") { + return ; + } + if (item.type === "tool") { + return ; + } + if (item.type === "thought") { + return ; + } + if (item.type === "metadata") { + return ; + } + return ; +} + +function MessageItem({ + agentName, + item, +}: { + agentName: string; + item: Extract; +}) { + const isAssistant = item.role === "assistant"; + const text = item.text.trim(); + return ( +
+
+ {isAssistant ? ( +
+ + + + {agentName} +
+ ) : null} +
+ {isAssistant ? ( + + ) : ( +

{text}

+ )} +
+
+
+ ); +} + +function ThoughtItem({ + item, +}: { + item: Extract; +}) { + return ( +
+ + + {item.title} + + +
+ +
+
+ ); +} + +function MetadataItem({ + item, +}: { + item: Extract; +}) { + return ( +
+ + + {item.title} + + {item.sections.length} section{item.sections.length === 1 ? "" : "s"} + + + +
+ {item.sections.map((section) => ( +
+ + {section.title} + + + 
+              {section.body.trim() || "No metadata."}
+
+
+ ))} +
+
+ ); +} + +function LifecycleItem({ + item, +}: { + item: Extract; +}) { + const isError = item.title.toLowerCase().includes("error"); + return ( +
+ {item.title} + {item.text ? - {item.text} : null} +
+ ); +} diff --git a/desktop/src/features/agents/ui/ManagedAgentSessionPanel.tsx b/desktop/src/features/agents/ui/ManagedAgentSessionPanel.tsx new file mode 100644 index 0000000000..18e642f6c3 --- /dev/null +++ b/desktop/src/features/agents/ui/ManagedAgentSessionPanel.tsx @@ -0,0 +1,244 @@ +import * as React from "react"; +import { + CircleAlert, + CircleDot, + Clock3, + Loader2, + TerminalSquare, + XCircle, +} from "lucide-react"; + +import type { ManagedAgent } from "@/shared/api/types"; +import { cn } from "@/shared/lib/cn"; +import { Badge } from "@/shared/ui/badge"; +import { Skeleton } from "@/shared/ui/skeleton"; +import { AgentSessionTranscriptList } from "./AgentSessionTranscriptList"; +import { RawEventRail } from "./RawEventRail"; +import type { + ConnectionState, + ObserverEvent, + TranscriptItem, +} from "./agentSessionTypes"; +import { buildTranscript } from "./agentSessionTranscript"; +import { shorten } from "./agentSessionUtils"; +import { useObserverEvents } from "./useObserverEvents"; + +type ManagedAgentSessionPanelProps = { + agent: ManagedAgent; + channelId?: string | null; + className?: string; + emptyDescription?: string; + showHeader?: boolean; + showRaw?: boolean; +}; + +export function ManagedAgentSessionPanel({ + agent, + channelId = null, + className, + emptyDescription = "Mention this agent in a channel to watch the next turn.", + showHeader = true, + showRaw = true, +}: ManagedAgentSessionPanelProps) { + const { connectionState, errorMessage, events } = useObserverEvents( + agent.observerUrl, + agent.status === "running", + ); + const scopedEvents = React.useMemo( + () => + channelId + ? events.filter((event) => event.channelId === channelId) + : events, + [channelId, events], + ); + const transcript = React.useMemo( + () => buildTranscript(scopedEvents), + [scopedEvents], + ); + const latestSessionId = React.useMemo( + () => + [...scopedEvents].reverse().find((event) => event.sessionId)?.sessionId, + [scopedEvents], + ); + + return ( +
+ {showHeader ? ( + + ) : null} + + +
+ ); +} + +function SessionHeader({ + connectionState, + eventCount, + hasObserver, + latestSessionId, +}: { + connectionState: ConnectionState; + eventCount: number; + hasObserver: boolean; + latestSessionId: string | null | undefined; +}) { + return ( +
+
+
+

+ Live ACP session +

+ +
+

+ {hasObserver + ? latestSessionId + ? `Session ${shorten(latestSessionId)}` + : "Waiting for the next agent turn." + : "Restart this local agent to attach the observer feed."} +

+
+ + {eventCount} event{eventCount === 1 ? "" : "s"} + +
+ ); +} + +function SessionBody({ + agentName, + connectionState, + emptyDescription, + errorMessage, + events, + hasObserver, + showRaw, + transcript, +}: { + agentName: string; + connectionState: ConnectionState; + emptyDescription: string; + errorMessage: string | null; + events: ObserverEvent[]; + hasObserver: boolean; + showRaw: boolean; + transcript: TranscriptItem[]; +}) { + return ( + <> + {!hasObserver ? ( + + ) : connectionState === "connecting" && events.length === 0 ? ( + + ) : ( +
+ + {showRaw ? : null} +
+ )} + + {errorMessage ? ( +

+ + {errorMessage} +

+ ) : null} + + ); +} + +function SessionLoadingSkeleton() { + return ( +
+
+
+ +
+
+
+
+ + +
+
+ + + +
+
+
+ + +
+
+ ); +} + +function ObserverStatusBadge({ state }: { state: ConnectionState }) { + const display = + state === "open" + ? { label: "Live", Icon: CircleDot, variant: "default" as const } + : state === "connecting" + ? { label: "Connecting", Icon: Loader2, variant: "secondary" as const } + : state === "error" + ? { + label: "Unavailable", + Icon: XCircle, + variant: "destructive" as const, + } + : state === "closed" + ? { label: "Closed", Icon: Clock3, variant: "secondary" as const } + : { label: "Idle", Icon: Clock3, variant: "secondary" as const }; + + return ( + + + {display.label} + + ); +} + +function EmptyObserverState() { + return ( +
+ +

Observer not attached

+

+ The live feed is available for local agents started after this update. +

+
+ ); +} diff --git a/desktop/src/features/agents/ui/RawEventRail.tsx b/desktop/src/features/agents/ui/RawEventRail.tsx new file mode 100644 index 0000000000..74a655b77e --- /dev/null +++ b/desktop/src/features/agents/ui/RawEventRail.tsx @@ -0,0 +1,53 @@ +import * as React from "react"; + +import { Button } from "@/shared/ui/button"; +import type { ObserverEvent } from "./agentSessionTypes"; +import { describeRawEvent } from "./agentSessionTranscript"; + +export function RawEventRail({ events }: { events: ObserverEvent[] }) { + const [expanded, setExpanded] = React.useState(false); + const visible = expanded ? events : events.slice(-18); + + return ( + + ); +} diff --git a/desktop/src/features/agents/ui/agentSessionToolCatalog.ts b/desktop/src/features/agents/ui/agentSessionToolCatalog.ts new file mode 100644 index 0000000000..4ff11e0b41 --- /dev/null +++ b/desktop/src/features/agents/ui/agentSessionToolCatalog.ts @@ -0,0 +1,286 @@ +import { + CheckCircle2, + CircleDot, + Clock3, + Hash, + MessageSquare, + Search, + Send, + Users, + Workflow, + XCircle, +} from "lucide-react"; + +import type { SproutToolInfo, ToolStatus } from "./agentSessionTypes"; + +export function normalizeToolStatus(status: string): ToolStatus { + const normalized = status.toLowerCase(); + if ( + normalized.includes("complete") || + normalized.includes("success") || + normalized === "done" + ) { + return "completed"; + } + if (normalized.includes("fail") || normalized.includes("error")) { + return "failed"; + } + if (normalized.includes("pending")) { + return "pending"; + } + return "executing"; +} + +export function getToolStatusDisplay(status: ToolStatus, isError: boolean) { + if (isError || status === "failed") { + return { + label: "Error", + Icon: XCircle, + state: "output-error" as const, + variant: "destructive" as const, + }; + } + if (status === "completed") { + return { + label: "Done", + Icon: CheckCircle2, + state: "output-available" as const, + variant: "secondary" as const, + }; + } + if (status === "pending") { + return { + label: "Pending", + Icon: CircleDot, + state: "input-streaming" as const, + variant: "secondary" as const, + }; + } + return { + label: "Running", + Icon: Clock3, + state: "input-available" as const, + variant: "secondary" as const, + }; +} + +const SPROUT_READ_TOOLS = new Set([ + "get_messages", + "get_channel_history", + "get_thread", + "search", + "get_feed", + "get_reactions", + "list_channels", + "get_channel", + "get_users", + "get_presence", + "list_channel_members", + "list_dms", + "get_canvas", + "list_workflows", + "get_workflow_runs", + "get_event", + "get_user_notes", + "get_contact_list", +]); + +const SPROUT_WRITE_TOOLS = new Set([ + "send_message", + "send_diff_message", + "edit_message", + "delete_message", + "add_reaction", + "remove_reaction", + "join_channel", + "leave_channel", + "update_channel", + "set_channel_topic", + "set_channel_purpose", + "open_dm", + "set_profile", + "set_presence", + "trigger_workflow", + "approve_step", + "create_channel", + "archive_channel", + "unarchive_channel", + "add_channel_member", + "remove_channel_member", + "add_dm_member", + "hide_dm", + "set_canvas", + "create_workflow", + "update_workflow", + "delete_workflow", + "set_channel_add_policy", + "vote_on_post", + "publish_note", + "set_contact_list", +]); + +const SPROUT_TOOL_NAMES = new Set([ + ...SPROUT_READ_TOOLS, + ...SPROUT_WRITE_TOOLS, +]); + +const SPROUT_TOOL_NAMES_BY_LENGTH = [...SPROUT_TOOL_NAMES].sort( + (left, right) => right.length - left.length, +); + +const SPROUT_TOOL_TITLE_ALIASES: Array<[RegExp, string]> = [ + [/\bsending message to channel\b/, "send_message"], + [/\bretrieving recent messages from channel\b/, "get_messages"], + [/\bgetting channel details\b/, "get_channel"], + [/\bgetting user information\b/, "get_users"], + [/\bsearching relay history\b/, "search"], + [/\bgetting thread\b/, "get_thread"], + [/\badding reaction\b/, "add_reaction"], + [/\bremoving reaction\b/, "remove_reaction"], +]; + +export function getSproutToolInfo(title: string): SproutToolInfo | null { + const name = normalizeToolName(title); + const isRead = SPROUT_READ_TOOLS.has(name); + const isWrite = SPROUT_WRITE_TOOLS.has(name); + if (!isRead && !isWrite) { + return null; + } + + if (name.includes("workflow") || name === "approve_step") { + return { + icon: Workflow, + label: isRead + ? "Reads workflow state from Sprout." + : "Updates workflow state in Sprout.", + tone: isWrite ? "write" : "read", + }; + } + if ( + name.includes("channel") || + name.includes("messages") || + name === "get_thread" + ) { + return { + icon: Hash, + label: isRead + ? "Reads channel context from the Sprout relay." + : "Changes channel state in the Sprout relay.", + tone: isWrite ? "write" : "read", + }; + } + if ( + name.includes("user") || + name.includes("member") || + name.includes("presence") + ) { + return { + icon: Users, + label: isRead + ? "Reads Sprout identity or presence data." + : "Updates Sprout identity or membership data.", + tone: isWrite ? "write" : "admin", + }; + } + if (name.includes("search") || name === "get_feed") { + return { + icon: Search, + label: "Searches relay-visible Sprout history.", + tone: "read", + }; + } + if ( + name.startsWith("send_") || + name.includes("reaction") || + name === "publish_note" + ) { + return { + icon: Send, + label: "Publishes relay-visible Sprout activity.", + tone: "write", + }; + } + + return { + icon: MessageSquare, + label: isRead ? "Reads from Sprout." : "Writes to Sprout.", + tone: isWrite ? "write" : "read", + }; +} + +export function normalizeToolName(title: string): string { + const knownName = findSproutToolName(title, true); + if (knownName) return knownName; + + const normalized = normalizeToolNameText(title) + .replace(/^sprout_mcp_/, "") + .replace(/^sprout_/, ""); + return normalized.match(/[a-z][a-z0-9_]+/)?.[0] ?? normalized; +} + +export function normalizeToolNameText(value: string): string { + return value + .trim() + .toLowerCase() + .replace(/[^a-z0-9_]+/g, "_") + .replace(/_+/g, "_") + .replace(/^_+|_+$/g, ""); +} + +export function findSproutToolName(value: string, includeShortNames: boolean) { + const alias = findSproutToolAlias(value); + if (alias) return alias; + + const normalized = normalizeToolNameText(value); + return ( + SPROUT_TOOL_NAMES_BY_LENGTH.find( + (name) => + (includeShortNames || name.length >= 8) && normalized.includes(name), + ) ?? null + ); +} + +function findSproutToolAlias(value: string) { + const normalizedPhrase = value + .trim() + .toLowerCase() + .replace(/[_-]+/g, " ") + .replace(/\s+/g, " "); + return ( + SPROUT_TOOL_TITLE_ALIASES.find(([pattern]) => + pattern.test(normalizedPhrase), + )?.[1] ?? null + ); +} + +export function isGenericToolTitle(value: string): boolean { + const normalized = normalizeToolNameText(value); + return ( + normalized.length === 0 || + normalized === "tool" || + normalized === "tool_call" || + normalized === "mcp_tool_call" || + normalized === "unknown" || + normalized === "read" || + normalized === "write" || + normalized === "execute" || + normalized === "completed" + ); +} + +export function formatToolTitle( + toolName: string, + fallbackTitle?: string, +): string { + const name = normalizeToolName(toolName); + if (SPROUT_READ_TOOLS.has(name) || SPROUT_WRITE_TOOLS.has(name)) { + return name + .split("_") + .map((part) => part.charAt(0).toUpperCase() + part.slice(1)) + .join(" "); + } + if (fallbackTitle && !isGenericToolTitle(fallbackTitle)) { + return fallbackTitle; + } + return toolName; +} diff --git a/desktop/src/features/agents/ui/agentSessionTranscript.ts b/desktop/src/features/agents/ui/agentSessionTranscript.ts new file mode 100644 index 0000000000..5f65ec4741 --- /dev/null +++ b/desktop/src/features/agents/ui/agentSessionTranscript.ts @@ -0,0 +1,327 @@ +import type { + ObserverEvent, + PromptSection, + ToolStatus, + TranscriptItem, +} from "./agentSessionTypes"; +import { + findSproutToolName, + isGenericToolTitle, + normalizeToolStatus, +} from "./agentSessionToolCatalog"; +import { asRecord, asString } from "./agentSessionUtils"; +import { + describeTurnStarted, + describeSessionResolved, + extractBlockText, + extractContentText, + extractPromptText, + extractToolArgs, + extractToolIdentity, + extractToolResult, + parsePromptText, +} from "./agentSessionTranscriptHelpers"; + +export { describeRawEvent } from "./agentSessionTranscriptHelpers"; + +export function buildTranscript(events: ObserverEvent[]): TranscriptItem[] { + const items: TranscriptItem[] = []; + const itemsById = new Map(); + + // Maps a logical message ID (e.g. `assistant:msg1`) to the *actual* key + // currently being appended to. When a non-message item interleaves, we + // seal the current key so subsequent chunks create a fresh entry. + const activeMessageKey = new Map(); + const sealedKeys = new Set(); + let continuationSeq = 0; + + /** Seal every currently-open message so the next chunk starts a new entry. */ + function sealOpenMessages() { + for (const [, currentKey] of activeMessageKey) { + if (!sealedKeys.has(currentKey)) { + sealedKeys.add(currentKey); + } + } + } + + function upsertMessage( + id: string, + role: "assistant" | "user", + title: string, + text: string, + timestamp: string, + ) { + const currentKey = activeMessageKey.get(id); + + // If there is an active (non-sealed) key, append to it. + if (currentKey && !sealedKeys.has(currentKey)) { + const existing = itemsById.get(currentKey); + if (existing?.type === "message") { + existing.text += text; + return; + } + } + + // Otherwise create a new entry (either first time, or continuation). + continuationSeq += 1; + const newKey = currentKey ? `${id}:c${continuationSeq}` : id; + const item: TranscriptItem = { + id: newKey, + type: "message", + role, + title, + text, + timestamp, + }; + items.push(item); + itemsById.set(newKey, item); + activeMessageKey.set(id, newKey); + } + + function upsertTextItem( + id: string, + type: "thought" | "lifecycle", + title: string, + text: string, + timestamp: string, + ) { + const existing = itemsById.get(id); + if (existing && existing.type === type) { + existing.text += text; + return; + } + sealOpenMessages(); + const item: TranscriptItem = { id, type, title, text, timestamp }; + items.push(item); + itemsById.set(id, item); + } + + function upsertMetadata( + id: string, + title: string, + sections: PromptSection[], + timestamp: string, + ) { + const existing = itemsById.get(id); + if (existing?.type === "metadata") { + existing.sections = sections; + return; + } + sealOpenMessages(); + const item: TranscriptItem = { + id, + type: "metadata", + title, + sections, + timestamp, + }; + items.push(item); + itemsById.set(id, item); + } + + function upsertTool( + id: string, + title: string, + toolName: string, + sproutToolName: string | null, + status: ToolStatus, + args: Record, + result: string, + isError: boolean, + timestamp: string, + ) { + const existing = itemsById.get(id); + const canonicalSproutToolName = + sproutToolName ?? findSproutToolName(toolName, true); + if (existing?.type === "tool") { + if (!isGenericToolTitle(title)) { + existing.title = title; + } + if (canonicalSproutToolName) { + existing.sproutToolName = canonicalSproutToolName; + existing.toolName = canonicalSproutToolName; + } else if (!existing.sproutToolName && !isGenericToolTitle(toolName)) { + existing.toolName = toolName; + } + existing.status = status; + existing.args = Object.keys(args).length > 0 ? args : existing.args; + if (result) existing.result = result; + existing.isError = isError || existing.isError; + return; + } + sealOpenMessages(); + const item: TranscriptItem = { + id, + type: "tool", + title, + toolName: canonicalSproutToolName ?? toolName, + sproutToolName: canonicalSproutToolName, + status, + args, + result, + isError, + timestamp, + }; + items.push(item); + itemsById.set(id, item); + } + + for (const event of events) { + if (event.kind === "turn_started") { + upsertTextItem( + `turn:${event.turnId ?? event.seq}`, + "lifecycle", + "Turn started", + describeTurnStarted(event.payload), + event.timestamp, + ); + continue; + } + + if (event.kind === "session_resolved") { + upsertTextItem( + `session:${event.turnId ?? event.seq}`, + "lifecycle", + "Session ready", + describeSessionResolved(event.payload), + event.timestamp, + ); + continue; + } + + if (event.kind === "acp_parse_error") { + upsertTextItem( + `parse-error:${event.seq}`, + "lifecycle", + "Wire parse error", + extractBlockText(event.payload), + event.timestamp, + ); + continue; + } + + if (event.kind !== "acp_read" && event.kind !== "acp_write") { + continue; + } + + const payload = asRecord(event.payload); + const method = asString(payload.method); + + if (event.kind === "acp_write" && method === "session/prompt") { + const promptText = extractPromptText(payload); + if (promptText) { + const parsedPrompt = parsePromptText(promptText); + if (parsedPrompt.userText) { + upsertMessage( + `prompt:${event.turnId ?? event.seq}`, + "user", + parsedPrompt.userTitle, + parsedPrompt.userText, + event.timestamp, + ); + } + if (parsedPrompt.sections.length > 0) { + upsertMetadata( + `prompt-context:${event.turnId ?? event.seq}`, + "Prompt context", + parsedPrompt.sections, + event.timestamp, + ); + } + } + continue; + } + + if (event.kind !== "acp_read" || method !== "session/update") { + continue; + } + + const params = asRecord(payload.params); + const update = asRecord(params.update); + const updateType = asString(update.sessionUpdate) ?? "unknown"; + const turnKey = event.turnId ?? event.sessionId ?? "unknown"; + const messageId = asString(update.messageId); + + if (updateType === "agent_message_chunk") { + upsertMessage( + `assistant:${messageId ?? turnKey}`, + "assistant", + "Assistant", + extractContentText(update.content), + event.timestamp, + ); + continue; + } + + if (updateType === "user_message_chunk") { + upsertMessage( + `user:${messageId ?? turnKey}`, + "user", + "User", + extractContentText(update.content), + event.timestamp, + ); + continue; + } + + if (updateType === "agent_thought_chunk") { + upsertTextItem( + `thinking:${messageId ?? turnKey}`, + "thought", + "Thinking", + extractContentText(update.content), + event.timestamp, + ); + continue; + } + + if (updateType === "tool_call") { + const toolId = asString(update.toolCallId) ?? `tool:${event.seq}`; + const identity = extractToolIdentity(update); + upsertTool( + `tool:${toolId}`, + identity.title, + identity.toolName, + identity.sproutToolName, + normalizeToolStatus(asString(update.status) ?? "executing"), + extractToolArgs(update), + extractToolResult(update), + false, + event.timestamp, + ); + continue; + } + + if (updateType === "tool_call_update") { + const toolId = asString(update.toolCallId) ?? `tool:${event.seq}`; + const status = normalizeToolStatus( + asString(update.status) ?? "completed", + ); + const identity = extractToolIdentity(update); + upsertTool( + `tool:${toolId}`, + identity.title, + identity.toolName, + identity.sproutToolName, + status, + extractToolArgs(update), + extractToolResult(update), + status === "failed", + event.timestamp, + ); + continue; + } + + if (updateType === "plan") { + upsertTextItem( + `plan:${turnKey}`, + "thought", + "Plan", + extractContentText(update.content) || JSON.stringify(update, null, 2), + event.timestamp, + ); + } + } + + return items; +} diff --git a/desktop/src/features/agents/ui/agentSessionTranscriptHelpers.ts b/desktop/src/features/agents/ui/agentSessionTranscriptHelpers.ts new file mode 100644 index 0000000000..72c3c4a80e --- /dev/null +++ b/desktop/src/features/agents/ui/agentSessionTranscriptHelpers.ts @@ -0,0 +1,219 @@ +import type { ObserverEvent, PromptSection } from "./agentSessionTypes"; +import { + findSproutToolName, + isGenericToolTitle, + normalizeToolName, +} from "./agentSessionToolCatalog"; +import { asRecord, asString, shorten, titleCase } from "./agentSessionUtils"; + +export function extractPromptText(payload: Record): string { + const params = asRecord(payload.params); + const prompt = params.prompt; + if (!Array.isArray(prompt)) return ""; + return prompt.map(extractBlockText).filter(Boolean).join("\n"); +} + +export function parsePromptText(text: string): { + sections: PromptSection[]; + userText: string; + userTitle: string; +} { + const sections = parsePromptSections(text); + if (sections.length === 0) { + return { sections: [], userText: text.trim(), userTitle: "Prompt" }; + } + + const eventSection = sections.find((section) => + section.title.toLowerCase().startsWith("sprout event"), + ); + const eventContent = eventSection + ? extractEventContent(eventSection.body) + : ""; + const eventKind = eventSection?.title.split(":").slice(1).join(":").trim(); + + return { + sections, + userText: eventContent, + userTitle: eventKind ? titleCase(eventKind) : "Sprout event", + }; +} + +function parsePromptSections(text: string): PromptSection[] { + const sections: PromptSection[] = []; + let current: PromptSection | null = null; + const preamble: string[] = []; + + for (const line of text.split(/\r?\n/)) { + const header = line.match(/^\[([^\]]+)]\s*$/); + if (header) { + if (current) { + sections.push({ + title: current.title, + body: current.body.trim(), + }); + } else if (preamble.join("\n").trim()) { + sections.push({ title: "Prompt", body: preamble.join("\n").trim() }); + } + current = { title: header[1], body: "" }; + continue; + } + + if (current) { + current.body += current.body ? `\n${line}` : line; + } else { + preamble.push(line); + } + } + + if (current) { + sections.push({ title: current.title, body: current.body.trim() }); + } else if (preamble.join("\n").trim()) { + sections.push({ title: "Prompt", body: preamble.join("\n").trim() }); + } + + return sections; +} + +function extractEventContent(body: string): string { + const contentMatch = body.match(/^Content:\s*(.*)$/m); + return contentMatch?.[1]?.trim() ?? ""; +} + +export function extractContentText(value: unknown): string { + if (typeof value === "string") return value; + if (Array.isArray(value)) return value.map(extractBlockText).join("\n"); + return extractBlockText(value); +} + +export function extractBlockText(value: unknown): string { + if (typeof value === "string") return value; + if (Array.isArray(value)) return value.map(extractBlockText).join("\n"); + const record = asRecord(value); + const nestedContent = record.content; + const rawOutput = record.rawOutput; + const nestedText = + nestedContent && typeof nestedContent === "object" + ? extractBlockText(nestedContent) + : ""; + const rawOutputText = + rawOutput === undefined || rawOutput === null + ? "" + : typeof rawOutput === "string" + ? rawOutput + : JSON.stringify(rawOutput, null, 2); + const directText = asString(record.text) ?? asString(record.content); + return directText || nestedText || rawOutputText || ""; +} + +export function extractToolArgs( + update: Record, +): Record { + const candidates = [ + update.args, + update.arguments, + update.input, + update.rawInput, + ]; + for (const candidate of candidates) { + if ( + candidate && + typeof candidate === "object" && + !Array.isArray(candidate) + ) { + return candidate as Record; + } + } + return {}; +} + +export function extractToolIdentity(update: Record): { + title: string; + toolName: string; + sproutToolName: string | null; +} { + const candidates = collectToolNameCandidates(update); + const knownName = + candidates + .map((candidate) => findSproutToolName(candidate, true)) + .find((candidate): candidate is string => Boolean(candidate)) ?? + findSproutToolName(JSON.stringify(update), false); + const firstSpecific = candidates.find( + (candidate) => !isGenericToolTitle(candidate), + ); + const title = + asString(update.title) ?? knownName ?? firstSpecific ?? "Tool call"; + return { + title, + toolName: knownName ?? normalizeToolName(firstSpecific ?? title), + sproutToolName: knownName, + }; +} + +function collectToolNameCandidates(update: Record): string[] { + const args = extractToolArgs(update); + const tool = asRecord(update.tool); + const input = asRecord(update.input); + const rawInput = asRecord(update.rawInput); + const candidates = [ + update.toolName, + update.tool_name, + update.name, + update.title, + update.kind, + tool.name, + tool.toolName, + args.toolName, + args.tool_name, + args.name, + args.method, + input.toolName, + input.tool_name, + input.name, + rawInput.toolName, + rawInput.tool_name, + rawInput.name, + ]; + + return candidates.flatMap((candidate) => { + const value = asString(candidate); + return value ? [value] : []; + }); +} + +export function extractToolResult(update: Record): string { + const contentText = extractContentText(update.content); + if (contentText) return contentText; + return extractBlockText(update.rawOutput); +} + +export function describeTurnStarted(payload: unknown): string { + const record = asRecord(payload); + const ids = Array.isArray(record.triggeringEventIds) + ? record.triggeringEventIds.filter( + (id): id is string => typeof id === "string", + ) + : []; + return ids.length > 0 + ? `Triggered by ${ids.map(shorten).join(", ")}.` + : "Heartbeat or internal turn."; +} + +export function describeSessionResolved(payload: unknown): string { + const record = asRecord(payload); + const sessionId = asString(record.sessionId); + const isNewSession = record.isNewSession === true; + if (!sessionId) { + return "Using existing ACP session."; + } + return `${isNewSession ? "Created" : "Using"} session ${shorten(sessionId)}.`; +} + +export function describeRawEvent(event: ObserverEvent): string { + const payload = asRecord(event.payload); + const method = asString(payload.method); + if (method === "session/update") { + const update = asRecord(asRecord(payload.params).update); + return asString(update.sessionUpdate) ?? method; + } + return method ?? event.kind; +} diff --git a/desktop/src/features/agents/ui/agentSessionTypes.ts b/desktop/src/features/agents/ui/agentSessionTypes.ts new file mode 100644 index 0000000000..b4b8c9f2ae --- /dev/null +++ b/desktop/src/features/agents/ui/agentSessionTypes.ts @@ -0,0 +1,75 @@ +import type { LucideIcon } from "lucide-react"; + +export type ObserverEvent = { + seq: number; + timestamp: string; + kind: string; + agentIndex: number | null; + channelId: string | null; + sessionId: string | null; + turnId: string | null; + payload: unknown; +}; + +export type ConnectionState = + | "idle" + | "connecting" + | "open" + | "closed" + | "error"; + +export type ToolStatus = "executing" | "completed" | "failed" | "pending"; + +export type TranscriptItem = + | { + id: string; + type: "message"; + role: "assistant" | "user"; + title: string; + text: string; + timestamp: string; + } + | { + id: string; + type: "thought"; + title: string; + text: string; + timestamp: string; + } + | { + id: string; + type: "lifecycle"; + title: string; + text: string; + timestamp: string; + } + | { + id: string; + type: "metadata"; + title: string; + sections: PromptSection[]; + timestamp: string; + } + | { + id: string; + type: "tool"; + title: string; + toolName: string; + sproutToolName: string | null; + status: ToolStatus; + args: Record; + result: string; + isError: boolean; + timestamp: string; + }; + +export type PromptSection = { + title: string; + body: string; +}; + +export type SproutToolInfo = { + icon: LucideIcon; + label: string; + tone: "read" | "write" | "admin"; +}; diff --git a/desktop/src/features/agents/ui/agentSessionUtils.ts b/desktop/src/features/agents/ui/agentSessionUtils.ts new file mode 100644 index 0000000000..5efa9d0013 --- /dev/null +++ b/desktop/src/features/agents/ui/agentSessionUtils.ts @@ -0,0 +1,81 @@ +export function getToolString( + record: Record, + keys: string[], +): string | null { + for (const key of keys) { + const value = record[key]; + if (typeof value === "string" && value.trim().length > 0) { + return value.trim(); + } + } + return null; +} + +export function getToolStringList( + record: Record, + keys: string[], +): string[] { + for (const key of keys) { + const value = record[key]; + if (typeof value === "string" && value.trim().length > 0) { + return [value.trim()]; + } + if (Array.isArray(value)) { + return value.filter( + (item): item is string => + typeof item === "string" && item.trim().length > 0, + ); + } + } + return []; +} + +export function getResultArray( + resultValue: unknown, + resultRecord: Record, + key: string, +) { + if (Array.isArray(resultValue)) return resultValue; + const value = resultRecord[key]; + return Array.isArray(value) ? value : null; +} + +export function formatCodeValue(value: string): string { + const trimmed = value.trim(); + if (!trimmed) return value; + try { + return JSON.stringify(JSON.parse(trimmed), null, 2); + } catch { + return value; + } +} + +export function titleCase(value: string): string { + return value + .replace(/[_-]+/g, " ") + .replace(/\s+/g, " ") + .trim() + .replace(/\b\w/g, (match) => match.toUpperCase()); +} + +export function asRecord(value: unknown): Record { + return value && typeof value === "object" + ? (value as Record) + : {}; +} + +export function asString(value: unknown): string | null { + return typeof value === "string" ? value : null; +} + +export function shorten(value: string) { + return value.length > 14 + ? `${value.slice(0, 8)}...${value.slice(-4)}` + : value; +} + +export function shortenMiddle(value: string, maxLength: number) { + if (value.length <= maxLength) return value; + const edgeLength = Math.max(4, Math.floor((maxLength - 3) / 2)); + return `${value.slice(0, edgeLength)}...${value.slice(-edgeLength)}`; +} diff --git a/desktop/src/features/agents/ui/useObserverEvents.ts b/desktop/src/features/agents/ui/useObserverEvents.ts new file mode 100644 index 0000000000..105b22b11a --- /dev/null +++ b/desktop/src/features/agents/ui/useObserverEvents.ts @@ -0,0 +1,65 @@ +import * as React from "react"; + +import type { ConnectionState, ObserverEvent } from "./agentSessionTypes"; + +const MAX_OBSERVER_EVENTS = 800; + +export function useObserverEvents( + observerUrl: string | null, + enabled: boolean, +) { + const [events, setEvents] = React.useState([]); + const [connectionState, setConnectionState] = + React.useState("idle"); + const [errorMessage, setErrorMessage] = React.useState(null); + + React.useEffect(() => { + setEvents([]); + setErrorMessage(null); + + if (!observerUrl || !enabled) { + setConnectionState("idle"); + return; + } + + setConnectionState("connecting"); + const source = new EventSource(observerUrl); + + source.onopen = () => { + setConnectionState("open"); + setErrorMessage(null); + }; + + source.onmessage = (event) => { + try { + const parsed = JSON.parse(event.data) as ObserverEvent; + setEvents((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."); + }; + + return () => { + source.close(); + setConnectionState("closed"); + }; + }, [enabled, observerUrl]); + + return { connectionState, errorMessage, events }; +} diff --git a/desktop/src/features/channels/ui/AgentSessionThreadPanel.tsx b/desktop/src/features/channels/ui/AgentSessionThreadPanel.tsx new file mode 100644 index 0000000000..2ea6eec669 --- /dev/null +++ b/desktop/src/features/channels/ui/AgentSessionThreadPanel.tsx @@ -0,0 +1,105 @@ +import type * as React from "react"; +import { Activity, Bot, CircleDot, X } from "lucide-react"; + +import { ManagedAgentSessionPanel } from "@/features/agents/ui/ManagedAgentSessionPanel"; +import type { Channel, ManagedAgent } from "@/shared/api/types"; +import { useStickToBottom } from "@/shared/hooks/useStickToBottom"; +import { Badge } from "@/shared/ui/badge"; +import { Button } from "@/shared/ui/button"; + +type AgentSessionThreadPanelProps = { + agent: ManagedAgent; + canResetWidth: boolean; + channel: Channel; + onClose: () => void; + onResetWidth: () => void; + onResizeStart: (event: React.PointerEvent) => void; + widthPx: number; +}; + +export function AgentSessionThreadPanel({ + agent, + canResetWidth, + channel, + onClose, + onResetWidth, + onResizeStart, + widthPx, +}: AgentSessionThreadPanelProps) { + const isLive = agent.status === "running" && Boolean(agent.observerUrl); + const { ref: scrollRef, onScroll } = useStickToBottom(); + + return ( + + ); +} diff --git a/desktop/src/features/channels/ui/BotActivityBar.tsx b/desktop/src/features/channels/ui/BotActivityBar.tsx new file mode 100644 index 0000000000..dcd8237b07 --- /dev/null +++ b/desktop/src/features/channels/ui/BotActivityBar.tsx @@ -0,0 +1,76 @@ +import { Bot, Loader2 } from "lucide-react"; + +import type { ManagedAgent } from "@/shared/api/types"; +import { cn } from "@/shared/lib/cn"; +import { Tooltip, TooltipContent, TooltipTrigger } from "@/shared/ui/tooltip"; + +type BotActivityBarProps = { + agents: ManagedAgent[]; + onOpenAgentSession: (pubkey: string) => void; + openAgentSessionPubkey: string | null; + typingBotPubkeys: string[]; +}; + +/** + * Compact right-aligned row of clickable bot pills. + * Only renders pills for bots that are currently typing (actively working). + */ +export function BotActivityBar({ + agents, + onOpenAgentSession, + openAgentSessionPubkey, + typingBotPubkeys, +}: BotActivityBarProps) { + if (typingBotPubkeys.length === 0) { + return null; + } + + const typingSet = new Set( + typingBotPubkeys.map((pubkey) => pubkey.toLowerCase()), + ); + + const typingAgents = agents.filter((agent) => + typingSet.has(agent.pubkey.toLowerCase()), + ); + + if (typingAgents.length === 0) { + return null; + } + + return ( +
+ {typingAgents.map((agent) => { + const isSelected = + openAgentSessionPubkey?.toLowerCase() === agent.pubkey.toLowerCase(); + return ( + + + + + + {agent.name} is working — click to view activity + + + ); + })} +
+ ); +} diff --git a/desktop/src/features/channels/ui/ChannelHeaderStatusBadge.tsx b/desktop/src/features/channels/ui/ChannelHeaderStatusBadge.tsx new file mode 100644 index 0000000000..a1d7d7551d --- /dev/null +++ b/desktop/src/features/channels/ui/ChannelHeaderStatusBadge.tsx @@ -0,0 +1,38 @@ +import type { EphemeralChannelDisplay } from "@/features/channels/lib/ephemeralChannel"; +import { EphemeralChannelBadge } from "@/features/channels/ui/EphemeralChannelBadge"; +import { PresenceBadge } from "@/features/presence/ui/PresenceBadge"; +import type { Channel, PresenceStatus } from "@/shared/api/types"; + +type ChannelHeaderStatusBadgeProps = { + channelType?: Channel["channelType"]; + ephemeralDisplay: EphemeralChannelDisplay | null; + presenceStatus: PresenceStatus | null; +}; + +export function ChannelHeaderStatusBadge({ + channelType, + ephemeralDisplay, + presenceStatus, +}: ChannelHeaderStatusBadgeProps) { + const ephemeralBadge = ephemeralDisplay ? ( + + ) : null; + + if (channelType === "dm" && presenceStatus) { + return ( + <> + + {ephemeralBadge} + + ); + } + + return ephemeralBadge; +} diff --git a/desktop/src/features/channels/ui/ChannelPane.tsx b/desktop/src/features/channels/ui/ChannelPane.tsx index 0b9ea96902..05ba65ad5b 100644 --- a/desktop/src/features/channels/ui/ChannelPane.tsx +++ b/desktop/src/features/channels/ui/ChannelPane.tsx @@ -5,11 +5,13 @@ import { MessageThreadPanel } from "@/features/messages/ui/MessageThreadPanel"; import { MessageTimeline } from "@/features/messages/ui/MessageTimeline"; import { TypingIndicatorRow } from "@/features/messages/ui/TypingIndicatorRow"; import { ChannelFindBar } from "@/features/search/ui/ChannelFindBar"; +import { AgentSessionThreadPanel } from "@/features/channels/ui/AgentSessionThreadPanel"; +import { BotActivityBar } from "@/features/channels/ui/BotActivityBar"; import type { useChannelFind } from "@/features/search/useChannelFind"; import type { MainTimelineEntry } from "@/features/messages/lib/threadPanel"; import type { TimelineMessage } from "@/features/messages/types"; import type { UserProfileLookup } from "@/features/profile/lib/identity"; -import type { Channel } from "@/shared/api/types"; +import type { Channel, ManagedAgent } from "@/shared/api/types"; const THREAD_PANEL_DEFAULT_WIDTH_PX = 380; const THREAD_PANEL_MIN_WIDTH_PX = 320; @@ -47,6 +49,8 @@ function getInitialThreadPanelWidth(): number { type ChannelPaneProps = { activeChannel: Channel | null; + agentSessionAgents: ManagedAgent[]; + botTypingPubkeys: string[]; channelFind: ReturnType; currentPubkey?: string; editTarget?: { @@ -62,11 +66,13 @@ type ChannelPaneProps = { messages: TimelineMessage[]; onCancelEdit?: () => void; onCancelThreadReply: () => void; + onCloseAgentSession: () => void; onCloseThread: () => void; onDelete?: (message: TimelineMessage) => void; onEdit?: (message: TimelineMessage) => void; onEditSave?: (content: string) => Promise; onExpandThreadReplies: (message: TimelineMessage) => void; + onOpenAgentSession: (pubkey: string) => void; onOpenThread: (message: TimelineMessage) => void; onSelectThreadReplyTarget: (message: TimelineMessage) => void; onSendMessage: ( @@ -90,6 +96,7 @@ type ChannelPaneProps = { personaLookup?: Map; profiles?: UserProfileLookup; openThreadHeadId: string | null; + openAgentSessionPubkey: string | null; threadHeadMessage: TimelineMessage | null; threadMessages: MainTimelineEntry[]; threadTypingPubkeys: string[]; @@ -102,6 +109,8 @@ type ChannelPaneProps = { export const ChannelPane = React.memo(function ChannelPane({ activeChannel, + agentSessionAgents, + botTypingPubkeys, channelFind, currentPubkey, editTarget = null, @@ -113,11 +122,13 @@ export const ChannelPane = React.memo(function ChannelPane({ messages, onCancelEdit, onCancelThreadReply, + onCloseAgentSession, onCloseThread, onDelete, onEdit, onEditSave, onExpandThreadReplies, + onOpenAgentSession, onOpenThread, onSelectThreadReplyTarget, onSendMessage, @@ -128,6 +139,7 @@ export const ChannelPane = React.memo(function ChannelPane({ personaLookup, profiles, openThreadHeadId, + openAgentSessionPubkey, targetMessageId, threadHeadMessage, threadMessages, @@ -199,6 +211,16 @@ export const ChannelPane = React.memo(function ChannelPane({ activeChannel.channelType === "forum" || isSending; + const selectedAgent = React.useMemo( + () => + openAgentSessionPubkey + ? (agentSessionAgents.find( + (agent) => agent.pubkey === openAgentSessionPubkey, + ) ?? null) + : null, + [agentSessionAgents, openAgentSessionPubkey], + ); + return (
@@ -267,12 +289,22 @@ export const ChannelPane = React.memo(function ChannelPane({ : "Select a channel" } /> - +
+ +
+ +
+
{threadHeadMessage ? ( @@ -303,6 +335,16 @@ export const ChannelPane = React.memo(function ChannelPane({ threadReplies={threadMessages} threadTypingPubkeys={threadTypingPubkeys} /> + ) : activeChannel && selectedAgent ? ( + ) : null}
); diff --git a/desktop/src/features/channels/ui/ChannelScreen.tsx b/desktop/src/features/channels/ui/ChannelScreen.tsx index 703732b6df..0114416155 100644 --- a/desktop/src/features/channels/ui/ChannelScreen.tsx +++ b/desktop/src/features/channels/ui/ChannelScreen.tsx @@ -1,12 +1,14 @@ import * as React from "react"; import { useAppShell } from "@/app/AppShellContext"; -import { ChatHeader } from "@/features/chat/ui/ChatHeader"; import { useActiveChannelHeader } from "@/features/channels/useActiveChannelHeader"; import { useChannelPaneHandlers } from "@/features/channels/useChannelPaneHandlers"; import { useChannelMembersQuery } from "@/features/channels/hooks"; -import { getChannelDescription } from "@/features/channels/lib/channelDescription"; -import { ChannelMembersBar } from "@/features/channels/ui/ChannelMembersBar"; -import { EphemeralChannelBadge } from "@/features/channels/ui/EphemeralChannelBadge"; +import { ChannelScreenEmptyState } from "@/features/channels/ui/ChannelScreenEmptyState"; +import { ChannelScreenHeader } from "@/features/channels/ui/ChannelScreenHeader"; +import { + ChannelPane, + ForumView, +} from "@/features/channels/ui/ChannelScreenLazyViews"; import { MembersSidebar } from "@/features/channels/ui/MembersSidebar"; import { useManagedAgentsQuery, @@ -29,7 +31,6 @@ import { buildThreadPanelData } from "@/features/messages/lib/threadPanel"; import { useFetchOlderMessages } from "@/features/messages/useFetchOlderMessages"; import { useLoadMissingAncestors } from "@/features/messages/useLoadMissingAncestors"; import { useChannelTyping } from "@/features/messages/useChannelTyping"; -import { PresenceBadge } from "@/features/presence/ui/PresenceBadge"; import { useUsersBatchQuery } from "@/features/profile/hooks"; import { mergeCurrentProfileIntoLookup } from "@/features/profile/lib/identity"; import type { @@ -40,16 +41,8 @@ import type { } from "@/shared/api/types"; import { useChannelFind } from "@/features/search/useChannelFind"; import { ViewLoadingFallback } from "@/shared/ui/ViewLoadingFallback"; - -const ChannelPane = React.lazy(async () => { - const module = await import("@/features/channels/ui/ChannelPane"); - return { default: module.ChannelPane }; -}); - -const ForumView = React.lazy(async () => { - const module = await import("@/features/forum/ui/ForumView"); - return { default: module.ForumView }; -}); +import { AgentSessionProvider } from "@/shared/context/AgentSessionContext"; +import { useChannelAgentSessions } from "./useChannelAgentSessions"; type ChannelScreenProps = { activeChannel: Channel | null; @@ -172,6 +165,19 @@ export function ChannelScreen({ enabled: messageProfilePubkeys.length > 0, }); const managedAgentsQuery = useManagedAgentsQuery(); + const { humanTypingPubkeys, botTypingPubkeys } = React.useMemo(() => { + const agentSet = new Set( + (managedAgentsQuery.data ?? []).map((a) => a.pubkey.toLowerCase()), + ); + return { + humanTypingPubkeys: mainTypingPubkeys.filter( + (pk) => !agentSet.has(pk.toLowerCase()), + ), + botTypingPubkeys: mainTypingPubkeys.filter((pk) => + agentSet.has(pk.toLowerCase()), + ), + }; + }, [mainTypingPubkeys, managedAgentsQuery.data]); const messageProfiles = React.useMemo(() => { const base = mergeCurrentProfileIntoLookup( @@ -322,8 +328,26 @@ export function ChannelScreen({ () => (canReact ? handleToggleReaction : undefined), [canReact, handleToggleReaction], ); + const { + channelAgentSessionAgents, + closeAgentSession: handleCloseAgentSession, + openAgentSession: handleOpenAgentSession, + openAgentSessionPubkey, + openThreadAndCloseAgentSession: handleOpenThreadAndCloseAgentSession, + } = useChannelAgentSessions({ + activeChannel, + activeChannelId, + channelMembers, + handleOpenThread, + managedAgents: managedAgentsQuery.data ?? [], + setExpandedThreadReplyIds, + setOpenThreadHeadId, + setThreadReplyTargetId, + setThreadScrollTargetId, + targetMessageId, + timelineMessages, + }); - const channelDescription = getChannelDescription(activeChannel); const shouldLoadTimeline = activeChannel !== null && activeChannel.channelType !== "forum"; const isTimelineLoading = @@ -336,9 +360,10 @@ export function ChannelScreen({ setExpandedThreadReplyIds(new Set()); setThreadScrollTargetId(null); setThreadReplyTargetId(null); + handleCloseAgentSession(); setEditTargetId(null); }, - [], + [handleCloseAgentSession], ); const handleThreadScrollTargetResolved = React.useCallback(() => { setThreadScrollTargetId(null); @@ -377,45 +402,16 @@ export function ChannelScreen({ useLoadMissingAncestors(activeChannel, resolvedMessages); - const activeChannelEphemeralBadge = activeChannelEphemeralDisplay ? ( - - ) : null; - - const headerStatusBadge = - activeChannel?.channelType === "dm" && activeDmPresenceStatus ? ( - <> - - {activeChannelEphemeralBadge} - - ) : ( - activeChannelEphemeralBadge - ); - return ( - <> - setIsMembersSidebarOpen((prev) => !prev)} - /> - ) : null - } - channelType={activeChannel?.channelType} - visibility={activeChannel?.visibility} - description={channelDescription} - statusBadge={headerStatusBadge} - title={activeChannelTitle} + + setIsMembersSidebarOpen((prev) => !prev)} />
@@ -435,6 +431,8 @@ export function ChannelScreen({ }> ) ) : ( -
-

- Select a channel to view messages. -

-
+ )}
@@ -493,7 +490,8 @@ export function ChannelScreen({ currentPubkey={currentPubkey} open={isMembersSidebarOpen} onOpenChange={setIsMembersSidebarOpen} + onViewActivity={handleOpenAgentSession} /> - + ); } diff --git a/desktop/src/features/channels/ui/ChannelScreenEmptyState.tsx b/desktop/src/features/channels/ui/ChannelScreenEmptyState.tsx new file mode 100644 index 0000000000..98fbeb8a00 --- /dev/null +++ b/desktop/src/features/channels/ui/ChannelScreenEmptyState.tsx @@ -0,0 +1,9 @@ +export function ChannelScreenEmptyState() { + return ( +
+

+ Select a channel to view messages. +

+
+ ); +} diff --git a/desktop/src/features/channels/ui/ChannelScreenHeader.tsx b/desktop/src/features/channels/ui/ChannelScreenHeader.tsx new file mode 100644 index 0000000000..e8062b6a96 --- /dev/null +++ b/desktop/src/features/channels/ui/ChannelScreenHeader.tsx @@ -0,0 +1,52 @@ +import { ChatHeader } from "@/features/chat/ui/ChatHeader"; +import type { EphemeralChannelDisplay } from "@/features/channels/lib/ephemeralChannel"; +import { getChannelDescription } from "@/features/channels/lib/channelDescription"; +import { ChannelHeaderStatusBadge } from "@/features/channels/ui/ChannelHeaderStatusBadge"; +import { ChannelMembersBar } from "@/features/channels/ui/ChannelMembersBar"; +import type { Channel, PresenceStatus } from "@/shared/api/types"; + +type ChannelScreenHeaderProps = { + activeChannel: Channel | null; + activeChannelEphemeralDisplay: EphemeralChannelDisplay | null; + activeChannelTitle: string; + activeDmPresenceStatus: PresenceStatus | null; + currentPubkey?: string; + onManageChannel: () => void; + onToggleMembers: () => void; +}; + +export function ChannelScreenHeader({ + activeChannel, + activeChannelEphemeralDisplay, + activeChannelTitle, + activeDmPresenceStatus, + currentPubkey, + onManageChannel, + onToggleMembers, +}: ChannelScreenHeaderProps) { + return ( + + ) : null + } + channelType={activeChannel?.channelType} + description={getChannelDescription(activeChannel)} + statusBadge={ + + } + title={activeChannelTitle} + visibility={activeChannel?.visibility} + /> + ); +} diff --git a/desktop/src/features/channels/ui/ChannelScreenLazyViews.ts b/desktop/src/features/channels/ui/ChannelScreenLazyViews.ts new file mode 100644 index 0000000000..4e4fc0f2af --- /dev/null +++ b/desktop/src/features/channels/ui/ChannelScreenLazyViews.ts @@ -0,0 +1,11 @@ +import * as React from "react"; + +export const ChannelPane = React.lazy(async () => { + const module = await import("@/features/channels/ui/ChannelPane"); + return { default: module.ChannelPane }; +}); + +export const ForumView = React.lazy(async () => { + const module = await import("@/features/forum/ui/ForumView"); + return { default: module.ForumView }; +}); diff --git a/desktop/src/features/channels/ui/MembersSidebar.tsx b/desktop/src/features/channels/ui/MembersSidebar.tsx index e580d07af2..6156462dc6 100644 --- a/desktop/src/features/channels/ui/MembersSidebar.tsx +++ b/desktop/src/features/channels/ui/MembersSidebar.tsx @@ -29,6 +29,7 @@ type MembersSidebarProps = { currentPubkey?: string; open: boolean; onOpenChange: (open: boolean) => void; + onViewActivity?: (pubkey: string) => void; }; export function MembersSidebar({ @@ -36,6 +37,7 @@ export function MembersSidebar({ currentPubkey, open, onOpenChange, + onViewActivity, }: MembersSidebarProps) { const channelId = channel?.id ?? null; const queryClient = useQueryClient(); @@ -168,6 +170,14 @@ export function MembersSidebar({ void handleAgentLifecycleAction(agent); }} onRemoveMember={handleRemoveMember} + onViewActivity={ + onViewActivity + ? (pubkey: string) => { + onOpenChange(false); + onViewActivity(pubkey); + } + : undefined + } presenceStatus={ memberPresenceQuery.data?.[member.pubkey.toLowerCase()] ?? null } diff --git a/desktop/src/features/channels/ui/MembersSidebarMemberCard.tsx b/desktop/src/features/channels/ui/MembersSidebarMemberCard.tsx index 10631be008..2828bf611d 100644 --- a/desktop/src/features/channels/ui/MembersSidebarMemberCard.tsx +++ b/desktop/src/features/channels/ui/MembersSidebarMemberCard.tsx @@ -1,4 +1,5 @@ import { + Activity, Ellipsis, Play, RotateCcw, @@ -43,6 +44,7 @@ type MembersSidebarMemberCardProps = { onChangeRole: (member: ChannelMember, role: string) => void; onManagedAgentAction: (agent: ManagedAgent) => void; onRemoveMember: (member: ChannelMember) => void; + onViewActivity?: (pubkey: string) => void; presenceStatus?: PresenceStatus | null; profileAvatarUrl?: string | null; }; @@ -80,13 +82,18 @@ export function MembersSidebarMemberCard({ onChangeRole, onManagedAgentAction, onRemoveMember, + onViewActivity, presenceStatus, profileAvatarUrl, }: MembersSidebarMemberCardProps) { const roleLabel = formatRoleLabel(member, memberIsBot); const disabled = isActionPending || isArchived; + const canViewActivity = + memberIsBot && + managedAgent?.backend.type === "local" && + Boolean(onViewActivity); const hasActions = memberIsBot - ? Boolean(managedAgent) || canRemoveMember + ? Boolean(managedAgent) || canRemoveMember || canViewActivity : canRemoveMember || canChangeRole; return ( @@ -135,6 +142,7 @@ export function MembersSidebarMemberCard({ ) : null} @@ -153,6 +162,7 @@ const PEOPLE_ROLES = ["admin", "member", "guest"] as const; function MemberActionsMenu({ canChangeRole, canRemoveMember, + canViewActivity, disabled, managedAgent, member, @@ -160,9 +170,11 @@ function MemberActionsMenu({ onChangeRole, onManagedAgentAction, onRemoveMember, + onViewActivity, }: { canChangeRole: boolean; canRemoveMember: boolean; + canViewActivity: boolean; disabled: boolean; managedAgent?: ManagedAgent; member: ChannelMember; @@ -170,6 +182,7 @@ function MemberActionsMenu({ onChangeRole: (member: ChannelMember, role: string) => void; onManagedAgentAction: (agent: ManagedAgent) => void; onRemoveMember: (member: ChannelMember) => void; + onViewActivity?: (pubkey: string) => void; }) { const showChangeRole = canChangeRole && !memberIsBot && member.role !== "owner"; @@ -189,8 +202,18 @@ function MemberActionsMenu({ align="end" onCloseAutoFocus={(event) => event.preventDefault()} > + {canViewActivity ? ( + onViewActivity?.(member.pubkey)} + > + + View activity + + ) : null} {memberIsBot && managedAgent ? ( <> + {canViewActivity ? : null} void; + managedAgents: ManagedAgent[]; + setExpandedThreadReplyIds: (value: Set) => void; + setOpenThreadHeadId: (value: string | null) => void; + setThreadReplyTargetId: (value: string | null) => void; + setThreadScrollTargetId: (value: string | null) => void; + targetMessageId: string | null; + timelineMessages: TimelineMessage[]; +}; + +export function useChannelAgentSessions({ + activeChannel, + activeChannelId, + channelMembers, + handleOpenThread, + managedAgents, + setExpandedThreadReplyIds, + setOpenThreadHeadId, + setThreadReplyTargetId, + setThreadScrollTargetId, + targetMessageId, + timelineMessages, +}: UseChannelAgentSessionsOptions) { + const [openAgentSessionPubkey, setOpenAgentSessionPubkey] = React.useState< + string | null + >(null); + const handledThreadTargetIdRef = React.useRef(null); + + const channelAgentSessionAgents = React.useMemo(() => { + if (!channelMembers) { + return []; + } + + const memberPubkeys = new Set( + channelMembers.map((member) => normalizePubkey(member.pubkey)), + ); + + return managedAgents.filter( + (agent) => + agent.backend.type === "local" && + memberPubkeys.has(normalizePubkey(agent.pubkey)), + ); + }, [channelMembers, managedAgents]); + + const closeAgentSession = React.useCallback(() => { + setOpenAgentSessionPubkey(null); + }, []); + + const openAgentSession = React.useCallback( + (pubkey: string) => { + setOpenThreadHeadId(null); + setExpandedThreadReplyIds(new Set()); + setThreadScrollTargetId(null); + setThreadReplyTargetId(null); + setOpenAgentSessionPubkey(pubkey); + }, + [ + setExpandedThreadReplyIds, + setOpenThreadHeadId, + setThreadReplyTargetId, + setThreadScrollTargetId, + ], + ); + + const selectAgentSession = React.useCallback((pubkey: string) => { + setOpenAgentSessionPubkey(pubkey); + }, []); + + const openThreadAndCloseAgentSession = React.useCallback( + (message: TimelineMessage) => { + setOpenAgentSessionPubkey(null); + handleOpenThread(message); + }, + [handleOpenThread], + ); + + React.useEffect(() => { + if (!targetMessageId) { + handledThreadTargetIdRef.current = null; + return; + } + + const targetKey = `${activeChannelId ?? "none"}:${targetMessageId}`; + if ( + handledThreadTargetIdRef.current !== null && + handledThreadTargetIdRef.current !== targetKey + ) { + handledThreadTargetIdRef.current = null; + } + + if ( + handledThreadTargetIdRef.current === targetKey || + !activeChannel || + activeChannel.channelType === "forum" + ) { + return; + } + + const targetMessage = + timelineMessages.find((message) => message.id === targetMessageId) ?? + null; + + if (!targetMessage?.parentId) { + return; + } + + const threadHeadId = targetMessage.rootId ?? targetMessage.parentId; + const messageById = new Map( + timelineMessages.map((message) => [message.id, message]), + ); + + if (!messageById.has(threadHeadId)) { + return; + } + + const expandedReplyIds = new Set(); + let ancestorId: string | null = targetMessage.parentId; + let guard = 0; + + while ( + ancestorId && + ancestorId !== threadHeadId && + guard < timelineMessages.length + ) { + expandedReplyIds.add(ancestorId); + ancestorId = messageById.get(ancestorId)?.parentId ?? null; + guard += 1; + } + + setOpenAgentSessionPubkey(null); + setOpenThreadHeadId(threadHeadId); + setThreadReplyTargetId(threadHeadId); + setThreadScrollTargetId(targetMessageId); + setExpandedThreadReplyIds(expandedReplyIds); + handledThreadTargetIdRef.current = targetKey; + }, [ + activeChannel, + activeChannelId, + setExpandedThreadReplyIds, + setOpenThreadHeadId, + setThreadReplyTargetId, + setThreadScrollTargetId, + targetMessageId, + timelineMessages, + ]); + + React.useEffect(() => { + if ( + openAgentSessionPubkey && + !channelAgentSessionAgents.some( + (agent) => agent.pubkey === openAgentSessionPubkey, + ) + ) { + setOpenAgentSessionPubkey(null); + } + }, [channelAgentSessionAgents, openAgentSessionPubkey]); + + return { + channelAgentSessionAgents, + closeAgentSession, + openAgentSession, + openAgentSessionPubkey, + openThreadAndCloseAgentSession, + selectAgentSession, + }; +} diff --git a/desktop/src/features/forum/ui/ForumComposer.tsx b/desktop/src/features/forum/ui/ForumComposer.tsx index 0dc540ba2e..c7e55f3979 100644 --- a/desktop/src/features/forum/ui/ForumComposer.tsx +++ b/desktop/src/features/forum/ui/ForumComposer.tsx @@ -35,7 +35,7 @@ type ForumComposerProps = { content: string, mentionPubkeys: string[], mediaTags?: string[][], - ) => void | Promise; + ) => undefined | Promise; /** When true, autocomplete renders below the input (for top-of-view composers). */ autocompleteBelow?: boolean; }; diff --git a/desktop/src/features/forum/ui/ForumThreadPanel.tsx b/desktop/src/features/forum/ui/ForumThreadPanel.tsx index 9985b09d35..9c858ec2ec 100644 --- a/desktop/src/features/forum/ui/ForumThreadPanel.tsx +++ b/desktop/src/features/forum/ui/ForumThreadPanel.tsx @@ -30,7 +30,7 @@ type ForumThreadPanelProps = { content: string, mentionPubkeys: string[], mediaTags?: string[][], - ) => void | Promise; + ) => undefined | Promise; onDeletePost?: (eventId: string) => void; onDeleteReply?: (eventId: string) => void; onTargetReached?: (eventId: string) => void; diff --git a/desktop/src/features/profile/ui/UserProfilePopover.tsx b/desktop/src/features/profile/ui/UserProfilePopover.tsx index c04b9c7063..5ff3036905 100644 --- a/desktop/src/features/profile/ui/UserProfilePopover.tsx +++ b/desktop/src/features/profile/ui/UserProfilePopover.tsx @@ -1,4 +1,5 @@ import * as React from "react"; +import { Activity } from "lucide-react"; import { useUserNotesQuery, @@ -12,6 +13,7 @@ import { usePresenceQuery } from "@/features/presence/hooks"; import { PresenceBadge } from "@/features/presence/ui/PresenceBadge"; import { formatRelativeTime } from "@/features/forum/lib/time"; import { rewriteRelayUrl } from "@/shared/lib/mediaUrl"; +import { useAgentSession } from "@/shared/context/AgentSessionContext"; import { Markdown } from "@/shared/ui/markdown"; import { Popover, PopoverContent, PopoverTrigger } from "@/shared/ui/popover"; import { BotIdenticon } from "@/features/messages/ui/BotIdenticon"; @@ -74,10 +76,15 @@ export function UserProfilePopover({ enabled: open, }); + const { onOpenAgentSession } = useAgentSession(); const relayAgent = relayAgentsQuery.data?.find((a) => a.pubkey === pubkey); const managedAgent = managedAgentsQuery.data?.find( (a) => a.pubkey === pubkey, ); + const canViewActivity = + role === "bot" && + managedAgent?.backend.type === "local" && + Boolean(onOpenAgentSession); const profile = profileQuery.data; const notes = notesQuery.data?.notes ?? []; const presenceStatus = presenceQuery.data?.[pubkey.toLowerCase()]; @@ -148,6 +155,21 @@ export function UserProfilePopover({

) : null} + {canViewActivity ? ( + + ) : null} +

{truncatePubkey(pubkey)}

diff --git a/desktop/src/shared/api/tauri.ts b/desktop/src/shared/api/tauri.ts index 1dd078a726..b245a74ca5 100644 --- a/desktop/src/shared/api/tauri.ts +++ b/desktop/src/shared/api/tauri.ts @@ -252,6 +252,7 @@ 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; @@ -857,6 +858,7 @@ 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, diff --git a/desktop/src/shared/api/types.ts b/desktop/src/shared/api/types.ts index 337cc1ee0d..e59e0c4f05 100644 --- a/desktop/src/shared/api/types.ts +++ b/desktop/src/shared/api/types.ts @@ -307,6 +307,7 @@ export type ManagedAgent = { lastExitCode: number | null; lastError: string | null; logPath: string; + observerUrl: string | null; startOnAppLaunch: boolean; backend: ManagedAgentBackend; backendAgentId: string | null; diff --git a/desktop/src/shared/context/AgentSessionContext.tsx b/desktop/src/shared/context/AgentSessionContext.tsx new file mode 100644 index 0000000000..8fd4903c4c --- /dev/null +++ b/desktop/src/shared/context/AgentSessionContext.tsx @@ -0,0 +1,32 @@ +import * as React from "react"; + +type AgentSessionContextValue = { + onOpenAgentSession: ((pubkey: string) => void) | null; +}; + +const AgentSessionContext = React.createContext({ + onOpenAgentSession: null, +}); + +export function AgentSessionProvider({ + children, + onOpenAgentSession, +}: { + children: React.ReactNode; + onOpenAgentSession: (pubkey: string) => void; +}) { + const value = React.useMemo( + () => ({ onOpenAgentSession }), + [onOpenAgentSession], + ); + + return ( + + {children} + + ); +} + +export function useAgentSession() { + return React.useContext(AgentSessionContext); +} diff --git a/desktop/src/shared/hooks/useStickToBottom.ts b/desktop/src/shared/hooks/useStickToBottom.ts new file mode 100644 index 0000000000..849bfd1903 --- /dev/null +++ b/desktop/src/shared/hooks/useStickToBottom.ts @@ -0,0 +1,60 @@ +import { useCallback, useEffect, useRef } from "react"; + +/** + * Keeps a scroll container pinned to the bottom as new content arrives, + * unless the user has scrolled up. Mirrors the "sticky scroll" pattern + * from goose's MessageTimeline. + * + * Attach `ref` to the scrollable container and `onScroll` as its scroll + * handler. The hook observes DOM mutations inside the container and + * auto-scrolls when the user is near the bottom (within 100 px). + * + * Scroll calls are batched via `requestAnimationFrame` so rapid streaming + * updates (e.g. token-by-token SSE) don't cause layout thrashing. + */ +export function useStickToBottom() { + const ref = useRef(null); + const isNearBottomRef = useRef(true); + + const onScroll = useCallback(() => { + const el = ref.current; + if (!el) return; + const { scrollTop, scrollHeight, clientHeight } = el; + isNearBottomRef.current = scrollHeight - scrollTop - clientHeight < 100; + }, []); + + useEffect(() => { + const el = ref.current; + if (!el) return; + + let rafId: number | null = null; + + const scrollIfSticky = () => { + // Coalesce to one scroll per animation frame. + if (rafId !== null) return; + rafId = requestAnimationFrame(() => { + rafId = null; + if (isNearBottomRef.current && ref.current) { + ref.current.scrollTo({ + top: ref.current.scrollHeight, + behavior: "smooth", + }); + } + }); + }; + + const observer = new MutationObserver(scrollIfSticky); + observer.observe(el, { + childList: true, + subtree: true, + characterData: true, + }); + + return () => { + observer.disconnect(); + if (rafId !== null) cancelAnimationFrame(rafId); + }; + }, []); + + return { ref, onScroll, isNearBottomRef }; +} diff --git a/desktop/src/testing/e2eBridge.ts b/desktop/src/testing/e2eBridge.ts index 4ce5acaac3..c0f6120dcf 100644 --- a/desktop/src/testing/e2eBridge.ts +++ b/desktop/src/testing/e2eBridge.ts @@ -331,6 +331,7 @@ 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" } @@ -681,6 +682,7 @@ 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, @@ -3642,6 +3644,9 @@ 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, @@ -3688,6 +3693,9 @@ 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; @@ -3703,6 +3711,7 @@ 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}`);