Step 14: Connect components — chamishim karsay zahav

Three connections built and tested:
- Haskell Shoresh Engine via Erlang Port (JSON lines over stdin/stdout)
- Prolog Talmudic Engine via Unix domain socket (JSON over /tmp/shorashim/talmudic.sock)
- Erlang Sanhedrin via native BEAM calls

New: Server.hs (JSON-line server loop), json_bridge.pl + socket_server.pl
(Prolog socket interface), ShoreshEnginePort + TalmudicEngineSocket +
SanhedrinClient (Elixir GenServer wrappers). BereishitCycle now routes
unresolved cases through Talmudic Engine and Sanhedrin. All connections
gracefully degrade — pipeline falls back to L1 data when engines unavailable.

46 Elixir tests, 33 Haskell tests, 57 Prolog tests — 0 failures.

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

vav_pipeline/lib/vav_pipeline/application.ex

@@ -4,18 +4,48 @@ defmodule VavPipeline.Application do
 
   Starts the supervision tree with:
   - ShabbatTimer (mandatory cessation monitor)
+  - ShoreshEnginePort (Erlang Port to Haskell engine, if binary found)
+  - TalmudicEngineSocket (Unix socket to Prolog engine)
+
+  Step 14 (Architecture §4): Component connections supervised.
+  The port and socket are optional — the pipeline degrades gracefully
+  if external engines are not available (clients fall back to stubs).
   """
 
   use Application
 
+  require Logger
+
   @impl true
   def start(_type, _args) do
-    children = [
-      # ShabbatTimer monitors Shabbat boundaries
-      {VavPipeline.ShabbatTimer, []}
-    ]
+    shoresh_binary = Application.get_env(:vav_pipeline, :shoresh_engine_binary, "shoresh-engine")
+    shoresh_cwd = Application.get_env(:vav_pipeline, :shoresh_engine_cwd, nil)
+    talmudic_socket = Application.get_env(:vav_pipeline, :talmudic_socket_path, "/tmp/shorashim/talmudic.sock")
+
+    children =
+      [
+        {VavPipeline.ShabbatTimer, []}
+      ]
+      |> maybe_add_shoresh_port(shoresh_binary, shoresh_cwd)
+      |> maybe_add_talmudic_socket(talmudic_socket)
 
     opts = [strategy: :one_for_one, name: VavPipeline.Supervisor]
     Supervisor.start_link(children, opts)
   end
+
+  defp maybe_add_shoresh_port(children, binary, cwd) do
+    case System.find_executable(binary) do
+      nil ->
+        Logger.warning("ShoreshEnginePort: binary '#{binary}' not in PATH — running without Haskell engine")
+        children
+
+      path ->
+        children ++ [{VavPipeline.ShoreshEnginePort, [binary: path, cwd: cwd]}]
+    end
+  end
+
+  defp maybe_add_talmudic_socket(children, socket_path) do
+    # Always add the socket client — it connects lazily on first request
+    children ++ [{VavPipeline.TalmudicEngineSocket, [socket_path: socket_path]}]
+  end
 end

vav_pipeline/lib/vav_pipeline/bereishit_cycle.ex

@@ -19,7 +19,8 @@ defmodule VavPipeline.BereishitCycle do
   Latency budget (Construction spec §2.5): 5000ms total
   """
 
-  alias VavPipeline.{Specification, CycleLog, Manna}
+  alias VavPipeline.{Specification, CycleLog, Manna,
+                      ShoreshEngineClient, TalmudicEngineClient, SanhedrinClient}
 
   require Logger
 
@@ -81,27 +82,22 @@ defmodule VavPipeline.BereishitCycle do
   """
   @spec vayomer(map(), VavPipeline.Shema.shema_context()) :: map()
   def vayomer(state, context) do
-    # In Phase 2: use L1 data from ShemaContext directly
-    # In Phase 3: call ShoreshEngineClient.process(state.raw, context)
-    roots =
-      context.verse.words
-      |> Enum.filter(fn w -> w.shoresh != nil end)
-      |> Enum.map(fn w ->
-        %{
-          shoresh: w.shoresh,
-          word: w.consonants,
-          position: w.position,
-          morph: w.morph,
-          gematria: Map.get(context.root_gematria, w.shoresh, 0)
-        }
-      end)
+    # Step 14: Call Shoresh Engine via port. Falls back to L1 data.
+    case ShoreshEngineClient.process(state.raw, context) do
+      {:ok, engine_result} ->
+        Map.merge(state, %{
+          roots_activated: get_result(engine_result, :roots_activated, "roots_activated", []),
+          parse_tree: get_result(engine_result, :parse_tree, "parse_tree", nil),
+          dual_readings: get_result(engine_result, :dual_readings, "dual_readings", []),
+          teamim_parse: context.teamim_marks,
+          ketiv_qere: context.ketiv_qere,
+          stage: :vayomer
+        })
 
-    Map.merge(state, %{
-      roots_activated: roots,
-      teamim_parse: context.teamim_marks,
-      ketiv_qere: context.ketiv_qere,
-      stage: :vayomer
-    })
+      {:error, reason} ->
+        Logger.warning("BereishitCycle.vayomer: engine error (#{reason}), using L1 fallback")
+        vayomer_fallback(state, context)
+    end
   end
 
   @doc """
