codex_web/

server.rs

use std::collections::HashSet;
use std::io::Error;
use std::net::Ipv4Addr;
use std::sync::Arc;

use http::HeaderValue;
use http::header::CACHE_CONTROL;
use http::header::CONTENT_SECURITY_POLICY;
use http::header::CONTENT_TYPE;
use http::header::COOKIE;
use http::header::HOST;
use http::header::ORIGIN;
use http::header::SET_COOKIE;
use serde_json::Value;
use serde_json::json;
use tokio::io::AsyncReadExt;
use tokio::io::AsyncWriteExt;
use tokio::net::TcpListener;
use tokio::net::TcpStream;
use tokio::task::JoinSet;
use tokio_util::sync::CancellationToken;
use topcoat::Result;
use topcoat::context::Cx;
use topcoat::context::app_context;
use topcoat::router::Body;
use topcoat::router::IntoResponse;
use topcoat::router::Response;
use topcoat::router::Router;
use topcoat::router::RouterBuilderDiscoverExt;
use topcoat::router::forbidden;
use topcoat::router::headers;
use topcoat::router::not_found;
use topcoat::router::page;
use topcoat::router::path_param;
use topcoat::router::route;
use topcoat::router::to_bytes;
use topcoat::router::unauthorized;
use topcoat::router::uri;

use crate::WebOptions;
use crate::auth::WebAuth;
use crate::bridge::AppServerBridge;
use crate::components::DashboardDocumentData;
use crate::components::DocumentData;
use crate::components::bootstrap_document;
use crate::components::dashboard_document;
use crate::components::document;
use crate::components::thread_title;
use crate::components::transcript_fragment;
use crate::daemon_config::AccountProfile;
use crate::daemon_config::DaemonConfig;
use crate::dashboard;
use crate::directory;
use crate::directory::DirectoryListRequest;
use crate::network::authority;
use crate::network::bind_listeners;
use crate::server_support::canonicalize_cwd;
use crate::server_support::chronological_turns;
use crate::server_support::generate_secret;
use crate::server_support::mcp_callback_url;
use crate::server_support::query_value;
use crate::server_support::thread_with_initial_turns;
use crate::ssh_tunnel::ManagedSshTunnel;
use crate::stream;

const MAX_RPC_BODY_BYTES: usize = 64 * 1024 * 1024;
const MAX_AUTH_BODY_BYTES: usize = 4 * 1024;
const APP_JS: &str = include_str!("../assets/app.js");
const APP_CSS: &str = include_str!("../assets/app.css");
const BOOTSTRAP_JS: &str = include_str!("../assets/bootstrap.js");
const HOME_JS: &str = include_str!("../assets/home.js");

struct WebState {
    auth: WebAuth,
    allowed_authorities: HashSet<String>,
    instance_id: String,
    cwd: String,
    bridge: Arc<AppServerBridge>,
    events: Arc<stream::LiveEventHub>,
    shutdown: CancellationToken,
    mcp_callback_port: Option<u16>,
    account_profile: String,
    proxy_active: bool,
}

#[topcoat::router::path_param]
struct ThreadId(str);

#[topcoat::router::path_param]
struct InstanceId(str);

#[topcoat::router::path_param]
struct CallbackId(str);

#[topcoat::router::path_param]
struct SessionId(str);

