diff --git a/.github/workflows/harness-image.yml b/.github/workflows/harness-image.yml
index 126233dc..24050e57 100644
--- a/.github/workflows/harness-image.yml
+++ b/.github/workflows/harness-image.yml
@@ -29,6 +29,8 @@ jobs:
     permissions:
       contents: read
       packages: write
+    outputs:
+      image-tag: ${{ steps.vars.outputs.image-tag }}
     steps:
       - name: Checkout
         uses: actions/checkout@v4
@@ -53,6 +55,10 @@ jobs:
           BRANCH="${{ github.ref_name }}"
           CLEANED_BRANCH_NAME=$(echo "$BRANCH" | tr '/' '-' | tr '[:upper:]' '[:lower:]')
           echo "cleaned-branch-name=$CLEANED_BRANCH_NAME" >> "$GITHUB_OUTPUT"
+          # The tag the downstream deploy should pull. dispatch-deploy only runs
+          # on workflow_dispatch, so this always matches the branch image pushed
+          # below (<branch>-latest), never the unrelated release tag.
+          echo "image-tag=${CLEANED_BRANCH_NAME}-latest" >> "$GITHUB_OUTPUT"
 
       - name: Docker metadata
         id: meta
@@ -92,4 +98,4 @@ jobs:
           repository: conductor-oss/oss-ci-util
           event-type: sdk_release
           client-payload: |-
-            {"tag": "${{ github.event.release.tag_name || 'latest' }}", "repo": "${{ github.repository }}"}
+            {"tag": "${{ needs.build-and-push.outputs.image-tag }}", "repo": "${{ github.repository }}"}
diff --git a/METRICS.md b/METRICS.md
index 9e7f9770..00f8820d 100644
--- a/METRICS.md
+++ b/METRICS.md
@@ -68,6 +68,27 @@ or `clean_dead_pids=True` to remove only files from PIDs that no longer exist.
 Both default to `False`. Use a dedicated metrics directory per worker process
 group.
 
+Cleanup is owned by the parent process and performed by
+`MetricsSettings.clean_metrics_directory()`, which is idempotent per process
+(the destructive step runs at most once per directory). Spawned workers only
+ensure the directory exists (via `create_metrics_collector`) and never delete
+`.db` files, so a newly started or restarted worker can never wipe metrics
+belonging to live sibling processes.
+
+`TaskHandler.__init__` calls `clean_metrics_directory()` for you before any
+worker is spawned, which is sufficient for worker-only apps. If the parent
+process *also* collects metrics before constructing the `TaskHandler` (for
+example, registering task/workflow definitions or starting workflows through an
+instrumented client), build the parent's collector with
+`create_metrics_collector_for_parent(metrics_settings)` instead of
+`create_metrics_collector`. The parent factory cleans the directory up front --
+before the collector's first write -- and then creates the collector. Because
+`clean_metrics_directory()` is idempotent, `TaskHandler`'s later call is then a
+no-op and cannot orphan the parent's own `.db` files. For a long-lived parent
+that shares the directory, prefer `clean_dead_pids=True` over
+`clean_directory=True` -- it never deletes a live process's file regardless of
+ordering.
+
 ## Selecting Canonical Metrics
 
 Set `WORKER_CANONICAL_METRICS` before the worker starts:
@@ -290,7 +311,13 @@ sum(rate(task_execute_time_seconds_count[5m])) by (taskType)
   implementations never mixes stale metric names.
 - Pass `clean_dead_pids=True` to `MetricsSettings` to remove `.db` files from
   PIDs that no longer exist.  Use `clean_directory=True` only when you are sure
-  no other live process shares the same directory.
+  no other live process shares the same directory. Cleanup is idempotent per
+  process and runs in the parent (via `clean_metrics_directory()`, which
+  `TaskHandler.__init__` calls) before workers spawn; workers never wipe the
+  directory, so restarts and newly spawned workers preserve sibling metrics. If
+  the parent also collects, build its collector with
+  `create_metrics_collector_for_parent()` (cleans up front, then creates) so it
+  doesn't orphan the parent's own files.
 - Restart workers after changing `WORKER_CANONICAL_METRICS`.
 
 ### High Cardinality
