defmodule ExecutionPlane.Process.Transport.LowerSimulation do @moduledoc """ Execution Plane-owned process transport simulation surface. The adapter replays configured stdout/stderr/exit frames through the normal process transport contract. It never spawns a process and is intended to be selected by higher-layer configuration, not by public request keywords. """ use GenServer alias ExecutionPlane.{Command, ProcessExit} alias ExecutionPlane.Process.Transport.{ Delivery, Error, Info, RunResult, Surface.Adapter, Surface.Capabilities } @behaviour Adapter @surface_kind :lower_simulation @event_tag :execution_plane_process @transport_option_keys ~w(scenario_ref stdout stdout_frames stderr stderr_frames exit exit_code)a @transport_option_key_aliases Map.new(@transport_option_keys, fn key -> {Atom.to_string(key), key} end) defstruct [ :command, :delivery, :event_tag, :exit, :scenario_ref, :surface_kind, :target_id, :lease_ref, :surface_ref, :boundary_class, :observability, :adapter_capabilities, :effective_capabilities, stdout_frames: [], stderr_frames: [], subscribers: %{}, status: :connected, played?: false, received_input: [] ] @impl Adapter def surface_kind, do: @surface_kind @impl Adapter def capabilities do Capabilities.new!( remote?: false, startup_kind: :attach, path_semantics: :local, supports_run?: true, supports_streaming_stdio?: true, supports_pty?: false, supports_user?: true, supports_env?: true, supports_cwd?: true, interrupt_kind: :none ) end @impl Adapter def normalize_transport_options(options) when is_list(options) do if Keyword.keyword?(options) do case validate_options(options) do {:ok, _scenario} -> {:ok, options} {:error, reason} -> {:error, {:invalid_transport_options, reason}} end else {:error, {:invalid_transport_options, options}} end end def normalize_transport_options(options) when is_map(options) do case Enum.reduce_while(options, [], &normalize_transport_option_pair/2) do :error -> {:error, {:invalid_transport_options, options}} normalized -> normalized |> Enum.reverse() |> normalize_transport_options() end end def normalize_transport_options(options), do: {:error, {:invalid_transport_options, options}} @spec start(keyword()) :: {:ok, pid()} | {:error, {:transport, Error.t()}} def start(opts) when is_list(opts), do: do_start(:start, opts) @spec start_link(keyword()) :: {:ok, pid()} | {:error, {:transport, Error.t()}} def start_link(opts) when is_list(opts), do: do_start(:start_link, opts) @spec run(Command.t(), keyword()) :: {:ok, RunResult.t()} | {:error, {:transport, Error.t()}} def run(%Command{} = command, opts) when is_list(opts) do with {:ok, scenario} <- scenario(opts) do stdout = IO.iodata_to_binary(scenario.stdout_frames) stderr = IO.iodata_to_binary(scenario.stderr_frames) {:ok, %RunResult{ invocation: command, stdout: stdout, stderr: stderr, output: output(stdout, stderr, Keyword.get(opts, :stderr, :separate)), exit: scenario.exit, stderr_mode: Keyword.get(opts, :stderr, :separate) }} end end @impl GenServer def init(opts) do case scenario(opts) do {:ok, scenario} -> delivery = Delivery.new(Keyword.get(opts, :event_tag, @event_tag)) state = %__MODULE__{ command: Keyword.get(opts, :command), delivery: delivery, event_tag: delivery.tagged_event_tag, exit: scenario.exit, scenario_ref: scenario.scenario_ref, surface_kind: Keyword.get(opts, :surface_kind, @surface_kind), target_id: Keyword.get(opts, :target_id), lease_ref: Keyword.get(opts, :lease_ref), surface_ref: Keyword.get(opts, :surface_ref), boundary_class: Keyword.get(opts, :boundary_class), observability: Keyword.get(opts, :observability, %{}), adapter_capabilities: Keyword.get(opts, :adapter_capabilities, capabilities()), effective_capabilities: Keyword.get(opts, :effective_capabilities, capabilities()), stdout_frames: scenario.stdout_frames, stderr_frames: scenario.stderr_frames } |> maybe_put_subscriber(Keyword.get(opts, :subscriber)) send(self(), :playback) {:ok, state} {:error, {:transport, %Error{}} = error} -> {:stop, error} end end @impl GenServer def handle_call({:send, input}, _from, state) do {:reply, :ok, %{state | received_input: [input | state.received_input]}} end def handle_call({:subscribe, pid, tag}, _from, state) do state = put_subscriber(state, pid, tag) if state.played? do replay_to(pid, tag, state) end {:reply, :ok, state} end def handle_call({:unsubscribe, pid}, _from, state) do {:reply, :ok, %{state | subscribers: Map.delete(state.subscribers, pid)}} end def handle_call(:interrupt, _from, state), do: {:reply, :ok, state} def handle_call(:end_input, _from, state), do: {:reply, :ok, state} def handle_call(:stderr, _from, state), do: {:reply, IO.iodata_to_binary(state.stderr_frames), state} def handle_call(:status, _from, state), do: {:reply, state.status, state} def handle_call(:info, _from, state), do: {:reply, info(state), state} @impl GenServer def handle_info(:playback, state) do Enum.each(state.subscribers, fn {pid, tag} -> replay_to(pid, tag, state) end) {:noreply, %{state | played?: true}} end defp do_start(fun, opts) do case apply(GenServer, fun, [__MODULE__, opts]) do {:ok, pid} -> {:ok, pid} {:error, {:transport, %Error{}} = error} -> {:error, error} {:error, reason} -> transport_error(Error.startup_failed(reason)) end end defp scenario(opts) do opts |> Keyword.get(:transport_options, []) |> validate_options() |> case do {:ok, scenario} -> {:ok, scenario} {:error, reason} -> invalid_options(reason) end end defp validate_options(options) when is_list(options) do if Keyword.keyword?(options) do with {:ok, scenario_ref} <- required_string(options, :scenario_ref), {:ok, stdout_frames} <- frames(options, :stdout, :stdout_frames), {:ok, stderr_frames} <- frames(options, :stderr, :stderr_frames), {:ok, exit} <- exit(options, stderr_frames) do {:ok, %{ scenario_ref: scenario_ref, stdout_frames: stdout_frames, stderr_frames: stderr_frames, exit: exit }} end else {:error, {:invalid_transport_options, options}} end end defp validate_options(options), do: {:error, {:invalid_transport_options, options}} defp required_string(options, key) do case Keyword.get(options, key) do value when is_binary(value) and value != "" -> {:ok, value} other -> {:error, {:missing_required_option, key, other}} end end defp frames(options, scalar_key, list_key) do case {Keyword.get(options, list_key), Keyword.get(options, scalar_key)} do {nil, nil} -> {:ok, []} {nil, value} when is_binary(value) -> {:ok, [value]} {values, _value} when is_list(values) -> validate_frame_list(values, list_key) {values, value} -> {:error, {:invalid_frames, list_key, values || value}} end end defp validate_frame_list(values, _key) do if Enum.all?(values, &is_binary/1) do {:ok, values} else {:error, {:invalid_frames, values}} end end defp exit(options, stderr_frames) do case Keyword.get(options, :exit, Keyword.get(options, :exit_code, 0)) do %ProcessExit{} = exit -> {:ok, exit} code when is_integer(code) -> {:ok, ProcessExit.from_reason(code, stderr: IO.iodata_to_binary(stderr_frames))} :normal -> {:ok, ProcessExit.from_reason(:normal, stderr: IO.iodata_to_binary(stderr_frames))} other -> {:error, {:invalid_exit, other}} end end defp output(stdout, _stderr, :separate), do: stdout defp output(stdout, stderr, :stdout), do: stdout <> stderr defp output(stdout, _stderr, _mode), do: stdout defp maybe_put_subscriber(state, nil), do: state defp maybe_put_subscriber(state, {pid, tag}) when is_pid(pid), do: put_subscriber(state, pid, tag) defp maybe_put_subscriber(state, pid) when is_pid(pid), do: put_subscriber(state, pid, pid) defp maybe_put_subscriber(state, _other), do: state defp put_subscriber(state, pid, tag) when is_pid(pid) do %{state | subscribers: Map.put(state.subscribers, pid, tag)} end defp replay_to(pid, tag, state) do Enum.each(state.stdout_frames, fn frame -> send(pid, {state.event_tag, tag, {:message, frame}}) end) Enum.each(state.stderr_frames, fn frame -> send(pid, {state.event_tag, tag, {:stderr, frame}}) end) send(pid, {state.event_tag, tag, {:exit, state.exit}}) end defp info(state) do %Info{ invocation: state.command, pid: self(), os_pid: nil, surface_kind: state.surface_kind, target_id: state.target_id, lease_ref: state.lease_ref, surface_ref: state.surface_ref, boundary_class: state.boundary_class, observability: state.observability, adapter_capabilities: state.adapter_capabilities, effective_capabilities: state.effective_capabilities, status: state.status, stdout_mode: :line, stdin_mode: :line, pty?: false, interrupt_mode: :signal, stderr: IO.iodata_to_binary(state.stderr_frames), delivery: state.delivery, adapter_metadata: %{ lower_simulation?: true, scenario_ref: state.scenario_ref, side_effect_policy: "deny_process_spawn", side_effect_result: "not_attempted" } } end defp normalize_transport_option_pair({key, value}, acc) when is_atom(key) do {:cont, [{key, value} | acc]} end defp normalize_transport_option_pair({key, value}, acc) when is_binary(key) do case Map.fetch(@transport_option_key_aliases, key) do {:ok, normalized_key} -> {:cont, [{normalized_key, value} | acc]} :error -> {:halt, :error} end end defp normalize_transport_option_pair(_other, _acc), do: {:halt, :error} defp invalid_options(reason) do transport_error(Error.invalid_options(reason)) end defp transport_error(%Error{} = error), do: {:error, {:transport, error}} end