Opens a long-lived SSE stream and owns its receive loop.

Dstar.Page streams run on this module; plain controllers use it the same way. It owns the parts of a stream's lifecycle that are easy to get subtly wrong in a hand-rolled receive:

• Keyed claims. With key:, open/2 claims {key, tabId} through the opt-in Dstar.Utility.StreamRegistry before SSE starts, and reports a refused claim as {:error, conn}.
• Takeover. When a newer stream for the same tab claims the key, the registry signals this stream with its exact claim generation. run/2 recognises only its own generation, releases the claim, then tears down. Stale signals from an earlier stream on a reused keep-alive process are ignored.
• Adapter plumbing. {:plug_conn, :sent} is skipped, and Bandit's {:bandit, _} flow-control messages are left in the mailbox for Bandit's own selective receive. A catch-all receive would swallow them and can stall an HTTP/2 stream.
• Liveness and release. An idle stream is probed every :idle_check ms and torn down once the client is gone. The claim is always released before :disconnect runs, and again if any callback raises.

Example

def stream(conn, _params) do
  user = conn.assigns.current_user

  case Dstar.Stream.open(conn, key: user.id) do
    {:ok, conn} ->
      Dstar.Stream.run(conn,
        connect: fn conn ->
          Phoenix.PubSub.subscribe(MyApp.PubSub, "user:#{user.id}")
          conn
        end,
        info: fn
          {:notification, n}, conn -> Dstar.Elements.append(conn, row(n), "#inbox")
          :logout, conn -> {:halt, conn}
        end,
        disconnect: fn _conn ->
          Phoenix.PubSub.unsubscribe(MyApp.PubSub, "user:#{user.id}")
        end
      )

    {:error, conn} ->
      # 400/413 (bad signals) or 503 (claim refused): plain HTTP, no SSE.
      conn
  end
end

Dstar.start_stream/2,3 and Dstar.Utility.StreamRegistry.release/1 remain available for loops that must own their receive.