use std::{
    collections::{HashMap, VecDeque},
    path::{Path, PathBuf},
    process::Stdio,
    sync::{Arc, Mutex},
    time::{Duration, Instant},
};

use anyhow::{Context, Result, bail};
use async_trait::async_trait;
use futures_util::StreamExt;
use serde_json::Value;
use tokio::{
    io::{AsyncBufReadExt, AsyncRead, AsyncWrite, AsyncWriteExt, BufReader},
    process::{Child, Command},
    sync::mpsc,
    time::sleep,
};

use crate::{
    ai_approval::{self, AiApprovalDecisionKind},
    app_state::AppState,
    approval_policy::{self, AutoApprovalDecision},
    claude_hook::{
        ClaudeHookStatusEvent, ClaudePermissionRequest, append_always_allow_command,
        claude_state_dir, ensure_runtime_dirs, request_path, response_from_choice, response_path,
    },
    claude_store::{load_claude_archive_summary, load_claude_messages},
    models::{
        AgentKind, ApprovalChoice, ApprovalKind, ApprovalRequest, ChatMessage, ProjectSummary,
        ResolvedProviderConfig, SessionSummary,
    },
    session_store::{load_session_archive_summary, load_session_messages},
};

#[async_trait]
pub trait AgentProvider: Send + Sync {
    async fn list_projects(&self) -> HashMap<String, ProjectSummary> {
        HashMap::new()
    }

    async fn list_sessions(&self) -> HashMap<String, SessionSummary> {
        HashMap::new()
    }

    async fn list_messages(&self, _session_id: &str) -> Option<Vec<ChatMessage>> {
        None
    }

    async fn default_runtime_ref(&self, _session_id: &str) -> Option<String> {
        None
    }

    async fn run_session(
        &self,
        state: Arc<AppState>,
        session: SessionSummary,
        input: ChatMessage,
        system_prompt: Option<String>,
        reply: ChatMessage,
        provider_config: Option<ResolvedProviderConfig>,
    ) -> Result<()>;

    async fn summarize_reply(
        &self,
        _state: Arc<AppState>,
        _session: SessionSummary,
        _content: String,
    ) -> Result<String> {
        bail!("summary is not supported by this agent")
    }
}

pub struct ProviderRegistry {
    providers: HashMap<AgentKind, Arc<dyn AgentProvider>>,
}

impl ProviderRegistry {
    pub fn new() -> Self {
        let mut providers: HashMap<AgentKind, Arc<dyn AgentProvider>> = HashMap::new();
        providers.insert(AgentKind::Codex, Arc::new(CodexProvider::new()));
        providers.insert(AgentKind::ClaudeCode, Arc::new(ClaudeCodeProvider::new()));
        providers.insert(AgentKind::OpenCode, Arc::new(OpenCodeProvider::new()));
        providers.insert(
            AgentKind::Custom,
            Arc::new(StubProvider::new(AgentKind::Custom)),
        );
        Self { providers }
    }

    pub fn get(&self, kind: AgentKind) -> Option<Arc<dyn AgentProvider>> {
        self.providers.get(&kind).cloned()
    }

    pub fn all(&self) -> Vec<Arc<dyn AgentProvider>> {
        self.providers.values().cloned().collect()
    }
}

struct StubProvider {
    kind: AgentKind,
}

impl StubProvider {
    fn new(kind: AgentKind) -> Self {
        Self { kind }
    }
}

#[async_trait]
impl AgentProvider for StubProvider {
    async fn run_session(
        &self,
        state: Arc<AppState>,
        session: SessionSummary,
        input: ChatMessage,
        _system_prompt: Option<String>,
        reply: ChatMessage,
        _provider_config: Option<ResolvedProviderConfig>,
    ) -> Result<()> {
        let prefix = match self.kind {
            AgentKind::ClaudeCode => "ClaudeCode",
            AgentKind::OpenCode => "OpenCode",
            AgentKind::Custom => "CustomAgent",
            AgentKind::Codex => "Codex",
        };

        let chunks = [
            format!("{prefix} 已接收任务。"),
            "当前 provider 架构已就位，这个 agent 还没有接入真实 CLI。".to_string(),
            format!("任务内容：{}", input.content),
        ];

        for chunk in chunks {
            sleep(Duration::from_millis(120)).await;
            state
                .emit_message_delta(&session.id, &reply.id, &chunk)
                .await;
        }

        state
            .finish_assistant_message(&session.id, &reply.id)
            .await
            .map_err(anyhow::Error::msg)?;
        Ok(())
    }

