bamboo-engine 2026.10.22

use std::borrow::Cow;
use std::collections::{BTreeMap, HashSet};
use std::error::Error as StdError;
use std::sync::Arc;

use tokio::sync::mpsc;
use tokio_util::sync::CancellationToken;

use crate::runtime::config::AgentLoopConfig;
use crate::runtime::runner::prompt_context::append_core_agent_directives;
use crate::runtime::runner::session_setup::prompt_envelope::{
    build_active_workflow_context_block, build_agent_hook_context_block,
    build_conversation_summary_context_block, build_external_memory_context_block,
    build_goal_context_block, build_history_boundary_context_block,
    build_instruction_overlay_context_block, build_plan_mode_context_block,
    build_plan_runtime_context_block, build_project_resources_context_block,
    build_root_orchestration_context_block_for_tools, build_session_identity_context_block,
    build_task_list_context_block, build_workspace_context_block,
};
use crate::runtime::runner::session_setup::prompt_setup::{
    build_stable_prompt_frame_with_sections, StablePrefixSection,
};
use bamboo_agent_core::agent::events::{ProviderPromptUsage, TokenBudgetUsage};
use bamboo_agent_core::tools::ToolSchema;
use bamboo_agent_core::{
    AgentError, AgentEvent, ContextBlock, ContextBlockPriority, ContextBlockStability,
    ContextBlockType, Message, MessagePhase, Role, Session,
};
use bamboo_compression::{PreparedContext, TiktokenTokenCounter, TokenCounter};
use bamboo_config::ContextManagementStrategy;
use bamboo_domain::{
    provider_transcript_boundary_sha256, CapabilityLoadingMode, CompressionEventKind,
    CompressionTriggerType, ModelContextEventKind, ModelContextResetReason, ProviderFamily,
    ProviderProtocol, ReasoningEffort, MAX_MODEL_CONTEXT_RENDERED_BYTES,
};
use bamboo_llm::provider::ResponsesRequestOptions;
use bamboo_llm::{
    CacheTtl, Continuation, LLMProvider, LLMRequestOptions, PromptCachePlan, PromptIR,
    ProviderVisibleToolFootprint, ProviderVisibleToolSegmentKind, Segment, SegmentRole,
};
use bamboo_tools::exposure::activated_discoverable_tools;
use sha2::{Digest, Sha256};

use super::PromptMemoryExposureFrame;

/// LLM-stream frame bundling per-request identification, observability, and
/// model configuration parameters.  Passed into [`execute_llm_stream`] to
/// keep its parameter count below the clippy threshold.
pub(in crate::runtime::runner) struct LlmStreamFrame<'a> {
    pub event_tx: &'a mpsc::Sender<AgentEvent>,
    pub cancel_token: &'a CancellationToken,
    pub session_id: &'a str,
    pub model: &'a str,
    pub provider_name: Option<&'a str>,
    pub provider_type: Option<&'a str>,
    pub reasoning_effort: Option<ReasoningEffort>,
    pub max_context_tokens: u32,
    pub max_output_tokens: u32,
    pub observation_progress_hint: Option<&'a str>,
    pub prompt_memory_exposure: Option<PromptMemoryExposureFrame<'a>>,
}

pub(super) const SESSION_RESPONSES_PREVIOUS_RESPONSE_ID_KEY: &str =
    "responses.previous_response_id";
const CONVERSATION_SUMMARY_START_MARKER: &str = "<!-- CONVERSATION_SUMMARY_START -->";
const INTERRUPTED_ASSISTANT_OUTPUT_KIND: &str = "interrupted_assistant_output";
const AGENT_LOOP_REQUEST_PURPOSE: &str = "agent_loop";
const PROMPT_CACHE_KEY_DOMAIN: &[u8] = b"bamboo/openai/responses/prompt-cache-key/v1\0";
const MAX_FINAL_REQUEST_CHECKPOINT_REPREPARES: usize = 2;
const LOADED_BROWSER_ACKNOWLEDGEMENT: &str =
    "Browser loaded. Its complete schema is in the tools for this request.";

struct ModelUsageGuard {
    runner: Arc<crate::runtime::hooks::HookRunner>,
    snapshot: Option<bamboo_domain::ModelCallUsage>,
    session_birth: chrono::DateTime<chrono::Utc>,
    root_session_id: Option<String>,
}

impl ModelUsageGuard {
    fn finish(
        &mut self,
        status: bamboo_domain::ModelCallStatus,
        usage: Option<crate::runtime::stream::handler::ProviderUsageSnapshot>,
    ) {
        if let Some(mut snapshot) = self.snapshot.take() {
            snapshot.status = status;
            snapshot.usage = reported_token_usage(usage);
            self.runner.notify_model_usage(
                snapshot,
                self.session_birth,
                self.root_session_id.clone(),
            );
        }
    }
}

impl Drop for ModelUsageGuard {
    fn drop(&mut self) {
        self.finish(
            bamboo_infrastructure::telemetry::dropped_model_call_status(),
            None,
        );
    }
}

fn usage_status(error: &AgentError) -> bamboo_domain::ModelCallStatus {
    if matches!(error, AgentError::Cancelled) {
        bamboo_domain::ModelCallStatus::Cancelled
    } else {
        bamboo_domain::ModelCallStatus::Error
    }
}

fn token_usage(
    value: crate::runtime::stream::handler::ProviderUsageSnapshot,
) -> bamboo_domain::ModelTokenUsage {
    fn representable(value: Option<u64>) -> Option<u64> {
        value.filter(|value| i64::try_from(*value).is_ok())
    }
    bamboo_domain::ModelTokenUsage {
        input_tokens: representable(value.input_tokens),
        output_tokens: representable(value.output_tokens),
        reasoning_tokens: representable(value.reasoning_tokens),
        cache_read_input_tokens: representable(value.cache_read_input_tokens),
        cache_creation_input_tokens: representable(value.cache_creation_input_tokens),
        cache_write_input_tokens: representable(value.cache_write_input_tokens),
    }
}

fn reported_token_usage(
    value: Option<crate::runtime::stream::handler::ProviderUsageSnapshot>,
) -> Option<bamboo_domain::ModelTokenUsage> {
    value
        .map(token_usage)
        .filter(|usage| usage != &Default::default())
}

fn context_management_telemetry(
    session: &Session,
    config: &AgentLoopConfig,
) -> crate::token_usage_log::ContextManagementTelemetry {
    let strategy = super::context_preparation::effective_context_pressure_strategy(
        session,
        &config.context_management,
    );
    let strategy = match strategy {
        ContextManagementStrategy::Summary => "summary",
        ContextManagementStrategy::MessageSummary => "message_summary",
        ContextManagementStrategy::RetrievalWindow => "retrieval_window",
    };
    let model_context_state = session.model_context_state.as_ref();
    let latest_retrieval_event = session
        .compression_events
        .iter()
        .rev()
        .find(|event| event.kind == CompressionEventKind::RetrievalWindow);
    let latest_retrieval_archive_trigger_type = latest_retrieval_event.map(|event| {
        match event.trigger_type {
            CompressionTriggerType::Auto => "auto",
            CompressionTriggerType::Manual => "manual",
            CompressionTriggerType::CriticalOverflow => "critical_overflow",
        }
        .to_string()
    });

    crate::token_usage_log::ContextManagementTelemetry {
        strategy: strategy.to_string(),
        model_context_epoch: model_context_state.map_or(0, |state| state.prefix_epoch),
        model_context_reset_reason: model_context_state
            .and_then(|state| state.last_reset_reason)
            .map(|reason| reason.as_str().to_string()),
        retrieval_archive_event_count: session
            .compression_events
            .iter()
            .filter(|event| event.kind == CompressionEventKind::RetrievalWindow)
            .count(),
        latest_retrieval_archive_event_id: latest_retrieval_event.map(|event| event.id.clone()),
        latest_retrieval_archive_trigger_type,
    }
}

fn interruption_kind(error: &AgentError) -> &'static str {
    match error {
        AgentError::Cancelled => "cancelled",
        AgentError::StreamTimeout(_) => "stream_timeout",
        AgentError::LLMOverflow(_) => "llm_overflow",
        AgentError::EmptyAssistantResponse { .. } => "empty_assistant_response",
        AgentError::LLM(_) | AgentError::LLMNonRetryable(_) => "llm_error",
        _ => "execution_error",
    }
}

/// Materialize already-received semantic output as a non-executable transcript
/// record before the stream error leaves the round.
///
/// Partial tool-call fragments are diagnostic metadata only.  They are never
/// assigned to `Message::tool_calls`, so a resume cannot replay an incomplete
/// call as though the model had completed it.
fn append_interrupted_assistant_output(
    session: &mut Session,
    partial: crate::runtime::stream::handler::InterruptedStreamOutput,
    error: &AgentError,
) -> bool {
    if partial.content.is_empty()
        && partial.reasoning_content.is_empty()
        && partial.partial_tool_calls.is_empty()
    {
        return false;
    }

    let partial_tool_calls = (!partial.partial_tool_calls.is_empty()).then(|| {
        serde_json::to_value(&partial.partial_tool_calls).unwrap_or(serde_json::Value::Null)
    });
    let mut message = Message::assistant_with_reasoning(
        partial.content,
        None,
        (!partial.reasoning_content.is_empty()).then_some(partial.reasoning_content),
    );
    if let Some(identity) = partial.visible_message {
        message = identity.apply_to(message);
    }
    message.phase = Some(MessagePhase::Commentary);
    message.reasoning_signature = None;
    message.metadata = Some(serde_json::json!({
        "runtime_kind": INTERRUPTED_ASSISTANT_OUTPUT_KIND,
        "interrupted": true,
        "interruption_kind": interruption_kind(error),
        "partial_tool_calls": partial_tool_calls,
    }));
    session.add_message(message);
    true
}

/// Remove the transient partial record before a retry.  A terminal attempt
/// leaves it in place for the shared execute-boundary checkpoint.
pub(crate) fn discard_latest_interrupted_assistant_output(
    session: &mut Session,
    attempt_tail_message_id: Option<&str>,
) -> Option<String> {
    let interrupted_message_id = session.messages.last().and_then(|message| {
        if Some(message.id.as_str()) == attempt_tail_message_id {
            return None;
        }
        (message
            .metadata
            .as_ref()
            .and_then(|metadata| metadata.get("runtime_kind"))
            .and_then(serde_json::Value::as_str)
            == Some(INTERRUPTED_ASSISTANT_OUTPUT_KIND))
        .then(|| message.id.clone())
    });
    if interrupted_message_id.is_some() {
        session.messages.pop();
        session.updated_at = chrono::Utc::now();
    }
    interrupted_message_id
}

