//! Per-session actor. Each session gets its own actor with isolated state.
//!
//! Prompt execution uses `ctx.spawn()` so the actor stays responsive for
//! `Cancel`, `SetMode`, and other messages between turns.

use crate::acp::client_bridge::ClientBridgeSender;
use crate::agent::agent_config::AgentConfig;
use crate::agent::core::{AgentMode, SessionRuntime, ToolConfig};
use crate::agent::execution::CycleOutcome;
use crate::agent::execution_context::ExecutionContext;
use crate::agent::messages::*;
use crate::agent::undo::{RedoResult, UndoError, UndoResult};
use crate::agent::utils::{format_prompt_user_text_only, render_prompt_for_display};
use crate::elicitation::ElicitationAction;
use crate::error::AgentError;
use crate::events::{AgentEventKind, SessionLimits};
use crate::hooks::HookNotice;
use crate::model::{AgentMessage, MessagePart};
use crate::session::runtime::RuntimeContext;
use crate::session::store::LLMConfig;
use agent_client_protocol::schema::{
    ExtResponse, PromptResponse, SetSessionModelResponse, StopReason,
};
use kameo::Actor;
use kameo::message::{Context, Message};
use kameo::reply::DelegatedReply;
use log::{debug, info, warn};
use querymt::chat::{ChatRole, ReasoningEffort};
use std::collections::{HashMap, HashSet};
use std::sync::Arc;
use tokio::sync::{OwnedSemaphorePermit, oneshot};
use tokio_util::sync::CancellationToken;
use tracing::{Instrument, info_span, instrument};
use uuid::Uuid;

fn parse_transport_model_id(model_id: &str, fallback_provider: &str) -> (String, String) {
    if let Some(slash_pos) = model_id.find('/') {
        (
            model_id[..slash_pos].to_string(),
            model_id[slash_pos + 1..].to_string(),
        )
    } else {
        // Backward compatibility for legacy clients that still send a bare model id.
        (fallback_provider.to_string(), model_id.to_string())
    }
}

const ATTACHMENTS_PLACEHOLDER: &str = "(attachments included)";
// TODO: use it once figured out intent updates
#[allow(dead_code)]
const INTENT_CONFIRMATION_TIMEOUT_SECS: u64 = 45;

