diff --git a/.repository-projection.json b/.repository-projection.json index ea90024e8..fd819f87a 100644 --- a/.repository-projection.json +++ b/.repository-projection.json @@ -3,11 +3,11 @@ "projection": "deixic-code", "projectionSchemaVersion": 1, "sourceRepository": "dx-corp/mono", - "sourceSha": "6f8bca8d69be1b576e96428243157a5cb5c3cf35", + "sourceSha": "6139cda76fc6bd4d07a40c8351468364c090e619", "destinationRepository": "dx-corp/code", - "priorProjectedBase": "873426648fb369290ceb39a5904738754ef0f46c", + "priorProjectedBase": "7f514fca918d431635a0ba92a825eb4bd0812cac", "definitionDigest": "064b6b9745d392dde277afd7dfdf028cff32b6178a8b9c99c518343219bab429", "toolDigest": "c8000985e1bd708bf1f8754b27cc1a2aad67af10ee23cf5317bd1fba670b4456", - "contentDigest": "0a31fb4dca7df2bcda6576943dc71fb0a3ecc735c3e86b8caacc8536e0e6d190", + "contentDigest": "0a9c99f245ca6f3b5929664ecdaa6dc22347566ee2a0af6f97bd8ac4c482bc64", "publicationEligible": true } diff --git a/packages/local-host-rs/src/telemetry/relay.rs b/packages/local-host-rs/src/telemetry/relay.rs index 4e9b7b18a..bcf23e1d7 100644 --- a/packages/local-host-rs/src/telemetry/relay.rs +++ b/packages/local-host-rs/src/telemetry/relay.rs @@ -1,4 +1,4 @@ -//! Bounded, content-free handoff from runtime event delivery to turn telemetry. +//! Process-wide, bounded, content-free handoff from runtime events to telemetry. //! //! The producer never waits and never performs telemetry I/O. Rich runtime //! payloads remain on their existing consumer path; this queue receives only @@ -11,20 +11,250 @@ use super::{ }; use crate::agent::{FromAgent, TokenUsage}; use maestro_runtime::ExecutionStatus; +use std::collections::HashMap; use std::sync::{ - Arc, RwLock, + Arc, OnceLock, RwLock, atomic::{AtomicU64, Ordering}, - mpsc::{SyncSender, TrySendError, sync_channel}, + mpsc::{Receiver, Sender, SyncSender, TrySendError, channel, sync_channel}, }; -const QUEUE_CAPACITY: usize = 64; +const QUEUE_CAPACITY: usize = 256; const WORKER_STACK_BYTES: usize = 256 * 1024; -const MAX_LABEL_BYTES: usize = 256; +const INLINE_LABEL_BYTES: usize = 80; +const MAX_BATCH: usize = 256; + +static TELEMETRY_HUB: OnceLock = OnceLock::new(); +static NEXT_RELAY_ID: AtomicU64 = AtomicU64::new(1); +#[cfg(test)] +static WORKER_STARTS: AtomicU64 = AtomicU64::new(0); + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(crate) struct InlineLabel { + len: u8, + bytes: [u8; INLINE_LABEL_BYTES], +} + +impl InlineLabel { + fn new(value: &str) -> Self { + let mut len = value.len().min(INLINE_LABEL_BYTES); + while !value.is_char_boundary(len) { + len -= 1; + } + let mut bytes = [0; INLINE_LABEL_BYTES]; + bytes[..len].copy_from_slice(&value.as_bytes()[..len]); + Self { + len: len as u8, + bytes, + } + } + + fn as_str(&self) -> &str { + std::str::from_utf8(&self.bytes[..usize::from(self.len)]) + .expect("inline telemetry labels are copied from valid UTF-8") + } +} + +struct TelemetryHub { + sender: SyncSender, + control_sender: Sender, + worker: std::thread::Thread, +} + +struct RelayRegistration { + relay_id: u64, + config: TurnTrackerConfig, + context: TurnTrackerContext, + identity_scope: Arc>>, + telemetry_host: Option, +} + +pub(crate) enum HubMessage { + Observe { + relay_id: u64, + queued: QueuedObservation, + }, +} + +enum ControlMessage { + Register(Box), + Unregister { relay_id: u64, attempted: u64 }, +} + +struct RelayWorkerState { + tracker: TurnTracker, + identity_scope: Arc>>, + telemetry_host: Option, + sequence_cursor: SequenceCursor, + closing_attempted: Option, +} + +#[derive(Debug, PartialEq, Eq)] +struct DrainReport { + observations: usize, + closed: Vec<(u64, u32)>, +} + +fn telemetry_hub() -> &'static TelemetryHub { + TELEMETRY_HUB.get_or_init(|| { + let (sender, receiver) = sync_channel(QUEUE_CAPACITY); + let (control_sender, control_receiver) = channel(); + let worker = std::thread::Builder::new() + .name("maestro-telemetry".to_owned()) + .stack_size(WORKER_STACK_BYTES) + .spawn(move || run_hub(receiver, control_receiver)) + .expect("spawn process telemetry worker"); + #[cfg(test)] + WORKER_STARTS.fetch_add(1, Ordering::Relaxed); + TelemetryHub { + sender, + control_sender, + worker: worker.thread().clone(), + } + }) +} + +fn run_hub(receiver: Receiver, control_receiver: Receiver) { + let mut states = HashMap::::new(); + let mut journal = TurnJournal::open(); + loop { + drain_hub_once(&receiver, &control_receiver, &mut states, &mut journal); + std::thread::park(); + } +} + +fn drain_control_messages( + control_receiver: &Receiver, + states: &mut HashMap, +) { + for message in control_receiver.try_iter() { + handle_control_message(message, states); + } +} + +fn drain_hub_once( + receiver: &Receiver, + control_receiver: &Receiver, + states: &mut HashMap, + journal: &mut Option, +) -> DrainReport { + // Control is sent before data by every relay. Drain it again after the + // data wakeup so a newly registered relay cannot lose its first event. + drain_control_messages(control_receiver, states); + let mut processed = 0; + for message in receiver.try_iter().take(MAX_BATCH) { + handle_hub_message(message, states, journal); + processed += 1; + } + let closed = finalize_closed_relays(states, journal); + DrainReport { + observations: processed, + closed, + } +} + +fn finalize_closed_relays( + states: &mut HashMap, + journal: &mut Option, +) -> Vec<(u64, u32)> { + let closing = states + .iter() + .filter_map(|(relay_id, state)| { + state + .closing_attempted + .map(|attempted| (*relay_id, attempted)) + }) + .collect::>(); + let mut closed = Vec::with_capacity(closing.len()); + for (relay_id, attempted) in closing { + let mut state = states + .remove(&relay_id) + .expect("closing telemetry relay must still be registered"); + recover_after_loss( + &mut state.tracker, + journal, + state.sequence_cursor.trailing_losses(attempted), + ); + closed.push((relay_id, state.tracker.turn_number())); + } + closed +} + +fn handle_hub_message( + message: HubMessage, + states: &mut HashMap, + journal: &mut Option, +) { + let HubMessage::Observe { relay_id, queued } = message; + let Some(state) = states.get_mut(&relay_id) else { + return; + }; + recover_after_loss( + &mut state.tracker, + journal, + state.sequence_cursor.observe(queued.sequence), + ); + let observation = queued.observation; + let snapshot = observation.needs_pending_snapshot(); + if observation.refreshes_identity() { + state.tracker.set_identity_scope( + state + .identity_scope + .read() + .expect("telemetry identity scope lock poisoned") + .clone(), + ); + } + if observation.refreshes_session() { + let session_id = state + .telemetry_host + .as_ref() + .and_then(|host| futures::executor::block_on(host.hook_session_id())); + state.tracker.set_session_id(session_id.unwrap_or_default()); + } + if let Some(completed) = handle_observation(&mut state.tracker, observation).as_ref() { + finish(journal, completed); + } + if snapshot { + observe_pending(journal, &state.tracker); + } +} + +fn handle_control_message(message: ControlMessage, states: &mut HashMap) { + match message { + ControlMessage::Register(registration) => { + let mut tracker = TurnTracker::new(registration.config); + tracker.update_context(registration.context); + states.insert( + registration.relay_id, + RelayWorkerState { + tracker, + identity_scope: registration.identity_scope, + telemetry_host: registration.telemetry_host, + sequence_cursor: SequenceCursor::default(), + closing_attempted: None, + }, + ); + } + ControlMessage::Unregister { + relay_id, + attempted, + } => { + if let Some(state) = states.get_mut(&relay_id) { + state.closing_attempted = Some(attempted); + } + } + } +} + +#[cfg(test)] +fn worker_start_count_for_test() -> u64 { + WORKER_STARTS.load(Ordering::Relaxed) +} #[derive(Debug, Clone)] pub(crate) struct ToolReceiptObservation { - call_id: String, - tool_name: String, + call_id: InlineLabel, + tool_name: InlineLabel, status: ExecutionStatus, duration_ms: Option, } @@ -32,31 +262,143 @@ pub(crate) struct ToolReceiptObservation { #[derive(Debug)] #[allow( clippy::large_enum_variant, - reason = "boxing would add producer-path allocation; the fixed queue is capped below 64 KiB" + reason = "boxing would allocate on the producer path; the process queue has a tested 128 KiB cap" )] pub(crate) enum TurnObservation { - Event(FromAgent), + Event(TelemetryEvent), Output { bytes: u64, first_observed_at: std::time::Instant, }, SideQuestionEnd { - side_id: String, + side_id: InlineLabel, answer_bytes: u64, usage: Option, failed: bool, }, ToolCall { - call_id: String, - tool: String, + call_id: InlineLabel, + tool: InlineLabel, requires_approval: bool, }, ToolEnd { - call_id: String, + call_id: InlineLabel, success: bool, receipt: Option, }, - Session(Option), + Session(Option), +} + +#[derive(Debug)] +#[allow( + clippy::large_enum_variant, + reason = "boxing would allocate on the producer path; the process queue has a tested 128 KiB cap" +)] +pub(crate) enum TelemetryEvent { + Ready { + model: InlineLabel, + provider: InlineLabel, + }, + ModelChanged { + model: InlineLabel, + provider: InlineLabel, + }, + BoostChanged { + status: maestro_runtime::agent::BoostStatus, + thinking: Option, + }, + ResponseStart { + response_id: InlineLabel, + }, + ResponseEnd { + response_id: InlineLabel, + usage: Option, + }, + TurnCompleted, + TurnInterrupted, + SideQuestionStart { + side_id: InlineLabel, + standalone: bool, + }, + CodexUsageState { + source: InlineLabel, + usage: Option, + }, + StreamObservation(SafeStreamObservation), + ContextCalibration { + request_id: InlineLabel, + generation: u64, + estimated_input_tokens: u64, + observed_input_tokens: u64, + }, + RequestRetryObservation, + RequestContextPrepared { + response_id: InlineLabel, + }, + OperationObservation(CompactOperationObservation), + TurnStarted, + RequestRetryScheduled { + attempt: u32, + delay_ms: u64, + rate_limited: bool, + }, + CompactionMeasured { + duration_ms: u64, + }, + ToolStart { + call_id: InlineLabel, + }, + Error { + fatal: bool, + terminal: bool, + retryable: bool, + }, + ProviderError { + kind: maestro_ai::ProviderStreamErrorKind, + }, + Compaction { + first_kept_entry_index: usize, + tokens_before: u64, + auto: bool, + }, +} + +#[derive(Debug, Clone, Copy)] +pub(crate) enum SafeStreamObservation { + Observed, + OpenFailed, + IdleTimeout, + Disconnect, + Retry, + Recovery, +} + +#[derive(Debug)] +pub(crate) enum CompactOperationObservation { + Admitted { + turn_id: InlineLabel, + thinking_level: InlineLabel, + experiment: Option, + }, + Prepared { + response_id: InlineLabel, + model_id: InlineLabel, + model_provider: InlineLabel, + message_count: u32, + input_size_bytes: Option, + }, + ReasoningUsage { + response_id: InlineLabel, + tokens: u64, + }, + GatewayReceipt { + response_id: InlineLabel, + request_id: InlineLabel, + record_id: InlineLabel, + lineage_id: InlineLabel, + provider_tools_sha256: Option, + provider_tool_count: Option, + }, } #[derive(Debug)] @@ -87,29 +429,29 @@ impl TurnObservation { matches!( self, Self::Event( - FromAgent::TurnStarted - | FromAgent::OperationObservation { .. } - | FromAgent::SideQuestionStart { .. } - | FromAgent::ResponseStart { .. } - | FromAgent::ResponseEnd { .. } + TelemetryEvent::TurnStarted + | TelemetryEvent::OperationObservation(_) + | TelemetryEvent::SideQuestionStart { .. } + | TelemetryEvent::ResponseStart { .. } + | TelemetryEvent::ResponseEnd { .. } ) ) } fn refreshes_identity(&self) -> bool { - matches!(self, Self::Event(FromAgent::ModelChanged { .. })) + matches!(self, Self::Event(TelemetryEvent::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 { .. } - } + TelemetryEvent::TurnStarted + | TelemetryEvent::ResponseStart { .. } + | TelemetryEvent::SideQuestionStart { .. } + | TelemetryEvent::OperationObservation( + CompactOperationObservation::Admitted { .. } + ) ) ) } @@ -117,10 +459,13 @@ impl TurnObservation { /// 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. +/// slots. All agents share one parked worker and one journal; aggregation, +/// serialization, locking, disk, and network work stay off the event relay. pub(crate) struct TurnTelemetryRelay { - sender: Option>, + sender: Option>, + control_sender: Option>, + worker: Option, + relay_id: u64, dropped: Arc, attempted: Arc, next_sequence: u64, @@ -138,6 +483,9 @@ impl TurnTelemetryRelay { if super::experiments_telemetry_disabled() { return Self { sender: None, + control_sender: None, + worker: None, + relay_id: 0, dropped: Arc::new(AtomicU64::new(0)), attempted: Arc::new(AtomicU64::new(0)), next_sequence: 0, @@ -146,57 +494,30 @@ impl TurnTelemetryRelay { }; } - let (sender, receiver) = sync_channel::(QUEUE_CAPACITY); + let sender = telemetry_hub().sender.clone(); + let control_sender = telemetry_hub().control_sender.clone(); + let worker = telemetry_hub().worker.clone(); + let relay_id = NEXT_RELAY_ID.fetch_add(1, Ordering::Relaxed); 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)), - ); - }); + let registered = control_sender + .send(ControlMessage::Register(Box::new(RelayRegistration { + relay_id, + config, + context, + identity_scope, + telemetry_host, + }))) + .is_ok(); + if registered { + worker.unpark(); + } Self { - sender: worker.ok().map(|_| sender), + sender: registered.then_some(sender), + control_sender: registered.then_some(control_sender), + worker: registered.then_some(worker), + relay_id, dropped, attempted, next_sequence: 0, @@ -242,7 +563,7 @@ impl TurnTelemetryRelay { | FromAgent::Status { .. } | FromAgent::HookBlocked { .. } => None, FromAgent::SessionInfo { session_id, .. } => Some(TurnObservation::Session( - session_id.as_deref().map(bounded_label), + session_id.as_deref().map(InlineLabel::new), )), FromAgent::ToolCall { call_id, @@ -250,8 +571,8 @@ impl TurnTelemetryRelay { requires_approval, .. } => Some(TurnObservation::ToolCall { - call_id: bounded_label(call_id), - tool: bounded_label(tool), + call_id: InlineLabel::new(call_id), + tool: InlineLabel::new(tool), requires_approval: *requires_approval, }), FromAgent::ToolEnd { @@ -260,11 +581,11 @@ impl TurnTelemetryRelay { receipt, .. } => Some(TurnObservation::ToolEnd { - call_id: bounded_label(call_id), + call_id: InlineLabel::new(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), + call_id: InlineLabel::new(&receipt.call_id), + tool_name: InlineLabel::new(&receipt.tool_name), status: receipt.status, duration_ms: receipt.duration_ms, }), @@ -276,7 +597,7 @@ impl TurnTelemetryRelay { error, .. } => Some(TurnObservation::SideQuestionEnd { - side_id: bounded_label(side_id), + side_id: InlineLabel::new(side_id), answer_bytes: answer.len() as u64, usage: usage.clone(), failed: error.is_some(), @@ -314,35 +635,48 @@ impl TurnTelemetryRelay { 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, + match sender.try_send(HubMessage::Observe { + relay_id: self.relay_id, + queued: QueuedObservation { + sequence, + observation, + }, }) { - self.record_drop(error); + Ok(()) => { + if let Some(worker) = &self.worker { + worker.unpark(); + } + } + Err(error) => self.record_drop(error), } } - fn record_drop(&self, error: TrySendError) { - if matches!(error, TrySendError::Full(_)) { - self.dropped.fetch_add(1, Ordering::Relaxed); - } + fn record_drop(&self, _error: TrySendError) { + self.dropped.fetch_add(1, Ordering::Relaxed); } #[cfg(test)] - pub(crate) fn saturated_for_test() -> (Self, std::sync::mpsc::Receiver) { + pub(crate) fn saturated_for_test() -> (Self, std::sync::mpsc::Receiver) { let (sender, receiver) = sync_channel(1); + let (control_sender, _control_receiver) = channel(); sender - .try_send(QueuedObservation { - sequence: 0, - observation: TurnObservation::Output { - bytes: 1, - first_observed_at: std::time::Instant::now(), + .try_send(HubMessage::Observe { + relay_id: 1, + queued: 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), + control_sender: Some(control_sender), + worker: None, + relay_id: 1, dropped: Arc::new(AtomicU64::new(0)), attempted: Arc::new(AtomicU64::new(1)), next_sequence: 1, @@ -354,6 +688,20 @@ impl TurnTelemetryRelay { } } +impl Drop for TurnTelemetryRelay { + fn drop(&mut self) { + if let Some(sender) = self.control_sender.take() { + let _ = sender.send(ControlMessage::Unregister { + relay_id: self.relay_id, + attempted: self.attempted.load(Ordering::Acquire), + }); + if let Some(worker) = &self.worker { + worker.unpark(); + } + } + } +} + fn recover_after_loss(tracker: &mut TurnTracker, journal: &mut Option, count: u64) { if count == 0 { return; @@ -386,7 +734,7 @@ fn handle_observation( observation: TurnObservation, ) -> Option { match observation { - TurnObservation::Event(event) => tracker.handle_event(&event), + TurnObservation::Event(event) => tracker.handle_event(&event.into_agent_event()), TurnObservation::Output { bytes, first_observed_at, @@ -399,13 +747,13 @@ fn handle_observation( answer_bytes, usage, failed, - } => tracker.finish_side_question(&side_id, answer_bytes, usage.as_ref(), failed), + } => tracker.finish_side_question(side_id.as_str(), answer_bytes, usage.as_ref(), failed), TurnObservation::ToolCall { call_id, tool, requires_approval, } => { - tracker.record_tool_call(&call_id, &tool, None, requires_approval); + tracker.record_tool_call(call_id.as_str(), tool.as_str(), None, requires_approval); None } TurnObservation::ToolEnd { @@ -414,7 +762,7 @@ fn handle_observation( receipt, } => { tracker.record_tool_end( - &call_id, + call_id.as_str(), success, receipt.as_ref().map(|receipt| { ( @@ -428,117 +776,291 @@ fn handle_observation( None } TurnObservation::Session(session_id) => { - tracker.set_session_id(session_id.unwrap_or_default()); + tracker.set_session_id( + session_id.map_or_else(String::new, |value| value.as_str().to_owned()), + ); None } } } -fn sanitized_event(event: &FromAgent) -> Option { +impl TelemetryEvent { + fn into_agent_event(self) -> FromAgent { + match self { + Self::Ready { model, provider } => FromAgent::Ready { + model: model.as_str().to_owned(), + provider: provider.as_str().to_owned(), + }, + Self::ModelChanged { model, provider } => FromAgent::ModelChanged { + model: model.as_str().to_owned(), + provider: provider.as_str().to_owned(), + }, + Self::BoostChanged { status, thinking } => FromAgent::BoostChanged { status, thinking }, + Self::ResponseStart { response_id } => FromAgent::ResponseStart { + response_id: response_id.as_str().to_owned(), + }, + Self::ResponseEnd { response_id, usage } => FromAgent::ResponseEnd { + response_id: response_id.as_str().to_owned(), + usage, + }, + Self::TurnCompleted => FromAgent::TurnCompleted { + response_id: String::new(), + coding_completion: None, + coding_child_records: Vec::new(), + }, + Self::TurnInterrupted => FromAgent::TurnInterrupted { + response_id: String::new(), + reason: String::new(), + }, + Self::SideQuestionStart { + side_id, + standalone, + } => FromAgent::SideQuestionStart { + side_id: side_id.as_str().to_owned(), + question: String::new(), + standalone, + }, + Self::CodexUsageState { source, usage } => FromAgent::CodexUsageState { + source: source.as_str().to_owned(), + usage, + }, + Self::StreamObservation(observation) => FromAgent::StreamObservation { + observation: observation.into_runtime(), + }, + Self::ContextCalibration { + request_id, + generation, + estimated_input_tokens, + observed_input_tokens, + } => FromAgent::ContextCalibration { + observation: maestro_context::context_usage::ContextCalibration { + request_id: request_id.as_str().to_owned(), + generation, + estimated_input_tokens, + observed_input_tokens, + }, + }, + Self::RequestRetryObservation => FromAgent::RequestRetryObservation, + Self::RequestContextPrepared { response_id } => FromAgent::RequestContextPrepared { + response_id: response_id.as_str().to_owned(), + }, + Self::OperationObservation(observation) => FromAgent::OperationObservation { + observation: observation.into_runtime(), + }, + Self::TurnStarted => FromAgent::TurnStarted, + Self::RequestRetryScheduled { + attempt, + delay_ms, + rate_limited, + } => FromAgent::RequestRetryScheduled { + attempt, + delay_ms, + rate_limited, + }, + Self::CompactionMeasured { duration_ms } => { + FromAgent::CompactionMeasured { duration_ms } + } + Self::ToolStart { call_id } => FromAgent::ToolStart { + call_id: call_id.as_str().to_owned(), + }, + Self::Error { + fatal, + terminal, + retryable, + } => FromAgent::Error { + message: String::new(), + fatal, + terminal, + retryable, + }, + Self::ProviderError { kind } => FromAgent::ProviderError { + kind, + message: String::new(), + }, + Self::Compaction { + first_kept_entry_index, + tokens_before, + auto, + } => FromAgent::Compaction { + summary: String::new(), + first_kept_entry_index, + tokens_before, + auto, + custom_instructions: None, + continuation: None, + timestamp: String::new(), + }, + } + } +} + +impl SafeStreamObservation { + fn from_runtime(observation: maestro_ai::StreamObservation) -> Self { + match observation { + maestro_ai::StreamObservation::Observed => Self::Observed, + maestro_ai::StreamObservation::OpenFailed => Self::OpenFailed, + maestro_ai::StreamObservation::IdleTimeout => Self::IdleTimeout, + maestro_ai::StreamObservation::Disconnect => Self::Disconnect, + maestro_ai::StreamObservation::Retry => Self::Retry, + maestro_ai::StreamObservation::Recovery => Self::Recovery, + } + } + + fn into_runtime(self) -> maestro_ai::StreamObservation { + match self { + Self::Observed => maestro_ai::StreamObservation::Observed, + Self::OpenFailed => maestro_ai::StreamObservation::OpenFailed, + Self::IdleTimeout => maestro_ai::StreamObservation::IdleTimeout, + Self::Disconnect => maestro_ai::StreamObservation::Disconnect, + Self::Retry => maestro_ai::StreamObservation::Retry, + Self::Recovery => maestro_ai::StreamObservation::Recovery, + } + } +} + +impl CompactOperationObservation { + fn into_runtime( + self, + ) -> maestro_runtime_contracts::operation_observation::OperationObservation { + use maestro_runtime_contracts::operation_observation::OperationObservation as O; + match self { + Self::Admitted { + turn_id, + thinking_level, + experiment, + } => O::Admitted { + turn_id: turn_id.as_str().to_owned(), + thinking_level: thinking_level.as_str().to_owned(), + experiment, + }, + Self::Prepared { + response_id, + model_id, + model_provider, + message_count, + input_size_bytes, + } => O::Prepared { + response_id: response_id.as_str().to_owned(), + model_id: model_id.as_str().to_owned(), + model_provider: model_provider.as_str().to_owned(), + message_count, + input_size_bytes, + }, + Self::ReasoningUsage { + response_id, + tokens, + } => O::ReasoningUsage { + response_id: response_id.as_str().to_owned(), + tokens, + }, + Self::GatewayReceipt { + response_id, + request_id, + record_id, + lineage_id, + provider_tools_sha256, + provider_tool_count, + } => O::GatewayReceipt { + response_id: response_id.as_str().to_owned(), + request_id: request_id.as_str().to_owned(), + record_id: record_id.as_str().to_owned(), + lineage_id: lineage_id.as_str().to_owned(), + provider_tools_sha256: provider_tools_sha256.map(|value| value.as_str().to_owned()), + provider_tool_count, + }, + } + } +} + +fn sanitized_event(event: &FromAgent) -> Option { Some(match event { - FromAgent::Ready { model, provider } => FromAgent::Ready { - model: bounded_label(model), - provider: bounded_label(provider), + FromAgent::Ready { model, provider } => TelemetryEvent::Ready { + model: InlineLabel::new(model), + provider: InlineLabel::new(provider), }, - FromAgent::ModelChanged { model, provider } => FromAgent::ModelChanged { - model: bounded_label(model), - provider: bounded_label(provider), + FromAgent::ModelChanged { model, provider } => TelemetryEvent::ModelChanged { + model: InlineLabel::new(model), + provider: InlineLabel::new(provider), }, - FromAgent::BoostChanged { status, thinking } => FromAgent::BoostChanged { + FromAgent::BoostChanged { status, thinking } => TelemetryEvent::BoostChanged { status: *status, thinking: *thinking, }, - FromAgent::ResponseStart { response_id } => FromAgent::ResponseStart { - response_id: bounded_label(response_id), + FromAgent::ResponseStart { response_id } => TelemetryEvent::ResponseStart { + response_id: InlineLabel::new(response_id), }, - FromAgent::ResponseEnd { response_id, usage } => FromAgent::ResponseEnd { - response_id: bounded_label(response_id), + FromAgent::ResponseEnd { response_id, usage } => TelemetryEvent::ResponseEnd { + response_id: InlineLabel::new(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::TurnCompleted { .. } => TelemetryEvent::TurnCompleted, + FromAgent::TurnInterrupted { .. } => TelemetryEvent::TurnInterrupted, FromAgent::SideQuestionStart { side_id, standalone, .. - } => FromAgent::SideQuestionStart { - side_id: bounded_label(side_id), - question: String::new(), + } => TelemetryEvent::SideQuestionStart { + side_id: InlineLabel::new(side_id), standalone: *standalone, }, - FromAgent::CodexUsageState { source, usage } => FromAgent::CodexUsageState { - source: bounded_label(source), + FromAgent::CodexUsageState { source, usage } => TelemetryEvent::CodexUsageState { + source: InlineLabel::new(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::StreamObservation { observation } => { + TelemetryEvent::StreamObservation(SafeStreamObservation::from_runtime(*observation)) + } + FromAgent::ContextCalibration { observation } => TelemetryEvent::ContextCalibration { + request_id: InlineLabel::new(&observation.request_id), + generation: observation.generation, + estimated_input_tokens: observation.estimated_input_tokens, + observed_input_tokens: observation.observed_input_tokens, }, - FromAgent::TurnStarted => FromAgent::TurnStarted, + FromAgent::RequestRetryObservation => TelemetryEvent::RequestRetryObservation, + FromAgent::RequestContextPrepared { response_id } => { + TelemetryEvent::RequestContextPrepared { + response_id: InlineLabel::new(response_id), + } + } + FromAgent::OperationObservation { observation } => { + TelemetryEvent::OperationObservation(sanitized_operation_observation(observation)) + } + FromAgent::TurnStarted => TelemetryEvent::TurnStarted, FromAgent::RequestRetryScheduled { attempt, delay_ms, rate_limited, - } => FromAgent::RequestRetryScheduled { + } => TelemetryEvent::RequestRetryScheduled { attempt: *attempt, delay_ms: *delay_ms, rate_limited: *rate_limited, }, - FromAgent::CompactionMeasured { duration_ms } => FromAgent::CompactionMeasured { + FromAgent::CompactionMeasured { duration_ms } => TelemetryEvent::CompactionMeasured { duration_ms: *duration_ms, }, - FromAgent::ToolStart { call_id } => FromAgent::ToolStart { - call_id: bounded_label(call_id), + FromAgent::ToolStart { call_id } => TelemetryEvent::ToolStart { + call_id: InlineLabel::new(call_id), }, FromAgent::Error { fatal, terminal, retryable, .. - } => FromAgent::Error { - message: String::new(), + } => TelemetryEvent::Error { fatal: *fatal, terminal: *terminal, retryable: *retryable, }, - FromAgent::ProviderError { kind, .. } => FromAgent::ProviderError { - kind: *kind, - message: String::new(), - }, + FromAgent::ProviderError { kind, .. } => TelemetryEvent::ProviderError { kind: *kind }, FromAgent::Compaction { first_kept_entry_index, tokens_before, auto, .. - } => FromAgent::Compaction { - summary: String::new(), + } => TelemetryEvent::Compaction { first_kept_entry_index: *first_kept_entry_index, tokens_before: *tokens_before, auto: *auto, - custom_instructions: None, - continuation: None, - timestamp: String::new(), }, _ => return None, }) @@ -546,16 +1068,16 @@ fn sanitized_event(event: &FromAgent) -> Option { fn sanitized_operation_observation( observation: &maestro_runtime_contracts::operation_observation::OperationObservation, -) -> maestro_runtime_contracts::operation_observation::OperationObservation { +) -> CompactOperationObservation { 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), + } => CompactOperationObservation::Admitted { + turn_id: InlineLabel::new(turn_id), + thinking_level: InlineLabel::new(thinking_level), experiment: experiment.clone().filter(|value| value.is_valid()), }, O::Prepared { @@ -564,18 +1086,18 @@ fn sanitized_operation_observation( 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), + } => CompactOperationObservation::Prepared { + response_id: InlineLabel::new(response_id), + model_id: InlineLabel::new(model_id), + model_provider: InlineLabel::new(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), + } => CompactOperationObservation::ReasoningUsage { + response_id: InlineLabel::new(response_id), tokens: *tokens, }, O::GatewayReceipt { @@ -585,28 +1107,17 @@ fn sanitized_operation_observation( 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), + } => CompactOperationObservation::GatewayReceipt { + response_id: InlineLabel::new(response_id), + request_id: InlineLabel::new(request_id), + record_id: InlineLabel::new(record_id), + lineage_id: InlineLabel::new(lineage_id), + provider_tools_sha256: provider_tools_sha256.as_deref().map(InlineLabel::new), 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::*; @@ -616,6 +1127,9 @@ mod tests { let (sender, receiver) = sync_channel(2); let relay = TurnTelemetryRelay { sender: Some(sender), + control_sender: None, + worker: None, + relay_id: 1, dropped: Arc::new(AtomicU64::new(0)), attempted: Arc::new(AtomicU64::new(0)), next_sequence: 0, @@ -637,65 +1151,195 @@ mod tests { assert!(receiver.try_recv().is_err()); } + #[test] + fn projected_records_cannot_retain_tool_arguments_or_errors() { + let mut relay = disabled_relay(); + let tool = relay + .project(&FromAgent::ToolCall { + call_id: "call-1".into(), + tool: "bash".into(), + args: serde_json::json!({"command": "super-secret-command"}), + requires_approval: true, + approval_inline_env: None, + }) + .unwrap(); + let error = relay + .project(&FromAgent::Error { + message: "super-secret-error".into(), + fatal: true, + terminal: true, + retryable: false, + }) + .unwrap(); + let projected = format!("{tool:?} {error:?}"); + assert!(!projected.contains("super-secret")); + assert!(!projected.contains("command")); + } + #[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())); + let value = "🦀".repeat(INLINE_LABEL_BYTES); + let bounded = InlineLabel::new(&value); + assert!(bounded.as_str().len() <= INLINE_LABEL_BYTES); + assert!(bounded.as_str().is_char_boundary(bounded.as_str().len())); + assert!(!bounded.as_str().is_empty()); } #[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(), + .try_send(HubMessage::Observe { + relay_id: 1, + queued: QueuedObservation { + sequence: 0, + observation: TurnObservation::Output { + bytes: 1, + first_observed_at: std::time::Instant::now(), + }, }, }) .unwrap(); let relay = TurnTelemetryRelay { sender: Some(sender), + control_sender: None, + worker: None, + relay_id: 1, 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(), + relay.record_drop(TrySendError::Full(HubMessage::Observe { + relay_id: 1, + queued: 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 disconnected_worker_is_counted_as_telemetry_loss() { + let (sender, receiver) = sync_channel(1); + drop(receiver); + let mut relay = TurnTelemetryRelay { + sender: Some(sender), + control_sender: None, + worker: None, + relay_id: 1, + dropped: Arc::new(AtomicU64::new(0)), + attempted: Arc::new(AtomicU64::new(0)), + next_sequence: 0, + pending_output_bytes: 0, + pending_first_output_at: None, + }; + relay.try_submit(TurnObservation::Event(TelemetryEvent::TurnStarted)); + assert_eq!(relay.dropped.load(Ordering::Relaxed), 1); + } + #[test] fn queue_storage_stays_small() { + let record_bytes = std::mem::size_of::(); + let queue_bytes = record_bytes * QUEUE_CAPACITY; + eprintln!("telemetry record: {record_bytes} bytes; queue: {queue_bytes} bytes"); assert!( - std::mem::size_of::() * QUEUE_CAPACITY <= 64 * 1024, - "the preallocated observation queue must stay within 64 KiB" + queue_bytes <= 128 * 1024, + "the process-wide observation queue must stay within 128 KiB" ); } + #[test] + fn every_relay_uses_the_single_process_worker() { + let before = worker_start_count_for_test(); + let first = test_relay(); + let second = test_relay(); + assert_eq!(worker_start_count_for_test(), before.max(1)); + drop((first, second)); + } + + #[test] + fn registration_is_applied_before_the_first_observation() { + let (sender, receiver) = sync_channel(1); + let (control_sender, control_receiver) = channel(); + control_sender + .send(ControlMessage::Register(Box::new(RelayRegistration { + relay_id: 41, + config: test_config(), + context: TurnTrackerContext::default(), + identity_scope: Arc::new(RwLock::new(None)), + telemetry_host: None, + }))) + .unwrap(); + sender + .send(HubMessage::Observe { + relay_id: 41, + queued: QueuedObservation { + sequence: 0, + observation: TurnObservation::Event(TelemetryEvent::TurnStarted), + }, + }) + .unwrap(); + + let mut states = HashMap::new(); + let mut journal = None; + assert_eq!( + drain_hub_once(&receiver, &control_receiver, &mut states, &mut journal).observations, + 1 + ); + + assert_eq!(states.get(&41).unwrap().tracker.turn_number(), 1); + } + + #[test] + fn unregister_drains_the_final_observation_before_removing_state() { + let (sender, receiver) = sync_channel(1); + let (control_sender, control_receiver) = channel(); + control_sender + .send(ControlMessage::Register(Box::new(RelayRegistration { + relay_id: 42, + config: test_config(), + context: TurnTrackerContext::default(), + identity_scope: Arc::new(RwLock::new(None)), + telemetry_host: None, + }))) + .unwrap(); + sender + .send(HubMessage::Observe { + relay_id: 42, + queued: QueuedObservation { + sequence: 0, + observation: TurnObservation::Event(TelemetryEvent::TurnStarted), + }, + }) + .unwrap(); + control_sender + .send(ControlMessage::Unregister { + relay_id: 42, + attempted: 1, + }) + .unwrap(); + + let mut states = HashMap::new(); + let mut journal = None; + let report = drain_hub_once(&receiver, &control_receiver, &mut states, &mut journal); + + assert_eq!(report.observations, 1); + assert_eq!(report.closed, vec![(42, 1)]); + assert!(!states.contains_key(&42)); + } + #[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 mut relay = disabled_relay(); let event = FromAgent::ResponseChunk { response_id: "response".into(), content: "one streamed token".into(), @@ -715,6 +1359,38 @@ mod tests { assert!(p99 < 1_000, "response chunk projection p99 was {p99} ns"); } + #[test] + #[ignore = "local performance probe"] + fn observation_submission_p99_is_sub_microsecond() { + const BATCH: u32 = 128; + const SAMPLES: usize = 256; + let (sender, receiver) = sync_channel(QUEUE_CAPACITY); + let mut relay = TurnTelemetryRelay { + sender: Some(sender), + control_sender: None, + worker: Some(std::thread::current()), + relay_id: 1, + 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 nanos_per_event = Vec::with_capacity(SAMPLES); + for _ in 0..SAMPLES { + let started = std::time::Instant::now(); + for _ in 0..BATCH { + relay.try_submit(TurnObservation::Event(TelemetryEvent::TurnStarted)); + } + nanos_per_event.push(started.elapsed().as_nanos() / u128::from(BATCH)); + assert_eq!(receiver.try_iter().count(), BATCH as usize); + } + nanos_per_event.sort_unstable(); + let p99 = nanos_per_event[SAMPLES * 99 / 100]; + eprintln!("observation submission p99: {p99} ns"); + assert!(p99 < 1_000, "observation submission p99 was {p99} ns"); + } + #[test] fn sequence_gaps_are_attributed_before_the_following_observation() { let mut cursor = SequenceCursor::default(); @@ -722,4 +1398,34 @@ mod tests { assert_eq!(cursor.observe(2), 1); assert_eq!(cursor.trailing_losses(4), 1); } + + fn test_relay() -> TurnTelemetryRelay { + TurnTelemetryRelay::spawn( + test_config(), + TurnTrackerContext::default(), + Arc::new(RwLock::new(None)), + None, + ) + } + + fn test_config() -> TurnTrackerConfig { + TurnTrackerConfig { + session_id: String::new(), + sampling_config: super::super::TailSamplingConfig::default(), + } + } + + fn disabled_relay() -> TurnTelemetryRelay { + TurnTelemetryRelay { + sender: None, + control_sender: None, + worker: None, + relay_id: 0, + dropped: Arc::new(AtomicU64::new(0)), + attempted: Arc::new(AtomicU64::new(0)), + next_sequence: 0, + pending_output_bytes: 0, + pending_first_output_at: None, + } + } }