Compare commits

..

20 Commits

Author SHA1 Message Date
9cfca1d008 flow: reference host driver flow-drive-host/flow-run-host + 4 tests
Some checks failed
Test, Build, and Deploy / test-build-deploy (push) Failing after 34s
Completes the host ABI from work-queue to driver loop: the host supplies only a
(kind payload) -> answer dispatch fn; flow-drive-host services one tick of pending
requests, flow-run-host ticks until quiescent (bounded). Tested via the art-dag
render -> human-review -> publish pipeline driven entirely by flow-run-host. The
art-dag integration is now: define dispatch, call flow-run-host. 166/166, 11 suites.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-06 19:33:04 +00:00
3cbf33d2d2 flow: host integration ABI (request/await/host-queue) + 11 tests (Phase 8)
Some checks failed
Test, Build, and Deploy / test-build-deploy (push) Failing after 38s
The seam for hooking flow to art-dag and human-in-the-loop later. (request kind
payload) suspends with a typed (flow-request kind payload) envelope and returns the
host's resume value; await-human/await-render sugar. (flow-host-requests) is the
host work queue: (id kind payload) for every suspended flow awaiting a host effect;
request?/request-kind/request-payload parse a tag. Tests include the art-dag-shaped
driver loop (render -> human-review -> publish). Host owns IO+persistence; flow only
requests (replay-safe). 162/162 across 11 suites.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-06 19:24:16 +00:00
c2d628e9c3 flow: README — API reference + deterministic-replay contract
Some checks failed
Test, Build, and Deploy / test-build-deploy (push) Failing after 1m10s
User-facing docs for the flow engine: the node model, every combinator, the
suspend/resume durability contract (escape-only call/cc -> deterministic replay),
lifecycle/introspection/hygiene API, fed-sx distribution, and substrate notes.
Doc-only; 151/151 unchanged.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-06 18:37:10 +00:00
aabb950256 flow: store hygiene flow/gc + flow/forget + 9 tests
Some checks are pending
Test, Build, and Deploy / test-build-deploy (push) Waiting to run
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>
2026-06-06 18:34:53 +00:00
2b47b2925c flow: end-to-end integration suite + 10 tests (Phase 7)
Some checks failed
Test, Build, and Deploy / test-build-deploy (push) Failing after 54s
Realistic flows composing every phase: an order pipeline (validate via attempt ->
payment suspend -> branch -> ledger federation via remote-node) and an onboarding
flow, each run through the full lifecycle including a simulated crash (export/wipe/
import) and a peer handoff mid-flow, with flow/pending|status|result introspection.
142/142 across 9 suites.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-06 18:17:40 +00:00
d9b9da3843 flow: railway attempt combinator — fail-value short-circuit + 10 tests (Phase 6)
Some checks failed
Test, Build, and Deploy / test-build-deploy (push) Failing after 55s
(attempt n1 n2 ...) threads like sequence but stops at the first node returning a
(fail ...) value, returning that failure. Makes the fail/recover error model
compose into validation/ETL pipelines (railway-oriented). 132/132 across 8 suites.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-06 18:09:21 +00:00
0a1b89c975 flow: bounded iteration combinators flow-while/flow-until + 6 tests
Some checks are pending
Test, Build, and Deploy / test-build-deploy (push) Waiting to run
(flow-while pred body max) / (flow-until pred body max) re-run body threading the
value while/until pred holds, capped at max steps for a deterministic bound (no
unbounded loops in pure SX). 122/122 across 7 suites.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-06 18:02:59 +00:00
0e6ba55647 flow: combinator library — tap, recover, map-flow + 11 tests (Phase 5 complete)
Some checks failed
Test, Build, and Deploy / test-build-deploy (push) Failing after 53s
tap: side-effecting pass-through (returns input). recover: fail-VALUE counterpart
of try-catch (run node; on (fail r) run handler on r). map-flow: run a node over
each item of a list, join results sequentially. 116/116 across 7 suites.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-06 17:57:48 +00:00
c1d24eb9b3 flow: operational introspection API — flow/status,result,list,pending + 12 tests
Some checks failed
Test, Build, and Deploy / test-build-deploy (push) Failing after 55s
flow/status id -> done|suspended|cancelled|unknown; flow/result id -> value or
error; flow/list -> (id status) per flow; flow/pending -> (id waiting-tag) for
suspended flows (operator view of what each awaits). Pure store introspection.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-06 17:53:23 +00:00
16cb727406 flow: replication + handoff across instances + 6 tests (Phase 4 complete)
Some checks failed
Test, Build, and Deploy / test-build-deploy (push) Failing after 42s
flow-replicate-to copies the plain-data store export to a peer's replica slot;
flow-restore-from imports it. Handoff = replicate, local instance dies, peer
restores and resumes by id. The replay log survives the move, so all resolved
suspends carry over. Same durable-data mechanism as crash recovery, across
instances. All four phases complete: 93/93.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-06 17:48:39 +00:00
f8722b3b08 flow: remote-failover — try peers in order, fall through to local + 6 tests
Some checks failed
Test, Build, and Deploy / test-build-deploy (push) Failing after 45s
(remote-failover addrs fn local) tries fn on each peer in order, moves to the next
on any raised error, and runs the local node if every peer fails. Threads input,
composes in sequences.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-06 17:44:04 +00:00
e1f802cfff flow: remote-node via mock fed-sx transport + 7 tests (Phase 4 begins)
Some checks failed
Test, Build, and Deploy / test-build-deploy (push) Failing after 1m4s
(remote-node addr fn) runs a node on a federation peer. Transport is the fed-sx
boundary, mocked by a peer registry (flow-peer-register!); raises
flow-remote-unreachable / flow-remote-no-fn. Composes with sequence/suspend/retry.
Also fixes conformance.sh to load remote.sx before api.sx.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-06 17:40:25 +00:00
97c7623743 flow: crash recovery — store export/import + resumable scan + 8 tests (Phase 3 complete)
Some checks failed
Test, Build, and Deploy / test-build-deploy (push) Failing after 58s
Records are name-keyed (defflow registers names); flow-store-export nulls live
procs to plain data, flow-store-import! restores, flow-resumable-ids scans for
paused flows. Resume re-resolves the proc by name, so a flow survives a wiped
store (simulated restart). The whole durable model persists only plain data.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-06 17:25:47 +00:00
e896deffc8 flow: Phase 3 suspend/resume/cancel via deterministic replay + 17 tests
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>
2026-06-06 17:20:09 +00:00
e762cc2e32 flow: timeout combinator — cooperative step budget + 7 tests (Phase 2 complete)
Some checks failed
Test, Build, and Deploy / test-build-deploy (push) Failing after 22s
(timeout budget node) bounds a node deterministically: nodes opt in via (tick),
budget ticks are allowed, the next raises flow-timeout. No scheduler/clock in pure
SX so the budget is a step count, not wall-clock. Budgets nest and are per-run.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-06 16:42:16 +00:00
4674620d7e flow: retry combinator — re-run node on raised exceptions + 6 tests
Some checks failed
Test, Build, and Deploy / test-build-deploy (push) Failing after 30s
(retry n node) re-runs up to n attempts on a raised exception; the last attempt's
exception propagates. Explicit (fail ...) values are NOT retried — they pass through.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-06 16:39:21 +00:00
f3da3b975a flow: try-catch combinator — reify raised exceptions + 6 tests
Some checks failed
Test, Build, and Deploy / test-build-deploy (push) Failing after 34s
(try-catch node handler) runs node; on a raised exception calls (handler error)
with the reified error via Scheme guard, returns the handler value.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-06 16:37:26 +00:00
1731476dc6 flow: error model — fail/failed?/fail-reason failure values + 6 tests
Some checks failed
Test, Build, and Deploy / test-build-deploy (push) Failing after 29s
Explicit (fail reason) values flow downstream as data and are inspected with
failed?/fail-reason — distinct from raised exceptions (retry/try-catch territory).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-06 16:35:40 +00:00
65cbdb8387 flow: branch combinator (conditional) + 6 tests
Some checks failed
Test, Build, and Deploy / test-build-deploy (push) Failing after 50s
Phase 2 control flow. (branch pred then else) selects then/else node by running
pred on the threaded input; named 'branch' since 'cond' is a Scheme special form.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-06 16:32:37 +00:00
91ffba9975 flow: Phase 1 declarative DAG — sequence/parallel/defflow combinators + 18 tests
Some checks failed
Test, Build, and Deploy / test-build-deploy (push) Failing after 31s
Flow combinators as a Scheme prelude loaded onto scheme-standard-env; a flow is a
Scheme procedure input->output, run inside the interpreter (sets up Phase 3 call/cc
suspend). flow/start entry point, conformance runner, scoreboard.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-06 16:22:22 +00:00
107 changed files with 1901 additions and 7052 deletions

View File