@@ -202,7 +198,7 @@ defmodule VavPipeline.BereishitCycle do
   """
   @spec erev_voker(VavPipeline.cycle_result(), VavPipeline.Shema.shema_context()) ::
           VavPipeline.cycle_result()
-  def erev_voker(result, _context) do
+  def erev_voker(result, context) do
     # Log the cycle
     phase =
       case result do
@@ -217,11 +213,133 @@ defmodule VavPipeline.BereishitCycle do
     # Enforce anti-hoarding
     Manna.enforce_anti_hoarding()
 
-    result
+    # Step 14: Route unresolved cases to Talmudic Engine / Sanhedrin
+    case result do
+      {:unresolved, data} ->
+        route_if_unresolved(data, context)
+
+      _ ->
+        result
+    end
+  end
+
+  # --- Routing ---
+
+  @doc false
+  defp route_if_unresolved(data, context) do
+    # Extract roots for the Talmudic Engine query
+    roots = Map.get(data, :roots_activated, [])
+    context_roots = Enum.map(roots, fn
+      %{shoresh: s} when is_binary(s) -> s
+      %{"shoresh" => s} when is_list(s) -> Enum.join(s, "-")
+      _ -> "unknown"
+    end)
+
+    case_ref = "cycle-#{:erlang.system_time(:millisecond)}"
+
+    talmudic_context = %{
+      context_roots: context_roots,
+      question: "Interpret roots #{inspect(context_roots)} in context"
+    }
+
+    # First: consult the Talmudic Engine
+    case TalmudicEngineClient.interpret(case_ref, talmudic_context) do
+      {:ok, %{"status" => "resolved"} = interpretation} ->
+        Logger.info("BereishitCycle: Talmudic Engine resolved case #{case_ref}")
+        {:ok, Map.merge(data, %{
+          talmudic_interpretation: interpretation,
+          resolution: :talmudic,
+          stage: :erev_voker
+        })}
+
+      {:ok, %{"status" => "teiku"}} ->
+        # Still unresolved — escalate to Sanhedrin
+        Logger.info("BereishitCycle: Talmudic Engine returned teiku, escalating to Sanhedrin")
+        escalate_to_sanhedrin(data, context, case_ref, context_roots)
+
+      {:ok, interpretation} ->
+        # Challenged or other status — include and return
+        {:unresolved, Map.merge(data, %{
+          talmudic_interpretation: interpretation,
+          stage: :erev_voker
+        })}
+
+      {:error, reason} ->
+        Logger.warning("BereishitCycle: Talmudic routing failed: #{reason}")
+        {:unresolved, Map.put(data, :routing_error, reason)}
+    end
+  end
+
+  defp escalate_to_sanhedrin(data, _context, case_ref, context_roots) do
+    sanhedrin_params = %{
+      case_ref: case_ref,
+      complexity: :simple,
+      question: "Deliberate on unresolved interpretation of #{inspect(context_roots)}",
+      context: %{
+        roots: context_roots,
+        named_results: Map.get(data, :named_results, [])
+      },
+      requested_by: "bereishit_cycle"
+    }
+
+    case SanhedrinClient.submit_case(sanhedrin_params) do
+      {:ok, decision} ->
+        outcome = Map.get(decision, :outcome, :teiku)
+        Logger.info("BereishitCycle: Sanhedrin decided #{outcome} for #{case_ref}")
+
+        case outcome do
+          :accepted ->
+            {:ok, Map.merge(data, %{
+              sanhedrin_decision: decision,
+              resolution: :sanhedrin,
+              stage: :erev_voker
+            })}
+
+          _ ->
+            {:unresolved, Map.merge(data, %{
+              sanhedrin_decision: decision,
+              stage: :erev_voker
+            })}
+        end
+
+      {:error, reason} ->
+        Logger.warning("BereishitCycle: Sanhedrin escalation failed: #{reason}")
+        {:unresolved, Map.put(data, :escalation_error, reason)}
+    end
+  end
+
+  # --- L1 Fallback (when port is unavailable) ---
+
+  defp vayomer_fallback(state, context) do
+    roots =
+      context.verse.words
+      |> Enum.filter(fn w -> w.shoresh != nil end)
+      |> Enum.map(fn w ->
+        %{
+          shoresh: w.shoresh,
+          word: w.consonants,
+          position: w.position,
+          morph: w.morph,
+          gematria: Map.get(context.root_gematria, w.shoresh, 0)
+        }
+      end)
+
+    Map.merge(state, %{
+      roots_activated: roots,
+      teamim_parse: context.teamim_marks,
+      ketiv_qere: context.ketiv_qere,
+      stage: :vayomer
+    })
   end
 
   # --- Private helpers ---
 
+  # Get a value from a map that may have atom or string keys
+  # (atom keys from L1 fallback, string keys from JSON engine response)
+  defp get_result(map, atom_key, string_key, default) do
+    Map.get(map, atom_key, Map.get(map, string_key, default))
+  end
+
   defp extract_content(input) do
     Map.get(input, :content, "") |> String.trim()
   end

vav_pipeline/lib/vav_pipeline/sanhedrin_client.ex

@@ -0,0 +1,141 @@
+defmodule VavPipeline.SanhedrinClient do
+  @moduledoc """
+  Client interface to the Erlang Sanhedrin deliberation engine.
+
+  Construction spec §3.2: Native BEAM message passing.
+  Protocol: Erlang records via escalation:submit_case/1.
+
+  The Sanhedrin runs in the same BEAM VM. This module translates
+  Elixir maps into Erlang record tuples and calls the Sanhedrin
+  API directly.
+  """
+
+  require Logger
+
+  @doc """
+  Submit a case for deliberation.
+
+  Construction spec §3.2: submit_case with case_ref, complexity,
+  question, context, requested_by.
+
+  Returns {:ok, decision_map} or {:error, reason}.
+  """
+  @spec submit_case(map()) :: {:ok, map()} | {:error, String.t()}
+  def submit_case(case_params) do
+    case_ref = Map.get(case_params, :case_ref, generate_ref())
+    complexity = Map.get(case_params, :complexity, :simple)
+    question = Map.get(case_params, :question, "")
+    context = Map.get(case_params, :context, %{})
+    requested_by = Map.get(case_params, :requested_by, "pipeline")
+
+    # Build Erlang record tuple matching #sanhedrin_case{}
+    # Record: {sanhedrin_case, case_ref, complexity, question, context,
+    #          requested_by, submitted_at}
+    erlang_case =
+      {:sanhedrin_case, case_ref, complexity, question, context,
+       requested_by, :erlang.system_time(:millisecond)}
+
+    try do
+      case :escalation.submit_case(erlang_case) do
+        {:ok, decision_ref} ->
+          # Decision is stored asynchronously by the panel.
+          # Wait briefly for the result, then fetch from clerk.
+          wait_for_decision(decision_ref, case_ref)
+
+        {:error, {:halted, halt_type, reason}} ->
+          {:error, "Halted (#{halt_type}): #{reason}"}
+
+        {:error, reason} ->
+          {:error, "Submission failed: #{inspect(reason)}"}
+      end
+    catch
+      :exit, {:noproc, _} ->
+        Logger.warning("SanhedrinClient: Sanhedrin not running, returning teiku")
+        {:ok, stub_teiku(case_ref)}
+
+      kind, reason ->
+        Logger.error("SanhedrinClient: #{kind} — #{inspect(reason)}")
+        {:error, "Sanhedrin call failed: #{inspect(reason)}"}
+    end
+  end
+
+  @doc """
+  Get a previously stored decision from the clerk.
+  """
+  @spec get_decision(binary()) :: {:ok, map()} | {:error, :not_found}
+  def get_decision(case_ref) do
+    try do
+      case :clerk.get_decision(case_ref) do
+        {:ok, decision} ->
+          {:ok, decision_to_map(decision)}
+
+        :not_found ->
+          {:error, :not_found}
+      end
+    catch
+      :exit, {:noproc, _} ->
+        {:error, :not_found}
+    end
+  end
+
+  # --- Private ---
+
+  defp wait_for_decision(decision_ref, case_ref) do
+    # The panel deliberates asynchronously. Poll clerk briefly.
+    wait_for_decision(decision_ref, case_ref, 10, 500)
+  end
+
+  defp wait_for_decision(_decision_ref, case_ref, 0, _interval) do
+    # Timeout — return teiku (unresolved)
+    Logger.warning("SanhedrinClient: deliberation timeout for #{case_ref}")
+    {:ok, stub_teiku(case_ref)}
+  end
+
+  defp wait_for_decision(decision_ref, case_ref, retries, interval) do
+    Process.sleep(interval)
+
+    case get_decision(case_ref) do
+      {:ok, decision} ->
+        {:ok, decision}
+
+      {:error, :not_found} ->
+        wait_for_decision(decision_ref, case_ref, retries - 1, interval)
+    end
+  end
+
+  defp decision_to_map({:decision, case_ref, outcome, majority, minority, panel_size,
+                        delib_time, unanimous_check}) do
+    %{
+      case_ref: case_ref,
+      outcome: outcome,
+      majority: majority,
+      minority: minority,
+      panel_size: panel_size,
+      deliberation_time_ms: delib_time,
+      unanimous_check: unanimous_check
+    }
+  end
+
+  defp decision_to_map(other), do: other
+
+  defp stub_teiku(case_ref) do
+    %{
+      case_ref: case_ref,
+      outcome: :teiku,
+      majority: %{position: "Sanhedrin not available", reasoning: "Engine not running", torah_refs: [], vote_count: 0},
+      minority: %{position: "", reasoning: "", torah_refs: [], vote_count: 0},
+      panel_size: 0,
+      deliberation_time_ms: 0,
+      unanimous_check: :passed
+    }
+  end
+
+  defp generate_ref do
+    :crypto.strong_rand_bytes(16)
+    |> Base.encode16(case: :lower)
+    |> then(fn hex ->
+      <<a::binary-size(8), b::binary-size(4), c::binary-size(4), d::binary-size(4), e::binary-size(12)>> = hex
+      "#{a}-#{b}-#{c}-#{d}-#{e}"
+    end)
+  end
+end

vav_pipeline/lib/vav_pipeline/shoresh_engine_client.ex

@@ -2,54 +2,29 @@ defmodule VavPipeline.ShoreshEngineClient do
   @moduledoc """
   Client interface to the Haskell Shoresh Engine.
 
