codex_mobile_bridge/state/

runtime.rs

use std::path::Path;
use std::sync::Arc;

use anyhow::{Context, Result, bail};
use serde_json::{Value, json};
use tokio::sync::{RwLock, mpsc};
use tokio::time::timeout;
use uuid::Uuid;

use super::{BridgeState, CLIENT_STATUS_TIMEOUT};
use crate::app_server::{AppServerInbound, AppServerLaunchConfig, AppServerManager};
use crate::bridge_protocol::{
    AppServerHandshakeSummary, GetRuntimeStatusRequest, PruneRuntimeRequest, RestartRuntimeRequest,
    RuntimeRecord, RuntimeStatusSnapshot, RuntimeSummary, StartRuntimeRequest, StopRuntimeRequest,
    now_millis, runtime_list_payload, status_payload,
};
use crate::config::expand_path;

pub(super) struct ManagedRuntime {
    pub(super) record: RuntimeRecord,
    pub(super) app_server: AppServerManager,
    pub(super) status: RwLock<RuntimeStatusSnapshot>,
}

impl ManagedRuntime {
    pub(super) async fn summary(&self) -> RuntimeSummary {
        RuntimeSummary::from_parts(&self.record, self.status.read().await.clone())
    }
}

impl BridgeState {
    pub async fn runtime_snapshot(&self) -> RuntimeStatusSnapshot {
        match self.get_runtime(&self.primary_runtime_id).await {
            Some(runtime) => runtime.status.read().await.clone(),
            None => RuntimeStatusSnapshot::stopped(self.primary_runtime_id.clone()),
        }
    }

    pub async fn runtime_summaries(&self) -> Vec<RuntimeSummary> {
        let runtimes = self.runtimes.read().await;
        let mut summaries = Vec::with_capacity(runtimes.len());
        for runtime in runtimes.values() {
            summaries.push(runtime.summary().await);
        }
        summaries.sort_by(|left, right| {
            right
                .is_primary
                .cmp(&left.is_primary)
                .then_with(|| left.display_name.cmp(&right.display_name))
        });
        summaries
    }

    pub async fn runtime_snapshot_for_client(&self) -> RuntimeStatusSnapshot {
        match timeout(CLIENT_STATUS_TIMEOUT, self.runtime_snapshot()).await {
            Ok(snapshot) => snapshot,
            Err(_) => {
                self.log_timeout_warning(
                    "runtime_snapshot:primary",
                    "读取 primary runtime 状态超时，回退到存储快照",
                );
                self.fallback_runtime_snapshot()
            }
        }
    }

    pub async fn runtime_summaries_for_client(&self) -> Vec<RuntimeSummary> {
        match timeout(CLIENT_STATUS_TIMEOUT, self.runtime_summaries()).await {
            Ok(summaries) => summaries,
            Err(_) => {
                self.log_timeout_warning(
                    "runtime_summaries:list",
                    "读取 runtime 列表超时，回退到存储快照",
                );
                self.fallback_runtime_summaries()
            }
        }
    }

    pub(super) async fn get_runtime_status(
        &self,
        request: GetRuntimeStatusRequest,
    ) -> Result<Value> {
        let runtime_id = request
            .runtime_id
            .unwrap_or_else(|| self.primary_runtime_id.clone());
        let runtime = match timeout(
            CLIENT_STATUS_TIMEOUT,
            self.require_runtime(Some(&runtime_id)),
        )
        .await
        {
            Ok(Ok(runtime)) => runtime.summary().await,
            Ok(Err(error)) => return Err(error),
            Err(_) => {
                self.log_timeout_warning(
                    &format!("runtime_status:{runtime_id}"),
                    &format!("读取 runtime={runtime_id} 状态超时，回退到存储快照"),
                );
                self.fallback_runtime_summary(&runtime_id)?
            }
        };
        Ok(json!({ "runtime": runtime }))
    }

