From c67672777bf9c008585afd69e86a345628d35f4e Mon Sep 17 00:00:00 2001 From: Scott Karp Date: Thu, 12 Mar 2026 17:39:21 -0400 Subject: [PATCH] =?UTF-8?q?Step=2014:=20Connect=20components=20=E2=80=94?= =?UTF-8?q?=20chamishim=20karsay=20zahav?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- shoresh-engine/shoresh-engine.cabal | 1 + shoresh-engine/src/Main.hs | 3 + shoresh-engine/src/Shoresh/Lexicon.hs | 2 + shoresh-engine/src/Shoresh/Server.hs | 301 ++++++++++++++++++ shoresh-engine/src/Shoresh/Types.hs | 19 +- talmudic_engine/src/interface/json_bridge.pl | 194 +++++++++++ .../src/interface/socket_server.pl | 145 +++++++++ talmudic_engine/src/main.pl | 2 + vav_pipeline/config/config.exs | 8 +- vav_pipeline/lib/vav_pipeline/application.ex | 38 ++- .../lib/vav_pipeline/bereishit_cycle.ex | 164 ++++++++-- .../lib/vav_pipeline/sanhedrin_client.ex | 141 ++++++++ .../lib/vav_pipeline/shoresh_engine_client.ex | 60 +--- .../lib/vav_pipeline/shoresh_engine_port.ex | 185 +++++++++++ .../vav_pipeline/talmudic_engine_client.ex | 117 ++++--- .../vav_pipeline/talmudic_engine_socket.ex | 172 ++++++++++ vav_pipeline/mix.exs | 3 +- vav_pipeline/mix.lock | 1 + 18 files changed, 1435 insertions(+), 121 deletions(-) create mode 100644 shoresh-engine/src/Shoresh/Server.hs create mode 100644 talmudic_engine/src/interface/json_bridge.pl create mode 100644 talmudic_engine/src/interface/socket_server.pl create mode 100644 vav_pipeline/lib/vav_pipeline/sanhedrin_client.ex create mode 100644 vav_pipeline/lib/vav_pipeline/shoresh_engine_port.ex create mode 100644 vav_pipeline/lib/vav_pipeline/talmudic_engine_socket.ex diff --git a/shoresh-engine/shoresh-engine.cabal b/shoresh-engine/shoresh-engine.cabal index 6621d800..d270bda7 100644 --- a/shoresh-engine/shoresh-engine.cabal +++ b/shoresh-engine/shoresh-engine.cabal @@ -32,6 +32,7 @@ library Shoresh.Havdalah Shoresh.Kriah Shoresh.KiTov + Shoresh.Server Luchot.Firmware hs-source-dirs: src default-language: Haskell2010 diff --git a/shoresh-engine/src/Main.hs b/shoresh-engine/src/Main.hs index 3e59b447..458ac026 100644 --- a/shoresh-engine/src/Main.hs +++ b/shoresh-engine/src/Main.hs @@ -27,6 +27,7 @@ import Shoresh.Teamim import Shoresh.KetivQere import Shoresh.Lexicon import Shoresh.Graph +import Shoresh.Server (runServer) import Luchot.Firmware main :: IO () @@ -34,6 +35,7 @@ main = do args <- getArgs case args of ["verify"] -> runVerification + ["server"] -> runServer ["verse", book, ch, vs] -> runVerse book ch vs ["root", rootText] -> runRoot rootText ["gematria", word] -> runGematria word @@ -45,6 +47,7 @@ printUsage = do putStrLn "" putStrLn "Usage:" putStrLn " shoresh-engine verify Run Step 7 verification" + putStrLn " shoresh-engine server JSON-line server (Erlang Port mode)" putStrLn " shoresh-engine verse Read and parse a verse from L1" putStrLn " shoresh-engine root Activate a root through all 7 binyanim" putStrLn " shoresh-engine gematria Compute gematria of a Hebrew word" diff --git a/shoresh-engine/src/Shoresh/Lexicon.hs b/shoresh-engine/src/Shoresh/Lexicon.hs index 63950685..810f04f0 100644 --- a/shoresh-engine/src/Shoresh/Lexicon.hs +++ b/shoresh-engine/src/Shoresh/Lexicon.hs @@ -79,6 +79,7 @@ data L1Word = L1Word , l1wShoresh :: !(Maybe Text) } deriving (Show, Generic) +instance ToJSON L1Word instance FromJSON L1Word where parseJSON = withObject "L1Word" $ \v -> L1Word <$> v .: "position" @@ -100,6 +101,7 @@ data L1Verse = L1Verse , l1vWords :: ![L1Word] } deriving (Show, Generic) +instance ToJSON L1Verse instance FromJSON L1Verse where parseJSON = withObject "L1Verse" $ \v -> L1Verse <$> v .: "ref" diff --git a/shoresh-engine/src/Shoresh/Server.hs b/shoresh-engine/src/Shoresh/Server.hs new file mode 100644 index 00000000..7930c8de --- /dev/null +++ b/shoresh-engine/src/Shoresh/Server.hs @@ -0,0 +1,301 @@ +{-# LANGUAGE DeriveGeneric #-} +{-# LANGUAGE OverloadedStrings #-} + +-- | Server — JSON-line port interface for pipeline integration. +-- +-- Step 14 (Architecture §4): "Connect components: Elixir pipeline +-- calls Haskell engine." +-- +-- Construction spec §3.1: Protocol is Erlang Port (JSON). +-- The engine reads one JSON line from stdin, processes it, +-- writes one JSON line to stdout, and repeats. +-- +-- Buffering is set to LineBuffering — critical for Erlang Port +-- communication. Haskell defaults to block-buffering on non-TTY +-- stdout, which would deadlock the port. +module Shoresh.Server + ( runServer + , handleRequest + ) where + +import qualified Data.ByteString.Char8 as BS8 +import qualified Data.ByteString.Lazy as BL +import qualified Data.ByteString.Lazy.Char8 as BLC +import qualified Data.Map.Strict as Map +import Data.Text (Text) +import qualified Data.Text as T +import Data.Aeson +import Data.Maybe (mapMaybe) +import System.IO (hSetBuffering, stdin, stdout, stderr, + BufferMode(..), hFlush, hIsEOF) +import System.Directory (doesFileExist) +import Control.Exception (catch, SomeException) + +import Shoresh.Types +import Shoresh.Gematria +import Shoresh.Binyanim +import Shoresh.Teamim +import Shoresh.Lexicon + + +-- | Run the JSON-line server loop. +-- Called by Main.hs when invoked with "server" argument. +runServer :: IO () +runServer = do + -- Critical: set line buffering on stdin/stdout. + -- Haskell defaults to block-buffering on non-TTY, which would + -- deadlock the Erlang Port. + hSetBuffering stdin LineBuffering + hSetBuffering stdout LineBuffering + hSetBuffering stderr LineBuffering + serverLoop + + +-- | Main server loop: read line, process, write line, repeat. +serverLoop :: IO () +serverLoop = do + eof <- hIsEOF stdin + if eof + then return () + else do + line <- BL.fromStrict <$> BS8.getLine + response <- processLine line + BLC.putStrLn (encode response) + hFlush stdout + serverLoop + + +-- | Process a single JSON line and return a response. +processLine :: BL.ByteString -> IO Value +processLine line = case eitherDecode line of + Left err -> return $ errorResponse "PARSE_ERROR" (T.pack err) + Right req -> handleRequest req + `catch` \e -> return $ errorResponse "INTERNAL_ERROR" + (T.pack $ show (e :: SomeException)) + + +-- | Handle a parsed request and produce a response. +handleRequest :: Value -> IO Value +handleRequest val = case parseOperation val of + Nothing -> return $ errorResponse "INVALID_OPERATION" "Missing or invalid 'operation' field" + Just "process" -> handleProcess val + Just "separate" -> handleSeparate val + Just "name" -> handleName val + Just "evaluate" -> handleEvaluate val + Just op -> return $ errorResponse "UNKNOWN_OPERATION" ("Unknown operation: " <> op) + + +-- | Extract the operation string from a request. +parseOperation :: Value -> Maybe Text +parseOperation (Object obj) = case obj .: "operation" of + Success t -> Just t + _ -> Nothing + where + (.:) o k = case Map.lookup k (toMap o) of + Just (String s) -> Success s + _ -> Error "not found" + toMap = \o -> case fromJSON (Object o) :: Result (Map.Map Text Value) of + Success m -> m + Error _ -> Map.empty +parseOperation _ = Nothing + + +-- | Handle "process" operation — root activation, te'amim parse, gematria. +-- Construction spec §3.1: Takes text + scope, returns roots_activated, +-- parse_tree, dual_readings. +handleProcess :: Value -> IO Value +handleProcess val = do + let reqText = extractText val + mscope = extractScope val + + -- Parse roots from input text + let roots = extractRoots reqText + + -- Build activated roots with gematria and binyanim + let activatedRoots = map activateRoot roots + + -- If scope is provided, load verse from L1 and parse te'amim + (parseTree, dualReadings) <- case mscope of + Nothing -> return (Null, toJSON ([] :: [Value])) + Just (book, ch, vs) -> do + let path = "l1/torah/" ++ T.unpack (bookToText book) + ++ "/" ++ show ch ++ "/" ++ show vs ++ ".json" + exists <- doesFileExist path + if not exists + then return (Null, toJSON ([] :: [Value])) + else do + mverse <- loadL1Verse path + case mverse of + Nothing -> return (Null, toJSON ([] :: [Value])) + Just verse -> do + let annotated = mapMaybe annotateL1Word (l1vWords verse) + tree = readTeamim annotated + ptJson = treeToJson tree + drJson = toJSON ([] :: [Value]) -- Phase 3: extract ketiv/qere + return (ptJson, drJson) + + return $ object + [ "status" .= ("ok" :: Text) + , "result" .= object + [ "roots_activated" .= activatedRoots + , "parse_tree" .= parseTree + , "dual_readings" .= dualReadings + ] + ] + + +-- | Handle "separate" operation — te'amim-based separation. +-- Construction spec §3.1 operation "separate". +handleSeparate :: Value -> IO Value +handleSeparate _val = do + -- Phase 2: returns stub separation + -- Full implementation requires the annotated words from context + return $ object + [ "status" .= ("ok" :: Text) + , "result" .= object + [ "groups" .= ([] :: [Value]) + ] + ] + + +-- | Handle "name" operation — assign names to categories. +handleName :: Value -> IO Value +handleName _val = do + return $ object + [ "status" .= ("ok" :: Text) + , "result" .= object + [ "name" .= ("unnamed" :: Text) + ] + ] + + +-- | Handle "evaluate" operation — ki tov check. +handleEvaluate :: Value -> IO Value +handleEvaluate _val = do + return $ object + [ "status" .= ("ok" :: Text) + , "result" .= object + [ "evaluation" .= ("teiku" :: Text) + ] + ] + + +-- ============================================================ +-- Helpers +-- ============================================================ + +-- | Extract text from the input field of a request. +extractText :: Value -> Text +extractText (Object obj) = case Map.lookup "input" (toMap obj) of + Just (Object inp) -> case Map.lookup "text" (toMap inp) of + Just (String t) -> t + _ -> "" + _ -> "" + where + toMap o = case fromJSON (Object o) :: Result (Map.Map Text Value) of + Success m -> m + Error _ -> Map.empty +extractText _ = "" + + +-- | Extract scope (book, chapter, verse) from the input field. +extractScope :: Value -> Maybe (Book, Int, Int) +extractScope (Object obj) = do + inp <- case Map.lookup "input" (toMap obj) of + Just (Object i) -> Just (toMap i) + _ -> Nothing + scopeObj <- case Map.lookup "scope" inp of + Just (Object s) -> Just (toMap s) + _ -> Nothing + bookVal <- Map.lookup "book" scopeObj + bookText <- case bookVal of + String s -> Just s + _ -> Nothing + book <- textToBook bookText + ch <- case Map.lookup "chapter" scopeObj of + Just (Number n) -> Just (floor n) + _ -> Nothing + vs <- case Map.lookup "verse" scopeObj of + Just (Number n) -> Just (floor n) + _ -> Nothing + return (book, ch, vs) + where + toMap o = case fromJSON (Object o) :: Result (Map.Map Text Value) of + Success m -> m + Error _ -> Map.empty +extractScope _ = Nothing + + +-- | Extract Hebrew roots from input text. +extractRoots :: Text -> [Shoresh] +extractRoots t = + let words' = T.words t + -- Try each word as a potential root + candidates = mapMaybe textToShoresh words' + -- Also try the whole text as a single root + wholeRoot = textToShoresh (T.filter (\c -> c >= '\x05D0' && c <= '\x05EA') t) + in case candidates of + [] -> case wholeRoot of + Just r -> [r] + Nothing -> [] + rs -> rs + + +-- | Activate a root: expand through binyanim, compute gematria, find neighbors. +activateRoot :: Shoresh -> Value +activateRoot sh = + let expanded = expand sh + gv = gematriaRoot sh + consonantNames = map consonantName [c1 sh, c2 sh, c3 sh] + defaultBinyan = case Map.lookup Qal expanded of + Just p -> T.pack (show (Shoresh.Types.binyan p)) + Nothing -> "Qal" + in object + [ "shoresh" .= consonantNames + , "shoresh_hebrew" .= shoreshToText sh + , "binyan" .= defaultBinyan + , "domain" .= ("unknown" :: Text) -- Phase 3: domain lookup + , "torah_refs" .= ([] :: [Value]) -- Phase 3: lexicon lookup + , "gematria_value" .= gv + , "gematria_neighbors" .= ([] :: [Value]) -- Phase 3: lexicon needed + , "neighbors" .= ([] :: [Value]) -- Phase 3: lexicon needed + ] + + +-- | Convert an L1 word to an AnnotatedWord for te'amim parsing. +annotateL1Word :: L1Word -> Maybe AnnotatedWord +annotateL1Word w = case l1wTeamim w of + [] -> Nothing + (t:_) -> case taamFromName t of + Nothing -> Nothing + Just taam -> Just AnnotatedWord + { awWord = l1wText w + , awTaam = taam + , awRole = taamRole taam + } + + +-- | Convert a ParsedVerse tree to JSON. +treeToJson :: ParsedVerse -> Value +treeToJson (Leaf w) = object + [ "type" .= ("leaf" :: Text) + , "word" .= awWord w + , "taam" .= T.pack (show (awTaam w)) + ] +treeToJson (Branch t l r) = object + [ "type" .= ("branch" :: Text) + , "taam" .= T.pack (show t) + , "left" .= treeToJson l + , "right" .= treeToJson r + ] + + +-- | Build an error response. +errorResponse :: Text -> Text -> Value +errorResponse code msg = object + [ "status" .= ("error" :: Text) + , "error" .= object + [ "code" .= code + , "message" .= msg + ] + ] diff --git a/shoresh-engine/src/Shoresh/Types.hs b/shoresh-engine/src/Shoresh/Types.hs index 4dce606b..f364accd 100644 --- a/shoresh-engine/src/Shoresh/Types.hs +++ b/shoresh-engine/src/Shoresh/Types.hs @@ -49,7 +49,7 @@ import Data.Text (Text) import qualified Data.Text as T import Data.Map.Strict (Map) import GHC.Generics (Generic) -import Data.Aeson (FromJSON(..), ToJSON(..), withText) +import Data.Aeson (FromJSON(..), ToJSON(..), FromJSONKey(..), ToJSONKey(..), withText) -- | The 22 Hebrew consonants. -- These are the building blocks. No more, no less. @@ -145,6 +145,8 @@ data Binyan instance ToJSON Binyan instance FromJSON Binyan +instance ToJSONKey Binyan +instance FromJSONKey Binyan -- | All seven binyanim as a list. allBinyanim :: [Binyan] @@ -298,6 +300,9 @@ data ShoreshEntry = ShoreshEntry , seGlosses :: !(Map Binyan Text) } deriving (Eq, Show, Generic) +instance ToJSON ShoreshEntry +instance FromJSON ShoreshEntry + -- | Raw input to the processing pipeline. data Input = Input @@ -305,18 +310,27 @@ data Input = Input , inputSource :: !Text } deriving (Show, Generic) +instance ToJSON Input +instance FromJSON Input + -- | A distinguished category after havdalah. data Category = Category { catName :: !Text , catContent :: !Text } deriving (Show, Generic) +instance ToJSON Category +instance FromJSON Category + -- | A processed result ready for evaluation. data ProcessedResult = ProcessedResult { prCategories :: ![Category] , prRoots :: ![ShoreshEntry] } deriving (Show, Generic) +instance ToJSON ProcessedResult +instance FromJSON ProcessedResult + -- | The evaluation outcome. -- Tov: fits the specification. -- LoTov: does not fit, with reason. @@ -326,3 +340,6 @@ data Evaluation | LoTov !Text | Teiku deriving (Eq, Show, Generic) + +instance ToJSON Evaluation +instance FromJSON Evaluation diff --git a/talmudic_engine/src/interface/json_bridge.pl b/talmudic_engine/src/interface/json_bridge.pl new file mode 100644 index 00000000..3cf7ad32 --- /dev/null +++ b/talmudic_engine/src/interface/json_bridge.pl @@ -0,0 +1,194 @@ +%% json_bridge.pl — JSON ↔ Prolog term conversion for Talmudic Engine. +%% +%% Construction §3.3: Bidirectional conversion between the JSON schema +%% defined in the spec and internal Prolog terms used by api.pl. +%% +%% Inbound: JSON dict → Prolog request term +%% Outbound: Prolog response term → JSON dict + +:- module(json_bridge, [ + json_to_request/2, + response_to_json/2, + parse_root_name/2 +]). + +:- use_module(library(json)). +:- use_module(library(lists)). + +%% ============================================================ +%% INBOUND: JSON → Prolog request +%% ============================================================ + +%% Convert a JSON dict (as parsed by SWI-Prolog's json library) to a +%% Prolog request term compatible with api:process_request/2. +%% +%% JSON format (Construction §3.3): +%% { "action": "interpret", "case": { "ref": "case-uuid", +%% "facts": [...], "question": "...", "context_roots": [...] } } +%% +%% Prolog format: +%% request(interpret, CaseRef, Domain, Roots, Question) + +json_to_request(Dict, request(Action, CaseRef, Domain, Roots, Question)) :- + get_dict(action, Dict, ActionStr), + atom_string(Action, ActionStr), + get_dict(case, Dict, CaseDict), + get_dict(ref, CaseDict, RefStr), + atom_string(CaseRef, RefStr), + %% Extract question + ( get_dict(question, CaseDict, QuestionStr) + -> atom_string(Question, QuestionStr) + ; Question = none + ), + %% Extract context_roots and derive domain + ( get_dict(context_roots, CaseDict, RootNames) + -> maplist(parse_root_name, RootNames, Roots) + ; Roots = [] + ), + %% Derive domain from roots (first root's known domain, or 'general') + ( Roots = [FirstRoot|_], + root_to_domain(FirstRoot, D) + -> Domain = D + ; Domain = general + ). + +%% Handle halt_check action (different format — no case, just action) +json_to_request(Dict, request(halt_check, Action)) :- + get_dict(action, Dict, "halt_check"), + get_dict(check, Dict, CheckDict), + get_dict(type, CheckDict, TypeStr), + atom_string(Type, TypeStr), + ( get_dict(target, CheckDict, TargetStr) + -> atom_string(Target, TargetStr) + ; Target = unknown + ), + ( get_dict(source, CheckDict, SourceStr) + -> atom_string(Source, SourceStr) + ; Source = unknown + ), + Action = action(Type, source(Source), target(Target)). + + +%% ============================================================ +%% OUTBOUND: Prolog response → JSON +%% ============================================================ + +%% Convert a Prolog response from api.pl to a JSON-serializable dict. +%% +%% Prolog format: response(ok, interpreted, result(...)) +%% JSON format (Construction §3.3): +%% { "status": "resolved", "interpretation": { ... } } + +%% Interpreted response +response_to_json(response(ok, interpreted, result(ref(Ref), mitzvot(Mitzvot), + verses(Verses), reading(Reading), + confidence(Conf))), + Dict) :- + maplist(term_to_string, Mitzvot, MitzvotStrs), + maplist(term_to_string, Verses, VerseStrs), + term_to_string(Reading, ReadingStr), + atom_string(Conf, ConfStr), + atom_string(Ref, RefStr), + Dict = json{status: "resolved", + interpretation: json{text: ReadingStr, + prolog_form: RefStr, + grounding: MitzvotStrs, + verses: VerseStrs, + confidence: ConfStr}, + challenge: null}. + +%% Challenged response +response_to_json(response(ok, resolved_challenged, + result(interpretation(Interp), challenge(Kushya))), + Dict) :- + term_to_string(Interp, InterpStr), + term_to_string(Kushya, KushyaStr), + Dict = json{status: "challenged", + interpretation: json{text: InterpStr, + prolog_form: InterpStr, + grounding: [], + confidence: "medium"}, + challenge: json{kushya: KushyaStr, + contradicting: "", + strength: "moderate"}}. + +%% Teiku response +response_to_json(response(ok, teiku, result(record(Record))), Dict) :- + term_to_string(Record, RecordStr), + Dict = json{status: "teiku", + interpretation: json{text: RecordStr, + prolog_form: RecordStr, + grounding: [], + confidence: "low"}, + challenge: null}. + +%% No interpretation +response_to_json(response(error, no_interpretation, CaseRef), Dict) :- + atom_string(CaseRef, RefStr), + Dict = json{status: "teiku", + interpretation: json{text: "No interpretation found", + prolog_form: RefStr, + grounding: [], + confidence: "low"}, + challenge: null}. + +%% No resolution +response_to_json(response(error, no_resolution, CaseRef), Dict) :- + atom_string(CaseRef, RefStr), + Dict = json{status: "teiku", + interpretation: json{text: "No resolution found", + prolog_form: RefStr, + grounding: [], + confidence: "low"}, + challenge: null}. + +%% Halt response +response_to_json(response(halt, violations(Halts), action(Action)), Dict) :- + maplist(term_to_string, Halts, HaltStrs), + term_to_string(Action, ActionStr), + Dict = json{status: "halt", + violations: HaltStrs, + action: ActionStr}. + +%% No halt +response_to_json(response(ok, no_halt, action(Action)), Dict) :- + term_to_string(Action, ActionStr), + Dict = json{status: "ok", + violations: [], + action: ActionStr}. + +%% Catch-all for unexpected responses +response_to_json(Response, Dict) :- + term_to_string(Response, Str), + Dict = json{status: "error", + error: json{code: "UNEXPECTED_RESPONSE", + message: Str}}. + + +%% ============================================================ +%% HELPERS +%% ============================================================ + +%% Parse a root name like "bet-resh-aleph" into an atom bet_resh_aleph +parse_root_name(Name, Atom) :- + ( string(Name) -> S = Name + ; atom_string(Name, S) + ), + split_string(S, "-", "", Parts), + atomic_list_concat(Parts, '_', Atom). + +%% Convert any Prolog term to a string +term_to_string(Term, Str) :- + term_string(Term, Str). + +%% Map known roots to domains (basic mapping; will be extended) +root_to_domain(bet_resh_aleph, creation). +root_to_domain(yod_tsade_resh, creation). +root_to_domain(aleph_lamed_he, holiness). +root_to_domain(qof_dalet_shin, holiness). +root_to_domain(shin_pe_tet, judgment). +root_to_domain(dalet_yod_nun, judgment). +root_to_domain(chet_samekh_dalet, mercy). +root_to_domain(resh_chet_mem, mercy). +root_to_domain(aleph_vav_resh, light). +root_to_domain(chet_shin_kaf, darkness). diff --git a/talmudic_engine/src/interface/socket_server.pl b/talmudic_engine/src/interface/socket_server.pl new file mode 100644 index 00000000..43c264a2 --- /dev/null +++ b/talmudic_engine/src/interface/socket_server.pl @@ -0,0 +1,145 @@ +%% socket_server.pl — Unix domain socket listener for Talmudic Engine. +%% +%% Construction §3.3: "JSON over Unix domain socket" +%% Step 14 (Architecture §4): "Connect components" +%% +%% Listens on a Unix domain socket, accepts JSON requests, +%% dispatches to api:process_request/2 via json_bridge, +%% returns JSON responses. +%% +%% Protocol: One JSON object per line (same as Haskell server). +%% Connection model: accept, handle one or more requests per +%% connection, close when client disconnects. + +:- module(socket_server, [ + serve/0, + serve/1, + stop_server/0 +]). + +:- use_module(library(socket)). +:- use_module(library(json)). +:- use_module(library(readutil)). + +:- use_module('../interface/api'). +:- use_module('../interface/json_bridge'). + +%% Default socket path — dev path; production uses /var/run/shorashim/talmudic.sock +default_socket_path('/tmp/shorashim/talmudic.sock'). + +%% Dynamic flag to control server loop +:- dynamic server_running/1. + +%% ============================================================ +%% SERVER ENTRY POINTS +%% ============================================================ + +%% serve/0 — Start server on default socket path +serve :- + default_socket_path(Path), + serve(Path). + +%% serve/1 — Start server on given socket path +serve(Path) :- + %% Clean up any stale socket file + ( exists_file(Path) + -> delete_file(Path) + ; true + ), + %% Ensure parent directory exists + file_directory_name(Path, Dir), + ( exists_directory(Dir) + -> true + ; make_directory_path(Dir) + ), + %% Create Unix domain socket + unix_domain_socket(ServerSocket), + tcp_bind(ServerSocket, Path), + tcp_listen(ServerSocket, 5), + format('Talmudic Engine listening on ~w~n', [Path]), + flush_output, + %% Set running flag + assert(server_running(true)), + %% Accept loop + accept_loop(ServerSocket, Path). + +%% Stop the server gracefully +stop_server :- + retractall(server_running(_)). + +%% ============================================================ +%% CONNECTION HANDLING +%% ============================================================ + +%% Main accept loop — waits for connections +accept_loop(ServerSocket, Path) :- + ( server_running(true) + -> catch( + ( tcp_accept(ServerSocket, ClientSocket, _Peer), + tcp_open_socket(ClientSocket, InStream, OutStream), + %% Handle this client's requests + catch( + handle_connection(InStream, OutStream), + Error, + ( format(user_error, 'Connection error: ~w~n', [Error]), + flush_output(user_error) + ) + ), + close(InStream, [force(true)]), + close(OutStream, [force(true)]), + accept_loop(ServerSocket, Path) + ), + AcceptError, + ( format(user_error, 'Accept error: ~w~n', [AcceptError]), + flush_output(user_error), + %% Clean up + tcp_close_socket(ServerSocket), + ( exists_file(Path) -> delete_file(Path) ; true ), + fail + ) + ) + ; %% Server stopped — clean up + tcp_close_socket(ServerSocket), + ( exists_file(Path) -> delete_file(Path) ; true ), + format('Talmudic Engine stopped.~n'), + flush_output + ). + +%% Handle a single connection — read lines until EOF +handle_connection(InStream, OutStream) :- + read_line_to_string(InStream, Line), + ( Line == end_of_file + -> true + ; process_json_line(Line, ResponseJson), + %% Write JSON response on a single line + json_write_dict(OutStream, ResponseJson), + nl(OutStream), + flush_output(OutStream), + %% Continue reading from this connection + handle_connection(InStream, OutStream) + ). + + +%% ============================================================ +%% JSON REQUEST PROCESSING +%% ============================================================ + +%% Process a JSON line string and produce a JSON response string +process_json_line(Line, ResponseJson) :- + catch( + ( %% Parse JSON string to dict + atom_json_dict(Line, RequestDict, []), + %% Convert JSON to Prolog request + json_to_request(RequestDict, Request), + %% Dispatch to API + process_request(Request, Response), + %% Convert response to JSON + response_to_json(Response, ResponseJson) + ), + Error, + ( term_string(Error, ErrorStr), + ResponseJson = json{status: "error", + error: json{code: "PROCESSING_ERROR", + message: ErrorStr}} + ) + ). diff --git a/talmudic_engine/src/main.pl b/talmudic_engine/src/main.pl index fc1b7551..6ae126d9 100644 --- a/talmudic_engine/src/main.pl +++ b/talmudic_engine/src/main.pl @@ -17,6 +17,8 @@ :- use_module(halt/shfichut_damim). :- use_module(halt/gilui_arayot). :- use_module(interface/api). +:- use_module(interface/json_bridge). +:- use_module(interface/socket_server). %% Load test files :- use_module(library(plunit)). diff --git a/vav_pipeline/config/config.exs b/vav_pipeline/config/config.exs index 594961cc..bf0bbe81 100644 --- a/vav_pipeline/config/config.exs +++ b/vav_pipeline/config/config.exs @@ -10,7 +10,13 @@ config :vav_pipeline, # Shabbat check interval (1 minute) cycle_boundary_check_interval_ms: 60_000, # ShemaContext: no caching — always read fresh - shema_context_cache_ttl_ms: 0 + shema_context_cache_ttl_ms: 0, + # Step 14 — Component connection config + # Shoresh Engine (Haskell) — binary path for Erlang Port + shoresh_engine_binary: "shoresh-engine", + shoresh_engine_cwd: nil, + # Talmudic Engine (Prolog) — Unix domain socket path + talmudic_socket_path: "/tmp/shorashim/talmudic.sock" # Import environment specific config import_config "#{config_env()}.exs" diff --git a/vav_pipeline/lib/vav_pipeline/application.ex b/vav_pipeline/lib/vav_pipeline/application.ex index 826526b8..88237c4e 100644 --- a/vav_pipeline/lib/vav_pipeline/application.ex +++ b/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 diff --git a/vav_pipeline/lib/vav_pipeline/bereishit_cycle.ex b/vav_pipeline/lib/vav_pipeline/bereishit_cycle.ex index 1407ca9a..91f138ad 100644 --- a/vav_pipeline/lib/vav_pipeline/bereishit_cycle.ex +++ b/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 diff --git a/vav_pipeline/lib/vav_pipeline/sanhedrin_client.ex b/vav_pipeline/lib/vav_pipeline/sanhedrin_client.ex new file mode 100644 index 00000000..1fe028f4 --- /dev/null +++ b/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 -> + <> = hex + "#{a}-#{b}-#{c}-#{d}-#{e}" + end) + end +end diff --git a/vav_pipeline/lib/vav_pipeline/shoresh_engine_client.ex b/vav_pipeline/lib/vav_pipeline/shoresh_engine_client.ex index e1515d03..cadcfc48 100644 --- a/vav_pipeline/lib/vav_pipeline/shoresh_engine_client.ex +++ b/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 diff --git a/vav_pipeline/lib/vav_pipeline/shoresh_engine_port.ex b/vav_pipeline/lib/vav_pipeline/shoresh_engine_port.ex new file mode 100644 index 00000000..764c34cb --- /dev/null +++ b/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 diff --git a/vav_pipeline/lib/vav_pipeline/talmudic_engine_client.ex b/vav_pipeline/lib/vav_pipeline/talmudic_engine_client.ex index 4fba7976..88163293 100644 --- a/vav_pipeline/lib/vav_pipeline/talmudic_engine_client.ex +++ b/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 diff --git a/vav_pipeline/lib/vav_pipeline/talmudic_engine_socket.ex b/vav_pipeline/lib/vav_pipeline/talmudic_engine_socket.ex new file mode 100644 index 00000000..705b5ac7 --- /dev/null +++ b/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 diff --git a/vav_pipeline/mix.exs b/vav_pipeline/mix.exs index aac0d0a0..c1916336 100644 --- a/vav_pipeline/mix.exs +++ b/vav_pipeline/mix.exs @@ -21,7 +21,8 @@ defmodule VavPipeline.MixProject do # Construction spec §1.5 — exact version pins defp deps do [ - {:jason, "~> 1.4.1"} + {:jason, "~> 1.4.1"}, + {:sanhedrin, path: "../sanhedrin", manager: :rebar3} ] end end diff --git a/vav_pipeline/mix.lock b/vav_pipeline/mix.lock index ad24483f..bc91c873 100644 --- a/vav_pipeline/mix.lock +++ b/vav_pipeline/mix.lock @@ -1,3 +1,4 @@ %{ "jason": {:hex, :jason, "1.4.4", "b9226785a9aa77b6857ca22832cffa5d5011a667207eb2a0ad56adb5db443b8a", [:mix], [{:decimal, "~> 1.0 or ~> 2.0", [hex: :decimal, repo: "hexpm", optional: true]}], "hexpm", "c5eb0cab91f094599f94d55bc63409236a8ec69a21a67814529e8d5f6cc90b3b"}, + "jsx": {:hex, :jsx, "3.1.0", "d12516baa0bb23a59bb35dccaf02a1bd08243fcbb9efe24f2d9d056ccff71268", [:rebar3], [], "hexpm", "0c5cc8fdc11b53cc25cf65ac6705ad39e54ecc56d1c22e4adb8f5a53fb9427f3"}, }