use std::collections::HashSet;
use std::io::Error;
use std::net::Ipv4Addr;
use std::sync::Arc;
use std::sync::RwLock;
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::Deserialize;
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");
const SETTINGS_JS: &str = include_str!("../assets/settings.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,
account_profile_name: String,
daemon_config: RwLock<DaemonConfig>,
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(),
account_profile_name: account_profile.name.clone(),
daemon_config: RwLock::new(daemon_config),
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();
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 accounts = account_profiles(state)?;
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,
mcp_servers: &data.mcp_servers,
accounts: &accounts,
active_account: &state.account_profile_name,
},
)
.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(GET "/assets/settings.js")]
async fn settings_js(cx: &Cx) -> Result<Response> {
public_state(cx)?;
static_asset(cx, "text/javascript; charset=utf-8", SETTINGS_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)
}
#[derive(Deserialize)]
#[serde(
tag = "action",
rename_all = "camelCase",
rename_all_fields = "camelCase"
)]
enum AccountProfilesRequest {
Add {
name: String,
label: String,
codex_home: std::path::PathBuf,
proxy: Option<String>,
use_ssh_tunnel: bool,
},
Remove {
name: String,
},
Move {
name: String,
direction: AccountMoveDirection,
},
}
#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
enum AccountMoveDirection {
Up,
Down,
}
#[route(POST "/i/{instance_id}/api/account-profiles")]
async fn update_account_profiles(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: AccountProfilesRequest = serde_json::from_slice(&bytes)?;
let mut config = state
.daemon_config
.write()
.map_err(|_| Error::other("daemon config lock is poisoned"))?;
let mut accounts = config.accounts.clone();
match request {
AccountProfilesRequest::Add {
name,
label,
codex_home,
proxy,
use_ssh_tunnel,
} => accounts.push(AccountProfile {
label: if label.trim().is_empty() {
name.clone()
} else {
label
},
name,
codex_home,
proxy: proxy.filter(|value| !value.trim().is_empty()),
use_ssh_tunnel,
}),
AccountProfilesRequest::Remove { name } => {
if config
.accounts
.iter()
.find(|account| account.name == name)
.is_some_and(|account| account.name == state.account_profile_name)
{
return json_response(
cx,
json!({ "error": "the active account cannot be removed until the daemon restarts" }),
);
}
accounts.retain(|account| account.name != name);
}
AccountProfilesRequest::Move { name, direction } => {
let Some(index) = accounts.iter().position(|account| account.name == name) else {
return json_response(cx, json!({ "error": "account profile not found" }));
};
let target = match direction {
AccountMoveDirection::Up => index.saturating_sub(1),
AccountMoveDirection::Down => (index + 1).min(accounts.len() - 1),
};
accounts.swap(index, target);
}
}
match config.save_accounts(accounts) {
Ok(()) => json_response(cx, json!({ "restartRequired": true })),
Err(error) => json_response(cx, json!({ "error": error.to_string() })),
}
}
#[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);
let accounts = account_profiles(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(),
accounts: &accounts,
active_account: &state.account_profile_name,
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 account_profiles(state: &WebState) -> Result<Vec<AccountProfile>> {
state
.daemon_config
.read()
.map(|config| config.accounts.clone())
.map_err(|_| Error::other("daemon config lock is poisoned").into())
}
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;