    pub(super) async fn start_runtime(&self, request: StartRuntimeRequest) -> Result<Value> {
        let runtime_id = match request.runtime_id.clone() {
            Some(runtime_id) => runtime_id,
            None => format!("runtime-{}", &Uuid::new_v4().simple().to_string()[..8]),
        };

        if self.get_runtime(&runtime_id).await.is_some() {
            if request.display_name.is_some()
                || request.codex_home.is_some()
                || request.codex_binary.is_some()
            {
                bail!("已存在 runtime，启动现有 runtime 时不能覆盖配置");
            }

            self.start_existing_runtime(&runtime_id).await?;
            let runtime = self
                .require_runtime(Some(&runtime_id))
                .await?
                .summary()
                .await;
            return Ok(json!({ "runtime": runtime }));
        }

        {
            let runtimes = self.runtimes.read().await;
            if runtimes.len() >= self.runtime_limit {
                bail!("已达到 runtime 上限 {}", self.runtime_limit);
            }
        }

        let codex_home = request
            .codex_home
            .as_deref()
            .map(|value| expand_path(Path::new(value)))
            .transpose()?
            .map(|value| value.to_string_lossy().to_string());
        let now = now_millis();
        let record = RuntimeRecord {
            runtime_id: runtime_id.clone(),
            display_name: request
                .display_name
                .filter(|value| !value.trim().is_empty())
                .unwrap_or_else(|| runtime_id.clone()),
            codex_home,
            codex_binary: request
                .codex_binary
                .filter(|value| !value.trim().is_empty())
                .unwrap_or_else(|| "codex".to_string()),
            is_primary: false,
            auto_start: request.auto_start.unwrap_or(false),
            created_at_ms: now,
            updated_at_ms: now,
        };

        self.storage.upsert_runtime(&record)?;
        self.register_runtime(record).await?;
        self.start_existing_runtime(&runtime_id).await?;

        let runtime = self
            .require_runtime(Some(&runtime_id))
            .await?
            .summary()
            .await;
        Ok(json!({ "runtime": runtime }))
    }

    pub(super) async fn stop_runtime(&self, request: StopRuntimeRequest) -> Result<Value> {
        let runtime = self.require_runtime(Some(&request.runtime_id)).await?;
        let current_status = runtime.status.read().await.clone();
        self.emit_runtime_status(
            &request.runtime_id,
            RuntimeStatusSnapshot {
                runtime_id: request.runtime_id.clone(),
                status: "stopping".to_string(),
                codex_home: current_status.codex_home,
                user_agent: current_status.user_agent,
                platform_family: current_status.platform_family,
                platform_os: current_status.platform_os,
                last_error: None,
                pid: current_status.pid,
                app_server_handshake: AppServerHandshakeSummary::inactive(),
                updated_at_ms: now_millis(),
            },
        )
        .await?;
        runtime.app_server.stop().await?;

        let runtime = self
            .require_runtime(Some(&request.runtime_id))
            .await?
            .summary()
            .await;
        Ok(json!({ "runtime": runtime }))
    }

    pub(super) async fn restart_runtime(&self, request: RestartRuntimeRequest) -> Result<Value> {
        let runtime = self.require_runtime(Some(&request.runtime_id)).await?;
        let current_status = runtime.status.read().await.clone();
        self.emit_runtime_status(
            &request.runtime_id,
            RuntimeStatusSnapshot {
                runtime_id: request.runtime_id.clone(),
                status: "stopping".to_string(),
                codex_home: current_status.codex_home,
                user_agent: current_status.user_agent,
                platform_family: current_status.platform_family,
                platform_os: current_status.platform_os,
                last_error: None,
                pid: current_status.pid,
                app_server_handshake: AppServerHandshakeSummary::inactive(),
                updated_at_ms: now_millis(),
            },
        )
        .await?;
        runtime.app_server.restart().await?;

        let runtime = self
            .require_runtime(Some(&request.runtime_id))
            .await?
            .summary()
            .await;
        Ok(json!({ "runtime": runtime }))
    }