-  Architecture §3.3: NIF or port to Haskell engine.
+  Architecture §3.3: Erlang Port to Haskell engine.
   Construction spec §3.1: JSON protocol over port.
 
-  Phase 2: Stub that returns data from L1/ShemaContext.
-  Phase 3 (Step 14): Full port/NIF connection to shoresh-engine binary.
+  Delegates to ShoreshEnginePort (the live Haskell process).
+  Falls back to L1/ShemaContext extraction if port is unavailable.
   """
 
   require Logger
 
   @doc """
   Process input through the Shoresh Engine.
-
-  Phase 2: Uses L1 data directly.
-  Phase 3: Sends JSON request to Haskell port.
+  Sends JSON request to Haskell port; falls back to L1 data.
   """
   @spec process(String.t(), map()) :: {:ok, map()} | {:error, String.t()}
-  def process(_raw_text, context) do
-    # Phase 2 stub — extract from ShemaContext
-    roots =
-      context.verse.words
-      |> Enum.filter(fn w -> w.shoresh != nil end)
-      |> Enum.map(fn w ->
-        %{
-          shoresh: w.shoresh,
-          binyan: infer_binyan(w.morph),
-          domain: :unknown,
-          torah_refs: [%{book: context.verse.book, chapter: context.verse.chapter, verse: context.verse.verse}],
-          gematria_value: Map.get(context.root_gematria, w.shoresh, 0)
-        }
-      end)
-
-    {:ok,
-     %{
-       roots_activated: roots,
-       parse_tree: nil,
-       dual_readings: context.ketiv_qere
-     }}
+  def process(raw_text, context) do
+    call_port(:process, raw_text, context)
   end
 
   @doc """
   Separate utterance using te'amim marks.
