# ruff: noqa: E402, I001
# The dotenv load below must precede application imports that read environment.
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 pathlib import Path
from time import perf_counter
from typing import Any

from dotenv import load_dotenv
from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.openapi.docs import get_redoc_html, get_swagger_ui_html
from fastapi.responses import HTMLResponse
from fastapi.staticfiles import StaticFiles

# 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.public_knowledge_cors import PublicKnowledgeCORSMiddleware
from app.gateway.routers import (
    admin_active_runs,
    admin_audit_retention as admin_audit_retention_routes,
    admin_users,
    agents,
    ai_writing,
    app_identity,
    artifacts,
    assistant_knowledge,
    assistants_compat,
    auth,
    service_tokens,
    browser_context,
    business_mapping,
    channels,
    chat_archive,
    checkpoint_migration,
    compact_menu_overrides,
    dashboard_sessions,
    deep_research,
    direct_chat,
    external_llmwiki,
    feedback,
    fixed_questions,
    flow_extract,
    force_situation,
    gongwen_documents,
    html_images,
    html_page_favorites,
    html_runtime,
    html_templates,
    intent,
    knowledge,
    knowledge_ingest,
    light_apps,
    llm_metrics,
    llmwiki,
    llmwiki_index,
    mcp,
    memory,
    menu_overrides,
    model_exchanges,
    model_management,
    models,
    multi_agent,
    notifications,
    open_3qfx,
    open_chat,
    parallel_agent_tasks,
    parallel_agents,
    position_roles,
    position_roundtable,
    positions,
    public_config,
    public_embed,
    public_knowledge_vector_search,
    public_leaderboard,
    public_scheduled_tasks,
    recommend,
    recommended_questions,
    report_collaboration,
    report_structures,
    roundtable_artifact_submissions,
    roundtable_chains,
    roundtable_diagnostics,
    roundtable_draft_shares,
    roundtable_drafts,
    roundtable_jobs,
    roundtable_rooms,
    roundtable_task_drafts,
    runs,
    scheduled_tasks,
    sensitive_words,
    sentiment_agent,
    skill_addresses,
    skill_knowledge,
    skill_view_diagnostics,
    skills,
    strategy_agents,
    stream_truncations,
    suggestions,
    system_settings,
    tags,
    task_analysis,
    task_analysis_qa,
    task_buttons,
    task_deeplink_details,
    task_deeplink_threads,
    task_reports,
    taskcop_tasks,
    thread_runs,
    thread_shares,
    threads,
    tool_metrics,
    upload_policy,
    uploads,
    user_preferences,
    user_prompts,
    web_app_sites,
    workflow_data_sources,
    workflow_embed,
    workflow_planning,
    workflow_resources,
    workflow_runs,
    workflow_studio,
    workflows,
    workflows_coze_compat,
    writing,
)
from app.gateway.runtime_executor import configure_runtime_executor  # noqa: E402
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__)

# Swagger UI and ReDoc are vendored so the API documentation works in isolated
# intranets.  The route intentionally lives below /docs so it is covered by the
# existing anonymous docs allow-list in the auth middleware.
_DOCS_STATIC_DIR = Path(__file__).with_name("static") / "docs"