@@ -571,12 +571,9 @@ and cek_run_with_io state =
Hashtbl.replace d "descent" (Number desc);
Dict d
| _ ->
let argsv = Sx_runtime.get_val request (String "args") in
(match Sx_persist_store.handle_op op argsv with
| Some resp -> resp
| None ->
let args = (match argsv with List l -> l | _ -> [argsv]) in
io_request op args)
let args = let a = Sx_runtime.get_val request (String "args") in
(match a with List l -> l | _ -> [a]) in
io_request op args
in
s := Sx_ref.cek_resume !s response;
loop ()
@@ -1543,12 +1540,7 @@ let rec dispatch env cmd =
| Some path -> load_library_file path | None -> ());
Nil
end
end else
(* durable-storage ops: service against on-disk store *)
let args = Sx_runtime.get_val request (String "args") in
(match Sx_persist_store.handle_op op args with
| Some resp -> resp
| None -> Nil (* non-import IO: resume with nil *)) in
end else Nil (* non-import IO: resume with nil *) in
s := Sx_ref.cek_resume !s response
done;
Sx_ref.cek_value !s
@@ -3901,10 +3893,7 @@ let http_mode port =
Dict d
| "io-sleep" | "sleep" -> Nil
| "import" -> Nil
| _ ->
(match Sx_persist_store.handle_op op args with
| Some resp -> resp
| None -> Nil));
| _ -> Nil);
(* Response cache — path → full HTTP response string.
Populated during pre-warm, serves cached responses in <0.1ms.
Thread-safe: reads are lock-free (Hashtbl.find_opt is atomic for

View File

@@ -1,293 +0,0 @@
(* sx_persist_store — host durable-storage adapter for lib/persist.
Production twin of `persist/serve` (lib/persist/durable.sx): it answers the
same `persist/...` IO ops, but backs them with real on-disk storage so writes
survive a process restart. Stateless-on-disk: every op reads/writes the
filesystem directly, so a fresh process recovers state with no warm-up — the
log on disk IS the state.
On-disk layout under the root dir (default ./persist-data, or $SX_PERSIST_DIR):
streams/<hex(stream)>.log append-only, one SX-serialized event per line
streams/<hex(stream)>.seq per-stream monotonic high-water counter (int)
kv/<hex(key)> one SX-serialized value per key
Invariants honoured (see plans/persist-on-sx.md Blocker spec):
1. last-seq is a per-stream monotonic counter stored in .seq, SEPARATE from
the rows — it keeps climbing across truncate, so a compacted stream never
reassigns a seq.
2. append never renumbers — the event already carries its :seq (log.sx does
last-seq+1); the host only bumps the high-water mark to max(hw, seq).
3. read returns surviving events in append order with :seq intact.
4. streams is the set of streams that ever had an append — keyed off the .seq
file, which truncate never deletes, so it survives full compaction.
5. values round-trip structurally via the SX serializer/parser. *)
open Sx_types
(* ---- root dir ---------------------------------------------------------- *)
let _root : string option ref = ref None
let set_root dir = _root := Some dir
let root_dir () =
match !_root with
| Some d -> d
| None -> (try Sys.getenv "SX_PERSIST_DIR" with Not_found -> "persist-data")
(* ---- filesystem helpers ------------------------------------------------ *)
let rec ensure_dir dir =
if dir = "" || dir = "." || dir = "/" || Sys.file_exists dir then ()
else begin
ensure_dir (Filename.dirname dir);
(try Unix.mkdir dir 0o755 with Unix.Unix_error (Unix.EEXIST, _, _) -> ())
end
let streams_dir () = Filename.concat (root_dir ()) "streams"
let kv_dir () = Filename.concat (root_dir ()) "kv"
let blobs_dir () = Filename.concat (root_dir ()) "blobs"
let read_file path =
let ic = open_in_bin path in
let n = in_channel_length ic in
let s = really_input_string ic n in
close_in ic;
s
(* Atomic write: temp file in the same dir then rename over the target. *)
let write_file_atomic path contents =
ensure_dir (Filename.dirname path);
let tmp = path ^ ".tmp" in
let oc = open_out_bin tmp in
output_string oc contents;
flush oc;
close_out oc;
Sys.rename tmp path
let append_line path line =
ensure_dir (Filename.dirname path);
let oc = open_out_gen [Open_append; Open_creat; Open_wronly] 0o644 path in
output_string oc line;
output_char oc '\n';
close_out oc
(* ---- name <-> filename (hex, reversible, fs-safe) ---------------------- *)
let hex_encode s =
let b = Buffer.create (String.length s * 2) in
String.iter (fun c -> Buffer.add_string b (Printf.sprintf "%02x" (Char.code c))) s;
Buffer.contents b
let hex_decode s =
let n = String.length s / 2 in
String.init n (fun i -> Char.chr (int_of_string ("0x" ^ String.sub s (i * 2) 2)))
let stream_log stream = Filename.concat (streams_dir ()) (hex_encode stream ^ ".log")
let stream_seq stream = Filename.concat (streams_dir ()) (hex_encode stream ^ ".seq")
let kv_path key = Filename.concat (kv_dir ()) (hex_encode key)
(* ---- value <-> SX text (round-trips through Sx_parser) ----------------- *)
let escape_str s =
let len = String.length s in
let buf = Buffer.create (len + 16) in
for i = 0 to len - 1 do
match s.[i] with
| '"' -> Buffer.add_string buf "\\\""
| '\\' -> Buffer.add_string buf "\\\\"
| '\n' -> Buffer.add_string buf "\\n"
| '\r' -> Buffer.add_string buf "\\r"
| '\t' -> Buffer.add_string buf "\\t"
| c -> Buffer.add_char buf c
done;
Buffer.contents buf
let rec serialize = function
| Nil -> "nil"
| Bool true -> "true"
| Bool false -> "false"
| Integer n -> string_of_int n
| Number n -> format_number n
| String s -> "\"" ^ escape_str s ^ "\""
| Symbol s -> "(quote " ^ s ^ ")"
| Keyword k -> ":" ^ k
| List items | ListRef { contents = items } ->
"(list" ^ (List.fold_left (fun acc v -> acc ^ " " ^ serialize v) "" items) ^ ")"
| Dict d ->
let pairs = Hashtbl.fold (fun k v acc ->
(Printf.sprintf ":%s %s" k (serialize v)) :: acc) d [] in
"{" ^ String.concat " " (List.sort String.compare pairs) ^ "}"
| _ -> "nil"
(* Parse one serialized value back. Empty / blank -> Nil. *)
let rec deserialize line =
let line = String.trim line in
if line = "" then Nil
else match Sx_parser.parse_all line with
| v :: _ -> eval_quote_lists v
| [] -> Nil
(* serialize emits lists as `(list ...)` and symbols as `(quote s)` so the
parser yields data, not a call — but the parser leaves those as AST. Walk
the parsed AST and collapse `(list ...)`/`(quote s)` back to values. *)
and eval_quote_lists v =
match v with
| List (Symbol "quote" :: x :: []) -> x
| List (Symbol "list" :: rest) -> List (List.map eval_quote_lists rest)
| List items -> List (List.map eval_quote_lists items)
| ListRef { contents = items } -> List (List.map eval_quote_lists items)
| Dict d ->
let d' = Hashtbl.create (Hashtbl.length d) in
Hashtbl.iter (fun k v -> Hashtbl.replace d' k (eval_quote_lists v)) d;
Dict d'
| other -> other
(* ---- seq counter ------------------------------------------------------- *)
let read_seq stream =
let p = stream_seq stream in
if Sys.file_exists p then (try int_of_string (String.trim (read_file p)) with _ -> 0)
else 0
let write_seq stream n = write_file_atomic (stream_seq stream) (string_of_int n)
let value_to_int = function
| Integer n -> n
| Number n -> int_of_float n
| _ -> 0
let event_seq ev =
match ev with
| Dict d -> (match Hashtbl.find_opt d "seq" with Some v -> value_to_int v | None -> 0)
| _ -> 0
(* ---- ops --------------------------------------------------------------- *)
let do_append stream ev =
ensure_dir (streams_dir ());
(* bump the monotonic high-water mark; create .seq on first append so the
stream shows up in `streams` and survives later truncation. *)
let hw = read_seq stream in
let s = event_seq ev in
write_seq stream (max hw s);
append_line (stream_log stream) (serialize ev)
let do_read stream =
let p = stream_log stream in
if not (Sys.file_exists p) then List []
else begin
let content = read_file p in
let lines = String.split_on_char '\n' content in
let evs = List.filter_map (fun l ->
if String.trim l = "" then None else Some (deserialize l)) lines in
List evs
end
let do_last_seq stream = Number (float_of_int (read_seq stream))
let list_dir_suffix dir suffix =
if not (Sys.file_exists dir) then []
else
Array.to_list (Sys.readdir dir)
|> List.filter (fun f -> Filename.check_suffix f suffix)
|> List.map (fun f -> hex_decode (Filename.chop_suffix f suffix))
|> List.sort String.compare
let do_streams () = List (List.map (fun s -> String s) (list_dir_suffix (streams_dir ()) ".seq"))
(* drop events with seq <= n; the .seq high-water counter is untouched. *)
let do_truncate stream n =
let p = stream_log stream in
if Sys.file_exists p then begin
let evs = match do_read stream with List l -> l | _ -> [] in
let kept = List.filter (fun ev -> event_seq ev > n) evs in
let body = String.concat "" (List.map (fun ev -> serialize ev ^ "\n") kept) in
write_file_atomic p body
end
let do_kv_get key =
let p = kv_path key in
if Sys.file_exists p then deserialize (read_file p) else Nil
let do_kv_put key v =
ensure_dir (kv_dir ());
write_file_atomic (kv_path key) (serialize v)
let do_kv_delete key =
let p = kv_path key in
if Sys.file_exists p then (try Sys.remove p with _ -> ())
let do_kv_has key = Bool (Sys.file_exists (kv_path key))
let do_kv_keys () =
if not (Sys.file_exists (kv_dir ())) then List []
else
List (
Array.to_list (Sys.readdir (kv_dir ()))
|> List.map hex_decode
|> List.sort String.compare
|> List.map (fun s -> String s))
(* ---- blob store (content-addressed) ------------------------------------ *)
(* Same pattern as the persist ops, but a SEPARATE adapter: large objects live
in a content-addressed directory keyed by a CIDv1 (raw codec, sha2-256).
persist only ever stores the returned ref ({:cid :size :mime}), never bytes.
blob/put is idempotent — identical bytes hash to the same cid + same file. *)
let codec_raw = 0x55
let blob_cid bytes =
let digest = Sx_cid.unhex (Sx_sha2.sha256_hex bytes) in
Sx_cid.cidv1 codec_raw (Sx_cid.multihash Sx_cid.mh_sha2_256 digest)
let blob_path cid = Filename.concat (blobs_dir ()) cid
let do_blob_put bytes =
let cid = blob_cid bytes in
let p = blob_path cid in
if not (Sys.file_exists p) then write_file_atomic p bytes;
String cid
let do_blob_get cid =
let p = blob_path cid in
if Sys.file_exists p then String (read_file p) else Nil
let do_blob_has cid = Bool (Sys.file_exists (blob_path cid))
(* ---- dispatch ---------------------------------------------------------- *)
let arglist = function
| List l | ListRef { contents = l } -> l
| Nil -> []
| v -> [v]
(* Returns Some response if op is a persist op this store owns, None otherwise. *)
let handle_op op args =
let a = arglist args in
let str = function String s -> s | v -> value_to_string v in
match op with
| "persist/append" ->
(match a with stream :: ev :: _ -> do_append (str stream) ev | _ -> ()); Some Nil
| "persist/read" ->
(match a with stream :: _ -> Some (do_read (str stream)) | _ -> Some (List []))
| "persist/last-seq" ->
(match a with stream :: _ -> Some (do_last_seq (str stream)) | _ -> Some (Number 0.0))
| "persist/streams" -> Some (do_streams ())
| "persist/truncate" ->
(match a with stream :: n :: _ -> do_truncate (str stream) (value_to_int n) | _ -> ()); Some Nil
| "persist/kv-get" ->
(match a with key :: _ -> Some (do_kv_get (str key)) | _ -> Some Nil)
| "persist/kv-put" ->
(match a with key :: v :: _ -> do_kv_put (str key) v | _ -> ()); Some Nil
| "persist/kv-delete" ->
(match a with key :: _ -> do_kv_delete (str key) | _ -> ()); Some Nil
| "persist/kv-has?" ->
(match a with key :: _ -> Some (do_kv_has (str key)) | _ -> Some (Bool false))
| "persist/kv-keys" -> Some (do_kv_keys ())
| "blob/put" ->
(match a with bytes :: _ -> Some (do_blob_put (str bytes)) | _ -> Some Nil)
| "blob/get" ->
(match a with cid :: _ -> Some (do_blob_get (str cid)) | _ -> Some Nil)
| "blob/has?" ->
(match a with cid :: _ -> Some (do_blob_has (str cid)) | _ -> Some (Bool false))
| _ -> None

View File

@@ -1,144 +0,0 @@
#!/usr/bin/env bash
# hosts/ocaml/test/persist_durable_test.sh
# Acceptance test for the host durable-storage adapter (Sx_persist_store).
#
# Exercises `persist/durable-backend` (REAL `perform`, not the mock) under the
# WORKTREE-built sx_server.exe, and asserts:
# 1. durable: writes land on disk and read back (the silent-data-loss repro
# from plans/persist-on-sx.md now returns correct values).
# 2. last-seq is monotonic across truncate (compaction never reassigns a seq).
# 3. kv ops round-trip and delete.
# 4. recovery: a REAL process restart (write, exit, fresh process, replay)
# recovers state from disk.
#
# Run from repo root or anywhere; locates the worktree binary relative to itself.
set -uo pipefail
HERE="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
ROOT="$(cd "$HERE/../../.." && pwd)" # repo/worktree root
cd "$ROOT"
SX="hosts/ocaml/_build/default/bin/sx_server.exe"
if [ ! -x "$SX" ]; then
echo "ERROR: worktree binary not found at $SX — build it first:" >&2
echo " (cd hosts/ocaml && dune build bin/sx_server.exe)" >&2
exit 1
fi
DATADIR="$(mktemp -d)"
trap 'rm -rf "$DATADIR"' EXIT
PASS=0
FAIL=0
check() { # check <label> <got> <expected>
if [ "$2" = "$3" ]; then
PASS=$((PASS + 1)); printf ' ok %-40s => %s\n' "$1" "$2"
else
FAIL=$((FAIL + 1)); printf ' FAIL %-40s got [%s] want [%s]\n' "$1" "$2" "$3"
fi
}
PRELUDE='(epoch 1)
(load "spec/stdlib.sx")
(load "lib/r7rs.sx")
(load "lib/persist/event.sx")
(load "lib/persist/backend.sx")
(load "lib/persist/log.sx")
(load "lib/persist/kv.sx")
(load "lib/persist/durable.sx")
(load "lib/persist/blob.sx")
(epoch 2)'
# run_eval <sx-expr-string>: prints the final (ok-len 2 ...) payload line.
run_eval() {
local expr="$1"
printf '%s\n(eval %s)\n' "$PRELUDE" "$expr" \
| SX_PERSIST_DIR="$DATADIR" timeout 60 "$SX" 2>/dev/null \
| awk '/^\(ok-len 2 / {getline; print; exit}'
}
# escape an SX program into a single-line double-quoted SX string literal for
# (eval "..."). The REPL reads one command per physical line, so newlines in the
# program are collapsed to spaces.
q() { printf '"%s"' "$(printf '%s' "$1" | tr '\n' ' ' | sed 's/\\/\\\\/g; s/"/\\"/g')"; }
echo "== durable: append/read/last-seq round-trip on disk =="
GOT=$(run_eval "$(q '(let ((b (persist/durable-backend)))
(begin
(persist/append b "s" "x" 0 {:v 1})
(persist/append b "s" "x" 0 {:v 2})
(list (persist/event-seq (persist/append b "s" "x" 0 {:v 3}))
(persist/count b "s")
(len (persist/read b "s")))))')")
check "append/count/read" "$GOT" "(3 3 3)"
echo "== last-seq monotonic across truncate =="
GOT=$(run_eval "$(q '(let ((b (persist/durable-backend)))
(begin
(persist/append b "t" "x" 0 {})
(persist/append b "t" "x" 0 {})
(persist/append b "t" "x" 0 {})
(persist/truncate b "t" 2)
(list (persist/last-seq b "t") (persist/count b "t"))))')")
check "last-seq survives truncate" "$GOT" "(3 1)"
echo "== streams set survives compaction =="
GOT=$(run_eval "$(q '(let ((b (persist/durable-backend)))
(sort ((get b "streams"))))')")
check "streams" "$GOT" '("s" "t")'
echo "== kv round-trip + delete =="
GOT=$(run_eval "$(q '(let ((b (persist/durable-backend)))
(begin
(persist/kv-put b "k" {:a 1 :b "two"})
(persist/kv-put b "gone" 9)
(persist/kv-delete b "gone")
(list (get (persist/kv-get b "k") :b)
(persist/kv-has? b "k")
(persist/kv-has? b "gone"))))')")
check "kv get/has/delete" "$GOT" '("two" true false)'
echo "== recovery: state survives a REAL process restart =="
# write in process A then let it exit; the next run is a brand-new process.
run_eval "$(q '(let ((b (persist/durable-backend)))
(begin
(persist/append b "r" "ev" 0 {:n 1})
(persist/append b "r" "ev" 0 {:n 2})
(persist/kv-put b "survive" "yes")
(persist/count b "r")))')" >/dev/null
# fresh process, same SX_PERSIST_DIR — must replay from disk.
GOT=$(run_eval "$(q '(let ((b (persist/durable-backend)))
(list (persist/count b "r")
(persist/last-seq b "r")
(get (get (nth (persist/read b "r") 1) :data) :n)
(persist/kv-get b "survive")))')")
check "recovered after restart" "$GOT" '(2 2 2 "yes")'
echo "== blob: content-addressed put/get/has? round-trip =="
GOT=$(run_eval "$(q '(let ((bs (persist/blob-store-backend)))
(let ((r (persist/blob-store bs "hello world" "text/plain")))
(list (persist/blob-size r)
(persist/blob-mime r)
(persist/blob-fetch bs r)
(persist/blob-exists? bs r))))')")
check "blob size/mime/fetch/exists" "$GOT" '(11 "text/plain" "hello world" true)'
echo "== blob: put is content-addressed (idempotent cid) =="
GOT=$(run_eval "$(q '(let ((bs (persist/blob-store-backend)))
(equal? (persist/blob-cid (persist/blob-store bs "same bytes" "x"))
(persist/blob-cid (persist/blob-store bs "same bytes" "x"))))')")
check "same bytes -> same cid" "$GOT" "true"
echo "== blob: bytes + ref-in-kv survive a REAL restart =="
# process A: store a blob, keep only its ref in the durable kv.
run_eval "$(q '(let ((b (persist/durable-backend)) (bs (persist/blob-store-backend)))
(begin (persist/kv-put b "logo" (persist/blob-store bs "PNGDATA" "image/png")) nil))')" >/dev/null
# fresh process: read the ref from kv, fetch the bytes from the blob store.
GOT=$(run_eval "$(q '(let ((b (persist/durable-backend)) (bs (persist/blob-store-backend)))
(let ((r (persist/kv-get b "logo")))
(list (persist/blob-fetch bs r) (persist/blob-exists? bs r) (persist/blob-mime r))))')")
check "blob recovered via ref after restart" "$GOT" '("PNGDATA" true "image/png")'
echo
echo "durable adapter: $PASS passed, $FAIL failed"
[ "$FAIL" -eq 0 ]

View File

@@ -1,38 +0,0 @@
; feed/acl — per-viewer visibility filtering. The same candidate stream yields
; different timelines for different viewers, so ACL is applied per request and
; pre-ACL timelines are never cached.
;
; permit? is injected: (permit? viewer activity) -> bool. Wire a real acl-sx
; predicate here; feed/permit-acl? is a self-contained default that reads an
; optional :visible-to allowlist on the activity.
;
; Requires: lib/feed/normalize.sx, lib/feed/stream.sx, lib/feed/fanout.sx
; (feed/-elem?), lib/feed/rank.sx (feed/top).
; default permit: actor always sees own activity; absent/nil :visible-to is
; public; otherwise viewer must be in the allowlist.
(define
feed/permit-acl?
(fn
(viewer a)
(or
(equal? viewer (get a :actor))
(let
((allowed (get a :visible-to nil)))
(if (= allowed nil) true (feed/-elem? viewer allowed))))))
(define feed/permit-public? (fn (viewer a) true))
; filter a stream to what viewer may read
(define
feed/visible
(fn
(stream viewer permit?)
(feed/filter stream (fn (a) (permit? viewer a)))))
; the capstone: candidate stream -> ACL for viewer -> rank -> top-N
(define
feed/timeline
(fn
(stream viewer permit? score-fn n)
(feed/top (feed/visible stream viewer permit?) score-fn n)))

View File

@@ -1,62 +0,0 @@
; feed/aggregate — group-by / counting via key-reduce. Keys must be strings
; (dict keys), so composite keys (actor, day) are joined into one string.
;
; Requires: lib/feed/normalize.sx, lib/feed/stream.sx.
; group activities into a dict: key-string -> (list of activities), order-preserving
(define
feed/group-by
(fn
(stream key-fn)
(reduce
(fn
(g a)
(let
((k (key-fn a)))
(assoc g k (append (get g k (list)) (list a)))))
{}
(feed/items stream))))
; key-string -> count
(define
feed/group-count
(fn
(stream key-fn)
(reduce
(fn
(g a)
(let
((k (key-fn a)))
(assoc g k (+ (get g k 0) 1))))
{}
(feed/items stream))))
; --- composite keys ---------------------------------------------------------
(define feed/day (fn (at window) (floor (/ at window))))
; (actor, day-bucket) -> "actor#day"
(define
feed/actor-day-key
(fn
(window)
(fn
(a)
(string-append
(get a :actor)
"#"
(number->string (feed/day (get a :at) window))))))
(define
feed/by-actor-day
(fn (stream window) (feed/group-count stream (feed/actor-day-key window))))
; per-actor activity counts
(define
feed/actor-counts
(fn (stream) (feed/group-count stream feed/actor)))
; per-object activity counts (engagement)
(define
feed/object-counts
(fn (stream) (feed/group-count stream feed/object)))

View File

@@ -1,24 +0,0 @@
; feed/api — ergonomic API over the stream layer for non-APL callers.
; A single mutable activity log; post appends, all returns it as a stream.
;
; Requires: lib/feed/normalize.sx, lib/feed/stream.sx (loaded by harness).
(define feed/-log (list))
; post — normalize then append. Returns the stored activity.
(define
feed/post
(fn
(raw)
(let
((a (feed/normalize raw)))
(begin (set! feed/-log (append feed/-log (list a))) a))))
; all — the whole log as a stream (insertion order)
(define feed/all (fn () (feed/stream feed/-log)))
; reset! — clear the log (test hygiene)
(define feed/reset! (fn () (begin (set! feed/-log (list)) nil)))
; size — number of posted activities
(define feed/size (fn () (len feed/-log)))

View File

@@ -1,125 +0,0 @@
#!/usr/bin/env bash
# lib/feed/conformance.sh — run feed test suites, emit scoreboard.json + scoreboard.md.
set -uo pipefail
cd "$(git rev-parse --show-toplevel)"
SX_SERVER="${SX_SERVER:-/root/rose-ash/hosts/ocaml/_build/default/bin/sx_server.exe}"
if [ ! -x "$SX_SERVER" ]; then
SX_SERVER="hosts/ocaml/_build/default/bin/sx_server.exe"
fi
if [ ! -x "$SX_SERVER" ]; then
echo "ERROR: sx_server.exe not found." >&2
exit 1
fi
SUITES=(basic fanout rank integration content notify home dedupe trending mute page thread)
OUT_JSON="lib/feed/scoreboard.json"
OUT_MD="lib/feed/scoreboard.md"
run_suite() {
local suite=$1
local file="lib/feed/tests/${suite}.sx"
local TMP
TMP=$(mktemp)
cat > "$TMP" << EPOCHS
(epoch 1)
(load "spec/stdlib.sx")
(load "lib/r7rs.sx")
(load "lib/apl/runtime.sx")
(load "lib/feed/normalize.sx")
(load "lib/feed/stream.sx")
(load "lib/feed/api.sx")
(load "lib/feed/fanout.sx")
(load "lib/feed/dedupe.sx")
(load "lib/feed/aggregate.sx")
(load "lib/feed/rank.sx")
(load "lib/feed/acl.sx")
(load "lib/feed/fed.sx")
(load "lib/feed/content.sx")
(load "lib/feed/notify.sx")
(load "lib/feed/home.sx")
(load "lib/feed/trending.sx")
(load "lib/feed/mute.sx")
(load "lib/feed/page.sx")
(load "lib/feed/thread.sx")
(epoch 2)
(eval "(define feed-test-pass 0)")
(eval "(define feed-test-fail 0)")
(eval "(define feed-test (fn (name got expected) (if (= got expected) (set! feed-test-pass (+ feed-test-pass 1)) (set! feed-test-fail (+ feed-test-fail 1)))))")
(epoch 3)
(load "${file}")
(epoch 4)
(eval "(list feed-test-pass feed-test-fail)")
EPOCHS
local OUTPUT
OUTPUT=$(timeout 300 "$SX_SERVER" < "$TMP" 2>/dev/null)
rm -f "$TMP"
local LINE
LINE=$(echo "$OUTPUT" | awk '/^\(ok-len 4 / {getline; print; exit}')
if [ -z "$LINE" ]; then
LINE=$(echo "$OUTPUT" | grep -E '^\(ok 4 \([0-9]+ [0-9]+\)\)' | tail -1 \
| sed -E 's/^\(ok 4 //; s/\)$//')
fi
local P F
P=$(echo "$LINE" | sed -E 's/^\(([0-9]+) ([0-9]+)\).*/\1/')
F=$(echo "$LINE" | sed -E 's/^\(([0-9]+) ([0-9]+)\).*/\2/')
P=${P:-0}
F=${F:-0}
echo "${P} ${F}"
}
declare -A SUITE_PASS
declare -A SUITE_FAIL
TOTAL_PASS=0
TOTAL_FAIL=0
echo "Running feed conformance suite..." >&2
for s in "${SUITES[@]}"; do
read -r p f < <(run_suite "$s")
SUITE_PASS[$s]=$p
SUITE_FAIL[$s]=$f
TOTAL_PASS=$((TOTAL_PASS + p))
TOTAL_FAIL=$((TOTAL_FAIL + f))
printf " %-12s %d/%d\n" "$s" "$p" "$((p+f))" >&2
done
# scoreboard.json
{
printf '{\n'
printf ' "suites": {\n'
first=1
for s in "${SUITES[@]}"; do
if [ $first -eq 0 ]; then printf ',\n'; fi
printf ' "%s": {"pass": %d, "fail": %d}' "$s" "${SUITE_PASS[$s]}" "${SUITE_FAIL[$s]}"
first=0
done
printf '\n },\n'
printf ' "total_pass": %d,\n' "$TOTAL_PASS"
printf ' "total_fail": %d,\n' "$TOTAL_FAIL"
printf ' "total": %d\n' "$((TOTAL_PASS + TOTAL_FAIL))"
printf '}\n'
} > "$OUT_JSON"
# scoreboard.md
{
printf '# feed Conformance Scoreboard\n\n'
printf '_Generated by `lib/feed/conformance.sh`_\n\n'
printf '| Suite | Pass | Fail | Total |\n'
printf '|-------|-----:|-----:|------:|\n'
for s in "${SUITES[@]}"; do
p=${SUITE_PASS[$s]}
f=${SUITE_FAIL[$s]}
printf '| %s | %d | %d | %d |\n' "$s" "$p" "$f" "$((p+f))"
done
printf '| **Total** | **%d** | **%d** | **%d** |\n' "$TOTAL_PASS" "$TOTAL_FAIL" "$((TOTAL_PASS + TOTAL_FAIL))"
} > "$OUT_MD"
echo "Wrote $OUT_JSON and $OUT_MD" >&2
echo "Total: $TOTAL_PASS pass, $TOTAL_FAIL fail" >&2
[ "$TOTAL_FAIL" -eq 0 ]

View File

@@ -1,68 +0,0 @@
; feed/content — TF-IDF relevance over activity :tags. Rare tags carry more
; signal, so an activity matching an uncommon tag ranks above one matching a
; common tag. Composes with rank.sx: feed/tfidf-score is just another scorer.
;
; Requires: lib/feed/normalize.sx, lib/feed/stream.sx, lib/feed/fanout.sx
; (feed/-distinct), lib/feed/rank.sx (feed/rank).
; document frequency: tag -> number of activities whose :tags contain it
; (a tag repeated within one activity counts once toward df)
(define
feed/tag-df
(fn
(stream)
(reduce
(fn
(df a)
(reduce
(fn (d t) (assoc d t (+ (get d t 0) 1)))
df
(feed/-distinct (get a :tags))))
{}
(feed/items stream))))
; inverse document frequency: tag -> log(N / df)
(define
feed/tag-idf
(fn
(stream)
(let
((n (feed/count stream)) (df (feed/tag-df stream)))
(reduce
(fn (idf t) (assoc idf t (log (/ n (get df t)))))
{}
(keys df)))))
; term frequency within one activity: tag -> occurrence count
(define
feed/-tf
(fn
(a)
(reduce
(fn (tf t) (assoc tf t (+ (get tf t 0) 1)))
{}
(get a :tags))))
; relevance of an activity to a query (list of tags) given precomputed idf:
; sum over query tags of tf(tag in activity) * idf(tag in corpus)
(define
feed/tfidf-score
(fn
(idf query)
(fn
(a)
(let
((tf (feed/-tf a)))
(reduce
(fn
(acc t)
(+ acc (* (get tf t 0) (get idf t 0))))
0
query)))))
; rank a stream by relevance to query tags (idf computed over the stream itself)
(define
feed/by-relevance
(fn
(stream query)
(feed/rank stream (feed/tfidf-score (feed/tag-idf stream) query))))

View File

@@ -1,76 +0,0 @@
; feed/dedupe — collapse duplicate items, keeping first occurrence per key.
; Each verb may want its own key (see briefing): "alice posted X" keys on
; (actor verb object) — distinct per actor; "alice liked X / bob liked X"
; collapse on (verb object) so the cross-actor likes fold into one.
;
; Requires: lib/feed/normalize.sx, lib/feed/stream.sx, lib/feed/fanout.sx
; (feed/-elem? lives in fanout.sx).
; generic: dedupe a stream by key-fn, first occurrence wins (stable)
(define
feed/-dedup-by
(fn
(items key-fn)
(get
(reduce
(fn
(st x)
(let
((k (key-fn x)))
(if (feed/-elem? k (get st :seen)) st {:seen (append (get st :seen) (list k)) :out (append (get st :out) (list x))})))
{:seen (list) :out (list)}
items)
:out)))
(define
feed/dedupe
(fn
(stream key-fn)
(feed/stream (feed/-dedup-by (feed/items stream) key-fn))))
; --- keys -------------------------------------------------------------------
(define
feed/activity-key
(fn (a) (list (get a :actor) (get a :verb) (get a :object))))
; collapse cross-actor duplicates of the same verb+object (e.g. likes)
(define feed/collapse-key (fn (a) (list (get a :verb) (get a :object))))
; per-receiver inbox key — one inbox event per (receiver, actor, verb, object)
(define
feed/event-key
(fn
(ev)
(let
((a (get ev :activity)))
(list (get ev :to) (get a :actor) (get a :verb) (get a :object)))))
; verbs whose duplicates collapse across actors (reactions, not authorship).
; rebindable: callers can (set! feed/collapse-verbs ...) to tune the policy.
(define
feed/collapse-verbs
(list "like" "favourite" "follow" "boost" "repost"))
; per-verb key: collapse-verbs fold on (verb object); the rest key on
; (actor verb object).
(define
feed/smart-key
(fn
(a)
(if
(feed/-elem? (get a :verb) feed/collapse-verbs)
(feed/collapse-key a)
(feed/activity-key a))))
; --- ready-made dedupers ----------------------------------------------------
(define feed/dedupe-activities (fn (s) (feed/dedupe s feed/activity-key)))
(define feed/dedupe-collapse (fn (s) (feed/dedupe s feed/collapse-key)))
; verb-aware: reactions collapse cross-actor, posts stay distinct per actor
(define feed/dedupe-smart (fn (s) (feed/dedupe s feed/smart-key)))
; dedupe an inbox: at most one event per receiver per (actor verb object)
(define feed/dedupe-inbox (fn (inbox) (feed/dedupe inbox feed/event-key)))

View File

@@ -1,114 +0,0 @@
; feed/fanout — THE SHOWCASE. Fan activities out to followers via the APL outer
; product (∘.×). activities ∘.× audience → an (activity × follower) matrix of
; inbox events; flatten to a vector; guard-keep only real follow edges.
;
; Requires: lib/apl/runtime.sx, lib/feed/normalize.sx, lib/feed/stream.sx.
;
; NOTE: apl-outer's combiner result is run through (if (scalar? r) (disclose r) r).
; A bare dict counts as a scalar (shape ()) and disclose nils it — so the combiner
; must (enclose ...) its event dict; apl-outer then discloses it back intact.
; --- graph: {followee -> (list of followers)} -------------------------------
(define feed/followers (fn (graph user) (get graph user (list))))
; build a graph from (follower followee) edges: "follower follows followee"
(define
feed/follow-graph
(fn
(edges)
(reduce
(fn
(g e)
(let
((follower (first e)) (followee (nth e 1)))
(assoc
g
followee
(append (feed/followers g followee) (list follower)))))
{}
edges)))
; --- helpers ----------------------------------------------------------------
; unwrap an apl-scalar (has :ravel) back to its value; pass activities through
(define
feed/-val
(fn
(x)
(if (and (= (type-of x) "dict") (has-key? x :ravel)) (disclose x) x)))
(define feed/-elem? (fn (x lst) (some (fn (y) (equal? x y)) lst)))
(define
feed/-distinct
(fn
(lst)
(if
(= (len lst) 0)
(list)
(get (apl-unique (make-array (list (len lst)) lst)) :ravel))))
; rank-2 matrix -> rank-1 stream of its ravel
(define feed/-flatten (fn (arr) (feed/stream (get arr :ravel))))
; distinct receivers across the whole graph, sorted for determinism
; (dict key order is unspecified, so sort to pin audience/recipient ordering)
(define
feed/audience
(fn
(graph)
(sort
(feed/-distinct
(reduce
(fn (acc k) (append acc (feed/followers graph k)))
(list)
(keys graph))))))
; --- the outer product ------------------------------------------------------
; one (activity, follower) inbox event, enclosed so apl-outer keeps the dict
(define feed/-mk-event (fn (a f) (enclose {:activity (feed/-val a) :to (feed/-val f)})))
; keep events where :to actually follows the activity's actor
(define
feed/-edge?
(fn
(graph)
(fn
(ev)
(feed/-elem?
(get ev :to)
(feed/followers graph (get (get ev :activity) :actor))))))
; fanout — activities ∘.× audience, flatten, guard-keep real edges
(define
feed/fanout
(fn
(stream graph)
(let
((matrix (apl-outer feed/-mk-event stream (feed/stream (feed/audience graph)))))
(feed/filter (feed/-flatten matrix) (feed/-edge? graph)))))
; --- inbox queries ----------------------------------------------------------
(define
feed/inbox-for
(fn
(inbox user)
(feed/filter inbox (fn (ev) (equal? (get ev :to) user)))))
(define
feed/recipients
(fn
(inbox)
(feed/-distinct (map (fn (ev) (get ev :to)) (feed/items inbox)))))
; the activities (unwrapped) destined for a user
(define
feed/inbox-activities
(fn
(inbox user)
(map
(fn (ev) (get ev :activity))
(feed/items (feed/inbox-for inbox user)))))

View File

@@ -1,60 +0,0 @@
; feed/fed — federation. Outbound: a local post fans out, then splits into local
; vs remote inboxes; remote events are handed to an injected send-fn. Inbound:
; peer activities merge into the local stream, deduped. Backfill: pull peer
; history via an injected fetch-fn and merge.
;
; remote? / send-fn / fetch-fn are injected so real fed-sx transport wires in here
; without feed depending on it.
;
; Requires: lib/feed/normalize.sx, lib/feed/stream.sx, lib/feed/fanout.sx,
; lib/feed/dedupe.sx.
; --- merge / ingest ---------------------------------------------------------
(define
feed/merge
(fn (s1 s2) (feed/stream (append (feed/items s1) (feed/items s2)))))
; merge a peer stream into local, dropping (actor verb object) duplicates
(define
feed/ingest
(fn (local peer) (feed/dedupe-activities (feed/merge local peer))))
; --- inbound ----------------------------------------------------------------
; peer pushes raw activities to the local inbox; normalize + ingest
(define
feed/inbound
(fn
(local raw-activities)
(feed/ingest local (feed/stream (map feed/normalize raw-activities)))))
; backfill on subscribe: pull peer history via fetch-fn, normalize, ingest
(define
feed/backfill
(fn (local fetch-fn peer-id) (feed/inbound local (fetch-fn peer-id))))
; --- outbound ---------------------------------------------------------------
; split an inbox into local vs remote deliveries by viewer-id predicate
(define feed/partition-inbox (fn (inbox remote?) {:local (feed/filter inbox (fn (ev) (not (remote? (get ev :to))))) :remote (feed/filter inbox (fn (ev) (remote? (get ev :to))))}))
; fan a stream out over the graph, then partition by locality
(define
feed/federate
(fn
(stream graph remote?)
(feed/partition-inbox (feed/fanout stream graph) remote?)))
; deliver: hand each remote event to send-fn, return the local inbox to enqueue
(define
feed/deliver
(fn
(stream graph remote? send-fn)
(let
((parts (feed/federate stream graph remote?)))
(begin
(for-each
(fn (ev) (send-fn (get ev :to) (get ev :activity)))
(feed/items (get parts :remote)))
(get parts :local)))))

View File

@@ -1,23 +0,0 @@
; feed/home — the capstone. A user's home timeline is the whole pipeline as one
; line: fan all activities out over the follow graph, take the events landing in
; the viewer's inbox, dedupe cross-posts, apply the viewer's ACL, rank, take N.
;
; Requires: fanout.sx, dedupe.sx, acl.sx (feed/timeline), rank.sx, stream.sx.
; the activities in a user's inbox, as a stream
(define
feed/inbox-stream
(fn (inbox user) (feed/stream (feed/inbox-activities inbox user))))
; fanout ∘ inbox ∘ dedupe ∘ ACL ∘ rank ∘ take
(define
feed/home
(fn
(stream graph viewer permit? score-fn n)
(feed/timeline
(feed/dedupe-activities
(feed/inbox-stream (feed/fanout stream graph) viewer))
viewer
permit?
score-fn
n)))

View File

@@ -1,44 +0,0 @@
; feed/mute — viewer-controlled filtering. ACL (acl.sx) is author-controlled
; visibility; mute is the reader's own preference: hide muted actors or tags.
; Like ACL it is per-viewer and applied per request, never cached.
;
; Requires: lib/feed/normalize.sx, lib/feed/stream.sx, lib/feed/fanout.sx
; (feed/-elem?).
; drop activities authored by a muted actor
(define
feed/mute-actors
(fn
(stream actors)
(feed/filter
stream
(fn (a) (not (feed/-elem? (get a :actor) actors))))))
; drop activities carrying any muted tag
(define
feed/mute-tags
(fn
(stream tags)
(feed/filter
stream
(fn (a) (not (some (fn (t) (feed/-elem? t tags)) (get a :tags)))))))
; drop activities about a muted object (thread mute)
(define
feed/mute-objects
(fn
(stream objects)
(feed/filter
stream
(fn (a) (not (feed/-elem? (get a :object) objects))))))
; apply a viewer preference bag: {:mute-actors (...) :mute-tags (...) :mute-objects (...)}
(define
feed/apply-prefs
(fn
(stream prefs)
(feed/mute-objects
(feed/mute-tags
(feed/mute-actors stream (get prefs :mute-actors (list)))
(get prefs :mute-tags (list)))
(get prefs :mute-objects (list)))))

View File

@@ -1,31 +0,0 @@
; feed/normalize — coerce arbitrary input into the canonical activity record.
; An activity is a small dict {:actor :verb :object :at :tags}; a stream is an
; APL vector of such dicts (see stream.sx). Extra keys on the raw input survive
; (e.g. :visible-to for ACL, peer metadata for federation) — :tags is the
; flexible bag but the record is not closed.
(define feed/activity-keys (list :actor :verb :object :at :tags))
(define
feed/normalize
(fn
(raw)
(let
((d (if (= (type-of raw) "dict") raw {})))
(merge d {:actor (get d :actor "") :object (get d :object nil) :at (get d :at 0) :tags (let ((t (get d :tags (list)))) (if (list? t) t (list t))) :verb (get d :verb "post")}))))
(define
feed/activity
(fn (actor verb object at tags) (feed/normalize {:actor actor :object object :at at :tags tags :verb verb})))
(define feed/actor (fn (a) (get a :actor)))
(define feed/verb (fn (a) (get a :verb)))
(define feed/object (fn (a) (get a :object)))
(define feed/at (fn (a) (get a :at)))
(define feed/tags (fn (a) (get a :tags)))
(define
feed/activity?
(fn
(a)
(and (= (type-of a) "dict") (has-key? a :actor) (has-key? a :verb))))

View File

@@ -1,45 +0,0 @@
; feed/notify — a notification feed is a thin layer over a recipient's inbox:
; the events directed at a user, optionally verb-filtered, and a digest that
; collapses "alice, bob and 1 other liked X" by (verb, object).
;
; Requires: lib/feed/normalize.sx, lib/feed/stream.sx, lib/feed/fanout.sx
; (feed/inbox-for, feed/-elem?).
; all inbox events for a user (their raw notifications)
(define feed/notifications (fn (inbox user) (feed/inbox-for inbox user)))
; restrict to notification-worthy verbs (e.g. (list "like" "reply" "follow"))
(define
feed/notify-verbs
(fn
(inbox user verbs)
(feed/filter
(feed/inbox-for inbox user)
(fn (ev) (feed/-elem? (get (get ev :activity) :verb) verbs)))))
; group key "verb|object" — deterministic, sortable
(define
feed/-notify-key
(fn
(ev)
(let
((a (get ev :activity)))
(string-append (get a :verb) "|" (get a :object)))))
; digest: one entry per (verb, object) with the distinct actors and a count,
; ordered by key for determinism.
(define
feed/notify-digest
(fn
(inbox user)
(let
((events (feed/items (feed/inbox-for inbox user))))
(let
((groups (reduce (fn (g ev) (let ((a (get ev :activity)) (k (feed/-notify-key ev))) (let ((cur (get g k {:object (get a :object) :actors (list) :verb (get a :verb)}))) (assoc g k (assoc cur :actors (append (get cur :actors) (list (get a :actor)))))))) {} events)))
(map
(fn
(k)
(let
((grp (get groups k)))
(assoc grp :count (len (get grp :actors)))))
(sort (keys groups)))))))

View File

@@ -1,50 +0,0 @@
; feed/page — pagination. Offset/limit for indexed access, and cursor-based
; (by :at) for recency feeds, which is stable under inserts: a cursor is the
; :at of the last item seen, and the next page is the newest items older than it.
;
; Requires: lib/feed/stream.sx (feed/recent, feed/take, feed/filter).
; --- offset / limit ---------------------------------------------------------
(define
feed/page
(fn
(stream offset limit)
(feed/stream (take (drop (feed/items stream) offset) limit))))
(define
feed/page-count
(fn (stream limit) (ceil (/ (feed/count stream) limit))))
; --- cursor (recency feeds) -------------------------------------------------
; activities strictly older than cursor (scroll down / load older)
(define
feed/before
(fn
(stream cursor)
(feed/filter stream (fn (a) (< (get a :at) cursor)))))
; activities strictly newer than cursor (load newer / "N new posts")
(define
feed/after
(fn
(stream cursor)
(feed/filter stream (fn (a) (> (get a :at) cursor)))))
; one page: the `limit` newest activities older than cursor, newest first
(define
feed/page-before
(fn
(stream cursor limit)
(feed/take (feed/recent (feed/before stream cursor)) limit)))
; cursor to fetch the next (older) page: :at of the last item of a page,
; or nil when the page is empty (end of feed)
(define
feed/next-cursor
(fn
(page)
(let
((items (feed/items page)))
(if (= (len items) 0) nil (get (last items) :at)))))

View File

@@ -1,92 +0,0 @@
; feed/rank — scoring + ranking. Scorers are (activity -> number). Ranking is a
; stable two-pass grade-down: first by :at descending (the tiebreak), then by
; score descending — so ties resolve by recency, then by input order. Fully
; deterministic on ties.
;
; Requires: lib/apl/runtime.sx, lib/feed/normalize.sx, lib/feed/stream.sx.
; --- scorers ----------------------------------------------------------------
; recency: half-life decay. score = 0.5 ^ (age / half-life). at==now -> 1.0.
(define
feed/recency
(fn
(now half-life)
(fn (a) (expt 0.5 (/ (- now (get a :at)) half-life)))))
; velocity: how many of this actor's activities fall in (at-window, at] —
; a burst of recent activity scores higher.
(define
feed/velocity
(fn
(stream window)
(fn
(a)
(len
(filter
(fn
(b)
(and
(equal? (get b :actor) (get a :actor))
(<= (get b :at) (get a :at))
(> (get b :at) (- (get a :at) window))))
(feed/items stream))))))
; engagement: how many activities in the stream touch this activity's :object
(define
feed/engagement
(fn
(stream)
(fn
(a)
(len
(filter
(fn (b) (equal? (get b :object) (get a :object)))
(feed/items stream))))))
; composite: weighted sum. parts = (list (list weight scorer) ...)
(define
feed/composite
(fn
(parts)
(fn
(a)
(reduce
(fn (acc p) (+ acc (* (first p) ((nth p 1) a))))
0
parts))))
; --- ranking ----------------------------------------------------------------
; stable reorder of items by key-fn, descending (grade-down is stable)
(define
feed/-desc-by
(fn
(items key-fn)
(let
((keys (make-array (list (len items)) (map key-fn items))))
(let
((order (get (apl-grade-down keys) :ravel)))
(map (fn (i) (nth items (- i 1))) order)))))
; rank by score descending; ties -> :at descending -> input order
(define
feed/rank
(fn
(stream score-fn)
(let
((by-at (feed/-desc-by (feed/items stream) feed/at)))
(feed/stream (feed/-desc-by by-at score-fn)))))
; attach a :score to each activity (for inspection / debugging)
(define
feed/with-scores
(fn
(stream score-fn)
(feed/stream
(map (fn (a) (assoc a :score (score-fn a))) (feed/items stream)))))
; top-N ranked timeline
(define
feed/top
(fn (stream score-fn n) (feed/take (feed/rank stream score-fn) n)))

View File

@@ -1,19 +0,0 @@
{
"suites": {
"basic": {"pass": 30, "fail": 0},
"fanout": {"pass": 29, "fail": 0},
"rank": {"pass": 24, "fail": 0},
"integration": {"pass": 22, "fail": 0},
"content": {"pass": 15, "fail": 0},
"notify": {"pass": 8, "fail": 0},
"home": {"pass": 6, "fail": 0},
"dedupe": {"pass": 9, "fail": 0},
"trending": {"pass": 11, "fail": 0},
"mute": {"pass": 9, "fail": 0},
"page": {"pass": 14, "fail": 0},
"thread": {"pass": 12, "fail": 0}
},
"total_pass": 189,
"total_fail": 0,
"total": 189
}

View File

@@ -1,19 +0,0 @@
# feed Conformance Scoreboard
_Generated by `lib/feed/conformance.sh`_
| Suite | Pass | Fail | Total |
|-------|-----:|-----:|------:|
| basic | 30 | 0 | 30 |
| fanout | 29 | 0 | 29 |
| rank | 24 | 0 | 24 |
| integration | 22 | 0 | 22 |
| content | 15 | 0 | 15 |
| notify | 8 | 0 | 8 |
| home | 6 | 0 | 6 |
| dedupe | 9 | 0 | 9 |
| trending | 11 | 0 | 11 |
| mute | 9 | 0 | 9 |
| page | 14 | 0 | 14 |
| thread | 12 | 0 | 12 |
| **Total** | **189** | **0** | **189** |

View File

@@ -1,75 +0,0 @@
; feed/stream — a stream is an APL vector (rank-1 array) whose ravel holds
; activity dicts. Operations lift APL primitives onto this shape: filter via
; compress (/), sort via grade (⍋), take via ↑, reverse via ⌽.
;
; Requires: lib/apl/runtime.sx, lib/feed/normalize.sx (loaded by harness).
(define feed/stream (fn (acts) (make-array (list (len acts)) acts)))
(define feed/items (fn (s) (get s :ravel)))
(define feed/count (fn (s) (len (get s :ravel))))
(define feed/empty (feed/stream (list)))
(define feed/empty? (fn (s) (= (feed/count s) 0)))
; filter — bool mask ∘ compress. pred : activity -> truthy
(define
feed/filter
(fn
(s pred)
(let
((items (get s :ravel)))
(let
((mask (make-array (list (len items)) (map (fn (a) (if (pred a) 1 0)) items))))
(apl-compress mask s)))))
; sort-by — ascending, stable on ties (grade-up is stable). key-fn : activity -> number
(define
feed/sort-by
(fn
(s key-fn)
(let
((items (get s :ravel)))
(let
((keys (make-array (list (len items)) (map key-fn items))))
(let
((order (get (apl-grade-up keys) :ravel)))
(feed/stream (map (fn (i) (nth items (- i 1))) order)))))))
(define feed/sort-by-at (fn (s) (feed/sort-by s feed/at)))
; newest-first: ascending sort then reverse (⌽)
(define feed/recent (fn (s) (apl-reverse (feed/sort-by-at s))))
; take N (↑), clamped to stream length so it never over-takes/pads
(define
feed/take
(fn
(s n)
(let
((c (feed/count s)))
(if (>= n c) s (apl-take (apl-scalar n) s)))))
(define feed/reverse (fn (s) (apl-reverse s)))
; common predicates
(define
feed/by-actor
(fn (s actor) (feed/filter s (fn (a) (equal? (get a :actor) actor)))))
(define
feed/by-verb
(fn (s verb) (feed/filter s (fn (a) (equal? (get a :verb) verb)))))
(define
feed/by-object
(fn
(s object)
(feed/filter s (fn (a) (equal? (get a :object) object)))))
; activities at or after timestamp t
(define
feed/since
(fn (s t) (feed/filter s (fn (a) (>= (get a :at) t)))))

View File

@@ -1,118 +0,0 @@
; Phase 1 — normalize, stream ops, api. Uses the feed-test harness
; (feed-test name got expected) provided by conformance.sh.
; ---------- normalize ----------
(feed-test
"normalize default actor"
(feed/actor (feed/normalize {}))
"")
(feed-test
"normalize default verb"
(feed/verb (feed/normalize {}))
"post")
(feed-test
"normalize default at"
(feed/at (feed/normalize {}))
0)
(feed-test
"normalize default object"
(feed/object (feed/normalize {}))
nil)
(feed-test
"normalize default tags"
(feed/tags (feed/normalize {}))
(list))
(feed-test
"normalize keeps actor"
(feed/actor (feed/normalize {:actor "alice"}))
"alice")
(feed-test
"normalize keeps verb"
(feed/verb (feed/normalize {:verb "like"}))
"like")
(feed-test
"normalize scalar tag -> list"
(feed/tags (feed/normalize {:tags "x"}))
(list "x"))
(feed-test
"normalize list tags kept"
(feed/tags (feed/normalize {:tags (list "a" "b")}))
(list "a" "b"))
(feed-test
"activity constructor at"
(feed/at (feed/activity "a" "post" "o" 5 (list)))
5)
(feed-test
"activity? on activity"
(feed/activity? (feed/normalize {:actor "a"}))
true)
(feed-test "activity? on number" (feed/activity? 5) false)
(feed-test "activity? on bare dict" (feed/activity? {:foo 1}) false)
; ---------- stream ----------
(define
S
(feed/stream
(list
(feed/activity "alice" "post" "p1" 30 (list))
(feed/activity "bob" "like" "p1" 10 (list))
(feed/activity "alice" "post" "p2" 20 (list)))))
(feed-test "stream count" (feed/count S) 3)
(feed-test "stream items len" (len (feed/items S)) 3)
(feed-test
"sort-by-at actors asc"
(map feed/actor (feed/items (feed/sort-by-at S)))
(list "bob" "alice" "alice"))
(feed-test
"recent newest first"
(map feed/at (feed/items (feed/recent S)))
(list 30 20 10))
(feed-test
"take 2 of recent"
(feed/count (feed/take (feed/recent S) 2))
2)
(feed-test
"take clamps past end"
(feed/count (feed/take S 10))
3)
(feed-test
"by-actor alice count"
(feed/count (feed/by-actor S "alice"))
2)
(feed-test
"by-verb like actor"
(map feed/actor (feed/items (feed/by-verb S "like")))
(list "bob"))
(feed-test
"by-object p1 count"
(feed/count (feed/by-object S "p1"))
2)
(feed-test
"since 20 count"
(feed/count (feed/since S 20))
2)
(feed-test
"reverse ats"
(map feed/at (feed/items (feed/reverse S)))
(list 20 10 30))
(feed-test "empty? on empty" (feed/empty? feed/empty) true)
(feed-test
"empty? on filtered-out"
(feed/empty? (feed/by-actor S "zzz"))
true)
; ---------- api ----------
(feed/reset!)
(feed/post {:actor "x" :at 1 :verb "post"})
(feed/post {:actor "y" :at 2 :verb "like"})
(feed-test "api size after posts" (feed/size) 2)
(feed-test "api all count" (feed/count (feed/all)) 2)
(feed-test
"post returns normalized verb"
(feed/verb (feed/post {:actor "z"}))
"post")
(feed-test "api size after third post" (feed/size) 3)

View File

@@ -1,85 +0,0 @@
; Follow-up — TF-IDF content ranking over :tags. (feed-test name got expected)
(define
corpus
(feed/stream
(list
(feed/normalize {:actor "u" :object "o1" :at 10 :tags (list "cats" "funny")})
(feed/normalize {:actor "u" :object "o2" :at 20 :tags (list "cats" "news")})
(feed/normalize {:actor "u" :object "o3" :at 30 :tags (list "politics" "news")})
(feed/normalize {:actor "u" :object "o4" :at 40 :tags (list "cats")}))))
; ---------- document frequency ----------
(feed-test "df cats" (get (feed/tag-df corpus) "cats") 3)
(feed-test "df news" (get (feed/tag-df corpus) "news") 2)
(feed-test "df funny" (get (feed/tag-df corpus) "funny") 1)
(feed-test "df politics" (get (feed/tag-df corpus) "politics") 1)
(feed-test "df full" (feed/tag-df corpus) {:news 2 :funny 1 :politics 1 :cats 3})
; ---------- inverse document frequency ----------
(feed-test
"idf news = log(4/2)"
(get (feed/tag-idf corpus) "news")
(log 2))
(feed-test
"idf funny = log(4/1)"
(get (feed/tag-idf corpus) "funny")
(log 4))
(feed-test
"rarer tag has higher idf"
(>
(get (feed/tag-idf corpus) "funny")
(get (feed/tag-idf corpus) "cats"))
true)
; ---------- tf-idf scoring ----------
(define idf (feed/tag-idf corpus))
(feed-test
"score query funny on o1"
((feed/tfidf-score idf (list "funny")) (feed/normalize {:actor "u" :object "x" :tags (list "cats" "funny")}))
(log 4))
(feed-test
"score query funny on non-match"
((feed/tfidf-score idf (list "funny")) (feed/normalize {:actor "u" :object "x" :tags (list "cats")}))
0)
(feed-test
"unknown query tag scores 0"
((feed/tfidf-score idf (list "zzz")) (feed/normalize {:actor "u" :object "x" :tags (list "cats")}))
0)
; ---------- ranking by relevance ----------
; query news: o2,o3 match (score log2), o1,o4 don't (0); ties break by :at desc
(feed-test
"by-relevance news order"
(map
(fn (a) (get a :object))
(feed/items (feed/by-relevance corpus (list "news"))))
(list "o3" "o2" "o4" "o1"))
; query funny: only o1 matches -> ranks first
(feed-test
"by-relevance funny first"
(get
(nth (feed/items (feed/by-relevance corpus (list "funny"))) 0)
:object)
"o1")
; query (cats news): o2 carries both tags -> highest combined tf-idf
(feed-test
"by-relevance cats+news top"
(get
(nth
(feed/items (feed/by-relevance corpus (list "cats" "news")))
0)
:object)
"o2")
(feed-test
"by-relevance preserves count"
(feed/count (feed/by-relevance corpus (list "cats")))
4)

View File

@@ -1,56 +0,0 @@
; Follow-up — verb-aware (smart) dedupe. (feed-test name got expected)
; reactions (like/follow) collapse cross-actor; posts stay distinct per actor
(define
M
(feed/stream
(list
(feed/activity "alice" "like" "X" 1 (list))
(feed/activity "bob" "like" "X" 2 (list))
(feed/activity "alice" "post" "P" 3 (list))
(feed/activity "bob" "post" "P" 4 (list))
(feed/activity "alice" "follow" "C" 5 (list))
(feed/activity "bob" "follow" "C" 6 (list))))) ; collapses
(feed-test
"smart dedupe total"
(feed/count (feed/dedupe-smart M))
4)
(feed-test
"smart keeps both posts"
(feed/count (feed/by-verb (feed/dedupe-smart M) "post"))
2)
(feed-test
"smart collapses likes to one"
(feed/count (feed/by-verb (feed/dedupe-smart M) "like"))
1)
(feed-test
"smart collapses follows to one"
(feed/count (feed/by-verb (feed/dedupe-smart M) "follow"))
1)
(feed-test
"collapsed like keeps first actor"
(map feed/actor (feed/items (feed/by-verb (feed/dedupe-smart M) "like")))
(list "alice"))
; contrast: plain activity dedupe keeps cross-actor likes distinct
(feed-test
"activity dedupe keeps both likes"
(feed/count (feed/by-verb (feed/dedupe-activities M) "like"))
2)
; contrast: blanket collapse folds the two posts (same verb+object) too
(feed-test
"collapse dedupe folds posts"
(feed/count (feed/by-verb (feed/dedupe-collapse M) "post"))
1)
; smart-key dispatch
(feed-test
"smart-key reaction -> (verb object)"
(feed/smart-key (feed/activity "alice" "like" "X" 0 (list)))
(list "like" "X"))
(feed-test
"smart-key post -> (actor verb object)"
(feed/smart-key (feed/activity "alice" "post" "P" 0 (list)))
(list "alice" "post" "P"))

View File

@@ -1,187 +0,0 @@
; Phase 2 — fanout via outer product + dedupe. (feed-test name got expected)
; ---------- graph ----------
; edges: (follower followee). bob,carol follow alice; carol,dave follow bob.
(define
G
(feed/follow-graph
(list
(list "bob" "alice")
(list "carol" "alice")
(list "carol" "bob")
(list "dave" "bob"))))
(feed-test "followers alice" (feed/followers G "alice") (list "bob" "carol"))
(feed-test "followers bob" (feed/followers G "bob") (list "carol" "dave"))
(feed-test "followers unknown" (feed/followers G "zzz") (list))
(feed-test "audience distinct" (feed/audience G) (list "bob" "carol" "dave"))
; ---------- fanout ----------
(define
S
(feed/stream
(list
(feed/activity "alice" "post" "p1" 10 (list))
(feed/activity "alice" "post" "p2" 20 (list))
(feed/activity "bob" "like" "p1" 30 (list)))))
(define IB (feed/fanout S G))
(feed-test "fanout total edges" (feed/count IB) 6)
(feed-test
"inbox bob count"
(feed/count (feed/inbox-for IB "bob"))
2)
(feed-test
"inbox carol count"
(feed/count (feed/inbox-for IB "carol"))
3)
(feed-test
"inbox dave count"
(feed/count (feed/inbox-for IB "dave"))
1)
(feed-test
"inbox alice (follows none)"
(feed/count (feed/inbox-for IB "alice"))
0)
(feed-test
"recipients order"
(feed/recipients IB)
(list "bob" "carol" "dave"))
(feed-test
"bob inbox objects"
(map (fn (a) (get a :object)) (feed/inbox-activities IB "bob"))
(list "p1" "p2"))
(feed-test
"dave inbox objects"
(map (fn (a) (get a :object)) (feed/inbox-activities IB "dave"))
(list "p1"))
(feed-test
"dave inbox verb"
(map (fn (a) (get a :verb)) (feed/inbox-activities IB "dave"))
(list "like"))
; empty graph → no audience → no edges
(feed-test
"empty graph fanout"
(feed/count (feed/fanout S {}))
0)
; actor nobody follows produces no edges
(define
Sghost
(feed/stream (list (feed/activity "ghost" "post" "g1" 5 (list)))))
(feed-test
"unfollowed actor fanout"
(feed/count (feed/fanout Sghost G))
0)
; ---------- high fanout (popular actor) ----------
(define
Gstar
(feed/follow-graph
(list
(list "u1" "star")
(list "u2" "star")
(list "u3" "star")
(list "u4" "star")
(list "u5" "star"))))
(define
Sstar
(feed/stream (list (feed/activity "star" "post" "s1" 1 (list)))))
(feed-test
"star fanout count"
(feed/count (feed/fanout Sstar Gstar))
5)
(feed-test "star audience size" (len (feed/audience Gstar)) 5)
; ---------- mutual follow ----------
(define Gmut (feed/follow-graph (list (list "a" "b") (list "b" "a"))))
(define
Smut
(feed/stream
(list
(feed/activity "a" "post" "pa" 1 (list))
(feed/activity "b" "post" "pb" 2 (list)))))
(define IBmut (feed/fanout Smut Gmut))
(feed-test "mutual total" (feed/count IBmut) 2)
(feed-test
"mutual a gets pb"
(map (fn (x) (get x :object)) (feed/inbox-activities IBmut "a"))
(list "pb"))
(feed-test
"mutual b gets pa"
(map (fn (x) (get x :object)) (feed/inbox-activities IBmut "b"))
(list "pa"))
; ---------- dedupe ----------
(define
Sdup2
(feed/stream
(list
(feed/activity "alice" "post" "p1" 1 (list))
(feed/activity "alice" "post" "p1" 9 (list))
(feed/activity "alice" "post" "p2" 2 (list)))))
(feed-test
"dedupe-activities collapses dup"
(feed/count (feed/dedupe-activities Sdup2))
2)
(feed-test
"dedupe-activities keeps distinct"
(map
(fn (a) (get a :object))
(feed/items (feed/dedupe-activities Sdup2)))
(list "p1" "p2"))
(define
Slikes
(feed/stream
(list
(feed/activity "alice" "like" "X" 1 (list))
(feed/activity "bob" "like" "X" 2 (list))
(feed/activity "carol" "like" "Y" 3 (list)))))
(feed-test
"collapse cross-actor likes"
(feed/count (feed/dedupe-collapse Slikes))
2)
(feed-test
"collapse keeps distinct objects"
(map
(fn (a) (get a :object))
(feed/items (feed/dedupe-collapse Slikes)))
(list "X" "Y"))
(feed-test
"activity-key shape"
(feed/activity-key (feed/activity "a" "post" "o" 0 (list)))
(list "a" "post" "o"))
(feed-test
"collapse-key shape"
(feed/collapse-key (feed/activity "a" "like" "o" 0 (list)))
(list "like" "o"))
; cross-post: alice posts p1 twice → bob's inbox has it twice → dedupe-inbox → once
(define
Scross
(feed/stream
(list
(feed/activity "alice" "post" "p1" 1 (list))
(feed/activity "alice" "post" "p1" 5 (list)))))
(define IBcross (feed/fanout Scross G))
(feed-test
"cross-post raw bob count"
(feed/count (feed/inbox-for IBcross "bob"))
2)
(feed-test
"cross-post deduped bob count"
(feed/count (feed/inbox-for (feed/dedupe-inbox IBcross) "bob"))
1)
(feed-test
"dedupe-inbox keeps distinct receivers"
(feed/count (feed/dedupe-inbox IBcross))
2)

View File

@@ -1,73 +0,0 @@
; Follow-up — feed/home capstone pipeline. (feed-test name got expected)
; alice follows star and bob (edges: follower followee)
(define
G
(feed/follow-graph (list (list "alice" "star") (list "alice" "bob"))))
; star posts s1 then s2; bob posts b1; star re-posts s1 (cross-post dup);
; zoe posts z1 (alice does NOT follow zoe)
(define
S
(feed/stream
(list
(feed/activity "star" "post" "s1" 10 (list))
(feed/activity "star" "post" "s2" 20 (list))
(feed/activity "bob" "post" "b1" 15 (list))
(feed/activity "star" "post" "s1" 5 (list))
(feed/activity "zoe" "post" "z1" 30 (list)))))
(define rec (feed/recency 100 10))
(feed-test
"home count (deduped, followed only)"
(feed/count (feed/home S G "alice" feed/permit-public? rec 10))
3)
(feed-test
"home order by recency"
(map
(fn (a) (get a :object))
(feed/items (feed/home S G "alice" feed/permit-public? rec 10)))
(list "s2" "b1" "s1"))
(feed-test
"home excludes unfollowed zoe"
(feed/-elem?
"z1"
(map
(fn (a) (get a :object))
(feed/items (feed/home S G "alice" feed/permit-public? rec 10))))
false)
(feed-test
"home top-2"
(map
(fn (a) (get a :object))
(feed/items (feed/home S G "alice" feed/permit-public? rec 2)))
(list "s2" "b1"))
(feed-test
"home dedupes cross-post (one s1)"
(len
(filter
(fn (o) (equal? o "s1"))
(map
(fn (a) (get a :object))
(feed/items
(feed/home S G "alice" feed/permit-public? rec 10)))))
1)
; ACL applied per-viewer in the home pipeline
(define
Sacl
(feed/stream
(list (feed/normalize {:actor "star" :object "pub" :at 20}) (feed/normalize {:actor "star" :object "sec" :visible-to (list "carol") :at 25}))))
(define Gacl (feed/follow-graph (list (list "alice" "star"))))
(feed-test
"home hides activity alice not permitted"
(map
(fn (a) (get a :object))
(feed/items (feed/home Sacl Gacl "alice" feed/permit-acl? rec 10)))
(list "pub"))

View File

@@ -1,155 +0,0 @@
; Phase 4 — visibility (ACL) + federation, and the end-to-end timeline.
; (feed-test name got expected)
; ---------- ACL visibility ----------
; pub: public. sec: bob, allows carol. dm: frank, allows dave.
(define
C
(feed/stream
(list
(feed/normalize {:actor "alice" :object "pub" :at 10})
(feed/normalize {:actor "bob" :object "sec" :visible-to (list "carol") :at 20})
(feed/normalize {:actor "frank" :object "dm" :visible-to (list "dave") :at 30}))))
(feed-test
"public visible to anyone"
(feed/count (feed/visible C "zoe" feed/permit-acl?))
1)
(feed-test
"carol sees allowlisted + public"
(feed/count (feed/visible C "carol" feed/permit-acl?))
2)
(feed-test
"dave sees dm + public"
(feed/count (feed/visible C "dave" feed/permit-acl?))
2)
(feed-test
"author always sees own private"
(feed/count (feed/visible C "frank" feed/permit-acl?))
2)
(feed-test
"permit-public? lets all through"
(feed/count (feed/visible C "zoe" feed/permit-public?))
3)
(feed-test
"visible objects for dave"
(map
(fn (a) (get a :object))
(feed/items (feed/visible C "dave" feed/permit-acl?)))
(list "pub" "dm"))
; per-viewer: same stream, different timelines
(feed-test
"zoe timeline differs from carol"
(not
(=
(feed/count (feed/visible C "zoe" feed/permit-acl?))
(feed/count (feed/visible C "carol" feed/permit-acl?))))
true)
; ---------- federation: merge / ingest ----------
(define
L
(feed/stream
(list
(feed/activity "alice" "post" "p1" 10 (list))
(feed/activity "alice" "post" "p2" 20 (list)))))
(define
P
(feed/stream
(list
(feed/activity "alice" "post" "p2" 20 (list))
(feed/activity "peer" "post" "p9" 25 (list)))))
(feed-test "merge concatenates" (feed/count (feed/merge L P)) 4)
(feed-test
"ingest dedupes overlap"
(feed/count (feed/ingest L P))
3)
(feed-test
"inbound normalizes + ingests"
(feed/count (feed/inbound L (list {:actor "peer" :object "p9" :at 25} {:actor "alice" :object "p1" :at 10})))
3)
; backfill via injected fetch-fn
(define peer-history (fn (peer-id) (list {:actor peer-id :object "h1" :at 1} {:actor peer-id :object "h2" :at 2})))
(feed-test
"backfill merges peer history"
(feed/count (feed/backfill L peer-history "remote"))
4)
(feed-test
"backfill objects present"
(map
(fn (a) (get a :object))
(feed/items
(feed/by-actor (feed/backfill L peer-history "remote") "remote")))
(list "h1" "h2"))
; ---------- federation: outbound partition ----------
; bob (local), alice@remote + carol@remote (remote) follow star
(define
Gf
(feed/follow-graph
(list
(list "bob" "star")
(list "alice@remote" "star")
(list "carol@remote" "star"))))
(define
Sf
(feed/stream (list (feed/activity "star" "post" "s1" 1 (list)))))
(define
remote?
(fn (id) (feed/-elem? id (list "alice@remote" "carol@remote"))))
(define parts (feed/federate Sf Gf remote?))
(feed-test "local deliveries" (feed/count (get parts :local)) 1)
(feed-test "remote deliveries" (feed/count (get parts :remote)) 2)
(feed-test
"local recipient is bob"
(feed/recipients (get parts :local))
(list "bob"))
; deliver: send-fn receives each remote event, local inbox returned
(define sent (list))
(define send-fn (fn (to act) (set! sent (append sent (list to)))))
(define local-inbox (feed/deliver Sf Gf remote? send-fn))
(feed-test "deliver returns local inbox" (feed/count local-inbox) 1)
(feed-test "deliver sent to both remotes" (len sent) 2)
(feed-test "deliver remote targets" sent (list "alice@remote" "carol@remote"))
; ---------- end-to-end: federated, ACL-filtered, ranked timeline ----------
(define
base
(feed/stream
(list
(feed/normalize {:actor "alice" :object "a1" :at 100})
(feed/normalize {:actor "bob" :object "b1" :visible-to (list "carol") :at 90})
(feed/normalize {:actor "eve" :object "e1" :visible-to (list "dave") :at 80}))))
(define federated (feed/inbound base (list {:actor "peer" :object "x1" :at 110})))
(define rec (feed/recency 120 10))
(define
carol-tl
(feed/timeline federated "carol" feed/permit-acl? rec 3))
; eve's :visible-to excludes carol -> filtered out; peer/alice public, bob allows carol
(feed-test "carol federated timeline count" (feed/count carol-tl) 3)
(feed-test
"carol timeline order (recency)"
(map (fn (a) (get a :object)) (feed/items carol-tl))
(list "x1" "a1" "b1"))
(feed-test
"eve dm excluded from carol"
(feed/-elem? "e1" (map (fn (a) (get a :object)) (feed/items carol-tl)))
false)
(feed-test
"dave sees eve dm not bob"
(map
(fn (a) (get a :object))
(feed/items
(feed/timeline federated "dave" feed/permit-acl? rec 5)))
(list "x1" "a1" "e1"))

View File

@@ -1,68 +0,0 @@
; Follow-up — viewer mute/block filtering. (feed-test name got expected)
(define
S
(feed/stream
(list
(feed/normalize {:actor "alice" :object "P1" :at 1 :tags (list "news")})
(feed/normalize {:actor "bob" :object "P2" :at 2 :tags (list "spam")})
(feed/normalize {:actor "alice" :object "P3" :at 3 :tags (list "cats")})
(feed/normalize {:actor "carol" :object "P4" :at 4 :tags (list "news" "spam")}))))
; ---------- mute actors ----------
(feed-test
"mute bob drops his post"
(map
(fn (a) (get a :object))
(feed/items (feed/mute-actors S (list "bob"))))
(list "P1" "P3" "P4"))
(feed-test
"mute alice drops two"
(feed/count (feed/mute-actors S (list "alice")))
2)
(feed-test
"mute nobody keeps all"
(feed/count (feed/mute-actors S (list)))
4)
; ---------- mute tags ----------
(feed-test
"mute spam tag drops two"
(map
(fn (a) (get a :object))
(feed/items (feed/mute-tags S (list "spam"))))
(list "P1" "P3"))
(feed-test
"mute news+cats leaves spam-only"
(map
(fn (a) (get a :object))
(feed/items (feed/mute-tags S (list "news" "cats"))))
(list "P2"))
; ---------- mute objects ----------
(feed-test
"mute object P3 (thread mute)"
(feed/count (feed/mute-objects S (list "P3")))
3)
; ---------- combined prefs ----------
(feed-test
"apply-prefs actors + tags"
(map
(fn (a) (get a :object))
(feed/items (feed/apply-prefs S {:mute-actors (list "bob") :mute-tags (list "cats")})))
(list "P1" "P4"))
(feed-test
"apply-prefs empty keeps all"
(feed/count (feed/apply-prefs S {}))
4)
(feed-test
"apply-prefs all three filters"
(map
(fn (a) (get a :object))
(feed/items (feed/apply-prefs S {:mute-objects (list "P3") :mute-actors (list "carol") :mute-tags (list "spam")})))
(list "P1"))

View File

@@ -1,69 +0,0 @@
; Follow-up — notification feed over an inbox. (feed-test name got expected)
; an inbox is a stream of {:to receiver :activity act} events
(define mk-ev (fn (to act) {:activity act :to to}))
(define
IB
(feed/stream
(list
(mk-ev "alice" (feed/activity "bob" "like" "P" 10 (list)))
(mk-ev "alice" (feed/activity "carol" "like" "P" 20 (list)))
(mk-ev "alice" (feed/activity "dave" "reply" "Q" 30 (list)))
(mk-ev "bob" (feed/activity "eve" "like" "R" 40 (list))))))
; ---------- raw notifications ----------
(feed-test
"alice notification count"
(feed/count (feed/notifications IB "alice"))
3)
(feed-test
"bob notification count"
(feed/count (feed/notifications IB "bob"))
1)
(feed-test
"zoe no notifications"
(feed/count (feed/notifications IB "zoe"))
0)
; ---------- verb filtering ----------
(feed-test
"alice likes only"
(feed/count (feed/notify-verbs IB "alice" (list "like")))
2)
(feed-test
"alice replies only"
(feed/count (feed/notify-verbs IB "alice" (list "reply")))
1)
(feed-test
"alice like+reply"
(feed/count (feed/notify-verbs IB "alice" (list "like" "reply")))
3)
(feed-test
"alice follow (none)"
(feed/count (feed/notify-verbs IB "alice" (list "follow")))
0)
; ---------- digest ----------
(define dig (feed/notify-digest IB "alice"))
(feed-test "digest group count" (len dig) 2)
(feed-test
"digest sorted by key (like|P before reply|Q)"
(map (fn (g) (get g :object)) dig)
(list "P" "Q"))
(feed-test
"like group actors"
(get (nth dig 0) :actors)
(list "bob" "carol"))
(feed-test "like group count" (get (nth dig 0) :count) 2)
(feed-test "like group verb" (get (nth dig 0) :verb) "like")
(feed-test "reply group count" (get (nth dig 1) :count) 1)
(feed-test
"reply group actors"
(get (nth dig 1) :actors)
(list "dave"))
(feed-test "empty digest for zoe" (feed/notify-digest IB "zoe") (list))

View File

@@ -1,86 +0,0 @@
; Follow-up — pagination (offset + cursor). (feed-test name got expected)
; ---------- offset / limit ----------
(define
O
(feed/stream
(list
(feed/activity "u" "post" "o1" 1 (list))
(feed/activity "u" "post" "o2" 2 (list))
(feed/activity "u" "post" "o3" 3 (list))
(feed/activity "u" "post" "o4" 4 (list))
(feed/activity "u" "post" "o5" 5 (list)))))
(feed-test
"page 1"
(map
(fn (a) (get a :object))
(feed/items (feed/page O 0 2)))
(list "o1" "o2"))
(feed-test
"page 2"
(map
(fn (a) (get a :object))
(feed/items (feed/page O 2 2)))
(list "o3" "o4"))
(feed-test
"page 3 (partial)"
(map
(fn (a) (get a :object))
(feed/items (feed/page O 4 2)))
(list "o5"))
(feed-test
"page past end empty"
(feed/count (feed/page O 10 2))
0)
(feed-test "page-count 5/2 = 3" (feed/page-count O 2) 3)
(feed-test "page-count 5/5 = 1" (feed/page-count O 5) 1)
; ---------- cursor (recency) ----------
(define
R
(feed/stream
(list
(feed/activity "u" "post" "a" 50 (list))
(feed/activity "u" "post" "b" 40 (list))
(feed/activity "u" "post" "c" 30 (list))
(feed/activity "u" "post" "d" 20 (list))
(feed/activity "u" "post" "e" 10 (list)))))
(define p1 (feed/page-before R 100 2))
(feed-test
"cursor page 1 newest first"
(map (fn (a) (get a :object)) (feed/items p1))
(list "a" "b"))
(feed-test "next cursor after page 1" (feed/next-cursor p1) 40)
(define p2 (feed/page-before R (feed/next-cursor p1) 2))
(feed-test
"cursor page 2"
(map (fn (a) (get a :object)) (feed/items p2))
(list "c" "d"))
(feed-test "next cursor after page 2" (feed/next-cursor p2) 20)
(define p3 (feed/page-before R (feed/next-cursor p2) 2))
(feed-test
"cursor page 3 (partial)"
(map (fn (a) (get a :object)) (feed/items p3))
(list "e"))
(feed-test
"empty page nil cursor"
(feed/next-cursor (feed/page-before R 5 2))
nil)
(feed-test
"after cursor loads newer"
(map
(fn (a) (get a :object))
(feed/items (feed/recent (feed/after R 30))))
(list "a" "b"))
(feed-test
"before cursor count"
(feed/count (feed/before R 30))
2)

View File

@@ -1,160 +0,0 @@
; Phase 3 — aggregation + ranking. (feed-test name got expected)
; ---------- aggregation ----------
(define
A
(feed/stream
(list
(feed/activity "alice" "post" "p1" 5 (list))
(feed/activity "alice" "post" "p2" 15 (list))
(feed/activity "bob" "post" "p3" 25 (list))
(feed/activity "alice" "like" "p1" 35 (list)))))
(feed-test "actor-counts" (feed/actor-counts A) {:alice 3 :bob 1})
(feed-test "object-counts" (feed/object-counts A) {:p2 1 :p3 1 :p1 2})
(feed-test
"group-by actor alice len"
(len (get (feed/group-by A feed/actor) "alice"))
3)
(feed-test
"group-count empty"
(feed/group-count feed/empty feed/actor)
{})
; day bucketing
(define
D
(feed/stream
(list
(feed/activity "alice" "post" "p1" 5 (list))
(feed/activity "alice" "post" "p2" 8 (list))
(feed/activity "alice" "post" "p3" 12 (list)))))
(feed-test "feed/day floor" (feed/day 12 10) 1)
(feed-test "feed/day same bucket" (feed/day 8 10) 0)
(feed-test "by-actor-day" (feed/by-actor-day D 10) {:alice#0 2 :alice#1 1})
; ---------- recency ----------
(define rec (feed/recency 100 10))
(feed-test
"recency at=now -> 1"
(rec (feed/activity "x" "post" "o" 100 (list)))
1)
(feed-test
"recency age=hl -> .5"
(rec (feed/activity "x" "post" "o" 90 (list)))
0.5)
(feed-test
"recency age=2hl -> .25"
(rec (feed/activity "x" "post" "o" 80 (list)))
0.25)
; ---------- velocity ----------
(define vel (feed/velocity D 10))
(feed-test
"velocity burst (at=12)"
(vel (feed/activity "alice" "post" "z" 12 (list)))
3)
(feed-test
"velocity mid (at=8)"
(vel (feed/activity "alice" "post" "z" 8 (list)))
2)
(feed-test
"velocity first (at=5)"
(vel (feed/activity "alice" "post" "z" 5 (list)))
1)
(feed-test
"velocity other actor"
(vel (feed/activity "bob" "post" "z" 12 (list)))
0)
; ---------- engagement ----------
(define eng (feed/engagement A))
(feed-test
"engagement p1"
(eng (feed/activity "x" "post" "p1" 0 (list)))
2)
(feed-test
"engagement p2"
(eng (feed/activity "x" "post" "p2" 0 (list)))
1)
; ---------- composite ----------
(define
cmp1
(feed/composite (list (list 2 (fn (a) (get a :at))))))
(feed-test
"composite single part"
(cmp1 (feed/activity "x" "post" "o" 5 (list)))
10)
(define
cmp2
(feed/composite
(list
(list 2 (fn (a) (get a :at)))
(list 3 (fn (a) 1)))))
(feed-test
"composite two parts"
(cmp2 (feed/activity "x" "post" "o" 5 (list)))
13)
; ---------- ranking ----------
(define
R
(feed/stream
(list
(feed/activity "u" "post" "oC" 80 (list))
(feed/activity "u" "post" "oA" 100 (list))
(feed/activity "u" "post" "oB" 90 (list)))))
(feed-test
"rank by recency objects"
(map (fn (a) (get a :object)) (feed/items (feed/rank R rec)))
(list "oA" "oB" "oC"))
(feed-test
"top-2 by recency"
(map (fn (a) (get a :object)) (feed/items (feed/top R rec 2)))
(list "oA" "oB"))
(feed-test "top-2 count" (feed/count (feed/top R rec 2)) 2)
; constant score -> tiebreak by :at descending
(define
T
(feed/stream
(list
(feed/activity "u" "post" "f" 10 (list))
(feed/activity "u" "post" "g" 30 (list))
(feed/activity "u" "post" "h" 20 (list)))))
(feed-test
"tiebreak at-desc"
(map
(fn (a) (get a :object))
(feed/items (feed/rank T (fn (a) 0))))
(list "g" "h" "f"))
; equal score AND equal :at -> stable input order
(define
E
(feed/stream
(list
(feed/activity "u" "post" "first" 50 (list))
(feed/activity "u" "post" "second" 50 (list)))))
(feed-test
"stable equal-key input order"
(map
(fn (a) (get a :object))
(feed/items (feed/rank E (fn (a) 0))))
(list "first" "second"))
(feed-test
"with-scores attaches score"
(get (nth (feed/items (feed/with-scores R rec)) 1) :score)
1)
(feed-test "rank preserves count" (feed/count (feed/rank A rec)) 4)

View File

@@ -1,49 +0,0 @@
; Follow-up — conversation threading via :reply-to closure. (feed-test name got expected)
(define
S
(feed/stream
(list
(feed/normalize {:actor "a" :object "root" :at 1})
(feed/normalize {:actor "b" :object "r1" :at 2 :verb "reply" :reply-to "root"})
(feed/normalize {:actor "c" :object "r2" :at 3 :verb "reply" :reply-to "root"})
(feed/normalize {:actor "d" :object "r3" :at 4 :verb "reply" :reply-to "r1"})
(feed/normalize {:actor "e" :object "x" :at 5}))))
; ---------- direct replies ----------
(feed-test "direct replies to root" (feed/reply-count S "root") 2)
(feed-test "direct replies to r1" (feed/reply-count S "r1") 1)
(feed-test "no replies to r3" (feed/reply-count S "r3") 0)
(feed-test
"replies objects to root"
(map (fn (a) (get a :object)) (feed/items (feed/replies S "root")))
(list "r1" "r2"))
; ---------- thread closure ----------
(feed-test
"thread objects root (transitive)"
(feed/thread-objects S "root")
(list "root" "r1" "r2" "r3"))
(feed-test
"thread root chronological"
(map (fn (a) (get a :object)) (feed/items (feed/thread S "root")))
(list "root" "r1" "r2" "r3"))
(feed-test "thread size root" (feed/thread-size S "root") 4)
(feed-test
"thread excludes unrelated x"
(feed/-elem?
"x"
(map (fn (a) (get a :object)) (feed/items (feed/thread S "root"))))
false)
; ---------- sub-thread ----------
(feed-test
"thread from r1 (sub-tree)"
(map (fn (a) (get a :object)) (feed/items (feed/thread S "r1")))
(list "r1" "r3"))
(feed-test "thread size r1" (feed/thread-size S "r1") 2)
(feed-test "leaf thread is itself" (feed/thread-size S "r3") 1)
(feed-test "unrelated thread is itself" (feed/thread-size S "x") 1)

View File

@@ -1,82 +0,0 @@
; Follow-up — trending objects/actors by recent activity. (feed-test name got expected)
; window (50,100]: X@60,X@70 (a), Y@80 (b), Z@90 (c); W@40 is too old
(define
S
(feed/stream
(list
(feed/activity "a" "post" "X" 60 (list))
(feed/activity "a" "post" "X" 70 (list))
(feed/activity "b" "post" "Y" 80 (list))
(feed/activity "c" "post" "Z" 90 (list))
(feed/activity "d" "post" "W" 40 (list)))))
; ---------- trending objects ----------
(feed-test
"trending count (3 in window)"
(len (feed/trending S 100 50 10))
3)
(feed-test
"trending top object"
(get
(nth (feed/trending S 100 50 10) 0)
:object)
"X")
(feed-test
"trending top count"
(get
(nth (feed/trending S 100 50 10) 0)
:count)
2)
(feed-test
"trending order (count desc, key asc tiebreak)"
(map
(fn (e) (get e :object))
(feed/trending S 100 50 10))
(list "X" "Y" "Z"))
(feed-test
"trending top-2"
(map
(fn (e) (get e :object))
(feed/trending S 100 50 2))
(list "X" "Y"))
(feed-test
"old object W excluded"
(feed/-elem?
"W"
(map
(fn (e) (get e :object))
(feed/trending S 100 50 10)))
false)
(feed-test
"narrow window keeps only newest"
(map
(fn (e) (get e :object))
(feed/trending S 100 15 10))
(list "Z"))
(feed-test
"empty window -> nothing"
(feed/trending S 100 5 10)
(list))
; ---------- trending actors ----------
(feed-test
"trending actor top"
(get
(nth (feed/trending-actors S 100 50 10) 0)
:actor)
"a")
(feed-test
"trending actor count"
(get
(nth (feed/trending-actors S 100 50 10) 0)
:count)
2)
(feed-test
"trending actors order"
(map
(fn (e) (get e :actor))
(feed/trending-actors S 100 50 10))
(list "a" "b" "c"))

View File

@@ -1,59 +0,0 @@
; feed/thread — conversation threading. A reply carries :reply-to <parent-object>
; (normalize preserves it). A thread is the transitive closure over :reply-to from
; a root object: root + replies + replies-to-replies, gathered chronologically.
;
; Requires: lib/feed/normalize.sx, lib/feed/stream.sx, lib/feed/fanout.sx
; (feed/-elem?, feed/-distinct).
; direct replies to an object
(define
feed/replies
(fn
(stream object)
(feed/filter stream (fn (a) (equal? (get a :reply-to) object)))))
(define
feed/reply-count
(fn (stream object) (feed/count (feed/replies stream object))))
; iterate f from x until the result stops growing (set-closure fixpoint)
(define
feed/-fixpoint
(fn
(f x)
(let
((nx (f x)))
(if (= (len nx) (len x)) x (feed/-fixpoint f nx)))))
; the set of object-ids in the thread rooted at `root`
(define
feed/thread-objects
(fn
(stream root)
(let
((all (feed/items stream)))
(feed/-fixpoint
(fn
(acc)
(feed/-distinct
(append
acc
(map
(fn (a) (get a :object))
(filter (fn (a) (feed/-elem? (get a :reply-to) acc)) all)))))
(list root)))))
; the full thread as a chronological stream (root + all descendants)
(define
feed/thread
(fn
(stream root)
(let
((objs (feed/thread-objects stream root)))
(feed/sort-by-at
(feed/filter stream (fn (a) (feed/-elem? (get a :object) objs)))))))
; how many activities are in the thread (root counts as 1)
(define
feed/thread-size
(fn (stream root) (feed/count (feed/thread stream root))))

View File

@@ -1,42 +0,0 @@
; feed/trending — what's hot right now: objects (or actors) ranked by activity
; count within a recency window. Deterministic: count descending, ties broken by
; key ascending (entries are pre-sorted by key, then stable grade-down by count).
;
; Requires: lib/feed/stream.sx, lib/feed/aggregate.sx (object/actor-counts),
; lib/feed/rank.sx (feed/-desc-by).
; activities within (now-window, now]
(define
feed/-recent
(fn
(stream now window)
(feed/filter
stream
(fn (a) (and (<= (get a :at) now) (> (get a :at) (- now window)))))))
; counts dict -> top-N entries {label key, :count n}, count desc, key asc
(define
feed/-top-counts
(fn
(counts label n)
(let
((entries (map (fn (k) (assoc {:count (get counts k)} label k)) (sort (keys counts)))))
(take (feed/-desc-by entries (fn (e) (get e :count))) n))))
; top-N trending objects in the window
(define
feed/trending
(fn
(stream now window n)
(feed/-top-counts
(feed/object-counts (feed/-recent stream now window))
:object n)))
; top-N most active actors in the window
(define
feed/trending-actors
(fn
(stream now window n)
(feed/-top-counts
(feed/actor-counts (feed/-recent stream now window))
:actor n)))

141
lib/flow/README.md Normal file
View File

@@ -0,0 +1,141 @@
# flow — durable DAG workflows on Scheme
`flow` is a workflow engine for rose-ash: content pipelines (write → review →
publish → federate), scheduled jobs, and multi-step user flows (signup, confirm,
onboard) that **survive process restarts**. It is a thin Scheme prelude over the
Scheme-on-SX guest (`lib/scheme/`); a flow runs *inside* the interpreter.
Run the suite: `bash lib/flow/conformance.sh`**151/151 across 10 suites**.
## Model
A **flow** is just a Scheme procedure of one argument — the upstream value:
```
node : input -> output
```
Combinators build composite nodes out of child nodes. A node that ignores its
argument is effectively a thunk. There is no separate "graph" object: composition
*is* function composition, so flows are values you can name, pass, and nest.
```scheme
(defflow publish
(sequence
(lambda (draft) (string-append draft "!"))
(branch (lambda (post) (>= (string-length post) 3))
(remote-node 'fed 'publish)
(flow-const 'rejected))))
(flow/start publish "hello") ; => federated, or a (flow-suspended id tag) state
```
## Building blocks (`spec.sx`)
| Combinator | Meaning |
|---|---|
| `(flow-node f)` / `(flow-id x)` / `(flow-const v)` | leaf nodes |
| `(sequence n ...)` | thread input left-to-right |
| `(parallel n ...)` | fan input to every child, join results into a list (sequential eval) |
| `(map-flow node)` | run `node` over each item of a list input, join results |
| `(flow-while pred body max)` / `(flow-until ...)` | bounded iteration (cap `max` steps) |
| `(defflow name body)` | bind + register a named flow (so it survives restart) |
## Control flow + errors (`spec.sx`)
| Combinator | Meaning |
|---|---|
| `(branch pred then else)` | `pred` on input selects `then`/`else` (`cond` is a Scheme special form) |
| `(retry n node)` | re-run on a *raised exception*, up to `n` attempts |
| `(timeout budget node)` | cooperative **step budget**: nodes call `(tick)`; the `(budget+1)`-th tick raises `flow-timeout` |
| `(try-catch node handler)` | catch a raised exception → `(handler error)` |
| `(fail reason)` / `(failed? x)` / `(fail-reason x)` | explicit failure *values* (flow downstream as data) |
| `(recover node handler)` | the fail-VALUE counterpart of try-catch |
| `(attempt n ...)` | railway sequence: stop at the first node returning a `(fail ...)` |
| `(tap effect)` | run a side effect, return input unchanged |
**Two error channels, on purpose.** Raised exceptions are for *bugs/transients*
(caught by `retry`/`try-catch`). `(fail reason)` values are for *expected business
outcomes* (validation rejected, declined) and compose via `attempt`/`recover`.
## Suspend / resume — the durable core (`spec.sx`, `store.sx`)
The guest Scheme's `call/cc` is **escape-only** — re-invoking a captured
continuation after it returns *hangs* the runtime. So flow does **not** serialize
continuations. Instead it uses **deterministic replay**:
- `(suspend tag)` — if `tag` is already in the replay log, return its logged value;
otherwise escape to the driver as `(flow-suspended tag)`.
- `resume` appends `(tag value)` to the log and **re-runs the flow from the start**.
Already-resolved suspends replay their values; the first unresolved one escapes
again (or the flow completes).
The entire persisted state is the replay log — plain data. No live continuation is
ever stored, so flows survive process restarts and even moves between instances.
> **Author contract:** suspend `tag`s must be unique and deterministic across
> replays, and **all** non-determinism / side effects must go through suspend
> points (so their results are logged) — otherwise they re-run on every replay.
### Lifecycle (`store.sx`)
```scheme
(flow/start flow input) ; raw result if it completes, else (flow-suspended id tag)
(flow/resume id value) ; inject value at the waiting tag, continue
(flow/cancel id) ; terminate; a later resume is rejected
```
### Introspection & hygiene
```scheme
(flow/status id) ; done | suspended | cancelled | unknown
(flow/result id) ; result if done, else (flow-error reason)
(flow/list) ; ((id status) ...)
(flow/pending) ; ((id waiting-tag) ...) — what each suspended flow awaits
(flow/gc) ; drop terminal records, keep live ones; returns count removed
(flow/forget id) ; drop one terminal record (refuses live flows)
```
### Crash recovery
```scheme
(flow-store-export) ; the store as plain data (live procs nulled)
(flow-store-import! d) ; restore the store from exported data
(flow-resumable-ids) ; ids of suspended flows to wake on restart
```
On restart the flow definitions are reloaded (`defflow` re-registers names) and the
exported store reimported; `resume` re-resolves each flow's procedure **by name**.
## Distribution via fed-sx (`remote.sx`)
```scheme
(flow-peer-register! addr table) ; mock a peer's exposed functions (fed-sx boundary)
(remote-node addr fn) ; run a node on a peer
(remote-failover addrs fn local) ; try peers in order, fall through to a local node
(flow-replicate-to addr) ; copy this store to a peer's replica slot
(flow-restore-from addr) ; import a peer's replica (handoff)
```
**Handoff** is crash recovery across instances: replicate → local instance dies →
peer restores the (plain-data) store and resumes. The replay log carries over, so
all resolved suspends survive the move.
## Files
| File | Contents |
|---|---|
| `spec.sx` | combinators (flow-combinators-src / flow-control-src / flow-suspend-src) |
| `store.sx` | durable store, lifecycle, crash recovery, introspection, hygiene |
| `remote.sx` | fed-sx transport (mock peer registry), failover, replication |
| `api.sx` | `flow-make-env` / `flow-run` SX helpers (one cached env, per-test reset) |
| `tests/*.sx` | 10 suites, 151 cases |
| `conformance.sh` | loads substrate + flow layer, runs every suite |
## Notes on the substrate
The guest Scheme (`lib/scheme/`, imported read-only) lacks dotted-rest params
`(a . rest)` and named `let`; combinators use `(lambda args ...)` variadics + top-
level recursion. `cons` is list-only (no dotted pairs), so log/assoc entries are
2-element lists. Strings box as `{:scm-string "..."}`. Timeout is a step budget
because there is no wall clock; `parallel` is sequential for the same reason.

65
lib/flow/api.sx Normal file
View File

@@ -0,0 +1,65 @@
;; lib/flow/api.sx — flow runtime entry points.
;;
;; Builds a Scheme env preloaded with the flow combinators (lib/flow/spec.sx),
;; the durable store + lifecycle (lib/flow/store.sx), the fed-sx remote layer
;; (lib/flow/remote.sx), and the host integration ABI (lib/flow/host.sx), and
;; provides SX helpers to run flow programs.
;;
;; Scheme-level API (available inside flow programs):
;; (flow/start flow input) — run a flow; raw result if it completes, else
;; (flow-suspended id tag). Defined in store.sx.
;; (flow/resume id value) — resume a suspended flow (store.sx)
;; (flow/cancel id) — cancel a flow (store.sx)
;; (suspend tag) — suspension point (spec.sx)
;; (request kind payload) — host request envelope over suspend (host.sx)
;; (remote-node addr fn) — node executed on a federation peer (remote.sx)
;;
;; SX-level helpers (for hosts and tests):
;; (flow-make-env) — fresh standard env + combinators + store + remote + host
;; (flow-run src) — eval a Scheme program string in a reset shared env
;; (flow-run-in env src) — eval a Scheme program string in a given env
;;
;; flow-run reuses ONE env (building the full standard env is expensive) and
;; resets the mutable flow globals before each program, so tests stay isolated
;; without paying for a fresh standard env each time. flow-registry persists (it
;; models reloaded flow definitions surviving a restart).
(define
flow-make-env
(fn
()
(let
((env (scheme-standard-env)))
(flow-load-combinators! env)
(flow-load-store! env)
(flow-load-remote! env)
(flow-load-host! env)
env)))
(define
flow-run-in
(fn (env src) (scheme-eval-program (scheme-parse-all src) env)))
(define
flow-reset-src
"(set! flow-store (list)) (set! flow-next-id 0) (set! flow-replay-log (list)) (set! flow-suspend-k #f) (set! flow-timeout-budget -1) (set! flow-peers (list)) (set! flow-replicas (list))")
(define flow-env-cache false)
(define
flow-shared-env
(fn
()
(begin
(if flow-env-cache nil (set! flow-env-cache (flow-make-env)))
flow-env-cache)))
(define
flow-run
(fn
(src)
(let
((env (flow-shared-env)))
(begin
(scheme-eval-program (scheme-parse-all flow-reset-src) env)
(scheme-eval-program (scheme-parse-all src) env)))))

103
lib/flow/conformance.sh Executable file
View File

@@ -0,0 +1,103 @@
#!/usr/bin/env bash
# flow-on-sx conformance runner — runs all flow test suites in one sx_server process.
#
# Usage:
# bash lib/flow/conformance.sh # run all suites
# bash lib/flow/conformance.sh -v # verbose (list each suite)
set -uo pipefail
cd "$(git rev-parse --show-toplevel)"
SX_SERVER="${SX_SERVER:-hosts/ocaml/_build/default/bin/sx_server.exe}"
if [ ! -x "$SX_SERVER" ]; then
SX_SERVER="/root/rose-ash/hosts/ocaml/_build/default/bin/sx_server.exe"
fi
if [ ! -x "$SX_SERVER" ]; then
echo "ERROR: sx_server.exe not found." >&2
exit 1
fi
VERBOSE="${1:-}"
# Suites: NAME RUNNER-FN PATH
SUITES=(
"basic flow-basic-tests-run! lib/flow/tests/basic.sx"
"control flow-ctl-tests-run! lib/flow/tests/control.sx"
"suspend flow-sus-tests-run! lib/flow/tests/suspend.sx"
"recovery flow-rec-tests-run! lib/flow/tests/recovery.sx"
"distributed flow-dist-tests-run! lib/flow/tests/distributed.sx"
"api flow-api-tests-run! lib/flow/tests/api.sx"
"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"
"host flow-hst-tests-run! lib/flow/tests/host.sx"
)
TMPFILE=$(mktemp); trap "rm -f $TMPFILE" EXIT
EPOCH=1
emit_load () { echo "(epoch $EPOCH)"; echo "(load \"$1\")"; EPOCH=$((EPOCH+1)); }
emit_eval () { echo "(epoch $EPOCH)"; echo "(eval \"$1\")"; EPOCH=$((EPOCH+1)); }
{
emit_load "lib/guest/lex.sx"
emit_load "lib/guest/reflective/env.sx"
emit_load "lib/guest/reflective/quoting.sx"
emit_load "lib/scheme/parser.sx"
emit_load "lib/scheme/eval.sx"
emit_load "lib/scheme/runtime.sx"
emit_load "lib/flow/spec.sx"
emit_load "lib/flow/store.sx"
emit_load "lib/flow/remote.sx"
emit_load "lib/flow/host.sx"
emit_load "lib/flow/api.sx"
for SUITE in "${SUITES[@]}"; do
read -r _NAME _RUNNER FILE <<< "$SUITE"
emit_load "$FILE"
emit_eval "($_RUNNER)"
done
} > "$TMPFILE"
OUTPUT=$(timeout 540 "$SX_SERVER" < "$TMPFILE" 2>&1 || true)
TOTAL_PASS=0
TOTAL_FAIL=0
FAILED_SUITES=()
LAST_DICT_LINES=$(echo "$OUTPUT" | grep -E '^\{:' || true)
I=0
while read -r LINE; do
[ -z "$LINE" ] && continue
P=$(echo "$LINE" | grep -oE ':passed [0-9]+' | awk '{print $2}')
F=$(echo "$LINE" | grep -oE ':failed [0-9]+' | awk '{print $2}')
[ -z "$P" ] && P=0
[ -z "$F" ] && F=0
SUITE_INFO="${SUITES[$I]}"
SUITE_NAME=$(echo "$SUITE_INFO" | awk '{print $1}')
TOTAL_PASS=$((TOTAL_PASS + P))
TOTAL_FAIL=$((TOTAL_FAIL + F))
if [ "$F" -gt 0 ]; then
FAILED_SUITES+=("$SUITE_NAME: $P/$((P+F))")
printf 'X %-12s %d/%d\n' "$SUITE_NAME" "$P" "$((P+F))"
echo "$LINE" | grep -oE ':name "[^"]*"' | sed 's/:name / fail: /'
elif [ "$VERBOSE" = "-v" ]; then
printf 'ok %-12s %d passed\n' "$SUITE_NAME" "$P"
fi
I=$((I+1))
done <<< "$LAST_DICT_LINES"
TOTAL=$((TOTAL_PASS + TOTAL_FAIL))
if [ "$TOTAL" -eq 0 ]; then
echo "ERROR: no suite results parsed. Raw output:" >&2
echo "$OUTPUT" >&2
exit 1
fi
if [ $TOTAL_FAIL -eq 0 ]; then
echo "ok $TOTAL_PASS/$TOTAL flow-on-sx tests passed (${#SUITES[@]} suites)"
else
echo "FAIL $TOTAL_PASS/$TOTAL passed, $TOTAL_FAIL failed:"
for S in "${FAILED_SUITES[@]}"; do echo " $S"; done
exit 1
fi

42
lib/flow/host.sx Normal file
View File

@@ -0,0 +1,42 @@
;; lib/flow/host.sx — the host integration ABI (Phase 8).
;;
;; `suspend` is flow's seam to the outside world, but a bare (suspend tag) is just a
;; signal — every author would invent their own tag shape. This layer defines a
;; stable request/response contract so a host (e.g. an art-dag driver, or a human
;; review UI) can hook in WITHOUT reverse-engineering ad-hoc tags.
;;
;; A flow asks the host to do something and waits for the answer:
;; (request kind payload) — suspend with a typed envelope (flow-request kind
;; payload); evaluates to the host's resume value.
;; (await-human prompt) — request kind=human (a decision point)
;; (await-render recipe) — request kind=render (e.g. an art-dag job)
;; (await-effect kind p) — request of an arbitrary kind
;;
;; The host drives flows by polling its work queue and resuming:
;; (flow-host-requests) — ((id kind payload) ...) for every SUSPENDED flow whose
;; waiting tag is a host request. The host dispatches by kind (render -> submit a
;; Celery job; human -> show UI), then calls (flow/resume id answer).
;; (request? tag) / (request-kind tag) / (request-payload tag) — parse one tag.
;;
;; Reference driver — the host only supplies `dispatch`, a (kind payload) -> answer:
;; (flow-drive-host dispatch) — one tick: service every CURRENTLY pending
;; request (snapshot), resuming each with (dispatch kind payload); returns the
;; count serviced. Resumes may create new requests — serviced on the next tick.
;; (flow-run-host dispatch maxticks) — tick until quiescent (no pending requests)
;; or maxticks reached; returns total requests serviced. Bounded for determinism.
;;
;; Contract: the host owns IO and persistence. flow stays deterministic — a flow
;; never performs IO itself, it only `request`s; the host performs the effect and
;; feeds the result back via resume (which the replay log records, so the effect is
;; not re-run on recovery). Persist with flow-store-export after each transition and
;; flow-store-import! on boot.
(define
flow-host-src
"(define (request kind payload) (suspend (list (quote flow-request) kind payload)))\n (define (request? tag) (and (pair? tag) (eq? (car tag) (quote flow-request))))\n (define (request-kind tag) (car (cdr tag)))\n (define (request-payload tag) (car (cdr (cdr tag))))\n (define (await-human prompt) (request (quote human) prompt))\n (define (await-render recipe) (request (quote render) recipe))\n (define (await-effect kind payload) (request kind payload))\n (define (flow-host-req-step pend)\n (if (null? pend)\n (list)\n (let ((id (car (car pend))) (tag (car (cdr (car pend)))))\n (if (request? tag)\n (cons (list id (request-kind tag) (request-payload tag))\n (flow-host-req-step (cdr pend)))\n (flow-host-req-step (cdr pend))))))\n (define (flow-host-requests) (flow-host-req-step (flow/pending)))\n (define (flow-drive-host-step reqs dispatch)\n (if (null? reqs)\n 0\n (begin\n (flow/resume (car (car reqs)) (dispatch (car (cdr (car reqs))) (car (cdr (cdr (car reqs))))))\n (+ 1 (flow-drive-host-step (cdr reqs) dispatch)))))\n (define (flow-drive-host dispatch) (flow-drive-host-step (flow-host-requests) dispatch))\n (define (flow-run-host dispatch maxticks)\n (if (<= maxticks 0)\n 0\n (let ((n (flow-drive-host dispatch)))\n (if (= n 0) 0 (+ n (flow-run-host dispatch (- maxticks 1)))))))")
(define
flow-load-host!
(fn
(env)
(begin (scheme-eval-program (scheme-parse-all flow-host-src) env) env)))

34
lib/flow/remote.sx Normal file
View File

@@ -0,0 +1,34 @@
;; lib/flow/remote.sx — distributed nodes via fed-sx (Phase 4).
;;
;; A node can execute on a federation peer. The transport is the fed-sx boundary;
;; it is MOCKED in tests by a peer registry mapping addr -> function table. In
;; production flow-transport would issue a fed-sx call; here it dispatches locally.
;;
;; (flow-peer-register! addr table) — register a mock peer. table is a list of
;; (fn-name proc) entries — the functions that peer exposes.
;; (flow-transport addr fn input) — invoke fn on the peer with input. Raises
;; (flow-remote-unreachable) if the addr is unknown, (flow-remote-no-fn) if the
;; peer does not expose fn.
;; (remote-node addr fn) — a node that runs fn on the peer at addr.
;; (remote-failover addrs fn local) — try fn on each peer in addrs in order; on a
;; raised error move to the next peer; if every peer fails, run the `local`
;; node as a fallback.
;;
;; Persistence across instances + handoff. Each instance runs the same flow
;; definitions, so the only thing that needs to cross the wire is the (plain-data)
;; store — exactly flow-store-export from store.sx. Replication pushes that export
;; to a peer's replica slot; handoff = restore the replica on the peer and resume.
;;
;; (flow-replicate-to addr) — copy this instance's store to peer addr's replica
;; (flow-restore-from addr) — import the replica from peer addr (#t / #f)
;; (flow-replica-get addr) — the raw replicated store at addr (or #f)
(define
flow-remote-src
"(define flow-peers (list))\n (define (flow-assoc key alist)\n (if (null? alist)\n #f\n (if (eq? (car (car alist)) key) (car (cdr (car alist))) (flow-assoc key (cdr alist)))))\n (define (flow-peer-register! addr table) (set! flow-peers (cons (list addr table) flow-peers)))\n (define (flow-transport addr fn input)\n (let ((table (flow-assoc addr flow-peers)))\n (if table\n (let ((proc (flow-assoc fn table)))\n (if proc (proc input) (raise (quote flow-remote-no-fn))))\n (raise (quote flow-remote-unreachable)))))\n (define (remote-node addr fn) (lambda (input) (flow-transport addr fn input)))\n (define (flow-failover-step addrs fn input local)\n (if (null? addrs)\n (local input)\n (guard (e (#t (flow-failover-step (cdr addrs) fn input local)))\n (flow-transport (car addrs) fn input))))\n (define (remote-failover addrs fn local)\n (lambda (input) (flow-failover-step addrs fn input local)))\n\n (define flow-replicas (list))\n (define (flow-replicas-remove addr reps)\n (if (null? reps)\n (list)\n (if (eq? (car (car reps)) addr)\n (flow-replicas-remove addr (cdr reps))\n (cons (car reps) (flow-replicas-remove addr (cdr reps))))))\n (define (flow-replicate-to addr)\n (set! flow-replicas (cons (list addr (flow-store-export)) (flow-replicas-remove addr flow-replicas))))\n (define (flow-replica-get addr) (flow-assoc addr flow-replicas))\n (define (flow-restore-from addr)\n (let ((data (flow-replica-get addr)))\n (if data (begin (flow-store-import! data) #t) #f)))")
(define
flow-load-remote!
(fn
(env)
(begin (scheme-eval-program (scheme-parse-all flow-remote-src) env) env)))

19
lib/flow/scoreboard.json Normal file
View File

@@ -0,0 +1,19 @@
{
"total": 166,
"passed": 166,
"failed": 0,
"suites": {
"basic": { "passed": 18, "total": 18 },
"control": { "passed": 31, "total": 31 },
"suspend": { "passed": 17, "total": 17 },
"recovery": { "passed": 8, "total": 8 },
"distributed": { "passed": 19, "total": 19 },
"api": { "passed": 12, "total": 12 },
"combinators": { "passed": 17, "total": 17 },
"railway": { "passed": 10, "total": 10 },
"integration": { "passed": 10, "total": 10 },
"hygiene": { "passed": 9, "total": 9 },
"host": { "passed": 15, "total": 15 }
},
"phases": { "phase1": "done", "phase2": "done", "phase3": "done", "phase4": "done", "phase5": "done", "phase6": "done", "phase7": "done", "phase8": "done" }
}

53
lib/flow/scoreboard.md Normal file
View File

@@ -0,0 +1,53 @@
# flow-on-sx Scoreboard
**All tests pass: 166 / 166 across 11 suites. Phases 1-8 complete.**
`bash lib/flow/conformance.sh`
## Per-suite breakdown
| Suite | Passing | Covers |
|-------|--------:|--------|
| basic | 18 | Phase 1: single nodes, linear sequence, data-flow threading, defflow, parallel fan/join, nested composition, publish-shaped flow |
| control | 31 | Phase 2: `branch` (6); error model `fail`/`failed?`/`fail-reason` (6); `try-catch` (6); `retry n` (6); `timeout` cooperative step budget (7) |
| suspend | 17 | Phase 3: suspend/resume/cancel via deterministic replay; multi-step, replay determinism, lifecycle guards, suspend-in-branch |
| recovery | 8 | Phase 3: crash recovery — store export/import, resumable scan, restart-at-every-step, replay-log survival |
| distributed | 19 | Phase 4: `remote-node` (7); `remote-failover` (6); replication + handoff across instances (6) |
| api | 12 | Phase 5: introspection — `flow/status`, `flow/result`, `flow/list`, `flow/pending` |
| 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) |
| host | 15 | Phase 8: host ABI — `request`/`await-human`/`await-render`, `flow-host-requests` queue, `flow-run-host` reference driver; art-dag-shaped render→review→publish loop |
## Architecture
Flow combinators are a **Scheme prelude** (`lib/flow/spec.sx`) loaded onto
`scheme-standard-env`. A flow is a Scheme procedure `input -> output`. The whole
flow executes inside the Scheme interpreter, so Phase 3's `suspend` (call/cc) will
capture the flow continuation directly.
- `lib/flow/spec.sx` — combinators: `flow-node`, `flow-id`, `flow-const`,
`sequence`, `parallel`, `defflow`; `flow-load-combinators!`.
- `lib/flow/api.sx``flow/start` (Scheme); `flow-make-env`, `flow-run`,
`flow-run-in` (SX helpers).
- `lib/flow/tests/basic.sx` — 18 cases.
- `lib/flow/conformance.sh` — loads substrate + flow layer, runs suites.
## Semantics notes
- **node** = 1-arg Scheme procedure; the upstream value is the argument. A node
ignoring its argument is effectively a thunk.
- **sequence** threads left-to-right; empty sequence = identity.
- **parallel** fans the same input to every branch and joins results into a list.
Evaluation is **sequential** for now; true concurrency arrives in Phase 3.
## Phases
- [x] Phase 1 — Declarative DAG + sequential execution (combinators + 18 tests, `flow/start`)
- [x] Phase 2 — Control flow + error handling (branch, error model, try-catch, retry, timeout)
- [x] Phase 3 — Suspend/resume (suspend/resume/cancel + crash recovery via deterministic replay)
- [x] Phase 4 — Distributed nodes via fed-sx (remote-node, failover, replication + handoff)
- [x] Phase 5 — Operational API + combinators (introspection, tap, recover, map-flow)
- [ ] Phase 3 — Suspend / resume (the showcase)
- [ ] Phase 4 — Distributed nodes via fed-sx

61
lib/flow/spec.sx Normal file
View File

@@ -0,0 +1,61 @@
;; lib/flow/spec.sx — flow combinators as a Scheme prelude.
;;
;; A flow is a Scheme procedure of one argument: the upstream value.
;; node : input -> output
;; A leaf node ignoring its argument is effectively a thunk. Combinators
;; build composite nodes out of child nodes. The whole flow runs INSIDE the
;; Scheme interpreter.
;;
;; Phase 1 combinators (flow-combinators-src):
;; flow-node / flow-id / flow-const / sequence / parallel / defflow
;; defflow both binds the flow and registers it by name (flow-register!, in
;; store.sx) so it can be re-resolved after a process restart.
;; map-flow (Phase 5): run a node over each item of a list input, join results.
;; flow-while / flow-until (Phase 5): bounded iteration — re-run body, threading
;; the value, while/until pred holds, up to `max` steps (deterministic bound; no
;; unbounded loops in pure SX).
;;
;; Phase 2 combinators (flow-control-src):
;; branch / fail / failed? / fail-reason / try-catch / retry / timeout / tick
;; tap (Phase 5): side-effecting pass-through (returns input unchanged).
;; recover (Phase 5): the fail-VALUE counterpart of try-catch.
;; attempt (Phase 6): railway sequence — thread nodes left-to-right but stop at
;; the first node that returns a (fail ...) value, returning that failure.
;;
;; Phase 3 suspend core (flow-suspend-src):
;; The guest Scheme's call/cc is ESCAPE-ONLY (re-invoking a captured k after it
;; returns hangs the runtime), so suspend/resume CANNOT re-enter a continuation.
;; Instead, durability uses DETERMINISTIC REPLAY: a flow re-runs from the start
;; on each resume; suspend points that have already been resolved replay their
;; logged value, and the first unresolved suspend escapes back to the driver.
;; The entire persisted state is the replay log (plain (tag value) data), which
;; survives process restart — no live continuation is ever serialized.
;;
;; (suspend tag) — if tag is in the replay log, return its value; else escape
;; to the driver as (flow-suspended tag). tags must be unique & deterministic
;; across replays. ALL effects/non-determinism must go through suspend so their
;; results are logged (otherwise they re-run on every replay).
;; (flow-drive flow input log) — run flow with the given replay log; returns
;; (flow-done result) or (flow-suspended tag).
(define
flow-combinators-src
"(define (flow-node f) f)\n (define (flow-id input) input)\n (define (flow-const v) (lambda (input) v))\n (define (flow-seq-step ns v)\n (if (null? ns) v (flow-seq-step (cdr ns) ((car ns) v))))\n (define sequence (lambda ns (lambda (input) (flow-seq-step ns input))))\n (define parallel (lambda ns (lambda (input) (map (lambda (n) (n input)) ns))))\n (define (map-flow node) (lambda (items) (map node items)))\n (define (flow-while-step pred body input n)\n (if (<= n 0)\n input\n (if (pred input) (flow-while-step pred body (body input) (- n 1)) input)))\n (define (flow-while pred body max) (lambda (input) (flow-while-step pred body input max)))\n (define (flow-until-step pred body input n)\n (if (<= n 0)\n input\n (if (pred input) input (flow-until-step pred body (body input) (- n 1)))))\n (define (flow-until pred body max) (lambda (input) (flow-until-step pred body input max)))\n (define-syntax defflow\n (syntax-rules ()\n ((defflow nm body)\n (begin (define nm body) (flow-register! (quote nm) nm)))))")
(define
flow-control-src
"(define (branch pred then else)\n (lambda (input) (if (pred input) (then input) (else input))))\n (define (fail reason) (list (quote flow-fail) reason))\n (define (failed? x) (and (pair? x) (eq? (car x) (quote flow-fail))))\n (define (fail-reason x) (car (cdr x)))\n (define (recover node handler)\n (lambda (input)\n (let ((r (node input)))\n (if (failed? r) (handler (fail-reason r)) r))))\n (define (tap effect)\n (lambda (input) (begin (effect input) input)))\n (define (flow-attempt-step ns v)\n (if (failed? v)\n v\n (if (null? ns) v (flow-attempt-step (cdr ns) ((car ns) v)))))\n (define attempt (lambda ns (lambda (input) (flow-attempt-step ns input))))\n (define (try-catch node handler)\n (lambda (input) (guard (e (#t (handler e))) (node input))))\n (define (flow-retry-step n node input)\n (guard (e (#t (if (<= n 1) (raise e) (flow-retry-step (- n 1) node input))))\n (node input)))\n (define (retry n node) (lambda (input) (flow-retry-step n node input)))\n (define flow-timeout-budget -1)\n (define (tick)\n (if (< flow-timeout-budget 0)\n 0\n (begin\n (set! flow-timeout-budget (- flow-timeout-budget 1))\n (if (< flow-timeout-budget 0)\n (raise (quote flow-timeout))\n flow-timeout-budget))))\n (define (timeout budget node)\n (lambda (input)\n (let ((saved flow-timeout-budget))\n (set! flow-timeout-budget budget)\n (guard (e (#t (begin (set! flow-timeout-budget saved) (raise e))))\n (let ((result (node input)))\n (set! flow-timeout-budget saved)\n result)))))")
(define
flow-suspend-src
"(define flow-replay-log (list))\n (define flow-suspend-k #f)\n (define (flow-log-lookup tag log)\n (if (null? log)\n (list #f #f)\n (if (eq? (car (car log)) tag)\n (list #t (car (cdr (car log))))\n (flow-log-lookup tag (cdr log)))))\n (define (suspend tag)\n (let ((hit (flow-log-lookup tag flow-replay-log)))\n (if (car hit)\n (car (cdr hit))\n (flow-suspend-k (list (quote flow-suspended) tag)))))\n (define (flow-drive flow input log)\n (set! flow-replay-log log)\n (call/cc\n (lambda (k)\n (set! flow-suspend-k k)\n (list (quote flow-done) (flow input)))))")
(define
flow-load-combinators!
(fn
(env)
(begin
(scheme-eval-program (scheme-parse-all flow-combinators-src) env)
(scheme-eval-program (scheme-parse-all flow-control-src) env)
(scheme-eval-program (scheme-parse-all flow-suspend-src) env)
env)))

45
lib/flow/store.sx Normal file

File diff suppressed because one or more lines are too long

79
lib/flow/tests/api.sx Normal file
View File

@@ -0,0 +1,79 @@
;; lib/flow/tests/api.sx — Phase 5: operational introspection API.
(define flow-api-pass 0)
(define flow-api-fail 0)
(define flow-api-fails (list))
(define
flow-api-test
(fn
(name actual expected)
(if
(= actual expected)
(set! flow-api-pass (+ flow-api-pass 1))
(begin
(set! flow-api-fail (+ flow-api-fail 1))
(append! flow-api-fails {:name name :expected expected :actual actual})))))
(define flow-a (fn (src) (flow-run src)))
;; ── flow/status ─────────────────────────────────────────────────
(flow-api-test "status: unknown id" (flow-a "(flow/status 999)") "unknown")
(flow-api-test
"status: suspended flow"
(flow-a
"(defflow w (lambda (x) (suspend (quote q)))) (define id (car (cdr (flow/start w 0)))) (flow/status id)")
"suspended")
(flow-api-test
"status: completed flow"
(flow-a
"(defflow w (sequence (lambda (x) (suspend (quote q))) (lambda (v) v))) (define id (car (cdr (flow/start w 0)))) (flow/resume id 5) (flow/status id)")
"done")
(flow-api-test
"status: cancelled flow"
(flow-a
"(defflow w (lambda (x) (suspend (quote q)))) (define id (car (cdr (flow/start w 0)))) (flow/cancel id) (flow/status id)")
"cancelled")
;; ── flow/result ─────────────────────────────────────────────────
(flow-api-test
"result: returns the value of a completed flow"
(flow-a
"(defflow w (sequence (lambda (x) (suspend (quote q))) (lambda (v) (list (quote got) v)))) (define id (car (cdr (flow/start w 0)))) (flow/resume id 9) (flow/result id)")
(list "got" 9))
(flow-api-test
"result: a still-suspended flow has no result"
(flow-a
"(defflow w (lambda (x) (suspend (quote q)))) (define id (car (cdr (flow/start w 0)))) (flow/result id)")
(list "flow-error" "not-done"))
(flow-api-test
"result: unknown id errors"
(flow-a "(flow/result 999)")
(list "flow-error" "no-such-flow"))
;; ── flow/list ───────────────────────────────────────────────────
(flow-api-test "list: empty store" (flow-a "(flow/list)") (list))
(flow-api-test
"list: reports id + status for each flow (newest first)"
(flow-a
"(defflow w (lambda (x) (suspend (quote q)))) (flow/start w 0) (flow/start (lambda (x) (* x 2)) 5) (flow/list)")
(list (list 2 "done") (list 1 "suspended")))
;; ── flow/pending ────────────────────────────────────────────────
(flow-api-test
"pending: lists suspended flows with their waiting tag"
(flow-a
"(defflow w (lambda (x) (suspend (quote review)))) (flow/start w 0) (flow/pending)")
(list (list 1 "review")))
(flow-api-test
"pending: excludes completed and cancelled flows"
(flow-a
"(defflow w (lambda (x) (suspend (quote q)))) (defflow v (sequence (lambda (x) (suspend (quote r))) (lambda (y) y))) (define i1 (car (cdr (flow/start w 0)))) (define i2 (car (cdr (flow/start v 0)))) (define i3 (car (cdr (flow/start w 0)))) (flow/resume i2 1) (flow/cancel i3) (flow/pending)")
(list (list 1 "q")))
(flow-api-test
"pending: operator can drain all pending flows"
(flow-a
"(defflow w (sequence (lambda (x) (suspend (quote q))) (lambda (v) (* v 10)))) (flow/start w 0) (flow/start w 0) (define ps (flow/pending)) (flow/resume (car (car ps)) 1) (flow/resume (car (car (cdr ps))) 2) (flow/list)")
(list (list 1 "done") (list 2 "done")))
(define flow-api-tests-run! (fn () {:total (+ flow-api-pass flow-api-fail) :passed flow-api-pass :failed flow-api-fail :fails flow-api-fails}))

121
lib/flow/tests/basic.sx Normal file
View File

@@ -0,0 +1,121 @@
;; lib/flow/tests/basic.sx — Phase 1: declarative DAG + sequential execution.
(define flow-basic-pass 0)
(define flow-basic-fail 0)
(define flow-basic-fails (list))
(define
flow-basic-test
(fn
(name actual expected)
(if
(= actual expected)
(set! flow-basic-pass (+ flow-basic-pass 1))
(begin
(set! flow-basic-fail (+ flow-basic-fail 1))
(append! flow-basic-fails {:name name :expected expected :actual actual})))))
;; Run a Scheme flow-program string and return its final value.
(define flow-b (fn (src) (flow-run src)))
;; Scheme strings are boxed as {:scm-string "..."}; unwrap to a host string.
(define flow-bs (fn (src) (get (flow-run src) :scm-string)))
;; ── single node ─────────────────────────────────────────────────
(flow-basic-test
"node: identity passes input through"
(flow-b "(flow/start flow-id 7)")
7)
(flow-basic-test
"node: const ignores input"
(flow-b "(flow/start (flow-const 99) 1)")
99)
(flow-basic-test
"node: bare lambda is a node"
(flow-b "(flow/start (lambda (x) (* x x)) 6)")
36)
;; ── linear sequence ─────────────────────────────────────────────
(flow-basic-test
"sequence: empty is identity"
(flow-b "(flow/start (sequence) 42)")
42)
(flow-basic-test
"sequence: single child"
(flow-b "(flow/start (sequence (lambda (x) (+ x 1))) 41)")
42)
(flow-basic-test
"sequence: two children thread"
(flow-b
"(flow/start (sequence (lambda (x) (+ x 1)) (lambda (x) (* x 10))) 4)")
50)
(flow-basic-test
"sequence: three children thread"
(flow-b
"(flow/start (sequence (lambda (x) (+ x 1)) (lambda (x) (* x 2)) (lambda (x) (- x 3))) 5)")
9)
;; ── data flow between nodes ─────────────────────────────────────
(flow-basic-test
"data flow: string accumulation"
(flow-bs
"(flow/start (sequence (lambda (s) (string-append s \"-a\")) (lambda (s) (string-append s \"-b\"))) \"x\")")
"x-a-b")
(flow-basic-test
"data flow: list build"
(flow-b
"(flow/start (sequence (lambda (x) (cons x (list))) (lambda (xs) (cons 0 xs))) 7)")
(list 0 7))
;; ── defflow ─────────────────────────────────────────────────────
(flow-basic-test
"defflow: names a flow"
(flow-b
"(defflow inc2 (sequence (lambda (x) (+ x 1)) (lambda (x) (+ x 1)))) (flow/start inc2 40)")
42)
(flow-basic-test
"defflow: reusable"
(flow-b
"(defflow dbl (lambda (x) (* x 2))) (+ (flow/start dbl 3) (flow/start dbl 10))")
26)
;; ── parallel (sequential semantics, join into list) ─────────────
(flow-basic-test
"parallel: fans input to all branches"
(flow-b
"(flow/start (parallel (lambda (x) (+ x 1)) (lambda (x) (* x 2)) (lambda (x) (- x 3))) 10)")
(list 11 20 7))
(flow-basic-test
"parallel: empty joins to empty list"
(flow-b "(flow/start (parallel) 5)")
(list))
(flow-basic-test
"parallel: single branch"
(flow-b "(flow/start (parallel (lambda (x) (* x x))) 9)")
(list 81))
;; ── nested composition ──────────────────────────────────────────
(flow-basic-test
"nested: sequence of sequences"
(flow-b
"(flow/start (sequence (sequence (lambda (x) (+ x 1)) (lambda (x) (+ x 1))) (sequence (lambda (x) (* x 3)))) 0)")
6)
(flow-basic-test
"nested: parallel inside sequence, join then reduce"
(flow-b
"(flow/start (sequence (parallel (lambda (x) (+ x 1)) (lambda (x) (* x 2))) (lambda (xs) (apply + xs))) 10)")
31)
(flow-basic-test
"nested: sequence inside parallel branch"
(flow-b
"(flow/start (parallel (sequence (lambda (x) (+ x 1)) (lambda (x) (* x 2))) (lambda (x) x)) 5)")
(list 12 5))
;; ── publish-shaped flow (the architecture sketch) ───────────────
(flow-basic-test
"publish: write -> (review | spell) -> join lengths"
(flow-b
"(defflow publish (sequence (lambda (draft) (string-append draft \"!\")) (parallel (lambda (c) (string-length c)) (lambda (c) (string-length (string-append c \"?\")))))) (flow/start publish \"hi\")")
(list 3 4))
(define flow-basic-tests-run! (fn () {:total (+ flow-basic-pass flow-basic-fail) :passed flow-basic-pass :failed flow-basic-fail :fails flow-basic-fails}))

View File

@@ -0,0 +1,108 @@
;; lib/flow/tests/combinators.sx — Phase 5: combinator library (tap, recover, map-flow, iteration).
(define flow-cmb-pass 0)
(define flow-cmb-fail 0)
(define flow-cmb-fails (list))
(define
flow-cmb-test
(fn
(name actual expected)
(if
(= actual expected)
(set! flow-cmb-pass (+ flow-cmb-pass 1))
(begin
(set! flow-cmb-fail (+ flow-cmb-fail 1))
(append! flow-cmb-fails {:name name :expected expected :actual actual})))))
(define flow-m (fn (src) (flow-run src)))
;; ── tap (side-effecting pass-through) ───────────────────────────
(flow-cmb-test
"tap: returns input unchanged"
(flow-m "(flow/start (tap (lambda (x) (* x 999))) 7)")
7)
(flow-cmb-test
"tap: runs the side effect"
(flow-m
"(define seen 0) (flow/start (tap (lambda (x) (set! seen x))) 42) seen")
42)
(flow-cmb-test
"tap: value flows on while the effect observes it"
(flow-m
"(define log 0) (flow/start (sequence (lambda (x) (+ x 1)) (tap (lambda (x) (set! log x))) (lambda (x) (* x 2))) 10) (list log (flow/result 1))")
(list 11 22))
;; ── recover (fail-value counterpart of try-catch) ───────────────
(flow-cmb-test
"recover: passes a non-fail value through"
(flow-m "(flow/start (recover (lambda (x) (* x 2)) (lambda (r) -1)) 5)")
10)
(flow-cmb-test
"recover: handles a fail value via the reason"
(flow-m
"(flow/start (recover (lambda (x) (fail (quote too-small))) (lambda (r) (list (quote recovered) r))) 1)")
(list "recovered" "too-small"))
(flow-cmb-test
"recover: handler can supply a default value"
(flow-m
"(flow/start (sequence (recover (lambda (x) (if (> x 0) x (fail (quote neg))) ) (flow-const 0)) (lambda (x) (* x 10))) -3)")
0)
(flow-cmb-test
"recover: does not catch raised exceptions (those are try-catch's job)"
(flow-m
"(flow/start (try-catch (recover (lambda (x) (raise (quote boom))) (flow-const 0)) (lambda (e) e)) 1)")
"boom")
;; ── map-flow (run a node over a list, join) ─────────────────────
(flow-cmb-test
"map-flow: applies the node to each item"
(flow-m "(flow/start (map-flow (lambda (x) (* x x))) (list 1 2 3 4))")
(list 1 4 9 16))
(flow-cmb-test
"map-flow: empty list joins to empty"
(flow-m "(flow/start (map-flow (lambda (x) (+ x 1))) (list))")
(list))
(flow-cmb-test
"map-flow: each item runs an independent sub-flow"
(flow-m
"(flow/start (map-flow (sequence (lambda (x) (+ x 1)) (lambda (x) (* x 2)))) (list 0 4 9))")
(list 2 10 20))
(flow-cmb-test
"map-flow: composes — fan over a list then reduce the join"
(flow-m
"(flow/start (sequence (map-flow (lambda (x) (* x 10))) (lambda (xs) (apply + xs))) (list 1 2 3))")
60)
;; ── flow-while / flow-until (bounded iteration) ─────────────────
(flow-cmb-test
"flow-while: iterates while the predicate holds"
(flow-m
"(flow/start (flow-while (lambda (x) (< x 10)) (lambda (x) (+ x 1)) 100) 0)")
10)
(flow-cmb-test
"flow-while: a false predicate leaves input unchanged"
(flow-m
"(flow/start (flow-while (lambda (x) (< x 0)) (lambda (x) (+ x 1)) 100) 5)")
5)
(flow-cmb-test
"flow-while: respects the max-iteration bound"
(flow-m "(flow/start (flow-while (lambda (x) #t) (lambda (x) (+ x 1)) 3) 0)")
3)
(flow-cmb-test
"flow-while: doubles until past a threshold"
(flow-m
"(flow/start (flow-while (lambda (x) (< x 50)) (lambda (x) (* x 2)) 100) 3)")
96)
(flow-cmb-test
"flow-until: iterates until the predicate becomes true"
(flow-m
"(flow/start (flow-until (lambda (x) (>= x 10)) (lambda (x) (+ x 3)) 100) 0)")
12)
(flow-cmb-test
"flow-until: composes inside a sequence"
(flow-m
"(flow/start (sequence (flow-until (lambda (x) (> x 100)) (lambda (x) (* x 3)) 100) (lambda (x) (- x 100))) 5)")
35)
(define flow-cmb-tests-run! (fn () {:total (+ flow-cmb-pass flow-cmb-fail) :passed flow-cmb-pass :failed flow-cmb-fail :fails flow-cmb-fails}))

179
lib/flow/tests/control.sx Normal file
View File

@@ -0,0 +1,179 @@
;; lib/flow/tests/control.sx — Phase 2: control flow + error handling.
(define flow-ctl-pass 0)
(define flow-ctl-fail 0)
(define flow-ctl-fails (list))
(define
flow-ctl-test
(fn
(name actual expected)
(if
(= actual expected)
(set! flow-ctl-pass (+ flow-ctl-pass 1))
(begin
(set! flow-ctl-fail (+ flow-ctl-fail 1))
(append! flow-ctl-fails {:name name :expected expected :actual actual})))))
(define flow-c (fn (src) (flow-run src)))
(define flow-cs (fn (src) (get (flow-run src) :scm-string)))
;; ── branch ──────────────────────────────────────────────────────
(flow-ctl-test
"branch: true selects then"
(flow-c
"(flow/start (branch (lambda (x) (> x 0)) (lambda (x) (* x 100)) (lambda (x) (- 0 x))) 5)")
500)
(flow-ctl-test
"branch: false selects else"
(flow-c
"(flow/start (branch (lambda (x) (> x 0)) (lambda (x) (* x 100)) (lambda (x) (- 0 x))) -3)")
3)
(flow-ctl-test
"branch: predicate sees the threaded input"
(flow-c
"(flow/start (sequence (lambda (x) (+ x 1)) (branch (lambda (x) (> x 3)) (flow-const 100) (flow-const 0))) 3)")
100)
(flow-ctl-test
"branch: branches are full nodes (sequence inside)"
(flow-c
"(flow/start (branch (lambda (x) (< x 10)) (sequence (lambda (x) (+ x 1)) (lambda (x) (* x 2))) (flow-const 0)) 4)")
10)
(flow-ctl-test
"branch: nested branch (3-way sign)"
(flow-c
"(defflow sign (branch (lambda (x) (> x 0)) (flow-const 1) (branch (lambda (x) (< x 0)) (flow-const -1) (flow-const 0)))) (list (flow/start sign 7) (flow/start sign -7) (flow/start sign 0))")
(list 1 -1 0))
(flow-ctl-test
"branch: publish-shaped approval gate"
(flow-cs
"(defflow publish (branch (lambda (post) (>= (string-length post) 3)) (lambda (post) (string-append post \" [published]\")) (lambda (post) (string-append post \" [rejected]\")))) (flow/start publish \"ok\")")
"ok [rejected]")
;; ── error model — explicit (fail reason) values ─────────────────
(flow-ctl-test
"fail: failed? is true for a failure value"
(flow-c "(failed? (fail 404))")
true)
(flow-ctl-test
"fail: fail-reason extracts the reason"
(flow-c "(fail-reason (fail 404))")
404)
(flow-ctl-test
"fail: failed? is false for a plain value"
(flow-c "(failed? 7)")
false)
(flow-ctl-test
"fail: failed? is false for an ordinary list"
(flow-c "(failed? (list 1 2 3))")
false)
(flow-ctl-test
"fail: a node may emit a failure as data"
(flow-c
"(defflow validate (lambda (s) (if (>= (string-length s) 3) s (fail (quote too-short))))) (failed? (flow/start validate \"hi\"))")
true)
(flow-ctl-test
"fail: failure flows downstream, branch recovers"
(flow-c
"(defflow guarded (sequence (lambda (s) (if (>= (string-length s) 3) (string-length s) (fail (quote too-short)))) (branch failed? (lambda (f) (list (quote recovered) (fail-reason f))) (lambda (n) (list (quote ok) n))))) (flow/start guarded \"hi\")")
(list "recovered" "too-short"))
;; ── try-catch — reify raised exceptions ─────────────────────────
(flow-ctl-test
"try-catch: no exception returns node result"
(flow-c "(flow/start (try-catch (lambda (x) (* x 2)) (lambda (e) -1)) 5)")
10)
(flow-ctl-test
"try-catch: handler runs on raise"
(flow-c
"(flow/start (try-catch (lambda (x) (raise (quote boom))) (flow-const 99)) 1)")
99)
(flow-ctl-test
"try-catch: handler receives the reified error"
(flow-c "(flow/start (try-catch (lambda (x) (raise 42)) (lambda (e) e)) 0)")
42)
(flow-ctl-test
"try-catch: catches exception from deep inside a sequence"
(flow-c
"(flow/start (try-catch (sequence (lambda (x) (+ x 1)) (lambda (x) (raise (quote deep)))) (flow-const -99)) 5)")
-99)
(flow-ctl-test
"try-catch: handler may convert to a failure value"
(flow-c
"(failed? (flow/start (try-catch (lambda (x) (raise (quote bad))) (lambda (e) (fail e))) 0))")
true)
(flow-ctl-test
"try-catch: composes — recover then continue"
(flow-c
"(flow/start (sequence (try-catch (lambda (x) (raise (quote x))) (flow-const 10)) (lambda (n) (* n 5))) 0)")
50)
;; ── retry — re-run on raised exceptions ─────────────────────────
(flow-ctl-test
"retry: succeeds after transient failures"
(flow-c
"(define ctr 0) (defflow flaky (lambda (x) (set! ctr (+ ctr 1)) (if (< ctr 3) (raise (quote nope)) (* x 10)))) (list (flow/start (retry 5 flaky) 7) ctr)")
(list 70 3))
(flow-ctl-test
"retry: exhausted re-raises (caught by try-catch)"
(flow-c
"(flow/start (try-catch (retry 2 (lambda (x) (raise (quote always)))) (flow-const (quote gaveup))) 0)")
"gaveup")
(flow-ctl-test
"retry: n=1 means a single attempt"
(flow-c
"(define ctr 0) (flow/start (try-catch (retry 1 (lambda (x) (set! ctr (+ ctr 1)) (raise (quote bad)))) (lambda (e) ctr)) 0)")
1)
(flow-ctl-test
"retry: success on first attempt does not re-run"
(flow-c
"(define ctr 0) (flow/start (sequence (retry 5 (lambda (x) (set! ctr (+ ctr 1)) (* x 2))) (lambda (n) ctr)) 21)")
1)
(flow-ctl-test
"retry: does not retry explicit failure values"
(flow-c
"(define ctr 0) (failed? (flow/start (retry 5 (lambda (x) (set! ctr (+ ctr 1)) (fail (quote bad)))) 0))")
true)
(flow-ctl-test
"retry: failure-value path runs node exactly once"
(flow-c
"(define ctr 0) (flow/start (sequence (retry 5 (lambda (x) (set! ctr (+ ctr 1)) (fail (quote bad)))) (lambda (f) ctr)) 0)")
1)
;; ── timeout — cooperative step budget ───────────────────────────
(flow-ctl-test
"timeout: work within budget completes"
(flow-c
"(define (cd n) (if (<= n 0) 99 (begin (tick) (cd (- n 1))))) (flow/start (try-catch (timeout 10 (lambda (x) (cd x))) (flow-const (quote timed-out))) 5)")
99)
(flow-ctl-test
"timeout: work exceeding budget raises flow-timeout"
(flow-c
"(define (cd n) (if (<= n 0) 99 (begin (tick) (cd (- n 1))))) (flow/start (try-catch (timeout 10 (lambda (x) (cd x))) (flow-const (quote timed-out))) 20)")
"timed-out")
(flow-ctl-test
"timeout: exact budget boundary completes"
(flow-c
"(define (cd n) (if (<= n 0) 99 (begin (tick) (cd (- n 1))))) (flow/start (try-catch (timeout 5 (lambda (x) (cd x))) (flow-const (quote timed-out))) 5)")
99)
(flow-ctl-test
"timeout: one tick over the budget raises"
(flow-c
"(define (cd n) (if (<= n 0) 99 (begin (tick) (cd (- n 1))))) (flow/start (try-catch (timeout 5 (lambda (x) (cd x))) (flow-const (quote timed-out))) 6)")
"timed-out")
(flow-ctl-test
"timeout: the raised error is identifiable"
(flow-c
"(define (cd n) (if (<= n 0) 99 (begin (tick) (cd (- n 1))))) (flow/start (try-catch (timeout 2 (lambda (x) (cd x))) (lambda (e) e)) 9)")
"flow-timeout")
(flow-ctl-test
"timeout: a node that never ticks is unbounded"
(flow-c "(flow/start (timeout 0 (lambda (x) (* x 2))) 5)")
10)
(flow-ctl-test
"timeout: budget is restored across sequential timeouts"
(flow-c
"(define (cd n) (if (<= n 0) 1 (begin (tick) (cd (- n 1))))) (flow/start (sequence (timeout 4 (lambda (x) (cd x))) (timeout 4 (lambda (x) (cd 3))) (lambda (x) (begin (tick) (+ x 100)))) 3)")
101)
(define flow-ctl-tests-run! (fn () {:total (+ flow-ctl-pass flow-ctl-fail) :passed flow-ctl-pass :failed flow-ctl-fail :fails flow-ctl-fails}))

View File

@@ -0,0 +1,120 @@
;; lib/flow/tests/distributed.sx — Phase 4: distributed nodes via fed-sx (mocked).
(define flow-dist-pass 0)
(define flow-dist-fail 0)
(define flow-dist-fails (list))
(define
flow-dist-test
(fn
(name actual expected)
(if
(= actual expected)
(set! flow-dist-pass (+ flow-dist-pass 1))
(begin
(set! flow-dist-fail (+ flow-dist-fail 1))
(append! flow-dist-fails {:name name :expected expected :actual actual})))))
(define flow-d (fn (src) (flow-run src)))
;; ── remote-node ─────────────────────────────────────────────────
(flow-dist-test
"remote: a node executes on a peer"
(flow-d
"(flow-peer-register! (quote edge) (list (list (quote double) (lambda (x) (* x 2))))) (flow/start (remote-node (quote edge) (quote double)) 21)")
42)
(flow-dist-test
"remote: remote nodes compose in a sequence"
(flow-d
"(flow-peer-register! (quote edge) (list (list (quote inc) (lambda (x) (+ x 1))) (list (quote double) (lambda (x) (* x 2))))) (flow/start (sequence (remote-node (quote edge) (quote inc)) (remote-node (quote edge) (quote double))) 4)")
10)
(flow-dist-test
"remote: a remote node mixes with local nodes"
(flow-d
"(flow-peer-register! (quote edge) (list (list (quote double) (lambda (x) (* x 2))))) (flow/start (sequence (lambda (x) (+ x 5)) (remote-node (quote edge) (quote double)) (lambda (x) (- x 1))) 10)")
29)
(flow-dist-test
"remote: unreachable peer raises flow-remote-unreachable"
(flow-d
"(flow/start (try-catch (remote-node (quote ghost) (quote double)) (lambda (e) e)) 1)")
"flow-remote-unreachable")
(flow-dist-test
"remote: unknown function on a peer raises flow-remote-no-fn"
(flow-d
"(flow-peer-register! (quote edge) (list (list (quote double) (lambda (x) (* x 2))))) (flow/start (try-catch (remote-node (quote edge) (quote missing)) (lambda (e) e)) 1)")
"flow-remote-no-fn")
(flow-dist-test
"remote: a remote node can suspend the flow (peer returns control)"
(flow-d
"(flow-peer-register! (quote edge) (list (list (quote review) (lambda (x) x)))) (flow/start (sequence (remote-node (quote edge) (quote review)) (lambda (x) (suspend (quote human))) (lambda (v) (list (quote published) v))) 7)")
(list "flow-suspended" 1 "human"))
(flow-dist-test
"remote: a transient remote failure is recoverable with retry"
(flow-d
"(define hits 0) (flow-peer-register! (quote edge) (list (list (quote flaky) (lambda (x) (begin (set! hits (+ hits 1)) (if (< hits 2) (raise (quote down)) (* x 3))))))) (list (flow/start (retry 3 (remote-node (quote edge) (quote flaky))) 7) hits)")
(list 21 2))
;; ── failover (retry on a different peer, fall through to local) ──
(flow-dist-test
"failover: first reachable peer serves the request"
(flow-d
"(flow-peer-register! (quote p2) (list (list (quote f) (lambda (x) (+ x 100))))) (flow/start (remote-failover (list (quote p2) (quote down)) (quote f) (flow-const (quote local))) 5)")
105)
(flow-dist-test
"failover: skips an unreachable peer to the next one"
(flow-d
"(flow-peer-register! (quote p2) (list (list (quote f) (lambda (x) (+ x 100))))) (flow/start (remote-failover (list (quote down) (quote p2)) (quote f) (flow-const (quote local))) 5)")
105)
(flow-dist-test
"failover: skips a peer whose function raises"
(flow-d
"(flow-peer-register! (quote bad) (list (list (quote f) (lambda (x) (raise (quote boom)))))) (flow-peer-register! (quote good) (list (list (quote f) (lambda (x) (* x 10))))) (flow/start (remote-failover (list (quote bad) (quote good)) (quote f) (flow-const 0)) 4)")
40)
(flow-dist-test
"failover: all peers fail, the local fallback runs"
(flow-d
"(flow/start (remote-failover (list (quote down1) (quote down2)) (quote f) (lambda (x) (* x -1))) 9)")
-9)
(flow-dist-test
"failover: threads the input through to the chosen peer"
(flow-d
"(flow-peer-register! (quote p) (list (list (quote f) (lambda (x) (list (quote got) x))))) (flow/start (sequence (lambda (x) (+ x 1)) (remote-failover (list (quote p)) (quote f) (flow-const 0))) 41)")
(list "got" 42))
(flow-dist-test
"failover: composes inside a larger sequence"
(flow-d
"(flow-peer-register! (quote p) (list (list (quote f) (lambda (x) (* x 2))))) (flow/start (sequence (remote-failover (list (quote down) (quote p)) (quote f) (flow-const 1)) (lambda (x) (+ x 3))) 5)")
13)
;; ── replication + handoff ───────────────────────────────────────
(flow-dist-test
"replicate: a peer holds the exported store"
(flow-d
"(defflow w (lambda (x) (suspend (quote q)))) (flow/start w 10) (flow-replicate-to (quote peerB)) (if (flow-replica-get (quote peerB)) (quote replicated) (quote missing))")
"replicated")
(flow-dist-test
"handoff: a peer resumes a flow after the local instance dies"
(flow-d
"(defflow w (sequence (lambda (x) (suspend (quote q))) (lambda (v) (list (quote done) v)))) (define id (car (cdr (flow/start w 10)))) (flow-replicate-to (quote peerB)) (set! flow-store (list)) (flow-restore-from (quote peerB)) (flow/resume id 55)")
(list "done" 55))
(flow-dist-test
"handoff: restored peer reports the flow as resumable"
(flow-d
"(defflow w (lambda (x) (suspend (quote q)))) (define id (car (cdr (flow/start w 10)))) (flow-replicate-to (quote peerB)) (set! flow-store (list)) (flow-restore-from (quote peerB)) (flow-resumable-ids)")
(list 1))
(flow-dist-test
"handoff: without restore the dead instance has lost the flow"
(flow-d
"(defflow w (lambda (x) (suspend (quote q)))) (define id (car (cdr (flow/start w 10)))) (flow-replicate-to (quote peerB)) (set! flow-store (list)) (flow/resume id 1)")
(list "flow-error" "no-such-flow"))
(flow-dist-test
"restore: from an unknown peer yields false"
(flow-d "(flow-restore-from (quote nowhere))")
false)
(flow-dist-test
"handoff: replication preserves the replay log across the move"
(flow-d
"(defflow two (sequence (lambda (x) (suspend (quote a))) (lambda (x) (suspend (quote b))) (lambda (x) (list x)))) (define id (car (cdr (flow/start two 0)))) (flow/resume id 11) (flow-replicate-to (quote peerB)) (set! flow-store (list)) (flow-restore-from (quote peerB)) (flow/resume id 22)")
(list 22))
(define flow-dist-tests-run! (fn () {:total (+ flow-dist-pass flow-dist-fail) :passed flow-dist-pass :failed flow-dist-fail :fails flow-dist-fails}))

106
lib/flow/tests/host.sx Normal file
View File

@@ -0,0 +1,106 @@
;; lib/flow/tests/host.sx — Phase 8: host integration ABI (request/await/host-queue/driver).
(define flow-hst-pass 0)
(define flow-hst-fail 0)
(define flow-hst-fails (list))
(define
flow-hst-test
(fn
(name actual expected)
(if
(= actual expected)
(set! flow-hst-pass (+ flow-hst-pass 1))
(begin
(set! flow-hst-fail (+ flow-hst-fail 1))
(append! flow-hst-fails {:name name :expected expected :actual actual})))))
(define flow-hst (fn (src) (flow-run src)))
;; ── request envelope ────────────────────────────────────────────
(flow-hst-test
"request: suspends with a typed envelope"
(flow-hst
"(car (cdr (cdr (flow/start (lambda (x) (request (quote render) x)) 5))))")
(list "flow-request" "render" 5))
(flow-hst-test
"request?: recognizes an envelope"
(flow-hst "(request? (list (quote flow-request) (quote human) 1))")
true)
(flow-hst-test
"request?: a plain tag is not a request"
(flow-hst "(request? (list (quote review) 1))")
false)
(flow-hst-test
"request-kind / request-payload: parse the envelope"
(flow-hst
"(define t (list (quote flow-request) (quote render) (list (quote recipe) 7))) (list (request-kind t) (request-payload t))")
(list "render" (list "recipe" 7)))
;; ── named decision points ───────────────────────────────────────
(flow-hst-test
"await-human: is a request of kind human"
(flow-hst
"(car (cdr (cdr (flow/start (lambda (x) (await-human x)) (quote approve?)))))")
(list "flow-request" "human" "approve?"))
(flow-hst-test
"await-render: is a request of kind render"
(flow-hst
"(car (cdr (cdr (flow/start (lambda (x) (await-render x)) (quote recipe)))))")
(list "flow-request" "render" "recipe"))
(flow-hst-test
"request: the host's resume value flows back into the flow"
(flow-hst
"(defflow f (sequence (lambda (x) (await-render x)) (lambda (art) (list (quote got) art)))) (define id (car (cdr (flow/start f 1)))) (flow/resume id (quote the-artifact))")
(list "got" "the-artifact"))
;; ── host work queue ─────────────────────────────────────────────
(flow-hst-test
"flow-host-requests: lists (id kind payload) for pending requests"
(flow-hst
"(flow/start (lambda (x) (await-render x)) 99) (flow-host-requests)")
(list (list 1 "render" 99)))
(flow-hst-test
"flow-host-requests: excludes bare (non-request) suspends"
(flow-hst
"(defflow a (lambda (x) (await-render x))) (defflow b (lambda (x) (suspend (quote plain)))) (flow/start a 1) (flow/start b 2) (flow-host-requests)")
(list (list 1 "render" 1)))
;; ── the art-dag-shaped host driver loop (manual resumes) ────────
(flow-hst-test
"host driver: render then human-review then publish"
(flow-hst
"(defflow pipeline (sequence (lambda (recipe) (await-render recipe)) (lambda (art) (await-human (list (quote review) art))) (branch (lambda (d) (eq? d (quote approve))) (flow-const (quote published)) (flow-const (fail (quote rejected)))))) (define id (car (cdr (flow/start pipeline 99)))) (define r1 (flow-host-requests)) (flow/resume id (list (quote art) 99)) (define r2 (flow-host-requests)) (flow/resume id (quote approve)) (list r1 r2 (flow/status id) (flow/result id))")
(list
(list (list 1 "render" 99))
(list (list 1 "human" (list "review" (list "art" 99))))
"done"
"published"))
(flow-hst-test
"host driver: rejection at the human gate yields a failure"
(flow-hst
"(defflow pipeline (sequence (lambda (recipe) (await-render recipe)) (lambda (art) (await-human (list (quote review) art))) (branch (lambda (d) (eq? d (quote approve))) (flow-const (quote published)) (flow-const (fail (quote rejected)))))) (define id (car (cdr (flow/start pipeline 1)))) (flow/resume id (quote artifact)) (failed? (flow/resume id (quote reject)))")
true)
;; ── reference driver: host supplies only a dispatch fn ──────────
(flow-hst-test
"flow-drive-host: one tick services every pending request"
(flow-hst
"(flow/start (lambda (x) (await-render x)) 5) (define n (flow-drive-host (lambda (k p) (list (quote done) p)))) (list n (flow/status 1) (flow/result 1))")
(list 1 "done" (list "done" 5)))
(flow-hst-test
"flow-run-host: drives a render -> human pipeline to completion"
(flow-hst
"(defflow pipeline (sequence (lambda (recipe) (await-render recipe)) (lambda (art) (await-human (list (quote review) art))) (branch (lambda (d) (eq? d (quote approve))) (flow-const (quote published)) (flow-const (fail (quote rejected)))))) (define id (car (cdr (flow/start pipeline 99)))) (define serviced (flow-run-host (lambda (kind payload) (if (eq? kind (quote render)) (list (quote art) payload) (quote approve))) 10)) (list serviced (flow/status id) (flow/result id))")
(list 2 "done" "published"))
(flow-hst-test
"flow-run-host: returns 0 when nothing is pending"
(flow-hst "(flow-run-host (lambda (k p) p) 5)")
0)
(flow-hst-test
"flow-run-host: respects the maxticks bound"
(flow-hst
"(defflow pipe2 (sequence (lambda (r) (await-render r)) (lambda (a) (await-human a)) (lambda (d) d))) (define id (car (cdr (flow/start pipe2 1)))) (define serviced (flow-run-host (lambda (k p) p) 1)) (list serviced (flow/status id))")
(list 1 "suspended"))
(define flow-hst-tests-run! (fn () {:total (+ flow-hst-pass flow-hst-fail) :passed flow-hst-pass :failed flow-hst-fail :fails flow-hst-fails}))

67
lib/flow/tests/hygiene.sx Normal file
View File

@@ -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}))

