"""Centralized accessors for singleton objects stored on ``app.state``.

**Getters** (used by routers): raise 503 when a required dependency is
missing, except ``get_store`` which returns ``None``.

Initialization is handled directly in ``app.py`` via :class:`AsyncExitStack`.
"""

from __future__ import annotations

import asyncio
import logging
from collections.abc import AsyncGenerator, Callable
from contextlib import AsyncExitStack, asynccontextmanager
from typing import TYPE_CHECKING, Any, TypeVar, cast

from fastapi import FastAPI, HTTPException, Request
from langgraph.types import Checkpointer

from deerflow.config.app_config import AppConfig
from deerflow.persistence.feedback import FeedbackRepository
from deerflow.runtime import RunContext, RunManager, StreamBridge
from deerflow.runtime.events.store.base import RunEventStore
from deerflow.runtime.runs.store.base import RunStore

logger = logging.getLogger(__name__)

if TYPE_CHECKING:
    from app.gateway.auth.local_provider import LocalAuthProvider
    from app.gateway.auth.repositories.sqlite import SQLiteUserRepository
    from deerflow.persistence.agents import AgentStore
    from deerflow.persistence.notifications import NotificationStore
    from deerflow.persistence.scheduled_tasks import ScheduledTaskStore
    from deerflow.persistence.skills import SkillStore
    from deerflow.persistence.tags import TagStore
    from deerflow.persistence.thread_meta.base import ThreadMetaStore


T = TypeVar("T")


def get_gateway_worker_topology() -> dict[str, Any]:
    """Detect the current worker count and source.

    Returns ``{"worker_count": int|None, "source": str, "is_multi_worker": bool}``.
    Sources checked in order: ``WEB_CONCURRENCY`` env var, ``/proc`` scan (Linux).
    """
    import os as _os

    wc = _os.environ.get("WEB_CONCURRENCY", "").strip()
    if wc and wc.isdigit():
        count = int(wc)
        return {"worker_count": count, "source": "WEB_CONCURRENCY", "is_multi_worker": count > 1}

    if not _os.name == "posix":
        return {"worker_count": None, "source": "unknown", "is_multi_worker": False}

    try:
        from pathlib import Path as _Path

        peers = 0
        proc = _Path("/proc")
        if proc.exists():
            for entry in 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 > 0:
            return {"worker_count": peers, "source": "process_scan", "is_multi_worker": peers > 1}
    except Exception:  # noqa: BLE001
        pass

    return {"worker_count": None, "source": "unknown", "is_multi_worker": False}


async def _background_reconcile_loop(service: Any, *, interval: int) -> None:
    """Periodically reconcile active conversation runs."""
    while True:
        try:
            await asyncio.sleep(interval)
            await service.reconcile_active_conversations()
        except asyncio.CancelledError:
            raise
        except Exception:  # noqa: BLE001
            logger.warning("[maf-roundtable] background reconcile failed", exc_info=True)


async def _startup_reconcile(service: Any) -> None:
    """One-shot reconcile of all active runs at gateway startup."""
    try:
        written = await service.reconcile_active_conversations()
        if written:
            logger.info("[maf-roundtable] startup reconcile wrote %d conversation events", written)
    except Exception:  # noqa: BLE001
        logger.warning("[maf-roundtable] startup reconcile failed", exc_info=True)


def resolve_roundtable_gateway_mode(raw: str | None) -> str:
    """解析 ``ROUNDTABLE_JOB_GATEWAY``：**默认 ``inprocess``（真模型后台挂起）**。

    只有显式写 ``simulated`` 才用模拟网关（前端联调用：真实进度、不调真模型）；
    缺省 / 空串 / 其它值一律落到 ``inprocess`` —— 部署时无需在 .env 里配置任何值，
    后台挂起开箱即用（Phase 3b 已在 Windows 本地与内网 Docker 验证）。
    """
    mode = (raw or "").strip().lower()
    return "simulated" if mode == "simulated" else "inprocess"


def get_config(request: Request) -> AppConfig:
    """Return the app-scoped ``AppConfig`` stored on ``app.state``."""
    config = getattr(request.app.state, "config", None)
    if config is None:
        raise HTTPException(status_code=503, detail="Configuration not available")
    return config


