use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex as StdMutex, Weak};

use chrono::{SecondsFormat, Utc};
use sqlx::SqlitePool;
use tiycore::agent::{
    Agent, AgentError, AgentMessage, PromoteError, QueueEvent, QueueMode, QueuedMessageHandle,
    ToolExecutionMode,
};
use tiycore::thinking::ThinkingLevel;
use tiycore::types::{Cost, InputType, Model, Provider, Usage};
use tokio::sync::mpsc;

use crate::core::app_state::GoalRuntimeState;
use crate::core::prompt;
use crate::core::subagent::HelperAgentOrchestrator;
use crate::core::tool_gateway::ToolGateway;
use crate::core::TIYCORE_REQUEST_MAX_RETRIES;
use crate::extensions::ExtensionsManager;
use crate::ipc::frontend_channels::ThreadStreamEvent;
use crate::model::errors::{AppError, ErrorSource};
use crate::persistence::repo::{message_repo, provider_repo, run_repo, tool_call_repo};

/// Deprecated: previously used as the hard limit for `message_repo::list_recent` in
/// `build_session_spec`.  Replaced by `message_repo::list_since_last_reset()` which
/// queries the DB for the reset boundary directly.  Retained for reference only.
#[allow(dead_code)]
const MESSAGE_HISTORY_LIMIT: i64 = 200;
pub(crate) const DEFAULT_CONTEXT_WINDOW: u32 = 128_000;
const DEFAULT_MAX_OUTPUT_TOKENS: u32 = 32_000;
pub(crate) const DEFAULT_FULL_TOOL_PROFILE: &str = "default_full";
pub(crate) const PLAN_READ_ONLY_TOOL_PROFILE: &str = "plan_read_only";
pub(crate) const STANDARD_TOOL_TIMEOUT_SECS: u64 = 120;
const SUBAGENT_TOOL_TIMEOUT_SECS: u64 = 600;
/// Main agent timeout is effectively unlimited (24 h) because user-interactive
/// tools like `clarify` and approval prompts must wait for human input without
/// being killed by the outer tiycore timeout.  The per-tool execution timeout
/// inside `tool_gateway` (120 s) still guards against runaway non-interactive
/// tool calls.
const MAIN_AGENT_TOOL_TIMEOUT_SECS: u64 = 86_400;
pub(crate) const CLARIFY_TOOL_NAME: &str = "clarify";
pub(crate) const PLAN_TOOL_NAME: &str = "update_plan";
pub(crate) const RENDER_TOOL_NAME: &str = "render";
pub(crate) const TASK_TOOL_NAMES: &[&str] = &["create_task", "update_task", "query_task"];
pub(crate) const PLAN_MODE_MISSING_CHECKPOINT_ERROR: &str =
    "Plan mode requires publishing a plan with update_plan before the run can finish.";
pub(crate) const TEXT_ATTACHMENT_MAX_CHARS: usize = 12_000;
const RUNTIME_QUEUE_MAX_MESSAGES: usize = 80;
const RUNTIME_QUEUE_MAX_EVENTS: usize = 80;

#[derive(Debug, Default)]
struct RuntimeQueueState {
    messages: Vec<RuntimeQueueMessageDto>,
    events: Vec<RuntimeQueueEventDto>,
    pending_removed_message_ids: Vec<String>,
    pending_promoted_message_ids: Vec<String>,
}

fn runtime_queue_timestamp() -> String {
    Utc::now().to_rfc3339_opts(SecondsFormat::Millis, true)
}

fn mark_runtime_queue_messages(
    messages: &mut [RuntimeQueueMessageDto],
    kind: AgentQueueMessageKind,
    count: usize,
    status: RuntimeQueueMessageStatus,
    updated_at: &str,
) -> Vec<RuntimeQueueMessageDto> {
    let mut remaining = count;
    let mut marked = Vec::new();
    for message in messages.iter_mut() {
        if remaining == 0 {
            break;
        }
        if message.kind == kind && message.status == RuntimeQueueMessageStatus::Pending {
            message.status = status;
            message.updated_at = updated_at.to_string();
            marked.push(message.clone());
            remaining = remaining.saturating_sub(1);
        }
    }
    marked
}

fn mark_runtime_queue_message_by_id(
    messages: &mut [RuntimeQueueMessageDto],
    message_id: &str,
    status: RuntimeQueueMessageStatus,
    updated_at: &str,
) -> Option<RuntimeQueueMessageDto> {
    let message = messages
        .iter_mut()
        .find(|message| message.id == message_id)?;
    message.status = status;
    message.updated_at = updated_at.to_string();
    Some(message.clone())
}

fn attach_runtime_queue_message_handle(
    messages: &mut [RuntimeQueueMessageDto],
    message_id: &str,
    handle: QueuedMessageHandle,
) {
    if let Some(message) = messages.iter_mut().find(|message| message.id == message_id) {
        message.handle = Some(handle);
    }
}

fn pending_runtime_queue_message_handle(
    state: &RuntimeQueueState,
    message_id: &str,
) -> Result<QueuedMessageHandle, AppError> {
    let Some(message) = state
        .messages
        .iter()
        .find(|message| message.id == message_id)
    else {
        return Err(AppError::recoverable(
            ErrorSource::Thread,
            "thread.queue_message.not_found",
            "The queued message is no longer available",
        ));
    };

    if message.status != RuntimeQueueMessageStatus::Pending {
        return Err(AppError::recoverable(
            ErrorSource::Thread,
            "thread.queue_message.not_pending",
            "The queued message has already been consumed, cleared, or cancelled",
        ));
    }

    message.handle.ok_or_else(|| {
        AppError::recoverable(
            ErrorSource::Thread,
            "thread.queue_message.handle_missing",
            "The queued message cannot be cancelled yet",
        )
    })
}

fn pending_follow_up_runtime_queue_message_handle(
    state: &RuntimeQueueState,
    message_id: &str,
) -> Result<QueuedMessageHandle, AppError> {
    let Some(message) = state
        .messages
        .iter()
        .find(|message| message.id == message_id)
    else {
        return Err(AppError::recoverable(
            ErrorSource::Thread,
            "thread.queue_message.not_found",
            "The queued message is no longer available",
        ));
    };

    if message.status != RuntimeQueueMessageStatus::Pending {
        return Err(AppError::recoverable(
            ErrorSource::Thread,
            "thread.queue_message.not_pending",
            "The queued message has already been consumed, cleared, or cancelled",
        ));
    }

    if message.kind != AgentQueueMessageKind::FollowUp {
        return Err(AppError::recoverable(
            ErrorSource::Thread,
            "thread.queue_message.promote_invalid_kind",
            "Only follow-up queue messages can be promoted to steering",
        ));
    }

    message.handle.ok_or_else(|| {
        AppError::recoverable(
            ErrorSource::Thread,
            "thread.queue_message.handle_missing",
            "The queued message cannot be promoted yet",
        )
    })
}

fn remove_pending_promoted_message_id(state: &mut RuntimeQueueState, message_id: &str) {
    state
        .pending_promoted_message_ids
        .retain(|pending_message_id| pending_message_id != message_id);
}

