diff --git a/homeassistant/components/backup/__init__.py b/homeassistant/components/backup/__init__.py
index 86e5b95d196490..f97805b19234f9 100644
--- a/homeassistant/components/backup/__init__.py
+++ b/homeassistant/components/backup/__init__.py
@@ -26,6 +26,8 @@
     BackupReaderWriterError,
     CoreBackupReaderWriter,
     CreateBackupEvent,
+    CreateBackupStage,
+    CreateBackupState,
     IdleEvent,
     IncorrectPasswordError,
     ManagerBackup,
@@ -49,6 +51,8 @@
     "BackupReaderWriter",
     "BackupReaderWriterError",
     "CreateBackupEvent",
+    "CreateBackupStage",
+    "CreateBackupState",
     "Folder",
     "IdleEvent",
     "IncorrectPasswordError",
diff --git a/homeassistant/components/hassio/backup.py b/homeassistant/components/hassio/backup.py
index 4aad984cc54ecc..43451e96b372ef 100644
--- a/homeassistant/components/hassio/backup.py
+++ b/homeassistant/components/hassio/backup.py
@@ -30,6 +30,8 @@
     BackupReaderWriter,
     BackupReaderWriterError,
     CreateBackupEvent,
+    CreateBackupStage,
+    CreateBackupState,
     Folder,
     IdleEvent,
     IncorrectPasswordError,
@@ -47,6 +49,7 @@
 from homeassistant.exceptions import HomeAssistantError
 from homeassistant.helpers.dispatcher import async_dispatcher_connect
 from homeassistant.util import dt as dt_util
+from homeassistant.util.enum import try_parse_enum
 
 from .const import DOMAIN, EVENT_SUPERVISOR_EVENT
 from .handler import get_supervisor_client
@@ -336,6 +339,7 @@ async def async_create_backup(
             self._async_wait_for_backup(
                 backup,
                 locations,
+                on_progress=on_progress,
                 remove_after_upload=locations == [LOCATION_CLOUD_BACKUP],
             ),
             name="backup_manager_create_backup",
@@ -349,6 +353,7 @@ async def _async_wait_for_backup(
         backup: supervisor_backups.NewBackup,
         locations: list[str | None],
         *,
+        on_progress: Callable[[CreateBackupEvent], None],
         remove_after_upload: bool,
     ) -> WrittenBackup:
         """Wait for a backup to complete."""
@@ -360,6 +365,14 @@ async def _async_wait_for_backup(
         def on_job_progress(data: Mapping[str, Any]) -> None:
             """Handle backup progress."""
             nonlocal backup_id
+            if not (stage := try_parse_enum(CreateBackupStage, data.get("stage"))):
+                _LOGGER.debug("Unknown create stage: %s", data.get("stage"))
+            else:
+                on_progress(
+                    CreateBackupEvent(
+                        reason=None, stage=stage, state=CreateBackupState.IN_PROGRESS
+                    )
+                )
             if data.get("done") is True:
                 backup_id = data.get("reference")
                 create_errors.extend(data.get("errors", []))
diff --git a/tests/components/hassio/test_backup.py b/tests/components/hassio/test_backup.py
index ab3335e00dc6e7..023a19a223f70d 100644
--- a/tests/components/hassio/test_backup.py
+++ b/tests/components/hassio/test_backup.py
@@ -1002,6 +1002,113 @@ async def test_reader_writer_create(
     assert response["event"] == {"manager_state": "idle"}
 
 
+@pytest.mark.usefixtures("hassio_client", "setup_integration")
+async def test_reader_writer_create_report_progress(
+    hass: HomeAssistant,
+    hass_ws_client: WebSocketGenerator,
+    freezer: FrozenDateTimeFactory,
+    supervisor_client: AsyncMock,
+) -> None:
+    """Test generating a backup."""
+    client = await hass_ws_client(hass)
+    freezer.move_to("2025-01-30 13:42:12.345678")
+    supervisor_client.backups.partial_backup.return_value.job_id = TEST_JOB_ID
+    supervisor_client.backups.backup_info.return_value = TEST_BACKUP_DETAILS
+    supervisor_client.jobs.get_job.return_value = TEST_JOB_NOT_DONE
+
+    await client.send_json_auto_id({"type": "backup/subscribe_events"})
+    response = await client.receive_json()
+    assert response["event"] == {"manager_state": "idle"}
+    response = await client.receive_json()
+    assert response["success"]
+
+    await client.send_json_auto_id(
+        {"type": "backup/generate", "agent_ids": ["hassio.local"], "name": "Test"}
+    )
+    response = await client.receive_json()
+    assert response["event"] == {
+        "manager_state": "create_backup",
+        "reason": None,
+        "stage": None,
+        "state": "in_progress",
+    }
+
+    response = await client.receive_json()
+    assert response["success"]
+    assert response["result"] == {"backup_job_id": TEST_JOB_ID}
+
+    supervisor_client.backups.partial_backup.assert_called_once_with(
+        DEFAULT_BACKUP_OPTIONS
+    )
+
+    supervisor_event_base = {"uuid": TEST_JOB_ID, "reference": "test_slug"}
+    supervisor_events = [
+        supervisor_event_base | {"done": False, "stage": "addon_repositories"},
+        supervisor_event_base | {"done": False, "stage": None},  # Will be skipped
+        supervisor_event_base | {"done": False, "stage": "unknown"},  # Will be skipped
+        supervisor_event_base | {"done": False, "stage": "home_assistant"},
+        supervisor_event_base | {"done": False, "stage": "addons"},
+        supervisor_event_base | {"done": True, "stage": "finishing_file"},
+    ]
+    expected_manager_events = [
+        "addon_repositories",
+        "home_assistant",
+        "addons",
+        "finishing_file",
+    ]
+
+    for supervisor_event in supervisor_events:
+        await client.send_json_auto_id(
+            {
+                "type": "supervisor/event",
+                "data": {"event": "job", "data": supervisor_event},
+            }
+        )
+
+    acks = 0
+    events = []
+    for _ in range(len(supervisor_events) + len(expected_manager_events)):
+        response = await client.receive_json()
+        if "event" in response:
+            events.append(response)
+            continue
+        assert response["success"]
+        acks += 1
+
+    assert acks == len(supervisor_events)
+    assert len(events) == len(expected_manager_events)
+
+    for i, event in enumerate(events):
+        assert event["event"] == {
+            "manager_state": "create_backup",
+            "reason": None,
+            "stage": expected_manager_events[i],
+            "state": "in_progress",
+        }
+
+    response = await client.receive_json()
+    assert response["event"] == {
+        "manager_state": "create_backup",
+        "reason": None,
+        "stage": "upload_to_agents",
+        "state": "in_progress",
+    }
+
+    response = await client.receive_json()
+    assert response["event"] == {
+        "manager_state": "create_backup",
+        "reason": None,
+        "stage": None,
+        "state": "completed",
+    }
+
+    supervisor_client.backups.download_backup.assert_not_called()
+    supervisor_client.backups.remove_backup.assert_not_called()
+
+    response = await client.receive_json()
+    assert response["event"] == {"manager_state": "idle"}
+
+
 @pytest.mark.usefixtures("hassio_client", "setup_integration")
 async def test_reader_writer_create_job_done(
     hass: HomeAssistant,