| `chat_opened` | `ChatLive.mount` on connected mount | none |

| `chat_message_sent` | `ChatLive.launch_turn` | `length_bucket` |

| `chat_turn_completed` | terminal `turn_updated` broadcast in `ChatLive` | `outcome`: `completed`, `failed`, `cancelled`; `duration_ms` from turn timestamps |

| `chat_message_queued` | `ChatLive` queues a message behind an active turn | `length_bucket`, `queue_depth`, `conversation_id` |

| `chat_message_received` | complete assistant message in `ChatLive`; completed run in `Chat.AccountTurns` | `length_bucket`, `modality`, `conversation_id` |

| `chat_stream_chunk` | streaming assistant deltas, throttled to one event per second per stream | `conversation_id`, `modality` |

| `chat_tool_called` | `Turns.TurnServer` tool request; `tool_call_started` in `Chat.AccountTurns` | `tool_name`, `turn_id`, `conversation_id`; never arguments |

| `chat_tokens_used` | once per turn at terminal state in `Turns.TurnServer` and `Chat.AccountTurns` | `input_tokens`, `output_tokens`, `model`, `provider`, `conversation_id`, `turn_id`, `outcome` |

| `chat_turn_failed` | non-completed terminal turn in `ChatLive`; failed or cancelled run in `Chat.AccountTurns` | `reason`, `outcome`, `conversation_id`, `turn_id` |

| `chat_voice_started` / `chat_voice_ended` | voice session lifecycle broadcasts in `ChatLive` | `conversation_id`; end adds `outcome`, `duration_ms` |

| `memory_saved` | `ProfileMemory.remember_explicit` | `disposition`: `stored`, `already_active` |

| `memory_viewed` | `MemoryLive.mount` on connected mount | none |

| `computer_paired` | `ComputersController.approve_pairing` success | `tier` |

defmodule OpenAgents.Analytics.Chat do

  @moduledoc """

  The chat lifecycle event vocabulary.

  Chat instrumentation spans a LiveView, a turn server, and the OpenRouter

  provider path, so the event names, the property names, and the token

  normalization live here instead of being spelled out at each call site. Every

  function returns `:ok` through `OpenAgents.Analytics.capture/3`, which drops

  sensitive keys and never fails its caller.

  Token counts are provider-reported totals for a whole turn. `tokens_used/3`

  is the only place a turn reports them, so a turn that ran several provider

  rounds or fell back to another API still produces one event.

  """

  alias OpenAgents.Analytics

  @stream_chunk_interval_ms 1_000

  @doc "A user message that waited behind an active turn."

  @spec message_queued(Analytics.distinct_id(), map()) :: :ok

  def message_queued(distinct_id, properties \\ %{}),

    do: Analytics.capture("chat_message_queued", distinct_id, properties)

  @doc "An assistant message that reached the user."

  @spec message_received(Analytics.distinct_id(), map()) :: :ok

  def message_received(distinct_id, properties \\ %{}),

    do: Analytics.capture("chat_message_received", distinct_id, properties)

  @doc """

  The provider-reported token counts for one turn.

  Accepts either naming the two OpenRouter APIs use (`input_tokens` and

  `output_tokens`, or `prompt_tokens` and `completion_tokens`). A usage map

  without any readable count captures nothing, because a missing count is not

  a zero.

  """

  @spec tokens_used(Analytics.distinct_id(), map() | nil, map()) :: :ok

  def tokens_used(distinct_id, usage, properties \\ %{})

  def tokens_used(distinct_id, usage, properties) when is_map(usage) do

    case token_counts(usage) do

      nil -> :ok

      counts -> Analytics.capture("chat_tokens_used", distinct_id, Map.merge(properties, counts))

    end

  end

  def tokens_used(_distinct_id, _usage, _properties), do: :ok

  @doc "A turn that ended anywhere other than completed."

  @spec turn_failed(Analytics.distinct_id(), map()) :: :ok

  def turn_failed(distinct_id, properties \\ %{}),

    do: Analytics.capture("chat_turn_failed", distinct_id, properties)

  @doc """

  One assistant stream chunk, rate-limited to one event per second.

  Callers hold the throttle: pass the value the previous call returned and keep

  the returned value. `nil` means nothing has been captured for this turn yet.

  """

  @spec stream_chunk(Analytics.distinct_id(), integer() | nil, map()) :: integer() | nil

  def stream_chunk(distinct_id, last_captured_at, properties \\ %{}) do

    now = System.monotonic_time(:millisecond)

    if throttled?(last_captured_at, now) do

      last_captured_at

    else

      Analytics.capture("chat_stream_chunk", distinct_id, properties)

      now

    end

  end

  @doc "A tool the assistant invoked during a turn."

  @spec tool_called(Analytics.distinct_id(), map()) :: :ok

  def tool_called(distinct_id, properties \\ %{}),

    do: Analytics.capture("chat_tool_called", distinct_id, properties)

  @doc "A voice session that began accepting audio."

  @spec voice_started(Analytics.distinct_id(), map()) :: :ok

  def voice_started(distinct_id, properties \\ %{}),

    do: Analytics.capture("chat_voice_started", distinct_id, properties)

  @doc "A voice session that stopped, with its duration when both ends are known."

  @spec voice_ended(Analytics.distinct_id(), map()) :: :ok

  def voice_ended(distinct_id, properties \\ %{}),

    do: Analytics.capture("chat_voice_ended", distinct_id, properties)

  @doc """

  The coarse size bucket for one message.

  Message text never enters an event property, so the bucket is what a reader

  gets: it separates a one-line ask from a pasted document without recording

  either.

  """

  @spec length_bucket(String.t()) :: String.t()

  def length_bucket(content) when is_binary(content) do

    cond do

      byte_size(content) < 100 -> "under_100"

      byte_size(content) < 1_000 -> "under_1k"

      byte_size(content) < 8_000 -> "under_8k"

      true -> "over_8k"

    end

  end

  defp throttled?(nil, _now), do: false

  defp throttled?(last_captured_at, now) when is_integer(last_captured_at),

    do: now - last_captured_at < @stream_chunk_interval_ms

  defp throttled?(_last_captured_at, _now), do: false

  defp token_counts(usage) do

    counts =

      %{

        "input_tokens" => count(usage, ["input_tokens", "prompt_tokens"]),

        "output_tokens" => count(usage, ["output_tokens", "completion_tokens"]),

        "total_tokens" => count(usage, ["total_tokens"])

      }

      |> Enum.reject(fn {_key, value} -> is_nil(value) end)

      |> Map.new()

    if counts == %{}, do: nil, else: counts

  end

  defp count(usage, keys) do

    Enum.find_value(keys, fn key ->

      case Map.get(usage, key) do

        value when is_integer(value) and value >= 0 -> value

        _other -> nil

      end

    end)

  end