fn session_previous_response_id(session: &Session) -> Option<&str> {
    session
        .metadata
        .get(SESSION_RESPONSES_PREVIOUS_RESPONSE_ID_KEY)
        .map(String::as_str)
        .map(str::trim)
        .filter(|value| !value.is_empty())
}

fn provider_supports_previous_response_id(provider_type: Option<&str>) -> bool {
    !matches!(provider_type.map(str::trim), Some("copilot"))
}

/// The engine's Responses-API request POLICY — store / verbosity / reasoning
/// summary / include list — defined in ONE place so the continuation gate below
/// and the planned request can never disagree about `store`.
fn engine_responses_policy() -> ResponsesRequestOptions {
    ResponsesRequestOptions {
        store: Some(false),
        // Encourage the model to emit visible narration alongside tool calls.
        text_verbosity: Some("high".to_string()),
        reasoning_summary: Some("auto".to_string()),
        include: Some(vec!["reasoning.encrypted_content".to_string()]),
        ..Default::default()
    }
}

/// Derive the privacy-preserving OpenAI Responses cache-affinity hint for an
/// agent-loop session. This is a routing hint only, not a cache-hit guarantee.
/// Length-prefixing the purpose and session beneath a versioned protocol domain
/// prevents concatenation ambiguity and cross-purpose reuse.
fn agent_loop_prompt_cache_key(
    session_id: Option<&str>,
    request_purpose: Option<&str>,
) -> Option<String> {
    let purpose = request_purpose.filter(|purpose| *purpose == AGENT_LOOP_REQUEST_PURPOSE)?;
    let session_id = session_id.filter(|session_id| !session_id.trim().is_empty())?;

    let mut hasher = Sha256::new();
    hasher.update(PROMPT_CACHE_KEY_DOMAIN);
    for component in [purpose.as_bytes(), session_id.as_bytes()] {
        hasher.update((component.len() as u64).to_be_bytes());
        hasher.update(component);
    }
    Some(hex::encode(hasher.finalize()))
}

/// Whether the stateful Responses continuation (`previous_response_id`) may be
/// used under the given request policy. Chaining requires the PREVIOUS turn to
/// have been stored upstream: with `store=false` the upstream never persists a
/// turn, so referencing its id on the next round fails with HTTP 400
/// `invalid_request_error` / `previous_response_not_found`. Since the engine
/// always sends the full input array alongside the id, omitting it under a
/// stateless policy loses nothing. Copilot's HTTP /responses endpoint rejects
/// the parameter outright, so it is excluded regardless of policy.
fn responses_continuation_enabled(
    policy: &ResponsesRequestOptions,
    provider_type: Option<&str>,
) -> bool {
    policy.store == Some(true) && provider_supports_previous_response_id(provider_type)
}

fn format_reqwest_transport_error(error: &reqwest::Error) -> String {
    let mut kinds = Vec::new();
    if error.is_timeout() {
        kinds.push("timeout");
    }
    if error.is_connect() {
        kinds.push("connect");
    }
    if error.is_request() {
        kinds.push("request");
    }
    if error.is_body() {
        kinds.push("body");
    }
    if error.is_decode() {
        kinds.push("decode");
    }
    if error.is_redirect() {
        kinds.push("redirect");
    }
    if error.is_builder() {
        kinds.push("builder");
    }
    if error.is_status() {
        kinds.push("status");
    }

    let kind = if kinds.is_empty() {
        "unknown".to_string()
    } else {
        kinds.join("+")
    };
    let url = error
        .url()
        .map(ToString::to_string)
        .unwrap_or_else(|| "<unknown>".to_string());

    let mut causes = Vec::new();
    let mut source = StdError::source(error);
    while let Some(cause) = source {
        causes.push(cause.to_string());
        source = cause.source();
        if causes.len() >= 4 {
            break;
        }
    }

    if causes.is_empty() {
        format!(
            "HTTP transport error [{}] for url ({}): {}",
            kind, url, error
        )
    } else {
        format!(
            "HTTP transport error [{}] for url ({}): {} | causes: {}",
            kind,
            url,
            error,
            causes.join(" | ")
        )
    }
}

fn is_http_body_failure(error: &bamboo_llm::provider::LLMError) -> bool {
    let bamboo_llm::provider::LLMError::Http(http) = error else {
        return false;
    };
    let mut source: Option<&(dyn StdError + 'static)> = Some(http);
    while let Some(cause) = source {
        if cause
            .downcast_ref::<reqwest::Error>()
            .is_some_and(reqwest::Error::is_body)
        {
            return true;
        }
        source = cause.source();
    }
    false
}

fn format_provider_error(error: bamboo_llm::provider::LLMError) -> String {
    match error {
        bamboo_llm::provider::LLMError::Http(http) => format_reqwest_transport_error(&http),
        other => other.to_string(),
    }
}

fn is_llm_overflow_error(message: &str) -> bool {
    let normalized = message.trim().to_ascii_lowercase();
    if normalized.is_empty() {
        return false;
    }

    let overflow_patterns = [
        "prompt too long",
        "context too long",
        "maximum context length",
        "maximum context size",
        "context length exceeded",
        "context window exceeded",
        "request too large",
        "too many tokens",
        "input is too long",
        "input too long",
        "token limit exceeded",
    ];

    overflow_patterns
        .iter()
        .any(|pattern| normalized.contains(pattern))
}

fn is_conversation_summary_message(message: &Message) -> bool {
    matches!(message.role, Role::System)
        && message.content.contains(CONVERSATION_SUMMARY_START_MARKER)
}

fn derive_system_remainder_message(
    message: &Message,
    stable_instructions: &str,
) -> Option<Message> {
    if !matches!(message.role, Role::System) || is_conversation_summary_message(message) {
        return None;
    }

    // Compare only the normalized user base. Every host-owned section — including
    // legacy workspace/instruction markers — is either in the canonical system
    // blocks or the typed model-context ledger and must never leak as a second
    // System remainder.
    let normalized = crate::runtime::runner::session_setup::prompt_setup::normalize_base_prompt(
        &message.content,
    );
    let trimmed = normalized.trim();
    if trimmed.is_empty() {
        return None;
    }

    let stable_normalized =
        crate::runtime::runner::session_setup::prompt_setup::normalize_base_prompt(
            stable_instructions,
        );
    let stable_trimmed = stable_normalized.trim();
    if stable_trimmed.is_empty() {
        return Some(Message::system(trimmed.to_string()));
    }

    if trimmed == stable_trimmed {
        return None;
    }

    if let Some(remainder) = trimmed.strip_prefix(stable_trimmed) {
        // Persisted ⊇ reference: the persisted message carries genuine extra
        // content beyond the assembled system field — re-emit only that tail.
        let remainder = remainder.trim();
        return (!remainder.is_empty()).then(|| Message::system(remainder.to_string()));
    }

    // Persisted ⊆ reference: the persisted message (typically the bare configured
    // base, since contexts are injected only at request time) sits at the head of
    // the assembled system field, and the remainder is framework-injected context
    // (workspace/skill/tool-guide/env) already present in the system field and the
    // relocated context messages. There is nothing extra to surface, so do not
    // re-emit the base into the conversation every round. The `\n` boundary check
    // avoids a mid-token false prefix (e.g. "base" vs "basement").
    if stable_trimmed.starts_with(&format!("{trimmed}\n")) {
        return None;
    }

    Some(Message::system(trimmed.to_string()))
}

struct PreparedRequestEnvelope {
    /// The single canonical request and SOLE source of truth: system field +
    /// immutable stable prefix + one chronological model transcript + the cache
    /// plan (`ir.cache`). `continuation` is filled in at dispatch time. Every
    /// provider renders its wire from this via `chat_stream_ir`; every wire view
    /// (chat / Responses-input / continuation-delta) is derived by the IR's
    /// lowering methods, so there are no parallel pre-baked message vecs.
    ir: PromptIR,
    /// Per-section breakdown of the cacheable stable prefix, kept for prompt-cache
    /// drift diagnostics (not sent to the provider; not part of the wire IR).
    stable_prefix_sections: Vec<StablePrefixSection>,
    ledger_changed: bool,
    prefix_epoch: u64,
    prefix_reset_reason: Option<bamboo_domain::ModelContextResetReason>,
}

#[derive(Debug, Clone, Copy, Default)]
pub(super) struct ProjectedRequestUsage {
    pub message_input_tokens: u32,
    pub tool_schema_input_tokens: u32,
    pub input_tokens: u32,
    pub tool_schema_serialized_bytes: usize,
    pub tool_schema_serialized_chars: usize,
    pub tool_schema_segment_count: usize,
    pub tool_schema_late_bound_segment_count: usize,
    pub ledger_rendered_bytes: usize,
    /// Complete token cost of model-context snapshot messages that carry the
    /// current archived-history boundary.
    pub history_boundary_input_tokens: u32,
    /// The transformed message vector after any provider-only history
    /// projection. Refit must subtract this count rather than durable bytes.
    pub prepared_message_input_tokens: u32,
}

fn measure_request_usage(
    session: &Session,
    envelope: &PreparedRequestEnvelope,
    tool_footprint: &ProviderVisibleToolFootprint,
) -> ProjectedRequestUsage {
    let counter = TiktokenTokenCounter::default();
    let messages = envelope.ir.flatten();
    let history_boundary_event_ids = session
        .model_context_state
        .as_ref()
        .into_iter()
        .flat_map(|state| state.events.iter())
        .filter(|event| {
            event.block_type == ContextBlockType::HistoryBoundary
                && event.kind == ModelContextEventKind::Snapshot
        })
        .map(|event| event.id.as_str())
        .collect::<HashSet<_>>();
    let history_boundary_input_tokens = messages
        .iter()
        .filter(|message| history_boundary_event_ids.contains(message.id.as_str()))
        .fold(0u32, |total, message| {
            total.saturating_add(counter.count_message(message))
        });
    let mut message_input_tokens = counter.count_messages(&messages);
    let mut replaced_anchors = HashSet::new();
    for group in &envelope.ir.provider_transcript_groups {
        if replaced_anchors.insert(group.anchor_message_id()) {
            if let Some(anchor) = messages
                .iter()
                .find(|message| message.id == group.anchor_message_id())
            {
                message_input_tokens =
                    message_input_tokens.saturating_sub(counter.count_message(anchor));
            }
        }
        let payloads = group
            .items()
            .iter()
            .map(|item| item.payload())
            .collect::<Vec<_>>();
        let wire = serde_json::to_string(&payloads).unwrap_or_default();
        message_input_tokens = message_input_tokens
            .saturating_add(counter.count_text(&wire))
            .saturating_add((group.items().len() as u32).saturating_mul(4));
    }
    let tool_estimate = counter.estimate_provider_visible_tool_segments(
        tool_footprint
            .segments
            .iter()
            .filter(|segment| segment.kind != ProviderVisibleToolSegmentKind::ProviderLateBound)
            .map(|segment| segment.serialized.as_str()),
    );
    let tool_schema_segment_count = tool_footprint
        .segments
        .iter()
        .filter(|segment| segment.kind != ProviderVisibleToolSegmentKind::ProviderLateBound)
        .count();
    let tool_schema_late_bound_segment_count = tool_footprint
        .segments
        .iter()
        .filter(|segment| segment.kind == ProviderVisibleToolSegmentKind::ProviderLateBound)
        .count();
    let input_tokens = message_input_tokens.saturating_add(tool_estimate.input_tokens);
    let ledger_rendered_bytes = session
        .model_context_state
        .as_ref()
        .map(|state| {
            state.events.iter().fold(0usize, |total, event| {
                total.saturating_add(event.rendered_text.len())
            })
        })
        .unwrap_or(0);
    ProjectedRequestUsage {
        message_input_tokens,
        tool_schema_input_tokens: tool_estimate.input_tokens,
        input_tokens,
        tool_schema_serialized_bytes: tool_estimate.serialized_bytes,
        tool_schema_serialized_chars: tool_estimate.serialized_chars,
        tool_schema_segment_count,
        tool_schema_late_bound_segment_count,
        ledger_rendered_bytes,
        history_boundary_input_tokens,
        prepared_message_input_tokens: 0,
    }
}

pub(in crate::runtime::runner) fn required_tool_for_session(
    session: &Session,
) -> Option<&'static str> {
    crate::runtime::runner::session_setup::legacy_instruction::explicit_activation_pending(session)
        .then_some("load_skill")
}