-
-  Phase 2: Returns groups based on melekh positions.
-  Phase 3: Delegates to Haskell Havdalah module.
   """
   @spec separate(map(), [map()]) :: [map()]
   def separate(utterance, teamim_marks) do
-    # Phase 2 stub — group by melekh divisions
     melekh_positions =
       teamim_marks
       |> Enum.filter(fn m -> m.role == "melekh" end)
@@ -60,29 +35,20 @@ defmodule VavPipeline.ShoreshEngineClient do
 
   @doc """
   Name a distinction.
-
-  Phase 2: Returns the primary root as the name.
-  Phase 3: Delegates to Haskell Kriah module.
   """
   @spec name(map()) :: map()
   def name(distinction) do
     Map.put(distinction, :name, Map.get(distinction, :shoresh, "unnamed"))
   end
 
-  # Infer binyan from morphological code
-  # H = Hebrew, V = verb, then binyan indicator
-  defp infer_binyan(morph) when is_binary(morph) do
-    cond do
-      String.contains?(morph, "Vq") -> :qal
-      String.contains?(morph, "VN") -> :nifal
-      String.contains?(morph, "Vp") -> :piel
-      String.contains?(morph, "VP") -> :pual
-      String.contains?(morph, "Vh") -> :hifil
-      String.contains?(morph, "VH") -> :hufal
-      String.contains?(morph, "Vt") -> :hitpael
-      true -> :qal
+  # --- Private ---
+
+  defp call_port(operation, text, context) do
+    if Process.whereis(VavPipeline.ShoreshEnginePort) do
+      apply(VavPipeline.ShoreshEnginePort, operation, [text, context])
+    else
+      {:error, "port not started"}
     end
   end
 
-  defp infer_binyan(_), do: :qal
 end

vav_pipeline/lib/vav_pipeline/shoresh_engine_port.ex

@@ -0,0 +1,185 @@
+defmodule VavPipeline.ShoreshEnginePort do
+  @moduledoc """
+  Erlang Port interface to the Haskell Shoresh Engine.
+
+  Step 14 (Architecture §4): "Connect components: Elixir pipeline
+  calls Haskell engine."
+
+  Construction spec §3.1: JSON protocol over Erlang Port.
+  Protocol: JSON line in, JSON line out.
+
+  The port launches `shoresh-engine server` as an OS process.
+  Communication is through stdin/stdout with line-delimited JSON.
+
+  Port crash isolation: if the Haskell process crashes, the BEAM
+  VM continues running. The GenServer will restart the port.
+  """
+
+  use GenServer
+
+  require Logger
+
+  @default_binary "shoresh-engine"
+  @call_timeout 5_000
+
+  # --- Client API ---
+
+  def start_link(opts \\ []) do
+    GenServer.start_link(__MODULE__, opts, name: __MODULE__)
+  end
+
+  @doc """
+  Send a process request to the Shoresh Engine.
+  Construction spec §3.1 — operation "process".
+  """
+  @spec process(String.t(), map()) :: {:ok, map()} | {:error, String.t()}
+  def process(text, context \\ %{}) do
+    request = build_request("process", text, context)
+    call_engine(request)
+  end
+
+  @doc """
+  Send a separate request to the Shoresh Engine.
+  Construction spec §3.1 — operation "separate".
+  """
+  @spec separate(String.t(), map()) :: {:ok, map()} | {:error, String.t()}
+  def separate(text, context \\ %{}) do
+    request = build_request("separate", text, context)
+    call_engine(request)
+  end
+
+  @doc """
+  Send a name request to the Shoresh Engine.
+  Construction spec §3.1 — operation "name".
+  """
+  @spec name(String.t(), map()) :: {:ok, map()} | {:error, String.t()}
+  def name(text, context \\ %{}) do
+    request = build_request("name", text, context)
+    call_engine(request)
+  end
+
+  @doc """
+  Send an evaluate request to the Shoresh Engine.
+  Construction spec §3.1 — operation "evaluate".
+  """
+  @spec evaluate(String.t(), map()) :: {:ok, map()} | {:error, String.t()}
+  def evaluate(text, context \\ %{}) do
+    request = build_request("evaluate", text, context)
+    call_engine(request)
+  end
+
+  # --- GenServer callbacks ---
+
+  @impl true
+  def init(opts) do
+    binary = Keyword.get(opts, :binary, @default_binary)
+    cwd = Keyword.get(opts, :cwd, nil)
+
+    port = open_port(binary, cwd)
+
+    {:ok, %{port: port, binary: binary, cwd: cwd, pending: nil}}
+  end
+
+  @impl true
+  def handle_call({:request, json_line}, from, state) do
+    Port.command(state.port, json_line <> "\n")
+    {:noreply, %{state | pending: from}}
+  end
+
+  @impl true
+  def handle_info({port, {:data, {:eol, data}}}, %{port: port, pending: from} = state)
+      when from != nil do
+    response = data |> IO.iodata_to_binary() |> Jason.decode()
+
+    reply =
+      case response do
+        {:ok, %{"status" => "ok", "result" => result}} ->
+          {:ok, result}
+
+        {:ok, %{"status" => "error", "error" => error}} ->
+          {:error, "Engine error: #{error["code"]} — #{error["message"]}"}
+
+        {:error, decode_error} ->
+          {:error, "JSON decode error: #{inspect(decode_error)}"}
+      end
+
+    GenServer.reply(from, reply)
+    {:noreply, %{state | pending: nil}}
+  end
+
+  def handle_info({port, {:exit_status, status}}, %{port: port} = state) do
+    Logger.error("ShoreshEnginePort: process exited with status #{status}")
+
+    if state.pending do
+      GenServer.reply(state.pending, {:error, "Engine process exited (status #{status})"})
+    end
+
+    # Restart the port
+    new_port = open_port(state.binary, state.cwd)
+    {:noreply, %{state | port: new_port, pending: nil}}
+  end
+
+  def handle_info({port, {:data, {:noeol, _partial}}}, %{port: port} = state) do
+    # Partial line — wait for more data
+    {:noreply, state}
+  end
+
+  def handle_info(msg, state) do
+    Logger.warning("ShoreshEnginePort: unexpected message: #{inspect(msg)}")
+    {:noreply, state}
+  end
+
+  @impl true
+  def terminate(_reason, state) do
+    Port.close(state.port)
+    :ok
+  end
+
+  # --- Private ---
+
+  defp open_port(binary, cwd) do
+    port_opts = [
+      :binary,
+      :exit_status,
+      {:line, 1_048_576},
+      {:args, ["server"]}
+    ]
+
+    port_opts =
+      if cwd do
+        [{:cd, cwd} | port_opts]
+      else
+        port_opts
+      end
+
+    Port.open({:spawn_executable, binary}, port_opts)
+  end
+
+  defp build_request(operation, text, context) do
+    request = %{
+      "operation" => operation,
+      "input" => %{
+        "text" => text
+      }
+    }
+
+    # Add scope from context if present
+    request =
+      if Map.has_key?(context, :verse) do
+        verse = context.verse
+        put_in(request, ["input", "scope"], %{
+          "book" => to_string(verse.book),
+          "chapter" => verse.chapter,
+          "verse" => verse.verse
+        })
+      else
+        request
+      end
+
+    Jason.encode!(request)
+  end
+
+  defp call_engine(json_line) do
+    GenServer.call(__MODULE__, {:request, json_line}, @call_timeout)
+  end
+end

vav_pipeline/lib/vav_pipeline/talmudic_engine_client.ex

@@ -4,73 +4,102 @@ defmodule VavPipeline.TalmudicEngineClient do
 
   Construction spec §3.3:
   Protocol: JSON over Unix domain socket
-  Socket: /var/run/shorashim/talmudic.sock
 
-  Phase 2: Stub that returns placeholder responses.
-  Phase 3 (Step 14): Full socket connection to talmudic_engine.
+  Delegates to TalmudicEngineSocket (the live Prolog process).
+  Falls back to stub responses if socket is unavailable.
   """
 
   require Logger
 
