anybody else's — ADMIN-001 names the same check at the route. It resolves

  and links; it never writes to the thread, mints for it, or returns it.

  Amended 2026-08-25 (issue #242): the stored linkage gained a reader.

  `OpenAgents.Gym.fetch_trial_thread/1` resolves a thread only through a

  stored trial row's `thread_id` — a link `record_trial/3` admitted through

  `get_for_user/2` at ingest — and serves the operator-gated `/gym/runs/:id`

  transcript view (`OpenAgentsWeb.GymRunLive`, recheck on mount and on every

  event). The trial row is the only key, so an arbitrary thread id has no

  path in; the function reads, and never writes to the thread, mints for it,

  or returns a grant. This is an operator read of another account's

  transcript — exactly the transcripts a bearer deliberately linked to a

  benchmark trial, and nothing else — and ADMIN-001 names it beside the

  ingest check. `test/openagents/gym_test.exs` proves the refusals (unknown

  trial, unlinked trial, deleted thread) and

  `test/openagents_web/live/gym_run_live_test.exs` proves the gate.

Evidence: `OpenAgents.Threads`, `OpenAgents.Threads.Thread`,

`OpenAgents.Threads.Event`, `OpenAgents.Inference.mint/1`,

`OpenAgents.Inference.expire_elapsed_for_owner/1`,

  and `PATCH /api/v1/gym/runs/:id`

  (`OpenAgentsWeb.GymRunController`, which rechecks the operator on every

  request over the bearer scope), and reading the scoreboard from `/gym`

  (`OpenAgentsWeb.GymLive`, recheck on mount and on every event). A run is a

  benchmark record — recipe digest, task, model, lane, reward, duration —

  never account data; the surface is operator-only because it is

  pre-release instrumentation, not because it reads across accounts. The one

  cross-record link a trial may carry, a `thread_id`, is verified at ingest:

  `OpenAgents.Gym.record_trial/3` admits a thread only when

  `OpenAgents.Threads.get_for_user/2` resolves it for the bearer's account,

  and an unknown thread and an unowned one refuse identically, so the Gym

  cannot be used to confirm that a foreign thread id exists.

  and a run's page from `/gym/runs/:id` (`OpenAgentsWeb.GymLive` and

  `OpenAgentsWeb.GymRunLive`, recheck on mount and on every event). A run

  is a benchmark record — recipe digest, task, model, lane, reward,

  duration — never account data. The one cross-record link a trial may

  carry, a `thread_id`, is verified at ingest: `OpenAgents.Gym.record_trial/3`

  admits a thread only when `OpenAgents.Threads.get_for_user/2` resolves it

  for the bearer's account, and an unknown thread and an unowned one refuse

  identically, so the Gym cannot be used to confirm that a foreign thread

  id exists. Since issue #242 that link is also read back: the run page

  streams a linked trial's thread transcript to the operator through

  `OpenAgents.Gym.fetch_trial_thread/1`, which resolves only through a

  stored, ingest-verified trial linkage — so this surface does read another

  account's data, exactly the transcripts a bearer deliberately linked to a

  benchmark trial, and nothing else (THREAD-001 names the same reader from

  the thread side).

Reading a private forum board and raising a repository's transparency tier to

