feat: Add local ACP session observer (#412)

This commit is contained in:
Bradley Axen
2026-04-29 04:21:58 +00:00
committed by GitHub
parent 2a01ca590d
commit 7c2c556d72
47 changed files with 3637 additions and 107 deletions
Generated
+1
View File
@@ -3621,6 +3621,7 @@ name = "sprout-acp"
version = "0.1.0"
dependencies = [
"anyhow",
"axum",
"chrono",
"clap",
"evalexpr",
+1
View File
@@ -0,0 +1 @@
hermit
Symlink
+1
View File
@@ -0,0 +1 @@
.flutter-3.41.7.pkg
+1
View File
@@ -0,0 +1 @@
.flutter-3.41.7.pkg
+1
View File
@@ -25,6 +25,7 @@ tokio = { workspace = true }
# WebSocket
tokio-tungstenite = { workspace = true }
axum = { workspace = true }
# Codec (bounded line reads)
tokio-util = { workspace = true }
+82 -5
View File
@@ -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<tokio::time::Instant>,
/// Optional local observer feed used by the desktop app.
observer: Option<ObserverHandle>,
/// Pool slot index for this agent process.
observer_agent_index: Option<usize>,
/// 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<ObserverHandle>, 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<String>, 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.
@@ -429,6 +463,33 @@ impl AcpClient {
}
};
self.cancel_with_cleanup_until(session_id, hard_deadline)
.await
}
/// Cancel a user-interrupted turn with a bounded grace window.
///
/// Some ACP servers currently keep streaming after `session/cancel`. For an
/// explicit Stop button, waiting until the original turn deadline can make
/// cancellation look broken. This variant gives the agent a short chance to
/// acknowledge cancellation, then returns a timeout so the caller can respawn
/// the agent process and actually stop the work.
pub async fn cancel_with_cleanup_grace(
&mut self,
session_id: &str,
grace: std::time::Duration,
) -> Result<StopReason, AcpError> {
let _ = self.current_hard_deadline.take();
let hard_deadline = tokio::time::Instant::now() + grace;
self.cancel_with_cleanup_until(session_id, hard_deadline)
.await
}
async fn cancel_with_cleanup_until(
&mut self,
session_id: &str,
hard_deadline: tokio::time::Instant,
) -> Result<StopReason, AcpError> {
// Validate precondition before any side effects — fail fast if there's
// no in-flight prompt (prevents writing permission responses or cancel
// notifications to the agent when no prompt is active).
@@ -456,10 +517,8 @@ impl AcpClient {
tracing::info!(target: "acp::cancel", "sent session/cancel for {session_id}");
// Use a fixed 30s idle timeout during cleanup — the cancel notification
// needs time to propagate and the agent may go silent while winding down.
// We do NOT use the caller's idle_timeout here: that value was tuned for
// normal prompt activity and may be very short (e.g. 5s). Using it during
// cancel would cause premature IdleTimeout before the cancelled response
// arrives, leaving the session in an inconsistent state.
// The separate hard_deadline bounds agents that keep producing output
// but ignore cancellation.
let cleanup_idle = std::time::Duration::from_secs(30);
let result = self
.read_until_response_with_idle_timeout(prompt_id, cleanup_idle, hard_deadline)
@@ -484,7 +543,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 +687,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 +701,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 +798,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 +812,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.
+19
View File
@@ -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<String>,
/// Local bind address for the ACP observer API.
///
/// Intended for the desktop app. The feed stays on localhost and is not
/// part of relay-visible Sprout history.
#[arg(long, env = "SPROUT_ACP_OBSERVER_ADDR")]
pub observer_addr: Option<String>,
/// Optional token required by the observer API.
#[arg(long, env = "SPROUT_ACP_OBSERVER_TOKEN")]
pub observer_token: Option<String>,
}
// ── 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<String>,
/// Local observer token, when configured.
pub observer_token: Option<String>,
}
/// 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,
}
}
+104 -9
View File
@@ -3,6 +3,7 @@
mod acp;
mod config;
mod filter;
mod observer;
mod pool;
mod queue;
mod relay;
@@ -19,7 +20,8 @@ use filter::SubscriptionRule;
use futures_util::FutureExt;
use nostr::ToBech32;
use pool::{
AgentPool, OwnedAgent, PromptContext, PromptOutcome, PromptResult, PromptSource, SessionState,
AgentPool, CancelMode, OwnedAgent, PromptContext, PromptOutcome, PromptResult, PromptSource,
SessionState,
};
use queue::{EventQueue, QueuedEvent, ThreadTags};
use relay::HarnessRelay;
@@ -529,6 +531,45 @@ async fn tokio_main() -> Result<()> {
let config = Config::from_cli().map_err(|e| anyhow::anyhow!("configuration error: {e}"))?;
tracing::info!("sprout-acp starting: {}", config.summary());
let (observer_control_tx, mut observer_control_rx) =
mpsc::channel::<observer::ObserverControlCommand>(32);
let observer = match config.observer_addr.as_deref() {
Some(addr) => {
match observer::spawn_observer_server(
addr,
config.observer_token.clone(),
Some(observer_control_tx.clone()),
)
.await
{
Ok(handle) => {
tracing::info!(target: "observer", "ACP observer listening on http://{}", handle.addr());
handle.emit(
"harness_started",
None,
&observer::ObserverContext::default(),
serde_json::json!({
"relayUrl": config.relay_url,
"agentCommand": config.agent_command,
"agentArgs": config.agent_args,
"parallelism": config.agents,
}),
);
Some(handle)
}
Err(error) => {
tracing::warn!(
target: "observer",
"failed to start local ACP observer at {addr}: {error}"
);
None
}
}
}
None => None,
};
// ── Step 1: Spawn N ACP agent subprocesses and initialize ─────────────────
//
// Finding #10: one agent failing to start must not kill the whole pool.
@@ -549,9 +590,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 +967,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);
});
}
@@ -989,6 +1039,29 @@ async fn tokio_main() -> Result<()> {
Some(Err(e)) = join_set.join_next(), if !join_set.is_empty() => {
Some(PoolEvent::Panic(e))
}
control = observer_control_rx.recv() => {
let _ = result_rx;
if let Some(command) = control {
match command {
observer::ObserverControlCommand::CancelTurn { channel_id, respond_to } => {
let fired = cancel_in_flight_task(&mut pool, channel_id, CancelMode::Stop);
if !fired {
tracing::warn!(
channel_id = %channel_id,
"observer cancel requested but no in-flight task — no-op"
);
}
let status = if fired {
observer::CancelTurnStatus::Sent
} else {
observer::CancelTurnStatus::NoActiveTurn
};
let _ = respond_to.send(observer::CancelTurnResponse { status });
}
}
}
None
}
// Remaining branches don't touch pool — evaluated when pool is idle.
sprout_event = relay.next_event() => {
let _ = result_rx; // end split borrow before relay handling
@@ -1162,7 +1235,7 @@ async fn tokio_main() -> Result<()> {
.await;
if let Some(owner) = owner {
if sprout_event.event.pubkey.to_hex() == *owner {
let fired = cancel_in_flight_task(&mut pool, sprout_event.channel_id);
let fired = cancel_in_flight_task(&mut pool, sprout_event.channel_id, CancelMode::Stop);
if !fired {
tracing::warn!(
channel_id = %sprout_event.channel_id,
@@ -1272,7 +1345,7 @@ async fn tokio_main() -> Result<()> {
}
};
if should_cancel {
cancel_in_flight_task(&mut pool, sprout_event.channel_id);
cancel_in_flight_task(&mut pool, sprout_event.channel_id, CancelMode::Interrupt);
}
}
// ── End mode gate ────────────────────────────────
@@ -1379,6 +1452,7 @@ async fn tokio_main() -> Result<()> {
&mut crash_history,
&respawn_tx,
&mut respawn_tasks,
observer.clone(),
) == LoopAction::Exit
{
break;
@@ -1393,6 +1467,7 @@ async fn tokio_main() -> Result<()> {
&mut crash_history,
&respawn_tx,
&mut respawn_tasks,
observer.clone(),
) == LoopAction::Exit
{
break;
@@ -1414,6 +1489,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");
@@ -1537,7 +1613,7 @@ enum LoopAction {
/// Send a cancel signal to the in-flight task for `channel_id`.
/// Returns `true` if a signal was sent, `false` if no in-flight task was found.
fn cancel_in_flight_task(pool: &mut AgentPool, channel_id: uuid::Uuid) -> bool {
fn cancel_in_flight_task(pool: &mut AgentPool, channel_id: uuid::Uuid, mode: CancelMode) -> bool {
let entry = pool
.task_map_mut()
.values_mut()
@@ -1545,7 +1621,7 @@ fn cancel_in_flight_task(pool: &mut AgentPool, channel_id: uuid::Uuid) -> bool {
if let Some(meta) = entry {
if let Some(tx) = meta.cancel_tx.take() {
let _ = tx.send(());
let _ = tx.send(mode);
tracing::info!(channel = %channel_id, "cancel signal sent to in-flight task");
return true;
}
@@ -1597,7 +1673,7 @@ fn dispatch_pending(
// Prompt text is now built inside run_prompt_task (needs async for
// context fetching). Pass None for prompt_text; batch carries the data.
let (cancel_tx, cancel_rx) = tokio::sync::oneshot::channel::<()>();
let (cancel_tx, cancel_rx) = tokio::sync::oneshot::channel::<CancelMode>();
let abort_handle = pool.join_set.spawn(async move {
pool::run_prompt_task(
@@ -1643,6 +1719,7 @@ fn handle_prompt_result(
crash_history: &mut [SlotCircuit],
respawn_tx: &mpsc::Sender<RespawnResult>,
respawn_tasks: &mut tokio::task::JoinSet<()>,
observer: Option<observer::ObserverHandle>,
) -> LoopAction {
let before = pool.task_map().len();
let agent_index = result.agent.index;
@@ -1722,6 +1799,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 +1853,7 @@ fn handle_prompt_result(
slot_history,
respawn_tx,
respawn_tasks,
observer,
) && pool.live_count() == 0
&& !any_respawn_in_flight(crash_history)
{
@@ -1808,6 +1887,7 @@ fn recover_panicked_agent(
crash_history: &mut [SlotCircuit],
respawn_tx: &mpsc::Sender<RespawnResult>,
respawn_tasks: &mut tokio::task::JoinSet<()>,
observer: Option<observer::ObserverHandle>,
) {
let task_id = join_error.id();
let Some(meta) = pool.task_map_mut().remove(&task_id) else {
@@ -1874,7 +1954,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 +1972,7 @@ fn drain_ready_join_results(
crash_history: &mut [SlotCircuit],
respawn_tx: &mpsc::Sender<RespawnResult>,
respawn_tasks: &mut tokio::task::JoinSet<()>,
observer: Option<observer::ObserverHandle>,
) -> LoopAction {
while let Some(Some(join_result)) = pool.join_set.join_next().now_or_never() {
if let Err(join_error) = join_result {
@@ -1907,6 +1988,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 +2073,7 @@ fn spawn_respawn_task(
slot: &mut SlotCircuit,
respawn_tx: &mpsc::Sender<RespawnResult>,
respawn_tasks: &mut tokio::task::JoinSet<()>,
observer: Option<observer::ObserverHandle>,
) -> bool {
let index = old_agent.index;
@@ -2027,7 +2110,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 +2127,24 @@ async fn spawn_and_init(
command: &str,
args: &[String],
extra_env: &[(String, String)],
agent_index: usize,
observer: Option<observer::ObserverHandle>,
) -> Result<AcpClient> {
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) => {
@@ -2554,6 +2647,8 @@ mod build_mcp_servers_tests {
respond_to: config::RespondTo::Anyone,
respond_to_allowlist: std::collections::HashSet::new(),
persona_env_vars: vec![],
observer_addr: None,
observer_token: None,
}
}
+418
View File
@@ -0,0 +1,418 @@
//! Local observer and control endpoint for ACP session activity.
//!
//! This is intentionally process-local infrastructure: it lets the desktop app
//! watch the raw ACP JSON-RPC stream and send tightly-scoped control commands
//! without sending private execution detail through the Sprout relay.
use std::{
collections::VecDeque,
convert::Infallible,
net::SocketAddr,
sync::{
atomic::{AtomicU64, Ordering},
Arc, Mutex,
},
};
use axum::{
extract::{Query, State},
http::{header, HeaderMap, HeaderValue, StatusCode},
response::{
sse::{Event, KeepAlive, Sse},
IntoResponse, Response,
},
routing::{get, post},
Json, Router,
};
use futures_util::{stream, StreamExt};
use serde::{Deserialize, Serialize};
use tokio::sync::{broadcast, mpsc, oneshot};
const OBSERVER_BUFFER_CAP: usize = 1_000;
/// Best-effort metadata attached to observer events.
#[derive(Clone, Debug, Default)]
pub struct ObserverContext {
/// Sprout channel UUID for the current turn, when channel-scoped.
pub channel_id: Option<String>,
/// ACP session ID associated with the current turn, once known.
pub session_id: Option<String>,
/// Local UUID for one prompt turn.
pub turn_id: Option<String>,
}
/// Handle used by the harness to publish local observer events.
#[derive(Clone)]
pub struct ObserverHandle {
inner: Arc<ObserverInner>,
}
struct ObserverInner {
tx: broadcast::Sender<ObserverEvent>,
buffer: Mutex<VecDeque<ObserverEvent>>,
seq: AtomicU64,
addr: SocketAddr,
}
/// Event delivered over the local observer SSE stream.
#[derive(Clone, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct ObserverEvent {
/// Monotonic process-local sequence number.
pub seq: u64,
/// RFC3339 UTC timestamp.
pub timestamp: String,
/// Observer event kind, for example `acp_read` or `turn_started`.
pub kind: String,
/// Pool slot index for the agent process that emitted the event.
pub agent_index: Option<usize>,
/// Sprout channel UUID for channel-scoped events.
pub channel_id: Option<String>,
/// ACP session ID when known.
pub session_id: Option<String>,
/// Local UUID for one prompt turn.
pub turn_id: Option<String>,
/// Raw or semantic event payload.
pub payload: serde_json::Value,
}
/// Commands accepted by the observer control loop.
#[derive(Debug)]
pub enum ObserverControlCommand {
/// Stop the active turn for a channel, if one exists.
CancelTurn {
channel_id: uuid::Uuid,
respond_to: oneshot::Sender<CancelTurnResponse>,
},
}
/// Response returned by observer control commands.
#[derive(Debug, Clone, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct CancelTurnResponse {
/// Result of attempting to send the cancel signal.
pub status: CancelTurnStatus,
}
/// Status for a cancel-turn request.
#[derive(Debug, Clone, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum CancelTurnStatus {
/// A cancel signal was sent to an in-flight channel turn.
Sent,
/// The channel had no active turn at the time of the request.
NoActiveTurn,
}
#[derive(Clone)]
struct ObserverServerState {
observer: ObserverHandle,
token: Option<String>,
control_tx: Option<mpsc::Sender<ObserverControlCommand>>,
}
#[derive(Deserialize)]
struct EventQuery {
token: Option<String>,
}
#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
struct CancelTurnRequest {
channel_id: uuid::Uuid,
}
/// Start the loopback observer HTTP server.
pub async fn spawn_observer_server(
bind_addr: &str,
token: Option<String>,
control_tx: Option<mpsc::Sender<ObserverControlCommand>>,
) -> anyhow::Result<ObserverHandle> {
let listener = tokio::net::TcpListener::bind(bind_addr).await?;
let addr = listener.local_addr()?;
if !addr.ip().is_loopback() {
anyhow::bail!("observer bind address must be loopback, got {addr}");
}
let (tx, _) = broadcast::channel(OBSERVER_BUFFER_CAP);
let observer = ObserverHandle {
inner: Arc::new(ObserverInner {
tx,
buffer: Mutex::new(VecDeque::with_capacity(OBSERVER_BUFFER_CAP)),
seq: AtomicU64::new(1),
addr,
}),
};
let state = ObserverServerState {
observer: observer.clone(),
token,
control_tx,
};
let app = Router::new()
.route("/events", get(events_handler))
.route("/health", get(health_handler))
.route("/control/cancel", post(cancel_turn_handler))
.with_state(state);
tokio::spawn(async move {
if let Err(error) = axum::serve(listener, app).await {
tracing::warn!(target: "observer", "observer server stopped: {error}");
}
});
Ok(observer)
}
impl ObserverHandle {
/// Return the bound loopback address.
pub fn addr(&self) -> SocketAddr {
self.inner.addr
}
/// Subscribe to live observer events.
pub fn subscribe(&self) -> broadcast::Receiver<ObserverEvent> {
self.inner.tx.subscribe()
}
/// Return the current replay buffer.
pub fn snapshot(&self) -> Vec<ObserverEvent> {
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()
}
}
}
/// Emit a local observer event.
pub fn emit(
&self,
kind: impl Into<String>,
agent_index: Option<usize>,
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<ObserverServerState>) -> Response {
serde_json::json!({
"ok": true,
"addr": state.observer.addr().to_string(),
"control": state.control_tx.is_some(),
})
.to_string()
.into_response()
}
async fn events_handler(
State(state): State<ObserverServerState>,
Query(query): Query<EventQuery>,
headers: HeaderMap,
) -> Response {
if !authorized(state.token.as_deref(), query.token.as_deref(), &headers) {
return (StatusCode::UNAUTHORIZED, "invalid observer token").into_response();
}
let origin = headers.get(header::ORIGIN).cloned();
if !origin_allowed(origin.as_ref()) {
return (StatusCode::FORBIDDEN, "forbidden: invalid origin").into_response();
}
let replay = state.observer.snapshot();
let replay_stream = stream::iter(replay.into_iter().map(event_to_sse));
let live_rx = state.observer.subscribe();
let live_stream = stream::unfold(live_rx, |mut rx| async move {
loop {
match rx.recv().await {
Ok(event) => return Some((event_to_sse(event), rx)),
Err(broadcast::error::RecvError::Lagged(_)) => continue,
Err(broadcast::error::RecvError::Closed) => return None,
}
}
});
let stream = replay_stream.chain(live_stream);
let sse = Sse::new(stream).keep_alive(KeepAlive::default());
let mut response = sse.into_response();
if let Some(origin) = origin {
response
.headers_mut()
.insert(header::ACCESS_CONTROL_ALLOW_ORIGIN, origin);
}
response.headers_mut().insert(
header::CACHE_CONTROL,
HeaderValue::from_static("no-store, no-cache"),
);
response
}
async fn cancel_turn_handler(
State(state): State<ObserverServerState>,
Query(query): Query<EventQuery>,
headers: HeaderMap,
Json(request): Json<CancelTurnRequest>,
) -> Response {
if !authorized(state.token.as_deref(), query.token.as_deref(), &headers) {
return (StatusCode::UNAUTHORIZED, "invalid observer token").into_response();
}
if !origin_allowed(headers.get(header::ORIGIN)) {
return (StatusCode::FORBIDDEN, "forbidden: invalid origin").into_response();
}
let Some(control_tx) = state.control_tx else {
return (
StatusCode::SERVICE_UNAVAILABLE,
"observer control is not available",
)
.into_response();
};
let (respond_to, response_rx) = oneshot::channel();
let command = ObserverControlCommand::CancelTurn {
channel_id: request.channel_id,
respond_to,
};
if control_tx.send(command).await.is_err() {
return (
StatusCode::SERVICE_UNAVAILABLE,
"observer control loop is not available",
)
.into_response();
}
match tokio::time::timeout(std::time::Duration::from_secs(2), response_rx).await {
Ok(Ok(response)) => Json(response).into_response(),
Ok(Err(_)) => (
StatusCode::SERVICE_UNAVAILABLE,
"observer control response was dropped",
)
.into_response(),
Err(_) => (StatusCode::GATEWAY_TIMEOUT, "observer control timed out").into_response(),
}
}
fn authorized(expected: Option<&str>, query_token: Option<&str>, headers: &HeaderMap) -> bool {
token_matches(expected, query_token) || token_matches(expected, bearer_token(headers))
}
fn bearer_token(headers: &HeaderMap) -> Option<&str> {
let value = headers.get(header::AUTHORIZATION)?.to_str().ok()?;
value.strip_prefix("Bearer ")
}
fn token_matches(expected: Option<&str>, actual: Option<&str>) -> bool {
match expected {
Some(expected) => actual == Some(expected),
None => true,
}
}
fn origin_allowed(origin: Option<&HeaderValue>) -> bool {
let Some(origin) = origin.and_then(|value| value.to_str().ok()) else {
return true;
};
origin == "tauri://localhost"
|| origin == "http://tauri.localhost"
|| origin == "https://tauri.localhost"
|| origin.starts_with("http://localhost:")
|| origin.starts_with("http://127.0.0.1:")
}
fn event_to_sse(event: ObserverEvent) -> Result<Event, Infallible> {
let data = serde_json::to_string(&event).unwrap_or_else(|error| {
serde_json::json!({
"seq": event.seq,
"timestamp": event.timestamp,
"kind": "observer_serialize_error",
"payload": {"error": error.to_string()},
})
.to_string()
});
Ok(Event::default().id(event.seq.to_string()).data(data))
}
/// Build observer context values from optional channel/session/turn IDs.
pub fn context_for(
channel_id: Option<uuid::Uuid>,
session_id: Option<String>,
turn_id: Option<String>,
) -> ObserverContext {
ObserverContext {
channel_id: channel_id.map(|id| id.to_string()),
session_id,
turn_id,
}
}
#[cfg(test)]
mod tests {
use axum::http::{HeaderMap, HeaderValue};
use super::{authorized, origin_allowed, token_matches};
#[test]
fn observer_token_requires_exact_match_when_configured() {
assert!(token_matches(Some("secret"), Some("secret")));
assert!(!token_matches(Some("secret"), Some("wrong")));
assert!(!token_matches(Some("secret"), None));
assert!(token_matches(None, None));
assert!(token_matches(None, Some("anything")));
}
#[test]
fn observer_authorization_accepts_bearer_token() {
let mut headers = HeaderMap::new();
headers.insert(
axum::http::header::AUTHORIZATION,
HeaderValue::from_static("Bearer secret"),
);
assert!(authorized(Some("secret"), None, &headers));
assert!(!authorized(Some("other"), None, &headers));
}
#[test]
fn observer_origin_allows_tauri_and_loopback_only() {
assert!(origin_allowed(None));
assert!(origin_allowed(Some(&HeaderValue::from_static(
"tauri://localhost"
))));
assert!(origin_allowed(Some(&HeaderValue::from_static(
"http://localhost:1420"
))));
assert!(origin_allowed(Some(&HeaderValue::from_static(
"http://127.0.0.1:1420"
))));
assert!(!origin_allowed(Some(&HeaderValue::from_static(
"https://example.com"
))));
}
}
+78 -8
View File
@@ -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,
@@ -53,7 +54,7 @@ pub struct TaskMeta {
pub recoverable_batch: Option<FlushBatch>,
/// Cancel signal for the in-flight prompt task.
/// `None` for heartbeat tasks (not cancellable) and after signal is consumed.
pub cancel_tx: Option<tokio::sync::oneshot::Sender<()>>,
pub cancel_tx: Option<tokio::sync::oneshot::Sender<CancelMode>>,
}
/// Agent-level model capabilities. Populated on first session creation.
@@ -153,6 +154,15 @@ pub enum PromptSource {
Heartbeat,
}
/// How an in-flight channel turn should be cancelled.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum CancelMode {
/// Stop the current turn and drop its triggering batch.
Stop,
/// Stop the current turn and requeue its triggering batch for a merged re-prompt.
Interrupt,
}
/// Outcome of a prompt task.
#[allow(dead_code)]
pub enum PromptOutcome {
@@ -596,7 +606,7 @@ pub async fn run_prompt_task(
prompt_text: Option<String>,
ctx: Arc<PromptContext>,
result_tx: mpsc::UnboundedSender<PromptResult>,
cancel_rx: Option<tokio::sync::oneshot::Receiver<()>>,
cancel_rx: Option<tokio::sync::oneshot::Receiver<CancelMode>>,
) {
// ── Determine source and resolve/create session ───────────────────────
@@ -605,6 +615,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<String> = 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 +724,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 ──────────────────────
@@ -884,51 +930,75 @@ pub async fn run_prompt_task(
ctx.idle_timeout,
ctx.max_turn_duration,
) => result,
_ = rx => {
mode = rx => {
let cancel_mode = mode.unwrap_or(CancelMode::Stop);
// Cancel signal received. Guard against Race 1: the turn may
// have completed naturally just as cancel fired.
if agent.acp.has_in_flight_prompt() {
// Prompt is genuinely in-flight — cancel it.
match agent.acp.cancel_with_cleanup(&session_id, ctx.idle_timeout).await {
match agent
.acp
.cancel_with_cleanup_grace(
&session_id,
std::time::Duration::from_secs(5),
)
.await
{
Ok(stop_reason) => {
log_stop_reason(&source, &stop_reason);
agent.state.invalidate(&source);
let retry_batch = match cancel_mode {
CancelMode::Interrupt => requeue_batch_if_queue(&ctx, batch),
CancelMode::Stop => None,
};
let _ = result_tx.send(PromptResult {
agent,
source,
outcome: PromptOutcome::Cancelled,
batch: requeue_batch_if_queue(&ctx, batch),
batch: retry_batch,
});
return;
}
Err(AcpError::AgentExited) => {
agent.state.invalidate_all();
let retry_batch = match cancel_mode {
CancelMode::Interrupt => requeue_batch_if_queue(&ctx, batch),
CancelMode::Stop => None,
};
let _ = result_tx.send(PromptResult {
agent,
source,
outcome: PromptOutcome::AgentExited,
batch: requeue_batch_if_queue(&ctx, batch),
batch: retry_batch,
});
return;
}
Err(AcpError::IdleTimeout(_) | AcpError::HardTimeout) => {
// Cancel drain timed out — agent state uncertain.
agent.state.invalidate(&source);
let retry_batch = match cancel_mode {
CancelMode::Interrupt => requeue_batch_if_queue(&ctx, batch),
CancelMode::Stop => None,
};
let _ = result_tx.send(PromptResult {
agent,
source,
outcome: PromptOutcome::Timeout,
batch: requeue_batch_if_queue(&ctx, batch),
batch: retry_batch,
});
return;
}
Err(e) => {
agent.state.invalidate(&source);
let retry_batch = match cancel_mode {
CancelMode::Interrupt => requeue_batch_if_queue(&ctx, batch),
CancelMode::Stop => None,
};
let _ = result_tx.send(PromptResult {
agent,
source,
outcome: PromptOutcome::Error(e),
batch: requeue_batch_if_queue(&ctx, batch),
batch: retry_batch,
});
return;
}
+2 -2
View File
@@ -45,7 +45,7 @@ const overrides = new Map([
["src-tauri/src/lib.rs", 710], // sprout-media:// proxy + Range headers + Sprout nest init (ensure_nest) in setup() + huddle command registration + PTT global shortcut handler + persona pack commands + app_handle storage for event emission
["src-tauri/src/commands/media.rs", 720], // ffmpeg video transcode + poster frame extraction + run_ffmpeg_with_timeout (find_ffmpeg, is_video_file, transcode_to_mp4, extract_poster_frame, transcode_and_extract_poster) + spawn_blocking wrappers + tests
["src-tauri/src/commands/agents.rs", 881], // remote agent lifecycle routing (local + provider branches) + scope enforcement + persona pack metadata wiring + mcp_toolsets field + NIP-OA auth_tag in deploy payload
["src-tauri/src/managed_agents/runtime.rs", 700], // KNOWN_AGENT_BINARIES const + process_belongs_to_us FFI (macOS proc_name + Linux /proc/comm) + terminate_process + start/stop/sync lifecycle + pack persona live-read + login shell PATH augmentation
["src-tauri/src/managed_agents/runtime.rs", 705], // KNOWN_AGENT_BINARIES const + process_belongs_to_us FFI (macOS proc_name + Linux /proc/comm) + terminate_process + start/stop/sync lifecycle + pack persona live-read + login shell PATH augmentation + observer endpoint wiring
["src-tauri/src/managed_agents/backend.rs", 530], // provider IPC, validation, discovery, binary resolution + tests
["src/features/huddle/HuddleContext.tsx", 650], // huddle lifecycle context + joinHuddle + connectAndSetupMedia shared helper + activeSpeakers/isReconnecting state + PTT (reusable AudioContext) + TTS subscription + mic level analyser (10fps throttle) + agent pubkey refresh
["src/features/agents/hooks.ts", 540], // agent query/mutation surface now includes built-in persona library activation + useUpdateManagedAgentMutation
@@ -68,7 +68,7 @@ const overrides = new Map([
["src-tauri/src/relay.rs", 510], // +4 lines for NIP-OA auth tag injection in profile sync (build_profile_event) + verification test
["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", 1140], // +14 lines for 3 NIP-AB pairing command wrappers + applyWorkspace + NIP-44 encrypt/decrypt wrappers
["src/shared/api/tauri.ts", 1140], // pairing command wrappers + applyWorkspace + NIP-44 encrypt/decrypt wrappers + observer_url field
]);
async function walkFiles(directory) {
@@ -0,0 +1,90 @@
use serde::{Deserialize, Serialize};
use tauri::State;
use crate::{app_state::AppState, relay::relay_error_message};
#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct CancelManagedAgentTurnResponse {
pub status: String,
}
#[derive(Debug, Deserialize)]
#[serde(rename_all = "camelCase")]
struct ObserverCancelTurnResponse {
status: String,
}
#[tauri::command]
pub async fn cancel_managed_agent_turn(
pubkey: String,
channel_id: String,
state: State<'_, AppState>,
) -> Result<CancelManagedAgentTurnResponse, String> {
let observer_url = {
let runtimes = state
.managed_agent_processes
.lock()
.map_err(|error| error.to_string())?;
let runtime = runtimes
.get(&pubkey)
.ok_or_else(|| format!("agent {pubkey} is not running locally"))?;
runtime
.observer_url
.clone()
.ok_or_else(|| format!("agent {pubkey} does not expose an observer control endpoint"))?
};
let (control_url, token) = observer_control_url(&observer_url)?;
let request = state
.http_client
.post(control_url)
.bearer_auth(token)
.json(&serde_json::json!({ "channelId": channel_id }));
let response = request
.send()
.await
.map_err(|error| format!("observer cancel request failed: {error}"))?;
if !response.status().is_success() {
return Err(relay_error_message(response).await);
}
let body = response
.json::<ObserverCancelTurnResponse>()
.await
.map_err(|error| format!("observer cancel response parse failed: {error}"))?;
Ok(CancelManagedAgentTurnResponse {
status: body.status,
})
}
fn observer_control_url(observer_url: &str) -> Result<(String, String), String> {
let mut url = url::Url::parse(observer_url)
.map_err(|error| format!("invalid observer URL for agent: {error}"))?;
let token = url
.query_pairs()
.find_map(|(key, value)| (key == "token").then(|| value.into_owned()))
.filter(|value| !value.is_empty())
.ok_or_else(|| "observer URL is missing its control token".to_string())?;
url.set_path("/control/cancel");
url.set_query(None);
Ok((url.to_string(), token))
}
#[cfg(test)]
mod tests {
use super::observer_control_url;
#[test]
fn derives_control_url_and_token_from_events_url() {
let (url, token) =
observer_control_url("http://127.0.0.1:1234/events?token=abc").expect("control url");
assert_eq!(url, "http://127.0.0.1:1234/control/cancel");
assert_eq!(token, "abc");
}
}
+2
View File
@@ -1,3 +1,4 @@
mod agent_control;
mod agent_discovery;
mod agent_models;
mod agent_settings;
@@ -18,6 +19,7 @@ pub mod tokens;
mod workflows;
mod workspace;
pub use agent_control::*;
pub use agent_discovery::*;
pub use agent_models::*;
pub use agent_settings::*;
+1
View File
@@ -412,6 +412,7 @@ pub fn run() {
revoke_token,
revoke_all_tokens,
list_relay_agents,
cancel_managed_agent_turn,
list_managed_agents,
create_managed_agent,
start_managed_agent,
@@ -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::*;
@@ -0,0 +1,25 @@
/// Loopback observer endpoint assigned to one local ACP harness process.
pub struct ObserverEndpoint {
/// Host:port bind address passed to `sprout-acp`.
pub addr: String,
/// Token required by observer requests.
pub token: String,
/// Event-stream URL including the observer token.
pub url: String,
}
/// Allocate an unused loopback port and random observer token.
pub fn allocate_observer_endpoint() -> Result<ObserverEndpoint, String> {
let listener = std::net::TcpListener::bind("127.0.0.1:0")
.map_err(|error| format!("failed to allocate observer port: {error}"))?;
let port = listener
.local_addr()
.map_err(|error| format!("failed to inspect observer port: {error}"))?
.port();
drop(listener);
let token = uuid::Uuid::new_v4().to_string();
let addr = format!("127.0.0.1:{port}");
let url = format!("http://127.0.0.1:{port}/events?token={token}");
Ok(ObserverEndpoint { addr, token, url })
}
@@ -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>), 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();
@@ -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<ManagedAgentSummary, String> {
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>), String> {
let log_path = managed_agent_log_path(app, &record.pubkey)?;
append_log_marker(
&log_path,
@@ -567,6 +571,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)]
@@ -585,7 +593,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(
@@ -616,7 +624,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();
@@ -628,7 +636,11 @@ pub fn start_managed_agent_process(
runtimes.insert(
record.pubkey.clone(),
ManagedAgentProcess { child, log_path },
ManagedAgentProcess {
child,
log_path,
observer_url,
},
);
Ok(())
}
@@ -127,6 +127,7 @@ pub struct ManagedAgentRecord {
pub struct ManagedAgentProcess {
pub child: Child,
pub log_path: PathBuf,
pub observer_url: Option<String>,
}
#[derive(Debug, Clone, Serialize)]
@@ -159,6 +160,7 @@ pub struct ManagedAgentSummary {
pub last_error: Option<String>,
pub start_on_app_launch: bool,
pub log_path: String,
pub observer_url: Option<String>,
}
#[derive(Debug, Deserialize)]
@@ -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<TranscriptItem, { type: "tool" }>;
}) {
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<HTMLDetailsElement>) => {
setIsExpanded(event.currentTarget.open);
},
[],
);
return (
<div className="not-prose w-full px-1">
<details
className="group w-full"
onToggle={handleToggle}
open={isExpanded}
>
<summary className="inline-flex max-w-full cursor-pointer list-none items-center gap-1.5 py-px">
{ToolIcon ? (
<ToolIcon
className={cn(
"h-4 w-4 shrink-0",
sproutTool ? "text-primary" : "text-muted-foreground",
)}
/>
) : null}
<span className="min-w-0 truncate text-sm font-medium">
{toolTitle}
</span>
{sproutTool ? (
<SproutToolInlineAction args={item.args} result={item.result} />
) : null}
{showStatus ? (
<span className="flex shrink-0 items-center gap-1 text-xs text-muted-foreground">
<status.Icon
className={cn(
"h-4 w-4",
item.status === "executing" && "animate-pulse",
)}
/>
{status.label}
</span>
) : null}
<ChevronDown className="h-3.5 w-3.5 shrink-0 text-muted-foreground transition-transform group-open:rotate-180" />
</summary>
<ToolDetailBlocks
args={item.args}
description={sproutTool?.label}
hasArgs={hasArgs}
hasResult={hasResult}
isError={item.isError}
result={item.result}
/>
</details>
</div>
);
}
function ToolDetailBlocks({
args,
description,
hasArgs,
hasResult,
isError,
result,
}: {
args: Record<string, unknown>;
description?: string;
hasArgs: boolean;
hasResult: boolean;
isError: boolean;
result: string;
}) {
return (
<div className="space-y-4 py-2 pl-5 text-popover-foreground outline-none">
{description ? (
<p className="max-w-2xl text-xs leading-5 text-muted-foreground">
{description}
</p>
) : null}
{hasArgs ? (
<ToolCodeBlock
label="Parameters"
tone="muted"
value={JSON.stringify(args, null, 2)}
/>
) : null}
{hasResult ? (
<ToolCodeBlock
label={isError ? "Error" : "Result"}
tone={isError ? "error" : "muted"}
value={result}
/>
) : null}
{!hasArgs && !hasResult ? (
<p className="text-sm text-muted-foreground/80">
Waiting for tool details.
</p>
) : null}
</div>
);
}
function ToolCodeBlock({
label,
tone,
value,
}: {
label: string;
tone: "muted" | "error";
value: string;
}) {
return (
<div className="space-y-2 overflow-hidden">
<h4 className="text-xs font-medium uppercase tracking-wide text-muted-foreground">
{label}
</h4>
<pre
className={cn(
"max-h-64 overflow-auto whitespace-pre-wrap break-words rounded-md px-3 py-2 font-mono text-xs leading-5",
tone === "error"
? "bg-destructive/10 text-destructive"
: "bg-muted/50 text-foreground",
)}
>
{formatCodeValue(value)}
</pre>
</div>
);
}
function SproutToolInlineAction({
args,
result,
}: {
args: Record<string, unknown>;
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 (
<button
className="inline-flex max-w-[14rem] shrink min-w-0 items-center gap-1 rounded-full border border-primary/20 bg-primary/[0.05] px-1.5 py-0.5 text-[11px] font-normal leading-none text-primary/90 transition-colors hover:border-primary/35 hover:bg-primary/10 hover:text-primary"
onClick={(event) => {
event.preventDefault();
event.stopPropagation();
action.onClick?.();
}}
title={action.title}
type="button"
>
<span className="shrink-0">{action.label}</span>
<span className="truncate">{action.value}</span>
<ArrowUpRight className="h-3 w-3 shrink-0" />
</button>
);
}
return (
<span
className="inline-flex max-w-[14rem] shrink min-w-0 items-center gap-1 rounded-full border border-border/60 bg-muted/40 px-1.5 py-0.5 text-[11px] font-normal leading-none text-muted-foreground"
title={action.title}
>
<span className="shrink-0">{action.label}</span>
<span className="truncate">{action.value}</span>
</span>
);
}
type SproutToolInlineActionModel = {
label: string;
value: string;
title: string;
onClick?: () => void;
};
function getSproutToolInlineAction({
args,
channelId,
channels,
openChannel,
resultValue,
}: {
args: Record<string, unknown>;
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)}`;
}
@@ -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 (
<div className="flex min-h-56 flex-col items-center justify-center px-6 py-10 text-center">
<Radio className="mx-auto h-5 w-5 text-muted-foreground" />
<p className="mt-3 text-sm font-medium">No ACP activity yet</p>
<p className="mt-1 text-sm text-muted-foreground">{emptyDescription}</p>
</div>
);
}
return (
<div
aria-label="Live ACP transcript"
aria-live="polite"
className="mx-auto w-full max-w-3xl py-1"
role="log"
>
{items.map((item) => (
<div className="mt-4 first:mt-0" key={item.id}>
<TranscriptItemView agentName={agentName} item={item} />
</div>
))}
</div>
);
}
function TranscriptItemView({
agentName,
item,
}: {
agentName: string;
item: TranscriptItem;
}) {
if (item.type === "message") {
return <MessageItem agentName={agentName} item={item} />;
}
if (item.type === "tool") {
return <ToolItem item={item} />;
}
if (item.type === "thought") {
return <ThoughtItem item={item} />;
}
if (item.type === "metadata") {
return <MetadataItem item={item} />;
}
return <LifecycleItem item={item} />;
}
function MessageItem({
agentName,
item,
}: {
agentName: string;
item: Extract<TranscriptItem, { type: "message" }>;
}) {
const isAssistant = item.role === "assistant";
const text = item.text.trim();
return (
<div
className={cn(
"flex px-1 py-1 animate-in fade-in duration-200 motion-reduce:animate-none",
isAssistant ? "flex-row" : "ml-auto flex-row-reverse",
)}
data-role={isAssistant ? "assistant-message" : "user-message"}
>
<div
className={cn(
"group relative min-w-0 flex flex-col gap-1",
isAssistant ? "w-full items-start" : "max-w-[85%] items-end",
)}
>
{isAssistant ? (
<div className="mb-0.5 flex items-center gap-1 text-xs">
<span className="flex h-5 w-5 items-center justify-center">
<Bot className="h-3.5 w-3.5 text-muted-foreground" />
</span>
<span className="font-normal text-foreground">{agentName}</span>
</div>
) : null}
<div
className={cn(
"w-full min-w-0 text-sm leading-relaxed",
!isAssistant && "rounded-2xl bg-muted p-3 text-foreground",
)}
>
{isAssistant ? (
<Markdown compact content={text || " "} />
) : (
<p className="whitespace-pre-wrap break-words">{text}</p>
)}
</div>
</div>
</div>
);
}
function ThoughtItem({
item,
}: {
item: Extract<TranscriptItem, { type: "thought" }>;
}) {
return (
<details className="group not-prose w-full px-1">
<summary className="inline-flex max-w-full cursor-pointer list-none items-center gap-1.5 py-px text-muted-foreground">
<Brain className="h-4 w-4" />
<span className="truncate text-sm font-medium">{item.title}</span>
<ChevronDown className="h-3.5 w-3.5 shrink-0 transition-transform group-open:rotate-180" />
</summary>
<div className="py-2 pl-5 text-sm leading-6 text-muted-foreground">
<Markdown compact content={item.text.trim() || " "} />
</div>
</details>
);
}
function MetadataItem({
item,
}: {
item: Extract<TranscriptItem, { type: "metadata" }>;
}) {
return (
<details className="group not-prose w-full px-1">
<summary className="inline-flex max-w-full cursor-pointer list-none items-center gap-1.5 py-px text-muted-foreground">
<TerminalSquare className="h-4 w-4" />
<span className="truncate text-sm font-medium">{item.title}</span>
<span className="shrink-0 text-xs">
{item.sections.length} section{item.sections.length === 1 ? "" : "s"}
</span>
<ChevronDown className="h-3.5 w-3.5 shrink-0 transition-transform group-open:rotate-180" />
</summary>
<div className="space-y-3 py-2 pl-5">
{item.sections.map((section) => (
<details
className="group/section"
key={`${section.title}:${section.body.slice(0, 48)}`}
>
<summary className="inline-flex max-w-full cursor-pointer list-none items-center gap-1.5 text-xs font-medium text-foreground/80">
<span className="truncate">{section.title}</span>
<ChevronDown className="h-3.5 w-3.5 shrink-0 text-muted-foreground transition-transform group-open/section:rotate-180" />
</summary>
<pre className="mt-2 max-h-56 overflow-auto whitespace-pre-wrap break-words rounded-md bg-muted/50 px-3 py-2 font-mono text-[11px] leading-5 text-muted-foreground">
{section.body.trim() || "No metadata."}
</pre>
</details>
))}
</div>
</details>
);
}
function LifecycleItem({
item,
}: {
item: Extract<TranscriptItem, { type: "lifecycle" }>;
}) {
const isError = item.title.toLowerCase().includes("error");
return (
<div
className={cn(
"px-4 py-2 text-center text-xs",
isError ? "text-destructive" : "text-muted-foreground",
)}
>
<span className="font-medium">{item.title}</span>
{item.text ? <span> - {item.text}</span> : null}
</div>
);
}
@@ -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 (
<section
className={cn(
"rounded-lg border border-border/70 bg-background/80 p-4 shadow-sm",
className,
)}
>
{showHeader ? (
<SessionHeader
connectionState={connectionState}
eventCount={scopedEvents.length}
hasObserver={Boolean(agent.observerUrl)}
latestSessionId={latestSessionId}
/>
) : null}
<SessionBody
agentName={agent.name}
connectionState={connectionState}
emptyDescription={emptyDescription}
errorMessage={errorMessage}
events={scopedEvents}
hasObserver={Boolean(agent.observerUrl)}
showRaw={showRaw}
transcript={transcript}
/>
</section>
);
}
function SessionHeader({
connectionState,
eventCount,
hasObserver,
latestSessionId,
}: {
connectionState: ConnectionState;
eventCount: number;
hasObserver: boolean;
latestSessionId: string | null | undefined;
}) {
return (
<div className="flex flex-col gap-3 sm:flex-row sm:items-start sm:justify-between">
<div className="min-w-0">
<div className="flex items-center gap-2">
<h3 className="text-sm font-semibold tracking-tight">
Live ACP session
</h3>
<ObserverStatusBadge state={connectionState} />
</div>
<p className="mt-1 text-sm text-muted-foreground">
{hasObserver
? latestSessionId
? `Session ${shorten(latestSessionId)}`
: "Waiting for the next agent turn."
: "Restart this local agent to attach the observer feed."}
</p>
</div>
<Badge className="w-fit font-mono" variant="outline">
{eventCount} event{eventCount === 1 ? "" : "s"}
</Badge>
</div>
);
}
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 ? (
<EmptyObserverState />
) : connectionState === "connecting" && events.length === 0 ? (
<SessionLoadingSkeleton />
) : (
<div
className={cn(
showRaw
? "mt-4 grid gap-4 xl:grid-cols-[minmax(0,1fr)_20rem]"
: "mt-0",
)}
>
<AgentSessionTranscriptList
agentName={agentName}
emptyDescription={emptyDescription}
items={transcript}
/>
{showRaw ? <RawEventRail events={events} /> : null}
</div>
)}
{errorMessage ? (
<p className="mt-4 inline-flex items-center gap-2 rounded-lg border border-destructive/30 bg-destructive/10 px-3 py-2 text-sm text-destructive">
<CircleAlert className="h-4 w-4" />
{errorMessage}
</p>
) : null}
</>
);
}
function SessionLoadingSkeleton() {
return (
<div className="mx-auto w-full max-w-3xl space-y-6 py-4">
<div className="flex justify-end">
<div className="max-w-[70%] space-y-2">
<Skeleton className="h-4 w-48 rounded-lg" />
</div>
</div>
<div className="space-y-2">
<div className="flex items-center gap-2">
<Skeleton className="h-5 w-5 rounded-full" />
<Skeleton className="h-3 w-16 rounded-full" />
</div>
<div className="space-y-2">
<Skeleton className="h-4 w-full rounded-lg" />
<Skeleton className="h-4 w-[86%] rounded-lg" />
<Skeleton className="h-4 w-[58%] rounded-lg" />
</div>
</div>
<div className="space-y-2">
<Skeleton className="h-4 w-44 rounded-lg" />
<Skeleton className="h-4 w-[68%] rounded-lg" />
</div>
</div>
);
}
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 (
<Badge className="gap-1.5" variant={display.variant}>
<display.Icon
className={cn("h-3.5 w-3.5", state === "connecting" && "animate-spin")}
/>
{display.label}
</Badge>
);
}
function EmptyObserverState() {
return (
<div className="mt-4 flex min-h-48 flex-col items-center justify-center px-6 py-8 text-center">
<TerminalSquare className="mx-auto h-5 w-5 text-muted-foreground" />
<p className="mt-3 text-sm font-medium">Observer not attached</p>
<p className="mt-1 text-sm text-muted-foreground">
The live feed is available for local agents started after this update.
</p>
</div>
);
}
@@ -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 (
<aside className="rounded-lg border border-border/70 bg-[#17171d] text-zinc-100">
<div className="flex items-center justify-between border-b border-white/10 px-3 py-2">
<div>
<p className="text-[11px] font-semibold uppercase tracking-[0.16em] text-zinc-400">
Raw ACP
</p>
<p className="text-xs text-zinc-500">JSON-RPC payloads</p>
</div>
<Button
className="h-7 text-zinc-300 hover:text-zinc-50"
onClick={() => setExpanded((current) => !current)}
size="sm"
variant="ghost"
>
{expanded ? "Latest" : "All"}
</Button>
</div>
<div className="max-h-[34rem] overflow-auto px-3 py-3">
{visible.length === 0 ? (
<p className="text-xs text-zinc-500">No raw events yet.</p>
) : (
<div className="space-y-2">
{visible.map((event) => (
<details
className="rounded-md border border-white/10 bg-white/[0.03] px-2 py-1.5"
key={event.seq}
>
<summary className="cursor-pointer select-none text-xs text-zinc-300">
<span className="font-mono text-zinc-500">#{event.seq}</span>{" "}
{describeRawEvent(event)}
</summary>
<pre className="mt-2 max-h-72 overflow-auto whitespace-pre-wrap break-words text-[11px] leading-5 text-zinc-300">
{JSON.stringify(event.payload, null, 2)}
</pre>
</details>
))}
</div>
)}
</div>
</aside>
);
}
@@ -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;
}
@@ -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<string, TranscriptItem>();
// 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<string, string>();
const sealedKeys = new Set<string>();
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<string, unknown>,
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;
}
@@ -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, unknown>): 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<string, unknown>,
): Record<string, unknown> {
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<string, unknown>;
}
}
return {};
}
export function extractToolIdentity(update: Record<string, unknown>): {
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, unknown>): 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, unknown>): 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;
}
@@ -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<string, unknown>;
result: string;
isError: boolean;
timestamp: string;
};
export type PromptSection = {
title: string;
body: string;
};
export type SproutToolInfo = {
icon: LucideIcon;
label: string;
tone: "read" | "write" | "admin";
};
@@ -0,0 +1,81 @@
export function getToolString(
record: Record<string, unknown>,
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<string, unknown>,
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<string, unknown>,
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<string, unknown> {
return value && typeof value === "object"
? (value as Record<string, unknown>)
: {};
}
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)}`;
}
@@ -0,0 +1,68 @@
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<ObserverEvent[]>([]);
const [connectionState, setConnectionState] =
React.useState<ConnectionState>("idle");
const [errorMessage, setErrorMessage] = React.useState<string | null>(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) => {
if (current.some((existing) => existing.seq === parsed.seq)) {
return current;
}
const next = [...current, parsed];
return next.length > MAX_OBSERVER_EVENTS
? next.slice(next.length - MAX_OBSERVER_EVENTS)
: next;
});
} catch (error) {
setErrorMessage(
error instanceof Error
? `Observer event parse failed: ${error.message}`
: "Observer event parse failed.",
);
}
};
source.onerror = () => {
setConnectionState((current) =>
current === "open" ? "closed" : "error",
);
setErrorMessage("Observer stream is not available right now.");
};
return () => {
source.close();
setConnectionState("closed");
};
}, [enabled, observerUrl]);
return { connectionState, errorMessage, events };
}
@@ -0,0 +1,150 @@
import type * as React from "react";
import { Activity, Bot, CircleDot, Octagon, X } from "lucide-react";
import { toast } from "sonner";
import { ManagedAgentSessionPanel } from "@/features/agents/ui/ManagedAgentSessionPanel";
import { cancelManagedAgentTurn } from "@/shared/api/agentControl";
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";
import { Tooltip, TooltipContent, TooltipTrigger } from "@/shared/ui/tooltip";
type AgentSessionThreadPanelProps = {
agent: ManagedAgent;
canResetWidth: boolean;
channel: Channel;
isWorking: boolean;
onClose: () => void;
onResetWidth: () => void;
onResizeStart: (event: React.PointerEvent<HTMLButtonElement>) => void;
widthPx: number;
};
export function AgentSessionThreadPanel({
agent,
canResetWidth,
channel,
isWorking,
onClose,
onResetWidth,
onResizeStart,
widthPx,
}: AgentSessionThreadPanelProps) {
const isLive = agent.status === "running" && Boolean(agent.observerUrl);
const { ref: scrollRef, onScroll } = useStickToBottom<HTMLDivElement>();
async function handleInterruptTurn() {
try {
const result = await cancelManagedAgentTurn(agent.pubkey, channel.id);
if (result.status === "sent") {
toast.success(`Stop signal sent to ${agent.name}.`);
} else {
toast.info(`${agent.name} has no active turn in this channel.`);
}
} catch (error) {
toast.error(
error instanceof Error
? error.message
: `Failed to stop ${agent.name}'s current turn.`,
);
}
}
return (
<aside
className="relative hidden h-full shrink-0 flex-col border-l border-border/80 bg-background lg:flex"
data-testid="agent-session-thread-panel"
style={{ width: `${widthPx}px` }}
>
<button
aria-label="Resize agent session panel"
className="group absolute inset-y-0 left-0 z-20 w-3 -translate-x-1/2 cursor-col-resize"
data-testid="agent-session-resize-handle"
onDoubleClick={canResetWidth ? onResetWidth : undefined}
onPointerDown={onResizeStart}
title={
canResetWidth
? "Drag to resize. Double-click to reset width."
: "Drag to resize."
}
type="button"
>
<span className="absolute inset-y-0 left-1/2 w-px -translate-x-1/2 bg-transparent transition-colors group-hover:bg-border/80" />
</button>
<div className="flex items-center gap-3 border-b border-border/70 px-4 py-2.5">
<Bot className="h-4 w-4 shrink-0 text-muted-foreground" />
<div className="min-w-0 flex-1">
<h2 className="truncate text-sm font-semibold tracking-tight">
{agent.name}
</h2>
<div className="flex items-center gap-1.5">
<Activity className="h-3 w-3 text-muted-foreground" />
<p className="truncate text-xs text-muted-foreground">
Agent activity log
</p>
</div>
</div>
{isLive ? (
<Badge className="shrink-0 gap-1" variant="default">
<CircleDot className="h-3 w-3" />
Live
</Badge>
) : (
<Badge className="shrink-0" variant="secondary">
Idle
</Badge>
)}
<Tooltip>
<TooltipTrigger asChild>
<Button
aria-label="Stop current agent turn"
data-testid="agent-session-stop-turn"
disabled={!isLive || !isWorking}
onClick={() => {
void handleInterruptTurn();
}}
size="sm"
type="button"
variant="outline"
>
<Octagon className="h-3.5 w-3.5" />
Stop
</Button>
</TooltipTrigger>
<TooltipContent side="bottom" className="text-xs">
{isWorking
? "Interrupt the current ACP turn without stopping the agent process."
: "No active turn to interrupt."}
</TooltipContent>
</Tooltip>
<Button
aria-label="Close activity panel"
data-testid="agent-session-close"
onClick={onClose}
size="icon"
type="button"
variant="ghost"
>
<X className="h-4 w-4" />
</Button>
</div>
<div
ref={scrollRef}
onScroll={onScroll}
className="min-h-0 flex-1 overflow-y-auto px-3 py-4"
>
<ManagedAgentSessionPanel
agent={agent}
channelId={channel.id}
className="border-0 bg-transparent p-0 shadow-none"
emptyDescription={`Mention ${agent.name} in the channel to see its work here.`}
showHeader={false}
showRaw={false}
/>
</div>
</aside>
);
}
@@ -0,0 +1,157 @@
import { Bot, Loader2 } from "lucide-react";
import type { ManagedAgent } from "@/shared/api/types";
import { cn } from "@/shared/lib/cn";
import {
DropdownMenu,
DropdownMenuContent,
DropdownMenuItem,
DropdownMenuLabel,
DropdownMenuTrigger,
} from "@/shared/ui/dropdown-menu";
import { Tooltip, TooltipContent, TooltipTrigger } from "@/shared/ui/tooltip";
type BotActivityBarProps = {
agents: ManagedAgent[];
onOpenAgentSession: (pubkey: string) => void;
openAgentSessionPubkey: string | null;
typingBotPubkeys: string[];
};
const COMPACT_THRESHOLD = 4;
const OVERFLOW_THRESHOLD = 6;
const MAX_VISIBLE_WITH_OVERFLOW = 5;
/**
* 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;
}
const { hiddenAgents, visibleAgents } = splitVisibleAgents(
typingAgents,
openAgentSessionPubkey,
);
const isCompact = typingAgents.length >= COMPACT_THRESHOLD;
return (
<div
className="flex min-w-0 shrink items-center justify-end gap-1 overflow-hidden"
data-testid="bot-activity-bar"
>
{visibleAgents.map((agent) => {
const isSelected =
openAgentSessionPubkey?.toLowerCase() === agent.pubkey.toLowerCase();
return (
<Tooltip key={agent.pubkey}>
<TooltipTrigger asChild>
<button
className={cn(
"inline-flex min-w-0 shrink items-center gap-1 rounded-full border py-1 text-xs font-medium transition-colors",
isCompact ? "max-w-[6.5rem] px-2" : "max-w-[9rem] px-2.5",
isSelected
? "border-primary/40 bg-primary/10 text-primary"
: "border-border/60 bg-background text-muted-foreground hover:border-primary/30 hover:bg-primary/5 hover:text-foreground",
)}
data-testid={`bot-chip-${agent.pubkey}`}
onClick={() => onOpenAgentSession(agent.pubkey)}
type="button"
>
<Bot className="h-3 w-3 shrink-0" />
<span className="min-w-0 truncate">{agent.name}</span>
<Loader2 className="h-3 w-3 shrink-0 animate-spin opacity-60" />
</button>
</TooltipTrigger>
<TooltipContent side="top" className="text-xs">
{agent.name} is working click to view activity
</TooltipContent>
</Tooltip>
);
})}
{hiddenAgents.length > 0 ? (
<DropdownMenu>
<DropdownMenuTrigger asChild>
<button
aria-label={`Show ${hiddenAgents.length} more working agents`}
className="inline-flex shrink-0 items-center gap-1 rounded-full border border-border/60 bg-background px-2 py-1 text-xs font-medium text-muted-foreground transition-colors hover:border-primary/30 hover:bg-primary/5 hover:text-foreground data-[state=open]:border-primary/40 data-[state=open]:bg-primary/10 data-[state=open]:text-primary"
data-testid="bot-chip-overflow"
type="button"
>
+{hiddenAgents.length}
</button>
</DropdownMenuTrigger>
<DropdownMenuContent align="end" className="w-56">
<DropdownMenuLabel className="px-2 py-1 text-xs text-muted-foreground">
More agents working
</DropdownMenuLabel>
{hiddenAgents.map((agent) => (
<DropdownMenuItem
className="cursor-pointer"
data-testid={`bot-chip-overflow-item-${agent.pubkey}`}
key={agent.pubkey}
onClick={() => onOpenAgentSession(agent.pubkey)}
>
<Bot className="h-4 w-4 text-muted-foreground" />
<span className="min-w-0 flex-1 truncate">{agent.name}</span>
<Loader2 className="h-3.5 w-3.5 animate-spin text-muted-foreground/70" />
</DropdownMenuItem>
))}
</DropdownMenuContent>
</DropdownMenu>
) : null}
</div>
);
}
function splitVisibleAgents(
typingAgents: ManagedAgent[],
openAgentSessionPubkey: string | null,
): { visibleAgents: ManagedAgent[]; hiddenAgents: ManagedAgent[] } {
if (typingAgents.length < OVERFLOW_THRESHOLD) {
return { visibleAgents: typingAgents, hiddenAgents: [] };
}
const selectedAgent = openAgentSessionPubkey
? typingAgents.find(
(agent) =>
agent.pubkey.toLowerCase() === openAgentSessionPubkey.toLowerCase(),
)
: null;
const visibleAgents = typingAgents.slice(0, MAX_VISIBLE_WITH_OVERFLOW);
if (
selectedAgent &&
!visibleAgents.some((agent) => agent.pubkey === selectedAgent.pubkey)
) {
visibleAgents[visibleAgents.length - 1] = selectedAgent;
}
const visibleSet = new Set(visibleAgents.map((agent) => agent.pubkey));
const hiddenAgents = typingAgents.filter(
(agent) => !visibleSet.has(agent.pubkey),
);
return { visibleAgents, hiddenAgents };
}
@@ -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 ? (
<EphemeralChannelBadge
display={ephemeralDisplay}
testId="chat-ephemeral-badge"
variant="header"
/>
) : null;
if (channelType === "dm" && presenceStatus) {
return (
<>
<PresenceBadge
data-testid="chat-presence-badge"
status={presenceStatus}
/>
{ephemeralBadge}
</>
);
}
return ephemeralBadge;
}
@@ -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<typeof useChannelFind>;
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<void>;
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<string, string>;
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 (
<div className="flex min-h-0 flex-1 overflow-hidden">
<div className="flex min-w-0 flex-1 flex-col overflow-hidden">
@@ -267,12 +289,22 @@ export const ChannelPane = React.memo(function ChannelPane({
: "Select a channel"
}
/>
<TypingIndicatorRow
channel={activeChannel}
currentPubkey={currentPubkey}
profiles={profiles}
typingPubkeys={typingPubkeys}
/>
<div className="relative bg-background">
<TypingIndicatorRow
channel={activeChannel}
currentPubkey={currentPubkey}
profiles={profiles}
typingPubkeys={typingPubkeys}
/>
<div className="absolute right-0 top-0 flex h-8 items-center pr-8 sm:pr-10">
<BotActivityBar
agents={agentSessionAgents}
onOpenAgentSession={onOpenAgentSession}
openAgentSessionPubkey={openAgentSessionPubkey}
typingBotPubkeys={botTypingPubkeys}
/>
</div>
</div>
</div>
{threadHeadMessage ? (
@@ -307,6 +339,20 @@ export const ChannelPane = React.memo(function ChannelPane({
threadReplies={threadMessages}
threadTypingPubkeys={threadTypingPubkeys}
/>
) : activeChannel && selectedAgent ? (
<AgentSessionThreadPanel
agent={selectedAgent}
canResetWidth={canResetThreadPanelWidth}
channel={activeChannel}
isWorking={botTypingPubkeys.some(
(pubkey) =>
pubkey.toLowerCase() === selectedAgent.pubkey.toLowerCase(),
)}
onClose={onCloseAgentSession}
onResetWidth={handleThreadPanelWidthReset}
onResizeStart={handleThreadPanelResizeStart}
widthPx={threadPanelWidthPx}
/>
) : null}
</div>
);
@@ -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,21 @@ export function ChannelScreen({
enabled: messageProfilePubkeys.length > 0,
});
const managedAgentsQuery = useManagedAgentsQuery();
const { humanTypingPubkeys, botTypingPubkeys } = React.useMemo(() => {
const localAgentSet = new Set(
(managedAgentsQuery.data ?? [])
.filter((agent) => agent.backend.type === "local")
.map((agent) => agent.pubkey.toLowerCase()),
);
return {
humanTypingPubkeys: mainTypingPubkeys.filter(
(pk) => !localAgentSet.has(pk.toLowerCase()),
),
botTypingPubkeys: mainTypingPubkeys.filter((pk) =>
localAgentSet.has(pk.toLowerCase()),
),
};
}, [mainTypingPubkeys, managedAgentsQuery.data]);
const messageProfiles = React.useMemo(() => {
const base =
mergeCurrentProfileIntoLookup(
@@ -322,8 +330,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 +362,10 @@ export function ChannelScreen({
setExpandedThreadReplyIds(new Set());
setThreadScrollTargetId(null);
setThreadReplyTargetId(null);
handleCloseAgentSession();
setEditTargetId(null);
},
[],
[handleCloseAgentSession],
);
const handleThreadScrollTargetResolved = React.useCallback(() => {
setThreadScrollTargetId(null);
@@ -377,45 +404,16 @@ export function ChannelScreen({
useLoadMissingAncestors(activeChannel, resolvedMessages);
const activeChannelEphemeralBadge = activeChannelEphemeralDisplay ? (
<EphemeralChannelBadge
display={activeChannelEphemeralDisplay}
testId="chat-ephemeral-badge"
variant="header"
/>
) : null;
const headerStatusBadge =
activeChannel?.channelType === "dm" && activeDmPresenceStatus ? (
<>
<PresenceBadge
data-testid="chat-presence-badge"
status={activeDmPresenceStatus}
/>
{activeChannelEphemeralBadge}
</>
) : (
activeChannelEphemeralBadge
);
return (
<>
<ChatHeader
actions={
activeChannel ? (
<ChannelMembersBar
channel={activeChannel}
currentPubkey={currentPubkey}
onManageChannel={openChannelManagement}
onToggleMembers={() => setIsMembersSidebarOpen((prev) => !prev)}
/>
) : null
}
channelType={activeChannel?.channelType}
visibility={activeChannel?.visibility}
description={channelDescription}
statusBadge={headerStatusBadge}
title={activeChannelTitle}
<AgentSessionProvider onOpenAgentSession={handleOpenAgentSession}>
<ChannelScreenHeader
activeChannel={activeChannel}
activeChannelEphemeralDisplay={activeChannelEphemeralDisplay}
activeChannelTitle={activeChannelTitle}
activeDmPresenceStatus={activeDmPresenceStatus}
currentPubkey={currentPubkey}
onManageChannel={openChannelManagement}
onToggleMembers={() => setIsMembersSidebarOpen((prev) => !prev)}
/>
<div className="flex min-h-0 min-w-0 flex-1 flex-col overflow-hidden">
@@ -435,6 +433,8 @@ export function ChannelScreen({
<React.Suspense fallback={<ViewLoadingFallback kind="channel" />}>
<ChannelPane
activeChannel={activeChannel}
agentSessionAgents={channelAgentSessionAgents}
botTypingPubkeys={botTypingPubkeys}
channelFind={channelFind}
currentPubkey={currentPubkey}
fetchOlder={fetchOlder}
@@ -454,17 +454,20 @@ export function ChannelScreen({
messages={timelineMessages}
onCancelEdit={handleCancelEdit}
onCancelThreadReply={handleCancelThreadReply}
onCloseAgentSession={handleCloseAgentSession}
onCloseThread={handleCloseThread}
onDelete={handleDelete}
onEdit={handleEdit}
onEditSave={handleEditSave}
onExpandThreadReplies={handleExpandThreadReplies}
onOpenThread={handleOpenThread}
onOpenAgentSession={handleOpenAgentSession}
onOpenThread={handleOpenThreadAndCloseAgentSession}
onSelectThreadReplyTarget={handleSelectThreadReplyTarget}
onSendMessage={handleSendMessage}
onSendThreadReply={handleSendThreadReply}
onThreadScrollTargetResolved={handleThreadScrollTargetResolved}
onToggleReaction={effectiveToggleReaction}
openAgentSessionPubkey={openAgentSessionPubkey}
openThreadHeadId={openThreadHeadId}
personaLookup={personaLookup}
profiles={messageProfiles}
@@ -475,16 +478,12 @@ export function ChannelScreen({
threadReplyTargetId={threadReplyTargetId}
threadReplyTargetMessage={threadReplyTargetMessage}
threadScrollTargetId={threadScrollTargetId}
typingPubkeys={mainTypingPubkeys}
typingPubkeys={humanTypingPubkeys}
/>
</React.Suspense>
)
) : (
<div className="flex min-h-0 flex-1 items-center justify-center px-6 py-8">
<p className="text-sm text-muted-foreground">
Select a channel to view messages.
</p>
</div>
<ChannelScreenEmptyState />
)}
</div>
@@ -493,7 +492,8 @@ export function ChannelScreen({
currentPubkey={currentPubkey}
open={isMembersSidebarOpen}
onOpenChange={setIsMembersSidebarOpen}
onViewActivity={handleOpenAgentSession}
/>
</>
</AgentSessionProvider>
);
}
@@ -0,0 +1,9 @@
export function ChannelScreenEmptyState() {
return (
<div className="flex min-h-0 flex-1 items-center justify-center px-6 py-8">
<p className="text-sm text-muted-foreground">
Select a channel to view messages.
</p>
</div>
);
}
@@ -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 (
<ChatHeader
actions={
activeChannel ? (
<ChannelMembersBar
channel={activeChannel}
currentPubkey={currentPubkey}
onManageChannel={onManageChannel}
onToggleMembers={onToggleMembers}
/>
) : null
}
channelType={activeChannel?.channelType}
description={getChannelDescription(activeChannel)}
statusBadge={
<ChannelHeaderStatusBadge
channelType={activeChannel?.channelType}
ephemeralDisplay={activeChannelEphemeralDisplay}
presenceStatus={activeDmPresenceStatus}
/>
}
title={activeChannelTitle}
visibility={activeChannel?.visibility}
/>
);
}
@@ -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 };
});
@@ -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
}
@@ -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({
<MemberActionsMenu
canChangeRole={canChangeRole}
canRemoveMember={canRemoveMember}
canViewActivity={canViewActivity}
disabled={disabled}
managedAgent={managedAgent}
member={member}
@@ -142,6 +150,7 @@ export function MembersSidebarMemberCard({
onChangeRole={onChangeRole}
onManagedAgentAction={onManagedAgentAction}
onRemoveMember={onRemoveMember}
onViewActivity={onViewActivity}
/>
) : null}
</div>
@@ -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 ? (
<DropdownMenuItem
data-testid={`sidebar-view-activity-${member.pubkey}`}
onClick={() => onViewActivity?.(member.pubkey)}
>
<Activity className="h-4 w-4" />
View activity
</DropdownMenuItem>
) : null}
{memberIsBot && managedAgent ? (
<>
{canViewActivity ? <DropdownMenuSeparator /> : null}
<DropdownMenuItem
data-testid={`sidebar-agent-action-${member.pubkey}`}
disabled={disabled}
@@ -0,0 +1,176 @@
import * as React from "react";
import type { TimelineMessage } from "@/features/messages/types";
import type { Channel, ChannelMember, ManagedAgent } from "@/shared/api/types";
import { normalizePubkey } from "@/shared/lib/pubkey";
type UseChannelAgentSessionsOptions = {
activeChannel: Channel | null;
activeChannelId: string | null;
channelMembers?: ChannelMember[];
handleOpenThread: (message: TimelineMessage) => void;
managedAgents: ManagedAgent[];
setExpandedThreadReplyIds: (value: Set<string>) => 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<string | null>(null);
const channelAgentSessionAgents = React.useMemo<ManagedAgent[]>(() => {
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<string>();
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,
};
}
@@ -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({
</p>
) : null}
{canViewActivity ? (
<button
className="flex w-full items-center gap-2 rounded-lg border border-border/60 px-3 py-2 text-left text-xs font-medium text-foreground transition-colors hover:bg-muted/50"
data-testid={`user-profile-view-activity-${pubkey}`}
onClick={() => {
setOpen(false);
onOpenAgentSession?.(pubkey);
}}
type="button"
>
<Activity className="h-3.5 w-3.5 text-muted-foreground" />
View activity log
</button>
) : null}
<p className="truncate font-mono text-[10px] text-muted-foreground/60">
{truncatePubkey(pubkey)}
</p>
+12
View File
@@ -0,0 +1,12 @@
import { invokeTauri } from "@/shared/api/tauri";
import type { CancelManagedAgentTurnResult } from "@/shared/api/types";
export async function cancelManagedAgentTurn(
pubkey: string,
channelId: string,
): Promise<CancelManagedAgentTurnResult> {
return invokeTauri<CancelManagedAgentTurnResult>(
"cancel_managed_agent_turn",
{ pubkey, channelId },
);
}
+2
View File
@@ -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;
@@ -862,6 +863,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,
+5
View File
@@ -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;
@@ -373,6 +374,10 @@ export type ManagedAgentLog = {
logPath: string;
};
export type CancelManagedAgentTurnResult = {
status: "sent" | "no_active_turn";
};
export type AcpProvider = {
id: string;
label: string;
@@ -0,0 +1,32 @@
import * as React from "react";
type AgentSessionContextValue = {
onOpenAgentSession: ((pubkey: string) => void) | null;
};
const AgentSessionContext = React.createContext<AgentSessionContextValue>({
onOpenAgentSession: null,
});
export function AgentSessionProvider({
children,
onOpenAgentSession,
}: {
children: React.ReactNode;
onOpenAgentSession: (pubkey: string) => void;
}) {
const value = React.useMemo(
() => ({ onOpenAgentSession }),
[onOpenAgentSession],
);
return (
<AgentSessionContext.Provider value={value}>
{children}
</AgentSessionContext.Provider>
);
}
export function useAgentSession() {
return React.useContext(AgentSessionContext);
}
@@ -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<T extends HTMLElement = HTMLDivElement>() {
const ref = useRef<T>(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 };
}
+11
View File
@@ -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,
@@ -3644,6 +3646,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,
@@ -3690,6 +3695,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;
@@ -3705,6 +3713,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}`);
@@ -4489,6 +4498,8 @@ export function maybeInstallE2eTauriMocks() {
return handleStopManagedAgent(
payload as Parameters<typeof handleStopManagedAgent>[0],
);
case "cancel_managed_agent_turn":
return { status: "sent" };
case "set_managed_agent_start_on_app_launch":
return handleSetManagedAgentStartOnAppLaunch(
payload as Parameters<