fix(runtime): wire hooks/cache/retry, session locking, MCP timeouts, cron injection

Build the hook executor from settings and enable prompt caching and a
retry policy at runtime init instead of hardcoding them off. Keep an
authoritative conversation snapshot so mid-query reads do not see an
empty shell, and route permission asks through per-request oneshots.
Own session grants in memory under a per-session file lock instead of
re-reading the file. Bound MCP stdio/ws requests with a 30s timeout and
drain pending requests when a server dies. Drain fired cron prompts on
an idle tick and inject them as user turns.

Author: lingcoder

crates/agents/src/prompt/git_context.rs

@@ -267,7 +267,6 @@ mod tests {
             .output()
             .unwrap();
 
-        // Create a commit
         fs::write(path.join("file.txt"), "content").unwrap();
         Command::new("git")
             .args(["add", "."])

crates/agents/src/runtime.rs

@@ -453,11 +453,7 @@ fn extract_topic(msg: &Message) -> Option<String> {
 
 /// Truncate a line to max length, adding "..." if truncated.
 fn truncate_line(s: &str, max: usize) -> String {
-    if s.len() <= max {
-        s.to_string()
-    } else {
-        format!("{}...", &s[..max.saturating_sub(3)])
-    }
+    crab_utils::text::truncate_chars_within(s, max, "...")
 }
 
 #[cfg(test)]

crates/agents/src/summarizer.rs

@@ -356,18 +356,25 @@ impl WorkerPool {
 
 /// Truncate a string to `max_len` chars, appending "…" if truncated.
 fn truncate_preview(s: &str, max_len: usize) -> String {
-    if s.len() <= max_len {
-        s.to_string()
-    } else {
-        format!("{}…", &s[..max_len])
-    }
+    crab_utils::text::truncate_chars(s, max_len, "…")
 }
 
 #[cfg(test)]
 mod tests {
     use super::*;
     use crab_session::Conversation;
 
+    #[test]
+    fn truncate_preview_multibyte_does_not_panic() {
+        // The previous byte-slice implementation panicked when `max_len`
+        // landed inside a multi-byte codepoint.
+        let prompt = "实现一个跨平台的终端用户界面组件".repeat(10);
+        let out = truncate_preview(&prompt, 80);
+        assert_eq!(out.chars().count(), 81);
+        assert!(out.ends_with('…'));
+        assert_eq!(truncate_preview("短", 80), "短");
+    }
+
     #[test]
     fn coordinator_creation() {
         let coord = WorkerPool::new("main".into(), "Main Agent".into());

crates/agents/src/teams/worker_pool.rs

@@ -25,7 +25,6 @@ use crab_tools::builtin::glob::GLOB_TOOL_NAME;
 use crab_tools::builtin::grep::GREP_TOOL_NAME;
 use crab_tools::builtin::notebook::NOTEBOOK_EDIT_TOOL_NAME;
 use crab_tools::builtin::read::READ_TOOL_NAME;
-use crab_tools::builtin::remote_trigger::REMOTE_TRIGGER_TOOL_NAME;
 use crab_tools::builtin::task::{
     TASK_CREATE_TOOL_NAME, TASK_GET_TOOL_NAME, TASK_LIST_TOOL_NAME, TASK_OUTPUT_TOOL_NAME,
     TASK_STOP_TOOL_NAME, TASK_UPDATE_TOOL_NAME,
@@ -483,7 +482,6 @@ fn default_registry_includes_all_builtin_tools() {
         CRON_CREATE_TOOL_NAME,
         CRON_DELETE_TOOL_NAME,
         CRON_LIST_TOOL_NAME,
-        REMOTE_TRIGGER_TOOL_NAME,
     ];
 
     for name in &expected {

crates/agents/tests/integration_m7b.rs

@@ -468,6 +468,9 @@ pub async fn run(cli: &Cli, resume_session_id: Option<String>) -> anyhow::Result
                 backend,
                 skill_dirs,
                 mcp_servers: settings.mcp_servers.clone(),
+                hooks: settings.hooks.clone(),
+                disable_hooks: cli.bare || settings.disable_all_hooks.unwrap_or(false),
+                cache_enabled: true,
                 settings_warnings,
                 prefill: cli.prefill.clone(),
                 open_resume_picker,

crates/cli/src/agent.rs

@@ -196,8 +196,12 @@ mod tests {
             request_id: None,
             is_confirmation: false,
         };
+        // `approve(Null)` still carries `Some(data)`, so it passes the
+        // presence check but fails schema validation against an object type.
         let resp = ElicitationResponse::approve(serde_json::Value::Null);
-        // Data is Null which is still Some; let's test with confirm
+        let err = validate_response(&req, &resp).unwrap_err();
+        assert!(err.contains("does not conform"));
+
         let resp2 = ElicitationResponse {
             approved: true,
             data: None,

crates/mcp/src/elicitation.rs

@@ -2,6 +2,7 @@ use std::collections::HashMap;
 use std::future::Future;
 use std::pin::Pin;
 use std::sync::Arc;
+use std::time::Duration;
 
 use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
 use tokio::process::{Child, Command};
@@ -10,6 +11,10 @@ use tokio::sync::{Mutex, oneshot};
 use crate::protocol::{JsonRpcNotification, JsonRpcRequest, JsonRpcResponse};
 use crate::transport::Transport;
 
+/// Per-request timeout — a request that gets no response within this window
+/// fails rather than hanging forever (e.g. the server died mid-request).
+const REQUEST_TIMEOUT: Duration = Duration::from_secs(30);
+
 /// Stdin/stdout transport for MCP servers launched as child processes.
 ///
 /// The MCP stdio protocol frames messages as `Content-Length: N\r\n\r\n{json}`
@@ -72,18 +77,23 @@ impl StdioTransport {
             loop {
                 match read_message(&mut reader).await {
                     Ok(Some(data)) => {
-                        // Try to parse as a response (has "id" field)
+                        // A response carries an "id"; a notification does not.
                         if let Ok(resp) = serde_json::from_str::<JsonRpcResponse>(&data) {
                             let mut map = pending_clone.lock().await;
                             if let Some(tx) = map.remove(&resp.id) {
                                 let _ = tx.send(resp);
                             }
+                        } else if let Ok(notif) = serde_json::from_str::<JsonRpcNotification>(&data)
+                        {
+                            // No consumer surface for stdio server notifications
+                            // yet; log so they're observable but not lost silently.
+                            tracing::debug!(
+                                method = notif.method,
+                                "received MCP server notification (no consumer wired)"
+                            );
                         }
-                        // Notifications from server are silently dropped for now.
-                        // TODO: emit server notifications through a channel.
                     }
                     Ok(None) => {
-                        // EOF — server process closed stdout
                         tracing::debug!("MCP server stdout closed");
                         break;
                     }
@@ -93,6 +103,10 @@ impl StdioTransport {
                     }
                 }
             }
+            // Reader is exiting (EOF or error): drop every pending sender so
+            // in-flight `rx.await` callers get an error promptly instead of
+            // hanging forever waiting on a dead server.
+            pending_clone.lock().await.clear();
         });
 
         Ok(Self {
@@ -142,10 +156,20 @@ impl Transport for StdioTransport {
             tracing::debug!(method = %req.method, id, "sending MCP request");
             self.write_message(&json).await?;
 
-            // Wait for the response from the reader task.
-            rx.await.map_err(|_| {
-                crab_core::Error::Other("MCP server closed connection before responding".into())
-            })
+            // Wait for the response, bounded by REQUEST_TIMEOUT so a dead or
+            // wedged server can't hang the caller indefinitely.
+            match tokio::time::timeout(REQUEST_TIMEOUT, rx).await {
+                Ok(Ok(resp)) => Ok(resp),
+                Ok(Err(_)) => Err(crab_core::Error::Other(
+                    "MCP server closed connection before responding".into(),
+                )),
+                Err(_) => {
+                    self.pending.lock().await.remove(&id);
+                    Err(crab_core::Error::Other(format!(
+                        "MCP server did not respond within {REQUEST_TIMEOUT:?}"
+                    )))
+                }
+            }
         })
     }
 
@@ -258,4 +282,46 @@ mod tests {
         let msg = read_message(&mut reader).await.unwrap().unwrap();
         assert_eq!(msg, "{}");
     }
+
+    #[test]
+    fn request_timeout_is_bounded() {
+        assert_eq!(REQUEST_TIMEOUT, Duration::from_secs(30));
+    }
+
+    /// A server that exits immediately (closing stdout) must cause an in-flight
+    /// request to fail promptly rather than hang forever. The reader task hits
+    /// EOF, drains `pending`, and the dropped sender resolves `rx.await` with an
+    /// error well before the 30s request timeout.
+    #[tokio::test]
+    async fn send_errors_when_server_dies_before_responding() {
+        let (command, args) = if cfg!(windows) {
+            (
+                "cmd".to_string(),
+                vec!["/C".to_string(), "exit".to_string()],
+            )
+        } else {
+            (
+                "sh".to_string(),
+                vec!["-c".to_string(), "exit 0".to_string()],
+            )
+        };
+
+        let transport = StdioTransport::spawn(&command, &args, None)
+            .await
+            .expect("spawn short-lived server");
+
+        let req = JsonRpcRequest::new("initialize", None);
+        // Bound the whole call so a regression (a real hang) fails the test
+        // instead of stalling the suite.
+        let result = tokio::time::timeout(Duration::from_secs(5), transport.send(req)).await;
+
+        assert!(
+            result.is_ok(),
+            "send hung instead of erroring on dead server"
+        );
+        assert!(
+            result.unwrap().is_err(),
+            "send should error when the server dies before responding"
+        );
+    }
 }

crates/mcp/src/transport/stdio.rs

@@ -19,6 +19,10 @@ use crate::transport::Transport;
 /// Connection timeout for the initial WebSocket handshake.
 const CONNECT_TIMEOUT: Duration = Duration::from_secs(30);
 
+/// Per-request timeout — a request that gets no response within this window
+/// fails rather than hanging forever (e.g. the server died mid-request).
+const REQUEST_TIMEOUT: Duration = Duration::from_secs(30);
+
 /// Receiver for server-initiated JSON-RPC notifications.
 ///
 /// Returned by [`WsTransport::take_notifications`] exactly once per
@@ -230,12 +234,21 @@ impl Transport for WsTransport {
             tracing::debug!(method = %req.method, id, url = %self.url, "sending WS request");
             self.send_text(&json).await?;
 
-            // Wait for the response from the reader task.
-            rx.await.map_err(|_| {
-                crab_core::Error::Other(
+            // Wait for the response, bounded by REQUEST_TIMEOUT so a dead or
+            // wedged server can't hang the caller indefinitely.
+            match tokio::time::timeout(REQUEST_TIMEOUT, rx).await {
+                Ok(Ok(resp)) => Ok(resp),
+                Ok(Err(_)) => Err(crab_core::Error::Other(
                     "WebSocket connection closed before response received".into(),
-                )
-            })
+                )),
+                Err(_) => {
+                    self.pending.lock().await.remove(&id);
+                    Err(crab_core::Error::Other(format!(
+                        "MCP server at {} did not respond within {REQUEST_TIMEOUT:?}",
+                        self.url
+                    )))
+                }
+            }
         })
     }
 

crates/mcp/src/transport/ws.rs

@@ -25,8 +25,12 @@ impl Default for ContextManager {
 
 impl ContextManager {
     /// Check usage and return the appropriate action.
+    ///
+    /// Occupancy is taken from [`Conversation::effective_token_estimate`],
+    /// which prefers the real `input_tokens` from the latest API response over
+    /// the char-based heuristic.
     pub fn check(&self, conversation: &Conversation) -> ContextAction {
-        let used = conversation.estimated_tokens();
+        let used = conversation.effective_token_estimate();
         let limit = conversation.context_window;
         if limit == 0 {
             return ContextAction::Ok;
@@ -176,6 +180,58 @@ mod tests {
         );
     }
 
+    #[test]
+    fn check_uses_real_input_tokens_over_char_estimate() {
+        use crab_core::model::TokenUsage;
+        let cm = ContextManager {
+            warn_threshold_percent: 30,
+            upgrade_threshold_percent: 50,
+            compact_threshold_percent: 80,
+        };
+        // Short content → tiny char estimate, but the API reported a large real
+        // prompt size (e.g. CJK that chars/4 underestimates). The compaction
+        // decision must honor the real token count.
+        let mut conv = Conversation::new("s".into(), String::new(), 1_000);
+        conv.push_user("短");
+        assert!(
+            matches!(cm.check(&conv), ContextAction::Ok),
+            "char estimate alone is below all thresholds"
+        );
+        conv.record_usage(TokenUsage {
+            input_tokens: 850,
+            ..TokenUsage::default()
+        });
+        assert!(
+            matches!(cm.check(&conv), ContextAction::NeedsCompaction { .. }),
+            "real input_tokens (850/1000) must trigger compaction"
+        );
+    }
+
+    #[test]
+    fn check_adds_post_usage_messages_to_real_tokens() {
+        use crab_core::model::TokenUsage;
+        let cm = ContextManager {
+            warn_threshold_percent: 30,
+            upgrade_threshold_percent: 50,
+            compact_threshold_percent: 80,
+        };
+        let mut conv = Conversation::new("s".into(), String::new(), 1_000);
+        conv.push_user("first");
+        conv.record_usage(TokenUsage {
+            input_tokens: 400,
+            ..TokenUsage::default()
+        });
+        // 400 real tokens = 40% → Warning band only.
+        assert!(matches!(cm.check(&conv), ContextAction::Warning { .. }));
+        // Append a large message (~500 tokens) not yet seen by any API call.
+        conv.push_user("x".repeat(2000));
+        // 400 + ~500 = ~900 → above compact threshold.
+        assert!(
+            matches!(cm.check(&conv), ContextAction::NeedsCompaction { .. }),
+            "appended messages must be folded into the real-token estimate"
+        );
+    }
+
     #[test]
     fn context_action_equality() {
         assert_eq!(ContextAction::Ok, ContextAction::Ok);