codex_web/
server.rs

1use std::io::Error;
2use std::net::Ipv4Addr;
3use std::sync::Arc;
4
5use http::HeaderValue;
6use http::header::CACHE_CONTROL;
7use http::header::CONTENT_TYPE;
8use serde_json::Value;
9use serde_json::json;
10use tokio::io::AsyncReadExt;
11use tokio::io::AsyncWriteExt;
12use tokio::net::TcpListener;
13use tokio::net::TcpStream;
14use tokio::task::JoinSet;
15use tokio_util::sync::CancellationToken;
16use topcoat::Result;
17use topcoat::context::Cx;
18use topcoat::context::app_context;
19use topcoat::router::Body;
20use topcoat::router::IntoResponse;
21use topcoat::router::Response;
22use topcoat::router::Router;
23use topcoat::router::RouterBuilderDiscoverExt;
24use topcoat::router::headers;
25use topcoat::router::not_found;
26use topcoat::router::page;
27use topcoat::router::path_param;
28use topcoat::router::route;
29use topcoat::router::to_bytes;
30use topcoat::router::uri;
31
32use crate::WebOptions;
33use crate::bridge::AppServerBridge;
34use crate::components::DocumentData;
35use crate::components::document;
36use crate::components::thread_title;
37use crate::components::transcript_fragment;
38use crate::network::authority;
39use crate::network::bind_listeners;
40use crate::server_support::canonicalize_cwd;
41use crate::server_support::chronological_turns;
42use crate::server_support::generate_secret;
43use crate::server_support::mcp_callback_url;
44use crate::server_support::query_value;
45use crate::server_support::thread_with_initial_turns;
46use crate::stream;
47
48const MAX_RPC_BODY_BYTES: usize = 64 * 1024 * 1024;
49const APP_JS: &str = include_str!("../assets/app.js");
50const APP_CSS: &str = include_str!("../assets/app.css");
51
52struct WebState {
53    secret: String,
54    instance_id: String,
55    cwd: String,
56    bridge: Arc<AppServerBridge>,
57    events: Arc<stream::LiveEventHub>,
58    shutdown: CancellationToken,
59    mcp_callback_port: Option<u16>,
60}
61
62#[topcoat::router::path_param]
63struct Secret(str);
64
65#[topcoat::router::path_param]
66struct ThreadId(str);
67
68#[topcoat::router::path_param]
69struct CallbackId(str);
70
71#[topcoat::router::path_param]
72struct SessionId(str);
73
74pub async fn run(options: WebOptions) -> std::io::Result<()> {
75    let cwd = canonicalize_cwd(options.cwd)?;
76    let secret = generate_secret();
77    let (listeners, _port) = bind_listeners(options.port).await?;
78    let addresses = listeners
79        .iter()
80        .filter_map(|listener| listener.local_addr().ok())
81        .collect::<Vec<_>>();
82    let mut config_overrides = options.config_overrides;
83    let mcp_callback_port = if let Some(callback_url) = mcp_callback_url(&addresses, &secret) {
84        let callback_listener = TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).await?;
85        let callback_port = callback_listener.local_addr()?.port();
86        drop(callback_listener);
87        config_overrides.push(format!("mcp_oauth_callback_port={callback_port}"));
88        config_overrides.push(format!("mcp_oauth_callback_url={callback_url:?}"));
89        Some(callback_port)
90    } else {
91        None
92    };
93    let bridge = AppServerBridge::start(
94        &options.codex_executable,
95        &cwd,
96        &config_overrides,
97        options.strict_config,
98    )
99    .await?;
100    let shutdown_token = CancellationToken::new();
101    let event_hub = stream::LiveEventHub::start(Arc::clone(&bridge)).await;
102    let state = Arc::new(WebState {
103        secret: secret.clone(),
104        instance_id: generate_secret()[..16].to_string(),
105        cwd: cwd.to_string_lossy().into_owned(),
106        bridge: Arc::clone(&bridge),
107        events: event_hub,
108        shutdown: shutdown_token.clone(),
109        mcp_callback_port,
110    });
111    let router = Router::builder().discover().app_context(state).build();
112    let service = topcoat::router::RouterService::new(router);
113    let mut servers = JoinSet::new();
114    for listener in listeners {
115        let service = service.clone();
116        let shutdown_token = shutdown_token.clone();
117        servers.spawn(async move {
118            topcoat::serve_until(listener, service, async move {
119                shutdown_token.cancelled().await;
120            })
121            .await
122        });
123    }
124
125    let urls = addresses
126        .iter()
127        .map(|address| format!("http://{}/s/{secret}", authority(*address)))
128        .collect::<Vec<_>>();
129    println!("\nCodex Web\n");
130    for url in &urls {
131        println!("  {url}");
132    }
133    println!("\nAnyone with a link can control this Codex process.\n");
134
135    if options.open_browser
136        && let Some(url) = urls.iter().find(|url| url.starts_with("http://127.0.0.1:"))
137        && let Err(error) = webbrowser::open(url)
138    {
139        tracing::warn!(%error, "failed to open Codex Web in a browser");
140    }
141
142    tokio::select! {
143        signal = tokio::signal::ctrl_c() => signal?,
144        result = servers.join_next() => {
145            match result {
146                Some(Ok(Ok(()))) | None => {}
147                Some(Ok(Err(error))) => return Err(error),
148                Some(Err(error)) => return Err(Error::other(error)),
149            }
150        }
151    }
152    shutdown_token.cancel();
153    // Event streams are intentionally long-lived, so a graceful listener drain can wait forever
154    // for browser connections after Ctrl-C. Stop accepting work, then cancel those connection
155    // tasks so the CLI returns promptly.
156    servers.abort_all();
157    while let Some(result) = servers.join_next().await {
158        match result {
159            Ok(Ok(())) => {}
160            Ok(Err(error)) => tracing::warn!(%error, "Codex Web listener stopped"),
161            Err(error) if error.is_cancelled() => {}
162            Err(error) => tracing::warn!(%error, "Codex Web listener task failed"),
163        }
164    }
165    bridge.shutdown().await;
166    Ok(())
167}
168
169#[page("/s/{secret}")]
170async fn home(cx: &Cx) -> Result {
171    let state = authorized_state(cx)?;
172    render_page(cx, state, None).await
173}
174
175#[page("/s/{secret}/thread/{thread_id}")]
176async fn thread_page(cx: &Cx) -> Result {
177    let state = authorized_state(cx)?;
178    render_page(cx, state, Some(path_param::<ThreadId>(cx))).await
179}
180
181#[route(GET "/s/{secret}/assets/app.js")]
182async fn app_js(cx: &Cx) -> Result<Response> {
183    authorized_state(cx)?;
184    static_asset(cx, "text/javascript; charset=utf-8", APP_JS)
185}
186
187#[route(GET "/s/{secret}/assets/app.css")]
188async fn app_css(cx: &Cx) -> Result<Response> {
189    authorized_state(cx)?;
190    static_asset(cx, "text/css; charset=utf-8", APP_CSS)
191}
192
193#[route(GET "/s/{secret}/events")]
194async fn event_stream(cx: &Cx) -> Result<Response> {
195    let state = authorized_state(cx)?;
196    let last_event_id = headers(cx)
197        .get("last-event-id")
198        .and_then(|value| value.to_str().ok())
199        .and_then(|value| value.parse().ok())
200        .or_else(|| {
201            query_value(uri(cx).query().unwrap_or_default(), "after")
202                .and_then(|value| value.parse().ok())
203        });
204    Ok(stream::response(Arc::clone(&state.events), last_event_id).await)
205}
206
207#[route(GET "/s/{secret}/oauth/callback/{callback_id}")]
208async fn mcp_oauth_callback(cx: &Cx) -> Result<Response> {
209    let state = authorized_state(cx)?;
210    let callback_port = state
211        .mcp_callback_port
212        .ok_or_else(|| Error::other("remote MCP OAuth callback is not configured"))?;
213    let callback_id = path_param::<CallbackId>(cx);
214    let query = uri(cx).query().unwrap_or_default();
215    let callback_path = format!("/s/{}/oauth/callback/{callback_id}?{query}", state.secret);
216    let mut stream = TcpStream::connect((Ipv4Addr::LOCALHOST, callback_port))
217        .await
218        .map_err(Error::other)?;
219    stream
220        .write_all(
221            format!(
222                "GET {callback_path} HTTP/1.1\r\nHost: 127.0.0.1:{callback_port}\r\nConnection: close\r\n\r\n"
223            )
224            .as_bytes(),
225        )
226        .await
227        .map_err(Error::other)?;
228    let mut encoded = Vec::new();
229    stream
230        .take(64 * 1024)
231        .read_to_end(&mut encoded)
232        .await
233        .map_err(Error::other)?;
234    let forwarded = String::from_utf8_lossy(&encoded);
235    let body = forwarded
236        .split_once("\r\n\r\n")
237        .map(|(_, body)| body)
238        .unwrap_or("MCP OAuth callback returned an invalid response")
239        .to_string();
240    let mut response = body.into_response(cx)?;
241    response.headers_mut().insert(
242        CONTENT_TYPE,
243        HeaderValue::from_static("text/html; charset=utf-8"),
244    );
245    response
246        .headers_mut()
247        .insert(CACHE_CONTROL, HeaderValue::from_static("no-store"));
248    Ok(response)
249}
250
251#[route(GET "/s/{secret}/api/session/{session_id}")]
252async fn session_fragment(cx: &Cx) -> Result<Response> {
253    let state = authorized_state(cx)?;
254    let session_id = path_param::<SessionId>(cx);
255    let base_seq = state.events.watermark();
256    let active_request = async {
257        if session_id == "new" {
258            None
259        } else {
260            let paginated = state
261                .bridge
262                .request(
263                    "thread/resume",
264                    json!({
265                        "threadId": session_id,
266                        "excludeTurns": true,
267                        "initialTurnsPage": {
268                            "limit": 30,
269                            "sortDirection": "desc",
270                            "itemsView": "full"
271                        }
272                    }),
273                )
274                .await;
275            match paginated {
276                Ok(response) => Some(response),
277                Err(_) => state
278                    .bridge
279                    .request("thread/resume", json!({ "threadId": session_id }))
280                    .await
281                    .ok(),
282            }
283        }
284    };
285    let (active, approvals) =
286        tokio::join!(active_request, state.bridge.outstanding_server_requests(),);
287    let active_thread = active.as_ref().and_then(|response| response.get("thread"));
288    let hydrated_thread = active.as_ref().and_then(thread_with_initial_turns);
289    let display_thread = hydrated_thread.as_ref().or(active_thread);
290    let fragment = transcript_fragment(cx, display_thread, &approvals).await?;
291    let active_turn_id = display_thread
292        .and_then(|active_thread| active_thread.get("turns"))
293        .and_then(Value::as_array)
294        .and_then(|turns| {
295            turns
296                .iter()
297                .rev()
298                .find(|turn| turn.get("status").and_then(Value::as_str) == Some("inProgress"))
299        })
300        .and_then(|turn| turn.get("id"))
301        .and_then(Value::as_str)
302        .unwrap_or_default();
303    let payload = json!({
304        "html": fragment.render(cx),
305        "baseSeq": base_seq,
306        "instanceId": state.instance_id,
307        "nextCursor": active.as_ref().and_then(|response| response.pointer("/initialTurnsPage/nextCursor")),
308        "threadId": display_thread
309            .and_then(|active_thread| active_thread.get("id"))
310            .and_then(Value::as_str)
311            .unwrap_or_default(),
312        "activeTurnId": active_turn_id,
313        "title": display_thread.map(thread_title).unwrap_or("New task"),
314        "cwd": display_thread
315            .and_then(|active_thread| active_thread.get("cwd"))
316            .and_then(Value::as_str)
317            .unwrap_or_default(),
318        "model": active
319            .as_ref()
320            .and_then(|response| response.get("model"))
321            .and_then(Value::as_str),
322        "reasoningEffort": active
323            .as_ref()
324            .and_then(|response| response.get("reasoningEffort"))
325            .and_then(Value::as_str),
326        "permissionMode": permission_mode(active.as_ref()),
327    });
328    let mut response = serde_json::to_vec(&payload)?.into_response(cx)?;
329    response
330        .headers_mut()
331        .insert(CONTENT_TYPE, HeaderValue::from_static("application/json"));
332    response
333        .headers_mut()
334        .insert(CACHE_CONTROL, HeaderValue::from_static("no-store"));
335    Ok(response)
336}
337
338#[route(GET "/s/{secret}/api/session/{session_id}/turns")]
339async fn earlier_turns(cx: &Cx) -> Result<Response> {
340    let state = authorized_state(cx)?;
341    let session_id = path_param::<SessionId>(cx);
342    let cursor = query_value(uri(cx).query().unwrap_or_default(), "cursor");
343    let page = state
344        .bridge
345        .request(
346            "thread/turns/list",
347            json!({
348                "threadId": session_id,
349                "cursor": cursor,
350                "limit": 30,
351                "sortDirection": "desc",
352                "itemsView": "full"
353            }),
354        )
355        .await
356        .map_err(Error::other)?;
357    let thread_value = json!({ "turns": chronological_turns(page.get("data")) });
358    let fragment = transcript_fragment(cx, Some(&thread_value), &[]).await?;
359    json_response(
360        cx,
361        json!({
362            "html": fragment.render(cx),
363            "nextCursor": page.get("nextCursor").cloned().unwrap_or(Value::Null)
364        }),
365    )
366}
367
368#[route(POST "/s/{secret}/api/shutdown")]
369async fn shutdown_process(cx: &Cx) -> Result<Response> {
370    let state = authorized_state(cx)?;
371    state.shutdown.cancel();
372    json_response(cx, json!({ "stopping": true }))
373}
374
375#[route(POST "/s/{secret}/api/rpc")]
376async fn rpc(cx: &Cx, body: Body) -> Result<Response> {
377    let state = authorized_state(cx)?;
378    let bytes = to_bytes(body, MAX_RPC_BODY_BYTES)
379        .await
380        .map_err(Error::other)?;
381    let message: Value = serde_json::from_slice(&bytes)?;
382    let envelope = if let Some(method) = message.get("method").and_then(Value::as_str) {
383        state
384            .bridge
385            .request_envelope(
386                method,
387                message.get("params").cloned().unwrap_or_else(|| json!({})),
388            )
389            .await
390            .map_err(Error::other)?
391    } else {
392        state.bridge.respond(message).await.map_err(Error::other)?;
393        json!({ "result": {} })
394    };
395    let mut response = serde_json::to_vec(&envelope)?.into_response(cx)?;
396    response
397        .headers_mut()
398        .insert(CONTENT_TYPE, HeaderValue::from_static("application/json"));
399    response
400        .headers_mut()
401        .insert(CACHE_CONTROL, HeaderValue::from_static("no-store"));
402    Ok(response)
403}
404
405async fn render_page(cx: &Cx, state: &WebState, thread_id: Option<&str>) -> Result {
406    let base_seq = state.events.watermark();
407    let threads_request = state.bridge.request(
408        "thread/list",
409        json!({ "limit": 100, "sortKey": "recency_at", "sortDirection": "desc" }),
410    );
411    let models_request = state.bridge.request("model/list", json!({ "limit": 100 }));
412    let collaboration_modes_request = state.bridge.request("collaborationMode/list", json!({}));
413    let mcp_servers_request = state.bridge.request(
414        "mcpServerStatus/list",
415        json!({ "limit": 100, "detail": "toolsAndAuthOnly" }),
416    );
417    let active_request = async {
418        match thread_id {
419            Some(thread_id) => state
420                .bridge
421                .request(
422                    "thread/resume",
423                    json!({
424                        "threadId": thread_id,
425                        "excludeTurns": true,
426                        "initialTurnsPage": {
427                            "limit": 30,
428                            "sortDirection": "desc",
429                            "itemsView": "full"
430                        }
431                    }),
432                )
433                .await
434                .ok(),
435            None => None,
436        }
437    };
438    let approvals_request = state.bridge.outstanding_server_requests();
439    let (threads, models, collaboration_modes, mcp_servers, active, approvals) = tokio::join!(
440        threads_request,
441        models_request,
442        collaboration_modes_request,
443        mcp_servers_request,
444        active_request,
445        approvals_request,
446    );
447    let threads = threads.unwrap_or_else(|_| json!({ "data": [] }));
448    let models = models.unwrap_or_else(|_| json!({ "data": [] }));
449    let collaboration_modes = collaboration_modes.unwrap_or_else(|_| json!({ "data": [] }));
450    let mcp_servers = mcp_servers.unwrap_or_else(|_| json!({ "data": [] }));
451    let hydrated_thread = active.as_ref().and_then(thread_with_initial_turns);
452    let active_thread = hydrated_thread
453        .as_ref()
454        .or_else(|| active.as_ref().and_then(|response| response.get("thread")));
455    document(
456        cx,
457        DocumentData {
458            secret: &state.secret,
459            instance_id: &state.instance_id,
460            base_seq,
461            threads: threads
462                .get("data")
463                .and_then(Value::as_array)
464                .map(Vec::as_slice)
465                .unwrap_or_default(),
466            active_thread,
467            active_model: active
468                .as_ref()
469                .and_then(|response| response.get("model"))
470                .and_then(Value::as_str)
471                .unwrap_or_default(),
472            active_effort: active
473                .as_ref()
474                .and_then(|response| response.get("reasoningEffort"))
475                .and_then(Value::as_str)
476                .unwrap_or_default(),
477            active_permission_mode: permission_mode(active.as_ref()),
478            models: models
479                .get("data")
480                .and_then(Value::as_array)
481                .map(Vec::as_slice)
482                .unwrap_or_default(),
483            collaboration_modes: collaboration_modes
484                .get("data")
485                .and_then(Value::as_array)
486                .map(Vec::as_slice)
487                .unwrap_or_default(),
488            approvals: &approvals,
489            workspace_cwd: &state.cwd,
490            mcp_servers: mcp_servers
491                .get("data")
492                .and_then(Value::as_array)
493                .map(Vec::as_slice)
494                .unwrap_or_default(),
495            initial_next_cursor: active
496                .as_ref()
497                .and_then(|response| response.pointer("/initialTurnsPage/nextCursor"))
498                .and_then(Value::as_str),
499        },
500    )
501    .await
502}
503
504fn permission_mode(response: Option<&Value>) -> &'static str {
505    let profile_id = response
506        .and_then(|response| response.pointer("/activePermissionProfile/id"))
507        .and_then(Value::as_str);
508    match profile_id {
509        Some(":danger-full-access") => "full-access",
510        Some(":read-only") => "read-only",
511        Some(":workspace") => "workspace",
512        _ => {
513            let approval = response
514                .and_then(|response| response.get("approvalPolicy"))
515                .and_then(Value::as_str);
516            let sandbox = response
517                .and_then(|response| response.pointer("/sandbox/type"))
518                .and_then(Value::as_str);
519            match (approval, sandbox) {
520                (Some("never"), Some("danger-full-access")) => "full-access",
521                (_, Some("read-only")) => "read-only",
522                _ => "workspace",
523            }
524        }
525    }
526}
527
528fn authorized_state(cx: &Cx) -> Result<&WebState> {
529    let state: &Arc<WebState> = app_context(cx);
530    if path_param::<Secret>(cx) != state.secret {
531        return Err(not_found().into());
532    }
533    Ok(state)
534}
535
536fn static_asset(cx: &Cx, content_type: &'static str, body: &'static str) -> Result<Response> {
537    let mut response = body.into_response(cx)?;
538    response
539        .headers_mut()
540        .insert(CONTENT_TYPE, HeaderValue::from_static(content_type));
541    response
542        .headers_mut()
543        .insert(CACHE_CONTROL, HeaderValue::from_static("no-store"));
544    Ok(response)
545}
546
547fn json_response(cx: &Cx, payload: Value) -> Result<Response> {
548    let mut response = serde_json::to_vec(&payload)?.into_response(cx)?;
549    response
550        .headers_mut()
551        .insert(CONTENT_TYPE, HeaderValue::from_static("application/json"));
552    response
553        .headers_mut()
554        .insert(CACHE_CONTROL, HeaderValue::from_static("no-store"));
555    Ok(response)
556}
557
558#[cfg(test)]
559#[path = "server_tests.rs"]
560mod tests;
codex_web/server.rs

codex_web/
server.rs
