"""Live paper-trading orchestrator (Milestone 4 end-to-end).

This is the piece that was missing: a single loop tying together

  1. ingest    — latest 1-minute Alpaca bars into the market schema (Postgres)
                 (plus a daily-bar backfill on startup so gap %% / RVOL work)
  2. watch     — Watcher.tick over the ingested bars, emitting signal_ready
  3. sync      — Alpaca paper account / positions / orders into the event DB
  4. execute   — TradingExecutionService.tick: approval requests, and order
                 submission (auto or via dashboard approval)
  5. dashboard — optional embedded API + UI at http://127.0.0.1:8010

Usage
-----
    python run_live_paper.py                 # continuous loop
    python run_live_paper.py --once          # single pass, then exit
    python run_live_paper.py --symbols AAPL,SNDL --auto-approve
    python run_live_paper.py --no-dashboard

Without ALPACA_API_KEY / ALPACA_SECRET_KEY the loop runs in dry mode:
network steps are skipped with clear messages, but watcher + dashboard
still work against whatever data already exists.
"""

from __future__ import annotations

import argparse
import json
import os
import signal
import sys
import threading
import time
import uuid
from datetime import datetime, timezone

from dotenv import load_dotenv

from alpaca_paper.client import AlpacaPaperClient
from alpaca_paper.execution import AlpacaPaperExecutor
from alpaca_paper.settings import AlpacaPaperSettings
from alpaca_paper.sync import AlpacaPaperSync
from api.main import DashboardState, serve_in_background
from research.ingestion.market_data import (
    IngestionResult,
    classify_session,
    data_feed,
    discover_active_symbols,
    ingest_daily_history,
    ingest_live_minute_bars,
)
from research.ingestion.scheduler import Scheduler
from research.ingestion.watcher_task import ResearchWatchlistProvider
from research.ingestion.discovery import run_discovery, screen_universe
from research.ingestion.alpaca_news import ingest_alpaca_news
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,
    eod_flatten_status,
    is_regular_hours,
)
from runtime.triggers import ArmedTriggerBook
from runtime.watcher import Watcher, WatcherConfig
from strategy.evaluation.premarket_range import (
    PremarketRangeConfig, premarket_range_pct, premarket_rank_multiplier)
from strategy.population import PopulationConfig
from strategy.exits import ExitConfig
from storage.event_schema import EventMode, ModuleTickEvent
from storage.event_store import EventStore
from trading_execution import ExecutionSettings, TradingExecutionService
from trading_mode import TradingModeSettings

# Fallback watchlist used ONLY if the sub-$20 screener is unavailable. Liquid
# small-caps that usually sit in the $1-20 band — NOT mega-caps (which the
# watcher's price band would just filter out, leaving nothing to trade).
DEFAULT_SYMBOLS = [
    "SOFI", "GRAB", "NU", "AAL", "PLUG", "RIOT",
    "MARA", "SNDL", "ACHR", "NOK", "SOUN", "BBAI",
]


def _flag(name: str, default: str = "1") -> bool:
    return os.environ.get(name, default).strip().lower() in {"1", "true", "yes", "on"}


# The process is ONE trading session. A dozen memory-only daily flags (_eod_done,
# _session_closed, entry caps, benches...) are set once and never reset on date
# rollover, so an uninterrupted day-2 of uptime would take zero entries and never
# EOD-flatten. Instead of resetting them all, the loop EXITS cleanly after a
# successful EOD flatten (and on SIGTERM) — compose's `restart: unless-stopped`
# brings back a fresh process whose boot recovery (_rebuild_daily_state) restores
# any state that matters. Set from the scheduler thread and the signal handler;
# consumed by the main loop so the existing `finally:` cleanup runs.
_clean_exit_requested = threading.Event()


def _request_exit_after_eod(ok: bool) -> bool:
    """Request a clean process exit after a SUCCESSFUL EOD flatten.

    Gated on TRADING_EXIT_AFTER_EOD_FLATTEN (default on). A FAILED flatten must
    keep retrying inside the window, so ``ok=False`` never requests exit.
    Returns whether exit was requested."""
    if not ok or not _flag("TRADING_EXIT_AFTER_EOD_FLATTEN", "1"):
        return False
    _clean_exit_requested.set()
    return True


def _sigterm_handler(signum, frame):  # noqa: ARG001 — signal API shape
    """SIGTERM (docker stop) = clean shutdown: drain through the main loop's
    ``finally:`` (threads joined, DB connections closed) instead of dying mid-pass."""
    print("[stop] SIGTERM received — draining for clean shutdown", flush=True)
    _clean_exit_requested.set()


def _now_session_date():
    session_date, _, _, _ = classify_session(datetime.now(timezone.utc))
    return session_date


def _learned_params_age_days(as_of, now: datetime | None = None) -> float | None:
    """Age (days) of a learned_params `as_of` ISO timestamp, or None if
    missing/unparseable. Naive timestamps are treated as UTC."""
    if not as_of:
        return None
    try:
        ts = datetime.fromisoformat(str(as_of))
    except (TypeError, ValueError):
        return None
    now = now or datetime.now(timezone.utc)
    if ts.tzinfo is None:
        ts = ts.replace(tzinfo=timezone.utc)
    if now.tzinfo is None:
        now = now.replace(tzinfo=timezone.utc)
    return (now - ts).total_seconds() / 86400.0


def _load_learned(now: datetime | None = None) -> dict:
    """Nightly-tuned params from data/learned_params.json (empty if absent).

    Warns (boot log + one gated Telegram note) if the file is older than
    LEARNED_PARAMS_MAX_AGE_DAYS (default 3) — a stale file means the nightly
    tuner stopped running. NEVER blocks boot: params load regardless."""
    path = os.path.join(os.environ.get("DATA_DIR", "./data"), "learned_params.json")
    try:
        with open(path) as fh:
            learned = json.load(fh)
    except Exception:  # noqa: BLE001
        return {}
    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:
        # 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:
            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 None:
            # file may live on a different container's volume — the DB heartbeat
            # is the durable record (see nightly_tune._write_heartbeat)
            try:
                from research.multi_schema import open_research_db
                r = open_research_db("market").execute(
                    "SELECT metadata_json FROM telemetry_events "
                    "WHERE event_type='tuner_heartbeat' ORDER BY created_at DESC LIMIT 1"
                ).fetchone()
                if r and r[0]:
                    hb_age = _learned_params_age_days(
                        json.loads(r[0]).get("ran_at"), now)
            except Exception:  # noqa: BLE001
                pass
        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


def _db_target_desc() -> str:
    """Human-readable Postgres target for the boot log (password masked)."""
    url = os.environ.get("DATABASE_URL")
    if url:
        if "://" in url and "@" in url:
            scheme, rest = url.split("://", 1)
            creds, host = rest.split("@", 1)
            user = creds.split(":", 1)[0]
            return f"{scheme}://{user}:***@{host}"
        return url
    host = os.environ.get("PGHOST", "localhost")
    port = os.environ.get("PGPORT", "5432")
    db = os.environ.get("PGDATABASE", "momentum")
    user = os.environ.get("PGUSER", "")
    return f"postgresql://{user}@{host}:{port}/{db}"


def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Momentum live paper-trading loop")
    parser.add_argument("--once", action="store_true", help="run one pass and exit")
    parser.add_argument(
        "--symbols",
        default=os.environ.get("WATCHER_SYMBOLS", ""),
        help="comma-separated symbols to watch (default: WATCHER_SYMBOLS env or built-ins)",
    )
    parser.add_argument(
        "--auto-approve",
        action="store_true",
        help="auto-approve and submit ready signals (otherwise approve from the dashboard)",
    )
    parser.add_argument(
        "--discover",
        action="store_true",
        help="also pull symbols from Alpaca's most-actives screener",
    )
    parser.add_argument("--no-dashboard", action="store_true", help="skip the embedded dashboard")
    return parser.parse_args(argv)


