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, 53 insertions, 15 deletions

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(