feat(grid-agent): add structured observability and replay (#127)

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<ConversationStore>,
     policy: Arc<PolicyGateway>,
     audit: Arc<MemoryPolicyAudit>,
+    observability: Mutex<Option<Arc<Observability>>>,
     behavior: BehaviorIngress,
     commands: mpsc::Sender<RuntimeControlCommand>,
     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<Observability>) {
+        *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<Option<EventDraft>, 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<T>(request: &PageRequest, values: Vec<T>) -> Page<T> {
     let start = usize::try_from(request.cursor.unwrap_or(0))
         .unwrap_or(usize::MAX)