kernel+sh: real anonymous pipes (concurrent, streaming, SIGPIPE)

Replace the shell's temp-file pipe hack with proper in-kernel FIFOs.

Kernel (src/pipe.asm, new):
- A small pool of bounded ring buffers (PIPE_MAX=2, 64 B each) with
  writers/readers refcounts. Pipe fds are $F0+idx*2 (read) / +1 (write);
  $FF stays "console".
- SYS_PIPE allocates one (writers=readers=1) and returns the read fd
  (write = read+1). getb/putb/close dispatch pipe fds here; sys_exit drops
  the refcounts held as PROC_STDIN/PROC_STDOUT.
- Blocking with SchedYield, which is the flow control: read blocks while
  empty with a writer (EOF once writers hit 0), write blocks while full
  with a reader, and if the last reader is gone the writer is killed
  (SIGPIPE -> exit 141). Cooperative-scheduler friendly.

Shell (c/sh.c):
- run_pipeline(): split on '|', make a pipe between adjacent stages, and
  fork ALL stages concurrently (no wait between), wiring stdin/stdout;
  then wait for all. Per-stage >/< still honored; orphaned pipe ends are
  closed on a lookup miss so EOF/EPIPE propagate.
- Drop the __pipe temp file and its 2 KB / serialized limits.

libc: pipe(). New c/ptest.c exercises the FIFO (write, read back, EOF).

Now works (old version couldn't): multi-stage a|b|c; streaming beyond 2 KB
(count 120 | wc = 3372 bytes through a 64 B buffer); early-exit SIGPIPE
(count 200 | true kills count instead of hanging/overflowing a file).

11 files changed, 349 insertions, 15 deletions

diff --git a/Makefile b/Makefile
index 62401b3..ceb9d51 100644
--- a/Makefile
+++ b/Makefile
@@ -20,7 +20,7 @@ $(BUILD)/%.o: src/%.asm | $(BUILD)
 # C programs: compiled with SDCC into ROM-bank blobs and INCBIN'd by programs.asm
 CBLOBS := c/chello.bin c/echo.bin c/true.bin c/false.bin c/uname.bin \
           c/pid.bin c/cat.bin c/wc.bin c/head.bin c/args.bin \
-          c/ls.bin c/save.bin c/rm.bin c/sh.bin c/ps.bin c/kill.bin c/spin.bin c/blk.bin c/mkdir.bin c/count.bin
+          c/ls.bin c/save.bin c/rm.bin c/sh.bin c/ps.bin c/kill.bin c/spin.bin c/blk.bin c/mkdir.bin c/count.bin c/ptest.bin
 $(BUILD)/programs.o: $(CBLOBS)
 
 # C programs also depend on the shared libc sources
diff --git a/c/gbos.h b/c/gbos.h
index 2a99c0e..a0619cc 100644
--- a/c/gbos.h
+++ b/c/gbos.h
@@ -12,6 +12,7 @@ char         *getargs(void);              /* this program's argument string */
 char          readc(void);                /* one input byte, EOF at end    */
 void          sexit(unsigned char code);
 unsigned char getpid(void);
+unsigned char pipe(void);         /* -> read fd; write end is read+1 */
 
 /* helpers (c/libc.c) */
 void          putu(unsigned int n);              /* print decimal          */
diff --git a/c/libc.s b/c/libc.s
index d499521..ec59bb7 100644
--- a/c/libc.s
+++ b/c/libc.s
@@ -1,6 +1,6 @@
 	.module libc
 	.globl _writes, _putc, _puts, _nl, _strlen, _readc, _sexit, _getpid, _getargs
-	.globl _open, _close, _fgetc, _fputc, _flist, _fremove
+	.globl _open, _close, _fgetc, _fputc, _flist, _fremove, _pipe
 	;; SDCC sm83 sdcccall(1): 1st arg -> A (byte) or DE (pointer); ret A / BC.
 	;; rst $30 (the trap) clobbers BC/DE/HL; SDCC treats them caller-saved, so
 	;; wrappers need not preserve them - but we compute cleanly regardless.
@@ -86,6 +86,12 @@ _getpid:
 	rst	#0x30
 	ret
 
+	;; unsigned char pipe(void) -> A (read fd; write end is read+1)
+_pipe:
+	ld	c, #28			; SYS_PIPE
+	rst	#0x30
+	ret
+
 	;; unsigned char open(const char *name /*DE*/, unsigned char mode /*A*/) -> A(fd)
 _open:
 	ld	b, a			; B = mode
diff --git a/c/ptest.c b/c/ptest.c
new file mode 100644
index 0000000..88b7d30
--- /dev/null
+++ b/c/ptest.c
@@ -0,0 +1,19 @@
+#include "gbos.h"
+/* ptest: exercise the kernel pipe in one process - write some bytes, read them
+   back, then close the write end and confirm the reader sees EOF. */
+void main(void) {
+    unsigned char pr = pipe();
+    unsigned char pw = pr + 1;
+    int c;
+    if (pr == NOFD) { puts("pipe: none free"); nl(); return; }
+    fputc(pw, 'H'); fputc(pw, 'i'); fputc(pw, '!');
+    c = fgetc(pr); putc(c);
+    c = fgetc(pr); putc(c);
+    c = fgetc(pr); putc(c);
+    nl();
+    close(pw);                 /* last writer gone */
+    c = fgetc(pr);
+    if (c < 0) puts("EOF"); else { puts("got "); putc(c); }
+    nl();
+    close(pr);
+}
diff --git a/c/sh.c b/c/sh.c
index df2c5e9..a8dd09d 100644
--- a/c/sh.c
+++ b/c/sh.c
@@ -95,6 +95,61 @@ done:
     if (myout != NOFD) close(myout);
 }
 
