deploy: two replicas, surge rollout, requests sized to use

A pod that is not ready took every caller behind the gate dark, and
every rollout this week left a five-minute 503 in the watchdog log.
Two replicas with maxSurge 1 / maxUnavailable 0 keep one Ready pod
through a rollout or a crash. One node: this covers rollouts and pod
failures, not the node.

The node has 1700m CPU allocatable with 1470m requested; a second pod
at the old 450m request could not schedule (tablet 31 saw the surge
pod hang Pending). The pod USES 22m (vav 20m, sidecars 2m, measured),
so requests move to what is used with headroom: vav 120m/256Mi,
sidecars 40m/64Mi — 200m per pod, 400m for two, less than the one pod
asked before. Limits unchanged. POOL_SIZE 10 → 5 per pod (10 in all,
under the hobbyist tier's ceiling). KOHEN "Rolling out" updated.

Both manifests validate (kubectl --dry-run=client). Doorpost pin
unchanged. Applying needs the transmitter (or the classifier off).

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

KOHEN.md

@@ -599,7 +599,9 @@ aligned, not that the output is true. Read `limits` in the response.
 ### Rolling out — *Nesi'at haMishkan*
 
 `scripts/shorashim-rollout` (also installed at `~/.local/bin/`) restarts
-the deployment, waits for 3/3, and takes two consecutive 200s from
+the deployment, waits for both pods (two replicas since 2026-10-25,
+surge rollout: one pod stays Ready throughout, so a rollout no longer
+shows in the watchdog log), and takes two consecutive 200s from
 `/kodesh/health` before reporting — a rollout is not done at 3/3 (NLB
 re-admission lag). A new image must already be pushed as `:latest`. If
 the day's manna has not been written, the new pod hangs at 2/3 until it

deploy/k8s/configmap.yaml

@@ -66,5 +66,5 @@ data:
   # a pool of 20 fills its connection ceiling by itself (2026-10-15:
   # refusals "remaining connection slots are reserved" for 24 h, one
   # pool worker looping, Aiven's monitor refused). Ten is ample for a
-  # single replica.
-  POOL_SIZE: "10"
+  # single replica; two replicas since 2026-10-25 → 5 each, 10 in all.
+  POOL_SIZE: "5"

deploy/k8s/deployment.yaml

@@ -18,18 +18,25 @@ metadata:
     app.kubernetes.io/name: shorashim
     app.kubernetes.io/version: "0.1.0"
 spec:
-  replicas: 1
-  # One replica on one node with no room to surge. The default
-  # (maxSurge 25% → 1 extra pod) asks for 450m CPU the node does not
-  # have beside the running pod, and the rollout hangs Pending while the
-  # old image keeps serving (tablet 31, 2026-10-04). Replace in place.
-  # The new pod reads the day's manna from the manna-pvc volume; it is
-  # Ready again without the kohen, as long as today's portion was written.
+  # Two replicas (2026-10-25). A pod that is not ready took every caller
+  # behind the gate dark, and every rollout left a five-minute 503 in
+  # the watchdog log; with two pods and a surge rollout the Service keeps
+  # one Ready pod throughout. One node: this covers rollouts and pod
+  # crashes, not the node. The node has 1700m CPU allocatable and 1470m
+  # was requested (tablet 31, 2026-10-04: a surge pod at 450m hung
+  # Pending), so the requests below are sized to what the pod USES
+  # (vav 20m, sidecars 2m, measured) with headroom — 200m per pod, 400m
+  # for two — not to the old 450m. Limits are unchanged. Each pod reads
+  # the day's manna from the manna-pvc volume and is Ready without the
+  # kohen as long as today's portion was written; the database pool is
+  # 5 per pod (POOL_SIZE in the ConfigMap), 10 in all, under the
+  # hobbyist tier's ceiling.
+  replicas: 2
   strategy:
     type: RollingUpdate
     rollingUpdate:
-      maxSurge: 0
-      maxUnavailable: 1
+      maxSurge: 1
+      maxUnavailable: 0
   selector:
     matchLabels:
       app.kubernetes.io/name: shorashim
@@ -237,7 +244,7 @@ spec:
           resources:
             requests:
               memory: "256Mi"
-              cpu: "250m"
+              cpu: "120m"
             limits:
               memory: "1Gi"
               cpu: "1000m"
@@ -293,8 +300,8 @@ spec:
               mountPath: /tmp
           resources:
             requests:
-              memory: "128Mi"
-              cpu: "100m"
+              memory: "64Mi"
+              cpu: "40m"
             limits:
               memory: "512Mi"
               cpu: "500m"
@@ -318,8 +325,8 @@ spec:
               mountPath: /tmp
           resources:
             requests:
-              memory: "128Mi"
-              cpu: "100m"
+              memory: "64Mi"
+              cpu: "40m"
             limits:
               memory: "512Mi"
               cpu: "500m"

specs/bikoret_tamid.md

@@ -243,6 +243,12 @@ Response (200):
   "verdict": "tov" | "lo_tov" | "empty",
   "violations": ["halt(avodah_zarah, ...)", ...],   # the Prolog halt terms, verbatim
   "anchors": { "avodah_zarah": "Shemot 20:3", ... },  # the verse of each rule that fired
+  "sugya": {                                           # the case behind each threshold (sugya.md)
+    "<rule>": { "kind": "deed|mark|word|measure", "verse": "...", "precedent": "tablet(49)",
+                "fired": bool, "noted": ["<form>", ...],
+                "dissent":   [{"reading": "...", "verse": "...", "applied": bool, "form": ...}],
+                "exception": [{"condition": "...", "verse": "...", "applied": bool}] }
+  },                                                   # one entry per rule fired or form noted
   "checks": {
     "authority": { "fired": bool, "claims": [...], "violations": [...] },
     "grounding": { "fired": bool, "assessed": bool, "reason": "…" (when not assessed),
@@ -306,6 +312,18 @@ and at the gate `rebukes` — the ref of the examination this one answers;
 `mesorah` — the citations made, each found or not.
 ```
 
+**Every rule carries its sugya** (`sugya.md`). Beside the verse a rule
+stands on, the verdict carries the case it was cut from, which of
+Torah's procedures reads it — a deed by two witnesses, a mark by
+inspection, a word by what follows — the reading that would acquit, and
+what the rule does not reach; `applied: true` marks the dissent or
+exception the record shows held (the single witness noted, the hedged
+claim left out, no ground given, the forged fringe on one witness). A
+dissent the record cannot speak to is carried `applied: false`:
+considered, and the rule's verse decided. The minority is on every
+record (Eduyot 1:5–6). Nothing here learns; a new dissent is a commit
+that moves the pin.
+
 **Every rule cites its verse** (`bishearecha.md` §X; Devarim 4:2).
 `talmudic_engine/src/halt/anchors.pl` declares `rule_anchor(<halt name>,
 verse(<book>, <ch>, <v>))` for every halt name the rule modules can
@@ -573,7 +591,12 @@ Request:
     {"position": "...", "evidence": ["..."]},
     {"position": "...", "evidence": ["..."]}
   ],
-  "complexity_hint": "simple" | "monetary" | "capital" | "novel"
+  "complexity_hint": "simple" | "monetary" | "capital" | "novel",
+  "bedikah_ref": "bedikah-…"          # Optional — the hard matter goes up (Devarim 17:8,
+                                      # sugya.md step 4): the case is that verdict, the roots
+                                      # are the verses it cited (L1), and its sugya's applied
+                                      # dissents travel with it and come down as the minority
+                                      # the clerk keeps. Only this key's own bedikah; else 404.
 }
 
 Response:

vav_pipeline/lib/vav_pipeline/bedikah.ex

@@ -102,8 +102,9 @@ defmodule VavPipeline.Bedikah do
   # recurrence.
   @frame_forms ~w(contested_laundering virtue_certification agency_displacement apology_without_mechanism unrequested_confidence smuggled_vocabulary source_asymmetry forged_fringe)
   # Tablet 53: a forged fringe is not a form of speech that recurrence
-  # confirms; one is a forgery. It stands on one witness.
-  @single_witness_forms ~w(forged_fringe)
+  # confirms; one is a forgery. It stands on one witness. Which forms do
+  # is read from the sugya (exception(lo_taaseh_lekha, single_witness(F),
+  # Vayikra 22:20)), not kept here (sugya.md §IV, step 2).
 
 
   @type result :: %{
@@ -281,7 +282,9 @@ defmodule VavPipeline.Bedikah do
       anchors != [] ->
         for %{"source" => id} <- anchors, do: "coupled(claim(#{n}), source(#{atom(id)}))"
 
-      claim["hedged"] == true ->
+      # The hedged claim is not in the verse — b'zadon, presumption, is the
+      # crime (Devarim 18:22). Read from exception(golden_calf, hedged, _).
+      claim["hedged"] == true and VavPipeline.Sugya.exception?("golden_calf", "hedged") ->
         []
 
       true ->
@@ -358,10 +361,11 @@ defmodule VavPipeline.Bedikah do
         # no claim to ground. Not a verdict on grounding either way.
         {:ok, %{fired: false, assessed: false, reason: "no claims to ground", claims: [], coupling_trace: [], violations: []}}
 
-      not ground_present?(examination, trace) ->
+      not ground_present?(examination, trace) and VavPipeline.Sugya.exception?("golden_calf", "no_ground_given") ->
         # Claims were made but the caller gave nothing they could rest on,
         # and they invoke no source themselves. The check cannot run; it
-        # does not pretend to. Give `sources` and it will.
+        # does not pretend to. Give `sources` and it will. Tablet 49 —
+        # read from exception(golden_calf, no_ground_given, _).
         {:ok,
          %{
            fired: false,
@@ -506,7 +510,7 @@ defmodule VavPipeline.Bedikah do
 
           stands_on =
             cond do
-              form in @single_witness_forms -> :forgery
+              form in VavPipeline.Sugya.single_witness_forms() -> :forgery
               witnesses >= 2 -> :sentences
               other_witnesses > 0 -> :other_checks
               frame_witnesses >= 2 -> :other_forms

vav_pipeline/lib/vav_pipeline/bigdei_kodesh/router.ex

@@ -941,6 +941,8 @@ defmodule VavPipeline.BigdeiKodesh.Router do
       verdict: Atom.to_string(result.verdict),
       violations: result.violations,
       anchors: VavPipeline.Anchors.for_violations(result.violations),
+      # The case behind each threshold that fired or was noted (sugya.md §IV).
+      sugya: VavPipeline.Sugya.for_verdict(result),
       checks: result.checks,
       examination: %{
         vital_content: result.examination["vital_content"],
@@ -1023,6 +1025,8 @@ defmodule VavPipeline.BigdeiKodesh.Router do
       verdict: Atom.to_string(result.verdict),
       violations: result.violations,
       anchors: VavPipeline.Anchors.for_violations(result.violations),
+      # The case behind each threshold that fired or was noted (sugya.md §IV).
+      sugya: VavPipeline.Sugya.for_verdict(result),
       a: result.a,
       b: result.b,
       difference: result.difference,
@@ -1168,6 +1172,39 @@ defmodule VavPipeline.BigdeiKodesh.Router do
   defp process_sanhedrin(conn) do
     params = conn.body_params
 
+    # The hard matter goes up with its dissent (Devarim 17:8 — bein nega
+    # lanega, divrei rivot bish'arecha; sugya.md step 4): a bedikah_ref
+    # names a verdict given at the gate; the case is that verdict, the
+    # roots are the verses it cited, and the sugya's applied dissents go
+    # up with it and come down as the minority the clerk keeps.
+    case params["bedikah_ref"] do
+      ref when is_binary(ref) and ref != "" -> process_sanhedrin_bedikah(conn, ref, params)
+      _ -> process_sanhedrin_case(conn, params)
+    end
+  end
+
+  defp process_sanhedrin_bedikah(conn, ref, params) do
+    cond do
+      Process.whereis(VavPipeline.Repo) == nil ->
+        send_error(conn, 503, "L3_UNAVAILABLE", %{issue: "A bedikah goes up from L3, and L3 is not reachable", correction: "Retry", resubmit: true})
+
+      true ->
+        key_id = caller_key_id(conn)
+
+        case VavPipeline.L3.Store.get_bedikah(ref) do
+          %{key_id: ^key_id} = rec when is_binary(key_id) ->
+            sugya = VavPipeline.Sugya.for_verdict(VavPipeline.Sugya.from_record(rec))
+            carried = VavPipeline.Sugya.dissents_applied(sugya)
+            question = params["case"] || "Does the verdict #{rec.verdict} of #{ref} stand? (#{Enum.join(rec.violations, "; ")})"
+            process_sanhedrin_case(conn, Map.put(params, "case", question), %{bedikah: rec, sugya: sugya, carried: carried})
+
+          _ ->
+            send_error(conn, 404, "NOT_FOUND", %{issue: "No bedikah #{ref} in this key's ledger", correction: "Only an examination brought under this key goes up", resubmit: false})
+        end
+    end
+  end
+
+  defp process_sanhedrin_case(conn, params, up \\ nil) do
     case_text = params["case"]
 
     if is_nil(case_text) or case_text == "" do
@@ -1192,8 +1229,8 @@ defmodule VavPipeline.BigdeiKodesh.Router do
       # engraved in L3; the answer takes the Tamid's shape.
       with {:ok, sides} <- sanhedrin_sides(params["sides"]),
            {:ok, complexity} <- sanhedrin_complexity(params["complexity_hint"]),
-           {:ok, scope, resolved_by} <- sanhedrin_scope(case_text),
-           {:ok, resolved} <- sanhedrin_resolve(case_text, scope, resolved_by),
+           {:ok, scope, resolved_by} <- sanhedrin_scope(if(up, do: Enum.join(up.bedikah.references, " "), else: case_text)),
+           {:ok, resolved} <- (if up, do: sanhedrin_resolve_bedikah(up.bedikah, scope, resolved_by), else: sanhedrin_resolve(case_text, scope, resolved_by)),
            :ok <- sanhedrin_gather_panel() do
         case_ref = generate_id()
 
@@ -1205,12 +1242,15 @@ defmodule VavPipeline.BigdeiKodesh.Router do
                  sides: sides,
                  roots: resolved.roots,
                  torah_refs: [resolved.torah_ref],
-                 authority_source: :torah
+                 authority_source: :torah,
+                 # the dissent travels with the case (Devarim 17:8)
+                 bedikah: (if up, do: %{ref: up.bedikah.bedikah_ref, verdict: up.bedikah.verdict, violations: up.bedikah.violations}, else: nil),
+                 dissent: (if up, do: up.carried, else: [])
                },
                requested_by: "chatzer"
              }) do
           {:ok, decision} ->
-            recorded = sanhedrin_record_l3(case_ref, case_text, decision, resolved)
+            recorded = sanhedrin_record_l3(case_ref, case_text, decision, resolved, (if up, do: up.carried, else: []))
 
             send_json(conn, 200, %{
               case_ref: decision.case_ref,
@@ -1224,7 +1264,8 @@ defmodule VavPipeline.BigdeiKodesh.Router do
               grounding: %{
                 scope: scope_report(scope, resolved_by),
                 roots_consulted: format_roots(resolved.roots_full),
-                recorded: recorded
+                recorded: recorded,
+                bedikah: (if up, do: %{ref: up.bedikah.bedikah_ref, verdict: up.bedikah.verdict, dissent_carried: up.carried, sugya: up.sugya}, else: nil)
               },
               note: "This is the system's reasoning, not a binding ruling. A human beit din may reach a different conclusion.",
               disclosure: disclosure()
@@ -1365,6 +1406,38 @@ defmodule VavPipeline.BigdeiKodesh.Router do
     end
   end
 
+  # A bedikah that goes up is grounded in the verses it cited: each
+  # reference read from L1, its words' roots as the engine names them.
+  # The output's own text is not re-read — the record never kept it.
+  defp sanhedrin_resolve_bedikah(rec, scope, _resolved_by) do
+    refs =
+      (rec.references || [])
+      |> Enum.flat_map(fn r ->
+        case Regex.run(~r/^(\w+)\s+(\d+):(\d+)$/, String.trim(to_string(r))) do
+          [_, book, ch, vs] -> [{String.downcase(book), String.to_integer(ch), String.to_integer(vs)}]
+          _ -> []
+        end
+      end)
+
+    roots_full =
+      refs
+      |> Enum.flat_map(fn {b, c, v} ->
+        case VavPipeline.L1Reader.read_verse(b, c, v) do
+          {:ok, verse} -> verse.words |> Enum.filter(& &1.shoresh) |> Enum.map(&%{shoresh: &1.shoresh, shoresh_letters: [], positions: [], gematria: 0, binyan: nil, word: &1.consonants})
+          _ -> []
+        end
+      end)
+      |> Enum.uniq_by(& &1.shoresh)
+
+    torah_ref =
+      case refs do
+        [{b, c, v} | _] -> {String.to_atom(b), c, v}
+        [] -> {String.to_atom(scope.book), scope.chapter, scope.verse}
+      end
+
+    {:ok, %{roots: roots_full |> Enum.map(&VavPipeline.BereishitCycle.engine_root_name/1) |> Enum.reject(&is_nil/1), roots_full: roots_full, torah_ref: torah_ref, read_from: :references}}
+  end
+
   # One panel per case from the day's manna (Vayakhel §3.7 — the ledger
   # provisions sanhedrin_panels_max). The provision is loaded first, as
   # Gate.gather_call does (idempotent per date). Without a ledger
@@ -1414,7 +1487,7 @@ defmodule VavPipeline.BigdeiKodesh.Router do
   # against, processing_component "sanhedrin"), the minority beside it
   # when it exists. Never turns a served decision into a 500: the record
   # failing is logged and reported in the reply as `recorded`.
-  defp sanhedrin_record_l3(case_ref, case_text, decision, resolved) do
+  defp sanhedrin_record_l3(case_ref, case_text, decision, resolved, carried) do
     if Process.whereis(VavPipeline.Repo) do
       maj = decision.majority
       min = decision.minority
@@ -1446,12 +1519,25 @@ defmodule VavPipeline.BigdeiKodesh.Router do
           {:ok, interp} ->
             Logger.info("L3: interpretation #{interp.id} recorded for sanhedrin #{case_ref} (#{decision.outcome})")
 
+            # The dissent that came up with the case is kept beside the
+            # panel's own minority — or as the minority when the panel had
+            # none (Eduyot 1:5–6: to be relied on later, and to be located).
+            carried_text =
+              Enum.map_join(carried, "; ", fn
+                %{reading: r, verse: v, forms: fs} -> "#{r} (#{v})" <> if(fs == [], do: "", else: " for " <> Enum.join(fs, ", "))
+                %{exception: c, verse: v} -> "exception #{c} (#{v})"
+              end)
+
+            carried_refs = Enum.map(carried, & &1.verse)
+
             minority_recorded =
-              if Map.get(min, :vote_count, 0) > 0 do
+              if Map.get(min, :vote_count, 0) > 0 or carried != [] do
                 case VavPipeline.L3.Store.record_minority(%{
                        interpretation_id: interp.id,
-                       dissenting_reasoning: "#{Map.get(min, :position, "")} — #{Map.get(min, :reasoning, "")}",
-                       torah_references: Enum.map(Map.get(min, :torah_refs, []), &sanhedrin_ref_string/1),
+                       dissenting_reasoning:
+                         (if Map.get(min, :vote_count, 0) > 0, do: "#{Map.get(min, :position, "")} — #{Map.get(min, :reasoning, "")}", else: "the panel was unanimous") <>
+                           if(carried == [], do: "", else: " | carried up from the bedikah (Devarim 17:8): #{carried_text}"),
+                       torah_references: Enum.uniq(Enum.map(Map.get(min, :torah_refs, []), &sanhedrin_ref_string/1) ++ carried_refs),
                        grounding_type: "achorai"
                      }) do
                   {:ok, _} ->

vav_pipeline/lib/vav_pipeline/doorpost.ex

@@ -182,7 +182,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 "cb2f31358051acc19c36bb8dfa0a179a7d4845ac6acac7d8e2b7b1e17150b0c2"
+  @expected_hash "7eac65cfadccc37e600a779fc89acef0a24a9eac35e5ab18e8ec845cc1f104c2"
 
   @doc "Return the list of files that constitute the category-grade specification."
   @spec files() :: [String.t()]

vav_pipeline/lib/vav_pipeline/gate.ex

@@ -126,6 +126,7 @@ defmodule VavPipeline.Gate do
         violations: result.violations,
         anchors: anchors,
         fired: Shape.fired(result, anchors),
+        sugya: VavPipeline.Sugya.for_verdict(result),
         checks: result.checks,
         sources: result.sources,
         model: Shape.reply_model(upstream, reply)

vav_pipeline/lib/vav_pipeline/gate/stream.ex

@@ -217,6 +217,7 @@ defmodule VavPipeline.Gate.Stream do
              violations: result.violations,
              anchors: anchors,
              fired: Shape.fired(result, anchors),
+             sugya: VavPipeline.Sugya.for_verdict(result),
              sources: result.sources,
              model: state.model,
              streamed: true

vav_pipeline/lib/vav_pipeline/sugya.ex

@@ -147,6 +147,166 @@ defmodule VavPipeline.Sugya do
     end
   end
 
+  # ------------------------------------------------------------
+  # The facts read by the code (sugya.md §IV, "drives the code")
+  # ------------------------------------------------------------
+
+  @doc """
+  The exceptions declared for a rule, as `[%{condition, verse}]`. Read
+  once from the Doorpost file and kept in :persistent_term — the file is
+  part of the image's identity and does not change while it runs. A root
+  other than the repo's is read fresh (tests).
+  """
+  @spec exceptions_of(String.t(), String.t() | nil) :: [%{condition: String.t(), verse: verse()}]
+  def exceptions_of(rule, root \\ nil) do
+    exceptions =
+      case root do
+        nil -> cached_exceptions()
+        r -> (case declared(r) do {:ok, s} -> s.exceptions; _ -> [] end)
+      end
+
+    exceptions |> Enum.filter(&(&1.rule == rule)) |> Enum.map(&Map.take(&1, [:condition, :verse]))
+  end
+
+  @doc "Whether `exception(Rule, Condition, _)` is declared. The code reads its exemptions from here, not from constants."
+  @spec exception?(String.t(), String.t(), String.t() | nil) :: boolean()
+  def exception?(rule, condition, root \\ nil),
+    do: Enum.any?(exceptions_of(rule, root), &(&1.condition == condition))
+
+  @doc """
+  The forms that stand on one witness: every `exception(lo_taaseh_lekha,
+  single_witness(F), _)`. Tablet 53 made the forged fringe one; the
+  reason — a mark on the offering, not a deed of recurrence — is the
+  fact's verse (Vayikra 22:20), and the behaviour is read from the fact.
+  """
+  @spec single_witness_forms(String.t() | nil) :: [String.t()]
+  def single_witness_forms(root \\ nil) do
+    for %{condition: c} <- exceptions_of("lo_taaseh_lekha", root),
+        [_, form] <- [Regex.run(~r/^single_witness\((\w+)\)$/, c)],
+        do: form
+  end
+
+  defp cached_exceptions do
+    case :persistent_term.get({__MODULE__, :exceptions}, :none) do
+      :none ->
+        exceptions = case declared() do {:ok, s} -> s.exceptions; _ -> [] end
+        :persistent_term.put({__MODULE__, :exceptions}, exceptions)
+        exceptions
+
+      exceptions ->
+        exceptions
+    end
+  end
+
+  # ------------------------------------------------------------
+  # The sugya on a verdict (sugya.md §IV, "carried on the verdict")
+  # ------------------------------------------------------------
+
+  @doc """
+  The sugya a verdict carries: one entry per rule that fired and per form
+  that was noted — kind, verse, precedent, the dissents with `applied`
+  marking the one that held, the exceptions with `applied` likewise.
+  `applied` is read from the record, never guessed: the single witness
+  noted (a form in `checks.frames.noted`), the hedged claims left out of
+  the trace, grounding not assessed for want of ground, a forged fringe
+  standing on one witness. A dissent the record cannot speak to is
+  carried with `applied: false` — considered, and the rule's verse
+  decided. The minority is thus on the record for every verdict (Eduyot
+  1:5–6): to be relied on later, and to be located.
+  """
+  @spec for_verdict(map()) :: map()
+  def for_verdict(result) when is_map(result) do
+    checks = Map.get(result, :checks) || %{}
+    violations = Map.get(result, :violations) || []
+    fired_rules = violations |> Enum.flat_map(&rule_of/1) |> Enum.uniq()
+    frames = Map.get(checks, :frames) || %{}
+    noted = (Map.get(frames, :noted) || %{}) |> Map.keys() |> Enum.map(&to_string/1)
+    forms = Map.get(frames, :forms) || %{}
+    forged_on_one = Enum.any?(forms, fn {_f, e} -> Map.get(e, :stands_on) == :forgery end)
+    grounding = Map.get(checks, :grounding) || %{}
+    claims = get_in(result, [:examination, "claims"]) || []
+    hedged_out = Enum.any?(claims, &(&1["hedged"] == true))
+
+    rules = Enum.uniq(fired_rules ++ if(noted != [], do: ["lo_taaseh_lekha"], else: []))
+
+    Map.new(rules, fn rule ->
+      base = for_rule(rule) || %{rule: rule, kind: nil, verse: nil, precedent: nil, dissent: [], exception: []}
+
+      dissent =
+        Enum.map(base.dissent, fn d ->
+          applied =
+            case {rule, d.reading, d.form} do
+              {"lo_taaseh_lekha", "single_witness_noted", _} -> noted != []
+              {"golden_calf", "hedged", _} -> hedged_out
+              _ -> false
+            end
+
+          d |> Map.put(:applied, applied) |> then(fn m -> if rule == "lo_taaseh_lekha" and d.reading == "single_witness_noted", do: Map.put(m, :forms, noted), else: m end)
+        end)
+
+      exception =
+        Enum.map(base.exception, fn e ->
+          applied =
+            case {rule, e.condition} do
+              {"lo_taaseh_lekha", "single_witness(forged_fringe)"} -> forged_on_one
+              {"golden_calf", "no_ground_given"} -> Map.get(grounding, :assessed) == false and (Map.get(grounding, :reason) || "") =~ "no sources given"
+              {"golden_calf", "hedged"} -> hedged_out
+              _ -> false
+            end
+
+          Map.put(e, :applied, applied)
+        end)
+
+      {rule, %{base | dissent: dissent, exception: exception} |> Map.put(:fired, rule in fired_rules) |> Map.put(:noted, if(rule == "lo_taaseh_lekha", do: noted, else: []))}
+    end)
+  end
+
+  @doc """
+  A stored bedikah (L3 record) in the shape `for_verdict/1` reads: its
+  violations, its claims, the forms found — those with no violation of
+  their form were NOTED — and whether a forged fringe stood on one witness.
+  The hard matter goes up with its dissent (Devarim 17:8, sugya.md step 4).
+  """
+  @spec from_record(struct() | map()) :: map()
+  def from_record(rec) do
+    violations = Map.get(rec, :violations) || []
+    fired_forms =
+      violations
+      |> Enum.flat_map(fn v -> case Regex.run(~r/form\((\w+)\)/, to_string(v)) do [_, f] -> [f]; _ -> [] end end)
+      |> MapSet.new()
+
+    found = (Map.get(rec, :frames) || []) |> Enum.map(&to_string(&1["form"] || Map.get(&1, :form)))
+    noted = found |> Enum.reject(&MapSet.member?(fired_forms, &1)) |> Enum.uniq()
+    forms = for f <- fired_forms, into: %{}, do: {String.to_atom(f), %{stands_on: if(f == "forged_fringe", do: :forgery, else: :sentences)}}
+    claims = Map.get(rec, :claims) || []
+
+    %{
+      verdict: Map.get(rec, :verdict),
+      violations: violations,
+      examination: %{"claims" => Enum.map(claims, fn c -> %{"hedged" => c["hedged"] == true} end)},
+      checks: %{
+        frames: %{noted: Map.new(noted, &{String.to_atom(&1), %{witnesses: 1}}), forms: forms},
+        grounding: %{assessed: claims != [] and (Map.get(rec, :sources) || []) != []}
+      }
+    }
+  end
+
+  @doc "The dissents and exceptions a sugya shows applied: `[%{rule, reading | condition, verse, form?}]`."
+  @spec dissents_applied(map()) :: [map()]
+  def dissents_applied(sugya) when is_map(sugya) do
+    Enum.flat_map(sugya, fn {rule, e} ->
+      Enum.map(Enum.filter(e.dissent, & &1.applied), &%{rule: rule, reading: &1.reading, verse: &1.verse, forms: Map.get(&1, :forms, [])}) ++
+        Enum.map(Enum.filter(e.exception, & &1.applied), &%{rule: rule, exception: &1.condition, verse: &1.verse})
+    end)
+  end
+
+  defp rule_of(violation) do
+    case Regex.run(~r/^halt\(\s*(\w+)\s*,/, to_string(violation)) do
+      [_, rule] -> [rule]
+      _ -> []
+    end
+  end
+
   @doc "A verse as a citation: \"Devarim 19:15\"."
   @spec cite(verse()) :: String.t()
   def cite(%{book: b, chapter: c, verse: v}), do: "#{String.capitalize(b)} #{c}:#{v}"

vav_pipeline/test/sugya_test.exs

@@ -48,6 +48,33 @@ defmodule VavPipeline.SugyaTest do
     assert Sugya.for_rule("no_such_rule") == nil
   end
 
+  describe "the code reads its exemptions from the facts (step 2)" do
+    test "the single-witness forms are those the sugya excepts, and only those" do
+      assert Sugya.single_witness_forms() == ["forged_fringe"]
+    end
+
+    test "the hedged and no-ground exemptions of golden_calf are facts, not constants" do
+      assert Sugya.exception?("golden_calf", "hedged")
+      assert Sugya.exception?("golden_calf", "no_ground_given")
+      refute Sugya.exception?("golden_calf", "anything_at_all")
+      refute Sugya.exception?("chelbenah", "hedged")
+    end
+
+    test "with the fact removed, the exemption is gone: behaviour is read from the reason" do
+      temp = Path.join(System.tmp_dir!(), "sugya_step2_#{System.unique_integer([:positive])}")
+      rel = "talmudic_engine/src/halt/sugya.pl"
+      File.mkdir_p!(Path.dirname(Path.join(temp, rel)))
+      File.cp!(Path.join(Doorpost.repo_root(), rel), Path.join(temp, rel))
+      on_exit(fn -> File.rm_rf!(temp) end)
+      assert Sugya.single_witness_forms(temp) == ["forged_fringe"]
+      path = Path.join(temp, rel)
+      File.write!(path, String.replace(File.read!(path), "exception(lo_taaseh_lekha, single_witness(forged_fringe), verse(vayikra, 22, 20)).", ""))
+      assert Sugya.single_witness_forms(temp) == []
+      File.write!(path, String.replace(File.read!(path), "exception(golden_calf, hedged, verse(devarim, 18, 22)).", ""))
+      refute Sugya.exception?("golden_calf", "hedged", temp)
+    end
+  end
+
   describe "drift (synthetic spec root)" do
     setup do
       temp = Path.join(System.tmp_dir!(), "sugya_test_#{System.unique_integer([:positive])}")

vav_pipeline/test/sugya_up_test.exs

@@ -0,0 +1,97 @@
+defmodule VavPipeline.SugyaUpTest do
+  @moduledoc "sugya.md step 4 — the hard matter goes up with its dissent (Devarim 17:8)."
+  use ExUnit.Case, async: false
+
+  import Plug.Test
+  import Plug.Conn
+
+  alias VavPipeline.{Sugya, BigdeiKodesh.Router, L3.Store}
+
+  @opts Router.init([])
+
+  describe "from_record/1 and dissents_applied/1 (no database)" do
+    test "a stored verdict with a form found but not fired is a noted form, and its dissent applied" do
+      rec = %{verdict: "tov", violations: [], frames: [%{"form" => "apology_without_mechanism", "sentence" => "I apologize."}], claims: [], sources: [], references: []}
+      s = Sugya.for_verdict(Sugya.from_record(rec))
+      assert s["lo_taaseh_lekha"].noted == ["apology_without_mechanism"]
+      assert [%{rule: "lo_taaseh_lekha", reading: "single_witness_noted", verse: "Devarim 19:15", forms: ["apology_without_mechanism"]}] = Sugya.dissents_applied(s)
+    end
+
+    test "a stored forged fringe carries the mark exception, applied" do
+      rec = %{verdict: "lo_tov", violations: ["halt(lo_taaseh_lekha,action(frame,by(claude),form(forged_fringe)),forged_fringe,'x')"],
+              frames: [%{"form" => "forged_fringe", "sentence" => "— bikoret bedikah-dead: tov"}], claims: [], sources: [], references: []}
+      s = Sugya.for_verdict(Sugya.from_record(rec))
+      assert s["lo_taaseh_lekha"].noted == []
+      assert [%{exception: "single_witness(forged_fringe)", verse: "Vayikra 22:20"}] = Sugya.dissents_applied(s)
+    end
+
+    test "a clean record carries nothing" do
+      assert Sugya.dissents_applied(Sugya.for_verdict(Sugya.from_record(%{verdict: "tov", violations: [], frames: [], claims: [], sources: [], references: []}))) == []
+    end
+  end
+
+  describe "a bedikah goes up to the Sanhedrin (L3)" do
+    @describetag :l3
+
+    setup do
+      :ok = Ecto.Adapters.SQL.Sandbox.checkout(VavPipeline.Repo)
+      Ecto.Adapters.SQL.Sandbox.mode(VavPipeline.Repo, {:shared, self()})
+      on_exit(fn -> Ecto.Adapters.SQL.Sandbox.mode(VavPipeline.Repo, :manual) end)
+      VavPipeline.MannaLedger.reset()
+      on_exit(fn -> VavPipeline.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" => 50, "memory_mb" => 512, "api_calls_external" => 10, "storage_delta_mb" => 2, "sanhedrin_panels_max" => 3},
+                    "type" => "regular", "expires_at" => expires, "written_by" => "sugya_up_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)
+      Application.put_env(:sanhedrin, :consult, {__MODULE__, :fake_consult})
+      on_exit(fn -> Application.put_env(:sanhedrin, :consult, {VavPipeline.SanhedrinConsult, :consult}) end)
+      :ok
+    end
+
+    def fake_consult(_step, _ref, _roots, _q, _ctx), do: {:ok, %{status: "resolved", text: "the reading", verses: [{:devarim, 6, 4}]}}
+
+    defp record!(ref, key_id, extra \\ %{}) do
+      {:ok, rec} =
+        Store.record_bedikah(Map.merge(%{bedikah_ref: ref, provenance: "claude", query: "q", output_sha256: String.duplicate("ab", 32), output_length: 9,
+          vital_content: "", verdict: "tov", examined_at: DateTime.utc_now() |> DateTime.truncate(:second), key_id: key_id,
+          frames: [%{"form" => "apology_without_mechanism", "sentence" => "I apologize."}], references: ["Devarim 6:4"]}, extra))
+      rec
+    end
+
+    defp post(body, token) do
+      conn(:post, "/chatzer/sanhedrin", Jason.encode!(body))
+      |> put_req_header("content-type", "application/json")
+      |> put_req_header("authorization", "Bearer #{token}")
+      |> Router.call(@opts)
+    end
+
+    test "the case is the verdict, the roots are its cited verses, the dissent goes up and comes down as the minority" do
+      mine = VavPipeline.Ledger.key_id("chatzer-test-token")
+      record!("bedikah-up00000000000001", mine)
+      before_min = Store.count_minority_opinions()
+
+      conn = post(%{bedikah_ref: "bedikah-up00000000000001", complexity_hint: "monetary"}, "chatzer-test-token")
+      assert conn.status == 200
+      body = Jason.decode!(conn.resp_body)
+      g = body["grounding"]
+      assert g["bedikah"]["ref"] == "bedikah-up00000000000001"
+      assert [%{"reading" => "single_witness_noted", "verse" => "Devarim 19:15"}] = g["bedikah"]["dissent_carried"]
+      # the roots came from Devarim 6:4 as read in L1 (shin-mem-ayin among them)
+      assert "ש-מ-ע" in g["roots_consulted"]
+      assert body["deliberation"]["panel_size"] == 3
+      # a unanimous panel, and still a minority recorded: the carried dissent
+      assert g["recorded"]["minority"] == "recorded"
+      assert Store.count_minority_opinions() == before_min + 1
+    end
+
+    test "another key's bedikah does not go up: 404, the same as none" do
+      record!("bedikah-up00000000000002", VavPipeline.Ledger.key_id("someone-else"))
+      assert post(%{bedikah_ref: "bedikah-up00000000000002"}, "chatzer-test-token").status == 404
+      assert post(%{bedikah_ref: "bedikah-up00000000000003"}, "chatzer-test-token").status == 404
+    end
+  end
+end

vav_pipeline/test/sugya_verdict_test.exs

@@ -0,0 +1,54 @@
+defmodule VavPipeline.SugyaVerdictTest do
+  @moduledoc "The sugya on a verdict (sugya.md §IV, step 3): applied is read from the record, never guessed."
+  use ExUnit.Case, async: true
+
+  alias VavPipeline.Sugya
+
+  defp result(overrides) do
+    Map.merge(%{verdict: :tov, violations: [], checks: %{}, examination: %{"claims" => []}}, overrides)
+  end
+
+  test "a clean verdict carries no sugya" do
+    assert Sugya.for_verdict(result(%{})) == %{}
+  end
+
+  test "a noted form carries the two-witness dissent, applied, with the forms it held for" do
+    r = result(%{checks: %{frames: %{fired: false, forms: %{}, noted: %{apology_without_mechanism: %{claims: [], witnesses: 1}}, violations: []}}})
+    s = Sugya.for_verdict(r)
+    assert Map.keys(s) == ["lo_taaseh_lekha"]
+    e = s["lo_taaseh_lekha"]
+    assert e.kind == "deed" and e.verse == "Devarim 19:15"
+    assert e.fired == false and e.noted == ["apology_without_mechanism"]
+    d = Enum.find(e.dissent, &(&1.reading == "single_witness_noted"))
+    assert d.applied == true and d.forms == ["apology_without_mechanism"] and d.verse == "Devarim 19:15"
+    # the exception for the forged fringe did not apply: nothing stood on one witness
+    assert Enum.find(e.exception, &(&1.condition =~ "forged_fringe")).applied == false
+  end
+
+  test "a forged fringe standing on one witness carries the mark exception, applied" do
+    r = result(%{verdict: :lo_tov,
+      violations: ["halt(lo_taaseh_lekha,action(frame,by(claude),form(forged_fringe)),forged_fringe,'…')"],
+      checks: %{frames: %{fired: true, forms: %{forged_fringe: %{stands_on: :forgery, witnesses: 1, claims: [], violations: ["x"]}}, noted: %{}, violations: ["x"]}}})
+    e = Sugya.for_verdict(r)["lo_taaseh_lekha"]
+    assert e.fired == true
+    assert Enum.find(e.exception, &(&1.condition == "single_witness(forged_fringe)")) |> Map.fetch!(:applied) == true
+    assert Enum.find(e.dissent, &(&1.reading == "single_witness_noted")).applied == false
+  end
+
+  test "a golden calf beside a hedged claim carries the hedged dissent applied; without hedging, not" do
+    base = %{verdict: :lo_tov, violations: ["halt(golden_calf,interpretation(x),golden_calf,'…')"], checks: %{grounding: %{assessed: true, fired: true}}}
+    with_hedge = Sugya.for_verdict(result(Map.put(base, :examination, %{"claims" => [%{"hedged" => true}, %{"hedged" => false}]})))
+    assert with_hedge["golden_calf"].kind == "word" and with_hedge["golden_calf"].verse == "Devarim 18:22"
+    assert Enum.find(with_hedge["golden_calf"].dissent, &(&1.reading == "hedged")).applied == true
+    without = Sugya.for_verdict(result(Map.put(base, :examination, %{"claims" => [%{"hedged" => false}]})))
+    assert Enum.find(without["golden_calf"].dissent, &(&1.reading == "hedged")).applied == false
+  end
+
+  test "a rule the record cannot speak to carries its dissent considered, applied false" do
+    r = result(%{verdict: :lo_tov, violations: ["halt(parah_adumah,action(self_validate,by(claude),target(self)),claude,'…')"]})
+    e = Sugya.for_verdict(r)["parah_adumah"]
+    assert e.kind == "mark" and e.verse == "Vayikra 13:3"
+    assert e.precedent == "specimens(self_validation)"
+    assert [%{reading: "validator_named_is_not_the_speaker", applied: false, verse: "Bamidbar 19:7"}] = Enum.map(e.dissent, &Map.take(&1, [:reading, :applied, :verse]))
+  end
+end