[pull] main from xintaofei:main

src-tauri/src/acp/lifecycle.rs

@@ -27,6 +27,7 @@ use crate::acp::types::{AcpEvent, ConnectionStatus, EventEnvelope};
 use crate::db::entities::conversation::ConversationStatus;
 use crate::db::error::DbError;
 use crate::db::service::conversation_service;
+use crate::logging::throttle::{LagLogThrottle, LAG_LOG_WINDOW};
 use crate::models::AgentType;
 use crate::web::event_bridge::{emit_with_state, EventEmitter};
 use tokio::sync::RwLock;
@@ -1512,6 +1513,7 @@ pub fn lifecycle_subscriber_task(
         // connection's first relevant event and torn down after a terminal
         // event by dropping the sender (worker drains its queue and exits).
         let mut workers: HashMap<String, mpsc::Sender<Arc<EventEnvelope>>> = HashMap::new();
+        let mut lag_throttle = LagLogThrottle::new(LAG_LOG_WINDOW);
         loop {
             match rx.recv().await {
                 Ok(envelope_arc) => {
@@ -1592,12 +1594,19 @@ pub fn lifecycle_subscriber_task(
                     // Lagged at the bus level. Now that the dispatcher
                     // filters and only blocks on the rare relevant events,
                     // this should only fire under genuine emit-rate spikes
-                    // exceeding the 4096 broadcast capacity.
-                    tracing::warn!(
-                        "[lifecycle][WARN] internal bus lagged, dropped {skipped} events \
-                         (emit rate exceeded broadcast capacity)"
-                    );
+                    // exceeding the 4096 broadcast capacity. The metric is the
+                    // authoritative loss record (unthrottled); the log line is
+                    // throttled to at most one per window.
                     metrics.lagged_count.fetch_add(skipped, Ordering::Relaxed);
+                    if let Some(s) = lag_throttle.record(skipped) {
+                        tracing::warn!(
+                            "[lifecycle][WARN] internal bus lagged: dropped {} events across \
+                             {} occurrence(s) in the last {}s (emit rate exceeded broadcast capacity)",
+                            s.dropped,
+                            s.occurrences,
+                            LAG_LOG_WINDOW.as_secs()
+                        );
+                    }
                 }
                 Err(broadcast::error::RecvError::Closed) => {
                     tracing::info!("[lifecycle] internal bus closed; dispatcher exiting");

src-tauri/src/automation/engine.rs

@@ -34,6 +34,7 @@ use crate::commands::folders::{
 use crate::db::entities::conversation::{self, ConversationStatus};
 use crate::db::service::automation_service;
 use crate::db::AppDatabase;
+use crate::logging::throttle::{LagLogThrottle, LAG_LOG_WINDOW};
 use crate::models::{
     AgentType, AutomationConfig, AutomationInfo, AutomationRunStatus, IsolationMode,
 };
@@ -231,14 +232,23 @@ pub async fn run_automation_engine(engine: Arc<AutomationEngine>) {
     let mut reconcile = delay_interval(RECONCILE_INTERVAL_SECS);
     let mut schedule = delay_interval(SCHEDULER_INTERVAL_SECS);
     let mut prune = delay_interval(PRUNE_INTERVAL_SECS);
+    let mut lag_throttle = LagLogThrottle::new(LAG_LOG_WINDOW);
 
     loop {
         tokio::select! {
             ev = rx.recv() => match ev {
                 Ok(env) => engine.on_event(&env).await,
                 // Dropped events under lag — the reconcile backstop recovers them.
                 Err(RecvError::Lagged(n)) => {
-                    tracing::warn!("[automation] event bus lagged {n}; reconcile will recover");
+                    if let Some(s) = lag_throttle.record(n) {
+                        tracing::warn!(
+                            "[automation] event bus lagged: dropped {} events across \
+                             {} occurrence(s) in the last {}s; reconcile will recover",
+                            s.dropped,
+                            s.occurrences,
+                            LAG_LOG_WINDOW.as_secs()
+                        );
+                    }
                 }
                 Err(RecvError::Closed) => break,
             },

src-tauri/src/chat_channel/event_subscriber.rs

@@ -17,6 +17,7 @@ use crate::acp::types::{AcpEvent, EventEnvelope};
 use crate::db::service::{
     app_metadata_service, chat_channel_message_log_service, chat_channel_service,
 };
+use crate::logging::throttle::{LagLogThrottle, LAG_LOG_WINDOW};
 
 /// Minimum interval between pushes for the same event type per channel (debounce).
 const DEBOUNCE_SECS: u64 = 5;
@@ -213,13 +214,22 @@ pub fn spawn_event_subscriber(
         let mut config = EventConfigCache::new();
         // One reqwest client, reused (and cheaply cloned) for every webhook POST.
         let webhook_client = super::webhook::make_webhook_client();
+        let mut lag_throttle = LagLogThrottle::new(LAG_LOG_WINDOW);
 
         loop {
             let envelope_arc = match rx.recv().await {
                 Ok(e) => e,
                 Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => {
-                    tracing::warn!("[ChatChannel] event subscriber lagged by {n} messages");
                     metrics.lagged_count.fetch_add(n, Ordering::Relaxed);
+                    if let Some(s) = lag_throttle.record(n) {
+                        tracing::warn!(
+                            "[ChatChannel] event subscriber lagged: dropped {} messages across \
+                             {} occurrence(s) in the last {}s",
+                            s.dropped,
+                            s.occurrences,
+                            LAG_LOG_WINDOW.as_secs()
+                        );
+                    }
                     continue;
                 }
                 Err(tokio::sync::broadcast::error::RecvError::Closed) => {

src-tauri/src/chat_channel/session_event_subscriber.rs

@@ -19,6 +19,7 @@ use crate::acp::types::{
 use crate::db::service::{
     app_metadata_service, conversation_service, sender_context_service, thread_binding_service,
 };
+use crate::logging::throttle::{LagLogThrottle, LAG_LOG_WINDOW};
 
 use super::manager::ChatChannelManager;
 
@@ -41,15 +42,24 @@ pub fn spawn_session_event_subscriber(
 
     tokio::spawn(async move {
         let mut last_heartbeat = Instant::now();
+        let mut lag_throttle = LagLogThrottle::new(LAG_LOG_WINDOW);
 
         loop {
             tokio::select! {
                 result = rx.recv() => {
                     let envelope_arc = match result {
                         Ok(e) => e,
                         Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => {
-                            tracing::warn!("[SessionEventSub] lagged {n} events");
                             metrics.lagged_count.fetch_add(n, Ordering::Relaxed);
+                            if let Some(s) = lag_throttle.record(n) {
+                                tracing::warn!(
+                                    "[SessionEventSub] lagged: dropped {} events across \
+                                     {} occurrence(s) in the last {}s",
+                                    s.dropped,
+                                    s.occurrences,
+                                    LAG_LOG_WINDOW.as_secs()
+                                );
+                            }
                             continue;
                         }
                         Err(_) => break,

src-tauri/src/logging/init.rs

@@ -39,12 +39,35 @@ pub struct LogGuard {
     _guard: Option<WorkerGuard>,
 }
 
+/// Standing per-target backstops appended to EVERY constructed filter — the
+/// default/configured level, a persisted level, AND an explicit `RUST_LOG` /
+/// `CODEG_LOG` override. Two entries, both appended last so they win over a
+/// broader global level:
+///
+/// - `kill_tree=warn` — the `kill_tree` crate emits one DEBUG line per inspected
+///   PID (~1500 on macOS, mostly EPERM on SIP-protected processes) on every
+///   `kill_tree()` call. Under a global `debug` level that firehose grew a
+///   server's log file to 217GB. We never want it: real kill failures are
+///   already surfaced at our own call sites via `error!`. Pinned here so even
+///   `RUST_LOG=debug` can't re-open it.
+/// - `codeg_lib::logging=off` — the logging stack must not log about itself
+///   (cross-thread feedback-loop backstop; the layer's thread-local guard
+///   handles the same-thread case).
+const TARGET_BACKSTOPS: &str = "kill_tree=warn,codeg_lib::logging=off";
+
+/// Directive string for an explicit env-override level `s`, with the standing
+/// [`TARGET_BACKSTOPS`] appended. Extracted so the backstop application on the
+/// env-override path is unit-testable without installing a global subscriber.
+fn env_override_directives(s: &str) -> String {
+    format!("{s},{TARGET_BACKSTOPS}")
+}
+
 /// Build the `EnvFilter` for the full settings: the global level followed by
-/// each non-empty per-target override (`target=level`), then always excluding
-/// the logging module's own target as a cross-thread feedback-loop backstop
-/// (the layer's thread-local guard handles the same-thread case). The backstop
-/// is appended last so it wins. `parse_lossy` silently drops any malformed
-/// target directive — the UI constrains the input to avoid that.
+/// each non-empty per-target override (`target=level`), then always appending
+/// [`TARGET_BACKSTOPS`] (the `kill_tree` clamp + the logging module's own target
+/// off). The backstops are appended last so they win. `parse_lossy` silently
+/// drops any malformed target directive — the UI constrains the input to avoid
+/// that.
 pub fn build_env_filter(settings: &LogSettings) -> EnvFilter {
     let mut directives = settings.level.directive().to_string();
     for t in &settings.targets {
@@ -56,7 +79,8 @@ pub fn build_env_filter(settings: &LogSettings) -> EnvFilter {
             directives.push_str(&format!(",{}={}", target, t.level.directive()));
         }
     }
-    directives.push_str(",codeg_lib::logging=off");
+    directives.push(',');
+    directives.push_str(TARGET_BACKSTOPS);
     EnvFilter::builder().parse_lossy(directives)
 }
 
@@ -159,7 +183,9 @@ fn build_subscriber(
     // same `env_level_override` precedence as `env_level_is_set`, so a level
     // applied here is exactly the one the UI reports as env-locked.
     let initial_filter = match env_level_override() {
-        Some(s) => EnvFilter::builder().parse_lossy(format!("{s},codeg_lib::logging=off")),
+        // Even an explicit env override gets the standing target backstops, so
+        // `RUST_LOG=debug` still can't re-open the kill_tree per-PID firehose.
+        Some(s) => EnvFilter::builder().parse_lossy(env_override_directives(&s)),
         // Phase 1 has no DB yet, so no persisted per-target overrides; just the
         // default level. Phase 2 (apply_persisted_level) rebuilds with targets.
         None => build_env_filter(&LogSettings {
@@ -331,6 +357,45 @@ mod tests {
         assert!(!rendered.contains("codeg_lib::logging=trace"), "{rendered}");
     }
 
+    #[test]
+    fn build_env_filter_pins_noisy_third_party_targets() {
+        // Every constructed filter clamps `kill_tree` to warn — its per-PID
+        // DEBUG firehose once grew a server log to 217GB — and keeps the
+        // logging module's own target off.
+        let rendered = build_env_filter(&LogSettings {
+            level: LogLevel::Info,
+            targets: Vec::new(),
+        })
+        .to_string();
+        assert!(
+            rendered.contains("kill_tree=warn"),
+            "kill_tree must be pinned to warn: {rendered}"
+        );
+        assert!(
+            rendered.contains("codeg_lib::logging=off"),
+            "logging backstop must remain: {rendered}"
+        );
+    }
+
+    #[test]
+    fn env_override_appends_backstops_so_debug_cannot_reopen_kill_tree() {
+        // The explicit RUST_LOG/CODEG_LOG path bypasses build_env_filter, so it
+        // must apply the same backstops itself — otherwise `RUST_LOG=debug`
+        // re-opens the kill_tree per-PID firehose.
+        assert_eq!(
+            env_override_directives("debug"),
+            "debug,kill_tree=warn,codeg_lib::logging=off"
+        );
+        // And the parsed filter still carries the clamp under a global debug.
+        let rendered = EnvFilter::builder()
+            .parse_lossy(env_override_directives("debug"))
+            .to_string();
+        assert!(
+            rendered.contains("kill_tree=warn"),
+            "kill_tree clamp must survive a global debug override: {rendered}"
+        );
+    }
+
     #[test]
     fn is_valid_target_grammar() {
         assert!(is_valid_target("codeg_lib::acp"));

src-tauri/src/logging/mod.rs

@@ -15,6 +15,7 @@
 pub mod hub;
 pub mod init;
 pub mod layer;
+pub mod throttle;
 
 use serde::{Deserialize, Serialize};