-  # Phase 3: Unix domain socket path
-  # @socket_path Application.compile_env(
-  #                :vav_pipeline,
-  #                :talmudic_socket,
-  #                "/var/run/shorashim/talmudic.sock"
-  #              )
-
   @doc """
   Send an interpretation request to the Talmudic Engine.
-
-  Phase 2: Returns a stub response.
-  Phase 3: Sends JSON over Unix socket.
   """
   @spec interpret(String.t(), map()) :: {:ok, map()} | {:error, String.t()}
-  def interpret(case_ref, _context) do
-    Logger.info("TalmudicEngineClient: interpret stub for #{case_ref}")
+  def interpret(case_ref, context) do
+    case call_socket(:interpret, case_ref, context) do
+      {:ok, result} ->
+        {:ok, result}
 
-    {:ok,
-     %{
-       status: :teiku,
-       interpretation: %{
-         text: "Phase 2 stub — Talmudic Engine not yet connected",
-         prolog_form: "stub(#{case_ref})",
-         grounding: [],
-         confidence: :low
-       },
-       challenge: nil
-     }}
+      {:error, reason} ->
+        Logger.warning("TalmudicEngineClient: socket unavailable (#{reason}), using stub")
+        {:ok, stub_interpretation(case_ref)}
+    end
   end
 
   @doc """
   Send a challenge request.
   """
   @spec challenge(String.t(), map()) :: {:ok, map()} | {:error, String.t()}
