diff --git a/apps/api/main.py b/apps/api/main.py
index 1f916268d..41dd9479f 100644
--- a/apps/api/main.py
+++ b/apps/api/main.py
@@ -839,6 +839,18 @@ async def _hourly_sweep_loop() -> None:
             _expire_stale_approvals()  # #798
         except Exception as exc:
             logger.warning("Approval expiry sweep error: %s", exc)
+        try:
+            # Coarse backstop for a dead or wedged scheduler thread. It only ever
+            # RE-starts a scheduler this process already started, so a web-role
+            # process never grows one here.
+            from scheduler import ensure_scheduler_running
+
+            # In a worker thread: a restart joins the outgoing generation for up
+            # to a few seconds and must not stall the event loop.
+            if await asyncio.to_thread(ensure_scheduler_running):
+                logger.warning("Scheduler watchdog restarted a dead or stale scheduler thread")
+        except Exception as exc:
+            logger.warning("Scheduler watchdog error: %s", exc)
         await asyncio.sleep(_SWEEP_INTERVAL_SECONDS)
 
 
diff --git a/apps/api/scheduler.py b/apps/api/scheduler.py
index a65a192fa..3e02daf80 100644
--- a/apps/api/scheduler.py
+++ b/apps/api/scheduler.py
@@ -175,11 +175,27 @@ def _warn_cronless_schedule_once(worker_id: str, trigger_id: str | None = None)
         )
 
 POLL_INTERVAL_SECONDS = 60  # check every minute
+# Stop event of the CURRENT scheduler generation. Every start_scheduler() binds a
+# FRESH event and hands it to the thread it launches by closure, so a stop aimed
+# at generation N can never terminate generation N+1. Before this, one shared
+# module-global event meant a stop() immediately followed by a start() (what the
+# cloud wrapper does when its advisory-lock connection blips) left the flag set,
+# start() early-returned because the old thread was still sleeping, and the old
+# thread then woke up, saw the flag and exited: no scheduler thread at all.
 _stop_event: threading.Event = threading.Event()
 _scheduler_thread: threading.Thread | None = None
-_scheduler_lock = threading.Lock()
+# Re-entrant so ensure_scheduler_running() can restart under the same lock it
+# uses to read the scheduler state, without a second thread slipping in between.
+_scheduler_lock = threading.RLock()
 _SCHEDULER_HEARTBEAT_STALE_AFTER_SECONDS = POLL_INTERVAL_SECONDS * 3.0
+# How long a start waits for a stopping generation to finish before it gives up
+# and launches a fresh one anyway. Bounded so a wedged tick cannot block boot.
+_SCHEDULER_STOP_JOIN_TIMEOUT_SECONDS = 5.0
 _scheduler_last_heartbeat_monotonic: float | None = None
+# True once this process started a scheduler. ensure_scheduler_running() only
+# ever RE-starts, so a web-role process that never ran one cannot grow a
+# scheduler thread by calling the watchdog.
+_scheduler_started_once = False
 SCHEDULE_MISSED_ERROR_CODE = "scheduler_missed"
 SCHEDULE_MISSED_ERROR = "Scheduled fire was missed or delayed by the scheduler."
 SPEND_CAP_ERROR_CODE = "spend_cap_exceeded"
@@ -1315,10 +1331,23 @@ def _tick() -> None:
             )
 
 
+def _safe_log(level: int, message: str, *args: Any, exc_info: bool = False) -> None:
+    """Log without ever letting a broken handler kill the scheduler thread.
+
+    A closed stdout pipe (seen after a container restart) raises inside
+    logging, and that exception used to propagate out of the poll loop and end
+    the scheduler for the lifetime of the process.
+    """
+    try:
+        logger.log(level, message, *args, exc_info=exc_info)
+    except Exception:
+        pass
+
+
 def _record_scheduler_heartbeat() -> None:
     global _scheduler_last_heartbeat_monotonic
     _scheduler_last_heartbeat_monotonic = time.monotonic()
-    logger.info("Scheduler heartbeat")
+    _safe_log(logging.INFO, "Scheduler heartbeat")
 
 
 def scheduler_heartbeat_status(*, now_monotonic: float | None = None) -> dict[str, Any]:
@@ -1340,6 +1369,11 @@ def scheduler_heartbeat_status(*, now_monotonic: float | None = None) -> dict[st
     return {
         "ok": running and not stale,
         "running": running,
+        # `running` is `alive and not stopping`, so a live thread that is winding
+        # down looks identical to no thread at all. A supervisor must be able to
+        # tell those apart: restarting or releasing a leader lock while a live
+        # thread may still be firing triggers double-fires runs. Additive.
+        "alive": alive,
         "thread": thread.name if thread is not None else None,
         "stopping": _stop_event.is_set(),
         "heartbeat_age_seconds": heartbeat_age,
@@ -1348,33 +1382,140 @@ def scheduler_heartbeat_status(*, now_monotonic: float | None = None) -> dict[st
     }
 
 
-def start_scheduler() -> None:
-    """Start the scheduler in a background daemon thread."""
-    global _scheduler_thread, _scheduler_last_heartbeat_monotonic
-    with _scheduler_lock:
-        if _scheduler_thread is not None and _scheduler_thread.is_alive():
-            return
-        _stop_event.clear()
-        _scheduler_last_heartbeat_monotonic = time.monotonic()
+def _start_scheduler_locked(*, replace_wedged: bool = True) -> bool:
+    """Launch a fresh scheduler generation. Caller must hold ``_scheduler_lock``.
+
+    ``replace_wedged=False`` refuses to launch a successor while the outgoing
+    generation is still alive after the bounded join, so an automatic caller can
+    never end up with two scheduler threads firing the same triggers. Returns
+    True when a new generation was launched.
+    """
+    global _scheduler_thread, _stop_event, _scheduler_last_heartbeat_monotonic
+    global _scheduler_started_once
 
-        def _loop() -> None:
-            logger.info("Scheduler started (poll interval: %ds)", POLL_INTERVAL_SECONDS)
-            while not _stop_event.is_set():
+    previous = _scheduler_thread
+    if previous is not None and previous.is_alive():
+        if not _stop_event.is_set():
+            # A healthy generation is already running: starting again is a no-op.
+            return False
+        # A stop is pending for that generation. Wait for it to finish instead of
+        # early-returning, otherwise the caller is left with a thread that is
+        # about to exit and no successor.
+        previous.join(timeout=_SCHEDULER_STOP_JOIN_TIMEOUT_SECONDS)
+        if previous.is_alive():
+            # Wedged inside a tick and unkillable from here: Python cannot stop a
+            # thread. Its stop flag stays set, so it exits as soon as that tick
+            # returns.
+            if not replace_wedged:
+                _safe_log(
+                    logging.ERROR,
+                    "Scheduler thread did not stop within %ss and is still alive; NOT starting a "
+                    "second generation, because two schedulers would fire the same triggers twice",
+                    _SCHEDULER_STOP_JOIN_TIMEOUT_SECONDS,
+                )
+                return False
+            _safe_log(
+                logging.WARNING,
+                "Previous scheduler thread did not stop within %ss; starting a new generation",
+                _SCHEDULER_STOP_JOIN_TIMEOUT_SECONDS,
+            )
+
+    stop_event = threading.Event()
+    _stop_event = stop_event
+    _scheduler_last_heartbeat_monotonic = time.monotonic()
+
+    def _loop() -> None:
+        _safe_log(logging.INFO, "Scheduler started (poll interval: %ds)", POLL_INTERVAL_SECONDS)
+        # Only `stop_event` ends this loop. Everything else, including logging
+        # and the sleep, is contained so an incidental exception cannot leave
+        # the deployment without a scheduler.
+        while not stop_event.is_set():
+            try:
                 _record_scheduler_heartbeat()
+                _tick()
+            except Exception as exc:
+                _safe_log(logging.ERROR, "Scheduler tick failed: %s", exc, exc_info=True)
+            try:
+                stop_event.wait(timeout=POLL_INTERVAL_SECONDS)
+            except Exception as exc:
+                _safe_log(logging.ERROR, "Scheduler sleep failed: %s", exc, exc_info=True)
                 try:
-                    _tick()
-                except Exception as exc:
-                    logger.exception("Scheduler tick failed: %s", exc)
-                _stop_event.wait(timeout=POLL_INTERVAL_SECONDS)
-            logger.info("Scheduler stopped")
+                    time.sleep(POLL_INTERVAL_SECONDS)
+                except Exception:
+                    pass
+        _safe_log(logging.INFO, "Scheduler stopped")
+
+    _scheduler_thread = threading.Thread(target=_loop, daemon=True, name="workeros-scheduler")
+    _scheduler_started_once = True
+    _scheduler_thread.start()
+    return True
+
+
+def start_scheduler() -> None:
+    """Start the scheduler in a background daemon thread.
 
-        _scheduler_thread = threading.Thread(target=_loop, daemon=True, name="workeros-scheduler")
-        _scheduler_thread.start()
+    Idempotent while a healthy generation is running. When the current
+    generation is stopping, it is joined (bounded) and replaced, so a
+    stop/start pair can never leave the process with no scheduler.
+    """
+    with _scheduler_lock:
+        _start_scheduler_locked()
 
 
 def stop_scheduler() -> None:
-    """Signal the scheduler to stop."""
-    _stop_event.set()
+    """Signal the current scheduler generation to stop."""
+    with _scheduler_lock:
+        _stop_event.set()
+
+
+def ensure_scheduler_running(*, now_monotonic: float | None = None) -> bool:
+    """Restart the scheduler when its thread is gone. Returns True if it did.
+
+    Three states, only one of which is recovered here:
+
+    - ok: a live generation with a fresh heartbeat. No-op.
+    - dead: no thread, a thread that is not alive, or one whose stop flag is set
+      and which exits within the bounded join. Recovered by launching a fresh
+      generation.
+    - wedged: alive, not stopping, heartbeat stale. Logged loudly and left
+      ALONE. Python cannot kill a thread, so starting a replacement would leave
+      two schedulers firing the same triggers and duplicating side effects
+      (emails, CRM writes), which is worse than the delay. A stale heartbeat is
+      also not proof of death: a recovery tick that works through many overdue
+      triggers legitimately runs long.
+
+    Safe to call repeatedly and from any thread: it holds ``_scheduler_lock``
+    for the whole check-and-restart, so it never spawns a second scheduler
+    thread. It only ever RE-starts, so a process that never started a scheduler
+    (WORKEROS_ROLE=web) stays without one.
+    """
+    with _scheduler_lock:
+        if not _scheduler_started_once:
+            return False
+        status = scheduler_heartbeat_status(now_monotonic=now_monotonic)
+        if status["ok"]:
+            return False
+        thread = _scheduler_thread
+        alive = bool(thread is not None and thread.is_alive())
+        if alive and not _stop_event.is_set():
+            _safe_log(
+                logging.ERROR,
+                "Scheduler thread is alive but its heartbeat is %.0fs old (stale after %.0fs); "
+                "leaving it alone, because replacing a live thread would run two schedulers",
+                status["heartbeat_age_seconds"] or 0.0,
+                _SCHEDULER_HEARTBEAT_STALE_AFTER_SECONDS,
+            )
+            return False
+        _safe_log(
+            logging.WARNING,
+            "Scheduler thread is %s; restarting it",
+            "stopping" if alive else "not running",
+        )
+        # Retire the outgoing generation first so it can never outlive its
+        # replacement, then start a fresh one under the same lock. A generation
+        # that refuses to die keeps the slot: no duplicate scheduler.
+        _stop_event.set()
+        return _start_scheduler_locked(replace_wedged=False)
 
 
 def scheduler_status() -> dict[str, Any]:
diff --git a/apps/api/services/health_ops.py b/apps/api/services/health_ops.py
index 800e78013..d62ed49dd 100644
--- a/apps/api/services/health_ops.py
+++ b/apps/api/services/health_ops.py
@@ -117,10 +117,28 @@ def _health_check_scheduler() -> Dict[str, Any]:
     if deploy != "local":
         return {"ok": True, "enabled": False, "deploy": deploy}
     try:
-        from scheduler import scheduler_status
-        return scheduler_status()
+        import scheduler
+        payload = dict(scheduler.scheduler_status())
     except Exception as exc:
         return {"ok": False, "error": str(exc)[:300]}
+    # Surface heartbeat staleness ADDITIVELY: a thread that is alive but wedged
+    # inside a tick still reports is_alive() True, so nothing here used to show
+    # that it stopped firing. `ok` deliberately keeps the scheduler_status()
+    # meaning, because a long recovery tick is not proof of death and flipping
+    # `ok` would change health semantics for every consumer. Leader-aware health
+    # is a separate follow-up. Read through getattr and never fail the whole
+    # check on it: several tests inject a minimal scheduler stub.
+    heartbeat_status = getattr(scheduler, "scheduler_heartbeat_status", None)
+    if heartbeat_status is None:
+        return payload
+    try:
+        heartbeat = heartbeat_status()
+    except Exception:
+        return payload
+    for key in ("alive", "heartbeat_age_seconds", "stale_after_seconds", "stale"):
+        if key in heartbeat:
+            payload[key] = heartbeat[key]
+    return payload
 
 
 def _run_health_checks() -> Dict[str, Any]:
diff --git a/apps/api/services/public_view.py b/apps/api/services/public_view.py
index aafdccca4..c0f8d4867 100644
--- a/apps/api/services/public_view.py
+++ b/apps/api/services/public_view.py
@@ -104,6 +104,21 @@ def _json_noindex(payload: Dict[str, Any], *, status_code: int = 200) -> JSONRes
     "skill_runtime_error": _RUNTIME_HEADLINE,
     "openai_call_failed": _RUNTIME_HEADLINE,
     "interrupted_by_restart": "This run was interrupted while the service restarted. Re-run the worker.",
+    # Platform faults. Without an entry here these fell back to the generic
+    # "This worker failed to run ... edit or re-run the worker", which blames the
+    # operator for OUR restart. Name the fault as ours and say what to do.
+    "scheduler_missed": (
+        "Floom's scheduler was delayed, so this scheduled run started later than its scheduled "
+        "time. The worker itself is fine and no action is needed."
+    ),
+    "executor_lost_mid_run": (
+        "Floom's execution service stopped while this run was in progress. This is a platform "
+        "fault, not a problem with the worker. Re-run it if the work did not complete."
+    ),
+    "run_claimed_without_dispatch": (
+        "Floom picked this run up but the platform stopped before the worker started. Nothing in "
+        "the worker ran. This is a platform fault; re-run it."
+    ),
     "context_mount_failed": _RUNTIME_HEADLINE,
     "mcp_connect_failed": _CONNECTION_HEADLINE,
     # Sandbox / timeout / resource.
diff --git a/apps/api/tests/test_platform_fault_operator_copy.py b/apps/api/tests/test_platform_fault_operator_copy.py
new file mode 100644
index 000000000..49d5afb2e
--- /dev/null
+++ b/apps/api/tests/test_platform_fault_operator_copy.py
@@ -0,0 +1,87 @@
+"""Platform faults must read as OUR fault, not as the operator's.
+
+Incident 2026-08-02: a customer's scheduled run was abandoned by a platform
+restart and the run card said "This worker failed to run. Check the run logs for
+details, then edit or re-run the worker." The worker was fine. The three codes
+that only ever mean "Floom stopped", scheduler_missed, executor_lost_mid_run and
+run_claimed_without_dispatch, had no headline entry and fell through to that
+generic operator copy.
+"""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+import pytest
+
+API_DIR = Path(__file__).resolve().parents[1]
+if str(API_DIR) not in sys.path:
+    sys.path.insert(0, str(API_DIR))
+
+from services.public_view import (  # noqa: E402
+    _OPERATOR_ERROR_GENERIC,
+    _operator_error_message,
+)
+
+PLATFORM_FAULT_CODES = [
+    "scheduler_missed",
+    "executor_lost_mid_run",
+    "run_claimed_without_dispatch",
+]
+
+
+@pytest.mark.parametrize("code", PLATFORM_FAULT_CODES)
+def test_platform_fault_codes_have_their_own_headline(code):
+    message = _operator_error_message(None, code)
+
+    assert message is not None
+    assert message != _OPERATOR_ERROR_GENERIC, f"{code} still falls back to the generic blame copy"
+    assert "Floom" in message, f"{code} must name the platform as the party at fault"
+
+
+def test_scheduler_missed_says_the_worker_is_fine():
+    message = _operator_error_message(None, "scheduler_missed")
+
+    assert message == (
+        "Floom's scheduler was delayed, so this scheduled run started later than its scheduled "
+        "time. The worker itself is fine and no action is needed."
+    )
+
+
+def test_executor_lost_mid_run_names_the_platform_fault():
+    message = _operator_error_message(None, "executor_lost_mid_run")
+
+    assert message == (
+        "Floom's execution service stopped while this run was in progress. This is a platform "
+        "fault, not a problem with the worker. Re-run it if the work did not complete."
+    )
+
+
+def test_run_claimed_without_dispatch_says_nothing_ran():
+    message = _operator_error_message(None, "run_claimed_without_dispatch")
+
+    assert message == (
+        "Floom picked this run up but the platform stopped before the worker started. Nothing in "
+        "the worker ran. This is a platform fault; re-run it."
+    )
+
+
+@pytest.mark.parametrize("code", PLATFORM_FAULT_CODES)
+def test_platform_fault_copy_wins_over_the_raw_internal_error(code):
+    """The raw text of these failures is internal jargon; the code decides."""
+    message = _operator_error_message(
+        "Scheduled fire was missed or delayed by the scheduler.", code
+    )
+
+    assert "Floom" in message
+    assert "edit or re-run the worker" not in message
+
+
+@pytest.mark.parametrize("code", PLATFORM_FAULT_CODES)
+def test_platform_fault_copy_never_blames_the_operator(code):
+    message = _operator_error_message(None, code)
+
+    assert "This worker failed to run" not in message
+    assert "edit or re-run the worker" not in message
+    assert "\u2014" not in message, "operator copy must not use em dashes"
diff --git a/apps/api/tests/test_scheduler_stop_start_race.py b/apps/api/tests/test_scheduler_stop_start_race.py
new file mode 100644
index 000000000..6f890897f
--- /dev/null
+++ b/apps/api/tests/test_scheduler_stop_start_race.py
@@ -0,0 +1,339 @@
+"""Regression: a stop() immediately followed by a start() must leave a scheduler.
+
+Incident 2026-08-02: the cloud wrapper calls stop_scheduler() and, milliseconds
+later, start_scheduler() again whenever its Postgres advisory-lock connection
+blips. With a single module-global stop event, start_scheduler() early-returned
+because the old thread was still sleeping in wait(POLL_INTERVAL_SECONDS), the
+stop flag stayed set, and the old thread then woke, saw the flag and exited.
+Result: no scheduler thread and a stop flag nothing ever cleared, so scheduled
+runs stopped firing until the process was restarted.
+"""
+
+from __future__ import annotations
+
+import importlib
+import sys
+import threading
+import time
+from pathlib import Path
+
+import pytest
+
+API_DIR = Path(__file__).resolve().parents[1]
+if str(API_DIR) not in sys.path:
+    sys.path.insert(0, str(API_DIR))
+
+
+def _fresh_scheduler():
+    """Import the real scheduler module.
+
+    Other tests in the suite replace ``sys.modules['scheduler']`` with a stub
+    namespace; pop it so we always load the genuine module.
+    """
+    sys.modules.pop("scheduler", None)
+    return importlib.import_module("scheduler")
+
+
+@pytest.fixture
+def scheduler():
+    module = _fresh_scheduler()
+    try:
+        yield module
+    finally:
+        module.stop_scheduler()
+        thread = module._scheduler_thread
+        if thread is not None:
+            thread.join(timeout=5)
+        module._scheduler_thread = None
+        module._scheduler_started_once = False
+
+
+class _Ticks:
+    """Counts scheduler ticks and lets a test wait for the next one."""
+
+    def __init__(self) -> None:
+        self.count = 0
+        self.seen = threading.Event()
+
+    def __call__(self) -> None:
+        self.count += 1
+        self.seen.set()
+
+    def wait_for_next(self, timeout: float = 5.0) -> bool:
+        self.seen.clear()
+        return self.seen.wait(timeout)
+
+    def wait_for_first(self, timeout: float = 5.0) -> bool:
+        return self.seen.wait(timeout)
+
+
+def test_start_after_stop_leaves_a_live_ticking_scheduler(scheduler, monkeypatch):
+    ticks = _Ticks()
+    monkeypatch.setattr(scheduler, "_tick", ticks)
+    # Long enough that the running generation is provably asleep in wait() when
+    # the stop/start pair lands, which is the exact race from the incident.
+    monkeypatch.setattr(scheduler, "POLL_INTERVAL_SECONDS", 30)
+
+    scheduler.start_scheduler()
+    assert ticks.wait_for_first(), "the scheduler must tick once after starting"
+    first_thread = scheduler._scheduler_thread
+    assert first_thread is not None and first_thread.is_alive()
+
+    ticks.seen.clear()
+    scheduler.stop_scheduler()
+    scheduler.start_scheduler()
+
+    assert ticks.wait_for_first(), "a fresh generation must tick after a stop/start race"
+    thread = scheduler._scheduler_thread
+    assert thread is not None
+    assert thread.is_alive(), "the stop/start race must not leave the process without a scheduler"
+    assert scheduler._stop_event.is_set() is False, "the new generation's stop flag must be clear"
+
+    status = scheduler.scheduler_heartbeat_status()
+    assert status["running"] is True
+    assert status["stopping"] is False
+
+
+def test_stop_only_stops_the_generation_it_targeted(scheduler, monkeypatch):
+    ticks = _Ticks()
+    monkeypatch.setattr(scheduler, "_tick", ticks)
+    monkeypatch.setattr(scheduler, "POLL_INTERVAL_SECONDS", 30)
+
+    scheduler.start_scheduler()
+    assert ticks.wait_for_first()
+    retired_stop_event = scheduler._stop_event
+    retired_thread = scheduler._scheduler_thread
+
+    scheduler.stop_scheduler()
+    scheduler.start_scheduler()
+    assert ticks.wait_for_first()
+
+    retired_thread.join(timeout=5)
+    assert retired_thread.is_alive() is False, "the stopped generation must exit"
+    assert retired_stop_event is not scheduler._stop_event, "each generation owns its stop event"
+    assert retired_stop_event.is_set() is True
+    assert scheduler._scheduler_thread.is_alive() is True
+
+
+def test_start_is_a_no_op_while_a_healthy_generation_runs(scheduler, monkeypatch):
+    ticks = _Ticks()
+    monkeypatch.setattr(scheduler, "_tick", ticks)
+    monkeypatch.setattr(scheduler, "POLL_INTERVAL_SECONDS", 30)
+
+    scheduler.start_scheduler()
+    assert ticks.wait_for_first()
+    thread = scheduler._scheduler_thread
+    stop_event = scheduler._stop_event
+
+    scheduler.start_scheduler()
+    scheduler.start_scheduler()
+
+    assert scheduler._scheduler_thread is thread, "a healthy scheduler must not be replaced"
+    assert scheduler._stop_event is stop_event
+
+
+def test_loop_survives_a_broken_logging_handler(scheduler, monkeypatch):
+    """A closed stdout pipe must not end the scheduler for the whole process."""
+    ticks = _Ticks()
+    monkeypatch.setattr(scheduler, "_tick", ticks)
+    monkeypatch.setattr(scheduler, "POLL_INTERVAL_SECONDS", 0.01)
+
+    def broken_log(*args, **kwargs):
+        raise ValueError("I/O operation on closed file")
+
+    monkeypatch.setattr(scheduler.logger, "log", broken_log)
+    monkeypatch.setattr(scheduler.logger, "info", broken_log)
+    monkeypatch.setattr(scheduler.logger, "error", broken_log)
+    monkeypatch.setattr(scheduler.logger, "exception", broken_log)
+
+    scheduler.start_scheduler()
+    assert ticks.wait_for_first()
+    assert ticks.wait_for_next(), "the loop must keep ticking with a broken log handler"
+    assert scheduler._scheduler_thread.is_alive() is True
+
+
+def test_ensure_scheduler_running_restarts_a_dead_thread(scheduler, monkeypatch):
+    ticks = _Ticks()
+    monkeypatch.setattr(scheduler, "_tick", ticks)
+    monkeypatch.setattr(scheduler, "POLL_INTERVAL_SECONDS", 30)
+
+    scheduler.start_scheduler()
+    assert ticks.wait_for_first()
+
+    # Kill the generation the way the incident did: stop it and let it exit,
+    # without anything starting a successor.
+    dead_thread = scheduler._scheduler_thread
+    scheduler.stop_scheduler()
+    dead_thread.join(timeout=5)
+    assert dead_thread.is_alive() is False
+
+    ticks.seen.clear()
+    assert scheduler.ensure_scheduler_running() is True
+    assert ticks.wait_for_first(), "the healed scheduler must tick"
+    healed = scheduler._scheduler_thread
+    assert healed is not dead_thread
+    assert healed.is_alive() is True
+
+
+def test_ensure_scheduler_running_is_idempotent(scheduler, monkeypatch):
+    ticks = _Ticks()
+    monkeypatch.setattr(scheduler, "_tick", ticks)
+    monkeypatch.setattr(scheduler, "POLL_INTERVAL_SECONDS", 30)
+
+    scheduler.start_scheduler()
+    assert ticks.wait_for_first()
+    thread = scheduler._scheduler_thread
+
+    before = _scheduler_thread_count()
+    assert scheduler.ensure_scheduler_running() is False
+    assert scheduler.ensure_scheduler_running() is False
+    assert scheduler.ensure_scheduler_running() is False
+
+    assert scheduler._scheduler_thread is thread, "a healthy scheduler must not be replaced"
+    assert _scheduler_thread_count() == before, "no duplicate scheduler threads"
+
+
+def test_ensure_scheduler_running_never_replaces_a_live_stale_thread(scheduler, monkeypatch):
+    """Python cannot kill a thread, so a wedged generation is left alone.
+
+    Replacing a live-but-stale scheduler would leave TWO schedulers firing the
+    same triggers and duplicating side effects, which is worse than the delay.
+    A stale heartbeat is also not proof of death: a recovery tick working
+    through many overdue triggers legitimately runs long.
+    """
+    ticks = _Ticks()
+    monkeypatch.setattr(scheduler, "_tick", ticks)
+    monkeypatch.setattr(scheduler, "POLL_INTERVAL_SECONDS", 30)
+
+    scheduler.start_scheduler()
+    assert ticks.wait_for_first()
+    wedged = scheduler._scheduler_thread
+
+    stale_now = (
+        scheduler._scheduler_last_heartbeat_monotonic
+        + scheduler._SCHEDULER_HEARTBEAT_STALE_AFTER_SECONDS
+        + 1.0
+    )
+    status = scheduler.scheduler_heartbeat_status(now_monotonic=stale_now)
+    assert status["ok"] is False and status["stale"] is True
+
+    assert scheduler.ensure_scheduler_running(now_monotonic=stale_now) is False
+    assert scheduler.ensure_scheduler_running(now_monotonic=stale_now) is False
+
+    assert scheduler._scheduler_thread is wedged
+    assert wedged.is_alive() is True
+    assert scheduler._stop_event.is_set() is False
+    assert _scheduler_thread_count() == 1, "a wedged scheduler must never be doubled"
+
+
+def test_ensure_scheduler_running_restarts_a_stopping_generation(scheduler, monkeypatch):
+    """A generation whose stop flag is set is dead, not wedged: replace it."""
+    ticks = _Ticks()
+    monkeypatch.setattr(scheduler, "_tick", ticks)
+    monkeypatch.setattr(scheduler, "POLL_INTERVAL_SECONDS", 30)
+
+    scheduler.start_scheduler()
+    assert ticks.wait_for_first()
+    outgoing = scheduler._scheduler_thread
+
+    scheduler.stop_scheduler()
+    ticks.seen.clear()
+
+    assert scheduler.ensure_scheduler_running() is True
+    assert ticks.wait_for_first()
+    outgoing.join(timeout=5)
+    assert outgoing.is_alive() is False
+    assert scheduler._scheduler_thread is not outgoing
+    assert scheduler._scheduler_thread.is_alive() is True
+    assert _scheduler_thread_count() == 1
+
+
+def test_ensure_scheduler_running_never_cold_starts_a_scheduler(scheduler):
+    """A web-role process that never started a scheduler must not grow one."""
+    assert scheduler._scheduler_started_once is False
+    assert scheduler.ensure_scheduler_running() is False
+    assert scheduler._scheduler_thread is None
+    assert _scheduler_thread_count() == 0
+
+
+def test_heartbeat_status_reports_alive_separately_from_running(scheduler, monkeypatch):
+    """`running` is `alive and not stopping`, so a winding-down thread needs its
+    own signal: a supervisor must not confuse it with "no thread at all"."""
+    ticks = _Ticks()
+    monkeypatch.setattr(scheduler, "_tick", ticks)
+    monkeypatch.setattr(scheduler, "POLL_INTERVAL_SECONDS", 30)
+
+    assert scheduler.scheduler_heartbeat_status()["alive"] is False
+
+    scheduler.start_scheduler()
+    assert ticks.wait_for_first()
+    running = scheduler.scheduler_heartbeat_status()
+    assert running["alive"] is True
+    assert running["running"] is True
+
+    thread = scheduler._scheduler_thread
+    scheduler.stop_scheduler()
+    thread.join(timeout=5)
+
+    stopped = scheduler.scheduler_heartbeat_status()
+    assert stopped["alive"] is False
+    assert stopped["running"] is False
+    assert stopped["stopping"] is True
+
+
+def test_health_surfaces_heartbeat_staleness_additively(scheduler, monkeypatch):
+    """/health gains the staleness fields without changing what `ok` means."""
+    monkeypatch.setenv("WORKEROS_DEPLOY", "local")
+    from services import health_ops
+
+    ticks = _Ticks()
+    monkeypatch.setattr(scheduler, "_tick", ticks)
+    monkeypatch.setattr(scheduler, "POLL_INTERVAL_SECONDS", 30)
+
+    scheduler.start_scheduler()
+    assert ticks.wait_for_first()
+
+    healthy = health_ops._health_check_scheduler()
+    # Backward compatible: still a superset of the scheduler_status() keys.
+    for key in ("ok", "running", "thread", "stopping"):
+        assert key in healthy
+    assert healthy["alive"] is True
+    assert healthy["ok"] is True
+    assert healthy["running"] is True
+    assert healthy["stale"] is False
+    assert healthy["heartbeat_age_seconds"] is not None
+
+    # The thread stays alive but stops ticking, which is what is_alive() alone
+    # could never see. It is now visible, and `ok` is deliberately unchanged so
+    # no consumer's health semantics shift in this PR.
+    scheduler._scheduler_last_heartbeat_monotonic = (
+        time.monotonic() - scheduler._SCHEDULER_HEARTBEAT_STALE_AFTER_SECONDS - 1.0
+    )
+
+    wedged = health_ops._health_check_scheduler()
+    assert wedged["stale"] is True
+    assert wedged["running"] is True
+    assert wedged["ok"] is True
+
+
+def test_health_still_reports_a_dead_scheduler_as_not_ok(scheduler, monkeypatch):
+    monkeypatch.setenv("WORKEROS_DEPLOY", "local")
+    from services import health_ops
+
+    ticks = _Ticks()
+    monkeypatch.setattr(scheduler, "_tick", ticks)
+    monkeypatch.setattr(scheduler, "POLL_INTERVAL_SECONDS", 30)
+
+    scheduler.start_scheduler()
+    assert ticks.wait_for_first()
+    thread = scheduler._scheduler_thread
+    scheduler.stop_scheduler()
+    thread.join(timeout=5)
+
+    dead = health_ops._health_check_scheduler()
+    assert dead["ok"] is False
+    assert dead["running"] is False
+
+
+def _scheduler_thread_count() -> int:
+    return len([t for t in threading.enumerate() if t.name == "workeros-scheduler" and t.is_alive()])