deerflow-code/offline-backend-20260512/backend/app/gateway/app.py

import asyncio
import logging
import os
import queue
import sys
import threading
from collections.abc import AsyncGenerator
from contextlib import asynccontextmanager
from datetime import datetime
from time import perf_counter

from dotenv import load_dotenv
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

# Load .env BEFORE any module that reads os.environ at import time.
# Without this, GATEWAY_CORS_ORIGINS / AUTH_JWT_SECRET / DEER_FLOW_AUTH_DISABLED
# placed in /workspace/.env are silently ignored when the container is started
# without explicit `-e` flags, leading to ephemeral JWT secrets, missing CORS
# origins, and "Failed to fetch" in the browser.
load_dotenv()

from app.gateway.auth.disabled_mode import is_auth_disabled  # noqa: E402
from app.gateway.auth_middleware import AuthMiddleware
from app.gateway.config import get_gateway_config
from app.gateway.csrf_middleware import CSRFMiddleware
from app.gateway.deps import langgraph_runtime
from app.gateway.routers import (
    ai_writing,
    admin_users,
    agents,
    artifacts,
    assistants_compat,
    auth,
    browser_context,
    business_mapping,
    channels,
    checkpoint_migration,
    feedback,
    html_page_favorites,
    intent,
    knowledge,
    knowledge_ingest,
    light_apps,
    llm_metrics,
    mcp,
    memory,
    menu_overrides,
    models,
    multi_agent,
    notifications,
    parallel_agent_tasks,
    parallel_agents,
    positions,
    public_config,
    public_embed,
    public_leaderboard,
    public_scheduled_tasks,
    recommend,
    recommended_questions,
    roundtable_chains,
    roundtable_draft_shares,
    roundtable_drafts,
    roundtable_jobs,
    runs,
    scheduled_tasks,
    sensitive_words,
    skills,
    suggestions,
    system_settings,
    tags,
    task_buttons,
    thread_runs,
    tool_metrics,
    thread_shares,
    threads,
    uploads,
    user_preferences,
    user_prompts,
    writing,
)
from deerflow.config import app_config as deerflow_app_config
from deerflow.config.app_config import apply_logging_level
from deerflow.runtime.scheduler import ScheduledTaskService, set_scheduled_task_service

AppConfig = deerflow_app_config.AppConfig
get_app_config = deerflow_app_config.get_app_config

# Default logging; lifespan overrides from config.yaml log_level.
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
)

logger = logging.getLogger(__name__)

# Upper bound (seconds) each lifespan shutdown hook is allowed to run.
# Bounds worker exit time so uvicorn's reload supervisor does not keep
# firing signals into a worker that is stuck waiting for shutdown cleanup.
_SHUTDOWN_HOOK_TIMEOUT_SECONDS = 5.0


async def _ensure_admin_user(app: FastAPI) -> None:
    """Startup hook: handle first boot and migrate orphan threads otherwise.

    After admin creation, migrate orphan threads from the LangGraph
    store (metadata.user_id unset) to the admin account. This is the
    "no-auth ? with-auth" upgrade path: users who ran DeerFlow without
    authentication have existing LangGraph thread data that needs an
    owner assigned.
        First boot (no admin exists):
            - Does NOT create any user accounts automatically.
            - The operator must visit ``/setup`` to create the first admin.

    Subsequent boots (admin already exists):
      - Runs the one-time "no-auth ? with-auth" orphan thread migration for
        existing LangGraph thread metadata that has no owner_id.

    No SQL persistence migration is needed: the four user_id columns
    (threads_meta, runs, run_events, feedback) only come into existence
    alongside the auth module via create_all, so freshly created tables
    never contain NULL-owner rows.
    """
    from sqlalchemy import select

    from app.gateway.deps import get_local_provider
    from deerflow.persistence.engine import get_session_factory
    from deerflow.persistence.user.model import UserRow

    try:
        provider = get_local_provider()
    except RuntimeError:
        # Auth persistence may not be initialized in some test/boot paths.
        # Skip admin migration work rather than failing gateway startup.
        logger.warning("Auth persistence not ready; skipping admin bootstrap check")
        return

    sf = get_session_factory()
    if sf is None:
        return

    admin_count = await provider.count_admin_users()

    if admin_count == 0:
        logger.info("=" * 60)
        logger.info("  First boot detected ? no admin account exists.")
        logger.info("  Visit /setup to complete admin account creation.")
        logger.info("=" * 60)
        return

    # Admin already exists ? run orphan thread migration for any
    # LangGraph thread metadata that pre-dates the auth module.
    async with sf() as session:
        stmt = select(UserRow).where(UserRow.system_role == "admin").limit(1)
        row = (await session.execute(stmt)).scalar_one_or_none()

    if row is None:
        return  # Should not happen (admin_count > 0 above), but be safe.

    admin_id = str(row.id)

    # LangGraph store orphan migration ? non-fatal.
    # This covers the "no-auth ? with-auth" upgrade path for users
    # whose existing LangGraph thread metadata has no user_id set.
    store = getattr(app.state, "store", None)
    if store is not None:
        try:
            migrated = await _migrate_orphaned_threads(store, admin_id)
            if migrated:
                logger.info("Migrated %d orphan LangGraph thread(s) to admin", migrated)
        except Exception:
            logger.exception("LangGraph thread migration failed (non-fatal)")


async def _iter_store_items(store, namespace, *, page_size: int = 500):
    """Paginated async iterator over a LangGraph store namespace.

    Replaces the old hardcoded ``limit=1000`` call with a cursor-style
    loop so that environments with more than one page of orphans do
    not silently lose data. Terminates when a page is empty OR when a
    short page arrives (indicating the last page).
    """
    offset = 0
    while True:
        batch = await store.asearch(namespace, limit=page_size, offset=offset)
        if not batch:
            return
        for item in batch:
            yield item
        if len(batch) < page_size:
            return
        offset += page_size