@@ -342,8 +369,15 @@ unreleased metrics harmonization work. For a summary, see the project
     across all processes (main, workers, MetricsProvider).
   - `MetricsSettings` gains `clean_directory` (default `False`) to wipe all
     `.db` files and `clean_dead_pids` (default `False`) to remove only `.db`
-    files from PIDs that no longer exist. Both are executed by the factory
-    against the resolved metrics directory.
+    files from PIDs that no longer exist. Cleanup is applied by
+    `MetricsSettings.clean_metrics_directory()`, which is idempotent per process
+    and invoked by the parent (`TaskHandler.__init__` calls it, or a
+    parent that also collects builds its collector via
+    `create_metrics_collector_for_parent()`, which cleans up front then creates)
+    before workers spawn.
+    `create_metrics_collector` (the per-worker path) is non-destructive and
+    only ensures the directory exists, so spawned/restarted workers never wipe
+    live sibling metrics.
   - `CONDUCTOR_MP_START_METHOD` env var (`spawn` / `fork` / `forkserver`;
     default `fork` on POSIX, `spawn` on Windows) to control the worker pool's
     multiprocessing start method (motivated by a `prometheus_client` lock-fork
diff --git a/harness/main.py b/harness/main.py
index 417ef7b6..c7e38b76 100644
--- a/harness/main.py
+++ b/harness/main.py
@@ -9,7 +9,7 @@
 from conductor.client.configuration.settings.metrics_settings import MetricsSettings
 from conductor.client.http.models.task_def import TaskDef
 from conductor.client.orkes_clients import OrkesClients
-from conductor.client.telemetry.metrics_factory import create_metrics_collector
+from conductor.client.telemetry.metrics_factory import create_metrics_collector_for_parent
 from conductor.client.workflow.conductor_workflow import ConductorWorkflow
 from conductor.client.workflow.task.simple_task import SimpleTask
 
@@ -77,7 +77,12 @@ def main() -> None:
         clean_dead_pids=True,
     )
 
-    metrics_collector = create_metrics_collector(metrics_settings)
+    # The parent collects metrics too (task/workflow registration below and the
+    # WorkflowGovernor), so use the parent factory: it cleans the directory up
+    # front -- before the collector's first write -- then creates the collector,
+    # avoiding orphaning the parent's own .db files. clean_metrics_directory() is
+    # idempotent per process, so TaskHandler's later call is a no-op.
+    metrics_collector = create_metrics_collector_for_parent(metrics_settings)
     print(f"Prometheus metrics server started on port {metrics_port} ({metrics_collector.collector_name()} metrics)")
     clients = OrkesClients(configuration, metrics_collector=metrics_collector)
 
