giles
e896deffc8
Some checks failed
Test, Build, and Deploy / test-build-deploy (push) Failing after 50s
Guest Scheme call/cc is escape-only (re-entry hangs), so durable resume uses deterministic replay: suspend escapes to the driver; resume re-runs the flow and replays resolved suspends from a (tag value) log. No live continuation is ever serialized — persisted state is plain data, survives restart. Adds flow/start (now state-returning, backward compatible), flow/resume, flow/cancel, store.sx. Harness reuses one env with a per-test reset (full env rebuild 66x was too slow). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
115 lines
5.3 KiB
Plaintext
115 lines
5.3 KiB
Plaintext
;; lib/flow/tests/suspend.sx — Phase 3: suspend / resume / cancel (deterministic replay).
|
|
|
|
(define flow-sus-pass 0)
|
|
(define flow-sus-fail 0)
|
|
(define flow-sus-fails (list))
|
|
|
|
(define
|
|
flow-sus-test
|
|
(fn
|
|
(name actual expected)
|
|
(if
|
|
(= actual expected)
|
|
(set! flow-sus-pass (+ flow-sus-pass 1))
|
|
(begin
|
|
(set! flow-sus-fail (+ flow-sus-fail 1))
|
|
(append! flow-sus-fails {:name name :expected expected :actual actual})))))
|
|
|
|
(define flow-s (fn (src) (flow-run src)))
|
|
|
|
;; ── flow/start ──────────────────────────────────────────────────
|
|
(flow-sus-test
|
|
"start: non-suspending flow returns the raw result"
|
|
(flow-s "(flow/start (lambda (x) (* x 2)) 5)")
|
|
10)
|
|
(flow-sus-test
|
|
"start: a suspending flow returns a flow-suspended state"
|
|
(flow-s
|
|
"(defflow w (sequence (lambda (x) (+ x 1)) (lambda (g) (suspend (quote await))) (lambda (c) c))) (car (flow/start w 10))")
|
|
"flow-suspended")
|
|
(flow-sus-test
|
|
"start: suspended state carries a numeric id"
|
|
(flow-s
|
|
"(defflow w (lambda (x) (suspend (quote await)))) (car (cdr (flow/start w 10)))")
|
|
1)
|
|
(flow-sus-test
|
|
"start: suspended state carries the suspend tag"
|
|
(flow-s
|
|
"(defflow w (lambda (x) (suspend (quote await)))) (car (cdr (cdr (flow/start w 10))))")
|
|
"await")
|
|
|
|
;; ── flow/resume ─────────────────────────────────────────────────
|
|
(flow-sus-test
|
|
"resume: injects the value and completes"
|
|
(flow-s
|
|
"(defflow w (sequence (lambda (x) (+ x 1)) (lambda (g) (suspend (quote await))) (lambda (c) (list (quote done) c)))) (define s (flow/start w 10)) (flow/resume (car (cdr s)) 777)")
|
|
(list "done" 777))
|
|
(flow-sus-test
|
|
"resume: injected value threads into the next node"
|
|
(flow-s
|
|
"(defflow w (sequence (lambda (x) (suspend (quote v))) (lambda (n) (* n 3)))) (define s (flow/start w 0)) (flow/resume (car (cdr s)) 14)")
|
|
42)
|
|
(flow-sus-test
|
|
"resume: replays earlier suspends (recompute is deterministic)"
|
|
(flow-s
|
|
"(define runs 0) (defflow w (sequence (lambda (x) (begin (set! runs (+ runs 1)) (+ x 1))) (lambda (g) (suspend (quote await))) (lambda (c) c))) (define s (flow/start w 10)) (flow/resume (car (cdr s)) 99) runs")
|
|
2)
|
|
|
|
;; ── multi-step suspension ───────────────────────────────────────
|
|
(flow-sus-test
|
|
"multi: first resume suspends at the next tag"
|
|
(flow-s
|
|
"(defflow two (sequence (lambda (x) (suspend (quote a))) (lambda (x) (suspend (quote b))) (lambda (x) (list (quote end) x)))) (define s (flow/start two 0)) (define s2 (flow/resume (car (cdr s)) 100)) (car (cdr (cdr s2)))")
|
|
"b")
|
|
(flow-sus-test
|
|
"multi: second resume completes with the latest value"
|
|
(flow-s
|
|
"(defflow two (sequence (lambda (x) (suspend (quote a))) (lambda (x) (suspend (quote b))) (lambda (x) (list (quote end) x)))) (define id (car (cdr (flow/start two 0)))) (flow/resume id 100) (flow/resume id 200)")
|
|
(list "end" 200))
|
|
|
|
;; ── error / lifecycle guards ────────────────────────────────────
|
|
(flow-sus-test
|
|
"resume: completed flow cannot be resumed again"
|
|
(flow-s
|
|
"(defflow w (lambda (x) (suspend (quote q)))) (define id (car (cdr (flow/start w 0)))) (flow/resume id 1) (flow/resume id 2)")
|
|
(list "flow-error" "not-suspended"))
|
|
(flow-sus-test
|
|
"resume: unknown id errors"
|
|
(flow-s "(flow/resume 999 1)")
|
|
(list "flow-error" "no-such-flow"))
|
|
|
|
;; ── flow/cancel ─────────────────────────────────────────────────
|
|
(flow-sus-test
|
|
"cancel: returns a flow-cancelled state"
|
|
(flow-s
|
|
"(defflow w (lambda (x) (suspend (quote q)))) (define id (car (cdr (flow/start w 0)))) (flow/cancel id)")
|
|
(list "flow-cancelled" 1))
|
|
(flow-sus-test
|
|
"cancel: a cancelled flow cannot be resumed (stale resume rejected)"
|
|
(flow-s
|
|
"(defflow w (lambda (x) (suspend (quote q)))) (define id (car (cdr (flow/start w 0)))) (flow/cancel id) (flow/resume id 5)")
|
|
(list "flow-error" "not-suspended"))
|
|
(flow-sus-test
|
|
"cancel: unknown id errors"
|
|
(flow-s "(flow/cancel 999)")
|
|
(list "flow-error" "no-such-flow"))
|
|
|
|
;; ── composition ─────────────────────────────────────────────────
|
|
(flow-sus-test
|
|
"suspend inside a branch arm"
|
|
(flow-s
|
|
"(defflow gate (branch (lambda (x) (> x 0)) (lambda (x) (suspend (quote approve))) (flow-const (quote rejected)))) (define s (flow/start gate 5)) (flow/resume (car (cdr s)) (quote approved))")
|
|
"approved")
|
|
(flow-sus-test
|
|
"two independent runs get independent ids"
|
|
(flow-s
|
|
"(defflow w (lambda (x) (suspend (quote q)))) (list (car (cdr (flow/start w 0))) (car (cdr (flow/start w 0))))")
|
|
(list 1 2))
|
|
(flow-sus-test
|
|
"suspend reason may be a structured value"
|
|
(flow-s
|
|
"(defflow w (lambda (x) (suspend (list (quote needs) (quote approval))))) (car (cdr (cdr (flow/start w 0))))")
|
|
(list "needs" "approval"))
|
|
|
|
(define flow-sus-tests-run! (fn () {:total (+ flow-sus-pass flow-sus-fail) :passed flow-sus-pass :failed flow-sus-fail :fails flow-sus-fails}))
|