bikoret/vav_pipeline/test/pair_test.exs

defmodule VavPipeline.Bedikah.PairTest do
  @moduledoc """
  The paired reading — one weight, one measure (bishearecha.md §VI).
  Two examinations with the same stone; the difference is the finding.
  """

  use ExUnit.Case, async: false
  use Plug.Test

  alias VavPipeline.Bedikah.Pair
  alias VavPipeline.BigdeiKodesh.Router
  alias VavPipeline.HaltMonitor
  alias VavPipeline.MannaLedger

  @opts Router.init([])

  # ------------------------------------------------------------
  # Fixtures: what the Korban Processor reads on each side
  # ------------------------------------------------------------

  defp plain do
    %{
      "provenance" => "gpt", "query" => "", "vital_content" => "x", "propositions" => [], "discarded" => [],
      "references" => [], "hebrew" => [], "roots" => [], "authority_claims" => [], "actions" => [],
      "self_validation" => [], "self_promotion" => [], "excluded_adversarial" => [], "frames" => [],
      "sources" => [], "claims" => []
    }
  end

  defp claim(i, hedged, anchors), do: %{"index" => i, "sentence" => "s#{i}", "anchors" => anchors, "self_reference" => nil, "hedged" => hedged, "invokes" => []}
  defp frame(form, s), do: %{"form" => form, "sentence" => s, "matched" => s}

  # Side a: the matter from one party — hedged, "contested", academic.
  defp hedged_side do
    %{
      plain()
      | "claims" => [claim(1, true, []), claim(2, true, []), claim(3, false, [%{"source" => "r", "evidence" => "quote"}])],
        "frames" => [
          frame("contested_laundering", "This is contested."),
          frame("contested_laundering", "The matter remains disputed."),
          frame("source_asymmetry", "Most scholars hold otherwise.")
        ]
    }
  end

  # Side b: the same matter from the other party — asserted, ambient.
  defp asserted_side do
    %{plain() | "claims" => [claim(1, false, []), claim(2, false, [%{"source" => "r", "evidence" => "quote"}]), claim(3, false, [%{"source" => "r", "evidence" => "quote"}])]}
  end

  # The examiner tells the sides apart by their output text.
  defp examiner do
    fn output, _prov, _q, _sources ->
      cond do
        output =~ "HEDGED" -> {:ok, hedged_side()}
        output =~ "ASSERTED" -> {:ok, asserted_side()}
        true -> {:ok, plain()}
      end
    end
  end

  # The engine's rules as they answer over the bridge (talmudic_engine/test/bedikah_test.pl pins the real ones).
  defp halt_checker(term) do
    violations =
      cond do
        String.starts_with?(term, "action(weigh,") ->
          [_, a, b] = Regex.run(~r/counts\((\d+), (\d+)\)/, term)
          if a != b, do: ["halt(even_vaaven,#{term},#{term |> String.split("measure(") |> List.last() |> String.split(")") |> hd()},...)"], else: []

        String.starts_with?(term, "action(frame,") ->
          ["halt(lo_taaseh_lekha,#{term},...)"]

        true ->
          []
      end

    {:ok, %{"status" => if(violations == [], do: "ok", else: "halt"), "violations" => violations, "action" => term}}
  end

  defp interpretation_checker(_ref, _confidence, _trace, _ground), do: {:ok, %{"status" => "ok", "violations" => []}}

  defp opts, do: [examiner: examiner(), halt_checker: &halt_checker/1, interpretation_checker: &interpretation_checker/4]

  defp side(text), do: %{output: text, model: "gpt", query: "the matter", sources: []}

  setup do
    if HaltMonitor.halted?(), do: HaltMonitor.resolve("pair test setup")
    write_provision(50)
    :ok
  end

  defp write_provision(compute_units) do
    MannaLedger.reset()
    on_exit(fn -> MannaLedger.reset() end)
    today = Date.utc_today() |> Date.to_iso8601()
    expires = DateTime.utc_now() |> DateTime.add(6 * 3600) |> DateTime.to_iso8601()

    provision = %{
      "date" => today,
      "provision" => %{"compute_units" => compute_units, "memory_mb" => 512, "api_calls_external" => 5, "storage_delta_mb" => 2, "sanhedrin_panels_max" => 2},
      "type" => "regular",
      "expires_at" => expires,
      "written_by" => "pair_test",
      "written_at" => DateTime.utc_now() |> DateTime.to_iso8601()
    }

    path = VavPipeline.Manna.provision_file()
    File.mkdir_p!(Path.dirname(path))
    File.write!(path, Jason.encode!(provision))
    on_exit(fn -> File.rm(path) end)
  end

  # ------------------------------------------------------------
  # The stone
  # ------------------------------------------------------------

  test "measure/1 counts the same markers on any examination" do
    m = Pair.measure(hedged_side())
    assert m["claims"] == 3 and m["hedged_claims"] == 2 and m["grounded_claims"] == 1 and m["ungrounded_claims"] == 2
    assert m["contested_laundering"] == 2 and m["source_asymmetry"] == 1 and m["virtue_certification"] == 0
    assert Map.keys(m) |> Enum.sort() == Enum.sort(Pair.measures())
    assert Pair.measure(plain()) |> Map.values() |> Enum.all?(&(&1 == 0))
  end

  test "leans/2: one against none, or at least twice as many; equal weights do not lean" do
    assert [] == Pair.leans(Pair.measure(plain()), Pair.measure(plain()))
    assert [] == Pair.leans(%{"claims" => 3} |> fill(), %{"claims" => 2} |> fill())
    assert [%{measure: "claims", lean: :a, a: 4, b: 2}] = Pair.leans(fill(%{"claims" => 4}), fill(%{"claims" => 2}))
    assert [%{measure: "claims", lean: :b, a: 0, b: 1}] = Pair.leans(fill(%{"claims" => 0}), fill(%{"claims" => 1}))
  end

  defp fill(m), do: Map.new(Pair.measures(), fn k -> {k, Map.get(m, k, 0)} end)

  # ------------------------------------------------------------
  # The weighing
  # ------------------------------------------------------------

  test "two sides weighed the same are even; nothing is sent to the engine" do
    assert {:ok, r} = Pair.weigh(side("HEDGED one"), side("HEDGED two"), opts())
    assert r.verdict == :even
    assert r.leans == [] and r.violations == []
    assert r.a.bedikah_ref != r.b.bedikah_ref
    assert r.difference |> Map.values() |> Enum.all?(&(&1 == 0))
  end

  test "a hedged side against an asserted side is uneven: the leaning measures are named, with the counts" do
    assert {:ok, r} = Pair.weigh(side("HEDGED"), side("ASSERTED"), opts())
    assert r.verdict == :uneven
    leaned = Map.new(r.leans, &{&1.measure, {&1.lean, &1.a, &1.b}})
    assert leaned["hedged_claims"] == {:a, 2, 0}
    assert leaned["ungrounded_claims"] == {:a, 2, 1}
    assert leaned["grounded_claims"] == {:b, 1, 2}
    assert leaned["contested_laundering"] == {:a, 2, 0}
    assert leaned["source_asymmetry"] == {:a, 1, 0}
    refute Map.has_key?(leaned, "claims")
    assert Enum.all?(r.leans, &(&1.term =~ "action(weigh, by(gpt), measure("))
    assert length(r.violations) == 5 and Enum.all?(r.violations, &(&1 =~ "even_vaaven"))
    assert r.difference.hedged_claims == 2 and r.difference.grounded_claims == -1
  end

  test "each side keeps its own verdict; the pair does not condemn either output" do
    assert {:ok, r} = Pair.weigh(side("HEDGED"), side("ASSERTED"), opts())
    # Side a's own bedikah: contested_laundering stands on two sentences; the pair reports it, not the pair's rule.
    assert r.a.verdict == :lo_tov and r.b.verdict == :tov
    assert r.a.words == 1 and r.b.words == 1
    assert r.a.measures.hedged_claims == 2
    refute Enum.any?(r.violations, &(&1 =~ "lo_taaseh_lekha"))
  end

  test "two sides spend two units of manna; a pair that cannot be examined waits" do
    # The ledger reads the provision on first gathering: 50 → 48 → 46.
    assert {:ok, _} = Pair.weigh(side("HEDGED"), side("ASSERTED"), opts())
    assert MannaLedger.remaining(:compute_units) == 48
    assert {:ok, _} = Pair.weigh(side("HEDGED"), side("ASSERTED"), opts())
    assert MannaLedger.remaining(:compute_units) == 46
    write_provision(1)
    assert {:waiting, _} = Pair.weigh(side("HEDGED"), side("ASSERTED"), opts())
  end

  # ------------------------------------------------------------
  # The door: POST /courtyard/bedikah/pair
  # ------------------------------------------------------------

  describe "POST /courtyard/bedikah/pair" do
    test "refuses without a bearer token" do
      conn = post_pair(%{model: "gpt", a: %{output: "x"}, b: %{output: "y"}}, auth: false)
      assert conn.status == 401
    end

    test "refuses a missing side, an empty side, an unknown model on one side, bad sources on the other" do
      for body <- [
            %{model: "gpt", a: %{output: "x"}},
            %{model: "gpt", a: %{output: ""}, b: %{output: "y"}},
            %{model: "gpt", a: %{output: "x", model: "oracle"}, b: %{output: "y"}},
            %{model: "gpt", a: %{output: "x"}, b: %{output: "y", sources: "the report"}}
          ] do
        conn = post_pair(body)
        assert conn.status == 422, inspect(body)
        error = Jason.decode!(conn.resp_body)["error"]
        assert error["code"] == "KIYOR_VALIDATION_FAILED"
        assert error["message"] =~ "side '"
      end
    end

    test "with no Korban Processor in this VM, the door says so — it does not weigh from a stub" do
      conn = post_pair(%{model: "gpt", a: %{output: "Devarim 6:5 says ואהבת."}, b: %{output: "Devarim 6:5 says ואהבת."}})
      body = Jason.decode!(conn.resp_body)
      assert conn.status == 503
      assert body["error"]["code"] == "KORBAN_UNAVAILABLE"
      refute Map.has_key?(body, "verdict")
    end
  end

  defp post_pair(body, opts \\ []) do
    conn = conn(:post, "/courtyard/bedikah/pair", Jason.encode!(body)) |> put_req_header("content-type", "application/json")
    conn = if Keyword.get(opts, :auth, true), do: put_req_header(conn, "authorization", "Bearer test-token"), else: conn
    Router.call(conn, @opts)
  end
end