From 7393b0bc83b19b427bddffc01f68510e4ec74957 Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" Date: Sun, 20 Sep 2026 15:49:10 +0000 Subject: [PATCH] chore: sync public mirror from internal --- .repository-projection.json | 6 +- packages/local-host-rs/src/agent/mod.rs | 139 ++-- packages/local-host-rs/src/telemetry/mod.rs | 2 + packages/local-host-rs/src/telemetry/relay.rs | 725 ++++++++++++++++++ .../local-host-rs/src/telemetry/tracker.rs | 103 +++ .../src/telemetry/wide_events.rs | 10 +- 6 files changed, 908 insertions(+), 77 deletions(-) create mode 100644 packages/local-host-rs/src/telemetry/relay.rs diff --git a/.repository-projection.json b/.repository-projection.json index 730402586..ea90024e8 100644 --- a/.repository-projection.json +++ b/.repository-projection.json @@ -3,11 +3,11 @@ "projection": "deixic-code", "projectionSchemaVersion": 1, "sourceRepository": "dx-corp/mono", - "sourceSha": "3b8d839ff70abae7d6aa87082079449a8d6eb43a", + "sourceSha": "6f8bca8d69be1b576e96428243157a5cb5c3cf35", "destinationRepository": "dx-corp/code", - "priorProjectedBase": "4690e79129c9b1874f9bce83c9b0e294894fd81d", + "priorProjectedBase": "873426648fb369290ceb39a5904738754ef0f46c", "definitionDigest": "064b6b9745d392dde277afd7dfdf028cff32b6178a8b9c99c518343219bab429", "toolDigest": "c8000985e1bd708bf1f8754b27cc1a2aad67af10ee23cf5317bd1fba670b4456", - "contentDigest": "f3cedd14d1124069f81086dd5bf7c8c7f53c4fbaebd3f6c20f51000dc13ffc44", + "contentDigest": "0a31fb4dca7df2bcda6576943dc71fb0a3ecc735c3e86b8caacc8536e0e6d190", "publicationEligible": true } diff --git a/packages/local-host-rs/src/agent/mod.rs b/packages/local-host-rs/src/agent/mod.rs index 6f619c6d8..824dfe7a0 100644 --- a/packages/local-host-rs/src/agent/mod.rs +++ b/packages/local-host-rs/src/agent/mod.rs @@ -667,97 +667,70 @@ pub(crate) use maestro_runtime::agent::managed_turn_lineage_id; /// Observe the same event stream delivered to the caller, preserving the /// canonical turn accounting that was attached to the actor before extraction. fn relay_runtime_events( - mut runtime_event_rx: tokio::sync::mpsc::UnboundedReceiver, + runtime_event_rx: tokio::sync::mpsc::UnboundedReceiver, config: &NativeAgentConfig, provider_name: String, telemetry_identity_scope: Arc>>, telemetry_host: Option, ) -> tokio::sync::mpsc::UnboundedReceiver { let (consumer_event_tx, event_rx) = tokio::sync::mpsc::unbounded_channel(); - let mut turn_tracker = - crate::telemetry::TurnTracker::new(crate::telemetry::TurnTrackerConfig { + let initial_identity_scope = telemetry_identity_scope + .read() + .expect("telemetry identity scope lock poisoned") + .clone(); + let telemetry = crate::telemetry::TurnTelemetryRelay::spawn( + crate::telemetry::TurnTrackerConfig { // A runtime run ID is not a captured conversation ID. Missing // session context stays unattributed in the cloud projection. session_id: String::new(), sampling_config: crate::telemetry::TailSamplingConfig::from_env(), - }); - turn_tracker.update_context(crate::telemetry::TurnTrackerContext { - model: Some(crate::telemetry::ModelInfo { - id: config.model.clone(), - provider: provider_name, - thinking_level: crate::telemetry::ThinkingLevel::Off, - }), - sandbox_mode: if config.sandbox_policy.is_some() { - crate::telemetry::SandboxMode::Local - } else { - crate::telemetry::SandboxMode::None }, - approval_mode: match config.approval_mode { - ApprovalMode::Yolo => crate::telemetry::ApprovalMode::Auto, - ApprovalMode::Selective | ApprovalMode::Safe => crate::telemetry::ApprovalMode::Prompt, - }, - mcp_servers: Vec::new(), - context_source_count: 0, - features: crate::telemetry::FeatureFlags { - safe_mode: config.approval_mode != ApprovalMode::Yolo, - ..Default::default() + crate::telemetry::TurnTrackerContext { + model: Some(crate::telemetry::ModelInfo { + id: config.model.clone(), + provider: provider_name, + thinking_level: crate::telemetry::ThinkingLevel::Off, + }), + sandbox_mode: if config.sandbox_policy.is_some() { + crate::telemetry::SandboxMode::Local + } else { + crate::telemetry::SandboxMode::None + }, + approval_mode: match config.approval_mode { + ApprovalMode::Yolo => crate::telemetry::ApprovalMode::Auto, + ApprovalMode::Selective | ApprovalMode::Safe => { + crate::telemetry::ApprovalMode::Prompt + } + }, + mcp_servers: Vec::new(), + context_source_count: 0, + features: crate::telemetry::FeatureFlags { + safe_mode: config.approval_mode != ApprovalMode::Yolo, + ..Default::default() + }, + identity_scope: initial_identity_scope, }, - identity_scope: telemetry_identity_scope - .read() - .expect("telemetry identity scope lock poisoned") - .clone(), - }); + telemetry_identity_scope, + telemetry_host, + ); + relay_runtime_events_with_telemetry(runtime_event_rx, telemetry, consumer_event_tx); + event_rx +} + +fn relay_runtime_events_with_telemetry( + mut runtime_event_rx: tokio::sync::mpsc::UnboundedReceiver, + mut telemetry: crate::telemetry::TurnTelemetryRelay, + consumer_event_tx: tokio::sync::mpsc::UnboundedSender, +) { tokio::spawn(async move { - let mut journal = crate::telemetry::TurnJournal::open(); while let Some(event) = runtime_event_rx.recv().await { - if matches!( - &event, - FromAgent::TurnStarted | FromAgent::ResponseStart { .. } - | FromAgent::SideQuestionStart { .. } - | FromAgent::OperationObservation { observation: maestro_runtime_contracts::operation_observation::OperationObservation::Admitted { .. } } - ) { - let session_id = match &telemetry_host { - Some(host) => host.hook_session_id().await, - None => None, - }; - // Clear a previous session when the owner has none; the - // collector pins the current turn's identity at its start. - turn_tracker.set_session_id(session_id.unwrap_or_default()); - } - if matches!(&event, FromAgent::ModelChanged { .. }) { - turn_tracker.set_identity_scope( - telemetry_identity_scope - .read() - .expect("telemetry identity scope lock poisoned") - .clone(), - ); - } - let completed = turn_tracker.handle_event(&event); - if let Some(completed) = completed.as_ref() { - if let Some(journal) = &mut journal { - journal.finish(completed); - } else { - crate::telemetry::record_canonical_turn_event(completed); - } - } - if matches!( - &event, - FromAgent::TurnStarted - | FromAgent::OperationObservation { .. } - | FromAgent::SideQuestionStart { .. } - | FromAgent::ResponseStart { .. } - | FromAgent::ResponseEnd { .. } - ) { - if let Some(journal) = &mut journal { - journal.observe(&turn_tracker.pending_snapshots()); - } - } + let observation = telemetry.project(&event); // A detached consumer does not end the actor's turn. Keep draining // until the runtime exits so terminal telemetry is still recorded. let _ = consumer_event_tx.send(event); + telemetry.submit(observation); } }); - event_rx } /// Bind a model transition before publishing its identity to turn telemetry. @@ -817,6 +790,28 @@ mod identity_transition_tests { .unwrap(); } + #[tokio::test] + async fn saturated_telemetry_never_delays_runtime_event_delivery() { + let (runtime_tx, runtime_rx) = tokio::sync::mpsc::unbounded_channel(); + let (consumer_tx, mut consumer_rx) = tokio::sync::mpsc::unbounded_channel(); + let (telemetry, _blocked_receiver) = + crate::telemetry::TurnTelemetryRelay::saturated_for_test(); + super::relay_runtime_events_with_telemetry(runtime_rx, telemetry, consumer_tx); + + runtime_tx + .send(super::FromAgent::ResponseStart { + response_id: "response".to_owned(), + }) + .unwrap(); + + let delivered = + tokio::time::timeout(std::time::Duration::from_millis(100), consumer_rx.recv()) + .await + .expect("a full telemetry queue must not delay the consumer") + .expect("runtime event relay closed unexpectedly"); + assert!(matches!(delivered, super::FromAgent::ResponseStart { .. })); + } + fn scope(org: &str, workspace: &str) -> Option { TelemetryIdentityScope::new(org, Some(workspace)) } diff --git a/packages/local-host-rs/src/telemetry/mod.rs b/packages/local-host-rs/src/telemetry/mod.rs index 99e66d7ce..a0177f02b 100644 --- a/packages/local-host-rs/src/telemetry/mod.rs +++ b/packages/local-host-rs/src/telemetry/mod.rs @@ -41,8 +41,10 @@ mod customer_export; mod onboarding; mod operation; +mod relay; mod staged_rollout; mod tracker; +pub(crate) use relay::TurnTelemetryRelay; pub(crate) use staged_rollout::TurnJournal; mod visibility; mod wide_events; diff --git a/packages/local-host-rs/src/telemetry/relay.rs b/packages/local-host-rs/src/telemetry/relay.rs new file mode 100644 index 000000000..4e9b7b18a --- /dev/null +++ b/packages/local-host-rs/src/telemetry/relay.rs @@ -0,0 +1,725 @@ +//! Bounded, content-free handoff from runtime event delivery to turn telemetry. +//! +//! The producer never waits and never performs telemetry I/O. Rich runtime +//! payloads remain on their existing consumer path; this queue receives only +//! the fields the turn collector needs. Streaming content is reduced to byte +//! counts before it can enter the queue. + +use super::{ + TelemetryIdentityScope, TurnJournal, TurnTracker, TurnTrackerConfig, TurnTrackerContext, + record_canonical_turn_event, +}; +use crate::agent::{FromAgent, TokenUsage}; +use maestro_runtime::ExecutionStatus; +use std::sync::{ + Arc, RwLock, + atomic::{AtomicU64, Ordering}, + mpsc::{SyncSender, TrySendError, sync_channel}, +}; + +const QUEUE_CAPACITY: usize = 64; +const WORKER_STACK_BYTES: usize = 256 * 1024; +const MAX_LABEL_BYTES: usize = 256; + +#[derive(Debug, Clone)] +pub(crate) struct ToolReceiptObservation { + call_id: String, + tool_name: String, + status: ExecutionStatus, + duration_ms: Option, +} + +#[derive(Debug)] +#[allow( + clippy::large_enum_variant, + reason = "boxing would add producer-path allocation; the fixed queue is capped below 64 KiB" +)] +pub(crate) enum TurnObservation { + Event(FromAgent), + Output { + bytes: u64, + first_observed_at: std::time::Instant, + }, + SideQuestionEnd { + side_id: String, + answer_bytes: u64, + usage: Option, + failed: bool, + }, + ToolCall { + call_id: String, + tool: String, + requires_approval: bool, + }, + ToolEnd { + call_id: String, + success: bool, + receipt: Option, + }, + Session(Option), +} + +#[derive(Debug)] +pub(crate) struct QueuedObservation { + sequence: u64, + observation: TurnObservation, +} + +#[derive(Default)] +struct SequenceCursor { + next: u64, +} + +impl SequenceCursor { + fn observe(&mut self, sequence: u64) -> u64 { + let missed = sequence.saturating_sub(self.next); + self.next = sequence.saturating_add(1); + missed + } + + fn trailing_losses(&self, attempted: u64) -> u64 { + attempted.saturating_sub(self.next) + } +} + +impl TurnObservation { + fn needs_pending_snapshot(&self) -> bool { + matches!( + self, + Self::Event( + FromAgent::TurnStarted + | FromAgent::OperationObservation { .. } + | FromAgent::SideQuestionStart { .. } + | FromAgent::ResponseStart { .. } + | FromAgent::ResponseEnd { .. } + ) + ) + } + + fn refreshes_identity(&self) -> bool { + matches!(self, Self::Event(FromAgent::ModelChanged { .. })) + } + + fn refreshes_session(&self) -> bool { + matches!( + self, + Self::Event( + FromAgent::TurnStarted + | FromAgent::ResponseStart { .. } + | FromAgent::SideQuestionStart { .. } + | FromAgent::OperationObservation { + observation: maestro_runtime_contracts::operation_observation::OperationObservation::Admitted { .. } + } + ) + ) + } +} + +/// A tiny producer facade. Its only hot-path state is a byte accumulator and +/// the first-output instant, so token chunks never allocate or occupy queue +/// slots. All aggregation, serialization, locking, disk, and network work is +/// owned by the dedicated worker thread. +pub(crate) struct TurnTelemetryRelay { + sender: Option>, + dropped: Arc, + attempted: Arc, + next_sequence: u64, + pending_output_bytes: u64, + pending_first_output_at: Option, +} + +impl TurnTelemetryRelay { + pub(crate) fn spawn( + config: TurnTrackerConfig, + context: TurnTrackerContext, + identity_scope: Arc>>, + telemetry_host: Option, + ) -> Self { + if super::experiments_telemetry_disabled() { + return Self { + sender: None, + dropped: Arc::new(AtomicU64::new(0)), + attempted: Arc::new(AtomicU64::new(0)), + next_sequence: 0, + pending_output_bytes: 0, + pending_first_output_at: None, + }; + } + + let (sender, receiver) = sync_channel::(QUEUE_CAPACITY); + let dropped = Arc::new(AtomicU64::new(0)); + let attempted = Arc::new(AtomicU64::new(0)); + let worker_attempted = Arc::clone(&attempted); + let worker = std::thread::Builder::new() + .name("maestro-telemetry".to_owned()) + .stack_size(WORKER_STACK_BYTES) + .spawn(move || { + let mut tracker = TurnTracker::new(config); + tracker.update_context(context); + let mut journal = TurnJournal::open(); + let mut sequence_cursor = SequenceCursor::default(); + while let Ok(queued) = receiver.recv() { + recover_after_loss( + &mut tracker, + &mut journal, + sequence_cursor.observe(queued.sequence), + ); + let observation = queued.observation; + let snapshot = observation.needs_pending_snapshot(); + if observation.refreshes_identity() { + tracker.set_identity_scope( + identity_scope + .read() + .expect("telemetry identity scope lock poisoned") + .clone(), + ); + } + if observation.refreshes_session() { + let session_id = telemetry_host + .as_ref() + .and_then(|host| futures::executor::block_on(host.hook_session_id())); + tracker.set_session_id(session_id.unwrap_or_default()); + } + let completed = handle_observation(&mut tracker, observation); + if let Some(completed) = completed.as_ref() { + finish(&mut journal, completed); + } + if snapshot { + observe_pending(&mut journal, &tracker); + } + } + recover_after_loss( + &mut tracker, + &mut journal, + sequence_cursor.trailing_losses(worker_attempted.load(Ordering::Acquire)), + ); + }); + + Self { + sender: worker.ok().map(|_| sender), + dropped, + attempted, + next_sequence: 0, + pending_output_bytes: 0, + pending_first_output_at: None, + } + } + + /// Project before the rich event is moved to its consumer, then enqueue + /// only after that consumer has been notified. Saturation is intentionally + /// lossy and never delays event delivery. + pub(crate) fn project(&mut self, event: &FromAgent) -> Option { + match event { + FromAgent::ResponseChunk { + content, + is_thinking, + .. + } => { + if !is_thinking && !content.is_empty() { + self.pending_first_output_at + .get_or_insert_with(std::time::Instant::now); + self.pending_output_bytes = self + .pending_output_bytes + .saturating_add(content.len() as u64); + } + None + } + FromAgent::ToolOutput { .. } + | FromAgent::LocalAssistantContent { .. } + | FromAgent::ConversationSnapshot { .. } + | FromAgent::ManagedAuthorizationRequest { .. } + | FromAgent::ManagedGatewayReceipt { .. } + | FromAgent::ModelChangeFailed { .. } + | FromAgent::SideQuestionChunk { .. } + | FromAgent::CodexSessionState { .. } + | FromAgent::CodexTurnState { .. } + | FromAgent::CodexCompatibility { .. } + | FromAgent::CodexNativeOperation { .. } + | FromAgent::CodexNativeDecision { .. } + | FromAgent::CodexTransportReceipt { .. } + | FromAgent::BatchStart { .. } + | FromAgent::BatchEnd { .. } + | FromAgent::Status { .. } + | FromAgent::HookBlocked { .. } => None, + FromAgent::SessionInfo { session_id, .. } => Some(TurnObservation::Session( + session_id.as_deref().map(bounded_label), + )), + FromAgent::ToolCall { + call_id, + tool, + requires_approval, + .. + } => Some(TurnObservation::ToolCall { + call_id: bounded_label(call_id), + tool: bounded_label(tool), + requires_approval: *requires_approval, + }), + FromAgent::ToolEnd { + call_id, + success, + receipt, + .. + } => Some(TurnObservation::ToolEnd { + call_id: bounded_label(call_id), + success: *success, + receipt: receipt.as_ref().map(|receipt| ToolReceiptObservation { + call_id: bounded_label(&receipt.call_id), + tool_name: bounded_label(&receipt.tool_name), + status: receipt.status, + duration_ms: receipt.duration_ms, + }), + }), + FromAgent::SideQuestionEnd { + side_id, + answer, + usage, + error, + .. + } => Some(TurnObservation::SideQuestionEnd { + side_id: bounded_label(side_id), + answer_bytes: answer.len() as u64, + usage: usage.clone(), + failed: error.is_some(), + }), + _ => sanitized_event(event).map(TurnObservation::Event), + } + } + + pub(crate) fn submit(&mut self, observation: Option) { + if self.sender.is_none() { + return; + } + let Some(observation) = observation else { + return; + }; + + if self.pending_output_bytes > 0 { + let bytes = std::mem::take(&mut self.pending_output_bytes); + let first_observed_at = self + .pending_first_output_at + .take() + .expect("nonzero output bytes require an observation time"); + self.try_submit(TurnObservation::Output { + bytes, + first_observed_at, + }); + } + self.try_submit(observation); + } + + fn try_submit(&mut self, observation: TurnObservation) { + let Some(sender) = &self.sender else { + return; + }; + let sequence = self.next_sequence; + self.next_sequence = self.next_sequence.saturating_add(1); + self.attempted.store(self.next_sequence, Ordering::Release); + if let Err(error) = sender.try_send(QueuedObservation { + sequence, + observation, + }) { + self.record_drop(error); + } + } + + fn record_drop(&self, error: TrySendError) { + if matches!(error, TrySendError::Full(_)) { + self.dropped.fetch_add(1, Ordering::Relaxed); + } + } + + #[cfg(test)] + pub(crate) fn saturated_for_test() -> (Self, std::sync::mpsc::Receiver) { + let (sender, receiver) = sync_channel(1); + sender + .try_send(QueuedObservation { + sequence: 0, + observation: TurnObservation::Output { + bytes: 1, + first_observed_at: std::time::Instant::now(), + }, + }) + .expect("test telemetry queue should start empty"); + ( + Self { + sender: Some(sender), + dropped: Arc::new(AtomicU64::new(0)), + attempted: Arc::new(AtomicU64::new(1)), + next_sequence: 1, + pending_output_bytes: 0, + pending_first_output_at: None, + }, + receiver, + ) + } +} + +fn recover_after_loss(tracker: &mut TurnTracker, journal: &mut Option, count: u64) { + if count == 0 { + return; + } + tracing::warn!( + dropped = count, + "telemetry queue saturated; recording turn as incomplete" + ); + for event in tracker.abandon_incomplete() { + finish(journal, &event); + } +} + +fn finish(journal: &mut Option, event: &super::CanonicalTurnEvent) { + if let Some(journal) = journal { + journal.finish(event); + } else { + record_canonical_turn_event(event); + } +} + +fn observe_pending(journal: &mut Option, tracker: &TurnTracker) { + if let Some(journal) = journal { + journal.observe(&tracker.pending_snapshots()); + } +} + +fn handle_observation( + tracker: &mut TurnTracker, + observation: TurnObservation, +) -> Option { + match observation { + TurnObservation::Event(event) => tracker.handle_event(&event), + TurnObservation::Output { + bytes, + first_observed_at, + } => { + tracker.record_output(bytes, first_observed_at); + None + } + TurnObservation::SideQuestionEnd { + side_id, + answer_bytes, + usage, + failed, + } => tracker.finish_side_question(&side_id, answer_bytes, usage.as_ref(), failed), + TurnObservation::ToolCall { + call_id, + tool, + requires_approval, + } => { + tracker.record_tool_call(&call_id, &tool, None, requires_approval); + None + } + TurnObservation::ToolEnd { + call_id, + success, + receipt, + } => { + tracker.record_tool_end( + &call_id, + success, + receipt.as_ref().map(|receipt| { + ( + receipt.call_id.as_str(), + receipt.tool_name.as_str(), + receipt.status, + receipt.duration_ms, + ) + }), + ); + None + } + TurnObservation::Session(session_id) => { + tracker.set_session_id(session_id.unwrap_or_default()); + None + } + } +} + +fn sanitized_event(event: &FromAgent) -> Option { + Some(match event { + FromAgent::Ready { model, provider } => FromAgent::Ready { + model: bounded_label(model), + provider: bounded_label(provider), + }, + FromAgent::ModelChanged { model, provider } => FromAgent::ModelChanged { + model: bounded_label(model), + provider: bounded_label(provider), + }, + FromAgent::BoostChanged { status, thinking } => FromAgent::BoostChanged { + status: *status, + thinking: *thinking, + }, + FromAgent::ResponseStart { response_id } => FromAgent::ResponseStart { + response_id: bounded_label(response_id), + }, + FromAgent::ResponseEnd { response_id, usage } => FromAgent::ResponseEnd { + response_id: bounded_label(response_id), + usage: usage.clone(), + }, + FromAgent::TurnCompleted { .. } => FromAgent::TurnCompleted { + response_id: String::new(), + coding_completion: None, + coding_child_records: Vec::new(), + }, + FromAgent::TurnInterrupted { .. } => FromAgent::TurnInterrupted { + response_id: String::new(), + reason: String::new(), + }, + FromAgent::SideQuestionStart { + side_id, + standalone, + .. + } => FromAgent::SideQuestionStart { + side_id: bounded_label(side_id), + question: String::new(), + standalone: *standalone, + }, + FromAgent::CodexUsageState { source, usage } => FromAgent::CodexUsageState { + source: bounded_label(source), + usage: usage.clone(), + }, + FromAgent::StreamObservation { observation } => FromAgent::StreamObservation { + observation: *observation, + }, + FromAgent::ContextCalibration { observation } => FromAgent::ContextCalibration { + observation: maestro_context::context_usage::ContextCalibration { + request_id: bounded_label(&observation.request_id), + generation: observation.generation, + estimated_input_tokens: observation.estimated_input_tokens, + observed_input_tokens: observation.observed_input_tokens, + }, + }, + FromAgent::RequestRetryObservation => FromAgent::RequestRetryObservation, + FromAgent::RequestContextPrepared { response_id } => FromAgent::RequestContextPrepared { + response_id: bounded_label(response_id), + }, + FromAgent::OperationObservation { observation } => FromAgent::OperationObservation { + observation: sanitized_operation_observation(observation), + }, + FromAgent::TurnStarted => FromAgent::TurnStarted, + FromAgent::RequestRetryScheduled { + attempt, + delay_ms, + rate_limited, + } => FromAgent::RequestRetryScheduled { + attempt: *attempt, + delay_ms: *delay_ms, + rate_limited: *rate_limited, + }, + FromAgent::CompactionMeasured { duration_ms } => FromAgent::CompactionMeasured { + duration_ms: *duration_ms, + }, + FromAgent::ToolStart { call_id } => FromAgent::ToolStart { + call_id: bounded_label(call_id), + }, + FromAgent::Error { + fatal, + terminal, + retryable, + .. + } => FromAgent::Error { + message: String::new(), + fatal: *fatal, + terminal: *terminal, + retryable: *retryable, + }, + FromAgent::ProviderError { kind, .. } => FromAgent::ProviderError { + kind: *kind, + message: String::new(), + }, + FromAgent::Compaction { + first_kept_entry_index, + tokens_before, + auto, + .. + } => FromAgent::Compaction { + summary: String::new(), + first_kept_entry_index: *first_kept_entry_index, + tokens_before: *tokens_before, + auto: *auto, + custom_instructions: None, + continuation: None, + timestamp: String::new(), + }, + _ => return None, + }) +} + +fn sanitized_operation_observation( + observation: &maestro_runtime_contracts::operation_observation::OperationObservation, +) -> maestro_runtime_contracts::operation_observation::OperationObservation { + use maestro_runtime_contracts::operation_observation::OperationObservation as O; + match observation { + O::Admitted { + turn_id, + thinking_level, + experiment, + } => O::Admitted { + turn_id: bounded_label(turn_id), + thinking_level: bounded_label(thinking_level), + experiment: experiment.clone().filter(|value| value.is_valid()), + }, + O::Prepared { + response_id, + model_id, + model_provider, + message_count, + input_size_bytes, + } => O::Prepared { + response_id: bounded_label(response_id), + model_id: bounded_label(model_id), + model_provider: bounded_label(model_provider), + message_count: *message_count, + input_size_bytes: *input_size_bytes, + }, + O::ReasoningUsage { + response_id, + tokens, + } => O::ReasoningUsage { + response_id: bounded_label(response_id), + tokens: *tokens, + }, + O::GatewayReceipt { + response_id, + request_id, + record_id, + lineage_id, + provider_tools_sha256, + provider_tool_count, + } => O::GatewayReceipt { + response_id: bounded_label(response_id), + request_id: bounded_label(request_id), + record_id: bounded_label(record_id), + lineage_id: bounded_label(lineage_id), + provider_tools_sha256: provider_tools_sha256.as_deref().map(bounded_label), + provider_tool_count: *provider_tool_count, + }, + } +} + +fn bounded_label(value: &str) -> String { + if value.len() <= MAX_LABEL_BYTES { + return value.to_owned(); + } + let mut end = MAX_LABEL_BYTES; + while !value.is_char_boundary(end) { + end -= 1; + } + value[..end].to_owned() +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn response_content_becomes_one_counter() { + let (sender, receiver) = sync_channel(2); + let relay = TurnTelemetryRelay { + sender: Some(sender), + dropped: Arc::new(AtomicU64::new(0)), + attempted: Arc::new(AtomicU64::new(0)), + next_sequence: 0, + pending_output_bytes: 0, + pending_first_output_at: None, + }; + let mut relay = relay; + assert!( + relay + .project(&FromAgent::ResponseChunk { + response_id: "response".into(), + content: "secret output".into(), + is_thinking: false, + }) + .is_none() + ); + assert_eq!(relay.pending_output_bytes, 13); + relay.submit(None); + assert!(receiver.try_recv().is_err()); + } + + #[test] + fn labels_are_utf8_safe_and_bounded() { + let value = "🦀".repeat(MAX_LABEL_BYTES); + let bounded = bounded_label(&value); + assert!(bounded.len() <= MAX_LABEL_BYTES); + assert!(bounded.is_char_boundary(bounded.len())); + } + + #[test] + fn full_queue_is_nonblocking_and_counted() { + let (sender, _receiver) = sync_channel(1); + sender + .try_send(QueuedObservation { + sequence: 0, + observation: TurnObservation::Output { + bytes: 1, + first_observed_at: std::time::Instant::now(), + }, + }) + .unwrap(); + let relay = TurnTelemetryRelay { + sender: Some(sender), + dropped: Arc::new(AtomicU64::new(0)), + attempted: Arc::new(AtomicU64::new(1)), + next_sequence: 1, + pending_output_bytes: 0, + pending_first_output_at: None, + }; + relay.record_drop(TrySendError::Full(QueuedObservation { + sequence: 1, + observation: TurnObservation::Output { + bytes: 2, + first_observed_at: std::time::Instant::now(), + }, + })); + assert_eq!(relay.dropped.load(Ordering::Relaxed), 1); + } + + #[test] + fn queue_storage_stays_small() { + assert!( + std::mem::size_of::() * QUEUE_CAPACITY <= 64 * 1024, + "the preallocated observation queue must stay within 64 KiB" + ); + } + + #[test] + #[ignore = "local performance probe"] + fn response_chunk_projection_p99_is_sub_microsecond() { + const BATCH: u32 = 1_024; + const SAMPLES: usize = 256; + let mut relay = TurnTelemetryRelay { + sender: None, + dropped: Arc::new(AtomicU64::new(0)), + attempted: Arc::new(AtomicU64::new(0)), + next_sequence: 0, + pending_output_bytes: 0, + pending_first_output_at: None, + }; + let event = FromAgent::ResponseChunk { + response_id: "response".into(), + content: "one streamed token".into(), + is_thinking: false, + }; + let mut nanos_per_event = Vec::with_capacity(SAMPLES); + for _ in 0..SAMPLES { + let started = std::time::Instant::now(); + for _ in 0..BATCH { + std::hint::black_box(relay.project(std::hint::black_box(&event))); + } + nanos_per_event.push(started.elapsed().as_nanos() / u128::from(BATCH)); + } + nanos_per_event.sort_unstable(); + let p99 = nanos_per_event[SAMPLES * 99 / 100]; + eprintln!("response chunk projection p99: {p99} ns"); + assert!(p99 < 1_000, "response chunk projection p99 was {p99} ns"); + } + + #[test] + fn sequence_gaps_are_attributed_before_the_following_observation() { + let mut cursor = SequenceCursor::default(); + assert_eq!(cursor.observe(0), 0); + assert_eq!(cursor.observe(2), 1); + assert_eq!(cursor.trailing_losses(4), 1); + } +} diff --git a/packages/local-host-rs/src/telemetry/tracker.rs b/packages/local-host-rs/src/telemetry/tracker.rs index 80d5054d7..30d4b13e3 100644 --- a/packages/local-host-rs/src/telemetry/tracker.rs +++ b/packages/local-host-rs/src/telemetry/tracker.rs @@ -534,6 +534,109 @@ impl TurnTracker { } } + pub(crate) fn record_output(&mut self, bytes: u64, first_observed_at: std::time::Instant) { + if bytes == 0 { + return; + } + if let Some(turn) = &mut self.current_turn { + turn.record_output_at(first_observed_at); + turn.add_output_size(bytes); + self.diagnostics.output_size_bytes = Some( + self.diagnostics + .output_size_bytes + .unwrap_or(0) + .saturating_add(bytes), + ); + } + } + + pub(crate) fn finish_side_question( + &mut self, + side_id: &str, + answer_bytes: u64, + usage: Option<&TokenUsage>, + failed: bool, + ) -> Option { + let mut side = self.side_questions.remove(side_id)?; + side.record_output(answer_bytes, std::time::Instant::now()); + side.handle_event(&FromAgent::ResponseEnd { + response_id: side_id.to_owned(), + usage: usage.cloned(), + }); + side.end_turn( + if failed { + TurnStatus::Error + } else { + TurnStatus::Success + }, + failed.then(|| ErrorDetails { + category: Some("runtime".into()), + message: None, + }), + ) + } + + pub(crate) fn record_tool_call( + &mut self, + call_id: &str, + tool: &str, + input_size_bytes: Option, + requires_approval: bool, + ) { + if requires_approval { + self.pending_approvals + .entry(call_id.to_owned()) + .or_insert_with(std::time::Instant::now); + } + if let Some(turn) = &mut self.current_turn { + turn.record_tool_start(tool, call_id, input_size_bytes); + } + } + + pub(crate) fn record_tool_end( + &mut self, + call_id: &str, + success: bool, + receipt: Option<(&str, &str, maestro_runtime::ExecutionStatus, Option)>, + ) { + if let Some(turn) = &mut self.current_turn { + let receipt = receipt.map(|(receipt_call_id, tool_name, status, duration_ms)| { + maestro_runtime::ExecutionReceipt { + code_authority: None, + call_id: receipt_call_id.to_owned(), + tool_name: tool_name.to_owned(), + source: maestro_runtime::ExecutionSource::Native, + status, + duration_ms, + policy: None, + details: maestro_runtime::ToolReceiptDetails::None, + } + }); + turn.record_tool_receipt(call_id, receipt.as_ref()); + turn.record_tool_end(call_id, success, None, None); + } + } + + /// Close any state whose causal stream became incomplete. This preserves + /// the last observed facts while preventing a later turn from inheriting + /// measurements across a queue gap. + pub(crate) fn abandon_incomplete(&mut self) -> Vec { + let snapshots = self.pending_snapshots(); + self.current_turn = None; + self.current_identity_scope = None; + self.current_response_id = None; + self.accumulated_usage = None; + self.cost_complete = true; + self.diagnostics = Default::default(); + self.response_started = None; + self.response_reasoning = None; + self.total_reasoning = None; + self.current_attempt = None; + self.pending_approvals.clear(); + self.side_questions.clear(); + snapshots + } + fn start_turn(&mut self) { self.turn_number += 1; self.diagnostics = super::operation::OperationDiagnostics { diff --git a/packages/local-host-rs/src/telemetry/wide_events.rs b/packages/local-host-rs/src/telemetry/wide_events.rs index 64cd63574..9c4c4d57b 100644 --- a/packages/local-host-rs/src/telemetry/wide_events.rs +++ b/packages/local-host-rs/src/telemetry/wide_events.rs @@ -635,9 +635,15 @@ impl TurnCollector { /// Observe output without retaining its content. pub fn record_output(&mut self) { + self.record_output_at(Instant::now()); + } + + /// Preserve the producer-observed first-output instant when collection is + /// performed by a delayed background worker. + pub(crate) fn record_output_at(&mut self, observed_at: Instant) { self.measurements.first_output_ms.get_or_insert_with(|| { - self.start_time - .elapsed() + observed_at + .saturating_duration_since(self.start_time) .as_millis() .min(u128::from(u64::MAX)) as u64 });