chore: sync public mirror from internal

.repository-projection.json

@@ -3,11 +3,11 @@
   "projection": "deixic-code",
   "projectionSchemaVersion": 1,
   "sourceRepository": "dx-corp/mono",
-  "sourceSha": "3b8d839ff70abae7d6aa87082079449a8d6eb43a",
+  "sourceSha": "6f8bca8d69be1b576e96428243157a5cb5c3cf35",
   "destinationRepository": "dx-corp/code",
-  "priorProjectedBase": "4690e79129c9b1874f9bce83c9b0e294894fd81d",
+  "priorProjectedBase": "873426648fb369290ceb39a5904738754ef0f46c",
   "definitionDigest": "064b6b9745d392dde277afd7dfdf028cff32b6178a8b9c99c518343219bab429",
   "toolDigest": "c8000985e1bd708bf1f8754b27cc1a2aad67af10ee23cf5317bd1fba670b4456",
-  "contentDigest": "f3cedd14d1124069f81086dd5bf7c8c7f53c4fbaebd3f6c20f51000dc13ffc44",
+  "contentDigest": "0a31fb4dca7df2bcda6576943dc71fb0a3ecc735c3e86b8caacc8536e0e6d190",
   "publicationEligible": true
 }

packages/local-host-rs/src/agent/mod.rs

