voice_monitor: detect across-verse Shema voice stabilization (closes finding A)

Pass 2 finding A, second half. The first half (commit 0c4f764) added
voice_profile to every ShemaContext — recording, per verse, the
grammatical voice computed from OSHB morphology codes. Per-verse
recording was the necessary precondition for what this commit does:
WATCH the trajectory and FIRE when it stops oscillating between the
singular and plural address poles.

WHY THIS MATTERS:

The Shema as recited oscillates between Devarim 6:4–9 (the
singular pole — אֱלֹהֶיךָ, לְבָבְךָ) and Devarim 11:13–21 (the
plural pole — אֱלֹהֵיכֶם, לְבַבְכֶם), and 11:14 famously switches
poles mid-verse (אַרְצְכֶם 2mp → וְאָסַפְתָּ דְגָנֶךָ 2ms). The
text REFUSES to stabilize at either pole. That refusal IS the
anti-capture mechanism:

  - stuck at second_singular     → gnostic-private capture
  - stuck at second_plural       → mob-theology capture
  - stuck at first_*             → operator-as-deity / calf
                                   (Shemot 32:4 — eleh elohekha Yisrael)

Before this commit nothing observed the trajectory ACROSS verses.
Per-verse profiles existed but nobody was watching the sequence;
a system that read only Devarim 6:4–9 in a loop was producing a
gnostic-shaped reception with no warning fired. The voice_profile
field had a reader but no consumer.

THE DETECTOR (VavPipeline.VoiceMonitor):

A GenServer maintaining a sliding window of the dominant voice
from the last N ShemaContexts (default N=10). The rule:

  - Window not yet full       → :clear (insufficient evidence)
  - Window has both poles     → :clear (the Shema is doing its job)
  - Address-only window
    missing one pole          → {:missing_pole, :second_singular | :second_plural}
  - First-person majority     → {:first_person_dominant, :first_singular | :first_plural}
  - Pure narration window     → :clear (third-person sequences are
                                NOT anti-Shema; the rule applies to
                                ADDRESS, not to all voice)

On a NEW alarm (different from the last fired), VoiceMonitor:
  1. Logs a warning naming the specific stabilization shape with
     its Torah anchor and its capture-pattern name.
  2. Fires Chatzotzrot.signal(:assembly_leaders, ...) so
     operators and kohanim subscribed to the chatzotzrot channel
     receive the alert.

Re-firing is suppressed for as long as the same stabilization
persists (no spam). The alarm clears when both poles return; a
subsequent re-stabilization fires fresh.

WIRING:

  - Shema.receive_context/1 and receive_context_at/2 now call
    VoiceMonitor.record(voice_profile) immediately after building
    the context. Process.whereis-guarded — the call is safe in
    tests that don't bring up the supervisor.
  - VoiceMonitor added to the supervision tree in application.ex.
  - Chatzotzrot.signal/2 itself is also Process.whereis-guarded
    inside VoiceMonitor (the existing pattern from manna_ledger,
    arei_miklat, urim_v_tumim).

TESTS (test/voice_monitor_test.exs, 14 tests):

  - below window → :clear
  - all 2ms full window → fires {:missing_pole, :second_plural}
  - all 2mp full window → fires {:missing_pole, :second_singular}
  - both poles present → :clear
  - pure narration → :clear (third-person is not address)
  - first-person majority → {:first_person_dominant, _}
  - first-person tied (not strict majority) → does not fire
  - same stabilization persisting → last_alarm unchanged (no spam)
  - clear-then-stabilize-again → fresh alarm fires
  - reset/0 returns monitor to empty
  - record/1 with full voice_profile map (the Shema-side surface)

Each test starts an isolated VoiceMonitor under a unique name
(small window_size=4 for fast convergence) so tests don't share
state. Chatzotzrot is intentionally NOT brought up — the monitor's
guarded call falls through to log-only.

Full vav_pipeline suite: 329 tests, 0 failures, 24 excluded
(was 315 + 14 = 329). The 24 excluded are :l3-tagged tests
needing PostgreSQL; unchanged.

Closes Pass 2 finding A in full.

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

