mnote/rust/crates/mnote-web/src/acp_session_manager.rs

/// ACP Session Manager — session lifecycle management.
///
/// Wraps an [`AcpClient`] to provide strongly-typed session operations:
/// create, prompt, cancel, and receive typed events from the agent.
///
/// Reference:
/// - `reference-code/hermes-vscode-main/src/sessionManager.ts` (primary)
/// - `reference-code/hermes-vscode-main/src/protocol.ts` (dedup logic)
use crate::acp_client::AcpClient;
use crate::acp_types::{
    ContentBlock, ContentBlockWrapper, SessionLoadParams, SessionNewParams, SessionNewResult,
    SessionPromptParams, SessionPromptResult, SessionUpdate, ToolCallStatus,
};
use serde_json::{json, Value};
use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use std::time::Duration;
use tokio::time::sleep;
use tracing::{debug, info, warn};

// ── Events ───────────────────────────────────────────

/// Strongly-typed event emitted by the session manager when a `session/update` arrives.
#[derive(Debug, Clone)]
pub enum AcpSessionEvent {
    /// Streaming text from the agent's response message.
    TextDelta { text: String },
    /// Streaming reasoning/thinking text.
    ThoughtDelta { text: String },
    /// 工具调用开始。
    ToolCall {
        tool_call_id: String,
        title: String,
        kind: String,
        status: ToolCallStatus,
        raw_input: Option<Value>,
        /// 工具涉及的文件路径。
        locations: Vec<String>,
    },
    /// 工具调用状态更新，可能携带结果内容。
    ToolCallUpdate {
        tool_call_id: String,
        status: ToolCallStatus,
        content: Option<Vec<ContentBlockWrapper>>,
    },
    /// 上下文用量更新。
    UsageUpdate { used: u64, size: u64 },
    /// Agent 发起权限请求。
    PermissionRequest {
        permission_id: String,
        tool_name: String,
        params: Value,
        decision: String,
    },
    /// 会话元数据更新，例如自动标题。
    SessionInfoUpdate { title: String },
    /// 计划条目更新。
    PlanUpdate { entries: Vec<String> },
    /// 连接关闭或异常。
    Disconnected { reason: String },
}

#[derive(Debug, Clone, Default)]
pub struct AcpMnoteToolContext {
    pub mnote_session_id: Option<String>,
    pub run_id: Option<String>,
    pub actor_id: Option<String>,
    pub trace_id: Option<String>,
    pub workspace_id: Option<String>,
    pub document_id: Option<String>,
}

/// Handler for session events.
pub type SessionEventHandler = Arc<dyn Fn(AcpSessionEvent) + Send + Sync + 'static>;

// ── Session state ────────────────────────────────────

#[derive(Debug, Clone, PartialEq)]
pub enum SessionState {
    Idle,
    Running,
    Cancelling,
    Closed,
}

// ── Pending permission ───────────────────────────────

/// 等待前端决策的 `session/request_permission`。
#[derive(Debug, Clone)]
pub struct PendingPermission {
    /// incoming ACP request 的原始 JSON-RPC id，可能是数字或字符串。
    pub jsonrpc_id: Value,
    /// `session/request_permission` 的原始参数。
    pub params: Value,
    /// 请求创建时间，用于后续超时判断。
    pub created_at: std::time::Instant,
}

// ── AcpSessionManager ────────────────────────────────

/// Manages ACP sessions — create, prompt, cancel, and event dispatch.
///
/// Currently supports one active session at a time. The internal [`AcpClient`]
/// handles the JSON-RPC wire protocol; this layer adds session semantics and
/// typed event dispatching.
pub struct AcpSessionManager {
    client: Arc<AcpClient>,
    session_id: Arc<Mutex<Option<String>>>,
    state: Arc<Mutex<SessionState>>,
    event_handler: Arc<Mutex<Option<SessionEventHandler>>>,
    /// Accumulated text for deduplication (per-turn).
    accumulated: Arc<Mutex<String>>,
    /// Whether we're currently inside a prompt (for dedup gating).
    in_prompt: Arc<Mutex<bool>>,
    /// Pending permission requests keyed by permission_id.
    pending_permissions: Arc<Mutex<HashMap<String, PendingPermission>>>,
}