fn promote_runtime_queue_message_by_id(
    messages: &mut [RuntimeQueueMessageDto],
    message_id: &str,
    handle: QueuedMessageHandle,
    updated_at: &str,
) -> Option<RuntimeQueueMessageDto> {
    let message = messages.iter_mut().find(|message| {
        message.id == message_id && message.status == RuntimeQueueMessageStatus::Pending
    })?;
    message.kind = AgentQueueMessageKind::Steer;
    message.handle = Some(handle);
    message.updated_at = updated_at.to_string();
    Some(message.clone())
}

fn trim_runtime_queue_state(state: &mut RuntimeQueueState) {
    if state.messages.len() > RUNTIME_QUEUE_MAX_MESSAGES {
        let drop_count = state.messages.len() - RUNTIME_QUEUE_MAX_MESSAGES;
        state.messages.drain(0..drop_count);
    }
    if state.events.len() > RUNTIME_QUEUE_MAX_EVENTS {
        let drop_count = state.events.len() - RUNTIME_QUEUE_MAX_EVENTS;
        state.events.drain(0..drop_count);
    }
}

fn build_runtime_queue_snapshot(
    stats: &tiycore::agent::QueueStats,
    state: &RuntimeQueueState,
) -> RuntimeQueueSnapshotDto {
    RuntimeQueueSnapshotDto {
        steering_depth: stats.steering_depth,
        follow_up_depth: stats.follow_up_depth,
        is_deferring_steering: stats.is_deferring_steering,
        messages: state.messages.clone(),
        events: state.events.clone(),
    }
}

fn runtime_queue_message_display_content(message: &RuntimeQueueMessageDto) -> String {
    message
        .metadata
        .as_ref()
        .and_then(|metadata| metadata.pointer("/composer/displayText"))
        .and_then(serde_json::Value::as_str)
        .filter(|display_text| !display_text.trim().is_empty())
        .map(ToString::to_string)
        .unwrap_or_else(|| message.content.clone())
}

fn queue_snapshot_to_value(snapshot: RuntimeQueueSnapshotDto) -> serde_json::Value {
    serde_json::to_value(snapshot).unwrap_or_else(|_| serde_json::json!({}))
}

#[derive(Debug, Clone)]
struct RuntimeQueueEventUpdate {
    event: RuntimeQueueEventDto,
    consumed_messages: Vec<RuntimeQueueMessageDto>,
}

fn update_runtime_queue_state_for_event(
    state: &mut RuntimeQueueState,
    event: &QueueEvent,
) -> RuntimeQueueEventUpdate {
    let now = runtime_queue_timestamp();
    let mut consumed_messages = Vec::new();
    let dto = match event {
        QueueEvent::Enqueued {
            kind,
            count,
            queue_depth,
        } => RuntimeQueueEventDto {
            id: uuid::Uuid::now_v7().to_string(),
            kind: AgentQueueMessageKind::from_tiy_queue_kind(*kind),
            action: RuntimeQueueEventAction::Enqueued,
            count: *count,
            queue_depth: Some(*queue_depth),
            remaining: None,
            count_dropped: None,
            created_at: now,
        },
        QueueEvent::Consumed {
            kind,
            count,
            remaining,
        } => {
            let queue_kind = AgentQueueMessageKind::from_tiy_queue_kind(*kind);
            consumed_messages = mark_runtime_queue_messages(
                &mut state.messages,
                queue_kind,
                *count,
                RuntimeQueueMessageStatus::Consumed,
                &now,
            );
            RuntimeQueueEventDto {
                id: uuid::Uuid::now_v7().to_string(),
                kind: queue_kind,
                action: RuntimeQueueEventAction::Consumed,
                count: *count,
                queue_depth: None,
                remaining: Some(*remaining),
                count_dropped: None,
                created_at: now,
            }
        }
        QueueEvent::Cleared {
            kind,
            count_dropped,
        } => {
            let queue_kind = AgentQueueMessageKind::from_tiy_queue_kind(*kind);
            mark_runtime_queue_messages(
                &mut state.messages,
                queue_kind,
                *count_dropped,
                RuntimeQueueMessageStatus::Cleared,
                &now,
            );
            RuntimeQueueEventDto {
                id: uuid::Uuid::now_v7().to_string(),
                kind: queue_kind,
                action: RuntimeQueueEventAction::Cleared,
                count: *count_dropped,
                queue_depth: None,
                remaining: None,
                count_dropped: Some(*count_dropped),
                created_at: now,
            }
        }
        QueueEvent::Removed {
            kind,
            count,
            remaining,
        } => {
            let queue_kind = AgentQueueMessageKind::from_tiy_queue_kind(*kind);
            let mut marked = 0usize;
            while marked < *count {
                // Search from the back for a pending ID whose message matches
                // the current kind, so that IDs for other kinds aren't
                // accidentally discarded when kinds are cancelled concurrently.
                let position = state
                    .pending_removed_message_ids
                    .iter()
                    .enumerate()
                    .rev()
                    .find(|(_, id)| {
                        state
                            .messages
                            .iter()
                            .any(|m| m.id == **id && m.kind == queue_kind)
                    })
                    .map(|(pos, _)| pos);

                let Some(pos) = position else {
                    break;
                };
                let message_id = state.pending_removed_message_ids.remove(pos);
                // Look up the message by ID and kind.  If it was already
                // marked Cancelled by the eager cancel path, count it as
                // matched so the FIFO fallback doesn't cancel an unrelated
                // pending message.
                let matched = state
                    .messages
                    .iter_mut()
                    .find(|message| message.id == message_id && message.kind == queue_kind);
                match matched {
                    Some(message) if message.status == RuntimeQueueMessageStatus::Cancelled => {
                        // Eager cancel already transitioned this message;
                        // update the timestamp but don't change status.
                        message.updated_at = now.clone();
                        marked += 1;
                    }
                    Some(message) if message.status == RuntimeQueueMessageStatus::Pending => {
                        message.status = RuntimeQueueMessageStatus::Cancelled;
                        message.updated_at = now.clone();
                        marked += 1;
                    }
                    _ => {
                        // Unexpected status — skip without counting so the ID
                        // is simply discarded.
                    }
                }
            }
            let promoted_positions: Vec<usize> = state
                .pending_promoted_message_ids
                .iter()
                .enumerate()
                .rev()
                .filter_map(|(pos, id)| {
                    state
                        .messages
                        .iter()
                        .any(|message| {
                            message.id == *id
                                && message.kind == queue_kind
                                && message.status == RuntimeQueueMessageStatus::Pending
                        })
                        .then_some(pos)
                })
                .take(count.saturating_sub(marked))
                .collect();
            for pos in promoted_positions {
                if let Some(message_id) = state.pending_promoted_message_ids.get(pos) {
                    if let Some(message) = state.messages.iter_mut().find(|message| {
                        message.id == *message_id
                            && message.kind == queue_kind
                            && message.status == RuntimeQueueMessageStatus::Pending
                    }) {
                        message.updated_at = now.clone();
                        marked += 1;
                    }
                }
            }
            if marked < *count {
                let mut remaining = count.saturating_sub(marked);
                for message in state.messages.iter_mut() {
                    if remaining == 0 {
                        break;
                    }
                    if message.kind == queue_kind
                        && message.status == RuntimeQueueMessageStatus::Pending
                        && !state
                            .pending_promoted_message_ids
                            .iter()
                            .any(|pending_message_id| pending_message_id == &message.id)
                    {
                        message.status = RuntimeQueueMessageStatus::Cancelled;
                        message.updated_at = now.clone();
                        remaining = remaining.saturating_sub(1);
                    }
                }
            }
            RuntimeQueueEventDto {
                id: uuid::Uuid::now_v7().to_string(),
                kind: queue_kind,
                action: RuntimeQueueEventAction::Removed,
                count: *count,
                queue_depth: None,
                remaining: Some(*remaining),
                count_dropped: None,
                created_at: now,
            }
        }
        QueueEvent::Transferred {
            from,
            to,
            count,
            source_remaining,
            target_queue_depth,
        } => {
            let source_kind = AgentQueueMessageKind::from_tiy_queue_kind(*from);
            let target_kind = AgentQueueMessageKind::from_tiy_queue_kind(*to);
            let mut moved = 0usize;
            while moved < *count {
                let position = state
                    .pending_promoted_message_ids
                    .iter()
                    .enumerate()
                    .rev()
                    .find(|(_, id)| {
                        state.messages.iter().any(|message| {
                            message.id == **id
                                && (message.kind == source_kind || message.kind == target_kind)
                                && message.status == RuntimeQueueMessageStatus::Pending
                        })
                    })
                    .map(|(pos, _)| pos);

                let Some(pos) = position else {
                    break;
                };
                let message_id = state.pending_promoted_message_ids.remove(pos);
                if let Some(message) = state.messages.iter_mut().find(|message| {
                    message.id == message_id && message.status == RuntimeQueueMessageStatus::Pending
                }) {
                    message.kind = target_kind;
                    message.updated_at = now.clone();
                    moved += 1;
                }
            }
            if moved < *count {
                for message in state.messages.iter_mut() {
                    if moved >= *count {
                        break;
                    }
                    if message.kind == source_kind
                        && message.status == RuntimeQueueMessageStatus::Pending
                    {
                        message.kind = target_kind;
                        message.updated_at = now.clone();
                        moved += 1;
                    }
                }
            }
            RuntimeQueueEventDto {
                id: uuid::Uuid::now_v7().to_string(),
                kind: target_kind,
                action: RuntimeQueueEventAction::Transferred,
                count: *count,
                queue_depth: Some(*target_queue_depth),
                remaining: Some(*source_remaining),
                count_dropped: None,
                created_at: now,
            }
        }
    };
    state.events.push(dto.clone());
    trim_runtime_queue_state(state);
    RuntimeQueueEventUpdate {
        event: dto,
        consumed_messages,
    }
}