pub(in crate::runtime::runner) fn effective_tool_schemas<'a>(
    session: &Session,
    tool_schemas: &'a [ToolSchema],
) -> Cow<'a, [ToolSchema]> {
    let Some(required_tool) = required_tool_for_session(session) else {
        return Cow::Borrowed(tool_schemas);
    };
    Cow::Owned(
        tool_schemas
            .iter()
            .filter(|schema| schema.function.name == required_tool)
            .cloned()
            .collect(),
    )
}

/// Per-request continuation handles, derived only from the current registry.
/// Reserved exact spellings retain their host-trusted lexical contract; schemas
/// do not attest implementation provenance and qualified shadows are distinct.
pub(super) fn build_exec_session_continuation_context_block(
    session: &Session,
    tool_schemas: &[ToolSchema],
) -> Option<ContextBlock> {
    if !session.allows_model_tool_execution("write_stdin")
        || !effective_tool_schemas(session, tool_schemas)
            .iter()
            .any(|schema| schema.function.name == "write_stdin")
    {
        return None;
    }
    let ids = bamboo_tools::tools::bash_runtime::uncollected_polled_shells_for_session(&session.id);
    if ids.is_empty() {
        return None;
    }
    Some(ContextBlock::new(
        ContextBlockType::RecoverySnapshot,
        ContextBlockPriority::Critical,
        ContextBlockStability::RoundDynamic,
        "Shell sessions awaiting a terminal result",
        format!(
            "Call write_stdin with each session_id below and empty chars to retrieve its terminal result. Continue polling while the tool result includes session_id.\nSession IDs: {}",
            serde_json::to_string(&ids).expect("shell session IDs are serializable"),
        ),
    ))
}

fn legacy_browser_results_for_projection(
    session: &Session,
    tool_schemas: &[ToolSchema],
    loading_mode: CapabilityLoadingMode,
) -> Option<BTreeMap<String, String>> {
    if loading_mode != CapabilityLoadingMode::LegacyFullCatalog
        || !tool_schemas
            .iter()
            .any(|schema| schema.function.name == "browser")
    {
        return None;
    }
    let results =
        crate::runtime::runner::loop_execution::legacy_browser_loaded_result_content(session);
    (!results.is_empty()).then_some(results)
}

fn project_legacy_browser_history(messages: &mut [Message], results: &BTreeMap<String, String>) {
    for message in messages {
        if message.role == Role::Tool
            && message.content_parts.is_none()
            && results
                .get(&message.id)
                .is_some_and(|original| original == &message.content)
        {
            message.content = LOADED_BROWSER_ACKNOWLEDGEMENT.to_string();
        }
    }
}

/// The Supervisor host context sees its complete approved logical catalog;
/// provider measurement and dispatch see exactly the existing wire projection.
fn provider_tool_schemas<'a>(
    session: &Session,
    tool_schemas: &'a [ToolSchema],
    loading_mode: CapabilityLoadingMode,
    required_tool: Option<&str>,
) -> Cow<'a, [ToolSchema]> {
    if session.supervisor_incarnation_id().is_some() {
        super::request_tool_schemas_for_loading_mode(
            session,
            tool_schemas,
            loading_mode,
            required_tool,
        )
    } else {
        Cow::Borrowed(tool_schemas)
    }
}

/// Project the exact PromptIR message input after ledger reconciliation without
/// mutating the live session. Context preparation uses this shadow pass to
/// reserve space for snapshots that are created only after ordinary message
/// fitting (notably when a retention reset seeds a fresh prefix epoch).
pub(super) async fn project_request_usage(
    session: &Session,
    prepared_context: &PreparedContext,
    config: &AgentLoopConfig,
    tool_schemas: &[ToolSchema],
    model: &str,
    llm: &Arc<dyn LLMProvider>,
) -> Result<ProjectedRequestUsage, AgentError> {
    let mut shadow = session.clone();
    let required_tool = required_tool_for_session(&shadow);
    let effective_tool_schemas = effective_tool_schemas(&shadow, tool_schemas);
    let loading_mode = llm.capability_loading_mode(model, required_tool).await;
    let browser_results = legacy_browser_results_for_projection(
        &shadow,
        effective_tool_schemas.as_ref(),
        loading_mode,
    );
    let prepared_message_input_tokens = if let Some(results) = browser_results.as_ref() {
        let mut projected = prepared_context.messages.clone();
        project_legacy_browser_history(&mut projected, results);
        TiktokenTokenCounter::default().count_messages(&projected)
    } else {
        TiktokenTokenCounter::default().count_messages(&prepared_context.messages)
    };
    let envelope = build_request_envelope_reconciled_for_loading_mode(
        &mut shadow,
        prepared_context,
        config,
        effective_tool_schemas.as_ref(),
        model,
        loading_mode,
        None,
    );
    let wire_schemas = provider_tool_schemas(
        &shadow,
        effective_tool_schemas.as_ref(),
        loading_mode,
        required_tool,
    );
    let tool_footprint = llm
        .provider_visible_tool_footprint(&envelope.ir, wire_schemas.as_ref(), model, required_tool)
        .await
        .map_err(|error| {
            AgentError::LLM(format!(
                "provider-visible tool footprint projection failed: {error}"
            ))
        })?;
    let mut usage = measure_request_usage(&shadow, &envelope, &tool_footprint);
    usage.prepared_message_input_tokens = prepared_message_input_tokens;
    Ok(usage)
}

#[cfg(test)]
fn build_request_envelope_reconciled(
    session: &mut Session,
    prepared_context: &PreparedContext,
    config: &AgentLoopConfig,
    tool_schemas: &[ToolSchema],
    model: &str,
) -> PreparedRequestEnvelope {
    build_request_envelope_reconciled_for_loading_mode(
        session,
        prepared_context,
        config,
        tool_schemas,
        model,
        CapabilityLoadingMode::LegacyFullCatalog,
        None,
    )
}

