refactor(daemon): run in-process invocations and MCP dispatch natively

crates/tracedecay-dashboard-api/src/lib.rs

@@ -145,7 +145,10 @@ pub use http_admission::{
     with_dashboard_http_admission,
 };
 pub use routes::router;
-pub use server::{bind_dashboard, serve_http_connection, serve_loopback, validate_dashboard_host};
+pub use server::{
+    ConnectionExecutor, bind_dashboard, serve_http_connection, serve_loopback,
+    validate_dashboard_host,
+};
 pub use state::{
     AdmittedDoctorReportV1, AutomationSchedulerReconcileFuture, AutomationSchedulerReconciler,
     DashboardAutomationObservationFuture, DashboardAutomationObservationRecorderV1, DashboardState,

crates/tracedecay-dashboard-api/src/server.rs

@@ -16,6 +16,7 @@ use hyper_util::service::TowerToHyperService;
 use tokio::task::JoinSet;
 use tokio_util::sync::CancellationToken;
 use tower::ServiceExt;
+use tracedecay_runtime_core::native_runtime::{NativeTaskOutcome, NativeTaskSet};
 
 use crate::state::config_error;
 use tracedecay_domain::errors::Result;
@@ -50,42 +51,99 @@ pub fn validate_dashboard_host(host: &str) -> Result<&str> {
 /// Matches `axum::serve(..).with_graceful_shutdown(..)` on an Asupersync
 /// socket. The drain has no deadline because callers own one. Accept errors
 /// never end the loop. Dropping the future aborts every connection.
-/// Connections run on `runtime`, because daemon routes served here still
-/// reach Tokio.
 pub async fn serve_loopback(
-    runtime: tokio::runtime::Handle,
+    executor: ConnectionExecutor,
     listener: TcpListener,
     router: Router,
     shutdown: impl Future<Output = ()>,
 ) {
     let graceful = CancellationToken::new();
-    let mut connections = JoinSet::new();
+    let mut connections = Connections::new(executor);
     let mut shutdown = std::pin::pin!(shutdown);
     loop {
         tokio::select! {
             biased;
             () = &mut shutdown => break,
-            Some(joined) = connections.join_next(), if !connections.is_empty() => {
-                observe_connection_join(joined);
-            }
+            Some(()) = connections.join_next(), if !connections.is_empty() => {}
             accepted = listener.accept() => match accepted {
                 Ok((stream, address)) => {
                     let router = router.clone();
                     let graceful = graceful.clone();
-                    connections.spawn_on(async move {
+                    connections.spawn(async move {
                         if let Err(error) = serve_http_connection(stream, router, true, graceful).await {
                             tracing::trace!(%error, %address, "loopback HTTP connection stopped");
                         }
-                    }, &runtime);
+                    });
                 }
                 Err(error) => wait_after_accept_error(&error).await,
             },
         }
     }
     drop(listener);
     graceful.cancel();
-    while let Some(joined) = connections.join_next().await {
-        observe_connection_join(joined);
+    while connections.join_next().await.is_some() {}
+}
+
+/// Where [`serve_loopback`] runs each connection.
+pub enum ConnectionExecutor {
+    Native,
+    /// For daemon routes that still reach `project_server_lifecycle` inside
+    /// their connection task.
+    Tokio(tokio::runtime::Handle),
+}
+
+enum Connections {
+    Native(NativeTaskSet<()>),
+    Tokio(tokio::runtime::Handle, JoinSet<()>),
+}
+
+impl Connections {
+    fn new(executor: ConnectionExecutor) -> Self {
+        match executor {
+            ConnectionExecutor::Native => Self::Native(NativeTaskSet::default()),
+            ConnectionExecutor::Tokio(runtime) => Self::Tokio(runtime, JoinSet::new()),
+        }
+    }
+
+    /// A refused connection's IO drops with it, closing the socket.
+    fn spawn(&mut self, connection: impl Future<Output = ()> + Send + 'static) {
+        match self {
+            Self::Native(set) => {
+                if let Err(error) = set.spawn(connection) {
+                    tracing::warn!(%error, "loopback HTTP connection not admitted");
+                }
+            }
+            Self::Tokio(runtime, set) => {
+                set.spawn_on(connection, runtime);
+            }
+        }
+    }
+
+    fn is_empty(&self) -> bool {
+        match self {
+            Self::Native(set) => set.is_empty(),
+            Self::Tokio(_, set) => set.is_empty(),
+        }
+    }
+
+    /// Joins the next connection to end, logging a failed one; `None` once
+    /// none remain.
+    async fn join_next(&mut self) -> Option<()> {
+        match self {
+            Self::Native(set) => {
+                if let NativeTaskOutcome::Panicked(payload) = set.join_next().await? {
+                    tracing::warn!(?payload, "loopback HTTP connection task failed");
+                }
+            }
+            Self::Tokio(_, set) => {
+                if let Err(error) = set.join_next().await?
+                    && !error.is_cancelled()
+                {
+                    tracing::warn!(%error, "loopback HTTP connection task failed");
+                }
+            }
+        }
+        Some(())
     }
 }
 
@@ -104,14 +162,6 @@ async fn wait_after_accept_error(error: &io::Error) {
     Sleep::after(wall_now(), Duration::from_secs(1)).await;
 }
 
-fn observe_connection_join(joined: std::result::Result<(), tokio::task::JoinError>) {
-    if let Err(error) = joined
-        && !error.is_cancelled()
-    {
-        tracing::warn!(%error, "loopback HTTP connection task failed");
-    }
-}
-
 /// Serves one HTTP/1.1 connection with `router` and starts a graceful
 /// shutdown once `graceful` is cancelled. hyper runs without an executor or a
 /// timer, so the IO is the only runtime seam.

crates/tracedecay-dashboard-api/src/server_tests.rs

@@ -8,7 +8,7 @@ use tokio::sync::{Notify, oneshot};
 
 use tracedecay_runtime_core::native_runtime::{self, NativeTask, NativeTaskOutcome};
 
-use crate::server::{bind_dashboard, serve_loopback};
+use crate::server::{ConnectionExecutor, bind_dashboard, serve_loopback};
 
 fn get_over_std_socket(addr: SocketAddr, path: &str) -> String {
     let mut stream = TcpStream::connect(addr).expect("connect loopback server");
@@ -51,9 +51,14 @@ fn loopback_server_drains_in_flight_request_after_shutdown_without_tokio_drivers
             .await
             .expect("bind loopback listener");
         let (stop, stopped) = oneshot::channel::<()>();
-        let server = tokio::spawn(serve_loopback(runtime.clone(), listener, router, async {
-            let _ = stopped.await;
-        }));
+        let server = tokio::spawn(serve_loopback(
+            ConnectionExecutor::Tokio(runtime.clone()),
+            listener,
+            router,
+            async {
+                let _ = stopped.await;
+            },
+        ));
         let client = tokio::task::spawn_blocking(move || get_over_std_socket(addr, "/slow"));
         entered.notified().await;
 
@@ -80,21 +85,23 @@ fn loopback_server_drains_in_flight_request_after_shutdown_without_tokio_drivers
     });
 }
 
-/// A native caller hands the serve loop its Tokio handle; the loop needs no
-/// ambient Tokio runtime to serve.
+/// The hosted dashboard serves every connection natively, with no Tokio
+/// runtime in the process path.
 #[test]
-fn loopback_server_serves_from_a_native_task_with_an_explicit_handle() {
-    let tokio = tokio::runtime::Builder::new_multi_thread()
-        .worker_threads(1)
-        .build()
-        .expect("Tokio runtime");
-    let router = Router::new().route("/ping", get(|| async { "pong" }));
+fn loopback_server_serves_native_connections_without_tokio() {
+    let router = Router::new().route(
+        "/ping",
+        get(|| async {
+            assert!(tokio::runtime::Handle::try_current().is_err());
+            "pong"
+        }),
+    );
     let (listener, addr) = native_runtime::block_on(bind_dashboard("127.0.0.1", 0))
         .expect("native root")
         .expect("bind loopback listener");
     let (stop, stopped) = oneshot::channel::<()>();
     let server = NativeTask::spawn(serve_loopback(
-        tokio.handle().clone(),
+        ConnectionExecutor::Native,
         listener,
         router,
         async {

crates/tracedecay-dashboard-api/src/test_transport.rs

@@ -9,9 +9,10 @@ use tracedecay_global_db::RegisteredGlobalDbLeaseV1;
 
 use crate::state::build_state_inner;
 use crate::{
-    DashboardAccessToken, DashboardApplicationRuntime, DashboardAutomationAuthorityV1,
-    DashboardAutomationWriter, DashboardDeliveryReadPortV1, DashboardGitCorrelationReadPortV1,
-    DashboardLcmReadPortV1, DashboardProfileCodeIndexWorkerSettingsPort, DashboardProjectContext,
+    ConnectionExecutor, DashboardAccessToken, DashboardApplicationRuntime,
+    DashboardAutomationAuthorityV1, DashboardAutomationWriter, DashboardDeliveryReadPortV1,
+    DashboardGitCorrelationReadPortV1, DashboardLcmReadPortV1,
+    DashboardProfileCodeIndexWorkerSettingsPort, DashboardProjectContext,
     DashboardSessionAuthoritiesV1, DashboardSessionMountV1, DashboardSessionResolutionV1,
     DashboardStateCompositionV1, PrAutoTrackManagedSummaryReader, bind_dashboard, code_read_api,
     router, serve_loopback, standalone_dashboard_automation_writer, with_dashboard_http_admission,
@@ -318,6 +319,6 @@ where
             .ok_or_else(|| {
                 crate::state::config_error("the test dashboard found no Tokio runtime")
             })?;
-    serve_loopback(runtime, listener, app, shutdown).await;
+    serve_loopback(ConnectionExecutor::Tokio(runtime), listener, app, shutdown).await;
     Ok(())
 }

crates/tracedecay-mcp/Cargo.toml

@@ -32,7 +32,7 @@ serde = { version = "1", features = ["derive"] }
 serde_json = "1"
 sha2 = "0.11"
 thiserror = "2"
-tokio = { version = "1", features = ["macros", "rt", "sync"] }
+tokio = { version = "1", features = ["macros", "sync"] }
 toml = "1"
 tracedecay-agent-hosts = { path = "../tracedecay-agent-hosts" }
 tracedecay-api = { path = "../tracedecay-api" }

crates/tracedecay-mcp/src/server/settlement.rs

@@ -4,7 +4,7 @@ use std::sync::Arc;
 use std::sync::atomic::{AtomicBool, AtomicU8, AtomicUsize, Ordering};
 
 use tracedecay_domain::errors::{Result, TraceDecayError};
-use tracedecay_runtime_core::native_runtime;
+use tracedecay_runtime_core::native_runtime::{self, NativeTaskOutcome, NativeTaskSet};
 
 const SETTLEMENT_NOT_STARTED: u8 = 0;
 const SETTLEMENT_SETTLING: u8 = 1;
@@ -155,9 +155,12 @@ impl Drop for DispatchCapacityLease {
     }
 }
 
+type DispatchId = u64;
+
 struct RetainedDispatchState {
-    tasks: tokio::task::JoinSet<Arc<DispatchExecutionSettlement>>,
-    active: HashMap<tokio::task::Id, ActiveDispatch>,
+    tasks: NativeTaskSet<(DispatchId, Arc<DispatchExecutionSettlement>)>,
+    active: HashMap<DispatchId, ActiveDispatch>,
+    next_id: DispatchId,
 }
 
 /// Daemon-owned lifetime authority for admitted MCP handlers.
@@ -186,8 +189,9 @@ impl RetainedDispatchRegistry {
             capacity,
             active_slots: Arc::new(AtomicUsize::new(0)),
             state: tokio::sync::Mutex::new(RetainedDispatchState {
-                tasks: tokio::task::JoinSet::new(),
+                tasks: NativeTaskSet::default(),
                 active: HashMap::new(),
+                next_id: 0,
             }),
             #[cfg(any(test, feature = "test-transport"))]
             retained_spawn_count: AtomicUsize::new(0),
@@ -239,10 +243,6 @@ impl RetainedDispatchRegistry {
         T: Send + 'static,
         F: Future<Output = Result<T>> + Send + 'static,
     {
-        // Tool dispatches still reach Tokio, so they settle there.
-        let runtime =
-            tracedecay_runtime_core::native_runtime::tokio_handle("mcp.dispatch.settlement")
-                .ok_or_else(dispatch_runtime_unavailable_error)?;
         let capacity_lease = self.acquire_capacity()?;
         let mut state = self.state.lock().await;
         Self::reap_finished(&mut state);
@@ -254,21 +254,26 @@ impl RetainedDispatchRegistry {
         let worker_settlement = Arc::clone(&settlement);
         let settlement_lease = DispatchSettlementLease(Arc::clone(&settlement));
         let (sender, receiver) = tokio::sync::oneshot::channel();
-        #[cfg(any(test, feature = "test-transport"))]
-        self.retained_spawn_count.fetch_add(1, Ordering::AcqRel);
-        let task = state.tasks.spawn_on(
-            async move {
+        let id = state.next_id;
+        state.next_id += 1;
+        // Tool dispatches run on the native runtime. Handlers move synchronous
+        // storage-port calls off the worker with `off_native_thread` and reach
+        // Tokio-only work through `native_runtime::tokio_handle`.
+        state
+            .tasks
+            .spawn(async move {
                 let _capacity_lease = capacity_lease;
                 worker_settlement.mark_settling();
                 let output = future.await;
                 drop(settlement_lease);
                 let _ = sender.send(output);
-                worker_settlement
-            },
-            &runtime,
-        );
+                (id, worker_settlement)
+            })
+            .map_err(|error| dispatch_runtime_unavailable_error(&error))?;
+        #[cfg(any(test, feature = "test-transport"))]
+        self.retained_spawn_count.fetch_add(1, Ordering::AcqRel);
         state.active.insert(
-            task.id(),
+            id,
             ActiveDispatch {
                 cancellation,
                 live_cancellable,
@@ -295,28 +300,27 @@ impl RetainedDispatchRegistry {
     }
 
     fn reap_finished(state: &mut RetainedDispatchState) {
-        while let Some(joined) = state.tasks.try_join_next_with_id() {
+        while let Some(joined) = state.tasks.try_join_next() {
             Self::settle_joined(state, joined);
         }
     }
 
     fn settle_joined(
         state: &mut RetainedDispatchState,
-        joined: std::result::Result<
-            (tokio::task::Id, Arc<DispatchExecutionSettlement>),
-            tokio::task::JoinError,
-        >,
+        joined: NativeTaskOutcome<(DispatchId, Arc<DispatchExecutionSettlement>)>,
     ) {
         match joined {
-            Ok((id, settlement)) => {
+            NativeTaskOutcome::Completed((id, settlement)) => {
                 settlement.mark_joined();
                 state.active.remove(&id);
             }
-            Err(error) => {
-                if let Some(active) = state.active.remove(&error.id()) {
-                    active.settlement.mark_joined();
-                }
-                tracing::error!(error = %error, "retained MCP dispatch task failed");
+            failed => {
+                // A failed task returns no id. Its dropped settlement lease
+                // already marked it joined, which identifies its entry.
+                state
+                    .active
+                    .retain(|_, active| active.settlement.snapshot() != DispatchSettlement::Joined);
+                tracing::error!(outcome = ?failed, "retained MCP dispatch task failed");
             }
         }
     }
@@ -330,7 +334,7 @@ impl RetainedDispatchRegistry {
                 let _ = active.cancellation.cancel(requested_at);
             }
         }
-        while let Some(joined) = state.tasks.join_next_with_id().await {
+        while let Some(joined) = state.tasks.join_next().await {
             Self::settle_joined(&mut state, joined);
         }
     }
@@ -783,11 +787,11 @@ fn dispatch_shutdown_error() -> TraceDecayError {
     )
 }
 
-fn dispatch_runtime_unavailable_error() -> TraceDecayError {
+fn dispatch_runtime_unavailable_error(error: &str) -> TraceDecayError {
     TraceDecayError::project_route(
         "tool_dispatch_runtime_unavailable",
         true,
-        "MCP tool dispatch found no Tokio runtime",
+        format!("MCP tool dispatch was not started: {error}"),
     )
 }
 

crates/tracedecay-rusqlite-runtime/Cargo.toml

@@ -24,9 +24,9 @@ ring = "0.17.14"
 sha2 = "0.11.0"
 thiserror = "2"
 asupersync.workspace = true
-# Synchronous storage ports wait on the writer from Tokio daemon workers; see
-# `recv_writer_reply`.
-tokio = { version = "1", default-features = false, features = ["rt", "rt-multi-thread"] }
+# `native_runtime::tokio_handle` hands the daemon's Tokio runtime to the sites
+# that still need one.
+tokio = { version = "1", default-features = false, features = ["rt"] }
 tracedecay-domain = { path = "../tracedecay-domain" }
 tracedecay-contracts = { path = "../tracedecay-contracts" }
 tracedecay-store = { path = "../tracedecay-store" }