fed-sx-m1: Step 7a — pure-functional projection driver + 12 tests

next/kernel/projection.erl

@@ -0,0 +1,54 @@
+-module(projection).
+-export([new/2, new/3, fold_activity/2, replay/2,
+         name/1, state/1, fold_fn/1]).
+
+%% Pure-functional projection driver per design §10.
+%%
+%% A projection is a property list:
+%%   [{name, atom}, {state, term}, {fold, fun}]
+%%
+%% The fold function is `fun (Activity, State) -> NewState`. v1
+%% uses Erlang funs as the fold body — the genesis bundle's SX
+%% `:fold` bodies are stored as binaries; an SX-source eval
+%% bridge will plug them into the same projection record once
+%% it lands (Step 7d). For now, callers supply Erlang funs
+%% directly when constructing a projection.
+%%
+%% `replay/2` is the cold-start primitive: fold an activity
+%% list (e.g. `log:entries/1`) through the projection from its
+%% initial state.
+
+new(Name, InitialState) ->
+    new(Name, InitialState, fun (_Activity, S) -> S end).
+
+new(Name, InitialState, FoldFn) ->
+    [{name, Name}, {state, InitialState}, {fold, FoldFn}].
+
+fold_activity(Proj, Activity) ->
+    Fn  = fold_fn(Proj),
+    S0  = state(Proj),
+    S1  = Fn(Activity, S0),
+    set_field(state, S1, Proj).
+
+replay(Proj, Activities) ->
+    fold_each(Proj, Activities).
+
+fold_each(Proj, []) -> Proj;
+fold_each(Proj, [A | Rest]) ->
+    fold_each(fold_activity(Proj, A), Rest).
+
+%% Accessors
+
+name(Proj)    -> field(name, Proj).
+state(Proj)   -> field(state, Proj).
+fold_fn(Proj) -> field(fold, Proj).
+
+%% Internal
+
+field(K, [{K, V} | _]) -> V;
+field(K, [_ | Rest]) -> field(K, Rest);
+field(_, []) -> erlang:error(badkey).
+
+set_field(K, V, [{K, _} | Rest]) -> [{K, V} | Rest];
+set_field(K, V, [P | Rest]) -> [P | set_field(K, V, Rest)];
+set_field(K, V, []) -> [{K, V}].

next/tests/projection_pure.sh

