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    evaluate_task_progress_with_dispatch(ctx, session, llm, frame, std::future::ready(()), || {})
53        .await
54}
55
56/// Internal evaluator entry point that observes the exact provider-dispatch
57/// boundary. Queueing and prompt preparation must not inflate evaluator
58/// duration metrics, so the callback runs immediately before the provider
59/// future is first polled.
60pub(crate) async fn evaluate_task_progress_with_dispatch<G, Fut, F>(
61    ctx: &TaskLoopContext,
62    session: &Session,
63    llm: Arc<dyn LLMProvider>,
64    frame: &TaskEvaluationFrame<'_>,
65    acquire_dispatch_guard: Fut,
66    on_dispatch: F,
67) -> Result<TaskEvaluationResult, AgentError>
68where
69    Fut: std::future::Future<Output = G>,
70    F: FnOnce(),
71{
72    use crate::runtime::stream::handler::{
73        await_stream_bootstrap, consume_llm_stream_silent_with_context,
74    };
75
76    let event_tx = frame.event_tx;
77    let session_id = frame.session_id;
78    let model = frame.model;
79    let reasoning_effort = frame.reasoning_effort;
80
81    let in_progress_count = ctx
82        .items
83        .iter()
84        .filter(|item| matches!(item.status, TaskItemStatus::InProgress))
85        .count();
86
87    if in_progress_count == 0 {
88        return Ok(skipped_evaluation("No in-progress tasks to evaluate"));
89    }
90
91    // When to evaluate is owned entirely by the caller (the loop spawns this only
92    // on a Task-tool write); this function just decides how. The single remaining
93    // guard above skips when there is nothing in progress to assess.
94    tracing::info!(
95        "[{}] Evaluating {} in-progress task items",
96        session_id,
97        in_progress_count
98    );
99
100    let _ = event_tx
101        .send(AgentEvent::TaskEvaluationStarted {
102            session_id: session_id.to_string(),
103            items_count: in_progress_count,
104            generation: Some(ctx.version),
105        })
106        .await;
107
108    let messages = build_task_evaluation_messages(ctx, session);
109    let prompt_tokens = estimate_prompt_tokens(&messages);
110    let tools = get_task_evaluation_tools();
111
112    // Use model from parameter (passed from config), not from session.
113    tracing::debug!("[{}] Task evaluation using model: {}", session_id, model);
114
115    let request_reasoning_effort = normalize_lightweight_reasoning_effort(reasoning_effort);
116    if request_reasoning_effort != reasoning_effort {
117        tracing::debug!(
118            "[{}] Task evaluation downgraded reasoning effort from {:?} to {:?} for lightweight request",
119            session_id,
120            reasoning_effort,
121            request_reasoning_effort
122        );
123    }
124
125    let request_options = LLMRequestOptions {
126        session_id: Some(session_id.to_string()),
127        reasoning_effort: request_reasoning_effort,
128        parallel_tool_calls: None,
129        required_tool: None,
130        responses: None,
131        request_purpose: Some("task_evaluation".to_string()),
132        cache: None,
133    };
134    let cancel_token = tokio_util::sync::CancellationToken::new();
135    let _dispatch_guard = acquire_dispatch_guard.await;
136    let timeout_context = frame.timeout_context.clone().begin_request();
137    on_dispatch();
138    let stream = match await_stream_bootstrap(
139        llm.chat_stream_with_options(&messages, &tools, Some(8192), model, Some(&request_options)),
140        &cancel_token,
141        session_id,
142        &timeout_context,
143    )
144    .await?
145    {
146        Ok(stream) => stream,
147        Err(error) => {
148            tracing::warn!("[{}] Task evaluation failed: {}", session_id, error);
149            return Ok(skipped_evaluation(&format!("Evaluation failed: {error}")));
150        }
151    };
152    let stream_output =
153        consume_llm_stream_silent_with_context(stream, &cancel_token, session_id, &timeout_context)
154            .await?;
155
156    Ok(outcomes::build_success_result(
157        stream_output,
158        event_tx,
159        session_id,
160        prompt_tokens,
161        ctx.version,
162    )
163    .await)
164}
bamboo_engine/runtime/task_evaluation/executor.rs

bamboo_engine/runtime/task_evaluation/
executor.rs
