"""Paper trading in-process service layer.

Replaces subprocess-based task execution (paper_trading.py router).
Provides:
  - make_engine(): Engine factory (mirrors CLI _make_engine)
  - run_task(): Async task runner for engine operations
  - run_all_task(): Run daily for all active sessions
  - task registry: dict backed by asyncio.Lock
  - AutoScheduler: asyncio-based scheduler (mirrors auto.py)
"""
from __future__ import annotations

import asyncio
import contextvars
import datetime as dt
import json
import os
import sys
import traceback
import uuid
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
from zoneinfo import ZoneInfo

# ── Shared task registry ──────────────────────────────────────────────────────

_tasks: dict[str, dict[str, Any]] = {}
_tasks_lock = asyncio.Lock()

# Per-session locks prevent concurrent runs on the same session
_session_locks: dict[str, asyncio.Lock] = {}


def _get_session_lock(session_name: str) -> asyncio.Lock:
    if session_name not in _session_locks:
        _session_locks[session_name] = asyncio.Lock()
    return _session_locks[session_name]


# ── Engine factory ────────────────────────────────────────────────────────────

def make_engine(session: Any, db_path: str) -> Any:
    """Create a PaperTradingEngine for the given session.

    Mirrors CLI _make_engine(). Reads credentials from env vars.
    """
    from apps.paper_trader.alpaca_broker import AlpacaBroker
    from apps.paper_trader.engine import PaperTradingEngine
    from apps.paper_trader.event_detector import EventDetector
    from apps.paper_trader.state import StateManager
    from apps.backtester.run import load_manifest, resolve_config
    from apps.paper_trader.backtest_sim import load_snapshot_store_for_session

    broker = AlpacaBroker.from_env()
    oracle_url = (
        os.environ.get("ORACLE_URL")
        or os.environ.get("STOCK_ORACLE_URL", "http://localhost:8000")
    )
    db_dsn = os.environ.get("DB_DSN") or os.environ.get("POSTGRES_DSN", "")
    state = StateManager(db_path)
    detector = EventDetector(db_dsn=db_dsn, oracle_url=oracle_url)
    manifest = load_manifest(session.config_path)
    config = resolve_config(manifest)
    snapshot_store = load_snapshot_store_for_session(config, oracle_url, db_dsn)
    return PaperTradingEngine(
        session=session, broker=broker, state=state, event_detector=detector,
        snapshot_store=snapshot_store,
    )


# ── Log helpers ───────────────────────────────────────────────────────────────

def _fmt_header(operation: str, session_name: str) -> str:
    ts = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC")
    return f"[{ts}] {operation} — session: {session_name}\n{'─' * 60}\n"


def _fmt_summary(summary: dict[str, Any]) -> str:
    """Format engine return dict into human-readable text."""
    if not summary:
        return "[No summary returned]\n"
    lines: list[str] = []
    for key, value in summary.items():
        if value is None:
            continue
        if isinstance(value, (list, dict)) and not value:
            continue
        if key == "reconciliation":
            # ReconciliationReport dataclass
            for attr in ("orphaned_alpaca", "ghost_local", "reconciled_exits", "stale_orders_cancelled"):
                v = getattr(value, attr, None)
                if v:
                    lines.append(f"  {attr}: {v}")
            continue
        if isinstance(value, dt.date):
            lines.append(f"  {key}: {value.isoformat()}")
        else:
            lines.append(f"  {key}: {value}")
    return "\n".join(lines) + "\n" if lines else "[done]\n"


# ── In-process task runner ────────────────────────────────────────────────────

async def run_task(
    session: Any,
    operation: str,
    db_path: str,
    date: str | None = None,
    force: bool = False,
) -> dict[str, Any]:
    """Create a task and launch the engine operation in-process.

    Returns the initial task dict immediately; the engine runs as an
    asyncio background task.
    """
    from libs.common.logging import bind_job_run_id
    job_run_id = str(uuid.uuid4())
    bind_job_run_id(job_run_id)
    task_id = str(uuid.uuid4())[:8]
    now = datetime.now(timezone.utc).isoformat()
    task: dict[str, Any] = {
        "task_id": task_id,
        "session_name": session.session_name,
        "operation": operation,
        "status": "running",
        "created_at": now,
        "started_at": now,
        "finished_at": None,
        "log": _fmt_header(operation, session.session_name),
        "error": None,
    }
    async with _tasks_lock:
        _tasks[task_id] = task

    asyncio.create_task(
        _execute_engine_task(task_id, session, operation, db_path, date, force, job_run_id)
    )
    return dict(task)


