diff --git a/backend/alembic/versions/e1f2a3b4c5d6_create_dataset_review_session_table.py b/backend/alembic/versions/e1f2a3b4c5d6_create_dataset_review_session_table.py
new file mode 100644
index 00000000..1277297f
--- /dev/null
+++ b/backend/alembic/versions/e1f2a3b4c5d6_create_dataset_review_session_table.py
@@ -0,0 +1,92 @@
+"""create dataset_review_sessions table
+
+Previously this table was created at runtime by init_db() →
+Base.metadata.create_all(). Migration f0e9d8c7b6a5 references it
+(altering the environment_id FK to add ondelete='CASCADE'), so the
+table must exist when that migration runs.
+
+This migration creates dataset_review_sessions with the FK already
+having ondelete='CASCADE' (matching the model). On fresh databases,
+f0e9d8c7b6a5's drop+recreate of this FK is a no-op. On databases
+upgraded from before the FK had CASCADE, f0e9d8c7b6a5 still applies
+the change.
+
+Revision ID: e1f2a3b4c5d6
+Revises: a5b6c7d8e9f0
+Create Date: 2026-06-11 17:10:00.000000
+
+"""
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = 'e1f2a3b4c5d6'
+down_revision: str | Sequence[str] | None = 'a5b6c7d8e9f0'
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    """Create dataset_review_sessions table."""
+    op.create_table(
+        'dataset_review_sessions',
+        sa.Column('session_id', sa.String(), nullable=False),
+        sa.Column('user_id', sa.String(), nullable=False),
+        sa.Column('environment_id', sa.String(), nullable=False),
+        sa.Column('source_kind', sa.String(), nullable=False),
+        sa.Column('source_input', sa.String(), nullable=False),
+        sa.Column('dataset_ref', sa.String(), nullable=False),
+        sa.Column('dataset_id', sa.Integer(), nullable=True),
+        sa.Column('dashboard_id', sa.Integer(), nullable=True),
+        sa.Column('readiness_state', sa.Enum(
+            'empty', 'importing', 'review_ready', 'semantic_source_review_needed',
+            'clarification_needed', 'clarification_active', 'mapping_review_needed',
+            'compiled_preview_ready', 'partially_ready', 'run_ready',
+            'run_in_progress', 'completed', 'recovery_required',
+            name='readinessstate', create_type=False,
+        ), nullable=False),
+        sa.Column('recommended_action', sa.Enum(
+            'import_from_superset', 'review_documentation', 'apply_semantic_source',
+            'start_clarification', 'answer_next_question', 'approve_mapping',
+            'generate_sql_preview', 'complete_required_values', 'launch_dataset',
+            'resume_session', 'export_outputs',
+            name='recommendedaction', create_type=False,
+        ), nullable=False),
+        sa.Column('version', sa.Integer(), nullable=False, server_default=sa.text('0')),
+        sa.Column('status', sa.Enum(
+            'active', 'paused', 'completed', 'archived', 'cancelled',
+            name='sessionstatus', create_type=False,
+        ), nullable=False),
+        sa.Column('current_phase', sa.Enum(
+            'intake', 'recovery', 'review', 'semantic_review', 'clarification',
+            'mapping_review', 'preview', 'launch', 'post_run',
+            name='sessionphase', create_type=False,
+        ), nullable=False),
+        sa.Column('active_task_id', sa.String(), nullable=True),
+        sa.Column('last_preview_id', sa.String(), nullable=True),
+        sa.Column('last_run_context_id', sa.String(), nullable=True),
+        sa.Column('created_at', sa.DateTime(), nullable=False),
+        sa.Column('updated_at', sa.DateTime(), nullable=False),
+        sa.Column('last_activity_at', sa.DateTime(), nullable=False),
+        sa.Column('closed_at', sa.DateTime(), nullable=True),
+        sa.ForeignKeyConstraint(
+            ['environment_id'], ['environments.id'],
+            ondelete='CASCADE', name='dataset_review_sessions_environment_id_fkey',
+        ),
+        sa.ForeignKeyConstraint(
+            ['user_id'], ['users.id'],
+            name='dataset_review_sessions_user_id_fkey',
+        ),
+        sa.PrimaryKeyConstraint('session_id'),
+    )
+
+
+def downgrade() -> None:
+    """Drop dataset_review_sessions table."""
+    op.drop_table('dataset_review_sessions')
+    op.execute('DROP TYPE IF EXISTS readinessstate')
+    op.execute('DROP TYPE IF EXISTS recommendedaction')
+    op.execute('DROP TYPE IF EXISTS sessionstatus')
+    op.execute('DROP TYPE IF EXISTS sessionphase')
diff --git a/backend/alembic/versions/f0e9d8c7b6a5_cascade_ondelete_all_env_fks.py b/backend/alembic/versions/f0e9d8c7b6a5_cascade_ondelete_all_env_fks.py
index 25e9130c..aab44b85 100644
--- a/backend/alembic/versions/f0e9d8c7b6a5_cascade_ondelete_all_env_fks.py
+++ b/backend/alembic/versions/f0e9d8c7b6a5_cascade_ondelete_all_env_fks.py
@@ -1,7 +1,7 @@
 """cascade ondelete for all remaining environments FK references
 
 Revision ID: f0e9d8c7b6a5
-Revises: a5b6c7d8e9f0, e5f4d3c2b1a
+Revises: e1f2a3b4c5d6, e5f4d3c2b1a
 Create Date: 2026-05-21 19:00:00.000000
 
 """
@@ -11,7 +11,7 @@ from alembic import op
 
 # revision identifiers, used by Alembic.
 revision: str = 'f0e9d8c7b6a5'
-down_revision: str | Sequence[str] | None = ('a5b6c7d8e9f0', 'e5f4d3c2b1a')
+down_revision: str | Sequence[str] | None = ('e1f2a3b4c5d6', 'e5f4d3c2b1a')
 branch_labels: str | Sequence[str] | None = None
 depends_on: str | Sequence[str] | None = None