impl AcpSessionManager {
    /// Create a new session manager wrapping an existing [`AcpClient`].
    ///
    /// Registers an internal notification handler that dispatches
    /// `session/update` notifications as typed [`AcpSessionEvent`]s.
    pub fn new(client: Arc<AcpClient>) -> Self {
        let session_id: Arc<Mutex<Option<String>>> = Arc::new(Mutex::new(None));
        let state: Arc<Mutex<SessionState>> = Arc::new(Mutex::new(SessionState::Idle));
        let event_handler: Arc<Mutex<Option<SessionEventHandler>>> = Arc::new(Mutex::new(None));
        let accumulated: Arc<Mutex<String>> = Arc::new(Mutex::new(String::new()));
        let in_prompt: Arc<Mutex<bool>> = Arc::new(Mutex::new(false));
        let pending_permissions: Arc<Mutex<HashMap<String, PendingPermission>>> =
            Arc::new(Mutex::new(HashMap::new()));

        // ── Incoming request handler ──────────────────
        // 处理 `session/request_permission`：先进入 pending，再由 HTTP
        // resolve-permission 端点决定最终响应。
        let event_handler_for_incoming = event_handler.clone();
        let pending_for_incoming = pending_permissions.clone();
        let client_for_incoming = client.clone();
        client.on_incoming_request(move |id, method, params| {
            if method != "session/request_permission" {
                // 未知方法交回 dispatch_message 回复 method-not-found。
                return false;
            }

            // 优先使用 agent 传来的 permission id；缺失时用 JSON-RPC id 派生稳定 id。
            let permission_id = params
                .get("permissionId")
                .or_else(|| params.get("permission_id"))
                .or_else(|| params.get("requestId"))
                .and_then(Value::as_str)
                .filter(|s| !s.is_empty())
                .map(ToOwned::to_owned)
                .unwrap_or_else(|| format!("acp_perm_{}", id));

            let tool_name = params
                .get("toolName")
                .or_else(|| params.get("tool"))
                .or_else(|| params.get("name"))
                .or_else(|| params.get("method"))
                .and_then(Value::as_str)
                .unwrap_or("session/request_permission")
                .to_string();

            // 记录到 pending map，等待前端 allow / deny。
            {
                let mut pending = pending_for_incoming.lock().unwrap();
                pending.insert(
                    permission_id.clone(),
                    PendingPermission {
                        jsonrpc_id: id.clone(),
                        params: params.clone(),
                        created_at: std::time::Instant::now(),
                    },
                );
                debug!(
                    "ACP permission pending: {} (jsonrpc_id={}, tool={})",
                    permission_id, id, tool_name
                );
            }

            // 向前端发出 `permission.requested`。
            {
                let handler = event_handler_for_incoming.lock().unwrap();
                if let Some(ref h) = *handler {
                    h(AcpSessionEvent::PermissionRequest {
                        permission_id: permission_id.clone(),
                        tool_name,
                        params,
                        decision: "requested".into(),
                    });
                }
            }

            // 60 秒没有决策时自动 deny，避免 agent 永久等待。
            let pending_for_timeout = pending_for_incoming.clone();
            let event_handler_for_timeout = event_handler_for_incoming.clone();
            let client_for_timeout = client_for_incoming.clone();
            let timeout_permission_id = permission_id.clone();
            let timeout_jsonrpc_id = id.clone();
            tokio::spawn(async move {
                sleep(Duration::from_secs(60)).await;
                let should_deny = {
                    let mut pending = pending_for_timeout.lock().unwrap();
                    if pending.remove(&timeout_permission_id).is_some() {
                        true
                    } else {
                        false
                    }
                };
                if should_deny {
                    warn!(
                        "ACP permission timeout: {} (jsonrpc_id={}), auto-denying",
                        timeout_permission_id, timeout_jsonrpc_id
                    );
                    // 向 ACP 回复 permission denied。
                    let _ = client_for_timeout
                        .respond_to_incoming_error(
                            timeout_jsonrpc_id.clone(),
                            -32000,
                            "permission denied by timeout",
                        )
                        .await;
                    // 同步 denied 事件给前端。
                    let handler = event_handler_for_timeout.lock().unwrap();
                    if let Some(ref h) = *handler {
                        h(AcpSessionEvent::PermissionRequest {
                            permission_id: timeout_permission_id,
                            tool_name: "session/request_permission".into(),
                            params: Value::Null,
                            decision: "denied".into(),
                        });
                    }
                }
            });

            true // handler took responsibility for responding
        });

        // ── Notification handler ──────────────────────
        // Only handles `session/update` (streaming events from agent).
        // Incoming request methods like `session/request_permission` are handled
        // above via `on_incoming_request`.
        let session_id_clone = session_id.clone();
        let event_handler_clone = event_handler.clone();
        let accumulated_clone = accumulated.clone();
        let in_prompt_clone = in_prompt.clone();

        client.on_notification(move |method, params| {
            if method != "session/update" {
                return;
            }
            let sid = session_id_clone.lock().unwrap().clone();
            if let Some(ref session_id) = sid {
                if let Some(msg_sid) = params.get("sessionId").and_then(|v| v.as_str()) {
                    if msg_sid != session_id {
                        return; // not our session
                    }
                }
            }

            // Parse the update
            let update: SessionUpdate = match serde_json::from_value(
                params.get("update").cloned().unwrap_or(Value::Null),
            ) {
                Ok(u) => u,
                Err(e) => {
                    warn!("ACP session/update parse error: {e}");
                    return;
                }
            };

            let is_in_prompt = *in_prompt_clone.lock().unwrap();
            let event = Self::session_update_to_event(&update, &accumulated_clone, is_in_prompt);

            if let Some(ev) = event {
                let handler = event_handler_clone.lock().unwrap();
                if let Some(ref h) = *handler {
                    h(ev);
                }
            }
        });

        Self {
            client,
            session_id,
            state,
            event_handler,
            accumulated,
            in_prompt,
            pending_permissions,
        }
    }

