diff --git a/homeassistant/components/recorder/queries.py b/homeassistant/components/recorder/queries.py
index 4acf43a491ef6..2e4b588a0b095 100644
--- a/homeassistant/components/recorder/queries.py
+++ b/homeassistant/components/recorder/queries.py
@@ -608,7 +608,8 @@ def delete_recorder_runs_rows(
     """Delete recorder_runs rows."""
     return lambda_stmt(
         lambda: delete(RecorderRuns)
-        .filter(RecorderRuns.start < purge_before)
+        .filter(RecorderRuns.end.is_not(None))
+        .filter(RecorderRuns.end < purge_before)
         .filter(RecorderRuns.run_id != current_run_id)
         .execution_options(synchronize_session=False)
     )
diff --git a/tests/components/recorder/test_purge.py b/tests/components/recorder/test_purge.py
index 245acf4603de8..3f5789cfdcd78 100644
--- a/tests/components/recorder/test_purge.py
+++ b/tests/components/recorder/test_purge.py
@@ -352,6 +352,8 @@ async def test_purge_old_recorder_runs(
     with session_scope(hass=hass) as session:
         recorder_runs = session.query(RecorderRuns)
         assert recorder_runs.count() == 7
+        # Make sure we have a run that is not closed
+        assert sum(run.end is None for run in recorder_runs) == 1
 
     purge_before = dt_util.utcnow()
 
@@ -376,7 +378,9 @@ async def test_purge_old_recorder_runs(
 
     with session_scope(hass=hass) as session:
         recorder_runs = session.query(RecorderRuns)
-        assert recorder_runs.count() == 1
+        assert recorder_runs.count() == 3
+        # Make sure we did not purge the unclosed run
+        assert sum(run.end is None for run in recorder_runs) == 1
 
 
 async def test_purge_old_statistics_runs(
diff --git a/tests/components/recorder/test_purge_v32_schema.py b/tests/components/recorder/test_purge_v32_schema.py
index 0754b2e911c4f..468fd38c85591 100644
--- a/tests/components/recorder/test_purge_v32_schema.py
+++ b/tests/components/recorder/test_purge_v32_schema.py
@@ -347,7 +347,7 @@ async def test_purge_old_recorder_runs(
 
     with session_scope(hass=hass) as session:
         recorder_runs = session.query(RecorderRuns)
-        assert recorder_runs.count() == 1
+        assert recorder_runs.count() == 3
 
 
 async def test_purge_old_statistics_runs(