View File

@@ -0,0 +1,115 @@
;; lib/flow/tests/integration.sx — Phase 7: end-to-end flows composing every phase.
(define flow-int-pass 0)
(define flow-int-fail 0)
(define flow-int-fails (list))
(define
flow-int-test
(fn
(name actual expected)
(if
(= actual expected)
(set! flow-int-pass (+ flow-int-pass 1))
(begin
(set! flow-int-fail (+ flow-int-fail 1))
(append! flow-int-fails {:name name :expected expected :actual actual})))))
(define flow-i (fn (src) (flow-run src)))
;; The order-processing flow, defined once per program via this prelude string:
;; validate amount (attempt: fail if <= 0)
;; -> suspend for payment confirmation (resume value = confirmed amount)
;; -> branch: confirmed>0 ? record on the ledger peer : declined failure
(define
order-prelude
"(flow-peer-register! (quote ledger) (list (list (quote record) (lambda (amt) (list (quote recorded) amt)))))\n (defflow order\n (attempt\n (lambda (amt) (if (> amt 0) amt (fail (quote invalid-amount))))\n (lambda (amt) (suspend (quote await-payment)))\n (branch (lambda (amt) (> amt 0))\n (remote-node (quote ledger) (quote record))\n (flow-const (fail (quote declined))))))")
;; ── happy path through every phase ──────────────────────────────
(flow-int-test
"order: validate -> suspend -> resume -> branch -> federate"
(flow-i
(str
order-prelude
"(define id (car (cdr (flow/start order 100)))) (flow/resume id 250)"))
(list "recorded" 250))
(flow-int-test
"order: starting suspends awaiting payment"
(flow-i
(str
order-prelude
"(define s (flow/start order 100)) (list (car s) (car (cdr (cdr s))))"))
(list "flow-suspended" "await-payment"))
(flow-int-test
"order: invalid amount fails up front and never suspends"
(flow-i
(str
order-prelude
"(define r (flow/start order -5)) (list (failed? r) (fail-reason r))"))
(list true "invalid-amount"))
(flow-int-test
"order: a declined payment yields a failure value"
(flow-i
(str
order-prelude
"(define id (car (cdr (flow/start order 100)))) (failed? (flow/resume id 0))"))
true)
;; ── crash recovery mid-flow ─────────────────────────────────────
(flow-int-test
"order: survives a simulated crash between suspend and resume"
(flow-i
(str
order-prelude
"(define id (car (cdr (flow/start order 100)))) (define saved (flow-store-export)) (set! flow-store (list)) (flow-store-import! saved) (flow/resume id 250)"))
(list "recorded" 250))
;; ── handoff to a peer mid-flow ──────────────────────────────────
(flow-int-test
"order: hands off to a peer that resumes and completes"
(flow-i
(str
order-prelude
"(define id (car (cdr (flow/start order 100)))) (flow-replicate-to (quote nodeB)) (set! flow-store (list)) (flow-restore-from (quote nodeB)) (flow/resume id 250)"))
(list "recorded" 250))
;; ── introspection during the flow's life ────────────────────────
(flow-int-test
"order: pending shows what the flow awaits, then result after resume"
(flow-i
(str
order-prelude
"(define id (car (cdr (flow/start order 100)))) (define p (flow/pending)) (flow/resume id 250) (list p (flow/status id) (flow/result id))"))
(list
(list (list 1 "await-payment"))
"done"
(list "recorded" 250)))
;; ── onboarding: two human steps + cancellation ──────────────────
(define
onboard-prelude
"(defflow onboard\n (sequence\n (lambda (user) (+ user 1))\n (lambda (x) (suspend (quote confirm-email)))\n (lambda (x) (suspend (quote complete-profile)))\n (lambda (x) (list (quote onboarded) x))))")
(flow-int-test
"onboard: two suspends resume in order to completion"
(flow-i
(str
onboard-prelude
"(define id (car (cdr (flow/start onboard 0)))) (flow/resume id 7) (flow/resume id 9)"))
(list "onboarded" 9))
(flow-int-test
"onboard: the second pending tag appears after the first resume"
(flow-i
(str
onboard-prelude
"(define id (car (cdr (flow/start onboard 0)))) (flow/resume id 7) (car (cdr (car (flow/pending))))"))
"complete-profile")
(flow-int-test
"onboard: cancelling abandons the flow"
(flow-i
(str
onboard-prelude
"(define id (car (cdr (flow/start onboard 0)))) (flow/cancel id) (list (flow/status id) (car (flow/resume id 7)))"))
(list "cancelled" "flow-error"))
(define flow-int-tests-run! (fn () {:total (+ flow-int-pass flow-int-fail) :passed flow-int-pass :failed flow-int-fail :fails flow-int-fails}))