    /// Register an event handler for session events.
    /// Only one handler at a time — subsequent calls replace the previous.
    pub fn on_event<F>(&self, handler: F)
    where
        F: Fn(AcpSessionEvent) + Send + Sync + 'static,
    {
        let mut guard = self.event_handler.lock().unwrap();
        *guard = Some(Arc::new(handler));
    }

    /// 按用户决策解析 pending permission。
    ///
    /// 根据 `permission_id` 找到 pending request，向 ACP 子进程回写 result/error，
    /// 发出 `permission.allowed` / `permission.denied` 事件，并从 pending map 移除。
    ///
    /// permission 存在且完成响应时返回 `Ok(())`，否则返回错误原因。
    pub async fn resolve_permission(
        &self,
        permission_id: &str,
        decision: &str,
    ) -> Result<(), String> {
        let pending = {
            let mut map = self.pending_permissions.lock().unwrap();
            map.remove(permission_id)
        };
        let pending = pending.ok_or_else(|| {
            format!("pending permission not found: {permission_id} (already timed out or invalid)")
        })?;

        let tool_name = pending
            .params
            .get("toolName")
            .or_else(|| pending.params.get("tool"))
            .or_else(|| pending.params.get("name"))
            .and_then(Value::as_str)
            .unwrap_or("session/request_permission")
            .to_string();

        let normalized_decision = match decision {
            "allow" | "allowed" => "allow",
            "deny" | "denied" => "deny",
            other => return Err(format!("unknown decision: {other} (expected allow/deny)")),
        };

        let response = permission_response_for_decision(&pending.params, normalized_decision)
            .unwrap_or_else(|| {
                if normalized_decision == "allow" {
                    Err((-32000, "permission allow option not available".into()))
                } else {
                    Err((-32000, "permission denied by user".into()))
                }
            });

        match response {
            Ok(result) => {
                self.client
                    .respond_to_incoming(pending.jsonrpc_id.clone(), result)
                    .await
                    .map_err(|e| format!("ACP respond failed: {e}"))?;
                info!(
                    "ACP permission resolved: {} decision={} (jsonrpc_id={})",
                    permission_id, normalized_decision, pending.jsonrpc_id
                );
            }
            Err((code, message)) => {
                self.client
                    .respond_to_incoming_error(pending.jsonrpc_id.clone(), code, &message)
                    .await
                    .map_err(|e| format!("ACP respond failed: {e}"))?;
                info!(
                    "ACP permission rejected: {} decision={} (jsonrpc_id={})",
                    permission_id, normalized_decision, pending.jsonrpc_id
                );
            }
        }

        // 发出最终 decision 事件。
        {
            let handler = self.event_handler.lock().unwrap();
            if let Some(ref h) = *handler {
                h(AcpSessionEvent::PermissionRequest {
                    permission_id: permission_id.to_string(),
                    tool_name,
                    params: pending.params,
                    decision: if normalized_decision == "allow" {
                        "allowed".into()
                    } else {
                        "denied".into()
                    },
                });
            }
        }

        Ok(())
    }

    /// Create a new ACP session.
    ///
    /// Sends `session/new` to the agent and stores the returned `sessionId`.
    /// The `page_context` is optional metadata about the current document.
    pub async fn create_session(
        &self,
        cwd: Option<&str>,
        _page_context: Option<Value>,
    ) -> Result<String, crate::acp_client::AcpError> {
        let project_root = std::path::Path::new(env!("CARGO_MANIFEST_DIR"))
            .ancestors()
            .nth(3)
            .unwrap_or(std::path::Path::new("/mnt/Data1T/mnote"))
            .to_string_lossy()
            .to_string();
        let params = SessionNewParams {
            cwd: Some(cwd.map(str::to_string).unwrap_or(project_root)),
            mcp_servers: Some(Vec::new()),
        };

        let result: SessionNewResult = self.client.request("session/new", params).await?;
        let mut sid_guard = self.session_id.lock().unwrap();
        *sid_guard = Some(result.session_id.clone());
        info!("ACP session created: {}", result.session_id);

        Ok(result.session_id)
    }

    /// Load/resume an existing ACP session.
    ///
    /// Sends `session/load { sessionId, cwd, mcpServers: [] }` to the agent.
    /// Returns `true` if the session was loaded and the internal session_id updated,
    /// `false` if the adapter returned null (session not found) or doesn't support the method.
    ///
    /// On success the internal `session_id` is set to the loaded id.
    ///
    /// Reference: `hermes-vscode-main/src/sessionManager.ts` `ensureSession()`
    pub async fn load_session(
        &self,
        session_id: &str,
        cwd: Option<&str>,
    ) -> Result<bool, crate::acp_client::AcpError> {
        let project_root = std::path::Path::new(env!("CARGO_MANIFEST_DIR"))
            .ancestors()
            .nth(3)
            .unwrap_or(std::path::Path::new("/mnt/Data1T/mnote"))
            .to_string_lossy()
            .to_string();
        let params = SessionLoadParams {
            session_id: session_id.to_string(),
            cwd: Some(cwd.map(str::to_string).unwrap_or(project_root)),
            mcp_servers: Some(Vec::new()),
        };

        match self
            .client
            .request::<_, serde_json::Value>("session/load", params)
            .await
        {
            Ok(result) => {
                // Successful load: adapter returned { sessionId: "..." }
                if let Some(sid) = result.get("sessionId").and_then(|v| v.as_str()) {
                    if !sid.is_empty() {
                        let mut sid_guard = self.session_id.lock().unwrap();
                        *sid_guard = Some(sid.to_string());
                        info!("ACP session loaded: {}", sid);
                        return Ok(true);
                    }
                }
                // Null or missing sessionId → session not found on adapter
                info!(
                    "ACP session/load returned null for session_id={}",
                    session_id
                );
                Ok(false)
            }
            Err(crate::acp_client::AcpError::JsonRpc { code, message }) => {
                // Adapter doesn't support session/load or session expired
                info!(
                    "ACP session/load not supported (code={code}, message={message}), will fallback"
                );
                Ok(false)
            }
            Err(e) => {
                warn!("ACP session/load transport error: {e}");
                Err(e)
            }
        }
    }

