fix(scheduler): survive stop/start races and self-heal a dead scheduler thread

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)
 
 

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]:

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]:

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.

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"