@asynccontextmanager
async def langgraph_runtime(app: FastAPI) -> AsyncGenerator[None, None]:
    """Bootstrap and tear down all LangGraph runtime singletons.

    Usage in ``app.py``::

        async with langgraph_runtime(app):
            yield
    """
    from deerflow.persistence.engine import close_engine, get_session_factory, init_engine_from_config
    from deerflow.runtime import make_store, make_stream_bridge
    from deerflow.runtime.checkpointer.async_provider import make_checkpointer
    from deerflow.runtime.events.store import make_run_event_store

    async with AsyncExitStack() as stack:
        config = getattr(app.state, "config", None)
        if config is None:
            raise RuntimeError("langgraph_runtime() requires app.state.config to be initialized")

        app.state.stream_bridge = await stack.enter_async_context(make_stream_bridge(config))

        # Initialize persistence engine BEFORE checkpointer so that
        # auto-create-database logic runs first (postgres backend).
        await init_engine_from_config(config.database)

        app.state.checkpointer = await stack.enter_async_context(make_checkpointer(config))
        app.state.store = await stack.enter_async_context(make_store(config))

        # Initialize repositories — one get_session_factory() call for all.
        sf = get_session_factory()
        if sf is not None:
            from deerflow.persistence.feedback import FeedbackRepository
            from deerflow.persistence.run import RunRepository

            app.state.run_store = RunRepository(sf)
            app.state.feedback_repo = FeedbackRepository(sf)
        else:
            from deerflow.runtime.runs.store.memory import MemoryRunStore

            app.state.run_store = MemoryRunStore()
            app.state.feedback_repo = None

        from deerflow.persistence.thread_meta import make_thread_store

        app.state.thread_store = make_thread_store(sf, app.state.store)

        from deerflow.persistence.scheduled_tasks import make_scheduled_task_store

        app.state.scheduled_task_store = make_scheduled_task_store(sf)

        from deerflow.persistence.html_pages import make_html_page_store

        app.state.html_page_store = make_html_page_store(sf)

        from deerflow.persistence.llm_metrics import make_llm_metrics_store

        app.state.llm_metrics_store = make_llm_metrics_store(sf)

        from deerflow.persistence.concurrency import make_concurrency_sample_store

        app.state.concurrency_sample_store = make_concurrency_sample_store(sf)

        from deerflow.persistence.tool_metrics import make_tool_metrics_store

        app.state.tool_metrics_store = make_tool_metrics_store(sf)

        from deerflow.persistence.agents import make_agent_store

        app.state.agent_store = make_agent_store(sf)

        from deerflow.persistence.skills import make_skill_store

        app.state.skill_store = make_skill_store(sf)

        from deerflow.persistence.notifications import (
            make_notification_store,
            set_notification_store,
        )

        app.state.notification_store = make_notification_store(sf)
        # Make the sender helper (harness layer) able to write without going
        # through FastAPI request context — used by cascade.py.
        set_notification_store(app.state.notification_store)
        from deerflow.persistence.recommended_questions import make_recommended_question_store

        app.state.recommended_question_store = make_recommended_question_store(sf)

        from deerflow.persistence.tags import make_tag_store

        app.state.tag_store = make_tag_store(sf)

        from deerflow.persistence.positions import make_position_store

        app.state.position_store = make_position_store(sf)

        from deerflow.services.position_sync import PositionSyncService

        app.state.position_sync = PositionSyncService(sf, app.state.position_store, app.state.tag_store)

        from deerflow.persistence.thread_shares import make_thread_share_store

        app.state.thread_share_store = make_thread_share_store(sf)

        from deerflow.persistence.embed_sessions import make_embed_session_store

        app.state.embed_session_store = make_embed_session_store(sf)

        from deerflow.persistence.ai_writing_sessions import (
            make_ai_writing_session_store,
        )
        from deerflow.persistence.ai_writing_sessions import (
            register_default_store as _register_ai_writing_default_store,
        )

        app.state.ai_writing_session_store = make_ai_writing_session_store(sf)
        # 让 LangGraph Server 路径下的 AI 写作节点能跨 app/harness 边界拿到 repo。
        # 老 /start /stream /resume 路由仍走 request.app.state，互不影响。
        _register_ai_writing_default_store(app.state.ai_writing_session_store)

        # AI 写作「后台挂起」执行器：把写作图作为后台任务驱动到定稿（离页/刷新/切号不打断）。
        # 用 app.state.checkpointer（与运行时 worker 同一份 saver）续上线程 checkpoint。
        from app.gateway.ai_writing_job_executor import AIWritingJobExecutor

        app.state.ai_writing_job_executor = AIWritingJobExecutor(app)
        # 进程重启续跑：把仍标记为 background 的会话重新拉起驱动（best-effort，不阻塞启动）。
        if app.state.ai_writing_session_store is not None:
            from app.gateway.ai_writing_job_executor import STATUS_BACKGROUND

            async def _reconcile_ai_writing_background() -> None:
                try:
                    sessions = await app.state.ai_writing_session_store.list_by_status(STATUS_BACKGROUND)
                    app.state.ai_writing_job_executor.reconcile(sessions=sessions)
                except Exception:  # noqa: BLE001
                    logger.warning("[ai-writing-bg] startup reconcile failed", exc_info=True)

            asyncio.create_task(_reconcile_ai_writing_background())

        from deerflow.persistence.article_types import make_article_type_store

        app.state.article_type_store = make_article_type_store(sf)

        from deerflow.persistence.research_templates import make_research_template_store

        app.state.research_template_store = make_research_template_store(sf)

        from deerflow.persistence.memory_config import make_user_memory_config_store

        app.state.memory_config_store = make_user_memory_config_store(sf)

        from deerflow.knowledge import make_knowledge_service

        app.state.knowledge_service = make_knowledge_service(sf, config.knowledge)

        # Seed the built-in extraction prompts as editable templates so operators
        # can immediately view/edit them (idempotent; never overwrites edits).
        if app.state.knowledge_service is not None:
            try:
                await app.state.knowledge_service.seed_default_templates()
            except Exception:
                logger.exception("Failed to seed knowledge extraction templates")

        # Background auto-sedimentation queue (phase 3) — drains finished
        # conversations off the chat path. Worker only does work when
        # ``knowledge.auto_ingest_enabled`` is set; the queue is always wired so
        # the middleware can enqueue without a hard dependency.
        from deerflow.knowledge.auto_ingest import AutoIngestQueue, set_auto_ingest_queue

        if app.state.knowledge_service is not None:
            _auto_ingest_queue = AutoIngestQueue(
                lambda: getattr(app.state, "knowledge_service", None),
                min_chars=config.knowledge.auto_ingest_min_chars,
            )
            _auto_ingest_queue.start()
            set_auto_ingest_queue(_auto_ingest_queue)
            app.state.knowledge_auto_ingest_queue = _auto_ingest_queue

        from deerflow.persistence.user_prompts import make_user_prompt_store

        app.state.user_prompt_store = make_user_prompt_store(sf)

        from deerflow.persistence.user_preferences import make_user_preferences_store

        app.state.user_preferences_store = make_user_preferences_store(sf)

        from deerflow.persistence.roundtable_drafts import make_roundtable_draft_store

        app.state.roundtable_draft_store = make_roundtable_draft_store(sf)

        # Task-scoped roundtable drafts (taskId 深链聊天记录，独立存、不按 user 分权)。
        from deerflow.persistence.roundtable_task_drafts import make_roundtable_task_draft_store

        app.state.roundtable_task_draft_store = make_roundtable_task_draft_store(sf)

        from deerflow.persistence.roundtable_draft_shares import make_roundtable_draft_share_store

        app.state.roundtable_draft_share_store = make_roundtable_draft_share_store(sf)

        from deerflow.persistence.roundtable_chains import make_roundtable_chain_store

        app.state.roundtable_chain_store = make_roundtable_chain_store(sf)

        # 独立的岗位协同 Session / Node 存储。它只索引业务流程状态，真实消息和
        # 文件仍复用 LangGraph thread/checkpoint 与现有 artifact API。
        from deerflow.persistence.position_roundtable import make_position_roundtable_store

        app.state.position_roundtable_store = make_position_roundtable_store(sf)

        # 岗位会商工作视角（名称 / 顺序 / 特殊意图岗）是全局配置，和组织/RBAC
        # 岗位分开保存。启动只补齐缺失内置项，不覆盖管理员的既有修改。
        from app.gateway.position_role_defaults import POSITION_ROLE_SEED
        from deerflow.persistence.position_roles import make_position_role_store

        app.state.position_role_store = make_position_role_store(sf)
        await app.state.position_role_store.ensure_seed(POSITION_ROLE_SEED)

        from deerflow.persistence.business_mapping import make_business_mapping_store

        app.state.business_mapping_store = make_business_mapping_store(sf)

        from deerflow.persistence.light_apps import make_light_app_store

        app.state.light_app_store = make_light_app_store(sf)

        from deerflow.persistence.menu_overrides import make_menu_override_store

        app.state.menu_override_store = make_menu_override_store(sf)

        from deerflow.persistence.sensitive_words import make_sensitive_word_store

        app.state.sensitive_word_store = make_sensitive_word_store(sf)

        from deerflow.persistence.dashboard_sessions import make_dashboard_session_store

        app.state.dashboard_session_store = make_dashboard_session_store(sf)

        from deerflow.persistence.task_buttons import make_task_button_store

        app.state.task_button_store = make_task_button_store(sf)

        from deerflow.persistence.taskcop_tasks import make_taskcop_task_store

        app.state.taskcop_task_store = make_taskcop_task_store(sf)

        from deerflow.persistence.task_reports import make_task_report_store

        app.state.task_report_store = make_task_report_store(sf)

        from deerflow.persistence.parallel_agent_tasks import make_parallel_agent_task_store

        app.state.parallel_agent_task_store = make_parallel_agent_task_store(sf)

        # ── MAF roundtable experiment (Microsoft Agent Framework 多智能体会商) ──
        # Independent, feature-flag-gated sibling of the legacy roundtable. Six
        # isolated tables (no FK to roundtable_*). The store is always wired
        # (so reads work for admin history review) but the **service** is only
        # constructed when ``maf_roundtable.enabled`` is on — that is the gate
        # the router checks; when off, write paths return 503 and the MAF SDK
        # is never imported. Initialization failures are logged but never
        # abort Gateway startup (the feature is experimental / opt-in).
        from deerflow.persistence.maf_roundtable import make_maf_roundtable_store

        app.state.maf_roundtable_store = make_maf_roundtable_store(sf)
        maf_enabled = False
        maf_dispatch_ready = False
        maf_dispatch_status = "configuration_unavailable"
        try:
            from app.maf_roundtable import MafRoundtableService

            maf_cfg = getattr(config, "maf_roundtable", None)
            maf_enabled = bool(getattr(maf_cfg, "enabled", False)) if maf_cfg is not None else False
            if maf_cfg is not None:
                from app.maf_roundtable.dispatch_contracts import MafDispatchCatalog

                maf_dispatch_ready, maf_dispatch_status = MafDispatchCatalog(maf_cfg).readiness()
                if maf_enabled and not maf_dispatch_ready:
                    logger.error(
                        "[maf-roundtable] enabled but no trusted manager route is configured; "
                        "forcing the conversation entry OFF (%s)",
                        maf_dispatch_status,
                    )
                    maf_enabled = False
            maf_admin_diag = (
                bool(getattr(maf_cfg, "admin_diagnostic_access", True)) if maf_cfg is not None else True
            )
            # P0-4: when MAF is switched on, its full safety stack must be in
            # place (write_guard mounted + provider complete + bridge resolvable
            # + MAF_ACTION_SECRET_KEY set). If any prerequisite is missing the
            # feature is forced OFF (clearly logged) — never "MAF on, approval
            # guard not mounted". Reads stay open for admin history review.
            if maf_enabled:
                from app.maf_roundtable.service import (
                    MafConfigurationError,
                    validate_maf_prerequisites,
                )

                try:
                    validate_maf_prerequisites()
                except MafConfigurationError as cfg_exc:
                    logger.error(
                        "[maf-roundtable] enabled but prerequisites unmet — "
                        "forcing the feature OFF: %s",
                        cfg_exc,
                    )
                    maf_enabled = False
                    maf_dispatch_ready = False
            if app.state.maf_roundtable_store is not None:
                app.state.maf_roundtable_service = MafRoundtableService(
                    app.state.maf_roundtable_store,
                    enabled=maf_enabled,
                    admin_diagnostic_access=maf_admin_diag,
                    config=maf_cfg,
                    agent_store=getattr(app.state, "agent_store", None),
                    app_config=config,
                )
                # Wire the WriteGuard bridge so the harness middleware can reach
                # the service when intercepting external-write tool calls.
                if app.state.maf_roundtable_service is not None:
                    from app.maf_roundtable.write_guard_bridge import register_write_guard_service

                    register_write_guard_service(app.state.maf_roundtable_service)
            else:
                app.state.maf_roundtable_service = None
        except Exception:  # noqa: BLE001 — experimental feature must never abort startup
            logger.exception("[maf-roundtable] service initialization failed; feature stays off")
            app.state.maf_roundtable_service = None

        # ── MAF roundtable V2 Conversation service + SSE event hub ──
        # P0-2: multi-worker guard — memory hub cannot broadcast across
        # workers. When detected, disable the conversation SSE write path
        # (reads stay open for admin history review).
        _topology = get_gateway_worker_topology()
        _maf_cfg = getattr(config, "maf_roundtable", None)
        _conv_backend = getattr(_maf_cfg, "conversation_event_backend", "memory")
        _require_single = getattr(_maf_cfg, "require_single_worker_for_conversation_sse", True)
        _stream_blocked = (
            _conv_backend == "memory"
            and _topology["is_multi_worker"]
            and _require_single
        )
        if _stream_blocked:
            logger.error(
                "[maf-roundtable] conversation_event_backend=memory with %d workers "
                "(%s) — SSE stream disabled. Set WEB_CONCURRENCY=1 or deploy Redis "
                "Pub/Sub before enabling multi-worker conversation SSE.",
                _topology["worker_count"],
                _topology["source"],
            )

        try:
            from app.maf_roundtable.conversation_event_hub import ConversationEventHub

            if _stream_blocked:
                app.state.maf_conversation_event_hub = None
            else:
                app.state.maf_conversation_event_hub = ConversationEventHub()

            if app.state.maf_roundtable_store is not None:
                from app.maf_roundtable.conversation_service import MafConversationService
                from app.maf_roundtable.dispatcher import get_dispatcher

                app.state.maf_conversation_service = MafConversationService(
                    app.state.maf_roundtable_store,
                    maf_config=_maf_cfg,
                    event_hub=app.state.maf_conversation_event_hub,
                    agent_store=getattr(app.state, "agent_store", None),
                    app_config=config,
                    dispatcher=get_dispatcher(app),
                    enabled=maf_enabled,
                )
            else:
                app.state.maf_conversation_service = None
        except Exception:  # noqa: BLE001
            logger.exception("[maf-roundtable] conversation service init failed")
            app.state.maf_conversation_service = None
            app.state.maf_conversation_event_hub = None

        app.state.maf_conversation_dispatch_ready = bool(maf_enabled and maf_dispatch_ready)
        app.state.maf_conversation_dispatch_status = maf_dispatch_status
        app.state.maf_conversation_stream_ready = not _stream_blocked and app.state.maf_conversation_event_hub is not None
        app.state.maf_conversation_stream_status = (
            "multi_worker_memory_hub_unsupported" if _stream_blocked
            else "ok" if app.state.maf_conversation_event_hub is not None
            else "hub_not_initialized"
        )
        app.state.maf_conversation_event_backend = _conv_backend

        # ── Background reconcile task for conversation event projection ──
        _reconcile_task: asyncio.Task | None = None
        if (
            app.state.maf_conversation_service is not None
            and maf_enabled
            and not _stream_blocked
        ):
            _reconcile_interval = getattr(_maf_cfg, "background_reconcile_interval_seconds", 10)
            _reconcile_task = asyncio.create_task(
                _background_reconcile_loop(
                    app.state.maf_conversation_service,
                    interval=max(5, int(_reconcile_interval)),
                )
            )
            app.state.maf_conversation_reconcile_task = _reconcile_task
            logger.info("[maf-roundtable] background reconcile started (interval=%ss)", _reconcile_interval)

            # Startup reconcile: scan active/recent runs once
            asyncio.create_task(_startup_reconcile(app.state.maf_conversation_service))

        from deerflow.persistence.roundtable_jobs import (
            make_roundtable_job_store,
        )
        from deerflow.persistence.roundtable_jobs import (
            register_default_store as _register_roundtable_job_default_store,
        )

        app.state.roundtable_job_store = make_roundtable_job_store(sf)
        # 让 LangGraph Server 路径下的圆桌编排图节点能跨 app/harness 边界拿到 repo
        # （Phase 2 后台编排图写进度用）。
        _register_roundtable_job_default_store(app.state.roundtable_job_store)

        # 圆桌会商「诊断日志」store：会商全流程各处报错 / 关键事件落库，供管理员诊断页筛选。
        # 同样登记进程级默认仓储——harness 侧编排引擎 / 进程内网关跨边界写诊断用。
        from deerflow.persistence.roundtable_diagnostics import (
            make_roundtable_diagnostic_store,
        )
        from deerflow.persistence.roundtable_diagnostics import (
            register_default_store as _register_roundtable_diagnostic_default_store,
        )

        app.state.roundtable_diagnostic_store = make_roundtable_diagnostic_store(sf)
        _register_roundtable_diagnostic_default_store(app.state.roundtable_diagnostic_store)

        # 圆桌后台作业执行器：把编排引擎作为后台任务跑。
        # 网关由环境变量 ``ROUNDTABLE_JOB_GATEWAY`` 选择（见 resolve_roundtable_gateway_mode）：
        #   - 缺省 / "inprocess"（Phase 3b）：进程内驱动真模型（leader/seat/report）——**默认**，
        #     部署时无需任何配置即可用后台挂起；
        #   - 显式 "simulated"（Phase 3a）：模拟网关，产生真实进度但不调真模型，供前端联调。
        if app.state.roundtable_job_store is not None:
            import os

            from app.gateway.roundtable_job_executor import RoundtableJobExecutor

            gateway_mode = resolve_roundtable_gateway_mode(os.getenv("ROUNDTABLE_JOB_GATEWAY"))
            if gateway_mode == "simulated":
                from deerflow.agents.roundtable_orchestrator import SimulatedRoundtableGateway

                # delay=3s/步：让模拟后台任务有可观察的进行中时长（切页/重连仍在跑）。
                gateway_factory = lambda p: SimulatedRoundtableGateway(  # noqa: E731
                    p.agents, mode=p.mode, delay=3.0
                )
                logger.info("[roundtable-jobs] gateway = simulated")
            else:
                from app.gateway.roundtable_inprocess_gateway import InProcessRoundtableGateway

                _app = app  # 闭包捕获 app，供网关访问 app.state（checkpointer/thread_store）
                gateway_factory = lambda p: InProcessRoundtableGateway(  # noqa: E731
                    _app,
                    user_id=p.user_id,
                    agents=p.agents,
                    model=p.model,
                    # 席位执行模式沿用 Step2 选择，后台席位 run 据此覆盖 policy 的 thinking/reasoning + 开子代理。
                    seat_thinking_enabled=p.seat_thinking_enabled,
                    seat_reasoning_effort=p.seat_reasoning_effort,
                    seat_subagent_enabled=p.seat_subagent_enabled,
                    seat_skill_directives=p.seat_skill_directives,
                    # 每席位推理深度覆盖（业务链条里各席位单独配）：run_seat 命中即覆盖作业级 seatMode。
                    seat_modes=p.seat_modes,
                    # 诊断日志上下文：网关每轮 leader/seat/report/summary/dashboard 报错均带这些归属。
                    job_id=p.job_id,
                    draft_id=p.draft_id,
                    task_id=p.task_id,
                )
                logger.info("[roundtable-jobs] gateway = inprocess (real models, default)")

            app.state.roundtable_job_executor = RoundtableJobExecutor(
                app.state.roundtable_job_store,
                gateway_factory=gateway_factory,
                draft_store=getattr(app.state, "roundtable_draft_store", None),
                # task 作业把最终报告写回**独立的** task-draft 存储（不分权）。
                task_draft_store=getattr(app.state, "roundtable_task_draft_store", None),
            )
        else:
            app.state.roundtable_job_executor = None

        # Run event store (has its own factory with config-driven backend selection)
        run_events_config = getattr(config, "run_events", None)
        app.state.run_event_store = make_run_event_store(run_events_config)

        # RunManager with store backing for persistence
        app.state.run_manager = RunManager(store=app.state.run_store)
        from deerflow.runtime import make_redis_concurrency_gate

        app.state.concurrency_gate = make_redis_concurrency_gate(config)

        try:
            yield
        finally:
            reconcile_task = getattr(app.state, "maf_conversation_reconcile_task", None)
            if reconcile_task is not None:
                reconcile_task.cancel()
                try:
                    await reconcile_task
                except asyncio.CancelledError:
                    pass
                except Exception:  # noqa: BLE001
                    logger.warning("[maf-roundtable] reconcile task shutdown error", exc_info=True)

            hub = getattr(app.state, "maf_conversation_event_hub", None)
            if hub is not None:
                try:
                    await hub.aclose()
                except Exception:  # noqa: BLE001
                    logger.warning("[maf-roundtable] hub aclose failed", exc_info=True)

            from deerflow.persistence.notifications import set_notification_store

            set_notification_store(None)
            queue = getattr(app.state, "knowledge_auto_ingest_queue", None)
            if queue is not None:
                from deerflow.knowledge.auto_ingest import set_auto_ingest_queue

                await queue.stop()
                set_auto_ingest_queue(None)
            await close_engine()
            gate = getattr(app.state, "concurrency_gate", None)
            if gate is not None:
                await gate.aclose()


