carries `"grant": null`, and the mint still resolves through

  `get_for_user/2`.

  Amended 2026-08-25 (issue #241): the owner-scoped lookup gained a second

  caller. `OpenAgents.Gym.record_trial/3` resolves a claimed trial thread

  through `get_for_user/2` before linking it to a benchmark trial, so the Gym

  admits only threads the bearer's account owns and confirms nothing about

  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.

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

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

`OpenAgents.Inference.expire_elapsed_for_owner/1`,

  `OpenAgents.ProfileMemory.forget_active/2`, which supersedes rather than

  deletes, so a retraction is another entry in the audit trail the same

  surface renders and never a row that quietly stops existing.

- Recording a graded Gym run under `POST /api/v3/gym/runs`

- Recording Gym runs and trials under `POST /api/v1/gym/runs`, the lifecycle

  routes `POST /api/v1/gym/runs/start`, `POST /api/v1/gym/runs/:id/trials`,

  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.

  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.

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

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

  those proofs produce. The harness runs elsewhere (the monorepo's bench

  lane); this is the record and the surface.

  ## Lifecycle

  A run has two ways in. The one-shot `record_run/1` writes a completed

  `graded` row, as it always has. The live lifecycle starts a run as

  `running` (`start_run/1`), upserts per-task trials against it

  (`record_trial/3`) — each optionally linked to the thread that carries

  its transcript — and closes it with `finalize_run/2` (grades, `graded`)

  or `abandon_run/1` (no grades, `abandoned`). Recipe-digest idempotency

  holds across both ways in: a resubmitted digest replays the existing row.

  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.

  ## PubSub

  `subscribe/0` joins the `"gym"` topic; `subscribe_run/1` joins

  `"gym:run:" <> run_id`. Both topics carry the same two messages:

    * `{:gym_run, %OpenAgents.Gym.Run{}}` — on record, start, finalize,

      abandon, and sweep. The struct is the run as stored, trials not

      loaded.

    * `{:gym_trial, %OpenAgents.Gym.Trial{}}` — on every trial upsert.

  Operator-only on every path for now: the Gym is a workbench for the

  people building the agent, not a public leaderboard. Widening it later is

  a deliberate act, not a default.

  import Ecto.Query

  alias OpenAgents.Accounts.User

  alias OpenAgents.Gym.Run

  alias OpenAgents.Gym.Trial

  alias OpenAgents.Repo

  alias OpenAgents.Threads

  @maximum_listed 200

  @maximum_trials_per_run 500

  @staleness_seconds 6 * 60 * 60

  @topic "gym"

  @doc "The most trials one run may hold. The bound the trial upsert enforces."

  @spec maximum_trials_per_run() :: pos_integer()

  def maximum_trials_per_run, do: @maximum_trials_per_run

  @doc "Subscribe to every run and trial change, on the `\"gym\"` topic."

  @spec subscribe() :: :ok | {:error, term()}

  def subscribe, do: Phoenix.PubSub.subscribe(OpenAgents.PubSub, @topic)

  @doc "Subscribe to one run's changes, on `\"gym:run:\" <> run_id`."

  @spec subscribe_run(String.t()) :: :ok | {:error, term()}

  def subscribe_run(run_id) when is_binary(run_id),

    do: Phoenix.PubSub.subscribe(OpenAgents.PubSub, run_topic(run_id))

  @doc """

  Record one run, idempotently by recipe digest.

  Record one completed run, idempotently by recipe digest.

  A resubmitted digest returns the existing row as `{:ok, run, replayed?:

  true}` rather than duplicating or refusing: the harness retries uploads,

  """

  @spec record_run(map()) :: {:ok, Run.t(), boolean()} | {:error, Ecto.Changeset.t()}

  def record_run(attributes) when is_map(attributes) do

    changeset = Run.changeset(%Run{}, attributes)

    changeset =

      %Run{}

      |> Run.changeset(attributes)

      |> Ecto.Changeset.put_change(:completed_at, DateTime.utc_now())

    case Repo.insert(changeset) do

      {:ok, run} ->

        {:ok, run, false}

    insert_or_replay(changeset)

  end

      {:error, %Ecto.Changeset{errors: errors} = failed} ->

        case Keyword.get(errors, :recipe_digest) do

          {_message, options} ->

            if options[:constraint] == :unique,

              do: replay(Ecto.Changeset.get_field(changeset, :recipe_digest), failed),

              else: {:error, failed}

  @doc """

  Register a run at suite start, as `running`.

          nil ->

            {:error, failed}

  Identity now, grades later: `suite`, `agent`, and `model` are required;

  `tasks_total` may carry the planned count. A digest given here replays an

  existing run the same way `record_run/1` does; an absent digest takes a

  generated `pending:` placeholder that `finalize_run/2` replaces.

  """

  @spec start_run(map()) :: {:ok, Run.t(), boolean()} | {:error, Ecto.Changeset.t()}

  def start_run(attributes) when is_map(attributes) do

    insert_or_replay(Run.start_changeset(%Run{}, attributes))

  end

  @doc """

  Fold the grades into a run and close it as `graded`.

  Accepts the same optional fields as `record_run/1`; a digest given here

  replaces the start-time placeholder. Refusals are typed for the door:

  `{:error, :already_graded, run}` for a run that is already terminalized

  with grades — a grade is written once — and `{:error, :digest_conflict,

  existing}` when the new digest already names a different run.

  """

  @spec finalize_run(Run.t(), map()) ::

          {:ok, Run.t()}

          | {:error, :already_graded, Run.t()}

          | {:error, :digest_conflict, Run.t()}

          | {:error, Ecto.Changeset.t()}

  def finalize_run(%Run{status: "graded"} = run, _attributes), do: {:error, :already_graded, run}

  def finalize_run(%Run{} = run, attributes) when is_map(attributes) do

    changeset = Run.finalize_changeset(run, attributes, DateTime.utc_now())

    case Repo.update(changeset) do

      {:ok, updated} ->

        broadcast_run(updated)

        {:ok, updated}

      {:error, %Ecto.Changeset{errors: errors} = failed} ->

        with {_message, options} <- Keyword.get(errors, :recipe_digest),

             true <- options[:constraint] == :unique,

             %Run{} = existing <-

               Repo.get_by(Run,

                 recipe_digest: Ecto.Changeset.get_field(changeset, :recipe_digest)

               ) do

          {:error, :digest_conflict, existing}

        else

          _other -> {:error, failed}

        end

    end

  end

  @doc "Runs, newest first, optionally filtered by suite. Bounded."

  @doc """

  Close a run without grades, as `abandoned`.

  Idempotent for an already-abandoned run; a graded run refuses, because a

  grade on record outranks a late abandonment.

  """

  @spec abandon_run(Run.t()) :: {:ok, Run.t()} | {:error, :already_graded, Run.t()}

  def abandon_run(%Run{status: "graded"} = run), do: {:error, :already_graded, run}

  def abandon_run(%Run{status: "abandoned"} = run), do: {:ok, run}

  def abandon_run(%Run{} = run) do

    {:ok, updated} = run |> Run.abandon_changeset(DateTime.utc_now()) |> Repo.update()

    broadcast_run(updated)

    {:ok, updated}

  end

  @doc """

  Upsert one trial of a run, by `(run_id, task)`.

  `bearer` is the account behind the request: a `thread_id`, when given, is

  admitted only if `OpenAgents.Threads.get_for_user/2` resolves it for that

  account, and an unknown thread and an unowned one refuse with the same

  `thread_id` error. A report that omits `thread_id` keeps an existing

  link rather than clearing it. Trials per run are bounded; a report for a

  task the run does not hold yet refuses once the bound is reached with

  `{:error, :trial_limit}`.

  """

  @spec record_trial(User.t(), Run.t(), map()) ::

          {:ok, Trial.t()} | {:error, :trial_limit} | {:error, Ecto.Changeset.t()}

  def record_trial(%User{} = bearer, %Run{} = run, attributes) when is_map(attributes) do

    changeset =

      %Trial{}

      |> Trial.changeset(attributes)

      |> Ecto.Changeset.put_change(:run_id, run.id)

      |> verify_thread(bearer)

    with {:ok, valid} <- applied(changeset),

         :ok <- within_trial_bound(run, valid.task) do

      {:ok, trial} =

        Repo.insert(changeset,

          on_conflict: {:replace, replaced_columns(attributes)},

          conflict_target: [:run_id, :task],

          returning: true

        )

      touch(run)

      broadcast_trial(trial)

      {:ok, trial}

    end

  end

  @doc "A run by id with its trials loaded, task order. Sweeps staleness first."

  @spec fetch_run(String.t()) :: {:ok, Run.t()} | :error

  def fetch_run(run_id) when is_binary(run_id) do

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

      sweep_stale()

      case Repo.get(Run, id) do

        %Run{} = run -> {:ok, Repo.preload(run, trials: trials_query())}

        nil -> :error

      end

    end

  end

  @doc "A run by id without trials, for the write paths. No sweep."

  @spec get_run(String.t()) :: Run.t() | nil

  def get_run(run_id) when is_binary(run_id) do

    case Ecto.UUID.cast(run_id) do

      {:ok, id} -> Repo.get(Run, id)

      :error -> nil

    end

  end

  @doc "A run's trials, task order."

  @spec list_trials(Run.t()) :: [Trial.t()]

  def list_trials(%Run{id: run_id}) do

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

  end

  @doc """

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

  Sweeps staleness first, so a read never lists a run that stopped

  reporting six hours ago as still running.

  """

  @spec list_runs(keyword()) :: [Run.t()]

  def list_runs(options \\ []) do

    sweep_stale()

    limit = options |> Keyword.get(:limit, 50) |> min(@maximum_listed) |> max(1)

    Run

    |> Repo.all()

  end

  defp filter_suite(query, suite) when is_binary(suite) and suite != "",

    do: where(query, [r], r.suite == ^suite)

  @doc """

  Sweep runs still `running` with no update for six hours to `abandoned`.

  defp filter_suite(query, _absent), do: query

  Lazy rather than scheduled: the read paths call it, which is enough for a

  scoreboard whose staleness only matters when somebody reads it. Each

  swept run is broadcast like any other terminal transition.

  """

  @spec sweep_stale() :: non_neg_integer()

  def sweep_stale do

    now = DateTime.utc_now()

    cutoff = DateTime.add(now, -@staleness_seconds, :second)

    {count, swept} =

      from(r in Run,

        where: r.status == "running" and r.updated_at < ^cutoff,

        select: r

      )

      |> Repo.update_all(set: [status: "abandoned", completed_at: now, updated_at: now])

    Enum.each(swept, &broadcast_run/1)

    count

  end

  defp insert_or_replay(changeset) do

    case Repo.insert(changeset) do

      {:ok, run} ->

        broadcast_run(run)

        {:ok, run, false}

      {:error, %Ecto.Changeset{errors: errors} = failed} ->

        case Keyword.get(errors, :recipe_digest) do

          {_message, options} ->

            if options[:constraint] == :unique,

              do: replay(Ecto.Changeset.get_field(changeset, :recipe_digest), failed),

              else: {:error, failed}

          nil ->

            {:error, failed}

        end

    end

  end

  defp replay(digest, failed) when is_binary(digest) do

    case Repo.get_by(Run, recipe_digest: digest) do

  end

  defp replay(_digest, failed), do: {:error, failed}

  defp filter_suite(query, suite) when is_binary(suite) and suite != "",

    do: where(query, [r], r.suite == ^suite)

  defp filter_suite(query, _absent), do: query

  defp verify_thread(changeset, bearer) do

    case Ecto.Changeset.get_change(changeset, :thread_id) do

      nil ->

        changeset

      thread_id ->

        # One refusal for an unknown thread and an unowned one, so linking

        # cannot be used to probe which thread ids exist.

        if Threads.get_for_user(bearer, thread_id) do

          changeset

        else

          Ecto.Changeset.add_error(

            changeset,

            :thread_id,

            "does not name a thread this account owns"

          )

        end

    end

  end

  defp applied(changeset) do

    case Ecto.Changeset.apply_action(changeset, :insert) do

      {:ok, trial} -> {:ok, trial}

      {:error, _failed} -> {:error, changeset}

    end

  end

  defp within_trial_bound(%Run{id: run_id}, task) do

    known? = Repo.exists?(from(t in Trial, where: t.run_id == ^run_id and t.task == ^task))

    cond do

      known? -> :ok

      count_trials(run_id) < @maximum_trials_per_run -> :ok

      true -> {:error, :trial_limit}

    end

  end

  defp count_trials(run_id) do

    Repo.aggregate(from(t in Trial, where: t.run_id == ^run_id), :count)

  end

  # A report that names no thread keeps an existing link; one that names a

  # thread (already verified) replaces it.

  defp replaced_columns(attributes) do

    if Map.has_key?(attributes, "thread_id") or Map.has_key?(attributes, :thread_id),

      do: [:state, :thread_id, :updated_at],

      else: [:state, :updated_at]

  end

  # A reporting run is not a stale run: every trial report moves the run's

  # `updated_at`, which is the clock the staleness sweep reads.

  defp touch(%Run{id: run_id}) do

    from(r in Run, where: r.id == ^run_id)

    |> Repo.update_all(set: [updated_at: DateTime.utc_now()])

  end

  defp trials_query, do: from(t in Trial, order_by: [asc: t.task])

  defp run_topic(run_id), do: "gym:run:" <> run_id

  defp broadcast_run(%Run{} = run) do

    broadcast(run.id, {:gym_run, run})

  end

  defp broadcast_trial(%Trial{} = trial) do

    broadcast(trial.run_id, {:gym_trial, trial})

  end

  defp broadcast(run_id, message) do

    Phoenix.PubSub.broadcast(OpenAgents.PubSub, @topic, message)

    Phoenix.PubSub.broadcast(OpenAgents.PubSub, run_topic(run_id), message)

  end

end

defmodule OpenAgents.Gym.Run do

  @moduledoc """

  One graded benchmark run of an agent against a suite.

  One benchmark run of an agent against a suite, alive or graded.

  A run is a record of measurement, never of execution: the Harbor harness

  (`docs/2026-08-24-harbor-terminal-bench-plan.md`) runs the trials and this

  passed, what it cost, and the digest of the exact recipe (CLI version,

  model catalog revision, plugin set, dataset version) that produced it.

  A run moves through a small status ladder. `running` is a run the harness

  registered at suite start; its grade columns are still empty. `graded` is

  the terminal state the one-shot ingest has always written and the state a

  finalize reaches; it requires the task counts and a `completed_at`.

  `abandoned` is the terminal state for a run that died without a grade —

  declared by the harness or applied by the lazy staleness sweep — so the

  scoreboard never shows a forever-running row.

  `recipe_digest` is unique: submitting the same run twice replays the first

  row rather than duplicating it, so a trend line never counts a run twice.

  The bounded `report` map carries per-task rows and anything else the

  harness wants to keep beside the headline numbers; it is data about the

  run, not a second transcript store.

  A run registered before its recipe is pinned carries a generated

  `pending:` placeholder until finalize supplies the real digest. The

  bounded `report` map carries per-task rows and anything else the harness

  wants to keep beside the headline numbers; it is data about the run, not a

  second transcript store. Per-trial rows with their thread links live in

  `OpenAgents.Gym.Trial`.

  """

  use Ecto.Schema

  import Ecto.Changeset

  alias OpenAgents.Gym.Trial

  @primary_key {:id, :binary_id, autogenerate: true}

  @statuses ~w(running graded abandoned)

  @bounded_fields [:suite, :agent, :model, :recipe_digest]

  @maximum_report_bytes 262_144

    field :agent_version, :string

    field :model, :string

    field :lane, :string

    field :status, :string, default: "graded"

    field :tasks_total, :integer

    field :tasks_passed, :integer

    field :input_tokens, :integer

    field :duration_seconds, :integer

    field :recipe_digest, :string

    field :report, :map, default: %{}

    field :completed_at, :utc_datetime_usec

    timestamps(type: :utc_datetime_usec, updated_at: false)

    has_many :trials, Trial, foreign_key: :run_id

    timestamps(type: :utc_datetime_usec)

  end

  @type t :: %__MODULE__{}

  def statuses, do: @statuses

  @doc """

  The one-shot graded row `POST /api/v1/gym/runs` has always written.

  The caller supplies the grades with the identity; the context stamps

  `completed_at`, and `status` keeps its `graded` default.

  """

  def changeset(run, attributes) do

    run

    |> cast(attributes, [

    |> unique_constraint(:recipe_digest)

  end

  @doc "Pass rate in [0.0, 1.0]; nil for an empty suite rather than a fake 1.0."

  @doc """

  Registration at suite start: identity without grades.

  A run registered before its recipe is pinned has no digest yet, and the

  column is unique and required, so an absent digest takes a generated

  `pending:` placeholder that finalize later replaces.

  """

  def start_changeset(run, attributes) do

    run

    |> cast(attributes, [

      :suite,

      :agent,

      :agent_version,

      :model,

      :lane,

      :tasks_total,

      :recipe_digest

    ])

    |> validate_required([:suite, :agent, :model])

    |> put_placeholder_digest()

    |> put_change(:status, "running")

    |> validate_bounded_fields()

    |> validate_number(:tasks_total, greater_than_or_equal_to: 0)

    |> check_constraint(:status, name: :gym_runs_status_check)

    |> unique_constraint(:recipe_digest)

  end

  @doc "Finalization: fold the grades in and close the run as `graded`."

  def finalize_changeset(run, attributes, now) do

    run

    |> cast(attributes, [

      :agent_version,

      :model,

      :tasks_total,

      :tasks_passed,

      :input_tokens,

      :output_tokens,

      :cost_microusd,

      :duration_seconds,

      :recipe_digest,

      :report

    ])

    |> put_change(:status, "graded")

    |> put_change(:completed_at, now)

    |> validate_required([:model, :tasks_total, :tasks_passed, :recipe_digest])

    |> validate_bounded_fields()

    |> validate_number(:tasks_total, greater_than_or_equal_to: 0)

    |> validate_number(:tasks_passed, greater_than_or_equal_to: 0)

    |> validate_passed_within_total()

    |> validate_report_bound()

    |> check_constraint(:tasks_passed, name: :gym_runs_task_counts_check)

    |> check_constraint(:status, name: :gym_runs_status_check)

    |> unique_constraint(:recipe_digest)

  end

  @doc "The gradeless terminal state, declared by the harness or swept."

  def abandon_changeset(run, now) do

    run

    |> change(%{status: "abandoned", completed_at: now})

    |> check_constraint(:status, name: :gym_runs_status_check)

  end

  @doc """

  Pass rate in [0.0, 1.0]; nil for an empty suite rather than a fake 1.0,

  and nil for a run that has no grades yet.

  """

  def score(%__MODULE__{tasks_total: nil}), do: nil

  def score(%__MODULE__{tasks_passed: nil}), do: nil

  def score(%__MODULE__{tasks_total: 0}), do: nil

  def score(%__MODULE__{tasks_total: total, tasks_passed: passed}), do: passed / total

  defp put_placeholder_digest(changeset) do

    case get_field(changeset, :recipe_digest) do

      nil -> put_change(changeset, :recipe_digest, "pending:" <> Ecto.UUID.generate())

      _present -> changeset

    end

  end

  defp validate_bounded_fields(changeset) do

    Enum.reduce(@bounded_fields, changeset, fn field, acc ->

      validate_length(acc, field, min: 1, max: 200, count: :bytes)

defmodule OpenAgents.Gym.Trial do

  @moduledoc """

  One task of a run: its name, its state, and — on the thread lane — the

  thread the coder opened for it.

  A trial is upserted by `(run_id, task)`: the harness reports `running`

  when the trial starts and reports again when the grade lands, and both

  reports name the same row. `thread_id` is the link between the Gym and

  the transcript that already streams through this server; it is verified

  against the bearer's account at ingest (`OpenAgents.Gym.record_trial/3`)

  and carries no foreign key, because a thread may be deleted with its

  account while the benchmark record stays. Local-lane trials have no

  thread and leave it nil — expected, not an error.

  """

  use Ecto.Schema

  import Ecto.Changeset

  alias OpenAgents.Gym.Run

  @primary_key {:id, :binary_id, autogenerate: true}

  @foreign_key_type :binary_id

  @states ~w(running passed failed ungraded)

  schema "gym_trials" do

    belongs_to :run, Run

    field :task, :string

    field :state, :string

    field :thread_id, :binary_id

    timestamps(type: :utc_datetime_usec)

  end

  @type t :: %__MODULE__{}

  def states, do: @states

  @doc """

  One trial report. `run_id` is set by the context, never cast from a

  caller, and the thread-ownership check lives in the context beside the

  account it needs.

  """

  def changeset(trial, attributes) do

    trial

    |> cast(attributes, [:task, :state, :thread_id])

    |> validate_required([:task, :state])

    |> validate_length(:task, min: 1, max: 200, count: :bytes)

    |> validate_inclusion(:state, @states)

    |> check_constraint(:task, name: :gym_trials_task_bound_check)

    |> check_constraint(:state, name: :gym_trials_state_check)

    |> foreign_key_constraint(:run_id)

    |> unique_constraint([:run_id, :task])

  end

end

    # Reporting it as `not_found` would tell a pusher their push is not on

    # record, which is a different and much worse claim.

    "push_record_unreadable" => {503, "The push record is temporarily unreadable"},

    # Gym run lifecycle. A harness that scripts against the lifecycle meets

    # two conflicts it must tell apart: the run it is closing was already

    # graded, and the digest it is pinning already names a different run.

    # Each carries the standing run beside the envelope so the caller reads

    # what it lost to rather than fetching again.

    "run_already_graded" => {409, "This run is already graded"},

    "recipe_digest_conflict" => {409, "That recipe digest already names another run"},

    "trace_body_too_large" => {413, "The trace body is larger than the maximum allowed size"}

  }

      "get /api/v1/agent" => {:required_bearer, :agent, :legacy},

      "post /api/v1/agents/:handle/box-control" => {:required_bearer, :agent, :legacy},

      "post /api/v1/gym/runs" => {:required_bearer, :gym, :envelope},

      "post /api/v1/gym/runs/start" => {:required_bearer, :gym, :envelope},

      "post /api/v1/gym/runs/:id/trials" => {:required_bearer, :gym, :envelope},

      "patch /api/v1/gym/runs/:id" => {:required_bearer, :gym, :envelope},

      "get /api/v1/gym/runs" => {:required_bearer, :gym, :envelope},

      "delete /api/v1/agents/:handle/box-control" => {:required_bearer, :agent, :legacy},

      "post /api/v1/agents/:handle/computer-control" => {:required_bearer, :agent, :legacy},

defmodule OpenAgentsWeb.GymRunController do

  @moduledoc """

  The door the bench harness posts graded runs through.

  The door the bench harness reports runs through — one-shot and live.

  Authority is the fleet-promotion shape without the privileged scope: an

  ordinary `forge:write` bearer plus live operator standing, rechecked on

  Idempotent by recipe digest: a retried upload answers `200` with the

  existing row where the first answered `201`, so harness retry policy

  needs no special casing.

  needs no special casing. The lifecycle routes add the live half: `start`

  registers a run as `running`, `create_trial` upserts one task's state

  (optionally linked to the thread carrying its transcript, verified

  against the bearer's account), and `update` closes the run — `graded`

  with the totals, or `abandoned` without them. The two `409` refusals

  carry the colliding run beside the envelope so a harness can read what

  it lost to.

  """

  use OpenAgentsWeb, :controller

  alias OpenAgents.Accounts

  alias OpenAgents.Gym

  alias OpenAgents.Gym.Run

  alias OpenAgents.Gym.Trial

  alias OpenAgentsWeb.ApiError

  def create(conn, params) do

    end

  end

  def start(conn, params) do

    with :ok <- operator(conn) do

      case Gym.start_run(params) do

        {:ok, run, replayed?} ->

          conn

          |> put_status(if(replayed?, do: :ok, else: :created))

          |> json(%{"run" => run_view(run), "replayed" => replayed?})

        {:error, changeset} ->

          ApiError.changeset(conn, changeset)

      end

    end

  end

  def create_trial(conn, %{"id" => run_id} = params) do

    with :ok <- operator(conn) do

      case Gym.get_run(run_id) do

        %Run{} = run ->

          case Gym.record_trial(conn.assigns.current_user, run, Map.delete(params, "id")) do

            {:ok, trial} ->

              json(conn, %{"trial" => trial_view(trial)})

            {:error, :trial_limit} ->

              ApiError.validation_failed(conn, %{

                "task" => ["this run already holds the maximum number of trials"]

              })

            {:error, changeset} ->

              ApiError.changeset(conn, changeset)

          end

        nil ->

          ApiError.not_found(conn)

      end

    end

  end

  def update(conn, %{"id" => run_id} = params) do

    with :ok <- operator(conn) do

      case Gym.get_run(run_id) do

        %Run{} = run -> close(conn, run, params)

        nil -> ApiError.not_found(conn)

      end

    end

  end

  defp close(conn, run, %{"status" => "graded"} = params) do

    case Gym.finalize_run(run, Map.drop(params, ["id", "status"])) do

      {:ok, updated} ->

        json(conn, %{"run" => run_view(updated)})

      {:error, :already_graded, graded} ->

        ApiError.refuse(conn, "run_already_graded", legacy: %{"run" => run_view(graded)})

      {:error, :digest_conflict, existing} ->

        ApiError.refuse(conn, "recipe_digest_conflict", legacy: %{"run" => run_view(existing)})

      {:error, changeset} ->

        ApiError.changeset(conn, changeset)

    end

  end

  defp close(conn, run, %{"status" => "abandoned"}) do

    case Gym.abandon_run(run) do

      {:ok, updated} ->

        json(conn, %{"run" => run_view(updated)})

      {:error, :already_graded, graded} ->

        ApiError.refuse(conn, "run_already_graded", legacy: %{"run" => run_view(graded)})

    end

  end

  defp close(conn, _run, _params) do

    ApiError.validation_failed(conn, %{"status" => ["must be graded or abandoned"]})

  end

  def index(conn, params) do

    with :ok <- operator(conn) do

      runs = Gym.list_runs(suite: params["suite"])

      "agent_version" => run.agent_version,

      "model" => run.model,

      "lane" => run.lane,

      "status" => run.status,

      "tasks_total" => run.tasks_total,

      "tasks_passed" => run.tasks_passed,

      "score" => Run.score(run),

      "cost_microusd" => run.cost_microusd,

      "duration_seconds" => run.duration_seconds,

      "recipe_digest" => run.recipe_digest,

      "recorded_at" => run.inserted_at

      "recorded_at" => run.inserted_at,

      "completed_at" => run.completed_at

    }

  end

  defp trial_view(%Trial{} = trial) do

    %{

      "id" => trial.id,

      "run_id" => trial.run_id,

      "task" => trial.task,

      "state" => trial.state,

      "thread_id" => trial.thread_id,

      "recorded_at" => trial.inserted_at,

      "updated_at" => trial.updated_at

    }

  end

end

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

  end

  # 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}"

  defp counts(_passed, _total), do: "—"

  defp tokens(nil, nil), do: "—"

  defp tokens(input, output), do: "#{format_count(input)} in / #{format_count(output)} out"

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

                <td>{run.lane || "—"}</td>

                <td class="font-semibold">{percent(Run.score(run))}</td>

                <td>{run.tasks_passed}/{run.tasks_total}</td>

                <td>{counts(run.tasks_passed, run.tasks_total)}</td>

                <td class="whitespace-nowrap">{elapsed(run.duration_seconds)}</td>

                <td class="whitespace-nowrap text-sm">

                  {tokens(run.input_tokens, run.output_tokens)}

    # The Gym's ingest and read: an ordinary forge:write bearer carries the

    # request, and the controller rechecks live operator standing on every

    # call — the fleet-promotion shape without a privileged scope, because

    # recording a benchmark row moves no money and deploys nothing.

    # recording a benchmark row moves no money and deploys nothing. The

    # lifecycle routes (start, trial upsert, finalize) share the exact same

    # posture.

    post "/gym/runs", GymRunController, :create

    post "/gym/runs/start", GymRunController, :start

    post "/gym/runs/:id/trials", GymRunController, :create_trial

    patch "/gym/runs/:id", GymRunController, :update

    get "/gym/runs", GymRunController, :index

  end

    20260825024500,

    20260825054906,

    20260825101934,

    20260825102408

    20260825102408,

    20260825113459

  ],

  "required_tables": [

    "users",

defmodule OpenAgents.Repo.Migrations.AddGymRunLifecycle do

  use Ecto.Migration

  def up do

    # A run now exists while it runs. `status` defaults to `graded` so every

    # legacy row — recorded only after the fact — keeps the meaning it had,

    # and `updated_at`/`completed_at` are backfilled from `inserted_at` for

    # the same reason: a one-shot row was complete the moment it was written.

    alter table(:gym_runs) do

      add :status, :text, null: false, default: "graded"

      add :completed_at, :utc_datetime_usec

      add :updated_at, :utc_datetime_usec

    end

    execute "UPDATE gym_runs SET updated_at = inserted_at, completed_at = inserted_at"

    execute "ALTER TABLE gym_runs ALTER COLUMN updated_at SET NOT NULL"

    # A running row has no grades yet; the changeset requires them by status

    # and the constraint below keeps a graded row honest at the database.

    execute "ALTER TABLE gym_runs ALTER COLUMN tasks_total DROP NOT NULL"

    execute "ALTER TABLE gym_runs ALTER COLUMN tasks_passed DROP NOT NULL"

    create constraint(:gym_runs, :gym_runs_status_check,

             check: "status IN ('running', 'graded', 'abandoned')"

           )

    create constraint(:gym_runs, :gym_runs_graded_shape_check,

             check:

               "status <> 'graded' OR (tasks_total IS NOT NULL AND tasks_passed IS NOT NULL AND completed_at IS NOT NULL)"

           )

    # The lazy staleness sweep reads exactly this slice.

    create index(:gym_runs, [:updated_at], where: "status = 'running'")

    create table(:gym_trials, primary_key: false) do

      add :id, :binary_id, primary_key: true

      add :run_id, references(:gym_runs, type: :binary_id, on_delete: :delete_all), null: false

      add :task, :text, null: false

      add :state, :text, null: false

      add :thread_id, :binary_id

      timestamps(type: :utc_datetime_usec)

    end

    create constraint(:gym_trials, :gym_trials_task_bound_check,

             check: "octet_length(task) BETWEEN 1 AND 200"

           )

    create constraint(:gym_trials, :gym_trials_state_check,

             check: "state IN ('running', 'passed', 'failed', 'ungraded')"

           )

    # One row per task per run: a re-reported trial updates in place.

    create unique_index(:gym_trials, [:run_id, :task])

  end

  def down do

    drop table(:gym_trials)

    drop index(:gym_runs, [:updated_at], where: "status = 'running'")

    drop constraint(:gym_runs, :gym_runs_graded_shape_check)

    drop constraint(:gym_runs, :gym_runs_status_check)

    execute "DELETE FROM gym_runs WHERE tasks_total IS NULL OR tasks_passed IS NULL"

    execute "ALTER TABLE gym_runs ALTER COLUMN tasks_total SET NOT NULL"

    execute "ALTER TABLE gym_runs ALTER COLUMN tasks_passed SET NOT NULL"

    alter table(:gym_runs) do

      remove :status

      remove :completed_at

      remove :updated_at

    end

  end

end

defmodule OpenAgents.GymTest do

  use OpenAgents.DataCase, async: true

  import Ecto.Query

  import OpenAgentsWeb.ConnCase, only: [github_user: 1]

  alias OpenAgents.Gym

  alias OpenAgents.Gym.Run

  alias OpenAgents.Gym.Trial

  alias OpenAgents.Repo

  alias OpenAgents.Threads

  defp attributes(overrides \\ %{}) do

    Map.merge(

    assert {:error, changeset} = Gym.record_run(attributes(%{"report" => huge}))

    assert %{report: [_message]} = errors_on(changeset)

  end

  defp start_attributes(overrides \\ %{}) do

    Map.merge(

      %{

        "suite" => "terminal-bench@2.0",

        "agent" => "openagents-coder",

        "model" => "ox-alpha",

        "lane" => "proxy",

        "tasks_total" => 5

      },

      overrides

    )

  end

  describe "start_run/1" do

    test "registers a running run with a placeholder digest and broadcasts it" do

      :ok = Gym.subscribe()

      assert {:ok, %Run{} = run, false} = Gym.start_run(start_attributes())

      assert run.status == "running"

      assert run.tasks_passed == nil

      assert run.completed_at == nil

      assert String.starts_with?(run.recipe_digest, "pending:")

      assert Run.score(run) == nil

      run_id = run.id

      assert_receive {:gym_run, %Run{id: ^run_id, status: "running"}}

    end

    test "a resubmitted digest replays the running row" do

      digest = "sha256:" <> String.duplicate("f", 64)

      {:ok, first, false} = Gym.start_run(start_attributes(%{"recipe_digest" => digest}))

      {:ok, second, true} = Gym.start_run(start_attributes(%{"recipe_digest" => digest}))

      assert second.id == first.id

      assert length(Gym.list_runs()) == 1

    end

    test "identity is required" do

      assert {:error, changeset} = Gym.start_run(%{"suite" => "terminal-bench@2.0"})

      assert %{agent: [_agent], model: [_model]} = errors_on(changeset)

    end

  end