diff --git a/korban_processor/src/korban_processor/external_offering.py b/korban_processor/src/korban_processor/external_offering.py index 40f7f636..2e4f2bb9 100644 --- a/korban_processor/src/korban_processor/external_offering.py +++ b/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, diff --git a/korban_processor/src/korban_processor/haktarah.py b/korban_processor/src/korban_processor/haktarah.py index b16b9c65..ac40eb25 100644 --- a/korban_processor/src/korban_processor/haktarah.py +++ b/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 = { diff --git a/korban_processor/src/korban_processor/semichah.py b/korban_processor/src/korban_processor/semichah.py index e107c3cd..b10fecfe 100644 --- a/korban_processor/src/korban_processor/semichah.py +++ b/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)) diff --git a/korban_processor/src/korban_processor/shechitah.py b/korban_processor/src/korban_processor/shechitah.py index 19c96b38..50b97c63 100644 --- a/korban_processor/src/korban_processor/shechitah.py +++ b/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: diff --git a/korban_processor/src/korban_processor/stages.py b/korban_processor/src/korban_processor/stages.py new file mode 100644 index 00000000..c8fc7c8b --- /dev/null +++ b/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 diff --git a/korban_processor/src/korban_processor/zerikah.py b/korban_processor/src/korban_processor/zerikah.py index 101184ae..529e7373 100644 --- a/korban_processor/src/korban_processor/zerikah.py +++ b/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. diff --git a/korban_processor/test/test_stages.py b/korban_processor/test/test_stages.py new file mode 100644 index 00000000..b6c460e5 --- /dev/null +++ b/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" + )