async def run_all_task(sessions: list[Any], db_path: str) -> dict[str, Any]:
    """Run run-daily for every given session as a single task."""
    task_id = str(uuid.uuid4())[:8]
    now = datetime.now(timezone.utc).isoformat()
    now_short = now[:19]
    task: dict[str, Any] = {
        "task_id": task_id,
        "session_name": "all",
        "operation": "run-all",
        "status": "running",
        "created_at": now,
        "started_at": now,
        "finished_at": None,
        "log": f"[{now_short} UTC] run-all — {len(sessions)} session(s)\n{'─' * 60}\n",
        "error": None,
    }
    async with _tasks_lock:
        _tasks[task_id] = task

    asyncio.create_task(_execute_run_all_task(task_id, sessions, db_path))
    return dict(task)


async def _execute_engine_task(
    task_id: str,
    session: Any,
    operation: str,
    db_path: str,
    date: str | None,
    force: bool,
    job_run_id: str | None = None,
) -> None:
    """Background coroutine: acquire session lock, run engine, update task."""
    target_date = dt.date.fromisoformat(date) if date else None
    # Capture context so the thread inherits job_run_id ContextVar
    ctx = contextvars.copy_context()

    async with _get_session_lock(session.session_name):
        extra_log, error = await asyncio.to_thread(
            ctx.run, _run_engine_sync, session, operation, db_path, target_date, force
        )

    finished = datetime.now(timezone.utc).isoformat()
    async with _tasks_lock:
        t = _tasks.get(task_id)
        if t:
            t["log"] = t.get("log", "") + extra_log
            t["finished_at"] = finished
            t["status"] = "failed" if error else "completed"
            t["error"] = error


def _run_engine_sync(
    session: Any,
    operation: str,
    db_path: str,
    target_date: dt.date | None,
    force: bool,
) -> tuple[str, str | None]:
    """Run the engine in a thread with its own event loop.

    Returns (log_text, error_message_or_None).
    """
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    try:
        engine = make_engine(session, db_path)
        if operation == "run":
            coro = engine.run_daily(target_date=target_date, force=force)
        elif operation == "run-close":
            coro = engine.run_reaction_close(target_date=target_date, force=force)
        elif operation == "run-open":
            coro = engine.run_next_open(target_date=target_date, force=force)
        else:
            return f"Unknown operation: {operation}\n", f"Unknown operation: {operation}"
        summary = loop.run_until_complete(coro)
        return _fmt_summary(summary), None
    except Exception as exc:
        tb = traceback.format_exc()
        return f"\nERROR: {type(exc).__name__}: {exc}\n{tb}\n", str(exc)
    finally:
        loop.close()


async def _execute_run_all_task(
    task_id: str,
    sessions: list[Any],
    db_path: str,
) -> None:
    """Background coroutine: run daily for each session sequentially."""
    any_error: str | None = None

    for session in sessions:
        header = f"\n▶ {session.session_name}\n"

        async with _tasks_lock:
            t = _tasks.get(task_id)
            if t:
                t["log"] = t.get("log", "") + header

        extra_log, error = await asyncio.to_thread(
            _run_engine_sync, session, "run", db_path, None, False
        )
        if error:
            any_error = error

        async with _tasks_lock:
            t = _tasks.get(task_id)
            if t:
                t["log"] = t.get("log", "") + extra_log

    finished = datetime.now(timezone.utc).isoformat()
    async with _tasks_lock:
        t = _tasks.get(task_id)
        if t:
            t["finished_at"] = finished
            t["status"] = "failed" if any_error else "completed"
            t["error"] = any_error


# ── Task registry accessors ───────────────────────────────────────────────────

def get_task(task_id: str) -> dict[str, Any] | None:
    return _tasks.get(task_id)


def get_task_log(task_id: str) -> dict[str, Any] | None:
    t = _tasks.get(task_id)
    if t is None:
        return None
    return {"log": t.get("log", ""), "status": t["status"]}


