vav_pipeline: commit 7 — L3 azazel_records + cross-language ingest GenServer

Pass 4 / Finding F, commit 7 of 10 from BUILD_AZAZEL.md. The
operator-internal half of property 9 (observable disposal). Crosses
the Python/Elixir boundary: the Python orchestrator drops completed
disposal records as JSON files; this commit's GenServer reads them
and persists to L3.

The public half (a /peah/azazel_log read endpoint) lands in commit 8;
the routing-from-Korban + Chatzotzrot wiring lands in commit 9. After
those, the Azazel Processor is end-to-end functional.

CREATED:

  vav_pipeline/priv/repo/migrations/20260428000001_create_azazel_records.exs
    azazel_records table — append-only, parallel to interpretations.
    Columns: case_ref, vidui (jsonb), routing, authorized_by,
    shaliach, triggering_call, goral_outcome_id, disposed_at.
    Indexes on case_ref, disposed_at, shaliach.

    Property-12-defending CHECK constraint:
      routing_must_be_eretz_gezerah CHECK (routing = 'eretz_gezerah').
    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.

    Append-only enforcement matches interpretations: REVOKE
    UPDATE/DELETE on the shorashim_app role.

  vav_pipeline/lib/vav_pipeline/l3/azazel_record.ex
    Ecto schema mirroring the Python AzazelRecord.to_storage_dict()
    shape. Changeset validates:
      - all required fields present
      - routing == "eretz_gezerah" (mirrors the DB CHECK at app layer)
      - vidui is a map with avonot/peshaim/chatot/confessor/
        confessed_at slots present
      - vidui's three categories not all empty (mirrors the Python
        Vidui.__post_init__ EmptyConfessionError)

  vav_pipeline/lib/vav_pipeline/azazel_ingest.ex      ~190 lines
    GenServer that polls a configurable pending-l3 directory
    (default /tmp/shorashim/azazel/pending_l3). For each *.json file:
      1. Read + JSON-parse
      2. Translate Python serialization → Ecto attrs
      3. Call recorder (default L3.Store.record_disposal/1)
      4. On success, shred the source file (property-10 follow-through)
      5. On failure, leave file in place for retry; log warning

    Recorder is injectable for tests — mock recorder lets us verify
    the pipeline without PostgreSQL.

    Poll interval default 5s, configurable. Polling rather than
    file_system events to keep the dependency surface minimal; the
    Azazel cadence is exceptional content, not high-throughput.

    Not added to the supervision tree in this commit — the producer
    side (Python orchestrator dropping files) is wired in commit 9.
    Adding to the supervisor before there's a producer means a
    poller doing nothing useful. Operator can start manually for
    early testing: VavPipeline.AzazelIngest.start_link(...).

CHANGED:

  vav_pipeline/lib/vav_pipeline/l3/store.ex
    + record_disposal/1
        Inserts an AzazelRecord. Returns {:ok, record} or
        {:error, changeset}.

    + count_disposals/1
        Total or since-a-timestamp count. Used by the public
        observability endpoint (commit 8).

    + disposal_summary/1
        Returns categorical info per disposal — case_ref, shaliach,
        triggering_call, disposed_at, categories_filled (which of
        avonot/peshaim/chatot had any items, NOT the items
        themselves). PROPERTY-9 PRIVACY BOUNDARY at the public
        side: items are operator-internal only, accessed via
        future /kodesh/* endpoints.

TESTS:

  vav_pipeline/test/azazel_ingest_test.exs           15 tests, no DB
    Happy path:
      - poll_now ingests valid file and shreds it
      - multiple files processed in sorted order
      - vidui preserved as map (jsonb shape)
    Retry semantics:
      - recorder error → file left in place + state.errors++
      - malformed JSON → file left in place
      - missing disposed_at → file left in place
    State accounting:
      - polls_completed, records_ingested, errors all tracked
      - state survives multiple polls
      - non-json files ignored
    Changeset (no DB):
      - valid attrs
      - rejects routing != eretz_gezerah
      - rejects vidui missing slots
      - rejects vidui all-empty (untagged disposal)
      - rejects missing required fields
      - accepts optional triggering_call + goral_outcome_id

  vav_pipeline/test/l3_store_test.exs                +9 :l3-tagged
    record_disposal:
      - inserts valid disposal
      - rejects routing != eretz_gezerah
      - rejects all-empty vidui
      - rejects missing required fields
    Counts:
      - count_disposals == 0 initially
      - count_disposals after several disposals
      - count_disposals(since) filters by timestamp
    Summary:
      - disposal_summary returns categorical info but NOT items
        (privacy boundary verification)
      - categories_filled accurately reflects vidui shape

Tests: vav_pipeline 353 passed, 0 failures, 33 excluded (was
329/24; +15 visible azazel_ingest + 9 :l3-tagged disposal store).
korban_processor + azazel_processor unchanged.

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

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

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

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