diff --git a/application/single_app/config.py b/application/single_app/config.py
index 26c151c6..e3f9a364 100644
--- a/application/single_app/config.py
+++ b/application/single_app/config.py
@@ -96,7 +96,7 @@
 EXECUTOR_TYPE = 'thread'
 EXECUTOR_MAX_WORKERS = 30
 SESSION_TYPE = 'filesystem'
-VERSION = "0.260.002"
+VERSION = "0.260.003"
 IS_DEVELOPMENT = is_development_env_enabled()
 
 SESSION_COOKIE_SAMESITE = os.getenv('SESSION_COOKIE_SAMESITE', 'Lax')
diff --git a/application/single_app/functions_data_management.py b/application/single_app/functions_data_management.py
index 0c8565bb..56a2505a 100644
--- a/application/single_app/functions_data_management.py
+++ b/application/single_app/functions_data_management.py
@@ -13202,6 +13202,7 @@ def _set_job_progress(
     total_steps,
     current_step=None,
     status=DATA_MANAGEMENT_STATUS_RUNNING,
+    step_status=None,
     allow_cancel_requested=False,
 ):
     total_steps = max(1, total_steps)
@@ -13230,13 +13231,27 @@ def _set_job_progress(
         saved_job.get("id"),
         current_step or "progress",
         saved_job,
-        status=status,
+        # A finished step stays "completed" even while the job itself keeps running.
+        status=step_status or status,
         message=message,
         details={"progress": saved_job.get("progress") if isinstance(saved_job.get("progress"), dict) else {}},
     )
     return saved_job
 
 
+def _complete_job_step(job, message, completed_steps, total_steps, current_step, **kwargs):
+    """Advance job progress and stamp the finished step as completed."""
+    return _set_job_progress(
+        job,
+        message,
+        completed_steps,
+        total_steps,
+        current_step=current_step,
+        step_status=DATA_MANAGEMENT_STATUS_COMPLETED,
+        **kwargs,
+    )
+
+
 def _get_backup_fernet(settings, key_reference=None):
     if not settings.get("encryption_enabled"):
         return None
@@ -17973,7 +17988,7 @@ def execute_restore_job(job, settings):
         container_client,
         fernet,
     ))
-    _set_job_progress(job, "Cosmos restore step completed", 1, total_steps, current_step="cosmos")
+    _complete_job_step(job, "Cosmos restore step completed", 1, total_steps, "cosmos")
 
     artifacts.extend(_execute_restore_search_resources(
         job,
@@ -17983,7 +17998,7 @@ def execute_restore_job(job, settings):
         container_client,
         fernet,
     ))
-    _set_job_progress(job, "AI Search restore step completed", 2, total_steps, current_step="ai_search")
+    _complete_job_step(job, "AI Search restore step completed", 2, total_steps, "ai_search")
 
     artifacts.extend(_execute_restore_source_blob_resources(
         job,
@@ -17993,7 +18008,7 @@ def execute_restore_job(job, settings):
         container_client,
         fernet,
     ))
-    _set_job_progress(job, "Source blob restore step completed", 3, total_steps, current_step="source_blobs")
+    _complete_job_step(job, "Source blob restore step completed", 3, total_steps, "source_blobs")
 
     warnings = list(restore_state.get("warnings") or [])
     failed_resource_names = [
@@ -18142,7 +18157,7 @@ def execute_backup_job(job, settings):
             warning,
             "Skipped disabled Cosmos backup scope",
         )
-    _set_job_progress(job, "Cosmos DB export step completed", 1, total_steps, current_step="cosmos")
+    _complete_job_step(job, "Cosmos DB export step completed", 1, total_steps, "cosmos")
 
     if backup_plan.get("include_ai_search"):
         artifacts.extend(_execute_backup_search_resources(
@@ -18164,7 +18179,7 @@ def execute_backup_job(job, settings):
             warning,
             "Skipped disabled AI Search backup scope",
         )
-    _set_job_progress(job, "AI Search export step completed", 2, total_steps, current_step="ai_search")
+    _complete_job_step(job, "AI Search export step completed", 2, total_steps, "ai_search")
 
     if backup_plan.get("include_source_blobs"):
         source_blob_service_client = _get_source_blob_service_client()
@@ -18233,7 +18248,7 @@ def execute_backup_job(job, settings):
             warning,
             "Skipped disabled source blob backup scope",
         )
