perf(daemon): serve Work/Workflow requests off native workers

crates/tracedecay-daemon-service/src/invocation/observability_producer.rs

@@ -238,13 +238,23 @@ impl DaemonInvocationService {
             .project_runtimes
             .get::<RegisteredWorkRuntime>(project_root?)
             .await?;
-        let workflow =
-            tracedecay_application::work::RegisteredWorkflowApplicationServicesV1::attach(
-                &runtime.database,
+        let identity = identity.clone();
+        // The port waits on the writer; keep that wait off native workers.
+        tracedecay_runtime_core::native_runtime::off_native_thread(async move {
+            let workflow =
+                tracedecay_application::work::RegisteredWorkflowApplicationServicesV1::attach(
+                    &runtime.database,
+                )
+                .ok()?;
+            tracedecay_contracts::WorkflowRunStoragePort::fan_out_binding(
+                workflow.effects(),
+                &identity,
             )
-            .ok()?;
-        tracedecay_contracts::WorkflowRunStoragePort::fan_out_binding(workflow.effects(), identity)
             .ok()
             .flatten()
+        })
+        .await
+        .ok()
+        .flatten()
     }
 }

crates/tracedecay-daemon-service/src/invocation/work.rs

@@ -106,8 +106,9 @@ pub async fn execute_work_application(
     deadline: Deadline,
     cancellation: CancellationContext,
 ) -> DaemonInvocationResponse {
+    let refused_request_id = request_id.clone();
     // Keep the Work dispatch and activity-publication frame out of invocation callers.
-    Box::pin(async move {
+    off_native_worker(refused_request_id, async move {
         let activity_database = registered.database.clone();
         let activity_root = project_root.clone();
         let mutates = work_invocation_mutates(&request);
@@ -146,6 +147,22 @@ pub async fn execute_work_application(
     .await
 }
 
+/// Serves a Work or Workflow request off native runtime workers. Their storage
+/// ports are synchronous and park the calling thread on every writer wait,
+/// and a parked worker drives no other task.
+async fn off_native_worker(
+    request_id: String,
+    request: impl Future<Output = DaemonInvocationResponse> + Send + 'static,
+) -> DaemonInvocationResponse {
+    match tracedecay_runtime_core::native_runtime::off_native_thread(request).await {
+        Ok(response) => response,
+        Err(error) => {
+            tracing::error!(%error, "Work request did not run on the blocking pool");
+            DaemonInvocationResponse::problem(request_id, DaemonInvocationProblem::Unavailable)
+        }
+    }
+}
+
 fn publish_committed_task_activity_in_background(
     database: tracedecay_global_db::RegisteredGlobalDbLeaseV1,
     project_root: PathBuf,

crates/tracedecay-daemon-service/src/invocation/work/workflow_dispatch.rs

@@ -30,8 +30,41 @@ use super::workflow_run_control::{
 use super::{RegisteredWorkRuntime, work_request_context, workflow_census};
 
 #[allow(clippy::too_many_arguments)]
-#[tracing::instrument(name = "daemon.service.workflow.execute", level = "trace", skip_all)]
 pub(crate) async fn execute_workflow_application(
+    registered: RegisteredWorkRuntime,
+    attempt_processes: Arc<WorkAttemptProcessRegistryV1>,
+    observability_producer: Option<
+        Arc<tracedecay_application::observability::BoundedObservabilityProducerV1>,
+    >,
+    project_root: PathBuf,
+    request_id: String,
+    request: WorkflowApplicationInvocation,
+    observed_at: UtcMicros,
+    deadline: Deadline,
+    cancellation: CancellationContext,
+    worktree_holder_admission: tracedecay_agent_hosts::native_integration::WorktreeHolderAdmissionFenceV1,
+) -> DaemonInvocationResponse {
+    super::off_native_worker(
+        request_id.clone(),
+        workflow_application(
+            registered,
+            attempt_processes,
+            observability_producer,
+            project_root,
+            request_id,
+            request,
+            observed_at,
+            deadline,
+            cancellation,
+            worktree_holder_admission,
+        ),
+    )
+    .await
+}
+
+#[allow(clippy::too_many_arguments)]
+#[tracing::instrument(name = "daemon.service.workflow.execute", level = "trace", skip_all)]
+async fn workflow_application(
     registered: RegisteredWorkRuntime,
     attempt_processes: Arc<WorkAttemptProcessRegistryV1>,
     observability_producer: Option<

crates/tracedecay-rusqlite-runtime/src/exact_sql/mod.rs

@@ -120,7 +120,7 @@ pub struct ExactSqlHandle {
 /// (a worker or a `block_on` caller; its blocking pool carries no runtime
 /// handle) still parks that thread. Each such wait is logged and counted in
 /// [`native_thread_writer_waits`] so the caller can be moved onto
-/// `native_runtime::spawn_blocking`.
+/// `native_runtime::off_native_thread` or `native_runtime::spawn_blocking`.
 #[track_caller]
 fn recv_writer_reply<T>(
     response: async_channel::Receiver<T>,

crates/tracedecay-rusqlite-runtime/src/native_runtime.rs

@@ -91,6 +91,44 @@ where
     })
 }
 
+/// Runs `future` where a thread may park: from a native runtime thread it
+/// moves the whole future to one blocking-pool thread and polls it there;
+/// anywhere else it polls in place.
+///
+/// For request handlers that call synchronous storage ports, which wait on
+/// the writer by parking their thread. Native wakes reach the pool thread
+/// through an ordinary `Waker`, so the handler's awaits still work. As with
+/// [`spawn_blocking`], dropping the returned future abandons only the wait:
+/// a moved future runs to completion.
+pub fn off_native_thread<F>(future: F) -> BoxFuture<'static, Result<F::Output, String>>
+where
+    F: Future + Send + 'static,
+    F::Output: Send + 'static,
+{
+    if Runtime::current_handle().is_none() {
+        return Box::pin(future.map(Ok));
+    }
+    spawn_blocking(move || poll_on_current_thread(future))
+}
+
+fn poll_on_current_thread<F: Future>(future: F) -> F::Output {
+    struct Unpark(std::thread::Thread);
+    impl std::task::Wake for Unpark {
+        fn wake(self: std::sync::Arc<Self>) {
+            self.0.unpark();
+        }
+    }
+    let waker = std::task::Waker::from(std::sync::Arc::new(Unpark(std::thread::current())));
+    let mut context = Context::from_waker(&waker);
+    let mut future = std::pin::pin!(future);
+    loop {
+        if let Poll::Ready(output) = future.as_mut().poll(&mut context) {
+            return output;
+        }
+        std::thread::park();
+    }
+}
+
 fn queue_blocking<T, F>(work: F) -> Result<(Cx, asupersync::channel::oneshot::Receiver<T>), String>
 where
     T: Send + 'static,
@@ -774,6 +812,36 @@ mod tests {
         );
     }
 
+    #[test]
+    fn off_native_thread_moves_a_native_caller_and_keeps_its_awaits_working() {
+        let off_native = || async {
+            let parked_thread = Runtime::current_handle().is_none();
+            let slept = sleep_until(Instant::now() + Duration::from_millis(5)).await;
+            let blocking = spawn_blocking(|| 7).await;
+            (parked_thread, slept, blocking)
+        };
+        let task = NativeTask::spawn(async move {
+            let caller_was_native = Runtime::current_handle().is_some();
+            (caller_was_native, off_native_thread(off_native()).await)
+        })
+        .expect("native task");
+        assert_eq!(
+            block_on(task.wait()).expect("process runtime"),
+            NativeTaskOutcome::Completed((true, Ok((true, Ok(()), Ok(7)))))
+        );
+        let in_place = std::thread::spawn(move || {
+            let thread = std::thread::current().id();
+            off_native_thread(async move { std::thread::current().id() == thread }).now_or_never()
+        })
+        .join()
+        .expect("plain thread");
+        assert_eq!(
+            in_place,
+            Some(Ok(true)),
+            "a non-native caller polls in place"
+        );
+    }
+
     #[tokio::test(flavor = "multi_thread")]
     async fn a_past_deadline_takes_a_ready_value_and_elapses_a_pending_one() {
         let past = Instant::now() - Duration::from_millis(1);

crates/tracedecay/src/daemon/invocation_tests/work_tests.rs

@@ -924,14 +924,18 @@ impl RegisteredWorkFixture {
             grant_digest.clone(),
             actor.clone(),
             UtcMicros(1),
-            UtcMicros(10_000),
+            // Workflow dispatch admits at the wall clock, not the request's
+            // observed time.
+            UtcMicros(tracedecay_contracts::now_micros().0 + 3_600_000_000),
             scope.clone(),
             tracedecay_contracts::WORK_APPLICATION_OPERATION_IDS_V1
                 .iter()
+                .chain(&tracedecay_contracts::WORKFLOW_APPLICATION_OPERATION_IDS)
                 .map(|(_, capability, _)| CapabilityId::new(*capability).expect("capability"))
                 .collect(),
             tracedecay_contracts::WORK_APPLICATION_OPERATION_IDS_V1
                 .iter()
+                .chain(&tracedecay_contracts::WORKFLOW_APPLICATION_OPERATION_IDS)
                 .map(|(_, _, use_case)| UseCaseId::new(*use_case).expect("use case"))
                 .collect(),
             DisclosureClass::Sensitive,
@@ -987,23 +991,165 @@ impl RegisteredWorkFixture {
         request_id: &'static str,
         request: WorkApplicationInvocationV1,
     ) -> DaemonInvocationOutcome {
-        self.service
-            .invoke(
-                &self.registry,
-                Some(self.project.path()),
-                None,
-                None,
-                None,
-                DaemonInvocationRequest::work_application(
-                    request_id,
+        invoke_registered(
+            self.service.clone(),
+            Arc::clone(&self.registry),
+            self.project.path().to_path_buf(),
+            DaemonInvocationRequest::work_application(
+                request_id,
+                request,
+                UtcMicros(100),
+                Deadline::new(UtcMicros(1_000)).expect("deadline"),
+                CancellationContext::active(format!("cancel.{request_id}")).expect("cancellation"),
+            ),
+        )
+        .await
+    }
+}
+
+async fn invoke_registered(
+    service: DaemonInvocationService,
+    registry: Arc<Mutex<LspSessionRegistry>>,
+    project_root: std::path::PathBuf,
+    request: DaemonInvocationRequest,
+) -> DaemonInvocationOutcome {
+    service
+        .invoke(&registry, Some(&project_root), None, None, None, request)
+        .await
+        .outcome
+}
+
+/// Broker-socket connections run on native runtime workers. Work and Workflow
+/// storage ports are synchronous and wait on the `SQLite` writer, so dispatch
+/// must run them off those workers: a parked worker drives no other task.
+#[tokio::test]
+async fn work_and_workflow_dispatch_never_park_a_native_worker_on_the_writer() {
+    let fixture = RegisteredWorkFixture::start("native-writer-waits").await;
+    let wall_clock = tracedecay_contracts::now_micros();
+    let workflow = |request_id: &str, request| {
+        DaemonInvocationRequest::workflow_application(
+            request_id,
+            request,
+            wall_clock,
+            Deadline::new(UtcMicros(wall_clock.0 + 600_000_000)).expect("deadline"),
+            CancellationContext::active(format!("cancel.{request_id}")).expect("cancellation"),
+        )
+    };
+    let digest = |byte: char| {
+        ManifestDigest::new(format!("sha256:{}", byte.to_string().repeat(64))).expect("digest")
+    };
+    let definition = tracedecay_domain::WorkflowDefinition::new(
+        tracedecay_domain::WorkflowDefinitionId::new("workflow.native-writer-waits")
+            .expect("definition id"),
+        1,
+        fixture.scope.project_id.clone(),
+        vec![tracedecay_domain::WorkflowStep {
+            step_id: tracedecay_domain::WorkflowStepId::new("prepare").expect("step id"),
+            operation: tracedecay_domain::WorkflowOperationRef::new("operation.work.start_attempt")
+                .expect("operation"),
+            predecessors: BTreeSet::new(),
+            inputs: Vec::new(),
+            outputs: Vec::new(),
+            fan_out: None,
+        }],
+        digest('a'),
+        digest('b'),
+        digest('c'),
+    )
+    .expect("workflow definition");
+    let requests = vec![
+        workflow(
+            "request.workflow.native-writer-waits.register",
+            tracedecay_daemon_protocol::WorkflowApplicationInvocation::RegisterDefinition(
+                tracedecay_contracts::WorkflowDefinitionRegisterRequest { definition },
+            ),
+        ),
+        workflow(
+            "request.workflow.native-writer-waits.get-run",
+            tracedecay_daemon_protocol::WorkflowApplicationInvocation::GetRun(
+                tracedecay_contracts::WorkflowRunGetRequest {
+                    run_id: tracedecay_domain::RunId::new("run.native-writer-waits")
+                        .expect("run id"),
+                },
+            ),
+        ),
+        DaemonInvocationRequest::work_application(
+            "request.work.native-writer-waits.list",
+            WorkApplicationInvocationV1::ListAttempts(
+                tracedecay_contracts::WorkAttemptListRequestV1 {
+                    page_size: 10,
+                    cursor: None,
+                },
+            ),
+            UtcMicros(100),
+            Deadline::new(UtcMicros(1_000)).expect("deadline"),
+            CancellationContext::active("cancel.request.work.native-writer-waits.list")
+                .expect("cancellation"),
+        ),
+    ];
+    let service = fixture.service.clone();
+    let registry = Arc::clone(&fixture.registry);
+    let project_root = fixture.project.path().to_path_buf();
+
+    let writer_waits = tracedecay_runtime_core::native_runtime::native_thread_writer_waits();
+    let served = tracedecay_runtime_core::native_runtime::NativeTask::spawn(async move {
+        let mut outcomes = Vec::new();
+        for request in requests {
+            outcomes.push(
+                invoke_registered(
+                    service.clone(),
+                    Arc::clone(&registry),
+                    project_root.clone(),
                     request,
-                    UtcMicros(100),
-                    Deadline::new(UtcMicros(1_000)).expect("deadline"),
-                    CancellationContext::active(format!("cancel.{request_id}"))
-                        .expect("cancellation"),
-                ),
+                )
+                .await,
+            );
+        }
+        let binding = service
+            .work_fan_out_binding(
+                Some(&project_root),
+                &tracedecay_domain::WorkAttemptIdentityV1::new(
+                    TaskId::new("task.native-writer-waits").expect("task id"),
+                    tracedecay_domain::RunId::new("run.native-writer-waits").expect("run id"),
+                    tracedecay_domain::AttemptId::new("attempt.native-writer-waits")
+                        .expect("attempt id"),
+                )
+                .expect("attempt identity"),
             )
-            .await
-            .outcome
-    }
+            .await;
+        (outcomes, binding)
+    })
+    .expect("native worker task")
+    .wait()
+    .await;
+    let tracedecay_runtime_core::native_runtime::NativeTaskOutcome::Completed((outcomes, binding)) =
+        served
+    else {
+        panic!("the native worker must serve every request: {served:?}");
+    };
+    assert_eq!(
+        tracedecay_runtime_core::native_runtime::native_thread_writer_waits(),
+        writer_waits,
+        "Work and Workflow dispatch must not park a native worker on the writer"
+    );
+
+    let [register, get_run, list] = outcomes.as_slice() else {
+        panic!("three outcomes: {outcomes:?}");
+    };
+    assert!(
+        matches!(
+            register,
+            DaemonInvocationOutcome::WorkflowApplication { .. }
+        ),
+        "registration must reach the workflow effect journal: {register:?}"
+    );
+    assert!(
+        matches!(get_run, DaemonInvocationOutcome::ApplicationProblem { .. }),
+        "an unknown run must be read from storage and refused: {get_run:?}"
+    );
+    assert!(
+        matches!(list, DaemonInvocationOutcome::WorkApplication { .. }),
+        "the attempt list must be answered: {list:?}"
+    );
+    assert_eq!(binding, None, "no workflow owns an unknown attempt");
 }