# ---------------------------------------------------------------------------
# Getters – called by routers per-request
# ---------------------------------------------------------------------------


def _require(attr: str, label: str) -> Callable[[Request], T]:
    """Create a FastAPI dependency that returns ``app.state.<attr>`` or 503."""

    def dep(request: Request) -> T:
        val = getattr(request.app.state, attr, None)
        if val is None:
            raise HTTPException(status_code=503, detail=f"{label} not available")
        return cast(T, val)

    dep.__name__ = dep.__qualname__ = f"get_{attr}"
    return dep


get_stream_bridge: Callable[[Request], StreamBridge] = _require("stream_bridge", "Stream bridge")
get_run_manager: Callable[[Request], RunManager] = _require("run_manager", "Run manager")
get_checkpointer: Callable[[Request], Checkpointer] = _require("checkpointer", "Checkpointer")
get_run_event_store: Callable[[Request], RunEventStore] = _require("run_event_store", "Run event store")
get_feedback_repo: Callable[[Request], FeedbackRepository] = _require("feedback_repo", "Feedback")
get_run_store: Callable[[Request], RunStore] = _require("run_store", "Run store")


def get_concurrency_gate(request: Request):
    """Return the optional cross-worker concurrency gate."""
    return getattr(request.app.state, "concurrency_gate", None)


