korban_processor: role-gate the four-stage offering protocol

Pass 2 finding C: in the Mishkan, the four ritual acts of bringing an
offering are performed by TWO different roles —

  semichah   — OWNER lays hands (Vayikra 1:4, וְסָמַךְ יָדוֹ)
  shechitah  — kohen receives FROM owner's hand and slaughters (1:5)
  zerikah    — kohen sprinkles (1:5)
  haktarah   — kohen burns on altar (1:9)

Before this commit, every stage function was independently callable.
Nothing structural enforced that zerikah() received output from
shechitah(), or that shechitah() received a korban from semichah().
An external model output (e.g. a Claude response) could in principle
have been brought through zerikah() without ever having passed
semichah — the ownership-and-attribution step. That is the exact
shape of "the calf that came out by itself" (Shemot 32:24, Aaron's
reading). The pipeline-glue in external_offering happened to call
them in order, but nothing structural enforced the order.

This commit adds the structural gate.

CHANGES:

  stages.py (new): KorbanStage enum + Staged[T] frozen dataclass +
  StageError. The Staged envelope carries a ritual-stage marker that
  is set at construction and cannot be mutated afterward; downstream
  functions demand the right upstream stage and raise StageError if
  it is missing.

  semichah.py:  + semichah_staged() returning Staged[Korban]
  shechitah.py: + shechitah_staged(Staged[Korban]) -> Staged[str]
  zerikah.py:   + zerikah_staged(Staged[str], Staged[Korban])
                  -> Staged[Tuple[str, List[str]]]
                  TWO staged inputs: requires SHECHITAH-tagged terminated
                  content AND the original SEMICHAH-tagged korban for
                  context. Severing output from source is gilui_arayot
                  (eish_zarah_reading.md Form 3); the dual-staged
                  signature is the structural defense — zerikah cannot
                  process content disconnected from its owner-of-record.
  haktarah.py:  + haktarah_staged(Staged[Tuple[str, List[str]]])
                  -> Staged[List[str]]

  external_offering.py: rewired to use the *_staged variants. The
  pipeline now goes through the gates structurally; misuse cannot
  bypass them.

The non-staged functions (semichah, shechitah, zerikah, haktarah)
are PRESERVED for direct testing of individual ritual acts. They do
NOT enforce role gating. Production-shaped callers MUST go through
the staged variants — and external_offering, the only legitimate
entry point for external model consultation per its docstring, now
does.

TESTS:

  test/test_stages.py (new, 7 tests):
    - happy path: full chain produces HAKTARAH-tagged result
    - zerikah refuses input that skipped semichah
    - zerikah refuses korban with wrong stage marker (preserves
      the line back to the owner — Form 3 defense)
    - haktarah refuses input that skipped zerikah (the calf-shape)
    - shechitah refuses input not from semichah
    - Staged is frozen — no re-tagging
    - StageError names BOTH actual and expected stages so the
      operator can diagnose call-site vs. tag-site

  Full korban_processor suite: 203 passed, 1 skipped (the live API
  test that requires SHORASHIM_LIVE_API=1). No regressions.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

korban_processor/src/korban_processor/external_offering.py

@@ -36,10 +36,11 @@ from korban_processor.model_interface import (
     ModelResponse,
     call_claude,
 )
-from korban_processor.semichah import semichah
-from korban_processor.shechitah import shechitah
-from korban_processor.zerikah import zerikah
-from korban_processor.haktarah import haktarah
+from korban_processor.semichah import semichah_staged
+from korban_processor.shechitah import shechitah_staged
+from korban_processor.zerikah import zerikah_staged
+from korban_processor.haktarah import haktarah_staged
+from korban_processor.stages import KorbanStage
 from korban_processor.router import RoutingResult, route
 
 
