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

bamboo_engine/runtime/task_evaluation/
executor.rs