def get_store(request: Request):
    """Return the global store (may be ``None`` if not configured)."""
    return getattr(request.app.state, "store", None)


def get_thread_store(request: Request) -> ThreadMetaStore:
    """Return the thread metadata store (SQL or memory-backed)."""
    val = getattr(request.app.state, "thread_store", None)
    if val is None:
        raise HTTPException(status_code=503, detail="Thread metadata store not available")
    return val


def get_scheduled_task_store(request: Request) -> ScheduledTaskStore:
    """Return the scheduled-task store."""
    val = getattr(request.app.state, "scheduled_task_store", None)
    if val is None:
        raise HTTPException(status_code=503, detail="Scheduled task store not available")
    return val


def get_llm_metrics_store(request: Request):
    """Return the LLM metrics store, or None when not configured."""
    return getattr(request.app.state, "llm_metrics_store", None)


def get_concurrency_sample_store(request: Request):
    """Return the concurrency-sample store, or None on the memory backend."""
    return getattr(request.app.state, "concurrency_sample_store", None)


def get_tool_metrics_store(request: Request):
    """Return the tool-call metrics store, or None when not configured."""
    return getattr(request.app.state, "tool_metrics_store", None)


def get_agent_store(request: Request) -> AgentStore:
    """Return the custom-agent store."""
    val = getattr(request.app.state, "agent_store", None)
    if val is None:
        raise HTTPException(status_code=503, detail="Agent store not available")
    return val