pub async fn run(options: WebOptions) -> std::io::Result<()> {
    let explicit_cwd = options.cwd.is_some();
    let cwd = options
        .cwd
        .map(canonicalize_cwd)
        .transpose()?
        .unwrap_or(std::env::current_dir()?.canonicalize()?);
    let auth = WebAuth::load(options.reset_token)?;
    let daemon_config = DaemonConfig::load(options.daemon_config.as_deref())?;
    let mut ssh_tunnel = match daemon_config.ssh_tunnel.as_ref() {
        Some(config) => Some(ManagedSshTunnel::start(config).await?),
        None => None,
    };
    let requested_port = if options.port == 0 {
        daemon_config.port.unwrap_or(0)
    } else {
        options.port
    };
    let (listeners, _port) = bind_listeners(requested_port).await?;
    let addresses = listeners
        .iter()
        .filter_map(|listener| listener.local_addr().ok())
        .collect::<Vec<_>>();
    let mut config_overrides = options.config_overrides;
    let callback_url = daemon_config
        .mcp_oauth_callback_url
        .clone()
        .or_else(|| mcp_callback_url(&addresses));
    let mcp_callback_port = if let Some(callback_url) = callback_url {
        let callback_listener = TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).await?;
        let callback_port = callback_listener.local_addr()?.port();
        drop(callback_listener);
        config_overrides.push(format!("mcp_oauth_callback_port={callback_port}"));
        config_overrides.push(format!("mcp_oauth_callback_url={callback_url:?}"));
        Some(callback_port)
    } else {
        None
    };
    let (account_profile, bridge, proxy_active) = start_account_bridge(
        &daemon_config.accounts,
        &options.app_server_command,
        &cwd,
        ssh_tunnel.as_ref().map(ManagedSshTunnel::proxy_url),
        &config_overrides,
        options.strict_config,
    )
    .await?;
    let shutdown_token = CancellationToken::new();
    let event_hub = stream::LiveEventHub::start(Arc::clone(&bridge)).await;
    let mut allowed_authorities = addresses
        .iter()
        .map(|address| authority(*address))
        .collect::<HashSet<_>>();
    if let Some(port) = addresses.first().map(std::net::SocketAddr::port) {
        allowed_authorities.insert(format!("localhost:{port}"));
    }
    let state = Arc::new(WebState {
        auth,
        allowed_authorities,
        instance_id: generate_secret()[..16].to_string(),
        cwd: directory::path_string(&cwd)?,
        bridge: Arc::clone(&bridge),
        events: event_hub,
        shutdown: shutdown_token.clone(),
        mcp_callback_port,
        account_profile: account_profile.label.clone(),
        proxy_active,
    });
    let router = Router::builder()
        .discover()
        .app_context(Arc::clone(&state))
        .build();
    let service = topcoat::router::RouterService::new(router);
    let mut servers = JoinSet::new();
    for listener in listeners {
        let service = service.clone();
        let shutdown_token = shutdown_token.clone();
        servers.spawn(async move {
            topcoat::serve_until(listener, service, async move {
                shutdown_token.cancelled().await;
            })
            .await
        });
    }

    let startup_path = if explicit_cwd {
        let return_to = format!("/new?cwd={}", urlencoding::encode(&state.cwd));
        format!("/?returnTo={}", urlencoding::encode(&return_to))
    } else {
        "/".to_string()
    };
    let urls = addresses
        .iter()
        .map(|address| {
            format!(
                "http://{}{}#bootstrap={}",
                authority(*address),
                startup_path,
                state.auth.bootstrap_token()
            )
        })
        .collect::<Vec<_>>();
    println!("\nCodex Web\n");
    for url in &urls {
        println!("  {url}");
    }
    println!("\nThis private link grants the server user's Codex and filesystem access.\n");
    println!("Account: {}", state.account_profile);
    println!(
        "OpenAI proxy: {}\n",
        if state.proxy_active {
            "enabled"
        } else {
            "disabled"
        }
    );

    if options.open_browser
        && let Some(url) = urls.iter().find(|url| url.starts_with("http://127.0.0.1:"))
        && let Err(error) = webbrowser::open(url)
    {
        tracing::warn!(%error, "failed to open Codex Web in a browser");
    }

    tokio::select! {
        signal = tokio::signal::ctrl_c() => signal?,
        result = servers.join_next() => {
            match result {
                Some(Ok(Ok(()))) | None => {}
                Some(Ok(Err(error))) => return Err(error),
                Some(Err(error)) => return Err(Error::other(error)),
            }
        }
    }
    shutdown_token.cancel();
    // Event streams are intentionally long-lived, so a graceful listener drain can wait forever
    // for browser connections after Ctrl-C. Stop accepting work, then cancel those connection
    // tasks so the CLI returns promptly.
    servers.abort_all();
    while let Some(result) = servers.join_next().await {
        match result {
            Ok(Ok(())) => {}
            Ok(Err(error)) => tracing::warn!(%error, "Codex Web listener stopped"),
            Err(error) if error.is_cancelled() => {}
            Err(error) => tracing::warn!(%error, "Codex Web listener task failed"),
        }
    }
    bridge.shutdown().await;
    if let Some(tunnel) = ssh_tunnel.as_mut() {
        tunnel.shutdown().await;
    }
    Ok(())
}