-  def challenge(case_ref, _context) do
-    Logger.info("TalmudicEngineClient: challenge stub for #{case_ref}")
+  def challenge(case_ref, context) do
+    case call_socket(:challenge, case_ref, context) do
+      {:ok, result} ->
+        {:ok, result}
 
-    {:ok,
-     %{
-       status: :teiku,
-       challenge: %{
-         kushya: "Phase 2 stub — no challenge generated",
-         contradicting: nil,
-         strength: :weak
-       }
-     }}
+      {:error, reason} ->
+        Logger.warning("TalmudicEngineClient: socket unavailable (#{reason}), using stub")
+        {:ok, stub_challenge(case_ref)}
+    end
   end
 
   @doc """
   Send a resolution request.
   """
   @spec resolve(String.t(), map()) :: {:ok, map()} | {:error, String.t()}
-  def resolve(case_ref, _context) do
-    Logger.info("TalmudicEngineClient: resolve stub for #{case_ref}")
+  def resolve(case_ref, context) do
+    case call_socket(:resolve, case_ref, context) do
+      {:ok, result} ->
+        {:ok, result}
 
-    {:ok,
-     %{
-       status: :teiku,
-       resolution: "Phase 2 stub — resolution deferred"
-     }}
+      {:error, reason} ->
+        Logger.warning("TalmudicEngineClient: socket unavailable (#{reason}), using stub")
+        {:ok, stub_resolution(case_ref)}
+    end
+  end
+
+  # --- Private ---
+
+  defp call_socket(operation, case_ref, context) do
+    if Process.whereis(VavPipeline.TalmudicEngineSocket) do
+      apply(VavPipeline.TalmudicEngineSocket, operation, [case_ref, context])
+    else
+      {:error, "socket not started"}
+    end
+  end
+
+  defp stub_interpretation(case_ref) do
+    %{
+      "status" => "teiku",
+      "interpretation" => %{
+        "text" => "Talmudic Engine not connected — teiku",
+        "prolog_form" => "stub(#{case_ref})",
+        "grounding" => [],
+        "confidence" => "low"
+      },
+      "challenge" => nil
+    }
+  end
+
+  defp stub_challenge(_case_ref) do
+    %{
+      "status" => "teiku",
+      "challenge" => %{
+        "kushya" => "No challenge generated — engine not connected",
+        "contradicting" => "",
+        "strength" => "weak"
+      }
+    }
+  end
+
+  defp stub_resolution(case_ref) do
+    %{
+      "status" => "teiku",
+      "interpretation" => %{
+        "text" => "Resolution deferred — engine not connected",
+        "prolog_form" => "stub(#{case_ref})",
+        "grounding" => [],
+        "confidence" => "low"
+      },
+      "challenge" => nil
+    }
   end
 end