fn append_runtime_queue_message(
    state: &mut RuntimeQueueState,
    kind: AgentQueueMessageKind,
    content: String,
    metadata: Option<serde_json::Value>,
) -> String {
    let now = runtime_queue_timestamp();
    let id = uuid::Uuid::now_v7().to_string();
    state.messages.push(RuntimeQueueMessageDto {
        id: id.clone(),
        kind,
        content,
        metadata,
        status: RuntimeQueueMessageStatus::Pending,
        created_at: now.clone(),
        updated_at: now,
        handle: None,
    });
    trim_runtime_queue_state(state);
    id
}

use crate::core::agent_session_compression::{
    persist_compression_markers_to_pool, run_auto_compression,
};
use crate::core::agent_session_events::handle_agent_event;
pub(crate) use crate::core::agent_session_history::*;
pub(crate) use crate::core::agent_session_tools::*;
pub use crate::core::agent_session_types::*;

pub async fn build_session_spec(
    pool: &SqlitePool,
    run_id: &str,
    thread_id: &str,
    workspace_path: &str,
    run_mode: &str,
    model_plan_value: &serde_json::Value,
) -> Result<AgentSessionSpec, AppError> {
    let raw_plan: RuntimeModelPlan =
        serde_json::from_value(model_plan_value.clone()).unwrap_or_default();
    let resolved_plan = resolve_model_plan(pool, raw_plan.clone()).await?;
    let tool_profile_name = resolve_tool_profile_name(&raw_plan, run_mode);
    // Load all messages since the last context reset marker (or all messages
    // if no reset exists).  The new `list_since_last_reset` queries the DB
    // directly for the reset boundary, removing the former hard-coded 200-row
    // limit (`MESSAGE_HISTORY_LIMIT`) that could silently discard older
    // context_reset markers.
    let history_messages = message_repo::list_since_last_reset(pool, thread_id).await?;

    // Load tool calls from all runs referenced by the history messages so that
    // convert_history_messages can reconstruct assistant-tool-call / tool-result
    // pairs that are not stored in the messages table.
    let history_run_ids: Vec<String> = history_messages
        .iter()
        .filter_map(|m| m.run_id.clone())
        .collect::<std::collections::HashSet<_>>()
        .into_iter()
        .collect();
    let history_tool_calls =
        tool_call_repo::list_parent_visible_by_run_ids(pool, &history_run_ids).await?;
    // No longer read after the calibration seed was removed; the
    // `find_latest_with_prompt_usage_by_thread_excluding_run` call
    // above is kept as documentation that historical run usage is
    // intentionally not used by the new `Usage::context_size()`
    // trigger (the first LLM response provides the source of truth).
    let _latest_historical_run =
        run_repo::find_latest_with_prompt_usage_by_thread_excluding_run(pool, thread_id, run_id)
            .await?;

    let composed_prompt =
        build_system_prompt(pool, &raw_plan, workspace_path, run_mode, thread_id).await?;
    let system_prompt = composed_prompt.text.clone();
    let cache_arbiter = composed_prompt.cache_arbiter;
    let extension_tools = ExtensionsManager::new(pool.clone())
        .list_runtime_agent_tools(Some(workspace_path))
        .await?;

    // Load custom subagent tools for the active profile
    let custom_subagent_tools = {
        use crate::core::subagent::runtime_orchestration::custom_subagent_as_tool;
        use crate::persistence::repo::{custom_subagent_repo, settings_repo};

        let active_profile_id = settings_repo::get(pool, "active_profile_id")
            .await
            .ok()
            .flatten()
            .and_then(|s| serde_json::from_str::<String>(&s.value_json).ok())
            .unwrap_or_default();

        if active_profile_id.is_empty() {
            vec![]
        } else {
            custom_subagent_repo::list_for_profile(pool, &active_profile_id)
                .await
                .unwrap_or_default()
                .iter()
                .map(|r| custom_subagent_as_tool(r))
                .collect()
        }
    };
    let web_search_enabled = crate::core::web_search_settings::load_web_search_settings(pool)
        .await
        .map(|settings| settings.is_ready())
        .unwrap_or(false);

    let mut runtime_tools = runtime_tools_with_custom_subagents(
        runtime_tools_with_web_search(
            runtime_tools_for_profile_with_extensions(&tool_profile_name, extension_tools),
            &tool_profile_name,
            web_search_enabled,
        ),
        custom_subagent_tools,
    );

    // Inject the main-agent-only `agent_judge` acceptance tool on demand: only
    // when this thread has a goal that has not yet passed Judge acceptance
    // (acceptance = status Complete AND judge_passed). It is appended after the
    // custom/extension merge so that the built-in tool name always wins and
    // cannot be shadowed by a custom or extension tool.
    if let Ok(Some(goal)) =
        crate::persistence::repo::goal_repo::find_by_thread_id(pool, thread_id).await
    {
        let already_verified =
            goal.status == crate::model::goal::GoalStatus::Complete && goal.judge_passed;
        if !already_verified {
            let judge_tool = crate::core::subagent::RuntimeOrchestrationTool::Judge.as_agent_tool();
            if !runtime_tools.iter().any(|t| t.name == judge_tool.name) {
                runtime_tools.push(judge_tool);
            }
        }
    }

    // The previous initial-context calibration seed was removed when we
    // switched auto-compression to the unified `Usage::context_size()`
    // trigger (see `should_compress_via_context_size`). The trigger no
    // longer needs a per-history-run calibration — it compares the most
    // recent LLM usage against the budget directly.

    Ok(AgentSessionSpec {
        run_id: run_id.to_string(),
        thread_id: thread_id.to_string(),
        workspace_path: workspace_path.to_string(),
        run_mode: run_mode.to_string(),
        tool_profile_name: tool_profile_name.clone(),
        runtime_tools,
        system_prompt,
        history_messages,
        history_tool_calls,
        model_plan: resolved_plan,
        initial_prompt: None,
        cache_arbiter,
    })
}