    /// Ensure an active ACP session exists.
    ///
    /// 1. If there's already an active session, return it immediately.
    /// 2. If a `stored_session_id` is provided, attempt `session/load` first.
    /// 3. Fall back to `session/new`.
    ///
    /// Reference: `hermes-vscode-main/src/sessionManager.ts` `ensureSession()`
    pub async fn ensure_session(
        &self,
        cwd: Option<&str>,
        stored_session_id: Option<&str>,
    ) -> Result<String, crate::acp_client::AcpError> {
        // 1. Reuse active session if one exists
        {
            let sid = self.session_id.lock().unwrap();
            if let Some(ref sid) = *sid {
                debug!("ACP ensure_session: reusing active session {}", sid);
                return Ok(sid.clone());
            }
        }

        // 2. Try to load a stored session
        if let Some(stored) = stored_session_id {
            if !stored.is_empty() {
                debug!("ACP ensure_session: attempting session/load for {}", stored);
                if self.load_session(stored, cwd).await.unwrap_or(false) {
                    return Ok(stored.to_string());
                }
                info!(
                    "ACP ensure_session: stored session {} not found, creating new",
                    stored
                );
            }
        }

        // 3. Fallback: create new session
        debug!("ACP ensure_session: creating new session");
        self.create_session(cwd, None).await
    }

    /// Send a prompt to the agent and stream events.
    ///
    /// The `prompt` is a list of content blocks (text + optional page context).
    /// Returns once the agent finishes (stopReason received) or on error.
    ///
    /// Sets `in_prompt = true` during the call to enable deduplication,
    /// then resets to `false` and clears accumulated text on completion.
    pub async fn run_prompt(
        &self,
        prompt: Vec<ContentBlock>,
    ) -> Result<SessionPromptResult, crate::acp_client::AcpError> {
        self.run_prompt_with_mnote_context(prompt, None).await
    }

    pub async fn run_prompt_with_mnote_context(
        &self,
        prompt: Vec<ContentBlock>,
        mnote_context: Option<AcpMnoteToolContext>,
    ) -> Result<SessionPromptResult, crate::acp_client::AcpError> {
        {
            let mut in_prompt = self.in_prompt.lock().unwrap();
            *in_prompt = true;
        }
        {
            let mut acc = self.accumulated.lock().unwrap();
            acc.clear();
        }
        {
            let mut state = self.state.lock().unwrap();
            *state = SessionState::Running;
        }

        let sid = self.session_id.lock().unwrap().clone();
        let session_id = sid.ok_or_else(|| {
            self.reset_prompt_state();
            crate::acp_client::AcpError::Internal(
                "no session created yet — call create_session first".into(),
            )
        })?;

        let mnote_context = mnote_context.unwrap_or_default();
        let params = SessionPromptParams {
            session_id: session_id.clone(),
            prompt,
            mnote_session_id: mnote_context.mnote_session_id,
            run_id: mnote_context.run_id,
            actor_id: mnote_context.actor_id,
            trace_id: mnote_context.trace_id,
            workspace_id: mnote_context.workspace_id,
            document_id: mnote_context.document_id,
        };

        debug!("ACP session/prompt (session={})", session_id);
        let result: Result<SessionPromptResult, crate::acp_client::AcpError> =
            self.client.request("session/prompt", params).await;
        if let Ok(ref prompt_result) = result {
            debug!(
                "ACP session/prompt done (session={}, stop_reason={:?})",
                session_id, prompt_result.stop_reason
            );
        }
        self.reset_prompt_state();

        result
    }

    /// Cancel the current prompt.
    ///
    /// Sends `session/cancel` notification to the agent.
    pub async fn cancel(&self) -> Result<(), crate::acp_client::AcpError> {
        let sid = self.session_id.lock().unwrap().clone();
        let session_id =
            sid.ok_or_else(|| crate::acp_client::AcpError::Internal("no active session".into()))?;

        {
            let mut state = self.state.lock().unwrap();
            *state = SessionState::Cancelling;
        }

        self.client
            .notification("session/cancel", json!({ "sessionId": session_id }))
            .await?;

        info!("ACP session cancelled: {}", session_id);
        Ok(())
    }

