Soju06
diff --git a/‎Makefile‎
Lines changed: 2 additions & 0 deletions b/‎Makefile‎
Lines changed: 2 additions & 0 deletions
diff --git a/‎app/db/alembic/versions/20260816_000000_add_account_pending_deletion.py‎
Lines changed: 81 additions & 0 deletions b/‎app/db/alembic/versions/20260816_000000_add_account_pending_deletion.py‎
Lines changed: 81 additions & 0 deletions
diff --git a/‎app/db/models.py‎
Lines changed: 25 additions & 0 deletions b/‎app/db/models.py‎
Lines changed: 25 additions & 0 deletions
diff --git a/‎app/main.py‎
Lines changed: 4 additions & 0 deletions b/‎app/main.py‎
Lines changed: 4 additions & 0 deletions
@@ -20,6 +20,7 @@ POSTGRES_PYTEST_TARGETS := \
 	tests/integration/test_db_commit_durability.py \
 	tests/test_request_logs_options_api.py \
 	tests/integration/test_account_usage_rollup.py \
+	tests/integration/test_account_deletion_background.py \
 	tests/integration/test_request_usage_time_rollup.py \
 	tests/integration/test_request_usage_rollup_parity.py \
 	tests/integration/test_migrations.py::test_request_usage_time_rollups_migration_upgrade_and_downgrade \
@@ -33,6 +34,7 @@ POSTGRES_PYTEST_TARGETS := \
 	tests/integration/test_repositories.py::test_replace_reauthorized_discards_pending_downgrade_evidence \
 	tests/integration/test_repositories.py::test_upsert_account_slot_discards_pending_downgrade_evidence_on_reimport \
 	tests/integration/test_migrations.py::test_account_plan_downgrade_observations_migration_upgrade_and_downgrade \
+	tests/integration/test_migrations.py::test_account_pending_deletion_migration_upgrade_and_downgrade \
 	tests/integration/test_usage_repository.py::test_bulk_history_since_primary_query_plan_is_index_only_postgresql \
 	tests/integration/test_usage_repository.py::test_bulk_history_since_cutoff_query_plan_is_index_only_postgresql \
 	tests/integration/test_usage_repository.py::test_bulk_history_since_secondary_query_plan_is_index_only_postgresql \
 