fn build_request_envelope_reconciled_for_loading_mode(
    session: &mut Session,
    prepared_context: &PreparedContext,
    config: &AgentLoopConfig,
    tool_schemas: &[ToolSchema],
    model: &str,
    loading_mode: CapabilityLoadingMode,
    observation_progress_hint: Option<&str>,
) -> PreparedRequestEnvelope {
    let requested_family = ProviderFamily::from_provider_type(config.provider_type.as_deref());
    let requested_protocol = requested_family.map(|family| match family {
        ProviderFamily::OpenAi | ProviderFamily::Copilot => ProviderProtocol::OpenAiResponsesV1,
        ProviderFamily::Anthropic => ProviderProtocol::AnthropicMessages2023_06_01,
    });
    let requested_provider_boundary_sha256 = provider_transcript_boundary_sha256(
        config.provider_name.as_deref(),
        config.provider_type.as_deref(),
    );
    let had_prior_provider_route = session.provider_transcript.active_family().is_some()
        || session.provider_transcript.active_protocol().is_some()
        || session
            .provider_transcript
            .active_provider_boundary_sha256()
            .is_some()
        || !session.provider_transcript.groups().is_empty();
    let provider_route_changed = match (
        requested_family,
        requested_protocol,
        requested_provider_boundary_sha256.as_deref(),
    ) {
        (Some(family), Some(protocol), Some(boundary)) => session
            .activate_provider_transcript_route(family, protocol, boundary)
            .expect("canonical provider family/protocol/boundary must be valid"),
        _ => session.deactivate_provider_transcript_route(),
    };
    let provider_changed = provider_route_changed && had_prior_provider_route;
    if provider_changed {
        session.reset_model_context_epoch(ModelContextResetReason::ProviderSwitch);
    }
    let activated = crate::runtime::runner::session_setup::tool_schemas::effective_guide_activation(
        config, session,
    );
    let (stable_frame, stable_prefix_sections) =
        build_stable_prompt_frame_with_sections(session, config, tool_schemas, &activated);
    let stable_instructions = stable_frame.stable_instructions.clone();

    // Host-owned context (recalled memory, task list, plan state, summary) is
    // reconciled into durable typed events. Initial snapshots lead the real
    // transcript; later changes append after the prior request boundary.
    // Session identity leads this run so a newly seeded ledger places it
    // immediately after the invariant tool-guide prefix. It deliberately does
    // not participate in `build_compression_context_blocks`: a fork/copy must
    // never inherit a stale Session ID through generated summary text.
    let mut context_blocks = vec![build_session_identity_context_block(session)];
    if let Some(block) =
        build_root_orchestration_context_block_for_tools(session, tool_schemas, loading_mode)
    {
        context_blocks.push(block);
    }
    let newly_activated = activated_discoverable_tools(session)
        .difference(&activated)
        .cloned()
        .collect::<std::collections::BTreeSet<_>>();
    if !newly_activated.is_empty() {
        let names = tool_schemas
            .iter()
            .filter(|schema| {
                newly_activated.contains(&bamboo_domain::canonical_tool_name(&schema.function.name))
            })
            .map(|schema| schema.function.name.clone())
            .collect::<Vec<_>>();
        if !names.is_empty() {
            let mut guide_context =
                bamboo_tools::guide::context::GuideBuildContext::from_system_prompt(
                    &stable_instructions,
                );
            guide_context.activated_discoverable_tools = newly_activated;
            guide_context.include_best_practices = false;
            let schemas = tool_schemas
                .iter()
                .filter(|schema| names.contains(&schema.function.name))
                .cloned()
                .collect::<Vec<_>>();
            let content = bamboo_tools::guide::EnhancedPromptBuilder::build_for_tools(
                Some(config.tool_registry.as_ref()),
                &names,
                &schemas,
                &guide_context,
            );
            context_blocks.push(ContextBlock::new(
                ContextBlockType::ToolGuide,
                ContextBlockPriority::High,
                ContextBlockStability::RoundDynamic,
                "Activated Tool Guidance",
                content,
            ));
        }
    }
    if let Some(block) = build_active_workflow_context_block(session) {
        context_blocks.push(block);
    }
    if let Some(block) = build_external_memory_context_block(session) {
        context_blocks.push(block);
    }
    if let Some(block) = build_project_resources_context_block(session) {
        context_blocks.push(block);
    }
    if let Some(block) = build_task_list_context_block(session) {
        context_blocks.push(block);
    }
    // Session goal rides the volatile tail (built directly from the active goal),
    // NOT injected into the system message — so a goal change never invalidates
    // the cached system prefix (goal-leak fix).
    if let Some(block) = build_goal_context_block(config.active_goal()) {
        context_blocks.push(block);
    }
    if let Some(block) = build_agent_hook_context_block(session) {
        context_blocks.push(block);
    }
    // Plan runtime + plan mode blocks are built DIRECTLY from session state (the
    // active PlanModeState + persisted plan artifacts), not reparsed from markers
    // injected into the system message — so the system prefix stays cache-stable
    // across plan transitions.
    if let Some(block) = build_plan_runtime_context_block(session, config.app_data_dir.as_deref()) {
        context_blocks.push(block);
    }
    if let Some(block) = build_plan_mode_context_block(session) {
        context_blocks.push(block);
    }
    if let Some(block) = build_conversation_summary_context_block(session) {
        context_blocks.push(block);
    }
    if let Some(block) = build_history_boundary_context_block(session) {
        context_blocks.push(block);
    }

    let mut system_remainder_messages = Vec::new();
    let mut conversation_messages = Vec::new();
    for message in &prepared_context.messages {
        if matches!(message.role, Role::System) {
            if let Some(remainder_message) =
                derive_system_remainder_message(message, &stable_instructions)
            {
                system_remainder_messages.push(remainder_message);
            }
        } else {
            conversation_messages.push(message.clone());
        }
    }
    // Advisory context belongs only to this request, after the complete tool
    // result batch. Reconcile it before ledger/native transcript anchoring.
    if let Some(hint) = observation_progress_hint {
        let mut message = Message::user(hint);
        if crate::runtime::runner::tool_execution::is_required_progress_feedback(hint) {
            message.metadata = Some(
                serde_json::json!({"runtime_kind":"observation_progress_feedback", "source":"runtime"}),
            );
        }
        conversation_messages.push(message);
    }

    // Canonical prompt structure — where Bamboo OWNS assembly and providers are
    // pure adapters.
    //
    // The tool/server guide (tool schemas + each connected MCP server's
    // `initialize` instructions — e.g. nova's targeting workflow) is relocated OUT
    // of the system prompt and INTO a fixed prefix MESSAGE: the system keeps only
    // invariant identity/directives, and the large, session-stable guide rides as
    // a typed context block at a known position with its own cache
    // breakpoint — so every provider family gets the same static-system structure.
    // The IR's lowering methods derive the chat / Responses-input / continuation
    // views from these runs; the engine no longer pre-bakes any of them.
    let section = |name: &str| -> String {
        stable_prefix_sections
            .iter()
            .find(|s| s.name == name)
            .map(|s| s.content.clone())
            .unwrap_or_default()
    };
    let tool_guide = section("tool_guide");
    let relocate_tool_guide = !tool_guide.trim().is_empty();
    // Keep ONLY cross-session-invariant content in the system field: the static
    // identity (`base`) plus framework core directives. Session-variable
    // context — workspace path, project instruction overlay (CLAUDE.md/AGENTS.md),
    // loaded skills — is relocated into context-block MESSAGES placed AFTER the
    // large, invariant tool guide (see `session_context_messages` below). This
    // keeps the cacheable prefix (system + tool guide) byte-identical across
    // sessions and across a mid-session workspace/skill injection, which is what
    // lets an automatic prefix cache (OpenAI/GLM-style, which has no explicit
    // breakpoints) actually hit on the big block instead of re-reading it.
    // Assemble the cross-session-invariant system field as DISCRETE structured
    // blocks (identity `base`, framework `core_directives`) instead of one glued
    // string. `system_blocks` is the structured form
    // (observability/analysis + a block-native provider can render one wire block
    // per entry); `lane_system` is their byte join for the legacy string wire
    // path. Session-variable context (workspace/instruction/skill) is still
    // relocated into context-block MESSAGES after the tool guide below.
    let mut system_blocks: Vec<bamboo_domain::PromptBlock> = [
        ("base", bamboo_domain::ContextBlockType::Base),
        (
            "core_directives",
            bamboo_domain::ContextBlockType::CoreDirectives,
        ),
    ]
    .into_iter()
    .filter_map(|(name, kind)| {
        let text = section(name);
        let text = if name == "core_directives" {
            append_core_agent_directives("", &text)
        } else {
            text
        };
        (!text.trim().is_empty()).then(|| {
            bamboo_domain::PromptBlock::new(name, kind, text)
                .with_stability(bamboo_domain::ContextBlockStability::Stable)
        })
    })
    .collect();
    // The single system cache breakpoint anchors on the last system block.
    if let Some(last) = system_blocks.last_mut() {
        last.cache_anchor = true;
    }
    let lane_system = system_blocks
        .iter()
        .map(|b| b.text.as_str())
        .collect::<Vec<_>>()
        .join("\n\n");
    let tool_guide_message = relocate_tool_guide.then(|| {
        ContextBlock::new(
            ContextBlockType::ToolGuide,
            ContextBlockPriority::High,
            ContextBlockStability::SessionStable,
            "Tool & Connected-Server Guide",
            tool_guide,
        )
        .render_runtime_context_message()
    });
    let tool_guide_breakpoint_id = tool_guide_message.as_ref().map(|m| m.id.clone());
    // Preserve the zero-tool carrier shape: with no relocated guide, the
    // invariant base + core directives remain a byte-authoritative string and
    // block-native providers see no structured system blocks. Mutable
    // workspace/env/skill material still lives in the ledger below.
    if !relocate_tool_guide {
        system_blocks.clear();
    }

    // Session-variable context rides AFTER the invariant tool guide so it never
    // shifts the cached head. Workspace and instructions are added below from
    // authoritative session metadata, not from these diagnostic sections.
    [
        (
            ContextBlockType::SkillContext,
            ContextBlockPriority::High,
            ContextBlockStability::RoundDynamic,
            "Loaded Skills",
            section("skill"),
        ),
        (
            ContextBlockType::EnvSnapshot,
            ContextBlockPriority::High,
            ContextBlockStability::SessionStable,
            "Environment Snapshot",
            section("env"),
        ),
    ]
    .into_iter()
    .filter(|(_, _, _, _, content)| !content.trim().is_empty())
    .map(|(block_type, priority, stability, title, content)| {
        ContextBlock::new(block_type, priority, stability, title, content)
    })
    .for_each(|block| context_blocks.push(block));
    if let Some(block) = build_workspace_context_block(session) {
        context_blocks.push(block);
    }
    if let Some(block) = build_instruction_overlay_context_block(session) {
        context_blocks.push(block);
    }

    // The Responses-API view (instructions = stable system, guide leading the input
    // array) is no longer pre-baked here: the IR carries the system field + the
    // ordered runs, and the OpenAI/Copilot adapter derives `instructions` /
    // `input_messages` from it via `PromptIR::responses_request_options`.
    let mut stable_prefix_messages = stable_frame.stable_prefix_messages.clone();
    if let Some(message) = tool_guide_message {
        stable_prefix_messages.push(message);
    }

    // Keep Responses item annotations append-only too: a rolling breakpoint on
    // the conversation tail would remove the marker from an item in the prior
    // request. The stable tool-guide boundary is the only generated message
    // breakpoint on the ledger path.
    let mut breakpoint_message_ids = Vec::new();
    // The relocated tool guide is large and session-stable, so it earns a
    // dedicated breakpoint. For legacy/continuation paths this id simply isn't
    // found in the message array, so the breakpoint is harmlessly ignored there.
    if let Some(id) = tool_guide_breakpoint_id {
        breakpoint_message_ids.push(id);
    }
    let cache_plan = PromptCachePlan {
        cache_tools: true,
        cache_system: true,
        breakpoint_message_ids,
        // Use the 1-hour extended cache TTL for the growing append-only prefix
        // (tools + system + tool guide + model transcript). The
        // default 5-minute TTL expires across any pause longer than 5 min
        // (waiting on the user, slow tools, long model think time), forcing a
        // full re-read of the large append-only history — including big tool
        // results — at full price. The 1-hour TTL survives those gaps; the 2x
        // write premium is paid once and amortized over many 0.1x cache reads.
        // Gated behind the `extended-cache-ttl-2025-04-11` beta header, which
        // the Anthropic provider adds whenever the plan's TTL is Extended.
        ttl: CacheTtl::Extended,
    };

    let mut real_transcript = system_remainder_messages;
    real_transcript.extend(conversation_messages);
    let cache_scope = serde_json::json!({
        "model": model,
        "provider_boundary_sha256": &requested_provider_boundary_sha256,
        "system": &lane_system,
        "stable_prefix": stable_prefix_messages
            .iter()
            .map(|message| (&message.role, message.content.as_str()))
            .collect::<Vec<_>>(),
        "tools": tool_schemas,
    });
    let cache_scope_sha256 = hex::encode(Sha256::digest(
        serde_json::to_vec(&cache_scope).expect("cache scope is serializable"),
    ));
    let ledger = super::context_ledger::reconcile_model_context(
        session,
        context_blocks,
        &real_transcript,
        cache_scope_sha256,
        prepared_context.truncation_occurred,
    );
    let mut model_transcript = ledger.transcript;
    if let Some(block) = build_exec_session_continuation_context_block(session, tool_schemas) {
        // This snapshot is current request context, never a second durable
        // handle inventory. Keeping it outside the ledger also removes every
        // stale ID from the next request as soon as its terminal poll consumes it.
        model_transcript.push(block.render_runtime_context_message());
    }

    // The single canonical request: stable prefix followed by one chronological
    // model transcript. Legacy lanes remain supported by PromptIR for external
    // callers, but the normal engine path no longer rebuilds them per round.
    let provider_transcript_groups = requested_family
        .zip(requested_protocol)
        .zip(requested_provider_boundary_sha256.as_deref())
        .map(|((family, protocol), boundary)| {
            session
                .provider_transcript
                .replayable_groups(family, protocol, boundary)
                .into_iter()
                .cloned()
                .collect()
        })
        .unwrap_or_default();
    let mut ir = PromptIR {
        system_text: lane_system,
        system_blocks,
        segments: vec![
            Segment::new(SegmentRole::StablePrefix, stable_prefix_messages),
            Segment::new(SegmentRole::ModelTranscript, model_transcript),
        ],
        provider_transcript_groups,
        cache: cache_plan,
        continuation: None,
    };
    if let Some(results) =
        legacy_browser_results_for_projection(session, tool_schemas, loading_mode)
    {
        if let Some(transcript) = ir
            .segments
            .iter_mut()
            .find(|segment| segment.role == SegmentRole::ModelTranscript)
        {
            project_legacy_browser_history(&mut transcript.messages, &results);
        }
    }

    // Prefix-drift diagnostics must observe only bytes that actually participate
    // in the cacheable head. Keeping dynamic assembly sections here would both
    // report false cache drift and persist raw workspace paths in diagnostic
    // snapshots.
    let stable_prefix_sections = stable_prefix_sections
        .into_iter()
        .filter(|section| matches!(section.name, "base" | "core_directives" | "tool_guide"))
        .collect();

    PreparedRequestEnvelope {
        ir,
        stable_prefix_sections,
        ledger_changed: ledger.changed || provider_changed,
        prefix_epoch: ledger.prefix_epoch,
        prefix_reset_reason: ledger.reset_reason,
    }
}

