feat(alerts): add saved reset first-seen alerts

Add configurable saved-reset first-seen alert rules backed by persisted metadata only.

Record saved-reset events once per upstream/reset expiration while preserving impacted pool targets and one delivery per channel. Expose the rule in admin alerts, safe email/webhook summaries, read models, and operator docs.

Author: masterkain

docs-site/src/content/docs/operators/alerts.mdx

@@ -62,6 +62,14 @@ Common upstream states include:
 
 When several upstreams are affected, summarize counts by state and Pool. Don't publish account identifiers, raw emails, or provider credential details.
 
+### Saved Reset First-Seen Alerts
+
+The `upstream_saved_reset_banked_first_seen` rule is an optional alert rule that operators configure from the Alerts surface. It is not implicit or always-on. When enabled, it watches persisted saved-reset metadata and opens a once-only incident for an upstream identity plus reset expiration the first time that expiration is seen for a matching Pool assignment.
+
+Evaluation is metadata-only. It reads stored upstream identity metadata and Pool assignments; it does not call provider APIs or refresh account data during alert evaluation. Incidents can name safe labels such as `example-pool` and `example-upstream`, sanitized expiration timestamps, rule kind, severity, and first-seen timing. They must not include raw provider credit objects, request or response bodies, bearer tokens, webhook secrets, prompts, account credentials, or other private content.
+
+If the same upstream identity affects multiple Pools, incident visibility follows impacted Pool targets and the current operator authorization model. Owners can see all impacted Pools; assigned admins see only the impacted Pools they are allowed to manage plus safe visible, hidden, and total counts.
+
 ## Pool API key signals
 
 Pool API key alerts are about runtime client access. A client can fail before routing if the key is paused, revoked, expired, scoped to the wrong Pool, or blocked by policy.

docs-site/src/content/docs/operators/upstreams.mdx

@@ -104,6 +104,8 @@ Auto redemption is a per-upstream opt-in policy available from the account-card
 
 Saved reset observations and redemption attempts stay metadata-only on `upstream_identities.metadata` under `saved_resets` and `saved_reset_redemption`. The saved-reset observation may include sanitized expiration rows in `available_expirations`, where each row has an `expires_at` timestamp and a pooler-owned `first_seen_at` timestamp. Legacy `available_expires_at` stays available for compatibility, and `next_expires_at`, `expires_observed_at`, and `expires_refresh_attempted_at` remain sanitized aggregate fields. Historical rows are backfilled with the migration time as `first_seen_at`; future observations preserve `first_seen_at` only while the same expiration remains present, and disappeared expirations are not retained. Missing or invalid provider expiration fields are not guessed.
 
+Operators can configure the optional `upstream_saved_reset_banked_first_seen` alert rule to notify once per upstream identity and saved-reset expiration when persisted metadata first reports a banked reset for a matching Pool assignment. The alert uses the same metadata-only observation boundary as the saved-reset UI; it does not call provider APIs during evaluation.
+
 Policy fields live on the upstream identity as `saved_reset_auto_redeem_enabled`, `saved_reset_auto_redeem_trigger_mode`, `saved_reset_auto_redeem_quota_threshold_percent`, `saved_reset_auto_redeem_min_blocked_minutes`, and `saved_reset_auto_redeem_keep_credits`. Do not copy provider payloads, raw provider credit objects, raw credit identifiers, titles, descriptions, request ids, redemption request identifiers, token material, secrets, or account credentials into metadata, logs, docs, or tests.
 
 ## Readiness States

lib/codex_pooler/admin/alert_notification_query.ex

@@ -22,6 +22,9 @@ defmodule CodexPooler.Admin.AlertNotificationQuery do
           required(:severity) => String.t(),
           required(:state) => String.t(),
           required(:impacted_pools) => [impacted_pool()],
+          required(:visible_impacted_pool_count) => non_neg_integer(),
+          required(:hidden_impacted_pool_count) => non_neg_integer(),
+          required(:total_impacted_pool_count) => non_neg_integer(),
           required(:last_seen_at) => DateTime.t(),
           required(:unread?) => boolean()
         }
@@ -110,14 +113,21 @@ defmodule CodexPooler.Admin.AlertNotificationQuery do
   @spec notification_rows([query_row()], [Ecto.UUID.t()]) :: [row()]
   defp notification_rows(rows, pool_ids) do
     impacted_pools_by_incident = impacted_pools_by_incident(rows, pool_ids)