+/* run a multi-stage pipeline (a | b | c): wire stages together with real
+   kernel pipes, launch them ALL concurrently, then wait for them. */
+static void run_pipeline(char **tok, unsigned char nt) {
+    unsigned char pids[8], npid = 0;
+    unsigned char cfd[8], ncfd = 0;      /* file-redirect fds; closed after all exit */
+    unsigned char start = 0, prev_r = NOFD;
+    for (;;) {
+        unsigned char end = start, last, i, cmdend, fin = NOFD, fout = NOFD;
+        unsigned char my_out = NOFD, next_r = NOFD, id, pid;
+        while (end < nt && !isop(tok[end], '|')) end++;
+        last = (end >= nt);
+        if (!last) {                     /* pipe from this stage to the next */
+            unsigned char pr = pipe();
+            if (pr != NOFD) { next_r = pr; my_out = pr + 1; }
+        }
+        cmdend = end;                    /* this stage's own >/< redirects */
+        for (i = start; i < end; i++) {
+            if (isop(tok[i], '>')) { if (i + 1 < end) fout = open(tok[i+1], O_WRITE); if (i < cmdend) cmdend = i; }
+            else if (isop(tok[i], '<')) { if (i + 1 < end) fin = open(tok[i+1], O_READ); if (i < cmdend) cmdend = i; }
+        }
+        id = lookup(tok[start]);
+        if (id == NOFD) {
+            puts(tok[start]); puts(": not found"); nl();
+            if (prev_r != NOFD) close(prev_r);   /* orphaned pipe ends -> EOF/EPIPE */
+            if (my_out != NOFD) close(my_out);
+            if (fin != NOFD) close(fin);
+            if (fout != NOFD) close(fout);
+        } else {
+            set_cmdline(tok, start, cmdend);
+            pid = fork();
+            if (pid == 0) {
+                if (fin != NOFD) setin(fin); else if (prev_r != NOFD) setin(prev_r);
+                if (fout != NOFD) setout(fout); else if (my_out != NOFD) setout(my_out);
+                exec(id);
+            }
+            if (npid < 8) pids[npid++] = pid;
+            if (fin != NOFD && ncfd < 8) cfd[ncfd++] = fin;
+            if (fout != NOFD && ncfd < 8) cfd[ncfd++] = fout;
+        }
+        prev_r = next_r;
+        if (last) break;
+        start = end + 1;
+    }
+    { unsigned char got = 0;             /* wait for all stages */
+      while (got < npid) {
+          unsigned char r = wait(), j, mine = 0;
+          if (r == NOFD) break;
+          for (j = 0; j < npid; j++) if (pids[j] == r) { mine = 1; break; }
+          if (mine) got++;
+          else { putc('['); putu(r); puts(" done]"); nl(); }
+      }
+    }
+    { unsigned char k; for (k = 0; k < ncfd; k++) close(cfd[k]); }
+}
+
 void main(void) {
     char line[80];
     char *tok[16];
@@ -133,16 +188,7 @@ void main(void) {
         pipepos = 0;
         for (i = 1; i < nt; i++)
             if (isop(tok[i], '|')) { pipepos = i; break; }
-        if (pipepos) {
-            unsigned char pw = open("__pipe", O_WRITE);
-            run(tok, 0, pipepos, NOFD, pw, 0);
-            close(pw);
-            { unsigned char pr = open("__pipe", O_READ);
-              run(tok, pipepos + 1, nt, pr, NOFD, 0);
-              close(pr); }
-            fremove("__pipe");
-        } else {
-            run(tok, 0, nt, NOFD, NOFD, bg);
-        }
+        if (pipepos) run_pipeline(tok, nt);
+        else run(tok, 0, nt, NOFD, NOFD, bg);
     }
 }