73
lib/flow/tests/railway.sx Normal file
View File

@@ -0,0 +1,73 @@
;; lib/flow/tests/railway.sx — Phase 6: railway-oriented composition (attempt).
(define flow-rail-pass 0)
(define flow-rail-fail 0)
(define flow-rail-fails (list))
(define
flow-rail-test
(fn
(name actual expected)
(if
(= actual expected)
(set! flow-rail-pass (+ flow-rail-pass 1))
(begin
(set! flow-rail-fail (+ flow-rail-fail 1))
(append! flow-rail-fails {:name name :expected expected :actual actual})))))
(define flow-r (fn (src) (flow-run src)))
;; ── attempt — short-circuit on the first (fail ...) ─────────────
(flow-rail-test
"attempt: threads like sequence when nothing fails"
(flow-r
"(flow/start (attempt (lambda (x) (+ x 1)) (lambda (x) (* x 10))) 4)")
50)
(flow-rail-test
"attempt: empty is identity"
(flow-r "(flow/start (attempt) 7)")
7)
(flow-rail-test
"attempt: returns the first failure"
(flow-r
"(failed? (flow/start (attempt (lambda (x) (fail (quote bad))) (lambda (x) (* x 10))) 4))")
true)
(flow-rail-test
"attempt: the failure carries its reason"
(flow-r
"(fail-reason (flow/start (attempt (lambda (x) x) (lambda (x) (fail (quote rejected)))) 4))")
"rejected")
(flow-rail-test
"attempt: nodes after a failure do not run"
(flow-r
"(define ran 0) (flow/start (attempt (lambda (x) (fail (quote stop))) (lambda (x) (begin (set! ran (+ ran 1)) x))) 0) ran")
0)
(flow-rail-test
"attempt: a failed input short-circuits immediately"
(flow-r
"(define ran 0) (fail-reason (flow/start (attempt (lambda (x) (begin (set! ran (+ ran 1)) x))) (fail (quote pre))))")
"pre")
(flow-rail-test
"attempt: middle failure halts the chain"
(flow-r
"(define ran 0) (flow/start (attempt (lambda (x) (+ x 1)) (lambda (x) (fail (quote mid))) (lambda (x) (begin (set! ran (+ ran 1)) x))) 5) ran")
0)
;; ── attempt + recover (rejoin the happy track) ──────────────────
(flow-rail-test
"attempt + recover: recover turns a failure into a value"
(flow-r
"(flow/start (recover (attempt (lambda (x) (if (> x 0) x (fail (quote non-positive)))) (lambda (x) (* x 2))) (flow-const 0)) -5)")
0)
(flow-rail-test
"attempt + recover: happy path passes recover through"
(flow-r
"(flow/start (recover (attempt (lambda (x) (if (> x 0) x (fail (quote non-positive)))) (lambda (x) (* x 2))) (flow-const 0)) 5)")
10)
(flow-rail-test
"attempt: validation pipeline reports the failing stage"
(flow-r
"(defflow validate (attempt (lambda (s) (if (>= (string-length s) 3) s (fail (quote too-short)))) (lambda (s) (if (<= (string-length s) 8) s (fail (quote too-long)))) (lambda (s) (list (quote ok) (string-length s))))) (list (fail-reason (flow/start validate \"hi\")) (flow/start validate \"hello\"))")
(list "too-short" (list "ok" 5)))
(define flow-rail-tests-run! (fn () {:total (+ flow-rail-pass flow-rail-fail) :passed flow-rail-pass :failed flow-rail-fail :fails flow-rail-fails}))