def list_tasks(session_name: str | None = None) -> list[dict[str, Any]]:
    tasks = list(_tasks.values())
    if session_name:
        tasks = [t for t in tasks if t.get("session_name") == session_name]
    tasks.sort(key=lambda t: t.get("created_at", ""), reverse=True)
    return tasks


# ── AutoScheduler ─────────────────────────────────────────────────────────────

_TZ_ET = ZoneInfo("America/New_York")

_SCHEDULE: list[dict[str, Any]] = [
    {"name": "pipeline_pre",  "et_hour": 7,  "et_min": 0,  "kind": "pipeline_pre",
     "label": "Pre-market pipeline"},
    {"name": "run_open",      "et_hour": 9,  "et_min": 35, "kind": "run_open",
     "label": "run-open (장 시작 직후)"},
    {"name": "run_close",     "et_hour": 15, "et_min": 45, "kind": "run_close",
     "label": "run-close (장 마감 직전)"},
    {"name": "pipeline_post", "et_hour": 16, "et_min": 30, "kind": "pipeline_post",
     "label": "Post-close pipeline"},
]

_PIPELINE_CMDS: list[list[str]] = [
    [sys.executable, "-m", "apps.pipeline.filing_poller.main"],
    [sys.executable, "-m", "apps.pipeline.filing_fetcher.main"],
    [sys.executable, "-m", "apps.pipeline.event_parser.main"],
    [sys.executable, "-m", "apps.pipeline.feature_builder.main"],
    [sys.executable, "-m", "apps.pipeline.label_generator.main"],
    [sys.executable, "-m", "apps.pipeline.label_generator.main",
     "--entry-convention", "reaction_close"],
]

_POST_PIPELINE_CMDS: list[list[str]] = [
    [sys.executable, "-m", "apps.pipeline.filing_poller.main"],
    [sys.executable, "-m", "apps.pipeline.filing_fetcher.main"],
    [sys.executable, "-m", "apps.pipeline.event_parser.main"],
    [sys.executable, "-m", "apps.pipeline.feature_builder.main"],
    [sys.executable, "-m", "apps.pipeline.label_generator.main",
     "--entry-convention", "reaction_close"],
    [sys.executable, "-m", "apps.pipeline.label_generator.main"],
]


def _state_file_path(db_path: str) -> Path:
    """Return path to the auto scheduler persistence file."""
    return Path(db_path).parent / ".paper_auto_state.json"


def load_saved_state(db_path: str) -> "dict[str, Any] | None":
    """Load previously persisted scheduler state. Returns None if absent/invalid."""
    sf = _state_file_path(db_path)
    if sf.exists():
        try:
            return json.loads(sf.read_text())
        except Exception:
            pass
    return None


