fed-sx-m1: Step 6c-schema-pure — pipeline:stage_schema/1,/2 with SchemaLookup callback + 14 tests

next/kernel/pipeline.erl

@@ -4,7 +4,8 @@
          inbound_stages/0, outbound_stages/0,
          stage_envelope/1,
          stage_signature/1, stage_signature/2,
-         stage_replay/1, stage_replay/2]).
+         stage_replay/1, stage_replay/2,
+         stage_schema/1, stage_schema/2]).
 
 %% Validation pipeline per design §14.
 %%
@@ -89,3 +90,46 @@ log_has_id(Id, [Act | Rest]) ->
         {ok, Id} -> true;
         _        -> log_has_id(Id, Rest)
     end.
+
+%% stage_schema/2 — validates the activity's :object against the
+%% schema registered for its :type. SchemaLookup is a caller-
+%% supplied fun (Type) -> {ok, SchemaFn} | not_found; SchemaFn is
+%% itself a fun (Object) -> bool. Returns:
+%%   ok                          when the schema accepts the object
+%%   {error, no_type}            when the activity has no :type
+%%   {error, schema_mismatch}    when SchemaFn returned false
+%%
+%% Open-world default: an unregistered Type returns ok so the
+%% pipeline doesn't block activities the kernel hasn't yet learned
+%% about. Tightening to strict-world happens later in milestone 2.
+%%
+%% Activities with no :object skip the schema check (some verbs
+%% legitimately carry no object).
+%%
+%% The Erlang-fun shape is the substrate-friendly stand-in for the
+%% SX-source :schema bodies stored in the genesis bundle. Once an
+%% SX-source eval bridge exists, the same stage shape will dispatch
+%% through it instead — no API change.
+stage_schema(Activity, SchemaLookup) ->
+    case envelope:get_field(type, Activity) of
+        not_found -> {error, no_type};
+        {ok, Type} ->
+            case SchemaLookup(Type) of
+                not_found -> ok;
+                {ok, SchemaFn} ->
+                    check_object_schema(Activity, SchemaFn)
+            end
+    end.
+
+check_object_schema(Activity, SchemaFn) ->
+    case envelope:get_field(object, Activity) of
+        not_found -> ok;
+        {ok, Obj} ->
+            case SchemaFn(Obj) of
+                true -> ok;
+                false -> {error, schema_mismatch}
+            end
+    end.
+
+stage_schema(SchemaLookup) ->
+    fun (Activity) -> stage_schema(Activity, SchemaLookup) end.

next/tests/pipeline_schema.sh

