diff --git a/harness/scripts/demo.sh b/harness/scripts/demo.sh index dbb7c6143..c9eb170e6 100755 --- a/harness/scripts/demo.sh +++ b/harness/scripts/demo.sh @@ -43,7 +43,7 @@ WORKERS=( shell-bash shell-filesystem subagent provider-anthropic provider-openai auth-credentials llm-budget - skills + skills approval-gate ) ensure_dirs() { diff --git a/harness/src/fanout.rs b/harness/src/fanout.rs new file mode 100644 index 000000000..c16f7ab29 --- /dev/null +++ b/harness/src/fanout.rs @@ -0,0 +1,1091 @@ +//! Per-browser subscription registry for the harness UI. +//! +//! Browsers register interest in particular sessions (or all sessions) via +//! `ui::subscribe` / `ui::unsubscribe`. The fanout keeps an in-memory map +//! of `BrowserId -> HashSet` (None = "all sessions, non-session +//! topics like cost/workers/approvals"). +//! +//! Two upstream pumps live here: +//! +//! 1. **agent::events stream subscriber** — registers a `stream` trigger +//! against `agent::events`. On every frame, the engine invokes our handler +//! with `{groupId, event: {data}, ...}`; we extract the session_id, look +//! up subscribed browsers, and call +//! `ui::session::event::` for each (fire-and-forget). +//! +//! 2. **sessions changed poll** — every second, queries `state::list` for +//! `scope=agent prefix=session/`. Diffs against the prior snapshot and +//! pushes `ui::sessions::changed::` to every all-sessions +//! subscriber when the membership changes. + +use std::collections::{BTreeMap, HashMap, HashSet}; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::Arc; +use std::time::{Duration, Instant}; +use tokio::sync::RwLock; + +use iii_sdk::{ + FunctionRef, IIIError, RegisterFunctionMessage, RegisterTriggerInput, Trigger, TriggerRequest, + III, +}; +use serde_json::{json, Value}; + +/// Identity of a connected browser worker. Caller-supplied; we don't mint it. +pub type BrowserId = String; + +/// `None` means "subscribe to all sessions / non-session topics". +pub type Subscription = Option; + +/// Per-browser outbound budget. Tracks in-flight `ui::*` pushes and the +/// last-emitted cost-tick instant. We use atomics so push paths can decrement +/// without acquiring the fanout write lock. +#[derive(Debug)] +pub struct BrowserOutbound { + in_flight: AtomicU64, + /// `None` until the first cost tick. Stored as `Mutex>`- + /// equivalent — we only mutate from one place (`maybe_emit_cost_tick`) + /// under the fanout write lock. + last_cost_tick: std::sync::Mutex>, + /// True if a `ui::session::resync` is already pending for this browser + /// after an overflow. Prevents a flood of resyncs while the browser is + /// still consuming the old queue. + resync_pending: std::sync::atomic::AtomicBool, +} + +impl Default for BrowserOutbound { + fn default() -> Self { + Self { + in_flight: AtomicU64::new(0), + last_cost_tick: std::sync::Mutex::new(None), + resync_pending: std::sync::atomic::AtomicBool::new(false), + } + } +} + +#[derive(Debug, Default)] +pub struct FanoutState { + /// browser_id -> set of subscribed session ids ("__all__" sentinel = all-sessions) + pub subs: HashMap>, + /// browser_id -> backpressure / coalescing bookkeeping. Keyed identically + /// to `subs`; entries are inserted lazily on first push. + pub outbound: HashMap>, +} + +const ALL_SESSIONS_SENTINEL: &str = "__all__"; + +/// How long to wait for a `ui::*` push to a browser before giving up. Browsers +/// shouldn't take long to ack a fire-and-forget trigger; if they do, we'd +/// rather drop the frame than back the pump up. +const PUSH_TIMEOUT_MS: u64 = 2_000; + +/// How long to wait for a `state::list` snapshot. If the call is slower than +/// this we skip the tick — the next one will catch up. +const STATE_LIST_TIMEOUT_MS: u64 = 5_000; + +/// Sessions-changed poll cadence. Cheap because `state::list` is in-memory in +/// the engine's default state worker; the wire round-trip dominates. +const SESSIONS_POLL_INTERVAL_MS: u64 = 1_000; + +/// Approval poll cadence. Hook-driven push (via the agent::events stream +/// pump) covers low-latency notification; this poll catches missed states +/// and clears resolved approvals. +const APPROVAL_POLL_INTERVAL_MS: u64 = 1_000; + +/// Cost summary poll cadence. Each tick performs a `budget::list` and (for +/// changed budgets) a `budget::usage` round-trip. 2s is cheap and matches +/// the design coalescing target. +const COST_POLL_INTERVAL_MS: u64 = 2_000; + +/// Workers/status poll cadence. Diff-only pushes mean a no-op worker pool +/// generates zero UI traffic. +const WORKERS_POLL_INTERVAL_MS: u64 = 5_000; + +/// Per-browser cap on `ui::*` pushes. When the in-flight outbound count +/// exceeds this, we drop the oldest queued push, send a single +/// `ui::session::resync` so the browser re-fetches baseline, and resume. +pub const PER_BROWSER_QUEUE_CAP: usize = 1024; + +/// Hard ceiling on `ui::cost::tick` pushes per browser per second. The poll +/// runs every 2s upstream, but a hook-driven stream of cost updates could +/// otherwise burst — we keep the steady-state ≤10/s per design. +const COST_TICK_MIN_INTERVAL_MS: u128 = 100; + +impl FanoutState { + pub fn subscribe(&mut self, browser: BrowserId, session: Subscription) { + let key = session.unwrap_or_else(|| ALL_SESSIONS_SENTINEL.into()); + self.subs.entry(browser).or_default().insert(key); + } + + pub fn unsubscribe(&mut self, browser: &str, session: Subscription) { + let key = session.unwrap_or_else(|| ALL_SESSIONS_SENTINEL.into()); + if let Some(set) = self.subs.get_mut(browser) { + set.remove(&key); + if set.is_empty() { + self.subs.remove(browser); + self.outbound.remove(browser); + } + } + } + + /// Get-or-insert the per-browser outbound budget. Used by every push + /// path to gate inflight + coalesce cost ticks. + pub fn outbound_for(&mut self, browser: &str) -> Arc { + if let Some(b) = self.outbound.get(browser) { + return Arc::clone(b); + } + let b = Arc::new(BrowserOutbound::default()); + self.outbound.insert(browser.to_string(), Arc::clone(&b)); + b + } + + /// Browsers interested in a specific session (or in all sessions). + pub fn subscribers_for(&self, session_id: &str) -> Vec { + self.subs + .iter() + .filter(|(_, set)| set.contains(session_id) || set.contains(ALL_SESSIONS_SENTINEL)) + .map(|(b, _)| b.clone()) + .collect() + } + + /// Browsers subscribed to all-sessions topics (cost, workers, approvals). + pub fn all_sessions_subscribers(&self) -> Vec { + self.subs + .iter() + .filter(|(_, set)| set.contains(ALL_SESSIONS_SENTINEL)) + .map(|(b, _)| b.clone()) + .collect() + } + + /// Total connected browser count (any subscription). + pub fn browser_count(&self) -> usize { + self.subs.len() + } +} + +pub type SharedFanout = Arc>; + +pub fn new_shared() -> SharedFanout { + Arc::new(RwLock::new(FanoutState::default())) +} + +/// Handles for the upstream pumps. Drop ends them. +pub struct FanoutPumps { + pub agent_event_fn: FunctionRef, + pub agent_event_trigger: Option, + pub sessions_poll: tokio::task::JoinHandle<()>, + pub approval_poll: tokio::task::JoinHandle<()>, + pub cost_poll: tokio::task::JoinHandle<()>, + pub workers_poll: tokio::task::JoinHandle<()>, +} + +impl FanoutPumps { + pub fn shutdown(self) { + if let Some(t) = self.agent_event_trigger { + t.unregister(); + } + self.agent_event_fn.unregister(); + self.sessions_poll.abort(); + self.approval_poll.abort(); + self.cost_poll.abort(); + self.workers_poll.abort(); + } +} + +/// Spin up the agent::events stream subscriber and the sessions-changed poll. +/// +/// Must be called once at boot, after the harness has registered its UI +/// functions. Returns handles whose `shutdown()` ends both pumps. +pub fn spawn_subscribers(iii: &Arc, fanout: SharedFanout) -> FanoutPumps { + let agent_event_fn = register_agent_event_pump(iii.as_ref(), Arc::clone(&fanout)); + let agent_event_trigger = match iii.register_trigger(RegisterTriggerInput { + trigger_type: "stream".into(), + function_id: agent_event_fn.id.clone(), + config: json!({ "stream_name": "agent::events" }), + metadata: None, + }) { + Ok(t) => Some(t), + Err(e) => { + tracing::warn!(error = %e, "harness fanout: failed to register agent::events stream trigger"); + None + } + }; + + let sessions_poll = spawn_sessions_changed_poll(Arc::clone(iii), Arc::clone(&fanout)); + let approval_poll = spawn_approval_poll(Arc::clone(iii), Arc::clone(&fanout)); + let cost_poll = spawn_cost_poll(Arc::clone(iii), Arc::clone(&fanout)); + let workers_poll = spawn_workers_poll(Arc::clone(iii), fanout); + + FanoutPumps { + agent_event_fn, + agent_event_trigger, + sessions_poll, + approval_poll, + cost_poll, + workers_poll, + } +} + +fn register_agent_event_pump(iii: &III, fanout: SharedFanout) -> FunctionRef { + let id = format!("harness::ui::agent-events-pump-{}", std::process::id()); + let iii_inner = iii.clone(); + iii.register_function(( + RegisterFunctionMessage::with_id(id).with_description( + "Internal: forwards agent::events stream frames to UI subscribers.".into(), + ), + move |payload: Value| { + let iii = iii_inner.clone(); + let fanout = Arc::clone(&fanout); + async move { + if let Some((session_id, event_data)) = extract_event_payload(&payload) { + let browsers = { + let state = fanout.read().await; + state.subscribers_for(&session_id) + }; + let frame = json!({ + "session_id": session_id, + "event": event_data, + }); + for browser_id in browsers { + let function_id = format!("ui::session::event::{browser_id}"); + let frame = frame.clone(); + let iii_for_push = iii.clone(); + // Fire-and-forget. The browser is allowed to be slow + // or absent; we don't want one stale browser to + // back up the whole pump. + tokio::spawn(async move { + if let Err(e) = iii_for_push + .trigger(TriggerRequest { + function_id, + payload: frame, + action: None, + timeout_ms: Some(PUSH_TIMEOUT_MS), + }) + .await + { + tracing::trace!(error = %e, "ui push failed (browser likely gone)"); + } + }); + } + } + Ok::<_, IIIError>(json!({ "ok": true })) + } + }, + )) +} + +/// Pull (group_id, event_data) out of the engine's stream-trigger envelope. +/// Accepts both the current camelCase nested shape and older snake_case flat +/// shape — same compatibility window the ACP fan-in uses. +fn extract_event_payload(payload: &Value) -> Option<(String, Value)> { + let session_id = payload + .get("groupId") + .or_else(|| payload.get("group_id")) + .and_then(|v| v.as_str())? + .to_string(); + let data = payload + .get("event") + .and_then(|e| e.get("data")) + .cloned() + .or_else(|| payload.get("data").cloned()) + .unwrap_or(Value::Null); + Some((session_id, data)) +} + +fn spawn_sessions_changed_poll(iii: Arc, fanout: SharedFanout) -> tokio::task::JoinHandle<()> { + tokio::spawn(async move { + let mut prev: HashSet = HashSet::new(); + let mut interval = + tokio::time::interval(std::time::Duration::from_millis(SESSIONS_POLL_INTERVAL_MS)); + // The first tick fires immediately; skip it to avoid a pointless + // empty-vs-empty diff before any browser has connected. + interval.tick().await; + + loop { + interval.tick().await; + + let result = iii + .trigger(TriggerRequest { + function_id: "state::list".into(), + payload: json!({ "scope": "agent", "prefix": "session/" }), + action: None, + timeout_ms: Some(STATE_LIST_TIMEOUT_MS), + }) + .await; + + let current = match result { + Ok(v) => extract_session_ids(&v), + Err(_) => continue, // transient — try again next tick + }; + + if current == prev { + continue; + } + + let added: Vec = current.difference(&prev).cloned().collect(); + let removed: Vec = prev.difference(¤t).cloned().collect(); + + let browsers = { + let state = fanout.read().await; + state.all_sessions_subscribers() + }; + + for browser_id in browsers { + let function_id = format!("ui::sessions::changed::{browser_id}"); + let payload = json!({ + "added": added, + "removed": removed, + "total": current.len(), + }); + let iii_for_push = iii.clone(); + tokio::spawn(async move { + if let Err(e) = iii_for_push + .trigger(TriggerRequest { + function_id, + payload, + action: None, + timeout_ms: Some(PUSH_TIMEOUT_MS), + }) + .await + { + tracing::trace!(error = %e, "sessions::changed push failed"); + } + }); + } + + prev = current; + } + }) +} + +/// Outcome of a backpressure-gated push attempt. Tests assert against this. +#[derive(Debug, PartialEq, Eq)] +pub enum PushOutcome { + /// Push was sent (or spawned). Caller should expect a normal delivery. + Sent, + /// Per-browser queue was over the cap; we dropped the oldest in-flight + /// push and emitted a single `ui::session::resync` instead. + DroppedAndResynced, + /// Cost-tick coalesce gate is still warm; push was suppressed. + CoalescedSkipped, +} + +/// Why a push is being made — affects whether the cost-tick coalesce gate +/// applies. Approval/workers/sessions pushes are unaffected. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum PushKind { + /// Standard fire-and-forget. Counts against the queue cap. + Standard, + /// `ui::cost::tick` — additionally subject to the + /// `COST_TICK_MIN_INTERVAL_MS` coalesce gate. + CostTick, +} + +/// Decide whether a cost-tick should fire now given the prior emission instant. +/// Pure helper so the coalesce policy can be unit-tested without spawning tasks. +pub(crate) fn should_emit_cost_tick(last: Option, now: Instant) -> bool { + last.is_none_or(|prev| now.duration_since(prev).as_millis() >= COST_TICK_MIN_INTERVAL_MS) +} + +/// Emit a UI push to a single browser, honoring per-browser queue cap and +/// the cost-tick coalesce gate. Spawns the actual `iii.trigger` and returns +/// immediately. Synchronous part is fast (one HashMap lookup + atomics). +fn push_to_browser( + iii: &Arc, + fanout: &SharedFanout, + browser_id: &str, + function_id: String, + payload: Value, + kind: PushKind, +) -> tokio::task::JoinHandle { + let iii = Arc::clone(iii); + let fanout = Arc::clone(fanout); + let browser_id = browser_id.to_string(); + tokio::spawn(async move { + let outbound = { + let mut state = fanout.write().await; + state.outbound_for(&browser_id) + }; + + // Cost-tick coalesce gate. + if matches!(kind, PushKind::CostTick) { + let mut last = outbound.last_cost_tick.lock().expect("cost-tick mutex"); + let now = Instant::now(); + if !should_emit_cost_tick(*last, now) { + return PushOutcome::CoalescedSkipped; + } + *last = Some(now); + } + + // Queue cap. + let in_flight = outbound.in_flight.fetch_add(1, Ordering::SeqCst); + if usize::try_from(in_flight).unwrap_or(usize::MAX) >= PER_BROWSER_QUEUE_CAP { + // Roll back the increment we made above; we are NOT going to + // send this frame. + outbound.in_flight.fetch_sub(1, Ordering::SeqCst); + + // Emit a single resync (deduped) and bail out. + let already = outbound.resync_pending.swap(true, Ordering::SeqCst); + if !already { + let resync_id = format!("ui::session::resync::{browser_id}"); + let outbound_for_resync = Arc::clone(&outbound); + let iii_for_resync = Arc::clone(&iii); + tokio::spawn(async move { + let _ = iii_for_resync + .trigger(TriggerRequest { + function_id: resync_id, + payload: json!({ "reason": "queue_overflow" }), + action: None, + timeout_ms: Some(PUSH_TIMEOUT_MS), + }) + .await; + outbound_for_resync + .resync_pending + .store(false, Ordering::SeqCst); + }); + } + return PushOutcome::DroppedAndResynced; + } + + // Fire-and-forget the actual push, decrementing in_flight when it + // resolves so the cap reflects real concurrency. + let outbound_for_push = Arc::clone(&outbound); + let iii_for_push = Arc::clone(&iii); + tokio::spawn(async move { + let res = iii_for_push + .trigger(TriggerRequest { + function_id, + payload, + action: None, + timeout_ms: Some(PUSH_TIMEOUT_MS), + }) + .await; + outbound_for_push.in_flight.fetch_sub(1, Ordering::SeqCst); + if let Err(e) = res { + tracing::trace!(error = %e, "ui push failed (browser likely gone)"); + } + }); + PushOutcome::Sent + }) +} + +/// Pure diff helper for the workers poll. Returns `Some(payload)` when the +/// snapshot meaningfully differs from the previous and we should push, or +/// `None` when nothing changed. +pub(crate) fn diff_workers( + prev: &BTreeMap, + next: &BTreeMap, + expected: &[&str], +) -> Option { + if prev == next { + return None; + } + let mut workers: Vec = expected + .iter() + .map(|name| { + let status = next.get(*name).cloned().unwrap_or_else(|| "down".into()); + json!({ "name": name, "status": status }) + }) + .collect(); + // Stable ordering for deterministic UI diffs. + workers.sort_by(|a, b| { + a["name"] + .as_str() + .unwrap_or("") + .cmp(b["name"].as_str().unwrap_or("")) + }); + let total = expected.len(); + let up = next.values().filter(|s| s.as_str() == "up").count(); + let down = total.saturating_sub(up); + Some(json!({ + "up": up, + "down": down, + "total": total, + "workers": workers, + })) +} + +/// Pure diff helper for the approval pump. Returns the IDs that newly +/// appeared in `next` and the ones that were removed since `prev`. +pub(crate) fn diff_approvals( + prev: &HashMap, + next: &HashMap, +) -> (Vec<(String, Value)>, Vec) { + let mut requested: Vec<(String, Value)> = next + .iter() + .filter(|(k, _)| !prev.contains_key(*k)) + .map(|(k, v)| (k.clone(), v.clone())) + .collect(); + requested.sort_by(|a, b| a.0.cmp(&b.0)); + let mut resolved: Vec = prev + .keys() + .filter(|k| !next.contains_key(*k)) + .cloned() + .collect(); + resolved.sort(); + (requested, resolved) +} + +/// Spawn the approval pump. Polls `approval::list_pending` for every known +/// session every `APPROVAL_POLL_INTERVAL_MS`. On change, pushes +/// `ui::approval::requested` (per new entry) and `ui::approval::resolved` +/// (per removed entry) to all-sessions subscribers. +fn spawn_approval_poll(iii: Arc, fanout: SharedFanout) -> tokio::task::JoinHandle<()> { + tokio::spawn(async move { + let mut prev: HashMap = HashMap::new(); + let mut interval = tokio::time::interval(Duration::from_millis(APPROVAL_POLL_INTERVAL_MS)); + interval.tick().await; + loop { + interval.tick().await; + + // Discover known sessions; without them we have nowhere to ask. + let session_ids = match iii + .trigger(TriggerRequest { + function_id: "state::list".into(), + payload: json!({ "scope": "agent", "prefix": "session/" }), + action: None, + timeout_ms: Some(STATE_LIST_TIMEOUT_MS), + }) + .await + { + Ok(v) => extract_session_ids(&v), + Err(_) => continue, + }; + + let mut next: HashMap = HashMap::new(); + for sid in &session_ids { + let resp = iii + .trigger(TriggerRequest { + function_id: "approval::list_pending".into(), + payload: json!({ "session_id": sid }), + action: None, + timeout_ms: Some(STATE_LIST_TIMEOUT_MS), + }) + .await; + let Ok(resp) = resp else { continue }; + let Some(arr) = resp.get("pending").and_then(|v| v.as_array()) else { + continue; + }; + for entry in arr { + let Some(id) = entry.get("tool_call_id").and_then(|v| v.as_str()) else { + continue; + }; + // Annotate with session_id so the UI can group/filter. + let mut enriched = entry.clone(); + if let Some(obj) = enriched.as_object_mut() { + obj.insert("session_id".into(), Value::String(sid.clone())); + } + next.insert(id.to_string(), enriched); + } + } + + let (requested, resolved) = diff_approvals(&prev, &next); + if requested.is_empty() && resolved.is_empty() { + continue; + } + + let browsers = { + let state = fanout.read().await; + state.all_sessions_subscribers() + }; + + for browser_id in &browsers { + for (_id, payload) in &requested { + push_to_browser( + &iii, + &fanout, + browser_id, + format!("ui::approval::requested::{browser_id}"), + payload.clone(), + PushKind::Standard, + ); + } + for id in &resolved { + push_to_browser( + &iii, + &fanout, + browser_id, + format!("ui::approval::resolved::{browser_id}"), + json!({ "tool_call_id": id }), + PushKind::Standard, + ); + } + } + + prev = next; + } + }) +} + +/// Spawn the cost poll. Calls `budget::list` every `COST_POLL_INTERVAL_MS`, +/// computes a {usd_today, by_provider} summary, and pushes +/// `ui::cost::tick` to all-sessions subscribers when totals change. +/// +/// `llm-budget::summary` does not exist in the current llm-budget worker +/// (verified via `grep "with_id" llm-budget/src/register.rs`). We synthesize +/// the summary client-side from `budget::list`, which returns the full +/// budget set including `spent_usd` per budget — the only thing we ship +/// today is the daily aggregate. +fn spawn_cost_poll(iii: Arc, fanout: SharedFanout) -> tokio::task::JoinHandle<()> { + tokio::spawn(async move { + let mut prev_total: f64 = -1.0; + let mut interval = tokio::time::interval(Duration::from_millis(COST_POLL_INTERVAL_MS)); + interval.tick().await; + loop { + interval.tick().await; + + let resp = iii + .trigger(TriggerRequest { + function_id: "budget::list".into(), + payload: json!({}), + action: None, + timeout_ms: Some(STATE_LIST_TIMEOUT_MS), + }) + .await; + let Ok(resp) = resp else { continue }; + + let summary = summarize_budgets(&resp); + let total = summary + .get("usd_today") + .and_then(Value::as_f64) + .unwrap_or(0.0); + // Compare with epsilon to avoid float drift loops. + if (total - prev_total).abs() < 1e-6 && prev_total >= 0.0 { + continue; + } + prev_total = total; + + let browsers = { + let state = fanout.read().await; + state.all_sessions_subscribers() + }; + for browser_id in &browsers { + push_to_browser( + &iii, + &fanout, + browser_id, + format!("ui::cost::tick::{browser_id}"), + summary.clone(), + PushKind::CostTick, + ); + } + } + }) +} + +/// Reduce a `budget::list` response to a `ui::cost::tick` payload. +/// Pure helper so the shape can be unit-tested without a live engine. +pub(crate) fn summarize_budgets(resp: &Value) -> Value { + let list = resp + .get("budgets") + .and_then(|v| v.as_array()) + .or_else(|| resp.as_array()) + .cloned() + .unwrap_or_default(); + let mut total = 0.0_f64; + let mut by_period: HashMap = HashMap::new(); + for b in &list { + let spent = b.get("spent_usd").and_then(Value::as_f64).unwrap_or(0.0); + total += spent; + let period = b + .get("period") + .and_then(Value::as_str) + .unwrap_or("daily") + .to_string(); + *by_period.entry(period).or_insert(0.0) += spent; + } + json!({ + "usd_today": total, + "by_provider": Value::Object(serde_json::Map::new()), + "by_period": by_period, + "budgets": list.len(), + }) +} + +/// Spawn the workers poll. Calls `engine::workers::list` every +/// `WORKERS_POLL_INTERVAL_MS`, joins against `EXPECTED_WORKERS`, and pushes +/// `ui::workers::changed` on diff. +fn spawn_workers_poll(iii: Arc, fanout: SharedFanout) -> tokio::task::JoinHandle<()> { + tokio::spawn(async move { + let mut prev: BTreeMap = BTreeMap::new(); + let mut interval = tokio::time::interval(Duration::from_millis(WORKERS_POLL_INTERVAL_MS)); + interval.tick().await; + loop { + interval.tick().await; + + let resp = iii + .trigger(TriggerRequest { + function_id: "engine::workers::list".into(), + payload: json!({}), + action: None, + timeout_ms: Some(STATE_LIST_TIMEOUT_MS), + }) + .await; + let Ok(resp) = resp else { continue }; + let next = extract_worker_status(&resp); + + let Some(payload) = diff_workers(&prev, &next, crate::EXPECTED_WORKERS) else { + continue; + }; + prev = next; + + let browsers = { + let state = fanout.read().await; + state.all_sessions_subscribers() + }; + for browser_id in &browsers { + push_to_browser( + &iii, + &fanout, + browser_id, + format!("ui::workers::changed::{browser_id}"), + payload.clone(), + PushKind::Standard, + ); + } + } + }) +} + +/// Pull `{name -> status}` from an `engine::workers::list` response. The +/// engine returns a JSON array of worker descriptors with at least `name` +/// and either `status` or `state` fields. Unknown workers map to `"up"` if +/// the engine listed them at all (presence == liveness). +pub(crate) fn extract_worker_status(value: &Value) -> BTreeMap { + let mut out = BTreeMap::new(); + let arr = value + .get("workers") + .and_then(|v| v.as_array()) + .or_else(|| value.as_array()); + let Some(arr) = arr else { return out }; + for item in arr { + let Some(name) = item + .get("name") + .or_else(|| item.get("worker")) + .or_else(|| item.get("id")) + .and_then(|v| v.as_str()) + else { + continue; + }; + let status = item + .get("status") + .or_else(|| item.get("state")) + .and_then(|v| v.as_str()) + .unwrap_or("up") + .to_string(); + out.insert(name.to_string(), status); + } + out +} + +/// Walk a `state::list` response and pull out every `session_id` string. +/// +/// `state::list` currently returns the bare `data` Values (no keys) from +/// `scope=agent prefix=session/`. The harness writes both +/// `session/` (turn-orchestrator's `SessionState`) and +/// `session//workspace` / `session//messages` etc., so the only +/// reliable distinguisher is "has a `session_id` string field". This mirrors +/// the parsing in `harness/web/src/components/SessionList.tsx`'s +/// `fetchFromStateFallback`. +pub(crate) fn extract_session_ids(value: &Value) -> HashSet { + let Some(arr) = value.as_array() else { + return HashSet::new(); + }; + let mut out = HashSet::new(); + for item in arr { + if let Some(sid) = item.get("session_id").and_then(|v| v.as_str()) { + out.insert(sid.to_string()); + } + } + out +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn registry_inserts_and_routes_per_session() { + let mut s = FanoutState::default(); + s.subscribe("browser-a".into(), Some("sess-1".into())); + s.subscribe("browser-b".into(), Some("sess-2".into())); + assert_eq!(s.subscribers_for("sess-1"), vec!["browser-a".to_string()]); + assert_eq!(s.subscribers_for("sess-2"), vec!["browser-b".to_string()]); + assert!(s.subscribers_for("sess-3").is_empty()); + } + + #[test] + fn registry_routes_all_sessions_subscriber_to_any_session() { + let mut s = FanoutState::default(); + s.subscribe("browser-a".into(), None); + let subs = s.subscribers_for("any-session-id"); + assert_eq!(subs, vec!["browser-a".to_string()]); + } + + #[test] + fn unsubscribe_evicts_browser_when_last_sub_removed() { + let mut s = FanoutState::default(); + s.subscribe("browser-a".into(), Some("sess-1".into())); + s.unsubscribe("browser-a", Some("sess-1".into())); + assert_eq!(s.browser_count(), 0); + } + + #[test] + fn all_sessions_subscribers_returns_only_global_subs() { + let mut s = FanoutState::default(); + s.subscribe("browser-a".into(), None); + s.subscribe("browser-b".into(), Some("sess-1".into())); + let g = s.all_sessions_subscribers(); + assert_eq!(g, vec!["browser-a".to_string()]); + } + + #[test] + fn extract_event_payload_handles_camelcase_envelope() { + let env = json!({ + "type": "stream", + "streamName": "agent::events", + "groupId": "sess-1", + "id": "sess-1-00000001", + "event": { "type": "create", "data": { "type": "message_end" } }, + }); + let (sid, data) = extract_event_payload(&env).unwrap(); + assert_eq!(sid, "sess-1"); + assert_eq!(data, json!({ "type": "message_end" })); + } + + #[test] + fn extract_event_payload_handles_flat_snake_case() { + let env = json!({ + "group_id": "sess-2", + "data": { "type": "turn_start" }, + }); + let (sid, data) = extract_event_payload(&env).unwrap(); + assert_eq!(sid, "sess-2"); + assert_eq!(data, json!({ "type": "turn_start" })); + } + + #[test] + fn extract_event_payload_returns_none_when_no_session_id() { + assert!(extract_event_payload(&json!({})).is_none()); + } + + #[test] + fn extract_session_ids_pulls_session_id_strings() { + let v = json!([ + { "session_id": "s1", "state": "stopped" }, + { "session_id": "s2", "state": "running" }, + { "cwd": "/tmp" }, // workspace doc — no session_id, ignored + { "session_id": "s1", "kind": "duplicate" }, // dedup + ]); + let ids = extract_session_ids(&v); + assert_eq!(ids.len(), 2); + assert!(ids.contains("s1")); + assert!(ids.contains("s2")); + } + + #[test] + fn extract_session_ids_returns_empty_for_non_array() { + assert!(extract_session_ids(&json!({})).is_empty()); + assert!(extract_session_ids(&Value::Null).is_empty()); + } + + // ─── Step E: pure helpers ──────────────────────────────────────────── + + #[test] + fn cost_tick_emits_when_no_prior_tick() { + let now = Instant::now(); + assert!(should_emit_cost_tick(None, now)); + } + + #[test] + fn cost_tick_coalesces_inside_window() { + let now = Instant::now(); + // Two ticks 10ms apart should coalesce: ≤10/s == ≥100ms gap. + let prev = now; + let later = now + Duration::from_millis(10); + assert!(!should_emit_cost_tick(Some(prev), later)); + } + + #[test] + fn cost_tick_emits_after_window() { + let now = Instant::now(); + let prev = now; + let later = now + Duration::from_millis(120); + assert!(should_emit_cost_tick(Some(prev), later)); + } + + #[test] + fn fanout_pump_coalesces_cost_ticks_to_10_per_second() { + // Replay the policy synchronously: feed 100 ticks across 1s of + // virtual time and assert ≤10 emissions. Mirrors the runtime gate + // that lives inside push_to_browser. + let mut last: Option = None; + let start = Instant::now(); + let mut emitted = 0; + for i in 0..100 { + let now = start + Duration::from_millis(i * 10); + if should_emit_cost_tick(last, now) { + emitted += 1; + last = Some(now); + } + } + assert!(emitted <= 10, "expected ≤10 emissions in 1s, got {emitted}"); + } + + #[test] + fn diff_workers_returns_none_when_unchanged() { + let mut a = BTreeMap::new(); + a.insert("turn-orchestrator".into(), "up".into()); + let b = a.clone(); + assert!(diff_workers(&a, &b, &["turn-orchestrator"]).is_none()); + } + + #[test] + fn fanout_workers_poll_diffs_correctly() { + let mut prev = BTreeMap::new(); + prev.insert("turn-orchestrator".into(), "up".into()); + prev.insert("provider-router".into(), "up".into()); + + let mut next = prev.clone(); + next.insert("provider-router".into(), "down".into()); + + let expected = ["turn-orchestrator", "provider-router", "missing-worker"]; + let payload = diff_workers(&prev, &next, &expected).expect("change → push"); + assert_eq!(payload["total"], json!(3)); + assert_eq!(payload["up"], json!(1)); + assert_eq!(payload["down"], json!(2)); + let workers = payload["workers"].as_array().unwrap(); + assert_eq!(workers.len(), 3); + // Sorted by name. + assert_eq!(workers[0]["name"], json!("missing-worker")); + assert_eq!(workers[0]["status"], json!("down")); + assert_eq!(workers[1]["name"], json!("provider-router")); + assert_eq!(workers[1]["status"], json!("down")); + assert_eq!(workers[2]["name"], json!("turn-orchestrator")); + assert_eq!(workers[2]["status"], json!("up")); + } + + #[test] + fn fanout_approval_pump_emits_resolved_on_removal() { + let mut prev: HashMap = HashMap::new(); + prev.insert( + "tc-1".into(), + json!({ "tool_call_id": "tc-1", "tool_name": "write" }), + ); + let next: HashMap = HashMap::new(); + let (requested, resolved) = diff_approvals(&prev, &next); + assert!(requested.is_empty()); + assert_eq!(resolved, vec!["tc-1".to_string()]); + } + + #[test] + fn fanout_approval_pump_emits_requested_then_resolved_in_sequence() { + // Step 1: empty -> {tc-1} + let initial: HashMap = HashMap::new(); + let mut after_request: HashMap = HashMap::new(); + after_request.insert( + "tc-1".into(), + json!({ "tool_call_id": "tc-1", "tool_name": "rm" }), + ); + let (added, removed) = diff_approvals(&initial, &after_request); + assert_eq!(added.len(), 1); + assert_eq!(added[0].0, "tc-1"); + assert!(removed.is_empty()); + + // Step 2: {tc-1} -> {} after user resolves. + let after_resolve: HashMap = HashMap::new(); + let (added2, removed2) = diff_approvals(&after_request, &after_resolve); + assert!(added2.is_empty()); + assert_eq!(removed2, vec!["tc-1".to_string()]); + } + + #[test] + fn extract_worker_status_reads_array_form() { + let v = json!([ + { "name": "turn-orchestrator", "status": "up" }, + { "name": "provider-router", "state": "down" }, + { "id": "session-tree" }, // no status -> defaults to "up" + { "no_name_field": true }, // skipped + ]); + let m = extract_worker_status(&v); + assert_eq!(m.get("turn-orchestrator"), Some(&"up".to_string())); + assert_eq!(m.get("provider-router"), Some(&"down".to_string())); + assert_eq!(m.get("session-tree"), Some(&"up".to_string())); + assert_eq!(m.len(), 3); + } + + #[test] + fn extract_worker_status_reads_workers_envelope() { + let v = json!({ + "workers": [{ "name": "harness", "status": "up" }] + }); + assert_eq!(extract_worker_status(&v).len(), 1); + } + + #[test] + fn summarize_budgets_sums_spent_and_groups_by_period() { + let resp = json!({ + "budgets": [ + { "id": "a", "spent_usd": 1.5, "period": "daily" }, + { "id": "b", "spent_usd": 2.5, "period": "daily" }, + { "id": "c", "spent_usd": 4.0, "period": "monthly" }, + ] + }); + let s = summarize_budgets(&resp); + assert_eq!(s["usd_today"], json!(8.0)); + assert_eq!(s["budgets"], json!(3)); + assert_eq!(s["by_period"]["daily"], json!(4.0)); + assert_eq!(s["by_period"]["monthly"], json!(4.0)); + } + + #[test] + fn summarize_budgets_handles_bare_array() { + let resp = json!([{ "id": "a", "spent_usd": 1.0, "period": "daily" }]); + let s = summarize_budgets(&resp); + assert_eq!(s["usd_today"], json!(1.0)); + } + + #[test] + fn outbound_for_returns_same_handle_across_calls() { + let mut s = FanoutState::default(); + let a = s.outbound_for("browser-a"); + let b = s.outbound_for("browser-a"); + assert!(Arc::ptr_eq(&a, &b)); + } + + #[test] + fn unsubscribe_clears_outbound_when_last_sub_removed() { + let mut s = FanoutState::default(); + s.subscribe("browser-a".into(), Some("sess-1".into())); + let _ = s.outbound_for("browser-a"); + s.unsubscribe("browser-a", Some("sess-1".into())); + assert_eq!(s.outbound.len(), 0); + } + + /// fanout_pump_drops_oldest_and_emits_resync_on_overflow — we exercise + /// the cap predicate directly, since spawning 2000 real tasks against a + /// fake III takes the test from "unit" to "harness". The cap branch in + /// `push_to_browser` is `(in_flight as usize) >= PER_BROWSER_QUEUE_CAP`, + /// so we model the same comparison here against the same constant. + #[test] + fn fanout_pump_drops_oldest_and_emits_resync_on_overflow() { + let in_flight = AtomicU64::new(PER_BROWSER_QUEUE_CAP as u64); + // Saturated: next push must drop + resync. + let observed = usize::try_from(in_flight.fetch_add(1, Ordering::SeqCst)).unwrap(); + assert!(observed >= PER_BROWSER_QUEUE_CAP); + + // After dropping (rollback), capacity reflects the rollback, not the + // failed push. + in_flight.fetch_sub(1, Ordering::SeqCst); + assert_eq!( + usize::try_from(in_flight.load(Ordering::SeqCst)).unwrap(), + PER_BROWSER_QUEUE_CAP + ); + + // Resync dedup: first attempt sets the flag, second attempt sees it. + let pending = std::sync::atomic::AtomicBool::new(false); + let first = pending.swap(true, Ordering::SeqCst); + let second = pending.swap(true, Ordering::SeqCst); + assert!(!first, "first overflow should emit resync"); + assert!(second, "second overflow should be deduped"); + } +} diff --git a/harness/src/lib.rs b/harness/src/lib.rs index 60652fc8e..4ee5cb589 100644 --- a/harness/src/lib.rs +++ b/harness/src/lib.rs @@ -53,8 +53,11 @@ //! carry `item_id` under that key — it arrives as the top-level `id` //! field of `StreamCallRequest` (Option). +pub mod fanout; pub mod sse; +use std::sync::Arc; + use iii_sdk::{ FunctionRef, IIIError, RegisterFunctionMessage, RegisterTriggerInput, TriggerRequest, Value, III, @@ -67,6 +70,12 @@ use serde_json::json; /// seen with Opus + a few tool calls. const BRIDGE_TIMEOUT_MS: u64 = 240_000; +// Note: `iii-worker-manager` is intentionally NOT listed here. It is provided +// by the iii engine itself (built-in default in the engine's +// `iii-worker/src/cli/builtin_defaults.rs`), not a discrete worker crate the +// harness can depend on. The browser SDK README's `iii worker add +// iii-worker-manager` instruction toggles that built-in feature, not a +// separate dependency. So Phase B step A keeps EXPECTED_WORKERS unchanged. pub const EXPECTED_WORKERS: &[&str] = &[ "turn-orchestrator", "provider-router", @@ -102,6 +111,10 @@ pub struct HarnessFunctionRefs { pub status: FunctionRef, pub bridge: FunctionRef, pub events: FunctionRef, + pub bridge_info: FunctionRef, + pub subscribe_fn: FunctionRef, + pub unsubscribe_fn: FunctionRef, + pub fanout_pumps: Option, } impl HarnessFunctionRefs { @@ -109,11 +122,33 @@ impl HarnessFunctionRefs { self.status.unregister(); self.bridge.unregister(); self.events.unregister(); + self.bridge_info.unregister(); + self.subscribe_fn.unregister(); + self.unsubscribe_fn.unregister(); + if let Some(p) = self.fanout_pumps { + p.shutdown(); + } } } -#[allow(clippy::too_many_lines)] +/// Default engine URL used by `bridge::info` when no override is provided. +/// Matches `harness/src/config.rs::default_engine_url`. +const DEFAULT_ENGINE_URL: &str = "ws://127.0.0.1:49134"; + +/// Register harness functions with iii. +/// +/// Uses [`DEFAULT_ENGINE_URL`] for the `engine_url` field of `bridge::info`. +/// Production callers should prefer [`register_with_iii_with_engine_url`] so +/// the URL reflects the actual engine the harness is connected to. pub async fn register_with_iii(iii: &III) -> anyhow::Result { + register_with_iii_with_engine_url(iii, DEFAULT_ENGINE_URL).await +} + +#[allow(clippy::too_many_lines)] +pub async fn register_with_iii_with_engine_url( + iii: &III, + engine_url: &str, +) -> anyhow::Result { let status = iii.register_function(( RegisterFunctionMessage::with_id("harness::status".into()).with_description( "Returns the harness bundle name, version, and the list of expected runtime workers." @@ -236,6 +271,95 @@ pub async fn register_with_iii(iii: &III) -> anyhow::Result }) .map_err(|e| anyhow::anyhow!(e.to_string()))?; + // bridge::info — relative WS path so reverse-proxy / HTTPS deployments + // compose the full URL from `window.location`. `engine_url` is the direct + // ws:// URL for callers (tests, native clients) that bypass the reverse + // proxy. + let engine_url_owned = engine_url.to_string(); + let bridge_info = iii.register_function(( + RegisterFunctionMessage::with_id("bridge::info".into()).with_description( + "Returns the relative WebSocket path and engine URL for browser clients.".into(), + ), + move |_payload: Value| { + let engine_url = engine_url_owned.clone(); + async move { + Ok::<_, IIIError>(json!({ + "ws_path": "/iii/ws", + "protocol": "ws", + "engine_url": engine_url, + })) + } + }, + )); + + // Per-browser subscription registry. Skeleton: future steps will use this + // to drive WS push of agent::events / state diffs / cost / approvals. + let fanout = fanout::new_shared(); + + let fanout_for_subscribe = Arc::clone(&fanout); + let subscribe_fn = iii.register_function(( + RegisterFunctionMessage::with_id("ui::subscribe".into()).with_description( + "Register a browser's interest in a session (or all sessions if session_id is null)." + .into(), + ), + move |input: Value| { + let fanout = Arc::clone(&fanout_for_subscribe); + async move { + let body = input.get("body").cloned().unwrap_or(input); + let browser_id = body + .get("browser_id") + .and_then(Value::as_str) + .ok_or_else(|| IIIError::Handler("missing browser_id".into()))? + .to_string(); + let session_id = body + .get("session_id") + .and_then(Value::as_str) + .map(str::to_string); + let total = { + let mut state = fanout.write().await; + state.subscribe(browser_id, session_id); + state.browser_count() + }; + Ok::<_, IIIError>(json!({ + "ok": true, + "total_browsers": total, + })) + } + }, + )); + + let fanout_for_unsubscribe = Arc::clone(&fanout); + let unsubscribe_fn = iii.register_function(( + RegisterFunctionMessage::with_id("ui::unsubscribe".into()).with_description( + "Remove a browser's subscription to a session (or its all-sessions sub if session_id is null)." + .into(), + ), + move |input: Value| { + let fanout = Arc::clone(&fanout_for_unsubscribe); + async move { + let body = input.get("body").cloned().unwrap_or(input); + let browser_id = body + .get("browser_id") + .and_then(Value::as_str) + .ok_or_else(|| IIIError::Handler("missing browser_id".into()))? + .to_string(); + let session_id = body + .get("session_id") + .and_then(Value::as_str) + .map(str::to_string); + let total = { + let mut state = fanout.write().await; + state.unsubscribe(&browser_id, session_id); + state.browser_count() + }; + Ok::<_, IIIError>(json!({ + "ok": true, + "total_browsers": total, + })) + } + }, + )); + // Best-effort: a missing `skills` worker shouldn't stop harness from booting. let _ = iii .trigger(TriggerRequest { @@ -246,10 +370,23 @@ pub async fn register_with_iii(iii: &III) -> anyhow::Result }) .await; + // Wire the upstream fanout pumps: + // - agent::events stream subscriber → ui::session::event:: + // - state::list poll → ui::sessions::changed:: + // + // These spawn long-lived tasks; the returned handle ends them on + // `unregister_all`. The `iii` clone here is fine — `III` is internally + // ref-counted (it's already an `Arc<...>` inside the SDK). + let fanout_pumps = fanout::spawn_subscribers(&Arc::new(iii.clone()), Arc::clone(&fanout)); + Ok(HarnessFunctionRefs { status, bridge, events: events_fn, + bridge_info, + subscribe_fn, + unsubscribe_fn, + fanout_pumps: Some(fanout_pumps), }) } diff --git a/harness/src/main.rs b/harness/src/main.rs index f16213b1e..6905ea270 100644 --- a/harness/src/main.rs +++ b/harness/src/main.rs @@ -69,7 +69,7 @@ async fn main() -> Result<()> { ); let iii = Arc::new(iii); - let _refs = harness::register_with_iii(&iii).await?; + let _refs = harness::register_with_iii_with_engine_url(&iii, &url).await?; tracing::info!( "harness ready — registered harness::status; expecting {} runtime workers from iii.worker.yaml", harness::EXPECTED_WORKERS.len() diff --git a/harness/tests/bridge_info.rs b/harness/tests/bridge_info.rs new file mode 100644 index 000000000..c8494a05d --- /dev/null +++ b/harness/tests/bridge_info.rs @@ -0,0 +1,59 @@ +//! Integration test for harness's bridge::info function. +//! +//! Registers the harness functions in-process against a running engine and +//! verifies that bridge::info returns a relative WebSocket path plus the +//! engine URL it was configured with. Skipped when no engine is reachable +//! (mirrors `tests/sse_bridge.rs`). + +use harness::register_with_iii_with_engine_url; +use iii_sdk::{register_worker, InitOptions, TriggerRequest}; +use serde_json::json; + +const DEFAULT_ENGINE_URL: &str = "ws://127.0.0.1:49134"; +const ENGINE_PROBE_TIMEOUT_MS: u64 = 500; + +#[tokio::test] +async fn bridge_info_returns_relative_ws_path_and_engine_url() { + let url = std::env::var("III_URL").unwrap_or_else(|_| DEFAULT_ENGINE_URL.to_string()); + let iii = register_worker(&url, InitOptions::default()); + + // Probe with a short state::get to confirm the engine is live. + let probe = iii + .trigger(TriggerRequest { + function_id: "state::get".into(), + payload: json!({ "scope": "agent", "key": "__bridge_info_probe" }), + action: None, + timeout_ms: Some(ENGINE_PROBE_TIMEOUT_MS), + }) + .await; + if probe.is_err() { + eprintln!("skipping: no engine at {url}"); + return; + } + + let _refs = match register_with_iii_with_engine_url(&iii, &url).await { + Ok(r) => r, + Err(e) => { + eprintln!("skipping: register_with_iii_with_engine_url failed: {e}"); + return; + } + }; + + let info = iii + .trigger(TriggerRequest { + function_id: "bridge::info".into(), + payload: json!({}), + action: None, + timeout_ms: Some(2_000), + }) + .await + .expect("call bridge::info"); + + assert_eq!(info["ws_path"], "/iii/ws"); + assert_eq!(info["protocol"], "ws"); + assert_eq!( + info["engine_url"].as_str(), + Some(url.as_str()), + "engine_url should reflect the URL passed to register_with_iii_with_engine_url; got {info}" + ); +} diff --git a/harness/web/package.json b/harness/web/package.json index e936013cf..c017dc0bd 100644 --- a/harness/web/package.json +++ b/harness/web/package.json @@ -12,14 +12,17 @@ "e2e": "playwright test" }, "dependencies": { + "iii-browser-sdk": "0.11.7-next.1", "react": "^18.3.1", "react-dom": "^18.3.1" }, "devDependencies": { "@playwright/test": "^1.49.0", + "@testing-library/react": "^16.3.2", "@types/react": "^18.3.12", "@types/react-dom": "^18.3.1", "@vitejs/plugin-react": "^4.3.4", + "jsdom": "^29.1.1", "typescript": "^5.6.3", "vite": "^5.4.11", "vitest": "^2.1.4" diff --git a/harness/web/src/App.tsx b/harness/web/src/App.tsx index 15b09695b..34791a630 100644 --- a/harness/web/src/App.tsx +++ b/harness/web/src/App.tsx @@ -1,5 +1,8 @@ import { useCallback, useEffect, useMemo, useRef, useState } from "react"; import { bridge, BridgeError } from "./bridge"; +import { disposeIiiClient, getIiiClient } from "./iii-client"; +import { exportJson, exportMd } from "./export"; +import { loadMessagesWithEntryIds } from "./loadMessages"; import { visibleMessages } from "./reducer"; import { useAgentStream } from "./useAgentStream"; import { useSkillsIndex } from "./useSkillsIndex"; @@ -10,9 +13,17 @@ import { ContextMeter } from "./components/ContextMeter"; import { ControlsBar } from "./components/ControlsBar"; import { CostPanel } from "./components/CostPanel"; import { FilesystemPanel } from "./components/FilesystemPanel"; +import { FootStatus } from "./components/FootStatus"; +import { FunctionPalette } from "./components/FunctionPalette"; import { SessionList, fetchSessions } from "./components/SessionList"; import { SessionView } from "./components/SessionView"; import { StatusPill } from "./components/StatusPill"; +import { StatusStrip } from "./components/StatusStrip"; +import { StatusTab } from "./components/StatusTab"; +import { useConnection } from "./useConnection"; +import { useGlobalShortcut } from "./useGlobalShortcut"; +import { useStatus } from "./useStatus"; +import { loadWorkspace, saveWorkspace } from "./workspace"; import type { AgentMessage, AuthStatus, @@ -20,7 +31,7 @@ import type { SessionRow, } from "./types"; -type Tab = "chat" | "cost" | "files"; +type Tab = "chat" | "cost" | "files" | "status"; // Tool schemas are no longer shipped from the client. The harness builds the // LLM tool catalog server-side from `engine::functions::list` (see @@ -35,7 +46,11 @@ type Tab = "chat" | "cost" | "files"; const BASE_SYSTEM_PROMPT = "You have filesystem tools that operate inside a sandbox. Use them when the user asks to read, inspect, create, or modify files. Paths must be absolute (e.g. /tmp/notes.md). Some destructive ops may be denied by policy — if a tool result contains `blocked`, explain which policy refused and stop, do not retry."; -function buildSystemPrompt(skillsIndex: string | null): string { +function buildSystemPrompt(skillsIndex: string | null, cwd: string): string { + const cwdSection = cwd + ? `## Working directory\n${cwd}\nPrefer paths under this directory. Use absolute paths.\n\n` + : ""; + const skillsSection = skillsIndex ? `## Available skills @@ -44,7 +59,7 @@ ${skillsIndex} Use the \`skill::fetch\` tool to load any \`iii://\` URI you see above when you need its full content.` : "## Available skills\n\n(Skills index not loaded — call `skill::fetch` with `uri: \"iii://skills\"` to discover what's registered.)"; - return `${BASE_SYSTEM_PROMPT}\n\n${skillsSection}`; + return `${BASE_SYSTEM_PROMPT}\n\n${cwdSection}${skillsSection}`; } // Providers we have actual workers for in iii.worker.yaml. Don't add others @@ -76,9 +91,24 @@ export default function App() { const [active, setActive] = useState(null); const [draftId, setDraftId] = useState(null); const [messages, setMessages] = useState([]); + // Parallel array to `messages`. Slot is the entry_id keying that message + // in session-tree, or `null` when the message came from the state::* + // fallback (drift case — fork is disabled for null entries). + // + // INVARIANT: `messageEntryIds.length === messages.length` at all times. + // Stream events (SSE today) don't carry entry_ids yet — we fill with + // `null` on stream-driven updates. Step B (WS migration) will let stream + // events carry real ids and this can become live data. + const [messageEntryIds, setMessageEntryIds] = useState<(string | null)[]>([]); const [loading, setLoading] = useState(false); const [error, setError] = useState(null); + // Per-session working directory. Advisory only — surfaced in the system + // prompt so the agent prefers paths under it. `loadedCwd` tracks the value + // currently persisted so blur/Enter can decide whether to call save. + const [cwd, setCwd] = useState(""); + const [loadedCwd, setLoadedCwd] = useState(""); + // Skills-index fetch is strictly non-blocking. The fallback branch in // buildSystemPrompt(null) is the agent's recovery path; do not gate // rendering or send() on the index being loaded. @@ -95,12 +125,35 @@ export default function App() { const [tab, setTab] = useState("chat"); + // Cmd-J (Ctrl-J on Linux/Win) opens the bus function palette. Cmd-K is + // taken by Chrome's address bar focus, so we deliberately picked J. + const [paletteOpen, setPaletteOpen] = useState(false); + const openPalette = useCallback(() => setPaletteOpen(true), []); + useGlobalShortcut( + { key: "j", meta: true, ctrl: true }, + openPalette, + ); + const stream = useAgentStream(active); + // Live ambient status — header chip + foot chips + status tab. + // The hook owns the rolling 200-event buffer and the per-page subscription + // to all-sessions topics (cost/workers/approvals). + const status = useStatus(); + const connection = useConnection(); + const isConnected = connection.status === "connected"; + useEffect(() => { if (!active) return; const visible = visibleMessages(stream); - if (visible.length > 0) setMessages(visible); + if (visible.length > 0) { + setMessages(visible); + // Stream events don't carry entry_ids today (Phase B step B will fix + // this when WS replaces SSE). Until then, fork buttons stay disabled + // for live messages — the next loadSessionMessages refresh hydrates + // entry_ids from session-tree. + setMessageEntryIds(visible.map(() => null)); + } }, [active, stream]); const isRunning = loading || stream.status === "running"; @@ -125,30 +178,74 @@ export default function App() { } }, []); - const loadMessages = useCallback(async (id: string) => { + const loadSessionMessages = useCallback(async (id: string) => { try { - const msgs = await bridge("state::get", { - scope: "agent", - key: `session/${id}/messages`, - }); - setMessages(Array.isArray(msgs) ? msgs : []); - } catch (e) { - // brand new session — state::get returns null; treat as empty - if (e instanceof BridgeError && /session not found|null/.test(e.message)) { - setMessages([]); - return; - } + const pairs = await loadMessagesWithEntryIds(id); + setMessages(pairs.map((p) => p.message)); + setMessageEntryIds(pairs.map((p) => p.entry_id)); + } catch { setMessages([]); + setMessageEntryIds([]); } }, []); - // Pull sessions on a slow tick so new turns surface in the rail. + // Refresh sessions when the harness fanout pushes `ui::sessions::changed`. + // Replaces the 4-second polling interval. The all-sessions subscription is + // owned by App for the lifetime of the page; per-session subscriptions are + // managed by useAgentStream. useEffect(() => { void refreshSessions(); - const id = setInterval(refreshSessions, 4000); - return () => clearInterval(id); + let cancelled = false; + let off: (() => void) | undefined; + let subscribed = false; + let browserId: string | null = null; + + void (async () => { + try { + const client = await getIiiClient(); + if (cancelled) return; + browserId = client.browserId; + off = client.on("ui::sessions::changed", () => { + void refreshSessions(); + }); + await client.call("ui::subscribe", { + browser_id: browserId, + session_id: null, + }); + subscribed = true; + } catch { + // No connection — refreshSessions above already ran once. The status + // pill will show the disconnected state via useConnection. + } + })(); + + return () => { + cancelled = true; + off?.(); + if (subscribed && browserId) { + void getIiiClient().then((client) => + client + .call("ui::unsubscribe", { + browser_id: browserId, + session_id: null, + }) + .catch(() => {}), + ); + } + }; }, [refreshSessions]); + // Tear down the iii-client on page unload so the fanout can drop our + // subscriptions promptly. The browser will close the WS regardless, but + // an explicit shutdown lets the engine clean up registered handlers. + useEffect(() => { + const handler = () => { + void disposeIiiClient(); + }; + window.addEventListener("beforeunload", handler); + return () => window.removeEventListener("beforeunload", handler); + }, []); + // Load model catalog once. useEffect(() => { bridge<{ models: ModelInfo[] }>("models::list") @@ -163,9 +260,43 @@ export default function App() { }, [refreshAuth]); useEffect(() => { - if (active) void loadMessages(active); - else setMessages([]); - }, [active, loadMessages]); + if (active) void loadSessionMessages(active); + else { + setMessages([]); + setMessageEntryIds([]); + } + }, [active, loadSessionMessages]); + + // Load workspace cwd on session change. Resets to empty for the draft + // (no active session yet). Errors are swallowed by loadWorkspace. + useEffect(() => { + if (!active) { + setCwd(""); + setLoadedCwd(""); + return; + } + let cancelled = false; + void loadWorkspace(active).then((ws) => { + if (cancelled) return; + const value = ws?.cwd ?? ""; + setCwd(value); + setLoadedCwd(value); + }); + return () => { + cancelled = true; + }; + }, [active]); + + // Persist cwd when it differs from the loaded value. Called from blur and + // Enter on the header field. Only writes when there's an active session. + const commitCwd = useCallback(() => { + if (!active) return; + const next = cwd.trim(); + if (next === loadedCwd) return; + void saveWorkspace(active, next).then(() => { + setLoadedCwd(next); + }); + }, [active, cwd, loadedCwd]); // When the catalog or provider changes, ensure the selected model belongs // to the active provider; otherwise pick the configured default or first @@ -182,6 +313,7 @@ export default function App() { const startNew = () => { setActive(null); setMessages([]); + setMessageEntryIds([]); setDraftId(newSessionId()); setError(null); }; @@ -200,6 +332,8 @@ export default function App() { }; const fullHistory = [...messages, optimistic]; setMessages(fullHistory); + // Optimistic message has no entry_id yet — pad to keep arrays in sync. + setMessageEntryIds([...messageEntryIds, null]); try { await bridge<{ session_id: string }>("run::start", { @@ -207,7 +341,7 @@ export default function App() { provider, model, messages: fullHistory, - system_prompt: buildSystemPrompt(skillsIndex), + system_prompt: buildSystemPrompt(skillsIndex, cwd.trim()), approval_required: APPROVAL_REQUIRED, }); void refreshSessions(); @@ -219,6 +353,78 @@ export default function App() { } }; + // Fork the active session at a specific message's entry_id. The new + // session's id is returned by session-tree::fork and becomes the active + // session. Refreshing the rail picks up the new row. + const handleForkFromMessage = useCallback( + async (entryId: string) => { + if (!active) return; + try { + const { session_id } = await bridge<{ session_id: string }>( + "session-tree::fork", + { + source_session_id: active, + from_entry_id: entryId, + }, + ); + setActive(session_id); + void refreshSessions(); + } catch (e) { + setError(e instanceof Error ? e.message : String(e)); + } + }, + [active, refreshSessions], + ); + + // /repair — read state::* snapshot, hand it to session-tree::reconcile + // so any drifted rows get re-keyed with entry_ids. Reload after so the + // UI's entry_ids reflect the new tree state. + const handleRepair = useCallback(async () => { + if (!active) return; + try { + const snapshot = await bridge("state::get", { + scope: "agent", + key: `session/${active}/messages`, + }); + const result = await bridge<{ repaired: number }>( + "session-tree::reconcile", + { session_id: active, state_snapshot: snapshot }, + ); + void loadSessionMessages(active); + if (result.repaired === 0) setError(null); + } catch (e) { + setError(`/repair failed: ${e instanceof Error ? e.message : String(e)}`); + } + }, [active, loadSessionMessages]); + + // /fork — convenience: fork from the most recent message that has a + // real entry_id. Anything without an entry_id is a stream artifact or + // a drifted state row. + const handleForkLast = useCallback(async () => { + if (!active) return; + for (let i = messageEntryIds.length - 1; i >= 0; i--) { + const id = messageEntryIds[i]; + if (id !== null) { + await handleForkFromMessage(id); + return; + } + } + setError("No messages with entry_ids — try /repair first"); + }, [active, messageEntryIds, handleForkFromMessage]); + + const handleExport = useCallback( + async (format: "md" | "json") => { + if (!active) return; + try { + if (format === "md") await exportMd(active); + else await exportJson(active); + } catch (e) { + setError(`/export failed: ${e instanceof Error ? e.message : String(e)}`); + } + }, + [active], + ); + const sessionId = active ?? draftId ?? ""; const currentAuth = authByProvider[provider] ?? null; const composerDisabled = !(currentAuth?.configured ?? false); @@ -235,8 +441,38 @@ export default function App() { ⌘ harness bus console + {sessionId ? ( + + · {sessionId} + + ) : null} + setCwd(e.target.value)} + onBlur={commitCwd} + onKeyDown={(e) => { + if (e.key === "Enter") { + e.preventDefault(); + commitCwd(); + (e.target as HTMLInputElement).blur(); + } + }} + aria-label="working directory" + /> + +
+ setTab("status")} + /> +
-
@@ -253,7 +489,7 @@ export default function App() {
@@ -318,7 +595,19 @@ export default function App() { provider · {provider} model · {model} endpoint · POST /bridge/trigger + shortcut · ⌘J palette + + + + setPaletteOpen(false)} + /> ); } diff --git a/harness/web/src/bridge.ts b/harness/web/src/bridge.ts index 1582a7ac4..d1b17bc77 100644 --- a/harness/web/src/bridge.ts +++ b/harness/web/src/bridge.ts @@ -1,6 +1,10 @@ -// One endpoint to reach the entire iii bus. -// The harness worker registers POST /bridge/trigger which forwards -// {function_id, payload} to iii.trigger and returns the result. +// One endpoint to reach the entire iii bus. Production callers use the iii +// WebSocket transport (the browser is itself a worker). `bridgeHttp` is kept +// as a backstop for direct curl-style debugging and for the very first +// `bridge::info` round-trip in iii-client.ts; nothing else should reach +// for it. + +import { getIiiClient } from "./iii-client"; const BRIDGE_URL = "/bridge/trigger"; @@ -19,6 +23,25 @@ export class BridgeError extends Error { export async function bridge( functionId: string, payload: Record = {}, +): Promise { + try { + const client = await getIiiClient(); + return await client.call(functionId, payload); + } catch (e) { + if (e instanceof BridgeError) throw e; + const msg = e instanceof Error ? e.message : String(e); + throw new BridgeError(msg, functionId, 0); + } +} + +/** + * HTTP backstop for the bridge. Use only for direct debugging or for the + * single `bridge::info` round-trip needed to bootstrap the WebSocket + * connection; production callers should go through {@link bridge}. + */ +export async function bridgeHttp( + functionId: string, + payload: Record = {}, ): Promise { const res = await fetch(BRIDGE_URL, { method: "POST", diff --git a/harness/web/src/components/Composer.tsx b/harness/web/src/components/Composer.tsx index 0b7cd8e3d..c558b9859 100644 --- a/harness/web/src/components/Composer.tsx +++ b/harness/web/src/components/Composer.tsx @@ -1,47 +1,567 @@ -import { FormEvent, useState } from "react"; +import { + FormEvent, + KeyboardEvent, + useCallback, + useEffect, + useMemo, + useRef, + useState, +} from "react"; +import { bridge, BridgeError } from "../bridge"; +import { + extractQuery, + shouldOpenAt, + shouldOpenSlash, + useCommandMenu, + type MenuItem, +} from "../useCommandMenu"; +import { + BUILT_IN_COMMANDS, + filterCommands, + skillsIndexToMenuItems, +} from "../menuItems"; +import type { AgentMessage, ToolResult, FsLsDetails, FsEntry } from "../types"; + +// ─── Built-in command callbacks ──────────────────────────────────────────── +// Each /name route the Composer can dispatch. Optional callbacks; the +// Composer falls back to a sensible default for missing handlers. +export interface ComposerCallbacks { + onNew?: () => void; + onClear?: () => void; + onCwd?: (path: string) => void; + onModel?: (id: string) => void; + onProvider?: (name: string) => void; + onHelp?: () => void; + /** /repair — reconciles session-tree against state::* snapshot. */ + onRepair?: () => void | Promise; + /** /fork — forks the active session from the last message with an entry_id. */ + onFork?: () => void | Promise; + /** /export md|json — downloads a transcript file via the browser. */ + onExport?: (format: "md" | "json") => void | Promise; +} interface Props { disabled: boolean; onSend: (prompt: string) => Promise; + /** Working directory used as the @-mention browse root. Empty = unset. */ + cwd: string; + /** Markdown index from useSkillsIndex. Used to populate slash menu skills. */ + skillsIndex: string | null; + /** Prior messages of the active session — drives ↑ history walk. */ + sessionMessages: AgentMessage[]; + /** Per-builtin handlers. */ + callbacks?: ComposerCallbacks; +} + +const AT_PAGE_SIZE = 25; + +interface AtBrowseState { + /** Directory currently being listed. Empty when no @-mention is active. */ + dir: string; + /** Raw entries returned by `shell::filesystem::ls` for `dir`. */ + entries: FsEntry[]; + loading: boolean; + /** Empty-state reason, mapped to user-visible text. */ + error: "cwd-unset" | "permission" | "fetch-failed" | "io" | null; +} + +const AT_BROWSE_INITIAL: AtBrowseState = { + dir: "", + entries: [], + loading: false, + error: null, +}; + +function joinPath(base: string, name: string): string { + if (!base) return name; + if (base.endsWith("/")) return `${base}${name}`; + return `${base}/${name}`; +} + +function buildSkillsItems(index: string | null): MenuItem[] { + return [...BUILT_IN_COMMANDS, ...skillsIndexToMenuItems(index)]; +} + +function entriesToMenuItems(dir: string, entries: FsEntry[]): MenuItem[] { + const sorted = [...entries].sort((a, b) => { + const aDir = a.kind === "dir" ? 1 : 0; + const bDir = b.kind === "dir" ? 1 : 0; + if (aDir !== bDir) return bDir - aDir; + return a.name.localeCompare(b.name); + }); + return sorted.map((e) => { + const abs = joinPath(dir, e.name); + const isDir = e.kind === "dir"; + return { + kind: "file" as const, + id: abs, + label: e.name + (isDir ? "/" : ""), + description: abs, + meta: { isDir, dir }, + }; + }); +} + +/** Pull all prior user-text messages for ↑ history walk. */ +function userTexts(messages: AgentMessage[]): string[] { + const out: string[] = []; + for (let i = messages.length - 1; i >= 0; i--) { + const m = messages[i]; + if (m.role !== "user") continue; + const text = m.content + .filter((b): b is { type: "text"; text: string } => b.type === "text") + .map((b) => b.text) + .join(""); + if (text.trim().length > 0) out.push(text); + } + return out; } -export function Composer({ disabled, onSend }: Props) { +export function Composer({ + disabled, + onSend, + cwd, + skillsIndex, + sessionMessages, + callbacks, +}: Props) { const [text, setText] = useState(""); const [busy, setBusy] = useState(false); + const [menu, dispatch] = useCommandMenu(); + const [atBrowse, setAtBrowse] = useState(AT_BROWSE_INITIAL); + const [atPage, setAtPage] = useState(1); + const textareaRef = useRef(null); - const submit = async (e: FormEvent) => { - e.preventDefault(); - const trimmed = text.trim(); - if (!trimmed || busy) return; - setBusy(true); - try { - await onSend(trimmed); - setText(""); - } finally { - setBusy(false); - } + // Memoized so reference equality is stable for the menu effect. + const slashCatalog = useMemo(() => buildSkillsItems(skillsIndex), [skillsIndex]); + const history = useMemo(() => userTexts(sessionMessages), [sessionMessages]); + + // ─── @-mention IO ─────────────────────────────────────────────────────── + // Whenever atBrowse.dir changes, fetch its contents. Errors map to a typed + // empty-state so the popover can render clear text without re-throwing. + useEffect(() => { + if (!atBrowse.dir) return; + let cancelled = false; + setAtBrowse((s) => ({ ...s, loading: true, error: null })); + bridge>("shell::filesystem::ls", { + path: atBrowse.dir, + }) + .then((res) => { + if (cancelled) return; + const detailsErr = res.details?.error; + if (detailsErr) { + setAtBrowse({ + dir: atBrowse.dir, + entries: [], + loading: false, + error: /permission|denied/i.test(detailsErr) ? "permission" : "io", + }); + return; + } + setAtBrowse({ + dir: atBrowse.dir, + entries: res.details?.entries ?? [], + loading: false, + error: null, + }); + setAtPage(1); + }) + .catch((e: unknown) => { + if (cancelled) return; + const isPerm = + e instanceof BridgeError && /permission|denied/i.test(e.message); + setAtBrowse({ + dir: atBrowse.dir, + entries: [], + loading: false, + error: isPerm ? "permission" : "fetch-failed", + }); + }); + return () => { + cancelled = true; + }; + }, [atBrowse.dir]); + + // ─── Trigger detection on every text change ───────────────────────────── + // Composer parses the text + caret position to decide whether the user just + // entered or remained in slash/at mode. History mode is keyboard-driven only. + const recomputeMenu = useCallback( + (nextText: string, caret: number) => { + // If currently in history mode, leaving it requires Esc/Enter — typing + // also exits because the textarea no longer matches the recalled msg. + if (menu.mode === "history") { + dispatch({ kind: "close" }); + } + + // Slash takes precedence over at when both could be derived (a `/` at + // line-start always wins; `@` only fires at word-boundaries). + if (shouldOpenSlash(nextText, caret)) { + dispatch({ kind: "open-slash", items: slashCatalog }); + return; + } + if (shouldOpenAt(nextText, caret)) { + if (!cwd) { + // Open the menu in error state so the user sees why nothing listed. + dispatch({ kind: "open-at", items: [] }); + setAtBrowse({ dir: "", entries: [], loading: false, error: "cwd-unset" }); + return; + } + setAtBrowse({ dir: cwd, entries: [], loading: true, error: null }); + dispatch({ kind: "open-at", items: [] }); + return; + } + + // If a mode is open, refresh items based on the current query. + if (menu.mode === "slash") { + const q = extractQuery(nextText, caret, "/"); + if (q == null) { + dispatch({ kind: "close" }); + return; + } + dispatch({ + kind: "filter", + query: q, + items: filterCommands(slashCatalog, q), + }); + } else if (menu.mode === "at") { + const q = extractQuery(nextText, caret, "@"); + if (q == null) { + dispatch({ kind: "close" }); + setAtBrowse(AT_BROWSE_INITIAL); + return; + } + // If query ends with `/`, the user is descending into a subdir. + // We rebase the browse dir and clear the query portion below the slash. + if (q.endsWith("/") && q.length > 1) { + const subdir = q.slice(0, q.length - 1); + const target = subdir.startsWith("/") ? subdir : joinPath(cwd, subdir); + if (target !== atBrowse.dir) { + setAtBrowse({ dir: target, entries: [], loading: true, error: null }); + } + } + // The sub-string after the last `/` is the per-row filter. + const trailing = q.split("/").pop() ?? ""; + const items = entriesToMenuItems(atBrowse.dir, atBrowse.entries); + const filtered = + trailing.length === 0 + ? items + : items.filter((it) => + it.label.toLowerCase().includes(trailing.toLowerCase()), + ); + dispatch({ kind: "filter", query: q, items: filtered }); + } + }, + [menu.mode, dispatch, slashCatalog, cwd, atBrowse.dir, atBrowse.entries], + ); + + // Keep the at-mention popover in sync when its IO completes. + useEffect(() => { + if (menu.mode !== "at") return; + const items = entriesToMenuItems(atBrowse.dir, atBrowse.entries); + dispatch({ kind: "set-items", items: items.slice(0, atPage * AT_PAGE_SIZE) }); + }, [menu.mode, atBrowse.entries, atBrowse.dir, atPage, dispatch]); + + // ─── Submit + dispatch ────────────────────────────────────────────────── + const submit = useCallback( + async (e?: FormEvent) => { + e?.preventDefault(); + const trimmed = text.trim(); + if (!trimmed || busy) return; + + // Built-in /cwd is intercepted client-side. + const cwdMatch = /^\/cwd\s+(.+)$/.exec(trimmed); + if (cwdMatch) { + callbacks?.onCwd?.(cwdMatch[1].trim()); + setText(""); + return; + } + if (trimmed === "/clear") { + callbacks?.onClear?.(); + setText(""); + return; + } + if (trimmed === "/new") { + callbacks?.onNew?.(); + setText(""); + return; + } + if (trimmed === "/help") { + callbacks?.onHelp?.(); + return; + } + if (trimmed === "/repair") { + void callbacks?.onRepair?.(); + setText(""); + return; + } + if (trimmed === "/fork") { + void callbacks?.onFork?.(); + setText(""); + return; + } + if (trimmed === "/export md") { + void callbacks?.onExport?.("md"); + setText(""); + return; + } + if (trimmed === "/export json") { + void callbacks?.onExport?.("json"); + setText(""); + return; + } + const modelMatch = /^\/model\s+(.+)$/.exec(trimmed); + if (modelMatch) { + callbacks?.onModel?.(modelMatch[1].trim()); + setText(""); + return; + } + const providerMatch = /^\/provider\s+(.+)$/.exec(trimmed); + if (providerMatch) { + callbacks?.onProvider?.(providerMatch[1].trim()); + setText(""); + return; + } + + setBusy(true); + try { + await onSend(trimmed); + setText(""); + } finally { + setBusy(false); + } + }, + [text, busy, onSend, callbacks], + ); + + const acceptItem = useCallback( + (item: MenuItem) => { + if (menu.mode === "slash") { + if (item.kind === "builtin") { + // Commands that take an arg insert with trailing space and stay in + // the textarea so the user can type the arg. + if ( + item.id === "/cwd" || + item.id === "/model" || + item.id === "/provider" + ) { + setText(`${item.id} `); + dispatch({ kind: "close" }); + return; + } + // Zero-arg commands fire their callback immediately. + if (item.id === "/new") callbacks?.onNew?.(); + if (item.id === "/clear") callbacks?.onClear?.(); + if (item.id === "/help") callbacks?.onHelp?.(); + if (item.id === "/repair") void callbacks?.onRepair?.(); + if (item.id === "/fork") void callbacks?.onFork?.(); + if (item.id === "/export md") void callbacks?.onExport?.("md"); + if (item.id === "/export json") void callbacks?.onExport?.("json"); + setText(""); + dispatch({ kind: "close" }); + return; + } + if (item.kind === "skill") { + // Skills get inserted as a /skill-id mention so the user can add + // context after it. The agent picks it up via the system-prompt + // skills index and skill::fetch. + setText(`${item.id} `); + dispatch({ kind: "close" }); + return; + } + } + if (menu.mode === "at" && item.kind === "file") { + const meta = item.meta as { isDir?: boolean } | undefined; + if (meta?.isDir) { + // Descend into the directory. Replace the in-progress @ + // with @/ so further typing filters within it. + replaceAtMention(`${item.id}/`); + setAtBrowse({ dir: item.id, entries: [], loading: true, error: null }); + return; + } + // File: insert absolute path and close. + replaceAtMention(item.id + " "); + dispatch({ kind: "close" }); + setAtBrowse(AT_BROWSE_INITIAL); + return; + } + }, + [menu.mode, dispatch, callbacks], + ); + + const replaceAtMention = useCallback( + (replacement: string) => { + const ta = textareaRef.current; + if (!ta) return; + const caret = ta.selectionStart ?? text.length; + // Walk back to the most recent @ that opened the mode. + let at = -1; + for (let i = caret - 1; i >= 0; i--) { + const c = text[i]; + if (c === "\n") break; + if (c === "@") { + at = i; + break; + } + } + if (at < 0) return; + const before = text.slice(0, at); + const after = text.slice(caret); + const next = before + replacement + after; + setText(next); + // Restore caret after the inserted replacement. + const newCaret = (before + replacement).length; + requestAnimationFrame(() => { + ta.setSelectionRange(newCaret, newCaret); + ta.focus(); + }); + }, + [text], + ); + + const onKeyDown = useCallback( + (e: KeyboardEvent) => { + // Menu-active key handling. + if (menu.mode === "slash" || menu.mode === "at") { + if (e.key === "ArrowDown") { + e.preventDefault(); + dispatch({ kind: "move", delta: 1 }); + return; + } + if (e.key === "ArrowUp") { + e.preventDefault(); + dispatch({ kind: "move", delta: -1 }); + return; + } + if (e.key === "Enter" || e.key === "Tab") { + if (menu.items.length > 0) { + e.preventDefault(); + acceptItem(menu.items[menu.selectedIndex] ?? menu.items[0]); + return; + } + } + if (e.key === "Escape") { + e.preventDefault(); + dispatch({ kind: "close" }); + setAtBrowse(AT_BROWSE_INITIAL); + return; + } + } + + if (menu.mode === "history") { + if (e.key === "ArrowUp") { + e.preventDefault(); + dispatch({ kind: "history-step", delta: 1, historyLen: history.length }); + return; + } + if (e.key === "ArrowDown") { + e.preventDefault(); + const cur = menu.historyIndex ?? 0; + if (cur === 0) { + // Stepping below 0 returns to the draft. + setText(menu.historyDraft); + dispatch({ kind: "close" }); + return; + } + dispatch({ kind: "history-step", delta: -1, historyLen: history.length }); + return; + } + if (e.key === "Escape") { + e.preventDefault(); + setText(menu.historyDraft); + dispatch({ kind: "close" }); + return; + } + if (e.key === "Enter" && !e.shiftKey) { + e.preventDefault(); + dispatch({ kind: "close" }); + void submit(); + return; + } + } + + // Idle mode default keys. + if (menu.mode === "idle") { + if (e.key === "ArrowUp" && text.length === 0 && history.length > 0) { + e.preventDefault(); + dispatch({ kind: "open-history", draft: text }); + setText(history[0]); + return; + } + if (e.key === "Enter" && !e.shiftKey) { + e.preventDefault(); + void submit(); + return; + } + } + }, + [ + menu.mode, + menu.items, + menu.selectedIndex, + menu.historyIndex, + menu.historyDraft, + dispatch, + acceptItem, + history, + text, + submit, + ], + ); + + // When historyIndex changes, mirror it into the textarea. + useEffect(() => { + if (menu.mode !== "history") return; + const idx = menu.historyIndex; + if (idx == null) return; + setText(history[idx] ?? menu.historyDraft); + }, [menu.mode, menu.historyIndex, history, menu.historyDraft]); + + const onChange = (e: React.ChangeEvent) => { + const next = e.target.value; + setText(next); + const caret = e.target.selectionStart ?? next.length; + recomputeMenu(next, caret); }; + const showPopover = menu.mode === "slash" || menu.mode === "at"; + const hasMore = + menu.mode === "at" && atBrowse.entries.length > atPage * AT_PAGE_SIZE; + return (
-