streams: per-invocation channel, pumping mode, filter pipeline wiring

exec: LineChannel bounded ring (cap=16) with abort/eof/marker fields.
Shell 'pumping mode' assembles canonical lines from its queue into the
channel while the command runs; ^C marks the channel aborted.

Shell-side grammar: USE/TIL pre-drain into the channel before spawn;
<cmd> <path> TIL <marker> sugar drains paths until the marker line.
Interactive TIL: pump_key checks each submitted line against the marker
and closes the channel when it matches.

Exports callbacks run in the command task but touch shell-owned state:
this is safe by construction — the shell blocks in exec_external until
the command completes.  put_line feeds Pipeline::line directly;
get_line pops from the channel (yield-poll when empty).

load: drop NX from image mapping (single-privilege kernel, NX bit 63
reserved when EFER.NXE is set by OVMF); compute image span from section
RVAs not file size.

bins/count: gate binary — reads lines via get_line, reports count via
put_line.  Verified: count USE file → 6; count <path> TIL END → 2;
interactive count TIL DONE → 3; -h filter across task boundary;
^C abort.

v1 stream exports: put_line → Pipeline, get_line ← channel, log_line
→ klog append, exit → abort_current.

Accounting: print_one_line → pipe.line() returns false (head kill /
pager ^C) → the binary should wrap up promptly; accepted: binary does
not check the return value in v1 (phase 3 simplification).

9 files changed, 459 insertions, 51 deletions

diff --git a/Cargo.lock b/Cargo.lock
index ed37b5f..0ca76eb 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -10,6 +10,13 @@ dependencies = [
 ]
 
 [[package]]