end

  import Ecto.Query

  alias OpenAgents.Accounts.User

  alias OpenAgents.Analytics

  alias OpenAgents.Analytics.Chat, as: ChatAnalytics

  alias OpenAgents.Chat.{AccountEvent, AccountRun, OpenRouter}

  alias OpenAgents.Conversations

  alias OpenAgents.Repo

    else

      {:provider_start_failed, run, reason} ->

        finish_run(run.id, {:error, :turn_start_failed})

        capture_run_outcome(run, user, {:error, :turn_start_failed})

        {:error, reason}

      error ->

    with %{id: conversation_id} <- Conversations.get_conversation_for_user(user),

         %AccountRun{} = run <- streaming_run(conversation_id) do

      stop_provider_task(run.id)

      ChatAnalytics.turn_failed(Analytics.distinct_id(user), %{

        "reason" => "cancelled",

        "outcome" => "cancelled",

        "conversation_id" => conversation_id,

        "turn_id" => run.id

      })

      cancel_run(run.id)

    else

      _no_active_turn -> {:error, :no_active_turn}

                 request,

                 fn event ->

                   persist_provider_event(run.id, event)

                   capture_stream_event(run, user, event)

                   notify(subscriber, {:openrouter_stream_event, run.id, event})

                 end,

                 tool_context: %{

             end

           finish_run(run.id, result)

           capture_run_outcome(run, user, result)

           notify(subscriber, {:account_chat_completed, run.id, result})

         end) do

      {:ok, pid} -> {:ok, pid}

  defp persist_provider_event(run_id, {kind, payload}),

    do: append_event(run_id, Atom.to_string(kind), normalize_payload(payload))

  # One process owns one run's stream, so the chunk throttle lives in that

  # process rather than in a shared counter. Tool starts are already one event

  # per call and need no throttle.

  defp capture_stream_event(run, user, {:text_delta, _delta}) do

    captured_at =

      ChatAnalytics.stream_chunk(

        Analytics.distinct_id(user),

        Process.get(:chat_stream_chunk_captured_at),

        %{"conversation_id" => run.conversation_id, "turn_id" => run.id, "modality" => "text"}

      )

    Process.put(:chat_stream_chunk_captured_at, captured_at)

    :ok

  end

  defp capture_stream_event(run, user, {:tool_call_started, payload}) when is_map(payload) do

    ChatAnalytics.tool_called(Analytics.distinct_id(user), %{

      "tool_name" => payload["name"] || "tool",

      "turn_id" => run.id,

      "conversation_id" => run.conversation_id,

      "modality" => "text"

    })

  end

  defp capture_stream_event(_run, _user, _event), do: :ok

  # The completion carries the provider's own totals for the whole run,

  # including any Chat Completions fallback, so this is the one place a run

  # reports tokens.

  defp capture_run_outcome(run, user, {:ok, completion}) when is_map(completion) do

    distinct_id = Analytics.distinct_id(user)

    identity = %{

      "conversation_id" => run.conversation_id,

      "turn_id" => run.id,

      "modality" => "text"

    }

    ChatAnalytics.message_received(

      distinct_id,

      Map.put(

        identity,

        "length_bucket",

        ChatAnalytics.length_bucket(completion["assistant_content"] || "")

      )

    )

    ChatAnalytics.tokens_used(

      distinct_id,

      completion["usage"],

      Map.merge(identity, %{

        "model" => completion["model"],

        "provider" => completion["provider"],

        "outcome" => "completed"

      })

    )

  end

  defp capture_run_outcome(run, user, {:error, reason}) do

    ChatAnalytics.turn_failed(Analytics.distinct_id(user), %{

      "reason" => error_code(reason),

      "outcome" => "failed",

      "conversation_id" => run.conversation_id,

      "turn_id" => run.id

    })

  end

  defp capture_run_outcome(_run, _user, _result), do: :ok

  defp finish_run(run_id, {:ok, completion}) do

    # Token counts are read before redaction, which blanks every field whose

    # name contains `token`, and are stored beside the redacted completion.

  @doc """

  Everything the operator analytics page shows.

  Returns `{:ok, shaped}` with six bounded projections, or

  Returns `{:ok, shaped}` with eight bounded projections, or

  `{:error, :not_configured | :unavailable}`. Each projection runs as its own

  HogQL query; a failure of any one fails the whole pull, because partial

  numbers presented next to each other read as complete.

      with {:ok, events} <- run(event_counts_sql(), "event_counts"),

           {:ok, funnel} <- run(funnel_sql(), "funnel"),

           {:ok, chat} <- run(chat_turns_sql(), "chat_turns"),

           {:ok, lifecycle} <- run(chat_lifecycle_sql(), "chat_lifecycle"),

           {:ok, tokens} <- run(chat_token_usage_sql(), "chat_token_usage"),

           {:ok, pages} <- run(top_pages_sql(), "top_pages"),

           {:ok, triage} <- run(triage_health_sql(), "triage_health"),

           {:ok, issue_flow} <- run(weekly_issue_flow_sql(), "weekly_issue_flow") do

           event_counts: shape_rows(events),

           funnel: shape_rows(funnel) |> List.first(%{}),

           chat_turns: shape_chat_turns(shape_rows(chat) |> List.first(%{})),

           chat_lifecycle: shape_chat_lifecycle(shape_rows(lifecycle) |> List.first(%{})),

           chat_token_usage: shape_chat_token_usage(shape_rows(tokens)),

           top_pages: shape_rows(pages),

           triage_health: shape_triage_health(shape_rows(triage) |> List.first(%{})),

           weekly_issue_flow: shape_rows(issue_flow)

    |> squash()

  end

  # Assistant deliveries and turn failures in one row, so a failure rate reads

  # against the turns that actually ran rather than against a separate pull.

  defp chat_lifecycle_sql do

    """

    SELECT

      countIf(event = 'chat_message_received') AS messages_received,

      countIf(event = 'chat_message_queued') AS messages_queued,

      countIf(event = 'chat_turn_failed') AS turns_failed,

      countIf(event = 'chat_turn_completed') AS turns_finished,

      if(

        turns_finished = 0,

        0,

        round(turns_failed * 100.0 / turns_finished, 1)

      ) AS turn_failure_percent

    FROM events

    WHERE timestamp >= now() - INTERVAL 1 DAY

      AND event IN (

        'chat_message_received',

        'chat_message_queued',

        'chat_turn_failed',

        'chat_turn_completed'

      )

    """

    |> squash()

  end

  # Token totals per model and provider. `chat_tokens_used` is captured once per

  # turn, so summing it here counts each turn's tokens once.

  defp chat_token_usage_sql do

    """

    SELECT

      properties.model AS model,

      properties.provider AS provider,

      count() AS turns,

      sum(properties.input_tokens) AS input_tokens,

      sum(properties.output_tokens) AS output_tokens,

      sum(properties.input_tokens) + sum(properties.output_tokens) AS total_tokens

    FROM events

    WHERE timestamp >= now() - INTERVAL 1 DAY AND event = 'chat_tokens_used'

    GROUP BY model, provider

    ORDER BY total_tokens DESC

    LIMIT 10

    """

    |> squash()

  end

  defp top_pages_sql do

    """

    SELECT properties.$current_url AS url, count() AS views

    }

  end

  defp shape_chat_lifecycle(row) do

    %{

      "messages_received" => count_value(row["messages_received"]),

      "messages_queued" => count_value(row["messages_queued"]),

      "turns_failed" => count_value(row["turns_failed"]),

      "turns_finished" => count_value(row["turns_finished"]),

      "turn_failure_percent" => numeric_value(row["turn_failure_percent"])

    }

  end

  defp shape_chat_token_usage(rows) do

    Enum.map(rows, fn row ->

      %{

        "model" => row["model"] || "unknown",

        "provider" => row["provider"] || "unknown",

        "turns" => count_value(row["turns"]),

        "input_tokens" => count_value(row["input_tokens"]),

        "output_tokens" => count_value(row["output_tokens"]),

        "total_tokens" => count_value(row["total_tokens"])

      }

    end)

  end

  defp shape_triage_health(row) do

    %{

      "median_first_maintainer_response_hours" =>

  use GenServer, restart: :temporary

  alias OpenAgents.{

    Analytics,

    Blueprint,

    Context.Composer,

    Conversations,

    ShadowPrograms

  }

  alias OpenAgents.Analytics.Chat, as: ChatAnalytics

  alias OpenAgents.Providers.{ProviderEvent, Request, ToolOutput}

  alias OpenAgents.Tools.{ConversationExecutionContext, Registry, Runner}

    :atomics.put(state.cancellation, 1, 1)

    cancel_task(state.task)

    _timer_result = Process.cancel_timer(state.timeout_reference)

    capture_tokens_used(state, total_usage(state), "cancelled")

    result = Conversations.cancel_turn(state.turn, total_usage(state))

    {:stop, :normal, result, state}

  end

             state.receipt,

             Map.put(attributes, :routing_receipt_id, routing_receipt.id)

           ) do

      capture_tool_called(state, call)

      {:noreply,

       %{

         state

    end

  end

  # Token totals are reported from the turn's terminal path, which every turn

  # reaches exactly once. Tool rounds and report continuations have already been

  # merged into the total by then, so a turn that called several tools still

  # reports one set of counts.

  defp capture_tokens_used(state, usage, outcome) do

    ChatAnalytics.tokens_used(owner_distinct_id(state.owner), usage, %{

      "model" => state.base_request.model_id,

      "provider" => state.provider.id(),

      "conversation_id" => state.turn.conversation_id,

      "turn_id" => state.turn.id,

      "outcome" => outcome,

      "modality" => "text"

    })

  end

  defp capture_tool_called(state, call) do

    ChatAnalytics.tool_called(owner_distinct_id(state.owner), %{

      "tool_name" => call.name,

      "turn_id" => state.turn.id,

      "conversation_id" => state.turn.conversation_id,

      "modality" => "text"

    })

  end

  defp owner_distinct_id(%{user_id: user_id}) when is_binary(user_id),

    do: Analytics.distinct_id(user_id)

  defp owner_distinct_id(%{id: id}) when is_binary(id),

    do: Analytics.distinct_id("visitor_#{id}")

  defp route_tool_call(nil, state) do

    Router.route(state.tool_snapshot, state.routing_policy, %{

      intent_digest: state.receipt.input_digest,

    do: stop_failed(state, reason)

  defp finalize_provider_result(:ok, %{terminal_event: :cancelled} = state) do

    capture_tokens_used(state, total_usage(state), "cancelled")

    _cancel_result = Conversations.cancel_turn(state.turn, total_usage(state))

    {:stop, :normal, state}

  end

  defp finalize_text_response(response_id, state) do

    case Conversations.complete_turn(state.turn, response_id, state.usage) do

      {:ok, _turn} ->

        capture_tokens_used(state, state.usage, "completed")

        _timer_result = Process.cancel_timer(state.timeout_reference)

        {:stop, :normal, state}

    :atomics.put(state.cancellation, 1, 1)

    cancel_task(state.task)

    _timer_result = Process.cancel_timer(state.timeout_reference)

    capture_tokens_used(state, total_usage(state), "failed")

    _failure_result = Conversations.fail_turn(state.turn, reason, total_usage(state))

    _incident = report_incident(state, reason)

    {:stop, :normal, state}

              </.card>

            </section>

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

              <.card id="analytics-chat-lifecycle">

                <div class="space-y-1">

                  <h2 id="lifecycle-heading" class="card-title">Chat lifecycle and tokens</h2>

                  <p class="text-muted-foreground">

                    Assistant deliveries, queued messages, and turn failures over the trailing

                    twenty-four hours. Token totals are provider-reported and counted once per turn.

                  </p>

                </div>

                <% lifecycle = @overview.chat_lifecycle %>

                <div class="grid gap-4 py-5 md:grid-cols-3">

                  <div class="rounded-md border border-border p-4">

                    <p class="text-sm text-muted-foreground">Assistant messages received</p>

                    <p id="lifecycle-messages-received" class="mt-2 text-2xl font-semibold">

                      {lifecycle["messages_received"]}

                    </p>

                    <p class="mt-1 text-sm text-muted-foreground">

                      {lifecycle["messages_queued"]} messages waited behind an active turn.

                    </p>

                  </div>

                  <div class="rounded-md border border-border p-4">

                    <p class="text-sm text-muted-foreground">Turn failure rate</p>

                    <p id="lifecycle-failure-rate" class="mt-2 text-2xl font-semibold">

                      {format_percent(lifecycle["turn_failure_percent"])}

                    </p>

                    <p class="mt-1 text-sm text-muted-foreground">

                      {lifecycle["turns_failed"]} of {lifecycle["turns_finished"]} finished turns

                      ended in failure or cancellation.

                    </p>

                  </div>

                  <div class="rounded-md border border-border p-4">

                    <p class="text-sm text-muted-foreground">Tokens used</p>

                    <p id="lifecycle-total-tokens" class="mt-2 text-2xl font-semibold">

                      {total_tokens(@overview.chat_token_usage)}

                    </p>

                    <p class="mt-1 text-sm text-muted-foreground">

                      Input plus output across every model.

                    </p>

                  </div>

                </div>

                <.table id="analytics-chat-tokens-table" rows={@overview.chat_token_usage}>

                  <:col :let={row} label="Model">{row["model"]}</:col>

                  <:col :let={row} label="Provider">{row["provider"]}</:col>

                  <:col :let={row} label="Turns">{row["turns"]}</:col>

                  <:col :let={row} label="Input">{row["input_tokens"]}</:col>

                  <:col :let={row} label="Output">{row["output_tokens"]}</:col>

                  <:col :let={row} label="Total">{row["total_tokens"]}</:col>

                </.table>

              </.card>

            </section>

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

              <.card id="analytics-event-volume">

                <h2 id="events-heading" class="card-title">Event volume</h2>

  defp format_duration(ms) when is_integer(ms), do: "#{ms}ms"

  defp total_tokens(rows) when is_list(rows),

    do: Enum.reduce(rows, 0, fn row, total -> total + (row["total_tokens"] || 0) end)

  defp format_hours(nil), do: "No data"

  defp format_hours(hours) when is_integer(hours), do: "#{hours}h"

    VoiceSessions

  }

  alias OpenAgents.Analytics.Chat, as: ChatAnalytics

  alias OpenAgents.ComputerActivity

  alias OpenAgents.Conversations.Message

  alias OpenAgents.Voice.Config, as: VoiceConfig

      |> assign(:oldest_message_id, first_id(messages))

      |> assign(:active_turn, active_turn)

      |> assign(:message_queue, [])

      |> assign(:stream_chunk_captured_at, nil)

      |> assign(:voice_enabled?, voice_config.enabled?)

      |> assign(:recording_config, Recordings.config())

      |> assign(:voice_session, voice_session)

  def handle_info({:message_updated, message}, socket) do

    {:noreply,

     socket

     |> capture_assistant_message(message)

     |> clear_live_voice_item(message.provider_item_id)

     |> refresh_job_rollup(message)

     |> stream_insert(:messages, message)}

  def handle_info({:turn_updated, turn}, socket) do

    if turn.status in ["completed", "failed", "cancelled"] do

      capture_turn_completed(turn, socket)

      capture_turn_failed(turn, socket)

      # The active turn ended: clear it, surface any error, and immediately start

      # the next queued message so a stacked run continues without the owner

      |> assign(:active_turn, nil)

      |> assign(:tool_activity, [])

      |> assign(:composer_error, turn.error_message)

      |> assign(:stream_chunk_captured_at, nil)

      |> push_event("composer:focus", %{})

      |> advance_queue()

    else

        do: clear_all_live_voice_items(socket),

        else: socket

    capture_voice_lifecycle(socket, voice_session)

    {:noreply,

     socket

     |> assign(:voice_session, voice_session)

      true ->

        item = %{id: System.unique_integer([:positive, :monotonic]), content: trimmed}

        ChatAnalytics.message_queued(

          Analytics.distinct_id(socket.assigns.current_user),

          %{

            "length_bucket" => length_bucket(trimmed),

            "queue_depth" => length(socket.assigns.message_queue) + 1,

            "conversation_id" => socket.assigns.conversation.id

          }

        )

        {:noreply,

         socket

         |> assign(:message_queue, socket.assigns.message_queue ++ [item])

    )

  end

  defp length_bucket(content) when is_binary(content) do

  # A failed or cancelled turn is reported beside the completion event so a

  # failure rate can be read from one event instead of a property filter.

  defp capture_turn_failed(%{status: "completed"}, _socket), do: :ok

  defp capture_turn_failed(turn, socket) do

    ChatAnalytics.turn_failed(

      Analytics.distinct_id(socket.assigns.current_user),

      %{

        "reason" => turn.error_code || turn.status,

        "outcome" => turn.status,

        "conversation_id" => turn.conversation_id,

        "turn_id" => turn.id

      }

    )

  end

  # Every assistant delta re-broadcasts the message, which makes this both the

  # stream-chunk signal and, at the terminal status, the one place an assistant

  # message is known to have reached the reader. Chunks are throttled inside

  # `OpenAgents.Analytics.Chat`; the throttle rides in an assign so it resets

  # with each turn.

  defp capture_assistant_message(

         socket,

         %Message{role: "assistant", status: "streaming"} = message

       ) do

    captured_at =

      ChatAnalytics.stream_chunk(

        Analytics.distinct_id(socket.assigns.current_user),

        socket.assigns.stream_chunk_captured_at,

        %{

          "conversation_id" => message.conversation_id,

          "modality" => message.modality

        }

      )

    assign(socket, :stream_chunk_captured_at, captured_at)

  end

  defp capture_assistant_message(

         socket,

         %Message{role: "assistant", status: "complete"} = message

       ) do

    ChatAnalytics.message_received(

      Analytics.distinct_id(socket.assigns.current_user),

      %{

        "length_bucket" => length_bucket(message.content || ""),

        "modality" => message.modality,

        "conversation_id" => message.conversation_id

      }

    )

    socket

  end

  defp capture_assistant_message(socket, _message), do: socket

  # Status broadcasts repeat through a call's life, so the transitions are read

  # against the session already in the assign: absent to live starts a call,

  # live to terminal ends one.

  defp capture_voice_lifecycle(socket, voice_session) do

    previous = socket.assigns.voice_session

    distinct_id = Analytics.distinct_id(socket.assigns.current_user)

    terminal? = voice_session.status in ~w(ended failed)

    cond do

      byte_size(content) < 100 -> "under_100"

      byte_size(content) < 1_000 -> "under_1k"

      byte_size(content) < 8_000 -> "under_8k"

      true -> "over_8k"

      is_nil(previous) and not terminal? ->

        ChatAnalytics.voice_started(distinct_id, %{

          "conversation_id" => voice_session.conversation_id

        })

      not is_nil(previous) and previous.status not in ~w(ended failed) and terminal? ->

        ChatAnalytics.voice_ended(distinct_id, %{

          "conversation_id" => voice_session.conversation_id,

          "outcome" => voice_session.status,

          "duration_ms" => voice_duration_ms(voice_session)

        })

      true ->

        :ok

    end

  end

  defp voice_duration_ms(%{

         started_at: %DateTime{} = started_at,

         ended_at: %DateTime{} = ended_at

       }),

       do: DateTime.diff(ended_at, started_at, :millisecond)

  defp voice_duration_ms(_voice_session), do: nil

  defp length_bucket(content), do: ChatAnalytics.length_bucket(content)

  defp tool_activity(nil, nil), do: []

  defp tool_activity(turn, _voice_session) when not is_nil(turn),

defmodule OpenAgents.Analytics.ChatTest do

  @moduledoc """

  The chat lifecycle vocabulary is the only place chat instrumentation names an

  event or normalizes a token count, so these assertions pin the two decisions

  that operator dashboards depend on: which counts survive normalization, and

  that a stream reports at most one chunk event per second.

  """

  use ExUnit.Case, async: false

  alias OpenAgents.Analytics.Chat, as: ChatAnalytics

  defmodule TestSink do

    def capture(event, distinct_id, properties) do

      send(:chat_analytics_test_process, {:captured, event, distinct_id, properties})

      :ok

    end

  end

  setup do

    Process.register(self(), :chat_analytics_test_process)

    original_token = Application.get_env(:openagents, :posthog_project_token)

    original_sink = Application.get_env(:openagents, :analytics_sink)

    Application.put_env(:openagents, :posthog_project_token, "phc_test_token")

    Application.put_env(:openagents, :analytics_sink, TestSink)

    on_exit(fn ->

      restore_env(:posthog_project_token, original_token)

      restore_env(:analytics_sink, original_sink)

    end)

    :ok

  end

  defp restore_env(key, nil), do: Application.delete_env(:openagents, key)

  defp restore_env(key, value), do: Application.put_env(:openagents, key, value)

  defp captured do

    assert_receive {:captured, event, distinct_id, properties}, 500

    %{event: event, distinct_id: distinct_id, properties: properties}

  end

  describe "tokens_used/3" do

    test "provider token counts reach PostHog with the turn's identity" do

      :ok =

        ChatAnalytics.tokens_used(

          "user_abc",

          %{"input_tokens" => 1_200, "output_tokens" => 300, "total_tokens" => 1_500},

          %{"model" => "anthropic/claude-sonnet-4.5", "provider" => "openrouter"}

        )

      captured = captured()

      assert captured.event == "chat_tokens_used"

      assert captured.distinct_id == "user_abc"

      assert captured.properties["input_tokens"] == 1_200

      assert captured.properties["output_tokens"] == 300

      assert captured.properties["total_tokens"] == 1_500

      assert captured.properties["model"] == "anthropic/claude-sonnet-4.5"

      assert captured.properties["provider"] == "openrouter"

    end

    test "the Chat Completions spelling of the same counts normalizes" do

      :ok =

        ChatAnalytics.tokens_used("user_abc", %{

          "prompt_tokens" => 40,

          "completion_tokens" => 9

        })

      properties = captured().properties

      assert properties["input_tokens"] == 40

      assert properties["output_tokens"] == 9

    end

    test "a turn the provider never counted reports nothing" do

      assert :ok == ChatAnalytics.tokens_used("user_abc", nil)

      assert :ok == ChatAnalytics.tokens_used("user_abc", %{})

      assert :ok == ChatAnalytics.tokens_used("user_abc", %{"input_tokens" => "many"})

      refute_receive {:captured, _event, _distinct_id, _properties}, 100

    end

  end

  describe "stream_chunk/3" do

    test "the first chunk reports and the next chunk within the window does not" do

      captured_at = ChatAnalytics.stream_chunk("user_abc", nil, %{"modality" => "text"})

      assert is_integer(captured_at)

      assert captured().event == "chat_stream_chunk"

      assert captured_at == ChatAnalytics.stream_chunk("user_abc", captured_at, %{})

      refute_receive {:captured, _event, _distinct_id, _properties}, 100

    end

    test "a chunk past the window reports again" do

      stale = System.monotonic_time(:millisecond) - 2_000

      captured_at = ChatAnalytics.stream_chunk("user_abc", stale, %{})

      assert captured_at > stale

      assert captured().event == "chat_stream_chunk"

    end

  end

  describe "the remaining lifecycle events" do

    test "each helper names its own event" do

      :ok = ChatAnalytics.message_queued("user_abc", %{"queue_depth" => 2})

      assert captured().event == "chat_message_queued"

      :ok = ChatAnalytics.message_received("user_abc")

      assert captured().event == "chat_message_received"

      :ok = ChatAnalytics.turn_failed("user_abc", %{"reason" => "provider_unavailable"})

      assert captured().event == "chat_turn_failed"

      :ok = ChatAnalytics.tool_called("user_abc", %{"tool_name" => "read_file"})

      assert captured().event == "chat_tool_called"

      :ok = ChatAnalytics.voice_started("user_abc")

      assert captured().event == "chat_voice_started"

      :ok = ChatAnalytics.voice_ended("user_abc", %{"duration_ms" => 4_000})

      assert captured().event == "chat_voice_ended"

    end

  end

  describe "length_bucket/1" do

    test "message size reads as a bucket rather than as content" do

      assert ChatAnalytics.length_bucket("hello") == "under_100"

      assert ChatAnalytics.length_bucket(String.duplicate("a", 500)) == "under_1k"

      assert ChatAnalytics.length_bucket(String.duplicate("a", 5_000)) == "under_8k"

      assert ChatAnalytics.length_bucket(String.duplicate("a", 9_000)) == "over_8k"

    end

  end

end

      assert {:error, :not_configured} = PostHog.overview()

    end

    test "shapes the six projections from one pull" do

    test "shapes the eight projections from one pull" do

      configure()

      Req.Test.expect(__MODULE__, fn conn ->

        })

      end)

      Req.Test.expect(__MODULE__, fn conn ->

        {:ok, body, conn} = Plug.Conn.read_body(conn)

        assert body =~ "'chat_message_received'"

        assert body =~ "turn_failure_percent"

        Req.Test.json(conn, %{

          "columns" => [

            "messages_received",

            "messages_queued",

            "turns_failed",

            "turns_finished",

            "turn_failure_percent"

          ],

          "results" => [[24, 2, 1, 24, 4.2]]

        })

      end)

      Req.Test.expect(__MODULE__, fn conn ->

        {:ok, body, conn} = Plug.Conn.read_body(conn)

        assert body =~ "'chat_tokens_used'"

        assert body =~ "GROUP BY model, provider"

        Req.Test.json(conn, %{

          "columns" => [

            "model",

            "provider",

            "turns",

            "input_tokens",

            "output_tokens",

            "total_tokens"

          ],

          "results" => [

            ["anthropic/claude-sonnet-4.5", "anthropic", 6, 4200, 1800, 6000],

            [nil, nil, 1, 10, 5.0, 15.0]

          ]

        })

      end)

      Req.Test.expect(__MODULE__, fn conn ->

        {:ok, body, conn} = Plug.Conn.read_body(conn)

        assert body =~ "'$pageview'"

               "max_duration_ms" => 7414

             }

      assert overview.chat_lifecycle == %{

               "messages_received" => 24,

               "messages_queued" => 2,

               "turns_failed" => 1,

               "turns_finished" => 24,

               "turn_failure_percent" => 4.2

             }

      assert overview.chat_token_usage == [

               %{

                 "model" => "anthropic/claude-sonnet-4.5",

                 "provider" => "anthropic",

                 "turns" => 6,

                 "input_tokens" => 4200,

                 "output_tokens" => 1800,

                 "total_tokens" => 6000

               },

               %{

                 "model" => "unknown",

                 "provider" => "unknown",

                 "turns" => 1,

                 "input_tokens" => 10,

                 "output_tokens" => 5,

                 "total_tokens" => 15

               }

             ]

      assert [%{"url" => "https://openagents.com/", "views" => 20} | _] = overview.top_pages

      assert overview.triage_health == %{

        request_options: [plug: {Req.Test, __MODULE__}]

      )

      # One pull is six questions.

      Req.Test.expect(__MODULE__, 6, fn conn -> respond_by_query(conn) end)

      # One pull is eight questions.

      Req.Test.expect(__MODULE__, 8, fn conn -> respond_by_query(conn) end)

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

      assert has_element?(view, "#analytics-weekly-issue-flow")

      assert has_element?(view, "#analytics-funnel")

      assert has_element?(view, "#analytics-chat-turns")

      assert has_element?(view, "#analytics-chat-lifecycle")

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

      assert html = render(view)

      assert html =~ "anthropic/claude-sonnet-4.5"

      assert html =~ "4.2%"

      assert html =~ "12.5h"

      assert html =~ "15.0%"

      assert html =~ "$pageview"

      assert html =~ "https://openagents.com/"

      # A second full pull backs the refresh click.

      Req.Test.expect(__MODULE__, 6, fn conn -> respond_by_query(conn) end)

      Req.Test.expect(__MODULE__, 8, fn conn -> respond_by_query(conn) end)

      assert view |> element("#analytics-refresh") |> render_click() =~ "LIVE POSTHOG"

      render_async(view)

    Req.Test.expect(__MODULE__, handler)

  end

  # The client asks six questions in a fixed order; each stub answers by

  # The client asks eight questions in a fixed order; each stub answers by

  # matching the HogQL in the request body rather than relying on call order.

  defp respond_by_query(conn) do

    {:ok, body, conn} = Plug.Conn.read_body(conn)

          "results" => [[3, 1, 2, 1, 6]]

        })

      body =~ "turn_failure_percent" ->

        Req.Test.json(conn, %{

          "columns" => [

            "messages_received",

            "messages_queued",

            "turns_failed",

            "turns_finished",

            "turn_failure_percent"