# 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) -> list[Any] | 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 None

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

    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 None

    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)
    return skills


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

    from deerflow.persistence.gauss_mysql_compat import is_gauss_mysql_product

    if is_gauss_mysql_product():
        logger.warning("Database product=gauss_mysql; skipping MySQL-specific 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 TimeoutError
                await asyncio.sleep(0.2)
            result = result_queue.get_nowait()
        except 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,
)
# Concurrency (system-pressure) sampler: how often to snapshot the in-flight run
# count, and how long to keep samples (drives the admin 并发量 line chart, which
# aggregates per minute/hour). Sampling is in-memory cheap (one tiny INSERT).
_CONCURRENCY_SAMPLE_ENABLED = _env_bool("CONCURRENCY_SAMPLE_ENABLED", True)
_CONCURRENCY_SAMPLE_INTERVAL_SECONDS = _env_float(
    "CONCURRENCY_SAMPLE_INTERVAL_SECONDS",
    15.0,
)
_CONCURRENCY_SAMPLE_RETENTION_HOURS = _env_float(
    "CONCURRENCY_SAMPLE_RETENTION_HOURS",
    168.0,  # 7 days
)
# Administrative audit data is operational rather than archival.  The actual
# cutoff lives with the shared query policy; these controls intentionally make
# cleanup a low-priority, incremental task instead of a large startup delete.
_ADMIN_AUDIT_RETENTION_INTERVAL_SECONDS = _env_float(
    "ADMIN_AUDIT_RETENTION_INTERVAL_SECONDS",
    15 * 60,
    minimum=60.0,
)
_ADMIN_AUDIT_RETENTION_STARTUP_DELAY_SECONDS = _env_float(
    "ADMIN_AUDIT_RETENTION_STARTUP_DELAY_SECONDS",
    300.0,
    minimum=0.0,
)
_ADMIN_AUDIT_RETENTION_BATCH_SIZE = min(
    500,
    _env_int("ADMIN_AUDIT_RETENTION_BATCH_SIZE", 100, minimum=1),
)
_ADMIN_AUDIT_RETENTION_MAX_BATCHES_PER_STORE = min(
    16,
    _env_int("ADMIN_AUDIT_RETENTION_MAX_BATCHES_PER_STORE", 4, minimum=1),
)
_ADMIN_AUDIT_RETENTION_INTER_BATCH_PAUSE_SECONDS = min(
    1.0,
    _env_float("ADMIN_AUDIT_RETENTION_INTER_BATCH_PAUSE_SECONDS", 0.05, minimum=0.0),
)
_ADMIN_AUDIT_RETENTION_MAX_SWEEP_SECONDS = min(
    60.0,
    _env_float("ADMIN_AUDIT_RETENTION_MAX_SWEEP_SECONDS", 8.0, minimum=0.1),
)
# 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 app.gateway.admin_query_window import ADMIN_QUERY_WINDOW_DAYS
    from deerflow.persistence.types import BEIJING_TZ

    today = datetime.now(BEIJING_TZ).date()
    start = today - timedelta(days=min(ADMIN_QUERY_WINDOW_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)


async def _concurrency_sampler_loop(app: FastAPI) -> None:
    """Periodically record the in-flight run count for the admin pressure chart.

    Runs in the gateway process, singleton-locked so only one worker samples.
    Each tick records ``len(run_manager.list_active())`` and, roughly hourly,
    purges samples older than the retention window. Best-effort throughout — a
    failed tick is logged and the loop continues.
    """
    if not _CONCURRENCY_SAMPLE_ENABLED:
        logger.info("Concurrency sampler disabled by CONCURRENCY_SAMPLE_ENABLED")
        return

    store = getattr(app.state, "concurrency_sample_store", None)
    run_mgr = getattr(app.state, "run_manager", None)
    concurrency_gate = getattr(app.state, "concurrency_gate", None)
    if store is None or run_mgr is None:
        logger.info("Concurrency sampler not started (no store / run manager — memory backend?)")
        return

    from deerflow.config.runtime_paths import runtime_home

    lock_handle = _acquire_singleton_lock(runtime_home() / ".concurrency_sampler.lock")
    if lock_handle is None:
        logger.info("Concurrency sampler: another gateway worker holds the lock; this worker skips")
        return

    interval = max(1.0, _CONCURRENCY_SAMPLE_INTERVAL_SECONDS)
    purge_every_ticks = max(1, int(3600 / interval))  # ~once per hour
    logger.info("Concurrency sampler started: interval=%.1fs retention=%.1fh", interval, _CONCURRENCY_SAMPLE_RETENTION_HOURS)
    ticks = 0
    try:
        while True:
            try:
                # Runtime toggle (admin can turn sampling off without a restart).
                # mtime-cached read, so this is cheap every tick.
                from deerflow.config.system_settings import get_concurrency_monitor_settings

                if get_concurrency_monitor_settings().enabled:
                    active_count = None
                    if concurrency_gate is not None:
                        active_count = await concurrency_gate.count_active()
                    if active_count is None:
                        active = await run_mgr.list_active()
                        active_count = len(active)
                    await store.record(active_count)
                    ticks += 1
                    if ticks % purge_every_ticks == 0:
                        from datetime import UTC, datetime, timedelta

                        cutoff = datetime.now(UTC) - timedelta(hours=_CONCURRENCY_SAMPLE_RETENTION_HOURS)
                        await store.purge_older_than(cutoff)
            except Exception:
                logger.exception("Concurrency sampler tick failed (non-fatal)")
            await asyncio.sleep(interval)
    finally:
        _release_singleton_lock(lock_handle)


async def _admin_audit_retention_loop(app: FastAPI) -> None:
    """Keep bounded admin-audit tables at the three-month retention target."""
    from app.gateway.admin_audit_retention import sweep_admin_audit_retention
    from app.gateway.admin_audit_retention_control import (
        is_paused,
        read_status,
        write_status,
    )
    from app.gateway.admin_query_window import ADMIN_AUDIT_RETENTION_DAYS
    from deerflow.config.runtime_paths import runtime_home
    from deerflow.persistence.engine import get_session_factory

    lock_handle = _acquire_singleton_lock(runtime_home() / ".admin_audit_retention.lock")
    if lock_handle is None:
        logger.info("Admin audit retention: another gateway worker holds the lock; this worker skips")
        return

    root = runtime_home()
    status = read_status(root)
    status.update(
        {
            "retention_days": ADMIN_AUDIT_RETENTION_DAYS,
            "interval_seconds": _ADMIN_AUDIT_RETENTION_INTERVAL_SECONDS,
            "startup_delay_seconds": _ADMIN_AUDIT_RETENTION_STARTUP_DELAY_SECONDS,
            "batch_size": _ADMIN_AUDIT_RETENTION_BATCH_SIZE,
            "max_rounds": _ADMIN_AUDIT_RETENTION_MAX_BATCHES_PER_STORE,
            "max_sweep_seconds": _ADMIN_AUDIT_RETENTION_MAX_SWEEP_SECONDS,
        }
    )

    def save_status(**updates: Any) -> None:
        status.update(updates)
        write_status(root, dict(status))

    async def publish_progress(progress: dict[str, Any]) -> None:
        await asyncio.to_thread(save_status, **progress)

    logger.info(
        "Admin audit retention started: interval=%ss startup_delay=%ss batch_size=%s "
        "max_batches_per_store=%s max_sweep_seconds=%s",
        _ADMIN_AUDIT_RETENTION_INTERVAL_SECONDS,
        _ADMIN_AUDIT_RETENTION_STARTUP_DELAY_SECONDS,
        _ADMIN_AUDIT_RETENTION_BATCH_SIZE,
        _ADMIN_AUDIT_RETENTION_MAX_BATCHES_PER_STORE,
        _ADMIN_AUDIT_RETENTION_MAX_SWEEP_SECONDS,
    )
    try:
        if _ADMIN_AUDIT_RETENTION_STARTUP_DELAY_SECONDS:
            # Do not compete with migrations, warmup, or users arriving just
            # after a Gateway restart. The scheduled passes are deliberately
            # maintenance work, not a startup requirement.
            save_status(
                status="scheduled",
                current_source=None,
                next_run_at=(
                    datetime.now().astimezone().timestamp()
                    + _ADMIN_AUDIT_RETENTION_STARTUP_DELAY_SECONDS
                ),
            )
            await asyncio.sleep(_ADMIN_AUDIT_RETENTION_STARTUP_DELAY_SECONDS)
        while True:
            if is_paused(root):
                save_status(status="paused", current_source=None, next_run_at=None)
                await asyncio.sleep(min(2.0, _ADMIN_AUDIT_RETENTION_INTERVAL_SECONDS))
                continue

            started_at = datetime.now().astimezone().isoformat(timespec="seconds")
            save_status(
                status="running",
                current_source=None,
                round=0,
                deleted={
                    "llm_metrics": 0,
                    "tool_metrics": 0,
                    "run_summaries": 0,
                    "leaderboard_snapshots": 0,
                },
                started_at=started_at,
                next_run_at=None,
                last_error=None,
            )
            try:
                result = await sweep_admin_audit_retention(
                    session_factory=get_session_factory(),
                    llm_metrics_store=getattr(app.state, "llm_metrics_store", None),
                    tool_metrics_store=getattr(app.state, "tool_metrics_store", None),
                    run_store=getattr(app.state, "run_store", None),
                    batch_size=_ADMIN_AUDIT_RETENTION_BATCH_SIZE,
                    max_batches_per_store=_ADMIN_AUDIT_RETENTION_MAX_BATCHES_PER_STORE,
                    inter_batch_pause_seconds=_ADMIN_AUDIT_RETENTION_INTER_BATCH_PAUSE_SECONDS,
                    max_sweep_seconds=_ADMIN_AUDIT_RETENTION_MAX_SWEEP_SECONDS,
                    should_stop=lambda: is_paused(root),
                    on_progress=publish_progress,
                )
                finished_at = datetime.now().astimezone().isoformat(timespec="seconds")
                paused = is_paused(root)
                last_sweep = {
                    "started_at": started_at,
                    "finished_at": finished_at,
                    "deleted": result,
                    "stopped_by_admin": paused,
                }
                save_status(
                    status="paused" if paused else "scheduled",
                    current_source=None,
                    round=status.get("round", 0),
                    deleted=result,
                    started_at=started_at,
                    finished_at=finished_at,
                    last_sweep=last_sweep,
                    next_run_at=None if paused else (
                        datetime.now().astimezone().timestamp()
                        + _ADMIN_AUDIT_RETENTION_INTERVAL_SECONDS
                    ),
                )
            except Exception:
                save_status(
                    status="error",
                    current_source=None,
                    last_error="清理轮次发生异常，请查看后台日志",
                    finished_at=datetime.now().astimezone().isoformat(timespec="seconds"),
                )
                logger.exception("Admin audit retention sweep failed (non-fatal)")

            # Wake periodically so an administrator can pause/resume without
            # waiting for the full 15-minute cleanup interval.
            remaining = _ADMIN_AUDIT_RETENTION_INTERVAL_SECONDS
            while remaining > 0 and not is_paused(root):
                sleep_for = min(2.0, remaining)
                await asyncio.sleep(sleep_for)
                remaining -= sleep_for
    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 _gateway_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)
        await configure_runtime_executor(app.state.config.runtime_performance)
        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")

        # One-time repair switch for a failed, unused Workflow Studio schema.
        # The helper is intentionally MySQL-only and drops an exact allow-list
        # in the configured current database; it never expands ``workflow_%``.
        # Keep this opt-in so established workflow data is never cleared during
        # an ordinary restart.
        if os.getenv("DEER_FLOW_REBUILD_WORKFLOW_SCHEMA", "").lower() in {"1", "true", "yes"}:
            from deerflow.persistence.engine import rebuild_workflow_schema_for_mysql

            logger.warning(
                "DEER_FLOW_REBUILD_WORKFLOW_SCHEMA is enabled; rebuilding the unused workflow schema"
            )
            await rebuild_workflow_schema_for_mysql()

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

        # The lightweight Wiki coordinator always runs: it upgrades persisted
        # generation prompts and applies strict focus filters even when the
        # optional local mirror is disabled. Mirror/index scans themselves
        # remain opt-in inside the loop, with per-library database leases for
        # multi-worker safety.
        from app.gateway.llmwiki_index_scheduler import local_wiki_index_scheduler_loop

        app.state.llmwiki_index_scheduler_task = asyncio.create_task(
            local_wiki_index_scheduler_loop(app)
        )

        # ?? 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.
        catalog_skills = None
        try:
            catalog_skills = await _sync_legacy_skills(app)
        except Exception:
            logger.exception("Skill ownership migration failed (non-fatal)")

        # Project the compact per-skill catalog metadata into MySQL before the
        # Gateway accepts requests. List APIs read indexed scalar columns and
        # never load skill markdown, display samples, or usage sidecars.
        try:
            from app.gateway.skill_catalog_cache import refresh_skill_catalog_projection

            catalog_size = await refresh_skill_catalog_projection(
                app.state.config,
                skills=catalog_skills,
            )
            logger.info("Skill MySQL catalog projection synchronized: skills=%d", catalog_size)
        except Exception:
            logger.exception("Skill MySQL catalog projection sync 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)"
            )

        # Strategy-selection functional agents (intent, six independent
        # analysis angles, judge, and solution editor).  Seed them before
        # _sync_legacy_agents so they are immediately visible and editable in
        # the existing agent-management UI on a fresh deployment.
        try:
            from app.gateway.routers._strategy_selection_seed import (
                ensure_strategy_selection_agents,
            )

            ensure_strategy_selection_agents()
        except Exception:
            logger.exception(
                "Strategy-selection functional agent seeding failed at startup (non-fatal)"
            )

        # 内置「模板数据转换器」agent(JSON → HTML 模板填充)。同样必须在
        # _sync_legacy_agents **之前**补建目录,后者把它 upsert 进 agents 表,
        # 第一次启动 /api/agents 与定时任务下拉框里就有它。
        try:
            from app.gateway.routers._template_builder_seed import (
                ensure_template_builder_agent,
            )

            ensure_template_builder_agent()
        except Exception:
            logger.exception(
                "Template-builder agent seeding failed at startup (non-fatal)"
            )

        # HTML 模板库:把随包内置模板 seed 进 {base_dir}/html-templates/,模板管理页
        # 与定时任务下拉即可见(缺失才复制,绝不覆盖管理页/线上修改)。
        try:
            from deerflow.runtime.scheduler import template_store

            template_store.ensure_seeded()
        except Exception:
            logger.exception(
                "HTML template store seeding failed at startup (non-fatal)"
            )

        # 实验性「页面研报设计师」：在 _sync_legacy_agents 前补齐 agent 目录与
        # 公共页面生成技能，首次启动即可在智能体管理页和普通对话中复用。
        try:
            from app.gateway.routers._page_report_designer_seed import (
                ensure_page_report_designer_agent,
            )

            ensure_page_report_designer_agent()
        except Exception:
            logger.exception(
                "Page-report designer agent seeding failed at startup (non-fatal)"
            )

        # 深度研究「收集员」内置 agent(对话式信息收集)。同样必须在
        # _sync_legacy_agents 之前补建目录,由后者 upsert 进 agents 表,
        # 深度研究页的收集对话与智能体管理页即可直接使用。
        try:
            from app.gateway.routers._deep_research_seed import (
                ensure_deep_research_collector_agent,
            )

            ensure_deep_research_collector_agent()
        except Exception:
            logger.exception(
                "Deep-research collector agent seeding failed at startup (non-fatal)"
            )

        # Built-in public-document stack: agent + workflow skill + the full
        # gongwen-assistant DOCX engine skill, all version-synced into
        # skills/public/ (see _gongwen_seed.py).  Seed before legacy agent
        # sync so it is immediately available to ordinary conversations.
        try:
            from app.gateway.routers._gongwen_seed import ensure_gongwen_agent

            ensure_gongwen_agent()
        except Exception:
            logger.exception(
                "Public-document agent seeding failed at startup (non-fatal)"
            )

        # 内置「认知态势图」public 技能：skills/ 不随 git 快照搬运，新机器
        # 部署后技能缺失，由种子资产在首次启动时补建（目录存在则不动）。
        try:
            from app.gateway.routers._cognitive_situation_seed import (
                ensure_cognitive_situation_skills,
            )

            ensure_cognitive_situation_skills()
        except Exception:
            logger.exception(
                "Cognitive-situation skill seeding failed at startup (non-fatal)"
            )

        # Workflow Studio owns a separate planning controller.  It must be
        # seeded before _sync_legacy_agents, so the planner is available to
        # every user's candidate-workflow request on a fresh deployment.
        try:
            from app.gateway.routers._workflow_planner_seed import (
                ensure_workflow_planner_agent,
            )

            ensure_workflow_planner_agent()
        except Exception:
            logger.exception(
                "Workflow planner agent seeding failed at startup (non-fatal)"
            )

        # 内置「强制检索输出助手」：必须在 _sync_legacy_agents 之前补建，
        # 使首次启动就能在智能体管理/广场中使用。每轮检索门禁由 harness
        # 的 ForcedResearchMiddleware 执行，而不是仅依赖 SOUL 提示词。
        try:
            from app.gateway.routers._forced_research_seed import (
                ensure_forced_research_agent,
            )

            ensure_forced_research_agent()
        except Exception:
            logger.exception(
                "Forced-research responder agent seeding failed at startup (non-fatal)"
            )

        # 内置「查询变慢排查助手」：Linux + MySQL 间歇性查询变慢。目录已存在则不覆盖。
        try:
            from app.gateway.routers._query_slowdown_diagnosis_seed import (
                ensure_query_slowdown_diagnosis_agent,
            )

            ensure_query_slowdown_diagnosis_agent()
        except Exception:
            logger.exception(
                "Query-slowdown diagnosis agent seeding failed at startup (non-fatal)"
            )
        # 内置「耗时排查助手」：根据执行记录和日志判断空等原因。目录已存在则不覆盖。
        try:
            from app.gateway.routers._latency_diagnosis_seed import (
                ensure_latency_diagnosis_agent,
            )

            ensure_latency_diagnosis_agent()
        except Exception:
            logger.exception(
                "Latency-diagnosis agent seeding failed at startup (non-fatal)"
            )

        # 内置 Elasticsearch 离线运维助手：随离线包提供部署知识、只读采集脚本
        # 和规则分析器。必须在 legacy agent 同步前补建，目录存在时不覆盖现场修改。
        try:
            from app.gateway.routers._elasticsearch_offline_ops_seed import (
                ensure_elasticsearch_offline_ops_agent,
            )

            ensure_elasticsearch_offline_ops_agent()
        except Exception:
            logger.exception(
                "Elasticsearch offline operations agent seeding failed at startup (non-fatal)"
            )
        # 内置「任务研判报告助手」（agentfx 任务深链固定使用）：同样必须在
        # _sync_legacy_agents 之前补建目录 + 补建配套 task-report-import 技能
        # （skills/ 被 .gitignore 排除），由后者 upsert 进 agents 表。
        try:
            from app.gateway.routers._agentfx_seed import ensure_agentfx_agent

            ensure_agentfx_agent()
        except Exception:
            logger.exception(
                "Agentfx analyst agent seeding failed at startup (non-fatal)"
            )

        # 内置「任务分析总控」（任务分析定时的专用总控：无人值守分解任务并派活）。
        try:
            from app.gateway.routers._task_analysis_seed import (
                ensure_task_analysis_functional_agents,
            )

            ensure_task_analysis_functional_agents()
        except Exception:
            logger.exception(
                "Task analysis coordinator 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)")

                # Synchronize every worker's strategy registry only after the
                # folder agents have been imported.  In memory mode this also
                # restores dynamic/default angle lifecycle from sidecars.
                try:
                    await strategy_agents.ensure_strategy_agent_registry(app)
                    logger.info("Strategy agent registry synchronized for worker")
                except Exception:
                    logger.exception("Strategy agent role seeding failed at startup (non-fatal)")
        except Exception:
            logger.exception("Legacy agent sync failed at startup (non-fatal)")

        # Refresh after database initialization and all built-in skill seeders;
        # import-time priming may have run before either was ready.
        try:
            from deerflow.agents.lead_agent.prompt import refresh_skills_system_prompt_cache_async

            await refresh_skills_system_prompt_cache_async()
        except Exception:
            logger.exception("Skills prompt cache warm-up 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)")

        # Concurrency (system-pressure) sampler — lightweight, singleton-locked,
        # runtime-toggleable via system_settings.concurrency_monitor.enabled.
        try:
            app.state.concurrency_sampler_task = asyncio.create_task(_concurrency_sampler_loop(app))
        except Exception:
            logger.exception("Concurrency sampler failed to start (non-fatal)")

        try:
            app.state.admin_audit_retention_task = asyncio.create_task(
                _admin_audit_retention_loop(app)
            )
        except Exception:
            logger.exception("Admin audit retention 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")

        # External-task chain-binding scheduler (任务分析定时)。
        task_analysis_enabled = os.environ.get("TASK_ANALYSIS_ENABLED", "true").strip().lower()
        if task_analysis_enabled in {"0", "false", "no", "off"}:
            logger.info("Task analysis service disabled by TASK_ANALYSIS_ENABLED")
        else:
            try:
                from deerflow.runtime.task_analysis import TaskAnalysisService

                task_analysis_store = getattr(app.state, "task_analysis_store", None)
                if task_analysis_store is not None:
                    ta_service = TaskAnalysisService(app, task_analysis_store)
                    app.state.task_analysis_service = ta_service
                    await ta_service.start()
                    logger.info("Task analysis service started")
            except Exception:
                logger.exception("Task analysis service failed to start")

        if getattr(app.state, "worker_pool", None) is not None:
            app.state.worker_pool.start()
        yield

        artifact_library_tasks = list(
            getattr(app.state, "artifact_library_sync_tasks", {}).values()
        )
        for task in artifact_library_tasks:
            task.cancel()
        if artifact_library_tasks:
            await asyncio.gather(*artifact_library_tasks, return_exceptions=True)

        llmwiki_index_scheduler_task = getattr(app.state, "llmwiki_index_scheduler_task", None)
        if llmwiki_index_scheduler_task is not None:
            llmwiki_index_scheduler_task.cancel()
            try:
                await llmwiki_index_scheduler_task
            except asyncio.CancelledError:
                pass
            except Exception:
                logger.exception("Local Wiki index scheduler shutdown error")
        llmwiki_index_jobs = list(getattr(app.state, "llmwiki_index_jobs", set()))
        for job in llmwiki_index_jobs:
            job.cancel()
        if llmwiki_index_jobs:
            await asyncio.gather(*llmwiki_index_jobs, return_exceptions=True)

        # ?? 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")

        concurrency_sampler_task = getattr(app.state, "concurrency_sampler_task", None)
        if concurrency_sampler_task is not None:
            concurrency_sampler_task.cancel()
            try:
                await concurrency_sampler_task
            except asyncio.CancelledError:
                pass
            except Exception:
                logger.exception("Concurrency sampler shutdown error")

        admin_audit_retention_task = getattr(app.state, "admin_audit_retention_task", None)
        if admin_audit_retention_task is not None:
            admin_audit_retention_task.cancel()
            try:
                await admin_audit_retention_task
            except asyncio.CancelledError:
                pass
            except Exception:
                logger.exception("Admin audit retention 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)

        try:
            ta_service = getattr(app.state, "task_analysis_service", None)
            if ta_service is not None:
                await ta_service.stop()
        except Exception:
            logger.exception("Failed to stop task analysis service")

        # 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")


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
    from app.gateway.worker_pool.lifecycle import managed_controller
    from app.gateway.perf_logging import event_loop_watchdog, register_stack_dump_signal

    register_stack_dump_signal()
    watchdog = asyncio.create_task(event_loop_watchdog(app), name="gateway-event-loop-watchdog")
    try:
        async with managed_controller(app):
            async with _gateway_lifespan(app):
                yield
    finally:
        watchdog.cancel()
        await asyncio.gather(watchdog, return_exceptions=True)
        from app.gateway.api_read_cache import close_api_read_cache

        await close_api_read_cache()


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

    Returns:
        Configured FastAPI application instance.
    """
    config = get_gateway_config()
    logger.info("API documentation endpoints enabled: %s", config.enable_docs)
    docs_kwargs = {
        "docs_url": None,
        "redoc_url": None,
        "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",
            },
        ],
    )

    @app.middleware("http")
    async def invalidate_catalog_cache_after_write(request: Request, call_next):
        response = await call_next(request)
        if request.method.upper() in {"POST", "PUT", "PATCH", "DELETE"} and response.status_code < 400:
            path = request.url.path.rstrip("/")
            collections: set[str] = set()
            if path == "/api/agents" or path.startswith("/api/agents/"):
                collections.update(("agents", "tags"))
            if path == "/api/skills" or path.startswith("/api/skills/"):
                collections.update(("skills", "tags"))
            if path == "/api/tags" or path.startswith("/api/tags/"):
                collections.update(("agents", "skills", "recommended_questions", "light_apps", "tags"))
            if path == "/api/positions" or path.startswith("/api/positions/"):
                collections.update(("agents", "skills", "recommended_questions"))
            if path == "/api/user-preferences" or path.startswith("/api/user-preferences/"):
                collections.add("user_preferences")
            if path == "/api/user-prompts" or path.startswith("/api/user-prompts/"):
                collections.add("user_prompts")
            if path.startswith("/api/recommended-questions"):
                collections.update(("recommended_questions", "tags"))
            if path == "/api/scheduled-tasks" or path.startswith("/api/scheduled-tasks/"):
                collections.add("tags")
            if path == "/api/light-apps" or path.startswith("/api/light-apps/"):
                collections.add("tags")
            if path == "/api/sensitive-words":
                collections.add("sensitive_words")
            config = getattr(request.app.state, "config", None)
            if config is not None and collections:
                from app.gateway.api_read_cache import invalidate_cached_collection

                await asyncio.gather(
                    *(
                        invalidate_cached_collection(
                            config,
                            collection,
                            force=True,
                        )
                        for collection in collections
                    )
                )
        return response

    if config.enable_docs:
        # Keep these paths relative to the page URL.  Nginx strips the
        # /deerflow deployment prefix before forwarding to this app, and
        # root-relative URLs would otherwise escape that prefix in browsers.
        app.mount(
            "/docs/static",
            StaticFiles(directory=_DOCS_STATIC_DIR),
            name="docs-static",
        )

        @app.get("/docs", include_in_schema=False)
        async def swagger_ui() -> HTMLResponse:
            return get_swagger_ui_html(
                openapi_url="./openapi.json",
                title=f"{app.title} - Swagger UI",
                swagger_js_url="./docs/static/swagger-ui-bundle.js",
                swagger_css_url="./docs/static/swagger-ui.css",
                swagger_favicon_url="./docs/static/favicon-32x32.png",
                swagger_ui_parameters=app.swagger_ui_parameters,
            )

        @app.get("/redoc", include_in_schema=False)
        async def redoc() -> HTMLResponse:
            return get_redoc_html(
                openapi_url="./openapi.json",
                title=f"{app.title} - ReDoc",
                redoc_js_url="./docs/static/redoc.standalone.js",
                redoc_favicon_url="./docs/static/favicon-32x32.png",
                with_google_fonts=False,
            )

    # Auth: reject unauthenticated requests to non-public paths (fail-closed safety net)
    from app.gateway.worker_pool.proxy import WorkerProxy

    app.add_middleware(WorkerProxy)
    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()

    # Keep standard local frontend origins available even when the startup
    # script supplies an explicit production/development allow-list. Vite can
    # choose 5173 or move to 5174 when that port is already occupied; excluding
    # the former makes every browser preflight fail after a backend restart.
    local_vite_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",
    ]

    # The legacy TaskCOP Vue application uses its own Vite development port.
    taskcop_vite_origins = [
        "http://localhost:8080",
        "http://127.0.0.1:8080",
    ]
    local_dev_origins = [*local_vite_origins, *taskcop_vite_origins]

    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 []
    if cors_origins:
        cors_origins = list(dict.fromkeys([*cors_origins, *local_dev_origins]))

    # 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",
    ]

    # Public pages (e.g. http://47.x.x.x:7010) calling this Gateway on localhost
    # trigger Chrome/Edge Private Network Access preflights. Without
    # allow_private_network=True the browser reports a generic CORS failure even
    # when Access-Control-Allow-Origin is correct.
    cors_kwargs = {
        "allow_credentials": True,
        "allow_methods": ["*"],
        "allow_headers": ["*"],
        "expose_headers": cors_expose_headers,
        "allow_private_network": True,
    }

    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".+",
            **cors_kwargs,
        )
    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,
                **cors_kwargs,
            )
    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 = local_dev_origins
        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,
            **cors_kwargs,
        )

    # This one anonymous API is intentionally callable by browser code from
    # every origin. Keep the override route-scoped so authenticated APIs retain
    # the deployment's normal CORS allow-list.
    app.add_middleware(PublicKnowledgeCORSMiddleware)

    # Include routers
    # Models API is mounted at /api/models
    app.include_router(models.router)
    # Administrator-managed config.yaml model settings and connection tests.
    app.include_router(model_management.router)

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

    # DeerFlow-native LLMWiki management/search BFF (conditionally backed by WeKnora).
    app.include_router(llmwiki.router)
    app.include_router(llmwiki_index.router)
    app.include_router(external_llmwiki.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)
    # Admin skill distillation matrix + global assistant knowledge base.
    app.include_router(skill_knowledge.router)
    app.include_router(assistant_knowledge.router)
    app.include_router(assistant_knowledge.export_router)
    from app.gateway.routers import wiki_packages

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

    # Web App site artifacts (multi-page site / flowchart / map) with signed
    # public static hosting at /api/public/site-preview/{token}/
    app.include_router(web_app_sites.router)

    # Uploads API is mounted at /api/threads/{thread_id}/uploads
    app.include_router(uploads.router)
    # Global database-backed upload limits at /api/upload-policy.
    app.include_router(upload_policy.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}
    from app.gateway.routers import analysis_projects

    app.include_router(analysis_projects.router)
    app.include_router(threads.router)
    # Dedicated taskId + goPath history lookup for single-agent task deep-link chats.
    app.include_router(task_deeplink_threads.router)
    app.include_router(task_deeplink_details.router)
    # Agents API is mounted at /api/agents
    app.include_router(agents.router)
    app.include_router(strategy_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)

    # Task-scoped roundtable drafts (taskId 深链聊天记录，独立存、不分权) at /api/roundtable-task-drafts
    app.include_router(roundtable_task_drafts.router)

    # Task-scoped artifact submission status; stores references only, never artifact content.
    app.include_router(roundtable_artifact_submissions.router)

    # Step3 流程图结构化抽取 + 校验 (SSE) at /api/flow-extract
    app.include_router(flow_extract.router)

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

    # 岗位协同独立会话 / 节点状态（不使用原圆桌总控或后台作业）
    # Canonical position-workspace persistence API.
    app.include_router(position_roundtable.router, prefix="/api")
    # Compatibility route for intranet gateways whose allowlist only exposes
    # the established `/api/multi-agent/*` namespace.
    app.include_router(
        position_roundtable.router,
        prefix="/api/multi-agent",
        include_in_schema=False,
    )

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

    # Position-roundtable work-view role configuration at /api/position-roles.
    app.include_router(position_roles.router)

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

    # Workflow Studio: definitions / runs / SSE / data sources / embed / Coze compat.
    # Static prefixes (node-types, resources, planning, runs, data-sources, embed) must be
    # registered before the /{workflow_id} catch-alls in workflows.router.
    app.include_router(workflow_resources.router)
    app.include_router(workflow_data_sources.router)
    app.include_router(workflow_embed.router)
    app.include_router(workflow_planning.router)
    app.include_router(workflow_runs.router)
    app.include_router(workflow_studio.router)
    app.include_router(workflows.router)
    app.include_router(workflows_coze_compat.router)
    app.include_router(workflows_coze_compat.chrome_router)

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

    # Compact/sidebar-summary menu overrides at /api/compact-menu-overrides
    app.include_router(compact_menu_overrides.router)

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

    # Public-document workbench: durable originals, review snapshots and DOCX versions.
    app.include_router(gongwen_documents.router)

    # Sentiment-analysis virtual agent AG-UI SSE proxy at /api/sentiment-agent
    app.include_router(sentiment_agent.router)

    # Skill URL/IP batch replace (技能地址管理) at /api/skill-addresses
    app.include_router(skill_addresses.router)

    # 大屏绘制智能体独立页面会话（按 taskId 存取，不分权）at /api/dashboard-sessions
    app.include_router(dashboard_sessions.router)

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

    # Admin exact-match question/answer short-circuit at /api/fixed-questions
    app.include_router(fixed_questions.router)

    # AgentScope report-collaboration workbench at /api/report-collaboration
    app.include_router(report_collaboration.router)

    # Deep-research report structures (工作流配置 / 报告结构) at /api/report-structures
    app.include_router(report_structures.router)

    # Temporary compatibility bridge for the TaskCOP situation-overview frontend.
    app.include_router(taskcop_tasks.router)

    # Task-scoped situation-report details used by dashboard drawing / task gates.
    app.include_router(task_reports.router)

    # Force-situation task detail documents (new task type; free-form JSON).
    app.include_router(force_situation.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)
    app.include_router(roundtable_rooms.router)
    app.include_router(roundtable_diagnostics.router)

    # Deep Research (深度研究) API at /api/deep-research
    app.include_router(deep_research.router)

    # The independent enterprise-research workbench is temporarily disabled.
    # Do not mount /api/enterprise-research while its MySQL schema is inactive.

    # 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)

    # External-task multi-agent scheduled analysis (/api/task-analysis)
    app.include_router(task_analysis.router)
    app.include_router(task_analysis_qa.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_runtime.router)
    app.include_router(html_images.router)
    app.include_router(html_page_favorites.router)
    # HTML 模板管理(模板管理页;读开放,写仅 admin)。
    app.include_router(html_templates.router)

    # Product knowledge base API is mounted at /api/knowledge
    app.include_router(knowledge.router)
    # Anonymous, section-level vector search over every published Wiki base
    # except the system conversation-deposit base.
    app.include_router(public_knowledge_vector_search.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)
    app.include_router(app_identity.router)
    app.include_router(app_identity.public_router)
    from app.gateway.worker_pool.router import router as worker_pool_router

    app.include_router(worker_pool_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)
    app.include_router(admin_audit_retention_routes.router)

    app.include_router(admin_active_runs.router)
    app.include_router(skill_view_diagnostics.router)
    app.include_router(model_exchanges.router)

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

    # Admin model-stream truncation event log
    app.include_router(stream_truncations.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)
    # Admin monthly archive/rollback for PostgreSQL chat checkpoints.
    app.include_router(chat_archive.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)
    app.include_router(service_tokens.router)

    # Feedback API is mounted at /api/threads/{thread_id}/runs/{run_id}/feedback
    app.include_router(feedback.router)
    # Aggregate user-feedback answer quality (not a complete capability score).
    app.include_router(feedback.agent_score_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)

    # 开放问答接口 /api/open/chat —— 无需登录,指定模型 + 智能体做问答,可关闭思考模式
    app.include_router(open_chat.router)

    # 极速问答 /api/direct-chat/stream —— 直连大模型流式输出,不走 lead agent / 工具 / 技能
    app.include_router(direct_chat.router)

    # 开放接口 /api/open/3qfx/ask —— 无需登录,按 taskId 直接发起 3qfx 第一次问答(后台运行,落成真实会话)
    app.include_router(open_3qfx.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?

    # Health probes are always unauthenticated (see
    # auth_middleware.PUBLIC_HEALTH_PATHS). Support the bare Gateway path,
    # API base URL, and LangGraph proxy base URL for deployment monitors.
    @app.api_route("/health", methods=["GET", "HEAD"], tags=["health"])
    @app.api_route("/api/health", methods=["GET", "HEAD"], tags=["health"])
    @app.api_route("/api/langgraph/health", methods=["GET", "HEAD"], tags=["health"])
    async def health_check() -> dict:
        """Health check endpoint.

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

    # WeKnora's embedded frontend uses root-relative routes such as /platform,
    # /assets, /locales and /api/v1/*. Register that proxy last so it stays a
    # true fallback and cannot intercept DeerFlow APIs.
    # Register the public-prefix routes before the unprefixed static fallback.
    # Otherwise proxy_router's /{proxied_path:path} route consumes every
    # /deerflow/* request before the prefixed platform/assets routes can match.
    app.include_router(llmwiki.proxy_router, prefix="/deerflow")
    app.include_router(llmwiki.proxy_router)

    from app.gateway.worker_pool.telemetry import RequestTelemetryMiddleware
    app.add_middleware(RequestTelemetryMiddleware)

    return app


# Create app instance for uvicorn
app = create_app()