sagents

defmodule <%= module %> do
  @moduledoc """
  LiveView shell over `<%= subscriber_session_module %>`.

  Each function takes a `socket`, calls the equivalent host-agnostic
  function on `<%= subscriber_session_alias %>`, and applies the resulting
  changes via `assign/2`. LiveView-specific concerns — streams (`stream/3`,
  `stream_insert/3`), flash messages, `connected?` gates, and orchestrating
  `AgentServer.resume/2` calls with their flash side effects — live here.

  Non-LiveView consumers (a GraphQL bridge GenServer, etc.) should bypass
  this module and call `<%= subscriber_session_alias %>` directly.

  ## Subscription model

  The host LiveView keeps a `Sagents.Subscriber` subs map in
  `socket.assigns.sagents_subs`. The host is also expected to:

    1. In `mount/3`, subscribe the LiveView process to the agent presence
       topic via `Phoenix.PubSub.subscribe(<%= app_module %>.PubSub,
       Sagents.Subscriber.presence_topic())` so `presence_diff` broadcasts
       drive auto-resubscribe on agent migration.

    2. In `handle_info/2`, route `{:DOWN, ref, :process, _, _}` and
       `%Phoenix.Socket.Broadcast{event: "presence_diff"}` to
       `handle_publisher_down/2` and `handle_presence_diff/2` here.

    3. For the action path (send_message, wake_agent, etc.), call
       `<%= coordinator_alias %>.ensure_agent_session_running(socket.assigns)`
       and apply the returned changes with `assign(socket, changes)`.

  ## Customization

  This module was generated for your application and has hardcoded references to:
  - `<%= conversations_module %>` - Database context
  - `<%= coordinator_module %>` - Agent coordination
  - `<%= subscriber_session_module %>` - Host-agnostic state model
  - `Sagents.AgentServer` - Agent server module
  - `Sagents.Subscriber` - Subscription bookkeeping
  """

  import Phoenix.LiveView, only: [stream: 4, stream_insert: 3, put_flash: 3, connected?: 1]
  import Phoenix.Component, only: [assign: 2, assign: 3]

  alias <%= conversations_module %>
  alias <%= coordinator_module %>
  alias <%= subscriber_session_module %>
  alias Sagents.AgentServer
  alias Sagents.Subscriber

  require Logger

  # ===========================================================================
  # State init / reset
  # ===========================================================================

  @doc """
  Initialize all agent-related assigns to their default empty state.

  Combines `<%= subscriber_session_alias %>.init_session_state/0` with the
  `:messages` LiveView stream initialization.
  """
  def init_agent_state(socket) do
    socket
    |> assign(<%= subscriber_session_alias %>.init_session_state())
    |> stream(:messages, [], reset: true)
  end

  @doc """
  Reset all agent-related state to default values and unsubscribe from any
  current agent.
  """
  def reset_conversation(socket) do
    socket =
      if connected?(socket) do
        unsubscribe_current_agent(socket)
      else
        socket
      end

    init_agent_state(socket)
  end

  defp unsubscribe_current_agent(socket) do
    agent_id = socket.assigns[:agent_id]
    subs = socket.assigns[:sagents_subs] || %{}

    if agent_id do
      new_subs = Subscriber.unsubscribe_from_agent(subs, agent_id)
      Logger.debug("Unsubscribed from agent #{agent_id}")
      assign(socket, :sagents_subs, new_subs)
    else
      socket
    end
  end

  # ===========================================================================
  # Load conversation (load path — subscribe-only)
  # ===========================================================================

  @doc """
  Load a conversation from the database and subscribe the LiveView to its
  agent's events.

  This is the **load path** — it never starts the agent. If the agent
  isn't running, the subscription is recorded as `:pending`; it
  auto-upgrades to `:subscribed` via `presence_diff` once the agent
  appears (e.g. when the user takes an action that calls
  `<%= coordinator_alias %>.ensure_agent_session_running/1`).

  ## Parameters

  - `socket` - The LiveView socket
  - `conversation_id` - ID of the conversation to load
  - `opts` - Keyword list of options:
    - `:scope` (required) - User scope for database queries
    - `:user_id` (optional) - User ID for presence tracking
    - `:conversations_module` (optional) - DB context module override
      (default: `<%= conversations_alias %>`)

  ## Returns

  - `{:ok, socket}` - Socket with conversation loaded and subscribed
  - `{:error, socket}` - Socket with flash error (when conversation not found)
  """
  def load_conversation(socket, conversation_id, opts) do
    scope = Keyword.fetch!(opts, :scope)
    user_id = Keyword.get(opts, :user_id)
    conversations = Keyword.get(opts, :conversations_module, <%= conversations_alias %>)

    try do
      socket = maybe_unsubscribe_previous(socket, conversation_id)

      conversation = conversations.get_conversation!(scope, conversation_id)
      agent_id = <%= coordinator_alias %>.conversation_agent_id(conversation_id)

      socket = maybe_subscribe_and_track(socket, agent_id, conversation_id, user_id)

      display_messages = conversations.load_display_messages(scope, conversation_id)
      has_messages = !Enum.empty?(display_messages)
      saved_todos = conversations.load_todos(scope, conversation_id)

      agent_status = AgentServer.get_status(agent_id)

      socket =
        socket
        |> assign(:conversation, conversation)
        |> assign(:conversation_id, conversation_id)
        |> assign(:agent_id, agent_id)
        |> assign(:todos, saved_todos)
        |> assign(:agent_status, agent_status)
        |> stream(:messages, display_messages, reset: true)
        |> assign(:has_messages, has_messages)

      socket =
        cond do
          agent_status == :interrupted ->
            info = AgentServer.get_info(agent_id)
            handle_status_interrupted(socket, info.interrupt_data)

          agent_status == :not_running and <%= conversations_alias %>.interrupted?(conversation) ->
            # Persisted state has a pending interrupt the user needs to see.
            # Auto-wake the agent so the boot broadcast surfaces the question
            # UI; clicking "Answer" then resumes against a live process.
            # Other (idle/historical) conversations stay lazy — opening them
            # for a read-only browse does not spin up an agent.
            auto_wake_for_pending_interrupt(socket)

          true ->
            socket
        end

      {:ok, socket}
    rescue
      Ecto.NoResultsError ->
        {:error, put_flash(socket, :error, "Conversation not found")}
    end
  end

  # Boots the agent for a conversation whose persisted metadata indicates a
  # pending interrupt. The LiveView is seeded as initial subscriber inside
  # ensure_agent_session_running, so the boot broadcast (carrying the restored
  # `:interrupted` status from sagents' derive_boot_status) reaches us via
  # handle_info and surfaces the interrupt UI without polling.
  defp auto_wake_for_pending_interrupt(socket) do
    case <%= coordinator_alias %>.ensure_agent_session_running(socket.assigns) do
      {:ok, changes} ->
        assign(socket, changes)

      {:error, reason} ->
        Logger.warning(
          "Auto-wake failed for interrupted conversation #{socket.assigns[:conversation_id]}: #{inspect(reason)}"
        )

        socket
    end
  end

  defp maybe_unsubscribe_previous(socket, conversation_id) do
    if connected?(socket) && socket.assigns[:conversation_id] &&
         socket.assigns.conversation_id != conversation_id do
      unsubscribe_current_agent(socket)
    else
      socket
    end
  end

  defp maybe_subscribe_and_track(socket, agent_id, conversation_id, user_id) do
    if connected?(socket) do
      socket = subscribe_to_agent(socket, agent_id)
      <%= subscriber_session_alias %>.maybe_track_viewer(conversation_id, user_id)
      socket
    else
      socket
    end
  end

  defp subscribe_to_agent(socket, agent_id) do
    subs = socket.assigns[:sagents_subs] || %{}
    new_subs = Subscriber.subscribe_to_agent(subs, agent_id)
    assign(socket, :sagents_subs, new_subs)
  end

  # ===========================================================================
  # Recovery (DOWN, presence_diff)
  # ===========================================================================

  @doc """
  Route a `{:DOWN, ref, :process, _pid, _reason}` from a producer crash
  to `<%= subscriber_session_alias %>.handle_publisher_down/3`. Pending
  subscriptions flip to `:pending` and we wait for `presence_diff` to drive
  resubscribe.
  """
  def handle_publisher_down(socket, ref, reason \\ :noproc) do
    changes = <%= subscriber_session_alias %>.handle_publisher_down(socket.assigns, ref, reason)

    if map_size(changes) > 0 do
      Logger.debug(
        "Producer DOWN — flipping subscription to pending (will re-subscribe on presence diff)"
      )
    end

    assign(socket, changes)
  end

  @doc """
  Route a Phoenix presence diff payload to
  `<%= subscriber_session_alias %>.handle_presence_diff/2`.
  """
  def handle_presence_diff(socket, payload),
    do: assign(socket, <%= subscriber_session_alias %>.handle_presence_diff(socket.assigns, payload))

  # ===========================================================================
  # Status handlers
  # ===========================================================================

  @doc "Status changed to :running."
  def handle_status_running(socket),
    do: assign(socket, <%= subscriber_session_alias %>.handle_status_running())

  @doc "Status changed to :idle."
  def handle_status_idle(socket),
    do: assign(socket, <%= subscriber_session_alias %>.handle_status_idle())

  @doc "Status changed to :cancelled."
  def handle_status_cancelled(socket),
    do: assign(socket, <%= subscriber_session_alias %>.handle_status_cancelled())

  @doc "Status changed to :error — also surfaces a flash."
  def handle_status_error(socket, reason) do
    socket
    |> assign(<%= subscriber_session_alias %>.handle_status_error(reason))
    |> put_flash(:error, <%= subscriber_session_alias %>.format_error_message(reason))
  end

  @doc "Status changed to :interrupted (waiting for human input)."
  def handle_status_interrupted(socket, interrupt_data),
    do: assign(socket, <%= subscriber_session_alias %>.handle_status_interrupted(interrupt_data))

  # ===========================================================================
  # Messaging handlers
  # ===========================================================================

  @doc "Streaming LLM deltas (incremental chunks)."
  def handle_llm_deltas(socket, deltas),
    do: assign(socket, <%= subscriber_session_alias %>.handle_llm_deltas(socket.assigns, deltas))

  @doc "Complete LLM message received."
  def handle_llm_message_complete(socket),
    do: assign(socket, <%= subscriber_session_alias %>.handle_llm_message_complete())

  @doc """
  Display message saved — reload from DB and stream-reset (or fall back to
  `stream_insert/3` if there's no conversation context yet).
  """
  def handle_display_message_saved(socket, display_msg) do
    socket =
      if socket.assigns[:conversation_id] && socket.assigns[:current_scope] do
        socket
        |> assign(:streaming_delta, nil)
        |> reload_messages_from_db()
      else
        stream_insert(socket, :messages, display_msg)
      end

    assign(socket, :has_messages, true)
  end

  @doc """
  Reload display messages from the database and stream-reset.
  """
  def reload_messages_from_db(socket) do
    if socket.assigns[:conversation_id] && socket.assigns[:current_scope] do
      messages =
        <%= conversations_alias %>.load_display_messages(
          socket.assigns.current_scope,
          socket.assigns.conversation_id
        )

      stream(socket, :messages, messages, reset: true)
    else
      socket
    end
  end

  @doc """
  Persist a message to the DB if a conversation exists, otherwise build
  an in-memory fallback. Returns the display message map.
  """
  def create_or_persist_message(socket, message_type, text) do
    if socket.assigns[:conversation_id] && socket.assigns[:current_scope] do
      case <%= conversations_alias %>.append_text_message(
             socket.assigns.current_scope,
             socket.assigns.conversation_id,
             message_type,
             text
           ) do
        {:ok, display_msg} ->
          display_msg

        {:error, reason} ->
          Logger.error("Failed to persist #{message_type} message: #{inspect(reason)}")
          create_fallback_message(message_type, text)
      end
    else
      create_fallback_message(message_type, text)
    end
  end

  defp create_fallback_message(message_type, text) do
    %{
      id: generate_id(),
      message_type: message_type,
      content_type: "text",
      content: %{"text" => text},
      timestamp: DateTime.utc_now()
    }
  end

  defp generate_id, do: :crypto.strong_rand_bytes(16) |> Base.encode16(case: :lower)

  # ===========================================================================
  # Tool execution handlers
  # ===========================================================================

  @doc "Tool call identified — set display_text + execution status."
  def handle_tool_call_identified(socket, tool_info) do
    assign(
      socket,
      <%= subscriber_session_alias %>.handle_tool_call_identified(socket.assigns, tool_info)
    )
  end

  @doc "Tool execution lifecycle update."
  def handle_tool_execution_update(socket, status, tool_info) do
    assign(
      socket,
      <%= subscriber_session_alias %>.handle_tool_execution_update(socket.assigns, status, tool_info)
    )
  end

  @doc "Display message updated — stream the new version."
  def handle_display_message_updated(socket, updated_msg),
    do: stream_insert(socket, :messages, updated_msg)

  # ===========================================================================
  # Lifecycle handlers
  # ===========================================================================

  @doc "Conversation title generated — write to DB and update assigns."
  def handle_conversation_title_generated(socket, new_title, agent_id) do
    assign(
      socket,
      <%= subscriber_session_alias %>.handle_conversation_title_generated(
        socket.assigns,
        new_title,
        agent_id
      )
    )
  end

  @doc "Agent shutdown event — clear agent_id and unsubscribe locally."
  def handle_agent_shutdown(socket, shutdown_data) do
    Logger.info("Agent #{socket.assigns[:agent_id]} shutting down: #{shutdown_data.reason}")
    assign(socket, <%= subscriber_session_alias %>.handle_agent_shutdown(socket.assigns, shutdown_data))
  end

  # ===========================================================================
  # HITL decision handlers
  # ===========================================================================

  @doc """
  Handle a single HITL approve/reject decision. Orchestrates the
  `AgentServer.resume/2` call (when the last pending tool is decided) and
  flash messages on top of the state changes from
  `<%= subscriber_session_alias %>.handle_hitl_decision/3`.
  """
  def handle_hitl_decision(socket, index, decision_type) do
    agent_id = socket.assigns[:agent_id]

    if is_nil(agent_id) do
      Logger.error("Cannot process HITL decision: agent_id is nil (agent may have shut down)")

      put_flash(
        socket,
        :error,
        "Agent is no longer running. Please send a new message to continue."
      )
    else
      decision_label = if decision_type == :approve, do: "approved", else: "rejected"

      <%= subscriber_session_alias %>.persist_hitl_decision(
        socket.assigns,
        socket.assigns.pending_tools,
        index,
        decision_label
      )

      Logger.info("#{String.capitalize(decision_label)} tool at index #{index}")

      case <%= subscriber_session_alias %>.handle_hitl_decision(socket.assigns, index, decision_type) do
        {:resume, accumulated, changes} ->
          case AgentServer.resume(agent_id, accumulated) do
            :ok ->
              socket
              |> assign(changes)
              |> assign(<%= subscriber_session_alias %>.hitl_resume_running_changes())
              |> put_flash(:info, "Agent resuming")

            {:error, reason} ->
              Logger.error("Failed to resume agent: #{inspect(reason)}")
              put_flash(socket, :error, "Failed to resume agent: #{inspect(reason)}")
          end

        {:more, changes} ->
          assign(socket, changes)
      end
    end
  end

  # ===========================================================================
  # Ask-user-question handlers
  # ===========================================================================

  @doc """
  Handle a single answer to a pending AskUserQuestion interrupt. Orchestrates
  the `AgentServer.resume/2` call (when the last pending question is
  answered) on top of the state changes from
  `<%= subscriber_session_alias %>.handle_question_response/2`.
  """
  def handle_question_response(socket, response) do
    agent_id = socket.assigns[:agent_id]

    if is_nil(agent_id) do
      Logger.error("Cannot process question response: agent_id is nil (agent may have shut down)")

      put_flash(
        socket,
        :error,
        "Agent is no longer running. Please send a new message to restart."
      )
    else
      case <%= subscriber_session_alias %>.handle_question_response(socket.assigns, response) do
        {:resume, resume_data, changes} ->
          case AgentServer.resume(agent_id, resume_data) do
            :ok ->
              socket
              |> assign(changes)
              |> assign(<%= subscriber_session_alias %>.question_resume_running_changes())

            {:error, reason} ->
              Logger.error("Failed to resume agent with question response: #{inspect(reason)}")
              put_flash(socket, :error, "Failed to submit response: #{inspect(reason)}")
          end

        {:more, changes} ->
          assign(socket, changes)
      end
    end
  end
end