1203 lines
54 KiB
Python
1203 lines
54 KiB
Python
"""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, 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.strategy_agent_roles import StrategyAgentRoleStore
|
||
from deerflow.persistence.tags import TagStore
|
||
from deerflow.persistence.thread_meta.base import ThreadMetaStore
|
||
|
||
|
||
T = TypeVar("T")
|
||
|
||
|
||
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
|
||
|
||
|
||
async def _enter_runtime_backend_or_fallback[BackendT](
|
||
stack: AsyncExitStack,
|
||
context_manager,
|
||
*,
|
||
label: str,
|
||
fallback_factory: Callable[[], BackendT],
|
||
) -> BackendT:
|
||
"""Enter a persistent runtime backend without making startup depend on it.
|
||
|
||
Checkpointer/Store ``setup()`` creates its own tables. Permissions,
|
||
incompatible DDL, or an unavailable persistence service must therefore
|
||
degrade to the in-process implementation instead of aborting FastAPI's
|
||
lifespan. The exception remains fully logged for operators.
|
||
"""
|
||
|
||
try:
|
||
return await stack.enter_async_context(context_manager)
|
||
except Exception:
|
||
logger.exception(
|
||
"%s initialization failed; continuing Gateway startup with an in-memory fallback",
|
||
label,
|
||
)
|
||
return fallback_factory()
|
||
|
||
|
||
@asynccontextmanager
|
||
async def langgraph_runtime(app: FastAPI, *, background: bool = True) -> AsyncGenerator[None, None]:
|
||
"""Bootstrap and tear down all LangGraph runtime singletons.
|
||
|
||
Usage in ``app.py``::
|
||
|
||
async with langgraph_runtime(app):
|
||
yield
|
||
"""
|
||
from langgraph.checkpoint.memory import InMemorySaver
|
||
from langgraph.store.memory import InMemoryStore
|
||
|
||
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 _enter_runtime_backend_or_fallback(
|
||
stack,
|
||
make_checkpointer(config),
|
||
label="LangGraph checkpointer",
|
||
fallback_factory=InMemorySaver,
|
||
)
|
||
app.state.store = await _enter_runtime_backend_or_fallback(
|
||
stack,
|
||
make_store(config),
|
||
label="LangGraph store",
|
||
fallback_factory=InMemoryStore,
|
||
)
|
||
|
||
# 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.fixed_questions import make_fixed_question_store
|
||
|
||
app.state.fixed_question_store = make_fixed_question_store(sf)
|
||
|
||
from deerflow.persistence.thread_meta import make_thread_store
|
||
|
||
app.state.thread_store = make_thread_store(sf, app.state.store)
|
||
|
||
from deerflow.persistence.task_deeplink_threads import (
|
||
make_task_deeplink_thread_store,
|
||
)
|
||
|
||
app.state.task_deeplink_thread_store = make_task_deeplink_thread_store(
|
||
sf,
|
||
app.state.store,
|
||
app.state.thread_store,
|
||
)
|
||
|
||
from deerflow.persistence.task_deeplink_details import make_task_deeplink_detail_store
|
||
|
||
app.state.task_deeplink_detail_store = make_task_deeplink_detail_store(sf)
|
||
|
||
from deerflow.persistence.artifact_library import make_artifact_library_store
|
||
|
||
app.state.artifact_library_store = make_artifact_library_store(sf)
|
||
# User-scoped refresh tasks are registered here by the artifacts router
|
||
# and cancelled by the Gateway lifespan before persistence shuts down.
|
||
app.state.artifact_library_sync_tasks = {}
|
||
app.state.artifact_library_sync_semaphore = asyncio.Semaphore(1)
|
||
|
||
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.stream_truncation import make_stream_truncation_store
|
||
|
||
app.state.stream_truncation_store = make_stream_truncation_store(sf)
|
||
|
||
from deerflow.persistence.agents import make_agent_store
|
||
|
||
app.state.agent_store = make_agent_store(sf)
|
||
|
||
from deerflow.persistence.strategy_agent_roles import (
|
||
make_strategy_agent_role_store,
|
||
)
|
||
|
||
app.state.strategy_agent_role_store = make_strategy_agent_role_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 background and 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)
|
||
try:
|
||
await app.state.position_role_store.ensure_seed(POSITION_ROLE_SEED)
|
||
except Exception:
|
||
logger.exception("Position-role seeding failed at startup; continuing without seed refresh")
|
||
|
||
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.workflows import make_workflow_store
|
||
|
||
app.state.workflow_store = make_workflow_store(sf)
|
||
|
||
from deerflow.persistence.workflow_data_sources import make_workflow_data_source_store
|
||
from deerflow.persistence.workflow_events import make_workflow_event_store
|
||
from deerflow.persistence.workflow_planning import make_workflow_planning_store
|
||
from deerflow.persistence.workflow_runs import make_workflow_run_store
|
||
|
||
app.state.workflow_run_store = make_workflow_run_store(sf)
|
||
app.state.workflow_event_store = make_workflow_event_store(sf, run_store=app.state.workflow_run_store)
|
||
app.state.workflow_data_source_store = make_workflow_data_source_store(sf)
|
||
# Candidate plans are durable user-editable drafts, deliberately kept
|
||
# separate from both the canvas draft and a formal workflow run.
|
||
app.state.workflow_planning_store = make_workflow_planning_store(sf)
|
||
from app.report_collaboration.execution.live_hub import PublishingReportCollaborationStore, ReportCollaborationLiveHub
|
||
from deerflow.persistence.report_collaboration import make_report_collaboration_store
|
||
|
||
app.state.report_collaboration_live_hub = ReportCollaborationLiveHub()
|
||
app.state.report_collaboration_store = PublishingReportCollaborationStore(
|
||
make_report_collaboration_store(sf),
|
||
app.state.report_collaboration_live_hub,
|
||
)
|
||
app.state.report_collaboration_executor = None
|
||
app.state.report_collaboration_dispatcher = None
|
||
rc_cfg = getattr(config, "report_collaboration", None)
|
||
if background and rc_cfg is not None and rc_cfg.enabled and rc_cfg.worker_enabled:
|
||
from app.report_collaboration.agentscope_runtime.member_kernel import AgentScopeMemberKernel
|
||
from app.report_collaboration.execution.dispatcher import ReportCollaborationDispatcher
|
||
from app.report_collaboration.execution.executor import ReportCollaborationExecutor
|
||
|
||
kernel = AgentScopeMemberKernel(app.state.report_collaboration_store, app_config=config, config=rc_cfg)
|
||
executor = ReportCollaborationExecutor(app.state.report_collaboration_store, kernel, config=rc_cfg)
|
||
dispatcher = ReportCollaborationDispatcher(app.state.report_collaboration_store, executor, config=rc_cfg)
|
||
app.state.report_collaboration_executor = executor
|
||
app.state.report_collaboration_dispatcher = dispatcher
|
||
dispatcher.start()
|
||
|
||
from deerflow.persistence.menu_overrides import make_menu_override_store
|
||
|
||
app.state.menu_override_store = make_menu_override_store(sf)
|
||
|
||
from deerflow.persistence.compact_menu_overrides import make_compact_menu_override_store
|
||
|
||
app.state.compact_menu_override_store = make_compact_menu_override_store(sf)
|
||
|
||
from deerflow.persistence.sensitive_words import make_sensitive_word_store
|
||
|
||
app.state.sensitive_word_store = make_sensitive_word_store(sf)
|
||
|
||
# Immutable originals and every repaired/generated public-document
|
||
# version live outside temporary deep-research sandbox artifacts.
|
||
from deerflow.persistence.gongwen_documents import make_gongwen_document_store
|
||
|
||
app.state.gongwen_document_store = make_gongwen_document_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.model_capabilities import (
|
||
import_legacy_reasoning_effort_capabilities,
|
||
make_model_capability_store,
|
||
)
|
||
|
||
app.state.model_capability_store = make_model_capability_store(sf)
|
||
imported_model_capabilities = await import_legacy_reasoning_effort_capabilities(
|
||
app.state.model_capability_store,
|
||
config.models,
|
||
)
|
||
if imported_model_capabilities:
|
||
logger.info(
|
||
"Imported legacy reasoning-effort capability for %d model(s)",
|
||
len(imported_model_capabilities),
|
||
)
|
||
|
||
from deerflow.persistence.upload_policy import make_upload_policy_store
|
||
from deerflow.uploads.policy import default_upload_policy, set_runtime_upload_policy
|
||
|
||
app.state.upload_policy_store = make_upload_policy_store(sf)
|
||
fallback_upload_policy = default_upload_policy()
|
||
try:
|
||
upload_policy_row = await app.state.upload_policy_store.ensure_seed(fallback_upload_policy)
|
||
set_runtime_upload_policy(upload_policy_row["policy"])
|
||
except Exception:
|
||
logger.exception("Upload-policy seeding failed at startup; using process-local defaults")
|
||
set_runtime_upload_policy(fallback_upload_policy)
|
||
|
||
from deerflow.persistence.knowledge_retrieval_settings import (
|
||
make_knowledge_retrieval_settings_store,
|
||
)
|
||
|
||
app.state.knowledge_retrieval_settings_store = make_knowledge_retrieval_settings_store(sf)
|
||
|
||
from deerflow.persistence.task_analysis import make_task_analysis_store
|
||
|
||
app.state.task_analysis_store = make_task_analysis_store(sf)
|
||
|
||
from deerflow.persistence.analysis_projects import AnalysisProjectStore
|
||
|
||
app.state.analysis_project_store = AnalysisProjectStore(sf)
|
||
|
||
from deerflow.persistence.knowledge_research import (
|
||
make_knowledge_research_store,
|
||
)
|
||
|
||
app.state.knowledge_research_store = make_knowledge_research_store(sf)
|
||
|
||
from app.gateway.analysis_project_memory import (
|
||
AnalysisProjectMemoryDispatcher,
|
||
AnalysisProjectMemoryExecutor,
|
||
)
|
||
from deerflow.persistence.analysis_project_memory import AnalysisProjectMemoryStore
|
||
|
||
app.state.analysis_project_memory_store = AnalysisProjectMemoryStore(sf)
|
||
app.state.analysis_project_memory_executor = AnalysisProjectMemoryExecutor(app)
|
||
app.state.analysis_project_memory_dispatcher = AnalysisProjectMemoryDispatcher(
|
||
app.state.analysis_project_memory_store,
|
||
app.state.analysis_project_memory_executor,
|
||
)
|
||
|
||
from deerflow.persistence.report_structures import make_report_structure_store
|
||
|
||
app.state.report_structure_store = make_report_structure_store(sf)
|
||
|
||
from deerflow.integrations.weknora.runtime import register_llmwiki_store
|
||
from deerflow.persistence.llmwiki import make_llmwiki_store
|
||
|
||
app.state.llmwiki_store = make_llmwiki_store(sf)
|
||
register_llmwiki_store(app.state.llmwiki_store)
|
||
|
||
# Skill distillation and the single global assistant knowledge base.
|
||
# Both repositories live in ``deerflow.*``; only this composition layer
|
||
# wires them to Gateway executors and remote adapters.
|
||
from deerflow.persistence.assistant_knowledge import (
|
||
make_assistant_knowledge_store,
|
||
register_assistant_knowledge_store,
|
||
)
|
||
from deerflow.persistence.skill_knowledge import make_skill_knowledge_store
|
||
|
||
app.state.assistant_knowledge_store = make_assistant_knowledge_store(sf)
|
||
register_assistant_knowledge_store(app.state.assistant_knowledge_store)
|
||
app.state.skill_knowledge_store = make_skill_knowledge_store(sf)
|
||
if background and app.state.skill_knowledge_store is not None and app.state.assistant_knowledge_store is not None:
|
||
from app.gateway.skill_knowledge_job_dispatcher import SkillKnowledgeJobDispatcher
|
||
from app.gateway.skill_knowledge_job_executor import SkillKnowledgeJobExecutor
|
||
|
||
app.state.skill_knowledge_executor = SkillKnowledgeJobExecutor(app)
|
||
app.state.skill_knowledge_dispatcher = SkillKnowledgeJobDispatcher(
|
||
app.state.skill_knowledge_store,
|
||
app.state.skill_knowledge_executor,
|
||
)
|
||
app.state.skill_knowledge_dispatcher.start()
|
||
from app.gateway.skill_knowledge_auto_sync import SkillKnowledgeAutoSyncService
|
||
|
||
app.state.skill_knowledge_auto_sync = SkillKnowledgeAutoSyncService(app)
|
||
app.state.skill_knowledge_auto_sync.start()
|
||
else:
|
||
app.state.skill_knowledge_executor = None
|
||
app.state.skill_knowledge_dispatcher = None
|
||
app.state.skill_knowledge_auto_sync = None
|
||
|
||
# DeerFlow-local Wiki mirror/vector index. The store is always wired so
|
||
# migrations and status inspection work while feature flags are off;
|
||
# network/search services only exist when explicitly enabled.
|
||
from deerflow.persistence.llmwiki_index import make_llmwiki_index_store
|
||
|
||
local_index_config = config.llmwiki.local_wiki_index
|
||
app.state.llmwiki_index_store = make_llmwiki_index_store(
|
||
sf,
|
||
write_batch_size=local_index_config.database_batch_size,
|
||
)
|
||
app.state.llmwiki_embedding = None
|
||
app.state.llmwiki_vector_cache = None
|
||
app.state.llmwiki_vector_search = None
|
||
app.state.llmwiki_sync_service = None
|
||
app.state.llmwiki_retrieval_service = None
|
||
from app.gateway.llmwiki_index_runtime import rebuild_local_wiki_index_runtime
|
||
|
||
# Build every embedding-dependent object through the same hot-swap
|
||
# path used by the administrator settings API. This keeps startup,
|
||
# uploaded-local selection and config.yaml reloads behaviorally equal.
|
||
await rebuild_local_wiki_index_runtime(app, force=True)
|
||
# Start only after the optional embedding runtime is ready. Otherwise
|
||
# persisted jobs reclaimed during startup would be permanently stored
|
||
# as lexical-only even when phase-two vector recall is configured.
|
||
if background:
|
||
app.state.analysis_project_memory_dispatcher.start()
|
||
|
||
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.force_situation import make_force_situation_store
|
||
|
||
app.state.force_situation_store = make_force_situation_store(sf)
|
||
|
||
from deerflow.persistence.roundtable_artifact_submissions import make_roundtable_artifact_submission_store
|
||
|
||
# Task-scoped artifact submission status: shared by task; actor is audit-only.
|
||
app.state.roundtable_artifact_submission_store = make_roundtable_artifact_submission_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)
|
||
|
||
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)
|
||
|
||
from deerflow.persistence.roundtable_rooms import make_roundtable_room_store
|
||
|
||
app.state.roundtable_room_store = make_roundtable_room_store(sf)
|
||
# 让 LangGraph Server 路径下的圆桌编排图节点能跨 app/harness 边界拿到 repo
|
||
# （Phase 2 后台编排图写进度用）。
|
||
_register_roundtable_job_default_store(app.state.roundtable_job_store)
|
||
|
||
# 意图回合单飞 store（Phase 4）：``/api/intent/stream`` 据此对 (thread_id,
|
||
# client_turn_id) 跨 worker 单飞 + 幂等回放，杜绝并发双发意图。sf 为 None（无 DB）
|
||
# 时为 None，路由侧判空跳过单飞、走原流程。
|
||
from deerflow.persistence.intent_turns import make_intent_turn_store
|
||
|
||
app.state.intent_turn_store = make_intent_turn_store(sf)
|
||
|
||
# 圆桌会商「诊断日志」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),
|
||
)
|
||
|
||
# Phase 3 dispatcher：跨 worker 领取 queued/租约过期作业 + outbox 补偿器。
|
||
# 每个 worker 进程都跑一个；谁先 claim_job 成功谁执行（DB 条件 UPDATE 互斥），
|
||
# 崩溃的 worker 租约过期后由其它 worker 接管。路由 start/resume 后 nudge 立即唤醒。
|
||
from app.gateway.roundtable_job_dispatcher import RoundtableJobDispatcher
|
||
|
||
app.state.roundtable_job_dispatcher = RoundtableJobDispatcher(
|
||
app.state.roundtable_job_store,
|
||
app.state.roundtable_job_executor,
|
||
draft_store=getattr(app.state, "roundtable_draft_store", None),
|
||
task_draft_store=getattr(app.state, "roundtable_task_draft_store", None),
|
||
)
|
||
if background:
|
||
app.state.roundtable_job_dispatcher.start()
|
||
else:
|
||
app.state.roundtable_job_executor = None
|
||
app.state.roundtable_job_dispatcher = None
|
||
|
||
# ── Deep Research (深度研究) stores + executor + dispatcher ──────────
|
||
# Five tables: sessions / jobs / events / sources / messages.
|
||
# When DB is available, the executor + dispatcher run background research
|
||
# jobs (lease-based claim, crash-recoverable) mirroring the roundtable pattern.
|
||
from deerflow.persistence.deep_research_events import (
|
||
make_deep_research_event_store,
|
||
)
|
||
from deerflow.persistence.deep_research_jobs import make_deep_research_job_store
|
||
from deerflow.persistence.deep_research_messages import (
|
||
make_deep_research_message_store,
|
||
)
|
||
from deerflow.persistence.deep_research_sessions import (
|
||
make_deep_research_session_store,
|
||
)
|
||
from deerflow.persistence.deep_research_sessions import (
|
||
register_default_store as _register_deep_research_session_default_store,
|
||
)
|
||
from deerflow.persistence.deep_research_sources import (
|
||
make_deep_research_source_store,
|
||
)
|
||
from deerflow.persistence.document_rewrite_versions import (
|
||
make_document_rewrite_version_store,
|
||
)
|
||
|
||
app.state.deep_research_session_store = make_deep_research_session_store(sf)
|
||
# LangGraph 内联写作管线（deep_research_report）跨 harness 边界落会话终态。
|
||
_register_deep_research_session_default_store(app.state.deep_research_session_store)
|
||
app.state.deep_research_job_store = make_deep_research_job_store(sf)
|
||
app.state.deep_research_event_store = make_deep_research_event_store(sf)
|
||
app.state.deep_research_source_store = make_deep_research_source_store(sf)
|
||
app.state.deep_research_message_store = make_deep_research_message_store(sf)
|
||
# Whole-document AI rewrites retain a durable original snapshot so a
|
||
# completed replacement can be safely undone after a refresh.
|
||
app.state.document_rewrite_version_store = make_document_rewrite_version_store(sf)
|
||
# Low-latency in-process SSE fan-out. The database event log remains
|
||
# authoritative for reconnect replay and cross-worker recovery.
|
||
from app.gateway.deep_research_live_hub import DeepResearchLiveHub
|
||
|
||
app.state.deep_research_live_hub = DeepResearchLiveHub()
|
||
|
||
if app.state.deep_research_job_store is not None:
|
||
from app.gateway.deep_research_job_dispatcher import (
|
||
DeepResearchJobDispatcher,
|
||
)
|
||
from app.gateway.deep_research_job_executor import (
|
||
DeepResearchJobExecutor,
|
||
)
|
||
|
||
app.state.deep_research_executor = DeepResearchJobExecutor(
|
||
app.state.deep_research_job_store,
|
||
app.state.deep_research_session_store,
|
||
app.state.deep_research_event_store,
|
||
app.state.deep_research_source_store,
|
||
live_publisher=app.state.deep_research_live_hub.publish,
|
||
live_subscriber_probe=app.state.deep_research_live_hub.has_subscribers,
|
||
checkpointer=app.state.checkpointer,
|
||
document_rewrite_version_store=app.state.document_rewrite_version_store,
|
||
app_config=config,
|
||
)
|
||
app.state.deep_research_dispatcher = DeepResearchJobDispatcher(
|
||
app.state.deep_research_job_store,
|
||
app.state.deep_research_executor,
|
||
)
|
||
if background:
|
||
app.state.deep_research_dispatcher.start()
|
||
else:
|
||
app.state.deep_research_executor = None
|
||
app.state.deep_research_dispatcher = None
|
||
|
||
# ── Workflow Studio runtime: live hub + executor + lease dispatcher ──
|
||
# Every worker runs a dispatcher; the atomic claim decides ownership and
|
||
# an expired lease is how a crashed worker's run gets picked back up.
|
||
workflow_cfg = getattr(config, "workflows", None)
|
||
if workflow_cfg is not None and workflow_cfg.enabled:
|
||
from app.gateway.workflow_dispatcher import WorkflowRunDispatcher
|
||
from app.gateway.workflow_executor import WorkflowRunExecutor
|
||
from app.gateway.workflow_live_hub import WorkflowLiveHub
|
||
|
||
app.state.workflow_live_hub = WorkflowLiveHub()
|
||
app.state.workflow_executor = WorkflowRunExecutor(
|
||
app,
|
||
run_store=app.state.workflow_run_store,
|
||
event_store=app.state.workflow_event_store,
|
||
workflow_store=app.state.workflow_store,
|
||
data_source_store=app.state.workflow_data_source_store,
|
||
live_publisher=app.state.workflow_live_hub.publish,
|
||
config=workflow_cfg,
|
||
)
|
||
app.state.workflow_dispatcher = WorkflowRunDispatcher(
|
||
app.state.workflow_run_store,
|
||
app.state.workflow_executor,
|
||
event_store=app.state.workflow_event_store,
|
||
live_publisher=app.state.workflow_live_hub.publish,
|
||
config=workflow_cfg,
|
||
)
|
||
if background:
|
||
app.state.workflow_dispatcher.start()
|
||
from app.gateway.workflow_retention import WorkflowRetentionCleaner
|
||
|
||
app.state.workflow_retention = WorkflowRetentionCleaner(
|
||
app.state.workflow_run_store,
|
||
app.state.workflow_event_store,
|
||
config=workflow_cfg,
|
||
)
|
||
if background:
|
||
app.state.workflow_retention.start()
|
||
else:
|
||
app.state.workflow_live_hub = None
|
||
app.state.workflow_executor = None
|
||
app.state.workflow_dispatcher = None
|
||
app.state.workflow_retention = 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)
|
||
retention = config.runtime_contention
|
||
app.state.run_manager.start_retention(retention_seconds=retention.completed_run_retention_seconds, interval_seconds=retention.run_sweep_seconds)
|
||
from deerflow.runtime import make_redis_concurrency_gate
|
||
|
||
app.state.concurrency_gate = make_redis_concurrency_gate(config)
|
||
|
||
try:
|
||
yield
|
||
finally:
|
||
await app.state.run_manager.aclose()
|
||
# Project-memory extraction can use the shared local Wiki encoder.
|
||
# Stop its jobs before retiring that runtime to avoid an embedding
|
||
# call racing shutdown.
|
||
analysis_project_memory_dispatcher = getattr(app.state, "analysis_project_memory_dispatcher", None)
|
||
if analysis_project_memory_dispatcher is not None:
|
||
await analysis_project_memory_dispatcher.stop()
|
||
from app.gateway.assistant_file_ingest import stop_file_jobs
|
||
|
||
await stop_file_jobs()
|
||
from deerflow.integrations.weknora.local_index.runtime import register_local_wiki_index_runtime
|
||
|
||
register_local_wiki_index_runtime(store=None, search=None, sync=None, retrieval=None)
|
||
from app.gateway.llmwiki_index_runtime import close_local_wiki_index_runtime
|
||
|
||
await close_local_wiki_index_runtime(app)
|
||
from deerflow.persistence.notifications import set_notification_store
|
||
|
||
set_notification_store(None)
|
||
dispatcher = getattr(app.state, "roundtable_job_dispatcher", None)
|
||
if dispatcher is not None:
|
||
await dispatcher.stop()
|
||
dr_dispatcher = getattr(app.state, "deep_research_dispatcher", None)
|
||
if dr_dispatcher is not None:
|
||
await dr_dispatcher.stop()
|
||
wf_dispatcher = getattr(app.state, "workflow_dispatcher", None)
|
||
if wf_dispatcher is not None:
|
||
await wf_dispatcher.stop()
|
||
rc_dispatcher = getattr(app.state, "report_collaboration_dispatcher", None)
|
||
if rc_dispatcher is not None:
|
||
await rc_dispatcher.stop()
|
||
wf_retention = getattr(app.state, "workflow_retention", None)
|
||
if wf_retention is not None:
|
||
await wf_retention.stop()
|
||
from deerflow.workflows.runtime.sql_runner import dispose_engines
|
||
|
||
await dispose_engines()
|
||
skill_knowledge_auto_sync = getattr(app.state, "skill_knowledge_auto_sync", None)
|
||
if skill_knowledge_auto_sync is not None:
|
||
await skill_knowledge_auto_sync.close()
|
||
skill_knowledge_dispatcher = getattr(app.state, "skill_knowledge_dispatcher", None)
|
||
if skill_knowledge_dispatcher is not None:
|
||
await skill_knowledge_dispatcher.close()
|
||
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_task_deeplink_thread_store(request: Request):
|
||
"""Return the dedicated taskId + goPath conversation-history index."""
|
||
val = getattr(request.app.state, "task_deeplink_thread_store", None)
|
||
if val is None:
|
||
raise HTTPException(status_code=503, detail="Task deep-link thread store not available")
|
||
return val
|
||
|
||
|
||
def get_task_deeplink_detail_store(request: Request):
|
||
"""Return the external task-detail store used by task deep links."""
|
||
val = getattr(request.app.state, "task_deeplink_detail_store", None)
|
||
if val is None:
|
||
raise HTTPException(status_code=503, detail="Task deep-link detail 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_stream_truncation_store(request: Request):
|
||
"""Return the stream-truncation event store, or None when not configured."""
|
||
return getattr(request.app.state, "stream_truncation_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_strategy_agent_role_store(request: Request) -> StrategyAgentRoleStore:
|
||
"""Return the global strategy-agent role registry."""
|
||
val = getattr(request.app.state, "strategy_agent_role_store", None)
|
||
if val is None:
|
||
raise HTTPException(status_code=503, detail="Strategy agent role 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 the caller identity, preferring the explicit Bearer header.
|
||
|
||
An iframe or another tab can rewrite the origin-wide cookie, while the
|
||
frontend deliberately sources its Bearer token from this tab's auth store.
|
||
"""
|
||
from app.gateway.auth.token_source import get_explicit_or_session_token
|
||
|
||
request_token = get_explicit_or_session_token(request)
|
||
if request_token:
|
||
return request_token
|
||
|
||
from app.gateway.proxy_path import app_relative_path
|
||
|
||
path = app_relative_path(request).rstrip("/")
|
||
if path.startswith("/api/llmwiki/knowledge-bases/") and path.endswith("/weknora/frame"):
|
||
query_token = (request.query_params.get("access_token") or "").strip()
|
||
if query_token:
|
||
return query_token
|
||
return 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
|
||
from app.gateway.auth.session_policy import session_policy_error
|
||
|
||
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(),
|
||
)
|
||
|
||
# Scoped service tokens are authorized by AuthMiddleware's strict path
|
||
# allow-list. They are intentionally not accepted by the separate
|
||
# LangGraph auth path, which still applies session_policy_error directly.
|
||
if payload.auth_method != "service" and (policy_error := session_policy_error(payload)):
|
||
raise HTTPException(
|
||
status_code=401,
|
||
detail=AuthErrorResponse(code=AuthErrorCode.TOKEN_INVALID, message=policy_error).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="登录已失效：密码已更改或账号已在其他位置登录，请重新登录").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
|