azazel commit 10 — closure pass: cross-channel classification-discipline integration test

Pass 4 / Finding F, commit 10 of 10 from BUILD_AZAZEL.md. Closes the
build with the integration test the staging plan called for: a
pipeline run that exercises BOTH the Korban path (processable input)
AND the Azazel path (empty-vital input) confirms classification
discipline (no input crosses channels).

Property 11 was already verified at the unit layer (the cross-package
test in azazel_processor/test/test_stages.py). This commit adds the
integration-layer verification: at the external_offering call site,
the two channels produce structurally distinct outputs that cannot
be confused with each other.

CREATED:

  korban_processor/test/test_classification_discipline_integration.py
                                                              4 tests
    - test_processable_path_produces_korban_not_azazel
        Vital-bearing response → ProcessedKorban with vital,
        disposal_record is None, ProcessedKorban is NOT an
        AzazelRecord
    - test_empty_vital_path_produces_azazel_not_processable_korban
        Empty-vital response → ProcessedKorban with empty
        vital_content, disposal_record is an AzazelRecord
    - test_two_runs_one_each_path_keep_outputs_separate
        Run both back-to-back with shared pending_l3 dir; verify
        outputs don't cross channels; only the empty-vital run
        drops a file
    - test_disposal_record_contains_no_processable_artifacts
        AzazelRecord has no vital_content, propositions, raw_korban,
        or discarded attributes; routing is the ERETZ_GEZERAH
        sentinel

Tests: korban_processor 211 passed, 1 skipped (was 207/1; +4
integration tests). All other test counts unchanged.

DOCUMENTATION (companion to this commit, not in the repo):

  ~/shorashim-conversation-review.md
    + Pass 4 section (300+ lines): finding A closure (across-verse
      voice stabilization detector), BUILD_AZAZEL design doc,
      9-commit Azazel build with property-by-property closure, full
      test counts (~907 across five languages). All 14 findings
      from the original A–N catalog now in final state.

  ~/.claude/projects/-Users-scottkarp/memory/MEMORY.md
    + Updated test counts (per layer)
    + Added Azazel Processor section under Phase 5 progress
    + Noted /peah/azazel_log endpoint as Phase 5 done

FINAL STATE OF THE AZAZEL BUILD:

  All 12 properties from azazel_engineering_reading.md §14 defended
  in code:
    1  parity provisioning      — import-time + runtime
    2  blind routing / goral    — deterministic per (id, day)
    3  enumerative commit        — operator-produced, three-fold
    4  role-separated operator   — IshItti + tevilah + CI invariant
    5  provisioned infrastructure — six checkpoints, persistence
    6  out-of-band acknowledgment — Chatzotzrot via Elixir bridge
    7  authorized routing        — kohen_gadol + shaliach attribution
    8  protocol-restricted access — envelope gate, fires first
    9  observable disposal        — L3 + /peah/azazel_log
    10 hard termination           — DESTROYED checkpoint, no recover
    11 classification discipline  — type system + integration test
    12 unmanaged destination      — sentinel + DB CHECK + CI invariant

  10 commits, ~3,300 lines added, ~150 tests across Python and
  Elixir. The Azazel Processor is the disposal twin of the Korban
  Processor; the two-goat architecture (Vayikra 16:8) is now built.

