bamboo-engine 2026.8.26

use std::borrow::Cow;
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, strip_existing_core_directives, strip_existing_external_memory,
    strip_existing_goal, strip_existing_plan_mode_instructions,
    strip_existing_plan_runtime_context, strip_existing_task_list,
};
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_plan_mode_context_block, build_plan_runtime_context_block,
    build_project_resources_context_block, build_task_list_context_block,
};
use crate::runtime::runner::session_setup::prompt_setup::{
    build_stable_prompt_frame_with_sections, StablePrefixSection,
};
use bamboo_agent_core::agent::events::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_domain::ReasoningEffort;
use bamboo_domain::MAX_MODEL_CONTEXT_RENDERED_BYTES;
use bamboo_llm::provider::ResponsesRequestOptions;
use bamboo_llm::{
    CacheTtl, Continuation, LLMProvider, LLMRequestOptions, PromptCachePlan, PromptIR, Segment,
    SegmentRole,
};
use bamboo_tools::exposure::activated_discoverable_tools;
use sha2::{Digest, Sha256};

/// 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,
}

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";

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(_) => "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),
    );
    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>,
) -> bool {
    let interrupted = session.messages.last().is_some_and(|message| {
        if Some(message.id.as_str()) == attempt_tail_message_id {
            return false;
        }
        message
            .metadata
            .as_ref()
            .and_then(|metadata| metadata.get("runtime_kind"))
            .and_then(serde_json::Value::as_str)
            == Some(INTERRUPTED_ASSISTANT_OUTPUT_KIND)
    });
    if interrupted {
        session.messages.pop();
        session.updated_at = chrono::Utc::now();
    }
    interrupted
}

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

    let without_external_memory = strip_existing_external_memory(&message.content);
    let without_task_list = strip_existing_task_list(&without_external_memory);
    let without_plan_mode = strip_existing_plan_mode_instructions(&without_task_list);
    let without_plan_runtime = strip_existing_plan_runtime_context(&without_plan_mode);
    // Strip a legacy goal block too: the goal now rides the volatile tail (built from
    // the active goal), so a stale `<!-- BAMBOO_GOAL_START -->` left in an old
    // persisted System message must not resurface as a duplicate in the remainder.
    let without_goal = strip_existing_goal(&without_plan_runtime);
    // Framework directives always live in the assembled system field and are not
    // part of the persisted base, so strip them from both sides before comparing.
    // The directives are inserted INTO the `base` section (between the base text
    // and the skill/tool-guide/workspace/env contexts) — not as a leading prefix
    // of the whole frame — so stripping them from both the persisted message and
    // the reference makes the comparison apples-to-apples (base+contexts vs
    // base+contexts). Without it, the directive-bearing reference no longer equals
    // the directive-free persisted base, so the dedup falls through and re-emits
    // the entire base as a redundant system message. Stripping both sides also
    // discards a STALE directive block in an old persisted message, so the current
    // directives (already in the system field) win.
    let without_directives = strip_existing_core_directives(&without_goal);
    let trimmed = without_directives.trim();
    if trimmed.is_empty() {
        return None;
    }

    let stable_without_directives = strip_existing_core_directives(stable_instructions);
    let stable_trimmed = stable_without_directives.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)]
pub(super) struct ProjectedRequestUsage {
    pub input_tokens: u32,
    pub ledger_rendered_bytes: usize,
}

fn measure_request_usage(
    session: &Session,
    envelope: &PreparedRequestEnvelope,
) -> ProjectedRequestUsage {
    let input_tokens = TiktokenTokenCounter::default().count_messages(&envelope.ir.flatten());
    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 {
        input_tokens,
        ledger_rendered_bytes,
    }
}

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

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(),
    )
}

/// 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) fn project_request_usage(
    session: &Session,
    prepared_context: &PreparedContext,
    config: &AgentLoopConfig,
    tool_schemas: &[ToolSchema],
    model: &str,
) -> ProjectedRequestUsage {
    let mut shadow = session.clone();
    let effective_tool_schemas = effective_tool_schemas(&shadow, tool_schemas);
    let envelope = build_request_envelope_reconciled(
        &mut shadow,
        prepared_context,
        config,
        effective_tool_schemas.as_ref(),
        model,
    );
    measure_request_usage(&shadow, &envelope)
}

