hatchery_node/
local_server.rs

1//! Session-local HTTP server pool (tesserax-http static + supervised http children).
2//!
3//! Ports bind loopback only. Instances are keyed by session id (same string the
4//! session browser uses). Lifecycle: start/stop via MCP door or restream control;
5//! session end stops every instance for that session.
6
7use serde_json::{json, Value};
8use std::collections::BTreeMap;
9use std::path::{Path, PathBuf};
10use std::process::Stdio;
11use std::sync::{Arc, Mutex, OnceLock};
12use std::time::Duration;
13use tokio::process::{Child, Command};
14
15/// Per-session port window size (ports are still ephemeral binds on 127.0.0.1:0).
16const MAX_PER_SESSION: usize = 8;
17const STOP_WAIT: Duration = Duration::from_secs(3);
18
19#[derive(Clone, Debug)]
20pub struct InstanceInfo {
21    pub id: String,
22    pub session: String,
23    pub kind: String,
24    pub url: String,
25    pub port: u16,
26    pub root: Option<String>,
27}
28
29enum Runtime {
30    Static {
31        // Keep the running server alive until stop.
32        server: tesserax::RunningServer,
33    },
34    Http {
35        child: Child,
36    },
37}
38
39struct Instance {
40    info: InstanceInfo,
41    runtime: Runtime,
42    log: Arc<Mutex<Vec<String>>>,
43}
44
45struct PoolState {
46    by_session: BTreeMap<String, BTreeMap<String, Instance>>,
47}
48
49pub struct LocalServerPool {
50    inner: Mutex<PoolState>,
51}
52
53impl LocalServerPool {
54    fn new() -> Self {
55        Self {
56            inner: Mutex::new(PoolState {
57                by_session: BTreeMap::new(),
58            }),
59        }
60    }
61
62    pub fn global() -> Arc<LocalServerPool> {
63        static POOL: OnceLock<Arc<LocalServerPool>> = OnceLock::new();
64        POOL.get_or_init(|| Arc::new(LocalServerPool::new())).clone()
65    }
66
67    pub fn kinds() -> Value {
68        json!([
69            {"kind": "static", "description": "Serve a directory with tesserax-http StaticDir (SPA fallback when index.html exists)."},
70            {"kind": "http", "description": "Run a user binary that listens on PORT (loopback). cwd and env are optional."}
71        ])
72    }
73
74    pub async fn start(
75        &self,
76        session: &str,
77        kind: &str,
78        root: Option<&str>,
79        cwd: Option<&str>,
80        bin: Option<&str>,
81        env: Option<&Value>,
82        spa: bool,
83    ) -> Result<InstanceInfo, String> {
84        if session.is_empty() || session.len() > 128 {
85            return Err("invalid session".into());
86        }
87        {
88            let g = self.inner.lock().map_err(|_| "lock")?;
89            let n = g.by_session.get(session).map(|m| m.len()).unwrap_or(0);
90            if n >= MAX_PER_SESSION {
91                return Err(format!("session already has {MAX_PER_SESSION} local servers"));
92            }
93        }
94        let id = format!("ls-{}", &uuid::Uuid::new_v4().to_string()[..8]);
95        let (info, runtime) = match kind {
96            "static" => {
97                let root = root.ok_or("static kind requires root")?;
98                let root_path = PathBuf::from(root);
99                if !root_path.is_absolute() {
100                    return Err("root must be an absolute path".into());
101                }
102                if !root_path.is_dir() {
103                    return Err("root is not a directory".into());
104                }
105                start_static(&id, session, &root_path, spa).await?
106            }
107            "http" => {
108                let bin = bin.or(cwd).ok_or("http kind requires bin or cwd")?;
109                let cwd_path = cwd.map(PathBuf::from);
110                start_http(&id, session, bin, cwd_path.as_deref(), env).await?
111            }
112            _ => return Err(format!("unknown kind: {kind}")),
113        };
114        let log = Arc::new(Mutex::new(Vec::new()));
115        {
116            let mut g = self.inner.lock().map_err(|_| "lock")?;
117            g.by_session
118                .entry(session.to_owned())
119                .or_default()
120                .insert(id.clone(), Instance { info: info.clone(), runtime, log });
121        }
122        Ok(info)
123    }
124
125    pub async fn stop(&self, session: &str, id: &str) -> Result<(), String> {
126        let inst = {
127            let mut g = self.inner.lock().map_err(|_| "lock")?;
128            g.by_session
129                .get_mut(session)
130                .and_then(|m| m.remove(id))
131                .ok_or_else(|| "not found".to_owned())?
132        };
133        stop_runtime(inst.runtime).await;
134        let mut g = self.inner.lock().map_err(|_| "lock")?;
135        if let Some(m) = g.by_session.get(session) {
136            if m.is_empty() {
137                g.by_session.remove(session);
138            }
139        }
140        Ok(())
141    }
142
143    pub fn list(&self, session: &str) -> Value {
144        let g = match self.inner.lock() {
145            Ok(g) => g,
146            Err(_) => return json!({"kinds": Self::kinds(), "instances": []}),
147        };
148        let instances: Vec<Value> = g
149            .by_session
150            .get(session)
151            .map(|m| {
152                m.values()
153                    .map(|i| {
154                        json!({
155                            "id": i.info.id,
156                            "kind": i.info.kind,
157                            "url": i.info.url,
158                            "port": i.info.port,
159                            "root": i.info.root,
160                        })
161                    })
162                    .collect()
163            })
164            .unwrap_or_default();
165        json!({ "kinds": Self::kinds(), "instances": instances, "session": session })
166    }
167
168    pub fn logs(&self, session: &str, id: &str, limit: usize) -> Result<Value, String> {
169        let g = self.inner.lock().map_err(|_| "lock")?;
170        let inst = g
171            .by_session
172            .get(session)
173            .and_then(|m| m.get(id))
174            .ok_or_else(|| "not found".to_owned())?;
175        let lines = inst.log.lock().map_err(|_| "lock")?;
176        let start = lines.len().saturating_sub(limit.max(1).min(200));
177        Ok(json!({ "id": id, "lines": lines[start..].to_vec() }))
178    }
179
180    pub async fn stop_session(&self, session: &str) {
181        let ids: Vec<String> = {
182            let g = match self.inner.lock() {
183                Ok(g) => g,
184                Err(_) => return,
185            };
186            g.by_session
187                .get(session)
188                .map(|m| m.keys().cloned().collect())
189                .unwrap_or_default()
190        };
191        for id in ids {
192            let _ = self.stop(session, &id).await;
193        }
194    }
195}
196
197async fn start_static(
198    id: &str,
199    session: &str,
200    root: &Path,
201    spa: bool,
202) -> Result<(InstanceInfo, Runtime), String> {
203    use tesserax::{ServerBuilder, Transport};
204    use tesserax_http::assets::StaticDir;
205    use tesserax_http::HttpExt;
206
207    let mut dir = StaticDir::new(root);
208    if spa {
209        let index = root.join("index.html");
210        if index.is_file() {
211            dir = dir.with_spa_fallback(index);
212        }
213    }
214    let built = ServerBuilder::new(format!("hatchery-local-{id}"))
215        .transport(Transport::local(0))
216        .with_static_dir(dir)
217        .build()
218        .await
219        .map_err(|e| format!("build static server: {e}"))?;
220    let running = built
221        .start()
222        .await
223        .map_err(|e| format!("start static server: {e}"))?;
224    let addr = running.local_addr();
225    if !addr.ip().is_loopback() {
226        running.shutdown();
227        let _ = running.wait().await;
228        return Err("refused non-loopback bind".into());
229    }
230    let port = addr.port();
231    let url = format!("http://127.0.0.1:{port}/");
232    Ok((
233        InstanceInfo {
234            id: id.to_owned(),
235            session: session.to_owned(),
236            kind: "static".into(),
237            url,
238            port,
239            root: Some(root.display().to_string()),
240        },
241        Runtime::Static { server: running },
242    ))
243}
244
245async fn start_http(
246    id: &str,
247    session: &str,
248    bin: &str,
249    cwd: Option<&Path>,
250    env: Option<&Value>,
251) -> Result<(InstanceInfo, Runtime), String> {
252    // Bind an ephemeral port ourselves and pass it as PORT so the child can listen.
253    let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
254        .await
255        .map_err(|e| format!("port alloc: {e}"))?;
256    let port = listener.local_addr().map_err(|e| e.to_string())?.port();
257    drop(listener);
258
259    let mut cmd = Command::new(bin);
260    if let Some(cwd) = cwd {
261        cmd.current_dir(cwd);
262    }
263    cmd.env("PORT", port.to_string())
264        .env("HOST", "127.0.0.1")
265        .stdin(Stdio::null())
266        .stdout(Stdio::null())
267        .stderr(Stdio::null())
268        .kill_on_drop(true);
269    crate::windows_process::hide_console_tokio(&mut cmd);
270    if let Some(Value::Object(map)) = env {
271        for (k, v) in map {
272            if let Some(s) = v.as_str() {
273                cmd.env(k, s);
274            }
275        }
276    }
277    let child = cmd.spawn().map_err(|e| format!("spawn http server: {e}"))?;
278    // Brief readiness: accept a TCP connect.
279    let deadline = tokio::time::Instant::now() + Duration::from_secs(5);
280    loop {
281        if tokio::net::TcpStream::connect(("127.0.0.1", port))
282            .await
283            .is_ok()
284        {
285            break;
286        }
287        if tokio::time::Instant::now() > deadline {
288            return Err("http server did not accept on PORT in time".into());
289        }
290        tokio::time::sleep(Duration::from_millis(50)).await;
291    }
292    let url = format!("http://127.0.0.1:{port}/");
293    Ok((
294        InstanceInfo {
295            id: id.to_owned(),
296            session: session.to_owned(),
297            kind: "http".into(),
298            url,
299            port,
300            root: cwd.map(|p| p.display().to_string()),
301        },
302        Runtime::Http { child },
303    ))
304}
305
306async fn stop_runtime(runtime: Runtime) {
307    match runtime {
308        Runtime::Static { server } => {
309            server.shutdown();
310            let _ = tokio::time::timeout(STOP_WAIT, server.wait()).await;
311        }
312        Runtime::Http { mut child } => {
313            let _ = child.kill().await;
314            let _ = tokio::time::timeout(STOP_WAIT, child.wait()).await;
315        }
316    }
317}
318
319/// Door / control dispatch. `op` is `local_server.start|stop|list|logs`.
320pub async fn handle_op(op: &str, payload: &Value) -> Result<Value, String> {
321    let pool = LocalServerPool::global();
322    let session = payload
323        .get("session")
324        .and_then(Value::as_str)
325        .ok_or("session required")?;
326    match op {
327        "local_server.list" => Ok(pool.list(session)),
328        "local_server.start" => {
329            let kind = payload
330                .get("kind")
331                .and_then(Value::as_str)
332                .ok_or("kind required")?;
333            let root = payload.get("root").and_then(Value::as_str);
334            let cwd = payload.get("cwd").and_then(Value::as_str);
335            let bin = payload
336                .get("bin")
337                .and_then(Value::as_str)
338                .or_else(|| payload.get("command").and_then(Value::as_str));
339            let spa = payload
340                .get("spa")
341                .and_then(Value::as_bool)
342                .unwrap_or(true);
343            let env = payload.get("env");
344            let info = pool
345                .start(session, kind, root, cwd, bin, env, spa)
346                .await?;
347            let open = payload
348                .get("open")
349                .and_then(Value::as_bool)
350                .unwrap_or(false);
351            if open {
352                #[cfg(feature = "d2b")]
353                {
354                    let browsers = crate::browser_capture::shared_map();
355                    let url = info.url.clone();
356                    let sess = session.to_owned();
357                    // Best-effort: navigate an existing session browser; ignore if none.
358                    if let Ok(bytes) = hatchery_core::encode_browser_navigate(&url) {
359                        let _ = browsers.input(&sess, &bytes).await;
360                    }
361                }
362            }
363            Ok(json!({
364                "id": info.id,
365                "url": info.url,
366                "port": info.port,
367                "kind": info.kind,
368            }))
369        }
370        "local_server.stop" => {
371            let id = payload
372                .get("id")
373                .and_then(Value::as_str)
374                .ok_or("id required")?;
375            pool.stop(session, id).await?;
376            Ok(json!({"ok": true, "id": id}))
377        }
378        "local_server.logs" => {
379            let id = payload
380                .get("id")
381                .and_then(Value::as_str)
382                .ok_or("id required")?;
383            let limit = payload
384                .get("limit")
385                .and_then(Value::as_u64)
386                .unwrap_or(50) as usize;
387            pool.logs(session, id, limit)
388        }
389        _ => Err("unknown local_server op".into()),
390    }
391}
392
393#[cfg(test)]
394mod tests {
395    use super::*;
396    use std::io::Write;
397
398    #[tokio::test]
399    async fn static_start_list_stop_on_loopback() {
400        let dir = tempfile::tempdir().unwrap();
401        let mut f = std::fs::File::create(dir.path().join("index.html")).unwrap();
402        write!(f, "<h1>hi</h1>").unwrap();
403        let pool = LocalServerPool::new();
404        let info = pool
405            .start("pty-test-1", "static", Some(dir.path().to_str().unwrap()), None, None, None, true)
406            .await
407            .unwrap();
408        assert!(info.url.starts_with("http://127.0.0.1:"));
409        let list = pool.list("pty-test-1");
410        assert_eq!(list["instances"].as_array().unwrap().len(), 1);
411        let body = reqwest::get(&info.url).await.unwrap().text().await.unwrap();
412        assert!(body.contains("hi"));
413        pool.stop("pty-test-1", &info.id).await.unwrap();
414        assert!(pool.list("pty-test-1")["instances"]
415            .as_array()
416            .unwrap()
417            .is_empty());
418    }
419}
hatchery_node/local_server.rs

hatchery_node/
local_server.rs
