diff --git a/app/core/worker.py b/app/core/worker.py
index dcf3209..744c14d 100644
--- a/app/core/worker.py
+++ b/app/core/worker.py
@@ -385,6 +385,66 @@ async def cleanup_audit_log_job(ctx: dict[str, Any]) -> None:
 register_job("cleanup_audit_log", cleanup_audit_log_job)
 
 
+# ── Trash cleanup job ───────────────────────────────────────────────────────
+
+async def cleanup_trash_job(ctx: dict[str, Any]) -> None:
+    """Permanently delete soft-deleted records older than 90 days.
+
+    Runs daily to clean up the trash. Iterates per-tenant for RLS compliance.
+    Default retention: 90 days in trash before permanent deletion.
+    """
+    from sqlalchemy import text as sa_text, delete as sa_delete
+    from datetime import datetime, timedelta
+
+    from app.core.db import get_worker_session_factory
+    from app.models.contact import Contact
+    from app.models.entity_attachment import EntityAttachment
+
+    factory = get_worker_session_factory()
+    async with factory() as db:
+        try:
+            tenant_result = await db.execute(sa_text("SELECT id FROM tenants"))
+            tenant_ids = [row[0] for row in tenant_result]
+
+            cutoff = datetime.utcnow() - timedelta(days=90)
+            total_deleted = 0
+
+            for tenant_id in tenant_ids:
+                await db.execute(
+                    sa_text("SELECT set_config('app.current_tenant_id', :tid, true)"),
+                    {"tid": str(tenant_id)},
+                )
+
+                # Delete soft-deleted contacts
+                result = await db.execute(
+                    sa_delete(Contact).where(
+                        Contact.deleted_at.is_not(None),
+                        Contact.deleted_at < cutoff,
+                    )
+                )
+                total_deleted += result.rowcount
+
+                # Delete soft-deleted entity attachments
+                result = await db.execute(
+                    sa_delete(EntityAttachment).where(
+                        EntityAttachment.deleted_at.is_not(None),
+                        EntityAttachment.deleted_at < cutoff,
+                    )
+                )
+                total_deleted += result.rowcount
+
+                await db.commit()
+
+            if total_deleted:
+                logger.info("Trash cleanup: permanently deleted %d old records", total_deleted)
+        except Exception:
+            logger.error("Trash cleanup failed", exc_info=True)
+            await db.rollback()
+
+
+register_job("cleanup_trash", cleanup_trash_job)
+
+
 class WorkerSettings:
     """ARQ worker settings."""
     functions = get_all_jobs()
@@ -419,6 +479,11 @@ class WorkerSettings:
             _wrap_cron_with_lock("cleanup_audit_log", cleanup_audit_log_job, ttl_seconds=300),
             hour=3, minute=0,
         ),
+        # Trash cleanup — daily at 04:00 (90 days retention)
+        cron(
+            _wrap_cron_with_lock("cleanup_trash", cleanup_trash_job, ttl_seconds=300),
+            hour=4, minute=0,
+        ),
         # Scheduled backup — daily at 02:00 (guarded by distributed lock)
         cron(
             _wrap_cron_with_lock("run_backup", get_job("run_backup"), ttl_seconds=600),