pub struct AgentSession {
    pub(crate) spec: AgentSessionSpec,
    pub(crate) pool: SqlitePool,
    pub(crate) tool_gateway: Arc<ToolGateway>,
    pub(crate) helper_orchestrator: Arc<HelperAgentOrchestrator>,
    pub(crate) event_tx: mpsc::UnboundedSender<ThreadStreamEvent>,
    pub(crate) agent: Arc<Agent>,
    cancel_requested: Arc<AtomicBool>,
    pub(crate) checkpoint_requested: AtomicBool,
    pub(crate) abort_signal: tiycore::agent::AbortSignal,
    /// Most recent `tiycore::types::Usage` observed from the last LLM
    /// call, shared with the `set_transform_context` closure. The
    /// closure compares `usage.context_size()` against the configured
    /// `CompressionSettings::budget()` to decide whether the next turn
    /// needs auto-compression. Shared with the event handler so the
    /// trigger stays in lockstep with the stream of usage updates.
    last_observed_usage: Arc<StdMutex<Option<Usage>>>,
    runtime_queue_state: Arc<StdMutex<RuntimeQueueState>>,
    /// Shared goal runtime state for tool call recording across command invocations.
    pub(crate) goal_runtime: Arc<std::sync::Mutex<GoalRuntimeState>>,
}

impl AgentSession {
    pub(crate) fn new(
        pool: SqlitePool,
        tool_gateway: Arc<ToolGateway>,
        helper_orchestrator: Arc<HelperAgentOrchestrator>,
        event_tx: mpsc::UnboundedSender<ThreadStreamEvent>,
        spec: AgentSessionSpec,
        max_turns: usize,
        goal_runtime: Arc<std::sync::Mutex<GoalRuntimeState>>,
    ) -> Arc<Self> {
        Arc::new_cyclic(|weak_self| {
            let agent = Arc::new(Agent::with_model(spec.model_plan.primary.model.clone()));
            // Unified trigger: start with no observed usage; the
            // `set_transform_context` closure defers its first decision
            // until the first LLM response reports `context_size`.
            let last_observed_usage = Arc::new(StdMutex::new(None::<Usage>));
            let runtime_queue_state = Arc::new(StdMutex::new(RuntimeQueueState::default()));
            agent.set_max_turns(max_turns);
            agent.set_max_retries(Some(TIYCORE_REQUEST_MAX_RETRIES));
            configure_agent(
                &agent,
                &spec,
                weak_self.clone(),
                Arc::clone(&last_observed_usage),
                Arc::clone(&runtime_queue_state),
                event_tx.clone(),
            );

            Self {
                spec,
                pool,
                tool_gateway,
                helper_orchestrator,
                event_tx,
                agent,
                cancel_requested: Arc::new(AtomicBool::new(false)),
                checkpoint_requested: AtomicBool::new(false),
                abort_signal: tiycore::agent::AbortSignal::new(),
                last_observed_usage,
                runtime_queue_state,
                goal_runtime,
            }
        })
    }

    pub fn start(self: Arc<Self>) -> tokio::task::JoinHandle<()> {
        tokio::spawn(async move {
            self.run().await;
        })
    }

    pub async fn cancel(&self) {
        self.cancel_requested.store(true, Ordering::SeqCst);
        // 1. Cascade-cancel all in-flight tool_gateway calls (including subagent ones
        //    via child tokens) so they return Cancelled immediately.
        self.abort_signal.cancel();
        // 2. Abort subagent Agent instances so their LLM streaming / run loops stop.
        self.helper_orchestrator.cancel_run(&self.spec.run_id).await;
        // 3. Yield to let already-aborted subagents finish cleanup and emit
        //    SubagentFailed before the main agent terminates.
        tokio::task::yield_now().await;
        // 4. Finally abort the main agent.
        self.agent.abort();
    }

    pub fn enqueue_queue_message(
        &self,
        kind: AgentQueueMessageKind,
        message: String,
        metadata: Option<serde_json::Value>,
    ) -> Result<RuntimeQueueSnapshotDto, AppError> {
        let text = message.trim();
        if text.is_empty() {
            return Err(AppError::validation(
                ErrorSource::Thread,
                "Queue message cannot be empty",
            ));
        }

        let message_id = {
            let mut state = self
                .runtime_queue_state
                .lock()
                .expect("runtime queue state poisoned");
            append_runtime_queue_message(&mut state, kind, text.to_string(), metadata)
        };

        let handle = match kind {
            AgentQueueMessageKind::Steer => self.agent.steer(AgentMessage::from(text)),
            AgentQueueMessageKind::FollowUp => self.agent.follow_up(AgentMessage::from(text)),
        };

        // Fetch queue stats BEFORE re-acquiring runtime_queue_state to avoid
        // an ABBA deadlock: the on_queue_event callback runs inside the
        // Agent's internal lock and acquires runtime_queue_state, while this
        // path would otherwise hold runtime_queue_state and then acquire the
        // Agent lock via queue_stats().
        let stats = self.agent.queue_stats();

        let mut state = self
            .runtime_queue_state
            .lock()
            .expect("runtime queue state poisoned");
        attach_runtime_queue_message_handle(&mut state.messages, &message_id, handle);
        Ok(build_runtime_queue_snapshot(&stats, &state))
    }

    pub fn runtime_queue_snapshot(&self) -> RuntimeQueueSnapshotDto {
        let stats = self.agent.queue_stats();
        let state = self
            .runtime_queue_state
            .lock()
            .expect("runtime queue state poisoned");
        build_runtime_queue_snapshot(&stats, &state)
    }