async fn start_account_bridge(
    accounts: &[AccountProfile],
    command: &crate::AppServerCommand,
    cwd: &std::path::Path,
    ssh_proxy: Option<&str>,
    config_overrides: &[String],
    strict_config: bool,
) -> std::io::Result<(AccountProfile, Arc<AppServerBridge>, bool)> {
    let mut failures = Vec::new();
    for profile in accounts {
        std::fs::create_dir_all(&profile.codex_home)?;
        let proxy = if profile.use_ssh_tunnel {
            ssh_proxy
        } else {
            profile.proxy.as_deref()
        };
        let mut overrides = config_overrides.to_vec();
        if proxy.is_some() {
            overrides.push("features.respect_system_proxy=true".to_string());
        }
        let bridge = match AppServerBridge::start(
            command,
            cwd,
            &profile.codex_home,
            proxy,
            &overrides,
            strict_config,
        )
        .await
        {
            Ok(bridge) => bridge,
            Err(error) => {
                failures.push(format!("{}: {error}", profile.name));
                continue;
            }
        };
        let account = bridge
            .request("account/read", json!({ "refreshToken": false }))
            .await;
        let available = account.as_ref().is_ok_and(|result| {
            result
                .get("account")
                .is_some_and(|account| !account.is_null())
                || result.get("requiresOpenaiAuth").and_then(Value::as_bool) == Some(false)
        });
        if available && !account_rate_limited(&bridge).await {
            return Ok((profile.clone(), bridge, proxy.is_some()));
        }
        failures.push(format!(
            "{}: {}",
            profile.name,
            if available {
                "rate limit exhausted"
            } else {
                "not signed in"
            }
        ));
        bridge.shutdown().await;
    }

    let profile = accounts
        .first()
        .ok_or_else(|| Error::new(std::io::ErrorKind::InvalidInput, "no accounts configured"))?;
    let proxy = if profile.use_ssh_tunnel {
        ssh_proxy
    } else {
        profile.proxy.as_deref()
    };
    let mut overrides = config_overrides.to_vec();
    if proxy.is_some() {
        overrides.push("features.respect_system_proxy=true".to_string());
    }
    tracing::warn!(attempts = %failures.join("; "), "all configured accounts need attention; using the first account for login");
    let bridge = AppServerBridge::start(
        command,
        cwd,
        &profile.codex_home,
        proxy,
        &overrides,
        strict_config,
    )
    .await?;
    Ok((profile.clone(), bridge, proxy.is_some()))
}

async fn account_rate_limited(bridge: &Arc<AppServerBridge>) -> bool {
    let Ok(result) = bridge.request("account/rateLimits/read", json!({})).await else {
        return false;
    };
    let snapshot = result.get("rateLimits").unwrap_or(&result);
    snapshot
        .get("rateLimitReachedType")
        .is_some_and(|value| !value.is_null())
        || snapshot.get("spendControlReached").and_then(Value::as_bool) == Some(true)
        || ["primary", "secondary"].into_iter().any(|window| {
            snapshot
                .get(window)
                .and_then(|window| window.get("usedPercent"))
                .and_then(Value::as_i64)
                .is_some_and(|used| used >= 100)
        })
}

#[page("/")]
async fn home(cx: &Cx) -> Result {
    public_state(cx)?;
    bootstrap_document(cx).await
}

