;;; gpu-worker.lsp -- TCP listener that dispatches bend-forms to local
;;; CUDA daemon binaries.
;;;
;;; Wire protocol: length-prefixed S-exp (see wire.lsp). Same shape
;;; the lumbda fleet workers in `ecdsa/lumbda/fleet/worker.lsp`
;;; already speak -- gpu-worker is the GPU-routing variant.
;;;
;;; Each accepted connection:
;;;   1. read one framed request   (wire-recv)
;;;   2. dispatch by op head       (handle-...)
;;;   3. send one framed reply     (wire-send)
;;;   4. close
;;;
;;; Daemons (long-lived --daemon binaries) are held open across all
;;; connections so CUDA context init pays once per worker startup,
;;; never per request. This is the architecture that makes the bend
;;; pattern actually faster than local CPU for repeated calls.
;;;
;;; Run:
;;;   lumbda gpu-worker.lsp                  ; default port 8320 (BEND)
;;;   lumbda gpu-worker.lsp --port 9001
;;;
;;; Port mnemonic — 8320 = BEND:
;;;   8 ~= B (implied infinity B flattened; bake a cake; baby & me)
;;;   3 ~= E (backward)
;;;   2 ~= N (pivoted 90 degrees)
;;;   0 ~= D (flattened)
;;;
;;; Requires the cuda binaries on disk; paths below.

(load "wire.lsp")

(define *worker-port* 8320)
(define *binary-shake-fanout*
  ;; Override via env or per host.
  "./shake256-fanout")

;; bend-cuda lives in the foxhop ecdsa repo and runs upstream-format
;; ops.bin against the CPU+GPU simulators, writing (cuda-sim-result …)
;; portals via --portal. No daemon mode: spawn-per-call.
;; BEND_CUDA env override > default bend-cuda path. AMD ROCm path
;; will land as BEND_ROCM env / bend-rocm binary; CPU fallback BEND_CPU.
(define *binary-bend-cuda*
  (or (get-environment-variable "BEND_CUDA")
      "/home/fox/git/www.foxhop.net/ecdsa/cuda/bend-cuda"))

;; cgbn-batch-worker — bend form B. CGBN bignum batch over BSHK protocol;
;; daemon mode mirrors shake256-fanout.cu (process-bin <in> <out>).
(define *binary-cgbn-batch*
  (or (get-environment-variable "CGBN_BATCH_WORKER")
      "./cgbn-batch-worker"))

;; secp256k1-batch-mul — bend form A. Batched scalar*G on secp256k1 via
;; vendored VanitySearch-Bitcrack GPUMath.h (AGPL-3.0). Binary wire uses
;; BSCP request / BSCR response magic to stay distinct from BSHK/BCGB.
(define *binary-secp256k1-batch*
  (or (get-environment-variable "SECP256K1_BATCH_WORKER")
      "./secp256k1-batch-mul"))

;; radix-sort — bend form G. Batched u64 ascending sort via NVIDIA CUB
;; DeviceRadixSort (header-only, ships with CUDA Toolkit). Binary wire
;; uses BSRT request / BSRR response magic, distinct from BSHK/BCGB/BSCP.
(define *binary-radix-sort*
  (or (get-environment-variable "RADIX_SORT_WORKER")
      "./radix-sort"))

;; blake3-fanout — Wave-2 form `cuda-blake3-tree`. Per-input BLAKE3,
;; one CUDA thread walks the Merkle chunk tree for its input. Binary
;; wire uses BSB3 request / BSR3 response magic, distinct from all
;; other forms (BSHK/BCGB/BSCP/BSRT). Primitives vendored from
;; Blaze-3/BLAKE3-gpu (MIT); driver AGPLv3.
(define *binary-blake3-fanout*
  (or (get-environment-variable "BLAKE3_FANOUT_WORKER")
      "./blake3-fanout"))

(define (parse-port-arg args)
  (let loop ((rest args))
    (cond
      ((null? rest) *worker-port*)
      ((null? (cdr rest)) *worker-port*)
      ((and (string? (car rest)) (string=? (car rest) "--port"))
       (or (string->number (car (cdr rest))) *worker-port*))
      (else (loop (cdr rest))))))

