Skip to content

feat(acp): per-turn stage latency instrumentation (#2386) #2460

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
brocoppler wants to merge 1 commit into
base: main
Choose a base branch
from
Open

feat(acp): per-turn stage latency instrumentation (#2386) #2460

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
98 changes: 98 additions & 0 deletions crates/buzz-acp/src/acp.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<std::time::Instant>,
/// 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<std::time::Instant>,
}

/// Recursively merge `overlay` into `base`, with `overlay` winning on scalar/shape
Expand Down Expand Up @@ -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,
})
}

Expand Down Expand Up @@ -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;
Expand All @@ -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;
Expand Down Expand Up @@ -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<u64> {
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.
///
Expand Down Expand Up @@ -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}");
}
Expand Down Expand Up @@ -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;
Expand Down
55 changes: 55 additions & 0 deletions crates/buzz-acp/src/lib.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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).
Expand Down Expand Up @@ -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}");
Expand Down Expand Up @@ -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<i64>| {
if let Some(ref observer) = observer {
let mut payload = serde_json::json!({
Expand Down Expand Up @@ -4897,6 +4932,7 @@ mod error_outcome_emission_tests {
turn_id: "test-turn-id".to_string(),
outcome,
batch: None,
timings: None,
};

handle_prompt_result(
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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,
Expand All @@ -5349,6 +5394,7 @@ mod error_outcome_emission_tests {
recently_active: true,
}),
batch: Some(batch),
timings: None,
};
handle_prompt_result(
&mut pool,
Expand Down Expand Up @@ -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,
Expand All @@ -5442,6 +5490,7 @@ mod error_outcome_emission_tests {
recently_active: true,
}),
batch: Some(batch),
timings: None,
};
handle_prompt_result(
&mut pool,
Expand Down Expand Up @@ -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),
Expand Down Expand Up @@ -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();
Expand All @@ -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(
Expand Down Expand Up @@ -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(
Expand Down
Loading