    pub fn cancel_runtime_queue_message(
        &self,
        message_id: &str,
    ) -> Result<RuntimeQueueSnapshotDto, AppError> {
        let handle = {
            let mut state = self
                .runtime_queue_state
                .lock()
                .expect("runtime queue state poisoned");
            let handle = pending_runtime_queue_message_handle(&state, message_id)?;
            state
                .pending_removed_message_ids
                .push(message_id.to_string());
            handle
        };

        if self.agent.cancel_queued_message(handle).is_none() {
            let mut state = self
                .runtime_queue_state
                .lock()
                .expect("runtime queue state poisoned");
            state
                .pending_removed_message_ids
                .retain(|pending_message_id| pending_message_id != message_id);
            return Err(AppError::recoverable(
                ErrorSource::Thread,
                "thread.queue_message.cancel_failed",
                "The queued message was already consumed or removed",
            ));
        }

        let now = runtime_queue_timestamp();
        // Fetch queue stats BEFORE re-acquiring runtime_queue_state to avoid
        // an ABBA deadlock (see enqueue_queue_message for details).
        let stats = self.agent.queue_stats();

        let mut state = self
            .runtime_queue_state
            .lock()
            .expect("runtime queue state poisoned");
        mark_runtime_queue_message_by_id(
            &mut state.messages,
            message_id,
            RuntimeQueueMessageStatus::Cancelled,
            &now,
        );
        Ok(build_runtime_queue_snapshot(&stats, &state))
    }

    pub fn promote_runtime_queue_message(
        &self,
        message_id: &str,
    ) -> Result<RuntimeQueueSnapshotDto, AppError> {
        let handle = {
            let mut state = self
                .runtime_queue_state
                .lock()
                .expect("runtime queue state poisoned");
            let handle = pending_follow_up_runtime_queue_message_handle(&state, message_id)?;
            state
                .pending_promoted_message_ids
                .push(message_id.to_string());
            handle
        };

        let new_handle = match self.agent.try_promote_follow_up_to_steering(handle.id) {
            Ok(new_handle) => new_handle,
            Err(PromoteError::NotFound) => {
                let mut state = self
                    .runtime_queue_state
                    .lock()
                    .expect("runtime queue state poisoned");
                remove_pending_promoted_message_id(&mut state, message_id);
                return Err(AppError::recoverable(
                    ErrorSource::Thread,
                    "thread.queue_message.promote_not_found",
                    "The queued message was already consumed or removed",
                ));
            }
            Err(PromoteError::QueueFull(error)) => {
                let mut state = self
                    .runtime_queue_state
                    .lock()
                    .expect("runtime queue state poisoned");
                remove_pending_promoted_message_id(&mut state, message_id);
                return Err(AppError::recoverable(
                    ErrorSource::Thread,
                    "thread.queue_message.promote_queue_full",
                    format!(
                        "The steering queue is full (depth={}, max={})",
                        error.current_depth, error.max_depth
                    ),
                ));
            }
        };

        let now = runtime_queue_timestamp();
        // Fetch queue stats BEFORE re-acquiring runtime_queue_state to avoid
        // an ABBA deadlock (see enqueue_queue_message for details).
        let stats = self.agent.queue_stats();

        let mut state = self
            .runtime_queue_state
            .lock()
            .expect("runtime queue state poisoned");
        remove_pending_promoted_message_id(&mut state, message_id);
        promote_runtime_queue_message_by_id(&mut state.messages, message_id, new_handle, &now);
        Ok(build_runtime_queue_snapshot(&stats, &state))
    }

    pub fn clear_runtime_queue(
        &self,
        kind: Option<AgentQueueMessageKind>,
    ) -> RuntimeQueueSnapshotDto {
        match kind {
            Some(AgentQueueMessageKind::Steer) => self.agent.clear_steering_queue(),
            Some(AgentQueueMessageKind::FollowUp) => self.agent.clear_follow_up_queue(),
            None => self.agent.clear_all_queues(),
        }

        let stats = self.agent.queue_stats();
        let state = self
            .runtime_queue_state
            .lock()
            .expect("runtime queue state poisoned");
        build_runtime_queue_snapshot(&stats, &state)
    }

    async fn run(self: Arc<Self>) {
        let current_message_id = Arc::new(StdMutex::new(None::<String>));
        let last_completed_message_id = Arc::new(StdMutex::new(None::<String>));
        let current_reasoning_message_id = Arc::new(StdMutex::new(None::<String>));
        let last_usage = Arc::new(StdMutex::new(None::<Usage>));
        let reasoning_buffer = Arc::new(StdMutex::new(String::new()));
        let run_id = self.spec.run_id.clone();
        let event_tx = self.event_tx.clone();
        let context_window = self
            .spec
            .model_plan
            .primary
            .model
            .context_window
            .to_string();
        let model_display_name = self.spec.model_plan.primary.model_name.clone();

        let message_id_ref = Arc::clone(&current_message_id);
        let last_completed_message_id_ref = Arc::clone(&last_completed_message_id);
        let reasoning_message_id_ref = Arc::clone(&current_reasoning_message_id);
        let last_usage_ref = Arc::clone(&last_usage);
        let last_observed_usage_ref = Arc::clone(&self.last_observed_usage);
        let reasoning_ref = Arc::clone(&reasoning_buffer);
        let current_turn_index: Arc<StdMutex<Option<usize>>> = Arc::new(StdMutex::new(None));
        let turn_index_ref = Arc::clone(&current_turn_index);
        let last_text_delta = Arc::new(StdMutex::new(None::<String>));
        let last_text_delta_ref = Arc::clone(&last_text_delta);
        let unsubscribe = self.agent.subscribe(move |event| {
            handle_agent_event(
                &run_id,
                &event_tx,
                &message_id_ref,
                &last_completed_message_id_ref,
                &reasoning_message_id_ref,
                &last_usage_ref,
                &last_observed_usage_ref,
                &reasoning_ref,
                &turn_index_ref,
                &last_text_delta_ref,
                &context_window,
                &model_display_name,
                event,
            );
        });

        let _ = self.event_tx.send(ThreadStreamEvent::RunStarted {
            run_id: self.spec.run_id.clone(),
            run_mode: self.spec.run_mode.clone(),
            started_at_ms: chrono::Utc::now().timestamp_millis(),
        });

        let result = if let Some(prompt) = self.spec.initial_prompt.clone() {
            // Phase 3: prepend RuntimeMessage (current_date) before the user's
            // turn so the LLM sees the wall-clock date without breaking the
            // system prompt's prefix cache. § 3.7 RuntimeMessagePlacement::BeforeLatestUser.
            let prompt_with_runtime = inject_runtime_context(&prompt).await;
            self.agent.prompt(prompt_with_runtime).await
        } else {
            self.agent.continue_().await
        };
        unsubscribe();

        match result {
            Ok(_) => {
                if self.cancel_requested.load(Ordering::SeqCst) {
                    let _ = self.event_tx.send(ThreadStreamEvent::RunCancelled {
                        run_id: self.spec.run_id.clone(),
                    });
                } else if self.checkpoint_requested.load(Ordering::SeqCst) {
                    let _ = self.event_tx.send(ThreadStreamEvent::RunCheckpointed {
                        run_id: self.spec.run_id.clone(),
                    });
                } else if let Some(error) = plan_mode_missing_checkpoint_error(
                    &self.spec.run_mode,
                    self.checkpoint_requested.load(Ordering::SeqCst),
                ) {
                    let _ = self.event_tx.send(ThreadStreamEvent::RunFailed {
                        run_id: self.spec.run_id.clone(),
                        error: error.to_string(),
                    });
                } else {
                    let _ = self.event_tx.send(ThreadStreamEvent::RunCompleted {
                        run_id: self.spec.run_id.clone(),
                    });
                }
            }
            Err(error) => {
                if self.cancel_requested.load(Ordering::SeqCst) {
                    let _ = self.event_tx.send(ThreadStreamEvent::RunCancelled {
                        run_id: self.spec.run_id.clone(),
                    });
                } else if self.checkpoint_requested.load(Ordering::SeqCst) {
                    let _ = self.event_tx.send(ThreadStreamEvent::RunCheckpointed {
                        run_id: self.spec.run_id.clone(),
                    });
                } else if let AgentError::MaxTurnsReached(max_turns) = &error {
                    let _ = self.event_tx.send(ThreadStreamEvent::RunLimitReached {
                        run_id: self.spec.run_id.clone(),
                        error: error.to_string(),
                        max_turns: *max_turns,
                    });
                } else {
                    let _ = self.event_tx.send(ThreadStreamEvent::RunFailed {
                        run_id: self.spec.run_id.clone(),
                        error: error.to_string(),
                    });
                }
            }
        }
    }