#[page("/i/{instance_id}")]
async fn dashboard_page(cx: &Cx) -> Result {
    let state = authorized_state(cx)?;
    let data = dashboard::load(&state.bridge).await;
    let base_path = base_path(state);
    dashboard_document(
        cx,
        DashboardDocumentData {
            base_path: &base_path,
            projects: &data.projects,
            threads: &data.threads,
            account: data.account.as_ref(),
            truncated: data.truncated,
            profile_label: &state.account_profile,
            proxy_active: state.proxy_active,
        },
    )
    .await
}

#[page("/i/{instance_id}/new")]
async fn new_thread_page(cx: &Cx) -> Result {
    let state = authorized_state(cx)?;
    let selected_cwd = query_value(uri(cx).query().unwrap_or_default(), "cwd")
        .map(|encoded| urlencoding::decode(&encoded).map(std::borrow::Cow::into_owned))
        .transpose()
        .map_err(Error::other)?
        .map(|path| directory::canonical_directory(&path))
        .transpose()?
        .map(|path| directory::path_string(&path))
        .transpose()?;
    render_page(cx, state, None, selected_cwd.as_deref()).await
}

#[page("/i/{instance_id}/thread/{thread_id}")]
async fn thread_page(cx: &Cx) -> Result {
    let state = authorized_state(cx)?;
    render_page(cx, state, Some(path_param::<ThreadId>(cx)), None).await
}

#[route(GET "/assets/app.js")]
async fn app_js(cx: &Cx) -> Result<Response> {
    public_state(cx)?;
    static_asset(cx, "text/javascript; charset=utf-8", APP_JS)
}

#[route(GET "/assets/app.css")]
async fn app_css(cx: &Cx) -> Result<Response> {
    public_state(cx)?;
    static_asset(cx, "text/css; charset=utf-8", APP_CSS)
}

#[route(GET "/assets/bootstrap.js")]
async fn bootstrap_js(cx: &Cx) -> Result<Response> {
    public_state(cx)?;
    static_asset(cx, "text/javascript; charset=utf-8", BOOTSTRAP_JS)
}

#[route(GET "/assets/home.js")]
async fn home_js(cx: &Cx) -> Result<Response> {
    public_state(cx)?;
    static_asset(cx, "text/javascript; charset=utf-8", HOME_JS)
}

#[route(POST "/auth/exchange")]
async fn exchange_token(cx: &Cx, body: Body) -> Result<Response> {
    let state = public_state(cx)?;
    authorize_mutation(cx, state)?;
    let bytes = to_bytes(body, MAX_AUTH_BODY_BYTES)
        .await
        .map_err(Error::other)?;
    let payload: Value = serde_json::from_slice(&bytes)?;
    let token = payload
        .get("token")
        .and_then(Value::as_str)
        .ok_or_else(unauthorized)?;
    if !state.auth.verify_bootstrap(token) {
        return Err(unauthorized().into());
    }
    let base_path = base_path(state);
    let mut response = json_response(
        cx,
        json!({
            "authenticated": true,
            "basePath": base_path,
        }),
    )?;
    response.headers_mut().insert(
        SET_COOKIE,
        HeaderValue::from_str(&state.auth.session_cookie(&base_path)?).map_err(Error::other)?,
    );
    Ok(response)
}

#[route(GET "/i/{instance_id}/events")]
async fn event_stream(cx: &Cx) -> Result<Response> {
    let state = authorized_state(cx)?;
    let last_event_id = headers(cx)
        .get("last-event-id")
        .and_then(|value| value.to_str().ok())
        .and_then(|value| value.parse().ok())
        .or_else(|| {
            query_value(uri(cx).query().unwrap_or_default(), "after")
                .and_then(|value| value.parse().ok())
        });
    Ok(stream::response(Arc::clone(&state.events), last_event_id).await)
}