vav_pipeline/lib/vav_pipeline/talmudic_engine_socket.ex

@@ -0,0 +1,172 @@
+defmodule VavPipeline.TalmudicEngineSocket do
+  @moduledoc """
+  Unix domain socket client to the SWI-Prolog Talmudic Engine.
+
+  Step 14 (Architecture §4): "Connect components: routes to
+  Prolog reasoning."
+
+  Construction spec §3.3: JSON over Unix domain socket.
+  Socket: /tmp/shorashim/talmudic.sock (dev)
+          /var/run/shorashim/talmudic.sock (production)
+
+  Connection model: connect, send one JSON line, read one JSON
+  line response, keep connection for subsequent requests.
+  Reconnects on connection loss.
+  """
+
+  use GenServer
+
+  require Logger
+
+  @default_socket_path "/tmp/shorashim/talmudic.sock"
+  @call_timeout 10_000
+  @connect_timeout 5_000
+
+  # --- Client API ---
+
+  def start_link(opts \\ []) do
+    GenServer.start_link(__MODULE__, opts, name: __MODULE__)
+  end
+
+  @doc """
+  Send an interpret request to the Talmudic Engine.
+  Construction spec §3.3 — action "interpret".
+  """
+  @spec interpret(String.t(), map()) :: {:ok, map()} | {:error, String.t()}
+  def interpret(case_ref, context) do
+    request = build_request("interpret", case_ref, context)
+    call_engine(request)
+  end
+
+  @doc """
+  Send a challenge request to the Talmudic Engine.
+  Construction spec §3.3 — action "challenge".
+  """
+  @spec challenge(String.t(), map()) :: {:ok, map()} | {:error, String.t()}
+  def challenge(case_ref, context) do
+    request = build_request("challenge", case_ref, context)
+    call_engine(request)
+  end
+
+  @doc """
+  Send a resolve request to the Talmudic Engine.
+  Construction spec §3.3 — action "resolve".
+  """
+  @spec resolve(String.t(), map()) :: {:ok, map()} | {:error, String.t()}
+  def resolve(case_ref, context) do
+    request = build_request("resolve", case_ref, context)
+    call_engine(request)
+  end
+
+  @doc """
+  Send a halt_check request to the Talmudic Engine.
+  """
+  @spec halt_check(map()) :: {:ok, map()} | {:error, String.t()}
+  def halt_check(action) do
+    request = Jason.encode!(%{
+      "action" => "halt_check",
+      "check" => action
+    })
+    call_engine(request)
+  end
+
+  # --- GenServer callbacks ---
+
+  @impl true
+  def init(opts) do
+    socket_path = Keyword.get(opts, :socket_path, @default_socket_path)
+    {:ok, %{socket_path: socket_path, socket: nil}}
+  end
+
+  @impl true
+  def handle_call({:request, json_line}, _from, state) do
+    case ensure_connected(state) do
+      {:ok, state} ->
+        case send_and_receive(state.socket, json_line) do
+          {:ok, response} ->
+            {:reply, {:ok, response}, state}
+
+          {:error, reason} ->
+            Logger.warning("TalmudicEngineSocket: send/receive failed: #{inspect(reason)}")
+            # Connection broken — close and retry once
+            :gen_tcp.close(state.socket)
+            state = %{state | socket: nil}
+
+            case ensure_connected(state) do
+              {:ok, state} ->
+                case send_and_receive(state.socket, json_line) do
+                  {:ok, response} -> {:reply, {:ok, response}, state}
+                  {:error, reason2} -> {:reply, {:error, "Retry failed: #{inspect(reason2)}"}, %{state | socket: nil}}
+                end
+
+              {:error, reason2} ->
+                {:reply, {:error, "Reconnect failed: #{inspect(reason2)}"}, state}
+            end
+
+        end
+
+      {:error, reason} ->
+        {:reply, {:error, "Cannot connect: #{inspect(reason)}"}, state}
+    end
+  end
+
+  @impl true
+  def terminate(_reason, state) do
+    if state.socket do
+      :gen_tcp.close(state.socket)
+    end
+    :ok
+  end
+
+  # --- Private ---
+
+  defp ensure_connected(%{socket: nil, socket_path: path} = state) do
+    case :gen_tcp.connect({:local, path}, 0, [:binary, active: false, packet: :line], @connect_timeout) do
+      {:ok, socket} ->
+        {:ok, %{state | socket: socket}}
+
+      {:error, reason} ->
+        {:error, reason}
+    end
+  end
+
+  defp ensure_connected(state), do: {:ok, state}
+
+  defp send_and_receive(socket, json_line) do
+    case :gen_tcp.send(socket, json_line <> "\n") do
+      :ok ->
+        case :gen_tcp.recv(socket, 0, @call_timeout) do
+          {:ok, data} ->
+            line = String.trim_trailing(data, "\n")
+            case Jason.decode(line) do
+              {:ok, parsed} -> {:ok, parsed}
+              {:error, err} -> {:error, "JSON decode: #{inspect(err)}"}
+            end
+
+          {:error, reason} ->
+            {:error, reason}
+        end
+
+      {:error, reason} ->
+        {:error, reason}
+    end
+  end
+
+  defp build_request(action, case_ref, context) do
+    roots = Map.get(context, :context_roots, [])
+    question = Map.get(context, :question, "")
+
+    Jason.encode!(%{
+      "action" => action,
+      "case" => %{
+        "ref" => case_ref,
+        "question" => question,
+        "context_roots" => roots
+      }
+    })
+  end
+
+  defp call_engine(json_line) do
+    GenServer.call(__MODULE__, {:request, json_line}, @call_timeout)
+  end
+end