@@ -0,0 +1,125 @@
+#!/usr/bin/env bash
+# next/tests/projection_pure.sh — Step 7a acceptance test.
+#
+# Exercises the pure-functional projection driver:
+#   new/2,3, fold_activity/2, replay/2, name/1, state/1, fold_fn/1.
+# Fold bodies are Erlang funs in v1; SX-source eval bridge will
+# plug into the same record later. 12 cases.
+
+set -uo pipefail
+cd "$(git rev-parse --show-toplevel)"
+
+SX_SERVER="${SX_SERVER:-hosts/ocaml/_build/default/bin/sx_server.exe}"
+if [ ! -x "$SX_SERVER" ]; then
+  SX_SERVER="/root/rose-ash/hosts/ocaml/_build/default/bin/sx_server.exe"
+fi
+if [ ! -x "$SX_SERVER" ]; then
+  echo "ERROR: sx_server.exe not found." >&2
+  exit 1
+fi
+
+VERBOSE="${1:-}"
+PASS=0; FAIL=0; ERRORS=""
+TMPFILE=$(mktemp); trap "rm -f $TMPFILE" EXIT
+
+cat > "$TMPFILE" <<'EPOCHS'
+(epoch 1)
+(load "lib/erlang/tokenizer.sx")
+(load "lib/erlang/parser.sx")
+(load "lib/erlang/parser-core.sx")
+(load "lib/erlang/parser-expr.sx")
+(load "lib/erlang/parser-module.sx")
+(load "lib/erlang/transpile.sx")
+(load "lib/erlang/runtime.sx")
+(load "lib/erlang/vm/dispatcher.sx")
+
+(epoch 2)
+(eval "(get (erlang-load-module (file-read \"next/kernel/projection.erl\")) :name)")
+
+;; new/2 sets initial state to the supplied value
+(epoch 10)
+(eval "(get (erlang-eval-ast \"P = projection:new(activity_log, init_state), projection:state(P) =:= init_state\") :name)")
+
+;; new/2 default fold is identity
+(epoch 11)
+(eval "(get (erlang-eval-ast \"P = projection:new(activity_log, base), P1 = projection:fold_activity(P, anything), projection:state(P1) =:= base\") :name)")
+
+;; new/3 stores supplied fold
+(epoch 12)
+(eval "(get (erlang-eval-ast \"P = projection:new(counter, 0, fun (_A, S) -> S + 1 end), is_function(projection:fold_fn(P))\") :name)")
+
+;; fold_activity threads through the fold fn
+(epoch 13)
+(eval "(erlang-eval-ast \"P = projection:new(counter, 0, fun (_A, S) -> S + 1 end), P1 = projection:fold_activity(P, x), projection:state(P1)\")")
+
+;; Two fold_activity calls accumulate
+(epoch 14)
+(eval "(erlang-eval-ast \"P = projection:new(counter, 0, fun (_A, S) -> S + 1 end), P1 = projection:fold_activity(P, a), P2 = projection:fold_activity(P1, b), projection:state(P2)\")")
+
+;; replay over a list
+(epoch 15)
+(eval "(erlang-eval-ast \"P = projection:new(counter, 0, fun (_A, S) -> S + 1 end), P1 = projection:replay(P, [a, b, c, d, e]), projection:state(P1)\")")
+
+;; replay over [] returns the projection unchanged (state preserved)
+(epoch 16)
+(eval "(erlang-eval-ast \"P = projection:new(counter, 99, fun (_A, S) -> S + 1 end), P1 = projection:replay(P, []), projection:state(P1)\")")
+
+;; Fold can read activity content (here append it)
+(epoch 17)
+(eval "(get (erlang-eval-ast \"P = projection:new(byname, [], fun (A, S) -> [A | S] end), P1 = projection:replay(P, [a, b, c]), projection:state(P1) =:= [c, b, a]\") :name)")
+
+;; Different projections are independent (different fold bodies)
+(epoch 18)
+(eval "(get (erlang-eval-ast \"P1 = projection:new(p_count, 0, fun (_A, S) -> S + 1 end), P2 = projection:new(p_collect, [], fun (A, S) -> [A | S] end), R1 = projection:replay(P1, [a, b, c]), R2 = projection:replay(P2, [a, b, c]), {projection:state(R1), projection:state(R2)} =:= {3, [c, b, a]}\") :name)")
+
+;; Name accessor
+(epoch 19)
+(eval "(get (erlang-eval-ast \"projection:name(projection:new(some_name, init)) =:= some_name\") :name)")
+
+;; Multi-step replay: aggregator by activity tag
+(epoch 20)
+(eval "(get (erlang-eval-ast \"By = fun (A, S) -> case A of {tag, T} -> [T | S]; _ -> S end end, P = projection:new(tag_log, [], By), P1 = projection:replay(P, [{tag, foo}, plain, {tag, bar}, {tag, baz}]), projection:state(P1) =:= [baz, bar, foo]\") :name)")
+EPOCHS
+
+OUTPUT=$(timeout 120 "$SX_SERVER" < "$TMPFILE" 2>/dev/null)
+
+check() {
+  local epoch="$1" desc="$2" expected="$3"
+  local actual
+  actual=$(echo "$OUTPUT" | awk -v e="$epoch" '
+    $0 ~ "^\\(ok-len " e " " { getline; print; exit }
+    $0 ~ "^\\(ok " e " "     { print; exit }
+    $0 ~ "^\\(error " e " "  { print; exit }
+  ')
+  [ -z "$actual" ] && actual="<no output for epoch $epoch>"
+  if echo "$actual" | grep -qF -- "$expected"; then
+    PASS=$((PASS+1))
+    [ "$VERBOSE" = "-v" ] && echo "  ok $desc"
+  else
+    FAIL=$((FAIL+1))
+    ERRORS+="  FAIL [$desc] (epoch $epoch) expected: $expected | actual: $actual
+"
+  fi
+}
+
+check  2  "module load name"                  "projection"
+check 10  "new/2 stores initial state"        "true"
+check 11  "default fold is identity"          "true"
+check 12  "new/3 stores fold fn"              "true"
+check 13  "fold_activity threads fn"          "1"
+check 14  "two folds accumulate"              "2"
+check 15  "replay over 5 activities"          "5"
+check 16  "replay over [] preserves state"    "99"
+check 17  "fold can read activity content"    "true"
+check 18  "different projections indep."      "true"
+check 19  "name accessor"                     "true"
+check 20  "tag-aware fold (replay)"           "true"
+
+TOTAL=$((PASS+FAIL))
+if [ $FAIL -eq 0 ]; then
+  echo "ok $PASS/$TOTAL next/tests/projection_pure.sh passed"
+else
+  echo "FAIL $PASS/$TOTAL passed, $FAIL failed:"
+  echo "$ERRORS"
+fi
+[ $FAIL -eq 0 ]

plans/fed-sx-milestone-1.md

@@ -455,6 +455,12 @@ publish(ActorId, ActivityRequest) ->
 
 ## Step 7 — Projection scheduler
 
+**Sub-deliverables:**
+- [x] **7a** — Pure-functional `next/kernel/projection.erl`: `new/2,3`, `fold_activity/2`, `replay/2`, `name/1`, `state/1`, `fold_fn/1`. Projection record is `[{name, _}, {state, _}, {fold, fun}]`; fold body is an Erlang fun in v1 (SX-source eval bridge deferred). `next/tests/projection_pure.sh` (12 cases).
+- [ ] **7b** — gen_server wrapper: `start_link/1`, named-per-projection, `async_fold/2`, `query/1`, `snapshot/1`.
+- [ ] **7c** — Broadcast hook from `outbox:publish` — feed `Signed` to every projection process.
+- [ ] **7d** — `sandbox:eval_pure/2` (Erlang sandbox-mode caller — gas budget + IO denial) once an SX-source eval bridge exists.
+
 **Deliverables:**
 
 ```erlang
@@ -971,6 +977,7 @@ A few things still under-specified; resolve as work begins.
 Newest first. One line per sub-deliverable commit. Erlang conformance gate
 (`bash lib/erlang/conformance.sh`) must remain 729/729 on every entry.
 
+- **2026-05-28** — Step 7a: `next/kernel/projection.erl` — pure-functional projection driver. Record shape `[{name, _}, {state, _}, {fold, fun}]`; `fold_activity/2` advances state by one activity; `replay/2` folds a whole list (mirrors `log:entries/1` semantics); `new/2` defaults to the identity fold and `new/3` accepts a custom Erlang fun. Multiple projections share no state — independent record values. Step 7 split into 7a (done) + 7b (gen_server-per-projection) + 7c (broadcast hook from outbox) + 7d (sandbox eval, needs SX-source bridge). `next/tests/projection_pure.sh` 12/12. Erlang conformance 729/729.
 - **2026-05-28** — Step 6d-publish: `outbox:publish/2(Request, Context)` orchestrates construct + sign + `pipeline:run_stages` + `log:append`. Stage list is `[stage_envelope, stage_signature(AS), stage_replay(LogState)]` — so a duplicate publish (same Request, same Published) halts at the replay stage and returns `{error, replay, LogState}` with the log unchanged; bad key material halts at `bad_signature`. Happy path returns `{ok, [{cid, Cid}, {activity, Signed}], NewLog}`. Projection-scheduler dispatch deferred to Step 7. `next/tests/outbox_publish.sh` 13/13 covers happy path, replay halt, sig halt, multi-publish progression, CID stability across fresh logs. Erlang conformance 729/729.
 - **2026-05-28** — Step 6d-cs: `next/kernel/outbox.erl` — envelope construction + signing. `construct/4` takes `(Type, ActorId, Published, Object)`, builds the canonical key-sorted property list, and derives the activity `:id` from `cid:to_string({activity_envelope, Skeleton})`. `sign/2` extracts key_id/algorithm/key-material from a KeySpec proplist, computes the v1 HMAC over canonical bytes, and appends the `:signature` pair. `cid_of/1` is a convenience accessor. Round-trip end-to-end through `envelope:verify_signature/2` verified (correct key passes, wrong key returns bad_signature). Step 6d split into 6d-cs (done) + 6d-publish (orchestration). `next/tests/outbox_construct.sh` 13/13. Erlang conformance 729/729.
 - **2026-05-28** — Step 6c-replay: `pipeline:stage_replay/2` (direct) + `stage_replay/1` (factory closed over LogState). Linear scan of `log:entries/1` checking for an existing entry with the same `:id`. Returns ok if new, `{error, replay}` on duplicate, `{error, no_id}` when the activity has no id field. Step 6c split into 6c-replay (done) + 6c-schema (deferred — blocked behind SX-source eval bridge for the activity-type :schema body). `next/tests/pipeline_replay.sh` 12/12 covers direct + factory + composition with stage_envelope. Erlang conformance 729/729.