-    _set_job_progress(job, "Source blob export step completed", 3, total_steps, current_step="source_blobs")
+    _complete_job_step(job, "Source blob export step completed", 3, total_steps, "source_blobs")
 
     _assert_backup_job_lease(job)
     artifacts = _backup_state_resource_artifacts(backup_state)
@@ -18424,13 +18439,13 @@ def execute_migration_job(job, settings):
             allow_cancel_requested=True,
         )
         raise
-    _set_job_progress(job, "Validated migration selection plan", 1, total_steps, current_step="plan")
+    _complete_job_step(job, "Validated migration selection plan", 1, total_steps, "plan")
     migration_state = job.get("migration_state") if isinstance(job.get("migration_state"), dict) else migration_state
     _record_data_management_job_event(
         job.get("id"),
         "migration-plan",
         job,
-        status=DATA_MANAGEMENT_STATUS_RUNNING,
+        status=DATA_MANAGEMENT_STATUS_COMPLETED,
         message="Migration selection plan validated",
         details={
             "migration_plan": plan_summary,
@@ -18552,7 +18567,7 @@ def preview_heartbeat(message, completed_count=0):
                 "Pinned server-owned migration inventory preview",
             )
 
-        _set_job_progress(job, "Migration inventory completed", 2, total_steps, current_step="inventory")
+        _complete_job_step(job, "Migration inventory completed", 2, total_steps, "inventory")
         _set_job_progress(job, "Validating migration destinations", 2, total_steps, current_step="preflight")
         migration_state = _run_data_management_migration_preflight(
             job,
@@ -18560,13 +18575,13 @@ def preview_heartbeat(message, completed_count=0):
             settings,
             migration_plan,
         )
-        _set_job_progress(job, "Destination migration preflight completed", 3, total_steps, current_step="preflight")
+        _complete_job_step(job, "Destination migration preflight completed", 3, total_steps, "preflight")
         migration_state = job.get("migration_state") if isinstance(job.get("migration_state"), dict) else migration_state
         _record_data_management_job_event(
             job.get("id"),
             "migration-preflight",
             job,
-            status=DATA_MANAGEMENT_STATUS_RUNNING,
+            status=DATA_MANAGEMENT_STATUS_COMPLETED,
             message="Verified source and destination migration access",
             details=migration_state.get("preflight") if isinstance(migration_state.get("preflight"), dict) else {},
         )
@@ -18578,7 +18593,7 @@ def preview_heartbeat(message, completed_count=0):
             settings,
             migration_plan,
         )
-        _set_job_progress(job, "Destination Cosmos capacity prepared", 4, total_steps, current_step="capacity")
+        _complete_job_step(job, "Destination Cosmos capacity prepared", 4, total_steps, "capacity")
         migration_state = job.get("migration_state") if isinstance(job.get("migration_state"), dict) else migration_state
 
         _set_job_progress(job, "Migrating Cosmos records", 4, total_steps, current_step="cosmos")
@@ -18601,11 +18616,11 @@ def preview_heartbeat(message, completed_count=0):
                 job.get("id"),
                 f"migration-cosmos-{target_type}",
                 job,
-                status=DATA_MANAGEMENT_STATUS_RUNNING,
+                status=DATA_MANAGEMENT_STATUS_COMPLETED,
                 message=f"Migrated {target_type.replace('_', ' ')} Cosmos records",
                 details={"target_type": target_type, "artifacts": copied},
             )
-        _set_job_progress(job, "Cosmos migration completed", 5, total_steps, current_step="cosmos")
+        _complete_job_step(job, "Cosmos migration completed", 5, total_steps, "cosmos")
         migration_state = job.get("migration_state") if isinstance(job.get("migration_state"), dict) else migration_state
 
         search_artifacts = []
