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).

1 files changed, 315 insertions, 27 deletions

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;