#[cfg(test)]
fn build_request_envelope(
    session: &Session,
    prepared_context: &PreparedContext,
    config: &AgentLoopConfig,
    tool_schemas: &[ToolSchema],
) -> PreparedRequestEnvelope {
    let mut shadow = session.clone();
    let model = config.model_name.as_deref().unwrap_or(&session.model);
    build_request_envelope_reconciled(&mut shadow, prepared_context, config, tool_schemas, model)
}

/// Session metadata key holding the latest [`RequestRenderObservability`] JSON.
const SESSION_REQUEST_RENDER_KEY: &str = "llm_request_render";

/// Structured, observable record of how the canonical request was rendered for
/// the provider this round — the single place the engine's request-shaping
/// decision is captured: the wire path, the system-field shape, per-lane message
/// counts, and the cache plan. Logged and persisted to session metadata so the
/// rendered request is inspectable after the fact.
#[derive(Debug, Clone, serde::Serialize)]
pub(crate) struct RequestRenderObservability {
    /// `"lanes"` (canonical structured path) or `"responses_continuation"` (delta).
    pub wire: &'static str,
    /// Structured system blocks (0 → the system field is a single joined string).
    pub system_block_count: usize,
    pub system_chars: usize,
    pub stable_prefix_messages: usize,
    pub model_transcript_messages: usize,
    pub dynamic_context_messages: usize,
    pub conversation_messages: usize,
    pub volatile_context_messages: usize,
    /// Messages actually sent: the continuation delta size, or the full chat list.
    pub request_message_count: usize,
    pub tool_count: usize,
    pub message_input_tokens: u32,
    pub tool_schema_input_tokens: u32,
    pub total_input_tokens: u32,
    pub tool_schema_serialized_bytes: usize,
    pub tool_schema_serialized_chars: usize,
    pub tool_schema_segment_count: usize,
    pub tool_schema_late_bound_segment_count: usize,
    pub cache_system: bool,
    pub cache_tools: bool,
    pub cache_breakpoints: usize,
    pub cache_ttl: &'static str,
    pub prefix_epoch: u64,
    pub prefix_reset_reason: Option<&'static str>,
}

impl RequestRenderObservability {
    fn log(&self, session_id: &str) {
        tracing::info!(
            "[{}] LLM request render: wire={} system_blocks={} system_chars={} stable_prefix_msgs={} model_transcript_msgs={} dynamic_ctx_msgs={} conversation_msgs={} volatile_ctx_msgs={} request_msgs={} tools={} known_input_tokens(message={}, tool_schema={}, total={}) tool_schema_shape(known_segments={}, late_bound_segments={}, bytes={}, chars={}) cache(system={}, tools={}, breakpoints={}, ttl={}) prefix_epoch={} prefix_reset_reason={}",
            session_id,
            self.wire,
            self.system_block_count,
            self.system_chars,
            self.stable_prefix_messages,
            self.model_transcript_messages,
            self.dynamic_context_messages,
            self.conversation_messages,
            self.volatile_context_messages,
            self.request_message_count,
            self.tool_count,
            self.message_input_tokens,
            self.tool_schema_input_tokens,
            self.total_input_tokens,
            self.tool_schema_segment_count,
            self.tool_schema_late_bound_segment_count,
            self.tool_schema_serialized_bytes,
            self.tool_schema_serialized_chars,
            self.cache_system,
            self.cache_tools,
            self.cache_breakpoints,
            self.cache_ttl,
            self.prefix_epoch,
            self.prefix_reset_reason.unwrap_or("none"),
        );
    }
}

/// The fully-resolved provider request the engine will dispatch, plus its
/// observability. Produced once per round by [`plan_llm_request`] — the single
/// home for turning the canonical [`PromptIR`] into a concrete provider request
/// and choosing the wire path. This consolidates what used to
/// be inline branching in `execute_llm_stream`, so the request-shaping capability
/// is one testable unit with a structured, inspectable result.
struct LlmRequestPlan {
    request_options: LLMRequestOptions,
    render: RequestRenderObservability,
}

fn cache_ttl_label(ttl: CacheTtl) -> &'static str {
    if ttl == CacheTtl::Extended {
        "1h"
    } else {
        "5m"
    }
}

/// Turn the canonical request envelope into a concrete provider request. The
/// envelope's `ir.continuation` must already be set (at dispatch time) when this is
/// a stateful Responses turn. This builds only the request POLICY — store /
/// verbosity / reasoning / include / cache — and never the prompt wire view: the
/// provider derives `instructions` / `input_messages` / the continuation delta from
/// the IR itself. The returned [`LlmRequestPlan::render`] records the decision.
fn plan_llm_request(
    envelope: &PreparedRequestEnvelope,
    session_id: &str,
    reasoning_effort: Option<ReasoningEffort>,
    tool_count: usize,
    required_tool: Option<&str>,
    usage: ProjectedRequestUsage,
) -> LlmRequestPlan {
    let is_continuation = envelope.ir.continuation.is_some();

    // The engine sets request POLICY only. The Responses prompt wire view
    // (instructions / input_messages / previous_response_id) is derived by the
    // OpenAI/Copilot adapter from the IR via `responses_request_options`.
    let mut responses_options = engine_responses_policy();
    responses_options.prompt_cache_key =
        agent_loop_prompt_cache_key(Some(session_id), Some(AGENT_LOOP_REQUEST_PURPOSE));
    responses_options.prefix_epoch = Some(envelope.prefix_epoch);
    responses_options.prefix_reset_reason = envelope.prefix_reset_reason;

    let request_options = LLMRequestOptions {
        session_id: Some(session_id.to_string()),
        reasoning_effort,
        parallel_tool_calls: Some(required_tool.is_none()),
        required_tool: required_tool.map(str::to_string),
        responses: Some(responses_options),
        request_purpose: Some(AGENT_LOOP_REQUEST_PURPOSE.to_string()),
        cache: Some(envelope.ir.cache.clone()),
    };

    // Observability is sourced from the IR (the single canonical structure), so the
    // recorded shape can't drift from what the lowering methods actually send.
    let render = RequestRenderObservability {
        wire: if is_continuation {
            "responses_continuation"
        } else {
            "model_transcript"
        },
        system_block_count: envelope.ir.system_blocks.len(),
        system_chars: envelope.ir.system_field().len(),
        stable_prefix_messages: envelope.ir.run(SegmentRole::StablePrefix).len(),
        model_transcript_messages: envelope.ir.run(SegmentRole::ModelTranscript).len(),
        dynamic_context_messages: envelope.ir.run(SegmentRole::DynamicContext).len(),
        conversation_messages: envelope.ir.run(SegmentRole::Conversation).len(),
        volatile_context_messages: envelope.ir.run(SegmentRole::VolatileTail).len(),
        request_message_count: if is_continuation {
            envelope.ir.continuation_delta().len()
        } else {
            envelope.ir.flatten().len()
        },
        tool_count,
        message_input_tokens: usage.message_input_tokens,
        tool_schema_input_tokens: usage.tool_schema_input_tokens,
        total_input_tokens: usage.input_tokens,
        tool_schema_serialized_bytes: usage.tool_schema_serialized_bytes,
        tool_schema_serialized_chars: usage.tool_schema_serialized_chars,
        tool_schema_segment_count: usage.tool_schema_segment_count,
        tool_schema_late_bound_segment_count: usage.tool_schema_late_bound_segment_count,
        cache_system: envelope.ir.cache.cache_system,
        cache_tools: envelope.ir.cache.cache_tools,
        cache_breakpoints: envelope.ir.cache.breakpoint_message_ids.len(),
        cache_ttl: cache_ttl_label(envelope.ir.cache.ttl),
        prefix_epoch: envelope.prefix_epoch,
        prefix_reset_reason: envelope
            .prefix_reset_reason
            .map(bamboo_domain::ModelContextResetReason::as_str),
    };

    LlmRequestPlan {
        request_options,
        render,
    }
}

