fed-sx-m1: Step 6c-replay — pipeline:stage_replay/1,/2 (factory + direct) + 12 tests

next/kernel/pipeline.erl

@@ -3,7 +3,8 @@
          validate_inbound/1, validate_outbound/1,
          inbound_stages/0, outbound_stages/0,
          stage_envelope/1,
-         stage_signature/1, stage_signature/2]).
+         stage_signature/1, stage_signature/2,
+         stage_replay/1, stage_replay/2]).
 
 %% Validation pipeline per design §14.
 %%
@@ -62,3 +63,29 @@ stage_signature(Activity, ActorState) ->
 %% ActorState isn't available at static-list build time).
 stage_signature(ActorState) ->
     fun (Activity) -> envelope:verify_signature(Activity, ActorState) end.
+
+%% stage_replay/2 — checks the in-memory log for an existing
+%% activity with the same :id. Returns ok if the activity is new,
+%% `{error, replay}` if the log already carries it, `{error, no_id}`
+%% if the activity has no :id field. The check is linear scan of
+%% log entries; the projection scheduler (Step 7) will eventually
+%% maintain a CID index that turns this into O(1).
+stage_replay(Activity, LogState) ->
+    case envelope:get_field(id, Activity) of
+        not_found -> {error, no_id};
+        {ok, Id} ->
+            case log_has_id(Id, log:entries(LogState)) of
+                true  -> {error, replay};
+                false -> ok
+            end
+    end.
+
+stage_replay(LogState) ->
+    fun (Activity) -> stage_replay(Activity, LogState) end.
+
+log_has_id(_, []) -> false;
+log_has_id(Id, [Act | Rest]) ->
+    case envelope:get_field(id, Act) of
+        {ok, Id} -> true;
+        _        -> log_has_id(Id, Rest)
+    end.