def get_skill_store(request: Request) -> SkillStore:
    """Return the custom-skill store."""
    val = getattr(request.app.state, "skill_store", None)
    if val is None:
        raise HTTPException(status_code=503, detail="Skill store not available")
    return val


def get_notification_store(request: Request) -> NotificationStore:
    """Return the in-app notification store."""
    val = getattr(request.app.state, "notification_store", None)
    if val is None:
        raise HTTPException(status_code=503, detail="Notification store not available")
    return val


def get_tag_store(request: Request) -> TagStore:
    """Return the tag store."""
    val = getattr(request.app.state, "tag_store", None)
    if val is None:
        raise HTTPException(status_code=503, detail="Tag store not available")
    return val


def get_position_store(request: Request):
    """Return the 岗位 (position) store."""
    val = getattr(request.app.state, "position_store", None)
    if val is None:
        raise HTTPException(status_code=503, detail="Position store not available")
    return val


def get_position_sync(request: Request):
    """Return the position → subscription sync service."""
    val = getattr(request.app.state, "position_sync", None)
    if val is None:
        raise HTTPException(status_code=503, detail="Position sync service not available")
    return val


def get_thread_share_store(request: Request):
    """Return the conversation share-link store."""
    val = getattr(request.app.state, "thread_share_store", None)
    if val is None:
        raise HTTPException(status_code=503, detail="Thread share store not available")
    return val