def build_runtime(args: argparse.Namespace) -> dict:
    """Wire up every component; safe without API keys."""
    load_dotenv()
    from runtime.env_overrides import load as _load_overrides
    _load_overrides()  # admin-panel overrides beat compose env

    explicit_symbols = [s.strip().upper() for s in args.symbols.split(",") if s.strip()]

    alpaca_settings = AlpacaPaperSettings.from_env()
    has_keys = alpaca_settings.is_configured
    client = AlpacaPaperClient(alpaca_settings) if has_keys else None

    # Broker failover: BROKER=tradier swaps the EXECUTION client for the
    # Tradier mirror (same surface/shapes); market data stays on Alpaca via
    # delegation. Default (unset/alpaca) is untouched.
    if os.environ.get("BROKER", "alpaca").strip().lower() == "tradier":
        from tradier_broker.client import TradierClient
        from tradier_broker.settings import TradierSettings
        tradier_settings = TradierSettings.from_env()
        if tradier_settings.is_configured:
            client = TradierClient(tradier_settings, data_client=client)
            print(f"[boot] BROKER=tradier: executing via Tradier "
                  f"({'SANDBOX' if tradier_settings.is_sandbox else 'LIVE'}), "
                  f"data via Alpaca", flush=True)
        else:
            print("[boot] WARNING: BROKER=tradier but TRADIER_ACCESS_TOKEN/"
                  "TRADIER_ACCOUNT_ID missing — staying on Alpaca", flush=True)

    # Datastore is Postgres (DATABASE_URL); WATCHER_EVENT_DB_PATH is a legacy
    # fallback the connection layer ignores.
    event_db = os.environ.get("WATCHER_EVENT_DB_PATH", "momentum")
    store = EventStore(event_db)
    research_con = open_research_db("market")
    # NBBO quote-size snapshots for tracked/held names — the free proxy for order-book
    # depth imbalance (the trader's "big seller at $6" read). Recording-only; a shadow
    # study consumes it later. See step_quotes.
    research_con.execute(
        """CREATE TABLE IF NOT EXISTS quote_snapshots (
               symbol TEXT NOT NULL,
               ts TIMESTAMPTZ NOT NULL DEFAULT now(),
               bid_price DOUBLE PRECISION, bid_size DOUBLE PRECISION,
               ask_price DOUBLE PRECISION, ask_size DOUBLE PRECISION)""")
    research_con.execute(
        "CREATE INDEX IF NOT EXISTS quote_snaps_sym_ts ON quote_snapshots(symbol, ts)")

    session_id = f"paper-{_now_session_date().isoformat()}-{uuid.uuid4().hex[:6]}"

    # Watchlist resolution. Explicit --symbols/WATCHER_SYMBOLS win; otherwise
    # discover the sub-$20 most-actives universe via the screener (the strategy
    # targets $1-20 small-caps, not the hardcoded mega-cap fallback). Falls back
    # to DEFAULT_SYMBOLS only if the screener is unavailable.
    price_min = float(os.environ.get("WATCHER_PRICE_MIN", "1"))
    price_max = float(os.environ.get("WATCHER_PRICE_MAX", "20"))
    discover_top = int(os.environ.get("DISCOVER_TOP", "20"))
    if explicit_symbols:
        symbols = explicit_symbols
        symbols_source = "explicit"
    elif client is not None and _flag("DISCOVER_ON_START", "1"):
        discovered = screen_universe(
            client, price_min=price_min, price_max=price_max, top=discover_top
        )
        symbols = discovered or list(DEFAULT_SYMBOLS)
        symbols_source = "screener" if discovered else "fallback"
    else:
        symbols = list(DEFAULT_SYMBOLS)
        symbols_source = "fallback"

    store.emit(
        ModuleTickEvent(
            timestamp=datetime.now(),
            mode=EventMode.PAPER,
            correlation_id=session_id,
            message=f"watchlist resolved via {symbols_source}: {len(symbols)} symbols",
            module="discovery",
            stage="boot",
            duration_ms=0.0,
            input_count=0,
            output_count=len(symbols),
            metrics={"source": symbols_source, "symbols": symbols},
            errors=(
                []
                if symbols_source != "fallback"
                else [{"error": "screener returned 0 names at boot — using fallback "
                       "watchlist; check Alpaca screener / data plan"}]
            ),
        )
    )

    _learned = _load_learned()
    if _learned:
        print(f"[boot] learned params: ready_score={_learned.get('ready_score_pct')} "
              f"min_bars={_learned.get('min_bars')} setups={_learned.get('setups')} "
              f"(tuned {str(_learned.get('as_of', ''))[:10]} over {_learned.get('sessions')} "
              f"sessions; backtest pnl {_learned.get('pnl')}, entry@{_learned.get('entry_min')}min)")

    ollama_cfg = OllamaConfig.from_env()

    # Catalyst advisory provider, TTL-cached so a pass doesn't re-query per symbol;
    # returns None on any fault so neither the (gated) dilution veto nor the score
    # blend ever block/skew on a DB blip. Shared by the watcher AND the execution
    # service (one cache). catalyst_score_provider maps the advisory -> a 0..1 score
    # OUTSIDE the pure evaluator, and is only handed to the watcher when the score
    # blend is explicitly enabled (so Phase 1 advisory never changes scoring).
    _catalyst_cache: dict = {"map": {}, "at": 0.0, "con": None}
    _catalyst_lock = threading.Lock()
    _CATALYST_TTL = 30.0

    def catalyst_provider(symbol: str):
        import time as _time
        from research.ingestion.news_enrichment import catalyst_map
        now = _time.monotonic()
        # Called from the TRIGGER thread (submit_breakout_now veto), the scheduler
        # thread (approvals), AND the watcher — so it must NOT touch the shared
        # research_con. Refresh under a lock using a DEDICATED connection (opened
        # lazily, only inside the lock => single-threaded use); the 30s TTL keeps
        # DB hits rare, and the read below is a lock-free atomic dict get.
        if (now - _catalyst_cache["at"]) >= _CATALYST_TTL:
            with _catalyst_lock:
                if (_time.monotonic() - _catalyst_cache["at"]) >= _CATALYST_TTL:
                    try:
                        if _catalyst_cache["con"] is None:
                            _catalyst_cache["con"] = open_research_db("market")
                        _catalyst_cache["map"] = catalyst_map(_catalyst_cache["con"])
                    except Exception:  # noqa: BLE001
                        _catalyst_cache["map"] = {}
                    _catalyst_cache["at"] = _time.monotonic()
        return _catalyst_cache["map"].get((symbol or "").upper())

    def catalyst_score_provider(symbol: str):
        from research.ingestion.news_enrichment import catalyst_score
        return catalyst_score(catalyst_provider(symbol))

    # Decision-time bid/ask spread provider (observe-only). One per-symbol quote,
    # TTL-cached so a watch pass doesn't re-fetch, mapped to spread_pct. Returns
    # None on any fault / missing quote / when disabled, so the watcher logs the
    # metric without ever blocking or breaking on a quote blip. Default ON when a
    # client exists; set SPREAD_PCT_ENABLED=0 to silence the extra quote calls.
    _spread_cache: dict[str, tuple[float, float | None]] = {}
    _SPREAD_TTL = 15.0

    def _spread_enabled() -> bool:
        return os.environ.get("SPREAD_PCT_ENABLED", "1").strip().lower() in {
            "1", "true", "yes", "on"}

    def spread_provider(symbol: str):
        if client is None or not _spread_enabled():
            return None
        import time as _time
        from strategy.evaluation.liquidity import compute_spread_pct
        now = _time.monotonic()
        hit = _spread_cache.get(symbol)
        if hit is not None and (now - hit[0]) < _SPREAD_TTL:
            return hit[1]
        val = None
        try:
            quote = (client.get_latest_quotes([symbol]) or {}).get(symbol) or {}
            val = compute_spread_pct(quote.get("bp"), quote.get("ap"))
        except Exception:  # noqa: BLE001
            val = None
        _spread_cache[symbol] = (now, val)
        return val

    watcher = Watcher(
        store,
        ResearchWatchlistProvider(
            research_con,
            limit=int(os.environ.get("WATCHER_MAX_SYMBOLS", "25")),
        ),
        WatcherConfig(
            session_id=session_id,
            mode=EventMode.PAPER,
            max_symbols=int(os.environ.get("WATCHER_MAX_SYMBOLS", "25")),
            # Defaults come from the nightly self-tuner (learned_params.json);
            # an explicit env var still wins.
            min_bars=int(os.environ.get("WATCHER_MIN_BARS", str(_learned.get("min_bars", 10)))),
            ready_score_pct=float(os.environ.get(
                "WATCHER_READY_SCORE", str(_learned.get("ready_score_pct", 60.0)))),
            min_quality=float(os.environ.get("WATCHER_MIN_QUALITY", "0.30")),
            # Same population gate the entry path enforces, applied before the
            # watcher scores anything — so it stops writing criteria receipts for
            # names that can never trade. Opt out with WATCHER_POPULATION_GATE=0.
            population=(PopulationConfig.from_env()
                        if _flag("WATCHER_POPULATION_GATE", "1") else None),
        ),
        catalyst_score_provider=(
            catalyst_score_provider if ollama_cfg.catalyst_score_enabled else None
        ),
        spread_provider=spread_provider,
    )

    exec_settings = ExecutionSettings.from_env()
    if args.auto_approve:
        exec_settings.auto_approve = True
    executor = AlpacaPaperExecutor(store, client=client, session_id=session_id) if client else None

    # Live last-trade price source for entry invalidation — tick-by-tick, not
    # bar close. Pulls the most recent trade from Alpaca's data API on demand;
    # a tiny TTL cache collapses redundant lookups within a single guard pass.
    # On any failure it returns None so the price-break check is simply skipped
    # for that pass (the wall-clock timeout still protects the entry) rather
    # than silently falling back to a stale bar close.
    _price_cache: dict[str, tuple[float, float]] = {}
    _PRICE_TTL = 2.0

    def latest_price(symbol: str):
        if client is None:
            return None
        import time as _time

        cached = _price_cache.get(symbol)
        now = _time.monotonic()
        if cached and (now - cached[1]) < _PRICE_TTL:
            return cached[0]
        try:
            trades = client.get_latest_trades([symbol])
            trade = trades.get(symbol) or trades.get(symbol.upper())
            price = float(trade["p"]) if trade and trade.get("p") is not None else None
        except Exception:
            price = None
        if price is not None:
            _price_cache[symbol] = (price, now)
        return price

    execution = (
        TradingExecutionService(
            store,
            executor=executor,
            settings=exec_settings,
            trading_mode=TradingModeSettings.from_env(),
            session_id=session_id,
            price_provider=latest_price,
            catalyst_provider=catalyst_provider,
        )
        if executor
        else None
    )
    sync = AlpacaPaperSync(store, client=client, session_id=session_id) if client else None

    # Fast armed-trigger book: keep the N most-promising gappers ready with a
    # pre-computed opening-range trigger/stop so step_trigger can fire the
    # instant live price crosses (instead of waiting for a bar close + cadence).
    book = ArmedTriggerBook(
        max_armed=int(os.environ.get("TRIGGER_MAX_ARMED", "6")),
        gap_min=float(os.environ.get("TRIGGER_GAP_MIN", "3.0")),
        gap_max=float(os.environ.get("TRIGGER_GAP_MAX", "1e9")),
        # TRADING_GAP_SQUEEZE_ENABLED=0 turns the >=150% squeeze exception off here too, so
        # one switch covers the trigger book, the watcher pre-filter and the entry gate.
        gap_squeeze_min=(float(os.environ.get("TRIGGER_GAP_SQUEEZE_MIN", "0"))
                         if os.environ.get("TRADING_GAP_SQUEEZE_ENABLED", "1").strip().lower()
                         not in {"0", "false", "no", "off"} else 0.0),
        gap_squeeze_rvol=float(os.environ.get("TRIGGER_GAP_SQUEEZE_RVOL", "5.0")),
        rvol_min=float(os.environ.get("TRIGGER_RVOL_MIN", "2.0")),
        min_range_pct=float(os.environ.get("TRIGGER_MIN_RANGE_PCT", "0.004")),
        min_dollar_vol=float(os.environ.get("TRIGGER_MIN_DOLLAR_VOL", "0")),
        max_price_age_s=float(os.environ.get("TRIGGER_PRICE_MAX_AGE", "12")),
        fresh_cross_tol=float(os.environ.get("TRIGGER_FRESH_CROSS_TOL", "0.015")),
    )

    return {
        "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
        # RVOL/liquidity signals are consolidated-scale like their (nightly-SIP) baselines
        "vol_ratios": {},
        "client": client,
        "has_keys": has_keys,
        "store": store,
        "research_con": research_con,
        "watcher": watcher,
        "execution": execution,
        "sync": sync,
        "book": book,
        "latest_price": latest_price,
        "catalyst_provider": catalyst_provider,
        "ollama_cfg": ollama_cfg,
        "session_id": session_id,
        "event_db": event_db,
    }