View File

@@ -0,0 +1,71 @@
;; lib/flow/tests/recovery.sx — Phase 3: crash recovery (store export/import + restart).
;;
;; "restart" is simulated within one program: (set! flow-store (list)) wipes the
;; in-memory store (process death), while flow-registry persists as it would after
;; reloading flow definitions. Recovery = import the exported (plain-data) store and
;; resume; the flow proc is re-resolved by name.
(define flow-rec-pass 0)
(define flow-rec-fail 0)
(define flow-rec-fails (list))
(define
flow-rec-test
(fn
(name actual expected)
(if
(= actual expected)
(set! flow-rec-pass (+ flow-rec-pass 1))
(begin
(set! flow-rec-fail (+ flow-rec-fail 1))
(append! flow-rec-fails {:name name :expected expected :actual actual})))))
(define flow-r (fn (src) (flow-run src)))
;; ── export / wipe / import ──────────────────────────────────────
(flow-rec-test
"export nulls the live procedure"
(flow-r
"(defflow w (lambda (x) (suspend (quote await)))) (flow/start w 10) (car (cdr (car (cdr (car (flow-store-export))))))")
false)
(flow-rec-test
"a wiped store loses the flow (process death)"
(flow-r
"(defflow w (lambda (x) (suspend (quote await)))) (define id (car (cdr (flow/start w 10)))) (set! flow-store (list)) (flow/resume id 1)")
(list "flow-error" "no-such-flow"))
(flow-rec-test
"import restores a wiped store and resume completes"
(flow-r
"(defflow w (sequence (lambda (x) (suspend (quote await))) (lambda (c) (list (quote done) c)))) (define id (car (cdr (flow/start w 10)))) (define saved (flow-store-export)) (set! flow-store (list)) (flow-store-import! saved) (flow/resume id 777)")
(list "done" 777))
;; ── resumable scan ──────────────────────────────────────────────
(flow-rec-test
"resumable-ids lists the suspended flow after import"
(flow-r
"(defflow w (lambda (x) (suspend (quote await)))) (define id (car (cdr (flow/start w 10)))) (define saved (flow-store-export)) (set! flow-store (list)) (flow-store-import! saved) (flow-resumable-ids)")
(list 1))
(flow-rec-test
"resumable-ids excludes completed flows"
(flow-r
"(defflow w (sequence (lambda (x) (suspend (quote await))) (lambda (c) c))) (define id (car (cdr (flow/start w 10)))) (flow/resume id 5) (flow-resumable-ids)")
(list))
(flow-rec-test
"resumable-ids excludes cancelled flows after import"
(flow-r
"(defflow w (lambda (x) (suspend (quote await)))) (define id (car (cdr (flow/start w 10)))) (flow/cancel id) (define saved (flow-store-export)) (set! flow-store (list)) (flow-store-import! saved) (flow-resumable-ids)")
(list))
;; ── restart at every step ───────────────────────────────────────
(flow-rec-test
"two suspends survive a restart between each step"
(flow-r
"(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)))) (define s1 (flow-store-export)) (set! flow-store (list)) (flow-store-import! s1) (flow/resume id 100) (define s2 (flow-store-export)) (set! flow-store (list)) (flow-store-import! s2) (flow/resume id 200)")
(list "end" 200))
(flow-rec-test
"import preserves the replay log (earlier value survives restart)"
(flow-r
"(defflow two (sequence (lambda (x) (suspend (quote a))) (lambda (x) (suspend (quote b))) (lambda (x) (list x)))) (define id (car (cdr (flow/start two 0)))) (flow/resume id 11) (define saved (flow-store-export)) (set! flow-store (list)) (flow-store-import! saved) (flow/resume id 22)")
(list 22))
(define flow-rec-tests-run! (fn () {:total (+ flow-rec-pass flow-rec-fail) :passed flow-rec-pass :failed flow-rec-fail :fails flow-rec-fails}))

114
lib/flow/tests/suspend.sx Normal file
View File

@@ -0,0 +1,114 @@
;; 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}))

View File

@@ -1,10 +0,0 @@
; persist/api — the public entry point. persist/open returns a backend (the
; in-memory one by default; pass a custom backend to inject file/pg/ipfs-ref).
; All facet functions take this backend as their first argument.
; Requires: lib/persist/backend.sx, lib/persist/log.sx, lib/persist/kv.sx.
(define
persist/open
(fn
(&rest args)
(if (= (len args) 0) (persist/mem-backend) (first args))))

View File

@@ -1,34 +0,0 @@
; persist/backend — the injected storage protocol. Every facet (log, kv,
; snapshot) goes through a backend dict, never touching storage directly, so
; file/pg/ipfs-ref backends swap in unchanged. A backend is a dict of fns:
; {:append :read :last-seq :truncate-through :streams
; :kv-get :kv-put :kv-delete :kv-has? :kv-keys}
; The in-memory backend is the test default. State is three dicts held in a
; closure and mutated with set!: logs (stream -> event list), seqs (stream ->
; last assigned seq — a monotonic high-water mark that survives compaction so
; truncating the log prefix never lets a future append reuse a seq), kv. The
; stream catalog comes from seqs, so a fully-compacted stream still lists.
(define
persist/mem-backend
(fn
()
(let ((logs {}) (seqs {}) (kv {})) {:truncate-through (fn (stream n) (let ((cur (get logs stream))) (set! logs (assoc logs stream (filter (fn (e) (> (persist/event-seq e) n)) (if cur cur (list))))))) :kv-keys (fn () (keys kv)) :read (fn (stream) (let ((cur (get logs stream))) (if cur cur (list)))) :kv-has? (fn (key) (has-key? kv key)) :last-seq (fn (stream) (let ((s (get seqs stream))) (if s s 0))) :streams (fn () (keys seqs)) :append (fn (stream event) (begin (let ((cur (get logs stream))) (set! logs (assoc logs stream (append (if cur cur (list)) event)))) (set! seqs (assoc seqs stream (persist/event-seq event))))) :kv-delete (fn (key) (set! kv (dissoc kv key))) :kv-put (fn (key val) (set! kv (assoc kv key val))) :kv-get (fn (key) (get kv key))})))
; protocol accessors — call a backend op by keyword
(define
persist/backend-append
(fn (b stream event) ((get b :append) stream event)))
(define persist/backend-read (fn (b stream) ((get b :read) stream)))
(define
persist/backend-last-seq
(fn (b stream) ((get b :last-seq) stream)))
(define persist/backend-streams (fn (b) ((get b :streams))))
(define
persist/backend-truncate
(fn (b stream n) ((get b :truncate-through) stream n)))
(define persist/backend-kv-get (fn (b key) ((get b :kv-get) key)))
(define persist/backend-kv-put (fn (b key val) ((get b :kv-put) key val)))
(define persist/backend-kv-delete (fn (b key) ((get b :kv-delete) key)))
(define persist/backend-kv-has? (fn (b key) ((get b :kv-has?) key)))
(define persist/backend-kv-keys (fn (b) ((get b :kv-keys))))

View File

@@ -1,40 +0,0 @@
; persist/batch — commit several events to a stream as one contiguous block.
; Each spec is (type at data). Plain append-batch always appends; the -expect
; form is the transactional commit: it checks the stream is still at `expected`
; before writing ANY event, so a batch is all-or-nothing under a concurrent
; writer (conflict is a value, not a partial write). For an order + its line
; items, an audit entry + its reason, etc. Requires: lib/persist/log.sx.
; append a list of (type at data) specs as one block; returns the stored events
; (a real cons-list, in order, with contiguous seqs)
(define
persist/append-batch
(fn
(b stream specs)
(reverse
(reduce
(fn
(acc spec)
(cons
(persist/append
b
stream
(first spec)
(nth spec 1)
(nth spec 2))
acc))
(list)
specs))))
; transactional batch: commit all specs only if the stream is still at expected,
; else return a conflict and write nothing
(define
persist/append-batch-expect
(fn
(b stream expected specs)
(let
((actual (persist/last-seq b stream)))
(if
(= actual expected)
(persist/append-batch b stream specs)
{:actual actual :expected expected :conflict true}))))

View File

