ExecutionPlane.Process.Transport behaviour (execution_plane v0.1.0)

Behaviour for the raw subprocess transport layer.

In addition to the long-lived subscriber-driven transport API, the transport layer also owns synchronous non-PTY command execution through run/2.

Subscribers receive tagged events only:

{event_tag, subscriber_pid, {:message, line}} for subscribe/2
{event_tag, ref, {:message, line}}
{event_tag, tag, {:data, chunk}}
{event_tag, tag, {:error, %ExecutionPlane.Process.Transport.Error{}}}
{event_tag, tag, {:stderr, chunk}}
{event_tag, tag, {:exit, %ExecutionPlane.ProcessExit{}}}

where tag is the subscriber pid when subscribe/2 is used and an explicit reference when subscribe/3 is used.

When :replay_stderr_on_subscribe? is enabled at startup, newly attached subscribers also receive the retained stderr tail immediately after subscription. When :buffer_events_until_subscribe? is enabled, stdout, stderr, and error events emitted before the first subscriber attaches are replayed in order.

Summary

Types

event_tag()

The tagged event atom prefix.

explicit_subscription_tag()

Caller-supplied tag for explicit subscriptions.

extracted_event()

Normalized transport event payload extracted from a mailbox message.

message()

Transport events delivered to subscribers.

subscription_tag()

Mailbox tag carried on tagged delivery.

surface_kind()

Generic execution-surface placement kind.

t()

Opaque transport reference.

Callbacks

close(t)

end_input(t)

force_close(t)

info(t)

interrupt(t)

run(t, keyword)

send(t, arg2)

start(keyword)

start_link(keyword)

status(t)

stderr(t)

subscribe(t, pid)

subscribe(t, pid, explicit_subscription_tag)

unsubscribe(t, pid)

Functions

close(transport)

Stops the transport.

delivery_info(transport)

Returns stable mailbox-delivery metadata for the current transport snapshot.

end_input(transport)

Closes stdin for EOF-driven CLIs.

extract_event(arg1)

Extracts a normalized transport event from a tagged mailbox message.

extract_event(arg1, tag)

Extracts a normalized transport event for a matching subscriber tag.

force_close(transport)

Forces the subprocess down immediately.

info(transport)

Returns the current transport metadata snapshot.

interrupt(transport)

Sends SIGINT to the subprocess.

run(command, opts \\ [])

Runs a one-shot non-PTY command and captures exact stdout, stderr, and exit data.

send(transport, message)

Sends data to the subprocess stdin.

start(opts)

Starts the default raw transport implementation.

start_link(opts)

Starts the default raw transport implementation and links it to the caller.

status(transport)

Returns transport connectivity status.

stderr(transport)

Returns the stderr ring buffer tail.

subscribe(transport, pid)

Subscribes a process using the subscriber pid as the default mailbox tag.

subscribe(transport, pid, tag)

Subscribes a process with an explicit reference tag.

unsubscribe(transport, pid)

Removes a subscriber.

Types

event_tag()

@type event_tag() :: atom()

The tagged event atom prefix.

explicit_subscription_tag()

@type explicit_subscription_tag() :: reference()

Caller-supplied tag for explicit subscriptions.

extracted_event()

@type extracted_event() ::
  {:message, binary()}
  | {:data, binary()}
  | {:error, ExecutionPlane.Process.Transport.Error.t()}
  | {:stderr, binary()}
  | {:exit, ExecutionPlane.ProcessExit.t()}

Normalized transport event payload extracted from a mailbox message.

message()

@type message() :: {event_tag(), subscription_tag(), extracted_event()}

Transport events delivered to subscribers.

subscription_tag()

@type subscription_tag() :: pid() | reference()

Mailbox tag carried on tagged delivery.

surface_kind()

@type surface_kind() ::
  :local_subprocess | :ssh_exec | :guest_bridge | :lower_simulation

Generic execution-surface placement kind.

t()

@type t() :: pid()

Opaque transport reference.

Callbacks

close(t)

@callback close(t()) :: :ok

end_input(t)

@callback end_input(t()) ::
  :ok | {:error, {:transport, ExecutionPlane.Process.Transport.Error.t()}}

force_close(t)