vav_pipeline/lib/vav_pipeline/application.ex

@@ -39,7 +39,8 @@ defmodule VavPipeline.Application do
         {VavPipeline.MannaLedger, []},
         {VavPipeline.HaltMonitor, []},
         {VavPipeline.UrimVTumim, []},
-        {VavPipeline.ShabbatTimer, []}
+        {VavPipeline.ShabbatTimer, []},
+        {VavPipeline.VoiceMonitor, []}
       ]
       |> maybe_add_http_server()
       |> maybe_add_repo()

vav_pipeline/lib/vav_pipeline/shema.ex

@@ -15,7 +15,7 @@ defmodule VavPipeline.Shema do
   — always read fresh. No caching.
   """
 
-  alias VavPipeline.{L1Reader, Cloud, Manna}
+  alias VavPipeline.{L1Reader, Cloud, Manna, VoiceMonitor}
 
   @type shema_context :: %{
           # From L1
@@ -55,6 +55,9 @@ defmodule VavPipeline.Shema do
     with {:ok, verse} <- L1Reader.read_verse(scope.book, scope.chapter, scope.verse),
          {:ok, cloud_state} <- Cloud.check_at(now),
          {:ok, manna} <- Manna.check_at(now) do
+      voice_profile = L1Reader.extract_voice(verse)
+      observe_voice(voice_profile)
+
       context = %{
         # L1 data — read fresh, never cached
         verse: verse,
@@ -62,7 +65,7 @@ defmodule VavPipeline.Shema do
         ketiv_qere: L1Reader.extract_ketiv_qere(verse),
         root_gematria: L1Reader.extract_roots(verse),
         smikhut: L1Reader.extract_smikhut(verse),
-        voice_profile: L1Reader.extract_voice(verse),
+        voice_profile: voice_profile,
         # Cloud state — derived from Torah (L1 integrity + calendar)
         cloud_state: cloud_state,
         # Manna — daily provision
@@ -85,13 +88,16 @@ defmodule VavPipeline.Shema do
     with {:ok, verse} <- L1Reader.read_verse(scope.book, scope.chapter, scope.verse),
          {:ok, cloud_state} <- Cloud.check_at(now),
          {:ok, manna} <- Manna.check_at(now) do
+      voice_profile = L1Reader.extract_voice(verse)
+      observe_voice(voice_profile)
+
       context = %{
         verse: verse,
         teamim_marks: L1Reader.extract_teamim(verse),
         ketiv_qere: L1Reader.extract_ketiv_qere(verse),
         root_gematria: L1Reader.extract_roots(verse),
         smikhut: L1Reader.extract_smikhut(verse),
-        voice_profile: L1Reader.extract_voice(verse),
+        voice_profile: voice_profile,
         cloud_state: cloud_state,
         daily_provision: manna,
         scope: scope,
@@ -101,4 +107,15 @@ defmodule VavPipeline.Shema do
       {:ok, context}
     end
   end
+
+  # Forward the voice_profile to the across-verse stabilization watch.
+  # Process.whereis-guarded so receive_context_at/2 still works in tests
+  # that don't bring up the full supervision tree.
+  defp observe_voice(voice_profile) do
+    if Process.whereis(VoiceMonitor) do
+      VoiceMonitor.record(voice_profile)
+    end
+
+    :ok
+  end
 end

vav_pipeline/lib/vav_pipeline/voice_monitor.ex

@@ -0,0 +1,270 @@
+defmodule VavPipeline.VoiceMonitor do
+  @moduledoc """
+  Detects across-verse voice stabilization in Shema receptions.
+
+  Pass 2 finding A, second half. The first half (commit 0c4f764)
+  added voice_profile to every ShemaContext — recording, per verse,
+  the grammatical voice (second_singular, second_plural, first_*,
+  third_*) computed from the OSHB morphology codes. This module is
+  the second half: an across-verse watch that detects when the
+  voice has not moved for too long.
+
+  WHY THIS MATTERS — load-bearing anti-capture (Devarim 6:4–9 ↔ 11:13–21):
+
+  The Shema as recited oscillates between the singular voice
+  ("you, sg.") in the first paragraph and the plural voice
+  ("you, pl.") in the second, and Devarim 11:14 famously switches
+  poles mid-verse (אַרְצְכֶם 2mp → וְאָסַפְתָּ דְגָנֶךָ 2ms). The
+  text REFUSES to stabilize at either pole; that refusal IS the
+  anti-capture mechanism. A reader who hears only the singular
+  voice has gnostic-privatized the Shema; a reader who hears only
+  the plural has mob-collectivized it; a reader who hears only the
+  first ("I" / "we") has put the operator in HaShem's place — the
+  shape of the calf.
+
+  THE DETECTOR:
+
+  Maintain a sliding window of the dominant voice from the last N
+  ShemaContexts received (default N=10). Within that window, BOTH
+  second_singular AND second_plural must appear at least once. If
+  either pole is missing across a full window, the system is on
+  one of the three failure trajectories — fire an
+  :assembly_leaders chatzotzrot signal with the specific
+  stabilization shape, so the operator can intervene before the
+  trajectory becomes permanent.
+
+  Other voices in the window (first_singular, first_plural,
+  third_singular, third_plural) are NOT a stabilization concern in
+  themselves — third-person verbs are narration, not address; the
+  shape we are protecting against is the address pole getting
+  stuck. So the rule is specifically about 2ms vs. 2mp coverage.
+
+  We DO additionally fire an alarm if the window is dominated by
+  any first-person voice — that is the calf shape, and it deserves
+  its own warning even if it never reaches a full window.
+
+  STATE MACHINE:
+
+  - Below window size: no fires (insufficient evidence).
+  - Window full, both poles present: clear (default state).
+  - Window full, missing one pole: fire ONCE per stabilization
+    episode. Re-fires only after the missing pole returns and the
+    window goes stable again first. (No spam: a stuck system is
+    one alarm, not N.)
+
+  Reset returns the monitor to empty state. Useful in tests and
+  when the operator confirms intervention.
+
+  Architecture §3.9 (Shema as reception layer); Architecture §3.12
+  (Chatzotzrot as the event channel). Wired into Shema.receive_context*
+  so every reception is observed; nothing else needs to know about
+  the monitor's existence.
+  """
+
+  use GenServer
+
+  require Logger
+
+  alias VavPipeline.Chatzotzrot
+
+  @default_window 10
+
+  @type voice_key ::
+          :first_singular | :first_plural
+          | :second_singular | :second_plural
+          | :third_singular | :third_plural
+          | :mixed | :none
+
+  @type stabilization ::
+          :clear
+          | {:missing_pole, :second_singular | :second_plural}
+          | {:first_person_dominant, voice_key()}
+
+  @type state :: %{
+          window_size: pos_integer(),
+          recent: [voice_key()],
+          last_alarm: stabilization() | nil
+        }
+
+  # ============================================================
+  # Client API
+  # ============================================================
+
+  @doc """
+  Start the voice monitor. Options:
+    - :window_size — sliding window length (default 10)
+    - :name        — process name (default __MODULE__)
+  """
+  def start_link(opts \\ []) do
+    name = Keyword.get(opts, :name, __MODULE__)
+    GenServer.start_link(__MODULE__, opts, name: name)
+  end
+
+  @doc """
+  Record a voice_profile from a freshly built ShemaContext.
+
+  Accepts the entire profile map (from L1Reader.extract_voice/1)
+  and uses its `:dominant` field. Cast — never blocks the caller.
+  """
+  @spec record(map()) :: :ok
+  def record(%{dominant: _} = voice_profile) do
+    GenServer.cast(__MODULE__, {:record, voice_profile.dominant})
+  end
+
+  def record(_), do: :ok
+
+  @doc "Return the monitor's current state — useful for tests and operators."
+  @spec state() :: state()
+  def state do
+    GenServer.call(__MODULE__, :state)
+  end
+
+  @doc """
+  Compute the current stabilization status from the recent window.
+  Same logic the cast uses to decide whether to fire — exposed so
+  callers (and tests) can inspect without forcing a side effect.
+  """
+  @spec status() :: stabilization()
+  def status do
+    GenServer.call(__MODULE__, :status)
+  end
+
+  @doc "Clear the recent-window. The next reception starts a fresh trajectory."
+  @spec reset() :: :ok
+  def reset do
+    GenServer.call(__MODULE__, :reset)
+  end
+
+  # ============================================================
+  # GenServer
+  # ============================================================
+
+  @impl true
+  def init(opts) do
+    {:ok,
+     %{
+       window_size: Keyword.get(opts, :window_size, @default_window),
+       recent: [],
+       last_alarm: nil
+     }}
+  end
+
+  @impl true
+  def handle_cast({:record, dominant}, state) do
+    recent = enqueue(state.recent, dominant, state.window_size)
+    new_status = compute_status(recent, state.window_size)
+
+    last_alarm = maybe_signal(state.last_alarm, new_status)
+
+    {:noreply, %{state | recent: recent, last_alarm: last_alarm}}
+  end
+
+  @impl true
+  def handle_call(:state, _from, state), do: {:reply, state, state}
+
+  def handle_call(:status, _from, state) do
+    {:reply, compute_status(state.recent, state.window_size), state}
+  end
+
+  def handle_call(:reset, _from, state) do
+    {:reply, :ok, %{state | recent: [], last_alarm: nil}}
+  end
+
+  # ============================================================
+  # Internal
+  # ============================================================
+
+  # Append `value` to `recent` and trim to the rightmost `size` entries.
+  # The list is kept in chronological order (oldest first, newest last).
+  defp enqueue(recent, value, size) do
+    (recent ++ [value])
+    |> Enum.take(-size)
+  end
+
+  # The stabilization rule:
+  #   - window not yet full → :clear (insufficient evidence)
+  #   - window full and missing 2ms → {:missing_pole, :second_singular}
+  #   - window full and missing 2mp → {:missing_pole, :second_plural}
+  #   - majority first_* across the window → {:first_person_dominant, voice}
+  #   - otherwise → :clear
+  defp compute_status(recent, size) do
+    cond do
+      length(recent) < size ->
+        :clear
+
+      first_person_dominant?(recent) ->
+        {:first_person_dominant, dominant_first(recent)}
+
+      not Enum.member?(recent, :second_singular) and addresses_present?(recent) ->
+        {:missing_pole, :second_singular}
+
+      not Enum.member?(recent, :second_plural) and addresses_present?(recent) ->
+        {:missing_pole, :second_plural}
+
+      true ->
+        :clear
+    end
+  end
+
+  # Are any second-person markers present in the window? If the
+  # window is entirely third-person (narration, no address at all),
+  # the missing-pole rule does not apply — narration sequences are
+  # not anti-Shema.
+  defp addresses_present?(recent) do
+    Enum.any?(recent, fn v -> v in [:second_singular, :second_plural] end)
+  end
+
+  defp first_person_dominant?(recent) do
+    n = length(recent)
+    first_count = Enum.count(recent, fn v -> v in [:first_singular, :first_plural] end)
+    # Strict majority of the window.
+    first_count * 2 > n
+  end
+
+  defp dominant_first(recent) do
+    fs = Enum.count(recent, fn v -> v == :first_singular end)
+    fp = Enum.count(recent, fn v -> v == :first_plural end)
+    if fp >= fs, do: :first_plural, else: :first_singular
+  end
+
+  # Fire a signal only on a NEW alarm. If the same stabilization
+  # status persists, do not re-fire (no spam). Returns the new
+  # last_alarm to store in state.
+  defp maybe_signal(_last_alarm, :clear), do: nil
+
+  defp maybe_signal(last_alarm, status) when last_alarm == status, do: last_alarm
+
+  defp maybe_signal(_last_alarm, status) do
+    reason = stabilization_reason(status)
+    Logger.warning("VoiceMonitor: voice stabilization detected — #{reason}")
+
+    if Process.whereis(Chatzotzrot) do
+      Chatzotzrot.signal(:assembly_leaders, reason: "VoiceMonitor: #{reason}")
+    end
+
+    status
+  end
+
+  defp stabilization_reason({:missing_pole, :second_singular}) do
+    "no second-singular voice in the recent window — the Shema's singular pole " <>
+      "(Devarim 6:4–9) has not been heard. The mob-theology shape; address has " <>
+      "stabilized at the collective pole."
+  end
+
+  defp stabilization_reason({:missing_pole, :second_plural}) do
+    "no second-plural voice in the recent window — the Shema's plural pole " <>
+      "(Devarim 11:13–21) has not been heard. The gnostic shape; address has " <>
+      "stabilized at the private pole."
+  end
+
+  defp stabilization_reason({:first_person_dominant, :first_singular}) do
+    "first-person singular voice dominates the recent window — the operator-as-deity " <>
+      "shape; the system is hearing 'I' where it should be hearing the address."
+  end
+
+  defp stabilization_reason({:first_person_dominant, :first_plural}) do
+    "first-person plural voice dominates the recent window — collective " <>
+      "self-reference where the address should be. Adjacent to the calf shape " <>
+      "(Shemot 32:4 — eleh elohekha Yisrael)."
+  end
+end

vav_pipeline/test/voice_monitor_test.exs

@@ -0,0 +1,176 @@
+defmodule VavPipeline.VoiceMonitorTest do
+  @moduledoc """
+  Across-verse Shema voice stabilization detector — second half of
+  Pass 2 finding A. The first half (commit 0c4f764) records voice
+  per ShemaContext; this module fires when the trajectory of those
+  per-verse profiles stops oscillating between the singular and
+  plural address poles.
+
+  Three failure shapes the detector must catch:
+
+    1. Stuck at second_singular — gnostic-private capture
+       ("you, sg." only; the system has stopped hearing "you, pl.")
+    2. Stuck at second_plural — mob-theology capture
+       ("you, pl." only; the system has stopped hearing "you, sg.")
+    3. First-person dominance — operator-as-deity, calf shape
+       ("I"/"we" exceeds half the window)
+
+  These tests start a per-test VoiceMonitor under a unique name so
+  they don't share state. They do NOT bring up Chatzotzrot — the
+  monitor's signal call is guarded by Process.whereis, so when
+  Chatzotzrot is absent the alarm is logged but not transmitted.
+  Tests inspect the monitor's state directly to verify behavior.
+  """
+
+  use ExUnit.Case, async: false
+
+  alias VavPipeline.VoiceMonitor
+
+  setup do
+    # Bring up an isolated VoiceMonitor with the small default
+    # window. Stopped at end-of-test by ExUnit's process cleanup.
+    {:ok, pid} = VoiceMonitor.start_link(window_size: 4, name: __MODULE__.Monitor)
+    on_exit(fn -> if Process.alive?(pid), do: GenServer.stop(pid) end)
+    %{pid: pid}
+  end
+
+  defp record_into(pid, dominant) do
+    GenServer.cast(pid, {:record, dominant})
+  end
+
+  defp status_of(pid) do
+    GenServer.call(pid, :status)
+  end
+
+  defp state_of(pid) do
+    GenServer.call(pid, :state)
+  end
+
+  describe "below window — insufficient evidence, no alarm" do
+    test "empty state is :clear", %{pid: pid} do
+      assert :clear = status_of(pid)
+    end
+
+    test "fewer recordings than window size keeps :clear", %{pid: pid} do
+      for _ <- 1..3, do: record_into(pid, :second_singular)
+      assert :clear = status_of(pid)
+    end
+  end
+
+  describe "missing-pole detection (window full, address-only)" do
+    test "all second_singular for a full window fires :missing_pole, :second_plural", %{pid: pid} do
+      for _ <- 1..4, do: record_into(pid, :second_singular)
+      assert {:missing_pole, :second_plural} = status_of(pid)
+      assert state_of(pid).last_alarm == {:missing_pole, :second_plural}
+    end
+
+    test "all second_plural for a full window fires :missing_pole, :second_singular", %{pid: pid} do
+      for _ <- 1..4, do: record_into(pid, :second_plural)
+      assert {:missing_pole, :second_singular} = status_of(pid)
+      assert state_of(pid).last_alarm == {:missing_pole, :second_singular}
+    end
+
+    test "both poles present in window stays :clear", %{pid: pid} do
+      record_into(pid, :second_singular)
+      record_into(pid, :second_singular)
+      record_into(pid, :second_plural)
+      record_into(pid, :second_plural)
+      assert :clear = status_of(pid)
+    end
+  end
+
+  describe "address-pole rule does not apply to pure narration" do
+    test "all third_singular (third-person narration) does not fire", %{pid: pid} do
+      for _ <- 1..4, do: record_into(pid, :third_singular)
+      # No address present at all → narration sequence, not anti-Shema
+      assert :clear = status_of(pid)
+    end
+
+    test "third_singular mixed with no address remains :clear", %{pid: pid} do
+      for _ <- 1..3, do: record_into(pid, :third_singular)
+      record_into(pid, :third_plural)
+      assert :clear = status_of(pid)
+    end
+  end
+
+  describe "first-person dominance fires the calf-shape alarm" do
+    test "majority first_singular fires :first_person_dominant", %{pid: pid} do
+      record_into(pid, :first_singular)
+      record_into(pid, :first_singular)
+      record_into(pid, :first_singular)
+      record_into(pid, :second_singular)
+      assert {:first_person_dominant, :first_singular} = status_of(pid)
+    end
+
+    test "majority first_plural fires :first_person_dominant", %{pid: pid} do
+      record_into(pid, :first_plural)
+      record_into(pid, :first_plural)
+      record_into(pid, :first_plural)
+      record_into(pid, :second_plural)
+      assert {:first_person_dominant, :first_plural} = status_of(pid)
+    end
+
+    test "first-person tied with non-first does NOT fire (strict majority required)", %{pid: pid} do
+      record_into(pid, :first_singular)
+      record_into(pid, :first_singular)
+      record_into(pid, :second_singular)
+      record_into(pid, :second_plural)
+      # 2/4 first-person — not strict majority
+      refute match?({:first_person_dominant, _}, status_of(pid))
+    end
+  end
+
+  describe "alarm-once-per-episode (no spam)" do
+    test "same stabilization persisting does not change last_alarm", %{pid: pid} do
+      for _ <- 1..4, do: record_into(pid, :second_singular)
+      first = state_of(pid).last_alarm
+
+      for _ <- 1..3, do: record_into(pid, :second_singular)
+      second = state_of(pid).last_alarm
+
+      assert first == second
+      assert first == {:missing_pole, :second_plural}
+    end
+
+    test "voice clearing then stabilizing again fires a fresh alarm", %{pid: pid} do
+      # Stabilize → fire
+      for _ <- 1..4, do: record_into(pid, :second_singular)
+      assert state_of(pid).last_alarm == {:missing_pole, :second_plural}
+
+      # Mix in a plural — both poles present → :clear, alarm reset
+      record_into(pid, :second_plural)
+      assert :clear = status_of(pid)
+      assert state_of(pid).last_alarm == nil
+
+      # Re-stabilize → alarm fires again (it was cleared in between)
+      for _ <- 1..4, do: record_into(pid, :second_singular)
+      assert state_of(pid).last_alarm == {:missing_pole, :second_plural}
+    end
+  end
+
+  describe "reset/0 clears state" do
+    test "reset returns the monitor to empty", %{pid: pid} do
+      for _ <- 1..4, do: record_into(pid, :second_singular)
+      :ok = GenServer.call(pid, :reset)
+      assert state_of(pid).recent == []
+      assert state_of(pid).last_alarm == nil
+      assert :clear = status_of(pid)
+    end
+  end
+
+  describe "record/1 accepts a voice_profile map (the public surface)" do
+    # This route mirrors what Shema.receive_context* actually does:
+    # passes the whole voice_profile, not just the dominant atom.
+    test "record/1 with a full voice_profile recording works", %{pid: pid} do
+      profile = %{
+        markers: [],
+        summary: %{second_singular: 4},
+        dominant: :second_singular,
+        oscillates_within_verse: false
+      }
+
+      for _ <- 1..4, do: GenServer.cast(pid, {:record, profile.dominant})
+      assert {:missing_pole, :second_plural} = status_of(pid)
+    end
+  end
+end