lc_agents/handoffs/
manager.rs

1//! HandoffManager + HandoffTool
2
3use std::cell::RefCell;
4use std::collections::HashMap;
5use std::sync::{Arc, Mutex};
6
7use async_trait::async_trait;
8use chrono::Utc;
9use serde_json::Value;
10
11use crate::base::AgentExecutor;
12use lc_core::tools::BaseTool;
13use lc_core::tools::ToolError;
14
15use super::handoff::{Handoff, HandoffContext, HandoffError, HandoffRecord, HandoffResult};
16
17// Per-invocation (task-local) handoff chain, independent per concurrent
18// `run()`. Previously the chain was a single field on the shared
19// `HandoffManager`, so two concurrent runs calling `run()`/`execute_handoff`
20// cleared and pushed into the same `Vec` — corrupting each other's cycle
21// detection and depth guard (0.22.0 audit H-A7). Each `run()` scopes its own
22// chain; nested handoffs on the same task tree push/pop into it, so a run
23// never observes another run's traversal. Direct `execute_handoff` calls
24// outside a `run()` scope see no chain and skip the guards.
25tokio::task_local! {
26    static HANDOFF_CHAIN: RefCell<Vec<String>>;
27}
28
29/// Internal state of the Handoff manager (a single Mutex avoids deadlocks from inconsistent multi-lock acquisition order)
30struct HandoffState {
31    agents: HashMap<String, Arc<AgentExecutor>>,
32    primary: Option<String>,
33    history: Vec<HandoffRecord>,
34}
35
36/// Handoff manager: registers multiple Agents and supports task handoff
37pub struct HandoffManager {
38    state: Mutex<HandoffState>,
39    /// Handoff depth limit; exceeding it is rejected (P1-7).
40    max_handoff_depth: usize,
41}
42
43/// Marker for the conversation-summary segment in handoff input (P2-4).
44const SUMMARY_MARKER: &str = "【交接摘要】";
45/// Marker for the task segment in handoff input (P2-4).
46const TASK_MARKER: &str = "【交接任务】";
47
48/// Folds the conversation summary from the handoff context into the target Agent's input (P2-4).
49///
50/// Falls back to bare task text when the summary is missing or empty, preserving old behavior.
51fn build_handoff_input(handoff: &Handoff) -> String {
52    let summary = handoff
53        .context
54        .as_ref()
55        .and_then(|c| c.conversation_summary.as_deref())
56        .unwrap_or_default();
57    if summary.is_empty() {
58        return handoff.task.clone();
59    }
60    format!(
61        "{SUMMARY_MARKER}\n{summary}\n\n{TASK_MARKER}\n{}",
62        handoff.task
63    )
64}
65
66impl HandoffManager {
67    /// Creates a new HandoffManager.
68    pub fn new() -> Self {
69        Self {
70            state: Mutex::new(HandoffState {
71                agents: HashMap::new(),
72                primary: None,
73                history: Vec::new(),
74            }),
75            max_handoff_depth: 10,
76        }
77    }
78
79    /// Sets the handoff depth limit (default 10, P1-7).
80    pub fn with_max_handoff_depth(mut self, depth: usize) -> Self {
81        self.max_handoff_depth = depth.max(1);
82        self
83    }
84
85    /// Registers an Agent
86    pub fn register_agent(
87        &self,
88        name: impl Into<String>,
89        executor: Arc<AgentExecutor>,
90    ) -> Result<(), HandoffError> {
91        self.state
92            .lock()
93            .unwrap_or_else(|e| e.into_inner())
94            .agents
95            .insert(name.into(), executor);
96        Ok(())
97    }
98
99    /// Sets the primary Agent
100    pub fn set_primary(&self, name: &str) -> Result<(), HandoffError> {
101        let mut state = self.state.lock().unwrap_or_else(|e| e.into_inner());
102        if !state.agents.contains_key(name) {
103            return Err(HandoffError::AgentNotFound(name.to_string()));
104        }
105        state.primary = Some(name.to_string());
106        Ok(())
107    }
108
109    /// Executes a handoff: gives the task to the target Agent
110    pub async fn execute_handoff(&self, handoff: Handoff) -> Result<HandoffResult, HandoffError> {
111        // Resolve executor + primary and apply the run-local chain guards.
112        let (executor, primary) = {
113            let state = self.state.lock().unwrap_or_else(|e| e.into_inner());
114
115            let executor = state
116                .agents
117                .get(&handoff.target_agent)
118                .ok_or_else(|| HandoffError::AgentNotFound(handoff.target_agent.clone()))?
119                .clone();
120            let primary = state.primary.clone();
121
122            // P1-7: depth + cycle guards against THIS run's chain. When there
123            // is no enclosing `run()` scope (a direct out-of-band call), the
124            // chain is absent and both guards are skipped.
125            HANDOFF_CHAIN
126                .try_with(|chain| {
127                    let mut chain = chain.borrow_mut();
128                    // P1-7: handoff depth limit
129                    if chain.len() >= self.max_handoff_depth {
130                        return Err(HandoffError::MaxHandoffDepthExceeded(
131                            self.max_handoff_depth,
132                        ));
133                    }
134                    // P1-7: cycle detection - the target is already in the chain, meaning an A→B→A cycle
135                    if chain.contains(&handoff.target_agent) {
136                        return Err(HandoffError::HandoffCycleDetected(
137                            handoff.target_agent.clone(),
138                        ));
139                    }
140                    chain.push(handoff.target_agent.clone());
141                    Ok(())
142                })
143                .unwrap_or(Ok(()))?;
144
145            (executor, primary)
146        };
147
148        // P2-4: fold the conversation summary from the handoff context into the target Agent's input, rather than transferring control raw.
149        let input = build_handoff_input(&handoff);
150        let result = executor
151            .invoke(input)
152            .await
153            .map_err(|e| HandoffError::ExecutionError(e.to_string()));
154
155        // Pop the current chain node whether it succeeds or fails, so a single failure doesn't poison subsequent handoffs
156        let result = match result {
157            Ok(r) => r,
158            Err(e) => {
159                let _ = HANDOFF_CHAIN.try_with(|chain| chain.borrow_mut().pop());
160                return Err(e);
161            }
162        };
163
164        let from = HANDOFF_CHAIN
165            .try_with(|chain| {
166                let mut chain = chain.borrow_mut();
167                chain.pop();
168                chain.last().cloned()
169            })
170            .unwrap_or(primary)
171            .unwrap_or_default();
172        {
173            let mut state = self.state.lock().unwrap_or_else(|e| e.into_inner());
174            state.history.push(HandoffRecord {
175                from_agent: from,
176                to_agent: handoff.target_agent.clone(),
177                task: handoff.task.clone(),
178                result: result.clone(),
179                timestamp: Utc::now().to_rfc3339(),
180            });
181        }
182
183        Ok(HandoffResult {
184            agent_name: handoff.target_agent,
185            result,
186            next_handoff: None,
187        })
188    }
189
190    /// Runs the primary Agent
191    pub async fn run(&self, input: String) -> Result<String, HandoffError> {
192        let (executor, primary) = {
193            let state = self.state.lock().unwrap_or_else(|e| e.into_inner());
194            let primary = state
195                .primary
196                .clone()
197                .ok_or_else(|| HandoffError::AgentNotFound("primary not set".to_string()))?;
198            let executor = state
199                .agents
200                .get(&primary)
201                .ok_or_else(|| HandoffError::AgentNotFound(primary.clone()))?
202                .clone();
203            // P1-7: primary is the chain's start, taking part in cycle detection
204            (executor, primary)
205        };
206        // Scope a fresh chain for this invocation so concurrent `run()`s never
207        // observe or corrupt one another's traversal state (0.22.0 H-A7).
208        HANDOFF_CHAIN
209            .scope(RefCell::new(vec![primary]), async move {
210                executor
211                    .invoke(input)
212                    .await
213                    .map_err(|e| HandoffError::ExecutionError(e.to_string()))
214            })
215            .await
216    }
217
218    /// Gets the handoff history
219    pub fn history(&self) -> Vec<HandoffRecord> {
220        self.state
221            .lock()
222            .unwrap_or_else(|e| e.into_inner())
223            .history
224            .clone()
225    }
226
227    /// Generates a HandoffTool for each registered Agent (for the primary Agent to call)
228    pub fn handoff_tools(self: &Arc<Self>) -> Vec<Arc<dyn BaseTool>> {
229        let state = self.state.lock().unwrap_or_else(|e| e.into_inner());
230        state
231            .agents
232            .keys()
233            .map(|name| Arc::new(HandoffTool::new(self.clone(), name.clone())) as Arc<dyn BaseTool>)
234            .collect()
235    }
236}
237
238impl Default for HandoffManager {
239    fn default() -> Self {
240        Self::new()
241    }
242}
243
244/// Handoff Tool - the handoff tool exposed to the LLM
245pub struct HandoffTool {
246    manager: Arc<HandoffManager>,
247    target_agent: String,
248    name: String,
249    description: String,
250}
251
252impl HandoffTool {
253    /// Creates a handoff tool targeting the given Agent.
254    pub fn new(manager: Arc<HandoffManager>, target_agent: impl Into<String>) -> Self {
255        let target_agent = target_agent.into();
256        let name = format!("handoff_to_{}", target_agent);
257        let description = format!("将任务交接给 {} agent", target_agent);
258        Self {
259            manager,
260            target_agent,
261            name,
262            description,
263        }
264    }
265}
266
267#[async_trait]
268impl BaseTool for HandoffTool {
269    fn name(&self) -> &str {
270        &self.name
271    }
272
273    fn description(&self) -> &str {
274        &self.description
275    }
276
277    async fn run(&self, input: String) -> Result<String, ToolError> {
278        // The input may be JSON {"task": "...", "summary": "..."} or plain text
279        let parsed = serde_json::from_str::<Value>(&input).ok();
280        let task = parsed
281            .as_ref()
282            .and_then(|v| v.get("task").and_then(|t| t.as_str()))
283            .map(|s| s.to_string())
284            .unwrap_or(input.clone());
285        // P2-4: the JSON may carry a summary, passed to the target Agent via the context.
286        let summary = parsed
287            .as_ref()
288            .and_then(|v| v.get("summary").and_then(|s| s.as_str()))
289            .map(|s| s.to_string());
290        let original_request = parsed
291            .as_ref()
292            .and_then(|v| v.get("original_request").and_then(|s| s.as_str()))
293            .unwrap_or(&input);
294
295        let context =
296            Some(HandoffContext::new(original_request).with_summary(summary.unwrap_or_default()));
297        let handoff = Handoff {
298            target_agent: self.target_agent.clone(),
299            task,
300            context,
301        };
302        // 0.20.0 S3.1:交接环/深度守卫是框架级「拒绝执行」的控制中止,不是工具执行
303        // 失败。用 ControlAbort 而非 ExecutionFailed 区分,使 AgentExecutor 能把
304        // 它硬失败上抛(agent 无法靠重规划绕过环检测),而不会像普通工具失败那样
305        // 被转成 observation 喂回循环。
306        let result = self
307            .manager
308            .execute_handoff(handoff)
309            .await
310            .map_err(|e| ToolError::ControlAbort(e.to_string()))?;
311        Ok(result.result)
312    }
313}
314
315#[cfg(test)]
316mod tests {
317    use super::*;
318    use crate::handoffs::HandoffContext;
319    use crate::types::{AgentAction, AgentFinish, AgentOutput, AgentStep, ToolInput};
320    use crate::{AgentError, BaseAgent, FunctionCallingAgent};
321    use lc_providers::{OpenAIChat, OpenAIConfig};
322
323    fn mock_executor() -> Arc<AgentExecutor> {
324        let llm = OpenAIChat::new(OpenAIConfig::default());
325        let agent = FunctionCallingAgent::new(llm, vec![], None);
326        Arc::new(AgentExecutor::new(
327            Arc::new(agent) as Arc<dyn BaseAgent>,
328            vec![],
329        ))
330    }
331
332    /// Offline-capable mock agent for P1-7 nested handoff tests (no real LLM).
333    ///
334    /// Behavior: the first round either issues one tool call (`first_action`),
335    /// or directly triggers one handoff (`direct_handoff`, bypassing
336    /// HandoffTool); later rounds return Finish.
337    struct OfflineAgent {
338        first_action: Option<(&'static str, String)>,
339        direct_handoff: Option<String>,
340        manager: Option<Arc<HandoffManager>>,
341    }
342
343    #[async_trait]
344    impl BaseAgent for OfflineAgent {
345        async fn plan(
346            &self,
347            intermediate_steps: &[AgentStep],
348            _inputs: &HashMap<String, String>,
349            _config: Option<&lc_core::runnables::RunnableConfig>,
350        ) -> Result<AgentOutput, AgentError> {
351            if !intermediate_steps.is_empty() {
352                return Ok(AgentOutput::Finish(AgentFinish::new(
353                    "done".to_string(),
354                    String::new(),
355                )));
356            }
357            if let Some(target) = &self.direct_handoff {
358                // First round triggers the handoff directly inside plan; cycle detection makes execute_handoff reject it immediately.
359                let mgr = self.manager.as_ref().unwrap();
360                let result = mgr
361                    .execute_handoff(Handoff {
362                        target_agent: target.clone(),
363                        task: "inner".to_string(),
364                        context: None,
365                    })
366                    .await;
367                return match result {
368                    Ok(r) => Ok(AgentOutput::Finish(AgentFinish::new(
369                        r.result,
370                        String::new(),
371                    ))),
372                    Err(e) => Err(AgentError::Other(format!("handoff failed: {}", e))),
373                };
374            }
375            if let Some((tool, input)) = &self.first_action {
376                return Ok(AgentOutput::Action(AgentAction {
377                    tool: tool.to_string(),
378                    tool_input: ToolInput::String {
379                        value: input.clone(),
380                    },
381                    log: String::new(),
382                }));
383            }
384            Ok(AgentOutput::Finish(AgentFinish::new(
385                "done".to_string(),
386                String::new(),
387            )))
388        }
389    }
390
391    #[test]
392    fn test_register_and_set_primary() {
393        let mgr = HandoffManager::new();
394        mgr.register_agent("researcher", mock_executor()).unwrap();
395        mgr.set_primary("researcher").unwrap();
396        // Non-existent agent
397        assert!(mgr.set_primary("nope").is_err());
398    }
399
400    #[test]
401    fn test_set_primary_without_register_errors() {
402        let mgr = HandoffManager::new();
403        assert!(mgr.set_primary("ghost").is_err());
404    }
405
406    #[tokio::test]
407    async fn test_execute_handoff_not_found() {
408        let mgr = HandoffManager::new();
409        let handoff = Handoff {
410            target_agent: "nope".to_string(),
411            task: "task".to_string(),
412            context: None,
413        };
414        assert!(mgr.execute_handoff(handoff).await.is_err());
415    }
416
417    #[tokio::test]
418    async fn test_run_without_primary_errors() {
419        let mgr = HandoffManager::new();
420        assert!(mgr.run("hi".to_string()).await.is_err());
421    }
422
423    #[test]
424    fn test_handoff_tool_name() {
425        let mgr = Arc::new(HandoffManager::new());
426        let tool = HandoffTool::new(mgr, "writer".to_string());
427        assert_eq!(tool.name(), "handoff_to_writer");
428        assert!(tool.description().contains("writer"));
429    }
430
431    #[test]
432    fn test_handoff_tools_generated() {
433        let mgr = HandoffManager::new();
434        mgr.register_agent("a", mock_executor()).unwrap();
435        mgr.register_agent("b", mock_executor()).unwrap();
436        let mgr = Arc::new(mgr);
437        let tools = mgr.handoff_tools();
438        assert_eq!(tools.len(), 2);
439    }
440
441    /// P1-7: A hands off to B, B hands back to A → cycle detected and error, no infinite loop.
442    #[tokio::test]
443    async fn test_handoff_cycle_detected() {
444        let manager = Arc::new(HandoffManager::new());
445        let agent_a = OfflineAgent {
446            first_action: Some(("handoff_to_b", "task for b".to_string())),
447            direct_handoff: None,
448            manager: None,
449        };
450        let agent_b = OfflineAgent {
451            first_action: None,
452            direct_handoff: Some("a".to_string()),
453            manager: Some(manager.clone()),
454        };
455        let tool_b = HandoffTool::new(manager.clone(), "b".to_string());
456        let executor_a = AgentExecutor::new(
457            Arc::new(agent_a) as Arc<dyn BaseAgent>,
458            vec![Arc::new(tool_b) as Arc<dyn BaseTool>],
459        );
460        let executor_b = AgentExecutor::new(Arc::new(agent_b) as Arc<dyn BaseAgent>, vec![]);
461
462        manager.register_agent("a", Arc::new(executor_a)).unwrap();
463        manager.register_agent("b", Arc::new(executor_b)).unwrap();
464        manager.set_primary("a").unwrap();
465
466        let err = manager.run("start".to_string()).await.unwrap_err();
467        assert!(
468            err.to_string().contains("handoff cycle"),
469            "should return a cycle-detection error, got: {}",
470            err
471        );
472    }
473
474    /// P1-7: handoff depth exceeds the limit → rejected with an error.
475    #[tokio::test]
476    async fn test_handoff_depth_limit() {
477        // Depth limit 1: primary(a) already takes one layer, so handing off to b exceeds it.
478        let manager = Arc::new(HandoffManager::new().with_max_handoff_depth(1));
479        let agent_a = OfflineAgent {
480            first_action: Some(("handoff_to_b", "t".to_string())),
481            direct_handoff: None,
482            manager: None,
483        };
484        let agent_b = OfflineAgent {
485            first_action: None,
486            direct_handoff: None,
487            manager: None,
488        };
489        let tool_b = HandoffTool::new(manager.clone(), "b".to_string());
490        let executor_a = AgentExecutor::new(
491            Arc::new(agent_a) as Arc<dyn BaseAgent>,
492            vec![Arc::new(tool_b) as Arc<dyn BaseTool>],
493        );
494        let executor_b = AgentExecutor::new(Arc::new(agent_b) as Arc<dyn BaseAgent>, vec![]);
495
496        manager.register_agent("a", Arc::new(executor_a)).unwrap();
497        manager.register_agent("b", Arc::new(executor_b)).unwrap();
498        manager.set_primary("a").unwrap();
499
500        let err = manager.run("start".to_string()).await.unwrap_err();
501        assert!(
502            err.to_string().contains("handoff depth"),
503            "should return a depth-exceeded error, got: {}",
504            err
505        );
506    }
507
508    /// Captures the input the target Agent receives (for P2-4 tests).
509    struct CaptureAgent {
510        received: Arc<Mutex<Option<String>>>,
511    }
512
513    #[async_trait]
514    impl BaseAgent for CaptureAgent {
515        async fn plan(
516            &self,
517            _intermediate_steps: &[AgentStep],
518            inputs: &HashMap<String, String>,
519            _config: Option<&lc_core::runnables::RunnableConfig>,
520        ) -> Result<AgentOutput, AgentError> {
521            *self.received.lock().unwrap_or_else(|e| e.into_inner()) = inputs.get("input").cloned();
522            Ok(AgentOutput::Finish(AgentFinish::new(
523                "done".to_string(),
524                String::new(),
525            )))
526        }
527    }
528
529    fn capture_executor(received: &Arc<Mutex<Option<String>>>) -> Arc<AgentExecutor> {
530        Arc::new(AgentExecutor::new(
531            Arc::new(CaptureAgent {
532                received: received.clone(),
533            }) as Arc<dyn BaseAgent>,
534            vec![],
535        ))
536    }
537
538    /// P2-4: when a handoff carries a summary, the target Agent's input contains the summary and task markers.
539    #[tokio::test]
540    async fn test_handoff_carries_conversation_summary() {
541        let received = Arc::new(Mutex::new(None));
542        let manager = HandoffManager::new();
543        manager
544            .register_agent("researcher", capture_executor(&received))
545            .unwrap();
546
547        let ctx = HandoffContext::new("原始请求").with_summary("此前对话要点A");
548        let handoff = Handoff {
549            target_agent: "researcher".to_string(),
550            task: "继续研究".to_string(),
551            context: Some(ctx),
552        };
553        manager.execute_handoff(handoff).await.unwrap();
554
555        let input = received
556            .lock()
557            .unwrap_or_else(|e| e.into_inner())
558            .clone()
559            .unwrap();
560        assert!(
561            input.contains("此前对话要点A"),
562            "目标应收到摘要,实际: {input}"
563        );
564        assert!(input.contains("继续研究"), "目标应收到任务,实际: {input}");
565        assert!(input.contains("【交接摘要】"), "应带摘要标记");
566    }
567
568    /// P2-4: when a handoff carries no summary, the target Agent receives bare task text (old behavior unchanged).
569    #[tokio::test]
570    async fn test_handoff_without_summary_bare_task() {
571        let received = Arc::new(Mutex::new(None));
572        let manager = HandoffManager::new();
573        manager
574            .register_agent("researcher", capture_executor(&received))
575            .unwrap();
576
577        let handoff = Handoff {
578            target_agent: "researcher".to_string(),
579            task: "只做这个".to_string(),
580            context: None,
581        };
582        manager.execute_handoff(handoff).await.unwrap();
583
584        let input = received
585            .lock()
586            .unwrap_or_else(|e| e.into_inner())
587            .clone()
588            .unwrap();
589        assert_eq!(input, "只做这个");
590    }
591
592    /// P2-4: an empty summary degrades to the bare task, not polluting the input.
593    #[tokio::test]
594    async fn test_handoff_empty_summary_bare_task() {
595        let received = Arc::new(Mutex::new(None));
596        let manager = HandoffManager::new();
597        manager
598            .register_agent("researcher", capture_executor(&received))
599            .unwrap();
600
601        let ctx = HandoffContext::new("原始请求").with_summary("");
602        let handoff = Handoff {
603            target_agent: "researcher".to_string(),
604            task: "taskY".to_string(),
605            context: Some(ctx),
606        };
607        manager.execute_handoff(handoff).await.unwrap();
608
609        let input = received
610            .lock()
611            .unwrap_or_else(|e| e.into_inner())
612            .clone()
613            .unwrap();
614        assert_eq!(input, "taskY");
615    }
616
617    /// P2-4: when HandoffTool's JSON input carries a summary, it reaches the target via the context.
618    #[tokio::test]
619    async fn test_handoff_tool_summary_json_flows_to_target() {
620        let received = Arc::new(Mutex::new(None));
621        let manager = Arc::new(HandoffManager::new());
622        manager
623            .register_agent("writer", capture_executor(&received))
624            .unwrap();
625
626        let tool = HandoffTool::new(manager, "writer".to_string());
627        let json = r#"{"task": "写总结", "summary": "会议要点S"}"#;
628        tool.run(json.to_string()).await.unwrap();
629
630        let input = received
631            .lock()
632            .unwrap_or_else(|e| e.into_inner())
633            .clone()
634            .unwrap();
635        assert!(
636            input.contains("会议要点S"),
637            "工具 JSON summary 应传到目标,实际: {input}"
638        );
639        assert!(input.contains("写总结"));
640    }
641
642    /// P2-4: HandoffTool plain-text input stays a bare-task transfer.
643    #[tokio::test]
644    async fn test_handoff_tool_plain_text_bare_task() {
645        let received = Arc::new(Mutex::new(None));
646        let manager = Arc::new(HandoffManager::new());
647        manager
648            .register_agent("writer", capture_executor(&received))
649            .unwrap();
650
651        let tool = HandoffTool::new(manager, "writer".to_string());
652        tool.run("纯文本任务".to_string()).await.unwrap();
653
654        let input = received
655            .lock()
656            .unwrap_or_else(|e| e.into_inner())
657            .clone()
658            .unwrap();
659        assert_eq!(input, "纯文本任务");
660    }
661}
lc_agents/handoffs/manager.rs

lc_agents/handoffs/
manager.rs