@callback force_close(t()) ::
  :ok | {:error, {:transport, ExecutionPlane.Process.Transport.Error.t()}}

info(t)

@callback info(t()) :: ExecutionPlane.Process.Transport.Info.t()

interrupt(t)

@callback interrupt(t()) ::
  :ok | {:error, {:transport, ExecutionPlane.Process.Transport.Error.t()}}

run(t, keyword)

@callback run(
  ExecutionPlane.Command.t(),
  keyword()
) ::
  {:ok, ExecutionPlane.Process.Transport.RunResult.t()}
  | {:error, {:transport, ExecutionPlane.Process.Transport.Error.t()}}

send(t, arg2)

@callback send(t(), iodata() | map() | list()) ::
  :ok | {:error, {:transport, ExecutionPlane.Process.Transport.Error.t()}}

start(keyword)

@callback start(keyword()) ::
  {:ok, t()}
  | {:error, {:transport, ExecutionPlane.Process.Transport.Error.t()}}

start_link(keyword)

@callback start_link(keyword()) ::
  {:ok, t()}
  | {:error, {:transport, ExecutionPlane.Process.Transport.Error.t()}}

status(t)

@callback status(t()) :: :connected | :disconnected | :error

stderr(t)

@callback stderr(t()) :: binary()

unsubscribe(t, pid)

@callback unsubscribe(t(), pid()) :: :ok

Functions

close(transport)

@spec close(t()) :: :ok

Stops the transport.

delivery_info(transport)

@spec delivery_info(t()) :: ExecutionPlane.Process.Transport.Delivery.t()

Returns stable mailbox-delivery metadata for the current transport snapshot.

end_input(transport)

@spec end_input(t()) ::
  :ok | {:error, {:transport, ExecutionPlane.Process.Transport.Error.t()}}

Closes stdin for EOF-driven CLIs.

Pipe-backed transports send :eof; PTY-backed transports send the terminal EOF byte (Ctrl-D).

extract_event(arg1)

@spec extract_event(term()) :: {:ok, extracted_event()} | :error

Extracts a normalized transport event from a tagged mailbox message.

extract_event(arg1, tag)

@spec extract_event(term(), subscription_tag()) :: {:ok, extracted_event()} | :error

Extracts a normalized transport event for a matching subscriber tag.

This is the stable core-owned way for adapters to consume tagged transport delivery without hard-coding the configured outer event atom.

force_close(transport)

@spec force_close(t()) ::
  :ok | {:error, {:transport, ExecutionPlane.Process.Transport.Error.t()}}

Forces the subprocess down immediately.

info(transport)

@spec info(t()) :: ExecutionPlane.Process.Transport.Info.t()

Returns the current transport metadata snapshot.

interrupt(transport)

@spec interrupt(t()) ::
  :ok | {:error, {:transport, ExecutionPlane.Process.Transport.Error.t()}}

Sends SIGINT to the subprocess.

run(command, opts \\ [])

@spec run(
  ExecutionPlane.Command.t(),
  keyword()
) ::
  {:ok, ExecutionPlane.Process.Transport.RunResult.t()}
  | {:error, {:transport, ExecutionPlane.Process.Transport.Error.t()}}

Runs a one-shot non-PTY command and captures exact stdout, stderr, and exit data.

send(transport, message)

@spec send(t(), iodata() | map() | list()) ::
  :ok | {:error, {:transport, ExecutionPlane.Process.Transport.Error.t()}}

Sends data to the subprocess stdin.

start(opts)

@spec start(keyword()) ::
  {:ok, t()}
  | {:error, {:transport, ExecutionPlane.Process.Transport.Error.t()}}

Starts the default raw transport implementation.

start_link(opts)

@spec start_link(keyword()) ::
  {:ok, t()}
  | {:error, {:transport, ExecutionPlane.Process.Transport.Error.t()}}

Starts the default raw transport implementation and links it to the caller.

status(transport)

@spec status(t()) :: :connected | :disconnected | :error

Returns transport connectivity status.

stderr(transport)

@spec stderr(t()) :: binary()

Returns the stderr ring buffer tail.

unsubscribe(transport, pid)

@spec unsubscribe(t(), pid()) :: :ok

Removes a subscriber.