flow: store hygiene flow/gc + flow/forget + 9 tests

flow/gc drops terminal (done/cancelled) records, keeps live suspended flows, returns
count removed; flow/forget id drops one terminal record and refuses live flows.
Bounds unbounded store growth (retention/GC). Bumped conformance sx_server timeout
to 540s for the 10-suite run under CPU contention. 151/151 across 10 suites.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

lib/flow/conformance.sh

@@ -30,6 +30,7 @@ SUITES=(
   "combinators flow-cmb-tests-run!       lib/flow/tests/combinators.sx"
   "railway     flow-rail-tests-run!      lib/flow/tests/railway.sx"
   "integration flow-int-tests-run!       lib/flow/tests/integration.sx"
+  "hygiene     flow-hyg-tests-run!       lib/flow/tests/hygiene.sx"
 )
 
 TMPFILE=$(mktemp); trap "rm -f $TMPFILE" EXIT
@@ -56,7 +57,7 @@ emit_eval () { echo "(epoch $EPOCH)"; echo "(eval \"$1\")"; EPOCH=$((EPOCH+1));
   done
 } > "$TMPFILE"
 
-OUTPUT=$(timeout 300 "$SX_SERVER" < "$TMPFILE" 2>&1 || true)
+OUTPUT=$(timeout 540 "$SX_SERVER" < "$TMPFILE" 2>&1 || true)
 
 TOTAL_PASS=0
 TOTAL_FAIL=0

lib/flow/scoreboard.json

@@ -1,6 +1,6 @@
 {
-  "total": 142,
+  "total": 151,
-  "passed": 142,
+  "passed": 151,
   "failed": 0,
   "suites": {
     "basic": { "passed": 18, "total": 18 },
@@ -11,7 +11,8 @@
     "api": { "passed": 12, "total": 12 },
     "combinators": { "passed": 17, "total": 17 },
     "railway": { "passed": 10, "total": 10 },
-    "integration": { "passed": 10, "total": 10 }
+    "integration": { "passed": 10, "total": 10 },
+    "hygiene": { "passed": 9, "total": 9 }
   },
   "phases": { "phase1": "done", "phase2": "done", "phase3": "done", "phase4": "done", "phase5": "done", "phase6": "done", "phase7": "done" }
 }

lib/flow/scoreboard.md

@@ -1,6 +1,6 @@
 # flow-on-sx Scoreboard
 
-**All tests pass: 142 / 142 across 9 suites. Phases 1-7 complete.**
+**All tests pass: 151 / 151 across 10 suites. Phases 1-7 complete.**
 
 `bash lib/flow/conformance.sh`
 
@@ -17,6 +17,7 @@
 | combinators | 17 | Phase 5: `tap`, `recover` (fail-value), `map-flow` fan-over-list, `flow-while`/`flow-until` bounded iteration |
 | railway | 10 | Phase 6: `attempt` — fail-value short-circuiting sequence + recover rejoin |
 | integration | 10 | Phase 7: end-to-end order + onboarding flows composing every phase (suspend, branch, federation, crash recovery, handoff, introspection) |
+| hygiene | 9 | Phase 5: `flow/gc` (prune terminal flows), `flow/forget` (drop one terminal record) |
 
 ## Architecture
 

lib/flow/tests/hygiene.sx

