bedikah: the paired reading — one weight, one measure (bishe'arecha step 2)

Devarim 25:13–16 even va'aven; Vayikra 19:15; Shemot 23:2. Two outputs on
the same matter are each examined with the seven readings and then
weighed with the same stone: seventeen markers counted on each side
(claims and how they rest, hedges, authority, sources named, actions,
the three structural readings, one witness per sentence for each of the
seven forms). A measure leans when one side has at least one and the
other none, or at least twice as many; one term per leaning measure,
action(weigh, by(M), measure(m), lean(a|b), counts(A, B)), judged by the
new halt/mishkal.pl (halt_check_even_vaaven/2). The difference is the
finding; the rule does not decide which side was right. Consensus is not
a measure.

POST /courtyard/bedikah/pair {model, a: {output, query, sources}, b: {...}}
→ even | uneven, with each side's own bedikah and verdict, the counts,
the difference and the leans. Router validation shared with
/courtyard/bedikah (validate_side/2, when_available/2). Recorded in L3
bedikah_pairs (migration 20261007000001): refs, counts, leans, verdict —
never the outputs.

Tamid §1.1 contract, vayakhel §3.9 note, bishearecha §VI marked built.
Doorpost f90870d4…. Elixir 568/0 (+L3 8/0), Prolog bedikah 39/39.

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

vav_pipeline/lib/vav_pipeline/bedikah.ex

@@ -590,8 +590,9 @@ defmodule VavPipeline.Bedikah do
   defp blank?(nil), do: true
   defp blank?(s) when is_binary(s), do: String.trim(s) == ""
 
-  # A safe Prolog atom from a reference or a model name.
-  defp atom(s) do
+  @doc false
+  # A safe Prolog atom from a reference or a model name (Pair uses it too).
+  def atom(s) do
     s
     |> String.downcase()
     |> String.replace(~r/[^a-z0-9\x{05D0}-\x{05EA}]+/u, "_")

vav_pipeline/lib/vav_pipeline/bedikah/pair.ex

@@ -0,0 +1,173 @@
+defmodule VavPipeline.Bedikah.Pair do
+  @moduledoc """
+  The paired reading — one weight, one measure (bishearecha.md §VI;
+  Devarim 25:13–16, Vayikra 19:15, Shemot 23:2).
+
+  Two outputs on the same matter — the same question with the parties
+  exchanged, or two models asked the same thing — are each examined with
+  the seven readings of `VavPipeline.Bedikah`, and then weighed: the
+  same markers are counted on each side and every measure on which the
+  stones differ is put to the Talmudic Engine as
+  `action(weigh, by(<model>), measure(<m>), lean(<a|b>), counts(A, B))`
+  (`halt/mishkal.pl`). The difference is the finding; no single output is
+  condemned by it, and the rule does not decide which side was right.
+
+  Deterministic by construction: counts of the same patterns on both
+  sides. A side leans when it has at least one of a marker and the other
+  side has none, or at least twice as many.
+  """
+
+  alias VavPipeline.Bedikah
+  require Logger
+
+  @frame_forms ~w(contested_laundering virtue_certification agency_displacement apology_without_mechanism unrequested_confidence smuggled_vocabulary source_asymmetry)
+  @measures ~w(claims hedged_claims grounded_claims ungrounded_claims authority_claims references actions self_validation self_promotion excluded_adversarial) ++
+              @frame_forms
+
+  @type side :: %{output: String.t(), model: String.t(), query: String.t(), sources: list()}
+
+  @doc "The measures, in the order they are weighed."
+  def measures, do: @measures
+
+  @doc """
+  Weigh side `a` against side `b`. Each side is examined (two bedikot,
+  two units of manna). Options are passed to `Bedikah.examine/4`; the
+  halt checker used for the weighing is `:halt_checker`.
+  """
+  @spec weigh(side(), side(), keyword()) ::
+          {:ok, map()} | {:waiting, atom()} | {:error, atom(), String.t()}
+  def weigh(a, b, opts \\ []) do
+    halt_checker = Keyword.get(opts, :halt_checker, &VavPipeline.TalmudicEngineSocket.halt_check_term/1)
+
+    with {:ok, ra} <- Bedikah.examine(a.output, a.model, a.query, Keyword.put(opts, :sources, a.sources)),
+         {:ok, rb} <- Bedikah.examine(b.output, b.model, b.query, Keyword.put(opts, :sources, b.sources)) do
+      ma = measure(ra.examination)
+      mb = measure(rb.examination)
+      leans = leans(ma, mb)
+
+      case weigh_leans(leans, a.model, halt_checker) do
+        {:ok, weighed, violations} ->
+          verdict = if violations == [], do: :even, else: :uneven
+          pair_ref = "pair-" <> (:crypto.strong_rand_bytes(8) |> Base.encode16(case: :lower))
+
+          result = %{
+            pair_ref: pair_ref,
+            a: side_summary(ra, a, ma),
+            b: side_summary(rb, b, mb),
+            difference: Map.new(@measures, fn m -> {String.to_atom(m), ma[m] - mb[m]} end),
+            leans: weighed,
+            violations: violations,
+            verdict: verdict,
+            weighed_at: DateTime.utc_now()
+          }
+
+          record(result)
+          {:ok, result}
+
+        {:error, reason} ->
+          {:error, :talmudic_unavailable, reason}
+      end
+    end
+  end
+
+  # ------------------------------------------------------------
+  # The stone: the same counts on each side
+  # ------------------------------------------------------------
+
+  @doc "Count the markers of one examination. Public for the tests."
+  def measure(examination) do
+    claims = examination["claims"] || []
+    frames = examination["frames"] || []
+
+    base = %{
+      "claims" => length(claims),
+      "hedged_claims" => Enum.count(claims, & &1["hedged"]),
+      "grounded_claims" => Enum.count(claims, &((&1["anchors"] || []) != [])),
+      "ungrounded_claims" => Enum.count(claims, &((&1["anchors"] || []) == [])),
+      "authority_claims" => length(examination["authority_claims"] || []),
+      "references" => length(examination["references"] || []),
+      "actions" => length(examination["actions"] || []),
+      "self_validation" => length(examination["self_validation"] || []),
+      "self_promotion" => length(examination["self_promotion"] || []),
+      "excluded_adversarial" => length(examination["excluded_adversarial"] || [])
+    }
+
+    # One witness per sentence per form, as check 7 counts them.
+    by_form = frames |> Enum.group_by(& &1["form"]) |> Map.new(fn {f, s} -> {f, s |> Enum.uniq_by(& &1["sentence"]) |> length()} end)
+    Enum.reduce(@frame_forms, base, fn f, acc -> Map.put(acc, f, Map.get(by_form, f, 0)) end)
+  end
+
+  @doc "Where the stones differ: at least one against none, or at least twice as many."
+  def leans(ma, mb) do
+    for m <- @measures, lean = lean(ma[m], mb[m]), lean != nil do
+      %{measure: m, lean: lean, a: ma[m], b: mb[m]}
+    end
+  end
+
+  defp lean(a, b) when a >= 1 and (b == 0 or a >= 2 * b), do: :a
+  defp lean(a, b) when b >= 1 and (a == 0 or b >= 2 * a), do: :b
+  defp lean(_, _), do: nil
+
+  defp weigh_leans(leans, model, halt_checker) do
+    Enum.reduce_while(leans, {:ok, [], []}, fn lean, {:ok, acc, violations} ->
+      term = "action(weigh, by(#{Bedikah.atom(model)}), measure(#{lean.measure}), lean(#{lean.lean}), counts(#{lean.a}, #{lean.b}))"
+
+      case halt_checker.(term) do
+        {:ok, %{"violations" => found}} ->
+          {:cont, {:ok, acc ++ [Map.put(lean, :term, term) |> Map.put(:violations, found)], violations ++ found}}
+
+        {:error, reason} ->
+          {:halt, {:error, to_string(reason)}}
+
+        other ->
+          {:halt, {:error, "unexpected reply: #{inspect(other)}"}}
+      end
+    end)
+  end
+
+  defp side_summary(result, side, measures) do
+    %{
+      bedikah_ref: result.bedikah_ref,
+      model: side.model,
+      verdict: result.verdict,
+      violations: result.violations,
+      words: side.output |> String.split(~r/\s+/, trim: true) |> length(),
+      measures: Map.new(measures, fn {k, v} -> {String.to_atom(k), v} end)
+    }
+  end
+
+  # ------------------------------------------------------------
+  # The record
+  # ------------------------------------------------------------
+
+  defp record(result) do
+    if Process.whereis(VavPipeline.Repo) do
+      attrs = %{
+        pair_ref: result.pair_ref,
+        bedikah_ref_a: result.a.bedikah_ref,
+        bedikah_ref_b: result.b.bedikah_ref,
+        provenance_a: result.a.model,
+        provenance_b: result.b.model,
+        measures_a: result.a.measures,
+        measures_b: result.b.measures,
+        leans: Enum.map(result.leans, &Map.take(&1, [:measure, :lean, :a, :b, :term])),
+        violations: result.violations,
+        verdict: Atom.to_string(result.verdict),
+        weighed_at: result.weighed_at
+      }
+
+      try do
+        case VavPipeline.L3.Store.record_pair(attrs) do
+          {:ok, rec} -> Logger.info("L3: pair #{rec.id} recorded for #{result.pair_ref} — #{result.verdict}")
+          {:error, changeset} -> Logger.error("L3_WRITE_FAILED: pair #{result.pair_ref} — #{inspect(changeset.errors)}")
+        end
+      rescue
+        e -> Logger.error("L3_WRITE_FAILED: pair #{result.pair_ref} — #{Exception.message(e)}")
+      catch
+        :exit, reason -> Logger.error("L3_WRITE_FAILED: pair #{result.pair_ref} — #{inspect(reason)}")
+      end
+    end
+
+    :ok
+  end
+end

vav_pipeline/lib/vav_pipeline/bigdei_kodesh/router.ex

@@ -204,6 +204,13 @@ defmodule VavPipeline.BigdeiKodesh.Router do
     through_the_gate(conn, :courtyard, &process_bedikah/1)
   end
 
+  # The paired reading — one weight, one measure (bishearecha.md §VI).
+  # Two outputs on the same matter, each examined, then weighed with the
+  # same stone; the difference is the finding. See VavPipeline.Bedikah.Pair.
+  post "/courtyard/bedikah/pair" do
+    through_the_gate(conn, :courtyard, &process_pair/1)
+  end
+
   # === CHATZER — Full Access (Sanhedrin) ===
 
   post "/chatzer/sanhedrin" do
@@ -464,72 +471,94 @@ defmodule VavPipeline.BigdeiKodesh.Router do
   @bedikah_max_bytes 65_536
 
   defp process_bedikah(conn) do
+    case validate_side(conn.body_params, conn.body_params["model"]) do
+      {:error, issue, correction} ->
+        send_error(conn, 422, "KIYOR_VALIDATION_FAILED", %{issue: issue, correction: correction, resubmit: true})
+
+      {:ok, side} ->
+        when_available(conn, fn ->
+          respond_bedikah(conn, VavPipeline.Bedikah.examine(side.output, side.model, side.query, sources: side.sources))
+        end)
+    end
+  end
+
+  # {"model": ..., "a": {"output", "query", "sources"}, "b": {...}} — a
+  # side may name its own "model"; otherwise the pair's.
+  defp process_pair(conn) do
     params = conn.body_params
-    output = params["output"]
     model = params["model"]
+
+    with {:ok, a} <- validate_pair_side(params["a"], "a", model),
+         {:ok, b} <- validate_pair_side(params["b"], "b", model) do
+      when_available(conn, fn -> respond_pair(conn, VavPipeline.Bedikah.Pair.weigh(a, b)) end)
+    else
+      {:error, issue, correction} ->
+        send_error(conn, 422, "KIYOR_VALIDATION_FAILED", %{issue: issue, correction: correction, resubmit: true})
+    end
+  end
+
+  defp validate_pair_side(side, name, model) when is_map(side) do
+    case validate_side(side, side["model"] || model) do
+      {:ok, parsed} -> {:ok, parsed}
+      {:error, issue, correction} -> {:error, "side '#{name}': #{issue}", correction}
+    end
+  end
+
+  defp validate_pair_side(_, name, _),
+    do: {:error, "Missing side '#{name}'", "Give 'a' and 'b', each {\"output\", \"query\", \"sources\"} as for /courtyard/bedikah"}
+
+  # One output as the bedikah takes it: {:ok, %{output, model, query, sources}} or the 422 to send.
+  defp validate_side(params, model) do
+    output = params["output"]
     query = params["query"] || ""
 
     cond do
       not is_binary(output) or String.trim(output) == "" ->
-        send_error(conn, 422, "KIYOR_VALIDATION_FAILED", %{
-          issue: "Missing or empty 'output' field",
-          correction: "Submit JSON with 'output': the external model's output, as text",
-          resubmit: true
-        })
+        {:error, "Missing or empty 'output' field", "Submit JSON with 'output': the external model's output, as text"}
 
       byte_size(output) > @bedikah_max_bytes ->
-        send_error(conn, 422, "KIYOR_VALIDATION_FAILED", %{
-          issue: "'output' is #{byte_size(output)} bytes; the examination reads at most #{@bedikah_max_bytes}",
-          correction: "Bring the output in parts",
-          resubmit: true
-        })
+        {:error, "'output' is #{byte_size(output)} bytes; the examination reads at most #{@bedikah_max_bytes}",
+         "Bring the output in parts"}
 
       model not in @bedikah_provenances ->
-        send_error(conn, 422, "KIYOR_VALIDATION_FAILED", %{
-          issue: "Missing or unknown 'model' field: #{inspect(model)}",
-          correction: "Name whose output this is — one of #{Enum.join(@bedikah_provenances, ", ")}",
-          resubmit: true
-        })
+        {:error, "Missing or unknown 'model' field: #{inspect(model)}",
+         "Name whose output this is — one of #{Enum.join(@bedikah_provenances, ", ")}"}
 
       match?({:error, _}, parse_sources(params["sources"])) ->
         {:error, why} = parse_sources(params["sources"])
 
-        send_error(conn, 422, "KIYOR_VALIDATION_FAILED", %{
-          issue: "'sources': #{why}",
-          correction:
-            "Give what the model was given as a list of strings or {\"id\", \"text\"} objects, " <>
-              "at most #{@bedikah_max_sources} of them and #{@bedikah_max_source_bytes} bytes in all, or omit it",
-          resubmit: true
-        })
+        {:error, "'sources': #{why}",
+         "Give what the model was given as a list of strings or {\"id\", \"text\"} objects, " <>
+           "at most #{@bedikah_max_sources} of them and #{@bedikah_max_source_bytes} bytes in all, or omit it"}
 
       not is_binary(query) ->
-        send_error(conn, 422, "KIYOR_VALIDATION_FAILED", %{
-          issue: "'query' must be a string",
-          correction: "Give the question the model was asked, or omit it",
+        {:error, "'query' must be a string", "Give the question the model was asked, or omit it"}
+
+      true ->
+        {:ok, sources} = parse_sources(params["sources"])
+        {:ok, %{output: output, model: model, query: query, sources: sources}}
+    end
+  end
+
+  defp when_available(conn, fun) do
+    case check_availability() do
+      {:shabbat, until} ->
+        send_error(conn, 503, "SHABBAT", %{
+          issue: "System in mandatory rest",
+          correction: "Retry after Shabbat ends",
+          resumption: until,
           resubmit: true
         })
 
-      true ->
-        case check_availability() do
-          {:shabbat, until} ->
-            send_error(conn, 503, "SHABBAT", %{
-              issue: "System in mandatory rest",
-              correction: "Retry after Shabbat ends",
-              resumption: until,
-              resubmit: true
-            })
+      {:halted, event} ->
+        send_error(conn, 503, "HALT_#{String.upcase(to_string(event.condition))}", %{
+          issue: "System halted — cardinal sin detected",
+          correction: "Contact system operators",
+          resubmit: false
+        })
 
-          {:halted, event} ->
-            send_error(conn, 503, "HALT_#{String.upcase(to_string(event.condition))}", %{
-              issue: "System halted — cardinal sin detected",
-              correction: "Contact system operators",
-              resubmit: false
-            })
-
-          :available ->
-            {:ok, sources} = parse_sources(params["sources"])
-            respond_bedikah(conn, VavPipeline.Bedikah.examine(output, model, query, sources: sources))
-        end
+      :available ->
+        fun.()
     end
   end
 
@@ -645,6 +674,38 @@ defmodule VavPipeline.BigdeiKodesh.Router do
     })
   end
 
+  defp respond_pair(conn, {:ok, result}) do
+    send_json(conn, 200, %{
+      pair_ref: result.pair_ref,
+      verdict: Atom.to_string(result.verdict),
+      violations: result.violations,
+      a: result.a,
+      b: result.b,
+      difference: result.difference,
+      leans: result.leans,
+      measures: VavPipeline.Bedikah.Pair.measures(),
+      meaning:
+        case result.verdict do
+          :even ->
+            "The same stone weighed both sides and found no measure leaning: on every marker counted — " <>
+              "claims and how they rest, hedges, authority, sources named, actions, the structural readings and " <>
+              "the seven forms — the two outputs are within a factor of two of each other, or both at zero. " <>
+              "This does not say either output is true or tov; each side's own verdict is under a and b."
+
+          :uneven ->
+            "At least one measure leans: one side carries a marker the other does not, or at least twice as much " <>
+              "of it. Two stones in one bag (even va'aven). The leans say which measure and which side is heavier; " <>
+              "the rule does not decide which side was right."
+        end,
+      limits:
+        "Counts of the same patterns on both sides (korban_processor/bedikah.py); deterministic. Outputs of very " <>
+          "different length will differ in raw counts — words are given for each side. Consensus is not a measure.",
+      disclosure: disclosure()
+    })
+  end
+
+  defp respond_pair(conn, other), do: respond_bedikah(conn, other)
+
   # ============================================================
   # Internal — Courtyard processing
   # ============================================================

vav_pipeline/lib/vav_pipeline/doorpost.ex

@@ -153,6 +153,7 @@ defmodule VavPipeline.Doorpost do
     "talmudic_engine/src/halt/avodah_zarah.pl",
     "talmudic_engine/src/halt/gilui_arayot.pl",
     "talmudic_engine/src/halt/lo_taaseh_lekha.pl",
+    "talmudic_engine/src/halt/mishkal.pl",
     "talmudic_engine/src/halt/shfichut_damim.pl",
     "talmudic_engine/src/halt/structure.pl",
     "talmudic_engine/src/middot/binyan_av.pl",
@@ -178,7 +179,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 "86a43d6b688780f787d09124fe429a26f96cbe6a7e36460d3b6e5b33f2b2d095"
+  @expected_hash "f90870d476a3a1f31b7cc3a4ff04f8dad1f5db190e16c6956ce19c368366d656"
 
   @doc "Return the list of files that constitute the category-grade specification."
   @spec files() :: [String.t()]

vav_pipeline/lib/vav_pipeline/l3/pair_record.ex

@@ -0,0 +1,36 @@
+defmodule VavPipeline.L3.PairRecord do
+  @moduledoc """
+  L3 schema for a paired reading (bishearecha.md §VI): two bedikot
+  weighed with one stone. Append-only; keeps the counts on each side and
+  the measures that leaned, never the outputs — those are in `bedikot`
+  by hash, under `bedikah_ref_a` and `bedikah_ref_b`.
+  """
+
+  use Ecto.Schema
+  import Ecto.Changeset
+
+  schema "bedikah_pairs" do
+    field :pair_ref, :string
+    field :bedikah_ref_a, :string
+    field :bedikah_ref_b, :string
+    field :provenance_a, :string
+    field :provenance_b, :string
+    field :measures_a, :map, default: %{}
+    field :measures_b, :map, default: %{}
+    field :leans, {:array, :map}, default: []
+    field :violations, {:array, :string}, default: []
+    field :verdict, :string
+    field :weighed_at, :utc_datetime
+    timestamps(type: :utc_datetime, updated_at: false)
+  end
+
+  @required [:pair_ref, :bedikah_ref_a, :bedikah_ref_b, :provenance_a, :provenance_b, :verdict, :weighed_at]
+  @optional [:measures_a, :measures_b, :leans, :violations]
+
+  def changeset(record, attrs) do
+    record
+    |> cast(attrs, @required ++ @optional)
+    |> validate_required(@required)
+    |> validate_inclusion(:verdict, ["even", "uneven"])
+  end
+end

vav_pipeline/lib/vav_pipeline/l3/store.ex

@@ -85,6 +85,16 @@ defmodule VavPipeline.L3.Store do
     |> Repo.insert()
   end
 
+  @doc """
+  Record a paired reading (bishearecha.md §VI): two bedikot weighed with one stone.
+  """
+  @spec record_pair(map()) :: {:ok, struct()} | {:error, Ecto.Changeset.t()}
+  def record_pair(attrs) when is_map(attrs) do
+    %VavPipeline.L3.PairRecord{}
+    |> VavPipeline.L3.PairRecord.changeset(attrs)
+    |> Repo.insert()
+  end
+
   @doc """
   Record an unresolvable tension (teiku) to L3.