@@ -18650,7 +18665,7 @@ def preview_heartbeat(message, completed_count=0):
                     )
         migration_state = job.get("migration_state") if isinstance(job.get("migration_state"), dict) else migration_state
         artifacts.extend(search_artifacts)
-        _set_job_progress(job, "AI Search migration completed", 6, total_steps, current_step="ai_search")
+        _complete_job_step(job, "AI Search migration completed", 6, total_steps, "ai_search")
         migration_state = job.get("migration_state") if isinstance(job.get("migration_state"), dict) else migration_state
 
         _set_job_progress(job, "Migrating source document blobs", 6, total_steps, current_step="source_blobs")
@@ -18663,7 +18678,7 @@ def preview_heartbeat(message, completed_count=0):
         )
         migration_state = job.get("migration_state") if isinstance(job.get("migration_state"), dict) else migration_state
         artifacts.extend(source_blob_artifacts)
-        _set_job_progress(job, "Source blob migration completed", 7, total_steps, current_step="source_blobs")
+        _complete_job_step(job, "Source blob migration completed", 7, total_steps, "source_blobs")
         migration_state = job.get("migration_state") if isinstance(job.get("migration_state"), dict) else migration_state
 
         _set_job_progress(job, "Reconciling source and destination", 7, total_steps, current_step="reconciliation")
@@ -18679,12 +18694,12 @@ def preview_heartbeat(message, completed_count=0):
         )
         migration_state = job.get("migration_state") if isinstance(job.get("migration_state"), dict) else migration_state
         artifacts.append(reconciliation_artifact)
-        _set_job_progress(job, "Migration reconciliation completed", 8, total_steps, current_step="reconciliation")
+        _complete_job_step(job, "Migration reconciliation completed", 8, total_steps, "reconciliation")
         _record_data_management_job_event(
             job.get("id"),
             "migration-reconciliation",
             job,
-            status=DATA_MANAGEMENT_STATUS_RUNNING,
+            status=DATA_MANAGEMENT_STATUS_COMPLETED,
             message="Reconciled migration source and destination identities",
             details=reconciliation_artifact,
         )
@@ -18762,7 +18777,7 @@ def preview_heartbeat(message, completed_count=0):
         settings,
         "Migration execution completed",
     )
-    _set_job_progress(job, "Migration execution completed", 10, total_steps, current_step="complete")
+    _complete_job_step(job, "Migration execution completed", 10, total_steps, "complete")
 
     artifact_summaries = summarize_backup_artifacts(artifacts)
     artifact_totals = _backup_artifact_totals(artifact_summaries)
diff --git a/application/single_app/static/js/admin/admin_data_management.js b/application/single_app/static/js/admin/admin_data_management.js
index 3e0bd9c8..481b2069 100644
--- a/application/single_app/static/js/admin/admin_data_management.js
+++ b/application/single_app/static/js/admin/admin_data_management.js
@@ -3853,10 +3853,12 @@ function getMigrationLiveMetrics(job) {
         const hasRecentProgress = timestampAgeSeconds(
             job?.last_progress_at || migrationState.last_progress_at
         ) <= 10;
-        metrics.push({
-            label: "Liveness",
-            value: hasRecentProgress ? "Running - progress active" : "Running - alive, no recent progress",
-        });
+        if (!isTerminalJobStatus(job?.status)) {
+            metrics.push({
+                label: "Liveness",
+                value: hasRecentProgress ? "Running - progress active" : "Running - alive, no recent progress",
+            });
+        }
     }
     return metrics;
 }