diff --git a/src/conductor/client/automator/task_handler.py b/src/conductor/client/automator/task_handler.py
index 6d24b516..c4bc3e28 100644
--- a/src/conductor/client/automator/task_handler.py
+++ b/src/conductor/client/automator/task_handler.py
@@ -251,6 +251,14 @@ def __init__(
         if metrics_settings is not None:
             os.environ["PROMETHEUS_MULTIPROC_DIR"] = metrics_settings.metrics_directory
             logger.info(f"Set PROMETHEUS_MULTIPROC_DIR={metrics_settings.metrics_directory}")
+            # Clean the shared metrics directory in the parent, before any
+            # worker process is spawned. This is a convenience for worker-only
+            # apps; clean_metrics_directory() is idempotent per process, so if
+            # the entrypoint already cleaned up front (recommended when the
+            # parent itself collects metrics), this call is a no-op and will not
+            # wipe .db files the parent already began writing. Spawned workers
+            # call create_metrics_collector (non-destructive) and never clean.
+            metrics_settings.clean_metrics_directory()
 
         # Store event listeners to pass to each worker process
         self.event_listeners = event_listeners or []
diff --git a/src/conductor/client/configuration/settings/metrics_settings.py b/src/conductor/client/configuration/settings/metrics_settings.py
index cdab29c3..b8705119 100644
--- a/src/conductor/client/configuration/settings/metrics_settings.py
+++ b/src/conductor/client/configuration/settings/metrics_settings.py
@@ -17,6 +17,20 @@
 CANONICAL_SUBDIR = "canonical"
 
 
+# Directories already cleaned in THIS process. Cleanup must run at most once
+# per directory per process so that a second caller (e.g. TaskHandler after the
+# owning entrypoint already cleaned up front) cannot wipe .db files that a live
+# process started writing after the first clean. Safe under spawn because only
+# the parent ever cleans; spawned workers never call clean_metrics_directory.
+_cleaned_metrics_directories: set = set()
+
+
+def _reset_cleaned_metrics_directories() -> None:
+    """Clear the per-process cleanup guard. Intended for tests so each starts
+    from a known state and does not leak entries via this module-level global."""
+    _cleaned_metrics_directories.clear()
+
+
 def _env_bool(name: str, default: bool) -> bool:
     value = os.environ.get(name, "")
     if not value:
@@ -100,6 +114,39 @@ def metrics_directory(self) -> str:
             return os.path.join(self.directory, self._subdir)
         return self.directory
 
+    def clean_metrics_directory(self) -> None:
+        """Prepare the shared metrics directory, at most once per process.
+
+        This is the destructive counterpart to metrics collection and must be
+        invoked only by the process that owns the metrics lifecycle, and as
+        early as possible -- before that process (or any worker it spawns)
+        creates a collector and starts writing ``.db`` files. It must never be
+        called by a spawned worker: a worker cannot know whether sibling
+        processes are already live and sharing this directory, so it must never
+        wipe ``.db`` files.
+
+        Idempotent per process: the destructive cleanup runs only on the first
+        call for a given directory. Later calls (e.g. ``TaskHandler`` invoking
+        this after the entrypoint already cleaned up front) only ensure the
+        directory exists, so they cannot wipe files a live process began
+        writing after the first clean.
+
+        On the first call, ensures the directory exists and applies the
+        configured cleanup:
+          - ``clean_directory``: remove all prometheus_client ``.db`` files.
+          - ``clean_dead_pids``: remove only ``.db`` files whose owning PID no
+            longer exists.
+        Both are no-ops when their respective flag is ``False``.
+        """
+        os.makedirs(self.metrics_directory, exist_ok=True)
+        if self.metrics_directory in _cleaned_metrics_directories:
+            return
+        _cleaned_metrics_directories.add(self.metrics_directory)
+        if self.clean_directory:
+            self._clean_stale_db_files()
+        if self.clean_dead_pids:
+            self._clean_dead_pid_files()
+
     def __set_dir(self, dir: str) -> None:
         if not os.path.isdir(dir):
             try:
diff --git a/src/conductor/client/telemetry/metrics_factory.py b/src/conductor/client/telemetry/metrics_factory.py
index 8523a201..0452e9c6 100644
--- a/src/conductor/client/telemetry/metrics_factory.py
+++ b/src/conductor/client/telemetry/metrics_factory.py
@@ -22,16 +22,6 @@
 )
 
 
-_cleaned_directories: set = set()
-
-
-def _reset_cleaned_directories() -> None:
-    """Clear the per-process cleanup guard.  Intended for tests so that each
-    test starts from a known-clean state and does not inherit or leak entries
-    via this module-level global."""
-    _cleaned_directories.clear()
-
-
 def create_metrics_collector(settings: MetricsSettings) -> MetricsCollectorBase:
     """
     Create the metrics collector indicated by ``settings.is_canonical``.
@@ -41,17 +31,17 @@ def create_metrics_collector(settings: MetricsSettings) -> MetricsCollectorBase:
 
     Returns a fully-initialised collector (legacy or canonical) that satisfies
     the MetricsCollector Protocol and can be registered as an event listener.
+
+    This is non-destructive: it only ensures the directory exists.  It never
+    deletes ``.db`` files, because it runs in every spawned worker and a worker
+    must not wipe metrics belonging to live sibling processes.  Directory
+    cleanup is owned by the parent via
+    ``MetricsSettings.clean_metrics_directory()`` (invoked once by
+    ``TaskHandler`` before workers spawn).
     """
     metrics_dir = settings.metrics_directory
     os.makedirs(metrics_dir, exist_ok=True)
 
