defmodule OpenAgents.TokenProductivity do

  @moduledoc """

  Productive token accounting: raw volume versus tokens that produced durable

  outcomes.

  Raw volume alone rewards burn. This module reads the four tables that hold

  token truth — `turn_receipts`, `voice_sessions`, `work_jobs`, and `scv_runs`

  — and splits the same tokens a second way: how many are attached to evidence

  that the work landed. Three buckets, strongest evidence first, each usage

  row counted at most once:

    * **Merged work** — an SCV run whose issue carries a merged pull request.

    * **Closed issues** — an SCV run whose issue closed without a merged pull

      request.

    * **Verified receipts** — an SCV run that succeeded with a digest-verified

      terminal receipt, or a work job that completed with its bounded report,

      where no stronger outcome evidence exists.

  Typed and spoken conversation tokens count toward raw volume only: a chat

  turn has no durable outcome record to attribute it to.

  ## Which usage counts

  The same double-count rule as `OpenAgents.Leaderboard` applies:

  `turn_receipts.usage` is already the merge of every tool-loop provider call

  in a typed turn, and `voice_sessions.usage` is already the merge of that

  session's responses, so `turn_provider_steps.usage` is never summed into raw

  totals. Provider steps are read separately — and only for the per-provider

  throughput table, where attributing tokens to the provider that produced

  them is the whole point. `work_jobs` rows come from the OpenCode-driven work

  machinery and `scv_runs` rows from the Codex driver; the two never describe

  the same run.

  ## Rates

  Stored usage maps spell their cache field three ways: `cached_input_tokens`

  (typed turns, Codex runs) and `input_cached_tokens` (voice) count cached

  tokens inside `input_tokens`, while `cache_read_tokens` (OpenCode) counts

  them separately. The cache hit rate therefore divides all cached tokens by

  `input_tokens` plus the exclusive `cache_read_tokens`, so both spellings

  land in the same denominator. The input share uses that same inclusive

  input count over input plus output.

  """

  import Ecto.Query

  alias OpenAgents.Conversations.ProviderStep

  alias OpenAgents.Conversations.TurnReceipt

  alias OpenAgents.Issues.Issue

  alias OpenAgents.PullRequests.PullRequest

  alias OpenAgents.Repo

  alias OpenAgents.SCV.Execution

  alias OpenAgents.Voice.Session

  alias OpenAgents.Work.Job

  @schema "openagents.token_productivity.v1"

  @empty_totals %{

    input_tokens: 0,

    output_tokens: 0,

    cached_input_tokens: 0,

    cache_read_tokens: 0,

    total_tokens: 0

  }

  @type totals :: %{

          input_tokens: non_neg_integer(),

          output_tokens: non_neg_integer(),

          cached_input_tokens: non_neg_integer(),

          cache_read_tokens: non_neg_integer(),

          total_tokens: non_neg_integer()

        }

  @type provider_row :: %{

          provider_id: String.t(),

          steps: non_neg_integer(),

          input_tokens: non_neg_integer(),

          output_tokens: non_neg_integer(),

          cached_input_tokens: non_neg_integer(),

          total_tokens: non_neg_integer(),

          duration_ms: non_neg_integer(),

          tokens_per_second: float() | nil

        }

  @type report :: %{

          schema: String.t(),

          generated_at: DateTime.t(),

          raw: totals(),

          sources: %{

            typed_turns: totals(),

            voice_sessions: totals(),

            work_jobs: totals(),

            scv_runs: totals()

          },

          productive: %{

            merged_work: totals(),

            closed_issues: totals(),

            verified_receipts: totals(),

            total_tokens: non_neg_integer(),

            share: float() | nil

          },

          cache: %{

            cached_input_tokens: non_neg_integer(),

            input_tokens: non_neg_integer(),

            hit_rate: float() | nil

          },

          split: %{

            input_tokens: non_neg_integer(),

            output_tokens: non_neg_integer(),

            input_share: float() | nil

          },

          providers: [provider_row()]

        }

  # SUM of one numeric JSON field, defensive against absent keys and

  # non-numeric values, exactly like the leaderboard's usage fragments. The

  # key is a compile-time literal, inlined so the operator stays jsonb ->> text.

  defmacrop summed(usage, key) when is_binary(key) do

    expression =

      "COALESCE(SUM(CASE WHEN ? ->> '#{key}' ~ '^[0-9]+$' THEN (? ->> '#{key}')::bigint ELSE 0 END), 0)"

    quote do

      fragment(unquote(expression), unquote(usage), unquote(usage))

    end

  end

  # Per-row effective total: providers that omit total_tokens still count.

  defmacrop summed_effective_total(usage) do

    quote do

      fragment(

        """

        COALESCE(SUM(GREATEST(

          CASE WHEN ? ->> 'total_tokens' ~ '^[0-9]+$' THEN (? ->> 'total_tokens')::bigint ELSE 0 END,

          CASE WHEN ? ->> 'input_tokens' ~ '^[0-9]+$' THEN (? ->> 'input_tokens')::bigint ELSE 0 END

            + CASE WHEN ? ->> 'output_tokens' ~ '^[0-9]+$' THEN (? ->> 'output_tokens')::bigint ELSE 0 END

        )), 0)

        """,

        unquote(usage),

        unquote(usage),

        unquote(usage),

        unquote(usage),

        unquote(usage),

        unquote(usage)

      )

    end

  end

  @doc "Computes the full report straight from PostgreSQL."

  @spec report() :: report()

  def report do

    sources = %{

      typed_turns: totals(from(receipt in TurnReceipt, as: :usage_row)),

      voice_sessions: totals(from(session in Session, as: :usage_row)),

      work_jobs: totals(from(job in Job, as: :usage_row)),

      scv_runs: totals(from(run in Execution, as: :usage_row))

    }

    raw =

      Enum.reduce(Map.values(sources), @empty_totals, &merge_totals/2)

    merged_work = totals(merged_work_query())

    closed_issues = totals(closed_issues_query())

    verified_receipts =

      merge_totals(totals(receipt_runs_query()), totals(completed_jobs_query()))

    productive_total =

      merged_work.total_tokens + closed_issues.total_tokens + verified_receipts.total_tokens

    cached = raw.cached_input_tokens

    input_inclusive = raw.input_tokens + raw.cache_read_tokens

    %{

      schema: @schema,

      generated_at: DateTime.utc_now(),

      raw: raw,

      sources: sources,

      productive: %{

        merged_work: merged_work,

        closed_issues: closed_issues,

        verified_receipts: verified_receipts,

        total_tokens: productive_total,

        share: ratio(productive_total, raw.total_tokens)

      },

      cache: %{

        cached_input_tokens: cached,

        input_tokens: input_inclusive,

        hit_rate: ratio(cached, input_inclusive)

      },

      split: %{

        input_tokens: input_inclusive,

        output_tokens: raw.output_tokens,

        input_share: ratio(input_inclusive, input_inclusive + raw.output_tokens)

      },

      providers: providers()

    }

  end

  @doc """

  Per-provider volume and throughput from completed provider steps.

  Provider steps carry each attempt's own usage and wall clock, so this is the

  one place they are read: tokens per second come from output tokens over the

  step's recorded duration.

  """

  @spec providers() :: [provider_row()]

  def providers do

    from(step in ProviderStep,

      as: :usage_row,

      where: step.status == "completed" and not is_nil(step.usage),

      group_by: step.provider_id,

      order_by: [desc: summed_effective_total(step.usage)],

      select: %{

        provider_id: step.provider_id,

        steps: count(step.id),

        input_tokens: summed(step.usage, "input_tokens"),

        output_tokens: summed(step.usage, "output_tokens"),

        cached_input_tokens: summed(step.usage, "cached_input_tokens"),

        total_tokens: summed_effective_total(step.usage),

        duration_ms:

          fragment(

            "COALESCE(SUM(EXTRACT(EPOCH FROM (? - ?)) * 1000) FILTER (WHERE ? IS NOT NULL), 0)::bigint",

            step.completed_at,

            step.started_at,

            step.completed_at

          )

      }

    )

    |> Repo.all()

    |> Enum.map(fn row ->

      row =

        Map.new(row, fn

          {:provider_id, value} -> {:provider_id, value}

          {key, value} -> {key, integer(value)}

        end)

      Map.put(row, :tokens_per_second, throughput(row.output_tokens, row.duration_ms))

    end)

  end

  # Every totals query names its usage-bearing binding :usage_row so one

  # select works for all of them.

  defp totals(query) do

    query

    |> select([usage_row: row], %{

      input_tokens: summed(row.usage, "input_tokens"),

      output_tokens: summed(row.usage, "output_tokens"),

      cached_input_tokens:

        summed(row.usage, "cached_input_tokens") +

          summed(row.usage, "input_cached_tokens") +

          summed(row.usage, "cache_read_tokens"),

      cache_read_tokens: summed(row.usage, "cache_read_tokens"),

      total_tokens: summed_effective_total(row.usage)

    })

    |> Repo.one()

    |> Map.new(fn {key, value} -> {key, integer(value)} end)

  end

  # A run's issue with a merged pull request is the strongest outcome evidence.

  defp merged_work_query do

    from(run in Execution,

      as: :usage_row,

      join: pull_request in PullRequest,

      on: pull_request.issue_id == run.issue_id,

      where: not is_nil(pull_request.merged_at)

    )

  end

  # Closed without a merged pull request: the issue itself was the outcome.

  defp closed_issues_query do

    from(run in Execution,

      as: :usage_row,

      join: issue in Issue,

      on: issue.id == run.issue_id,

      left_join: pull_request in PullRequest,

      on: pull_request.issue_id == run.issue_id,

      where: issue.state == "closed",

      where: is_nil(pull_request.id) or is_nil(pull_request.merged_at)

    )

  end

  # Succeeded runs carry a digest-verified terminal receipt; count the ones no

  # stronger bucket already counted.

  defp receipt_runs_query do

    from(run in Execution,

      as: :usage_row,

      left_join: issue in Issue,

      on: issue.id == run.issue_id,

      left_join: pull_request in PullRequest,

      on: pull_request.issue_id == run.issue_id,

      where: run.status == "succeeded",

      where: is_nil(issue.id) or issue.state != "closed",

      where: is_nil(pull_request.id) or is_nil(pull_request.merged_at)

    )

  end

  # A completed work job's terminal row must carry its bounded report — that

  # report is the receipt.

  defp completed_jobs_query do

    from(job in Job, as: :usage_row, where: job.status == "completed")

  end

  defp merge_totals(left, right) do

    Map.new(@empty_totals, fn {key, _zero} ->

      {key, Map.fetch!(left, key) + Map.fetch!(right, key)}

    end)

  end

  defp ratio(_numerator, denominator) when denominator in [0, nil], do: nil

  defp ratio(numerator, denominator), do: numerator / denominator

  defp throughput(_output_tokens, duration_ms) when duration_ms <= 0, do: nil

  defp throughput(output_tokens, duration_ms), do: output_tokens / (duration_ms / 1000)

  defp integer(value) when is_integer(value), do: value

  defp integer(%Decimal{} = value), do: Decimal.to_integer(value)

  defp integer(_value), do: 0

