From 80f5d51c381184c91f71d083c4b59e6f77dec7e1 Mon Sep 17 00:00:00 2001 From: Brad Groux <3053586+BradGroux@users.noreply.github.com> Date: Wed, 22 Jul 2026 13:53:55 -0500 Subject: [PATCH] perf(acp): trace mention-to-reply latency Refs #2386 Co-authored-by: Brad Groux Signed-off-by: Brad Groux --- benchmarks/agent-reply-latency/README.md | 82 +++ crates/buzz-acp/README.md | 9 + crates/buzz-acp/src/acp.rs | 102 +++ crates/buzz-acp/src/latency.rs | 640 ++++++++++++++++++ crates/buzz-acp/src/lib.rs | 85 ++- crates/buzz-acp/src/observer.rs | 135 +++- crates/buzz-acp/src/relay.rs | 88 ++- crates/buzz-acp/src/setup_mode.rs | 13 +- .../features/agents/ui/agentSessionTypes.ts | 2 + 9 files changed, 1143 insertions(+), 13 deletions(-) create mode 100644 benchmarks/agent-reply-latency/README.md create mode 100644 crates/buzz-acp/src/latency.rs diff --git a/benchmarks/agent-reply-latency/README.md b/benchmarks/agent-reply-latency/README.md new file mode 100644 index 0000000000..9734c2527f --- /dev/null +++ b/benchmarks/agent-reply-latency/README.md @@ -0,0 +1,82 @@ +# Agent mention-to-reply latency + +`buzz-acp` emits a content-free `mention_reply_latency` observer event after it +has seen both the first self-authored reply on the relay and `turn_completed`. +The event contains one sample plus rolling p50, p95, and max summaries for the +most recent 100 samples on the same warm or cold path. + +This first slice measures `harness_relay_receipt` to +`harness_relay_fanout`, recorded explicitly as `measurementStart` and +`measurementEnd`. It does not yet claim sender-publish-to-recipient-render +end-to-end timing. + +## Boundaries + +All duration math uses `monotonicMs`, a process-local monotonic clock. RFC3339 +timestamps and Nostr `created_at` values are correlation metadata only and must +not be used for cross-process duration math. + +| Metric | Start | End | +| --- | --- | --- | +| `receiveToQueueMs` | relay frame accepted by `buzz-acp` | event admitted to the channel queue | +| `queueWaitMs` | event admitted to the channel queue | turn task started | +| `sessionResolveMs` | turn task started | ACP session created or reused | +| `postSessionSetupMs` | ACP session created or reused | ACP `session/prompt` written | +| `turnSetupMs` | turn task started | ACP `session/prompt` written | +| `timeToFirstOutputMs` | ACP `session/prompt` written | first semantic ACP model/tool output frame | +| `firstOutputToReplyMs` | first semantic ACP model/tool output frame | first self-authored kind-9 reply observed on the relay | +| `turnDurationMs` | turn task started | turn task completed | +| `totalMs` | relay frame accepted by `buzz-acp` | first self-authored kind-9 reply observed on the relay | + +`path` is `cold` when the turn created a new ACP session, `warm` when it reused +one, and `unknown` when session resolution was not observed. Summaries never mix +these paths. + +## Deterministic check + +The unit fixture drives fixed semantic boundaries through the collector and +asserts stage math, nearest-rank percentiles, warm/cold separation, flat-thread +correlation, trace expiry, and content redaction: + +```bash +. ./bin/activate-hermit +cargo test -p buzz-acp latency --no-fail-fast +cargo test -p buzz-acp observer_emits_derived_latency_sample --no-fail-fast +``` + +## Live benchmark + +Managed Desktop agents already start with `BUZZ_ACP_RELAY_OBSERVER=true`. For a +standalone harness, set that variable and make sure the agent has a resolvable +owner so encrypted observer frames can be published. Keep the default +`--ignore-self` behavior enabled; that lets mention-only subscriptions observe +self-authored replies for telemetry without routing them back into the agent. + +1. Record the Buzz commit, relay, agent runtime, provider, model, machine, and + network condition. +2. Open the managed-agent session's raw event rail. +3. Send a fixed prompt such as `Reply with exactly: pong.` and wait for its + `mention_reply_latency` event before sending the next trial. +4. For warm trials, reuse the same channel session. For cold trials, issue + `!rotate`, wait for rotation to finish, then send the fixed prompt. +5. Run at least 20 sequential trials per path. Read the last event for each path; + its `summary` contains `windowSize` and per-stage `samples`, `p50`, `p95`, and + `max` in milliseconds. + +Keep real-provider runs scheduled or opt-in rather than merge-blocking. Provider +startup, credentials, cost, network conditions, and relay placement make those +runs useful operational evidence but unsuitable as deterministic CI. + +## Privacy and current scope + +Semantic latency events include IDs, path classification, durations, and sample +counts. They do not include message content, prompts, model output, credentials, +or tool arguments. Existing raw `acp_read`/`acp_write` observer events retain +their current behavior and are outside this telemetry's redaction guarantee. + +The start boundary is harness receipt, not the sender's publish timestamp. The +final reply boundary is relay fanout back to the harness, not the Buzz CLI's +publish-start timestamp or a separate recipient's render timestamp. This first +instrumentation change establishes a measurable baseline; issue #2386 should +remain open until the missing outer boundaries, production baselines, explicit +budgets, and a regression job are landed. diff --git a/crates/buzz-acp/README.md b/crates/buzz-acp/README.md index b9c8406242..1857b93a7b 100644 --- a/crates/buzz-acp/README.md +++ b/crates/buzz-acp/README.md @@ -251,6 +251,15 @@ Each channel has at most one prompt in flight. Multiple channels can be processe > **Note:** On startup, the harness replays all unprocessed @mentions since the last run. Expect a burst of activity if there are stale events in the channel. +### Agent response timing + +With `BUZZ_ACP_RELAY_OBSERVER=true`, the observer feed includes content-free +`mention_reply_latency` samples and rolling warm/cold p50, p95, and max stage +summaries. See the +[mention-to-reply benchmark](../../benchmarks/agent-reply-latency/README.md) for +the event boundaries, repeatable live procedure, privacy scope, and current +limitations. + ## Using Any ACP Agent The harness works with any agent that implements the [ACP spec](https://agentclientprotocol.com/) over stdio. The requirements are: diff --git a/crates/buzz-acp/src/acp.rs b/crates/buzz-acp/src/acp.rs index b553adaabb..626c281002 100644 --- a/crates/buzz-acp/src/acp.rs +++ b/crates/buzz-acp/src/acp.rs @@ -172,6 +172,10 @@ pub struct AcpClient { observer_agent_index: Option, /// Best-effort context attached to raw ACP wire events. observer_context: ObserverContext, + /// Whether the current prompt has produced its first semantic ACP frame. + /// Reset immediately before each `session/prompt` write so cancel cleanup + /// cannot emit a duplicate first-output boundary. + first_turn_output_observed: bool, /// Most recently observed `_meta.goose.activeRunId` from a /// `session/update` notification of kind `session_info_update`. /// @@ -489,6 +493,7 @@ impl AcpClient { observer: None, observer_agent_index: None, observer_context: ObserverContext::default(), + first_turn_output_observed: false, active_run_id: None, steer_rx: None, goose_usage: UsageTracker::default(), @@ -684,6 +689,7 @@ impl AcpClient { // prompt so that any setup notifications recorded earlier are not // misattributed to this turn. self.goose_usage.begin_turn(session_id); + self.first_turn_output_observed = false; self.last_prompt_id = Some(self.next_id); let id = self.next_id; @@ -702,6 +708,7 @@ impl AcpClient { self.current_hard_deadline = None; return Err(e); } + self.observe("prompt_dispatched", serde_json::json!({ "requestId": id })); let result = self .read_until_response_with_idle_timeout( @@ -1407,6 +1414,10 @@ impl AcpClient { continue; } }; + if !self.first_turn_output_observed && is_semantic_turn_output(&msg) { + self.first_turn_output_observed = true; + self.observe("turn_first_output", serde_json::json!({})); + } self.observe("acp_read", msg.clone()); let activity_now = Instant::now(); @@ -1760,6 +1771,26 @@ impl AcpClient { } } +/// Whether an ACP frame represents the first model/tool output for the current +/// turn. Lifecycle, usage, keepalive, and capability updates are intentionally +/// excluded so they cannot make time-to-first-output look artificially fast. +fn is_semantic_turn_output(msg: &serde_json::Value) -> bool { + if msg.get("method").and_then(|value| value.as_str()) != Some("session/update") { + return false; + } + + matches!( + msg["params"]["update"]["sessionUpdate"].as_str(), + Some( + "agent_message_chunk" + | "agent_thought_chunk" + | "tool_call" + | "tool_call_update" + | "plan" + ) + ) +} + /// Build `session/prompt` params from one or more text content blocks. fn build_prompt_params(session_id: &str, prompt_blocks: &[&str]) -> serde_json::Value { let blocks: Vec = prompt_blocks @@ -2590,6 +2621,77 @@ mod tests { .expect("failed to spawn test script") } + #[tokio::test] + async fn prompt_emits_content_free_dispatch_and_first_output_boundaries() { + let script = r#" +IFS= read -r line +printf '%s\n' '{"jsonrpc":"2.0","method":"session/update","params":{"sessionId":"test-session","update":{"sessionUpdate":"keepalive"}}}' +printf '%s\n' '{"jsonrpc":"2.0","method":"session/update","params":{"sessionId":"test-session","update":{"sessionUpdate":"agent_message_chunk","content":{"type":"text","text":"private model output"}}}}' +printf '%s\n' '{"jsonrpc":"2.0","id":0,"result":{"stopReason":"end_turn"}}' +"#; + let mut client = spawn_script(script).await; + let observer = ObserverHandle::in_process(); + client.set_observer(Some(observer.clone()), 0); + client.set_observer_context(crate::observer::context_for_turn( + None, + Some("test-session".into()), + "turn-1".into(), + "2026-07-22T00:00:00Z".into(), + )); + + let result = client + .session_prompt_with_idle_timeout( + "test-session", + "private prompt", + std::time::Duration::from_secs(1), + std::time::Duration::from_secs(5), + ) + .await; + assert_eq!(result.unwrap(), StopReason::EndTurn); + + let events = observer.snapshot(); + let semantic: Vec<_> = events + .iter() + .filter(|event| event.kind == "prompt_dispatched" || event.kind == "turn_first_output") + .collect(); + assert_eq!( + semantic + .iter() + .filter(|event| event.kind == "prompt_dispatched") + .count(), + 1 + ); + assert_eq!( + semantic + .iter() + .filter(|event| event.kind == "turn_first_output") + .count(), + 1 + ); + let first_output = semantic + .iter() + .find(|event| event.kind == "turn_first_output") + .expect("first semantic output boundary"); + let keepalive = events + .iter() + .find(|event| { + event.kind == "acp_read" + && event.payload["params"]["update"]["sessionUpdate"] == "keepalive" + }) + .expect("keepalive raw observer frame"); + assert!( + keepalive.seq < first_output.seq, + "keepalive must not close the first semantic output boundary" + ); + assert!(semantic.windows(2).all(|pair| { + pair[0].monotonic_ms <= pair[1].monotonic_ms && pair[0].seq < pair[1].seq + })); + + let serialized = serde_json::to_string(&semantic).unwrap(); + assert!(!serialized.contains("private prompt")); + assert!(!serialized.contains("private model output")); + } + #[tokio::test] async fn idle_timeout_fires_on_silent_process() { let mut client = spawn_script("sleep 10").await; diff --git a/crates/buzz-acp/src/latency.rs b/crates/buzz-acp/src/latency.rs new file mode 100644 index 0000000000..6c58f3632b --- /dev/null +++ b/crates/buzz-acp/src/latency.rs @@ -0,0 +1,640 @@ +//! Content-free mention-to-reply latency correlation. +//! +//! The collector consumes semantic observer boundaries emitted by the harness +//! and derives one `mention_reply_latency` sample after both the first reply and +//! turn completion are observed. Durations use the process-local monotonic +//! observer clock; RFC3339/Nostr timestamps remain correlation metadata only. + +use std::collections::{HashMap, HashSet, VecDeque}; + +use serde_json::{json, Map, Value}; + +use crate::observer::{ObserverContext, ObserverEvent}; + +const TRACE_TTL_MS: u64 = 15 * 60 * 1_000; +const SUMMARY_WINDOW: usize = 100; + +/// Saturating millisecond conversion for semantic timing payloads. +pub(crate) fn duration_ms(duration: std::time::Duration) -> u64 { + u64::try_from(duration.as_millis()).unwrap_or(u64::MAX) +} + +#[derive(Clone, Debug)] +struct TriggerTiming { + received_ms: u64, + queued_ms: Option, +} + +#[derive(Clone, Debug, Default)] +struct TurnTiming { + agent_index: Option, + channel_id: Option, + session_id: Option, + triggering_event_ids: Vec, + started_ms: u64, + session_resolved_ms: Option, + prompt_dispatched_ms: Option, + first_output_ms: Option, + reply_observed_ms: Option, + completed_ms: Option, + is_new_session: Option, +} + +#[derive(Clone, Copy, Debug)] +struct SampleMetrics { + receive_to_queue_ms: u64, + queue_wait_ms: u64, + session_resolve_ms: Option, + post_session_setup_ms: Option, + turn_setup_ms: Option, + time_to_first_output_ms: Option, + first_output_to_reply_ms: Option, + turn_duration_ms: u64, + total_ms: u64, +} + +/// Derived sample ready to be emitted into the observer feed. +pub(crate) struct CompletedLatency { + pub(crate) agent_index: Option, + pub(crate) context: ObserverContext, + pub(crate) payload: Value, +} + +/// Stateful correlator for semantic observer events. +#[derive(Default)] +pub(crate) struct LatencyCollector { + triggers: HashMap, + event_to_turn: HashMap, + turns: HashMap, + windows: HashMap>, +} + +impl LatencyCollector { + /// Ingest one observer event and return a completed latency sample, if this + /// event closes a trace. Unknown/raw event kinds are deliberately ignored. + pub(crate) fn ingest(&mut self, event: &ObserverEvent) -> Option { + self.prune(event.monotonic_ms); + + let mut candidate_turn_id = event.turn_id.clone(); + match event.kind.as_str() { + "event_received" => { + let event_id = string_field(&event.payload, "eventId")?; + let receipt_to_observer_ms = + u64_field(&event.payload, "receiptToObserverMs").unwrap_or_default(); + self.triggers.insert( + event_id, + TriggerTiming { + received_ms: event.monotonic_ms.saturating_sub(receipt_to_observer_ms), + queued_ms: None, + }, + ); + return None; + } + "event_queued" => { + let event_id = string_field(&event.payload, "eventId")?; + self.triggers.get_mut(&event_id)?.queued_ms = Some(event.monotonic_ms); + return None; + } + "turn_started" => { + let turn_id = event.turn_id.clone()?; + let triggering_event_ids = string_array_field(&event.payload, "triggeringEventIds"); + for event_id in &triggering_event_ids { + self.event_to_turn.insert(event_id.clone(), turn_id.clone()); + } + self.turns.insert( + turn_id.clone(), + TurnTiming { + agent_index: event.agent_index, + channel_id: event.channel_id.clone(), + session_id: event.session_id.clone(), + triggering_event_ids, + started_ms: event.monotonic_ms, + ..TurnTiming::default() + }, + ); + candidate_turn_id = Some(turn_id); + } + "session_resolved" => { + if let Some(turn_id) = event.turn_id.as_deref() { + if let Some(turn) = self.turns.get_mut(turn_id) { + turn.session_id = event.session_id.clone(); + turn.session_resolved_ms.get_or_insert(event.monotonic_ms); + turn.is_new_session = + event.payload.get("isNewSession").and_then(Value::as_bool); + } + } + } + "prompt_dispatched" => { + if let Some(turn_id) = event.turn_id.as_deref() { + if let Some(turn) = self.turns.get_mut(turn_id) { + turn.prompt_dispatched_ms.get_or_insert(event.monotonic_ms); + } + } + } + "turn_first_output" => { + if let Some(turn_id) = event.turn_id.as_deref() { + if let Some(turn) = self.turns.get_mut(turn_id) { + turn.first_output_ms.get_or_insert(event.monotonic_ms); + } + } + } + "reply_observed" => { + if candidate_turn_id.is_none() { + candidate_turn_id = ["parentEventId", "rootEventId"] + .iter() + .filter_map(|key| string_field(&event.payload, key)) + .find_map(|event_id| self.event_to_turn.get(&event_id).cloned()); + } + if candidate_turn_id.is_none() { + // Buzz keeps human-facing conversations flat, so a reply can + // point at the thread root instead of the mention that started + // this turn. At most one turn is active per channel; retain + // completed traces briefly and select the newest channel turn + // when thread tags cannot provide a direct event correlation. + candidate_turn_id = event.channel_id.as_deref().and_then(|channel_id| { + self.turns + .iter() + .filter(|(_, turn)| turn.channel_id.as_deref() == Some(channel_id)) + .max_by_key(|(_, turn)| turn.started_ms) + .map(|(turn_id, _)| turn_id.clone()) + }); + } + if let Some(turn_id) = candidate_turn_id.as_deref() { + if let Some(turn) = self.turns.get_mut(turn_id) { + // First published reply is the user-visible response + // latency boundary. Later progress/final posts do not + // overwrite it. + turn.reply_observed_ms.get_or_insert(event.monotonic_ms); + } + } + } + "turn_completed" => { + if let Some(turn_id) = event.turn_id.as_deref() { + if let Some(turn) = self.turns.get_mut(turn_id) { + turn.completed_ms.get_or_insert(event.monotonic_ms); + } + } + } + _ => return None, + } + + candidate_turn_id.and_then(|turn_id| self.try_complete(&turn_id)) + } + + fn try_complete(&mut self, turn_id: &str) -> Option { + let turn = self.turns.get(turn_id)?.clone(); + let reply_ms = turn.reply_observed_ms?; + let completed_ms = turn.completed_ms?; + let (primary_event_id, trigger, queued_ms) = turn + .triggering_event_ids + .iter() + .filter_map(|event_id| { + self.triggers.get(event_id).and_then(|timing| { + timing + .queued_ms + .map(|queued_ms| (event_id.clone(), timing.clone(), queued_ms)) + }) + }) + .min_by_key(|(_, timing, _)| timing.received_ms)?; + + let metrics = SampleMetrics { + receive_to_queue_ms: queued_ms.saturating_sub(trigger.received_ms), + queue_wait_ms: turn.started_ms.saturating_sub(queued_ms), + session_resolve_ms: turn + .session_resolved_ms + .map(|resolved_ms| resolved_ms.saturating_sub(turn.started_ms)), + post_session_setup_ms: turn + .session_resolved_ms + .zip(turn.prompt_dispatched_ms) + .map(|(resolved_ms, prompt_ms)| prompt_ms.saturating_sub(resolved_ms)), + turn_setup_ms: turn + .prompt_dispatched_ms + .map(|prompt_ms| prompt_ms.saturating_sub(turn.started_ms)), + time_to_first_output_ms: turn + .prompt_dispatched_ms + .zip(turn.first_output_ms) + .map(|(prompt_ms, output_ms)| output_ms.saturating_sub(prompt_ms)), + first_output_to_reply_ms: turn + .first_output_ms + .map(|output_ms| reply_ms.saturating_sub(output_ms)), + turn_duration_ms: completed_ms.saturating_sub(turn.started_ms), + total_ms: reply_ms.saturating_sub(trigger.received_ms), + }; + + let path = match turn.is_new_session { + Some(true) => "cold", + Some(false) => "warm", + None => "unknown", + }; + let window = self.windows.entry(path.to_string()).or_default(); + if window.len() >= SUMMARY_WINDOW { + window.pop_front(); + } + window.push_back(metrics); + + let mut payload = Map::new(); + payload.insert("traceId".into(), Value::String(primary_event_id)); + payload.insert( + "measurementStart".into(), + Value::String("harness_relay_receipt".into()), + ); + payload.insert( + "measurementEnd".into(), + Value::String("harness_relay_fanout".into()), + ); + payload.insert( + "triggeringEventIds".into(), + json!(turn.triggering_event_ids), + ); + payload.insert("path".into(), Value::String(path.to_string())); + insert_u64( + &mut payload, + "receiveToQueueMs", + metrics.receive_to_queue_ms, + ); + insert_u64(&mut payload, "queueWaitMs", metrics.queue_wait_ms); + insert_optional_u64(&mut payload, "sessionResolveMs", metrics.session_resolve_ms); + insert_optional_u64( + &mut payload, + "postSessionSetupMs", + metrics.post_session_setup_ms, + ); + insert_optional_u64(&mut payload, "turnSetupMs", metrics.turn_setup_ms); + insert_optional_u64( + &mut payload, + "timeToFirstOutputMs", + metrics.time_to_first_output_ms, + ); + insert_optional_u64( + &mut payload, + "firstOutputToReplyMs", + metrics.first_output_to_reply_ms, + ); + insert_u64(&mut payload, "turnDurationMs", metrics.turn_duration_ms); + insert_u64(&mut payload, "totalMs", metrics.total_ms); + payload.insert("summary".into(), build_summary(window)); + + self.turns.remove(turn_id); + for event_id in &turn.triggering_event_ids { + self.triggers.remove(event_id); + self.event_to_turn.remove(event_id); + } + + Some(CompletedLatency { + agent_index: turn.agent_index, + context: ObserverContext { + channel_id: turn.channel_id, + session_id: turn.session_id, + turn_id: Some(turn_id.to_string()), + started_at: None, + }, + payload: Value::Object(payload), + }) + } + + fn prune(&mut self, now_ms: u64) { + self.triggers + .retain(|_, timing| now_ms.saturating_sub(timing.received_ms) <= TRACE_TTL_MS); + self.turns + .retain(|_, turn| now_ms.saturating_sub(turn.started_ms) <= TRACE_TTL_MS); + let active_turns: HashSet<&str> = self.turns.keys().map(String::as_str).collect(); + let active_triggers: HashSet<&str> = self.triggers.keys().map(String::as_str).collect(); + self.event_to_turn.retain(|event_id, turn_id| { + active_triggers.contains(event_id.as_str()) && active_turns.contains(turn_id.as_str()) + }); + } +} + +fn string_field(payload: &Value, key: &str) -> Option { + payload.get(key)?.as_str().map(ToOwned::to_owned) +} + +fn u64_field(payload: &Value, key: &str) -> Option { + payload.get(key)?.as_u64() +} + +fn string_array_field(payload: &Value, key: &str) -> Vec { + payload + .get(key) + .and_then(Value::as_array) + .into_iter() + .flatten() + .filter_map(Value::as_str) + .map(ToOwned::to_owned) + .collect() +} + +fn insert_u64(payload: &mut Map, key: &str, value: u64) { + payload.insert(key.to_string(), Value::from(value)); +} + +fn insert_optional_u64(payload: &mut Map, key: &str, value: Option) { + if let Some(value) = value { + insert_u64(payload, key, value); + } +} + +fn build_summary(window: &VecDeque) -> Value { + let mut summary = Map::new(); + summary.insert("windowSize".into(), Value::from(window.len() as u64)); + insert_stats( + &mut summary, + "receiveToQueueMs", + window.iter().map(|sample| Some(sample.receive_to_queue_ms)), + ); + insert_stats( + &mut summary, + "queueWaitMs", + window.iter().map(|sample| Some(sample.queue_wait_ms)), + ); + insert_stats( + &mut summary, + "turnSetupMs", + window.iter().map(|sample| sample.turn_setup_ms), + ); + insert_stats( + &mut summary, + "sessionResolveMs", + window.iter().map(|sample| sample.session_resolve_ms), + ); + insert_stats( + &mut summary, + "postSessionSetupMs", + window.iter().map(|sample| sample.post_session_setup_ms), + ); + insert_stats( + &mut summary, + "timeToFirstOutputMs", + window.iter().map(|sample| sample.time_to_first_output_ms), + ); + insert_stats( + &mut summary, + "firstOutputToReplyMs", + window.iter().map(|sample| sample.first_output_to_reply_ms), + ); + insert_stats( + &mut summary, + "turnDurationMs", + window.iter().map(|sample| Some(sample.turn_duration_ms)), + ); + insert_stats( + &mut summary, + "totalMs", + window.iter().map(|sample| Some(sample.total_ms)), + ); + Value::Object(summary) +} + +fn insert_stats(summary: &mut Map, key: &str, values: I) +where + I: Iterator>, +{ + let mut values: Vec = values.flatten().collect(); + if values.is_empty() { + return; + } + values.sort_unstable(); + let p50 = nearest_rank(&values, 50); + let p95 = nearest_rank(&values, 95); + let max = values.last().copied().unwrap_or_default(); + summary.insert( + key.to_string(), + json!({ + "samples": values.len(), + "p50": p50, + "p95": p95, + "max": max, + }), + ); +} + +fn nearest_rank(sorted: &[u64], percentile: usize) -> u64 { + let rank = (percentile * sorted.len()).div_ceil(100); + sorted[rank.saturating_sub(1).min(sorted.len() - 1)] +} + +#[cfg(test)] +mod tests { + use super::*; + + fn event( + kind: &str, + monotonic_ms: u64, + turn_id: Option<&str>, + payload: Value, + ) -> ObserverEvent { + ObserverEvent { + seq: monotonic_ms, + monotonic_ms, + timestamp: "2026-07-22T00:00:00Z".to_string(), + kind: kind.to_string(), + agent_index: Some(0), + channel_id: Some("11111111-1111-1111-1111-111111111111".to_string()), + session_id: turn_id.map(|_| "session-1".to_string()), + turn_id: turn_id.map(ToOwned::to_owned), + started_at: None, + payload, + } + } + + fn complete_sample( + collector: &mut LatencyCollector, + event_id: &str, + turn_id: &str, + offset: u64, + total_ms: u64, + is_new_session: bool, + ) -> CompletedLatency { + let received = offset; + let queued = received + 3; + let started = queued + 10; + let session_resolved = started + 2; + let prompt = started + 5; + let first_output = prompt + 15; + let reply = received + total_ms; + let completed = reply + 5; + + assert!(collector + .ingest(&event( + "event_received", + received, + None, + json!({"eventId": event_id, "receiptToObserverMs": 0}), + )) + .is_none()); + assert!(collector + .ingest(&event( + "event_queued", + queued, + None, + json!({"eventId": event_id}), + )) + .is_none()); + assert!(collector + .ingest(&event( + "turn_started", + started, + Some(turn_id), + json!({"triggeringEventIds": [event_id]}), + )) + .is_none()); + assert!(collector + .ingest(&event( + "session_resolved", + session_resolved, + Some(turn_id), + json!({"isNewSession": is_new_session}), + )) + .is_none()); + assert!(collector + .ingest(&event( + "prompt_dispatched", + prompt, + Some(turn_id), + json!({}), + )) + .is_none()); + assert!(collector + .ingest(&event( + "turn_first_output", + first_output, + Some(turn_id), + json!({}), + )) + .is_none()); + assert!(collector + .ingest(&event( + "reply_observed", + reply, + Some(turn_id), + json!({"eventId": "reply-id"}), + )) + .is_none()); + collector + .ingest(&event( + "turn_completed", + completed, + Some(turn_id), + json!({}), + )) + .expect("completed latency sample") + } + + #[test] + fn correlates_content_free_stage_durations() { + let mut collector = LatencyCollector::default(); + let completed = complete_sample(&mut collector, "trigger-1", "turn-1", 7, 48, false); + + assert_eq!(completed.payload["traceId"], "trigger-1"); + assert_eq!(completed.payload["path"], "warm"); + assert_eq!(completed.payload["receiveToQueueMs"], 3); + assert_eq!(completed.payload["queueWaitMs"], 10); + assert_eq!(completed.payload["sessionResolveMs"], 2); + assert_eq!(completed.payload["postSessionSetupMs"], 3); + assert_eq!(completed.payload["turnSetupMs"], 5); + assert_eq!(completed.payload["timeToFirstOutputMs"], 15); + assert_eq!(completed.payload["firstOutputToReplyMs"], 15); + assert_eq!(completed.payload["turnDurationMs"], 40); + assert_eq!(completed.payload["totalMs"], 48); + assert_eq!(completed.payload["summary"]["totalMs"]["p50"], 48); + assert_eq!(completed.payload["summary"]["totalMs"]["p95"], 48); + assert_eq!(completed.payload["summary"]["totalMs"]["max"], 48); + + let serialized = serde_json::to_string(&completed.payload).unwrap(); + for forbidden in ["message", "content", "prompt", "modelOutput", "credential"] { + assert!( + !serialized.contains(forbidden), + "performance telemetry must not contain {forbidden}: {serialized}" + ); + } + } + + #[test] + fn keeps_warm_and_cold_percentile_windows_separate() { + let mut collector = LatencyCollector::default(); + let first = complete_sample(&mut collector, "warm-1", "turn-1", 0, 20, false); + assert_eq!(first.payload["summary"]["windowSize"], 1); + + let second = complete_sample(&mut collector, "warm-2", "turn-2", 100, 40, false); + assert_eq!(second.payload["summary"]["windowSize"], 2); + assert_eq!(second.payload["summary"]["totalMs"]["p50"], 20); + assert_eq!(second.payload["summary"]["totalMs"]["p95"], 40); + assert_eq!(second.payload["summary"]["totalMs"]["max"], 40); + + let cold = complete_sample(&mut collector, "cold-1", "turn-3", 200, 80, true); + assert_eq!(cold.payload["path"], "cold"); + assert_eq!(cold.payload["summary"]["windowSize"], 1); + assert_eq!(cold.payload["summary"]["totalMs"]["p50"], 80); + } + + #[test] + fn correlates_flat_thread_reply_after_completion_from_channel() { + let mut collector = LatencyCollector::default(); + collector.ingest(&event( + "event_received", + 10, + None, + json!({"eventId": "trigger-1", "receiptToObserverMs": 2}), + )); + collector.ingest(&event( + "event_queued", + 10, + None, + json!({"eventId": "trigger-1"}), + )); + collector.ingest(&event( + "turn_started", + 15, + Some("turn-1"), + json!({"triggeringEventIds": ["trigger-1"]}), + )); + collector.ingest(&event("turn_completed", 30, Some("turn-1"), json!({}))); + + let completed = collector + .ingest(&event( + "reply_observed", + 35, + None, + json!({ + "eventId": "reply-1", + "parentEventId": "older-thread-root", + }), + )) + .expect("channel fallback should recover the completed flat-thread turn"); + assert_eq!(completed.payload["traceId"], "trigger-1"); + assert_eq!(completed.payload["totalMs"], 27); + } + + #[test] + fn prunes_incomplete_traces_after_ttl() { + let mut collector = LatencyCollector::default(); + collector.ingest(&event( + "event_received", + 10, + None, + json!({"eventId": "stale", "receiptToObserverMs": 1}), + )); + collector.ingest(&event( + "event_queued", + 10, + None, + json!({"eventId": "stale"}), + )); + collector.ingest(&event( + "turn_started", + 20, + Some("stale-turn"), + json!({"triggeringEventIds": ["stale"]}), + )); + collector.ingest(&event( + "event_received", + TRACE_TTL_MS + 21, + None, + json!({"eventId": "fresh", "receiptToObserverMs": 1}), + )); + + assert!(!collector.triggers.contains_key("stale")); + assert!(!collector.turns.contains_key("stale-turn")); + assert!(!collector.event_to_turn.contains_key("stale")); + } +} diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index 923e132167..494ee7656b 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -4,6 +4,7 @@ mod acp; mod config; mod engram_fetch; mod filter; +mod latency; mod observer; mod pool; mod queue; @@ -1348,10 +1349,15 @@ async fn tokio_main() -> Result<()> { .filter(|s| !s.is_empty()) .and_then(|s| buzz_sdk::nip_oa::parse_auth_tag(&s).ok()); - let mut relay = - HarnessRelay::connect(&config.relay_url, &config.keys, &pubkey_hex, relay_auth_tag) - .await - .map_err(|e| anyhow::anyhow!("relay connect error: {e}"))?; + let mut relay = HarnessRelay::connect( + &config.relay_url, + &config.keys, + &pubkey_hex, + relay_auth_tag, + config.relay_observer && config.ignore_self, + ) + .await + .map_err(|e| anyhow::anyhow!("relay connect error: {e}"))?; // Tell the relay background task the watermark so it can use // `since = watermark - 5s` on the first REQ instead of `since=now`. @@ -1827,6 +1833,7 @@ async fn tokio_main() -> Result<()> { let _ = result_rx; // end split borrow before relay handling match buzz_event { Some(buzz_event) => { + let event_received_at = buzz_event.received_at; let kind_u32 = buzz_event.event.kind.as_u16() as u32; if kind_u32 == KIND_MEMBER_ADDED_NOTIFICATION @@ -1941,7 +1948,38 @@ async fn tokio_main() -> Result<()> { continue; } - if config.ignore_self && buzz_event.event.pubkey.to_hex() == pubkey_hex { + let is_self_authored = buzz_event.event.pubkey.to_hex() == pubkey_hex; + if is_self_authored + && kind_u32 == KIND_STREAM_MESSAGE + && config.relay_observer + { + if let Some(ref observer) = observer { + let thread_tags = queue::parse_thread_tags(&buzz_event.event); + let active_turn = pool + .task_map() + .values() + .find(|meta| meta.channel_id == Some(buzz_event.channel_id)); + let agent_index = active_turn.map(|meta| meta.agent_index); + let turn_id = active_turn.map(|meta| meta.turn_id.clone()); + observer.emit( + "reply_observed", + agent_index, + &observer::context_for( + Some(buzz_event.channel_id), + None, + turn_id, + ), + serde_json::json!({ + "eventId": buzz_event.event.id.to_hex(), + "relayCreatedAt": buzz_event.event.created_at.as_secs(), + "rootEventId": thread_tags.root_event_id, + "parentEventId": thread_tags.parent_event_id, + }), + ); + } + } + + if config.ignore_self && is_self_authored { tracing::debug!(channel_id = %buzz_event.channel_id, "dropping self-authored event"); continue; } @@ -2093,6 +2131,7 @@ async fn tokio_main() -> Result<()> { // buzz_event.event (needed for mode gate below). let author_hex = buzz_event.event.pubkey.to_hex(); let event_id_hex = buzz_event.event.id.to_hex(); + let event_created_at = buzz_event.event.created_at.as_secs(); // Clone for the non-cancelling steer fork, which // needs the event to render the steer body. The // clone is unconditional because we don't know @@ -2105,6 +2144,24 @@ async fn tokio_main() -> Result<()> { // backed payload) so the cost is negligible. let event_for_steer = buzz_event.event.clone(); let prompt_tag_for_steer = prompt_tag.clone(); + if let Some(ref observer) = observer { + observer.emit( + "event_received", + None, + &observer::context_for( + Some(buzz_event.channel_id), + None, + None, + ), + serde_json::json!({ + "eventId": event_id_hex.clone(), + "relayCreatedAt": event_created_at, + "receiptToObserverMs": latency::duration_ms( + event_received_at.elapsed() + ), + }), + ); + } let accepted = queue.push(QueuedEvent { channel_id: buzz_event.channel_id, event: buzz_event.event, @@ -2117,6 +2174,21 @@ async fn tokio_main() -> Result<()> { // guard's cleanup may race with this add, leaving a // cosmetic stale 👀. Acceptable — see ReactionGuard docs. if accepted { + if let Some(ref observer) = observer { + observer.emit( + "event_queued", + None, + &observer::context_for( + Some(buzz_event.channel_id), + None, + None, + ), + serde_json::json!({ + "eventId": event_id_hex.clone(), + "relayCreatedAt": event_created_at, + }), + ); + } let rc = ctx.rest_client.clone(); let eid = event_id_hex.clone(); tokio::spawn(async move { @@ -4264,6 +4336,7 @@ mod observer_chunk_coalescer_tests { ) -> observer::ObserverEvent { observer::ObserverEvent { seq, + monotonic_ms: seq, timestamp: format!("2026-04-29T04:00:0{seq}Z"), kind: "acp_read".to_string(), agent_index: Some(0), @@ -4292,6 +4365,7 @@ mod observer_chunk_coalescer_tests { fn non_chunk_event(seq: u64) -> observer::ObserverEvent { observer::ObserverEvent { seq, + monotonic_ms: seq, timestamp: format!("2026-04-29T04:00:0{seq}Z"), kind: "turn_started".to_string(), agent_index: Some(0), @@ -5514,6 +5588,7 @@ mod observer_payload_trim_tests { fn event_with_payload(kind: &str, payload: serde_json::Value) -> observer::ObserverEvent { observer::ObserverEvent { seq: 1, + monotonic_ms: 1, timestamp: "2026-06-16T00:00:00Z".to_string(), kind: kind.to_string(), agent_index: Some(0), diff --git a/crates/buzz-acp/src/observer.rs b/crates/buzz-acp/src/observer.rs index 7029e5af6d..f53fcf4679 100644 --- a/crates/buzz-acp/src/observer.rs +++ b/crates/buzz-acp/src/observer.rs @@ -10,6 +10,7 @@ use std::{ atomic::{AtomicU64, Ordering}, Arc, Mutex, }, + time::Instant, }; use serde::Serialize; @@ -40,6 +41,8 @@ struct ObserverInner { tx: broadcast::Sender, buffer: Mutex>, seq: AtomicU64, + monotonic_origin: Instant, + latency: Mutex, } fn new_observer_handle() -> ObserverHandle { @@ -49,6 +52,8 @@ fn new_observer_handle() -> ObserverHandle { tx, buffer: Mutex::new(VecDeque::with_capacity(OBSERVER_BUFFER_CAP)), seq: AtomicU64::new(1), + monotonic_origin: Instant::now(), + latency: Mutex::new(crate::latency::LatencyCollector::default()), }), } } @@ -59,6 +64,11 @@ fn new_observer_handle() -> ObserverHandle { pub struct ObserverEvent { /// Monotonic process-local sequence number. pub seq: u64, + /// Milliseconds since this process-local observer feed was created. + /// + /// This clock is authoritative for durations between events from the same + /// harness process. It must not be compared across processes. + pub monotonic_ms: u64, /// RFC3339 UTC timestamp. pub timestamp: String, /// Observer event kind, for example `acp_read` or `turn_started`. @@ -110,6 +120,8 @@ impl ObserverHandle { ) { let event = ObserverEvent { seq: self.inner.seq.fetch_add(1, Ordering::Relaxed), + monotonic_ms: u64::try_from(self.inner.monotonic_origin.elapsed().as_millis()) + .unwrap_or(u64::MAX), timestamp: chrono::Utc::now().to_rfc3339(), kind: kind.into(), agent_index, @@ -132,7 +144,27 @@ impl ObserverHandle { } } - let _ = self.inner.tx.send(event); + let _ = self.inner.tx.send(event.clone()); + + // Build content-free mention-to-reply traces synchronously from the + // semantic observer boundaries. Synchronous ingestion preserves the + // source event order and guarantees the derived sample is buffered + // immediately after the event that completed it. + let completed = match self.inner.latency.lock() { + Ok(mut latency) => latency.ingest(&event), + Err(error) => { + tracing::warn!(target: "observer", "latency collector lock poisoned: {error}"); + None + } + }; + if let Some(completed) = completed { + self.emit( + "mention_reply_latency", + completed.agent_index, + &completed.context, + completed.payload, + ); + } } } @@ -164,3 +196,104 @@ pub fn context_for_turn( started_at: Some(started_at), } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn observer_serializes_process_local_monotonic_time() { + let observer = ObserverHandle::in_process(); + observer.emit( + "first", + None, + &ObserverContext::default(), + serde_json::json!({}), + ); + observer.emit( + "second", + None, + &ObserverContext::default(), + serde_json::json!({}), + ); + + let events = observer.snapshot(); + assert_eq!(events.len(), 2); + assert!(events[0].monotonic_ms <= events[1].monotonic_ms); + assert_eq!( + serde_json::to_value(&events[0]).unwrap()["monotonicMs"], + events[0].monotonic_ms + ); + } + + #[test] + fn observer_emits_derived_latency_sample() { + let observer = ObserverHandle::in_process(); + let channel_id = uuid::Uuid::parse_str("11111111-1111-1111-1111-111111111111").unwrap(); + let trigger_context = context_for(Some(channel_id), None, None); + let turn_context = context_for_turn( + Some(channel_id), + Some("session-1".into()), + "turn-1".into(), + "2026-07-22T00:00:00Z".into(), + ); + + observer.emit( + "event_received", + None, + &trigger_context, + serde_json::json!({"eventId": "trigger-1", "receiptToObserverMs": 0}), + ); + observer.emit( + "event_queued", + None, + &trigger_context, + serde_json::json!({"eventId": "trigger-1"}), + ); + observer.emit( + "turn_started", + Some(0), + &turn_context, + serde_json::json!({"triggeringEventIds": ["trigger-1"]}), + ); + observer.emit( + "session_resolved", + Some(0), + &turn_context, + serde_json::json!({"isNewSession": false}), + ); + observer.emit( + "prompt_dispatched", + Some(0), + &turn_context, + serde_json::json!({}), + ); + observer.emit( + "turn_first_output", + Some(0), + &turn_context, + serde_json::json!({}), + ); + observer.emit( + "reply_observed", + Some(0), + &turn_context, + serde_json::json!({"eventId": "reply-1"}), + ); + observer.emit( + "turn_completed", + Some(0), + &turn_context, + serde_json::json!({}), + ); + + let events = observer.snapshot(); + let sample = events + .iter() + .find(|event| event.kind == "mention_reply_latency") + .expect("derived latency sample"); + assert_eq!(sample.payload["traceId"], "trigger-1"); + assert_eq!(sample.payload["path"], "warm"); + assert_eq!(sample.payload["summary"]["windowSize"], 1); + } +} diff --git a/crates/buzz-acp/src/relay.rs b/crates/buzz-acp/src/relay.rs index ce7d565f12..978e6537a1 100644 --- a/crates/buzz-acp/src/relay.rs +++ b/crates/buzz-acp/src/relay.rs @@ -115,7 +115,7 @@ use std::time::Instant; use buzz_core::kind::{ KIND_AGENT_OBSERVER_FRAME, KIND_MEMBER_ADDED_NOTIFICATION, KIND_MEMBER_REMOVED_NOTIFICATION, - KIND_TYPING_INDICATOR, + KIND_STREAM_MESSAGE, KIND_TYPING_INDICATOR, }; use futures_util::{SinkExt, StreamExt}; use nostr::{Event, EventBuilder, Keys, Kind, RelayUrl, Tag}; @@ -418,6 +418,8 @@ pub struct BuzzEvent { pub channel_id: Uuid, /// The underlying Nostr event. pub event: Event, + /// Process-local instant when the relay frame was accepted by the harness. + pub received_at: Instant, } /// Errors from relay operations. @@ -586,6 +588,7 @@ impl HarnessRelay { keys: &Keys, agent_pubkey_hex: &str, auth_tag: Option, + observe_self_replies: bool, ) -> Result { // Perform the initial connection and auth handshake, retrying // transient failures (dropped handshake, timeout) with bounded @@ -616,6 +619,7 @@ impl HarnessRelay { bg_relay_url, bg_agent_pubkey_hex, bg_auth_tag, + observe_self_replies, ) .await; }); @@ -981,6 +985,9 @@ struct BgState { membership_sub_active: bool, /// Whether the observer control subscription is active. observer_control_sub_active: bool, + /// Whether channel REQs also include self-authored kind-9 replies so the + /// observer can close mention-to-reply traces at relay fanout. + observe_self_replies: bool, /// Oldest dropped channel-event timestamp per channel, keyed by channel_id. /// Mirrors `membership_dropped_since` but for ordinary channel events. /// On reconnect resubscribe, `since` = min(last_seen, channel_dropped_since). @@ -1058,6 +1065,7 @@ impl BgState { membership_last_seen: None, membership_sub_active: false, observer_control_sub_active: false, + observe_self_replies: false, channel_dropped_since: HashMap::new(), proactive_resubscribe_needed: false, startup_watermark: None, @@ -1529,8 +1537,10 @@ async fn run_background_task( relay_url: String, agent_pubkey_hex: String, auth_tag: Option, + observe_self_replies: bool, ) { let mut state = BgState::new(); + state.observe_self_replies = observe_self_replies; let handshake_ok = process_handshake_buffer( &mut ws, @@ -2090,6 +2100,7 @@ async fn handle_ws_message( let buzz_event = BuzzEvent { channel_id: channel_uuid, event: *event, + received_at: Instant::now(), }; let cap = event_tx.max_capacity(); let used = cap - event_tx.capacity(); @@ -2131,6 +2142,7 @@ async fn handle_ws_message( let buzz_event = BuzzEvent { channel_id, event: *event, + received_at: Instant::now(), }; // Warn at 80% capacity. let cap = event_tx.max_capacity(); @@ -3141,13 +3153,16 @@ async fn wait_for_reconnect( /// - `kinds` is included only when `filter.kinds` is `Some`; `None` = wildcard. /// - `#p` is included only when `filter.require_mention` is `true`. /// - `#h` is always included (channel-scoped subscription). +/// - When relay observation is enabled, mention-only subscriptions add a +/// second OR-filter for self-authored kind-9 replies. The harness consumes +/// those events as telemetry and never dispatches them back to the agent. /// - On first subscribe (`since` is `None`) adds `since=now` to avoid replaying /// history. On reconnect (`since` is `Some`) subtracts [`SINCE_SKEW_SECS`]. /// /// Returns `true` if the REQ was successfully written to the WebSocket. async fn send_subscribe( ws: &mut WsStream, - _state: &BgState, + state: &BgState, channel_id: Uuid, agent_pubkey_hex: &str, since: Option, @@ -3181,7 +3196,16 @@ async fn send_subscribe( }; req_filter.insert("since".into(), json!(since_ts)); - let req = json!(["REQ", sub_id, Value::Object(req_filter)]); + let mut req = vec![json!("REQ"), json!(sub_id), Value::Object(req_filter)]; + if filter.require_mention && state.observe_self_replies { + req.push(json!({ + "kinds": [KIND_STREAM_MESSAGE], + "authors": [agent_pubkey_hex], + "#h": [channel_id.to_string()], + "since": since_ts, + })); + } + let req = Value::Array(req); match serde_json::to_string(&req) { Ok(text) => { @@ -4346,6 +4370,64 @@ mod tests { } } + #[tokio::test] + async fn observer_subscription_adds_self_reply_filter() { + let (mut client, mut server) = test_ws_pair().await; + let mut state = BgState::new(); + state.observe_self_replies = true; + let channel_id = Uuid::new_v4(); + let filter = ChannelFilter { + kinds: Some(vec![KIND_STREAM_MESSAGE]), + require_mention: true, + }; + + assert!( + send_subscribe( + &mut client, + &state, + channel_id, + "agent-pubkey", + Some(1_000), + &filter, + ) + .await + ); + + let frame = next_test_frame(&mut server).await; + assert_eq!(frame.as_array().map(Vec::len), Some(4)); + assert_eq!(frame[2]["#p"], json!(["agent-pubkey"])); + assert_eq!(frame[3]["authors"], json!(["agent-pubkey"])); + assert_eq!(frame[3]["kinds"], json!([KIND_STREAM_MESSAGE])); + assert_eq!(frame[3]["#h"], json!([channel_id.to_string()])); + assert_eq!(frame[3]["since"], 995); + } + + #[tokio::test] + async fn ordinary_subscription_does_not_add_self_reply_filter() { + let (mut client, mut server) = test_ws_pair().await; + let state = BgState::new(); + let channel_id = Uuid::new_v4(); + let filter = ChannelFilter { + kinds: Some(vec![KIND_STREAM_MESSAGE]), + require_mention: true, + }; + + assert!( + send_subscribe( + &mut client, + &state, + channel_id, + "agent-pubkey", + Some(1_000), + &filter, + ) + .await + ); + + let frame = next_test_frame(&mut server).await; + assert_eq!(frame.as_array().map(Vec::len), Some(3)); + } + fn seed_test_subscription(state: &mut BgState, channel_id: Uuid) { apply_command_to_state( state, diff --git a/crates/buzz-acp/src/setup_mode.rs b/crates/buzz-acp/src/setup_mode.rs index dda7857260..ebd4daa8ca 100644 --- a/crates/buzz-acp/src/setup_mode.rs +++ b/crates/buzz-acp/src/setup_mode.rs @@ -326,10 +326,15 @@ pub(crate) async fn run_setup_listener(config: Config, payload: SetupPayload) -> .unwrap_or_default() .as_secs(); - let mut relay = - HarnessRelay::connect(&config.relay_url, &config.keys, &pubkey_hex, relay_auth_tag) - .await - .map_err(|e| anyhow::anyhow!("setup-mode relay connect error: {e}"))?; + let mut relay = HarnessRelay::connect( + &config.relay_url, + &config.keys, + &pubkey_hex, + relay_auth_tag, + false, + ) + .await + .map_err(|e| anyhow::anyhow!("setup-mode relay connect error: {e}"))?; if let Err(e) = relay.set_startup_watermark(startup_watermark).await { tracing::warn!("setup-mode: failed to set startup watermark: {e}"); diff --git a/desktop/src/features/agents/ui/agentSessionTypes.ts b/desktop/src/features/agents/ui/agentSessionTypes.ts index 578f98076c..1d4cec07e5 100644 --- a/desktop/src/features/agents/ui/agentSessionTypes.ts +++ b/desktop/src/features/agents/ui/agentSessionTypes.ts @@ -2,6 +2,8 @@ import type { LucideIcon } from "lucide-react"; export type ObserverEvent = { seq: number; + /** Process-local monotonic milliseconds; never compare across harnesses. */ + monotonicMs?: number; timestamp: string; kind: string; agentIndex: number | null;