-    if metrics_dir not in _cleaned_directories:
-        _cleaned_directories.add(metrics_dir)
-        if settings.clean_directory:
-            settings._clean_stale_db_files()
-        if settings.clean_dead_pids:
-            settings._clean_dead_pid_files()
-
     if settings.is_canonical:
         from conductor.client.telemetry.canonical_metrics_collector import CanonicalMetricsCollector
         logger.info("WORKER_CANONICAL_METRICS is true — using CanonicalMetricsCollector (dir=%s)", metrics_dir)
@@ -60,3 +50,22 @@ def create_metrics_collector(settings: MetricsSettings) -> MetricsCollectorBase:
     from conductor.client.telemetry.legacy_metrics_collector import LegacyMetricsCollector
     logger.info("Using LegacyMetricsCollector (dir=%s; set WORKER_CANONICAL_METRICS=true for canonical)", metrics_dir)
     return LegacyMetricsCollector(settings)
+
+
+def create_metrics_collector_for_parent(settings: MetricsSettings) -> MetricsCollectorBase:
+    """Owner/parent entrypoint: prepare the shared metrics directory, then build
+    the collector.
+
+    Use this from the process that owns the metrics lifecycle (the one that
+    spawns workers, and that may itself collect metrics before spawning). It
+    cleans the directory up front -- honoring ``settings.clean_directory`` /
+    ``settings.clean_dead_pids`` -- so the parent's own first write is not
+    orphaned, then delegates to ``create_metrics_collector``.
+
+    Do NOT call this from a spawned worker: workers must use
+    ``create_metrics_collector`` (non-destructive) so they can never wipe ``.db``
+    files belonging to live sibling processes. ``clean_metrics_directory()`` is
+    idempotent per process, so a later ``TaskHandler`` call is a no-op.
+    """
+    settings.clean_metrics_directory()
+    return create_metrics_collector(settings)
diff --git a/tests/unit/automator/test_task_handler.py b/tests/unit/automator/test_task_handler.py
index 26dd26f7..51d50a7a 100644
--- a/tests/unit/automator/test_task_handler.py
+++ b/tests/unit/automator/test_task_handler.py
@@ -1,4 +1,5 @@
 import multiprocessing
+import tempfile
 import unittest
 from unittest.mock import Mock
 from unittest.mock import patch
@@ -6,6 +7,7 @@
 from conductor.client.automator.task_handler import TaskHandler
 from conductor.client.automator.task_runner import TaskRunner
 from conductor.client.configuration.configuration import Configuration
+from conductor.client.configuration.settings.metrics_settings import MetricsSettings
 from tests.unit.resources.workers import ClassWorker
 
 
@@ -41,6 +43,23 @@ def test_start_processes(self):
                         isinstance(process, multiprocessing.Process)
                     )
 