#[route(GET "/oauth/callback/{callback_id}")]
async fn mcp_oauth_callback(cx: &Cx) -> Result<Response> {
    let state = public_state(cx)?;
    let callback_port = state
        .mcp_callback_port
        .ok_or_else(|| Error::other("remote MCP OAuth callback is not configured"))?;
    let callback_id = path_param::<CallbackId>(cx);
    let query = uri(cx).query().unwrap_or_default();
    if !valid_callback_id(callback_id) || query.bytes().any(|byte| byte.is_ascii_control()) {
        return Err(not_found().into());
    }
    let callback_path = format!("/oauth/callback/{callback_id}?{query}");
    let mut stream = TcpStream::connect((Ipv4Addr::LOCALHOST, callback_port))
        .await
        .map_err(Error::other)?;
    stream
        .write_all(
            format!(
                "GET {callback_path} HTTP/1.1\r\nHost: 127.0.0.1:{callback_port}\r\nConnection: close\r\n\r\n"
            )
            .as_bytes(),
        )
        .await
        .map_err(Error::other)?;
    let mut encoded = Vec::new();
    stream
        .take(64 * 1024)
        .read_to_end(&mut encoded)
        .await
        .map_err(Error::other)?;
    let forwarded = String::from_utf8_lossy(&encoded);
    let body = forwarded
        .split_once("\r\n\r\n")
        .map(|(_, body)| body)
        .unwrap_or("MCP OAuth callback returned an invalid response")
        .to_string();
    let mut response = body.into_response(cx)?;
    response.headers_mut().insert(
        CONTENT_TYPE,
        HeaderValue::from_static("text/plain; charset=utf-8"),
    );
    response
        .headers_mut()
        .insert(CACHE_CONTROL, HeaderValue::from_static("no-store"));
    response.headers_mut().insert(
        CONTENT_SECURITY_POLICY,
        HeaderValue::from_static("default-src 'none'; frame-ancestors 'none'"),
    );
    Ok(response)
}

#[route(GET "/i/{instance_id}/api/session/{session_id}")]
async fn session_fragment(cx: &Cx) -> Result<Response> {
    let state = authorized_state(cx)?;
    let session_id = path_param::<SessionId>(cx);
    let base_seq = state.events.watermark();
    let active_request = async {
        if session_id == "new" {
            None
        } else {
            let paginated = state
                .bridge
                .request(
                    "thread/resume",
                    json!({
                        "threadId": session_id,
                        "excludeTurns": true,
                        "initialTurnsPage": {
                            "limit": 30,
                            "sortDirection": "desc",
                            "itemsView": "full"
                        }
                    }),
                )
                .await;
            match paginated {
                Ok(response) => Some(response),
                Err(_) => state
                    .bridge
                    .request("thread/resume", json!({ "threadId": session_id }))
                    .await
                    .ok(),
            }
        }
    };
    let (active, approvals) =
        tokio::join!(active_request, state.bridge.outstanding_server_requests(),);
    let active_thread = active.as_ref().and_then(|response| response.get("thread"));
    let hydrated_thread = active.as_ref().and_then(thread_with_initial_turns);
    let display_thread = hydrated_thread.as_ref().or(active_thread);
    let fragment = transcript_fragment(cx, display_thread, &approvals).await?;
    let active_turn_id = display_thread
        .and_then(|active_thread| active_thread.get("turns"))
        .and_then(Value::as_array)
        .and_then(|turns| {
            turns
                .iter()
                .rev()
                .find(|turn| turn.get("status").and_then(Value::as_str) == Some("inProgress"))
        })
        .and_then(|turn| turn.get("id"))
        .and_then(Value::as_str)
        .unwrap_or_default();
    let payload = json!({
        "html": fragment.render(cx),
        "baseSeq": base_seq,
        "instanceId": state.instance_id,
        "nextCursor": active.as_ref().and_then(|response| response.pointer("/initialTurnsPage/nextCursor")),
        "threadId": display_thread
            .and_then(|active_thread| active_thread.get("id"))
            .and_then(Value::as_str)
            .unwrap_or_default(),
        "activeTurnId": active_turn_id,
        "title": display_thread.map(thread_title).unwrap_or("New task"),
        "cwd": display_thread
            .and_then(|active_thread| active_thread.get("cwd"))
            .and_then(Value::as_str)
            .unwrap_or_default(),
        "model": active
            .as_ref()
            .and_then(|response| response.get("model"))
            .and_then(Value::as_str),
        "reasoningEffort": active
            .as_ref()
            .and_then(|response| response.get("reasoningEffort"))
            .and_then(Value::as_str),
        "permissionMode": permission_mode(active.as_ref()),
    });
    let mut response = serde_json::to_vec(&payload)?.into_response(cx)?;
    response
        .headers_mut()
        .insert(CONTENT_TYPE, HeaderValue::from_static("application/json"));
    response
        .headers_mut()
        .insert(CACHE_CONTROL, HeaderValue::from_static("no-store"));
    Ok(response)
}

