@@ -58,3 +58,35 @@ number of large tenants have millions.  An unindexed UPDATE/DELETE on
 take tens of seconds, blocking the API request under `@transaction.atomic`.
 
 Always add an index on FK columns in partitioned tables.
+
+## Stale `partitioned_tables` tracking (COST-8415)
+
+Each tenant schema has a `partitioned_tables` registry. Creating/dropping partitions
+goes through that table and `trfn_partition_manager`. When a migration removes
+partitioned models (or drops parents) without cleaning the registry, tracking rows
+can remain. The monthly pre-create task **skips** those stale defaults (it does not
+delete them).
+
+This is a **rare** event — typically when a provider or partitioned model family is
+removed from the product — not something that accumulates day-to-day.
+
+### If you remove partitioned models
+
+1. Drop/remove the physical tables (as the migration already does).
+2. Clear leftover registry rows **only** when the physical `schema.table_name` no
+   longer exists. A `DELETE` on `partitioned_tables` while a same-named relation
+   still exists will `DETACH` / `TRUNCATE` / `DROP` it via the trigger.
+3. Prefer the management command (dry-run first, then `--delete` after reviewing
+   samples):
+
+```bash
+# Inventory only (default)
+pipenv run python koku/manage.py reconcile_stale_partition_tracking --schema orgEXAMPLE
+# After reviewing log samples:
+pipenv run python koku/manage.py reconcile_stale_partition_tracking --schema orgEXAMPLE --delete
+```
+
+Or call `koku.pg_partition.reconcile_stale_partition_tracking(schema_name, delete=...)`
+from a one-off ops path / migration RunPython with the same rule: never delete a
+row whose relation still exists in `pg_class` (name reuse / detached tables stay
+as report-only).
@@ -13,6 +13,7 @@
 import time
 import uuid
 from contextlib import contextmanager
+from typing import Any
 
 import ciso8601
 from dateutil.relativedelta import relativedelta
@@ -2141,3 +2142,137 @@ def _handle_partitions(self, schema_name, table_names, start_date, end_date):  #
                     LOG.debug(f"created = {created}")
                     if created:
                         LOG.info(f"Created partition {newpart.schema_name}.{newpart.table_name}")
