use std::sync::{
atomic::{AtomicBool, Ordering},
Arc,
};
use bamboo_agent_core::tools::{FunctionSchema, ToolCall, ToolSchema};
use bamboo_agent_core::{AgentError, AgentEvent, GoldCheckpoint, GoldConfidence, GoldDecision};
use bamboo_agent_core::{Message, Role, Session};
use bamboo_compression::{TiktokenTokenCounter, TokenCounter};
use bamboo_domain::ReasoningEffort;
use bamboo_llm::{LLMProvider, LLMRequestOptions};
use chrono::Utc;
use serde_json::json;
use tokio::sync::{mpsc, Notify};
use tokio_util::sync::CancellationToken;
use crate::runtime::config::GoldConfig;
use crate::runtime::stream::handler::{
await_stream_bootstrap, consume_llm_stream_silent_with_context, StreamTimeoutContext,
};
use crate::runtime::task_context::TaskLoopContext;
use bamboo_metrics::TokenUsage as MetricsTokenUsage;
/// Evaluation-scoped frame bundling parameters that identify and configure a
/// single gold evaluation pass. Passed into [`evaluate_gold`] to keep its
/// parameter count below the clippy threshold.
pub struct GoldEvalFrame<'a> {
pub event_tx: &'a mpsc::Sender<AgentEvent>,
pub session_id: &'a str,
pub model: &'a str,
pub timeout_context: StreamTimeoutContext,
pub reasoning_effort: Option<ReasoningEffort>,
pub checkpoint: GoldCheckpoint,
pub iteration: u32,
}
#[derive(Debug, Clone)]
pub struct GoldEvaluationResult {
pub checkpoint: GoldCheckpoint,
pub iteration: u32,
pub decision: GoldDecision,
pub confidence: GoldConfidence,
pub reasoning: String,
/// Concrete information still missing to achieve the goal, if any.
pub missing_information: Vec<String>,
/// The single most useful next action for the agent to take, if the
/// evaluator can name one. Used to give the auto-continue nudge direction.
pub next_action: Option<String>,
pub prompt_tokens: u64,
pub completion_tokens: u64,
}
#[derive(Debug, Clone)]
pub(crate) struct AsyncGoldEvaluationRequest {
pub(crate) session_id: String,
pub(crate) round_number: usize,
pub(crate) model_name: String,
pub(crate) timeout_context: StreamTimeoutContext,
pub(crate) reasoning_effort: Option<ReasoningEffort>,
pub(crate) checkpoint: GoldCheckpoint,
pub(crate) session_snapshot: Session,
pub(crate) task_context_snapshot: Option<TaskLoopContext>,
pub(crate) gold_config: GoldConfig,
}
#[derive(Debug, Clone)]
pub(crate) struct AsyncGoldEvaluationResult {
pub(crate) round_number: usize,
pub(crate) model_name: String,
pub(crate) evaluation_result: GoldEvaluationResult,
}
// A known local routing refusal is not an evaluator verdict. Other evaluator
// errors retain the existing low-confidence fallback for async Gold.
#[derive(Debug)]
pub(crate) enum GoldEvaluationError {
Routing(AgentError),
Evaluation(AgentError),
}
impl From<AgentError> for GoldEvaluationError {
fn from(error: AgentError) -> Self {
Self::Evaluation(error)
}
}
impl std::fmt::Display for GoldEvaluationError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::Routing(error) | Self::Evaluation(error) => write!(f, "{error}"),
}
}
}
impl GoldEvaluationError {
pub(crate) fn into_agent_error(self) -> AgentError {
match self {
Self::Routing(error) | Self::Evaluation(error) => error,
}
}
}
fn estimate_prompt_tokens(messages: &[Message]) -> u64 {
let counter = TiktokenTokenCounter::default();
u64::from(counter.count_messages(messages))
}
fn estimate_completion_tokens(content: &str, tool_calls: &[ToolCall]) -> u64 {
let counter = TiktokenTokenCounter::default();
let mut completion_surface = content.to_string();
for call in tool_calls {
if !completion_surface.is_empty() {
completion_surface.push('\n');
}
completion_surface.push_str(&call.function.name);
completion_surface.push('\n');
completion_surface.push_str(&call.function.arguments);
}
u64::from(counter.count_text(&completion_surface))
}
#[allow(clippy::too_many_arguments)]
pub(crate) fn build_async_gold_evaluation_request(
task_context: &Option<TaskLoopContext>,
session: &Session,
session_id: &str,
round_number: usize,
model_name: Option<&str>,
reasoning_effort: Option<ReasoningEffort>,
checkpoint: GoldCheckpoint,
gold_config: &GoldConfig,
timeout_context: StreamTimeoutContext,
) -> Result<Option<AsyncGoldEvaluationRequest>, AgentError> {
if !gold_config.enabled {
return Ok(None);
}
let model_name = gold_config
.model_name
.as_deref()
.or(model_name)
.ok_or_else(|| AgentError::LLM("gold evaluation model_name is required".to_string()))?;
Ok(Some(AsyncGoldEvaluationRequest {
session_id: session_id.to_string(),
round_number,
model_name: model_name.to_string(),
timeout_context,
reasoning_effort,
checkpoint,
session_snapshot: session.clone(),
task_context_snapshot: task_context.clone(),
gold_config: gold_config.clone(),
}))
}
// The standalone test helper owns its own claim. The real runner shares it
// with its InFlight state so run cleanup cannot emit a second terminal event.
#[cfg(test)]
pub(crate) async fn execute_async_gold_evaluation(
request: AsyncGoldEvaluationRequest,
llm: Arc<dyn LLMProvider>,
event_tx: mpsc::Sender<AgentEvent>,
configured_limit: usize,
) -> Result<AsyncGoldEvaluationResult, AgentError> {
execute_async_gold_evaluation_with_terminal(
request,
llm,
event_tx,
configured_limit,
Arc::new(GoldEvaluationTerminal::default()),
)
.await
}
pub(crate) async fn execute_async_gold_evaluation_with_terminal(
request: AsyncGoldEvaluationRequest,
llm: Arc<dyn LLMProvider>,
event_tx: mpsc::Sender<AgentEvent>,
configured_limit: usize,
terminal_claimed: Arc<GoldEvaluationTerminal>,
) -> Result<AsyncGoldEvaluationResult, AgentError> {
let budget_provider = llm.clone();
let budget_model = request.model_name.clone();
let acquire_dispatch_guard = async move {
crate::runtime::runner::auxiliary_budget::acquire(
&budget_provider,
&budget_model,
configured_limit,
)
.await
};
let evaluation_result = match evaluate_gold_with_dispatch_and_terminal(
&request.session_snapshot,
request.task_context_snapshot.as_ref(),
&request.gold_config,
llm,
&GoldEvalFrame {
event_tx: &event_tx,
session_id: &request.session_id,
model: &request.model_name,
timeout_context: request.timeout_context.clone(),
reasoning_effort: request.reasoning_effort,
checkpoint: request.checkpoint,
iteration: request.round_number as u32,
},
acquire_dispatch_guard,
Some(terminal_claimed.as_ref()),
)
.await
{
Ok(result) => result,
Err(GoldEvaluationError::Routing(error)) => return Err(error),
Err(GoldEvaluationError::Evaluation(error)) => GoldEvaluationResult {
checkpoint: request.checkpoint,
iteration: request.round_number as u32,
decision: GoldDecision::Continue,
confidence: GoldConfidence::Low,
reasoning: format!("Gold evaluation failed: {error}"),
missing_information: Vec::new(),
next_action: None,
prompt_tokens: 0,
completion_tokens: 0,
},
};
Ok(AsyncGoldEvaluationResult {
round_number: request.round_number,
model_name: request.model_name,
evaluation_result,
})
}
pub async fn evaluate_gold(
session: &Session,
task_context: Option<&TaskLoopContext>,
config: &GoldConfig,
llm: Arc<dyn LLMProvider>,
frame: &GoldEvalFrame<'_>,
) -> Result<GoldEvaluationResult, AgentError> {
evaluate_gold_with_dispatch(
session,
task_context,
config,
llm,
frame,
std::future::ready(()),
)
.await
.map_err(GoldEvaluationError::into_agent_error)
}
pub(crate) async fn evaluate_gold_with_dispatch<G, Fut>(
session: &Session,
task_context: Option<&TaskLoopContext>,
config: &GoldConfig,
llm: Arc<dyn LLMProvider>,
frame: &GoldEvalFrame<'_>,
acquire_dispatch_guard: Fut,
) -> Result<GoldEvaluationResult, GoldEvaluationError>
where
Fut: std::future::Future<Output = G>,
{
evaluate_gold_with_dispatch_and_terminal(
session,
task_context,
config,
llm,
frame,
acquire_dispatch_guard,
None,
)
.await
}
// One claim is shared by an async Gold evaluator and its cleanup. The
// notification lets a losing sender stop waiting for channel capacity.
#[derive(Default)]
pub(crate) struct GoldEvaluationTerminal {
claimed: AtomicBool,
published: Notify,
}
pub(crate) async fn emit_gold_cancelled_once(
event_tx: &mpsc::Sender<AgentEvent>,
terminal: &GoldEvaluationTerminal,
session_id: &str,
reason: &str,
) {
// Register before checking the flag so a concurrent publication cannot be
// missed between the check and polling the notification.
let published = terminal.published.notified();
tokio::pin!(published);
published.as_mut().enable();
if terminal.claimed.load(Ordering::Acquire) {
return;
}
let permit = tokio::select! {
biased;
_ = &mut published => return,
result = event_tx.reserve() => match result {
Ok(permit) => permit,
Err(_) => return,
},
};
let event = AgentEvent::GoldEvaluationCancelled {
session_id: session_id.to_string(),
reason: reason.to_string(),
};
if terminal
.claimed
.compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
.is_ok()
{
// Enqueue and notify are synchronous after the claim: an aborted task
// cannot yield in between them. This is not a consumer acknowledgement.
permit.send(event);
terminal.published.notify_waiters();
}
}
async fn evaluate_gold_with_dispatch_and_terminal<G, Fut>(
session: &Session,
task_context: Option<&TaskLoopContext>,
config: &GoldConfig,
llm: Arc<dyn LLMProvider>,
frame: &GoldEvalFrame<'_>,
acquire_dispatch_guard: Fut,
terminal_claimed: Option<&GoldEvaluationTerminal>,
) -> Result<GoldEvaluationResult, GoldEvaluationError>
where
Fut: std::future::Future<Output = G>,
{
if let Some(error) = llm.known_pre_dispatch_error() {
return Err(GoldEvaluationError::Routing(AgentError::LLM(
error.to_string(),
)));
}
// Bind frame fields as locals so the rest of the function body stays unchanged.
let event_tx = frame.event_tx;
let session_id = frame.session_id;
let model = frame.model;
let reasoning_effort = frame.reasoning_effort;
let checkpoint = frame.checkpoint;
let iteration = frame.iteration;
let _ = event_tx
.send(AgentEvent::GoldEvaluationStarted {
session_id: session_id.to_string(),
checkpoint,
iteration,
})
.await;
let messages = build_gold_messages(session, task_context, config, checkpoint);
let prompt_tokens = estimate_prompt_tokens(&messages);
let tools = get_gold_evaluation_tools();
let request_options = LLMRequestOptions {
session_id: Some(session_id.to_string()),
reasoning_effort,
parallel_tool_calls: None,
required_tool: None,
responses: None,
request_purpose: Some("gold_evaluation".to_string()),
cache: None,
};
let cancel_token = CancellationToken::new();
let _dispatch_guard = acquire_dispatch_guard.await;
let timeout_context = frame.timeout_context.clone().begin_request();
let stream = await_stream_bootstrap(
llm.chat_stream_with_options(
&messages,
&tools,
Some(config.max_output_tokens),
model,
Some(&request_options),
),
&cancel_token,
session_id,
&timeout_context,
)
.await?;
let stream = match stream {
Ok(stream) => stream,
Err(error) if llm.local_dispatch_refused() => {
// No provider request was dispatched. Free the scarce slot before
// waiting for a potentially backpressured event consumer.
drop(_dispatch_guard);
if let Some(terminal_claimed) = terminal_claimed {
emit_gold_cancelled_once(event_tx, terminal_claimed, session_id, "route_rejected")
.await;
} else {
let _ = event_tx
.send(AgentEvent::GoldEvaluationCancelled {
session_id: session_id.to_string(),
reason: "route_rejected".to_string(),
})
.await;
}
return Err(GoldEvaluationError::Routing(AgentError::LLM(
error.to_string(),
)));
}
Err(error) => {
return Err(GoldEvaluationError::Evaluation(AgentError::LLM(
error.to_string(),
)))
}
};
let stream_output =
consume_llm_stream_silent_with_context(stream, &cancel_token, session_id, &timeout_context)
.await?;
let result = parse_gold_evaluation(
&stream_output.content,
&stream_output.tool_calls,
checkpoint,
iteration,
prompt_tokens,
);
let _ = event_tx
.send(AgentEvent::GoldEvaluationCompleted {
session_id: session_id.to_string(),
checkpoint: result.checkpoint,
iteration: result.iteration,
decision: result.decision,
confidence: result.confidence,
reasoning: result.reasoning.clone(),
})
.await;
Ok(result)
}
pub fn build_gold_messages(
session: &Session,
task_context: Option<&TaskLoopContext>,
config: &GoldConfig,
checkpoint: GoldCheckpoint,
) -> Vec<Message> {
let mut messages = Vec::new();
let mut system_prompt = String::from(
"You are a gold progress evaluator. Judge whether the agent has already achieved the user's goal, should continue execution, needs user input, is blocked, or is exhausted.\n\nRules:\n1. This phase is observe-only: do not mutate state or invent actions.\n2. You must call report_gold_evaluation exactly once.\n3. Use achieved only when the user's actual goal is satisfied.\n4. Use continue when more agent work is still appropriate.\n5. Use need_input only when missing user input is the true next blocker.\n6. Use blocked only for a concrete blocking condition.\n7. Use exhausted for loops, budget exhaustion, or clear inability to make progress.\n8. Keep reasoning short, concrete, and evidence-based."
);
if let Some(extra) = config
.evaluation_prompt
.as_deref()
.map(str::trim)
.filter(|value| !value.is_empty())
{
system_prompt.push_str("\n\nAdditional instructions:\n");
system_prompt.push_str(extra);
}
messages.push(Message::system(system_prompt));
let task_summary = task_context
.map(TaskLoopContext::format_for_prompt)
.filter(|value| !value.trim().is_empty())
.unwrap_or_else(|| "## Current Task List\nNo task list available.".to_string());
let pending_question_summary = session
.pending_question
.as_ref()
.map(|question| {
let options = if question.options.is_empty() {
"none".to_string()
} else {
question.options.join(" | ")
};
let tool_name = if question.tool_name.trim().is_empty() {
"unknown".to_string()
} else {
question.tool_name.clone()
};
format!(
"question={} | options={} | tool={} | source={:?}",
question.question, options, tool_name, question.source
)
})
.unwrap_or_else(|| "none".to_string());
let runtime_summary = session
.agent_runtime_state
.as_ref()
.map(|state| {
format!(
"status={:?} | current_round={} | round_cap={} | suspend_reason={} | waiting_for_children={}",
state.status,
state.round.current_round,
state
.round
.max_rounds
.map(|value| value.to_string())
.unwrap_or_else(|| "unlimited".to_string()),
state
.suspension
.as_ref()
.map(|s| s.reason.clone())
.unwrap_or_else(|| "none".to_string()),
state.waiting_for_children.is_some()
)
})
.unwrap_or_else(|| "runtime_state=none".to_string());
let recent_messages = format_recent_messages(session, 6);
let goal_section = config
.effective_goal()
.map(|goal| format!("## Goal\n{goal}"))
.unwrap_or_else(|| {
"## Goal\nNo explicit goal set. Judge against the user's request inferred from the conversation.".to_string()
});
let user_prompt = format!(
"## Gold Checkpoint\ncheckpoint={}\n\n{}\n\n## Runtime\n{}\n\n## Pending Question\n{}\n\n{}\n\n## Recent Conversation\n{}\n\n## Instruction\nReport the best current Gold judgment for this checkpoint by measuring progress against the goal above. Remember: Phase 1 is observe-only, so only report decision/confidence/reasoning.",
checkpoint.as_str(),
goal_section,
runtime_summary,
pending_question_summary,
task_summary,
recent_messages,
);
messages.push(Message::user(user_prompt));
messages
}
fn format_recent_messages(session: &Session, limit: usize) -> String {
let start = session.messages.len().saturating_sub(limit);
let mut lines = Vec::new();
for message in session.messages.iter().skip(start) {
let role = match message.role {
Role::System => "system",
Role::User => "user",
Role::Assistant => "assistant",
Role::Tool => "tool",
};
let mut content = message.content.trim().replace('\n', " ");
if content.chars().count() > 240 {
content = format!("{}…", content.chars().take(240).collect::<String>());
}
if content.is_empty() {
content = "<empty>".to_string();
}
lines.push(format!("- [{}] {}", role, content));
}
if lines.is_empty() {
"- <no messages>".to_string()
} else {
lines.join("\n")
}
}
pub fn get_gold_evaluation_tools() -> Vec<ToolSchema> {
vec![ToolSchema {
schema_type: "function".to_string(),
function: FunctionSchema {
name: "report_gold_evaluation".to_string(),
description: "Report the current Gold evaluation decision for the session".to_string(),
parameters: json!({
"type": "object",
"properties": {
"decision": {
"type": "string",
"enum": ["continue", "achieved", "blocked", "need_input", "exhausted"]
},
"confidence": {
"type": "string",
"enum": ["low", "medium", "high"]
},
"reasoning": {
"type": "string",
"description": "Short concrete reasoning for the decision"
},
"missing_information": {
"type": "array",
"items": { "type": "string" },
"description": "Concrete pieces of information still missing to achieve the goal. Empty when nothing is missing."
},
"next_action": {
"type": "string",
"description": "The single most useful next action the agent should take. Provide when decision is continue."
}
},
"required": ["decision", "confidence", "reasoning"],
"additionalProperties": false
}),
},
}]
}
pub fn parse_gold_evaluation(
content: &str,
tool_calls: &[ToolCall],
checkpoint: GoldCheckpoint,
iteration: u32,
prompt_tokens: u64,
) -> GoldEvaluationResult {
let completion_tokens = estimate_completion_tokens(content, tool_calls);
let parsed = parse_gold_result_from_tool_calls(tool_calls);
let parsed = parsed.unwrap_or_else(|| {
let fallback_reasoning = content.trim().to_string();
ParsedGoldResult {
decision: GoldDecision::Continue,
confidence: GoldConfidence::Low,
reasoning: if fallback_reasoning.is_empty() {
"Gold evaluation returned no structured result; defaulting to continue.".to_string()
} else {
fallback_reasoning
},
missing_information: Vec::new(),
next_action: None,
}
});
GoldEvaluationResult {
checkpoint,
iteration,
decision: parsed.decision,
confidence: parsed.confidence,
reasoning: parsed.reasoning,
missing_information: parsed.missing_information,
next_action: parsed.next_action,
prompt_tokens,
completion_tokens,
}
}
struct ParsedGoldResult {
decision: GoldDecision,
confidence: GoldConfidence,
reasoning: String,
missing_information: Vec<String>,
next_action: Option<String>,
}
fn parse_gold_result_from_tool_calls(tool_calls: &[ToolCall]) -> Option<ParsedGoldResult> {
for tool_call in tool_calls {
if tool_call.function.name != "report_gold_evaluation" {
continue;
}
let Ok(args) = serde_json::from_str::<serde_json::Value>(&tool_call.function.arguments)
else {
continue;
};
let decision = match args.get("decision").and_then(|value| value.as_str()) {
Some("continue") => GoldDecision::Continue,
Some("achieved") => GoldDecision::Achieved,
Some("blocked") => GoldDecision::Blocked,
Some("need_input") => GoldDecision::NeedInput,
Some("exhausted") => GoldDecision::Exhausted,
_ => continue,
};
let confidence = match args.get("confidence").and_then(|value| value.as_str()) {
Some("low") => GoldConfidence::Low,
Some("medium") => GoldConfidence::Medium,
Some("high") => GoldConfidence::High,
_ => GoldConfidence::Low,
};
let reasoning = args
.get("reasoning")
.and_then(|value| value.as_str())
.map(str::trim)
.filter(|value| !value.is_empty())
.unwrap_or("Gold evaluation produced no reasoning")
.to_string();
let missing_information = args
.get("missing_information")
.and_then(|value| value.as_array())
.map(|items| {
items
.iter()
.filter_map(|item| item.as_str())
.map(str::trim)
.filter(|value| !value.is_empty())
.map(str::to_string)
.collect::<Vec<_>>()
})
.unwrap_or_default();
let next_action = args
.get("next_action")
.and_then(|value| value.as_str())
.map(str::trim)
.filter(|value| !value.is_empty())
.map(str::to_string);
return Some(ParsedGoldResult {
decision,
confidence,
reasoning,
missing_information,
next_action,
});
}
None
}
pub(crate) fn apply_gold_evaluation_result(
session: &mut Session,
result: &GoldEvaluationResult,
) -> MetricsTokenUsage {
let evaluation_count = session
.metadata
.get("gold.evaluation_count")
.and_then(|value| value.parse::<u64>().ok())
.unwrap_or(0)
.saturating_add(1);
let summary = json!({
"checkpoint": result.checkpoint.as_str(),
"iteration": result.iteration,
"decision": result.decision.as_str(),
"confidence": result.confidence.as_str(),
"reasoning": result.reasoning,
"recorded_at": Utc::now().to_rfc3339(),
});
session
.metadata
.insert("gold.last_evaluation".to_string(), summary.to_string());
session.metadata.insert(
"gold.last_decision".to_string(),
result.decision.as_str().to_string(),
);
session.metadata.insert(
"gold.last_confidence".to_string(),
result.confidence.as_str().to_string(),
);
session
.metadata
.insert("gold.last_reasoning".to_string(), result.reasoning.clone());
session.metadata.insert(
"gold.last_checkpoint".to_string(),
result.checkpoint.as_str().to_string(),
);
session.metadata.insert(
"gold.last_iteration".to_string(),
result.iteration.to_string(),
);
session.metadata.insert(
"gold.evaluation_count".to_string(),
evaluation_count.to_string(),
);
session.updated_at = Utc::now();
let mut usage = MetricsTokenUsage {
prompt_tokens: result.prompt_tokens,
completion_tokens: result.completion_tokens,
..Default::default()
};
usage.recompute_total();
usage
}
#[cfg(test)]
mod tests {
use super::*;
use bamboo_agent_core::tools::FunctionCall;
// This witness supports both the old return shape and the corrected one,
// allowing the same assertions to run as a RED test on the baseline.
trait GoldOutcomeWitness {
fn is_routing_refusal(&self) -> bool;
fn ordinary_failure_reasoning(&self) -> Option<&str>;
}
impl GoldOutcomeWitness for AsyncGoldEvaluationResult {
fn is_routing_refusal(&self) -> bool {
false
}
fn ordinary_failure_reasoning(&self) -> Option<&str> {
Some(&self.evaluation_result.reasoning)
}
}
impl GoldOutcomeWitness for Result<AsyncGoldEvaluationResult, AgentError> {
fn is_routing_refusal(&self) -> bool {
self.is_err()
}
fn ordinary_failure_reasoning(&self) -> Option<&str> {
self.as_ref()
.ok()
.map(|result| result.evaluation_result.reasoning.as_str())
}
}
fn synthetic_request(model: &str) -> AsyncGoldEvaluationRequest {
AsyncGoldEvaluationRequest {
session_id: "gold-preflight-fixture".into(),
round_number: 1,
model_name: model.into(),
timeout_context: StreamTimeoutContext::default(),
reasoning_effort: None,
checkpoint: GoldCheckpoint::PostRound,
session_snapshot: Session::new("gold-preflight-fixture", model),
task_context_snapshot: None,
gold_config: GoldConfig {
enabled: true,
..Default::default()
},
}
}
#[tokio::test]
async fn gold_terminal_wakes_losing_sender_without_another_channel_slot() {
let (event_tx, mut event_rx) = mpsc::channel(1);
event_tx
.send(AgentEvent::Token {
content: "occupy".into(),
})
.await
.unwrap();
let terminal = GoldEvaluationTerminal::default();
let mut route = Box::pin(emit_gold_cancelled_once(
&event_tx,
&terminal,
"synthetic-session",
"route_rejected",
));
let mut cleanup = Box::pin(emit_gold_cancelled_once(
&event_tx,
&terminal,
"synthetic-session",
"run_cancelled",
));
// Poll both to Pending while the same one-slot channel is full.
assert!(futures::poll!(route.as_mut()).is_pending());
assert!(futures::poll!(cleanup.as_mut()).is_pending());
assert!(!terminal.claimed.load(Ordering::Acquire));
assert!(matches!(event_rx.try_recv(), Ok(AgentEvent::Token { .. })));
// Both must return while the one terminal event still fills the only
// channel slot. The losing reservation cannot wait for another drain.
tokio::time::timeout(std::time::Duration::from_secs(5), async {
tokio::join!(route, cleanup);
})
.await
.expect("the losing terminal sender must be notified");
assert!(terminal.claimed.load(Ordering::Acquire));
assert!(matches!(
event_rx.try_recv(),
Ok(AgentEvent::GoldEvaluationCancelled { ref session_id, ref reason })
if session_id == "synthetic-session"
&& (reason == "route_rejected" || reason == "run_cancelled")
));
assert!(event_rx.try_recv().is_err(), "exactly one terminal event");
}
#[tokio::test]
async fn gold_terminal_closed_channel_never_claims_a_terminal() {
let (event_tx, mut event_rx) = mpsc::channel(1);
event_tx
.send(AgentEvent::Token {
content: "occupy".into(),
})
.await
.unwrap();
let terminal = GoldEvaluationTerminal::default();
let mut route = Box::pin(emit_gold_cancelled_once(
&event_tx,
&terminal,
"synthetic-session",
"route_rejected",
));
let mut cleanup = Box::pin(emit_gold_cancelled_once(
&event_tx,
&terminal,
"synthetic-session",
"run_cancelled",
));
assert!(futures::poll!(route.as_mut()).is_pending());
assert!(futures::poll!(cleanup.as_mut()).is_pending());
event_rx.close();
tokio::time::timeout(std::time::Duration::from_secs(5), async {
tokio::join!(route, cleanup);
})
.await
.expect("closed channel must unblock both reservations");
assert!(!terminal.claimed.load(Ordering::Acquire));
assert!(matches!(event_rx.try_recv(), Ok(AgentEvent::Token { .. })));
assert!(
event_rx.try_recv().is_err(),
"no terminal could be enqueued"
);
}
#[tokio::test]
async fn gold_terminal_cancelled_waiter_leaves_claim_available_for_cleanup() {
let (event_tx, mut event_rx) = mpsc::channel(1);
event_tx
.send(AgentEvent::Token {
content: "occupy".into(),
})
.await
.unwrap();
let terminal = GoldEvaluationTerminal::default();
let mut route = Box::pin(emit_gold_cancelled_once(
&event_tx,
&terminal,
"synthetic-session",
"route_rejected",
));
let mut cleanup = Box::pin(emit_gold_cancelled_once(
&event_tx,
&terminal,
"synthetic-session",
"run_cancelled",
));
assert!(futures::poll!(route.as_mut()).is_pending());
assert!(futures::poll!(cleanup.as_mut()).is_pending());
drop(route);
assert!(!terminal.claimed.load(Ordering::Acquire));
assert!(matches!(event_rx.try_recv(), Ok(AgentEvent::Token { .. })));
tokio::time::timeout(std::time::Duration::from_secs(5), cleanup)
.await
.expect("cleanup must claim the released slot");
assert!(terminal.claimed.load(Ordering::Acquire));
assert!(matches!(
event_rx.try_recv(),
Ok(AgentEvent::GoldEvaluationCancelled { ref session_id, ref reason })
if session_id == "synthetic-session" && reason == "run_cancelled"
));
assert!(event_rx.try_recv().is_err(), "exactly one terminal event");
}
#[tokio::test]
async fn rejected_fast_never_waits_for_budget_or_emits_a_gold_verdict() {
let routing = crate::app_context::model_routing_test_support::RoutingContext::new(None);
let mut config = routing.config.read().await.clone();
config.provider_instances.get_mut("fast-b").unwrap().enabled = false;
let (model, provider) = crate::model_config_helper::resolve_fast_model_for_runner(
Some(&config),
"chat-a",
&routing.registry,
);
let model = model.unwrap();
let provider = provider.unwrap();
// Also verify trait-object propagation through the role decorator.
let mut reference = bamboo_domain::ProviderModelRef::new("fast-b", model.as_str());
reference.reasoning_effort = Some(bamboo_domain::ReasoningEffort::Low);
let decorated = bamboo_llm::ResolvedModel::from_ref(provider.clone(), &reference).provider;
for selected in [provider, decorated] {
let held =
crate::runtime::runner::auxiliary_budget::acquire(&selected, &model, 1).await;
let (event_tx, mut event_rx) = mpsc::channel(8);
let outcome = tokio::time::timeout(
std::time::Duration::from_secs(2),
execute_async_gold_evaluation(synthetic_request(&model), selected, event_tx, 1),
)
.await
.expect("rejected Fast must not wait for an auxiliary permit");
assert!(
outcome.is_routing_refusal(),
"refusal must not become a Gold verdict"
);
assert!(
event_rx.try_recv().is_err(),
"no Gold Started or Completed event"
);
drop(held);
}
assert!(routing.requests("fast-b").is_empty());
assert!(routing.requests("chat-a").is_empty());
}
#[tokio::test]
async fn managed_gold_route_revoked_after_started_has_route_terminal_and_releases_permit() {
use crate::app_context::model_routing_test_support::CaptureProvider;
use bamboo_domain::ProviderModelRef;
use bamboo_llm::{Config, ProviderRegistry};
use std::time::Duration;
use tokio::sync::RwLock;
// This managed handle is only constructed and validated. If routing
// incorrectly dispatches, the only delegate is a synthetic capture;
// no LLM request is sent to the loopback address in the config.
let dir = tempfile::tempdir().unwrap();
let mut config = Config::default();
config.provider = "gold-a".into();
config.default_provider_instance = Some("gold-a".into());
config.http_proxy = "http://127.0.0.1:9".into();
config.https_proxy.clear();
config.proxy_auth = None;
config.provider_instances.insert(
"gold-a".into(),
serde_json::from_value(serde_json::json!({
"provider_type": "openai",
"enabled": true,
"api_key": "synthetic-only",
"base_url": "http://127.0.0.1:1",
"model": "main-model",
"runtime_models": ["main-model", "explicit-gold"]
}))
.unwrap(),
);
let registry = Arc::new(
ProviderRegistry::from_config(&config, dir.path().to_path_buf())
.await
.unwrap(),
);
let target = ProviderModelRef::new("gold-a", "explicit-gold");
let (selected, origin) = registry.provider_for_model_with_origin(&target).unwrap();
let capture = Arc::new(CaptureProvider::default());
let live = Arc::new(RwLock::new(config));
let guard = bamboo_llm::gold_admission::guarded_gold_target_with_origin(
live.clone(),
origin.expect("the factory-built registry is managed"),
target,
selected,
capture.clone(),
);
assert!(guard.known_pre_dispatch_error().is_none());
let held =
crate::runtime::runner::auxiliary_budget::acquire(&guard, "explicit-gold", 1).await;
let (event_tx, mut event_rx) = mpsc::channel(1);
let runner_guard = guard.clone();
let runner_tx = event_tx.clone();
let handle = tokio::spawn(async move {
execute_async_gold_evaluation(
synthetic_request("explicit-gold"),
runner_guard,
runner_tx,
1,
)
.await
});
let started = tokio::time::timeout(Duration::from_secs(5), event_rx.recv())
.await
.expect("Gold should start before waiting for its auxiliary slot")
.expect("Gold event sender must remain open");
assert!(matches!(
started,
AgentEvent::GoldEvaluationStarted {
ref session_id,
checkpoint: GoldCheckpoint::PostRound,
iteration: 1,
} if session_id == "gold-preflight-fixture"
));
assert!(!handle.is_finished(), "the held permit blocks dispatch");
assert!(capture.requests.lock().unwrap().is_empty());
// Fill the event channel to ensure a failure event cannot be sent yet.
// The refused evaluation must release its auxiliary slot before it
// waits for the event sink to accept that event.
event_tx
.send(AgentEvent::Token {
content: "saturation-fixture".into(),
})
.await
.unwrap();
live.write()
.await
.provider_instances
.get_mut("gold-a")
.unwrap()
.extra
.insert("runtime_models".into(), serde_json::json!(["main-model"]));
drop(held);
tokio::time::timeout(Duration::from_secs(5), async {
while !guard.local_dispatch_refused() {
tokio::task::yield_now().await;
}
})
.await
.expect("the managed guard must refuse the revoked route");
assert!(
capture.requests.lock().unwrap().is_empty(),
"no provider call"
);
let available = tokio::time::timeout(
Duration::from_secs(5),
crate::runtime::runner::auxiliary_budget::acquire(&guard, "explicit-gold", 1),
)
.await
.expect("the refused evaluation must release its auxiliary permit before its event send");
drop(available);
assert!(matches!(
event_rx.try_recv(),
Ok(AgentEvent::Token { ref content }) if content == "saturation-fixture"
));
let outcome = tokio::time::timeout(Duration::from_secs(5), handle)
.await
.expect("the revoked evaluation must stop")
.expect("the evaluation task must not panic");
let error = outcome.expect_err("a revoked route must not become a verdict");
match error {
AgentError::LLM(message) => assert!(
message.contains("no longer admitted"),
"unexpected routing error: {message}"
),
other => panic!("expected an LLM routing error, got {other:?}"),
}
assert!(guard.local_dispatch_refused());
assert!(
capture.requests.lock().unwrap().is_empty(),
"no provider call"
);
// This describes the auxiliary evaluation's route refusal, not a user
// cancellation of the owning run and not a completed Gold verdict.
let terminal = event_rx
.try_recv()
.expect("Started must have a route terminal");
let terminal = serde_json::to_value(terminal).unwrap();
assert_eq!(terminal["type"], "gold_evaluation_cancelled");
assert_eq!(terminal["session_id"], "gold-preflight-fixture");
assert_eq!(terminal["reason"], "route_rejected");
assert!(event_rx.try_recv().is_err(), "exactly one terminal event");
}
#[tokio::test]
async fn ordinary_provider_failure_keeps_existing_async_gold_behavior() {
use crate::app_context::model_routing_test_support::{Failure, RoutingContext};
let routing = RoutingContext::new(None);
*routing.providers["fast-b"].failure.lock().unwrap() = Some(Failure::Network);
let provider: Arc<dyn LLMProvider> = routing.providers["fast-b"].clone();
let (event_tx, mut event_rx) = mpsc::channel(8);
let outcome =
execute_async_gold_evaluation(synthetic_request("cheap-b"), provider, event_tx, 1)
.await;
assert!(outcome
.ordinary_failure_reasoning()
.unwrap()
.contains("Gold evaluation failed"));
assert!(matches!(
event_rx.try_recv(),
Ok(AgentEvent::GoldEvaluationStarted { .. })
));
assert!(event_rx.try_recv().is_err());
assert_eq!(routing.requests("fast-b").len(), 1);
assert!(routing.requests("chat-a").is_empty());
}
#[tokio::test]
async fn network_error_is_not_reclassified_after_provider_state_changes() {
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
struct SwitchAfterNetworkError {
changed: AtomicBool,
calls: AtomicUsize,
}
#[async_trait::async_trait]
impl LLMProvider for SwitchAfterNetworkError {
fn known_pre_dispatch_error(&self) -> Option<bamboo_llm::LLMError> {
self.changed
.load(Ordering::SeqCst)
.then(|| bamboo_llm::LLMError::Auth("different later provider state".into()))
}
async fn chat_stream(
&self,
_: &[bamboo_agent_core::Message],
_: &[bamboo_agent_core::ToolSchema],
_: Option<u32>,
_: &str,
) -> Result<bamboo_llm::LLMStream, bamboo_llm::LLMError> {
self.calls.fetch_add(1, Ordering::SeqCst);
self.changed.store(true, Ordering::SeqCst);
Err(bamboo_llm::LLMError::RequestFailed(
"synthetic network failure".into(),
))
}
}
let provider = Arc::new(SwitchAfterNetworkError {
changed: AtomicBool::new(false),
calls: AtomicUsize::new(0),
});
let (event_tx, mut event_rx) = mpsc::channel(8);
let outcome = execute_async_gold_evaluation(
synthetic_request("fixture-model"),
provider.clone(),
event_tx,
1,
)
.await
.expect("an ordinary request error retains the previous async fallback");
assert!(outcome
.evaluation_result
.reasoning
.contains("synthetic network failure"));
assert_eq!(provider.calls.load(Ordering::SeqCst), 1);
assert!(provider.known_pre_dispatch_error().is_some());
assert!(matches!(
event_rx.try_recv(),
Ok(AgentEvent::GoldEvaluationStarted { .. })
));
assert!(event_rx.try_recv().is_err());
}
fn report_call(arguments: serde_json::Value) -> ToolCall {
ToolCall {
id: "call-1".to_string(),
tool_type: "function".to_string(),
function: FunctionCall {
name: "report_gold_evaluation".to_string(),
arguments: arguments.to_string(),
},
}
}
#[test]
fn parse_gold_result_from_tool_calls_reads_structured_fields() {
let parsed = parse_gold_result_from_tool_calls(&[report_call(json!({
"decision": "blocked",
"confidence": "high",
"reasoning": "Missing credentials",
"missing_information": ["API key", " ", "Database URL"],
"next_action": " Ask the user for the API key "
}))])
.expect("gold result should parse");
assert_eq!(parsed.decision, GoldDecision::Blocked);
assert_eq!(parsed.confidence, GoldConfidence::High);
assert_eq!(parsed.reasoning, "Missing credentials");
assert_eq!(
parsed.missing_information,
vec!["API key".to_string(), "Database URL".to_string()]
);
assert_eq!(
parsed.next_action.as_deref(),
Some("Ask the user for the API key")
);
}
#[test]
fn parse_gold_result_from_tool_calls_ignores_other_tools() {
let parsed = parse_gold_result_from_tool_calls(&[ToolCall {
id: "call-1".to_string(),
tool_type: "function".to_string(),
function: FunctionCall {
name: "other_tool".to_string(),
arguments: "{}".to_string(),
},
}]);
assert!(parsed.is_none());
}
#[test]
fn apply_gold_evaluation_result_updates_metadata_keys() {
let mut session = Session::new("session-1", "model");
let result = GoldEvaluationResult {
checkpoint: GoldCheckpoint::PostRound,
iteration: 2,
decision: GoldDecision::Achieved,
confidence: GoldConfidence::Medium,
reasoning: "Goal satisfied".to_string(),
missing_information: Vec::new(),
next_action: None,
prompt_tokens: 10,
completion_tokens: 5,
};
let usage = apply_gold_evaluation_result(&mut session, &result);
assert_eq!(
session
.metadata
.get("gold.last_decision")
.map(String::as_str),
Some("achieved")
);
assert_eq!(
session
.metadata
.get("gold.last_confidence")
.map(String::as_str),
Some("medium")
);
assert_eq!(
session
.metadata
.get("gold.last_checkpoint")
.map(String::as_str),
Some("post_round")
);
assert_eq!(
session
.metadata
.get("gold.evaluation_count")
.map(String::as_str),
Some("1")
);
assert_eq!(usage.prompt_tokens, 10);
assert_eq!(usage.completion_tokens, 5);
assert_eq!(usage.total_tokens, 15);
}
}