    /// Close the session and the underlying ACP client.
    pub async fn close(&self) {
        {
            let mut state = self.state.lock().unwrap();
            *state = SessionState::Closed;
        }

        if let Some(handler) = self.event_handler.lock().unwrap().take() {
            handler(AcpSessionEvent::Disconnected {
                reason: "session closed".into(),
            });
        }
    }

    /// Get the current session ID, if any.
    pub async fn session_id(&self) -> Option<String> {
        self.session_id.lock().unwrap().clone()
    }

    /// Get the current session state.
    pub async fn state(&self) -> SessionState {
        self.state.lock().unwrap().clone()
    }

    fn reset_prompt_state(&self) {
        {
            let mut state = self.state.lock().unwrap();
            *state = SessionState::Idle;
        }
        {
            let mut in_prompt = self.in_prompt.lock().unwrap();
            *in_prompt = false;
        }
        {
            let mut acc = self.accumulated.lock().unwrap();
            acc.clear();
        }
    }

    // ── Internal: session/update → event mapping ─────

    /// Convert a parsed [`SessionUpdate`] into an [`AcpSessionEvent`],
    /// applying text deduplication for streaming text.
    ///
    /// Reference: `hermes-vscode-main/src/protocol.ts` `extractTextContent()`,
    /// `deduplicateChunk()`, `parseToolCall()`, `parseToolCallUpdate()`,
    /// `parseUsageUpdate()`, `parseSessionInfoUpdate()`
    fn session_update_to_event(
        update: &SessionUpdate,
        accumulated: &Arc<Mutex<String>>,
        is_in_prompt: bool,
    ) -> Option<AcpSessionEvent> {
        match update {
            SessionUpdate::AgentMessageChunk { content, .. }
            | SessionUpdate::AgentThoughtChunk { content, .. } => {
                let discrim = match update {
                    SessionUpdate::AgentMessageChunk { .. } => "msg",
                    SessionUpdate::AgentThoughtChunk { .. } => "thought",
                    _ => unreachable!(),
                };

                // Deduplication (reference: protocol.ts deduplicateChunk)
                if is_in_prompt {
                    let mut acc = accumulated.lock().unwrap();
                    let text = &content.text;
                    let event = if text == acc.as_str() {
                        // Exact full resend → drop
                        None
                    } else if text.len() > 10 && text.starts_with(acc.as_str()) {
                        // Superset resend → emit only the tail
                        let new_part = text[acc.len()..].to_string();
                        if new_part.is_empty() {
                            None
                        } else {
                            *acc = text.clone();
                            Some(new_part)
                        }
                    } else if text.len() > 10 && acc.ends_with(text) {
                        // Partial resend → drop
                        None
                    } else {
                        // Normal delta
                        let new_acc = format!("{}{}", acc, text);
                        *acc = new_acc;
                        Some(text.clone())
                    };

                    return event.map(|t| match discrim {
                        "msg" => AcpSessionEvent::TextDelta { text: t },
                        "thought" => AcpSessionEvent::ThoughtDelta { text: t },
                        _ => unreachable!(),
                    });
                }

                // Not in prompt — emit directly (historical playback)
                let text = content.text.clone();
                Some(match discrim {
                    "msg" => AcpSessionEvent::TextDelta { text },
                    "thought" => AcpSessionEvent::ThoughtDelta { text },
                    _ => unreachable!(),
                })
            }

            SessionUpdate::ToolCall {
                tool_call_id,
                title,
                kind,
                status,
                raw_input,
                locations,
                ..
            } => {
                let title = title.clone().unwrap_or_else(|| "tool".into());
                let kind_str = match kind {
                    Some(k) => format!("{:?}", k).to_lowercase(),
                    None => "other".into(),
                };
                let status = status.clone().unwrap_or(ToolCallStatus::Pending);
                let locations: Vec<String> = locations.iter().map(|l| l.path.clone()).collect();
                Some(AcpSessionEvent::ToolCall {
                    tool_call_id: tool_call_id.clone(),
                    title,
                    kind: kind_str,
                    status,
                    raw_input: raw_input.clone(),
                    locations,
                })
            }

            SessionUpdate::ToolCallUpdate {
                tool_call_id,
                status,
                content,
                ..
            } => {
                let status = status.clone().unwrap_or(ToolCallStatus::Completed);
                Some(AcpSessionEvent::ToolCallUpdate {
                    tool_call_id: tool_call_id.clone(),
                    status,
                    content: content.clone(),
                })
            }

            SessionUpdate::UsageUpdate { used, size, .. } => Some(AcpSessionEvent::UsageUpdate {
                used: *used,
                size: *size,
            }),

            SessionUpdate::SessionInfoUpdate { title, .. } => {
                Some(AcpSessionEvent::SessionInfoUpdate {
                    title: title.clone(),
                })
            }

            SessionUpdate::Plan { entries, .. } => {
                let summaries: Vec<String> = entries.iter().map(|e| e.content.clone()).collect();
                Some(AcpSessionEvent::PlanUpdate { entries: summaries })
            }

            SessionUpdate::Unknown { .. } => {
                warn!("ACP unknown session/update variant");
                None
            }
        }
    }
}