async def _sync_legacy_skills(app: FastAPI) -> None:
    """Idempotently register existing on-disk ``skills/custom/*`` directories
    in the ``skills`` DB table, owned by the first admin and published=True.

    Without this step, after upgrading to per-user skill ownership the existing
    custom skill folders would not be visible to anyone (the SkillStore
    visibility filter relies on a DB row). Running this once on every boot is
    cheap (folder scan + one ``ensure_legacy`` per name) and idempotent.

    Non-fatal: a failure here MUST NOT block server startup.
    """
    from sqlalchemy import select

    from deerflow.persistence.engine import get_session_factory
    from deerflow.persistence.user.model import UserRow
    from deerflow.skills.storage import get_or_new_skill_storage
    from deerflow.skills.types import SkillCategory

    sf = get_session_factory()
    if sf is None:
        return

    skill_store = getattr(app.state, "skill_store", None)
    if skill_store is None:
        return

    async with sf() as session:
        stmt = select(UserRow).where(UserRow.system_role == "admin").limit(1)
        row = (await session.execute(stmt)).scalar_one_or_none()
    if row is None:
        # No admin yet ? leave ownership null; first admin is responsible for
        # claiming via the UI once they exist.
        admin_id: str | None = None
    else:
        admin_id = str(row.id)

    config = getattr(app.state, "config", None)
    storage = get_or_new_skill_storage(app_config=config)
    try:
        skills = storage.load_skills(enabled_only=False)
    except Exception:
        logger.exception("_sync_legacy_skills: could not list on-disk skills")
        return

    inserted = 0
    for skill in skills:
        if skill.category != SkillCategory.CUSTOM:
            continue
        try:
            existing = await skill_store.get_any(skill.name)
            if existing is not None:
                continue
            await skill_store.ensure_legacy(
                {"name": skill.name, "owner_user_id": admin_id, "published": True}
            )
            inserted += 1
        except Exception:
            logger.warning("_sync_legacy_skills: could not register %s", skill.name, exc_info=True)
    if inserted:
        logger.info("Skill ownership migration: registered %d legacy custom skill(s)", inserted)


async def _migrate_orphaned_threads(store, admin_user_id: str) -> int:
    """Migrate LangGraph store threads with no user_id to the given admin.

    Uses cursor pagination so all orphans are migrated regardless of
    count. Returns the number of rows migrated.
    """
    migrated = 0
    async for item in _iter_store_items(store, ("threads",)):
        metadata = item.value.get("metadata", {})
        if not metadata.get("user_id"):
            metadata["user_id"] = admin_user_id
            item.value["metadata"] = metadata
            await store.aput(("threads",), item.key, item.value)
            migrated += 1
    return migrated


async def _run_db_migrations() -> None:
    """Run Alembic migrations at startup so schema changes are applied automatically."""
    from pathlib import Path

    from alembic import command
    from alembic.config import Config

    import deerflow.persistence as _persistence_pkg
    from deerflow.persistence.engine import get_engine

    engine = get_engine()
    if engine is None:
        logger.info("Persistence backend=memory ? skipping Alembic migrations")
        return

    if os.getenv("DEER_FLOW_SKIP_ALEMBIC", "").lower() in {"1", "true", "yes"}:
        logger.warning("DEER_FLOW_SKIP_ALEMBIC is set; skipping Alembic migrations")
        return

    db_url = engine.url.render_as_string(hide_password=False).replace("%", "%%")
    migrations_dir = str(Path(_persistence_pkg.__file__).parent / "migrations")
    result_queue: queue.Queue[BaseException | None] = queue.Queue(maxsize=1)

    def _run_sync() -> None:
        try:
            cfg = Config()
            cfg.set_main_option("script_location", migrations_dir)
            cfg.set_main_option("sqlalchemy.url", db_url)
            command.upgrade(cfg, "head")
        except BaseException as exc:
            result_queue.put(exc)
        else:
            result_queue.put(None)

    try:
        try:
            timeout_s = float(os.getenv("DEER_FLOW_ALEMBIC_TIMEOUT", "60"))
        except ValueError:
            timeout_s = 60.0

        worker = threading.Thread(target=_run_sync, name="deerflow-alembic-upgrade", daemon=True)
        worker.start()
        try:
            deadline = asyncio.get_running_loop().time() + timeout_s
            while result_queue.empty():
                if asyncio.get_running_loop().time() >= deadline:
                    raise asyncio.TimeoutError
                await asyncio.sleep(0.2)
            result = result_queue.get_nowait()
        except asyncio.TimeoutError:
            logger.error(
                "Database migration did not finish within %.0fs; continuing startup on create_all schema. "
                "Likely cause: MySQL metadata-lock wait or a slow DDL. Set DEER_FLOW_SKIP_ALEMBIC=1 "
                "to skip startup migrations, or inspect MySQL with SHOW FULL PROCESSLIST.",
                timeout_s,
            )
            return
        if result is not None:
            raise result
        logger.info("Database migrations applied (head)")
    except Exception:
        logger.exception("Database migration failed ? schema may be out of date")


# Skill curator ?????
# - ????:?????????"??????";??????? should_run_now() ???
# - ????:?????????????,???????
_CURATOR_CHECK_INTERVAL_SECONDS = 3600  # 1 ??
_CURATOR_STARTUP_DELAY_SECONDS = 300  # 5 ??