+    def test_metrics_directory_cleaned_once_in_parent_init(self):
+        """TaskHandler.__init__ (parent, pre-spawn) invokes
+        clean_metrics_directory exactly once. Spawned workers rely on the
+        non-destructive create_metrics_collector path and never clean, so this
+        parent-owned call is the single point where .db files are wiped."""
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            metrics_settings = MetricsSettings(directory=tmp_dir, clean_directory=True)
+            metrics_settings.clean_metrics_directory = Mock()
+
+            task_handler = TaskHandler(
+                configuration=Configuration(),
+                workers=[ClassWorker('task')],
+                metrics_settings=metrics_settings,
+            )
+            with task_handler:
+                metrics_settings.clean_metrics_directory.assert_called_once_with()
+
 
 def _get_valid_task_handler():
     return TaskHandler(
diff --git a/tests/unit/telemetry/test_metrics_factory.py b/tests/unit/telemetry/test_metrics_factory.py
index 1a324245..a5ad8435 100644
--- a/tests/unit/telemetry/test_metrics_factory.py
+++ b/tests/unit/telemetry/test_metrics_factory.py
@@ -8,10 +8,13 @@
 import unittest
 from unittest import mock
 
-from conductor.client.configuration.settings.metrics_settings import MetricsSettings
+from conductor.client.configuration.settings.metrics_settings import (
+    MetricsSettings,
+    _reset_cleaned_metrics_directories,
+)
 from conductor.client.telemetry.metrics_factory import (
     create_metrics_collector,
-    _reset_cleaned_directories,
+    create_metrics_collector_for_parent,
 )
 from conductor.client.telemetry.legacy_metrics_collector import LegacyMetricsCollector
 from conductor.client.telemetry.canonical_metrics_collector import CanonicalMetricsCollector
@@ -20,7 +23,7 @@
 class TestMetricsFactory(unittest.TestCase):
 
     def setUp(self):
-        _reset_cleaned_directories()
+        _reset_cleaned_metrics_directories()
         self._saved_env = {}
         for key in ("WORKER_CANONICAL_METRICS", "WORKER_LEGACY_METRICS"):
             self._saved_env[key] = os.environ.pop(key, None)
@@ -34,7 +37,7 @@ def tearDown(self):
                 os.environ[key] = val
         if os.path.exists(self.metrics_dir):
             shutil.rmtree(self.metrics_dir)
-        _reset_cleaned_directories()
+        _reset_cleaned_metrics_directories()
 
     def _make_settings(self, **kwargs):
         kwargs.setdefault("directory", self.metrics_dir)
@@ -137,53 +140,90 @@ def test_both_implementations_satisfy_same_interface(self):
                 f"CanonicalMetricsCollector missing method: {method_name}",
             )
 
-    def test_cleanup_runs_once_per_directory(self):
-        """clean_directory cleanup fires only on the first collector for a given
-        directory; a second collector on the same directory must skip it so that
-        live .db files written by existing workers are never wiped."""
-        settings = MetricsSettings(directory=self.metrics_dir, clean_directory=True)
+    def test_create_collector_is_non_destructive(self):
+        """create_metrics_collector runs in every spawned worker, so it must
+        NEVER delete .db files -- doing so would wipe live sibling metrics.
+        Even with clean_directory=True, the factory leaves existing files
+        untouched; cleanup is the parent's responsibility."""
+        db_path = os.path.join(self.metrics_dir, "gauge_livesum_12345.db")
+        with open(db_path, "w") as f:
+            f.write("fake")
 
-        with mock.patch.object(
-            MetricsSettings, "_clean_stale_db_files", autospec=True
-        ) as clean_spy:
-            create_metrics_collector(settings)
-            create_metrics_collector(settings)
+        settings = MetricsSettings(directory=self.metrics_dir, clean_directory=True)
+        create_metrics_collector(settings)
 