_SESSION_OPEN_CACHE: dict = {}


def _session_open(con, symbol: str, session_date) -> float | None:
    """First regular-hours bar's open for symbol/day — the TRUE session open.

    query_minute_bars returns a recent WINDOW, so its first bar drifts intraday;
    this queries the actual 09:30 bar once and caches it (immutable once printed).
    """
    key = (symbol, str(session_date))
    v = _SESSION_OPEN_CACHE.get(key)
    if v is not None:
        return v
    try:
        r = con.execute(
            "SELECT open, timestamp FROM minute_bars WHERE symbol=? AND session_date=? "
            "AND is_regular_hours ORDER BY timestamp LIMIT 1", [symbol, session_date]
        ).fetchall()
        if r:
            # trust it ONLY if it is really the 09:30 ET opening bar: mid-day-
            # discovered names briefly have bars starting mid-session, and treating
            # that first bar as "the open" is how TVRD/FRTT armed as fake 20%
            # gappers on 2026-07-08. minute_bars.timestamp is UTC-naive, so the
            # open is 13:30Z in summer (EDT) but 14:30Z in winter (EST) — convert
            # to ET instead of a fixed UTC window, else the whole fast path arms
            # nothing all winter.
            from datetime import datetime, timezone
            from zoneinfo import ZoneInfo
            raw = r[0][1]
            ts = raw if isinstance(raw, datetime) else datetime.fromisoformat(str(raw))
            if ts.tzinfo is None:
                ts = ts.replace(tzinfo=timezone.utc)
            et_hhmm = ts.astimezone(ZoneInfo("America/New_York")).strftime("%H:%M")
            if "09:30" <= et_hhmm <= "09:33":
                _SESSION_OPEN_CACHE[key] = float(r[0][0])
                return _SESSION_OPEN_CACHE[key]
    except Exception:  # noqa: BLE001 — fail-open; gap falls back to window/day_open
        pass
    return None


