refactor(dispatch): F-CONC audit batch (5 concurrency findings)

crates/pitboss-cli/src/control/server.rs

@@ -654,6 +654,14 @@ async fn serve_connection(
                     }
                 } else {
                     let reply = dispatch_op(&state, op).await;
+                    // F-CONC-11 #490: broadcasting op replies to every
+                    // subscriber (including the SPA's SSE bridge) is
+                    // intentional — see the PR-Q routing block above for
+                    // the design rationale and the `OpUnknownState` /
+                    // current-state-leak analysis. Audit re-validation
+                    // confirmed `current_state` is the same coarse enum
+                    // already published via `WorkerStatus`, so the
+                    // "subscriber visibility" surface is not a leak.
                     state
                         .root
                         .broadcast_control_event(EventEnvelope {

crates/pitboss-cli/src/dispatch/layer.rs

@@ -109,12 +109,22 @@ pub struct WorkerRegistry {
 }
 
 impl WorkerRegistry {
-    /// Construct an empty registry with a fresh 64-slot `done_tx` channel.
-    /// The capacity matches the pre-split `LayerState::new` default; the
-    /// channel is mostly there for `wait_for_worker` consumers and a
+    /// Construct an empty registry with a fresh 1024-slot `done_tx` channel.
+    /// The channel is mostly there for `wait_for_worker` consumers and a
     /// `Lagged` receiver re-syncs from the `states` map.
+    ///
+    /// **Capacity (F-CONC-12 #491):** sized to absorb a mass-cancel storm
+    /// across a fully-saturated depth-2 tree (root + N sub-leads, each
+    /// with their own worker pool) without the slowest `wait_for_*`
+    /// subscriber lagging into the recovery rescan. The `Lagged` →
+    /// `refresh_in_flight_from_maps` path in
+    /// `hierarchical.rs::await_workers_drained` remains correct and is
+    /// kept as belt-and-braces, not the expected fast path. The pre-F-CONC-12
+    /// capacity of 64 matched the historical single-layer default, but
+    /// post depth-2 every root subscriber sees every worker's completion
+    /// across the whole tree.
     pub fn new() -> Self {
-        let (done_tx, _) = broadcast::channel(64);
+        let (done_tx, _) = broadcast::channel(1024);
         Self {
             states: RwLock::new(HashMap::new()),
             cancels: RwLock::new(HashMap::new()),

crates/pitboss-cli/src/dispatch/signals.rs

@@ -68,6 +68,12 @@ fn send_group_signal(pid: u32, sig: libc::c_int, name: &'static str) -> Result<(
 ///   2nd SIGINT within window → terminate
 /// After the window, re-armed: a single later SIGINT is treated as a fresh first.
 ///
+/// **Drain-mode auto-escalation (opt-in, F-CONC-7 #497):** set the env
+/// var `PITBOSS_CTRLC_DRAIN_ESCALATE_SECS=<n>` (positive integer seconds)
+/// to make the watcher escalate to terminate after `n` seconds of drain
+/// without a second Ctrl-C. Default behaviour (env unset / unparseable /
+/// zero) is to remain in drain mode indefinitely.
+///
 /// **Idempotent.** Calling more than once in the same process is a no-op
 /// after the first install — pitboss has two callsites today (flat-mode
 /// and hierarchical entry points) and a future third callsite would
@@ -102,14 +108,61 @@ pub fn install_ctrl_c_watcher(cancel: CancelToken) {
                     return;
                 }
                 _ => {
-                    tracing::info!("drain window expired; continuing in drain mode");
-                    // Loop again: if another Ctrl-C arrives later, start a new window.
+                    // F-CONC-7 #497: optionally auto-escalate to terminate
+                    // after a configurable drain-mode timeout, so operators
+                    // who walk away after a single Ctrl-C don't leave the
+                    // run hanging behind a stuck worker. Opt-in via env var
+                    // to preserve the "drain forever" default behaviour.
+                    if let Some(escalate_after) = drain_escalate_timeout() {
+                        tracing::warn!(
+                            "drain window expired; auto-escalating to terminate in {:?} \
+                             (PITBOSS_CTRLC_DRAIN_ESCALATE_SECS) unless another Ctrl-C arrives",
+                            escalate_after,
+                        );
+                        tokio::select! {
+                            () = tokio::time::sleep(escalate_after) => {
+                                cancel.terminate_with_reason(
+                                    pitboss_core::store::TerminateReason::OperatorCtrlC,
+                                );
+                                tracing::warn!(
+                                    "drain-mode timeout reached — terminating subprocesses",
+                                );
+                                return;
+                            }
+                            res = tokio::signal::ctrl_c() => {
+                                if res.is_ok() {
+                                    cancel.terminate_with_reason(
+                                        pitboss_core::store::TerminateReason::OperatorCtrlC,
+                                    );
+                                    tracing::warn!(
+                                        "received Ctrl-C during drain-mode timeout — terminating",
+                                    );
+                                }
+                                return;
+                            }
+                        }
+                    } else {
+                        tracing::info!("drain window expired; continuing in drain mode");
+                        // Loop again: if another Ctrl-C arrives later, start a new window.
+                    }
                 }
             }
         }
     });
 }
 
+/// Read the optional drain-mode auto-escalation timeout from
+/// `PITBOSS_CTRLC_DRAIN_ESCALATE_SECS`. Returns `None` if the env var is
+/// unset, empty, or unparseable as a positive integer — keeping the
+/// pre-F-CONC-7 "drain forever" semantics as the default.
+fn drain_escalate_timeout() -> Option<Duration> {
+    std::env::var("PITBOSS_CTRLC_DRAIN_ESCALATE_SECS")
+        .ok()
+        .and_then(|s| s.trim().parse::<u64>().ok())
+        .filter(|n| *n > 0)
+        .map(Duration::from_secs)
+}
+
 // ── Kill-with-reason cascade (Task 4.5) ─────────────────────────────────────
 
 /// Cancel any actor in the tree (worker or sub-lead) and optionally deliver
@@ -172,6 +225,15 @@ pub async fn cancel_actor_with_reason(
 /// under a held `state.subleads` read lock — O(N) per cancel call,
 /// blocking new sublead registration. The `worker_layer_index` lookup
 /// mirrors `resolve_layer_for_caller`'s routing for KV ops (#150 audit).
+///
+/// **Lock-acquisition order (project invariant):** `state.subleads` →
+/// `state.worker_layer_index` → `sub_layer.workers.cancels`. Every
+/// callsite that holds any pair of these locks acquires them in this
+/// order. Inverting it under a `subleads.write()` would deadlock; today
+/// no such path exists because every `subleads.write()` callsite uses a
+/// single-statement guard (`register_sublead`, `reconcile_terminated_sublead`).
+/// Mirror of the convention documented in `hierarchical.rs::await_workers_drained`.
+/// (F-CONC-4 #494)
 async fn cancel_and_find_parent(
     state: &Arc<DispatchState>,
     target: &str,
@@ -297,6 +359,7 @@ pub fn install_cascade_cancel_watcher(state: Arc<DispatchState>) {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use serial_test::serial;
 
     #[test]
     fn freeze_rejects_pid_zero() {
@@ -310,6 +373,45 @@ mod tests {
         assert!(err.to_string().contains("pid is 0"));
     }
 
+    /// F-CONC-7 #497: the opt-in drain-mode auto-escalation timeout
+    /// returns `None` when the env var is unset (preserving the
+    /// pre-fix "drain forever" default) and `Some(_)` only for a
+    /// positive integer.
+    #[test]
+    #[serial(env)]
+    fn drain_escalate_timeout_is_opt_in() {
+        // SAFETY: `#[serial(env)]` serializes against other env-touching
+        // tests so we don't race readers in tracing/reqwest/locale init.
+        let key = "PITBOSS_CTRLC_DRAIN_ESCALATE_SECS";
+        std::env::remove_var(key);
+        assert_eq!(drain_escalate_timeout(), None, "unset → no escalation");
+
+        std::env::set_var(key, "");
+        assert_eq!(drain_escalate_timeout(), None, "empty → no escalation");
+
+        std::env::set_var(key, "0");
+        assert_eq!(drain_escalate_timeout(), None, "zero → no escalation");
+
+        std::env::set_var(key, "not-a-number");
+        assert_eq!(
+            drain_escalate_timeout(),
+            None,
+            "unparseable → no escalation"
+        );
+
+        std::env::set_var(key, "300");
+        assert_eq!(drain_escalate_timeout(), Some(Duration::from_secs(300)));
+
+        std::env::set_var(key, "  60  ");
+        assert_eq!(
+            drain_escalate_timeout(),
+            Some(Duration::from_secs(60)),
+            "surrounding whitespace tolerated"
+        );
+
+        std::env::remove_var(key);
+    }
+
     /// End-to-end: spawn a sleeping child, SIGSTOP it, confirm
     /// /proc reports stopped state, SIGCONT, confirm runnable,
     /// then clean up. Linux-only because /proc isn't available on

crates/pitboss-cli/src/dispatch/state.rs

@@ -566,6 +566,32 @@ impl DispatchState {
     /// `dispatch/sublead.rs:340-383` so the watcher install + eager
     /// cascade can never drift out of order.  Pinned by the integration
     /// tests in `tests/cancel_cascade_flows.rs`.
+    ///
+    /// **Step order is load-bearing — do not reorder (F-CONC-8 #498):**
+    ///
+    /// - `insert` MUST happen before `cascade_to`. The cascade-watcher
+    ///   installed by `install_cascade_cancel_watcher` walks
+    ///   `cascade_to_subleads`, which reads `self.subleads`; if the
+    ///   sublead were not yet inserted, a concurrent root-cancel could
+    ///   miss this sub-layer and the eager `cascade_to` at step 3 would
+    ///   be the only signal that ever reaches it.
+    /// - `install_sublead_cancel_watcher` MUST happen before the eager
+    ///   `cascade_to`. The per-sub-layer watcher is fire-once on the
+    ///   sub-layer's own `CancelToken`; if `cascade_to` runs first and
+    ///   already toggles drain/terminate, the watcher started afterwards
+    ///   still observes the canceled state and fans out to workers as
+    ///   intended (the watcher's `select!` over `await_drain`/`await_terminate`
+    ///   resolves immediately for an already-canceled token). The
+    ///   ordering exists so the watcher is *registered* on the canonical
+    ///   pre-cancel state for the (more common) non-canceled path.
+    ///
+    /// Between steps 1 and 3 there is a benign window where a concurrent
+    /// root cancellation's `cascade_to_subleads` walk could fire and
+    /// no-op (root not yet canceled when walked) — the eager `cascade_to`
+    /// at step 3 closes this window. The sublead subprocess is not
+    /// launched until after `register_sublead` returns
+    /// (`dispatch/sublead.rs::spawn_sublead_session`), so no worker
+    /// observes the transient state.
     pub async fn register_sublead(&self, sublead_id: String, sub_layer: Arc<LayerState>) {
         self.subleads
             .write()