zagens_runtime/runtime_serve/
http.rs

1//! HTTP sidecar bootstrap — DI assembly and axum serve loop (D16 E1-d).
2
3use std::net::SocketAddr;
4use std::path::PathBuf;
5use std::sync::Arc;
6use std::time::{SystemTime, UNIX_EPOCH};
7
8use anyhow::{Context, Result, anyhow};
9use sha2::{Digest, Sha256};
10use tokio::io::AsyncBufReadExt;
11use tokio::io::BufReader;
12use tokio::net::TcpListener;
13use tokio::sync::Mutex;
14use tokio_util::sync::CancellationToken;
15
16use crate::automation_manager::{AutomationManager, AutomationSchedulerConfig, spawn_scheduler};
17use crate::config::Config;
18use crate::runtime_api::{ResumeTaskTracker, RuntimeApiState, build_router};
19use crate::runtime_threads::{RuntimeThreadManager, RuntimeThreadManagerConfig};
20use crate::session_manager::{SessionManager, default_sessions_dir};
21use crate::task_manager::{TaskManager, TaskManagerConfig};
22
23#[derive(Debug, Clone)]
24pub struct RuntimeApiOptions {
25    pub host: String,
26    pub port: u16,
27    pub workers: usize,
28    /// Additional CORS origins to allow on top of the built-in defaults
29    /// (`http://localhost:{3000,1420}`, `http://127.0.0.1:{3000,1420}`,
30    /// `tauri://localhost`). Populated by `--cors-origin` (repeatable),
31    /// `DEEPSEEK_CORS_ORIGINS` (comma-separated), and `[runtime_api]
32    /// cors_origins` in `config.toml`. Whalescale#255 / #561.
33    pub cors_origins: Vec<String>,
34    /// Optional bearer token required for `/v1/*` routes. If omitted here,
35    /// `run_http_server` also checks `DEEPSEEK_RUNTIME_TOKEN`.
36    pub auth_token: Option<String>,
37}
38
39impl Default for RuntimeApiOptions {
40    fn default() -> Self {
41        Self {
42            host: "127.0.0.1".to_string(),
43            port: 7878,
44            workers: 8,
45            cors_origins: Vec::new(),
46            auth_token: None,
47        }
48    }
49}
50
51/// Start the runtime API server.
52///
53/// `options.port == 0` is now accepted and means "let the OS pick an ephemeral
54/// port". The actually bound port is reported back to the supervisor via the
55/// `DS_PICK_READY` line (`port: <bound>`) and through the `local_addr().port()`
56/// log line below; Zagens desktop consumes it via `tokio::sync::watch::<u16>`
57/// (see `crates/desktop/src/sidecar.rs` D2 work). The guard that previously
58/// rejected port 0 was removed in this commit (D2 follow-up).
59pub async fn run_http_server(
60    config: Config,
61    workspace: PathBuf,
62    options: RuntimeApiOptions,
63) -> Result<()> {
64    let t0 = std::time::Instant::now();
65    eprintln!("[deepseek-runtime] starting HTTP API (task manager, threads, scheduler)…");
66
67    let task_cfg = TaskManagerConfig::from_runtime(
68        &config,
69        workspace.clone(),
70        config.default_text_model.clone(),
71        Some(options.workers),
72    );
73    let manager_cfg = RuntimeThreadManagerConfig::from_task_data_dir(task_cfg.data_dir.clone());
74    let sb_config = config.clone();
75    let sb_workspace = workspace.clone();
76    let runtime_threads = Arc::new(
77        tokio::task::spawn_blocking(move || {
78            RuntimeThreadManager::open(sb_config, sb_workspace, manager_cfg)
79        })
80        .await
81        .map_err(|e| anyhow!("RuntimeThreadManager::open panicked: {e}"))??,
82    );
83    eprintln!(
84        "[deepseek-runtime] RuntimeThreadManager::open ok (+{:?})",
85        t0.elapsed()
86    );
87    let task_manager =
88        TaskManager::start_with_runtime_manager(task_cfg, config.clone(), runtime_threads.clone())
89            .await?;
90    eprintln!(
91        "[deepseek-runtime] TaskManager::start ok (+{:?})",
92        t0.elapsed()
93    );
94    let automations = Arc::new(Mutex::new(AutomationManager::default_location()?));
95    runtime_threads.attach_automation_manager(automations.clone());
96    let scheduler_cancel = CancellationToken::new();
97    let scheduler_handle = spawn_scheduler(
98        automations.clone(),
99        task_manager.clone(),
100        scheduler_cancel.clone(),
101        AutomationSchedulerConfig::default(),
102    );
103
104    let sessions_dir = default_sessions_dir()
105        .unwrap_or_else(|_| zagens_config::user_data_path_or_relative("sessions"));
106    let runtime_token = options
107        .auth_token
108        .clone()
109        .or_else(|| std::env::var("DEEPSEEK_RUNTIME_TOKEN").ok())
110        .filter(|token| !token.trim().is_empty());
111    let auth_enabled = runtime_token.is_some();
112
113    let process_started_at_ms = SystemTime::now()
114        .duration_since(UNIX_EPOCH)
115        .unwrap_or_default()
116        .as_millis();
117    let token_fingerprint = {
118        let mut hasher = Sha256::new();
119        hasher.update(runtime_token.as_deref().unwrap_or(""));
120        let hash = hasher.finalize();
121        let fp: String = hash[..16].iter().map(|b| format!("{b:02x}")).collect();
122        Arc::new(fp)
123    };
124    let shared_session_manager = Arc::new(
125        SessionManager::new(sessions_dir.clone()).context("Failed to create SessionManager")?,
126    );
127
128    let mut shared_mcp_pool = crate::mcp::McpPool::from_config_path(&config.mcp_config_path())
129        .context("Failed to load MCP config for shared pool")?;
130    if let Some(network_toml) = config.network.clone() {
131        let decider = crate::network_policy::NetworkPolicyDecider::with_default_audit(
132            network_toml.into_runtime(),
133        );
134        shared_mcp_pool = shared_mcp_pool.with_network_policy(decider);
135    }
136    let shared_mcp_pool = Arc::new(tokio::sync::Mutex::new(shared_mcp_pool));
137    crate::mcp_shared::install_shared_mcp_pool(Arc::clone(&shared_mcp_pool));
138
139    let token_fp = token_fingerprint.as_ref().clone();
140    let state = RuntimeApiState::new(
141        config.clone(),
142        workspace,
143        task_manager,
144        runtime_threads,
145        options.cors_origins.clone(),
146        config.mcp_config_path(),
147        automations,
148        runtime_token,
149        process_started_at_ms,
150        token_fingerprint,
151        shared_session_manager,
152        ResumeTaskTracker::new(),
153        shared_mcp_pool,
154    );
155    let app = build_router(state);
156
157    let addr: SocketAddr = format!("{}:{}", options.host, options.port)
158        .parse()
159        .with_context(|| format!("Invalid bind address '{}:{}'", options.host, options.port))?;
160    let listener = TcpListener::bind(addr)
161        .await
162        .with_context(|| format!("Failed to bind {addr}"))?;
163    // Report the actual bound port so callers that pass `--port 0` (or hit ephemeral fallback)
164    // can discover the real listener; `options.port` may differ from `local_addr().port()`.
165    let bound_addr = listener
166        .local_addr()
167        .with_context(|| "Failed to read bound local_addr from TcpListener")?;
168    let bound_port = bound_addr.port();
169
170    eprintln!(
171        "[deepseek-runtime] bound {bound_addr}, serving (+{:?}) — output also on stderr (see sidecar.log if launched from Zagens)",
172        t0.elapsed()
173    );
174    eprintln!("Runtime API listening on http://{bound_addr}");
175    eprintln!("Security: this server is local-first. Do not expose it to untrusted networks.");
176    if auth_enabled {
177        eprintln!("Runtime API auth: bearer token required for /v1/* routes.");
178    }
179
180    // Signal READY to the supervisor via stdout (line protocol).
181    // Zagens's supervisor waits for this line before considering the sidecar healthy.
182    // `port` MUST be the actually bound port (not the requested one) so the desktop
183    // shell can discover ephemeral ports from `--port 0`.
184    let ready_line = serde_json::json!({
185        "port": bound_port,
186        "pid": std::process::id(),
187        "token_fp": token_fp,
188        "version": env!("CARGO_PKG_VERSION"),
189    });
190    println!("DS_PICK_READY {ready_line}");
191    let _ = std::io::Write::flush(&mut std::io::stdout());
192
193    let started_at = std::time::Instant::now();
194    tokio::spawn(async move {
195        let stdin = BufReader::new(tokio::io::stdin());
196        let mut lines = stdin.lines();
197        while let Ok(Some(line)) = lines.next_line().await {
198            let op: serde_json::Value = match serde_json::from_str(&line) {
199                Ok(v) => v,
200                Err(_) => continue,
201            };
202            match op.get("op").and_then(|v| v.as_str()) {
203                Some("ping") => {
204                    let seq = op.get("seq").and_then(|v| v.as_u64()).unwrap_or(0);
205                    let pong = serde_json::json!({
206                        "op": "pong",
207                        "seq": seq,
208                        "pid": std::process::id(),
209                        "uptime_ms": started_at.elapsed().as_millis(),
210                    });
211                    println!("DS_PICK_PONG {pong}");
212                    let _ = std::io::Write::flush(&mut std::io::stdout());
213                }
214                Some("drain") => {
215                    let drain_resp = serde_json::json!({
216                        "op": "drain",
217                        "state": "draining",
218                    });
219                    println!("DS_PICK_DRAIN {drain_resp}");
220                    let _ = std::io::Write::flush(&mut std::io::stdout());
221                    break;
222                }
223                _ => {}
224            }
225        }
226    });
227
228    eprintln!("[deepseek-runtime] axum::serve started, listening on {bound_addr}");
229    let serve_result = axum::serve(listener, app)
230        .await
231        .map_err(|e| anyhow!("Runtime API server error: {e}"));
232    eprintln!(
233        "[deepseek-runtime] axum::serve returned: {:?}",
234        serve_result
235            .as_ref()
236            .map(|_| "ok")
237            .map_err(|e| format!("{e:#}"))
238    );
239    scheduler_cancel.cancel();
240    scheduler_handle.abort();
241    serve_result
242}
zagens_runtime/runtime_serve/http.rs

zagens_runtime/runtime_serve/
http.rs