+    impacted_pool_counts_by_incident = impacted_pool_counts_by_incident(rows, pool_ids)
 
     Enum.map(rows, fn {incident, receipt} ->
+      counts =
+        Map.get(impacted_pool_counts_by_incident, incident.id, empty_impacted_pool_counts())
+
       %{
         id: incident.id,
         rule_kind: incident.rule_kind,
         severity: incident.severity,
         state: incident.state,
         impacted_pools: Map.get(impacted_pools_by_incident, incident.id, []),
+        visible_impacted_pool_count: counts.visible,
+        hidden_impacted_pool_count: counts.hidden,
+        total_impacted_pool_count: counts.total,
         last_seen_at: incident.last_seen_at,
         unread?: Alerts.incident_notification_unread?(incident, receipt)
       }
@@ -147,10 +157,47 @@ defmodule CodexPooler.Admin.AlertNotificationQuery do
     )
     |> Enum.group_by(& &1.incident_id)
     |> Map.new(fn {incident_id, pools} ->
-      {incident_id, Enum.map(pools, &Map.take(&1, [:id, :slug, :name]))}
+      unique_pools = Enum.uniq_by(pools, & &1.id)
+
+      {incident_id, Enum.map(unique_pools, &Map.take(&1, [:id, :slug, :name]))}
+    end)
+  end
+
+  @spec impacted_pool_counts_by_incident([query_row()], [Ecto.UUID.t()]) :: %{
+          optional(Ecto.UUID.t()) => %{
+            required(:visible) => non_neg_integer(),
+            required(:hidden) => non_neg_integer(),
+            required(:total) => non_neg_integer()
+          }
+        }
+  defp impacted_pool_counts_by_incident([], _pool_ids), do: %{}
+
+  defp impacted_pool_counts_by_incident(rows, pool_ids) do
+    incident_ids = rows |> Enum.map(fn {incident, _receipt} -> incident.id end) |> Enum.uniq()
+    visible_pool_ids = MapSet.new(pool_ids)
+
+    Repo.all(
+      from target in AlertIncidentTarget,
+        where: target.incident_id in ^incident_ids,
+        select: %{incident_id: target.incident_id, pool_id: target.pool_id}
+    )
+    |> Enum.group_by(& &1.incident_id)
+    |> Map.new(fn {incident_id, targets} ->
+      unique_pool_ids = targets |> Enum.map(& &1.pool_id) |> MapSet.new()
+      visible_count = Enum.count(unique_pool_ids, &MapSet.member?(visible_pool_ids, &1))
+      total_count = MapSet.size(unique_pool_ids)
+
+      {incident_id,
+       %{
+         visible: visible_count,
+         hidden: max(total_count - visible_count, 0),
+         total: total_count
+       }}
     end)
   end
 
+  defp empty_impacted_pool_counts, do: %{visible: 0, hidden: 0, total: 0}
+
   @spec visible_pool_ids(Scope.t()) :: {:ok, [Ecto.UUID.t()]} | {:error, Alerts.access_error()}
   defp visible_pool_ids(scope) do
     case Alerts.list_manageable_pools(scope) do

lib/codex_pooler/alerts.ex

@@ -100,6 +100,10 @@ defmodule CodexPooler.Alerts do
           IncidentLifecycle.record_result()
   defdelegate record_incident_match(attrs), to: IncidentLifecycle
 
+  @spec record_incident_once(IncidentLifecycle.match_attrs() | map()) ::
+          IncidentLifecycle.record_once_result()
+  defdelegate record_incident_once(attrs), to: IncidentLifecycle
+
   @spec safe_projected_metadata_for_admin(map()) :: map()
   defdelegate safe_projected_metadata_for_admin(metadata), to: IncidentNotifications
 

lib/codex_pooler/alerts/email_delivery.ex

