fix(interpreter): admit concurrent pipeline forks

crates/bashkit/docs/threat-model.md

@@ -130,7 +130,7 @@ stdout capture limit remains downstream of pipelines and redirects.
 | sleep bypasses timeout (TM-DOS-057) | `sleep N` ignores `ExecutionLimits::timeout` | Host-backed timeout; non-JS wasm blocking sleep is clamped to the execution deadline | **MITIGATED** |
 | Interactive terminal flooding (TM-DOS-119) | Unbounded type-ahead, undrained output, or a `vi` buffer grown by pastes | Caps on unread input (1 MiB), retained output (4 MiB), scrollback (1000 rows), transcript (1 MiB), line length, `vi` buffer (8 MiB) and undo history; `:w` goes through VFS limits | MITIGATED |
 | Input wait vs. timeout (TM-DOS-120) | Abusing the keyboard-wait exclusion to run work past the deadline | Only time blocked reading the terminal is excluded; CPU work and `sleep` still count | MITIGATED |
-| Concurrent pipeline amplification (TM-DOS-124) | An endless loop piped into `head`, a producer whose reader quit, or recursive pipelines `f() { f \| f; }` | 4 KiB pipes with backpressure; a writer whose reader is gone ends with 141 (SIGPIPE); streaming nesting counts toward `max_subshell_depth` and falls back to sequential stages past 4 levels; stages share the session budget, timeout and cancellation | MITIGATED |
+| Concurrent pipeline amplification (TM-DOS-124) | An endless loop piped into `head`, a producer whose reader quit, or recursive pipelines `f() { f \| f; }` | 4 KiB pipes with backpressure; a writer whose reader is gone ends with 141 (SIGPIPE); streaming nesting counts toward `max_subshell_depth` and falls back to sequential stages past 4 levels; stages share the session budget, timeout and cancellation; producers atomically claim the shared job cap before construction, borrow the AST and lease owned fork storage/future layouts before allocation | MITIGATED |
 | make amplification (TM-DOS-126) | Exponential recursive variables, self-including makefiles, deep `$(MAKE)` recursion, huge graphs | Expansion depth 200 and 4 MiB per expansion; substitution checks its exact size before construction, with shared capacity leases retained by expanded values and work/deadline checkpoints while matching and copying; include depth 16; target, query and glob caps per run; `$(MAKE)` stops at level 4; recipes run under the session budget and timeout | MITIGATED |
 | Random generator amplification (TM-DOS-123) | `openssl rand` asked for gigabytes | Requests capped at 1 MiB; fixed-size `uuidgen`/`$SRANDOM` | MITIGATED |
 | Child-shell recursion (TM-DOS-125) | Script that re-runs itself via `sh` | Child shells count against function depth, max 8 nested | MITIGATED |

crates/bashkit/src/fs/proc_sub_fds.rs

@@ -90,6 +90,17 @@ impl ProcSubFs {
         })
     }
 
+    /// Fork metadata only: substitution byte payloads remain Arc-shared.
+    pub(crate) fn fork_storage_bytes(&self) -> usize {
+        std::mem::size_of::<Self>()
+            .saturating_add(2 * std::mem::size_of::<usize>())
+            .saturating_add(
+                self.table()
+                    .len()
+                    .saturating_mul(std::mem::size_of::<(i32, ProcSubData)>()),
+            )
+    }
+
     /// Same filesystem, a copy of the open fds (a forked job).
     pub(crate) fn fork(&self) -> Arc<Self> {
         Arc::new(Self {

crates/bashkit/src/interpreter/completion.rs

@@ -250,6 +250,20 @@ pub(crate) struct CompletionState {
     pub(crate) bindings: super::readline_bind::Bindings,
 }
 
+impl CompletionState {
+    pub(super) fn fork_storage(
+        &self,
+        storage: &mut super::pipeline_admission::ForkStorage<'_>,
+    ) -> Result<()> {
+        storage.slice::<(String, CompSpec)>(self.specs.len())?;
+        for (name, spec) in &self.specs {
+            storage.add(name.len())?;
+            storage.add(spec.text_bytes())?;
+        }
+        self.bindings.fork_storage(storage)
+    }
+}
+
 /// Parsed `complete`/`compgen` options.
 #[derive(Default)]
 struct CompArgs {

crates/bashkit/src/interpreter/jobs.rs

@@ -341,13 +341,13 @@ pub struct JobControl {
     live: Arc<AtomicUsize>,
 }
 
-/// One live job's claim on the session-wide job cap; released on drop
-/// (finish or abort).
-pub struct JobSlot(Arc<AtomicUsize>);
+/// A claim on the session-wide job cap; released on drop (finish or abort).
+/// Pipelines retain a group claim until every producer owner is dropped.
+pub struct JobSlot(Arc<AtomicUsize>, usize);
 
 impl Drop for JobSlot {
     fn drop(&mut self) {
-        self.0.fetch_sub(1, AtomicOrdering::AcqRel);
+        self.0.fetch_sub(self.1, AtomicOrdering::AcqRel);
     }
 }
 
@@ -371,12 +371,19 @@ impl JobControl {
 
     /// Claim a slot under `max` live jobs session-wide (TM-DOS-122).
     pub fn try_claim(&self, max: usize) -> Option<JobSlot> {
+        self.try_claim_many(max, 1)
+    }
+
+    /// Admit every producer before constructing a concurrent pipeline.
+    /// One drop-owned claim keeps its retained forks counted until completion
+    /// or cancellation; nested pipelines and background jobs share this cap.
+    pub(crate) fn try_claim_many(&self, max: usize, count: usize) -> Option<JobSlot> {
         self.live
             .fetch_update(AtomicOrdering::AcqRel, AtomicOrdering::Acquire, |n| {
-                (n < max).then_some(n + 1)
+                n.checked_add(count).filter(|next| *next <= max)
             })
             .ok()
-            .map(|_| JobSlot(Arc::clone(&self.live)))
+            .map(|_| JobSlot(Arc::clone(&self.live), count))
     }
 
     /// Lock the table. Never held across an `.await`.
@@ -462,6 +469,22 @@ pub fn new_shared_job_table() -> SharedJobTable {
 
 #[cfg(test)]
 mod tests {
+    #[test]
+    fn pipeline_admission_claims_atomically_across_forks_and_refunds_on_drop() {
+        let jobs = super::new_shared_job_table();
+        let child = jobs.fork();
+        let occupied = jobs.try_claim(2).unwrap();
+        assert!(child.try_claim_many(2, 2).is_none());
+        let remaining = child.try_claim(2).unwrap();
+        assert!(jobs.try_claim(2).is_none());
+        drop(occupied);
+        drop(remaining);
+        let group = child.try_claim_many(2, 2).unwrap();
+        assert!(jobs.try_claim(2).is_none());
+        drop(group);
+        assert!(jobs.try_claim_many(2, 2).is_some());
+    }
+
     use super::*;
     use std::time::Duration;