agent_session_manager

defmodule ASM.Execution.Config do
  @moduledoc """
  Normalized execution-mode, execution-surface, and execution-environment
  configuration with precedence-aware merging.

  This module is the explicit carrier boundary between the ASM kernel, the
  frozen lower contract packet, and the Wave 5 durable session vocabulary:

  - `BoundarySessionDescriptor.v1`
  - `ExecutionRoute.v1`
  - `AttachGrant.v1`
  - `CredentialHandleRef.v1`
  - `ExecutionEvent.v1`
  - `ExecutionOutcome.v1`
  - `ProcessExecutionIntent.v1`
  - `JsonRpcExecutionIntent.v1`

  The named boundary metadata groups carried above that packet are:

  - `descriptor`
  - `route`
  - `attach_grant`
  - `replay`
  - `approval`
  - `callback`
  - `identity`

  `execution_surface` and `execution_environment` remain the family-facing
  mapped carrier IR here. The detailed lower intent interiors stay provisional
  until Wave 3 prove-out.
  """

  alias ASM.{Error, Execution.Environment, Permission}
  alias ASM.Schema.RemoteNode, as: RemoteNodeSchema
  alias CliSubprocessCore.ExecutionSurface

  @boundary_contract_keys [
    "descriptor",
    "route",
    "attach_grant",
    "replay",
    "approval",
    "callback",
    "identity"
  ]
  @execution_plane_contracts [
    "BoundarySessionDescriptor.v1",
    "ExecutionRoute.v1",
    "AttachGrant.v1",
    "CredentialHandleRef.v1",
    "ExecutionEvent.v1",
    "ExecutionOutcome.v1",
    "ProcessExecutionIntent.v1",
    "JsonRpcExecutionIntent.v1"
  ]
  @provisional_minimal_lane_contracts [
    "ProcessExecutionIntent.v1",
    "JsonRpcExecutionIntent.v1"
  ]

  @execution_surface_keys [
    :contract_version,
    :surface_kind,
    :transport_options,
    :target_id,
    :lease_ref,
    :surface_ref,
    :boundary_class,
    :observability
  ]
  @valid_execution_modes [:local, :remote_node]
  @valid_bootstrap_modes [:require_prestarted, :ensure_started]
  @legacy_execution_surface_keys [
    :surface_kind,
    :transport_options,
    :lease_ref,
    :surface_ref,
    :target_id,
    :boundary_class,
    :observability
  ]

  @enforce_keys [:execution_mode, :transport_call_timeout_ms]
  defstruct execution_mode: :local,
            transport_call_timeout_ms: 5_000,
            execution_surface: %ExecutionSurface{},
            execution_environment: %Environment{},
            provider_permission_mode: nil,
            remote: nil

  @type remote_t :: %{
          required(:remote_node) => atom(),
          required(:remote_cookie) => atom() | nil,
          required(:remote_connect_timeout_ms) => pos_integer(),
          required(:remote_rpc_timeout_ms) => pos_integer(),
          required(:remote_boot_lease_timeout_ms) => pos_integer(),
          required(:remote_bootstrap_mode) => :require_prestarted | :ensure_started,
          required(:remote_cwd) => String.t() | nil
        }

  @type t :: %__MODULE__{
          execution_mode: :local | :remote_node,
          transport_call_timeout_ms: pos_integer(),
          execution_surface: ExecutionSurface.t(),
          execution_environment: Environment.t(),
          provider_permission_mode: atom() | nil,
          remote: remote_t() | nil
        }

  @spec execution_plane_contracts() :: [String.t(), ...]
  def execution_plane_contracts, do: @execution_plane_contracts

  @spec boundary_contract_keys() :: [String.t(), ...]
  def boundary_contract_keys, do: @boundary_contract_keys

  @spec provisional_minimal_lane_contracts() :: [String.t(), ...]
  def provisional_minimal_lane_contracts, do: @provisional_minimal_lane_contracts

  @spec resolve(keyword(), keyword(), keyword()) :: {:ok, t()} | {:error, Error.t()}
  def resolve(session_stream_opts, run_stream_opts, opts \\ [])
      when is_list(session_stream_opts) and is_list(run_stream_opts) and is_list(opts) do
    app_cfg = app_config()
    session_driver_opts = normalize_keyword(Keyword.get(session_stream_opts, :driver_opts, []))
    run_driver_opts = normalize_keyword(Keyword.get(run_stream_opts, :driver_opts, []))
    merged_driver_opts = Keyword.merge(session_driver_opts, run_driver_opts)

    explicit_driver? = Keyword.get(opts, :explicit_driver?, false)

    with :ok <- reject_legacy_execution_surface_keys(session_stream_opts, run_stream_opts),
         {:ok, execution_mode} <-
           resolve_execution_mode(app_cfg, session_stream_opts, run_stream_opts, explicit_driver?),
         {:ok, transport_call_timeout_ms} <-
           resolve_transport_call_timeout(
             app_cfg,
             session_stream_opts,
             run_stream_opts,
             merged_driver_opts
           ),
         {:ok, session_execution_surface} <- resolve_execution_surface(session_stream_opts),
         {:ok, run_execution_surface} <- resolve_execution_surface(run_stream_opts),
         {:ok, execution_surface} <-
           merge_execution_surfaces(session_execution_surface, run_execution_surface),
         {:ok, session_execution_environment_attrs} <-
           resolve_execution_environment_attrs(session_stream_opts),
         {:ok, run_execution_environment_attrs} <-
           resolve_execution_environment_attrs(run_stream_opts),
         {:ok, execution_environment} <-
           merge_execution_environments(
             session_execution_environment_attrs,
             run_execution_environment_attrs
           ),
         {:ok, provider_permission_mode} <-
           resolve_provider_permission_mode(
             execution_environment.permission_mode,
             Keyword.get(opts, :provider)
           ),
         {:ok, remote} <-
           resolve_remote_config(
             execution_mode,
             app_cfg,
             session_stream_opts,
             run_stream_opts,
             merged_driver_opts
           ) do
      {:ok,
       %__MODULE__{
         execution_mode: execution_mode,
         transport_call_timeout_ms: transport_call_timeout_ms,
         execution_surface: execution_surface,
         execution_environment: execution_environment,
         provider_permission_mode: provider_permission_mode,
         remote: remote
       }}
    end
  end

  @spec to_execution_surface(t()) :: ExecutionSurface.t()
  def to_execution_surface(%__MODULE__{
        execution_surface: %ExecutionSurface{} = execution_surface
      }) do
    execution_surface
  end

  @spec to_execution_environment(t()) :: Environment.t()
  def to_execution_environment(%__MODULE__{
        execution_environment: %Environment{} = execution_environment
      }) do
    execution_environment
  end

  defp resolve_execution_mode(_app_cfg, _session_stream_opts, _run_stream_opts, true) do
    {:ok, :local}
  end

  defp resolve_execution_mode(app_cfg, session_stream_opts, run_stream_opts, false) do
    mode =
      Keyword.get(run_stream_opts, :execution_mode) ||
        Keyword.get(session_stream_opts, :execution_mode) ||
        Keyword.get(app_cfg, :execution_mode, :local)

    if mode in @valid_execution_modes do
      {:ok, mode}
    else
      {:error,
       config_error(
         "invalid execution_mode #{inspect(mode)}; expected one of #{inspect(@valid_execution_modes)}"
       )}
    end
  end

  defp resolve_transport_call_timeout(
         app_cfg,
         session_stream_opts,
         run_stream_opts,
         merged_driver_opts
       ) do
    timeout_ms =
      Keyword.get(merged_driver_opts, :remote_transport_call_timeout_ms) ||
        Keyword.get(merged_driver_opts, :transport_call_timeout_ms) ||
        Keyword.get(run_stream_opts, :transport_call_timeout_ms) ||
        Keyword.get(session_stream_opts, :transport_call_timeout_ms) ||
        Keyword.get(app_cfg, :transport_call_timeout_ms, 5_000)

    case normalize_pos_integer(timeout_ms) do
      {:ok, value} ->
        {:ok, value}

      :error ->
        {:error,
         config_error(
           "transport_call_timeout_ms must be a positive integer, got: #{inspect(timeout_ms)}"
         )}
    end
  end

  defp resolve_remote_config(
         :local,
         _app_cfg,
         _session_stream_opts,
         _run_stream_opts,
         _driver_opts
       ),
       do: {:ok, nil}

  defp resolve_remote_config(
         :remote_node,
         app_cfg,
         session_stream_opts,
         run_stream_opts,
         driver_opts
       ) do
    remote_node =
      Keyword.get(driver_opts, :remote_node) ||
        Keyword.get(run_stream_opts, :remote_node) ||
        Keyword.get(session_stream_opts, :remote_node) ||
        Keyword.get(app_cfg, :remote_node)

    with {:ok, remote_connect_timeout_ms} <-
           normalize_timeout(
             driver_opts,
             run_stream_opts,
             session_stream_opts,
             app_cfg,
             :remote_connect_timeout_ms,
             5_000
           ),
         {:ok, remote_rpc_timeout_ms} <-
           normalize_timeout(
             driver_opts,
             run_stream_opts,
             session_stream_opts,
             app_cfg,
             :remote_rpc_timeout_ms,
             15_000
           ),
         {:ok, remote_boot_lease_timeout_ms} <-
           normalize_timeout(
             driver_opts,
             run_stream_opts,
             session_stream_opts,
             app_cfg,
             :remote_boot_lease_timeout_ms,
             10_000
           ),
         {:ok, remote_bootstrap_mode} <-
           normalize_bootstrap_mode(
             Keyword.get(driver_opts, :remote_bootstrap_mode) ||
               Keyword.get(run_stream_opts, :remote_bootstrap_mode) ||
               Keyword.get(session_stream_opts, :remote_bootstrap_mode) ||
               Keyword.get(app_cfg, :remote_bootstrap_mode, :require_prestarted)
           ) do
      build_remote_config(
        remote_node,
        Keyword.get(driver_opts, :remote_cookie),
        remote_connect_timeout_ms,
        remote_rpc_timeout_ms,
        remote_boot_lease_timeout_ms,
        remote_bootstrap_mode,
        Keyword.get(driver_opts, :remote_cwd)
      )
    end
  end

  defp normalize_timeout(driver_opts, run_stream_opts, session_stream_opts, app_cfg, key, default) do
    value =
      Keyword.get(driver_opts, key) ||
        Keyword.get(run_stream_opts, key) ||
        Keyword.get(session_stream_opts, key) ||
        Keyword.get(app_cfg, key, default)

    case normalize_pos_integer(value) do
      {:ok, timeout} ->
        {:ok, timeout}

      :error ->
        {:error, config_error("#{key} must be a positive integer, got: #{inspect(value)}")}
    end
  end

  defp normalize_bootstrap_mode(mode) when mode in @valid_bootstrap_modes, do: {:ok, mode}

  defp normalize_bootstrap_mode(mode) do
    {:error,
     config_error(
       "remote_bootstrap_mode must be one of #{inspect(@valid_bootstrap_modes)}, got: #{inspect(mode)}"
     )}
  end

  defp normalize_pos_integer(value) when is_integer(value) and value > 0, do: {:ok, value}
  defp normalize_pos_integer(_value), do: :error

  defp build_remote_config(
         remote_node,
         remote_cookie,
         remote_connect_timeout_ms,
         remote_rpc_timeout_ms,
         remote_boot_lease_timeout_ms,
         remote_bootstrap_mode,
         remote_cwd
       ) do
    attrs =
      %{
        remote_connect_timeout_ms: remote_connect_timeout_ms,
        remote_rpc_timeout_ms: remote_rpc_timeout_ms,
        remote_boot_lease_timeout_ms: remote_boot_lease_timeout_ms,
        remote_bootstrap_mode: remote_bootstrap_mode
      }
      |> maybe_put(:remote_node, remote_node)
      |> maybe_put(:remote_cookie, remote_cookie)
      |> maybe_put(:remote_cwd, remote_cwd)

    case RemoteNodeSchema.parse(attrs) do
      {:ok, remote} ->
        {:ok, remote}

      {:error, {:invalid_remote_node_config, details}} ->
        {:error, config_error(details.message)}
    end
  end

  defp maybe_put(map, _key, nil), do: map
  defp maybe_put(map, key, value), do: Map.put(map, key, value)

  defp reject_legacy_execution_surface_keys(session_stream_opts, run_stream_opts) do
    legacy_keys =
      @legacy_execution_surface_keys
      |> Enum.filter(
        &(Keyword.has_key?(session_stream_opts, &1) or Keyword.has_key?(run_stream_opts, &1))
      )

    if legacy_keys == [] do
      :ok
    else
      {:error,
       config_error(
         "legacy execution-surface keys are not supported: " <>
           Enum.map_join(legacy_keys, ", ", &inspect/1) <>
           ". Use :execution_surface instead."
       )}
    end
  end

  defp resolve_execution_surface(opts) when is_list(opts) do
    case Keyword.get(opts, :execution_surface) do
      nil ->
        {:ok, nil}

      %ExecutionSurface{} = execution_surface ->
        normalize_execution_surface_input(execution_surface)

      execution_surface when is_list(execution_surface) ->
        if Keyword.keyword?(execution_surface) do
          normalize_execution_surface_input(execution_surface)
        else
          {:error,
           config_error(
             "execution_surface must be a CliSubprocessCore.ExecutionSurface, keyword list, or map, got: #{inspect(execution_surface)}"
           )}
        end

      %{} = execution_surface ->
        normalize_execution_surface_input(execution_surface)

      execution_surface ->
        {:error,
         config_error(
           "execution_surface must be a CliSubprocessCore.ExecutionSurface, keyword list, or map, got: #{inspect(execution_surface)}"
         )}
    end
  end

  defp merge_execution_surfaces(nil, nil), do: normalize_execution_surface_attrs([])

  defp merge_execution_surfaces(attrs, nil) when is_list(attrs),
    do: normalize_execution_surface_attrs(attrs)

  defp merge_execution_surfaces(nil, attrs) when is_list(attrs),
    do: normalize_execution_surface_attrs(attrs)

  defp merge_execution_surfaces(
         session_execution_surface_attrs,
         run_execution_surface_attrs
       ) do
    session_transport_options =
      Keyword.get(session_execution_surface_attrs, :transport_options, [])

    run_transport_options = Keyword.get(run_execution_surface_attrs, :transport_options, [])
    session_observability = Keyword.get(session_execution_surface_attrs, :observability, %{})
    run_observability = Keyword.get(run_execution_surface_attrs, :observability, %{})

    normalize_execution_surface_attrs(
      session_execution_surface_attrs
      |> Keyword.merge(run_execution_surface_attrs)
      |> Keyword.put(
        :transport_options,
        Keyword.merge(session_transport_options, run_transport_options)
      )
      |> Keyword.put(:observability, Map.merge(session_observability, run_observability))
    )
  end

  defp normalize_execution_surface_struct(%ExecutionSurface{} = execution_surface) do
    execution_surface
    |> execution_surface_attrs()
    |> normalize_execution_surface_attrs()
  end

  defp normalize_execution_surface_attrs(attrs) when is_list(attrs) do
    case ExecutionSurface.new(attrs) do
      {:ok, %ExecutionSurface{} = execution_surface} ->
        {:ok, execution_surface}

      {:error, reason} ->
        {:error, config_error("execution_surface is invalid: #{inspect(reason)}")}
    end
  end

  defp normalize_execution_surface_input(%ExecutionSurface{} = execution_surface) do
    with {:ok, normalized} <- normalize_execution_surface_struct(execution_surface) do
      {:ok, execution_surface_attrs(normalized)}
    end
  end

  defp normalize_execution_surface_input(execution_surface) when is_list(execution_surface) do
    normalized_keys = Keyword.keys(execution_surface)

    with {:ok, normalized} <- normalize_execution_surface_attrs(execution_surface) do
      {:ok, execution_surface_attrs(normalized, normalized_keys)}
    end
  end

  defp normalize_execution_surface_input(execution_surface) when is_map(execution_surface) do
    normalized_keys = execution_surface_present_keys(execution_surface)

    with {:ok, normalized} <-
           execution_surface
           |> execution_surface_attrs()
           |> normalize_execution_surface_attrs() do
      {:ok, execution_surface_attrs(normalized, normalized_keys)}
    end
  end

  defp execution_surface_attrs(%ExecutionSurface{} = execution_surface) do
    execution_surface_attrs(execution_surface, @execution_surface_keys)
  end

  defp execution_surface_attrs(attrs) when is_map(attrs) do
    [
      contract_version: Map.get(attrs, :contract_version, Map.get(attrs, "contract_version")),
      surface_kind: Map.get(attrs, :surface_kind, Map.get(attrs, "surface_kind")),
      transport_options: Map.get(attrs, :transport_options, Map.get(attrs, "transport_options")),
      target_id: Map.get(attrs, :target_id, Map.get(attrs, "target_id")),
      lease_ref: Map.get(attrs, :lease_ref, Map.get(attrs, "lease_ref")),
      surface_ref: Map.get(attrs, :surface_ref, Map.get(attrs, "surface_ref")),
      boundary_class: Map.get(attrs, :boundary_class, Map.get(attrs, "boundary_class")),
      observability: Map.get(attrs, :observability, Map.get(attrs, "observability", %{}))
    ]
  end

  defp execution_surface_attrs(%ExecutionSurface{} = execution_surface, keys) do
    [
      contract_version: execution_surface.contract_version,
      surface_kind: execution_surface.surface_kind,
      transport_options: execution_surface.transport_options,
      target_id: execution_surface.target_id,
      lease_ref: execution_surface.lease_ref,
      surface_ref: execution_surface.surface_ref,
      boundary_class: execution_surface.boundary_class,
      observability: execution_surface.observability
    ]
    |> Keyword.take(keys)
  end

  defp execution_surface_present_keys(attrs) when is_map(attrs) do
    Enum.filter(@execution_surface_keys, fn key ->
      Map.has_key?(attrs, key) or Map.has_key?(attrs, Atom.to_string(key))
    end)
  end

  defp resolve_execution_environment_attrs(opts) when is_list(opts) do
    with {:ok, nested_attrs} <-
           normalize_execution_environment_input(Keyword.get(opts, :execution_environment)),
         {:ok, top_level_attrs} <- normalize_top_level_execution_environment_attrs(opts) do
      {:ok, merge_present_attrs(nested_attrs, top_level_attrs, Environment.keys())}
    end
  end

  defp normalize_execution_environment_input(nil), do: {:ok, []}

  defp normalize_execution_environment_input(input),
    do: normalize_execution_environment_attrs(input)

  defp normalize_top_level_execution_environment_attrs(opts) when is_list(opts) do
    attrs =
      Enum.reduce(Environment.keys(), [], fn key, acc ->
        if Keyword.has_key?(opts, key) do
          [{key, Keyword.get(opts, key)} | acc]
        else
          acc
        end
      end)
      |> Enum.reverse()

    normalize_execution_environment_attrs(attrs)
  end

  defp normalize_execution_environment_attrs(attrs) do
    case Environment.normalize_attrs(attrs) do
      {:ok, normalized_attrs} ->
        {:ok, normalized_attrs}

      {:error, reason} ->
        {:error, config_error("execution_environment is invalid: #{inspect(reason)}")}
    end
  end

  defp merge_execution_environments(session_attrs, run_attrs)
       when is_list(session_attrs) and is_list(run_attrs) do
    merged_attrs = merge_present_attrs(session_attrs, run_attrs, Environment.keys())

    case Environment.new(merged_attrs) do
      {:ok, %Environment{} = environment} ->
        {:ok, environment}

      {:error, reason} ->
        {:error, config_error("execution_environment is invalid: #{inspect(reason)}")}
    end
  end

  defp merge_present_attrs(left_attrs, right_attrs, keys)
       when is_list(left_attrs) and is_list(right_attrs) and is_list(keys) do
    Enum.reduce(keys, [], fn key, acc ->
      cond do
        Keyword.has_key?(right_attrs, key) ->
          [{key, Keyword.get(right_attrs, key)} | acc]

        Keyword.has_key?(left_attrs, key) ->
          [{key, Keyword.get(left_attrs, key)} | acc]

        true ->
          acc
      end
    end)
    |> Enum.reverse()
  end

  defp resolve_provider_permission_mode(nil, _provider), do: {:ok, nil}

  defp resolve_provider_permission_mode(permission_mode, provider) when is_atom(provider) do
    case Permission.normalize(provider, permission_mode) do
      {:ok, %{native: native}} ->
        {:ok, native}

      {:error, %Error{} = error} ->
        {:error, error}
    end
  end

  defp resolve_provider_permission_mode(_permission_mode, _provider), do: {:ok, nil}

  defp normalize_keyword(value) when is_list(value), do: value
  defp normalize_keyword(_value), do: []

  defp app_config do
    defaults = [
      execution_mode: :local,
      remote_connect_timeout_ms: 5_000,
      remote_rpc_timeout_ms: 15_000,
      remote_boot_lease_timeout_ms: 10_000,
      remote_bootstrap_mode: :require_prestarted,
      transport_call_timeout_ms: 5_000
    ]

    Keyword.merge(defaults, Application.get_env(:agent_session_manager, __MODULE__, []))
  end

  defp config_error(message) do
    Error.new(:config_invalid, :config, message)
  end
end