{-# LANGUAGE BlockArguments #-}
{-# LANGUAGE OverloadedStrings #-}

-- | Persistent REPL agent as a bus participant.
--
-- 'runConnector' spawns a process (via 'openProc'), attaches to the
-- free-agent bus, and loops: await addressed posts → commit body to repl
-- stdin → block on the next stdout frame → scribe reply.
--
-- Framing is port-side: the repl's prompt grammar ('procMarks') decides
-- turn boundaries, and the emit end blocks until a complete frame arrives.
-- No polling; the timeouts here are deadlines around blocking reads, not
-- backoff loops.
--
-- Turn correlation is content-decided, not geometry-decided.  Each addressed
-- post is treated as an /ask/ whose bus 'stamp' is the correlation id.  The
-- response post carries that id in its 'thread' and is addressed back to the
-- asker.  Zero-frame and two-frame violations are reported as in-band
-- diagnostic posts with the same ask id, so downstream consumers can detect
-- them by the stamped output grammar rather than by timing.
module Free.Agent.Connector
  ( ConnectorConfig (..),
    defaultConnectorConfig,
    runConnector,

    -- * Turn handler (exported for oracles)
    TurnResult (..),
    runTurn,
  )
where

import Circuit.Agent (Name, Post (..), PostId, deliversTo, mkPost)
import Circuit.Agent.Framing (Stamped, stamp, stamped)
import Circuit.Agent.StdPorts
  ( ProcConfig (..),
    ProcEnds (..),
    defaultProcConfig,
    ghciMarks,
    openProc,
  )
import Circuit.Category (K (..))
import Circuit.Layer (run)
import Circuit.Poles (HasDual (..), In (..), Out (..), Poles (..))
import Circuit.Syntax (eval)
import Control.Concurrent (threadDelay)
import Control.Concurrent.STM (atomically)
import Control.Monad (unless, void)
import Data.Foldable (traverse_)
import Data.Maybe (fromMaybe)
import Data.Text (Text)
import Data.Text qualified as T
import Data.Text.Encoding (decodeUtf8, encodeUtf8)
import Free.Agent.Bus (Bus, awaitSince, closeBus, openBus, scribeIO)
import System.IO (hPutStrLn, stderr)
import System.Timeout (timeout)
import Prelude

data ConnectorConfig = ConnectorConfig
  { ConnectorConfig -> Text
connName :: Name,
    ConnectorConfig -> String
connBusRoot :: FilePath,
    ConnectorConfig -> ProcConfig
connRepl :: ProcConfig,
    -- | Deadline (microseconds) for the repl's first frame.
    ConnectorConfig -> Int
connStartupTimeout :: Int,
    -- | Deadline (microseconds) for one command's reply frame.
    ConnectorConfig -> Int
connCommandTimeout :: Int,
    -- | Microseconds to wait for a second frame after the first, used to
    -- detect two-frame commits.  Should be short: the second frame, if it
    -- belongs to the same turn, is already enqueued by the pumper.
    ConnectorConfig -> Int
connExtraFrameTimeout :: Int
  }

defaultConnectorConfig :: Name -> ConnectorConfig
defaultConnectorConfig :: Text -> ConnectorConfig
defaultConnectorConfig Text
name =
  ConnectorConfig
    { connName :: Text
connName = Text
name,
      connBusRoot :: String
connBusRoot = String
"/tmp/free-agent-bus",
      connRepl :: ProcConfig
connRepl =
        ProcConfig
defaultProcConfig
          { procCommand = "cabal",
            procArgs = ["repl"],
            procMarks = ghciMarks
          },
      connStartupTimeout :: Int
connStartupTimeout = Int
180_000_000,
      connCommandTimeout :: Int
connCommandTimeout = Int
60_000_000,
      connExtraFrameTimeout :: Int
connExtraFrameTimeout = Int
10_000
    }

runConnector :: ConnectorConfig -> IO ()
runConnector :: ConnectorConfig -> IO ()
runConnector ConnectorConfig
cfg = do
  let name :: Text
name = ConnectorConfig -> Text
connName ConnectorConfig
cfg
  bus <- String -> IO (Bus Text)
forall a. PostBody a => String -> IO (Bus a)
openBus (ConnectorConfig -> String
connBusRoot ConnectorConfig
cfg)
  repl <- runK (eval (openProc encodeUtf8 decodeUtf8 (connRepl cfg))) ()
  void $ scribeIO bus (mkPost name [] ("starting repl: " <> T.pack (procCommand (connRepl cfg)) <> " " <> T.unwords (map T.pack (procArgs (connRepl cfg)))))

  -- Drain the initial frame (up to the first prompt).
  mFirst <- timeout (connStartupTimeout cfg) (emitText (procStdio repl))
  whenNothing mFirst $
    hPutStrLn stderr "connector: startup deadline expired before first prompt"

  err0 <- drainStderr (procStderr repl)
  unless (T.null err0) $
    hPutStrLn stderr ("connector stderr: " <> T.unpack (T.take 200 err0))

  -- Main loop.
  void $ scribeIO bus (mkPost name [] "repl ready")
  loop bus repl name 0
  closeBus bus
  procClose repl
  where
    whenNothing :: Maybe a -> f () -> f ()
whenNothing Maybe a
m f ()
act = f () -> (a -> f ()) -> Maybe a -> f ()
forall b a. b -> (a -> b) -> Maybe a -> b
maybe f ()
act (f () -> a -> f ()
forall a b. a -> b -> a
const (() -> f ()
forall a. a -> f a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ())) Maybe a
m
    loop :: Bus Text -> ProcEnds Text Text Text -> Text -> PostId -> IO ()
loop Bus Text
bus ProcEnds Text Text Text
repl Text
name PostId
lastId = do
      ps <- STM [Stamped Text] -> IO [Stamped Text]
forall a. STM a -> IO a
atomically (Bus Text -> [Text] -> PostId -> STM [Stamped Text]
forall a. Bus a -> [Text] -> PostId -> STM [Stamped a]
awaitSince Bus Text
bus [Text
name] PostId
lastId)
      if null ps
        then threadDelay 500_000 >> loop bus repl name lastId
        else do
          done <- processPosts cfg bus repl ps
          let newId = [PostId] -> PostId
forall a. Ord a => [a] -> a
forall (t :: * -> *) a. (Foldable t, Ord a) => t a -> a
maximum ((Stamped Text -> PostId) -> [Stamped Text] -> [PostId]
forall a b. (a -> b) -> [a] -> [b]
map ((UTCTime, PostId) -> PostId
forall a b. (a, b) -> b
snd ((UTCTime, PostId) -> PostId)
-> (Stamped Text -> (UTCTime, PostId)) -> Stamped Text -> PostId
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Stamped Text -> (UTCTime, PostId)
forall r a. Stamped r a -> r
stamp) [Stamped Text]
ps) PostId -> PostId -> PostId
forall a. Num a => a -> a -> a
+ PostId
1
          unless done $ loop bus repl name newId

processPosts ::
  ConnectorConfig ->
  Bus Text ->
  ProcEnds Text Text Text ->
  [Stamped Text] ->
  IO Bool
processPosts :: ConnectorConfig
-> Bus Text -> ProcEnds Text Text Text -> [Stamped Text] -> IO Bool
processPosts ConnectorConfig
cfg Bus Text
bus ProcEnds Text Text Text
repl = [Stamped Text] -> IO Bool
go
  where
    name :: Text
name = ConnectorConfig -> Text
connName ConnectorConfig
cfg
    go :: [Stamped Text] -> IO Bool
go [] = Bool -> IO Bool
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Bool
False
    go (Stamped Text
stored : [Stamped Text]
rest) = do
      if Post Text -> [Text] -> Bool
forall a. Post a -> [Text] -> Bool
deliversTo (Stamped Text -> Post Text
forall r a. Stamped r a -> a
stamped Stamped Text
stored) [Text
name] Bool -> Bool -> Bool
|| Text
name Text -> [Text] -> Bool
forall a. Eq a => a -> [a] -> Bool
forall (t :: * -> *) a. (Foldable t, Eq a) => a -> t a -> Bool
`elem` Post Text -> [Text]
forall a. Post a -> [Text]
to (Stamped Text -> Post Text
forall r a. Stamped r a -> a
stamped Stamped Text
stored)
        then do
          let cmd :: Text
cmd = Post Text -> Text
forall a. Post a -> a
body (Stamped Text -> Post Text
forall r a. Stamped r a -> a
stamped Stamped Text
stored)
          if Text
cmd Text -> Text -> Bool
forall a. Eq a => a -> a -> Bool
== Text
"quit" Bool -> Bool -> Bool
|| Text
cmd Text -> Text -> Bool
forall a. Eq a => a -> a -> Bool
== Text
":quit"
            then Bus Text -> Post Text -> IO (Stamped Text)
forall a. Bus a -> Post a -> IO (Stamped a)
scribeIO Bus Text
bus (Text -> [Text] -> Text -> Post Text
forall a. Text -> [Text] -> a -> Post a
mkPost Text
name [] Text
"closing") IO (Stamped Text) -> IO Bool -> IO Bool
forall a b. IO a -> IO b -> IO b
forall (m :: * -> *) a b. Monad m => m a -> m b -> m b
>> Bool -> IO Bool
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Bool
True
            else do
              replies <- ConnectorConfig
-> ProcEnds Text Text Text -> Stamped Text -> IO [Post Text]
runTurn ConnectorConfig
cfg ProcEnds Text Text Text
repl Stamped Text
stored
              traverse_ (scribeIO bus) replies
              go rest
        else [Stamped Text] -> IO Bool
go [Stamped Text]
rest

-- | The result of one critical-section turn, before scribing.
data TurnResult
  = -- | Exactly one response frame.
    TurnOk Text
  | -- | No frame arrived before the deadline.
    TurnZeroFrame
  | -- | More than one frame arrived for the same ask.
    TurnMultiFrame [Text]
  deriving (TurnResult -> TurnResult -> Bool
(TurnResult -> TurnResult -> Bool)
-> (TurnResult -> TurnResult -> Bool) -> Eq TurnResult
forall a. (a -> a -> Bool) -> (a -> a -> Bool) -> Eq a
$c== :: TurnResult -> TurnResult -> Bool
== :: TurnResult -> TurnResult -> Bool
$c/= :: TurnResult -> TurnResult -> Bool
/= :: TurnResult -> TurnResult -> Bool
Eq, Int -> TurnResult -> String -> String
[TurnResult] -> String -> String
TurnResult -> String
(Int -> TurnResult -> String -> String)
-> (TurnResult -> String)
-> ([TurnResult] -> String -> String)
-> Show TurnResult
forall a.
(Int -> a -> String -> String)
-> (a -> String) -> ([a] -> String -> String) -> Show a
$cshowsPrec :: Int -> TurnResult -> String -> String
showsPrec :: Int -> TurnResult -> String -> String
$cshow :: TurnResult -> String
show :: TurnResult -> String
$cshowList :: [TurnResult] -> String -> String
showList :: [TurnResult] -> String -> String
Show)

-- | Run one critical-section turn for an addressed post.
--
-- The ask is identified by the incoming post's bus 'stamp'.  The response
-- carries that id in its 'thread' and is addressed back to the asker.  This
-- makes correlation content-decided: a dropped, doubled, or misordered frame
-- is observable from the bus log, not inferred from pipe adjacency.
runTurn ::
  ConnectorConfig ->
  ProcEnds Text Text Text ->
  Stamped Text ->
  IO [Post Text]
runTurn :: ConnectorConfig
-> ProcEnds Text Text Text -> Stamped Text -> IO [Post Text]
runTurn ConnectorConfig
cfg ProcEnds Text Text Text
repl Stamped Text
stored = do
  let ask :: Post Text
ask = Stamped Text -> Post Text
forall r a. Stamped r a -> a
stamped Stamped Text
stored
      asker :: Text
asker = Post Text -> Text
forall a. Post a -> Text
from Post Text
ask
      askId :: PostId
askId = (UTCTime, PostId) -> PostId
forall a b. (a, b) -> b
snd (Stamped Text -> (UTCTime, PostId)
forall r a. Stamped r a -> r
stamp Stamped Text
stored)
      name :: Text
name = ConnectorConfig -> Text
connName ConnectorConfig
cfg
      stdio :: Poles (K IO) Text Text
stdio = ProcEnds Text Text Text -> Poles (K IO) Text Text
forall a b c. ProcEnds a b c -> Poles (K IO) a b
procStdio ProcEnds Text Text Text
repl

  Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
unless (Text -> Bool
T.null (Post Text -> Text
forall a. Post a -> a
body Post Text
ask)) (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ Poles (K IO) Text Text -> Text -> IO ()
forall b. Poles (K IO) Text b -> Text -> IO ()
commitText Poles (K IO) Text Text
stdio (Post Text -> Text
forall a. Post a -> a
body Post Text
ask)

  mFirst <- Int -> IO Text -> IO (Maybe Text)
forall a. Int -> IO a -> IO (Maybe a)
timeout (ConnectorConfig -> Int
connCommandTimeout ConnectorConfig
cfg) (Poles (K IO) Text Text -> IO Text
forall a. Poles (K IO) a Text -> IO Text
emitText Poles (K IO) Text Text
stdio)
  case mFirst of
    Maybe Text
Nothing -> [Post Text] -> IO [Post Text]
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure [Text -> Text -> PostId -> Text -> Post Text
stampedPost Text
name Text
asker PostId
askId Text
"<zero-frame>"]
    Just Text
out -> do
      mSecond <- Int -> IO Text -> IO (Maybe Text)
forall a. Int -> IO a -> IO (Maybe a)
timeout (ConnectorConfig -> Int
connExtraFrameTimeout ConnectorConfig
cfg) (Poles (K IO) Text Text -> IO Text
forall a. Poles (K IO) a Text -> IO Text
emitText Poles (K IO) Text Text
stdio)
      case mSecond of
        Maybe Text
Nothing -> [Post Text] -> IO [Post Text]
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure [Text -> Text -> PostId -> Text -> Post Text
replyPost Text
name Text
asker PostId
askId Text
out]
        Just Text
out2 ->
          [Post Text] -> IO [Post Text]
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure
            [ Text -> Text -> PostId -> Text -> Post Text
replyPost Text
name Text
asker PostId
askId Text
out,
              Text -> Text -> PostId -> Text -> Post Text
stampedPost Text
name Text
asker PostId
askId (Text
"<extra-frame> " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
out2)
            ]

replyPost :: Name -> Name -> PostId -> Text -> Post Text
replyPost :: Text -> Text -> PostId -> Text -> Post Text
replyPost Text
name Text
asker PostId
askId Text
bodyText =
  (Text -> [Text] -> Text -> Post Text
forall a. Text -> [Text] -> a -> Post a
mkPost Text
name [Text
asker] Text
bodyText) {thread = [askId]}

stampedPost :: Name -> Name -> PostId -> Text -> Post Text
stampedPost :: Text -> Text -> PostId -> Text -> Post Text
stampedPost Text
name Text
asker PostId
askId Text
bodyText =
  (Text -> [Text] -> Text -> Post Text
forall a. Text -> [Text] -> a -> Post a
mkPost Text
name [Text
asker] Text
bodyText) {thread = [askId]}

-- | Commit one 'Text' token through an 'Poles' conjoint.
commitText :: Poles (K IO) Text b -> Text -> IO ()
commitText :: forall b. Poles (K IO) Text b -> Text -> IO ()
commitText Poles (K IO) Text b
e Text
t = K IO Text () -> Text -> IO ()
forall {k} (m :: k -> *) a (b :: k). K m a b -> a -> m b
runK (In (K IO) Text -> forall x. Out (K IO) x -> K IO Text x
forall {k1} {k2} (arr :: k1 -> k2 -> *) (a :: k1).
In arr a -> forall (x :: k2). Out arr x -> arr a x
commit (Poles (K IO) Text b -> In (K IO) Text
forall {k1} {k2} (arr :: k1 -> k2 -> *) (a :: k1) (b :: k2).
Poles arr a b -> In arr a
conjoint Poles (K IO) Text b
e) Out (K IO) ()
outU) Text
t
  where
    Poles In (K IO) ()
_ Out (K IO) ()
outU = Poles (K IO) () ()
forall {k} (bot :: k) (arr :: k -> k -> *).
HasDual bot arr =>
Poles arr bot bot
open

-- | Emit one 'Text' frame from an 'Poles' companion.  Blocks until a
-- complete frame arrives.
emitText :: Poles (K IO) a Text -> IO Text
emitText :: forall a. Poles (K IO) a Text -> IO Text
emitText Poles (K IO) a Text
e = K IO () Text -> () -> IO Text
forall {k} (m :: k -> *) a (b :: k). K m a b -> a -> m b
runK (Out (K IO) Text -> forall x. In (K IO) x -> K IO x Text
forall {k1} {k2} (arr :: k1 -> k2 -> *) (a :: k2).
Out arr a -> forall (x :: k1). In arr x -> arr x a
emit (Poles (K IO) a Text -> Out (K IO) Text
forall {k1} {k2} (arr :: k1 -> k2 -> *) (a :: k1) (b :: k2).
Poles arr a b -> Out arr b
companion Poles (K IO) a Text
e) In (K IO) ()
inU) ()
  where
    Poles In (K IO) ()
inU Out (K IO) ()
_ = Poles (K IO) () ()
forall {k} (bot :: k) (arr :: k -> k -> *).
HasDual bot arr =>
Poles arr bot bot
open

-- | Drain pending stderr lines, bounded: returns after ~100ms of quiet.
-- Stderr is diagnostics, not dialogue — a bounded drain, not a blocking
-- read, so a silent stderr cannot stall the turn.
drainStderr :: Poles (K IO) a Text -> IO Text
drainStderr :: forall a. Poles (K IO) a Text -> IO Text
drainStderr Poles (K IO) a Text
e = [Text] -> IO Text
go []
  where
    go :: [Text] -> IO Text
go [Text]
acc = do
      m <- Int -> IO Text -> IO (Maybe Text)
forall a. Int -> IO a -> IO (Maybe a)
timeout Int
100_000 (Poles (K IO) a Text -> IO Text
forall a. Poles (K IO) a Text -> IO Text
emitText Poles (K IO) a Text
e)
      case m of
        Maybe Text
Nothing -> Text -> IO Text
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ([Text] -> Text
T.unlines ([Text] -> [Text]
forall a. [a] -> [a]
reverse [Text]
acc))
        Just Text
l -> [Text] -> IO Text
go (Text
l Text -> [Text] -> [Text]
forall a. a -> [a] -> [a]
: [Text]
acc)