diff --git a/crates/metacrate-grid-agent/README.md b/crates/metacrate-grid-agent/README.md index 75e1819..533f464 100644 --- a/crates/metacrate-grid-agent/README.md +++ b/crates/metacrate-grid-agent/README.md @@ -57,3 +57,7 @@ and metadata-only operator controls are specified in The versioned integrated/TCP protocol, roles, framing, bounds, TLS remote-mode requirements, events, and management methods are specified in [`../../docs/grid-agent-control-plane.md`](../../docs/grid-agent-control-plane.md). +The unified event schema, pseudonymous correlations, bounded JSONL rotation, +fixed-cardinality metrics, diagnostic-capture warning, and non-executing replay +contract are specified in +[`../../docs/grid-agent-observability.md`](../../docs/grid-agent-observability.md). diff --git a/crates/metacrate-grid-agent/src/control_plane.rs b/crates/metacrate-grid-agent/src/control_plane.rs index e6fbff5..dd16a4a 100644 --- a/crates/metacrate-grid-agent/src/control_plane.rs +++ b/crates/metacrate-grid-agent/src/control_plane.rs @@ -4,6 +4,7 @@ use crate::backend::BackendFuture; use crate::config::SecretString; +use crate::observability::{MetricsSnapshot, StructuredEvent}; use libremetaverse_types::compat::{CancellationToken, CancellationTokenSource}; use serde::{Deserialize, Serialize}; use std::collections::{BTreeMap, BTreeSet, VecDeque}; @@ -244,11 +245,13 @@ pub struct AuditEventView { #[serde(tag = "kind", content = "data", rename_all = "snake_case")] pub enum ControlPayload { Health(HealthView), + Metrics(MetricsSnapshot), Runtime(RuntimeView), Sessions(Page), ScheduledJobs(Page), PendingApprovals(Page), AuditEvents(Page), + ObservabilityEvents(Page), Accepted { operation_id: String }, Completed, Subscribed { current_sequence: u64 }, @@ -266,6 +269,7 @@ pub enum ConversationChannelView { #[serde(tag = "method", content = "parameters", rename_all = "snake_case")] pub enum ControlRequest { Health, + Metrics, Runtime, ListSessions { page: PageRequest, @@ -279,6 +283,9 @@ pub enum ControlRequest { ListAuditEvents { page: PageRequest, }, + ListObservabilityEvents { + page: PageRequest, + }, SubscribeEvents { after_sequence: Option, }, @@ -326,11 +333,13 @@ impl ControlRequest { !matches!( self, Self::Health + | Self::Metrics | Self::Runtime | Self::ListSessions { .. } | Self::ListScheduledJobs { .. } | Self::ListPendingApprovals { .. } | Self::ListAuditEvents { .. } + | Self::ListObservabilityEvents { .. } | Self::SubscribeEvents { .. } ) } @@ -340,7 +349,8 @@ impl ControlRequest { Self::ListSessions { page } | Self::ListScheduledJobs { page } | Self::ListPendingApprovals { page } - | Self::ListAuditEvents { page } => page.valid(), + | Self::ListAuditEvents { page } + | Self::ListObservabilityEvents { page } => page.valid(), Self::CancelRequest { target_request_id } => valid_identifier(target_request_id), Self::CancelAction { action_id } => valid_identifier(action_id), Self::DecideApproval { approval_id, .. } => *approval_id != 0, @@ -952,7 +962,7 @@ enum ClientFrame { #[serde(tag = "frame", content = "body", rename_all = "snake_case")] enum ServerFrame { Hello(ServerHello), - Response(ControlResponseEnvelope), + Response(Box), Event(ControlEvent), } @@ -1245,7 +1255,10 @@ where true, ), ); - if outbound.try_send(ServerFrame::Response(response)).is_err() { + if outbound + .try_send(ServerFrame::Response(Box::new(response))) + .is_err() + { break; } continue; @@ -1256,7 +1269,7 @@ where tasks.spawn(async move { let (response, subscription) = request_core.request(envelope).await; if request_outbound - .try_send(ServerFrame::Response(response)) + .try_send(ServerFrame::Response(Box::new(response))) .is_err() { request_cancel.cancel(); @@ -1534,7 +1547,7 @@ async fn tcp_client_reader( match frame { Ok(ServerFrame::Response(response)) => { if let Some(sender) = lock(&shared.pending).remove(&response.request_id) { - let _ = sender.send(response); + let _ = sender.send(*response); } } Ok(ServerFrame::Event(event)) => { @@ -1744,11 +1757,13 @@ fn role_name(role: ControlRole) -> &'static str { fn request_name(request: &ControlRequest) -> &'static str { match request { ControlRequest::Health => "health", + ControlRequest::Metrics => "metrics", ControlRequest::Runtime => "runtime", ControlRequest::ListSessions { .. } => "list_sessions", ControlRequest::ListScheduledJobs { .. } => "list_scheduled_jobs", ControlRequest::ListPendingApprovals { .. } => "list_pending_approvals", ControlRequest::ListAuditEvents { .. } => "list_audit_events", + ControlRequest::ListObservabilityEvents { .. } => "list_observability_events", ControlRequest::SubscribeEvents { .. } => "subscribe_events", ControlRequest::CancelRequest { .. } => "cancel_request", ControlRequest::PauseAutonomy => "pause_autonomy", diff --git a/crates/metacrate-grid-agent/src/control_runtime.rs b/crates/metacrate-grid-agent/src/control_runtime.rs index 8ab99a1..455bf64 100644 --- a/crates/metacrate-grid-agent/src/control_runtime.rs +++ b/crates/metacrate-grid-agent/src/control_runtime.rs @@ -10,6 +10,10 @@ use crate::control_plane::{ SessionMetadataView, }; use crate::conversation::{ConversationChannel, ConversationKey, ConversationStore}; +use crate::observability::{ + CorrelationIds, EventDraft, EventFamily, EventOrigin, EventSeverity, MetricsSnapshot, + Observability, pseudonymous_identifier, +}; use crate::policy::{ ApprovalId, AuthenticatedPrincipal, MemoryPolicyAudit, PolicyFinalOutcome, PolicyGateway, PolicyReasonCode, @@ -79,6 +83,7 @@ pub struct AgentControlTarget { conversations: Arc, policy: Arc, audit: Arc, + observability: Mutex>>, behavior: BehaviorIngress, commands: mpsc::Sender, command_capacity: usize, @@ -121,6 +126,7 @@ impl AgentControlTarget { conversations, policy, audit, + observability: Mutex::new(None), behavior, commands, command_capacity, @@ -131,6 +137,11 @@ impl AgentControlTarget { )) } + /// Attaches the unified recorder used by the live service and control API. + pub fn attach_observability(&self, observability: Arc) { + *lock(&self.observability) = Some(observability); + } + pub fn update_session(&self, session: SessionStatus) { let mut state = lock(&self.state); state.session = session; @@ -205,6 +216,24 @@ impl AgentControlTarget { protocol_version: CONTROL_PROTOCOL_VERSION, })) } + ControlRequest::Metrics => { + let mut metrics = lock(&self.observability) + .as_ref() + .map_or_else(MetricsSnapshot::default, |observer| { + observer.metrics().snapshot() + }); + let state = lock(&self.state); + metrics.ready = state.session.agent_ready; + metrics.active_sessions = + u64::try_from(self.conversations.list_metadata().len()).unwrap_or(u64::MAX); + metrics.queue_depth = u64::try_from( + self.command_capacity + .saturating_sub(self.commands.capacity()), + ) + .unwrap_or(u64::MAX); + metrics.queue_capacity = u64::try_from(self.command_capacity).unwrap_or(u64::MAX); + Ok(ControlPayload::Metrics(metrics)) + } ControlRequest::Runtime => { let state = lock(&self.state).clone(); let usage = self.policy.global_budget_usage(); @@ -284,7 +313,7 @@ impl AgentControlTarget { .map(|(index, record)| AuditEventView { sequence: u64::try_from(index).unwrap_or(u64::MAX).saturating_add(1), unix_millis: record.recorded_unix_millis, - principal: record.principal.as_str().to_owned(), + principal: pseudonymous_identifier(record.principal.as_str()), operation: record.tool.as_str().to_owned(), outcome: policy_outcome_name(record.final_outcome).to_owned(), authorization_id: record.authorization_id, @@ -292,6 +321,14 @@ impl AgentControlTarget { .collect(); Ok(ControlPayload::AuditEvents(page_values(&page, values))) } + ControlRequest::ListObservabilityEvents { page } => { + let values = lock(&self.observability) + .as_ref() + .map_or_else(Vec::new, |observer| observer.snapshot()); + Ok(ControlPayload::ObservabilityEvents(page_values( + &page, values, + ))) + } ControlRequest::PauseAutonomy => { let response = self.enqueue("pause", RuntimeControlCommand::Pause)?; self.behavior.pause(); @@ -452,11 +489,185 @@ impl ControlTarget for AgentControlTarget { false, )); } - self.execute_now(&context, request) + let operation = runtime_request_name(&request); + let correlation = request_correlation(&request); + let domain_event = runtime_domain_event(&request).ok().flatten(); + let result = self.execute_now(&context, request); + if let Some(observability) = lock(&self.observability).as_ref() { + let (severity, result_code) = match &result { + Ok(_) => (EventSeverity::Info, "accepted"), + Err(error) => ( + if error.retryable { + EventSeverity::Warning + } else { + EventSeverity::Error + }, + control_error_name(error.code), + ), + }; + let event = EventDraft::new( + EventFamily::ControlCommand, + severity, + "control", + EventOrigin::Control, + ) + .and_then(|event| event.correlation(correlation)) + .and_then(|event| event.result_code(result_code)) + .and_then(|event| event.code_field("operation", operation)) + .and_then(|event| { + event.code_field( + "role", + match context.role { + crate::control_plane::ControlRole::Observer => "observer", + crate::control_plane::ControlRole::Operator => "operator", + }, + ) + }) + .and_then(|event| event.redacted("operator_payload")); + if let Ok(event) = event { + let _ = observability.record(event); + } + if result.is_ok() + && let Some(event) = domain_event + { + let _ = observability.record(event); + } + } + result }) } } +fn runtime_domain_event( + request: &ControlRequest, +) -> Result, crate::observability::ObservabilityError> { + let event = match request { + ControlRequest::SetRoamingJob { job_id, enabled } => Some( + EventDraft::new( + EventFamily::ScheduledJob, + EventSeverity::Info, + "scheduler", + EventOrigin::Control, + )? + .correlation(CorrelationIds { + action_id: Some(job_id.clone()), + ..CorrelationIds::default() + })? + .result_code("accepted")? + .code_field("state", if *enabled { "enabled" } else { "disabled" })?, + ), + ControlRequest::ExpireConversation { avatar_id, channel } => Some( + EventDraft::new( + EventFamily::SessionExpired, + EventSeverity::Info, + "conversation", + EventOrigin::Control, + )? + .correlation(CorrelationIds { + avatar_id: Some(avatar_id.clone()), + ..CorrelationIds::default() + })? + .result_code("completed")? + .reason_code("operator_expired")? + .code_field( + "channel", + match channel { + ConversationChannelView::PublicChat => "public_chat", + ConversationChannelView::DirectIm => "direct_im", + }, + )?, + ), + ControlRequest::DecideApproval { + approval_id, + approve, + } => Some( + EventDraft::new( + EventFamily::ApprovalDecision, + EventSeverity::Info, + "policy", + EventOrigin::Control, + )? + .correlation(CorrelationIds { + action_id: Some(format!("approval-{approval_id}")), + ..CorrelationIds::default() + })? + .result_code("completed")? + .code_field("decision", if *approve { "approved" } else { "denied" })?, + ), + ControlRequest::GracefulShutdown => Some( + EventDraft::new( + EventFamily::Shutdown, + EventSeverity::Info, + "service", + EventOrigin::Control, + )? + .result_code("requested")?, + ), + _ => None, + }; + Ok(event) +} + +fn request_correlation(request: &ControlRequest) -> CorrelationIds { + let action_id = match request { + ControlRequest::CancelAction { action_id } => Some(action_id.clone()), + ControlRequest::DecideApproval { approval_id, .. } => { + Some(format!("approval-{approval_id}")) + } + _ => None, + }; + CorrelationIds { + action_id, + ..CorrelationIds::default() + } +} + +const fn runtime_request_name(request: &ControlRequest) -> &'static str { + match request { + ControlRequest::Health => "health", + ControlRequest::Metrics => "metrics", + ControlRequest::Runtime => "runtime", + ControlRequest::ListSessions { .. } => "list_sessions", + ControlRequest::ListScheduledJobs { .. } => "list_scheduled_jobs", + ControlRequest::ListPendingApprovals { .. } => "list_pending_approvals", + ControlRequest::ListAuditEvents { .. } => "list_audit_events", + ControlRequest::ListObservabilityEvents { .. } => "list_observability_events", + ControlRequest::SubscribeEvents { .. } => "subscribe_events", + ControlRequest::CancelRequest { .. } => "cancel_request", + ControlRequest::PauseAutonomy => "pause_autonomy", + ControlRequest::ResumeAutonomy => "resume_autonomy", + ControlRequest::CancelAction { .. } => "cancel_action", + ControlRequest::DecideApproval { approve: true, .. } => "approve_proposal", + ControlRequest::DecideApproval { approve: false, .. } => "deny_proposal", + ControlRequest::ForceReconnect => "force_reconnect", + ControlRequest::ExpireConversation { .. } => "expire_conversation", + ControlRequest::SetRoamingJob { enabled: true, .. } => "enable_roaming_job", + ControlRequest::SetRoamingJob { enabled: false, .. } => "disable_roaming_job", + ControlRequest::InjectOperatorMessage { .. } => "inject_operator_message", + ControlRequest::GracefulShutdown => "graceful_shutdown", + } +} + +const fn control_error_name(code: ControlErrorCode) -> &'static str { + match code { + ControlErrorCode::AuthenticationFailed => "authentication_failed", + ControlErrorCode::VersionMismatch => "version_mismatch", + ControlErrorCode::PermissionDenied => "permission_denied", + ControlErrorCode::InvalidRequest => "invalid_request", + ControlErrorCode::Replay => "replay", + ControlErrorCode::NotFound => "not_found", + ControlErrorCode::Conflict => "conflict", + ControlErrorCode::Cancelled => "cancelled", + ControlErrorCode::TimedOut => "timed_out", + ControlErrorCode::Busy => "busy", + ControlErrorCode::Backpressure => "backpressure", + ControlErrorCode::FrameTooLarge => "frame_too_large", + ControlErrorCode::IdleTimeout => "idle_timeout", + ControlErrorCode::TransportClosed => "transport_closed", + ControlErrorCode::Internal => "internal", + } +} + fn page_values(request: &PageRequest, values: Vec) -> Page { let start = usize::try_from(request.cursor.unwrap_or(0)) .unwrap_or(usize::MAX) diff --git a/crates/metacrate-grid-agent/src/control_runtime_tests.rs b/crates/metacrate-grid-agent/src/control_runtime_tests.rs index 7f74c69..d036958 100644 --- a/crates/metacrate-grid-agent/src/control_runtime_tests.rs +++ b/crates/metacrate-grid-agent/src/control_runtime_tests.rs @@ -1,13 +1,13 @@ use crate::behavior::{BehaviorController, BehaviorError, EmbodimentFuture, EmbodimentSink}; use crate::control_plane::{ CONTROL_PROTOCOL_VERSION, ControlLimits, ControlPayload, ControlPlane, ControlRequest, - ControlRequestEnvelope, + ControlRequestEnvelope, PageRequest, }; use crate::control_runtime::{AgentControlTarget, RuntimeControlCommand}; use crate::conversation::{ConversationLimits, ConversationStore}; use crate::perception::WorldPosition; use crate::policy::{MemoryPolicyAudit, PolicyAuditSink, PolicyGateway, PolicyLimits}; -use crate::{AgentConfig, EmbodiedPose, SecretString}; +use crate::{AgentConfig, EmbodiedPose, Observability, ObservabilityLimits, SecretString}; use libremetaverse_types::UUID; use libremetaverse_types::compat::CancellationToken; use std::collections::BTreeSet; @@ -123,13 +123,15 @@ async fn production_target_projects_state_and_routes_real_mutations() { let (target, mut commands) = AgentControlTarget::new(conversations, policy, audit, behavior.ingress(), 8) .expect("target"); + let observability = Observability::memory(ObservabilityLimits::default()).expect("events"); + target.attach_observability(observability); target.update_region( Some("00000000-0000-4000-8000-000000000001".into()), Some("Test Region".into()), Some([128.0, 128.0, 24.0]), ); let plane = ControlPlane::new( - target, + target.clone(), SecretString::new("test.operator", "operator-token").expect("token"), Some(SecretString::new("test.observer", "observer-token").expect("token")), ControlLimits::default(), @@ -144,6 +146,13 @@ async fn production_target_projects_state_and_routes_real_mutations() { }; assert_eq!(runtime.region_name.as_deref(), Some("Test Region")); assert_eq!(runtime.control_queue_capacity, 8); + let metrics = operator + .request(envelope("metrics", ControlRequest::Metrics)) + .await; + let Ok(ControlPayload::Metrics(metrics)) = metrics.result else { + panic!("metrics projection") + }; + assert_eq!(metrics.queue_capacity, 8); assert!( operator @@ -163,5 +172,30 @@ async fn production_target_projects_state_and_routes_real_mutations() { .await; assert!(denied.result.is_err()); assert!(commands.try_recv().is_err()); + let events = observer + .request(envelope( + "events", + ControlRequest::ListObservabilityEvents { + page: PageRequest { + cursor: None, + limit: 100, + }, + }, + )) + .await; + let Ok(ControlPayload::ObservabilityEvents(events)) = events.result else { + panic!("observability events") + }; + assert!( + events + .items + .iter() + .any(|event| event.family == "control_command") + ); + assert!(events.items.iter().all(|event| { + !serde_json::to_string(event) + .expect("event JSON") + .contains("operator-token") + })); behavior.shutdown().await.expect("shutdown behavior"); } diff --git a/crates/metacrate-grid-agent/src/conversation.rs b/crates/metacrate-grid-agent/src/conversation.rs index c0735de..d105aa5 100644 --- a/crates/metacrate-grid-agent/src/conversation.rs +++ b/crates/metacrate-grid-agent/src/conversation.rs @@ -232,6 +232,7 @@ impl MemoryRecord { /// Stable, content-free reasons suitable for audit and observability. #[derive(Clone, Copy, Debug, Eq, PartialEq)] pub enum MemoryReason { + SessionCreated, PublicExpired, DirectImExpired, TurnCompacted, @@ -542,6 +543,10 @@ impl ConversationStore { tool_results: 0, }; state.sessions.insert(key, session); + if let Some(session) = state.sessions.get(&key) { + let event = event_for(MemoryReason::SessionCreated, session); + push_event(&mut state.events, event); + } } let effective_wall = state.sessions.get(&key).map_or(wall_now, |session| { wall_now.max(session.last_active_unix_millis) diff --git a/crates/metacrate-grid-agent/src/interaction.rs b/crates/metacrate-grid-agent/src/interaction.rs index 2458b35..1beea77 100644 --- a/crates/metacrate-grid-agent/src/interaction.rs +++ b/crates/metacrate-grid-agent/src/interaction.rs @@ -374,8 +374,20 @@ pub enum DeliveryOutcome { PolicyDenied, } +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum InferenceOutcome { + Completed, + Cancelled, + TimedOut, + Failed, + PolicyRejected, +} + #[derive(Clone, Debug, Eq, PartialEq)] pub enum InteractionObservation { + SessionMemory { + event: crate::conversation::MemoryEvent, + }, Suppressed { delivery_id: BoundedText, reason: SuppressionReason, @@ -390,6 +402,17 @@ pub enum InteractionObservation { delivery_id: BoundedText, avatar_id: UUID, }, + InferenceStarted { + delivery_id: BoundedText, + session_id: BoundedText, + prompt_messages: usize, + }, + InferenceFinished { + delivery_id: BoundedText, + session_id: BoundedText, + outcome: InferenceOutcome, + duration_millis: u64, + }, Delivery { delivery_id: BoundedText, session_id: BoundedText, @@ -1044,6 +1067,12 @@ async fn process_batch( (InteractionChannel::DirectIm, false) => InteractionOrigin::UnprivilegedIm, }; let intent = classify_intent(trigger.channel, authorized, &body); + for event in update.events.iter().cloned() { + observe( + &observations, + InteractionObservation::SessionMemory { event }, + ); + } observe( &observations, InteractionObservation::IntentRouted { @@ -1104,12 +1133,38 @@ async fn process_batch( intent, messages, }; + observe( + &observations, + InteractionObservation::InferenceStarted { + delivery_id: delivery_id.clone(), + session_id: update.session_id.clone(), + prompt_messages: request.messages.len(), + }, + ); + let inference_started = Instant::now(); let model = tokio::select! { () = cancellation.cancelled() => Err(InteractionModelError::Cancelled), result = tokio::time::timeout(settings.model_timeout, responder.respond(request, cancellation.clone())) => { result.unwrap_or(Err(InteractionModelError::Timeout)) } }; + let inference_outcome = match &model { + Ok(_) => InferenceOutcome::Completed, + Err(InteractionModelError::Cancelled) => InferenceOutcome::Cancelled, + Err(InteractionModelError::Timeout) => InferenceOutcome::TimedOut, + Err(InteractionModelError::Failed) => InferenceOutcome::Failed, + Err(InteractionModelError::PolicyRejected) => InferenceOutcome::PolicyRejected, + }; + observe( + &observations, + InteractionObservation::InferenceFinished { + delivery_id: delivery_id.clone(), + session_id: update.session_id.clone(), + outcome: inference_outcome, + duration_millis: u64::try_from(inference_started.elapsed().as_millis()) + .unwrap_or(u64::MAX), + }, + ); match model { Ok(response) => { let Some(safe) = safe_visible_response(response.as_str(), settings.max_response_bytes) diff --git a/crates/metacrate-grid-agent/src/interaction_tests.rs b/crates/metacrate-grid-agent/src/interaction_tests.rs index f8427a6..4a259b0 100644 --- a/crates/metacrate-grid-agent/src/interaction_tests.rs +++ b/crates/metacrate-grid-agent/src/interaction_tests.rs @@ -237,6 +237,9 @@ async fn observation_for( | InteractionObservation::Delivery { delivery_id: id, .. } => id.as_str() == delivery_id, + InteractionObservation::SessionMemory { .. } + | InteractionObservation::InferenceStarted { .. } + | InteractionObservation::InferenceFinished { .. } => false, }; if matches { return observation; diff --git a/crates/metacrate-grid-agent/src/lib.rs b/crates/metacrate-grid-agent/src/lib.rs index 2261300..690b1c8 100644 --- a/crates/metacrate-grid-agent/src/lib.rs +++ b/crates/metacrate-grid-agent/src/lib.rs @@ -12,6 +12,7 @@ pub mod control_runtime; pub mod conversation; pub mod interaction; pub mod llm; +pub mod observability; pub mod perception; pub mod policy; pub mod service; @@ -30,6 +31,8 @@ mod conversation_tests; #[cfg(test)] mod interaction_tests; #[cfg(test)] +mod observability_tests; +#[cfg(test)] mod perception_tests; #[cfg(test)] mod policy_tests; @@ -74,17 +77,25 @@ pub use conversation::{ }; pub use interaction::{ DeliveryFuture, DeliveryOutcome, ImDialogKind, InboundInteraction, InboundSource, - InteractionChannel, InteractionCoordinator, InteractionDeliveryError, InteractionError, - InteractionHandle, InteractionIngress, InteractionIntent, InteractionModelError, - InteractionObservation, InteractionOrigin, InteractionPacingError, InteractionResponder, - InteractionSettings, InteractionSink, OutboundInteraction, PacerFuture, PolicyLlmResponder, - ResponderFuture, ResponsePacer, ResponseRequest, SuppressionReason, VisibleResponse, - split_utf8, + InferenceOutcome, InteractionChannel, InteractionCoordinator, InteractionDeliveryError, + InteractionError, InteractionHandle, InteractionIngress, InteractionIntent, + InteractionModelError, InteractionObservation, InteractionOrigin, InteractionPacingError, + InteractionResponder, InteractionSettings, InteractionSink, OutboundInteraction, PacerFuture, + PolicyLlmResponder, ResponderFuture, ResponsePacer, ResponseRequest, SuppressionReason, + VisibleResponse, split_utf8, }; pub use llm::{ Completion, CompletionMessage, ContentPart, ImageDetail, LlmClient, LlmError, LlmTransportLimits, ToolDefinition, ToolSchema, Usage, }; +pub use observability::{ + AgentMetrics, CorrelationIds, DiagnosticDirection, EventDraft, EventFamily, EventOrigin, + EventSeverity, EventSubscription, JournalConfig, LatencyMetrics, MetricsSnapshot, + OBSERVABILITY_SCHEMA_VERSION, Observability, ObservabilityError, ObservabilityLimits, + ObservabilityRuntime, OutcomeCode, OutcomeMetrics, ReplayAction, ReplayState, + SensitiveDiagnosticPart, StructuredEvent, UnifiedPolicyAudit, event_json, + pseudonymous_identifier, replay_journal, +}; pub use perception::{ AGENT_STATE_TOOL, AgentSnapshot, EnvironmentSnapshot, INVENTORY_SEARCH_TOOL, InventoryMetadata, LOCATION_TOOL, LandmarkMetadata, NEARBY_AVATARS_TOOL, ObservedAvatar, ObservedObject, @@ -105,9 +116,9 @@ pub use policy::{ pub use service::{AgentService, ServiceError, ServiceHandle, ServiceState}; pub use session::{ GridSession, GridSessionBackend, ReconnectPolicy, SessionControl, SessionFailure, - SessionFailureKind, SessionFuture, SessionObservation, SessionSignal, SessionState, - SessionStatus, SessionSupervisor, SessionSupervisorError, SessionSupervisorHandle, SessionWork, - WorkDisposition, WorkKind, + SessionFailureKind, SessionFuture, SessionObservation, SessionReason, SessionSignal, + SessionState, SessionStatus, SessionSupervisor, SessionSupervisorError, + SessionSupervisorHandle, SessionWork, WorkDisposition, WorkKind, }; pub use tool_loop::{ HistorySummarizer, SessionGeneration, ToolExecution, ToolExecutor, ToolFuture, ToolLoop, diff --git a/crates/metacrate-grid-agent/src/main.rs b/crates/metacrate-grid-agent/src/main.rs index 0c872cb..dc75095 100644 --- a/crates/metacrate-grid-agent/src/main.rs +++ b/crates/metacrate-grid-agent/src/main.rs @@ -7,7 +7,7 @@ use std::path::PathBuf; #[cfg(feature = "live-grid")] use std::sync::Arc; #[cfg(feature = "live-grid")] -use std::time::{SystemTime, UNIX_EPOCH}; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; const MAX_ARGUMENTS: usize = 8; @@ -184,22 +184,28 @@ async fn run_live( let readiness = tokio::time::timeout(config.timeouts.startup, async { loop { match handle.next_observation().await { - Some(SessionObservation::Transition { status, .. }) if status.agent_ready => { + Some(event @ SessionObservation::Transition { status, .. }) + if status.agent_ready => + { + record_session_observation(&live.observability, &event); return Ok::<(), CliError>(()); } - Some(SessionObservation::Transition { - status: - metacrate_grid_agent::SessionStatus { - state: SessionState::AuthenticationBlocked, - .. - }, - .. - }) => { + Some( + event @ SessionObservation::Transition { + status: + metacrate_grid_agent::SessionStatus { + state: SessionState::AuthenticationBlocked, + .. + }, + .. + }, + ) => { + record_session_observation(&live.observability, &event); return Err(CliError( "grid authentication/configuration requires operator action".into(), )); } - Some(_) => {} + Some(event) => record_session_observation(&live.observability, &event), None => { return Err(CliError( "session supervisor stopped before readiness".into(), @@ -239,6 +245,7 @@ async fn run_live( live.behavior.ingress(), config.limits.control_queue, )?; + control_target.attach_observability(live.observability.clone()); control_target.update_session(handle.status()); let erased_target: Arc = control_target.clone(); let (control_plane, _integrated_client, control_server) = match config.mode { @@ -292,6 +299,7 @@ async fn run_live( } event = handle.next_observation() => { let Some(event) = event else { break; }; + record_session_observation(&live.observability, &event); if let SessionObservation::Transition { status, reason, retry_in } = event { control_target.update_session(status); control_plane.publish(ControlEventKind::StateChanged { @@ -309,14 +317,17 @@ async fn run_live( } event = live.interaction.next_observation() => { let Some(event) = event else { break; }; + record_interaction_observation(&live.observability, &event); println!("grid interaction event={event:?}"); } event = live.perception_observations.recv() => { let Some(event) = event else { break; }; + record_perception_observation(&live.observability, &event); println!("grid perception event={event:?}"); } event = live.behavior.next_observation() => { let Some(event) = event else { break; }; + record_behavior_observation(&live.observability, &event); if let BehaviorObservation::Transition { to, .. } = &event { control_target.update_behavior(*to); control_plane.publish(ControlEventKind::StateChanged { @@ -358,6 +369,14 @@ async fn run_live( } } } + let shutdown_event = metacrate_grid_agent::EventDraft::new( + metacrate_grid_agent::EventFamily::Shutdown, + metacrate_grid_agent::EventSeverity::Info, + "service", + metacrate_grid_agent::EventOrigin::Service, + )? + .result_code("requested")?; + let _ = live.observability.record(shutdown_event); control_target.mark_stopping(); let session_result = handle.shutdown().await; let interaction_result = live.interaction.shutdown().await; @@ -391,6 +410,7 @@ struct LiveInteractions { conversations: Arc, policy: Arc, audit: Arc, + observability: Arc, } #[cfg(feature = "live-grid")] @@ -402,8 +422,9 @@ fn start_live_interactions( use metacrate_grid_agent::{ AuthorizedBackendRouter, AuthorizedToolBackend, BehaviorBackend, BehaviorController, ConversationStore, InteractionCoordinator, LlmClient, LlmTransportLimits, - MemoryPolicyAudit, PerceptionBackend, PolicyGateway, PolicyLimits, PolicyLlmResponder, - ToolLoopLimits, behavior_policy_tools, perception_policy_tools, + MemoryPolicyAudit, Observability, ObservabilityLimits, PerceptionBackend, PolicyAuditSink, + PolicyGateway, PolicyLimits, PolicyLlmResponder, ToolLoopLimits, UnifiedPolicyAudit, + behavior_policy_tools, perception_policy_tools, }; let transport_limits = LlmTransportLimits { @@ -435,6 +456,16 @@ fn start_live_interactions( let behavior_ingress = behavior.ingress(); let client = Arc::new(LlmClient::new(config.llm.clone(), transport_limits)?); let audit = Arc::new(MemoryPolicyAudit::new(config.limits.observable_queue)?); + let observability = Observability::memory(ObservabilityLimits { + ring_events: config.limits.observable_queue, + subscriber_queue: config.limits.observable_queue.min(512), + journal_queue: config.limits.observable_queue, + ..ObservabilityLimits::default() + })?; + let audit_sink: Arc = Arc::new(UnifiedPolicyAudit::new( + audit.clone(), + observability.clone(), + )); let mut tools = perception_policy_tools()?; tools.extend(behavior_policy_tools(&config.behavior)?); let routes = tools @@ -453,7 +484,7 @@ fn start_live_interactions( config.authorized_avatar_uuids.clone(), tools, PolicyLimits::default(), - audit.clone(), + audit_sink, )?); let loop_limits = ToolLoopLimits { max_tool_calls_per_turn: config.limits.max_tool_calls, @@ -500,5 +531,497 @@ fn start_live_interactions( conversations: conversation, policy: gateway, audit, + observability, }) } + +#[cfg(feature = "live-grid")] +fn record_session_observation( + observer: &metacrate_grid_agent::Observability, + observation: &metacrate_grid_agent::SessionObservation, +) { + use metacrate_grid_agent::{ + EventDraft, EventFamily, EventOrigin, EventSeverity, SessionObservation, + }; + let SessionObservation::Transition { + status, + reason, + retry_in, + } = observation + else { + return; + }; + observer.metrics().set_ready(status.agent_ready); + if matches!( + reason, + metacrate_grid_agent::SessionReason::TransientTransport + | metacrate_grid_agent::SessionReason::Maintenance + | metacrate_grid_agent::SessionReason::Kicked + | metacrate_grid_agent::SessionReason::SimulatorDisconnected + | metacrate_grid_agent::SessionReason::ServerFailure + | metacrate_grid_agent::SessionReason::OperatorForceReconnect + ) { + observer.metrics().record_reconnect(); + } + let event = EventDraft::new( + EventFamily::LifecycleTransition, + if status.agent_ready { + EventSeverity::Info + } else { + EventSeverity::Warning + }, + "session", + EventOrigin::Grid, + ) + .and_then(|event| event.reason_code(session_reason_code(*reason))) + .and_then(|event| { + event + .retry_count(status.consecutive_failures) + .code_field("to", status.state.as_str()) + }) + .and_then(|event| event.field("generation", serde_json::Value::from(status.generation))) + .and_then(|event| event.field("ready", serde_json::Value::from(status.agent_ready))) + .and_then(|event| { + event.field( + "transport_connected", + serde_json::Value::from(status.transport_connected), + ) + }) + .and_then(|event| { + event.field( + "retry_millis", + serde_json::Value::from(retry_in.map_or(0, |duration| { + u64::try_from(duration.as_millis()).unwrap_or(u64::MAX) + })), + ) + }); + if let Ok(event) = event { + let _ = observer.record(event); + } +} + +#[cfg(feature = "live-grid")] +#[allow(clippy::too_many_lines)] +fn record_interaction_observation( + observer: &metacrate_grid_agent::Observability, + observation: &metacrate_grid_agent::InteractionObservation, +) { + use metacrate_grid_agent::{ + CorrelationIds, EventDraft, EventFamily, EventOrigin, EventSeverity, InferenceOutcome, + InteractionObservation, MemoryReason, OutcomeCode, + }; + let event = match observation { + InteractionObservation::SessionMemory { event } => { + let (family, state) = match event.reason { + MemoryReason::SessionCreated => (EventFamily::SessionCreated, "created"), + MemoryReason::PublicExpired + | MemoryReason::DirectImExpired + | MemoryReason::SessionLimitEviction + | MemoryReason::TotalByteEviction + | MemoryReason::OperatorDeleted + | MemoryReason::OperatorExpired => (EventFamily::SessionExpired, "expired"), + MemoryReason::TurnCompacted + | MemoryReason::ByteCompacted + | MemoryReason::ToolResultCompacted => { + (EventFamily::LifecycleTransition, "compacted") + } + MemoryReason::CorruptSnapshotQuarantined + | MemoryReason::UnsupportedSnapshotQuarantined + | MemoryReason::SnapshotTooLargeQuarantined + | MemoryReason::PermissionsNotVerified => { + (EventFamily::LifecycleTransition, "quarantined") + } + }; + EventDraft::new( + family, + EventSeverity::Info, + "conversation", + EventOrigin::Service, + ) + .and_then(|draft| { + draft.correlation(CorrelationIds { + avatar_id: event.avatar_id.map(|id| id.to_string()), + session_id: event.session_id.as_ref().map(|id| id.as_str().to_owned()), + ..CorrelationIds::default() + }) + }) + .and_then(|draft| draft.reason_code(memory_reason_code(event.reason))) + .and_then(|draft| draft.code_field("state", state)) + } + InteractionObservation::Suppressed { + delivery_id, + reason, + } => EventDraft::new( + EventFamily::InboundMessage, + EventSeverity::Info, + "interaction", + EventOrigin::Grid, + ) + .and_then(|event| { + event.correlation(CorrelationIds { + request_id: Some(delivery_id.as_str().to_owned()), + ..CorrelationIds::default() + }) + }) + .and_then(|event| event.result_code("suppressed")) + .and_then(|event| event.reason_code(suppression_reason_code(*reason))) + .and_then(|event| event.redacted("message_content")), + InteractionObservation::IntentRouted { + delivery_id, + session_id, + origin, + intent, + } => EventDraft::new( + EventFamily::InboundMessage, + EventSeverity::Info, + "interaction", + EventOrigin::Grid, + ) + .and_then(|event| { + event.correlation(CorrelationIds { + session_id: Some(session_id.as_str().to_owned()), + request_id: Some(delivery_id.as_str().to_owned()), + ..CorrelationIds::default() + }) + }) + .and_then(|event| event.result_code("routed")) + .and_then(|event| event.code_field("origin", interaction_origin_code(*origin))) + .and_then(|event| event.code_field("intent", interaction_intent_code(*intent))) + .and_then(|event| event.redacted("message_content")), + InteractionObservation::AttentionRequested { + delivery_id, + avatar_id, + } => EventDraft::new( + EventFamily::ModelActionSummary, + EventSeverity::Info, + "behavior", + EventOrigin::Service, + ) + .and_then(|event| { + event.correlation(CorrelationIds { + avatar_id: Some(avatar_id.to_string()), + request_id: Some(delivery_id.as_str().to_owned()), + ..CorrelationIds::default() + }) + }) + .and_then(|event| event.code_field("summary", "face_speaker")), + InteractionObservation::InferenceStarted { + delivery_id, + session_id, + prompt_messages, + } => EventDraft::new( + EventFamily::InferenceRequest, + EventSeverity::Info, + "llm", + EventOrigin::Service, + ) + .and_then(|event| { + event.correlation(CorrelationIds { + session_id: Some(session_id.as_str().to_owned()), + request_id: Some(delivery_id.as_str().to_owned()), + ..CorrelationIds::default() + }) + }) + .and_then(|event| event.field("prompt_messages", serde_json::Value::from(*prompt_messages))) + .and_then(|event| event.redacted("prompt_content")), + InteractionObservation::InferenceFinished { + delivery_id, + session_id, + outcome, + duration_millis, + } => { + observer + .metrics() + .record_inference_latency(Duration::from_millis(*duration_millis)); + let (severity, result, metric) = match outcome { + InferenceOutcome::Completed => { + (EventSeverity::Info, "completed", OutcomeCode::Completed) + } + InferenceOutcome::Cancelled => { + (EventSeverity::Warning, "cancelled", OutcomeCode::Cancelled) + } + InferenceOutcome::TimedOut => { + (EventSeverity::Warning, "timed_out", OutcomeCode::Failed) + } + InferenceOutcome::Failed => (EventSeverity::Error, "failed", OutcomeCode::Failed), + InferenceOutcome::PolicyRejected => ( + EventSeverity::Warning, + "policy_rejected", + OutcomeCode::Denied, + ), + }; + observer.metrics().record_inference_outcome(metric); + EventDraft::new( + EventFamily::InferenceResult, + severity, + "llm", + EventOrigin::Model, + ) + .and_then(|event| { + event.correlation(CorrelationIds { + session_id: Some(session_id.as_str().to_owned()), + request_id: Some(delivery_id.as_str().to_owned()), + ..CorrelationIds::default() + }) + }) + .map(|event| event.duration_millis(*duration_millis)) + .and_then(|event| event.result_code(result)) + .and_then(|event| event.redacted("response_content")) + } + InteractionObservation::Delivery { + delivery_id, + session_id, + channel, + outcome, + delivered_parts, + } => EventDraft::new( + EventFamily::OutboundMessage, + EventSeverity::Info, + "interaction", + EventOrigin::Service, + ) + .and_then(|event| { + event.correlation(CorrelationIds { + session_id: Some(session_id.as_str().to_owned()), + request_id: Some(delivery_id.as_str().to_owned()), + ..CorrelationIds::default() + }) + }) + .and_then(|event| event.result_code(delivery_outcome_code(*outcome))) + .and_then(|event| event.code_field("channel", interaction_channel_code(*channel))) + .and_then(|event| event.field("delivered_parts", serde_json::Value::from(*delivered_parts))) + .and_then(|event| event.redacted("message_content")), + }; + if let Ok(event) = event { + let _ = observer.record(event); + } +} + +#[cfg(feature = "live-grid")] +fn record_perception_observation( + observer: &metacrate_grid_agent::Observability, + observation: &metacrate_grid_agent::PerceptionObservation, +) { + use metacrate_grid_agent::{ + CorrelationIds, EventDraft, EventFamily, EventOrigin, EventSeverity, PerceptionOutcome, + }; + observer + .metrics() + .record_tool_latency(Duration::from_millis(observation.duration_millis)); + let result = match observation.outcome { + PerceptionOutcome::Completed => "completed", + PerceptionOutcome::Rejected => "rejected", + }; + let event = EventDraft::new( + EventFamily::ToolResult, + EventSeverity::Info, + "perception", + EventOrigin::Service, + ) + .and_then(|event| { + event.correlation(CorrelationIds { + action_id: Some(observation.call_id.as_str().to_owned()), + ..CorrelationIds::default() + }) + }) + .map(|event| event.duration_millis(observation.duration_millis)) + .and_then(|event| event.result_code(result)) + .and_then(|event| event.identifier_field("tool", observation.tool.as_str())) + .and_then(|event| { + event.field( + "result_bytes", + serde_json::Value::from(observation.result_bytes), + ) + }) + .and_then(|event| event.field("cache_hit", serde_json::Value::from(observation.cache_hit))) + .and_then(|event| event.redacted("tool_result")); + if let Ok(event) = event { + let _ = observer.record(event); + } +} + +#[cfg(feature = "live-grid")] +fn record_behavior_observation( + observer: &metacrate_grid_agent::Observability, + observation: &metacrate_grid_agent::BehaviorObservation, +) { + use metacrate_grid_agent::{ + BehaviorObservation, CorrelationIds, EventDraft, EventFamily, EventOrigin, EventSeverity, + }; + let event = match observation { + BehaviorObservation::Transition { from, to, .. } => EventDraft::new( + EventFamily::BehaviorTransition, + EventSeverity::Info, + "behavior", + EventOrigin::Service, + ) + .and_then(|event| event.code_field("from", behavior_mode_code(*from))) + .and_then(|event| event.code_field("to", behavior_mode_code(*to))), + BehaviorObservation::Action { + action_id, + generation, + action, + duration_millis, + outcome, + .. + } => EventDraft::new( + EventFamily::ToolResult, + EventSeverity::Info, + "behavior", + EventOrigin::Service, + ) + .and_then(|event| { + event.correlation(CorrelationIds { + action_id: Some(action_id.clone()), + ..CorrelationIds::default() + }) + }) + .map(|event| event.duration_millis(*duration_millis)) + .and_then(|event| event.result_code(behavior_outcome_code(*outcome))) + .and_then(|event| event.identifier_field("tool", action)) + .and_then(|event| { + event.field( + "generation", + serde_json::Value::from(generation.unwrap_or(0)), + ) + }), + }; + if let Ok(event) = event { + let _ = observer.record(event); + } +} + +#[cfg(feature = "live-grid")] +const fn session_reason_code(reason: metacrate_grid_agent::SessionReason) -> &'static str { + use metacrate_grid_agent::SessionReason; + match reason { + SessionReason::Startup => "startup", + SessionReason::LoginSucceeded => "login_succeeded", + SessionReason::AgentReady => "agent_ready", + SessionReason::ReadinessLost => "readiness_lost", + SessionReason::TransientTransport => "transient_transport", + SessionReason::Maintenance => "maintenance", + SessionReason::Kicked => "kicked", + SessionReason::SimulatorDisconnected => "simulator_disconnected", + SessionReason::ServerFailure => "server_failure", + SessionReason::InvalidCredentials => "invalid_credentials", + SessionReason::InvalidConfiguration => "invalid_configuration", + SessionReason::StableSessionReset => "stable_session_reset", + SessionReason::OperatorPause => "operator_pause", + SessionReason::OperatorResume => "operator_resume", + SessionReason::OperatorForceReconnect => "operator_force_reconnect", + SessionReason::OperatorLogout => "operator_logout", + SessionReason::ShutdownRequested => "shutdown_requested", + SessionReason::ShutdownComplete => "shutdown_complete", + } +} + +#[cfg(feature = "live-grid")] +const fn memory_reason_code(reason: metacrate_grid_agent::MemoryReason) -> &'static str { + use metacrate_grid_agent::MemoryReason; + match reason { + MemoryReason::SessionCreated => "session_created", + MemoryReason::PublicExpired => "public_expired", + MemoryReason::DirectImExpired => "direct_im_expired", + MemoryReason::TurnCompacted => "turn_compacted", + MemoryReason::ByteCompacted => "byte_compacted", + MemoryReason::ToolResultCompacted => "tool_result_compacted", + MemoryReason::SessionLimitEviction => "session_limit_eviction", + MemoryReason::TotalByteEviction => "total_byte_eviction", + MemoryReason::OperatorDeleted => "operator_deleted", + MemoryReason::OperatorExpired => "operator_expired", + MemoryReason::CorruptSnapshotQuarantined => "corrupt_snapshot_quarantined", + MemoryReason::UnsupportedSnapshotQuarantined => "unsupported_snapshot_quarantined", + MemoryReason::SnapshotTooLargeQuarantined => "snapshot_too_large_quarantined", + MemoryReason::PermissionsNotVerified => "permissions_not_verified", + } +} + +#[cfg(feature = "live-grid")] +const fn suppression_reason_code(reason: metacrate_grid_agent::SuppressionReason) -> &'static str { + use metacrate_grid_agent::SuppressionReason; + match reason { + SuppressionReason::SelfEcho => "self_echo", + SuppressionReason::Duplicate => "duplicate", + SuppressionReason::Muted => "muted", + SuppressionReason::UnsupportedSource => "unsupported_source", + SuppressionReason::UnsupportedDialog => "unsupported_dialog", + SuppressionReason::AmbientPublicChat => "ambient_public_chat", + SuppressionReason::SenderQueueFull => "sender_queue_full", + SuppressionReason::SenderLimit => "sender_limit", + SuppressionReason::Disconnected => "disconnected", + SuppressionReason::UnsafeResponse => "unsafe_response", + SuppressionReason::Cancelled => "cancelled", + } +} + +#[cfg(feature = "live-grid")] +const fn interaction_origin_code(origin: metacrate_grid_agent::InteractionOrigin) -> &'static str { + use metacrate_grid_agent::InteractionOrigin; + match origin { + InteractionOrigin::Public => "public", + InteractionOrigin::UnprivilegedIm => "unprivileged_im", + InteractionOrigin::AuthorizedIm => "authorized_im", + } +} + +#[cfg(feature = "live-grid")] +const fn interaction_intent_code(intent: metacrate_grid_agent::InteractionIntent) -> &'static str { + use metacrate_grid_agent::InteractionIntent; + match intent { + InteractionIntent::Informational => "informational", + InteractionIntent::PublicCommandDenied => "public_command_denied", + InteractionIntent::PolicyGatedCommand => "policy_gated_command", + InteractionIntent::PolicyGatedLslRequest => "policy_gated_lsl_request", + } +} + +#[cfg(feature = "live-grid")] +const fn delivery_outcome_code(outcome: metacrate_grid_agent::DeliveryOutcome) -> &'static str { + use metacrate_grid_agent::DeliveryOutcome; + match outcome { + DeliveryOutcome::Succeeded => "succeeded", + DeliveryOutcome::Failed => "failed", + DeliveryOutcome::TimedOut => "timed_out", + DeliveryOutcome::PolicyDenied => "policy_denied", + } +} + +#[cfg(feature = "live-grid")] +const fn interaction_channel_code( + channel: metacrate_grid_agent::InteractionChannel, +) -> &'static str { + use metacrate_grid_agent::InteractionChannel; + match channel { + InteractionChannel::PublicChat => "public_chat", + InteractionChannel::DirectIm => "direct_im", + } +} + +#[cfg(feature = "live-grid")] +const fn behavior_mode_code(mode: metacrate_grid_agent::BehaviorMode) -> &'static str { + use metacrate_grid_agent::BehaviorMode; + match mode { + BehaviorMode::Offline => "offline", + BehaviorMode::Settling => "settling", + BehaviorMode::Available => "available", + BehaviorMode::Engaged => "engaged", + BehaviorMode::Executing => "executing", + BehaviorMode::Roaming => "roaming", + BehaviorMode::Paused => "paused", + BehaviorMode::Recovering => "recovering", + } +} + +#[cfg(feature = "live-grid")] +const fn behavior_outcome_code(outcome: metacrate_grid_agent::BehaviorOutcome) -> &'static str { + use metacrate_grid_agent::BehaviorOutcome; + match outcome { + BehaviorOutcome::Completed => "completed", + BehaviorOutcome::Cancelled => "cancelled", + BehaviorOutcome::TimedOut => "timed_out", + BehaviorOutcome::Stuck => "stuck", + BehaviorOutcome::Rejected => "rejected", + BehaviorOutcome::Preempted => "preempted", + } +} diff --git a/crates/metacrate-grid-agent/src/observability.rs b/crates/metacrate-grid-agent/src/observability.rs new file mode 100644 index 0000000..043a33d --- /dev/null +++ b/crates/metacrate-grid-agent/src/observability.rs @@ -0,0 +1,1679 @@ +//! Bounded, redacted observability, journal, metrics, and deterministic replay. +//! +//! Events describe externally observable inputs, decisions, actions, and outcomes. +//! They deliberately contain no prompt bodies, private message bodies, credentials, +//! capability URLs, inventory payloads, or hidden model reasoning. Ordering is total +//! inside one recorder; concurrently produced events reflect recorder admission order. + +#![allow(clippy::missing_errors_doc)] + +use serde::{Deserialize, Serialize}; +use serde_json::{Map, Value}; +use sha2::{Digest, Sha256}; +use std::collections::{BTreeMap, VecDeque}; +use std::error::Error; +use std::fmt; +use std::fs::{self, File, OpenOptions}; +use std::io::{BufRead, BufReader, Read, Seek, SeekFrom, Write}; +use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; +use std::sync::{Arc, Mutex, MutexGuard, Weak}; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; +use tokio::sync::{mpsc, oneshot}; +use tokio::task::JoinHandle; + +use crate::policy::{ + MemoryPolicyAudit, PolicyAuditError, PolicyAuditRecord, PolicyAuditSink, PolicyDisposition, + PolicyFinalOutcome, +}; + +pub const OBSERVABILITY_SCHEMA_VERSION: u16 = 1; +const MAX_EVENT_BYTES_HARD: usize = 16 * 1024; +const MAX_RING_EVENTS_HARD: usize = 65_536; +const MAX_SUBSCRIBERS_HARD: usize = 1_024; +const MAX_FIELD_COUNT: usize = 32; +const MAX_CODE_BYTES: usize = 96; +const MAX_ID_BYTES: usize = 128; +const MAX_DIAGNOSTIC_ENTRIES_HARD: usize = 4_096; +const JOURNAL_EVENT_ID_RESERVATION: u64 = 1 << 32; + +#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum EventSeverity { + Trace, + Info, + Warning, + Error, +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum EventOrigin { + Grid, + Model, + Policy, + Scheduler, + Control, + Operator, + Service, +} + +/// Stable event families. The wire schema stores their snake-case code so +/// unknown future families remain readable by older replay implementations. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum EventFamily { + LifecycleTransition, + InboundMessage, + OutboundMessage, + SessionCreated, + SessionExpired, + InferenceRequest, + InferenceResult, + ModelActionSummary, + ToolProposed, + PolicyDecision, + ApprovalDecision, + ToolStarted, + ToolProgress, + ToolResult, + BehaviorTransition, + ScheduledJob, + ControlCommand, + Shutdown, + DiagnosticEnvelope, +} + +impl EventFamily { + #[must_use] + pub const fn as_str(self) -> &'static str { + match self { + Self::LifecycleTransition => "lifecycle_transition", + Self::InboundMessage => "inbound_message", + Self::OutboundMessage => "outbound_message", + Self::SessionCreated => "session_created", + Self::SessionExpired => "session_expired", + Self::InferenceRequest => "inference_request", + Self::InferenceResult => "inference_result", + Self::ModelActionSummary => "model_action_summary", + Self::ToolProposed => "tool_proposed", + Self::PolicyDecision => "policy_decision", + Self::ApprovalDecision => "approval_decision", + Self::ToolStarted => "tool_started", + Self::ToolProgress => "tool_progress", + Self::ToolResult => "tool_result", + Self::BehaviorTransition => "behavior_transition", + Self::ScheduledJob => "scheduled_job", + Self::ControlCommand => "control_command", + Self::Shutdown => "shutdown", + Self::DiagnosticEnvelope => "diagnostic_envelope", + } + } +} + +#[derive(Clone, Default, Eq, PartialEq, Serialize, Deserialize)] +pub struct CorrelationIds { + #[serde(skip_serializing_if = "Option::is_none")] + pub avatar_id: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub session_id: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub request_id: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub action_id: Option, +} + +impl CorrelationIds { + fn validate(&self) -> bool { + [ + self.avatar_id.as_deref(), + self.session_id.as_deref(), + self.request_id.as_deref(), + self.action_id.as_deref(), + ] + .into_iter() + .flatten() + .all(valid_identifier) + } + + fn pseudonymized(mut self) -> Self { + self.avatar_id = self.avatar_id.map(|value| pseudonymous_identifier(&value)); + self.session_id = self.session_id.map(|value| pseudonymous_identifier(&value)); + self.request_id = self.request_id.map(|value| pseudonymous_identifier(&value)); + self.action_id = self.action_id.map(|value| pseudonymous_identifier(&value)); + self + } +} + +impl fmt::Debug for CorrelationIds { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("CorrelationIds") + .field("avatar", &self.avatar_id.is_some()) + .field("session", &self.session_id.is_some()) + .field("request", &self.request_id.is_some()) + .field("action", &self.action_id.is_some()) + .finish() + } +} + +/// One forward-compatible JSONL event. Unknown keys inside `fields` survive a +/// deserialize/serialize cycle and unknown `family` values remain replayable. +#[derive(Clone, PartialEq, Serialize, Deserialize)] +pub struct StructuredEvent { + pub schema_version: u16, + pub event_id: u64, + pub unix_millis: u64, + pub severity: EventSeverity, + pub component: String, + pub family: String, + pub correlation: CorrelationIds, + pub origin: EventOrigin, + #[serde(skip_serializing_if = "Option::is_none")] + pub duration_millis: Option, + pub retry_count: u32, + #[serde(skip_serializing_if = "Option::is_none")] + pub result_code: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub reason_code: Option, + pub redaction_flags: Vec, + #[serde(default)] + pub fields: BTreeMap, + /// Unknown top-level properties retained for schema evolution. + #[serde(flatten)] + pub extensions: BTreeMap, +} + +impl fmt::Debug for StructuredEvent { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("StructuredEvent") + .field("schema_version", &self.schema_version) + .field("event_id", &self.event_id) + .field("unix_millis", &self.unix_millis) + .field("severity", &self.severity) + .field("component_bytes", &self.component.len()) + .field("family_bytes", &self.family.len()) + .field("correlation", &self.correlation) + .field("origin", &self.origin) + .field("duration_millis", &self.duration_millis) + .field("retry_count", &self.retry_count) + .field("has_result_code", &self.result_code.is_some()) + .field("has_reason_code", &self.reason_code.is_some()) + .field("redaction_count", &self.redaction_flags.len()) + .field("field_count", &self.fields.len()) + .field("extension_count", &self.extensions.len()) + .finish_non_exhaustive() + } +} + +/// Validated event before the recorder assigns identity and time. +#[derive(Clone, Debug)] +pub struct EventDraft { + severity: EventSeverity, + component: String, + family: EventFamily, + correlation: CorrelationIds, + origin: EventOrigin, + duration_millis: Option, + retry_count: u32, + result_code: Option, + reason_code: Option, + redaction_flags: Vec, + fields: BTreeMap, +} + +impl EventDraft { + pub fn new( + family: EventFamily, + severity: EventSeverity, + component: &'static str, + origin: EventOrigin, + ) -> Result { + if !valid_code(component) { + return Err(ObservabilityError::InvalidEvent("invalid component")); + } + Ok(Self { + severity, + component: component.to_owned(), + family, + correlation: CorrelationIds::default(), + origin, + duration_millis: None, + retry_count: 0, + result_code: None, + reason_code: None, + redaction_flags: Vec::new(), + fields: BTreeMap::new(), + }) + } + + pub fn correlation(mut self, correlation: CorrelationIds) -> Result { + if !correlation.validate() { + return Err(ObservabilityError::InvalidEvent( + "invalid correlation identifier", + )); + } + self.correlation = correlation.pseudonymized(); + Ok(self) + } + + #[must_use] + pub const fn duration_millis(mut self, duration_millis: u64) -> Self { + self.duration_millis = Some(duration_millis); + self + } + + #[must_use] + pub const fn retry_count(mut self, retry_count: u32) -> Self { + self.retry_count = retry_count; + self + } + + pub fn result_code(mut self, code: &'static str) -> Result { + self.result_code = Some(checked_code(code.to_owned())?); + Ok(self) + } + + pub fn reason_code(mut self, code: &'static str) -> Result { + self.reason_code = Some(checked_code(code.to_owned())?); + Ok(self) + } + + pub fn redacted(mut self, flag: &'static str) -> Result { + let flag = checked_code(flag.to_owned())?; + if !self.redaction_flags.contains(&flag) { + self.redaction_flags.push(flag); + } + Ok(self) + } + + /// Adds a non-sensitive bounded scalar. Free-form text and nested payloads + /// are intentionally rejected at this boundary. + pub fn field(mut self, name: &'static str, value: Value) -> Result { + let name = checked_code(name.to_owned())?; + if self.fields.len() == MAX_FIELD_COUNT + || !matches!(value, Value::Null | Value::Bool(_) | Value::Number(_)) + { + return Err(ObservabilityError::InvalidEvent("unsafe event field")); + } + self.fields.insert(name, value); + Ok(self) + } + + /// Adds a readable, compile-time stable enum/state code. + pub fn code_field( + mut self, + name: &'static str, + value: &'static str, + ) -> Result { + let name = checked_code(name.to_owned())?; + let value = checked_code(value.to_owned())?; + if self.fields.len() == MAX_FIELD_COUNT { + return Err(ObservabilityError::InvalidEvent("too many event fields")); + } + self.fields.insert(name, Value::String(value)); + Ok(self) + } + + /// Adds a stable SHA-256 pseudonym for a runtime identifier. + pub fn identifier_field( + mut self, + name: &'static str, + value: &str, + ) -> Result { + let name = checked_code(name.to_owned())?; + if value.is_empty() || value.len() > 16 * 1024 || self.fields.len() == MAX_FIELD_COUNT { + return Err(ObservabilityError::InvalidEvent("invalid event identifier")); + } + self.fields + .insert(name, Value::String(pseudonymous_identifier(value))); + Ok(self) + } +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub struct ObservabilityLimits { + pub ring_events: usize, + pub subscriber_queue: usize, + pub max_subscribers: usize, + pub journal_queue: usize, + pub max_event_bytes: usize, + pub max_diagnostic_entries: usize, +} + +impl Default for ObservabilityLimits { + fn default() -> Self { + Self { + ring_events: 2_048, + subscriber_queue: 256, + max_subscribers: 64, + journal_queue: 1_024, + max_event_bytes: 8 * 1024, + max_diagnostic_entries: 128, + } + } +} + +impl ObservabilityLimits { + #[must_use] + pub const fn is_valid(self) -> bool { + self.ring_events > 0 + && self.ring_events <= MAX_RING_EVENTS_HARD + && self.subscriber_queue > 0 + && self.subscriber_queue <= 8_192 + && self.max_subscribers > 0 + && self.max_subscribers <= MAX_SUBSCRIBERS_HARD + && self.journal_queue > 0 + && self.journal_queue <= 16_384 + && self.max_event_bytes >= 512 + && self.max_event_bytes <= MAX_EVENT_BYTES_HARD + && self.max_diagnostic_entries <= MAX_DIAGNOSTIC_ENTRIES_HARD + } +} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct JournalConfig { + pub directory: PathBuf, + pub max_segment_bytes: u64, + pub max_segments: usize, + pub max_total_bytes: u64, + pub sync_each_record: bool, +} + +impl JournalConfig { + fn validate(&self, max_event_bytes: usize) -> Result<(), ObservabilityError> { + let event = u64::try_from(max_event_bytes).unwrap_or(u64::MAX); + if self.max_segment_bytes < event + || self.max_segment_bytes > 256 * 1024 * 1024 + || self.max_segments == 0 + || self.max_segments > 1_024 + || self.max_total_bytes < self.max_segment_bytes + || self.max_total_bytes > 16 * 1024 * 1024 * 1024 + { + return Err(ObservabilityError::UnsafeLimits); + } + Ok(()) + } +} + +#[derive(Debug)] +pub enum ObservabilityError { + UnsafeLimits, + InvalidEvent(&'static str), + Io(std::io::Error), + Json(serde_json::Error), + JournalClosed, + CorruptJournal { path: PathBuf, line: usize }, +} + +impl fmt::Display for ObservabilityError { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::UnsafeLimits => formatter.write_str("unsafe observability limits"), + Self::InvalidEvent(reason) => write!(formatter, "invalid observable event: {reason}"), + Self::Io(error) => write!(formatter, "observability I/O failed: {error}"), + Self::Json(error) => write!(formatter, "observability JSON failed: {error}"), + Self::JournalClosed => formatter.write_str("observability journal is closed"), + Self::CorruptJournal { path, line } => { + write!( + formatter, + "corrupt journal {} at line {line}", + path.display() + ) + } + } + } +} + +impl Error for ObservabilityError { + fn source(&self) -> Option<&(dyn Error + 'static)> { + match self { + Self::Io(error) => Some(error), + Self::Json(error) => Some(error), + _ => None, + } + } +} + +impl From for ObservabilityError { + fn from(value: std::io::Error) -> Self { + Self::Io(value) + } +} + +impl From for ObservabilityError { + fn from(value: serde_json::Error) -> Self { + Self::Json(value) + } +} + +#[derive(Clone, Debug, Default, Eq, PartialEq, Serialize, Deserialize)] +pub struct OutcomeMetrics { + pub allowed: u64, + pub denied: u64, + pub approval_required: u64, + pub completed: u64, + pub failed: u64, + pub cancelled: u64, +} + +#[derive(Clone, Debug, Default, Eq, PartialEq, Serialize, Deserialize)] +pub struct LatencyMetrics { + pub samples: u64, + pub total_millis: u64, + pub maximum_millis: u64, +} + +#[derive(Clone, Debug, Default, Eq, PartialEq, Serialize, Deserialize)] +pub struct MetricsSnapshot { + pub ready: bool, + pub reconnects: u64, + pub active_sessions: u64, + pub active_tasks: u64, + pub queue_depth: u64, + pub queue_capacity: u64, + pub inference_latency: LatencyMetrics, + pub tool_latency: LatencyMetrics, + pub inference_outcomes: OutcomeMetrics, + pub tool_outcomes: OutcomeMetrics, + pub policy_outcomes: OutcomeMetrics, + pub rate_limit_used: u64, + pub rate_limit_capacity: u64, + pub recorded_events: u64, + pub dropped_ring_events: u64, + pub dropped_subscriber_events: u64, + pub dropped_journal_events: u64, + pub diagnostic_entries: u64, +} + +#[derive(Debug, Default)] +struct AtomicLatency { + samples: AtomicU64, + total: AtomicU64, + maximum: AtomicU64, +} + +impl AtomicLatency { + fn record(&self, millis: u64) { + self.samples.fetch_add(1, Ordering::Relaxed); + self.total.fetch_add(millis, Ordering::Relaxed); + self.maximum.fetch_max(millis, Ordering::Relaxed); + } + + fn snapshot(&self) -> LatencyMetrics { + LatencyMetrics { + samples: self.samples.load(Ordering::Relaxed), + total_millis: self.total.load(Ordering::Relaxed), + maximum_millis: self.maximum.load(Ordering::Relaxed), + } + } +} + +#[derive(Debug, Default)] +struct AtomicOutcomes { + allowed: AtomicU64, + denied: AtomicU64, + approval_required: AtomicU64, + completed: AtomicU64, + failed: AtomicU64, + cancelled: AtomicU64, +} + +impl AtomicOutcomes { + fn record(&self, code: OutcomeCode) { + let counter = match code { + OutcomeCode::Allowed => &self.allowed, + OutcomeCode::Denied => &self.denied, + OutcomeCode::ApprovalRequired => &self.approval_required, + OutcomeCode::Completed => &self.completed, + OutcomeCode::Failed => &self.failed, + OutcomeCode::Cancelled => &self.cancelled, + }; + counter.fetch_add(1, Ordering::Relaxed); + } + + fn snapshot(&self) -> OutcomeMetrics { + OutcomeMetrics { + allowed: self.allowed.load(Ordering::Relaxed), + denied: self.denied.load(Ordering::Relaxed), + approval_required: self.approval_required.load(Ordering::Relaxed), + completed: self.completed.load(Ordering::Relaxed), + failed: self.failed.load(Ordering::Relaxed), + cancelled: self.cancelled.load(Ordering::Relaxed), + } + } +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum OutcomeCode { + Allowed, + Denied, + ApprovalRequired, + Completed, + Failed, + Cancelled, +} + +/// Fixed-cardinality, lock-free metrics. There are no caller-defined labels. +#[derive(Debug, Default)] +pub struct AgentMetrics { + ready: AtomicBool, + reconnects: AtomicU64, + active_sessions: AtomicU64, + active_tasks: AtomicU64, + queue_depth: AtomicU64, + queue_capacity: AtomicU64, + inference_latency: AtomicLatency, + tool_latency: AtomicLatency, + inference_outcomes: AtomicOutcomes, + tool_outcomes: AtomicOutcomes, + policy_outcomes: AtomicOutcomes, + rate_limit_used: AtomicU64, + rate_limit_capacity: AtomicU64, + recorded_events: AtomicU64, + dropped_ring_events: AtomicU64, + dropped_subscriber_events: AtomicU64, + dropped_journal_events: AtomicU64, + diagnostic_entries: AtomicU64, +} + +impl AgentMetrics { + pub fn set_health(&self, ready: bool, reconnects: u64) { + self.ready.store(ready, Ordering::Relaxed); + self.reconnects.store(reconnects, Ordering::Relaxed); + } + + pub fn set_ready(&self, ready: bool) { + self.ready.store(ready, Ordering::Relaxed); + } + + pub fn record_reconnect(&self) { + self.reconnects.fetch_add(1, Ordering::Relaxed); + } + + pub fn set_load( + &self, + active_sessions: u64, + active_tasks: u64, + queue_depth: u64, + queue_capacity: u64, + ) { + self.active_sessions + .store(active_sessions, Ordering::Relaxed); + self.active_tasks.store(active_tasks, Ordering::Relaxed); + self.queue_depth.store(queue_depth, Ordering::Relaxed); + self.queue_capacity.store(queue_capacity, Ordering::Relaxed); + } + + pub fn set_rate_limit(&self, used: u64, capacity: u64) { + self.rate_limit_used.store(used, Ordering::Relaxed); + self.rate_limit_capacity.store(capacity, Ordering::Relaxed); + } + + pub fn record_inference_latency(&self, duration: Duration) { + self.inference_latency.record(millis(duration)); + } + + pub fn record_tool_latency(&self, duration: Duration) { + self.tool_latency.record(millis(duration)); + } + + pub fn record_tool_outcome(&self, outcome: OutcomeCode) { + self.tool_outcomes.record(outcome); + } + + pub fn record_inference_outcome(&self, outcome: OutcomeCode) { + self.inference_outcomes.record(outcome); + } + + pub fn record_policy_outcome(&self, outcome: OutcomeCode) { + self.policy_outcomes.record(outcome); + } + + #[must_use] + pub fn snapshot(&self) -> MetricsSnapshot { + MetricsSnapshot { + ready: self.ready.load(Ordering::Relaxed), + reconnects: self.reconnects.load(Ordering::Relaxed), + active_sessions: self.active_sessions.load(Ordering::Relaxed), + active_tasks: self.active_tasks.load(Ordering::Relaxed), + queue_depth: self.queue_depth.load(Ordering::Relaxed), + queue_capacity: self.queue_capacity.load(Ordering::Relaxed), + inference_latency: self.inference_latency.snapshot(), + tool_latency: self.tool_latency.snapshot(), + inference_outcomes: self.inference_outcomes.snapshot(), + tool_outcomes: self.tool_outcomes.snapshot(), + policy_outcomes: self.policy_outcomes.snapshot(), + rate_limit_used: self.rate_limit_used.load(Ordering::Relaxed), + rate_limit_capacity: self.rate_limit_capacity.load(Ordering::Relaxed), + recorded_events: self.recorded_events.load(Ordering::Relaxed), + dropped_ring_events: self.dropped_ring_events.load(Ordering::Relaxed), + dropped_subscriber_events: self.dropped_subscriber_events.load(Ordering::Relaxed), + dropped_journal_events: self.dropped_journal_events.load(Ordering::Relaxed), + diagnostic_entries: self.diagnostic_entries.load(Ordering::Relaxed), + } + } +} + +#[derive(Debug)] +struct RecorderState { + ring: VecDeque, + subscribers: BTreeMap>, + next_subscriber: u64, +} + +#[derive(Debug)] +enum JournalCommand { + Event(Vec), + Shutdown(oneshot::Sender>), +} + +/// Nonblocking recorder shared by service components and control clients. +pub struct Observability { + limits: ObservabilityLimits, + next_event: AtomicU64, + state: Mutex, + metrics: AgentMetrics, + journal: Option>, + event_id_limit: u64, +} + +/// Policy audit sink that preserves the existing fail-closed bounded audit and +/// mirrors content-free decisions into the unified event stream. +#[derive(Debug)] +pub struct UnifiedPolicyAudit { + memory: Arc, + observability: Arc, +} + +impl UnifiedPolicyAudit { + #[must_use] + pub const fn new(memory: Arc, observability: Arc) -> Self { + Self { + memory, + observability, + } + } + + #[must_use] + pub fn memory(&self) -> Arc { + Arc::clone(&self.memory) + } +} + +impl PolicyAuditSink for UnifiedPolicyAudit { + fn emit(&self, record: PolicyAuditRecord) -> Result<(), PolicyAuditError> { + self.memory.emit(record.clone())?; + let correlation = CorrelationIds { + avatar_id: record.origin_avatar_id.map(|id| id.to_string()), + session_id: Some(record.session_id.as_str().to_owned()), + request_id: Some(record.correlation_id.as_str().to_owned()), + action_id: record + .authorization_id + .map(|id| format!("authorization-{id}")), + }; + let disposition = policy_disposition_code(record.disposition); + let reason = policy_reason_code(record.reason); + let outcome = policy_outcome_code(record.final_outcome); + let proposed = EventDraft::new( + EventFamily::ToolProposed, + EventSeverity::Info, + "policy", + EventOrigin::Model, + ) + .and_then(|event| event.correlation(correlation.clone())) + .and_then(|event| event.identifier_field("tool", record.tool.as_str())) + .and_then(|event| { + event.identifier_field("arguments_digest", record.arguments_hash.as_str()) + }) + .and_then(|event| event.redacted("tool_arguments")); + if let Ok(proposed) = proposed { + let _ = self.observability.record(proposed); + } + let family = match record.final_outcome { + PolicyFinalOutcome::ApprovalGranted => EventFamily::ApprovalDecision, + PolicyFinalOutcome::Completed + | PolicyFinalOutcome::Rejected + | PolicyFinalOutcome::Failed + | PolicyFinalOutcome::AmbiguousMutation => EventFamily::ToolResult, + _ => EventFamily::PolicyDecision, + }; + let severity = if matches!( + record.final_outcome, + PolicyFinalOutcome::Failed | PolicyFinalOutcome::AmbiguousMutation + ) { + EventSeverity::Error + } else if matches!(record.disposition, PolicyDisposition::Denied) { + EventSeverity::Warning + } else { + EventSeverity::Info + }; + let event = EventDraft::new(family, severity, "policy", EventOrigin::Policy) + .and_then(|event| event.correlation(correlation)) + .and_then(|event| event.result_code(outcome)) + .and_then(|event| event.reason_code(reason)) + .and_then(|event| event.identifier_field("tool", record.tool.as_str())) + .and_then(|event| event.code_field("disposition", disposition)) + .and_then(|event| { + event.field("tool_calls", Value::from(record.applied_budget.tool_calls)) + }) + .and_then(|event| { + event.field( + "movement_millimeters", + Value::from(record.applied_budget.movement_millimeters), + ) + }) + .and_then(|event| event.redacted("tool_arguments")); + if let Ok(event) = event { + let _ = self.observability.record(event); + } + let metric = match record.final_outcome { + PolicyFinalOutcome::Denied | PolicyFinalOutcome::Rejected => OutcomeCode::Denied, + PolicyFinalOutcome::ApprovalRequired => OutcomeCode::ApprovalRequired, + PolicyFinalOutcome::Completed => OutcomeCode::Completed, + PolicyFinalOutcome::Failed | PolicyFinalOutcome::AmbiguousMutation => { + OutcomeCode::Failed + } + PolicyFinalOutcome::ApprovalGranted | PolicyFinalOutcome::Authorized => { + OutcomeCode::Allowed + } + }; + self.observability.metrics.record_policy_outcome(metric); + if family == EventFamily::ToolResult { + self.observability.metrics.record_tool_outcome(metric); + } + Ok(()) + } +} + +impl fmt::Debug for Observability { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("Observability") + .field("limits", &self.limits) + .field("metrics", &self.metrics.snapshot()) + .finish_non_exhaustive() + } +} + +impl Observability { + pub fn memory(limits: ObservabilityLimits) -> Result, ObservabilityError> { + if !limits.is_valid() { + return Err(ObservabilityError::UnsafeLimits); + } + Ok(Arc::new(Self::create(limits, None, 1, u64::MAX))) + } + + /// Creates a recorder with an owned asynchronous journal worker. + pub fn journaled( + limits: ObservabilityLimits, + config: JournalConfig, + ) -> Result<(Arc, ObservabilityRuntime), ObservabilityError> { + if !limits.is_valid() { + return Err(ObservabilityError::UnsafeLimits); + } + config.validate(limits.max_event_bytes)?; + fs::create_dir_all(&config.directory)?; + let active = config.directory.join("events-current.jsonl.tmp"); + recover_active(&active, config.max_segment_bytes)?; + let scanned_next = scan_last_event_id(&config.directory, limits.max_event_bytes)? + .checked_add(1) + .ok_or(ObservabilityError::InvalidEvent("event ID space exhausted"))?; + let (next_event, event_id_limit) = reserve_event_ids(&config.directory, scanned_next)?; + let (sender, receiver) = mpsc::channel(limits.journal_queue); + let observer = Arc::new(Self::create( + limits, + Some(sender), + next_event, + event_id_limit, + )); + let metrics = Arc::downgrade(&observer); + let stop = Arc::new(AtomicBool::new(false)); + let worker_stop = Arc::clone(&stop); + let task = tokio::task::spawn_blocking(move || { + journal_worker(config, receiver, metrics, &worker_stop) + }); + Ok(( + observer, + ObservabilityRuntime { + task: Some(task), + stop, + }, + )) + } + + fn create( + limits: ObservabilityLimits, + journal: Option>, + next_event: u64, + event_id_limit: u64, + ) -> Self { + Self { + limits, + next_event: AtomicU64::new(next_event), + state: Mutex::new(RecorderState { + ring: VecDeque::with_capacity(limits.ring_events), + subscribers: BTreeMap::new(), + next_subscriber: 1, + }), + metrics: AgentMetrics::default(), + journal, + event_id_limit, + } + } + + #[must_use] + pub const fn metrics(&self) -> &AgentMetrics { + &self.metrics + } + + /// Records without awaiting disk or subscribers. Overflow is counted and + /// old ring entries are evicted; slow subscribers are disconnected. + pub fn record(&self, draft: EventDraft) -> Result { + let event_id = self + .next_event + .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| { + (current < self.event_id_limit) + .then(|| current.checked_add(1)) + .flatten() + }) + .map_err(|_| ObservabilityError::InvalidEvent("event ID space exhausted"))?; + let event = StructuredEvent { + schema_version: OBSERVABILITY_SCHEMA_VERSION, + event_id, + unix_millis: unix_millis(), + severity: draft.severity, + component: draft.component, + family: draft.family.as_str().to_owned(), + correlation: draft.correlation, + origin: draft.origin, + duration_millis: draft.duration_millis, + retry_count: draft.retry_count, + result_code: draft.result_code, + reason_code: draft.reason_code, + redaction_flags: draft.redaction_flags, + fields: draft.fields, + extensions: BTreeMap::new(), + }; + let bytes = serde_json::to_vec(&event)?; + if bytes.len().saturating_add(1) > self.limits.max_event_bytes { + return Err(ObservabilityError::InvalidEvent( + "serialized event exceeds bound", + )); + } + self.metrics.recorded_events.fetch_add(1, Ordering::Relaxed); + { + let mut state = lock(&self.state); + if state.ring.len() == self.limits.ring_events { + state.ring.pop_front(); + self.metrics + .dropped_ring_events + .fetch_add(1, Ordering::Relaxed); + } + state.ring.push_back(event.clone()); + state.subscribers.retain(|_, subscriber| { + if subscriber.try_send(event.clone()).is_ok() { + true + } else { + self.metrics + .dropped_subscriber_events + .fetch_add(1, Ordering::Relaxed); + false + } + }); + } + if let Some(journal) = &self.journal + && journal.try_send(JournalCommand::Event(bytes)).is_err() + { + self.metrics + .dropped_journal_events + .fetch_add(1, Ordering::Relaxed); + } + Ok(event_id) + } + + #[must_use] + pub fn snapshot(&self) -> Vec { + lock(&self.state).ring.iter().cloned().collect() + } + + pub fn subscribe( + self: &Arc, + after_event_id: Option, + ) -> Result { + let mut state = lock(&self.state); + if state.subscribers.len() == self.limits.max_subscribers { + return Err(ObservabilityError::InvalidEvent("subscriber limit reached")); + } + let retained_first = state + .ring + .front() + .map_or(self.next_event.load(Ordering::Relaxed), |event| { + event.event_id + }); + let requested = after_event_id.map_or(retained_first, |id| id.saturating_add(1)); + let gap = (requested < retained_first).then_some((requested, retained_first)); + let initial = state + .ring + .iter() + .filter(|event| event.event_id >= requested.max(retained_first)) + .cloned() + .collect(); + let (sender, receiver) = mpsc::channel(self.limits.subscriber_queue); + let id = state.next_subscriber; + state.next_subscriber = state.next_subscriber.saturating_add(1); + state.subscribers.insert(id, sender); + Ok(EventSubscription { + id, + owner: Arc::downgrade(self), + gap, + initial, + receiver, + }) + } + + /// Records an explicit diagnostic envelope containing only sizes and + /// SHA-256 digests. Content is never retained, even when capture is enabled. + pub fn capture_diagnostic( + &self, + direction: DiagnosticDirection, + request_id: impl Into, + parts: &[SensitiveDiagnosticPart<'_>], + ) -> Result { + let current = self.metrics.diagnostic_entries.load(Ordering::Relaxed); + if current >= u64::try_from(self.limits.max_diagnostic_entries).unwrap_or(u64::MAX) { + return Err(ObservabilityError::InvalidEvent( + "diagnostic retention limit reached", + )); + } + let request_id = request_id.into(); + if !valid_identifier(&request_id) || parts.len() > 64 { + return Err(ObservabilityError::InvalidEvent( + "invalid diagnostic envelope", + )); + } + let total_bytes = parts + .iter() + .fold(0usize, |sum, part| sum.saturating_add(part.contents.len())); + let mut hasher = Sha256::new(); + for part in parts { + hasher.update(part.role.as_bytes()); + hasher.update([0]); + hasher.update(part.contents.as_bytes()); + hasher.update([0xff]); + } + let digest = format!("sha256:{}", hex_bytes(&hasher.finalize())); + let draft = EventDraft::new( + EventFamily::DiagnosticEnvelope, + EventSeverity::Warning, + "llm", + EventOrigin::Model, + )? + .correlation(CorrelationIds { + request_id: Some(request_id), + ..CorrelationIds::default() + })? + .redacted("content_omitted")? + .redacted("diagnostic_capture")? + .code_field("direction", direction.as_str())? + .field("part_count", Value::from(parts.len()))? + .field("content_bytes", Value::from(total_bytes))? + .identifier_field("content_digest", &digest)?; + self.metrics + .diagnostic_entries + .fetch_add(1, Ordering::Relaxed); + self.record(draft) + } +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum DiagnosticDirection { + Prompt, + Response, +} + +impl DiagnosticDirection { + const fn as_str(self) -> &'static str { + match self { + Self::Prompt => "prompt", + Self::Response => "response", + } + } +} + +pub struct SensitiveDiagnosticPart<'a> { + pub role: &'a str, + pub contents: &'a str, +} + +impl fmt::Debug for SensitiveDiagnosticPart<'_> { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("SensitiveDiagnosticPart") + .field("role_bytes", &self.role.len()) + .field("content_bytes", &self.contents.len()) + .field("contents", &"[REDACTED]") + .finish() + } +} + +pub struct EventSubscription { + id: u64, + owner: Weak, + gap: Option<(u64, u64)>, + initial: VecDeque, + receiver: mpsc::Receiver, +} + +impl EventSubscription { + #[must_use] + pub fn take_gap(&mut self) -> Option<(u64, u64)> { + self.gap.take() + } + + pub async fn recv(&mut self) -> Option { + if let Some(event) = self.initial.pop_front() { + Some(event) + } else { + self.receiver.recv().await + } + } +} + +impl Drop for EventSubscription { + fn drop(&mut self) { + if let Some(owner) = self.owner.upgrade() { + lock(&owner.state).subscribers.remove(&self.id); + } + } +} + +pub struct ObservabilityRuntime { + task: Option>>, + stop: Arc, +} + +impl ObservabilityRuntime { + pub async fn shutdown(mut self, observer: &Observability) -> Result<(), ObservabilityError> { + if let Some(sender) = &observer.journal { + let (reply, receive) = oneshot::channel(); + sender + .send(JournalCommand::Shutdown(reply)) + .await + .map_err(|_| ObservabilityError::JournalClosed)?; + receive + .await + .map_err(|_| ObservabilityError::JournalClosed)??; + } + if let Some(task) = self.task.take() { + task.await + .map_err(|_| ObservabilityError::JournalClosed)??; + } + Ok(()) + } +} + +impl Drop for ObservabilityRuntime { + fn drop(&mut self) { + // `spawn_blocking` tasks cannot be cancelled once running. Signal the + // polling worker so dropping this guard cannot leave an orphan thread. + self.stop.store(true, Ordering::Release); + } +} + +#[derive(Clone, Debug, Default, Eq, PartialEq, Serialize, Deserialize)] +pub struct ReplayAction { + pub action_id: String, + pub events: Vec, + pub final_result: Option, +} + +#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)] +pub struct ReplayState { + pub last_event_id: u64, + pub lifecycle_state: Option, + pub behavior_state: Option, + pub sessions_created: u64, + pub sessions_expired: u64, + pub policy_decisions: u64, + pub tool_outcomes: u64, + pub unavailable_private_content: bool, + pub actions: BTreeMap, + pub timeline: Vec, +} + +/// Reads journal files in segment order and reconstructs observable state only. +/// It never invokes a backend, model, policy gateway, or mutation executor. +pub fn replay_journal( + directory: impl AsRef, + max_event_bytes: usize, + max_events: usize, +) -> Result { + if max_event_bytes == 0 || max_event_bytes > MAX_EVENT_BYTES_HARD || max_events == 0 { + return Err(ObservabilityError::UnsafeLimits); + } + let mut paths = journal_paths(directory.as_ref())?; + paths.sort(); + let mut replay = ReplayState::default(); + for path in paths { + let active = path + .file_name() + .is_some_and(|name| name == "events-current.jsonl.tmp"); + read_journal_file(&path, active, max_event_bytes, max_events, &mut replay)?; + } + replay.timeline.sort_by_key(|event| event.event_id); + replay.last_event_id = replay.timeline.last().map_or(0, |event| event.event_id); + Ok(replay) +} + +fn read_journal_file( + path: &Path, + allow_truncated_final: bool, + max_event_bytes: usize, + max_events: usize, + replay: &mut ReplayState, +) -> Result<(), ObservabilityError> { + let file = File::open(path)?; + let mut reader = BufReader::new(file); + let mut line = Vec::new(); + let mut line_number = 0usize; + loop { + line.clear(); + let bytes = reader.read_until(b'\n', &mut line)?; + if bytes == 0 { + break; + } + line_number = line_number.saturating_add(1); + if line.len() > max_event_bytes { + return Err(ObservabilityError::CorruptJournal { + path: path.to_owned(), + line: line_number, + }); + } + let terminated = line.last() == Some(&b'\n'); + if !terminated && allow_truncated_final { + break; + } + let event: StructuredEvent = + serde_json::from_slice(&line).map_err(|_| ObservabilityError::CorruptJournal { + path: path.to_owned(), + line: line_number, + })?; + if replay.timeline.len() == max_events { + return Err(ObservabilityError::UnsafeLimits); + } + apply_replay(replay, &event); + replay.timeline.push(event); + } + Ok(()) +} + +fn apply_replay(replay: &mut ReplayState, event: &StructuredEvent) { + replay.unavailable_private_content |= event.redaction_flags.iter().any(|flag| { + matches!( + flag.as_str(), + "content_omitted" | "private_content" | "inventory_payload" + ) + }); + match event.family.as_str() { + "lifecycle_transition" => { + replay.lifecycle_state = event + .fields + .get("to") + .and_then(Value::as_str) + .map(str::to_owned); + } + "behavior_transition" => { + replay.behavior_state = event + .fields + .get("to") + .and_then(Value::as_str) + .map(str::to_owned); + } + "session_created" => replay.sessions_created = replay.sessions_created.saturating_add(1), + "session_expired" => replay.sessions_expired = replay.sessions_expired.saturating_add(1), + "policy_decision" | "approval_decision" => { + replay.policy_decisions = replay.policy_decisions.saturating_add(1); + } + "tool_result" => replay.tool_outcomes = replay.tool_outcomes.saturating_add(1), + _ => {} + } + if let Some(action_id) = &event.correlation.action_id { + let action = replay + .actions + .entry(action_id.clone()) + .or_insert_with(|| ReplayAction { + action_id: action_id.clone(), + ..ReplayAction::default() + }); + action.events.push(event.event_id); + if event.family == "tool_result" { + action.final_result.clone_from(&event.result_code); + } + } +} + +#[allow(clippy::needless_pass_by_value)] // The blocking worker owns the weak lifecycle handle. +fn journal_worker( + config: JournalConfig, + mut receiver: mpsc::Receiver, + observer: Weak, + stop: &AtomicBool, +) -> Result<(), ObservabilityError> { + let mut writer = JournalWriter::open(config)?; + loop { + if stop.load(Ordering::Acquire) { + return writer.finish(); + } + let command = match receiver.try_recv() { + Ok(command) => command, + Err(mpsc::error::TryRecvError::Empty) => { + std::thread::sleep(Duration::from_millis(10)); + continue; + } + Err(mpsc::error::TryRecvError::Disconnected) => return writer.finish(), + }; + match command { + JournalCommand::Event(bytes) => { + if let Err(error) = writer.append(&bytes) { + if let Some(observer) = observer.upgrade() { + observer + .metrics + .dropped_journal_events + .fetch_add(1, Ordering::Relaxed); + } + return Err(error); + } + } + JournalCommand::Shutdown(reply) => { + let result = writer.finish(); + let reply_result = result + .as_ref() + .map_err(|error| { + ObservabilityError::InvalidEvent(match error { + ObservabilityError::Io(_) => "journal flush failed", + _ => "journal shutdown failed", + }) + }) + .copied(); + let _ = reply.send(reply_result); + return result; + } + } + } +} + +struct JournalWriter { + config: JournalConfig, + active_path: PathBuf, + file: File, + bytes: u64, + next_segment: u64, +} + +impl JournalWriter { + fn open(config: JournalConfig) -> Result { + fs::create_dir_all(&config.directory)?; + let active_path = config.directory.join("events-current.jsonl.tmp"); + recover_active(&active_path, config.max_segment_bytes)?; + let file = OpenOptions::new() + .create(true) + .append(true) + .read(true) + .open(&active_path)?; + let bytes = file.metadata()?.len(); + let next_segment = next_segment_id(&config.directory)?; + Ok(Self { + config, + active_path, + file, + bytes, + next_segment, + }) + } + + fn append(&mut self, event: &[u8]) -> Result<(), ObservabilityError> { + let required = u64::try_from(event.len().saturating_add(1)).unwrap_or(u64::MAX); + if self.bytes > 0 && self.bytes.saturating_add(required) > self.config.max_segment_bytes { + self.rotate()?; + } + self.file.write_all(event)?; + self.file.write_all(b"\n")?; + self.file.flush()?; + if self.config.sync_each_record { + self.file.sync_data()?; + } + self.bytes = self.bytes.saturating_add(required); + enforce_retention(&self.config) + } + + fn rotate(&mut self) -> Result<(), ObservabilityError> { + self.file.sync_all()?; + let archived = self + .config + .directory + .join(format!("events-{:020}.jsonl", self.next_segment)); + self.next_segment = self.next_segment.saturating_add(1); + fs::rename(&self.active_path, archived)?; + self.file = OpenOptions::new() + .create_new(true) + .append(true) + .read(true) + .open(&self.active_path)?; + self.bytes = 0; + enforce_retention(&self.config)?; + Ok(()) + } + + fn finish(mut self) -> Result<(), ObservabilityError> { + self.file.flush()?; + self.file.sync_all()?; + enforce_retention(&self.config) + } +} + +fn recover_active(path: &Path, max_segment_bytes: u64) -> Result<(), ObservabilityError> { + let Some(length) = fs::metadata(path).ok().map(|metadata| metadata.len()) else { + return Ok(()); + }; + if length > max_segment_bytes { + return Err(ObservabilityError::UnsafeLimits); + } + let mut file = OpenOptions::new().read(true).write(true).open(path)?; + if length == 0 { + return Ok(()); + } + file.seek(SeekFrom::End(-1))?; + let mut byte = [0u8; 1]; + file.read_exact(&mut byte)?; + if byte[0] == b'\n' { + return Ok(()); + } + file.seek(SeekFrom::Start(0))?; + let mut bytes = Vec::with_capacity(usize::try_from(length).unwrap_or(0)); + file.read_to_end(&mut bytes)?; + let valid = bytes + .iter() + .rposition(|byte| *byte == b'\n') + .map_or(0, |index| index.saturating_add(1)); + file.set_len(u64::try_from(valid).unwrap_or(0))?; + file.sync_all()?; + Ok(()) +} + +fn enforce_retention(config: &JournalConfig) -> Result<(), ObservabilityError> { + let mut segments = journal_paths(&config.directory)? + .into_iter() + .filter(|path| { + path.extension() + .is_some_and(|extension| extension == "jsonl") + }) + .collect::>(); + segments.sort(); + let active_bytes = fs::metadata(config.directory.join("events-current.jsonl.tmp")) + .map_or(0, |metadata| metadata.len()); + let mut total = segments.iter().try_fold(active_bytes, |sum, path| { + Ok::<_, std::io::Error>(sum.saturating_add(fs::metadata(path)?.len())) + })?; + while segments.len() > config.max_segments || total > config.max_total_bytes { + let oldest = segments.remove(0); + let length = fs::metadata(&oldest)?.len(); + fs::remove_file(oldest)?; + total = total.saturating_sub(length); + } + Ok(()) +} + +/// Reserves a large, durable ID range before admitting journaled events. This +/// prevents IDs that were observed in memory but dropped by a saturated disk +/// queue from being reused after a restart. Immutable reservation markers keep +/// the update atomic on Unix and Windows without replacing an open file. +fn reserve_event_ids( + directory: &Path, + scanned_next: u64, +) -> Result<(u64, u64), ObservabilityError> { + loop { + let mut reservations = reservation_paths(directory)?; + reservations.sort_by_key(|(limit, _)| *limit); + let reserved_next = reservations.last().map_or(1, |(limit, _)| *limit); + let next = scanned_next.max(reserved_next); + if next == u64::MAX { + return Err(ObservabilityError::InvalidEvent("event ID space exhausted")); + } + let limit = next.saturating_add(JOURNAL_EVENT_ID_RESERVATION); + let marker = directory.join(format!("events-id-reservation-{limit:020}")); + match OpenOptions::new() + .create_new(true) + .write(true) + .open(&marker) + { + Ok(mut file) => { + writeln!(file, "{next}..{limit}")?; + file.sync_all()?; + reservations.push((limit, marker)); + reservations.sort_by_key(|(reserved_limit, _)| *reserved_limit); + while reservations.len() > 1 { + let (_, oldest) = reservations.remove(0); + fs::remove_file(oldest)?; + } + return Ok((next, limit)); + } + Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => {} + Err(error) => return Err(error.into()), + } + } +} + +fn reservation_paths(directory: &Path) -> Result, ObservabilityError> { + let mut paths = Vec::new(); + for entry in fs::read_dir(directory)? { + let entry = entry?; + if !entry.file_type()?.is_file() { + continue; + } + let name = entry.file_name(); + let name = name.to_string_lossy(); + let Some(limit) = name + .strip_prefix("events-id-reservation-") + .and_then(|value| value.parse::().ok()) + else { + continue; + }; + paths.push((limit, entry.path())); + } + Ok(paths) +} + +fn next_segment_id(directory: &Path) -> Result { + let retained = journal_paths(directory)? + .into_iter() + .filter_map(|path| { + let name = path.file_name()?.to_str()?; + name.strip_prefix("events-")? + .strip_suffix(".jsonl")? + .parse::() + .ok() + }) + .max(); + let retained_next = retained.map_or(Ok(0), |identifier| { + identifier + .checked_add(1) + .ok_or(ObservabilityError::InvalidEvent( + "journal segment ID space exhausted", + )) + })?; + Ok(unix_millis().max(retained_next)) +} + +fn scan_last_event_id(directory: &Path, max_event_bytes: usize) -> Result { + let mut maximum = 0u64; + for path in journal_paths(directory)? { + let active = path + .file_name() + .is_some_and(|name| name == "events-current.jsonl.tmp"); + let mut reader = BufReader::new(File::open(&path)?); + let mut line = Vec::new(); + let mut line_number = 0usize; + loop { + line.clear(); + let bytes = reader.read_until(b'\n', &mut line)?; + if bytes == 0 { + break; + } + line_number = line_number.saturating_add(1); + if line.len() > max_event_bytes { + return Err(ObservabilityError::CorruptJournal { + path: path.clone(), + line: line_number, + }); + } + if line.last() != Some(&b'\n') && active { + break; + } + let event: StructuredEvent = + serde_json::from_slice(&line).map_err(|_| ObservabilityError::CorruptJournal { + path: path.clone(), + line: line_number, + })?; + maximum = maximum.max(event.event_id); + } + } + Ok(maximum) +} + +fn journal_paths(directory: &Path) -> Result, ObservabilityError> { + if !directory.exists() { + return Ok(Vec::new()); + } + let mut paths = Vec::new(); + for entry in fs::read_dir(directory)? { + let entry = entry?; + if !entry.file_type()?.is_file() { + continue; + } + let name = entry.file_name(); + let name = name.to_string_lossy(); + if (name.starts_with("events-") && name.ends_with(".jsonl")) + || name == "events-current.jsonl.tmp" + { + paths.push(entry.path()); + } + } + Ok(paths) +} + +fn checked_code(value: String) -> Result { + if valid_code(&value) { + Ok(value) + } else { + Err(ObservabilityError::InvalidEvent("invalid stable code")) + } +} + +const fn policy_disposition_code(value: PolicyDisposition) -> &'static str { + match value { + PolicyDisposition::Allowed => "allowed", + PolicyDisposition::Denied => "denied", + PolicyDisposition::ApprovalRequired => "approval_required", + } +} + +const fn policy_outcome_code(value: PolicyFinalOutcome) -> &'static str { + match value { + PolicyFinalOutcome::Denied => "denied", + PolicyFinalOutcome::ApprovalRequired => "approval_required", + PolicyFinalOutcome::ApprovalGranted => "approval_granted", + PolicyFinalOutcome::Authorized => "authorized", + PolicyFinalOutcome::Completed => "completed", + PolicyFinalOutcome::Rejected => "rejected", + PolicyFinalOutcome::Failed => "failed", + PolicyFinalOutcome::AmbiguousMutation => "ambiguous_mutation", + } +} + +const fn policy_reason_code(value: crate::policy::PolicyReasonCode) -> &'static str { + use crate::policy::PolicyReasonCode; + match value { + PolicyReasonCode::Allowed => "allowed", + PolicyReasonCode::UnknownTool => "unknown_tool", + PolicyReasonCode::InvalidArguments => "invalid_arguments", + PolicyReasonCode::OriginDenied => "origin_denied", + PolicyReasonCode::CapabilityDenied => "capability_denied", + PolicyReasonCode::ForbiddenMilestoneOperation => "forbidden_milestone_operation", + PolicyReasonCode::ToolCostExceeded => "tool_cost_exceeded", + PolicyReasonCode::PrincipalBudgetExceeded => "principal_budget_exceeded", + PolicyReasonCode::GlobalBudgetExceeded => "global_budget_exceeded", + PolicyReasonCode::ApprovalRequired => "approval_required", + PolicyReasonCode::ApprovalUnknown => "approval_unknown", + PolicyReasonCode::ApprovalNotGranted => "approval_not_granted", + PolicyReasonCode::ApprovalExpired => "approval_expired", + PolicyReasonCode::ApprovalReplayed => "approval_replayed", + PolicyReasonCode::ApprovalMismatch => "approval_mismatch", + PolicyReasonCode::ApprovalUnexpected => "approval_unexpected", + PolicyReasonCode::SchedulerGrantUnknown => "scheduler_grant_unknown", + PolicyReasonCode::SchedulerGrantExpired => "scheduler_grant_expired", + PolicyReasonCode::SchedulerGrantExhausted => "scheduler_grant_exhausted", + PolicyReasonCode::SchedulerPrivilegeEscalation => "scheduler_privilege_escalation", + PolicyReasonCode::AuditUnavailable => "audit_unavailable", + } +} + +fn valid_code(value: &str) -> bool { + !value.is_empty() + && value.len() <= MAX_CODE_BYTES + && value + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'_' | b'-' | b'.')) +} + +fn valid_identifier(value: &str) -> bool { + !value.is_empty() + && value.len() <= MAX_ID_BYTES + && value + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'_' | b'-' | b'.' | b':')) +} + +/// Returns a stable, non-reversible correlation value for runtime identifiers. +#[must_use] +pub fn pseudonymous_identifier(value: &str) -> String { + let digest = Sha256::digest(value.as_bytes()); + format!("id:{}", &hex_bytes(&digest)[..24]) +} + +fn millis(duration: Duration) -> u64 { + u64::try_from(duration.as_millis()).unwrap_or(u64::MAX) +} + +fn hex_bytes(bytes: &[u8]) -> String { + const HEX: &[u8; 16] = b"0123456789abcdef"; + let mut result = String::with_capacity(bytes.len().saturating_mul(2)); + for byte in bytes { + result.push(char::from(HEX[usize::from(byte >> 4)])); + result.push(char::from(HEX[usize::from(byte & 0x0f)])); + } + result +} + +fn unix_millis() -> u64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map_or(0, millis) +} + +fn lock(value: &Mutex) -> MutexGuard<'_, T> { + value + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) +} + +/// Converts a structured event to a generic JSON object for clients that want +/// to preserve fields introduced by a newer schema. +pub fn event_json(event: &StructuredEvent) -> Result, ObservabilityError> { + match serde_json::to_value(event)? { + Value::Object(object) => Ok(object), + _ => Err(ObservabilityError::InvalidEvent( + "event did not encode as an object", + )), + } +} diff --git a/crates/metacrate-grid-agent/src/observability_tests.rs b/crates/metacrate-grid-agent/src/observability_tests.rs new file mode 100644 index 0000000..29c23ec --- /dev/null +++ b/crates/metacrate-grid-agent/src/observability_tests.rs @@ -0,0 +1,423 @@ +use crate::observability::{ + CorrelationIds, DiagnosticDirection, EventDraft, EventFamily, EventOrigin, EventSeverity, + JournalConfig, OBSERVABILITY_SCHEMA_VERSION, Observability, ObservabilityLimits, OutcomeCode, + SensitiveDiagnosticPart, StructuredEvent, UnifiedPolicyAudit, replay_journal, +}; +use crate::policy::{ + MemoryPolicyAudit, OriginClass, PolicyAuditRecord, PolicyAuditSink, PolicyDisposition, + PolicyFinalOutcome, PolicyReasonCode, ResourceCost, +}; +use crate::types::{BoundedText, MAX_IDENTIFIER_BYTES}; +use serde_json::{Value, json}; +use std::fs::{self, OpenOptions}; +use std::io::Write; +use std::path::PathBuf; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::time::Duration; + +static NEXT_TEMP: AtomicU64 = AtomicU64::new(1); +const SECRET_CANARY: &str = "OBSERVABILITY-SECRET-CANARY-9f7d2"; + +fn limits() -> ObservabilityLimits { + ObservabilityLimits { + ring_events: 8, + subscriber_queue: 2, + max_subscribers: 2, + journal_queue: 8, + max_event_bytes: 4_096, + max_diagnostic_entries: 4, + } +} + +fn temp_dir(label: &str) -> PathBuf { + let path = std::env::temp_dir().join(format!( + "metacrate-observability-{label}-{}-{}", + std::process::id(), + NEXT_TEMP.fetch_add(1, Ordering::Relaxed) + )); + fs::create_dir_all(&path).expect("create test directory"); + path +} + +fn draft(family: EventFamily, request: &str) -> EventDraft { + EventDraft::new(family, EventSeverity::Info, "test", EventOrigin::Service) + .expect("draft") + .correlation(CorrelationIds { + request_id: Some(request.to_owned()), + ..CorrelationIds::default() + }) + .expect("correlation") + .result_code("completed") + .expect("result") + .redacted("content_omitted") + .expect("redaction") + .field("count", Value::from(1)) + .expect("field") +} + +#[test] +fn golden_schema_covers_every_event_family_and_preserves_extensions() { + let observer = Observability::memory(ObservabilityLimits { + ring_events: 32, + ..limits() + }) + .expect("observer"); + let families = [ + EventFamily::LifecycleTransition, + EventFamily::InboundMessage, + EventFamily::OutboundMessage, + EventFamily::SessionCreated, + EventFamily::SessionExpired, + EventFamily::InferenceRequest, + EventFamily::InferenceResult, + EventFamily::ModelActionSummary, + EventFamily::ToolProposed, + EventFamily::PolicyDecision, + EventFamily::ApprovalDecision, + EventFamily::ToolStarted, + EventFamily::ToolProgress, + EventFamily::ToolResult, + EventFamily::BehaviorTransition, + EventFamily::ScheduledJob, + EventFamily::ControlCommand, + EventFamily::Shutdown, + EventFamily::DiagnosticEnvelope, + ]; + for (index, family) in families.into_iter().enumerate() { + observer + .record(draft(family, &format!("request-{index}"))) + .expect("record family"); + } + let events = observer.snapshot(); + assert_eq!(events.len(), families.len()); + let event = events.last().expect("last event"); + let value = serde_json::to_value(event).expect("JSON"); + assert_eq!(value["schema_version"], OBSERVABILITY_SCHEMA_VERSION); + assert_eq!(value["family"], "diagnostic_envelope"); + assert_eq!(value["redaction_flags"], json!(["content_omitted"])); + + let mut value = value; + value["future_top_level"] = json!({"compatible": true}); + value["fields"]["future_scalar"] = json!(42); + let decoded: StructuredEvent = serde_json::from_value(value).expect("future event"); + assert_eq!(decoded.extensions["future_top_level"]["compatible"], true); + assert_eq!(decoded.fields["future_scalar"], 42); + let encoded = serde_json::to_value(decoded).expect("round trip"); + assert_eq!(encoded["future_top_level"]["compatible"], true); +} + +#[test] +fn secret_canaries_are_unrepresentable_and_debug_is_redacted() { + let observer = Observability::memory(limits()).expect("observer"); + for sensitive in [ + SECRET_CANARY, + "Bearer OBSERVABILITY-SECRET-CANARY-9f7d2", + "https://grid.example/CAPS/OBSERVABILITY-SECRET-CANARY-9f7d2", + "password=OBSERVABILITY-SECRET-CANARY-9f7d2", + "api_key=OBSERVABILITY-SECRET-CANARY-9f7d2", + ] { + assert!( + EventDraft::new( + EventFamily::InboundMessage, + EventSeverity::Info, + "test", + EventOrigin::Grid + ) + .expect("draft") + .field("unsafe", Value::String(sensitive.to_owned())) + .is_err() + ); + } + let part = SensitiveDiagnosticPart { + role: SECRET_CANARY, + contents: SECRET_CANARY, + }; + assert!(!format!("{part:?}").contains(SECRET_CANARY)); + observer + .capture_diagnostic(DiagnosticDirection::Prompt, "request-safe", &[part]) + .expect("diagnostic metadata"); + let rendered = serde_json::to_string(&observer.snapshot()).expect("render events"); + assert!(!rendered.contains(SECRET_CANARY)); + assert!(!format!("{observer:?}").contains(SECRET_CANARY)); + assert!(rendered.contains("content_digest")); + assert!(rendered.contains("content_omitted")); + let mut external = serde_json::to_value(&observer.snapshot()[0]).expect("external event"); + external["component"] = Value::String(SECRET_CANARY.to_owned()); + external["family"] = Value::String(SECRET_CANARY.to_owned()); + external["result_code"] = Value::String(SECRET_CANARY.to_owned()); + external["reason_code"] = Value::String(SECRET_CANARY.to_owned()); + external["redaction_flags"] = serde_json::json!([SECRET_CANARY]); + external["future_secret"] = Value::String(SECRET_CANARY.to_owned()); + let external: StructuredEvent = serde_json::from_value(external).expect("external event"); + assert!(!format!("{external:?}").contains(SECRET_CANARY)); +} + +#[tokio::test] +async fn ring_overflow_reports_gap_and_slow_subscribers_are_disconnected() { + let observer = Observability::memory(ObservabilityLimits { + ring_events: 3, + subscriber_queue: 1, + ..limits() + }) + .expect("observer"); + for index in 0..5 { + observer + .record(draft( + EventFamily::ToolProgress, + &format!("request-{index}"), + )) + .expect("record"); + } + assert_eq!(observer.snapshot().len(), 3); + assert_eq!(observer.metrics().snapshot().dropped_ring_events, 2); + let mut replay = observer.subscribe(Some(0)).expect("subscription"); + assert_eq!(replay.take_gap(), Some((1, 3))); + assert_eq!(replay.recv().await.expect("replay").event_id, 3); + + let mut slow = observer.subscribe(None).expect("slow subscription"); + for index in 5..9 { + observer + .record(draft( + EventFamily::ToolProgress, + &format!("request-{index}"), + )) + .expect("record"); + } + while slow.recv().await.is_some() {} + assert!(observer.metrics().snapshot().dropped_subscriber_events >= 1); +} + +#[tokio::test] +async fn journal_rotates_recovers_truncated_tail_and_replays_without_execution() { + let directory = temp_dir("journal"); + let config = JournalConfig { + directory: directory.clone(), + max_segment_bytes: 700, + max_segments: 3, + max_total_bytes: 2_100, + sync_each_record: false, + }; + let (observer, runtime) = Observability::journaled( + ObservabilityLimits { + max_event_bytes: 512, + ..limits() + }, + config, + ) + .expect("journal observer"); + for index in 0..12 { + let family = if index == 0 { + EventFamily::LifecycleTransition + } else if index == 1 { + EventFamily::BehaviorTransition + } else { + EventFamily::ToolResult + }; + let mut event = draft(family, &format!("request-{index}")); + if family == EventFamily::LifecycleTransition || family == EventFamily::BehaviorTransition { + event = event.code_field("to", "available").expect("state"); + } + if family == EventFamily::ToolResult { + event = event + .correlation(CorrelationIds { + action_id: Some(format!("action-{index}")), + ..CorrelationIds::default() + }) + .expect("action"); + } + observer.record(event).expect("journal event"); + } + runtime.shutdown(&observer).await.expect("journal shutdown"); + let last_before_restart = observer.snapshot().last().expect("last event").event_id; + let (restarted, restarted_runtime) = Observability::journaled( + ObservabilityLimits { + max_event_bytes: 512, + ..limits() + }, + JournalConfig { + directory: directory.clone(), + max_segment_bytes: 700, + max_segments: 3, + max_total_bytes: 2_100, + sync_each_record: false, + }, + ) + .expect("restart journal"); + let restarted_id = restarted + .record(draft(EventFamily::Shutdown, "restart")) + .expect("restart event"); + assert!(restarted_id > last_before_restart); + restarted_runtime + .shutdown(&restarted) + .await + .expect("restart shutdown"); + let active = directory.join("events-current.jsonl.tmp"); + OpenOptions::new() + .append(true) + .open(&active) + .expect("active") + .write_all(b"{\"schema_version\":1") + .expect("torn record"); + let replay = replay_journal(&directory, 4_096, 64).expect("recover replay"); + assert!(!replay.timeline.is_empty()); + assert!(replay.tool_outcomes > 0); + assert!(replay.unavailable_private_content); + assert!( + replay + .actions + .values() + .all(|action| action.final_result.is_some()) + ); + // Three retained segments, one active file, and one durable ID reservation. + assert!(fs::read_dir(&directory).expect("directory").count() <= 5); + let journal_bytes = fs::read_dir(&directory) + .expect("directory") + .filter_map(Result::ok) + .filter(|entry| { + let name = entry.file_name(); + let name = name.to_string_lossy(); + name.ends_with(".jsonl") || name == "events-current.jsonl.tmp" + }) + .map(|entry| entry.metadata().expect("metadata").len()) + .sum::(); + assert!(journal_bytes <= 2_100); + fs::remove_dir_all(directory).expect("remove test directory"); +} + +#[test] +fn metrics_have_fixed_fields_and_saturating_latency_summaries() { + let observer = Observability::memory(limits()).expect("observer"); + let metrics = observer.metrics(); + metrics.set_health(true, 3); + metrics.set_load(4, 5, 6, 7); + metrics.set_rate_limit(8, 9); + metrics.record_inference_latency(Duration::from_millis(11)); + metrics.record_inference_latency(Duration::from_millis(13)); + metrics.record_tool_latency(Duration::from_millis(17)); + metrics.record_inference_outcome(OutcomeCode::Failed); + metrics.record_tool_outcome(OutcomeCode::Completed); + metrics.record_policy_outcome(OutcomeCode::Denied); + let snapshot = metrics.snapshot(); + assert!(snapshot.ready); + assert_eq!(snapshot.reconnects, 3); + assert_eq!(snapshot.active_sessions, 4); + assert_eq!(snapshot.queue_capacity, 7); + assert_eq!(snapshot.inference_latency.samples, 2); + assert_eq!(snapshot.inference_latency.total_millis, 24); + assert_eq!(snapshot.inference_latency.maximum_millis, 13); + assert_eq!(snapshot.tool_latency.maximum_millis, 17); + assert_eq!(snapshot.inference_outcomes.failed, 1); + assert_eq!(snapshot.tool_outcomes.completed, 1); + assert_eq!(snapshot.policy_outcomes.denied, 1); + assert_eq!( + serde_json::to_value(snapshot) + .expect("metrics") + .as_object() + .expect("object") + .len(), + 18 + ); +} + +#[test] +fn unified_policy_audit_correlates_outcomes_without_exposing_inputs() { + let probe_observer = Observability::memory(limits()).expect("probe observer"); + let probe = EventDraft::new( + EventFamily::ToolResult, + EventSeverity::Info, + "policy", + EventOrigin::Policy, + ) + .and_then(|event| { + event.correlation(CorrelationIds { + session_id: Some(SECRET_CANARY.to_owned()), + request_id: Some(SECRET_CANARY.to_owned()), + action_id: Some("authorization-7".to_owned()), + ..CorrelationIds::default() + }) + }) + .and_then(|event| event.result_code("completed")) + .and_then(|event| event.reason_code("allowed")) + .and_then(|event| event.identifier_field("tool", SECRET_CANARY)) + .and_then(|event| event.code_field("disposition", "allowed")) + .and_then(|event| event.field("tool_calls", Value::from(1))) + .and_then(|event| event.field("movement_millimeters", Value::from(0))) + .and_then(|event| event.redacted("tool_arguments")) + .expect("policy event draft"); + probe_observer.record(probe).expect("policy event record"); + + let observer = Observability::memory(limits()).expect("observer"); + let memory = std::sync::Arc::new(MemoryPolicyAudit::new(8).expect("policy audit")); + let audit = UnifiedPolicyAudit::new(memory.clone(), observer.clone()); + audit + .emit(PolicyAuditRecord { + recorded_unix_millis: 1, + authorization_id: Some(7), + approval_id: None, + origin: OriginClass::AuthorizedIm, + origin_avatar_id: None, + principal: BoundedText::::new("principal", SECRET_CANARY) + .expect("principal"), + session_id: BoundedText::::new("session", SECRET_CANARY) + .expect("session"), + correlation_id: BoundedText::::new("correlation", SECRET_CANARY) + .expect("correlation"), + tool: BoundedText::::new("tool", SECRET_CANARY).expect("tool"), + disposition: PolicyDisposition::Allowed, + reason: PolicyReasonCode::Allowed, + applied_budget: ResourceCost::one_call(), + arguments_hash: BoundedText::<64>::new("hash", SECRET_CANARY).expect("hash"), + final_outcome: PolicyFinalOutcome::Completed, + }) + .expect("emit audit"); + assert_eq!(memory.snapshot().len(), 1); + let rendered = serde_json::to_string(&observer.snapshot()).expect("events"); + assert!(!rendered.contains(SECRET_CANARY)); + let families = observer + .snapshot() + .iter() + .map(|event| event.family.clone()) + .collect::>(); + assert!( + families.iter().any(|family| family == "tool_result"), + "{families:?}" + ); + assert_eq!(observer.metrics().snapshot().policy_outcomes.completed, 1); +} + +#[tokio::test] +async fn nonblocking_flood_remains_bounded_when_journal_queue_is_saturated() { + let directory = temp_dir("flood"); + let flood_limits = ObservabilityLimits { + ring_events: 16, + subscriber_queue: 1, + max_subscribers: 1, + journal_queue: 1, + max_event_bytes: 4_096, + max_diagnostic_entries: 0, + }; + let (observer, runtime) = Observability::journaled( + flood_limits, + JournalConfig { + directory: directory.clone(), + max_segment_bytes: 4_096, + max_segments: 2, + max_total_bytes: 8_192, + sync_each_record: true, + }, + ) + .expect("journal observer"); + let _slow = observer.subscribe(None).expect("subscriber"); + for index in 0..10_000 { + observer + .record(draft(EventFamily::ToolProgress, &format!("flood-{index}"))) + .expect("bounded record"); + } + assert_eq!(observer.snapshot().len(), flood_limits.ring_events); + let snapshot = observer.metrics().snapshot(); + assert_eq!(snapshot.recorded_events, 10_000); + assert!(snapshot.dropped_journal_events > 0); + assert!(snapshot.dropped_subscriber_events > 0); + runtime.shutdown(&observer).await.expect("shutdown"); + fs::remove_dir_all(directory).expect("remove test directory"); +} diff --git a/crates/metacrate-grid-agent/tests/dependency_policy.rs b/crates/metacrate-grid-agent/tests/dependency_policy.rs index 905b032..72f71e7 100644 --- a/crates/metacrate-grid-agent/tests/dependency_policy.rs +++ b/crates/metacrate-grid-agent/tests/dependency_policy.rs @@ -48,7 +48,7 @@ fn runtime_source_has_no_subprocess_or_native_abi_escape_hatch() { let mut files = Vec::with_capacity(20); collect_rust_files(&source, &mut files); assert!( - files.len() <= 24, + files.len() <= 26, "source-file count needs a reviewed bound update" ); for path in files { diff --git a/docs/grid-agent-control-plane.md b/docs/grid-agent-control-plane.md index 4f1cb0d..e5cf865 100644 --- a/docs/grid-agent-control-plane.md +++ b/docs/grid-agent-control-plane.md @@ -48,11 +48,11 @@ The JSON request envelope is stable and versioned. For example: {"version":1,"request_id":"health-1","request":{"method":"health"}} ``` -Observers can call `health`, `runtime`, `list_sessions`, -`list_scheduled_jobs`, `list_pending_approvals`, `list_audit_events`, and -`subscribe_events`. Operators can additionally call `cancel_request`, -`pause_autonomy`, `resume_autonomy`, `cancel_action`, `decide_approval`, -`force_reconnect`, `expire_conversation`, `set_roaming_job`, +Observers can call `health`, `metrics`, `runtime`, `list_sessions`, +`list_scheduled_jobs`, `list_pending_approvals`, `list_audit_events`, +`list_observability_events`, and `subscribe_events`. Operators can additionally +call `cancel_request`, `pause_autonomy`, `resume_autonomy`, `cancel_action`, +`decide_approval`, `force_reconnect`, `expire_conversation`, `set_roaming_job`, `inject_operator_message`, and `graceful_shutdown`. Cancellation is a mutation and is operator-only. List requests use an opaque numeric cursor and a page size of 1 through 100. @@ -62,6 +62,9 @@ region and pose fields when known, behavior mode, control-queue utilization, and aggregate budget use. Conversation responses contain metadata only. Audit and approval responses exclude arguments, prompt contents, credentials, authorization headers, capability URLs, model reasoning, and filesystem data. +Unified events use pseudonymous correlation IDs and fixed-cardinality metrics; +their schema, journal, redaction, and replay rules are documented in +[`grid-agent-observability.md`](grid-agent-observability.md). `cancel_action` binds to the exact bounded action ID carried by behavior audit observations; it can cancel a queued or executing embodied action without preempting unrelated work. The built-in roaming job ID is `default-roaming`. diff --git a/docs/grid-agent-observability.md b/docs/grid-agent-observability.md new file mode 100644 index 0000000..2f6bafc --- /dev/null +++ b/docs/grid-agent-observability.md @@ -0,0 +1,118 @@ +# Grid-agent observability, audit, metrics, and replay + +The grid agent records externally observable decisions and outcomes without +claiming access to private model reasoning. The stable JSON schema version is +`1`. `StructuredEvent` is the transport-neutral representation used by the +in-memory ring, optional JSONL journal, control API, and replay reader. + +## Event contract + +Every event has `schema_version`, monotonic `event_id`, `unix_millis`, +`severity`, `component`, `family`, `origin`, correlation IDs, optional duration, +retry count, result/reason codes, redaction flags, bounded scalar fields, and +forward-compatible extension fields. Version 1 families cover lifecycle and +behavior transitions; inbound/outbound messages; conversation creation/expiry; +inference requests/results; model action summaries; tool proposals, +policy/approval decisions and tool execution; scheduled jobs; control commands; +shutdown; and diagnostic envelopes. + +One recorder assigns a total admission order. This preserves order for an +action and gives concurrent producers a deterministic recorded order, but does +not claim causal order between tasks before recorder admission. Correlation +values are stable SHA-256 pseudonyms, so an operator can join a timeline without +retaining original avatar, session, request, or action identifiers. Runtime tool +names are likewise pseudonymized. Readable state/reason/result values are +compile-time enumeration codes. + +Later readers ignore unknown top-level and `fields` keys. The Rust reader +retains both across a decode/encode cycle. + +```json +{"schema_version":1,"event_id":42,"unix_millis":1787025600000,"severity":"info","component":"policy","family":"tool_result","correlation":{"session_id":"id:83c1...","request_id":"id:bd10...","action_id":"id:11a0..."},"origin":"policy","duration_millis":18,"retry_count":0,"result_code":"completed","reason_code":"allowed","redaction_flags":["tool_arguments"],"fields":{"tool_calls":1}} +``` + +## Privacy boundary + +The event builder cannot accept free-form string fields. Message/prompt bodies, +model responses, authorization headers, API keys, grid passwords, capability +and asset URLs, inventory payloads, and hidden chain-of-thought have no runtime +event representation. Debug output omits field and extension values. Control +messages retain operation, role, and bounded metadata only. + +Diagnostic capture is explicit and bounded by `max_diagnostic_entries`. It +emits a warning envelope with part count, total bytes, and a digest; prompt and +response text is never retained. It carries `diagnostic_capture` and +`content_omitted` flags. Hashes and timing can still reveal equality, so enabling +capture remains a privacy decision. + +## Memory, subscribers, and JSONL journal + +`Observability::memory` creates a fixed ring. Eviction increments +`dropped_ring_events`. Each subscriber has a bounded queue. Resuming before +retained history reports an explicit missing interval. Slow subscribers are +disconnected and counted; producers never await them. + +`Observability::journaled` adds an optional bounded background writer. Producers +use `try_send`, so slow disks cannot block chat, lifecycle, or control handling. +Queue loss increments `dropped_journal_events`. The active file is +`events-current.jsonl.tmp`; complete segments are atomically renamed to +`events-NNN.jsonl`. Segment count and combined active/archive bytes are bounded, +with oldest complete segments removed first. `sync_each_record` trades latency +for durability. `ObservabilityRuntime::shutdown` drains, flushes, and syncs. + +A crash may tear only the final active record. Startup truncates that fragment +to the last newline. A single immutable reservation marker allocates the next +bounded ID range before events are admitted, so IDs observed by subscribers but +dropped by a saturated journal queue are not reused after restart. Malformed +complete records, oversized records, and exhausted ID ranges fail closed. + +```rust,no_run +use metacrate_grid_agent::{JournalConfig, Observability, ObservabilityLimits}; +use std::path::PathBuf; + +# async fn example() -> Result<(), Box> { +let (events, runtime) = Observability::journaled( + ObservabilityLimits::default(), + JournalConfig { + directory: PathBuf::from("data/grid-agent/audit"), + max_segment_bytes: 4 * 1024 * 1024, + max_segments: 8, + max_total_bytes: 32 * 1024 * 1024, + sync_each_record: false, + }, +)?; +runtime.shutdown(&events).await?; +# Ok(()) } +``` + +## Metrics and control views + +Metrics use fixed atomics with no caller-defined labels or external telemetry. +The snapshot contains readiness, reconnects, active sessions/tasks, queue use, +inference/tool latency count/sum/max, fixed inference/tool/policy outcome +buckets, rate-limit use, recorded events, and ring/subscriber/journal drop +counters. + +Control protocol v1 provides read-only `metrics` and paginated +`list_observability_events` requests in addition to the policy audit. Observer +and operator roles may read them; sensitive payloads cannot enter either view. + +## Deterministic replay + +`replay_journal` reads retained segments and the recovered active file with +explicit event-size/count limits. It rebuilds lifecycle and behavior state, +session counts, policy/tool outcomes, and per-action observable timelines. +Redaction flags set `unavailable_private_content`, making missing context clear. + +Replay is data-only: it has no grid backend, LLM client, policy gateway, or tool +executor, so it cannot resend messages or repeat mutations. Rotation means it +reconstructs the retained window, not already expired history. + +## Focused verification + +```sh +cargo test --locked -p metacrate-grid-agent --lib observability_tests +cargo test --locked -p metacrate-grid-agent --test dependency_policy +cargo clippy --locked -p metacrate-grid-agent --all-targets -- -D warnings +RUSTDOCFLAGS="-D warnings" cargo doc --locked -p metacrate-grid-agent --no-deps +```