@@ -0,0 +1,137 @@
+#!/usr/bin/env bash
+# next/tests/pipeline_schema.sh — Step 6c-schema-pure test.
+#
+# Exercises stage_schema/2 (direct call) and stage_schema/1
+# (factory). The SchemaLookup callback returns either
+# {ok, SchemaFn} or not_found; open-world default means
+# not_found resolves to ok. 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
+
+# Common: a strict Pin schema requires Object to have :path and :cid
+# `PinSchema = fun (Obj) -> ...`.
+PRELUDE='PinSchema = fun (Obj) -> case envelope:get_field(path, Obj) of {ok, _} -> case envelope:get_field(cid, Obj) of {ok, _} -> true; _ -> false end; _ -> false end end, PinLookup = fun (pin) -> {ok, PinSchema}; (_) -> not_found end,'
+
+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/envelope.erl\")) :name)")
+(epoch 3)
+(eval "(get (erlang-load-module (file-read \"next/kernel/pipeline.erl\")) :name)")
+
+;; Open-world default: unknown type returns ok
+(epoch 10)
+(eval "(get (erlang-eval-ast \"NoLookup = fun (_) -> not_found end, pipeline:stage_schema([{type, foo}, {object, bar}], NoLookup) =:= ok\") :name)")
+
+;; Activity without :type -> {error, no_type}
+(epoch 11)
+(eval "(get (erlang-eval-ast \"NoLookup = fun (_) -> not_found end, pipeline:stage_schema([{object, x}], NoLookup) =:= {error, no_type}\") :name)")
+
+;; Known type, schema passes -> ok
+(epoch 12)
+(eval "(get (erlang-eval-ast \"${PRELUDE} Act = [{type, pin}, {object, [{path, <<47,97>>}, {cid, <<98>>}]}], pipeline:stage_schema(Act, PinLookup) =:= ok\") :name)")
+
+;; Known type, schema fails -> {error, schema_mismatch}
+(epoch 13)
+(eval "(get (erlang-eval-ast \"${PRELUDE} Act = [{type, pin}, {object, [{path, <<47,97>>}]}], pipeline:stage_schema(Act, PinLookup) =:= {error, schema_mismatch}\") :name)")
+
+;; Activity with no :object skips schema check
+(epoch 14)
+(eval "(get (erlang-eval-ast \"${PRELUDE} pipeline:stage_schema([{type, pin}], PinLookup) =:= ok\") :name)")
+
+;; stage_schema/1 returns a function
+(epoch 15)
+(eval "(get (erlang-eval-ast \"is_function(pipeline:stage_schema(fun (_) -> not_found end))\") :name)")
+
+;; Factory + activity -> applies the lookup
+(epoch 16)
+(eval "(get (erlang-eval-ast \"${PRELUDE} Stage = pipeline:stage_schema(PinLookup), Stage([{type, pin}, {object, [{path, <<1>>}, {cid, <<2>>}]}]) =:= ok\") :name)")
+
+;; Factory + bad activity -> schema_mismatch
+(epoch 17)
+(eval "(get (erlang-eval-ast \"${PRELUDE} Stage = pipeline:stage_schema(PinLookup), Stage([{type, pin}, {object, [{path, <<1>>}]}]) =:= {error, schema_mismatch}\") :name)")
+
+;; Composed with stage_envelope via run_stages: bad envelope halts first
+(epoch 18)
+(eval "(get (erlang-eval-ast \"${PRELUDE} Stages = [fun (A) -> pipeline:stage_envelope(A) end, pipeline:stage_schema(PinLookup)], case pipeline:run_stages([{type, pin}], Stages) of {error, {missing_field, _}} -> ok; _ -> bad end\") :name)")
+
+;; Composed: envelope ok + schema fail -> schema_mismatch
+(epoch 19)
+(eval "(get (erlang-eval-ast \"${PRELUDE} Act = [{id, 1}, {type, pin}, {actor, alice}, {published, 1}, {signature, [{key_id, k}, {algorithm, e}, {value, v}]}, {object, [{path, <<1>>}]}], Stages = [fun (A) -> pipeline:stage_envelope(A) end, pipeline:stage_schema(PinLookup)], pipeline:run_stages(Act, Stages) =:= {error, schema_mismatch}\") :name)")
+
+;; Schema fn receives the object (verify by mutating an Erlang process flag isn't reliable; instead capture & test inside the schema)
+(epoch 20)
+(eval "(get (erlang-eval-ast \"Captor = fun (Obj) -> envelope:get_field(target, Obj) =:= {ok, mark} end, Lookup = fun (_) -> {ok, Captor} end, pipeline:stage_schema([{type, t}, {object, [{target, mark}]}], Lookup) =:= ok\") :name)")
+
+;; Multiple types registered: only matching one consulted
+(epoch 21)
+(eval "(get (erlang-eval-ast \"PinF = fun (_) -> true end, NoteF = fun (_) -> false end, Multi = fun (pin) -> {ok, PinF}; (note) -> {ok, NoteF}; (_) -> not_found end, {pipeline:stage_schema([{type, pin}, {object, ignored}], Multi), pipeline:stage_schema([{type, note}, {object, ignored}], Multi), pipeline:stage_schema([{type, other}, {object, ignored}], Multi)} =:= {ok, {error, schema_mismatch}, ok}\") :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  "envelope module loaded"             "envelope"
+check  3  "pipeline module loaded"             "pipeline"
+check 10  "open-world default for unknown"     "true"
+check 11  "no :type -> no_type error"          "true"
+check 12  "schema accepts -> ok"               "true"
+check 13  "schema rejects -> mismatch"         "true"
+check 14  "no :object skips check"             "true"
+check 15  "stage_schema/1 returns fun"         "true"
+check 16  "factory + ok"                       "true"
+check 17  "factory + mismatch"                 "true"
+check 18  "envelope halt before schema"        "ok"
+check 19  "envelope ok + schema mismatch"      "true"
+check 20  "schema fn receives object"          "true"
+check 21  "multi-type lookup dispatches"       "true"
+
+TOTAL=$((PASS+FAIL))
+if [ $FAIL -eq 0 ]; then
+  echo "ok $PASS/$TOTAL next/tests/pipeline_schema.sh passed"
+else
+  echo "FAIL $PASS/$TOTAL passed, $FAIL failed:"
+  echo "$ERRORS"
+fi
+[ $FAIL -eq 0 ]

plans/fed-sx-milestone-1.md

@@ -390,7 +390,7 @@ projection fold maintains it.)
 - [x] **6b-env** — `pipeline:stage_envelope/1` delegating to `envelope:validate_shape/1`; wired into both `inbound_stages` and `outbound_stages`. `next/tests/pipeline_envelope.sh` (12 cases); pipeline_driver.sh updated to test the driver in isolation.
 - [x] **6b-sig** — `pipeline:stage_signature/2` (direct call) + `stage_signature/1` (factory returning a context-bound stage fun). Not wired into default stage lists since ActorState isn't available at static-list build time; callers compose by `Stages = [..., pipeline:stage_signature(AS)]`. `next/tests/pipeline_signature.sh` (11 cases) covers direct + factory + composition + halt behaviour with stage_envelope.
 - [x] **6c-replay** — `pipeline:stage_replay/2` (direct) + `stage_replay/1` (factory closed over LogState). Checks the log entries for an existing activity with the same `:id`. Returns `{error, replay}` on duplicate, `{error, no_id}` when missing. `next/tests/pipeline_replay.sh` (12 cases).
-- [ ] **6c-schema** — `stage_activity_schema/1` (registry lookup of activity-type, evaluate :schema body) — blocked behind SX-source eval bridge.
+- [x] **6c-schema-pure** — `pipeline:stage_schema/2` (direct) + `stage_schema/1` (factory closed over a SchemaLookup callback). SchemaLookup is `fun(Type) -> {ok, SchemaFn} | not_found`; SchemaFn is `fun(Object) -> bool`. Open-world default: unknown type → ok; no :object skips the check. `next/tests/pipeline_schema.sh` (14 cases). SX-source eval bridge will plug into the same shape later.
 - [x] **6d-cs** — `outbox:construct/4` (skeleton + CID-derived :id via `cid:to_string`) + `outbox:sign/2` (HMAC over canonical bytes, append :signature pair from KeySpec) + `cid_of/1` accessor. Verified end-to-end: construct→sign→envelope:verify_signature passes; wrong key material fails with bad_signature. `next/tests/outbox_construct.sh` (13 cases).
 - [x] **6d-publish** — `outbox:publish/2(Request, Context)` orchestrates construct + sign + `pipeline:run_stages([envelope, signature, replay])` + `log:append`. Returns `{ok, [{cid, _}, {activity, _}], NewLog}` or `{error, Reason, LogState}` on stage halt. Replay catches duplicate publishes; bad key material surfaces `bad_signature`. `next/tests/outbox_publish.sh` (13 cases).
 - [ ] **6e** — HTTP handler for POST /activity glue (depends on Step 8 http server)
@@ -1000,6 +1000,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 6c-schema-pure: `pipeline:stage_schema/2` accepts (Activity, SchemaLookup) where SchemaLookup is a caller-supplied callback `fun(Type) -> {ok, SchemaFn} | not_found`. Open-world default — unregistered types resolve to ok so the pipeline doesn't block activities the kernel hasn't yet learned about (tightened to strict-world in milestone 2). Activities without `:object` skip the schema check. `stage_schema/1` returns a 1-arity stage fun closed over SchemaLookup for composition with run_stages. Halt order verified end-to-end: envelope-shape errors precede schema; envelope-ok + schema-fail surfaces `schema_mismatch`. The Erlang-fun shape is the substrate-friendly stand-in for the SX `:schema` bodies in genesis; same stage shape will dispatch through an SX-source eval bridge once it exists. `next/tests/pipeline_schema.sh` 14/14. Erlang conformance 729/729.
 - **2026-05-28** — Step 8d-dispatch-get: format-aware versions of every GET response builder. `actor_doc_response_for/2`, `artifact_response_for/2`, `projection_response_for/2`, `projections_list_response_for/1`. Each produces `{"key":"value"}` (json/activity_json), `(key "value")` (sx), raw payload bytes (cbor stub), or the existing text form. `dispatch` refactored to `/3` with a backward-compat `dispatch/2` wrapper. Route extracts Format via `accept_format_from/1` once at the top and threads it through dispatch. End-to-end GETs with `Accept: application/json` / `application/sx` verified for all three dynamic-prefix routes + the projections-list bare-path route. Step 8d effectively complete — format dispatch + Content-Type live on every non-static response. `next/tests/http_get_format.sh` 17/17. Erlang conformance 729/729.
 - **2026-05-28** — Step 8d-dispatch-post: `handle_post_activity` extracts the Accept format via `accept_format_from/1` and threads it into `publish_if_kernel/2`. Both success paths emit format-specific bodies: `cid_response_for/2` produces `{"cid":"<cid>"}\n` (json/activity_json), `(cid "<cid>")\n` (sx), raw CID bytes (cbor), or the existing text form; `post_activity_response_for/1` mirrors for the kernel-absent stub. Each response carries the matching Content-Type. End-to-end POSTs with `Accept: application/json` / `application/sx` verified through the full HTTP→nx_kernel→publish→cid_response_for chain. `next/tests/http_post_format.sh` 13/13. Erlang conformance 729/729.
 - **2026-05-28** — Step 8d-content-type: `content_type_for/1` maps format atoms to MIME-type binaries — text/plain (10b), application/json (16b), application/activity+json (25b), application/sx (14b), application/cbor (16b); unknown formats fall through to text/plain. `ok_response/2(Body, Format)` constructs a 200 response with `{headers, [{<<"content-type">>, MIME}]}`. Lowercase header key matches how the BIF wrapper normalises request headers. `ok_response/1` still produces the empty-headers shape — backward compat preserved. `next/tests/http_content_type.sh` 13/13. Erlang conformance 729/729.