From 410de12008c2f77f17bdcc55fc92f7ea6d0ffea3 Mon Sep 17 00:00:00 2001 From: Broc Oppler Date: Wed, 22 Jul 2026 20:28:18 -0700 Subject: [PATCH] feat(acp): per-turn stage latency instrumentation Measure each turn's pipeline stages: relay-accept lag (wall clock, 1s), admission (receipt -> queue), queue wait (receipt -> dispatch), session setup, first model output, and turn total, threaded from the relay loop through QueuedEvent/BatchEvent into run_prompt_task. Emitted on three carriers keyed by turn_id: optional kind:44200 payload fields, the turn_completed observer frame, and one pool::metrics info summary line. Reply-publish/relay-OK stages are out-of-band today (#2459 tracks that observability). Co-Authored-By: Claude Fable 5 Signed-off-by: Broc Oppler --- crates/buzz-acp/src/acp.rs | 98 ++++++ crates/buzz-acp/src/lib.rs | 55 ++++ crates/buzz-acp/src/pool.rs | 327 ++++++++++++++++++++- crates/buzz-acp/src/queue.rs | 158 ++++++++++ crates/buzz-core/src/agent_turn_metric.rs | 103 +++++++ desktop/src-tauri/src/archive/mod_tests.rs | 7 + docs/nips/NIP-AM.md | 14 +- 7 files changed, 752 insertions(+), 10 deletions(-) diff --git a/crates/buzz-acp/src/acp.rs b/crates/buzz-acp/src/acp.rs index b553adaabb..2f2eacba19 100644 --- a/crates/buzz-acp/src/acp.rs +++ b/crates/buzz-acp/src/acp.rs @@ -200,6 +200,14 @@ pub struct AcpClient { /// deltas. Both goose and buzz-agent emit this notification; goose gates /// on client capability advertisement, buzz-agent emits unconditionally. goose_usage: UsageTracker, + /// Instant captured immediately before the current turn's `session/prompt` + /// request was written. Reset alongside the usage tracker at turn begin; + /// consumed by [`take_first_output_latency_ms`](Self::take_first_output_latency_ms). + prompt_sent_at: Option, + /// Instant of the FIRST `agent_message_chunk` observed after the current + /// turn's prompt was sent. Reset at turn begin; consumed by + /// [`take_first_output_latency_ms`](Self::take_first_output_latency_ms). + first_output_at: Option, } /// Recursively merge `overlay` into `base`, with `overlay` winning on scalar/shape @@ -492,6 +500,8 @@ impl AcpClient { active_run_id: None, steer_rx: None, goose_usage: UsageTracker::default(), + prompt_sent_at: None, + first_output_at: None, }) } @@ -684,6 +694,10 @@ impl AcpClient { // prompt so that any setup notifications recorded earlier are not // misattributed to this turn. self.goose_usage.begin_turn(session_id); + // Reset per-turn first-output tracking alongside the usage tracker so + // a prior turn's chunk timing can never leak into this turn. + self.prompt_sent_at = None; + self.first_output_at = None; self.last_prompt_id = Some(self.next_id); let id = self.next_id; @@ -697,6 +711,9 @@ impl AcpClient { }); tracing::debug!(target: "acp::wire", "→ {}", &serde_json::to_string(&msg).unwrap_or_default()); + // Captured immediately before the write: first-output latency is + // measured from the moment the prompt hits the agent's stdin. + self.prompt_sent_at = Some(std::time::Instant::now()); if let Err(e) = self.write_ndjson(&msg).await { self.last_prompt_id = None; self.current_hard_deadline = None; @@ -780,6 +797,24 @@ impl AcpClient { self.goose_usage.take() } + /// Consume the first-output latency for the turn that just ended: elapsed + /// milliseconds from the `session/prompt` write to the first + /// `agent_message_chunk` the agent streamed back. + /// + /// Returns `None` when no prompt was sent or the agent produced no message + /// chunk (e.g. failed, cancelled before output, or tool-only turns). + /// Clears both marks so a later call cannot report a stale turn's latency. + pub fn take_first_output_latency_ms(&mut self) -> Option { + let sent = self.prompt_sent_at.take(); + let first = self.first_output_at.take(); + match (sent, first) { + (Some(sent), Some(first)) => { + Some(first.saturating_duration_since(sent).as_millis() as u64) + } + _ => None, + } + } + /// Install a per-turn steer request channel for goose-native /// non-cancelling mid-turn delivery. /// @@ -1529,6 +1564,12 @@ impl AcpClient { match update_type { "agent_message_chunk" => { + // First model output for the in-flight turn — the mark is + // reset when the next prompt begins, so only the first chunk + // per turn lands here. + if self.first_output_at.is_none() { + self.first_output_at = Some(std::time::Instant::now()); + } if let Some(text) = update["content"]["text"].as_str() { tracing::info!(target: "acp::stream", "{text}"); } @@ -3073,6 +3114,63 @@ mod tests { }) } + /// Build a `session/update` notification carrying an `agent_message_chunk`. + fn agent_message_chunk_msg(text: &str) -> serde_json::Value { + serde_json::json!({ + "jsonrpc": "2.0", + "method": "session/update", + "params": { + "sessionId": "test-session", + "update": { + "sessionUpdate": "agent_message_chunk", + "content": { "text": text }, + }, + } + }) + } + + /// First `agent_message_chunk` after a prompt is the first-output mark; + /// later chunks don't move it. `take_first_output_latency_ms` measures + /// from the prompt-sent mark, drains both marks, and returns `None` when + /// either mark is missing. + #[tokio::test] + async fn first_output_latency_marks_first_chunk_and_drains_on_take() { + let mut client = spawn_inert_client().await; + assert!( + client.take_first_output_latency_ms().is_none(), + "no prompt sent — no latency" + ); + + // Chunk with no prompt-sent mark: recorded, but take() yields None + // (and drains the stray mark). + let _ = client.handle_session_update(&agent_message_chunk_msg("early")); + assert!(client.first_output_at.is_some()); + assert!(client.take_first_output_latency_ms().is_none()); + assert!(client.first_output_at.is_none(), "take must drain the mark"); + + // Simulate a prompt write, then two chunks — the FIRST chunk wins. + client.prompt_sent_at = + Some(std::time::Instant::now() - std::time::Duration::from_millis(80)); + let _ = client.handle_session_update(&agent_message_chunk_msg("chunk one")); + let first_mark = client.first_output_at; + let _ = client.handle_session_update(&agent_message_chunk_msg("chunk two")); + assert_eq!( + client.first_output_at, first_mark, + "later chunks must not move the first-output mark" + ); + + let latency = client + .take_first_output_latency_ms() + .expect("both marks set — latency must be reported"); + assert!( + latency >= 80, + "latency must cover the sent→chunk gap: {latency}" + ); + + // Drained: a second take reports nothing. + assert!(client.take_first_output_latency_ms().is_none()); + } + #[tokio::test] async fn active_run_id_sets_on_string() { let mut client = spawn_inert_client().await; diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index 862732f478..fc8c8d76b8 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -1858,6 +1858,17 @@ async fn tokio_main() -> Result<()> { let _ = result_rx; // end split borrow before relay handling match buzz_event { Some(buzz_event) => { + // Turn-stage timing: capture the harness-receipt + // instant before any admission gates run, plus the + // wall-clock relay-accept lag (1s resolution) from + // the event's `created_at`. Both ride the queued + // event into per-turn stage metrics. + let relay_received_at = std::time::Instant::now(); + let relay_lag_secs = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_secs() + .saturating_sub(buzz_event.event.created_at.as_secs()); let kind_u32 = buzz_event.event.kind.as_u16() as u32; if kind_u32 == KIND_MEMBER_ADDED_NOTIFICATION @@ -2145,6 +2156,8 @@ async fn tokio_main() -> Result<()> { event: buzz_event.event, received_at: std::time::Instant::now(), prompt_tag, + relay_received_at, + relay_lag_secs, }); // 👀 — immediate "seen" reaction, only if the event // was actually queued (not dropped by DedupMode::Drop). @@ -2768,10 +2781,14 @@ fn try_native_steer( // steering (which is to inject only what's new). let (header, closing) = queue::native_steer_framing(); let event_id_hex = event.id.to_hex(); + // Timing fields are placeholders: this BatchEvent exists only to render + // the steer body via `format_event_block` and never enters queue/metrics. let be = queue::BatchEvent { event, prompt_tag: prompt_tag.clone(), received_at: std::time::Instant::now(), + relay_received_at: std::time::Instant::now(), + relay_lag_secs: 0, }; let event_block = queue::format_event_block(channel_id, None, &be, None); let body = format!("{header}\n\n[Buzz event: {prompt_tag}]\n{event_block}\n\n{closing}"); @@ -3109,6 +3126,24 @@ fn handle_prompt_result( PromptSource::Heartbeat => None, }; let turn_id = result.turn_id.clone(); + // One-line per-turn stage-latency summary, keyed by turn_id. `None` + // stages (unobserved for this turn) are omitted from the record. + if let Some(ref timings) = result.timings { + tracing::info!( + target: "pool::metrics", + turn_id = %turn_id, + channel_id = ?channel_id, + outcome = outcome_label, + relay_lag_secs = timings.relay_lag_secs, + admission_ms = timings.admission_ms, + queue_wait_ms = timings.queue_wait_ms, + session_setup_ms = timings.session_setup_ms, + first_output_ms = timings.first_output_ms, + turn_total_ms = timings.turn_total_ms, + session_reused = timings.session_reused, + "turn stage timing" + ); + } let emit_turn_error = |error_msg: &str, error_code: Option| { if let Some(ref observer) = observer { let mut payload = serde_json::json!({ @@ -4897,6 +4932,7 @@ mod error_outcome_emission_tests { turn_id: "test-turn-id".to_string(), outcome, batch: None, + timings: None, }; handle_prompt_result( @@ -5063,6 +5099,7 @@ mod error_outcome_emission_tests { turn_id: "test-turn-id".to_string(), outcome, batch: None, + timings: None, }; handle_prompt_result( &mut pool, @@ -5113,6 +5150,8 @@ mod error_outcome_emission_tests { event, prompt_tag: "test".into(), received_at: std::time::Instant::now(), + relay_received_at: std::time::Instant::now(), + relay_lag_secs: 0, }], cancelled_events: vec![], cancel_reason: None, @@ -5153,6 +5192,7 @@ mod error_outcome_emission_tests { turn_id: "test-turn-id".to_string(), outcome, batch: Some(batch), + timings: None, }; handle_prompt_result( &mut pool, @@ -5219,6 +5259,8 @@ mod error_outcome_emission_tests { event, prompt_tag: "test".into(), received_at: std::time::Instant::now(), + relay_received_at: std::time::Instant::now(), + relay_lag_secs: 0, }], cancelled_events: vec![], cancel_reason: None, @@ -5258,6 +5300,7 @@ mod error_outcome_emission_tests { turn_id: "test-turn-id".to_string(), outcome, batch: Some(batch), + timings: None, }; handle_prompt_result( &mut pool, @@ -5337,6 +5380,8 @@ mod error_outcome_emission_tests { .unwrap(), prompt_tag: "test".into(), received_at: std::time::Instant::now(), + relay_received_at: std::time::Instant::now(), + relay_lag_secs: 0, }], cancelled_events: vec![], cancel_reason: None, @@ -5349,6 +5394,7 @@ mod error_outcome_emission_tests { recently_active: true, }), batch: Some(batch), + timings: None, }; handle_prompt_result( &mut pool, @@ -5430,6 +5476,8 @@ mod error_outcome_emission_tests { .unwrap(), prompt_tag: "test".into(), received_at: std::time::Instant::now(), + relay_received_at: std::time::Instant::now(), + relay_lag_secs: 0, }], cancelled_events: vec![], cancel_reason: None, @@ -5442,6 +5490,7 @@ mod error_outcome_emission_tests { recently_active: true, }), batch: Some(batch), + timings: None, }; handle_prompt_result( &mut pool, @@ -5508,6 +5557,8 @@ mod error_outcome_emission_tests { event: original_event.clone(), prompt_tag: "test".into(), received_at: std::time::Instant::now(), + relay_received_at: std::time::Instant::now(), + relay_lag_secs: 0, }], cancelled_events: vec![], cancel_reason: Some(CancelReason::Steer), @@ -5536,6 +5587,8 @@ mod error_outcome_emission_tests { channel_id, event: new_event.clone(), received_at: std::time::Instant::now(), + relay_received_at: std::time::Instant::now(), + relay_lag_secs: 0, prompt_tag: "test".into(), }); let config = test_config(); @@ -5556,6 +5609,7 @@ mod error_outcome_emission_tests { turn_id: "test-turn-id".to_string(), outcome: PromptOutcome::CancelDrainTimeout(grace), batch: Some(batch), + timings: None, }; handle_prompt_result( @@ -5688,6 +5742,7 @@ mod error_outcome_emission_tests { // `classify_control_cancel_failure` — `handle_prompt_result` // never sees one to requeue. batch: None, + timings: None, }; handle_prompt_result( diff --git a/crates/buzz-acp/src/pool.rs b/crates/buzz-acp/src/pool.rs index fb08096792..1970209944 100644 --- a/crates/buzz-acp/src/pool.rs +++ b/crates/buzz-acp/src/pool.rs @@ -21,7 +21,7 @@ use std::collections::{HashMap, HashSet}; use std::sync::{Arc, Mutex}; -use std::time::Duration; +use std::time::{Duration, Instant}; use tokio::sync::mpsc; use tokio::task::{JoinHandle, JoinSet}; @@ -226,6 +226,83 @@ pub struct PromptResult { pub outcome: PromptOutcome, /// Present on failure in Queue mode, for requeue. pub batch: Option, + /// Stage latencies for this turn, when measured. `None` only for results + /// synthesized outside `run_prompt_task` (e.g. tests). + pub timings: Option, +} + +/// Per-turn stage latencies. All values come from monotonic clocks except +/// [`relay_lag_secs`](Self::relay_lag_secs) (wall-clock, 1s resolution). +/// +/// A field is `None` when the stage was not observed for the turn: heartbeat +/// turns carry no relay/admission/queue stages, early failures never reach +/// session resolution or first output. No message content is recorded here. +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct TurnTimings { + /// Wall-clock lag between event `created_at` and harness receipt, + /// minimum over the batch. + pub relay_lag_secs: Option, + /// Harness receipt to queue admission for the batch's oldest event + /// (dedup/author/rule gates). + pub admission_ms: Option, + /// Harness receipt of the batch's oldest event to prompt-task start + /// (queueing, retries, and dispatch wait). + pub queue_wait_ms: Option, + /// Prompt-task start to session resolution (reused or freshly created). + pub session_setup_ms: Option, + /// `session/prompt` write to the first `agent_message_chunk`. + pub first_output_ms: Option, + /// Prompt-task start to turn completion (any outcome). + pub turn_total_ms: Option, + /// `true` when the turn ran on an existing session, `false` when a new + /// session was created for it. + pub session_reused: Option, +} + +impl TurnTimings { + /// Serialize the known stages into a camelCase JSON object for observer + /// frames, omitting unknown (`None`) stages. + fn to_observer_json(&self) -> serde_json::Value { + let mut map = serde_json::Map::new(); + let mut put_u64 = |key: &str, value: Option| { + if let Some(v) = value { + map.insert(key.to_string(), serde_json::json!(v)); + } + }; + put_u64("relayLagSecs", self.relay_lag_secs); + put_u64("admissionMs", self.admission_ms); + put_u64("queueWaitMs", self.queue_wait_ms); + put_u64("sessionSetupMs", self.session_setup_ms); + put_u64("firstOutputMs", self.first_output_ms); + put_u64("turnTotalMs", self.turn_total_ms); + if let Some(reused) = self.session_reused { + map.insert("sessionReused".to_string(), serde_json::json!(reused)); + } + serde_json::Value::Object(map) + } +} + +/// Milliseconds from `from` to `to`, saturating at zero if `to` is earlier. +fn duration_ms(from: Instant, to: Instant) -> u64 { + to.saturating_duration_since(from).as_millis() as u64 +} + +/// Complete a turn's [`TurnTimings`] at an exit site of `run_prompt_task`: +/// stamps the total duration and first-output latency (consumed from the +/// [`AcpClient`]), records the result on the completion guard so the +/// `turn_completed` observer frame carries it, and returns the final value +/// for the kind:44200 payload and the [`PromptResult`]. +fn finalize_turn_timings( + base: &TurnTimings, + turn_started: Instant, + acp: &mut AcpClient, + guard: &TurnCompletionGuard, +) -> TurnTimings { + let mut timings = base.clone(); + timings.first_output_ms = acp.take_first_output_latency_ms(); + timings.turn_total_ms = Some(duration_ms(turn_started, Instant::now())); + guard.set_timings(timings.clone()); + timings } /// Whether the prompt came from a channel event or a heartbeat. @@ -1187,6 +1264,7 @@ fn send_prompt_result( source: PromptSource, outcome: PromptOutcome, batch: Option, + timings: Option, ) { agent.acp.clear_steer_rx(); let _ = result_tx.send(PromptResult { @@ -1195,6 +1273,7 @@ fn send_prompt_result( turn_id: turn_id.to_owned(), outcome, batch, + timings, }); } @@ -1229,6 +1308,19 @@ pub async fn run_prompt_task( PromptSource::Heartbeat => None, }; let turn_started_at = chrono::Utc::now().to_rfc3339(); + // Monotonic twin of `turn_started_at` — anchor for all stage latencies. + let turn_started = Instant::now(); + // Pre-task stages, measured from the batch's oldest event (the one that + // has waited longest and whose receipt anchors `queue_wait_ms`). + // Heartbeats have no batch and therefore no relay/admission/queue stages. + let mut timings = TurnTimings::default(); + if let Some(b) = &batch { + timings.relay_lag_secs = b.events.iter().map(|be| be.relay_lag_secs).min(); + if let Some(oldest) = b.events.iter().min_by_key(|be| be.relay_received_at) { + timings.admission_ms = Some(duration_ms(oldest.relay_received_at, oldest.received_at)); + timings.queue_wait_ms = Some(duration_ms(oldest.relay_received_at, turn_started)); + } + } agent.acp.set_observer_context(observer::context_for_turn( observer_channel_id, None, @@ -1253,8 +1345,10 @@ pub async fn run_prompt_task( // Emits `turn_completed` on any exit path. Captures observer handle and // metadata now, before the agent is moved into PromptResult. It must be // declared before `liveness_guard`: Rust drops locals in reverse order, so - // liveness is aborted before completion makes the turn terminal. - let _turn_guard = TurnCompletionGuard::new( + // liveness is aborted before completion makes the turn terminal. Exit + // sites record stage timings on it via `finalize_turn_timings` so the + // `turn_completed` frame carries them. + let turn_guard = TurnCompletionGuard::new( agent.acp.observer_handle(), agent.acp.observer_agent_index(), observer_channel_id, @@ -1442,6 +1536,12 @@ pub async fn run_prompt_task( } Err(AcpError::AgentExited) => { agent.state.invalidate_all(); + let turn_timings = finalize_turn_timings( + &timings, + turn_started, + &mut agent.acp, + &turn_guard, + ); send_prompt_result( &result_tx, &turn_id, @@ -1449,12 +1549,19 @@ pub async fn run_prompt_task( source, PromptOutcome::AgentExited, requeue_batch_if_queue(&ctx, batch), + Some(turn_timings), ); return; } Err(e) => { // Session creation failed; pending canvas was never committed, // so the next retry will re-fetch a fresh revision. + let turn_timings = finalize_turn_timings( + &timings, + turn_started, + &mut agent.acp, + &turn_guard, + ); send_prompt_result( &result_tx, &turn_id, @@ -1462,6 +1569,7 @@ pub async fn run_prompt_task( source, PromptOutcome::Error(e), requeue_batch_if_queue(&ctx, batch), + Some(turn_timings), ); return; } @@ -1484,6 +1592,12 @@ pub async fn run_prompt_task( } Err(AcpError::AgentExited) => { agent.state.invalidate_all(); + let turn_timings = finalize_turn_timings( + &timings, + turn_started, + &mut agent.acp, + &turn_guard, + ); send_prompt_result( &result_tx, &turn_id, @@ -1491,10 +1605,17 @@ pub async fn run_prompt_task( source, PromptOutcome::AgentExited, None, + Some(turn_timings), ); return; } Err(e) => { + let turn_timings = finalize_turn_timings( + &timings, + turn_started, + &mut agent.acp, + &turn_guard, + ); send_prompt_result( &result_tx, &turn_id, @@ -1502,6 +1623,7 @@ pub async fn run_prompt_task( source, PromptOutcome::Error(e), None, + Some(turn_timings), ); return; } @@ -1509,6 +1631,10 @@ pub async fn run_prompt_task( } } }; + // Session resolution stage complete (covers core/canvas fetches and any + // session creation above; ~0 for a reused session). + timings.session_setup_ms = Some(duration_ms(turn_started, Instant::now())); + timings.session_reused = Some(!is_new_session); agent.acp.set_observer_context(observer::context_for_turn( observer_channel_id, Some(session_id.clone()), @@ -1576,6 +1702,8 @@ pub async fn run_prompt_task( } Err(AcpError::AgentExited) => { agent.state.invalidate_all(); + let turn_timings = + finalize_turn_timings(&timings, turn_started, &mut agent.acp, &turn_guard); send_prompt_result( &result_tx, &turn_id, @@ -1583,6 +1711,7 @@ pub async fn run_prompt_task( source, PromptOutcome::AgentExited, requeue_batch_if_queue(&ctx, batch), + Some(turn_timings), ); return; } @@ -1602,6 +1731,12 @@ pub async fn run_prompt_task( } Err(AcpError::AgentExited) => { agent.state.invalidate_all(); + let turn_timings = finalize_turn_timings( + &timings, + turn_started, + &mut agent.acp, + &turn_guard, + ); send_prompt_result( &result_tx, &turn_id, @@ -1609,6 +1744,7 @@ pub async fn run_prompt_task( source, PromptOutcome::AgentExited, requeue_batch_if_queue(&ctx, batch), + Some(turn_timings), ); return; } @@ -1620,6 +1756,8 @@ pub async fn run_prompt_task( agent.state.invalidate(&source); } } + let turn_timings = + finalize_turn_timings(&timings, turn_started, &mut agent.acp, &turn_guard); send_prompt_result( &result_tx, &turn_id, @@ -1627,6 +1765,7 @@ pub async fn run_prompt_task( source, PromptOutcome::Timeout(TimeoutKind::Idle), requeue_batch_if_queue(&ctx, batch), + Some(turn_timings), ); return; } @@ -1638,6 +1777,8 @@ pub async fn run_prompt_task( ctx.max_turn_duration.as_secs() ); agent.state.invalidate_all(); + let turn_timings = + finalize_turn_timings(&timings, turn_started, &mut agent.acp, &turn_guard); send_prompt_result( &result_tx, &turn_id, @@ -1645,6 +1786,7 @@ pub async fn run_prompt_task( source, PromptOutcome::Timeout(TimeoutKind::Hard { recently_active }), requeue_batch_if_queue(&ctx, batch), + Some(turn_timings), ); return; } @@ -1654,6 +1796,8 @@ pub async fn run_prompt_task( "initial_message failed for channel {cid}: {e} — invalidating session" ); agent.state.invalidate(&source); + let turn_timings = + finalize_turn_timings(&timings, turn_started, &mut agent.acp, &turn_guard); send_prompt_result( &result_tx, &turn_id, @@ -1661,6 +1805,7 @@ pub async fn run_prompt_task( source, PromptOutcome::Error(e), requeue_batch_if_queue(&ctx, batch), + Some(turn_timings), ); return; } @@ -1741,6 +1886,8 @@ pub async fn run_prompt_task( // Should not happen — batch is None only for heartbeats which have prompt_text. // Return the agent to the pool to prevent a permanent slot leak. tracing::error!("run_prompt_task: no batch and no prompt_text — returning agent"); + let turn_timings = + finalize_turn_timings(&timings, turn_started, &mut agent.acp, &turn_guard); send_prompt_result( &result_tx, &turn_id, @@ -1748,6 +1895,7 @@ pub async fn run_prompt_task( source, PromptOutcome::Error(AcpError::Protocol("no batch and no prompt_text".into())), None, + Some(turn_timings), ); return; }; @@ -1827,6 +1975,12 @@ pub async fn run_prompt_task( requeue_cancelled_batch(&ctx, control_signal, batch); let usage = agent.acp.take_turn_usage(); + let turn_timings = finalize_turn_timings( + &timings, + turn_started, + &mut agent.acp, + &turn_guard, + ); publish_agent_turn_metric( &ctx, usage, @@ -1834,6 +1988,7 @@ pub async fn run_prompt_task( &session_id, &turn_id, Some(buzz_core::agent_turn_metric::StopReason::Cancelled), + &turn_timings, ) .await; send_prompt_result( @@ -1843,6 +1998,7 @@ pub async fn run_prompt_task( source, PromptOutcome::Cancelled, retry_batch, + Some(turn_timings), ); return; } @@ -1863,6 +2019,12 @@ pub async fn run_prompt_task( } let usage = agent.acp.take_turn_usage(); + let turn_timings = finalize_turn_timings( + &timings, + turn_started, + &mut agent.acp, + &turn_guard, + ); publish_agent_turn_metric( &ctx, usage, @@ -1870,6 +2032,7 @@ pub async fn run_prompt_task( &session_id, &turn_id, Some(buzz_core::agent_turn_metric::StopReason::Error), + &turn_timings, ) .await; send_prompt_result( @@ -1879,6 +2042,7 @@ pub async fn run_prompt_task( source, failure.outcome, failure.retry_batch, + Some(turn_timings), ); return; } @@ -1918,6 +2082,12 @@ pub async fn run_prompt_task( &control_signal, ); let usage = agent.acp.take_turn_usage(); + let turn_timings = finalize_turn_timings( + &timings, + turn_started, + &mut agent.acp, + &turn_guard, + ); publish_agent_turn_metric( &ctx, usage, @@ -1925,6 +2095,7 @@ pub async fn run_prompt_task( &session_id, &turn_id, Some(buzz_core::agent_turn_metric::StopReason::EndTurn), + &turn_timings, ) .await; send_prompt_result( @@ -1934,6 +2105,7 @@ pub async fn run_prompt_task( source, PromptOutcome::Ok(StopReason::EndTurn), None, // turn succeeded — batch was processed, no requeue + Some(turn_timings), ); return; } @@ -1980,6 +2152,8 @@ pub async fn run_prompt_task( let core_stop = acp_stop_to_core(&stop_reason); let usage = agent.acp.take_turn_usage(); + let turn_timings = + finalize_turn_timings(&timings, turn_started, &mut agent.acp, &turn_guard); publish_agent_turn_metric( &ctx, usage, @@ -1987,6 +2161,7 @@ pub async fn run_prompt_task( &session_id, &turn_id, Some(core_stop), + &turn_timings, ) .await; @@ -1997,12 +2172,15 @@ pub async fn run_prompt_task( source, PromptOutcome::Ok(stop_reason), None, + Some(turn_timings), ); } Err(AcpError::AgentExited) => { tracing::error!(target: "pool::prompt", "agent {} exited during prompt", agent.index); agent.state.invalidate_all(); let usage = agent.acp.take_turn_usage(); + let turn_timings = + finalize_turn_timings(&timings, turn_started, &mut agent.acp, &turn_guard); publish_agent_turn_metric( &ctx, usage, @@ -2010,6 +2188,7 @@ pub async fn run_prompt_task( &session_id, &turn_id, Some(buzz_core::agent_turn_metric::StopReason::Error), + &turn_timings, ) .await; send_prompt_result( @@ -2019,6 +2198,7 @@ pub async fn run_prompt_task( source, PromptOutcome::AgentExited, requeue_batch_if_queue(&ctx, batch), + Some(turn_timings), ); } Err(AcpError::IdleTimeout(_)) => { @@ -2035,6 +2215,8 @@ pub async fn run_prompt_task( Ok(stop_reason) => { log_stop_reason(&source, &stop_reason); let usage = agent.acp.take_turn_usage(); + let turn_timings = + finalize_turn_timings(&timings, turn_started, &mut agent.acp, &turn_guard); publish_agent_turn_metric( &ctx, usage, @@ -2042,6 +2224,7 @@ pub async fn run_prompt_task( &session_id, &turn_id, Some(buzz_core::agent_turn_metric::StopReason::Cancelled), + &turn_timings, ) .await; // Timeout triggers respawn in handle_prompt_result — @@ -2053,6 +2236,7 @@ pub async fn run_prompt_task( source, PromptOutcome::Timeout(TimeoutKind::Idle), requeue_batch_if_queue(&ctx, batch), + Some(turn_timings), ); } Err(AcpError::AgentExited) => { @@ -2063,6 +2247,8 @@ pub async fn run_prompt_task( ); agent.state.invalidate_all(); let usage = agent.acp.take_turn_usage(); + let turn_timings = + finalize_turn_timings(&timings, turn_started, &mut agent.acp, &turn_guard); publish_agent_turn_metric( &ctx, usage, @@ -2070,6 +2256,7 @@ pub async fn run_prompt_task( &session_id, &turn_id, Some(buzz_core::agent_turn_metric::StopReason::Error), + &turn_timings, ) .await; send_prompt_result( @@ -2079,6 +2266,7 @@ pub async fn run_prompt_task( source, PromptOutcome::AgentExited, requeue_batch_if_queue(&ctx, batch), + Some(turn_timings), ); } Err(e) => { @@ -2088,6 +2276,8 @@ pub async fn run_prompt_task( ); agent.state.invalidate(&source); let usage = agent.acp.take_turn_usage(); + let turn_timings = + finalize_turn_timings(&timings, turn_started, &mut agent.acp, &turn_guard); publish_agent_turn_metric( &ctx, usage, @@ -2095,6 +2285,7 @@ pub async fn run_prompt_task( &session_id, &turn_id, Some(buzz_core::agent_turn_metric::StopReason::Error), + &turn_timings, ) .await; send_prompt_result( @@ -2104,6 +2295,7 @@ pub async fn run_prompt_task( source, PromptOutcome::Timeout(TimeoutKind::Idle), requeue_batch_if_queue(&ctx, batch), + Some(turn_timings), ); } } @@ -2117,6 +2309,8 @@ pub async fn run_prompt_task( ); agent.state.invalidate_all(); let usage = agent.acp.take_turn_usage(); + let turn_timings = + finalize_turn_timings(&timings, turn_started, &mut agent.acp, &turn_guard); publish_agent_turn_metric( &ctx, usage, @@ -2124,6 +2318,7 @@ pub async fn run_prompt_task( &session_id, &turn_id, Some(buzz_core::agent_turn_metric::StopReason::Error), + &turn_timings, ) .await; send_prompt_result( @@ -2133,6 +2328,7 @@ pub async fn run_prompt_task( source, PromptOutcome::Timeout(TimeoutKind::Hard { recently_active }), requeue_batch_if_queue(&ctx, batch), + Some(turn_timings), ); } Err(e) => { @@ -2144,6 +2340,8 @@ pub async fn run_prompt_task( agent.state.invalidate(&source); } let usage = agent.acp.take_turn_usage(); + let turn_timings = + finalize_turn_timings(&timings, turn_started, &mut agent.acp, &turn_guard); publish_agent_turn_metric( &ctx, usage, @@ -2151,6 +2349,7 @@ pub async fn run_prompt_task( &session_id, &turn_id, Some(buzz_core::agent_turn_metric::StopReason::Error), + &turn_timings, ) .await; send_prompt_result( @@ -2160,6 +2359,7 @@ pub async fn run_prompt_task( source, PromptOutcome::Error(e), requeue_batch_if_queue(&ctx, batch), + Some(turn_timings), ); } } @@ -3232,6 +3432,10 @@ struct TurnCompletionGuard { agent_index: Option, channel_id: Option, turn_id: String, + /// Stage timings recorded by `finalize_turn_timings` at the exit site, + /// carried into the `turn_completed` frame payload. Interior mutability + /// because the guard is held by shared reference for the whole turn. + timings: Mutex>, } impl TurnCompletionGuard { @@ -3246,6 +3450,15 @@ impl TurnCompletionGuard { agent_index, channel_id, turn_id, + timings: Mutex::new(None), + } + } + + /// Record the turn's stage timings for the `turn_completed` frame. + /// Later calls overwrite earlier ones; the exit-site value wins. + fn set_timings(&self, timings: TurnTimings) { + if let Ok(mut slot) = self.timings.lock() { + *slot = Some(timings); } } } @@ -3254,12 +3467,13 @@ impl Drop for TurnCompletionGuard { fn drop(&mut self) { if let Some(observer) = self.observer.take() { let context = observer::context_for(self.channel_id, None, Some(self.turn_id.clone())); - observer.emit( - "turn_completed", - self.agent_index, - &context, - serde_json::json!({}), - ); + let payload = self + .timings + .lock() + .ok() + .and_then(|slot| slot.as_ref().map(TurnTimings::to_observer_json)) + .unwrap_or_else(|| serde_json::json!({})); + observer.emit("turn_completed", self.agent_index, &context, payload); } } } @@ -3289,6 +3503,7 @@ async fn publish_agent_turn_metric( session_id: &str, turn_id: &str, stop_reason: Option, + timings: &TurnTimings, ) { use buzz_core::agent_turn_metric::{AgentTurnMetricPayload, TokenCounts}; use nostr::{EventBuilder, Kind, Tag}; @@ -3335,6 +3550,13 @@ async fn publish_agent_turn_metric( cumulative: cumulative_counts, delta_reliable: usage.delta_reliable, stop_reason, + relay_lag_secs: timings.relay_lag_secs, + admission_ms: timings.admission_ms, + queue_wait_ms: timings.queue_wait_ms, + session_setup_ms: timings.session_setup_ms, + first_output_ms: timings.first_output_ms, + turn_total_ms: timings.turn_total_ms, + session_reused: timings.session_reused, }; let ciphertext = match buzz_core::agent_turn_metric::encrypt_agent_turn_metric( &ctx.agent_keys, @@ -4086,6 +4308,8 @@ mod tests { event, prompt_tag: "@mention".into(), received_at: std::time::Instant::now(), + relay_received_at: std::time::Instant::now(), + relay_lag_secs: 0, }], cancelled_events: vec![], cancel_reason: None, @@ -4412,6 +4636,8 @@ mod tests { event, prompt_tag: "test".into(), received_at: std::time::Instant::now(), + relay_received_at: std::time::Instant::now(), + relay_lag_secs: 0, }], cancelled_events: vec![], cancel_reason: None, @@ -4979,6 +5205,7 @@ mod tests { source, PromptOutcome::Error(AcpError::Protocol("simulated session-create error".into())), None, + None, ); // Receive the PromptResult back from the channel. @@ -5037,6 +5264,7 @@ mod tests { source, PromptOutcome::Ok(StopReason::EndTurn), None, + None, ); let mut result = result_rx.recv().await.expect("PromptResult must be sent"); @@ -5088,6 +5316,7 @@ mod tests { "sess-1", "turn-1", Some(buzz_core::agent_turn_metric::StopReason::EndTurn), + &TurnTimings::default(), ) .await; } @@ -5116,6 +5345,7 @@ mod tests { "sess-1", "turn-1", Some(buzz_core::agent_turn_metric::StopReason::EndTurn), + &TurnTimings::default(), ) .await; } @@ -5148,6 +5378,7 @@ mod tests { "sess-1", "turn-1", Some(buzz_core::agent_turn_metric::StopReason::EndTurn), + &TurnTimings::default(), ) .await; } @@ -5181,6 +5412,7 @@ mod tests { "sess-cancel", "turn-cancel", Some(buzz_core::agent_turn_metric::StopReason::Cancelled), + &TurnTimings::default(), ) .await; } @@ -5214,10 +5446,87 @@ mod tests { "sess-ba", "turn-ba", Some(buzz_core::agent_turn_metric::StopReason::EndTurn), + &TurnTimings::default(), ) .await; } + // ── Turn stage-timing tests ──────────────────────────────────────────── + + /// `TurnTimings::to_observer_json` emits camelCase keys for known stages + /// and omits unknown (`None`) stages entirely. + #[test] + fn test_turn_timings_observer_json_omits_unknown_stages() { + assert_eq!( + TurnTimings::default().to_observer_json(), + serde_json::json!({}), + "all-None timings must serialize to an empty object" + ); + + let timings = TurnTimings { + relay_lag_secs: Some(3), + admission_ms: Some(40), + queue_wait_ms: Some(1200), + session_setup_ms: None, + first_output_ms: None, + turn_total_ms: Some(9000), + session_reused: Some(true), + }; + assert_eq!( + timings.to_observer_json(), + serde_json::json!({ + "relayLagSecs": 3, + "admissionMs": 40, + "queueWaitMs": 1200, + "turnTotalMs": 9000, + "sessionReused": true, + }) + ); + } + + /// `finalize_turn_timings` stamps `turn_total_ms`, consumes the (absent) + /// first-output latency, and records the finalized value on the + /// completion guard for the `turn_completed` frame. + #[tokio::test] + async fn test_finalize_turn_timings_stamps_total_and_records_on_guard() { + let mut acp = AcpClient::spawn( + "bash", + &["-c".to_string(), "sleep 10".to_string()], + &[], + false, + ) + .await + .expect("failed to spawn test agent"); + + let base = TurnTimings { + queue_wait_ms: Some(500), + session_reused: Some(false), + ..TurnTimings::default() + }; + let guard = TurnCompletionGuard::new(None, Some(0), None, "turn-t".to_string()); + let turn_started = Instant::now() - Duration::from_millis(50); + + let finalized = finalize_turn_timings(&base, turn_started, &mut acp, &guard); + + assert_eq!(finalized.queue_wait_ms, Some(500), "base stages preserved"); + assert_eq!(finalized.session_reused, Some(false)); + assert!( + finalized.turn_total_ms.is_some_and(|ms| ms >= 50), + "turn_total_ms must cover the elapsed turn: {:?}", + finalized.turn_total_ms + ); + assert_eq!( + finalized.first_output_ms, None, + "no prompt was sent — first output latency must be absent" + ); + let recorded = guard.timings.lock().expect("guard timings lock").clone(); + assert_eq!( + recorded, + Some(finalized), + "guard must carry the finalized timings for turn_completed" + ); + } + fn make_prompt_context_no_owner() -> PromptContext { let agent_keys = nostr::Keys::generate(); make_prompt_context_impl(&agent_keys, None) diff --git a/crates/buzz-acp/src/queue.rs b/crates/buzz-acp/src/queue.rs index 029bf86dbf..cd6a3d2e6f 100644 --- a/crates/buzz-acp/src/queue.rs +++ b/crates/buzz-acp/src/queue.rs @@ -49,6 +49,13 @@ pub struct QueuedEvent { pub received_at: Instant, /// Tag identifying which rule (or mode) matched this event. pub prompt_tag: String, + /// When the relay handed this event to the harness main loop, before any + /// admission gates ran. `received_at - relay_received_at` is the + /// admission latency (dedup/author/rule gates) for turn-stage timing. + pub relay_received_at: Instant, + /// Wall-clock relay-accept lag captured at receipt (1s resolution): + /// unix seconds now minus the event's `created_at`, saturating at 0. + pub relay_lag_secs: u64, } /// A single event inside a [`FlushBatch`]. @@ -57,6 +64,10 @@ pub struct BatchEvent { pub event: Event, pub prompt_tag: String, pub received_at: Instant, + /// See [`QueuedEvent::relay_received_at`]; preserved across requeues. + pub relay_received_at: Instant, + /// See [`QueuedEvent::relay_lag_secs`]; preserved across requeues. + pub relay_lag_secs: u64, } /// Why a batch's prior turn was cancelled — controls how `format_prompt` @@ -341,6 +352,8 @@ impl EventQueue { event: qe.event, prompt_tag: qe.prompt_tag, received_at: qe.received_at, + relay_received_at: qe.relay_received_at, + relay_lag_secs: qe.relay_lag_secs, }) .collect(); // Relay replay delivers stored events newest-first (`ORDER BY @@ -480,6 +493,8 @@ impl EventQueue { event: be.event, prompt_tag: be.prompt_tag, received_at: be.received_at, // preserve original timestamp (#46) + relay_received_at: be.relay_received_at, + relay_lag_secs: be.relay_lag_secs, }); } // Enforce per-channel cap: trim oldest (back) events if requeue pushed @@ -515,6 +530,8 @@ impl EventQueue { event: be.event, prompt_tag: be.prompt_tag, received_at: be.received_at, + relay_received_at: be.relay_received_at, + relay_lag_secs: be.relay_lag_secs, }); } // Enforce per-channel cap: trim newest (back) events if over limit. @@ -1647,6 +1664,8 @@ mod tests { event: make_event(content), received_at: Instant::now(), prompt_tag: "test".into(), + relay_received_at: Instant::now(), + relay_lag_secs: 0, } } @@ -1657,6 +1676,8 @@ mod tests { event: make_event(content), received_at: Instant::now() - age, prompt_tag: "test".into(), + relay_received_at: Instant::now() - age, + relay_lag_secs: 0, } } @@ -1677,6 +1698,8 @@ mod tests { event, received_at: Instant::now(), prompt_tag: "test".into(), + relay_received_at: Instant::now(), + relay_lag_secs: 0, } } @@ -1872,6 +1895,8 @@ mod tests { event, prompt_tag: "@mention".into(), received_at: Instant::now(), + relay_received_at: Instant::now(), + relay_lag_secs: 0, }], cancelled_events: vec![], cancel_reason: None, @@ -1902,11 +1927,15 @@ mod tests { event: make_event("the new message"), prompt_tag: "@mention".into(), received_at: Instant::now(), + relay_received_at: Instant::now(), + relay_lag_secs: 0, }], cancelled_events: vec![BatchEvent { event: make_event("the original task"), prompt_tag: "@mention".into(), received_at: Instant::now(), + relay_received_at: Instant::now(), + relay_lag_secs: 0, }], cancel_reason: reason, } @@ -2034,17 +2063,23 @@ mod tests { event: make_event("new one"), prompt_tag: "@mention".into(), received_at: Instant::now(), + relay_received_at: Instant::now(), + relay_lag_secs: 0, }, BatchEvent { event: make_event("new two"), prompt_tag: "@mention".into(), received_at: Instant::now(), + relay_received_at: Instant::now(), + relay_lag_secs: 0, }, ], cancelled_events: vec![BatchEvent { event: make_event("original"), prompt_tag: "@mention".into(), received_at: Instant::now(), + relay_received_at: Instant::now(), + relay_lag_secs: 0, }], cancel_reason: Some(CancelReason::Steer), }; @@ -2090,11 +2125,15 @@ mod tests { event: steering, prompt_tag: "@mention".into(), received_at: Instant::now(), + relay_received_at: Instant::now(), + relay_lag_secs: 0, }], cancelled_events: vec![BatchEvent { event: original, prompt_tag: "@mention".into(), received_at: Instant::now(), + relay_received_at: Instant::now(), + relay_lag_secs: 0, }], cancel_reason: Some(CancelReason::Steer), }; @@ -2183,16 +2222,22 @@ mod tests { event: e1, prompt_tag: "tag-a".into(), received_at: Instant::now(), + relay_received_at: Instant::now(), + relay_lag_secs: 0, }, BatchEvent { event: e2, prompt_tag: "tag-b".into(), received_at: Instant::now(), + relay_received_at: Instant::now(), + relay_lag_secs: 0, }, BatchEvent { event: e3, prompt_tag: "tag-c".into(), received_at: Instant::now(), + relay_received_at: Instant::now(), + relay_lag_secs: 0, }, ], cancelled_events: vec![], @@ -2222,6 +2267,8 @@ mod tests { event, prompt_tag: "test".into(), received_at: Instant::now(), + relay_received_at: Instant::now(), + relay_lag_secs: 0, }], cancelled_events: vec![], cancel_reason: None, @@ -2245,6 +2292,8 @@ mod tests { event, prompt_tag: "test".into(), received_at: Instant::now(), + relay_received_at: Instant::now(), + relay_lag_secs: 0, }], cancelled_events: vec![], cancel_reason: None, @@ -2277,6 +2326,8 @@ mod tests { event, prompt_tag: "test".into(), received_at: Instant::now(), + relay_received_at: Instant::now(), + relay_lag_secs: 0, }], cancelled_events: vec![], cancel_reason: None, @@ -2307,6 +2358,8 @@ mod tests { event, prompt_tag: "test".into(), received_at: Instant::now(), + relay_received_at: Instant::now(), + relay_lag_secs: 0, }], cancelled_events: vec![], cancel_reason: None, @@ -2334,6 +2387,8 @@ mod tests { event, prompt_tag: "test".into(), received_at: Instant::now(), + relay_received_at: Instant::now(), + relay_lag_secs: 0, }], cancelled_events: vec![], cancel_reason: None, @@ -2358,6 +2413,8 @@ mod tests { event, prompt_tag: "test".into(), received_at: Instant::now(), + relay_received_at: Instant::now(), + relay_lag_secs: 0, }], cancelled_events: vec![], cancel_reason: None, @@ -2414,6 +2471,8 @@ mod tests { event, prompt_tag: "test".into(), received_at: Instant::now(), + relay_received_at: Instant::now(), + relay_lag_secs: 0, }], cancelled_events: vec![], cancel_reason: None, @@ -2452,6 +2511,8 @@ mod tests { event, prompt_tag: "test".into(), received_at: Instant::now(), + relay_received_at: Instant::now(), + relay_lag_secs: 0, }], cancelled_events: vec![], cancel_reason: None, @@ -2680,6 +2741,8 @@ mod tests { channel_id: ch, event: make_event("old-msg"), received_at: old_time, + relay_received_at: old_time, + relay_lag_secs: 0, prompt_tag: "test".into(), }); @@ -2695,6 +2758,47 @@ mod tests { assert_eq!(batch2.events[0].received_at, original_received_at); } + /// Turn-stage timing fields (`relay_received_at`, `relay_lag_secs`) survive + /// push → flush, `requeue_preserve_timestamps` → flush, and `requeue`'s + /// push-front — the batch always carries the original receipt marks. + #[test] + fn test_relay_timing_fields_preserved_through_flush_and_requeue() { + let mut q = EventQueue::new(DedupMode::Queue); + let ch = Uuid::new_v4(); + let relay_time = Instant::now() - Duration::from_secs(30); + + q.push(QueuedEvent { + channel_id: ch, + event: make_event("timed-msg"), + received_at: relay_time + Duration::from_millis(250), + prompt_tag: "test".into(), + relay_received_at: relay_time, + relay_lag_secs: 7, + }); + + let batch = q.flush_next().expect("flush"); + assert_eq!(batch.events[0].relay_received_at, relay_time); + assert_eq!(batch.events[0].relay_lag_secs, 7); + + // requeue_preserve_timestamps → flush: marks unchanged. + q.requeue_preserve_timestamps(batch); + q.mark_complete(ch); + let batch2 = q.flush_next().expect("flush after requeue_preserve"); + assert_eq!(batch2.events[0].relay_received_at, relay_time); + assert_eq!(batch2.events[0].relay_lag_secs, 7); + + // requeue (backoff path) pushes the event back with marks intact; + // inspect the queue directly since retry_after throttles the flush. + assert!( + q.requeue(batch2).is_none(), + "attempt 1 must be requeued, not dead-lettered" + ); + q.mark_complete(ch); + let requeued = q.queues[&ch].front().expect("requeued event"); + assert_eq!(requeued.relay_received_at, relay_time); + assert_eq!(requeued.relay_lag_secs, 7); + } + #[test] fn test_requeue_preserve_timestamps_no_retry_after() { let mut q = EventQueue::new(DedupMode::Queue); @@ -2969,6 +3073,8 @@ mod tests { event, prompt_tag: "test".into(), received_at: Instant::now(), + relay_received_at: Instant::now(), + relay_lag_secs: 0, }], cancelled_events: vec![], cancel_reason: None, @@ -3000,6 +3106,8 @@ mod tests { event, prompt_tag: "dm".into(), received_at: Instant::now(), + relay_received_at: Instant::now(), + relay_lag_secs: 0, }], cancelled_events: vec![], cancel_reason: None, @@ -3038,6 +3146,8 @@ mod tests { event, prompt_tag: "@mention".into(), received_at: Instant::now(), + relay_received_at: Instant::now(), + relay_lag_secs: 0, }], cancelled_events: vec![], cancel_reason: None, @@ -3066,6 +3176,8 @@ mod tests { event, prompt_tag: "@mention".into(), received_at: Instant::now(), + relay_received_at: Instant::now(), + relay_lag_secs: 0, }], cancelled_events: vec![], cancel_reason: None, @@ -3110,6 +3222,8 @@ mod tests { event, prompt_tag: "dm".into(), received_at: Instant::now(), + relay_received_at: Instant::now(), + relay_lag_secs: 0, }], cancelled_events: vec![], cancel_reason: None, @@ -3159,6 +3273,8 @@ mod tests { event, prompt_tag: "@mention".into(), received_at: Instant::now(), + relay_received_at: Instant::now(), + relay_lag_secs: 0, }], cancelled_events: vec![], cancel_reason: None, @@ -3366,6 +3482,8 @@ mod tests { event, prompt_tag: "dm".into(), received_at: Instant::now(), + relay_received_at: Instant::now(), + relay_lag_secs: 0, }], cancelled_events: vec![], cancel_reason: None, @@ -3423,6 +3541,8 @@ mod tests { event, prompt_tag: "dm".into(), received_at: Instant::now(), + relay_received_at: Instant::now(), + relay_lag_secs: 0, }], cancelled_events: vec![], cancel_reason: None, @@ -3463,6 +3583,8 @@ mod tests { event, prompt_tag: "test".into(), received_at: Instant::now(), + relay_received_at: Instant::now(), + relay_lag_secs: 0, }], cancelled_events: vec![], cancel_reason: None, @@ -3487,6 +3609,8 @@ mod tests { event, prompt_tag: "test".into(), received_at: Instant::now(), + relay_received_at: Instant::now(), + relay_lag_secs: 0, }], cancelled_events: vec![], cancel_reason: None, @@ -3510,6 +3634,8 @@ mod tests { event, prompt_tag: "test".into(), received_at: Instant::now(), + relay_received_at: Instant::now(), + relay_lag_secs: 0, }], cancelled_events: vec![], cancel_reason: None, @@ -3876,6 +4002,8 @@ mod tests { event, prompt_tag: "@mention".into(), received_at: Instant::now(), + relay_received_at: Instant::now(), + relay_lag_secs: 0, }], cancelled_events: vec![], cancel_reason: None, @@ -3918,6 +4046,8 @@ mod tests { event, prompt_tag: "@mention".into(), received_at: Instant::now(), + relay_received_at: Instant::now(), + relay_lag_secs: 0, }], cancelled_events: vec![], cancel_reason: None, @@ -3952,6 +4082,8 @@ mod tests { event, prompt_tag: "test".into(), received_at: Instant::now(), + relay_received_at: Instant::now(), + relay_lag_secs: 0, }], cancelled_events: vec![], cancel_reason: None, @@ -3981,6 +4113,8 @@ mod tests { event, prompt_tag: "test".into(), received_at: Instant::now(), + relay_received_at: Instant::now(), + relay_lag_secs: 0, }], cancelled_events: vec![], cancel_reason: None, @@ -4023,6 +4157,8 @@ mod tests { event, prompt_tag: "@mention".into(), received_at: Instant::now(), + relay_received_at: Instant::now(), + relay_lag_secs: 0, }], cancelled_events: vec![], cancel_reason: None, @@ -4059,6 +4195,8 @@ mod tests { event, prompt_tag: "@mention".into(), received_at: Instant::now(), + relay_received_at: Instant::now(), + relay_lag_secs: 0, }], cancelled_events: vec![], cancel_reason: None, @@ -4095,11 +4233,15 @@ mod tests { event: plain, prompt_tag: "test".into(), received_at: Instant::now(), + relay_received_at: Instant::now(), + relay_lag_secs: 0, }, BatchEvent { event: threaded, prompt_tag: "@mention".into(), received_at: Instant::now(), + relay_received_at: Instant::now(), + relay_lag_secs: 0, }, ], cancelled_events: vec![], @@ -4132,11 +4274,15 @@ mod tests { event: threaded, prompt_tag: "@mention".into(), received_at: Instant::now(), + relay_received_at: Instant::now(), + relay_lag_secs: 0, }, BatchEvent { event: plain, prompt_tag: "test".into(), received_at: Instant::now(), + relay_received_at: Instant::now(), + relay_lag_secs: 0, }, ], cancelled_events: vec![], @@ -4164,6 +4310,8 @@ mod tests { event: make_event(content), prompt_tag: "test".into(), received_at: Instant::now(), + relay_received_at: Instant::now(), + relay_lag_secs: 0, }], cancelled_events: vec![], cancel_reason: None, @@ -4241,6 +4389,8 @@ mod tests { event: make_event("another message"), prompt_tag: "test".into(), received_at: Instant::now(), + relay_received_at: Instant::now(), + relay_lag_secs: 0, }); assert_eq!(slash_command_for_batch(&multi, &[]), None); @@ -4250,6 +4400,8 @@ mod tests { event: make_event("interrupted"), prompt_tag: "test".into(), received_at: Instant::now(), + relay_received_at: Instant::now(), + relay_lag_secs: 0, }); assert_eq!(slash_command_for_batch(&cancelled, &[]), None); @@ -4458,6 +4610,8 @@ mod tests { event: make_event("hi"), prompt_tag: "test".into(), received_at: Instant::now(), + relay_received_at: Instant::now(), + relay_lag_secs: 0, }], cancelled_events: vec![], cancel_reason: None, @@ -4487,6 +4641,8 @@ mod tests { event: make_event("hi"), prompt_tag: "test".into(), received_at: Instant::now(), + relay_received_at: Instant::now(), + relay_lag_secs: 0, }], cancelled_events: vec![], cancel_reason: None, @@ -4515,6 +4671,8 @@ mod tests { event: make_event("hi"), prompt_tag: "test".into(), received_at: Instant::now(), + relay_received_at: Instant::now(), + relay_lag_secs: 0, }], cancelled_events: vec![], cancel_reason: None, diff --git a/crates/buzz-core/src/agent_turn_metric.rs b/crates/buzz-core/src/agent_turn_metric.rs index 037f54f5ab..b25b3c5661 100644 --- a/crates/buzz-core/src/agent_turn_metric.rs +++ b/crates/buzz-core/src/agent_turn_metric.rs @@ -125,6 +125,43 @@ pub struct AgentTurnMetricPayload { /// Why the turn ended. Unrecognized values MUST be treated as `Unknown`. pub stop_reason: Option, + + /// Wall-clock lag (whole seconds, 1s resolution) between the triggering + /// event's `created_at` and harness receipt — minimum over the batch. + /// Omitted when unknown (e.g. heartbeat turns with no triggering event). + #[serde(skip_serializing_if = "Option::is_none")] + pub relay_lag_secs: Option, + + /// Milliseconds from harness receipt to queue admission for the batch's + /// oldest event (author gate, rule matching). Omitted when unknown. + #[serde(skip_serializing_if = "Option::is_none")] + pub admission_ms: Option, + + /// Milliseconds from harness receipt of the batch's oldest event to + /// prompt-task start (queueing + dispatch wait). Omitted when unknown. + #[serde(skip_serializing_if = "Option::is_none")] + pub queue_wait_ms: Option, + + /// Milliseconds from prompt-task start to session resolution (reused or + /// freshly created). Omitted when unknown. + #[serde(skip_serializing_if = "Option::is_none")] + pub session_setup_ms: Option, + + /// Milliseconds from the `session/prompt` write to the first + /// `agent_message_chunk`. Omitted when the agent streamed no message + /// (e.g. failed or cancelled before output). + #[serde(skip_serializing_if = "Option::is_none")] + pub first_output_ms: Option, + + /// Milliseconds from prompt-task start to turn completion (any outcome). + /// Omitted when unknown. + #[serde(skip_serializing_if = "Option::is_none")] + pub turn_total_ms: Option, + + /// `true` when the turn ran on an existing session, `false` when a new + /// session was created for it. Omitted when unknown. + #[serde(skip_serializing_if = "Option::is_none")] + pub session_reused: Option, } fn default_delta_reliable() -> bool { @@ -223,6 +260,13 @@ mod tests { }), delta_reliable: true, stop_reason: Some(StopReason::EndTurn), + relay_lag_secs: Some(2), + admission_ms: Some(35), + queue_wait_ms: Some(120), + session_setup_ms: Some(950), + first_output_ms: Some(1800), + turn_total_ms: Some(15400), + session_reused: Some(false), } } @@ -317,6 +361,51 @@ mod tests { assert_eq!(back, counts); } + #[test] + fn turn_timing_fields_round_trip_and_skip_when_absent() { + // Known timings serialize as camelCase keys; None timings are omitted + // entirely (skip_serializing_if), and payloads written before the + // fields existed still parse with all timings None. + let payload = sample_payload(); + let json = serde_json::to_string(&payload).unwrap(); + assert!(json.contains("\"relayLagSecs\":2")); + assert!(json.contains("\"admissionMs\":35")); + assert!(json.contains("\"queueWaitMs\":120")); + assert!(json.contains("\"sessionSetupMs\":950")); + assert!(json.contains("\"firstOutputMs\":1800")); + assert!(json.contains("\"turnTotalMs\":15400")); + assert!(json.contains("\"sessionReused\":false")); + let back: AgentTurnMetricPayload = serde_json::from_str(&json).unwrap(); + assert_eq!(back, payload); + + let mut sparse = sample_payload(); + sparse.relay_lag_secs = None; + sparse.admission_ms = None; + sparse.queue_wait_ms = None; + sparse.session_setup_ms = None; + sparse.first_output_ms = None; + sparse.turn_total_ms = None; + sparse.session_reused = None; + let json = serde_json::to_string(&sparse).unwrap(); + for key in [ + "relayLagSecs", + "admissionMs", + "queueWaitMs", + "sessionSetupMs", + "firstOutputMs", + "turnTotalMs", + "sessionReused", + ] { + assert!(!json.contains(key), "{key} must be omitted when None"); + } + + // Pre-timing payloads (no timing keys at all) still deserialize. + let legacy = r#"{"harness":"goose","timestamp":"2026-07-01T20:11:03Z"}"#; + let parsed: AgentTurnMetricPayload = serde_json::from_str(legacy).expect("parse"); + assert_eq!(parsed.turn_total_ms, None); + assert_eq!(parsed.session_reused, None); + } + #[test] fn unknown_stop_reason_maps_to_unknown_not_error() { // NIP-AM: consumers MUST treat unrecognized stopReason values as Unknown; @@ -368,6 +457,13 @@ mod tests { cumulative: None, delta_reliable: true, stop_reason: None, + relay_lag_secs: None, + admission_ms: None, + queue_wait_ms: None, + session_setup_ms: None, + first_output_ms: None, + turn_total_ms: None, + session_reused: None, } } @@ -391,6 +487,13 @@ mod tests { }), delta_reliable: true, stop_reason: None, + relay_lag_secs: None, + admission_ms: None, + queue_wait_ms: None, + session_setup_ms: None, + first_output_ms: None, + turn_total_ms: None, + session_reused: None, } } diff --git a/desktop/src-tauri/src/archive/mod_tests.rs b/desktop/src-tauri/src/archive/mod_tests.rs index 288d2ab34c..a6d2fced49 100644 --- a/desktop/src-tauri/src/archive/mod_tests.rs +++ b/desktop/src-tauri/src/archive/mod_tests.rs @@ -647,6 +647,13 @@ fn make_turn_metric_event(owner_keys: &Keys, agent_keys: &Keys) -> Event { cumulative: None, delta_reliable: true, stop_reason: None, + relay_lag_secs: None, + admission_ms: None, + queue_wait_ms: None, + session_setup_ms: None, + first_output_ms: None, + turn_total_ms: None, + session_reused: None, }; let ciphertext = encrypt_agent_turn_metric(agent_keys, &owner_keys.public_key(), &payload).unwrap(); diff --git a/docs/nips/NIP-AM.md b/docs/nips/NIP-AM.md index ff636fb802..acbb067eee 100644 --- a/docs/nips/NIP-AM.md +++ b/docs/nips/NIP-AM.md @@ -108,7 +108,19 @@ The `content` field decrypts to a UTF-8 JSON object: // "turn" object unreliable for this event. "deltaReliable": true, - "stopReason": "end_turn" // optional + "stopReason": "end_turn", // optional + + // Stage latencies for this turn, all optional (present only when the + // publisher measured them). Monotonic-clock millisecond durations except + // relayLagSecs (wall-clock seconds, 1s resolution). Correlation/timing + // metadata only — never message content. + "relayLagSecs": 1 | null, // relay accept -> harness receipt + "admissionMs": 4 | null, // receipt -> queue admission + "queueWaitMs": 12 | null, // admission -> turn dispatch + "sessionSetupMs": 850 | null, // dispatch -> ACP session resolved + "firstOutputMs": 2900 | null, // prompt sent -> first model output + "turnTotalMs": 9400 | null, // dispatch -> turn completion + "sessionReused": true | null // warm (true) vs cold (false) session } ```