giles
c21be815f3
lib/host/ra.sx — a PURE-SX seam runner (advertises {effect,branch,each,suspend}) with an INJECTED
erl-eval (real = er-to-sx-deep ∘ erlang-eval-ast; mock in unit tests), so it loads in the plain host
(Erlang refs resolve lazily inside lambdas) and is unit-testable without the Erlang runtime.
host/ra--{atom,bin,erl-src,start-expr,resume-expr,parse,make-runner,resume,real-eval}: marshals our
canonical activity → Erlang source (CID as <<"…">> binary, atoms single-quoted), starts a named
next/ flow via flow_store, parses (ok Id (flow_done V))→{:status done :effects V :flow-id} /
(ok Id (flow_suspended T))→{:status suspended :resume {:id :tag}}.
DUAL-RUNNER ROUTING (flows.sx): host/flow--required-caps now handles a {:erl-flow :needs} DAG
(declared caps, since a foreign flow can't be introspected); host/flow--select-runner picks the
cheapest runner whose capabilities cover the DAG's needs. The capability model is now REAL with two
runners — an {effect,branch} composition lands on exec-runner; a {suspend} DAG routes to RA.
Verified: ra 9/9 (mock erl-eval) + plans/ra-integration.sh 4/4 (the REAL module driving live
flow_store: urgent→done, newsletter→suspended with a resume handle, digest_sent effect-as-data).
Full host conformance 607/607; next/tests/triggers_e2e.sh 10/10 baseline intact.
FINDING → RA-LIVE deferred: gen_servers don't persist across separate erlang-eval-ast calls (flow
README), so true cross-call suspend/resume needs a PERSISTENT next/ kernel process. The runner +
marshalling + suspend/resume mechanics are proven; RA-live is process lifecycle + wiring, documented.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
64 lines
3.9 KiB
Bash
Executable File
64 lines
3.9 KiB
Bash
Executable File
#!/usr/bin/env bash
|
|
# RA INTEGRATION — the REAL lib/host/ra.sx runner (host/ra--make-runner + host/ra--real-eval)
|
|
# driving next/'s live flow_store durable flow, end-to-end. (ra-spike.sh proved the path with
|
|
# inline Erlang; this proves the MODULE.) urgent→done, newsletter→suspended, resume→done.
|
|
set -uo pipefail
|
|
cd "$(git rev-parse --show-toplevel)"
|
|
SX_SERVER="${SX_SERVER:-hosts/ocaml/_build/default/bin/sx_server.exe}"
|
|
[ -x "$SX_SERVER" ] || SX_SERVER="/root/rose-ash/hosts/ocaml/_build/default/bin/sx_server.exe"
|
|
PASS=0; FAIL=0
|
|
TMPFILE=$(mktemp); trap "rm -f $TMPFILE" EXIT
|
|
|
|
cat > "$TMPFILE" <<'EPOCHS'
|
|
(epoch 1)
|
|
(load "lib/erlang/tokenizer.sx")
|
|
(load "lib/erlang/parser.sx")
|
|
(load "lib/erlang/parser-core.sx")
|
|
(load "lib/erlang/parser-expr.sx")
|
|
(load "lib/erlang/parser-module.sx")
|
|
(load "lib/erlang/transpile.sx")
|
|
(load "lib/erlang/runtime.sx")
|
|
(load "lib/erlang/vm/dispatcher.sx")
|
|
(load "lib/host/ra.sx")
|
|
(epoch 2)
|
|
(eval "(er-load-gen-server!)")
|
|
(eval "(get (erlang-load-module (file-read \"next/kernel/envelope.erl\")) :name)")
|
|
(eval "(get (erlang-load-module (file-read \"next/flow/flow.erl\")) :name)")
|
|
(eval "(get (erlang-load-module (file-read \"next/flow/flow_spec.erl\")) :name)")
|
|
(eval "(get (erlang-load-module (file-read \"next/flow/flow_store.erl\")) :name)")
|
|
(eval "(get (erlang-load-module (file-read \"next/flow/flows/blog_publish_digest.erl\")) :name)")
|
|
(epoch 3)
|
|
;; gen_servers don't persist across separate erlang-eval-ast calls (flow README), so the injected
|
|
;; erl-eval boots the store + registers the flow inline on EVERY call (like the e2e). This proves
|
|
;; the MODULE's marshalling/dispatch/parse against the REAL flow; TRUE cross-call resume (suspend,
|
|
;; return, resume later) needs a PERSISTENT next/ kernel process — the RA-live deployment step.
|
|
(eval "(define ra/boot-eval (fn (src) (er-to-sx-deep (erlang-eval-ast (str \"flow_store:start_link(), FF = fun(_) -> [f1, f2, f3] end, flow_store:register_flow(bd, blog_publish_digest:build([{fetch_followers, FF}])), \" src)))))")
|
|
(eval "(define ra/R (host/ra--make-runner ra/boot-eval))")
|
|
(eval "(define ra/urgent {:verb \"create\" :actor \"alice\" :id \"u1\" :object-type \"article\" :category \"urgent\"})")
|
|
(eval "(define ra/news {:verb \"create\" :actor \"alice\" :id \"n1\" :object-type \"article\" :category \"newsletter\"})")
|
|
;; ── urgent: the real MODULE runner → done ──
|
|
(epoch 10)
|
|
(eval "(get ((get ra/R :run) {:erl-flow \"bd\"} {:activity ra/urgent}) :status)")
|
|
;; ── newsletter: the real MODULE runner → suspended (durable wait) ──
|
|
(epoch 20)
|
|
(eval "(get ((get ra/R :run) {:erl-flow \"bd\"} {:activity ra/news}) :status)")
|
|
;; ── the suspended result carries a resume handle {:id :tag morning} ──
|
|
(epoch 21)
|
|
(eval "(str (get (get ((get ra/R :run) {:erl-flow \"bd\"} {:activity ra/news}) :resume) :tag))")
|
|
;; ── the done result carries the flow's effect-as-data (digest_sent) ──
|
|
(epoch 30)
|
|
(eval "(str (first (get ((get ra/R :run) {:erl-flow \"bd\"} {:activity ra/urgent}) :effects)))")
|
|
EPOCHS
|
|
|
|
OUT=$(timeout 360 "$SX_SERVER" < "$TMPFILE" 2>/dev/null)
|
|
grab() { echo "$OUT" | awk -v e="$1" '$0 ~ "^\\(ok " e " "{print;exit} $0 ~ "^\\(ok-len " e " "{getline;print;exit} $0 ~ "^\\(error " e " "{print;exit}'; }
|
|
ck() { local a; a=$(grab "$1"); if echo "$a" | grep -qF -- "$2"; then PASS=$((PASS+1)); echo " ok [$3]"; else FAIL=$((FAIL+1)); echo " FAIL [$3] want '$2' got: $a"; fi; }
|
|
|
|
echo "── RA integration (the real host/ra.sx module) ──────"
|
|
ck 10 "done" "urgent → real RA runner → done"
|
|
ck 20 "suspended" "newsletter → real RA runner → suspended"
|
|
ck 21 "morning" "suspended result carries a resume handle (:tag morning)"
|
|
ck 30 "digest_sent" "done result carries the flow's effect-as-data"
|
|
echo "─────────────────────────────────────────────────────"
|
|
echo "PASS=$PASS FAIL=$FAIL"
|