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

2 files changed, 74 insertions, 0 deletions

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
+}