feat: deepen async concurrency seams

CONTEXT.md

@@ -70,7 +70,10 @@ _Avoid_: action, command, operation.
 
 **Runner**:
 The component that orders tasks, applies skip rules, and drives each task's
-command entries to completion, emitting events as it goes (`TaskRunner`).
+command entries to completion, emitting events as it goes (`TaskRunner`). Holds
+a `CancellationToken` shared with nested Sub-config Runners; cancel aborts
+in-flight Tokio tasks (OS subprocess teardown is a follow-up). Uses
+`tokio::task::JoinSet` for parallel layer and parallel Command admission.
 _Avoid_: engine (too broad), executor (means something narrower here).
 
 **Command executor**:
@@ -99,10 +102,10 @@ justifies a real plugin seam — see ADR-0006).
 
 **Task event**:
 A message describing execution progress (`TaskEvent`) — lifecycle and
-per-line/per-file output alike. `task_name` is an `Arc<str>` interned once per
-Task so per-line output clones a refcount, not a new allocation. Emitted
-through the **Task event sink**; the TUI and plain logger consume events from
-the channel-backed adapter.
+per-line/per-file output alike. `task_name` and `command_desc` are `Arc<str>`
+interned once per Task / Command entry so per-line output clones refcounts,
+not new allocations. Emitted through the **Task event sink**; the TUI and
+plain logger consume events from the channel-backed adapter.
 _Avoid_: message, log, signal.
 
 **History**:
@@ -128,8 +131,8 @@ Declarative gate on whether a Task runs — `only_if` (all must pass) and
 `skip_if` (any triggers skip). Each field accepts a path string, a list of
 path strings (backward compatible), or rich objects: `{ path }`, `{ env }`,
 `{ command }`, `{ mode }`. Evaluated in `engine::conditions` together with OS
-filter and install History skip; the Runner calls `evaluate_skip` before
-spawning work.
+filter and install History skip; the Runner calls async `evaluate_skip` before
+spawning work. Shell `{ command }` conditions run via **Host blocking**.
 _Avoid_: when clause, if guard, predicate.
 
 **Task graph**:
@@ -142,24 +145,30 @@ _Avoid_: dependency resolver, DAG, scheduler (do not reuse "scheduler" for
 concurrency — see **Concurrency gate**).
 
 **Task event sink**:
-The seam for emitting **Task events**. Adapters: **ChannelSink** (mpsc →
-TUI/plain) and **NullSink** (benches). The Runner and `CommandContext` both
-emit through this interface — not a raw sender. Subprocess line readers may
-coalesce stdout/stderr into a **`CommandOutputBatch`** Task event before
-emit (single-line `CommandOutput` remains for sparse progress); batching is
-an engine policy, not a third sink adapter.
+The seam for emitting **Task events**. Adapters: **ChannelSink** (bounded
+mpsc, capacity 8192, backpressure via `block_in_place` + `blocking_send` on
+the multi-thread runtime → TUI/plain) and **NullSink**
+(benches). The Runner and `CommandContext` both emit through this interface —
+not a raw sender. Subprocess line readers may coalesce stdout/stderr into a
+**`CommandOutputBatch`** Task event before emit (single-line `CommandOutput`
+remains for sparse progress); batching is an engine policy, not a third sink
+adapter.
 _Avoid_: logger, event bus, observer.
 
 **Concurrency gate**:
 The Runner's global cap on in-flight leaf Command executor work, driven by
 `num_threads` (default: physical CPUs − 1). Permits are per Command entry;
 `machine_setup` does not hold a permit so nested Sub-configs can share the
-gate. Sync File ops work runs via `spawn_blocking` so it does not block Tokio
-workers. Owns a shared Rayon FS apply pool (same size as the permit limit),
-created lazily on first tree-apply use. Also owns **Exclusive lanes** (package
-managers) and a separate **tree-apply** K=1 semaphore so only one
-`pool.install` runs at a time (tree-apply does not reuse Exclusive lanes).
-Does **not** order Tasks by dependency — that remains the **Task graph**.
+gate. **`admit`** acquires an Exclusive lane (if any), an optional per-Task
+sub-quota for `parallel: true` Tasks (half the gate limit, ceil, minimum 1),
+then a work permit. Sync File ops work runs via `spawn_blocking` so it does
+not block Tokio workers; sync config load, shell conditions, and clone fs ops
+use **Host blocking** (`engine::host_blocking`). Owns a shared Rayon FS apply
+pool (same size as the permit limit), created lazily on first tree-apply use.
+Also owns **Exclusive lanes** (package managers) and a separate **tree-apply**
+K=1 semaphore so only one `pool.install` runs at a time (tree-apply does not
+reuse Exclusive lanes). Does **not** order Tasks by dependency — that remains
+the **Task graph**.
 _Avoid_: scheduler (do not call the FS pool the "scheduler").
 
 **Exclusive lane**:
