raxol_agent

if Code.ensure_loaded?(Plug.Router) do
  defmodule Raxol.Agent.SessionStreamServer do
    @moduledoc """
    HTTP/SSE server for remote agent session observation.

    Provides REST endpoints for session listing and SSE endpoints for
    real-time event streaming. Built on Plug for zero Phoenix dependency.

    Requires `plug` as a dependency (available transitively via raxol).

    ## Endpoints

    - `GET /sessions` -- list active sessions (JSON)
    - `GET /sessions/:id` -- session info (JSON)
    - `GET /sessions/:id/events` -- SSE stream of real-time events
    - `GET /sessions/:id/history` -- recent event history (JSON)

    ## Usage

        # Start as part of supervision tree (requires Bandit or Plug.Cowboy)
        children = [
          {Raxol.Agent.SessionStreamer, []},
          {Bandit, plug: Raxol.Agent.SessionStreamServer, port: 4001}
        ]

        # Or start manually
        {:ok, _} = Bandit.start_link(
          plug: {Raxol.Agent.SessionStreamServer, streamer: MyStreamer},
          port: 4001
        )

    ## SSE Format

    Events are sent as Server-Sent Events with JSON data:

        event: text_delta
        data: {"text":"Hello"}

        event: tool_use
        data: {"name":"read_file","arguments":{"path":"mix.exs"},"id":"t1"}

        event: state_change
        data: {"from":"thinking","to":"acting"}
    """

    use Plug.Router

    plug(:match)

    plug(Plug.Parsers,
      parsers: [:json],
      pass: ["application/json"],
      json_decoder: Jason
    )

    plug(:dispatch)

    @doc false
    def init(opts), do: opts

    # -- Routes ------------------------------------------------------------------

    get "/sessions" do
      streamer = get_streamer(conn)
      sessions = Raxol.Agent.SessionStreamer.list_sessions(streamer)

      session_list =
        Enum.map(sessions, fn id ->
          %{id: to_string(id), active: true}
        end)

      send_json(conn, 200, %{sessions: session_list})
    end

    get "/sessions/:id/events" do
      streamer = get_streamer(conn)
      session_id = parse_session_id(conn.params["id"])

      conn =
        conn
        |> put_resp_header("content-type", "text/event-stream")
        |> put_resp_header("cache-control", "no-cache")
        |> put_resp_header("connection", "keep-alive")
        |> put_resp_header("access-control-allow-origin", "*")
        |> send_chunked(200)

      Raxol.Agent.SessionStreamer.subscribe(session_id, streamer)

      sse_loop(conn, session_id)
    end

    get "/sessions/:id/history" do
      streamer = get_streamer(conn)
      session_id = parse_session_id(conn.params["id"])
      events = Raxol.Agent.SessionStreamer.history(session_id, streamer)

      json_events =
        Enum.map(events, fn event ->
          {event_type, data} = serialize_event(event)
          %{event: event_type, data: data}
        end)

      send_json(conn, 200, %{session_id: to_string(session_id), events: json_events})
    end

    get "/sessions/:id" do
      session_id = parse_session_id(conn.params["id"])

      case lookup_session_info(session_id) do
        {:ok, info} ->
          send_json(conn, 200, info)

        {:error, :not_found} ->
          send_json(conn, 404, %{error: "session not found"})
      end
    end

    match _ do
      send_json(conn, 404, %{error: "not found"})
    end

    # -- SSE Loop ----------------------------------------------------------------

    defp sse_loop(conn, session_id) do
      receive do
        {:session_event, ^session_id, event} ->
          {event_type, data} = serialize_event(event)

          case Jason.encode(data) do
            {:ok, json} ->
              case chunk(conn, "event: #{event_type}\ndata: #{json}\n\n") do
                {:ok, conn} ->
                  sse_loop(conn, session_id)

                {:error, _reason} ->
                  conn
              end

            {:error, _} ->
              sse_loop(conn, session_id)
          end
      after
        30_000 ->
          # Send keepalive comment
          case chunk(conn, ": keepalive\n\n") do
            {:ok, conn} -> sse_loop(conn, session_id)
            {:error, _} -> conn
          end
      end
    end

    # -- Helpers ----------------------------------------------------------------

    defp get_streamer(conn) do
      case conn.private do
        %{streamer: s} -> s
        _ -> Raxol.Agent.SessionStreamer
      end
    end

    defp parse_session_id(id) do
      case Integer.parse(id) do
        {n, ""} -> n
        _ -> id
      end
    end

    defp lookup_session_info(session_id) do
      if Process.whereis(Raxol.Agent.Registry) do
        do_lookup_session_info(session_id)
      else
        {:error, :not_found}
      end
    end

    defp do_lookup_session_info(session_id) do
      case Registry.lookup(Raxol.Agent.Registry, {:process, session_id}) do
        [{pid, _}] ->
          try do
            status = Raxol.Agent.Process.get_status(pid)

            {:ok,
             %{id: to_string(session_id), status: to_string(status.status), pid: inspect(pid)}}
          catch
            :exit, _ -> {:error, :not_found}
          end

        [] ->
          case Registry.lookup(Raxol.Agent.Registry, session_id) do
            [{_pid, _}] -> {:ok, %{id: to_string(session_id), status: "active"}}
            [] -> {:error, :not_found}
          end
      end
    end

    @passthrough_events [:tool_use, :tool_result, :state_change, :turn_complete, :done]

    defp serialize_event({:text_delta, text}), do: {"text_delta", %{text: text}}
    defp serialize_event({:error, reason}), do: {"error", %{reason: inspect(reason)}}

    defp serialize_event({type, info}) when type in @passthrough_events do
      {Atom.to_string(type), info}
    end

    defp serialize_event(other), do: {"unknown", %{data: inspect(other)}}

    defp send_json(conn, status, data) do
      case Jason.encode(data) do
        {:ok, json} ->
          conn
          |> put_resp_content_type("application/json")
          |> send_resp(status, json)

        {:error, _} ->
          conn
          |> put_resp_content_type("application/json")
          |> send_resp(500, ~s({"error":"encoding_failed"}))
      end
    end
  end
end