@@ -1,66 +0,0 @@
; persist/blob — large objects (images, media) are NOT persist's to hold. They
; live in a content-addressed store (artdag/IPFS); persist stores only a
; reference: {:cid :size :mime}. The blob store is a SEPARATE injected
; dependency with its own transport (perform in production, a mock content store
; in tests), distinct from the event/kv backend. The invariant: a blob ref that
; lands in the log or kv carries the CID + metadata and never the bytes.
; Requires: lib/persist/backend.sx.
(define persist/blob-ref (fn (cid size mime) {:mime mime :size size :cid cid}))
(define persist/blob-ref? (fn (r) (has-key? r :cid)))
(define persist/blob-cid (fn (r) (get r :cid)))
(define persist/blob-size (fn (r) (get r :size)))
(define persist/blob-mime (fn (r) (get r :mime)))
; blob store protocol over an injectable transport
(define persist/blob-io (fn (transport) {:put (fn (bytes mime) (transport {:op "blob/put" :args (list bytes mime)})) :get (fn (cid) (transport {:op "blob/get" :args (list cid)})) :has? (fn (cid) (transport {:op "blob/has?" :args (list cid)}))}))
; production blob store — transport is the kernel's perform
(define
persist/blob-store-backend
(fn () (persist/blob-io (fn (req) (perform req)))))
; store bytes via the blob backend; return ONLY the ref (cid + metadata) — this
; is what the caller persists in the log/kv. The bytes never enter persist.
(define
persist/blob-store
(fn
(blob bytes mime)
(let
((cid ((get blob :put) bytes mime)))
(persist/blob-ref cid (len bytes) mime))))
(define
persist/blob-fetch
(fn (blob ref) ((get blob :get) (persist/blob-cid ref))))
(define
persist/blob-exists?
(fn (blob ref) ((get blob :has?) (persist/blob-cid ref))))
; mock content-addressed store (stands in for artdag/IPFS). CID is a
; deterministic content address: identical bytes dedupe to one CID. A real
; store computes a SHA3/IPFS CID host-side; the prefix keeps the mock readable.
(define persist/blob-cid-of (fn (bytes) (str "cid:" bytes)))
(define
persist/blob-serve
(fn
(store req)
(let
((op (get req :op)) (args (get req :args)))
(cond
((equal? op "blob/put")
(let
((cid (persist/blob-cid-of (first args))))
(begin (persist/backend-kv-put store cid (first args)) cid)))
((equal? op "blob/get") (persist/backend-kv-get store (first args)))
((equal? op "blob/has?")
(persist/backend-kv-has? store (first args)))
(else (error (str "persist/blob-serve: unknown op " op)))))))
(define
persist/blob-mock-transport
(fn (store) (fn (req) (persist/blob-serve store req))))
(define
persist/mock-blob
(fn (store) (persist/blob-io (persist/blob-mock-transport store))))

View File

@@ -1,35 +0,0 @@
; persist/catalog — enumerate the streams a backend holds. The catalog is the
; set of streams ever appended to (from the seq high-water marks), so a stream
; whose log has been fully compacted still appears. $-prefixed streams are
; reserved for internal indexes (e.g. the $global commit index) and are hidden
; from the public catalog; use streams-all to see them. For admin, global ops,
; and cross-stream tooling. Requires: lib/persist/backend.sx, lib/persist/log.sx.
(define persist/reserved-stream? (fn (s) (starts-with? s "$")))
; every stream including reserved internal indexes
(define persist/streams-all (fn (b) (persist/backend-streams b)))
; public streams (reserved internal indexes hidden)
(define
persist/streams
(fn
(b)
(filter
(fn (s) (not (persist/reserved-stream? s)))
(persist/streams-all b))))
(define persist/stream-count (fn (b) (len (persist/streams b))))
(define
persist/stream-exists?
(fn (b stream) (contains? (persist/streams b) stream)))
; total logical events across all public streams (sum of high-water marks)
(define
persist/total-events
(fn
(b)
(reduce
(fn (acc s) (+ acc (persist/last-seq b s)))
0
(persist/streams b))))

View File

@@ -1,43 +0,0 @@
; persist/compaction — once a snapshot subsumes a log prefix, those events are
; dead weight: replay starts from the snapshot, so events with seq <= the
; snapshot's seq are never folded again. Compaction checkpoints then truncates
; that prefix. The seq counter is monotonic (backend high-water mark) so future
; appends keep climbing — the surviving tail keeps its original seqs and replay
; from the snapshot still equals a full replay of the pre-compaction log.
; Policy is explicit: compact when the uncompacted tail reaches `every` events.
; Requires: lib/persist/snapshot.sx, lib/persist/log.sx.
; events accumulated since the last snapshot for name
(define
persist/uncompacted
(fn
(b stream name seed)
(-
(persist/last-seq b stream)
(persist/project-seq (persist/snapshot-load b name seed)))))
; policy: should we compact yet? tail since snapshot >= every
(define
persist/should-compact?
(fn
(b stream name every seed)
(>= (persist/uncompacted b stream name seed) every)))
; checkpoint then drop the snapshotted prefix; returns the new snapshot state
(define
persist/compact
(fn
(b stream name step seed)
(let
((state (persist/checkpoint b stream name step seed)))
(begin (persist/truncate b stream (persist/project-seq state)) state))))
; compact only if the policy fires; always returns the current snapshot state
(define
persist/maybe-compact
(fn
(b stream name step seed every)
(if
(persist/should-compact? b stream name every seed)
(persist/compact b stream name step seed)
(persist/snapshot-load b name seed))))

View File

@@ -1,24 +0,0 @@
; persist/concurrency — optimistic concurrency for the log facet. The caller
; passes the seq it believes is current (the last-seq it last observed). If the
; stream has advanced since, the append is refused and a conflict VALUE is
; returned — never a crash, never a silent overwrite. The caller re-reads the
; tail and retries. This is the substrate-level answer to "two writers, one
; stream": the loser gets a result it can act on.
; Requires: lib/persist/log.sx.
(define
persist/append-expect
(fn
(b stream expected type at data)
(let
((actual (persist/last-seq b stream)))
(if
(= actual expected)
(persist/append b stream type at data)
{:actual actual :expected expected :conflict true}))))
(define
persist/conflict?
(fn (r) (if (has-key? r :conflict) (get r :conflict) false)))
(define persist/conflict-expected (fn (r) (get r :expected)))
(define persist/conflict-actual (fn (r) (get r :actual)))

View File

@@ -1,128 +0,0 @@
#!/usr/bin/env bash
# lib/persist/conformance.sh — run persist test suites, emit scoreboard.json + scoreboard.md.
set -uo pipefail
cd "$(git rev-parse --show-toplevel)"
SX_SERVER="${SX_SERVER:-/root/rose-ash/hosts/ocaml/_build/default/bin/sx_server.exe}"
if [ ! -x "$SX_SERVER" ]; then
SX_SERVER="hosts/ocaml/_build/default/bin/sx_server.exe"
fi
if [ ! -x "$SX_SERVER" ]; then
echo "ERROR: sx_server.exe not found." >&2
exit 1
fi
SUITES=(event log kv project subscribe concurrency snapshot compaction durable blob view cas catalog query batch upcast idempotency global example-acl recovery)
OUT_JSON="lib/persist/scoreboard.json"
OUT_MD="lib/persist/scoreboard.md"
run_suite() {
local suite=$1
local file="lib/persist/tests/${suite}.sx"
local TMP
TMP=$(mktemp)
cat > "$TMP" << EPOCHS
(epoch 1)
(load "spec/stdlib.sx")
(load "lib/r7rs.sx")
(load "lib/persist/event.sx")
(load "lib/persist/backend.sx")
(load "lib/persist/log.sx")
(load "lib/persist/kv.sx")
(load "lib/persist/project.sx")
(load "lib/persist/concurrency.sx")
(load "lib/persist/snapshot.sx")
(load "lib/persist/compaction.sx")
(load "lib/persist/durable.sx")
(load "lib/persist/blob.sx")
(load "lib/persist/view.sx")
(load "lib/persist/catalog.sx")
(load "lib/persist/query.sx")
(load "lib/persist/batch.sx")
(load "lib/persist/upcast.sx")
(load "lib/persist/idempotency.sx")
(load "lib/persist/global.sx")
(load "lib/persist/examples/acl.sx")
(load "lib/persist/subscribe.sx")
(load "lib/persist/api.sx")
(epoch 2)
(eval "(define persist-test-pass 0)")
(eval "(define persist-test-fail 0)")
(eval "(define persist-test (fn (name got expected) (if (equal? got expected) (set! persist-test-pass (+ persist-test-pass 1)) (set! persist-test-fail (+ persist-test-fail 1)))))")
(epoch 3)
(load "${file}")
(epoch 4)
(eval "(list persist-test-pass persist-test-fail)")
EPOCHS
local OUTPUT
OUTPUT=$(timeout 300 "$SX_SERVER" < "$TMP" 2>/dev/null)
rm -f "$TMP"
local LINE
LINE=$(echo "$OUTPUT" | awk '/^\(ok-len 4 / {getline; print; exit}')
if [ -z "$LINE" ]; then
LINE=$(echo "$OUTPUT" | grep -E '^\(ok 4 \([0-9]+ [0-9]+\)\)' | tail -1 \
| sed -E 's/^\(ok 4 //; s/\)$//')
fi
local P F
P=$(echo "$LINE" | sed -E 's/^\(([0-9]+) ([0-9]+)\).*/\1/')
F=$(echo "$LINE" | sed -E 's/^\(([0-9]+) ([0-9]+)\).*/\2/')
P=${P:-0}
F=${F:-0}
echo "${P} ${F}"
}
declare -A SUITE_PASS
declare -A SUITE_FAIL
TOTAL_PASS=0
TOTAL_FAIL=0
echo "Running persist conformance suite..." >&2
for s in "${SUITES[@]}"; do
read -r p f < <(run_suite "$s")
SUITE_PASS[$s]=$p
SUITE_FAIL[$s]=$f
TOTAL_PASS=$((TOTAL_PASS + p))
TOTAL_FAIL=$((TOTAL_FAIL + f))
printf " %-12s %d/%d\n" "$s" "$p" "$((p+f))" >&2
done
# scoreboard.json
{
printf '{\n'
printf ' "suites": {\n'
first=1
for s in "${SUITES[@]}"; do
if [ $first -eq 0 ]; then printf ',\n'; fi
printf ' "%s": {"pass": %d, "fail": %d}' "$s" "${SUITE_PASS[$s]}" "${SUITE_FAIL[$s]}"
first=0
done
printf '\n },\n'
printf ' "total_pass": %d,\n' "$TOTAL_PASS"
printf ' "total_fail": %d,\n' "$TOTAL_FAIL"
printf ' "total": %d\n' "$((TOTAL_PASS + TOTAL_FAIL))"
printf '}\n'
} > "$OUT_JSON"
# scoreboard.md
{
printf '# persist Conformance Scoreboard\n\n'
printf '_Generated by `lib/persist/conformance.sh`_\n\n'
printf '| Suite | Pass | Fail | Total |\n'
printf '|-------|-----:|-----:|------:|\n'
for s in "${SUITES[@]}"; do
p=${SUITE_PASS[$s]}
f=${SUITE_FAIL[$s]}
printf '| %s | %d | %d | %d |\n' "$s" "$p" "$f" "$((p+f))"
done
printf '| **Total** | **%d** | **%d** | **%d** |\n' "$TOTAL_PASS" "$TOTAL_FAIL" "$((TOTAL_PASS + TOTAL_FAIL))"
} > "$OUT_MD"
echo "Wrote $OUT_JSON and $OUT_MD" >&2
echo "Total: $TOTAL_PASS pass, $TOTAL_FAIL fail" >&2
[ "$TOTAL_FAIL" -eq 0 ]

View File

@@ -1,71 +0,0 @@
; persist/durable — a backend whose every op crosses the kernel's IO-suspension
; boundary. Each op performs an IO request {:op "persist/..." :args (...)};
; under the real kernel `perform` suspends the CEK machine and the host (file,
; pg, ipfs-ref) services the request and resumes with the result — so the facet
; code above (log/kv/project/snapshot/compaction) never changes. The TRANSPORT
; is injectable: production passes the kernel's perform; tests pass a mock
; servicer over an in-memory disk. Same request shapes either way, so the whole
; existing facet stack runs unchanged on the mock-durable backend.
; Requires: lib/persist/backend.sx.
; request encoders — the exact payloads the durable backend performs
(define persist/req-append (fn (stream event) {:op "persist/append" :args (list stream event)}))
(define persist/req-read (fn (stream) {:op "persist/read" :args (list stream)}))
(define persist/req-last-seq (fn (stream) {:op "persist/last-seq" :args (list stream)}))
(define persist/req-streams (fn () {:op "persist/streams" :args (list)}))
(define persist/req-truncate (fn (stream n) {:op "persist/truncate" :args (list stream n)}))
(define persist/req-kv-get (fn (key) {:op "persist/kv-get" :args (list key)}))
(define persist/req-kv-put (fn (key val) {:op "persist/kv-put" :args (list key val)}))
(define persist/req-kv-delete (fn (key) {:op "persist/kv-delete" :args (list key)}))
(define persist/req-kv-has? (fn (key) {:op "persist/kv-has?" :args (list key)}))
(define persist/req-kv-keys (fn () {:op "persist/kv-keys" :args (list)}))
; a backend parameterized over a transport (req -> response)
(define persist/io-backend (fn (transport) {:truncate-through (fn (stream n) (transport (persist/req-truncate stream n))) :kv-keys (fn () (transport (persist/req-kv-keys))) :read (fn (stream) (transport (persist/req-read stream))) :kv-has? (fn (key) (transport (persist/req-kv-has? key))) :last-seq (fn (stream) (transport (persist/req-last-seq stream))) :streams (fn () (transport (persist/req-streams))) :append (fn (stream event) (transport (persist/req-append stream event))) :kv-delete (fn (key) (transport (persist/req-kv-delete key))) :kv-put (fn (key val) (transport (persist/req-kv-put key val))) :kv-get (fn (key) (transport (persist/req-kv-get key)))}))
; production backend — transport is the kernel's perform (suspends; host resumes)
(define
persist/durable-backend
(fn () (persist/io-backend (fn (req) (perform req)))))
; reference host: service one request against a disk (any backend protocol impl).
; This is what a real host plugs into the kernel's IO resolver, and the mock-IO
; harness for tests: it never touches a real disk, just an in-memory backend.
(define
persist/serve
(fn
(disk req)
(let
((op (get req :op)) (args (get req :args)))
(cond
((equal? op "persist/append")
(persist/backend-append disk (first args) (nth args 1)))
((equal? op "persist/read")
(persist/backend-read disk (first args)))
((equal? op "persist/last-seq")
(persist/backend-last-seq disk (first args)))
((equal? op "persist/streams") (persist/backend-streams disk))
((equal? op "persist/truncate")
(persist/backend-truncate disk (first args) (nth args 1)))
((equal? op "persist/kv-get")
(persist/backend-kv-get disk (first args)))
((equal? op "persist/kv-put")
(persist/backend-kv-put disk (first args) (nth args 1)))
((equal? op "persist/kv-delete")
(persist/backend-kv-delete disk (first args)))
((equal? op "persist/kv-has?")
(persist/backend-kv-has? disk (first args)))
((equal? op "persist/kv-keys") (persist/backend-kv-keys disk))
(else (error (str "persist/serve: unknown op " op)))))))
; mock transport: a perform-replacement that services against a disk in-process
(define
persist/mock-transport
(fn (disk) (fn (req) (persist/serve disk req))))
; a durable backend wired to a mock disk — exercises the full io-backend path
; (request-encode -> serve -> disk) with no suspension, so the existing facet
; suite runs against it unchanged.
(define
persist/mock-durable
(fn (disk) (persist/io-backend (persist/mock-transport disk))))

View File

@@ -1,13 +0,0 @@
; persist/event — an event is the unit of the log facet:
; {:stream :seq :type :at :data}
; stream = which append-only stream, seq = 1-based position within it,
; type = event kind, at = caller-supplied timestamp (never a clock here:
; replay must stay pure), data = payload dict.
(define persist/event (fn (stream seq type at data) {:data data :type type :at at :stream stream :seq seq}))
(define persist/event-stream (fn (e) (get e :stream)))
(define persist/event-seq (fn (e) (get e :seq)))
(define persist/event-type (fn (e) (get e :type)))
(define persist/event-at (fn (e) (get e :at)))
(define persist/event-data (fn (e) (get e :data)))

View File

