diff --git a/vav_pipeline/lib/vav_pipeline/azazel_ingest.ex b/vav_pipeline/lib/vav_pipeline/azazel_ingest.ex new file mode 100644 index 00000000..32957dff --- /dev/null +++ b/vav_pipeline/lib/vav_pipeline/azazel_ingest.ex @@ -0,0 +1,222 @@ +defmodule VavPipeline.AzazelIngest do + @moduledoc """ + Cross-language bridge between the Python Azazel Processor and the + Elixir L3 layer. + + The Python orchestrator (azazel_processor/src/azazel_processor/ + azazel.py) serializes completed AzazelRecords to JSON files in a + pending-l3 directory. This GenServer watches that directory, + reads each file, calls L3.Store.record_disposal/1, and on success + shreds the source file. + + The pattern matches Shorashim's other cross-language file bridges + (the manna provision file). Recoverable: if either the Python side + or this GenServer crashes mid-write, the file remains and is + re-processed on the next poll. + + POLL-BASED, NOT FileSystem.event-BASED: + + We do not bring the :file_system dependency into vav_pipeline. + Polling is simpler and adequate for the disposal cadence (the + Azazel channel is for exceptional content, not a high-throughput + stream). Default poll interval 5 seconds; configurable. + + RETRY SEMANTICS: + + If L3.Store.record_disposal/1 fails (validation error, DB + unavailable), the file is LEFT IN PLACE — it will be retried on + the next poll. The reasoning: L3 is append-only; a write that + failed for a transient reason should be retried, and a write + that failed for a permanent reason (schema mismatch) should + remain visible to operators rather than be silently destroyed. + + Operator intervention path: if a file persists across many polls + with the same error, the operator inspects it directly in the + pending directory and decides whether to fix the schema, fix the + record, or move it to an error-archive subdirectory for analysis. + + PROPERTY COVERAGE: + + Property 9 (operator-internal half) — the L3 record exists. The + public-side endpoint (commit 8) reads from the same table. + """ + + use GenServer + + require Logger + + alias VavPipeline.L3.Store + + @default_poll_interval_ms 5_000 + @default_pending_dir "/tmp/shorashim/azazel/pending_l3" + + # ============================================================ + # Client API + # ============================================================ + + @doc """ + Start the AzazelIngest GenServer. Options: + - :pending_dir — directory to watch (default + /tmp/shorashim/azazel/pending_l3) + - :poll_interval_ms — milliseconds between polls (default 5000) + - :name — process name (default __MODULE__) + - :recorder — function used to write the L3 record; + defaults to &Store.record_disposal/1. + Tests inject mocks here so they don't + need PostgreSQL. + """ + def start_link(opts \\ []) do + name = Keyword.get(opts, :name, __MODULE__) + GenServer.start_link(__MODULE__, opts, name: name) + end + + @doc """ + Run a single poll synchronously. Useful for tests and for + operator-triggered manual flushes. + """ + def poll_now(server \\ __MODULE__) do + GenServer.call(server, :poll_now) + end + + @doc "Return the GenServer's current configuration and state." + def state(server \\ __MODULE__) do + GenServer.call(server, :state) + end + + # ============================================================ + # GenServer + # ============================================================ + + @impl true + def init(opts) do + state = %{ + pending_dir: Keyword.get(opts, :pending_dir, @default_pending_dir), + poll_interval_ms: Keyword.get(opts, :poll_interval_ms, @default_poll_interval_ms), + recorder: Keyword.get(opts, :recorder, &Store.record_disposal/1), + polls_completed: 0, + records_ingested: 0, + errors: 0 + } + + File.mkdir_p!(state.pending_dir) + schedule_poll(state.poll_interval_ms) + {:ok, state} + end + + @impl true + def handle_call(:poll_now, _from, state) do + new_state = do_poll(state) + {:reply, :ok, new_state} + end + + def handle_call(:state, _from, state), do: {:reply, state, state} + + @impl true + def handle_info(:poll, state) do + new_state = do_poll(state) + schedule_poll(new_state.poll_interval_ms) + {:noreply, new_state} + end + + # ============================================================ + # Internal + # ============================================================ + + defp schedule_poll(interval_ms) do + Process.send_after(self(), :poll, interval_ms) + end + + defp do_poll(state) do + files = list_pending(state.pending_dir) + + {ingested, errors} = + Enum.reduce(files, {0, 0}, fn path, {i, e} -> + case ingest_file(path, state.recorder) do + :ok -> {i + 1, e} + {:error, _reason} -> {i, e + 1} + end + end) + + %{ + state + | polls_completed: state.polls_completed + 1, + records_ingested: state.records_ingested + ingested, + errors: state.errors + errors + } + end + + defp list_pending(dir) do + case File.ls(dir) do + {:ok, names} -> + names + |> Enum.filter(&String.ends_with?(&1, ".json")) + |> Enum.sort() + |> Enum.map(&Path.join(dir, &1)) + + {:error, _reason} -> + [] + end + end + + defp ingest_file(path, recorder) do + with {:ok, body} <- File.read(path), + {:ok, raw} <- Jason.decode(body), + {:ok, attrs} <- normalize_record(raw), + {:ok, _record} <- recorder.(attrs) do + # 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. + File.rm(path) + :ok + else + {:error, reason} -> + Logger.warning( + "AzazelIngest: failed to ingest #{path} — #{inspect(reason)}. " <> + "File left in place for retry." + ) + + {:error, reason} + end + end + + # Translate the Python-side JSON shape into Ecto-friendly attrs. + # The Python AzazelRecord.to_storage_dict() shape is: + # { + # "case_ref": str, + # "vidui": {avonot, peshaim, chatot, confessor, confessed_at}, + # "routing": "eretz_gezerah", + # "authorized_by": str, + # "shaliach": str, + # "triggering_call": str | null, + # "goral_outcome_id": str | null, + # "disposed_at": iso8601 string + # } + # We pass `vidui` through as-is (JSONB column) and parse the + # disposed_at into a UTC DateTime. + defp normalize_record(raw) when is_map(raw) do + with {:ok, disposed_at} <- parse_iso8601(Map.get(raw, "disposed_at")) do + {:ok, + %{ + case_ref: Map.get(raw, "case_ref"), + vidui: Map.get(raw, "vidui"), + routing: Map.get(raw, "routing"), + authorized_by: Map.get(raw, "authorized_by"), + shaliach: Map.get(raw, "shaliach"), + triggering_call: Map.get(raw, "triggering_call"), + goral_outcome_id: Map.get(raw, "goral_outcome_id"), + disposed_at: disposed_at + }} + end + end + + defp normalize_record(_), do: {:error, :not_a_map} + + defp parse_iso8601(nil), do: {:error, :missing_disposed_at} + + defp parse_iso8601(s) when is_binary(s) do + case DateTime.from_iso8601(s) do + {:ok, dt, _offset} -> {:ok, DateTime.truncate(dt, :second)} + {:error, _} = err -> err + end + end +end diff --git a/vav_pipeline/lib/vav_pipeline/l3/azazel_record.ex b/vav_pipeline/lib/vav_pipeline/l3/azazel_record.ex new file mode 100644 index 00000000..b3961779 --- /dev/null +++ b/vav_pipeline/lib/vav_pipeline/l3/azazel_record.ex @@ -0,0 +1,125 @@ +defmodule VavPipeline.L3.AzazelRecord do + @moduledoc """ + L3 schema for Azazel disposal records — the gravestone for content + that went through the disposal channel. + + Vayikra 16:8, 16:22 — the goat goes to eretz gezerah and does not + return; the RECORD of the disposal remains as permanent oral + tradition. Append-only: once a disposal is recorded, it is fixed. + + THE PAYLOAD IS NOT HERE. Property 10 (hard termination) — the + AzazelRecord carries: + - case_ref opaque identifier of WHAT was disposed + - vidui operator-produced three-fold enumeration + - routing ALWAYS the literal "eretz_gezerah" + - authorized_by kohen_gadol identifier (property 7) + - shaliach ish itti identifier (property 4) + - triggering_call which subsystem invoked disposal (audit) + - goral_outcome_id if the goral was cast, its SHA-256 digest + (audit-recomputable per property 2) + - disposed_at orchestrator-clock timestamp + + The vidui is stored as JSONB with the shape: + { + "avonot": [string, ...], + "peshaim": [string, ...], + "chatot": [string, ...], + "confessor": string, + "confessed_at": iso8601 string + } + + Mirrors the Python AzazelRecord.to_storage_dict() (in + azazel_processor/src/azazel_processor/azazel_record.py). If either + side adds/removes a field, BOTH must move together — the cross- + language file watcher (VavPipeline.AzazelIngest) reads the Python + serialization and constructs an Ecto changeset from it. + """ + + use Ecto.Schema + import Ecto.Changeset + + schema "azazel_records" do + field :case_ref, :string + field :vidui, :map + field :routing, :string + field :authorized_by, :string + field :shaliach, :string + field :triggering_call, :string + field :goral_outcome_id, :string + field :disposed_at, :utc_datetime + + timestamps(type: :utc_datetime, updated_at: false) + end + + @required_fields [ + :case_ref, + :vidui, + :routing, + :authorized_by, + :shaliach, + :disposed_at + ] + + @optional_fields [:triggering_call, :goral_outcome_id] + + def changeset(record, attrs) do + record + |> cast(attrs, @required_fields ++ @optional_fields) + |> validate_required(@required_fields) + |> validate_inclusion(:routing, ["eretz_gezerah"], + message: + "routing must be the canonical opaque value 'eretz_gezerah'. " <> + "Property 12 (unmanaged destination) — there are no other " <> + "legitimate routing values." + ) + |> validate_vidui_shape() + end + + # The vidui must have all three categorical slots present (lists + # may be empty individually, but the absence of a slot is not OK). + # Plus a confessor and confessed_at. Mirrors the Python Vidui's + # __post_init__ validation. + defp validate_vidui_shape(changeset) do + case get_change(changeset, :vidui) do + nil -> + changeset + + vidui when is_map(vidui) -> + required_slots = ~w(avonot peshaim chatot confessor confessed_at) + missing = Enum.reject(required_slots, &Map.has_key?(vidui, &1)) + + cond do + missing != [] -> + add_error( + changeset, + :vidui, + "missing required slot(s): #{Enum.join(missing, ", ")}" + ) + + all_three_categories_empty?(vidui) -> + add_error( + changeset, + :vidui, + "all three categories empty — untagged disposal " <> + "(property 3 violation)" + ) + + true -> + changeset + end + + _ -> + add_error(changeset, :vidui, "must be a map") + end + end + + defp all_three_categories_empty?(vidui) do + Enum.all?(["avonot", "peshaim", "chatot"], fn slot -> + case Map.get(vidui, slot) do + [] -> true + nil -> true + _ -> false + end + end) + end +end diff --git a/vav_pipeline/lib/vav_pipeline/l3/store.ex b/vav_pipeline/lib/vav_pipeline/l3/store.ex index bddc19a2..6d716b79 100644 --- a/vav_pipeline/lib/vav_pipeline/l3/store.ex +++ b/vav_pipeline/lib/vav_pipeline/l3/store.ex @@ -18,7 +18,14 @@ defmodule VavPipeline.L3.Store do """ alias VavPipeline.Repo - alias VavPipeline.L3.{CouplingTrace, Interpretation, MinorityOpinion, TeikuRecord} + + alias VavPipeline.L3.{ + AzazelRecord, + CouplingTrace, + Interpretation, + MinorityOpinion, + TeikuRecord + } import Ecto.Query @@ -79,6 +86,30 @@ defmodule VavPipeline.L3.Store do |> Repo.insert() end + @doc """ + Record an Azazel disposal to L3. + + Accepts a map with the fields from the AzazelRecord schema — + case_ref, vidui (map with avonot/peshaim/chatot/confessor/ + confessed_at), routing (must be "eretz_gezerah"), authorized_by, + shaliach, optionally triggering_call and goral_outcome_id, and + disposed_at (UTC datetime). + + Returns {:ok, record} or {:error, changeset}. + + Pass 4 / Finding F, commit 7. The Python Azazel orchestrator + serializes AzazelRecord.to_storage_dict() to a JSON file under + the pending-l3 directory; VavPipeline.AzazelIngest reads each + file, calls this function, and deletes the source file on + success. The cross-language bridge for property 9 (operator- + internal observability). + """ + def record_disposal(attrs) when is_map(attrs) do + %AzazelRecord{} + |> AzazelRecord.changeset(attrs) + |> Repo.insert() + end + @doc """ Resolve a teiku — update resolved_at and resolution. @@ -149,4 +180,74 @@ defmodule VavPipeline.L3.Store do def count_minority_opinions do Repo.aggregate(MinorityOpinion, :count, :id) end + + @doc """ + Count Azazel disposals since a given timestamp (or all-time when nil). + Used by the public observability endpoint (commit 8) — counts only, + no payloads, per property 9's privacy boundary. + """ + def count_disposals(since \\ nil) + + def count_disposals(nil) do + Repo.aggregate(AzazelRecord, :count, :id) + end + + def count_disposals(%DateTime{} = since) do + AzazelRecord + |> where([r], r.disposed_at >= ^since) + |> Repo.aggregate(:count, :id) + end + + @doc """ + Categorical summary of Azazel disposals since a given timestamp. + + Returns a list of %{shaliach, count, vidui_categories_filled} maps. + vidui_categories_filled is which of avonot/peshaim/chatot had any + items, NOT the items themselves — property 9's public-side privacy + boundary. The detailed vidui is operator-internal (queryable via + /kodesh/* endpoints, commit 8). + """ + def disposal_summary(since \\ nil) do + query = + AzazelRecord + |> select([r], %{ + case_ref: r.case_ref, + shaliach: r.shaliach, + vidui: r.vidui, + triggering_call: r.triggering_call, + disposed_at: r.disposed_at + }) + |> order_by([r], desc: r.disposed_at) + + query = + case since do + nil -> query + %DateTime{} = ts -> where(query, [r], r.disposed_at >= ^ts) + end + + query + |> Repo.all() + |> Enum.map(fn row -> + %{ + case_ref: row.case_ref, + shaliach: row.shaliach, + triggering_call: row.triggering_call, + disposed_at: row.disposed_at, + # Categorical only — what slots were filled, NOT the items + categories_filled: vidui_categories_filled(row.vidui) + } + end) + end + + defp vidui_categories_filled(vidui) when is_map(vidui) do + Enum.reduce(["avonot", "peshaim", "chatot"], [], fn slot, acc -> + case Map.get(vidui, slot) do + list when is_list(list) and list != [] -> [slot | acc] + _ -> acc + end + end) + |> Enum.reverse() + end + + defp vidui_categories_filled(_), do: [] end diff --git a/vav_pipeline/priv/repo/migrations/20260428000001_create_azazel_records.exs b/vav_pipeline/priv/repo/migrations/20260428000001_create_azazel_records.exs new file mode 100644 index 00000000..948925f8 --- /dev/null +++ b/vav_pipeline/priv/repo/migrations/20260428000001_create_azazel_records.exs @@ -0,0 +1,95 @@ +defmodule VavPipeline.Repo.Migrations.CreateAzazelRecords do + @moduledoc """ + L3 — Azazel disposal records. Append-only, parallel to interpretations. + + Vayikra 16: the disposed thing goes to eretz gezerah and does not + return; the RECORD of the disposal remains as permanent oral + tradition. The two-goat architecture: this table parallels + interpretations the way the Azazel Processor parallels the Korban + Processor. + + PROPERTY 9 — observable disposal: + The public fact of disposal is part of the implementation, not + incidental to it. This table is the operator-internal half; + the /peah/azazel_log endpoint (commit 8) is the public half. + + PROPERTY 10 — hard termination: + The DISPOSED PAYLOAD IS NOT IN THIS TABLE. Only the vidui (what + was named) and the routing decision (always eretz_gezerah). A + schema CHECK constraint enforces routing = 'eretz_gezerah' so + a future migration cannot accidentally introduce a + destination-handle column. + + Append-only enforcement matches interpretations: REVOKE + UPDATE/DELETE on the shorashim_app role. + """ + + use Ecto.Migration + + def up do + create table(:azazel_records) do + # The opaque case_ref — content-addressable identifier of what + # was disposed (NOT the content itself). + add :case_ref, :text, null: false + + # The vidui — operator-produced enumerative commit, three-fold + # (avonot/peshaim/chatot) per Vayikra 16:21. Stored as JSONB + # so the structure is queryable but the content is not parsed + # by Postgres-side logic. + add :vidui, :jsonb, null: false + + # The routing — ALWAYS the literal string "eretz_gezerah". The + # CHECK constraint below makes this structural. + add :routing, :text, null: false + + # Property 7 — authorized routing + add :authorized_by, :text, null: false # kohen_gadol identifier + add :shaliach, :text, null: false # ish itti identifier + + # Audit metadata + add :triggering_call, :text # which subsystem invoked + add :goral_outcome_id, :text # if goral was cast, the + # SHA-256 outcome digest + # (audit-recomputable) + + # When the disposal happened (per the orchestrator's clock, not + # the L3 write clock — those can differ if the file watcher + # picks up a backlog). + add :disposed_at, :utc_datetime, null: false + + timestamps(type: :utc_datetime, updated_at: false) + end + + # Property 12 / property 7: routing must be the canonical opaque + # value. A future migration that wants to add a destination + # handle will fail this check; the operator must explicitly + # justify removing the constraint, which is the right friction. + create constraint( + :azazel_records, + :routing_must_be_eretz_gezerah, + check: "routing = 'eretz_gezerah'" + ) + + create index(:azazel_records, [:case_ref]) + create index(:azazel_records, [:disposed_at]) + # Index on shaliach for operator queries ("what did this handler + # dispose of?") — useful for tevilah audit and for property 4 + # accountability. + create index(:azazel_records, [:shaliach]) + + # Append-only enforcement — same pattern as interpretations. + execute """ + DO $$ + BEGIN + IF EXISTS (SELECT FROM pg_roles WHERE rolname = 'shorashim_app') THEN + REVOKE UPDATE, DELETE ON azazel_records FROM shorashim_app; + END IF; + END + $$; + """ + end + + def down do + drop table(:azazel_records) + end +end diff --git a/vav_pipeline/test/azazel_ingest_test.exs b/vav_pipeline/test/azazel_ingest_test.exs new file mode 100644 index 00000000..ec6a8327 --- /dev/null +++ b/vav_pipeline/test/azazel_ingest_test.exs @@ -0,0 +1,336 @@ +defmodule VavPipeline.AzazelIngestTest do + @moduledoc """ + Tests for the cross-language Azazel ingest GenServer (Pass 4 / + Finding F, commit 7). These tests verify the file-watcher logic + and the AzazelRecord changeset validation; they do NOT require + PostgreSQL because the recorder is injected as a mock. + + L3-side tests (the actual record_disposal write) live in + l3_store_test.exs tagged :l3. + """ + + use ExUnit.Case, async: false + + alias VavPipeline.AzazelIngest + alias VavPipeline.L3.AzazelRecord + + # ============================================================ + # Test helpers + # ============================================================ + + setup do + # Per-test pending directory so tests don't share state. + dir = + System.tmp_dir!() + |> Path.join("azazel-ingest-test-#{:erlang.unique_integer([:positive])}") + + File.mkdir_p!(dir) + on_exit(fn -> File.rm_rf!(dir) end) + %{pending_dir: dir} + end + + defp valid_record_json(case_ref \\ "case-001") do + %{ + "case_ref" => case_ref, + "vidui" => %{ + "avonot" => [], + "peshaim" => [], + "chatot" => ["empty-vital offering: test"], + "confessor" => "kohen-gadol", + "confessed_at" => "2026-04-28T12:00:00Z" + }, + "routing" => "eretz_gezerah", + "authorized_by" => "kohen-gadol", + "shaliach" => "aharon", + "triggering_call" => "external_offering", + "goral_outcome_id" => nil, + "disposed_at" => "2026-04-28T12:00:01Z" + } + |> Jason.encode!() + end + + defp drop_file(dir, name, body) do + File.write!(Path.join(dir, name), body) + end + + defp start_ingest(dir, opts \\ []) do + name = :"ingest_#{:erlang.unique_integer([:positive])}" + + captured = :ets.new(:captured, [:public, :bag]) + + recorder = fn attrs -> + :ets.insert(captured, {:record, attrs}) + {:ok, attrs} + end + + {:ok, pid} = + AzazelIngest.start_link( + Keyword.merge( + [ + name: name, + pending_dir: dir, + poll_interval_ms: 60_000, # don't auto-poll during tests + recorder: recorder + ], + opts + ) + ) + + on_exit(fn -> if Process.alive?(pid), do: GenServer.stop(pid) end) + %{server: name, captured: captured, recorder: recorder} + end + + # ============================================================ + # Happy path + # ============================================================ + + test "poll_now ingests a valid file and shreds it", %{pending_dir: dir} do + drop_file(dir, "case-001.json", valid_record_json()) + %{server: server, captured: captured} = start_ingest(dir) + + AzazelIngest.poll_now(server) + + # File was shredded + refute File.exists?(Path.join(dir, "case-001.json")) + + # Record was passed to the recorder + assert [{:record, attrs}] = :ets.tab2list(captured) + assert attrs.case_ref == "case-001" + assert attrs.routing == "eretz_gezerah" + assert attrs.authorized_by == "kohen-gadol" + assert attrs.shaliach == "aharon" + assert %DateTime{} = attrs.disposed_at + end + + test "poll_now ingests multiple files in sorted order", %{pending_dir: dir} do + drop_file(dir, "case-002.json", valid_record_json("case-002")) + drop_file(dir, "case-001.json", valid_record_json("case-001")) + drop_file(dir, "case-003.json", valid_record_json("case-003")) + + %{server: server, captured: captured} = start_ingest(dir) + AzazelIngest.poll_now(server) + + refs = + :ets.tab2list(captured) + |> Enum.map(fn {:record, attrs} -> attrs.case_ref end) + + assert refs == ["case-001", "case-002", "case-003"] + end + + test "ingest preserves the vidui as a map (jsonb column)", %{pending_dir: dir} do + drop_file(dir, "case-vidui.json", valid_record_json()) + %{server: server, captured: captured} = start_ingest(dir) + + AzazelIngest.poll_now(server) + + [{:record, attrs}] = :ets.tab2list(captured) + assert is_map(attrs.vidui) + assert attrs.vidui["chatot"] == ["empty-vital offering: test"] + assert attrs.vidui["confessor"] == "kohen-gadol" + end + + # ============================================================ + # Retry semantics — failed records left in place + # ============================================================ + + test "file is left in place when recorder returns error", %{pending_dir: dir} do + drop_file(dir, "case-fail.json", valid_record_json()) + + name = :"ingest_fail_#{:erlang.unique_integer([:positive])}" + + failing_recorder = fn _ -> {:error, :db_unavailable} end + + {:ok, pid} = + AzazelIngest.start_link( + name: name, + pending_dir: dir, + poll_interval_ms: 60_000, + recorder: failing_recorder + ) + + on_exit(fn -> if Process.alive?(pid), do: GenServer.stop(pid) end) + + AzazelIngest.poll_now(name) + + # File still present for retry + assert File.exists?(Path.join(dir, "case-fail.json")) + state = AzazelIngest.state(name) + assert state.errors == 1 + assert state.records_ingested == 0 + end + + test "file is left in place when JSON is malformed", %{pending_dir: dir} do + drop_file(dir, "case-malformed.json", "not-json{") + %{server: server} = start_ingest(dir) + + AzazelIngest.poll_now(server) + + # File still present (operator can investigate) + assert File.exists?(Path.join(dir, "case-malformed.json")) + state = AzazelIngest.state(server) + assert state.errors == 1 + end + + test "file is left in place when disposed_at is missing", %{pending_dir: dir} do + raw = + %{"case_ref" => "case-no-ts", "vidui" => %{}, "routing" => "eretz_gezerah"} + |> Jason.encode!() + + drop_file(dir, "case-no-ts.json", raw) + %{server: server} = start_ingest(dir) + + AzazelIngest.poll_now(server) + + assert File.exists?(Path.join(dir, "case-no-ts.json")) + end + + # ============================================================ + # State accounting + # ============================================================ + + test "state tracks polls_completed, records_ingested, errors", %{pending_dir: dir} do + drop_file(dir, "case-001.json", valid_record_json("case-001")) + drop_file(dir, "case-002.json", valid_record_json("case-002")) + drop_file(dir, "case-bad.json", "garbage") + + %{server: server} = start_ingest(dir) + + AzazelIngest.poll_now(server) + + state = AzazelIngest.state(server) + assert state.polls_completed == 1 + assert state.records_ingested == 2 + assert state.errors == 1 + end + + test "state survives multiple polls", %{pending_dir: dir} do + %{server: server} = start_ingest(dir) + AzazelIngest.poll_now(server) + AzazelIngest.poll_now(server) + AzazelIngest.poll_now(server) + + state = AzazelIngest.state(server) + assert state.polls_completed == 3 + end + + test "non-json files are ignored", %{pending_dir: dir} do + File.write!(Path.join(dir, "README.txt"), "not a record") + drop_file(dir, "case-001.json", valid_record_json()) + + %{server: server, captured: captured} = start_ingest(dir) + AzazelIngest.poll_now(server) + + # JSON file processed; README left alone + refute File.exists?(Path.join(dir, "case-001.json")) + assert File.exists?(Path.join(dir, "README.txt")) + assert length(:ets.tab2list(captured)) == 1 + end + + # ============================================================ + # Changeset validation (no DB) + # ============================================================ + + describe "AzazelRecord.changeset/2" do + test "valid attrs produce a valid changeset" do + attrs = %{ + case_ref: "case-cs-001", + vidui: %{ + "avonot" => [], + "peshaim" => [], + "chatot" => ["something"], + "confessor" => "kohen", + "confessed_at" => "2026-04-28T12:00:00Z" + }, + routing: "eretz_gezerah", + authorized_by: "kohen-gadol", + shaliach: "aharon", + disposed_at: ~U[2026-04-28 12:00:01Z] + } + + cs = AzazelRecord.changeset(%AzazelRecord{}, attrs) + assert cs.valid? + end + + test "rejects routing other than eretz_gezerah" do + attrs = base_attrs(%{routing: "void"}) + cs = AzazelRecord.changeset(%AzazelRecord{}, attrs) + refute cs.valid? + assert {:routing, _} = List.keyfind(cs.errors, :routing, 0) + end + + test "rejects vidui missing avonot slot" do + attrs = + base_attrs(%{ + vidui: %{ + "peshaim" => [], + "chatot" => ["x"], + "confessor" => "kohen", + "confessed_at" => "2026-04-28T12:00:00Z" + } + }) + + cs = AzazelRecord.changeset(%AzazelRecord{}, attrs) + refute cs.valid? + assert {:vidui, {msg, _}} = List.keyfind(cs.errors, :vidui, 0) + assert msg =~ "avonot" + end + + test "rejects vidui with all three categories empty" do + attrs = + base_attrs(%{ + vidui: %{ + "avonot" => [], + "peshaim" => [], + "chatot" => [], + "confessor" => "kohen", + "confessed_at" => "2026-04-28T12:00:00Z" + } + }) + + cs = AzazelRecord.changeset(%AzazelRecord{}, attrs) + refute cs.valid? + assert {:vidui, {msg, _}} = List.keyfind(cs.errors, :vidui, 0) + assert msg =~ "untagged" + end + + test "rejects missing required fields" do + cs = AzazelRecord.changeset(%AzazelRecord{}, %{}) + refute cs.valid? + + missing = Enum.map(cs.errors, fn {k, _} -> k end) + for required <- [:case_ref, :vidui, :routing, :authorized_by, :shaliach, :disposed_at] do + assert required in missing, "expected #{required} to be missing" + end + end + + test "accepts optional triggering_call and goral_outcome_id" do + attrs = base_attrs(%{ + triggering_call: "external_offering", + goral_outcome_id: String.duplicate("a", 64) + }) + + cs = AzazelRecord.changeset(%AzazelRecord{}, attrs) + assert cs.valid? + end + + defp base_attrs(overrides \\ %{}) do + Map.merge( + %{ + case_ref: "case-base-001", + vidui: %{ + "avonot" => [], + "peshaim" => [], + "chatot" => ["something"], + "confessor" => "kohen", + "confessed_at" => "2026-04-28T12:00:00Z" + }, + routing: "eretz_gezerah", + authorized_by: "kohen-gadol", + shaliach: "aharon", + disposed_at: ~U[2026-04-28 12:00:01Z] + }, + overrides + ) + end + end +end diff --git a/vav_pipeline/test/l3_store_test.exs b/vav_pipeline/test/l3_store_test.exs index 4cbd8637..b7d8671b 100644 --- a/vav_pipeline/test/l3_store_test.exs +++ b/vav_pipeline/test/l3_store_test.exs @@ -11,7 +11,7 @@ defmodule VavPipeline.L3.StoreTest do @moduletag :l3 alias VavPipeline.L3.Store - alias VavPipeline.L3.{Interpretation, MinorityOpinion, TeikuRecord} + alias VavPipeline.L3.{AzazelRecord, Interpretation, MinorityOpinion, TeikuRecord} setup do :ok = Ecto.Adapters.SQL.Sandbox.checkout(VavPipeline.Repo) @@ -293,6 +293,122 @@ defmodule VavPipeline.L3.StoreTest do assert "rejected" in statuses end + # === AZAZEL DISPOSAL RECORDS (Pass 4 / Finding F, commit 7) === + # These tests exercise the L3.Store.record_disposal/1 path against + # an actual database. Property 9 (operator-internal half) — the + # disposal becomes permanent oral tradition. + + @valid_disposal %{ + case_ref: "azazel_case_001", + vidui: %{ + "avonot" => [], + "peshaim" => [], + "chatot" => ["empty-vital offering: claude response on bereishit 1:1"], + "confessor" => "kohen-gadol-aharon", + "confessed_at" => "2026-04-28T12:00:00Z" + }, + routing: "eretz_gezerah", + authorized_by: "kohen-gadol-aharon", + shaliach: "ish-itti-001", + triggering_call: "external_offering.consult_external_model", + disposed_at: ~U[2026-04-28 12:00:01Z] + } + + test "record_disposal inserts a valid disposal" do + assert {:ok, %AzazelRecord{} = record} = Store.record_disposal(@valid_disposal) + assert record.id != nil + assert record.case_ref == "azazel_case_001" + assert record.routing == "eretz_gezerah" + assert record.shaliach == "ish-itti-001" + assert record.authorized_by == "kohen-gadol-aharon" + assert is_map(record.vidui) + assert record.vidui["confessor"] == "kohen-gadol-aharon" + end + + test "record_disposal rejects routing other than eretz_gezerah" do + attrs = Map.put(@valid_disposal, :routing, "void") + assert {:error, changeset} = Store.record_disposal(attrs) + assert %{routing: _} = errors_on(changeset) + end + + test "record_disposal rejects vidui with all categories empty" do + attrs = + Map.put(@valid_disposal, :vidui, %{ + "avonot" => [], + "peshaim" => [], + "chatot" => [], + "confessor" => "kohen", + "confessed_at" => "2026-04-28T12:00:00Z" + }) + + assert {:error, changeset} = Store.record_disposal(attrs) + errors = errors_on(changeset) + assert errors[:vidui] != nil + end + + test "record_disposal rejects missing required fields" do + assert {:error, changeset} = Store.record_disposal(%{routing: "eretz_gezerah"}) + refute changeset.valid? + end + + test "count_disposals returns 0 when none recorded" do + assert Store.count_disposals() == 0 + end + + test "count_disposals returns total after several disposals" do + {:ok, _} = Store.record_disposal(@valid_disposal) + {:ok, _} = Store.record_disposal(Map.put(@valid_disposal, :case_ref, "azazel_case_002")) + {:ok, _} = Store.record_disposal(Map.put(@valid_disposal, :case_ref, "azazel_case_003")) + assert Store.count_disposals() == 3 + end + + test "count_disposals(since) filters by disposed_at" do + early = + @valid_disposal + |> Map.put(:case_ref, "early") + |> Map.put(:disposed_at, ~U[2026-04-01 00:00:00Z]) + + late = + @valid_disposal + |> Map.put(:case_ref, "late") + |> Map.put(:disposed_at, ~U[2026-04-30 00:00:00Z]) + + {:ok, _} = Store.record_disposal(early) + {:ok, _} = Store.record_disposal(late) + + assert Store.count_disposals(~U[2026-04-15 00:00:00Z]) == 1 + end + + test "disposal_summary returns categorical info but NOT vidui items" do + {:ok, _} = Store.record_disposal(@valid_disposal) + + [summary] = Store.disposal_summary() + + # Has the expected structural fields + assert summary.case_ref == "azazel_case_001" + assert summary.shaliach == "ish-itti-001" + assert summary.triggering_call == "external_offering.consult_external_model" + assert "chatot" in summary.categories_filled + + # Privacy boundary (property 9): the actual vidui items are NOT + # in the summary. The endpoint exposes which categories were + # filled, not what was named. + assert :items not in Map.keys(summary) + refute Enum.any?(Map.values(summary), fn v -> + v == "empty-vital offering: claude response on bereishit 1:1" + end) + end + + test "disposal_summary categories_filled accurately reflects vidui" do + avonot_only = put_in(@valid_disposal, [:vidui, "avonot"], ["halt avodah_zarah: x"]) + avonot_only = put_in(avonot_only, [:vidui, "chatot"], []) + avonot_only = Map.put(avonot_only, :case_ref, "avonot-only") + {:ok, _} = Store.record_disposal(avonot_only) + + [summary] = Store.disposal_summary() + assert summary.categories_filled == ["avonot"] + end + # Helper to extract error messages from changeset defp errors_on(changeset) do Ecto.Changeset.traverse_errors(changeset, fn {msg, opts} ->