def get_roundtable_draft_share_store(request: Request):
    """Return the roundtable-draft share-link store."""
    val = getattr(request.app.state, "roundtable_draft_share_store", None)
    if val is None:
        raise HTTPException(status_code=503, detail="Roundtable draft share store not available")
    return val


def get_embed_session_store(request: Request):
    """Return the iframe embed session → thread mapping store."""
    val = getattr(request.app.state, "embed_session_store", None)
    if val is None:
        raise HTTPException(status_code=503, detail="Embed session store not available")
    return val


def get_ai_writing_session_store(request: Request):
    """Return the AI writing session store, or None when not configured."""
    return getattr(request.app.state, "ai_writing_session_store", None)


def get_ai_writing_job_executor(request: Request):
    """Return the AI writing background-suspend executor, or None when unavailable."""
    return getattr(request.app.state, "ai_writing_job_executor", None)


def get_article_type_store(request: Request):
    """Return the article type store, or None when not configured."""
    return getattr(request.app.state, "article_type_store", None)


def get_research_template_store(request: Request):
    """Return the research template store, or None when not configured."""
    return getattr(request.app.state, "research_template_store", None)


def get_memory_config_store(request: Request):
    """Return the per-user memory config store."""
    val = getattr(request.app.state, "memory_config_store", None)
    if val is None:
        raise HTTPException(status_code=503, detail="Memory config store not available")
    return val