@@ -0,0 +1,81 @@
+"""add account pending-deletion marker columns
+
+Revision ID: 20260816_000000_add_account_pending_deletion
+Revises: 20260812_120000_add_sticky_abandonment_scope
+Create Date: 2026-08-16
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = "20260816_000000_add_account_pending_deletion"
+down_revision = "20260812_120000_add_sticky_abandonment_scope"
+branch_labels = None
+depends_on = None
+
+_TABLE = "accounts"
+_INDEX = "idx_accounts_delete_requested_at"
+
+
+def _columns(bind) -> set[str]:
+    return {column["name"] for column in sa.inspect(bind).get_columns(_TABLE)}
+
+
+def _indexes(bind) -> set[str]:
+    return {index["name"] for index in sa.inspect(bind).get_indexes(_TABLE)}
+
+
+def upgrade() -> None:
+    bind = op.get_bind()
+    columns = _columns(bind)
+    if "delete_requested_at" not in columns:
+        op.add_column(_TABLE, sa.Column("delete_requested_at", sa.DateTime(), nullable=True))
+    if "delete_history_requested" not in columns:
+        op.add_column(
+            _TABLE,
+            sa.Column(
+                "delete_history_requested",
+                sa.Boolean(),
+                nullable=False,
+                server_default=sa.false(),
+            ),
+        )
+    if _INDEX not in _indexes(bind):
+        # Pending-deletion queue probe/order support; partial so it is empty
+        # (and free) in the steady state with no pending deletions.
+        op.create_index(
+            _INDEX,
+            _TABLE,
+            ["delete_requested_at", "id"],
+            postgresql_where=sa.text("delete_requested_at IS NOT NULL"),
+            sqlite_where=sa.text("delete_requested_at IS NOT NULL"),
+        )
+
+
+def downgrade() -> None:
+    bind = op.get_bind()
+    columns = _columns(bind)
+    if "delete_requested_at" in columns:
+        # The marker columns are the deletion queue's only durable state:
+        # dropping them while deletions are queued would silently abandon
+        # acknowledged deletions and hand the parent build unusable
+        # (credential-wiped, partially drained) account rows it would list
+        # again. Refuse instead — let the worker finish (or supersede the
+        # deletions via re-import/reauth) before downgrading.
+        pending = bind.execute(
+            sa.text(f"SELECT COUNT(*) FROM {_TABLE} WHERE delete_requested_at IS NOT NULL")  # noqa: S608
+        ).scalar()
+        if pending:
+            raise RuntimeError(
+                f"cannot downgrade {revision}: {pending} account(s) are still queued for "
+                "background deletion; wait for the deletion worker to finish (or supersede "
+                "the deletions with a credential re-import) before downgrading"
+            )
+    if _INDEX in _indexes(bind):
+        op.drop_index(_INDEX, table_name=_TABLE)
+    if "delete_history_requested" in columns:
+        op.drop_column(_TABLE, "delete_history_requested")
+    if "delete_requested_at" in columns:
+        op.drop_column(_TABLE, "delete_requested_at")
@@ -137,6 +137,20 @@ class Account(Base):
         server_default=false(),
         nullable=False,
     )
+    # Pending-deletion marker: set by the fast DELETE path, consumed by the
+    # background deletion worker, cleared only by a credential replacement
+    # (re-import/reauth) that supersedes the deletion. Non-NULL rows are
+    # hidden from account listings and are already unroutable (the fast path
+    # also sets status=DEACTIVATED).
+    delete_requested_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
+    # Frozen at the first delete request (repeat requests do not escalate):
+    # True selects the history-deleting variant in the background worker.
+    delete_history_requested: Mapped[bool] = mapped_column(
+        Boolean,
+        default=False,
+        server_default=false(),
+        nullable=False,
+    )
 
     api_key_assignments: Mapped[list["ApiKeyAccountAssignment"]] = relationship(
         "ApiKeyAccountAssignment",
@@ -2121,6 +2135,17 @@ class HttpBridgeRetryCircuit(Base):
     postgresql_include=["used_percent", "reset_at", "window_minutes", "id"],
 )
 Index("idx_accounts_email", Account.email)
+# Pending-deletion queue: every replica probes ``delete_requested_at IS NOT
+# NULL LIMIT 1`` each worker interval and the leader orders the queue by
+# (delete_requested_at, id); the partial index keeps both reads off the full
+# accounts table and is empty in the steady state (no pending deletions).
+Index(
+    "idx_accounts_delete_requested_at",
+    Account.delete_requested_at,
+    Account.id,
+    postgresql_where=text("delete_requested_at IS NOT NULL"),
+    sqlite_where=text("delete_requested_at IS NOT NULL"),
+)
 Index("idx_api_keys_name", ApiKey.name)
 Index("idx_logs_account_time", RequestLog.account_id, RequestLog.requested_at)
 Index("idx_logs_model_source_time", RequestLog.model_source_id, RequestLog.requested_at)
 
@@ -63,6 +63,7 @@
 from app.core.utils.time import utcnow
 from app.db.session import SessionLocal, close_db, close_session, init_background_db, init_db
 from app.modules.accounts import api as accounts_api
+from app.modules.accounts.deletion import build_account_deletion_scheduler
 from app.modules.accounts.repository import AccountsRepository
 from app.modules.accounts.usage_rollup_scheduler import build_account_usage_rollup_scheduler
 from app.modules.api_keys import api as api_keys_api
@@ -488,6 +489,7 @@ async def lifespan(app: FastAPI):
     automations_scheduler = build_automations_scheduler()
     rate_limit_reset_credits_scheduler = build_rate_limit_reset_credits_scheduler()
     account_usage_rollup_scheduler = build_account_usage_rollup_scheduler()
+    account_deletion_scheduler = build_account_deletion_scheduler()
     data_retention_scheduler = build_data_retention_scheduler()
     telemetry_scheduler = build_telemetry_scheduler()
     start_live_usage_ingestor()
@@ -501,6 +503,7 @@ async def lifespan(app: FastAPI):
     await automations_scheduler.start()
     await rate_limit_reset_credits_scheduler.start()
     await account_usage_rollup_scheduler.start()
+    await account_deletion_scheduler.start()
     await data_retention_scheduler.start()
     await telemetry_scheduler.start()
     if settings.metrics_enabled and PROMETHEUS_AVAILABLE:
@@ -718,6 +721,7 @@ async def _activate_bridge_membership(svc: RingMembershipService, iid: str) -> N
         await stop_live_usage_ingestor()
         await rate_limit_reset_credits_scheduler.stop()
         await account_usage_rollup_scheduler.stop()
+        await account_deletion_scheduler.stop()
         await data_retention_scheduler.stop()
         await telemetry_scheduler.stop()
         # Release the scheduler leader lease only after every leader-gated