@@ -1,79 +0,0 @@
; persist/examples/acl — a WORKED MIGRATION REFERENCE. A subsystem (acl grants:
; who may access what) currently hand-rolls an in-memory mutable map that loses
; every grant on restart and keeps no audit trail. This shows the same subsystem
; rebuilt on persist. It is the template other subsystem loops copy; it does NOT
; touch the real lib/acl (out of this loop's scope).
;
; BEFORE — hand-rolled, ephemeral, no history, no concurrency safety:
; (define acl-grants {}) ; resource -> principal list (mutable)
; (define acl-grant! (fn (r p) (set! acl-grants (assoc acl-grants r (cons p (get acl-grants r))))))
; (define acl-revoke! (fn (r p) (set! acl-grants (assoc acl-grants r (remove p ...)))))
; (define acl-can? (fn (r p) (contains? (get acl-grants r) p)))
; ;; vanishes on restart; "when/why was X granted?" is unanswerable.
;
; AFTER — on persist. Grants/revokes are EVENTS (history matters), the current
; grant set is a PROJECTION, checks read a materialized VIEW, and the audit trail
; is a time-windowed query. Every fn takes a backend `b`, so the same code runs
; on the in-memory backend today and the durable backend unchanged.
; Requires: lib/persist/log.sx, lib/persist/project.sx, lib/persist/view.sx,
; lib/persist/query.sx.
(define acl/stream (fn (resource) (str "acl/" resource)))
; write side — grant/revoke append events (the history is the source of truth)
(define
acl/grant
(fn
(b resource principal at)
(persist/append b (acl/stream resource) "granted" at {:principal principal})))
(define
acl/revoke
(fn
(b resource principal at)
(persist/append b (acl/stream resource) "revoked" at {:principal principal})))
; fold step: grant adds a principal (once), revoke removes it
(define
acl/step
(fn
(set e)
(let
((p (get (persist/event-data e) :principal)))
(if
(equal? (persist/event-type e) "granted")
(if (contains? set p) set (append set p))
(filter (fn (x) (not (equal? x p))) set)))))
; read side — current grant set + membership check (replays the log)
(define
acl/grants
(fn
(b resource)
(persist/project-fold b (acl/stream resource) acl/step (list))))
(define
acl/can?
(fn (b resource principal) (contains? (acl/grants b resource) principal)))
; materialized view — attach to a hub for O(1) checks that stay current on write
(define
acl/view
(fn
(resource)
(persist/view
(str "acl-current/" resource)
(acl/stream resource)
acl/step
(list))))
(define
acl/can-fast?
(fn
(b resource principal)
(contains? (persist/view-peek b (acl/view resource)) principal)))
; audit — grants/revokes for a resource in a time window (the new capability the
; hand-rolled version could never answer)
(define
acl/audit-window
(fn
(b resource from to)
(persist/read-window b (acl/stream resource) from to)))

View File

@@ -1,55 +0,0 @@
; persist/global — a global commit ordering across streams. Per-stream seqs only
; order within a stream; a unified timeline (e.g. feed's home feed, a global
; audit trail) needs a single order across streams. `persist/gappend` appends to
; the target stream and then records a pointer in a reserved $global index whose
; own seq IS the global commit position. Reading the index in order and
; resolving each pointer yields every event in commit order. This is opt-in:
; streams that don't need global ordering use plain persist/append and never
; touch $global. Determinism: the order is the $global append order, replayed
; identically. Requires: lib/persist/log.sx, lib/persist/catalog.sx.
(define persist/global-stream "$global")
; append with a global commit position. Returns the stored stream event; the
; event's global position is the seq of its pointer in $global.
(define
persist/gappend
(fn
(b stream type at data)
(let
((ev (persist/append b stream type at data)))
(begin (persist/append b persist/global-stream "ref" at {:stream stream :seq (persist/event-seq ev)}) ev))))
; the global index: pointer events in commit order (each pointer's seq = gpos)
(define persist/global-log (fn (b) (persist/read b persist/global-stream)))
; the current global commit position (count of globally-ordered appends)
(define
persist/global-pos
(fn (b) (persist/last-seq b persist/global-stream)))
; resolve a pointer event to the actual stream event it references
(define
persist/resolve-ref
(fn
(b ptr)
(let
((d (persist/event-data ptr)))
(first (persist/read-from b (get d :stream) (get d :seq))))))
; every globally-ordered event, in commit order
(define
persist/read-global
(fn
(b)
(map (fn (ptr) (persist/resolve-ref b ptr)) (persist/global-log b))))
; pointer events at or after a global position (incremental global consumers)
(define
persist/global-from
(fn (b gpos) (persist/read-from b persist/global-stream gpos)))
; fold over all events in global commit order
(define
persist/project-global
(fn (b step seed) (reduce step seed (persist/read-global b))))

View File

@@ -1,28 +0,0 @@
; persist/idempotency — exactly-once append under retries. A command retried
; after a network blip must not append its event twice. The caller supplies an
; idempotency key; the first append for that (stream, key) stores the event and
; remembers the key in the kv facet; a repeat returns the SAME event without
; appending. Because the marker lives in kv, idempotency holds across a restart
; too. Keyed per stream. Requires: lib/persist/log.sx, lib/persist/kv.sx.
(define persist/idem-key (fn (stream key) (str "idem/" stream "/" key)))
; true if an append-once has already been recorded for (stream, key)
(define
persist/seen?
(fn (b stream key) (persist/kv-has? b (persist/idem-key stream key))))
; append at most once per (stream, key). Returns the stored event either way —
; freshly appended on first use, the remembered one on a repeat.
(define
persist/append-once
(fn
(b stream key type at data)
(let
((k (persist/idem-key stream key)))
(if
(persist/kv-has? b k)
(persist/kv-get b k)
(let
((ev (persist/append b stream type at data)))
(begin (persist/kv-put b k ev) ev))))))

View File

@@ -1,44 +0,0 @@
; persist/kv — the kv facet: current-state values, no history. For things
; whose history does NOT matter (stock counts, config, profiles, session
; blobs) and where projections materialize their read models.
; Requires: lib/persist/backend.sx.
(define persist/kv-get (fn (b key) (persist/backend-kv-get b key)))
(define
persist/kv-put
(fn (b key val) (begin (persist/backend-kv-put b key val) val)))
(define persist/kv-delete (fn (b key) (persist/backend-kv-delete b key)))
(define persist/kv-has? (fn (b key) (persist/backend-kv-has? b key)))
(define persist/kv-keys (fn (b) (persist/backend-kv-keys b)))
; get with a default when the key is absent
(define
persist/kv-get-or
(fn
(b key dflt)
(if (persist/kv-has? b key) (persist/kv-get b key) dflt)))
; read-modify-write: apply f to the current value (or dflt if absent), store result
(define
persist/kv-update
(fn
(b key dflt f)
(persist/kv-put b key (f (persist/kv-get-or b key dflt)))))
; compare-and-swap: set key to new ONLY if its current value equals expected.
; Returns new on success, or a conflict value {:conflict true :expected :actual}
; the caller can re-read and retry on. The kv analogue of log append-expect.
(define
persist/kv-cas
(fn
(b key expected new)
(let
((actual (persist/kv-get b key)))
(if (equal? actual expected) (persist/kv-put b key new) {:actual actual :expected expected :conflict true}))))
; create-only: put a value only if the key is absent; conflict if it exists
(define
persist/kv-put-new
(fn
(b key val)
(if (persist/kv-has? b key) {:actual (persist/kv-get b key) :conflict true :reason "exists"} (persist/kv-put b key val))))

View File

@@ -1,43 +0,0 @@
; persist/log — the log facet: append-only event streams. seq is assigned from
; a monotonic per-stream high-water mark (1-based) held by the backend, so it
; keeps climbing even after the log prefix is compacted away. Reads return the
; events currently stored, oldest-first.
; Requires: lib/persist/event.sx, lib/persist/backend.sx.
; logical last seq assigned in a stream (0 if none) — survives compaction
(define
persist/last-seq
(fn (b stream) (persist/backend-last-seq b stream)))
; number of events physically stored in a stream (shrinks on compaction)
(define
persist/count
(fn (b stream) (len (persist/backend-read b stream))))
; append an event, auto-assigning the next seq. Returns the stored event.
(define
persist/append
(fn
(b stream type at data)
(let
((seq (+ 1 (persist/last-seq b stream))))
(let
((ev (persist/event stream seq type at data)))
(begin (persist/backend-append b stream ev) ev)))))
; read all events currently stored in a stream, oldest-first
(define persist/read (fn (b stream) (persist/backend-read b stream)))
; read events with seq >= from
(define
persist/read-from
(fn
(b stream from)
(filter
(fn (e) (>= (persist/event-seq e) from))
(persist/read b stream))))
; drop events with seq <= n (compaction); the seq counter is untouched
(define
persist/truncate
(fn (b stream n) (persist/backend-truncate b stream n)))

View File

@@ -1,30 +0,0 @@
; persist/project — a projection folds a stream's events into a read model.
; A projection state is {:value v :seq s} where s is the last seq folded in,
; so a projection can resume incrementally from where it left off (replay only
; the tail). step : (value event) -> value. Determinism: step must be pure —
; time lives on the event (event-at), never a clock here.
; Requires: lib/persist/event.sx, lib/persist/log.sx.
; fold the tail (events with seq > prior's seq) onto a prior projection state
(define
persist/project-resume
(fn
(b stream step prior)
(let
((tail (persist/read-from b stream (+ 1 (get prior :seq)))))
(reduce (fn (acc e) {:value (step (get acc :value) e) :seq (persist/event-seq e)}) prior tail))))
; project the whole stream from seed
(define
persist/project
(fn (b stream step seed) (persist/project-resume b stream step {:value seed :seq 0})))
(define persist/project-value (fn (p) (get p :value)))
(define persist/project-seq (fn (p) (get p :seq)))
; convenience: project and return just the value
(define
persist/project-fold
(fn
(b stream step seed)
(persist/project-value (persist/project b stream step seed))))

View File

@@ -1,54 +0,0 @@
; persist/query — read-side helpers over a stream: slice by seq range, filter by
; timestamp / type / predicate. Pure reads composed from persist/read, no
; backend changes. The log is bad at ad-hoc relational queries (project into a
; kv read model for those) but these cover the common log scans: an audit window
; by time, a type filter, a since-cursor for incremental consumers.
; Requires: lib/persist/log.sx.
; events with seq in [from, to] inclusive
(define
persist/read-between
(fn
(b stream from to)
(filter
(fn
(e)
(and (>= (persist/event-seq e) from) (<= (persist/event-seq e) to)))
(persist/read b stream))))
; events at or after a timestamp (events carry :at; never a clock here)
(define
persist/read-since
(fn
(b stream at)
(filter (fn (e) (>= (persist/event-at e) at)) (persist/read b stream))))
; events whose :at is in [from, to] inclusive — an audit window
(define
persist/read-window
(fn
(b stream from to)
(filter
(fn
(e)
(and (>= (persist/event-at e) from) (<= (persist/event-at e) to)))
(persist/read b stream))))
; events matching a predicate (e -> truthy)
(define
persist/read-where
(fn (b stream pred) (filter pred (persist/read b stream))))
; events of a given type
(define
persist/read-by-type
(fn
(b stream type)
(filter
(fn (e) (equal? (persist/event-type e) type))
(persist/read b stream))))
; count events matching a predicate
(define
persist/count-where
(fn (b stream pred) (len (persist/read-where b stream pred))))

View File

@@ -1,27 +0,0 @@
{
"suites": {
"event": {"pass": 6, "fail": 0},
"log": {"pass": 9, "fail": 0},
"kv": {"pass": 13, "fail": 0},
"project": {"pass": 9, "fail": 0},
"subscribe": {"pass": 9, "fail": 0},
"concurrency": {"pass": 8, "fail": 0},
"snapshot": {"pass": 11, "fail": 0},
"compaction": {"pass": 11, "fail": 0},
"durable": {"pass": 15, "fail": 0},
"blob": {"pass": 14, "fail": 0},
"view": {"pass": 11, "fail": 0},
"cas": {"pass": 11, "fail": 0},
"catalog": {"pass": 10, "fail": 0},
"query": {"pass": 9, "fail": 0},
"batch": {"pass": 10, "fail": 0},
"upcast": {"pass": 9, "fail": 0},
"idempotency": {"pass": 9, "fail": 0},
"global": {"pass": 11, "fail": 0},
"example-acl": {"pass": 10, "fail": 0},
"recovery": {"pass": 6, "fail": 0}
},
"total_pass": 201,
"total_fail": 0,
"total": 201
}

View File

@@ -1,27 +0,0 @@
# persist Conformance Scoreboard
_Generated by `lib/persist/conformance.sh`_
| Suite | Pass | Fail | Total |
|-------|-----:|-----:|------:|
| event | 6 | 0 | 6 |
| log | 9 | 0 | 9 |
| kv | 13 | 0 | 13 |
| project | 9 | 0 | 9 |
| subscribe | 9 | 0 | 9 |
| concurrency | 8 | 0 | 8 |
| snapshot | 11 | 0 | 11 |
| compaction | 11 | 0 | 11 |
| durable | 15 | 0 | 15 |
| blob | 14 | 0 | 14 |
| view | 11 | 0 | 11 |
| cas | 11 | 0 | 11 |
| catalog | 10 | 0 | 10 |
| query | 9 | 0 | 9 |
| batch | 10 | 0 | 10 |
| upcast | 9 | 0 | 9 |
| idempotency | 9 | 0 | 9 |
| global | 11 | 0 | 11 |
| example-acl | 10 | 0 | 10 |
| recovery | 6 | 0 | 6 |
| **Total** | **201** | **0** | **201** |

View File

@@ -1,40 +0,0 @@
; persist/snapshot — checkpoint a projection so a read model rebuilds as
; snapshot + tail instead of a full replay. A snapshot is just a projection
; state {:value :seq} stored in the kv facet under a namespaced key. The
; headline property (tested both ways): snapshot + tail == full replay. Replay
; is pure — it depends only on the stored snapshot and the log tail, never a
; clock. Requires: lib/persist/project.sx, lib/persist/kv.sx.
(define persist/snapshot-key (fn (name) (str "snapshot/" name)))
; load the stored snapshot for name, or a fresh {:value seed :seq 0} if none
(define
persist/snapshot-load
(fn
(b name seed)
(persist/kv-get-or b (persist/snapshot-key name) {:value seed :seq 0})))
; store a projection state as the snapshot for name; returns the state
(define
persist/snapshot-save
(fn (b name state) (persist/kv-put b (persist/snapshot-key name) state)))
(define
persist/snapshot-exists?
(fn (b name) (persist/kv-has? b (persist/snapshot-key name))))
; replay = snapshot + tail: load the snapshot then fold events after it
(define
persist/replay
(fn
(b stream name step seed)
(persist/project-resume b stream step (persist/snapshot-load b name seed))))
; replay then persist the new snapshot; returns the updated state
(define
persist/checkpoint
(fn
(b stream name step seed)
(let
((state (persist/replay b stream name step seed)))
(begin (persist/snapshot-save b name state) state))))

View File

@@ -1,21 +0,0 @@
; persist/subscribe — a subscription hub wraps a backend with per-stream
; callbacks fired after each append. The canonical use: a callback re-runs a
; projection (or bumps a kv counter) so read models update incrementally on
; write instead of being recomputed on read.
; callback signature: (backend stream event) -> ignored
; Publish goes through the hub; direct persist/append on the backend bypasses
; subscribers by design (bulk loads, replay).
; Requires: lib/persist/log.sx.
(define persist/hub (fn (b) (let ((subs {})) {:subscriber-count (fn (stream) (let ((cs (get subs stream))) (if cs (len cs) 0))) :publish (fn (stream type at data) (let ((ev (persist/append b stream type at data))) (begin (for-each (fn (cb) (cb b stream ev)) (let ((cs (get subs stream))) (if cs cs (list)))) ev))) :subscribe (fn (stream cb) (let ((cur (get subs stream))) (set! subs (assoc subs stream (append (if cur cur (list)) cb))))) :backend b})))
(define persist/hub-backend (fn (h) (get h :backend)))
(define
persist/subscribe
(fn (h stream cb) ((get h :subscribe) stream cb)))
(define
persist/publish
(fn (h stream type at data) ((get h :publish) stream type at data)))
(define
persist/subscriber-count
(fn (h stream) ((get h :subscriber-count) stream)))

View File

@@ -1,122 +0,0 @@
; Extension — atomic batch append: contiguous seqs, transactional all-or-nothing.
(persist-test
"batch assigns contiguous seqs"
(let
((b (persist/open)))
(let
((evs (persist/append-batch b "s" (list (list "a" 0 {}) (list "b" 0 {}) (list "c" 0 {})))))
(list
(persist/event-seq (first evs))
(persist/event-seq (nth evs 2)))))
(list 1 3))
(persist-test
"batch returns events in order"
(let
((b (persist/open)))
(let
((evs (persist/append-batch b "s" (list (list "a" 0 {}) (list "b" 0 {})))))
(list
(persist/event-type (first evs))
(persist/event-type (nth evs 1)))))
(list "a" "b"))
(persist-test
"batch grows the stream by its size"
(let
((b (persist/open)))
(begin
(persist/append-batch
b
"s"
(list
(list "a" 0 {})
(list "b" 0 {})
(list "c" 0 {})))
(persist/count b "s")))
3)
(persist-test
"batch continues an existing stream"
(let
((b (persist/open)))
(begin
(persist/append b "s" "x" 0 {})
(let
((evs (persist/append-batch b "s" (list (list "a" 0 {}) (list "b" 0 {})))))
(persist/event-seq (first evs)))))
2)
(persist-test
"empty batch is a no-op"
(let
((b (persist/open)))
(begin (persist/append-batch b "s" (list)) (persist/count b "s")))
0)
(persist-test
"batch-expect with correct seq commits all"
(let
((b (persist/open)))
(begin
(persist/append-batch-expect
b
"s"
0
(list
(list "a" 0 {})
(list "b" 0 {})))
(persist/count b "s")))
2)
(persist-test
"batch-expect with stale seq writes nothing"
(let
((b (persist/open)))
(begin
(persist/append b "s" "x" 0 {})
(persist/append-batch-expect
b
"s"
0
(list
(list "a" 0 {})
(list "b" 0 {})))
(persist/count b "s")))
1)
(persist-test
"batch-expect stale returns a conflict"
(let
((b (persist/open)))
(begin
(persist/append b "s" "x" 0 {})
(persist/conflict?
(persist/append-batch-expect
b
"s"
0
(list (list "a" 0 {}))))))
true)
(persist-test
"batch data is preserved"
(let
((b (persist/open)))
(begin
(persist/append-batch
b
"order"
(list
(list "placed" 0 {:id 1})
(list "line" 0 {:sku "x"})))
(get
(persist/event-data (nth (persist/read b "order") 1))
:sku)))
"x")
(persist-test
"batch works on the durable backend"
(let
((db (persist/mock-durable (persist/mem-backend))))
(begin
(persist/append-batch
db
"s"
(list
(list "a" 0 {})
(list "b" 0 {})))
(persist/last-seq db "s")))
2)

View File

@@ -1,112 +0,0 @@
; Phase 4 — blob backend: store the ref, never the bytes. Bytes live in a
; separate content-addressed store (mock here).
(persist-test
"blob-ref carries cid"
(persist/blob-cid (persist/blob-ref "c1" 10 "image/png"))
"c1")
(persist-test
"blob-ref carries size"
(persist/blob-size (persist/blob-ref "c1" 10 "image/png"))
10)
(persist-test
"blob-ref carries mime"
(persist/blob-mime (persist/blob-ref "c1" 10 "image/png"))
"image/png")
(persist-test
"blob-ref? true for a ref"
(persist/blob-ref? (persist/blob-ref "c1" 1 "x"))
true)
(persist-test
"blob-ref? false for a plain dict"
(persist/blob-ref? {:n 1})
false)
(persist-test
"store returns a ref, not the bytes"
(let
((blob (persist/mock-blob (persist/mem-backend))))
(persist/blob-ref? (persist/blob-store blob "PNGDATA" "image/png")))
true)
(persist-test
"store records the byte length as size"
(let
((blob (persist/mock-blob (persist/mem-backend))))
(persist/blob-size (persist/blob-store blob "12345" "text/plain")))
5)
(persist-test
"fetch round-trips the bytes via the ref"
(let
((blob (persist/mock-blob (persist/mem-backend))))
(let
((ref (persist/blob-store blob "PAYLOAD" "text/plain")))
(persist/blob-fetch blob ref)))
"PAYLOAD")
(persist-test
"exists? true after store"
(let
((blob (persist/mock-blob (persist/mem-backend))))
(let
((ref (persist/blob-store blob "X" "text/plain")))
(persist/blob-exists? blob ref)))
true)
(persist-test
"content addressing: same bytes dedupe to same cid"
(let
((blob (persist/mock-blob (persist/mem-backend))))
(equal?
(persist/blob-cid (persist/blob-store blob "SAME" "text/plain"))
(persist/blob-cid (persist/blob-store blob "SAME" "text/plain"))))
true)
(persist-test
"different bytes get different cids"
(let
((blob (persist/mock-blob (persist/mem-backend))))
(equal?
(persist/blob-cid (persist/blob-store blob "A" "text/plain"))
(persist/blob-cid (persist/blob-store blob "B" "text/plain"))))
false)
; ---------- the invariant: persist holds the ref, never the bytes ----------
(persist-test
"a blob ref stored in kv is a ref"
(let
((db (persist/mock-durable (persist/mem-backend)))
(blob (persist/mock-blob (persist/mem-backend))))
(begin
(persist/kv-put
db
"avatar"
(persist/blob-store blob "BIGIMAGE" "image/png"))
(persist/blob-ref? (persist/kv-get db "avatar"))))
true)
(persist-test
"the kv value does not contain the bytes"
(let
((db (persist/mock-durable (persist/mem-backend)))
(blob (persist/mock-blob (persist/mem-backend))))
(begin
(persist/kv-put
db
"avatar"
(persist/blob-store blob "BIGIMAGE" "image/png"))
(has-key? (persist/kv-get db "avatar") :bytes)))
false)
(persist-test
"a blob ref stored in the log is a ref, bytes fetched separately"
(let
((db (persist/mock-durable (persist/mem-backend)))
(store (persist/mem-backend)))
(let
((blob (persist/mock-blob store)))
(begin
(persist/append
db
"uploads"
"added"
0
(persist/blob-store blob "FILEBYTES" "application/pdf"))
(let
((ref (persist/event-data (first (persist/read db "uploads")))))
(list (persist/blob-ref? ref) (persist/blob-fetch blob ref))))))
(list true "FILEBYTES"))

View File

@@ -1,96 +0,0 @@
; Extension — kv compare-and-swap: atomic current-state updates. Uses
; persist/conflict? from concurrency.sx.
(persist-test
"cas on absent key with nil expected succeeds"
(let ((b (persist/open))) (persist/kv-cas b "k" nil 1))
1)
(persist-test
"cas with matching expected succeeds"
(let
((b (persist/open)))
(begin
(persist/kv-put b "k" 5)
(persist/kv-cas b "k" 5 6)
(persist/kv-get b "k")))
6)
(persist-test
"cas with stale expected returns a conflict"
(let
((b (persist/open)))
(begin
(persist/kv-put b "k" 5)
(persist/conflict? (persist/kv-cas b "k" 4 6))))
true)
(persist-test
"a conflicting cas does not write"
(let
((b (persist/open)))
(begin
(persist/kv-put b "k" 5)
(persist/kv-cas b "k" 4 6)
(persist/kv-get b "k")))
5)
(persist-test
"cas conflict carries expected and actual"
(let
((b (persist/open)))
(begin
(persist/kv-put b "k" 5)
(let
((r (persist/kv-cas b "k" 4 6)))
(list (persist/conflict-expected r) (persist/conflict-actual r)))))
(list 4 5))
(persist-test
"two cas racers: first wins, second conflicts"
(let
((b (persist/open)))
(begin
(persist/kv-put b "stock" 10)
(persist/kv-cas b "stock" 10 9)
(persist/conflict? (persist/kv-cas b "stock" 10 9))))
true)
(persist-test
"retry after cas conflict succeeds"
(let
((b (persist/open)))
(begin
(persist/kv-put b "stock" 10)
(persist/kv-cas b "stock" 10 9)
(let
((r (persist/kv-cas b "stock" 10 9)))
(if
(persist/conflict? r)
(persist/kv-cas b "stock" (persist/conflict-actual r) 8)
r))))
8)
(persist-test
"put-new on absent key succeeds"
(let ((b (persist/open))) (persist/kv-put-new b "k" 1))
1)
(persist-test
"put-new on existing key conflicts"
(let
((b (persist/open)))
(begin
(persist/kv-put b "k" 1)
(persist/conflict? (persist/kv-put-new b "k" 2))))
true)
(persist-test
"put-new does not overwrite"
(let
((b (persist/open)))
(begin
(persist/kv-put b "k" 1)
(persist/kv-put-new b "k" 2)
(persist/kv-get b "k")))
1)
(persist-test
"cas works on the durable backend"
(let
((db (persist/mock-durable (persist/mem-backend))))
(begin
(persist/kv-put db "k" 1)
(persist/kv-cas db "k" 1 2)
(persist/kv-get db "k")))
2)

View File

@@ -1,86 +0,0 @@
; Extension — stream catalog: enumerate streams, count, existence, totals.
(persist-test
"empty backend has no streams"
(persist/stream-count (persist/open))
0)
(persist-test
"stream-exists? false when absent"
(persist/stream-exists? (persist/open) "orders")
false)
(persist-test
"append registers a stream"
(let
((b (persist/open)))
(begin
(persist/append b "orders" "x" 0 {})
(persist/stream-exists? b "orders")))
true)
(persist-test
"stream-count counts distinct streams"
(let
((b (persist/open)))
(begin
(persist/append b "a" "x" 0 {})
(persist/append b "b" "x" 0 {})
(persist/append b "a" "x" 0 {})
(persist/stream-count b)))
2)
(persist-test
"compacted-away stream still lists"
(let
((b (persist/open)))
(begin
(persist/append b "a" "x" 0 {})
(persist/checkpoint b "a" "snap" (fn (acc e) acc) 0)
(persist/truncate b "a" 1)
(list (persist/count b "a") (persist/stream-exists? b "a"))))
(list 0 true))
(persist-test
"kv-only backend lists no streams"
(let
((b (persist/open)))
(begin (persist/kv-put b "k" 1) (persist/stream-count b)))
0)
(persist-test
"total-events sums high-water marks"
(let
((b (persist/open)))
(begin
(persist/append b "a" "x" 0 {})
(persist/append b "a" "x" 0 {})
(persist/append b "b" "x" 0 {})
(persist/total-events b)))
3)
(persist-test
"total-events counts compacted events too"
(let
((b (persist/open)))
(begin
(persist/append b "a" "x" 0 {})
(persist/append b "a" "x" 0 {})
(persist/checkpoint b "a" "snap" (fn (acc e) acc) 0)
(persist/truncate b "a" 2)
(persist/total-events b)))
2)
(persist-test
"catalog works on the durable backend"
(let
((db (persist/mock-durable (persist/mem-backend))))
(begin
(persist/append db "a" "x" 0 {})
(persist/append db "b" "x" 0 {})
(persist/stream-count db)))
2)
(persist-test
"catalog survives restart"
(let
((disk (persist/mem-backend)))
(begin
(let
((db (persist/mock-durable disk)))
(begin
(persist/append db "a" "x" 0 {})
(persist/append db "b" "x" 0 {})))
(persist/stream-count (persist/mock-durable disk))))
2)

View File

@@ -1,124 +0,0 @@
; Phase 3 — compaction: drop the snapshotted prefix; replay determinism holds.
(define comp-count (fn (acc e) (+ acc 1)))
(persist-test
"uncompacted counts events since snapshot"
(let
((b (persist/open)))
(begin
(persist/append b "s" "x" 0 {})
(persist/append b "s" "x" 0 {})
(persist/uncompacted b "s" "snap" 0)))
2)
(persist-test
"should-compact? false below threshold"
(let
((b (persist/open)))
(begin
(persist/append b "s" "x" 0 {})
(persist/should-compact? b "s" "snap" 3 0)))
false)
(persist-test
"should-compact? true at threshold"
(let
((b (persist/open)))
(begin
(persist/append b "s" "x" 0 {})
(persist/append b "s" "x" 0 {})
(persist/append b "s" "x" 0 {})
(persist/should-compact? b "s" "snap" 3 0)))
true)
(persist-test
"compact truncates the snapshotted prefix"
(let
((b (persist/open)))
(begin
(persist/append b "s" "x" 0 {})
(persist/append b "s" "x" 0 {})
(persist/append b "s" "x" 0 {})
(persist/compact b "s" "snap" comp-count 0)
(persist/count b "s")))
0)
(persist-test
"compact preserves logical last-seq"
(let
((b (persist/open)))
(begin
(persist/append b "s" "x" 0 {})
(persist/append b "s" "x" 0 {})
(persist/compact b "s" "snap" comp-count 0)
(persist/last-seq b "s")))
2)
(persist-test
"append after compaction continues the seq"
(let
((b (persist/open)))
(begin
(persist/append b "s" "x" 0 {})
(persist/append b "s" "x" 0 {})
(persist/compact b "s" "snap" comp-count 0)
(persist/event-seq (persist/append b "s" "x" 0 {}))))
3)
(persist-test
"replay after compaction == full count before compaction"
(let
((b (persist/open)))
(begin
(persist/append b "s" "x" 0 {})
(persist/append b "s" "x" 0 {})
(persist/append b "s" "x" 0 {})
(persist/compact b "s" "snap" comp-count 0)
(persist/append b "s" "x" 0 {})
(persist/append b "s" "x" 0 {})
(persist/project-value
(persist/replay b "s" "snap" comp-count 0))))
5)
(persist-test
"determinism: post-compaction replay value equals uncompacted full replay"
(let
((b (persist/open)) (c (persist/open)))
(begin
(persist/append b "s" "x" 0 {})
(persist/append b "s" "x" 0 {})
(persist/append b "s" "x" 0 {})
(persist/append c "s" "x" 0 {})
(persist/append c "s" "x" 0 {})
(persist/append c "s" "x" 0 {})
(persist/compact b "s" "snap" comp-count 0)
(persist/append b "s" "x" 0 {})
(persist/append c "s" "x" 0 {})
(equal?
(persist/project-value
(persist/replay b "s" "snap" comp-count 0))
(persist/project-fold c "s" comp-count 0))))
true)
(persist-test
"maybe-compact below threshold does not truncate"
(let
((b (persist/open)))
(begin
(persist/append b "s" "x" 0 {})
(persist/maybe-compact b "s" "snap" comp-count 0 5)
(persist/count b "s")))
1)
(persist-test
"maybe-compact at threshold truncates"
(let
((b (persist/open)))
(begin
(persist/append b "s" "x" 0 {})
(persist/append b "s" "x" 0 {})
(persist/maybe-compact b "s" "snap" comp-count 0 2)
(persist/count b "s")))
0)
(persist-test
"compact is idempotent on an empty tail"
(let
((b (persist/open)))
(begin
(persist/append b "s" "x" 0 {})
(persist/compact b "s" "snap" comp-count 0)
(persist/project-value
(persist/compact b "s" "snap" comp-count 0))))
1)

View File

@@ -1,96 +0,0 @@
; Phase 2 — optimistic concurrency: conflict is a real result, not a crash.
(persist-test
"append-expect 0 on empty stream succeeds"
(persist/event-seq
(persist/append-expect
(persist/open)
"s"
0
"x"
0
{}))
1)
(persist-test
"append-expect with correct seq succeeds"
(let
((b (persist/open)))
(begin
(persist/append b "s" "x" 0 {})
(persist/event-seq
(persist/append-expect b "s" 1 "x" 0 {}))))
2)
(persist-test
"append-expect with stale seq returns a conflict"
(let
((b (persist/open)))
(begin
(persist/append b "s" "x" 0 {})
(persist/append b "s" "x" 0 {})
(persist/conflict?
(persist/append-expect b "s" 1 "x" 0 {}))))
true)
(persist-test
"a successful append is not a conflict"
(persist/conflict?
(persist/append-expect
(persist/open)
"s"
0
"x"
0
{}))
false)
(persist-test
"conflict carries expected and actual"
(let
((b (persist/open)))
(begin
(persist/append b "s" "x" 0 {})
(persist/append b "s" "x" 0 {})
(let
((r (persist/append-expect b "s" 0 "x" 0 {})))
(list (persist/conflict-expected r) (persist/conflict-actual r)))))
(list 0 2))
(persist-test
"a conflicting append does not write"
(let
((b (persist/open)))
(begin
(persist/append b "s" "x" 0 {})
(persist/append-expect b "s" 0 "x" 0 {})
(persist/count b "s")))
1)
(persist-test
"two writers: first wins, second conflicts"
(let
((b (persist/open)))
(let
((seen (persist/last-seq b "s")))
(begin
(persist/append-expect b "s" seen "x" 0 {:who "A"})
(persist/conflict?
(persist/append-expect b "s" seen "x" 0 {:who "B"})))))
true)
(persist-test
"retry after conflict succeeds"
(let
((b (persist/open)))
(let
((seen (persist/last-seq b "s")))
(begin
(persist/append-expect b "s" seen "x" 0 {:who "A"})
(let
((r (persist/append-expect b "s" seen "x" 0 {:who "B"})))
(if
(persist/conflict? r)
(persist/event-seq
(persist/append-expect
b
"s"
(persist/conflict-actual r)
"x"
0
{:who "B"}))
(persist/event-seq r))))))
2)

View File

@@ -1,163 +0,0 @@
; Phase 4 — durable backend over the IO-suspension boundary, tested with a mock
; transport (the mock-IO harness for the durable protocol). The whole facet
; stack must run unchanged on mock-durable, and a "crash/restart" (drop the
; backend, keep the disk) must recover state by replay.
(define dur-count (fn (acc e) (+ acc 1)))
; ---------- request encoders ----------
(persist-test
"req-append encodes op + args"
(persist/req-append "s" {:k 1})
{:op "persist/append" :args (list "s" {:k 1})})
(persist-test
"req-kv-put encodes op + args"
(persist/req-kv-put "k" 7)
{:op "persist/kv-put" :args (list "k" 7)})
; ---------- serve round-trips against a disk ----------
(persist-test
"serve append then serve read"
(let
((disk (persist/mem-backend)))
(begin
(persist/serve
disk
(persist/req-append
"s"
(persist/event "s" 1 "x" 0 {:n 1})))
(get
(persist/event-data
(first (persist/serve disk (persist/req-read "s"))))
:n)))
1)
(persist-test
"serve kv-put then kv-get"
(let
((disk (persist/mem-backend)))
(begin
(persist/serve disk (persist/req-kv-put "k" 42))
(persist/serve disk (persist/req-kv-get "k"))))
42)
(persist-test
"serve unknown op is a clear error"
(let
((disk (persist/mem-backend)))
(guard (e (true "errored")) (persist/serve disk {:op "persist/bogus" :args (list)})))
"errored")
; ---------- full facet stack on mock-durable ----------
(persist-test
"log facet works on mock-durable"
(let
((db (persist/mock-durable (persist/mem-backend))))
(begin
(persist/append db "s" "x" 0 {})
(persist/append db "s" "x" 0 {})
(persist/count db "s")))
2)
(persist-test
"seq assignment works on mock-durable"
(let
((db (persist/mock-durable (persist/mem-backend))))
(begin
(persist/append db "s" "x" 0 {})
(persist/event-seq (persist/append db "s" "x" 0 {}))))
2)
(persist-test
"kv facet works on mock-durable"
(let
((db (persist/mock-durable (persist/mem-backend))))
(begin (persist/kv-put db "k" 5) (persist/kv-get db "k")))
5)
(persist-test
"projection works on mock-durable"
(let
((db (persist/mock-durable (persist/mem-backend))))
(begin
(persist/append db "s" "x" 0 {})
(persist/append db "s" "x" 0 {})
(persist/append db "s" "x" 0 {})
(persist/project-fold db "s" dur-count 0)))
3)
(persist-test
"snapshot + replay work on mock-durable"
(let
((db (persist/mock-durable (persist/mem-backend))))
(begin
(persist/append db "s" "x" 0 {})
(persist/append db "s" "x" 0 {})
(persist/checkpoint db "s" "snap" dur-count 0)
(persist/append db "s" "x" 0 {})
(persist/project-value
(persist/replay db "s" "snap" dur-count 0))))
3)
(persist-test
"compaction works on mock-durable"
(let
((db (persist/mock-durable (persist/mem-backend))))
(begin
(persist/append db "s" "x" 0 {})
(persist/append db "s" "x" 0 {})
(persist/compact db "s" "snap" dur-count 0)
(list (persist/count db "s") (persist/last-seq db "s"))))
(list 0 2))
; ---------- crash / restart replay ----------
(persist-test
"restart recovers log state from the disk"
(let
((disk (persist/mem-backend)))
(begin
(let
((db (persist/mock-durable disk)))
(begin
(persist/append db "s" "x" 0 {})
(persist/append db "s" "x" 0 {})))
(let
((db2 (persist/mock-durable disk)))
(persist/project-fold db2 "s" dur-count 0))))
2)
(persist-test
"restart continues the seq counter"
(let
((disk (persist/mem-backend)))
(begin
(let
((db (persist/mock-durable disk)))
(begin
(persist/append db "s" "x" 0 {})
(persist/append db "s" "x" 0 {})))
(let
((db2 (persist/mock-durable disk)))
(persist/event-seq (persist/append db2 "s" "x" 0 {})))))
3)
(persist-test
"restart recovers a kv value"
(let
((disk (persist/mem-backend)))
(begin
(let
((db (persist/mock-durable disk)))
(persist/kv-put db "cfg" "on"))
(let ((db2 (persist/mock-durable disk))) (persist/kv-get db2 "cfg"))))
"on")
(persist-test
"restart from snapshot equals full replay"
(let
((disk (persist/mem-backend)))
(begin
(let
((db (persist/mock-durable disk)))
(begin
(persist/append db "s" "x" 0 {})
(persist/append db "s" "x" 0 {})
(persist/checkpoint db "s" "snap" dur-count 0)
(persist/append db "s" "x" 0 {})))
(let
((db2 (persist/mock-durable disk)))
(equal?
(persist/project-value
(persist/replay db2 "s" "snap" dur-count 0))
(persist/project-fold db2 "s" dur-count 0)))))
true)

View File

@@ -1,30 +0,0 @@
; Phase 1 — event record accessors. Uses the persist-test harness
; (persist-test name got expected) provided by conformance.sh.
(persist-test
"event-stream"
(persist/event-stream
(persist/event "s" 1 "t" 0 {}))
"s")
(persist-test
"event-seq"
(persist/event-seq (persist/event "s" 3 "t" 0 {}))
3)
(persist-test
"event-type"
(persist/event-type
(persist/event "s" 1 "create" 0 {}))
"create")
(persist-test
"event-at"
(persist/event-at (persist/event "s" 1 "t" 42 {}))
42)
(persist-test
"event-data"
(persist/event-data
(persist/event "s" 1 "t" 0 {:x 9}))
{:x 9})
(persist-test
"event is a dict with all fields"
(len (keys (persist/event "s" 1 "t" 0 {})))
5)

View File

@@ -1,104 +0,0 @@
; Reference migration — acl grants on persist. Proves the AFTER behaviour,
; including the capabilities the hand-rolled BEFORE version could not provide
; (durability across restart + an audit trail).
(persist-test
"grant then can?"
(let
((b (persist/open)))
(begin
(acl/grant b "doc-1" "alice" 0)
(acl/can? b "doc-1" "alice")))
true)
(persist-test
"no grant means no access"
(acl/can? (persist/open) "doc-1" "alice")
false)
(persist-test
"revoke removes access"
(let
((b (persist/open)))
(begin
(acl/grant b "doc-1" "alice" 0)
(acl/revoke b "doc-1" "alice" 1)
(acl/can? b "doc-1" "alice")))
false)
(persist-test
"multiple principals tracked independently"
(let
((b (persist/open)))
(begin
(acl/grant b "doc-1" "alice" 0)
(acl/grant b "doc-1" "bob" 1)
(acl/revoke b "doc-1" "alice" 2)
(list (acl/can? b "doc-1" "alice") (acl/can? b "doc-1" "bob"))))
(list false true))
(persist-test
"granting twice is idempotent in the set"
(let
((b (persist/open)))
(begin
(acl/grant b "doc-1" "alice" 0)
(acl/grant b "doc-1" "alice" 1)
(len (acl/grants b "doc-1"))))
1)
(persist-test
"grants on different resources are isolated"
(let
((b (persist/open)))
(begin
(acl/grant b "doc-1" "alice" 0)
(acl/grant b "doc-2" "bob" 0)
(list (acl/can? b "doc-1" "bob") (acl/can? b "doc-2" "bob"))))
(list false true))
(persist-test
"audit window answers when-was-it-granted (new capability)"
(let
((b (persist/open)))
(begin
(acl/grant b "doc-1" "alice" 100)
(acl/revoke b "doc-1" "alice" 200)
(acl/grant b "doc-1" "bob" 300)
(len (acl/audit-window b "doc-1" 150 300))))
2)
(persist-test
"materialized view stays current on publish"
(let
((b (persist/open)))
(let
((h (persist/view-attach (persist/hub b) (acl/view "doc-1"))))
(begin
(persist/publish
h
(acl/stream "doc-1")
"granted"
0
{:principal "alice"})
(acl/can-fast? b "doc-1" "alice"))))
true)
(persist-test
"grants survive restart on the durable backend (the headline win)"
(let
((disk (persist/mem-backend)))
(begin
(let
((db (persist/mock-durable disk)))
(begin
(acl/grant db "doc-1" "alice" 0)
(acl/grant db "doc-1" "bob" 1)))
(let
((db2 (persist/mock-durable disk)))
(list (acl/can? db2 "doc-1" "alice") (acl/can? db2 "doc-1" "bob")))))
(list true true))
(persist-test
"revoke before restart is still revoked after"
(let
((disk (persist/mem-backend)))
(begin
(let
((db (persist/mock-durable disk)))
(begin
(acl/grant db "doc-1" "alice" 0)
(acl/revoke db "doc-1" "alice" 1)))
(acl/can? (persist/mock-durable disk) "doc-1" "alice")))
false)

View File

@@ -1,123 +0,0 @@
; Extension — global commit ordering across streams.
(persist-test
"gappend returns the stream event with its local seq"
(let
((b (persist/open)))
(persist/event-seq
(persist/gappend b "orders" "placed" 0 {})))
1)
(persist-test
"global-pos advances per gappend regardless of stream"
(let
((b (persist/open)))
(begin
(persist/gappend b "orders" "placed" 0 {})
(persist/gappend b "users" "joined" 0 {})
(persist/gappend b "orders" "placed" 0 {})
(persist/global-pos b)))
3)
(persist-test
"read-global returns events in commit order across streams"
(let
((b (persist/open)))
(begin
(persist/gappend b "orders" "placed" 0 {:n 1})
(persist/gappend b "users" "joined" 0 {:n 2})
(persist/gappend b "orders" "placed" 0 {:n 3})
(let
((g (persist/read-global b)))
(list
(get (persist/event-data (nth g 0)) :n)
(get (persist/event-data (nth g 1)) :n)
(get (persist/event-data (nth g 2)) :n)))))
(list 1 2 3))
(persist-test
"read-global resolves to the right streams"
(let
((b (persist/open)))
(begin
(persist/gappend b "orders" "placed" 0 {})
(persist/gappend b "users" "joined" 0 {})
(let
((g (persist/read-global b)))
(list
(persist/event-stream (nth g 0))
(persist/event-stream (nth g 1))))))
(list "orders" "users"))
(persist-test
"project-global folds across all streams in order"
(let
((b (persist/open)))
(begin
(persist/gappend b "a" "x" 0 {:v 10})
(persist/gappend b "b" "x" 0 {:v 20})
(persist/gappend b "a" "x" 0 {:v 30})
(persist/project-global
b
(fn (acc e) (+ acc (get (persist/event-data e) :v)))
0)))
60)
(persist-test
"global index is hidden from the public catalog"
(let
((b (persist/open)))
(begin
(persist/gappend b "orders" "placed" 0 {})
(persist/gappend b "users" "joined" 0 {})
(list (persist/stream-count b) (persist/stream-exists? b "$global"))))
(list 2 false))
(persist-test
"streams-all reveals the reserved index"
(let
((b (persist/open)))
(begin
(persist/gappend b "orders" "placed" 0 {})
(contains? (persist/streams-all b) "$global")))
true)
(persist-test
"global-from gives pointers at or after a position"
(let
((b (persist/open)))
(begin
(persist/gappend b "a" "x" 0 {})
(persist/gappend b "a" "x" 0 {})
(persist/gappend b "a" "x" 0 {})
(len (persist/global-from b 2))))
2)
(persist-test
"plain append does not touch the global index"
(let
((b (persist/open)))
(begin
(persist/append b "orders" "placed" 0 {})
(persist/gappend b "orders" "placed" 0 {})
(persist/global-pos b)))
1)
(persist-test
"global ordering works on the durable backend"
(let
((db (persist/mock-durable (persist/mem-backend))))
(begin
(persist/gappend db "a" "x" 0 {:v 1})
(persist/gappend db "b" "x" 0 {:v 2})
(persist/project-global
db
(fn (acc e) (+ acc (get (persist/event-data e) :v)))
0)))
3)
(persist-test
"global order survives restart (determinism)"
(let
((disk (persist/mem-backend)))
(begin
(let
((db (persist/mock-durable disk)))
(begin
(persist/gappend db "a" "x" 0 {:v 1})
(persist/gappend db "b" "x" 0 {:v 2})))
(persist/project-global
(persist/mock-durable disk)
(fn (acc e) (+ acc (get (persist/event-data e) :v)))
0)))
3)

View File

@@ -1,92 +0,0 @@
; Extension — exactly-once append under retries.
(persist-test
"seen? false before first append"
(persist/seen? (persist/open) "orders" "cmd-1")
false)
(persist-test
"append-once appends on first use"
(let
((b (persist/open)))
(begin
(persist/append-once b "orders" "cmd-1" "placed" 0 {})
(persist/count b "orders")))
1)
(persist-test
"seen? true after first append"
(let
((b (persist/open)))
(begin
(persist/append-once b "orders" "cmd-1" "placed" 0 {})
(persist/seen? b "orders" "cmd-1")))
true)
(persist-test
"repeat with same key does not append again"
(let
((b (persist/open)))
(begin
(persist/append-once b "orders" "cmd-1" "placed" 0 {})
(persist/append-once b "orders" "cmd-1" "placed" 0 {})
(persist/append-once b "orders" "cmd-1" "placed" 0 {})
(persist/count b "orders")))
1)
(persist-test
"repeat returns the same event (same seq)"
(let
((b (persist/open)))
(let
((e1 (persist/append-once b "orders" "cmd-1" "placed" 0 {})))
(persist/event-seq
(persist/append-once b "orders" "cmd-1" "placed" 0 {}))))
1)
(persist-test
"different keys append separately"
(let
((b (persist/open)))
(begin
(persist/append-once b "orders" "cmd-1" "placed" 0 {})
(persist/append-once b "orders" "cmd-2" "placed" 0 {})
(persist/count b "orders")))
2)
(persist-test
"idempotency is per-stream"
(let
((b (persist/open)))
(begin
(persist/append-once b "a" "cmd-1" "x" 0 {})
(persist/append-once b "b" "cmd-1" "x" 0 {})
(list (persist/count b "a") (persist/count b "b"))))
(list 1 1))
(persist-test
"stored data is preserved on first append"
(let
((b (persist/open)))
(get
(persist/event-data
(persist/append-once b "s" "k" "x" 0 {:n 9}))
:n))
9)
(persist-test
"idempotency survives restart on the durable backend"
(let
((disk (persist/mem-backend)))
(begin
(persist/append-once
(persist/mock-durable disk)
"orders"
"cmd-1"
"placed"
0
{})
(let
((db2 (persist/mock-durable disk)))
(begin
(persist/append-once
db2
"orders"
"cmd-1"
"placed"
0
{})
(persist/count db2 "orders")))))
1)

View File

@@ -1,86 +0,0 @@
; Phase 1 — kv facet: get/put/delete/has?/keys, get-or, update.
(persist-test "absent key reads nil" (persist/kv-get (persist/open) "x") nil)
(persist-test
"has? false when absent"
(persist/kv-has? (persist/open) "x")
false)
(persist-test
"put then get"
(let
((b (persist/open)))
(begin (persist/kv-put b "x" 7) (persist/kv-get b "x")))
7)
(persist-test
"put returns value"
(let ((b (persist/open))) (persist/kv-put b "x" 9))
9)
(persist-test
"has? true after put"
(let
((b (persist/open)))
(begin (persist/kv-put b "x" 1) (persist/kv-has? b "x")))
true)
(persist-test
"put overwrites"
(let
((b (persist/open)))
(begin
(persist/kv-put b "x" 1)
(persist/kv-put b "x" 2)
(persist/kv-get b "x")))
2)
(persist-test
"delete removes key"
(let
((b (persist/open)))
(begin
(persist/kv-put b "x" 1)
(persist/kv-delete b "x")
(persist/kv-has? b "x")))
false)
(persist-test
"delete then get is nil"
(let
((b (persist/open)))
(begin
(persist/kv-put b "x" 1)
(persist/kv-delete b "x")
(persist/kv-get b "x")))
nil)
(persist-test
"keys lists stored keys"
(let
((b (persist/open)))
(begin
(persist/kv-put b "a" 1)
(persist/kv-put b "b" 2)
(len (persist/kv-keys b))))
2)
(persist-test
"get-or returns default when absent"
(persist/kv-get-or (persist/open) "x" 99)
99)
(persist-test
"get-or returns value when present"
(let
((b (persist/open)))
(begin
(persist/kv-put b "x" 5)
(persist/kv-get-or b "x" 99)))
5)
(persist-test
"kv-update applies fn over default"
(let
((b (persist/open)))
(begin
(persist/kv-update b "n" 0 (fn (v) (+ v 1)))
(persist/kv-update b "n" 0 (fn (v) (+ v 1)))
(persist/kv-get b "n")))
2)
(persist-test
"kv facet does not touch log"
(let
((b (persist/open)))
(begin (persist/kv-put b "x" 1) (persist/count b "x")))
0)

View File

@@ -1,81 +0,0 @@
; Phase 1 — log facet: append/read/read-from, sequential seq, stream isolation.
; Note: map returns an array-backed list not equal? to a (list ...) literal,
; so assertions build their compared list with list/nth, not map.
(persist-test
"empty stream reads empty"
(len (persist/read (persist/open) "orders"))
0)
(persist-test
"last-seq empty is 0"
(persist/last-seq (persist/open) "orders")
0)
(persist-test
"append returns event with seq 1"
(persist/event-seq
(persist/append (persist/open) "orders" "placed" 0 {:id 1}))
1)
(persist-test
"append assigns sequential seqs"
(let
((b (persist/open)))
(begin
(persist/append b "orders" "placed" 0 {})
(persist/append b "orders" "placed" 1 {})
(persist/event-seq
(persist/append b "orders" "placed" 2 {}))))
3)
(persist-test
"read returns events oldest-first"
(let
((b (persist/open)))
(begin
(persist/append b "s" "a" 0 {:n 1})
(persist/append b "s" "b" 0 {:n 2})
(let
((es (persist/read b "s")))
(list
(get (persist/event-data (nth es 0)) :n)
(get (persist/event-data (nth es 1)) :n)))))
(list 1 2))
(persist-test
"count tracks appends"
(let
((b (persist/open)))
(begin
(persist/append b "s" "a" 0 {})
(persist/append b "s" "a" 0 {})
(persist/count b "s")))
2)
(persist-test
"streams are isolated"
(let
((b (persist/open)))
(begin
(persist/append b "s1" "a" 0 {})
(persist/append b "s2" "a" 0 {})
(persist/append b "s2" "a" 0 {})
(list (persist/count b "s1") (persist/count b "s2"))))
(list 1 2))
(persist-test
"read-from filters by seq"
(let
((b (persist/open)))
(begin
(persist/append b "s" "a" 0 {})
(persist/append b "s" "a" 0 {})
(persist/append b "s" "a" 0 {})
(let
((es (persist/read-from b "s" 2)))
(list
(persist/event-seq (nth es 0))
(persist/event-seq (nth es 1))))))
(list 2 3))
(persist-test
"read-from past end is empty"
(let
((b (persist/open)))
(begin
(persist/append b "s" "a" 0 {})
(len (persist/read-from b "s" 5))))
0)

View File

@@ -1,115 +0,0 @@
; Phase 2 — projections: fold a stream into a read model, resume incrementally.
(persist-test
"project empty stream returns seed value"
(persist/project-fold
(persist/open)
"s"
(fn (acc e) (+ acc 1))
0)
0)
(persist-test
"project empty stream seq is 0"
(persist/project-seq
(persist/project (persist/open) "s" (fn (a e) a) 0))
0)
(persist-test
"project counts events"
(let
((b (persist/open)))
(begin
(persist/append b "s" "x" 0 {})
(persist/append b "s" "x" 0 {})
(persist/append b "s" "x" 0 {})
(persist/project-fold
b
"s"
(fn (acc e) (+ acc 1))
0)))
3)
(persist-test
"project sums event data"
(let
((b (persist/open)))
(begin
(persist/append b "ledger" "credit" 0 {:amt 10})
(persist/append b "ledger" "credit" 1 {:amt 5})
(persist/append b "ledger" "debit" 2 {:amt 3})
(persist/project-fold
b
"ledger"
(fn
(bal e)
(if
(equal? (persist/event-type e) "credit")
(+ bal (get (persist/event-data e) :amt))
(- bal (get (persist/event-data e) :amt))))
0)))
12)
(persist-test
"project tracks last seq"
(let
((b (persist/open)))
(begin
(persist/append b "s" "x" 0 {})
(persist/append b "s" "x" 0 {})
(persist/project-seq (persist/project b "s" (fn (a e) a) 0))))
2)
(persist-test
"resume folds only the tail"
(let
((b (persist/open)))
(begin
(persist/append b "s" "x" 0 {})
(persist/append b "s" "x" 0 {})
(let
((p1 (persist/project b "s" (fn (acc e) (+ acc 1)) 0)))
(begin
(persist/append b "s" "x" 0 {})
(persist/project-value
(persist/project-resume
b
"s"
(fn (acc e) (+ acc 1))
p1))))))
3)
(persist-test
"resume with no new events is a no-op"
(let
((b (persist/open)))
(begin
(persist/append b "s" "x" 0 {})
(let
((p1 (persist/project b "s" (fn (acc e) (+ acc 1)) 0)))
(persist/project-value
(persist/project-resume b "s" (fn (acc e) (+ acc 1)) p1)))))
1)
(persist-test
"resume advances seq"
(let
((b (persist/open)))
(begin
(persist/append b "s" "x" 0 {})
(let
((p1 (persist/project b "s" (fn (a e) a) 0)))
(begin
(persist/append b "s" "x" 0 {})
(persist/append b "s" "x" 0 {})
(persist/project-seq
(persist/project-resume b "s" (fn (a e) a) p1))))))
3)
(persist-test
"full project equals seed-resume from zero"
(let
((b (persist/open)))
(begin
(persist/append b "s" "x" 0 {})
(persist/append b "s" "x" 0 {})
(equal?
(persist/project b "s" (fn (acc e) (+ acc 1)) 0)
(persist/project-resume
b
"s"
(fn (acc e) (+ acc 1))
{:value 0 :seq 0}))))
true)

View File

@@ -1,101 +0,0 @@
; Extension — read-side query helpers. Assertions count / index, not map vs list.
(define q-seqs (fn (es) (map persist/event-seq es)))
(persist-test
"read-between slices a seq range"
(let
((b (persist/open)))
(begin
(persist/append b "s" "x" 0 {})
(persist/append b "s" "x" 0 {})
(persist/append b "s" "x" 0 {})
(persist/append b "s" "x" 0 {})
(let
((es (persist/read-between b "s" 2 3)))
(list
(len es)
(persist/event-seq (first es))
(persist/event-seq (nth es 1))))))
(list 2 2 3))
(persist-test
"read-between is inclusive of endpoints"
(let
((b (persist/open)))
(begin
(persist/append b "s" "x" 0 {})
(persist/append b "s" "x" 0 {})
(persist/append b "s" "x" 0 {})
(len (persist/read-between b "s" 1 3))))
3)
(persist-test
"read-since filters by timestamp"
(let
((b (persist/open)))
(begin
(persist/append b "s" "x" 100 {})
(persist/append b "s" "x" 200 {})
(persist/append b "s" "x" 300 {})
(len (persist/read-since b "s" 200))))
2)
(persist-test
"read-window is an inclusive time range"
(let
((b (persist/open)))
(begin
(persist/append b "s" "x" 100 {})
(persist/append b "s" "x" 200 {})
(persist/append b "s" "x" 300 {})
(persist/append b "s" "x" 400 {})
(len (persist/read-window b "s" 200 300))))
2)
(persist-test
"read-by-type filters by event type"
(let
((b (persist/open)))
(begin
(persist/append b "s" "created" 0 {})
(persist/append b "s" "updated" 0 {})
(persist/append b "s" "created" 0 {})
(len (persist/read-by-type b "s" "created"))))
2)
(persist-test
"read-where filters by predicate over data"
(let
((b (persist/open)))
(begin
(persist/append b "s" "x" 0 {:amt 5})
(persist/append b "s" "x" 0 {:amt 15})
(persist/append b "s" "x" 0 {:amt 25})
(len
(persist/read-where
b
"s"
(fn (e) (> (get (persist/event-data e) :amt) 10))))))
2)
(persist-test
"count-where counts matches"
(let
((b (persist/open)))
(begin
(persist/append b "s" "a" 0 {})
(persist/append b "s" "b" 0 {})
(persist/append b "s" "a" 0 {})
(persist/count-where
b
"s"
(fn (e) (equal? (persist/event-type e) "a")))))
2)
(persist-test
"queries return empty on empty stream"
(len (persist/read-since (persist/open) "s" 0))
0)
(persist-test
"queries work on the durable backend"
(let
((db (persist/mock-durable (persist/mem-backend))))
(begin
(persist/append db "s" "x" 100 {})
(persist/append db "s" "x" 200 {})
(len (persist/read-since db "s" 150))))
1)

View File

@@ -1,126 +0,0 @@
; Phase 4 — crash/restart integration. A whole subsystem (an order ledger:
; event log + a kv read model kept by a subscription + a periodic snapshot + an
; invoice blob ref) on the durable backend must survive a restart. "Crash" =
; drop every in-process object (backend, hub, projections); "restart" = rebuild
; them over the SAME disk + blob store. Nothing but the disk and content store
; carries across, exactly as a real process restart.
(define rec-count (fn (acc e) (+ acc 1)))
(persist-test
"log survives restart and seq continues"
(let
((disk (persist/mem-backend)))
(begin
(let
((db (persist/mock-durable disk)))
(begin
(persist/append db "orders" "placed" 0 {:id "a"})
(persist/append db "orders" "placed" 1 {:id "b"})))
(let
((db2 (persist/mock-durable disk)))
(list
(persist/project-fold db2 "orders" rec-count 0)
(persist/event-seq
(persist/append db2 "orders" "placed" 2 {:id "c"}))))))
(list 2 3))
(persist-test
"subscription-driven kv read model survives restart"
(let
((disk (persist/mem-backend)))
(begin
(let
((h (persist/hub (persist/mock-durable disk))))
(begin
(persist/subscribe
h
"orders"
(fn
(bk s e)
(persist/kv-update
bk
"order-count"
0
(fn (n) (+ n 1)))))
(persist/publish h "orders" "placed" 0 {})
(persist/publish h "orders" "placed" 1 {})))
(let
((db2 (persist/mock-durable disk)))
(persist/kv-get db2 "order-count"))))
2)
(persist-test
"snapshot taken before crash drives replay after restart"
(let
((disk (persist/mem-backend)))
(begin
(let
((db (persist/mock-durable disk)))
(begin
(persist/append db "orders" "placed" 0 {})
(persist/append db "orders" "placed" 1 {})
(persist/checkpoint db "orders" "count" rec-count 0)
(persist/append db "orders" "placed" 2 {})))
(let
((db2 (persist/mock-durable disk)))
(equal?
(persist/project-value
(persist/replay db2 "orders" "count" rec-count 0))
(persist/project-fold db2 "orders" rec-count 0)))))
true)
(persist-test
"compacted log still replays correctly after restart"
(let
((disk (persist/mem-backend)))
(begin
(let
((db (persist/mock-durable disk)))
(begin
(persist/append db "orders" "placed" 0 {})
(persist/append db "orders" "placed" 1 {})
(persist/append db "orders" "placed" 2 {})
(persist/compact db "orders" "count" rec-count 0)
(persist/append db "orders" "placed" 3 {})))
(let
((db2 (persist/mock-durable disk)))
(persist/project-value
(persist/replay db2 "orders" "count" rec-count 0)))))
4)
(persist-test
"invoice blob ref survives restart, bytes fetched from content store"
(let
((disk (persist/mem-backend)) (store (persist/mem-backend)))
(begin
(let
((db (persist/mock-durable disk)) (blob (persist/mock-blob store)))
(persist/kv-put
db
"invoice"
(persist/blob-store blob "INVOICEPDF" "application/pdf")))
(let
((db2 (persist/mock-durable disk))
(blob2 (persist/mock-blob store)))
(persist/blob-fetch blob2 (persist/kv-get db2 "invoice")))))
"INVOICEPDF")
(persist-test
"two independent restarts converge to the same state (determinism)"
(let
((disk (persist/mem-backend)))
(begin
(let
((db (persist/mock-durable disk)))
(begin
(persist/append db "orders" "placed" 0 {})
(persist/append db "orders" "placed" 1 {})
(persist/append db "orders" "placed" 2 {})))
(equal?
(persist/project-fold
(persist/mock-durable disk)
"orders"
rec-count
0)
(persist/project-fold
(persist/mock-durable disk)
"orders"
rec-count
0))))
true)

View File

@@ -1,114 +0,0 @@
; Phase 3 — snapshots + replay. Headline: snapshot + tail == full replay.
(define snap-count (fn (acc e) (+ acc 1)))
(persist-test
"no snapshot loads fresh seed state"
(persist/snapshot-load (persist/open) "feed" 0)
{:value 0 :seq 0})
(persist-test
"snapshot-exists? false initially"
(persist/snapshot-exists? (persist/open) "feed")
false)
(persist-test
"checkpoint stores a snapshot"
(let
((b (persist/open)))
(begin
(persist/append b "s" "x" 0 {})
(persist/checkpoint b "s" "snap" snap-count 0)
(persist/snapshot-exists? b "snap")))
true)
(persist-test
"checkpoint value equals full projection"
(let
((b (persist/open)))
(begin
(persist/append b "s" "x" 0 {})
(persist/append b "s" "x" 0 {})
(persist/append b "s" "x" 0 {})
(persist/project-value
(persist/checkpoint b "s" "snap" snap-count 0))))
3)
(persist-test
"checkpoint records the last seq"
(let
((b (persist/open)))
(begin
(persist/append b "s" "x" 0 {})
(persist/append b "s" "x" 0 {})
(persist/project-seq
(persist/checkpoint b "s" "snap" snap-count 0))))
2)
(persist-test
"replay after checkpoint only folds the tail"
(let
((b (persist/open)))
(begin
(persist/append b "s" "x" 0 {})
(persist/append b "s" "x" 0 {})
(persist/checkpoint b "s" "snap" snap-count 0)
(persist/append b "s" "x" 0 {})
(persist/project-value
(persist/replay b "s" "snap" snap-count 0))))
3)
(persist-test
"snapshot + tail == full replay (value)"
(let
((b (persist/open)))
(begin
(persist/append b "s" "x" 0 {})
(persist/append b "s" "x" 0 {})
(persist/checkpoint b "s" "snap" snap-count 0)
(persist/append b "s" "x" 0 {})
(persist/append b "s" "x" 0 {})
(equal?
(persist/project-value
(persist/replay b "s" "snap" snap-count 0))
(persist/project-fold b "s" snap-count 0))))
true)
(persist-test
"snapshot + tail == full replay (whole state)"
(let
((b (persist/open)))
(begin
(persist/append b "s" "x" 0 {})
(persist/checkpoint b "s" "snap" snap-count 0)
(persist/append b "s" "x" 0 {})
(persist/append b "s" "x" 0 {})
(equal?
(persist/replay b "s" "snap" snap-count 0)
(persist/project b "s" snap-count 0))))
true)
(persist-test
"replay determinism: two replays from same snapshot agree"
(let
((b (persist/open)))
(begin
(persist/append b "s" "x" 0 {})
(persist/checkpoint b "s" "snap" snap-count 0)
(persist/append b "s" "x" 0 {})
(equal?
(persist/replay b "s" "snap" snap-count 0)
(persist/replay b "s" "snap" snap-count 0))))
true)
(persist-test
"re-checkpoint advances the snapshot"
(let
((b (persist/open)))
(begin
(persist/append b "s" "x" 0 {})
(persist/checkpoint b "s" "snap" snap-count 0)
(persist/append b "s" "x" 0 {})
(persist/checkpoint b "s" "snap" snap-count 0)
(persist/project-seq (persist/snapshot-load b "snap" 0))))
2)
(persist-test
"snapshots are keyed independently"
(let
((b (persist/open)))
(begin
(persist/append b "s" "x" 0 {})
(persist/checkpoint b "s" "a" snap-count 0)
(persist/snapshot-exists? b "b")))
false)

View File

@@ -1,130 +0,0 @@
; Phase 2 — subscription hub: callbacks fire on publish, drive read models.
(persist-test
"no subscribers initially"
(persist/subscriber-count (persist/hub (persist/open)) "s")
0)
(persist-test
"subscribe registers a callback"
(let
((h (persist/hub (persist/open))))
(begin
(persist/subscribe h "s" (fn (b s e) nil))
(persist/subscriber-count h "s")))
1)
(persist-test
"publish appends to the log"
(let
((b (persist/open)))
(let
((h (persist/hub b)))
(begin
(persist/publish h "s" "x" 0 {})
(persist/publish h "s" "x" 0 {})
(persist/count b "s"))))
2)
(persist-test
"publish returns the stored event"
(let
((h (persist/hub (persist/open))))
(persist/event-seq (persist/publish h "s" "x" 0 {:id 1})))
1)
(persist-test
"callback fires on publish — drives a kv read model"
(let
((b (persist/open)))
(let
((h (persist/hub b)))
(begin
(persist/subscribe
h
"s"
(fn
(bk s e)
(persist/kv-update
bk
"count"
0
(fn (n) (+ n 1)))))
(persist/publish h "s" "x" 0 {})
(persist/publish h "s" "x" 0 {})
(persist/publish h "s" "x" 0 {})
(persist/kv-get b "count"))))
3)
(persist-test
"callback receives the event"
(let
((b (persist/open)))
(let
((h (persist/hub b)))
(begin
(persist/subscribe
h
"s"
(fn (bk s e) (persist/kv-put bk "last" (persist/event-type e))))
(persist/publish h "s" "created" 0 {})
(persist/kv-get b "last"))))
"created")
(persist-test
"subscriptions are per-stream"
(let
((b (persist/open)))
(let
((h (persist/hub b)))
(begin
(persist/subscribe
h
"s1"
(fn
(bk s e)
(persist/kv-update bk "n" 0 (fn (n) (+ n 1)))))
(persist/publish h "s2" "x" 0 {})
(persist/kv-get-or b "n" 0))))
0)
(persist-test
"multiple subscribers all fire"
(let
((b (persist/open)))
(let
((h (persist/hub b)))
(begin
(persist/subscribe
h
"s"
(fn
(bk s e)
(persist/kv-update bk "a" 0 (fn (n) (+ n 1)))))
(persist/subscribe
h
"s"
(fn
(bk s e)
(persist/kv-update bk "b" 0 (fn (n) (+ n 10)))))
(persist/publish h "s" "x" 0 {})
(list (persist/kv-get b "a") (persist/kv-get b "b")))))
(list 1 10))
(persist-test
"incremental read model via resume in callback"
(let
((b (persist/open)))
(let
((h (persist/hub b)))
(begin
(persist/kv-put b "proj" {:value 0 :seq 0})
(persist/subscribe
h
"s"
(fn
(bk s e)
(persist/kv-put
bk
"proj"
(persist/project-resume
bk
s
(fn (acc ev) (+ acc 1))
(persist/kv-get bk "proj")))))
(persist/publish h "s" "x" 0 {})
(persist/publish h "s" "x" 0 {})
(persist/project-value (persist/kv-get b "proj")))))
2)

View File

@@ -1,115 +0,0 @@
; Extension — event schema evolution via upcasters.
; v1 "placed" events had {:total N}; v2 wants {:amount N :currency "GBP"}.
(define up-placed (fn (e) (persist/upcast-data e {:amount (get (persist/event-data e) :total) :currency "GBP"})))
(persist-test
"unregistered type passes through unchanged"
(let
((reg (persist/upcasters)))
(persist/event-data
(persist/upcast
reg
(persist/event "s" 1 "other" 0 {:x 1}))))
{:x 1})
(persist-test
"registered upcaster lifts an old event"
(let
((reg (persist/register-upcaster (persist/upcasters) "placed" up-placed)))
(get
(persist/event-data
(persist/upcast
reg
(persist/event "s" 1 "placed" 0 {:total 50})))
:amount))
50)
(persist-test
"upcaster adds the new field"
(let
((reg (persist/register-upcaster (persist/upcasters) "placed" up-placed)))
(get
(persist/event-data
(persist/upcast
reg
(persist/event "s" 1 "placed" 0 {:total 50})))
:currency))
"GBP")
(persist-test
"upcast preserves stream/seq/type/at"
(let
((reg (persist/register-upcaster (persist/upcasters) "placed" up-placed)))
(let
((e (persist/upcast reg (persist/event "orders" 7 "placed" 99 {:total 1}))))
(list
(persist/event-seq e)
(persist/event-at e)
(persist/event-type e))))
(list 7 99 "placed"))
(persist-test
"registry is immutable — register returns a new dict"
(let
((r0 (persist/upcasters)))
(begin
(persist/register-upcaster r0 "placed" up-placed)
(has-key? r0 "placed")))
false)
(persist-test
"read-upcast lifts every event in a stream"
(let
((b (persist/open))
(reg
(persist/register-upcaster (persist/upcasters) "placed" up-placed)))
(begin
(persist/append b "orders" "placed" 0 {:total 10})
(persist/append b "orders" "placed" 0 {:total 20})
(let
((es (persist/read-upcast b "orders" reg)))
(list
(get (persist/event-data (nth es 0)) :amount)
(get (persist/event-data (nth es 1)) :amount)))))
(list 10 20))
(persist-test
"project-upcast folds over the current shape"
(let
((b (persist/open))
(reg
(persist/register-upcaster (persist/upcasters) "placed" up-placed)))
(begin
(persist/append b "orders" "placed" 0 {:total 10})
(persist/append b "orders" "placed" 0 {:total 20})
(persist/project-upcast
b
"orders"
reg
(fn (acc e) (+ acc (get (persist/event-data e) :amount)))
0)))
30)
(persist-test
"mixed old and new events fold uniformly"
(let
((b (persist/open))
(reg
(persist/register-upcaster (persist/upcasters) "placed" up-placed)))
(begin
(persist/append b "orders" "placed" 0 {:total 5})
(persist/append b "orders" "placed" 0 {:total 7 :amount 7})
(persist/project-upcast
b
"orders"
reg
(fn (acc e) (+ acc (get (persist/event-data e) :amount)))
0)))
12)
(persist-test
"upcast works on the durable backend"
(let
((db (persist/mock-durable (persist/mem-backend)))
(reg
(persist/register-upcaster (persist/upcasters) "placed" up-placed)))
(begin
(persist/append db "orders" "placed" 0 {:total 42})
(get
(persist/event-data
(nth (persist/read-upcast db "orders" reg) 0))
:amount)))
42)

View File

@@ -1,105 +0,0 @@
; Extension — materialized views: stay current on write, read O(1) via peek.
(define vw-count (fn (acc e) (+ acc 1)))
(define vw (persist/view "order-count" "orders" vw-count 0))
(persist-test "view-name" (persist/view-name vw) "order-count")
(persist-test "view-stream" (persist/view-stream vw) "orders")
(persist-test
"view-value folds the stream"
(let
((b (persist/open)))
(begin
(persist/append b "orders" "x" 0 {})
(persist/append b "orders" "x" 0 {})
(persist/view-value b vw)))
2)
(persist-test
"view-refresh persists a snapshot that peek then reads"
(let
((b (persist/open)))
(begin
(persist/append b "orders" "x" 0 {})
(persist/view-refresh b vw)
(persist/view-peek b vw)))
1)
(persist-test
"peek lags an un-refreshed tail"
(let
((b (persist/open)))
(begin
(persist/append b "orders" "x" 0 {})
(persist/view-refresh b vw)
(persist/append b "orders" "x" 0 {})
(persist/view-peek b vw)))
1)
(persist-test
"view-value sees the whole stream even after a stale snapshot"
(let
((b (persist/open)))
(begin
(persist/append b "orders" "x" 0 {})
(persist/view-refresh b vw)
(persist/append b "orders" "x" 0 {})
(persist/view-value b vw)))
2)
(persist-test
"attached view stays current on publish — peek needs no manual refresh"
(let
((b (persist/open)))
(let
((h (persist/view-attach (persist/hub b) vw)))
(begin
(persist/publish h "orders" "x" 0 {})
(persist/publish h "orders" "x" 0 {})
(persist/publish h "orders" "x" 0 {})
(persist/view-peek b vw))))
3)
(persist-test
"attached view advances the snapshot seq incrementally"
(let
((b (persist/open)))
(let
((h (persist/view-attach (persist/hub b) vw)))
(begin
(persist/publish h "orders" "x" 0 {})
(persist/publish h "orders" "x" 0 {})
(persist/project-seq
(persist/snapshot-load b "order-count" 0)))))
2)
(persist-test
"attach only reacts to its own stream"
(let
((b (persist/open)))
(let
((h (persist/view-attach (persist/hub b) vw)))
(begin
(persist/publish h "other" "x" 0 {})
(persist/view-peek b vw))))
0)
(persist-test
"materialized view works on the durable backend"
(let
((db (persist/mock-durable (persist/mem-backend))))
(let
((h (persist/view-attach (persist/hub db) vw)))
(begin
(persist/publish h "orders" "x" 0 {})
(persist/publish h "orders" "x" 0 {})
(persist/view-peek db vw))))
2)
(persist-test
"view sum over event data"
(let
((b (persist/open))
(sumv
(persist/view
"rev"
"sales"
(fn (acc e) (+ acc (get (persist/event-data e) :amt)))
0)))
(begin
(persist/append b "sales" "sale" 0 {:amt 10})
(persist/append b "sales" "sale" 1 {:amt 25})
(persist/view-value b sumv)))
35)

View File

@@ -1,44 +0,0 @@
; persist/upcast — event schema evolution. An append-only log keeps events
; forever, so old events have old shapes. Rather than migrate stored data (you
; can't rewrite history) or branch every projection on version, register an
; upcaster per event type: a pure (event -> event) that lifts an old event to
; the current shape. Reads pass through the registry so projections see ONE
; shape. The registry is an immutable dict the consumer threads (no global
; mutable state). Requires: lib/persist/event.sx, lib/persist/log.sx.
(define persist/upcasters (fn () {}))
(define persist/register-upcaster (fn (reg type fn) (assoc reg type fn)))
; apply the registered upcaster for an event's type, or pass it through unchanged
(define
persist/upcast
(fn
(reg e)
(let ((f (get reg (persist/event-type e)))) (if f (f e) e))))
; read a stream with every event lifted to current shape
(define
persist/read-upcast
(fn
(b stream reg)
(map (fn (e) (persist/upcast reg e)) (persist/read b stream))))
; project over upcasted events — projections never see a legacy shape
(define
persist/project-upcast
(fn
(b stream reg step seed)
(reduce step seed (persist/read-upcast b stream reg))))
; helper: upcast an event's :data by merging in/overriding fields, keeping the
; record's stream/seq/type/at. Common upcaster body.
(define
persist/upcast-data
(fn
(e new-data)
(persist/event
(persist/event-stream e)
(persist/event-seq e)
(persist/event-type e)
(persist/event-at e)
(merge (persist/event-data e) new-data))))

View File

@@ -1,49 +0,0 @@
; persist/view — a materialized view: the consumer-facing read model. It bundles
; a stream, a fold (step + seed) and a snapshot name. Attached to a hub it
; refreshes incrementally on every publish, so the materialized value stays
; current on write and reads are O(1) snapshot loads (persist/view-peek) instead
; of a full fold. This is what feed indices, mod audit rollups, search counters,
; etc. sit on. Requires: lib/persist/snapshot.sx, lib/persist/subscribe.sx.
(define persist/view (fn (name stream step seed) {:name name :step step :stream stream :seed seed}))
(define persist/view-name (fn (v) (get v :name)))
(define persist/view-stream (fn (v) (get v :stream)))
; bring the view's snapshot up to date with the log tail; returns the state
(define
persist/view-refresh
(fn
(b v)
(persist/checkpoint
b
(get v :stream)
(get v :name)
(get v :step)
(get v :seed))))
; current materialized value — refreshes first, so never stale
(define
persist/view-value
(fn (b v) (persist/project-value (persist/view-refresh b v))))
; O(1) read of the last persisted snapshot value WITHOUT folding the tail. Equal
; to view-value when the view is attached (kept current on every publish);
; otherwise may lag the log by the un-refreshed tail.
(define
persist/view-peek
(fn
(b v)
(persist/project-value
(persist/snapshot-load b (get v :name) (get v :seed)))))
; attach to a hub: refresh the view on every publish to its stream
(define
persist/view-attach
(fn
(h v)
(begin
(persist/subscribe
h
(persist/view-stream v)
(fn (bk s e) (persist/view-refresh bk v)))
h)))

View File

@@ -1,94 +0,0 @@
# host-persist loop agent (single agent, builds the durable storage host)
Role: make `lib/persist`'s durable backend **actually durable**. The persist
substrate (`lib/persist/**`, 201/201 tests) performs `{:op "persist/..." :args}`
IO requests for every storage op; under `sx_server.exe` today nothing services
them, so writes silently vanish. You build the **host-side adapter** that answers
those ops against real on-disk storage — the one piece standing between persist
and "all subsystems share a durable substrate."
```
worktree: /root/rose-ash-loops/host-persist
branch: loops/host-persist (push origin/loops/host-persist; NEVER main/architecture)
```
## The authoritative contract — read this first, every restart
`plans/persist-on-sx.md`**Blockers → "OPEN — host durable-storage adapter"**.
That entry is the spec: the silent-data-loss repro, the full op contract table,
the hard invariants (monotonic `last-seq`, etc.), the blob adapter shape, where
to register in `sx_server.ml`, and the acceptance test. Do not restate it here —
read it there and implement it. The reference implementation to mirror is
`persist/serve` in `lib/persist/durable.sx` (same op names, same shapes).
## Restart baseline — check before iterating
1. Read the Blocker spec (above) + this briefing.
2. `git log --oneline -8` on `loops/host-persist` to see what's done.
3. Is there a worktree-local build? `ls hosts/ocaml/_build/default/bin/sx_server.exe`.
Fresh worktrees have none — the first build is the first task.
4. If an acceptance suite exists (e.g. `hosts/ocaml/test/persist_durable_*` or a
`lib/persist/tests/durable-real.sx`), run it against the **worktree-built**
binary. Green before new work.
## The queue (phases)
- **Phase 0 — reproduce.** Confirm the silent-data-loss repro from the spec under
this worktree. Builds your mental model; costs one short run.
- **Phase 1 — storage module.** A new OCaml module under `hosts/ocaml/` that
implements the op contract over **real persistent storage**. Start simple and
correct: a filesystem-backed store (one append-only file per stream + a kv
file + a per-stream seq high-water file), or SQLite if the toolchain has it.
Honour every invariant in the spec — especially: `last-seq` is a monotonic
counter stored separately from rows so it survives `truncate`; values
round-trip structurally.
- **Phase 2 — register.** Wire a `"persist/..."` arm into the kernel's IO
resolver (`Sx_types._cek_io_resolver`, ~line 3864 of `hosts/ocaml/bin/sx_server.ml`)
and/or the `cek_run_with_io` bridge path (~528576), dispatching to the storage
module. Op names are the contract — do not rename.
- **Phase 3 — acceptance.** New tests that use `persist/durable-backend` (REAL
`perform`, not the mock) run under the freshly-built worktree binary: the
`durable` + `recovery` semantics must pass, and a **real process restart**
(start the built server, write, stop it, start again, replay) must recover
state from disk. Put host-owned tests under a host path (e.g.
`hosts/ocaml/test/`) — do not churn persist's existing suites.
- **Phase 4 — blob adapter.** Same pattern for `blob/put|get|has?` backed by a
content-addressed directory; persist stores only the ref.
Every iteration: implement → build → test → commit (short factual message) →
push → update `plans/persist-on-sx.md` (tick the Blocker toward CLOSED, append a
dated Progress-log line, newest first) → next.
## Ground rules (hard)
- **Build is your job** (unlike the persist loop). But build **only in this
worktree's `_build`** via `dune` from `/root/rose-ash-loops/host-persist`.
**NEVER overwrite the shared binary** at
`/root/rose-ash/hosts/ocaml/_build/default/bin/sx_server.exe` — every sibling
loop uses it; clobbering it breaks them all. Point acceptance tests at the
worktree binary (`hosts/ocaml/_build/default/bin/sx_server.exe` *inside this
worktree*).
- **First build is slow** (full OCaml). The `sx_build` MCP tool has a ~600s
watchdog that may kill it — prefer `dune build bin/sx_server.exe` (or `@all`)
run via Bash with `run_in_background: true` and a long timeout, then poll.
- **NEVER `pkill sx_server`** — siblings share the process/binary. Start your own
server on a throwaway path/port for restart tests and stop only that PID; bound
every run with `timeout`.
- **Scope:** `hosts/**`, host-owned test files, and the Blocker entry +
Progress log in `plans/persist-on-sx.md`. Do **not** modify `lib/persist/**`
source (the persist loop owns it; its API is your contract, not your code) —
if you need an upstream change, leave a note in the Blocker entry.
- **Determinism:** replay from disk must equal the in-memory semantics; same log
→ same state.
- **Commits:** one feature per commit; push to `origin/loops/host-persist`.
- **SX files:** `sx-tree` MCP tools ONLY, `file:` not `path:`, `sx_validate`
after edits. (Most of your work is OCaml — edit those with normal tools.)
## Definition of done
The Blocker entry flips to **CLOSED**: `persist/durable-backend` writes land on
disk, survive a real server restart, and the durable + recovery acceptance suites
are green against the worktree-built binary. At that point a subsystem migrated
per `lib/persist/examples/acl.sx` is genuinely durable.
Go. Read the Blocker spec; reproduce the gap; build the storage module.

View File

@@ -1,115 +0,0 @@
# persist-on-sx loop agent (single agent, queue-driven)
Role: iterates `plans/persist-on-sx.md` forever. **Durable state on the SX kernel**
— the foundation substrate every other subsystem currently fakes with an in-memory
mutable list. Event log (append-only streams) + kv (current-state) over one
injectable backend; pure projections; snapshots; durable IO at the kernel's
`perform` boundary. This is **substrate-level**, not a guest language.
```
description: persist-on-sx queue loop
subagent_type: general-purpose
run_in_background: true
isolation: worktree
```
## Prompt
You are the sole background agent working `plans/persist-on-sx.md`. Isolated
worktree `/root/rose-ash-loops/persist` on branch `loops/persist`, forever, one
commit per feature. Push to `origin/loops/persist` after every commit. Never touch
`main` or `architecture`.
## Restart baseline — check before iterating
1. Read `plans/persist-on-sx.md` — roadmap + Progress log. Note the scope table:
persist owns the **log** + **kv** facets; blobs are delegated (store the CID,
not the bytes); cache is out of scope. Do not event-source everything.
2. `ls lib/persist/` — pick up from the most advanced file.
3. If `lib/persist/tests/*.sx` exist, run them via `bash lib/persist/conformance.sh`.
Green before new work.
4. If `lib/persist/scoreboard.md` exists, that's your baseline.
5. **Learn the substrate before writing durable code.** persist sits on the kernel's
IO-suspension surface — the third CEK phase: `perform`, `cek-step-loop`,
`cek-resume`, `make-cek-suspended`. Study how IO is requested and resumed, and
how `spec/harness.sx` mocks an IO platform for tests (assert-io-*). Phases 13
need NO real IO — the in-memory backend is pure SX. Real durable IO (Phase 4)
goes through `perform` and is tested against the mock-IO harness, not a real disk.
Verify the actual exported names with sx_find_all / grep before relying on them.
## The queue
Phase order per `plans/persist-on-sx.md`:
- **Phase 1** — log + kv + in-memory backend (event record, injectable backend
protocol, append/read, kv get/put/delete, api).
- **Phase 2** — projections (`fold step seed`) + subscriptions; concurrency
conflict as a real result.
- **Phase 3** — snapshots + replay (checkpoint, replay = snapshot + tail,
determinism).
- **Phase 4** — durable backend via kernel IO (`perform`), blob-ref interface,
crash/restart replay against the mock-IO harness.
Within a phase, pick the checkbox that unlocks the most tests per effort.
Every iteration: implement → test → commit → tick `[ ]` → Progress log → next.
## Ground rules (hard)
- **Scope:** only `lib/persist/**` and `plans/persist-on-sx.md`. Do **not** edit
`spec/`, `hosts/`, `shared/`, or any `lib/<lang>/`. You may **import** the
kernel's IO-suspension + platform-IO surface only. **Do NOT add host primitives.**
If a durable IO op you need doesn't exist, it belongs in `hosts/` (out of scope) →
Blockers entry with a minimal repro, and stop on that item.
- **NEVER call `sx_build`.** 600s watchdog. If the sx_server binary is broken →
Blockers entry, stop. Run tests by invoking the sx_server binary directly from a
conformance.sh (model it on an existing one, e.g. `lib/apl/conformance.sh`),
pointing `SX_SERVER` at `/root/rose-ash/hosts/ocaml/_build/default/bin/sx_server.exe`
— fresh worktrees have no `_build/`.
- **Determinism:** replay must be pure — same log → same state. No clocks/randomness
inside projections; timestamps live on the event, passed in.
- **Shared-file issues** → plan's Blockers with minimal repro; don't fix here.
- **SX files:** `sx-tree` MCP tools ONLY. **They take `file:` not `path:`** — a
wrong key yields `Yojson Type_error("Expected string, got null")`, which looks
like a broken binary but is just a param mismatch. `sx_validate` after edits.
Path-based edits (`sx_replace_node`) count comment headers in their indices and
can clobber the wrong node — re-read after, or prefer `sx_write_file` for small
files.
- **Unicode in `.sx`:** raw UTF-8 only, never `\uXXXX` escapes.
- **Commit granularity:** one feature per commit. Short factual messages
(`persist: kv facet get/put/delete + 6 tests`). Push to `origin/loops/persist`.
- **Plan file:** update Progress log (newest first) + tick boxes every commit.
## persist-specific gotchas
- **Two facets, not one.** Don't force current-state values (a stock count, a
config value, a session blob) through the event log — that's the kv facet. Event
log is for things whose *history* matters.
- **Backend is injected.** The in-memory backend is the test default; never hardwire
it. Every op goes through the backend protocol so file/pg/ipfs swap in unchanged.
- **Optimistic concurrency is a real result.** A conflicting append returns a
conflict value the caller can retry on — not a crash, not a silent overwrite.
- **Blobs by reference only.** persist stores a content-address/CID + metadata. The
bytes live in a content-addressed store (artdag/IPFS). Never put large payloads in
the log.
- **Replay determinism is the headline property.** Snapshot + tail must equal full
replay. Test it explicitly, both directions.
## General gotchas (all loops)
- SX `do` = R7RS iteration. Use `begin` for multi-expr sequences.
- `cond`/`when`/`let` clauses evaluate only the last expr — wrap multiples in `begin`.
- `let` is parallel, not sequential — nest `let`s when a binding references an earlier one.
- `env-bind!` creates a binding; `env-set!` mutates an existing one (walks scope chain).
- `sx_validate` after every structural edit.
- Namespace-prefix all helpers (`persist/...`) — short/host-colliding names get
silently shadowed or hang the runtime.
## Style
- No comments in `.sx` unless non-obvious.
- No new planning docs — update `plans/persist-on-sx.md` inline.
- Short, factual commit messages.
- One feature per iteration. Commit. Log. Push. Next.
Go. Start by reading the plan; find the first unchecked `[ ]`; implement it.

View File

@@ -1,82 +0,0 @@
# commerce-on-sx: Catalog, cart, pricing & orders on miniKanren
> **DRAFT outline.** The revenue vertical. Depends on `persist-on-sx` (durable
> orders) and `flow-on-sx` (checkout as a durable flow). Don't start before
> persist-on-sx Phase 1 is green.
rose-ash's revenue engine — market (catalog), cart (checkout), orders (SumUp
payment, reconciliation) — has no SX subsystem. The hard part of commerce isn't
CRUD; it's **pricing**: discounts, bundles, tax, membership rates, promotions that
stack (or don't). These are relations, and a relational engine can run them in
multiple directions — forward ("what's the total?") and backward ("what promo code
yields this total?", "which line item triggered the discount?").
That's a miniKanren fit. Pricing/promotion rules are relational; cart and order
*lifecycle* (reserve → pay → fulfil → reconcile) is a durable `flow`; the order
ledger is a `persist` stream. Commerce is the first real **composition** subsystem.
End-state: a catalog model, a relational pricing/promotion engine, a cart with
deterministic totals, and an order lifecycle flow with payment-webhook
reconciliation — all auditable via the event log.
## Status (rolling)
`bash lib/commerce/conformance.sh`**0/0** (not yet started)
## Ground rules
- **Scope:** only `lib/commerce/**` and `plans/commerce-on-sx.md`. May **import**
from `lib/minikanren/`, and (once they exist) `lib/persist/` + `lib/flow/`. Do not
edit substrates.
- **Architecture:** prices/promotions are miniKanren relations over catalog facts;
a cart total is a *deterministic* query result (first solution under a fixed rule
order). Order lifecycle is a `flow` that suspends at the payment IO boundary.
Money is integer minor units — never floats.
- **Determinism:** promotion stacking must have explicit, tested precedence;
totals must be reproducible from the cart + catalog snapshot.
- **Commits:** one feature per commit. Progress log + tick boxes.
## Architecture sketch
```
Catalog + cart Total / order
product(id,price,tags) {:subtotal :discounts :tax :total}
│ ▲
▼ │
lib/commerce/catalog.sx lib/commerce/price.sx
— product / variant / stock facts — miniKanren pricing relations
│ — promo stacking, membership rates
▼ ▲
lib/commerce/cart.sx lib/commerce/order.sx (flow + store)
— line items, quantities — reserve→pay→fulfil→reconcile
│ — SumUp webhook = flow resume
▼ │
lib/commerce/api.sx ── (commerce/add) (commerce/total) (commerce/checkout) ──┘
```
## Phase 1 — Catalog + cart + deterministic totals
- [ ] `catalog.sx` — product/variant/stock as facts
- [ ] `cart.sx` — line items, add/remove/qty
- [ ] `price.sx` — base pricing relation, subtotal; tax
- [ ] `api.sx` + tests + scoreboard + conformance.sh
## Phase 2 — Promotions (relational)
- [ ] promo rules: percentage, fixed, bundle, member rate
- [ ] explicit stacking precedence; "best price" backward query
- [ ] tests: stacking order, mutually-exclusive promos, member vs guest
## Phase 3 — Order lifecycle (flow + store)
- [ ] order flow: reserve stock → await payment → fulfil
- [ ] payment webhook resumes the suspended flow
- [ ] order ledger as a `persist` stream; idempotent reconciliation
## Phase 4 — Reconciliation + federation
- [ ] mismatch detection (paid≠ordered) as queries over the ledger
- [ ] cross-instance catalog (federated marketplace) — out-of-scope stub
- [ ] tests: webhook replay, partial refund, double-charge guard
## Progress log
(loop fills this in)
## Blockers
(loop fills this in)

Some files were not shown because too many files have changed in this diff Show More