+
+
+# Classification labels for partition tracking reconciliation (COST-8415).
+TRACKING_SAFE_ORPHAN = "safe_orphan"
+TRACKING_PHYSICAL_PRESENT = "physical_present"
+
+
+def relation_exists(schema_name: str, table_name: str) -> bool:
+    """Return True if schema_name.table_name exists in pg_class."""
+    sql = """
+SELECT EXISTS (
+    SELECT 1
+      FROM pg_class c
+      JOIN pg_namespace n
+        ON n.oid = c.relnamespace
+     WHERE n.nspname = %s::name
+       AND c.relname = %s::name
+)
+"""
+    cur = conn_execute(sql, (schema_name, table_name))
+    return bool(cur.fetchone()[0])
+
+
+def _is_default_partition_parameters(partition_parameters: Any) -> bool:
+    if not isinstance(partition_parameters, dict):
+        return False
+    return bool(partition_parameters.get("default"))
+
+
+def classify_partition_tracking(schema_name: str) -> list[dict[str, Any]]:
+    """
+    Classify each partitioned_tables row in schema_name against PostgreSQL.
+
+    safe_orphan: tracking row whose physical table_name does not exist — DELETE is a
+    registry-only change (trfn_partition_manager takes no DDL action).
+    physical_present: relation exists (live, detached, or name reuse) — never auto-delete.
+    """
+    with schema_context(schema_name):
+        records = list(PartitionedTable.objects.filter(schema_name=schema_name).order_by("table_name"))
+
+    classified = []
+    for record in records:
+        exists = relation_exists(schema_name, record.table_name)
+        classified.append(
+            {
+                "id": record.id,
+                "schema_name": schema_name,
+                "table_name": record.table_name,
+                "partition_of_table_name": record.partition_of_table_name,
+                "active": record.active,
+                "is_default": _is_default_partition_parameters(record.partition_parameters),
+                "classification": TRACKING_PHYSICAL_PRESENT if exists else TRACKING_SAFE_ORPHAN,
+            }
+        )
+    return classified
+
+
+def _delete_safe_orphan_tracking(schema_name: str, candidate_ids: list[int]) -> int:
+    """
+    Delete candidate tracking rows only when the physical relation is still absent.
+
+    Existence is enforced in the same DELETE statement (NOT EXISTS on pg_class) so a
+    concurrent CREATE cannot be selected for deletion after a separate Python check.
+    """
+    if not candidate_ids:
+        return 0
+    sql = """
+DELETE FROM partitioned_tables AS pt
+ WHERE pt.id = ANY(%s)
+   AND pt.schema_name = %s
+   AND NOT EXISTS (
+         SELECT 1
+           FROM pg_class c
+           JOIN pg_namespace n
+             ON n.oid = c.relnamespace
+          WHERE n.nspname = pt.schema_name::name
+            AND c.relname = pt.table_name::name
+       )
+"""
+    with schema_context(schema_name):
+        cur = conn_execute(sql, (candidate_ids, schema_name))
+        return cur.rowcount
+
+
+def reconcile_stale_partition_tracking(schema_name: str, *, delete: bool = False, sample: int = 5) -> dict[str, Any]:
+    """
+    Inventory (and optionally delete) safe_orphan partition tracking rows for one schema.
+
+    Only deletes rows whose physical table is still absent at delete time. Never deletes
+    physical_present rows. Default is dry-run (delete=False).
+
+    Returns:
+        dict with keys: schema_name, delete, counts (by classification), deleted,
+        samples (up to sample rows per classification).
+    """
+    classified = classify_partition_tracking(schema_name)
+    counts = {TRACKING_SAFE_ORPHAN: 0, TRACKING_PHYSICAL_PRESENT: 0}
+    samples: dict[str, list[dict[str, Any]]] = {TRACKING_SAFE_ORPHAN: [], TRACKING_PHYSICAL_PRESENT: []}
+    for row in classified:
+        label = row["classification"]
+        counts[label] = counts.get(label, 0) + 1
+        if len(samples.get(label, [])) < sample:
+            samples.setdefault(label, []).append(
+                {
+                    "table_name": row["table_name"],
+                    "partition_of_table_name": row["partition_of_table_name"],
+                    "is_default": row["is_default"],
+                    "active": row["active"],
+                }
+            )
+
+    deleted = 0
+    if delete:
+        safe_ids = [row["id"] for row in classified if row["classification"] == TRACKING_SAFE_ORPHAN]
+        deleted = _delete_safe_orphan_tracking(schema_name, safe_ids)
+
+    LOG.info(
+        log_json(
+            msg="partition tracking reconcile",
+            schema=schema_name,
+            delete=delete,
+            deleted=deleted,
+            safe_orphan=counts.get(TRACKING_SAFE_ORPHAN, 0),
+            physical_present=counts.get(TRACKING_PHYSICAL_PRESENT, 0),
+            samples=samples,
+        )
+    )
+    return {
+        "schema_name": schema_name,
+        "delete": delete,
+        "counts": counts,
+        "deleted": deleted,
+        "samples": samples,
+    }
@@ -1489,3 +1489,115 @@ def hold_read_lock():
         self.assertEqual(counts["created"], 1)
         self.assertEqual(counts["missing"], [])
         self.assertTrue(any("'outcome': 'created'" in line for line in logs.output))
