diff --git a/lib/api_organizations/src/plan.rs b/lib/api_organizations/src/plan.rs
index 591ae48a4..9389efa84 100644
--- a/lib/api_organizations/src/plan.rs
+++ b/lib/api_organizations/src/plan.rs
@@ -5,7 +5,7 @@ use bencher_endpoint::{
     CorsResponse, Delete, Endpoint, Get, Patch, Post, ResponseCreated, ResponseDeleted, ResponseOk,
 };
 use bencher_json::{
-    DateTime, MeteredPlanId, OrganizationResourceId,
+    DateTime, OrganizationResourceId, PlanStatus,
     organization::plan::{JsonNewPlan, JsonPlan, JsonUpdatePlan},
 };
 use bencher_rbac::organization::Permission;
@@ -14,7 +14,7 @@ use bencher_schema::{
     context::ApiContext,
     error::{
         BencherResource, bad_request_error, forbidden_error, issue_error, resource_conflict_err,
-        resource_conflict_error, resource_not_found_err,
+        resource_conflict_error, resource_not_found_err, service_unavailable_error,
     },
     model::{
         organization::{
@@ -235,16 +235,9 @@ async fn post_inner(
         .rbac
         .is_allowed_organization(auth_user, Permission::Manage, &query_organization)
         .map_err(forbidden_error)?;
-    // Check to make sure the organization doesn't already have a plan
-    if let Ok(query_plan) =
-        QueryPlan::belonging_to(&query_organization).first::<QueryPlan>(auth_conn!(context))
-    {
-        return Err(resource_conflict_error(
-            BencherResource::Plan,
-            (query_organization, query_plan),
-            "Organization already has a plan",
-        ));
-    }
+    // Block creating a plan when the organization still has an active plan; prune a
+    // stale lapsed metered plan row first so the organization can subscribe again.
+    prune_or_conflict_existing_plan(context, biller, &query_organization).await?;
 
     let JsonNewPlan {
         checkout,
@@ -311,10 +304,6 @@ async fn post_inner(
             .as_ref()
             .parse()
             .map_err(resource_not_found_err!(Plan, subscription_id))?;
-        // Grant the first period's included usage credit so a new metered org has
-        // its credit immediately (best-effort; the daily sweep handles later
-        // periods). A no-op for plans without a base fee.
-        grant_initial_credit(biller, &metered_plan_id).await;
         InsertPlan::metered_plan(write_conn!(context), metered_plan_id, &query_organization)?;
         QueryPlan::belonging_to(&query_organization)
             .first::<QueryPlan>(auth_conn!(context))
@@ -329,6 +318,87 @@ async fn post_inner(
     }
 }
 
+/// Block plan creation when the organization still has a plan that is active, trialing,
+/// or recoverable (a dunning `past_due`/`unpaid`/`incomplete`/`paused` subscription still
+/// exists in Stripe). Only a *terminal* metered subscription (canceled or
+/// incomplete-expired), or one Stripe reports as gone, leaves a stale local plan row safe
+/// to prune so the organization can subscribe again now that the daily reconciliation
+/// sweep is gone; pruning a still-live subscription would orphan it and let the org
+/// create a duplicate (double billing). Licensed (Self-Hosted) plan rows do not lapse on
+/// their own, so they always block.
+async fn prune_or_conflict_existing_plan(
+    context: &ApiContext,
+    biller: &Biller,
+    query_organization: &QueryOrganization,
+) -> Result<(), HttpError> {
+    let Ok(query_plan) =
+        QueryPlan::belonging_to(query_organization).first::<QueryPlan>(auth_conn!(context))
+    else {
+        return Ok(());
+    };
+
+    // A licensed (Self-Hosted) plan carries no metered subscription and always blocks.
+    let Some(metered_plan_id) = &query_plan.metered_plan else {
+        return Err(plan_conflict(query_organization, query_plan));
+    };
+
+    // Resolve the live metered subscription status (`None` => Stripe reports it gone). A
+    // transient Stripe error surfaces as 503 rather than being mistaken for "gone", so we
+    // never prune a subscription that might still be live.
+    let status = biller
+        .metered_plan_status(metered_plan_id)
+        .await
+        .map_err(service_unavailable_error)?;
+
+    match existing_plan_action(status) {
+        ExistingPlan::Conflict => Err(plan_conflict(query_organization, query_plan)),
+        ExistingPlan::Prune => {
+            diesel::delete(schema::plan::table.filter(schema::plan::id.eq(query_plan.id)))
+                .execute(write_conn!(context))
+                .map_err(resource_conflict_err!(Plan, query_plan))?;
+            Ok(())
+        },
+    }
+}
+
+/// The conflict error returned when an organization already has a plan that blocks
+/// creating a new one.
+fn plan_conflict(query_organization: &QueryOrganization, query_plan: QueryPlan) -> HttpError {
+    resource_conflict_error(
+        BencherResource::Plan,
+        (query_organization.clone(), query_plan),
+        "Organization already has a plan",
+    )
+}
+
+/// Whether an organization's existing metered plan row blocks creating a new plan, or is
+/// a stale row to prune. `status` is `None` when Stripe reports the subscription gone
+/// (404), else its live status.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum ExistingPlan {
+    Conflict,
+    Prune,
+}
+
+/// Prune (allow re-subscribe) only when the subscription is gone or *terminal* (canceled
+/// / incomplete-expired). An active, trialing, or recoverable (dunning) subscription
+/// still exists in Stripe and must block, so we never orphan a live subscription and let
+/// the org create a duplicate. Matched exhaustively so a new `PlanStatus` forces a
+/// decision here.
+fn existing_plan_action(status: Option<PlanStatus>) -> ExistingPlan {
+    match status {
+        None | Some(PlanStatus::Canceled | PlanStatus::IncompleteExpired) => ExistingPlan::Prune,
+        Some(
+            PlanStatus::Active
+            | PlanStatus::Trialing
+            | PlanStatus::PastDue
+            | PlanStatus::Unpaid
+            | PlanStatus::Incomplete
+            | PlanStatus::Paused,
+        ) => ExistingPlan::Conflict,
+    }
+}
+
 #[derive(Deserialize, JsonSchema)]
 pub struct OrgPlanQuery {
     pub remote: Option<bool>,
@@ -400,21 +470,6 @@ async fn delete_inner(
     delete_plan_result
 }
 
-/// Best-effort grant of the first period's included usage credit for a new
-/// metered subscription. Idempotent and self-guarding (a no-op for plans without
-/// a base fee); never blocks plan creation.
-async fn grant_initial_credit(biller: &Biller, metered_plan_id: &MeteredPlanId) {
-    let _credit = biller
-        .ensure_period_credit(metered_plan_id)
-        .await
-        .inspect_err(|e| {
-            #[cfg(feature = "sentry")]
-            sentry::capture_error(e);
-            #[cfg(not(feature = "sentry"))]
-            let _ = e;
-        });
-}
-
 async fn delete_plan(
     context: &ApiContext,
     biller: &Biller,
@@ -463,3 +518,36 @@ async fn delete_plan(
 
     Ok(())
 }
+
+#[cfg(test)]
+mod tests {
+    use bencher_json::PlanStatus;
+
+    use super::{ExistingPlan, existing_plan_action};
+
+    #[test]
+    fn existing_plan_action_decides() {
+        // Gone in Stripe (404) or terminal: prune the stale row.
+        assert_eq!(existing_plan_action(None), ExistingPlan::Prune);
+        assert_eq!(
+            existing_plan_action(Some(PlanStatus::Canceled)),
+            ExistingPlan::Prune,
+        );
+        assert_eq!(
+            existing_plan_action(Some(PlanStatus::IncompleteExpired)),
+            ExistingPlan::Prune,
+        );
+        // Active, trialing, or recoverable (dunning) still exists in Stripe: block, so we
+        // never orphan a live subscription and let the org create a duplicate.
+        for status in [
+            PlanStatus::Active,
+            PlanStatus::Trialing,
+            PlanStatus::PastDue,
+            PlanStatus::Unpaid,
+            PlanStatus::Incomplete,
+            PlanStatus::Paused,
+        ] {
+            assert_eq!(existing_plan_action(Some(status)), ExistingPlan::Conflict);
+        }
+    }
+}
diff --git a/lib/api_organizations/src/usage.rs b/lib/api_organizations/src/usage.rs
index e61bfd536..25627f3d7 100644
--- a/lib/api_organizations/src/usage.rs
+++ b/lib/api_organizations/src/usage.rs
@@ -219,10 +219,11 @@ fn free_plan_usage(
 
 /// Estimate billable usage for a metered (Bencher Cloud) plan.
 ///
-/// On Pro, only Private Project metrics are metered; Public Project metrics are free
-/// and unlimited. On legacy Team (and metered Enterprise) plans, both Public and
-/// Private Project metrics are metered. This mirrors `PlanKind::check_usage` so the
-/// estimate matches what is actually billed. Bare metal runner minutes are metered
+/// Legacy Team (and metered Enterprise) plans are metered on both Public and Private
+/// Project metrics, so the returned metrics figure matches their bill. Pro now bills on
+/// active series rather than metrics, so for Pro this returns only its Private Project
+/// metrics as an informational figure that no longer matches the bill; surfacing the
+/// active-series count here is a follow-up. Bare metal runner minutes are metered
 /// regardless of visibility and plan level.
 fn metered_plan_usage(
     conn: &mut DbConnection,
diff --git a/lib/bencher_config/src/config_tx.rs b/lib/bencher_config/src/config_tx.rs
index 73ec6fe82..0b301e221 100644
--- a/lib/bencher_config/src/config_tx.rs
+++ b/lib/bencher_config/src/config_tx.rs
@@ -24,7 +24,7 @@ use bencher_schema::{
     context::RateLimiting,
     model::{
         runner::job::{reprocess_completed_jobs, spawn_heartbeat_timeout},
-        server::{QueryServer, spawn_credit_grants},
+        server::QueryServer,
     },
     write_conn,
 };
@@ -173,22 +173,6 @@ impl ConfigTx {
         #[cfg(feature = "plus")]
         spawn_stats(log, server.app_private()).await?;
 
-        // Bencher Cloud only (a Biller is configured): keep each metered
-        // subscription's monthly usage credit granted and reconcile lapses on a
-        // daily sweep, off the report-ingestion path.
-        #[cfg(feature = "plus")]
-        {
-            let context = server.app_private();
-            if let Some(biller) = context.biller.clone() {
-                spawn_credit_grants(
-                    log.clone(),
-                    context.database.path.clone(),
-                    context.database.busy_timeout,
-                    biller,
-                );
-            }
-        }
-
         Ok(server)
     }
 }
diff --git a/lib/bencher_json/src/organization/plan.rs b/lib/bencher_json/src/organization/plan.rs
index 2e274ada8..981f194b0 100644
--- a/lib/bencher_json/src/organization/plan.rs
+++ b/lib/bencher_json/src/organization/plan.rs
@@ -12,6 +12,12 @@ use crate::{BigInt, OrganizationUuid, system::payment::JsonCustomer};
 
 pub const DEFAULT_PRICE_NAME: &str = "default";
 pub const METRICS_METER_NAME: &str = "metrics";
+/// Stripe meter for monthly-active series (distinct testbed x benchmark x measure).
+/// Backs the Pro tiered price (tier 1 flat fee plus per-series step-ups) and is billed
+/// with `last` aggregation: after each report we post the org's cumulative
+/// period-to-date series count, so the final post of the period is the period total and
+/// a missed post self-heals on the next report.
+pub const ACTIVE_SERIES_METER_NAME: &str = "active_series";
 pub const RUNNER_MINUTES_METER_NAME: &str = "runner_minutes";
 
 #[typeshare::typeshare]
diff --git a/lib/bencher_json/src/system/config/plus/cloud/billing.rs b/lib/bencher_json/src/system/config/plus/cloud/billing.rs
index ed319fc12..7fbef9a0c 100644
--- a/lib/bencher_json/src/system/config/plus/cloud/billing.rs
+++ b/lib/bencher_json/src/system/config/plus/cloud/billing.rs
@@ -25,6 +25,8 @@ pub struct JsonProducts {
     // Legacy self-serve paid tier, retained for grandfathered customers.
     pub team: JsonProduct,
     pub enterprise: JsonProduct,
+    // Shared metered metrics product for legacy Team/Enterprise plans. Pro bills on
+    // its own tiered active-series price (its `metered` map), not on this meter.
     pub metrics: JsonProduct,
     pub bare_metal: JsonProduct,
 }
@@ -37,37 +39,28 @@ pub struct JsonProduct {
     pub metered: HashMap<String, String>,
     #[serde(default)]
     pub licensed: HashMap<String, String>,
-    // Flat recurring base prices.
-    // Empty for products with no flat base fee.
-    #[serde(default)]
-    pub base: HashMap<String, String>,
-    #[serde(default)]
-    pub trial_coupon: Option<String>,
 }
 
 #[cfg(test)]
 mod tests {
     use super::{JsonProduct, JsonProducts};
 
-    // Existing team/enterprise/bare_metal config has no `base` key; it must still
-    // deserialize, defaulting `base` to an empty map (backward compatibility).
+    // A product given only an `id` still deserializes, defaulting `metered` and
+    // `licensed` to empty maps (backward compatibility).
     #[test]
-    fn product_without_base_defaults_empty() {
-        let json =
-            r#"{"id":"prod_x","metered":{"default":"price_m"},"licensed":{"default":"price_l"}}"#;
+    fn product_without_pricing_defaults_empty() {
+        let json = r#"{"id":"prod_x"}"#;
         let product: JsonProduct = serde_json::from_str(json).unwrap();
-        assert!(product.base.is_empty());
-        assert!(product.trial_coupon.is_none());
-        assert_eq!(
-            product.metered.get("default").map(String::as_str),
-            Some("price_m"),
-        );
+        assert!(product.metered.is_empty());
+        assert!(product.licensed.is_empty());
     }
 
+    // Pro carries a single tiered metered price (base fee plus per-series step-ups);
+    // the shared metrics product carries the legacy Team/Enterprise metered price.
     #[test]
-    fn products_with_pro_base_and_trial_coupon() {
+    fn products_with_pro_tiered_price() {
         let json = r#"{
-            "pro": {"id":"p","base":{"default":"price_b"},"trial_coupon":"coupon_x"},
+            "pro": {"id":"p","metered":{"default":"price_pro_tiered"}},
             "team": {"id":"t","metered":{},"licensed":{}},
             "enterprise": {"id":"e","metered":{},"licensed":{}},
             "metrics": {"id":"m","metered":{"default":"price_mm"},"licensed":{"default":"price_ml"}},
@@ -75,16 +68,14 @@ mod tests {
         }"#;
         let products: JsonProducts = serde_json::from_str(json).unwrap();
         assert_eq!(
-            products.pro.base.get("default").map(String::as_str),
-            Some("price_b"),
+            products.pro.metered.get("default").map(String::as_str),
+            Some("price_pro_tiered"),
         );
-        assert_eq!(products.pro.trial_coupon.as_deref(), Some("coupon_x"));
-        assert!(products.pro.metered.is_empty());
+        assert!(products.pro.licensed.is_empty());
         assert_eq!(
             products.metrics.metered.get("default").map(String::as_str),
             Some("price_mm"),
         );
-        assert!(products.team.base.is_empty());
-        assert!(products.team.trial_coupon.is_none());
+        assert!(products.team.metered.is_empty());
     }
 }
diff --git a/lib/bencher_schema/migrations/2026-06-24-120000_series_last_seen/down.sql b/lib/bencher_schema/migrations/2026-06-24-120000_series_last_seen/down.sql
new file mode 100644
index 000000000..f08f88c0b
--- /dev/null
+++ b/lib/bencher_schema/migrations/2026-06-24-120000_series_last_seen/down.sql
@@ -0,0 +1 @@
+DROP TABLE IF EXISTS series_last_seen;
diff --git a/lib/bencher_schema/migrations/2026-06-24-120000_series_last_seen/up.sql b/lib/bencher_schema/migrations/2026-06-24-120000_series_last_seen/up.sql
new file mode 100644
index 000000000..92e6b03e5
--- /dev/null
+++ b/lib/bencher_schema/migrations/2026-06-24-120000_series_last_seen/up.sql
@@ -0,0 +1,53 @@
+-- Cache of each monitored series' most recent activity, for value-based billing on
+-- monthly-active series (a series is a distinct testbed x benchmark x measure) and
+-- for telemetry. `last_seen` is the greatest `report.created` (the server-side
+-- ingestion time, not the user-supplied `end_time`) ever recorded for the series, kept
+-- monotonic on ingest so reprocessing an older report cannot lower it.
+--
+-- `organization_id` and `project_id` are denormalized so a billing read is a single
+-- index range scan over (organization_id, last_seen), without joining through the
+-- entity tables. The series key (testbed_id, benchmark_id, measure_id) is globally
+-- unique on its own because each entity is per-project and id-unique, so it is the
+-- primary key.
+--
+-- Hard-deleting a testbed, benchmark, or measure cascades its series rows away and can
+-- lower the current period's active-series count; an accepted exception to monotonic
+-- billing, since entity deletion requires first deleting all of its reports.
+CREATE TABLE series_last_seen (
+    organization_id INTEGER NOT NULL,
+    project_id INTEGER NOT NULL,
+    testbed_id INTEGER NOT NULL,
+    benchmark_id INTEGER NOT NULL,
+    measure_id INTEGER NOT NULL,
+    last_seen BIGINT NOT NULL,
+    PRIMARY KEY (testbed_id, benchmark_id, measure_id),
+    FOREIGN KEY (organization_id) REFERENCES organization (id) ON DELETE CASCADE,
+    FOREIGN KEY (project_id) REFERENCES project (id) ON DELETE CASCADE,
+    FOREIGN KEY (testbed_id) REFERENCES testbed (id) ON DELETE CASCADE,
+    FOREIGN KEY (benchmark_id) REFERENCES benchmark (id) ON DELETE CASCADE,
+    FOREIGN KEY (measure_id) REFERENCES measure (id) ON DELETE CASCADE
+);
+
+CREATE INDEX index_series_last_seen_org_last_seen
+    ON series_last_seen (organization_id, last_seen);
+
+-- Index the remaining cascaded foreign keys so parent deletes do not full-scan this
+-- table. `testbed_id` is already the primary key's prefix and `organization_id` is
+-- covered by the billing index above.
+CREATE INDEX index_series_last_seen_project ON series_last_seen (project_id);
+CREATE INDEX index_series_last_seen_benchmark ON series_last_seen (benchmark_id);
+CREATE INDEX index_series_last_seen_measure ON series_last_seen (measure_id);
+
+-- Backfill from existing metrics: one row per distinct series with the latest
+-- `report.created` it has produced. Runs once, against the table just created empty
+-- above and before the server serves any request, so it needs no conflict handling
+-- (like `metric_count_by_report`). Correct on day one, so the cache equals a fresh
+-- COUNT(DISTINCT testbed, benchmark, measure) over the same metrics.
+INSERT INTO series_last_seen (organization_id, project_id, testbed_id, benchmark_id, measure_id, last_seen)
+SELECT p.organization_id, p.id, r.testbed_id, rb.benchmark_id, m.measure_id, MAX(r.created)
+FROM metric m
+INNER JOIN report_benchmark rb ON m.report_benchmark_id = rb.id
+INNER JOIN report r ON rb.report_id = r.id
+INNER JOIN benchmark b ON rb.benchmark_id = b.id
+INNER JOIN project p ON b.project_id = p.id
+GROUP BY r.testbed_id, rb.benchmark_id, m.measure_id;
diff --git a/lib/bencher_schema/src/macros/sql.rs b/lib/bencher_schema/src/macros/sql.rs
index ea5b257da..cc70ff614 100644
--- a/lib/bencher_schema/src/macros/sql.rs
+++ b/lib/bencher_schema/src/macros/sql.rs
@@ -25,3 +25,13 @@ diesel::define_sql_function! {
     /// (silent truncation).
     fn min(a: diesel::sql_types::Integer, b: diesel::sql_types::Integer) -> diesel::sql_types::Integer;
 }
+
+diesel::define_sql_function! {
+    /// `SQLite` scalar `MAX(a, b)`: returns the greater of two `BigInt` values.
+    ///
+    /// Used by the `series_last_seen` upsert to keep `last_seen` monotonic: reprocessing
+    /// an older report (an older `report.created`) must not lower a series' recorded
+    /// activity. `BigInt` matches the `DateTime` SQL representation (whole-second Unix
+    /// timestamps; see `DateTime`'s `ToSql`).
+    fn max(a: diesel::sql_types::BigInt, b: diesel::sql_types::BigInt) -> diesel::sql_types::BigInt;
+}
diff --git a/lib/bencher_schema/src/model/organization/plan.rs b/lib/bencher_schema/src/model/organization/plan.rs
index 78c16574b..732241c7f 100644
--- a/lib/bencher_schema/src/model/organization/plan.rs
+++ b/lib/bencher_schema/src/model/organization/plan.rs
@@ -91,7 +91,7 @@ impl QueryPlan {
         biller: Option<&Biller>,
         api_actor: &ApiActor,
         query_organization: &QueryOrganization,
-    ) -> Result<Option<(CustomerId, PlanLevel)>, HttpError> {
+    ) -> Result<Option<MeteredPlan>, HttpError> {
         let Some(biller) = biller else {
             return Ok(None);
         };
@@ -106,35 +106,25 @@ impl QueryPlan {
             return Ok(None);
         };
 
-        let (plan_status, customer_id, level) = biller
-            .get_metered_plan_status(&metered_plan_id)
+        let billing = biller
+            .get_metered_plan_billing(&metered_plan_id)
             .await
             .map_err(not_found_error)?;
 
         // A canceled/lapsed (inactive) subscription gracefully downgrades to
         // Free: return `None` so the caller falls through to the public/no-plan
         // logic instead of hard-erroring.
-        if plan_status.is_active() {
-            Ok(Some((customer_id, level)))
+        if billing.status.is_active() {
+            Ok(Some(MeteredPlan {
+                customer_id: billing.customer_id,
+                level: billing.level,
+                current_period_start: billing.current_period_start,
+                current_period_end: billing.current_period_end,
+            }))
         } else {
             Ok(None)
         }
     }
-
-    /// All organization plans with a metered (Stripe) subscription. Used by the
-    /// daily billing sweep to ensure each period's credit and reconcile canceled
-    /// subscriptions.
-    pub fn all_metered(conn: &mut DbConnection) -> diesel::QueryResult<Vec<Self>> {
-        schema::plan::table
-            .filter(schema::plan::metered_plan.is_not_null())
-            .load::<Self>(conn)
-    }
-
-    /// Delete a plan row by id. Used by the daily billing sweep to prune a plan
-    /// whose subscription has fully lapsed.
-    pub fn delete(conn: &mut DbConnection, plan_id: PlanId) -> diesel::QueryResult<usize> {
-        diesel::delete(schema::plan::table.filter(schema::plan::id.eq(plan_id))).execute(conn)
-    }
 }
 
 #[derive(Debug, diesel::Insertable)]
@@ -226,8 +216,19 @@ impl InsertPlan {
     }
 }
 
+/// An active metered (Stripe) subscription's billing context, carried by
+/// [`PlanKind::Metered`]: who to bill (`customer_id`), the tier (`level`), and the
+/// current billing period (the active-series count window for the post-report Pro
+/// series push).
+pub struct MeteredPlan {
+    pub customer_id: CustomerId,
+    pub level: PlanLevel,
+    pub current_period_start: DateTime,
+    pub current_period_end: DateTime,
+}
+
 pub enum PlanKind {
-    Metered(CustomerId, PlanLevel),
+    Metered(MeteredPlan),
     Licensed(LicenseUsage),
     None,
 }
@@ -260,7 +261,7 @@ impl PlanKind {
         }
         match self {
             Self::None => Priority::Free,
-            Self::Metered(_, _) => Priority::Plus,
+            Self::Metered(_) => Priority::Plus,
             Self::Licensed(license_usage) => match license_usage.level {
                 PlanLevel::Free => Priority::Free,
                 PlanLevel::Pro | PlanLevel::Team | PlanLevel::Enterprise => Priority::Plus,
@@ -268,6 +269,25 @@ impl PlanKind {
         }
     }
 
+    /// For a Pro metered plan, the Stripe customer and current billing period needed to
+    /// post the post-report active-series usage. `None` for any non-Pro plan, so only
+    /// Pro triggers a series post. Pairs with [`metered_bills_active_series`].
+    pub fn metered_series_billing(&self) -> Option<(CustomerId, DateTime, DateTime)> {
+        match self {
+            Self::Metered(MeteredPlan {
+                customer_id,
+                level,
+                current_period_start,
+                current_period_end,
+            }) if metered_bills_active_series(*level) => Some((
+                customer_id.clone(),
+                *current_period_start,
+                *current_period_end,
+            )),
+            Self::Metered(_) | Self::Licensed(_) | Self::None => None,
+        }
+    }
+
     async fn new(
         context: &ApiContext,
         biller: Option<&Biller>,
@@ -276,11 +296,11 @@ impl PlanKind {
         query_organization: &QueryOrganization,
         visibility: Visibility,
     ) -> Result<Self, HttpError> {
-        if let Some((customer_id, level)) =
+        if let Some(metered_plan) =
             QueryPlan::get_active_metered_plan(context, biller, api_actor, query_organization)
                 .await?
         {
-            Ok(Self::Metered(customer_id, level))
+            Ok(Self::Metered(metered_plan))
         } else if let Some(license_usage) = LicenseUsage::get(
             actor_conn!(context, api_actor),
             licensor,
@@ -396,7 +416,17 @@ impl PlanKind {
         usage: u32,
     ) -> Result<(), HttpError> {
         match self {
-            Self::Metered(customer_id, level) => {
+            Self::Metered(MeteredPlan {
+                customer_id, level, ..
+            }) => {
+                // Pro bills on active series via its tiered price (posted after each
+                // report), not per-metric, so it records no metrics usage. Legacy Team
+                // (and metered Enterprise) plans bill all metrics on the `metrics`
+                // meter. Bare metal runner time is metered separately via the runner
+                // channel.
+                if metered_bills_active_series(level) {
+                    return Ok(());
+                }
                 let Some(biller) = biller else {
                     return Err(issue_error(
                         "No Biller when checking usage",
@@ -404,14 +434,6 @@ impl PlanKind {
                         PlanKindError::NoBiller,
                     ));
                 };
-                // Private Project Metrics are always metered. Public Project Metrics
-                // are free only on Pro; legacy Team (and metered Enterprise) plans
-                // are billed for public metrics too. Bare metal runner time is
-                // metered separately (regardless of visibility) via the runner
-                // channel.
-                if project.visibility.is_public() && !metered_bills_public_metrics(level) {
-                    return Ok(());
-                }
                 if let Err(e) = biller.record_metrics_usage(&customer_id, usage).await {
                     #[cfg(feature = "otel")]
                     bencher_otel::ApiMeter::increment(
@@ -447,17 +469,15 @@ impl PlanKind {
     }
 }
 
-/// Whether a metered (Stripe) subscription is billed for *public* project metrics.
+/// Whether a metered (Stripe) subscription counts *public* project metrics in the usage
+/// estimate shown by the usage endpoint.
 ///
-/// Public Project Metrics are free only on Pro (the new self-serve tier, the only
-/// paid tier with a flat base fee). Legacy Team (and metered Enterprise, which the
-/// base-fee heuristic resolves to `Team`) plans are billed for public metrics too.
-/// Private Project Metrics are always billed regardless of level, so this only
-/// governs the public case.
+/// Only legacy Team (and metered Enterprise, which the price heuristic resolves to
+/// `Team`) plans bill public metrics, so their estimate counts all metrics; Pro's
+/// estimate counts only private metrics. This governs only the metrics figure shown for
+/// metered plans. Pro itself now bills on active series, not metrics (see
+/// [`metered_bills_active_series`]).
 pub fn metered_bills_public_metrics(level: PlanLevel) -> bool {
-    // Public metrics are free on Free and Pro (Pro is the self-serve tier that
-    // includes them); only the legacy Team tier (and metered Enterprise, which the
-    // base-fee heuristic resolves to `Team`) is billed for public metrics.
     // Matched exhaustively so a new `PlanLevel` variant forces a decision here.
     match level {
         PlanLevel::Free | PlanLevel::Pro => false,
@@ -465,6 +485,25 @@ pub fn metered_bills_public_metrics(level: PlanLevel) -> bool {
     }
 }
 
+/// Whether a metered (Stripe) subscription is billed on the active-series meter.
+///
+/// Active-series billing is the Pro plan only: Pro's tiered price bills the base fee and
+/// the per-series step-ups on the `active_series` meter, and Pro records no per-metric
+/// `metrics` usage (see [`PlanKind::check_usage`]). Legacy Team and metered Enterprise
+/// plans stay on the `metrics` meter and have no series usage recorded.
+///
+/// Also gates the post-report series push: after a Pro report we count the
+/// organization's active series and post the period-to-date total to the
+/// `active_series` meter.
+///
+/// Matched exhaustively so a new `PlanLevel` variant forces a decision here.
+pub fn metered_bills_active_series(level: PlanLevel) -> bool {
+    match level {
+        PlanLevel::Pro => true,
+        PlanLevel::Free | PlanLevel::Team | PlanLevel::Enterprise => false,
+    }
+}
+
 pub struct LicenseUsage {
     pub entitlements: Entitlements,
     pub usage: u32,
@@ -530,9 +569,9 @@ impl LicenseUsage {
 
 #[cfg(test)]
 mod tests {
-    use bencher_json::PlanLevel;
+    use bencher_json::{DateTime, PlanLevel};
 
-    use super::metered_bills_public_metrics;
+    use super::{MeteredPlan, PlanKind, metered_bills_active_series, metered_bills_public_metrics};
 
     #[test]
     fn does_not_bill_public_metrics() {
@@ -545,8 +584,39 @@ mod tests {
     fn bills_public_metrics() {
         // Legacy Team and metered Enterprise are billed for public metrics. Free is
         // included for completeness even though metered plans only resolve to
-        // Pro/Team via the base-fee heuristic.
+        // Pro/Team via the Pro-price heuristic.
         assert!(metered_bills_public_metrics(PlanLevel::Team));
         assert!(metered_bills_public_metrics(PlanLevel::Enterprise));
     }
+
+    #[test]
+    fn bills_active_series_pro_only() {
+        // Active-series billing is the Pro plan only.
+        assert!(metered_bills_active_series(PlanLevel::Pro));
+        assert!(!metered_bills_active_series(PlanLevel::Free));
+        assert!(!metered_bills_active_series(PlanLevel::Team));
+        assert!(!metered_bills_active_series(PlanLevel::Enterprise));
+    }
+
+    #[test]
+    fn metered_series_billing_pro_only() {
+        let metered = |level| {
+            PlanKind::Metered(MeteredPlan {
+                customer_id: "cus_test".into(),
+                level,
+                current_period_start: DateTime::TEST,
+                current_period_end: DateTime::TEST,
+            })
+        };
+        // Only Pro resolves a post-report series-billing context; other metered tiers
+        // and the non-metered kinds do not.
+        assert!(metered(PlanLevel::Pro).metered_series_billing().is_some());
+        assert!(metered(PlanLevel::Team).metered_series_billing().is_none());
+        assert!(
+            metered(PlanLevel::Enterprise)
+                .metered_series_billing()
+                .is_none()
+        );
+        assert!(PlanKind::None.metered_series_billing().is_none());
+    }
 }
diff --git a/lib/bencher_schema/src/model/project/mod.rs b/lib/bencher_schema/src/model/project/mod.rs
index 20f10acb4..0170e0df2 100644
--- a/lib/bencher_schema/src/model/project/mod.rs
+++ b/lib/bencher_schema/src/model/project/mod.rs
@@ -47,6 +47,7 @@ pub mod metric_boundary;
 pub mod plot;
 pub mod project_role;
 pub mod report;
+pub mod series;
 pub mod testbed;
 pub mod threshold;
 
diff --git a/lib/bencher_schema/src/model/project/report/mod.rs b/lib/bencher_schema/src/model/project/report/mod.rs
index 80885999b..a4dbd1682 100644
--- a/lib/bencher_schema/src/model/project/report/mod.rs
+++ b/lib/bencher_schema/src/model/project/report/mod.rs
@@ -23,7 +23,10 @@ use crate::model::spec::SpecId;
 #[cfg(feature = "plus")]
 use crate::model::{
     organization::plan::PlanKind,
-    project::testbed::{RunJob, RunTestbed},
+    project::{
+        series::count_active,
+        testbed::{RunJob, RunTestbed},
+    },
     runner::{PendingInsertJob, SourceIp},
 };
 use crate::{
@@ -472,6 +475,10 @@ impl QueryReport {
     ) -> Result<(), HttpError> {
         #[cfg(feature = "plus")]
         let mut usage = 0;
+        // Capture the Pro active-series billing context (customer + period) before
+        // `check_usage` consumes the plan kind; `None` for any non-Pro plan.
+        #[cfg(feature = "plus")]
+        let series_billing = plan_kind.metered_series_billing();
 
         let mut report_results = ReportResults::new(
             self.project_id,
@@ -480,6 +487,11 @@ impl QueryReport {
             self.testbed_id,
             self.spec_id,
             self.id,
+            #[cfg(feature = "plus")]
+            results::SeriesCacheContext {
+                organization_id: query_project.organization_id,
+                report_created: self.created,
+            },
         );
         let processed = report_results
             .process(
@@ -506,10 +518,100 @@ impl QueryReport {
             .check_usage(context.biller.as_ref(), query_project, usage)
             .await?;
 
+        // Pro active-series billing: once the report's metrics and series cache are
+        // committed, count the organization's period-to-date active series on the request
+        // connection and post it to Stripe in a detached task (only the Stripe post is
+        // off the hot path). Skipped if processing failed (the cache rolled back).
+        #[cfg(feature = "plus")]
+        if processed.is_ok()
+            && let Some((customer_id, period_start, period_end)) = series_billing
+        {
+            post_series_usage(
+                log,
+                context,
+                query_project,
+                customer_id,
+                period_start,
+                period_end,
+            )
+            .await;
+        }
+
         processed
     }
 }
 
+/// Post a Pro organization's period-to-date active-series count to the `active_series`
+/// meter after a report. Counts on the request connection (a single indexed range scan);
+/// only the Stripe meter post is detached, so it never blocks returning the report.
+/// Best-effort: a count or post failure is logged (and reported via
+/// `ActiveSeriesBilledFailed`) but never surfaced. The next report re-posts the
+/// cumulative count, so a dropped post self-heals within the period. The one gap is a
+/// failed post on a period's final report with no later report before the period closes,
+/// which under-bills by the delta: an accepted availability-over-accuracy tradeoff now
+/// that the reconciliation sweep is gone.
+#[cfg(feature = "plus")]
+async fn post_series_usage(
+    log: &Logger,
+    context: &ApiContext,
+    query_project: &QueryProject,
+    customer_id: bencher_billing::CustomerId,
+    period_start: DateTime,
+    period_end: DateTime,
+) {
+    let Some(biller) = context.biller.clone() else {
+        return;
+    };
+    let organization_id = query_project.organization_id;
+    // Acquire a read connection and count on the request path (a single indexed range
+    // scan). Best-effort: a connection or count failure logs and returns without
+    // surfacing, since the report itself is already committed.
+    let mut conn = match context.database.get_public_conn().await {
+        Ok(conn) => conn,
+        Err(e) => {
+            slog::warn!(
+                log,
+                "Failed to acquire a connection to count active series for organization ({organization_id}): {e}"
+            );
+            return;
+        },
+    };
+    let count = match count_active(&mut conn, organization_id, period_start, period_end) {
+        Ok(count) => count,
+        Err(e) => {
+            slog::warn!(
+                log,
+                "Failed to count active series for organization ({organization_id}): {e}"
+            );
+            return;
+        },
+    };
+    drop(conn);
+
+    // Stamp the meter post with the count time so the `active_series` meter's `last`
+    // aggregation orders deterministically across concurrent reports.
+    let counted_at = context.clock.now();
+    let log = log.clone();
+    tokio::spawn(async move {
+        if let Err(e) = biller
+            .record_series_usage(&customer_id, count, counted_at)
+            .await
+        {
+            #[cfg(feature = "otel")]
+            bencher_otel::ApiMeter::increment(bencher_otel::ApiCounter::ActiveSeriesBilledFailed);
+            slog::warn!(
+                log,
+                "Failed to record active-series usage for organization ({organization_id}): {e}"
+            );
+            #[cfg(feature = "sentry")]
+            sentry::capture_error(&e);
+        } else {
+            #[cfg(feature = "otel")]
+            bencher_otel::ApiMeter::increment(bencher_otel::ApiCounter::ActiveSeriesBilled);
+        }
+    });
+}
+
 type ResultsQuery = (
     Iteration,
     QueryBenchmark,
diff --git a/lib/bencher_schema/src/model/project/report/results/mod.rs b/lib/bencher_schema/src/model/project/report/results/mod.rs
index 2a17c10fb..cc5bac470 100644
--- a/lib/bencher_schema/src/model/project/report/results/mod.rs
+++ b/lib/bencher_schema/src/model/project/report/results/mod.rs
@@ -12,8 +12,13 @@ use diesel::RunQueryDsl as _;
 use dropshot::HttpError;
 use slog::Logger;
 
+#[cfg(feature = "plus")]
+use bencher_json::DateTime;
+
 use crate::macros::sql::last_insert_rowid;
 use crate::model::spec::SpecId;
+#[cfg(feature = "plus")]
+use crate::model::{organization::OrganizationId, project::series::upsert_series_last_seen};
 use crate::{
     auth_conn,
     context::ApiContext,
@@ -44,11 +49,27 @@ pub struct ReportResults {
     pub testbed_id: TestbedId,
     pub spec_id: Option<SpecId>,
     pub report_id: ReportId,
+    // The owning organization and report end time written into the active-series cache
+    // on ingest. Bundled so they travel together and the constructor stays within its
+    // argument count.
+    #[cfg(feature = "plus")]
+    pub series_cache: SeriesCacheContext,
     pub benchmark_cache: HashMap<BenchmarkNameId, BenchmarkId>,
     pub measure_cache: HashMap<MeasureNameId, MeasureId>,
     pub detector_cache: HashMap<MeasureId, Option<Detector>>,
 }
 
+/// The report context the active-series cache write needs: the owning organization
+/// (denormalized into each `series_last_seen` row so a billing read is a single index
+/// scan) and the report's server-side creation time (written as each ingested series'
+/// `last_seen`). Creation time, not the user-supplied `end_time`, is used so a report
+/// cannot dodge active-series billing by claiming a far-future `end_time`.
+#[cfg(feature = "plus")]
+pub struct SeriesCacheContext {
+    pub organization_id: OrganizationId,
+    pub report_created: DateTime,
+}
+
 impl ReportResults {
     pub fn new(
         project_id: ProjectId,
@@ -57,6 +78,7 @@ impl ReportResults {
         testbed_id: TestbedId,
         spec_id: Option<SpecId>,
         report_id: ReportId,
+        #[cfg(feature = "plus")] series_cache: SeriesCacheContext,
     ) -> Self {
         Self {
             project_id,
@@ -65,6 +87,8 @@ impl ReportResults {
             testbed_id,
             spec_id,
             report_id,
+            #[cfg(feature = "plus")]
+            series_cache,
             benchmark_cache: HashMap::new(),
             measure_cache: HashMap::new(),
             detector_cache: HashMap::new(),
@@ -165,6 +189,11 @@ impl ReportResults {
         let write_start = context.clock.now();
 
         write_transaction!(context, |conn| {
+            // Series (testbed x benchmark x measure) seen in this iteration, upserted
+            // into the active-series cache in this same transaction so the cache cannot
+            // drift from the metrics it bills.
+            #[cfg(feature = "plus")]
+            let mut series_keys: Vec<(BenchmarkId, MeasureId)> = Vec::new();
             for prepared in prepared_benchmarks {
                 // Insert report_benchmark
                 diesel::insert_into(schema::report_benchmark::table)
@@ -172,9 +201,13 @@ impl ReportResults {
                     .execute(conn)?;
                 let report_benchmark_id: ReportBenchmarkId =
                     diesel::select(last_insert_rowid()).get_result(conn)?;
+                #[cfg(feature = "plus")]
+                let benchmark_id = prepared.insert_report_benchmark.benchmark_id;
 
                 // Insert all metrics for this benchmark
                 for prepared_metric in prepared.metrics {
+                    #[cfg(feature = "plus")]
+                    series_keys.push((benchmark_id, prepared_metric.measure_id));
                     let insert_metric = InsertMetric::from_json(
                         report_benchmark_id,
                         prepared_metric.measure_id,
@@ -195,6 +228,22 @@ impl ReportResults {
             // Upsert metric count summary (count computed before acquiring write lock)
             super::upsert_metric_count(conn, self.report_id, iteration_metric_count)?;
 
+            // Refresh each seen series' last_seen to this report's creation time, in the
+            // same transaction as the metric inserts above. The (benchmark, measure) pairs
+            // are distinct within an iteration; repeats across iterations are idempotent.
+            #[cfg(feature = "plus")]
+            for (benchmark_id, measure_id) in series_keys {
+                upsert_series_last_seen(
+                    conn,
+                    self.series_cache.organization_id,
+                    self.project_id,
+                    self.testbed_id,
+                    benchmark_id,
+                    measure_id,
+                    self.series_cache.report_created,
+                )?;
+            }
+
             diesel::QueryResult::Ok(())
         })
         .map_err(|e| {
diff --git a/lib/bencher_schema/src/model/project/series.rs b/lib/bencher_schema/src/model/project/series.rs
new file mode 100644
index 000000000..9c10fcdd9
--- /dev/null
+++ b/lib/bencher_schema/src/model/project/series.rs
@@ -0,0 +1,1145 @@
+#![cfg(feature = "plus")]
+
+//! Cache of each monitored series' most recent activity.
+//!
+//! A *series* is a distinct `(testbed, benchmark, measure)` an organization reports
+//! to. `series_last_seen` stores, per series, the greatest `report.created` (the
+//! server-side ingestion time, not the user-supplied `end_time`) ever recorded for it
+//! (`last_seen`). The cache is written on ingest in the same transaction as the metric
+//! inserts and backfilled from existing metrics by the migration. `last_seen` only ever
+//! rises (`MAX`): reprocessing an older report cannot lower it, and deleting a report
+//! does NOT lower it either; a series that reported during a period was active that
+//! period and stays billed for it. The one way a series leaves the count early is
+//! hard-deleting its testbed, benchmark, or measure, which cascades its rows away and
+//! can lower the current period's count. Accepted: entity deletion requires first
+//! deleting all of that entity's reports (destroying the org's own history), and an
+//! inactive series stops billing next period anyway.
+//!
+//! The cache exists so that billing on monthly-active series (and the matching telemetry
+//! figure) is a single index range scan over `(organization_id, last_seen)` instead of a
+//! `COUNT(DISTINCT ...)` over every metric. Absent report deletions it equals that
+//! `COUNT(DISTINCT ...)` (the test oracle, [`oracle_count`]).
+
+use bencher_json::{DateTime, project::Visibility};
+use diesel::{ExpressionMethods as _, QueryDsl as _, RunQueryDsl as _};
+use dropshot::HttpError;
+
+use crate::{
+    context::DbConnection,
+    error::issue_error,
+    model::{
+        organization::OrganizationId,
+        project::{ProjectId, benchmark::BenchmarkId, measure::MeasureId, testbed::TestbedId},
+    },
+    schema::{self, series_last_seen as series_table},
+};
+
+/// Record that a series produced a metric at `last_seen`, keeping the stored value
+/// the greater of its current and new value.
+///
+/// Called once per distinct `(testbed, benchmark, measure)` in an ingested report,
+/// inside the same write transaction as that report's metric inserts. `last_seen` is the
+/// report's server-side creation time. `MAX` keeps it monotonic: reprocessing an older
+/// report cannot lower a series' recorded activity, and a repeat within the same report
+/// (across iterations) is an idempotent no-op.
+///
+/// Does not open its own transaction; callers run it inside the ingest write
+/// transaction so the cache cannot drift from the metrics, and to avoid a nested
+/// `SQLite` savepoint per series.
+pub fn upsert_series_last_seen(
+    conn: &mut DbConnection,
+    organization_id: OrganizationId,
+    project_id: ProjectId,
+    testbed_id: TestbedId,
+    benchmark_id: BenchmarkId,
+    measure_id: MeasureId,
+    last_seen: DateTime,
+) -> diesel::QueryResult<()> {
+    use crate::macros::sql::max;
+
+    diesel::insert_into(series_table::table)
+        .values((
+            series_table::organization_id.eq(organization_id),
+            series_table::project_id.eq(project_id),
+            series_table::testbed_id.eq(testbed_id),
+            series_table::benchmark_id.eq(benchmark_id),
+            series_table::measure_id.eq(measure_id),
+            series_table::last_seen.eq(last_seen),
+        ))
+        .on_conflict((
+            series_table::testbed_id,
+            series_table::benchmark_id,
+            series_table::measure_id,
+        ))
+        .do_update()
+        // Only `last_seen` is refreshed; `organization_id`/`project_id` are written once
+        // at insert. Safe because a project cannot move organizations (no such
+        // operation); if that ever changes, refresh them here too. Visibility is not
+        // stored, so visibility changes are handled at read time by the project join.
+        .set(series_table::last_seen.eq(max(series_table::last_seen, last_seen)))
+        .execute(conn)?;
+    Ok(())
+}
+
+/// Count an organization's monthly-active series in `[start_time, end_time]` (all
+/// project visibilities).
+///
+/// This is the Pro billable figure: Pro is billed for all of its active series
+/// regardless of project visibility. Mirrors `QueryMetric::usage`.
+pub fn count_active(
+    conn: &mut DbConnection,
+    organization_id: OrganizationId,
+    start_time: DateTime,
+    end_time: DateTime,
+) -> Result<u32, HttpError> {
+    count_inner(conn, organization_id, start_time, end_time, None)
+}
+
+/// Count an organization's monthly-active series in `[start_time, end_time]`,
+/// restricted to private projects.
+///
+/// Not currently used for billing (Pro bills all visibilities via [`count_active`]);
+/// kept for a private-only view. Mirrors `QueryMetric::private_usage`.
+pub fn count_active_private(
+    conn: &mut DbConnection,
+    organization_id: OrganizationId,
+    start_time: DateTime,
+    end_time: DateTime,
+) -> Result<u32, HttpError> {
+    count_inner(
+        conn,
+        organization_id,
+        start_time,
+        end_time,
+        Some(Visibility::Private),
+    )
+}
+
+/// Count an organization's monthly-active series whose `last_seen` falls within
+/// `[start_time, end_time]`, optionally restricted to a project `visibility`.
+///
+/// Soft-deleted projects are excluded (their series stop billing). The
+/// `(organization_id, last_seen)` index makes the org-and-window filter a single
+/// range scan; the join to `project` only applies the visibility and soft-delete
+/// filters.
+fn count_inner(
+    conn: &mut DbConnection,
+    organization_id: OrganizationId,
+    start_time: DateTime,
+    end_time: DateTime,
+    visibility: Option<Visibility>,
+) -> Result<u32, HttpError> {
+    let mut query = series_table::table
+        .inner_join(schema::project::table)
+        .filter(series_table::organization_id.eq(organization_id))
+        .filter(series_table::last_seen.ge(start_time))
+        .filter(series_table::last_seen.le(end_time))
+        .filter(schema::project::deleted.is_null())
+        .select(diesel::dsl::count_star())
+        .into_boxed();
+    if let Some(visibility) = visibility {
+        query = query.filter(schema::project::visibility.eq(visibility));
+    }
+    query
+        .get_result::<i64>(conn)
+        .map_err(|e| {
+            issue_error(
+                "Failed to count active series",
+                &format!(
+                    "Failed to count active series (visibility: {visibility:?}) for organization ({organization_id}) between {start_time} and {end_time}."
+                ),
+                e,
+            )
+        })?
+        .try_into()
+        .map_err(|e| {
+            issue_error(
+                "Failed to count active series",
+                &format!(
+                    "Failed to count active series (visibility: {visibility:?}) for organization ({organization_id}) between {start_time} and {end_time}."
+                ),
+                e,
+            )
+        })
+}
+
+/// Test-only oracle: the distinct-series count the cache must equal, computed from
+/// scratch over the raw metric rows in plain Rust (independent of the cache's SQL).
+///
+/// Counts distinct `(testbed, benchmark, measure)` whose latest `report.created`
+/// falls in `[start_time, end_time]`, excluding soft-deleted projects, optionally
+/// restricted to a `visibility`. Never run at runtime; [`count_inner`] is the
+/// runtime source and this pins it in tests (shared with the ingest tests).
+#[cfg(test)]
+pub(crate) fn oracle_count(
+    conn: &mut DbConnection,
+    organization_id: OrganizationId,
+    start_time: DateTime,
+    end_time: DateTime,
+    visibility: Option<Visibility>,
+) -> u32 {
+    use std::collections::HashMap;
+
+    let rows: Vec<(
+        TestbedId,
+        BenchmarkId,
+        MeasureId,
+        DateTime,
+        Visibility,
+        Option<DateTime>,
+    )> = schema::metric::table
+        .inner_join(
+            schema::report_benchmark::table
+                .inner_join(schema::benchmark::table.inner_join(schema::project::table))
+                .inner_join(schema::report::table),
+        )
+        .filter(schema::report::project_id.eq(schema::project::id))
+        .filter(schema::project::organization_id.eq(organization_id))
+        .select((
+            schema::report::testbed_id,
+            schema::report_benchmark::benchmark_id,
+            schema::metric::measure_id,
+            schema::report::created,
+            schema::project::visibility,
+            schema::project::deleted,
+        ))
+        .load(conn)
+        .expect("Failed to load metric rows for oracle");
+
+    // Reduce to the latest creation time per series, applying the same project filters
+    // the cache read applies.
+    let mut last_by_series: HashMap<(TestbedId, BenchmarkId, MeasureId), DateTime> = HashMap::new();
+    for (testbed_id, benchmark_id, measure_id, created_row, project_visibility, deleted) in rows {
+        if deleted.is_some() {
+            continue;
+        }
+        // `Visibility` has no `PartialEq`; compare discriminants (it is a unit enum).
+        if let Some(want) = visibility
+            && project_visibility as i32 != want as i32
+        {
+            continue;
+        }
+        // `DateTime` has no `Ord`; compare by timestamp (whole seconds, exactly what
+        // SQL stores and compares).
+        last_by_series
+            .entry((testbed_id, benchmark_id, measure_id))
+            .and_modify(|latest| {
+                if created_row.timestamp() > latest.timestamp() {
+                    *latest = created_row;
+                }
+            })
+            .or_insert(created_row);
+    }
+
+    u32::try_from(
+        last_by_series
+            .values()
+            .filter(|last_seen| {
+                last_seen.timestamp() >= start_time.timestamp()
+                    && last_seen.timestamp() <= end_time.timestamp()
+            })
+            .count(),
+    )
+    .expect("series count exceeds u32")
+}
+
+#[cfg(test)]
+mod tests {
+    use bencher_json::{DateTime, project::Visibility};
+    use diesel::{ExpressionMethods as _, QueryDsl as _, RunQueryDsl as _};
+
+    use super::{count_active, count_active_private, oracle_count, upsert_series_last_seen};
+    use crate::{
+        context::DbConnection,
+        macros::sql::last_insert_rowid,
+        model::{
+            organization::OrganizationId,
+            project::{
+                ProjectId,
+                benchmark::BenchmarkId,
+                branch::{head::HeadId, version::VersionId},
+                measure::MeasureId,
+                report::ReportId,
+                testbed::TestbedId,
+            },
+        },
+        schema,
+        test_util::{
+            archive_benchmark, archive_measure, archive_testbed, create_benchmark,
+            create_branch_with_head, create_measure, create_report_benchmark, create_testbed,
+            create_version, setup_test_db,
+        },
+    };
+
+    /// A fully set-up project: the ids threaded into every report this test ingests.
+    #[derive(Clone, Copy)]
+    struct Proj {
+        org: OrganizationId,
+        project: ProjectId,
+        head: HeadId,
+        version: VersionId,
+    }
+
+    /// A monotonic source of globally unique UUID strings, so every seeded row gets a
+    /// distinct UUID without hand-tracking ranges per entity kind.
+    struct Uuids(u32);
+
+    impl Uuids {
+        fn next(&mut self) -> String {
+            let n = self.0;
+            self.0 += 1;
+            format!("00000000-0000-0000-0000-{n:012x}")
+        }
+    }
+
+    /// Whole-second timestamp (storage truncates sub-second), strictly ordered by
+    /// `secs` so SQL and Rust comparisons agree.
+    fn at(secs: i64) -> DateTime {
+        DateTime::try_from(1_700_000_000i64 + secs).expect("valid timestamp")
+    }
+
+    fn make_org(conn: &mut DbConnection, uuids: &mut Uuids) -> OrganizationId {
+        let uuid = uuids.next();
+        diesel::insert_into(schema::organization::table)
+            .values((
+                schema::organization::uuid.eq(&uuid),
+                schema::organization::name.eq(format!("Org {uuid}")),
+                schema::organization::slug.eq(format!("org-{uuid}")),
+                schema::organization::created.eq(DateTime::TEST),
+                schema::organization::modified.eq(DateTime::TEST),
+            ))
+            .execute(conn)
+            .expect("insert organization");
+        diesel::select(last_insert_rowid())
+            .get_result(conn)
+            .expect("organization id")
+    }
+
+    fn make_project(
+        conn: &mut DbConnection,
+        uuids: &mut Uuids,
+        org: OrganizationId,
+        visibility: Visibility,
+    ) -> ProjectId {
+        let uuid = uuids.next();
+        diesel::insert_into(schema::project::table)
+            .values((
+                schema::project::uuid.eq(&uuid),
+                schema::project::organization_id.eq(org),
+                schema::project::name.eq(format!("Project {uuid}")),
+                schema::project::slug.eq(format!("project-{uuid}")),
+                schema::project::visibility.eq(visibility),
+                schema::project::created.eq(DateTime::TEST),
+                schema::project::modified.eq(DateTime::TEST),
+            ))
+            .execute(conn)
+            .expect("insert project");
+        diesel::select(last_insert_rowid())
+            .get_result(conn)
+            .expect("project id")
+    }
+
+    fn make_proj(
+        conn: &mut DbConnection,
+        uuids: &mut Uuids,
+        org: OrganizationId,
+        visibility: Visibility,
+    ) -> Proj {
+        let project = make_project(conn, uuids, org, visibility);
+        let branch_uuid = uuids.next();
+        let head_uuid = uuids.next();
+        let version_uuid = uuids.next();
+        let branch =
+            create_branch_with_head(conn, project, &branch_uuid, "main", "main", &head_uuid);
+        let version = create_version(conn, project, &version_uuid, 1, None);
+        Proj {
+            org,
+            project,
+            head: branch.head_id,
+            version,
+        }
+    }
+
+    /// Add a second branch (a new head) to a project, e.g. a PR branch.
+    fn make_head(conn: &mut DbConnection, uuids: &mut Uuids, project: ProjectId) -> HeadId {
+        let branch_uuid = uuids.next();
+        let head_uuid = uuids.next();
+        let slug = uuids.next();
+        create_branch_with_head(conn, project, &branch_uuid, &slug, &slug, &head_uuid).head_id
+    }
+
+    fn make_testbed(conn: &mut DbConnection, uuids: &mut Uuids, project: ProjectId) -> TestbedId {
+        let uuid = uuids.next();
+        create_testbed(conn, project, &uuid, &uuid, &uuid)
+    }
+
+    fn make_benchmark(
+        conn: &mut DbConnection,
+        uuids: &mut Uuids,
+        project: ProjectId,
+    ) -> BenchmarkId {
+        let uuid = uuids.next();
+        create_benchmark(conn, project, &uuid, &uuid, &uuid)
+    }
+
+    fn make_measure(conn: &mut DbConnection, uuids: &mut Uuids, project: ProjectId) -> MeasureId {
+        let uuid = uuids.next();
+        create_measure(conn, project, &uuid, &uuid, &uuid)
+    }
+
+    /// Ingest one report on `head` that produces a single metric for the series
+    /// `(testbed, benchmark, measure)` at `end_time`: it writes the report, the
+    /// `report_benchmark`, the metric (which the oracle reads), and the series upsert
+    /// (the function under test), mirroring the real ingest path.
+    #[expect(
+        clippy::too_many_arguments,
+        reason = "test helper threads the full series key"
+    )]
+    fn ingest_on(
+        conn: &mut DbConnection,
+        uuids: &mut Uuids,
+        proj: Proj,
+        head: HeadId,
+        testbed: TestbedId,
+        benchmark: BenchmarkId,
+        measure: MeasureId,
+        end_time: DateTime,
+    ) {
+        let report_uuid = uuids.next();
+        diesel::insert_into(schema::report::table)
+            .values((
+                schema::report::uuid.eq(&report_uuid),
+                schema::report::project_id.eq(proj.project),
+                schema::report::head_id.eq(head),
+                schema::report::version_id.eq(proj.version),
+                schema::report::testbed_id.eq(testbed),
+                schema::report::adapter.eq(0),
+                schema::report::start_time.eq(end_time),
+                schema::report::end_time.eq(end_time),
+                schema::report::created.eq(end_time),
+            ))
+            .execute(conn)
+            .expect("insert report");
+        let report: ReportId = diesel::select(last_insert_rowid())
+            .get_result(conn)
+            .expect("report id");
+        let rb_uuid = uuids.next();
+        let report_benchmark = create_report_benchmark(conn, &rb_uuid, report, 0, benchmark);
+        let metric_uuid = uuids.next();
+        diesel::insert_into(schema::metric::table)
+            .values((
+                schema::metric::uuid.eq(&metric_uuid),
+                schema::metric::report_benchmark_id.eq(report_benchmark),
+                schema::metric::measure_id.eq(measure),
+                schema::metric::value.eq(1.0f64),
+            ))
+            .execute(conn)
+            .expect("insert metric");
+        upsert_series_last_seen(
+            conn,
+            proj.org,
+            proj.project,
+            testbed,
+            benchmark,
+            measure,
+            end_time,
+        )
+        .expect("upsert series");
+    }
+
+    /// Ingest on the project's primary head.
+    fn ingest(
+        conn: &mut DbConnection,
+        uuids: &mut Uuids,
+        proj: Proj,
+        testbed: TestbedId,
+        benchmark: BenchmarkId,
+        measure: MeasureId,
+        end_time: DateTime,
+    ) {
+        ingest_on(
+            conn, uuids, proj, proj.head, testbed, benchmark, measure, end_time,
+        );
+    }
+
+    /// The whole-of-time window: every seeded series is in range.
+    fn always() -> (DateTime, DateTime) {
+        (at(-1), at(1_000_000))
+    }
+
+    // ----- Identity and counting -----
+
+    #[test]
+    fn iterations_do_not_inflate() {
+        let mut conn = setup_test_db();
+        let mut uuids = Uuids(1);
+        let org = make_org(&mut conn, &mut uuids);
+        let proj = make_proj(&mut conn, &mut uuids, org, Visibility::Public);
+        let (testbed, benchmark, measure) = (
+            make_testbed(&mut conn, &mut uuids, proj.project),
+            make_benchmark(&mut conn, &mut uuids, proj.project),
+            make_measure(&mut conn, &mut uuids, proj.project),
+        );
+        // Three reports (as if three iterations) of the same series.
+        for _ in 0..3 {
+            ingest(
+                &mut conn,
+                &mut uuids,
+                proj,
+                testbed,
+                benchmark,
+                measure,
+                at(0),
+            );
+        }
+        let (start, end) = always();
+        assert_eq!(count_active(&mut conn, org, start, end).unwrap(), 1);
+        assert_eq!(oracle_count(&mut conn, org, start, end, None), 1);
+    }
+
+    #[test]
+    fn multiplicity_is_correct() {
+        let mut conn = setup_test_db();
+        let mut uuids = Uuids(1);
+        let org = make_org(&mut conn, &mut uuids);
+        let proj = make_proj(&mut conn, &mut uuids, org, Visibility::Public);
+        let mut testbeds = Vec::new();
+        let mut benchmarks = Vec::new();
+        let mut measures = Vec::new();
+        for _ in 0..2 {
+            testbeds.push(make_testbed(&mut conn, &mut uuids, proj.project));
+            benchmarks.push(make_benchmark(&mut conn, &mut uuids, proj.project));
+            measures.push(make_measure(&mut conn, &mut uuids, proj.project));
+        }
+        for &testbed in &testbeds {
+            for &benchmark in &benchmarks {
+                for &measure in &measures {
+                    ingest(
+                        &mut conn,
+                        &mut uuids,
+                        proj,
+                        testbed,
+                        benchmark,
+                        measure,
+                        at(0),
+                    );
+                }
+            }
+        }
+        let (start, end) = always();
+        // 2 testbeds x 2 benchmarks x 2 measures = 8 series.
+        assert_eq!(count_active(&mut conn, org, start, end).unwrap(), 8);
+        assert_eq!(oracle_count(&mut conn, org, start, end, None), 8);
+    }
+
+    #[test]
+    fn branch_is_excluded_from_series() {
+        let mut conn = setup_test_db();
+        let mut uuids = Uuids(1);
+        let org = make_org(&mut conn, &mut uuids);
+        let proj = make_proj(&mut conn, &mut uuids, org, Visibility::Public);
+        let (testbed, benchmark, measure) = (
+            make_testbed(&mut conn, &mut uuids, proj.project),
+            make_benchmark(&mut conn, &mut uuids, proj.project),
+            make_measure(&mut conn, &mut uuids, proj.project),
+        );
+        // Same series on `main` plus three PR branches (distinct heads).
+        ingest(
+            &mut conn,
+            &mut uuids,
+            proj,
+            testbed,
+            benchmark,
+            measure,
+            at(0),
+        );
+        for _ in 0..3 {
+            let head = make_head(&mut conn, &mut uuids, proj.project);
+            ingest_on(
+                &mut conn,
+                &mut uuids,
+                proj,
+                head,
+                testbed,
+                benchmark,
+                measure,
+                at(0),
+            );
+        }
+        let (start, end) = always();
+        assert_eq!(count_active(&mut conn, org, start, end).unwrap(), 1);
+        assert_eq!(oracle_count(&mut conn, org, start, end, None), 1);
+    }
+
+    #[test]
+    fn cross_project_within_org_counts_separately() {
+        let mut conn = setup_test_db();
+        let mut uuids = Uuids(1);
+        let org = make_org(&mut conn, &mut uuids);
+        // Two projects in the same org, each with its own (id-distinct) entities.
+        for _ in 0..2 {
+            let proj = make_proj(&mut conn, &mut uuids, org, Visibility::Public);
+            let testbed = make_testbed(&mut conn, &mut uuids, proj.project);
+            let benchmark = make_benchmark(&mut conn, &mut uuids, proj.project);
+            let measure = make_measure(&mut conn, &mut uuids, proj.project);
+            ingest(
+                &mut conn,
+                &mut uuids,
+                proj,
+                testbed,
+                benchmark,
+                measure,
+                at(0),
+            );
+        }
+        let (start, end) = always();
+        assert_eq!(count_active(&mut conn, org, start, end).unwrap(), 2);
+        assert_eq!(oracle_count(&mut conn, org, start, end, None), 2);
+    }
+
+    #[test]
+    fn archived_entities_that_still_report_count() {
+        let mut conn = setup_test_db();
+        let mut uuids = Uuids(1);
+        let org = make_org(&mut conn, &mut uuids);
+        let proj = make_proj(&mut conn, &mut uuids, org, Visibility::Public);
+        let (testbed, benchmark, measure) = (
+            make_testbed(&mut conn, &mut uuids, proj.project),
+            make_benchmark(&mut conn, &mut uuids, proj.project),
+            make_measure(&mut conn, &mut uuids, proj.project),
+        );
+        archive_testbed(&mut conn, testbed);
+        archive_benchmark(&mut conn, benchmark);
+        archive_measure(&mut conn, measure);
+        // An archived series that still produces a metric is still active.
+        ingest(
+            &mut conn,
+            &mut uuids,
+            proj,
+            testbed,
+            benchmark,
+            measure,
+            at(0),
+        );
+        let (start, end) = always();
+        assert_eq!(count_active(&mut conn, org, start, end).unwrap(), 1);
+        assert_eq!(oracle_count(&mut conn, org, start, end, None), 1);
+    }
+
+    #[test]
+    fn rename_is_a_new_series() {
+        let mut conn = setup_test_db();
+        let mut uuids = Uuids(1);
+        let org = make_org(&mut conn, &mut uuids);
+        let proj = make_proj(&mut conn, &mut uuids, org, Visibility::Public);
+        let testbed = make_testbed(&mut conn, &mut uuids, proj.project);
+        let measure = make_measure(&mut conn, &mut uuids, proj.project);
+        // get_or_create is name-keyed, so a mid-period rename creates a new benchmark
+        // id: the same testbed/measure under two benchmark ids is two series.
+        let benchmark_before = make_benchmark(&mut conn, &mut uuids, proj.project);
+        let benchmark_after = make_benchmark(&mut conn, &mut uuids, proj.project);
+        ingest(
+            &mut conn,
+            &mut uuids,
+            proj,
+            testbed,
+            benchmark_before,
+            measure,
+            at(0),
+        );
+        ingest(
+            &mut conn,
+            &mut uuids,
+            proj,
+            testbed,
+            benchmark_after,
+            measure,
+            at(1),
+        );
+        let (start, end) = always();
+        assert_eq!(count_active(&mut conn, org, start, end).unwrap(), 2);
+        assert_eq!(oracle_count(&mut conn, org, start, end, None), 2);
+    }
+
+    #[test]
+    fn soft_deleted_project_is_excluded() {
+        let mut conn = setup_test_db();
+        let mut uuids = Uuids(1);
+        let org = make_org(&mut conn, &mut uuids);
+        let proj = make_proj(&mut conn, &mut uuids, org, Visibility::Public);
+        let (testbed, benchmark, measure) = (
+            make_testbed(&mut conn, &mut uuids, proj.project),
+            make_benchmark(&mut conn, &mut uuids, proj.project),
+            make_measure(&mut conn, &mut uuids, proj.project),
+        );
+        ingest(
+            &mut conn,
+            &mut uuids,
+            proj,
+            testbed,
+            benchmark,
+            measure,
+            at(0),
+        );
+        let (start, end) = always();
+        assert_eq!(count_active(&mut conn, org, start, end).unwrap(), 1);
+        // Soft-delete the project: its series stop billing.
+        diesel::update(schema::project::table.filter(schema::project::id.eq(proj.project)))
+            .set(schema::project::deleted.eq(Some(DateTime::TEST)))
+            .execute(&mut conn)
+            .expect("soft-delete project");
+        assert_eq!(count_active(&mut conn, org, start, end).unwrap(), 0);
+        assert_eq!(oracle_count(&mut conn, org, start, end, None), 0);
+    }
+
+    // ----- Time and period -----
+
+    #[test]
+    fn boundary_start_is_inclusive() {
+        let mut conn = setup_test_db();
+        let mut uuids = Uuids(1);
+        let org = make_org(&mut conn, &mut uuids);
+        let proj = make_proj(&mut conn, &mut uuids, org, Visibility::Public);
+        let (testbed, benchmark, measure) = (
+            make_testbed(&mut conn, &mut uuids, proj.project),
+            make_benchmark(&mut conn, &mut uuids, proj.project),
+            make_measure(&mut conn, &mut uuids, proj.project),
+        );
+        // last_seen exactly at the window start is counted (`>=`).
+        ingest(
+            &mut conn,
+            &mut uuids,
+            proj,
+            testbed,
+            benchmark,
+            measure,
+            at(10),
+        );
+        assert_eq!(count_active(&mut conn, org, at(10), at(20)).unwrap(), 1);
+        assert_eq!(oracle_count(&mut conn, org, at(10), at(20), None), 1);
+        // Just after the window end is not counted.
+        assert_eq!(count_active(&mut conn, org, at(0), at(9)).unwrap(), 0);
+    }
+
+    #[test]
+    fn period_rollover_resets() {
+        let mut conn = setup_test_db();
+        let mut uuids = Uuids(1);
+        let org = make_org(&mut conn, &mut uuids);
+        let proj = make_proj(&mut conn, &mut uuids, org, Visibility::Public);
+        let (testbed, benchmark, measure) = (
+            make_testbed(&mut conn, &mut uuids, proj.project),
+            make_benchmark(&mut conn, &mut uuids, proj.project),
+            make_measure(&mut conn, &mut uuids, proj.project),
+        );
+        // Seen in period 1, never again.
+        ingest(
+            &mut conn,
+            &mut uuids,
+            proj,
+            testbed,
+            benchmark,
+            measure,
+            at(5),
+        );
+        // Period 1 counts it; period 2 does not.
+        assert_eq!(count_active(&mut conn, org, at(0), at(10)).unwrap(), 1);
+        assert_eq!(count_active(&mut conn, org, at(11), at(20)).unwrap(), 0);
+        assert_eq!(oracle_count(&mut conn, org, at(11), at(20), None), 0);
+    }
+
+    #[test]
+    fn backdated_report_does_not_lower_last_seen() {
+        let mut conn = setup_test_db();
+        let mut uuids = Uuids(1);
+        let org = make_org(&mut conn, &mut uuids);
+        let proj = make_proj(&mut conn, &mut uuids, org, Visibility::Public);
+        let (testbed, benchmark, measure) = (
+            make_testbed(&mut conn, &mut uuids, proj.project),
+            make_benchmark(&mut conn, &mut uuids, proj.project),
+            make_measure(&mut conn, &mut uuids, proj.project),
+        );
+        ingest(
+            &mut conn,
+            &mut uuids,
+            proj,
+            testbed,
+            benchmark,
+            measure,
+            at(50),
+        );
+        // A late, backdated report must not lower the recorded activity (MAX).
+        ingest(
+            &mut conn,
+            &mut uuids,
+            proj,
+            testbed,
+            benchmark,
+            measure,
+            at(10),
+        );
+        // The late window still counts it; the early window does not (it was not
+        // resurrected back to at(10)).
+        assert_eq!(count_active(&mut conn, org, at(40), at(60)).unwrap(), 1);
+        assert_eq!(count_active(&mut conn, org, at(0), at(20)).unwrap(), 0);
+        assert_eq!(oracle_count(&mut conn, org, at(40), at(60), None), 1);
+        assert_eq!(oracle_count(&mut conn, org, at(0), at(20), None), 0);
+    }
+
+    // ----- Visibility split -----
+
+    #[test]
+    fn count_active_and_private_split_by_visibility() {
+        let mut conn = setup_test_db();
+        let mut uuids = Uuids(1);
+        let org = make_org(&mut conn, &mut uuids);
+        for visibility in [Visibility::Public, Visibility::Private] {
+            let proj = make_proj(&mut conn, &mut uuids, org, visibility);
+            let testbed = make_testbed(&mut conn, &mut uuids, proj.project);
+            let benchmark = make_benchmark(&mut conn, &mut uuids, proj.project);
+            let measure = make_measure(&mut conn, &mut uuids, proj.project);
+            ingest(
+                &mut conn,
+                &mut uuids,
+                proj,
+                testbed,
+                benchmark,
+                measure,
+                at(0),
+            );
+        }
+        let (start, end) = always();
+        assert_eq!(count_active(&mut conn, org, start, end).unwrap(), 2);
+        assert_eq!(count_active_private(&mut conn, org, start, end).unwrap(), 1);
+        assert_eq!(oracle_count(&mut conn, org, start, end, None), 2);
+        assert_eq!(
+            oracle_count(&mut conn, org, start, end, Some(Visibility::Private)),
+            1
+        );
+    }
+
+    // ----- Edge cases -----
+
+    #[test]
+    fn zero_when_no_series() {
+        let mut conn = setup_test_db();
+        let mut uuids = Uuids(1);
+        let org = make_org(&mut conn, &mut uuids);
+        // An active org with no metrics this period bills zero series, without panic.
+        let (start, end) = always();
+        assert_eq!(count_active(&mut conn, org, start, end).unwrap(), 0);
+        assert_eq!(count_active_private(&mut conn, org, start, end).unwrap(), 0);
+    }
+
+    #[test]
+    fn org_scoped_count_excludes_other_orgs() {
+        let mut conn = setup_test_db();
+        let mut uuids = Uuids(1);
+        let org_a = make_org(&mut conn, &mut uuids);
+        let org_b = make_org(&mut conn, &mut uuids);
+        for org in [org_a, org_b] {
+            let proj = make_proj(&mut conn, &mut uuids, org, Visibility::Public);
+            let testbed = make_testbed(&mut conn, &mut uuids, proj.project);
+            let benchmark = make_benchmark(&mut conn, &mut uuids, proj.project);
+            let measure = make_measure(&mut conn, &mut uuids, proj.project);
+            ingest(
+                &mut conn,
+                &mut uuids,
+                proj,
+                testbed,
+                benchmark,
+                measure,
+                at(0),
+            );
+        }
+        let (start, end) = always();
+        assert_eq!(count_active(&mut conn, org_a, start, end).unwrap(), 1);
+        assert_eq!(count_active(&mut conn, org_b, start, end).unwrap(), 1);
+    }
+
+    #[test]
+    fn duplicate_upsert_converges_without_error() {
+        let mut conn = setup_test_db();
+        let mut uuids = Uuids(1);
+        let org = make_org(&mut conn, &mut uuids);
+        let proj = make_proj(&mut conn, &mut uuids, org, Visibility::Public);
+        let (testbed, benchmark, measure) = (
+            make_testbed(&mut conn, &mut uuids, proj.project),
+            make_benchmark(&mut conn, &mut uuids, proj.project),
+            make_measure(&mut conn, &mut uuids, proj.project),
+        );
+        // Two bare upserts of the same series key converge to one row, last_seen = MAX,
+        // with no duplicate-key error.
+        upsert_series_last_seen(
+            &mut conn,
+            org,
+            proj.project,
+            testbed,
+            benchmark,
+            measure,
+            at(5),
+        )
+        .unwrap();
+        upsert_series_last_seen(
+            &mut conn,
+            org,
+            proj.project,
+            testbed,
+            benchmark,
+            measure,
+            at(3),
+        )
+        .unwrap();
+        let (start, end) = always();
+        assert_eq!(count_active(&mut conn, org, start, end).unwrap(), 1);
+        // last_seen stayed at the max (5), so the [4, 10] window still counts it.
+        assert_eq!(count_active(&mut conn, org, at(4), at(10)).unwrap(), 1);
+    }
+
+    #[test]
+    fn series_upsert_rolls_back_with_its_transaction() {
+        use diesel::Connection as _;
+
+        let mut conn = setup_test_db();
+        let mut uuids = Uuids(1);
+        let org = make_org(&mut conn, &mut uuids);
+        let proj = make_proj(&mut conn, &mut uuids, org, Visibility::Public);
+        let (testbed, benchmark, measure) = (
+            make_testbed(&mut conn, &mut uuids, proj.project),
+            make_benchmark(&mut conn, &mut uuids, proj.project),
+            make_measure(&mut conn, &mut uuids, proj.project),
+        );
+        let (start, end) = always();
+        // The cache upsert runs inside the caller's transaction (it opens none of its
+        // own), so a failure anywhere in the ingest transaction rolls back the metric
+        // and its series row together: the cache cannot drift from the metrics.
+        let result = conn.transaction::<(), diesel::result::Error, _>(|conn| {
+            ingest_on(
+                conn,
+                &mut uuids,
+                proj,
+                proj.head,
+                testbed,
+                benchmark,
+                measure,
+                at(0),
+            );
+            Err(diesel::result::Error::RollbackTransaction)
+        });
+        assert!(result.is_err());
+        assert_eq!(count_active(&mut conn, org, start, end).unwrap(), 0);
+        assert_eq!(oracle_count(&mut conn, org, start, end, None), 0);
+    }
+
+    #[test]
+    fn entity_delete_removes_series() {
+        let mut conn = setup_test_db();
+        let mut uuids = Uuids(1);
+        let org = make_org(&mut conn, &mut uuids);
+        let proj = make_proj(&mut conn, &mut uuids, org, Visibility::Public);
+        let (testbed, benchmark, measure) = (
+            make_testbed(&mut conn, &mut uuids, proj.project),
+            make_benchmark(&mut conn, &mut uuids, proj.project),
+            make_measure(&mut conn, &mut uuids, proj.project),
+        );
+        ingest(
+            &mut conn,
+            &mut uuids,
+            proj,
+            testbed,
+            benchmark,
+            measure,
+            at(0),
+        );
+        let (start, end) = always();
+        assert_eq!(count_active(&mut conn, org, start, end).unwrap(), 1);
+        // Hard-deleting an entity cascades its series rows away and lowers the current
+        // period's count: the documented, accepted exception to monotonic billing (see
+        // the module docs). Mirror the delete endpoint's reports-first requirement:
+        // remove the benchmark's metric and report_benchmark rows, then the benchmark.
+        diesel::delete(schema::metric::table)
+            .execute(&mut conn)
+            .expect("delete metrics");
+        diesel::delete(schema::report_benchmark::table)
+            .execute(&mut conn)
+            .expect("delete report benchmarks");
+        diesel::delete(schema::benchmark::table.filter(schema::benchmark::id.eq(benchmark)))
+            .execute(&mut conn)
+            .expect("delete benchmark");
+        assert_eq!(count_active(&mut conn, org, start, end).unwrap(), 0);
+        // The cascade removed the cache row itself, not just filtered it from the count.
+        let rows: i64 = super::series_table::table
+            .count()
+            .get_result(&mut conn)
+            .expect("count series rows");
+        assert_eq!(rows, 0);
+    }
+
+    #[test]
+    fn backfill_equals_oracle() {
+        use diesel::sql_query;
+
+        // Mirrors the backfill in migration 2026-06-24-120000_series_last_seen/up.sql.
+        const BACKFILL: &str = "\
+INSERT INTO series_last_seen (organization_id, project_id, testbed_id, benchmark_id, measure_id, last_seen)
+SELECT p.organization_id, p.id, r.testbed_id, rb.benchmark_id, m.measure_id, MAX(r.created)
+FROM metric m
+INNER JOIN report_benchmark rb ON m.report_benchmark_id = rb.id
+INNER JOIN report r ON rb.report_id = r.id
+INNER JOIN benchmark b ON rb.benchmark_id = b.id
+INNER JOIN project p ON b.project_id = p.id
+GROUP BY r.testbed_id, rb.benchmark_id, m.measure_id";
+
+        let mut conn = setup_test_db();
+        let mut uuids = Uuids(1);
+        let org = make_org(&mut conn, &mut uuids);
+        let proj = make_proj(&mut conn, &mut uuids, org, Visibility::Public);
+        let mut testbeds = Vec::new();
+        let mut benchmarks = Vec::new();
+        for _ in 0..2 {
+            testbeds.push(make_testbed(&mut conn, &mut uuids, proj.project));
+            benchmarks.push(make_benchmark(&mut conn, &mut uuids, proj.project));
+        }
+        let measure = make_measure(&mut conn, &mut uuids, proj.project);
+        // Seed metrics WITHOUT the cache, then clear the cache and run the migration's
+        // backfill SQL to prove it reconstructs the same counts as the oracle, with the
+        // latest report.created per series.
+        ingest(
+            &mut conn,
+            &mut uuids,
+            proj,
+            testbeds[0],
+            benchmarks[0],
+            measure,
+            at(10),
+        );
+        ingest(
+            &mut conn,
+            &mut uuids,
+            proj,
+            testbeds[0],
+            benchmarks[0],
+            measure,
+            at(30),
+        );
+        ingest(
+            &mut conn,
+            &mut uuids,
+            proj,
+            testbeds[1],
+            benchmarks[1],
+            measure,
+            at(12),
+        );
+        diesel::delete(super::series_table::table)
+            .execute(&mut conn)
+            .expect("clear cache");
+
+        sql_query(BACKFILL)
+            .execute(&mut conn)
+            .expect("run backfill");
+
+        let (start, end) = always();
+        assert_eq!(count_active(&mut conn, org, start, end).unwrap(), 2);
+        assert_eq!(oracle_count(&mut conn, org, start, end, None), 2);
+        // Backfilled last_seen is the latest (30), so the early-only window excludes it.
+        assert_eq!(count_active(&mut conn, org, at(0), at(15)).unwrap(), 1);
+        assert_eq!(oracle_count(&mut conn, org, at(0), at(15), None), 1);
+    }
+
+    // ----- Property test: the backbone -----
+
+    /// Deterministic `SplitMix64`, seeded so the property test is reproducible (no
+    /// wall-clock, no external RNG).
+    struct Rng(u64);
+
+    impl Rng {
+        fn next_u64(&mut self) -> u64 {
+            self.0 = self.0.wrapping_add(0x9E37_79B9_7F4A_7C15);
+            let mut z = self.0;
+            z = (z ^ (z >> 30)).wrapping_mul(0xBF58_476D_1CE4_E5B9);
+            z = (z ^ (z >> 27)).wrapping_mul(0x94D0_49BB_1331_11EB);
+            z ^ (z >> 31)
+        }
+
+        fn below(&mut self, n: u64) -> usize {
+            usize::try_from(self.next_u64() % n).expect("fits usize")
+        }
+    }
+
+    #[test]
+    fn cache_equals_oracle_property() {
+        // One public and one private project, each with a small pool of entities, plus
+        // an extra head per project so branch variation is exercised.
+        struct Pool {
+            proj: Proj,
+            heads: Vec<HeadId>,
+            testbeds: Vec<TestbedId>,
+            benchmarks: Vec<BenchmarkId>,
+            measures: Vec<MeasureId>,
+        }
+
+        let mut conn = setup_test_db();
+        let mut uuids = Uuids(1);
+        let mut rng = Rng(0x5EED);
+        let org = make_org(&mut conn, &mut uuids);
+
+        let mut pools: Vec<Pool> = Vec::new();
+        for visibility in [Visibility::Public, Visibility::Private] {
+            let proj = make_proj(&mut conn, &mut uuids, org, visibility);
+            let heads = vec![proj.head, make_head(&mut conn, &mut uuids, proj.project)];
+            let mut testbeds = Vec::new();
+            let mut benchmarks = Vec::new();
+            let mut measures = Vec::new();
+            for _ in 0..3 {
+                testbeds.push(make_testbed(&mut conn, &mut uuids, proj.project));
+                benchmarks.push(make_benchmark(&mut conn, &mut uuids, proj.project));
+                measures.push(make_measure(&mut conn, &mut uuids, proj.project));
+            }
+            pools.push(Pool {
+                proj,
+                heads,
+                testbeds,
+                benchmarks,
+                measures,
+            });
+        }
+
+        for round in 0..300 {
+            // Pick a random series in a random project and ingest at a random time.
+            let (proj, head, testbed, benchmark, measure, end_time) = {
+                let pool = &pools[rng.below(pools.len() as u64)];
+                (
+                    pool.proj,
+                    pool.heads[rng.below(pool.heads.len() as u64)],
+                    pool.testbeds[rng.below(pool.testbeds.len() as u64)],
+                    pool.benchmarks[rng.below(pool.benchmarks.len() as u64)],
+                    pool.measures[rng.below(pool.measures.len() as u64)],
+                    at(i64::try_from(rng.below(100)).expect("fits i64")),
+                )
+            };
+            ingest_on(
+                &mut conn, &mut uuids, proj, head, testbed, benchmark, measure, end_time,
+            );
+
+            // Periodically assert the invariant over several random windows.
+            if round % 7 == 0 {
+                for _ in 0..4 {
+                    let a = i64::try_from(rng.below(100)).expect("fits i64");
+                    let b = i64::try_from(rng.below(100)).expect("fits i64");
+                    let (start, end) = (at(a.min(b)), at(a.max(b)));
+                    assert_eq!(
+                        count_active(&mut conn, org, start, end).unwrap(),
+                        oracle_count(&mut conn, org, start, end, None),
+                        "count_active != oracle at round {round} window [{a}, {b}]",
+                    );
+                    assert_eq!(
+                        count_active_private(&mut conn, org, start, end).unwrap(),
+                        oracle_count(&mut conn, org, start, end, Some(Visibility::Private)),
+                        "count_active_private != oracle at round {round} window [{a}, {b}]",
+                    );
+                }
+            }
+        }
+    }
+}
diff --git a/lib/bencher_schema/src/model/runner/job.rs b/lib/bencher_schema/src/model/runner/job.rs
index 0a6770ca6..9411443b7 100644
--- a/lib/bencher_schema/src/model/runner/job.rs
+++ b/lib/bencher_schema/src/model/runner/job.rs
@@ -469,9 +469,7 @@ fn resolve_timeout(requested: Option<Timeout>, plan_kind: &PlanKind, is_claimed:
     }
     match plan_kind {
         PlanKind::None => requested.map_or(Timeout::FREE_MAX, |t| t.clamp_max(Timeout::FREE_MAX)),
-        PlanKind::Metered(_, _) | PlanKind::Licensed(_) => {
-            requested.unwrap_or(Timeout::PLUS_DEFAULT)
-        },
+        PlanKind::Metered(_) | PlanKind::Licensed(_) => requested.unwrap_or(Timeout::PLUS_DEFAULT),
     }
 }
 
@@ -505,7 +503,7 @@ mod tests {
     use super::*;
     use crate::{
         macros::sql::last_insert_rowid,
-        model::organization::plan::LicenseUsage,
+        model::organization::plan::{LicenseUsage, MeteredPlan},
         test_util::{
             create_base_entities, create_branch_with_head, create_head_version, create_testbed,
             create_version, setup_test_db,
@@ -513,7 +511,12 @@ mod tests {
     };
 
     fn metered_plan() -> PlanKind {
-        PlanKind::Metered("cus_test".into(), PlanLevel::Pro)
+        PlanKind::Metered(MeteredPlan {
+            customer_id: "cus_test".into(),
+            level: PlanLevel::Pro,
+            current_period_start: DateTime::TEST,
+            current_period_end: DateTime::TEST,
+        })
     }
 
     fn licensed_plan(level: PlanLevel) -> PlanKind {
diff --git a/lib/bencher_schema/src/model/server/mod.rs b/lib/bencher_schema/src/model/server/mod.rs
index 2eef67fd5..5babe6a87 100644
--- a/lib/bencher_schema/src/model/server/mod.rs
+++ b/lib/bencher_schema/src/model/server/mod.rs
@@ -3,4 +3,4 @@ mod plus;
 
 pub use backup::{ServerBackup, ServerBackupError};
 #[cfg(feature = "plus")]
-pub use plus::{QueryServer, ServerId, spawn_credit_grants};
+pub use plus::{QueryServer, ServerId};
diff --git a/lib/bencher_schema/src/model/server/plus/credit.rs b/lib/bencher_schema/src/model/server/plus/credit.rs
deleted file mode 100644
index f8b7ccb09..000000000
--- a/lib/bencher_schema/src/model/server/plus/credit.rs
+++ /dev/null
@@ -1,196 +0,0 @@
-#![cfg(feature = "plus")]
-
-use std::cmp;
-use std::path::PathBuf;
-
-use bencher_billing::Biller;
-use bencher_json::PlanStatus;
-use chrono::{Duration, NaiveTime, Utc};
-use diesel::Connection as _;
-use slog::Logger;
-
-use crate::context::DbConnection;
-use crate::model::organization::plan::QueryPlan;
-
-use super::configure_standalone_connection;
-
-/// Time of day (UTC) at which the daily credit sweep runs. A fixed time so the
-/// sweep happens at the same time every day rather than drifting with server
-/// restarts.
-const CREDIT_SWEEP_TIME: NaiveTime = NaiveTime::MIN;
-
-/// How long to wait before retrying a failed database connection during the daily
-/// credit sweep, so a transient failure does not skip the whole day's run.
-const CREDIT_SWEEP_RETRY_DELAY: std::time::Duration = std::time::Duration::from_mins(1);
-
-/// Maximum number of connection attempts for a single daily sweep before giving
-/// up until the next scheduled run.
-const CREDIT_SWEEP_RETRY_LIMIT: usize = 5;
-
-/// Daily background sweep (Bencher Cloud only) that keeps each metered
-/// subscription's included usage credit granted for the current billing period
-/// and prunes the local plan row once a subscription has fully lapsed. This keeps
-/// credit granting off the report-ingestion hot path, free of billing-side Stripe
-/// calls.
-pub fn spawn_credit_grants(log: Logger, db_path: PathBuf, busy_timeout: u32, biller: Biller) {
-    tokio::spawn(async move {
-        loop {
-            // Sleep until the next occurrence of the daily sweep time (UTC) so the
-            // sweep runs at a fixed time of day, not relative to server start.
-            let now = Utc::now().naive_utc().time();
-            let sleep_time = match now.cmp(&CREDIT_SWEEP_TIME) {
-                cmp::Ordering::Less => CREDIT_SWEEP_TIME - now,
-                cmp::Ordering::Equal => Duration::days(1),
-                cmp::Ordering::Greater => Duration::days(1) - (now - CREDIT_SWEEP_TIME),
-            }
-            .to_std()
-            .unwrap_or(std::time::Duration::from_hours(24));
-            tokio::time::sleep(sleep_time).await;
-
-            // Open a configured connection for this sweep, retrying briefly on a
-            // transient failure rather than skipping the whole day's run.
-            let mut attempt = 0;
-            let conn = loop {
-                attempt += 1;
-                match DbConnection::establish(db_path.to_string_lossy().as_ref()) {
-                    Ok(mut conn) => {
-                        match configure_standalone_connection(&mut conn, busy_timeout) {
-                            Ok(()) => break Some(conn),
-                            Err(e) => slog::error!(
-                                log,
-                                "Failed to configure database connection PRAGMAs for credit sweep: {e}"
-                            ),
-                        }
-                    },
-                    Err(e) => slog::error!(
-                        log,
-                        "Failed to establish database connection for credit sweep: {e}"
-                    ),
-                }
-                if attempt >= CREDIT_SWEEP_RETRY_LIMIT {
-                    break None;
-                }
-                tokio::time::sleep(CREDIT_SWEEP_RETRY_DELAY).await;
-            };
-            let Some(mut conn) = conn else {
-                slog::error!(
-                    log,
-                    "Giving up on credit sweep until the next scheduled run after {CREDIT_SWEEP_RETRY_LIMIT} failed connection attempts"
-                );
-                continue;
-            };
-
-            let plans = match QueryPlan::all_metered(&mut conn) {
-                Ok(plans) => plans,
-                Err(e) => {
-                    slog::error!(log, "Failed to load metered plans for credit sweep: {e}");
-                    continue;
-                },
-            };
-
-            for plan in plans {
-                let Some(metered_plan_id) = plan.metered_plan.clone() else {
-                    continue;
-                };
-                let plan_id = plan.id;
-                match biller.get_metered_plan_status(&metered_plan_id).await {
-                    Ok((status, _, _)) => match credit_sweep_action(status) {
-                        // Active (or trialing): ensure this period's included credit.
-                        // Idempotent, and a no-op for metered plans without a base fee.
-                        CreditSweepAction::EnsureCredit => {
-                            if let Err(e) = biller.ensure_period_credit(&metered_plan_id).await {
-                                slog::warn!(
-                                    log,
-                                    "Failed to ensure period credit for {metered_plan_id}: {e}"
-                                );
-                                #[cfg(feature = "sentry")]
-                                sentry::capture_error(&e);
-                            }
-                        },
-                        // Terminal: prune the local plan row so the org reads as Free
-                        // and we stop querying a dead subscription.
-                        CreditSweepAction::Prune => {
-                            if let Err(e) = QueryPlan::delete(&mut conn, plan_id) {
-                                slog::warn!(
-                                    log,
-                                    "Failed to prune lapsed plan {metered_plan_id}: {e}"
-                                );
-                            }
-                        },
-                        // Recoverable/transient: leave the plan row intact so the
-                        // subscription can recover; grant nothing this run.
-                        CreditSweepAction::Skip => {},
-                    },
-                    Err(e) => {
-                        slog::warn!(log, "Failed to fetch status for {metered_plan_id}: {e}");
-                        #[cfg(feature = "sentry")]
-                        sentry::capture_error(&e);
-                    },
-                }
-            }
-        }
-    });
-}
-
-/// The action the daily sweep takes for a metered plan, derived from its
-/// subscription status. Extracted as a pure function so the branching is
-/// unit-testable without Stripe or a database.
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
-enum CreditSweepAction {
-    /// Active or trialing: ensure this period's included usage credit.
-    EnsureCredit,
-    /// Terminally dead (canceled or incomplete-expired): prune the local plan row.
-    Prune,
-    /// Recoverable or transient (incomplete, past due, paused, unpaid): leave the
-    /// plan row in place and grant nothing; the subscription may still recover.
-    Skip,
-}
-
-/// Decide the sweep action from the subscription status. Only terminal states are
-/// pruned: deleting the local plan row severs the only org-to-subscription link
-/// (no webhook re-creates it), so a subscription that is merely retrying payment
-/// (`PastDue`/`Unpaid`), awaiting its first charge (`Incomplete`), or paused must
-/// be left intact to recover. The exhaustive match forces a deliberate choice if a
-/// new `PlanStatus` is added.
-fn credit_sweep_action(status: PlanStatus) -> CreditSweepAction {
-    match status {
-        PlanStatus::Active | PlanStatus::Trialing => CreditSweepAction::EnsureCredit,
-        PlanStatus::Canceled | PlanStatus::IncompleteExpired => CreditSweepAction::Prune,
-        PlanStatus::Incomplete | PlanStatus::PastDue | PlanStatus::Paused | PlanStatus::Unpaid => {
-            CreditSweepAction::Skip
-        },
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use bencher_json::PlanStatus;
-
-    use super::{CreditSweepAction, credit_sweep_action};
-
-    #[test]
-    fn active_statuses_ensure_credit() {
-        for status in [PlanStatus::Active, PlanStatus::Trialing] {
-            assert_eq!(credit_sweep_action(status), CreditSweepAction::EnsureCredit);
-        }
-    }
-
-    #[test]
-    fn terminal_statuses_prune() {
-        for status in [PlanStatus::Canceled, PlanStatus::IncompleteExpired] {
-            assert_eq!(credit_sweep_action(status), CreditSweepAction::Prune);
-        }
-    }
-
-    #[test]
-    fn recoverable_statuses_skip() {
-        for status in [
-            PlanStatus::Incomplete,
-            PlanStatus::PastDue,
-            PlanStatus::Paused,
-            PlanStatus::Unpaid,
-        ] {
-            assert_eq!(credit_sweep_action(status), CreditSweepAction::Skip);
-        }
-    }
-}
diff --git a/lib/bencher_schema/src/model/server/plus/mod.rs b/lib/bencher_schema/src/model/server/plus/mod.rs
index a1fbb7a21..d838cd51d 100644
--- a/lib/bencher_schema/src/model/server/plus/mod.rs
+++ b/lib/bencher_schema/src/model/server/plus/mod.rs
@@ -26,11 +26,8 @@ use crate::{
     schema::{self, server as server_table},
 };
 
-mod credit;
 mod stats;
 
-pub use credit::spawn_credit_grants;
-
 crate::macros::typed_id::typed_id!(ServerId);
 
 const SERVER_ID: ServerId = ServerId(1);
diff --git a/lib/bencher_schema/src/schema.rs b/lib/bencher_schema/src/schema.rs
index f1d6669e1..75579f553 100644
--- a/lib/bencher_schema/src/schema.rs
+++ b/lib/bencher_schema/src/schema.rs
@@ -325,6 +325,17 @@ diesel::table! {
     }
 }
 
+diesel::table! {
+    series_last_seen (testbed_id, benchmark_id, measure_id) {
+        organization_id -> Integer,
+        project_id -> Integer,
+        testbed_id -> Integer,
+        benchmark_id -> Integer,
+        measure_id -> Integer,
+        last_seen -> BigInt,
+    }
+}
+
 diesel::table! {
     server (id) {
         id -> Integer,
@@ -483,6 +494,11 @@ diesel::joinable!(report_benchmark -> benchmark (benchmark_id));
 diesel::joinable!(report_benchmark -> report (report_id));
 diesel::joinable!(runner_spec -> runner (runner_id));
 diesel::joinable!(runner_spec -> spec (spec_id));
+diesel::joinable!(series_last_seen -> benchmark (benchmark_id));
+diesel::joinable!(series_last_seen -> measure (measure_id));
+diesel::joinable!(series_last_seen -> organization (organization_id));
+diesel::joinable!(series_last_seen -> project (project_id));
+diesel::joinable!(series_last_seen -> testbed (testbed_id));
 diesel::joinable!(sso -> organization (organization_id));
 diesel::joinable!(testbed -> project (project_id));
 diesel::joinable!(testbed -> spec (spec_id));
@@ -522,6 +538,7 @@ diesel::allow_tables_to_appear_in_same_query!(
     report_benchmark,
     runner,
     runner_spec,
+    series_last_seen,
     server,
     spec,
     sso,
diff --git a/plus/api_runners/src/channel.rs b/plus/api_runners/src/channel.rs
index 204f9d8b9..1751ec024 100644
--- a/plus/api_runners/src/channel.rs
+++ b/plus/api_runners/src/channel.rs
@@ -505,11 +505,10 @@ impl BillingState {
                         self.customer = CachedCustomer::None;
                         return Ok(None);
                     };
-                    let (status, customer_id, _level) =
-                        biller.get_metered_plan_status(&metered_plan_id).await?;
-                    if status.is_active() {
-                        self.customer = CachedCustomer::Some(customer_id.clone());
-                        Ok(Some(customer_id))
+                    let billing = biller.get_metered_plan_billing(&metered_plan_id).await?;
+                    if billing.status.is_active() {
+                        self.customer = CachedCustomer::Some(billing.customer_id.clone());
+                        Ok(Some(billing.customer_id))
                     } else {
                         self.customer = CachedCustomer::None;
                         Ok(None)
diff --git a/plus/bencher_billing/src/biller.rs b/plus/bencher_billing/src/biller.rs
index 1e6564b0a..85aa2d5ea 100644
--- a/plus/bencher_billing/src/biller.rs
+++ b/plus/bencher_billing/src/biller.rs
@@ -4,39 +4,33 @@ use std::{
 };
 
 use bencher_json::{
-    Email, Entitlements, LicensedPlanId, MeteredPlanId, OrganizationUuid, PlanLevel, PlanStatus,
+    DateTime, Email, Entitlements, LicensedPlanId, MeteredPlanId, OrganizationUuid, PlanLevel,
+    PlanStatus,
     organization::plan::{
-        JsonCardDetails, JsonPlan, METRICS_METER_NAME, RUNNER_MINUTES_METER_NAME,
+        ACTIVE_SERIES_METER_NAME, JsonCardDetails, JsonPlan, METRICS_METER_NAME,
+        RUNNER_MINUTES_METER_NAME,
     },
     system::{
         config::JsonBilling,
         payment::{JsonCard, JsonCheckout, JsonCustomer},
     },
 };
-use stripe::{Client as StripeClient, IdempotencyKey, RequestStrategy, StripeRequest as _};
+use stripe::Client as StripeClient;
 use stripe_billing::{
     BillingMeterEvent, Subscription, SubscriptionId, SubscriptionItem, SubscriptionStatus,
-    billing_credit_grant::{
-        CreateBillingCreditGrant, CreateBillingCreditGrantAmount,
-        CreateBillingCreditGrantAmountMonetary, CreateBillingCreditGrantAmountType,
-        CreateBillingCreditGrantApplicabilityConfig,
-        CreateBillingCreditGrantApplicabilityConfigScope,
-        CreateBillingCreditGrantApplicabilityConfigScopePrices, ListBillingCreditGrant,
-    },
     billing_meter_event::CreateBillingMeterEvent,
     subscription::{
-        CancelSubscription, CreateSubscription, CreateSubscriptionItems, DiscountsDataParam,
-        RetrieveSubscription, UpdateSubscription,
+        CancelSubscription, CreateSubscription, CreateSubscriptionItems, ListSubscription,
+        ListSubscriptionStatus, RetrieveSubscription, UpdateSubscription,
     },
 };
 use stripe_checkout::{
     CheckoutSessionId, CheckoutSessionMode, CheckoutSessionUiMode,
     checkout_session::{
         CreateCheckoutSession, CreateCheckoutSessionConsentCollection,
-        CreateCheckoutSessionConsentCollectionTermsOfService, CreateCheckoutSessionDiscounts,
-        CreateCheckoutSessionLineItems, CreateCheckoutSessionLineItemsAdjustableQuantity,
-        CreateCheckoutSessionPaymentMethodTypes, CreateCheckoutSessionSubscriptionData,
-        RetrieveCheckoutSession,
+        CreateCheckoutSessionConsentCollectionTermsOfService, CreateCheckoutSessionLineItems,
+        CreateCheckoutSessionLineItemsAdjustableQuantity, CreateCheckoutSessionPaymentMethodTypes,
+        CreateCheckoutSessionSubscriptionData, RetrieveCheckoutSession,
     },
 };
 use stripe_core::customer::{CreateCustomer, ListCustomer};
@@ -62,13 +56,14 @@ const METER_CUSTOMER_KEY: &str = "stripe_customer_id";
 /// Stripe meter event payload key for the usage value.
 const METER_VALUE_KEY: &str = "value";
 
-/// Credit grant metadata key marking a grant as set by Bencher Cloud.
-const CREDIT_BENCHER_CLOUD_KEY: &str = "bencher_cloud";
-/// Credit grant metadata key holding the billing-period start, used to
-/// idempotently grant exactly one included-usage credit per period.
-const CREDIT_PERIOD_KEY: &str = "period_start";
-/// Descriptive name shown in the Stripe Dashboard for the included usage credit.
-const CREDIT_NAME: &str = "Bencher Cloud included usage credit";
+/// Free-trial length (in days) for a new Pro subscription. Set per subscription and per
+/// Checkout Session (a subscription-level trial), not as a per-price default (which is
+/// incompatible with Checkout). A native trial generates no invoice for the period, so
+/// all usage (active series and bare-metal runner minutes) is free during the trial, not
+/// just the base fee. Granted once per customer: a customer with any prior subscription
+/// (any status, including canceled) gets no trial (see [`Biller::trial_period_days`]),
+/// so canceling and re-subscribing cannot restart the trial.
+const PRO_TRIAL_PERIOD_DAYS: u32 = 30;
 
 #[derive(Clone)]
 pub struct Biller {
@@ -76,15 +71,17 @@ pub struct Biller {
     products: Products,
 }
 
-/// Outcome of [`Biller::ensure_period_credit`].
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
-pub enum PeriodCredit {
-    /// A new included-usage credit was granted for the current period.
-    Granted,
-    /// This period's credit already existed; no new grant was created.
-    AlreadyGranted,
-    /// The subscription has no flat base fee, so no credit applies.
-    NotApplicable,
+/// A metered subscription's current billing snapshot, fetched in one call: its
+/// [`PlanStatus`] (the active/lapsed gate), the Stripe [`CustomerId`] to post usage
+/// for, the [`PlanLevel`], and the current period bounds (the active-series count
+/// window).
+#[derive(Debug, Clone)]
+pub struct MeteredPlanBilling {
+    pub status: PlanStatus,
+    pub customer_id: CustomerId,
+    pub level: PlanLevel,
+    pub current_period_start: DateTime,
+    pub current_period_end: DateTime,
 }
 
 /// The configuration price-name key (e.g. `default`) selecting which configured
@@ -113,11 +110,10 @@ impl fmt::Display for PriceName {
 #[derive(Debug, Clone)]
 enum PlusPlan {
     Free,
-    // Bencher Cloud metered tier with a flat monthly base fee and an included
-    // usage credit. Carries the metered metrics price name, which resolves
-    // against the `metrics` product; the base price and trial coupon come from
-    // the `pro` product and are looked up separately. This tier has no licensed
-    // (Self-Hosted) form.
+    // Bencher Cloud self-serve tier. Carries the price name for the single tiered
+    // active-series price (on the `pro` product) that bills both the flat monthly base
+    // fee (tier 1) and the per-series step-ups. Gets a native free trial. This tier has
+    // no licensed (Self-Hosted) form.
     Pro(PriceName),
     Team(PlusUsage),
     Enterprise(PlusUsage),
@@ -166,29 +162,11 @@ impl PlusPlan {
             .ok_or_else(|| BillingError::PriceNotFound(price_name.to_string()))
     }
 
-    // The flat recurring base fee for this plan, if any.
-    fn base_price<'a>(&self, products: &'a Products) -> Result<Option<&'a Price>, BillingError> {
-        match self {
-            Self::Pro(price_name) => products
-                .pro
-                .base
-                .get(price_name.as_str())
-                .map(Some)
-                .ok_or_else(|| BillingError::PriceNotFound(price_name.to_string())),
-            Self::Free | Self::Team(_) | Self::Enterprise(_) => Ok(None),
-        }
-    }
-
-    // Whether this plan carries a flat recurring base fee (and so receives the
-    // free-trial coupon and the monthly included-usage credit).
-    fn has_base_fee(&self) -> bool {
-        matches!(self, Self::Pro(_))
-    }
-
-    // The configured free-trial coupon for this plan's product, if any.
-    fn trial_coupon<'a>(&self, products: &'a Products) -> Option<&'a str> {
+    // The native free-trial length (in days) for this plan, if any. Only the Pro
+    // self-serve tier gets a trial.
+    fn trial_period_days(&self) -> Option<u32> {
         match self {
-            Self::Pro(_) => products.pro.trial_coupon.as_deref(),
+            Self::Pro(_) => Some(PRO_TRIAL_PERIOD_DAYS),
             Self::Free | Self::Team(_) | Self::Enterprise(_) => None,
         }
     }
@@ -199,9 +177,11 @@ impl PlusPlan {
     ) -> Result<(&Price, Option<Entitlements>), BillingError> {
         Ok(match self {
             PlusPlan::Free => return Err(BillingError::ProductLevelFree),
+            // Pro bills on its own tiered active-series price (base fee + step-ups),
+            // which lives on the `pro` product, not the shared `metrics` product.
             PlusPlan::Pro(price_name) => (
                 products
-                    .metrics
+                    .pro
                     .metered
                     .get(price_name.as_str())
                     .ok_or_else(|| BillingError::PriceNotFound(price_name.to_string()))?,
@@ -316,8 +296,7 @@ impl Biller {
             PlusPlan::metered(plan_level, price_name)
         };
         let bare_metal_price = plus_plan.bare_metal_price(&self.products)?;
-        let base_price = plus_plan.base_price(&self.products)?;
-        let trial_coupon = self.trial_coupon(&plus_plan);
+        let trial_period_days = self.trial_period_days(&customer, &plus_plan).await?;
         let (plus_price, entitlements) = plus_plan.into_plus_price(&self.products)?;
 
         let mut plus_line_item = CreateCheckoutSessionLineItems {
@@ -337,17 +316,9 @@ impl Biller {
             ..Default::default()
         };
 
-        // Flat monthly base fee billed alongside the metered usage.
-        let mut line_items = Vec::new();
-        if let Some(base_price) = base_price {
-            line_items.push(CreateCheckoutSessionLineItems {
-                price: Some(base_price.id.to_string()),
-                quantity: Some(1),
-                ..Default::default()
-            });
-        }
-        line_items.push(plus_line_item);
-        line_items.push(bare_metal_line_item);
+        // Pro's tiered active-series price carries the base fee, so there is no
+        // separate base line item.
+        let line_items = vec![plus_line_item, bare_metal_line_item];
 
         let mut consent = CreateCheckoutSessionConsentCollection::new();
         // https://bencher.dev/legal/subscription/
@@ -360,8 +331,11 @@ impl Biller {
                 .into_iter()
                 .collect(),
         );
+        // Native free trial set at the subscription level (Checkout-compatible), not as
+        // a per-price default. No charge during the trial; billing starts at trial end.
+        sub_data.trial_period_days = trial_period_days;
 
-        let mut create_checkout_session = CreateCheckoutSession::new()
+        let create_checkout_session = CreateCheckoutSession::new()
             .ui_mode(CheckoutSessionUiMode::HostedPage)
             .customer(customer.to_string())
             .payment_method_types(vec![CreateCheckoutSessionPaymentMethodTypes::Card])
@@ -371,15 +345,6 @@ impl Biller {
             .consent_collection(consent)
             .subscription_data(sub_data)
             .success_url(return_url);
-        // Free trial: waive the first month's base fee via a one-time coupon.
-        // Metered usage still bills, offset by the monthly included-usage credit.
-        if let Some(coupon) = trial_coupon {
-            create_checkout_session =
-                create_checkout_session.discounts(vec![CreateCheckoutSessionDiscounts {
-                    coupon: Some(coupon),
-                    promotion_code: None,
-                }]);
-        }
         let mut checkout_session = create_checkout_session.send(&self.client).await?;
 
         Ok(JsonCheckout {
@@ -520,25 +485,17 @@ impl Biller {
         plus_plan: PlusPlan,
     ) -> Result<Subscription, BillingError> {
         let bare_metal_price = plus_plan.bare_metal_price(&self.products)?;
-        let base_price = plus_plan.base_price(&self.products)?;
-        let trial_coupon = self.trial_coupon(&plus_plan);
+        let trial_period_days = self.trial_period_days(&customer_id, &plus_plan).await?;
         let (plus_price, entitlements) = plus_plan.into_plus_price(&self.products)?;
 
-        // Flat monthly base fee billed alongside the metered usage.
-        let mut items = Vec::new();
-        if let Some(base_price) = base_price {
-            let mut base_item = CreateSubscriptionItems::new();
-            base_item.price = Some(base_price.id.to_string());
-            base_item.quantity = Some(1);
-            items.push(base_item);
-        }
+        // Pro's tiered active-series price carries the base fee, so there is no
+        // separate base item.
         let mut plus_item = CreateSubscriptionItems::new();
         plus_item.price = Some(plus_price.id.to_string());
         plus_item.quantity = entitlements.map(Into::into);
-        items.push(plus_item);
         let mut bare_metal_item = CreateSubscriptionItems::new();
         bare_metal_item.price = Some(bare_metal_price.id.to_string());
-        items.push(bare_metal_item);
+        let items = vec![plus_item, bare_metal_item];
 
         let mut create_subscription = CreateSubscription::new()
             .customer(customer_id.to_string())
@@ -549,13 +506,10 @@ impl Biller {
                     .into_iter()
                     .collect::<HashMap<String, String>>(),
             );
-        // Free trial: waive the first month's base fee via a one-time coupon.
-        if let Some(coupon) = trial_coupon {
-            create_subscription = create_subscription.discounts(vec![DiscountsDataParam {
-                coupon: Some(coupon),
-                discount: None,
-                promotion_code: None,
-            }]);
+        // Native free trial (subscription-level): no charge during the trial; billing
+        // starts at trial end.
+        if let Some(days) = trial_period_days {
+            create_subscription = create_subscription.trial_period_days(days);
         }
         create_subscription
             .send(&self.client)
@@ -563,6 +517,39 @@ impl Biller {
             .map_err(Into::into)
     }
 
+    /// The native free-trial length for a new subscription: the plan's trial, granted
+    /// only if the customer has no prior subscription. This makes the trial
+    /// once-per-customer: canceling and re-subscribing (the plan POST prunes a
+    /// terminally lapsed plan row) cannot restart it. Plans without a trial skip the
+    /// Stripe lookup entirely. A Stripe error propagates and fails plan creation
+    /// rather than silently granting or denying the trial.
+    async fn trial_period_days(
+        &self,
+        customer_id: &CustomerId,
+        plus_plan: &PlusPlan,
+    ) -> Result<Option<u32>, BillingError> {
+        let Some(days) = plus_plan.trial_period_days() else {
+            return Ok(None);
+        };
+        Ok((!self.customer_has_prior_subscription(customer_id).await?).then_some(days))
+    }
+
+    /// Whether the customer has ever had a subscription, in any status (including
+    /// canceled and incomplete-expired). One existence check is enough, so a single
+    /// result is requested.
+    async fn customer_has_prior_subscription(
+        &self,
+        customer_id: &CustomerId,
+    ) -> Result<bool, BillingError> {
+        let subscriptions = ListSubscription::new()
+            .customer(customer_id.to_string())
+            .status(ListSubscriptionStatus::All)
+            .limit(1)
+            .send(&self.client)
+            .await?;
+        Ok(!subscriptions.data.is_empty())
+    }
+
     pub async fn get_metered_plan(
         &self,
         metered_plan_id: &MeteredPlanId,
@@ -580,22 +567,23 @@ impl Biller {
     }
 
     /// Derive a paid subscription's plan level from its line items. Pro is the only
-    /// paid tier with a flat base fee, so a Pro base-fee item identifies the plan
-    /// level without relying on which Stripe product the metered item sits on.
+    /// paid tier that carries the tiered active-series price (on the `pro` product), so
+    /// a Pro active-series item identifies the plan level; any other paid metered
+    /// subscription is Team.
     fn subscription_plan_level(&self, subscription: &Subscription) -> PlanLevel {
-        let pro_base_price_ids: HashSet<&PriceId> = self
+        let pro_price_ids: HashSet<&PriceId> = self
             .products
             .pro
-            .base
+            .metered
             .values()
             .map(|price| &price.id)
             .collect();
-        let has_base_fee = subscription
+        let is_pro = subscription
             .items
             .data
             .iter()
-            .any(|item| pro_base_price_ids.contains(&item.price.id));
-        plan_level_from_base_fee(has_base_fee)
+            .any(|item| pro_price_ids.contains(&item.price.id));
+        plan_level_from_pro_price(is_pro)
     }
 
     async fn get_plan(&self, subscription_id: &SubscriptionId) -> Result<JsonPlan, BillingError> {
@@ -654,7 +642,11 @@ impl Biller {
             subscription_id,
             subscription.default_payment_method.as_ref(),
         )?;
-        let unit_amount = Self::get_plan_unit_amount(&subscription_item)?;
+        let unit_amount = Self::get_plan_unit_amount(
+            &subscription_item,
+            self.products
+                .tier_base_fee_cents(&subscription_item.price.id),
+        )?;
 
         let status = Self::map_status(&subscription.status);
 
@@ -745,12 +737,20 @@ impl Biller {
         })
     }
 
-    fn get_plan_unit_amount(subscription_item: &SubscriptionItem) -> Result<u64, BillingError> {
+    /// The plan's displayed per-period unit amount (cents). For a flat price this is the
+    /// price's `unit_amount`. A tiered price (the Pro active-series price) exposes no
+    /// flat `unit_amount`, so the caller passes `fallback_cents` (the base monthly fee,
+    /// tier 1 `flat_amount`) to use instead.
+    fn get_plan_unit_amount(
+        subscription_item: &SubscriptionItem,
+        fallback_cents: Option<i64>,
+    ) -> Result<u64, BillingError> {
         let price = &subscription_item.price;
-        let Some(unit_amount) = price.unit_amount else {
-            return Err(BillingError::NoUnitAmount(price.id.clone()));
-        };
-        u64::try_from(unit_amount).map_err(Into::into)
+        let cents = price
+            .unit_amount
+            .or(fallback_cents)
+            .ok_or_else(|| BillingError::NoUnitAmount(price.id.clone()))?;
+        u64::try_from(cents).map_err(Into::into)
     }
 
     fn get_subscription_item(
@@ -800,17 +800,70 @@ impl Biller {
         }
     }
 
-    pub async fn get_metered_plan_status(
+    /// One-fetch billing snapshot: status, the Stripe customer to post usage for, the
+    /// plan level, and the current period bounds. Expands the subscription's items
+    /// because the period bounds live on them (all items share one period). Used on the
+    /// report path to gate access and to supply the active-series count window.
+    pub async fn get_metered_plan_billing(
         &self,
         metered_plan_id: &MeteredPlanId,
-    ) -> Result<(PlanStatus, CustomerId, PlanLevel), BillingError> {
+    ) -> Result<MeteredPlanBilling, BillingError> {
         let subscription_id: SubscriptionId = metered_plan_id.as_ref().into();
-        let subscription = self.get_subscription(&subscription_id).await?;
-        Ok((
+        let subscription = self
+            .get_subscription_expand(&subscription_id, vec!["items".into()])
+            .await?;
+        let (status, customer_id, level) = self.subscription_status_snapshot(&subscription);
+        // All subscription items share the same billing period.
+        let item = subscription
+            .items
+            .data
+            .first()
+            .ok_or_else(|| BillingError::NoSubscriptionItem(subscription_id.clone()))?;
+        let current_period_start = item.current_period_start.try_into().map_err(|e| {
+            BillingError::DateTime(subscription_id.clone(), item.current_period_start, e)
+        })?;
+        let current_period_end = item.current_period_end.try_into().map_err(|e| {
+            BillingError::DateTime(subscription_id.clone(), item.current_period_end, e)
+        })?;
+        Ok(MeteredPlanBilling {
+            status,
+            customer_id,
+            level,
+            current_period_start,
+            current_period_end,
+        })
+    }
+
+    /// Resolve a metered subscription's live status for re-subscription gating:
+    /// `Some(status)` when the subscription exists, and `None` when Stripe reports it no
+    /// longer exists (a definitive 404, safe to treat as gone). Returns an error only for
+    /// indeterminate failures (network, 5xx, rate limit), so a transient outage is never
+    /// mistaken for "gone" (which would risk pruning a still-live subscription). The
+    /// caller distinguishes terminal from recoverable (dunning) statuses.
+    pub async fn metered_plan_status(
+        &self,
+        metered_plan_id: &MeteredPlanId,
+    ) -> Result<Option<PlanStatus>, BillingError> {
+        let subscription_id: SubscriptionId = metered_plan_id.as_ref().into();
+        match self.get_subscription(&subscription_id).await {
+            Ok(subscription) => Ok(Some(Self::map_status(&subscription.status))),
+            Err(BillingError::Stripe(stripe::StripeError::Stripe(_, 404))) => Ok(None),
+            Err(e) => Err(e),
+        }
+    }
+
+    /// Derive a subscription's status, customer, and plan level for
+    /// [`Self::get_metered_plan_billing`]. Deriving the level does not require the
+    /// subscription's items to be expanded (item price ids are present without it).
+    fn subscription_status_snapshot(
+        &self,
+        subscription: &Subscription,
+    ) -> (PlanStatus, CustomerId, PlanLevel) {
+        (
             Self::map_status(&subscription.status),
             subscription.customer.id().clone(),
-            self.subscription_plan_level(&subscription),
-        ))
+            self.subscription_plan_level(subscription),
+        )
     }
 
     pub async fn get_licensed_plan_status(
@@ -840,188 +893,64 @@ impl Biller {
         customer_id: &CustomerId,
         quantity: u32,
     ) -> Result<BillingMeterEvent, BillingError> {
-        self.record_metered_usage(METRICS_METER_NAME, customer_id, quantity)
+        self.record_metered_usage(METRICS_METER_NAME, customer_id, quantity, None)
             .await
     }
 
+    /// Post an organization's cumulative period-to-date active-series count to the
+    /// `active_series` meter (which backs the Pro tiered price), stamped with `when` (the
+    /// time the count was taken). Billed with `last` aggregation: the explicit timestamp
+    /// orders concurrent posts by `when` (to one-second granularity) rather than by
+    /// Stripe's receipt order, so reports more than a second apart bill the later count
+    /// deterministically; a same-second tie self-heals on the next report. Posting after
+    /// each report makes the final post of the period the period total, and a missed post
+    /// self-heals on the next report. Mirrors [`Self::record_metrics_usage`].
+    pub async fn record_series_usage(
+        &self,
+        customer_id: &CustomerId,
+        quantity: u32,
+        when: DateTime,
+    ) -> Result<BillingMeterEvent, BillingError> {
+        self.record_metered_usage(
+            ACTIVE_SERIES_METER_NAME,
+            customer_id,
+            quantity,
+            Some(when.timestamp()),
+        )
+        .await
+    }
+
     pub async fn record_runner_usage(
         &self,
         customer_id: &CustomerId,
         minutes: u32,
     ) -> Result<BillingMeterEvent, BillingError> {
-        self.record_metered_usage(RUNNER_MINUTES_METER_NAME, customer_id, minutes)
+        self.record_metered_usage(RUNNER_MINUTES_METER_NAME, customer_id, minutes, None)
             .await
     }
 
+    /// `timestamp` (Unix seconds) is set on the meter event only for `last`-aggregation
+    /// meters (the `active_series` meter), where it makes ordering deterministic; the
+    /// `sum`-aggregation meters (metrics, runner minutes) pass `None` and use Stripe's
+    /// receipt time.
     async fn record_metered_usage(
         &self,
         meter_name: &str,
         customer_id: &CustomerId,
         quantity: u32,
+        timestamp: Option<i64>,
     ) -> Result<BillingMeterEvent, BillingError> {
-        CreateBillingMeterEvent::new(
+        let mut event = CreateBillingMeterEvent::new(
             meter_name,
             HashMap::from([
                 (METER_CUSTOMER_KEY.to_owned(), customer_id.to_string()),
                 (METER_VALUE_KEY.to_owned(), quantity.to_string()),
             ]),
-        )
-        .send(&self.client)
-        .await
-        .map_err(Into::into)
-    }
-
-    // The first-period base-fee waiver coupon for this plan: returned only when
-    // the plan has a base fee and its product configures a trial coupon.
-    fn trial_coupon(&self, plus_plan: &PlusPlan) -> Option<String> {
-        if !plus_plan.has_base_fee() {
-            return None;
-        }
-        plus_plan
-            .trial_coupon(&self.products)
-            .map(ToOwned::to_owned)
-    }
-
-    /// Idempotently grant the current billing period's included usage credit for a
-    /// metered subscription that carries a flat base fee. The credit equals that
-    /// base fee and forms a single fungible pool scoped to the subscription's
-    /// metered usage prices, expiring at period end (use-it-or-lose-it). A no-op if
-    /// the subscription has no base fee or this period's grant already exists.
-    pub async fn ensure_period_credit(
-        &self,
-        metered_plan_id: &MeteredPlanId,
-    ) -> Result<PeriodCredit, BillingError> {
-        let subscription_id: SubscriptionId = metered_plan_id.as_ref().into();
-        let subscription = self
-            .get_subscription_expand(&subscription_id, vec!["items".into()])
-            .await?;
-
-        // The included credit equals the subscription's flat base fee. A metered
-        // plan without a configured base price (e.g. a grandfathered Team plan) is
-        // a safe no-op.
-        let base_cents: HashMap<&PriceId, i64> = self
-            .products
-            .all_base_prices()
-            .filter_map(|price| price.unit_amount.map(|cents| (&price.id, cents)))
-            .collect();
-        let Some(value_cents) = matched_base_cents(
-            subscription.items.data.iter().map(|item| &item.price.id),
-            &base_cents,
-        ) else {
-            return Ok(PeriodCredit::NotApplicable);
-        };
-
-        let customer_id = subscription.customer.id().clone();
-        // All subscription items share the same billing period.
-        let item = subscription
-            .items
-            .data
-            .first()
-            .ok_or_else(|| BillingError::NoSubscriptionItem(subscription_id.clone()))?;
-        let period_start = item.current_period_start;
-        let period_end = item.current_period_end;
-
-        // The credit applies to the metered usage prices (every item except the
-        // flat base fee), forming a single fungible pool.
-        let credit_price_ids: Vec<String> = subscription
-            .items
-            .data
-            .iter()
-            .map(|item| &item.price.id)
-            .filter(|id| !base_cents.contains_key(*id))
-            .map(ToString::to_string)
-            .collect();
-
-        // Dedup: skip if a Bencher-managed grant for this period already exists.
-        // Stripe lists grants newest-first by `created` and cannot filter by
-        // metadata, and a customer may have unrelated grants (manual dashboard
-        // credits, refunds), so scan a page of recent grants for our marker plus
-        // period rather than trusting the single newest. Our current-period grant
-        // is recent, so it sits well within one page. The idempotency key guards
-        // concurrent attempts; this guards across the sweep's daily cadence.
-        let marker = period_start.to_string();
-        let grants = ListBillingCreditGrant::new()
-            .customer(customer_id.to_string())
-            .limit(100)
-            .send(&self.client)
-            .await?;
-        if period_already_granted(
-            grants.data.iter().map(|grant| {
-                (
-                    grant
-                        .metadata
-                        .get(CREDIT_BENCHER_CLOUD_KEY)
-                        .map(String::as_str),
-                    grant.metadata.get(CREDIT_PERIOD_KEY).map(String::as_str),
-                )
-            }),
-            &marker,
-        ) {
-            return Ok(PeriodCredit::AlreadyGranted);
+        );
+        if let Some(timestamp) = timestamp {
+            event = event.timestamp(timestamp);
         }
-
-        self.create_credit_grant(
-            &customer_id,
-            value_cents,
-            credit_price_ids,
-            period_start,
-            period_end,
-        )
-        .await?;
-        Ok(PeriodCredit::Granted)
-    }
-
-    async fn create_credit_grant(
-        &self,
-        customer_id: &CustomerId,
-        value_cents: i64,
-        credit_price_ids: Vec<String>,
-        period_start: stripe_types::Timestamp,
-        period_end: stripe_types::Timestamp,
-    ) -> Result<(), BillingError> {
-        let prices = credit_price_ids
-            .into_iter()
-            .map(CreateBillingCreditGrantApplicabilityConfigScopePrices::new)
-            .collect();
-        let scope = CreateBillingCreditGrantApplicabilityConfigScope {
-            price_type: None,
-            prices: Some(prices),
-        };
-        let amount = CreateBillingCreditGrantAmount {
-            monetary: Some(CreateBillingCreditGrantAmountMonetary::new(
-                Currency::USD,
-                value_cents,
-            )),
-            type_: CreateBillingCreditGrantAmountType::Monetary,
-        };
-        let metadata = HashMap::from([
-            (CREDIT_BENCHER_CLOUD_KEY.to_owned(), "true".to_owned()),
-            (CREDIT_PERIOD_KEY.to_owned(), period_start.to_string()),
-        ]);
-        // Idempotency key from customer + period so concurrent grant attempts
-        // (e.g. plan creation overlapping the daily sweep) collapse to one grant.
-        // The list-check above handles dedup across periods (beyond the key's TTL).
-        let idempotency_key =
-            IdempotencyKey::new(format!("bencher-credit:{customer_id}:{period_start}"))?;
-        // Do not set `effective_at`. Stripe rejects a timestamp before its server
-        // "now", and `period_start` is in the past for an already-active
-        // subscription. Omitting it defaults to Stripe's "now" (also sidestepping
-        // clock skew between us and Stripe). The credit still covers this period's
-        // metered usage, which is invoiced at `expires_at` (period end).
-        CreateBillingCreditGrant::new(
-            amount,
-            CreateBillingCreditGrantApplicabilityConfig { scope },
-        )
-        .customer(customer_id.to_string())
-        .expires_at(period_end)
-        .metadata(metadata)
-        .name(CREDIT_NAME)
-        .customize()
-        .request_strategy(RequestStrategy::Idempotent(idempotency_key))
-        .send(&self.client)
-        .await
-        .map(|_grant| ())
-        .map_err(Into::into)
+        event.send(&self.client).await.map_err(Into::into)
     }
 
     /// Immediately cancel a metered subscription. Used by the admin-only DELETE
@@ -1037,7 +966,7 @@ impl Biller {
     /// Schedule (`true`) or clear (`false`) a metered subscription's
     /// cancel-at-period-end. With `true` the customer keeps access through the
     /// period they have already paid for; the subscription stays active until it
-    /// then lapses (reconciled lazily on read and by the daily sweep). With
+    /// then lapses (reconciled lazily on read). With
     /// `false` a scheduled cancellation is cleared, resuming the subscription.
     /// Used by the self-service PATCH path.
     pub async fn set_metered_cancel_at_period_end(
@@ -1072,41 +1001,17 @@ impl Biller {
     }
 }
 
-/// A paid subscription's plan level. Pro is the only paid tier that carries a flat
-/// base fee (see `PlusPlan::base_price`), so a subscription with a configured
-/// base-fee item is Pro; any other paid metered subscription is Team. This is
-/// independent of which Stripe product the metered-metrics item sits on.
-fn plan_level_from_base_fee(has_base_fee: bool) -> PlanLevel {
-    if has_base_fee {
+/// A paid subscription's plan level. Pro is the only paid tier that carries the tiered
+/// active-series price (on the `pro` product), so a subscription with that item is Pro;
+/// any other paid metered subscription is Team.
+fn plan_level_from_pro_price(is_pro: bool) -> PlanLevel {
+    if is_pro {
         PlanLevel::Pro
     } else {
         PlanLevel::Team
     }
 }
 
-/// The included-credit amount (the matched flat base price's cents) for a
-/// subscription, given a map of configured base price ID to cents. `None` if the
-/// subscription carries no configured base fee. Gates the included-usage credit to
-/// base-fee subscriptions and sizes it to the base fee.
-fn matched_base_cents<'a>(
-    mut item_price_ids: impl Iterator<Item = &'a PriceId>,
-    base_cents: &HashMap<&'a PriceId, i64>,
-) -> Option<i64> {
-    item_price_ids.find_map(|id| base_cents.get(id).copied())
-}
-
-/// Whether a Bencher-managed credit grant for the given billing-period marker
-/// already exists. Each item is `(bencher_cloud_marker, period_start)`; only
-/// grants carrying our marker count, so unrelated grants (manual credits,
-/// refunds) are ignored even when newer. (Per-period dedup for
-/// `ensure_period_credit`.)
-fn period_already_granted<'a>(
-    mut grants: impl Iterator<Item = (Option<&'a str>, Option<&'a str>)>,
-    marker: &'a str,
-) -> bool {
-    grants.any(|(managed, period)| managed == Some("true") && period == Some(marker))
-}
-
 fn into_payment_card(card: JsonCard) -> CreatePaymentMethodCardDetailsParams {
     let JsonCard {
         number,
@@ -1128,7 +1033,10 @@ mod tests {
 
     use bencher_json::{
         Entitlements, OrganizationUuid, PlanLevel, PlanStatus, UserUuid,
-        organization::plan::{DEFAULT_PRICE_NAME, METRICS_METER_NAME, RUNNER_MINUTES_METER_NAME},
+        organization::plan::{
+            ACTIVE_SERIES_METER_NAME, DEFAULT_PRICE_NAME, METRICS_METER_NAME,
+            RUNNER_MINUTES_METER_NAME,
+        },
         system::{
             config::{JsonBilling, JsonProduct, JsonProducts},
             payment::{JsonCard, JsonCustomer},
@@ -1141,82 +1049,44 @@ mod tests {
 
     use rustls::crypto::aws_lc_rs::default_provider;
 
-    use crate::{Biller, PeriodCredit};
-
-    use super::{
-        METER_CUSTOMER_KEY, METER_VALUE_KEY, matched_base_cents, period_already_granted,
-        plan_level_from_base_fee,
-    };
-
-    #[test]
-    fn matched_base_cents_returns_base_fee() {
-        let base: stripe_product::PriceId = "price_base".parse().unwrap();
-        let metered: stripe_product::PriceId = "price_metered".parse().unwrap();
-        let other: stripe_product::PriceId = "price_other".parse().unwrap();
-        let base_cents: HashMap<&stripe_product::PriceId, i64> = HashMap::from([(&base, 2_000)]);
-        // A subscription carrying the base price yields that base price's cents.
-        assert_eq!(
-            matched_base_cents([&metered, &base].into_iter(), &base_cents),
-            Some(2_000),
-        );
-        // A subscription without a configured base price yields nothing.
-        assert_eq!(
-            matched_base_cents([&metered, &other].into_iter(), &base_cents),
-            None,
-        );
-    }
+    use crate::Biller;
 
-    #[test]
-    fn period_already_granted_matches_marker() {
-        // A Bencher-managed grant for the period counts.
-        assert!(period_already_granted(
-            [(Some("true"), Some("100")), (None, None)].into_iter(),
-            "100",
-        ));
-        // A non-Bencher grant on top does not hide a real Bencher grant deeper down.
-        assert!(period_already_granted(
-            [(None, Some("100")), (Some("true"), Some("100"))].into_iter(),
-            "100",
-        ));
-        // A non-Bencher grant for the period (no marker) is not counted as ours.
-        assert!(!period_already_granted(
-            [(None, Some("100"))].into_iter(),
-            "100",
-        ));
-        // A different period does not match.
-        assert!(!period_already_granted(
-            [(Some("true"), Some("100"))].into_iter(),
-            "200",
-        ));
-        assert!(!period_already_granted(
-            std::iter::empty::<(Option<&str>, Option<&str>)>(),
-            "100",
-        ));
-    }
+    use super::{METER_CUSTOMER_KEY, METER_VALUE_KEY, plan_level_from_pro_price};
 
     #[test]
-    fn plan_level_from_base_fee_resolves_tier() {
-        // Pro is the only paid tier with a base fee, so it resolves to Pro; any
-        // other paid metered subscription resolves to Team.
-        assert_eq!(plan_level_from_base_fee(true), PlanLevel::Pro);
-        assert_eq!(plan_level_from_base_fee(false), PlanLevel::Team);
+    fn plan_level_from_pro_price_resolves_tier() {
+        // Pro is the only paid tier carrying the tiered active-series price, so it
+        // resolves to Pro; any other paid metered subscription resolves to Team.
+        assert_eq!(plan_level_from_pro_price(true), PlanLevel::Pro);
+        assert_eq!(plan_level_from_pro_price(false), PlanLevel::Team);
     }
 
     #[test]
     fn get_plan_unit_amount_reads_price() {
         let mut item = make_subscription_item("price_metrics");
         item.price.unit_amount = Some(1);
-        assert_eq!(Biller::get_plan_unit_amount(&item).unwrap(), 1);
+        assert_eq!(Biller::get_plan_unit_amount(&item, None).unwrap(), 1);
     }
 
     #[test]
     fn get_plan_unit_amount_missing_errors() {
-        // `make_subscription_item` builds a price with `unit_amount: None`.
+        // `make_subscription_item` builds a price with `unit_amount: None` and no tier
+        // fallback, so the amount is unresolved.
         let item = make_subscription_item("price_metrics");
-        let err = Biller::get_plan_unit_amount(&item).unwrap_err();
+        let err = Biller::get_plan_unit_amount(&item, None).unwrap_err();
         assert!(matches!(err, crate::BillingError::NoUnitAmount(_)));
     }
 
+    #[test]
+    fn get_plan_unit_amount_tiered_uses_fallback() {
+        // A tiered price exposes no flat `unit_amount`; the base-tier fee is used.
+        let item = make_subscription_item("price_pro_tiered");
+        assert_eq!(
+            Biller::get_plan_unit_amount(&item, Some(2_000)).unwrap(),
+            2_000,
+        );
+    }
+
     const TEST_BILLING_KEY: &str = "TEST_BILLING_KEY";
 
     fn billing_key() -> Option<String> {
@@ -1227,12 +1097,10 @@ mod tests {
         JsonProducts {
             pro: JsonProduct {
                 id: "prod_UizgEJP4gIENBi".into(),
-                metered: HashMap::new(),
-                licensed: HashMap::new(),
-                base: hmap! {
-                    "default".to_owned() => "price_1TjXgeKal5vzTlmhZzr88bki".to_owned(),
+                metered: hmap! {
+                    "default".to_owned() => "price_1Tmo29Kal5vzTlmh9Qk5vtLi".to_owned(),
                 },
-                trial_coupon: Some("m8uHY6oa".to_owned()),
+                licensed: HashMap::new(),
             },
             team: JsonProduct {
                 id: "prod_NKz5B9dGhDiSY1".into(),
@@ -1242,8 +1110,6 @@ mod tests {
                 licensed: hmap! {
                     "default".to_owned() => "price_1O4XlwKal5vzTlmh0n0wtplQ".to_owned(),
                 },
-                base: HashMap::new(),
-                trial_coupon: None,
             },
             enterprise: JsonProduct {
                 id: "prod_NLC7fDet2C8Nmk".into(),
@@ -1253,8 +1119,6 @@ mod tests {
                 licensed: hmap! {
                     "default".to_owned() => "price_1O4Xo1Kal5vzTlmh1KrcEbq0".to_owned(),
                 },
-                base: HashMap::new(),
-                trial_coupon: None,
             },
             metrics: JsonProduct {
                 id: "prod_UjlAPYSw7n3RDq".into(),
@@ -1264,8 +1128,6 @@ mod tests {
                 licensed: hmap! {
                     "default".to_owned() => "price_1TkHeSKal5vzTlmhPKXPOW7c".to_owned(),
                 },
-                base: HashMap::new(),
-                trial_coupon: None,
             },
             bare_metal: JsonProduct {
                 id: "prod_U6a28ecwqRZHYz".into(),
@@ -1275,8 +1137,6 @@ mod tests {
                 licensed: hmap! {
                     "default".to_owned() => "price_1TCWdkKal5vzTlmh9fdahWqY".to_owned(),
                 },
-                base: HashMap::new(),
-                trial_coupon: None,
             },
         }
     }
@@ -1327,7 +1187,6 @@ mod tests {
 
     #[expect(
         clippy::too_many_arguments,
-        clippy::cognitive_complexity,
         reason = "test helper exercising the full metered subscription lifecycle"
     )]
     async fn metered_subscription(
@@ -1344,9 +1203,9 @@ mod tests {
             .create_metered_subscription(
                 organization,
                 customer_id,
-                payment_method_id,
+                payment_method_id.clone(),
                 plan_level,
-                price_name,
+                price_name.clone(),
             )
             .await
             .unwrap();
@@ -1357,8 +1216,8 @@ mod tests {
 
         let metered_plan_id = &subscription_id.as_ref().parse().unwrap();
         let json_plan = biller.get_metered_plan(metered_plan_id).await.unwrap();
-        // The plan level is derived from base-fee presence: Pro carries a flat base
-        // fee and resolves to Pro; tiers without a base fee resolve to Team.
+        // The plan level is derived from the tiered active-series price: Pro carries it
+        // and resolves to Pro; tiers without it resolve to Team.
         let expected_level = if plan_level == PlanLevel::Pro {
             PlanLevel::Pro
         } else {
@@ -1366,29 +1225,25 @@ mod tests {
         };
         assert_eq!(json_plan.level, expected_level);
 
-        let (plan_status, customer_id, status_level) = biller
-            .get_metered_plan_status(metered_plan_id)
+        let billing = biller
+            .get_metered_plan_billing(metered_plan_id)
             .await
             .unwrap();
-        assert_eq!(plan_status, PlanStatus::Active);
-        // The status path derives the same level as the full plan fetch.
-        assert_eq!(status_level, expected_level);
-
-        // Only Pro carries a flat base fee, so only Pro grants an included-usage
-        // credit; the call is a no-op for Team/Enterprise. Granting twice must be
-        // idempotent (regression: a past `effective_at` was rejected by Stripe).
-        let first_credit = biller.ensure_period_credit(metered_plan_id).await.unwrap();
-        let second_credit = biller.ensure_period_credit(metered_plan_id).await.unwrap();
-        if plan_level == PlanLevel::Pro {
-            assert_eq!(first_credit, PeriodCredit::Granted);
-            assert_eq!(second_credit, PeriodCredit::AlreadyGranted);
+        // Pro starts in a native free trial, so it is `Trialing`; the other tiers have no
+        // trial and are `Active` immediately.
+        let expected_status = if plan_level == PlanLevel::Pro {
+            PlanStatus::Trialing
         } else {
-            assert_eq!(first_credit, PeriodCredit::NotApplicable);
-            assert_eq!(second_credit, PeriodCredit::NotApplicable);
-        }
+            PlanStatus::Active
+        };
+        assert_eq!(billing.status, expected_status);
+        // The billing path derives the same level as the full plan fetch.
+        assert_eq!(billing.level, expected_level);
+        let customer_id = billing.customer_id;
 
-        record_runner_usage(biller, &customer_id, runner_minutes).await;
         record_metrics_usage(biller, &customer_id, metrics_quantity).await;
+        record_series_usage(biller, &customer_id, metrics_quantity).await;
+        record_runner_usage(biller, &customer_id, runner_minutes).await;
 
         // PATCH path: schedule cancel-at-period-end, then clear it (resume). The
         // subscription stays active throughout.
@@ -1408,11 +1263,54 @@ mod tests {
             .cancel_metered_subscription(&subscription_id.parse().unwrap())
             .await
             .unwrap();
-        let (plan_status, _, _) = biller
-            .get_metered_plan_status(metered_plan_id)
+        let billing = biller
+            .get_metered_plan_billing(metered_plan_id)
+            .await
+            .unwrap();
+        assert_eq!(billing.status, PlanStatus::Canceled);
+
+        if plan_level == PlanLevel::Pro {
+            resubscribe_has_no_trial(
+                biller,
+                organization,
+                customer_id,
+                payment_method_id,
+                price_name,
+            )
+            .await;
+        }
+    }
+
+    /// Re-subscribing after cancellation must not restart the free trial: the
+    /// customer's prior (canceled) subscription disqualifies it, so the new Pro
+    /// subscription is `Active` immediately, not `Trialing` (trial cycling guard).
+    async fn resubscribe_has_no_trial(
+        biller: &Biller,
+        organization: OrganizationUuid,
+        customer_id: CustomerId,
+        payment_method_id: PaymentMethodId,
+        price_name: String,
+    ) {
+        let resubscription = biller
+            .create_metered_subscription(
+                organization,
+                customer_id,
+                payment_method_id,
+                PlanLevel::Pro,
+                price_name,
+            )
+            .await
+            .unwrap();
+        let metered_plan_id = &resubscription.id.as_ref().parse().unwrap();
+        let billing = biller
+            .get_metered_plan_billing(metered_plan_id)
+            .await
+            .unwrap();
+        assert_eq!(billing.status, PlanStatus::Active);
+        biller
+            .cancel_metered_subscription(metered_plan_id)
             .await
             .unwrap();
-        assert_eq!(plan_status, PlanStatus::Canceled);
     }
 
     async fn licensed_subscription(
@@ -1489,6 +1387,24 @@ mod tests {
         );
     }
 
+    async fn record_series_usage(biller: &Biller, customer_id: &CustomerId, quantity: u32) {
+        // Live Stripe test: a meter event needs a real, recent timestamp (a fixed
+        // `DateTime::TEST` would be rejected as outside Stripe's accepted window).
+        let event = biller
+            .record_series_usage(customer_id, quantity, bencher_json::DateTime::now())
+            .await
+            .unwrap();
+        assert_eq!(event.event_name, ACTIVE_SERIES_METER_NAME);
+        assert_eq!(
+            event.payload.get(METER_CUSTOMER_KEY),
+            Some(&customer_id.to_string()),
+        );
+        assert_eq!(
+            event.payload.get(METER_VALUE_KEY),
+            Some(&quantity.to_string()),
+        );
+    }
+
     fn make_price(price_id: &str) -> stripe_shared::Price {
         stripe_shared::Price {
             active: false,
diff --git a/plus/bencher_billing/src/lib.rs b/plus/bencher_billing/src/lib.rs
index 0cfe422f1..5251e9380 100644
--- a/plus/bencher_billing/src/lib.rs
+++ b/plus/bencher_billing/src/lib.rs
@@ -10,5 +10,5 @@ mod biller;
 mod error;
 mod products;
 
-pub use biller::{Biller, PeriodCredit};
+pub use biller::{Biller, MeteredPlanBilling};
 pub use error::BillingError;
diff --git a/plus/bencher_billing/src/products.rs b/plus/bencher_billing/src/products.rs
index e994f3999..41b637162 100644
--- a/plus/bencher_billing/src/products.rs
+++ b/plus/bencher_billing/src/products.rs
@@ -11,12 +11,15 @@ use crate::BillingError;
 
 #[derive(Clone)]
 pub struct Products {
+    // Bencher Cloud self-serve tier. Its `metered` map holds the single tiered
+    // active-series price (tier 1 flat base fee plus per-series step-ups) that bills
+    // both the base monthly fee and the per-series overage.
     pub pro: Product,
     // Legacy self-serve paid tier, retained for grandfathered customers.
     pub team: Product,
     pub enterprise: Product,
-    // Holds the Pro plan's metered metrics price (its own product so checkout
-    // shows a distinct "Bencher Metrics" line item).
+    // Shared metered metrics product for legacy Team/Enterprise plans. Pro bills on
+    // its own tiered active-series price, not on this meter.
     pub metrics: Product,
     pub bare_metal: Product,
 }
@@ -40,29 +43,37 @@ impl Products {
         })
     }
 
-    // The price IDs that identify a subscription's plan item (excluding flat base
-    // fees and bare_metal), used by `get_plan` to filter subscription items down to
-    // that one item. These are the metered + licensed prices across the `metrics`
+    // The price IDs that identify a subscription's plan item (excluding bare_metal),
+    // used by `get_plan` to filter subscription items down to that one item. These are
+    // the metered + licensed prices across the `pro` (tiered active-series), `metrics`
     // (shared metered metrics), `team`, and `enterprise` products. The Team and
     // Enterprise metered prices are retained so a subscription still on its own
     // product (not yet moved to `metrics`) continues to resolve.
     pub fn plan_price_ids(&self) -> HashSet<&PriceId> {
-        [&self.metrics, &self.team, &self.enterprise]
+        [&self.pro, &self.metrics, &self.team, &self.enterprise]
             .into_iter()
             .flat_map(|product| product.metered.values().chain(product.licensed.values()))
             .map(|price| &price.id)
             .collect()
     }
 
-    /// All configured flat base prices across every product. The included usage
-    /// credit equals the base fee of whichever base price a subscription carries.
-    pub fn all_base_prices(&self) -> impl Iterator<Item = &StripePrice> {
-        self.pro
-            .base
-            .values()
-            .chain(self.team.base.values())
-            .chain(self.enterprise.base.values())
-            .chain(self.bare_metal.base.values())
+    /// The base monthly fee (in cents) for a tiered price, read from its first tier's
+    /// `flat_amount`. The Pro tiered active-series price exposes no flat `unit_amount`,
+    /// so `get_plan` reports this as the plan's `unit_amount`. `None` if the price is
+    /// unknown or not tiered. Relies on prices being retrieved with `tiers` expanded
+    /// (see [`Product::pricing`]).
+    pub fn tier_base_fee_cents(&self, price_id: &PriceId) -> Option<i64> {
+        [
+            &self.pro,
+            &self.metrics,
+            &self.team,
+            &self.enterprise,
+            &self.bare_metal,
+        ]
+        .into_iter()
+        .flat_map(|product| product.metered.values().chain(product.licensed.values()))
+        .find(|price| &price.id == price_id)
+        .and_then(|price| price.tiers.as_ref()?.first()?.flat_amount)
     }
 }
 
@@ -74,13 +85,6 @@ pub struct Product {
         reason = "retained for future Stripe API use"
     )]
     pub product: StripeProduct,
-    // Stripe coupon ID for this product's free trial (waives the first month's
-    // base fee). `None` if the product has no trial.
-    pub trial_coupon: Option<String>,
-    // Flat recurring base prices. Excluded from `plan_price_ids` (never the metered
-    // "plan item"), but used by `get_plan` to detect the Pro base fee and so resolve
-    // the plan level.
-    pub base: HashMap<String, StripePrice>,
     pub metered: HashMap<String, StripePrice>,
     pub licensed: HashMap<String, StripePrice>,
 }
@@ -91,20 +95,15 @@ impl Product {
             id,
             metered,
             licensed,
-            base,
-            trial_coupon,
         } = product;
 
         let product_id: ProductId = id.as_str().into();
         let product = RetrieveProduct::new(product_id).send(client).await?;
         let metered = Self::pricing(client, metered).await?;
         let licensed = Self::pricing(client, licensed).await?;
-        let base = Self::pricing(client, base).await?;
 
         Ok(Self {
             product,
-            trial_coupon,
-            base,
             metered,
             licensed,
         })
@@ -117,7 +116,12 @@ impl Product {
         let mut biller_pricing = HashMap::with_capacity(pricing.len());
         for (price_name, price_id) in pricing {
             let price_id: PriceId = price_id.as_str().into();
-            let price = RetrievePrice::new(price_id).send(client).await?;
+            // Expand `tiers` so a tiered price (the Pro active-series price) exposes its
+            // tier `flat_amount` (the base monthly fee). Harmless for non-tiered prices.
+            let price = RetrievePrice::new(price_id)
+                .expand(vec!["tiers".to_owned()])
+                .send(client)
+                .await?;
             biller_pricing.insert(price_name, price);
         }
         Ok(biller_pricing)
diff --git a/plus/bencher_otel/src/api_meter.rs b/plus/bencher_otel/src/api_meter.rs
index c91396e44..803eefc0c 100644
--- a/plus/bencher_otel/src/api_meter.rs
+++ b/plus/bencher_otel/src/api_meter.rs
@@ -88,6 +88,8 @@ pub enum ApiCounter {
     MetricsCreate(Priority),
     MetricsBilled,
     MetricsBilledFailed,
+    ActiveSeriesBilled,
+    ActiveSeriesBilledFailed,
 
     UserIp,
     UserIpNotFound,
@@ -176,6 +178,8 @@ impl ApiCounter {
 
             Self::MetricsCreate(_) | Self::MetricsBilled | Self::MetricsBilledFailed => "{metric}",
 
+            Self::ActiveSeriesBilled | Self::ActiveSeriesBilledFailed => "{series}",
+
             Self::UserIp
             | Self::UserIpNotFound
             | Self::RequestMax(_, _)
@@ -239,6 +243,8 @@ impl ApiCounter {
             Self::MetricsCreate(_) => "metrics.create",
             Self::MetricsBilled => "metrics.billed",
             Self::MetricsBilledFailed => "metrics.billed.failed",
+            Self::ActiveSeriesBilled => "active_series.billed",
+            Self::ActiveSeriesBilledFailed => "active_series.billed.failed",
 
             Self::UserIp => "user.ip",
             Self::UserIpNotFound => "user.ip.not_found",
@@ -333,6 +339,8 @@ impl ApiCounter {
             Self::MetricsCreate(_) => "Counts the number of metrics created",
             Self::MetricsBilled => "Counts the number of metrics successfully billed",
             Self::MetricsBilledFailed => "Counts the number of metrics billing failures",
+            Self::ActiveSeriesBilled => "Counts the number of active series successfully billed",
+            Self::ActiveSeriesBilledFailed => "Counts the number of active series billing failures",
 
             Self::UserIp => "Counts the number of user IP address found occurrences",
             Self::UserIpNotFound => "Counts the number of user IP address not found occurrences",
@@ -458,6 +466,8 @@ impl ApiCounter {
             | Self::UserKeyRevokeBlocked
             | Self::MetricsBilled
             | Self::MetricsBilledFailed
+            | Self::ActiveSeriesBilled
+            | Self::ActiveSeriesBilledFailed
             | Self::EmailSend
             | Self::OciBlobPush
             | Self::OciBlobPull
diff --git a/services/api/openapi.json b/services/api/openapi.json
index eeca82ae3..97a040f27 100644
--- a/services/api/openapi.json
+++ b/services/api/openapi.json
@@ -15187,13 +15187,6 @@
       "JsonProduct": {
         "type": "object",
         "properties": {
-          "base": {
-            "default": {},
-            "type": "object",
-            "additionalProperties": {
-              "type": "string"
-            }
-          },
           "id": {
             "type": "string"
           },
@@ -15210,11 +15203,6 @@
             "additionalProperties": {
               "type": "string"
             }
-          },
-          "trial_coupon": {
-            "nullable": true,
-            "default": null,
-            "type": "string"
           }
         },
         "required": [