fn permission_response_for_decision(
    params: &Value,
    decision: &str,
) -> Option<Result<Value, (i64, String)>> {
    let option_id = permission_option_id_by_decision(params, decision);
    match (decision, option_id) {
        ("allow", Some(option_id)) | ("deny", Some(option_id)) => Some(Ok(json!({
            "outcome": "selected",
            "optionId": option_id
        }))),
        ("deny", None) => Some(Err((-32000, "permission denied by user".into()))),
        _ => None,
    }
}

fn permission_option_id_by_decision(params: &Value, decision: &str) -> Option<String> {
    let options = params.get("options").and_then(Value::as_array)?;
    let preferred: &[&str] = if decision == "allow" {
        &["allow_once", "allow", "approve", "yes"]
    } else {
        &["deny_once", "reject_once", "deny", "reject", "no"]
    };

    for keyword in preferred.iter().copied() {
        if let Some(option_id) = options.iter().find_map(|option| {
            let id = permission_option_id(option)?;
            let haystack = format!(
                "{} {}",
                id.to_ascii_lowercase(),
                option
                    .get("kind")
                    .and_then(Value::as_str)
                    .unwrap_or("")
                    .to_ascii_lowercase()
            );
            if haystack.contains(keyword) {
                Some(id)
            } else {
                None
            }
        }) {
            return Some(option_id);
        }
    }

    if decision == "allow" {
        return options.iter().find_map(|option| {
            let id = permission_option_id(option)?;
            let lower = id.to_ascii_lowercase();
            if lower.contains("deny") || lower.contains("reject") || lower == "no" {
                None
            } else {
                Some(id)
            }
        });
    }

    None
}

fn permission_option_id(option: &Value) -> Option<String> {
    option
        .get("optionId")
        .or_else(|| option.get("option_id"))
        .or_else(|| option.get("id"))
        .and_then(Value::as_str)
        .filter(|value| !value.trim().is_empty())
        .map(ToOwned::to_owned)
}

// ── Tests ────────────────────────────────────────────

#[cfg(test)]
mod tests {
    use super::*;
    use crate::acp_client::AcpClient;

    /// Creates a minimal ACP mock server for testing session operations.
    async fn spawn_mock_acp() -> Arc<AcpClient> {
        let script = r#"
import * as readline from 'node:readline';
import { stdin as input, stdout as output } from 'node:process';
const rl = readline.createInterface({ input, output, terminal: false });
rl.on('line', (line) => {
    const msg = JSON.parse(line);
    if (!msg.id) return;
    if (msg.method === 'session/new') {
        process.stdout.write(JSON.stringify({
            jsonrpc: '2.0', id: msg.id,
            result: { sessionId: 'test_session_1' }
        }) + '\n');
    } else if (msg.method === 'session/prompt') {
        const sessionId = msg.params?.sessionId || 'test';
        process.stdout.write(JSON.stringify({
            jsonrpc: '2.0',
            method: 'session/update',
            params: {
                sessionId,
                update: {
                    sessionUpdate: 'agent_message_chunk',
                    content: { type: 'text', text: 'Hello from mock ACP' }
                }
            }
        }) + '\n');
        process.stdout.write(JSON.stringify({
            jsonrpc: '2.0', id: msg.id,
            result: { stopReason: 'end_turn' }
        }) + '\n');
    } else if (msg.method === 'session/cancel') {
        // No response for notification
    } else if (msg.method === 'initialize') {
        process.stdout.write(JSON.stringify({
            jsonrpc: '2.0', id: msg.id,
            result: { protocolVersion: 1, agentCapabilities: {}, agentInfo: { name: 'mock', version: '1.0' }, authMethods: [] }
        }) + '\n');
    }
});
        "#;
        let dir = std::env::temp_dir();
        let script_path = dir.join("acp_session_test_mock.mjs");
        std::fs::write(&script_path, script).expect("write mock");
        let client = AcpClient::spawn("node", &[script_path.to_str().unwrap()])
            .await
            .expect("spawn");
        Arc::new(client)
    }

    #[tokio::test]
    async fn test_create_session() {
        let client = spawn_mock_acp().await;
        let mgr = AcpSessionManager::new(client);
        let sid = mgr
            .create_session(Some("/test"), None)
            .await
            .expect("create_session");
        assert_eq!(sid, "test_session_1");
        assert_eq!(mgr.session_id().await, Some("test_session_1".into()));
    }

    #[tokio::test]
    async fn test_run_prompt() {
        let client = spawn_mock_acp().await;
        let mgr = AcpSessionManager::new(client);
        mgr.create_session(Some("/test"), None)
            .await
            .expect("create_session");

        let prompt = vec![ContentBlock::Text {
            text: "Hello agent".into(),
        }];
        let result = mgr.run_prompt(prompt).await.expect("run_prompt");
        assert_eq!(format!("{:?}", result.stop_reason), "EndTurn".to_string());
        // After prompt, state should be idle again
        assert_eq!(mgr.state().await, SessionState::Idle);
    }

    #[tokio::test]
    async fn test_cancel() {
        let client = spawn_mock_acp().await;
        let mgr = AcpSessionManager::new(client);
        mgr.create_session(Some("/test"), None)
            .await
            .expect("create_session");
        mgr.cancel().await.expect("cancel");
    }