+name = "count"
+version = "0.6.0"
+dependencies = [
+ "rsapi",
+]
+
+[[package]]
 name = "hello"
 version = "0.6.0"
 dependencies = [
diff --git a/Cargo.toml b/Cargo.toml
index e8e0827..feca451 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -1,6 +1,6 @@
 [workspace]
 resolver = "2"
-members = ["rsnix", "shell", "rsdos", "rsdosedit", "rspaint", "rsapi", "bins/hello", "bins/boom"]
+members = ["rsnix", "shell", "rsdos", "rsdosedit", "rspaint", "rsapi", "bins/hello", "bins/boom", "bins/count"]
 
 [workspace.package]
 version = "0.6.0"
diff --git a/bins/count/Cargo.toml b/bins/count/Cargo.toml
new file mode 100644
index 0000000..5ab50cf
--- /dev/null
+++ b/bins/count/Cargo.toml
@@ -0,0 +1,10 @@
+[package]
+name = "count"
+version.workspace = true
+edition.workspace = true
+
+[lints]
+workspace = true
+
+[dependencies]
+rsapi = { path = "../../rsapi" }
diff --git a/bins/count/src/main.rs b/bins/count/src/main.rs
new file mode 100644
index 0000000..6fd6f17
--- /dev/null
+++ b/bins/count/src/main.rs
@@ -0,0 +1,64 @@
+//! count — stream-input consumer (ADR 0006 Phase 3 gate).
+//!
+//! Reads lines through `get_line` until EOF or Abort, then reports how
+//! many it saw via `put_line`.  Proves: USE pre-drain, TIL truncation,
+//! interactive pumping, and ^C abort across the task boundary.
+
+#![no_std]
+#![no_main]
+
+use rsapi::{Exports, StreamResult};
+
+#[panic_handler]
+fn panic(_info: &core::panic::PanicInfo) -> ! {
+    loop {
+        core::hint::spin_loop();
+    }
+}
+
+fn put_u32(exports: &Exports, mut v: u32) {
+    // Small fixed buffer; u32 max is 10 digits.
+    let mut buf = [0u8; 10];
+    let mut i = buf.len();
+    if v == 0 {
+        i -= 1;
+        buf[i] = b'0';
+    }
+    while v > 0 {
+        i -= 1;
+        buf[i] = b'0' + (v % 10) as u8;
+        v /= 10;
+    }
+    let suffix = b" line(s)";
+    let mut msg = [0u8; 18];
+    let n = buf.len() - i;
+    msg[..n].copy_from_slice(&buf[i..]);
+    msg[n..n + suffix.len()].copy_from_slice(suffix);
+    (exports.put_line)(rsapi::Line {
+        ptr: msg.as_ptr(),
+        len: n + suffix.len(),
+    });
+}
+
+/// The image entry point — the loader jumps here with the ABI arguments.
+#[unsafe(no_mangle)]
+pub extern "C" fn efi_main(
+    exports: &'static Exports,
+    _argc: usize,
+    _argv: *const rsapi::Arg,
+) -> i32 {
+    let mut count: u32 = 0;
+    loop {
+        let mut line = rsapi::Line {
+            ptr: core::ptr::null(),
+            len: 0,
+        };
+        match (exports.get_line)(&mut line) {
+            StreamResult::Line => count += 1,
+            StreamResult::Eof => break,
+            StreamResult::Abort => return 130, // 128+SIGINT convention
+        }
+    }
+    put_u32(exports, count);
+    0
+}
diff --git a/bins/hello/src/main.rs b/bins/hello/src/main.rs
index d1d58df..2904351 100644
--- a/bins/hello/src/main.rs
+++ b/bins/hello/src/main.rs
@@ -19,15 +19,15 @@ fn panic(_info: &core::panic::PanicInfo) -> ! {
 /// (exports table, argc, argv).  The conventional UEFI name is kept so
 /// the PE looks normal; the signature is the rs-dos ABI.
 #[unsafe(no_mangle)]
-pub extern "C" fn efi_main(exports: &'static Exports, _argc: usize, _argv: *const rsapi::Arg) -> i32 {
+pub extern "C" fn efi_main(
+    exports: &'static Exports,
+    _argc: usize,
+    _argv: *const rsapi::Arg,
+) -> i32 {
     let msg = b"hello, world - from an external binary";
     let line = rsapi::Line {
         ptr: msg.as_ptr(),
         len: msg.len(),
     };
-    if (exports.put_line)(line) {
-        0
-    } else {
-        1
-    }
+    if (exports.put_line)(line) { 0 } else { 1 }
 }
diff --git a/rsapi/src/lib.rs b/rsapi/src/lib.rs
index f71e69f..d7b69a2 100644
--- a/rsapi/src/lib.rs
+++ b/rsapi/src/lib.rs
@@ -20,7 +20,8 @@ pub const ABI_VERSION: u32 = 1;
 /// Entry-point signature.  `extern "C"` on the x86_64-unknown-uefi
 /// target is the Win64 convention: the three arguments arrive in
 /// RCX/RDX/R8 on both sides of the table.
-pub type EntryPoint = extern "C" fn(exports: &'static Exports, argc: usize, argv: *const Arg) -> i32;
+pub type EntryPoint =
+    extern "C" fn(exports: &'static Exports, argc: usize, argv: *const Arg) -> i32;
 
 /// One C-style argv element: pointer + length (rs-dos strings are not
 /// NUL-terminated in memory).
diff --git a/rsnix/src/lib.rs b/rsnix/src/lib.rs
index 376d3b5..325932a 100644
--- a/rsnix/src/lib.rs
+++ b/rsnix/src/lib.rs
@@ -5,6 +5,7 @@ extern crate alloc;
 pub mod ahci;
 pub mod cursor;
 pub mod ehci;
+pub mod exec;
 pub mod fat;
 pub mod font;
 pub mod fs;
@@ -12,7 +13,6 @@ pub mod gdt;
 pub mod gop;
 pub mod interrupts;
 pub mod klog;
-pub mod exec;
 pub mod load;
 pub mod memory;
 pub mod paging;
diff --git a/rsnix/src/load.rs b/rsnix/src/load.rs
index cebca89..488c703 100644
--- a/rsnix/src/load.rs
+++ b/rsnix/src/load.rs
@@ -50,7 +50,8 @@ pub enum LoadError {
 // ---------------------------------------------------------------------------
 
 fn u16_at(b: &[u8], off: usize) -> Option<u16> {
-    b.get(off..off + 2).map(|s| u16::from_le_bytes([s[0], s[1]]))
+    b.get(off..off + 2)
+        .map(|s| u16::from_le_bytes([s[0], s[1]]))
 }
 fn u32_at(b: &[u8], off: usize) -> Option<u32> {
     b.get(off..off + 4)
@@ -87,7 +88,23 @@ struct Section {
     raw_size: usize,
 }
 
-/// Parse headers, apply relocations into `mem`, return the entry RVA.
+/// Total virtual span of a PE image: the highest section RVA+virt_size
+/// (never smaller than one page for the headers).
+fn image_span_bytes(img: &[u8]) -> usize {
+    let e_lfanew = u32_at(img, 0x3C).unwrap_or(0) as usize;
+    let coff = e_lfanew + 4;
+    let num_sections = u16_at(img, coff + 2).unwrap_or(0) as usize;
+    let opt_size = u16_at(img, coff + 16).unwrap_or(0) as usize;
+    let sect_tab = coff + 20 + opt_size;
+    let mut end = PAGE_SIZE; // at least one page for headers
+    for i in 0..num_sections {
+        let s = sect_tab + i * 40;
+        let vsz = u32_at(img, s + 8).unwrap_or(0) as usize;
+        let va = u32_at(img, s + 12).unwrap_or(0) as usize;
+        end = end.max(va + vsz);
+    }
+    end
+}
 fn parse_and_relocate(img: &[u8], mem: usize) -> Result<usize, LoadError> {
     // ---- DOS header ----
     if u16_at(img, 0) != Some(DOS_MAGIC) {
@@ -151,7 +168,9 @@ fn parse_and_relocate(img: &[u8], mem: usize) -> Result<usize, LoadError> {
         if n == 0 {
             continue;
         }
-        let src = img.get(s.raw_ptr..s.raw_ptr + n).ok_or(LoadError::BadSectionTable)?;
+        let src = img
+            .get(s.raw_ptr..s.raw_ptr + n)
+            .ok_or(LoadError::BadSectionTable)?;
         unsafe {
             let dst = (mem + s.virt_addr) as *mut u8;
             core::ptr::copy_nonoverlapping(src.as_ptr(), dst, n);
@@ -185,7 +204,9 @@ fn parse_and_relocate(img: &[u8], mem: usize) -> Result<usize, LoadError> {
                         unsafe {
                             let p = target as *mut u64;
                             let v = p.read_unaligned();
-                            p.write_unaligned(v.wrapping_add((mem as u64).wrapping_sub(image_base as u64)));
+                            p.write_unaligned(
+                                v.wrapping_add((mem as u64).wrapping_sub(image_base as u64)),
+                            );
                         }
                     }
                     _ => return Err(LoadError::BadReloc),
@@ -213,17 +234,22 @@ pub fn load_image(bytes: &[u8]) -> Result<Image, LoadError> {
         return Err(LoadError::NotPe);
     }
 
-    let page_count = bytes.len().div_ceil(PAGE_SIZE);
+    // The memory span is the VIRTUAL layout: the highest section RVA +
+    // virt_size (headers live at rva 0).  Sections are scattered across
+    // RVAs, so this can exceed the file size.
+    let image_span = image_span_bytes(bytes);
+    let page_count = image_span.div_ceil(PAGE_SIZE);
     // Allocate physical pages at a fixed window (identity mapping makes
     // the link address = runtime address after relocation).
-    let base = crate::memory::alloc_pages_above(page_count, LOAD_BASE).ok_or(LoadError::AllocFailed)?;
+    let base =
+        crate::memory::alloc_pages_above(page_count, LOAD_BASE).ok_or(LoadError::AllocFailed)?;
     if base >= LOAD_LIMIT {
         crate::memory::free_pages(base, page_count);
         return Err(LoadError::TooBig);
     }
-    // Identity-map them user+writable+nx-default (relocations need write;
-    // we do not enforce W^X here — single-privilege-level kernel).
-    if !paging::map_range(base, page_count, paging::PAGE_USER | paging::PAGE_WRITABLE | paging::PAGE_NX) {
+    // Identity-map user+writable: single-privilege-level kernel, no W^X —
+    // the image contains code that must execute and data that relocates.
+    if !paging::map_range(base, page_count, paging::PAGE_USER | paging::PAGE_WRITABLE) {
         crate::memory::free_pages(base, page_count);
         return Err(LoadError::MapFailed);
     }
@@ -250,15 +276,27 @@ pub fn load_image(bytes: &[u8]) -> Result<Image, LoadError> {
 /// Zero the part of each section between raw_size and virt_size (BSS).
 fn zero_bss_tails(img: &[u8], mem: usize) {
     // Re-walk the section table minimally (headers are small).
-    let Some(e_lfanew) = u32_at(img, 0x3C) else { return };
+    let Some(e_lfanew) = u32_at(img, 0x3C) else {
+        return;
+    };
     let coff = e_lfanew as usize + 4;
-    let Some(num_sections) = u16_at(img, coff + 2) else { return };
-    let Some(opt_size) = u16_at(img, coff + 16) else { return };
+    let Some(num_sections) = u16_at(img, coff + 2) else {
+        return;
+    };
+    let Some(opt_size) = u16_at(img, coff + 16) else {
+        return;
+    };
     let sect_tab = coff + 20 + opt_size as usize;
     for i in 0..num_sections as usize {
-        let Some(virt_size) = u32_at(img, sect_tab + i * 40 + 8) else { return };
-        let Some(raw_size) = u32_at(img, sect_tab + i * 40 + 16) else { return };
-        let Some(virt_addr) = u32_at(img, sect_tab + i * 40 + 12) else { return };
+        let Some(virt_size) = u32_at(img, sect_tab + i * 40 + 8) else {
+            return;
+        };
+        let Some(raw_size) = u32_at(img, sect_tab + i * 40 + 16) else {
+            return;
+        };
+        let Some(virt_addr) = u32_at(img, sect_tab + i * 40 + 12) else {
+            return;
+        };
         if virt_size > raw_size {
             unsafe {
                 core::ptr::write_bytes(
diff --git a/shell/src/lib.rs b/shell/src/lib.rs
index 9ca999e..5248ea9 100644
--- a/shell/src/lib.rs
+++ b/shell/src/lib.rs
@@ -711,7 +711,13 @@ fn emit_string(pipe: &mut Pipeline, s: &str) {
 
 /// Commands that may consume an input stream (`USE`/`TIL`).
 fn accepts_input(cmd: &str) -> bool {
-    matches!(cmd, "cat" | "hexdump")
+    // Built-in stream consumers, plus any external binary in /bin/cli —
+    // externals get injected streams by construction (ADR 0006), so the
+    // keyword is never ambiguous for them.
+    if matches!(cmd, "cat" | "hexdump") {
+        return true;
+    }
+    rsnix::exec::resolve("cli", cmd).is_some()
 }
 
 fn execute(s: &str) {
@@ -748,8 +754,10 @@ write `{} <path> TIL <marker>`",
     // Resolution rule (ADR 0006): \bin\<kind>\<name> first, else the
     // built-in.  The frozen core (help/clear/echo/poweroff/exit/halt/
     // rescan) is exempt — it exists to fix the system and never migrates.
-    if !matches!(cmd, "help" | "clear" | "echo" | "halt" | "poweroff" | "exit" | "rescan")
-        && exec_external(&parsed)
+    if !matches!(
+        cmd,
+        "help" | "clear" | "echo" | "halt" | "poweroff" | "exit" | "rescan"
+    ) && exec_external(&parsed)
     {
         return;
     }
@@ -801,30 +809,133 @@ write `{} <path> TIL <marker>`",
 }
 
 // ---------------------------------------------------------------------------
-// External binaries (ADR 0006 Phase 2)
+// External binaries (ADR 0006 Phase 3)
+//
+// Resolution: \bin\cli\<name> on each mounted volume; else built-in.
+//
+// Stream injection: the Exports callbacks run in the COMMAND task but
+// touch shell-owned state.  This is safe by construction — the invoking
+// shell blocks in exec_external until the command completes, so nothing
+// else touches the Pipeline or the channel.  One invocation at a time.
 //
-// Resolution: \bin\cli\<name> on each mounted volume; else "unknown
-// command".  The Exports table is constructed per invocation: v1 wires
-// put_line to the active terminal and get_line to EOF (stream injection
-// through the filter pipeline lands in Phase 3).
+//   put_line → Pipeline::line directly (output side needs no queue).
+//   get_line ← a bounded per-invocation ring; while the command runs,
+//              the blocked shell "pumps": drains its key queue, assembles
+//              canonical lines, pushes them into the ring.  ^C marks the
+//              channel aborted instead of killing a stream consumer.
 // ---------------------------------------------------------------------------
 
 static mut LAST_EXIT_CODE: i32 = 0;
 
+/// The pipeline of the running external command (None when idle).
+static mut ACTIVE_PIPELINE: Option<Pipeline> = None;
+
 /// Last exit code of an external command ($? equivalent for scripts).
 pub fn last_exit_code() -> i32 {
     unsafe { core::ptr::addr_of!(LAST_EXIT_CODE).read() }
 }
 
-/// ABI callback: print one line to the invoking terminal.
+/// Bounded line channel between the pumping shell and the command task.
+struct LineChannel {
+    lines: Vec<String>,
+    len: usize,
+    cap: usize,
+    /// ^C seen: get_line returns Abort once drained.
+    aborted: bool,
+    /// Source exhausted (EOF / marker hit).
+    eof: bool,
+    /// Interactive TIL: exact line that closes the input (pump checks it).
+    marker: Option<String>,
+}
+
+impl LineChannel {
+    const CAP: usize = 16;
+    fn new() -> Self {
+        LineChannel {
+            lines: Vec::with_capacity(Self::CAP),
+            len: 0,
+            cap: Self::CAP,
+            aborted: false,
+            eof: false,
+            marker: None,
+        }
+    }
+    /// Shell side: push one assembled line.  Returns false when full
+    /// (the pump waits and retries).
+    fn push(&mut self, s: &str) -> bool {
+        if self.len >= self.cap {
+            return false;
+        }
+        self.lines.push(String::from(s));
+        self.len += 1;
+        true
+    }
+    /// Command side: pop one line.
+    fn pop(&mut self) -> Option<String> {
+        if self.len == 0 {
+            return None;
+        }
+        let s = self.lines.remove(0);
+        self.len -= 1;
+        Some(s)
+    }
+}
+
+/// The current invocation.  Only touched by the shell (pump + setup) and
+/// by the ABI callbacks (command task) — never concurrently, see above.
+static mut INVOCATION: Option<LineChannel> = None;
+
+/// Set when the input source is interactive (tty): the pump assembles
+/// canonical lines from keys.  File/marker sources are pre-drained into
+/// the channel before spawn instead.
+fn with_invocation<R>(f: impl FnOnce(&mut LineChannel) -> R) -> R {
+    let inv = unsafe {
+        match &mut *core::ptr::addr_of_mut!(INVOCATION) {
+            Some(ch) => ch,
+            None => panic!("exec: no active invocation"),
+        }
+    };
+    f(inv)
+}
+
+/// ABI callback: print one line through the filter pipeline / sink.
 fn abi_put_line(line: rsapi::Line) -> bool {
-    term_println(line.as_str());
-    true
+    let pipe = unsafe {
+        match &mut *core::ptr::addr_of_mut!(ACTIVE_PIPELINE) {
+            Some(p) => p,
+            None => panic!("exec: no active pipeline"),
+        }
+    };
+    pipe.line(line.as_str())
 }
 
-/// ABI callback: read one input line — no stream source wired yet (Phase 3).
-fn abi_get_line(_out: &mut rsapi::Line) -> rsapi::StreamResult {
-    rsapi::StreamResult::Eof
+/// ABI callback: read one input line from the invocation channel.
+fn abi_get_line(out: &mut rsapi::Line) -> rsapi::StreamResult {
+    loop {
+        let (line, status) = with_invocation(|ch| match ch.pop() {
+            Some(l) => (Some(l), rsapi::StreamResult::Line),
+            None if ch.aborted => (None, rsapi::StreamResult::Abort),
+            None if ch.eof => (None, rsapi::StreamResult::Eof),
+            None => (None, rsapi::StreamResult::Line), // empty but open: pump
+        });
+        match status {
+            rsapi::StreamResult::Abort => return rsapi::StreamResult::Abort,
+            rsapi::StreamResult::Eof => return rsapi::StreamResult::Eof,
+            rsapi::StreamResult::Line => {
+                if let Some(l) = line {
+                    out.ptr = l.as_ptr();
+                    out.len = l.len();
+                    // Leak the String into the command's lifetime: lines are
+                    // one-shot and the task dies soon; freeing across tasks
+                    // is not worth the bookkeeping.
+                    core::mem::forget(l);
+                    return rsapi::StreamResult::Line;
+                }
+                // Channel empty: yield so the shell pump can refill.
+                rsnix::sched::yield_now();
+            }
+        }
+    }
 }
 
 /// ABI callback: kernel log.
@@ -837,8 +948,7 @@ fn abi_exit(code: i32) -> ! {
     unsafe {
         LAST_EXIT_CODE = code;
     }
-    // Unwind the binary by longjmp-style task abort: the trampoline never
-    // returns here.  We report the code via the static and kill the task.
+    // Kill the calling task; exec_external sees the abort notification.
     rsnix::sched::abort_current("exit")
 }
 
@@ -871,37 +981,199 @@ fn exec_external(p: &Parsed) -> bool {
             return true;
         }
     };
+
+    // Grammar stays shell-side (ADR 0006): the binary gets clean argv and
+    // injected streams.  Build its pipeline exactly like a built-in's.
+    let pipe = Pipeline::new(&p.filters, &p.output);
+    #[allow(unused_mut)]
+    // Input channel: pre-drain file sources; tty sources are pumped.
+    let mut channel = LineChannel::new();
+    match &p.input {
+        Input::None => {}
+        Input::Use(path) => {
+            drain_file_into(&mut channel, path, None);
+            channel.eof = true;
+        }
+        Input::UseTil(path, marker) => {
+            drain_file_into(&mut channel, path, Some(marker));
+            channel.eof = true;
+        }
+        Input::Til(marker) => {
+            if p.args.is_empty() {
+                // Interactive: read from tty until marker — the pump
+                // checks each submitted line against it.
+                channel.marker = Some(marker.clone());
+            } else {
+                for path in &p.args {
+                    drain_file_into(&mut channel, path, Some(marker));
+                }
+                channel.eof = true;
+            }
+        }
+    }
+
     let mut argv: Vec<String> = Vec::with_capacity(p.args.len() + 1);
     argv.push(p.cmd.clone());
     argv.extend(p.args.iter().cloned());
     let name = alloc::format!("{}", p.cmd);
+
+    // Publish the invocation state before spawn (the command may call
+    // put_line immediately).
+    unsafe {
+        *core::ptr::addr_of_mut!(ACTIVE_PIPELINE) = Some(pipe);
+        *core::ptr::addr_of_mut!(INVOCATION) = Some(channel);
+    }
+
+    let mut cmd_done = false;
+    let mut exit_code: i64 = 0;
     match rsnix::exec::spawn(image, &EXPORTS_V1, &argv, &name) {
-        Ok(id) => {
-            // Wait for the completion message (TaskMsg::User(exit_code)).
+        Ok(_id) => {
+            // Pumping mode: assemble canonical lines from our queue into
+            // the channel while the command runs.  The WM pump delivers
+            // keys to us only while our window is focused.
             loop {
                 match rsnix::sched::recv_blocking() {
                     rsnix::sched::TaskMsg::User(code) => {
-                        unsafe {
-                            LAST_EXIT_CODE = code as i32;
-                        }
-                        break;
+                        exit_code = if code == u64::MAX { -1 } else { code as i64 };
+                        cmd_done = true;
                     }
                     rsnix::sched::TaskMsg::Key(rsnix::ps2::KeyEvent::Ctrl('c')) => {
-                        // ^C while waiting: kill the command task.
-                        let _ = rsnix::sched::kill(id);
-                        term_err("^C");
-                        break;
+                        // ^C: mark the channel aborted; stream consumers stop
+                        // at their next get_line/put_line-false.  Compute-
+                        // bound commands stay blind to it (accepted policy).
+                        with_invocation(|ch| ch.aborted = true);
+                    }
+                    rsnix::sched::TaskMsg::Key(ev) => {
+                        // Canonical assembly for the interactive source:
+                        // mirror tty::read_line's key handling, minus echo
+                        // (the terminal already echoed it? no — the shell
+                        // must echo here since nothing else does).
+                        pump_key(ev);
                     }
                     _ => {}
                 }
+                if cmd_done {
+                    break;
+                }
             }
         }
-        Err(rsnix::exec::ExecError::Busy) => term_err(&alloc::format!("{}: task table full", p.cmd)),
+        Err(rsnix::exec::ExecError::Busy) => {
+            term_err(&alloc::format!("{}: task table full", p.cmd))
+        }
         Err(_) => term_err(&alloc::format!("{}: cannot start", p.cmd)),
     }
+
+    unsafe {
+        let pipe_ptr = core::ptr::addr_of_mut!(ACTIVE_PIPELINE);
+        if let Some(mut pipe) = (*pipe_ptr).take() {
+            pipe.end();
+        }
+        *core::ptr::addr_of_mut!(INVOCATION) = None;
+        LAST_EXIT_CODE = exit_code as i32;
+    }
     true
 }
 
+/// One key of canonical assembly during pumping.  Lines land in the
+/// channel; ^D closes it; Enter submits the line.  Echo goes to the
+/// terminal (this is the command's stdin).
+fn pump_key(ev: rsnix::ps2::KeyEvent) {
+    use rsnix::ps2::KeyEvent;
+    // Static assembly buffer: only touched between keys by the shell.
+    static mut PUMP_BUF: Option<String> = None;
+    let buf = unsafe {
+        let p = core::ptr::addr_of_mut!(PUMP_BUF);
+        if (*p).is_none() {
+            *p = Some(String::new());
+        }
+        (*p).as_mut().unwrap()
+    };
+    match ev {
+        KeyEvent::Enter => {
+            gop_println!();
+            let line = core::mem::take(buf);
+            with_invocation(|ch| {
+                // Interactive TIL: the submitted line itself is checked
+                // against the marker (it is consumed, not delivered).
+                if let Some(m) = &ch.marker
+                    && line.as_str() == m
+                {
+                    ch.eof = true;
+                } else {
+                    ch.push(line.as_str());
+                }
+            });
+        }
+        KeyEvent::Ctrl('c') => {
+            buf.clear();
+            with_invocation(|ch| ch.aborted = true);
+        }
+        KeyEvent::Ctrl('d') => {
+            if buf.is_empty() {
+                with_invocation(|ch| ch.eof = true);
+            }
+        }
+        KeyEvent::Char(c) if c.is_ascii_graphic() || c == ' ' => {
+            buf.push(c);
+            gop_print!("{}", c);
+        }
+        KeyEvent::Backspace => {
+            if buf.pop().is_some() {
+                gop_print!("\x08");
+            }
+        }
+        _ => {}
+    }
+}
+
+/// Pre-drain a file into the channel; `marker` stops before that line
+/// (USE <path> TIL sugar).  The channel is left open — the caller sets
+/// `eof` when this is the whole source.
+fn drain_file_into(ch: &mut LineChannel, path: &str, marker: Option<&str>) {
+    let h = match rsnix::fs::resolve(path) {
+        Some(h) => h,
+        None => {
+            term_err(&alloc::format!("{}: no such file", path));
+            ch.eof = true;
+            return;
+        }
+    };
+    drain_handle(ch, h, marker);
+}
+
+/// Shared drain: read `h` in chunks, push complete lines; stop at marker.
+fn drain_handle(ch: &mut LineChannel, h: usize, marker: Option<&str>) {
+    let mut buf = [0u8; 512];
+    let mut pos = 0u64;
+    let mut pending = String::new();
+    'outer: loop {
+        match rsnix::fs::read(h, &mut buf, pos) {
+            Ok(0) | Err(_) => break,
+            Ok(n) => {
+                pos += n as u64;
+                pending.push_str(&String::from_utf8_lossy(&buf[..n]));
+                while let Some(idx) = pending.find('\n') {
+                    let line: String = pending.drain(..=idx).collect();
+                    let line = line.trim_end_matches('\n').to_string();
+                    if let Some(m) = marker
+                        && line == m
+                    {
+                        break 'outer;
+                    }
+                    while !ch.push(&line) {
+                        rsnix::sched::yield_now();
+                    }
+                }
+            }
+        }
+    }
+    if marker.is_none() && !pending.is_empty() {
+        while !ch.push(&pending) {
+            rsnix::sched::yield_now();
+        }
+    }
+}
+
 // ---------------------------------------------------------------------------
 // Built-in commands
 // ---------------------------------------------------------------------------
@@ -973,6 +1245,7 @@ an operator: echo \"WRITING\" prints WRITING.
 
 fn cmd_echo(p: &Parsed) {
     let mut pipe = Pipeline::new(&p.filters, &p.output);
+    #[allow(unused_mut)]
     let line = p.args.join(" ");
     let _ = pipe.line(&line);
     pipe.end();
@@ -980,6 +1253,7 @@ fn cmd_echo(p: &Parsed) {
 
 fn cmd_hello(p: &Parsed) {
     let mut pipe = Pipeline::new(&p.filters, &p.output);
+    #[allow(unused_mut)]
     let _ = pipe.line("World!");
     pipe.end();
 }
@@ -1036,6 +1310,7 @@ fn cmd_clear() {
 
 fn cmd_info(p: &Parsed) {
     let mut pipe = Pipeline::new(&p.filters, &p.output);
+    #[allow(unused_mut)]
     let mut s = String::from("rs-dos v");
     s.push_str(env!("CARGO_PKG_VERSION"));
     s.push_str(" - rsnix kernel");
@@ -1047,6 +1322,7 @@ fn cmd_info(p: &Parsed) {
 fn cmd_ls(p: &Parsed) {
     let path = p.args.first().map(|s| s.as_str()).unwrap_or("/");
     let mut pipe = Pipeline::new(&p.filters, &p.output);
+    #[allow(unused_mut)]
     match rsnix::fs::resolve(path) {
         None => term_err(&alloc::format!("ls: {}: no such file or directory", path)),
         Some(h) => {
@@ -1111,6 +1387,7 @@ enum ReadResult {
 
 fn cmd_cat(p: &Parsed) {
     let mut pipe = Pipeline::new(&p.filters, &p.output);
+    #[allow(unused_mut)]
     match &p.input {
         Input::None => {
             if p.args.is_empty() {
@@ -1318,6 +1595,7 @@ impl HexCtx<'_> {
 
 fn cmd_hexdump(p: &Parsed) {
     let mut pipe = Pipeline::new(&p.filters, &p.output);
+    #[allow(unused_mut)]
     match &p.input {
         Input::None => {
             if p.args.is_empty() {
@@ -1633,6 +1911,7 @@ fn cmd_mem(p: &Parsed) {
     let ps = rsnix::memory::page_stats();
 
     let mut pipe = Pipeline::new(&p.filters, &p.output);
+    #[allow(unused_mut)]
     let _ = pipe.line(&alloc::format!(
         "memory map (key=0x{:x}, {} entries)",
         key,
@@ -1699,6 +1978,7 @@ fn cmd_mem(p: &Parsed) {
 fn cmd_mmap(p: &Parsed) {
     let (map, key) = rsnix::memory::get_map();
     let mut pipe = Pipeline::new(&p.filters, &p.output);
+    #[allow(unused_mut)]
     let _ = pipe.line(&alloc::format!(
         "memory map (key=0x{:x}, {} entries):",
         key,
@@ -1744,6 +2024,7 @@ fn cmd_stress(p: &Parsed) {
     };
 
     let mut pipe = Pipeline::new(&p.filters, &p.output);
+    #[allow(unused_mut)]
     let mut allocated = 0usize;
     {
         let mut blocks: [Option<alloc::vec::Vec<u8>>; 64] = core::array::from_fn(|_| None);
@@ -1931,6 +2212,7 @@ fn cmd_heapdbg(_p: &Parsed) {
 fn cmd_date(p: &Parsed) {
     let t = rsnix::rtc::now();
     let mut pipe = Pipeline::new(&p.filters, &p.output);
+    #[allow(unused_mut)]
     let line = alloc::format!(
         "{}-{:02}-{:02} {:02}:{:02}:{:02}",
         t.year,
@@ -2184,6 +2466,7 @@ fn cmd_bg(p: &Parsed) {
 
 fn cmd_jobs(p: &Parsed) {
     let mut pipe = Pipeline::new(&p.filters, &p.output);
+    #[allow(unused_mut)]
     let lines = rsnix::sched::jobs();
     if lines.is_empty() {
         let _ = pipe.line("jobs: no tasks");
@@ -2440,6 +2723,7 @@ fn usb_if_label(c: u8, s: u8, p: u8) -> &'static str {
 fn cmd_pci(p: &Parsed) {
     let devices = rsnix::pci::probe();
     let mut pipe = Pipeline::new(&p.filters, &p.output);
+    #[allow(unused_mut)]
     let _ = pipe.line(&alloc::format!("pci: {} device(s)", devices.len()));
     let _ = pipe.line("  bus dev fn  class              vendor:device   if");
     for d in &devices {
@@ -2466,11 +2750,13 @@ fn cmd_usb(p: &Parsed) {
         rsnix::usb::rescan();
         rsnix::ehci::rescan();
         let mut pipe = Pipeline::new(&p.filters, &p.output);
+        #[allow(unused_mut)]
         let _ = pipe.line("usb: rescanning ports");
         pipe.end();
         return;
     }
     let mut pipe = Pipeline::new(&p.filters, &p.output);
+    #[allow(unused_mut)]
     for l in rsnix::usb::diagnose().lines() {
         if !pipe.line(l) {
             break;
@@ -2479,6 +2765,7 @@ fn cmd_usb(p: &Parsed) {
     pipe.end();
     // EHCI (USB2) controller status - the real hardware path.
     let mut pipe = Pipeline::new(&p.filters, &p.output);
+    #[allow(unused_mut)]
     for l in rsnix::ehci::diagnose().lines() {
         if !pipe.line(l) {
             break;
@@ -2487,6 +2774,7 @@ fn cmd_usb(p: &Parsed) {
     pipe.end();
     // UHCI (USB1.1 companion) — full/low-speed keyboards and mice.
     let mut pipe = Pipeline::new(&p.filters, &p.output);
+    #[allow(unused_mut)]
     for l in rsnix::uhci::diagnose().lines() {
         if !pipe.line(l) {
             break;