diff --git a/korban_processor/src/korban_processor/external_offering.py b/korban_processor/src/korban_processor/external_offering.py index 2e4f2bb9..37295e4a 100644 --- a/korban_processor/src/korban_processor/external_offering.py +++ b/korban_processor/src/korban_processor/external_offering.py @@ -22,7 +22,11 @@ This module is the only entry point for external model consultation. No internal component calls an external model directly. """ -from dataclasses import dataclass +import hashlib +import json +import os +import pathlib +from dataclasses import dataclass, field from typing import Optional from korban_processor.korban import ( @@ -43,21 +47,97 @@ from korban_processor.haktarah import haktarah_staged from korban_processor.stages import KorbanStage from korban_processor.router import RoutingResult, route +# Property 1 (parity provisioning): the disposal channel must be +# importable at the same scale as the processing channel. If +# azazel_processor cannot be imported, external_offering refuses to +# load — the two-goat architecture is symmetric or it is broken. +# This is the strongest formulation of "provisioned in advance, from +# the same pool"; the failure manifests as ImportError at module load, +# not as a runtime gap discovered when an empty-vital korban arrives. +try: + from azazel_processor import ( + AzazelInput, + AzazelRecord, + IshItti, + authorize_disposal, + confess_empty_vital, + dispatch_for_disposal, + ) +except ImportError as e: # pragma: no cover — bootstrap-time check + raise ImportError( + "external_offering requires azazel_processor (the disposal " + "channel) to be importable. Property 1 of " + "azazel_engineering_reading.md §14 — disposal capacity must " + "be provisioned at parity with processing capacity, from the " + "same pool. Without azazel_processor on the import path, the " + "two-goat architecture is broken; refuse to load. " + f"Underlying ImportError: {e}" + ) + + +# Default directory the Python orchestrator drops completed +# AzazelRecords into for the Elixir AzazelIngest GenServer to pick +# up. Mirrors the manna-provision file pattern Shorashim already uses +# for cross-language state transfer. +DEFAULT_AZAZEL_PENDING_L3_DIR = pathlib.Path("/tmp/shorashim/azazel/pending_l3") + @dataclass class ExternalConsultation: """The full record of an external-model consultation. - All four fields are preserved so the full chain is auditable: - - request: what was asked and with what system prompt - - raw_response: the unprocessed output from the external model - - processed: the korban after the offering protocol - - routing: where the processed korban was sent + Fields are preserved so the full chain is auditable: + - request: what was asked and with what system prompt + - raw_response: the unprocessed output from the external model + - processed: the korban after the offering protocol + - routing: where the processed korban was sent + - disposal_record: present only if the consultation was routed + to disposal (zerikah produced no vital + content). When non-None, the caller knows + the input was DISPOSED rather than + processed; processed.vital_content will be + empty in this case. """ request: ModelRequest raw_response: ModelResponse processed: ProcessedKorban routing: RoutingResult + disposal_record: Optional[AzazelRecord] = None + + +def _content_addressable_case_ref(content: str, source_query: str) -> str: + """Produce a case_ref from the input content. Stable hash so audit + can verify the routing was for THIS input (without retaining the + input itself — property 10).""" + h = hashlib.sha256() + h.update(source_query.encode("utf-8")) + h.update(b"|") + h.update(content.encode("utf-8")) + return f"ext-consult-{h.hexdigest()[:16]}" + + +def _make_pending_l3_recorder(pending_dir: pathlib.Path): + """Return a recorder callable that writes the AzazelRecord to a + JSON file in the pending-l3 directory. The Elixir AzazelIngest + GenServer (commit 7) watches this directory and persists each + file to L3. + + Cross-language bridge: same pattern as the manna provision file. + Recoverable: if either side crashes mid-write, the file remains + and is re-processed on the next poll. + """ + pending_dir.mkdir(parents=True, exist_ok=True) + + def recorder(record: AzazelRecord) -> None: + path = pending_dir / f"{record.case_ref}.json" + # Atomic-ish: write to temp then rename, so a crash mid-write + # leaves either the previous state or nothing — never a + # half-written file. + temp_path = path.with_suffix(".json.tmp") + temp_path.write_text(json.dumps(record.to_storage_dict())) + os.replace(temp_path, path) + + return recorder def consult_external_model( @@ -67,6 +147,10 @@ def consult_external_model( offering_type: OfferingType = OfferingType.OLAH, system_prompt: Optional[str] = None, caller: Optional[callable] = None, + *, + azazel_kohen_gadol: str = "kohen-gadol-default", + azazel_shaliach_id: str = "ish-itti-default", + azazel_pending_l3_dir: pathlib.Path = DEFAULT_AZAZEL_PENDING_L3_DIR, ) -> ExternalConsultation: """Submit a query to an external model, then process the response as korban. @@ -147,9 +231,99 @@ def consult_external_model( routing = route(processed, offering_type) + # Property 1 (parity provisioning) END-TO-END: when zerikah + # produces no vital content, the korban is routed to disposal + # rather than silently dropped. The two-goat architecture is + # provisioned at intake (property 1 — both modules importable) + # AND at runtime (this branch — disposal is invoked when the + # processing channel cannot metabolize the input). + # + # Before this commit, an empty-vital korban was returned as-is + # with empty vital_content and the discarded sentences in + # `discarded` — the caller knew the offering was empty but the + # rejection had no permanent record. After: the empty-vital + # case is named (vidui), routed (eretz_gezerah), recorded + # (L3 azazel_records via the file-watcher bridge), and + # acknowledged (chatzotzrot :moadim by AzazelIngest). + disposal_record = None + if not vital.strip(): + disposal_record = _route_to_azazel( + korban=korban, + source_query=source_query, + kohen_gadol=azazel_kohen_gadol, + shaliach_id=azazel_shaliach_id, + pending_l3_dir=azazel_pending_l3_dir, + ) + return ExternalConsultation( request=request, raw_response=raw_response, processed=processed, routing=routing, + disposal_record=disposal_record, ) + + +def _route_to_azazel( + *, + korban: Korban, + source_query: str, + kohen_gadol: str, + shaliach_id: str, + pending_l3_dir: pathlib.Path, +) -> AzazelRecord: + """Run the full Azazel disposal protocol for an empty-vital korban. + + Property 1 — parity provisioning at runtime: an empty-vital + korban is not silently discarded; it goes to disposal with the + full protocol envelope. + + Each call activates a fresh shaliach, performs the disposal, + then tevilah's the shaliach. The pending-l3 file the recorder + drops is picked up by the Elixir AzazelIngest GenServer (commit + 7) which writes the L3 record AND fires the Chatzotzrot + :moadim signal (property 6 — out-of-band acknowledgment). + """ + shaliach = IshItti(shaliach_id) + shaliach.activate() + + # Build the vidui — operator-produced enumerative commit. We + # name the case at the call site here; the helper routes it + # into chatot (target-missing). + summary = ( + f"{korban.provenance.value} response on " + f"{source_query!r} — zerikah extracted no vital content" + ) + vidui = confess_empty_vital(summary, confessor=kohen_gadol) + + case_ref = _content_addressable_case_ref( + korban.raw_content, source_query + ) + + azazel_input = AzazelInput( + case_ref=case_ref, + payload=korban, + classification="empty_vital", + triggering_call="external_offering.consult_external_model", + ) + + recorder = _make_pending_l3_recorder(pending_l3_dir) + + # Property 8 envelope: this is an event-driven disposal during + # the year (not a Yom Kippur batch), so we open authorize_disposal + # explicitly. The scope is exactly this one disposal cycle. + with authorize_disposal(): + record = dispatch_for_disposal( + azazel_input, + handler=shaliach, + confess_with=vidui, + authorized_by=kohen_gadol, + recorder=recorder, + ) + + # Tevilah — the shaliach washes before rejoining the camp + # (Vayikra 16:26). Property 4 follow-through: state must be + # cleared before the next disposal can use this handler. + shaliach.tevilah() + + return record diff --git a/korban_processor/test/test_external_offering.py b/korban_processor/test/test_external_offering.py index b33a98b4..dca6c0fe 100644 --- a/korban_processor/test/test_external_offering.py +++ b/korban_processor/test/test_external_offering.py @@ -160,6 +160,137 @@ def test_system_prompt_is_forwarded(): assert result.request.system_prompt == "You are a Torah scholar. Cite verse references." +# ============================================================ +# Empty-vital → Azazel disposal routing (commit 9, properties 1 + 6) +# ============================================================ +# Property 1 at runtime: an empty-vital korban is routed to disposal, +# not silently dropped. The disposal record IS the permanent +# acknowledgment that the system received this input and could not +# metabolize it. + +import json +import pathlib +import tempfile + +from azazel_processor import AzazelRecord + + +def _empty_vital_response(): + """A model response that contains no extractable Torah content + (no Torah refs, no Hebrew, no insight markers, no roots). + zerikah will return empty vital for this.""" + return "ok thanks. let me know if you want me to do anything else." + + +def test_empty_vital_korban_is_routed_to_azazel(): + with tempfile.TemporaryDirectory(prefix="azazel-pending-") as d: + pending = pathlib.Path(d) + + result = consult_external_model( + query="anything", + source_query="empty_vital_test_001", + caller=_fake_caller(_empty_vital_response()), + azazel_kohen_gadol="kohen-gadol-test", + azazel_shaliach_id="ish-itti-test", + azazel_pending_l3_dir=pending, + ) + + # The processed korban is empty (zerikah returned no vital) + assert result.processed.vital_content == "" + + # The disposal_record field is now populated + assert result.disposal_record is not None + assert isinstance(result.disposal_record, AzazelRecord) + assert result.disposal_record.shaliach == "ish-itti-test" + assert result.disposal_record.authorized_by == "kohen-gadol-test" + assert result.disposal_record.triggering_call == ( + "external_offering.consult_external_model" + ) + + # A pending-l3 file was dropped for AzazelIngest to pick up + files = list(pending.glob("*.json")) + assert len(files) == 1 + + body = json.loads(files[0].read_text()) + assert body["case_ref"] == result.disposal_record.case_ref + assert body["routing"] == "eretz_gezerah" + assert body["shaliach"] == "ish-itti-test" + # The vidui is on the pending file (operator-internal channel) + assert body["vidui"]["confessor"] == "kohen-gadol-test" + assert body["vidui"]["chatot"] # non-empty + + +def test_processable_response_does_not_create_disposal_record(): + """The companion to the above: an offering with extractable Torah + content goes through the processing channel and the disposal_record + field stays None.""" + with tempfile.TemporaryDirectory(prefix="azazel-pending-") as d: + result = consult_external_model( + query="explain bereishit 1:1", + source_query="processable_test_001", + caller=_fake_caller( + "Bereishit 1:1 — the shoresh ב-ר-א means to create. The verse " + "indicates beginnings." + ), + azazel_pending_l3_dir=pathlib.Path(d), + ) + + assert result.disposal_record is None + assert result.processed.vital_content != "" + # No file was dropped — nothing for AzazelIngest to pick up + assert list(pathlib.Path(d).glob("*.json")) == [] + + +def test_pending_l3_file_does_not_contain_payload(): + """Property 10 follow-through: the file the orchestrator drops + for the Elixir bridge contains the vidui (what was disposed) but + not the disposed thing itself. The model's actual response text + must not appear in the pending-l3 file.""" + secret_phrase = "this-string-must-not-leak-to-the-pending-l3-file" + + with tempfile.TemporaryDirectory(prefix="azazel-pending-") as d: + pending = pathlib.Path(d) + consult_external_model( + query="any", + source_query="leak_check_001", + caller=_fake_caller(secret_phrase), # response is just the secret + azazel_pending_l3_dir=pending, + ) + + files = list(pending.glob("*.json")) + assert len(files) == 1 + body_text = files[0].read_text() + assert secret_phrase not in body_text + + +def test_case_ref_is_content_addressable(): + """Audit can verify the routing was for a specific input by + recomputing the case_ref. Property 7's audit-trail half — the + routing decision is owned and traceable.""" + response = _empty_vital_response() + + with tempfile.TemporaryDirectory(prefix="azazel-pending-") as d: + pending = pathlib.Path(d) + result1 = consult_external_model( + query="any", + source_query="audit_test_001", + caller=_fake_caller(response), + azazel_pending_l3_dir=pending, + ) + + # Same content + same source_query → same case_ref + with tempfile.TemporaryDirectory(prefix="azazel-pending-") as d: + pending = pathlib.Path(d) + result2 = consult_external_model( + query="different-query", # query is not in the case_ref + source_query="audit_test_001", + caller=_fake_caller(response), + azazel_pending_l3_dir=pending, + ) + + assert result1.disposal_record.case_ref == result2.disposal_record.case_ref + + # ============================================================ # Live API integration test — opt-in only # ============================================================ diff --git a/vav_pipeline/lib/vav_pipeline/azazel_ingest.ex b/vav_pipeline/lib/vav_pipeline/azazel_ingest.ex index 32957dff..2ac5f7ed 100644 --- a/vav_pipeline/lib/vav_pipeline/azazel_ingest.ex +++ b/vav_pipeline/lib/vav_pipeline/azazel_ingest.ex @@ -45,7 +45,7 @@ defmodule VavPipeline.AzazelIngest do require Logger - alias VavPipeline.L3.Store + alias VavPipeline.{Chatzotzrot, L3.Store} @default_poll_interval_ms 5_000 @default_pending_dir "/tmp/shorashim/azazel/pending_l3" @@ -163,6 +163,20 @@ defmodule VavPipeline.AzazelIngest do {:ok, raw} <- Jason.decode(body), {:ok, attrs} <- normalize_record(raw), {:ok, _record} <- recorder.(attrs) do + # Property 6 (out-of-band acknowledgment) — fire the Chatzotzrot + # signal AFTER the L3 write succeeds. The chatzotzrot is the + # side channel; the disposal flow itself does not return + # acknowledgment to the caller (the caller already knows they + # handed the input to disposal). + # + # :moadim is the right signal type — Yom Kippur completion + # semantics. This disposal cycle is closed; the public fact + # (property 9) is now witnessable via /peah/azazel_log. + # + # Process.whereis-guarded so this works in tests that don't + # bring up the full supervision tree. + fire_chatzotzrot(attrs) + # Property 10 follow-through: shred the source file after # successful L3 write. The disposal is now permanent oral # tradition; the pending artifact must not linger. @@ -179,6 +193,34 @@ defmodule VavPipeline.AzazelIngest do end end + defp fire_chatzotzrot(attrs) do + if Process.whereis(Chatzotzrot) do + summary = chatzotzrot_summary(attrs) + + Chatzotzrot.signal(:moadim, + reason: "Azazel disposal completed: #{summary}" + ) + end + + :ok + end + + defp chatzotzrot_summary(attrs) do + case_ref = Map.get(attrs, :case_ref) || "unknown" + shaliach = Map.get(attrs, :shaliach) || "unknown" + vidui = Map.get(attrs, :vidui) || %{} + + categories = + Enum.filter(["avonot", "peshaim", "chatot"], fn slot -> + case Map.get(vidui, slot) do + list when is_list(list) and list != [] -> true + _ -> false + end + end) + + "case=#{case_ref} shaliach=#{shaliach} categories=#{Enum.join(categories, ",")}" + end + # Translate the Python-side JSON shape into Ecto-friendly attrs. # The Python AzazelRecord.to_storage_dict() shape is: # { diff --git a/vav_pipeline/test/azazel_ingest_test.exs b/vav_pipeline/test/azazel_ingest_test.exs index ec6a8327..28b5c155 100644 --- a/vav_pipeline/test/azazel_ingest_test.exs +++ b/vav_pipeline/test/azazel_ingest_test.exs @@ -226,6 +226,83 @@ defmodule VavPipeline.AzazelIngestTest do assert length(:ets.tab2list(captured)) == 1 end + # ============================================================ + # Property 6 — out-of-band acknowledgment via Chatzotzrot + # ============================================================ + + test "successful ingest fires Chatzotzrot :moadim signal", %{pending_dir: dir} do + # The supervision-tree-started Chatzotzrot is already running. + # Subscribe this test process as an :operators recipient and + # observe the signal AzazelIngest fires. + VavPipeline.Chatzotzrot.subscribe(:all) + + drop_file(dir, "case-ack-001.json", valid_record_json("case-ack-001")) + %{server: server} = start_ingest(dir) + AzazelIngest.poll_now(server) + + # Drain any unrelated signals from prior tests; we want OUR signal. + receive_until_ours("case-ack-001") + end + + test "chatzotzrot fire does NOT include vidui item content (privacy)" do + # Property 9's privacy boundary applies to the chatzotzrot + # summary too — the operator-side signal channel must not leak + # the operator-internal vidui items. + secret_phrase = "must-not-leak-via-chatzotzrot-summary-channel" + + body = + %{ + "case_ref" => "leak-001", + "vidui" => %{ + "avonot" => [], + "peshaim" => [], + "chatot" => [secret_phrase], + "confessor" => "kohen", + "confessed_at" => "2026-04-28T12:00:00Z" + }, + "routing" => "eretz_gezerah", + "authorized_by" => "kohen-gadol", + "shaliach" => "aharon", + "triggering_call" => "test", + "goral_outcome_id" => nil, + "disposed_at" => "2026-04-28T12:00:01Z" + } + |> Jason.encode!() + + VavPipeline.Chatzotzrot.subscribe(:all) + + dir = + System.tmp_dir!() |> Path.join("azazel-leak-#{:erlang.unique_integer([:positive])}") + + File.mkdir_p!(dir) + on_exit(fn -> File.rm_rf!(dir) end) + File.write!(Path.join(dir, "leak-001.json"), body) + + %{server: server} = start_ingest(dir) + AzazelIngest.poll_now(server) + + signal = receive_until_ours("leak-001") + refute signal.details.reason =~ secret_phrase + end + + # Drain any unrelated chatzotzrot signals (from other tests sharing + # the supervision-tree Chatzotzrot) until we see ours, identified by + # case_ref in the reason. Returns the matching signal. + defp receive_until_ours(case_ref) do + receive do + {:chatzotzrot, signal} -> + if String.contains?(signal.details.reason || "", case_ref) and + signal.signal_type == :moadim do + assert signal.details.reason =~ "Azazel disposal completed" + signal + else + receive_until_ours(case_ref) + end + after + 2_000 -> flunk("Did not receive chatzotzrot signal for #{case_ref}") + end + end + # ============================================================ # Changeset validation (no DB) # ============================================================