    /// Persist context_summary and context_reset markers to the DB after
    /// auto-compression.
    ///
    /// `boundary_message_id` is the id of the **first DB-backed message we
    /// intend to keep** in the next run's loaded history. It is stored inside
    /// the reset marker's metadata as `boundaryMessageId` so that
    /// `list_since_last_reset` can use it as the lower bound — this is the
    /// only way to keep the DB view aligned with the in-memory `cut_point`,
    /// because UUID v7 timing means the reset marker itself is written *after*
    /// some messages that the cut_point means to preserve.
    ///
    /// The reset marker is written first, then the summary, so both rows have
    /// `id > boundary_message_id` and the next `list_since_last_reset(…)` call
    /// will return `[boundary_message, …, reset_marker, summary_marker]` in
    /// chronological order.
    pub async fn persist_compression_markers(
        &self,
        thread_id: &str,
        summary: &str,
        source: &str,
        boundary_message_id: Option<&str>,
    ) -> Result<(), crate::model::errors::AppError> {
        persist_compression_markers_to_pool(
            &self.pool,
            thread_id,
            summary,
            source,
            boundary_message_id,
        )
        .await
    }

    pub(super) async fn next_plan_revision(&self) -> Result<u32, AppError> {
        let messages =
            message_repo::list_recent(&self.pool, &self.spec.thread_id, None, 256).await?;
        let next = messages
            .iter()
            .filter(|message| message.message_type == "plan")
            .filter_map(|message| {
                message
                    .metadata_json
                    .as_deref()
                    .and_then(|raw| serde_json::from_str::<serde_json::Value>(raw).ok())
                    .and_then(|value| {
                        value
                            .get("planRevision")
                            .and_then(serde_json::Value::as_u64)
                            .and_then(|value| u32::try_from(value).ok())
                    })
            })
            .max()
            .unwrap_or(0);
        Ok(next.saturating_add(1))
    }
}

fn configure_agent(
    agent: &Arc<Agent>,
    spec: &AgentSessionSpec,
    weak_self: Weak<AgentSession>,
    last_observed_usage: Arc<StdMutex<Option<Usage>>>,
    runtime_queue_state: Arc<StdMutex<RuntimeQueueState>>,
    event_tx: mpsc::UnboundedSender<ThreadStreamEvent>,
) {
    agent.set_system_prompt(spec.system_prompt.clone());
    agent.replace_messages(convert_history_messages(
        &spec.history_messages,
        &spec.history_tool_calls,
        &spec.model_plan.primary.model,
    ));
    agent.set_tools(spec.runtime_tools.clone());
    agent.set_tool_execution(ToolExecutionMode::Sequential);
    agent.set_thinking_level(spec.model_plan.thinking_level);
    agent.set_transport(spec.model_plan.transport);
    agent.set_security_config(main_agent_security_config());

    // Context compression: when the most recent LLM call reported a
    // unified context size above the input budget, summarise the older
    // messages with the primary model, persist markers to DB, and keep
    // only recent messages. Falls back to pure truncation if the LLM
    // call fails.
    //
    // The trigger is `Usage::context_size()` from tiycore 0.2.10-rc.2 —
    // the cross-protocol unified "context occupancy" value
    // (input + output + cache_read + cache_write). The previous heuristic
    // (`estimate_total_tokens(messages) + system_prompt_estimated_tokens`
    // scaled by an observed `ContextTokenCalibration` ratio) was removed
    // because it could disagree with the actual provider usage and let an
    // over-budget call slip through, and because the heuristic + CJK
    // weight drift made the trigger point unpredictable.
    //
    // Two correctness hazards addressed in [`run_auto_compression`]:
    //
    // 1. UUID v7 timing. The reset marker we write to DB uses `now_v7()`,
    //    but `cut_point` in-memory points at a slice that includes
    //    messages the current run persisted EARLIER than this call. A
    //    naive `list_since_last_reset WHERE id >= reset_id` would exclude
    //    those earlier messages and effectively "lose" the current user
    //    prompt on the next reload. We therefore resolve a DB-backed
    //    boundary id conservatively covering all recent messages and
    //    attach it to the reset marker's metadata.
    //
    // 2. Summary-of-summary decay. If a previous auto-compression already
    //    injected a `<context_summary>` as the head of `messages`, naively
    //    re-summarising `old_messages` would re-summarise an already-
    //    summarised prefix, losing detail each pass. Instead we detect
    //    the prior summary, treat it as a pinned prefix, and ask the
    //    model to **merge** the prior summary with the delta of messages
    //    since then.
    let compression_settings = crate::core::context_compression::CompressionSettings::new(
        spec.model_plan.primary.model.context_window,
    );
    let primary_model_role = spec.model_plan.primary.clone();
    let compression_weak_self = weak_self.clone();
    let compression_thread_id = spec.thread_id.clone();
    let compression_run_id = spec.run_id.clone();
    let compression_response_language = spec.model_plan.raw.response_language.clone();
    let last_observed_usage_for_trigger = Arc::clone(&last_observed_usage);
    agent.set_transform_context(move |messages| {
        // Cheap pass-through check first: only clone the heavy captured
        // state (ResolvedModelRole, String ids, Weak) when compression
        // will actually run. For long sessions with many turns this
        // avoids per-turn heap allocations when the thread is still
        // well under budget.
        let settings = compression_settings.clone();
        let snapshot: Option<Usage> = last_observed_usage_for_trigger
            .lock()
            .ok()
            .and_then(|guard| *guard);
        // `should_compress_via_context_size` returns `false` for
        // `None` (no observed usage yet) — we defer the first decision
        // until the first LLM response reports its `context_size`,
        // matching the design note in `context_compression`.
        let needs_compression = crate::core::context_compression::should_compress_via_context_size(
            snapshot.as_ref(),
            &settings,
        ) && !messages.is_empty();

        let model_role = if needs_compression {
            Some(primary_model_role.clone())
        } else {
            None
        };
        let weak = if needs_compression {
            Some(compression_weak_self.clone())
        } else {
            None
        };
        let thread_id = if needs_compression {
            Some(compression_thread_id.clone())
        } else {
            None
        };
        let run_id = if needs_compression {
            Some(compression_run_id.clone())
        } else {
            None
        };
        let response_language = if needs_compression {
            Some(compression_response_language.clone())
        } else {
            None
        };

        async move {
            if !needs_compression {
                return messages;
            }
            // Unwraps are sound: all `Some(_)` are populated together
            // under `needs_compression`, so either all six are `Some`
            // (compression path) or we returned above.
            run_auto_compression(
                messages,
                settings,
                model_role.expect("model_role populated when compressing"),
                weak.expect("weak populated when compressing"),
                thread_id.expect("thread_id populated when compressing"),
                run_id.expect("run_id populated when compressing"),
                response_language.expect("response_language populated when compressing"),
            )
            .await
        }
    });

    if let Some(api_key) = spec.model_plan.primary.api_key.clone() {
        agent.set_api_key(api_key);
    }

    // Set session ID for prompt caching (used as prompt_cache_key in OpenAI Responses API).
    // Using thread_id ensures the same conversation thread shares a cache across runs.
    agent.set_session_id(&spec.thread_id);

    // Consume at most the oldest follow-up message per turn. Steering keeps
    // tiycore's default mode because it is configured independently.
    agent.set_follow_up_mode(QueueMode::OneAtATime);

    {
        let run_id = spec.run_id.clone();
        let event_tx = event_tx.clone();
        let agent_ref = Arc::downgrade(agent);
        let runtime_queue_state = Arc::clone(&runtime_queue_state);
        agent.set_on_queue_event(move |event| {
            let Some(agent_ref) = agent_ref.upgrade() else {
                return;
            };
            // Fetch queue stats BEFORE acquiring runtime_queue_state to
            // avoid an ABBA deadlock: this callback runs inside the Agent's
            // internal lock, and client methods acquire runtime_queue_state
            // before calling agent.queue_stats().  By reading stats first
            // we never hold both locks simultaneously in conflicting order.
            let stats = agent_ref.queue_stats();
            let (snapshot, consumed_messages) = {
                let mut state = runtime_queue_state
                    .lock()
                    .expect("runtime queue state poisoned");
                let update = update_runtime_queue_state_for_event(&mut state, &event);
                let _event = update.event;
                (
                    build_runtime_queue_snapshot(&stats, &state),
                    update.consumed_messages,
                )
            };
            let _ = event_tx.send(ThreadStreamEvent::QueueUpdated {
                run_id: run_id.clone(),
                queue: queue_snapshot_to_value(snapshot),
            });
            for message in consumed_messages {
                let content = runtime_queue_message_display_content(&message);
                let metadata = message.metadata.clone();
                let _ = event_tx.send(ThreadStreamEvent::UserMessageRecorded {
                    run_id: run_id.clone(),
                    message_id: uuid::Uuid::now_v7().to_string(),
                    content,
                    created_at: message.updated_at,
                    metadata,
                });
            }
        });
    }

    // Inject default TiyCode identification headers for all LLM API requests.
    agent.set_custom_headers(crate::core::tiycode_default_headers());

    // Merge per-model provider_options into every outgoing LLM request payload.
    {
        let provider_options = spec.model_plan.primary.provider_options.clone();
        if provider_options.is_some() {
            agent.set_on_payload(move |payload, _model| {
                let provider_options = provider_options.clone();
                Box::pin(async move {
                    let mut p = payload;
                    if let Some(ref opts) = provider_options {
                        p = merge_payload(p, opts);
                    }
                    Some(p)
                })
            });
        }
    }

    agent.set_tool_executor(move |tool_name, tool_call_id, tool_input, _update_cb| {
        let weak_self = weak_self.clone();
        let tool_name = tool_name.to_string();
        let tool_call_id = tool_call_id.to_string();
        let tool_input = tool_input.clone();

        async move {
            match weak_self.upgrade() {
                Some(session) => {
                    session
                        .execute_tool_call(&tool_name, &tool_call_id, &tool_input)
                        .await
                }
                None => agent_error_result("agent session unavailable"),
            }
        }
    });
}