fn persist_request_render_metadata(session: &mut Session, render: &RequestRenderObservability) {
    if let Ok(value) = serde_json::to_string(render) {
        session
            .metadata
            .insert(SESSION_REQUEST_RENDER_KEY.to_string(), value);
    }
}

// Bound only the optional recovery snapshot. Ordinary Session handling and
// attachment limits are unchanged. Serialize into a counting sink before any
// clone so an oversized transcript never creates a second large allocation.
fn live_bootstrap_snapshot(session: &Session) -> Option<Session> {
    const MAX_BYTES: usize = 4 * 1024 * 1024;
    struct Count(usize);
    impl std::io::Write for Count {
        fn write(&mut self, bytes: &[u8]) -> std::io::Result<usize> {
            self.0 = self
                .0
                .checked_add(bytes.len())
                .filter(|size| *size <= MAX_BYTES)
                .ok_or_else(|| std::io::Error::other("live bootstrap snapshot too large"))?;
            Ok(bytes.len())
        }
        fn flush(&mut self) -> std::io::Result<()> {
            Ok(())
        }
    }
    fn skipped(session: &Session) -> Option<Count> {
        let mut count = Count(0);
        let mut add = |bytes: usize| {
            count.0 = count
                .0
                .checked_add(bytes)
                .filter(|size| *size <= MAX_BYTES)?;
            Some(())
        };
        if let Some(value) = &session.force_manual_compression {
            add(value.len())?;
        }
        if let Some((model, _)) = &session.resolved_token_budget {
            add(model.len())?;
        }
        if let Some(runtime) = &session.agent_runtime_state {
            for id in runtime
                .children
                .active_ids
                .iter()
                .chain(&runtime.children.completed_ids)
            {
                add(id.len())?;
                add(std::mem::size_of::<String>())?;
            }
        }
        Some(count)
    }
    let mut compact = skipped(session)?;
    serde_json::to_writer(&mut compact, session).ok()?;
    let mut pretty = skipped(session)?;
    serde_json::to_writer_pretty(&mut pretty, session).ok()?;
    Some(session.clone())
}

pub(super) async fn execute_llm_stream(
    session: &mut Session,
    config: &AgentLoopConfig,
    llm: &Arc<dyn LLMProvider>,
    prepared_context: &PreparedContext,
    tool_schemas: &[ToolSchema],
    frame: &LlmStreamFrame<'_>,
) -> Result<
    (
        crate::runtime::stream::handler::StreamHandlingOutput,
        u128,
        u64,
    ),
    AgentError,
