feat(019): discovery on its own thread + tuner heartbeat (honest stale alert)

Telemetry: discovery took 6-30s ON the single scheduler, freezing ingest/arm
so the trigger read stale bars — the real 'late edge detection'. Now discovery
runs on its own daemon thread with a dedicated DB connection; the watchlist it
mutates is a lock-guarded Watchlist (list-like drop-in, no torn reads); the
scheduler discover step auto-disables (single-writer). Loop is now 3 fast
threads (entries/exits/discovery) + slow scheduler for tolerant work.

Also: nightly_tune writes a heartbeat every run, so the boot alert distinguishes
'tuner not running' (real) from 'ran, no profitable ORB config, kept params'
(expected — the strategy has no edge). Kills the misleading Telegram alarm.
543 tests green.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

run_live_paper.py

@@ -56,6 +56,7 @@ from research.ingestion.news_enrichment import enrich_recent_news
 from config import OllamaConfig
 from research.multi_schema import open_research_db
 from runtime.exit_manager import LiveExitManager
+from runtime.watchlist import Watchlist
 from runtime.halt_guard import is_halt_gap
 from runtime.market_calendar import (
     days_to_next_session,
@@ -150,14 +151,33 @@ def _load_learned(now: datetime | None = None) -> dict:
     age = _learned_params_age_days(learned.get("as_of"), now)
     max_age = float(os.environ.get("LEARNED_PARAMS_MAX_AGE_DAYS", "3"))
     if age is not None and age > max_age:
-        msg = (f"learned_params is {age:.0f} days stale (as_of "
-               f"{str(learned.get('as_of'))[:10]}) — tuner not running?")
-        print(f"[boot] WARNING: {msg}", flush=True)
+        # Distinguish a REAL failure (tuner not running) from the EXPECTED case
+        # (tuner runs nightly but finds no profitable ORB config, so it keeps
+        # the old params — the common state when the strategy has no edge). The
+        # tuner writes a heartbeat every run regardless of outcome.
+        hb_age = None
         try:
-            from runtime.notifier import send_telegram
-            send_telegram(f"⚠️ {msg}")
-        except Exception:  # noqa: BLE001 — never let a note block boot
-            pass
+            hb_path = os.path.join(os.environ.get("DATA_DIR", "./data"), "tuner_heartbeat.json")
+            if os.path.exists(hb_path):
+                with open(hb_path) as fh:
+                    hb_age = _learned_params_age_days(json.load(fh).get("ran_at"), now)
+        except Exception:  # noqa: BLE001
+            hb_age = None
+        if hb_age is not None and hb_age <= max_age:
+            # tuner IS running — params are old only because nothing profitable
+            # was found. Informational, not an alarm.
+            msg = (f"tuner running (last {hb_age:.0f}d ago) but no profitable ORB "
+                   f"config found — params kept (as_of {str(learned.get('as_of'))[:10]})")
+            print(f"[boot] note: {msg}", flush=True)
+        else:
+            msg = (f"learned_params {age:.0f}d stale AND tuner heartbeat "
+                   f"{'missing' if hb_age is None else f'{hb_age:.0f}d old'} — tuner NOT running")
+            print(f"[boot] WARNING: {msg}", flush=True)
+            try:
+                from runtime.notifier import send_telegram
+                send_telegram(f"⚠️ {msg}")
+            except Exception:  # noqa: BLE001 — never let a note block boot
+                pass
     return learned
 
 
@@ -433,7 +453,7 @@ def build_runtime(args: argparse.Namespace) -> dict:
     )
 
     return {
-        "symbols": symbols,
+        "symbols": Watchlist(symbols),  # plan 019: thread-safe (discovery mutates on its own thread)
         "symbols_source": symbols_source,
         # per-symbol IEX->consolidated volume correction (research.ingestion.volume_ratio):
         # measured each morning from the delayed free SIP feed, applied at ingest so live
@@ -501,7 +521,7 @@ def main(argv: list[str] | None = None) -> int:
     args = parse_args(argv)
     rt = build_runtime(args)
     ollama_cfg = rt["ollama_cfg"]
-    symbols: list[str] = rt["symbols"]
+    symbols: Watchlist = rt["symbols"]
     client = rt["client"]
 
     print(f"[boot] session={rt['session_id']} db={_db_target_desc()}")
@@ -522,8 +542,7 @@ def main(argv: list[str] | None = None) -> int:
             provider = rt["watcher"].provider
             if hasattr(provider, "add_symbol"):
                 provider.add_symbol(symbol)
-            if symbol not in symbols:
-                symbols.append(symbol)
+            symbols.append(symbol)  # Watchlist dedupes
 
         state = DashboardState(
             rt["event_db"],
@@ -549,10 +568,10 @@ def main(argv: list[str] | None = None) -> int:
             )
             if discovered:
                 print(f"[boot] screener added: {', '.join(discovered)}")
-                symbols = sorted(set(symbols) | set(discovered))
+                symbols.replace(sorted(set(symbols.snapshot()) | set(discovered)))
             else:
                 print("[boot] screener unavailable — using static symbol list")
-        daily = ingest_daily_history(rt["research_con"], client, symbols, days=30)
+        daily = ingest_daily_history(rt["research_con"], client, list(symbols), days=30)
         print(f"[boot] daily backfill: {daily.daily_rows} rows"
               + (f" (errors: {daily.errors})" if daily.errors else ""))
 
@@ -794,11 +813,11 @@ def main(argv: list[str] | None = None) -> int:
         done = command_consumer.poll()
         return f"operator commands executed: {len(done)}" if done else None
 
-    def step_discover():
+    def step_discover(con=None):
         if not client:
             return "skipped (no keys)"
         res = run_discovery(
-            rt["store"], rt["research_con"], client, _now_session_date(),
+            rt["store"], con or rt["research_con"], client, _now_session_date(),
             mode=EventMode.PAPER, correlation_id=rt["session_id"],
             price_min=float(os.environ.get("WATCHER_PRICE_MIN", "1")),
             price_max=float(os.environ.get("WATCHER_PRICE_MAX", "20")),
@@ -806,14 +825,14 @@ def main(argv: list[str] | None = None) -> int:
         )
         # fold newly-screened names into the live watchlist so the next ingest
         # pulls their bars and the watcher evaluates them
-        added = [s for s in res.universe if s not in symbols]
-        symbols.extend(added)
+        before = set(symbols.snapshot())
+        symbols.extend(res.universe)
+        added = [s for s in res.universe if s not in before]
         # Keep the watchlist BOUNDED so it rotates toward current movers instead
         # of growing all session. (Ingestion is now batched so an over-cap list
         # no longer 400s, but an unbounded list is still wasteful and stale.)
         max_uni = int(os.environ.get("WATCHER_MAX_UNIVERSE", "80"))
-        if len(symbols) > max_uni:
-            del symbols[: len(symbols) - max_uni]
+        symbols.trim(max_uni)
         return (f"universe={len(res.universe)} gappers={len(res.gappers)}"
                 + (f" +{len(added)} new" if added else ""))
 
@@ -1277,9 +1296,14 @@ def main(argv: list[str] | None = None) -> int:
     scheduler = Scheduler()
     scheduler.add("news", step_news, float(os.environ.get("ALPACA_NEWS_INTERVAL_SECONDS", "120")),
                   enabled=_flag("ALPACA_NEWS_ENABLED", "1"))
+    # SINGLE-WRITER (plan 019): the scheduler discover step runs ONLY when the
+    # dedicated discovery thread is OFF. Discovery takes 6-30s (telemetry) and
+    # on the scheduler it froze ingest/arm -> the trigger read stale bars. The
+    # thread moves that block off the time-critical path. Thread default ON.
     scheduler.add("discover", step_discover,
                   float(os.environ.get("DISCOVER_INTERVAL_SECONDS", "300")),
-                  enabled=_flag("DISCOVER_ENABLED", "1"))
+                  enabled=(_flag("DISCOVER_ENABLED", "1")
+                           and not _flag("DISCOVER_THREAD_ENABLED", "1")))
     scheduler.add("vratio", step_vratio, float(os.environ.get("VOLUME_RATIO_INTERVAL_SECONDS", "300")),
                   enabled=_flag("VOLUME_RATIO_ENABLED", "1"))
     scheduler.add("ingest", step_ingest, float(os.environ.get("LIVE_BARS_INTERVAL_SECONDS", "60")),
@@ -1369,6 +1393,28 @@ def main(argv: list[str] | None = None) -> int:
 
         threading.Thread(target=exit_loop, name="exits", daemon=True).start()
         print(f"[boot] exit thread started (every {_exit_interval:.1f}s) — scheduler exits step OFF")
+    # discovery on its OWN thread (plan 019): its 6-30s pass no longer blocks
+    # ingest/arm, so the trigger thread never reads stale bars. Own DB
+    # connection (shared facade isn't thread-safe); the watchlist it mutates is
+    # a thread-safe Watchlist; scheduler discover step disabled when this runs.
+    if _flag("DISCOVER_THREAD_ENABLED", "1") and _flag("DISCOVER_ENABLED", "1") and client is not None:
+        from runtime.exit_thread import run_exit_loop as _run_interval_loop  # generic driver
+        _disc_interval = float(os.environ.get("DISCOVER_INTERVAL_SECONDS", "300"))
+        _disc_con: dict = {"c": None}
+
+        def discover_loop():
+            try:
+                _disc_con["c"] = open_research_db("market")
+            except Exception as exc:  # noqa: BLE001
+                print(f"[discover-thread] dedicated DB unavailable, using shared: {exc}", flush=True)
+
+            class _DiscStep:
+                def manage(_self):        # adapt to the generic loop driver
+                    step_discover(con=_disc_con["c"])
+            _run_interval_loop(_DiscStep(), stop_event, _disc_interval)
+
+        threading.Thread(target=discover_loop, name="discover", daemon=True).start()
+        print(f"[boot] discovery thread started (every {_disc_interval:.0f}s) — scheduler discover step OFF")
     # news enrichment on its OWN thread (never blocks the scheduler's trade steps)
     if ollama_cfg.enabled and ollama_cfg.enrichment_enabled:
         threading.Thread(target=enrich_loop, name="enrich", daemon=True).start()