`glass` are operator reads that widen with the same allowlist

  A trial's thread link is verified at ingest: the thread must exist and

  belong to the bearer's account, and an unknown thread and an unowned one

  refuse identically, so the check confirms nothing about threads the

  account cannot see. A run still `running` whose last update is older

  than six hours is swept to `abandoned` lazily on the read paths, so the

  scoreboard never shows a forever-running row.

  account cannot see. The operator-gated gym surface reads a linked

  transcript back through `fetch_trial_thread/1`, which resolves only

  through that stored, verified linkage. A run still `running` whose last

  update is older than six hours is swept to `abandoned` lazily on the read

  paths, so the scoreboard never shows a forever-running row.

  ## PubSub

    trials_query() |> where([t], t.run_id == ^run_id) |> Repo.all()

  end

  @doc """

  The thread a trial's transcript lives on, read through the trial linkage.

  This is the read path for the operator-gated gym surface. The `/gym`

  viewer is any operator, not necessarily the thread's owner, so it cannot

  read through the account-scoped `OpenAgents.Threads.fetch_readable/2`.

  Reading here is sound because the linkage is the authority:

  `record_trial/3` admitted the `thread_id` only after

  `OpenAgents.Threads.get_for_user/2` resolved it for the reporting

  bearer's account (INVARIANTS THREAD-001, ADMIN-001), so every stored

  linkage names a thread its reporter owned and deliberately attached to a

  benchmark trial. The trial row is the only key — an arbitrary thread id

  has no path in — and the function reads; it never writes to the thread,

  mints for it, or returns a grant.

  Returns `:error` for an unknown trial, a trial with no linkage, and a

  linked thread that no longer exists — a thread may be deleted with its

  account while the benchmark record stays.

  """

  @spec fetch_trial_thread(String.t()) :: {:ok, Threads.Thread.t()} | :error

  def fetch_trial_thread(trial_id) when is_binary(trial_id) do

    with {:ok, id} <- Ecto.UUID.cast(trial_id),

         %Trial{thread_id: thread_id} when is_binary(thread_id) <- Repo.get(Trial, id),

         %Threads.Thread{} = thread <- Repo.get(Threads.Thread, thread_id) do

      {:ok, thread}

    else

      _no_verified_linkage -> :error

    end

  end

  @doc """

  Runs, newest first, optionally filtered by suite. Bounded.

    Phoenix.PubSub.subscribe(OpenAgents.PubSub, topic(thread_id))

  end

  @doc """

  Drop a `subscribe/1` subscription.

  For a viewer that moves between transcripts on one socket, so events from

  the thread it left stop arriving instead of being filtered forever.

  """

  @spec unsubscribe(Thread.t() | String.t()) :: :ok

  def unsubscribe(%Thread{id: thread_id}), do: unsubscribe(thread_id)

  def unsubscribe(thread_id) when is_binary(thread_id) do

    Phoenix.PubSub.unsubscribe(OpenAgents.PubSub, topic(thread_id))

  end

  defp topic(thread_id), do: "thread:" <> thread_id

  defp broadcast(%Event{} = event) do

defmodule OpenAgentsWeb.AI.ThreadTranscript do

  @moduledoc """

  One thread transcript event, rendered through the shared AI Elements

  components — the same vocabulary `/chat` renders with.

  This module is the mapping from the thread event vocabulary onto the chat

  components: `turn.user` becomes a user `OpenAgentsWeb.AI.Conversation`

  message, `turn.reasoning` becomes the `OpenAgentsWeb.AI.Reasoning`

  reasoning disclosure, `tool.ran` becomes the tool block `ChatLive` renders

  tool activity with, and `turn.assistant` becomes an assistant message with

  the app's Markdown rendering. It is a shared function component rather

  than a private helper of the gym page so `OpenAgentsWeb.ThreadShowLive`

  can adopt the same rendering later instead of keeping a second vocabulary.

  Every payload field is client-written, so every read is defensive, exactly

  as `OpenAgentsWeb.ThreadShowLive` reads the same payloads: a missing or

  oddly typed field degrades the event to the neutral bounded raw row, an

  unknown event type takes the raw row too, and nothing here can crash the

  transcript around a malformed payload.

  """

  use OpenAgentsWeb, :html

  import OpenAgentsWeb.AI.Conversation, only: [message: 1, message_content: 1]

  import OpenAgentsWeb.AI.Reasoning,

    only: [

      reasoning: 1,

      reasoning_trigger: 1,

      reasoning_content: 1,

      tool: 1,

      tool_header: 1,

      tool_content: 1,

      tool_input: 1,

      tool_output: 1

    ]

  @bounded_json_characters 2_000

  @doc """

  One transcript entry, dispatched on its event type.

  `event` is an `OpenAgents.Threads.Event`. `id`, when given, prefixes the

  DOM ids of the inner components.

  """

  attr :id, :string, default: nil

  attr :event, :any, required: true

  def transcript_event(%{event: %{event_type: "turn.user"} = event} = assigns) do

    case text(event.payload) do

      nil ->

        raw_row(assigns)

      text ->

        assigns = assign(assigns, text: text, steered: steered?(event.payload))

        ~H"""

        <.message from="user" class="items-end">

          <.badge :if={@steered} variant={:warning} class="message-provenance">steered</.badge>

          <.message_content>

            <div class="whitespace-pre-wrap">{@text}</div>

          </.message_content>

        </.message>

        """

    end

  end

  def transcript_event(%{event: %{event_type: "turn.reasoning"} = event} = assigns) do

    case text(event.payload) do

      nil ->

        raw_row(assigns)

      text ->

        assigns = assign(assigns, :text, text)

        ~H"""

        <.reasoning id={@id && "#{@id}-reasoning"}>

          <.reasoning_trigger />

          <.reasoning_content text={@text} />

        </.reasoning>

        """

    end

  end

  def transcript_event(%{event: %{event_type: "tool.ran"} = event} = assigns) do

    case string(event.payload, ["tool", "name"]) do

      nil ->

        raw_row(assigns)

      tool_name ->

        assigns =

          assign(assigns,

            tool_name: tool_name,

            state: tool_state(string(event.payload, ["status"])),

            arguments: bounded_json(Map.get(event.payload, "arguments")),

            result: bounded_json(Map.get(event.payload, "result")),

            error: bounded_json(Map.get(event.payload, "error"))

          )

        ~H"""

        <.tool id={@id && "#{@id}-tool"} class="mb-0">

          <.tool_header type="dynamic-tool" tool_name={@tool_name} state={@state} />

          <.tool_content>

            <.tool_input :if={@arguments} input={@arguments} />

            <.tool_output :if={@result || @error} output={@result} error_text={@error} />

            <p :if={!@arguments && !@result && !@error} class="text-muted-foreground text-xs">

              The report carried no arguments and no result.

            </p>

          </.tool_content>

        </.tool>

        """

    end

  end

  def transcript_event(%{event: %{event_type: "turn.assistant"} = event} = assigns) do

    case text(event.payload) do

      nil ->

        raw_row(assigns)

      text ->

        assigns = assign(assigns, :text, text)

        ~H"""

        <.message from="assistant">

          <.message_content text={@text} />

        </.message>

        """

    end

  end

  def transcript_event(assigns), do: raw_row(assigns)

  # The neutral row: whatever the type, whatever the payload, it renders as

  # the recorded fact rather than crashing the transcript around it.

  defp raw_row(assigns) do

    assigns = assign(assigns, :json, bounded_json(assigns.event.payload))

    ~H"""

    <details class="rounded-md border border-border" data-transcript-raw="true">

      <summary class="flex cursor-pointer select-none items-center gap-2 px-3 py-2 text-xs">

        <code class="font-mono">{@event.event_type}</code>

        <span class="text-muted-foreground"><.time_ago at={@event.emitted_at} /></span>

      </summary>

      <pre

        :if={@json}

        class="overflow-x-auto border-t border-border px-3 py-2 font-mono text-xs"

      ><code phx-no-format>{@json}</code></pre>

    </details>

    """

  end

  # ── payload readers ──────────────────────────────────────────────────────

  defp text(payload), do: string(payload, ["text"])

  defp string(payload, keys) when is_map(payload) do

    Enum.find_value(keys, fn key ->

      case Map.get(payload, key) do

        value when is_binary(value) and value != "" -> value

        _other -> nil

      end

    end)

  end

  defp string(_payload, _keys), do: nil

  defp steered?(payload) when is_map(payload), do: Map.get(payload, "steered") == true

  defp steered?(_payload), do: false

  defp bounded_json(nil), do: nil

  defp bounded_json(value) do

    case Jason.encode(value, pretty: true) do

      {:ok, json} when byte_size(json) > @bounded_json_characters ->

        String.slice(json, 0, @bounded_json_characters) <> "\n…"

      {:ok, json} ->

        json

      {:error, _unencodable} ->

        inspect(value, limit: 50, printable_limit: @bounded_json_characters)

    end

  end

  # The client-written status words mapped onto the AI SDK tool-part states

  # `tool_header/1` reads, the same narrowing `ChatLive` applies to durable

  # step statuses. `tool.ran` is past tense, so an absent or unrecognized

  # status reads as completed rather than inventing an error.

  defp tool_state("ok"), do: "output-available"

  defp tool_state("success"), do: "output-available"

  defp tool_state("succeeded"), do: "output-available"

  defp tool_state("error"), do: "output-error"

  defp tool_state("failed"), do: "output-error"

  defp tool_state("refused"), do: "output-denied"

  defp tool_state("denied"), do: "output-denied"

  defp tool_state("running"), do: "input-available"

  defp tool_state(_other), do: "output-available"

end

defmodule OpenAgentsWeb.GymLive do

  @moduledoc """

  The Gym: graded benchmark runs of our agents, operator-only.

  The Gym: graded benchmark runs of our agents, operator-only, live.

  Read-only over `OpenAgents.Gym` — the harness runs elsewhere and posts

  results through `POST /api/v1/gym/runs`; this surface is the scoreboard

  that capability work (models, plugins, harness changes) is read against.

  results through `POST /api/v1/gym/runs` and the lifecycle routes; this

  surface is the scoreboard that capability work (models, plugins, harness

  changes) is read against. It subscribes to the gym topic, so a running

  suite appears the moment the harness registers it, its trial tally moves

  as trials report, and the run flips to the graded table in place when it

  finalizes — no reload anywhere.

  Operator-gated the same way `/chat` is: the route sits behind the

  `:operator` pipeline, the mount re-checks, and every event re-checks,

  because a long-lived socket outlives the decision that opened it.

  The graded table stays a stream. The running section is a bounded assign

  instead, because each entry carries a live trial tally that mutates on

  every `{:gym_trial, _}` broadcast, and the collection is bounded by the

  number of suites running at once rather than by history.

  """

  use OpenAgentsWeb, :live_view

  import OpenAgentsWeb.AI.Conversation, only: [shimmer: 1]

  alias OpenAgents.Accounts

  alias OpenAgents.Gym

  alias OpenAgents.Gym.Run

  @impl true

  def mount(_params, _session, socket) do

    if Accounts.admin?(socket.assigns.current_user) do

      # Subscribe before the snapshot read, so a run or trial that lands

      # between the two arrives as a message rather than being missed.

      if connected?(socket), do: Gym.subscribe()

      {:ok, load(socket, nil)}

    else

      {:ok, redirect(socket, to: ~p"/")}

    end

  end

  @impl true

  def handle_info({:gym_run, %Run{}}, socket) do

    # Start, finalize, abandon, and sweep all reload the page's two

    # sections under the current filter, which is what moves a finalized

    # run from the running section into the graded table in place.

    if Accounts.admin?(socket.assigns.current_user) do

      {:noreply, load(socket, socket.assigns.suite)}

    else

      {:noreply, redirect(socket, to: ~p"/")}

    end

  end

  def handle_info({:gym_trial, trial}, socket) do

    if Accounts.admin?(socket.assigns.current_user) do

      running =

        Enum.map(socket.assigns.running, fn entry ->

          if entry.run.id == trial.run_id, do: running_entry(entry.run), else: entry

        end)

      {:noreply, assign(socket, :running, running)}

    else

      {:noreply, redirect(socket, to: ~p"/")}

    end

  end

  defp presence(""), do: nil

  defp presence(suite) when is_binary(suite), do: suite

  defp load(socket, suite) do

    runs = Gym.list_runs(suite: suite)

    {running, finished} = Enum.split_with(runs, &(&1.status == "running"))

    socket

    |> assign(:page_title, "Gym")

    |> assign(:suite, suite)

    |> assign(:suites, Gym.suites())

    |> assign(:running, Enum.map(running, &running_entry/1))

    |> assign(:runs_empty?, runs == [])

    |> stream(:runs, runs, reset: true)

    |> assign(:table_empty?, finished == [])

    |> stream(:runs, finished, reset: true)

  end

  # A running run with its live trial tally: how many trials the harness

  # has reported and how many have passed so far.

  defp running_entry(run) do

    trials = Gym.list_trials(run)

    %{

      run: run,

      reported: length(trials),

      passed: Enum.count(trials, &(&1.state == "passed"))

    }

  end

  defp percent(nil), do: "—"

    if(rest == 0, do: "#{minutes}m", else: "#{minutes}m #{rest}s")

  end

  defp elapsed_since(started_at),

    do: elapsed(max(DateTime.diff(DateTime.utc_now(), started_at, :second), 0))

  # A run still `running` (or swept to `abandoned`) has no grades yet.

  defp counts(passed, total) when is_integer(passed) and is_integer(total),

    do: "#{passed}/#{total}"

          </select>

        </form>

        <section :if={@running != []} id="gym-running" class="space-y-3" aria-label="Running now">

          <h2 class="text-sm font-medium text-muted-foreground">Running now</h2>

          <.link

            :for={entry <- @running}

            navigate={~p"/gym/runs/#{entry.run.id}"}

            id={"gym-running-#{entry.run.id}"}

            class="block"

          >

            <.card class="text-sm transition-colors hover:border-primary">

              <div class="flex flex-wrap items-center gap-x-6 gap-y-2">

                <span class="font-mono">{entry.run.suite}</span>

                <span>

                  {entry.run.agent}

                  <span :if={entry.run.agent_version} class="text-muted-foreground">

                    @{entry.run.agent_version}

                  </span>

                </span>

                <span class="font-mono">{entry.run.model}</span>

                <span>{entry.run.lane || "—"}</span>

                <span class="tabular-nums" data-tally>

                  {entry.passed} passed / {entry.reported} reported

                </span>

                <span class="whitespace-nowrap text-muted-foreground">

                  {elapsed_since(entry.run.inserted_at)}

                </span>

                <.shimmer text="Running" tag="span" class="text-xs" />

              </div>

            </.card>

          </.link>

        </section>

        <div :if={@runs_empty?}>

          <.empty title="No runs recorded yet">

            No graded runs have been posted. The harness records one with

          </.empty>

        </div>

        <div :if={!@runs_empty?} class="overflow-x-auto">

        <div :if={!@table_empty?} class="overflow-x-auto">

          <table class="table">

            <thead>

              <tr>

            <tbody id="gym-runs" phx-update="stream">

              <tr :for={{id, run} <- @streams.runs} id={id}>

                <td class="whitespace-nowrap">

                  {Calendar.strftime(run.inserted_at, "%Y-%m-%d %H:%M")}

                  <.link navigate={~p"/gym/runs/#{run.id}"} class="hover:underline">

                    {Calendar.strftime(run.inserted_at, "%Y-%m-%d %H:%M")}

                  </.link>

                </td>

                <td class="font-mono text-sm">{run.suite}</td>

                <td>

defmodule OpenAgentsWeb.GymRunLive do

  @moduledoc """

  One Gym run, live: the run header, its trials as they report, and the

  selected trial's transcript streaming through the same conversation

  components `/chat` renders with.

  Operator-only the way `/gym` is: the route sits in the `:operator_chat`

  live session, the mount re-checks `OpenAgents.Accounts.admin?/1`, and

  every event re-checks it. An unknown run id redirects to `/gym` rather

  than confirming anything, matching how the operator surfaces route a

  reader back to the main flow.

  The transcript read path is `OpenAgents.Gym.fetch_trial_thread/1`: the

  viewer is any operator, not the thread's owner, so the read resolves only

  through a stored trial linkage that was ownership-verified at ingest

  (INVARIANTS THREAD-001, ADMIN-001). The snapshot-to-live order follows

  the projection protocol `OpenAgentsWeb.ThreadShowLive` documents:

  subscribe to the thread's topic first, then read the snapshot, then let

  buffered broadcasts drain with a monotonic `last_event_id` dedup, so an

  event is never dropped and never doubled. Trials on a lane that leaves no

  thread render a state-only placeholder instead.

  """

  use OpenAgentsWeb, :live_view

  import OpenAgentsWeb.AI.Conversation,

    only: [conversation: 1, conversation_content: 1, shimmer: 1]

  import OpenAgentsWeb.AI.ThreadTranscript, only: [transcript_event: 1]

  alias OpenAgents.Accounts

  alias OpenAgents.Gym

  alias OpenAgents.Gym.Run

  alias OpenAgents.Threads

  # Transcript pages are capped at 50 by the context; forty pages bounds the

  # snapshot at 2,000 events, the same bound `ThreadShowLive` holds.

  @maximum_pages 40

  @impl true

  def mount(%{"id" => run_id}, _session, socket) do

    if Accounts.admin?(socket.assigns.current_user) do

      # Subscribe before the snapshot read, so a trial report that lands

      # between the two arrives as a message rather than being missed.

      if connected?(socket), do: Gym.subscribe_run(run_id)

      case Gym.fetch_run(run_id) do

        {:ok, run} ->

          socket =

            socket

            |> assign(:page_title, "Gym run")

            |> assign(:run, run)

            |> assign(:trials, run.trials)

            |> assign(:selected_trial_id, nil)

            |> assign(:thread, nil)

            |> assign(:last_event_id, 0)

            |> assign(:transcript, :none)

            |> assign(:events_empty?, true)

            |> stream(:events, [])

            |> select_trial(default_trial(run.trials))

            |> restream_trials()

          {:ok, socket}

        :error ->

          {:ok, redirect(socket, to: ~p"/gym")}

      end

    else

      {:ok, redirect(socket, to: ~p"/")}

    end

  end

  @impl true

  def handle_event("select_trial", %{"id" => trial_id}, socket) do

    if Accounts.admin?(socket.assigns.current_user) do

      socket =

        case Enum.find(socket.assigns.trials, &(&1.id == trial_id)) do

          nil -> socket

          trial -> socket |> select_trial(trial) |> restream_trials()

        end

      {:noreply, socket}

    else

      {:noreply, redirect(socket, to: ~p"/")}

    end

  end

  @impl true

  def handle_info({:gym_run, %Run{} = run}, socket) do

    cond do

      !Accounts.admin?(socket.assigns.current_user) ->

        {:noreply, redirect(socket, to: ~p"/")}

      run.id == socket.assigns.run.id ->

        # The broadcast carries the run as stored, trials not loaded; the

        # trial list lives in its own assign, so only the header moves.

        {:noreply, assign(socket, :run, run)}

      true ->

        {:noreply, socket}

    end

  end

  def handle_info({:gym_trial, trial}, socket) do

    if Accounts.admin?(socket.assigns.current_user) do

      trials = upsert_trial(socket.assigns.trials, trial)

      socket = socket |> assign(:trials, trials) |> restream_trials()

      socket =

        cond do

          # The first reported trial becomes the selection, so an operator

          # watching an empty run is attached the moment work starts.

          socket.assigns.selected_trial_id == nil ->

            select_trial(socket, default_trial(trials))

          # The selected trial gained its thread link after selection.

          trial.id == socket.assigns.selected_trial_id and

            socket.assigns.transcript != :live and is_binary(trial.thread_id) ->

            select_trial(socket, trial)

          true ->

            socket

        end

      {:noreply, socket}

    else

      {:noreply, redirect(socket, to: ~p"/")}

    end

  end

  def handle_info({:thread_event, event}, socket) do

    cond do

      !Accounts.admin?(socket.assigns.current_user) ->

        {:noreply, redirect(socket, to: ~p"/")}

      socket.assigns.thread == nil or event.thread_id != socket.assigns.thread.id ->

        {:noreply, socket}

      event.id <= socket.assigns.last_event_id ->

        {:noreply, socket}

      true ->

        {:noreply,

         socket

         |> assign(:last_event_id, event.id)

         |> assign(:events_empty?, false)

         |> stream_insert(:events, event)}

    end

  end

  # ── selection ────────────────────────────────────────────────────────────

  defp default_trial(trials),

    do: Enum.find(trials, &(&1.state == "running")) || List.first(trials)

  defp select_trial(socket, nil) do

    socket

    |> detach_thread()

    |> assign(selected_trial_id: nil, transcript: :none, events_empty?: true, last_event_id: 0)

    |> stream(:events, [], reset: true)

  end

  defp select_trial(socket, trial) do

    socket = socket |> detach_thread() |> assign(:selected_trial_id, trial.id)

    if trial.thread_id == nil do

      socket

      |> assign(transcript: :no_thread, events_empty?: true, last_event_id: 0)

      |> stream(:events, [], reset: true)

    else

      attach_thread(socket, trial)

    end

  end

  defp attach_thread(socket, trial) do

    case Gym.fetch_trial_thread(trial.id) do

      {:ok, thread} ->

        # Attach the live subscriber before reading the snapshot: an append

        # that lands between the two arrives as a buffered message and is

        # deduped by id, so the gap cannot lose an event.

        if connected?(socket), do: Threads.subscribe(thread)

        events = transcript_snapshot(thread)

        socket

        |> assign(:thread, thread)

        |> assign(:transcript, :live)

        |> assign(:last_event_id, last_id(events))

        |> assign(:events_empty?, events == [])

        |> stream(:events, events, reset: true)

      :error ->

        socket

        |> assign(transcript: :unavailable, events_empty?: true, last_event_id: 0)

        |> stream(:events, [], reset: true)

    end

  end

  defp detach_thread(socket) do

    case socket.assigns[:thread] do

      nil ->

        socket

      thread ->

        if connected?(socket), do: Threads.unsubscribe(thread)

        assign(socket, :thread, nil)

    end

  end

  defp upsert_trial(trials, trial) do

    trials

    |> Enum.reject(&(&1.id == trial.id or &1.task == trial.task))

    |> then(&[trial | &1])

    |> Enum.sort_by(& &1.task)

  end

  defp restream_trials(socket) do

    socket

    |> assign(:trials_empty?, socket.assigns.trials == [])

    |> stream(:trials, socket.assigns.trials, reset: true)

  end

  # ── snapshot ─────────────────────────────────────────────────────────────

  defp transcript_snapshot(thread), do: transcript_snapshot(thread, nil, @maximum_pages, [])

  defp transcript_snapshot(_thread, _after_id, 0, pages),

    do: pages |> Enum.reverse() |> List.flatten()

  defp transcript_snapshot(thread, after_id, remaining, pages) do

    page = Threads.list_events(thread, after: after_id)

    case last_id(page) do

      0 -> transcript_snapshot(thread, after_id, 0, pages)

      last -> transcript_snapshot(thread, last, remaining - 1, [page | pages])

    end

  end

  defp last_id([]), do: 0

  defp last_id(events), do: List.last(events).id

  # ── presentation ─────────────────────────────────────────────────────────

  defp percent(nil), do: "—"

  defp percent(score), do: "#{Float.round(score * 100, 1)}%"

  defp status_variant("running"), do: :info

  defp status_variant("graded"), do: :success

  defp status_variant("abandoned"), do: :dim

  defp status_variant(_status), do: :default

  defp trial_variant("passed"), do: :success

  defp trial_variant("failed"), do: :danger

  defp trial_variant(_running_or_ungraded), do: :dim

  @impl true

  def render(assigns) do

    ~H"""

    <Layouts.app

      flash={@flash}

      sidebar_sections={assigns[:sidebar_sections]}

      current_scope={@current_scope}

    >

      <main id="gym-run" class="mx-auto w-full max-w-6xl space-y-6 px-4 py-10">

        <header class="space-y-3">

          <div class="flex flex-wrap items-center gap-3">

            <.badge id="gym-run-status" variant={status_variant(@run.status)}>

              {@run.status}

            </.badge>

            <span class="font-mono text-xs text-muted-foreground">{@run.id}</span>

          </div>

          <h1 class="text-2xl font-semibold tracking-tight">

            {@run.suite}

            <span class="text-muted-foreground">·</span>

            {@run.agent}<span

              :if={@run.agent_version}

              class="text-muted-foreground"

            >@{@run.agent_version}</span>

          </h1>

          <dl

            id="gym-run-facts"

            class="flex flex-wrap gap-x-6 gap-y-1 text-sm text-muted-foreground"

          >

            <div class="flex gap-1.5">

              <dt>Model</dt>

              <dd class="font-mono text-xs leading-5">{@run.model}</dd>

            </div>

            <div class="flex gap-1.5">

              <dt>Lane</dt>

              <dd>{@run.lane || "—"}</dd>

            </div>

            <div class="flex gap-1.5">

              <dt>Score</dt>

              <dd id="gym-run-score" class="tabular-nums">{percent(Run.score(@run))}</dd>

            </div>

            <div class="flex gap-1.5">

              <dt>Recipe</dt>

              <dd class="max-w-xs truncate font-mono text-xs leading-5" title={@run.recipe_digest}>

                {@run.recipe_digest}

              </dd>

            </div>

            <div class="flex gap-1.5">

              <dt>Started</dt>

              <dd><.time_ago at={@run.inserted_at} /></dd>

            </div>

            <div :if={@run.completed_at} class="flex gap-1.5">

              <dt>Completed</dt>

              <dd><.time_ago at={@run.completed_at} /></dd>

            </div>

          </dl>

        </header>

        <div class="grid gap-6 lg:grid-cols-[minmax(16rem,1fr)_2fr]">

          <section aria-label="Trials" class="space-y-3">

            <h2 class="text-sm font-medium text-muted-foreground">Trials</h2>

            <.empty :if={@trials_empty?} id="gym-run-trials-empty" title="No trials reported yet">

              Trials appear here as the harness launches them.

            </.empty>

            <div id="gym-run-trials" phx-update="stream" class="space-y-2">

              <button

                :for={{dom_id, trial} <- @streams.trials}

                id={dom_id}

                type="button"

                phx-click="select_trial"

                phx-value-id={trial.id}

                data-state={trial.state}

                data-selected={to_string(@selected_trial_id == trial.id)}

                class={[

                  "flex w-full items-center justify-between gap-3 rounded-md border px-3 py-2 text-left text-sm transition-colors",

                  if(@selected_trial_id == trial.id,

                    do: "border-primary bg-secondary",

                    else: "border-border hover:bg-secondary"

                  )

                ]}

              >

                <span class="min-w-0 truncate font-mono text-xs">{trial.task}</span>

                <%= if trial.state == "running" do %>

                  <.shimmer text="running" tag="span" class="shrink-0 text-xs" />

                <% else %>

                  <.badge variant={trial_variant(trial.state)} class="shrink-0">

                    {trial.state}

                  </.badge>

                <% end %>

              </button>

            </div>

          </section>

          <section aria-label="Transcript" class="space-y-3">

            <h2 class="text-sm font-medium text-muted-foreground">Transcript</h2>

            <%= case @transcript do %>

              <% :none -> %>

                <.empty id="gym-transcript-none" title="No trial selected">

                  Select a trial to read its transcript.

                </.empty>

              <% :no_thread -> %>

                <.empty id="gym-transcript-no-thread" title="No transcript">

                  This trial's lane left no transcript.

                </.empty>

              <% :unavailable -> %>

                <.empty id="gym-transcript-unavailable" title="Transcript unavailable">

                  The linked thread no longer exists.

                </.empty>

              <% :live -> %>

                <div class="flex h-[36rem] flex-col overflow-hidden rounded-md border border-border">

                  <.conversation id="gym-conversation" aria-label="Trial transcript">

                    <.conversation_content id="gym-conversation-content">

                      <.empty :if={@events_empty?} id="gym-transcript-empty" title="No events yet">

                        The transcript fills as the trial works.

                      </.empty>

                      <div id="gym-transcript-events" phx-update="stream" class="contents">

                        <div

                          :for={{dom_id, event} <- @streams.events}

                          id={dom_id}

                          data-kind={event.event_type}

                        >

                          <.transcript_event id={dom_id} event={event} />

                        </div>

                      </div>

                    </.conversation_content>

                  </.conversation>

                </div>

            <% end %>

          </section>

        </div>

      </main>

    </Layouts.app>

    """

  end

end

  defp policy(%{path: "/gym"}),

    do: declaration(:operator, "configured operator GitHub ID", "gym:read", false)

  defp policy(%{path: "/gym/runs/:id"}),

    do: declaration(:operator, "configured operator GitHub ID", "gym:read", false)

  defp policy(%{path: "/admin/analytics"}),

    do: declaration(:operator, "configured operator GitHub ID", "analytics:read", false)

      # than a second gating mechanism. Widening it is a decision, not a

      # default.

      live "/gym", GymLive, :index

      live "/gym/runs/:id", GymRunLive, :show

    end

  end

      assert Gym.fetch_run("not-a-uuid") == :error

    end

  end

  describe "fetch_trial_thread/1" do

    test "reads a thread only through a stored, verified linkage" do

      bearer = github_user("gym-thread-reader")

      {:ok, thread} = Threads.open(bearer, "Run the linked trial")

      run = running_run()

      {:ok, linked} =

        Gym.record_trial(bearer, run, %{

          "task" => "linked",

          "state" => "running",

          "thread_id" => thread.id

        })

      assert {:ok, fetched} = Gym.fetch_trial_thread(linked.id)

      assert fetched.id == thread.id

    end

    test "a trial without a linkage is refused" do

      bearer = github_user("gym-thread-unlinked")

      run = running_run()

      {:ok, unlinked} =

        Gym.record_trial(bearer, run, %{"task" => "local-lane", "state" => "running"})

      assert Gym.fetch_trial_thread(unlinked.id) == :error

    end

    test "an unknown trial id is refused, however it is spelled" do

      assert Gym.fetch_trial_thread(Ecto.UUID.generate()) == :error

      assert Gym.fetch_trial_thread("not-a-uuid") == :error

    end

    test "a linkage whose thread was deleted with its account is refused" do

      bearer = github_user("gym-thread-deleted")

      {:ok, thread} = Threads.open(bearer, "Run then delete")

      run = running_run()

      {:ok, linked} =

        Gym.record_trial(bearer, run, %{

          "task" => "deleted",

          "state" => "running",

          "thread_id" => thread.id

        })

      {:ok, _deleted} = Repo.delete(thread)

      assert Gym.fetch_trial_thread(linked.id) == :error

    end

  end

end

    {:record_event, 3} => :thread_struct,

    {:record_events, 2} => :thread_struct,

    {:spend, 1} => :thread_struct,

    {:subscribe, 1} => :thread_struct

    {:subscribe, 1} => :thread_struct,

    {:unsubscribe, 1} => :thread_struct

  }

  # Every module that reaches a token-returning `OpenAgents.Threads` export.

    end

  end

  describe "live updates" do

    test "a running run appears, its tally moves, and it flips to graded in place", %{

      conn: conn

    } do

      conn = log_in_admin_user(conn, "gym-live-operator")

      {:ok, view, _html} = live(conn, ~p"/gym")

      bearer = github_user("gym-live-bearer")

      {:ok, run, false} =