async fn resolve_model_plan(
    pool: &SqlitePool,
    raw_plan: RuntimeModelPlan,
) -> Result<ResolvedRuntimeModelPlan, AppError> {
    let primary = raw_plan.primary.clone().ok_or_else(|| {
        AppError::recoverable(
            ErrorSource::Settings,
            "settings.model_plan.primary_missing",
            "Run model plan is missing the primary model",
        )
    })?;

    let mut primary = resolve_runtime_model_role(pool, primary).await?;
    let mut auxiliary = match raw_plan.auxiliary.clone() {
        Some(role) => Some(resolve_runtime_model_role(pool, role).await?),
        None => None,
    };
    let mut lightweight = match raw_plan.lightweight.clone() {
        Some(role) => Some(resolve_runtime_model_role(pool, role).await?),
        None => None,
    };

    let thinking_level = raw_plan
        .thinking_level
        .as_deref()
        .map(ThinkingLevel::from)
        .unwrap_or(ThinkingLevel::Off);

    apply_thinking_level_to_model_role(&mut primary, thinking_level);
    if let Some(role) = auxiliary.as_mut() {
        apply_thinking_level_to_model_role(role, thinking_level);
    }
    if let Some(role) = lightweight.as_mut() {
        apply_thinking_level_to_model_role(role, thinking_level);
    }

    Ok(ResolvedRuntimeModelPlan {
        thinking_level,
        transport: parse_transport(raw_plan.transport.as_deref()),
        raw: raw_plan,
        primary,
        auxiliary,
        lightweight,
    })
}

fn apply_thinking_level_to_model_role(role: &mut ResolvedModelRole, thinking_level: ThinkingLevel) {
    role.model.reasoning = thinking_level != ThinkingLevel::Off && role.model.reasoning;
}

pub async fn resolve_runtime_model_role(
    pool: &SqlitePool,
    role: RuntimeModelRole,
) -> Result<ResolvedModelRole, AppError> {
    let provider = provider_repo::find_by_id(pool, &role.provider_id)
        .await?
        .ok_or_else(|| AppError::not_found(ErrorSource::Settings, "provider"))?;

    let provider_name = role
        .provider_name
        .clone()
        .unwrap_or_else(|| provider.display_name.clone());
    let model_name = role
        .model_display_name
        .clone()
        .unwrap_or_else(|| role.model.clone());
    let context_window = parse_positive_u32(role.context_window.as_deref(), DEFAULT_CONTEXT_WINDOW);
    let max_output_tokens =
        parse_positive_u32(role.max_output_tokens.as_deref(), DEFAULT_MAX_OUTPUT_TOKENS);

    let mut builder = Model::builder()
        .id(&role.model)
        .name(&model_name)
        .provider(Provider::from(role.provider_type.clone()))
        .base_url(&role.base_url)
        .reasoning(role.supports_reasoning.unwrap_or(false))
        .context_window(context_window)
        .max_tokens(max_output_tokens)
        .input({
            let mut input = vec![InputType::Text];
            if role.supports_image_input.unwrap_or(false) {
                input.push(InputType::Image);
            }
            input
        })
        .cost(Cost::default());

    {
        let mut headers = crate::core::tiycode_default_headers();
        if let Some(user_headers) = role.custom_headers.clone() {
            headers.extend(user_headers);
        }
        builder = builder.headers(headers);
    }

    let maybe_compat = default_openai_compatible_compat(&role.provider_type);

    // reasoning_content_constrained must propagate to all provider types,
    // not only "openai-compatible".  Some providers (e.g. DeepSeek V4 Pro
    // accessed via ZenMux) require every assistant message to carry
    // reasoning_content when thinking is enabled, and the API rejects
    // requests that omit it with HTTP 400.
    if let Some(true) = role.reasoning_content_constrained {
        let mut compat = maybe_compat.unwrap_or_default();
        compat.thinking.content_constrained = true;
        builder = builder.compat(compat);
    } else if let Some(compat) = maybe_compat {
        builder = builder.compat(compat);
    }

    let model = builder.build().map_err(|error| {
        AppError::internal(
            ErrorSource::Settings,
            format!("failed to build runtime model: {error}"),
        )
    })?;

    Ok(ResolvedModelRole {
        provider_id: role.provider_id,
        model_record_id: role.model_record_id,
        model_id: role.model_id,
        model_name,
        provider_type: role.provider_type,
        provider_name,
        api_key: provider.api_key_encrypted,
        provider_options: normalize_provider_options(role.provider_options),
        model,
    })
}

