defmodule PhoenixKitAI do @moduledoc """ Main context for PhoenixKit AI system. Provides AI endpoint management and usage tracking for AI API requests. ## Architecture Each **Endpoint** is a unified configuration that combines: - Provider credentials (api_key, base_url, provider_settings) - Model selection (single model per endpoint) - Generation parameters (temperature, max_tokens, etc.) Users create as many endpoints as needed, each representing one complete AI configuration ready for making API requests. ## Core Functions ### System Management - `enabled?/0` - Check if AI module is enabled - `enable_system/0` - Enable the AI module - `disable_system/0` - Disable the AI module - `get_config/0` - Get module configuration with statistics ### Endpoint CRUD - `list_endpoints/1` - List all endpoints with filters - `get_endpoint!/1` - Get endpoint by UUID (raises) - `get_endpoint/1` - Get endpoint by UUID - `create_endpoint/1` - Create new endpoint - `update_endpoint/2` - Update existing endpoint - `delete_endpoint/1` - Delete endpoint ### Completion API - `ask/3` - Simple single-turn completion - `complete/3` - Multi-turn chat completion - `embed/3` - Generate embeddings ### Usage Tracking - `list_requests/1` - List requests with pagination/filters - `create_request/1` - Log a new request - `get_usage_stats/1` - Get aggregated statistics - `get_dashboard_stats/0` - Get stats for dashboard display ## Usage Examples # Enable the module PhoenixKitAI.enable_system() # Create an endpoint {:ok, endpoint} = PhoenixKitAI.create_endpoint(%{ name: "Claude Fast", provider: "openrouter", api_key: "sk-or-v1-...", model: "anthropic/claude-3-haiku", temperature: 0.7 }) # Use the endpoint {:ok, response} = PhoenixKitAI.ask(endpoint.uuid, "Hello!") # Extract the response text {:ok, text} = PhoenixKitAI.extract_content(response) ## Configuration # Persist message + response content in request metadata (default: true). # Disable for deployments with PII / data-retention obligations — token # counts, latency, model, and cost are still recorded. config :phoenix_kit_ai, capture_request_content: false # Capture process memory in request caller_context (default: false). config :phoenix_kit_ai, capture_request_memory: true # Allow endpoint base_url to point at private/loopback IPs (default: false). # Required for self-hosted Ollama / intranet inference. config :phoenix_kit_ai, allow_internal_endpoint_urls: true """ use PhoenixKit.Module import Ecto.Query, warn: false require Logger alias PhoenixKit.Dashboard.Tab alias PhoenixKit.PubSub.Manager, as: PubSub alias PhoenixKit.Settings alias PhoenixKit.Utils.Date, as: UtilsDate alias PhoenixKit.Utils.UUID, as: UUIDUtils alias PhoenixKitAI.Endpoint alias PhoenixKitAI.Prompt alias PhoenixKitAI.Request # =========================================== # PUBSUB TOPICS # =========================================== @endpoints_topic "phoenix_kit:ai:endpoints" @prompts_topic "phoenix_kit:ai:prompts" @requests_topic "phoenix_kit:ai:requests" @doc """ Returns the PubSub topic for AI endpoints. Subscribe to this topic to receive real-time updates. """ @spec endpoints_topic() :: String.t() def endpoints_topic, do: @endpoints_topic @doc """ Returns the PubSub topic for AI prompts. """ @spec prompts_topic() :: String.t() def prompts_topic, do: @prompts_topic @doc """ Returns the PubSub topic for AI requests/usage. """ @spec requests_topic() :: String.t() def requests_topic, do: @requests_topic @doc """ Subscribes the current process to AI endpoint changes. """ @spec subscribe_endpoints() :: :ok | {:error, term()} def subscribe_endpoints do PubSub.subscribe(@endpoints_topic) end @doc """ Subscribes the current process to AI prompt changes. """ @spec subscribe_prompts() :: :ok | {:error, term()} def subscribe_prompts do PubSub.subscribe(@prompts_topic) end @doc """ Subscribes the current process to AI request/usage changes. """ @spec subscribe_requests() :: :ok | {:error, term()} def subscribe_requests do PubSub.subscribe(@requests_topic) end # =========================================== # HELPERS # =========================================== defp repo do PhoenixKit.RepoHelper.repo() end # Textual-UUID check (36 chars, hyphenated). Core's `UUIDUtils.valid?/1` # delegates to `Ecto.UUID.cast/1`, which also accepts *16-byte* raw # binaries — that means any 16-character slug would be mistaken for a # UUID in UUID-vs-slug dispatch helpers. Guarding with byte_size keeps # the dispatch correct without touching core. defp textual_uuid?(string) when is_binary(string) do byte_size(string) == 36 and UUIDUtils.valid?(string) end defp broadcast_endpoint_change(result, event) do case result do {:ok, endpoint} -> PubSub.broadcast(@endpoints_topic, {event, endpoint}) {:ok, endpoint} error -> error end end defp broadcast_prompt_change(result, event) do case result do {:ok, prompt} -> PubSub.broadcast(@prompts_topic, {event, prompt}) {:ok, prompt} error -> error end end defp broadcast_request_change(result, event) do case result do {:ok, request} -> PubSub.broadcast(@requests_topic, {event, request}) {:ok, request} error -> error end end # =========================================== # ACTIVITY LOGGING HELPERS # =========================================== # Log a successful endpoint mutation. The failure-branch row is # written by `log_failed_endpoint_mutation/3` lower down. Together # they ensure the audit feed records every user-initiated mutation # attempt, including ones that failed due to validation, FK # violations, or DB outages. `metadata.db_pending: true` # distinguishes attempted-but-failed from completed actions. defp log_endpoint_activity({:ok, endpoint} = result, action, opts) do log_activity(action, "endpoint", endpoint.uuid, opts, %{"name" => endpoint.name}) result end defp log_endpoint_activity(error, _action, _opts), do: error defp log_prompt_activity({:ok, prompt} = result, action, opts) do log_activity(action, "prompt", prompt.uuid, opts, %{"name" => prompt.name}) result end defp log_prompt_activity(error, _action, _opts), do: error # Failure-branch audit row. Writes `metadata.db_pending: true` plus # the validation `error_keys` so the audit feed can distinguish # attempted-but-failed from completed mutations. Returns the result # unchanged so callers can continue piping. No-op on `{:ok, _}`. defp log_failed_endpoint_mutation( {:error, %Ecto.Changeset{} = changeset} = result, action, opts ) do log_activity( action, "endpoint", Map.get(changeset.data, :uuid), opts, failure_metadata(changeset, :name) ) result end defp log_failed_endpoint_mutation(result, _action, _opts), do: result defp log_failed_prompt_mutation( {:error, %Ecto.Changeset{} = changeset} = result, action, opts ) do log_activity( action, "prompt", Map.get(changeset.data, :uuid), opts, failure_metadata(changeset, :name) ) result end defp log_failed_prompt_mutation(result, _action, _opts), do: result # PII-safe failure metadata — `name` is the only changeset field we # surface (the resource's display string, already public in the admin # UI); `error_keys` is the list of failed validation keys, never the # rejected values themselves. defp failure_metadata(changeset, name_field) do %{ "name" => Ecto.Changeset.get_field(changeset, name_field), "db_pending" => true, "error_keys" => changeset.errors |> Keyword.keys() |> Enum.map(&Atom.to_string/1) } end # Enable/disable toggle — logs only when the flag actually changes. defp maybe_log_endpoint_toggle({:ok, endpoint} = result, was_enabled, opts) do cond do was_enabled == endpoint.enabled -> result endpoint.enabled -> log_toggle(result, "endpoint.enabled", endpoint, "endpoint", opts) true -> log_toggle(result, "endpoint.disabled", endpoint, "endpoint", opts) end end defp maybe_log_endpoint_toggle(error, _, _), do: error defp maybe_log_prompt_toggle({:ok, prompt} = result, was_enabled, opts) do cond do was_enabled == prompt.enabled -> result prompt.enabled -> log_toggle(result, "prompt.enabled", prompt, "prompt", opts) true -> log_toggle(result, "prompt.disabled", prompt, "prompt", opts) end end defp maybe_log_prompt_toggle(error, _, _), do: error defp log_toggle(result, action, resource, resource_type, opts) do log_activity(action, resource_type, resource.uuid, opts, %{"name" => resource.name}) result end # Unified logger — guarded by Code.ensure_loaded?/1 and rescued so # activity failures never crash the primary operation. No-op on hosts # without PhoenixKit.Activity available. # # The `:undefined_table` case is silently skipped: hosts that haven't # run the core PhoenixKit migrations yet simply don't have the # `phoenix_kit_activities` table, so logging would be noise on every # mutation. Any other failure is logged so real bugs aren't hidden. defp log_activity(action, resource_type, resource_uuid, opts, extra) do if Code.ensure_loaded?(PhoenixKit.Activity) do metadata = %{"actor_role" => Keyword.get(opts, :actor_role, "user")} |> Map.merge(extra) PhoenixKit.Activity.log(%{ action: action, module: "ai", mode: Keyword.get(opts, :mode, "manual"), actor_uuid: Keyword.get(opts, :actor_uuid), resource_type: resource_type, resource_uuid: resource_uuid, metadata: metadata }) end rescue e in Postgrex.Error -> if Map.get(e.postgres || %{}, :code) == :undefined_table do # Host hasn't run the core activity migration yet — silent no-op. :activity_log_unavailable else log_activity_failure(action, e) end e -> log_activity_failure(action, e) end defp log_activity_failure(action, exception) do require Logger Logger.warning( "[PhoenixKitAI] activity log failed for #{action}: #{Exception.message(exception)}" ) :activity_log_failed end # =========================================== # SYSTEM MANAGEMENT # =========================================== @doc """ Checks if the AI module is enabled. """ @impl PhoenixKit.Module @spec enabled?() :: boolean() def enabled? do Settings.get_boolean_setting("ai_enabled", false) rescue _ -> false catch # `Settings.get_boolean_setting/2` can hit a shutting-down pool # in tests and exit on `DBConnection.Holder.checkout/3`. The # convention is "must return false as fallback" — that includes # process exits, not just exceptions. :exit, _ -> false end @doc """ Enables the AI module. """ @impl PhoenixKit.Module @spec enable_system() :: {:ok, term()} | {:error, term()} def enable_system do Settings.update_boolean_setting_with_module("ai_enabled", true, module_key()) end @doc """ Disables the AI module. """ @impl PhoenixKit.Module @spec disable_system() :: {:ok, term()} | {:error, term()} def disable_system do Settings.update_boolean_setting_with_module("ai_enabled", false, module_key()) end @doc """ One-shot auto-migrator for legacy `endpoint.api_key` values into `PhoenixKit.Integrations` connections. Mirrors the pattern of `PhoenixKit.Integrations.run_legacy_migrations/0` — call it at host-app boot to fold pre-Integrations endpoint api_keys into the named-connection model. Safe to call multiple times: multiple idempotency guards short-circuit on already-migrated state. ## What it does For each `phoenix_kit_ai_endpoints` row whose `provider` is the bare string `"openrouter"` (i.e., NOT already pointing at a named Integrations connection like `"openrouter:my-key"`) AND whose `api_key` is non-empty: 1. Group by api_key value — endpoints sharing a key share one connection (dedup). 2. Create a `PhoenixKit.Integrations` connection per distinct key. Naming: `"openrouter:default"` if there's exactly one key in the deployment; `"openrouter:imported-1"`, `"openrouter:imported-2"` (1-indexed by first-seen order) if there are multiple. 3. Update each endpoint's `provider` field to point at the new connection key (e.g., `"openrouter:default"`). The legacy `api_key` column is NEVER cleared — it stays on each row as a safety net. `OpenRouterClient.resolve_api_key/2` prefers Integrations, so post-migration endpoints stop firing the legacy warning; if Integrations later breaks for any reason, the column still has the value and the fallback path keeps working. ## Idempotency guards (any one short-circuits) - The `ai_legacy_api_key_migration_completed_at` setting is set → already ran, skip. - ANY `integration:openrouter:*` key already exists in `phoenix_kit_settings` (operator already set up Integrations manually) → mark completed and skip. - NO endpoints have `provider == "openrouter"` with a non-empty `api_key` → nothing to migrate, mark completed. ## Failure modes Top-level `try/rescue/catch :exit` so DB outages, race conditions, or any unexpected exception NEVER crashes the host app's boot. Per-key-group operations are isolated — one bad group doesn't abort the others. Partial migration is safe because un-migrated endpoints still resolve via the legacy fallback path. ## Configuration No options. Disable by simply not calling the function. """ @spec run_legacy_api_key_migration() :: :ok def run_legacy_api_key_migration do case do_run_legacy_api_key_migration() do :skipped -> :ok {:migrated, count} -> require Logger Logger.info( "[PhoenixKitAI] Auto-migrated #{count} endpoint(s) from legacy api_key " <> "to PhoenixKit.Integrations connections" ) :ok end rescue e -> require Logger Logger.warning( "[PhoenixKitAI] Legacy api_key migration crashed (host boot continues): " <> Exception.message(e) ) :ok catch :exit, _reason -> :ok end defp do_run_legacy_api_key_migration do cond do not Code.ensure_loaded?(PhoenixKit.Integrations) -> :skipped legacy_api_key_migration_completed?() -> :skipped any_openrouter_integration_exists?() -> # The auto-migrator only ever creates `integration:openrouter:*` # rows (its target population is endpoints with `provider == # "openrouter"` and a non-empty `api_key`). If any OpenRouter # connection already exists in `phoenix_kit_settings`, the # operator either set Integrations up manually OR a previous # run completed — either way the desired end state is reached. # Mark complete and skip future runs. # # **Scope clarification** (PR #6 review finding #8): this gate # is OpenRouter-only. A multi-provider deployment that has, # say, a Mistral integration set up but still has unmigrated # legacy OpenRouter `api_key` columns will NOT trip this guard # — the migration runs normally for the OpenRouter rows. # `any_openrouter_integration_exists?/0` mirrors the # migration's actual target population, not a global "any # integration anywhere" check. mark_legacy_api_key_migration_complete() :skipped true -> attempt_legacy_api_key_migration() end end defp legacy_api_key_migration_completed? do Settings.get_setting("ai_legacy_api_key_migration_completed_at", nil) != nil rescue # Settings table missing in this environment — treat as not completed # but the next guard (any_openrouter_integration_exists?) will trip # on the same missing infra and we'll skip safely. _ -> false end defp any_openrouter_integration_exists? do query = from(s in "phoenix_kit_settings", where: like(s.key, "integration:openrouter:%"), select: count(s.uuid) ) repo().one(query) > 0 rescue # If the settings table or column shape isn't what we expect, fall # through to "yes, skip" — safer than risking a partial migration # against an unfamiliar schema. _ -> true end defp attempt_legacy_api_key_migration do candidates = list_legacy_api_key_endpoints() if Enum.empty?(candidates) do mark_legacy_api_key_migration_complete() :skipped else grouped_by_key = Enum.group_by(candidates, & &1.api_key) total_groups = map_size(grouped_by_key) # Snapshot the existing OpenRouter connections once so the # `:already_exists` fallback is a map lookup rather than a fresh # `list_connections/1` query per group. On a deployment with N # distinct legacy keys re-running migration, this collapses # `O(groups × connections)` reads down to one. Connections # created during the loop come back via `add_connection`'s # `{:ok, %{uuid: _}}` directly, so they don't need the cache. existing_uuids = snapshot_connection_uuids("openrouter") migrated_count = grouped_by_key |> Enum.with_index(1) |> Enum.reduce(0, fn {{api_key, endpoints}, index}, acc -> name = legacy_connection_name(total_groups, index) acc + migrate_endpoint_group(name, api_key, endpoints, existing_uuids) end) mark_legacy_api_key_migration_complete() {:migrated, migrated_count} end end defp snapshot_connection_uuids(provider) do PhoenixKit.Integrations.list_connections(provider) |> Map.new(fn conn -> {conn.name, conn.uuid} end) rescue _ -> %{} end defp list_legacy_api_key_endpoints do # Only touch endpoints whose provider is the bare "openrouter" # string. Endpoints already pointing at a named connection (e.g. # "openrouter:my-key") have been migrated by hand or by an earlier # run — leave them alone. query = from(e in Endpoint, where: e.provider == "openrouter" and not is_nil(e.api_key) and e.api_key != "", select: %{uuid: e.uuid, api_key: e.api_key, name: e.name} ) repo().all(query) rescue _ -> [] end defp legacy_connection_name(1, _index), do: "default" defp legacy_connection_name(_total, index), do: "imported-#{index}" defp migrate_endpoint_group(name, api_key, endpoints, existing_uuids) do full_key = "openrouter:#{name}" # Two-step write under core's strict-UUID Integrations API: # add_connection/3 creates (or surfaces) the row and returns its # uuid; save_setup/3 then stores the legacy api_key against that # uuid. `:already_exists` on re-runs is fine — pull the existing # row's uuid out of the snapshot taken before the loop started. integration_uuid = case PhoenixKit.Integrations.add_connection("openrouter", name) do {:ok, %{uuid: uuid}} -> uuid {:error, :already_exists} -> Map.get(existing_uuids, name) _ -> nil end if is_nil(integration_uuid) do require Logger Logger.warning( "[PhoenixKitAI] Skipping legacy api_key group (#{length(endpoints)} endpoints) — " <> "could not resolve integration uuid" ) 0 else case PhoenixKit.Integrations.save_setup(integration_uuid, %{"api_key" => api_key}) do {:ok, _saved} -> count = update_endpoints_provider(endpoints, full_key, integration_uuid) if count > 0 do log_migration_activity(:credentials_migrated, %{ "endpoint_count" => count, "integration_uuid" => integration_uuid, "connection_name" => name }) end count {:error, _reason} -> require Logger Logger.warning( "[PhoenixKitAI] Skipping legacy api_key group (#{length(endpoints)} endpoints) — " <> "save_setup failed" ) 0 end end rescue _ -> 0 end defp update_endpoints_provider(endpoints, new_provider, integration_uuid) do uuids = Enum.map(endpoints, & &1.uuid) # Atomic write: set the new provider/integration_uuid AND clear # the legacy api_key column in the same UPDATE statement. The # integration row created above (`save_setup`) holds the same # api_key the endpoint had — it's the canonical credential # source now. Keeping a duplicate on the endpoint row would only # rot quietly (admin rotates the integration's key, endpoint's # column drifts) and mask config drift if the integration ever # breaks. Cleared to "" rather than NULL because the column is # NOT NULL (V34); both the runtime fallback chain # (`maybe_get_credentials("") => {:error, :not_configured}`) and # the recovery-card render condition treat "" as "no fallback". set_clause = [provider: new_provider, api_key: "", updated_at: DateTime.utc_now()] |> maybe_add_integration_uuid(integration_uuid) {count, _} = from(e in Endpoint, where: e.uuid in ^uuids) |> repo().update_all(set: set_clause) count rescue _ -> 0 end # Caller (`update_endpoints_provider/3`) is only reachable via the # `is_nil(integration_uuid) -> 0` short-circuit one frame up, so the # `nil` clause here is unreachable per dialyzer's typeflow. Empty # string is still possible (call sites that pass through raw form # input) — keep that branch. defp maybe_add_integration_uuid(set_clause, ""), do: set_clause defp maybe_add_integration_uuid(set_clause, uuid) when is_binary(uuid) do Keyword.put(set_clause, :integration_uuid, uuid) end defp mark_legacy_api_key_migration_complete do Settings.update_setting_with_module( "ai_legacy_api_key_migration_completed_at", DateTime.utc_now() |> DateTime.to_iso8601(), module_key() ) :ok rescue _ -> :ok end # =========================================== # Combined migrate_legacy/0 callback # =========================================== @doc """ Combined boot-time migration entry point. Runs both legacy data transitions for AI: 1. **Local api_key → Integrations row + integration_uuid** (delegates to `run_legacy_api_key_migration/0`). Endpoints with bare `provider == "openrouter"` and a non-empty `api_key` get grouped, get an Integration row created, and have both `provider` AND `integration_uuid` updated to point at it. 2. **`provider`-string → `integration_uuid`** (sweep). Endpoints with `integration_uuid IS NULL` whose `provider` field resolves to a real integration uuid (either bare provider, `provider:name` shape, or a uuid stuffed in the string column from pre-V107 form saves) get their `integration_uuid` populated. V107's migration backfilled most of these at install time; this pass catches stragglers (e.g., endpoints created post-form-update but pre-V107). Both kinds log to `PhoenixKit.Activity` per migrated record / group with `mode: "auto"` and module `"ai"`. PII-safe: never logs `api_key` values. Idempotent — run on every host-app boot. Designed to be invoked via the orchestrator (`PhoenixKit.ModuleRegistry.run_all_legacy_migrations/0`), but can be called directly for ad-hoc migration runs. """ @impl PhoenixKit.Module @spec migrate_legacy() :: {:ok, map()} | {:error, term()} def migrate_legacy do credentials_result = run_legacy_api_key_migration() references_result = sweep_provider_string_to_integration_uuid() summary = %{ credentials_migration: credentials_result, reference_migration: references_result } # Surface inner sweep failures at the outer return shape so the # orchestrator (`PhoenixKit.ModuleRegistry.run_all_legacy_migrations/0`) # can pattern-match on `{:ok, _}` to decide "ran cleanly". Wrapping # `:error` inside `{:ok, %{reference_migration: :error}}` would # silently swallow the failure for any caller that doesn't drill # into the map. case references_result do :error -> {:error, {:sweep_failed, summary}} _ -> {:ok, summary} end rescue e -> require Logger Logger.warning("[PhoenixKitAI] migrate_legacy/0 raised: #{Exception.message(e)}") {:error, e} end defp sweep_provider_string_to_integration_uuid do endpoints = list_endpoints_needing_uuid_promotion() if Enum.empty?(endpoints) do :nothing_to_migrate else migrated = Enum.reduce(endpoints, 0, fn endpoint, acc -> case promote_provider_to_integration_uuid(endpoint) do :ok -> acc + 1 _ -> acc end end) if migrated > 0 do log_migration_activity(:reference_migrated, %{ "endpoint_count" => migrated, "source" => "boot_sweep" }) end {:migrated, migrated} end rescue e -> require Logger Logger.warning(fn -> "[PhoenixKitAI] reference sweep crashed: " <> "exception=#{inspect(e.__struct__)}, " <> "message=#{Exception.message(e)}" end) :error end defp list_endpoints_needing_uuid_promotion do # Endpoints with NULL integration_uuid but a provider field that # might resolve. Skip rows whose provider is the bare default # ("openrouter") because credentials migration handles those when # they have an api_key — and a bare provider with no api_key # has nothing to promote to. query = from(e in Endpoint, where: is_nil(e.integration_uuid) and not is_nil(e.provider) and e.provider != "" and e.provider != "openrouter", select: %{uuid: e.uuid, provider: e.provider} ) repo().all(query) rescue _ -> [] end defp promote_provider_to_integration_uuid(%{uuid: endpoint_uuid, provider: provider}) do integration_uuid = resolve_provider_to_uuid(provider) if is_nil(integration_uuid) do :no_match else {count, _} = from(e in Endpoint, where: e.uuid == ^endpoint_uuid and is_nil(e.integration_uuid)) |> repo().update_all( set: [integration_uuid: integration_uuid, updated_at: DateTime.utc_now()] ) if count > 0, do: :ok, else: :no_op end rescue _ -> :error end defp resolve_provider_to_uuid(provider) when is_binary(provider) do # Delegates to core's dual-input primitive. Previously this helper # carried its own regex + dispatch + provider:name split — a # near-clone of `OpenRouterClient.lookup_uuid_for_provider/1`. # Centralising into `Integrations.resolve_to_uuid/1` removes the # duplication and eliminates the "fourth provider tempts a third # copy-paste" risk. case PhoenixKit.Integrations.resolve_to_uuid(provider) do {:ok, uuid} -> uuid _ -> nil end rescue _ -> nil end defp resolve_provider_to_uuid(_), do: nil defp log_migration_activity(action_atom, metadata) do if Code.ensure_loaded?(PhoenixKit.Activity) do PhoenixKit.Activity.log(%{ action: "integration.legacy_migrated", module: module_key(), mode: "auto", resource_type: "endpoint", metadata: Map.merge(metadata, %{ "migration_kind" => Atom.to_string(action_atom), "actor_role" => "system" }) }) end :ok rescue _ -> :ok end @doc """ Gets the AI module configuration with statistics. Stat queries are wrapped in a try/rescue so that environments without a live Repo connection (early boot, test cases without sandbox checkout) still get a well-formed map — matching the defensive pattern in `enabled?/0`. """ @impl PhoenixKit.Module @spec get_config() :: %{ enabled: boolean(), endpoints_count: non_neg_integer(), total_requests: non_neg_integer(), total_tokens: non_neg_integer() } def get_config do %{ enabled: enabled?(), endpoints_count: safe_count(&count_endpoints/0), total_requests: safe_count(&count_requests/0), total_tokens: safe_count(&sum_tokens/0) } end defp safe_count(fun) do fun.() rescue _ -> 0 catch :exit, _ -> 0 end # =========================================== # MODULE BEHAVIOUR CALLBACKS # =========================================== @impl PhoenixKit.Module @spec module_key() :: String.t() def module_key, do: "ai" @impl PhoenixKit.Module @spec module_name() :: String.t() def module_name, do: "AI" @impl PhoenixKit.Module @spec permission_metadata() :: map() def permission_metadata do %{ key: module_key(), label: "AI", icon: "hero-sparkles", description: "AI endpoints, prompts, and usage tracking" } end @impl PhoenixKit.Module @spec admin_tabs() :: [PhoenixKit.Dashboard.Tab.t()] def admin_tabs do [ %Tab{ id: :admin_ai, label: "AI", icon: "hero-cpu-chip", path: "ai", priority: 640, level: :admin, permission: module_key(), match: :prefix, group: :admin_modules, subtab_display: :when_active, highlight_with_subtabs: false, redirect_to_first_subtab: true }, %Tab{ id: :admin_ai_endpoints, label: "Endpoints", icon: "hero-server-stack", path: "ai/endpoints", priority: 641, level: :admin, permission: module_key(), parent: :admin_ai }, %Tab{ id: :admin_ai_prompts, label: "Prompts", icon: "hero-document-text", path: "ai/prompts", priority: 642, level: :admin, permission: module_key(), parent: :admin_ai }, %Tab{ id: :admin_ai_playground, label: "Playground", icon: "hero-beaker", path: "ai/playground", priority: 643, level: :admin, permission: module_key(), parent: :admin_ai }, %Tab{ id: :admin_ai_usage, label: "Usage", icon: "hero-chart-bar", path: "ai/usage", priority: 644, level: :admin, permission: module_key(), parent: :admin_ai } ] end @impl PhoenixKit.Module @dialyzer {:nowarn_function, css_sources: 0} @spec css_sources() :: [atom()] def css_sources, do: [:phoenix_kit_ai] @impl PhoenixKit.Module @spec required_integrations() :: [String.t()] def required_integrations, do: ["openrouter"] @impl PhoenixKit.Module @spec version() :: String.t() def version, do: "0.1.5" @impl PhoenixKit.Module @spec route_module() :: module() def route_module, do: PhoenixKitAI.Routes # =========================================== # ENDPOINT CRUD # =========================================== @doc """ Lists all AI endpoints. ## Options - `:provider` - Filter by provider type - `:enabled` - Filter by enabled status - `:preload` - Associations to preload ## Examples PhoenixKitAI.list_endpoints() PhoenixKitAI.list_endpoints(provider: "openrouter", enabled: true) """ @spec list_endpoints(keyword()) :: {[Endpoint.t()], non_neg_integer()} def list_endpoints(opts \\ []) do sort_by = Keyword.get(opts, :sort_by, :sort_order) sort_dir = Keyword.get(opts, :sort_dir, :asc) # Always paginate, default to page 1 and ensure it's > 0 page = Keyword.get(opts, :page, 1) |> max(1) page_size = Keyword.get(opts, :page_size, 20) # Build base query with filters (no sorting yet) base_query = from(e in Endpoint) base_query = case Keyword.get(opts, :provider) do nil -> base_query provider -> where(base_query, [e], e.provider == ^provider) end base_query = case Keyword.get(opts, :enabled) do nil -> base_query enabled -> where(base_query, [e], e.enabled == ^enabled) end # Count on base query BEFORE applying sorting (which may add group_by) total = repo().aggregate(base_query, :count) # Now apply sorting (may add group_by for usage/tokens/cost/last_used) query = apply_endpoint_sorting(base_query, sort_by, sort_dir) query = case Keyword.get(opts, :preload) do nil -> query preloads -> preload(query, ^preloads) end offset = (page - 1) * page_size endpoints = query |> limit(^page_size) |> offset(^offset) |> repo().all() # Always return the same shape {endpoints, total} end defp apply_endpoint_sorting(query, :usage, dir) do # Sort by total request count using a subquery to avoid GROUP BY issues stats_subquery = from(r in Request, where: not is_nil(r.endpoint_uuid), group_by: r.endpoint_uuid, select: %{endpoint_uuid: r.endpoint_uuid, count: count()} ) from(e in query, left_join: s in subquery(stats_subquery), on: s.endpoint_uuid == e.uuid, order_by: [{^dir, coalesce(s.count, 0)}] ) end defp apply_endpoint_sorting(query, :tokens, dir) do # Sort by total tokens used stats_subquery = from(r in Request, where: not is_nil(r.endpoint_uuid), group_by: r.endpoint_uuid, select: %{endpoint_uuid: r.endpoint_uuid, total: coalesce(sum(r.total_tokens), 0)} ) from(e in query, left_join: s in subquery(stats_subquery), on: s.endpoint_uuid == e.uuid, order_by: [{^dir, coalesce(s.total, 0)}] ) end defp apply_endpoint_sorting(query, :cost, dir) do # Sort by total cost stats_subquery = from(r in Request, where: not is_nil(r.endpoint_uuid), group_by: r.endpoint_uuid, select: %{endpoint_uuid: r.endpoint_uuid, total: coalesce(sum(r.cost_cents), 0)} ) from(e in query, left_join: s in subquery(stats_subquery), on: s.endpoint_uuid == e.uuid, order_by: [{^dir, coalesce(s.total, 0)}] ) end defp apply_endpoint_sorting(query, :last_used, dir) do # Sort by most recent request time stats_subquery = from(r in Request, where: not is_nil(r.endpoint_uuid), group_by: r.endpoint_uuid, select: %{endpoint_uuid: r.endpoint_uuid, last_used: max(r.inserted_at)} ) from(e in query, left_join: s in subquery(stats_subquery), on: s.endpoint_uuid == e.uuid, order_by: [{^dir, s.last_used}] ) end defp apply_endpoint_sorting(query, field, dir) when field in [:name, :enabled, :model, :sort_order] do order_by(query, [e], [{^dir, field(e, ^field)}]) end defp apply_endpoint_sorting(query, _field, _dir) do # Default sorting order_by(query, [e], asc: e.sort_order, desc: e.inserted_at) end @doc """ Returns usage statistics for each endpoint. Returns a map of endpoint_uuid => %{request_count, total_tokens, total_cost, last_used_at} """ def get_endpoint_usage_stats do query = from(r in Request, where: not is_nil(r.endpoint_uuid), group_by: r.endpoint_uuid, select: { r.endpoint_uuid, %{ request_count: count(), total_tokens: coalesce(sum(r.total_tokens), 0), total_cost: coalesce(sum(r.cost_cents), 0), last_used_at: max(r.inserted_at) } } ) query |> repo().all() |> Map.new() end @doc """ Gets a single endpoint by UUID. Raises `Ecto.NoResultsError` if the endpoint does not exist. """ @spec get_endpoint!(String.t()) :: Endpoint.t() def get_endpoint!(id) do case get_endpoint(id) do nil -> raise Ecto.NoResultsError, queryable: Endpoint endpoint -> endpoint end end @doc """ Gets a single endpoint by UUID. Accepts a UUID string (e.g., "550e8400-e29b-41d4-a716-446655440000"). Returns `nil` if the endpoint does not exist. """ @spec get_endpoint(term()) :: Endpoint.t() | nil def get_endpoint(id) when is_binary(id) do if textual_uuid?(id) do repo().get_by(Endpoint, uuid: id) else nil end end def get_endpoint(_), do: nil @doc """ Resolves an endpoint from an ID (UUID string) or Endpoint struct. ## Examples {:ok, endpoint} = PhoenixKitAI.resolve_endpoint("019abc12-3456-7def-8901-234567890abc") {:ok, endpoint} = PhoenixKitAI.resolve_endpoint(endpoint) """ @spec resolve_endpoint(term()) :: {:ok, Endpoint.t()} | {:error, :endpoint_not_found | :invalid_endpoint_identifier} def resolve_endpoint(id) when is_binary(id) do case get_endpoint(id) do nil -> {:error, :endpoint_not_found} endpoint -> {:ok, endpoint} end end def resolve_endpoint(%Endpoint{} = endpoint), do: {:ok, endpoint} def resolve_endpoint(_), do: {:error, :invalid_endpoint_identifier} @doc """ Creates a new AI endpoint. ## Examples {:ok, endpoint} = PhoenixKitAI.create_endpoint(%{ name: "Claude Fast", provider: "openrouter", api_key: "sk-or-v1-...", model: "anthropic/claude-3-haiku", temperature: 0.7 }) """ @spec create_endpoint(map(), keyword()) :: {:ok, Endpoint.t()} | {:error, Ecto.Changeset.t()} def create_endpoint(attrs, opts \\ []) do %Endpoint{} |> Endpoint.changeset(attrs) |> repo().insert() |> broadcast_endpoint_change(:endpoint_created) |> log_endpoint_activity("endpoint.created", opts) |> log_failed_endpoint_mutation("endpoint.created", opts) end @doc """ Updates an existing AI endpoint. Accepts an `:actor_uuid` option so the mutation can be attributed in the activity feed. If the change toggles the `enabled` flag an additional `endpoint.enabled` / `endpoint.disabled` entry is logged. """ @spec update_endpoint(Endpoint.t(), map(), keyword()) :: {:ok, Endpoint.t()} | {:error, Ecto.Changeset.t()} def update_endpoint(%Endpoint{} = endpoint, attrs, opts \\ []) do was_enabled = endpoint.enabled changeset = Endpoint.changeset(endpoint, attrs) # Skip the activity row when the update is a no-op (no field actually # changed). The toggle log keeps its own guard so an enable/disable # via a bare `%{enabled: x}` still attributes correctly. has_changes = changeset.changes != %{} changeset |> repo().update() |> broadcast_endpoint_change(:endpoint_updated) |> maybe_log_endpoint_update(has_changes, opts) |> maybe_log_endpoint_toggle(was_enabled, opts) |> log_failed_endpoint_mutation("endpoint.updated", opts) end defp maybe_log_endpoint_update({:ok, _} = result, true, opts) do log_endpoint_activity(result, "endpoint.updated", opts) end defp maybe_log_endpoint_update(result, _has_changes, _opts), do: result @doc """ Deletes an AI endpoint. """ @spec delete_endpoint(Endpoint.t(), keyword()) :: {:ok, Endpoint.t()} | {:error, Ecto.Changeset.t()} def delete_endpoint(%Endpoint{} = endpoint, opts \\ []) do repo().delete(endpoint) |> broadcast_endpoint_change(:endpoint_deleted) |> log_endpoint_activity("endpoint.deleted", opts) |> log_failed_endpoint_mutation("endpoint.deleted", opts) end @doc """ Returns an endpoint changeset for use in forms. """ @spec change_endpoint(Endpoint.t(), map()) :: Ecto.Changeset.t() def change_endpoint(%Endpoint{} = endpoint, attrs \\ %{}) do Endpoint.changeset(endpoint, attrs) end @doc """ Marks an endpoint as validated by updating its last_validated_at timestamp. """ @spec mark_endpoint_validated(Endpoint.t()) :: {:ok, Endpoint.t()} | {:error, Ecto.Changeset.t()} def mark_endpoint_validated(%Endpoint{} = endpoint) do endpoint |> Endpoint.validation_changeset() |> repo().update() end @doc """ Counts the total number of endpoints. """ @spec count_endpoints() :: non_neg_integer() def count_endpoints do repo().aggregate(Endpoint, :count) end @doc """ Counts the number of enabled endpoints. """ @spec count_enabled_endpoints() :: non_neg_integer() def count_enabled_endpoints do query = from(e in Endpoint, where: e.enabled == true) repo().aggregate(query, :count) end # =========================================== # PROMPT CRUD # =========================================== @doc """ Lists all AI prompts. ## Options - `:sort_by` - Field to sort by (default: :sort_order) - `:sort_dir` - Sort direction, :asc or :desc (default: :asc) - `:enabled` - Filter by enabled status ## Examples PhoenixKitAI.list_prompts() PhoenixKitAI.list_prompts(sort_by: :name, sort_dir: :asc) PhoenixKitAI.list_prompts(enabled: true) """ @spec list_prompts(keyword()) :: [Prompt.t()] | {[Prompt.t()], non_neg_integer()} def list_prompts(opts \\ []) do sort_by = Keyword.get(opts, :sort_by, :sort_order) sort_dir = Keyword.get(opts, :sort_dir, :asc) page = Keyword.get(opts, :page) page_size = Keyword.get(opts, :page_size, 20) query = from(p in Prompt) query = case Keyword.get(opts, :enabled) do nil -> query enabled -> where(query, [p], p.enabled == ^enabled) end query = order_by(query, [p], [{^sort_dir, field(p, ^sort_by)}]) # If page is provided, return paginated results with total count if page do total = repo().aggregate(query, :count) offset = (page - 1) * page_size prompts = query |> limit(^page_size) |> offset(^offset) |> repo().all() {prompts, total} else # No pagination - return all (backwards compatible) repo().all(query) end end @doc """ Lists only enabled prompts. Convenience wrapper for `list_prompts(enabled: true)`. ## Examples PhoenixKitAI.list_enabled_prompts() """ @spec list_enabled_prompts() :: [Prompt.t()] | {[Prompt.t()], non_neg_integer()} def list_enabled_prompts do list_prompts(enabled: true) end @doc """ Gets a single prompt by UUID. Raises `Ecto.NoResultsError` if the prompt does not exist. """ @spec get_prompt!(String.t()) :: Prompt.t() def get_prompt!(id) do case get_prompt(id) do nil -> raise Ecto.NoResultsError, queryable: Prompt prompt -> prompt end end @doc """ Gets a single prompt by UUID. Accepts a UUID string (e.g., "550e8400-e29b-41d4-a716-446655440000"). Returns `nil` if the prompt does not exist. """ @spec get_prompt(term()) :: Prompt.t() | nil def get_prompt(id) when is_binary(id) do if textual_uuid?(id) do repo().get_by(Prompt, uuid: id) else nil end end def get_prompt(_), do: nil @doc """ Gets a prompt by slug. Returns `nil` if the prompt does not exist. """ @spec get_prompt_by_slug(String.t()) :: Prompt.t() | nil def get_prompt_by_slug(slug) when is_binary(slug) do repo().get_by(Prompt, slug: slug) end @doc """ Creates a new AI prompt. ## Examples {:ok, prompt} = PhoenixKitAI.create_prompt(%{ name: "Translator", content: "Translate the following text to {{Language}}:\\n\\n{{Text}}" }) """ @spec create_prompt(map(), keyword()) :: {:ok, Prompt.t()} | {:error, Ecto.Changeset.t()} def create_prompt(attrs, opts \\ []) do %Prompt{} |> Prompt.changeset(attrs) |> repo().insert() |> broadcast_prompt_change(:prompt_created) |> log_prompt_activity("prompt.created", opts) |> log_failed_prompt_mutation("prompt.created", opts) end @doc """ Updates an existing AI prompt. Accepts an `:actor_uuid` option so the mutation can be attributed in the activity feed. If the change toggles the `enabled` flag an additional `prompt.enabled` / `prompt.disabled` entry is logged. """ @spec update_prompt(Prompt.t(), map(), keyword()) :: {:ok, Prompt.t()} | {:error, Ecto.Changeset.t()} def update_prompt(%Prompt{} = prompt, attrs, opts \\ []) do was_enabled = prompt.enabled changeset = Prompt.changeset(prompt, attrs) has_changes = changeset.changes != %{} changeset |> repo().update() |> broadcast_prompt_change(:prompt_updated) |> maybe_log_prompt_update(has_changes, opts) |> maybe_log_prompt_toggle(was_enabled, opts) |> log_failed_prompt_mutation("prompt.updated", opts) end defp maybe_log_prompt_update({:ok, _} = result, true, opts) do log_prompt_activity(result, "prompt.updated", opts) end defp maybe_log_prompt_update(result, _has_changes, _opts), do: result @doc """ Deletes an AI prompt. """ @spec delete_prompt(Prompt.t(), keyword()) :: {:ok, Prompt.t()} | {:error, Ecto.Changeset.t()} def delete_prompt(%Prompt{} = prompt, opts \\ []) do repo().delete(prompt) |> broadcast_prompt_change(:prompt_deleted) |> log_prompt_activity("prompt.deleted", opts) |> log_failed_prompt_mutation("prompt.deleted", opts) end @doc """ Returns a prompt changeset for use in forms. """ @spec change_prompt(Prompt.t(), map()) :: Ecto.Changeset.t() def change_prompt(%Prompt{} = prompt, attrs \\ %{}) do Prompt.changeset(prompt, attrs) end @doc """ Increments the usage count for a prompt and updates last_used_at. """ @spec record_prompt_usage(Prompt.t()) :: {:ok, Prompt.t()} | {:error, Ecto.Changeset.t()} def record_prompt_usage(%Prompt{} = prompt) do prompt |> Prompt.usage_changeset() |> repo().update() end @doc """ Counts the total number of prompts. """ @spec count_prompts() :: non_neg_integer() def count_prompts do repo().aggregate(Prompt, :count) end @doc """ Counts the number of enabled prompts. """ @spec count_enabled_prompts() :: non_neg_integer() def count_enabled_prompts do query = from(p in Prompt, where: p.enabled == true) repo().aggregate(query, :count) end @doc """ Resolves a prompt from various input types. Accepts: - UUID string (e.g., "019abc12-3456-7def-8901-234567890abc") - String slug (e.g., "my-prompt") - Prompt struct (returned as-is) Returns `{:ok, prompt}` or `{:error, reason}`. """ def resolve_prompt(%Prompt{} = prompt), do: {:ok, prompt} def resolve_prompt(id_or_slug) when is_binary(id_or_slug) do if textual_uuid?(id_or_slug) do # It's a UUID case get_prompt(id_or_slug) do nil -> {:error, {:prompt_error, :not_found}} prompt -> {:ok, prompt} end else # It's a slug case get_prompt_by_slug(id_or_slug) do nil -> {:error, {:prompt_error, :not_found}} prompt -> {:ok, prompt} end end end def resolve_prompt(_), do: {:error, {:prompt_error, :invalid_identifier}} @doc """ Renders a prompt by replacing variables with provided values. Returns `{:ok, rendered_text}` or `{:error, reason}`. """ def render_prompt(prompt_uuid, variables \\ %{}) do with {:ok, prompt} <- resolve_prompt(prompt_uuid) do Prompt.render(prompt, variables) end end @doc """ Increments the usage count for a prompt and updates last_used_at. """ def increment_prompt_usage(prompt_uuid) do with {:ok, prompt} <- resolve_prompt(prompt_uuid) do record_prompt_usage(prompt) end end @doc """ Makes an AI completion using a prompt template. The prompt content is rendered with the provided variables and sent as the user message. """ def ask_with_prompt(endpoint_uuid, prompt_uuid, variables \\ %{}, opts \\ []) do with {:ok, prompt} <- resolve_prompt(prompt_uuid), {:ok, _} <- validate_prompt(prompt), {:ok, rendered} <- Prompt.render(prompt, variables), {:ok, system_prompt} <- Prompt.render_system_prompt(prompt, variables) do # Pass prompt info to ask for request logging opts_with_prompt = opts |> Keyword.put(:prompt_uuid, prompt.uuid) |> Keyword.put(:prompt_name, prompt.name) # Include system prompt if the prompt template defines one opts_with_prompt = if system_prompt do Keyword.put_new(opts_with_prompt, :system, system_prompt) else opts_with_prompt end case ask(endpoint_uuid, rendered, opts_with_prompt) do {:ok, response} -> # Only increment usage on successful completion increment_prompt_usage(prompt_uuid) {:ok, response} error -> error end end end @doc """ Makes an AI completion with a prompt template as the system message. The prompt is rendered and used as the system message, with the user_message as the user message. """ def complete_with_system_prompt(endpoint_uuid, prompt_uuid, variables, user_message, opts \\ []) do with {:ok, prompt} <- resolve_prompt(prompt_uuid), {:ok, _} <- validate_prompt(prompt), {:ok, system_prompt} <- Prompt.render(prompt, variables) do # Build messages with system prompt messages = [ %{role: "system", content: system_prompt}, %{role: "user", content: user_message} ] # Pass prompt info to complete for request logging opts_with_prompt = opts |> Keyword.put(:prompt_uuid, prompt.uuid) |> Keyword.put(:prompt_name, prompt.name) case complete(endpoint_uuid, messages, opts_with_prompt) do {:ok, response} -> # Only increment usage on successful completion increment_prompt_usage(prompt_uuid) {:ok, response} error -> error end end end @doc """ Validates that a prompt is ready for use. Returns `{:ok, prompt}` if valid, or `{:error, reason}` if not. """ def validate_prompt(prompt) do cond do prompt.content == nil or prompt.content == "" -> {:error, {:prompt_error, :empty_content}} prompt.enabled == false -> {:error, {:prompt_error, :disabled}} true -> {:ok, prompt} end end @doc """ Duplicates a prompt with a new name. """ @spec duplicate_prompt(String.t(), String.t()) :: {:ok, Prompt.t()} | {:error, term()} def duplicate_prompt(prompt_uuid, new_name) when is_binary(new_name) do with {:ok, prompt} <- resolve_prompt(prompt_uuid) do create_prompt(%{ name: new_name, description: prompt.description, content: prompt.content, enabled: prompt.enabled, sort_order: prompt.sort_order, metadata: prompt.metadata }) end end @doc """ Enables a prompt. """ @spec enable_prompt(String.t()) :: {:ok, Prompt.t()} | {:error, term()} def enable_prompt(prompt_uuid) do with {:ok, prompt} <- resolve_prompt(prompt_uuid) do update_prompt(prompt, %{enabled: true}) end end @doc """ Disables a prompt. """ @spec disable_prompt(String.t()) :: {:ok, Prompt.t()} | {:error, term()} def disable_prompt(prompt_uuid) do with {:ok, prompt} <- resolve_prompt(prompt_uuid) do update_prompt(prompt, %{enabled: false}) end end @doc """ Gets the variables defined in a prompt. """ def get_prompt_variables(prompt_uuid) do with {:ok, prompt} <- resolve_prompt(prompt_uuid) do {:ok, prompt.variables || []} end end @doc """ Previews a rendered prompt without making an AI call. """ def preview_prompt(prompt_uuid, variables \\ %{}) do render_prompt(prompt_uuid, variables) end @doc """ Validates that all required variables are provided for a prompt. """ def validate_prompt_variables(prompt_uuid, variables) do with {:ok, prompt} <- resolve_prompt(prompt_uuid) do Prompt.validate_variables(prompt, variables) end end @doc """ Searches prompts by name, description, or content. """ def search_prompts(query, opts \\ []) when is_binary(query) do pattern = "%#{query}%" limit = Keyword.get(opts, :limit, 50) base_query = from(p in Prompt, where: ilike(p.name, ^pattern) or ilike(p.description, ^pattern) or ilike(p.content, ^pattern), order_by: [asc: p.sort_order, desc: p.inserted_at], limit: ^limit ) base_query = case Keyword.get(opts, :enabled) do nil -> base_query enabled -> where(base_query, [p], p.enabled == ^enabled) end repo().all(base_query) end @doc """ Finds all prompts that use a specific variable. """ def get_prompts_with_variable(variable_name) when is_binary(variable_name) do query = from(p in Prompt, where: ^variable_name in p.variables, order_by: [asc: p.sort_order, desc: p.inserted_at] ) repo().all(query) end @doc """ Validates that the content has valid variable syntax. """ def validate_prompt_content(content) when is_binary(content) do all_patterns = Regex.scan(~r/\{\{([^}]+)\}\}/, content) invalid = all_patterns |> Enum.map(fn [_full, inner] -> inner end) |> Enum.reject(fn inner -> Regex.match?(~r/^\w+$/, inner) end) if Enum.empty?(invalid) do :ok else {:error, invalid} end end def validate_prompt_content(_), do: {:error, {:prompt_error, :content_not_string}} @doc """ Gets usage statistics for all prompts. """ def get_prompt_usage_stats(opts \\ []) do query = from(p in Prompt, select: %{ prompt: p, usage_count: p.usage_count, last_used_at: p.last_used_at }, order_by: [desc: p.usage_count, desc: p.last_used_at] ) query = case Keyword.get(opts, :enabled) do nil -> query enabled -> where(query, [p], p.enabled == ^enabled) end query = case Keyword.get(opts, :limit) do nil -> query limit -> limit(query, ^limit) end repo().all(query) end @doc """ Resets the usage statistics for a prompt. """ def reset_prompt_usage(prompt_uuid) do with {:ok, prompt} <- resolve_prompt(prompt_uuid) do prompt |> Ecto.Changeset.change(%{usage_count: 0, last_used_at: nil}) |> repo().update() end end @doc """ Updates the sort order for multiple prompts. Accepts prompt UUIDs. """ def reorder_prompts(order_list) when is_list(order_list) do repo().transaction(fn -> Enum.each(order_list, fn {id, sort_order} -> build_prompt_uuid_query(id) |> repo().update_all(set: [sort_order: sort_order]) end) end) :ok end defp build_prompt_uuid_query(id) when is_binary(id) do if textual_uuid?(id) do from(p in Prompt, where: p.uuid == ^id) else from(p in Prompt, where: false) end end defp build_prompt_uuid_query(_), do: from(p in Prompt, where: false) # =========================================== # USAGE TRACKING (REQUESTS) # =========================================== @doc """ Lists AI requests with pagination and filters. ## Options - `:page` - Page number (default: 1) - `:page_size` - Results per page (default: 20) - `:endpoint_uuid` - Filter by endpoint - `:user_uuid` - Filter by user - `:status` - Filter by status - `:model` - Filter by model - `:source` - Filter by source (from metadata) - `:since` - Filter by date (requests after this date) - `:preload` - Associations to preload ## Returns `{requests, total_count}` """ def list_requests(opts \\ []) do page = Keyword.get(opts, :page, 1) page_size = Keyword.get(opts, :page_size, 20) offset = (page - 1) * page_size sort_by = Keyword.get(opts, :sort_by, :inserted_at) sort_dir = Keyword.get(opts, :sort_dir, :desc) base_query = from(r in Request) base_query = apply_request_filters(base_query, opts) base_query = apply_request_sorting(base_query, sort_by, sort_dir) total = repo().aggregate(base_query, :count) query = base_query |> limit(^page_size) |> offset(^offset) query = case Keyword.get(opts, :preload) do nil -> query preloads -> preload(query, ^preloads) end requests = repo().all(query) {requests, total} end defp apply_request_sorting(query, field, dir) when field in [ :inserted_at, :model, :total_tokens, :latency_ms, :cost_cents, :status, :endpoint_name ] do order_by(query, [r], [{^dir, field(r, ^field)}]) end defp apply_request_sorting(query, _field, _dir) do order_by(query, [r], desc: r.inserted_at) end @doc """ Gets a single request by UUID. """ def get_request!(id) do case get_request(id) do nil -> raise Ecto.NoResultsError, queryable: Request request -> request end end @doc """ Gets a single request by UUID. Accepts a UUID string (e.g., "550e8400-e29b-41d4-a716-446655440000"). Returns `nil` if the request does not exist. """ def get_request(id) when is_binary(id) do if textual_uuid?(id) do repo().get_by(Request, uuid: id) else nil end end def get_request(_), do: nil @doc """ Creates a new AI request record. Used to log every AI API call for tracking and statistics. """ def create_request(attrs) do %Request{} |> Request.changeset(attrs) |> repo().insert() |> broadcast_request_change(:request_created) end @doc """ Counts the total number of requests. """ @spec count_requests() :: non_neg_integer() def count_requests do repo().aggregate(Request, :count) end @doc """ Sums the total tokens used across all requests. """ @spec sum_tokens() :: non_neg_integer() def sum_tokens do repo().aggregate(Request, :sum, :total_tokens) || 0 end defp apply_request_filters(query, opts) do query |> maybe_filter_by(:endpoint_uuid, Keyword.get(opts, :endpoint_uuid)) |> maybe_filter_by(:user_uuid, Keyword.get(opts, :user_uuid)) |> maybe_filter_by(:status, Keyword.get(opts, :status)) |> maybe_filter_by(:model, Keyword.get(opts, :model)) |> maybe_filter_by(:source, Keyword.get(opts, :source)) |> maybe_filter_since(Keyword.get(opts, :since)) end defp maybe_filter_by(query, _field, nil), do: query defp maybe_filter_by(query, :endpoint_uuid, uuid) when is_binary(uuid) do where(query, [r], r.endpoint_uuid == ^uuid) end defp maybe_filter_by(query, :user_uuid, uuid) when is_binary(uuid) do where(query, [r], r.user_uuid == ^uuid) end defp maybe_filter_by(query, :status, status), do: where(query, [r], r.status == ^status) defp maybe_filter_by(query, :model, model), do: where(query, [r], r.model == ^model) defp maybe_filter_by(query, :source, source), do: where(query, [r], fragment("?->>'source' = ?", r.metadata, ^source)) defp maybe_filter_since(query, nil), do: query defp maybe_filter_since(query, date), do: where(query, [r], r.inserted_at >= ^date) @doc """ Returns filter options for requests (distinct endpoints, models, and sources). """ def get_request_filter_options do endpoints_query = from(r in Request, where: not is_nil(r.endpoint_uuid) and not is_nil(r.endpoint_name), distinct: true, select: {r.endpoint_uuid, r.endpoint_name}, order_by: r.endpoint_name ) models_query = from(r in Request, where: not is_nil(r.model), distinct: r.model, select: r.model, order_by: r.model ) # Query unique sources from metadata JSONB field sources_query = from(r in Request, where: not is_nil(fragment("?->>'source'", r.metadata)), distinct: fragment("?->>'source'", r.metadata), select: fragment("?->>'source'", r.metadata), order_by: fragment("?->>'source'", r.metadata) ) %{ endpoints: repo().all(endpoints_query), models: repo().all(models_query), statuses: Request.valid_statuses(), sources: repo().all(sources_query) } end # =========================================== # STATISTICS # =========================================== @doc """ Gets aggregated usage statistics. ## Options - `:since` - Start date for statistics - `:until` - End date for statistics - `:endpoint_uuid` - Filter by endpoint ## Returns Map with statistics including total_requests, total_tokens, success_rate, etc. """ def get_usage_stats(opts \\ []) do base_query = from(r in Request) base_query = apply_request_filters(base_query, opts) total_requests = repo().aggregate(base_query, :count) total_tokens = repo().aggregate(base_query, :sum, :total_tokens) || 0 total_cost = repo().aggregate(base_query, :sum, :cost_cents) || 0 avg_latency = repo().aggregate(base_query, :avg, :latency_ms) success_query = where(base_query, [r], r.status == "success") success_count = repo().aggregate(success_query, :count) success_rate = if total_requests > 0 do Float.round(success_count / total_requests * 100, 1) else 0.0 end %{ total_requests: total_requests, total_tokens: total_tokens, total_cost_cents: total_cost, success_count: success_count, error_count: total_requests - success_count, success_rate: success_rate, avg_latency_ms: decimal_to_int(avg_latency) } end # Convert Decimal or number to integer, handling nil defp decimal_to_int(nil), do: nil defp decimal_to_int(%Decimal{} = d), do: d |> Decimal.round() |> Decimal.to_integer() defp decimal_to_int(n) when is_float(n), do: round(n) defp decimal_to_int(n) when is_integer(n), do: n @doc """ Gets dashboard statistics for display. Returns stats for the last 30 days plus all-time totals. """ def get_dashboard_stats do thirty_days_ago = UtilsDate.utc_now() |> DateTime.add(-30, :day) today_start = Date.utc_today() |> DateTime.new!(~T[00:00:00], "Etc/UTC") all_time = get_usage_stats() last_30_days = get_usage_stats(since: thirty_days_ago) today = get_usage_stats(since: today_start) tokens_by_model = get_tokens_by_model(since: thirty_days_ago) requests_by_day = get_requests_by_day(since: thirty_days_ago) %{ all_time: all_time, last_30_days: last_30_days, today: today, tokens_by_model: tokens_by_model, requests_by_day: requests_by_day } end @doc """ Gets token usage grouped by model. """ def get_tokens_by_model(opts \\ []) do base_query = from(r in Request) base_query = apply_request_filters(base_query, opts) query = from(r in subquery(base_query), where: not is_nil(r.model) and r.model != "", group_by: r.model, select: %{ model: r.model, total_tokens: sum(r.total_tokens), request_count: count() }, order_by: [desc: sum(r.total_tokens)] ) repo().all(query) end @doc """ Gets request counts grouped by day. """ def get_requests_by_day(opts \\ []) do base_query = from(r in Request) base_query = apply_request_filters(base_query, opts) query = from(r in subquery(base_query), group_by: fragment("DATE(?)", r.inserted_at), select: %{ date: fragment("DATE(?)", r.inserted_at), count: count(), tokens: sum(r.total_tokens) }, order_by: [asc: fragment("DATE(?)", r.inserted_at)] ) repo().all(query) end # =========================================== # COMPLETION API # =========================================== alias PhoenixKitAI.Completion @doc """ Makes a chat completion request using a configured endpoint. ## Parameters - `endpoint_uuid` - Endpoint UUID string or Endpoint struct - `messages` - List of message maps with `:role` and `:content` - `opts` - Optional parameter overrides ## Options All standard completion parameters plus: - `:source` - Override auto-detected source for request tracking ## Examples {:ok, response} = PhoenixKitAI.complete(endpoint_uuid, [ %{role: "user", content: "Hello!"} ]) # With system message {:ok, response} = PhoenixKitAI.complete(endpoint_uuid, [ %{role: "system", content: "You are a helpful assistant."}, %{role: "user", content: "What is 2+2?"} ]) # With parameter overrides {:ok, response} = PhoenixKitAI.complete(endpoint_uuid, messages, temperature: 0.5, max_tokens: 500 ) # With custom source for tracking {:ok, response} = PhoenixKitAI.complete(endpoint_uuid, messages, source: "MyModule" ) ## Returns - `{:ok, response}` - Full API response including usage stats - `{:error, reason}` - Error atom or tagged tuple. See `PhoenixKitAI.Errors` for the vocabulary and translation. """ @spec complete(String.t() | Endpoint.t(), list(map()), keyword()) :: {:ok, map()} | {:error, term()} def complete(endpoint_uuid, messages, opts \\ []) do with {:ok, endpoint} <- resolve_endpoint(endpoint_uuid), {:ok, _} <- validate_endpoint(endpoint) do # Capture caller info (source + stacktrace + context) {auto_source, stacktrace, caller_context} = capture_caller_info() # Allow manual override of source, but all debug info is always captured source = Keyword.get(opts, :source) || auto_source # Extract prompt info if present (from ask_with_prompt, complete_with_system_prompt) prompt_info = %{ prompt_uuid: Keyword.get(opts, :prompt_uuid), prompt_name: Keyword.get(opts, :prompt_name) } merged_opts = merge_endpoint_opts(endpoint, opts) case Completion.chat_completion(endpoint, messages, merged_opts) do {:ok, response} -> log_request( endpoint, messages, response, source, stacktrace, caller_context, prompt_info ) {:ok, response} {:error, reason} -> log_failed_request( endpoint, messages, reason, source, stacktrace, caller_context, prompt_info ) {:error, reason} end end end @doc """ Simple helper for single-turn chat completion. ## Parameters - `endpoint_uuid` - Endpoint UUID string or Endpoint struct - `prompt` - User prompt string - `opts` - Optional parameter overrides and system message ## Options All options from `complete/3` plus: - `:system` - System message string - `:source` - Override auto-detected source for request tracking ## Examples # Simple question {:ok, response} = PhoenixKitAI.ask(endpoint_uuid, "What is the capital of France?") # With system message {:ok, response} = PhoenixKitAI.ask(endpoint_uuid, "Translate: Hello", system: "You are a translator. Translate to French." ) # With custom source for tracking {:ok, response} = PhoenixKitAI.ask(endpoint_uuid, "Hello!", source: "Languages" ) # Extract just the text content {:ok, response} = PhoenixKitAI.ask(endpoint_uuid, "Hello!") {:ok, text} = PhoenixKitAI.extract_content(response) ## Returns Same as `complete/3` """ @spec ask(String.t() | Endpoint.t(), String.t(), keyword()) :: {:ok, map()} | {:error, term()} def ask(endpoint_uuid, prompt, opts \\ []) when is_binary(prompt) do {system, opts} = Keyword.pop(opts, :system) messages = case system do nil -> [%{role: "user", content: prompt}] sys -> [%{role: "system", content: sys}, %{role: "user", content: prompt}] end complete(endpoint_uuid, messages, opts) end @doc """ Makes an embeddings request using a configured endpoint. ## Parameters - `endpoint_uuid` - Endpoint UUID string or Endpoint struct - `input` - Text or list of texts to embed - `opts` - Optional parameter overrides ## Options - `:dimensions` - Override embedding dimensions - `:source` - Override auto-detected source for request tracking ## Examples # Single text {:ok, response} = PhoenixKitAI.embed(endpoint_uuid, "Hello, world!") # Multiple texts {:ok, response} = PhoenixKitAI.embed(endpoint_uuid, ["Hello", "World"]) # With dimension override {:ok, response} = PhoenixKitAI.embed(endpoint_uuid, "Hello", dimensions: 512) # With custom source for tracking {:ok, response} = PhoenixKitAI.embed(endpoint_uuid, "Hello", source: "SemanticSearch" ) ## Returns - `{:ok, response}` - Response with embeddings - `{:error, reason}` - Error atom or tagged tuple. """ @spec embed(String.t() | Endpoint.t(), String.t() | list(String.t()), keyword()) :: {:ok, map()} | {:error, term()} def embed(endpoint_uuid, input, opts \\ []) do with {:ok, endpoint} <- resolve_endpoint(endpoint_uuid), {:ok, _} <- validate_endpoint(endpoint) do # Capture caller info (source + stacktrace + context) {auto_source, stacktrace, caller_context} = capture_caller_info() # Allow manual override of source, but all debug info is always captured source = Keyword.get(opts, :source) || auto_source merged_opts = merge_embedding_opts(endpoint, opts) case Completion.embeddings(endpoint, input, merged_opts) do {:ok, response} -> log_embedding_request(endpoint, input, response, source, stacktrace, caller_context) {:ok, response} {:error, reason} -> log_failed_embedding_request(endpoint, reason, source, stacktrace, caller_context) {:error, reason} end end end @doc """ Extracts the text content from a completion response. ## Examples {:ok, response} = PhoenixKitAI.ask(endpoint_uuid, "Hello!") {:ok, text} = PhoenixKitAI.extract_content(response) # => "Hello! How can I help you today?" """ defdelegate extract_content(response), to: Completion @doc """ Extracts usage information from a response. ## Examples {:ok, response} = PhoenixKitAI.complete(endpoint_uuid, messages) usage = PhoenixKitAI.extract_usage(response) # => %{prompt_tokens: 10, completion_tokens: 15, total_tokens: 25} """ defdelegate extract_usage(response), to: Completion # Private helpers for completion API defp validate_endpoint(endpoint) do cond do endpoint.model == nil or endpoint.model == "" -> {:error, :endpoint_no_model} endpoint.enabled == false -> {:error, :endpoint_disabled} true -> case endpoint_credential_status(endpoint) do :ok -> {:ok, endpoint} {:error, _} = err -> err end end end # Mirrors the lookup ladder that `OpenRouterClient.resolve_api_key/1` # walks at request time so validation can't disagree with the actual # credential resolution: integration_uuid first, then the legacy # `provider` column (which carried a uuid pre-V107), then the legacy # `api_key` column. Validation only fails when ALL three sources are # empty — same as the request path. The error reason distinguishes # between "you pinned an integration that was deleted" (orphan) and # "you never wired anything up" so the user-facing message is honest. # # **Intentional behaviour change vs pre-strict-UUID** (PR #6 review # finding #2): the old shape short-circuited on # `not Integrations.connected?(endpoint.provider)` — an integration # with stored credentials but in a not-`"connected"` state would # fail validation pre-API. The new ladder does NOT consult # `connected?/1`: as long as a credential exists in any of the # three sources, we let the upstream provider be the source of # truth on whether the key actually authenticates. This matches # how the request path resolves credentials — validation only # disagrees with the eventual request when one of the three # sources changes between validate and dispatch, which is a # narrower window than the old shape's "stale `connected?` state" # window. A row that's `"error"` or `"configured"` (creds present, # never validated) now reaches the upstream API and gets a real # 401/403 instead of a pre-emptive `:integration_not_configured`. # If a caller specifically wants "is this key healthy right now", # they should call `PhoenixKit.Integrations.validate_connection/2` # directly — `endpoint_credential_status/1` is the # `is this endpoint dispatchable` check, not a health check. defp endpoint_credential_status(endpoint) do cond do match?({:ok, _}, lookup_credentials(endpoint.integration_uuid)) -> :ok match?({:ok, _}, lookup_credentials(endpoint.provider)) -> :ok is_binary(endpoint.api_key) and endpoint.api_key != "" -> :ok is_binary(endpoint.integration_uuid) and endpoint.integration_uuid != "" -> {:error, :integration_deleted} true -> {:error, :integration_not_configured} end end defp lookup_credentials(nil), do: {:error, :not_configured} defp lookup_credentials(""), do: {:error, :not_configured} defp lookup_credentials(key) when is_binary(key) do PhoenixKit.Integrations.get_credentials(key) end defp merge_endpoint_opts(endpoint, opts) do # Endpoint defaults, then user overrides base_opts = [ temperature: endpoint.temperature, max_tokens: endpoint.max_tokens, top_p: endpoint.top_p, top_k: endpoint.top_k, frequency_penalty: endpoint.frequency_penalty, presence_penalty: endpoint.presence_penalty, repetition_penalty: endpoint.repetition_penalty, stop: endpoint.stop, seed: endpoint.seed, # Reasoning/thinking parameters (for models like DeepSeek R1, Qwen QwQ) reasoning_enabled: endpoint.reasoning_enabled, reasoning_effort: endpoint.reasoning_effort, reasoning_max_tokens: endpoint.reasoning_max_tokens, reasoning_exclude: endpoint.reasoning_exclude ] # Filter out nil values and merge with user opts base_opts |> Enum.reject(fn {_k, v} -> is_nil(v) end) |> Keyword.merge(opts) end defp merge_embedding_opts(endpoint, opts) do base_opts = [dimensions: endpoint.dimensions] base_opts |> Enum.reject(fn {_k, v} -> is_nil(v) end) |> Keyword.merge(opts) end # =========================================== # CALLER INFO CAPTURE (for source tracking & debugging) # =========================================== @doc false # Captures full debug context: source, stacktrace, and caller context. # # Memory capture is opt-in via `config :phoenix_kit_ai, capture_request_memory: true` # to avoid bloating JSONB metadata on every request. defp capture_caller_info do keys = if Application.get_env(:phoenix_kit_ai, :capture_request_memory, false) do [:current_stacktrace, :memory] else [:current_stacktrace] end info = Process.info(self(), keys) stack = Keyword.fetch!(info, :current_stacktrace) memory = Keyword.get(info, :memory) # Format stacktrace for storage formatted_stack = format_stacktrace(stack) # Extract clean source from first non-internal caller source = extract_source(stack) # Build caller context with additional debug info caller_context = build_caller_context(memory) {source, formatted_stack, caller_context} end defp format_stacktrace(stack) do stack # Limit depth for storage |> Enum.take(20) |> Enum.map(fn {mod, fun, arity, location} -> mod_str = Atom.to_string(mod) |> String.replace_prefix("Elixir.", "") file = Keyword.get(location, :file, ~c"unknown") |> to_string() line = Keyword.get(location, :line, 0) "#{mod_str}.#{fun}/#{arity} (#{file}:#{line})" end) end defp extract_source(stack) do # Modules to skip (PhoenixKitAI internals, Elixir/Erlang core) skip_prefixes = ["PhoenixKitAI", "Elixir.PhoenixKitAI"] skip_modules = [Process, :proc_lib, :gen_server, :gen, :elixir, :erl_eval] caller = Enum.find(stack, fn {mod, _fun, _arity, _loc} -> mod_str = Atom.to_string(mod) not Enum.any?(skip_prefixes, &String.starts_with?(mod_str, &1)) and mod not in skip_modules end) case caller do {mod, fun, _arity, _loc} -> mod_str = Atom.to_string(mod) |> String.replace_prefix("Elixir.", "") "#{mod_str}.#{fun}" nil -> nil end end defp build_caller_context(memory) do # Get Phoenix request_id from Logger metadata (if in request context) logger_meta = Logger.metadata() base = %{ request_id: Keyword.get(logger_meta, :request_id), node: node() |> Atom.to_string(), pid: self() |> inspect() } if is_integer(memory), do: Map.put(base, :memory_bytes, memory), else: base end # =========================================== # REQUEST LOGGING # =========================================== defp log_request(endpoint, messages, response, source, stacktrace, caller_context, prompt_info) do usage = Completion.extract_usage(response) # Extract response content response_content = case Completion.extract_content(response) do {:ok, content} -> content _ -> nil end # Reasoning models (DeepSeek-R1, Mistral Magistral, OpenAI o-series, # Anthropic extended thinking) return the chain-of-thought alongside # the final answer in a `reasoning` / `reasoning_content` / # `thinking` field. Capture it next to `response` so the request # detail page can surface it later — currently dropped on the floor # when only `content` is extracted. Subject to the same # `capture_request_content?` gate as `response` (see below) — when # content capture is off, reasoning is dropped too. Reasoning can # contain anything the prompt mentioned, so it's PII-equivalent. response_reasoning = Completion.extract_reasoning(response) # User-content persistence is opt-out. Default `true` preserves the # debugging shape we've shipped so far; deployments with PII or # data-retention obligations can flip it off via # `config :phoenix_kit_ai, capture_request_content: false`. Token # counts, latency, model, and cost are always recorded. capture_content = capture_request_content?() normalized = if capture_content, do: normalize_messages(messages), else: nil base_metadata = %{ temperature: endpoint.temperature, max_tokens: endpoint.max_tokens, # Debug context (source tracking) source: source, stacktrace: stacktrace, caller_context: caller_context } metadata = if capture_content do request_payload = %{ model: endpoint.model, messages: normalized, temperature: endpoint.temperature, max_tokens: endpoint.max_tokens } Map.merge(base_metadata, %{ messages: normalized, response: response_content, response_reasoning: response_reasoning, request_payload: request_payload }) else Map.put(base_metadata, :content_redacted, true) end create_request(%{ endpoint_uuid: endpoint.uuid, endpoint_name: endpoint.name, prompt_uuid: prompt_info[:prompt_uuid], prompt_name: prompt_info[:prompt_name], model: endpoint.model, request_type: "chat", input_tokens: usage.prompt_tokens, output_tokens: usage.completion_tokens, total_tokens: usage.total_tokens, cost_cents: usage.cost_cents, latency_ms: response["latency_ms"], status: "success", metadata: metadata }) end defp log_failed_request( endpoint, messages, reason, source, stacktrace, caller_context, prompt_info ) do capture_content = capture_request_content?() metadata = %{ # Original reason atom/tuple — preserved alongside the # human-readable error_message so callers can still filter on # the machine-readable shape. error_reason: inspect(reason), # Debug context (source tracking) source: source, stacktrace: stacktrace, caller_context: caller_context } |> maybe_add_content(:messages, capture_content, fn -> normalize_messages(messages) end) create_request(%{ endpoint_uuid: endpoint.uuid, endpoint_name: endpoint.name, prompt_uuid: prompt_info[:prompt_uuid], prompt_name: prompt_info[:prompt_name], model: endpoint.model, request_type: "chat", status: "error", error_message: error_reason_to_string(reason), metadata: metadata }) end defp maybe_add_content(metadata, _key, false, _build), do: Map.put(metadata, :content_redacted, true) defp maybe_add_content(metadata, key, true, build), do: Map.put(metadata, key, build.()) # Render a `{:error, reason}` value into a string suitable for the # `error_message` :string column. `Errors.message/1` is total — atoms, # tagged tuples, and strings all collapse to a translated string. defp error_reason_to_string(reason), do: PhoenixKitAI.Errors.message(reason) # Whether to persist user/assistant message content to request metadata. # Defaults to `true` for parity with the shipped behaviour. Deployments # with retention or PII concerns set # `config :phoenix_kit_ai, capture_request_content: false`. defp capture_request_content? do Application.get_env(:phoenix_kit_ai, :capture_request_content, true) end # Normalize messages to ensure consistent format for storage defp normalize_messages(messages) do Enum.map(messages, fn msg -> %{ "role" => to_string(msg[:role] || msg["role"]), "content" => msg[:content] || msg["content"] } end) end defp log_embedding_request(endpoint, input, response, source, stacktrace, caller_context) do usage = Completion.extract_usage(response) input_count = if is_list(input), do: length(input), else: 1 create_request(%{ endpoint_uuid: endpoint.uuid, endpoint_name: endpoint.name, model: endpoint.model, request_type: "embedding", input_tokens: usage.prompt_tokens, total_tokens: usage.total_tokens, cost_cents: usage.cost_cents, latency_ms: response["latency_ms"], status: "success", metadata: %{ input_count: input_count, dimensions: endpoint.dimensions, # Debug context (source tracking) source: source, stacktrace: stacktrace, caller_context: caller_context } }) end defp log_failed_embedding_request(endpoint, reason, source, stacktrace, caller_context) do create_request(%{ endpoint_uuid: endpoint.uuid, endpoint_name: endpoint.name, model: endpoint.model, request_type: "embedding", status: "error", error_message: error_reason_to_string(reason), metadata: %{ error_reason: inspect(reason), # Debug context (source tracking) source: source, stacktrace: stacktrace, caller_context: caller_context } }) end end