end

defmodule OpenAgentsWeb.AdminTokensLive do

  @moduledoc """

  Operator view of productive token usage versus raw volume.

  Every number is computed by `OpenAgents.TokenProductivity` at request time

  straight from PostgreSQL, the same tables the leaderboard reads, so the

  surface adds no second aggregation authority. Read-only and operator-gated.

  It shows aggregate token counts and rates only. It never renders

  conversation content, objectives, reports, or anything a person wrote.

  """

  use OpenAgentsWeb, :live_view

  alias OpenAgents.Accounts

  alias OpenAgents.TokenProductivity

  @impl true

  def mount(_params, _session, socket) do

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

      socket =

        socket

        |> assign(:page_title, "Operator · Tokens")

        |> assign(:status, :loading)

        |> assign(:report, nil)

      if connected?(socket) do

        send(self(), :load)

      end

      {:ok, socket}

    else

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

    end

  end

  @impl true

  def handle_event("refresh", _params, socket) do

    # Re-checked per event, not only at mount: a long-lived socket outlives the

    # decision that opened it.

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

      {:noreply,

       socket

       |> assign(:status, :loading)

       |> load()}

    else

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

    end

  end

  @impl true

  def handle_info(:load, socket), do: {:noreply, load(socket)}

  defp load(%{assigns: %{status: :loading}} = socket) do

    assign(socket, status: :loaded, report: TokenProductivity.report())

  end

  # A refresh while a load is already resolving must not clobber the newer

  # state with an older response.

  defp load(socket), do: socket

  @impl true

  def render(assigns) do

    ~H"""

    <Layouts.app

      flash={@flash}

      sidebar_sections={assigns[:sidebar_sections]}

      current_scope={@current_scope}

      title="Token productivity"

    >

      <main id="admin-tokens-page" class="app-shell admin-shell">

        <section class="admin space-y-8" aria-labelledby="tokens-heading">

          <header class="admin-heading">

            <h1 id="tokens-heading">Token productivity</h1>

            <p>

              Raw token volume next to the tokens that produced durable outcomes —

              merged work, closed issues, and verified receipts. Aggregate counts

              only; no conversation or run content is readable from this page.

            </p>

            <div class="admin-totals">

              <.badge

                :if={@status == :loaded && @report}

                variant={:dim}

                id="tokens-generated-at"

              >

                GENERATED {Calendar.strftime(@report.generated_at, "%Y-%m-%d %H:%M UTC")}

              </.badge>

              <.text_button id="tokens-refresh" phx-click="refresh" disabled={@status == :loading}>

                {if(@status == :loading, do: "REFRESHING…", else: "REFRESH")}

              </.text_button>

            </div>

          </header>

          <.alert :if={@status == :loading} id="tokens-loading" appearance={:row}>

            Computing token totals…

          </.alert>

          <div :if={@status == :loaded && @report} class="space-y-8">

            <section aria-labelledby="productive-heading">

              <.card id="tokens-productive">

                <h2 id="productive-heading" class="card-title">Productive versus raw</h2>

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

                  <.badge>{format_tokens(@report.raw.total_tokens)} RAW</.badge>

                  <.badge variant={:success}>

                    {format_tokens(@report.productive.total_tokens)} PRODUCTIVE

                  </.badge>

                  <.badge :if={@report.productive.share} variant={:info}>

                    {format_share(@report.productive.share)} PRODUCTIVE SHARE

                  </.badge>

                </div>

                <ul class="divide-y divide-border">

                  <li class="flex items-center justify-between gap-4 py-3">

                    <span>Merged work</span>

                    <span class="font-semibold">

                      {format_tokens(@report.productive.merged_work.total_tokens)}

                    </span>

                  </li>

                  <li class="flex items-center justify-between gap-4 py-3">

                    <span>Closed issues</span>

                    <span class="font-semibold">

                      {format_tokens(@report.productive.closed_issues.total_tokens)}

                    </span>

                  </li>

                  <li class="flex items-center justify-between gap-4 py-3">

                    <span>Verified receipts</span>

                    <span class="font-semibold">

                      {format_tokens(@report.productive.verified_receipts.total_tokens)}

                    </span>

                  </li>

                </ul>

              </.card>

            </section>

            <section aria-labelledby="rates-heading">

              <.card id="tokens-rates">

                <h2 id="rates-heading" class="card-title">Cache and split</h2>

                <ul class="divide-y divide-border">

                  <li class="flex items-center justify-between gap-4 py-3">

                    <span>Cache hit rate</span>

                    <span class="font-semibold">{format_share(@report.cache.hit_rate)}</span>

                  </li>

                  <li class="flex items-center justify-between gap-4 py-3">

                    <span>Cached input tokens</span>

                    <span class="font-semibold">

                      {format_tokens(@report.cache.cached_input_tokens)}

                    </span>

                  </li>

                  <li class="flex items-center justify-between gap-4 py-3">

                    <span>Input share</span>

                    <span class="font-semibold">{format_share(@report.split.input_share)}</span>

                  </li>

                  <li class="flex items-center justify-between gap-4 py-3">

                    <span>Input / output tokens</span>

                    <span class="font-semibold">

                      {format_tokens(@report.split.input_tokens)} / {format_tokens(

                        @report.split.output_tokens

                      )}

                    </span>

                  </li>

                </ul>

              </.card>

            </section>

            <section aria-labelledby="sources-heading">

              <.card id="tokens-sources">

                <h2 id="sources-heading" class="card-title">Raw volume by source</h2>

                <.table id="tokens-sources-table" rows={source_rows(@report.sources)}>

                  <:col :let={row} label="Source">{row.label}</:col>

                  <:col :let={row} label="Input">{format_tokens(row.totals.input_tokens)}</:col>

                  <:col :let={row} label="Output">{format_tokens(row.totals.output_tokens)}</:col>

                  <:col :let={row} label="Total">{format_tokens(row.totals.total_tokens)}</:col>

                </.table>

              </.card>

            </section>

            <section aria-labelledby="providers-heading">

              <.card id="tokens-providers">

                <h2 id="providers-heading" class="card-title">Provider throughput</h2>

                <p :if={@report.providers == []} class="text-muted-foreground">

                  No completed provider steps recorded yet.

                </p>

                <.table

                  :if={@report.providers != []}

                  id="tokens-providers-table"

                  rows={@report.providers}

                >

                  <:col :let={row} label="Provider">{row.provider_id}</:col>

                  <:col :let={row} label="Steps">{row.steps}</:col>

                  <:col :let={row} label="Input">{format_tokens(row.input_tokens)}</:col>

                  <:col :let={row} label="Output">{format_tokens(row.output_tokens)}</:col>

                  <:col :let={row} label="Tokens/s">{format_rate(row.tokens_per_second)}</:col>

                </.table>

              </.card>

            </section>

          </div>

        </section>

      </main>

    </Layouts.app>

    """

  end

  defp source_rows(sources) do

    [

      %{label: "Typed turns", totals: sources.typed_turns},

      %{label: "Voice sessions", totals: sources.voice_sessions},

      %{label: "Work jobs", totals: sources.work_jobs},

      %{label: "SCV runs", totals: sources.scv_runs}

    ]

  end

  defp format_tokens(count) when is_integer(count) do

    count

    |> Integer.to_charlist()

    |> Enum.reverse()

    |> Enum.chunk_every(3)

    |> Enum.join(",")

    |> String.reverse()

  end

  defp format_share(nil), do: "n/a"

  defp format_share(share) when is_float(share),

    do: "#{:erlang.float_to_binary(share * 100, decimals: 1)}%"

  defp format_rate(nil), do: "n/a"

  defp format_rate(rate) when is_float(rate),

    do: :erlang.float_to_binary(rate, decimals: 1)