The Pass 2 backlog is fully closed. Every actionable finding from
the original A–N catalog is now in a final state.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Scott KarpandClaude Opus 4.7 committed 2026-04-28 10:29:24 -04:00
1 parent 1a27a283ae
commit 1e59d9a752
1 file changed
+154
@@ -0,0 +1,154 @@
"""
End-to-end classification-discipline test (BUILD_AZAZEL.md commit 10).
Property 11 verified at the integration layer: a pipeline run that
exercises BOTH the Korban path (processable input) AND the Azazel
path (empty-vital input) confirms no input crossed channels and the
two channel outputs cannot be confused with each other.
The unit-level test in azazel_processor/test/test_stages.py
(test_korban_function_refuses_azazel_staged_value etc.) verifies
the type system gate. THIS test verifies the integration: at the
external_offering level, a single pipeline run produces either a
ProcessedKorban OR an AzazelRecord (never both with non-empty
content), and the two outputs are non-interconvertible.
"""
import pathlib
import tempfile
from azazel_processor import AzazelRecord
from korban_processor.external_offering import (
ExternalConsultation,
consult_external_model,
)
from korban_processor.korban import ProcessedKorban
from korban_processor.model_interface import ModelRequest, ModelResponse
def _fake_caller(text):
def caller(_request: ModelRequest) -> ModelResponse:
return ModelResponse(
content=text,
model="mock",
stop_reason="end_turn",
input_tokens=10,
output_tokens=20,
)
return caller
def test_processable_path_produces_korban_not_azazel():
"""A response with extractable Torah goes through korban; the
AzazelRecord field stays None."""
with tempfile.TemporaryDirectory() as d:
result = consult_external_model(
query="any",
source_query="processable_001",
caller=_fake_caller(
"Bereishit 1:1 — the shoresh ב-ר-א means to create. "
"The verse indicates beginnings."
),
azazel_pending_l3_dir=pathlib.Path(d),
)
assert isinstance(result.processed, ProcessedKorban)
assert result.processed.vital_content != ""
assert result.disposal_record is None
# And critically: a ProcessedKorban is NOT an AzazelRecord
assert not isinstance(result.processed, AzazelRecord)
def test_empty_vital_path_produces_azazel_not_processable_korban():
"""An empty-vital response goes through azazel; processed is
still a ProcessedKorban (empty), but disposal_record is the
actual record of what happened."""
with tempfile.TemporaryDirectory() as d:
result = consult_external_model(
query="any",
source_query="empty_vital_001",
caller=_fake_caller("ok thanks. let me know."),
azazel_pending_l3_dir=pathlib.Path(d),
)
assert isinstance(result.processed, ProcessedKorban)
assert result.processed.vital_content == "" # nothing processable
assert isinstance(result.disposal_record, AzazelRecord)
def test_two_runs_one_each_path_keep_outputs_separate():
"""Run two consultations back-to-back — one processable, one
empty-vital. Verify the outputs are completely separate; nothing
from the korban side crosses into the azazel side or vice versa."""
with tempfile.TemporaryDirectory() as d:
pending = pathlib.Path(d)
processable = consult_external_model(
query="any",
source_query="separate_processable",
caller=_fake_caller(
"Vayikra 11:44 — the shoresh ק-ד-ש means holy. The verse "
"indicates separation."
),
azazel_pending_l3_dir=pending,
)
empty_vital = consult_external_model(
query="any",
source_query="separate_empty_vital",
caller=_fake_caller("ok"),
azazel_pending_l3_dir=pending,
)
# Processable path: ProcessedKorban with vital, no disposal record
assert processable.processed.vital_content != ""
assert processable.disposal_record is None
# Empty-vital path: ProcessedKorban without vital, has disposal record
assert empty_vital.processed.vital_content == ""
assert empty_vital.disposal_record is not None
# Cross-channel non-interconvertibility: the AzazelRecord
# from the empty-vital run is not a ProcessedKorban; the
# ProcessedKorban from the processable run is not an
# AzazelRecord.
assert not isinstance(processable.processed, AzazelRecord)
assert not isinstance(empty_vital.disposal_record, ProcessedKorban)
# The case_refs are content-addressable and different
assert empty_vital.disposal_record.case_ref != ""
# Only ONE pending-l3 file (from the empty-vital run).
# Assertion MUST be inside the with-block — pending is
# deleted on TemporaryDirectory exit.
files = list(pending.glob("*.json"))
assert len(files) == 1
def test_disposal_record_contains_no_processable_artifacts():
"""The empty-vital run's disposal_record must not carry anything
from the (non-existent) korban-side processing — no propositions,
no vital content, no routing metadata that suggests it went
through the korban path."""
with tempfile.TemporaryDirectory() as d:
result = consult_external_model(
query="any",
source_query="disposal_isolation_001",
caller=_fake_caller("ok"),
azazel_pending_l3_dir=pathlib.Path(d),
)
record = result.disposal_record
# No processable-shape attributes
for attr in ["vital_content", "propositions", "raw_korban", "discarded"]:
assert not hasattr(record, attr), (
f"AzazelRecord has korban-shaped attribute {attr!r} — "
f"property 11 (classification discipline) violated"
)
# The routing is the disposal-only sentinel
from azazel_processor import is_eretz_gezerah
assert is_eretz_gezerah(record.routing)