fn emit_hook_notices(config: &AgentConfig, session_id: &str, notices: Vec<HookNotice>) {
    for notice in notices {
        config.emit_event(
            session_id,
            AgentEventKind::HookNotice {
                event_name: notice.event_name,
                message: notice.message,
                is_error: notice.is_error,
            },
        );
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum IntentUpdateDecision {
    NoChange,
    Update,
    AskUser,
}

fn summarize_intent_candidate(input: &str) -> Option<String> {
    let trimmed = input.trim();
    if trimmed.is_empty() || trimmed == ATTACHMENTS_PLACEHOLDER {
        return None;
    }

    let single_line = trimmed.split_whitespace().collect::<Vec<_>>().join(" ");
    if single_line.is_empty() {
        return None;
    }

    Some(querymt_utils::str_utils::truncate_with_ellipsis(
        &single_line,
        240,
    ))
}

fn normalize_intent_text(text: &str) -> String {
    text.trim()
        .to_lowercase()
        .split_whitespace()
        .collect::<Vec<_>>()
        .join(" ")
}

fn decide_intent_update(current_intent: &str, candidate: &str) -> IntentUpdateDecision {
    let candidate_trimmed = candidate.trim();
    if candidate_trimmed.is_empty() {
        return IntentUpdateDecision::NoChange;
    }

    let normalized_current = normalize_intent_text(current_intent);
    let normalized_candidate = normalize_intent_text(candidate_trimmed);

    if normalized_current == normalized_candidate {
        return IntentUpdateDecision::NoChange;
    }

    let low_signal_prefixes = [
        "do this",
        "do that",
        "i want that",
        "i want this",
        "can you",
        "could you",
        "please",
    ];
    if low_signal_prefixes
        .iter()
        .any(|prefix| normalized_candidate.starts_with(prefix))
        && normalized_candidate.split_whitespace().count() <= 10
    {
        return IntentUpdateDecision::NoChange;
    }

    let current_words: HashSet<&str> = normalized_current.split_whitespace().collect();
    let candidate_words: HashSet<&str> = normalized_candidate.split_whitespace().collect();
    let overlap = candidate_words.intersection(&current_words).count();

    let explicit_shift_markers = [
        "new goal",
        "change of plan",
        "instead",
        "let's switch",
        "let us switch",
        "different objective",
        "pivot",
    ];
    if explicit_shift_markers
        .iter()
        .any(|marker| normalized_candidate.contains(marker))
    {
        return IntentUpdateDecision::AskUser;
    }

    let explicit_update_markers = [
        "session objective:",
        "objective:",
        "intent:",
        "my goal is",
        "the goal is",
    ];
    if explicit_update_markers
        .iter()
        .any(|marker| normalized_candidate.contains(marker))
    {
        return IntentUpdateDecision::Update;
    }

    if overlap == 0 && candidate_words.len() >= 6 {
        return IntentUpdateDecision::AskUser;
    }

    if candidate_words.len() >= 16 {
        return IntentUpdateDecision::AskUser;
    }

    IntentUpdateDecision::NoChange
}

// TODO: use it once figured out intent updates
#[allow(dead_code)]
async fn request_intent_update_confirmation(
    config: &AgentConfig,
    session_id: &str,
    current_intent: &str,
    proposed_intent: &str,
) -> Result<bool, AgentError> {
    let elicitation_id = Uuid::new_v4().to_string();
    let (response_tx, response_rx) = oneshot::channel();

    {
        let mut pending = config.pending_elicitations.lock().await;
        pending.insert(elicitation_id.clone(), response_tx);
    }

    let requested_schema = serde_json::json!({
        "type": "object",
        "properties": {
            "selection": {
                "type": "string",
                "title": "Intent",
                "description": "Choose whether to update the session objective.",
                "oneOf": [
                    {
                        "const": "update",
                        "title": "Update objective"
                    },
                    {
                        "const": "keep",
                        "title": "Keep current objective"
                    }
                ]
            }
        },
        "required": ["selection"]
    });

    config
        .event_sink
        .emit_durable(
            session_id,
            AgentEventKind::ElicitationRequested {
                elicitation_id: elicitation_id.clone(),
                session_id: session_id.to_string(),
                message: format!(
                    "This prompt may change the session objective.\n\nCurrent objective:\n{}\n\nProposed objective:\n{}",
                    current_intent, proposed_intent
                ),
                requested_schema,
                source: "builtin:intent_update".to_string(),
            },
        )
        .await
        .map_err(|e| AgentError::Internal(e.to_string()))?;

    let response = tokio::time::timeout(
        std::time::Duration::from_secs(INTENT_CONFIRMATION_TIMEOUT_SECS),
        response_rx,
    )
    .await;

    if response.is_err() {
        let mut pending = config.pending_elicitations.lock().await;
        pending.remove(&elicitation_id);
    }

    let Some(response) = response.ok().and_then(|received| received.ok()) else {
        return Ok(false);
    };

    if response.action != ElicitationAction::Accept {
        return Ok(false);
    }

    let selection = response
        .content
        .as_ref()
        .and_then(|content| content.get("selection"))
        .and_then(|value| value.as_str())
        .unwrap_or("keep");

    Ok(selection == "update")
}

async fn maybe_update_session_intent(
    config: &AgentConfig,
    session_id: &str,
    exec_ctx: &mut ExecutionContext,
    user_text: &str,
) -> Result<(), AgentError> {
    let Some(candidate) = summarize_intent_candidate(user_text) else {
        return Ok(());
    };

    let Some(current_intent) = exec_ctx
        .state
        .current_intent
        .as_ref()
        .map(|intent| intent.summary.clone())
    else {
        exec_ctx
            .state
            .update_intent_snapshot(candidate, None, None)
            .await
            .map_err(|e| AgentError::Internal(e.to_string()))?;
        return Ok(());
    };

    match decide_intent_update(&current_intent, &candidate) {
        IntentUpdateDecision::NoChange => Ok(()),
        IntentUpdateDecision::Update => {
            exec_ctx
                .state
                .update_intent_snapshot(candidate, None, None)
                .await
                .map_err(|e| AgentError::Internal(e.to_string()))?;
            Ok(())
        }
        IntentUpdateDecision::AskUser => {
            // TODO(intent): Re-enable explicit user confirmation for intent shifts
            // once we have a better UX/policy for ambiguous objective changes.
            let _ = (config, session_id, current_intent);
            Ok(())
        }
    }
}

/// Per-session actor. Each session gets its own actor with isolated state.
#[derive(Actor)]
pub struct SessionActor {
    // ── Shared config (read at turn start) ───────────────────────
    pub(crate) config: Arc<AgentConfig>,

    // ── Session identity ─────────────────────────────────────────
    pub(crate) session_id: String,

    // ── Session state (OWNED, NO LOCKS) ──────────────────────────
    pub(crate) runtime: Arc<SessionRuntime>,
    pub(crate) mode: AgentMode,
    pub(crate) reasoning_effort: Option<ReasoningEffort>,
    pub(crate) tool_config: ToolConfig,

    // ── Cancellation ─────────────────────────────────────────────
    /// Tracks the active prompt generation and cancellation token.
    ///
    /// Generation advances when a new `Prompt` is accepted so stale
    /// `PromptFinished` notifications can be ignored safely.
    pub(crate) turn_state: TurnState,

    // ── Client bridge (for SessionUpdate notifications) ──────────
    pub(crate) bridge: Option<ClientBridgeSender>,

    // ── Execution tracking ───────────────────────────────────────
    pub(crate) prompt_running: bool,

    // ── Mesh (remote sessions only) ──────────────────────────────
    /// Present when this actor was spawned on a mesh node via
    /// `RemoteNodeManager`. Used by `SubscribeEvents` to look up the
    /// remote `EventRelayActor` in the Kademlia DHT.
    #[cfg(feature = "remote")]
    pub(crate) mesh: Option<crate::agent::remote::MeshHandle>,

    // Tracks EventForwarder task handles by relay actor so unsubscribe can abort.
    pub(crate) relay_forwarder_handles: HashMap<u64, tokio::task::JoinHandle<()>>,
}

#[derive(Clone)]
pub(crate) struct TurnState {
    pub(crate) generation: u64,
    pub(crate) token: CancellationToken,
}

impl TurnState {
    fn new() -> Self {
        Self {
            generation: 0,
            token: CancellationToken::new(),
        }
    }
}

impl SessionActor {
    /// Create a new SessionActor. Call `kameo::actor::spawn()` to start it.
    pub fn new(config: Arc<AgentConfig>, session_id: String, runtime: Arc<SessionRuntime>) -> Self {
        let tool_config = config.tool_config.clone();
        // Read current default mode from the shared mutex
        let mode = config
            .default_mode
            .lock()
            .map(|m| *m)
            .unwrap_or(AgentMode::Build);
        // Read current default reasoning effort (lock-free via ArcSwap)
        let reasoning_effort = **config.default_reasoning_effort.load();
        Self {
            config,
            session_id,
            runtime,
            mode,
            reasoning_effort,
            tool_config,
            turn_state: TurnState::new(),
            bridge: None,
            prompt_running: false,
            #[cfg(feature = "remote")]
            mesh: None,
            relay_forwarder_handles: HashMap::new(),
        }
    }

    /// Attach a [`MeshHandle`] to this actor (builder pattern, for remote sessions).
    ///
    /// Called by `RemoteNodeManager` after `new()` so the `SubscribeEvents`
    /// handler can reach the Kademlia DHT without querying global state.
    #[cfg(feature = "remote")]
    pub fn with_mesh(mut self, mesh: Option<crate::agent::remote::MeshHandle>) -> Self {
        self.mesh = mesh;
        self
    }

    /// Helper method to extract system prompt from current session config.
    async fn get_session_system_prompt(&self) -> Vec<String> {
        if let Ok(Some(current_config)) = self
            .config
            .provider
            .history_store()
            .get_session_llm_config(&self.session_id)
            .await
            && let Some(params) = &current_config.params
            && let Some(system_array) = params.get("system").and_then(|v| v.as_array())
        {
            let mut system_parts = Vec::new();
            for item in system_array {
                if let Some(s) = item.as_str() {
                    system_parts.push(s.to_string());
                }
            }
            if !system_parts.is_empty() {
                return system_parts;
            }
        }
        self.config.provider.initial_config().system.clone()
    }
}

// ══════════════════════════════════════════════════════════════════════════
//  Message Handlers
// ══════════════════════════════════════════════════════════════════════════

// ── Cancel ───────────────────────────────────────────────────────────────
//
// Cancellation flow:
// 1. Cancel message cancels turn_state.token.
// 2. Detached prompt task observes token.is_cancelled() and exits early.
// 3. PromptFinished applies only when generation matches the current turn.
// 4. Matching PromptFinished resets token and clears prompt_running.

impl Message<Cancel> for SessionActor {
    type Reply = ();

    async fn handle(&mut self, _msg: Cancel, _ctx: &mut Context<Self, Self::Reply>) -> Self::Reply {
        debug!("Session {}: Cancel received", self.session_id);
        self.turn_state.token.cancel();
        self.config
            .emit_event(&self.session_id, AgentEventKind::Cancelled);
    }
}

// ── PromptFinished (internal) ────────────────────────────────────────────

impl Message<PromptFinished> for SessionActor {
    type Reply = ();

    async fn handle(
        &mut self,
        msg: PromptFinished,
        _ctx: &mut Context<Self, Self::Reply>,
    ) -> Self::Reply {
        if msg.generation != self.turn_state.generation {
            debug!(
                "Session {} PromptFinished: stale generation={}, current={} (ignored)",
                self.session_id, msg.generation, self.turn_state.generation
            );
            return;
        }

        self.prompt_running = false;
        // Reset the token so the next turn starts with a clean (uncancelled) token.
        self.turn_state.token = CancellationToken::new();
        debug!(
            "Session {} PromptFinished: prompt_running=false, token reset (generation={})",
            self.session_id, msg.generation
        );
    }
}

// ── SetMode ──────────────────────────────────────────────────────────────

impl Message<SetMode> for SessionActor {
    type Reply = ();

    async fn handle(&mut self, msg: SetMode, _ctx: &mut Context<Self, Self::Reply>) -> Self::Reply {
        self.mode = msg.mode;
        self.config.emit_event(
            &self.session_id,
            AgentEventKind::SessionModeChanged { mode: msg.mode },
        );
    }
}

// ── GetMode ──────────────────────────────────────────────────────────────

impl Message<GetMode> for SessionActor {
    type Reply = Result<AgentMode, kameo::error::Infallible>;

    async fn handle(
        &mut self,
        _msg: GetMode,
        _ctx: &mut Context<Self, Self::Reply>,
    ) -> Self::Reply {
        Ok(self.mode)
    }
}

// ── SetReasoningEffort ────────────────────────────────────────────────────

impl Message<SetReasoningEffort> for SessionActor {
    type Reply = Result<(), AgentError>;

    async fn handle(
        &mut self,
        msg: SetReasoningEffort,
        _ctx: &mut Context<Self, Self::Reply>,
    ) -> Self::Reply {
        // Store in-memory (None = cleared / provider default)
        self.reasoning_effort = msg.effort;

        // Always update the LLM config row so the provider picks up the change
        // (including clearing: None serializes as absent, removing the field).
        let current_config = self
            .config
            .provider
            .history_store()
            .get_session_llm_config(&self.session_id)
            .await
            .map_err(|e| AgentError::Internal(e.to_string()))?;

        if let Some(current) = current_config {
            // Reconstruct LLMParams from the stored config
            let mut params = if let Some(ref params_json) = current.params {
                serde_json::from_value::<querymt::LLMParams>(params_json.clone())
                    .unwrap_or_default()
            } else {
                querymt::LLMParams::new()
            };
            // Set provider/model back (stripped during storage) and apply new effort
            params = params.provider(&current.provider).model(&current.model);
            params.reasoning_effort = msg.effort;

            self.set_llm_config_impl(params).await?;
        }

        Ok(())
    }
}

// ── GetReasoningEffort ───────────────────────────────────────────────────

impl Message<GetReasoningEffort> for SessionActor {
    type Reply = Result<Option<ReasoningEffort>, kameo::error::Infallible>;

    async fn handle(
        &mut self,
        _msg: GetReasoningEffort,
        _ctx: &mut Context<Self, Self::Reply>,
    ) -> Self::Reply {
        Ok(self.reasoning_effort)
    }
}

// ── SetProvider ──────────────────────────────────────────────────────────

impl Message<SetProvider> for SessionActor {
    type Reply = Result<(), AgentError>;

    async fn handle(
        &mut self,
        msg: SetProvider,
        _ctx: &mut Context<Self, Self::Reply>,
    ) -> Self::Reply {
        let system_prompt = self.get_session_system_prompt().await;
        let mut config = querymt::LLMParams::new()
            .provider(&msg.provider)
            .model(&msg.model);
        for prompt_part in system_prompt {
            config = config.system(prompt_part);
        }
        self.set_llm_config_impl(config).await
    }
}

// ── SetLlmConfig ─────────────────────────────────────────────────────────

impl Message<SetLlmConfig> for SessionActor {
    type Reply = Result<(), AgentError>;

    async fn handle(
        &mut self,
        msg: SetLlmConfig,
        _ctx: &mut Context<Self, Self::Reply>,
    ) -> Self::Reply {
        self.set_llm_config_impl(msg.config).await
    }
}

impl SessionActor {
    async fn set_llm_config_impl(&self, config: querymt::LLMParams) -> Result<(), AgentError> {
        let provider_name = config
            .provider
            .as_ref()
            .ok_or(AgentError::ProviderRequired)?;

        let is_known_local_provider = self
            .config
            .provider
            .plugin_registry()
            .get(provider_name)
            .await
            .is_some();

        if !is_known_local_provider {
            let provider_node_id = self
                .config
                .provider
                .history_store()
                .get_session_provider_node_id(&self.session_id)
                .await
                .map_err(|e| AgentError::Internal(e.to_string()))?;

            if provider_node_id.is_none() {
                return Err(AgentError::UnknownProvider {
                    name: provider_name.clone(),
                });
            }
        }

        let llm_config = self
            .config
            .provider
            .history_store()
            .create_or_get_llm_config(&config)
            .await
            .map_err(|e| AgentError::Internal(e.to_string()))?;

        self.config
            .provider
            .history_store()
            .set_session_llm_config(&self.session_id, llm_config.id)
            .await
            .map_err(|e| AgentError::Internal(e.to_string()))?;

        let context_limit =
            crate::model_info::get_model_info(&llm_config.provider, &llm_config.model)
                .and_then(|m| m.context_limit());

        self.config.emit_event(
            &self.session_id,
            AgentEventKind::ProviderChanged {
                provider: llm_config.provider.clone(),
                model: llm_config.model.clone(),
                config_id: llm_config.id,
                context_limit,
                provider_node_id: None,
            },
        );
        Ok(())
    }
}

// ── SetSessionModel ──────────────────────────────────────────────────────

impl Message<SetSessionModel> for SessionActor {
    type Reply = Result<SetSessionModelResponse, AgentError>;

    async fn handle(
        &mut self,
        msg: SetSessionModel,
        _ctx: &mut Context<Self, Self::Reply>,
    ) -> Self::Reply {
        use agent_client_protocol::schema::SetSessionModelResponse;

        let session_id = msg.req.session_id.to_string();
        let _session = self
            .config
            .provider
            .history_store()
            .get_session(&session_id)
            .await
            .map_err(|e| AgentError::Internal(e.to_string()))?
            .ok_or_else(|| AgentError::SessionNotFound {
                session_id: session_id.clone(),
            })?;

        let model_id = msg.req.model_id.to_string();
        let current_config = self
            .config
            .provider
            .history_store()
            .get_session_llm_config(&session_id)
            .await
            .map_err(|e| AgentError::Internal(e.to_string()))?;
        let fallback_provider = current_config
            .as_ref()
            .map(|c| c.provider.as_str())
            .unwrap_or("anthropic");

        let (provider, model) = parse_transport_model_id(&model_id, fallback_provider);

        let system_prompt = self.get_session_system_prompt().await;
        let mut llm_config_input = querymt::LLMParams::new()
            .provider(provider.clone())
            .model(model.clone());
        for prompt_part in system_prompt {
            llm_config_input = llm_config_input.system(prompt_part);
        }

        let llm_config = self
            .config
            .provider
            .history_store()
            .create_or_get_llm_config(&llm_config_input)
            .await
            .map_err(|e| AgentError::Internal(e.to_string()))?;

        self.config
            .provider
            .history_store()
            .set_session_llm_config(&session_id, llm_config.id)
            .await
            .map_err(|e| AgentError::Internal(e.to_string()))?;

        // Persist the provider_node_id so the session knows where to route LLM calls.
        let provider_node_id = msg.provider_node_id.as_ref().map(ToString::to_string);
        if let Err(e) = self
            .config
            .provider
            .history_store()
            .set_session_provider_node_id(&session_id, provider_node_id.as_deref())
            .await
        {
            log::warn!("SetSessionModel: failed to persist provider_node_id: {}", e);
        }

        let context_limit =
            crate::model_info::get_model_info(&llm_config.provider, &llm_config.model)
                .and_then(|m| m.context_limit());

        self.config.emit_event(
            &session_id,
            AgentEventKind::ProviderChanged {
                provider: llm_config.provider.clone(),
                model: llm_config.model.clone(),
                config_id: llm_config.id,
                context_limit,
                provider_node_id,
            },
        );

        Ok(SetSessionModelResponse::new())
    }
}

// ── Tool Config Messages ─────────────────────────────────────────────────

impl Message<SetToolPolicy> for SessionActor {
    type Reply = ();

    async fn handle(
        &mut self,
        msg: SetToolPolicy,
        _ctx: &mut Context<Self, Self::Reply>,
    ) -> Self::Reply {
        self.tool_config.policy = msg.policy;
    }
}

impl Message<SetAllowedTools> for SessionActor {
    type Reply = ();

    async fn handle(
        &mut self,
        msg: SetAllowedTools,
        _ctx: &mut Context<Self, Self::Reply>,
    ) -> Self::Reply {
        self.tool_config.allowlist = Some(msg.tools.into_iter().collect());
    }
}

impl Message<ClearAllowedTools> for SessionActor {
    type Reply = ();

    async fn handle(
        &mut self,
        _msg: ClearAllowedTools,
        _ctx: &mut Context<Self, Self::Reply>,
    ) -> Self::Reply {
        self.tool_config.allowlist = None;
    }
}

impl Message<SetDeniedTools> for SessionActor {
    type Reply = ();

    async fn handle(
        &mut self,
        msg: SetDeniedTools,
        _ctx: &mut Context<Self, Self::Reply>,
    ) -> Self::Reply {
        self.tool_config.denylist = msg.tools.into_iter().collect();
    }
}

impl Message<ClearDeniedTools> for SessionActor {
    type Reply = ();

    async fn handle(
        &mut self,
        _msg: ClearDeniedTools,
        _ctx: &mut Context<Self, Self::Reply>,
    ) -> Self::Reply {
        self.tool_config.denylist.clear();
    }
}

// ── State Queries ────────────────────────────────────────────────────────

impl Message<GetSessionLimits> for SessionActor {
    type Reply = Result<Option<SessionLimits>, kameo::error::Infallible>;

    async fn handle(
        &mut self,
        _msg: GetSessionLimits,
        _ctx: &mut Context<Self, Self::Reply>,
    ) -> Self::Reply {
        Ok(self.config.get_session_limits())
    }
}

impl Message<GetRuntimeStatus> for SessionActor {
    type Reply = Result<SessionRuntimeStatus, kameo::error::Infallible>;

    async fn handle(
        &mut self,
        _msg: GetRuntimeStatus,
        _ctx: &mut Context<Self, Self::Reply>,
    ) -> Self::Reply {
        let status = if !self.prompt_running {
            SessionRuntimeStatus::Idle
        } else if self.turn_state.token.is_cancelled() {
            SessionRuntimeStatus::CancelRequested
        } else {
            SessionRuntimeStatus::Running
        };
        Ok(status)
    }
}

impl Message<GetLlmConfig> for SessionActor {
    type Reply = Result<Option<LLMConfig>, AgentError>;

    async fn handle(
        &mut self,
        _msg: GetLlmConfig,
        _ctx: &mut Context<Self, Self::Reply>,
    ) -> Self::Reply {
        self.config
            .provider
            .history_store()
            .get_session_llm_config(&self.session_id)
            .await
            .map_err(|e| AgentError::Internal(e.to_string()))
    }
}

// ── Undo / Redo ──────────────────────────────────────────────────────────

impl Message<Undo> for SessionActor {
    type Reply = Result<UndoResult, UndoError>;

    async fn handle(&mut self, msg: Undo, _ctx: &mut Context<Self, Self::Reply>) -> Self::Reply {
        let backend = self
            .config
            .snapshot_backend
            .as_ref()
            .ok_or(UndoError::NoSnapshotBackend)?;

        let worktree = self
            .runtime
            .cwd
            .as_ref()
            .ok_or(UndoError::NoWorkingDirectory)?
            .to_path_buf();

        // Delegate to free function using our owned state
        crate::agent::undo::undo_impl(
            backend.as_ref(),
            &self.config.provider,
            &self.session_id,
            &msg.message_id,
            &worktree,
        )
        .await
    }
}

impl Message<Redo> for SessionActor {
    type Reply = Result<RedoResult, UndoError>;

    async fn handle(&mut self, _msg: Redo, _ctx: &mut Context<Self, Self::Reply>) -> Self::Reply {
        let backend = self
            .config
            .snapshot_backend
            .as_ref()
            .ok_or(UndoError::NoSnapshotBackend)?;

        let worktree = self
            .runtime
            .cwd
            .as_ref()
            .ok_or(UndoError::NoWorkingDirectory)?
            .to_path_buf();

        crate::agent::undo::redo_impl(
            backend.as_ref(),
            &self.config.provider,
            &self.session_id,
            &worktree,
        )
        .await
    }
}

impl Message<ForkAtMessage> for SessionActor {
    type Reply = Result<String, AgentError>;

    async fn handle(
        &mut self,
        msg: ForkAtMessage,
        _ctx: &mut Context<Self, Self::Reply>,
    ) -> Self::Reply {
        self.config
            .provider
            .history_store()
            .fork_session(
                &self.session_id,
                &msg.message_id,
                crate::session::domain::ForkOrigin::User,
            )
            .await
            .map_err(|e| AgentError::Internal(e.to_string()))
    }
}

impl Message<ExtMethod> for SessionActor {
    type Reply = Result<ExtResponse, AgentError>;

    async fn handle(
        &mut self,
        _msg: ExtMethod,
        _ctx: &mut Context<Self, Self::Reply>,
    ) -> Self::Reply {
        let raw_value = serde_json::value::RawValue::from_string("null".to_string())
            .map_err(|e| AgentError::Serialization(e.to_string()))?;
        Ok(ExtResponse::new(std::sync::Arc::from(raw_value)))
    }
}

impl Message<ExtNotification> for SessionActor {
    type Reply = Result<(), AgentError>;

    async fn handle(
        &mut self,
        _msg: ExtNotification,
        _ctx: &mut Context<Self, Self::Reply>,
    ) -> Self::Reply {
        Ok(())
    }
}

impl Message<SetBridge> for SessionActor {
    type Reply = ();

    async fn handle(
        &mut self,
        msg: SetBridge,
        _ctx: &mut Context<Self, Self::Reply>,
    ) -> Self::Reply {
        self.bridge = Some(msg.bridge);
    }
}

impl Message<Shutdown> for SessionActor {
    type Reply = ();

    async fn handle(
        &mut self,
        _msg: Shutdown,
        ctx: &mut Context<Self, Self::Reply>,
    ) -> Self::Reply {
        ctx.stop();
    }
}

// ── Remote-ready messages ────────────────────────────────────────────────

/// Get the full message history for this session.
///
/// Reads from the local `SessionStore`. When called remotely, the result
/// is serialized and sent back over the kameo mesh.
impl Message<crate::agent::messages::GetHistory> for SessionActor {
    type Reply = Result<Vec<AgentMessage>, AgentError>;

    async fn handle(
        &mut self,
        _msg: crate::agent::messages::GetHistory,
        _ctx: &mut Context<Self, Self::Reply>,
    ) -> Self::Reply {
        self.config
            .provider
            .history_store()
            .get_history(&self.session_id)
            .await
            .map_err(|e| AgentError::Internal(e.to_string()))
    }
}

/// Return the full durable event stream for this session from the local journal.
impl Message<crate::agent::messages::GetEventStream> for SessionActor {
    type Reply = Result<Vec<crate::events::AgentEvent>, AgentError>;

    async fn handle(
        &mut self,
        _msg: crate::agent::messages::GetEventStream,
        _ctx: &mut Context<Self, Self::Reply>,
    ) -> Self::Reply {
        self.config
            .event_sink
            .journal()
            .load_session_stream(&self.session_id, None, None)
            .await
            .map(|events| {
                events
                    .into_iter()
                    .map(crate::events::AgentEvent::from)
                    .collect()
            })
            .map_err(|e| AgentError::Internal(e.to_string()))
    }
}

/// Subscribe a remote event relay to this session's events.
///
/// When the kameo swarm is bootstrapped, this handler:
/// 1. Resolves `relay_actor_id` → `RemoteActorRef<EventRelayActor>` via swarm
/// 2. Starts an `EventForwarder` background task subscribed to the EventFanout
///
/// Without a swarm (swarm not bootstrapped), it logs and returns Ok so the
/// message round-trips correctly for local tests.
impl Message<crate::agent::messages::SubscribeEvents> for SessionActor {
    type Reply = Result<(), AgentError>;

    async fn handle(
        &mut self,
        msg: crate::agent::messages::SubscribeEvents,
        _ctx: &mut Context<Self, Self::Reply>,
    ) -> Self::Reply {
        #[cfg(feature = "remote")]
        {
            use crate::agent::remote::event_forwarder::EventForwarder;
            use crate::agent::remote::event_relay::EventRelayActor;

            if let Some(ref mesh) = self.mesh {
                let relay_name = msg.relay_dht_name.clone();
                match mesh
                    .lookup_actor::<EventRelayActor>(relay_name.clone())
                    .await
                {
                    Ok(Some(relay_ref)) => {
                        // Abort previous forwarder for this relay_actor_id if any
                        if let Some(prev_handle) =
                            self.relay_forwarder_handles.remove(&msg.relay_actor_id)
                        {
                            prev_handle.abort();
                            log::debug!(
                                "Session {}: SubscribeEvents relay_actor_id={} — aborted previous forwarder",
                                self.session_id,
                                msg.relay_actor_id
                            );
                        }

                        let fanout = self.config.event_sink.fanout().clone();
                        let handle = EventForwarder::start(
                            fanout,
                            relay_ref,
                            format!("session:{}", self.session_id),
                            self.session_id.clone(),
                        );
                        self.relay_forwarder_handles
                            .insert(msg.relay_actor_id, handle);

                        // Catch-up for subscribe-after-ready race:
                        // if the workspace index finished before this relay subscribed,
                        // re-emit WorkspaceIndexReady now that forwarding is installed.
                        if let Some(workspace) = self.runtime.workspace_handle.get()
                            && let Some(index) = workspace.file_index()
                            && let Err(e) = self
                                .config
                                .emit_event_persisted(
                                    &self.session_id,
                                    crate::events::AgentEventKind::WorkspaceIndexReady {
                                        workspace_root: index.root.display().to_string(),
                                    },
                                )
                                .await
                        {
                            warn!(
                                "Session {}: failed to emit WorkspaceIndexReady catch-up after SubscribeEvents: {}",
                                self.session_id, e
                            );
                        }

                        log::debug!(
                            "Session {}: SubscribeEvents — EventForwarder started for relay '{}' (relay_actor_id={})",
                            self.session_id,
                            relay_name,
                            msg.relay_actor_id,
                        );
                    }
                    Ok(None) => {
                        warn!(
                            "Session {}: SubscribeEvents — no relay actor found under '{}' in DHT yet",
                            self.session_id, relay_name
                        );
                    }
                    Err(e) => {
                        warn!(
                            "Session {}: SubscribeEvents relay lookup failed: {} (continuing without relay)",
                            self.session_id, e
                        );
                    }
                }
            } else {
                info!(
                    "Session {}: SubscribeEvents relay_actor_id={} — no mesh, event relay skipped",
                    self.session_id, msg.relay_actor_id
                );
            }
        }

        #[cfg(not(feature = "remote"))]
        {
            info!(
                "Session {}: SubscribeEvents relay_actor_id={} (remote feature not enabled)",
                self.session_id, msg.relay_actor_id
            );
        }

        Ok(())
    }
}

/// Unsubscribe a previously registered event relay.
///
/// Aborts the `EventForwarder` background task for `relay_actor_id`.
/// Without a swarm, this is a no-op (nothing was registered).
impl Message<crate::agent::messages::UnsubscribeEvents> for SessionActor {
    type Reply = Result<(), AgentError>;

    async fn handle(
        &mut self,
        msg: crate::agent::messages::UnsubscribeEvents,
        _ctx: &mut Context<Self, Self::Reply>,
    ) -> Self::Reply {
        if let Some(handle) = self.relay_forwarder_handles.remove(&msg.relay_actor_id) {
            handle.abort();
            info!(
                "Session {}: UnsubscribeEvents relay_actor_id={} — forwarder task aborted",
                self.session_id, msg.relay_actor_id
            );
        } else {
            info!(
                "Session {}: UnsubscribeEvents relay_actor_id={} had no registered forwarder",
                self.session_id, msg.relay_actor_id
            );
        }
        Ok(())
    }
}

/// Set planning context on a delegate session.
///
/// Appends the parent session's planning summary to this session's
/// system prompt. Used by the delegation orchestrator to inject context
/// without requiring direct access to the session's `SessionStore`.
impl Message<crate::agent::messages::SetPlanningContext> for SessionActor {
    type Reply = Result<(), AgentError>;

    async fn handle(
        &mut self,
        msg: crate::agent::messages::SetPlanningContext,
        _ctx: &mut Context<Self, Self::Reply>,
    ) -> Self::Reply {
        info!(
            "Session {}: SetPlanningContext ({} bytes)",
            self.session_id,
            msg.summary.len()
        );

        // Get current system prompt and append the planning summary
        let mut system_prompt = self.get_session_system_prompt().await;
        system_prompt.push(msg.summary);

        // Rebuild the LLM config with the updated system prompt
        let current_config = self
            .config
            .provider
            .history_store()
            .get_session_llm_config(&self.session_id)
            .await
            .map_err(|e| AgentError::Internal(e.to_string()))?;

        if let Some(current) = current_config {
            // Reconstruct full LLMParams from the stored config JSON.
            // This preserves provider-specific custom params (e.g. text_only,
            // n_ctx, flash_attention for llama_cpp) that are stored via
            // #[serde(flatten)] on LLMParams.custom.  Previously this built
            // a fresh LLMParams with only provider/model/system, silently
            // dropping all custom params — causing delegate sessions to lose
            // their inference configuration.
            let mut llm_params = if let Some(ref params_json) = current.params {
                serde_json::from_value::<querymt::LLMParams>(params_json.clone())
                    .unwrap_or_default()
            } else {
                querymt::LLMParams::new()
            };
            // Re-add provider/model (stored in separate columns, stripped from params JSON)
            llm_params = llm_params.provider(&current.provider).model(&current.model);
            // Replace system prompt with updated version (original + planning context)
            llm_params.system = system_prompt;

            let new_config = self
                .config
                .provider
                .history_store()
                .create_or_get_llm_config(&llm_params)
                .await
                .map_err(|e| AgentError::Internal(e.to_string()))?;

            self.config
                .provider
                .history_store()
                .set_session_llm_config(&self.session_id, new_config.id)
                .await
                .map_err(|e| AgentError::Internal(e.to_string()))?;
        }

        Ok(())
    }
}

// ── File Proxy ───────────────────────────────────────────────────────────

impl Message<crate::agent::messages::GetFileIndex> for SessionActor {
    type Reply = Result<
        crate::agent::file_proxy::GetFileIndexResponse,
        crate::agent::file_proxy::FileProxyError,
    >;

    async fn handle(
        &mut self,
        _msg: crate::agent::messages::GetFileIndex,
        _ctx: &mut Context<Self, Self::Reply>,
    ) -> Self::Reply {
        use crate::agent::file_proxy::{FileProxyError, GetFileIndexResponse};

        let handle = self
            .runtime
            .workspace_handle
            .get()
            .ok_or(FileProxyError::IndexNotReady)?;
        let index = handle.file_index().ok_or(FileProxyError::IndexNotReady)?;
        Ok(GetFileIndexResponse {
            files: index.files.clone(),
            generated_at: index.generated_at,
            workspace_root: index.root.display().to_string(),
        })
    }
}

impl Message<crate::agent::messages::ReadRemoteFile> for SessionActor {
    type Reply = Result<
        crate::agent::file_proxy::ReadRemoteFileResponse,
        crate::agent::file_proxy::FileProxyError,
    >;

    async fn handle(
        &mut self,
        msg: crate::agent::messages::ReadRemoteFile,
        _ctx: &mut Context<Self, Self::Reply>,
    ) -> Self::Reply {
        use crate::agent::file_proxy::{FileProxyError, ReadRemoteFileResponse};
        use crate::tools::builtins::read_shared::render_read_output;
        use base64::Engine as _;

        let cwd = self
            .runtime
            .cwd
            .as_ref()
            .ok_or(FileProxyError::NoWorkingDirectory)?;

        let path = std::path::Path::new(&msg.path);
        let joined = if path.is_absolute() {
            path.to_path_buf()
        } else {
            cwd.join(path)
        };

        let resolved = joined
            .canonicalize()
            .map_err(|e| FileProxyError::PathResolution(e.to_string()))?;

        let root = crate::index::resolve_workspace_root(cwd);
        if !resolved.starts_with(&root) {
            return Err(FileProxyError::PathOutsideWorkspace(
                resolved.display().to_string(),
            ));
        }

        let blocks = render_read_output(&resolved, msg.offset, msg.limit)
            .await
            .map_err(FileProxyError::ReadError)?;

        // Map the single Content block returned by render_read_output to the
        // appropriate ReadRemoteFileResponse variant.
        for block in blocks {
            match block {
                querymt::chat::Content::Image { mime_type, data } => {
                    return Ok(ReadRemoteFileResponse::Image {
                        mime_type,
                        base64_data: base64::engine::general_purpose::STANDARD.encode(&data),
                    });
                }
                querymt::chat::Content::Text { text } => {
                    return Ok(ReadRemoteFileResponse::Text(text));
                }
                _ => {}
            }
        }

        Ok(ReadRemoteFileResponse::Binary)
    }
}

// ── Prompt (the big one) ─────────────────────────────────────────────────

impl Message<Prompt> for SessionActor {
    type Reply = DelegatedReply<Result<PromptResponse, AgentError>>;

    async fn handle(&mut self, msg: Prompt, ctx: &mut Context<Self, Self::Reply>) -> Self::Reply {
        if self.prompt_running {
            // Allow new prompts through even while one is running.
            // execution_permit still serializes actual turn execution, so this
            // behaves like queueing instead of fail-fast rejection.
            if self.turn_state.token.is_cancelled() {
                debug!(
                    "Session {}: prompt_running=true, cancelled=true → allowing new prompt through",
                    self.session_id
                );
            } else {
                debug!(
                    "Session {}: prompt_running=true, cancelled=false → queueing behind running prompt",
                    self.session_id
                );
            }
        } else {
            debug!(
                "Session {}: prompt_running=false → accepting new prompt",
                self.session_id
            );
        }

        self.prompt_running = true;

        // Advance generation for this prompt so stale PromptFinished messages from
        // older detached tasks cannot mutate current state.
        self.turn_state.generation = self.turn_state.generation.saturating_add(1);
        let prompt_generation = self.turn_state.generation;

        // Only create a fresh token when the previous token was already cancelled.
        // Otherwise keep the token so Cancel still reaches the currently-running task
        // while additional prompts wait behind the execution permit.
        if self.turn_state.token.is_cancelled() {
            self.turn_state.token = CancellationToken::new();
            debug!(
                "Session {}: set prompt_running=true, generation={}, fresh token created (previous was cancelled)",
                self.session_id, prompt_generation
            );
        } else {
            debug!(
                "Session {}: set prompt_running=true, generation={}, reusing existing token",
                self.session_id, prompt_generation
            );
        }

        // Capture everything needed for the detached task
        let config = self.config.clone();
        let session_id = self.session_id.clone();
        let runtime = self.runtime.clone();
        let cancel_token = self.turn_state.token.clone();
        let bridge = self.bridge.clone();
        let mode = self.mode;
        let tool_config = self.tool_config.clone();
        let actor_ref = ctx.actor_ref().clone();
        let span_session_id = session_id.clone();

        // Capture the current span (kameo's `actor.handle_message`) so the
        // spawned task's span tree remains a child of the actor message span.
        // Without this, `tokio::spawn` breaks the tracing context and the
        // entire execution trace appears as an orphaned root.
        let parent_span = tracing::Span::current();

        ctx.spawn(
            async move {
                let result = execute_prompt_detached(DetachedPromptExecution {
                    req: msg.req,
                    session_id: session_id.clone(),
                    runtime,
                    config,
                    cancel_token,
                    bridge,
                    mode,
                    tool_config,
                    execution_origin: crate::agent::execution_context::ExecutionOrigin::Interactive,
                })
                .await;

                debug!("Session {}: sending PromptFinished to actor", session_id);
                // Reset prompt_running flag via message back to actor
                if let Err(e) = actor_ref
                    .tell(PromptFinished {
                        generation: prompt_generation,
                    })
                    .await
                {
                    debug!(
                        "Failed to send PromptFinished message to actor ({}): {:?}. \
                         Actor may have been shutdown — next prompt will reset via generation guard.",
                        session_id, e
                    );
                } else {
                    debug!("Session {}: PromptFinished sent successfully", session_id);
                }

                result
            }
            .instrument(info_span!(
                parent: &parent_span,
                "agent.prompt.task",
                session_id = %span_session_id,
                prompt_generation,
                execution_origin = ?crate::agent::execution_context::ExecutionOrigin::Interactive,
                mode = %mode,
            )),
        )
    }
}

// ── ScheduledPrompt ──────────────────────────────────────────────────────
//
// Fired by SchedulerActor. Similar to Prompt but:
// 1. Builds a PromptRequest from the schedule's prompt_text.
// 2. Sets ExecutionOrigin::Scheduled on the execution context.
// 3. Emits ScheduledExecutionCompleted / ScheduledExecutionFailed terminal
//    events so the SchedulerActor can correlate cycle outcomes.

impl Message<ScheduledPrompt> for SessionActor {
    type Reply = DelegatedReply<Result<PromptResponse, AgentError>>;

    async fn handle(
        &mut self,
        msg: ScheduledPrompt,
        ctx: &mut Context<Self, Self::Reply>,
    ) -> Self::Reply {
        debug!(
            "Session {}: ScheduledPrompt received (schedule={})",
            self.session_id, msg.schedule_public_id
        );

        if self.prompt_running {
            if self.turn_state.token.is_cancelled() {
                debug!(
                    "Session {}: scheduled prompt: prompt_running=true, cancelled=true → allowing through",
                    self.session_id
                );
            } else {
                debug!(
                    "Session {}: scheduled prompt: prompt_running=true → queueing behind running prompt",
                    self.session_id
                );
            }
        }

        self.prompt_running = true;
        self.turn_state.generation = self.turn_state.generation.saturating_add(1);
        let prompt_generation = self.turn_state.generation;

        if self.turn_state.token.is_cancelled() {
            self.turn_state.token = CancellationToken::new();
        }

        // Build a PromptRequest from the schedule's prompt text
        let prompt_request = agent_client_protocol::schema::PromptRequest::new(
            agent_client_protocol::schema::SessionId::from(self.session_id.clone()),
            vec![agent_client_protocol::schema::ContentBlock::Text(
                agent_client_protocol::schema::TextContent::new(msg.prompt_text),
            )],
        );

        let config = self.config.clone();
        let session_id = self.session_id.clone();
        let runtime = self.runtime.clone();
        let cancel_token = self.turn_state.token.clone();
        let bridge = self.bridge.clone();
        let mode = self.mode;
        let tool_config = self.tool_config.clone();
        let schedule_public_id = msg.schedule_public_id.clone();
        let actor_ref = ctx.actor_ref().clone();
        let span_session_id = session_id.clone();

        // Capture the current span (kameo's `actor.handle_message`) so the
        // spawned task's span tree remains a child of the actor message span.
        let parent_span = tracing::Span::current();

        ctx.spawn(
            async move {
                let result = execute_prompt_detached(DetachedPromptExecution {
                    req: prompt_request,
                    session_id: session_id.clone(),
                    runtime,
                    config: config.clone(),
                    cancel_token,
                    bridge,
                    mode,
                    tool_config,
                    execution_origin: crate::agent::execution_context::ExecutionOrigin::Scheduled {
                        schedule_public_id: schedule_public_id.clone(),
                    },
                })
                .await;

                // Emit explicit scheduled terminal events for SchedulerActor correlation
                let turn_id = Uuid::new_v4().to_string();
                match &result {
                    Ok(_) => {
                        config.emit_event(
                            &session_id,
                            AgentEventKind::ScheduledExecutionCompleted {
                                schedule_public_id,
                                turn_id,
                            },
                        );
                    }
                    Err(e) => {
                        config.emit_event(
                            &session_id,
                            AgentEventKind::ScheduledExecutionFailed {
                                schedule_public_id,
                                turn_id: None,
                                error: e.to_string(),
                            },
                        );
                    }
                }

                if let Err(e) = actor_ref
                    .tell(PromptFinished {
                        generation: prompt_generation,
                    })
                    .await
                {
                    warn!(
                        "Failed to send PromptFinished after scheduled execution: {:?}",
                        e
                    );
                }

                result
            }
            .instrument(info_span!(
                parent: &parent_span,
                "agent.scheduled_prompt.task",
                session_id = %span_session_id,
                prompt_generation,
                execution_origin = "scheduled",
                mode = %mode,
            )),
        )
    }
}

/// Execute a prompt in a detached task (called from `ctx.spawn()`).
///
/// This function gathers all needed state upfront and runs the full execution cycle.
/// It does NOT access the actor — everything is passed as parameters.
struct DetachedPromptExecution {
    req: agent_client_protocol::schema::PromptRequest,
    session_id: String,
    runtime: Arc<SessionRuntime>,
    config: Arc<AgentConfig>,
    cancel_token: CancellationToken,
    bridge: Option<ClientBridgeSender>,
    mode: AgentMode,
    tool_config: ToolConfig,
    /// Origin of this execution — interactive or scheduled.
    execution_origin: crate::agent::execution_context::ExecutionOrigin,
}

enum PermitWaitOutcome {
    Acquired(OwnedSemaphorePermit),
    TimedOut,
    Cancelled,
}

async fn acquire_execution_permit_with_timeout(
    permit: &Arc<tokio::sync::Semaphore>,
    cancel_token: &CancellationToken,
    timeout: std::time::Duration,
) -> Result<PermitWaitOutcome, AgentError> {
    tokio::select! {
        permit = tokio::time::timeout(timeout, permit.clone().acquire_owned()) => match permit {
            Ok(result) => result
                .map(PermitWaitOutcome::Acquired)
                .map_err(|_| AgentError::SessionSemaphoreClosed),
            Err(_) => Ok(PermitWaitOutcome::TimedOut),
        },
        _ = cancel_token.cancelled() => Ok(PermitWaitOutcome::Cancelled),
    }
}

#[instrument(
    name = "agent.prompt.execute",
    skip(exec),
    fields(
        session_id = %exec.session_id,
        execution_origin = ?exec.execution_origin,
        mode = %exec.mode,
        has_bridge = exec.bridge.is_some(),
        cwd = tracing::field::display(
            exec.runtime
                .cwd
                .as_ref()
                .map(|p| p.display().to_string())
                .unwrap_or_else(|| "none".to_string())
        )
    )
)]
async fn execute_prompt_detached(
    exec: DetachedPromptExecution,
) -> Result<PromptResponse, AgentError> {
    let DetachedPromptExecution {
        req,
        session_id,
        runtime,
        config,
        cancel_token,
        bridge,
        mode,
        tool_config,
        execution_origin,
    } = exec;
    debug!(
        "Prompt request for session {} with {} block(s)",
        session_id,
        req.prompt.len()
    );

    // Acquire execution permit (blocking with timeout)
    let _permit = match acquire_execution_permit_with_timeout(
        &runtime.execution_permit,
        &cancel_token,
        std::time::Duration::from_millis(100),
    )
    .await?
    {
        PermitWaitOutcome::Acquired(permit) => permit,
        PermitWaitOutcome::Cancelled => {
            return Ok(PromptResponse::new(StopReason::Cancelled));
        }
        PermitWaitOutcome::TimedOut => {
            debug!(
                "Session {} is busy, waiting for previous operation to complete...",
                session_id
            );
            config.emit_event(
                &session_id,
                AgentEventKind::SessionQueued {
                    reason: "waiting for previous operation to complete".to_string(),
                },
            );
            let timeout_duration = std::time::Duration::from_secs(config.execution_timeout_secs);
            match acquire_execution_permit_with_timeout(
                &runtime.execution_permit,
                &cancel_token,
                timeout_duration,
            )
            .await?
            {
                PermitWaitOutcome::Acquired(permit) => permit,
                PermitWaitOutcome::Cancelled => {
                    return Ok(PromptResponse::new(StopReason::Cancelled));
                }
                PermitWaitOutcome::TimedOut => {
                    return Err(AgentError::SessionTimeout {
                        details: format!(
                            "session_id={}, timeout={}s",
                            session_id, config.execution_timeout_secs
                        ),
                    });
                }
            }
        }
    };

    // Clean up revert state if a new prompt is sent while in reverted state.
    // Also prunes the event journal so undone turns don't reappear on reload.
    // IMPORTANT: this must run BEFORE loading RuntimeContext so that
    // `get_history` / `load_working_context` only sees the surviving messages.
    if let Err(e) = crate::agent::undo::cleanup_revert_on_prompt(
        &config.provider,
        &**config.event_sink.journal(),
        &session_id,
    )
    .await
    {
        warn!("Failed to clean up revert state: {}", e);
    }

    // Get Session Context (turn-pinned)
    let session_handle = config
        .provider
        .with_session(&session_id)
        .await
        .map_err(|e| AgentError::Internal(e.to_string()))?;

    // Create and load RuntimeContext
    let mut runtime_context =
        RuntimeContext::new(config.provider.history_store(), session_id.clone())
            .await
            .map_err(|e| AgentError::Internal(e.to_string()))?;
    runtime_context
        .load_working_context()
        .await
        .map_err(|e| AgentError::Internal(e.to_string()))?;

    // Create execution context — attach the cancellation token so it propagates
    // into individual tool calls for cooperative cancellation.
    // The knowledge store is propagated so knowledge tools can access it.
    let turn_id = Uuid::new_v4().to_string();
    let mut exec_ctx = ExecutionContext::new(
        session_id.clone(),
        runtime.clone(),
        runtime_context,
        session_handle,
        tool_config,
    )
    .with_cancellation_token(cancel_token.clone())
    .with_execution_origin(execution_origin)
    .with_knowledge_store(config.knowledge_store())
    .with_event_sink(config.event_sink.clone())
    .with_workspace_query_bridge(bridge.clone())
    .with_turn_id(turn_id.clone())
    .with_turn_mode(mode);

    // 4. Slash command expansion (before storing user messages)
    let mut req = req;
    if let Some(agent_client_protocol::schema::ContentBlock::Text(text_content)) =
        req.prompt.first()
        && let Some(expansion) =
            crate::slash_commands::try_expand(&text_content.text, &config.slash_command_registry)
    {
        log::info!(
            "Session {}: expanding slash command '/{}'",
            session_id,
            expansion.invocation.name
        );
        // Replace the first text block with the expanded prompt
        req.prompt[0] = agent_client_protocol::schema::ContentBlock::Text(
            agent_client_protocol::schema::TextContent::new(expansion.prompt_text),
        );
    }

    // 5. Store User Messages
    // Keep separate projections for user-visible events vs LLM replay context.
    let display_content = render_prompt_for_display(&req.prompt);

    // User text only for intent snapshot (clean, no attachments)
    let user_text = format_prompt_user_text_only(&req.prompt);

    let message_id = Uuid::new_v4().to_string();

    let hook_result = config
        .hooks
        .run_user_prompt_submit(crate::hooks::UserPromptSubmitRequest {
            session_id: session_id.clone(),
            turn_id: turn_id.clone(),
            cwd: exec_ctx.cwd().map(|p| p.to_path_buf()),
            model: exec_ctx
                .llm_config()
                .map(|cfg| cfg.model.clone())
                .unwrap_or_default(),
            permission_mode: exec_ctx.permission_mode().to_string(),
            prompt: user_text.clone(),
        })
        .await
        .map_err(|e| AgentError::Internal(e.to_string()))?;
    emit_hook_notices(&config, &session_id, hook_result.notices.clone());
    if hook_result.should_block {
        let reason = hook_result
            .block_reason
            .unwrap_or_else(|| "prompt blocked by hook".to_string());
        config.emit_event(
            &session_id,
            AgentEventKind::Error {
                message: reason.clone(),
            },
        );
        return Err(AgentError::Internal(reason));
    }
    if !hook_result.additional_contexts.is_empty() {
        log::debug!(
            "Session {}: ignoring {} hook additional_context item(s) for MVP",
            session_id,
            hook_result.additional_contexts.len()
        );
    }

    config.emit_event(
        &session_id,
        AgentEventKind::PromptReceived {
            content: display_content.clone(),
            message_id: Some(message_id.clone()),
        },
    );

    maybe_update_session_intent(&config, &session_id, &mut exec_ctx, &user_text).await?;

    let agent_msg = AgentMessage {
        id: message_id,
        session_id: session_id.clone(),
        role: ChatRole::User,
        parts: vec![MessagePart::Prompt {
            blocks: req.prompt.clone(),
        }],
        created_at: time::OffsetDateTime::now_utc().unix_timestamp(),
        parent_message_id: None,
        source_provider: None,
        source_model: None,
    };

    if let Err(e) = exec_ctx.add_message(agent_msg).await {
        config.emit_event(
            &session_id,
            AgentEventKind::Error {
                message: e.to_string(),
            },
        );
        return Err(AgentError::Internal(e.to_string()));
    }

    config.emit_event(
        &session_id,
        AgentEventKind::UserMessageStored {
            content: display_content.clone(),
        },
    );

    config.emit_event(&session_id, AgentEventKind::TurnStarted);

    debug!(
        "Session {}: user_message_stored, starting pre-turn snapshot",
        session_id
    );

    // Pre-turn snapshot for undo/redo (off critical path)
    if let Some(ref backend) = config.snapshot_backend
        && let Some(worktree) = runtime.cwd.as_ref()
    {
        let turn_id = Uuid::new_v4().to_string();
        let backend = Arc::clone(backend);
        let worktree = worktree.to_path_buf();
        let worktree_display = worktree.display().to_string();
        let session_id_for_task = session_id.clone();
        let turn_id_for_task = turn_id.clone();

        let task = tokio::spawn(
            async move {
                let started = std::time::Instant::now();
                let snapshot_result = backend.track(&worktree).await.map_err(|e| e.to_string());
                let elapsed = started.elapsed();
                match &snapshot_result {
                    Ok(snapshot_id) => debug!(
                        "Session {}: pre-turn snapshot ready in {:?} (turn_id={}, snapshot_id={})",
                        session_id_for_task, elapsed, turn_id_for_task, snapshot_id
                    ),
                    Err(err) => warn!(
                        "Session {}: pre-turn snapshot failed in {:?} (turn_id={}): {}",
                        session_id_for_task, elapsed, turn_id_for_task, err
                    ),
                }
                (turn_id_for_task, snapshot_result)
            }
            .instrument(info_span!(
                "agent.snapshot.pre_turn.track",
                session_id = %session_id,
                turn_id = %turn_id,
                worktree = %worktree_display
            )),
        );

        *runtime.pre_turn_snapshot_task.lock() = Some(task);
    } else {
        debug!(
            "Session {}: no snapshot backend, skipping pre-turn snapshot",
            session_id
        );
        *runtime.pre_turn_snapshot_task.lock() = None;
    }

    debug!(
        "Session {}: entering execute_cycle_state_machine, cancelled={}",
        session_id,
        exec_ctx.cancellation_token.is_cancelled()
    );

    // Execute Agent Loop using State Machine
    let result = crate::agent::execution::execute_cycle_state_machine(
        &config,
        &mut exec_ctx,
        bridge.clone(),
        mode,
    )
    .await;

    debug!(
        "Session {}: state machine returned: {:?}",
        session_id,
        result
            .as_ref()
            .map(|o| format!("{:?}", o))
            .unwrap_or_else(|e| e.to_string())
    );

    debug!("Session {}: post-turn snapshot start", session_id);

    // Post-turn snapshot
    if let Some(ref backend) = config.snapshot_backend
        && let Some(worktree) = runtime.cwd.as_ref()
    {
        if let Err(e) = ensure_pre_turn_snapshot_ready(&mut exec_ctx, "post_turn_snapshot").await {
            warn!(
                "Failed to resolve pre-turn snapshot before post-turn processing: {}",
                e
            );
        }

        let turn_snapshot_data = runtime.turn_snapshot.lock().take();
        if let Some((turn_id, pre_snapshot_id)) = turn_snapshot_data {
            match backend.track(worktree).await {
                Ok(post_snapshot_id) => {
                    if pre_snapshot_id != post_snapshot_id {
                        match backend
                            .diff(worktree, &pre_snapshot_id, &post_snapshot_id)
                            .await
                        {
                            Ok(changed) if !changed.is_empty() => {
                                let patch_part = MessagePart::TurnSnapshotPatch {
                                    turn_id: turn_id.clone(),
                                    snapshot_id: post_snapshot_id.clone(),
                                    changed_paths: changed
                                        .iter()
                                        .map(|p| p.to_string_lossy().to_string())
                                        .collect(),
                                };
                                let snapshot_msg = AgentMessage {
                                    id: Uuid::new_v4().to_string(),
                                    session_id: session_id.clone(),
                                    role: ChatRole::Assistant,
                                    parts: vec![patch_part],
                                    created_at: time::OffsetDateTime::now_utc().unix_timestamp(),
                                    parent_message_id: None,
                                    source_provider: None,
                                    source_model: None,
                                };
                                if let Err(e) = exec_ctx.add_message(snapshot_msg).await {
                                    warn!("Failed to store turn snapshot patch: {}", e);
                                }
                            }
                            Ok(_) => {}
                            Err(e) => warn!("Failed to diff turn snapshots: {}", e),
                        }
                    }
                }
                Err(e) => warn!("Post-turn snapshot failed: {}", e),
            }
            let _ = backend.gc(worktree, &config.snapshot_gc_config).await;
        }
    } else {
        debug!(
            "Session {}: no snapshot backend, skipping post-turn snapshot",
            session_id
        );
    }

    debug!("Session {}: post-turn snapshot done", session_id);

    match result {
        Ok(CycleOutcome::Completed) => Ok(PromptResponse::new(StopReason::EndTurn)),
        Ok(CycleOutcome::Cancelled) => Ok(PromptResponse::new(StopReason::Cancelled)),
        Ok(CycleOutcome::Stopped(stop_reason)) => Ok(PromptResponse::new(stop_reason)),
        Err(e) => {
            config.emit_event(
                &session_id,
                AgentEventKind::Error {
                    message: e.to_string(),
                },
            );
            Err(AgentError::Internal(e.to_string()))
        }
    }
}