diff --git a/include/gbos.inc b/include/gbos.inc
index 9b44b33..aca92e3 100644
--- a/include/gbos.inc
+++ b/include/gbos.inc
@@ -125,7 +125,8 @@ DEF SYS_BPEEK   EQU 24   ; [debug] read a disk block      (B=block, E=off -> A)
 DEF SYS_MKDIR   EQU 25   ; make a directory               (DE=path -> A=0/$FF)
 DEF SYS_CHDIR   EQU 26   ; change directory               (DE=path -> A=0/$FF)
 DEF SYS_OPENDIR EQU 27   ; point ls at a directory        (DE=path -> A=0/$FF)
-DEF SYS_MAX     EQU 28
+DEF SYS_PIPE    EQU 28   ; make a pipe                    (-> A=read fd; write=A+1)
+DEF SYS_MAX     EQU 29
 ; SYS_KILL (9) is now implemented (B=pid -> A=0/$FF; pid 1 is immortal)
 ; (SYS_OPEN=4 / SYS_CLOSE=5 are now implemented by the filesystem)
 
@@ -179,6 +180,11 @@ DEF OF_MODE  EQU 2
 DEF OF_POS   EQU 3
 DEF OF_SIZE  EQU 5
 
+; anonymous pipes
+DEF PIPE_MAX     EQU 2      ; concurrent pipes
+DEF PIPE_BUFSZ   EQU 64     ; ring buffer bytes per pipe
+DEF PIPE_FD_BASE EQU $F0    ; pipe fds: $F0+idx*2 (read), +1 (write); $FF=console
+
 ; -----------------------------------------------------------------------------
 ; Program table indices (see programs.asm). exec() takes one of these in B.
 ; -----------------------------------------------------------------------------
@@ -204,5 +210,6 @@ DEF PROG_SPIN   EQU 18
 DEF PROG_BLK    EQU 19
 DEF PROG_MKDIR  EQU 20
 DEF PROG_COUNT  EQU 21
+DEF PROG_PTEST  EQU 22
 
     ENDC
