fix: add bootstrap step

koharu-app/src/app.rs

@@ -36,6 +36,23 @@ use crate::session::ProjectSession;
 /// cover a minute of pipeline chatter at ~4 events/sec.
 const EVENT_BUS_CAPACITY: usize = 256;
 
+#[derive(Clone)]
+pub struct AppSharedState {
+    pub jobs: Arc<DashMap<String, JobSummary>>,
+    pub downloads: Arc<DashMap<String, DownloadProgress>>,
+    pub bus: Arc<EventBus>,
+}
+
+impl Default for AppSharedState {
+    fn default() -> Self {
+        Self {
+            jobs: Arc::new(DashMap::new()),
+            downloads: Arc::new(DashMap::new()),
+            bus: EventBus::new(EVENT_BUS_CAPACITY),
+        }
+    }
+}
+
 /// Top-level app.
 pub struct App {
     pub config: Arc<ArcSwap<AppConfig>>,
@@ -59,6 +76,19 @@ impl App {
         runtime: Arc<RuntimeManager>,
         cpu: bool,
         version: &'static str,
+    ) -> Result<Self> {
+        Self::new_with_shared_state(config, runtime, cpu, AppSharedState::default(), version)
+    }
+
+    /// Construct with caller-provided shared registries/event bus. This is
+    /// used by the HTTP bootstrap flow so the server can expose downloads and
+    /// SSE state before the full `App` is ready.
+    pub fn new_with_shared_state(
+        config: AppConfig,
+        runtime: Arc<RuntimeManager>,
+        cpu: bool,
+        shared: AppSharedState,
+        version: &'static str,
     ) -> Result<Self> {
         let backend = shared_llama_backend(&runtime)?;
         let llm = Arc::new(llm::Model::new((*runtime).clone(), cpu, backend));
@@ -68,9 +98,9 @@ impl App {
             runtime,
             registry: Arc::new(Registry::new()),
             session: Arc::new(ArcSwapOption::empty()),
-            jobs: Arc::new(DashMap::new()),
-            downloads: Arc::new(DashMap::new()),
-            bus: EventBus::new(EVENT_BUS_CAPACITY),
+            jobs: shared.jobs,
+            downloads: shared.downloads,
+            bus: shared.bus,
             llm,
             renderer,
             autosave: Mutex::new(None),

koharu-app/src/lib.rs

@@ -18,7 +18,7 @@ pub mod renderer;
 pub mod session;
 pub mod utils;
 
-pub use app::App;
+pub use app::{App, AppSharedState};
 pub use blobs::BlobStore;
 pub use config::AppConfig;
 pub use pipeline::{

koharu-rpc/src/api.rs

@@ -5,6 +5,11 @@
 
 use axum::Router;
 use axum::extract::DefaultBodyLimit;
+use axum::extract::{Request, State};
+use axum::http::StatusCode;
+use axum::middleware;
+use axum::middleware::Next;
+use axum::response::{IntoResponse, Response};
 use utoipa_axum::router::OpenApiRouter;
 
 use crate::AppState;
@@ -16,8 +21,14 @@ const MAX_BODY_SIZE: usize = 1024 * 1024 * 1024;
 /// State threaded through every `State<ApiState>` extractor.
 pub type ApiState = AppState;
 
-/// Build the router + OpenAPI doc. Called by the bin and by `router()`.
-pub fn api() -> (Router<ApiState>, utoipa::openapi::OpenApi) {
+fn bootstrap_api() -> OpenApiRouter<ApiState> {
+    OpenApiRouter::default()
+        .merge(routes::downloads::router())
+        .merge(routes::operations::router())
+        .merge(events::router())
+}
+
+fn app_api() -> OpenApiRouter<ApiState> {
     OpenApiRouter::default()
         .merge(routes::history::router())
         .merge(routes::pages::router())
@@ -27,18 +38,38 @@ pub fn api() -> (Router<ApiState>, utoipa::openapi::OpenApi) {
         .merge(routes::fonts::router())
         .merge(routes::llm::router())
         .merge(routes::pipelines::router())
-        .merge(routes::downloads::router())
-        .merge(routes::operations::router())
-        .merge(events::router())
         .merge(binary::router())
-        .split_for_parts()
+}
+
+/// Build the router + OpenAPI doc. Called by the bin and by `router()`.
+pub fn api() -> (Router<ApiState>, utoipa::openapi::OpenApi) {
+    bootstrap_api().merge(app_api()).split_for_parts()
+}
+
+async fn require_ready(State(app): State<ApiState>, request: Request, next: Next) -> Response {
+    if !app.is_ready() {
+        return (
+            StatusCode::SERVICE_UNAVAILABLE,
+            crate::ApiError::new(
+                StatusCode::SERVICE_UNAVAILABLE,
+                "app is still bootstrapping",
+            ),
+        )
+            .into_response();
+    }
+    next.run(request).await
 }
 
 /// Ready-to-serve router. `app` becomes shared state. All routes live under
 /// `/api/v1` so the UI can reach them through a single proxy prefix.
 pub fn router(app: ApiState) -> Router {
-    let (inner, _) = api();
+    let (bootstrap, _) = bootstrap_api().split_for_parts();
+    let (guarded, _) = app_api().split_for_parts();
+    let bootstrap = bootstrap.with_state(app.clone());
+    let guarded = guarded
+        .with_state(app.clone())
+        .layer(middleware::from_fn_with_state(app, require_ready));
     Router::new()
-        .nest("/api/v1", inner.with_state(app))
+        .nest("/api/v1", bootstrap.merge(guarded))
         .layer(DefaultBodyLimit::max(MAX_BODY_SIZE))
 }

koharu-rpc/src/bootstrap.rs

@@ -0,0 +1,78 @@
+use std::ops::Deref;
+use std::sync::{Arc, OnceLock};
+
+use dashmap::DashMap;
+use koharu_app::bus::EventBus;
+use koharu_app::{App, AppSharedState};
+use koharu_core::{AppEvent, DownloadProgress, JobSummary};
+use koharu_runtime::RuntimeManager;
+
+pub struct BootstrapManager {
+    app: OnceLock<Arc<App>>,
+    runtime: Arc<RuntimeManager>,
+    shared: AppSharedState,
+}
+
+impl BootstrapManager {
+    pub fn new(runtime: Arc<RuntimeManager>) -> Arc<Self> {
+        Arc::new(Self {
+            app: OnceLock::new(),
+            runtime,
+            shared: AppSharedState::default(),
+        })
+    }
+
+    pub fn app(&self) -> Option<Arc<App>> {
+        self.app.get().cloned()
+    }
+
+    pub fn is_ready(&self) -> bool {
+        self.app.get().is_some()
+    }
+
+    pub fn set_app(&self, app: Arc<App>) -> Result<(), Arc<App>> {
+        self.app.set(app)
+    }
+
+    pub fn runtime(&self) -> Arc<RuntimeManager> {
+        self.runtime.clone()
+    }
+
+    pub fn shared_state(&self) -> AppSharedState {
+        self.shared.clone()
+    }
+
+    pub fn jobs(&self) -> Arc<DashMap<String, JobSummary>> {
+        self.shared.jobs.clone()
+    }
+
+    pub fn downloads(&self) -> Arc<DashMap<String, DownloadProgress>> {
+        self.shared.downloads.clone()
+    }
+
+    pub fn bus(&self) -> Arc<EventBus> {
+        self.shared.bus.clone()
+    }
+
+    pub fn spawn_download_forwarder(&self) {
+        let mut rx = self.runtime.subscribe_downloads();
+        let downloads = self.downloads();
+        let bus = self.bus();
+        tokio::spawn(async move {
+            while let Ok(progress) = rx.recv().await {
+                downloads.insert(progress.id.clone(), progress.clone());
+                bus.publish(AppEvent::DownloadProgress(progress));
+            }
+        });
+    }
+}
+
+impl Deref for BootstrapManager {
+    type Target = App;
+
+    fn deref(&self) -> &Self::Target {
+        self.app
+            .get()
+            .expect("bootstrap routes must guard app access until ready")
+    }
+}

koharu-rpc/src/events.rs

@@ -36,9 +36,11 @@ pub fn router() -> OpenApiRouter<AppState> {
 
 /// Build the current registry snapshot.
 fn snapshot_from(app: &AppState) -> SnapshotEvent {
-    let jobs: Vec<JobSummary> = app.jobs.iter().map(|e| e.value().clone()).collect();
+    let jobs_state = app.jobs();
+    let downloads_state = app.downloads();
+    let jobs: Vec<JobSummary> = jobs_state.iter().map(|e| e.value().clone()).collect();
     let downloads: Vec<DownloadProgress> =
-        app.downloads.iter().map(|e| e.value().clone()).collect();
+        downloads_state.iter().map(|e| e.value().clone()).collect();
     SnapshotEvent { jobs, downloads }
 }
 
@@ -56,7 +58,8 @@ async fn events(
     headers: HeaderMap,
 ) -> Sse<impl Stream<Item = Result<Event, axum::Error>>> {
     let resume_from = last_event_id(&headers);
-    let mut rx = BroadcastStream::new(app.bus.subscribe());
+    let bus = app.bus();
+    let mut rx = BroadcastStream::new(bus.subscribe());
     let app_for_stream = app.clone();
 
     let stream = try_stream! {
@@ -66,10 +69,10 @@ async fn events(
                 // in the ring buffer. If there's a gap (the buffer no longer
                 // contains `last_id + 1`), emit a snapshot first so the client
                 // can rebuild state — then stream whatever tail we still have.
-                let replay = app.bus.replay_since(last_id);
+                let replay = bus.replay_since(last_id);
                 let has_gap = match replay.first() {
                     Some(first) => first.seq > last_id + 1,
-                    None => app.bus.latest_seq().map(|s| s > last_id).unwrap_or(false),
+                    None => bus.latest_seq().map(|s| s > last_id).unwrap_or(false),
                 };
                 if has_gap {
                     yield snapshot_frame(&app)?;

koharu-rpc/src/lib.rs

@@ -11,6 +11,7 @@
 
 pub mod api;
 pub mod binary;
+pub mod bootstrap;
 pub mod error;
 pub mod events;
 pub mod mcp;
@@ -21,7 +22,8 @@ pub mod server;
 use std::sync::Arc;
 
 pub use api::{ApiState, api, router};
+pub use bootstrap::BootstrapManager;
 pub use error::{ApiError, ApiResult};
 
 /// Concrete state threaded through every `State<ApiState>` extractor.
-pub type AppState = Arc<koharu_app::App>;
+pub type AppState = Arc<BootstrapManager>;

koharu-rpc/src/mcp/mod.rs

@@ -30,15 +30,23 @@ use serde::{Deserialize, Serialize};
 use std::sync::atomic::AtomicBool;
 use uuid::Uuid;
 
+use crate::AppState;
+
 /// Server state handed to each tool call. Carries the shared `App`.
 #[derive(Clone)]
 pub struct KoharuServer {
-    app: Arc<App>,
+    state: AppState,
 }
 
 impl KoharuServer {
-    pub fn new(app: Arc<App>) -> Self {
-        Self { app }
+    pub fn new(state: AppState) -> Self {
+        Self { state }
+    }
+
+    fn app(&self) -> Result<Arc<App>, rmcp::ErrorData> {
+        self.state
+            .app()
+            .ok_or_else(|| rmcp::ErrorData::internal_error("app is still bootstrapping", None))
     }
 }
 
@@ -107,20 +115,23 @@ impl KoharuServer {
         &self,
         Parameters(input): Parameters<ApplyInput>,
     ) -> Result<JsonOutput<ApplyOutput>, rmcp::ErrorData> {
+        let app = self.app()?;
         let op: Op = serde_json::from_value(input.op).map_err(err)?;
-        let epoch = self.app.apply(op).map_err(err)?;
+        let epoch = app.apply(op).map_err(err)?;
         Ok(JsonOutput(ApplyOutput { epoch }))
     }
 
     #[tool(name = "koharu.undo", description = "Undo the most recent op")]
     async fn undo(&self) -> Result<JsonOutput<UndoOutput>, rmcp::ErrorData> {
-        let epoch = self.app.undo().map_err(err)?;
+        let app = self.app()?;
+        let epoch = app.undo().map_err(err)?;
         Ok(JsonOutput(UndoOutput { epoch }))
     }
 
     #[tool(name = "koharu.redo", description = "Redo the most recent undo")]
     async fn redo(&self) -> Result<JsonOutput<UndoOutput>, rmcp::ErrorData> {
-        let epoch = self.app.redo().map_err(err)?;
+        let app = self.app()?;
+        let epoch = app.redo().map_err(err)?;
         Ok(JsonOutput(UndoOutput { epoch }))
     }
 
@@ -132,9 +143,9 @@ impl KoharuServer {
         &self,
         Parameters(input): Parameters<OpenProjectInput>,
     ) -> Result<JsonOutput<OpenProjectOutput>, rmcp::ErrorData> {
+        let app = self.app()?;
         let path = Utf8PathBuf::from(input.path);
-        let session = self
-            .app
+        let session = app
             .open_project(path, input.create_name)
             .await
             .map_err(err)?;
@@ -149,7 +160,8 @@ impl KoharuServer {
         description = "Close the active project"
     )]
     async fn close_project(&self) -> Result<JsonOutput<serde_json::Value>, rmcp::ErrorData> {
-        self.app.close_project().await.map_err(err)?;
+        let app = self.app()?;
+        app.close_project().await.map_err(err)?;
         Ok(JsonOutput(serde_json::Value::Null))
     }
 
@@ -161,8 +173,8 @@ impl KoharuServer {
         &self,
         Parameters(input): Parameters<StartPipelineInput>,
     ) -> Result<JsonOutput<StartPipelineOutput>, rmcp::ErrorData> {
-        let session = self
-            .app
+        let app = self.app()?;
+        let session = app
             .current_session()
             .ok_or_else(|| rmcp::ErrorData::invalid_request("no project open", None))?;
         let spec = PipelineSpec {
@@ -180,11 +192,11 @@ impl KoharuServer {
         };
         let job_id = Uuid::new_v4().to_string();
         let cancel = Arc::new(AtomicBool::new(false));
-        let registry = self.app.registry.clone();
-        let runtime = self.app.runtime.clone();
-        let llm = self.app.llm.clone();
-        let renderer = self.app.renderer.clone();
-        let cpu = self.app.cpu_only();
+        let registry = app.registry.clone();
+        let runtime = app.runtime.clone();
+        let llm = app.llm.clone();
+        let renderer = app.renderer.clone();
+        let cpu = app.cpu_only();
         tokio::spawn(async move {
             let _ = koharu_app::pipeline::run(
                 session, registry, runtime, cpu, llm, renderer, spec, cancel, None,
@@ -217,11 +229,11 @@ impl ServerHandler for KoharuServer {
 // ---------------------------------------------------------------------------
 
 /// Mount the MCP endpoint at `/mcp` on `router`.
-pub fn mount(router: axum::Router, app: Arc<App>) -> axum::Router {
+pub fn mount(router: axum::Router, state: AppState) -> axum::Router {
     let manager = Arc::new(LocalSessionManager::default());
     let factory = {
-        let app = app.clone();
-        move || -> Result<KoharuServer, std::io::Error> { Ok(KoharuServer::new(app.clone())) }
+        let state = state.clone();
+        move || -> Result<KoharuServer, std::io::Error> { Ok(KoharuServer::new(state.clone())) }
     };
     let service =
         StreamableHttpService::new(factory, manager, StreamableHttpServerConfig::default());

koharu-rpc/src/routes/downloads.rs

@@ -37,7 +37,8 @@ pub struct ListDownloadsResponse {
     responses((status = 200, body = ListDownloadsResponse))
 )]
 async fn list_downloads(State(app): State<AppState>) -> ApiResult<Json<ListDownloadsResponse>> {
-    let downloads = app.downloads.iter().map(|e| e.value().clone()).collect();
+    let downloads_state = app.downloads();
+    let downloads = downloads_state.iter().map(|e| e.value().clone()).collect();
     Ok(Json(ListDownloadsResponse { downloads }))
 }
 
@@ -72,7 +73,7 @@ async fn start_download(
         .all()
         .find(|p| p.id == req.model_id)
         .ok_or_else(|| ApiError::not_found(format!("unknown package {}", req.model_id)))?;
-    let runtime = app.runtime.clone();
+    let runtime = app.runtime();
     tokio::spawn(async move {
         if let Err(e) = (pkg.ensure)(&runtime).await {
             tracing::error!(package = pkg.id, "download failed: {e:#}");

koharu-rpc/src/routes/operations.rs

@@ -57,7 +57,8 @@ pub struct ListOperationsResponse {
     responses((status = 200, body = ListOperationsResponse))
 )]
 async fn list_operations(State(app): State<AppState>) -> ApiResult<Json<ListOperationsResponse>> {
-    let operations = app.jobs.iter().map(|e| e.value().clone()).collect();
+    let jobs = app.jobs();
+    let operations = jobs.iter().map(|e| e.value().clone()).collect();
     Ok(Json(ListOperationsResponse { operations }))
 }
 
@@ -75,6 +76,6 @@ async fn cancel_operation(
         flag.store(true, Ordering::Relaxed);
     }
     // Best-effort download cancel: drop the registry row.
-    app.downloads.remove(&id);
+    app.downloads().remove(&id);
     Ok(StatusCode::NO_CONTENT)
 }

koharu-rpc/src/server.rs

@@ -11,10 +11,10 @@ use axum::body::Body;
 use axum::extract::Request;
 use axum::http::{HeaderValue, StatusCode, header::CONTENT_TYPE};
 use axum::response::{IntoResponse, Response};
-use koharu_app::App;
 use tokio::net::TcpListener;
 use tower_http::cors::CorsLayer;
 
+use crate::AppState;
 use crate::api;
 
 /// Function that maps a URL path (e.g. `"/index.html"`) to `(bytes, mime)`.
@@ -22,14 +22,14 @@ use crate::api;
 pub type AssetResolver = Arc<dyn Fn(&str) -> Option<(Vec<u8>, String)> + Send + Sync>;
 
 /// Wrap `router(app)` with CORS + mount MCP at `/mcp`.
-pub fn router_for(app: Arc<App>) -> Router {
+pub fn router_for(app: AppState) -> Router {
     let base = api::router(app.clone()).layer(CorsLayer::very_permissive());
     crate::mcp::mount(base, app)
 }
 
 /// Same as `router_for` but installs `resolver` as a fallback, serving
 /// embedded frontend assets for unmatched GET requests.
-pub fn router_with_assets(app: Arc<App>, resolver: AssetResolver) -> Router {
+pub fn router_with_assets(app: AppState, resolver: AssetResolver) -> Router {
     router_for(app).fallback(move |req: Request<Body>| {
         let resolver = resolver.clone();
         async move { serve_asset(resolver, req).await }
@@ -53,7 +53,7 @@ async fn serve_asset(resolver: AssetResolver, req: Request<Body>) -> Response {
 }
 
 /// Serve HTTP on an already-bound listener. Tauri-friendly.
-pub async fn serve_with_listener(listener: TcpListener, app: Arc<App>) -> Result<()> {
+pub async fn serve_with_listener(listener: TcpListener, app: AppState) -> Result<()> {
     axum::serve(listener, router_for(app)).await?;
     Ok(())
 }
@@ -62,7 +62,7 @@ pub async fn serve_with_listener(listener: TcpListener, app: Arc<App>) -> Result
 /// production build to serve the bundled UI.
 pub async fn serve_with_listener_and_assets(
     listener: TcpListener,
-    app: Arc<App>,
+    app: AppState,
     resolver: AssetResolver,
 ) -> Result<()> {
     axum::serve(listener, router_with_assets(app, resolver)).await?;

koharu/src/app.rs

@@ -6,7 +6,7 @@ use std::sync::Arc;
 use anyhow::{Context, Result};
 use clap::Parser;
 use koharu_app::{App, AppConfig, config as app_config};
-use koharu_rpc::server;
+use koharu_rpc::{BootstrapManager, server};
 use koharu_runtime::{ComputePolicy, RuntimeHttpConfig, RuntimeManager};
 use tokio::net::TcpListener;
 use tracing_subscriber::layer::SubscriberExt;
@@ -14,10 +14,35 @@ use tracing_subscriber::util::SubscriberInitExt;
 
 use crate::cli::Cli;
 
+async fn bootstrap_app(
+    state: Arc<BootstrapManager>,
+    config: AppConfig,
+    cpu_only: bool,
+) -> Result<()> {
+    let runtime = state.runtime();
+    runtime
+        .prepare()
+        .await
+        .context("failed to prepare runtime")?;
+
+    let app = Arc::new(App::new_with_shared_state(
+        config,
+        runtime,
+        cpu_only,
+        state.shared_state(),
+        crate::version::current(),
+    )?);
+    koharu_llm::suppress_native_logs();
+    app.spawn_llm_forwarder();
+    state
+        .set_app(app)
+        .map_err(|_| anyhow::anyhow!("app already initialized"))?;
+    Ok(())
+}
+
 pub async fn run() -> Result<()> {
     let cli = Cli::parse();
 
-    // ── Platform & logging ───────────────────────────────────────────
     #[cfg(target_os = "windows")]
     {
         let attached = crate::windows::attach_parent_console();
@@ -41,7 +66,6 @@ pub async fn run() -> Result<()> {
         koharu_llm::providers::disable_keyring();
     }
 
-    // ── Config ───────────────────────────────────────────────────────
     let config: AppConfig = app_config::load()?;
     let http = RuntimeHttpConfig {
         connect_timeout_secs: config.http.connect_timeout.max(1),
@@ -61,27 +85,16 @@ pub async fn run() -> Result<()> {
             .context("failed to download runtime packages");
     }
 
-    // ── Runtime + App ────────────────────────────────────────────────
-    let runtime = RuntimeManager::new_with_http(config.data.path.as_std_path(), compute, http)?;
-    runtime
-        .prepare()
-        .await
-        .context("failed to prepare runtime")?;
+    let state = BootstrapManager::new(Arc::new(RuntimeManager::new_with_http(
+        config.data.path.as_std_path(),
+        compute,
+        http,
+    )?));
+    state.spawn_download_forwarder();
 
     #[cfg(target_os = "windows")]
     crate::windows::register_khr().ok();
 
-    let app = Arc::new(App::new(
-        config,
-        Arc::new(runtime),
-        cli.cpu,
-        crate::version::current(),
-    )?);
-    koharu_llm::suppress_native_logs();
-    app.spawn_download_forwarder();
-    app.spawn_llm_forwarder();
-
-    // ── Server ───────────────────────────────────────────────────────
     let default_port = if cfg!(debug_assertions) { 9999 } else { 0 };
     let bind_host = cli.host.as_deref().unwrap_or("127.0.0.1");
     let bind_port = cli.port.unwrap_or(default_port);
@@ -89,31 +102,23 @@ pub async fn run() -> Result<()> {
     let port = listener.local_addr()?.port();
     tracing::info!(port, "starting server");
 
-    // Extract the embedded UI assets up-front so both headless and GUI modes
-    // can serve them on the HTTP fallback. Headless deployments point a
-    // browser at the listener port and get the full app; the GUI build uses
-    // the same bundle inside Tauri's webview.
     let mut context = tauri::generate_context!();
     let assets = crate::assets::from_context(&mut context);
+    let server_state = state.clone();
+    tauri::async_runtime::spawn(async move {
+        if let Err(e) = server::serve_with_listener_and_assets(listener, server_state, assets).await
+        {
+            tracing::error!("server error: {e:#}");
+        }
+    });
 
     if cli.headless {
-        tauri::async_runtime::spawn(async move {
-            if let Err(e) = server::serve_with_listener_and_assets(listener, app, assets).await {
-                tracing::error!("server error: {e:#}");
-            }
-        });
         tracing::info!(port, "headless: open http://127.0.0.1:{port}/ in a browser");
+        bootstrap_app(state, config, cli.cpu).await?;
         tokio::signal::ctrl_c().await?;
         return Ok(());
     }
 
-    // ── GUI ──────────────────────────────────────────────────────────
-    tauri::async_runtime::spawn(async move {
-        if let Err(e) = server::serve_with_listener_and_assets(listener, app, assets).await {
-            tracing::error!("server error: {e:#}");
-        }
-    });
-
     tauri::Builder::default()
         .plugin(tauri_plugin_dialog::init())
         .plugin(tauri_plugin_fs::init())
@@ -121,6 +126,12 @@ pub async fn run() -> Result<()> {
         .plugin(tauri_plugin_updater::Builder::new().build())
         .plugin(tauri_plugin_process::init())
         .setup(move |handle| {
+            tauri::async_runtime::spawn(async move {
+                if let Err(e) = bootstrap_app(state, config, cli.cpu).await {
+                    tracing::error!("bootstrap error: {e:#}");
+                }
+            });
+
             let cfg = handle.config();
             let url: tauri::Url = if cfg!(debug_assertions) {
                 cfg.build

tests/integration-tests/src/harness.rs

@@ -17,7 +17,7 @@ use anyhow::{Context, Result};
 use camino::Utf8PathBuf;
 use koharu_app::{App, AppConfig};
 use koharu_client::apis::configuration::Configuration;
-use koharu_rpc::server;
+use koharu_rpc::{BootstrapManager, server};
 use koharu_runtime::{ComputePolicy, RuntimeHttpConfig, RuntimeManager};
 use tokio::net::TcpListener;
 use tokio::sync::OnceCell;
@@ -93,15 +93,26 @@ impl TestApp {
 
         // Shared runtime (prepared once per process).
         let runtime = shared_runtime().await?;
-        let app = Arc::new(App::new(config, runtime, true, "test")?);
+        let state = BootstrapManager::new(runtime.clone());
+        state.spawn_download_forwarder();
+        let app = Arc::new(App::new_with_shared_state(
+            config,
+            runtime,
+            true,
+            state.shared_state(),
+            "test",
+        )?);
         app.spawn_llm_forwarder();
+        state
+            .set_app(app.clone())
+            .map_err(|_| anyhow::anyhow!("test app already initialized"))?;
 
         let listener = TcpListener::bind("127.0.0.1:0").await?;
         let addr = listener.local_addr()?;
         let base_url = format!("http://{addr}/api/v1");
         let server = tokio::spawn({
-            let app = app.clone();
-            async move { server::serve_with_listener(listener, app).await }
+            let state = state.clone();
+            async move { server::serve_with_listener(listener, state).await }
         });
 
         let client_config = Configuration {