    #[tokio::test]
    async fn test_event_handler() {
        use std::sync::atomic::{AtomicBool, Ordering};
        let client = spawn_mock_acp().await;
        let mgr = AcpSessionManager::new(client);
        let received = Arc::new(AtomicBool::new(false));
        let r = received.clone();

        mgr.on_event(move |ev| {
            if matches!(ev, AcpSessionEvent::TextDelta { .. }) {
                r.store(true, Ordering::SeqCst);
            }
        });

        mgr.create_session(Some("/test"), None)
            .await
            .expect("create_session");
        let prompt = vec![ContentBlock::Text {
            text: "Hello".into(),
        }];
        mgr.run_prompt(prompt).await.expect("run_prompt");

        // Give the notification handler time to process
        sleep(Duration::from_millis(200)).await;
        assert!(
            received.load(Ordering::SeqCst),
            "should have received TextDelta"
        );
    }

    /// Creates a mock ACP that supports session/load (success case).
    async fn spawn_mock_acp_with_load() -> Arc<AcpClient> {
        let script = r#"
import * as readline from 'node:readline';
import { stdin as input, stdout as output } from 'node:process';
const rl = readline.createInterface({ input, output, terminal: false });
rl.on('line', (line) => {
    const msg = JSON.parse(line);
    if (!msg.id) return;
    if (msg.method === 'session/load') {
        process.stdout.write(JSON.stringify({
            jsonrpc: '2.0', id: msg.id,
            result: { sessionId: msg.params?.sessionId || 'loaded_session_1' }
        }) + '\n');
    } else if (msg.method === 'session/new') {
        process.stdout.write(JSON.stringify({
            jsonrpc: '2.0', id: msg.id,
            result: { sessionId: 'test_session_1' }
        }) + '\n');
    } else if (msg.method === 'session/prompt') {
        const sessionId = msg.params?.sessionId || 'test';
        process.stdout.write(JSON.stringify({
            jsonrpc: '2.0',
            method: 'session/update',
            params: {
                sessionId,
                update: {
                    sessionUpdate: 'agent_message_chunk',
                    content: { type: 'text', text: 'Hello from mock ACP' }
                }
            }
        }) + '\n');
        process.stdout.write(JSON.stringify({
            jsonrpc: '2.0', id: msg.id,
            result: { stopReason: 'end_turn' }
        }) + '\n');
    } else if (msg.method === 'session/cancel') {
        // No response for notification
    } else if (msg.method === 'initialize') {
        process.stdout.write(JSON.stringify({
            jsonrpc: '2.0', id: msg.id,
            result: { protocolVersion: 1, agentCapabilities: {}, agentInfo: { name: 'mock', version: '1.0' }, authMethods: [] }
        }) + '\n');
    }
});
        "#;
        let dir = std::env::temp_dir();
        let script_path = dir.join("acp_session_load_success_mock.mjs");
        std::fs::write(&script_path, script).expect("write mock");
        let client = AcpClient::spawn("node", &[script_path.to_str().unwrap()])
            .await
            .expect("spawn");
        Arc::new(client)
    }

    /// Creates a mock ACP that returns null for session/load (session expired/not found)
    async fn spawn_mock_acp_with_load_null() -> Arc<AcpClient> {
        let script = r#"
import * as readline from 'node:readline';
import { stdin as input, stdout as output } from 'node:process';
const rl = readline.createInterface({ input, output, terminal: false });
rl.on('line', (line) => {
    const msg = JSON.parse(line);
    if (!msg.id) return;
    if (msg.method === 'session/load') {
        // Return null result → session not found on adapter
        process.stdout.write(JSON.stringify({
            jsonrpc: '2.0', id: msg.id,
            result: null
        }) + '\n');
    } else if (msg.method === 'session/new') {
        process.stdout.write(JSON.stringify({
            jsonrpc: '2.0', id: msg.id,
            result: { sessionId: 'fallback_session_1' }
        }) + '\n');
    } else if (msg.method === 'session/prompt') {
        const sessionId = msg.params?.sessionId || 'fallback_session_1';
        process.stdout.write(JSON.stringify({
            jsonrpc: '2.0',
            method: 'session/update',
            params: {
                sessionId,
                update: {
                    sessionUpdate: 'agent_message_chunk',
                    content: { type: 'text', text: 'Hello from fallback session' }
                }
            }
        }) + '\n');
        process.stdout.write(JSON.stringify({
            jsonrpc: '2.0', id: msg.id,
            result: { stopReason: 'end_turn' }
        }) + '\n');
    } else if (msg.method === 'session/cancel') {
        // No response for notification
    } else if (msg.method === 'initialize') {
        process.stdout.write(JSON.stringify({
            jsonrpc: '2.0', id: msg.id,
            result: { protocolVersion: 1, agentCapabilities: {}, agentInfo: { name: 'mock', version: '1.0' }, authMethods: [] }
        }) + '\n');
    }
});
        "#;
        let dir = std::env::temp_dir();
        let script_path = dir.join("acp_session_load_null_mock.mjs");
        std::fs::write(&script_path, script).expect("write mock");
        let client = AcpClient::spawn("node", &[script_path.to_str().unwrap()])
            .await
            .expect("spawn");
        Arc::new(client)
    }