diff --git a/src/boot.asm b/src/boot.asm
index 7eed4a8..23511c4 100644
--- a/src/boot.asm
+++ b/src/boot.asm
@@ -55,6 +55,7 @@ KernelInit:
     ld [wCurProc], a
     ld a, HIGH(wKernelPCB)
     ld [wCurProc+1], a
+    call pipe_init              ; mark all pipes free
     call fs_init                ; mount the disk (format on first boot; persists)
     call term_init              ; set up the LCD text terminal
     call osk_init               ; build the on-screen keyboard (window layer)
diff --git a/src/fs.asm b/src/fs.asm
index a4ad77e..1c6d299 100644
--- a/src/fs.asm
+++ b/src/fs.asm
@@ -761,6 +761,13 @@ sys_open::
 ; =============================================================================
 sys_close::
     ld a, b
+    call is_pipe_fd
+    jr c, .notpipe
+    call pipe_close
+    xor a
+    ret
+.notpipe
+    ld a, b
     cp OF_MAX
     jr nc, .done
     ld [wFsSlot], a             ; save fd
@@ -787,6 +794,9 @@ sys_close::
 ; =============================================================================
 sys_getb::
     ld a, b
+    call is_pipe_fd
+    jp nc, pipe_read            ; A = pipe read fd
+    ld a, b
     cp OF_MAX
     jr nc, .eof
     call OFPtr
@@ -868,6 +878,10 @@ sys_getb::
 sys_putb::
     ld a, e
     ld [wFsByte], a
+    ld [wPipeByte], a
+    ld a, b
+    call is_pipe_fd
+    jp nc, pipe_write           ; A = pipe write fd, byte in wPipeByte
     ld a, b
     cp OF_MAX
     jr nc, .done