@@ -29,12 +29,17 @@ defmodule CodexPooler.Alerts.EmailDelivery do
   )
   @safe_summary_keys ~w(
     assignment_count
+    available_count
     enabled_assignment_count
     impacted_pool_count
     model
+    path_style
     quota_state
     reason_code
+    reset_expires_at
+    reset_first_seen_at
     routing_usable
+    source
     status
     target_state
     threshold_used_percent

lib/codex_pooler/alerts/evaluation_candidate.ex

@@ -0,0 +1,254 @@
+defmodule CodexPooler.Alerts.EvaluationCandidate do
+  @moduledoc """
+  Builds generic alert evaluation match/clear candidates and stable dedupe keys.
+  """
+
+  alias CodexPooler.Alerts.IncidentLifecycle
+  alias CodexPooler.Alerts.Schemas.AlertRule
+  alias CodexPooler.Upstreams.Quota
+
+  @type action :: :match | :clear
+  @type candidate :: %{
+          required(:action) => action(),
+          required(:dedupe_key) => String.t(),
+          required(:rule_id) => Ecto.UUID.t(),
+          required(:rule_kind) => AlertRule.rule_kind(),
+          optional(:match_attrs) => IncidentLifecycle.match_attrs(),
+          optional(:clear_attrs) => IncidentLifecycle.clear_attrs()
+        }
+
+  @spec pool_match(AlertRule.t(), map(), String.t(), DateTime.t()) :: candidate()
+  def pool_match(%AlertRule{} = rule, projection, reason_code, %DateTime{} = timestamp) do
+    dedupe_key = dedupe_key_for_rule(rule, nil)
+
+    match(rule, dedupe_key, timestamp, %{
+      pool_id: rule.pool_id,
+      safe_evidence_snapshot: %{
+        "reason_code" => reason_code,
+        "pool_id" => rule.pool_id,
+        "model" => rule.model,
+        "assignment_count" => projection.assignment_count,
+        "enabled_assignment_count" => projection.enabled_assignment_count,
+        "usable_assignment_count" => projection.usable_assignment_count,
+        "state_counts" => stringify_count_keys(projection.state_counts),
+        "min_usable_assignments" => rule.min_usable_assignments,
+        "target_state" => rule.target_state
+      },
+      targets: [target(rule, rule.pool_id, %{reason_code: reason_code})]
+    })
+  end
+
+  @spec threshold(AlertRule.t(), map(), DateTime.t()) :: candidate()
+  def threshold(%AlertRule{} = rule, assignment, %DateTime{} = timestamp) do
+    dedupe_key = dedupe_key_for_rule(rule, assignment.upstream_identity_id)
+    threshold = rule.threshold_used_percent || Decimal.new(100)
+    selector = rule.window_selector || "any"
+    windows = selected_windows(assignment.quota_windows, selector, rule.model, timestamp)
+    match = Enum.find(windows, &threshold_match?(&1, threshold, timestamp))
+
+    if match do
+      upstream_match(rule, dedupe_key, assignment, "quota_threshold", timestamp, %{
+        window_selector: selector,
+        threshold_used_percent: decimal_string(threshold),
+        used_percent: decimal_float(match.used_percent),
+        reset_at: iso8601_or_nil(match.reset_at),
+        quota_key: match.quota_key,
+        window_kind: match.window_kind,
+        quota_scope: match.quota_scope,
+        model: match.model,
+        upstream_model: match.upstream_model
+      })
+    else
+      clear(rule, dedupe_key, timestamp)
+    end
+  end
+
+  @spec auth_state(AlertRule.t(), map(), DateTime.t()) :: candidate()
+  def auth_state(%AlertRule{} = rule, assignment, %DateTime{} = timestamp) do
+    dedupe_key = dedupe_key_for_rule(rule, assignment.upstream_identity_id)
+    target_state = rule.target_state
+
+    if assignment.identity_status == target_state do
+      upstream_match(rule, dedupe_key, assignment, target_state, timestamp, %{
+        target_state: target_state,
+        identity_status: assignment.identity_status
+      })
+    else
+      clear(rule, dedupe_key, timestamp)
+    end
+  end
+
+  @spec clear(AlertRule.t(), String.t(), DateTime.t()) :: candidate()
+  def clear(%AlertRule{} = rule, dedupe_key, %DateTime{} = timestamp) do
+    %{
+      action: :clear,
+      dedupe_key: dedupe_key,
+      rule_id: rule.id,
+      rule_kind: rule.rule_kind,
+      clear_attrs: %{dedupe_key: dedupe_key, cleared_at: timestamp}
+    }
+  end
+
+  @spec dedupe_key_for_rule(AlertRule.t(), Ecto.UUID.t() | nil) :: String.t()
+  def dedupe_key_for_rule(
+        %AlertRule{rule_kind: "upstream_quota_threshold"} = rule,
+        upstream_identity_id
+      ) do
+    Enum.join(
+      [
+        "alerts",
+        "v1",
+        rule.rule_kind,
+        "upstream_identity",
+        upstream_identity_id || "none",
+        "window",
+        rule.window_selector || "any",
+        "threshold",
+        decimal_string(rule.threshold_used_percent || Decimal.new(100)),
+        "model",
+        rule.model || "any"
+      ],
+      ":"
+    )
+  end
+
+  def dedupe_key_for_rule(
+        %AlertRule{rule_kind: "upstream_auth_state"} = rule,
+        upstream_identity_id
+      ) do
+    Enum.join(
+      [
+        "alerts",
+        "v1",
+        rule.rule_kind,
+        "upstream_identity",
+        upstream_identity_id || "none",
+        "state",
+        rule.target_state || "unknown"
+      ],
+      ":"
+    )
+  end
+
+  def dedupe_key_for_rule(%AlertRule{} = rule, _upstream_identity_id) do
+    Enum.join(
+      [
+        "alerts",
+        "v1",
+        rule.rule_kind,
+        "pool",
+        rule.pool_id,
+        "model",
+        rule.model || "any",
+        "min",
+        rule.min_usable_assignments || "none",
+        "state",
+        rule.target_state || "none"
+      ],
+      ":"
+    )
+  end
+
+  defp upstream_match(rule, dedupe_key, assignment, reason_code, timestamp, metadata) do
+    match(rule, dedupe_key, timestamp, %{
+      upstream_identity_id: assignment.upstream_identity_id,
+      safe_evidence_snapshot:
+        Map.merge(metadata, %{
+          reason_code: reason_code,
+          pool_id: rule.pool_id,
+          upstream_identity_id: assignment.upstream_identity_id,
+          pool_upstream_assignment_id: assignment.assignment_id,
+          model: rule.model
+        }),
+      targets: [target(rule, rule.pool_id, metadata)]
+    })
+  end
+
+  defp match(rule, dedupe_key, timestamp, attrs) do
+    match_attrs = %{
+      dedupe_key: dedupe_key,
+      scope_type: rule.scope_type,
+      rule_kind: rule.rule_kind,
+      severity: severity_for(rule),
+      pool_id: Map.get(attrs, :pool_id),
+      upstream_identity_id: Map.get(attrs, :upstream_identity_id),
+      safe_evidence_snapshot: Map.fetch!(attrs, :safe_evidence_snapshot),
+      targets: Map.fetch!(attrs, :targets),
+      matched_at: timestamp
+    }
+
+    %{
+      action: :match,
+      dedupe_key: dedupe_key,
+      rule_id: rule.id,
+      rule_kind: rule.rule_kind,
+      match_attrs: match_attrs
+    }
+  end
+
+  defp threshold_match?(%Quota.AccountQuotaWindow{} = window, threshold, timestamp) do
+    Quota.Windows.fresh_window?(window, timestamp) and not is_nil(window.used_percent) and
+      Decimal.compare(window.used_percent, threshold) != :lt
+  end
+
+  defp selected_windows(windows, "any", model, timestamp) do
+    opts = model_opts(model) ++ [at: timestamp]
+
+    windows
+    |> Quota.Windows.quota_window_selection_data_from_windows(opts)
+    |> Map.fetch!(:routing_windows)
+  end
+
+  defp selected_windows(windows, selector, model, timestamp) do
+    windows
+    |> selected_windows("any", model, timestamp)
+    |> Enum.filter(&(window_selector(&1) == selector))
+  end
+
+  defp window_selector(%Quota.AccountQuotaWindow{} = window) do
+    cond do
+      model_window?(window) and window.window_kind == "secondary" -> "model_secondary"
+      model_window?(window) -> "model_primary"
+      window.window_kind == "secondary" -> "account_secondary"
+      true -> "account_primary"
+    end
+  end
+
+  defp model_window?(%Quota.AccountQuotaWindow{} = window) do
+    window.quota_scope in ["model", "upstream_model"] or present_string?(window.model) or
+      present_string?(window.upstream_model)
+  end
+
+  defp target(rule, pool_id, metadata) do
+    %{rule_id: rule.id, pool_id: pool_id, metadata: stringify_metadata(metadata)}
+  end
+
+  defp severity_for(%AlertRule{
+         rule_kind: "upstream_auth_state",
+         target_state: "reauth_required"
+       }),
+       do: "critical"
+
+  defp severity_for(%AlertRule{severity: severity}), do: severity
+
+  defp model_opts(nil), do: []
+  defp model_opts(model), do: [model: model]
+
+  defp stringify_count_keys(counts),
+    do: Map.new(counts, fn {key, value} -> {to_string(key), value} end)
+
+  defp stringify_metadata(metadata),
+    do: Map.new(metadata, fn {key, value} -> {to_string(key), value} end)
+
+  defp decimal_string(%Decimal{} = value), do: Decimal.to_string(value, :normal)
+  defp decimal_string(value) when is_integer(value), do: Integer.to_string(value)
+
+  defp decimal_float(%Decimal{} = value), do: Decimal.to_float(value)
+  defp decimal_float(_value), do: nil
+
+  defp iso8601_or_nil(%DateTime{} = value), do: DateTime.to_iso8601(value)
+  defp iso8601_or_nil(_value), do: nil
+
+  defp present_string?(value) when is_binary(value), do: String.trim(value) != ""
+  defp present_string?(_value), do: false
+end