bamboo_engine/runtime/

gold_evaluation.rs

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);
    }
}