Safe Haskell	None
Language	GHC2024

Circuit.Agent.StdPorts

Contents

Configuration
Stream marks
Process ports
The mark machine
Ends / seat view (client-facing)
In-memory test harness

Description

Process ports: stdin stdout stderr as free dual poles.

A StdPorts handle is a persistent child process viewed through three independent Poles seats plus a close action. Pipes all the way down: no FIFO, no log files, no byte-offset polling.

Internal IO is ByteString. Constructors take encodedecode 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 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.

Synopsis

data ProcConfig = ProcConfig {
- procCommand :: String
- procArgs :: [String]
- procWorkingDir :: FilePath
- procMarks :: ProcMarks
}
defaultProcConfig :: ProcConfig
newtype ProcMarks = ProcMarks [Text]
splitFrame :: ProcMarks -> ByteString -> Maybe (ByteString, ByteString)
lineMarks :: ProcMarks
ghciMarks :: ProcMarks
hermesMarks :: ProcMarks
sseMarks :: ProcMarks
data StdPorts a b c = StdPorts {
- stdIn :: In (K IO) a
- stdOut :: Out (K IO) b
- stdErr :: Out (K IO) c
- stdClose :: IO ()
}
openStdPorts :: (a -> ByteString) -> (ByteString -> a) -> ProcConfig -> Trace Either (K IO) () (StdPorts a a a)
frameAgent :: ProcMarks -> (ByteString -> a) -> Agent (->) (ByteString, [a]) (Maybe ByteString) [a]
frameProcess :: ProcMarks -> (ByteString -> a) -> Process (Maybe ByteString) [a]
data ProcEnds a b c = ProcEnds {
- procStdio :: Poles (K IO) a b
- procStderr :: Poles (K IO) a c
- procClose :: IO ()
}
stdioEnds :: StdPorts a b c -> Poles (K IO) a b
stderrEnds :: StdPorts a b c -> Poles (K IO) a c
openProc :: (a -> ByteString) -> (ByteString -> a) -> ProcConfig -> Trace Either (K IO) () (ProcEnds a a a)
portsEnds :: StdPorts a b c -> Trace (,) (K IO) (a, ((), ())) ((), (b, c))
echo :: (a -> IO a) -> IO (StdPorts a a ())

Configuration

data ProcConfig Source #

Constructors

ProcConfig
Fields procCommand :: String procArgs :: [String] procWorkingDir :: FilePath procMarks :: ProcMarks Boundary grammar of the stdout stream. Stderr is line-framed (`lineMarks`); stderr is diagnostics, not dialogue.

Instances

Instances details

Eq ProcConfig Source #
Instance details Defined in Circuit.Agent.StdPorts Methods (==) :: ProcConfig -> ProcConfig -> Bool # (/=) :: ProcConfig -> ProcConfig -> Bool #
Show ProcConfig Source #
Instance details Defined in Circuit.Agent.StdPorts Methods showsPrec :: Int -> ProcConfig -> ShowS # show :: ProcConfig -> String # showList :: [ProcConfig] -> ShowS #

defaultProcConfig :: ProcConfig Source #

Stream marks

newtype ProcMarks Source #

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.

Constructors

ProcMarks [Text]

Instances

Instances details

Eq ProcMarks Source #
Instance details Defined in Circuit.Agent.StdPorts Methods (==) :: ProcMarks -> ProcMarks -> Bool # (/=) :: ProcMarks -> ProcMarks -> Bool #
Show ProcMarks Source #
Instance details Defined in Circuit.Agent.StdPorts Methods showsPrec :: Int -> ProcMarks -> ShowS # show :: ProcMarks -> String # showList :: [ProcMarks] -> ShowS #

splitFrame :: ProcMarks -> ByteString -> Maybe (ByteString, ByteString) Source #

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.

lineMarks :: ProcMarks Source #

Line framing: the newline is the mark, one frame per line. For line protocols (ACP's JSON-RPC) and for stderr diagnostics.

ghciMarks :: ProcMarks Source #

The ghci prompt grammar.

hermesMarks :: ProcMarks Source #

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.)

sseMarks :: ProcMarks Source #

The SSE (Server-Sent Events) level-0 grammar: the blank line ends an event frame. Payload is the event:/data: block, mark stripped.

Process ports

data StdPorts a b c Source #

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.

Constructors

StdPorts
Fields stdIn :: In (K IO) a stdOut :: Out (K IO) b stdErr :: Out (K IO) c stdClose :: IO ()

openStdPorts :: (a -> ByteString) -> (ByteString -> a) -> ProcConfig -> Trace Either (K IO) () (StdPorts a a a) Source #

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.

The mark machine

frameAgent :: ProcMarks -> (ByteString -> a) -> Agent (->) (ByteString, [a]) (Maybe ByteString) [a] Source #

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"]]

frameProcess :: ProcMarks -> (ByteString -> a) -> Process (Maybe ByteString) [a] Source #

The same machine as a Process: chunk stream in, frame lists out, state carried implicitly.

Ends / seat view (client-facing)

data ProcEnds a b c Source #

Client view of a process: two Poles sharing stdin, plus resource close.

Constructors

ProcEnds
Fields procStdio :: Poles (K IO) a b procStderr :: Poles (K IO) a c procClose :: IO ()

stdioEnds :: StdPorts a b c -> Poles (K IO) a b Source #

Stdin commit + stdout emit as matched Poles.

stderrEnds :: StdPorts a b c -> Poles (K IO) a c Source #

Stdin commit + stderr emit as matched Poles.

openProc :: (a -> ByteString) -> (ByteString -> a) -> ProcConfig -> Trace Either (K IO) () (ProcEnds a a a) Source #

Open a process and return the dual-seat client view as a Trace Either.

portsEnds :: StdPorts a b c -> Trace (,) (K IO) (a, ((), ())) ((), (b, c)) Source #

The wire view of StdPorts: one nested tensor morphism.

In-memory test harness

echo :: (a -> IO a) -> IO (StdPorts a a ()) Source #

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"