Compare commits

...

43 Commits

Author SHA1 Message Date
200b93c1f6 persist: Blocker spec for the host durable-storage adapter
Some checks failed
Test, Build, and Deploy / test-build-deploy (push) Failing after 36s
Document the one gap to real durability: a hosts/ servicer for the persist/*
IO ops. Includes the silent-data-loss repro (durable-backend currently no-ops
under sx_server's default resolver), the full op contract table, hard
invariants (monotonic last-seq, etc.), the blob adapter shape, where to
register in sx_server.ml, and an acceptance test (swap transport, run durable +
recovery suites against real storage, survive a real restart).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-06 20:52:44 +00:00
84d5732b38 persist: worked reference migration — acl grants on persist + 10 tests
Some checks failed
Test, Build, and Deploy / test-build-deploy (push) Failing after 41s
examples/acl.sx: a tested template migrating an ACL-grants store from a
hand-rolled ephemeral map to persist — grants/revokes as events, current set as
a projection, O(1) checks via a materialized view, audit via read-window.
Header carries the BEFORE->AFTER diff. Proves grants survive restart on the
durable backend (the capability the BEFORE version lacked). The pattern other
subsystem loops copy; does not touch the real lib/acl. 201/201.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-06 20:43:15 +00:00
a37a158d01 persist: global commit ordering across streams + 11 tests
Some checks failed
Test, Build, and Deploy / test-build-deploy (push) Failing after 46s
global.sx: persist/gappend records a pointer in a reserved $global index whose
seq is the global commit position; read-global/project-global replay every
event in commit order; global-from for incremental consumers. Opt-in (plain
append untouched); $-prefixed streams now reserved + hidden from the public
catalog (streams-all reveals them). Gives feed its unified timeline.
Deterministic across restart. 191/191.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-06 20:41:01 +00:00
3e90c780e9 persist: exactly-once append under retries + 9 tests
Some checks failed
Test, Build, and Deploy / test-build-deploy (push) Failing after 41s
idempotency.sx: persist/append-once appends at most once per (stream,
idempotency key), returning the same event on a repeat. The marker lives in the
kv facet, so idempotency holds across a restart (verified on durable).
persist/seen? check. 180/180.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-06 19:28:21 +00:00
0f6dbdfc7d persist: event schema evolution via upcasters + 9 tests
Some checks failed
Test, Build, and Deploy / test-build-deploy (push) Failing after 37s
upcast.sx: register a pure (event -> event) upcaster per type in an immutable
registry; read-upcast/project-upcast lift legacy events to the current shape on
read so projections see one shape (no version branching, no history rewrite).
upcast-data helper merges new :data fields. 171/171.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-06 19:26:35 +00:00
62a1485302 persist: atomic batch append — contiguous block + transactional guard + 10 tests
Some checks failed
Test, Build, and Deploy / test-build-deploy (push) Failing after 43s
batch.sx: persist/append-batch commits (type at data) specs as one contiguous
block; persist/append-batch-expect checks the stream is still at expected
before writing any event, so the batch is all-or-nothing under a concurrent
writer (conflict is a value, not a partial write). 162/162.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-06 19:24:35 +00:00
4e521e3d7a persist: read-side query helpers — seq/time/type/predicate scans + 9 tests
Some checks failed
Test, Build, and Deploy / test-build-deploy (push) Failing after 28s
query.sx: read-between (seq range), read-since/read-window (by :at),
read-by-type, read-where, count-where. Pure scans over persist/read for audit
windows, type filters, since-cursors. 152/152.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-06 19:22:03 +00:00
a00439da6e persist: stream catalog — enumerate streams + 10 tests
Some checks failed
Test, Build, and Deploy / test-build-deploy (push) Failing after 30s
New backend op :streams (from seq high-water marks, so compacted streams still
list), threaded through mem-backend + durable serve/io-backend. catalog.sx:
persist/streams, stream-count, stream-exists?, total-events. 143/143.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-06 19:20:22 +00:00
8e16ba6b04 persist: kv compare-and-swap + create-only put + 11 tests
Some checks failed
Test, Build, and Deploy / test-build-deploy (push) Failing after 43s
kv.sx: persist/kv-cas sets a key only if its current value equals expected,
else returns {:conflict :expected :actual}; persist/kv-put-new is create-only.
The kv analogue of log append-expect — atomic current-state for sessions, acl
grants, stock counts. 133/133.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-06 19:17:53 +00:00
ecdaeea223 persist: materialized views — stay current on write, O(1) read + 11 tests
Some checks failed
Test, Build, and Deploy / test-build-deploy (push) Failing after 27s
view.sx: persist/view bundles stream + fold + snapshot name; view-attach
subscribes it to a hub so each publish refreshes the snapshot incrementally,
making view-peek an O(1) current read. view-value always folds the tail so it
is never stale. The consumer read-model abstraction (feed indices, audit
rollups, search counters). 122/122.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-06 19:16:16 +00:00
4be6988963 persist: crash/restart recovery integration + migration notes — Phase 4 complete
Some checks failed
Test, Build, and Deploy / test-build-deploy (push) Failing after 37s
recovery.sx: 6-test end-to-end crash/restart of an order ledger (log +
subscription kv read model + snapshot + compaction + invoice blob ref) on the
durable backend; everything survives a restart over the same disk + content
store, seq continues, two restarts converge. Migration notes (mem → durable
under a live subsystem) added to the plan. Roadmap done, 111/111.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-06 19:14:01 +00:00
1c7b602978 persist: blob backend — store the ref/CID, never the bytes + 14 tests
Some checks failed
Test, Build, and Deploy / test-build-deploy (push) Has been cancelled
blob.sx: a blob ref is {:cid :size :mime}; the blob store is a separate
injected dependency (perform in prod, mock content store in tests).
persist/blob-store puts bytes and returns only the ref; bytes live in a
content-addressed store (artdag/IPFS). Tests assert refs in log/kv never carry
the bytes + content-address dedup. 105/105.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-06 19:11:48 +00:00
90c2a57975 persist: durable backend over the perform IO boundary + 15 tests
Some checks failed
Test, Build, and Deploy / test-build-deploy (push) Has been cancelled
durable.sx: io-backend with an injectable transport — persist/durable-backend
performs each op as {:op "persist/..." :args (...)} (kernel suspends, host
resumes); persist/mock-durable services via persist/serve over an in-memory
disk. Identical request shapes mean the whole facet/projection/snapshot/
compaction stack runs unchanged on the durable backend. Crash/restart replay
recovers log+kv+snapshot. 91/91.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-06 19:09:12 +00:00
aff7d1e84f persist: compaction — drop snapshotted prefix, monotonic seq + 11 tests
Some checks failed
Test, Build, and Deploy / test-build-deploy (push) Failing after 1m0s
Backend now tracks last-seq as a monotonic high-water mark (survives
truncation) and exposes :truncate-through. compaction.sx: persist/compact
checkpoints then drops events with seq <= snapshot seq; should-compact?/
maybe-compact give an explicit every-N policy. Determinism: post-compaction
replay value == uncompacted full replay. Phase 3 complete, 76/76.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-06 18:42:06 +00:00
b0874b1282 persist: snapshots — checkpoint + replay = snapshot + tail + 11 tests
Some checks are pending
Test, Build, and Deploy / test-build-deploy (push) Waiting to run
snapshot.sx: snapshot is a projection state {:value :seq} stored in kv under
snapshot/<name>. persist/checkpoint replays and saves; persist/replay folds
only the tail after the snapshot. Tests assert snapshot+tail == full replay
both ways + determinism. 65/65.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-06 18:39:41 +00:00
156d6f12ec persist: optimistic concurrency — conflict as a real result + 8 tests
Some checks failed
Test, Build, and Deploy / test-build-deploy (push) Has been cancelled
concurrency.sx: persist/append-expect refuses an append when the stream
advanced past the caller's expected seq, returning {:conflict :expected
:actual} instead of crashing or overwriting. persist/conflict? + accessors.
Phase 2 complete, 54/54.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-06 18:37:49 +00:00
03da8d4328 persist: subscription hub — read models update on publish + 9 tests
Some checks failed
Test, Build, and Deploy / test-build-deploy (push) Failing after 1m6s
subscribe.sx: persist/hub wraps a backend; persist/publish appends then fires
per-stream callbacks (backend stream event). Direct persist/append bypasses
subscribers (bulk load/replay). Callbacks drive kv counters / project-resume. 46/46.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-06 18:36:16 +00:00
a6864178c3 persist: projections — fold stream into read model, incremental resume + 9 tests
Some checks failed
Test, Build, and Deploy / test-build-deploy (push) Failing after 1m9s
project.sx: projection state {:value :seq}; persist/project folds the whole
stream, persist/project-resume folds only the tail so read models update
incrementally. Pure step (value event)->value. 37/37.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-06 18:34:52 +00:00
314cc37030 persist: Phase 1 — log + kv facets on injectable in-memory backend + 28 tests
Some checks failed
Test, Build, and Deploy / test-build-deploy (push) Failing after 1m3s
event/backend/log/kv/api over one injected backend protocol (mem default).
log: append/read/read-from, sequential per-stream seq, stream isolation.
kv: get/put/delete/has?/keys/get-or/update. conformance.sh + 3 suites, 28/28.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-06 18:32:51 +00:00
b80cc32363 briefings: add persist-on-sx loop briefing
Some checks failed
Test, Build, and Deploy / test-build-deploy (push) Failing after 1m2s
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-06 18:24:52 +00:00
1902cce57f plans: rename store-on-sx → persist-on-sx; clarify it's persistence not shop, and scope (log+kv facets, blobs delegated, cache excluded)
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-06 18:20:14 +00:00
ff537bfba2 plans: six subsystem outline plans for the SX rewrite (store, commerce, identity, content, events, host)
Gap analysis from the five-subsystem set (acl/feed/flow/mod/search):
- store-on-sx: event-sourcing foundation the others fake with in-memory lists (build first)
- commerce-on-sx: catalog/cart/pricing/orders on miniKanren (+ store + flow)
- identity-on-sx: OAuth2/sessions/membership on Erlang (the core acl assumes)
- content-on-sx: documents/blocks/CRDT on Smalltalk
- events-on-sx: calendar/ticketing on Datalog + flow-driven delivery
- host-on-sx: the web boundary — off Quart onto native server+SXTP now, dream-on-sx next

All DRAFT outlines; substrate choices proposed, not final.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-06 17:39:29 +00:00
1e4cf25015 Merge loops/feed into architecture: feed-on-sx activity feed engine on APL
Some checks failed
Test, Build, and Deploy / test-build-deploy (push) Failing after 42s
Activity feeds as APL array math on lib/apl/ — timelines, fanout, ranking,
visibility, federation. Roadmap (4 phases) + 8 extensions, 189/189 tests.

- Phase 1: stream model (normalize, filter/sort/take/reverse)
- Phase 2: fanout via outer product (∘.×), edge-guard, dedupe
- Phase 3: aggregation + ranking (recency/velocity/engagement, top-N)
- Phase 4: per-viewer ACL + federation (injected permit?/transport)
- Extensions: TF-IDF, notifications, home capstone, smart-dedupe,
  trending, mute, pagination, threading

Purely additive under lib/feed/**; no conflicts.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-06 17:23:42 +00:00
9c4a5d1913 feed: conversation threading — :reply-to transitive closure (thread/replies/thread-size) + 12 tests
Some checks failed
Test, Build, and Deploy / test-build-deploy (push) Failing after 42s
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-06 17:00:10 +00:00
f91ac82434 feed: pagination — offset/limit + cursor-by-at (before/after/page-before/next-cursor) + 14 tests
Some checks failed
Test, Build, and Deploy / test-build-deploy (push) Failing after 44s
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-06 16:58:36 +00:00
5136249ae5 feed: viewer mute/block — mute actors/tags/objects + apply-prefs bag + 9 tests
Some checks failed
Test, Build, and Deploy / test-build-deploy (push) Failing after 42s
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-06 16:57:05 +00:00
6fc61147a8 feed: trending objects/actors by recent activity window, deterministic tiebreak + 11 tests
Some checks failed
Test, Build, and Deploy / test-build-deploy (push) Failing after 40s
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-06 16:55:55 +00:00
0122c41ecb feed: verb-aware smart dedupe — reactions collapse cross-actor, posts stay per-actor + 9 tests
Some checks failed
Test, Build, and Deploy / test-build-deploy (push) Failing after 54s
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-06 16:54:21 +00:00
58656b03e4 feed: feed/home capstone — fanout∘inbox∘dedupe∘ACL∘rank∘take as one line + 6 tests
Some checks failed
Test, Build, and Deploy / test-build-deploy (push) Failing after 42s
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-06 16:53:15 +00:00
b0feb7b01b feed: notification feed — per-recipient inbox, verb filter, (verb,object) digest + 8 tests
Some checks failed
Test, Build, and Deploy / test-build-deploy (push) Failing after 54s
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-06 16:51:53 +00:00
a979297959 feed: TF-IDF content ranking over :tags — tag-df/idf, tfidf-score, by-relevance + 15 tests
Some checks failed
Test, Build, and Deploy / test-build-deploy (push) Failing after 43s
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-06 16:50:36 +00:00
37226cf6eb feed: Phase 4 visibility + federation — per-viewer ACL, fanout partition, inbound/backfill/ingest, e2e feed/timeline + 22 tests
Some checks failed
Test, Build, and Deploy / test-build-deploy (push) Failing after 36s
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-06 16:48:27 +00:00
50a7f31a39 feed: Phase 3 aggregation + ranking — group-by, recency/velocity/engagement scorers, composite, top-N via stable grade-down + 24 tests
Some checks failed
Test, Build, and Deploy / test-build-deploy (push) Failing after 42s
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-06 16:44:04 +00:00
915f51b2b6 feed: Phase 2 fanout via outer product — activities ∘.× audience, flatten, edge-guard, dedupe + 29 tests
Some checks failed
Test, Build, and Deploy / test-build-deploy (push) Failing after 21s
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-06 16:40:34 +00:00
e7501bdf8f feed: Phase 1 stream model — normalize, APL-backed filter/sort/take/reverse, post/all api + 30 tests
Some checks failed
Test, Build, and Deploy / test-build-deploy (push) Failing after 49s
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-06 16:31:36 +00:00
c3a0727645 plans: five rose-ash subsystem plans + three loop briefings
Some checks failed
Test, Build, and Deploy / test-build-deploy (push) Failing after 19s
Plans for acl-on-sx (Datalog), flow-on-sx (Scheme), feed-on-sx (APL),
mod-on-sx (Prolog), search-on-sx (Haskell). Each is a 4-phase queue
sitting on its respective guest language, targeting rose-ash needs:
access control, durable workflows, activity feeds, moderation, search.
Federation extension in Phase 4 of each (plugs into fed-sx).

Briefings for the three loops we're kicking off now: acl-loop,
flow-loop, feed-loop. mod-sx and search-sx briefings will follow
once the first three have surfaced any shared infrastructure
worth extracting to lib/guest/.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-06-06 15:55:39 +00:00
1b94082a71 Merge loops/erlang into architecture: Erlang substrate fixes (FFI + tokenizer + charlists + integer literals)
Four small, contained substrate fixes that came out of the fed-sx-m1 milestone work — all scoped to
lib/erlang/, no other-language regressions:

  c6f397c3  register binary_to_list/1 + list_to_binary/1 BIFs (+9 ffi tests, 738/738)
  9fe5c904  $X char literals decode to char code in tokenizer (+12 eval tests, 750/750)
  5098a8f0  atom_to_list/integer_to_list return Erlang charlists; list_to_* accept both (+9 eval, 759/759)
  bcabed6b  integer literals truncate to strict int (was float; broke integer->char)

Together these complete the byte-level term-codec primitive set:
  binary_to_list / list_to_binary (iolist-aware; round-trips for free)
  $X char literals decoding to int char codes
  atom_to_list / integer_to_list returning standard Erlang charlists
  integer literals coercing to strict int (not float)

Any Erlang-on-SX consumer that needs to construct/deconstruct byte sequences or work with charlists now
does so with standard Erlang semantics. Scoreboard: 759/759 (full Erlang suite).

Loop branch loops/erlang stays alive for future Erlang substrate work; this just lands the closed deliverables.
2026-06-06 15:45:46 +00:00
57184daaee briefings: add kernel-on-sx loop briefing
Some checks failed
Test, Build, and Deploy / test-build-deploy (push) Failing after 42s
Sibling to apl-loop / common-lisp-loop / scheme-loop. Captures the
queue-driven kernel loop pattern (Phase B stratification entry-point,
env-as-value successor, motivates lib/guest/reflective/).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-06-06 15:28:09 +00:00
d9e2627b89 Merge loops/go into architecture: Go-on-SX, 609/609 across 11 phases, loop closed
Some checks failed
Test, Build, and Deploy / test-build-deploy (push) Failing after 24s
2026-06-06 15:17:17 +00:00
bcabed6bce erlang: integer literals truncate to strict int (was float; broke integer->char)
Some checks failed
Test, Build, and Deploy / test-build-deploy (push) Failing after 30s
2026-06-06 08:05:57 +00:00
5098a8f015 erlang: atom_to_list/integer_to_list return Erlang charlists; list_to_* accept both (+9 net eval, 759/759) 2026-06-06 08:04:45 +00:00
9fe5c9044d erlang: $X char literals decode to char code in tokenizer (+12 eval tests, 750/750) 2026-06-06 08:03:46 +00:00
c6f397c3d9 erlang: register binary_to_list/1 + list_to_binary/1 BIFs (+9 ffi tests, 738/738) 2026-06-06 08:02:36 +00:00
97 changed files with 7559 additions and 36 deletions

View File

@@ -1561,7 +1561,66 @@
(er-register-pure-bif! "crypto" "hash" 2 er-bif-crypto-hash)
(er-register-pure-bif! "cid" "from_bytes" 1 er-bif-cid-from-bytes)
(er-register-pure-bif! "cid" "to_string" 1 er-bif-cid-to-string)
;; ── binary_to_list / list_to_binary (Step 3b — term codec) ──────
;; Standard Erlang semantics:
;; binary_to_list(<<B1,B2,...>>) -> [B1, B2, ...] (Erlang cons of ints)
;; list_to_binary(IoList) -> <<...>> (flattens nested
;; iolists; elements are byte ints 0-255 or binaries)
;; Bad arg / out-of-range byte / non-iolist element -> error:badarg.
(define er-bif-binary-to-list
(fn (vs)
(let ((v (nth vs 0)))
(cond
(not (er-binary? v))
(raise (er-mk-error-marker (er-mk-atom "badarg")))
:else
(let ((bs (get v :bytes)) (out (er-mk-nil)))
(for-each
(fn (i)
(set! out (er-mk-cons (nth bs (- (- (len bs) 1) i)) out)))
(range 0 (len bs)))
out)))))
;; Walk an Erlang iolist, appending bytes to `acc` (a mutable SX list).
;; Accepts: nil, cons-of-X, binary, integer in 0..255. Anything else
;; signals failure by setting (nth fail 0) to true.
(define er-iolist-walk!
(fn (v acc fail)
(cond
(nth fail 0) nil
(er-nil? v) nil
(er-cons? v)
(do (er-iolist-walk! (get v :head) acc fail)
(er-iolist-walk! (get v :tail) acc fail))
(er-binary? v)
(for-each
(fn (i) (append! acc (nth (get v :bytes) i)))
(range 0 (len (get v :bytes))))
(= (type-of v) "number")
(cond
(and (>= v 0) (<= v 255)) (append! acc v)
:else (set-nth! fail 0 true))
:else (set-nth! fail 0 true))))
(define er-bif-list-to-binary
(fn (vs)
(let ((v (nth vs 0)) (acc (list)) (fail (list false)))
(cond
(not (or (er-nil? v) (er-cons? v) (er-binary? v)))
(raise (er-mk-error-marker (er-mk-atom "badarg")))
:else
(do
(er-iolist-walk! v acc fail)
(cond
(nth fail 0)
(raise (er-mk-error-marker (er-mk-atom "badarg")))
:else (er-mk-binary acc)))))))
(er-register-bif! "file" "list_dir" 1 er-bif-file-list-dir)
(er-register-pure-bif! "erlang" "binary_to_list" 1 er-bif-binary-to-list)
(er-register-pure-bif! "erlang" "list_to_binary" 1 er-bif-list-to-binary)
(er-mk-atom "ok")))
;; Register everything at load time.

View File

@@ -1,18 +1,18 @@
{
"language": "erlang",
"total_pass": 729,
"total": 729,
"total_pass": 761,
"total": 761,
"suites": [
{"name":"tokenize","pass":62,"total":62,"status":"ok"},
{"name":"parse","pass":52,"total":52,"status":"ok"},
{"name":"eval","pass":385,"total":385,"status":"ok"},
{"name":"eval","pass":408,"total":408,"status":"ok"},
{"name":"runtime","pass":93,"total":93,"status":"ok"},
{"name":"ring","pass":4,"total":4,"status":"ok"},
{"name":"ping-pong","pass":4,"total":4,"status":"ok"},
{"name":"bank","pass":8,"total":8,"status":"ok"},
{"name":"echo","pass":7,"total":7,"status":"ok"},
{"name":"fib","pass":8,"total":8,"status":"ok"},
{"name":"ffi","pass":28,"total":28,"status":"ok"},
{"name":"ffi","pass":37,"total":37,"status":"ok"},
{"name":"vm","pass":78,"total":78,"status":"ok"}
]
}

View File

@@ -1,19 +1,19 @@
# Erlang-on-SX Scoreboard
**Total: 729 / 729 tests passing**
**Total: 761 / 761 tests passing**
| | Suite | Pass | Total |
|---|---|---|---|
| ✅ | tokenize | 62 | 62 |
| ✅ | parse | 52 | 52 |
| ✅ | eval | 385 | 385 |
| ✅ | eval | 408 | 408 |
| ✅ | runtime | 93 | 93 |
| ✅ | ring | 4 | 4 |
| ✅ | ping-pong | 4 | 4 |
| ✅ | bank | 8 | 8 |
| ✅ | echo | 7 | 7 |
| ✅ | fib | 8 | 8 |
| ✅ | ffi | 28 | 28 |
| ✅ | ffi | 37 | 37 |
| ✅ | vm | 78 | 78 |

View File

@@ -228,9 +228,10 @@
(er-eval-test "tuple_size 0" (ev "tuple_size({})") 0)
;; ── BIFs: atom / list conversions ───────────────────────────────
(er-eval-test "atom_to_list" (ev "atom_to_list(hello)") "hello")
(er-eval-test "atom_to_list -> charlist length" (ev "length(atom_to_list(hello))") 5)
(er-eval-test "atom_to_list -> head $h" (ev "hd(atom_to_list(hello))") 104)
(er-eval-test "list_to_atom roundtrip"
(nm (ev "list_to_atom(atom_to_list(foo))")) "foo")
(nm (ev "list_to_atom(atom_to_list(foo))")) "foo") ;; round-trip via charlist
(er-eval-test "list_to_atom fresh"
(nm (ev "list_to_atom(\"bar\")")) "bar")
@@ -1060,11 +1061,13 @@
(er-eval-test "list_to_tuple roundtrip"
(ev "tuple_size(list_to_tuple([10, 20, 30]))") 3)
(er-eval-test "integer_to_list" (ev "integer_to_list(42)") "42")
(er-eval-test "integer_to_list neg" (ev "integer_to_list(-99)") "-99")
(er-eval-test "integer_to_list -> charlist length" (ev "length(integer_to_list(42))") 2)
(er-eval-test "integer_to_list 42 head $4" (ev "hd(integer_to_list(42))") 52)
(er-eval-test "integer_to_list neg -> charlist length" (ev "length(integer_to_list(-99))") 3)
(er-eval-test "integer_to_list -99 head $-" (ev "hd(integer_to_list(-99))") 45)
(er-eval-test "list_to_integer" (ev "list_to_integer(\"123\")") 123)
(er-eval-test "list_to_integer roundtrip"
(ev "list_to_integer(integer_to_list(7))") 7)
(ev "list_to_integer(integer_to_list(7))") 7) ;; round-trip via charlist
(er-eval-test "is_function fun"
(nm (ev "F = fun (X) -> X end, is_function(F)")) "true")
@@ -1341,6 +1344,42 @@
(get (nth (get er-rt-cap-result :elements) 4) :name) "true")
;; ── $X char literals (Step 3b substrate fix 2026-06-04) ──────────
(er-eval-test "char $A" (ev "$A") 65)
(er-eval-test "char $a" (ev "$a") 97)
(er-eval-test "char $0 is digit, not escape-NUL" (ev "$0") 48)
(er-eval-test "char $\\n is newline (10)" (ev "$\\n") 10)
(er-eval-test "char $\\t is tab (9)" (ev "$\\t") 9)
(er-eval-test "char $\\r is CR (13)" (ev "$\\r") 13)
(er-eval-test "char $\\s is space (32)" (ev "$\\s") 32)
(er-eval-test "char $\\0 is NUL (0)" (ev "$\\0") 0)
(er-eval-test "char $\\\\ is backslash (92)" (ev "$\\\\") 92)
(er-eval-test "[$h,$i] head is 104" (ev "hd([$h, $i])") 104)
(er-eval-test "list_to_binary char-list -> bytes"
(ev "byte_size(list_to_binary([$f, $e, $d]))") 3)
(er-eval-test "list_to_binary char-list round-trip"
(nm (ev "list_to_binary([$h, $i]) =:= <<104, 105>>")) "true")
;; ── atom_to_list / integer_to_list charlist semantics (Step 3b substrate fix #3) ──
(er-eval-test "atom_to_list hd is char code"
(ev "hd(atom_to_list(hi))") 104)
(er-eval-test "atom_to_list maps to bytes via list_to_binary"
(ev "byte_size(list_to_binary(atom_to_list(hello)))") 5)
(er-eval-test "atom_to_list -> list_to_binary -> bytes content"
(nm (ev "list_to_binary(atom_to_list(ok)) =:= <<111, 107>>")) "true")
(er-eval-test "integer_to_list 12345 -> 5 chars"
(ev "length(integer_to_list(12345))") 5)
(er-eval-test "integer_to_list -> bytes -> back"
(ev "list_to_integer(integer_to_list(99999))") 99999)
(er-eval-test "list_to_atom from charlist"
(nm (ev "list_to_atom([$f, $o, $o])")) "foo")
(er-eval-test "list_to_atom from SX-string back-compat"
(nm (ev "list_to_atom(\"bar\")")) "bar")
(er-eval-test "list_to_integer from charlist"
(ev "list_to_integer([$1, $0, $0])") 100)
(define
er-eval-test-summary
(str "eval " er-eval-test-pass "/" er-eval-test-count))

View File

@@ -160,6 +160,51 @@
(ffi-nm (ffi-ev "element(2, file:list_dir(\"/no/such/dir/xyz\"))"))
"enoent")
(er-ffi-test
"binary_to_list <<1,2,3>> length"
(ffi-ev "length(binary_to_list(<<1,2,3,4,5>>))")
5)
(er-ffi-test
"binary_to_list hd byte"
(ffi-ev "hd(binary_to_list(<<7,8,9>>))")
7)
(er-ffi-test
"binary_to_list empty -> []"
(ffi-nm (ffi-ev "case binary_to_list(<<>>) of [] -> empty end"))
"empty")
(er-ffi-test
"list_to_binary flat list bytes"
(ffi-ev "byte_size(list_to_binary([1,2,3]))")
3)
(er-ffi-test
"list_to_binary nested iolist"
(ffi-ev "byte_size(list_to_binary([1, <<2,3>>, [4, [5]]]))")
5)
(er-ffi-test
"list_to_binary round-trip via binary_to_list"
(ffi-nm (ffi-ev "list_to_binary(binary_to_list(<<10,20,30>>)) =:= <<10,20,30>>"))
"true")
(er-ffi-test
"binary_to_list non-binary -> error:badarg"
(ffi-nm (ffi-ev "try binary_to_list(42) catch error:badarg -> ok end"))
"ok")
(er-ffi-test
"list_to_binary out-of-range byte -> error:badarg"
(ffi-nm (ffi-ev "try list_to_binary([300]) catch error:badarg -> ok end"))
"ok")
(er-ffi-test
"list_to_binary non-iolist -> error:badarg"
(ffi-nm (ffi-ev "try list_to_binary(42) catch error:badarg -> ok end"))
"ok")
;; ── Still deferred (no host primitive): httpc (HTTP client, v2),
;; sqlite-* (v2 indexes). Assert NOT registered so a future iteration
;; that wires them without updating this suite fails fast.

View File

@@ -229,13 +229,37 @@
(= ch "$")
(do
(er-advance! 1)
(if
(and (< pos src-len) (= (er-cur) "\\"))
(do
(er-advance! 1)
(when (< pos src-len) (er-advance! 1)))
(when (< pos src-len) (er-advance! 1)))
(er-emit! "integer" (slice src start pos) start)
;; Emit the char's decimal code as the integer token value
;; (was: raw "$X" text — parse-number then returned nil).
(let
((code (cond
(>= pos src-len) 0
(= (er-cur) "\\")
(do
(er-advance! 1)
(let ((esc (if (< pos src-len) (er-cur) "")))
(when (< pos src-len) (er-advance! 1))
(cond
(= esc "n") 10
(= esc "t") 9
(= esc "r") 13
(= esc "s") 32
(= esc "b") 8
(= esc "e") 27
(= esc "f") 12
(= esc "v") 11
(= esc "d") 127
(= esc "0") 0
(= esc "\\") 92
(= esc "\"") 34
(= esc "'") 39
(= esc "") 0
:else (char->integer (nth (string->list esc) 0)))))
:else
(let ((c (er-cur)))
(er-advance! 1)
(char->integer (nth (string->list c) 0))))))
(er-emit! "integer" (str code) start))
(scan!))
(er-lower? ch)
(do

View File

@@ -107,7 +107,12 @@
(let
((ty (get node :type)))
(cond
(= ty "integer") (parse-number (get node :value))
(= ty "integer")
(let ((n (parse-number (get node :value))))
(cond
(= n nil) (error (str "Erlang: invalid integer literal: "
(get node :value)))
:else (truncate n)))
(= ty "float") (parse-number (get node :value))
(= ty "atom") (er-mk-atom (get node :value))
(= ty "string") (get node :value)
@@ -821,16 +826,30 @@
(len (get v :elements))
(error "Erlang: tuple_size: not a tuple")))))
(define er-string->charlist
(fn (s)
(let ((cs (string->list s)) (out (er-mk-nil)))
(for-each
(fn (i)
(set! out (er-mk-cons
(char->integer (nth cs (- (- (len cs) 1) i)))
out)))
(range 0 (len cs)))
out)))
(define
er-bif-atom-to-list
(fn
(vs)
(let
((v (er-bif-arg1 vs "atom_to_list")))
;; Standard Erlang: atom_to_list/1 returns an Erlang charlist
;; (list of integer char codes). Was: SX string of :name —
;; unusable from Erlang-land for [Char|T] / ++ / binary segments.
(if
(er-atom? v)
(get v :name)
(error "Erlang: atom_to_list: not an atom")))))
(er-string->charlist (get v :name))
(raise (er-mk-error-marker (er-mk-atom "badarg")))))))
(define
er-bif-list-to-atom
@@ -838,10 +857,11 @@
(vs)
(let
((v (er-bif-arg1 vs "list_to_atom")))
(if
(= (type-of v) "string")
(er-mk-atom v)
(error "Erlang: list_to_atom: not a string")))))
;; Accept Erlang charlist (cons of ints) or SX string.
(let ((s (er-source-to-string v)))
(cond
(= s nil) (raise (er-mk-error-marker (er-mk-atom "badarg")))
:else (er-mk-atom s))))))
;; ── lists module ─────────────────────────────────────────────────
(define
@@ -1597,10 +1617,12 @@
(vs)
(let
((v (er-bif-arg1 vs "integer_to_list")))
;; Standard Erlang: integer_to_list/1 returns an Erlang charlist
;; (e.g. integer_to_list(42) -> [$4, $2] -> [52, 50]).
(cond
(not (= (type-of v) "number"))
(raise (er-mk-error-marker (er-mk-atom "badarg")))
:else (str v)))))
:else (er-string->charlist (str v))))))
(define
er-bif-list-to-integer
@@ -1608,15 +1630,14 @@
(vs)
(let
((v (er-bif-arg1 vs "list_to_integer")))
(cond
(not (= (type-of v) "string"))
(raise (er-mk-error-marker (er-mk-atom "badarg")))
:else (let
((n (parse-number v)))
(cond
(= n nil)
(raise (er-mk-error-marker (er-mk-atom "badarg")))
:else n))))))
;; Accept Erlang charlist (cons of ints) or SX string.
(let ((s (er-source-to-string v)))
(cond
(= s nil) (raise (er-mk-error-marker (er-mk-atom "badarg")))
:else (let ((n (parse-number s)))
(cond
(= n nil) (raise (er-mk-error-marker (er-mk-atom "badarg")))
:else n)))))))
(define
er-bif-is-function

38
lib/feed/acl.sx Normal file
View File

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

62
lib/feed/aggregate.sx Normal file
View File

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

24
lib/feed/api.sx Normal file
View File

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

125
lib/feed/conformance.sh Executable file
View File

@@ -0,0 +1,125 @@
#!/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 ]

68
lib/feed/content.sx Normal file
View File

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

76
lib/feed/dedupe.sx Normal file
View File

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

114
lib/feed/fanout.sx Normal file
View File

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

60
lib/feed/fed.sx Normal file
View File

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

23
lib/feed/home.sx Normal file
View File

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

44
lib/feed/mute.sx Normal file
View File

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

31
lib/feed/normalize.sx Normal file
View File

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

45
lib/feed/notify.sx Normal file
View File

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

50
lib/feed/page.sx Normal file
View File

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

92
lib/feed/rank.sx Normal file
View File

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

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

@@ -0,0 +1,19 @@
{
"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
}

19
lib/feed/scoreboard.md Normal file
View File

@@ -0,0 +1,19 @@
# 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** |

75
lib/feed/stream.sx Normal file
View File

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

118
lib/feed/tests/basic.sx Normal file
View File

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

85
lib/feed/tests/content.sx Normal file
View File

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

56
lib/feed/tests/dedupe.sx Normal file
View File

@@ -0,0 +1,56 @@
; 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"))

187
lib/feed/tests/fanout.sx Normal file
View File

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

73
lib/feed/tests/home.sx Normal file
View File

@@ -0,0 +1,73 @@
; 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

@@ -0,0 +1,155 @@
; 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"))

68
lib/feed/tests/mute.sx Normal file
View File

@@ -0,0 +1,68 @@
; 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"))

69
lib/feed/tests/notify.sx Normal file
View File

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

86
lib/feed/tests/page.sx Normal file
View File

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

160
lib/feed/tests/rank.sx Normal file
View File

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

49
lib/feed/tests/thread.sx Normal file
View File

@@ -0,0 +1,49 @@
; 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

@@ -0,0 +1,82 @@
; 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"))

59
lib/feed/thread.sx Normal file
View File

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

42
lib/feed/trending.sx Normal file
View File

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

10
lib/persist/api.sx Normal file
View File

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

34
lib/persist/backend.sx Normal file
View File

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

40
lib/persist/batch.sx Normal file
View File

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

66
lib/persist/blob.sx Normal file
View File

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

35
lib/persist/catalog.sx Normal file
View File

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

43
lib/persist/compaction.sx Normal file
View File

@@ -0,0 +1,43 @@
; 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

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

128
lib/persist/conformance.sh Executable file
View File

@@ -0,0 +1,128 @@
#!/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 ]

71
lib/persist/durable.sx Normal file
View File

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

13
lib/persist/event.sx Normal file
View File

@@ -0,0 +1,13 @@
; 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

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

55
lib/persist/global.sx Normal file
View File

@@ -0,0 +1,55 @@
; 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

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

44
lib/persist/kv.sx Normal file
View File

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

43
lib/persist/log.sx Normal file
View File

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

30
lib/persist/project.sx Normal file
View File

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

54
lib/persist/query.sx Normal file
View File

@@ -0,0 +1,54 @@
; 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

@@ -0,0 +1,27 @@
{
"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
}

27
lib/persist/scoreboard.md Normal file
View File

@@ -0,0 +1,27 @@
# 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** |

40
lib/persist/snapshot.sx Normal file
View File

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

21
lib/persist/subscribe.sx Normal file
View File

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

122
lib/persist/tests/batch.sx Normal file
View File

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

112
lib/persist/tests/blob.sx Normal file
View File

@@ -0,0 +1,112 @@
; 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"))

96
lib/persist/tests/cas.sx Normal file
View File

@@ -0,0 +1,96 @@
; 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

@@ -0,0 +1,86 @@
; 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

@@ -0,0 +1,124 @@
; 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

@@ -0,0 +1,96 @@
; 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

@@ -0,0 +1,163 @@
; 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

@@ -0,0 +1,30 @@
; 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

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

123
lib/persist/tests/global.sx Normal file
View File

@@ -0,0 +1,123 @@
; 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

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

86
lib/persist/tests/kv.sx Normal file
View File

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

81
lib/persist/tests/log.sx Normal file
View File

@@ -0,0 +1,81 @@
; 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

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

101
lib/persist/tests/query.sx Normal file
View File

@@ -0,0 +1,101 @@
; 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

@@ -0,0 +1,126 @@
; 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

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

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

115
lib/persist/tests/upcast.sx Normal file
View File

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

105
lib/persist/tests/view.sx Normal file
View File

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

44
lib/persist/upcast.sx Normal file
View File

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

49
lib/persist/view.sx Normal file
View File

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

102
plans/acl-on-sx.md Normal file
View File

@@ -0,0 +1,102 @@
# acl-on-sx: Access Control on Datalog
rose-ash needs fine-grained, explainable, federation-aware access control. Subjects
(users, groups, roles, services) × actions (read, edit, comment, moderate, federate)
× resources (pages, posts, threads, peers). Decisions must come with a trace — not just
permit/deny, but **why**.
Datalog's bottom-up rule engine produces transparent permit/deny chains: the proof tree
is the audit trail. Inheritance over groups + resource hierarchies is recursive Datalog
in one rule. Federation extends naturally — fed-sx replicates ACL facts, peers reason
over the union.
End-state: a Datalog-on-SX layer specifically for ACL, with explanation API, audit log,
and federation extension. Reuses `lib/datalog/` evaluator and term model where possible.
## Status (rolling)
`bash lib/acl/conformance.sh`**0/0** (not yet started)
## Ground rules
- **Scope:** only touch `lib/acl/**` and `plans/acl-on-sx.md`. Do **not** edit `spec/`,
`hosts/`, `shared/`, `lib/datalog/**`, or other `lib/<lang>/`. You may **import**
from `lib/datalog/` (its public API in `lib/datalog/datalog.sx`); do **not** copy or
modify Datalog code.
- **Shared-file issues** go under "Blockers" with a minimal repro; do not fix here.
- **SX files:** use `sx-tree` MCP tools only.
- **Architecture:** thin layer on top of `lib/datalog/`. Define schema, surface API,
audit + federation hooks. The rule engine itself is Datalog's.
- **Watch for shared patterns** going into `lib/guest/` — both acl-sx and mod-sx need
rule-engine plumbing. If you find shared shape, flag it for extraction (don't
extract yet — wait for mod-sx to start).
- **Commits:** one feature per commit. Keep Progress log updated and tick boxes.
## Architecture sketch
```
ACL declarations (SX) User query
│ │
▼ ▼
lib/acl/schema.sx lib/acl/api.sx
— subject sorts — (acl/permit? subj act res)
— resource sorts — (acl/explain subj act res)
— action sorts — (acl/audit subj act res :allowed?)
— fact schema │
│ ▼
▼ lib/acl/engine.sx
lib/acl/facts.sx — builds Datalog query
— actor(id, kind) — invokes lib/datalog/
— resource(id, kind) — extracts proof tree
— member_of(actor, group) │
— child_of(res, parent) ▼
— grant(actor, act, res) lib/acl/audit.sx
— deny (actor, act, res) — persistent decision log
— query API
```
## Phase 1 — Direct grants
- [ ] `lib/acl/schema.sx` — sorts: subject {user, group, role, service}, action,
resource {page, post, thread, peer}
- [ ] `lib/acl/facts.sx``actor`, `resource`, `grant`, `deny` predicates as Datalog
EDB
- [ ] `lib/acl/engine.sx``(permit? subj act res db)` reduces to Datalog query
- [ ] `lib/acl/api.sx` — public `(acl/permit? ...)` taking implicit current db
- [ ] `lib/acl/tests/direct.sx` — 15+ cases: direct grant, missing grant, explicit deny
- [ ] `lib/acl/scoreboard.{json,md}` baseline
- [ ] `lib/acl/conformance.sh` runs the suite
## Phase 2 — Inheritance
- [ ] `member_of(actor, group)` chain — group grants apply to members (transitive)
- [ ] `child_of(res, parent)` chain — parent grants apply to children (transitive)
- [ ] role expansion — role contains list of (action, resource) tuples
- [ ] deny-overrides — explicit deny wins over inherited allow
- [ ] `lib/acl/tests/inherit.sx` — 25+ cases: nested groups, deep resource trees,
conflict resolution, deny precedence
- [ ] document the deny-overrides choice in plan
## Phase 3 — Explanation + audit
- [ ] `(acl/explain subj act res)``{:allowed? T :proof <tree>}`
- [ ] proof tree extracts from Datalog's derivation
- [ ] `lib/acl/audit.sx` — append-only decision log (in-memory + serializer for disk)
- [ ] `(acl/audit-tail n)` for recent decisions
- [ ] `lib/acl/tests/explain.sx` — proof correctness, audit completeness
## Phase 4 — Federation
- [ ] peer trust facts — `peer(addr, kind)`, `trust(peer, level)`
- [ ] delegated grants — `delegate(peer, actor, action, resource)`
- [ ] cross-instance permit chain — query asks local + queries trusted peers via fed-sx
- [ ] revocation propagation — fact retraction across federation
- [ ] `lib/acl/tests/fed.sx` — federated grant chains (mock fed-sx transport in tests)
## Progress log
(loop fills this in)
## Blockers
(loop fills this in)

View File

@@ -0,0 +1,93 @@
# acl-on-sx loop agent (single agent, queue-driven)
Role: iterates `plans/acl-on-sx.md` forever. **First subsystem loop after fed-sx.**
Sits on `lib/datalog/` — rule engine reused, schema/api/audit/federation added on
top. The deliverable isn't "implement Datalog ACL"; it's *also* to surface shared
rule-engine plumbing into `lib/guest/` (the mod-sx loop will be the second consumer,
validating extraction).
```
description: acl-on-sx queue loop
subagent_type: general-purpose
run_in_background: true
isolation: worktree
```
## Prompt
You are the sole background agent working `/root/rose-ash/plans/acl-on-sx.md`.
Isolated worktree, forever, one commit per feature. Push to `origin/loops/acl`
after every commit.
## Restart baseline — check before iterating
1. Read `plans/acl-on-sx.md` — roadmap + Progress log.
2. `ls lib/acl/` — pick up from the most advanced file.
3. If `lib/acl/tests/*.sx` exist, run them via `bash lib/acl/conformance.sh`. Green
before new work.
4. If `lib/acl/scoreboard.md` exists, that's your baseline.
5. Read `lib/datalog/datalog.sx` public API once — that's your substrate.
## The queue
Phase order per `plans/acl-on-sx.md`:
- **Phase 1** — direct grants. Schema, EDB facts, engine, api, 15+ tests
- **Phase 2** — inheritance (member_of, child_of, role expansion, deny-overrides)
- **Phase 3** — explanation + audit (proof tree, audit log)
- **Phase 4** — federation (peer trust, delegation, cross-instance permit chain)
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/acl/**` and `plans/acl-on-sx.md`. Do **not** edit `spec/`,
`hosts/`, `shared/`, other `lib/<lang>/` dirs, `lib/stdlib.sx`, or `lib/` root.
May **import** from `lib/datalog/` only (its public API).
- **NEVER call `sx_build`.** 600s watchdog. If sx_server binary broken → Blockers
entry, stop.
- **Shared-file issues** → plan's Blockers with minimal repro.
- **SX files:** `sx-tree` MCP tools ONLY. `sx_validate` after edits.
- **Worktree:** commit, then push to `origin/loops/acl`. Never touch `main` or
`architecture`.
- **Commit granularity:** one feature per commit. Short factual messages
(`acl: child_of resource inheritance + 8 tests`).
- **Plan file:** update Progress log + tick boxes every commit.
- **Watch for shared infrastructure** with future mod-sx (Prolog moderation). If you
build a generic rule-engine adapter, note it in Progress log so the eventual
`lib/guest/rules/` extraction has both consumers identified.
## ACL-specific gotchas
- **Datalog is bottom-up.** No goal-directed search. Don't reach for cut or
backtracking — that's mod-sx's job. Your decisions emerge from fixpoint.
- **Deny-overrides** is the policy: if both an allow and deny rule fire, deny wins.
Encode this via stratified negation; document the choice clearly in plan.
- **Inheritance termination:** recursive rules with `member_of` chains must
terminate. Datalog guarantees this absent function symbols — don't introduce them
in your schema.
- **Proof tree shape:** Datalog's derivation graph is a DAG, not a tree, when the
same fact is derived multiple ways. For audit, pick one canonical derivation
(shortest, or first); document choice.
- **Federation isn't transitive trust.** A peer's `delegate(...)` fact only applies
if local `trust(peer, level)` covers the action class. Re-check trust on every
query, not at fact-ingestion time.
## 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`.
- `env-bind!` creates a binding; `env-set!` mutates an existing one (walks scope chain).
- `sx_validate` after every structural edit.
- `list?` returns false on raw JS Arrays — host data must be SX-converted.
## Style
- No comments in `.sx` unless non-obvious.
- No new planning docs — update `plans/acl-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

@@ -0,0 +1,99 @@
# feed-on-sx loop agent (single agent, queue-driven)
Role: iterates `plans/feed-on-sx.md` forever. **Activity feeds on APL** — timelines,
notifications, fanout, ranking, all as APL array math on activity vectors. Densest
possible expression of feed composition. Sits on `lib/apl/` (450+/450+ tests
already); adds a feed-shaped vocabulary on top.
```
description: feed-on-sx queue loop
subagent_type: general-purpose
run_in_background: true
isolation: worktree
```
## Prompt
You are the sole background agent working `/root/rose-ash/plans/feed-on-sx.md`.
Isolated worktree, forever, one commit per feature. Push to `origin/loops/feed`
after every commit.
## Restart baseline — check before iterating
1. Read `plans/feed-on-sx.md` — roadmap + Progress log.
2. `ls lib/feed/` — pick up from the most advanced file.
3. If `lib/feed/tests/*.sx` exist, run them via `bash lib/feed/conformance.sh`. Green
before new work.
4. If `lib/feed/scoreboard.md` exists, that's your baseline.
5. Read `lib/apl/apl.sx` public API once — that's your substrate. Familiarize
yourself with at least: ` / ⌽ ↑ ↓ ⌷ ∊ ∘.× /\ ⍋` (you will use all of these).
## The queue
Phase order per `plans/feed-on-sx.md`:
- **Phase 1** — stream model + basic ops (record schema, filter, sort, take)
- **Phase 2** — **THE SHOWCASE**: fanout via outer product. activities `∘.×`
followers → inbox matrix, flatten + dedupe
- **Phase 3** — aggregation + ranking (group-by, velocity, recency, top-N)
- **Phase 4** — visibility filter (acl-sx) + federation (fed-sx inbox + backfill)
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/feed/**` and `plans/feed-on-sx.md`. Do **not** edit `spec/`,
`hosts/`, `shared/`, other `lib/<lang>/` dirs, `lib/stdlib.sx`, or `lib/` root.
May **import** from `lib/apl/` only (its public API).
- **NEVER call `sx_build`.** 600s watchdog. If sx_server binary broken → Blockers
entry, stop.
- **Shared-file issues** → plan's Blockers with minimal repro.
- **SX files:** `sx-tree` MCP tools ONLY. `sx_validate` after edits.
- **Unicode in `.sx`:** raw UTF-8 only, never `\uXXXX` escapes. APL glyphs land
directly in source.
- **Worktree:** commit, then push to `origin/loops/feed`. Never touch `main` or
`architecture`.
- **Commit granularity:** one feature per commit. Short factual messages
(`feed: outer-product fanout + dedupe by (actor,verb,object) + 9 tests`).
- **Plan file:** update Progress log + tick boxes every commit.
## feed-specific gotchas
- **Activities are heterogeneous.** Different verbs carry different shapes
(`:object` might be page-id, post-id, user-id). Don't over-normalize — keep
`:tags` as a flexible bag. APL operations over heterogeneous records work fine
via dict lookups; only the indexed fields need uniform shape.
- **Fanout produces matrices fast.** N activities × M followers → NM items. Apply
filter/dedupe early, not after materialization. Use guard predicates *inside*
the outer product where possible (compose with `∘.{a v ⊢ ...}`).
- **Dedupe key isn't always `(actor,verb,object)`.** For "alice liked X" and "bob
liked X" the dedupe key is `(verb,object)` (collapse the actors into a list).
For "alice posted X" each `:actor` is distinct. Each verb may want its own
dedupe rule; codify these in `lib/feed/dedupe.sx`.
- **Recency decay matters more than score precision.** Use a simple half-life decay
(e.g. score × 0.5^(age/window)) rather than a clever curve. Calibrate the
window via tests, not theory.
- **Ranking should be deterministic on ties.** Always include a tiebreaker (id, or
hash). Otherwise tests will flake.
- **The ACL filter is per-viewer.** A timeline is computed *for* a user; the same
candidate stream produces different timelines for different viewers. Don't
cache pre-ACL timelines.
## 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`.
- `env-bind!` creates a binding; `env-set!` mutates an existing one (walks scope chain).
- `sx_validate` after every structural edit.
- `list?` returns false on raw JS Arrays — host data must be SX-converted.
## Style
- No comments in `.sx` unless non-obvious.
- No new planning docs — update `plans/feed-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

@@ -0,0 +1,98 @@
# flow-on-sx loop agent (single agent, queue-driven)
Role: iterates `plans/flow-on-sx.md` forever. **Durable workflows on Scheme** — the
call/cc + delimited continuation showcase that justifies pulling R7RS into
production. art-dag's natural successor: DAG-of-tasks with pause/resume across
process restarts. fed-sx extension turns local flows into distributed ones.
```
description: flow-on-sx queue loop
subagent_type: general-purpose
run_in_background: true
isolation: worktree
```
## Prompt
You are the sole background agent working `/root/rose-ash/plans/flow-on-sx.md`.
Isolated worktree, forever, one commit per feature. Push to `origin/loops/flow`
after every commit.
## Restart baseline — check before iterating
1. Read `plans/flow-on-sx.md` — roadmap + Progress log.
2. `ls lib/flow/` — pick up from the most advanced file.
3. If `lib/flow/tests/*.sx` exist, run them via `bash lib/flow/conformance.sh`. Green
before new work.
4. If `lib/flow/scoreboard.md` exists, that's your baseline.
5. Read `lib/scheme/scheme.sx` public API once — that's your substrate.
## The queue
Phase order per `plans/flow-on-sx.md`:
- **Phase 1** — declarative DAG: `defflow`, `sequence`, `parallel`, sync runtime,
basic api
- **Phase 2** — control flow + error handling: `cond`, `retry`, `timeout`,
`try-catch`
- **Phase 3** — **THE SHOWCASE**: `suspend`/`resume` via `call/cc`, persistent
store, crash recovery
- **Phase 4** — distributed nodes via fed-sx (remote-node, handoff, replication)
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/flow/**` and `plans/flow-on-sx.md`. Do **not** edit `spec/`,
`hosts/`, `shared/`, other `lib/<lang>/` dirs, `lib/stdlib.sx`, or `lib/` root.
May **import** from `lib/scheme/` only (its public API).
- **NEVER call `sx_build`.** 600s watchdog. If sx_server binary broken → Blockers
entry, stop.
- **Shared-file issues** → plan's Blockers with minimal repro.
- **SX files:** `sx-tree` MCP tools ONLY. `sx_validate` after edits.
- **Worktree:** commit, then push to `origin/loops/flow`. Never touch `main` or
`architecture`.
- **Commit granularity:** one feature per commit. Short factual messages
(`flow: retry combinator with exponential backoff + 6 tests`).
- **Plan file:** update Progress log + tick boxes every commit.
## flow-specific gotchas
- **Continuations must be re-entrant.** Phase 3's `suspend` captures a continuation
that may be re-entered after a process restart. That means: no captured file
descriptors, no captured sockets, no captured live runtime references that won't
survive serialization. State referenced by the continuation must be plain SX data
or live in the flow store.
- **call/cc, not call-with-escape-continuation.** R7RS distinguishes. Use the full
call/cc for resume; escape-only continuations cannot be re-entered. Read
`lib/scheme/r7rs.md` (or equivalent) to confirm semantics.
- **`parallel` in Phase 1 is sequential.** Don't try threading until Phase 3+. Just
evaluate branches in order, collect results, return joined value. Document the
semantics clearly so users don't assume true concurrency.
- **Retry doesn't retry continuations.** If a node has already suspended, retry on
resume doesn't re-run it from scratch — it resumes. `retry` only applies to
exceptions raised before suspend. Be explicit in the API.
- **Cancellation invalidates the continuation.** `(flow/cancel id)` must remove the
stored continuation so a stale `resume` cannot wake it. Document semantics.
- **Timeouts in pure SX are tricky.** Without a scheduler, `timeout` is a budget on
step count or wall-clock probed at safe points. Pick one approach (probably step
budget for determinism) and document.
## 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`.
- `env-bind!` creates a binding; `env-set!` mutates an existing one (walks scope chain).
- `sx_validate` after every structural edit.
- `list?` returns false on raw JS Arrays — host data must be SX-converted.
## Style
- No comments in `.sx` unless non-obvious.
- No new planning docs — update `plans/flow-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

@@ -0,0 +1,106 @@
# kernel-on-sx loop agent (single agent, queue-driven)
Role: iterates `plans/kernel-on-sx.md` forever. **First chisel of the Phase B stratification work** — natural successor to env-as-value, validates SX's reflection story (first-class environments, evaluators, operatives). Goal isn't just "implement Kernel"; it's *also* to surface common patterns into `lib/guest/` (specifically motivating a future `lib/guest/reflective/` sub-layer). One feature per commit.
```
description: kernel-on-sx queue loop
subagent_type: general-purpose
run_in_background: true
isolation: worktree
```
## DO NOT START WITHOUT THE PREREQUISITES
This loop **must not** start until the lib-guest core kits are in place. Kernel's parser consumes `lib/guest/core/lex.sx` and `lib/guest/core/pratt.sx` (s-expression-shaped, minimal demand); its evaluator's pattern dispatch consumes `lib/guest/core/match.sx`.
**Pre-flight check:**
```
ls /root/rose-ash/lib/guest/lex.sx /root/rose-ash/lib/guest/pratt.sx \
/root/rose-ash/lib/guest/match.sx /root/rose-ash/lib/guest/ast.sx
```
If any of those `lib/guest/*.sx` files are missing, **stop and report**. Do not start.
## Prompt
You are the sole background agent working `/root/rose-ash/plans/kernel-on-sx.md`. You run in an isolated git worktree on branch `loops/kernel`. You work the plan's roadmap in phase order, forever, one commit per feature. Push to `origin/loops/kernel` after every commit.
## Restart baseline — check before iterating
1. Read `plans/kernel-on-sx.md` — Roadmap + Progress log + Blockers tell you where you are.
2. Run the pre-flight check above. If any lib/guest kit is missing, stop immediately and update the plan's Blockers section.
3. `ls lib/kernel/` — pick up from the most advanced file that exists. If the directory does not exist, you are at Phase 1.
4. If `lib/kernel/tests/*.sx` exist, run them via the epoch protocol against `sx_server.exe`. They must be green before new work.
## The queue
Phase order per `plans/kernel-on-sx.md`:
- **Phase 1** — Parser (s-expression reader, minimal — consumes `lib/guest/lex` + `lib/guest/pratt`)
- **Phase 2** — Core evaluator with first-class environments
- **Phase 3** — `$vau` / `$lambda` / `wrap` / `unwrap` (the operativeapplicative distinction)
- **Phase 4** — Standard environment construction
- **Phase 5** — Encapsulations (Kernel's opaque-type idiom)
- **Phase 6** — Hygienic operatives (Shutt's later work — operatives that don't capture)
- **Phase 7** — Propose `lib/guest/reflective/` (extraction phase — see chiselling discipline)
Within a phase, pick the checkbox with the best tests-per-effort ratio.
Every iteration: implement → test → commit → tick `[ ]` in plan → append Progress log → push → next.
## Lib/guest chiselling discipline (the defining feature of this loop)
You are not just implementing Kernel — you are *chiselling* the substrate to surface what `lib/guest/reflective/` should contain. Every commit must end with a one-line **"chisel note"** appended to the plan's Progress log entry, in this format:
```
chisel: <one of: consumes-X | shapes-reflective | proposes-Y | nothing>
```
- `consumes-X` — this commit used an existing `lib/guest/X` kit (e.g., `consumes-pratt`, `consumes-match`).
- `shapes-reflective` — this commit revealed something about what `lib/guest/reflective/` should look like (e.g., env-reification helper signatures, applicative-vs-operative dispatch protocol). Add a paragraph to the plan's "lib/guest feedback loop" section describing the insight.
- `proposes-Y` — this commit revealed a gap in another existing kit (e.g., `match.sx` doesn't quite handle X). Open a Blockers entry describing the gap.
- `nothing` — pure Kernel work that didn't touch the substrate or lib/guest story (rare; if you write this twice in a row, stop and reflect on why).
**Phase 7 (extraction)** is **gated** by the two-consumer rule. Kernel alone is one consumer. The natural second consumer is a future MetaScheme port, a Common-Lisp meta-evaluator port, or a Kernel dialect (cKanren-style). **Until a second consumer exists, do NOT actually extract** — instead, mark Phase 7 `[partial — pending second consumer]` and document the proposed `lib/guest/reflective/` API surface in the plan's progress log. The extraction itself happens later, when a second consumer materialises.
This discipline is the point of the loop, not a bookkeeping tax. The chisel notes are what tell us — at the end of Kernel's run — whether a `lib/guest/reflective/` sub-layer is real or just one-language-shaped.
## Ground rules (hard)
- **Scope:** only `lib/kernel/**` and `plans/kernel-on-sx.md`. Do **not** edit `spec/`, `hosts/`, `shared/`, `lib/guest/**` (read-only consumer at this phase), or other `lib/<lang>/`.
- **Consume `lib/guest/core/`** wherever it covers a need. Hand-rolling defeats the chiselling goal.
- **Do not extract into `lib/guest/reflective/` from this loop.** That's Phase 7 territory, gated by the two-consumer rule. Until there's a second consumer, document the API surface only.
- **Substrate gaps** (env-as-value not exposing X, `eval` semantics drift, JIT not handling reflective patterns) → Blockers entry with minimal repro. Do **not** fix substrate from this loop. Substrate work belongs to `sx-improvements.md` / `jit-perf-regression.md`.
- **NEVER call `sx_build`.** 600s watchdog will kill you. If `sx_server.exe` is broken, add a Blockers entry and stop.
- **SX files:** `sx-tree` MCP tools ONLY. `sx_validate` after every edit. Never `Edit`/`Read`/`Write` on `.sx`.
- **Worktree:** commit, then push to `origin/loops/kernel`. Never touch `main`. Never push to `architecture`.
- **Commit granularity:** one feature per commit. Short factual messages: `kernel: $vau operative + 6 tests`.
- **Plan file:** update Progress log + tick boxes every commit. Include the chisel note.
- **If blocked** for two iterations on the same issue, add to Blockers and move on.
## Kernel-specific gotchas
- **Operatives don't evaluate their arguments.** `$vau` builds an operative; the body sees the *unevaluated* argument expressions plus the dynamic environment. This is the opposite of every other guest in the set. `(define-via-vau)` builds a binding by calling `eval` inside the body on the (still-syntax) argument.
- **Applicatives wrap operatives.** `(wrap op)` produces an applicative that evaluates its args first, then calls `op` with the values. `$lambda` is sugar for `wrap``$vau`.
- **Dynamic vs static environments.** Operative body sees both: the static env where the `$vau` was created (closure-style), AND the dynamic env where the call happens (passed as the env-param). Different from lexical-only languages.
- **No special forms in the evaluator.** `$if`, `$define!`, `$lambda` are all just operatives bound in the standard environment. The evaluator is `lookup-and-call` — no hardcoded switch on symbols. This is the whole point: the language is reified as data.
- **`eval` is a primitive callable on user environments.** This is where SX's env-as-value matters most. If env-as-value isn't fully landed in the substrate, this is where it'll break.
- **Encapsulations (Phase 5) are Kernel's opaque-types idiom.** `make-encapsulation-type` returns three operatives: encapsulator (constructs), predicate (tests), decapsulator (extracts). Used to define promises, streams, modules.
- **Hygienic operatives (Phase 6) are research-grade.** Shutt's later work. Operatives that don't accidentally capture caller bindings. Likely uses scope sets / frame stamps. Treat as exploration, not implementation-deadline.
## 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`.
- `env-bind!` creates a binding; `env-set!` mutates an existing one (walks scope chain).
- `sx_validate` after every structural edit.
- `list?` returns false on raw JS Arrays — host data must be SX-converted.
- Shell heredoc `||` gets eaten — escape or use `case`.
## Style
- No comments in `.sx` unless non-obvious.
- No new planning docs — update `plans/kernel-on-sx.md` inline.
- Short, factual commit messages with chisel note: `kernel: $vau operative + 6 tests [shapes-reflective]`.
- One feature per iteration. Commit. Log. Push. Next.
Go. Run the pre-flight check. If lib/guest kits are missing, stop. Otherwise read the plan, find the first unchecked `[ ]`, implement it. Remember: every commit ends with a chisel note, and Phase 7 extraction waits for a second consumer.

View File

@@ -0,0 +1,115 @@
# 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.

82
plans/commerce-on-sx.md Normal file
View File

@@ -0,0 +1,82 @@
# 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)

82
plans/content-on-sx.md Normal file
View File

@@ -0,0 +1,82 @@
# content-on-sx: Documents, blocks & collaborative editing on Smalltalk
> **DRAFT outline.** The CMS vertical — blog, WYSIWYG editor, Ghost sync. Depends
> on `persist-on-sx` (document history as an event log). Ghost/CMS sync stays a thin
> external adapter (Python/FFI) until a native replacement exists.
rose-ash's `blog` domain is content management: a block-based WYSIWYG editor,
navigation, Ghost CMS sync. A document is a tree of live blocks; editing is a
stream of operations; collaboration needs conflict-free merge. That is an object
model — blocks are objects, edits are messages, and a document is the object graph
responding to them. Smalltalk's "everything is an object responding to messages"
maps directly to a block/WYSIWYG model, and a semilattice (CRDT) merge keeps
concurrent edits conflict-free.
End-state: a Smalltalk-on-SX document model (typed blocks, structural ops),
operation log + CRDT merge for collaborative editing, versioning/history via the
event store, and a render boundary to HTML/SX. External CMS (Ghost) sync is an
injected adapter, not core.
## Status (rolling)
`bash lib/content/conformance.sh`**0/0** (not yet started)
## Ground rules
- **Scope:** only `lib/content/**` and `plans/content-on-sx.md`. May **import**
from `lib/smalltalk/`, and (once it exists) `lib/persist/`. Do not edit substrates.
- **Architecture:** a document is an ordered tree of blocks (objects); an edit is a
message (`insert`/`update`/`move`/`delete`); concurrent edits merge via a
commutative (CRDT/semilattice) operation so order doesn't matter. History is the
`persist` event stream; any version is a replay.
- **Determinism:** merge must be commutative + idempotent (test: apply ops in any
order / twice → same document).
- **Commits:** one feature per commit. Progress log + tick boxes.
## Architecture sketch
```
Edit op Rendered document
(insert block after id) ... HTML / SX tree
│ ▲
▼ │
lib/content/block.sx lib/content/render.sx
— typed blocks as objects — block tree → HTML/SX
— heading/text/image/embed — (reuses SX render boundary)
│ ▲
▼ │
lib/content/doc.sx lib/content/merge.sx
— ordered block tree — CRDT/semilattice op merge
— apply op, structural moves — concurrent-edit reconciliation
│ ▲
▼ │
lib/content/api.sx ── (content/edit) (content/render) (content/history) ──┐
│ │
├── op log + versions → persist │
└── Ghost/CMS sync → injected external adapter (thin, non-core) ──┘
```
## Phase 1 — Block document model
- [ ] `block.sx` — typed block objects
- [ ] `doc.sx` — ordered tree, apply edit op, structural moves
- [ ] `render.sx` — block tree → HTML/SX
- [ ] `api.sx` + tests + scoreboard + conformance.sh
## Phase 2 — Op log + versioning
- [ ] edit ops as `persist` events; replay to any version
- [ ] `(content/history doc)`, diff between versions
## Phase 3 — Collaborative merge (CRDT)
- [ ] commutative/idempotent op merge
- [ ] concurrent-edit tests (any order, double-apply → identical)
## Phase 4 — External sync + federation
- [ ] Ghost/CMS sync via injected adapter (import/export)
- [ ] federated documents (peer-authored blocks) — trust-gated stub
- [ ] tests: round-trip import/export, conflict on concurrent external edit
## Progress log
(loop fills this in)
## Blockers
(loop fills this in)

81
plans/events-on-sx.md Normal file
View File

@@ -0,0 +1,81 @@
# events-on-sx: Calendar, ticketing & notification delivery on Datalog
> **DRAFT outline.** The events vertical + the shared notification-delivery edge.
> Depends on `persist-on-sx` (bookings ledger) and `flow-on-sx` (reminders, retrying
> delivery). Pairs with `commerce-on-sx` for paid tickets.
rose-ash's `events` domain is calendar + ticketing: recurring events, availability,
capacity, bookings. Scheduling is constraint reasoning — "is this slot free given
recurrence, capacity, and the attendee's other bookings?" — which is rule
evaluation over facts. Datalog expresses availability, recurrence expansion, and
capacity as rules; a booking is a transaction; reminders and digests are durable
`flow`s. Notification *delivery* (email/push) — needed here and by `feed/notify`
is folded in as an injected transport, extractable later.
End-state: a Datalog-on-SX events layer with recurrence expansion, availability +
capacity rules, transactional booking, and a flow-driven notification dispatcher
(reminders, digests, retries) over an injected transport.
## Status (rolling)
`bash lib/events/conformance.sh`**0/0** (not yet started)
## Ground rules
- **Scope:** only `lib/events/**` and `plans/events-on-sx.md`. May **import** from
`lib/datalog/`, and (once they exist) `lib/persist/` + `lib/flow/`. Do not edit
substrates.
- **Architecture:** events/availability/capacity are Datalog facts + rules;
recurrence expands to occurrence facts within a window; a booking checks rules
then appends a `persist` event (idempotent, capacity-safe). Notifications are flows
that suspend on transport IO and retry on failure.
- **Determinism:** recurrence expansion + availability must be reproducible for a
fixed window + ruleset; capacity checks must be race-safe (no overbooking).
- **Commits:** one feature per commit. Progress log + tick boxes.
## Architecture sketch
```
Event + booking Result
event(id,start,rrule,capacity) {:booked | :full | :conflict} + reminders
│ ▲
▼ │
lib/events/calendar.sx lib/events/availability.sx
— event facts, recurrence (RRULE) — free/busy + capacity rules (Datalog)
— expand occurrences in window │
│ ▲
▼ │
lib/events/booking.sx lib/events/notify.sx (flow)
— transactional, capacity-safe — reminders / digests, retry on fail
— bookings → persist ledger — injected transport (email/push)
│ │
▼ ▼
lib/events/api.sx ── (events/schedule) (events/book) (events/agenda) ──────┘
```
## Phase 1 — Calendar + recurrence
- [ ] `calendar.sx` — event facts, RRULE expansion in a window
- [ ] `availability.sx` — free/busy rules
- [ ] `api.sx` + tests + scoreboard + conformance.sh
## Phase 2 — Ticketing + booking
- [ ] capacity rules; transactional booking → `persist` (no overbooking)
- [ ] paid tickets compose with `commerce` order flow
- [ ] tests: capacity edge, double-book guard, conflict detection
## Phase 3 — Notification delivery (flow)
- [ ] `notify.sx` — reminder/digest flows over injected transport
- [ ] retry/backoff on transport failure (flow suspend/resume)
- [ ] tests: delivery success, retry path, idempotent re-send
- [ ] NOTE: shared with `feed/notify` — candidate for later extraction to a
`delivery-on-sx` once a second consumer is real
## Phase 4 — Federation
- [ ] cross-instance events (peer calendar) — trust-gated stub
- [ ] tests: federated agenda merge
## Progress log
(loop fills this in)
## Blockers
(loop fills this in)

176
plans/feed-on-sx.md Normal file
View File

@@ -0,0 +1,176 @@
# feed-on-sx: Activity Feeds on APL
Timelines, notifications, activity aggregation. The math is array math: filter, sort,
reduce, scan, outer product. APL is the densest possible expression of feed
composition — a fanout-and-rank pipeline reads as a single line.
rose-ash needs: per-user home timeline, notification feed, activity stream digestion,
backfill for new follows, deduplication across cross-posts. Every operation is an
array-shaped transformation.
End-state: an APL-flavored layer on `lib/apl/` with feed-specific combinators
(`fanout`, `dedupe`, `score`, `rank`), an SX adapter for callers who don't want raw
APL, ACL visibility filtering via `lib/acl/`, federation via fed-sx.
## Status (rolling)
`bash lib/feed/conformance.sh`**189/189** (Phases 14 + TF-IDF, notifications, home, smart-dedupe, trending, mute, pagination, threading)
## Ground rules
- **Scope:** only touch `lib/feed/**` and `plans/feed-on-sx.md`. Do **not** edit
`spec/`, `hosts/`, `shared/`, `lib/apl/**`, or other `lib/<lang>/`. You may
**import** from `lib/apl/` (public API in `lib/apl/apl.sx`); do **not** modify APL.
- **Shared-file issues** go under "Blockers" with a minimal repro; do not fix here.
- **SX files:** use `sx-tree` MCP tools only.
- **Architecture:** an activity is a small dict (`{:actor :verb :object :at :tags}`); a
stream is an APL vector of such dicts. Operations are APL primitives lifted onto
this shape. SX adapter exposes ergonomic API to non-APL callers.
- **Unicode:** raw UTF-8 in `.sx` files. APL glyphs land directly.
- **Commits:** one feature per commit. Keep Progress log updated and tick boxes.
## Architecture sketch
```
Raw activities (any shape) Per-user view
│ ▲
▼ │
lib/feed/normalize.sx lib/feed/timeline.sx
— {:actor :verb :object — (timeline user)
:at :tags} record — applies filter ∘ rank ∘ take
│ ▲
▼ │
lib/feed/stream.sx lib/feed/rank.sx
— APL vector of activities — velocity, recency
— filter, sort, take — TF-IDF-ish over :tags
│ ▲
▼ │
lib/feed/fanout.sx lib/feed/dedupe.sx
— followers vector — group by :object
— activities ∘.× followers — collapse cross-posts
— flatten + dedupe
lib/feed/api.sx lib/feed/fed.sx
— (feed/post activity) — inbox via fed-sx
— (feed/timeline user) — backfill on subscribe
— (feed/notify user)
```
## Phase 1 — Stream model + basic ops
- [x] `lib/feed/normalize.sx` — activity record schema; coerce arbitrary inputs
- [x] `lib/feed/stream.sx` — APL vector representation; filter by predicate; sort by
`:at`; take N (`↑`); reverse (`⌽`)
- [x] `lib/feed/api.sx``(feed/post activity)`, `(feed/all)`
- [x] `lib/feed/tests/basic.sx` — 30 cases: normalize defaults, filter, sort, take, api
- [x] `lib/feed/scoreboard.{json,md}`
- [x] `lib/feed/conformance.sh`
## Phase 2 — Fanout via outer product
- [x] follower graph: `followers user → vector of user ids` (`feed/follow-graph`,
`feed/followers`; graph = `{followee -> (followers)}` dict)
- [x] fanout: activities `∘.×` audience → matrix via `apl-outer feed/-mk-event`
- [x] flatten to inbox events vector (`feed/-flatten` rank-2 → rank-1)
- [x] dedupe — `feed/dedupe-inbox` by `(to, actor, verb, object)`; also
`feed/dedupe-activities` `(actor verb object)` and `feed/dedupe-collapse`
`(verb object)` for cross-actor likes
- [x] `lib/feed/tests/fanout.sx` — 29 cases: small graph, mutual follow, star
(high-fanout), empty graph, unfollowed actor, cross-post dedupe
## Phase 3 — Aggregation + ranking
- [x] group-by — `feed/group-by`/`feed/group-count` key-reduce; `feed/by-actor-day`
buckets `(actor, day)` via `feed/day` (string-joined keys)
- [x] velocity score — `feed/velocity` counts actor's activities in `(at-window, at]`
- [x] recency score — `feed/recency` half-life decay `0.5^(age/hl)`
- [x] composite rank — `feed/composite` weighted sum of `(weight scorer)` parts
- [x] top-N per timeline — `feed/top` = rank then take
- [x] `lib/feed/tests/rank.sx` — 24 cases: decay shape, velocity burst, stable
tie-break, top-N, composite
## Phase 4 — Visibility filter + federation
`lib/acl/` and fed-sx don't exist yet and are out of scope (import `lib/apl/`
only), so ACL/transport are injected: `permit?`, `remote?`, `send-fn`, `fetch-fn`
are function parameters. Real acl-sx / fed-sx wire in at the call site unchanged.
- [x] ACL filter — `feed/visible stream viewer permit?`; default `feed/permit-acl?`
reads `:visible-to` allowlist (+ author-sees-own); per-viewer, never cached
- [x] fed-sx outbound — `feed/federate`/`feed/deliver` fan out then partition
local vs remote inboxes; remote events handed to injected `send-fn`
- [x] fed-sx inbound — `feed/inbound` normalizes + `feed/ingest` dedupes peer
activities into the local stream
- [x] backfill on subscribe — `feed/backfill local fetch-fn peer-id`
- [x] `lib/feed/tests/integration.sx` — 22 cases incl. end-to-end
`feed/timeline` (federated → ACL for viewer → recency rank → top-N)
## Progress log
- **Phase 1 done (30/30).** Stream = APL rank-1 array whose ravel holds activity
dicts. `normalize.sx` (record schema + accessors), `stream.sx` (filter via `/`
compress, sort via `⍋` grade-up [stable], take via `↑`, reverse via `⌽`,
by-actor/verb/object/since predicates), `api.sx` (mutable log: post/all/reset!/size).
Substrate: `apl-compress`, `apl-grade-up`, `apl-take`, `apl-reverse`, `make-array`.
Grade-up returns 1-based indices (⎕IO=1), is stable on ties → deterministic sort.
- **Phase 2 done (59/59 total).** `fanout.sx` (graph + `apl-outer` showcase),
`dedupe.sx` (per-key dedupe, first-wins stable). Key APL gotcha: `scalar?` is
true for ANY dict and `disclose` nils a non-array dict, so an apl-outer combiner
MUST `enclose` its event dict — apl-outer discloses it back intact. `apl-unique`
preserves first-occurrence order; dict `keys` order is NOT stable, so
`feed/audience` sorts (else recipient ordering flakes). `apl-compress` needs a
rank-1 array, so the (activity×follower) matrix is flattened to its ravel before
the edge-guard filter.
- **Phase 3 done (83/83 total).** `aggregate.sx` (group-by/count, day buckets) +
`rank.sx` (recency/velocity/engagement scorers, composite, top-N). `sort` is
single-arg ascending only — no comparator — so ranking uses a stable two-pass
`apl-grade-down` (by :at desc, then by score desc) for deterministic tie-breaks.
Dict keys must be strings, so composite group keys are string-joined ("actor#day").
- **Phase 4 done (105/105 total).** `acl.sx` (per-viewer `feed/visible`,
`feed/timeline` capstone) + `fed.sx` (merge/ingest/inbound/backfill/federate/
deliver). ACL/transport are dependency-injected (permit?/remote?/send-fn/fetch-fn)
since lib/acl + fed-sx don't exist. `feed/normalize` now MERGEs defaults over the
raw dict (was projecting to 5 keys) so extra metadata (:visible-to, peer fields)
survives — matches the "flexible bag" principle.
## Roadmap is complete (all 4 phases). Possible follow-ups:
- Wire real acl-sx once `lib/acl/` exists (swap injected `permit?`).
- Wire real fed-sx transport (swap `send-fn`/`fetch-fn`).
- [x] TF-IDF over `:tags` for content ranking — `content.sx`: `feed/tag-df`,
`feed/tag-idf` (log N/df), `feed/tfidf-score`, `feed/by-relevance`; 15 tests.
Composes as a scorer with rank.sx. (120/120 total.)
- [x] Notification feed (verb-filtered, per-recipient) — `notify.sx`:
`feed/notifications`, `feed/notify-verbs`, `feed/notify-digest` (collapses
"X, Y liked Z" by (verb,object), sorted-deterministic); 8 tests. (128/128 total.)
- [x] **Capstone** `feed/home` — the whole pipeline as one line: fanout ∘ inbox ∘
dedupe ∘ ACL ∘ rank ∘ take (`home.sx`); 6 tests incl. per-viewer ACL + cross-post
dedupe. (134/134 total.)
- [x] Per-verb dedupe rules (briefing gotcha #3) — `feed/dedupe-smart` /
`feed/smart-key`: reactions (like/follow/boost/...) collapse cross-actor on
(verb,object); posts stay distinct per actor. `feed/collapse-verbs` is
rebindable policy; 9 tests. (143/143 total.)
- [x] Trending — `feed/trending` / `feed/trending-actors`: objects/actors ranked
by activity count in a recency window, count-desc with key-asc tiebreak
(`trending.sx`); 11 tests. (154/154 total.)
- [x] Mute/block — `feed/mute-actors` / `feed/mute-tags` / `feed/mute-objects` /
`feed/apply-prefs`: viewer-controlled per-request filtering (complements ACL's
author-controlled visibility) (`mute.sx`); 9 tests. (163/163 total.)
- [x] Pagination — `feed/page`/`feed/page-count` (offset) + `feed/before`/
`feed/after`/`feed/page-before`/`feed/next-cursor` (cursor by :at, stable under
inserts) (`page.sx`); 14 tests. (177/177 total.)
- [x] Threading — `feed/replies`/`feed/reply-count`/`feed/thread`/
`feed/thread-objects`/`feed/thread-size`: conversation closure over `:reply-to`
(transitive fixpoint), chronological (`thread.sx`); 12 tests. (189/189 total.)
(none)
## Notes for next iteration
- sx-tree MCP tools take `file:` NOT `path:` (CLAUDE.md is stale). Wrong key →
`Yojson Type_error("Expected string, got null")`. Looks like a broken binary, isn't.
- sx_server binary lives in main repo: `/root/rose-ash/hosts/ocaml/_build/default/bin/sx_server.exe`
(worktree has no `_build`). conformance.sh already points there with relative fallback.
- Phase 2 substrate verified available: `apl-outer` (∘.×), `apl-member` (∊),
`apl-unique`, `apl-iota` (1-based).

108
plans/flow-on-sx.md Normal file
View File

@@ -0,0 +1,108 @@
# flow-on-sx: Durable DAG Workflows on Scheme
rose-ash needs workflows that survive restarts: content pipelines (write → review →
publish → federate), scheduled jobs (digest emails), multi-step user flows (signup,
confirm, onboard). art-dag is the precedent — DAG-of-tasks with pause/resume at IO
boundaries.
Scheme's `call/cc` + delimited continuations make pause/resume natural: a `suspend`
captures the continuation, serializes it as part of the flow record, and `resume`
re-enters at exactly that point. No state-machine bookkeeping by hand. R7RS-small is
already at 2644/2644 (see kernel/architecture status).
End-state: a Scheme-on-SX layer over the existing scheme runtime, with combinators
for sequence/parallel/branch/retry/timeout/suspend, persistent flow store, and a
federation extension via fed-sx for remote-node execution.
## Status (rolling)
`bash lib/flow/conformance.sh`**0/0** (not yet started)
## Ground rules
- **Scope:** only touch `lib/flow/**` and `plans/flow-on-sx.md`. Do **not** edit
`spec/`, `hosts/`, `shared/`, `lib/scheme/**`, or other `lib/<lang>/`. You may
**import** from `lib/scheme/` (public API via `lib/scheme/scheme.sx`); do **not**
modify Scheme.
- **Shared-file issues** go under "Blockers" with a minimal repro; do not fix here.
- **SX files:** use `sx-tree` MCP tools only.
- **Architecture:** flow combinators are Scheme macros + procedures. Runtime is a
driver loop that walks the flow graph and invokes `call/cc` at `suspend` points.
Persistence layer serializes the continuation + open file/socket placeholders are
forbidden (continuations must be resumable across process restart).
- **art-dag awareness:** read `plans/art-dag*` if it exists for design lineage; do not
import code.
- **Commits:** one feature per commit. Keep Progress log updated and tick boxes.
## Architecture sketch
```
(defflow publish
(sequence
(write-content)
(parallel
(review)
(spell-check))
(cond approved?
(sequence (publish) (federate))
(notify-author))))
lib/flow/spec.sx lib/flow/runtime.sx lib/flow/store.sx
— defflow — driver loop — append-only flow log
— sequence/parallel — node dispatch — checkpoint serialize
— cond/retry/timeout — call/cc at suspend — restart loader
— suspend/resume │ │
▼ ▼
lib/flow/api.sx lib/flow/remote.sx
— (flow/start name args) — fed-sx adapter
— (flow/resume id value) — node-on-peer execution
— (flow/cancel id) — failure handling
```
## Phase 1 — Declarative DAG + sequential execution
- [ ] `lib/flow/spec.sx``defflow` macro, `sequence` combinator
- [ ] node = Scheme thunk; output threads to next node (data flow)
- [ ] `parallel` combinator (sequential semantics for now — TRUE parallelism in Phase 3)
- [ ] runtime executes a flow synchronously, returns final value
- [ ] `lib/flow/api.sx``(flow/start name args)` entry point
- [ ] `lib/flow/tests/basic.sx` — 15+ cases: linear sequence, nested sequences,
data flow between nodes, parallel-with-join
- [ ] `lib/flow/scoreboard.{json,md}`
- [ ] `lib/flow/conformance.sh`
## Phase 2 — Control flow + error handling
- [ ] `cond` combinator — predicate selects branch
- [ ] `retry n [backoff]` — re-runs node up to n times on exception
- [ ] `timeout ms` — bounds node execution
- [ ] `try-catch` — exception handler with reified error
- [ ] error model — exceptions vs explicit `(fail :reason ...)` results
- [ ] `lib/flow/tests/control.sx` — 25+ cases: each combinator + composition
## Phase 3 — Suspend / resume (the showcase)
- [ ] `(suspend reason)``call/cc` captures continuation, returns flow-id to caller
- [ ] `lib/flow/store.sx` — serialize flow state (continuation + open vars)
- [ ] `(flow/resume id value)` — load continuation, inject value, re-enter
- [ ] `(flow/cancel id)` — explicit termination
- [ ] crash recovery — on restart, scan store for paused flows, mark resumable
- [ ] `lib/flow/tests/suspend.sx` — pause-resume scenarios, cancellation, "restart"
scenarios (simulated by re-loading store)
## Phase 4 — Distributed nodes via fed-sx
- [ ] `(remote-node addr fn args)` — execute node on a federation peer
- [ ] failure semantics — retry on different peer, fall through to local
- [ ] persistence across instances — flow state replicates via fed-sx
- [ ] handoff — flow started here can resume on a peer if the local instance is down
- [ ] `lib/flow/tests/distributed.sx` — federated flow scenarios (mock fed-sx in tests)
## Progress log
(loop fills this in)
## Blockers
(loop fills this in)

100
plans/host-on-sx.md Normal file
View File

@@ -0,0 +1,100 @@
# host-on-sx: The SX web host — off Quart, onto the kernel (Dream-bound)
> **DRAFT outline.** The integration boundary that turns the subsystem libraries
> into running services, and the strangler path off Python/Quart. This is the
> dependency hub — it imports every subsystem. Decision recorded below: native
> server + SXTP **now**, `dream-on-sx` framework layer **next**, Python only at the
> external-integration edges.
The subsystems (`feed`, `search`, `acl`, `mod`, `flow`, `commerce`, `identity`,
`content`, `events`) are libraries. Something has to receive an HTTP request, route
it, call the right subsystem, and serialize the response. Today that's Python/Quart
— the one large non-SX component in the stack: separate runtime, deploy, and
failure mode. The goal is to move the web/host/domain layer onto the SX substrate
and retire Quart, **incrementally (strangler-fig), never big-bang.**
This is already underway: a native OCaml HTTP server is live in prod on
`sx.rose-ash.com` (~3ms cached, ~323 req/s, ~2MB RSS), `defhandler`/`defpage`
exist, and a partial **SXTP** protocol is specced. That is the unblocked near-term
host — no `ocaml-on-sx` dependency.
## Two layers, two timelines
1. **Now (unblocked): native server + SXTP adapter + SX handlers.** Route rose-ash
endpoints onto the SX host one at a time. Each migrated endpoint is an SX
handler dispatching to a subsystem; Quart proxies the rest until cut over.
2. **Next: `dream-on-sx` as the framework layer.** Dream gives Quart-grade
ergonomics — typed routing, middleware stacks, sessions, CSRF. It is gated on
`ocaml-on-sx` Phases 15 + minimal stdlib. **This plan is the concrete target
user that un-parks `dream-on-sx`** (see `plans/dream-on-sx.md`): "the subsystems
need an HTTP front door" is the real feature pulling Dream. Until then, do not
block migration on Dream — the native server is sufficient.
3. **Always: Python only at the edges.** External integrations — SumUp payments,
Ghost CMS, ActivityPub crypto, IPFS/Kubo — ride Python libraries today. They
stay as thin injected adapters (Python/FFI) behind subsystem interfaces until
native replacements exist. "Drop Quart" ≠ "drop every line of Python."
## Status (rolling)
`bash lib/host/conformance.sh`**0/0** (not yet started)
## Ground rules
- **Scope:** `lib/host/**` and `plans/host-on-sx.md`. May **import** every subsystem
+ the kernel's server/SXTP surface. Do **not** edit `spec/`, `hosts/`, `shared/`,
or subsystem internals — wire to their public APIs only. Host-primitive / server
changes belong in `hosts/` (out of scope) → Blockers.
- **Architecture:** a route maps (method, path) → handler; a handler is an SX fn
`request -> response` that calls subsystem APIs; middleware is composed handlers
(auth via `identity`, permission via `acl`, mute via subsystem prefs). SXTP is the
wire format between host and subsystem-as-service.
- **Migration discipline:** each endpoint moved must be behavior-equivalent to its
Quart original (golden-response test before flip). Keep a migration ledger.
- **Commits:** one feature per commit. Progress log + tick boxes.
## Architecture sketch
```
HTTP request HTTP response
│ ▲
▼ │
native OCaml http server (prod) ──────► lib/host/router.sx
(hosts/ — out of scope) — (method,path) → handler
│ ▲
▼ │
lib/host/middleware.sx lib/host/handler.sx
— auth(identity) ∘ acl ∘ mute ∘ ... — request → subsystem call → response
│ ▲
▼ │
lib/host/sxtp.sx subsystem APIs (feed/search/commerce/…)
— wire format, host↔service — called via public interfaces
└── external edges: SumUp / Ghost / AP / IPFS → injected Python/FFI adapters
```
## Phase 1 — Router + handler + one real endpoint
- [ ] `router.sx` — route table, (method,path) match
- [ ] `handler.sx` — request/response model, subsystem dispatch
- [ ] migrate ONE read endpoint (e.g. a feed timeline) end-to-end, golden test
- [ ] `conformance.sh` + scoreboard
## Phase 2 — Middleware + SXTP
- [ ] `middleware.sx` — composable auth/acl/mute/error layers
- [ ] `sxtp.sx` — host↔subsystem wire format (align with existing spec)
- [ ] migrate a write endpoint (auth + permission + action)
## Phase 3 — Strangler migration ledger
- [ ] enumerate Quart endpoints; track migrated vs proxied
- [ ] golden-response harness vs the live Quart responses
- [ ] cut over a whole domain (smallest: `likes` or `relations`) as proof
## Phase 4 — Dream framework layer (gated)
- [ ] gate: `ocaml-on-sx` Phases 15 + minimal stdlib green
- [ ] adopt `dream-on-sx` routing/middleware/session ergonomics over the same handlers
- [ ] re-home external adapters as native where replacements land
## Progress log
(loop fills this in)
## Blockers
(loop fills this in)

84
plans/identity-on-sx.md Normal file
View File

@@ -0,0 +1,84 @@
# identity-on-sx: OAuth2, sessions & membership on Erlang
> **DRAFT outline.** The identity core `acl-on-sx` assumes already exists. `acl`
> answers "may X do Y"; identity answers "who is X, and how did they prove it."
> Depends on `persist-on-sx` (grant/audit ledger). Pairs with `acl-on-sx`.
rose-ash's `account` domain is the OAuth2 authorization server every other app is
a client of: silent SSO, per-app first-party cookies, grant verification,
membership. Sessions and grants are **long-lived, concurrent, individually
addressable, and expire on their own** — that is the actor model. Erlang's
processes + mailboxes map cleanly: a session is a process, token issue/refresh/
revoke are messages, expiry is a process timeout, and SSO is one process answering
many apps.
End-state: an Erlang-on-SX layer with the OAuth2 authorization-code + silent
(`prompt=none`) flows as message protocols, a session/grant registry, token
lifecycle (issue/refresh/revoke/introspect), and membership state — all auditable
through the event log, all authorization questions delegated to `acl-on-sx`.
## Status (rolling)
`bash lib/identity/conformance.sh`**0/0** (not yet started)
## Ground rules
- **Scope:** only `lib/identity/**` and `plans/identity-on-sx.md`. May **import**
from `lib/erlang/`, and (once they exist) `lib/persist/` + `lib/acl/`. Do not edit
substrates.
- **Architecture:** a session/grant is a process holding its own state; the
registry routes messages by subject/client id. Tokens are opaque + introspected,
not self-validating (revocation must be real). Authorization decisions are NOT
made here — `identity` proves identity, `acl` decides permission.
- **Security:** revocation is immediate (kill the process / tombstone the grant);
no decision relies on a token that outlived its grant. Negative answers are
explicit, never "absence of a yes."
- **Commits:** one feature per commit. Progress log + tick boxes.
## Architecture sketch
```
Auth request Token / session
(authorize client scope subject) {:access :refresh :expires :grant}
│ ▲
▼ │
lib/identity/oauth.sx lib/identity/token.sx
— authz-code + prompt=none flows — issue / refresh / revoke / introspect
— as Erlang message protocols — opaque tokens, grant-backed
│ ▲
▼ │
lib/identity/session.sx lib/identity/registry.sx
— session = process, expiry=timeout — route by subject/client; SSO fan-out
│ │
▼ ▼
lib/identity/api.sx ── (identity/login) (identity/grant?) (identity/revoke) ──┐
│ │
└──────── grant + audit events → persist ; permission? → acl ──────────┘
```
## Phase 1 — Sessions + tokens
- [ ] `session.sx` — session process, create/lookup/expire
- [ ] `token.sx` — issue/introspect/revoke (opaque, grant-backed)
- [ ] `registry.sx` — route by subject/client
- [ ] `api.sx` + tests + scoreboard + conformance.sh
## Phase 2 — OAuth2 flows
- [ ] authorization-code flow as a message protocol
- [ ] refresh + rotation; revocation cascades to issued tokens
- [ ] tests: full code exchange, refresh, revoke-then-use (must fail)
## Phase 3 — Silent SSO + membership
- [ ] `prompt=none` cross-app login (one session, many clients)
- [ ] membership state + per-app grant projection
- [ ] grant verification delegated cache (mirror Redis-cache pattern)
## Phase 4 — Audit + federation
- [ ] every issue/refresh/revoke is a `persist` event; `(identity/audit subject)`
- [ ] federated identity (peer-asserted subject) — advisory, trust-gated stub
- [ ] tests: audit completeness, cross-instance subject mapping
## Progress log
(loop fills this in)
## Blockers
(loop fills this in)

112
plans/mod-on-sx.md Normal file
View File

@@ -0,0 +1,112 @@
# mod-on-sx: Moderation on Prolog
rose-ash needs moderation infrastructure: reports flagged by users, automated
classifications (spam, abuse), tiered escalation (auto → human → appeal), audit
trails. Each decision is the conclusion of a backtracking search over evidence and
policy rules — exactly what Prolog does.
Where acl-sx says "may this happen?", mod-sx says "should this stay?" The former is
a positive decision (proof of grant); the latter often a negative one (proof of
violation), and policy chains naturally backtrack: if the first rule doesn't apply,
try the next.
End-state: a Prolog-on-SX layer for moderation policy declaration and evaluation,
with persistent report lifecycle, audit log, escalation state machine, and
federation extension.
## Status (rolling)
`bash lib/mod/conformance.sh`**0/0** (not yet started)
## Ground rules
- **Scope:** only touch `lib/mod/**` and `plans/mod-on-sx.md`. Do **not** edit
`spec/`, `hosts/`, `shared/`, `lib/prolog/**`, or other `lib/<lang>/`. You may
**import** from `lib/prolog/` (public API in `lib/prolog/prolog.sx`); do **not**
modify Prolog.
- **Shared-file issues** go under "Blockers" with a minimal repro; do not fix here.
- **SX files:** use `sx-tree` MCP tools only.
- **Architecture:** policies are Prolog rules over `report(...)` and `evidence(...)`
facts. Decisions are query results. Proof trees become audit records. The state
machine for report lifecycle is separate (an SX module on top).
- **Shared with acl-sx:** rule-engine plumbing may be liftable into `lib/guest/`.
Watch for it; flag in Progress log but do not extract until both subsystems are
past Phase 2.
- **Commits:** one feature per commit. Keep Progress log updated and tick boxes.
## Architecture sketch
```
Report Decision
{:by :about :reason :at} {:action :proof :next-state}
│ ▲
▼ │
lib/mod/schema.sx lib/mod/engine.sx
— report/4, evidence/2, — query Prolog with report fact
classification/3 predicates — extract proof tree
│ ▲
▼ │
lib/mod/policy.sx lib/mod/lifecycle.sx
— rule syntax → Prolog — state machine
— action heads: — open → triaged → decided
{:keep :hide :remove — appeal handling
:escalate :ban} │
│ ▼
▼ lib/mod/audit.sx
lib/mod/api.sx — append-only decision log
— (mod/report ...) — proof tree persistence
— (mod/decide report) — query API
— (mod/appeal id)
lib/mod/fed.sx
— cross-instance reports via fed-sx
— decision sharing / trust model
```
## Phase 1 — Report representation + simple policy
- [ ] `lib/mod/schema.sx``report(id, by, about, reason)`, `evidence(id, kind, val)`,
`policy-action(report, action)` predicates as Prolog facts/rules
- [ ] `lib/mod/policy.sx` — rule declarations: `(defrule action :when conditions)`
desugars to Prolog clause
- [ ] `lib/mod/engine.sx``(decide report-id)` runs Prolog query, returns first
matching action
- [ ] `lib/mod/api.sx``(mod/report by about reason)`, `(mod/decide id)`
- [ ] `lib/mod/tests/decide.sx` — 15+ cases: spam keyword → hide, repeated reports →
escalate, no rule matches → keep
- [ ] `lib/mod/scoreboard.{json,md}`
- [ ] `lib/mod/conformance.sh`
## Phase 2 — Evidence + audit trail
- [ ] evidence accumulation — additional facts asserted before query
- [ ] proof tree from Prolog derivation tree
- [ ] `lib/mod/audit.sx` — append-only log (decision + proof + evidence snapshot)
- [ ] `(mod/audit id)` retrieval
- [ ] `lib/mod/tests/audit.sx` — proof correctness, trail completeness
## Phase 3 — Escalation + lifecycle state machine
- [ ] state machine: `:open → :triaged → :decided → :appealed → :final`
- [ ] auto-tier: first-pass rules decide quick cases
- [ ] human-tier: rules that emit `:escalate` move to next state
- [ ] appeal: re-runs with appeal evidence, may override prior decision
- [ ] `(mod/appeal id new-evidence)` API
- [ ] `lib/mod/tests/escalation.sx` — full lifecycle traversal cases
## Phase 4 — Federation
- [ ] cross-instance reports — peer raises report about local content (or vice versa)
- [ ] decision sharing — actions taken locally propagate to peers via fed-sx
- [ ] trust model — peer's decision is advisory unless `(trust peer :mod)` is granted
- [ ] revocation — undo applied moderation if proof was invalidated
- [ ] `lib/mod/tests/fed.sx` — federated decision chains (mock fed-sx in tests)
## Progress log
(loop fills this in)
## Blockers
(loop fills this in)

411
plans/persist-on-sx.md Normal file
View File

@@ -0,0 +1,411 @@
# persist-on-sx: Durable state on the SX kernel
> **DRAFT outline.** Foundation subsystem — the durable substrate the other five
> currently fake with in-memory mutable lists. Build this first.
>
> **"persist" = persistence / data store, NOT the shop.** The shop/commerce vertical
> is `commerce-on-sx`.
rose-ash needs durable state: every subsystem (feed log, flow store, mod audit,
search index, acl grants, sessions) today hand-rolls an in-memory structure that
vanishes on restart. `persist-on-sx` is the one durable substrate they share. It
lives directly on the SX kernel's IO-suspension primitives (`perform`/`cek-resume`
— the third CEK phase) so a read/write `perform`s and the kernel persists at the
boundary. Concrete storage backends are injected.
## Does it cover ALL persistence? No — and on purpose.
Event-sourcing-everything is a known trap (replay cost, event schema evolution,
awkward ad-hoc queries, 5MB images in a log). So persist owns the **durable
source-of-truth substrate**, exposed as **two facets over one backend protocol**,
with two things explicitly delegated out:
| Shape | Owner | Notes |
|-------|-------|-------|
| **Event streams** (append-only, history matters) | persist — **log facet** | feed activities, mod audit, order ledger, flow state, content edits |
| **Current-state values** (KV / document, no history) | persist — **kv facet** | profiles, stock counts, config, session blobs; also where projections materialize |
| **Snapshots / read models** (derived, queryable) | persist — projections → kv/log | rebuildable from the log; persisted so you don't replay to answer a query |
| **Blobs / large objects** (images, media) | **delegated** → content-addressed store (artdag/IPFS already) | persist stores the *reference/CID*, never the bytes |
| **Cache** (ephemeral, evictable) | **out of scope** | not persistence — different lifecycle (Redis-shaped) |
| **Ad-hoc relational query** | the subsystem, over a projected read model | the log is bad at "all orders by X in March"; project into a queryable kv/SQL backend |
So: persist is the **single durable substrate** for state that's either a stream of
changes or a current value — but it does **not** force everything into an event
log, it does **not** hold blobs (only their content-addressed refs), and it does
**not** do caching. Those boundaries are the whole point of calling it a substrate
rather than "the database."
End-state: `log` (append/read streams) + `kv` (get/put/delete by key) facets, an
injectable backend protocol (mem → file → Postgres → IPFS-ref), pure projections
with incremental snapshots, optimistic concurrency, and a subscription hook so
read models (feeds, indices, audit logs) update incrementally.
## Status (rolling)
`bash lib/persist/conformance.sh`**201/201** (Phases 14 complete + extensions + a reference migration)
## Ground rules
- **Scope:** only `lib/persist/**` and `plans/persist-on-sx.md`. May **import** the
kernel's IO-suspension surface (`perform`, platform IO ops) — verify what's
exported first. Do not add host primitives; a missing durable IO op is a Blockers
entry (it belongs in `hosts/`, out of scope).
- **Architecture:** an event is `{:stream :seq :type :at :data}`; the log is an
ordered append-only vector; a projection is `(fold step seed events)`; a kv value
is `(get/put/delete key)`. Both facets sit on one injected backend
`{:append :read :kv-get :kv-put :snapshot-read :snapshot-write}`. The in-memory
backend is the test default; real backends wire in unchanged.
- **Determinism:** replay is pure — same log → same state, always. No clocks or
randomness inside projections; time lives on the event.
- **Blobs:** store the content-address/CID and metadata; never the bytes. The blob
backend is a separate injected dependency.
- **Commits:** one feature per commit. Progress log + tick boxes.
## Architecture sketch
```
Command / write Read model / value
(append stream type data) (project stream step seed)
(kv-put key value) (kv-get key)
│ ▲
▼ │
lib/persist/event.sx lib/persist/project.sx
— {:stream :seq :type :at :data} — fold step seed; incremental from snapshot
│ ▲
▼ │
lib/persist/log.sx lib/persist/kv.sx lib/persist/snapshot.sx
— append/read — get/put/delete — checkpoint; replay = snapshot + tail
— optimistic seq — current-state
│ │ ▲
└──────────────────┴── (perform → backend) ───┘
lib/persist/backend.sx lib/persist/api.sx
— injected protocol — (persist/append) (persist/project)
— mem | file | pg | ipfs-ref — (persist/kv-get/put) (persist/subscribe)
└── blobs → content-addressed store (artdag/IPFS), by reference only
```
## Phase 1 — Log + kv + in-memory backend
- [x] `event.sx` — event record, stream/seq helpers
- [x] `backend.sx` — injectable protocol + in-memory impl (log + kv)
- [x] `log.sx``append` (optimistic seq), `read`, `read-from`
- [x] `kv.sx``get`/`put`/`delete` current-state
- [x] `api.sx` + tests + scoreboard + conformance.sh
## Phase 2 — Projections + subscriptions
- [x] `project.sx``(project stream step seed)`, incremental fold
- [x] subscription hook — projection / kv read model re-runs on append
- [x] concurrency conflict surfaced as a real result, not a crash
## Phase 3 — Snapshots + replay
- [x] `snapshot.sx` — checkpoint a projection; replay = snapshot + tail
- [x] compaction policy; replay-determinism tests
## Phase 4 — Durable backends via kernel IO
- [x] file/log backend driven through `perform` (IO-suspension boundary)
- [x] blob backend interface (store ref/CID; bytes live in artdag/IPFS)
- [x] crash/restart replay test (mock IO platform)
- [x] migration notes for swapping mem → durable under a live subsystem
### Migration notes — mem → durable under a live subsystem
The facet API takes the backend as its first argument and never names a concrete
backend, so swapping storage is a one-line change at the open site:
```
(persist/open) ; in-memory (test / ephemeral)
(persist/mock-durable (persist/mem-backend)); durable protocol, in-process disk
(persist/durable-backend) ; production: ops cross perform → host
```
Everything above the backend — `append`/`read`/`project`/`subscribe`/`snapshot`
/`compact` — is byte-identical across all three. A subsystem migrates by:
1. **Pick the seam.** The subsystem holds one backend value (today an in-memory
list). Replace its construction with `persist/open`/`durable-backend`; leave
every call site untouched.
2. **Backfill.** For an existing in-memory store, replay its current state into
the durable backend once (append historical events / `kv-put` current
values) before cutting reads over. New writes go to durable from then on.
3. **Read models rebuild themselves.** A projection is pure `(fold step seed)`;
after cutover, `persist/replay` (snapshot + tail) reconstructs every read
model from the durable log — no bespoke migration of derived state.
4. **Blobs first, by reference.** Move large payloads into the content store and
store only `persist/blob-ref`s; the log/kv stay small, so the backfill in (2)
never copies bytes.
5. **Concurrency is already handled.** Two writers racing a stream get a
`persist/conflict?` result, not corruption — the same on mem or durable, so
no new code is needed at cutover.
The only behavioural difference durable introduces is that each op crosses the
kernel IO-suspension boundary (`perform`): under the real kernel the call
suspends and the host resumes it transparently, so the facet code is unaware.
Tests prove this by routing the identical request shapes through `persist/serve`
over an in-process disk (the mock-IO harness).
## Extensions (post-roadmap)
- [x] `view.sx` — materialized views: bundle stream + fold + snapshot name;
`view-attach` keeps the snapshot current on every publish so `view-peek` is an
O(1) read. The consumer-facing read-model abstraction (feed indices, audit
rollups, search counters).
- [x] `kv.sx` CAS — `persist/kv-cas` (compare-and-swap) + `persist/kv-put-new`
(create-only): atomic current-state updates, conflict as a real value (kv
analogue of log `append-expect`). For sessions, acl grants, stock counts.
- [x] `catalog.sx` — stream catalog: `persist/streams`/`stream-count`/
`stream-exists?`/`total-events`. Backend `:streams` op (from seq high-water
marks, so compacted streams still list), threaded through mem + durable.
- [x] `query.sx` — read-side scans: `read-between` (seq range), `read-since`/
`read-window` (by `:at`), `read-by-type`, `read-where`, `count-where`. Pure
reads for audit windows / type filters / since-cursors.
- [x] `batch.sx``persist/append-batch` commits a list of `(type at data)`
specs as one contiguous block; `persist/append-batch-expect` is transactional
(all-or-nothing guarded by optimistic concurrency). For an order + its line
items as one commit.
- [x] `upcast.sx` — event schema evolution: register a pure `(event -> event)`
upcaster per type; `read-upcast`/`project-upcast` lift old events to the
current shape on read so projections see one shape. Immutable registry;
`upcast-data` helper merges new `:data` fields. Addresses the schema-evolution
trap without rewriting history.
- [x] `idempotency.sx` — exactly-once append under retries: `persist/append-once`
keyed by a caller idempotency key (per stream), returning the same event on a
repeat. Marker lives in kv, so idempotency holds across restart. `seen?` check.
- [x] `global.sx` — global commit ordering across streams (the primitive feed's
unified timeline needs). `persist/gappend` records a pointer in a reserved
`$global` index whose seq is the commit position; `read-global`/
`project-global` replay every event in commit order; `global-from` for
incremental consumers. Opt-in (plain `append` never touches it); reserved
index hidden from the public catalog. Deterministic across restart.
## Consumers (post-foundation, not in scope here)
feed/-log, flow store, mod/audit, search index, acl grants, identity sessions all
become `persist` log or kv. Track each migration in that subsystem's plan.
**Reference migration:** `lib/persist/examples/acl.sx` is a worked, tested
template — an ACL-grants store rebuilt on persist (grants/revokes as events,
current set as a projection, O(1) checks via a materialized view, an audit-window
query). It carries an explicit BEFORE (hand-rolled ephemeral map) → AFTER
diff in its header and proves the headline win (grants survive restart) on the
durable backend. Other subsystem loops copy this pattern; it does not touch the
real `lib/acl`.
## Progress log
- **Reference migration: acl grants (201/201).** `lib/persist/examples/acl.sx`
a worked, in-scope template migrating an ACL-grants store from a hand-rolled
ephemeral map to persist: grants/revokes as events, current set as a
projection, O(1) checks via a materialized view, audit via `read-window`.
Header carries the BEFORE→AFTER diff. 10 tests, incl. grants surviving restart
on the durable backend (the capability the BEFORE version lacked). The pattern
other subsystem loops copy.
- **Ext: global commit ordering (191/191).** `global.sx``persist/gappend`
records a pointer in a reserved `$global` index (its seq = global commit
position); `read-global`/`project-global` resolve pointers to events in commit
order; `global-from` for incremental global consumers. Opt-in; `$`-streams are
now reserved + hidden from the public catalog (`streams-all` reveals them).
Gives feed its cross-stream timeline. 11 tests incl. durable + restart
determinism.
- **Ext: exactly-once append (180/180).** `idempotency.sx`
`persist/append-once` appends at most once per (stream, idempotency key),
returning the same event on a repeat; the marker lives in kv so it survives
restart (verified on durable). `persist/seen?` check. 9 tests.
- **Ext: event schema evolution (171/171).** `upcast.sx` — per-type pure
`(event -> event)` upcasters in an immutable registry; `read-upcast`/
`project-upcast` lift legacy events to the current shape on read so
projections never branch on version. `upcast-data` merges new `:data` fields
keeping stream/seq/type/at. 9 tests incl. mixed old/new + durable.
- **Ext: atomic batch append (162/162).** `batch.sx``persist/append-batch`
commits `(type at data)` specs as one contiguous block (real cons-list, in
order); `persist/append-batch-expect` checks the stream is still at expected
before writing any event, so the batch is all-or-nothing under a concurrent
writer. 10 tests incl. conflict-writes-nothing + durable.
- **Ext: read-side query helpers (152/152).** `query.sx``read-between` (seq
range), `read-since`/`read-window` (by `:at`), `read-by-type`, `read-where`,
`count-where`. Pure scans over `persist/read`; for ad-hoc relational queries
consumers still project into a kv read model. 9 tests incl. durable.
- **Ext: stream catalog (143/143).** New backend op `:streams` (keys of the seq
high-water-mark dict, threaded through mem-backend + durable serve/io-backend)
so fully-compacted streams still enumerate. `catalog.sx`:
`persist/streams`/`stream-count`/`stream-exists?`/`total-events`. 10 tests
incl. durable + restart.
- **Ext: kv compare-and-swap (133/133).** `persist/kv-cas` sets a key only if
its current value equals expected, else returns `{:conflict :expected
:actual}`; `persist/kv-put-new` is create-only. The kv analogue of log
`append-expect` — atomic current-state for sessions/acl/stock. 11 tests incl.
racer + retry + durable backend.
- **Ext: materialized views (122/122).** `view.sx``persist/view` bundles
stream + step + seed + snapshot name; `view-attach` subscribes it to a hub so
every publish refreshes the snapshot incrementally; `view-peek` is then an
O(1) current read (no fold), `view-value` always folds the tail so it's never
stale. 11 tests incl. on durable backend + a sum-over-data view.
- **Phase 4c+4d (111/111) — Phase 4 complete, roadmap done.** `recovery.sx` — a
6-test crash/restart integration: an order ledger (event log + subscription
kv read model + snapshot + compaction + invoice blob ref) over the durable
backend, where "crash" drops every in-process object and "restart" rebuilds
over the same disk + content store. Log, read model, snapshot, compacted
replay, and blob ref all survive; seq continues; two restarts converge
(determinism). Migration notes (mem → durable under a live subsystem) added
inline above.
- **Phase 4b (105/105).** `blob.sx` — large objects stay out of persist. A blob
ref is `{:cid :size :mime}`; the blob store is a SEPARATE injected dependency
(`persist/blob-io` over an injectable transport, perform in prod / mock
content store in tests). `persist/blob-store` puts bytes and returns ONLY the
ref; `persist/blob-fetch` retrieves bytes via the ref. Mock store is
content-addressed (same bytes dedupe). 14 tests assert the invariant: a ref in
the log/kv carries the CID, never the bytes (`has-key? :bytes` is false).
- **Phase 4a (91/91).** `durable.sx` — a backend whose every op crosses the
kernel IO boundary via `(perform {:op "persist/..." :args (...)})`. The
transport is injectable: `persist/durable-backend` uses the kernel's
`perform` (suspends; host resumes); `persist/mock-durable` uses
`persist/serve` over an in-memory disk. `persist/serve` is the reference host
+ the mock-IO harness. Because the request shapes are identical, the ENTIRE
facet stack (log/kv/project/snapshot/compaction) runs unchanged on
mock-durable — verified. Crash/restart (drop backend, keep disk) recovers log
+ kv + snapshot by replay; seq counter continues. 15 tests. See Blockers for
why end-to-end perform suspension isn't exercised under sx_server.exe.
- **Phase 3b (76/76) — Phase 3 complete.** Backend refactor: `last-seq` is now
a monotonic per-stream high-water mark (backend `seqs` dict), not physical
length, so a compacted log keeps assigning climbing seqs. Added backend
`:truncate-through` + `persist/truncate`. `compaction.sx``persist/compact`
checkpoints then drops events with seq <= snapshot seq; `should-compact?`/
`maybe-compact` give an explicit "compact every N tail events" policy. 11
tests: post-compaction replay value == uncompacted full replay (determinism),
seq continuity after truncation, idempotence. `persist/count` = physical
stored count (shrinks on compaction) vs `persist/last-seq` = logical.
- **Phase 3a (65/65).** `snapshot.sx` — a snapshot is a projection state
`{:value :seq}` stored in the kv facet under `snapshot/<name>`.
`persist/checkpoint` replays + saves; `persist/replay` = snapshot + tail.
11 tests assert the headline both ways: snapshot+tail == full replay (value
and whole state), plus replay determinism.
- **Phase 2c (54/54) — Phase 2 complete.** `concurrency.sx` — optimistic
concurrency: `persist/append-expect b stream expected ...` refuses the append
if the stream advanced past `expected`, returning a conflict VALUE
`{:conflict true :expected :actual}` (never a crash, never a silent
overwrite). `persist/conflict?` + accessors; caller re-reads actual and
retries. 8 tests incl. two-writer race + retry.
- **Phase 2b (46/46).** `subscribe.sx``persist/hub` wraps a backend with
per-stream callbacks. `persist/publish` appends then fires subscribers
`(backend stream event)`; direct `persist/append` bypasses them by design
(bulk load/replay). Canonical use: callback re-runs `project-resume` or bumps
a kv counter so read models update on write. 9 tests.
- **Phase 2a (37/37).** `project.sx` — projection state `{:value :seq}`;
`persist/project` folds whole stream from seed, `persist/project-resume`
folds only the tail (seq > prior seq) so read models update incrementally.
step is pure `(value event) -> value`. 9 tests incl. resume==full-from-zero.
- **Phase 1 complete (28/28).** `event.sx` (event record + accessors),
`backend.sx` (injectable protocol + in-memory log/kv impl, closure state via
set!), `log.sx` (append/read/read-from, sequential per-stream seq, stream
isolation), `kv.sx` (get/put/delete/has?/keys/get-or/update), `api.sx`
(`persist/open` — mem default, backend injectable). conformance.sh + three
suites (event/log/kv). Gotcha logged in Blockers: `map` returns an
array-backed list not `equal?` to a `(list ...)` literal — assertions build
compared lists with list/nth.
## Blockers
### OPEN — host durable-storage adapter (the only gap to real durability)
**Owner:** a `hosts/` loop (NOT this one — `lib/persist/**` is the scope fence,
and `sx_build` is forbidden here). **Without it, durable persistence silently
drops all writes.**
**Symptom / minimal repro.** `persist/durable-backend` performs
`{:op "persist/..." :args (...)}` for every storage op. Under `sx_server.exe`
the kernel's default IO resolver answers unknown ops with `nil` — so the durable
backend does not error, it *silently no-ops*:
```
; load event/backend/log/durable, then:
(let ((b (persist/durable-backend)))
(begin (persist/append b "s" "x" 0 {})
(persist/append b "s" "x" 0 {})
(list (persist/event-seq (persist/append b "s" "x" 0 {}))
(persist/count b "s")
(persist/read b "s"))))
; => (1 0 nil) ; every append gets seq 1, nothing stored, reads empty — DATA LOSS
```
The in-memory backend (`persist/open`) is correct and complete; this gap is
*only* the production transport.
**What to build.** A host servicer that answers the `persist/*` IO ops against a
real store (sqlite/files/pg). It is the production twin of `persist/serve`
(`lib/persist/durable.sx`) — same op names, same request/response shapes — so
mirror that function and back it with durable storage instead of a mem-backend.
**Op contract** (request `{:op :args}` → response). `args` is a positional list;
events are dicts `{:stream :seq :type :at :data}`:
| op | args | returns | semantics |
|----|------|---------|-----------|
| `persist/append` | `(stream event)` | (ignored) | store `event` in `stream` |
| `persist/read` | `(stream)` | event list (oldest-first) | currently-stored events |
| `persist/last-seq` | `(stream)` | number | **monotonic high-water mark** (see below) |
| `persist/streams` | `()` | stream-name list | every stream ever appended to |
| `persist/truncate` | `(stream n)` | (ignored) | drop events with `seq <= n` |
| `persist/kv-get` | `(key)` | value or nil | |
| `persist/kv-put` | `(key val)` | (ignored) | upsert |
| `persist/kv-delete`| `(key)` | (ignored) | remove key |
| `persist/kv-has?` | `(key)` | boolean | |
| `persist/kv-keys` | `()` | key list | |
**Hard invariants** (the facets above rely on these; mem-backend + `persist/serve`
are the reference):
1. **`last-seq` is a per-stream monotonic counter, NOT the row count.** It must
keep climbing after `truncate`, so a compacted stream never reassigns a seq.
Store the counter separately from the rows.
2. `append` is the only seq-assigner upstream (`log.sx` does `last-seq + 1`); the
host must not renumber.
3. `read` returns events in append order with `:seq` intact (post-truncate it
returns only the surviving tail).
4. `streams` is the set of streams that ever had an append (survives full
compaction) — keep it keyed off the seq counters, like mem-backend's `seqs`.
5. Values round-trip structurally: dicts/lists/numbers/strings/nil/booleans in =
same out (event `:data`, kv values, blob refs).
**Blobs** are a *separate* adapter with the same pattern: ops `blob/put`
`(bytes mime)` → cid, `blob/get` `(cid)` → bytes, `blob/has?` `(cid)` → bool
(see `lib/persist/blob.sx` / `persist/blob-serve`). Back it with the
content-addressed store (artdag/IPFS); persist only ever stores the returned ref.
**Where to register.** `hosts/ocaml/bin/sx_server.ml`:
- the in-process resolver `Sx_types._cek_io_resolver` (~line 3864) — add a
`"persist/..."` match arm dispatching to the new storage module (used by
SSR/`eval_with_io`); and/or
- the bridge path in `cek_run_with_io` (~line 528576), which currently forwards
unknown ops via `io_request op args` to the external bridge — a Python-bridge
handler is the alternative home if storage lives Python-side.
Pick one home; the op names are the contract, not the location.
**Acceptance test.** Swap the transport: point a `persist/io-backend` at the new
host servicer (instead of `persist/serve` over a mem disk) and run the existing
`durable` + `recovery` suites — they must stay green, and state must survive an
actual process restart (kill the server, restart, replay → recovered). That is
exactly what `lib/persist/tests/durable.sx` and `recovery.sx` already assert
against the mock; the host adapter just makes the disk real.
---
- **Phase 4 perform-suspension not exercised end-to-end under sx_server.exe (by
design, not a bug).** The CEK suspension primitives (`cek-step-loop`,
`cek-resume`, `cek-suspended?`, `cek-io-request`) and a settable SX-level IO
hook are only bound by the `run_tests` OCaml binary (out of scope: hosts/, and
sx_build is forbidden). Under `sx_server.exe`, an unhandled `perform` resolves
through the OCaml io-request/io-response stdin bridge (production path) — not
callable from the pure-eval conformance harness. Resolution: the durable
backend's transport is injectable, so the production path is one line
`(perform req)` (kernel-handled) and ALL durable logic is tested through the
mock transport (`persist/serve` over an in-memory disk). The single untested
line is the kernel primitive itself. No host primitive needed; nothing to fix.
- **Not a blocker, a testing convention:** `map` returns an array-backed list
that is NOT `equal?` to a `(list ...)` cons-literal (two `map` results do
compare equal to each other). When asserting list-shaped results against a
`(list ...)` literal, build the compared value with `list`/`nth`/`cons`, not
`map`. `into`/list-coercion needs the IO bridge and is unusable in the
pure-eval harness.

106
plans/search-on-sx.md Normal file
View File

@@ -0,0 +1,106 @@
# search-on-sx: Full-text + structured search on Haskell
rose-ash needs search across pages, posts, threads, federated content. Tokenize,
index, query, rank, filter by visibility. Typed ADTs make query parsing clean,
lazy lists make posting-list iteration efficient, and Haskell-on-SX is at 1514/1514.
End-state: a Haskell-on-SX layer with inverted index, query AST, boolean +
phrase + ranked queries (TF-IDF, BM25), ACL-aware post-filter, and a federation
extension that merges per-peer indices.
## Status (rolling)
`bash lib/search/conformance.sh`**0/0** (not yet started)
## Ground rules
- **Scope:** only touch `lib/search/**` and `plans/search-on-sx.md`. Do **not** edit
`spec/`, `hosts/`, `shared/`, `lib/haskell/**`, or other `lib/<lang>/`. You may
**import** from `lib/haskell/` (public API in `lib/haskell/haskell.sx`); do **not**
modify Haskell.
- **Shared-file issues** go under "Blockers" with a minimal repro; do not fix here.
- **SX files:** use `sx-tree` MCP tools only.
- **Architecture:** index = `Map Term [(DocId, [Pos])]`. Query AST = ADT. Eval =
fold of posting lists with set ops + ranking math. Ranking is pure (no IO until
result emission).
- **Commits:** one feature per commit. Keep Progress log updated and tick boxes.
## Architecture sketch
```
Document Query
{:id :text :tags} "alice AND bob OR phrase \"x y\""
│ │
▼ ▼
lib/search/tokenize.sx lib/search/parse.sx
— tokenize :: Text → [Term] — parse :: Text → Query
— normalize (lowercase, strip) — Query = Term | And | Or
— (optionally) stem | Not | Phrase
│ │
▼ ▼
lib/search/index.sx lib/search/eval.sx
— Map Term [(DocId, [Pos])] — eval :: Index → Query → [DocId]
— insert / delete / lookup — boolean + phrase positions
— persistence (optional later) │
│ ▼
└────────────────► lib/search/rank.sx
— TF-IDF / BM25 scoring
— top-N
lib/search/api.sx
— (search/index doc)
— (search/query q)
— (search/top n q)
lib/search/fed.sx
— federated query (merge peer results)
— ACL filter post-merge
```
## Phase 1 — Tokenize + index
- [ ] `lib/search/tokenize.sx` — normalize (lowercase, strip punctuation), split on
whitespace, return positions
- [ ] `lib/search/index.sx` — inverted index data structure (typed `Map` from
haskell lib); `insert`, `delete`, `lookup`
- [ ] `lib/search/api.sx``(search/index doc)`, `(search/lookup term)`
- [ ] `lib/search/tests/index.sx` — 15+ cases: tokenize, insert + lookup, update,
delete, multi-doc
- [ ] `lib/search/scoreboard.{json,md}`
- [ ] `lib/search/conformance.sh`
## Phase 2 — Query AST + boolean evaluation
- [ ] Query ADT: `Term Text | And Query Query | Or Query Query | Not Query |
Phrase [Text]`
- [ ] `lib/search/parse.sx` — query syntax parser (boolean operators, quoted phrases)
- [ ] `lib/search/eval.sx` — boolean eval via set ops on posting lists
- [ ] phrase eval — adjacency check using positions
- [ ] `lib/search/tests/boolean.sx` — 25+ cases: term, and, or, not, phrase,
composition, parser edge cases
## Phase 3 — Ranking
- [ ] document frequency tracking — extend index with `df` per term
- [ ] TF-IDF scoring
- [ ] BM25 scoring (configurable k1, b)
- [ ] top-N retrieval (heap-based)
- [ ] `lib/search/tests/rank.sx` — 20+ cases: TF-IDF behavior, BM25 vs TF-IDF,
ranking stability, top-N correctness
## Phase 4 — ACL filter + federation
- [ ] post-filter — each candidate result tested via `(acl/permit? viewer :read doc)`
- [ ] federated query — fan out to peer instances via fed-sx, merge results
- [ ] merge policy — interleave by rank, dedupe by `(peer, doc-id)`
- [ ] `lib/search/tests/integration.sx` — federated search with ACL filter
## Progress log
(loop fills this in)
## Blockers
(loop fills this in)