Chatzotzrot: silver trumpets with BEAM routing, subscription, and zikaron

Bamidbar 10:2 — Aseh lekha shtei chatzotz'rot kesef, mikshah ta'aseh otam.
Two trumpets, hammered from a single piece. Five signals, five patterns.

- Upgraded from simple function module to GenServer with state
- Recipient routing: :all (assembly_all, breaking_camp, moadim),
  :kohanim (assembly_leaders, war), :operators (war)
- PubSub via BEAM messaging: subscribe/1, unsubscribe/1
  Signals delivered as {:chatzotzrot, record} messages
- Zikaron (remembrance — Bamidbar 10:10): signal history with
  type filtering and limit. V'hayu lakhem l'zikaron.
- Dead subscriber cleanup via Process.monitor/1
- Added to supervision tree in application.ex
- 30 new tests: patterns, urgency, routing, subscription,
  zikaron history, dead process cleanup, format compliance
- All 125 Elixir tests pass, 0 failures

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

vav_pipeline/lib/vav_pipeline/application.ex

@@ -4,6 +4,7 @@ defmodule VavPipeline.Application do
 
   Starts the supervision tree with:
   - Repo (L3 PostgreSQL — oral tradition persistence)
+  - Chatzotzrot (silver trumpets — signal system)
   - ShabbatTimer (mandatory cessation monitor)
   - ShoreshEnginePort (Erlang Port to Haskell engine, if binary found)
   - TalmudicEngineSocket (Unix socket to Prolog engine)
@@ -25,6 +26,7 @@ defmodule VavPipeline.Application do
 
     children =
       [
+        {VavPipeline.Chatzotzrot, []},
         {VavPipeline.ShabbatTimer, []}
       ]
       |> maybe_add_repo()

vav_pipeline/lib/vav_pipeline/chatzotzrot.ex

@@ -1,15 +1,30 @@
 defmodule VavPipeline.Chatzotzrot do
   @moduledoc """
-  Signal system — the silver trumpets.
+  Two silver trumpets. Five signals. Bamidbar 10:1-10.
 
-  Construction spec §3.5 — Chatzotzrot Signal Format:
+  Architecture §3.12, Construction §3.5:
-  Signal types: assembly_all, assembly_leaders, breaking_camp, war, moadim
+  Each signal is a different KIND of event, not a different severity.
-  Patterns: tekiah, teruah, tekiah_teruah, teruah_teruah, tekiah_tekiah
+  The trumpets are mikshah — hammered from a single piece of silver.
 
-  In Phase 2, signals are logged. In production, they will
+  Signal types and their trumpet patterns:
-  be distributed via BEAM messaging.
+    assembly_all     → tekiah          — all operators and users
+    assembly_leaders → teruah          — kohanim and senior kohanim only
+    breaking_camp    → tekiah_teruah   — system is moving
+    war              → teruah_teruah   — system under attack
+    moadim           → tekiah_tekiah   — milestones and festivals
+
+  V'hayu lakhem l'zikaron lifnei Eloheikhem — Bamidbar 10:10.
+  "They shall be for you a remembrance before your God."
+  The signal system serves MEMORY, not just notification.
+  Every signal is recorded in the zikaron (signal history).
+
+  Recipients subscribe by group (:all, :kohanim, :operators).
+  Signals are routed to the correct groups and delivered as
+  {:chatzotzrot, signal_record} messages via BEAM messaging.
   """
 
+  use GenServer
+
   require Logger
 
   @type signal_type ::
@@ -21,16 +36,212 @@ defmodule VavPipeline.Chatzotzrot do
 
   @type pattern :: :tekiah | :teruah | :tekiah_teruah | :teruah_teruah | :tekiah_tekiah
 