    pub(super) async fn prune_runtime(&self, request: PruneRuntimeRequest) -> Result<Value> {
        if request.runtime_id == self.primary_runtime_id {
            bail!("primary runtime 不能删除");
        }

        let runtime = self.require_runtime(Some(&request.runtime_id)).await?;
        runtime.app_server.stop().await?;

        {
            let mut runtimes = self.runtimes.write().await;
            runtimes.remove(&request.runtime_id);
        }
        self.storage.remove_runtime(&request.runtime_id)?;
        self.emit_runtime_list().await?;

        Ok(json!({ "pruned": true, "runtimeId": request.runtime_id }))
    }

    pub(super) async fn start_existing_runtime(&self, runtime_id: &str) -> Result<()> {
        let runtime = self.require_runtime(Some(runtime_id)).await?;
        runtime.app_server.start().await
    }

    pub(super) async fn register_runtime(&self, record: RuntimeRecord) -> Result<()> {
        let managed = Arc::new(Self::build_runtime(record, self.inbound_tx.clone()));
        {
            let mut runtimes = self.runtimes.write().await;
            runtimes.insert(managed.record.runtime_id.clone(), managed);
        }
        self.emit_runtime_list().await?;
        Ok(())
    }

    pub(super) fn build_runtime(
        record: RuntimeRecord,
        inbound_tx: mpsc::UnboundedSender<AppServerInbound>,
    ) -> ManagedRuntime {
        ManagedRuntime {
            status: RwLock::new(RuntimeStatusSnapshot::stopped(record.runtime_id.clone())),
            app_server: AppServerManager::new(
                AppServerLaunchConfig {
                    runtime_id: record.runtime_id.clone(),
                    codex_binary: record.codex_binary.clone(),
                    codex_home: record.codex_home.as_ref().map(std::path::PathBuf::from),
                },
                inbound_tx,
            ),
            record,
        }
    }

    pub(super) async fn get_runtime(&self, runtime_id: &str) -> Option<Arc<ManagedRuntime>> {
        self.runtimes.read().await.get(runtime_id).cloned()
    }

    pub(super) async fn require_runtime(
        &self,
        runtime_id: Option<&str>,
    ) -> Result<Arc<ManagedRuntime>> {
        let target = runtime_id.unwrap_or(&self.primary_runtime_id);
        self.get_runtime(target)
            .await
            .with_context(|| format!("未找到 runtime: {target}"))
    }

    pub(super) async fn emit_runtime_status(
        &self,
        runtime_id: &str,
        status: RuntimeStatusSnapshot,
    ) -> Result<()> {
        let status_label = status.status.clone();
        let runtime = self.require_runtime(Some(runtime_id)).await?;
        {
            let mut guard = runtime.status.write().await;
            *guard = status;
        }
        let summary = runtime.summary().await;
        self.emit_event(
            "runtime_status_changed",
            Some(runtime_id),
            None,
            status_payload(&summary),
        )?;
        if matches!(status_label.as_str(), "stopping" | "stopped" | "error") {
            let reason = match status_label.as_str() {
                "stopping" => "runtime 正在停止，线程状态待重新加载",
                "stopped" => "runtime 已停止，线程状态待重新加载",
                _ => "runtime 异常退出，线程状态待重新加载",
            };
            self.downgrade_runtime_threads(runtime_id, reason)?;
        }
        Ok(())
    }

    pub(super) async fn emit_runtime_process_changed(
        &self,
        runtime_id: &str,
        pid: Option<u32>,
        running: bool,
    ) -> Result<()> {
        let runtime = self.require_runtime(Some(runtime_id)).await?;
        {
            let mut guard = runtime.status.write().await;
            guard.pid = pid.map(|value| value as i32);
            guard.updated_at_ms = now_millis();
            if !running && guard.status == "starting" {
                guard.status = "stopped".to_string();
                guard.last_error = None;
                guard.app_server_handshake = AppServerHandshakeSummary::inactive();
            }
        }
        self.emit_event(
            "runtime_process_changed",
            Some(runtime_id),
            None,
            json!({
                "runtimeId": runtime_id,
                "pid": pid,
                "running": running,
                "updatedAtMs": now_millis(),
            }),
        )?;
        Ok(())
    }