def _acquire_singleton_lock(lock_path):
    """???????????,??? worker ?? leader ???

    ?? -> ??????????(?????????,??????/????);
    ????(???? worker ??)????? -> ?? None?
    ??????? OS ????????
    """
    try:
        lock_path.parent.mkdir(parents=True, exist_ok=True)
        handle = open(lock_path, "a+")
    except OSError:
        logger.exception("Curator scheduler: ???????")
        return None
    try:
        if sys.platform == "win32":
            import msvcrt

            handle.seek(0)
            msvcrt.locking(handle.fileno(), msvcrt.LK_NBLCK, 1)
        else:
            import fcntl

            fcntl.flock(handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
    except OSError:
        handle.close()
        return None
    return handle


def _release_singleton_lock(handle) -> None:
    """????? _acquire_singleton_lock ?????????"""
    try:
        if sys.platform == "win32":
            import msvcrt

            handle.seek(0)
            msvcrt.locking(handle.fileno(), msvcrt.LK_UNLCK, 1)
        else:
            import fcntl

            fcntl.flock(handle.fileno(), fcntl.LOCK_UN)
    except OSError:
        pass
    finally:
        try:
            handle.close()
        except OSError:
            pass


async def _curator_scheduler_loop() -> None:
    """??????:?? _CURATOR_CHECK_INTERVAL_SECONDS ???????????

    ? worker ???????? leader ?? ?? ???????? worker ????
    ????,?? worker ????,?? curator ?????????? LLM?
    ? cancel ? CancelledError ??? ``except Exception``(?? BaseException
    ??,?????),?????????? finally ?????
    """
    from deerflow.config.runtime_paths import runtime_home
    from deerflow.skills.curator import maybe_run_curator

    lock_handle = _acquire_singleton_lock(runtime_home() / ".curator_scheduler.lock")
    if lock_handle is None:
        logger.info("Curator scheduler: ??? worker ??????,? worker ??")
        return

    logger.info("Curator scheduler: ?????,? worker ?? curator ??")
    try:
        # ?????????????,??????
        await asyncio.sleep(_CURATOR_STARTUP_DELAY_SECONDS)
        while True:
            try:
                await maybe_run_curator()
            except Exception:
                logger.exception("Curator scheduler tick failed (non-fatal)")
            await asyncio.sleep(_CURATOR_CHECK_INTERVAL_SECONDS)
    finally:
        _release_singleton_lock(lock_handle)


def _env_bool(name: str, default: bool) -> bool:
    raw = os.getenv(name)
    if raw is None:
        return default
    return raw.strip().lower() not in {"0", "false", "no", "off"}


def _env_int(name: str, default: int, *, alias: str | None = None, minimum: int = 1) -> int:
    raw = os.getenv(name)
    if raw is None and alias:
        raw = os.getenv(alias)
    if raw is None:
        raw = str(default)
    try:
        value = int(raw)
    except ValueError:
        return default
    return max(minimum, value)


def _env_float(name: str, default: float, *, alias: str | None = None, minimum: float = 0.0) -> float:
    raw = os.getenv(name)
    if raw is None and alias:
        raw = os.getenv(alias)
    if raw is None:
        raw = str(default)
    try:
        value = float(raw)
    except ValueError:
        return default
    return max(minimum, value)


_LEADERBOARD_SNAPSHOT_BACKGROUND_ENABLED = _env_bool("ADMIN_LEADERBOARD_BACKGROUND_ENABLED", True)
_LEADERBOARD_SNAPSHOT_PREWARM_DAYS = _env_int(
    "ADMIN_LEADERBOARD_BACKGROUND_PREWARM_DAYS",
    30,
)
_LEADERBOARD_SNAPSHOT_BATCH_SIZE = _env_int(
    "ADMIN_LEADERBOARD_BACKGROUND_BATCH_SIZE",
    1,
)
_LEADERBOARD_SNAPSHOT_INTERVAL_SECONDS = _env_float(
    "ADMIN_LEADERBOARD_BACKGROUND_INTERVAL_SECONDS",
    300.0,
)
_LEADERBOARD_SNAPSHOT_SLEEP_SECONDS = _env_float(
    "ADMIN_LEADERBOARD_BACKGROUND_SLEEP_SECONDS",
    2.0,
)
_LEADERBOARD_SNAPSHOT_RUNNING_TTL_SECONDS = _env_int(
    "ADMIN_LEADERBOARD_BACKGROUND_RUNNING_TTL_SECONDS",
    1800,
)
_LEADERBOARD_SNAPSHOT_STARTUP_DELAY_SECONDS = _env_float(
    "ADMIN_LEADERBOARD_BACKGROUND_STARTUP_DELAY_SECONDS",
    60.0,
)
# Hour (local Beijing time, 0-23) after which the scheduler force-recomputes the
# *previous* day's snapshot once, so records that landed near midnight are fully
# captured. Default 1 ? "???? 1 ????????".
_LEADERBOARD_FINALIZE_HOUR = min(
    23,
    _env_int("ADMIN_LEADERBOARD_FINALIZE_HOUR", 1, minimum=0),
)


def _leaderboard_scheduler_sql_for_log(stmt) -> str:
    try:
        return str(stmt.compile(compile_kwargs={"literal_binds": True}))
    except Exception:
        return str(stmt)


async def _execute_leaderboard_scheduler_query(session, step: str, stmt, **context):
    sql = _leaderboard_scheduler_sql_for_log(stmt)
    logger.info(
        "Leaderboard snapshot DB query start: step=%s stat_date=%s job=%s settings_hash=%s started_at=%s sql=%s",
        step,
        context.get("stat_date") or context.get("stat_dates") or "",
        context.get("job_key") or "",
        context.get("settings_hash") or "",
        datetime.now().isoformat(),
        sql,
    )
    start = perf_counter()
    try:
        result = await session.execute(stmt)
    except Exception:
        logger.exception(
            "Leaderboard snapshot DB query failed: step=%s stat_date=%s job=%s settings_hash=%s elapsed_ms=%.2f",
            step,
            context.get("stat_date") or context.get("stat_dates") or "",
            context.get("job_key") or "",
            context.get("settings_hash") or "",
            (perf_counter() - start) * 1000,
        )
        raise
    logger.info(
        "Leaderboard snapshot DB query done: step=%s stat_date=%s job=%s settings_hash=%s finished_at=%s elapsed_ms=%.2f",
        step,
        context.get("stat_date") or context.get("stat_dates") or "",
        context.get("job_key") or "",
        context.get("settings_hash") or "",
        datetime.now().isoformat(),
        (perf_counter() - start) * 1000,
    )
    return result


async def _time_leaderboard_scheduler_db_operation(step: str, operation, *, sql: str = "", **context):
    logger.info(
        "Leaderboard snapshot DB query start: step=%s stat_date=%s job=%s settings_hash=%s started_at=%s sql=%s",
        step,
        context.get("stat_date") or context.get("stat_dates") or "",
        context.get("job_key") or "",
        context.get("settings_hash") or "",
        datetime.now().isoformat(),
        sql,
    )
    start = perf_counter()
    try:
        result = await operation()
    except Exception:
        logger.exception(
            "Leaderboard snapshot DB query failed: step=%s stat_date=%s job=%s settings_hash=%s elapsed_ms=%.2f",
            step,
            context.get("stat_date") or context.get("stat_dates") or "",
            context.get("job_key") or "",
            context.get("settings_hash") or "",
            (perf_counter() - start) * 1000,
        )
        raise
    logger.info(
        "Leaderboard snapshot DB query done: step=%s stat_date=%s job=%s settings_hash=%s finished_at=%s elapsed_ms=%.2f",
        step,
        context.get("stat_date") or context.get("stat_dates") or "",
        context.get("job_key") or "",
        context.get("settings_hash") or "",
        datetime.now().isoformat(),
        (perf_counter() - start) * 1000,
    )
    return result


def _recent_leaderboard_stat_dates(days: int) -> list[str]:
    from datetime import datetime, timedelta

    from deerflow.persistence.types import BEIJING_TZ

    today = datetime.now(BEIJING_TZ).date()
    start = today - timedelta(days=max(1, days) - 1)
    return [(start + timedelta(days=i)).isoformat() for i in range((today - start).days + 1)]


async def _leaderboard_snapshot_due_dates(
    session_factory,
    *,
    settings_hash: str,
    limit: int,
    running_ttl_seconds: int,
) -> list[str]:
    from datetime import UTC, datetime, timedelta

    from sqlalchemy import and_, or_, select

    from deerflow.persistence.admin_stats.model import AdminLeaderboardDailyStatRow

    cutoff = datetime.now(UTC) - timedelta(seconds=running_ttl_seconds)
    async with session_factory() as session:
        stmt = (
            select(AdminLeaderboardDailyStatRow.stat_date)
            .where(
                AdminLeaderboardDailyStatRow.settings_hash == settings_hash,
                or_(
                    AdminLeaderboardDailyStatRow.status.in_(["queued", "error"]),
                    and_(
                        AdminLeaderboardDailyStatRow.status == "running",
                        AdminLeaderboardDailyStatRow.updated_at < cutoff,
                    ),
                ),
            )
            .order_by(AdminLeaderboardDailyStatRow.stat_date.desc())
            .limit(limit)
        )
        rows = (
            await _execute_leaderboard_scheduler_query(
                session,
                "scheduler.due_dates",
                stmt,
                settings_hash=settings_hash,
            )
        ).all()
    return [str(row[0]) for row in rows]


async def _claim_leaderboard_snapshot(
    session_factory,
    *,
    stat_date: str,
    settings_hash: str,
    running_ttl_seconds: int,
) -> bool:
    from datetime import UTC, datetime

    from app.gateway.routers.admin_users import _snapshot_is_fresh
    from deerflow.persistence.admin_stats.model import AdminLeaderboardDailyStatRow
    from deerflow.persistence.types import BEIJING_TZ

    async with session_factory() as session:
        row = await _time_leaderboard_scheduler_db_operation(
            "scheduler.claim.get",
            lambda: session.get(
                AdminLeaderboardDailyStatRow,
                {"stat_date": stat_date, "settings_hash": settings_hash},
            ),
            sql="SELECT admin_leaderboard_daily_stats by primary key",
            stat_date=stat_date,
            settings_hash=settings_hash,
        )
        if row is None or _snapshot_is_fresh(row, stat_date):
            return False
        if row.status == "running":
            updated_at = row.updated_at
            if isinstance(updated_at, datetime):
                if updated_at.tzinfo is None:
                    updated_at = updated_at.replace(tzinfo=BEIJING_TZ)
                age = (datetime.now(BEIJING_TZ) - updated_at.astimezone(BEIJING_TZ)).total_seconds()
                if age < running_ttl_seconds:
                    return False
        row.status = "running"
        row.message = "?????????????"
        row.error = None
        row.updated_at = datetime.now(UTC)
        await _time_leaderboard_scheduler_db_operation(
            "scheduler.claim.commit",
            lambda: session.commit(),
            sql="COMMIT",
            stat_date=stat_date,
            settings_hash=settings_hash,
        )
    return True


async def _maybe_finalize_previous_leaderboard_day(session_factory, settings_hash: str) -> bool:
    """Once per day after ``_LEADERBOARD_FINALIZE_HOUR``, force-recompute yesterday.

    A ``ready`` past-day snapshot is otherwise treated as fresh forever and never
    rebuilt. But the previous day's snapshot is usually generated before midnight,
    so it can miss runs/tool-calls that only land in the DB around the day
    boundary. This re-queues yesterday (``force=True``) a single time; the rebuild
    stamps ``generated_at`` to today, which stops it from re-firing for the rest of
    the day.
    """
    from datetime import timedelta

    from app.gateway.routers.admin_users import (
        _as_beijing,
        _queue_daily_leaderboard_snapshots,
    )
    from deerflow.persistence.admin_stats.finalize import should_finalize_previous_day
    from deerflow.persistence.admin_stats.model import AdminLeaderboardDailyStatRow
    from deerflow.persistence.types import BEIJING_TZ

    now = datetime.now(BEIJING_TZ)
    yesterday = (now.date() - timedelta(days=1)).isoformat()
    async with session_factory() as session:
        row = await session.get(
            AdminLeaderboardDailyStatRow,
            {"stat_date": yesterday, "settings_hash": settings_hash},
        )
    generated_at = _as_beijing(row.generated_at) if (row is not None and row.generated_at is not None) else None
    if not should_finalize_previous_day(
        status=(row.status if row is not None else None),
        generated_at=generated_at,
        now=now,
        finalize_hour=_LEADERBOARD_FINALIZE_HOUR,
    ):
        return False
    queued = await _queue_daily_leaderboard_snapshots(
        session_factory,
        [yesterday],
        settings_hash,
        force=True,
        message="???????????????????????????",
    )
    if queued:
        logger.info(
            "Leaderboard snapshot finalize: queued previous day %s for full recompute (hash=%s)",
            yesterday,
            settings_hash[:10],
        )
        return True
    return False


async def _run_leaderboard_snapshot_tick() -> tuple[int, int]:
    from app.gateway.routers.admin_users import (
        _daily_job_key,
        _queue_daily_leaderboard_snapshots,
        _refresh_daily_leaderboard_snapshot,
        _resolve_leaderboard_snapshot_scope,
    )
    from deerflow.config.system_settings import load_system_settings
    from deerflow.persistence.engine import get_session_factory

    session_factory = get_session_factory()
    if session_factory is None:
        return 0, 0

    settings = load_system_settings().leaderboard
    excluded, include_scheduled, include_admins, include_failed, settings_hash = _resolve_leaderboard_snapshot_scope(settings)
    stat_dates = _recent_leaderboard_stat_dates(_LEADERBOARD_SNAPSHOT_PREWARM_DAYS)
    queued_dates = await _queue_daily_leaderboard_snapshots(
        session_factory,
        stat_dates,
        settings_hash,
        message="?????????????/?????",
    )

    # ??????? 1 ???????"???"????????????????????
    try:
        await _maybe_finalize_previous_leaderboard_day(session_factory, settings_hash)
    except Exception:
        logger.exception("Leaderboard snapshot finalize check failed (non-fatal)")

    due_dates = await _leaderboard_snapshot_due_dates(
        session_factory,
        settings_hash=settings_hash,
        limit=_LEADERBOARD_SNAPSHOT_BATCH_SIZE,
        running_ttl_seconds=_LEADERBOARD_SNAPSHOT_RUNNING_TTL_SECONDS,
    )
    processed = 0
    for stat_date in due_dates:
        claimed = await _claim_leaderboard_snapshot(
            session_factory,
            stat_date=stat_date,
            settings_hash=settings_hash,
            running_ttl_seconds=_LEADERBOARD_SNAPSHOT_RUNNING_TTL_SECONDS,
        )
        if not claimed:
            continue
        logger.info("Leaderboard snapshot scheduler: building date=%s hash=%s", stat_date, settings_hash[:10])
        await _refresh_daily_leaderboard_snapshot(
            _daily_job_key(stat_date, settings_hash),
            session_factory=session_factory,
            settings=settings,
            stat_date=stat_date,
            settings_hash=settings_hash,
            excluded=excluded,
            include_scheduled=include_scheduled,
            include_admins=include_admins,
            include_failed=include_failed,
        )
        processed += 1
        if _LEADERBOARD_SNAPSHOT_SLEEP_SECONDS > 0 and processed < len(due_dates):
            await asyncio.sleep(_LEADERBOARD_SNAPSHOT_SLEEP_SECONDS)
    return processed, len(queued_dates)


async def _leaderboard_snapshot_scheduler_loop() -> None:
    """Background task inside the existing gateway process for daily snapshots."""
    if not _LEADERBOARD_SNAPSHOT_BACKGROUND_ENABLED:
        logger.info("Leaderboard snapshot scheduler disabled by ADMIN_LEADERBOARD_BACKGROUND_ENABLED")
        return

    from deerflow.config.runtime_paths import runtime_home

    lock_handle = _acquire_singleton_lock(runtime_home() / ".leaderboard_snapshot_scheduler.lock")
    if lock_handle is None:
        logger.info("Leaderboard snapshot scheduler: another gateway worker holds the lock; this worker skips")
        return

    logger.info(
        "Leaderboard snapshot scheduler started: prewarm_days=%d batch_size=%d interval=%.1fs",
        _LEADERBOARD_SNAPSHOT_PREWARM_DAYS,
        _LEADERBOARD_SNAPSHOT_BATCH_SIZE,
        _LEADERBOARD_SNAPSHOT_INTERVAL_SECONDS,
    )
    try:
        if _LEADERBOARD_SNAPSHOT_STARTUP_DELAY_SECONDS > 0:
            await asyncio.sleep(_LEADERBOARD_SNAPSHOT_STARTUP_DELAY_SECONDS)
        while True:
            try:
                processed, queued = await _run_leaderboard_snapshot_tick()
                if processed or queued:
                    logger.info(
                        "Leaderboard snapshot scheduler tick: queued=%d processed=%d",
                        queued,
                        processed,
                    )
            except Exception:
                logger.exception("Leaderboard snapshot scheduler tick failed (non-fatal)")
            await asyncio.sleep(_LEADERBOARD_SNAPSHOT_INTERVAL_SECONDS)
    finally:
        _release_singleton_lock(lock_handle)


def _warn_if_multi_worker() -> None:
    """?? uvicorn / gunicorn ? worker ???? ERROR ???

    ? agent / canvas / AI ??????? Gateway ????? ``StreamBridge``
    ?? run_id ? SSE ?????? map?? worker ????? worker A ??
    ``/runs/stream``???? worker B ? join/resume??? in-process map ??
    ???? 404 / ???

    ????? startup ? ERROR ????**???? fail-fast** ?? ??
    ???????????????? sticky session??????????
    """
    # uvicorn / gunicorn ??????? worker ?????????????
    # ?????? sibling ??????????? WEB_CONCURRENCY?gunicorn
    # / uvicorn ????
    web_concurrency = os.environ.get("WEB_CONCURRENCY", "").strip()
    if web_concurrency and web_concurrency.isdigit() and int(web_concurrency) > 1:
        logger.error(
            "? ??? WEB_CONCURRENCY=%s?? worker ????AI ?? / ? agent "
            "??????????????? worker ????????? / "
            "SSE ?? / resume 400??????????? WEB_CONCURRENCY=1 "
            "????????? sticky session??? AIWritingPage.tsx ????",
            web_concurrency,
        )
        return

    # ???? /proc ??? uvicorn ?????? Linux ??????
    if not sys.platform.startswith("linux"):
        return
    try:
        from pathlib import Path

        peers = 0
        for entry in Path("/proc").iterdir():
            if not entry.name.isdigit():
                continue
            try:
                cmdline = (entry / "cmdline").read_bytes().replace(b"\x00", b" ").decode("utf-8", "ignore")
            except OSError:
                continue
            if "uvicorn" in cmdline and "app.gateway.app" in cmdline:
                peers += 1
        if peers > 1:
            logger.error(
                "? ??? %d ??? uvicorn ???? worker ????? AI ?? "
                "????? / SSE ??????????? worker ??????? "
                "?? sticky session?",
                peers,
            )
    except Exception:
        # ????????? ?? ??????
        pass


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
    """Application lifespan handler."""

    # ?????? worker ?? ?? ?? AI ?? / ? agent ???????
    # ??????? worker ???????? ERROR ?????
    _warn_if_multi_worker()

    # Load config and check necessary environment variables at startup
    try:
        app.state.config = get_app_config()
        apply_logging_level(app.state.config.log_level)
        logger.info("Configuration loaded successfully")
    except Exception as e:
        error_msg = f"Failed to load configuration during gateway startup: {e}"
        logger.exception(error_msg)
        raise RuntimeError(error_msg) from e
    config = get_gateway_config()
    logger.info(f"Starting API Gateway on {config.host}:{config.port}")

    # Initialize LangGraph runtime components (StreamBridge, RunManager, checkpointer, store)
    async with langgraph_runtime(app):
        logger.info("LangGraph runtime initialised")

        # Run Alembic migrations before anything else touches the schema.
        await _run_db_migrations()

        # ?? recall_budget cache?? DB ????,? get_effective_memory_config ?????
        try:
            from deerflow.config.memory_config import load_recall_budget_cache

            _mc_store = getattr(app.state, "memory_config_store", None)
            if _mc_store is not None:
                _recall_budgets = await _mc_store.load_all()
                load_recall_budget_cache(_recall_budgets)
                logger.info("Memory config cache warmed: %d user override(s)", len(_recall_budgets))
        except Exception:
            logger.exception("Memory config cache warm-up failed (non-fatal)")

        # Ensure admin user exists (auto-create on first boot)
        # Must run AFTER langgraph_runtime so app.state.store is available for thread migration
        await _ensure_admin_user(app)

        # Migrate "ownerless" on-disk skills into the new SkillStore table ?
        # makes the multitenancy switch transparent for existing deployments.
        try:
            await _sync_legacy_skills(app)
        except Exception:
            logger.exception("Skill ownership migration failed (non-fatal)")

        # ????? agent(intent / recommender / coordinator)???????
        # ??? _sync_legacy_agents **??**??????????????????
        # ??? upsert ? agents ?,?????? /api/agents ?????,
        # ????????????????.deer-flow/ ? gitignore,??? clone
        # ????????,seeder ????? _roundtable_seed_assets/ ??????
        try:
            from app.gateway.routers._roundtable_seed import (
                ensure_roundtable_functional_agents,
            )

            ensure_roundtable_functional_agents()
        except Exception:
            logger.exception(
                "Roundtable functional agent seeding failed at startup (non-fatal)"
            )

        # AI ????? agent(researcher / outliner / writer / editor)???????
        # ???:??? _sync_legacy_agents ??,?????????????
        # upsert ? agents ?,admin ?????????????????? agent?
        try:
            from app.gateway.routers._ai_writing_seed import (
                ensure_ai_writing_functional_agents,
            )

            ensure_ai_writing_functional_agents()
        except Exception:
            logger.exception(
                "AI-writing functional agent seeding failed at startup (non-fatal)"
            )

        # ?????????????????3Q/6BF/7BF/8BF??agent/chain ?????
        try:
            business_mapping_store = getattr(app.state, "business_mapping_store", None)
            if business_mapping_store is not None:
                inserted = await business_mapping_store.ensure_seed_defaults()
                if inserted:
                    logger.info("Seeded %d business mapping rows", inserted)
        except Exception:
            logger.exception("Business mapping seeding failed at startup (non-fatal)")

        # ???? .deer-flow/agents/{id}/ ???????? agent ?? agents ??
        # ??????????????? sync(????? + N ? agent ??? DB
        # SELECT/UPDATE),??????? + ?????? TTL ??,????
        # /api/agents ?????
        try:
            from app.gateway.routers.agents import (
                _mark_legacy_sync_fresh,
                _sync_legacy_agents,
            )
            from app.gateway.services import _mark_legacy_sync_fresh as _services_mark_sync_fresh

            agent_store = getattr(app.state, "agent_store", None)
            if agent_store is not None:
                await _sync_legacy_agents(agent_store)
                _mark_legacy_sync_fresh()
                # ``services.py`` keeps an independent TTL gate (used by
                # ``start_run`` ? ``_enforce_agent_access``). Without this
                # ack, the first ``/runs/stream`` after startup would
                # re-run the heavy sync even though we just did it.
                _services_mark_sync_fresh()
                logger.info("Legacy folder-backed agents synced at startup")
                # 内置功能型 agent(roundtable-* / ai-writing-*)按用途默认打标签,
                # 供管理员「内置智能体」筛选按用途归类。幂等、只追加、失败不阻塞启动。
                try:
                    from app.gateway.routers._builtin_agent_tag_seed import (
                        ensure_builtin_agent_purpose_tags,
                    )

                    tag_store = getattr(app.state, "tag_store", None)
                    if tag_store is not None:
                        await ensure_builtin_agent_purpose_tags(agent_store, tag_store)
                except Exception:
                    logger.exception("Built-in agent purpose-tag seeding failed at startup (non-fatal)")
        except Exception:
            logger.exception("Legacy agent sync failed at startup (non-fatal)")

        # ?? skill curator ???:??????,?????????????
        # (per-user interval_hours ??????????????)?
        try:
            app.state.curator_scheduler_task = asyncio.create_task(_curator_scheduler_loop())
        except Exception:
            logger.exception("Skill curator scheduler failed to start (non-fatal)")

        # ?? AI ????????:? config.yaml ? ai_writing.cleanup ???,
        # ?????? 3 ????,? retention_days(?? 7 ?)?????,??
        # ?? MySQL ??? + SQLite checkpointer ?? thread ???? worker
        # ?????? leader,?????? worker ??????
        try:
            from app.gateway.ai_writing_cleanup import cleanup_scheduler_loop

            app.state.ai_writing_cleanup_task = asyncio.create_task(cleanup_scheduler_loop(app))
        except Exception:
            logger.exception("AI writing cleanup scheduler failed to start (non-fatal)")

        try:
            app.state.leaderboard_snapshot_scheduler_task = asyncio.create_task(_leaderboard_snapshot_scheduler_loop())
        except Exception:
            logger.exception("Leaderboard snapshot scheduler failed to start (non-fatal)")

        # Start IM channel service if any channels are configured
        try:
            from app.channels.service import start_channel_service

            channel_service = await start_channel_service(app.state.config)
            logger.info("Channel service started: %s", channel_service.get_status())
        except Exception:
            logger.exception("No IM channels configured or channel service failed to start")

        scheduled_tasks_enabled = os.environ.get("SCHEDULED_TASKS_ENABLED", "true").strip().lower()
        if scheduled_tasks_enabled in {"0", "false", "no", "off"}:
            logger.info("Scheduled task service disabled by SCHEDULED_TASKS_ENABLED=%s", scheduled_tasks_enabled)
        else:
            try:
                scheduled_task_store = getattr(app.state, "scheduled_task_store", None)
                if scheduled_task_store is not None:
                    scheduler_service = ScheduledTaskService(app, scheduled_task_store)
                    app.state.scheduled_task_service = scheduler_service
                    set_scheduled_task_service(scheduler_service)
                    await scheduler_service.start()
                    logger.info("Scheduled task service started")
            except Exception:
                logger.exception("Scheduled task service failed to start")

        yield

        # ?? curator ???
        curator_task = getattr(app.state, "curator_scheduler_task", None)
        if curator_task is not None:
            curator_task.cancel()
            try:
                await curator_task
            except asyncio.CancelledError:
                pass
            except Exception:
                logger.exception("Curator scheduler shutdown error")

        # ?? AI ???????
        cleanup_task = getattr(app.state, "ai_writing_cleanup_task", None)
        if cleanup_task is not None:
            cleanup_task.cancel()
            try:
                await cleanup_task
            except asyncio.CancelledError:
                pass
            except Exception:
                logger.exception("AI writing cleanup scheduler shutdown error")

        leaderboard_snapshot_task = getattr(app.state, "leaderboard_snapshot_scheduler_task", None)
        if leaderboard_snapshot_task is not None:
            leaderboard_snapshot_task.cancel()
            try:
                await leaderboard_snapshot_task
            except asyncio.CancelledError:
                pass
            except Exception:
                logger.exception("Leaderboard snapshot scheduler shutdown error")

        # ?? 7 ????AI ?? graph ???? checkpointer / in-flight task?
        # thread ??? LangGraph Server ? checkpointer ???lifespan ???
        # ``langgraph_runtime()`` ? contextmanager ???????????????

        try:
            scheduler_service = getattr(app.state, "scheduled_task_service", None)
            if scheduler_service is not None:
                await scheduler_service.stop()
        except Exception:
            logger.exception("Failed to stop scheduled task service")
        finally:
            set_scheduled_task_service(None)

        # Stop channel service on shutdown (bounded to prevent worker hang)
        try:
            from app.channels.service import stop_channel_service

            await asyncio.wait_for(
                stop_channel_service(),
                timeout=_SHUTDOWN_HOOK_TIMEOUT_SECONDS,
            )
        except TimeoutError:
            logger.warning(
                "Channel service shutdown exceeded %.1fs; proceeding with worker exit.",
                _SHUTDOWN_HOOK_TIMEOUT_SECONDS,
            )
        except Exception:
            logger.exception("Failed to stop channel service")

    logger.info("Shutting down API Gateway")


def create_app() -> FastAPI:
    """Create and configure the FastAPI application.

    Returns:
        Configured FastAPI application instance.
    """
    config = get_gateway_config()
    docs_kwargs = {"docs_url": "/docs", "redoc_url": "/redoc", "openapi_url": "/openapi.json"} if config.enable_docs else {"docs_url": None, "redoc_url": None, "openapi_url": None}

    app = FastAPI(
        title="DeerFlow API Gateway",
        description="""
## DeerFlow API Gateway

API Gateway for DeerFlow - A LangGraph-based AI agent backend with sandbox execution capabilities.

### Features

- **Models Management**: Query and retrieve available AI models
- **MCP Configuration**: Manage Model Context Protocol (MCP) server configurations
- **Memory Management**: Access and manage global memory data for personalized conversations
- **Skills Management**: Query and manage skills and their enabled status
- **Artifacts**: Access thread artifacts and generated files
- **Health Monitoring**: System health check endpoints

### Architecture

LangGraph requests are handled by nginx reverse proxy.
This gateway provides custom endpoints for models, MCP configuration, skills, and artifacts.
        """,
        version="0.1.0",
        lifespan=lifespan,
        **docs_kwargs,
        openapi_tags=[
            {
                "name": "models",
                "description": "Operations for querying available AI models and their configurations",
            },
            {
                "name": "mcp",
                "description": "Manage Model Context Protocol (MCP) server configurations",
            },
            {
                "name": "memory",
                "description": "Access and manage global memory data for personalized conversations",
            },
            {
                "name": "skills",
                "description": "Manage skills and their configurations",
            },
            {
                "name": "artifacts",
                "description": "Access and download thread artifacts and generated files",
            },
            {
                "name": "uploads",
                "description": "Upload and manage user files for threads",
            },
            {
                "name": "threads",
                "description": "Manage DeerFlow thread-local filesystem data",
            },
            {
                "name": "agents",
                "description": "Create and manage custom agents with per-agent config and prompts",
            },
            {
                "name": "suggestions",
                "description": "Generate follow-up question suggestions for conversations",
            },
            {
                "name": "channels",
                "description": "Manage IM channel integrations (Feishu, Slack, Telegram)",
            },
            {
                "name": "assistants-compat",
                "description": "LangGraph Platform-compatible assistants API (stub)",
            },
            {
                "name": "runs",
                "description": "LangGraph Platform-compatible runs lifecycle (create, stream, cancel)",
            },
            {
                "name": "scheduled-tasks",
                "description": "Create, manage, and execute user scheduled agent tasks",
            },
            {
                "name": "health",
                "description": "Health check and system status endpoints",
            },
        ],
    )

    # Auth: reject unauthenticated requests to non-public paths (fail-closed safety net)
    app.add_middleware(AuthMiddleware)

    # CSRF: Double Submit Cookie pattern for state-changing requests
    app.add_middleware(CSRFMiddleware)

    # CORS
    if is_auth_disabled():
        logger.warning(
            "DEER_FLOW_AUTH_DISABLED is set ? API authentication is bypassed. "
            "Do not use this in production."
        )

    def _cors_allow_any_origin() -> bool:
        """Mirror any browser Origin (works with credentials); cannot use literal '*'."""
        v = os.environ.get("GATEWAY_CORS_ALLOW_ALL", "").strip().lower()
        return v in ("1", "true", "yes", "on") or is_auth_disabled()

    cors_origins_env = os.environ.get("GATEWAY_CORS_ORIGINS", "").strip()
    cors_origins: list[str] = [o.strip() for o in cors_origins_env.split(",") if o.strip()] if cors_origins_env else []

    # Headers the LangGraph SDK / browser clients need to read from cross-origin
    # responses. Without exposing them, the SDK's `response.headers.get(...)` returns
    # null on cross-origin streaming responses, breaking run-id extraction and reconnect.
    cors_expose_headers = [
        "Content-Disposition",
        "Content-Location",
        "Location",
        "X-Run-Id",
        "X-Thread-Id",
    ]

    if _cors_allow_any_origin():
        logger.warning(
            "Permissive CORS enabled ? any Origin is echoed back (GATEWAY_CORS_ALLOW_ALL or "
            "DEER_FLOW_AUTH_DISABLED). Do not expose publicly."
        )
        app.add_middleware(
            CORSMiddleware,
            allow_origins=[],
            # fullmatch() on the Origin header; browsers send scheme://host[:port]
            allow_origin_regex=r".+",
            allow_credentials=True,
            allow_methods=["*"],
            allow_headers=["*"],
            expose_headers=cors_expose_headers,
        )
    elif cors_origins:
        if "*" in cors_origins:
            logger.error(
                "GATEWAY_CORS_ORIGINS contains '*' with allow_credentials=True ? invalid. "
                "Set GATEWAY_CORS_ALLOW_ALL=1 instead, or list explicit origins."
            )
            cors_origins = [o for o in cors_origins if o != "*"]
        if cors_origins:
            app.add_middleware(
                CORSMiddleware,
                allow_origins=cors_origins,
                allow_credentials=True,
                allow_methods=["*"],
                allow_headers=["*"],
                expose_headers=cors_expose_headers,
            )
    else:
        # Safe-by-default fallback for local frontend development. Without any CORS
        # middleware, browser preflight OPTIONS requests will hit routes directly and
        # fail with 405, which blocks login and all JSON POST APIs.
        dev_origins = [
            "http://localhost:5173",
            "http://127.0.0.1:5173",
            "http://localhost:5174",
            "http://127.0.0.1:5174",
            "http://localhost:3000",
            "http://127.0.0.1:3000",
        ]
        logger.warning(
            "No CORS configuration found; enabling local dev origins only: %s. "
            "Set GATEWAY_CORS_ORIGINS or GATEWAY_CORS_ALLOW_ALL for deployment.",
            ", ".join(dev_origins),
        )
        app.add_middleware(
            CORSMiddleware,
            allow_origins=dev_origins,
            allow_credentials=True,
            allow_methods=["*"],
            allow_headers=["*"],
            expose_headers=cors_expose_headers,
        )

    # Include routers
    # Models API is mounted at /api/models
    app.include_router(models.router)

    # MCP API is mounted at /api/mcp
    app.include_router(mcp.router)

    # Memory API is mounted at /api/memory
    app.include_router(memory.router)

    # Skills API is mounted at /api/skills
    app.include_router(skills.router)

    # Artifacts API is mounted at /api/threads/{thread_id}/artifacts
    app.include_router(artifacts.router)

    # Uploads API is mounted at /api/threads/{thread_id}/uploads
    app.include_router(uploads.router)

    # Optional Chrome-extension page capture API is mounted at /api/browser-context
    app.include_router(browser_context.router)

    # Thread cleanup API is mounted at /api/threads/{thread_id}
    app.include_router(threads.router)

    # Agents API is mounted at /api/agents
    app.include_router(agents.router)

    # Recommended Questions API is mounted at /api/recommended-questions
    app.include_router(recommended_questions.router)

    app.include_router(user_prompts.router)

    # Per-user UI preferences at /api/user-preferences
    app.include_router(user_preferences.router)

    # Roundtable-planning drafts + recommendation history at /api/roundtable-drafts
    app.include_router(roundtable_drafts.router)

    # Roundtable-planning business chains at /api/roundtable-chains
    app.include_router(roundtable_chains.router)

    # Global business ? agent / chain mapping at /api/business-mapping
    app.include_router(business_mapping.router)

    # Light-app registry (?????) at /api/light-apps
    app.include_router(light_apps.router)

    # Sidebar-menu overrides (菜单管理) at /api/menu-overrides
    app.include_router(menu_overrides.router)

    # Sensitive-word map (敏感词管理 / 脱敏词) at /api/sensitive-words
    app.include_router(sensitive_words.router)

    # Task-workspace configurable jump buttons (按钮管理) at /api/task-buttons
    app.include_router(task_buttons.router)

    # Standalone parallel multi-agent panel: fan-out streaming + saved tasks
    app.include_router(parallel_agents.router)
    app.include_router(parallel_agent_tasks.router)

    # Roundtable-planning background jobs at /api/roundtable-jobs
    app.include_router(roundtable_jobs.router)

    # Suggestions API is mounted at /api/threads/{thread_id}/suggestions
    app.include_router(suggestions.router)

    # Writing rewrite API is mounted at /api/writing
    app.include_router(writing.router)

    # AI Writing multi-agent API is mounted at /api/ai-writing
    app.include_router(ai_writing.router)

    # Scheduled task API is mounted at /api/scheduled-tasks
    app.include_router(scheduled_tasks.router)

    # Public, unauthenticated read-only scheduled task API
    # (/api/public/scheduled-tasks/*). Lets external systems query tasks and
    # surface results without a DeerFlow login.
    app.include_router(public_scheduled_tasks.router)

    # Public read-only user leaderboard publish page (/api/public/leaderboard)
    app.include_router(public_leaderboard.router)

    # Public embed Q&A status polling (/api/public/embed/*)
    app.include_router(public_embed.router)
    app.include_router(public_config.router)
    app.include_router(html_page_favorites.router)

    # Product knowledge base API is mounted at /api/knowledge
    app.include_router(knowledge.router)

    # External knowledge ingest (publish md artifacts) at /api/knowledge-ingest
    app.include_router(knowledge_ingest.router)

    # System settings API is mounted at /api/system-settings
    app.include_router(system_settings.router)

    # Admin LLM metrics audit + export
    app.include_router(llm_metrics.router)

    # Admin window into other users' threads / messages / memory
    app.include_router(admin_users.router)
    app.include_router(admin_users.self_router)

    # Admin tool / skill call metrics audit + export
    app.include_router(tool_metrics.router)

    # Channels API is mounted at /api/channels
    app.include_router(channels.router)

    # Admin checkpoint migration API: SQLite legacy checkpoints -> Postgres.
    app.include_router(checkpoint_migration.router)

    # Assistants compatibility API (LangGraph Platform stub)
    app.include_router(assistants_compat.router)

    # Auth API is mounted at /api/v1/auth
    app.include_router(auth.router)

    # Feedback API is mounted at /api/threads/{thread_id}/runs/{run_id}/feedback
    app.include_router(feedback.router)

    # Notifications API is mounted at /api/notifications
    app.include_router(notifications.router)

    # Tags API is mounted at /api/tags
    app.include_router(tags.router)

    # 岗位 (position) management API is mounted at /api/positions
    app.include_router(positions.router)

    # Thread Runs API (LangGraph Platform-compatible runs lifecycle)
    app.include_router(thread_runs.router)

    app.include_router(thread_shares.router)
    # Public, unauthenticated read-only share snapshots (/api/public/shares/*).
    app.include_router(thread_shares.public_router)

    # ?????????import ???? / view ????????
    app.include_router(roundtable_draft_shares.router)
    # Public, unauthenticated read-only roundtable result (/api/public/roundtable-shares/*).
    app.include_router(roundtable_draft_shares.public_router)

    # Stateless Runs API (stream/wait without a pre-existing thread)
    app.include_router(runs.router)

    # --- ?????????? /page/workspace/roundtable/planning?---
    # ??? app/gateway/routers/{multi_agent,intent,recommend}.py?
    # ??? frontend-web/docs/multi-agent-backend-dev.md
    app.include_router(multi_agent.router)   # Step 2?init(SSE) + run/stream(leader|special)
    app.include_router(intent.router)        # Step 1?/api/intent/init + /stream
    app.include_router(recommend.router)     # Step 1?2?/api/recommend/stream?? init?
    # Step 3 ????/?????? artifacts.router?/api/threads/{id}/artifacts?

    # ?????????(? auth_middleware._PUBLIC_PATH_PREFIXES)?
    # ????? ``/health`` ?? API ??? ``/api/health``,?????
    # ?? base URL ????,?????????? 401?
    @app.get("/health", tags=["health"])
    @app.get("/api/health", tags=["health"])
    async def health_check() -> dict:
        """Health check endpoint.

        Returns:
            Service health status information.
        """
        return {"status": "healthy", "service": "deer-flow-gateway"}

    return app


# Create app instance for uvicorn
app = create_app()