end

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

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

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

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

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

    do: declaration(:operator, "configured operator GitHub ID", "forge:promote", true)

      ] do

      live "/", AdminLive, :index

      live "/analytics", AdminAnalyticsLive, :index

      live "/tokens", AdminTokensLive, :index

      live "/forge", AdminForgeLive, :index

      live "/recordings", AdminRecordingsLive, :index

      live "/scv/accounts", AdminScvAccountsLive, :index

defmodule OpenAgents.TokenProductivityTest do

  @moduledoc """

  Productive attribution is exclusive bucketing over durable outcome evidence,

  so the tests that matter are the boundaries: which bucket a run lands in,

  what never counts as productive, and that provider steps stay out of the raw

  totals they would double-count.

  """

  use OpenAgents.DataCase, async: false

  import OpenAgents.IssuesFixtures

  alias OpenAgents.Accounts

  alias OpenAgents.Context.Composer

  alias OpenAgents.Conversations

  alias OpenAgents.Conversations.ProviderStep

  alias OpenAgents.Issues

  alias OpenAgents.Providers.Request

  alias OpenAgents.PullRequests.PullRequest

  alias OpenAgents.SCV.DriverAccount

  alias OpenAgents.SCV.Execution

  alias OpenAgents.TokenProductivity

  alias OpenAgents.Work.Job

  test "sums raw volume by source without reading provider steps" do

    user = account("raw-sources")

    conversation = conversation_for(user)

    %{turn: turn, receipt: receipt} = begin_typed_turn(conversation, "Raw volume turn.")

    # The provider step carries its own copy of the same tokens the receipt

    # merges; counting both would double the raw total.

    {:ok, _step} =

      Conversations.record_provider_step_completion(receipt, "response-raw", %{

        "input_tokens" => 100,

        "output_tokens" => 40,

        "total_tokens" => 140

      })

    {:ok, _turn} =

      Conversations.complete_turn(turn, "response-raw", %{

        "input_tokens" => 100,

        "output_tokens" => 40,

        "total_tokens" => 140

      })

    insert_job(conversation, "completed", %{

      "input_tokens" => 30,

      "output_tokens" => 10,

      "total_tokens" => 40

    })

    account = driver_account("raw-sources-driver")

    insert_run(account, 1, "failed", nil, %{

      "input_tokens" => 8,

      "output_tokens" => 2,

      "total_tokens" => 10

    })

    report = TokenProductivity.report()

    assert report.sources.typed_turns.total_tokens == 140

    assert report.sources.work_jobs.total_tokens == 40

    assert report.sources.scv_runs.total_tokens == 10

    assert report.sources.voice_sessions.total_tokens == 0

    assert report.raw.total_tokens == 190

    assert report.raw.input_tokens == 138

    assert report.raw.output_tokens == 52

  end

  test "buckets productive tokens by strongest outcome evidence, each row once" do

    repository = repository_fixture()

    account = driver_account("productive-driver")

    # Merged work: the run's issue carries a merged pull request.

    merged_issue = issue_fixture(repository, %{title: "merged work"})

    insert_merged_pull_request(repository, merged_issue)

    insert_run(account, 1, "succeeded", merged_issue.id, %{

      "input_tokens" => 70,

      "output_tokens" => 30,

      "total_tokens" => 100

    })

    # Closed issue: closed without a merged pull request.

    closed_issue = issue_fixture(repository, %{title: "closed issue"})

    {:ok, _closed} = Issues.update_issue(closed_issue, %{"state" => "closed"})

    insert_run(account, 2, "succeeded", closed_issue.id, %{

      "input_tokens" => 14,

      "output_tokens" => 6,

      "total_tokens" => 20

    })

    # Verified receipt: succeeded with its terminal receipt, no stronger

    # evidence.

    insert_run(account, 3, "succeeded", nil, %{

      "input_tokens" => 7,

      "output_tokens" => 3,

      "total_tokens" => 10

    })

    # A failed run is raw volume only.

    insert_run(account, 4, "failed", nil, %{

      "input_tokens" => 300,

      "output_tokens" => 100,

      "total_tokens" => 400

    })

    # Completed work jobs carry their bounded report; failed ones count as raw

    # volume only.

    conversation = conversation_for(account("productive-jobs"))

    insert_job(conversation, "completed", %{"input_tokens" => 4, "output_tokens" => 1})

    insert_job(conversation, "failed", %{"input_tokens" => 900, "output_tokens" => 100})

    report = TokenProductivity.report()

    assert report.productive.merged_work.total_tokens == 100

    assert report.productive.closed_issues.total_tokens == 20

    assert report.productive.verified_receipts.total_tokens == 15

    assert report.productive.total_tokens == 135

    assert report.raw.total_tokens == 1535

    assert_in_delta report.productive.share, 135 / 1535, 0.000001

  end

  test "cache hit rate spans inclusive and exclusive cached-token spellings" do

    user = account("cache-rate")

    # Typed turns count cached tokens inside input_tokens.

    complete_typed_turn(user, "Cached typed turn.", %{

      "input_tokens" => 80,

      "output_tokens" => 20,

      "total_tokens" => 100,

      "cached_input_tokens" => 60

    })

    # OpenCode-style usage counts cache reads outside input_tokens.

    account = driver_account("cache-rate-driver")

    insert_run(account, 1, "succeeded", nil, %{

      "input_tokens" => 10,

      "output_tokens" => 5,

      "total_tokens" => 15,

      "cache_read_tokens" => 110

    })

    report = TokenProductivity.report()

    assert report.cache.cached_input_tokens == 170

    assert report.cache.input_tokens == 200

    assert_in_delta report.cache.hit_rate, 170 / 200, 0.000001

    assert_in_delta report.split.input_share, 200 / 225, 0.000001

  end

  test "provider throughput reads completed steps grouped by provider" do

    user = account("provider-throughput")

    conversation = conversation_for(user)

    # begin_typed_turn records a "started" step at sequence 1; it must stay

    # invisible to the throughput table.

    %{receipt: receipt} = begin_typed_turn(conversation, "Throughput turn.")

    Repo.insert!(%ProviderStep{

      turn_receipt_id: receipt.id,

      sequence: 2,

      provider_id: "test.provider",

      model_id: "model-v1",

      status: "completed",

      provider_response_id: "response-throughput",

      usage: %{

        "input_tokens" => 980,

        "output_tokens" => 260,

        "total_tokens" => 1240,

        "cached_input_tokens" => 850

      },

      started_at: ~U[2026-08-20 12:00:00.000000Z],

      completed_at: ~U[2026-08-20 12:00:10.000000Z]

    })

    assert [row] = TokenProductivity.providers()

    assert row.provider_id == "test.provider"

    assert row.steps == 1

    assert row.input_tokens == 980

    assert row.output_tokens == 260

    assert row.cached_input_tokens == 850

    assert row.total_tokens == 1240

    assert row.duration_ms == 10_000

    assert_in_delta row.tokens_per_second, 26.0, 0.000001

  end

  test "an empty database reports zeros and no rates" do

    report = TokenProductivity.report()

    assert report.raw.total_tokens == 0

    assert report.productive.total_tokens == 0

    assert report.productive.share == nil

    assert report.cache.hit_rate == nil

    assert report.split.input_share == nil

    assert report.providers == []

  end

  defp account(key) do

    digest = :crypto.hash(:sha256, key)

    github_id = digest |> binary_part(0, 7) |> :binary.decode_unsigned()

    suffix = digest |> Base.encode16(case: :lower) |> binary_part(0, 12)

    {:ok, user} =

      Accounts.upsert_github_user(%{

        github_id: github_id,

        github_login: "test-#{suffix}",

        github_avatar_url: "https://avatars.githubusercontent.com/u/#{github_id}?v=4"

      })

    user

  end

  defp conversation_for(user) do

    {:ok, conversation} = Conversations.ensure_conversation(user)

    conversation

  end

  defp complete_typed_turn(user, content, usage) do

    conversation = conversation_for(user)

    %{turn: turn} = begin_typed_turn(conversation, content)

    {:ok, _turn} = Conversations.complete_turn(turn, "response-#{turn.id}", usage)

    :ok

  end

  defp begin_typed_turn(conversation, content) do

    {:ok, records} = Conversations.create_turn(conversation, content)

    context = Composer.compose!()

    messages = Conversations.provider_messages(conversation.id)

    request = %Request{

      model_id: "model-v1",

      instructions: context.instructions,

      input: messages

    }

    {:ok, inference} =

      Conversations.begin_inference(records.turn, context, request, "test.provider", [])

    inference

  end

  defp driver_account(label) do

    operator = account("operator-#{label}")

    %DriverAccount{}

    |> DriverAccount.create_changeset(%{

      operator_id: operator.id,

      label: label,

      secret_ref: "file:#{label}-#{System.unique_integer([:positive])}"

    })

    |> Repo.insert!()

  end

  defp insert_run(%DriverAccount{} = account, generation, status, issue_id, usage) do

    now = DateTime.utc_now()

    report = "Run receipt for generation #{generation}."

    Repo.insert!(%Execution{

      driver_account_id: account.id,

      issue_id: issue_id,

      driver: "codex_app_server",

      principal: "scv:codex_app_server:#{account.id}",

      repository_revision: String.duplicate("a", 40),

      objective: "Bounded test objective.",

      permission_profile: "read_only",

      model: "gpt-5.6-luna",

      reasoning_effort: "low",

      status: status,

      owner_node: "test@node",

      generation: generation,

      lease_expires_at: now,

      report: report,

      report_digest: "sha256:" <> (:crypto.hash(:sha256, report) |> Base.encode16(case: :lower)),

      usage: usage,

      started_at: now,

      completed_at: now

    })

  end

  defp insert_job(conversation, status, usage) do

    Repo.insert!(%Job{

      conversation_id: conversation.id,

      owner_visitor_id: conversation.visitor_id,

      surface: "text",

      goal: "Bounded test goal.",

      status: status,

      report: "Terminal report.",

      usage: usage,

      started_at: DateTime.utc_now(),

      completed_at: DateTime.utc_now()

    })

  end

  defp insert_merged_pull_request(repository, issue) do

    Repo.insert!(%PullRequest{

      repository_id: repository.id,

      issue_id: issue.id,

      head_repository_id: repository.id,

      head_ref: "scv/merged-work",

      head_sha: String.duplicate("b", 40),

      base_ref: "main",

      base_sha: String.duplicate("c", 40),

      state: "closed",

      draft: false,

      merged_at: DateTime.utc_now()

    })

  end