    #[tokio::test]
    async fn test_load_session_success() {
        let client = spawn_mock_acp_with_load().await;
        let mgr = AcpSessionManager::new(client);
        let loaded = mgr
            .load_session("stored_session_1", Some("/test"))
            .await
            .expect("load_session");
        assert!(loaded, "session should be loaded successfully");
        assert_eq!(mgr.session_id().await, Some("stored_session_1".into()));
    }

    #[tokio::test]
    async fn test_load_session_null_fallback() {
        let client = spawn_mock_acp_with_load_null().await;
        let mgr = AcpSessionManager::new(client);

        // load_session returns false for null result
        let loaded = mgr
            .load_session("expired_session", Some("/test"))
            .await
            .expect("load_session");
        assert!(!loaded, "session should not be loaded");

        // No session_id set after failed load
        assert_eq!(mgr.session_id().await, None);
    }

    #[tokio::test]
    async fn test_ensure_session_reuses_active() {
        let client = spawn_mock_acp().await;
        let mgr = AcpSessionManager::new(client);

        // First, create a session
        let created = mgr
            .create_session(Some("/test"), None)
            .await
            .expect("create_session");
        assert_eq!(created, "test_session_1");

        // ensure_session should reuse the active session, not create a new one
        let ensured = mgr
            .ensure_session(Some("/test"), Some("stored_but_ignored"))
            .await
            .expect("ensure_session");
        assert_eq!(ensured, "test_session_1");
        assert_eq!(mgr.session_id().await, Some("test_session_1".into()));
    }

    #[tokio::test]
    async fn test_ensure_session_loads_stored() {
        let client = spawn_mock_acp_with_load().await;
        let mgr = AcpSessionManager::new(client);

        // No active session, stored id provided → should load
        let ensured = mgr
            .ensure_session(Some("/test"), Some("stored_session_1"))
            .await
            .expect("ensure_session");
        assert_eq!(ensured, "stored_session_1");
        assert_eq!(mgr.session_id().await, Some("stored_session_1".into()));
    }

    #[tokio::test]
    async fn test_ensure_session_fallback_to_new() {
        let client = spawn_mock_acp_with_load_null().await;
        let mgr = AcpSessionManager::new(client);

        // No active session, stored id provided but load returns null → fallback to new
        let ensured = mgr
            .ensure_session(Some("/test"), Some("expired_session"))
            .await
            .expect("ensure_session");
        assert_eq!(ensured, "fallback_session_1");
        assert_eq!(mgr.session_id().await, Some("fallback_session_1".into()));
    }

    #[tokio::test]
    async fn test_ensure_session_no_stored_creates_new() {
        let client = spawn_mock_acp().await;
        let mgr = AcpSessionManager::new(client);

        // No active session, no stored id → should create new
        let ensured = mgr
            .ensure_session(Some("/test"), None)
            .await
            .expect("ensure_session");
        assert_eq!(ensured, "test_session_1");
        assert_eq!(mgr.session_id().await, Some("test_session_1".into()));
    }

    #[tokio::test]
    async fn test_ensure_session_fallback_then_prompt_works() {
        let client = spawn_mock_acp_with_load_null().await;
        let mgr = AcpSessionManager::new(client);

        // ensure_session with expired stored id → should create fallback
        mgr.ensure_session(Some("/test"), Some("expired_session"))
            .await
            .expect("ensure_session");

        // Verify prompt still works on the fallback session
        let prompt = vec![ContentBlock::Text {
            text: "Hello after fallback".into(),
        }];
        let result = mgr
            .run_prompt(prompt)
            .await
            .expect("run_prompt after fallback");
        assert_eq!(format!("{:?}", result.stop_reason), "EndTurn".to_string());
    }

    #[test]
    fn test_thought_chunk_maps_to_thought_delta() {
        let accumulated = Arc::new(Mutex::new(String::new()));
        let update = SessionUpdate::AgentThoughtChunk {
            session_update: SessionUpdate::AGENT_THOUGHT_CHUNK.into(),
            content: crate::acp_types::TextContent {
                content_type: "text".into(),
                text: "internal reasoning".into(),
            },
        };

        let event = AcpSessionManager::session_update_to_event(&update, &accumulated, true)
            .expect("thought chunk should emit event");

        match event {
            AcpSessionEvent::ThoughtDelta { text } => assert_eq!(text, "internal reasoning"),
            other => panic!("expected ThoughtDelta, got {other:?}"),
        }
    }

    #[test]
    fn permission_response_selects_allow_and_deny_options() {
        let params = json!({
            "options": [
                {"optionId": "allow_once", "name": "Allow Once", "kind": "allow_once"},
                {"optionId": "reject_once", "name": "Reject Once", "kind": "reject_once"}
            ]
        });

        let allow = permission_response_for_decision(&params, "allow")
            .expect("allow response")
            .expect("allow should select an option");
        assert_eq!(allow["outcome"], "selected");
        assert_eq!(allow["optionId"], "allow_once");

        let deny = permission_response_for_decision(&params, "deny")
            .expect("deny response")
            .expect("deny should select an option");
        assert_eq!(deny["outcome"], "selected");
        assert_eq!(deny["optionId"], "reject_once");
    }
}