azazel_processor: commit 3 — shaliach + tevilah (property 4, role-separated operator)

Pass 4 / Finding F, commit 3 of 10 from BUILD_AZAZEL.md.

Vayikra 16:21 — וְשִׁלַּח בְּיַד־אִישׁ עִתִּי הַמִּדְבָּרָה.
And he shall send the goat into the wilderness by the hand of a
designated man (ish itti).

Vayikra 16:26 — וְהַמְשַׁלֵּחַ … יְכַבֵּס בְּגָדָיו וְרָחַץ אֶת־בְּשָׂרוֹ
בַּמַּיִם וְאַחֲרֵי־כֵן יָבוֹא אֶל־הַמַּחֲנֶה.
The sender washes his clothes and bathes his flesh (tevilah), and
AFTERWARD comes into the camp.

Yoma 67a: the ish itti is a dedicated operator role, prepared in
advance, distinct from the kohanim who handle sacred processing.
Contact with the disposal channel produces state that must be
reset before the operator rejoins the trusted boundary.

THE THREE ROLES IN SHORASHIM:

  Owner       — submits the offering (upstream caller)
  Kohen       — performs shechitah/zerikah/haktarah (korban_processor
                stages.py enforces owner/kohen separation)
  Shaliach    — receives input the kohen REFUSED and disposes of it
                through the Azazel protocol (THIS module)

Without tevilah, payload references would persist across disposals;
disposal-N would contaminate disposal-N+1 with information from N.
The Mishkan analog: the ish itti who returns from the wilderness
without tevilah brings desert-state into the camp.

CREATED:

  src/azazel_processor/shaliach.py        ~210 lines

    IshItti class:
      __init__       — sets _immutable_identifier (no other immutable
                       attrs; everything else must be cleared by tevilah)
      activate()     — begin a disposal; refuses if already active (no
                       state-mixing across disposals)
      hold_payload() — register payload ref for later destruction
      tag()/get_tag()— intermediate metadata during the disposal route
      tevilah()      — Vayikra 16:26: clears ALL non-_immutable_
                       attributes (discovered dynamically via vars(),
                       so future state additions are auto-cleared)
      identifier     — propagates onto AzazelRecords (commit 7) for
                       shaliach attribution

    ShaliachError    — raised on role-separation violations
    assert_shaliach_active(handler) — runtime gate every disposal-channel
                       function calls at entry; the type-system gate is
                       AzazelStaged (property 11). Refuses non-IshItti
                       callers AND inactive handlers. Identity-checked
                       (isinstance), not duck-typed.

THE LOAD-BEARING CI INVARIANT (test_tevilah_clears_all_non_immutable_state):

Walks vars(handler) post-tevilah and asserts every non-_immutable_
attribute is in a 'fresh' state (empty container, False, None, 0).
A future contributor who adds `self._cache = {}` (or any state
attribute) without extending tevilah() to clear it breaks this and
the test fires before the regression ships. The discovery is
DYNAMIC — no hand-maintained list of attributes to clear; tevilah
is implemented to clear all non-_immutable_ attrs and the test
verifies it does.

  test/test_shaliach.py                   23 tests covering:
    - construction (named identifier required, validates non-empty,
      non-whitespace, string)
    - identifier immutability across activate→tevilah→reactivate
    - lifecycle: starts inactive; activate marks active; re-activate
      without tevilah raises; tevilah→reactivate succeeds
    - hold_payload/tag require active; both work when active
    - tevilah clears payload refs, intermediate tags, sets inactive
    - tevilah preserves _immutable_identifier
    - THE CI INVARIANT (dynamic walk)
    - assert_shaliach_active: passes for active IshItti, refuses
      inactive, refuses non-IshItti, refuses duck-typed impostor
    - integration: stale-state-across-disposals prevented end-to-end

NOTE ON PYTEST INVOCATION:

When running azazel_processor tests, use `--rootdir=.` from the
package directory. Without it, pytest auto-discovers /Users/scottkarp/
shorashim as rootdir and ignores the local pyproject.toml's
testpaths config. Documented here so future runs match.

  cd azazel_processor
  PYTHONPATH=src:../korban_processor/src python3 -m pytest --rootdir=.

Tests: azazel_processor 71 passed (was 48, +23). korban_processor
unchanged at 203 passed, 1 skipped. No regressions.

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

azazel_processor/test/test_shaliach.py