@@ -667,97 +667,70 @@ pub(crate) use maestro_runtime::agent::managed_turn_lineage_id;
 /// Observe the same event stream delivered to the caller, preserving the
 /// canonical turn accounting that was attached to the actor before extraction.
 fn relay_runtime_events(
-    mut runtime_event_rx: tokio::sync::mpsc::UnboundedReceiver<FromAgent>,
+    runtime_event_rx: tokio::sync::mpsc::UnboundedReceiver<FromAgent>,
     config: &NativeAgentConfig,
     provider_name: String,
     telemetry_identity_scope: Arc<RwLock<Option<crate::telemetry::TelemetryIdentityScope>>>,
     telemetry_host: Option<maestro_runtime::agent::NativeExecutionHostHandle>,
 ) -> tokio::sync::mpsc::UnboundedReceiver<FromAgent> {
     let (consumer_event_tx, event_rx) = tokio::sync::mpsc::unbounded_channel();
-    let mut turn_tracker =
-        crate::telemetry::TurnTracker::new(crate::telemetry::TurnTrackerConfig {
+    let initial_identity_scope = telemetry_identity_scope
+        .read()
+        .expect("telemetry identity scope lock poisoned")
+        .clone();
+    let telemetry = crate::telemetry::TurnTelemetryRelay::spawn(
+        crate::telemetry::TurnTrackerConfig {
             // A runtime run ID is not a captured conversation ID. Missing
             // session context stays unattributed in the cloud projection.
             session_id: String::new(),
             sampling_config: crate::telemetry::TailSamplingConfig::from_env(),
-        });
-    turn_tracker.update_context(crate::telemetry::TurnTrackerContext {
-        model: Some(crate::telemetry::ModelInfo {
-            id: config.model.clone(),
-            provider: provider_name,
-            thinking_level: crate::telemetry::ThinkingLevel::Off,
-        }),
-        sandbox_mode: if config.sandbox_policy.is_some() {
-            crate::telemetry::SandboxMode::Local
-        } else {
-            crate::telemetry::SandboxMode::None
         },
-        approval_mode: match config.approval_mode {
-            ApprovalMode::Yolo => crate::telemetry::ApprovalMode::Auto,
-            ApprovalMode::Selective | ApprovalMode::Safe => crate::telemetry::ApprovalMode::Prompt,
-        },
-        mcp_servers: Vec::new(),
-        context_source_count: 0,
-        features: crate::telemetry::FeatureFlags {
-            safe_mode: config.approval_mode != ApprovalMode::Yolo,
-            ..Default::default()
+        crate::telemetry::TurnTrackerContext {
+            model: Some(crate::telemetry::ModelInfo {
+                id: config.model.clone(),
+                provider: provider_name,
+                thinking_level: crate::telemetry::ThinkingLevel::Off,
+            }),
+            sandbox_mode: if config.sandbox_policy.is_some() {
+                crate::telemetry::SandboxMode::Local
+            } else {
+                crate::telemetry::SandboxMode::None
+            },
+            approval_mode: match config.approval_mode {
+                ApprovalMode::Yolo => crate::telemetry::ApprovalMode::Auto,
+                ApprovalMode::Selective | ApprovalMode::Safe => {
+                    crate::telemetry::ApprovalMode::Prompt
+                }
+            },
+            mcp_servers: Vec::new(),
+            context_source_count: 0,
+            features: crate::telemetry::FeatureFlags {
+                safe_mode: config.approval_mode != ApprovalMode::Yolo,
+                ..Default::default()
+            },
+            identity_scope: initial_identity_scope,
         },
-        identity_scope: telemetry_identity_scope
-            .read()
-            .expect("telemetry identity scope lock poisoned")
-            .clone(),
-    });
+        telemetry_identity_scope,
+        telemetry_host,
+    );
+    relay_runtime_events_with_telemetry(runtime_event_rx, telemetry, consumer_event_tx);
+    event_rx
+}
+
+fn relay_runtime_events_with_telemetry(
+    mut runtime_event_rx: tokio::sync::mpsc::UnboundedReceiver<FromAgent>,
+    mut telemetry: crate::telemetry::TurnTelemetryRelay,
+    consumer_event_tx: tokio::sync::mpsc::UnboundedSender<FromAgent>,
+) {
     tokio::spawn(async move {
-        let mut journal = crate::telemetry::TurnJournal::open();
         while let Some(event) = runtime_event_rx.recv().await {
-            if matches!(
-                &event,
-                FromAgent::TurnStarted | FromAgent::ResponseStart { .. }
-                    | FromAgent::SideQuestionStart { .. }
-                    | FromAgent::OperationObservation { observation: maestro_runtime_contracts::operation_observation::OperationObservation::Admitted { .. } }
-            ) {
-                let session_id = match &telemetry_host {
-                    Some(host) => host.hook_session_id().await,
-                    None => None,
-                };
-                // Clear a previous session when the owner has none; the
-                // collector pins the current turn's identity at its start.
-                turn_tracker.set_session_id(session_id.unwrap_or_default());
-            }
-            if matches!(&event, FromAgent::ModelChanged { .. }) {
-                turn_tracker.set_identity_scope(
-                    telemetry_identity_scope
-                        .read()
-                        .expect("telemetry identity scope lock poisoned")
-                        .clone(),
-                );
-            }
-            let completed = turn_tracker.handle_event(&event);
-            if let Some(completed) = completed.as_ref() {
-                if let Some(journal) = &mut journal {
-                    journal.finish(completed);
-                } else {
-                    crate::telemetry::record_canonical_turn_event(completed);
-                }
-            }
-            if matches!(
-                &event,
-                FromAgent::TurnStarted
-                    | FromAgent::OperationObservation { .. }
-                    | FromAgent::SideQuestionStart { .. }
-                    | FromAgent::ResponseStart { .. }
-                    | FromAgent::ResponseEnd { .. }
-            ) {
-                if let Some(journal) = &mut journal {
-                    journal.observe(&turn_tracker.pending_snapshots());
-                }
-            }
+            let observation = telemetry.project(&event);
             // A detached consumer does not end the actor's turn. Keep draining
             // until the runtime exits so terminal telemetry is still recorded.
             let _ = consumer_event_tx.send(event);
+            telemetry.submit(observation);
         }
     });
-    event_rx
 }
 
 /// Bind a model transition before publishing its identity to turn telemetry.
@@ -817,6 +790,28 @@ mod identity_transition_tests {
             .unwrap();
     }
 
+    #[tokio::test]
+    async fn saturated_telemetry_never_delays_runtime_event_delivery() {
+        let (runtime_tx, runtime_rx) = tokio::sync::mpsc::unbounded_channel();
+        let (consumer_tx, mut consumer_rx) = tokio::sync::mpsc::unbounded_channel();
+        let (telemetry, _blocked_receiver) =
+            crate::telemetry::TurnTelemetryRelay::saturated_for_test();
+        super::relay_runtime_events_with_telemetry(runtime_rx, telemetry, consumer_tx);
+
+        runtime_tx
+            .send(super::FromAgent::ResponseStart {
+                response_id: "response".to_owned(),
+            })
+            .unwrap();
+
+        let delivered =
+            tokio::time::timeout(std::time::Duration::from_millis(100), consumer_rx.recv())
+                .await
+                .expect("a full telemetry queue must not delay the consumer")
+                .expect("runtime event relay closed unexpectedly");
+        assert!(matches!(delivered, super::FromAgent::ResponseStart { .. }));
+    }
+
     fn scope(org: &str, workspace: &str) -> Option<TelemetryIdentityScope> {
         TelemetryIdentityScope::new(org, Some(workspace))
     }

packages/local-host-rs/src/telemetry/mod.rs

@@ -41,8 +41,10 @@
 mod customer_export;
 mod onboarding;
 mod operation;
+mod relay;
 mod staged_rollout;
 mod tracker;
+pub(crate) use relay::TurnTelemetryRelay;
 pub(crate) use staged_rollout::TurnJournal;
 mod visibility;
 mod wide_events;