fn build_request_envelope_reconciled(
    session: &mut Session,
    prepared_context: &PreparedContext,
    config: &AgentLoopConfig,
    tool_schemas: &[ToolSchema],
    model: &str,
) -> PreparedRequestEnvelope {
    let activated = activated_discoverable_tools(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.
    let mut context_blocks = Vec::new();
    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);
    }

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

    // 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 (workspace path, project instruction overlay,
    // loaded skills), relocated to ride AFTER the invariant tool guide so it never
    // shifts the cached head. Each block is session-stable and emitted only when
    // present; keeping them as separate blocks lets an unchanged block stay inside
    // the shared prefix even when a sibling changes. The instruction overlay keeps
    // Critical priority so its authority survives the move out of the system field.
    [
        (
            ContextBlockType::Workspace,
            ContextBlockPriority::High,
            "Project & Workspace",
            [section("project"), section("workspace")]
                .into_iter()
                .filter(|value| !value.trim().is_empty())
                .collect::<Vec<_>>()
                .join("\n\n"),
        ),
        (
            ContextBlockType::InstructionOverlay,
            ContextBlockPriority::Critical,
            "Project Instructions",
            section("instruction"),
        ),
        (
            ContextBlockType::SkillContext,
            ContextBlockPriority::High,
            "Loaded Skills",
            section("skill"),
        ),
        (
            ContextBlockType::EnvSnapshot,
            ContextBlockPriority::High,
            "Environment Snapshot",
            section("env"),
        ),
    ]
    .into_iter()
    .filter(|(_, _, _, content)| !content.trim().is_empty())
    .map(|(block_type, priority, title, content)| {
        ContextBlock::new(
            block_type,
            priority,
            ContextBlockStability::SessionStable,
            title,
            content,
        )
    })
    .for_each(|block| 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,
        "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,
    );

    // 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 ir = PromptIR {
        system_text: lane_system,
        system_blocks,
        segments: vec![
            Segment::new(SegmentRole::StablePrefix, stable_prefix_messages),
            Segment::new(SegmentRole::ModelTranscript, ledger.transcript),
        ],
        cache: cache_plan,
        continuation: None,
    };

    PreparedRequestEnvelope {
        ir,
        stable_prefix_sections,
        ledger_changed: ledger.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 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={} 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.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>,
) -> 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,
        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);
    }
}

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), 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 max_context_tokens = frame.max_context_tokens;
    let 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 required_tool = required_tool_for_session(session);
    let effective_tool_schemas = effective_tool_schemas(session, tool_schemas);
    let tool_schemas = effective_tool_schemas.as_ref();
    // 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 = responses_continuation_enabled(&responses_policy, provider_type);
    // 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 mut prepared_envelope =
        build_request_envelope_reconciled(session, prepared_context, config, tool_schemas, model);
    // `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 final_usage = measure_request_usage(session, &prepared_envelope);
    let request_input_limit = max_context_tokens.saturating_sub(max_output_tokens);
    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;
        return Err(AgentError::Budget(format!(
            "final PromptIR message input exceeds ledger-safe limits: input_tokens={}, input_limit={request_input_limit}, ledger_bytes={}, ledger_byte_limit={MAX_MODEL_CONTEXT_RENDERED_BYTES}",
            final_usage.input_tokens, 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. Fail closed on checkpoint
    // failure instead of sending an unrepeatable request.
    if prepared_envelope.ledger_changed {
        if let Some(persistence) = config.persistence.as_ref() {
            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;
                return Err(AgentError::LLM(format!(
                    "model-context ledger checkpoint failed before provider dispatch: {error}"
                )));
            }
        }
    }
    // 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,
    );
    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 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.
    let stream = crate::runtime::stream::handler::await_stream_bootstrap(
        llm.chat_stream_ir(
            &prepared_envelope.ir,
            tool_schemas,
            Some(max_output_tokens),
            model,
            Some(&planned.request_options),
        ),
        cancel_token,
        session_id,
        &timeout_context,
    )
    .await?
    .map_err(|error| {
        let message = format_provider_error(error);
        if is_llm_overflow_error(&message) {
            AgentError::LLMOverflow(message)
        } else {
            AgentError::LLM(message)
        }
    })?;

    // 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: 0,
    };

    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 stream_output_result =
        if crate::runtime::runner::session_setup::skill_context::explicit_activation_pending(
            session,
        ) {
            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
        };
    let stream_output = match stream_output_result {
        Ok(output) => output,
        Err(failure) => {
            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);
        }
    };

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

    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,
            );
            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();

    Ok((stream_output, llm_duration))
}

#[cfg(test)]
mod tests;