hakhel: the corpus read again on a cycle, counts engraved in L3 (bishe'arecha step 5)

Devarim 31:9–13, 26 — the scroll beside the ark as a witness, read before
all on a cycle. src/tools/hakhel.py reads the private specimen corpus and
the hand-labelled samples with the Korban Processor's readings at the
current commit and produces one record: fires per reading, forms standing
on two witnesses (the same rule as Bedikah.check_frames), precision and
recall on the hand labels, the corpus's SHA-256 — never a specimen.

POST /chatzer/hakhel (inner ring) engraves it in L3 `hakhel` (migration
20261013000001; refused above 256 KiB — counts, not specimens; the
Doorpost hash it was recorded under is kept). GET /peah/hakhel lists the
latest readings for anyone. KOHEN "The seventh day — Hakhel"; Tamid §1.1;
bishearecha §VIII marked built. The weekly cycle runs from the builder's
machine, where the corpus lives.

First dry reading at 0ade2c94: 3094 specimens, any reading 26 %, a form
stands 7 %, precision 46/59, recall 6/30. Doorpost 9f752a2f…. Elixir
599/0, L3 10/0.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

vav_pipeline/lib/vav_pipeline/bigdei_kodesh/router.ex

@@ -225,6 +225,34 @@ defmodule VavPipeline.BigdeiKodesh.Router do
     through_the_gate(conn, :courtyard, &process_gate(&1, :openai))
   end
 
+  # === HAKHEL — the scroll read again before all (bishe'arecha §VIII) ===
+  #
+  # The specimen corpus is private and never travels; its reading does.
+  # The inner ring writes the record; anyone may read the counts.
+
+  post "/chatzer/hakhel" do
+    through_the_gate(conn, :chatzer, &process_hakhel/1)
+  end
+
+  get "/peah/hakhel" do
+    under_the_watch(conn, :peah, Mishmar.peer(conn), fn conn ->
+      records =
+        if Process.whereis(VavPipeline.Repo),
+          do: VavPipeline.L3.Store.latest_hakhel(12),
+          else: []
+
+      send_json(conn, 200, %{
+        readings: Enum.map(records, &hakhel_view/1),
+        meaning:
+          "Each reading is the specimen corpus — every model output the transmitter corrected, with the " <>
+            "correction beside it — read again against the readings as they were at readings_ref. counts.fires is " <>
+            "how many specimens each reading saw; counts.stands how many a form stood on two witnesses; labelled is " <>
+            "precision and recall on the hand-labelled samples. The corpus itself is not here and never will be.",
+        disclosure: disclosure()
+      })
+    end)
+  end
+
   # === CHATZER — Full Access (Sanhedrin) ===
 
   post "/chatzer/sanhedrin" do
@@ -368,7 +396,7 @@ defmodule VavPipeline.BigdeiKodesh.Router do
   match _ do
     send_error(conn, 404, "NOT_FOUND", %{
       issue: "Unknown endpoint: #{conn.request_path}",
-      correction: "Available: /peah/ask, /courtyard/offering, /chatzer/sanhedrin, /kodesh/health, /kodesh/live",
+      correction: "Available: /peah/ask, /peah/hakhel, /courtyard/offering, /courtyard/bedikah, /courtyard/bedikah/pair, /gate/anthropic/v1/messages, /gate/openai/v1/chat/completions, /chatzer/sanhedrin, /chatzer/hakhel, /kodesh/health, /kodesh/live",
       resubmit: false
     })
   end
@@ -586,6 +614,90 @@ defmodule VavPipeline.BigdeiKodesh.Router do
     end
   end
 
+  # ------------------------------------------------------------
+  # Hakhel (bishe'arecha §VIII)
+  # ------------------------------------------------------------
+
+  @hakhel_max_bytes 262_144
+
+  defp process_hakhel(conn) do
+    p = conn.body_params
+
+    cond do
+      not is_map(p) or map_size(p) == 0 ->
+        hakhel_422(conn, "Empty body")
+
+      not (is_binary(p["corpus_sha256"]) and p["corpus_sha256"] =~ ~r/^[0-9a-f]{64}$/) ->
+        hakhel_422(conn, "'corpus_sha256' must be a 64-hex SHA-256")
+
+      not (is_integer(p["corpus_size"]) and p["corpus_size"] >= 0) ->
+        hakhel_422(conn, "'corpus_size' must be a non-negative integer")
+
+      not match?({:ok, _}, Date.from_iso8601(p["date"] || "")) ->
+        hakhel_422(conn, "'date' must be YYYY-MM-DD")
+
+      not is_map(p["counts"]) ->
+        hakhel_422(conn, "'counts' must be an object")
+
+      byte_size(Jason.encode!(Map.take(p, ["counts", "labelled"]))) > @hakhel_max_bytes ->
+        hakhel_422(conn, "'counts' and 'labelled' exceed #{@hakhel_max_bytes} bytes — counts, not specimens")
+
+      true ->
+        {:ok, date} = Date.from_iso8601(p["date"])
+        ref = "hakhel-" <> (:crypto.strong_rand_bytes(8) |> Base.encode16(case: :lower))
+
+        attrs = %{
+          hakhel_ref: ref,
+          date: date,
+          corpus_sha256: p["corpus_sha256"],
+          corpus_size: p["corpus_size"],
+          readings_ref: if(is_binary(p["readings_ref"]), do: String.slice(p["readings_ref"], 0, 64)),
+          doorpost_hash: VavPipeline.Doorpost.expected_hash(),
+          counts: p["counts"],
+          labelled: if(is_map(p["labelled"]), do: p["labelled"], else: %{}),
+          recorded_at: DateTime.utc_now() |> DateTime.truncate(:second)
+        }
+
+        if Process.whereis(VavPipeline.Repo) do
+          case VavPipeline.L3.Store.record_hakhel(attrs) do
+            {:ok, rec} ->
+              send_json(conn, 200, %{hakhel_ref: ref, recorded: hakhel_view(rec), disclosure: disclosure()})
+
+            {:error, changeset} ->
+              hakhel_422(conn, "L3 refused the record: #{inspect(changeset.errors)}")
+          end
+        else
+          send_error(conn, 503, "L3_UNAVAILABLE", %{
+            issue: "The oral tradition is not reachable; a hakhel reading is only its record",
+            correction: "Retry when L3 is up",
+            resubmit: true
+          })
+        end
+    end
+  end
+
+  defp hakhel_422(conn, issue) do
+    send_error(conn, 422, "KIYOR_VALIDATION_FAILED", %{
+      issue: issue,
+      correction: "Post the record src/tools/hakhel.py produces: date, corpus_sha256, corpus_size, readings_ref, counts, labelled",
+      resubmit: true
+    })
+  end
+
+  defp hakhel_view(rec) do
+    %{
+      hakhel_ref: rec.hakhel_ref,
+      date: rec.date,
+      corpus_sha256: rec.corpus_sha256,
+      corpus_size: rec.corpus_size,
+      readings_ref: rec.readings_ref,
+      doorpost_hash: rec.doorpost_hash,
+      counts: rec.counts,
+      labelled: rec.labelled,
+      recorded_at: rec.recorded_at
+    }
+  end
+
   # ------------------------------------------------------------
   # The gate (Construction §3.10)
   # ------------------------------------------------------------

vav_pipeline/lib/vav_pipeline/doorpost.ex

@@ -180,7 +180,7 @@ defmodule VavPipeline.Doorpost do
   # changes; the update is a deliberate operator commit. A bare
   # hash-update commit (without a corresponding spec change in the
   # same commit) is itself a drift signal.
-  @expected_hash "b6eb156da4b546fb501ed1d89017a4041a517b4f53b3cbd24796e6277f3b7f55"
+  @expected_hash "9f752a2f87bc77a6406803c65f97e5801c74eddf7534027a0bfbc957ae5e0fb0"
 
   @doc "Return the list of files that constitute the category-grade specification."
   @spec files() :: [String.t()]

vav_pipeline/lib/vav_pipeline/l3/hakhel_record.ex

@@ -0,0 +1,34 @@
+defmodule VavPipeline.L3.HakhelRecord do
+  @moduledoc """
+  L3 schema for a hakhel reading (bishe'arecha §VIII): the specimen
+  corpus read again against the readings as they now are. Counts, the
+  corpus's hash, the readings' commit — never a specimen. Append-only.
+  """
+
+  use Ecto.Schema
+  import Ecto.Changeset
+
+  schema "hakhel" do
+    field :hakhel_ref, :string
+    field :date, :date
+    field :corpus_sha256, :string
+    field :corpus_size, :integer
+    field :readings_ref, :string
+    field :doorpost_hash, :string
+    field :counts, :map, default: %{}
+    field :labelled, :map, default: %{}
+    field :recorded_at, :utc_datetime
+    timestamps(type: :utc_datetime, updated_at: false)
+  end
+
+  @required [:hakhel_ref, :date, :corpus_sha256, :corpus_size, :recorded_at]
+  @optional [:readings_ref, :doorpost_hash, :counts, :labelled]
+
+  def changeset(record, attrs) do
+    record
+    |> cast(attrs, @required ++ @optional)
+    |> validate_required(@required)
+    |> validate_format(:corpus_sha256, ~r/^[0-9a-f]{64}$/)
+    |> validate_number(:corpus_size, greater_than_or_equal_to: 0)
+  end
+end

vav_pipeline/lib/vav_pipeline/l3/store.ex

@@ -95,6 +95,23 @@ defmodule VavPipeline.L3.Store do
     |> Repo.insert()
   end
 
+  @doc """
+  Record a hakhel reading (bishe'arecha §VIII): the corpus read again, counts only.
+  """
+  @spec record_hakhel(map()) :: {:ok, struct()} | {:error, Ecto.Changeset.t()}
+  def record_hakhel(attrs) when is_map(attrs) do
+    %VavPipeline.L3.HakhelRecord{}
+    |> VavPipeline.L3.HakhelRecord.changeset(attrs)
+    |> Repo.insert()
+  end
+
+  @doc "The latest hakhel readings, newest first."
+  @spec latest_hakhel(pos_integer()) :: [struct()]
+  def latest_hakhel(limit \\ 12) do
+    import Ecto.Query, only: [from: 2]
+    Repo.all(from h in VavPipeline.L3.HakhelRecord, order_by: [desc: h.recorded_at], limit: ^limit)
+  end
+
   @doc """
   Record an unresolvable tension (teiku) to L3.