@@ -0,0 +1,319 @@
+"""
+Tests for shaliach + tevilah — property 4 (role-separated operator)
+of azazel_engineering_reading.md §14.
+
+The disposal handler (IshItti, "designated man") is the ONLY role
+authorized to invoke disposal-channel functions. The handler
+accumulates state during a disposal (payload references, intermediate
+tags) that MUST be cleared by tevilah before the handler can be
+re-used or before the same process touches non-azazel code.
+
+The CI invariant — test_tevilah_clears_all_non_immutable_state —
+catches future contributors who add state-bearing attributes to
+IshItti without extending tevilah() to clear them. Without that
+invariant, the property would silently degrade as the codebase
+evolves.
+"""
+
+import pytest
+
+from azazel_processor.shaliach import (
+    IshItti,
+    ShaliachError,
+    assert_shaliach_active,
+)
+
+
+# ============================================================
+# Construction
+# ============================================================
+
+
+def test_handler_requires_named_identifier():
+    h = IshItti("aharon")
+    assert h.identifier == "aharon"
+
+
+def test_handler_refuses_empty_identifier():
+    with pytest.raises(ShaliachError):
+        IshItti("")
+
+
+def test_handler_refuses_whitespace_identifier():
+    with pytest.raises(ShaliachError):
+        IshItti("   ")
+
+
+def test_handler_refuses_non_string_identifier():
+    with pytest.raises(ShaliachError):
+        IshItti(None)  # type: ignore[arg-type]
+    with pytest.raises(ShaliachError):
+        IshItti(123)  # type: ignore[arg-type]
+
+
+def test_identifier_is_immutable():
+    """The identifier is set at construction and propagates onto
+    AzazelRecords for shaliach attribution. It must not be
+    re-assignable, and it must survive tevilah."""
+    h = IshItti("aharon")
+    h.activate()
+    h.tevilah()
+    assert h.identifier == "aharon"
+
+
+# ============================================================
+# Activation lifecycle
+# ============================================================
+
+
+def test_handler_starts_inactive():
+    h = IshItti("aharon")
+    assert h.is_active() is False
+
+
+def test_activate_marks_handler_active():
+    h = IshItti("aharon")
+    h.activate()
+    assert h.is_active() is True
+
+
+def test_activate_when_already_active_raises():
+    """Re-activation without tevilah would mix two disposals' state.
+    Refuse — the handler must wash before rejoining (Vayikra 16:26)."""
+    h = IshItti("aharon")
+    h.activate()
+    with pytest.raises(ShaliachError) as exc:
+        h.activate()
+    assert "tevilah" in str(exc.value).lower()
+
+
+def test_tevilah_then_reactivate_succeeds():
+    """The handler can perform many disposals across its lifetime,
+    one at a time, with mandatory tevilah between."""
+    h = IshItti("aharon")
+    h.activate()
+    h.tevilah()
+    assert h.is_active() is False
+    h.activate()
+    assert h.is_active() is True
+
+
+# ============================================================
+# Hold payload + tag — only when active
+# ============================================================
+
+
+def test_hold_payload_requires_active():
+    h = IshItti("aharon")
+    with pytest.raises(ShaliachError):
+        h.hold_payload({"data": "x"})
+
+
+def test_hold_payload_tracks_references():
+    h = IshItti("aharon")
+    h.activate()
+    h.hold_payload({"data": "first"})
+    h.hold_payload({"data": "second"})
+    assert h.held_payload_count() == 2
+
+
+def test_tag_requires_active():
+    h = IshItti("aharon")
+    with pytest.raises(ShaliachError):
+        h.tag("checkpoint", "STAGED")
+
+
+def test_tag_round_trip():
+    h = IshItti("aharon")
+    h.activate()
+    h.tag("checkpoint", "STAGED")
+    assert h.get_tag("checkpoint") == "STAGED"
+    assert h.get_tag("missing") is None
+    assert h.get_tag("missing", "default") == "default"
+
+
+# ============================================================
+# Tevilah — the load-bearing reset
+# ============================================================
+
+
+def test_tevilah_clears_payload_refs():
+    h = IshItti("aharon")
+    h.activate()
+    h.hold_payload({"data": "x"})
+    assert h.held_payload_count() == 1
+
+    h.tevilah()
+    assert h.held_payload_count() == 0
+
+
+def test_tevilah_clears_intermediate_tags():
+    h = IshItti("aharon")
+    h.activate()
+    h.tag("checkpoint", "STAGED")
+    h.tag("vidui_id", "abc")
+
+    h.tevilah()
+    h.activate()  # re-activate to allow get_tag (which doesn't actually require active, but for clarity)
+    assert h.get_tag("checkpoint") is None
+    assert h.get_tag("vidui_id") is None
+
+
+def test_tevilah_returns_handler_to_inactive():
+    h = IshItti("aharon")
+    h.activate()
+    h.tevilah()
+    assert h.is_active() is False
+
+
+def test_tevilah_preserves_immutable_identifier():
+    h = IshItti("aharon")
+    h.activate()
+    h.tevilah()
+    assert h.identifier == "aharon"
+
+
+# ============================================================
+# THE LOAD-BEARING CI INVARIANT
+# ============================================================
+
+
+def test_tevilah_clears_all_non_immutable_state():
+    """Property 4's structural defense: ANY attribute on IshItti
+    that does not start with `_immutable_` MUST be reset to a fresh
+    value (or removed entirely) by tevilah().
+
+    A future contributor who adds `self._cache = {}` (or any other
+    state attribute) to IshItti without extending tevilah() to
+    clear it breaks the property — and this test fires before the
+    regression ships.
+
+    The discovery is dynamic: we walk the handler's `vars()` after
+    pre-tevilah state has been seeded, then verify each non-immutable
+    attribute is in a 'fresh' state post-tevilah.
+    """
+    h = IshItti("aharon")
+    h.activate()
+
+    # Seed every state-bearing attribute we know about.
+    h.hold_payload({"data": "first"})
+    h.hold_payload({"data": "second"})
+    h.tag("checkpoint", "STAGED")
+    h.tag("disposal_id", "abc-123")
+
+    # Snapshot the pre-tevilah attribute set.
+    pre_tevilah = dict(vars(h))
+    # Sanity: at least one non-immutable attribute should be non-fresh.
+    non_immutable = {
+        k: v for k, v in pre_tevilah.items() if not k.startswith("_immutable_")
+    }
+    assert any(_is_non_fresh(v) for v in non_immutable.values()), (
+        "Test setup error: no non-fresh state to clear"
+    )
+
+    h.tevilah()
+
+    # Post-tevilah: every non-immutable attribute must be 'fresh'.
+    for attr_name, value in vars(h).items():
+        if attr_name.startswith("_immutable_"):
+            continue
+        assert _is_fresh(value), (
+            f"tevilah() did not reset {attr_name}: value is {value!r}. "
+            f"If this attribute was added without extending tevilah, "
+            f"that is the property-4 regression. Either prefix the "
+            f"name with `_immutable_` (if it is set once at "
+            f"construction) or extend tevilah() to clear it."
+        )
+
+    # All immutable attributes must STILL be present.
+    for attr_name in pre_tevilah:
+        if attr_name.startswith("_immutable_"):
+            assert attr_name in vars(h), (
+                f"tevilah() incorrectly removed immutable attribute {attr_name!r}."
+            )
+
+
+def _is_fresh(value) -> bool:
+    """A value is 'fresh' if it is the canonical empty value for its
+    type — empty container, False, None, 0."""
+    if value is None or value is False:
+        return True
+    if value == 0:
+        return True
+    if isinstance(value, (list, tuple, set, dict, str, bytes)) and len(value) == 0:
+        return True
+    return False
+
+
+def _is_non_fresh(value) -> bool:
+    return not _is_fresh(value)
+
+
+# ============================================================
+# assert_shaliach_active — the role gate
+# ============================================================
+
+
+def test_assert_shaliach_active_passes_for_active_ish_itti():
+    h = IshItti("aharon")
+    h.activate()
+    # Should not raise
+    assert_shaliach_active(h)
+
+
+def test_assert_shaliach_active_refuses_inactive_handler():
+    h = IshItti("aharon")
+    with pytest.raises(ShaliachError) as exc:
+        assert_shaliach_active(h)
+    assert "not active" in str(exc.value).lower()
+
+
+def test_assert_shaliach_active_refuses_non_ish_itti():
+    """A kohen-shaped or owner-shaped value passed where a shaliach
+    is expected — refuse. Property 4: cross-role invocation is
+    forbidden."""
+    not_a_shaliach = "string-not-an-IshItti"
+    with pytest.raises(ShaliachError) as exc:
+        assert_shaliach_active(not_a_shaliach)  # type: ignore[arg-type]
+    assert "ishitti" in str(exc.value).lower() or "shaliach" in str(exc.value).lower()
+
+
+def test_assert_shaliach_active_refuses_object_impersonating_handler():
+    """A duck-typed object with the same method names is NOT an
+    IshItti. The role gate is identity, not interface."""
+
+    class DuckIshItti:
+        identifier = "fake"
+
+        def is_active(self):
+            return True
+
+    with pytest.raises(ShaliachError):
+        assert_shaliach_active(DuckIshItti())  # type: ignore[arg-type]
+
+
+# ============================================================
+# Stale-state-across-disposals detection
+# ============================================================
+
+
+def test_stale_state_across_disposals_is_prevented():
+    """The integration property: state from disposal-N must not
+    leak into disposal-N+1. tevilah is the gate; this test verifies
+    it works end-to-end."""
+    h = IshItti("aharon")
+
+    # Disposal 1
+    h.activate()
+    h.hold_payload({"id": 1, "data": "first"})
+    h.tag("disposal_id", "first")
+    h.tevilah()
+
+    # Disposal 2 — must start fresh
+    h.activate()
+    assert h.held_payload_count() == 0
+    assert h.get_tag("disposal_id") is None
+    h.hold_payload({"id": 2, "data": "second"})
+    h.tag("disposal_id", "second")
+    assert h.held_payload_count() == 1
+    assert h.get_tag("disposal_id") == "second"