end

defmodule OpenAgentsWeb.AdminTokensLiveTest do

  @moduledoc """

  `/admin/tokens` gates like every operator surface and renders aggregate

  token counts and rates only. Nothing on the page names conversation, run,

  or report content.

  """

  use OpenAgentsWeb.ConnCase, async: false

  import Phoenix.LiveViewTest

  describe "access" do

    test "the operator reaches the surface", %{conn: conn} do

      conn = log_in_admin_user(conn, "tokens-operator")

      {:ok, _view, html} = live(conn, ~p"/admin/tokens")

      assert html =~ "Token productivity"

    end

    test "an ordinary authenticated account is redirected and told nothing", %{conn: conn} do

      conn = log_in_github_user(conn, "tokens-ordinary")

      assert {:error, {:redirect, %{to: "/"}}} = live(conn, ~p"/admin/tokens")

      response = get(conn, ~p"/admin/tokens")

      assert redirected_to(response) == ~p"/"

    end

    test "an unauthenticated visitor is redirected", %{conn: conn} do

      assert {:error, {:redirect, %{to: "/"}}} = live(conn, ~p"/admin/tokens")

    end

  end

  describe "report" do

    test "renders the aggregate sections and refreshes", %{conn: conn} do

      conn = log_in_admin_user(conn, "tokens-loaded")

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

      render(view)

      assert has_element?(view, "#tokens-productive")

      assert has_element?(view, "#tokens-rates")

      assert has_element?(view, "#tokens-sources-table")

      assert has_element?(view, "#tokens-generated-at")

      html = render(view)

      assert html =~ "Merged work"

      assert html =~ "Closed issues"

      assert html =~ "Verified receipts"

      assert html =~ "Cache hit rate"

      assert html =~ "Input share"

      assert view |> element("#tokens-refresh") |> render_click() =~ "Token productivity"

      assert has_element?(view, "#tokens-generated-at")

    end

  end

end