(define *daemons* '())    ; alist (op-name . (stdin-port . stdout-port))

;;; -- daemon pool (per-tier stubs marked) -----------------------
;;;
;;; spawn-process-stdio returns (stdin-port . stdout-port) for a
;;; long-running subprocess. Implementation varies by tier:
;;;   Python: subprocess.Popen with stdin/stdout=PIPE
;;;   C:      fork + pipe + exec + dup2
;;;   asm:    syscall fork + pipe + execve
;;;
;;; lumbda's fleet doesn't have it today; the first place it's
;;; needed is here. Should land in each tier's primitive table
;;; alongside `tcp-*`.

;; start-daemon optionally takes extra-args (a list of strings) appended
;; after "--daemon" in the spawn argv. Day-4: secp256k1 daemon flips to
;; pass '("--window-w" "4") because v3 (Day-3 windowed-G ladder) clean-
;; wins at n>=100k vs v1; see plans/form-A-day4-progress.md.
(define (start-daemon binary-path . extra-args)
  (let* ((argv (cons "--daemon"
                     (if (null? extra-args) '() (car extra-args))))
         (pair (spawn-process-stdio binary-path argv))
         (ready (read-line (cdr pair))))
    (if (string=? ready "ready")
        pair
        (error "daemon failed to ready:" ready))))

;; Send `process[-bin] in out` to daemon, await `done` or `error`.

(define (daemon-process daemon-pair in-portal out-portal use-bin?)
  (let ((cmd (if use-bin? "process-bin " "process ")))
    (display cmd (car daemon-pair))
    (display in-portal (car daemon-pair))
    (display " " (car daemon-pair))
    (display out-portal (car daemon-pair))
    (newline (car daemon-pair))
    (flush-port (car daemon-pair))
    (let ((line (read-line (cdr daemon-pair))))
      (cond
        ((eq? line #f) (cons 'error "daemon closed"))
        ((>= (string-length line) 5)
         (cond
           ((string=? (substring line 0 4) "done") 'ok)
           ((string=? (substring line 0 5) "error") (cons 'error line))
           (else (cons 'error (string-append "?: " line)))))
        (else (cons 'error (string-append "?: " line)))))))

(define (register-daemon! op-name binary-path . extra-args)
  (let ((pair (if (null? extra-args)
                  (start-daemon binary-path)
                  (start-daemon binary-path (car extra-args)))))
    (set! *daemons*
          (cons (cons op-name pair) *daemons*)))
  (display "gpu-worker: ready ") (display op-name)
  (display " <- ") (display binary-path)
  (cond
    ((not (null? extra-args))
     (display " extra-args=") (display (car extra-args))))
  (newline))

;;; -- op handler -- cuda-shake-fanout ----------------------------

(define (gensym-path prefix suffix)
  ;; Each tier already has current-time-ms or gensym; if neither is
  ;; present, fall back to a counter. Simple uniqueness only.
  (string-append prefix "-" (number->string (current-time-ms)) suffix))

(define (write-shake-input-portal! path inputs out-bytes)
  (let ((port (open-output-file path)))
    (display "(cuda-shake-fanout\n" port)
    (display "  (output-bytes " port) (display out-bytes port) (display ")\n" port)
    (display "  (inputs\n" port)
    (for-each (lambda (h)
                (display "    \"" port) (display h port) (display "\"\n" port))
              inputs)
    (display "))\n" port)
    (close-port port)))

(define (read-shake-output-portal path)
  (let ((sexp (read-from-string (file->string path))))
    (let loop ((children (cdr sexp)))
      (cond
        ((null? children) '())
        ((and (pair? (car children)) (eq? (car (car children)) 'hashes))
         (cdr (car children)))
        (else (loop (cdr children)))))))

;; If x looks like (quote (...)) -- bend's serialized form -- return

;; the inner list. Otherwise return x unchanged.

(define (unquote-list x)
  (cond
    ((and (pair? x) (eq? (car x) 'quote) (pair? (cdr x)))
     (car (cdr x)))
    (else x)))

(define (handle-cuda-shake-fanout args)
  (let* ((inputs (unquote-list (car args)))
         (out-bytes (car (cdr args)))
         (in-path (gensym-path "/tmp/bend-in" ".portal"))
         (out-path (gensym-path "/tmp/bend-out" ".portal"))
         (daemon (cdr (assoc 'cuda-shake-fanout *daemons*))))
    (write-shake-input-portal! in-path inputs out-bytes)
    (let ((status (daemon-process daemon in-path out-path #f)))
      (cond
        ((eq? status 'ok)
         (let ((result (read-shake-output-portal out-path)))
           (delete-file in-path)
           (delete-file out-path)
           (list 'ok result)))
        (else
         (delete-file in-path) (delete-file out-path)
         (list 'error (cdr status)))))))

;;; -- op handler -- cuda-sim-ops-bin -----------------------------
;;;
;;; Request: (cuda-sim-ops-bin <ops-bin-path> <n-batches>)
;;; Spawns bend-cuda <path> <n-batches> --portal <tmp>, drains its
;;; stdout to EOF (process exit), reads the portal, returns the
;;; parsed (cuda-sim-result ...) form unchanged so callers can pull
;;; out gates / phase / timings.

(define (drain-to-eof port)
  ;; Read until read-line returns #f (or eof-object). Discard lines.
  (let loop ()
    (let ((line (read-line port)))
      (cond
        ((eq? line #f) #t)
        ((eof-object? line) #t)
        (else (loop))))))

;; Pull a numeric value out of a parsed portal expression for telemetry.
;; Walks (a b c ...) looking for (name <number>); returns #f if missing.
(define (portal-find-number form name)
  ;; Guard: form may arrive as #f when read-from-string sees an empty
  ;; or malformed portal file (bend-cuda crashed mid-write, OOM, etc).
  ;; A bare (cdr #f) here historically crashed the gpu-worker with
  ;; "error: not a pair: #f", taking the port-8320 listener down.
  (cond
    ((not (pair? form)) #f)
    (else
     (let loop ((rest (cdr form)))
       (cond
         ((null? rest) #f)
         ((not (pair? rest)) #f)
         ((and (pair? (car rest)) (eq? (car (car rest)) name)
               (pair? (cdr (car rest))) (number? (car (cdr (car rest)))))
          (car (cdr (car rest))))
         (else (loop (cdr rest))))))))

;; Walk through (timing-ms (cpu-total X) (gpu-kernel Y)) shape.
(define (portal-timing form which)
  (cond
    ((not (pair? form)) #f)
    (else
     (let loop ((rest (cdr form)))
       (cond
         ((null? rest) #f)
         ((not (pair? rest)) #f)
         ((and (pair? (car rest)) (eq? (car (car rest)) 'timing-ms))
          (portal-find-number (car rest) which))
         (else (loop (cdr rest))))))))

(define (handle-cuda-sim-ops-bin args)
  (let* ((ops-path (car args))
         (n-batches (car (cdr args)))
         (portal-path (gensym-path "/tmp/bend-sim-ops" ".portal"))
         (t-start (current-time-ms)))
    (cond
      ((not (file-exists? ops-path))
       (display ";;; bend ERROR cuda-sim-ops-bin ops-bin-missing ")
       (display ops-path) (newline)
       (list 'error (list 'ops-bin-missing ops-path)))
      (else
       (display ";;; bend RECV cuda-sim-ops-bin ops=")
       (display ops-path)
       (display " n-batches=") (display n-batches)
       (display " t-ms=") (display t-start) (newline)
       (let* ((argv (list ops-path
                          (number->string n-batches)
                          "--portal" portal-path))
              (pair (spawn-process-stdio *binary-bend-cuda* argv)))
         ;; bend-cuda does not read stdin; drain stdout until exit.
         (drain-to-eof (cdr pair))
         (close-port (car pair))
         (close-port (cdr pair))
         (let ((wall-ms (- (current-time-ms) t-start)))
           (cond
             ((file-exists? portal-path)
              (let* ((result (read-from-string (file->string portal-path)))
                     (cpu-ms (portal-timing result 'cpu-total))
                     (gpu-ms (portal-timing result 'gpu-kernel))
                     (mismatches (portal-find-number result 'mismatches)))
                (delete-file portal-path)
                (display ";;; bend DONE cuda-sim-ops-bin")
                (display " n-batches=") (display n-batches)
                (display " wall-ms=") (display wall-ms)
                (display " cpu-ms=") (display (or cpu-ms 'NA))
                (display " gpu-ms=") (display (or gpu-ms 'NA))
                (display " mismatches=") (display (or mismatches 'NA))
                (cond
                  ((and (number? cpu-ms) (number? gpu-ms) (> gpu-ms 0))
                   (display " gpu/cpu=")
                   (display (/ cpu-ms gpu-ms))))
                (newline)
                (list 'ok result)))
             (else
              (display ";;; bend FAIL cuda-sim-ops-bin no-portal wall-ms=")
              (display wall-ms) (newline)
              (list 'error (list 'no-portal portal-path))))))))))

;;; -- op handler -- health --------------------------------------
;;;
;;; (health) returns
;;;   (ok (load-avg L) (vram-free-mb V) (uptime-ms U))
;;;
;;; L = 1-minute load average parsed from /proc/loadavg
;;; V = MB of free VRAM from `nvidia-smi --query-gpu=memory.free`
;;; U = current-time-ms (millisecond wallclock, lets a client detect
;;;     a worker that hung & restarted between probes)
;;;
;;; Cheap (sub-millisecond load-avg + a single nvidia-smi exec for
;;; VRAM). Cache TTL on a client makes per-dispatch polling free.

(define (parse-leading-number s)
  ;; Longest digit-or-decimal-point run from index 0; 0 on failure.
  (let* ((n (string-length s))
         (end (let scan ((i 0))
                (cond
                  ((>= i n) i)
                  ((let ((c (string-ref s i)))
                     (or (char-numeric? c) (char=? c #\.)))
                   (scan (+ i 1)))
                  (else i)))))
    (cond
      ((= end 0) 0)
      (else (or (string->number (substring s 0 end)) 0)))))

(define (health-load-avg)
  (let ((s (file->string "/proc/loadavg")))
    (cond
      ((eq? s #f) 0)
      (else (parse-leading-number s)))))

(define (health-vram-free-mb)
  ;; nvidia-smi --query-gpu=memory.free --format=csv,noheader,nounits
  ;; one number per GPU, one per line; first line suffices.
  ;; Returns 0 when nvidia-smi missing (host without an NVIDIA GPU)
  ;; or when the spawn output cannot be parsed — never raises.
  (cond
    ((not (file-exists? "/usr/bin/nvidia-smi")) 0)
    (else
     (let ((pair (spawn-process-stdio "/usr/bin/nvidia-smi"
                   '("--query-gpu=memory.free"
                     "--format=csv,noheader,nounits"))))
       (cond
         ((eq? pair #f) 0)
         (else
          (let ((line (read-line (cdr pair))))
            (close-port (car pair))
            (close-port (cdr pair))
            (cond
              ((eq? line #f) 0)
              ((eof-object? line) 0)
              (else (parse-leading-number line))))))))))

;;; -- pool / queue health -------------------------------------------
;;;
;;; bend co-lives with an bend-emit-pool on each foxhop production host.
;;; When the pool dies but bend stays up, .lsp cells pile up un-emitted
;;; and bend sits idle waiting for .ready bins that never arrive. The
;;; feeder daemon polls bend's (health) to know whether to push more
;;; work — adding pool/queue counts here means a single RPC tells the
;;; feeder everything: bend health, pool health, queue depth.
;;;
;;; BEND_QUEUE_DIR env var — set when bend is launched on a host with
;;; an associated pool. Absent on hosts that only run bend without a
;;; pool. When absent, queue counts return -1 (caller treats as
;;; "unknown / not applicable").

(define (run-and-count cmd-args)
  ;; Spawn cmd-args, read all stdout lines, return integer line count.
  ;; Returns 0 on spawn failure. Used to count ls / pgrep output.
  (cond
    ((not (file-exists? (car cmd-args))) 0)
    (else
     (let ((pair (spawn-process-stdio (car cmd-args) (cdr cmd-args))))
       (cond
         ((eq? pair #f) 0)
         (else
          (let loop ((count 0))
            (let ((line (read-line (cdr pair))))
              (cond
                ((or (eq? line #f) (eof-object? line))
                 (close-port (car pair))
                 (close-port (cdr pair))
                 count)
                (else (loop (+ count 1))))))))))))

(define (health-pool-procs)
  ;; Count of bend-emit-pool processes on this host. 0 = pool dead.
  (let* ((pair (spawn-process-stdio "/usr/bin/pgrep"
                 '("-cf" "bend-emit-pool"))))
    (cond
      ((eq? pair #f) 0)
      (else
       (let ((line (read-line (cdr pair))))
         (close-port (car pair))
         (close-port (cdr pair))
         (cond
           ((or (eq? line #f) (eof-object? line)) 0)
           (else (or (string->number (parse-trim line)) 0))))))))

(define (health-dispatcher-procs)
  ;; Count of bend-dispatcher processes on this host. <=1 = dispatcher
  ;; functionally dead (only stale wrapper bash matched). Live
  ;; multi-dispatcher = parent + 2 worker forks = ≥3 procs.
  (let* ((pair (spawn-process-stdio "/usr/bin/pgrep"
                 '("-cf" "bend-dispatcher"))))
    (cond
      ((eq? pair #f) 0)
      (else
       (let ((line (read-line (cdr pair))))
         (close-port (car pair))
         (close-port (cdr pair))
         (cond
           ((or (eq? line #f) (eof-object? line)) 0)
           (else (or (string->number (parse-trim line)) 0))))))))

(define (parse-trim s)
  ;; Strip leading/trailing whitespace from a single-line string.
  (let* ((n (string-length s))
         (lo (let ll ((i 0))
               (cond
                 ((>= i n) i)
                 ((char-whitespace? (string-ref s i)) (ll (+ i 1)))
                 (else i))))
         (hi (let lh ((i (- n 1)))
               (cond
                 ((< i lo) lo)
                 ((char-whitespace? (string-ref s i)) (lh (- i 1)))
                 (else (+ i 1))))))
    (substring s lo hi)))

(define (health-queue-count ext)
  ;; Count of files matching $BEND_QUEUE_DIR/*.<ext>. Returns -1 when
  ;; BEND_QUEUE_DIR env is unset (caller treats as unknown).
  (let ((qdir (get-environment-variable "BEND_QUEUE_DIR")))
    (cond
      ((or (eq? qdir #f) (string=? qdir "")) -1)
      (else
       (let* ((pattern (string-append qdir "/*." ext))
              (pair (spawn-process-stdio "/bin/sh"
                      (list "-c"
                            (string-append "ls " pattern " 2>/dev/null | wc -l")))))
         (cond
           ((eq? pair #f) 0)
           (else
            (let ((line (read-line (cdr pair))))
              (close-port (car pair))
              (close-port (cdr pair))
              (cond
                ((or (eq? line #f) (eof-object? line)) 0)
                (else (or (string->number (parse-trim line)) 0)))))))))))

(define (health-feeder-paused)
  ;; 1 if $BEND_QUEUE_DIR/FEEDER_PAUSE marker present (operator wants
  ;; this host out of the rotation; feeder should not push or restart
  ;; supervisors here). 0 if no marker. -1 if BEND_QUEUE_DIR unset.
  (let ((qdir (get-environment-variable "BEND_QUEUE_DIR")))
    (cond
      ((or (eq? qdir #f) (string=? qdir "")) -1)
      ((file-exists? (string-append qdir "/FEEDER_PAUSE")) 1)
      (else 0))))

(define (handle-health)
  (list 'ok
        (list 'load-avg (health-load-avg))
        (list 'vram-free-mb (health-vram-free-mb))
        (list 'uptime-ms (current-time-ms))
        (list 'pool-procs (health-pool-procs))
        (list 'dispatcher-procs (health-dispatcher-procs))
        (list 'queue-ready (health-queue-count "ready"))
        (list 'queue-emitting (health-queue-count "emitting"))
        (list 'queue-done (health-queue-count "done"))
        (list 'feeder-paused (health-feeder-paused))))

;;; -- dispatch --------------------------------------------------

(define (handle-request sexp)
  (cond
    ((not (pair? sexp)) (list 'error "not a form"))
    (else
     (let ((op (car sexp)) (args (cdr sexp)))
       (cond
         ((eq? op 'cuda-shake-fanout) (handle-cuda-shake-fanout args))
         ((eq? op 'cuda-sim-ops-bin) (handle-cuda-sim-ops-bin args))
         ((eq? op 'health) (handle-health))
         ((eq? op 'ping) (list 'ok 'pong))
         (else (list 'error (list 'unknown-op op))))))))

;; Binary wire mode: payload starts with magic "BSHK" then a
;; daemon-binary-portal blob (u32 out_bytes | u32 n | n x (u32 len + bytes)).
;; Worker writes the blob to disk, calls daemon process-bin, reads the
;; binary result, prepends "BSHR" magic, wire-send-raws it back.

(define (handle-binary-shake client payload)
  (let* ((daemon (cdr (assoc 'cuda-shake-fanout *daemons*)))
         (in-path (gensym-path "/tmp/bend-bin-in" ".bin"))
         (out-path (gensym-path "/tmp/bend-bin-out" ".bin"))
         (portal-blob (substring payload 4 (string-length payload))))
    (write-binary-file in-path portal-blob)
    (let ((status (daemon-process daemon in-path out-path #t)))
      (cond
        ((eq? status 'ok)
         (let ((result-blob (read-binary-file out-path)))
           (if (file-exists? in-path) (delete-file in-path))
           (if (file-exists? out-path) (delete-file out-path))
           (wire-send-raw client (string-append "BSHR" result-blob))))
        (else
         (if (file-exists? in-path) (delete-file in-path))
         (if (file-exists? out-path) (delete-file out-path))
         (wire-send-raw client (string-append "BERR" (cdr status))))))))

;; Binary wire for bend form B (cuda-bignum-cgbn).
;; Payload begins with "BCGB"; pass entire blob through to the daemon,
;; which expects the same magic + header it received from the client.
(define (handle-binary-cgbn client payload)
  (let* ((daemon (cdr (assoc 'cuda-bignum-cgbn *daemons*)))
         (in-path (gensym-path "/tmp/bend-cgbn-in" ".bin"))
         (out-path (gensym-path "/tmp/bend-cgbn-out" ".bin"))
         (t-start (current-time-ms)))
    (write-binary-file in-path payload)
    (display ";;; bend RECV cuda-bignum-cgbn bytes=")
    (display (string-length payload))
    (display " t-ms=") (display t-start) (newline)
    (let ((status (daemon-process daemon in-path out-path #t)))
      (let ((wall-ms (- (current-time-ms) t-start)))
        (cond
          ((eq? status 'ok)
           (let ((result-blob (read-binary-file out-path)))
             (if (file-exists? in-path) (delete-file in-path))
             (if (file-exists? out-path) (delete-file out-path))
             (display ";;; bend DONE cuda-bignum-cgbn")
             (display " wall-ms=") (display wall-ms)
             (display " out-bytes=") (display (string-length result-blob))
             (newline)
             (wire-send-raw client result-blob)))
          (else
           (display ";;; bend FAIL cuda-bignum-cgbn wall-ms=")
           (display wall-ms) (newline)
           (if (file-exists? in-path) (delete-file in-path))
           (if (file-exists? out-path) (delete-file out-path))
           (wire-send-raw client (string-append "BERR" (cdr status)))))))))

;; Binary wire for bend form A (cuda-secp256k1-batched-mul).
;; Payload begins with "BSCP"; pass entire blob through to the daemon,
;; which expects the same magic + header. Response payload begins
;; with "BSCR" on success or "BERR" on failure.
(define (handle-binary-secp client payload)
  (let* ((daemon (cdr (assoc 'cuda-secp256k1-batched-mul *daemons*)))
         (in-path (gensym-path "/tmp/bend-secp-in" ".bin"))
         (out-path (gensym-path "/tmp/bend-secp-out" ".bin"))
         (t-start (current-time-ms)))
    (write-binary-file in-path payload)
    (display ";;; bend RECV cuda-secp256k1-batched-mul bytes=")
    (display (string-length payload))
    (display " t-ms=") (display t-start) (newline)
    (let ((status (daemon-process daemon in-path out-path #t)))
      (let ((wall-ms (- (current-time-ms) t-start)))
        (cond
          ((eq? status 'ok)
           (let ((result-blob (read-binary-file out-path)))
             (if (file-exists? in-path) (delete-file in-path))
             (if (file-exists? out-path) (delete-file out-path))
             (display ";;; bend DONE cuda-secp256k1-batched-mul")
             (display " wall-ms=") (display wall-ms)
             (display " out-bytes=") (display (string-length result-blob))
             (newline)
             (wire-send-raw client result-blob)))
          (else
           (display ";;; bend FAIL cuda-secp256k1-batched-mul wall-ms=")
           (display wall-ms) (newline)
           (if (file-exists? in-path) (delete-file in-path))
           (if (file-exists? out-path) (delete-file out-path))
           (wire-send-raw client (string-append "BERR" (cdr status)))))))))

;; Binary wire for Wave-2 form `cuda-blake3-tree` (BLAKE3 batched fan-out).
;; Payload begins with "BSB3". The .cu daemon (blake3-fanout.cu) expects
;; the same magic + header in its on-disk binary file (matches the BCGB /
;; BSCP / BSRT pattern, not the BSHK strip-magic pattern), so we pass the
;; entire payload through verbatim. The daemon's response file already
;; begins with "BSR3"; pass through unchanged.
(define (handle-binary-blake3 client payload)
  (let* ((daemon (cdr (assoc 'cuda-blake3-tree *daemons*)))
         (in-path (gensym-path "/tmp/bend-blake3-in" ".bin"))
         (out-path (gensym-path "/tmp/bend-blake3-out" ".bin"))
         (t-start (current-time-ms)))
    (write-binary-file in-path payload)
    (display ";;; bend RECV cuda-blake3-tree bytes=")
    (display (string-length payload))
    (display " t-ms=") (display t-start) (newline)
    (let ((status (daemon-process daemon in-path out-path #t)))
      (let ((wall-ms (- (current-time-ms) t-start)))
        (cond
          ((eq? status 'ok)
           (let ((result-blob (read-binary-file out-path)))
             (if (file-exists? in-path) (delete-file in-path))
             (if (file-exists? out-path) (delete-file out-path))
             (display ";;; bend DONE cuda-blake3-tree")
             (display " wall-ms=") (display wall-ms)
             (display " out-bytes=") (display (string-length result-blob))
             (newline)
             (wire-send-raw client result-blob)))
          (else
           (display ";;; bend FAIL cuda-blake3-tree wall-ms=")
           (display wall-ms) (newline)
           (if (file-exists? in-path) (delete-file in-path))
           (if (file-exists? out-path) (delete-file out-path))
           (wire-send-raw client (string-append "BERR" (cdr status)))))))))

;; Binary wire for bend form G (cuda-radix-sort).
;; Payload begins with "BSRT"; pass entire blob through to the daemon,
;; which expects the same magic + header. Response begins with "BSRR"
;; on success or "BERR" on failure.
(define (handle-binary-sort client payload)
  (let* ((daemon (cdr (assoc 'cuda-radix-sort *daemons*)))
         (in-path (gensym-path "/tmp/bend-sort-in" ".bin"))
         (out-path (gensym-path "/tmp/bend-sort-out" ".bin"))
         (t-start (current-time-ms)))
    (write-binary-file in-path payload)
    (display ";;; bend RECV cuda-radix-sort bytes=")
    (display (string-length payload))
    (display " t-ms=") (display t-start) (newline)
    (let ((status (daemon-process daemon in-path out-path #t)))
      (let ((wall-ms (- (current-time-ms) t-start)))
        (cond
          ((eq? status 'ok)
           (let ((result-blob (read-binary-file out-path)))
             (if (file-exists? in-path) (delete-file in-path))
             (if (file-exists? out-path) (delete-file out-path))
             (display ";;; bend DONE cuda-radix-sort")
             (display " wall-ms=") (display wall-ms)
             (display " out-bytes=") (display (string-length result-blob))
             (newline)
             (wire-send-raw client result-blob)))
          (else
           (display ";;; bend FAIL cuda-radix-sort wall-ms=")
           (display wall-ms) (newline)
           (if (file-exists? in-path) (delete-file in-path))
           (if (file-exists? out-path) (delete-file out-path))
           (wire-send-raw client (string-append "BERR" (cdr status)))))))))

;; Accept one client, handle one request, close. Returns #t to keep
;; serving, #f when the server should stop.
(define (handle-one server)
  (let ((client (tcp-accept server)))
    (cond
      ((eq? client #f) #t)
      (else
       (let ((payload (wire-recv-raw client)))
         (cond
           ((eq? payload #f) (tcp-close client) #t)
           ((and (>= (string-length payload) 4)
                 (string=? (substring payload 0 4) "BSHK"))
            (handle-binary-shake client payload)
            (tcp-close client) #t)
           ((and (>= (string-length payload) 4)
                 (string=? (substring payload 0 4) "BCGB"))
            (handle-binary-cgbn client payload)
            (tcp-close client) #t)
           ((and (>= (string-length payload) 4)
                 (string=? (substring payload 0 4) "BSCP"))
            (handle-binary-secp client payload)
            (tcp-close client) #t)
           ((and (>= (string-length payload) 4)
                 (string=? (substring payload 0 4) "BSRT"))
            (handle-binary-sort client payload)
            (tcp-close client) #t)
           ((and (>= (string-length payload) 4)
                 (string=? (substring payload 0 4) "BSB3"))
            (handle-binary-blake3 client payload)
            (tcp-close client) #t)
           (else
            (let* ((req (read-from-string payload))
                   (resp (handle-request req)))
              (wire-send client resp)
              (tcp-close client)
              #t))))))))

;;; -- VRAM-aware admission control ------------------------------
;;;
;;; bend-cuda holds ~bin-size GB of VRAM during its run (per
;;; foxhop empirical: bin file size on disk ≈ peak VRAM used by
;;; bend-cuda loading + running that bin).
;;;
;;; DYNAMIC algorithm: query running EMA per bend-form (op kind),
;;; combine with current nvidia-smi reading + projected new-cell
;;; cost. Fork iff (current_vram + projected_cell_vram + safety) < gpu_total.
;;; No static max-children cap — the algorithm picks N based on
;;; observed per-form VRAM usage. Larger cells = fewer concurrent;
;;; tiny cells = many concurrent. Self-tuning to each bend form.

(define *gpu-total-mib*
  (let ((env (get-environment-variable "LUMBDA_GPU_TOTAL_MIB")))
    (cond
      ((and env (> (string-length env) 0)) (string->number env))
      (else 24576))))   ; 24 GB default (RTX 3090)

(define *gpu-headroom-mib*
  (let ((env (get-environment-variable "LUMBDA_GPU_HEADROOM_MIB")))
    (cond
      ((and env (> (string-length env) 0)) (string->number env))
      (else 1024))))    ; 1 GB safety pad

;; Running max per-cell VRAM observed. Conservative — uses largest
;; bin seen, not average. Single-slot mutable. Initial 4096 MiB =
;; pessimistic seed until we observe real cells.
;; Seed lowered 4096→2500 (2026-06-11) — real K=2 bin avg is ~2.4GB.
;; Overprojection at 4096 throttled admit to 2-3 concurrent for big bins
;; even when 4-5 would fit; lower seed lets autoscaler use full GPU headroom.
;; record-cell-vram! in child still updates running max if larger seen.
(define *vram-per-cell-max-mib* 2500)

(define (vram-used-mib)
  ;; nvidia-smi --query-gpu=memory.used --format=csv,noheader,nounits
  ;; returns one integer line per GPU; sum if multi-GPU host.
  (let* ((pair (spawn-process-stdio
                "/usr/bin/nvidia-smi"
                '("--query-gpu=memory.used"
                  "--format=csv,noheader,nounits"))))
    (cond
      ((eq? pair #f) 0)  ;; fail-open: assume 0 if nvidia-smi missing
      (else
       (close-port (car pair))  ;; close stdin to child
       (let loop ((sum 0))
         (let ((line (read-line (cdr pair))))
           (cond
             ((eof-object? line) (close-port (cdr pair)) sum)
             (else
              (let ((n (string->number (string-trim line))))
                (loop (+ sum (cond (n n) (else 0)))))))))))))

(define (string-trim s)
  ;; trim leading + trailing whitespace.
  (let* ((len (string-length s))
         (start (let loop ((i 0))
                  (cond ((>= i len) len)
                        ((char-whitespace? (string-ref s i)) (loop (+ i 1)))
                        (else i))))
         (end (let loop ((i (- len 1)))
                (cond ((< i start) start)
                      ((char-whitespace? (string-ref s i)) (loop (- i 1)))
                      (else (+ i 1))))))
    (substring s start end)))

(define (file-size-mib path)
  ;; Stat a bin path; return its size in MiB. Returns 0 on missing
  ;; file (fail-open). Used as cheap proxy for per-cell VRAM cost
  ;; before forking — bin file on disk ≈ peak VRAM used by bend-cuda
  ;; loading + running that bin (per foxhop empirical).
  (let* ((pair (spawn-process-stdio
                "/usr/bin/stat" (list "-c" "%s" path))))
    (cond
      ((eq? pair #f) 0)
      (else
       (close-port (car pair))
       (let ((line (read-line (cdr pair))))
         (close-port (cdr pair))
         (cond
           ((or (eof-object? line) (not line)) 0)
           (else
            (let ((n (string->number (string-trim line))))
              (cond ((not n) 0)
                    (else (quotient n 1048576))))))))))) ; bytes → MiB

(define (admit-fork? projected-cell-mib)
  ;; Dynamic admit decision: fork iff current VRAM + projected cell
  ;; + safety pad < total GPU memory. Self-tuning to per-cell sizes
  ;; observed (caller updates *vram-per-cell-max-mib* after each fork).
  (let* ((used (vram-used-mib))
         (need (+ used projected-cell-mib *gpu-headroom-mib*)))
    (< need *gpu-total-mib*)))

(define (wait-admit projected-cell-mib)
  ;; Block until admission lets the next fork through. Reap zombies
  ;; while waiting so they don't pile up.
  (let loop ()
    (waitpid-nonblock)
    (cond
      ((admit-fork? projected-cell-mib) #t)
      (else (sleep 1) (loop)))))

(define (record-cell-vram! observed-mib)
  ;; Track max observed cell VRAM. Conservative — uses max not mean,
  ;; so admission stays safe under heterogeneous workloads.
  (cond
    ((> observed-mib *vram-per-cell-max-mib*)
     (set! *vram-per-cell-max-mib* observed-mib))))

;;; -- entry -----------------------------------------------------

;;; run-loop — fork-per-accept pattern.
;;;
;;; Accept connection → check VRAM budget → fork child handler →
;;; parent reaps zombies + returns to accept. Child runs handle-one
;;; (which spawns bend-cuda + waits + responds) then exits — its
;;; bend-cuda subprocess holds VRAM for the duration of that one
;;; request, naturally isolated from sibling children.
;;;
;;; This gives concurrent dispatch on a single PID without lumbda
;;; needing threading primitives. Linux COW handles parent->child
;;; memory; OS scheduler distributes across cores.
(define (run-loop server)
  (waitpid-nonblock)               ;; reap any completed child
  (wait-admit *vram-per-cell-max-mib*)  ;; dynamic: admit if room for largest seen cell
  (let ((client (tcp-accept server)))
    (cond
      ((eq? client #f) (run-loop server))
      (else
       (let ((pid (fork-self)))
         (cond
           ((eq? pid #f)
            ;; fork failed — fall back to serial handle
            (handle-one-client client server)
            (run-loop server))
           ((eq? pid 0)
            ;; child: handle this one client, then exit
            (handle-one-client client server)
            (exit 0))
           (else
            ;; parent: close our copy of client fd, loop to accept
            (tcp-close client)
            (run-loop server))))))))

;;; handle-one-client — same dispatch as the old handle-one body but
;;; takes the already-accepted client port as arg (no accept call).
;;;
;;; After reading the request, update *vram-per-cell-max-mib* with the
;;; bin file size as a proxy for per-cell VRAM. Self-tunes admission
;;; for subsequent forks. NOTE: this runs IN THE CHILD process; the
;;; parent's *vram-per-cell-max-mib* won't see it (forks don't share
;;; memory). For the algorithm to learn cross-fork, the parent needs
;;; to read the bin size pre-fork (TODO: peek request before forking,
;;; or have child write a small note for parent to read).
(define (handle-one-client client server)
  (let ((payload (wire-recv-raw client)))
    (cond
      ((eq? payload #f) (tcp-close client))
      ((and (>= (string-length payload) 4)
            (string=? (substring payload 0 4) "BSHK"))
       (handle-binary-shake client payload)
       (tcp-close client))
      ((and (>= (string-length payload) 4)
            (string=? (substring payload 0 4) "BCGB"))
       (handle-binary-cgbn client payload)
       (tcp-close client))
      ((and (>= (string-length payload) 4)
            (string=? (substring payload 0 4) "BSCP"))
       (handle-binary-secp client payload)
       (tcp-close client))
      ((and (>= (string-length payload) 4)
            (string=? (substring payload 0 4) "BSRT"))
       (handle-binary-sort client payload)
       (tcp-close client))
      ((and (>= (string-length payload) 4)
            (string=? (substring payload 0 4) "BSB3"))
       (handle-binary-blake3 client payload)
       (tcp-close client))
      (else
       (let* ((req (read-from-string payload))
              (resp (handle-request req)))
         (wire-send client resp)
         (tcp-close client))))))

;; Optional registration — only spawn the daemon when its binary is
;; reachable. Lets a worker host serve a subset of forms without
;; failing to start because some bend form's daemon isn't installed.
(define (maybe-register-daemon! op-name binary-path . extra-args)
  (cond
    ((file-exists? binary-path)
     (if (null? extra-args)
         (register-daemon! op-name binary-path)
         (register-daemon! op-name binary-path (car extra-args))))
    (else
     (display ";;; gpu-worker: skipping ") (display op-name)
     (display " - binary not found at ") (display binary-path)
     (newline))))

;; Day-4 daemon flag set for the secp256k1 binary. v3 (--window-w 4)
;; landed 13.83 Mkeys/s @ n=1M (1.73x v1, 1.55x v4). Day-4 v4 stack
;; regressed -12% vs v3; daemon stays on v3 until warp-scan Phase B/D
;; lands (see plans/form-A-day4-progress.md).
(define *secp-daemon-extra-args* '("--window-w" "4"))

(define (main)
  (let ((port (parse-port-arg *argv*)))
    (set! *worker-port* port)
    (maybe-register-daemon! 'cuda-shake-fanout *binary-shake-fanout*)
    (maybe-register-daemon! 'cuda-bignum-cgbn *binary-cgbn-batch*)
    (maybe-register-daemon! 'cuda-secp256k1-batched-mul
                            *binary-secp256k1-batch*
                            *secp-daemon-extra-args*)
    (maybe-register-daemon! 'cuda-radix-sort *binary-radix-sort*)
    (maybe-register-daemon! 'cuda-blake3-tree *binary-blake3-fanout*)
    (let ((server (tcp-listen port)))
      (cond
        ((eq? server #f)
         (display ";;; ERROR -- tcp-listen failed on port ")
         (display port) (newline))
        (else
         (display "gpu-worker listening on port ")
         (display port) (newline)
         (run-loop server))))))

;; (main)  ; uncomment to run; needs the cuda-shake-fanout binary
;;            on disk + spawn-process-stdio primitive per tier