> {
    // Bind frame fields as locals so the rest of the function body stays unchanged.
    let event_tx = frame.event_tx;
    let cancel_token = frame.cancel_token;
    let mut max_context_tokens = frame.max_context_tokens;
    let mut max_output_tokens = frame.max_output_tokens;
    let model = frame.model;
    let provider_name = frame.provider_name;
    let provider_type = frame.provider_type;
    let reasoning_effort = frame.reasoning_effort;
    let session_id = frame.session_id;

    let llm_started_at = std::time::Instant::now();
    let base_tool_schemas = tool_schemas;
    let mut prepared_context = prepared_context.clone();
    let mut required_tool = required_tool_for_session(session);
    let mut effective_schemas = effective_tool_schemas(session, base_tool_schemas);
    // Stateful chaining is gated on the request policy: a `store=false` turn is
    // never persisted upstream, so its id must not be sent back (it would 400
    // with `previous_response_not_found`) nor kept in session metadata.
    let responses_policy = engine_responses_policy();
    let continuation_enabled = bamboo_domain::ChildContextBinding::from_session(session)
        .map_err(|error| AgentError::Budget(error.to_string()))?
        .is_none()
        && responses_continuation_enabled(&responses_policy, provider_type);
    let mut checkpoint_reprepares = 0usize;
    let mut observation_progress_hint = frame.observation_progress_hint;
    let (mut prepared_envelope, previous_response_id, final_usage, loading_mode) = loop {
        let tool_schemas = effective_schemas.as_ref();
        // Owned (not borrowed) so the immutable borrow of `session` ends here and
        // the drift diagnostic below can take `&mut session`.
        let previous_response_id = if continuation_enabled {
            session_previous_response_id(session).map(str::to_string)
        } else {
            None
        };
        let previous_model_context_state = session.model_context_state.clone();
        let previous_provider_transcript = session.provider_transcript.clone();
        let loading_mode = llm.capability_loading_mode(model, required_tool).await;
        let prepared_envelope = build_request_envelope_reconciled_for_loading_mode(
            session,
            &prepared_context,
            config,
            tool_schemas,
            model,
            loading_mode,
            observation_progress_hint,
        );
        // `prepare_round_context` reserves the already-durable ledger history. The
        // reconciliation above can append a new host-state snapshot, so verify the
        // exact final IR again before checkpoint/provider dispatch. Failing closed
        // preserves the context-window contract without committing an unsendable
        // ledger candidate.
        let wire_schemas =
            provider_tool_schemas(session, tool_schemas, loading_mode, required_tool);
        let tool_footprint = match llm
            .provider_visible_tool_footprint(
                &prepared_envelope.ir,
                wire_schemas.as_ref(),
                model,
                required_tool,
            )
            .await
        {
            Ok(footprint) => footprint,
            Err(error) => {
                session.model_context_state = previous_model_context_state;
                session.provider_transcript = previous_provider_transcript;
                return Err(AgentError::LLM(format!(
                    "provider-visible tool footprint projection failed before dispatch: {error}"
                )));
            }
        };
        let final_usage = measure_request_usage(session, &prepared_envelope, &tool_footprint);
        let mut request_input_limit = max_context_tokens.saturating_sub(max_output_tokens);
        if bamboo_domain::ChildContextBinding::from_session(session)
            .map_err(|error| AgentError::Budget(error.to_string()))?
            .is_some()
        {
            request_input_limit =
                request_input_limit.min(prepared_context.token_usage.budget_limit);
        }
        if final_usage.input_tokens > request_input_limit
            || final_usage.ledger_rendered_bytes > MAX_MODEL_CONTEXT_RENDERED_BYTES
        {
            session.model_context_state = previous_model_context_state;
            session.provider_transcript = previous_provider_transcript;
            // Legacy optional advice can be omitted, but the exact recovery
            // control facts must reach the model or fail before dispatch. A
            // budget retry must never silently claim that feedback was sent.
            let required_progress_feedback = observation_progress_hint
                .is_some_and(crate::runtime::runner::tool_execution::is_required_progress_feedback);
            if !required_progress_feedback && observation_progress_hint.take().is_some() {
                continue;
            }
            return Err(AgentError::Budget(format!(
                "final known provider-visible request exceeds ledger-safe limits: message_input_tokens={}, tool_schema_input_tokens={}, input_tokens={}, input_limit={request_input_limit}, tool_schema_known_segments={}, tool_schema_late_bound_segments={}, tool_schema_bytes={}, tool_schema_chars={}, ledger_bytes={}, ledger_byte_limit={MAX_MODEL_CONTEXT_RENDERED_BYTES}",
                final_usage.message_input_tokens,
                final_usage.tool_schema_input_tokens,
                final_usage.input_tokens,
                final_usage.tool_schema_segment_count,
                final_usage.tool_schema_late_bound_segment_count,
                final_usage.tool_schema_serialized_bytes,
                final_usage.tool_schema_serialized_chars,
                final_usage.ledger_rendered_bytes,
            )));
        }
        // Reconciliation changes the next outbound model transcript. It must become
        // durable before any provider request is attempted, otherwise a crash/retry
        // could allocate a different event sequence. The append-safe checkpoint can
        // itself merge a concurrent durable suffix into `session`; after every
        // successful checkpoint, rebuild the bounded context and final envelope from
        // that reconciled Session instead of dispatching the stale candidate.
        if prepared_envelope.ledger_changed {
            if let Some(persistence) = config.persistence.as_ref() {
                if checkpoint_reprepares >= MAX_FINAL_REQUEST_CHECKPOINT_REPREPARES {
                    session.model_context_state = previous_model_context_state;
                    session.provider_transcript = previous_provider_transcript;
                    return Err(AgentError::LLM(format!(
                        "model-context request could not stabilize after {MAX_FINAL_REQUEST_CHECKPOINT_REPREPARES} append-safe checkpoint reprepares"
                    )));
                }
                if let Err(error) = persistence.checkpoint_runtime_session(session).await {
                    // Reconciliation is an in-memory transaction until its durable
                    // checkpoint succeeds. Restore the prior state so an in-process
                    // retry cannot mistake the failed candidate for a committed
                    // ledger and bypass persistence on its next attempt.
                    session.model_context_state = previous_model_context_state;
                    session.provider_transcript = previous_provider_transcript;
                    return Err(AgentError::LLM(format!(
                        "model-context ledger checkpoint failed before provider dispatch: {error}"
                    )));
                }
                checkpoint_reprepares += 1;
                required_tool = required_tool_for_session(session);
                effective_schemas = effective_tool_schemas(session, base_tool_schemas);
                let reprepared = Box::pin(super::context_preparation::prepare_round_context(
                    session,
                    config,
                    model,
                    session_id,
                    effective_schemas.as_ref(),
                    llm,
                    Some(event_tx),
                ))
                .await?;
                max_context_tokens = reprepared.budget.max_context_tokens;
                max_output_tokens = reprepared.budget.max_output_tokens;
                prepared_context = reprepared.prepared_context;
                continue;
            }
        }
        break (
            prepared_envelope,
            previous_response_id,
            final_usage,
            loading_mode,
        );
    };
    let wire_schemas = provider_tool_schemas(
        session,
        effective_schemas.as_ref(),
        loading_mode,
        required_tool,
    );
    let tool_schemas = wire_schemas.as_ref();
    // Freeze the already-reconciled request's Session before the following
    // diagnostic-only mutations. The final IR, schemas and options below are
    // then retained by the very same pinned provider future; no retry may
    // rebuild them from this snapshot.
    let live_bootstrap = match (
        config.live_bootstrap_host.as_ref(),
        config.persistence.as_ref(),
        config.session_inbox.as_ref(),
    ) {
        (Some(host), Some(persistence), Some(inbox)) => persistence
            .root_actor_writer()
            .filter(|owner| {
                owner.fence.actor_id == session.id
                    && owner.created_at == session.created_at
                    && session.kind == bamboo_domain::SessionKind::Root
                    && session.authority_identity.is_ordinary()
            })
            .and_then(|owner| {
                live_bootstrap_snapshot(session).map(|snapshot| {
                    (
                        host.clone(),
                        persistence.clone(),
                        inbox.clone(),
                        owner,
                        snapshot,
                    )
                })
            }),
        _ => None,
    };
    // Side-channel diagnostic: record whether the cacheable stable prefix drifted
    // from the previous round (esp. shrinks, which drop cached content). Never
    // affects what is sent below.
    super::prefix_drift::record_prefix_drift(
        session,
        config.app_data_dir.as_deref(),
        &prepared_envelope.stable_prefix_sections,
    );
    // Set the stateful Responses continuation on the IR (boundary = the last
    // assistant turn in the model transcript) BEFORE planning, so the provider — or
    // the default lowering — derives the delta itself from addressable runs and the
    // plan's wire label reflects it.
    if let Some(response_id) = previous_response_id.as_deref() {
        let last_committed_assistant_id = prepared_envelope
            .ir
            .run(SegmentRole::ModelTranscript)
            .iter()
            .rev()
            .find(|message| matches!(message.role, Role::Assistant))
            .map(|message| message.id.clone());
        prepared_envelope.ir.continuation = Some(Continuation {
            previous_response_id: response_id.to_string(),
            last_committed_assistant_id,
        });
    }

    // Single home for turning the canonical envelope into a concrete provider
    // request (request POLICY + observability). The ledger keeps every previously
    // emitted item byte-stable while appending real turns and context events.
    let planned = plan_llm_request(
        &prepared_envelope,
        session_id,
        reasoning_effort,
        tool_schemas.len(),
        required_tool,
        final_usage,
    );
    if !continuation_enabled {
        tracing::debug!(
            "[{}] Responses API previous_response_id disabled (store={:?}, provider={})",
            session_id,
            responses_policy.store,
            provider_name.unwrap_or("unknown")
        );
    }
    // Structured, inspectable record of how this request was rendered.
    planned.render.log(session_id);
    persist_request_render_metadata(session, &planned.render);

    let mut timeout_context = crate::runtime::stream::handler::StreamTimeoutContext::new(
        config.stream_timeout,
        provider_name,
        Some(model),
    )
    .allow_turn_retry_before_semantic_output()
    .begin_request();

    // ONE canonical dispatch: every provider renders the IR. The default lowering
    // (chat / continuation-delta) and the provider overrides (Anthropic block-native
    // system, OpenAI/Copilot Responses view) all derive their wire from this IR.
    // The provider future itself is covered by the same transport-idle policy as
    // the returned stream. This bounds proxies that accept the request but never
    // return response headers, a phase the per-frame watchdog cannot observe.
    if let Some(binding) = bamboo_domain::ChildContextBinding::from_session(session)
        .map_err(|error| AgentError::Budget(error.to_string()))?
    {
        // Check the provider-bound IR after reconciliation/checkpoint/reprepare,
        // rather than treating a retained Session message as proof of delivery.
        binding
            .validate_messages(&session.id, &prepared_envelope.ir.body_chat())
            .map_err(|error| AgentError::Budget(error.to_string()))?;
        if prepared_envelope.ir.continuation.is_some() {
            return Err(AgentError::Budget(
                bamboo_domain::ChildContextPacketError::Invalid.to_string(),
            ));
        }
    }
    if cancel_token.is_cancelled() {
        return Err(AgentError::Cancelled);
    }
    // Start only at the actual dispatch boundary: failed prompt preflight is
    // not an LLM request. The same span remains alive through stream consumption.
    let observation =
        bamboo_infrastructure::telemetry::Observation::llm(provider_type, model, session_id);
    let call_id = uuid::Uuid::new_v4().to_string();
    observation.model_call_id(&call_id);
    let trace_ids = observation.trace_ids();
    let mut usage_guard = ModelUsageGuard {
        runner: config.hook_runner.clone(),
        session_birth: session.created_at,
        root_session_id: bamboo_infrastructure::telemetry::session_reference(
            &session.root_session_id,
        ),
        snapshot: bamboo_infrastructure::telemetry::session_reference(session_id).map(
            |session_id| bamboo_domain::ModelCallUsage {
                schema_version: 1,
                event_id: call_id.clone(),
                call_id,
                run_id: bamboo_infrastructure::telemetry::current_run_id(&session_id),
                session_id,
                provider: provider_type.and_then(bamboo_infrastructure::telemetry::safe_provider),
                model: bamboo_infrastructure::telemetry::safe_identifier(model),
                status: bamboo_domain::ModelCallStatus::Cancelled,
                trace_id: trace_ids.as_ref().map(|ids| ids.0.clone()),
                span_id: trace_ids.map(|ids| ids.1),
                usage: None,
            },
        ),
    };
    let (http_retry_observer, live_offer) = if live_bootstrap.is_some() {
        let (observer, receiver) =
            bamboo_llm::retry::InitialHttpRetryObserver::with_live_bootstrap();
        (observer, Some(receiver))
    } else {
        (
            Arc::new(bamboo_llm::retry::InitialHttpRetryObserver::default()),
            None,
        )
    };
    let bootstrap_result = {
        let bootstrap = observation.scope(bamboo_llm::retry::with_upstream_retry(
            config.upstream_retry,
            crate::runtime::stream::handler::await_stream_bootstrap_observed(
                llm.chat_stream_ir(
                    &prepared_envelope.ir,
                    tool_schemas,
                    Some(max_output_tokens),
                    model,
                    Some(&planned.request_options),
                ),
                cancel_token,
                session_id,
                &timeout_context,
                http_retry_observer.clone(),
            ),
        ));
        tokio::pin!(bootstrap);
        let mut live_guard: Option<Box<dyn Send>> = None;
        let result = match (live_bootstrap, live_offer) {
            (Some((host, persistence, inbox, owner, snapshot)), Some(receiver)) => {
                tokio::select! {
                    biased;
                    _ = cancel_token.cancelled() => Err(AgentError::Cancelled),
                    result = &mut bootstrap => result,
                    received = receiver => {
                        if let Ok(offer) = received {
                            let checkpoint = tokio::select! {
                                biased;
                                _ = cancel_token.cancelled() => Err(std::io::Error::other("run cancelled")),
                                result = persistence.checkpoint_live_bootstrap(&snapshot, inbox.clone(), offer.gate()) => result,
                            };
                            if let Ok(proof) = checkpoint {
                                if offer.matches_proof(&proof) && !cancel_token.is_cancelled() {
                                    let registered = tokio::select! {
                                        biased;
                                        _ = cancel_token.cancelled() => Err(std::io::Error::other("run cancelled")),
                                        result = host.register(owner, proof, persistence, inbox, offer) => result,
                                    };
                                    if let Ok(guard) = registered {
                                        live_guard = Some(guard);
                                    }
                                }
                            }
                        }
                        bootstrap.await
                    }
                }
            }
            _ => bootstrap.await,
        };
        drop(live_guard);
        result
    };
    let stream = bootstrap_result.and_then(|result| {
        result.map_err(|error| {
            if matches!(error, bamboo_llm::provider::LLMError::RequestFailed(_)) {
                if let Some(diagnostic) = http_retry_observer
                    .terminal_diagnostic()
                    .filter(|d| d.matches_error_message(&error.to_string()))
                {
                    // The existing terminal Session checkpoint carries this
                    // safe receipt. No provider error text enters the DTO.
                    session.record_llm_request_diagnostic(diagnostic);
                }
            }
            if matches!(
                error,
                bamboo_llm::provider::LLMError::RequestFailed(_)
                    | bamboo_llm::provider::LLMError::Auth(_)
            ) {
                // Authentication types outrank text in an upstream body:
                // even overflow-looking text cannot authorize another call.
                return AgentError::LLMNonRetryable(error.to_string());
            }
            // Request or response body errors can happen after request
            // bytes were sent. Their source text must not trigger model
            // context recovery that replays the request.
            let body_failure = is_http_body_failure(&error);
            let message = format_provider_error(error);
            if body_failure {
                AgentError::LLMNonRetryable(message)
            } else if is_llm_overflow_error(&message) {
                AgentError::LLMOverflow(message)
            } else if http_retry_observer.request_budget_managed() {
                // Terminal auth, request-validation and transport failures
                // must not reset the same provider-owned request budget.
                AgentError::LLMNonRetryable(message)
            } else {
                AgentError::LLM(message)
            }
        })
    });
    let stream = match stream {
        Ok(stream) => stream,
        Err(error) => {
            usage_guard.finish(usage_status(&error), None);
            observation.finish(crate::runtime::runner::telemetry_error_outcome(&error));
            return Err(error);
        }
    };
    let provider_manages_retry_budget = http_retry_observer.request_budget_managed();
    if provider_manages_retry_budget {
        timeout_context.managed_response_established();
    }

    // A successful stream bootstrap is the first point at which the final
    // provider-visible PromptIR is known to have been accepted. Capture the
    // typed fresh-selection snapshot once here; retries with the same round id
    // are harmless because SQLite atomically preserves the first snapshot.
    // Historical compact text already present in the append-only context ledger
    // is intentionally not reparsed or backfilled into this round's observation.
    if let (Some(collector), Some(exposure)) = (
        config.metrics_collector.as_ref(),
        frame.prompt_memory_exposure,
    ) {
        collector.prompt_memory_exposure(exposure.provenance.observation(
            exposure.round_id,
            session_id,
            chrono::Utc::now(),
        ));
    }

    // Carry the last completed provider result through prompt preparation. A
    // newly prepared budget is not evidence of a real zero-cache provider
    // response, so consumers must not flash it as Cache 0%.
    let previous_cache_read_input_tokens = session
        .token_usage
        .as_ref()
        .map(|usage| usage.cache_read_input_tokens)
        .unwrap_or(0);
    let previous_provider_prompt_usage = session
        .token_usage
        .as_ref()
        .and_then(|usage| usage.provider_prompt_usage)
        .map(|mut usage| {
            usage.retained_from_previous_call = true;
            usage
        });

    // Send token budget update AFTER LLM call succeeds.
    // This timing gives frontend time to subscribe to /events endpoint.
    let usage = TokenBudgetUsage {
        system_tokens: prepared_context.token_usage.system_tokens,
        summary_tokens: prepared_context.token_usage.summary_tokens,
        window_tokens: prepared_context.token_usage.window_tokens,
        total_tokens: prepared_context.token_usage.total_tokens,
        max_context_tokens,
        budget_limit: prepared_context.token_usage.budget_limit,
        truncation_occurred: prepared_context.truncation_occurred,
        segments_removed: prepared_context.segments_removed,
        prompt_cached_tool_outputs: prepared_context.prompt_cached_tool_outputs,
        prompt_cached_tool_tokens_saved: prepared_context.prompt_cached_tool_tokens_saved,
        thinking_tokens: 0,
        cache_read_input_tokens: previous_cache_read_input_tokens,
        provider_prompt_usage: previous_provider_prompt_usage,
    };

    session.token_usage = Some(usage.clone());

    let budget_event = AgentEvent::TokenBudgetUpdated { usage };
    if let Err(error) = event_tx.send(budget_event).await {
        tracing::warn!(
            "[{}] Failed to send token budget event: {}",
            session_id,
            error
        );
    }

    // A single structured explicit workflow has a fail-closed first step: the
    // model must call load_skill before any answer tokens become user-visible.
    // Keep that first stream silent; the pipeline verifies and executes the
    // model-issued call, then later rounds stream normally once activation is
    // mirrored into the runner-owned Session.
    let activation_pending =
        crate::runtime::runner::session_setup::legacy_instruction::explicit_activation_pending(
            session,
        );
    // The opt-in evidence check must publish only the final, checked answer.
    // Buffer this response until tool calls tell us whether it is a candidate.
    let stream_output_result = observation
        .scope(async {
            if activation_pending || config.features_final_evidence_check {
                crate::runtime::stream::handler::consume_llm_stream_silent_with_context_and_partial(
                    stream,
                    cancel_token,
                    session_id,
                    &timeout_context,
                )
                .await
            } else {
                crate::runtime::stream::handler::consume_llm_stream_with_context_and_partial(
                    stream,
                    event_tx,
                    cancel_token,
                    session_id,
                    &timeout_context,
                )
                .await
            }
        })
        .await;
    let mut stream_output = match stream_output_result {
        Ok(output) => output,
        Err(mut failure) => {
            // A generic SSE/body error is no safer to replay than a watchdog
            // expiry once text, reasoning or a tool-call delta was observed.
            let replay_unsafe = provider_manages_retry_budget
                || !failure.partial_output.content.is_empty()
                || !failure.partial_output.reasoning_content.is_empty()
                || !failure.partial_output.partial_tool_calls.is_empty();
            failure.error = match failure.error {
                AgentError::LLM(message) if replay_unsafe => AgentError::LLMNonRetryable(message),
                other => other,
            };
            if let Some(usage) = failure.partial_output.provider_usage {
                observation.model_usage(&token_usage(usage));
            }
            usage_guard.finish(
                usage_status(&failure.error),
                failure.partial_output.provider_usage,
            );
            observation.finish(crate::runtime::runner::telemetry_error_outcome(
                &failure.error,
            ));
            let appended = append_interrupted_assistant_output(
                session,
                *failure.partial_output,
                &failure.error,
            );
            if appended {
                tracing::warn!(
                    "[{}] Preserved interrupted partial assistant output in the live transcript",
                    session_id
                );
            }
            return Err(failure.error);
        }
    };

    if let Some(usage) = stream_output.provider_usage {
        observation.model_usage(&token_usage(usage));
    }
    usage_guard.finish(
        bamboo_domain::ModelCallStatus::Success,
        stream_output.provider_usage,
    );
    observation.finish(bamboo_infrastructure::telemetry::Outcome::Success);

    if config.features_final_evidence_check
        && !activation_pending
        && !stream_output.tool_calls.is_empty()
    {
        if stream_output.visible_message.is_none() && !stream_output.reasoning_content.is_empty() {
            let message = Message::assistant(String::new(), None);
            stream_output.visible_message =
                Some(crate::runtime::stream::handler::VisibleMessageIdentity {
                    message_id: message.id,
                    created_at: message.created_at,
                });
        }
        if let Some(identity) = stream_output.visible_message.as_ref() {
            crate::runtime::stream::handler::publish_buffered_response(
                event_tx,
                identity,
                &stream_output.content,
                Some(&stream_output.reasoning_content),
            )
            .await;
        }
    }

    // Update session token usage with actual output/thinking/cache stats from the LLM response.
    if let Some(ref mut usage) = session.token_usage {
        usage.thinking_tokens = stream_output.thinking_tokens as u32;
        usage.cache_read_input_tokens = stream_output.cache_read_input_tokens as u32;
        let provider_prompt_usage = ProviderPromptUsage {
            input_tokens: stream_output.input_tokens,
            cache_creation_input_tokens: stream_output.cache_creation_input_tokens,
            cache_read_input_tokens: stream_output.cache_read_input_tokens,
            retained_from_previous_call: false,
        };
        // This legacy DTO has no per-field unknown representation. Avoid
        // publishing an exact-looking cache ratio when the authoritative
        // provider snapshot omitted or invalidated its input or cache count.
        let prompt_fields_known = stream_output.provider_usage.is_none_or(|snapshot| {
            snapshot.input_tokens.is_some() && snapshot.cache_read_input_tokens.is_some()
        });
        usage.provider_prompt_usage = (prompt_fields_known
            && provider_prompt_usage.total_input_tokens() > 0)
            .then_some(provider_prompt_usage);
    }

    if let Some(usage) = session.token_usage.clone() {
        let final_budget_event = AgentEvent::TokenBudgetUpdated { usage };
        if let Err(error) = event_tx.send(final_budget_event).await {
            tracing::warn!(
                "[{}] Failed to send final token budget event: {}",
                session_id,
                error
            );
        }
    }

    let cache_write_input_tokens = stream_output
        .provider_usage
        .and_then(|usage| usage.cache_write_input_tokens)
        .unwrap_or(0);
    if stream_output.cache_creation_input_tokens > 0
        || stream_output.cache_read_input_tokens > 0
        || cache_write_input_tokens > 0
    {
        tracing::info!(
            "[{}] Provider prompt cache: creation={}, read={}, write={}, output={}, thinking={}",
            session_id,
            stream_output.cache_creation_input_tokens,
            stream_output.cache_read_input_tokens,
            cache_write_input_tokens,
            stream_output.output_tokens,
            stream_output.thinking_tokens,
        );
    }

    // Append a per-call record to the session's dedicated, append-only
    // `token-usage.jsonl` (next to session.json) for offline cache/cost
    // analysis. `session.token_usage` only keeps the latest overwritten
    // snapshot; this log keeps the full per-round history. `cache_creation`
    // lives only on the stream output (not on the budget snapshot), so it is
    // read from there.
    //
    // Gated to dev: active in any debug build, and in release only when the
    // `token-usage-log` feature is enabled. `cfg!(...)` keeps the code compiled
    // either way (no unused-binding churn); the compiler eliminates the block in
    // a release build with the feature off, so nothing is written.
    if cfg!(any(debug_assertions, feature = "token-usage-log")) {
        if let Some(persistence) = config.persistence.as_ref() {
            let record = crate::token_usage_log::TokenUsageRecord::new(
                chrono::Utc::now().to_rfc3339(),
                session_id,
                model,
                provider_name.unwrap_or(""),
                session.messages.len(),
                session.token_usage.as_ref(),
                stream_output.cache_creation_input_tokens,
                stream_output.cache_read_input_tokens,
                cache_write_input_tokens,
                stream_output.input_tokens,
                stream_output.output_tokens,
                stream_output.thinking_tokens,
                context_management_telemetry(session, config),
            );
            match record.to_json_line() {
                Ok(line) => {
                    if let Err(error) = persistence
                        .append_token_usage_record(session_id, &line)
                        .await
                    {
                        tracing::warn!(
                            "[{}] Failed to append token-usage record: {}",
                            session_id,
                            error
                        );
                    }
                }
                Err(error) => {
                    tracing::warn!(
                        "[{}] Failed to serialize token-usage record: {}",
                        session_id,
                        error
                    );
                }
            }
        }
    }

    if continuation_enabled {
        if let Some(response_id) = stream_output
            .response_id
            .as_deref()
            .map(str::trim)
            .filter(|value| !value.is_empty())
        {
            session.metadata.insert(
                SESSION_RESPONSES_PREVIOUS_RESPONSE_ID_KEY.to_string(),
                response_id.to_string(),
            );
        } else {
            session
                .metadata
                .remove(SESSION_RESPONSES_PREVIOUS_RESPONSE_ID_KEY);
        }
    } else {
        session
            .metadata
            .remove(SESSION_RESPONSES_PREVIOUS_RESPONSE_ID_KEY);
    }

    let llm_duration = llm_started_at.elapsed().as_millis();
    let prompt_tokens = super::token_estimation::estimate_prompt_tokens(&prepared_context.messages);

    Ok((stream_output, llm_duration, prompt_tokens))
}

#[cfg(test)]
mod tests;