{-# LANGUAGE LambdaCase #-}
{-# LANGUAGE OverloadedStrings #-}
{-# LANGUAGE ScopedTypeVariables #-}

-- | Process ports: stdin / stdout / stderr as free dual poles.
--
-- A 'StdPorts' handle is a persistent child process viewed through three
-- independent 'Circuit.Poles' seats plus a close action.  Pipes all the way
-- down: no FIFO, no log files, no byte-offset polling.
--
-- Internal I/O is 'ByteString'.  Constructors take encode/decode adapters:
--
-- @
-- openStdPorts encodeUtf8 decodeUtf8 cfg  -- 'Text'
-- openStdPorts id id cfg                  -- 'ByteString'
-- @
--
-- = Stream marks
--
-- The boundary grammar of a process stream is a type, 'ProcMarks' — the
-- level-0 grammar of 'Circuit.Agent.Mark' surfaced at the process boundary.
-- A mark announces the end of a frame; everything before it is the payload.
-- 'splitFrame' is the stateless parse of the mark machine: the byte buffer
-- is the whole state.
--
-- A pump thread per output handle frames the raw stream and closes each
-- payload into a queue ('openIO' 'Unbounded').  The emit ends are the queue
-- companions: an emit /blocks/ until a complete frame arrives — the queue's
-- @readTQueue@ retry IS the blocking boundary.  Arrival is decided by
-- content, never inferred from quiet; there is no timeout and no empty
-- poll result on the emit side.
--
-- = Resource lifecycle
--
-- 'openStdPorts' returns a @'Loop' 'Either'@ where the feedback state is
-- the process.  'Left' = process alive, 'Right' = ports delivered.  The
-- 'StdPorts' ends are self-contained — they capture the pipe handles, the
-- queues, and the pump threads.  'stdClose' terminates the process, kills
-- the pumps, and closes the handles.
module Circuit.Agent.StdPorts
  ( -- * Configuration
    ProcConfig (..),
    defaultProcConfig,

    -- * Stream marks
    ProcMarks (..),
    splitFrame,
    lineMarks,
    ghciMarks,
    hermesMarks,
    sseMarks,

    -- * Process ports
    StdPorts (..),
    openStdPorts,

    -- * The mark machine
    frameAgent,
    frameProcess,

    -- * Ends / seat view (client-facing)
    ProcEnds (..),
    stdioEnds,
    stderrEnds,
    openProc,
    portsEnds,

    -- * In-memory test harness
    echo,
  )
where

import Circuit.Agent (Agent, run1)
import Circuit.Agent.Ends (ChannelPolicy (..), Queue (..), openChannel, openIO)
import Circuit.Category (K (..), (.>))
import Circuit.Poles (HasDual (..), In (..), Out (..), Poles (..), commit, emit, open)
import Circuit.Poly (Eval (..))
import Circuit.Process (Process, iterateSystem, systemAsProcess)
import Circuit.System (fromEvalSystem)
import Circuit.Tensor (Tensor (..))
import Circuit.Trace (Trace, base, yank)
import Control.Concurrent (forkIO, killThread)
import Control.Exception (IOException, try)
import Control.Monad (void)
import Data.ByteString qualified as BS
import Data.Foldable (traverse_)
import Data.IORef
import Data.List (minimumBy)
import Data.Maybe (fromMaybe)
import Data.Ord (comparing)
import Data.Text (Text)
import Data.Text.Encoding (encodeUtf8)
import System.IO
  ( BufferMode (NoBuffering),
    Handle,
    hClose,
    hFlush,
    hSetBuffering,
  )
import System.Process
import Prelude

-- $setup
--
-- >>> import Circuit.Process (iterateSystem)
-- >>> import Circuit.Poles (Poles (..), HasDual (..), commit, emit, open)
-- >>> import Circuit.Category (K (..))
-- >>> import Data.Text.Encoding (decodeUtf8)

-- ---------------------------------------------------------------------------
-- Configuration
-- ---------------------------------------------------------------------------

data ProcConfig = ProcConfig
  { ProcConfig -> String
procCommand :: String,
    ProcConfig -> [String]
procArgs :: [String],
    ProcConfig -> String
procWorkingDir :: FilePath,
    -- | Boundary grammar of the stdout stream.  Stderr is line-framed
    -- ('lineMarks'); stderr is diagnostics, not dialogue.
    ProcConfig -> ProcMarks
procMarks :: ProcMarks
  }
  deriving (Int -> ProcConfig -> ShowS
[ProcConfig] -> ShowS
ProcConfig -> String
(Int -> ProcConfig -> ShowS)
-> (ProcConfig -> String)
-> ([ProcConfig] -> ShowS)
-> Show ProcConfig
forall a.
(Int -> a -> ShowS) -> (a -> String) -> ([a] -> ShowS) -> Show a
$cshowsPrec :: Int -> ProcConfig -> ShowS
showsPrec :: Int -> ProcConfig -> ShowS
$cshow :: ProcConfig -> String
show :: ProcConfig -> String
$cshowList :: [ProcConfig] -> ShowS
showList :: [ProcConfig] -> ShowS
Show, ProcConfig -> ProcConfig -> Bool
(ProcConfig -> ProcConfig -> Bool)
-> (ProcConfig -> ProcConfig -> Bool) -> Eq ProcConfig
forall a. (a -> a -> Bool) -> (a -> a -> Bool) -> Eq a
$c== :: ProcConfig -> ProcConfig -> Bool
== :: ProcConfig -> ProcConfig -> Bool
$c/= :: ProcConfig -> ProcConfig -> Bool
/= :: ProcConfig -> ProcConfig -> Bool
Eq)

defaultProcConfig :: ProcConfig
defaultProcConfig :: ProcConfig
defaultProcConfig =
  ProcConfig
    { procCommand :: String
procCommand = String
"cabal",
      procArgs :: [String]
procArgs = [String
"repl"],
      procWorkingDir :: String
procWorkingDir = String
".",
      procMarks :: ProcMarks
procMarks = ProcMarks
ghciMarks
    }

-- ---------------------------------------------------------------------------
-- Stream marks
-- ---------------------------------------------------------------------------

-- | The boundary grammar of a process stream: a finite set of marks, each
-- a glyph sequence announcing the end of a frame.  This is the level-0
-- grammar of the process boundary — the free boundary @K + payload@ with
-- @K@ finite, the stateless 'splitFrame' of the mark machine.  Anything
-- stateful (turn counters, roles) lives above this layer.
newtype ProcMarks = ProcMarks [Text]
  deriving (Int -> ProcMarks -> ShowS
[ProcMarks] -> ShowS
ProcMarks -> String
(Int -> ProcMarks -> ShowS)
-> (ProcMarks -> String)
-> ([ProcMarks] -> ShowS)
-> Show ProcMarks
forall a.
(Int -> a -> ShowS) -> (a -> String) -> ([a] -> ShowS) -> Show a
$cshowsPrec :: Int -> ProcMarks -> ShowS
showsPrec :: Int -> ProcMarks -> ShowS
$cshow :: ProcMarks -> String
show :: ProcMarks -> String
$cshowList :: [ProcMarks] -> ShowS
showList :: [ProcMarks] -> ShowS
Show, ProcMarks -> ProcMarks -> Bool
(ProcMarks -> ProcMarks -> Bool)
-> (ProcMarks -> ProcMarks -> Bool) -> Eq ProcMarks
forall a. (a -> a -> Bool) -> (a -> a -> Bool) -> Eq a
$c== :: ProcMarks -> ProcMarks -> Bool
== :: ProcMarks -> ProcMarks -> Bool
$c/= :: ProcMarks -> ProcMarks -> Bool
/= :: ProcMarks -> ProcMarks -> Bool
Eq)

-- | Line framing: the newline is the mark, one frame per line.  For line
-- protocols (ACP's JSON-RPC) and for stderr diagnostics.
lineMarks :: ProcMarks
lineMarks :: ProcMarks
lineMarks = [Text] -> ProcMarks
ProcMarks [Text
"\n"]

-- | The ghci prompt grammar.
ghciMarks :: ProcMarks
ghciMarks :: ProcMarks
ghciMarks = [Text] -> ProcMarks
ProcMarks [Text
"ghci> ", Text
"\955> "]

-- | The hermes CLI prompt grammar: @❯@ between separator lines when ready.
-- ANSI decorations ride inside the payload; the mark itself is a bare glyph.
-- (Probe card: @coffee\/loom\/hermes-boundary-probe.md@.)
hermesMarks :: ProcMarks
hermesMarks :: ProcMarks
hermesMarks = [Text] -> ProcMarks
ProcMarks [Text
"❯"]

-- | The SSE (Server-Sent Events) level-0 grammar: the blank line ends an
-- event frame.  Payload is the @event:\/data:@ block, mark stripped.
sseMarks :: ProcMarks
sseMarks :: ProcMarks
sseMarks = [Text] -> ProcMarks
ProcMarks [Text
"\n\n"]

-- | The stateless mark parse: the earliest mark occurrence in the buffer,
-- giving @(payload, rest)@ — payload before the mark (mark stripped), rest
-- after it.  'Nothing' when no mark has arrived yet.
--
-- Marks are matched on bytes ('encodeUtf8').  UTF-8 is self-synchronising:
-- a valid multi-byte encoding never occurs inside another character, so a
-- byte-level mark cannot false-match, and a mark split across read chunks
-- is found once the buffer accumulates its final byte.
splitFrame :: ProcMarks -> BS.ByteString -> Maybe (BS.ByteString, BS.ByteString)
splitFrame :: ProcMarks -> ByteString -> Maybe (ByteString, ByteString)
splitFrame (ProcMarks [Text]
marks) ByteString
buf =
  case [(Int, ByteString, ByteString, ByteString)]
hits of
    [] -> Maybe (ByteString, ByteString)
forall a. Maybe a
Nothing
    [(Int, ByteString, ByteString, ByteString)]
_ ->
      let (Int
_, ByteString
markBytes, ByteString
before, ByteString
rest) = ((Int, ByteString, ByteString, ByteString)
 -> (Int, ByteString, ByteString, ByteString) -> Ordering)
-> [(Int, ByteString, ByteString, ByteString)]
-> (Int, ByteString, ByteString, ByteString)
forall (t :: * -> *) a.
Foldable t =>
(a -> a -> Ordering) -> t a -> a
minimumBy (((Int, ByteString, ByteString, ByteString) -> Int)
-> (Int, ByteString, ByteString, ByteString)
-> (Int, ByteString, ByteString, ByteString)
-> Ordering
forall a b. Ord a => (b -> a) -> b -> b -> Ordering
comparing (\(Int
n, ByteString
_, ByteString
_, ByteString
_) -> Int
n)) [(Int, ByteString, ByteString, ByteString)]
hits
       in (ByteString, ByteString) -> Maybe (ByteString, ByteString)
forall a. a -> Maybe a
Just (ByteString
before, Int -> ByteString -> ByteString
BS.drop (ByteString -> Int
BS.length ByteString
markBytes) ByteString
rest)
  where
    hits :: [(Int, ByteString, ByteString, ByteString)]
hits =
      [ (ByteString -> Int
BS.length ByteString
before, ByteString
markBytes, ByteString
before, ByteString
rest)
      | Text
mark <- [Text]
marks,
        let markBytes :: ByteString
markBytes = Text -> ByteString
encodeUtf8 Text
mark,
        Bool -> Bool
not (ByteString -> Bool
BS.null ByteString
markBytes),
        let (ByteString
before, ByteString
rest) = ByteString -> ByteString -> (ByteString, ByteString)
BS.breakSubstring ByteString
markBytes ByteString
buf,
        ByteString
markBytes ByteString -> ByteString -> Bool
`BS.isPrefixOf` ByteString
rest
      ]

-- ---------------------------------------------------------------------------
-- Process ports
-- ---------------------------------------------------------------------------

-- | A process token with three free dual seats: stdin commit, stdout emit,
-- and stderr emit.
--
-- The emit seats block: an emit returns the next complete frame, waiting on
-- the queue when the stream has not produced one yet.  There is no
-- empty-read result — quiet is not an opinion here.
data StdPorts a b c = StdPorts
  { forall a b c. StdPorts a b c -> In (K IO) a
stdIn :: In (K IO) a,
    forall a b c. StdPorts a b c -> Out (K IO) b
stdOut :: Out (K IO) b,
    forall a b c. StdPorts a b c -> Out (K IO) c
stdErr :: Out (K IO) c,
    forall a b c. StdPorts a b c -> IO ()
stdClose :: IO ()
  }

-- | Spawn a process and open its ports as a 'Trace' 'Either'.
--
-- @encode@ converts a token to bytes written to stdin (+ newline).
-- @decode@ converts a framed payload's bytes back to a token.
--
-- The feedback state is the process.  'Left' = process alive, 'Right' =
-- ports delivered, resources captured by 'stdClose'.
openStdPorts ::
  (a -> BS.ByteString) ->
  (BS.ByteString -> a) ->
  ProcConfig ->
  Trace Either (K IO) () (StdPorts a a a)
openStdPorts :: forall a.
(a -> ByteString)
-> (ByteString -> a)
-> ProcConfig
-> Trace Either (K IO) () (StdPorts a a a)
openStdPorts a -> ByteString
encode ByteString -> a
decode ProcConfig
cfg = Trace
  Either
  (K IO)
  (Either (StdPorts a a a) ())
  (Either (StdPorts a a a) (StdPorts a a a))
-> Trace Either (K IO) () (StdPorts a a a)
forall (t :: * -> * -> *) (arr :: * -> * -> *) s a b.
Trace t arr (t s a) (t s b) -> Trace t arr a b
yank (K IO
  (Either (StdPorts a a a) ())
  (Either (StdPorts a a a) (StdPorts a a a))
-> Trace
     Either
     (K IO)
     (Either (StdPorts a a a) ())
     (Either (StdPorts a a a) (StdPorts a a a))
forall (arr :: * -> * -> *) a b (t :: * -> * -> *).
arr a b -> Trace t arr a b
base ((Either (StdPorts a a a) ()
 -> IO (Either (StdPorts a a a) (StdPorts a a a)))
-> K IO
     (Either (StdPorts a a a) ())
     (Either (StdPorts a a a) (StdPorts a a a))
forall {k} (m :: k -> *) a (b :: k). (a -> m b) -> K m a b
K Either (StdPorts a a a) ()
-> IO (Either (StdPorts a a a) (StdPorts a a a))
step))
  where
    step :: Either (StdPorts a a a) ()
-> IO (Either (StdPorts a a a) (StdPorts a a a))
step (Right ()) = do
      let procSpec :: CreateProcess
procSpec =
            (String -> [String] -> CreateProcess
proc (ProcConfig -> String
procCommand ProcConfig
cfg) (ProcConfig -> [String]
procArgs ProcConfig
cfg))
              { cwd = Just (procWorkingDir cfg),
                std_in = CreatePipe,
                std_out = CreatePipe,
                std_err = CreatePipe
              }
      (Just stdinH, Just stdoutH, Just stderrH, ph) <- CreateProcess
-> IO (Maybe Handle, Maybe Handle, Maybe Handle, ProcessHandle)
createProcess CreateProcess
procSpec
      hSetBuffering stdinH NoBuffering
      hSetBuffering stdoutH NoBuffering
      hSetBuffering stderrH NoBuffering

      -- Mark-carrying stdout uses the linear (empty-residual) mediator so
      -- halt marks are preserved in order.  Diagnostic stderr uses a bounded
      -- weakening mediator: old diagnostics may be dropped when the reader
      -- falls behind, but the process dialogue must never lose a halt.
      qOut <- openChannel Linear
      qErr <- openChannel (NewestN 100)
      outTid <- forkIO (pumpFrames (procMarks cfg) decode stdoutH (sink qOut))
      errTid <- forkIO (pumpFrames lineMarks decode stderrH (sink qErr))

      let ports =
            StdPorts
              { stdIn :: In (K IO) a
stdIn = (forall x. Out (K IO) x -> K IO a x) -> In (K IO) a
forall {k} {k1} (arr :: k -> k1 -> *) (a :: k).
(forall (x :: k1). Out arr x -> arr a x) -> In arr a
In ((forall x. Out (K IO) x -> K IO a x) -> In (K IO) a)
-> (forall x. Out (K IO) x -> K IO a x) -> In (K IO) a
forall a b. (a -> b) -> a -> b
$ \Out (K IO) x
o -> (a -> IO x) -> K IO a x
forall {k} (m :: k -> *) a (b :: k). (a -> m b) -> K m a b
K ((a -> IO x) -> K IO a x) -> (a -> IO x) -> K IO a x
forall a b. (a -> b) -> a -> b
$ \a
a -> do
                  Handle -> ByteString -> IO ()
BS.hPutStr Handle
stdinH (a -> ByteString
encode a
a ByteString -> ByteString -> ByteString
forall a. Semigroup a => a -> a -> a
<> ByteString
"\n")
                  Handle -> IO ()
hFlush Handle
stdinH
                  K IO a x -> a -> IO x
forall {k} (m :: k -> *) a (b :: k). K m a b -> a -> m b
runK (Out (K IO) x -> forall x. In (K IO) x -> K IO x x
forall {k1} {k2} (arr :: k1 -> k2 -> *) (a :: k2).
Out arr a -> forall (x :: k1). In arr x -> arr x a
emit Out (K IO) x
o (StdPorts a a a -> In (K IO) a
forall a b c. StdPorts a b c -> In (K IO) a
stdIn StdPorts a a a
ports)) a
a,
                stdOut :: Out (K IO) a
stdOut = Poles (K IO) a a -> Out (K IO) a
forall {k1} {k2} (arr :: k1 -> k2 -> *) (a :: k1) (b :: k2).
Poles arr a b -> Out arr b
companion Poles (K IO) a a
qOut,
                stdErr :: Out (K IO) a
stdErr = Poles (K IO) a a -> Out (K IO) a
forall {k1} {k2} (arr :: k1 -> k2 -> *) (a :: k1) (b :: k2).
Poles arr a b -> Out arr b
companion Poles (K IO) a a
qErr,
                stdClose :: IO ()
stdClose = do
                  IO (Either IOException ()) -> IO ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (IO (Either IOException ()) -> IO ())
-> IO (Either IOException ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ forall e a. Exception e => IO a -> IO (Either e a)
try @IOException (Handle -> IO ()
hClose Handle
stdinH)
                  ProcessHandle -> IO ()
terminateProcess ProcessHandle
ph
                  ThreadId -> IO ()
killThread ThreadId
outTid
                  ThreadId -> IO ()
killThread ThreadId
errTid
              }
      pure (Left ports)
    step (Left StdPorts a a a
ports) =
      Either (StdPorts a a a) (StdPorts a a a)
-> IO (Either (StdPorts a a a) (StdPorts a a a))
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (StdPorts a a a -> Either (StdPorts a a a) (StdPorts a a a)
forall a b. b -> Either a b
Right StdPorts a a a
ports)

-- | Commit one token to a queue end, plugged with unit poles.
sink :: Poles (K IO) a a -> a -> IO ()
sink :: forall a. Poles (K IO) a a -> a -> IO ()
sink Poles (K IO) a a
q = K IO a () -> a -> IO ()
forall {k} (m :: k -> *) a (b :: k). K m a b -> a -> m b
runK (In (K IO) a -> forall x. Out (K IO) x -> K IO a 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) a a -> In (K IO) a
forall {k1} {k2} (arr :: k1 -> k2 -> *) (a :: k1) (b :: k2).
Poles arr a b -> In arr a
conjoint Poles (K IO) a a
q) Out (K IO) ()
outU)
  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 :: Poles (K IO) () ()

-- | The pumper as an agent: a Moore machine from maybe-chunks to frame
-- lists.  The carrier is @(buffer, pending)@ — the unexplained suffix and
-- the frames awaiting observation.  @Just chunk@ is the percept,
-- 'Nothing' the end-of-stream mark: the EOF flush is content, decided by
-- the same stateless grammar.  This is the level-0 mark machine made
-- literal; 'pumpFrames' is merely its IO interpretation at a 'Handle'.
--
-- >>> iterateSystem (frameAgent lineMarks decodeUtf8) ("", []) [Just "a\n", Just "b", Nothing]
-- [["a"],[],["b"]]
frameAgent :: ProcMarks -> (BS.ByteString -> a) -> Agent (->) (BS.ByteString, [a]) (Maybe BS.ByteString) [a]
frameAgent :: forall a.
ProcMarks
-> (ByteString -> a)
-> Agent (->) (ByteString, [a]) (Maybe ByteString) [a]
frameAgent ProcMarks
marks ByteString -> a
decode = ((ByteString, [a])
 -> Eval (Mono (Maybe ByteString) [a]) (ByteString, [a]))
-> System (->) (ByteString, [a]) (Mono (Maybe ByteString) [a])
forall (p :: Poly) s.
SystemEval p =>
(s -> Eval p s) -> System (->) s p
fromEvalSystem (((ByteString, [a])
  -> Eval (Mono (Maybe ByteString) [a]) (ByteString, [a]))
 -> System (->) (ByteString, [a]) (Mono (Maybe ByteString) [a]))
-> ((ByteString, [a])
    -> Eval (Mono (Maybe ByteString) [a]) (ByteString, [a]))
-> System (->) (ByteString, [a]) (Mono (Maybe ByteString) [a])
forall a b. (a -> b) -> a -> b
$ \(ByteString
buf, [a]
pending) ->
  (Eval ('Const [a]) (ByteString, [a]),
 Eval ('Exp (Maybe ByteString)) (ByteString, [a]))
-> Eval (Mono (Maybe ByteString) [a]) (ByteString, [a])
forall (p1 :: Poly) x (q :: Poly).
(Eval p1 x, Eval q x) -> Eval ('Prod p1 q) x
EP
    ( [a] -> Eval ('Const [a]) (ByteString, [a])
forall c x. c -> Eval ('Const c) x
EK [a]
pending,
      (Maybe ByteString -> (ByteString, [a]))
-> Eval ('Exp (Maybe ByteString)) (ByteString, [a])
forall a x. (a -> x) -> Eval ('Exp a) x
EE ((Maybe ByteString -> (ByteString, [a]))
 -> Eval ('Exp (Maybe ByteString)) (ByteString, [a]))
-> (Maybe ByteString -> (ByteString, [a]))
-> Eval ('Exp (Maybe ByteString)) (ByteString, [a])
forall a b. (a -> b) -> a -> b
$ \case
        Just ByteString
chunk ->
          let ([ByteString]
fs, ByteString
buf') = ByteString -> ([ByteString], ByteString)
peel (ByteString
buf ByteString -> ByteString -> ByteString
forall a. Semigroup a => a -> a -> a
<> ByteString
chunk)
           in (ByteString
buf', (ByteString -> a) -> [ByteString] -> [a]
forall a b. (a -> b) -> [a] -> [b]
map ByteString -> a
decode [ByteString]
fs)
        Maybe ByteString
Nothing ->
          (ByteString
BS.empty, [ByteString -> a
decode ByteString
buf | Bool -> Bool
not (ByteString -> Bool
BS.null ByteString
buf)])
    )
  where
    peel :: ByteString -> ([ByteString], ByteString)
peel ByteString
buf = case ProcMarks -> ByteString -> Maybe (ByteString, ByteString)
splitFrame ProcMarks
marks ByteString
buf of
      Maybe (ByteString, ByteString)
Nothing -> ([], ByteString
buf)
      Just (ByteString
p, ByteString
rest) ->
        let ([ByteString]
ps, ByteString
rest') = ByteString -> ([ByteString], ByteString)
peel ByteString
rest
         in (ByteString
p ByteString -> [ByteString] -> [ByteString]
forall a. a -> [a] -> [a]
: [ByteString]
ps, ByteString
rest')

-- | The same machine as a 'Process': chunk stream in, frame lists out,
-- state carried implicitly.
frameProcess :: ProcMarks -> (BS.ByteString -> a) -> Process (Maybe BS.ByteString) [a]
frameProcess :: forall a.
ProcMarks -> (ByteString -> a) -> Process (Maybe ByteString) [a]
frameProcess ProcMarks
marks ByteString -> a
decode = System (->) (ByteString, [a]) (Mono (Maybe ByteString) [a])
-> (ByteString, [a]) -> Process (Maybe ByteString) [a]
forall s i o. System (->) s (Mono i o) -> s -> Process i o
systemAsProcess (ProcMarks
-> (ByteString -> a)
-> System (->) (ByteString, [a]) (Mono (Maybe ByteString) [a])
forall a.
ProcMarks
-> (ByteString -> a)
-> Agent (->) (ByteString, [a]) (Maybe ByteString) [a]
frameAgent ProcMarks
marks ByteString -> a
decode) (ByteString
BS.empty, [])

-- | The pumper: the IO interpretation of 'frameAgent' at a 'Handle'.
-- Blocking reads deliver percepts; payloads are sunk into the queue.
-- A mark split across reads completes in the buffer; end-of-stream is the
-- 'Nothing' percept, flushing a partial frame as the final token.
pumpFrames :: ProcMarks -> (BS.ByteString -> a) -> Handle -> (a -> IO ()) -> IO ()
pumpFrames :: forall a.
ProcMarks -> (ByteString -> a) -> Handle -> (a -> IO ()) -> IO ()
pumpFrames ProcMarks
marks ByteString -> a
decode Handle
h a -> IO ()
snk = (ByteString, [a]) -> IO ()
go (ByteString
BS.empty, [])
  where
    sys :: Agent (->) (ByteString, [a]) (Maybe ByteString) [a]
sys = ProcMarks
-> (ByteString -> a)
-> Agent (->) (ByteString, [a]) (Maybe ByteString) [a]
forall a.
ProcMarks
-> (ByteString -> a)
-> Agent (->) (ByteString, [a]) (Maybe ByteString) [a]
frameAgent ProcMarks
marks ByteString -> a
decode
    go :: (ByteString, [a]) -> IO ()
go (ByteString, [a])
st = do
      r <- forall e a. Exception e => IO a -> IO (Either e a)
try @IOException (Handle -> Int -> IO ByteString
BS.hGetSome Handle
h Int
4096)
      let input = case Either IOException ByteString
r of
            Left IOException
_ -> Maybe ByteString
forall a. Maybe a
Nothing
            Right ByteString
bs
              | ByteString -> Bool
BS.null ByteString
bs -> Maybe ByteString
forall a. Maybe a
Nothing
              | Bool
otherwise -> ByteString -> Maybe ByteString
forall a. a -> Maybe a
Just ByteString
bs
          (outs, st') = run1 sys st input
      traverse_ snk outs
      case input of
        Maybe ByteString
Nothing -> () -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
        Just ByteString
_ -> (ByteString, [a]) -> IO ()
go (ByteString, [a])
st'

-- ---------------------------------------------------------------------------
-- Ends / seat view
-- ---------------------------------------------------------------------------

-- | Client view of a process: two 'Poles' sharing stdin, plus resource close.
data ProcEnds a b c = ProcEnds
  { forall a b c. ProcEnds a b c -> Poles (K IO) a b
procStdio :: Poles (K IO) a b,
    forall a b c. ProcEnds a b c -> Poles (K IO) a c
procStderr :: Poles (K IO) a c,
    forall a b c. ProcEnds a b c -> IO ()
procClose :: IO ()
  }

-- | Stdin commit + stdout emit as matched 'Poles'.
stdioEnds :: StdPorts a b c -> Poles (K IO) a b
stdioEnds :: forall a b c. StdPorts a b c -> Poles (K IO) a b
stdioEnds StdPorts a b c
pp = In (K IO) a -> Out (K IO) b -> Poles (K IO) a b
forall {k} {k1} (arr :: k -> k1 -> *) (a :: k) (b :: k1).
In arr a -> Out arr b -> Poles arr a b
Poles (StdPorts a b c -> In (K IO) a
forall a b c. StdPorts a b c -> In (K IO) a
stdIn StdPorts a b c
pp) (StdPorts a b c -> Out (K IO) b
forall a b c. StdPorts a b c -> Out (K IO) b
stdOut StdPorts a b c
pp)

-- | Stdin commit + stderr emit as matched 'Poles'.
stderrEnds :: StdPorts a b c -> Poles (K IO) a c
stderrEnds :: forall a b c. StdPorts a b c -> Poles (K IO) a c
stderrEnds StdPorts a b c
pp = In (K IO) a -> Out (K IO) c -> Poles (K IO) a c
forall {k} {k1} (arr :: k -> k1 -> *) (a :: k) (b :: k1).
In arr a -> Out arr b -> Poles arr a b
Poles (StdPorts a b c -> In (K IO) a
forall a b c. StdPorts a b c -> In (K IO) a
stdIn StdPorts a b c
pp) (StdPorts a b c -> Out (K IO) c
forall a b c. StdPorts a b c -> Out (K IO) c
stdErr StdPorts a b c
pp)

-- | Open a process and return the dual-seat client view as a 'Trace' 'Either'.
openProc ::
  (a -> BS.ByteString) ->
  (BS.ByteString -> a) ->
  ProcConfig ->
  Trace Either (K IO) () (ProcEnds a a a)
openProc :: forall a.
(a -> ByteString)
-> (ByteString -> a)
-> ProcConfig
-> Trace Either (K IO) () (ProcEnds a a a)
openProc a -> ByteString
encode ByteString -> a
decode ProcConfig
cfg =
  (a -> ByteString)
-> (ByteString -> a)
-> ProcConfig
-> Trace Either (K IO) () (StdPorts a a a)
forall a.
(a -> ByteString)
-> (ByteString -> a)
-> ProcConfig
-> Trace Either (K IO) () (StdPorts a a a)
openStdPorts a -> ByteString
encode ByteString -> a
decode ProcConfig
cfg Trace Either (K IO) () (StdPorts a a a)
-> Syntax
     (SigCompose :+: SigYank Either)
     (K IO)
     (StdPorts a a a)
     (ProcEnds a a a)
-> Syntax
     (SigCompose :+: SigYank Either) (K IO) () (ProcEnds a a a)
forall {k} (arr :: k -> k -> *) (a :: k) (b :: k) (c :: k).
Category arr =>
arr a b -> arr b c -> arr a c
.> K IO (StdPorts a a a) (ProcEnds a a a)
-> Syntax
     (SigCompose :+: SigYank Either)
     (K IO)
     (StdPorts a a a)
     (ProcEnds a a a)
forall (arr :: * -> * -> *) a b (t :: * -> * -> *).
arr a b -> Trace t arr a b
base K IO (StdPorts a a a) (ProcEnds a a a)
forall {a} {b} {c}. K IO (StdPorts a b c) (ProcEnds a b c)
portsToProcEnds
  where
    portsToProcEnds :: K IO (StdPorts a b c) (ProcEnds a b c)
portsToProcEnds = (StdPorts a b c -> IO (ProcEnds a b c))
-> K IO (StdPorts a b c) (ProcEnds a b c)
forall {k} (m :: k -> *) a (b :: k). (a -> m b) -> K m a b
K ((StdPorts a b c -> IO (ProcEnds a b c))
 -> K IO (StdPorts a b c) (ProcEnds a b c))
-> (StdPorts a b c -> IO (ProcEnds a b c))
-> K IO (StdPorts a b c) (ProcEnds a b c)
forall a b. (a -> b) -> a -> b
$ \StdPorts a b c
pp ->
      ProcEnds a b c -> IO (ProcEnds a b c)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure
        ProcEnds
          { procStdio :: Poles (K IO) a b
procStdio = StdPorts a b c -> Poles (K IO) a b
forall a b c. StdPorts a b c -> Poles (K IO) a b
stdioEnds StdPorts a b c
pp,
            procStderr :: Poles (K IO) a c
procStderr = StdPorts a b c -> Poles (K IO) a c
forall a b c. StdPorts a b c -> Poles (K IO) a c
stderrEnds StdPorts a b c
pp,
            procClose :: IO ()
procClose = StdPorts a b c -> IO ()
forall a b c. StdPorts a b c -> IO ()
stdClose StdPorts a b c
pp
          }

-- | The wire view of 'StdPorts': one nested 'tensor' morphism.
portsEnds :: StdPorts a b c -> Trace (,) (K IO) (a, ((), ())) ((), (b, c))
portsEnds :: forall a b c.
StdPorts a b c -> Trace (,) (K IO) (a, ((), ())) ((), (b, c))
portsEnds StdPorts a b c
pp = Syntax (SigCompose :+: SigYank (,)) (K IO) a ()
-> Syntax (SigCompose :+: SigYank (,)) (K IO) ((), ()) (b, c)
-> Syntax
     (SigCompose :+: SigYank (,)) (K IO) (a, ((), ())) ((), (b, c))
forall a b c d.
Syntax (SigCompose :+: SigYank (,)) (K IO) a b
-> Syntax (SigCompose :+: SigYank (,)) (K IO) c d
-> Syntax (SigCompose :+: SigYank (,)) (K IO) (a, c) (b, d)
forall {k} (t :: k -> k -> k) (arr :: k -> k -> *) (a :: k)
       (b :: k) (c :: k) (d :: k).
Tensor t arr =>
arr a b -> arr c d -> arr (t a c) (t b d)
tensor Syntax (SigCompose :+: SigYank (,)) (K IO) a ()
commitM (Syntax (SigCompose :+: SigYank (,)) (K IO) () b
-> Syntax (SigCompose :+: SigYank (,)) (K IO) () c
-> Syntax (SigCompose :+: SigYank (,)) (K IO) ((), ()) (b, c)
forall a b c d.
Syntax (SigCompose :+: SigYank (,)) (K IO) a b
-> Syntax (SigCompose :+: SigYank (,)) (K IO) c d
-> Syntax (SigCompose :+: SigYank (,)) (K IO) (a, c) (b, d)
forall {k} (t :: k -> k -> k) (arr :: k -> k -> *) (a :: k)
       (b :: k) (c :: k) (d :: k).
Tensor t arr =>
arr a b -> arr c d -> arr (t a c) (t b d)
tensor Syntax (SigCompose :+: SigYank (,)) (K IO) () b
outM Syntax (SigCompose :+: SigYank (,)) (K IO) () c
errM)
  where
    commitM :: Syntax (SigCompose :+: SigYank (,)) (K IO) a ()
commitM = K IO a () -> Syntax (SigCompose :+: SigYank (,)) (K IO) a ()
forall (arr :: * -> * -> *) a b (t :: * -> * -> *).
arr a b -> Trace t arr a b
base (In (K IO) a -> forall x. Out (K IO) x -> K IO a x
forall {k1} {k2} (arr :: k1 -> k2 -> *) (a :: k1).
In arr a -> forall (x :: k2). Out arr x -> arr a x
commit (StdPorts a b c -> In (K IO) a
forall a b c. StdPorts a b c -> In (K IO) a
stdIn StdPorts a b c
pp) Out (K IO) ()
outUIn)
    outM :: Syntax (SigCompose :+: SigYank (,)) (K IO) () b
outM = K IO () b -> Syntax (SigCompose :+: SigYank (,)) (K IO) () b
forall (arr :: * -> * -> *) a b (t :: * -> * -> *).
arr a b -> Trace t arr a b
base (Out (K IO) b -> forall x. In (K IO) x -> K IO x b
forall {k1} {k2} (arr :: k1 -> k2 -> *) (a :: k2).
Out arr a -> forall (x :: k1). In arr x -> arr x a
emit (StdPorts a b c -> Out (K IO) b
forall a b c. StdPorts a b c -> Out (K IO) b
stdOut StdPorts a b c
pp) In (K IO) ()
inUOut)
    errM :: Syntax (SigCompose :+: SigYank (,)) (K IO) () c
errM = K IO () c -> Syntax (SigCompose :+: SigYank (,)) (K IO) () c
forall (arr :: * -> * -> *) a b (t :: * -> * -> *).
arr a b -> Trace t arr a b
base (Out (K IO) c -> forall x. In (K IO) x -> K IO x c
forall {k1} {k2} (arr :: k1 -> k2 -> *) (a :: k2).
Out arr a -> forall (x :: k1). In arr x -> arr x a
emit (StdPorts a b c -> Out (K IO) c
forall a b c. StdPorts a b c -> Out (K IO) c
stdErr StdPorts a b c
pp) In (K IO) ()
inUErr)
    Poles In (K IO) ()
_inUIn Out (K IO) ()
outUIn = Poles (K IO) () ()
forall {k} (bot :: k) (arr :: k -> k -> *).
HasDual bot arr =>
Poles arr bot bot
open
    Poles In (K IO) ()
inUOut Out (K IO) ()
_ = Poles (K IO) () ()
forall {k} (bot :: k) (arr :: k -> k -> *).
HasDual bot arr =>
Poles arr bot bot
open
    Poles In (K IO) ()
inUErr Out (K IO) ()
_ = Poles (K IO) () ()
forall {k} (bot :: k) (arr :: k -> k -> *).
HasDual bot arr =>
Poles arr bot bot
open

-- ---------------------------------------------------------------------------
-- In-memory test harness
-- ---------------------------------------------------------------------------

-- | An in-memory echo repl: commit a token, emit the transformed result.
-- No process, no files.  For fast bus-connector testing.
--
-- >>> pp <- echo pure :: IO (StdPorts String String ())
-- >>> runK (commit (stdIn pp) (companion (open :: Poles (K IO) () ()))) "hi"
-- >>> runK (emit (stdOut pp) (conjoint (open :: Poles (K IO) () ()))) ()
-- "hi"
echo :: (a -> IO a) -> IO (StdPorts a a ())
echo :: forall a. (a -> IO a) -> IO (StdPorts a a ())
echo a -> IO a
f = do
  ref <- Maybe a -> IO (IORef (Maybe a))
forall a. a -> IO (IORef a)
newIORef Maybe a
forall a. Maybe a
Nothing
  let ports =
        StdPorts
          { stdIn :: In (K IO) a
stdIn = (forall x. Out (K IO) x -> K IO a x) -> In (K IO) a
forall {k} {k1} (arr :: k -> k1 -> *) (a :: k).
(forall (x :: k1). Out arr x -> arr a x) -> In arr a
In ((forall x. Out (K IO) x -> K IO a x) -> In (K IO) a)
-> (forall x. Out (K IO) x -> K IO a x) -> In (K IO) a
forall a b. (a -> b) -> a -> b
$ \Out (K IO) x
o -> (a -> IO x) -> K IO a x
forall {k} (m :: k -> *) a (b :: k). (a -> m b) -> K m a b
K ((a -> IO x) -> K IO a x) -> (a -> IO x) -> K IO a x
forall a b. (a -> b) -> a -> b
$ \a
a -> do
              result <- a -> IO a
f a
a
              writeIORef ref (Just result)
              runK (emit o (stdIn ports)) a,
            stdOut :: Out (K IO) a
stdOut = (forall x. In (K IO) x -> K IO x a) -> Out (K IO) a
forall {k} {k1} (arr :: k -> k1 -> *) (a :: k1).
(forall (x :: k). In arr x -> arr x a) -> Out arr a
Out ((forall x. In (K IO) x -> K IO x a) -> Out (K IO) a)
-> (forall x. In (K IO) x -> K IO x a) -> Out (K IO) a
forall a b. (a -> b) -> a -> b
$ \In (K IO) x
_ -> (x -> IO a) -> K IO x a
forall {k} (m :: k -> *) a (b :: k). (a -> m b) -> K m a b
K ((x -> IO a) -> K IO x a) -> (x -> IO a) -> K IO x a
forall a b. (a -> b) -> a -> b
$ \x
_ -> do
              m <- IORef (Maybe a) -> IO (Maybe a)
forall a. IORef a -> IO a
readIORef IORef (Maybe a)
ref
              writeIORef ref Nothing
              pure (fromMaybe (error "echo: no input yet") m),
            stdErr :: Out (K IO) ()
stdErr = (forall x. In (K IO) x -> K IO x ()) -> Out (K IO) ()
forall {k} {k1} (arr :: k -> k1 -> *) (a :: k1).
(forall (x :: k). In arr x -> arr x a) -> Out arr a
Out ((forall x. In (K IO) x -> K IO x ()) -> Out (K IO) ())
-> (forall x. In (K IO) x -> K IO x ()) -> Out (K IO) ()
forall a b. (a -> b) -> a -> b
$ \In (K IO) x
_ -> (x -> IO ()) -> K IO x ()
forall {k} (m :: k -> *) a (b :: k). (a -> m b) -> K m a b
K ((x -> IO ()) -> K IO x ()) -> (x -> IO ()) -> K IO x ()
forall a b. (a -> b) -> a -> b
$ \x
_ -> () -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (),
            stdClose :: IO ()
stdClose = () -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
          }
  pure ports