diff --git a/lib/towerops/llm/network_snapshot.ex b/lib/towerops/llm/network_snapshot.ex index 0de2c180..b89d07d6 100644 --- a/lib/towerops/llm/network_snapshot.ex +++ b/lib/towerops/llm/network_snapshot.ex @@ -263,11 +263,29 @@ defmodule Towerops.LLM.NetworkSnapshot do %{high_utilization: high} end - defp build_gaiia(_organization_id) do - # The reconciliation worker has its own cache; rather than recompute - # here we surface only the *count* of currently active reconciliation - # findings (cheap query) and let the LLM see them as existing insights. - %{untracked: 0, ghosts: 0, mismatches: 0} + defp build_gaiia(organization_id) do + rows = + Insight + |> where([i], i.organization_id == ^organization_id) + |> where([i], i.type == "reconciliation_finding" and i.status == "active") + |> select([i], %{ + finding_type: fragment("?->>'finding_type'", i.metadata), + item_count: fragment("(?->>'count')::int", i.metadata) + }) + |> Repo.all() + + %{ + untracked: item_count_for(rows, "untracked"), + ghosts: item_count_for(rows, "ghost"), + mismatches: item_count_for(rows, "mismatch") + } + end + + defp item_count_for(rows, type) do + Enum.find_value(rows, 0, fn + %{finding_type: ^type, item_count: count} -> count + _ -> nil + end) end defp build_agents(organization_id) do diff --git a/lib/towerops/workers/ai_network_insight_worker.ex b/lib/towerops/workers/ai_network_insight_worker.ex index 5e431816..49e7c7e6 100644 --- a/lib/towerops/workers/ai_network_insight_worker.ex +++ b/lib/towerops/workers/ai_network_insight_worker.ex @@ -32,14 +32,19 @@ defmodule Towerops.Workers.AiNetworkInsightWorker do @impl Oban.Worker def perform(%Oban.Job{}) do - Repo.transaction(fn -> - Organization - |> select([o], o.id) - |> Repo.stream() - |> Enum.each(&process_organization/1) - end) + case Repo.transaction(fn -> + Organization + |> select([o], o.id) + |> Repo.stream() + |> Enum.each(&process_organization/1) + end) do + {:ok, _} -> + :ok - :ok + {:error, reason} -> + Logger.error("AI network insight transaction failed: #{inspect(reason)}") + {:error, reason} + end end defp process_organization(organization_id) do @@ -56,7 +61,7 @@ defmodule Towerops.Workers.AiNetworkInsightWorker do :ok {:error, reason} -> - Logger.warning("AI network insight failed for org #{organization_id}: #{inspect(reason)}") + Logger.error("AI network insight LLM call failed for org #{organization_id}: #{inspect(reason)}") :ok end rescue @@ -93,7 +98,7 @@ defmodule Towerops.Workers.AiNetworkInsightWorker do :ok {:error, changeset} -> - Logger.warning("Failed to persist ai_observation: #{inspect(changeset.errors)}") + Logger.error("Failed to persist ai_observation: #{inspect(changeset.errors)}") :ok end end @@ -106,15 +111,19 @@ defmodule Towerops.Workers.AiNetworkInsightWorker do end defp record_usage(organization_id, response, model) do - _ = - Usage.record(%{ - organization_id: organization_id, - model: model, - purpose: "network_insight", - prompt_tokens: Map.get(response, :prompt_tokens) || 0, - completion_tokens: Map.get(response, :completion_tokens) || 0 - }) + case Usage.record(%{ + organization_id: organization_id, + model: model, + purpose: "network_insight", + prompt_tokens: Map.get(response, :prompt_tokens) || 0, + completion_tokens: Map.get(response, :completion_tokens) || 0 + }) do + {:ok, _} -> + :ok - :ok + {:error, changeset} -> + Logger.error("Failed to record LLM usage: #{inspect(changeset.errors)}") + :ok + end end end diff --git a/lib/towerops/workers/insight_llm_enrichment_worker.ex b/lib/towerops/workers/insight_llm_enrichment_worker.ex index bbf3ddb7..155a88ed 100644 --- a/lib/towerops/workers/insight_llm_enrichment_worker.ex +++ b/lib/towerops/workers/insight_llm_enrichment_worker.ex @@ -34,24 +34,33 @@ 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) - # 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 - }) + case Insights.apply_llm_enrichment(insight, parsed, model) do + {:ok, _} -> + record_usage(insight.organization_id, response, model) - :ok + {:error, changeset} -> + Logger.error("Failed to persist LLM enrichment for insight #{insight.id}: #{inspect(changeset.errors)}") + end {:error, reason} -> - Logger.warning("LLM enrichment failed for insight #{insight.id}: #{inspect(reason)}") + Logger.error("LLM enrichment failed for insight #{insight.id}: #{inspect(reason)}") + end + end + + defp record_usage(organization_id, response, model) do + case Usage.record(%{ + organization_id: organization_id, + model: model, + purpose: "insight_enrichment", + prompt_tokens: Map.get(response, :prompt_tokens) || 0, + completion_tokens: Map.get(response, :completion_tokens) || 0 + }) do + {:ok, _} -> + :ok + + {:error, changeset} -> + Logger.error("Failed to record LLM usage: #{inspect(changeset.errors)}") :ok end end diff --git a/lib/towerops_web/live/insights_live/index.ex b/lib/towerops_web/live/insights_live/index.ex index 3dbd4744..93c43d94 100644 --- a/lib/towerops_web/live/insights_live/index.ex +++ b/lib/towerops_web/live/insights_live/index.ex @@ -107,16 +107,22 @@ defmodule ToweropsWeb.InsightsLive.Index do @impl true def handle_event("regenerate_insights", _params, socket) do if Scope.superuser?(socket.assigns.current_scope) do - Enum.each(@regeneration_workers, fn worker -> - %{} |> worker.new() |> Oban.insert() - end) + {ok_count, fail_count} = insert_regeneration_jobs() - {:noreply, - put_flash( - socket, - :info, - t("Insight regeneration queued. New insights will appear shortly.") - )} + flash = + if fail_count == 0 do + {:info, t("Insight regeneration queued. New insights will appear shortly.")} + else + {:error, + "%{ok} worker queued, %{fail} worker failed to queue. Check logs." + |> ngettext( + "%{ok} workers queued, %{fail} workers failed to queue. Check logs.", + ok_count + ) + |> then(&Gettext.dngettext(ToweropsWeb.Gettext, "", &1, &1, ok_count: ok_count, fail: fail_count))} + end + + {:noreply, put_flash(socket, elem(flash, 0), elem(flash, 1))} else {:noreply, put_flash(socket, :error, t("Not authorized."))} end @@ -139,6 +145,21 @@ defmodule ToweropsWeb.InsightsLive.Index do end end + defp insert_regeneration_jobs do + require Logger + + Enum.reduce(@regeneration_workers, {0, 0}, fn worker, {ok, fail} -> + case %{} |> worker.new() |> Oban.insert() do + {:ok, _} -> + {ok + 1, fail} + + {:error, changeset} -> + Logger.error("Failed to insert #{inspect(worker)}: #{inspect(changeset.errors)}") + {ok, fail + 1} + end + end) + end + defp load_insights(socket) do org_id = socket.assigns.organization.id opts = [status: socket.assigns.filter_status]