    pub(super) async fn emit_runtime_degraded(
        &self,
        runtime_id: &str,
        message: String,
    ) -> Result<()> {
        self.emit_event(
            "runtime_degraded",
            Some(runtime_id),
            None,
            json!({
                "runtimeId": runtime_id,
                "message": message,
                "updatedAtMs": now_millis(),
            }),
        )
    }

    pub(super) async fn emit_runtime_list(&self) -> Result<()> {
        let runtimes = self.runtime_summaries().await;
        self.emit_event("runtime_list", None, None, runtime_list_payload(&runtimes))
    }

    pub(super) fn fallback_runtime_snapshot(&self) -> RuntimeStatusSnapshot {
        self.storage
            .get_runtime(&self.primary_runtime_id)
            .ok()
            .flatten()
            .map(|record| self.build_fallback_status(&record))
            .unwrap_or_else(|| RuntimeStatusSnapshot::stopped(self.primary_runtime_id.clone()))
    }

    pub(super) fn fallback_runtime_summaries(&self) -> Vec<RuntimeSummary> {
        let mut summaries = self
            .storage
            .list_runtimes()
            .unwrap_or_default()
            .into_iter()
            .map(|record| RuntimeSummary::from_parts(&record, self.build_fallback_status(&record)))
            .collect::<Vec<_>>();
        summaries.sort_by(|left, right| {
            right
                .is_primary
                .cmp(&left.is_primary)
                .then_with(|| left.display_name.cmp(&right.display_name))
        });
        summaries
    }

    pub(super) fn fallback_runtime_summary(&self, runtime_id: &str) -> Result<RuntimeSummary> {
        let record = self
            .storage
            .get_runtime(runtime_id)?
            .with_context(|| format!("未找到 runtime: {runtime_id}"))?;
        Ok(RuntimeSummary::from_parts(
            &record,
            self.build_fallback_status(&record),
        ))
    }

    fn build_fallback_status(&self, record: &RuntimeRecord) -> RuntimeStatusSnapshot {
        RuntimeStatusSnapshot {
            runtime_id: record.runtime_id.clone(),
            status: if record.auto_start {
                "unknown".to_string()
            } else {
                "stopped".to_string()
            },
            codex_home: record.codex_home.clone(),
            user_agent: None,
            platform_family: None,
            platform_os: None,
            last_error: None,
            pid: None,
            app_server_handshake: AppServerHandshakeSummary::inactive(),
            updated_at_ms: now_millis(),
        }
    }

    pub(super) async fn transition_runtime_status(
        &self,
        runtime_id: &str,
        status_label: String,
        last_error: Option<String>,
        handshake: AppServerHandshakeSummary,
    ) -> Result<()> {
        let runtime = self.require_runtime(Some(runtime_id)).await?;
        let current_status = runtime.status.read().await.clone();
        self.emit_runtime_status(
            runtime_id,
            RuntimeStatusSnapshot {
                runtime_id: runtime_id.to_string(),
                status: status_label,
                codex_home: current_status.codex_home,
                user_agent: current_status.user_agent,
                platform_family: current_status.platform_family,
                platform_os: current_status.platform_os,
                last_error,
                pid: current_status.pid,
                app_server_handshake: handshake,
                updated_at_ms: now_millis(),
            },
        )
        .await
    }

    fn downgrade_runtime_threads(&self, runtime_id: &str, reason: &str) -> Result<()> {
        for thread in self
            .storage
            .downgrade_runtime_threads(runtime_id, Some(reason))?
        {
            self.emit_event(
                "thread/status/changed",
                Some(runtime_id),
                Some(&thread.id),
                json!({
                    "runtimeId": runtime_id,
                    "threadId": thread.id,
                    "status": thread.status_info.raw,
                    "statusInfo": thread.status_info,
                }),
            )?;
        }
        Ok(())
    }
}