/// Build a runtime-context block (current date / timestamp) and prepend it to
/// the user prompt. Implements § 3.7 RuntimeMessagePlacement::BeforeLatestUser
/// at the message-content level — keeping the system prompt prefix-cache stable.
///
/// **Implicit dedup / PinOutsideWindow**:
///
/// The runtime block is **never persisted to the messages table** —
/// `agent_run_manager.rs::start_run` writes `display_prompt` (or the raw user
/// prompt), not the wrapped string we hand to `agent.prompt(...)`. Consequences:
///
/// 1. Each turn starts with a clean user prompt and is wrapped fresh — no need
///    for an explicit `dedup_id` lookup; the previous turn's runtime context is
///    not in the DB to be deduped.
/// 2. The compaction summary input (`build_compact_summary_messages`) reads
///    from `messages` and therefore sees no `<runtime_context>` blocks —
///    equivalent to `CompactionPolicy::PinOutsideWindow` at the storage layer
///    without an extra column.
/// 3. The wall-clock date enters the LLM context only via this prepend; if
///    a feature later needs server-authoritative time across the full message
///    history, a `compaction_pinned` column on `messages` would be required.
pub(crate) async fn inject_runtime_context(user_prompt: &str) -> String {
    use crate::core::prompt::{CurrentDateInjector, RuntimeMessageInjector, SystemClock};
    use std::sync::Arc;

    // The CurrentDateInjector source is fixed; only Surface gates apply.
    // We construct a minimal BuildCx to satisfy the trait signature.
    let injector = CurrentDateInjector::new(Arc::new(SystemClock));
    // Build a dummy BuildCx — CurrentDateInjector doesn't read it.
    let placeholder_pool = match sqlx::SqlitePool::connect_lazy("sqlite::memory:") {
        Ok(p) => p,
        Err(_) => return user_prompt.to_string(),
    };
    let cx = crate::core::prompt::BuildCx {
        pool: &placeholder_pool,
        workspace_path: "",
        thread_id: None,
        run_id: None,
        raw_plan: None,
        run_mode: crate::core::prompt::RunMode::Default,
        helper_profile: None,
        custom_subagent_slug: None,
        response_language: None,
        target_model: crate::core::prompt::ModelTarget::AnthropicClaude {
            context_window: 200_000,
            supports_cache_control: false,
        },
        clock: Arc::new(SystemClock),
        signals: Arc::new(crate::core::prompt::SignalCache::new()),
        renderer: Arc::new(crate::core::prompt::MarkdownRenderer),
    };

    match injector.build_message(&cx).await {
        Some(msg) => format!("{}\n\n{}", msg.text, user_prompt),
        None => user_prompt.to_string(),
    }
}

async fn build_system_prompt(
    pool: &SqlitePool,
    raw_plan: &RuntimeModelPlan,
    workspace_path: &str,
    run_mode: &str,
    thread_id: &str,
) -> Result<crate::core::prompt::ComposedPrompt, AppError> {
    use crate::core::prompt::{
        BuildCx, Composer, DefaultCacheMarkerArbiter, MarkdownRenderer, ModelTarget, NoopRedactor,
        PromptBudget, PromptSurface, RunMode, SourceExecPolicy, SystemClock,
    };
    use std::sync::Arc;

    let rm = RunMode::from_str(run_mode);
    let registry = Arc::new(prompt::registry::default_registry());
    let arbiter = Arc::new(DefaultCacheMarkerArbiter::new(4));
    let composer = Composer::new(
        registry,
        SourceExecPolicy::default(),
        Arc::new(NoopRedactor),
    )
    .with_cache_arbiter(arbiter);

    let cx = BuildCx {
        pool,
        workspace_path,
        thread_id: Some(thread_id),
        run_id: None,
        raw_plan: Some(raw_plan),
        run_mode: rm,
        helper_profile: None,
        custom_subagent_slug: None,
        response_language: None,
        // Cache markers enabled: Composer places Ephemeral markers at StablePrefix
        // and SessionStable layer boundaries for Anthropic prompt-prefix caching.
        target_model: ModelTarget::AnthropicClaude {
            context_window: 200_000,
            supports_cache_control: true,
        },
        clock: Arc::new(SystemClock),
        signals: Arc::new(crate::core::prompt::SignalCache::new()),
        renderer: Arc::new(MarkdownRenderer),
    };

    let surface = PromptSurface::MainAgent { run_mode: rm };
    let budget = PromptBudget::for_model(&cx.target_model, &surface);
    let composed = composer.build(&surface, &cx, &budget).await?;
    Ok(composed)
}

/// Security config for the **main** agent.  Uses a very large tool timeout so
/// that user-interactive tools (clarify, approval) are never killed by the
/// outer tiycore `tokio::select!` timeout.
fn main_agent_security_config() -> tiycore::types::SecurityConfig {
    let mut security = tiycore::types::SecurityConfig::default();
    security.agent.tool_execution_timeout_secs = MAIN_AGENT_TOOL_TIMEOUT_SECS;
    security.url = crate::core::tiycode_url_policy();
    security
}

/// Security config for **sub-agents** (helpers).  Keeps the tighter 600 s
/// timeout because sub-agents never surface user-interactive tools.
pub(crate) fn runtime_security_config() -> tiycore::types::SecurityConfig {
    let mut security = tiycore::types::SecurityConfig::default();
    security.agent.tool_execution_timeout_secs = SUBAGENT_TOOL_TIMEOUT_SECS;
    security.url = crate::core::tiycode_url_policy();
    security
}

pub(crate) fn standard_tool_timeout() -> std::time::Duration {
    std::time::Duration::from_secs(STANDARD_TOOL_TIMEOUT_SECS)
}

pub(crate) fn merge_payload(
    mut base: serde_json::Value,
    patch: &serde_json::Value,
) -> serde_json::Value {
    merge_json_value(&mut base, patch);
    base
}

fn merge_json_value(base: &mut serde_json::Value, patch: &serde_json::Value) {
    match (base, patch) {
        (serde_json::Value::Object(base_map), serde_json::Value::Object(patch_map)) => {
            for (key, patch_value) in patch_map {
                if let Some(base_value) = base_map.get_mut(key) {
                    merge_json_value(base_value, patch_value);
                } else {
                    base_map.insert(key.clone(), patch_value.clone());
                }
            }
        }
        (base_value, patch_value) => {
            *base_value = patch_value.clone();
        }
    }
}

#[cfg(test)]
#[path = "agent_session_tests.rs"]
mod agent_session_tests;
