chore: sync public mirror from internal

.repository-projection.json

@@ -3,11 +3,11 @@
   "projection": "deixic-code",
   "projectionSchemaVersion": 1,
   "sourceRepository": "dx-corp/mono",
-  "sourceSha": "028183748c33b32f658596ed0f96ef1c60424961",
+  "sourceSha": "43eb0edd715c489c97e0fb1fe22f52a44dbcfd95",
   "destinationRepository": "dx-corp/code",
-  "priorProjectedBase": "a2b3e4a7f7f38bf2b195b8402bf5b804450a0486",
+  "priorProjectedBase": "44c5b0b9c35bc88ea604bae376b1a1b9c4d0d7a5",
   "definitionDigest": "82936441c776e3e8edb5d215a75007ec9714a233f489d460075d79d5ef5ba32f",
   "toolDigest": "45502ff0478e541d02f34cc39ac332935f8d3c0030fa221afe5bcc3b5e51f88e",
-  "contentDigest": "e6e93ea0769f4bd63e5aea305914b79869813d5138b1161347463c2055779d3d",
+  "contentDigest": "fd67fb0c9b781bf5e60fd477ccb05c531bb038c0c4bcb52ccf50947d8c5bcd1e",
   "publicationEligible": true
 }

packages/local-host-rs/src/hosted_runner.rs

@@ -89,6 +89,10 @@ pub(super) const HOSTED_RUNNER_DRAIN_FINALIZATION_PENDING_STATUS: &str =
 const DEFAULT_HEARTBEAT_INTERVAL_MS: u64 = 15_000;
 const CONNECTION_IDLE_MS: i64 = (DEFAULT_HEARTBEAT_INTERVAL_MS as i64) * 3;
 const MAINTENANCE_PUMP_INTERVAL: Duration = Duration::from_millis(100);
+// Connected transports notify the pump after enqueueing every event. Keep a
+// slower safety drain for missed notifications without waking 10 times a
+// second throughout an otherwise idle hosted session.
+const NOTIFIED_MAINTENANCE_PUMP_INTERVAL: Duration = Duration::from_secs(5);
 const THREAD_PERSISTENCE_RECOVERY_RETRY_DELAY: Duration = Duration::from_millis(100);
 const MAX_EVENTS: usize = 1024;
 // Response retries are short-lived transport retries; retain enough completed
@@ -579,6 +583,12 @@ pub trait HostedRunnerHeadlessMessageExecutor: Send + Sync {
         Ok(None)
     }
 
+    /// Pending response acknowledgements and failed ledger writes need the
+    /// short maintenance cadence even when transport events are notified.
+    fn has_pending_maintenance_work(&self) -> bool {
+        false
+    }
+
     /// Report whether a runtime that was previously connected has lost its
     /// transport. Hosted mode owns one child generation, so the event pump
     /// treats this as terminal while leaving `drain` available for export.
@@ -1135,6 +1145,24 @@ impl HostedRunnerHeadlessMessageExecutor for AgentSupervisorHostedRunnerMessageE
         Ok(supervisor.event_notification())
     }
 