diff --git a/src/pipe.asm b/src/pipe.asm
new file mode 100644
index 0000000..f9473a4
--- /dev/null
+++ b/src/pipe.asm
@@ -0,0 +1,215 @@
+; =============================================================================
+; pipe.asm - anonymous pipes: bounded in-kernel FIFOs with blocking ends.
+;
+; A pipe fd is $F0 + idx*2 (+1 for the write end); $FF stays "console". getb/
+; putb/close dispatch pipe fds here. read blocks (yields) while empty and there
+; is still a writer, then returns EOF once all writers are gone; write blocks
+; while full and there is a reader, and if the last reader is gone the writer is
+; killed (SIGPIPE). refcounts (writers/readers) are set to 1 by pipe() - the one
+; producer and one consumer of a shell pipeline - and dropped on close/exit.
+; =============================================================================
+INCLUDE "include/gbos.inc"
+
+SECTION "pipe_bss", WRAM0
+wPipeW::    DS PIPE_MAX                 ; writers refcount (0 = end/free)
+wPipeR::    DS PIPE_MAX                 ; readers refcount
+wPipeHead:: DS PIPE_MAX                 ; read index
+wPipeTail:: DS PIPE_MAX                 ; write index
+wPipeCnt::  DS PIPE_MAX                 ; bytes buffered
+wPipeBuf::  DS PIPE_MAX * PIPE_BUFSZ    ; ring buffers
+wPipeByte:: DS 1                        ; scratch
+
+SECTION "pipe", ROM0
+
+; pipe_init - mark every pipe free (writers=readers=0). Called at boot.
+pipe_init::
+    ld hl, wPipeW
+    ld b, PIPE_MAX * 2                  ; wPipeW followed by wPipeR
+    xor a
+.z
+    ld [hl+], a
+    dec b
+    jr nz, .z
+    ret
+
+; sys_pipe - allocate a pipe -> A = read fd (write = read+1), or $FF.
+sys_pipe::
+    jp pipe_alloc
+
+; is_pipe_fd - A = fd. CF set (carry) if it is NOT a pipe fd. Preserves A.
+is_pipe_fd::
+    cp PIPE_FD_BASE
+    ret c                               ; < base -> not a pipe (CF set)
+    cp PIPE_FD_BASE + PIPE_MAX*2
+    ccf                                 ; >= top -> CF set (not pipe); else CF clear
+    ret
+
+; pipe_alloc - -> A = read fd, or $FF if none free. Empties + writers=readers=1.
+pipe_alloc::
+    ld c, 0
+.scan
+    ld hl, wPipeW
+    ld b, 0
+    add hl, bc
+    ld a, [hl]
+    or a
+    jr nz, .next
+    ld hl, wPipeR
+    add hl, bc
+    ld a, [hl]
+    or a
+    jr z, .found
+.next
+    inc c
+    ld a, c
+    cp PIPE_MAX
+    jr c, .scan
+    ld a, $FF
+    ret
+.found
+    ld b, 0
+    ld hl, wPipeW
+    add hl, bc
+    ld [hl], 1
+    ld hl, wPipeR
+    add hl, bc
+    ld [hl], 1
+    ld hl, wPipeHead
+    add hl, bc
+    ld [hl], 0
+    ld hl, wPipeTail
+    add hl, bc
+    ld [hl], 0
+    ld hl, wPipeCnt
+    add hl, bc
+    ld [hl], 0
+    ld a, c
+    add a                               ; idx*2
+    add PIPE_FD_BASE
+    ret
+
+; pipe_close - A = fd. If a pipe fd, drop the appropriate refcount.
+pipe_close::
+    call is_pipe_fd
+    ret c
+    sub PIPE_FD_BASE
+    ld c, a
+    srl c                               ; C = idx
+    and 1                               ; A = 1 if write end
+    ld hl, wPipeR
+    jr z, .have
+    ld hl, wPipeW
+.have
+    ld b, 0
+    add hl, bc
+    ld a, [hl]
+    or a
+    ret z
+    dec [hl]
+    ret
+
+; pipe_bufptr - C = idx, A = pos -> HL = &wPipeBuf[idx*PIPE_BUFSZ + pos]. keeps BC.
+; (idx*64 + pos <= 255, so the whole offset fits in one byte.)
+pipe_bufptr:
+    ld e, a                             ; E = pos
+    ld a, c
+    swap a
+    add a
+    add a                               ; A = idx*64
+    add e                               ; + pos
+    ld e, a
+    ld d, 0
+    ld hl, wPipeBuf
+    add hl, de
+    ret
+
+; pipe_read - A = read fd -> A = byte, CF set on EOF. Blocks while empty w/ writer.
+pipe_read::
+    sub PIPE_FD_BASE
+    srl a
+    ld c, a                             ; C = idx
+.retry
+    ld hl, wPipeCnt
+    ld b, 0
+    add hl, bc
+    ld a, [hl]
+    or a
+    jr nz, .data
+    ld hl, wPipeW                       ; empty: any writers left?
+    add hl, bc
+    ld a, [hl]
+    or a
+    jr z, .eof                          ; none -> EOF
+    call SchedYield                     ; block: let the writer run
+    jr .retry
+.data
+    ld hl, wPipeHead
+    add hl, bc
+    ld a, [hl]                          ; head
+    push af
+    call pipe_bufptr                    ; A=head, C=idx -> HL = &buf byte
+    ld a, [hl]
+    ld [wPipeByte], a
+    pop af                              ; head
+    inc a
+    cp PIPE_BUFSZ
+    jr c, .hok
+    xor a
+.hok
+    ld hl, wPipeHead
+    add hl, bc
+    ld [hl], a
+    ld hl, wPipeCnt
+    add hl, bc
+    dec [hl]
+    ld a, [wPipeByte]
+    and a                               ; CF = 0
+    ret
+.eof
+    scf
+    ret
+
+; pipe_write - A = write fd, byte in wPipeByte. Blocks while full w/ reader;
+; SIGPIPE (exit) if no reader.
+pipe_write::
+    sub PIPE_FD_BASE
+    srl a
+    ld c, a                             ; C = idx
+.retry
+    ld hl, wPipeCnt
+    ld b, 0
+    add hl, bc
+    ld a, [hl]
+    cp PIPE_BUFSZ
+    jr c, .space
+    ld hl, wPipeR                       ; full: any readers left?
+    add hl, bc
+    ld a, [hl]
+    or a
+    jr z, .epipe                        ; none -> SIGPIPE
+    call SchedYield                     ; block: let the reader drain
+    jr .retry
+.space
+    ld hl, wPipeTail
+    add hl, bc
+    ld a, [hl]                          ; tail
+    push af
+    call pipe_bufptr                    ; A=tail, C=idx -> HL = &buf byte
+    ld a, [wPipeByte]
+    ld [hl], a
+    pop af                              ; tail
+    inc a
+    cp PIPE_BUFSZ
+    jr c, .tok
+    xor a
+.tok
+    ld hl, wPipeTail
+    add hl, bc
+    ld [hl], a
+    ld hl, wPipeCnt
+    add hl, bc
+    inc [hl]
+    ret
+.epipe
+    ld b, 141                           ; 128 + SIGPIPE(13)
+    jp sys_exit
diff --git a/src/programs.asm b/src/programs.asm
index 40d9034..e659cee 100644
--- a/src/programs.asm
+++ b/src/programs.asm
@@ -106,6 +106,9 @@ ProgMkdir:
 SECTION "prog_count",  ROMX[$4000], BANK[23]
 ProgCount:
     INCBIN "c/count.bin"