@@ -3869,15 +3871,18 @@ function getBackupLiveMetrics(job) {
     const totals = backupState.totals && typeof backupState.totals === "object" ? backupState.totals : {};
     const telemetry = backupState.telemetry && typeof backupState.telemetry === "object" ? backupState.telemetry : {};
     const sourceCapacity = backupState.source_capacity && typeof backupState.source_capacity === "object" ? backupState.source_capacity : {};
-    const metrics = [
-        { label: "Current container", value: telemetry.current_container || "Waiting" },
+    const metrics = [];
+    if (!isTerminalJobStatus(job?.status)) {
+        metrics.push({ label: "Current container", value: telemetry.current_container || "Waiting" });
+    }
+    metrics.push(
         { label: "Checkpoint position", value: formatNumber(telemetry.checkpoint_position || totals.checkpoint_count || 0) },
         { label: "Processed", value: formatNumber(telemetry.records_processed || totals.processed_count || 0) },
         { label: "Transferred", value: formatBytes(telemetry.bytes || totals.bytes || 0) },
         { label: "Request units", value: formatNumber(telemetry.request_units || totals.request_units || 0) },
         { label: "Retries / throttles", value: `${formatNumber(telemetry.retries || totals.retry_attempt_count || 0)} / ${formatNumber(telemetry.throttles || totals.throttle_count || 0)}` },
         { label: "Skipped / failed", value: `${formatNumber(totals.skipped_count || 0)} / ${formatNumber(totals.failed_count || 0)}` },
-    ];
+    );
     if (telemetry.elapsed_seconds !== undefined || totals.elapsed_seconds !== undefined) {
         metrics.push({ label: "Elapsed", value: `${formatNumber(telemetry.elapsed_seconds ?? totals.elapsed_seconds ?? 0)}s` });
     }
@@ -3923,6 +3928,10 @@ function getRestoreLiveMetrics(job) {
     return metrics;
 }
 
+function isTerminalJobStatus(status) {
+    return ["completed", "completed_with_warnings", "failed", "canceled"].includes(String(status || ""));
+}
+
 function timestampAgeSeconds(value) {
     const timestamp = Date.parse(value || "");
     if (!Number.isFinite(timestamp)) {
diff --git a/docs/explanation/fixes/DATA_MANAGEMENT_JOB_STEP_STATUS_FIX.md b/docs/explanation/fixes/DATA_MANAGEMENT_JOB_STEP_STATUS_FIX.md
new file mode 100644
index 00000000..9274cd5c
--- /dev/null
+++ b/docs/explanation/fixes/DATA_MANAGEMENT_JOB_STEP_STATUS_FIX.md
@@ -0,0 +1,177 @@
+# Data Management Job Step Status Fix
+
+**Fixed in version: 0.260.003**
+
+Fixes #1272. Related: #1258, #1271, #1276.
+
+## Issue Description
+
+On a finished Data Management job, the details panel misrepresented state in two ways:
+
+1. Timeline entries for steps that had clearly finished — `Cosmos DB export step completed`,
+   `AI Search export step completed`, `Source blob export step completed` — still displayed a
+   `running` badge.
+2. The live metrics grid still showed **Current container: Waiting** on a job that had already
+   reached `completed_with_warnings`.
+
+Together these made a completed job look stuck. Admins reported the history reading as
+"queued, running, then complete" at the job level while individual steps never advanced past
+`running`.
+
+## Root Cause
+
+### Step badges stuck on `running`
+
+`_set_job_progress()` used a single `status` value for two different things: the **job**
+document and the **timeline event** it recorded.
+
+```python
+def _set_job_progress(job, message, completed_steps, total_steps,
+                      current_step=None, status=DATA_MANAGEMENT_STATUS_RUNNING, ...):
+    job.update({"status": status, ...})
+    saved_job = _save_data_management_job(job)
+    _record_data_management_job_event(
+        saved_job.get("id"), current_step or "progress", saved_job,
+        status=status,          # job status, not step outcome
+        ...
+    )
+```
+
+Every call site that announced a finished step relied on the default `status`, which is
+`DATA_MANAGEMENT_STATUS_RUNNING`:
+
+```python
+_set_job_progress(job, "Cosmos DB export step completed", 1, total_steps, current_step="cosmos")
+```
+
+The event was historically accurate — the *job* was running when the step finished — but the
+timeline presents that badge as the *step's* state. There was no way to express "this step is
+done, the job is not" because both shared one field.
+
+Four migration events recorded through `_record_data_management_job_event()` had the same
+problem: `migration-plan`, `migration-preflight`, `migration-cosmos-{target_type}`, and
+`migration-reconciliation` all describe finished work but were stamped
+`DATA_MANAGEMENT_STATUS_RUNNING`.
+
+The `queued` half of the report was **not** a defect. Every event call site passes an explicit
+status, so the `status=DATA_MANAGEMENT_STATUS_QUEUED` default on
+`_record_data_management_job_event()` is never actually stranded on a step. The `queued` and
+`*-retry-queued` entries genuinely describe queueing actions.
+
+### Stale "Current container"
+
+`_execute_backup_source_blob_resource` writes its final checkpoint after the worker pool
+drains, so `telemetry.current_container` is empty at completion. The frontend rendered its
+empty-state label unconditionally:
+
+```javascript
+{ label: "Current container", value: telemetry.current_container || "Waiting" }
+```
+
+`getMigrationLiveMetrics` had the equivalent problem with a hardcoded
+`Liveness: Running - ...` row.
+
+## Technical Details
+
+### Files Modified
+
+| File | Change |
+|------|--------|
+| `application/single_app/functions_data_management.py` | Added `step_status` to `_set_job_progress`; added `_complete_job_step`; moved 15 step-completion call sites onto it; corrected 4 migration outcome events |
+| `application/single_app/static/js/admin/admin_data_management.js` | Added `isTerminalJobStatus`; gated live-only telemetry rows |
+| `functional_tests/test_data_management_job_step_status.py` | New coverage |
+| `application/single_app/config.py` | Version bump |
+
+### Code Changes
+
+Step status is now independent of job status:
+
+```python
+def _set_job_progress(job, message, completed_steps, total_steps, current_step=None,
+                      status=DATA_MANAGEMENT_STATUS_RUNNING, step_status=None,
+                      allow_cancel_requested=False):
+    ...
+    _record_data_management_job_event(
+        saved_job.get("id"), current_step or "progress", saved_job,
+        # A finished step stays "completed" even while the job itself keeps running.
+        status=step_status or status,
+        ...
+    )
+
+
+def _complete_job_step(job, message, completed_steps, total_steps, current_step, **kwargs):
+    """Advance job progress and stamp the finished step as completed."""
+    return _set_job_progress(
+        job, message, completed_steps, total_steps,
+        current_step=current_step,
+        step_status=DATA_MANAGEMENT_STATUS_COMPLETED,
+        **kwargs,
+    )
+```
+
+`step_status` defaults to `None` and falls back to `status`, so every call site that was not
+deliberately migrated keeps its previous behavior. Steps that *start* still report `running`,
+and genuinely terminal calls such as `Restore completed` still propagate their terminal status
+to both the job and the event.
+
+Resulting lifecycle per step: `running` when the step begins, `completed` when it finishes,
+while the job stays `running` until it actually completes.
+
+### Frontend
+
+```javascript
+function isTerminalJobStatus(status) {
+    return ["completed", "completed_with_warnings", "failed", "canceled"].includes(String(status || ""));
+}
+```
+
+`getBackupLiveMetrics` only emits **Current container** when the job is not terminal, and
+`getMigrationLiveMetrics` only emits **Liveness** when the job is not terminal. Cumulative
+metrics (processed, transferred, request units, retries, skipped/failed) are unchanged and
+still render on finished jobs.
+
+## Validation
+
+### Test Results
+
+`functional_tests/test_data_management_job_step_status.py` — **18 passed**.
+
+Coverage:
+
+- `test_finished_step_is_completed_while_job_keeps_running` — `_complete_job_step` records
+  `completed` while the job document stays `running`.
+- `test_in_progress_step_still_reports_running` — started steps are unaffected.
+- `test_terminal_job_progress_stamps_terminal_step_status` — an explicit terminal job status
+  still reaches the final event.
+- `test_step_completions_use_the_completed_step_helper` — parameterized across all 12
+  step-completion messages.
+- `test_migration_outcome_events_are_not_stamped_running` — the 4 migration outcome events.
+- `test_terminal_jobs_hide_live_current_container` — the frontend guard precedes the
+  **Current container** row.
+
+Broader suite (`-k data_management`) — **175 passed**, with only the two known pre-existing
+issues: the `swagger_wrapper` collection error in `test_admin_endpoint.py` and
+`test_backup_recovery_and_admin_progress_are_bounded_and_sanitized`.
+
+### Regression Probe
+
+Reverting `status=step_status or status` back to `status=status` failed
+`test_finished_step_is_completed_while_job_keeps_running` with
+`AssertionError: assert 'running' == 'completed'`, confirming the test fails for the right
+reason. The fix was then restored and all 18 tests passed.
+
+### Before / After
+
+| Timeline entry | Before | After |
+|----------------|--------|-------|
+| `Cosmos DB export step completed` | `running` | `completed` |
+| `AI Search export step completed` | `running` | `completed` |
+| `Source blob export step completed` | `running` | `completed` |
+| `Migration reconciliation completed` | `running` | `completed` |
+| Started step (e.g. `Migrating Cosmos records`) | `running` | `running` |
+| Job status while steps complete | `running` | `running` |
+
+| Panel row on a finished job | Before | After |
+|------------------------------|--------|-------|
+| Current container | `Waiting` | hidden |
+| Liveness (migration) | `Running - ...` | hidden |
diff --git a/docs/explanation/release_notes.md b/docs/explanation/release_notes.md
index 332b082b..2c9f83af 100644
--- a/docs/explanation/release_notes.md
+++ b/docs/explanation/release_notes.md
@@ -2,6 +2,25 @@
 
 For feature-focused and fix-focused drill-downs by version, see [Features by Version](/explanation/features/) and [Fixes by Version](/explanation/fixes/).
 
+### **(v0.260.003)**
+
+#### Bug Fixes
+
+*   **Data Management Timeline Steps Now Show Their Own Status**
+    *   Fixed completed job steps showing a `running` badge on the Data Management job timeline. Events such as "Cosmos DB export step completed" and "Migration reconciliation completed" now read `completed`.
+    *   Root cause was `_set_job_progress` stamping the **job** status onto every step event it recorded, so a finished step inherited `running` because the job itself was still running.
+    *   Step status is now decoupled from job status: steps that start report `running`, steps that finish report `completed`, and the job continues running until it genuinely finishes.
+    *   Applies to backup, restore, and migration timelines.
+    *   (Ref: `functions_data_management.py`, `_set_job_progress`, `_complete_job_step`, `_record_data_management_job_event`)
+
+#### User Interface Enhancements
+
+*   **Finished Jobs No Longer Look Stuck**
+    *   Completed backup jobs no longer display **Current container: Waiting**, which made a finished job look like it was still churning.
+    *   Migration jobs no longer display a **Liveness: Running** row after reaching a terminal status.
+    *   Live-only telemetry is now hidden once a job is `completed`, `completed_with_warnings`, `failed`, or `canceled`.
+    *   (Ref: `admin_data_management.js`, `getBackupLiveMetrics`, `getMigrationLiveMetrics`, `isTerminalJobStatus`)
+
 ### **(v0.260.002)**
 
 #### Bug Fixes
diff --git a/functional_tests/test_data_management_job_step_status.py b/functional_tests/test_data_management_job_step_status.py
new file mode 100644
index 00000000..03ec928b
--- /dev/null
+++ b/functional_tests/test_data_management_job_step_status.py
@@ -0,0 +1,218 @@
+#!/usr/bin/env python3
+# test_data_management_job_step_status.py
+"""
+Functional test for Data Management per-step timeline status.
+Version: 0.260.003
+Implemented in: 0.260.003
+
+This test ensures finished job steps are recorded as completed instead of
+inheriting the running job status, and that terminal jobs stop rendering
+live-only telemetry such as "Current container: Waiting".
+"""
+
+import copy
+import importlib.util
+from pathlib import Path
+import sys
+import types
+
+import pytest
+
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+APP_ROOT = REPO_ROOT / "application" / "single_app"
+MODULE_PATH = APP_ROOT / "functions_data_management.py"
+ADMIN_JS_PATH = APP_ROOT / "static" / "js" / "admin" / "admin_data_management.js"
+sys.path.insert(0, str(APP_ROOT))
+sys.path.insert(0, str(Path(__file__).resolve().parent))
+
+from test_support.versioning import assert_app_version_at_least
+
+
+class StubContainer:
+    """Accept the container calls made during module import without storing state."""
+
+    def create_item(self, body):
+        return copy.deepcopy(body)
+
+    def upsert_item(self, body):
+        return copy.deepcopy(body)
+
+    def query_items(self, *_args, **_kwargs):
+        return []
+
+
+def load_data_management_module(monkeypatch):
+    """Load production job-progress helpers with inert Cosmos dependencies."""
+    container = StubContainer()
+    config_module = types.ModuleType("config")
+    config_module.CLIENTS = {}
+    config_module.VERSION = "0.260.003"
+    config_module.cosmos_data_management_jobs_container = container
+    config_module.cosmos_data_management_job_items_container = container
+    config_module.cosmos_settings_container = container
+    config_module.cosmos_data_management_backup_item_states_container = container
+    monkeypatch.setitem(sys.modules, "config", config_module)
+
+    appinsights_module = types.ModuleType("functions_appinsights")
+    appinsights_module.log_event = lambda *_args, **_kwargs: None
+    monkeypatch.setitem(sys.modules, "functions_appinsights", appinsights_module)
+
+    throughput_module = types.ModuleType("functions_cosmos_throughput")
+
+    class FakeCosmosThroughputError(Exception):
+        pass
+
+    throughput_module.CosmosThroughputError = FakeCosmosThroughputError
+    throughput_module.get_container_throughput = lambda *_args, **_kwargs: {}
+    throughput_module.get_database_throughput = lambda *_args, **_kwargs: {}
+    throughput_module.set_database_throughput = lambda *_args, **_kwargs: {}
+    monkeypatch.setitem(sys.modules, "functions_cosmos_throughput", throughput_module)
+
+    module_name = "data_management_step_status_test_module"
+    spec = importlib.util.spec_from_file_location(module_name, MODULE_PATH)
+    module = importlib.util.module_from_spec(spec)
+    sys.modules[module_name] = module
+    assert spec and spec.loader
+    spec.loader.exec_module(module)
+    monkeypatch.delitem(sys.modules, module_name, raising=False)
+    return module
+
+
+def capture_progress_events(monkeypatch, module):
+    """Bypass lease/persistence plumbing and record emitted timeline events."""
+    recorded = []
+    monkeypatch.setattr(module, "_assert_data_management_job_lease", lambda *_a, **_k: None)
+    monkeypatch.setattr(module, "_save_data_management_job", lambda job: copy.deepcopy(job))
+
+    def record_event(job_id, step_name, job, status=None, message=None, details=None):
+        recorded.append({
+            "job_id": job_id,
+            "step_name": step_name,
+            "status": status,
+            "message": message,
+            "job_status": job.get("status"),
+        })
+
+    monkeypatch.setattr(module, "_record_data_management_job_event", record_event)
+    return recorded
+
+
+def build_job():
+    return {"id": "job-1", "operation": "backup", "status": "running"}
+
+
+def test_finished_step_is_completed_while_job_keeps_running(monkeypatch):
+    """Record a finished step as completed without ending the job."""
+    module = load_data_management_module(monkeypatch)
+    recorded = capture_progress_events(monkeypatch, module)
+    job = build_job()
+
+    saved = module._complete_job_step(job, "Cosmos DB export step completed", 1, 4, "cosmos")
+
+    assert len(recorded) == 1
+    assert recorded[0]["step_name"] == "cosmos"
+    assert recorded[0]["status"] == "completed"
+    assert recorded[0]["job_status"] == "running", "Completing a step must not end the job."
+    assert saved["status"] == "running"
+    assert saved["progress"]["completed_steps"] == 1
+
+
+def test_in_progress_step_still_reports_running(monkeypatch):
+    """Keep the running badge for steps that have only started."""
+    module = load_data_management_module(monkeypatch)
+    recorded = capture_progress_events(monkeypatch, module)
+    job = build_job()
+
+    module._set_job_progress(job, "Migrating Cosmos records", 4, 10, current_step="cosmos")
+
+    assert recorded[0]["status"] == "running"
+    assert recorded[0]["job_status"] == "running"
+
+
+def test_terminal_job_progress_stamps_terminal_step_status(monkeypatch):
+    """Let an explicit terminal job status flow through to the final event."""
+    module = load_data_management_module(monkeypatch)
+    recorded = capture_progress_events(monkeypatch, module)
+    job = build_job()
+
+    module._set_job_progress(
+        job,
+        "Restore completed",
+        4,
+        4,
+        current_step="completed",
+        status=module.DATA_MANAGEMENT_STATUS_COMPLETED_WITH_WARNINGS,
+    )
+
+    assert recorded[0]["status"] == "completed_with_warnings"
+    assert recorded[0]["job_status"] == "completed_with_warnings"
+
+
+@pytest.mark.parametrize("message", [
+    "Cosmos DB export step completed",
+    "AI Search export step completed",
+    "Source blob export step completed",
+    "Cosmos restore step completed",
+    "AI Search restore step completed",
+    "Source blob restore step completed",
+    "Cosmos migration completed",
+    "AI Search migration completed",
+    "Source blob migration completed",
+    "Migration inventory completed",
+    "Migration reconciliation completed",
+    "Destination migration preflight completed",
+])
+def test_step_completions_use_the_completed_step_helper(message):
+    """Keep every step-completion call site on the completed-status helper."""
+    source = MODULE_PATH.read_text(encoding="utf-8")
+    call_index = source.index(f'"{message}"')
+    call_start = source.rindex("\n", 0, source.rindex("(", 0, call_index))
+    call = source[call_start:call_index]
+    assert "_complete_job_step" in call, (
+        f"'{message}' must be recorded with _complete_job_step, got: {call.strip()}"
+    )
+
+
+def test_migration_outcome_events_are_not_stamped_running():
+    """Stop labeling finished migration outcomes with the running job status."""
+    source = MODULE_PATH.read_text(encoding="utf-8")
+    outcome_events = [
+        '"migration-plan"',
+        '"migration-preflight"',
+        '"migration-reconciliation"',
+        'f"migration-cosmos-{target_type}"',
+    ]
+    for event in outcome_events:
+        event_index = source.index(event)
+        block = source[event_index:event_index + 260]
+        assert "DATA_MANAGEMENT_STATUS_COMPLETED" in block, (
+            f"{event} records finished work and must not use the running status."
+        )
+        assert "status=DATA_MANAGEMENT_STATUS_RUNNING" not in block
+
+
+def test_terminal_jobs_hide_live_current_container():
+    """Stop showing 'Current container: Waiting' once a job has finished."""
+    source = ADMIN_JS_PATH.read_text(encoding="utf-8")
+    assert "function isTerminalJobStatus(status)" in source
+    for status in ("completed", "completed_with_warnings", "failed", "canceled"):
+        assert f'"{status}"' in source
+
+    metrics_index = source.index("function getBackupLiveMetrics(job)")
+    metrics_block = source[metrics_index:metrics_index + 1400]
+    guard_index = metrics_block.index("isTerminalJobStatus(job?.status)")
+    current_container_index = metrics_block.index('label: "Current container"')
+    assert guard_index < current_container_index, (
+        "Current container must be gated behind the terminal-status guard."
+    )
+    assert "if (!isTerminalJobStatus(job?.status))" in metrics_block
+
+
+def test_step_status_fix_ships_in_supported_version():
+    """Keep the fix traceable to the version that introduced it."""
+    assert_app_version_at_least("0.260.003")
+
+
+if __name__ == "__main__":
+    sys.exit(pytest.main([__file__, "-q"]))