def get_knowledge_service(request: Request):
    """Return the knowledge base service (503 when disabled / memory backend)."""
    val = getattr(request.app.state, "knowledge_service", None)
    if val is None:
        raise HTTPException(status_code=503, detail="Knowledge base not available")
    return val


def get_run_context(request: Request) -> RunContext:
    """Build a :class:`RunContext` from ``app.state`` singletons.

    Returns a *base* context with infrastructure dependencies.
    """
    config = get_config(request)
    return RunContext(
        checkpointer=get_checkpointer(request),
        store=get_store(request),
        event_store=get_run_event_store(request),
        run_events_config=getattr(config, "run_events", None),
        thread_store=get_thread_store(request),
        app_config=config,
    )


# ---------------------------------------------------------------------------
# Auth helpers (used by authz.py and auth middleware)
# ---------------------------------------------------------------------------

# Cached singletons to avoid repeated instantiation per request
_cached_local_provider: LocalAuthProvider | None = None
_cached_repo: SQLiteUserRepository | None = None


def get_request_access_token(request: Request) -> str | None:
    """Extract access token from cookie or Authorization header.

    Priority:
    1. ``access_token`` cookie (browser session mode)
    2. ``Authorization: Bearer <token>`` header (API client mode)
    """
    cookie_token = request.cookies.get("access_token")
    if cookie_token:
        return cookie_token

    authorization = request.headers.get("authorization", "").strip()
    if not authorization:
        return None

    scheme, _, token = authorization.partition(" ")
    if scheme.lower() != "bearer":
        return None

    token = token.strip()
    return token or None