#[route(GET "/i/{instance_id}/api/session/{session_id}/turns")]
async fn earlier_turns(cx: &Cx) -> Result<Response> {
    let state = authorized_state(cx)?;
    let session_id = path_param::<SessionId>(cx);
    let cursor = query_value(uri(cx).query().unwrap_or_default(), "cursor");
    let page = state
        .bridge
        .request(
            "thread/turns/list",
            json!({
                "threadId": session_id,
                "cursor": cursor,
                "limit": 30,
                "sortDirection": "desc",
                "itemsView": "full"
            }),
        )
        .await
        .map_err(Error::other)?;
    let thread_value = json!({ "turns": chronological_turns(page.get("data")) });
    let fragment = transcript_fragment(cx, Some(&thread_value), &[]).await?;
    json_response(
        cx,
        json!({
            "html": fragment.render(cx),
            "nextCursor": page.get("nextCursor").cloned().unwrap_or(Value::Null)
        }),
    )
}

#[route(POST "/i/{instance_id}/api/shutdown")]
async fn shutdown_process(cx: &Cx) -> Result<Response> {
    let state = authorized_state(cx)?;
    authorize_mutation(cx, state)?;
    state.shutdown.cancel();
    json_response(cx, json!({ "stopping": true }))
}

#[route(POST "/i/{instance_id}/api/rpc")]
async fn rpc(cx: &Cx, body: Body) -> Result<Response> {
    let state = authorized_state(cx)?;
    authorize_mutation(cx, state)?;
    let bytes = to_bytes(body, MAX_RPC_BODY_BYTES)
        .await
        .map_err(Error::other)?;
    let mut message: Value = serde_json::from_slice(&bytes)?;
    if message.get("method").and_then(Value::as_str) == Some("thread/start")
        && let Some(cwd) = message
            .pointer("/params/cwd")
            .and_then(Value::as_str)
            .map(ToOwned::to_owned)
    {
        let canonical = directory::canonical_directory(&cwd)?;
        if let Some(params) = message.get_mut("params").and_then(Value::as_object_mut) {
            params.insert(
                "cwd".to_string(),
                Value::String(directory::path_string(&canonical)?),
            );
        }
    }
    let envelope = if let Some(method) = message.get("method").and_then(Value::as_str) {
        state
            .bridge
            .request_envelope(
                method,
                message.get("params").cloned().unwrap_or_else(|| json!({})),
            )
            .await
            .map_err(Error::other)?
    } else {
        state.bridge.respond(message).await.map_err(Error::other)?;
        json!({ "result": {} })
    };
    let mut response = serde_json::to_vec(&envelope)?.into_response(cx)?;
    response
        .headers_mut()
        .insert(CONTENT_TYPE, HeaderValue::from_static("application/json"));
    response
        .headers_mut()
        .insert(CACHE_CONTROL, HeaderValue::from_static("no-store"));
    Ok(response)
}

