ocserve_http/
pty.rs

1//! `/pty/*` — terminal session routes (web-UI integrated terminal, VS Code,
2//! SDK clients). Freeze parity: shapes captured live against opencode
3//! 1.18.31 (2026-10-08, bench/pty/PTY-PLAN.md).
4//!
5//! Routes:
6//!   GET    /pty/shells                → Shell[] (no process spawned)
7//!   GET    /pty                       → Info[] (running only — legacy surface)
8//!   POST   /pty                       → Info (spawns; login shells get -l)
9//!   GET    /pty/{id}                  → Info | 404 tag | 400 Params envelope
10//!   PUT    /pty/{id}                  → Info (title/size; PATCH not a route)
11//!   DELETE /pty/{id}                  → true
12//!   POST   /pty/{id}/connect-token    → {ticket,expires_in} (header + origin gated)
13//!   GET    /pty/{id}/connect          → websocket: replay → meta([0x00]JSON)
14//!                                        → live; close 4404 on not-found/exited
15
16use axum::extract::ws::{Message, WebSocket, WebSocketUpgrade};
17use axum::extract::{Path, Query, State};
18use axum::http::{HeaderMap, StatusCode, header};
19use axum::response::{IntoResponse, Response};
20use serde_json::{Value, json};
21use std::sync::Arc;
22
23use crate::{ApiError, AppState, HttpError};
24
25/// v1 pty ids: `pty_` + 26 chars (schema/src/pty.ts IDSchema isStartsWith).
26/// Malformed path params get the Effect Query Params envelope (probed).
27fn check_pty_id(id: &str) -> Result<(), ApiError> {
28    let ok = id.starts_with("pty_") && id.len() > 4;
29    if ok {
30        return Ok(());
31    }
32    Err(ApiError {
33        status: StatusCode::BAD_REQUEST,
34        name: "BadRequest",
35        message: format!("Expected a string starting with \"pty\", got \"{id}\"\n  at [\"ptyID\"]"),
36    })
37}
38
39fn not_found(id: &str) -> HttpError {
40    // Tagged envelope (probed): {"_tag":"PtyNotFoundError","ptyID","message"}
41    HttpError::TaggedData {
42        status: StatusCode::NOT_FOUND,
43        tag: "PtyNotFoundError",
44        fields: json!({"ptyID": id, "message": format!("PTY session not found: {id}")}),
45    }
46}
47
48pub async fn get_shells(State(st): State<Arc<AppState>>) -> Response {
49    axum::Json(st.pty.shells()).into_response()
50}
51
52pub async fn list(State(st): State<Arc<AppState>>) -> Response {
53    axum::Json(Value::Array(st.pty.list_running())).into_response()
54}
55
56pub async fn create(State(st): State<Arc<AppState>>, body: Option<axum::Json<Value>>) -> Response {
57    let input = body.map(|b| b.0).unwrap_or_else(|| json!({}));
58    match st.pty.create(&input) {
59        Ok(info) => axum::Json(info).into_response(),
60        Err(e) => (
61            StatusCode::INTERNAL_SERVER_ERROR,
62            axum::Json(json!({"name": "UnknownError", "data": {"message": e}})),
63        )
64            .into_response(),
65    }
66}
67
68pub async fn get(Path(id): Path<String>, State(st): State<Arc<AppState>>) -> Response {
69    if let Err(e) = check_pty_id(&id) {
70        return e.into_response();
71    }
72    match st.pty.get_running(&id) {
73        Ok(info) => axum::Json(info).into_response(),
74        Err(_) => not_found(&id).into_response(),
75    }
76}
77
78pub async fn update(
79    Path(id): Path<String>,
80    State(st): State<Arc<AppState>>,
81    body: Option<axum::Json<Value>>,
82) -> Response {
83    if let Err(e) = check_pty_id(&id) {
84        return e.into_response();
85    }
86    let input = body.map(|b| b.0).unwrap_or_else(|| json!({}));
87    match st.pty.update(&id, &input) {
88        Ok(info) => axum::Json(info).into_response(),
89        Err(_) => not_found(&id).into_response(),
90    }
91}
92
93pub async fn remove(Path(id): Path<String>, State(st): State<Arc<AppState>>) -> Response {
94    if let Err(e) = check_pty_id(&id) {
95        return e.into_response();
96    }
97    match st.pty.remove(&id) {
98        Ok(v) => axum::Json(v).into_response(),
99        Err(_) => not_found(&id).into_response(),
100    }
101}
102
103/// POST /pty/{id}/connect-token — upstream requires the marker header
104/// (`x-opencode-ticket: 1`) AND a permitted Origin, then the session must
105/// exist. Wrong header/origin → 403 tag; unknown id (after the gate) → 404.
106pub async fn connect_token(
107    Path(id): Path<String>,
108    State(st): State<Arc<AppState>>,
109    headers: HeaderMap,
110) -> Response {
111    if let Err(e) = check_pty_id(&id) {
112        return e.into_response();
113    }
114    let marker_ok = headers
115        .get("x-opencode-ticket")
116        .and_then(|v| v.to_str().ok())
117        == Some("1");
118    let origin_ok = match headers.get(header::ORIGIN).and_then(|v| v.to_str().ok()) {
119        None => true, // no Origin: allowed (CLI clients)
120        Some(o) => {
121            crate::ui::is_allowed_cors_origin(o, &st.cors_extra)
122                || headers
123                    .get(header::HOST)
124                    .and_then(|v| v.to_str().ok())
125                    .map(|h| o.ends_with(h))
126                    .unwrap_or(false)
127        }
128    };
129    if !marker_ok || !origin_ok {
130        return (HttpError::TaggedData {
131            status: StatusCode::FORBIDDEN,
132            tag: "PtyForbiddenError",
133            fields: json!({"message": "Invalid PTY connect token request"}),
134        })
135        .into_response();
136    }
137    if st.pty.get_running(&id).is_err() {
138        return not_found(&id).into_response();
139    }
140    axum::Json(st.pty.issue_ticket(&id)).into_response()
141}
142
143#[derive(serde::Deserialize)]
144pub struct ConnectQuery {
145    pub cursor: Option<String>,
146    pub ticket: Option<String>,
147    #[allow(dead_code)]
148    pub directory: Option<String>,
149    #[allow(dead_code)]
150    pub workspace: Option<String>,
151}
152
153/// GET /pty/{id}/connect — websocket. Upstream: if a ticket is presented it
154/// must consume; if not, allowed origins only. Then attach, replay
155/// (64 KiB chunks), meta frame [0x00 + {"cursor":N}], live streaming;
156/// close code 4404 when the session is not-found/exited.
157pub async fn connect(
158    Path(id): Path<String>,
159    Query(q): Query<ConnectQuery>,
160    State(st): State<Arc<AppState>>,
161    headers: HeaderMap,
162    ws: WebSocketUpgrade,
163) -> Response {
164    if let Err(e) = check_pty_id(&id) {
165        return e.into_response();
166    }
167    if let Some(ticket) = q.ticket.as_deref() {
168        let origin_ok = match headers.get(header::ORIGIN).and_then(|v| v.to_str().ok()) {
169            None => true,
170            Some(o) => crate::ui::is_allowed_cors_origin(o, &st.cors_extra),
171        };
172        if !origin_ok || !st.pty.consume_ticket(&id, ticket) {
173            return StatusCode::FORBIDDEN.into_response();
174        }
175    } else {
176        // no ticket: permitted origin required (probed: localhost origin OK)
177        if let Some(o) = headers.get(header::ORIGIN).and_then(|v| v.to_str().ok())
178            && !crate::ui::is_allowed_cors_origin(o, &st.cors_extra)
179        {
180            return StatusCode::FORBIDDEN.into_response();
181        }
182    }
183    let cursor = q.cursor.and_then(|c| c.parse::<i64>().ok());
184    match st.pty.attach(&id, cursor) {
185        Err(_) => {
186            // upgrade then immediately close 4404 (upstream behavior)
187            ws.on_upgrade(move |mut socket| async move {
188                let _ = socket
189                    .send(Message::Close(Some(axum::extract::ws::CloseFrame {
190                        code: 4404,
191                        reason: "session not found".into(),
192                    })))
193                    .await;
194            })
195        }
196        Ok(attachment) => {
197            let st2 = st.clone();
198            let id2 = id.clone();
199            ws.on_upgrade(move |socket| pty_ws_loop(socket, st2, id2, attachment))
200        }
201    }
202}
203
204async fn pty_ws_loop(
205    mut socket: WebSocket,
206    st: Arc<AppState>,
207    id: String,
208    mut attachment: ocserve_pty::Attachment,
209) {
210    // 1) replay in 64 KiB chunks (upstream PtyProtocol.REPLAY_CHUNK)
211    const REPLAY_CHUNK: usize = 64 * 1024;
212    let replay = attachment.replay.clone();
213    let mut rest = replay.as_str();
214    while !rest.is_empty() {
215        let cut = floor_char_boundary(rest, REPLAY_CHUNK);
216        if socket
217            .send(Message::Text(rest[..cut].to_string().into()))
218            .await
219            .is_err()
220        {
221            st.pty.detach(&id, attachment.sub_id);
222            return;
223        }
224        rest = &rest[cut..];
225    }
226    // 2) meta frame: 0x00 then JSON {"cursor":N}
227    let meta = format!("\u{0}{}", json!({"cursor": attachment.cursor}));
228    if socket.send(Message::Text(meta.into())).await.is_err() {
229        st.pty.detach(&id, attachment.sub_id);
230        return;
231    }
232    // 3) live: pump subscription → socket; inbound → pty write; close on end.
233    loop {
234        tokio::select! {
235            msg = attachment.rx.recv() => match msg {
236                Some(ocserve_pty::Msg::Data(chunk)) => {
237                    if socket.send(Message::Text(chunk.into())).await.is_err() { break; }
238                }
239                Some(ocserve_pty::Msg::End { .. }) | None => {
240                    let _ = socket.send(Message::Close(Some(axum::extract::ws::CloseFrame {
241                        code: 1000, reason: "".into(),
242                    }))).await;
243                    break;
244                }
245            },
246            inbound = socket.recv() => match inbound {
247                Some(Ok(Message::Text(t))) => attachment_write(&st, &id, t.as_str()),
248                Some(Ok(Message::Binary(b))) => {
249                    attachment_write(&st, &id, &String::from_utf8_lossy(&b));
250                }
251                Some(Ok(Message::Close(_))) | None => break,
252                Some(Ok(_)) => {}
253                Some(Err(_)) => break,
254            },
255        }
256    }
257    st.pty.detach(&id, attachment.sub_id);
258}
259
260fn attachment_write(st: &Arc<AppState>, id: &str, data: &str) {
261    st.pty.write_session(id, data);
262}
263
264/// Largest char boundary ≤ `want`. When a single char is wider than `want`
265/// (only possible for want < 4), returns 0 rather than a non-boundary index —
266/// the previous `want.min(len)` fallback returned an index that is NOT a char
267/// boundary and would panic any caller slicing at it.
268fn floor_char_boundary(s: &str, want: usize) -> usize {
269    if want >= s.len() {
270        return s.len();
271    }
272    let mut end = want;
273    while end > 0 && !s.is_char_boundary(end) {
274        end -= 1;
275    }
276    end
277}
ocserve_http/pty.rs

ocserve_http/
pty.rs