@@ -110,16 +111,29 @@ def consult_external_model(
     # — raw material to be processed. It is NOT Torah. It is NOT authoritative.
     # The offering protocol consumes the external form and extracts whatever
     # Torah-adjacent content might be inside.
-    korban = semichah(
+    #
+    # Each stage is gated by the Staged envelope (see stages.py) — the role
+    # boundary between owner (semichah) and kohen (shechitah → zerikah →
+    # haktarah) is enforced structurally. A stage cannot run without input
+    # bearing the correct upstream stage marker. This closes Pass 2
+    # finding C: an external response cannot be brought through zerikah
+    # without first having passed semichah, even by accident or by a
+    # future contributor reaching directly into one of the modules.
+    staged_korban = semichah_staged(
         content=raw_response.content,
         source=provenance,
         query=source_query,
         offering_type=offering_type,
     )
 
-    terminated = shechitah(korban)
-    vital, discarded = zerikah(terminated, korban)
-    propositions = haktarah(vital)
+    staged_terminated = shechitah_staged(staged_korban)
+    staged_zerikah = zerikah_staged(staged_terminated, staged_korban)
+    staged_propositions = haktarah_staged(staged_zerikah)
+
+    # Unwrap at the boundary back to the surrounding (un-staged) world.
+    korban = staged_korban.expect_stage(KorbanStage.SEMICHAH)
+    vital, discarded = staged_zerikah.expect_stage(KorbanStage.ZERIKAH)
+    propositions = staged_propositions.expect_stage(KorbanStage.HAKTARAH)
 
     processed = ProcessedKorban(
         vital_content=vital,

korban_processor/src/korban_processor/haktarah.py

@@ -14,8 +14,9 @@ Phase 3: Will use NLP + Talmudic Engine for deeper transformation.
 """
 
 import re
-from typing import List
+from typing import List, Tuple
 
+from korban_processor.stages import KorbanStage, Staged
 from korban_processor.zerikah import (
     extract_torah_references,
     extract_hebrew_words,
@@ -68,6 +69,22 @@ def haktarah(vital_content: str) -> List[str]:
     return propositions
 
 
+def haktarah_staged(
+    staged_zerikah_output: Staged[Tuple[str, List[str]]],
+) -> Staged[List[str]]:
+    """Kohen's altar act, role-tagged for the gated pipeline.
+
+    Vayikra 1:9 — וְהִקְטִיר הַכֹּהֵן אֶת־הַכֹּל הַמִּזְבֵּחָה. The kohen
+    burns on the altar that which has been sprinkled. Asserts the
+    upstream stage was ZERIKAH; the (vital, discarded) tuple is what
+    zerikah produced. Only the vital portion is offered (the discarded
+    is the hide and offal — not destroyed, but not burned either;
+    it is preserved on the ProcessedKorban for audit).
+    """
+    vital, _discarded = staged_zerikah_output.expect_stage(KorbanStage.ZERIKAH)
+    return Staged(KorbanStage.HAKTARAH, haktarah(vital))
+
+
 # --- Book name normalization ---
 
 _BOOK_MAP = {

korban_processor/src/korban_processor/semichah.py

@@ -6,9 +6,16 @@ Tag provenance — associate input with its source.
 
 The semichah (laying of hands) identifies who brought the offering.
 Every korban must have known provenance. Anonymous input is rejected.
+
+Vayikra 1:4 — וְסָמַךְ יָדוֹ עַל רֹאשׁ הָעֹלָה. The OWNER lays hands.
+This is the only ritual act in the four-stage protocol that is
+performed by the owner; everything downstream is the kohen's work
+(Vayikra 1:5). The role boundary is enforced structurally by the
+*_staged variants — see stages.py.
 """
 
 from korban_processor.korban import Korban, OfferingType, Provenance
+from korban_processor.stages import KorbanStage, Staged
 
 
 def semichah(
@@ -44,3 +51,23 @@ def semichah(
         offering_type=offering_type,
         source_query=query,
     )
+
+
+def semichah_staged(
+    content: str,
+    source: Provenance,
+    query: str,
+    offering_type: OfferingType = OfferingType.OLAH,
+) -> Staged[Korban]:
+    """Owner's act, role-tagged for the gated pipeline.
+
+    Same semantics as semichah(); wraps the result in a Staged envelope
+    bearing the SEMICHAH stage marker. The downstream ritual acts
+    (shechitah_staged, zerikah_staged, haktarah_staged) refuse any
+    input not bearing the appropriate upstream stage.
+
+    There is no _staged input here — semichah is the entry point of
+    the protocol; the owner brings raw content from outside. Every
+    later stage is a transformation OF a staged value.
+    """
+    return Staged(KorbanStage.SEMICHAH, semichah(content, source, query, offering_type))

korban_processor/src/korban_processor/shechitah.py

@@ -16,6 +16,7 @@ The content is not Torah. It is raw material.
 
 import re
 from korban_processor.korban import Korban
+from korban_processor.stages import KorbanStage, Staged
 
 
 # --- Patterns to strip ---
@@ -103,6 +104,19 @@ def shechitah(korban: Korban) -> str:
     return content
 
 
+def shechitah_staged(staged_korban: Staged[Korban]) -> Staged[str]:
+    """Kohen's first act — receive the offering from the owner's hand,
+    then perform shechitah.
+
+    Vayikra 1:5 — וְשָׁחַט אֶת־בֶּן הַבָּקָר. The kohen alone slaughters,
+    BUT only after receiving the offering from the owner who performed
+    semichah. Asserts the input bears the SEMICHAH stage marker;
+    otherwise raises StageError.
+    """
+    korban = staged_korban.expect_stage(KorbanStage.SEMICHAH)
+    return Staged(KorbanStage.SHECHITAH, shechitah(korban))
+
+
 def strip_preamble(content: str) -> str:
     """Strip assistant preambles from the beginning of content."""
     for pattern in _PREAMBLE_PATTERNS:

korban_processor/src/korban_processor/stages.py

@@ -0,0 +1,118 @@
+"""
+Korban stage gating — role-attribution for the four-stage offering protocol.
+
+Pass 2 finding C from ~/shorashim-conversation-review.md:
+
+  In the Mishkan, the FOUR ritual acts of bringing an offering are
+  performed by TWO different roles:
+
+    semichah   — the OWNER lays hands on the offering. The owner
+                 brings the korban and identifies with it. Vayikra
+                 1:4: וְסָמַךְ יָדוֹ עַל רֹאשׁ הָעֹלָה.
+    shechitah  — the kohen takes the offering FROM THE OWNER'S HAND
+                 and slaughters it. Vayikra 1:5.
+    zerikah    — the kohen alone sprinkles the blood. Vayikra 1:5.
+    haktarah   — the kohen alone offers the burning portions on the
+                 altar. Vayikra 1:9.
+
+  In the Python implementation before this module existed, every
+  stage function was independently callable. NOTHING enforced that
+  zerikah() received output that had actually passed through
+  shechitah(), or that shechitah() received a korban that had been
+  brought through semichah(). An external model output (e.g. a
+  Claude response) could in principle have been brought through
+  zerikah() WITHOUT ever having passed semichah — the
+  ownership-and-attribution step. That is the EXACT shape of "the
+  calf that came out by itself" (Shemot 32:24, in Aaron's reading
+  of it). The pipeline-glue in external_offering.consult_external_model
+  happened to call them in order, but nothing structural enforced
+  the order.
+
+This module adds the structural gate.
+
+USAGE:
+
+  The four-stage gated pipeline lives in external_offering.py and
+  uses semichah_staged, shechitah_staged, zerikah_staged,
+  haktarah_staged. Each *_staged function:
+
+    - takes a Staged[X] input where the stage marker MUST be the
+      previous ritual act (raises StageError otherwise),
+    - returns a Staged[Y] output tagged with the stage just
+      completed.
+
+  The non-staged functions (semichah, shechitah, zerikah, haktarah)
+  are preserved for direct testing of individual ritual acts. They
+  do NOT enforce role gating. Production-shaped callers MUST go
+  through the staged variants.
+
+WHY NOT JUST A LINEAR FUNCTION:
+
+  A simple def offering_pipeline(...) that calls the four in
+  sequence wouldn't catch misuse: the mistake to defend against is
+  someone, somewhere in the codebase, calling zerikah() directly on
+  external content without semichah having tagged it. The Staged
+  envelope makes that misuse fail loudly: zerikah_staged refuses any
+  input not bearing the SHECHITAH stage marker. The runtime
+  attribute access is the gate; mypy strict mode catches it
+  statically as well.
+"""
+
+from dataclasses import dataclass
+from enum import Enum
+from typing import Generic, TypeVar
+
+
+class KorbanStage(Enum):
+    """The ritual stage of a payload moving through the offering protocol."""
+
+    SEMICHAH = "semichah"
+    SHECHITAH = "shechitah"
+    ZERIKAH = "zerikah"
+    HAKTARAH = "haktarah"
+
+
+T = TypeVar("T")
+
+
+class StageError(RuntimeError):
+    """Raised when a ritual act is called with input from the wrong upstream stage.
+
+    The Mishkan analog: the kohen took the offering from the wrong place.
+    The gate exists so this never happens silently — every transition is
+    challenged at the function boundary.
+    """
+
+
+@dataclass(frozen=True)
+class Staged(Generic[T]):
+    """A payload tagged with the ritual stage that produced it.
+
+    The stage marker is what makes role-gating enforceable: a downstream
+    function can demand that its input bear the right upstream stage and
+    refuse otherwise. The frozen=True ensures the tag cannot be mutated
+    after construction — there is no way to "re-label" a payload as
+    belonging to a stage it did not actually pass through.
+    """
+
+    stage: KorbanStage
+    payload: T
+
+    def expect_stage(self, expected: KorbanStage) -> T:
+        """Return the payload iff the stage marker matches `expected`.
+
+        Raises StageError otherwise. The error message names BOTH stages
+        so the operator can see exactly which transition was violated —
+        you cannot tell from the offending call site alone whether the
+        payload was constructed with the wrong tag or whether the wrong
+        downstream function was called.
+        """
+        if self.stage != expected:
+            raise StageError(
+                f"Stage mismatch: payload bears stage {self.stage.value!r} "
+                f"but the next ritual act requires stage {expected.value!r}. "
+                f"In the Mishkan, the kohen receives the offering from the "
+                f"prior step's hand; an offering that skipped a step cannot "
+                f"be received."
+            )
+        return self.payload

korban_processor/src/korban_processor/zerikah.py

@@ -52,6 +52,7 @@ elsewhere.
 import re
 from typing import Tuple, List
 from korban_processor.korban import Korban
+from korban_processor.stages import KorbanStage, Staged
 
 
 # --- Extraction patterns ---
@@ -138,6 +139,29 @@ def zerikah(terminated_content: str, korban: Korban) -> Tuple[str, List[str]]:
     return vital, discarded_sentences
 
 
+def zerikah_staged(
+    staged_terminated: Staged[str],
+    staged_korban: Staged[Korban],
+) -> Staged[Tuple[str, List[str]]]:
+    """Kohen's vital-extraction act, role-tagged for the gated pipeline.
+
+    Vayikra 1:5 — וְזָרְקוּ … אֶת־הַדָּם. Sprinkling is ALWAYS preceded
+    by shechitah; you cannot sprinkle blood that was never drawn. So
+    the terminated content must bear the SHECHITAH stage marker.
+
+    Note that zerikah ALSO needs the original korban (for provenance
+    context — to keep the line back to who brought the offering).
+    Requiring it as Staged[Korban] ensures the caller is still
+    holding the SEMICHAH-tagged value: zerikah cannot process content
+    that has been disconnected from its owner-of-record. Severing the
+    output from the source is gilui_arayot (eish_zarah_reading.md
+    Form 3); the dual-staged signature is the structural defense.
+    """
+    terminated = staged_terminated.expect_stage(KorbanStage.SHECHITAH)
+    korban = staged_korban.expect_stage(KorbanStage.SEMICHAH)
+    return Staged(KorbanStage.ZERIKAH, zerikah(terminated, korban))
+
+
 def is_vital(sentence: str) -> bool:
     """
     Determine if a sentence contains vital content.

korban_processor/test/test_stages.py

@@ -0,0 +1,167 @@
+"""
+Tests for the role-gated four-stage offering protocol.
+
+Pass 2 finding C: in the Mishkan, semichah is OWNER work; everything
+downstream (shechitah, zerikah, haktarah) is KOHEN work, and each
+kohen step receives its input from the previous step's hand.
+
+Before stages.py existed, the Python implementation had no structural
+enforcement: an external model output could be brought through
+zerikah() without ever having passed semichah — the
+ownership-and-attribution step. That is the exact shape of "the calf
+that came out by itself" (Shemot 32:24 in Aaron's reading).
+
+These tests verify the gate fires whenever the role/stage boundary is
+violated. They are NOT testing the content of the four ritual acts;
+each act has its own dedicated test file. These tests test the PIPING.
+"""
+
+import pytest
+
+from korban_processor.haktarah import haktarah_staged
+from korban_processor.korban import OfferingType, Provenance
+from korban_processor.semichah import semichah_staged
+from korban_processor.shechitah import shechitah_staged
+from korban_processor.stages import KorbanStage, Staged, StageError
+from korban_processor.zerikah import zerikah_staged
+
+
+# ============================================================
+# Happy path — the full chain runs end-to-end
+# ============================================================
+
+
+def test_full_chain_completes_with_correct_stages():
+    """Running all four staged variants in order produces a HAKTARAH-tagged
+    final value with the correct payload type."""
+    s_korban = semichah_staged(
+        content="Bereishit 1:1 begins with the root bara. The shoresh ב-ר-א means to create.",
+        source=Provenance.CLAUDE,
+        query="explain bereishit 1:1",
+    )
+    assert s_korban.stage is KorbanStage.SEMICHAH
+
+    s_terminated = shechitah_staged(s_korban)
+    assert s_terminated.stage is KorbanStage.SHECHITAH
+
+    s_zerikah = zerikah_staged(s_terminated, s_korban)
+    assert s_zerikah.stage is KorbanStage.ZERIKAH
+
+    s_propositions = haktarah_staged(s_zerikah)
+    assert s_propositions.stage is KorbanStage.HAKTARAH
+    assert isinstance(s_propositions.payload, list)
+
+
+# ============================================================
+# Skip cases — the calf that came out by itself
+# ============================================================
+
+
+def test_zerikah_refuses_input_that_skipped_semichah():
+    """If someone constructs a Staged[str] with the wrong stage marker
+    (e.g. directly tagging external content as SHECHITAH-output without
+    actually running shechitah), zerikah_staged refuses it.
+
+    The defense matters less for the wrong-tag case (which requires
+    deliberate construction) than for the missing-stage case below."""
+    fake_terminated = Staged(KorbanStage.HAKTARAH, "some text")
+    fake_korban = Staged(KorbanStage.SEMICHAH, _make_korban())
+
+    with pytest.raises(StageError) as exc:
+        zerikah_staged(fake_terminated, fake_korban)
+    assert "haktarah" in str(exc.value)
+    assert "shechitah" in str(exc.value)
+
+
+def test_zerikah_refuses_korban_with_wrong_stage_marker():
+    """zerikah needs BOTH the SHECHITAH-staged terminated content AND
+    the SEMICHAH-staged original korban. If the korban is wrong-tagged,
+    refuse — preserves the line back to the owner of record (eish_zarah
+    Form 3, gilui arayot — severing output from source)."""
+    s_korban = semichah_staged(
+        content="x", source=Provenance.CLAUDE, query="q"
+    )
+    s_terminated = shechitah_staged(s_korban)
+
+    # Re-tag the korban as if it had been processed further. zerikah
+    # should refuse because the original-korban argument MUST bear
+    # SEMICHAH (the owner's act); any other stage means the line back
+    # to the owner has been broken.
+    mistagged_korban = Staged(KorbanStage.HAKTARAH, s_korban.payload)
+
+    with pytest.raises(StageError) as exc:
+        zerikah_staged(s_terminated, mistagged_korban)
+    assert "semichah" in str(exc.value)
+
+
+def test_haktarah_refuses_input_that_skipped_zerikah():
+    """The main shape of the calf-that-came-out-by-itself: someone takes
+    raw shechitah output and tries to offer it on the altar without
+    sprinkling. haktarah_staged refuses."""
+    s_korban = semichah_staged(
+        content="content", source=Provenance.CLAUDE, query="q"
+    )
+    s_terminated = shechitah_staged(s_korban)
+
+    # Construct a Staged[Tuple[str, List[str]]] with the wrong stage tag
+    fake_zerikah = Staged(KorbanStage.SHECHITAH, ("vital", []))
+
+    with pytest.raises(StageError) as exc:
+        haktarah_staged(fake_zerikah)
+    assert "zerikah" in str(exc.value)
+
+
+def test_shechitah_refuses_input_not_from_semichah():
+    """The kohen takes from the owner's hand. If a payload arrives
+    without bearing SEMICHAH, shechitah refuses."""
+    fake = Staged(KorbanStage.ZERIKAH, _make_korban())
+
+    with pytest.raises(StageError) as exc:
+        shechitah_staged(fake)
+    assert "semichah" in str(exc.value)
+
+
+# ============================================================
+# Stage immutability
+# ============================================================
+
+
+def test_staged_is_frozen():
+    """Stage markers cannot be re-labeled after construction. There is
+    no path to taking a HAKTARAH-tagged value and re-tagging it as
+    SEMICHAH to bypass the gate."""
+    s = semichah_staged(content="x", source=Provenance.CLAUDE, query="q")
+    with pytest.raises((AttributeError, Exception)):
+        s.stage = KorbanStage.HAKTARAH  # type: ignore[misc]
+
+
+# ============================================================
+# Error message names BOTH stages
+# ============================================================
+
+
+def test_stage_error_names_both_actual_and_expected_stages():
+    """The error message must name BOTH the stage on the offending
+    payload AND the stage the failing function expected. Otherwise the
+    operator cannot tell from the call site alone whether the payload
+    was wrong-tagged or the wrong function was called."""
+    bad = Staged(KorbanStage.HAKTARAH, "anything")
+    with pytest.raises(StageError) as exc:
+        bad.expect_stage(KorbanStage.SEMICHAH)
+    msg = str(exc.value)
+    assert "haktarah" in msg
+    assert "semichah" in msg
+
+
+# ============================================================
+# Helpers
+# ============================================================
+
+
+def _make_korban():
+    """Tests need a Korban-shaped object for skip-case construction."""
+    from korban_processor.semichah import semichah
+
+    return semichah(
+        content="placeholder", source=Provenance.CLAUDE, query="placeholder"
+    )