@@ -216,6 +225,14 @@ thin per-kind Command executors that supply a per-file strategy. Does not move
 privilege planning into File ops (ADR-0002).
 _Avoid_: unified tree command, generic file command.
 
+**Host blocking**:
+The seam for running synchronous host work off Tokio worker threads via
+`tokio::task::spawn_blocking` — config load in Sub-config, shell condition
+evaluation, clone directory create/remove, and similar blocking I/O. Tree-op
+already uses its own `spawn_blocking` path; main-thread pre-runtime config load
+in `main.rs` stays synchronous.
+_Avoid_: blocking the async runtime, inline fs in async executors.
+
 **Command bench**:
 The measurement module for Command executor / Tree materialization / Runner
 wall-clock speed — Criterion microbenches plus thin Runner smoke over

docs/adr/0003-concurrency-gate-global.md

@@ -11,17 +11,26 @@ Task — so sequential and `parallel: true` Tasks share the same rule. The
 Sub-config Runners share the parent's gate and their leaf commands acquire
 normally. That avoids a deadlock when `num_threads: 1` and a parent Task would
 otherwise hold the only permit across nested work. Sync File ops (`copy` /
-`symlink`) run via `spawn_blocking`. We accepted that a fat `parallel: true`
-Task can starve siblings; per-Task sub-quotas are a future fix. The gate also
+`symlink`) run via `spawn_blocking`. Per-Task sub-quotas are implemented for
+`parallel: true` Tasks: each such Task gets a semaphore of
+`max(1, gate_limit.div_ceil(2))` so parallel commands within one Task cannot
+consume the entire gate. Sequential Tasks pass no sub-quota. The gate also
 owns a shared Rayon pool of the same size for in-tree DirectFs file apply
 (ADR-0004) so sibling commands do not each spawn a private worker set. The
 pool is created lazily on first tree-apply use, not when the gate is built.
 
-## Tree-apply admission (accepted 2026-09-05 — not yet implemented)
+## Tree-apply admission (accepted 2026-09-05 — implemented)
 
 A second semaphore on the Concurrency gate admits **at most one** concurrent
 tree `pool.install` (K=1). Leaf Command permits and Exclusive lanes are
 unchanged. This prevents sibling tree Command entries from oversubscribing the
 shared Rayon pool. Splitting pool width or reusing package-manager Exclusive
 lanes for trees was rejected. K>1 remains a possible later knob; default is
 exclusive tree-apply.
+
+## Per-Task sub-quotas (implemented 2026-09-05)
+
+When a Task has `parallel: true`, the Runner creates a per-Task semaphore of
+`max(1, gate_limit.div_ceil(2))`. Each parallel Command entry acquires through
+`ConcurrencyGate::admit` with that quota before the global work permit.
+Sequential Tasks omit the sub-quota.

docs/adr/0005-task-event-sink.md

@@ -7,11 +7,19 @@ wall-clock from event clone/channel cost and keeps one seam for lifecycle and
 CommandOutput alike. Fan-out to multiple simultaneous consumers was rejected as
 overbuilt for a single UI consumer.
 
-## Coalesced subprocess output (accepted 2026-09-05 — not yet implemented)
+## Bounded ChannelSink (implemented 2026-09-05)
+
+**ChannelSink** uses a bounded `mpsc` channel with capacity **8192**.
+`emit` applies backpressure (`try_send`, then on `Full`: `block_in_place` +
+`blocking_send` on the multi-thread runtime, plain `blocking_send` outside a
+runtime). Direct `blocking_send` on a Tokio worker panics — do not restore
+that path. Closed receivers are ignored. **NullSink** is unchanged.
+
+## Coalesced subprocess output (implemented 2026-09-05)
 
 Subprocess line reading may buffer stdout/stderr and emit a
 **`CommandOutputBatch`** Task event (multiple lines per flush) instead of one
 `CommandOutput` per line. Single-line `CommandOutput` stays for sparse progress
 (e.g. FileProgress). Coalescing lives in the engine line reader, not in a new
-sink adapter. TUI/plain must expand batches. Bench the ChannelSink path (not
-only NullSink) when landing.
+sink adapter. TUI/plain expand batches. Bench the ChannelSink path (not only
+NullSink) when landing.

