diff --git a/lib/towerops/llm/usage.ex b/lib/towerops/llm/usage.ex new file mode 100644 index 00000000..a10190c6 --- /dev/null +++ b/lib/towerops/llm/usage.ex @@ -0,0 +1,145 @@ +defmodule Towerops.LLM.Usage do + @moduledoc """ + Per-day, per-org, per-purpose LLM token-usage tracking. The schema + doubles as the public API: every consumer of `Towerops.LLM.complete/2` + should call `record/1` after a successful response so the + organization's daily spend stays queryable in one place. + + Today's only caller is `Towerops.Workers.InsightLlmEnrichmentWorker`. + Future callers — billing-anomaly summaries, monitoring alert + narratives, network-wide weekly reports — should pass their own + `:purpose` so the per-feature spend can be split out. + + Roll up monthly cost with: + + SELECT day, organization_id, model, purpose, + SUM(prompt_tokens), SUM(completion_tokens), SUM(request_count) + FROM llm_usage + WHERE day >= now() - interval '30 days' + GROUP BY 1, 2, 3, 4 + ORDER BY 1 DESC; + """ + use Ecto.Schema + + import Ecto.Changeset + import Ecto.Query + + alias Towerops.Repo + + @primary_key {:id, :binary_id, autogenerate: true} + @foreign_key_type :binary_id + + schema "llm_usage" do + field :day, :date + field :model, :string + field :purpose, :string + field :prompt_tokens, :integer, default: 0 + field :completion_tokens, :integer, default: 0 + field :request_count, :integer, default: 0 + + belongs_to :organization, Towerops.Organizations.Organization + + timestamps(type: :utc_datetime) + end + + @castable [ + :organization_id, + :day, + :model, + :purpose, + :prompt_tokens, + :completion_tokens, + :request_count + ] + + def changeset(usage, attrs) do + usage + |> cast(attrs, @castable) + |> validate_required([:day, :model, :purpose]) + |> validate_number(:prompt_tokens, greater_than_or_equal_to: 0) + |> validate_number(:completion_tokens, greater_than_or_equal_to: 0) + |> validate_number(:request_count, greater_than_or_equal_to: 1) + end + + @doc """ + Idempotently upserts one day's usage for the given + `(organization_id, day, model, purpose)` tuple. Each call is one LLM + request — the row's counters get incremented atomically. + + ## Required fields + - `:model` — string, e.g. `"deepseek-v4-pro"` + - `:purpose` — string, e.g. `"insight_enrichment"` + + ## Optional fields + - `:organization_id` — binary_id, nil for system-level analyses + - `:prompt_tokens`, `:completion_tokens` — non-negative integers, + default to 0 if the API didn't surface a usage block + - `:day` — Date, defaults to today (UTC) + """ + @spec record(map()) :: {:ok, %__MODULE__{}} | {:error, Ecto.Changeset.t()} + def record(attrs) when is_map(attrs) do + today = attrs[:day] || Date.utc_today() + prompt = attrs[:prompt_tokens] || 0 + completion = attrs[:completion_tokens] || 0 + + base_attrs = %{ + organization_id: attrs[:organization_id], + day: today, + model: attrs[:model], + purpose: attrs[:purpose], + prompt_tokens: prompt, + completion_tokens: completion, + request_count: 1 + } + + changeset = changeset(%__MODULE__{}, base_attrs) + + if changeset.valid? do + Repo.insert( + changeset, + on_conflict: [ + inc: [ + prompt_tokens: prompt, + completion_tokens: completion, + request_count: 1 + ] + ], + conflict_target: [:organization_id, :day, :model, :purpose] + ) + else + {:error, changeset} + end + end + + @doc """ + Returns rolled-up usage for an organization across the given window + (default last 30 days). Useful for the billing/admin dashboards. + """ + @spec summarize(Ecto.UUID.t() | nil, keyword()) :: [map()] + def summarize(organization_id, opts \\ []) do + days = Keyword.get(opts, :days, 30) + cutoff = Date.add(Date.utc_today(), -days) + + from(u in __MODULE__, + where: u.day >= ^cutoff, + group_by: [u.day, u.model, u.purpose], + order_by: [desc: u.day], + select: %{ + day: u.day, + model: u.model, + purpose: u.purpose, + prompt_tokens: sum(u.prompt_tokens), + completion_tokens: sum(u.completion_tokens), + request_count: sum(u.request_count) + } + ) + |> maybe_filter_org(organization_id) + |> Repo.all() + end + + defp maybe_filter_org(query, nil), do: query + + defp maybe_filter_org(query, org_id) do + where(query, [u], u.organization_id == ^org_id) + end +end diff --git a/lib/towerops/preseem/insight.ex b/lib/towerops/preseem/insight.ex index 613ec8b3..8cbd8eeb 100644 --- a/lib/towerops/preseem/insight.ex +++ b/lib/towerops/preseem/insight.ex @@ -30,8 +30,6 @@ defmodule Towerops.Preseem.Insight do field :recommended_action, :string field :llm_model, :string field :llm_enriched_at, :utc_datetime - field :llm_prompt_tokens, :integer - field :llm_completion_tokens, :integer belongs_to :organization, Towerops.Organizations.Organization belongs_to :preseem_access_point, Towerops.Preseem.AccessPoint @@ -62,9 +60,7 @@ defmodule Towerops.Preseem.Insight do :llm_summary, :recommended_action, :llm_model, - :llm_enriched_at, - :llm_prompt_tokens, - :llm_completion_tokens + :llm_enriched_at ]) |> validate_required([:organization_id, :type, :urgency, :channel, :title]) |> validate_inclusion(:type, @valid_types) diff --git a/lib/towerops/preseem/insights.ex b/lib/towerops/preseem/insights.ex index 00ec1468..821c080c 100644 --- a/lib/towerops/preseem/insights.ex +++ b/lib/towerops/preseem/insights.ex @@ -115,19 +115,20 @@ defmodule Towerops.Preseem.Insights do @doc """ Apply LLM-generated enrichment to an insight. Sets `llm_summary`, - `recommended_action`, `llm_model`, `llm_enriched_at` (now), and the - prompt/completion token counts when the LLM client surfaces them. + `recommended_action`, `llm_model`, and `llm_enriched_at` (now). + + Token-usage tracking is intentionally *not* on the insight row — + see `Towerops.LLM.Usage` for the dedicated per-day/org/purpose + rollup table that all LLM consumers should record to. """ - def apply_llm_enrichment(%Insight{} = insight, %{summary: summary} = parsed, model, usage \\ %{}) + def apply_llm_enrichment(%Insight{} = insight, %{summary: summary} = parsed, model) when is_binary(summary) and is_binary(model) do insight |> Insight.changeset(%{ llm_summary: summary, recommended_action: Map.get(parsed, :action), llm_model: model, - llm_enriched_at: Towerops.Time.now(), - llm_prompt_tokens: Map.get(usage, :prompt_tokens), - llm_completion_tokens: Map.get(usage, :completion_tokens) + llm_enriched_at: Towerops.Time.now() }) |> Repo.update() end diff --git a/lib/towerops/workers/insight_llm_enrichment_worker.ex b/lib/towerops/workers/insight_llm_enrichment_worker.ex index ba414822..f6202b88 100644 --- a/lib/towerops/workers/insight_llm_enrichment_worker.ex +++ b/lib/towerops/workers/insight_llm_enrichment_worker.ex @@ -12,6 +12,7 @@ defmodule Towerops.Workers.InsightLlmEnrichmentWorker do alias Towerops.LLM alias Towerops.LLM.InsightPrompt + alias Towerops.LLM.Usage alias Towerops.Preseem.Insights require Logger @@ -33,13 +34,20 @@ defmodule Towerops.Workers.InsightLlmEnrichmentWorker do case LLM.complete(messages, max_tokens: 400, temperature: 0.2) do {:ok, %{content: content, model: model} = response} -> {:ok, parsed} = InsightPrompt.parse(content) + {:ok, _} = Insights.apply_llm_enrichment(insight, parsed, model) - usage = %{ - prompt_tokens: Map.get(response, :prompt_tokens), - completion_tokens: Map.get(response, :completion_tokens) - } + # Token-usage tracking lives in its own table — every LLM + # consumer in the codebase records here so the per-org daily + # cost is queryable in one place. + Usage.record(%{ + organization_id: insight.organization_id, + model: model, + purpose: "insight_enrichment", + prompt_tokens: Map.get(response, :prompt_tokens) || 0, + completion_tokens: Map.get(response, :completion_tokens) || 0 + }) - Insights.apply_llm_enrichment(insight, parsed, model, usage) + :ok {:error, reason} -> Logger.warning("LLM enrichment failed for insight #{insight.id}: #{inspect(reason)}") diff --git a/priv/repo/migrations/20260510170000_create_llm_usage.exs b/priv/repo/migrations/20260510170000_create_llm_usage.exs new file mode 100644 index 00000000..bbbda571 --- /dev/null +++ b/priv/repo/migrations/20260510170000_create_llm_usage.exs @@ -0,0 +1,41 @@ +defmodule Towerops.Repo.Migrations.CreateLlmUsage do + use Ecto.Migration + + def change do + create table(:llm_usage, primary_key: false) do + add :id, :binary_id, primary_key: true + # Nullable: future system-level analysis ("which orgs are + # at-risk this week?") may not be scoped to a single org. + add :organization_id, + references(:organizations, type: :binary_id, on_delete: :delete_all) + + add :day, :date, null: false + add :model, :string, null: false + # "insight_enrichment", "alert_summary", "billing_anomaly", etc. + # Free-form so new callers don't need a schema change. + add :purpose, :string, null: false + + add :prompt_tokens, :integer, null: false, default: 0 + add :completion_tokens, :integer, null: false, default: 0 + add :request_count, :integer, null: false, default: 0 + + timestamps(type: :utc_datetime) + end + + # One row per (org, day, model, purpose) so callers can UPSERT and + # increment cheaply. + create unique_index(:llm_usage, [:organization_id, :day, :model, :purpose], + name: :llm_usage_org_day_model_purpose_idx, + nulls_distinct: false + ) + + create index(:llm_usage, [:day]) + create index(:llm_usage, [:organization_id, :day]) + + # Drop the per-insight token columns — superseded by the new table. + alter table(:preseem_insights) do + remove :llm_prompt_tokens, :integer + remove :llm_completion_tokens, :integer + end + end +end diff --git a/test/towerops/llm/usage_test.exs b/test/towerops/llm/usage_test.exs new file mode 100644 index 00000000..26875043 --- /dev/null +++ b/test/towerops/llm/usage_test.exs @@ -0,0 +1,132 @@ +defmodule Towerops.LLM.UsageTest do + use Towerops.DataCase, async: true + + import Towerops.AccountsFixtures + import Towerops.OrganizationsFixtures + + alias Towerops.LLM.Usage + + setup do + user = user_fixture() + org = organization_fixture(user.id) + %{org: org} + end + + describe "record/1" do + test "inserts a new row when no (org, day, model, purpose) exists yet", %{org: org} do + assert {:ok, row} = + Usage.record(%{ + organization_id: org.id, + model: "deepseek-v4-pro", + purpose: "insight_enrichment", + prompt_tokens: 100, + completion_tokens: 25 + }) + + assert row.prompt_tokens == 100 + assert row.completion_tokens == 25 + assert row.request_count == 1 + assert row.day == Date.utc_today() + end + + test "upserts and increments counters on subsequent calls", %{org: org} do + attrs = %{ + organization_id: org.id, + model: "deepseek-v4-pro", + purpose: "insight_enrichment", + prompt_tokens: 50, + completion_tokens: 10 + } + + {:ok, _} = Usage.record(attrs) + {:ok, _} = Usage.record(attrs) + {:ok, _} = Usage.record(attrs) + + [row] = Repo.all(Usage) + assert row.prompt_tokens == 150 + assert row.completion_tokens == 30 + assert row.request_count == 3 + end + + test "splits per purpose so feature-level cost is queryable", %{org: org} do + base = %{organization_id: org.id, model: "deepseek-v4-pro", prompt_tokens: 10, completion_tokens: 5} + + {:ok, _} = Usage.record(Map.put(base, :purpose, "insight_enrichment")) + {:ok, _} = Usage.record(Map.put(base, :purpose, "billing_anomaly")) + + assert Repo.aggregate(Usage, :count, :id) == 2 + end + + test "splits per model so multi-LLM cost is queryable", %{org: org} do + base = %{organization_id: org.id, purpose: "insight_enrichment", prompt_tokens: 10, completion_tokens: 5} + + {:ok, _} = Usage.record(Map.put(base, :model, "deepseek-v4-pro")) + {:ok, _} = Usage.record(Map.put(base, :model, "gpt-5")) + + assert Repo.aggregate(Usage, :count, :id) == 2 + end + + test "accepts a nil organization_id (system-level analyses)" do + assert {:ok, row} = + Usage.record(%{ + organization_id: nil, + model: "deepseek-v4-pro", + purpose: "fleet_summary", + prompt_tokens: 1000, + completion_tokens: 250 + }) + + assert is_nil(row.organization_id) + end + + test "defaults missing token fields to 0", %{org: org} do + assert {:ok, row} = + Usage.record(%{ + organization_id: org.id, + model: "deepseek-v4-pro", + purpose: "insight_enrichment" + }) + + assert row.prompt_tokens == 0 + assert row.completion_tokens == 0 + assert row.request_count == 1 + end + + test "rejects invalid changesets" do + assert {:error, %Ecto.Changeset{}} = + Usage.record(%{model: nil, purpose: nil}) + end + end + + describe "summarize/2" do + test "rolls up usage for one organization across the configured window", %{org: org} do + {:ok, _} = + Usage.record(%{ + organization_id: org.id, + model: "deepseek-v4-pro", + purpose: "insight_enrichment", + prompt_tokens: 100, + completion_tokens: 25 + }) + + {:ok, _} = + Usage.record(%{ + organization_id: org.id, + model: "deepseek-v4-pro", + purpose: "insight_enrichment", + prompt_tokens: 50, + completion_tokens: 10 + }) + + [summary] = Usage.summarize(org.id, days: 7) + assert summary.day == Date.utc_today() + assert summary.prompt_tokens == 150 + assert summary.completion_tokens == 35 + assert summary.request_count == 2 + end + + test "returns nothing for an org with no usage", %{org: org} do + assert Usage.summarize(org.id) == [] + end + end +end diff --git a/test/towerops/workers/insight_llm_enrichment_worker_test.exs b/test/towerops/workers/insight_llm_enrichment_worker_test.exs index fcd525d7..ff1739f6 100644 --- a/test/towerops/workers/insight_llm_enrichment_worker_test.exs +++ b/test/towerops/workers/insight_llm_enrichment_worker_test.exs @@ -1,6 +1,7 @@ defmodule Towerops.Workers.InsightLlmEnrichmentWorkerTest do use Towerops.DataCase, async: false + alias Towerops.LLM.Usage alias Towerops.Preseem.AccessPoint alias Towerops.Preseem.Insight alias Towerops.Repo @@ -72,9 +73,9 @@ defmodule Towerops.Workers.InsightLlmEnrichmentWorkerTest do assert updated.llm_enriched_at end - test "perform/1 persists prompt + completion token counts when the LLM returns usage", + test "perform/1 records prompt + completion tokens to llm_usage (not on the insight)", %{organization: org, access_point: ap} do - insight = insert_active_insight(org, ap) + _insight = insert_active_insight(org, ap) Process.put( :llm_mock_response, @@ -90,9 +91,39 @@ defmodule Towerops.Workers.InsightLlmEnrichmentWorkerTest do assert :ok = InsightLlmEnrichmentWorker.perform(%Oban.Job{}) - updated = Repo.get!(Insight, insight.id) - assert updated.llm_prompt_tokens == 123 - assert updated.llm_completion_tokens == 45 + [usage] = Repo.all(Usage) + assert usage.organization_id == org.id + assert usage.model == "deepseek-v4-pro" + assert usage.purpose == "insight_enrichment" + assert usage.prompt_tokens == 123 + assert usage.completion_tokens == 45 + assert usage.request_count == 1 + assert usage.day == Date.utc_today() + end + + test "perform/1 increments the existing day's row when the worker runs twice", + %{organization: org, access_point: ap} do + insert_active_insight(org, ap, %{title: "first"}) + insert_active_insight(org, ap, %{title: "second"}) + + Process.put( + :llm_mock_response, + {:ok, + %{ + content: ~s|{"summary":"x","action":"y"}|, + model: "deepseek-v4-pro", + prompt_tokens: 100, + completion_tokens: 25 + }} + ) + + assert :ok = InsightLlmEnrichmentWorker.perform(%Oban.Job{}) + + [usage] = Repo.all(Usage) + # Both insights enriched — counters incremented atomically. + assert usage.prompt_tokens == 200 + assert usage.completion_tokens == 50 + assert usage.request_count == 2 end test "perform/1 leaves insight unenriched on LLM error and does not crash", %{organization: org, access_point: ap} do