From 7fe80254348a05be87fa845785aafbfcb614f7d6 Mon Sep 17 00:00:00 2001 From: Scott Karp Date: Tue, 17 Mar 2026 15:50:36 -0400 Subject: [PATCH] Chatzotzrot: silver trumpets with BEAM routing, subscription, and zikaron MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- vav_pipeline/lib/vav_pipeline/application.ex | 2 + vav_pipeline/lib/vav_pipeline/chatzotzrot.ex | 267 +++++++++++++++-- vav_pipeline/test/chatzotzrot_test.exs | 288 +++++++++++++++++++ 3 files changed, 533 insertions(+), 24 deletions(-) create mode 100644 vav_pipeline/test/chatzotzrot_test.exs diff --git a/vav_pipeline/lib/vav_pipeline/application.ex b/vav_pipeline/lib/vav_pipeline/application.ex index 40d995f9..83a25d21 100644 --- a/vav_pipeline/lib/vav_pipeline/application.ex +++ b/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() diff --git a/vav_pipeline/lib/vav_pipeline/chatzotzrot.ex b/vav_pipeline/lib/vav_pipeline/chatzotzrot.ex index 81167e07..91be1caf 100644 --- a/vav_pipeline/lib/vav_pipeline/chatzotzrot.ex +++ b/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: - Signal types: assembly_all, assembly_leaders, breaking_camp, war, moadim - Patterns: tekiah, teruah, tekiah_teruah, teruah_teruah, tekiah_tekiah + Architecture §3.12, Construction §3.5: + Each signal is a different KIND of event, not a different severity. + The trumpets are mikshah — hammered from a single piece of silver. - In Phase 2, signals are logged. In production, they will - be distributed via BEAM messaging. + Signal types and their trumpet patterns: + 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 """ - Send a chatzotzrot signal. + @type recipient_group :: :all | :kohanim | :operators - 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() - def signal(type, opts \\ []) do - signal = %{ + @spec signal(signal_type(), keyword()) :: signal_record() + def signal(type, opts \\ []) when is_map_key(@signals, type) do + 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 diff --git a/vav_pipeline/test/chatzotzrot_test.exs b/vav_pipeline/test/chatzotzrot_test.exs new file mode 100644 index 00000000..498de2f0 --- /dev/null +++ b/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