#[route(POST "/i/{instance_id}/api/fs/list")]
async fn list_directory(cx: &Cx, body: Body) -> Result<Response> {
    let state = authorized_state(cx)?;
    authorize_mutation(cx, state)?;
    let bytes = to_bytes(body, MAX_AUTH_BODY_BYTES)
        .await
        .map_err(Error::other)?;
    let request: DirectoryListRequest = serde_json::from_slice(&bytes)?;
    match tokio::task::spawn_blocking(move || directory::list(request)).await {
        Ok(Ok(listing)) => {
            let payload = serde_json::to_value(listing)?;
            json_response(cx, payload)
        }
        Ok(Err(error)) => json_response(cx, json!({ "error": error.to_string() })),
        Err(error) => Err(Error::other(error).into()),
    }
}

async fn render_page(
    cx: &Cx,
    state: &WebState,
    thread_id: Option<&str>,
    selected_cwd: Option<&str>,
) -> Result {
    let base_seq = state.events.watermark();
    let threads_request = state.bridge.request(
        "thread/list",
        json!({ "limit": 100, "sortKey": "recency_at", "sortDirection": "desc" }),
    );
    let models_request = state.bridge.request("model/list", json!({ "limit": 100 }));
    let collaboration_modes_request = state.bridge.request("collaborationMode/list", json!({}));
    let mcp_servers_request = state.bridge.request(
        "mcpServerStatus/list",
        json!({ "limit": 100, "detail": "toolsAndAuthOnly" }),
    );
    let active_request = async {
        match thread_id {
            Some(thread_id) => state
                .bridge
                .request(
                    "thread/resume",
                    json!({
                        "threadId": thread_id,
                        "excludeTurns": true,
                        "initialTurnsPage": {
                            "limit": 30,
                            "sortDirection": "desc",
                            "itemsView": "full"
                        }
                    }),
                )
                .await
                .ok(),
            None => None,
        }
    };
    let approvals_request = state.bridge.outstanding_server_requests();
    let (threads, models, collaboration_modes, mcp_servers, active, approvals) = tokio::join!(
        threads_request,
        models_request,
        collaboration_modes_request,
        mcp_servers_request,
        active_request,
        approvals_request,
    );
    let threads = threads.unwrap_or_else(|_| json!({ "data": [] }));
    let models = models.unwrap_or_else(|_| json!({ "data": [] }));
    let collaboration_modes = collaboration_modes.unwrap_or_else(|_| json!({ "data": [] }));
    let mcp_servers = mcp_servers.unwrap_or_else(|_| json!({ "data": [] }));
    let hydrated_thread = active.as_ref().and_then(thread_with_initial_turns);
    let active_thread = hydrated_thread
        .as_ref()
        .or_else(|| active.as_ref().and_then(|response| response.get("thread")));
    let workspace_cwd = page_workspace_cwd(active_thread, selected_cwd, &state.cwd);
    let base_path = base_path(state);
    document(
        cx,
        DocumentData {
            base_path: &base_path,
            instance_id: &state.instance_id,
            base_seq,
            threads: threads
                .get("data")
                .and_then(Value::as_array)
                .map(Vec::as_slice)
                .unwrap_or_default(),
            active_thread,
            active_model: active
                .as_ref()
                .and_then(|response| response.get("model"))
                .and_then(Value::as_str)
                .unwrap_or_default(),
            active_effort: active
                .as_ref()
                .and_then(|response| response.get("reasoningEffort"))
                .and_then(Value::as_str)
                .unwrap_or_default(),
            active_permission_mode: permission_mode(active.as_ref()),
            models: models
                .get("data")
                .and_then(Value::as_array)
                .map(Vec::as_slice)
                .unwrap_or_default(),
            collaboration_modes: collaboration_modes
                .get("data")
                .and_then(Value::as_array)
                .map(Vec::as_slice)
                .unwrap_or_default(),
            approvals: &approvals,
            workspace_cwd,
            mcp_servers: mcp_servers
                .get("data")
                .and_then(Value::as_array)
                .map(Vec::as_slice)
                .unwrap_or_default(),
            initial_next_cursor: active
                .as_ref()
                .and_then(|response| response.pointer("/initialTurnsPage/nextCursor"))
                .and_then(Value::as_str),
        },
    )
    .await
}

