bamboo_engine/runtime/task_evaluation/
executor.rs

1use std::sync::Arc;
2
3use tokio::sync::mpsc;
4
5use bamboo_agent_core::{AgentError, AgentEvent, Session};
6use bamboo_domain::ReasoningEffort;
7use bamboo_domain::TaskItemStatus;
8use bamboo_llm::{LLMProvider, LLMRequestOptions};
9
10use super::super::task_context::TaskLoopContext;
11use super::message_builder::build_task_evaluation_messages;
12use super::schema::get_task_evaluation_tools;
13use super::token_estimation::estimate_prompt_tokens;
14use super::TaskEvaluationResult;
15
16mod outcomes;
17
18pub struct TaskEvaluationFrame<'a> {
19    pub event_tx: &'a mpsc::Sender<AgentEvent>,
20    pub session_id: &'a str,
21    pub model: &'a str,
22    pub reasoning_effort: Option<ReasoningEffort>,
23    pub timeout_context: &'a crate::runtime::stream::handler::StreamTimeoutContext,
24}
25
26fn skipped_evaluation(reasoning: &str) -> TaskEvaluationResult {
27    TaskEvaluationResult {
28        needs_evaluation: false,
29        updates: Vec::new(),
30        reasoning: reasoning.to_string(),
31        prompt_tokens: 0,
32        completion_tokens: 0,
33    }
34}
35
36fn normalize_lightweight_reasoning_effort(
37    reasoning_effort: Option<ReasoningEffort>,
38) -> Option<ReasoningEffort> {
39    reasoning_effort.map(|effort| match effort {
40        ReasoningEffort::Xhigh | ReasoningEffort::Max => ReasoningEffort::High,
41        other => other,
42    })
43}
44
45/// 执行 TaskList 评估
46pub async fn evaluate_task_progress(
47    ctx: &TaskLoopContext,
48    session: &Session,
49    llm: Arc<dyn LLMProvider>,
50    frame: &TaskEvaluationFrame<'_>,
51) -> Result<TaskEvaluationResult, AgentError> {
52    use crate::runtime::stream::handler::consume_llm_stream_silent_with_context;
53
54    let event_tx = frame.event_tx;
55    let session_id = frame.session_id;
56    let model = frame.model;
57    let reasoning_effort = frame.reasoning_effort;
58    let timeout_context = frame.timeout_context;
59
60    let in_progress_count = ctx
61        .items
62        .iter()
63        .filter(|item| matches!(item.status, TaskItemStatus::InProgress))
64        .count();
65
66    if in_progress_count == 0 {
67        return Ok(skipped_evaluation("No in-progress tasks to evaluate"));
68    }
69
70    // When to evaluate is owned entirely by the caller (the loop spawns this only
71    // on a Task-tool write); this function just decides how. The single remaining
72    // guard above skips when there is nothing in progress to assess.
73    tracing::info!(
74        "[{}] Evaluating {} in-progress task items",
75        session_id,
76        in_progress_count
77    );
78
79    let _ = event_tx
80        .send(AgentEvent::TaskEvaluationStarted {
81            session_id: session_id.to_string(),
82            items_count: in_progress_count,
83            generation: Some(ctx.version),
84        })
85        .await;
86
87    let messages = build_task_evaluation_messages(ctx, session);
88    let prompt_tokens = estimate_prompt_tokens(&messages);
89    let tools = get_task_evaluation_tools();
90
91    // Use model from parameter (passed from config), not from session.
92    tracing::debug!("[{}] Task evaluation using model: {}", session_id, model);
93
94    let request_reasoning_effort = normalize_lightweight_reasoning_effort(reasoning_effort);
95    if request_reasoning_effort != reasoning_effort {
96        tracing::debug!(
97            "[{}] Task evaluation downgraded reasoning effort from {:?} to {:?} for lightweight request",
98            session_id,
99            reasoning_effort,
100            request_reasoning_effort
101        );
102    }
103
104    let request_options = LLMRequestOptions {
105        session_id: Some(session_id.to_string()),
106        reasoning_effort: request_reasoning_effort,
107        parallel_tool_calls: None,
108        required_tool: None,
109        responses: None,
110        request_purpose: Some("task_evaluation".to_string()),
111        cache: None,
112    };
113    match llm
114        .chat_stream_with_options(&messages, &tools, Some(8192), model, Some(&request_options))
115        .await
116    {
117        Ok(stream) => {
118            let stream_output = consume_llm_stream_silent_with_context(
119                stream,
120                &tokio_util::sync::CancellationToken::new(),
121                session_id,
122                timeout_context,
123            )
124            .await?;
125
126            Ok(outcomes::build_success_result(
127                stream_output,
128                event_tx,
129                session_id,
130                prompt_tokens,
131                ctx.version,
132            )
133            .await)
134        }
135        Err(error) => {
136            tracing::warn!("[{}] Task evaluation failed: {}", session_id, error);
137            Ok(skipped_evaluation(&format!("Evaluation failed: {}", error)))
138        }
139    }
140}
bamboo_engine/runtime/task_evaluation/executor.rs

bamboo_engine/runtime/task_evaluation/
executor.rs