#[instrument(
    name = "agent.snapshot.pre_turn.ensure",
    skip(exec_ctx),
    fields(session_id = %exec_ctx.session_id, reason = reason)
)]
pub(crate) async fn ensure_pre_turn_snapshot_ready(
    exec_ctx: &mut ExecutionContext,
    reason: &'static str,
) -> Result<(), AgentError> {
    let pending_task = exec_ctx.runtime.pre_turn_snapshot_task.lock().take();
    let Some(task) = pending_task else {
        return Ok(());
    };

    let span = info_span!(
        "agent.snapshot.pre_turn.resolve",
        session_id = %exec_ctx.session_id,
        reason = reason
    );

    match task.instrument(span).await {
        Ok((turn_id, Ok(snapshot_id))) => {
            *exec_ctx.runtime.turn_snapshot.lock() = Some((turn_id.clone(), snapshot_id.clone()));

            let start_part = MessagePart::TurnSnapshotStart {
                turn_id,
                snapshot_id,
            };
            let snapshot_msg = AgentMessage {
                id: Uuid::new_v4().to_string(),
                session_id: exec_ctx.session_id.clone(),
                role: ChatRole::Assistant,
                parts: vec![start_part],
                created_at: time::OffsetDateTime::now_utc().unix_timestamp(),
                parent_message_id: None,
                source_provider: None,
                source_model: None,
            };
            exec_ctx.add_message(snapshot_msg).await.map_err(|e| {
                AgentError::Internal(format!("Failed to store turn snapshot start: {}", e))
            })?;
        }
        Ok((_turn_id, Err(err))) => {
            warn!("Pre-turn snapshot task finished with error: {}", err);
        }
        Err(join_err) => {
            warn!("Pre-turn snapshot task join failed: {}", join_err);
        }
    }

    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::agent::agent_config_builder::AgentConfigBuilder;
    use crate::agent::core::ToolPolicy;
    use crate::session::backend::StorageBackend;
    use crate::session::store::SessionStore;
    use crate::test_utils::{
        MockLlmProvider, MockSessionStore, SharedLlmProvider, TestProviderFactory, mock_llm_config,
        mock_plugin_registry, mock_session,
    };
    use kameo::actor::Spawn;
    use querymt::LLMParams;
    use std::collections::HashMap;
    use std::sync::Arc;
    use tokio::sync::Mutex;
    use tokio_util::sync::CancellationToken;

    // ── Shared fixture ───────────────────────────────────────────────────────

    struct ActorFixture {
        config: Arc<AgentConfig>,
        actor_ref: kameo::actor::ActorRef<SessionActor>,
        _session_id: String,
        _temp_dir: tempfile::TempDir,
    }

    impl ActorFixture {
        async fn new() -> Self {
            Self::with_session_id("test-session").await
        }

        async fn with_session_id(session_id: &str) -> Self {
            let provider = Arc::new(Mutex::new(MockLlmProvider::new()));
            let shared = SharedLlmProvider {
                inner: provider.clone(),
                tools: vec![].into_boxed_slice(),
            };
            let factory = Arc::new(TestProviderFactory { provider: shared });
            let (plugin_registry, temp_dir) =
                mock_plugin_registry(factory).expect("plugin registry");

            let llm_config = mock_llm_config();
            let session = mock_session(session_id);
            let mut store = MockSessionStore::new();
            let session_clone = session.clone();
            store
                .expect_get_session()
                .returning(move |_| Ok(Some(session_clone.clone())))
                .times(0..);
            let llm_for_mock = llm_config.clone();
            store
                .expect_get_session_llm_config()
                .returning(move |_| Ok(Some(llm_for_mock.clone())))
                .times(0..);
            store
                .expect_get_llm_config()
                .returning(move |_| Ok(Some(llm_config.clone())))
                .times(0..);
            store
                .expect_create_or_get_llm_config()
                .returning(|_| Ok(mock_llm_config()))
                .times(0..);
            store
                .expect_set_session_llm_config()
                .returning(|_, _| Ok(()))
                .times(0..);
            store
                .expect_set_session_provider_node_id()
                .returning(|_, _| Ok(()))
                .times(0..);
            store
                .expect_get_session_provider_node_id()
                .returning(|_| Ok(None))
                .times(0..);
            store
                .expect_list_sessions()
                .returning(|| Ok(vec![]))
                .times(0..);

            let store: Arc<dyn SessionStore> = Arc::new(store);
            let storage = Arc::new(
                crate::session::sqlite_storage::SqliteStorage::connect(":memory:".into())
                    .await
                    .expect("create event store"),
            );

            let config = Arc::new(
                AgentConfigBuilder::new(
                    Arc::new(plugin_registry),
                    store.clone(),
                    storage.event_journal(),
                    LLMParams::new().provider("mock").model("mock-model"),
                )
                .with_tool_policy(ToolPolicy::ProviderOnly)
                .build(),
            );

            let runtime = crate::agent::core::SessionRuntime::new(
                None,
                HashMap::new(),
                crate::agent::core::McpToolState::empty(),
            );
            let actor = SessionActor::new(config.clone(), session_id.to_string(), runtime);
            let actor_ref = SessionActor::spawn(actor);

            Self {
                config,
                actor_ref,
                _session_id: session_id.to_string(),
                _temp_dir: temp_dir,
            }
        }
    }

    // ── Actor message handler tests ──────────────────────────────────────────

    #[tokio::test]
    async fn test_set_and_get_mode() {
        let f = ActorFixture::new().await;

        // Default mode is Build
        let mode = f.actor_ref.ask(GetMode).await.expect("ask GetMode");
        assert_eq!(mode, AgentMode::Build);

        // Switch to Plan
        f.actor_ref
            .tell(SetMode {
                mode: AgentMode::Plan,
            })
            .await
            .expect("tell SetMode");
        // Give the actor time to process
        tokio::time::sleep(tokio::time::Duration::from_millis(10)).await;

        let mode = f.actor_ref.ask(GetMode).await.expect("ask GetMode");
        assert_eq!(mode, AgentMode::Plan);
    }

    #[tokio::test]
    async fn test_cancel_emits_event() {
        let f = ActorFixture::new().await;
        let mut rx = f.config.event_sink.fanout().subscribe();

        f.actor_ref.tell(Cancel).await.expect("tell Cancel");
        // Give the actor time to process
        tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;

        // Look for Cancelled event
        let mut found_cancel = false;
        while let Ok(envelope) = rx.try_recv() {
            if matches!(envelope.kind(), crate::events::AgentEventKind::Cancelled) {
                found_cancel = true;
            }
        }
        assert!(found_cancel, "Expected Cancelled event on event fanout");
    }

    #[tokio::test]
    async fn test_get_runtime_status_idle_when_not_running() {
        let f = ActorFixture::new().await;

        let status = f
            .actor_ref
            .ask(GetRuntimeStatus)
            .await
            .expect("ask GetRuntimeStatus");

        assert_eq!(status, SessionRuntimeStatus::Idle);
    }

    #[tokio::test]
    async fn test_get_runtime_status_remains_idle_after_cancel_without_running_prompt() {
        let f = ActorFixture::new().await;

        f.actor_ref.tell(Cancel).await.expect("tell Cancel");

        let status = f
            .actor_ref
            .ask(GetRuntimeStatus)
            .await
            .expect("ask GetRuntimeStatus");

        assert_eq!(status, SessionRuntimeStatus::Idle);
    }

    #[tokio::test]
    async fn test_get_runtime_status_cancel_requested_when_prompt_running_and_cancelled() {
        let f = ActorFixture::new().await;

        // Simulate prompt_running=true and token cancelled (the state after Cancel
        // is received while a prompt execution task is still in flight).
        f.actor_ref
            .tell(PromptFinished { generation: 0 })
            .await
            .expect("tell PromptFinished");
        tokio::time::sleep(tokio::time::Duration::from_millis(10)).await;

        // Now set prompt_running=true and cancel the token
        f.actor_ref.tell(Cancel).await.expect("tell Cancel");
        tokio::time::sleep(tokio::time::Duration::from_millis(10)).await;

        // We can't directly set prompt_running from outside, but we can verify
        // the status logic: when prompt_running=false and token is cancelled,
        // status should be Idle (not CancelRequested) because there's no running
        // prompt to cancel.
        let status = f
            .actor_ref
            .ask(GetRuntimeStatus)
            .await
            .expect("ask GetRuntimeStatus");

        // prompt_running is false (no prompt was sent), so status is Idle
        // even though the token was cancelled.
        assert_eq!(status, SessionRuntimeStatus::Idle);
    }

    #[tokio::test]
    async fn test_set_mode_emits_event() {
        let f = ActorFixture::new().await;
        let mut rx = f.config.event_sink.fanout().subscribe();

        f.actor_ref
            .tell(SetMode {
                mode: AgentMode::Plan,
            })
            .await
            .expect("tell SetMode");
        tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;

        let mut found = false;
        while let Ok(envelope) = rx.try_recv() {
            if let crate::events::AgentEventKind::SessionModeChanged { mode } = envelope.kind() {
                assert_eq!(*mode, AgentMode::Plan);
                found = true;
            }
        }
        assert!(found, "Expected SessionModeChanged event");
    }

    #[tokio::test]
    async fn test_set_provider_unknown_fails() {
        let f = ActorFixture::new().await;
        let result = f
            .actor_ref
            .ask(SetProvider {
                provider: "nonexistent-provider".to_string(),
                model: "some-model".to_string(),
            })
            .await;

        assert!(
            result.is_err(),
            "expected error for unknown provider, got ok"
        );
    }

    #[tokio::test]
    async fn test_set_tool_policy() {
        let f = ActorFixture::new().await;

        f.actor_ref
            .tell(SetToolPolicy {
                policy: ToolPolicy::BuiltInOnly,
            })
            .await
            .expect("tell SetToolPolicy");
        tokio::time::sleep(tokio::time::Duration::from_millis(10)).await;

        // Verify the policy was set by checking GetLlmConfig (indirect; we can't read tool_config directly)
        // Instead: setting allowed tools should persist without error
        f.actor_ref
            .tell(SetAllowedTools {
                tools: vec!["shell".to_string(), "read_tool".to_string()],
            })
            .await
            .expect("tell SetAllowedTools");
        tokio::time::sleep(tokio::time::Duration::from_millis(10)).await;

        // Clear should also succeed
        f.actor_ref
            .tell(ClearAllowedTools)
            .await
            .expect("tell ClearAllowedTools");
    }

    #[tokio::test]
    async fn test_set_denied_tools() {
        let f = ActorFixture::new().await;

        f.actor_ref
            .tell(SetDeniedTools {
                tools: vec!["shell".to_string()],
            })
            .await
            .expect("tell SetDeniedTools");
        tokio::time::sleep(tokio::time::Duration::from_millis(10)).await;

        f.actor_ref
            .tell(ClearDeniedTools)
            .await
            .expect("tell ClearDeniedTools");
    }

    #[tokio::test]
    async fn test_get_session_limits_returns_ok() {
        let f = ActorFixture::new().await;
        let limits = f
            .actor_ref
            .ask(GetSessionLimits)
            .await
            .expect("ask GetSessionLimits");
        // No limits middleware configured, so None is correct
        assert!(limits.is_none());
    }

    #[tokio::test]
    async fn test_get_llm_config() {
        let f = ActorFixture::new().await;
        let result = f
            .actor_ref
            .ask(GetLlmConfig)
            .await
            .expect("ask GetLlmConfig");
        // The mock store returns a config — None is fine (mock may not set it)
        // Just verify no error occurred (already asserted by .expect above)
        drop(result);
    }

    #[tokio::test]
    async fn test_set_reasoning_effort_allows_remote_only_provider() {
        let provider = Arc::new(Mutex::new(MockLlmProvider::new()));
        let shared = SharedLlmProvider {
            inner: provider,
            tools: vec![].into_boxed_slice(),
        };
        let factory = Arc::new(TestProviderFactory { provider: shared });
        let (plugin_registry, _temp_dir) = mock_plugin_registry(factory).expect("plugin registry");

        let session = mock_session("remote-session");
        let remote_config = crate::session::store::LLMConfig {
            id: 42,
            name: None,
            provider: "codex".to_string(),
            model: "gpt-5.4".to_string(),
            params: Some(serde_json::json!({
                "reasoning_effort": "medium"
            })),
            created_at: Some(time::OffsetDateTime::now_utc()),
            updated_at: Some(time::OffsetDateTime::now_utc()),
            provider_node_id: Some("12D3KooWRemoteNode".to_string()),
        };

        let mut store = MockSessionStore::new();
        let session_clone = session.clone();
        store
            .expect_get_session()
            .returning(move |_| Ok(Some(session_clone.clone())))
            .times(0..);
        let remote_config_for_get = remote_config.clone();
        store
            .expect_get_session_llm_config()
            .returning(move |_| Ok(Some(remote_config_for_get.clone())))
            .times(1..);
        let remote_config_for_id = remote_config.clone();
        store
            .expect_get_llm_config()
            .returning(move |_| Ok(Some(remote_config_for_id.clone())))
            .times(0..);
        store
            .expect_get_session_provider_node_id()
            .withf(|session_id| session_id == "remote-session")
            .returning(|_| Ok(Some("12D3KooWRemoteNode".to_string())))
            .times(1..);
        store
            .expect_create_or_get_llm_config()
            .withf(|config| {
                config.provider.as_deref() == Some("codex")
                    && config.model.as_deref() == Some("gpt-5.4")
                    && config.reasoning_effort == Some(ReasoningEffort::High)
            })
            .returning(|_| {
                Ok(crate::session::store::LLMConfig {
                    id: 43,
                    name: None,
                    provider: "codex".to_string(),
                    model: "gpt-5.4".to_string(),
                    params: Some(serde_json::json!({"reasoning_effort": "high"})),
                    created_at: Some(time::OffsetDateTime::now_utc()),
                    updated_at: Some(time::OffsetDateTime::now_utc()),
                    provider_node_id: Some("12D3KooWRemoteNode".to_string()),
                })
            })
            .times(1);
        store
            .expect_set_session_llm_config()
            .withf(|session_id, config_id| session_id == "remote-session" && *config_id == 43)
            .returning(|_, _| Ok(()))
            .times(1);
        store
            .expect_set_session_provider_node_id()
            .returning(|_, _| Ok(()))
            .times(0..);
        store
            .expect_list_sessions()
            .returning(|| Ok(vec![]))
            .times(0..);

        let store: Arc<dyn SessionStore> = Arc::new(store);
        let storage = Arc::new(
            crate::session::sqlite_storage::SqliteStorage::connect(":memory:".into())
                .await
                .expect("create event store"),
        );

        let config = Arc::new(
            AgentConfigBuilder::new(
                Arc::new(plugin_registry),
                store,
                storage.event_journal(),
                LLMParams::new().provider("mock").model("mock-model"),
            )
            .with_tool_policy(ToolPolicy::ProviderOnly)
            .build(),
        );

        let runtime = crate::agent::core::SessionRuntime::new(
            None,
            HashMap::new(),
            crate::agent::core::McpToolState::empty(),
        );
        let actor = SessionActor::new(config, "remote-session".to_string(), runtime);
        let actor_ref = SessionActor::spawn(actor);

        actor_ref
            .ask(SetReasoningEffort {
                effort: Some(ReasoningEffort::High),
            })
            .await
            .expect("remote session should accept reasoning effort update");
    }

    #[tokio::test]
    async fn test_ext_method_returns_null() {
        let f = ActorFixture::new().await;
        let null_params = std::sync::Arc::from(
            serde_json::value::RawValue::from_string("null".to_string()).unwrap(),
        );
        let result = f
            .actor_ref
            .ask(ExtMethod {
                req: agent_client_protocol::schema::ExtRequest::new("custom_method", null_params),
            })
            .await
            .expect("ask ExtMethod");
        // The default implementation returns a null JSON value
        assert_eq!(result.0.get(), "null");
    }

    #[tokio::test]
    async fn test_ext_notification_succeeds() {
        let f = ActorFixture::new().await;
        let null_params = std::sync::Arc::from(
            serde_json::value::RawValue::from_string("null".to_string()).unwrap(),
        );
        f.actor_ref
            .ask(ExtNotification {
                notif: agent_client_protocol::schema::ExtNotification::new(
                    "my_notification",
                    null_params,
                ),
            })
            .await
            .expect("ask ExtNotification");
    }

    #[tokio::test]
    async fn test_subscribe_unsubscribe_events_no_panic() {
        let f = ActorFixture::new().await;
        // Without remote feature, these are no-ops that return Ok(())
        f.actor_ref
            .ask(crate::agent::messages::SubscribeEvents {
                relay_actor_id: 42,
                relay_dht_name: "event_relay::test::peer-X".to_string(),
            })
            .await
            .expect("ask SubscribeEvents");

        f.actor_ref
            .ask(crate::agent::messages::UnsubscribeEvents {
                relay_actor_id: 42,
                relay_dht_name: "event_relay::test::peer-X".to_string(),
            })
            .await
            .expect("ask UnsubscribeEvents");
    }

    #[tokio::test]
    async fn test_prompt_finished_resets_state() {
        let f = ActorFixture::new().await;
        // Send PromptFinished — even when not "running", should be a no-op that doesn't panic
        f.actor_ref
            .tell(PromptFinished { generation: 0 })
            .await
            .expect("tell PromptFinished");
        tokio::time::sleep(tokio::time::Duration::from_millis(10)).await;

        // GetMode should still work, confirming actor is alive
        let mode = f.actor_ref.ask(GetMode).await.expect("ask GetMode");
        assert_eq!(mode, AgentMode::Build);
    }

    #[tokio::test]
    async fn test_set_bridge_succeeds() {
        let f = ActorFixture::new().await;
        // We can't easily create a real ClientBridgeSender in unit tests,
        // but setting None→Some via SetBridge message should not panic or fail.
        // We verify the actor is alive by asking GetMode after.
        // (SetBridge requires a real sender; skip the actual send but confirm actor health)
        let mode = f
            .actor_ref
            .ask(GetMode)
            .await
            .expect("ask GetMode after potential bridge set");
        assert_eq!(mode, AgentMode::Build);
    }

    #[tokio::test]
    async fn test_shutdown_stops_actor() {
        let f = ActorFixture::new().await;
        // Shutdown the actor
        f.actor_ref.tell(Shutdown).await.expect("tell Shutdown");
        // Give the actor time to stop
        tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;
        // After shutdown, the actor should no longer accept messages
        let result = f.actor_ref.ask(GetMode).await;
        assert!(result.is_err(), "actor should be stopped after Shutdown");
    }

    #[tokio::test]
    async fn test_multiple_sessions_independent() {
        let f1 = ActorFixture::with_session_id("session-alpha").await;
        let f2 = ActorFixture::with_session_id("session-beta").await;

        f1.actor_ref
            .tell(SetMode {
                mode: AgentMode::Plan,
            })
            .await
            .expect("tell SetMode on f1");
        tokio::time::sleep(tokio::time::Duration::from_millis(10)).await;

        let mode1 = f1.actor_ref.ask(GetMode).await.expect("GetMode f1");
        let mode2 = f2.actor_ref.ask(GetMode).await.expect("GetMode f2");

        assert_eq!(mode1, AgentMode::Plan);
        assert_eq!(
            mode2,
            AgentMode::Build,
            "f2 should not be affected by f1's mode change"
        );
    }

    // ── Token state machine tests ─────────────────────────────────────────────

    #[test]
    fn parse_transport_model_id_splits_on_first_slash() {
        let (provider, model) = parse_transport_model_id(
            "llama_cpp/hf:unsloth/Qwen3.5-4B-GGUF:Q5_K_M.gguf",
            "anthropic",
        );
        assert_eq!(provider, "llama_cpp");
        assert_eq!(model, "hf:unsloth/Qwen3.5-4B-GGUF:Q5_K_M.gguf");

        let (provider, model) =
            parse_transport_model_id("openrouter/openai/gpt-5.3-codex", "anthropic");
        assert_eq!(provider, "openrouter");
        assert_eq!(model, "openai/gpt-5.3-codex");
    }

    #[test]
    fn parse_transport_model_id_uses_fallback_for_legacy_bare_ids() {
        let (provider, model) = parse_transport_model_id("claude-opus-4-6", "anthropic");
        assert_eq!(provider, "anthropic");
        assert_eq!(model, "claude-opus-4-6");
    }

    #[test]
    fn stale_prompt_finished_generation_is_ignored() {
        let mut turn_state = TurnState::new();
        turn_state.generation = 4;

        let running_token = turn_state.token.clone();
        let stale_generation = 3;

        // Simulate PromptFinished for an older prompt while generation 4 is current.
        if stale_generation == turn_state.generation {
            turn_state.token = CancellationToken::new();
        }

        turn_state.token.cancel();
        assert!(
            running_token.is_cancelled(),
            "Stale PromptFinished must not replace the current turn token"
        );
    }

    /// Simulates the cancel-then-resume scenario:
    ///
    /// 1. Prompt starts       → prompt_running=true, token shared with running task
    /// 2. Cancel arrives      → token cancelled (reaches the running task)
    /// 3. New prompt arrives  → token.is_cancelled() → fresh token created for new turn
    /// 4. Old task finishes   → PromptFinished resets prompt_running + token
    ///
    /// The new prompt's detached task waits on the execution_permit semaphore
    /// (held by the old task) so no two prompts run concurrently.
    #[test]
    fn cancel_flag_allows_new_prompt_while_old_task_winds_down() {
        let mut cancel_token = CancellationToken::new();

        // Step 1: Prompt starts — give the running task a clone of the token.
        let mut prompt_running = true;
        let _running_task_token = cancel_token.clone(); // simulates the spawned task's handle
        assert!(prompt_running);
        assert!(!cancel_token.is_cancelled());

        // Step 2: Cancel arrives — cancels the shared token, which the running task holds.
        cancel_token.cancel();
        assert!(cancel_token.is_cancelled());
        assert!(
            _running_task_token.is_cancelled(),
            "Cancel must reach the running task's token clone"
        );
        // prompt_running is still true because PromptFinished hasn't arrived yet.

        // Step 3: New prompt arrives while old task winds down.
        // The Prompt handler creates a fresh token only because the previous one is cancelled.
        let allow_new_prompt = if prompt_running {
            cancel_token.is_cancelled() // cancelled → allow new prompt
        } else {
            true
        };
        assert!(
            allow_new_prompt,
            "A cancelled session must accept new prompts even before PromptFinished arrives"
        );

        if cancel_token.is_cancelled() {
            cancel_token = CancellationToken::new();
        }
        assert!(
            !cancel_token.is_cancelled(),
            "New prompt's cancel_token should start clean"
        );

        // Step 4: PromptFinished arrives — resets prompt_running.
        // In the real actor the token is also reset here, but for this
        // scenario it was already reset in Step 3 (cancelled path).
        prompt_running = false;
        assert!(!prompt_running);
        assert!(!cancel_token.is_cancelled());
    }

    /// Verifies the non-cancelled path: a second Prompt arriving while the first
    /// task is still running must NOT replace the cancel token, so Cancel still
    /// reaches the first task.
    #[test]
    fn cancel_reaches_running_task_when_second_prompt_queued() {
        let mut cancel_token = CancellationToken::new();

        // Step 1: First prompt starts.
        let running_task_token = cancel_token.clone();
        assert!(!cancel_token.is_cancelled());

        // Step 2: Second prompt arrives while first is still running.
        // Because the token is NOT cancelled, the Prompt handler keeps the existing token.
        let should_replace = cancel_token.is_cancelled();
        assert!(
            !should_replace,
            "Must not replace token while task is alive"
        );
        // token unchanged — running task still holds the same token.

        // Step 3: Cancel arrives.
        cancel_token.cancel();
        assert!(
            running_task_token.is_cancelled(),
            "Cancel must reach the first task even after a second Prompt was queued"
        );

        // Step 4: PromptFinished — reset token for the next clean turn.
        cancel_token = CancellationToken::new();
        assert!(
            !cancel_token.is_cancelled(),
            "Token reset after PromptFinished"
        );
    }

    #[test]
    fn summarize_intent_candidate_ignores_empty_and_attachment_placeholder() {
        assert_eq!(summarize_intent_candidate("   \n  "), None);
        assert_eq!(
            summarize_intent_candidate(ATTACHMENTS_PLACEHOLDER),
            None,
            "attachment-only prompts should not become intent"
        );
    }

    #[test]
    fn decide_intent_update_respects_low_signal_clarifications() {
        let decision = decide_intent_update("Build a Rust CLI to parse logs", "can you fix this?");
        assert_eq!(
            decision,
            IntentUpdateDecision::NoChange,
            "short clarification prompts should not rewrite session objective"
        );
    }

    #[test]
    fn decide_intent_update_auto_updates_on_explicit_objective_marker() {
        let decision = decide_intent_update(
            "Build a Rust CLI to parse logs",
            "objective: migrate this to a TypeScript dashboard",
        );
        assert_eq!(decision, IntentUpdateDecision::Update);
    }

    #[test]
    fn decide_intent_update_asks_on_explicit_shift_language() {
        let decision = decide_intent_update(
            "Build a Rust CLI to parse logs",
            "Instead, let's switch to building a web UI for analytics.",
        );
        assert_eq!(decision, IntentUpdateDecision::AskUser);
    }

    #[test]
    fn summarize_intent_candidate_ukrainian_no_panic() {
        // Regression test for issue #525: byte 237 landed inside 'ч' (2-byte char).
        let ukrainian = "зроби коміт без gpg, до цього стану та почни виправленя. \
                         А ще, зараз проблема з подвійним натисканням шифт + клік, \
                         щоб переместити річ у сховище";
        let result = summarize_intent_candidate(ukrainian).unwrap();
        assert!(result.len() <= 240);
        assert!(result.is_char_boundary(result.len()));
    }

    #[test]
    fn summarize_intent_candidate_cjk_no_panic() {
        let cjk = "你好".repeat(200); // 2 chars * 200 = 400 chars, each 3 bytes = 1200 bytes
        let result = summarize_intent_candidate(&cjk).unwrap();
        assert!(result.len() <= 240);
        assert!(result.is_char_boundary(result.len()));
    }
}