fn page_workspace_cwd<'a>(
    active_thread: Option<&'a Value>,
    selected_cwd: Option<&'a str>,
    daemon_cwd: &'a str,
) -> &'a str {
    active_thread
        .and_then(|thread| thread.get("cwd"))
        .and_then(Value::as_str)
        .or(selected_cwd)
        .unwrap_or(daemon_cwd)
}

fn permission_mode(response: Option<&Value>) -> &'static str {
    let profile_id = response
        .and_then(|response| response.pointer("/activePermissionProfile/id"))
        .and_then(Value::as_str);
    match profile_id {
        Some(":danger-full-access") => "full-access",
        Some(":read-only") => "read-only",
        Some(":workspace") => "workspace",
        _ => {
            let approval = response
                .and_then(|response| response.get("approvalPolicy"))
                .and_then(Value::as_str);
            let sandbox = response
                .and_then(|response| response.pointer("/sandbox/type"))
                .and_then(Value::as_str);
            match (approval, sandbox) {
                (Some("never"), Some("danger-full-access")) => "full-access",
                (_, Some("read-only")) => "read-only",
                _ => "workspace",
            }
        }
    }
}

fn authorized_state(cx: &Cx) -> Result<&WebState> {
    let state = instance_state(cx)?;
    if !is_authorized(cx, state) {
        return Err(not_found().into());
    }
    Ok(state)
}

fn instance_state(cx: &Cx) -> Result<&WebState> {
    let state = public_state(cx)?;
    if path_param::<InstanceId>(cx) != state.instance_id {
        return Err(not_found().into());
    }
    Ok(state)
}

fn base_path(state: &WebState) -> String {
    format!("/i/{}", state.instance_id)
}

fn valid_callback_id(callback_id: &str) -> bool {
    callback_id.len() == 12
        && callback_id
            .bytes()
            .all(|byte| byte.is_ascii_alphanumeric() || byte == b'-' || byte == b'_')
}

fn public_state(cx: &Cx) -> Result<&WebState> {
    let state: &Arc<WebState> = app_context(cx);
    let host = headers(cx)
        .get(HOST)
        .and_then(|value| value.to_str().ok())
        .ok_or_else(forbidden)?;
    if !state.allowed_authorities.contains(host) {
        return Err(forbidden().into());
    }
    Ok(state)
}

fn is_authorized(cx: &Cx, state: &WebState) -> bool {
    state.auth.authorize_cookie_header(
        headers(cx)
            .get(COOKIE)
            .and_then(|value| value.to_str().ok()),
    )
}

fn authorize_mutation(cx: &Cx, state: &WebState) -> Result<()> {
    let host = headers(cx)
        .get(HOST)
        .and_then(|value| value.to_str().ok())
        .ok_or_else(forbidden)?;
    let origin = headers(cx)
        .get(ORIGIN)
        .and_then(|value| value.to_str().ok())
        .ok_or_else(forbidden)?;
    if !state.allowed_authorities.contains(host) || origin != format!("http://{host}") {
        return Err(forbidden().into());
    }
    Ok(())
}

fn static_asset(cx: &Cx, content_type: &'static str, body: &'static str) -> Result<Response> {
    let mut response = body.into_response(cx)?;
    response
        .headers_mut()
        .insert(CONTENT_TYPE, HeaderValue::from_static(content_type));
    response
        .headers_mut()
        .insert(CACHE_CONTROL, HeaderValue::from_static("no-store"));
    Ok(response)
}

fn json_response(cx: &Cx, payload: Value) -> Result<Response> {
    let mut response = serde_json::to_vec(&payload)?.into_response(cx)?;
    response
        .headers_mut()
        .insert(CONTENT_TYPE, HeaderValue::from_static("application/json"));
    response
        .headers_mut()
        .insert(CACHE_CONTROL, HeaderValue::from_static("no-store"));
    Ok(response)
}

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