class AutoScheduler:
    """In-process auto scheduler. Runs inside FastAPI; no separate daemon process needed.

    Pipeline steps remain subprocesses (via asyncio.create_subprocess_exec).
    Trading operations use PaperTradingEngine in-process.
    State is persisted to disk so FastAPI restart can auto-resume.
    """

    def __init__(self) -> None:
        self._task: asyncio.Task | None = None  # type: ignore[type-arg]
        self._sessions: list[str] = []
        self._dry_run: bool = False
        self._db_path: str = ""
        self._log_lines: list[str] = []
        self._completed: set[str] = set()

    # ── Public interface ───────────────────────────────────────────────────────

    @property
    def running(self) -> bool:
        return self._task is not None and not self._task.done()

    def start(self, sessions: list[str], dry_run: bool, db_path: str) -> None:
        if self.running:
            raise RuntimeError("AutoScheduler already running")
        self._sessions = sessions
        self._dry_run = dry_run
        self._db_path = db_path
        self._log_lines = []
        self._completed = set()
        self._save_state()
        self._task = asyncio.create_task(self._run_loop())

    def stop(self) -> None:
        """User-initiated stop. Cancels task and clears persisted state."""
        if self._task and not self._task.done():
            self._task.cancel()
        self._clear_state()

    def shutdown(self) -> None:
        """Server shutdown stop. Cancels task but keeps persisted state for auto-restart."""
        if self._task and not self._task.done():
            self._task.cancel()

    def _save_state(self) -> None:
        if not self._db_path:
            return
        try:
            _state_file_path(self._db_path).write_text(json.dumps({
                "running": True,
                "sessions": self._sessions,
                "dry_run": self._dry_run,
                "db_path": self._db_path,
            }))
        except Exception:
            pass

    def _clear_state(self) -> None:
        if not self._db_path:
            return
        try:
            sf = _state_file_path(self._db_path)
            if sf.exists():
                sf.unlink()
        except Exception:
            pass

    def get_log(self, lines: int = 200) -> str:
        return "\n".join(self._log_lines[-lines:])

    def get_log_tail(self, lines: int = 80) -> list[str]:
        return self._log_lines[-lines:]

    @property
    def log_line_count(self) -> int:
        return len(self._log_lines)

    # ── Helpers ────────────────────────────────────────────────────────────────

    def _emit_lifecycle(self, event_name: str, phase: str, job_run_id: str, **kwargs: Any) -> None:
        try:
            from apps.web.services.events_store import EventsStore
            payload: dict[str, Any] = {
                "ts_utc": datetime.now(timezone.utc).isoformat(),
                "source": "auto_scheduler",
                "level": "INFO",
                "event": event_name,
                "category": "lifecycle",
                "job_run_id": job_run_id,
                "phase": phase,
                "sessions": self._sessions,
            }
            payload.update(kwargs)
            EventsStore.get().write(payload)
        except Exception:
            pass

    def _log(self, msg: str) -> None:
        ts = datetime.now(timezone.utc).strftime("%H:%M UTC")
        self._log_lines.append(f"{ts}  {msg}")
        # Mirror to EventsStore so log messages survive server restarts and are queryable
        try:
            from apps.web.services.events_store import EventsStore
            EventsStore.get().write({
                "ts_utc": datetime.now(timezone.utc).isoformat(),
                "source": "auto_scheduler",
                "level": "WARN" if any(w in msg for w in ("ERROR", "FAILED", "✗")) else "INFO",
                "event": "scheduler_log",
                "message": msg,
                "category": "lifecycle",
            })
        except Exception:
            pass

    def _now_et(self) -> dt.datetime:
        return dt.datetime.now(tz=_TZ_ET)

    def _et_dt_for(self, date: dt.date, ev: dict[str, Any]) -> dt.datetime:
        return dt.datetime(
            date.year, date.month, date.day,
            ev["et_hour"], ev["et_min"], tzinfo=_TZ_ET,
        )

    def _is_trading_day(self, date: dt.date) -> bool:
        try:
            from libs.common.time_utils import is_trading_day
            return is_trading_day(date)
        except Exception:
            return date.weekday() < 5  # Mon–Fri fallback

    def _next_trading_day(self, from_date: dt.date) -> dt.date:
        check = from_date + dt.timedelta(days=1)
        for _ in range(14):
            if self._is_trading_day(check):
                return check
            check += dt.timedelta(days=1)
        raise RuntimeError("No trading day found within 14 days")

    def _prev_trading_day(self, from_date: dt.date) -> dt.date:
        check = from_date - dt.timedelta(days=1)
        for _ in range(14):
            if self._is_trading_day(check):
                return check
            check -= dt.timedelta(days=1)
        raise RuntimeError("No previous trading day within 14 days")

    def _get_active_sessions(self) -> list[str]:
        try:
            from apps.paper_trader.state import StateManager
            return [
                s.session_name
                for s in StateManager(self._db_path).list_sessions()
                if s.status == "active"
            ]
        except Exception:
            return []

    @staticmethod
    def _fmt_countdown(seconds: float) -> str:
        if seconds <= 0:
            return "now"
        h = int(seconds // 3600)
        m = int((seconds % 3600) // 60)
        s = int(seconds % 60)
        if h > 0:
            return f"{h}h {m:02d}m"
        if m > 0:
            return f"{m}m {s:02d}s"
        return f"{s}s"

    # ── Pipeline execution (subprocess) ───────────────────────────────────────

    async def _run_pipeline(self, cmds: list[list[str]], phase_run_id: str = "") -> bool:
        project_root = str(Path(__file__).parent.parent.parent)
        sub_env = {**os.environ, "JOB_RUN_ID": phase_run_id} if phase_run_id else None
        for cmd in cmds:
            label = " ".join(cmd[2:] if cmd[:2] == [sys.executable, "-m"] else cmd)
            if self._dry_run:
                self._log(f"[DRY] {label}")
                continue
            self._log(f"  ▶ {label}")
            try:
                proc = await asyncio.create_subprocess_exec(
                    *cmd,
                    stdout=asyncio.subprocess.PIPE,
                    stderr=asyncio.subprocess.STDOUT,
                    cwd=project_root,
                    env=sub_env,
                )
                stdout, _ = await proc.communicate()
                ok = proc.returncode == 0
                self._log(f"  {'OK' if ok else 'FAILED'} (exit {proc.returncode})")
                # Forward all pipeline JSON log lines to EventsStore; surface notable ones to _log_lines
                try:
                    from apps.web.services.events_store import EventsStore, _infer_category
                    _store = EventsStore.get()
                except Exception:
                    _store = None
                for line in (stdout or b"").decode("utf-8", errors="replace").splitlines():
                    line = line.strip()
                    if not line:
                        continue
                    try:
                        rec = json.loads(line)
                        lvl = rec.get("level", "")
                        evt = rec.get("event", "")
                        # Forward to EventsStore (all levels)
                        if _store is not None:
                            rec["source"] = "pipeline"
                            rec.setdefault("job_run_id", phase_run_id)  # defensive: subprocess may not call configure_logging
                            _store.write(rec)
                        # Surface errors and fallback events to the in-memory user log
                        cat = _infer_category(evt, rec.get("category")) if _store is not None else None
                        if lvl in ("error", "critical"):
                            self._log(f"  ERROR: {evt} {rec}")
                        elif cat == "fallback" or (isinstance(cat, str) and "fallback" in evt.lower()):
                            self._log(f"  FALLBACK: {evt}")
                    except Exception:
                        pass
                if not ok:
                    self._log(f"  PIPELINE HALTED: {label} failed — downstream steps skipped")
                    return False
            except Exception as exc:
                self._log(f"  ERROR running {label}: {exc}")
                return False
        return True

    # ── Trading execution (in-process) ────────────────────────────────────────

    async def _run_trading(self, operation: str, sessions: list[str]) -> None:
        from apps.paper_trader.state import StateManager

        state_mgr = StateManager(self._db_path)
        for session_name in sessions:
            session = state_mgr.get_session(session_name)
            if session is None or session.status != "active":
                self._log(f"  {session_name}: skipped (not active)")
                continue
            if self._dry_run:
                self._log(f"  [DRY] {operation} — {session_name}")
                continue
            self._log(f"  ▶ {operation} — {session_name}")
            try:
                _ctx = contextvars.copy_context()

                def _run_sync(s: Any = session) -> dict[str, Any]:
                    loop = asyncio.new_event_loop()
                    asyncio.set_event_loop(loop)
                    try:
                        engine = make_engine(s, self._db_path)
                        if operation == "run-open":
                            coro = engine.run_next_open()
                        elif operation == "run-close":
                            coro = engine.run_reaction_close()
                        else:
                            raise ValueError(f"Unknown operation: {operation}")
                        return loop.run_until_complete(coro)
                    finally:
                        loop.close()

                summary = await asyncio.to_thread(_ctx.run, _run_sync)
                status = summary.get("status", "?")
                self._log(f"  ✓ {session_name}: {status}")
            except Exception as exc:
                self._log(f"  ERROR {session_name}: {type(exc).__name__}: {exc}")

    # ── Snapshot refresh after pipeline_post ─────────────────────────────────

    async def _refresh_active_snapshots(self) -> None:
        """Incremental snapshot refresh after pipeline_post completes."""
        from apps.paper_trader.state import StateManager
        from apps.backtester.run import load_manifest, resolve_config
        from libs.export.canonical_snapshots import incremental_update_canonical_snapshot

        state_mgr = StateManager(self._db_path)
        seen: set[str] = set()
        for s in self._sessions:
            session = state_mgr.get_session(s)
            if session is None:
                continue
            try:
                manifest = load_manifest(session.config_path)
                config = resolve_config(manifest)
                sid = getattr(config, "dataset_snapshot_id", None)
                if not sid or sid in seen:
                    continue
                seen.add(sid)
                self._log(f"  Refreshing snapshot: {sid}")
                await incremental_update_canonical_snapshot(sid)
                self._log(f"  ✓ Snapshot refreshed: {sid}")
            except Exception as exc:
                self._log(f"  Snapshot refresh error ({s}): {exc}")

    # ── Catch-up on startup ───────────────────────────────────────────────────

    async def _run_catchup(self) -> None:
        """Run missed pipeline steps and paper trading phases on startup."""
        now_et = self._now_et()
        today = now_et.date()
        catchup_items: list[tuple[str, list[list[str]]]] = []

        prev_td = self._prev_trading_day(today)
        prev_post_close_et = self._et_dt_for(prev_td, _SCHEDULE[3])  # pipeline_post 16:30
        if prev_post_close_et < now_et:
            catchup_items.append((f"Post-close pipeline ({prev_td})", _POST_PIPELINE_CMDS))

        if self._is_trading_day(today):
            pre_market_et = self._et_dt_for(today, _SCHEDULE[0])  # pipeline_pre 07:00
            if pre_market_et < now_et:
                catchup_items.append((f"Pre-market pipeline ({today})", _PIPELINE_CMDS))

        if catchup_items:
            self._log("━━━ Catch-up: 놓친 파이프라인 실행 ━━━")
            for label, cmds in catchup_items:
                self._log(f"▶ {label}")
                catchup_run_id = str(uuid.uuid4())
                await self._run_pipeline(cmds, catchup_run_id)
            self._log("━━━ Catch-up 완료 ━━━")

        # Paper trading phase catchup: run missed open/close phases if their window has passed
        # but the phase hasn't been recorded in processed_phases. Prevents a crash/restart from
        # silently skipping a trading phase because its scheduled time has already passed.
        if self._is_trading_day(today) and not self._dry_run:
            sessions = self._sessions or self._get_active_sessions()
            if sessions and self._db_path:
                from apps.paper_trader.state import StateManager
                state_mgr = StateManager(self._db_path)

                run_open_et = self._et_dt_for(today, _SCHEDULE[1])   # run_open 9:35
                run_close_et = self._et_dt_for(today, _SCHEDULE[2])  # run_close 15:45
                market_close_et = now_et.replace(hour=16, minute=0, second=0, microsecond=0)
                cutoff_et = now_et.replace(hour=20, minute=0, second=0, microsecond=0)

                # run_open catchup: fired if 9:35 ≤ now < 16:00 and "next_open" not in phases
                if run_open_et <= now_et < market_close_et:
                    already_ran = any(
                        (sess := state_mgr.get_session(s)) is not None
                        and state_mgr.is_phase_processed(sess.session_id, today, "next_open")
                        for s in sessions
                    )
                    if not already_ran:
                        self._log("━━━ Catch-up: run_open 누락 (9:35 ET 이후 시작) — 지금 실행 ━━━")
                        await self._run_trading("run-open", sessions)
                        self._completed.add("run_open")
                        self._log("━━━ Catch-up run_open 완료 ━━━")

                # run_close catchup: fired if 15:45 ≤ now < 20:00 and "reaction_close" not in phases
                if run_close_et <= now_et < cutoff_et:
                    already_ran = any(
                        (sess := state_mgr.get_session(s)) is not None
                        and state_mgr.is_phase_processed(sess.session_id, today, "reaction_close")
                        for s in sessions
                    )
                    if not already_ran:
                        self._log("━━━ Catch-up: run_close 누락 (15:45 ET 이후 시작) — 지금 실행 ━━━")
                        await self._run_trading("run-close", sessions)
                        self._completed.add("run_close")
                        self._log("━━━ Catch-up run_close 완료 ━━━")

    # ── Main scheduler loop ───────────────────────────────────────────────────

    async def _run_loop(self) -> None:
        """Main scheduler loop. Mirrors auto.py run_auto()."""
        resolved = self._sessions or self._get_active_sessions()
        if not resolved:
            self._log("No active sessions found. Stopping.")
            return

        self._log(f"Auto scheduler started — sessions: {', '.join(resolved)}")
        if self._dry_run:
            self._log("DRY RUN — commands will not execute")

        try:
            await self._run_catchup()
        except Exception as exc:
            self._log(f"Catch-up error: {exc}")

        last_schedule_date: dt.date | None = None

        try:
            while True:
                now_et = self._now_et()
                today = now_et.date()

                # New day → reset completed set and refresh session list
                if last_schedule_date != today:
                    self._completed.clear()
                    last_schedule_date = today

                    # Re-read active sessions from DB so new sessions are picked up
                    # and removed/closed sessions are dropped automatically.
                    if not self._sessions:  # only auto-refresh if not pinned via --session
                        fresh = self._get_active_sessions()
                        if fresh != resolved:
                            added = set(fresh) - set(resolved)
                            removed = set(resolved) - set(fresh)
                            if added:
                                self._log(f"Sessions added: {', '.join(sorted(added))}")
                            if removed:
                                self._log(f"Sessions removed: {', '.join(sorted(removed))}")
                            resolved = fresh
                        if not resolved:
                            self._log("No active sessions. Waiting for next day.")

                    self._log(f"━━━ {today.strftime('%a %Y-%m-%d')} — sessions: {', '.join(resolved) or 'none'} ━━━")

                    if not self._is_trading_day(today):
                        next_td = self._next_trading_day(today)
                        self._log(f"Non-trading day. Next trading day: {next_td}")
                    else:
                        # Mark already-past events as skipped
                        for ev in _SCHEDULE:
                            if self._et_dt_for(today, ev) <= now_et:
                                self._completed.add(ev["name"])
                                self._log(f"Skipping past event: {ev['label']}")

                if not self._is_trading_day(today):
                    await asyncio.sleep(1800)
                    continue

                pending = [ev for ev in _SCHEDULE if ev["name"] not in self._completed]
                if not pending:
                    next_td = self._next_trading_day(today)
                    first = _SCHEDULE[0]
                    wake_et = self._et_dt_for(next_td, first)
                    wait = (wake_et - now_et).total_seconds()
                    self._log(
                        f"All done today. Sleeping until "
                        f"{wake_et.strftime('%I:%M %p ET')} on {next_td} "
                        f"— {self._fmt_countdown(wait)}"
                    )
                    await asyncio.sleep(min(wait, 3600))
                    continue

                next_ev = pending[0]
                next_et = self._et_dt_for(today, next_ev)
                wait = (next_et - now_et).total_seconds()

                if wait > 90:
                    self._log(
                        f"Next: {next_ev['label']} at "
                        f"{next_et.strftime('%I:%M %p ET')} "
                        f"— {self._fmt_countdown(wait)}"
                    )
                    # Re-check every 10 min at most
                    await asyncio.sleep(min(wait - 60, 600))
                    continue

                if wait > 0:
                    self._log(
                        f"Firing in {self._fmt_countdown(wait)}: {next_ev['label']}"
                    )
                    await asyncio.sleep(wait)

                # ── Execute event ──────────────────────────────────────────
                self._log(f"▶ {next_ev['label']}")
                pipeline_ok = True
                phase_run_id = str(uuid.uuid4())
                phase_start = datetime.now(timezone.utc)
                try:
                    from libs.common.logging import bind_job_run_id
                    bind_job_run_id(phase_run_id)
                except Exception:
                    pass
                self._emit_lifecycle("phase_started", next_ev["kind"], phase_run_id)
                try:
                    kind = next_ev["kind"]
                    if kind == "pipeline_pre":
                        pipeline_ok = await self._run_pipeline(_PIPELINE_CMDS, phase_run_id)
                        if pipeline_ok:
                            await self._refresh_active_snapshots()
                    elif kind == "run_open":
                        await self._run_trading("run-open", resolved)
                    elif kind == "run_close":
                        await self._run_trading("run-close", resolved)
                    elif kind == "pipeline_post":
                        pipeline_ok = await self._run_pipeline(_POST_PIPELINE_CMDS, phase_run_id)
                        if pipeline_ok:
                            await self._refresh_active_snapshots()
                except Exception as exc:
                    self._log(f"ERROR executing {next_ev['name']}: {exc}")
                    pipeline_ok = False

                phase_status = "success" if pipeline_ok else "failed"
                duration_s = (datetime.now(timezone.utc) - phase_start).total_seconds()
                self._emit_lifecycle("phase_completed", next_ev["kind"], phase_run_id,
                                     status=phase_status, duration_s=round(duration_s, 1))

                if not pipeline_ok:
                    self._log(f"✗ FAILED: {next_ev['label']} — NOT marking completed (will not retry today)")
                else:
                    self._completed.add(next_ev["name"])
                    self._log(f"✓ Done: {next_ev['label']}")

        except asyncio.CancelledError:
            self._log("Auto scheduler stopped.")
            raise


# ── Module-level singleton ────────────────────────────────────────────────────

auto_scheduler = AutoScheduler()