-  @doc """
+  @type recipient_group :: :all | :kohanim | :operators
-  Send a chatzotzrot signal.
 
-  Returns the signal record for logging/testing.
+  @type signal_record :: %{
+          signal_type: signal_type(),
+          pattern: pattern(),
+          timestamp: String.t(),
+          source: :chatzotzrot,
+          details: %{reason: String.t(), urgency: :immediate | :next_cycle | :informational}
+        }
+
+  # === Signal → Pattern mapping (Bamidbar 10, Architecture §3.12) ===
+  @signals %{
+    assembly_all: :tekiah,
+    assembly_leaders: :teruah,
+    breaking_camp: :tekiah_teruah,
+    war: :teruah_teruah,
+    moadim: :tekiah_tekiah
+  }
+
+  # === Signal → Recipient groups ===
+  # Not all signals go to all listeners.
+  # The kohanim blow — but who hears depends on the signal type.
+  @routing %{
+    assembly_all: [:all],
+    assembly_leaders: [:kohanim],
+    breaking_camp: [:all],
+    war: [:operators, :kohanim],
+    moadim: [:all]
+  }
+
+  # ============================================================
+  # Client API
+  # ============================================================
+
+  @doc "Start the chatzotzrot GenServer."
+  def start_link(opts \\ []) do
+    name = Keyword.get(opts, :name, __MODULE__)
+    GenServer.start_link(__MODULE__, opts, name: name)
+  end
+
+  @doc """
+  Sound a chatzotzrot signal.
+
+  Options:
+    - :reason — why this signal is being sounded (string)
+    - :urgency — override default urgency (:immediate | :next_cycle | :informational)
+
+  Returns the signal record.
   """
-  @spec signal(signal_type(), keyword()) :: map()
+  @spec signal(signal_type(), keyword()) :: signal_record()
-  def signal(type, opts \\ []) do
+  def signal(type, opts \\ []) when is_map_key(@signals, type) do
-    signal = %{
+    GenServer.call(__MODULE__, {:signal, type, opts})
+  end
+
+  @doc """
+  Subscribe the calling process to receive signals for a recipient group.
+
+  Groups:
+    - :all       — receives assembly_all, breaking_camp, moadim
+    - :kohanim   — receives assembly_leaders, war
+    - :operators  — receives war
+
+  A process may subscribe to multiple groups.
+  Signals are delivered as messages: {:chatzotzrot, signal_record}
+  """
+  @spec subscribe(recipient_group()) :: :ok
+  def subscribe(group) when group in [:all, :kohanim, :operators] do
+    GenServer.call(__MODULE__, {:subscribe, group, self()})
+  end
+
+  @doc """
+  Unsubscribe the calling process from a recipient group.
+  """
+  @spec unsubscribe(recipient_group()) :: :ok
+  def unsubscribe(group) when group in [:all, :kohanim, :operators] do
+    GenServer.call(__MODULE__, {:unsubscribe, group, self()})
+  end
+
+  @doc """
+  Get the signal history (zikaron).
+
+  V'hayu lakhem l'zikaron — "They shall be for you a remembrance."
+  Every signal ever sounded is recorded.
+
+  Options:
+    - :type — filter by signal type
+    - :limit — max number of records to return (default: all)
+
+  Returns signals in reverse chronological order (newest first).
+  """
+  @spec zikaron(keyword()) :: [signal_record()]
+  def zikaron(opts \\ []) do
+    GenServer.call(__MODULE__, {:zikaron, opts})
+  end
+
+  @doc """
+  Count total signals sounded since startup.
+  """
+  @spec count() :: non_neg_integer()
+  def count do
+    GenServer.call(__MODULE__, :count)
+  end
+
+  # ============================================================
+  # GenServer callbacks
+  # ============================================================
+
+  @impl true
+  def init(_opts) do
+    state = %{
+      # Subscribers by group: %{group => MapSet.t(pid)}
+      subscribers: %{
+        all: MapSet.new(),
+        kohanim: MapSet.new(),
+        operators: MapSet.new()
+      },
+      # Signal history — zikaron (remembrance). Newest first.
+      zikaron: [],
+      # Total count
+      count: 0
+    }
+
+    Logger.info("Chatzotzrot: two silver trumpets ready — mikshah, hammered from a single piece")
+    {:ok, state}
+  end
+
+  @impl true
+  def handle_call({:signal, type, opts}, _from, state) do
+    record = build_signal(type, opts)
+
+    # Route to recipient groups
+    groups = Map.fetch!(@routing, type)
+    recipients = collect_recipients(groups, state.subscribers)
+
+    # Deliver signal — each recipient gets {:chatzotzrot, record}
+    live_recipients = Enum.filter(recipients, &Process.alive?/1)
+    Enum.each(live_recipients, fn pid -> send(pid, {:chatzotzrot, record}) end)
+    delivered_count = length(live_recipients)
+
+    Logger.info(
+      "Chatzotzrot: #{type} (#{record.pattern}) — #{record.details.reason} — " <>
+        "delivered to #{delivered_count} recipient(s)"
+    )
+
+    new_state = %{
+      state
+      | zikaron: [record | state.zikaron],
+        count: state.count + 1
+    }
+
+    {:reply, record, new_state}
+  end
+
+  @impl true
+  def handle_call({:subscribe, group, pid}, _from, state) do
+    Process.monitor(pid)
+
+    new_subscribers =
+      Map.update!(state.subscribers, group, &MapSet.put(&1, pid))
+
+    {:reply, :ok, %{state | subscribers: new_subscribers}}
+  end
+
+  @impl true
+  def handle_call({:unsubscribe, group, pid}, _from, state) do
+    new_subscribers =
+      Map.update!(state.subscribers, group, &MapSet.delete(&1, pid))
+
+    {:reply, :ok, %{state | subscribers: new_subscribers}}
+  end
+
+  @impl true
+  def handle_call({:zikaron, opts}, _from, state) do
+    records = filter_zikaron(state.zikaron, opts)
+    {:reply, records, state}
+  end
+
+  @impl true
+  def handle_call(:count, _from, state) do
+    {:reply, state.count, state}
+  end
+
+  @impl true
+  def handle_info({:DOWN, _ref, :process, pid, _reason}, state) do
+    # Clean up dead subscribers from all groups
+    new_subscribers =
+      Map.new(state.subscribers, fn {group, pids} ->
+        {group, MapSet.delete(pids, pid)}
+      end)
+
+    {:noreply, %{state | subscribers: new_subscribers}}
+  end
+
+  @impl true
+  def handle_info(_msg, state) do
+    {:noreply, state}
+  end
+
+  # ============================================================
+  # Internal functions
+  # ============================================================
+
+  defp build_signal(type, opts) do
+    %{
       signal_type: type,
-      pattern: pattern_for(type),
+      pattern: Map.fetch!(@signals, type),
       timestamp: DateTime.utc_now() |> DateTime.to_iso8601(),
       source: :chatzotzrot,
       details: %{
@@ -38,21 +249,29 @@ defmodule VavPipeline.Chatzotzrot do
         urgency: Keyword.get(opts, :urgency, urgency_for(type))
       }
     }
-
-    Logger.info("Chatzotzrot: #{type} (#{signal.pattern}) — #{signal.details.reason}")
-
-    signal
   end
 
-  # Signal type → trumpet pattern mapping (Bamidbar 10, Architecture §3.12)
-  defp pattern_for(:assembly_all), do: :tekiah
-  defp pattern_for(:assembly_leaders), do: :teruah
-  defp pattern_for(:breaking_camp), do: :tekiah_teruah
-  defp pattern_for(:war), do: :teruah_teruah
-  defp pattern_for(:moadim), do: :tekiah_tekiah
-
   defp urgency_for(:breaking_camp), do: :immediate
   defp urgency_for(:war), do: :immediate
   defp urgency_for(:assembly_all), do: :next_cycle
   defp urgency_for(_), do: :informational
+
+  defp collect_recipients(groups, subscribers) do
+    groups
+    |> Enum.flat_map(fn group -> MapSet.to_list(Map.fetch!(subscribers, group)) end)
+    |> Enum.uniq()
+  end
+
+  defp filter_zikaron(zikaron, opts) do
+    type_filter = Keyword.get(opts, :type)
+    limit = Keyword.get(opts, :limit)
+
+    zikaron
+    |> then(fn records ->
+      if type_filter, do: Enum.filter(records, &(&1.signal_type == type_filter)), else: records
+    end)
+    |> then(fn records ->
+      if limit, do: Enum.take(records, limit), else: records
+    end)
+  end
 end

vav_pipeline/test/chatzotzrot_test.exs

@@ -0,0 +1,288 @@
+defmodule VavPipeline.ChatzotzrotTest do
+  @moduledoc """
+  Tests for the Chatzotzrot signaling system.
+
+  Bamidbar 10:2 — Aseh lekha shtei chatzotz'rot kesef,
+  mikshah ta'aseh otam.
+
+  "Make for yourself two trumpets of silver;
+  of hammered work you shall make them."
+
+  Five signals. Five patterns. Routing by recipient group.
+  Every signal recorded in the zikaron (remembrance).
+  """
+
+  use ExUnit.Case, async: false
+
+  alias VavPipeline.Chatzotzrot
+
+  # ============================================================
+  # Signal Type → Pattern Mapping (Bamidbar 10, Architecture §3.12)
+  # ============================================================
+  describe "signal patterns" do
+    test "assembly_all → tekiah" do
+      signal = Chatzotzrot.signal(:assembly_all)
+      assert signal.signal_type == :assembly_all
+      assert signal.pattern == :tekiah
+      assert signal.source == :chatzotzrot
+      assert signal.timestamp != nil
+    end
+
+    test "assembly_leaders → teruah" do
+      signal = Chatzotzrot.signal(:assembly_leaders)
+      assert signal.signal_type == :assembly_leaders
+      assert signal.pattern == :teruah
+    end
+
+    test "breaking_camp → tekiah_teruah" do
+      signal = Chatzotzrot.signal(:breaking_camp)
+      assert signal.signal_type == :breaking_camp
+      assert signal.pattern == :tekiah_teruah
+    end
+
+    test "war → teruah_teruah" do
+      signal = Chatzotzrot.signal(:war)
+      assert signal.signal_type == :war
+      assert signal.pattern == :teruah_teruah
+    end
+
+    test "moadim → tekiah_tekiah" do
+      signal = Chatzotzrot.signal(:moadim)
+      assert signal.signal_type == :moadim
+      assert signal.pattern == :tekiah_tekiah
+    end
+  end
+
+  # ============================================================
+  # Urgency (Construction §3.5)
+  # ============================================================
+  describe "urgency" do
+    test "breaking_camp is immediate" do
+      signal = Chatzotzrot.signal(:breaking_camp)
+      assert signal.details.urgency == :immediate
+    end
+
+    test "war is immediate" do
+      signal = Chatzotzrot.signal(:war)
+      assert signal.details.urgency == :immediate
+    end
+
+    test "assembly_all is next_cycle" do
+      signal = Chatzotzrot.signal(:assembly_all)
+      assert signal.details.urgency == :next_cycle
+    end
+
+    test "assembly_leaders is informational" do
+      signal = Chatzotzrot.signal(:assembly_leaders)
+      assert signal.details.urgency == :informational
+    end
+
+    test "moadim is informational" do
+      signal = Chatzotzrot.signal(:moadim)
+      assert signal.details.urgency == :informational
+    end
+
+    test "urgency can be overridden" do
+      signal = Chatzotzrot.signal(:moadim, urgency: :immediate)
+      assert signal.details.urgency == :immediate
+    end
+  end
+
+  # ============================================================
+  # Custom reason (Construction §3.5)
+  # ============================================================
+  describe "reason" do
+    test "default reason is the signal type name" do
+      signal = Chatzotzrot.signal(:war)
+      assert signal.details.reason == "war"
+    end
+
+    test "custom reason is preserved" do
+      signal = Chatzotzrot.signal(:war, reason: "DDoS detected on API layer")
+      assert signal.details.reason == "DDoS detected on API layer"
+    end
+  end
+
+  # ============================================================
+  # Subscription and Routing (Architecture §3.12)
+  # ============================================================
+  describe "subscription and routing" do
+    test "subscriber to :all receives assembly_all signal" do
+      Chatzotzrot.subscribe(:all)
+      Chatzotzrot.signal(:assembly_all)
+
+      assert_receive {:chatzotzrot, %{signal_type: :assembly_all, pattern: :tekiah}}
+    end
+
+    test "subscriber to :all receives breaking_camp signal" do
+      Chatzotzrot.subscribe(:all)
+      Chatzotzrot.signal(:breaking_camp)
+
+      assert_receive {:chatzotzrot, %{signal_type: :breaking_camp, pattern: :tekiah_teruah}}
+    end
+
+    test "subscriber to :all receives moadim signal" do
+      Chatzotzrot.subscribe(:all)
+      Chatzotzrot.signal(:moadim)
+
+      assert_receive {:chatzotzrot, %{signal_type: :moadim, pattern: :tekiah_tekiah}}
+    end
+
+    test "subscriber to :all does NOT receive assembly_leaders" do
+      Chatzotzrot.subscribe(:all)
+      Chatzotzrot.signal(:assembly_leaders)
+
+      refute_receive {:chatzotzrot, %{signal_type: :assembly_leaders}}, 100
+    end
+
+    test "subscriber to :all does NOT receive war" do
+      Chatzotzrot.subscribe(:all)
+      Chatzotzrot.signal(:war)
+
+      refute_receive {:chatzotzrot, %{signal_type: :war}}, 100
+    end
+
+    test "subscriber to :kohanim receives assembly_leaders" do
+      Chatzotzrot.subscribe(:kohanim)
+      Chatzotzrot.signal(:assembly_leaders)
+
+      assert_receive {:chatzotzrot, %{signal_type: :assembly_leaders, pattern: :teruah}}
+    end
+
+    test "subscriber to :kohanim receives war" do
+      Chatzotzrot.subscribe(:kohanim)
+      Chatzotzrot.signal(:war)
+
+      assert_receive {:chatzotzrot, %{signal_type: :war, pattern: :teruah_teruah}}
+    end
+
+    test "subscriber to :operators receives war" do
+      Chatzotzrot.subscribe(:operators)
+      Chatzotzrot.signal(:war)
+
+      assert_receive {:chatzotzrot, %{signal_type: :war, pattern: :teruah_teruah}}
+    end
+
+    test "subscriber to :operators does NOT receive assembly_all" do
+      Chatzotzrot.subscribe(:operators)
+      Chatzotzrot.signal(:assembly_all)
+
+      refute_receive {:chatzotzrot, _}, 100
+    end
+
+    test "process subscribed to multiple groups receives signal once" do
+      # War routes to both :operators and :kohanim
+      Chatzotzrot.subscribe(:operators)
+      Chatzotzrot.subscribe(:kohanim)
+      Chatzotzrot.signal(:war)
+
+      assert_receive {:chatzotzrot, %{signal_type: :war}}
+      refute_receive {:chatzotzrot, %{signal_type: :war}}, 100
+    end
+
+    test "unsubscribe stops signal delivery" do
+      Chatzotzrot.subscribe(:all)
+      Chatzotzrot.signal(:assembly_all)
+      assert_receive {:chatzotzrot, %{signal_type: :assembly_all}}
+
+      Chatzotzrot.unsubscribe(:all)
+      Chatzotzrot.signal(:assembly_all)
+      refute_receive {:chatzotzrot, _}, 100
+    end
+  end
+
+  # ============================================================
+  # Zikaron — Remembrance (Bamidbar 10:10)
+  # ============================================================
+  describe "zikaron (signal history)" do
+    test "signals are recorded in zikaron" do
+      initial_count = Chatzotzrot.count()
+
+      Chatzotzrot.signal(:assembly_all, reason: "test zikaron")
+      Chatzotzrot.signal(:war, reason: "test zikaron war")
+
+      history = Chatzotzrot.zikaron()
+      assert length(history) >= 2
+
+      # Newest first
+      [newest | _] = history
+      assert newest.signal_type == :war
+      assert newest.details.reason == "test zikaron war"
+
+      assert Chatzotzrot.count() >= initial_count + 2
+    end
+
+    test "zikaron can be filtered by type" do
+      Chatzotzrot.signal(:moadim, reason: "zikaron filter test")
+      Chatzotzrot.signal(:war, reason: "zikaron filter test")
+
+      moadim_history = Chatzotzrot.zikaron(type: :moadim)
+      assert Enum.all?(moadim_history, &(&1.signal_type == :moadim))
+    end
+
+    test "zikaron can be limited" do
+      Chatzotzrot.signal(:assembly_all)
+      Chatzotzrot.signal(:assembly_all)
+      Chatzotzrot.signal(:assembly_all)
+
+      limited = Chatzotzrot.zikaron(limit: 2)
+      assert length(limited) == 2
+    end
+
+    test "zikaron can be filtered and limited" do
+      Chatzotzrot.signal(:moadim, reason: "combined filter 1")
+      Chatzotzrot.signal(:moadim, reason: "combined filter 2")
+      Chatzotzrot.signal(:moadim, reason: "combined filter 3")
+
+      result = Chatzotzrot.zikaron(type: :moadim, limit: 1)
+      assert length(result) == 1
+      assert hd(result).signal_type == :moadim
+    end
+  end
+
+  # ============================================================
+  # Dead subscriber cleanup
+  # ============================================================
+  describe "dead subscriber cleanup" do
+    test "dead processes are removed from subscriber list" do
+      # Spawn a process that subscribes then dies
+      pid =
+        spawn(fn ->
+          Chatzotzrot.subscribe(:all)
+
+          receive do
+            :stop -> :ok
+          end
+        end)
+
+      # Give it time to subscribe
+      Process.sleep(50)
+      send(pid, :stop)
+      Process.sleep(50)
+
+      # Signal should not crash — dead subscriber is cleaned up
+      signal = Chatzotzrot.signal(:assembly_all)
+      assert signal.signal_type == :assembly_all
+    end
+  end
+
+  # ============================================================
+  # Construction §3.5 — signal format compliance
+  # ============================================================
+  describe "signal format (Construction §3.5)" do
+    test "signal record has all required fields" do
+      signal = Chatzotzrot.signal(:assembly_all, reason: "format test")
+
+      assert Map.has_key?(signal, :signal_type)
+      assert Map.has_key?(signal, :pattern)
+      assert Map.has_key?(signal, :timestamp)
+      assert Map.has_key?(signal, :source)
+      assert Map.has_key?(signal, :details)
+      assert Map.has_key?(signal.details, :reason)
+      assert Map.has_key?(signal.details, :urgency)
+
+      assert signal.source == :chatzotzrot
+      assert is_binary(signal.timestamp)
+    end
+  end
+end