-        self.assertEqual(
-            clean_spy.call_count,
-            1,
-            "stale-db cleanup must run exactly once per directory per process",
+        self.assertTrue(
+            os.path.exists(db_path),
+            "create_metrics_collector must not delete .db files (worker path)",
         )
 
-    def test_cleanup_runs_again_after_reset(self):
-        """Resetting the per-process guard allows cleanup to run again, proving
-        the global state is what gates the skip (and that setUp/tearDown isolate
-        tests from each other)."""
-        settings = MetricsSettings(directory=self.metrics_dir, clean_directory=True)
+    def test_create_for_parent_cleans_then_returns_collector(self):
+        """create_metrics_collector_for_parent is the parent/owner entrypoint: it
+        cleans the directory up front (honoring clean_directory) and then returns
+        a valid collector, so the parent's own first write is not orphaned."""
+        db_path = os.path.join(self.metrics_dir, "gauge_livesum_12345.db")
+        with open(db_path, "w") as f:
+            f.write("fake")
 
-        with mock.patch.object(
-            MetricsSettings, "_clean_stale_db_files", autospec=True
-        ) as clean_spy:
-            create_metrics_collector(settings)
-            _reset_cleaned_directories()
-            create_metrics_collector(settings)
+        settings = MetricsSettings(directory=self.metrics_dir, clean_directory=True)
+        collector = create_metrics_collector_for_parent(settings)
 
-        self.assertEqual(clean_spy.call_count, 2)
+        self.assertFalse(
+            os.path.exists(db_path),
+            "create_metrics_collector_for_parent must clean .db files with clean_directory=True",
+        )
+        self.assertIsInstance(collector, LegacyMetricsCollector)
 
     def test_clean_directory_removes_db_files(self):
-        """clean_directory=True actually removes .db files from the metrics dir."""
+        """MetricsSettings.clean_metrics_directory() removes .db files when
+        clean_directory=True."""
         db_path = os.path.join(self.metrics_dir, "gauge_livesum_12345.db")
         with open(db_path, "w") as f:
             f.write("fake")
 
         settings = MetricsSettings(directory=self.metrics_dir, clean_directory=True)
-        create_metrics_collector(settings)
+        settings.clean_metrics_directory()
 
         self.assertFalse(
             os.path.exists(db_path),
             "clean_directory=True should remove existing .db files",
         )
 
+    def test_clean_metrics_directory_noop_without_flags(self):
+        """clean_metrics_directory() leaves .db files untouched when neither
+        clean_directory nor clean_dead_pids is set."""
+        db_path = os.path.join(self.metrics_dir, "gauge_livesum_12345.db")
+        with open(db_path, "w") as f:
+            f.write("fake")
+
+        settings = MetricsSettings(directory=self.metrics_dir)
+        settings.clean_metrics_directory()
+
+        self.assertTrue(
+            os.path.exists(db_path),
+            "clean_metrics_directory() must not delete files when no flag is set",
+        )
+
+    def test_clean_metrics_directory_idempotent_per_process(self):
+        """The destructive cleanup runs only on the first call per directory.
+        A second call must not wipe .db files a live process (e.g. the parent's
+        collector) started writing after the first clean -- this is what lets
+        the entrypoint clean up front and TaskHandler safely call it again."""
+        settings = MetricsSettings(directory=self.metrics_dir, clean_directory=True)
+        settings.clean_metrics_directory()
+
+        # Simulate a live process creating a .db file after the first clean.
+        live_db = os.path.join(self.metrics_dir, f"gauge_livesum_{os.getpid()}.db")
+        with open(live_db, "w") as f:
+            f.write("fake")
+
+        settings.clean_metrics_directory()
+
+        self.assertTrue(
+            os.path.exists(live_db),
+            "second clean_metrics_directory() call must be a no-op and keep the file",
+        )
+
     def test_clean_dead_pids_removes_dead_pid_file(self):
         """clean_dead_pids=True removes .db files whose PID no longer exists."""
         dead_pid = 2_000_000_000
@@ -192,7 +232,7 @@ def test_clean_dead_pids_removes_dead_pid_file(self):
             f.write("fake")
 
         settings = MetricsSettings(directory=self.metrics_dir, clean_dead_pids=True)
-        create_metrics_collector(settings)
+        settings.clean_metrics_directory()
 
         self.assertFalse(
             os.path.exists(db_path),
@@ -207,7 +247,7 @@ def test_clean_dead_pids_keeps_live_pid_file(self):
             f.write("fake")
 
         settings = MetricsSettings(directory=self.metrics_dir, clean_dead_pids=True)
-        create_metrics_collector(settings)
+        settings.clean_metrics_directory()
 
         self.assertTrue(
             os.path.exists(db_path),
@@ -223,7 +263,7 @@ def test_clean_dead_pids_keeps_file_on_permission_error(self, _mock_kill):
             f.write("fake")
 
         settings = MetricsSettings(directory=self.metrics_dir, clean_dead_pids=True)
-        create_metrics_collector(settings)
+        settings.clean_metrics_directory()
 
         self.assertTrue(
             os.path.exists(db_path),