+SECTION "prog_ptest",  ROMX[$4000], BANK[24]
+ProgPtest:
+    INCBIN "c/ptest.bin"
 
 ; -----------------------------------------------------------------------------
 ; PROG_SH (bank 3) - the shell, now written in C (c/sh.c): parses >/</| and
@@ -192,6 +195,8 @@ ProgramTable::
     dw ProgMkdir
     db LOW(BANK(ProgCount)), HIGH(BANK(ProgCount))
     dw ProgCount
+    db LOW(BANK(ProgPtest)), HIGH(BANK(ProgPtest))
+    dw ProgPtest
 
 ; -----------------------------------------------------------------------------
 ; NameTable (ROM0) - command name -> program id, searched by sys_lookup.
@@ -242,4 +247,6 @@ NameTable::
     db PROG_MKDIR
     db "count", 0
     db PROG_COUNT
+    db "ptest", 0
+    db PROG_PTEST
     db 0
diff --git a/src/syscall.asm b/src/syscall.asm
index cb2c996..c602097 100644
--- a/src/syscall.asm
+++ b/src/syscall.asm
@@ -62,6 +62,7 @@ SyscallTable:
     dw sys_mkdir                ; 25 MKDIR
     dw sys_chdir                ; 26 CHDIR
     dw sys_opendir              ; 27 OPENDIR
+    dw sys_pipe                 ; 28 PIPE
 
 ; -----------------------------------------------------------------------------
 sys_nosys:
@@ -256,7 +257,24 @@ sys_yield:
 ;   - wake our parent if it is blocked in wait()
 ;   - schedule away forever (RAM bank + PCB slot are freed by the reaper)
 ; -----------------------------------------------------------------------------
-sys_exit:
+sys_exit::
+    ; release any pipe ends held as stdin/stdout (so EOF/free fire)
+    ld a, [wCurProc]
+    add PROC_STDIN
+    ld l, a
+    ld a, [wCurProc+1]
+    adc 0
+    ld h, a
+    ld a, [hl]
+    call pipe_close
+    ld a, [wCurProc]
+    add PROC_STDOUT
+    ld l, a
+    ld a, [wCurProc+1]
+    adc 0
+    ld h, a
+    ld a, [hl]
+    call pipe_close
     ; PROC_STATE = PS_ZOMBIE
     ld a, [wCurProc]
     ld l, a