execution_plane_process

defmodule ExecutionPlane.Process.Transport do
  @moduledoc """
  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.
  """

  alias ExecutionPlane.{Command, ProcessExit, TaskSupport}
  alias ExecutionPlane.Process.Transport.{Delivery, Error, Info, RunResult}
  alias ExecutionPlane.Process.Transport.Surface
  alias ExecutionPlane.Process.Transport.Surface.Capabilities

  @default_call_timeout_ms 5_000
  @default_force_close_timeout_ms 5_000

  @typedoc "Opaque transport reference."
  @type t :: pid()

  @typedoc "Mailbox tag carried on tagged delivery."
  @type subscription_tag :: pid() | reference()

  @typedoc "Caller-supplied tag for explicit subscriptions."
  @type explicit_subscription_tag :: reference()

  @typedoc "The tagged event atom prefix."
  @type event_tag :: atom()

  @typedoc "Generic execution-surface placement kind."
  @type surface_kind :: :local_subprocess | :ssh_exec | :guest_bridge | :lower_simulation

  @typedoc "Transport events delivered to subscribers."
  @type message :: {event_tag(), subscription_tag(), extracted_event()}

  @typedoc "Normalized transport event payload extracted from a mailbox message."
  @type extracted_event ::
          {:message, binary()}
          | {:data, binary()}
          | {:error, Error.t()}
          | {:stderr, binary()}
          | {:exit, ProcessExit.t()}

  @callback start(keyword()) :: {:ok, t()} | {:error, {:transport, Error.t()}}
  @callback start_link(keyword()) :: {:ok, t()} | {:error, {:transport, Error.t()}}
  @callback run(Command.t(), keyword()) ::
              {:ok, RunResult.t()} | {:error, {:transport, Error.t()}}
  @callback send(t(), iodata() | map() | list()) :: :ok | {:error, {:transport, Error.t()}}
  @callback subscribe(t(), pid()) :: :ok | {:error, {:transport, Error.t()}}
  @callback subscribe(t(), pid(), explicit_subscription_tag()) ::
              :ok | {:error, {:transport, Error.t()}}
  @callback unsubscribe(t(), pid()) :: :ok
  @callback close(t()) :: :ok
  @callback force_close(t()) :: :ok | {:error, {:transport, Error.t()}}
  @callback interrupt(t()) :: :ok | {:error, {:transport, Error.t()}}
  @callback status(t()) :: :connected | :disconnected | :error
  @callback end_input(t()) :: :ok | {:error, {:transport, Error.t()}}
  @callback stderr(t()) :: binary()
  @callback info(t()) :: Info.t()

  @doc """
  Starts the default raw transport implementation.
  """
  @spec start(keyword()) :: {:ok, t()} | {:error, {:transport, Error.t()}}
  def start(opts) when is_list(opts) do
    case Surface.resolve(opts) do
      {:ok, %{dispatch: dispatch, adapter_options: adapter_options} = resolved} ->
        with :ok <- validate_start_capabilities(resolved, opts) do
          dispatch.start.(adapter_options)
        end

      {:error, reason} ->
        transport_error(reason)
    end
  end

  @doc """
  Starts the default raw transport implementation and links it to the caller.
  """
  @spec start_link(keyword()) :: {:ok, t()} | {:error, {:transport, Error.t()}}
  def start_link(opts) when is_list(opts) do
    case Surface.resolve(opts) do
      {:ok, %{dispatch: dispatch, adapter_options: adapter_options} = resolved} ->
        with :ok <- validate_start_capabilities(resolved, opts) do
          dispatch.start_link.(adapter_options)
        end

      {:error, reason} ->
        transport_error(reason)
    end
  end

  @doc """
  Runs a one-shot non-PTY command and captures exact stdout, stderr, and exit
  data.
  """
  @spec run(Command.t(), keyword()) :: {:ok, RunResult.t()} | {:error, {:transport, Error.t()}}
  def run(%Command{} = command, opts \\ []) when is_list(opts) do
    case Surface.resolve(opts) do
      {:ok, %{dispatch: dispatch, adapter_options: adapter_options} = resolved} ->
        with :ok <- validate_run_capabilities(resolved, command) do
          dispatch.run.(command, adapter_options)
        end

      {:error, reason} ->
        transport_error(reason)
    end
  end

  @doc """
  Sends data to the subprocess stdin.
  """
  @spec send(t(), iodata() | map() | list()) :: :ok | {:error, {:transport, Error.t()}}
  def send(transport, message) when is_pid(transport) do
    case safe_call(transport, {:send, message}) do
      {:ok, result} -> result
      {:error, reason} -> transport_error(reason)
    end
  end

  @doc """
  Subscribes a process using the subscriber pid as the default mailbox tag.
  """
  @spec subscribe(t(), pid()) :: :ok | {:error, {:transport, Error.t()}}
  def subscribe(transport, pid) when is_pid(transport) and is_pid(pid) do
    do_subscribe(transport, pid, pid)
  end

  @doc """
  Subscribes a process with an explicit reference tag.
  """
  @spec subscribe(t(), pid(), explicit_subscription_tag()) ::
          :ok | {:error, {:transport, Error.t()}}
  def subscribe(transport, pid, tag)
      when is_pid(transport) and is_pid(pid) and is_reference(tag),
      do: do_subscribe(transport, pid, tag)

  def subscribe(_transport, _pid, tag) do
    transport_error(Error.invalid_options({:invalid_subscriber, tag}))
  end

  @doc """
  Removes a subscriber.
  """
  @spec unsubscribe(t(), pid()) :: :ok
  def unsubscribe(transport, pid) when is_pid(transport) and is_pid(pid) do
    case safe_call(transport, {:unsubscribe, pid}) do
      {:ok, :ok} -> :ok
      {:error, _reason} -> :ok
    end
  end

  @doc """
  Stops the transport.
  """
  @spec close(t()) :: :ok
  def close(transport) when is_pid(transport) do
    GenServer.stop(transport, :normal)
  catch
    :exit, {:noproc, _} -> :ok
    :exit, :noproc -> :ok
  end

  @doc """
  Forces the subprocess down immediately.
  """
  @spec force_close(t()) :: :ok | {:error, {:transport, Error.t()}}
  def force_close(transport) when is_pid(transport) do
    do_force_close(transport, @default_force_close_timeout_ms)
  end

  @doc """
  Sends SIGINT to the subprocess.
  """
  @spec interrupt(t()) :: :ok | {:error, {:transport, Error.t()}}
  def interrupt(transport) when is_pid(transport) do
    case safe_call(transport, :interrupt) do
      {:ok, result} ->
        normalize_call_result(result)

      {:error, reason} ->
        transport_error(reason)
    end
  end

  @doc """
  Returns transport connectivity status.
  """
  @spec status(t()) :: :connected | :disconnected | :error
  def status(transport) when is_pid(transport) do
    case safe_call(transport, :status) do
      {:ok, status} when status in [:connected, :disconnected, :error] -> status
      {:ok, _other} -> :error
      {:error, _reason} -> :disconnected
    end
  end

  @doc """
  Closes stdin for EOF-driven CLIs.

  Pipe-backed transports send `:eof`; PTY-backed transports send the terminal
  EOF byte (`Ctrl-D`).
  """
  @spec end_input(t()) :: :ok | {:error, {:transport, Error.t()}}
  def end_input(transport) when is_pid(transport) do
    case safe_call(transport, :end_input) do
      {:ok, result} -> result
      {:error, reason} -> transport_error(reason)
    end
  end

  @doc """
  Returns the stderr ring buffer tail.
  """
  @spec stderr(t()) :: binary()
  def stderr(transport) when is_pid(transport) do
    case safe_call(transport, :stderr) do
      {:ok, data} when is_binary(data) -> data
      _ -> ""
    end
  end

  @doc """
  Returns the current transport metadata snapshot.
  """
  @spec info(t()) :: Info.t()
  def info(transport) when is_pid(transport) do
    case safe_call(transport, :info) do
      {:ok, %Info{} = info} -> info
      _other -> Info.disconnected()
    end
  end

  defp validate_start_capabilities(
         %{adapter_capabilities: %Capabilities{} = capabilities, surface: surface},
         opts
       )
       when is_list(opts) do
    [
      {capabilities.supports_streaming_stdio?, :streaming_stdio},
      {not Keyword.get(opts, :pty?, false) or capabilities.supports_pty?, :pty},
      {not has_user?(opts) or capabilities.supports_user?, :user},
      {not has_env?(opts) or capabilities.supports_env?, :env},
      {not has_cwd?(opts) or capabilities.supports_cwd?, :cwd}
    ]
    |> validate_family_capabilities(surface.surface_kind)
  end

  defp validate_run_capabilities(
         %{adapter_capabilities: %Capabilities{} = capabilities, surface: surface},
         %Command{} = command
       ) do
    [
      {capabilities.supports_run?, :run},
      {not has_user?(command) or capabilities.supports_user?, :user},
      {not has_env?(command) or capabilities.supports_env?, :env},
      {not has_cwd?(command) or capabilities.supports_cwd?, :cwd}
    ]
    |> validate_family_capabilities(surface.surface_kind)
  end

  defp validate_family_capabilities(checks, surface_kind) when is_list(checks) do
    Enum.reduce_while(checks, :ok, fn {allowed?, capability}, :ok ->
      case require_family_capability(allowed?, capability, surface_kind) do
        :ok -> {:cont, :ok}
        error -> {:halt, error}
      end
    end)
  end

  defp require_family_capability(true, _capability, _surface_kind), do: :ok

  defp require_family_capability(false, capability, surface_kind) do
    transport_error(Error.unsupported_capability(capability, surface_kind))
  end

  defp has_user?(opts) when is_list(opts) do
    case Keyword.get(opts, :user) do
      user when is_binary(user) and user != "" -> true
      _other -> false
    end
  end

  defp has_user?(%Command{user: user}) when is_binary(user) and user != "", do: true
  defp has_user?(%Command{}), do: false

  defp has_env?(opts) when is_list(opts) do
    clear_env? = Keyword.get(opts, :clear_env?, false)

    case Keyword.get(opts, :env, %{}) do
      env when is_map(env) -> map_size(env) > 0 or clear_env?
      env when is_list(env) -> env != [] or clear_env?
      _other -> clear_env?
    end
  end

  defp has_env?(%Command{env: env, clear_env?: clear_env?}) when is_map(env),
    do: map_size(env) > 0 or clear_env?

  defp has_env?(%Command{}), do: false

  defp has_cwd?(opts) when is_list(opts) do
    case Keyword.get(opts, :cwd) do
      cwd when is_binary(cwd) and cwd != "" -> true
      _other -> false
    end
  end

  defp has_cwd?(%Command{cwd: cwd}) when is_binary(cwd) and cwd != "", do: true
  defp has_cwd?(%Command{}), do: false

  @doc """
  Extracts a normalized transport event from a tagged mailbox message.
  """
  @spec extract_event(term()) :: {:ok, extracted_event()} | :error
  def extract_event({event_tag, _tag, event}) when is_atom(event_tag),
    do: extract_tagged_event(event)

  def extract_event(_message), do: :error

  @doc """
  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.
  """
  @spec extract_event(term(), subscription_tag()) :: {:ok, extracted_event()} | :error
  def extract_event({event_tag, tag, event}, tag)
      when is_atom(event_tag) and (is_pid(tag) or is_reference(tag)) do
    extract_tagged_event(event)
  end

  def extract_event(_message, _tag), do: :error

  @doc """
  Returns stable mailbox-delivery metadata for the current transport snapshot.
  """
  @spec delivery_info(t()) :: Delivery.t()
  def delivery_info(transport) do
    case info(transport) do
      %Info{delivery: %Delivery{} = delivery} -> delivery
      _other -> Delivery.new(:execution_plane_process)
    end
  end

  defp extract_tagged_event({:message, line}) when is_binary(line), do: {:ok, {:message, line}}
  defp extract_tagged_event({:data, chunk}) when is_binary(chunk), do: {:ok, {:data, chunk}}
  defp extract_tagged_event({:error, %Error{} = error}), do: {:ok, {:error, error}}
  defp extract_tagged_event({:stderr, chunk}) when is_binary(chunk), do: {:ok, {:stderr, chunk}}
  defp extract_tagged_event({:exit, %ProcessExit{} = exit}), do: {:ok, {:exit, exit}}
  defp extract_tagged_event(_event), do: :error

  defp do_subscribe(transport, pid, tag)
       when is_pid(transport) and is_pid(pid) and (is_pid(tag) or is_reference(tag)) do
    case safe_call(transport, {:subscribe, pid, tag}) do
      {:ok, result} -> result
      {:error, reason} -> transport_error(reason)
    end
  end

  defp safe_call(transport, message, timeout \\ @default_call_timeout_ms)

  defp safe_call(transport, message, timeout)
       when is_pid(transport) and is_integer(timeout) and timeout >= 0 do
    case TaskSupport.async_nolink(fn ->
           try do
             {:ok, GenServer.call(transport, message, :infinity)}
           catch
             :exit, reason -> {:error, normalize_call_exit(reason)}
           end
         end) do
      {:ok, task} ->
        await_task_result(task, timeout)

      {:error, reason} ->
        {:error, normalize_call_task_start_error(reason)}
    end
  end

  defp await_task_result(task, timeout) do
    case TaskSupport.await(task, timeout, :brutal_kill) do
      {:ok, result} -> result
      {:exit, reason} -> {:error, normalize_call_exit(reason)}
      {:error, :timeout} -> {:error, Error.timeout()}
    end
  end

  defp normalize_call_task_start_error(:noproc), do: Error.transport_stopped()
  defp normalize_call_task_start_error(reason), do: Error.call_exit(reason)

  defp normalize_call_exit({:noproc, _}), do: Error.not_connected()
  defp normalize_call_exit(:noproc), do: Error.not_connected()
  defp normalize_call_exit({:normal, _}), do: Error.not_connected()
  defp normalize_call_exit({:shutdown, _}), do: Error.not_connected()
  defp normalize_call_exit({:timeout, _}), do: Error.timeout()
  defp normalize_call_exit(reason), do: Error.call_exit(reason)

  defp normalize_call_result(:ok), do: :ok
  defp normalize_call_result({:error, {:transport, %Error{}}} = error), do: error
  defp normalize_call_result({:error, %Error{} = error}), do: transport_error(error)
  defp normalize_call_result({:error, reason}), do: transport_error(reason)

  defp normalize_call_result(other),
    do: transport_error(Error.transport_error({:unexpected_task_result, other}))

  defp do_force_close(transport, timeout_ms)
       when is_pid(transport) and is_integer(timeout_ms) and timeout_ms >= 0 do
    GenServer.stop(transport, :normal, timeout_ms)
    :ok
  catch
    :exit, reason ->
      transport_error(normalize_force_close_exit(reason))
  end

  defp normalize_force_close_exit({:noproc, _}), do: Error.not_connected()
  defp normalize_force_close_exit(:noproc), do: Error.not_connected()
  defp normalize_force_close_exit({:normal, _}), do: Error.not_connected()
  defp normalize_force_close_exit({:shutdown, _}), do: Error.not_connected()
  defp normalize_force_close_exit({:timeout, {GenServer, :stop, _}}), do: Error.timeout()
  defp normalize_force_close_exit({:timeout, _}), do: Error.timeout()
  defp normalize_force_close_exit(reason), do: Error.call_exit(reason)

  defp transport_error({:transport, %Error{}} = error), do: {:error, error}
  defp transport_error(%Error{} = error), do: {:error, {:transport, error}}
  defp transport_error(reason), do: {:error, {:transport, Error.transport_error(reason)}}
end