def get_local_provider() -> LocalAuthProvider:
    """Get or create the cached LocalAuthProvider singleton.

    Must be called after ``init_engine_from_config()`` — the shared
    session factory is required to construct the user repository.
    """
    global _cached_local_provider, _cached_repo
    if _cached_repo is None:
        from app.gateway.auth.repositories.sqlite import SQLiteUserRepository
        from deerflow.persistence.engine import get_session_factory

        sf = get_session_factory()
        if sf is None:
            raise RuntimeError("get_local_provider() called before init_engine_from_config(); cannot access users table")
        _cached_repo = SQLiteUserRepository(sf)
    if _cached_local_provider is None:
        from app.gateway.auth.local_provider import LocalAuthProvider

        _cached_local_provider = LocalAuthProvider(repository=_cached_repo)
    return _cached_local_provider


async def get_current_user_from_request(request: Request):
    """Get the current authenticated user from request token.

    Raises HTTPException 401 if not authenticated.
    """
    from app.gateway.auth import decode_token
    from app.gateway.auth.errors import AuthErrorCode, AuthErrorResponse, TokenError, token_error_to_code

    access_token = get_request_access_token(request)
    if not access_token:
        raise HTTPException(
            status_code=401,
            detail=AuthErrorResponse(code=AuthErrorCode.NOT_AUTHENTICATED, message="Not authenticated").model_dump(),
        )

    payload = decode_token(access_token)
    if isinstance(payload, TokenError):
        raise HTTPException(
            status_code=401,
            detail=AuthErrorResponse(code=token_error_to_code(payload), message=f"Token error: {payload.value}").model_dump(),
        )

    provider = get_local_provider()
    user = await provider.get_user(payload.sub)
    if user is None:
        raise HTTPException(
            status_code=401,
            detail=AuthErrorResponse(code=AuthErrorCode.USER_NOT_FOUND, message="User not found").model_dump(),
        )

    # Token version mismatch → password was changed, token is stale
    if user.token_version != payload.ver:
        raise HTTPException(
            status_code=401,
            detail=AuthErrorResponse(code=AuthErrorCode.TOKEN_INVALID, message="Token revoked (password changed)").model_dump(),
        )

    return user


async def get_optional_user_from_request(request: Request):
    """Get optional authenticated user from request.

    Returns None if not authenticated.
    """
    try:
        return await get_current_user_from_request(request)
    except HTTPException:
        return None


async def get_current_user(request: Request) -> str | None:
    """Extract user_id from request cookie, or None if not authenticated.

    Thin adapter that returns the string id for callers that only need
    identification (e.g., ``feedback.py``). Full-user callers should use
    ``get_current_user_from_request`` or ``get_optional_user_from_request``.
    """
    user = await get_optional_user_from_request(request)
    return str(user.id) if user else None