+
+
+class TestReconcileStalePartitionTracking(IamTestCase):
+    """Safe cleanup of partitioned_tables rows whose physical tables are gone (COST-8415)."""
+
+    SCHEMA_NAME = "org1234567"
+    LIVE_TABLE = "__pg_partition_reconcile_live"
+    STALE_TABLE = "__pg_partition_reconcile_stale"
+
+    def setUp(self):
+        super().setUp()
+        self._create_partitioned_table(self.LIVE_TABLE)
+        self._create_partitioned_table(self.STALE_TABLE)
+        _execute(f"drop table {self.SCHEMA_NAME}.{self.STALE_TABLE} cascade")
+
+    def tearDown(self):
+        with schema_context(self.SCHEMA_NAME):
+            PartitionedTable.objects.filter(
+                schema_name=self.SCHEMA_NAME,
+                partition_of_table_name__in=[self.LIVE_TABLE, self.STALE_TABLE],
+            ).delete()
+        _execute(f"drop table if exists {self.SCHEMA_NAME}.{self.LIVE_TABLE} cascade")
+        _execute(f"drop table if exists {self.SCHEMA_NAME}.{self.STALE_TABLE} cascade")
+        # Name-reuse helper tables from individual tests
+        _execute(f"drop table if exists {self.SCHEMA_NAME}.{self.STALE_TABLE}_default cascade")
+        super().tearDown()
+
+    def _create_partitioned_table(self, table_name):
+        _execute(
+            f"""
+create table if not exists {self.SCHEMA_NAME}.{table_name} (
+    id bigserial,
+    usage_start date not null,
+    primary key (usage_start, id)
+)
+partition by range (usage_start);
+"""
+        )
+        with schema_context(self.SCHEMA_NAME):
+            ppart.get_or_create_partition(
+                {
+                    "schema_name": self.SCHEMA_NAME,
+                    "table_name": f"{table_name}_default",
+                    "partition_of_table_name": table_name,
+                    "partition_type": PartitionedTable.RANGE,
+                    "partition_col": "usage_start",
+                    "partition_parameters": {"default": True},
+                    "active": True,
+                }
+            )
+
+    def _tracking(self, table_name):
+        with schema_context(self.SCHEMA_NAME):
+            return PartitionedTable.objects.filter(schema_name=self.SCHEMA_NAME, table_name=table_name).first()
+
+    def test_dry_run_classifies_safe_orphan_and_does_not_delete(self):
+        result = ppart.reconcile_stale_partition_tracking(self.SCHEMA_NAME, delete=False)
+        self.assertFalse(result["delete"])
+        self.assertEqual(result["deleted"], 0)
+        self.assertGreaterEqual(result["counts"][ppart.TRACKING_SAFE_ORPHAN], 1)
+        self.assertIsNotNone(self._tracking(f"{self.STALE_TABLE}_default"))
+        self.assertIsNotNone(self._tracking(f"{self.LIVE_TABLE}_default"))
+
+    def test_delete_removes_only_safe_orphan_rows(self):
+        result = ppart.reconcile_stale_partition_tracking(self.SCHEMA_NAME, delete=True)
+        self.assertGreaterEqual(result["deleted"], 1)
+        self.assertIsNone(self._tracking(f"{self.STALE_TABLE}_default"))
+        self.assertIsNotNone(self._tracking(f"{self.LIVE_TABLE}_default"))
+        self.assertTrue(ppart.relation_exists(self.SCHEMA_NAME, f"{self.LIVE_TABLE}_default"))
+
+    def test_name_reuse_is_physical_present_and_not_deleted(self):
+        """A same-named default attached to another parent must not be deleted."""
+        replacement = "__pg_partition_reconcile_replacement"
+        _execute(
+            f"""
+create table {self.SCHEMA_NAME}.{replacement} (
+    id bigserial,
+    usage_start date not null,
+    primary key (usage_start, id)
+)
+partition by range (usage_start);
+"""
+        )
+        _execute(
+            f"create table {self.SCHEMA_NAME}.{self.STALE_TABLE}_default "
+            f"partition of {self.SCHEMA_NAME}.{replacement} default"
+        )
+        try:
+            classified = ppart.classify_partition_tracking(self.SCHEMA_NAME)
+            stale_default = next(row for row in classified if row["table_name"] == f"{self.STALE_TABLE}_default")
+            self.assertEqual(stale_default["classification"], ppart.TRACKING_PHYSICAL_PRESENT)
+            self.assertEqual(stale_default["partition_of_table_name"], self.STALE_TABLE)
+
+            result = ppart.reconcile_stale_partition_tracking(self.SCHEMA_NAME, delete=True)
+            self.assertIsNotNone(self._tracking(f"{self.STALE_TABLE}_default"))
+            self.assertTrue(ppart.relation_exists(self.SCHEMA_NAME, f"{self.STALE_TABLE}_default"))
+            # Live default still present; name-reused stale default was not among deleted safe orphans
+            self.assertNotIn(
+                f"{self.STALE_TABLE}_default",
+                [s["table_name"] for s in result["samples"].get(ppart.TRACKING_SAFE_ORPHAN, [])],
+            )
+        finally:
+            _execute(f"drop table if exists {self.SCHEMA_NAME}.{replacement} cascade")
+
+    def test_delete_sql_skips_row_when_relation_exists(self):
+        """DELETE ... NOT EXISTS must not remove tracking for a present relation."""
+        live_default = self._tracking(f"{self.LIVE_TABLE}_default")
+        self.assertIsNotNone(live_default)
+        deleted = ppart._delete_safe_orphan_tracking(self.SCHEMA_NAME, [live_default.id])
+        self.assertEqual(deleted, 0)
+        self.assertIsNotNone(self._tracking(f"{self.LIVE_TABLE}_default"))
+        self.assertTrue(ppart.relation_exists(self.SCHEMA_NAME, f"{self.LIVE_TABLE}_default"))
@@ -0,0 +1,89 @@
+#
+# Copyright 2026 Red Hat Inc.
+# SPDX-License-Identifier: Apache-2.0
+#
+"""Inventory and safely delete orphaned partitioned_tables tracking rows (COST-8415).
+
+Only deletes rows whose physical table does not exist in PostgreSQL. DELETE on
+partitioned_tables would otherwise DETACH/TRUNCATE/DROP a same-named relation via
+trfn_partition_manager — those rows are reported only.
+
+Default is dry-run. Pass --delete after reviewing counts and samples.
+"""
+from __future__ import annotations
+
+import argparse
+import logging
+from typing import Any
+
+from django.core.management.base import BaseCommand
+from django.core.management.base import CommandError
+
+from api.iam.models import Tenant
+from koku.pg_partition import reconcile_stale_partition_tracking
+from koku.pg_partition import TRACKING_PHYSICAL_PRESENT
+from koku.pg_partition import TRACKING_SAFE_ORPHAN
+
+LOG = logging.getLogger(__name__)
+
+
+def _tenant_schema_names(*, schema: str | None = None) -> list[str]:
+    excluded = {"public", Tenant._TEMPLATE_SCHEMA}
+    if schema is not None:
+        if not schema:
+            raise CommandError("--schema must not be empty.")
+        if schema in excluded:
+            raise CommandError(f"Schema {schema} is not a customer tenant schema.")
+        if not Tenant.objects.filter(schema_name=schema).exists():
+            raise CommandError(f"Tenant schema {schema} not found.")
+        return [schema]
+    return [name for name in Tenant.objects.values_list("schema_name", flat=True) if name and name not in excluded]
+
+
+class Command(BaseCommand):
+    help = (
+        "Inventory (and optionally delete) partitioned_tables rows whose physical "
+        "tables no longer exist. Dry-run by default; pass --delete to mutate."
+    )
+
+    def add_arguments(self, parser: argparse.ArgumentParser) -> None:
+        parser.add_argument(
+            "--delete",
+            action="store_true",
+            default=False,
+            help="Delete verified safe_orphan tracking rows. Without this flag, only report.",
+        )
+        parser.add_argument(
+            "--schema",
+            type=str,
+            default=None,
+            help="Limit to one tenant schema (default: all customer schemas).",
+        )
+        parser.add_argument(
+            "--sample",
+            type=int,
+            default=5,
+            help="Max sample rows to log per classification per schema (default: 5).",
+        )
+
+    def handle(self, *args: Any, delete: bool, schema: str | None, sample: int, **kwargs: Any) -> None:
+        if sample < 0:
+            raise CommandError("--sample must be >= 0")
+
+        schemas = _tenant_schema_names(schema=schema)
+        mode = "DELETE" if delete else "DRY RUN"
+        self.stdout.write(f"partition tracking reconcile ({mode}) schemas={len(schemas)}")
+
+        totals = {TRACKING_SAFE_ORPHAN: 0, TRACKING_PHYSICAL_PRESENT: 0, "deleted": 0}
+        for schema_name in schemas:
+            result = reconcile_stale_partition_tracking(schema_name, delete=delete, sample=sample)
+            totals[TRACKING_SAFE_ORPHAN] += result["counts"].get(TRACKING_SAFE_ORPHAN, 0)
+            totals[TRACKING_PHYSICAL_PRESENT] += result["counts"].get(TRACKING_PHYSICAL_PRESENT, 0)
+            totals["deleted"] += result["deleted"]
+
+        self.stdout.write(
+            f"Done ({mode}): safe_orphan={totals[TRACKING_SAFE_ORPHAN]} "
+            f"physical_present={totals[TRACKING_PHYSICAL_PRESENT]} deleted={totals['deleted']}"
+        )
+        if not delete and totals[TRACKING_SAFE_ORPHAN]:
+            self.stdout.write("Review samples in logs, then re-run with --delete to remove safe_orphan rows only.")