+    fn has_pending_maintenance_work(&self) -> bool {
+        !self
+            .queued_responses
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner)
+            .is_empty()
+            || !self
+                .queued_unkeyed_responses
+                .lock()
+                .unwrap_or_else(std::sync::PoisonError::into_inner)
+                .is_empty()
+            || !self
+                .memory_completed_responses
+                .lock()
+                .unwrap_or_else(std::sync::PoisonError::into_inner)
+                .is_empty()
+    }
+
     fn disconnected_after_ready(&self) -> Result<bool, HostedRunnerError> {
         let supervisor = self
             .supervisor
@@ -1982,40 +2010,50 @@ async fn pump_agent_events(
     cancelled: CancellationToken,
     maintenance_interval: Duration,
 ) {
-    let mut interval = tokio::time::interval(maintenance_interval);
+    let mut first_tick = true;
     loop {
-        let should_pump = tokio::select! {
-            () = cancelled.cancelled() => break,
-            _ = interval.tick() => true,
-            result = wait_for_event_notification(&shared) => {
-                if let Err(error) = result {
-                    shared.publish_runtime_error("event_pump_failed", error);
-                    break;
-                }
-                true
+        let notification = match shared.message_executor.event_notification() {
+            Ok(notification) => notification,
+            Err(error) => {
+                shared.publish_runtime_error("event_pump_failed", error);
+                break;
             }
         };
+        let can_use_safety_tick = notification.is_some()
+            && !shared
+                .thread_persistence_retry_pending
+                .load(Ordering::Acquire)
+            && !shared.message_executor.has_pending_maintenance_work();
+        let tick_interval = if first_tick {
+            Duration::ZERO
+        } else if can_use_safety_tick {
+            NOTIFIED_MAINTENANCE_PUMP_INTERVAL.max(maintenance_interval)
+        } else {
+            maintenance_interval
+        };
+        first_tick = false;
+        tokio::select! {
+            () = cancelled.cancelled() => break,
+            () = tokio::time::sleep(tick_interval) => {},
+            () = async {
+                if let Some(notification) = notification {
+                    notification.notified().await;
+                } else {
+                    std::future::pending::<()>().await;
+                }
+            } => {},
+        }
         let lifecycle = shared.mutation_lifecycle.clone();
         let _lifecycle = tokio::select! {
             () = cancelled.cancelled() => break,
             lifecycle = lifecycle.lock() => lifecycle,
         };
-        if should_pump && matches!(pump_tick(&shared), PumpTick::Stop) {
+        if matches!(pump_tick(&shared), PumpTick::Stop) {
             break;
         }
     }
 }
 
-async fn wait_for_event_notification(shared: &SharedRunner) -> Result<(), HostedRunnerError> {
-    let notification = shared.message_executor.event_notification()?;
-    if let Some(notification) = notification {
-        notification.notified().await;
-        Ok(())
-    } else {
-        std::future::pending().await
-    }
-}
-
 #[derive(Debug, PartialEq, Eq)]
 enum PumpTick {
     Continue,

packages/local-host-rs/src/hosted_runner/tests.rs

@@ -2592,13 +2592,19 @@ impl HostedRunnerHeadlessMessageExecutor for PumpOnlyRuntimeExecutor {
 struct NotifyingPumpRuntimeExecutor {
     queued: Arc<Mutex<Vec<FromAgentMessage>>>,
     notification: Arc<Notify>,
+    drain_count: AtomicUsize,
+    expose_notification: bool,
+    pending_maintenance: AtomicBool,
 }
 
 impl Default for NotifyingPumpRuntimeExecutor {
     fn default() -> Self {
         Self {
             queued: Arc::new(Mutex::new(Vec::new())),
             notification: Arc::new(Notify::new()),
+            drain_count: AtomicUsize::new(0),
+            expose_notification: true,
+            pending_maintenance: AtomicBool::new(false),
         }
     }
 }
@@ -2631,6 +2637,7 @@ impl HostedRunnerHeadlessMessageExecutor for NotifyingPumpRuntimeExecutor {
     }
 
     fn drain(&self) -> Result<HostedRunnerDrainResult, HostedRunnerError> {
+        self.drain_count.fetch_add(1, Ordering::Relaxed);
         Ok(HostedRunnerDrainResult {
             messages: std::mem::take(&mut *self.queued.lock().expect("queued ready event")),
             consumed_response_keys: Vec::new(),
@@ -2639,7 +2646,13 @@ impl HostedRunnerHeadlessMessageExecutor for NotifyingPumpRuntimeExecutor {
     }
 
     fn event_notification(&self) -> Result<Option<Arc<Notify>>, HostedRunnerError> {
-        Ok(Some(Arc::clone(&self.notification)))
+        Ok(self
+            .expose_notification
+            .then(|| Arc::clone(&self.notification)))
+    }
+
+    fn has_pending_maintenance_work(&self) -> bool {
+        self.pending_maintenance.load(Ordering::Relaxed)
     }
 }
 
@@ -7776,6 +7789,160 @@ async fn transport_notification_publishes_ready_before_maintenance_tick() {
     handle.shutdown().await;
 }
 
+#[tokio::test(start_paused = true)]
+async fn notified_event_pump_has_a_bounded_idle_drain_rate() {
+    let workspace = tempdir().expect("workspace");
+    let executor = Arc::new(NotifyingPumpRuntimeExecutor::default());
+    let shared = SharedRunner::new_with_message_executor_and_restore(
+        test_config(workspace.path().to_path_buf()),
+        executor.clone(),
+        None,
+    );
+    let cancelled = CancellationToken::new();
+    let pump = tokio::spawn(pump_agent_events(
+        shared,
+        cancelled.clone(),
+        MAINTENANCE_PUMP_INTERVAL,
+    ));
+    tokio::task::yield_now().await;
+    let initial_drains = executor.drain_count.load(Ordering::Relaxed);
+    assert_eq!(initial_drains, 1, "the pump must drain on startup");
+
+    // Advance in small steps so Tokio cannot collapse a missed interval into
+    // one callback. This measures the work a real idle minute would schedule.
+    for _ in 0..600 {
+        tokio::time::advance(Duration::from_millis(100)).await;
+        tokio::task::yield_now().await;
+    }
+    let idle_drains = executor.drain_count.load(Ordering::Relaxed) - initial_drains;
+    assert!(
+        idle_drains <= 12,
+        "notified transport drained {idle_drains} times in one idle minute"
+    );
+
+    executor.queue_ready_event();
+    tokio::task::yield_now().await;
+    assert_eq!(
+        executor.drain_count.load(Ordering::Relaxed),
+        initial_drains + idle_drains + 1,
+        "a real event must wake the pump without waiting for maintenance"
+    );
+    cancelled.cancel();
+    pump.await.expect("pump task");
+}
+
+#[tokio::test(start_paused = true)]
+async fn executor_without_notification_keeps_short_maintenance_fallback() {
+    let workspace = tempdir().expect("workspace");
+    let executor = Arc::new(NotifyingPumpRuntimeExecutor {
+        expose_notification: false,
+        ..Default::default()
+    });
+    let shared = SharedRunner::new_with_message_executor_and_restore(
+        test_config(workspace.path().to_path_buf()),
+        executor.clone(),
+        None,
+    );
+    let cancelled = CancellationToken::new();
+    let pump = tokio::spawn(pump_agent_events(
+        shared,
+        cancelled.clone(),
+        MAINTENANCE_PUMP_INTERVAL,
+    ));
+    tokio::task::yield_now().await;
+    let initial_drains = executor.drain_count.load(Ordering::Relaxed);
+    assert_eq!(initial_drains, 1);
+    for _ in 0..10 {
+        tokio::time::advance(Duration::from_millis(100)).await;
+        tokio::task::yield_now().await;
+    }
+    assert_eq!(
+        executor.drain_count.load(Ordering::Relaxed),
+        initial_drains + 10
+    );
+    cancelled.cancel();
+    pump.await.expect("pump task");
+}
+
+#[tokio::test(start_paused = true)]
+async fn notified_event_pump_keeps_short_retry_while_work_is_pending() {
+    let workspace = tempdir().expect("workspace");
+    let executor = Arc::new(NotifyingPumpRuntimeExecutor::default());
+    executor.pending_maintenance.store(true, Ordering::Relaxed);
+    let shared = SharedRunner::new_with_message_executor_and_restore(
+        test_config(workspace.path().to_path_buf()),
+        executor.clone(),
+        None,
+    );
+    let cancelled = CancellationToken::new();
+    let pump = tokio::spawn(pump_agent_events(
+        shared,
+        cancelled.clone(),
+        MAINTENANCE_PUMP_INTERVAL,
+    ));
+    tokio::task::yield_now().await;
+    let initial_drains = executor.drain_count.load(Ordering::Relaxed);
+    assert_eq!(initial_drains, 1);
+    for _ in 0..10 {
+        tokio::time::advance(Duration::from_millis(100)).await;
+        tokio::task::yield_now().await;
+    }
+    assert_eq!(
+        executor.drain_count.load(Ordering::Relaxed),
+        initial_drains + 10,
+        "pending acknowledgements and persistence retries need the 100 ms cadence"
+    );
+    cancelled.cancel();
+    pump.await.expect("pump task");
+}
+
+#[cfg(target_os = "linux")]
+#[tokio::test]
+#[ignore = "manual one-minute process CPU sample; use --ignored --nocapture"]
+async fn notified_event_pump_idle_cpu_probe() {
+    fn process_cpu() -> Duration {
+        let mut clock = std::mem::MaybeUninit::<libc::timespec>::uninit();
+        // SAFETY: clock points to writable timespec storage and the clock ID
+        // is a Linux process CPU clock.
+        assert_eq!(
+            unsafe { libc::clock_gettime(libc::CLOCK_PROCESS_CPUTIME_ID, clock.as_mut_ptr()) },
+            0,
+            "read process CPU clock"
+        );
+        // SAFETY: a successful clock_gettime initialized the whole timespec.
+        let clock = unsafe { clock.assume_init() };
+        Duration::new(clock.tv_sec as u64, clock.tv_nsec as u32)
+    }
+
+    let workspace = tempdir().expect("workspace");
+    let executor = Arc::new(NotifyingPumpRuntimeExecutor::default());
+    let shared = SharedRunner::new_with_message_executor_and_restore(
+        test_config(workspace.path().to_path_buf()),
+        executor.clone(),
+        None,
+    );
+    let cancelled = CancellationToken::new();
+    let pump = tokio::spawn(pump_agent_events(
+        shared,
+        cancelled.clone(),
+        MAINTENANCE_PUMP_INTERVAL,
+    ));
+    tokio::task::yield_now().await;
+    let before_drains = executor.drain_count.load(Ordering::Relaxed);
+    let before_cpu = process_cpu();
+    tokio::time::sleep(Duration::from_mins(1)).await;
+    let cpu = process_cpu()
+        .checked_sub(before_cpu)
+        .expect("process CPU clock is monotonic");
+    let drains = executor.drain_count.load(Ordering::Relaxed) - before_drains;
+    println!(
+        "idle_60s_process_cpu_ms={} drain_calls={drains}",
+        cpu.as_millis()
+    );
+    cancelled.cancel();
+    pump.await.expect("pump task");
+}
+
 #[tokio::test]
 async fn event_pump_publishes_agent_events_without_a_follow_up_http_request() {
     let workspace = tempdir().expect("workspace");