@@ -0,0 +1,67 @@
+;; lib/flow/tests/hygiene.sx — Phase 5: store hygiene (flow/gc, flow/forget).
+
+(define flow-hyg-pass 0)
+(define flow-hyg-fail 0)
+(define flow-hyg-fails (list))
+
+(define
+  flow-hyg-test
+  (fn
+    (name actual expected)
+    (if
+      (= actual expected)
+      (set! flow-hyg-pass (+ flow-hyg-pass 1))
+      (begin
+        (set! flow-hyg-fail (+ flow-hyg-fail 1))
+        (append! flow-hyg-fails {:name name :expected expected :actual actual})))))
+
+(define flow-h (fn (src) (flow-run src)))
+
+;; ── flow/gc ─────────────────────────────────────────────────────
+(flow-hyg-test
+  "gc: empty store removes nothing"
+  (flow-h "(flow/gc)")
+  0)
+(flow-hyg-test
+  "gc: removes a done flow, keeps a suspended one"
+  (flow-h
+    "(defflow w (lambda (x) (suspend (quote q)))) (flow/start w 0) (flow/start (lambda (x) x) 5) (define removed (flow/gc)) (list removed (flow/list))")
+  (list 1 (list (list 1 "suspended"))))
+(flow-hyg-test
+  "gc: removes a cancelled flow"
+  (flow-h
+    "(defflow w (lambda (x) (suspend (quote q)))) (define id (car (cdr (flow/start w 0)))) (flow/cancel id) (flow/gc)")
+  1)
+(flow-hyg-test
+  "gc: a kept suspended flow is still resumable"
+  (flow-h
+    "(defflow w (sequence (lambda (x) (suspend (quote q))) (lambda (v) (* v 2)))) (define id (car (cdr (flow/start w 0)))) (flow/start (lambda (x) x) 1) (flow/gc) (flow/resume id 21)")
+  42)
+(flow-hyg-test
+  "gc: counts every terminal flow it drops"
+  (flow-h
+    "(flow/start (lambda (x) x) 1) (flow/start (lambda (x) x) 2) (defflow w (lambda (x) (suspend (quote q)))) (flow/start w 0) (flow/gc)")
+  2)
+
+;; ── flow/forget ─────────────────────────────────────────────────
+(flow-hyg-test
+  "forget: drops a completed flow"
+  (flow-h
+    "(defflow w (sequence (lambda (x) (suspend (quote q))) (lambda (v) v))) (define id (car (cdr (flow/start w 0)))) (flow/resume id 7) (list (flow/forget id) (flow/status id))")
+  (list true "unknown"))
+(flow-hyg-test
+  "forget: refuses to drop a live (suspended) flow"
+  (flow-h
+    "(defflow w (lambda (x) (suspend (quote q)))) (define id (car (cdr (flow/start w 0)))) (list (flow/forget id) (flow/status id))")
+  (list false "suspended"))
+(flow-hyg-test
+  "forget: drops a cancelled flow"
+  (flow-h
+    "(defflow w (lambda (x) (suspend (quote q)))) (define id (car (cdr (flow/start w 0)))) (flow/cancel id) (list (flow/forget id) (flow/status id))")
+  (list true "unknown"))
+(flow-hyg-test
+  "forget: unknown id yields false"
+  (flow-h "(flow/forget 999)")
+  false)
+
+(define flow-hyg-tests-run! (fn () {:total (+ flow-hyg-pass flow-hyg-fail) :passed flow-hyg-pass :failed flow-hyg-fail :fails flow-hyg-fails}))

plans/flow-on-sx.md

@@ -16,7 +16,7 @@ federation extension via fed-sx for remote-node execution.
 
 ## Status (rolling)
 
-`bash lib/flow/conformance.sh` → **142/142** (Phases 1-7 complete)
+`bash lib/flow/conformance.sh` → **151/151** (Phases 1-7 complete; +store hygiene)
 
 ## Ground rules
 
@@ -141,6 +141,9 @@ something operators and authors actually use. Accumulation, not a rewrite.
 - [x] introspection API — `flow/status id`, `flow/result id`, `flow/list`,
   `flow/pending` (operator view of what each suspended flow awaits). 12 tests in
   `tests/api.sx`.
+- [x] store hygiene — `flow/gc` drops terminal (done/cancelled) records keeping
+  live suspended flows (returns count); `flow/forget id` drops one terminal record
+  and refuses live flows. Bounds unbounded store growth. 9 tests in `tests/hygiene.sx`.
 - [x] `tap` — side-effecting pass-through node (logging/metrics) that returns input
 - [x] `recover` — complement to try-catch for the fail-VALUE channel: run node; if it
   yields `(fail ...)`, run a recovery node on the reason