def main(argv: list[str] | None = None) -> int:
    args = parse_args(argv)
    rt = build_runtime(args)
    ollama_cfg = rt["ollama_cfg"]
    symbols: Watchlist = rt["symbols"]
    client = rt["client"]

    print(f"[boot] session={rt['session_id']} db={_db_target_desc()}")
    print(f"[boot] watching ({rt['symbols_source']}): {', '.join(symbols)}")
    if not rt["has_keys"]:
        print("[boot] no ALPACA_API_KEY/SECRET — dry mode: ingestion, sync and "
              "order submission are skipped; watcher runs on existing data")

    # dashboard
    server = None
    if not args.no_dashboard and _flag("DASHBOARD_ENABLED", "1"):
        host = os.environ.get("DASHBOARD_HOST", "127.0.0.1")
        port = int(os.environ.get("DASHBOARD_PORT", "8010"))

        def inject_symbol(symbol: str) -> None:
            # add to the watcher's provider so it's evaluated next tick, and
            # to the live symbol list so the next ingest pass pulls its bars
            provider = rt["watcher"].provider
            if hasattr(provider, "add_symbol"):
                provider.add_symbol(symbol)
            symbols.append(symbol)  # Watchlist dedupes

        state = DashboardState(
            rt["event_db"],
            execution_service=rt["execution"],
            execution_mode=TradingModeSettings.from_env().execution_mode,
            research_con=rt["research_con"],
            watch_inject=inject_symbol,
        )
        try:
            server, _ = serve_in_background(state, host, port)
            print(f"[boot] dashboard at http://{host}:{port}")
        except OSError as exc:
            print(f"[boot] dashboard disabled ({exc})")

    # startup: daily history backfill so gap%/RVOL have a baseline
    if client:
        if args.discover:
            discovered = discover_active_symbols(
                client,
                top=10,
                price_min=float(os.environ.get("WATCHER_PRICE_MIN", "1")),
                price_max=float(os.environ.get("WATCHER_PRICE_MAX", "20")),
            )
            if discovered:
                print(f"[boot] screener added: {', '.join(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, list(symbols), days=30)
        print(f"[boot] daily backfill: {daily.daily_rows} rows"
              + (f" (errors: {daily.errors})" if daily.errors else ""))

    lookback = int(os.environ.get("LIVE_BARS_LOOKBACK_MINUTES", "240"))
    refresh_lb = int(os.environ.get("LIVE_BARS_REFRESH_MINUTES", "20"))
    _ingest_first = {"v": True}
    # freshness watchdog: warn when the newest bar is older than this during RTH.
    _ingest_stale_secs = float(os.environ.get("INGEST_STALE_SECONDS", "180"))
    # block arming on stale data? OFF by default — observability first; a
    # mis-tuned threshold must not silently halt trading.
    _ingest_stale_block = os.environ.get("INGEST_STALE_BLOCK", "0").strip() in {"1", "true", "yes"}
    _data_stale = {"v": False}

    def _minutes_since_premarket() -> float:
        try:
            from zoneinfo import ZoneInfo
            now_et = datetime.now(ZoneInfo("America/New_York"))
            pre_et = now_et.replace(hour=4, minute=0, second=0, microsecond=0)
            return max(0.0, (now_et - pre_et).total_seconds() / 60.0)
        except Exception:  # noqa: BLE001
            return 0.0

    def _first_pass_lookback() -> int:
        # The first ingest must cover the whole session-so-far so the
        # opening-range bars land in the DB (even on a mid-session restart);
        # after that we only pull the last few minutes — closed bars never
        # change, and re-fetching a 4-hour window for 80 symbols every minute
        # was the ~46s step that froze the loop.
        # "Session" starts at the 04:00 ET premarket, not 09:30: KITT was found at
        # 09:35 on 2026-09-25 and got bars from 09:13 only, losing the 09:00
        # squeeze and ~6M premarket shares — its volume, gap context and rvol
        # were all computed from the tail of the move.
        return int(max(lookback, _minutes_since_premarket() + 15))

    def _freshest_bar_age():
        """Seconds since the newest stored minute bar (UTC), or None. The bar
        timestamp is its START, and IEX lags ~1 bar, so ~90-120s is NORMAL during
        RTH — only flag well beyond that."""
        from datetime import timezone
        try:
            row = rt["research_con"].execute(
                "SELECT max(timestamp) FROM minute_bars WHERE session_date = ?",
                [_now_session_date()]).fetchone()
            if not row or not row[0]:
                return None
            ts = row[0]
            if isinstance(ts, str):
                ts = datetime.fromisoformat(ts.replace("Z", "+00:00"))
            if ts.tzinfo is None:
                ts = ts.replace(tzinfo=timezone.utc)
            return (datetime.now(timezone.utc) - ts).total_seconds()
        except Exception:  # noqa: BLE001
            return None

    def _is_rth():
        # holiday/early-close aware (was weekday<5 only, which traded holidays)
        return is_regular_hours()

    def _symbol_bar_age(symbol):
        """Seconds since this symbol's newest minute bar today (UTC), or None."""
        from datetime import timezone
        try:
            row = rt["research_con"].execute(
                "SELECT max(timestamp) FROM minute_bars "
                "WHERE symbol = ? AND session_date = ?",
                [symbol, _now_session_date()]).fetchone()
            if not row or not row[0]:
                return None
            ts = row[0]
            if isinstance(ts, str):
                ts = datetime.fromisoformat(ts.replace("Z", "+00:00"))
            if ts.tzinfo is None:
                ts = ts.replace(tzinfo=timezone.utc)
            return (datetime.now(timezone.utc) - ts).total_seconds()
        except Exception:  # noqa: BLE001
            return None

    def _recently_halted(symbol):
        """Halt heuristic for the entry guard: during RTH this symbol has gone
        silent (freshest bar older than the configured gap) WHILE ingestion is
        otherwise healthy (some name printed recently). For the liquid gappers we
        trade, a multi-minute silence is a trading halt, not thin coverage — and
        entering a halted name is the "+100% then halts down" trap. Conservative
        on purpose: a false positive just skips one entry (it retries next tick)."""
        ex = rt.get("execution")
        if ex is None or not ex.settings.halt_guard_enabled or not _is_rth():
            return False
        gap_s = ex.settings.halt_max_bar_gap_min * 60.0
        # is_halt_gap also gates the global-ingestion-health check, so a feed-wide
        # lag can't flag the whole book (pure + unit-tested in runtime.halt_guard)
        return is_halt_gap(_symbol_bar_age(symbol), _freshest_bar_age(), gap_s)

    def step_vratio():
        """Measure today's per-name IEX->SIP volume capture ratio from the delayed
        (free) SIP feed and keep it for ingest scaling. Merge, never clear: a name
        whose sample got too thin keeps its last good ratio for the session.
        Self-disables when the live feed is already SIP (paid plan) — the estimator
        is the free-tier fallback, and scaling consolidated volume would double-count
        (ingest_live_minute_bars refuses too; this just saves the API calls)."""
        if not client:
            return "skipped (no keys)"
        if data_feed(client) != "iex":
            return "off (live feed is SIP)"
        from research.ingestion.volume_ratio import compute_ratios, persist_ratios
        fresh = compute_ratios(client, list(symbols))
        if fresh:
            rt["vol_ratios"].update(fresh)
            persist_ratios(rt["research_con"], fresh)
        known = rt["vol_ratios"]
        if not known:
            return "no ratios yet"
        med = sorted(known.values())[len(known) // 2]
        return f"{len(fresh)} fresh / {len(known)} known, median x{med:.0f}"

    _ingested_syms: set = set()

    def step_ingest():
        if not client:
            return "skipped (no keys)"
        # PER-SYMBOL first pass: a global one-shot meant symbols discovered
        # mid-session only ever got the 20-min refresh window — no 09:30 bars, so
        # their opening range and overnight gap were computed from mid-day bars
        # (the 2026-07-08 TVRD/FRTT fake-gapper hole). New names get the full
        # session; known names get the incremental refresh.
        new_syms = [s for s in symbols if s not in _ingested_syms]
        known_syms = [s for s in symbols if s in _ingested_syms]
        res = IngestionResult() if not known_syms else ingest_live_minute_bars(
            rt["research_con"], client, known_syms, lookback_minutes=refresh_lb,
            volume_scale=rt["vol_ratios"])
        if new_syms:
            res_new = ingest_live_minute_bars(
                rt["research_con"], client, new_syms,
                lookback_minutes=_first_pass_lookback(),
                volume_scale=rt["vol_ratios"])
            res.minute_rows += res_new.minute_rows
            res.errors.extend(res_new.errors)
            res.per_symbol.update(res_new.per_symbol)
            _ingested_syms.update(new_syms)
        # Freshness watchdog: 45% of passes historically returned 0 rows with NO
        # error — indistinguishable from a feed outage. Make staleness VISIBLE
        # (emit + display); hard-block arming only if explicitly enabled, so a
        # mis-tuned threshold can't silently halt trading.
        age = _freshest_bar_age()
        stale = bool(_is_rth() and age is not None and age > _ingest_stale_secs)
        _data_stale["v"] = stale
        if stale:
            rt["store"].emit(ModuleTickEvent(
                timestamp=datetime.now(), mode=EventMode.PAPER,
                correlation_id=rt["session_id"],
                message=f"INGEST STALE: freshest bar {age:.0f}s old (> {_ingest_stale_secs:.0f}s)",
                module="ingestion", stage="stale", duration_ms=0.0,
                input_count=len(symbols), output_count=res.minute_rows,
                metrics={"bar_age_s": age}, errors=[{"error": "stale_data"}]))
        rt["store"].emit(
            ModuleTickEvent(
                timestamp=datetime.now(),
                mode=EventMode.PAPER,
                correlation_id=rt["session_id"],
                message=f"ingested {res.minute_rows} minute rows across {len(res.symbols)} symbols",
                module="ingestion",
                stage="completed",
                duration_ms=0.0,
                input_count=len(symbols),
                output_count=res.minute_rows,
                metrics={"per_symbol": res.per_symbol, "symbols_with_data": res.symbols},
                errors=[{"error": e} for e in res.errors],
            )
        )
        note = (f" STALE({age:.0f}s)" if stale else "")
        _lbdesc = f"{refresh_lb}m" + (f"+{len(new_syms)} new full" if new_syms else "")
        return (f"{res.minute_rows} rows (lookback {_lbdesc}){note}"
                + (f", errors={res.errors}" if res.errors else ""))

    def step_watch():
        res = rt["watcher"].tick(_now_session_date())
        return (f"evaluated={res.evaluated} ready={res.ready} blocked={len(res.blocked)}"
                + (f" errors={res.errors}" if res.errors else ""))

    def step_sync():
        if not rt["sync"]:
            return "skipped (no keys)"
        res = rt["sync"].sync_all()
        failed = [k for k, v in res.items() if v is None]
        return "ok" if not failed else f"DEGRADED ({', '.join(failed)})"

    def step_execute():
        if not rt["execution"]:
            return "skipped (no keys)"
        # Hand the trigger book's current ranks to the slow auto path so it honours
        # concentrate-by-rank identically to the fast path (plan 005). None-safe: an
        # unranked/absent symbol -> factor 1.0 (no-op). Built inline from the book's
        # thread-safe trigger dict — no accessor added to runtime/triggers.py.
        _bk = rt.get("book")
        rank_map = ({sym: t.rank for sym, t in _bk.triggers.items()} if _bk else None)
        res = rt["execution"].tick(rank_map=rank_map)
        backed = len(res.get("backed_out") or [])
        # `auto` is the raw approve_order() results: a REAL broker submission carries a
        # broker_order_id; rejections (e.g. the daily-entry cap) are tagged rejected=True.
        # Count only real submissions as executed, and surface rejections explicitly so a
        # tripped cap is visible instead of masquerading as auto_executed=1.
        auto = res.get("auto_executed") or []
        executed = [r for r in auto if r.get("broker_order_id")]
        rejected = [r for r in auto if r.get("rejected")]
        line = (f"approvals_requested={res['approvals_requested']} "
                f"auto_executed={len(executed)}")
        if rejected:
            hist: dict[str, int] = {}
            for r in rejected:
                k = r.get("reason") or "?"
                hist[k] = hist.get(k, 0) + 1
            reasons = ", ".join(f"{k}x{v}" for k, v in sorted(hist.items()))
            line += f" auto_rejected={len(rejected)} ({reasons})"
        if backed:
            syms = ", ".join(b["symbol"] for b in res["backed_out"])
            line += f" backed_out={backed} ({syms})"
        return line

    def step_guard():
        # Fast invalidation watch: re-check armed (unfilled) entries against the
        # live last-trade price and the wall-clock timeout. Runs far more often
        # than `execute` so a break of the entry trigger is caught tick-by-tick
        # instead of waiting for the next execute pass.
        if not rt["execution"]:
            return None
        backed = rt["execution"].expire_stale_entries()
        if not backed:
            return None
        return "backed_out " + ", ".join(
            f"{b['symbol']} ({b['reason']})" for b in backed
        )

    # Operator command bus (plan 007): consume kill-switch commands emitted by the
    # dashboard / qabot on the same 5s cadence as the guard. flatten_all routes
    # through close_session DIRECTLY (never step_eod_flatten) so it can't trip the
    # plan-003 daily-restart exit. None in dry mode (no execution service).
    from runtime.operator_commands import OperatorCommandConsumer
    command_consumer = (
        OperatorCommandConsumer(rt["store"], rt["execution"], session_id=rt["session_id"])
        if rt["execution"] is not None else None
    )

    def step_commands():
        if command_consumer is None:
            return None
        done = command_consumer.poll()
        return f"operator commands executed: {len(done)}" if done else None

    def step_discover(con=None):
        if not client:
            return "skipped (no keys)"
        res = run_discovery(
            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")),
            top=int(os.environ.get("DISCOVER_TOP", "20")),
        )
        # fold newly-screened names into the live watchlist so the next ingest
        # pulls their bars and the watcher evaluates them
        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"))
        symbols.trim(max_uni)
        return (f"universe={len(res.universe)} gappers={len(res.gappers)}"
                + (f" +{len(added)} new" if added else ""))

    # --- fast armed-trigger path -------------------------------------------
    book = rt["book"]
    price_min_v = float(os.environ.get("WATCHER_PRICE_MIN", "1"))
    price_max_v = float(os.environ.get("WATCHER_PRICE_MAX", "20"))
    orb_bars_v = int(os.environ.get("ORB_BARS", "5"))
    stop_cushion_v = float(os.environ.get("TRIGGER_STOP_CUSHION", "0.005"))

    def step_arm():
        """Rank the most-promising gappers and pre-compute each one's opening-
        range breakout trigger/stop, so step_trigger can fire on a live cross."""
        from research import query as rq
        from research.ingestion.discovery import recent_news_map
        from research.ingestion.signals import scan_gappers
        from strategy.evaluation.structure import opening_range

        # don't arm new triggers on stale data when blocking is enabled — a break
        # computed off a frozen feed could fire into a name that has already moved.
        if _ingest_stale_block and _data_stale["v"]:
            return "skipped: data stale (INGEST_STALE_BLOCK)"
        session_date = _now_session_date()
        # fresh catalysts (the discover step ingests the feeds; we just read them)
        news_map = recent_news_map(
            rt["research_con"], lookback_hours=int(os.environ.get("NEWS_LOOKBACK_HOURS", "8")))
        news_boost = float(os.environ.get("NEWS_RANK_BOOST", "2.0"))
        # GATED graded ranking from the LLM catalyst read. When the score blend is
        # off, ranking is the existing binary `news_boost if catalyst else 1.0`.
        # When on, a bullish catalyst boosts by `1 + gain*score` and a confirmed
        # dilutive offering is sunk below the armed cut (multiplied by a small
        # penalty) so the book stops arming the gap-and-fade traps.
        cat_map: dict = {}
        if ollama_cfg.catalyst_score_enabled:
            from research.ingestion.news_enrichment import catalyst_map as _cmap
            try:
                cat_map = _cmap(rt["research_con"])
            except Exception:  # noqa: BLE001
                cat_map = {}
        news_rank_gain = float(os.environ.get("NEWS_RANK_BOOST_GAIN", "2.0"))
        dilutive_penalty = float(os.environ.get("NEWS_DILUTIVE_PENALTY", "0.1"))
        # premarket travel: the one measured predictor of a >=50% session (3.1x
        # lift at >=15%). See strategy/evaluation/premarket_range.py for the study.
        pm_cfg = PremarketRangeConfig.from_env()

        _PEGGED = ("acqui", "merger", "buyout", "to be acquired", "take-private")

        def _headline_pegged(sym: str) -> bool:
            """Buyout/merger headlines PEG the price (CRNX 2026-07-07: pinned at
            $83.50 by an $85 buyout) — the opposite of a momentum catalyst. Never
            rank-boost them."""
            h = (news_map.get(sym) or "").lower()
            return any(k in h for k in _PEGGED)

        def _cat_mult(sym: str, has_headline: bool) -> float:
            if has_headline and _headline_pegged(sym):
                return 1.0
            if not ollama_cfg.catalyst_score_enabled:
                return news_boost if has_headline else 1.0
            from research.ingestion.news_enrichment import catalyst_score as _cscore
            adv = cat_map.get((sym or "").upper())
            if not adv:
                # No LLM advisory for THIS symbol -> fall back to the binary
                # headline boost instead of dropping news from the ranking.
                # Returning a flat 1.0 here meant that whenever the enrichment
                # was down, cat_map was empty for EVERY symbol and news stopped
                # influencing the rank at all — silently. That is exactly what
                # happened from 2026-08-03 (Ollama host/model misconfigured)
                # through 2026-09-13: 40 days of live trading, two days of which
                # preceded the first real-money fill, with zero news input to the
                # ranking and no error anywhere to show for it.
                return news_boost if has_headline else 1.0
            if adv.get("is_dilutive") and float(adv.get("conviction") or 0.0) >= (
                    ollama_cfg.dilution_veto_min_conviction):
                return dilutive_penalty
            return 1.0 + news_rank_gain * (_cscore(adv) or 0.0)
        # optional MANUAL per-symbol priority nudge (SYMBOL_WEIGHTS=ABCD:1.2,XYZ:1.1)
        # — a small boost to names you want watched; keep modest so it doesn't
        # override the gap/volume logic.
        weights: dict[str, float] = {}
        for part in os.environ.get("SYMBOL_WEIGHTS", "").split(","):
            if ":" in part:
                s, w = part.split(":", 1)
                try:
                    weights[s.strip().upper()] = float(w)
                except ValueError:
                    pass
        candidates: list[dict] = []
        # rank ALL session symbols by gap*rvol (thresholds applied later, per
        # trigger, so the board still shows the field pre-gap)
        gappers = scan_gappers(
            rt["research_con"], session_date,
            min_gap_pct=0.0, min_relative_volume=0.0,
            price_min=price_min_v, price_max=price_max_v,
            limit=max(12, book.max_armed + 6),
        )
        for g in gappers:
            bars = rq.query_minute_bars(rt["research_con"], g.symbol, session_date)
            hi, lo, complete = opening_range(bars, orb_bars=orb_bars_v)
            stop = lo * (1.0 - stop_cushion_v) if lo else None
            # premarket travel off the SAME frame — no extra query
            pm_range = premarket_range_pct(bars)
            catalyst = news_map.get(g.symbol, "")
            # recent $-volume (last 5 regular-hours bars) — the liquidity gate.
            # rvol is relative; this is the absolute "can it absorb an order" check.
            dollar_vol = 0.0
            day_open = None
            try:
                rth = bars[bars["is_regular_hours"]] if "is_regular_hours" in bars else bars
                tail = rth.tail(5)
                dollar_vol = float((tail["close"] * tail["volume"]).sum())
                if len(rth):  # session open = first regular-hours bar's open
                    day_open = float(rth.iloc[0]["open"])
            except Exception:  # noqa: BLE001
                dollar_vol = 0.0
            # STATIC overnight gap (session open vs prev close) — the population
            # definition. The scanner's g.gap_pct is DYNAMIC (latest price vs prev
            # close): intraday rallies masqueraded as gappers through it (TVRD 6%
            # true gap entered 2026-07-07; audit 2026-07-08). None until the open.
            # day_open above comes from query_minute_bars, which returns a WINDOW —
            # its "first RTH bar" is a mid-session bar later in the day, which let
            # TVRD (+4%) and FRTT arm on 2026-07-08 as fake "20% gappers". Use the
            # authoritative first-RTH-bar open, cached (immutable once set).
            true_open = _session_open(rt["research_con"], g.symbol, session_date)
            if true_open:
                day_open = true_open
            gap_open_pct = None
            if true_open and g.prev_close and g.prev_close > 0:
                gap_open_pct = (true_open / g.prev_close - 1.0) * 100.0
            elif _is_rth():
                # session underway but this symbol's opening bars are not ingested
                # yet (mid-day discovery): its "opening range" would be a fake
                # mid-day window and its gap unknowable. Do not arm until the
                # per-symbol backfill lands (usually the next ingest pass).
                hi = None
                stop = None
            candidates.append({
                "symbol": g.symbol,
                "gap_pct": g.gap_pct,
                "gap_open_pct": gap_open_pct,
                "rvol": g.relative_volume,
                "rvol_tod": g.relative_volume_tod,
                "rvol_med": g.relative_volume_med,
                "trigger": hi,
                "stop": stop,
                "range_pct": ((hi - lo) / hi) if (hi and lo and hi > 0) else 0.0,
                "cum_volume": g.cumulative_volume,
                "dollar_vol": dollar_vol,
                "catalyst": catalyst,
                "day_open": day_open,
                # a fresh catalyst is WHY a small-cap runs — boost it up the rank;
                # times an optional manual per-symbol weight (default 1.0)
                "pm_range_pct": pm_range,
                "_score": (g.gap_pct * max(g.relative_volume, 0.1)
                           * _cat_mult(g.symbol, bool(catalyst))
                           * premarket_rank_multiplier(pm_range, pm_cfg)
                           * weights.get(g.symbol, 1.0)),
                "complete": complete,
            })
        # pre-open / before any gappers form: surface the queued watchlist so the
        # board shows what we're about to watch instead of looking blank
        if not candidates:
            for s in symbols[: book.max_armed]:
                candidates.append({"symbol": s, "gap_pct": 0.0, "rvol": 0.0,
                                   "trigger": None, "stop": None,
                                   "range_pct": 0.0, "complete": False})
        # catalyst-backed gappers float to the top of the armed set
        candidates.sort(key=lambda c: c.get("_score", 0.0), reverse=True)
        book.arm(candidates)
        # Armed names must also get watcher scoring: the trigger book arms from
        # the live gapper scan, so names discovery never surfaced carried NO
        # criteria trail — 2026-08-01 four of ten armed rows had a blank
        # reasoning panel. extra_symbols is the provider's standing hook.
        try:
            _prov = rt["watcher"].provider
            for c in candidates:
                if c.get("symbol") and c.get("complete"):
                    _prov.add_symbol(c["symbol"])
        except Exception:  # noqa: BLE001 — observability aid, never blocks arming
            pass
        armed = sum(1 for t in book.triggers.values() if t.state == "armed")
        n_news = sum(1 for c in candidates if c.get("catalyst"))
        # NOTE: no pm-range counter here. The arm step never emits a module_tick
        # (verified against 2026-09-11 and 2026-09-14: only discovery, ingestion,
        # enrichment and triggers do), so anything added to this string reaches
        # container stdout and nothing else. Adding a tick just for it would put
        # ~4k events/day back after the watcher filter removed 88% of them. The
        # per-candidate pm_range_pct above is on the armed rows for the board;
        # persist it on the approval payload if it needs to be measurable per trade.
        return (f"tracking={len(book.triggers)} armed={armed}"
                + (f" news={n_news}" if n_news else ""))

    def step_quotes():
        """Record NBBO bid/ask SIZES for tracked + held names (order-book-imbalance
        proxy — full L2 depth needs a paid feed; NBBO sizes are the free approximation).
        Recording only, zero trading impact; premarket included (that's where the
        edge data lives). A future shadow study tests ask/bid-size imbalance as a
        pop-vs-fade predictor."""
        syms = set(book.triggers.keys())
        if rt["execution"] is not None:
            try:
                syms |= set(rt["execution"]._held_symbols())
            except Exception:  # noqa: BLE001
                pass
        if not syms:
            return None
        try:
            quotes = client.get_latest_quotes(sorted(syms))
        except Exception as exc:  # noqa: BLE001
            return f"quotes fetch failed: {exc}"
        n = 0
        for sym, q in (quotes or {}).items():
            if not q:
                continue
            rt["research_con"].execute(
                "INSERT INTO quote_snapshots (symbol, bid_price, bid_size, ask_price, ask_size) "
                "VALUES (?, ?, ?, ?, ?)",
                [sym, q.get("bp"), q.get("bs"), q.get("ap"), q.get("as")])
            n += 1
        return f"{n} quotes ({len(syms)} names)"

    # The trigger runs on its OWN thread, not the scheduler, so a slow ingest
    # can never freeze breakout detection. It only touches the (thread-safe)
    # book, execution service, and event store — never research_con.
    trigger_interval = float(os.environ.get("TRIGGER_INTERVAL_SECONDS", "4"))
    stop_event = threading.Event()

    def _batch_prices(syms):
        """One batched last-trade call for all armed names (8 syms ≈ 1 sym cost)."""
        if client is None or not syms:
            return {}
        try:
            trades = client.get_latest_trades(list(syms))
        except Exception:  # noqa: BLE001
            return {}
        out = {}
        for s in syms:
            tr = trades.get(s) or trades.get(s.upper())
            if tr and tr.get("p") is not None:
                try:
                    out[s] = float(tr["p"])
                except (TypeError, ValueError):
                    pass
        return out

    _block_logged: set[str] = set()  # dedup TRIGGER BLOCKED spam (re-checked every pass)

    def _trigger_pass():
        snap = book.snapshot()
        syms = [r["symbol"] for r in snap]
        if not syms:
            return
        for s, px in _batch_prices(syms).items():
            book.update_price(s, px)
        fired: list[str] = []
        blocked: list[str] = []
        if rt["execution"] is not None:
            try:
                book.mark_filled(rt["execution"]._held_symbols())
            except Exception:  # noqa: BLE001
                pass
            for t in book.fires():
                res = rt["execution"].submit_breakout_now(
                    t.symbol, t.trigger, t.stop, last_price=t.price,
                    cum_volume=t.cum_volume, halted=_recently_halted(t.symbol),
                    day_open=t.day_open, rank=t.rank,
                    # STATIC overnight gap only — never the dynamic book gap
                    gap_pct=(t.gap_open_pct / 100.0) if t.gap_open_pct is not None else None,
                    rvol=t.squeeze_rvol, price=t.price,
                )
                if res.get("ok"):
                    book.mark_fired(t.symbol)
                    fired.append(t.symbol)
                elif res.get("skipped") in ("over_extended", "over_extended_day",
                                            "halted_symbol"):
                    # surface the anti-chase / halt skips once per symbol (they
                    # re-check every 4s; routine dedup skips stay quiet entirely)
                    key = f"{t.symbol}:{res['skipped']}"
                    if key not in _block_logged:
                        _block_logged.add(key)
                        blocked.append(key)
        snap = book.snapshot()
        armed_n = sum(1 for r in snap if r["state"] == "armed")
        rt["store"].emit(
            ModuleTickEvent(
                timestamp=datetime.now(), mode=EventMode.PAPER,
                correlation_id=rt["session_id"],
                message=f"triggers armed={armed_n} fired={len(fired)}",
                module="triggers", stage="snapshot", duration_ms=0.0,
                input_count=len(syms), output_count=len(fired),
                metrics={"triggers": snap, "armed": armed_n}, errors=[],
            )
        )
        if fired:
            print(f"[{datetime.now():%H:%M:%S}] TRIGGER FIRED {', '.join(fired)}", flush=True)
        if blocked:
            print(f"[{datetime.now():%H:%M:%S}] TRIGGER BLOCKED {', '.join(blocked)}", flush=True)

    def trigger_loop():
        while not stop_event.wait(trigger_interval):
            try:
                _trigger_pass()
            except Exception as exc:  # noqa: BLE001
                print(f"[trigger] error: {exc}", flush=True)

    # --- live exit management (breakeven / trail / scale / first-red) -------
    exit_cfg = ExitConfig.from_env()

    # Exits run on their OWN thread (plan 018) with a DEDICATED research
    # connection — the shared rt["research_con"] is not thread-safe, and the
    # exit loop reads bars concurrently with the scheduler. Open it lazily so a
    # DB hiccup at boot doesn't stop the loop; None falls back to the shared
    # connection (still correct, just not concurrency-isolated).
    _exit_con: dict = {"c": None}

    def _rth_bars(sym):
        from research import query as rq
        con = _exit_con["c"] or rt["research_con"]
        b = rq.query_minute_bars(con, sym, _now_session_date())
        if b is not None and len(b) and "is_regular_hours" in b.columns:
            b = b[b["is_regular_hours"] == True].reset_index(drop=True)  # noqa: E712
        return b

    exit_mgr = (
        LiveExitManager(client, rt["store"], _rth_bars, cfg=exit_cfg,
                        session_id=rt["session_id"])
        if client else None
    )
    print(f"[boot] exit management: {exit_cfg.describe()}")
    _es = rt["execution"].settings if rt["execution"] is not None else None
    if _es is not None:
        print(
            f"[boot] entry gates: VWAP {'ENFORCED' if _es.require_above_vwap else 'shadow-only'}"
            f", anti-chase {'ON (auto path)' if _es.unified_entry else 'fast-path only'}"
            # rank concentration now runs on BOTH paths through the shared gate
            # (plan 005) — no longer fast-path-only, so no warning is needed.
            f", rank top-N {_es.concentrate_top_n or 'off'} (both paths)"
            f", fill={_es.entry_fill_model}"
            f", daily-cap {_es.max_fresh_entries_per_day or 'off'}"
            f", risk-cap ${_es.max_risk_dollars:.0f}"
        )

    # Telegram push alerts — gated, decoupled (polls the event store), and
    # fire-and-forget so it can never impede trading.
    from runtime.notifier import TelegramNotifier, send_telegram
    notifier = TelegramNotifier(rt["store"])
    if notifier.enabled:
        print("[boot] telegram alerts: ON")
        send_telegram("\U0001F916 Momentum loop started — push alerts armed.")
    else:
        print("[boot] telegram alerts: off (TELEGRAM_BOT_TOKEN/CHAT_ID unset)")

    def step_notify():
        n = notifier.poll()
        return f"telegram: {n} alert(s)" if n else None

    def step_retention():
        from runtime.retention import maybe_prune
        out = maybe_prune(rt["store"])
        return None if out in ("idle", "done-today") else out

    def step_telemetry():
        from runtime.telemetry_push import push_snapshot
        return push_snapshot(
            rt["store"], broker=os.environ.get("BROKER", "alpaca").strip().lower())

    def step_manage_exits():
        if exit_mgr is None:
            return None
        acts = exit_mgr.manage()
        return ", ".join(acts) if acts else None

    # --- end-of-day auto-flatten (a day-trading book shouldn't gap overnight) --
    eod_enabled = _flag("TRADING_EOD_FLATTEN_ENABLED", "1")
    # Legacy absolute time (15:55) is kept ONLY to derive the normal lead minutes
    # (16:00 - 15:55 = 5) — the window itself is now relative to the ACTUAL close
    # so it fires on early-close half-days too. Before a multi-day closure the
    # lead widens so the book is flat with margin (the ATPC naked-holiday case).
    try:
        eod_h, eod_m = (int(x) for x in
                        os.environ.get("TRADING_EOD_FLATTEN_TIME", "15:55").split(":"))
    except Exception:  # noqa: BLE001
        eod_h, eod_m = 15, 55
    _normal_lead = max(1, 16 * 60 - (eod_h * 60 + eod_m))            # 15:55 -> 5
    _pre_lead = int(os.environ.get("TRADING_PRE_CLOSURE_FLATTEN_LEAD_MIN", "20"))
    # Restart-churn guard: the daily-restart exit fires INSIDE the EOD window and
    # compose restarts within seconds — the fresh process must see today's
    # already-successful flatten (event-store truth) or it would flatten-nothing,
    # exit again, and churn until the window closes.
    _eod_done = {"v": rt["execution"] is not None
                 and rt["execution"].eod_flatten_done_today()}
    if _eod_done["v"]:
        print("[boot] EOD flatten already succeeded today — no re-flatten, no re-exit")

    def step_eod_flatten():
        if not eod_enabled or _eod_done["v"] or rt["execution"] is None:
            return None
        try:
            from zoneinfo import ZoneInfo
            now_et = datetime.now(ZoneInfo("America/New_York"))
        except Exception:  # noqa: BLE001
            return None
        # Fire inside [close - lead, close): after the close a market order just
        # queues uselessly. The lead WIDENS before a multi-day closure so a thin
        # name has time to fill and a transient failure still has a wide retry
        # window. Skips weekends/holidays (session_close_hm -> None) — flattening
        # into a closed market can't fill the sell.
        due, pre_closure = eod_flatten_status(now_et, _normal_lead, _pre_lead)
        if not due:
            return None
        reason = "eod_flatten_preclose" if pre_closure else "eod_flatten"
        res = rt["execution"].close_session(reason)
        # only mark done if it actually succeeded — otherwise RETRY on the next
        # pass within the window (a flatten that errored once left positions naked
        # overnight). close_session re-blocks entries each call, so retrying is safe.
        ok = not res["errors"]
        _eod_done["v"] = ok
        closed = ", ".join(res["closed_positions"]) or "none"
        tag = "PRE-HOLIDAY EOD FLATTEN" if pre_closure else "EOD FLATTEN"
        if res["errors"]:
            # A failed flatten before a MULTI-DAY gap = naked carry across the
            # closure (the silent 6/18 Juneteenth failure). Scream about it.
            gap = days_to_next_session(now_et.date())
            sev = f"‼ NAKED-CARRY RISK ({gap}-day gap ahead): " if pre_closure else ""
            print(f"[{datetime.now():%H:%M:%S}] {sev}{tag} could not flatten: {res['errors']}", flush=True)
            send_telegram(f"⚠️ {tag} FAILED to flatten: {res['errors']} — {sev or 'will retry'}")
            return f"{sev}{tag} closed [{closed}] errors={res['errors']} (will retry)"
        # success — push the once-per-day EOD P&L summary
        try:
            from storage.projections import query_session_pnl
            notifier.send_eod_summary(now_et.date().isoformat(),
                                      query_session_pnl(rt["store"]))
        except Exception:  # noqa: BLE001
            pass
        # the process is the session: a successful flatten (including "already
        # flat, nothing to close") ends it — exit cleanly, compose restarts fresh
        if _request_exit_after_eod(ok):
            msg = ("♻️ EOD flatten complete — exiting for daily restart "
                   "(compose brings the loop back fresh).")
            print(f"[{datetime.now():%H:%M:%S}] {msg}", flush=True)
            send_telegram(msg)
        return f"{tag} closed [{closed}]"

    # --- open catch-up flatten: a day-trading book should be FLAT by the open;
    # close anything carried from a prior session (a missed/failed EOD flatten)
    # before it bleeds for days — the ATPC-over-Juneteenth backstop. Does NOT
    # halt the session; today's fresh names are left to trade normally.
    catchup_enabled = _flag("TRADING_OPEN_CATCHUP_FLATTEN_ENABLED", "1")
    _catchup_done = {"v": False}

    def step_open_catchup_flatten():
        if not catchup_enabled or _catchup_done["v"] or rt["execution"] is None:
            return None
        try:
            from zoneinfo import ZoneInfo
            now_et = datetime.now(ZoneInfo("America/New_York"))
        except Exception:  # noqa: BLE001
            return None
        if not is_regular_hours(now_et):       # only act while the market is open
            return None
        res = rt["execution"].flatten_overnight_carries(now_et.date(), "open_catchup")
        if res["errors"]:
            print(f"[{datetime.now():%H:%M:%S}] ‼ OPEN CATCH-UP could not flatten carry: "
                  f"{res['errors']}", flush=True)
            return f"OPEN CATCH-UP carries={res['carries']} closed={res['closed']} errors={res['errors']} (will retry)"
        # clean (no carries) or all carries closed -> done for the session
        _catchup_done["v"] = True
        if res["carries"]:
            print(f"[{datetime.now():%H:%M:%S}] OPEN CATCH-UP flattened stale carries: "
                  f"{res['closed']}", flush=True)
            return f"OPEN CATCH-UP flattened overnight carries {res['closed']}"
        return None

    def step_news():
        """Land latest Alpaca/Benzinga news for the tracked universe into
        raw_news_items (feeds recent_news_map catalyst prioritisation + the Ollama
        enrichment). Off the hot path; never blocks trading."""
        if not client:
            return "skipped (no keys)"
        res = ingest_alpaca_news(
            rt["research_con"], client, symbols=symbols,
            limit=int(os.environ.get("ALPACA_NEWS_LIMIT", "50")),
        )
        if res.error:
            return f"ERROR {res.error}"
        if res.new_items:
            return f"+{len(res.new_items)} new / {res.item_count} fetched"
        return None

    # Local-LLM (Ollama) news/catalyst enrichment — advisory layer. Runs on its
    # OWN daemon thread with its OWN DB connection (started below), NOT the
    # scheduler thread: a slow/hung Ollama pass (up to batch_limit x timeout) must
    # never stall exits / guard / eod-flatten, which share the scheduler thread,
    # and research_con is a single psycopg2 connection unsafe to share across
    # threads. Gated by ollama.enabled AND enrichment_enabled; no-op if Ollama down.
    def enrich_loop():
        try:
            econ = open_research_db("market")
        except Exception as exc:  # noqa: BLE001
            print(f"[enrich] no DB connection, enrichment off: {exc}", flush=True)
            return
        # enrich_recent_news NEVER raises — it reports via its counts dict. The
        # old loop discarded that return value, so a totally dead enrichment
        # (Ollama unreachable, model absent) looked identical to a healthy idle
        # one and ran unnoticed for 40 days. Emit a module_tick so the failure is
        # visible to the monitoring board and queryable in the event store.
        consecutive_bad = 0
        while not stop_event.wait(float(ollama_cfg.enrichment_interval_seconds)):
            try:
                counts = enrich_recent_news(econ, ollama_cfg) or {}
            except Exception as exc:  # noqa: BLE001
                counts = {"enriched": 0, "skipped": 0, "errors": -1}
                print(f"[enrich] error: {exc}", flush=True)
            errors = int(counts.get("errors") or 0)
            enriched = int(counts.get("enriched") or 0)
            # "bad" = the pass tried to classify and every attempt failed.
            bad = errors != 0 and enriched == 0
            consecutive_bad = consecutive_bad + 1 if bad else 0
            if bad:
                print(f"[enrich] {errors} failures, 0 enriched "
                      f"(host={ollama_cfg.host} model={ollama_cfg.model})", flush=True)
            # Report on the first bad pass and then hourly, so a persistent
            # outage stays visible without flooding the store every interval.
            period = max(1, int(3600 / max(1.0, float(ollama_cfg.enrichment_interval_seconds))))
            if enriched or (bad and (consecutive_bad == 1 or consecutive_bad % period == 0)):
                rt["store"].emit(ModuleTickEvent(
                    timestamp=datetime.now(),
                    mode=EventMode.PAPER,
                    correlation_id=rt["session_id"],
                    message=(f"news enrichment DOWN: {errors} failures, 0 enriched "
                             f"({consecutive_bad} passes)" if bad
                             else f"news enrichment: +{enriched} classified"),
                    module="enrichment",
                    stage="classify",
                    duration_ms=0.0,
                    input_count=enriched + errors,
                    output_count=enriched,
                    metrics={"enriched": enriched, "errors": errors,
                             "skipped": int(counts.get("skipped") or 0),
                             "host": ollama_cfg.host, "model": ollama_cfg.model,
                             "consecutive_failed_passes": consecutive_bad},
                    errors=([{"error": "news catalyst enrichment produced nothing — "
                              f"check Ollama at {ollama_cfg.host} and that model "
                              f"'{ollama_cfg.model}' is pulled. News ranking is "
                              "running on the binary headline boost until it recovers."}]
                            if bad else []),
                ))

    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")
                           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")),
                  enabled=_flag("LIVE_BARS_ENABLED", "1"))
    scheduler.add("watch", step_watch, float(os.environ.get("WATCHER_INTERVAL_SECONDS", "30")),
                  enabled=_flag("WATCHER_ENABLED", "1"))
    scheduler.add("arm", step_arm, float(os.environ.get("TRIGGER_ARM_INTERVAL_SECONDS", "20")),
                  enabled=_flag("TRIGGER_ENABLED", "1"))
    scheduler.add("sync", step_sync, float(os.environ.get("ALPACA_PAPER_SYNC_INTERVAL_SECONDS", "60")),
                  enabled=_flag("ALPACA_PAPER_SYNC_ENABLED", "1"))
    scheduler.add("quotes", step_quotes, float(os.environ.get("QUOTE_SNAPSHOT_INTERVAL_SECONDS", "15")),
                  enabled=_flag("QUOTE_SNAPSHOT_ENABLED", "1"))
    scheduler.add("execute", step_execute, float(os.environ.get("TRADING_EXECUTION_INTERVAL_SECONDS", "30")),
                  enabled=_flag("TRADING_EXECUTION_ENABLED", "1"))
    scheduler.add("guard", step_guard, float(os.environ.get("TRADING_ENTRY_GUARD_INTERVAL_SECONDS", "5")),
                  enabled=_flag("TRADING_EXECUTION_ENABLED", "1"))
    scheduler.add("command", step_commands,
                  float(os.environ.get("OPERATOR_COMMAND_INTERVAL_SECONDS", "5")),
                  enabled=(rt["execution"] is not None))
    # SINGLE-CALLER INVARIANT (plan 018): the scheduler exits step runs ONLY when
    # the dedicated exit thread is OFF — otherwise manage() would be called from
    # two threads (racing instance state + double-submit). Thread default ON.
    scheduler.add("exits", step_manage_exits, float(os.environ.get("TRADING_EXIT_MANAGE_INTERVAL_SECONDS", "12")),
                  enabled=(_flag("TRADING_EXECUTION_ENABLED", "1")
                           and not _flag("TRADING_EXIT_THREAD_ENABLED", "1")))
    scheduler.add("eod", step_eod_flatten, float(os.environ.get("TRADING_EOD_INTERVAL_SECONDS", "30")),
                  enabled=_flag("TRADING_EXECUTION_ENABLED", "1"))
    scheduler.add("notify", step_notify, float(os.environ.get("TELEGRAM_NOTIFY_INTERVAL_SECONDS", "45")),
                  enabled=notifier.enabled)
    scheduler.add("catchup", step_open_catchup_flatten,
                  float(os.environ.get("TRADING_OPEN_CATCHUP_INTERVAL_SECONDS", "30")),
                  enabled=_flag("TRADING_EXECUTION_ENABLED", "1"))
    # nightly event-store prune (runtime/retention.py): cheap hourly check,
    # actual DELETE once/day at RETENTION_HOUR
    scheduler.add("retention", step_retention, 1800.0,
                  enabled=_flag("RETENTION_ENABLED", "1"))
    # external metrics push (runtime/telemetry_push.py): off unless
    # TELEMETRY_PUSH_URL is set
    scheduler.add("telemetry", step_telemetry,
                  float(os.environ.get("TELEMETRY_PUSH_INTERVAL_SECONDS", "300")),
                  enabled=bool(os.environ.get("TELEMETRY_PUSH_URL", "").strip()))

    def run_pass():
        # force every task due, in pipeline order
        for task in scheduler.tasks:
            task.last_run = -10**9
        results = {}
        for name in ("discover", "ingest", "watch", "arm", "sync", "execute", "exits", "eod", "catchup"):
            for task in scheduler.tasks:
                if task.name == name and task.enabled:
                    results.update(scheduler_run_one(task))
        stamp = datetime.now().strftime("%H:%M:%S")
        line = " | ".join(f"{k}: {v}" for k, v in results.items())
        print(f"[{stamp}] {line}")

    def scheduler_run_one(task):
        task.last_run = time.monotonic()
        task.run_count += 1
        try:
            out = task.func()
            task.last_error = None
            return {task.name: out}
        except Exception as exc:  # noqa: BLE001
            task.error_count += 1
            task.last_error = str(exc)
            return {task.name: f"ERROR {exc}"}

    if args.once:
        run_pass()
        if server:
            server.shutdown()
        return 0

    print("[boot] entering live loop (Ctrl-C to stop)")
    # start the fast trigger on its own thread so slow ingest can't freeze it
    trigger_thread = None
    if _flag("TRIGGER_ENABLED", "1") and client is not None:
        trigger_thread = threading.Thread(target=trigger_loop, name="trigger", daemon=True)
        trigger_thread.start()
        print(f"[boot] fast trigger thread started (every {trigger_interval:.0f}s)")
    # exits on their OWN thread (plan 018): sub-second reaction, independent of
    # the scheduler queue. Gated so the slow scheduler "exits" step is disabled
    # whenever this runs (single-caller invariant — see below).
    _exit_thread_on = (_flag("TRADING_EXIT_THREAD_ENABLED", "1")
                       and exit_mgr is not None and client is not None)
    if _exit_thread_on:
        from runtime.exit_thread import run_exit_loop
        _exit_interval = float(os.environ.get("TRADING_EXIT_THREAD_INTERVAL_SECONDS", "1.0"))

        def exit_loop():
            try:  # dedicated connection so bars reads don't race the scheduler
                _exit_con["c"] = open_research_db("market")
            except Exception as exc:  # noqa: BLE001 — fall back to shared con
                print(f"[exit-thread] dedicated DB unavailable, using shared: {exc}", flush=True)
            run_exit_loop(exit_mgr, stop_event, _exit_interval)

        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")
    # Whole-market premarket squeeze scan (Alpaca snapshots) on its OWN light
    # thread. The movers/most-actives screeners only see the regular session, so
    # premarket squeezes stayed invisible until 09:30 (KITT 2026-09-25: ran at
    # 09:00, found 09:35). Discovery only: names join the watchlist and flow
    # through the normal new-symbol ingest (full-session first pass); every
    # trading gate still applies. Same thread-safety pattern as plan 019: own DB
    # connection, thread-safe Watchlist, single writer per step.
    if (_flag("DISCOVER_SNAPSHOT_SCAN_ENABLED", "1") and _flag("DISCOVER_ENABLED", "1")
            and client is not None):
        from research.ingestion.snapshot_scan import (
            SnapshotScanConfig, SnapshotScanner, emit_discoveries, in_scan_window)
        from runtime.exit_thread import run_exit_loop as _run_scan_loop  # generic driver
        _scan_cfg = SnapshotScanConfig.from_env()
        _scanner = SnapshotScanner(client, _scan_cfg)
        _scan_con: dict = {"c": None}

        def step_snapshot_scan():
            now = datetime.now(timezone.utc)
            if not in_scan_window(now):   # 04:00-16:00 ET, weekdays
                return
            res = _scanner.sweep(now, already_watched=set(symbols.snapshot()))
            names = [h.symbol for h in res.injected]
            if names:
                emit_discoveries(rt["store"], res.injected, mode=EventMode.PAPER,
                                 correlation_id=rt["session_id"],
                                 scan_duration_s=res.duration_s)
                # prior close / 20d volume baseline BEFORE the name is watched,
                # so its first gap/RVOL read isn't blank
                try:
                    ingest_daily_history(_scan_con["c"] or rt["research_con"], client,
                                         names, days=30)
                except Exception as exc:  # noqa: BLE001
                    res.errors.append(f"daily backfill failed: {exc}"[:300])
                symbols.extend(names)   # next step_ingest: new-symbol full-session pass
                symbols.trim(int(os.environ.get("WATCHER_MAX_UNIVERSE", "80")))
            if names or res.errors:
                rt["store"].emit(ModuleTickEvent(
                    timestamp=datetime.now(), mode=EventMode.PAPER,
                    correlation_id=rt["session_id"],
                    message=(f"snapshot scan: {res.scanned} scanned, {len(res.hits)} "
                             f"qualify, +{len(names)} new"),
                    module="discovery", stage="snapshot_scan",
                    duration_ms=res.duration_s * 1000.0,
                    input_count=res.scanned, output_count=len(names),
                    metrics={"requests": res.requests, "injected": names,
                             "top": [{"symbol": h.symbol, **h.as_data()}
                                     for h in res.hits[:10]]},
                    errors=[{"error": e} for e in res.errors],
                ))
            if names:
                print(f"[snapscan] +{', '.join(names)} ({res.scanned} scanned, "
                      f"{res.duration_s:.1f}s, {res.requests} req)", flush=True)

        def snapshot_scan_loop():
            try:
                _scan_con["c"] = open_research_db("market")
            except Exception as exc:  # noqa: BLE001
                print(f"[snapscan-thread] dedicated DB unavailable, using shared: {exc}", flush=True)

            class _ScanStep:
                def manage(_self):        # adapt to the generic loop driver
                    step_snapshot_scan()
            _run_scan_loop(_ScanStep(), stop_event, _scan_cfg.interval_s)

        threading.Thread(target=snapshot_scan_loop, name="snapscan", daemon=True).start()
        print(f"[boot] snapshot squeeze scan started (every {_scan_cfg.interval_s:.0f}s, "
              f">= {_scan_cfg.min_pct:g}% , cap {_scan_cfg.max_per_sweep}/sweep, 04:00-16:00 ET)")
    # 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()
        print(f"[boot] news enrichment thread started (every {ollama_cfg.enrichment_interval_seconds}s, model {ollama_cfg.model})")
    # SIGTERM (docker stop) drains through the loop's finally: instead of a
    # mid-pass kill — leaves the book protected by broker-side stop legs.
    signal.signal(signal.SIGTERM, _sigterm_handler)
    try:
        # one immediate full pass, then interval-driven
        run_pass()
        boot_date = _now_session_date()
        while not _clean_exit_requested.is_set():
            # DATE ROLLOVER: a process booted AFTER the EOD flatten never gets an
            # exit-after-flatten, runs overnight, and day-2 takes ZERO entries
            # (every memory-only daily flag — _session_closed, caps, benches — is
            # stale; hit live 2026-07-21: booted 17:11 on the 20th, silently dead
            # at the 21st's open). Exit cleanly at midnight instead; compose
            # restarts a fresh process whose boot recovery rebuilds the new day.
            if _now_session_date() != boot_date:
                print("[stop] session date rolled over — clean restart for the new day")
                _clean_exit_requested.set()
                break
            results = scheduler.run_pending()
            if results:
                stamp = datetime.now().strftime("%H:%M:%S")
                line = " | ".join(
                    f"{k}: {v if not isinstance(v, Exception) else f'ERROR {v}'}"
                    for k, v in results.items()
                )
                print(f"[{stamp}] {line}")
            time.sleep(1.0)
        # clean exit (EOD daily restart or SIGTERM): fall through to finally
        print("[stop] clean exit — shutting down (compose restarts the loop fresh)")
    except KeyboardInterrupt:
        print("\n[stop] shutting down")
    finally:
        stop_event.set()
        if trigger_thread is not None:
            trigger_thread.join(timeout=5)
        if server:
            server.shutdown()
        rt["store"].close()
        rt["research_con"].close()
    return 0


if __name__ == "__main__":
    sys.exit(main())