    async fn summarize_reply(
        &self,
        _state: Arc<AppState>,
        _session: SessionSummary,
        content: String,
    ) -> Result<String> {
        let summary = content.chars().take(60).collect::<String>();
        Ok(summary)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::models::{MessageRole, SessionStatus};

    #[test]
    fn provider_registry_has_provider_for_every_agent_kind() {
        let registry = ProviderRegistry::new();

        for kind in AgentKind::ALL {
            assert!(
                registry.get(kind).is_some(),
                "missing provider for {kind:?}"
            );
        }
    }

    #[test]
    fn codex_streaming_state_parses_content_status_and_completion() {
        let mut state = CodexAppServerStreamingState::default();

        match state.ingest_value(&serde_json::json!({
            "method": "item/agentMessage/delta",
            "params": {
                "itemId": "msg-1",
                "delta": "hello"
            }
        })) {
            CodexAppServerEvent::Content(text) => assert_eq!(text, "hello"),
            _ => panic!("expected content delta"),
        }

        match state.ingest_value(&serde_json::json!({
            "method": "item/completed",
            "params": {
                "item": {
                    "id": "msg-1",
                    "type": "agentMessage",
                    "text": "hello world"
                }
            }
        })) {
            CodexAppServerEvent::Content(text) => assert_eq!(text, "hello world"),
            _ => panic!("expected completed content"),
        }

        match state.ingest_value(&serde_json::json!({
            "method": "turn/completed",
            "params": {
                "turn": {
                    "status": "completed"
                }
            }
        })) {
            CodexAppServerEvent::TurnCompleted => {}
            _ => panic!("expected turn completion"),
        }

        assert_eq!(state.finish_text().as_deref(), Some("hello world"));
    }

    #[test]
    fn codex_streaming_state_maps_approval_requests() {
        let mut state = CodexAppServerStreamingState::default();

        match state.ingest_value(&serde_json::json!({
            "id": 7,
            "method": "item/commandExecution/requestApproval",
            "params": {
                "command": "cargo test",
                "reason": "run tests",
                "availableDecisions": ["accept", "acceptForSession", "cancel"]
            }
        })) {
            CodexAppServerEvent::ApprovalRequested(pending) => {
                assert_eq!(pending.request.request_id, "7");
                assert_eq!(pending.request.command.as_deref(), Some("cargo test"));
                assert_eq!(pending.request.reason.as_deref(), Some("run tests"));
                assert!(pending.request.allow_accept_for_session);
                assert!(pending.request.allow_cancel);
                assert!(pending.request.resolvable);
            }
            _ => panic!("expected approval request"),
        }
    }

    #[test]
    fn codex_streaming_state_maps_file_change_approval_requests() {
        let mut state = CodexAppServerStreamingState::default();

        match state.ingest_value(&serde_json::json!({
            "id": 8,
            "method": "item/fileChange/requestApproval",
            "params": {
                "reason": "Would you like to make the following edits?",
                "availableDecisions": ["accept", "acceptForSession", "cancel"],
                "fileChanges": [
                    { "path": "src/adapter.rs" },
                    { "path": "src/models.rs" }
                ]
            }
        })) {
            CodexAppServerEvent::ApprovalRequested(pending) => {
                assert_eq!(pending.request.request_id, "8");
                assert!(matches!(pending.request.kind, ApprovalKind::FileChange));
                assert_eq!(
                    pending.request.command.as_deref(),
                    Some("edit files: src/adapter.rs, src/models.rs")
                );
                assert_eq!(
                    pending.request.reason.as_deref(),
                    Some("Would you like to make the following edits?")
                );
                assert!(pending.request.allow_accept_for_session);
                assert!(pending.request.allow_cancel);
                assert!(pending.request.resolvable);
            }
            _ => panic!("expected file change approval request"),
        }
    }

    #[test]
    fn codex_streaming_state_maps_legacy_apply_patch_approval_requests() {
        let mut state = CodexAppServerStreamingState::default();

        match state.ingest_value(&serde_json::json!({
            "id": "patch-1",
            "method": "applyPatchApproval",
            "params": {
                "files": ["src/adapter.rs"]
            }
        })) {
            CodexAppServerEvent::ApprovalRequested(pending) => {
                assert_eq!(pending.request.request_id, "patch-1");
                assert!(matches!(pending.request.kind, ApprovalKind::ApplyPatch));
                assert_eq!(
                    pending.request.command.as_deref(),
                    Some("edit files: src/adapter.rs")
                );
                assert!(pending.request.allow_accept_for_session);
                assert!(pending.request.allow_cancel);
                assert!(pending.request.resolvable);
            }
            _ => panic!("expected apply patch approval request"),
        }
    }

    #[test]
    fn codex_approval_result_serializes_file_change_decisions() {
        assert_eq!(
            approval_result_json(&ApprovalChoice::Accept, &ApprovalKind::FileChange),
            serde_json::json!({ "decision": "accept" })
        );
        assert_eq!(
            approval_result_json(&ApprovalChoice::AcceptForSession, &ApprovalKind::FileChange),
            serde_json::json!({ "decision": "acceptForSession" })
        );
        assert_eq!(
            approval_result_json(&ApprovalChoice::AcceptForSession, &ApprovalKind::ApplyPatch),
            serde_json::json!({ "decision": "approved_for_session" })
        );
    }

    #[test]
    fn claude_streaming_state_parses_assistant_result_and_errors() {
        let mut state = ClaudeStreamingState::default();

        let rendered = state
            .ingest_line(
                r#"{"type":"assistant","message":{"content":[{"type":"text","text":"partial"}]}}"#,
            )
            .expect("valid assistant event");
        assert_eq!(rendered.as_deref(), Some("partial"));
        assert_eq!(state.finish_text().as_deref(), Some("partial"));

        let rendered = state
            .ingest_line(r#"{"type":"result","result":"final answer","is_error":false}"#)
            .expect("valid result event");
        assert_eq!(rendered.as_deref(), Some("final answer"));
        assert_eq!(state.finish_text().as_deref(), Some("final answer"));

        let error = state
            .ingest_line(r#"{"type":"result","result":"failed","is_error":true}"#)
            .expect_err("error result should fail");
        assert!(error.to_string().contains("failed"));
    }

    #[test]
    fn claude_streaming_state_parses_stream_events() {
        let mut state = ClaudeStreamingState::default();

        let rendered = state
            .ingest_line(
                r#"{"type":"stream_event","event":{"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}}"#,
            )
            .expect("valid content block start");
        assert_eq!(rendered, None);

        let rendered = state
            .ingest_line(
                r#"{"type":"stream_event","event":{"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"hello"}}}"#,
            )
            .expect("valid content block delta");
        assert_eq!(rendered.as_deref(), Some("hello"));
        assert_eq!(state.finish_text().as_deref(), Some("hello"));

        let rendered = state
            .ingest_line(
                r#"{"type":"stream_event","event":{"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":" world"}}}"#,
            )
            .expect("valid second content block delta");
        assert_eq!(rendered.as_deref(), Some("hello world"));

        let rendered = state
            .ingest_line(
                r#"{"type":"stream_event","event":{"type":"content_block_stop","index":0}}"#,
            )
            .expect("valid content block stop");
        assert_eq!(rendered.as_deref(), Some("hello world"));
        assert_eq!(state.finish_text().as_deref(), Some("hello world"));
    }

    #[test]
    fn context_migration_instructions_include_recent_history_without_current_turn() {
        let current_input = ChatMessage {
            id: "input-now".to_string(),
            session_id: "session".to_string(),
            role: MessageRole::User,
            content: "最新问题".to_string(),
            created_at: chrono::Utc::now(),
        };
        let pending_reply = ChatMessage {
            id: "reply-now".to_string(),
            session_id: "session".to_string(),
            role: MessageRole::Assistant,
            content: String::new(),
            created_at: chrono::Utc::now(),
        };
        let messages = vec![
            ChatMessage {
                id: "old-user".to_string(),
                session_id: "session".to_string(),
                role: MessageRole::User,
                content: "先看 provider 配置".to_string(),
                created_at: chrono::Utc::now(),
            },
            ChatMessage {
                id: "old-assistant".to_string(),
                session_id: "session".to_string(),
                role: MessageRole::Assistant,
                content: "已确认请求体发的是 gpt-5.5".to_string(),
                created_at: chrono::Utc::now(),
            },
            current_input.clone(),
            pending_reply.clone(),
        ];

        let instructions =
            build_context_migration_instructions(&messages, &current_input, &pending_reply)
                .expect("history should be generated");

        assert!(instructions.contains("先看 provider 配置"));
        assert!(instructions.contains("已确认请求体发的是 gpt-5.5"));
        assert!(!instructions.contains("最新问题"));
    }

    #[test]
    fn claude_runtime_ref_selection_respects_resume_and_model_switch() {
        let session_id = "local-session";
        let existing_runtime_ref = "claude-runtime";

        assert_eq!(
            select_claude_runtime_ref(session_id, None, false),
            session_id
        );
        assert_eq!(
            select_claude_runtime_ref(session_id, Some(existing_runtime_ref), true),
            existing_runtime_ref
        );

        let migrated = select_claude_runtime_ref(session_id, Some(existing_runtime_ref), false);
        assert_ne!(migrated, session_id);
        assert_ne!(migrated, existing_runtime_ref);
        assert!(uuid::Uuid::parse_str(&migrated).is_ok());
    }

    #[test]
    fn context_migration_instructions_keep_recent_messages_when_history_is_long() {
        let current_input = ChatMessage {
            id: "input-now".to_string(),
            session_id: "session".to_string(),
            role: MessageRole::User,
            content: "当前输入".to_string(),
            created_at: chrono::Utc::now(),
        };
        let pending_reply = ChatMessage {
            id: "reply-now".to_string(),
            session_id: "session".to_string(),
            role: MessageRole::Assistant,
            content: String::new(),
            created_at: chrono::Utc::now(),
        };
        let messages = (0..15)
            .map(|index| ChatMessage {
                id: format!("msg-{index}"),
                session_id: "session".to_string(),
                role: if index % 2 == 0 {
                    MessageRole::User
                } else {
                    MessageRole::Assistant
                },
                content: format!("历史消息 {index}"),
                created_at: chrono::Utc::now(),
            })
            .collect::<Vec<_>>();

        let instructions =
            build_context_migration_instructions(&messages, &current_input, &pending_reply)
                .expect("history should be generated");

        assert!(!instructions.contains("历史消息 0"));
        assert!(!instructions.contains("历史消息 2"));
        assert!(instructions.contains("历史消息 14"));
    }

    #[test]
    fn codex_thread_action_resumes_when_provider_matches_even_if_model_changes() {
        assert_eq!(
            decide_codex_thread_action(Some("thread-1"), Some("provider-a"), Some("provider-a")),
            CodexThreadDecision::Resume
        );
    }

    #[test]
    fn codex_thread_action_migrates_when_provider_changes() {
        assert_eq!(
            decide_codex_thread_action(Some("thread-1"), Some("provider-a"), Some("provider-b")),
            CodexThreadDecision::StartWithMigration
        );
        assert_eq!(
            decide_codex_thread_action(None, Some("provider-a"), Some("provider-a")),
            CodexThreadDecision::StartFresh
        );
    }

    #[tokio::test]
    async fn stub_provider_completes_when_called_directly() {
        let state = Arc::new(AppState::new().await);
        let provider = StubProvider::new(AgentKind::Custom);
        let session = SessionSummary {
            id: "stub-session".to_string(),
            project_id: "project".to_string(),
            title: "Stub".to_string(),
            agent: AgentKind::Custom,
            brief_reply_mode: false,
            status: SessionStatus::Running,
            updated_at: chrono::Utc::now(),
            unread_count: 0,
            last_message_preview: None,
            pending_approval: None,
            provider_id: None,
        };
        let input = ChatMessage {
            id: "input".to_string(),
            session_id: session.id.clone(),
            role: MessageRole::User,
            content: "ping".to_string(),
            created_at: chrono::Utc::now(),
        };
        let reply = ChatMessage {
            id: "reply".to_string(),
            session_id: session.id.clone(),
            role: MessageRole::Assistant,
            content: String::new(),
            created_at: chrono::Utc::now(),
        };

        let result = provider
            .run_session(Arc::clone(&state), session, input, None, reply, None)
            .await
            .expect_err("direct call without seeded reply should expose state contract");
        assert!(result.to_string().contains("unknown message"));
    }

    #[test]
    fn opencode_streaming_state_parses_session_text_tools_and_errors() {
        let mut state = OpenCodeStreamingState::default();

        match state
            .ingest_line(
                r#"{"type":"step_start","timestamp":1767036059338,"sessionID":"ses_123","part":{"type":"step-start"}}"#,
            )
            .expect("valid step start")
        {
            OpenCodeStreamEvent::Session(session_id) => assert_eq!(session_id, "ses_123"),
            _ => panic!("expected session event"),
        }

        match state
            .ingest_line(
                r#"{"type":"text","timestamp":1767036064268,"sessionID":"ses_123","part":{"type":"text","text":"hello"}}"#,
            )
            .expect("valid text")
        {
            OpenCodeStreamEvent::Content(text) => assert_eq!(text, "hello"),
            _ => panic!("expected text content"),
        }

        match state
            .ingest_line(
                r#"{"type":"tool_use","timestamp":1767036064269,"sessionID":"ses_123","part":{"tool":"bash","status":"completed","input":{"command":"cargo test"}}}"#,
            )
            .expect("valid tool use")
        {
            OpenCodeStreamEvent::Status(status) => assert!(status.contains("cargo test")),
            _ => panic!("expected tool status"),
        }

        match state
            .ingest_line(
                r#"{"type":"step_finish","timestamp":1767036064273,"sessionID":"ses_123","part":{"type":"step-finish","reason":"stop"}}"#,
            )
            .expect("valid step finish")
        {
            OpenCodeStreamEvent::None => {}
            _ => panic!("expected no content on final step"),
        }
        assert_eq!(state.finish_text().as_deref(), Some("hello"));

        match state
            .ingest_line(
                r#"{"type":"error","timestamp":1767036065000,"sessionID":"ses_123","error":{"name":"APIError","data":{"message":"rate limited"}}}"#,
            )
            .expect("valid error")
        {
            OpenCodeStreamEvent::Error(message) => assert!(message.contains("rate limited")),
            _ => panic!("expected error event"),
        }
    }
}

#[derive(Clone)]
struct CachedMessageSet {
    fingerprint: u64,
    messages: Vec<ChatMessage>,
}

#[derive(Clone)]
struct CachedCodexArchive {
    checked_at: Instant,
    fingerprint: u64,
    projects: HashMap<String, ProjectSummary>,
    sessions: HashMap<String, SessionSummary>,
    session_files: HashMap<String, PathBuf>,
    messages: HashMap<String, CachedMessageSet>,
}

struct CodexProvider {
    cache: Mutex<Option<CachedCodexArchive>>,
}

impl CodexProvider {
    const ARCHIVE_CACHE_TTL: Duration = Duration::from_secs(2);

    fn new() -> Self {
        Self {
            cache: Mutex::new(None),
        }
    }

    fn ensure_archive(&self) -> CachedCodexArchive {
        {
            let cache = self.cache.lock().expect("codex cache poisoned");
            if let Some(existing) = cache.as_ref()
                && existing.checked_at.elapsed() < Self::ARCHIVE_CACHE_TTL
            {
                return existing.clone();
            }
        }

        let summary = load_session_archive_summary();
        let mut cache = self.cache.lock().expect("codex cache poisoned");
        if let Some(existing) = cache.as_mut()
            && existing.fingerprint == summary.fingerprint
        {
            existing.checked_at = Instant::now();
            return existing.clone();
        }

        let messages = cache
            .as_ref()
            .map(|existing| existing.messages.clone())
            .unwrap_or_default();
        let archive = CachedCodexArchive {
            checked_at: Instant::now(),
            fingerprint: summary.fingerprint,
            projects: summary.projects,
            sessions: summary.sessions,
            session_files: summary.session_files,
            messages,
        };
        *cache = Some(archive.clone());
        archive
    }

    fn load_messages_for_session(&self, session_id: &str) -> Option<Vec<ChatMessage>> {
        let mut archive = self.ensure_archive();
        let path = archive.session_files.get(session_id)?.clone();
        let loaded = load_session_messages(&path)?;
        if let Some(existing) = archive.messages.get(session_id)
            && existing.fingerprint == loaded.fingerprint
        {
            return Some(existing.messages.clone());
        }

        archive.messages.insert(
            session_id.to_string(),
            CachedMessageSet {
                fingerprint: loaded.fingerprint,
                messages: loaded.messages.clone(),
            },
        );
        let mut cache = self.cache.lock().expect("codex cache poisoned");
        *cache = Some(archive);
        Some(loaded.messages)
    }
}

#[async_trait]
impl AgentProvider for CodexProvider {
    async fn list_projects(&self) -> HashMap<String, ProjectSummary> {
        self.ensure_archive().projects
    }

    async fn list_sessions(&self) -> HashMap<String, SessionSummary> {
        self.ensure_archive().sessions
    }

    async fn list_messages(&self, session_id: &str) -> Option<Vec<ChatMessage>> {
        self.load_messages_for_session(session_id)
    }

    async fn default_runtime_ref(&self, session_id: &str) -> Option<String> {
        self.ensure_archive()
            .sessions
            .contains_key(session_id)
            .then(|| session_id.to_string())
    }

    async fn run_session(
        &self,
        state: Arc<AppState>,
        session: SessionSummary,
        input: ChatMessage,
        system_prompt: Option<String>,
        reply: ChatMessage,
        provider_config: Option<ResolvedProviderConfig>,
    ) -> Result<()> {
        run_codex(
            state,
            &session,
            &input,
            system_prompt.as_deref(),
            &reply,
            provider_config,
        )
        .await
    }

    async fn summarize_reply(
        &self,
        state: Arc<AppState>,
        session: SessionSummary,
        content: String,
    ) -> Result<String> {
        summarize_with_codex(state, &session, &content).await
    }
}

#[derive(Clone)]
struct CachedClaudeArchive {
    checked_at: Instant,
    fingerprint: u64,
    projects: HashMap<String, ProjectSummary>,
    sessions: HashMap<String, SessionSummary>,
    session_files: HashMap<String, PathBuf>,
    messages: HashMap<String, CachedMessageSet>,
}

struct ClaudeCodeProvider {
    cache: Mutex<Option<CachedClaudeArchive>>,
}

impl ClaudeCodeProvider {
    const ARCHIVE_CACHE_TTL: Duration = Duration::from_secs(2);

    fn new() -> Self {
        Self {
            cache: Mutex::new(None),
        }
    }

    fn ensure_archive(&self) -> CachedClaudeArchive {
        {
            let cache = self.cache.lock().expect("claude cache poisoned");
            if let Some(existing) = cache.as_ref()
                && existing.checked_at.elapsed() < Self::ARCHIVE_CACHE_TTL
            {
                return existing.clone();
            }
        }

        let summary = load_claude_archive_summary();
        let mut cache = self.cache.lock().expect("claude cache poisoned");
        if let Some(existing) = cache.as_mut()
            && existing.fingerprint == summary.fingerprint
        {
            existing.checked_at = Instant::now();
            return existing.clone();
        }

        let messages = cache
            .as_ref()
            .map(|existing| existing.messages.clone())
            .unwrap_or_default();
        let archive = CachedClaudeArchive {
            checked_at: Instant::now(),
            fingerprint: summary.fingerprint,
            projects: summary.projects,
            sessions: summary.sessions,
            session_files: summary.session_files,
            messages,
        };
        *cache = Some(archive.clone());
        archive
    }

    fn load_messages_for_session(&self, session_id: &str) -> Option<Vec<ChatMessage>> {
        let mut archive = self.ensure_archive();
        let path = archive.session_files.get(session_id)?.clone();
        let loaded = load_claude_messages(&path)?;
        if let Some(existing) = archive.messages.get(session_id)
            && existing.fingerprint == loaded.fingerprint
        {
            return Some(existing.messages.clone());
        }

        archive.messages.insert(
            session_id.to_string(),
            CachedMessageSet {
                fingerprint: loaded.fingerprint,
                messages: loaded.messages.clone(),
            },
        );
        let mut cache = self.cache.lock().expect("claude cache poisoned");
        *cache = Some(archive);
        Some(loaded.messages)
    }
}

#[async_trait]
impl AgentProvider for ClaudeCodeProvider {
    async fn list_projects(&self) -> HashMap<String, ProjectSummary> {
        self.ensure_archive().projects
    }

    async fn list_sessions(&self) -> HashMap<String, SessionSummary> {
        self.ensure_archive().sessions
    }

    async fn list_messages(&self, session_id: &str) -> Option<Vec<ChatMessage>> {
        self.load_messages_for_session(session_id)
    }

    async fn default_runtime_ref(&self, session_id: &str) -> Option<String> {
        self.ensure_archive()
            .sessions
            .contains_key(session_id)
            .then(|| session_id.to_string())
    }

    async fn run_session(
        &self,
        state: Arc<AppState>,
        session: SessionSummary,
        input: ChatMessage,
        system_prompt: Option<String>,
        reply: ChatMessage,
        provider_config: Option<ResolvedProviderConfig>,
    ) -> Result<()> {
        run_claude_code(
            state,
            &session,
            &input,
            system_prompt.as_deref(),
            &reply,
            provider_config,
        )
        .await
    }

    async fn summarize_reply(
        &self,
        state: Arc<AppState>,
        session: SessionSummary,
        content: String,
    ) -> Result<String> {
        summarize_with_claude_code(state, &session, &content).await
    }
}

struct OpenCodeProvider;

impl OpenCodeProvider {
    fn new() -> Self {
        Self
    }
}

#[async_trait]
impl AgentProvider for OpenCodeProvider {
    async fn list_projects(&self) -> HashMap<String, ProjectSummary> {
        match OpenCodeHttpClient::start().await {
            Ok(mut client) => {
                let result = client.list_projects().await.unwrap_or_default();
                client.shutdown().await;
                result
            }
            Err(_) => HashMap::new(),
        }
    }

    async fn list_sessions(&self) -> HashMap<String, SessionSummary> {
        match OpenCodeHttpClient::start().await {
            Ok(mut client) => {
                let result = client.list_sessions().await.unwrap_or_default();
                client.shutdown().await;
                result
            }
            Err(_) => HashMap::new(),
        }
    }

    async fn list_messages(&self, session_id: &str) -> Option<Vec<ChatMessage>> {
        let mut client = OpenCodeHttpClient::start().await.ok()?;
        let result = client.list_messages(session_id).await.ok();
        client.shutdown().await;
        result
    }

    async fn default_runtime_ref(&self, _session_id: &str) -> Option<String> {
        None
    }

    async fn run_session(
        &self,
        state: Arc<AppState>,
        session: SessionSummary,
        input: ChatMessage,
        system_prompt: Option<String>,
        reply: ChatMessage,
        provider_config: Option<ResolvedProviderConfig>,
    ) -> Result<()> {
        run_opencode(
            state,
            &session,
            &input,
            system_prompt.as_deref(),
            &reply,
            provider_config,
        )
        .await
    }

    async fn summarize_reply(
        &self,
        state: Arc<AppState>,
        session: SessionSummary,
        content: String,
    ) -> Result<String> {
        summarize_with_opencode(state, &session, &content).await
    }
}

fn spawn_codex_app_server(cwd: &Path) -> Result<Child> {
    spawn_codex_app_server_with_config(cwd, None, None)
}

fn spawn_codex_app_server_with_config(
    cwd: &Path,
    provider_config: Option<&ResolvedProviderConfig>,
    codex_provider_name: Option<&str>,
) -> Result<Child> {
    let binary = codex_binary_path();
    let mut command = Command::new(&binary);
    command
        .args(["app-server", "--listen", "stdio://"])
        .current_dir(cwd)
        .stdin(Stdio::piped())
        .stdout(Stdio::piped())
        .stderr(Stdio::piped())
        .kill_on_drop(true);

    let mut c_overrides: Vec<String> = Vec::new();
    let push_arg = |cmd: &mut Command, vec: &mut Vec<String>, kv: String| {
        cmd.args(["-c", &kv]);
        vec.push(kv);
    };
    let toml_string = |value: &str| serde_json::to_string(value).expect("string serializes");

    // Force manual approval routing for app-server clients. `approval_policy`
    // alone is not enough when Codex config enables an automatic reviewer.
    push_arg(
        &mut command,
        &mut c_overrides,
        "approval_policy=\"on-request\"".to_string(),
    );
    push_arg(
        &mut command,
        &mut c_overrides,
        "approvals_reviewer=\"user\"".to_string(),
    );

    // Apply provider configuration via -c config overrides while preserving the
    // user's CODEX_HOME for MCP, trust, and other Codex settings.
    if let Some(config) = provider_config {
        let name = codex_provider_name.unwrap_or("omni-bridge");
        let display_name =
            find_codex_provider_display_name(name).unwrap_or_else(|| name.to_string());

        push_arg(
            &mut command,
            &mut c_overrides,
            format!("model_provider={}", toml_string(name)),
        );
        push_arg(
            &mut command,
            &mut c_overrides,
            format!("model_providers.{name}.name={}", toml_string(&display_name)),
        );
        push_arg(
            &mut command,
            &mut c_overrides,
            format!(
                "model_providers.{name}.base_url={}",
                toml_string(&config.base_url)
            ),
        );
        push_arg(
            &mut command,
            &mut c_overrides,
            format!("model_providers.{name}.requires_openai_auth=false"),
        );
        push_arg(
            &mut command,
            &mut c_overrides,
            format!("model_providers.{name}.wire_api=\"responses\""),
        );
        if !config.api_key.is_empty() {
            let auth_env_var = "OMNI_CODEX_PROVIDER_KEY";
            command.env(auth_env_var, &config.api_key);
            push_arg(
                &mut command,
                &mut c_overrides,
                format!(
                    "model_providers.{name}.env_key={}",
                    toml_string(auth_env_var)
                ),
            );
        }
        if let Some(ref model) = config.model {
            push_arg(
                &mut command,
                &mut c_overrides,
                format!("model={}", toml_string(model)),
            );
            push_arg(
                &mut command,
                &mut c_overrides,
                format!("model_providers.{name}.model={}", toml_string(model)),
            );
        }
    }
    eprintln!("[codex] full -c args: {c_overrides:#?}");

    command
        .spawn()
        .with_context(|| format!("failed to spawn `{}`", binary.display()))
}

fn find_codex_provider_name() -> Option<String> {
    let path = codex_home_dir().join("config.toml");
    let content = std::fs::read_to_string(path).ok()?;
    for line in content.lines() {
        let trimmed = line.trim();
        if trimmed.starts_with("model_provider") && trimmed.contains('=') {
            let value = trimmed.splitn(2, '=').nth(1)?.trim();
            return Some(value.trim_matches('"').trim_matches('\'').to_string());
        }
    }
    None
}

/// Read the `name` field from a model_providers section in config.toml.
fn find_codex_provider_display_name(provider_key: &str) -> Option<String> {
    let path = codex_home_dir().join("config.toml");
    let content = std::fs::read_to_string(path).ok()?;
    let section_header = format!("[model_providers.{provider_key}]");
    let mut in_section = false;
    for line in content.lines() {
        let trimmed = line.trim();
        if trimmed == section_header {
            in_section = true;
            continue;
        }
        if in_section {
            if trimmed.starts_with('[') && trimmed.ends_with(']') {
                break;
            }
            if trimmed.starts_with("name") && trimmed.contains('=') {
                let value = trimmed.splitn(2, '=').nth(1)?.trim();
                return Some(value.trim_matches('"').trim_matches('\'').to_string());
            }
        }
    }
    None
}

/// Read the model_provider from codex's session data file for a given session ID.
/// Session files are stored at ~/.codex/sessions/YYYY/MM/DD/rollout-*-<session_id>.jsonl
fn read_codex_session_provider(session_id: &str) -> Option<String> {
    let sessions_dir = codex_home_dir().join("sessions");
    find_session_file(&sessions_dir, session_id).and_then(|path| {
        let content = std::fs::read_to_string(path).ok()?;
        let first_line = content.lines().next()?;
        let meta: serde_json::Value = serde_json::from_str(first_line).ok()?;
        meta.pointer("/payload/model_provider")
            .and_then(|v| v.as_str())
            .map(String::from)
    })
}

fn find_session_file(dir: &Path, session_id: &str) -> Option<PathBuf> {
    for entry in std::fs::read_dir(dir).ok()? {
        let entry = entry.ok()?;
        let path = entry.path();
        if path.is_dir() {
            if let Some(found) = find_session_file(&path, session_id) {
                return Some(found);
            }
        } else {
            let name = path.file_name()?.to_string_lossy();
            if name.contains(session_id) && name.ends_with(".jsonl") {
                return Some(path);
            }
        }
    }
    None
}

/// Path to the codex home directory (~/.codex)
fn codex_home_dir() -> PathBuf {
    if let Some(home) = std::env::var_os("HOME").map(PathBuf::from) {
        home.join(".codex")
    } else {
        PathBuf::from(".codex")
    }
}

async fn start_codex_thread(
    stdin: &mut (impl tokio::io::AsyncWrite + Unpin),
    next_request_id: &mut u64,
    stdout_rx: &mut mpsc::UnboundedReceiver<std::result::Result<String, String>>,
    raw_stdout: &mut String,
    state: &AppState,
    session: &SessionSummary,
    developer_instructions: &Option<String>,
) -> Result<serde_json::Value> {
    let request_id = send_json_rpc_request(
        stdin,
        next_request_id,
        "thread/start",
        serde_json::json!({
            "cwd": state
                .project_root_path_for_session(&session.id)
                .await
                .map_err(anyhow::Error::msg)?,
            "approvalPolicy": "on-request",
            "approvalsReviewer": "user",
            "sandbox": "workspace-write",
            "persistExtendedHistory": false,
            "developerInstructions": developer_instructions,
        }),
    )
    .await?;
    wait_for_json_rpc_response(stdout_rx, raw_stdout, request_id).await
}

fn brief_reply_developer_prompt(session: &SessionSummary) -> Option<&'static str> {
    session.brief_reply_mode.then_some(
        "回复要求：请简短说明你做了什么和结果，尽量不超过 50 个汉字。只保留关键动作、结果或结论，避免展开解释。",
    )
}

fn log_codex_thread_response(session_id: &str, response: &Value) {
    let thread_id = response
        .pointer("/thread/id")
        .or_else(|| response.pointer("/thread/threadId"))
        .and_then(Value::as_str)
        .unwrap_or("<missing>");
    let approval_policy = response
        .pointer("/thread/approvalPolicy")
        .or_else(|| response.pointer("/thread/approval_policy"))
        .or_else(|| response.pointer("/approvalPolicy"))
        .or_else(|| response.pointer("/approval_policy"))
        .and_then(Value::as_str)
        .unwrap_or("<missing>");
    let approvals_reviewer = response
        .pointer("/thread/approvalsReviewer")
        .or_else(|| response.pointer("/thread/approvals_reviewer"))
        .or_else(|| response.pointer("/approvalsReviewer"))
        .or_else(|| response.pointer("/approvals_reviewer"))
        .and_then(Value::as_str)
        .unwrap_or("<missing>");
    eprintln!(
        "[codex] thread response session={session_id} thread={thread_id} approvalPolicy={approval_policy} approvalsReviewer={approvals_reviewer}"
    );
}

fn turn_system_prompt<'a>(
    session: &SessionSummary,
    system_prompt: Option<&'a str>,
) -> Option<String> {
    let mut prompts = Vec::new();
    if let Some(prompt) = brief_reply_developer_prompt(session) {
        prompts.push(prompt.to_string());
    }
    if let Some(prompt) = system_prompt
        .map(str::trim)
        .filter(|value| !value.is_empty())
    {
        prompts.push(prompt.to_string());
    }
    (!prompts.is_empty()).then(|| prompts.join("\n\n"))
}

fn combine_developer_instructions(
    base: Option<String>,
    migration: Option<String>,
) -> Option<String> {
    match (base, migration) {
        (Some(base), Some(migration)) => Some(format!("{base}\n\n{migration}")),
        (Some(base), None) => Some(base),
        (None, Some(migration)) => Some(migration),
        (None, None) => None,
    }
}

fn build_context_migration_instructions(
    messages: &[ChatMessage],
    current_input: &ChatMessage,
    pending_reply: &ChatMessage,
) -> Option<String> {
    const MAX_MESSAGES: usize = 12;
    const MAX_CHARS_PER_MESSAGE: usize = 600;
    const MAX_TOTAL_CHARS: usize = 4000;

    let mut selected = messages
        .iter()
        .filter(|message| message.id != current_input.id && message.id != pending_reply.id)
        .filter_map(|message| {
            let content = message.content.trim();
            if content.is_empty() {
                return None;
            }
            let role = match message.role {
                crate::models::MessageRole::User => "用户",
                crate::models::MessageRole::Assistant => "助手",
                crate::models::MessageRole::System => "系统",
            };
            let truncated = truncate_chars(content, MAX_CHARS_PER_MESSAGE);
            Some(format!("{role}: {truncated}"))
        })
        .collect::<Vec<_>>();

    if selected.is_empty() {
        return None;
    }

    if selected.len() > MAX_MESSAGES {
        selected = selected.split_off(selected.len() - MAX_MESSAGES);
    }

    let mut body = String::new();
    for line in selected {
        let extra_len = if body.is_empty() {
            line.chars().count()
        } else {
            line.chars().count() + 1
        };
        if body.chars().count() + extra_len > MAX_TOTAL_CHARS {
            break;
        }
        if !body.is_empty() {
            body.push('\n');
        }
        body.push_str(&line);
    }

    if body.is_empty() {
        return None;
    }

    Some(format!(
        "这是一次模型或提供方切换后的新线程。请把下面历史对话当作延续上下文，仅用于理解先前约束、决策和未完成事项，不要重复复述给用户。\n\n历史对话摘录：\n{body}"
    ))
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum CodexThreadDecision {
    StartFresh,
    Resume,
    StartWithMigration,
}

fn decide_codex_thread_action(
    existing_runtime_ref: Option<&str>,
    stored_provider_name: Option<&str>,
    current_provider_name: Option<&str>,
) -> CodexThreadDecision {
    if existing_runtime_ref.is_none() {
        return CodexThreadDecision::StartFresh;
    }

    if stored_provider_name == current_provider_name {
        CodexThreadDecision::Resume
    } else {
        CodexThreadDecision::StartWithMigration
    }
}

fn truncate_chars(text: &str, max_chars: usize) -> String {
    let mut result = String::new();
    for (index, ch) in text.chars().enumerate() {
        if index >= max_chars {
            result.push_str("...");
            break;
        }
        result.push(ch);
    }
    result
}

fn codex_binary_path() -> PathBuf {
    if let Some(path) = std::env::var_os("ECHO_MATE_CODEX_BIN")
        .map(PathBuf::from)
        .filter(|path| !path.as_os_str().is_empty())
    {
        return path;
    }

    if let Some(path) = find_executable_in_path("codex") {
        return path;
    }

    let mut candidates = Vec::new();
    if let Some(home) = std::env::var_os("HOME").map(PathBuf::from) {
        candidates.push(home.join(".bun/bin/codex"));
        candidates.push(home.join(".local/bin/codex"));

        let node_versions = home.join(".nvm/versions/node");
        if let Ok(entries) = std::fs::read_dir(node_versions) {
            let mut codex_bins = entries
                .filter_map(Result::ok)
                .map(|entry| entry.path().join("bin/codex"))
                .filter(|path| path.is_file())
                .collect::<Vec<_>>();
            codex_bins.sort();
            candidates.extend(codex_bins.into_iter().rev());
        }
    }
    candidates.push(PathBuf::from("/usr/local/bin/codex"));
    candidates.push(PathBuf::from("/usr/bin/codex"));

    candidates
        .into_iter()
        .find(|path| path.is_file())
        .unwrap_or_else(|| PathBuf::from("codex"))
}

fn opencode_binary_path() -> PathBuf {
    if let Some(path) = std::env::var_os("ECHO_MATE_OPENCODE_BIN")
        .map(PathBuf::from)
        .filter(|path| !path.as_os_str().is_empty())
    {
        return path;
    }

    find_executable_in_path("opencode").unwrap_or_else(|| PathBuf::from("opencode"))
}

fn opencode_config_path() -> PathBuf {
    if let Some(home) = std::env::var_os("HOME").map(PathBuf::from) {
        home.join(".config/opencode/opencode.json")
    } else {
        PathBuf::from(".config/opencode/opencode.json")
    }
}

/// Create a temporary opencode config file with the given provider's base_url.
/// Returns the path to the temp config file.
fn write_opencode_overlay_config(config: &ResolvedProviderConfig) -> Result<PathBuf> {
    let original_path = opencode_config_path();
    let original = std::fs::read_to_string(&original_path)
        .unwrap_or_else(|_| r#"{"provider":{}}"#.to_string());
    let modified = modify_opencode_config(&original, config);
    let overlay = crate::bridge_settings::project_tmp_dir("opencode-overlay");
    std::fs::create_dir_all(&overlay)?;
    let path = overlay.join("opencode.json");
    std::fs::write(&path, &modified)?;
    Ok(path)
}

/// Modify opencode config: find a provider with matching base_url AND npm package, and update it,
/// or create a new provider if none found.
fn modify_opencode_config(original: &str, config: &ResolvedProviderConfig) -> String {
    let Ok(mut value) = serde_json::from_str::<serde_json::Value>(original) else {
        return original.to_string();
    };

    let providers = value.get_mut("provider").and_then(|v| v.as_object_mut());
    let Some(providers) = providers else {
        return original.to_string();
    };

    let npm = match config.format {
        crate::models::ApiFormat::AnthropicMessages => "@ai-sdk/anthropic",
        _ => "@ai-sdk/openai-compatible",
    };

    // Find provider to update: must match both base_url AND npm package
    // This prevents updating an OpenAI-compatible provider when we need Anthropic
    let target_key = providers
        .iter()
        .find(|(_, v)| {
            let base_url_match = v
                .get("options")
                .and_then(|o| o.get("baseURL"))
                .and_then(|u| u.as_str())
                == Some(&config.base_url);
            let npm_match = v.get("npm").and_then(|n| n.as_str()) == Some(npm);
            base_url_match && npm_match
        })
        .map(|(k, _)| k.clone());

    if let Some(key) = target_key {
        // Update existing provider with matching base_url and npm
        if let Some(provider) = providers.get_mut(&key).and_then(|v| v.as_object_mut()) {
            let options = provider
                .entry("options")
                .or_insert_with(|| serde_json::json!({}));
            if let Some(obj) = options.as_object_mut() {
                obj.insert("baseURL".to_string(), serde_json::json!(config.base_url));
                if !config.api_key.is_empty() {
                    obj.insert("apiKey".to_string(), serde_json::json!(config.api_key));
                }
            }
            provider.insert("npm".to_string(), serde_json::json!(npm));
        }
    } else {
        // Create new provider with a unique name based on npm package
        let provider_name = match config.format {
            crate::models::ApiFormat::AnthropicMessages => "omni-bridge-anthropic",
            crate::models::ApiFormat::Codex => "omni-bridge-codex",
            _ => "omni-bridge",
        };
        let mut options = serde_json::json!({ "baseURL": config.base_url });
        if !config.api_key.is_empty() {
            options["apiKey"] = serde_json::json!(config.api_key);
        }
        providers.insert(
            provider_name.to_string(),
            serde_json::json!({ "npm": npm, "options": options }),
        );
    }

    serde_json::to_string_pretty(&value).unwrap_or_else(|_| original.to_string())
}

struct OpenCodeHttpClient {
    base_url: String,
    http: reqwest::Client,
    child: Child,
    stderr_task: tokio::task::JoinHandle<std::result::Result<String, std::io::Error>>,
}

impl OpenCodeHttpClient {
    async fn start() -> Result<Self> {
        Self::start_with_config(None).await
    }

    async fn start_with_config(provider_config: Option<&ResolvedProviderConfig>) -> Result<Self> {
        let mut command = Command::new(opencode_binary_path());
        command
            .arg("serve")
            .arg("--hostname")
            .arg("127.0.0.1")
            .stdin(Stdio::null())
            .stdout(Stdio::piped())
            .stderr(Stdio::piped())
            .kill_on_drop(true);

        // Apply provider configuration via OPENCODE_CONFIG overlay file
        if let Some(config) = provider_config {
            match write_opencode_overlay_config(config) {
                Ok(path) => {
                    command.env("OPENCODE_CONFIG", &path);
                    eprintln!(
                        "[opencode] using overlay config: base_url={}",
                        config.base_url
                    );
                }
                Err(err) => {
                    eprintln!("[opencode] warning: failed to create overlay config: {err}");
                }
            }
            if let Some(ref model) = config.model {
                command.env("OPENCODE_MODEL", model);
            }
        }

        let mut child = command
            .spawn()
            .context("failed to spawn `opencode serve`")?;
        let stdout = child
            .stdout
            .take()
            .context("opencode server did not expose stdout")?;
        let stderr = child
            .stderr
            .take()
            .context("opencode server did not expose stderr")?;
        let stderr_task = tokio::spawn(async move {
            let mut reader = BufReader::new(stderr).lines();
            let mut output = String::new();
            while let Some(line) = reader.next_line().await? {
                if !output.is_empty() {
                    output.push('\n');
                }
                output.push_str(&strip_ansi(&line));
            }
            Ok::<_, std::io::Error>(output)
        });
        let base_url = read_opencode_server_url(stdout).await?;
        Ok(Self {
            base_url,
            http: reqwest::Client::new(),
            child,
            stderr_task,
        })
    }

    async fn shutdown(&mut self) {
        let _ = self.child.kill().await;
        self.stderr_task.abort();
    }

    async fn list_projects(&self) -> Result<HashMap<String, ProjectSummary>> {
        let sessions = self.get_json("/session").await?;
        let mut projects = HashMap::new();
        for session in sessions.as_array().into_iter().flatten() {
            let Some(directory) = session.get("directory").and_then(Value::as_str) else {
                continue;
            };
            let project_id = crate::session_store::project_id_for_path(directory);
            let updated_at = opencode_time(session.pointer("/time/updated"));
            let preview = session
                .get("title")
                .and_then(Value::as_str)
                .map(ToString::to_string);
            projects
                .entry(project_id.clone())
                .and_modify(|project: &mut ProjectSummary| {
                    project.session_count += 1;
                    if updated_at > project.updated_at {
                        project.updated_at = updated_at;
                        project.last_session_preview = preview.clone();
                    }
                })
                .or_insert_with(|| ProjectSummary {
                    id: project_id,
                    name: project_name_from_path(directory),
                    root_path: directory.to_string(),
                    updated_at,
                    session_count: 1,
                    last_session_preview: preview,
                    git_branch: None,
                    git_status: None,
                });
        }
        Ok(projects)
    }

    async fn list_sessions(&self) -> Result<HashMap<String, SessionSummary>> {
        let sessions = self.get_json("/session").await?;
        Ok(sessions
            .as_array()
            .into_iter()
            .flatten()
            .filter_map(opencode_session_summary)
            .map(|session| (session.id.clone(), session))
            .collect())
    }

    async fn list_messages(&self, session_id: &str) -> Result<Vec<ChatMessage>> {
        let value = self
            .get_json(&format!("/session/{}/message", url_path_escape(session_id)))
            .await?;
        Ok(opencode_messages_from_value(session_id, &value))
    }

    async fn latest_assistant_text(&self, session_id: &str, project_root: &Path) -> Result<String> {
        let value = self
            .get_json_with_query(
                &format!("/session/{}/message", url_path_escape(session_id)),
                &[("directory", project_root.display().to_string())],
            )
            .await?;
        Ok(opencode_messages_from_value(session_id, &value)
            .into_iter()
            .rev()
            .find(|message| matches!(message.role, crate::models::MessageRole::Assistant))
            .map(|message| message.content)
            .unwrap_or_default())
    }

    async fn create_session(&self, project_root: &Path, title: &str) -> Result<String> {
        let value = self
            .post_json_with_query(
                "/session",
                &[("directory", project_root.display().to_string())],
                serde_json::json!({
                    "title": title,
                    "permission": [{
                        "permission": "bash",
                        "pattern": "*",
                        "action": "ask",
                    }],
                }),
            )
            .await?;
        value
            .get("id")
            .and_then(Value::as_str)
            .map(ToString::to_string)
            .context("opencode session create did not return id")
    }

    async fn prompt_async(
        &self,
        session_id: &str,
        project_root: &Path,
        prompt: &str,
        system: Option<&str>,
        model: Option<&str>,
        provider_id: Option<&str>,
    ) -> Result<()> {
        let mut body = serde_json::json!({
            "parts": [{
                "type": "text",
                "text": prompt,
            }],
        });
        if let Some(system) = system {
            body["system"] = serde_json::Value::String(system.to_string());
        }
        if let Some(model) = model {
            // opencode expects model as an object with provider and model fields
            // Use the provider_id if provided, otherwise default to "omni-bridge"
            // Note: The provider name must match what modify_opencode_config creates
            let provider = provider_id.unwrap_or("omni-bridge");
            body["model"] = serde_json::json!({
                "provider": provider,
                "model": model,
            });
        }
        self.post_json_with_query(
            &format!("/session/{}/prompt_async", url_path_escape(session_id)),
            &[("directory", project_root.display().to_string())],
            body,
        )
        .await?;
        Ok(())
    }

    async fn reply_permission(&self, request_id: &str, choice: &ApprovalChoice) -> Result<()> {
        let reply = match choice {
            ApprovalChoice::Accept => "once",
            ApprovalChoice::AcceptForSession | ApprovalChoice::AlwaysAllow => "always",
            ApprovalChoice::Decline | ApprovalChoice::Cancel => "reject",
        };
        self.post_json(
            &format!("/permission/{}/reply", url_path_escape(request_id)),
            serde_json::json!({ "reply": reply }),
        )
        .await?;
        Ok(())
    }

    async fn event_stream(
        &self,
        project_root: &Path,
    ) -> Result<mpsc::UnboundedReceiver<Result<String>>> {
        let response = self
            .http
            .get(format!("{}/event", self.base_url))
            .query(&[("directory", project_root.display().to_string())])
            .send()
            .await
            .context("failed to connect opencode event stream")?
            .error_for_status()
            .context("opencode event stream failed")?;
        let (tx, rx) = mpsc::unbounded_channel();
        tokio::spawn(async move {
            let mut stream = response.bytes_stream();
            let mut buffer = Vec::<u8>::new();
            while let Some(chunk) = stream.next().await {
                match chunk {
                    Ok(bytes) => {
                        buffer.extend_from_slice(&bytes);
                        while let Some(index) = buffer.iter().position(|byte| *byte == b'\n') {
                            let line = buffer.drain(..=index).collect::<Vec<_>>();
                            let line = String::from_utf8_lossy(&line).trim().to_string();
                            let _ = tx.send(Ok(line));
                        }
                    }
                    Err(error) => {
                        let _ = tx.send(Err(error.into()));
                        return;
                    }
                }
            }
        });
        Ok(rx)
    }

    async fn get_json(&self, path: &str) -> Result<Value> {
        self.get_json_with_query(path, &[]).await
    }

    async fn get_json_with_query(&self, path: &str, query: &[(&str, String)]) -> Result<Value> {
        let response = self
            .http
            .get(format!("{}{}", self.base_url, path))
            .query(query)
            .send()
            .await?
            .error_for_status()?;
        Ok(response.json().await?)
    }

    async fn post_json(&self, path: &str, body: Value) -> Result<Value> {
        self.post_json_with_query(path, &[], body).await
    }

    async fn post_json_with_query(
        &self,
        path: &str,
        query: &[(&str, String)],
        body: Value,
    ) -> Result<Value> {
        let url = format!("{}{}", self.base_url, path);
        let response = self.http.post(&url).query(query).json(&body).send().await?;
        let status = response.status();
        if status.is_success() {
            Ok(response.json().await.unwrap_or(Value::Null))
        } else {
            let body_text = response.text().await.unwrap_or_default();
            eprintln!("[opencode] HTTP {status} {url} body={body_text}");
            bail!("opencode HTTP {status}: {body_text}");
        }
    }
}

async fn read_opencode_server_url(stdout: impl AsyncRead + Unpin) -> Result<String> {
    let mut lines = BufReader::new(stdout).lines();
    let deadline = tokio::time::sleep(Duration::from_secs(10));
    tokio::pin!(deadline);
    loop {
        tokio::select! {
            line = lines.next_line() => {
                let Some(line) = line? else {
                    bail!("opencode server exited before printing URL");
                };
                if let Some(url) = extract_first_http_url(&strip_ansi(&line)) {
                    return Ok(url);
                }
            }
            _ = &mut deadline => {
                bail!("timed out waiting for opencode server URL");
            }
        }
    }
}

fn extract_first_http_url(text: &str) -> Option<String> {
    text.split_whitespace()
        .find(|part| part.starts_with("http://") || part.starts_with("https://"))
        .map(|part| part.trim_end_matches('/').to_string())
}

pub fn find_executable_in_path(name: &str) -> Option<PathBuf> {
    let paths = std::env::var_os("PATH")?;
    std::env::split_paths(&paths)
        .map(|dir| dir.join(name))
        .find(|path| path.is_file())
}

async fn run_codex(
    state: Arc<AppState>,
    session: &SessionSummary,
    input: &ChatMessage,
    system_prompt: Option<&str>,
    reply: &ChatMessage,
    provider_config: Option<ResolvedProviderConfig>,
) -> Result<()> {
    let project_root = state
        .project_root_path_for_session(&session.id)
        .await
        .map(PathBuf::from)
        .map_err(anyhow::Error::msg)?;
    // Resolve the codex provider name for this session.
    // When the bridge explicitly resolved a provider_id (message-level or session-level),
    // use it directly as the codex provider name — the -c overrides will create the
    // matching model_providers entry. This avoids mismatches where the fallback chain
    // picks a different provider name than what the bridge actually configured.
    //
    // Fallback chain when the bridge did not inject a provider config:
    // 1. Codex session data file (model_provider in session_meta)
    // 2. Current config.toml's model_provider
    // 3. No override
    let codex_provider_name = match provider_config.as_ref().and_then(|c| c.provider_id.clone()) {
        Some(id) => Some(id),
        None => read_codex_session_provider(&session.id).or_else(find_codex_provider_name),
    };
    let mut child = spawn_codex_app_server_with_config(
        &project_root,
        provider_config.as_ref(),
        codex_provider_name.as_deref(),
    )
    .context("failed to spawn `codex app-server`")?;

    let mut stdin = child
        .stdin
        .take()
        .context("codex process did not expose stdin")?;
    let stdout = child
        .stdout
        .take()
        .context("codex process did not expose stdout")?;
    let stderr = child
        .stderr
        .take()
        .context("codex process did not expose stderr")?;

    let stderr_task = tokio::spawn(async move {
        let mut reader = BufReader::new(stderr).lines();
        let mut output = String::new();
        while let Some(line) = reader.next_line().await? {
            eprintln!("[codex:stderr] {line}");
            if !output.is_empty() {
                output.push('\n');
            }
            output.push_str(&line);
        }
        Ok::<_, std::io::Error>(output)
    });

    let (stdout_tx, mut stdout_rx) = mpsc::unbounded_channel();
    tokio::spawn(async move {
        let mut lines = BufReader::new(stdout).lines();
        loop {
            match lines.next_line().await {
                Ok(Some(line)) => {
                    let _ = stdout_tx.send(Ok(line));
                }
                Ok(None) => break,
                Err(error) => {
                    let _ = stdout_tx.send(Err(error.to_string()));
                    break;
                }
            }
        }
    });

    let (approval_tx, mut approval_rx) = mpsc::unbounded_channel();
    state.set_approval_sender(&session.id, approval_tx).await;

    let mut next_request_id = 1_u64;
    let mut raw_stdout = String::new();
    let mut parsed = CodexAppServerStreamingState::default();

    send_json_rpc_request(
        &mut stdin,
        &mut next_request_id,
        "initialize",
        serde_json::json!({
            "clientInfo": {
                "name": "omni-code-bridge",
                "title": "omni-code-bridge",
                "version": env!("CARGO_PKG_VERSION"),
            },
            "capabilities": {
                "experimentalApi": true,
            }
        }),
    )
    .await?;
    wait_for_json_rpc_response(&mut stdout_rx, &mut raw_stdout, next_request_id - 1).await?;

    let current_model = provider_config.as_ref().and_then(|c| c.model.clone());
    let stored_model = state.codex_model(&session.id).await;
    let existing_runtime_ref = state.provider_session_ref(&session.id).await;
    let stored_provider_name = state.codex_provider_name(&session.id).await;
    let thread_decision = decide_codex_thread_action(
        existing_runtime_ref.as_deref(),
        stored_provider_name.as_deref(),
        codex_provider_name.as_deref(),
    );
    let migration_instructions = if thread_decision == CodexThreadDecision::StartWithMigration {
        state
            .list_messages(&session.id)
            .await
            .and_then(|messages| build_context_migration_instructions(&messages, input, reply))
    } else {
        None
    };
    let base_developer_instructions = turn_system_prompt(session, system_prompt);
    let developer_instructions =
        combine_developer_instructions(base_developer_instructions.clone(), migration_instructions);

    if thread_decision == CodexThreadDecision::StartWithMigration {
        eprintln!(
            "[codex] not resuming thread for session={}: provider changed old_provider={:?} new_provider={:?} old_model={:?} new_model={:?}",
            session.id, stored_provider_name, codex_provider_name, stored_model, current_model,
        );
        state.set_provider_session_ref(&session.id, None).await;
        state.set_codex_provider_name(&session.id, None).await;
        state.set_codex_model(&session.id, None).await;
    } else if thread_decision == CodexThreadDecision::Resume && stored_model != current_model {
        eprintln!(
            "[codex] attempting thread/resume across model change for session={}: provider={:?} old_model={:?} new_model={:?}",
            session.id, codex_provider_name, stored_model, current_model,
        );
    }
    let existing_runtime_ref = if thread_decision == CodexThreadDecision::Resume {
        existing_runtime_ref
    } else {
        None
    };

    // Try to resume existing thread; if it fails (e.g. provider mismatch), fall back to thread/start.
    let thread_response = if let Some(thread_id) = existing_runtime_ref.as_deref() {
        let resume_id = send_json_rpc_request(
            &mut stdin,
            &mut next_request_id,
            "thread/resume",
            serde_json::json!({
                "threadId": thread_id,
                "approvalPolicy": "on-request",
                "approvalsReviewer": "user",
                "sandbox": "workspace-write",
                "persistExtendedHistory": false,
                "developerInstructions": developer_instructions,
            }),
        )
        .await?;
        match wait_for_json_rpc_response(&mut stdout_rx, &mut raw_stdout, resume_id).await {
            Ok(resp) if resp.get("error").is_none() => resp,
            Ok(resp) => {
                eprintln!(
                    "[codex] thread/resume failed, falling back to thread/start: {}",
                    resp
                );
                state.set_codex_provider_name(&session.id, None).await;
                state.set_codex_model(&session.id, None).await;
                let fallback_developer_instructions = combine_developer_instructions(
                    base_developer_instructions.clone(),
                    state.list_messages(&session.id).await.and_then(|messages| {
                        build_context_migration_instructions(&messages, input, reply)
                    }),
                );
                start_codex_thread(
                    &mut stdin,
                    &mut next_request_id,
                    &mut stdout_rx,
                    &mut raw_stdout,
                    &state,
                    &session,
                    &fallback_developer_instructions,
                )
                .await?
            }
            Err(e) => {
                eprintln!("[codex] thread/resume failed, falling back to thread/start: {e}");
                state.set_codex_provider_name(&session.id, None).await;
                state.set_codex_model(&session.id, None).await;
                let fallback_developer_instructions = combine_developer_instructions(
                    base_developer_instructions.clone(),
                    state.list_messages(&session.id).await.and_then(|messages| {
                        build_context_migration_instructions(&messages, input, reply)
                    }),
                );
                start_codex_thread(
                    &mut stdin,
                    &mut next_request_id,
                    &mut stdout_rx,
                    &mut raw_stdout,
                    &state,
                    &session,
                    &fallback_developer_instructions,
                )
                .await?
            }
        }
    } else {
        start_codex_thread(
            &mut stdin,
            &mut next_request_id,
            &mut stdout_rx,
            &mut raw_stdout,
            &state,
            &session,
            &developer_instructions,
        )
        .await?
    };
    log_codex_thread_response(&session.id, &thread_response);
    let thread_id = thread_response
        .pointer("/thread/id")
        .and_then(Value::as_str)
        .map(ToString::to_string)
        .or_else(|| {
            thread_response
                .pointer("/thread/threadId")
                .and_then(Value::as_str)
                .map(ToString::to_string)
        })
        .or(existing_runtime_ref)
        .context("codex app-server did not provide a thread id")?;
    state
        .set_provider_session_ref(&session.id, Some(thread_id.clone()))
        .await;
    // Store the provider name only for bridge-managed provider sessions.
    if provider_config.is_some() {
        let name = codex_provider_name.unwrap_or_else(|| "omni-bridge".to_string());
        state.set_codex_provider_name(&session.id, Some(name)).await;
        state.set_codex_model(&session.id, current_model).await;
    }
    parsed.session_id = Some(thread_id.clone());

    let turn_request_id = send_json_rpc_request(
        &mut stdin,
        &mut next_request_id,
        "turn/start",
        serde_json::json!({
            "threadId": thread_id,
            "input": [{
                "type": "text",
                "text": input.content,
                "text_elements": [],
            }],
        }),
    )
    .await?;

    let idle_deadline = Duration::from_secs(15);
    let idle_sleep = tokio::time::sleep(idle_deadline);
    tokio::pin!(idle_sleep);
    let mut pending_approval: Option<PendingApproval> = None;
    let mut queued_approvals: VecDeque<PendingApproval> = VecDeque::new();
    let mut turn_finished = false;
    let mut last_rendered = String::new();

    loop {
        tokio::select! {
            maybe_line = stdout_rx.recv() => {
                let Some(line) = maybe_line else {
                    break;
                };
                let line = line.map_err(anyhow::Error::msg)?;
                eprintln!("[codex:stdout] {line}");
                idle_sleep.as_mut().reset(tokio::time::Instant::now() + idle_deadline);
                if !raw_stdout.is_empty() {
                    raw_stdout.push('\n');
                }
                raw_stdout.push_str(&line);

                let value = serde_json::from_str::<Value>(&line)
                    .unwrap_or_else(|_| serde_json::json!({ "raw_line": line }));
                if value.get("id").and_then(jsonrpc_id_to_string).as_deref()
                    == Some(&turn_request_id.to_string())
                {
                    if let Some(error) = value.get("error") {
                        let message = error
                            .get("message")
                            .and_then(Value::as_str)
                            .unwrap_or("codex turn/start failed");
                        bail!("{message}");
                    }
                    continue;
                }

                let previous_status = parsed.current_status().map(ToString::to_string);
                match parsed.ingest_value(&value) {
                    CodexAppServerEvent::None => {}
                    CodexAppServerEvent::Content(text) => {
                        push_incremental_text(&state, &session.id, &reply.id, &mut last_rendered, &text).await;
                    }
                    CodexAppServerEvent::Status(status) => {
                        parsed.latest_status = Some(status);
                    }
                    CodexAppServerEvent::ApprovalRequested(pending) => {
                        if let Some(choice) = auto_approve_codex_request(&pending.request, &project_root).await? {
                            send_json_rpc_response(
                                &mut stdin,
                                &pending.request.request_id,
                                approval_result_json(&choice, &pending.request.kind),
                            )
                            .await?;
                        } else {
                            if pending_approval.is_some() {
                                queued_approvals.push_back(pending);
                            } else {
                                state.raise_approval(&session.id, pending.request.clone()).await;
                                pending_approval = Some(pending);
                            }
                        }
                    }
                    CodexAppServerEvent::ApprovalResolved { request_id } => {
                        let resolved_current = pending_approval
                            .as_ref()
                            .map(|pending| pending.request.request_id == request_id)
                            .unwrap_or(false);
                        if resolved_current {
                            let pending = pending_approval
                                .take()
                                .context("approval state disappeared unexpectedly")?;
                            let choice = pending.last_choice.unwrap_or(ApprovalChoice::Accept);
                            state.resolve_approval(&session.id, &request_id, choice).await;
                            if let Some(next) = queued_approvals.pop_front() {
                                state.raise_approval(&session.id, next.request.clone()).await;
                                pending_approval = Some(next);
                            }
                        } else if pending_approval.is_none() {
                            state.resolve_approval(&session.id, &request_id, ApprovalChoice::Accept).await;
                        } else {
                            eprintln!(
                                "[approval] ignoring resolved request for non-current approval session={} request={}",
                                session.id, request_id
                            );
                        }
                    }
                    CodexAppServerEvent::TurnCompleted => {
                        turn_finished = true;
                        break;
                    }
                    CodexAppServerEvent::TurnFailed(message) => {
                        bail!("{message}");
                    }
                }
                let next_status = parsed.current_status().map(ToString::to_string);
                if next_status.is_some() && next_status != previous_status {
                    state
                        .emit_system_message(&session.id, next_status.unwrap_or_default())
                        .await;
                }
            }
            Some(choice) = approval_rx.recv(), if pending_approval.is_some() => {
                let request = pending_approval
                    .as_mut()
                    .context("approval state disappeared unexpectedly")?;
                request.last_choice = Some(choice.clone());
                send_json_rpc_response(
                    &mut stdin,
                    &request.request.request_id,
                    approval_result_json(&choice, &request.request.kind),
                )
                .await?;
            }
            _ = &mut idle_sleep => {
                let previous_status = parsed.current_status().map(ToString::to_string);
                if let Some(text) = parsed.mark_idle_waiting() {
                    push_incremental_text(&state, &session.id, &reply.id, &mut last_rendered, &text).await;
                }
                let next_status = parsed.current_status().map(ToString::to_string);
                if next_status.is_some() && next_status != previous_status {
                    state
                        .emit_system_message(&session.id, next_status.unwrap_or_default())
                        .await;
                }
                idle_sleep.as_mut().reset(tokio::time::Instant::now() + idle_deadline);
            }
        }
    }

    let text = parsed
        .finish_text()
        .or_else(|| fallback_text(&raw_stdout))
        .context("codex response did not include assistant text")?;
    push_incremental_text(&state, &session.id, &reply.id, &mut last_rendered, &text).await;

    if turn_finished {
        stderr_task.abort();
        state
            .finish_assistant_message(&session.id, &reply.id)
            .await
            .map_err(anyhow::Error::msg)?;
        return Ok(());
    }

    let status = child.wait().await?;
    let stderr = stderr_task
        .await
        .context("failed to join codex stderr reader")??;
    if !status.success() {
        let code = status.code().unwrap_or(1);
        let stderr = stderr.trim();
        if stderr.is_empty() {
            bail!("codex exited with status {code}");
        } else {
            // stderr may contain retry messages like "Reconnecting... 1/5".
            // Use the last line as the actual error message.
            let last_line = stderr.lines().last().unwrap_or(stderr);
            bail!("codex exited with status {code}: {last_line}");
        }
    }

    state
        .finish_assistant_message(&session.id, &reply.id)
        .await
        .map_err(anyhow::Error::msg)?;
    Ok(())
}

async fn run_claude_code(
    state: Arc<AppState>,
    session: &SessionSummary,
    input: &ChatMessage,
    system_prompt: Option<&str>,
    reply: &ChatMessage,
    provider_config: Option<ResolvedProviderConfig>,
) -> Result<()> {
    let state_dir = claude_state_dir();
    ensure_runtime_dirs(&state_dir).await?;
    let project_root = state
        .project_root_path_for_session(&session.id)
        .await
        .map(PathBuf::from)
        .map_err(anyhow::Error::msg)?;
    let current_provider_id = provider_config.as_ref().and_then(|c| c.provider_id.clone());
    let current_model = provider_config.as_ref().and_then(|c| c.model.clone());
    let stored_provider_id = state.claude_provider_id(&session.id).await;
    let stored_model = state.claude_model(&session.id).await;
    let existing_runtime_ref = state.provider_session_ref(&session.id).await;
    let can_resume_existing_session = existing_runtime_ref.is_some()
        && stored_provider_id == current_provider_id
        && stored_model == current_model;
    let migration_instructions = if existing_runtime_ref.is_some() && !can_resume_existing_session {
        state
            .list_messages(&session.id)
            .await
            .and_then(|messages| build_context_migration_instructions(&messages, input, reply))
    } else {
        None
    };
    if existing_runtime_ref.is_some() && !can_resume_existing_session {
        eprintln!(
            "[claude] not resuming session for session={}: provider/model changed old_provider={:?} new_provider={:?} old_model={:?} new_model={:?}",
            session.id, stored_provider_id, current_provider_id, stored_model, current_model,
        );
        state.set_provider_session_ref(&session.id, None).await;
        state.set_claude_provider_id(&session.id, None).await;
        state.set_claude_model(&session.id, None).await;
    }
    let runtime_ref = select_claude_runtime_ref(
        &session.id,
        existing_runtime_ref.as_deref(),
        can_resume_existing_session,
    );
    state
        .set_provider_session_ref(&session.id, Some(runtime_ref.clone()))
        .await;
    let run_id = uuid::Uuid::new_v4().to_string();
    let current_exe = std::env::current_exe().context("failed to resolve current executable")?;
    let hook_command = format!(
        "{} claude-permission-hook --state-dir {} --session-id {} --run-id {}",
        shell_quote(current_exe.to_string_lossy().as_ref()),
        shell_quote(state_dir.display().to_string().as_str()),
        shell_quote(session.id.as_str()),
        shell_quote(run_id.as_str()),
    );
    let hook_command = format!(
        "{hook_command} --project-root {}",
        shell_quote(project_root.display().to_string().as_str())
    );
    // Build settings, including env overrides to suppress global ~/.claude/settings.json env
    let mut settings_value = serde_json::json!({
        "permissions": {
            "allow": ["Bash(*)", "WebSearch(*)", "WebFetch(*)"],
            "defaultMode": "acceptEdits"
        },
        "hooks": {
            "PreToolUse": [{
                "hooks": [{"type": "command", "command": hook_command}]
            }],
            "PostToolUse": [{
                "hooks": [{"type": "command", "command": hook_command}]
            }]
        }
    });

    // Apply provider configuration
    if let Some(ref config) = provider_config {
        // Inject env overrides into settings so they take precedence over
        // ~/.claude/settings.json env (ANTHROPIC_AUTH_TOKEN, ANTHROPIC_BASE_URL, etc.)
        let mut env_overrides = serde_json::Map::new();
        if !config.api_key.is_empty() {
            env_overrides.insert(
                "ANTHROPIC_AUTH_TOKEN".to_string(),
                serde_json::json!(config.api_key),
            );
            env_overrides.insert(
                "ANTHROPIC_API_KEY".to_string(),
                serde_json::json!(config.api_key),
            );
        }
        if !config.base_url.is_empty() {
            env_overrides.insert(
                "ANTHROPIC_BASE_URL".to_string(),
                serde_json::json!(config.base_url),
            );
        }
        if !env_overrides.is_empty() {
            settings_value["env"] = serde_json::Value::Object(env_overrides);
        }
    }

    let mut command = Command::new("claude");
    command
        .arg("-p")
        .arg("--verbose")
        .arg("--output-format")
        .arg("stream-json")
        .arg("--include-partial-messages")
        .arg("--settings")
        .arg(settings_value.to_string());

    // Also set env vars directly on the process as a fallback
    if let Some(ref config) = provider_config {
        if !config.api_key.is_empty() {
            command.env("ANTHROPIC_AUTH_TOKEN", &config.api_key);
            command.env("ANTHROPIC_API_KEY", &config.api_key);
        }
        if !config.base_url.is_empty() {
            command.env("ANTHROPIC_BASE_URL", &config.base_url);
        }
        if let Some(ref model) = config.model {
            command.arg("--model").arg(model);
        }
    }

    let combined_system_prompt = {
        let base = turn_system_prompt(session, system_prompt);
        match (base, migration_instructions) {
            (Some(base), Some(migration)) => Some(format!("{base}\n\n{migration}")),
            (Some(base), None) => Some(base),
            (None, Some(migration)) => Some(migration),
            (None, None) => None,
        }
    };
    if let Some(system_prompt) = combined_system_prompt {
        command.arg("--append-system-prompt").arg(system_prompt);
    }
    if can_resume_existing_session {
        command.arg("-r").arg(&runtime_ref);
    } else {
        command.arg("--session-id").arg(&runtime_ref);
    }
    command
        .arg(&input.content)
        .current_dir(&project_root)
        .stdin(Stdio::null())
        .stdout(Stdio::piped())
        .stderr(Stdio::piped())
        .kill_on_drop(true);

    let mut child = command.spawn().context("failed to spawn `claude`")?;
    state
        .set_claude_provider_id(&session.id, current_provider_id)
        .await;
    state.set_claude_model(&session.id, current_model).await;
    let stdout = child
        .stdout
        .take()
        .context("claude process did not expose stdout")?;
    let stderr = child
        .stderr
        .take()
        .context("claude process did not expose stderr")?;

    let stderr_task = tokio::spawn(async move {
        let mut reader = BufReader::new(stderr).lines();
        let mut output = String::new();
        while let Some(line) = reader.next_line().await? {
            if !output.is_empty() {
                output.push('\n');
            }
            output.push_str(&line);
        }
        Ok::<_, std::io::Error>(output)
    });

    let (approval_tx, mut approval_rx) = mpsc::unbounded_channel();
    state.set_approval_sender(&session.id, approval_tx).await;

    let mut last_rendered = String::new();
    let mut parsed = ClaudeStreamingState::default();
    let mut final_text = None;
    let mut reader = BufReader::new(stdout).lines();
    let mut pending_approval: Option<String> = None;
    let mut seen_permission_requests = std::collections::HashSet::new();
    let mut seen_status_events = std::collections::HashSet::new();
    let mut permission_poll = tokio::time::interval(Duration::from_millis(200));

    loop {
        tokio::select! {
            maybe_line = reader.next_line() => {
                let Some(line) = maybe_line? else {
                    break;
                };
                let clean_line = strip_ansi(&line);
                if clean_line.trim().is_empty() {
                    continue;
                }
                let previous_status = parsed.current_status().map(ToString::to_string);
                if let Some(text) = parsed.ingest_line(&clean_line)? {
                    final_text = parsed.finish_text();
                    push_incremental_text(&state, &session.id, &reply.id, &mut last_rendered, &text).await;
                }
                let next_status = parsed.current_status().map(ToString::to_string);
                if next_status.is_some() && next_status != previous_status {
                    state
                        .emit_system_message(&session.id, next_status.unwrap_or_default())
                        .await;
                }
            }
            _ = permission_poll.tick() => {
                if let Some(event) = next_claude_status_event(&state_dir, &run_id, &mut seen_status_events).await? {
                    let previous_status = parsed.current_status().map(ToString::to_string);
                    if let Some(text) = parsed.ingest_external_status(event.summary) {
                        push_incremental_text(&state, &session.id, &reply.id, &mut last_rendered, &text).await;
                    }
                    let next_status = parsed.current_status().map(ToString::to_string);
                    if next_status.is_some() && next_status != previous_status {
                        state
                            .emit_system_message(&session.id, next_status.unwrap_or_default())
                            .await;
                    }
                }
                if pending_approval.is_none() {
                    if let Some(request) = next_claude_permission_request(&state_dir, &run_id, &mut seen_permission_requests).await? {
                        let approval = request.as_approval_request();
                        if let Some(choice) = auto_approve_codex_request(&approval, &project_root).await? {
                            let mut response = response_from_choice(&choice);
                            response.request_id = approval.request_id.clone();
                            tokio::fs::write(
                                response_path(&state_dir, &approval.request_id),
                                serde_json::to_vec_pretty(&response)?,
                            )
                            .await?;
                            state.resolve_approval(&session.id, &approval.request_id, choice).await;
                        } else {
                            pending_approval = Some(approval.request_id.clone());
                            state.raise_approval(&session.id, approval).await;
                        }
                    }
                }
            }
            Some(choice) = approval_rx.recv(), if pending_approval.is_some() => {
                let request_id = pending_approval.clone().context("approval state disappeared unexpectedly")?;
                let mut response = response_from_choice(&choice);
                response.request_id = request_id.clone();
                if matches!(choice, ApprovalChoice::AlwaysAllow) {
                    let runtime_request_path = request_path(&state_dir, &request_id);
                    if let Ok(body) = tokio::fs::read(&runtime_request_path).await {
                        if let Ok(request) = serde_json::from_slice::<ClaudePermissionRequest>(&body) {
                            if let Some(command) = request.as_approval_request().command {
                                append_always_allow_command(&state_dir, &command).await?;
                            }
                        }
                    }
                }
                tokio::fs::write(
                    response_path(&state_dir, &request_id),
                    serde_json::to_vec_pretty(&response)?,
                )
                .await?;
                state.resolve_approval(&session.id, &request_id, choice).await;
                pending_approval = None;
            }
        }
    }

    let status = child.wait().await?;
    let stderr = stderr_task
        .await
        .context("failed to join claude stderr reader")??;

    if !status.success() {
        let code = status.code().unwrap_or(1);
        let stderr = stderr.trim();
        if stderr.is_empty() {
            bail!("claude exited with status {code}");
        } else {
            bail!("claude exited with status {code}: {stderr}");
        }
    }

    final_text = final_text.or_else(|| parsed.finish_text());
    if final_text.is_none() {
        bail!("claude response did not include assistant text");
    }

    state
        .finish_assistant_message(&session.id, &reply.id)
        .await
        .map_err(anyhow::Error::msg)?;
    Ok(())
}

async fn run_opencode(
    state: Arc<AppState>,
    session: &SessionSummary,
    input: &ChatMessage,
    system_prompt: Option<&str>,
    reply: &ChatMessage,
    provider_config: Option<ResolvedProviderConfig>,
) -> Result<()> {
    let project_root = state
        .project_root_path_for_session(&session.id)
        .await
        .map(PathBuf::from)
        .map_err(anyhow::Error::msg)?;

    let mut client = OpenCodeHttpClient::start_with_config(provider_config.as_ref()).await?;
    let opencode_session_id =
        if let Some(session_id) = state.provider_session_ref(&session.id).await {
            session_id
        } else {
            let session_id = client
                .create_session(&project_root, &session.title)
                .await
                .context("failed to create opencode session")?;
            state
                .set_provider_session_ref(&session.id, Some(session_id.clone()))
                .await;
            session_id
        };

    let mut last_rendered = String::new();
    let mut full_text = String::new();
    let (approval_tx, mut approval_rx) = mpsc::unbounded_channel();
    state.set_approval_sender(&session.id, approval_tx).await;
    let system = turn_system_prompt(session, system_prompt);
    let model = provider_config.as_ref().and_then(|c| c.model.as_deref());
    // Use opencode_provider_name which matches what modify_opencode_config creates
    let opencode_provider = provider_config
        .as_ref()
        .and_then(|c| c.opencode_provider_name.as_deref());
    client
        .prompt_async(
            &opencode_session_id,
            &project_root,
            &input.content,
            system.as_deref(),
            model,
            opencode_provider,
        )
        .await
        .context("failed to send opencode prompt")?;
    let mut event_stream = client
        .event_stream(&project_root)
        .await
        .context("failed to subscribe to opencode events")?;
    let mut pending_approval: Option<PendingApproval> = None;

    loop {
        tokio::select! {
            maybe_line = event_stream.recv() => {
                let Some(line) = maybe_line else {
                    break;
                };
                let line = line?;
                let Some(event) = parse_sse_json_event(&line) else {
                    continue;
                };
                let event_type = event.get("type").and_then(Value::as_str).unwrap_or_default();
                let properties = event.get("properties").unwrap_or(&Value::Null);
                let event_session_id = properties
                    .get("sessionID")
                    .and_then(Value::as_str)
                    .unwrap_or_default();
                if !event_session_id.is_empty() && event_session_id != opencode_session_id {
                    continue;
                }

                match event_type {
                    "message.part.delta" => {
                        let delta = properties.get("delta").and_then(Value::as_str).unwrap_or_default();
                        if properties.get("field").and_then(Value::as_str).unwrap_or("text") == "text"
                            && !delta.is_empty()
                        {
                            full_text.push_str(delta);
                            push_incremental_text(&state, &session.id, &reply.id, &mut last_rendered, &full_text).await;
                        }
                    }
                    "message.part.updated" => {
                        if let Some(status) = render_opencode_part_status(properties.get("part").unwrap_or(&Value::Null)) {
                            state.emit_system_message(&session.id, status).await;
                        }
                    }
                    "permission.asked" => {
                        let request = opencode_permission_to_approval(properties);
                        if let Some(choice) = auto_approve_opencode_request(&request, &project_root).await? {
                            client.reply_permission(&request.request_id, &choice).await?;
                            state.resolve_approval(&session.id, &request.request_id, choice).await;
                        } else {
                            pending_approval = Some(PendingApproval {
                                request: request.clone(),
                                last_choice: None,
                            });
                            state.raise_approval(&session.id, request).await;
                        }
                    }
                    "permission.replied" => {
                        let request_id = properties
                            .get("requestID")
                            .and_then(Value::as_str)
                            .unwrap_or_default();
                        if let Some(pending) = pending_approval.take() {
                            let choice = pending.last_choice.unwrap_or(ApprovalChoice::Accept);
                            state.resolve_approval(&session.id, request_id, choice).await;
                        }
                    }
                    "session.error" => {
                        bail!("{}", render_opencode_error(properties));
                    }
                    "session.idle" => break,
                    "session.status" => {
                        if let Some(status) = properties.pointer("/status/type").and_then(Value::as_str)
                            && status != "idle"
                        {
                            state.emit_system_message(&session.id, format!("[opencode] {status}")).await;
                        }
                    }
                    _ => {}
                }
            }
            Some(choice) = approval_rx.recv(), if pending_approval.is_some() => {
                let request = pending_approval
                    .as_mut()
                    .context("approval state disappeared unexpectedly")?;
                request.last_choice = Some(choice.clone());
                client.reply_permission(&request.request.request_id, &choice).await?;
            }
        }
    }

    if full_text.trim().is_empty() {
        full_text = client
            .latest_assistant_text(&opencode_session_id, &project_root)
            .await
            .unwrap_or_default();
    }
    let text = (!full_text.trim().is_empty())
        .then(|| full_text.trim().to_string())
        .context("opencode response did not include assistant text")?;
    push_incremental_text(&state, &session.id, &reply.id, &mut last_rendered, &text).await;
    client.shutdown().await;
    // Clean up overlay config if we created one
    if provider_config.is_some() {
        let overlay = crate::bridge_settings::project_tmp_dir("opencode-overlay");
        let _ = std::fs::remove_dir_all(&overlay);
    }
    state
        .finish_assistant_message(&session.id, &reply.id)
        .await
        .map_err(anyhow::Error::msg)?;
    Ok(())
}

async fn summarize_with_codex(
    state: Arc<AppState>,
    session: &SessionSummary,
    content: &str,
) -> Result<String> {
    let project_root = state
        .project_root_path_for_session(&session.id)
        .await
        .map(PathBuf::from)
        .map_err(anyhow::Error::msg)?;
    let mut child = spawn_codex_app_server(&project_root)
        .context("failed to spawn `codex app-server` for summary")?;

    let mut stdin = child
        .stdin
        .take()
        .context("codex summary process did not expose stdin")?;
    let stdout = child
        .stdout
        .take()
        .context("codex summary process did not expose stdout")?;
    let stderr = child
        .stderr
        .take()
        .context("codex summary process did not expose stderr")?;

    let stderr_task = tokio::spawn(async move {
        let mut reader = BufReader::new(stderr).lines();
        let mut output = String::new();
        while let Some(line) = reader.next_line().await? {
            if !output.is_empty() {
                output.push('\n');
            }
            output.push_str(&line);
        }
        Ok::<_, std::io::Error>(output)
    });

    let (stdout_tx, mut stdout_rx) = mpsc::unbounded_channel();
    tokio::spawn(async move {
        let mut lines = BufReader::new(stdout).lines();
        loop {
            match lines.next_line().await {
                Ok(Some(line)) => {
                    let _ = stdout_tx.send(Ok(line));
                }
                Ok(None) => break,
                Err(error) => {
                    let _ = stdout_tx.send(Err(error.to_string()));
                    break;
                }
            }
        }
    });

    let mut next_request_id = 1_u64;
    let mut raw_stdout = String::new();
    let mut parsed = CodexAppServerStreamingState::default();

    send_json_rpc_request(
        &mut stdin,
        &mut next_request_id,
        "initialize",
        serde_json::json!({
            "clientInfo": {
                "name": "omni-code-bridge",
                "title": "omni-code-bridge",
                "version": env!("CARGO_PKG_VERSION"),
            },
            "capabilities": {
                "experimentalApi": true,
            }
        }),
    )
    .await?;
    wait_for_json_rpc_response(&mut stdout_rx, &mut raw_stdout, next_request_id - 1).await?;

    let thread_request_id = send_json_rpc_request(
        &mut stdin,
        &mut next_request_id,
        "thread/start",
        serde_json::json!({
            "cwd": state
                .project_root_path_for_session(&session.id)
                .await
                .map_err(anyhow::Error::msg)?,
            "approvalPolicy": "on-request",
            "approvalsReviewer": "user",
            "sandbox": "workspace-write",
            "persistExtendedHistory": false,
        }),
    )
    .await?;
    let thread_response =
        wait_for_json_rpc_response(&mut stdout_rx, &mut raw_stdout, thread_request_id).await?;
    let thread_id = thread_response
        .pointer("/thread/id")
        .and_then(Value::as_str)
        .map(ToString::to_string)
        .or_else(|| {
            thread_response
                .pointer("/thread/threadId")
                .and_then(Value::as_str)
                .map(ToString::to_string)
        })
        .context("codex summary thread did not provide a thread id")?;

    let turn_request_id = send_json_rpc_request(
        &mut stdin,
        &mut next_request_id,
        "turn/start",
        serde_json::json!({
            "threadId": thread_id,
            "input": [{
                "type": "text",
                "text": summary_prompt(content),
                "text_elements": [],
            }],
        }),
    )
    .await?;

    loop {
        let Some(line) = stdout_rx.recv().await else {
            break;
        };
        let line = line.map_err(anyhow::Error::msg)?;
        if !raw_stdout.is_empty() {
            raw_stdout.push('\n');
        }
        raw_stdout.push_str(&line);

        let value = serde_json::from_str::<Value>(&line)
            .unwrap_or_else(|_| serde_json::json!({ "raw_line": line }));
        if value.get("id").and_then(jsonrpc_id_to_string).as_deref()
            == Some(&turn_request_id.to_string())
        {
            if let Some(error) = value.get("error") {
                let message = error
                    .get("message")
                    .and_then(Value::as_str)
                    .unwrap_or("codex summary failed");
                bail!("{message}");
            }
            continue;
        }

        match parsed.ingest_value(&value) {
            CodexAppServerEvent::TurnCompleted => break,
            CodexAppServerEvent::TurnFailed(message) => bail!("{message}"),
            CodexAppServerEvent::ApprovalRequested(_) => {
                bail!("codex summary unexpectedly requested approval")
            }
            _ => {}
        }
    }

    let status = child.wait().await?;
    let stderr = stderr_task
        .await
        .context("failed to join codex summary stderr reader")??;
    if !status.success() {
        let code = status.code().unwrap_or(1);
        let stderr = stderr.trim();
        if stderr.is_empty() {
            bail!("codex summary exited with status {code}");
        } else {
            bail!("codex summary exited with status {code}: {stderr}");
        }
    }

    let text = parsed
        .finish_text()
        .or_else(|| fallback_text(&raw_stdout))
        .context("codex summary did not include assistant text")?;
    Ok(normalize_summary_output(&text))
}

async fn summarize_with_claude_code(
    state: Arc<AppState>,
    session: &SessionSummary,
    content: &str,
) -> Result<String> {
    let project_root = state
        .project_root_path_for_session(&session.id)
        .await
        .map(PathBuf::from)
        .map_err(anyhow::Error::msg)?;

    let mut child = Command::new("claude");
    child
        .arg("-p")
        .arg("--output-format")
        .arg("stream-json")
        .arg(summary_prompt(content))
        .current_dir(project_root)
        .stdin(Stdio::null())
        .stdout(Stdio::piped())
        .stderr(Stdio::piped())
        .kill_on_drop(true);

    let mut child = child
        .spawn()
        .context("failed to spawn `claude` for summary")?;
    let stdout = child
        .stdout
        .take()
        .context("claude summary process did not expose stdout")?;
    let stderr = child
        .stderr
        .take()
        .context("claude summary process did not expose stderr")?;

    let stderr_task = tokio::spawn(async move {
        let mut reader = BufReader::new(stderr).lines();
        let mut output = String::new();
        while let Some(line) = reader.next_line().await? {
            if !output.is_empty() {
                output.push('\n');
            }
            output.push_str(&line);
        }
        Ok::<_, std::io::Error>(output)
    });

    let mut parsed = ClaudeStreamingState::default();
    let mut final_text = None;
    let mut reader = BufReader::new(stdout).lines();

    while let Some(line) = reader.next_line().await? {
        let clean_line = strip_ansi(&line);
        if clean_line.trim().is_empty() {
            continue;
        }
        if parsed.ingest_line(&clean_line)?.is_some() {
            final_text = parsed.finish_text();
        }
    }

    let status = child.wait().await?;
    let stderr = stderr_task
        .await
        .context("failed to join claude summary stderr reader")??;
    if !status.success() {
        let code = status.code().unwrap_or(1);
        let stderr = stderr.trim();
        if stderr.is_empty() {
            bail!("claude summary exited with status {code}");
        } else {
            bail!("claude summary exited with status {code}: {stderr}");
        }
    }

    let text = final_text
        .or_else(|| parsed.finish_text())
        .context("claude summary did not include assistant text")?;
    Ok(normalize_summary_output(&text))
}

async fn summarize_with_opencode(
    state: Arc<AppState>,
    session: &SessionSummary,
    content: &str,
) -> Result<String> {
    let project_root = state
        .project_root_path_for_session(&session.id)
        .await
        .map(PathBuf::from)
        .map_err(anyhow::Error::msg)?;
    let mut client = OpenCodeHttpClient::start().await?;
    let session_id = client
        .create_session(&project_root, "omni-code summary")
        .await
        .context("failed to create opencode summary session")?;
    client
        .prompt_async(
            &session_id,
            &project_root,
            &summary_prompt(content),
            None,
            None,
            None,
        )
        .await
        .context("failed to send opencode summary prompt")?;
    wait_for_opencode_session_idle(&mut client, &session_id, &project_root).await?;
    let text = client
        .latest_assistant_text(&session_id, &project_root)
        .await
        .context("opencode summary did not include assistant text")?;
    client.shutdown().await;
    Ok(normalize_summary_output(&text))
}

fn summary_prompt(content: &str) -> String {
    format!(
        "请把下面这段 AI 回复压缩成 50 到 60 个汉字，只输出最终摘要，不要标题、不要引号、不要列表、不要解释、不要补充新信息。\n\n原文如下：\n{content}"
    )
}

fn normalize_summary_output(text: &str) -> String {
    text.lines()
        .map(str::trim)
        .find(|line| !line.is_empty())
        .unwrap_or_default()
        .trim_matches('"')
        .trim_matches('“')
        .trim_matches('”')
        .trim()
        .to_string()
}

async fn push_incremental_text(
    state: &AppState,
    session_id: &str,
    message_id: &str,
    last_rendered: &mut String,
    next_text: &str,
) {
    if let Some(delta) = next_text.strip_prefix(last_rendered.as_str()) {
        if !delta.is_empty() {
            state
                .emit_message_delta(session_id, message_id, delta)
                .await;
            *last_rendered = next_text.to_string();
        }
    }
}

#[derive(Default)]
struct CodexAppServerStreamingState {
    display_blocks: Vec<String>,
    assistant_blocks: Vec<String>,
    partial_agent_messages: HashMap<String, String>,
    latest_status: Option<String>,
    session_id: Option<String>,
}

enum CodexAppServerEvent {
    None,
    Content(String),
    Status(String),
    ApprovalRequested(PendingApproval),
    ApprovalResolved { request_id: String },
    TurnCompleted,
    TurnFailed(String),
}

#[derive(Clone)]
struct PendingApproval {
    request: ApprovalRequest,
    last_choice: Option<ApprovalChoice>,
}

impl CodexAppServerStreamingState {
    fn current_status(&self) -> Option<&str> {
        self.latest_status.as_deref()
    }

    fn ingest_value(&mut self, value: &Value) -> CodexAppServerEvent {
        let Some(map) = value.as_object() else {
            return CodexAppServerEvent::None;
        };

        if map.get("id").is_some() && map.get("method").is_some() {
            return self.ingest_app_server_request(value);
        }

        if let Some(method) = map.get("method").and_then(Value::as_str) {
            let params = map.get("params").unwrap_or(&Value::Null);
            return match method {
                "thread/started" => {
                    if self.session_id.is_none() {
                        self.session_id = params
                            .pointer("/thread/id")
                            .and_then(Value::as_str)
                            .map(ToString::to_string);
                    }
                    CodexAppServerEvent::None
                }
                "mcpServer/startupStatus/updated" => render_codex_startup_status(params)
                    .map(CodexAppServerEvent::Status)
                    .unwrap_or(CodexAppServerEvent::None),
                "thread/status/changed" => render_codex_thread_status(params)
                    .map(CodexAppServerEvent::Status)
                    .unwrap_or(CodexAppServerEvent::None),
                "turn/started" => CodexAppServerEvent::Status("[turn] started".to_string()),
                "turn/plan/updated" => render_codex_plan_update(params)
                    .map(CodexAppServerEvent::Status)
                    .unwrap_or(CodexAppServerEvent::None),
                "item/fileChange/outputDelta" => render_codex_file_change_delta(params)
                    .map(CodexAppServerEvent::Status)
                    .unwrap_or(CodexAppServerEvent::None),
                "turn/diff/updated" => render_codex_turn_diff(params)
                    .map(CodexAppServerEvent::Status)
                    .unwrap_or(CodexAppServerEvent::None),
                "thread/tokenUsage/updated" | "account/rateLimits/updated" => {
                    CodexAppServerEvent::None
                }
                "item/agentMessage/delta" => {
                    let item_id = params
                        .get("itemId")
                        .and_then(Value::as_str)
                        .unwrap_or_default()
                        .to_string();
                    let delta = params
                        .get("delta")
                        .and_then(Value::as_str)
                        .unwrap_or_default();
                    if item_id.is_empty() || delta.is_empty() {
                        CodexAppServerEvent::None
                    } else {
                        let entry = self.partial_agent_messages.entry(item_id).or_default();
                        entry.push_str(delta);
                        self.render_assistant_text()
                            .map(CodexAppServerEvent::Content)
                            .unwrap_or(CodexAppServerEvent::None)
                    }
                }
                "item/started" | "item/completed" => self
                    .ingest_app_server_item(params.get("item").unwrap_or(&Value::Null), method)
                    .map(CodexAppServerEvent::Content)
                    .unwrap_or(CodexAppServerEvent::None),
                "serverRequest/resolved" => {
                    let request_id = params
                        .get("requestId")
                        .and_then(jsonrpc_id_to_string)
                        .unwrap_or_default();
                    if request_id.is_empty() {
                        CodexAppServerEvent::None
                    } else {
                        self.latest_status = None;
                        CodexAppServerEvent::ApprovalResolved { request_id }
                    }
                }
                "turn/completed" => {
                    let status = params
                        .pointer("/turn/status")
                        .and_then(Value::as_str)
                        .unwrap_or_default();
                    match status {
                        "completed" | "interrupted" => CodexAppServerEvent::TurnCompleted,
                        "failed" => {
                            let message = params
                                .pointer("/turn/error/message")
                                .and_then(Value::as_str)
                                .unwrap_or("codex turn failed")
                                .to_string();
                            CodexAppServerEvent::TurnFailed(message)
                        }
                        _ => CodexAppServerEvent::None,
                    }
                }
                "error" => {
                    let error = &params["error"];
                    let message = error
                        .get("message")
                        .and_then(Value::as_str)
                        .unwrap_or("codex app-server error");
                    let details = error
                        .get("additionalDetails")
                        .and_then(Value::as_str)
                        .unwrap_or("");
                    let full = if details.is_empty() {
                        message.to_string()
                    } else {
                        format!("{message}: {details}")
                    };
                    CodexAppServerEvent::TurnFailed(full)
                }
                _ => CodexAppServerEvent::Status(format!(
                    "[debug:codex:method] unhandled method={method}"
                )),
            };
        }

        CodexAppServerEvent::None
    }

    fn ingest_app_server_request(&mut self, value: &Value) -> CodexAppServerEvent {
        let method = value
            .get("method")
            .and_then(Value::as_str)
            .unwrap_or_default();
        let request_id = value
            .get("id")
            .and_then(jsonrpc_id_to_string)
            .unwrap_or_default();
        let params = value.get("params").unwrap_or(&Value::Null);
        if request_id.is_empty() {
            return CodexAppServerEvent::None;
        }

        let request = match method {
            "item/commandExecution/requestApproval" => Some(PendingApproval {
                request: ApprovalRequest {
                    request_id: request_id.clone(),
                    kind: ApprovalKind::CommandExecution,
                    command: params
                        .get("command")
                        .and_then(Value::as_str)
                        .map(ToString::to_string),
                    reason: params
                        .get("reason")
                        .and_then(Value::as_str)
                        .map(ToString::to_string),
                    allow_accept_for_session: approval_decisions_contain(
                        params.get("availableDecisions"),
                        "acceptForSession",
                    ),
                    allow_cancel: approval_decisions_contain(
                        params.get("availableDecisions"),
                        "cancel",
                    ),
                    resolvable: true,
                },
                last_choice: None,
            }),
            "item/fileChange/requestApproval" => Some(PendingApproval {
                request: ApprovalRequest {
                    request_id: request_id.clone(),
                    kind: ApprovalKind::FileChange,
                    command: render_file_change_approval_command(params),
                    reason: params
                        .get("reason")
                        .and_then(Value::as_str)
                        .map(ToString::to_string)
                        .or_else(|| Some("Apply proposed edits".to_string())),
                    allow_accept_for_session: approval_decisions_contain(
                        params.get("availableDecisions"),
                        "acceptForSession",
                    ),
                    allow_cancel: approval_decisions_contain(
                        params.get("availableDecisions"),
                        "cancel",
                    ),
                    resolvable: true,
                },
                last_choice: None,
            }),
            "execCommandApproval" => Some(PendingApproval {
                request: ApprovalRequest {
                    request_id: request_id.clone(),
                    kind: ApprovalKind::ExecCommand,
                    command: params
                        .get("command")
                        .and_then(Value::as_array)
                        .map(|parts| {
                            parts
                                .iter()
                                .filter_map(Value::as_str)
                                .collect::<Vec<_>>()
                                .join(" ")
                        })
                        .filter(|command| !command.is_empty()),
                    reason: params
                        .get("reason")
                        .and_then(Value::as_str)
                        .map(ToString::to_string),
                    allow_accept_for_session: true,
                    allow_cancel: true,
                    resolvable: true,
                },
                last_choice: None,
            }),
            "applyPatchApproval" => Some(PendingApproval {
                request: ApprovalRequest {
                    request_id: request_id.clone(),
                    kind: ApprovalKind::ApplyPatch,
                    command: render_file_change_approval_command(params),
                    reason: params
                        .get("reason")
                        .and_then(Value::as_str)
                        .map(ToString::to_string)
                        .or_else(|| Some("Apply proposed edits".to_string())),
                    allow_accept_for_session: true,
                    allow_cancel: true,
                    resolvable: true,
                },
                last_choice: None,
            }),
            "item/permissions/requestApproval" => Some(PendingApproval {
                request: ApprovalRequest {
                    request_id: request_id.clone(),
                    kind: ApprovalKind::Permissions,
                    command: None,
                    reason: params
                        .get("reason")
                        .and_then(Value::as_str)
                        .map(ToString::to_string),
                    allow_accept_for_session: false,
                    allow_cancel: true,
                    resolvable: true,
                },
                last_choice: None,
            }),
            _ => None,
        };

        if let Some(request) = request {
            self.latest_status = Some(render_approval_summary(&request.request));
            CodexAppServerEvent::ApprovalRequested(request)
        } else {
            CodexAppServerEvent::Status(format!(
                "[debug:codex:request] unhandled request method={method}"
            ))
        }
    }

    fn ingest_app_server_item(&mut self, item: &Value, method: &str) -> Option<String> {
        let item_type = item.get("type").and_then(Value::as_str).unwrap_or_default();
        match item_type {
            "agentMessage" if method == "item/completed" => {
                let item_id = item.get("id").and_then(Value::as_str).unwrap_or_default();
                let text = item.get("text").and_then(Value::as_str)?.trim().to_string();
                if text.is_empty() {
                    return None;
                }
                self.partial_agent_messages.remove(item_id);
                self.display_blocks.push(text.clone());
                self.assistant_blocks.push(text);
                self.render_assistant_text()
            }
            "agentMessage" => None,
            "userMessage" => None,
            "reasoning" => {
                self.latest_status = Some(if method == "item/started" {
                    "[reasoning] thinking".to_string()
                } else {
                    "[reasoning] complete".to_string()
                });
                None
            }
            "commandExecution" => {
                self.latest_status = Some(render_command_summary(
                    item.get("command")
                        .and_then(Value::as_str)
                        .unwrap_or_default(),
                    match item
                        .get("status")
                        .and_then(Value::as_str)
                        .unwrap_or_default()
                    {
                        "inProgress" => "running",
                        other => other,
                    },
                    item.get("exitCode").and_then(Value::as_i64),
                ));
                None
            }
            "webSearch" => {
                self.latest_status = Some(render_web_search_summary(
                    item.get("query")
                        .and_then(Value::as_str)
                        .unwrap_or_default(),
                    method,
                ));
                None
            }
            _ => {
                self.latest_status =
                    render_generic_item_summary(item_type, item, method).or_else(|| {
                        Some(format!(
                            "[debug:codex:item] unhandled item_type={} phase={}",
                            item_type,
                            if method == "item/started" {
                                "started"
                            } else {
                                "completed"
                            }
                        ))
                    });
                None
            }
        }
    }

    fn mark_idle_waiting(&mut self) -> Option<String> {
        let waiting = "[status] waiting for approval or blocked by sandbox/network";
        if self.latest_status.as_deref() == Some(waiting) {
            None
        } else {
            self.latest_status = Some(waiting.to_string());
            None
        }
    }

    fn finish_text(&self) -> Option<String> {
        let text = self.assistant_blocks.join("\n\n---\n\n");
        if text.is_empty() { None } else { Some(text) }
    }

    fn render_assistant_text(&self) -> Option<String> {
        let mut blocks = self.display_blocks.clone();
        if let Some((_, partial)) = self.partial_agent_messages.iter().last() {
            let partial = partial.trim();
            if !partial.is_empty() {
                blocks.push(partial.to_string());
            }
        }
        let text = blocks.join("\n\n---\n\n").trim().to_string();
        if text.is_empty() { None } else { Some(text) }
    }
}

async fn send_json_rpc_request(
    writer: &mut (impl AsyncWrite + Unpin),
    next_request_id: &mut u64,
    method: &str,
    params: Value,
) -> Result<u64> {
    let request_id = *next_request_id;
    *next_request_id += 1;
    write_json_line(
        writer,
        &serde_json::json!({
            "jsonrpc": "2.0",
            "id": request_id,
            "method": method,
            "params": params,
        }),
    )
    .await?;
    eprintln!("[codex:jsonrpc:send] id={request_id} method={method}");
    Ok(request_id)
}

async fn send_json_rpc_response(
    writer: &mut (impl AsyncWrite + Unpin),
    request_id: &str,
    result: Value,
) -> Result<()> {
    let id = request_id
        .parse::<u64>()
        .map(Value::from)
        .unwrap_or_else(|_| Value::String(request_id.to_string()));
    write_json_line(
        writer,
        &serde_json::json!({
            "id": id,
            "result": result,
        }),
    )
    .await?;
    eprintln!("[codex:jsonrpc:send-response] id={request_id}");
    Ok(())
}

async fn write_json_line(writer: &mut (impl AsyncWrite + Unpin), value: &Value) -> Result<()> {
    let mut bytes = serde_json::to_vec(value)?;
    bytes.push(b'\n');
    writer.write_all(&bytes).await?;
    writer.flush().await?;
    Ok(())
}

async fn wait_for_json_rpc_response(
    stdout_rx: &mut mpsc::UnboundedReceiver<std::result::Result<String, String>>,
    raw_stdout: &mut String,
    request_id: u64,
) -> Result<Value> {
    loop {
        let line = stdout_rx
            .recv()
            .await
            .context("codex app-server closed before responding")?
            .map_err(anyhow::Error::msg)?;
        if !raw_stdout.is_empty() {
            raw_stdout.push('\n');
        }
        raw_stdout.push_str(&line);
        eprintln!("[codex:jsonrpc:recv] {line}");
        let value: Value = serde_json::from_str(&line)
            .with_context(|| "codex app-server produced invalid JSON")?;
        if value.get("id").and_then(jsonrpc_id_to_string).as_deref()
            == Some(&request_id.to_string())
        {
            if let Some(error) = value.get("error") {
                let message = error
                    .get("message")
                    .and_then(Value::as_str)
                    .unwrap_or("codex app-server request failed");
                bail!("{message}");
            }
            return Ok(value.get("result").cloned().unwrap_or(Value::Null));
        }
    }
}

fn fallback_text(raw_stdout: &str) -> Option<String> {
    raw_stdout
        .lines()
        .filter_map(|line| serde_json::from_str::<Value>(line).ok())
        .filter_map(|value| {
            value
                .pointer("/params/item/text")
                .and_then(Value::as_str)
                .map(str::trim)
                .filter(|text| !text.is_empty())
                .map(ToString::to_string)
        })
        .next_back()
}

fn jsonrpc_id_to_string(value: &Value) -> Option<String> {
    match value {
        Value::String(value) => Some(value.clone()),
        Value::Number(value) => Some(value.to_string()),
        _ => None,
    }
}

fn approval_decisions_contain(value: Option<&Value>, expected: &str) -> bool {
    value
        .and_then(Value::as_array)
        .map(|items| {
            items.iter().any(|item| {
                item.as_str() == Some(expected)
                    || item
                        .as_object()
                        .map(|map| map.contains_key(expected))
                        .unwrap_or(false)
            })
        })
        .unwrap_or(false)
}

fn render_file_change_approval_command(params: &Value) -> Option<String> {
    let files = params
        .get("files")
        .or_else(|| params.get("fileChanges"))
        .or_else(|| params.get("changes"))
        .and_then(Value::as_array)
        .map(|items| {
            items
                .iter()
                .filter_map(file_change_path)
                .take(5)
                .collect::<Vec<_>>()
        })
        .filter(|paths| !paths.is_empty());

    if let Some(paths) = files {
        let suffix = if paths.len() == 5 { "..." } else { "" };
        return Some(format!("edit files: {}{}", paths.join(", "), suffix));
    }

    params
        .get("path")
        .or_else(|| params.get("file"))
        .or_else(|| params.pointer("/file/path"))
        .and_then(Value::as_str)
        .map(|path| format!("edit file: {path}"))
        .or_else(|| Some("apply proposed edits".to_string()))
}

fn file_change_path(value: &Value) -> Option<String> {
    value
        .as_str()
        .map(ToString::to_string)
        .or_else(|| {
            value
                .get("path")
                .and_then(Value::as_str)
                .map(ToString::to_string)
        })
        .or_else(|| {
            value
                .get("file")
                .and_then(Value::as_str)
                .map(ToString::to_string)
        })
        .or_else(|| {
            value
                .get("relativePath")
                .and_then(Value::as_str)
                .map(ToString::to_string)
        })
}

fn render_approval_summary(request: &ApprovalRequest) -> String {
    let command = request
        .command
        .as_deref()
        .filter(|command| !command.is_empty())
        .unwrap_or("unknown command");
    let reason = request.reason.as_deref().unwrap_or("等待用户审批");
    format!("[approval] {reason}: {command}")
}

fn render_command_summary(command: &str, status: &str, exit_code: Option<i64>) -> String {
    let command = if command.trim().is_empty() {
        "command"
    } else {
        command
    };
    match exit_code {
        Some(code) => format!("[command:{status}] {command} (exit {code})"),
        None => format!("[command:{status}] {command}"),
    }
}

fn render_web_search_summary(query: &str, method: &str) -> String {
    let query = if query.trim().is_empty() {
        "search"
    } else {
        query
    };
    match method {
        "item/started" => format!("[web] searching: {query}"),
        _ => format!("[web] finished: {query}"),
    }
}

fn render_codex_startup_status(params: &Value) -> Option<String> {
    let status = params
        .get("status")
        .or_else(|| params.pointer("/startupStatus/status"))
        .and_then(Value::as_str)?;
    match status {
        "ready" | "completed" => None,
        other => Some(format!("[mcp] startup {other}")),
    }
}

fn render_codex_thread_status(params: &Value) -> Option<String> {
    let status = params
        .get("status")
        .or_else(|| params.pointer("/thread/status"))
        .and_then(Value::as_str)?;
    let summary = match status {
        "running" => "[thread] running",
        "waiting" => "[thread] waiting",
        "idle" => return None,
        other => return Some(format!("[thread] {other}")),
    };
    Some(summary.to_string())
}

fn render_codex_plan_update(params: &Value) -> Option<String> {
    let steps = params
        .get("steps")
        .or_else(|| params.pointer("/plan/steps"))
        .or_else(|| params.get("items"))
        .or_else(|| params.pointer("/plan/items"))
        .and_then(Value::as_array)?;
    let previews = steps
        .iter()
        .filter_map(|step| {
            step.get("title")
                .or_else(|| step.get("step"))
                .or_else(|| step.get("content"))
                .or_else(|| step.get("text"))
                .and_then(Value::as_str)
                .map(str::trim)
                .filter(|text| !text.is_empty())
                .map(ToString::to_string)
        })
        .take(3)
        .collect::<Vec<_>>();
    let count = steps.len();
    if previews.is_empty() {
        Some(format!("[plan] updated ({count} steps)"))
    } else {
        Some(format!("[plan] {} ({count} steps)", previews.join(" | ")))
    }
}

fn render_codex_file_change_delta(params: &Value) -> Option<String> {
    let files = collect_file_paths(params);
    let preview = if files.is_empty() {
        params
            .get("delta")
            .and_then(Value::as_str)
            .or_else(|| params.get("output").and_then(Value::as_str))
            .or_else(|| params.get("text").and_then(Value::as_str))
            .map(|text| truncate_tool_text(text, 120))
            .filter(|text| !text.is_empty())?
    } else {
        format_file_preview(&files)
    };
    Some(format!("[file:delta] {preview}"))
}

fn render_codex_turn_diff(params: &Value) -> Option<String> {
    let files = collect_file_paths(params);
    let mut segments = Vec::new();
    if !files.is_empty() {
        segments.push(format_file_preview(&files));
    }

    let added = find_first_i64(
        params,
        &[
            "added",
            "insertions",
            "linesAdded",
            "additions",
            "totalAdded",
        ],
    );
    let removed = find_first_i64(
        params,
        &[
            "removed",
            "deletions",
            "linesRemoved",
            "totalRemoved",
            "deletionsCount",
        ],
    );

    match (added, removed) {
        (Some(added), Some(removed)) => segments.push(format!("+{added} -{removed}")),
        (Some(added), None) => segments.push(format!("+{added}")),
        (None, Some(removed)) => segments.push(format!("-{removed}")),
        (None, None) => {}
    }

    let summary = params
        .get("summary")
        .or_else(|| params.get("title"))
        .or_else(|| params.get("description"))
        .and_then(Value::as_str)
        .map(|text| truncate_tool_text(text, 80))
        .filter(|text| !text.is_empty());
    if let Some(summary) = summary {
        segments.push(summary);
    }

    if segments.is_empty() {
        None
    } else {
        Some(format!("[diff] {}", segments.join(" | ")))
    }
}

fn render_generic_item_summary(item_type: &str, item: &Value, method: &str) -> Option<String> {
    let normalized = item_type.trim();
    if normalized.is_empty() {
        return None;
    }

    let phase = if method == "item/started" {
        "running"
    } else {
        "done"
    };

    if normalized.to_ascii_lowercase().contains("todo") {
        let todo_items = item
            .get("todos")
            .or_else(|| item.get("items"))
            .and_then(Value::as_array)
            .map(|items| {
                items
                    .iter()
                    .filter_map(|entry| {
                        entry
                            .get("content")
                            .or_else(|| entry.get("text"))
                            .or_else(|| entry.get("title"))
                            .and_then(Value::as_str)
                            .map(str::trim)
                            .filter(|text| !text.is_empty())
                            .map(ToString::to_string)
                    })
                    .take(3)
                    .collect::<Vec<_>>()
            })
            .unwrap_or_default();
        let count = item
            .get("todos")
            .or_else(|| item.get("items"))
            .and_then(Value::as_array)
            .map(|items| items.len())
            .unwrap_or(todo_items.len());
        return Some(if todo_items.is_empty() {
            format!("[todo:{phase}] {count} items")
        } else {
            format!("[todo:{phase}] {} ({count} items)", todo_items.join(" | "))
        });
    }

    let files = collect_file_paths(item);
    if !files.is_empty() {
        let preview = files.iter().take(3).cloned().collect::<Vec<_>>().join(", ");
        return Some(format!(
            "[file:{phase}] {}{}",
            preview,
            if files.len() > 3 {
                format!(" +{} more", files.len() - 3)
            } else {
                String::new()
            }
        ));
    }

    let output = item
        .get("stdout")
        .or_else(|| item.get("stderr"))
        .or_else(|| item.get("output"))
        .and_then(Value::as_str)
        .map(|text| truncate_tool_text(text, 120));
    if let Some(output) = output.filter(|text| !text.is_empty()) {
        return Some(format!("[{normalized}:{phase}] {output}"));
    }

    let title = item
        .get("title")
        .or_else(|| item.get("description"))
        .or_else(|| item.get("name"))
        .and_then(Value::as_str)
        .map(|text| truncate_tool_text(text, 120));
    title.map(|title| format!("[{normalized}:{phase}] {title}"))
}

fn collect_file_paths(value: &Value) -> Vec<String> {
    let mut results = Vec::new();
    collect_file_paths_inner(value, &mut results);
    results.sort();
    results.dedup();
    results
}

fn format_file_preview(files: &[String]) -> String {
    let preview = files.iter().take(3).cloned().collect::<Vec<_>>().join(", ");
    if files.len() > 3 {
        format!("{preview} +{} more", files.len() - 3)
    } else {
        preview
    }
}

fn find_first_i64(value: &Value, keys: &[&str]) -> Option<i64> {
    match value {
        Value::Object(map) => {
            for key in keys {
                if let Some(number) = map.get(*key).and_then(Value::as_i64) {
                    return Some(number);
                }
            }
            for nested in map.values() {
                if let Some(number) = find_first_i64(nested, keys) {
                    return Some(number);
                }
            }
            None
        }
        Value::Array(items) => items.iter().find_map(|item| find_first_i64(item, keys)),
        _ => None,
    }
}

fn collect_file_paths_inner(value: &Value, results: &mut Vec<String>) {
    match value {
        Value::Object(map) => {
            for key in ["filePath", "file_path", "path", "target_file"] {
                if let Some(path) = map.get(key).and_then(Value::as_str) {
                    let path = path.trim();
                    if !path.is_empty() && looks_like_file_path(path) {
                        results.push(path.to_string());
                    }
                }
            }
            for nested in map.values() {
                collect_file_paths_inner(nested, results);
            }
        }
        Value::Array(items) => {
            for item in items {
                collect_file_paths_inner(item, results);
            }
        }
        _ => {}
    }
}

fn looks_like_file_path(path: &str) -> bool {
    path.contains('/') || path.contains('.') || path.contains('\\')
}

fn truncate_tool_text(text: &str, max_chars: usize) -> String {
    let text = text.trim().replace('\n', " ");
    let mut chars = text.chars();
    let truncated: String = chars.by_ref().take(max_chars).collect();
    if chars.next().is_some() {
        format!("{truncated}...")
    } else {
        truncated
    }
}

#[derive(Default)]
struct ClaudeStreamingState {
    display_blocks: Vec<String>,
    assistant_blocks: Vec<String>,
    latest_status: Option<String>,
    partial_text: Option<String>,
    stream_text: String,
}

impl ClaudeStreamingState {
    fn current_status(&self) -> Option<&str> {
        self.latest_status.as_deref()
    }

    fn ingest_line(&mut self, line: &str) -> Result<Option<String>> {
        let value = serde_json::from_str::<Value>(line)
            .with_context(|| "Claude stream produced invalid JSON")?;
        let Some(map) = value.as_object() else {
            return Ok(None);
        };

        match map.get("type").and_then(Value::as_str).unwrap_or_default() {
            "assistant" => Ok(self.ingest_assistant_text(
                map.get("message")
                    .and_then(extract_text_from_json)
                    .or_else(|| map.get("content").and_then(extract_text_from_json))
                    .or_else(|| map.get("text").and_then(extract_text_from_json)),
            )),
            "stream_event" => {
                Ok(self.ingest_stream_event(map.get("event").or_else(|| map.get("stream_event"))))
            }
            "system" => {
                self.latest_status = map
                    .get("subtype")
                    .and_then(Value::as_str)
                    .map(|subtype| format!("[claude] {subtype}"));
                Ok(None)
            }
            "result" => {
                if map
                    .get("is_error")
                    .and_then(Value::as_bool)
                    .unwrap_or(false)
                {
                    let message = map
                        .get("result")
                        .and_then(extract_text_from_json)
                        .unwrap_or_else(|| "claude run failed".to_string());
                    bail!("{message}");
                }
                if let Some(text) = map
                    .get("result")
                    .and_then(extract_text_from_json)
                    .or_else(|| map.get("output_text").and_then(extract_text_from_json))
                    .or_else(|| map.get("text").and_then(extract_text_from_json))
                {
                    self.partial_text = None;
                    if self.assistant_blocks.last() != Some(&text) {
                        self.display_blocks.push(text.clone());
                        self.assistant_blocks.push(text);
                    }
                    self.latest_status = None;
                    Ok(self.render_assistant_text())
                } else {
                    Ok(None)
                }
            }
            other => {
                self.latest_status = Some(format!("[debug:claude:type] unhandled type={other}"));
                Ok(None)
            }
        }
    }

    fn ingest_stream_event(&mut self, event: Option<&Value>) -> Option<String> {
        let Some(event) = event.and_then(Value::as_object) else {
            return None;
        };

        match event
            .get("type")
            .and_then(Value::as_str)
            .unwrap_or_default()
        {
            "content_block_start" => {
                if let Some(text) = event
                    .get("content_block")
                    .and_then(extract_text_from_json)
                    .or_else(|| event.get("block").and_then(extract_text_from_json))
                {
                    self.stream_text.push_str(&text);
                    self.partial_text = Some(self.stream_text.clone());
                    self.latest_status = None;
                    return self.render_assistant_text();
                }
                None
            }
            "content_block_delta" => {
                let text = event
                    .get("delta")
                    .and_then(extract_stream_delta_text)
                    .or_else(|| {
                        event
                            .get("text")
                            .and_then(Value::as_str)
                            .map(str::to_string)
                    })?;
                self.stream_text.push_str(&text);
                self.partial_text = Some(self.stream_text.clone());
                self.latest_status = None;
                self.render_assistant_text()
            }
            "content_block_stop" | "message_stop" => {
                let text = self.stream_text.trim().to_string();
                if !text.is_empty() && self.assistant_blocks.last() != Some(&text) {
                    self.display_blocks.push(text.clone());
                    self.assistant_blocks.push(text);
                }
                self.partial_text = None;
                self.stream_text.clear();
                self.latest_status = None;
                self.render_assistant_text()
            }
            "message_start" | "message_delta" => {
                self.latest_status = None;
                None
            }
            _ => None,
        }
    }

    fn ingest_assistant_text(&mut self, text: Option<String>) -> Option<String> {
        let text = text?.trim().to_string();
        if text.is_empty() {
            return None;
        }
        self.partial_text = Some(text);
        self.latest_status = None;
        self.render_assistant_text()
    }

    fn ingest_external_status(&mut self, summary: String) -> Option<String> {
        self.latest_status = Some(summary);
        None
    }

    fn finish_text(&self) -> Option<String> {
        if let Some(text) = self.partial_text.as_ref() {
            let text = text.trim();
            if !text.is_empty() {
                return Some(text.to_string());
            }
        }
        let text = self.assistant_blocks.join("\n\n---\n\n");
        if text.trim().is_empty() {
            None
        } else {
            Some(text)
        }
    }

    fn render_assistant_text(&self) -> Option<String> {
        let mut blocks = self.display_blocks.clone();
        if let Some(text) = self.partial_text.as_ref() {
            let text = text.trim();
            if !text.is_empty() {
                blocks.push(text.to_string());
            }
        }
        let text = blocks.join("\n\n---\n\n").trim().to_string();
        if text.is_empty() { None } else { Some(text) }
    }
}

#[cfg(test)]
#[derive(Default)]
struct OpenCodeStreamingState {
    session_id: Option<String>,
    assistant_text: String,
    latest_status: Option<String>,
    finished: bool,
}

#[cfg(test)]
enum OpenCodeStreamEvent {
    None,
    Content(String),
    Status(String),
    Session(String),
    Error(String),
}

#[cfg(test)]
impl OpenCodeStreamingState {
    fn ingest_line(&mut self, line: &str) -> Result<OpenCodeStreamEvent> {
        let value = serde_json::from_str::<Value>(line)
            .with_context(|| "opencode produced invalid JSON")?;
        let Some(map) = value.as_object() else {
            return Ok(OpenCodeStreamEvent::None);
        };

        if self.session_id.is_none()
            && let Some(session_id) = value.get("sessionID").and_then(Value::as_str)
        {
            self.session_id = Some(session_id.to_string());
            return Ok(OpenCodeStreamEvent::Session(session_id.to_string()));
        }

        match map.get("type").and_then(Value::as_str).unwrap_or_default() {
            "text" => {
                let text = value
                    .pointer("/part/text")
                    .or_else(|| value.get("text"))
                    .and_then(Value::as_str)
                    .unwrap_or_default();
                if text.is_empty() {
                    return Ok(OpenCodeStreamEvent::None);
                }
                self.assistant_text.push_str(text);
                Ok(OpenCodeStreamEvent::Content(self.assistant_text.clone()))
            }
            "step_start" | "step-start" => Ok(OpenCodeStreamEvent::Status(
                "[opencode] step started".to_string(),
            )),
            "step_finish" | "step-finish" => {
                self.finished = true;
                self.latest_status = None;
                Ok(OpenCodeStreamEvent::None)
            }
            "tool_use" | "tool-use" => render_opencode_tool_status(&value)
                .map(OpenCodeStreamEvent::Status)
                .map(Ok)
                .unwrap_or(Ok(OpenCodeStreamEvent::None)),
            "error" => Ok(OpenCodeStreamEvent::Error(render_opencode_error(&value))),
            "session" => value
                .get("id")
                .or_else(|| value.get("sessionID"))
                .and_then(Value::as_str)
                .map(|session_id| {
                    self.session_id = Some(session_id.to_string());
                    OpenCodeStreamEvent::Session(session_id.to_string())
                })
                .map(Ok)
                .unwrap_or(Ok(OpenCodeStreamEvent::None)),
            other if other.is_empty() => Ok(OpenCodeStreamEvent::None),
            other => Ok(OpenCodeStreamEvent::Status(format!(
                "[debug:opencode:type] unhandled type={other}"
            ))),
        }
    }

    fn finish_text(&self) -> Option<String> {
        let text = self.assistant_text.trim();
        if text.is_empty() {
            None
        } else {
            Some(text.to_string())
        }
    }
}

#[cfg(test)]
fn render_opencode_tool_status(value: &Value) -> Option<String> {
    let part = value.get("part").unwrap_or(value);
    let tool = part
        .get("tool")
        .or_else(|| part.get("name"))
        .or_else(|| value.get("tool"))
        .and_then(Value::as_str)
        .unwrap_or("tool");
    let status = part
        .get("status")
        .or_else(|| value.get("status"))
        .and_then(Value::as_str)
        .unwrap_or("running");
    let command = part
        .pointer("/input/command")
        .or_else(|| part.pointer("/args/command"))
        .or_else(|| part.get("command"))
        .and_then(Value::as_str)
        .map(|text| truncate_tool_text(text, 120));
    Some(match command.filter(|text| !text.is_empty()) {
        Some(command) => format!("[opencode:{tool}:{status}] {command}"),
        None => format!("[opencode:{tool}:{status}]"),
    })
}

fn render_opencode_error(value: &Value) -> String {
    value
        .pointer("/error/data/message")
        .or_else(|| value.pointer("/error/message"))
        .or_else(|| value.get("message"))
        .and_then(Value::as_str)
        .unwrap_or("opencode run failed")
        .to_string()
}

fn parse_sse_json_event(line: &str) -> Option<Value> {
    line.strip_prefix("data:")
        .map(str::trim)
        .filter(|data| !data.is_empty() && *data != "[DONE]")
        .and_then(|data| serde_json::from_str::<Value>(data).ok())
}

fn opencode_session_summary(value: &Value) -> Option<SessionSummary> {
    let id = value.get("id").and_then(Value::as_str)?.to_string();
    let directory = value.get("directory").and_then(Value::as_str)?;
    let title = value
        .get("title")
        .and_then(Value::as_str)
        .filter(|title| !title.trim().is_empty())
        .unwrap_or("opencode session")
        .to_string();
    Some(SessionSummary {
        id,
        project_id: crate::session_store::project_id_for_path(directory),
        title: title.clone(),
        agent: AgentKind::OpenCode,
        brief_reply_mode: false,
        status: crate::models::SessionStatus::Idle,
        updated_at: opencode_time(value.pointer("/time/updated")),
        unread_count: 0,
        last_message_preview: Some(title),
        pending_approval: None,
        provider_id: None,
    })
}

fn opencode_messages_from_value(session_id: &str, value: &Value) -> Vec<ChatMessage> {
    value
        .as_array()
        .into_iter()
        .flatten()
        .filter_map(|entry| {
            let info = entry.get("info")?;
            let role = match info.get("role").and_then(Value::as_str)? {
                "user" => crate::models::MessageRole::User,
                "assistant" => crate::models::MessageRole::Assistant,
                _ => return None,
            };
            let content = opencode_parts_text(entry.get("parts").unwrap_or(&Value::Null));
            if content.trim().is_empty() {
                return None;
            }
            Some(ChatMessage {
                id: info
                    .get("id")
                    .and_then(Value::as_str)
                    .unwrap_or_default()
                    .to_string(),
                session_id: session_id.to_string(),
                role,
                content,
                created_at: opencode_time(info.pointer("/time/created")),
            })
        })
        .collect()
}

fn opencode_parts_text(parts: &Value) -> String {
    parts
        .as_array()
        .into_iter()
        .flatten()
        .filter_map(|part| match part.get("type").and_then(Value::as_str) {
            Some("text") => part.get("text").and_then(Value::as_str),
            _ => None,
        })
        .collect::<Vec<_>>()
        .join("")
        .trim()
        .to_string()
}

fn opencode_time(value: Option<&Value>) -> chrono::DateTime<chrono::Utc> {
    let millis = value.and_then(Value::as_i64).unwrap_or(0);
    chrono::DateTime::<chrono::Utc>::from_timestamp_millis(millis).unwrap_or_else(chrono::Utc::now)
}

fn project_name_from_path(path: &str) -> String {
    Path::new(path)
        .file_name()
        .and_then(|name| name.to_str())
        .filter(|name| !name.is_empty())
        .unwrap_or(path)
        .to_string()
}

fn select_claude_runtime_ref(
    session_id: &str,
    existing_runtime_ref: Option<&str>,
    can_resume_existing_session: bool,
) -> String {
    if can_resume_existing_session {
        existing_runtime_ref
            .map(ToString::to_string)
            .unwrap_or_else(|| session_id.to_string())
    } else if existing_runtime_ref.is_some() {
        uuid::Uuid::new_v4().to_string()
    } else {
        session_id.to_string()
    }
}

fn opencode_permission_to_approval(value: &Value) -> ApprovalRequest {
    let request_id = value
        .get("id")
        .and_then(Value::as_str)
        .unwrap_or_default()
        .to_string();
    let permission = value
        .get("permission")
        .and_then(Value::as_str)
        .unwrap_or("permission");
    let patterns = value
        .get("patterns")
        .and_then(Value::as_array)
        .map(|items| {
            items
                .iter()
                .filter_map(Value::as_str)
                .collect::<Vec<_>>()
                .join(" ")
        })
        .filter(|text| !text.is_empty());
    let metadata_command = value
        .pointer("/metadata/command")
        .or_else(|| value.pointer("/metadata/description"))
        .and_then(Value::as_str)
        .map(ToString::to_string);
    let command = metadata_command.or(patterns);
    ApprovalRequest {
        request_id,
        kind: if command.is_some() {
            ApprovalKind::ExecCommand
        } else {
            ApprovalKind::Permissions
        },
        command,
        reason: Some(format!("OpenCode 请求 {permission} 权限")),
        allow_accept_for_session: true,
        allow_cancel: true,
        resolvable: true,
    }
}

async fn auto_approve_opencode_request(
    request: &ApprovalRequest,
    project_root: &Path,
) -> Result<Option<ApprovalChoice>> {
    match auto_approve_codex_request(request, project_root).await? {
        Some(ApprovalChoice::Accept) => Ok(Some(ApprovalChoice::Accept)),
        Some(other) => Ok(Some(other)),
        None => Ok(None),
    }
}

fn render_opencode_part_status(part: &Value) -> Option<String> {
    match part.get("type").and_then(Value::as_str)? {
        "tool" => {
            let tool = part.get("tool").and_then(Value::as_str).unwrap_or("tool");
            let status = part
                .pointer("/state/status")
                .and_then(Value::as_str)
                .unwrap_or("running");
            let command = part
                .pointer("/state/input/command")
                .or_else(|| part.pointer("/state/title"))
                .and_then(Value::as_str)
                .map(|text| truncate_tool_text(text, 120));
            Some(match command.filter(|text| !text.is_empty()) {
                Some(command) => format!("[opencode:{tool}:{status}] {command}"),
                None => format!("[opencode:{tool}:{status}]"),
            })
        }
        "step-start" => Some("[opencode] step started".to_string()),
        "step-finish" => Some("[opencode] step finished".to_string()),
        "patch" => render_opencode_patch_status(part),
        _ => None,
    }
}

fn render_opencode_patch_status(part: &Value) -> Option<String> {
    let files = part.get("files").and_then(Value::as_array)?;
    let paths = files
        .iter()
        .filter_map(Value::as_str)
        .take(3)
        .collect::<Vec<_>>();
    if paths.is_empty() {
        None
    } else {
        Some(format!("[opencode:patch] {}", paths.join(", ")))
    }
}

async fn wait_for_opencode_session_idle(
    client: &mut OpenCodeHttpClient,
    session_id: &str,
    project_root: &Path,
) -> Result<()> {
    let mut stream = client.event_stream(project_root).await?;
    while let Some(line) = stream.recv().await {
        let line = line?;
        let Some(event) = parse_sse_json_event(&line) else {
            continue;
        };
        let event_type = event
            .get("type")
            .and_then(Value::as_str)
            .unwrap_or_default();
        let properties = event.get("properties").unwrap_or(&Value::Null);
        let event_session_id = properties
            .get("sessionID")
            .and_then(Value::as_str)
            .unwrap_or_default();
        if !event_session_id.is_empty() && event_session_id != session_id {
            continue;
        }
        match event_type {
            "session.idle" => return Ok(()),
            "session.error" => bail!("{}", render_opencode_error(properties)),
            _ => {}
        }
    }
    bail!("opencode event stream closed before session became idle")
}

fn url_path_escape(value: &str) -> String {
    value
        .bytes()
        .flat_map(|byte| match byte {
            b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'_' | b'.' | b'~' => {
                vec![byte as char]
            }
            other => format!("%{other:02X}").chars().collect::<Vec<_>>(),
        })
        .collect()
}

fn extract_text_from_json(value: &Value) -> Option<String> {
    match value {
        Value::String(text) => {
            let text = text.trim();
            if text.is_empty() {
                None
            } else {
                Some(text.to_string())
            }
        }
        Value::Array(items) => {
            let text = items
                .iter()
                .filter_map(|item| match item.get("type").and_then(Value::as_str) {
                    Some("text") => item
                        .get("text")
                        .and_then(Value::as_str)
                        .map(ToString::to_string),
                    _ => extract_text_from_json(item),
                })
                .collect::<Vec<_>>()
                .join("");
            let text = text.trim();
            if text.is_empty() {
                None
            } else {
                Some(text.to_string())
            }
        }
        Value::Object(map) => map
            .get("content")
            .and_then(extract_text_from_json)
            .or_else(|| map.get("message").and_then(extract_text_from_json))
            .or_else(|| map.get("text").and_then(extract_text_from_json))
            .or_else(|| map.get("result").and_then(extract_text_from_json))
            .or_else(|| map.get("output_text").and_then(extract_text_from_json)),
        _ => None,
    }
}

fn extract_stream_delta_text(value: &Value) -> Option<String> {
    match value {
        Value::String(text) => Some(text.to_string()),
        Value::Array(items) => {
            let text = items
                .iter()
                .filter_map(extract_stream_delta_text)
                .collect::<Vec<_>>()
                .join("");
            Some(text)
        }
        Value::Object(map) => map
            .get("text")
            .and_then(Value::as_str)
            .map(str::to_string)
            .or_else(|| map.get("delta").and_then(extract_stream_delta_text))
            .or_else(|| map.get("content").and_then(extract_stream_delta_text)),
        _ => None,
    }
}

fn strip_ansi(text: &str) -> String {
    let mut result = String::with_capacity(text.len());
    let mut chars = text.chars().peekable();
    while let Some(ch) = chars.next() {
        if ch == '\u{1b}' {
            if matches!(chars.peek(), Some('[') | Some(']')) {
                chars.next();
                while let Some(next) = chars.next() {
                    if ('@'..='~').contains(&next) {
                        break;
                    }
                }
                continue;
            }
        }
        result.push(ch);
    }
    result
}

async fn next_claude_permission_request(
    state_dir: &std::path::Path,
    run_id: &str,
    seen: &mut std::collections::HashSet<String>,
) -> Result<Option<ClaudePermissionRequest>> {
    let mut entries = match tokio::fs::read_dir(state_dir.join("requests")).await {
        Ok(entries) => entries,
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
        Err(error) => return Err(error.into()),
    };

    while let Some(entry) = entries.next_entry().await? {
        let path = entry.path();
        let Some(stem) = path.file_stem().and_then(|value| value.to_str()) else {
            continue;
        };
        if seen.contains(stem) {
            continue;
        }
        let body = tokio::fs::read(&path).await?;
        if body.is_empty() {
            continue;
        }
        let request: ClaudePermissionRequest = serde_json::from_slice(&body)
            .with_context(|| format!("failed to parse {}", path.display()))?;
        if request.run_id != run_id {
            continue;
        }
        seen.insert(request.request_id.clone());
        return Ok(Some(request));
    }
    Ok(None)
}

async fn next_claude_status_event(
    state_dir: &std::path::Path,
    run_id: &str,
    seen: &mut std::collections::HashSet<String>,
) -> Result<Option<ClaudeHookStatusEvent>> {
    let mut entries = match tokio::fs::read_dir(state_dir.join("events")).await {
        Ok(entries) => entries,
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
        Err(error) => return Err(error.into()),
    };

    while let Some(entry) = entries.next_entry().await? {
        let path = entry.path();
        let Some(stem) = path.file_stem().and_then(|value| value.to_str()) else {
            continue;
        };
        if seen.contains(stem) {
            continue;
        }
        let body = tokio::fs::read(&path).await?;
        if body.is_empty() {
            let _ = tokio::fs::remove_file(&path).await;
            continue;
        }
        let event: ClaudeHookStatusEvent = serde_json::from_slice(&body)
            .with_context(|| format!("failed to parse {}", path.display()))?;
        if event.run_id != run_id {
            continue;
        }
        seen.insert(event.event_id.clone());
        let _ = tokio::fs::remove_file(&path).await;
        return Ok(Some(event));
    }
    Ok(None)
}

fn shell_quote(value: &str) -> String {
    format!("'{}'", value.replace('\'', r"'\''"))
}

async fn auto_approve_codex_request(
    request: &ApprovalRequest,
    project_root: &std::path::Path,
) -> Result<Option<ApprovalChoice>> {
    match request.kind {
        ApprovalKind::CommandExecution | ApprovalKind::ExecCommand => {
            let Some(command) = request.command.as_deref() else {
                return Ok(None);
            };
            if let Some(decision) = approval_policy::should_auto_approve(command, project_root) {
                return Ok(match decision {
                    AutoApprovalDecision::Accept => Some(ApprovalChoice::Accept),
                });
            }
            let decision = match ai_approval::review_request(request, project_root).await {
                Ok(decision) => decision,
                Err(error) => {
                    eprintln!(
                        "AI approval review failed; falling back to user approval: {error:?}"
                    );
                    None
                }
            };
            Ok(match decision.map(|decision| decision.decision) {
                Some(AiApprovalDecisionKind::Accept) => Some(ApprovalChoice::Accept),
                Some(AiApprovalDecisionKind::Decline) => Some(ApprovalChoice::Decline),
                Some(AiApprovalDecisionKind::AskUser) | None => None,
            })
        }
        ApprovalKind::FileChange | ApprovalKind::ApplyPatch | ApprovalKind::Permissions => Ok(None),
    }
}

fn approval_result_json(choice: &ApprovalChoice, kind: &ApprovalKind) -> Value {
    match kind {
        ApprovalKind::CommandExecution | ApprovalKind::FileChange => serde_json::json!({
            "decision": match choice {
                ApprovalChoice::Accept | ApprovalChoice::AlwaysAllow => Value::String("accept".to_string()),
                ApprovalChoice::AcceptForSession => Value::String("acceptForSession".to_string()),
                ApprovalChoice::Decline => Value::String("decline".to_string()),
                ApprovalChoice::Cancel => Value::String("cancel".to_string()),
            }
        }),
        ApprovalKind::ExecCommand => serde_json::json!({
            "decision": match choice {
                ApprovalChoice::Accept | ApprovalChoice::AlwaysAllow => Value::String("approved".to_string()),
                ApprovalChoice::AcceptForSession => Value::String("approved_for_session".to_string()),
                ApprovalChoice::Decline => Value::String("denied".to_string()),
                ApprovalChoice::Cancel => Value::String("abort".to_string()),
            }
        }),
        ApprovalKind::ApplyPatch => serde_json::json!({
            "decision": match choice {
                ApprovalChoice::Accept | ApprovalChoice::AlwaysAllow => Value::String("approved".to_string()),
                ApprovalChoice::AcceptForSession => Value::String("approved_for_session".to_string()),
                ApprovalChoice::Decline => Value::String("denied".to_string()),
                ApprovalChoice::Cancel => Value::String("abort".to_string()),
            }
        }),
        ApprovalKind::Permissions => match choice {
            ApprovalChoice::Accept
            | ApprovalChoice::AcceptForSession
            | ApprovalChoice::AlwaysAllow => serde_json::json!({
                "permissions": {
                    "network": { "enabled": true }
                },
                "scope": if matches!(choice, ApprovalChoice::AcceptForSession) {
                    "session"
                } else {
                    "turn"
                },
            }),
            ApprovalChoice::Decline | ApprovalChoice::Cancel => serde_json::json!({
                "permissions": {},
                "scope": "turn",
            }),
        },
    }
}

fn is_macos() -> bool {
    cfg!(target_os = "macos")
}

fn is_windows() -> bool {
    cfg!(target_os = "windows")
}

async fn command_exists(name: &str) -> bool {
    tokio::process::Command::new(if cfg!(target_os = "windows") {
        "where"
    } else {
        "which"
    })
    .arg(name)
    .output()
    .await
    .map(|o| o.status.success())
    .unwrap_or(false)
}

async fn try_install_with_npm(
    agent: AgentKind,
    npm_package: &str,
    binary_name: &str,
) -> Option<crate::models::AgentInstallResult> {
    if !command_exists("npm").await {
        return None;
    }

    let result = tokio::process::Command::new("npm")
        .args(["install", "-g", npm_package])
        .output()
        .await;

    match result {
        Ok(output) if output.status.success() => {
            let path =
                find_executable_in_path(binary_name).unwrap_or_else(|| PathBuf::from(binary_name));
            Some(crate::models::AgentInstallResult {
                agent,
                success: true,
                message: Some(format!("installed successfully via npm ({npm_package})")),
                installed_path: Some(path.display().to_string()),
            })
        }
        _ => None,
    }
}

async fn try_install_with_brew(
    agent: AgentKind,
    brew_package: &str,
    binary_name: &str,
) -> Option<crate::models::AgentInstallResult> {
    if !is_macos() || !command_exists("brew").await {
        return None;
    }

    let result = tokio::process::Command::new("brew")
        .args(["install", brew_package])
        .output()
        .await;

    match result {
        Ok(output) if output.status.success() => {
            let path =
                find_executable_in_path(binary_name).unwrap_or_else(|| PathBuf::from(binary_name));
            Some(crate::models::AgentInstallResult {
                agent,
                success: true,
                message: Some(format!("installed successfully via brew ({brew_package})")),
                installed_path: Some(path.display().to_string()),
            })
        }
        _ => None,
    }
}

async fn try_install_with_script(
    agent: AgentKind,
    unix_script_url: &str,
    windows_script_url: Option<&str>,
    binary_name: &str,
) -> Option<crate::models::AgentInstallResult> {
    let result = if is_windows() {
        let Some(url) = windows_script_url else {
            return None;
        };
        tokio::process::Command::new("powershell")
            .args(["-Command", &format!("irm {url} | iex")])
            .output()
            .await
    } else {
        tokio::process::Command::new("sh")
            .args(["-c", &format!("curl -fsSL {unix_script_url} | sh")])
            .output()
            .await
    };

    let used_url = if is_windows() {
        windows_script_url.unwrap_or(unix_script_url)
    } else {
        unix_script_url
    };

    match result {
        Ok(output) if output.status.success() => {
            let path =
                find_executable_in_path(binary_name).unwrap_or_else(|| PathBuf::from(binary_name));
            Some(crate::models::AgentInstallResult {
                agent,
                success: true,
                message: Some(format!("installed successfully via script ({used_url})")),
                installed_path: Some(path.display().to_string()),
            })
        }
        _ => None,
    }
}

pub fn manual_install_hint(agent: AgentKind) -> String {
    match agent {
        AgentKind::Codex => "Please install manually:\n  \
             npm: npm install -g @openai/codex\n  \
             brew: brew install --cask codex\n  \
             script: curl -fsSL https://chatgpt.com/codex/install.sh | sh\n  \
             Windows: powershell -Command \"irm https://chatgpt.com/codex/install.ps1 | iex\""
            .to_string(),
        AgentKind::ClaudeCode => "Please install manually:\n  \
             script: curl -fsSL https://claude.ai/install.sh | bash\n  \
             brew: brew install --cask claude-code\n  \
             Windows: powershell -Command \"irm https://claude.ai/install.ps1 | iex\"\n  \
             npm (deprecated): npm install -g @anthropic-ai/claude-code"
            .to_string(),
        AgentKind::OpenCode => "Please install manually:\n  \
             npm: npm i -g opencode-ai@latest\n  \
             brew: brew install anomalyco/tap/opencode\n  \
             script: curl -fsSL https://opencode.ai/install | bash\n  \
             Windows: scoop install opencode"
            .to_string(),
        AgentKind::Custom => "Custom agent does not support auto-install".to_string(),
    }
}

pub async fn install_agent(agent: AgentKind) -> crate::models::AgentInstallResult {
    let (binary_name, npm_package, brew_package, unix_script, windows_script) = match agent {
        AgentKind::Codex => (
            "codex",
            "@openai/codex",
            "codex",
            "https://chatgpt.com/codex/install.sh",
            Some("https://chatgpt.com/codex/install.ps1"),
        ),
        AgentKind::ClaudeCode => (
            "claude",
            "@anthropic-ai/claude-code",
            "claude-code",
            "https://claude.ai/install.sh",
            Some("https://claude.ai/install.ps1"),
        ),
        AgentKind::OpenCode => (
            "opencode",
            "opencode-ai",
            "anomalyco/tap/opencode",
            "https://opencode.ai/install",
            None,
        ),
        AgentKind::Custom => {
            return crate::models::AgentInstallResult {
                agent,
                success: false,
                message: Some("Custom agent does not support auto-install".to_string()),
                installed_path: None,
            };
        }
    };

    if let Some(result) = try_install_with_npm(agent, npm_package, binary_name).await {
        return result;
    }
    if let Some(result) = try_install_with_brew(agent, brew_package, binary_name).await {
        return result;
    }
    if let Some(result) =
        try_install_with_script(agent, unix_script, windows_script, binary_name).await
    {
        return result;
    }

    crate::models::AgentInstallResult {
        agent,
        success: false,
        message: Some(manual_install_hint(agent)),
        installed_path: None,
    }
}