src/engine/commands/catalog.rs

@@ -48,13 +48,13 @@ pub fn create_executor(entry: &CommandEntry) -> Executor {
 }
 
 /// Display label for a Command entry (used by `Display` and Task events).
-pub fn description(entry: &CommandEntry) -> String {
+pub fn description(entry: &CommandEntry) -> std::sync::Arc<str> {
     match entry {
-        CommandEntry::Copy(args) => args.to_string(),
-        CommandEntry::Symlink(args) => args.to_string(),
-        CommandEntry::Clone(args) => args.to_string(),
-        CommandEntry::Run(args) => args.to_string(),
-        CommandEntry::MachineSetup(args) => args.to_string(),
+        CommandEntry::Copy(args) => std::sync::Arc::from(args.to_string()),
+        CommandEntry::Symlink(args) => std::sync::Arc::from(args.to_string()),
+        CommandEntry::Clone(args) => std::sync::Arc::from(args.to_string()),
+        CommandEntry::Run(args) => std::sync::Arc::from(args.to_string()),
+        CommandEntry::MachineSetup(args) => std::sync::Arc::from(args.to_string()),
     }
 }
 

src/engine/commands/clone.rs

@@ -69,7 +69,8 @@ impl CloneCommand {
         ));
 
         if let Some(parent) = target.parent() {
-            std::fs::create_dir_all(parent)?;
+            let parent = parent.to_path_buf();
+            crate::engine::host_blocking::run(move || std::fs::create_dir_all(&parent)).await??;
         }
 
         run_git_command(
@@ -95,7 +96,8 @@ impl CloneCommand {
 
         if target.exists() {
             ctx.log_progress(format!("remove {}", display_path(&target)));
-            std::fs::remove_dir_all(&target)?;
+            let target = target.to_path_buf();
+            crate::engine::host_blocking::run(move || std::fs::remove_dir_all(&target)).await??;
         }
 
         Ok(())

src/engine/commands/copy.rs

@@ -206,6 +206,7 @@ mod tests {
             default_shell: crate::config::types::Shell::Bash,
             task_name: Arc::<str>::from("t"),
             depth: 0,
+            cancel: tokio_util::sync::CancellationToken::new(),
         }
     }
 

src/engine/commands/setup.rs

@@ -49,7 +49,10 @@ async fn run_sub_config(args: &MachineSetupArgs, ctx: &CommandContext) -> Result
 
     ctx.log_info(format!("Loading sub-config: {config_str}"));
 
-    let config = crate::config::load_config(&config_str)?;
+    let config_str = config_str.into_owned();
+    let sub_config_dir = crate::config::resolve_config_dir(&config_str, &ctx.config_dir);
+    let config = crate::engine::host_blocking::run(move || crate::config::load_config(&config_str))
+        .await??;
 
     let run_set = match (&args.task, args.with_deps) {
         (Some(task_name), true) => Some(crate::config::selection::expand_for_mode(
@@ -63,12 +66,11 @@ async fn run_sub_config(args: &MachineSetupArgs, ctx: &CommandContext) -> Result
 
     // Resolve the sub-config's directory for its own relative paths. URLs
     // and unresolvable paths fall back to the parent's config_dir.
-    let sub_config_dir = crate::config::resolve_config_dir(&config_str, &ctx.config_dir);
-
     let runner = crate::engine::runner::TaskRunner::new(config, ctx.mode, Arc::clone(&ctx.events))
         .with_gate(Arc::clone(&ctx.gate))
         .with_config_dir(sub_config_dir)
-        .with_depth(ctx.depth + 1);
+        .with_depth(ctx.depth + 1)
+        .with_cancel(ctx.cancel.clone());
 
     match (run_set, &args.task) {
         (Some(tasks), _) => runner.run_tasks(&tasks, args.force).await,

src/engine/commands/symlink.rs

@@ -164,7 +164,7 @@ mod tests {
         dir: &Path,
     ) -> (
         CommandContext,
-        tokio::sync::mpsc::UnboundedReceiver<crate::engine::event::TaskEvent>,
+        tokio::sync::mpsc::Receiver<crate::engine::event::TaskEvent>,
     ) {
         let (events, rx) = crate::engine::sink::ChannelSink::channel();
         let ctx = CommandContext {
@@ -178,6 +178,7 @@ mod tests {
             default_shell: crate::config::types::Shell::Bash,
             task_name: std::sync::Arc::<str>::from("t"),
             depth: 0,
+            cancel: tokio_util::sync::CancellationToken::new(),
         };
         (ctx, rx)
     }

src/engine/concurrency.rs

@@ -107,6 +107,62 @@ impl ConcurrencyGate {
             .expect("ConcurrencyGate semaphore is never closed")
     }
 
+    /// Per-Task sub-quota for `parallel: true` Tasks — half the gate limit (ceil), at least 1.
+    pub fn task_quota_limit(&self) -> usize {
+        std::cmp::max(1, self.limit.div_ceil(2))
+    }
+
+    /// Lane first, then optional per-Task quota, then work permit (ADR-0010 + sub-quotas).
+    ///
+    /// Calls `on_lane_wait(lane)` only when the lane is already held (contended path).
+    pub async fn admit(
+        self: &Arc<Self>,
+        lane: Option<ExclusiveLane>,
+        occupies_slot: bool,
+        task_quota: Option<&Arc<Semaphore>>,
+        on_lane_wait: impl FnOnce(ExclusiveLane),
+    ) -> AdmitPermits {
+        let lane_permit = if let Some(lane) = lane {
+            match self.try_acquire_lane(lane) {
+                Some(permit) => Some(permit),
+                None => {
+                    on_lane_wait(lane);
+                    Some(self.acquire_lane(lane).await)
+                }
+            }
+        } else {
+            None
+        };
+
+        let task_permit = if let Some(quota) = task_quota {
+            #[expect(
+                clippy::expect_used,
+                reason = "per-Task quota semaphore is never closed"
+            )]
+            Some(
+                quota
+                    .clone()
+                    .acquire_owned()
+                    .await
+                    .expect("per-Task quota semaphore is never closed"),
+            )
+        } else {
+            None
+        };
+
+        let work_permit = if occupies_slot {
+            Some(self.acquire().await)
+        } else {
+            None
+        };
+
+        AdmitPermits {
+            _lane: lane_permit,
+            _task: task_permit,
+            _work: work_permit,
+        }
+    }
+
     /// Try to take an Exclusive lane without waiting.
     pub fn try_acquire_lane(&self, lane: ExclusiveLane) -> Option<OwnedSemaphorePermit> {
         self.lanes[lane.index()].clone().try_acquire_owned().ok()
@@ -139,6 +195,13 @@ impl ConcurrencyGate {
     }
 }
 
+/// Permits held for one Command entry admission (lane, optional Task quota, work).
+pub struct AdmitPermits {
+    _lane: Option<OwnedSemaphorePermit>,
+    _task: Option<OwnedSemaphorePermit>,
+    _work: Option<OwnedSemaphorePermit>,
+}
+
 fn build_fs_pool(limit: usize) -> ThreadPool {
     #[expect(
         clippy::expect_used,
@@ -168,6 +231,39 @@ pub fn resolve_limit(num_threads: Option<usize>) -> usize {
 mod tests {
     use super::*;
 
+    #[test]
+    fn task_quota_limit_is_half_ceil_min_one() {
+        let gate = ConcurrencyGate::from_num_threads(Some(2));
+        assert_eq!(gate.task_quota_limit(), 1);
+        let gate = ConcurrencyGate::from_num_threads(Some(5));
+        assert_eq!(gate.task_quota_limit(), 3);
+        let gate = ConcurrencyGate::from_num_threads(Some(1));
+        assert_eq!(gate.task_quota_limit(), 1);
+    }
+
+    #[tokio::test]
+    async fn task_quota_serializes_parallel_commands_when_one() {
+        let gate = Arc::new(ConcurrencyGate::from_num_threads(Some(2)));
+        assert_eq!(gate.task_quota_limit(), 1);
+        let quota = Arc::new(Semaphore::new(gate.task_quota_limit()));
+        let gate2 = Arc::clone(&gate);
+        let quota2 = Arc::clone(&quota);
+        let first = tokio::spawn(async move {
+            let _p = gate2.admit(None, true, Some(&quota2), |_| {}).await;
+            tokio::time::sleep(std::time::Duration::from_millis(50)).await;
+        });
+        tokio::task::yield_now().await;
+        let gate3 = Arc::clone(&gate);
+        let quota3 = Arc::clone(&quota);
+        let second = tokio::spawn(async move {
+            gate3.admit(None, true, Some(&quota3), |_| {}).await;
+        });
+        tokio::task::yield_now().await;
+        assert!(!second.is_finished());
+        first.await.unwrap();
+        second.await.unwrap();
+    }
+
     #[test]
     fn resolve_limit_respects_explicit() {
         assert_eq!(resolve_limit(Some(4)), 4);