refactor(config): eliminate global mutable state, wire DeerFlowContext into runtime

- Freeze all config models (AppConfig + 15 sub-configs) with frozen=True
- Purify from_file() — remove 9 load_*_from_dict() side-effect calls
- Replace mtime/reload/push/pop machinery with single ContextVar + init_app_config()
- Delete 10 sub-module globals and their getters/setters/loaders
- Migrate 50+ consumers from get_*_config() to get_app_config().xxx

- Expand DeerFlowContext: app_config + thread_id + agent_name (frozen dataclass)
- Wire into Gateway runtime (worker.py) and DeerFlowClient via context= parameter
- Remove sandbox_id from runtime.context — flows through ThreadState.sandbox only
- Middleware/tools access runtime.context directly via Runtime[DeerFlowContext] generic
- resolve_context() retained at server entry points for LangGraph Server fallback

.env.example

@@ -24,6 +24,7 @@ INFOQUEST_API_KEY=your-infoquest-api-key
 # SLACK_BOT_TOKEN=your-slack-bot-token
 # SLACK_APP_TOKEN=your-slack-app-token
 # TELEGRAM_BOT_TOKEN=your-telegram-bot-token
+# DISCORD_BOT_TOKEN=your-discord-bot-token
 
 # Enable LangSmith to monitor and debug your LLM calls, agent runs, and tool executions.
 # LANGSMITH_TRACING=true
@@ -33,9 +34,5 @@ INFOQUEST_API_KEY=your-infoquest-api-key
 
 # GitHub API Token
 # GITHUB_TOKEN=your-github-token
-
-# Database (only needed when config.yaml has database.backend: postgres)
-# DATABASE_URL=postgresql://deerflow:password@localhost:5432/deerflow
-#
 # WECOM_BOT_ID=your-wecom-bot-id
 # WECOM_BOT_SECRET=your-wecom-bot-secret

backend/CLAUDE.md

@@ -179,7 +179,9 @@ Setup: Copy `config.example.yaml` to `config.yaml` in the **project root** direc
 
 **Config Versioning**: `config.example.yaml` has a `config_version` field. On startup, `AppConfig.from_file()` compares user version vs example version and emits a warning if outdated. Missing `config_version` = version 0. Run `make config-upgrade` to auto-merge missing fields. When changing the config schema, bump `config_version` in `config.example.yaml`.
 
-**Config Caching**: `get_app_config()` caches the parsed config, but automatically reloads it when the resolved config path changes or the file's mtime increases. This keeps Gateway and LangGraph reads aligned with `config.yaml` edits without requiring a manual process restart.
+**Config Lifecycle**: All config models are `frozen=True` (immutable after construction). `AppConfig.from_file()` is a pure function — no side effects on sub-module globals. `get_app_config()` is backed by a single `ContextVar`, set once via `init_app_config()` at process startup. To update config at runtime (e.g., Gateway API updates MCP/Skills), construct a new `AppConfig.from_file()` and call `init_app_config()` again. No mtime detection, no auto-reload.
+
+**DeerFlowContext**: Per-invocation typed context for the agent execution path, injected via LangGraph `Runtime[DeerFlowContext]`. Holds `app_config: AppConfig`, `thread_id: str`, `agent_name: str | None`. Gateway runtime and `DeerFlowClient` construct full `DeerFlowContext` at invoke time; LangGraph Server path uses a fallback via `resolve_context()`. Middleware and tools access context through `resolve_context(runtime)` which returns a typed `DeerFlowContext` regardless of entry point. Mutable runtime state (`sandbox_id`) flows through `ThreadState.sandbox`, not context.
 
 Configuration priority:
 1. Explicit `config_path` argument

backend/Dockerfile

@@ -13,9 +13,6 @@ FROM python:3.12-slim-bookworm AS builder
 ARG NODE_MAJOR=22
 ARG APT_MIRROR
 ARG UV_INDEX_URL
-# Optional extras to install (e.g. "postgres" for PostgreSQL support)
-# Usage: docker build --build-arg UV_EXTRAS=postgres ...
-ARG UV_EXTRAS
 
 # Optionally override apt mirror for restricted networks (e.g. APT_MIRROR=mirrors.aliyun.com)
 RUN if [ -n "${APT_MIRROR}" ]; then \
@@ -46,9 +43,8 @@ WORKDIR /app
 COPY backend ./backend
 
 # Install dependencies with cache mount
-# When UV_EXTRAS is set (e.g. "postgres"), installs optional dependencies.
 RUN --mount=type=cache,target=/root/.cache/uv \
-    sh -c "cd backend && UV_INDEX_URL=${UV_INDEX_URL:-https://pypi.org/simple} uv sync ${UV_EXTRAS:+--extra $UV_EXTRAS}"
+    sh -c "cd backend && UV_INDEX_URL=${UV_INDEX_URL:-https://pypi.org/simple} uv sync"
 
 # ── Stage 2: Dev ──────────────────────────────────────────────────────────────
 # Retains compiler toolchain from builder so startup-time `uv sync` can build

backend/app/channels/discord.py

@@ -0,0 +1,273 @@
+"""Discord channel integration using discord.py."""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+import threading
+from typing import Any
+
+from app.channels.base import Channel
+from app.channels.message_bus import InboundMessageType, MessageBus, OutboundMessage, ResolvedAttachment
+
+logger = logging.getLogger(__name__)
+
+_DISCORD_MAX_MESSAGE_LEN = 2000
+
+
+class DiscordChannel(Channel):
+    """Discord bot channel.
+
+    Configuration keys (in ``config.yaml`` under ``channels.discord``):
+        - ``bot_token``: Discord Bot token.
+        - ``allowed_guilds``: (optional) List of allowed Discord guild IDs. Empty = allow all.
+    """
+
+    def __init__(self, bus: MessageBus, config: dict[str, Any]) -> None:
+        super().__init__(name="discord", bus=bus, config=config)
+        self._bot_token = str(config.get("bot_token", "")).strip()
+        self._allowed_guilds: set[int] = set()
+        for guild_id in config.get("allowed_guilds", []):
+            try:
+                self._allowed_guilds.add(int(guild_id))
+            except (TypeError, ValueError):
+                continue
+
+        self._client = None
+        self._thread: threading.Thread | None = None
+        self._discord_loop: asyncio.AbstractEventLoop | None = None
+        self._main_loop: asyncio.AbstractEventLoop | None = None
+        self._discord_module = None
+
+    async def start(self) -> None:
+        if self._running:
+            return
+
+        try:
+            import discord
+        except ImportError:
+            logger.error("discord.py is not installed. Install it with: uv add discord.py")
+            return
+
+        if not self._bot_token:
+            logger.error("Discord channel requires bot_token")
+            return
+
+        intents = discord.Intents.default()
+        intents.messages = True
+        intents.guilds = True
+        intents.message_content = True
+
+        client = discord.Client(
+            intents=intents,
+            allowed_mentions=discord.AllowedMentions.none(),
+        )
+        self._client = client
+        self._discord_module = discord
+        self._main_loop = asyncio.get_event_loop()
+
+        @client.event
+        async def on_message(message) -> None:
+            await self._on_message(message)
+
+        self._running = True
+        self.bus.subscribe_outbound(self._on_outbound)
+
+        self._thread = threading.Thread(target=self._run_client, daemon=True)
+        self._thread.start()
+        logger.info("Discord channel started")
+
+    async def stop(self) -> None:
+        self._running = False
+        self.bus.unsubscribe_outbound(self._on_outbound)
+
+        if self._client and self._discord_loop and self._discord_loop.is_running():
+            close_future = asyncio.run_coroutine_threadsafe(self._client.close(), self._discord_loop)
+            try:
+                await asyncio.wait_for(asyncio.wrap_future(close_future), timeout=10)
+            except TimeoutError:
+                logger.warning("[Discord] client close timed out after 10s")
+            except Exception:
+                logger.exception("[Discord] error while closing client")
+
+        if self._thread:
+            self._thread.join(timeout=10)
+            self._thread = None
+
+        self._client = None
+        self._discord_loop = None
+        self._discord_module = None
+        logger.info("Discord channel stopped")
+
+    async def send(self, msg: OutboundMessage) -> None:
+        target = await self._resolve_target(msg)
+        if target is None:
+            logger.error("[Discord] target not found for chat_id=%s thread_ts=%s", msg.chat_id, msg.thread_ts)
+            return
+
+        text = msg.text or ""
+        for chunk in self._split_text(text):
+            send_future = asyncio.run_coroutine_threadsafe(target.send(chunk), self._discord_loop)
+            await asyncio.wrap_future(send_future)
+
+    async def send_file(self, msg: OutboundMessage, attachment: ResolvedAttachment) -> bool:
+        target = await self._resolve_target(msg)
+        if target is None:
+            logger.error("[Discord] target not found for file upload chat_id=%s thread_ts=%s", msg.chat_id, msg.thread_ts)
+            return False
+
+        if self._discord_module is None:
+            return False
+
+        try:
+            fp = open(str(attachment.actual_path), "rb")  # noqa: SIM115
+            file = self._discord_module.File(fp, filename=attachment.filename)
+            send_future = asyncio.run_coroutine_threadsafe(target.send(file=file), self._discord_loop)
+            await asyncio.wrap_future(send_future)
+            logger.info("[Discord] file uploaded: %s", attachment.filename)
+            return True
+        except Exception:
+            logger.exception("[Discord] failed to upload file: %s", attachment.filename)
+            return False
+
+    async def _on_message(self, message) -> None:
+        if not self._running or not self._client:
+            return
+
+        if message.author.bot:
+            return
+
+        if self._client.user and message.author.id == self._client.user.id:
+            return
+
+        guild = message.guild
+        if self._allowed_guilds:
+            if guild is None or guild.id not in self._allowed_guilds:
+                return
+
+        text = (message.content or "").strip()
+        if not text:
+            return
+
+        if self._discord_module is None:
+            return
+
+        if isinstance(message.channel, self._discord_module.Thread):
+            chat_id = str(message.channel.parent_id or message.channel.id)
+            thread_id = str(message.channel.id)
+        else:
+            thread = await self._create_thread(message)
+            if thread is None:
+                return
+            chat_id = str(message.channel.id)
+            thread_id = str(thread.id)
+
+        msg_type = InboundMessageType.COMMAND if text.startswith("/") else InboundMessageType.CHAT
+        inbound = self._make_inbound(
+            chat_id=chat_id,
+            user_id=str(message.author.id),
+            text=text,
+            msg_type=msg_type,
+            thread_ts=thread_id,
+            metadata={
+                "guild_id": str(guild.id) if guild else None,
+                "channel_id": str(message.channel.id),
+                "message_id": str(message.id),
+            },
+        )
+        inbound.topic_id = thread_id
+
+        if self._main_loop and self._main_loop.is_running():
+            future = asyncio.run_coroutine_threadsafe(self.bus.publish_inbound(inbound), self._main_loop)
+            future.add_done_callback(lambda f: logger.exception("[Discord] publish_inbound failed", exc_info=f.exception()) if f.exception() else None)
+
+    def _run_client(self) -> None:
+        self._discord_loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(self._discord_loop)
+        try:
+            self._discord_loop.run_until_complete(self._client.start(self._bot_token))
+        except Exception:
+            if self._running:
+                logger.exception("Discord client error")
+        finally:
+            try:
+                if self._client and not self._client.is_closed():
+                    self._discord_loop.run_until_complete(self._client.close())
+            except Exception:
+                logger.exception("Error during Discord shutdown")
+
+    async def _create_thread(self, message):
+        try:
+            thread_name = f"deerflow-{message.author.display_name}-{message.id}"[:100]
+            return await message.create_thread(name=thread_name)
+        except Exception:
+            logger.exception("[Discord] failed to create thread for message=%s (threads may be disabled or missing permissions)", message.id)
+            try:
+                await message.channel.send("Could not create a thread for your message. Please check that threads are enabled in this channel.")
+            except Exception:
+                pass
+            return None
+
+    async def _resolve_target(self, msg: OutboundMessage):
+        if not self._client or not self._discord_loop:
+            return None
+
+        target_ids: list[str] = []
+        if msg.thread_ts:
+            target_ids.append(msg.thread_ts)
+        if msg.chat_id and msg.chat_id not in target_ids:
+            target_ids.append(msg.chat_id)
+
+        for raw_id in target_ids:
+            target = await self._get_channel_or_thread(raw_id)
+            if target is not None:
+                return target
+        return None
+
+    async def _get_channel_or_thread(self, raw_id: str):
+        if not self._client or not self._discord_loop:
+            return None
+
+        try:
+            target_id = int(raw_id)
+        except (TypeError, ValueError):
+            return None
+
+        get_future = asyncio.run_coroutine_threadsafe(self._fetch_channel(target_id), self._discord_loop)
+        try:
+            return await asyncio.wrap_future(get_future)
+        except Exception:
+            logger.exception("[Discord] failed to resolve target id=%s", raw_id)
+            return None
+
+    async def _fetch_channel(self, target_id: int):
+        if not self._client:
+            return None
+
+        channel = self._client.get_channel(target_id)
+        if channel is not None:
+            return channel
+
+        try:
+            return await self._client.fetch_channel(target_id)
+        except Exception:
+            return None
+
+    @staticmethod
+    def _split_text(text: str) -> list[str]:
+        if not text:
+            return [""]
+
+        chunks: list[str] = []
+        remaining = text
+        while len(remaining) > _DISCORD_MAX_MESSAGE_LEN:
+            split_at = remaining.rfind("\n", 0, _DISCORD_MAX_MESSAGE_LEN)
+            if split_at <= 0:
+                split_at = _DISCORD_MAX_MESSAGE_LEN
+            chunks.append(remaining[:split_at])
+            remaining = remaining[split_at:].lstrip("\n")
+
+        if remaining:
+            chunks.append(remaining)
+
+        return chunks

backend/app/channels/manager.py

@@ -35,6 +35,7 @@ STREAM_UPDATE_MIN_INTERVAL_SECONDS = 0.35
 THREAD_BUSY_MESSAGE = "This conversation is already processing another request. Please wait for it to finish and try again."
 
 CHANNEL_CAPABILITIES = {
+    "discord": {"supports_streaming": False},
     "feishu": {"supports_streaming": True},
     "slack": {"supports_streaming": False},
     "telegram": {"supports_streaming": False},

backend/app/channels/service.py

@@ -15,6 +15,7 @@ logger = logging.getLogger(__name__)
 
 # Channel name → import path for lazy loading
 _CHANNEL_REGISTRY: dict[str, str] = {
+    "discord": "app.channels.discord:DiscordChannel",
     "feishu": "app.channels.feishu:FeishuChannel",
     "slack": "app.channels.slack:SlackChannel",
     "telegram": "app.channels.telegram:TelegramChannel",
@@ -66,9 +67,9 @@ class ChannelService:
     @classmethod
     def from_app_config(cls) -> ChannelService:
         """Create a ChannelService from the application config."""
-        from deerflow.config.app_config import get_app_config
+        from deerflow.config.app_config import AppConfig
 
-        config = get_app_config()
+        config = AppConfig.current()
         channels_config = {}
         # extra fields are allowed by AppConfig (extra="allow")
         extra = config.model_extra or {}

backend/app/gateway/app.py

@@ -1,23 +1,16 @@
 import logging
-import os
 from collections.abc import AsyncGenerator
 from contextlib import asynccontextmanager
-from datetime import UTC
 
 from fastapi import FastAPI
-from fastapi.middleware.cors import CORSMiddleware
 
-from app.gateway.auth_middleware import AuthMiddleware
 from app.gateway.config import get_gateway_config
-from app.gateway.csrf_middleware import CSRFMiddleware
 from app.gateway.deps import langgraph_runtime
 from app.gateway.routers import (
     agents,
     artifacts,
     assistants_compat,
-    auth,
     channels,
-    feedback,
     mcp,
     memory,
     models,
@@ -28,7 +21,7 @@ from app.gateway.routers import (
     threads,
     uploads,
 )
-from deerflow.config.app_config import get_app_config
+from deerflow.config.app_config import AppConfig
 
 # Configure logging
 logging.basicConfig(
@@ -40,132 +33,13 @@ logging.basicConfig(
 logger = logging.getLogger(__name__)
 
 
-async def _ensure_admin_user(app: FastAPI) -> None:
-    """Auto-create the admin user on first boot if no users exist.
-
-    After admin creation, migrate orphan threads from the LangGraph
-    store (metadata.owner_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.
-
-    No SQL persistence migration is needed: the four owner_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. "Existing persistence DB + new auth"
-    is not a supported upgrade path — fresh install or wipe-and-retry.
-
-    Multi-worker safe: relies on SQLite UNIQUE constraint to resolve
-    races during admin creation. Only the worker that successfully
-    creates/updates the admin prints the password; losers silently skip.
-    """
-    import secrets
-
-    from app.gateway.auth.credential_file import write_initial_credentials
-    from app.gateway.deps import get_local_provider
-
-    def _announce_credentials(email: str, password: str, *, label: str, headline: str) -> None:
-        """Write the password to a 0600 file and log the path (never the secret)."""
-        cred_path = write_initial_credentials(email, password, label=label)
-        logger.info("=" * 60)
-        logger.info("  %s", headline)
-        logger.info("  Credentials written to: %s (mode 0600)", cred_path)
-        logger.info("  Change it after login: Settings -> Account")
-        logger.info("=" * 60)
-
-    provider = get_local_provider()
-    user_count = await provider.count_users()
-
-    admin = None
-
-    if user_count == 0:
-        password = secrets.token_urlsafe(16)
-        try:
-            admin = await provider.create_user(email="admin@deerflow.dev", password=password, system_role="admin", needs_setup=True)
-        except ValueError:
-            return  # Another worker already created the admin.
-        _announce_credentials(admin.email, password, label="initial", headline="Admin account created on first boot")
-    else:
-        # Admin exists but setup never completed — reset password so operator
-        # can always find it in the console without needing the CLI.
-        # Multi-worker guard: if admin was created less than 30s ago, another
-        # worker just created it and will print the password — skip reset.
-        admin = await provider.get_user_by_email("admin@deerflow.dev")
-        if admin and admin.needs_setup:
-            import time
-
-            age = time.time() - admin.created_at.replace(tzinfo=UTC).timestamp()
-            if age >= 30:
-                from app.gateway.auth.password import hash_password_async
-
-                password = secrets.token_urlsafe(16)
-                admin.password_hash = await hash_password_async(password)
-                admin.token_version += 1
-                await provider.update_user(admin)
-                _announce_credentials(admin.email, password, label="reset", headline="Admin account setup incomplete — password reset")
-
-    if admin is None:
-        return  # Nothing to bind orphans to.
-
-    admin_id = str(admin.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 owner_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 _migrate_orphaned_threads(store, admin_user_id: str) -> int:
-    """Migrate LangGraph store threads with no owner_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("owner_id"):
-            metadata["owner_id"] = admin_user_id
-            item.value["metadata"] = metadata
-            await store.aput(("threads",), item.key, item.value)
-            migrated += 1
-    return migrated
-
-
 @asynccontextmanager
 async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
     """Application lifespan handler."""
 
     # Load config and check necessary environment variables at startup
     try:
-        get_app_config()
+        AppConfig.current()
         logger.info("Configuration loaded successfully")
     except Exception as e:
         error_msg = f"Failed to load configuration during gateway startup: {e}"
@@ -178,10 +52,6 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
     async with langgraph_runtime(app):
         logger.info("LangGraph runtime initialised")
 
-        # 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)
-
         # Start IM channel service if any channels are configured
         try:
             from app.channels.service import start_channel_service
@@ -293,31 +163,7 @@ This gateway provides custom endpoints for models, MCP configuration, skills, an
         ],
     )
 
-    # Auth: reject unauthenticated requests to non-public paths (fail-closed safety net)
+    # CORS is handled by nginx - no need for FastAPI middleware
-    app.add_middleware(AuthMiddleware)
-
-    # CSRF: Double Submit Cookie pattern for state-changing requests
-    app.add_middleware(CSRFMiddleware)
-
-    # CORS: when GATEWAY_CORS_ORIGINS is set (dev without nginx), add CORS middleware.
-    # In production, nginx handles CORS and no middleware is needed.
-    cors_origins_env = os.environ.get("GATEWAY_CORS_ORIGINS", "")
-    if cors_origins_env:
-        cors_origins = [o.strip() for o in cors_origins_env.split(",") if o.strip()]
-        # Validate: wildcard origin with credentials is a security misconfiguration
-        for origin in cors_origins:
-            if origin == "*":
-                logger.error("GATEWAY_CORS_ORIGINS contains wildcard '*' with allow_credentials=True. This is a security misconfiguration — browsers will reject the response. Use explicit scheme://host:port origins instead.")
-                cors_origins = [o for o in cors_origins if o != "*"]
-                break
-        if cors_origins:
-            app.add_middleware(
-                CORSMiddleware,
-                allow_origins=cors_origins,
-                allow_credentials=True,
-                allow_methods=["*"],
-                allow_headers=["*"],
-            )
 
     # Include routers
     # Models API is mounted at /api/models
@@ -353,12 +199,6 @@ This gateway provides custom endpoints for models, MCP configuration, skills, an
     # Assistants compatibility API (LangGraph Platform stub)
     app.include_router(assistants_compat.router)
 
-    # Auth API is mounted at /api/v1/auth
-    app.include_router(auth.router)
-
-    # Feedback API is mounted at /api/threads/{thread_id}/runs/{run_id}/feedback
-    app.include_router(feedback.router)
-
     # Thread Runs API (LangGraph Platform-compatible runs lifecycle)
     app.include_router(thread_runs.router)
 

backend/app/gateway/auth/__init__.py

@@ -1,42 +0,0 @@
-"""Authentication module for DeerFlow.
-
-This module provides:
-- JWT-based authentication
-- Provider Factory pattern for extensible auth methods
-- UserRepository interface for storage backends (SQLite)
-"""
-
-from app.gateway.auth.config import AuthConfig, get_auth_config, set_auth_config
-from app.gateway.auth.errors import AuthErrorCode, AuthErrorResponse, TokenError
-from app.gateway.auth.jwt import TokenPayload, create_access_token, decode_token
-from app.gateway.auth.local_provider import LocalAuthProvider
-from app.gateway.auth.models import User, UserResponse
-from app.gateway.auth.password import hash_password, verify_password
-from app.gateway.auth.providers import AuthProvider
-from app.gateway.auth.repositories.base import UserRepository
-
-__all__ = [
-    # Config
-    "AuthConfig",
-    "get_auth_config",
-    "set_auth_config",
-    # Errors
-    "AuthErrorCode",
-    "AuthErrorResponse",
-    "TokenError",
-    # JWT
-    "TokenPayload",
-    "create_access_token",
-    "decode_token",
-    # Password
-    "hash_password",
-    "verify_password",
-    # Models
-    "User",
-    "UserResponse",
-    # Providers
-    "AuthProvider",
-    "LocalAuthProvider",
-    # Repository
-    "UserRepository",
-]

backend/app/gateway/auth/config.py

@@ -1,57 +0,0 @@
-"""Authentication configuration for DeerFlow."""
-
-import logging
-import os
-import secrets
-
-from dotenv import load_dotenv
-from pydantic import BaseModel, Field
-
-load_dotenv()
-
-logger = logging.getLogger(__name__)
-
-
-class AuthConfig(BaseModel):
-    """JWT and auth-related configuration. Parsed once at startup.
-
-    Note: the ``users`` table now lives in the shared persistence
-    database managed by ``deerflow.persistence.engine``. The old
-    ``users_db_path`` config key has been removed — user storage is
-    configured through ``config.database`` like every other table.
-    """
-
-    jwt_secret: str = Field(
-        ...,
-        description="Secret key for JWT signing. MUST be set via AUTH_JWT_SECRET.",
-    )
-    token_expiry_days: int = Field(default=7, ge=1, le=30)
-    oauth_github_client_id: str | None = Field(default=None)
-    oauth_github_client_secret: str | None = Field(default=None)
-
-
-_auth_config: AuthConfig | None = None
-
-
-def get_auth_config() -> AuthConfig:
-    """Get the global AuthConfig instance. Parses from env on first call."""
-    global _auth_config
-    if _auth_config is None:
-        jwt_secret = os.environ.get("AUTH_JWT_SECRET")
-        if not jwt_secret:
-            jwt_secret = secrets.token_urlsafe(32)
-            os.environ["AUTH_JWT_SECRET"] = jwt_secret
-            logger.warning(
-                "⚠ AUTH_JWT_SECRET is not set — using an auto-generated ephemeral secret. "
-                "Sessions will be invalidated on restart. "
-                "For production, add AUTH_JWT_SECRET to your .env file: "
-                'python -c "import secrets; print(secrets.token_urlsafe(32))"'
-            )
-        _auth_config = AuthConfig(jwt_secret=jwt_secret)
-    return _auth_config
-
-
-def set_auth_config(config: AuthConfig) -> None:
-    """Set the global AuthConfig instance (for testing)."""
-    global _auth_config
-    _auth_config = config

backend/app/gateway/auth/credential_file.py

@@ -1,48 +0,0 @@
-"""Write initial admin credentials to a restricted file instead of logs.
-
-Logging secrets to stdout/stderr is a well-known CodeQL finding
-(py/clear-text-logging-sensitive-data) — in production those logs
-get collected into ELK/Splunk/etc and become a secret sprawl
-source. This helper writes the credential to a 0600 file that only
-the process user can read, and returns the path so the caller can
-log **the path** (not the password) for the operator to pick up.
-"""
-
-from __future__ import annotations
-
-import os
-from pathlib import Path
-
-from deerflow.config.paths import get_paths
-
-_CREDENTIAL_FILENAME = "admin_initial_credentials.txt"
-
-
-def write_initial_credentials(email: str, password: str, *, label: str = "initial") -> Path:
-    """Write the admin email + password to ``{base_dir}/admin_initial_credentials.txt``.
-
-    The file is created **atomically** with mode 0600 via ``os.open``
-    so the password is never world-readable, even for the single syscall
-    window between ``write_text`` and ``chmod``.
-
-    ``label`` distinguishes "initial" (fresh creation) from "reset"
-    (password reset) in the file header so an operator picking up the
-    file after a restart can tell which event produced it.
-
-    Returns the absolute :class:`Path` to the file.
-    """
-    target = get_paths().base_dir / _CREDENTIAL_FILENAME
-    target.parent.mkdir(parents=True, exist_ok=True)
-
-    content = (
-        f"# DeerFlow admin {label} credentials\n# This file is generated on first boot or password reset.\n# Change the password after login via Settings -> Account,\n# then delete this file.\n#\nemail: {email}\npassword: {password}\n"
-    )
-
-    # Atomic 0600 create-or-truncate. O_TRUNC (not O_EXCL) so the
-    # reset-password path can rewrite an existing file without a
-    # separate unlink-then-create dance.
-    fd = os.open(target, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
-    with os.fdopen(fd, "w", encoding="utf-8") as fh:
-        fh.write(content)
-
-    return target.resolve()

backend/app/gateway/auth/errors.py

@@ -1,44 +0,0 @@
-"""Typed error definitions for auth module.
-
-AuthErrorCode: exhaustive enum of all auth failure conditions.
-TokenError: exhaustive enum of JWT decode failures.
-AuthErrorResponse: structured error payload for HTTP responses.
-"""
-
-from enum import StrEnum
-
-from pydantic import BaseModel
-
-
-class AuthErrorCode(StrEnum):
-    """Exhaustive list of auth error conditions."""
-
-    INVALID_CREDENTIALS = "invalid_credentials"
-    TOKEN_EXPIRED = "token_expired"
-    TOKEN_INVALID = "token_invalid"
-    USER_NOT_FOUND = "user_not_found"
-    EMAIL_ALREADY_EXISTS = "email_already_exists"
-    PROVIDER_NOT_FOUND = "provider_not_found"
-    NOT_AUTHENTICATED = "not_authenticated"
-
-
-class TokenError(StrEnum):
-    """Exhaustive list of JWT decode failure reasons."""
-
-    EXPIRED = "expired"
-    INVALID_SIGNATURE = "invalid_signature"
-    MALFORMED = "malformed"
-
-
-class AuthErrorResponse(BaseModel):
-    """Structured error response — replaces bare `detail` strings."""
-
-    code: AuthErrorCode
-    message: str
-
-
-def token_error_to_code(err: TokenError) -> AuthErrorCode:
-    """Map TokenError to AuthErrorCode — single source of truth."""
-    if err == TokenError.EXPIRED:
-        return AuthErrorCode.TOKEN_EXPIRED
-    return AuthErrorCode.TOKEN_INVALID

backend/app/gateway/auth/jwt.py

@@ -1,55 +0,0 @@
-"""JWT token creation and verification."""
-
-from datetime import UTC, datetime, timedelta
-
-import jwt
-from pydantic import BaseModel
-
-from app.gateway.auth.config import get_auth_config
-from app.gateway.auth.errors import TokenError
-
-
-class TokenPayload(BaseModel):
-    """JWT token payload."""
-
-    sub: str  # user_id
-    exp: datetime
-    iat: datetime | None = None
-    ver: int = 0  # token_version — must match User.token_version
-
-
-def create_access_token(user_id: str, expires_delta: timedelta | None = None, token_version: int = 0) -> str:
-    """Create a JWT access token.
-
-    Args:
-        user_id: The user's UUID as string
-        expires_delta: Optional custom expiry, defaults to 7 days
-        token_version: User's current token_version for invalidation
-
-    Returns:
-        Encoded JWT string
-    """
-    config = get_auth_config()
-    expiry = expires_delta or timedelta(days=config.token_expiry_days)
-
-    now = datetime.now(UTC)
-    payload = {"sub": user_id, "exp": now + expiry, "iat": now, "ver": token_version}
-    return jwt.encode(payload, config.jwt_secret, algorithm="HS256")
-
-
-def decode_token(token: str) -> TokenPayload | TokenError:
-    """Decode and validate a JWT token.
-
-    Returns:
-        TokenPayload if valid, or a specific TokenError variant.
-    """
-    config = get_auth_config()
-    try:
-        payload = jwt.decode(token, config.jwt_secret, algorithms=["HS256"])
-        return TokenPayload(**payload)
-    except jwt.ExpiredSignatureError:
-        return TokenError.EXPIRED
-    except jwt.InvalidSignatureError:
-        return TokenError.INVALID_SIGNATURE
-    except jwt.PyJWTError:
-        return TokenError.MALFORMED

backend/app/gateway/auth/local_provider.py

@@ -1,87 +0,0 @@
-"""Local email/password authentication provider."""
-
-from app.gateway.auth.models import User
-from app.gateway.auth.password import hash_password_async, verify_password_async
-from app.gateway.auth.providers import AuthProvider
-from app.gateway.auth.repositories.base import UserRepository
-
-
-class LocalAuthProvider(AuthProvider):
-    """Email/password authentication provider using local database."""
-
-    def __init__(self, repository: UserRepository):
-        """Initialize with a UserRepository.
-
-        Args:
-            repository: UserRepository implementation (SQLite)
-        """
-        self._repo = repository
-
-    async def authenticate(self, credentials: dict) -> User | None:
-        """Authenticate with email and password.
-
-        Args:
-            credentials: dict with 'email' and 'password' keys
-
-        Returns:
-            User if authentication succeeds, None otherwise
-        """
-        email = credentials.get("email")
-        password = credentials.get("password")
-
-        if not email or not password:
-            return None
-
-        user = await self._repo.get_user_by_email(email)
-        if user is None:
-            return None
-
-        if user.password_hash is None:
-            # OAuth user without local password
-            return None
-
-        if not await verify_password_async(password, user.password_hash):
-            return None
-
-        return user
-
-    async def get_user(self, user_id: str) -> User | None:
-        """Get user by ID."""
-        return await self._repo.get_user_by_id(user_id)
-
-    async def create_user(self, email: str, password: str | None = None, system_role: str = "user", needs_setup: bool = False) -> User:
-        """Create a new local user.
-
-        Args:
-            email: User email address
-            password: Plain text password (will be hashed)
-            system_role: Role to assign ("admin" or "user")
-            needs_setup: If True, user must complete setup on first login
-
-        Returns:
-            Created User instance
-        """
-        password_hash = await hash_password_async(password) if password else None
-        user = User(
-            email=email,
-            password_hash=password_hash,
-            system_role=system_role,
-            needs_setup=needs_setup,
-        )
-        return await self._repo.create_user(user)
-
-    async def get_user_by_oauth(self, provider: str, oauth_id: str) -> User | None:
-        """Get user by OAuth provider and ID."""
-        return await self._repo.get_user_by_oauth(provider, oauth_id)
-
-    async def count_users(self) -> int:
-        """Return total number of registered users."""
-        return await self._repo.count_users()
-
-    async def update_user(self, user: User) -> User:
-        """Update an existing user."""
-        return await self._repo.update_user(user)
-
-    async def get_user_by_email(self, email: str) -> User | None:
-        """Get user by email."""
-        return await self._repo.get_user_by_email(email)

backend/app/gateway/auth/models.py

@@ -1,41 +0,0 @@
-"""User Pydantic models for authentication."""
-
-from datetime import UTC, datetime
-from typing import Literal
-from uuid import UUID, uuid4
-
-from pydantic import BaseModel, ConfigDict, EmailStr, Field
-
-
-def _utc_now() -> datetime:
-    """Return current UTC time (timezone-aware)."""
-    return datetime.now(UTC)
-
-
-class User(BaseModel):
-    """Internal user representation."""
-
-    model_config = ConfigDict(from_attributes=True)
-
-    id: UUID = Field(default_factory=uuid4, description="Primary key")
-    email: EmailStr = Field(..., description="Unique email address")
-    password_hash: str | None = Field(None, description="bcrypt hash, nullable for OAuth users")
-    system_role: Literal["admin", "user"] = Field(default="user")
-    created_at: datetime = Field(default_factory=_utc_now)
-
-    # OAuth linkage (optional)
-    oauth_provider: str | None = Field(None, description="e.g. 'github', 'google'")
-    oauth_id: str | None = Field(None, description="User ID from OAuth provider")
-
-    # Auth lifecycle
-    needs_setup: bool = Field(default=False, description="True for auto-created admin until setup completes")
-    token_version: int = Field(default=0, description="Incremented on password change to invalidate old JWTs")
-
-
-class UserResponse(BaseModel):
-    """Response model for user info endpoint."""
-
-    id: str
-    email: str
-    system_role: Literal["admin", "user"]
-    needs_setup: bool = False

backend/app/gateway/auth/password.py

@@ -1,33 +0,0 @@
-"""Password hashing utilities using bcrypt directly."""
-
-import asyncio
-
-import bcrypt
-
-
-def hash_password(password: str) -> str:
-    """Hash a password using bcrypt."""
-    return bcrypt.hashpw(password.encode("utf-8"), bcrypt.gensalt()).decode("utf-8")
-
-
-def verify_password(plain_password: str, hashed_password: str) -> bool:
-    """Verify a password against its hash."""
-    return bcrypt.checkpw(plain_password.encode("utf-8"), hashed_password.encode("utf-8"))
-
-
-async def hash_password_async(password: str) -> str:
-    """Hash a password using bcrypt (non-blocking).
-
-    Wraps the blocking bcrypt operation in a thread pool to avoid
-    blocking the event loop during password hashing.
-    """
-    return await asyncio.to_thread(hash_password, password)
-
-
-async def verify_password_async(plain_password: str, hashed_password: str) -> bool:
-    """Verify a password against its hash (non-blocking).
-
-    Wraps the blocking bcrypt operation in a thread pool to avoid
-    blocking the event loop during password verification.
-    """
-    return await asyncio.to_thread(verify_password, plain_password, hashed_password)

backend/app/gateway/auth/providers.py

@@ -1,24 +0,0 @@
-"""Auth provider abstraction."""
-
-from abc import ABC, abstractmethod
-
-
-class AuthProvider(ABC):
-    """Abstract base class for authentication providers."""
-
-    @abstractmethod
-    async def authenticate(self, credentials: dict) -> "User | None":
-        """Authenticate user with given credentials.
-
-        Returns User if authentication succeeds, None otherwise.
-        """
-        ...
-
-    @abstractmethod
-    async def get_user(self, user_id: str) -> "User | None":
-        """Retrieve user by ID."""
-        ...
-
-
-# Import User at runtime to avoid circular imports
-from app.gateway.auth.models import User  # noqa: E402

backend/app/gateway/auth/repositories/base.py

@@ -1,97 +0,0 @@
-"""User repository interface for abstracting database operations."""
-
-from abc import ABC, abstractmethod
-
-from app.gateway.auth.models import User
-
-
-class UserNotFoundError(LookupError):
-    """Raised when a user repository operation targets a non-existent row.
-
-    Subclass of :class:`LookupError` so callers that already catch
-    ``LookupError`` for "missing entity" can keep working unchanged,
-    while specific call sites can pin to this class to distinguish
-    "concurrent delete during update" from other lookups.
-    """
-
-
-class UserRepository(ABC):
-    """Abstract interface for user data storage.
-
-    Implement this interface to support different storage backends
-    (SQLite)
-    """
-
-    @abstractmethod
-    async def create_user(self, user: User) -> User:
-        """Create a new user.
-
-        Args:
-            user: User object to create
-
-        Returns:
-            Created User with ID assigned
-
-        Raises:
-            ValueError: If email already exists
-        """
-        ...
-
-    @abstractmethod
-    async def get_user_by_id(self, user_id: str) -> User | None:
-        """Get user by ID.
-
-        Args:
-            user_id: User UUID as string
-
-        Returns:
-            User if found, None otherwise
-        """
-        ...
-
-    @abstractmethod
-    async def get_user_by_email(self, email: str) -> User | None:
-        """Get user by email.
-
-        Args:
-            email: User email address
-
-        Returns:
-            User if found, None otherwise
-        """
-        ...
-
-    @abstractmethod
-    async def update_user(self, user: User) -> User:
-        """Update an existing user.
-
-        Args:
-            user: User object with updated fields
-
-        Returns:
-            Updated User
-
-        Raises:
-            UserNotFoundError: If no row exists for ``user.id``. This is
-                a hard failure (not a no-op) so callers cannot mistake a
-                concurrent-delete race for a successful update.
-        """
-        ...
-
-    @abstractmethod
-    async def count_users(self) -> int:
-        """Return total number of registered users."""
-        ...
-
-    @abstractmethod
-    async def get_user_by_oauth(self, provider: str, oauth_id: str) -> User | None:
-        """Get user by OAuth provider and ID.
-
-        Args:
-            provider: OAuth provider name (e.g. 'github', 'google')
-            oauth_id: User ID from the OAuth provider
-
-        Returns:
-            User if found, None otherwise
-        """
-        ...

backend/app/gateway/auth/repositories/sqlite.py

@@ -1,122 +0,0 @@
-"""SQLAlchemy-backed UserRepository implementation.
-
-Uses the shared async session factory from
-``deerflow.persistence.engine`` — the ``users`` table lives in the
-same database as ``threads_meta``, ``runs``, ``run_events``, and
-``feedback``.
-
-Constructor takes the session factory directly (same pattern as the
-other four repositories in ``deerflow.persistence.*``). Callers
-construct this after ``init_engine_from_config()`` has run.
-"""
-
-from __future__ import annotations
-
-from datetime import UTC
-from uuid import UUID
-
-from sqlalchemy import func, select
-from sqlalchemy.exc import IntegrityError
-from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
-
-from app.gateway.auth.models import User
-from app.gateway.auth.repositories.base import UserNotFoundError, UserRepository
-from deerflow.persistence.user.model import UserRow
-
-
-class SQLiteUserRepository(UserRepository):
-    """Async user repository backed by the shared SQLAlchemy engine."""
-
-    def __init__(self, session_factory: async_sessionmaker[AsyncSession]) -> None:
-        self._sf = session_factory
-
-    # ── Converters ────────────────────────────────────────────────────
-
-    @staticmethod
-    def _row_to_user(row: UserRow) -> User:
-        return User(
-            id=UUID(row.id),
-            email=row.email,
-            password_hash=row.password_hash,
-            system_role=row.system_role,  # type: ignore[arg-type]
-            # SQLite loses tzinfo on read; reattach UTC so downstream
-            # code can compare timestamps reliably.
-            created_at=row.created_at if row.created_at.tzinfo else row.created_at.replace(tzinfo=UTC),
-            oauth_provider=row.oauth_provider,
-            oauth_id=row.oauth_id,
-            needs_setup=row.needs_setup,
-            token_version=row.token_version,
-        )
-
-    @staticmethod
-    def _user_to_row(user: User) -> UserRow:
-        return UserRow(
-            id=str(user.id),
-            email=user.email,
-            password_hash=user.password_hash,
-            system_role=user.system_role,
-            created_at=user.created_at,
-            oauth_provider=user.oauth_provider,
-            oauth_id=user.oauth_id,
-            needs_setup=user.needs_setup,
-            token_version=user.token_version,
-        )
-
-    # ── CRUD ──────────────────────────────────────────────────────────
-
-    async def create_user(self, user: User) -> User:
-        """Insert a new user. Raises ``ValueError`` on duplicate email."""
-        row = self._user_to_row(user)
-        async with self._sf() as session:
-            session.add(row)
-            try:
-                await session.commit()
-            except IntegrityError as exc:
-                await session.rollback()
-                raise ValueError(f"Email already registered: {user.email}") from exc
-        return user
-
-    async def get_user_by_id(self, user_id: str) -> User | None:
-        async with self._sf() as session:
-            row = await session.get(UserRow, user_id)
-            return self._row_to_user(row) if row is not None else None
-
-    async def get_user_by_email(self, email: str) -> User | None:
-        stmt = select(UserRow).where(UserRow.email == email)
-        async with self._sf() as session:
-            result = await session.execute(stmt)
-            row = result.scalar_one_or_none()
-            return self._row_to_user(row) if row is not None else None
-
-    async def update_user(self, user: User) -> User:
-        async with self._sf() as session:
-            row = await session.get(UserRow, str(user.id))
-            if row is None:
-                # Hard fail on concurrent delete: callers (reset_admin,
-                # password change handlers, _ensure_admin_user) all
-                # fetched the user just before this call, so a missing
-                # row here means the row vanished underneath us. Silent
-                # success would let the caller log "password reset" for
-                # a row that no longer exists.
-                raise UserNotFoundError(f"User {user.id} no longer exists")
-            row.email = user.email
-            row.password_hash = user.password_hash
-            row.system_role = user.system_role
-            row.oauth_provider = user.oauth_provider
-            row.oauth_id = user.oauth_id
-            row.needs_setup = user.needs_setup
-            row.token_version = user.token_version
-            await session.commit()
-        return user
-
-    async def count_users(self) -> int:
-        stmt = select(func.count()).select_from(UserRow)
-        async with self._sf() as session:
-            return await session.scalar(stmt) or 0
-
-    async def get_user_by_oauth(self, provider: str, oauth_id: str) -> User | None:
-        stmt = select(UserRow).where(UserRow.oauth_provider == provider, UserRow.oauth_id == oauth_id)
-        async with self._sf() as session:
-            result = await session.execute(stmt)
-            row = result.scalar_one_or_none()
-            return self._row_to_user(row) if row is not None else None

backend/app/gateway/auth/reset_admin.py

@@ -1,91 +0,0 @@
-"""CLI tool to reset an admin password.
-
-Usage:
-    python -m app.gateway.auth.reset_admin
-    python -m app.gateway.auth.reset_admin --email admin@example.com
-
-Writes the new password to ``.deer-flow/admin_initial_credentials.txt``
-(mode 0600) instead of printing it, so CI / log aggregators never see
-the cleartext secret.
-"""
-
-from __future__ import annotations
-
-import argparse
-import asyncio
-import secrets
-import sys
-
-from sqlalchemy import select
-
-from app.gateway.auth.credential_file import write_initial_credentials
-from app.gateway.auth.password import hash_password
-from app.gateway.auth.repositories.sqlite import SQLiteUserRepository
-from deerflow.persistence.user.model import UserRow
-
-
-async def _run(email: str | None) -> int:
-    from deerflow.config import get_app_config
-    from deerflow.persistence.engine import (
-        close_engine,
-        get_session_factory,
-        init_engine_from_config,
-    )
-
-    config = get_app_config()
-    await init_engine_from_config(config.database)
-    try:
-        sf = get_session_factory()
-        if sf is None:
-            print("Error: persistence engine not available (check config.database).", file=sys.stderr)
-            return 1
-
-        repo = SQLiteUserRepository(sf)
-
-        if email:
-            user = await repo.get_user_by_email(email)
-        else:
-            # Find first admin via direct SELECT — repository does not
-            # expose a "first admin" helper and we do not want to add
-            # one just for this CLI.
-            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:
-                user = None
-            else:
-                user = await repo.get_user_by_id(row.id)
-
-        if user is None:
-            if email:
-                print(f"Error: user '{email}' not found.", file=sys.stderr)
-            else:
-                print("Error: no admin user found.", file=sys.stderr)
-            return 1
-
-        new_password = secrets.token_urlsafe(16)
-        user.password_hash = hash_password(new_password)
-        user.token_version += 1
-        user.needs_setup = True
-        await repo.update_user(user)
-
-        cred_path = write_initial_credentials(user.email, new_password, label="reset")
-        print(f"Password reset for: {user.email}")
-        print(f"Credentials written to: {cred_path} (mode 0600)")
-        print("Next login will require setup (new email + password).")
-        return 0
-    finally:
-        await close_engine()
-
-
-def main() -> None:
-    parser = argparse.ArgumentParser(description="Reset admin password")
-    parser.add_argument("--email", help="Admin email (default: first admin found)")
-    args = parser.parse_args()
-
-    exit_code = asyncio.run(_run(args.email))
-    sys.exit(exit_code)
-
-
-if __name__ == "__main__":
-    main()

backend/app/gateway/auth_middleware.py

@@ -1,117 +0,0 @@
-"""Global authentication middleware — fail-closed safety net.
-
-Rejects unauthenticated requests to non-public paths with 401. When a
-request passes the cookie check, resolves the JWT payload to a real
-``User`` object and stamps it into both ``request.state.user`` and the
-``deerflow.runtime.user_context`` contextvar so that repository-layer
-owner filtering works automatically via the sentinel pattern.
-
-Fine-grained permission checks remain in authz.py decorators.
-"""
-
-from collections.abc import Callable
-
-from fastapi import HTTPException, Request, Response
-from starlette.middleware.base import BaseHTTPMiddleware
-from starlette.responses import JSONResponse
-from starlette.types import ASGIApp
-
-from app.gateway.auth.errors import AuthErrorCode, AuthErrorResponse
-from app.gateway.authz import _ALL_PERMISSIONS, AuthContext
-from deerflow.runtime.user_context import reset_current_user, set_current_user
-
-# Paths that never require authentication.
-_PUBLIC_PATH_PREFIXES: tuple[str, ...] = (
-    "/health",
-    "/docs",
-    "/redoc",
-    "/openapi.json",
-)
-
-# Exact auth paths that are public (login/register/status check).
-# /api/v1/auth/me, /api/v1/auth/change-password etc. are NOT public.
-_PUBLIC_EXACT_PATHS: frozenset[str] = frozenset(
-    {
-        "/api/v1/auth/login/local",
-        "/api/v1/auth/register",
-        "/api/v1/auth/logout",
-        "/api/v1/auth/setup-status",
-    }
-)
-
-
-def _is_public(path: str) -> bool:
-    stripped = path.rstrip("/")
-    if stripped in _PUBLIC_EXACT_PATHS:
-        return True
-    return any(path.startswith(prefix) for prefix in _PUBLIC_PATH_PREFIXES)
-
-
-class AuthMiddleware(BaseHTTPMiddleware):
-    """Strict auth gate: reject requests without a valid session.
-
-    Two-stage check for non-public paths:
-
-    1. Cookie presence — return 401 NOT_AUTHENTICATED if missing
-    2. JWT validation via ``get_optional_user_from_request`` — return 401
-       TOKEN_INVALID if the token is absent, malformed, expired, or the
-       signed user does not exist / is stale
-
-    On success, stamps ``request.state.user`` and the
-    ``deerflow.runtime.user_context`` contextvar so that repository-layer
-    owner filters work downstream without every route needing a
-    ``@require_auth`` decorator. Routes that need per-resource
-    authorization (e.g. "user A cannot read user B's thread by guessing
-    the URL") should additionally use ``@require_permission(...,
-    owner_check=True)`` for explicit enforcement — but authentication
-    itself is fully handled here.
-    """
-
-    def __init__(self, app: ASGIApp) -> None:
-        super().__init__(app)
-
-    async def dispatch(self, request: Request, call_next: Callable) -> Response:
-        if _is_public(request.url.path):
-            return await call_next(request)
-
-        # Non-public path: require session cookie
-        if not request.cookies.get("access_token"):
-            return JSONResponse(
-                status_code=401,
-                content={
-                    "detail": AuthErrorResponse(
-                        code=AuthErrorCode.NOT_AUTHENTICATED,
-                        message="Authentication required",
-                    ).model_dump()
-                },
-            )
-
-        # Strict JWT validation: reject junk/expired tokens with 401
-        # right here instead of silently passing through. This closes
-        # the "junk cookie bypass" gap (AUTH_TEST_PLAN test 7.5.8):
-        # without this, non-isolation routes like /api/models would
-        # accept any cookie-shaped string as authentication.
-        #
-        # We call the *strict* resolver so that fine-grained error
-        # codes (token_expired, token_invalid, user_not_found, …)
-        # propagate from AuthErrorCode, not get flattened into one
-        # generic code. BaseHTTPMiddleware doesn't let HTTPException
-        # bubble up, so we catch and render it as JSONResponse here.
-        from app.gateway.deps import get_current_user_from_request
-
-        try:
-            user = await get_current_user_from_request(request)
-        except HTTPException as exc:
-            return JSONResponse(status_code=exc.status_code, content={"detail": exc.detail})
-
-        # Stamp both request.state.user (for the contextvar pattern)
-        # and request.state.auth (so @require_permission's "auth is
-        # None" branch short-circuits instead of running the entire
-        # JWT-decode + DB-lookup pipeline a second time per request).
-        request.state.user = user
-        request.state.auth = AuthContext(user=user, permissions=_ALL_PERMISSIONS)
-        token = set_current_user(user)
-        try:
-            return await call_next(request)
-        finally:
-            reset_current_user(token)

backend/app/gateway/authz.py

@@ -1,262 +0,0 @@
-"""Authorization decorators and context for DeerFlow.
-
-Inspired by LangGraph Auth system: https://github.com/langchain-ai/langgraph/blob/main/libs/sdk-py/langgraph_sdk/auth/__init__.py
-
-**Usage:**
-
-1. Use ``@require_auth`` on routes that need authentication
-2. Use ``@require_permission("resource", "action", filter_key=...)`` for permission checks
-3. The decorator chain processes from bottom to top
-
-**Example:**
-
-    @router.get("/{thread_id}")
-    @require_auth
-    @require_permission("threads", "read", owner_check=True)
-    async def get_thread(thread_id: str, request: Request):
-        # User is authenticated and has threads:read permission
-        ...
-
-**Permission Model:**
-
-- threads:read   - View thread
-- threads:write  - Create/update thread
-- threads:delete - Delete thread
-- runs:create   - Run agent
-- runs:read     - View run
-- runs:cancel   - Cancel run
-"""
-
-from __future__ import annotations
-
-import functools
-from collections.abc import Callable
-from typing import TYPE_CHECKING, Any, ParamSpec, TypeVar
-
-from fastapi import HTTPException, Request
-
-if TYPE_CHECKING:
-    from app.gateway.auth.models import User
-
-P = ParamSpec("P")
-T = TypeVar("T")
-
-
-# Permission constants
-class Permissions:
-    """Permission constants for resource:action format."""
-
-    # Threads
-    THREADS_READ = "threads:read"
-    THREADS_WRITE = "threads:write"
-    THREADS_DELETE = "threads:delete"
-
-    # Runs
-    RUNS_CREATE = "runs:create"
-    RUNS_READ = "runs:read"
-    RUNS_CANCEL = "runs:cancel"
-
-
-class AuthContext:
-    """Authentication context for the current request.
-
-    Stored in request.state.auth after require_auth decoration.
-
-    Attributes:
-        user: The authenticated user, or None if anonymous
-        permissions: List of permission strings (e.g., "threads:read")
-    """
-
-    __slots__ = ("user", "permissions")
-
-    def __init__(self, user: User | None = None, permissions: list[str] | None = None):
-        self.user = user
-        self.permissions = permissions or []
-
-    @property
-    def is_authenticated(self) -> bool:
-        """Check if user is authenticated."""
-        return self.user is not None
-
-    def has_permission(self, resource: str, action: str) -> bool:
-        """Check if context has permission for resource:action.
-
-        Args:
-            resource: Resource name (e.g., "threads")
-            action: Action name (e.g., "read")
-
-        Returns:
-            True if user has permission
-        """
-        permission = f"{resource}:{action}"
-        return permission in self.permissions
-
-    def require_user(self) -> User:
-        """Get user or raise 401.
-
-        Raises:
-            HTTPException 401 if not authenticated
-        """
-        if not self.user:
-            raise HTTPException(status_code=401, detail="Authentication required")
-        return self.user
-
-
-def get_auth_context(request: Request) -> AuthContext | None:
-    """Get AuthContext from request state."""
-    return getattr(request.state, "auth", None)
-
-
-_ALL_PERMISSIONS: list[str] = [
-    Permissions.THREADS_READ,
-    Permissions.THREADS_WRITE,
-    Permissions.THREADS_DELETE,
-    Permissions.RUNS_CREATE,
-    Permissions.RUNS_READ,
-    Permissions.RUNS_CANCEL,
-]
-
-
-async def _authenticate(request: Request) -> AuthContext:
-    """Authenticate request and return AuthContext.
-
-    Delegates to deps.get_optional_user_from_request() for the JWT→User pipeline.
-    Returns AuthContext with user=None for anonymous requests.
-    """
-    from app.gateway.deps import get_optional_user_from_request
-
-    user = await get_optional_user_from_request(request)
-    if user is None:
-        return AuthContext(user=None, permissions=[])
-
-    # In future, permissions could be stored in user record
-    return AuthContext(user=user, permissions=_ALL_PERMISSIONS)
-
-
-def require_auth[**P, T](func: Callable[P, T]) -> Callable[P, T]:
-    """Decorator that authenticates the request and sets AuthContext.
-
-    Must be placed ABOVE other decorators (executes after them).
-
-    Usage:
-        @router.get("/{thread_id}")
-        @require_auth  # Bottom decorator (executes first after permission check)
-        @require_permission("threads", "read")
-        async def get_thread(thread_id: str, request: Request):
-            auth: AuthContext = request.state.auth
-            ...
-
-    Raises:
-        ValueError: If 'request' parameter is missing
-    """
-
-    @functools.wraps(func)
-    async def wrapper(*args: Any, **kwargs: Any) -> Any:
-        request = kwargs.get("request")
-        if request is None:
-            raise ValueError("require_auth decorator requires 'request' parameter")
-
-        # Authenticate and set context
-        auth_context = await _authenticate(request)
-        request.state.auth = auth_context
-
-        return await func(*args, **kwargs)
-
-    return wrapper
-
-
-def require_permission(
-    resource: str,
-    action: str,
-    owner_check: bool = False,
-    require_existing: bool = False,
-) -> Callable[[Callable[P, T]], Callable[P, T]]:
-    """Decorator that checks permission for resource:action.
-
-    Must be used AFTER @require_auth.
-
-    Args:
-        resource: Resource name (e.g., "threads", "runs")
-        action: Action name (e.g., "read", "write", "delete")
-        owner_check: If True, validates that the current user owns the resource.
-                     Requires 'thread_id' path parameter and performs ownership check.
-        require_existing: Only meaningful with ``owner_check=True``. If True, a
-                          missing ``threads_meta`` row counts as a denial (404)
-                          instead of "untracked legacy thread, allow". Use on
-                          **destructive / mutating** routes (DELETE, PATCH,
-                          state-update) so a deleted thread can't be re-targeted
-                          by another user via the missing-row code path.
-
-    Usage:
-        # Read-style: legacy untracked threads are allowed
-        @require_permission("threads", "read", owner_check=True)
-        async def get_thread(thread_id: str, request: Request):
-            ...
-
-        # Destructive: thread row MUST exist and be owned by caller
-        @require_permission("threads", "delete", owner_check=True, require_existing=True)
-        async def delete_thread(thread_id: str, request: Request):
-            ...
-
-    Raises:
-        HTTPException 401: If authentication required but user is anonymous
-        HTTPException 403: If user lacks permission
-        HTTPException 404: If owner_check=True but user doesn't own the thread
-        ValueError: If owner_check=True but 'thread_id' parameter is missing
-    """
-
-    def decorator(func: Callable[P, T]) -> Callable[P, T]:
-        @functools.wraps(func)
-        async def wrapper(*args: Any, **kwargs: Any) -> Any:
-            request = kwargs.get("request")
-            if request is None:
-                raise ValueError("require_permission decorator requires 'request' parameter")
-
-            auth: AuthContext = getattr(request.state, "auth", None)
-            if auth is None:
-                auth = await _authenticate(request)
-                request.state.auth = auth
-
-            if not auth.is_authenticated:
-                raise HTTPException(status_code=401, detail="Authentication required")
-
-            # Check permission
-            if not auth.has_permission(resource, action):
-                raise HTTPException(
-                    status_code=403,
-                    detail=f"Permission denied: {resource}:{action}",
-                )
-
-            # Owner check for thread-specific resources.
-            #
-            # 2.0-rc moved thread metadata into the SQL persistence layer
-            # (``threads_meta`` table). We verify ownership via
-            # ``ThreadMetaStore.check_access``: it returns True for
-            # missing rows (untracked legacy thread) and for rows whose
-            # ``owner_id`` is NULL (shared / pre-auth data), so this is
-            # strict-deny rather than strict-allow — only an *existing*
-            # row with a *different* owner_id triggers 404.
-            if owner_check:
-                thread_id = kwargs.get("thread_id")
-                if thread_id is None:
-                    raise ValueError("require_permission with owner_check=True requires 'thread_id' parameter")
-
-                from app.gateway.deps import get_thread_meta_repo
-
-                thread_meta_repo = get_thread_meta_repo(request)
-                allowed = await thread_meta_repo.check_access(
-                    thread_id,
-                    str(auth.user.id),
-                    require_existing=require_existing,
-                )
-                if not allowed:
-                    raise HTTPException(
-                        status_code=404,
-                        detail=f"Thread {thread_id} not found",
-                    )
-
-            return await func(*args, **kwargs)
-
-        return wrapper
-
-    return decorator

backend/app/gateway/csrf_middleware.py

@@ -1,112 +0,0 @@
-"""CSRF protection middleware for FastAPI.
-
-Per RFC-001:
-State-changing operations require CSRF protection.
-"""
-
-import secrets
-from collections.abc import Callable
-
-from fastapi import Request, Response
-from starlette.middleware.base import BaseHTTPMiddleware
-from starlette.responses import JSONResponse
-from starlette.types import ASGIApp
-
-CSRF_COOKIE_NAME = "csrf_token"
-CSRF_HEADER_NAME = "X-CSRF-Token"
-CSRF_TOKEN_LENGTH = 64  # bytes
-
-
-def is_secure_request(request: Request) -> bool:
-    """Detect whether the original client request was made over HTTPS."""
-    return request.headers.get("x-forwarded-proto", request.url.scheme) == "https"
-
-
-def generate_csrf_token() -> str:
-    """Generate a secure random CSRF token."""
-    return secrets.token_urlsafe(CSRF_TOKEN_LENGTH)
-
-
-def should_check_csrf(request: Request) -> bool:
-    """Determine if a request needs CSRF validation.
-
-    CSRF is checked for state-changing methods (POST, PUT, DELETE, PATCH).
-    GET, HEAD, OPTIONS, and TRACE are exempt per RFC 7231.
-    """
-    if request.method not in ("POST", "PUT", "DELETE", "PATCH"):
-        return False
-
-    path = request.url.path.rstrip("/")
-    # Exempt /api/v1/auth/me endpoint
-    if path == "/api/v1/auth/me":
-        return False
-    return True
-
-
-_AUTH_EXEMPT_PATHS: frozenset[str] = frozenset(
-    {
-        "/api/v1/auth/login/local",
-        "/api/v1/auth/logout",
-        "/api/v1/auth/register",
-    }
-)
-
-
-def is_auth_endpoint(request: Request) -> bool:
-    """Check if the request is to an auth endpoint.
-
-    Auth endpoints don't need CSRF validation on first call (no token).
-    """
-    return request.url.path.rstrip("/") in _AUTH_EXEMPT_PATHS
-
-
-class CSRFMiddleware(BaseHTTPMiddleware):
-    """Middleware that implements CSRF protection using Double Submit Cookie pattern."""
-
-    def __init__(self, app: ASGIApp) -> None:
-        super().__init__(app)
-
-    async def dispatch(self, request: Request, call_next: Callable) -> Response:
-        _is_auth = is_auth_endpoint(request)
-
-        if should_check_csrf(request) and not _is_auth:
-            cookie_token = request.cookies.get(CSRF_COOKIE_NAME)
-            header_token = request.headers.get(CSRF_HEADER_NAME)
-
-            if not cookie_token or not header_token:
-                return JSONResponse(
-                    status_code=403,
-                    content={"detail": "CSRF token missing. Include X-CSRF-Token header."},
-                )
-
-            if not secrets.compare_digest(cookie_token, header_token):
-                return JSONResponse(
-                    status_code=403,
-                    content={"detail": "CSRF token mismatch."},
-                )
-
-        response = await call_next(request)
-
-        # For auth endpoints that set up session, also set CSRF cookie
-        if _is_auth and request.method == "POST":
-            # Generate a new CSRF token for the session
-            csrf_token = generate_csrf_token()
-            is_https = is_secure_request(request)
-            response.set_cookie(
-                key=CSRF_COOKIE_NAME,
-                value=csrf_token,
-                httponly=False,  # Must be JS-readable for Double Submit Cookie pattern
-                secure=is_https,
-                samesite="strict",
-            )
-
-        return response
-
-
-def get_csrf_token(request: Request) -> str | None:
-    """Get the CSRF token from the current request's cookies.
-
-    This is useful for server-side rendering where you need to embed
-    token in forms or headers.
-    """
-    return request.cookies.get(CSRF_COOKIE_NAME)

backend/app/gateway/deps.py

@@ -1,8 +1,7 @@
 """Centralized accessors for singleton objects stored on ``app.state``.
 
 **Getters** (used by routers): raise 503 when a required dependency is
-missing, except ``get_store`` and ``get_thread_meta_repo`` which return
+missing, except ``get_store`` which returns ``None``.
-``None``.
 
 Initialization is handled directly in ``app.py`` via :class:`AsyncExitStack`.
 """
@@ -11,15 +10,10 @@ from __future__ import annotations
 
 from collections.abc import AsyncGenerator
 from contextlib import AsyncExitStack, asynccontextmanager
-from typing import TYPE_CHECKING
 
 from fastapi import FastAPI, HTTPException, Request
 
-from deerflow.runtime import RunContext, RunManager
+from deerflow.runtime import RunManager, StreamBridge
-
-if TYPE_CHECKING:
-    from app.gateway.auth.local_provider import LocalAuthProvider
-    from app.gateway.auth.repositories.sqlite import SQLiteUserRepository
 
 
 @asynccontextmanager
@@ -32,194 +26,45 @@ async def langgraph_runtime(app: FastAPI) -> AsyncGenerator[None, None]:
             yield
     """
     from deerflow.agents.checkpointer.async_provider import make_checkpointer
-    from deerflow.config import get_app_config
-    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.events.store import make_run_event_store
 
     async with AsyncExitStack() as stack:
         app.state.stream_bridge = await stack.enter_async_context(make_stream_bridge())
-
-        # Initialize persistence engine BEFORE checkpointer so that
-        # auto-create-database logic runs first (postgres backend).
-        config = get_app_config()
-        await init_engine_from_config(config.database)
-
         app.state.checkpointer = await stack.enter_async_context(make_checkpointer())
         app.state.store = await stack.enter_async_context(make_store())
-
+        app.state.run_manager = RunManager()
-        # Initialize repositories — one get_session_factory() call for all.
+        yield
-        sf = get_session_factory()
-        if sf is not None:
-            from deerflow.persistence.feedback import FeedbackRepository
-            from deerflow.persistence.run import RunRepository
-            from deerflow.persistence.thread_meta import ThreadMetaRepository
-
-            app.state.run_store = RunRepository(sf)
-            app.state.feedback_repo = FeedbackRepository(sf)
-            app.state.thread_meta_repo = ThreadMetaRepository(sf)
-        else:
-            from deerflow.persistence.thread_meta import MemoryThreadMetaStore
-            from deerflow.runtime.runs.store.memory import MemoryRunStore
-
-            app.state.run_store = MemoryRunStore()
-            app.state.feedback_repo = None
-            app.state.thread_meta_repo = MemoryThreadMetaStore(app.state.store)
-
-        # 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)
-
-        try:
-            yield
-        finally:
-            await close_engine()
 
 
 # ---------------------------------------------------------------------------
-# Getters -- called by routers per-request
+# Getters – called by routers per-request
 # ---------------------------------------------------------------------------
 
 
-def _require(attr: str, label: str):
+def get_stream_bridge(request: Request) -> StreamBridge:
-    """Create a FastAPI dependency that returns ``app.state.<attr>`` or 503."""
+    """Return the global :class:`StreamBridge`, or 503."""
-
+    bridge = getattr(request.app.state, "stream_bridge", None)
-    def dep(request: Request):
+    if bridge is None:
-        val = getattr(request.app.state, attr, None)
+        raise HTTPException(status_code=503, detail="Stream bridge not available")
-        if val is None:
+    return bridge
-            raise HTTPException(status_code=503, detail=f"{label} not available")
-        return val
-
-    dep.__name__ = dep.__qualname__ = f"get_{attr}"
-    return dep
 
 
-get_stream_bridge = _require("stream_bridge", "Stream bridge")
+def get_run_manager(request: Request) -> RunManager:
-get_run_manager = _require("run_manager", "Run manager")
+    """Return the global :class:`RunManager`, or 503."""
-get_checkpointer = _require("checkpointer", "Checkpointer")
+    mgr = getattr(request.app.state, "run_manager", None)
-get_run_event_store = _require("run_event_store", "Run event store")
+    if mgr is None:
-get_feedback_repo = _require("feedback_repo", "Feedback")
+        raise HTTPException(status_code=503, detail="Run manager not available")
-get_run_store = _require("run_store", "Run store")
+    return mgr
+
+
+def get_checkpointer(request: Request):
+    """Return the global checkpointer, or 503."""
+    cp = getattr(request.app.state, "checkpointer", None)
+    if cp is None:
+        raise HTTPException(status_code=503, detail="Checkpointer not available")
+    return cp
 
 
 def get_store(request: Request):
     """Return the global store (may be ``None`` if not configured)."""
     return getattr(request.app.state, "store", None)
-
-
-get_thread_meta_repo = _require("thread_meta_repo", "Thread metadata store")
-
-
-def get_run_context(request: Request) -> RunContext:
-    """Build a :class:`RunContext` from ``app.state`` singletons.
-
-    Returns a *base* context with infrastructure dependencies.  Callers that
-    need per-run fields (e.g. ``follow_up_to_run_id``) should use
-    ``dataclasses.replace(ctx, follow_up_to_run_id=...)`` before passing it
-    to :func:`run_agent`.
-    """
-    from deerflow.config import get_app_config
-
-    return RunContext(
-        checkpointer=get_checkpointer(request),
-        store=get_store(request),
-        event_store=get_run_event_store(request),
-        run_events_config=getattr(get_app_config(), "run_events", None),
-        thread_meta_repo=get_thread_meta_repo(request),
-    )
-
-
-# ---------------------------------------------------------------------------
-# 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_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 the request cookie.
-
-    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 = request.cookies.get("access_token")
-    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

backend/app/gateway/langgraph_auth.py

@@ -1,106 +0,0 @@
-"""LangGraph Server auth handler — shares JWT logic with Gateway.
-
-Loaded by LangGraph Server via langgraph.json ``auth.path``.
-Reuses the same ``decode_token`` / ``get_auth_config`` as Gateway,
-so both modes validate tokens with the same secret and rules.
-
-Two layers:
-  1. @auth.authenticate — validates JWT cookie, extracts user_id,
-     and enforces CSRF on state-changing methods (POST/PUT/DELETE/PATCH)
-  2. @auth.on — returns metadata filter so each user only sees own threads
-"""
-
-import secrets
-
-from langgraph_sdk import Auth
-
-from app.gateway.auth.errors import TokenError
-from app.gateway.auth.jwt import decode_token
-from app.gateway.deps import get_local_provider
-
-auth = Auth()
-
-# Methods that require CSRF validation (state-changing per RFC 7231).
-_CSRF_METHODS = frozenset({"POST", "PUT", "DELETE", "PATCH"})
-
-
-def _check_csrf(request) -> None:
-    """Enforce Double Submit Cookie CSRF check for state-changing requests.
-
-    Mirrors Gateway's CSRFMiddleware logic so that LangGraph routes
-    proxied directly by nginx have the same CSRF protection.
-    """
-    method = getattr(request, "method", "") or ""
-    if method.upper() not in _CSRF_METHODS:
-        return
-
-    cookie_token = request.cookies.get("csrf_token")
-    header_token = request.headers.get("x-csrf-token")
-
-    if not cookie_token or not header_token:
-        raise Auth.exceptions.HTTPException(
-            status_code=403,
-            detail="CSRF token missing. Include X-CSRF-Token header.",
-        )
-
-    if not secrets.compare_digest(cookie_token, header_token):
-        raise Auth.exceptions.HTTPException(
-            status_code=403,
-            detail="CSRF token mismatch.",
-        )
-
-
-@auth.authenticate
-async def authenticate(request):
-    """Validate the session cookie, decode JWT, and check token_version.
-
-    Same validation chain as Gateway's get_current_user_from_request:
-      cookie → decode JWT → DB lookup → token_version match
-    Also enforces CSRF on state-changing methods.
-    """
-    # CSRF check before authentication so forged cross-site requests
-    # are rejected early, even if the cookie carries a valid JWT.
-    _check_csrf(request)
-
-    token = request.cookies.get("access_token")
-    if not token:
-        raise Auth.exceptions.HTTPException(
-            status_code=401,
-            detail="Not authenticated",
-        )
-
-    payload = decode_token(token)
-    if isinstance(payload, TokenError):
-        raise Auth.exceptions.HTTPException(
-            status_code=401,
-            detail=f"Token error: {payload.value}",
-        )
-
-    user = await get_local_provider().get_user(payload.sub)
-    if user is None:
-        raise Auth.exceptions.HTTPException(
-            status_code=401,
-            detail="User not found",
-        )
-    if user.token_version != payload.ver:
-        raise Auth.exceptions.HTTPException(
-            status_code=401,
-            detail="Token revoked (password changed)",
-        )
-
-    return payload.sub
-
-
-@auth.on
-async def add_owner_filter(ctx: Auth.types.AuthContext, value: dict):
-    """Inject owner_id metadata on writes; filter by owner_id on reads.
-
-    Gateway stores thread ownership as ``metadata.owner_id``.
-    This handler ensures LangGraph Server enforces the same isolation.
-    """
-    # On create/update: stamp owner_id into metadata
-    metadata = value.setdefault("metadata", {})
-    metadata["owner_id"] = ctx.user.identity
-
-    # Return filter dict — LangGraph applies it to search/read/delete
-    return {"owner_id": ctx.user.identity}

backend/app/gateway/routers/artifacts.py

@@ -7,7 +7,6 @@ from urllib.parse import quote
 from fastapi import APIRouter, HTTPException, Request
 from fastapi.responses import FileResponse, PlainTextResponse, Response
 
-from app.gateway.authz import require_permission
 from app.gateway.path_utils import resolve_thread_virtual_path
 
 logger = logging.getLogger(__name__)
@@ -82,7 +81,6 @@ def _extract_file_from_skill_archive(zip_path: Path, internal_path: str) -> byte
     summary="Get Artifact File",
     description="Retrieve an artifact file generated by the AI agent. Text and binary files can be viewed inline, while active web content is always downloaded.",
 )
-@require_permission("threads", "read", owner_check=True)
 async def get_artifact(thread_id: str, path: str, request: Request, download: bool = False) -> Response:
     """Get an artifact file by its path.
 

backend/app/gateway/routers/auth.py

@@ -1,418 +0,0 @@
-"""Authentication endpoints."""
-
-import logging
-import os
-import time
-from ipaddress import ip_address, ip_network
-
-from fastapi import APIRouter, Depends, HTTPException, Request, Response, status
-from fastapi.security import OAuth2PasswordRequestForm
-from pydantic import BaseModel, EmailStr, Field, field_validator
-
-from app.gateway.auth import (
-    UserResponse,
-    create_access_token,
-)
-from app.gateway.auth.config import get_auth_config
-from app.gateway.auth.errors import AuthErrorCode, AuthErrorResponse
-from app.gateway.csrf_middleware import is_secure_request
-from app.gateway.deps import get_current_user_from_request, get_local_provider
-
-logger = logging.getLogger(__name__)
-
-router = APIRouter(prefix="/api/v1/auth", tags=["auth"])
-
-
-# ── Request/Response Models ──────────────────────────────────────────────
-
-
-class LoginResponse(BaseModel):
-    """Response model for login — token only lives in HttpOnly cookie."""
-
-    expires_in: int  # seconds
-    needs_setup: bool = False
-
-
-# Top common-password blocklist. Drawn from the public SecLists "10k worst
-# passwords" set, lowercased + length>=8 only (shorter ones already fail
-# the min_length check). Kept tight on purpose: this is the **lower bound**
-# defense, not a full HIBP / passlib check, and runs in-process per request.
-_COMMON_PASSWORDS: frozenset[str] = frozenset(
-    {
-        "password",
-        "password1",
-        "password12",
-        "password123",
-        "password1234",
-        "12345678",
-        "123456789",
-        "1234567890",
-        "qwerty12",
-        "qwertyui",
-        "qwerty123",
-        "abc12345",
-        "abcd1234",
-        "iloveyou",
-        "letmein1",
-        "welcome1",
-        "welcome123",
-        "admin123",
-        "administrator",
-        "passw0rd",
-        "p@ssw0rd",
-        "monkey12",
-        "trustno1",
-        "sunshine",
-        "princess",
-        "football",
-        "baseball",
-        "superman",
-        "batman123",
-        "starwars",
-        "dragon123",
-        "master123",
-        "shadow12",
-        "michael1",
-        "jennifer",
-        "computer",
-    }
-)
-
-
-def _password_is_common(password: str) -> bool:
-    """Case-insensitive blocklist check.
-
-    Lowercases the input so trivial mutations like ``Password`` /
-    ``PASSWORD`` are also rejected. Does not normalize digit substitutions
-    (``p@ssw0rd`` is included as a literal entry instead) — keeping the
-    rule cheap and predictable.
-    """
-    return password.lower() in _COMMON_PASSWORDS
-
-
-def _validate_strong_password(value: str) -> str:
-    """Pydantic field-validator body shared by Register + ChangePassword.
-
-    Constraint = function, not type-level mixin. The two request models
-    have no "is-a" relationship; they only share the password-strength
-    rule. Lifting it into a free function lets each model bind it via
-    ``@field_validator(field_name)`` without inheritance gymnastics.
-    """
-    if _password_is_common(value):
-        raise ValueError("Password is too common; choose a stronger password.")
-    return value
-
-
-class RegisterRequest(BaseModel):
-    """Request model for user registration."""
-
-    email: EmailStr
-    password: str = Field(..., min_length=8)
-
-    _strong_password = field_validator("password")(classmethod(lambda cls, v: _validate_strong_password(v)))
-
-
-class ChangePasswordRequest(BaseModel):
-    """Request model for password change (also handles setup flow)."""
-
-    current_password: str
-    new_password: str = Field(..., min_length=8)
-    new_email: EmailStr | None = None
-
-    _strong_password = field_validator("new_password")(classmethod(lambda cls, v: _validate_strong_password(v)))
-
-
-class MessageResponse(BaseModel):
-    """Generic message response."""
-
-    message: str
-
-
-# ── Helpers ───────────────────────────────────────────────────────────────
-
-
-def _set_session_cookie(response: Response, token: str, request: Request) -> None:
-    """Set the access_token HttpOnly cookie on the response."""
-    config = get_auth_config()
-    is_https = is_secure_request(request)
-    response.set_cookie(
-        key="access_token",
-        value=token,
-        httponly=True,
-        secure=is_https,
-        samesite="lax",
-        max_age=config.token_expiry_days * 24 * 3600 if is_https else None,
-    )
-
-
-# ── Rate Limiting ────────────────────────────────────────────────────────
-# In-process dict — not shared across workers. Sufficient for single-worker deployments.
-
-_MAX_LOGIN_ATTEMPTS = 5
-_LOCKOUT_SECONDS = 300  # 5 minutes
-
-# ip → (fail_count, lock_until_timestamp)
-_login_attempts: dict[str, tuple[int, float]] = {}
-
-
-def _trusted_proxies() -> list:
-    """Parse ``AUTH_TRUSTED_PROXIES`` env var into a list of ip_network objects.
-
-    Comma-separated CIDR or single-IP entries. Empty / unset = no proxy is
-    trusted (direct mode). Invalid entries are skipped with a logger warning.
-    Read live so env-var overrides take effect immediately and tests can
-    ``monkeypatch.setenv`` without poking a module-level cache.
-    """
-    raw = os.getenv("AUTH_TRUSTED_PROXIES", "").strip()
-    if not raw:
-        return []
-    nets = []
-    for entry in raw.split(","):
-        entry = entry.strip()
-        if not entry:
-            continue
-        try:
-            nets.append(ip_network(entry, strict=False))
-        except ValueError:
-            logger.warning("AUTH_TRUSTED_PROXIES: ignoring invalid entry %r", entry)
-    return nets
-
-
-def _get_client_ip(request: Request) -> str:
-    """Extract the real client IP for rate limiting.
-
-    Trust model:
-
-    - The TCP peer (``request.client.host``) is always the baseline. It is
-      whatever the kernel reports as the connecting socket — unforgeable
-      by the client itself.
-    - ``X-Real-IP`` is **only** honored if the TCP peer is in the
-      ``AUTH_TRUSTED_PROXIES`` allowlist (set via env var, comma-separated
-      CIDR or single IPs). When set, the gateway is assumed to be behind a
-      reverse proxy (nginx, Cloudflare, ALB, …) that overwrites
-      ``X-Real-IP`` with the original client address.
-    - With no ``AUTH_TRUSTED_PROXIES`` set, ``X-Real-IP`` is silently
-      ignored — closing the bypass where any client could rotate the
-      header to dodge per-IP rate limits in dev / direct-gateway mode.
-
-    ``X-Forwarded-For`` is intentionally NOT used because it is naturally
-    client-controlled at the *first* hop and the trust chain is harder to
-    audit per-request.
-    """
-    peer_host = request.client.host if request.client else None
-
-    trusted = _trusted_proxies()
-    if trusted and peer_host:
-        try:
-            peer_ip = ip_address(peer_host)
-            if any(peer_ip in net for net in trusted):
-                real_ip = request.headers.get("x-real-ip", "").strip()
-                if real_ip:
-                    return real_ip
-        except ValueError:
-            # peer_host wasn't a parseable IP (e.g. "unknown") — fall through
-            pass
-
-    return peer_host or "unknown"
-
-
-def _check_rate_limit(ip: str) -> None:
-    """Raise 429 if the IP is currently locked out."""
-    record = _login_attempts.get(ip)
-    if record is None:
-        return
-    fail_count, lock_until = record
-    if fail_count >= _MAX_LOGIN_ATTEMPTS:
-        if time.time() < lock_until:
-            raise HTTPException(
-                status_code=429,
-                detail="Too many login attempts. Try again later.",
-            )
-        del _login_attempts[ip]
-
-
-_MAX_TRACKED_IPS = 10000
-
-
-def _record_login_failure(ip: str) -> None:
-    """Record a failed login attempt for the given IP."""
-    # Evict expired lockouts when dict grows too large
-    if len(_login_attempts) >= _MAX_TRACKED_IPS:
-        now = time.time()
-        expired = [k for k, (c, t) in _login_attempts.items() if c >= _MAX_LOGIN_ATTEMPTS and now >= t]
-        for k in expired:
-            del _login_attempts[k]
-        # If still too large, evict cheapest-to-lose half: below-threshold
-        # IPs (lock_until=0.0) sort first, then earliest-expiring lockouts.
-        if len(_login_attempts) >= _MAX_TRACKED_IPS:
-            by_time = sorted(_login_attempts.items(), key=lambda kv: kv[1][1])
-            for k, _ in by_time[: len(by_time) // 2]:
-                del _login_attempts[k]
-
-    record = _login_attempts.get(ip)
-    if record is None:
-        _login_attempts[ip] = (1, 0.0)
-    else:
-        new_count = record[0] + 1
-        lock_until = time.time() + _LOCKOUT_SECONDS if new_count >= _MAX_LOGIN_ATTEMPTS else 0.0
-        _login_attempts[ip] = (new_count, lock_until)
-
-
-def _record_login_success(ip: str) -> None:
-    """Clear failure counter for the given IP on successful login."""
-    _login_attempts.pop(ip, None)
-
-
-# ── Endpoints ─────────────────────────────────────────────────────────────
-
-
-@router.post("/login/local", response_model=LoginResponse)
-async def login_local(
-    request: Request,
-    response: Response,
-    form_data: OAuth2PasswordRequestForm = Depends(),
-):
-    """Local email/password login."""
-    client_ip = _get_client_ip(request)
-    _check_rate_limit(client_ip)
-
-    user = await get_local_provider().authenticate({"email": form_data.username, "password": form_data.password})
-
-    if user is None:
-        _record_login_failure(client_ip)
-        raise HTTPException(
-            status_code=status.HTTP_401_UNAUTHORIZED,
-            detail=AuthErrorResponse(code=AuthErrorCode.INVALID_CREDENTIALS, message="Incorrect email or password").model_dump(),
-        )
-
-    _record_login_success(client_ip)
-    token = create_access_token(str(user.id), token_version=user.token_version)
-    _set_session_cookie(response, token, request)
-
-    return LoginResponse(
-        expires_in=get_auth_config().token_expiry_days * 24 * 3600,
-        needs_setup=user.needs_setup,
-    )
-
-
-@router.post("/register", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
-async def register(request: Request, response: Response, body: RegisterRequest):
-    """Register a new user account (always 'user' role).
-
-    Admin is auto-created on first boot. This endpoint creates regular users.
-    Auto-login by setting the session cookie.
-    """
-    try:
-        user = await get_local_provider().create_user(email=body.email, password=body.password, system_role="user")
-    except ValueError:
-        raise HTTPException(
-            status_code=status.HTTP_400_BAD_REQUEST,
-            detail=AuthErrorResponse(code=AuthErrorCode.EMAIL_ALREADY_EXISTS, message="Email already registered").model_dump(),
-        )
-
-    token = create_access_token(str(user.id), token_version=user.token_version)
-    _set_session_cookie(response, token, request)
-
-    return UserResponse(id=str(user.id), email=user.email, system_role=user.system_role)
-
-
-@router.post("/logout", response_model=MessageResponse)
-async def logout(request: Request, response: Response):
-    """Logout current user by clearing the cookie."""
-    response.delete_cookie(key="access_token", secure=is_secure_request(request), samesite="lax")
-    return MessageResponse(message="Successfully logged out")
-
-
-@router.post("/change-password", response_model=MessageResponse)
-async def change_password(request: Request, response: Response, body: ChangePasswordRequest):
-    """Change password for the currently authenticated user.
-
-    Also handles the first-boot setup flow:
-    - If new_email is provided, updates email (checks uniqueness)
-    - If user.needs_setup is True and new_email is given, clears needs_setup
-    - Always increments token_version to invalidate old sessions
-    - Re-issues session cookie with new token_version
-    """
-    from app.gateway.auth.password import hash_password_async, verify_password_async
-
-    user = await get_current_user_from_request(request)
-
-    if user.password_hash is None:
-        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=AuthErrorResponse(code=AuthErrorCode.INVALID_CREDENTIALS, message="OAuth users cannot change password").model_dump())
-
-    if not await verify_password_async(body.current_password, user.password_hash):
-        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=AuthErrorResponse(code=AuthErrorCode.INVALID_CREDENTIALS, message="Current password is incorrect").model_dump())
-
-    provider = get_local_provider()
-
-    # Update email if provided
-    if body.new_email is not None:
-        existing = await provider.get_user_by_email(body.new_email)
-        if existing and str(existing.id) != str(user.id):
-            raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=AuthErrorResponse(code=AuthErrorCode.EMAIL_ALREADY_EXISTS, message="Email already in use").model_dump())
-        user.email = body.new_email
-
-    # Update password + bump version
-    user.password_hash = await hash_password_async(body.new_password)
-    user.token_version += 1
-
-    # Clear setup flag if this is the setup flow
-    if user.needs_setup and body.new_email is not None:
-        user.needs_setup = False
-
-    await provider.update_user(user)
-
-    # Re-issue cookie with new token_version
-    token = create_access_token(str(user.id), token_version=user.token_version)
-    _set_session_cookie(response, token, request)
-
-    return MessageResponse(message="Password changed successfully")
-
-
-@router.get("/me", response_model=UserResponse)
-async def get_me(request: Request):
-    """Get current authenticated user info."""
-    user = await get_current_user_from_request(request)
-    return UserResponse(id=str(user.id), email=user.email, system_role=user.system_role, needs_setup=user.needs_setup)
-
-
-@router.get("/setup-status")
-async def setup_status():
-    """Check if admin account exists. Always False after first boot."""
-    user_count = await get_local_provider().count_users()
-    return {"needs_setup": user_count == 0}
-
-
-# ── OAuth Endpoints (Future/Placeholder) ─────────────────────────────────
-
-
-@router.get("/oauth/{provider}")
-async def oauth_login(provider: str):
-    """Initiate OAuth login flow.
-
-    Redirects to the OAuth provider's authorization URL.
-    Currently a placeholder - requires OAuth provider implementation.
-    """
-    if provider not in ["github", "google"]:
-        raise HTTPException(
-            status_code=status.HTTP_400_BAD_REQUEST,
-            detail=f"Unsupported OAuth provider: {provider}",
-        )
-
-    raise HTTPException(
-        status_code=status.HTTP_501_NOT_IMPLEMENTED,
-        detail="OAuth login not yet implemented",
-    )
-
-
-@router.get("/callback/{provider}")
-async def oauth_callback(provider: str, code: str, state: str):
-    """OAuth callback endpoint.
-
-    Handles the OAuth provider's callback after user authorization.
-    Currently a placeholder.
-    """
-    raise HTTPException(
-        status_code=status.HTTP_501_NOT_IMPLEMENTED,
-        detail="OAuth callback not yet implemented",
-    )

backend/app/gateway/routers/feedback.py

@@ -1,132 +0,0 @@
-"""Feedback endpoints — create, list, stats, delete.
-
-Allows users to submit thumbs-up/down feedback on runs,
-optionally scoped to a specific message.
-"""
-
-from __future__ import annotations
-
-import logging
-from typing import Any
-
-from fastapi import APIRouter, HTTPException, Request
-from pydantic import BaseModel, Field
-
-from app.gateway.authz import require_permission
-from app.gateway.deps import get_current_user, get_feedback_repo, get_run_store
-
-logger = logging.getLogger(__name__)
-router = APIRouter(prefix="/api/threads", tags=["feedback"])
-
-
-# ---------------------------------------------------------------------------
-# Request / response models
-# ---------------------------------------------------------------------------
-
-
-class FeedbackCreateRequest(BaseModel):
-    rating: int = Field(..., description="Feedback rating: +1 (positive) or -1 (negative)")
-    comment: str | None = Field(default=None, description="Optional text feedback")
-    message_id: str | None = Field(default=None, description="Optional: scope feedback to a specific message")
-
-
-class FeedbackResponse(BaseModel):
-    feedback_id: str
-    run_id: str
-    thread_id: str
-    owner_id: str | None = None
-    message_id: str | None = None
-    rating: int
-    comment: str | None = None
-    created_at: str = ""
-
-
-class FeedbackStatsResponse(BaseModel):
-    run_id: str
-    total: int = 0
-    positive: int = 0
-    negative: int = 0
-
-
-# ---------------------------------------------------------------------------
-# Endpoints
-# ---------------------------------------------------------------------------
-
-
-@router.post("/{thread_id}/runs/{run_id}/feedback", response_model=FeedbackResponse)
-@require_permission("threads", "write", owner_check=True, require_existing=True)
-async def create_feedback(
-    thread_id: str,
-    run_id: str,
-    body: FeedbackCreateRequest,
-    request: Request,
-) -> dict[str, Any]:
-    """Submit feedback (thumbs-up/down) for a run."""
-    if body.rating not in (1, -1):
-        raise HTTPException(status_code=400, detail="rating must be +1 or -1")
-
-    user_id = await get_current_user(request)
-
-    # Validate run exists and belongs to thread
-    run_store = get_run_store(request)
-    run = await run_store.get(run_id)
-    if run is None:
-        raise HTTPException(status_code=404, detail=f"Run {run_id} not found")
-    if run.get("thread_id") != thread_id:
-        raise HTTPException(status_code=404, detail=f"Run {run_id} not found in thread {thread_id}")
-
-    feedback_repo = get_feedback_repo(request)
-    return await feedback_repo.create(
-        run_id=run_id,
-        thread_id=thread_id,
-        rating=body.rating,
-        owner_id=user_id,
-        message_id=body.message_id,
-        comment=body.comment,
-    )
-
-
-@router.get("/{thread_id}/runs/{run_id}/feedback", response_model=list[FeedbackResponse])
-@require_permission("threads", "read", owner_check=True)
-async def list_feedback(
-    thread_id: str,
-    run_id: str,
-    request: Request,
-) -> list[dict[str, Any]]:
-    """List all feedback for a run."""
-    feedback_repo = get_feedback_repo(request)
-    return await feedback_repo.list_by_run(thread_id, run_id)
-
-
-@router.get("/{thread_id}/runs/{run_id}/feedback/stats", response_model=FeedbackStatsResponse)
-@require_permission("threads", "read", owner_check=True)
-async def feedback_stats(
-    thread_id: str,
-    run_id: str,
-    request: Request,
-) -> dict[str, Any]:
-    """Get aggregated feedback stats (positive/negative counts) for a run."""
-    feedback_repo = get_feedback_repo(request)
-    return await feedback_repo.aggregate_by_run(thread_id, run_id)
-
-
-@router.delete("/{thread_id}/runs/{run_id}/feedback/{feedback_id}")
-@require_permission("threads", "delete", owner_check=True, require_existing=True)
-async def delete_feedback(
-    thread_id: str,
-    run_id: str,
-    feedback_id: str,
-    request: Request,
-) -> dict[str, bool]:
-    """Delete a feedback record."""
-    feedback_repo = get_feedback_repo(request)
-    # Verify feedback belongs to the specified thread/run before deleting
-    existing = await feedback_repo.get(feedback_id)
-    if existing is None:
-        raise HTTPException(status_code=404, detail=f"Feedback {feedback_id} not found")
-    if existing.get("thread_id") != thread_id or existing.get("run_id") != run_id:
-        raise HTTPException(status_code=404, detail=f"Feedback {feedback_id} not found in run {run_id}")
-    deleted = await feedback_repo.delete(feedback_id)
-    if not deleted:
-        raise HTTPException(status_code=404, detail=f"Feedback {feedback_id} not found")
-    return {"success": True}

backend/app/gateway/routers/mcp.py

@@ -6,7 +6,8 @@ from typing import Literal
 from fastapi import APIRouter, HTTPException
 from pydantic import BaseModel, Field
 
-from deerflow.config.extensions_config import ExtensionsConfig, get_extensions_config, reload_extensions_config
+from deerflow.config.app_config import AppConfig
+from deerflow.config.extensions_config import ExtensionsConfig
 
 logger = logging.getLogger(__name__)
 router = APIRouter(prefix="/api", tags=["mcp"])
@@ -90,9 +91,9 @@ async def get_mcp_configuration() -> McpConfigResponse:
         }
         ```
     """
-    config = get_extensions_config()
+    ext = AppConfig.current().extensions
 
-    return McpConfigResponse(mcp_servers={name: McpServerConfigResponse(**server.model_dump()) for name, server in config.mcp_servers.items()})
+    return McpConfigResponse(mcp_servers={name: McpServerConfigResponse(**server.model_dump()) for name, server in ext.mcp_servers.items()})
 
 
 @router.put(
@@ -143,12 +144,12 @@ async def update_mcp_configuration(request: McpConfigUpdateRequest) -> McpConfig
             logger.info(f"No existing extensions config found. Creating new config at: {config_path}")
 
         # Load current config to preserve skills configuration
-        current_config = get_extensions_config()
+        current_ext = AppConfig.current().extensions
 
         # Convert request to dict format for JSON serialization
         config_data = {
             "mcpServers": {name: server.model_dump() for name, server in request.mcp_servers.items()},
-            "skills": {name: {"enabled": skill.enabled} for name, skill in current_config.skills.items()},
+            "skills": {name: {"enabled": skill.enabled} for name, skill in current_ext.skills.items()},
         }
 
         # Write the configuration to file
@@ -161,8 +162,9 @@ async def update_mcp_configuration(request: McpConfigUpdateRequest) -> McpConfig
         # will detect config file changes via mtime and reinitialize MCP tools automatically
 
         # Reload the configuration and update the global cache
-        reloaded_config = reload_extensions_config()
+        AppConfig.init(AppConfig.from_file())
-        return McpConfigResponse(mcp_servers={name: McpServerConfigResponse(**server.model_dump()) for name, server in reloaded_config.mcp_servers.items()})
+        reloaded_ext = AppConfig.current().extensions
+        return McpConfigResponse(mcp_servers={name: McpServerConfigResponse(**server.model_dump()) for name, server in reloaded_ext.mcp_servers.items()})
 
     except Exception as e:
         logger.error(f"Failed to update MCP configuration: {e}", exc_info=True)

backend/app/gateway/routers/memory.py

@@ -12,7 +12,7 @@ from deerflow.agents.memory.updater import (
     reload_memory_data,
     update_memory_fact,
 )
-from deerflow.config.memory_config import get_memory_config
+from deerflow.config.app_config import AppConfig
 
 router = APIRouter(prefix="/api", tags=["memory"])
 
@@ -311,7 +311,7 @@ async def get_memory_config_endpoint() -> MemoryConfigResponse:
         }
         ```
     """
-    config = get_memory_config()
+    config = AppConfig.current().memory
     return MemoryConfigResponse(
         enabled=config.enabled,
         storage_path=config.storage_path,
@@ -336,7 +336,7 @@ async def get_memory_status() -> MemoryStatusResponse:
     Returns:
         Combined memory configuration and current data.
     """
-    config = get_memory_config()
+    config = AppConfig.current().memory
     memory_data = get_memory_data()
 
     return MemoryStatusResponse(

backend/app/gateway/routers/models.py

@@ -1,7 +1,7 @@
 from fastapi import APIRouter, HTTPException
 from pydantic import BaseModel, Field
 
-from deerflow.config import get_app_config
+from deerflow.config.app_config import AppConfig
 
 router = APIRouter(prefix="/api", tags=["models"])
 
@@ -58,7 +58,7 @@ async def list_models() -> ModelsListResponse:
         }
         ```
     """
-    config = get_app_config()
+    config = AppConfig.current()
     models = [
         ModelResponse(
             name=model.name,
@@ -101,7 +101,7 @@ async def get_model(model_name: str) -> ModelResponse:
         }
         ```
     """
-    config = get_app_config()
+    config = AppConfig.current()
     model = config.get_model_config(model_name)
     if model is None:
         raise HTTPException(status_code=404, detail=f"Model '{model_name}' not found")

backend/app/gateway/routers/skills.py

@@ -8,7 +8,8 @@ from pydantic import BaseModel, Field
 
 from app.gateway.path_utils import resolve_thread_virtual_path
 from deerflow.agents.lead_agent.prompt import refresh_skills_system_prompt_cache_async
-from deerflow.config.extensions_config import ExtensionsConfig, SkillStateConfig, get_extensions_config, reload_extensions_config
+from deerflow.config.app_config import AppConfig
+from deerflow.config.extensions_config import ExtensionsConfig, SkillStateConfig
 from deerflow.skills import Skill, load_skills
 from deerflow.skills.installer import SkillAlreadyExistsError, install_skill_from_archive
 from deerflow.skills.manager import (
@@ -325,19 +326,19 @@ async def update_skill(skill_name: str, request: SkillUpdateRequest) -> SkillRes
             config_path = Path.cwd().parent / "extensions_config.json"
             logger.info(f"No existing extensions config found. Creating new config at: {config_path}")
 
-        extensions_config = get_extensions_config()
+        ext = AppConfig.current().extensions
-        extensions_config.skills[skill_name] = SkillStateConfig(enabled=request.enabled)
+        ext.skills[skill_name] = SkillStateConfig(enabled=request.enabled)
 
         config_data = {
-            "mcpServers": {name: server.model_dump() for name, server in extensions_config.mcp_servers.items()},
+            "mcpServers": {name: server.model_dump() for name, server in ext.mcp_servers.items()},
-            "skills": {name: {"enabled": skill_config.enabled} for name, skill_config in extensions_config.skills.items()},
+            "skills": {name: {"enabled": skill_config.enabled} for name, skill_config in ext.skills.items()},
         }
 
         with open(config_path, "w", encoding="utf-8") as f:
             json.dump(config_data, f, indent=2)
 
         logger.info(f"Skills configuration updated and saved to: {config_path}")
-        reload_extensions_config()
+        AppConfig.init(AppConfig.from_file())
         await refresh_skills_system_prompt_cache_async()
 
         skills = load_skills(enabled_only=False)

backend/app/gateway/routers/suggestions.py

@@ -1,11 +1,10 @@
 import json
 import logging
 
-from fastapi import APIRouter, Request
+from fastapi import APIRouter
 from langchain_core.messages import HumanMessage, SystemMessage
 from pydantic import BaseModel, Field
 
-from app.gateway.authz import require_permission
 from deerflow.models import create_chat_model
 
 logger = logging.getLogger(__name__)
@@ -99,13 +98,12 @@ def _format_conversation(messages: list[SuggestionMessage]) -> str:
     summary="Generate Follow-up Questions",
     description="Generate short follow-up questions a user might ask next, based on recent conversation context.",
 )
-@require_permission("threads", "read", owner_check=True)
+async def generate_suggestions(thread_id: str, request: SuggestionsRequest) -> SuggestionsResponse:
-async def generate_suggestions(thread_id: str, body: SuggestionsRequest, request: Request) -> SuggestionsResponse:
+    if not request.messages:
-    if not body.messages:
         return SuggestionsResponse(suggestions=[])
 
-    n = body.n
+    n = request.n
-    conversation = _format_conversation(body.messages)
+    conversation = _format_conversation(request.messages)
     if not conversation:
         return SuggestionsResponse(suggestions=[])
 
@@ -122,7 +120,7 @@ async def generate_suggestions(thread_id: str, body: SuggestionsRequest, request
     user_content = f"Conversation Context:\n{conversation}\n\nGenerate {n} follow-up questions"
 
     try:
-        model = create_chat_model(name=body.model_name, thinking_enabled=False)
+        model = create_chat_model(name=request.model_name, thinking_enabled=False)
         response = await model.ainvoke([SystemMessage(content=system_instruction), HumanMessage(content=user_content)])
         raw = _extract_response_text(response.content)
         suggestions = _parse_json_string_list(raw) or []

backend/app/gateway/routers/thread_runs.py

@@ -19,8 +19,7 @@ from fastapi import APIRouter, HTTPException, Query, Request
 from fastapi.responses import Response, StreamingResponse
 from pydantic import BaseModel, Field
 
-from app.gateway.authz import require_permission
+from app.gateway.deps import get_checkpointer, get_run_manager, get_stream_bridge
-from app.gateway.deps import get_checkpointer, get_run_event_store, get_run_manager, get_run_store, get_stream_bridge
 from app.gateway.services import sse_consumer, start_run
 from deerflow.runtime import RunRecord, serialize_channel_values
 
@@ -54,7 +53,6 @@ class RunCreateRequest(BaseModel):
     after_seconds: float | None = Field(default=None, description="Delayed execution")
     if_not_exists: Literal["reject", "create"] = Field(default="create", description="Thread creation policy")
     feedback_keys: list[str] | None = Field(default=None, description="LangSmith feedback keys")
-    follow_up_to_run_id: str | None = Field(default=None, description="Run ID this message follows up on. Auto-detected from latest successful run if not provided.")
 
 
 class RunResponse(BaseModel):
@@ -94,7 +92,6 @@ def _record_to_response(record: RunRecord) -> RunResponse:
 
 
 @router.post("/{thread_id}/runs", response_model=RunResponse)
-@require_permission("runs", "create", owner_check=True, require_existing=True)
 async def create_run(thread_id: str, body: RunCreateRequest, request: Request) -> RunResponse:
     """Create a background run (returns immediately)."""
     record = await start_run(body, thread_id, request)
@@ -102,7 +99,6 @@ async def create_run(thread_id: str, body: RunCreateRequest, request: Request) -
 
 
 @router.post("/{thread_id}/runs/stream")
-@require_permission("runs", "create", owner_check=True, require_existing=True)
 async def stream_run(thread_id: str, body: RunCreateRequest, request: Request) -> StreamingResponse:
     """Create a run and stream events via SSE.
 
@@ -130,7 +126,6 @@ async def stream_run(thread_id: str, body: RunCreateRequest, request: Request) -
 
 
 @router.post("/{thread_id}/runs/wait", response_model=dict)
-@require_permission("runs", "create", owner_check=True, require_existing=True)
 async def wait_run(thread_id: str, body: RunCreateRequest, request: Request) -> dict:
     """Create a run and block until it completes, returning the final state."""
     record = await start_run(body, thread_id, request)
@@ -156,7 +151,6 @@ async def wait_run(thread_id: str, body: RunCreateRequest, request: Request) ->
 
 
 @router.get("/{thread_id}/runs", response_model=list[RunResponse])
-@require_permission("runs", "read", owner_check=True)
 async def list_runs(thread_id: str, request: Request) -> list[RunResponse]:
     """List all runs for a thread."""
     run_mgr = get_run_manager(request)
@@ -165,7 +159,6 @@ async def list_runs(thread_id: str, request: Request) -> list[RunResponse]:
 
 
 @router.get("/{thread_id}/runs/{run_id}", response_model=RunResponse)
-@require_permission("runs", "read", owner_check=True)
 async def get_run(thread_id: str, run_id: str, request: Request) -> RunResponse:
     """Get details of a specific run."""
     run_mgr = get_run_manager(request)
@@ -176,7 +169,6 @@ async def get_run(thread_id: str, run_id: str, request: Request) -> RunResponse:
 
 
 @router.post("/{thread_id}/runs/{run_id}/cancel")
-@require_permission("runs", "cancel", owner_check=True, require_existing=True)
 async def cancel_run(
     thread_id: str,
     run_id: str,
@@ -214,7 +206,6 @@ async def cancel_run(
 
 
 @router.get("/{thread_id}/runs/{run_id}/join")
-@require_permission("runs", "read", owner_check=True)
 async def join_run(thread_id: str, run_id: str, request: Request) -> StreamingResponse:
     """Join an existing run's SSE stream."""
     bridge = get_stream_bridge(request)
@@ -235,7 +226,6 @@ async def join_run(thread_id: str, run_id: str, request: Request) -> StreamingRe
 
 
 @router.api_route("/{thread_id}/runs/{run_id}/stream", methods=["GET", "POST"], response_model=None)
-@require_permission("runs", "read", owner_check=True)
 async def stream_existing_run(
     thread_id: str,
     run_id: str,
@@ -275,54 +265,3 @@ async def stream_existing_run(
             "X-Accel-Buffering": "no",
         },
     )
-
-
-# ---------------------------------------------------------------------------
-# Messages / Events / Token usage endpoints
-# ---------------------------------------------------------------------------
-
-
-@router.get("/{thread_id}/messages")
-@require_permission("runs", "read", owner_check=True)
-async def list_thread_messages(
-    thread_id: str,
-    request: Request,
-    limit: int = Query(default=50, le=200),
-    before_seq: int | None = Query(default=None),
-    after_seq: int | None = Query(default=None),
-) -> list[dict]:
-    """Return displayable messages for a thread (across all runs)."""
-    event_store = get_run_event_store(request)
-    return await event_store.list_messages(thread_id, limit=limit, before_seq=before_seq, after_seq=after_seq)
-
-
-@router.get("/{thread_id}/runs/{run_id}/messages")
-@require_permission("runs", "read", owner_check=True)
-async def list_run_messages(thread_id: str, run_id: str, request: Request) -> list[dict]:
-    """Return displayable messages for a specific run."""
-    event_store = get_run_event_store(request)
-    return await event_store.list_messages_by_run(thread_id, run_id)
-
-
-@router.get("/{thread_id}/runs/{run_id}/events")
-@require_permission("runs", "read", owner_check=True)
-async def list_run_events(
-    thread_id: str,
-    run_id: str,
-    request: Request,
-    event_types: str | None = Query(default=None),
-    limit: int = Query(default=500, le=2000),
-) -> list[dict]:
-    """Return the full event stream for a run (debug/audit)."""
-    event_store = get_run_event_store(request)
-    types = event_types.split(",") if event_types else None
-    return await event_store.list_events(thread_id, run_id, event_types=types, limit=limit)
-
-
-@router.get("/{thread_id}/token-usage")
-@require_permission("threads", "read", owner_check=True)
-async def thread_token_usage(thread_id: str, request: Request) -> dict:
-    """Thread-level token usage aggregation."""
-    run_store = get_run_store(request)
-    agg = await run_store.aggregate_tokens_by_thread(thread_id)
-    return {"thread_id": thread_id, **agg}

backend/app/gateway/routers/threads.py

@@ -18,34 +18,23 @@ import uuid
 from typing import Any
 
 from fastapi import APIRouter, HTTPException, Request
-from pydantic import BaseModel, Field, field_validator
+from pydantic import BaseModel, Field
 
-from app.gateway.authz import require_permission
+from app.gateway.deps import get_checkpointer, get_store
-from app.gateway.deps import get_checkpointer
-from app.gateway.utils import sanitize_log_param
 from deerflow.config.paths import Paths, get_paths
 from deerflow.runtime import serialize_channel_values
 
+# ---------------------------------------------------------------------------
+# Store namespace
+# ---------------------------------------------------------------------------
+
+THREADS_NS: tuple[str, ...] = ("threads",)
+"""Namespace used by the Store for thread metadata records."""
+
 logger = logging.getLogger(__name__)
 router = APIRouter(prefix="/api/threads", tags=["threads"])
 
 
-# Metadata keys that the server controls; clients are not allowed to set
-# them. Pydantic ``@field_validator("metadata")`` strips them on every
-# inbound model below so a malicious client cannot reflect a forged
-# owner identity through the API surface. Defense-in-depth — the
-# row-level invariant is still ``threads_meta.owner_id`` populated from
-# the auth contextvar; this list closes the metadata-blob echo gap.
-_SERVER_RESERVED_METADATA_KEYS: frozenset[str] = frozenset({"owner_id", "user_id"})
-
-
-def _strip_reserved_metadata(metadata: dict[str, Any] | None) -> dict[str, Any]:
-    """Return ``metadata`` with server-controlled keys removed."""
-    if not metadata:
-        return metadata or {}
-    return {k: v for k, v in metadata.items() if k not in _SERVER_RESERVED_METADATA_KEYS}
-
-
 # ---------------------------------------------------------------------------
 # Response / request models
 # ---------------------------------------------------------------------------
@@ -74,11 +63,8 @@ class ThreadCreateRequest(BaseModel):
     """Request body for creating a thread."""
 
     thread_id: str | None = Field(default=None, description="Optional thread ID (auto-generated if omitted)")
-    assistant_id: str | None = Field(default=None, description="Associate thread with an assistant")
     metadata: dict[str, Any] = Field(default_factory=dict, description="Initial metadata")
 
-    _strip_reserved = field_validator("metadata")(classmethod(lambda cls, v: _strip_reserved_metadata(v)))
-
 
 class ThreadSearchRequest(BaseModel):
     """Request body for searching threads."""
@@ -107,8 +93,6 @@ class ThreadPatchRequest(BaseModel):
 
     metadata: dict[str, Any] = Field(default_factory=dict, description="Metadata to merge")
 
-    _strip_reserved = field_validator("metadata")(classmethod(lambda cls, v: _strip_reserved_metadata(v)))
-
 
 class ThreadStateUpdateRequest(BaseModel):
     """Request body for updating thread state (human-in-the-loop resume)."""
@@ -151,16 +135,61 @@ def _delete_thread_data(thread_id: str, paths: Paths | None = None) -> ThreadDel
         raise HTTPException(status_code=422, detail=str(exc)) from exc
     except FileNotFoundError:
         # Not critical — thread data may not exist on disk
-        logger.debug("No local thread data to delete for %s", sanitize_log_param(thread_id))
+        logger.debug("No local thread data to delete for %s", thread_id)
         return ThreadDeleteResponse(success=True, message=f"No local data for {thread_id}")
     except Exception as exc:
-        logger.exception("Failed to delete thread data for %s", sanitize_log_param(thread_id))
+        logger.exception("Failed to delete thread data for %s", thread_id)
         raise HTTPException(status_code=500, detail="Failed to delete local thread data.") from exc
 
-    logger.info("Deleted local thread data for %s", sanitize_log_param(thread_id))
+    logger.info("Deleted local thread data for %s", thread_id)
     return ThreadDeleteResponse(success=True, message=f"Deleted local thread data for {thread_id}")
 
 
+async def _store_get(store, thread_id: str) -> dict | None:
+    """Fetch a thread record from the Store; returns ``None`` if absent."""
+    item = await store.aget(THREADS_NS, thread_id)
+    return item.value if item is not None else None
+
+
+async def _store_put(store, record: dict) -> None:
+    """Write a thread record to the Store."""
+    await store.aput(THREADS_NS, record["thread_id"], record)
+
+
+async def _store_upsert(store, thread_id: str, *, metadata: dict | None = None, values: dict | None = None) -> None:
+    """Create or refresh a thread record in the Store.
+
+    On creation the record is written with ``status="idle"``.  On update only
+    ``updated_at`` (and optionally ``metadata`` / ``values``) are changed so
+    that existing fields are preserved.
+
+    ``values`` carries the agent-state snapshot exposed to the frontend
+    (currently just ``{"title": "..."}``).
+    """
+    now = time.time()
+    existing = await _store_get(store, thread_id)
+    if existing is None:
+        await _store_put(
+            store,
+            {
+                "thread_id": thread_id,
+                "status": "idle",
+                "created_at": now,
+                "updated_at": now,
+                "metadata": metadata or {},
+                "values": values or {},
+            },
+        )
+    else:
+        val = dict(existing)
+        val["updated_at"] = now
+        if metadata:
+            val.setdefault("metadata", {}).update(metadata)
+        if values:
+            val.setdefault("values", {}).update(values)
+        await _store_put(store, val)
+
+
 def _derive_thread_status(checkpoint_tuple) -> str:
     """Derive thread status from checkpoint metadata."""
     if checkpoint_tuple is None:
@@ -186,19 +215,23 @@ def _derive_thread_status(checkpoint_tuple) -> str:
 
 
 @router.delete("/{thread_id}", response_model=ThreadDeleteResponse)
-@require_permission("threads", "delete", owner_check=True, require_existing=True)
 async def delete_thread_data(thread_id: str, request: Request) -> ThreadDeleteResponse:
     """Delete local persisted filesystem data for a thread.
 
     Cleans DeerFlow-managed thread directories, removes checkpoint data,
-    and removes the thread_meta row from the configured ThreadMetaStore
+    and removes the thread record from the Store.
-    (sqlite or memory).
     """
-    from app.gateway.deps import get_thread_meta_repo
-
     # Clean local filesystem
     response = _delete_thread_data(thread_id)
 
+    # Remove from Store (best-effort)
+    store = get_store(request)
+    if store is not None:
+        try:
+            await store.adelete(THREADS_NS, thread_id)
+        except Exception:
+            logger.debug("Could not delete store record for thread %s (not critical)", thread_id)
+
     # Remove checkpoints (best-effort)
     checkpointer = getattr(request.app.state, "checkpointer", None)
     if checkpointer is not None:
@@ -206,15 +239,7 @@ async def delete_thread_data(thread_id: str, request: Request) -> ThreadDeleteRe
             if hasattr(checkpointer, "adelete_thread"):
                 await checkpointer.adelete_thread(thread_id)
         except Exception:
-            logger.debug("Could not delete checkpoints for thread %s (not critical)", sanitize_log_param(thread_id))
+            logger.debug("Could not delete checkpoints for thread %s (not critical)", thread_id)
-
-    # Remove thread_meta row (best-effort) — required for sqlite backend
-    # so the deleted thread no longer appears in /threads/search.
-    try:
-        thread_meta_repo = get_thread_meta_repo(request)
-        await thread_meta_repo.delete(thread_id)
-    except Exception:
-        logger.debug("Could not delete thread_meta for %s (not critical)", sanitize_log_param(thread_id))
 
     return response
 
@@ -223,40 +248,43 @@ async def delete_thread_data(thread_id: str, request: Request) -> ThreadDeleteRe
 async def create_thread(body: ThreadCreateRequest, request: Request) -> ThreadResponse:
     """Create a new thread.
 
-    Writes a thread_meta record (so the thread appears in /threads/search)
+    The thread record is written to the Store (for fast listing) and an
-    and an empty checkpoint (so state endpoints work immediately).
+    empty checkpoint is written to the checkpointer (for state reads).
     Idempotent: returns the existing record when ``thread_id`` already exists.
     """
-    from app.gateway.deps import get_thread_meta_repo
+    store = get_store(request)
-
     checkpointer = get_checkpointer(request)
-    thread_meta_repo = get_thread_meta_repo(request)
     thread_id = body.thread_id or str(uuid.uuid4())
     now = time.time()
-    # ``body.metadata`` is already stripped of server-reserved keys by
-    # ``ThreadCreateRequest._strip_reserved`` — see the model definition.
 
-    # Idempotency: return existing record when already present
+    # Idempotency: return existing record from Store when already present
-    existing_record = await thread_meta_repo.get(thread_id)
+    if store is not None:
-    if existing_record is not None:
+        existing_record = await _store_get(store, thread_id)
-        return ThreadResponse(
+        if existing_record is not None:
-            thread_id=thread_id,
+            return ThreadResponse(
-            status=existing_record.get("status", "idle"),
+                thread_id=thread_id,
-            created_at=str(existing_record.get("created_at", "")),
+                status=existing_record.get("status", "idle"),
-            updated_at=str(existing_record.get("updated_at", "")),
+                created_at=str(existing_record.get("created_at", "")),
-            metadata=existing_record.get("metadata", {}),
+                updated_at=str(existing_record.get("updated_at", "")),
-        )
+                metadata=existing_record.get("metadata", {}),
+            )
 
-    # Write thread_meta so the thread appears in /threads/search immediately
+    # Write thread record to Store
-    try:
+    if store is not None:
-        await thread_meta_repo.create(
+        try:
-            thread_id,
+            await _store_put(
-            assistant_id=getattr(body, "assistant_id", None),
+                store,
-            metadata=body.metadata,
+                {
-        )
+                    "thread_id": thread_id,
-    except Exception:
+                    "status": "idle",
-        logger.exception("Failed to write thread_meta for %s", sanitize_log_param(thread_id))
+                    "created_at": now,
-        raise HTTPException(status_code=500, detail="Failed to create thread")
+                    "updated_at": now,
+                    "metadata": body.metadata,
+                },
+            )
+        except Exception:
+            logger.exception("Failed to write thread %s to store", thread_id)
+            raise HTTPException(status_code=500, detail="Failed to create thread")
 
     # Write an empty checkpoint so state endpoints work immediately
     config = {"configurable": {"thread_id": thread_id, "checkpoint_ns": ""}}
@@ -273,10 +301,10 @@ async def create_thread(body: ThreadCreateRequest, request: Request) -> ThreadRe
         }
         await checkpointer.aput(config, empty_checkpoint(), ckpt_metadata, {})
     except Exception:
-        logger.exception("Failed to create checkpoint for thread %s", sanitize_log_param(thread_id))
+        logger.exception("Failed to create checkpoint for thread %s", thread_id)
         raise HTTPException(status_code=500, detail="Failed to create thread")
 
-    logger.info("Thread created: %s", sanitize_log_param(thread_id))
+    logger.info("Thread created: %s", thread_id)
     return ThreadResponse(
         thread_id=thread_id,
         status="idle",
@@ -290,91 +318,166 @@ async def create_thread(body: ThreadCreateRequest, request: Request) -> ThreadRe
 async def search_threads(body: ThreadSearchRequest, request: Request) -> list[ThreadResponse]:
     """Search and list threads.
 
-    Delegates to the configured ThreadMetaStore implementation
+    Two-phase approach:
-    (SQL-backed for sqlite/postgres, Store-backed for memory mode).
-    """
-    from app.gateway.deps import get_thread_meta_repo
 
-    repo = get_thread_meta_repo(request)
+    **Phase 1 — Store (fast path, O(threads))**: returns threads that were
-    rows = await repo.search(
+    created or run through this Gateway.  Store records are tiny metadata
-        metadata=body.metadata or None,
+    dicts so fetching all of them at once is cheap.
-        status=body.status,
+
-        limit=body.limit,
+    **Phase 2 — Checkpointer supplement (lazy migration)**: threads that
-        offset=body.offset,
+    were created directly by LangGraph Server (and therefore absent from the
-    )
+    Store) are discovered here by iterating the shared checkpointer.  Any
-    return [
+    newly found thread is immediately written to the Store so that the next
-        ThreadResponse(
+    search skips Phase 2 for that thread — the Store converges to a full
-            thread_id=r["thread_id"],
+    index over time without a one-shot migration job.
-            status=r.get("status", "idle"),
+    """
-            created_at=r.get("created_at", ""),
+    store = get_store(request)
-            updated_at=r.get("updated_at", ""),
+    checkpointer = get_checkpointer(request)
-            metadata=r.get("metadata", {}),
+
-            values={"title": r["display_name"]} if r.get("display_name") else {},
+    # -----------------------------------------------------------------------
-            interrupts={},
+    # Phase 1: Store
-        )
+    # -----------------------------------------------------------------------
-        for r in rows
+    merged: dict[str, ThreadResponse] = {}
-    ]
+
+    if store is not None:
+        try:
+            items = await store.asearch(THREADS_NS, limit=10_000)
+        except Exception:
+            logger.warning("Store search failed — falling back to checkpointer only", exc_info=True)
+            items = []
+
+        for item in items:
+            val = item.value
+            merged[val["thread_id"]] = ThreadResponse(
+                thread_id=val["thread_id"],
+                status=val.get("status", "idle"),
+                created_at=str(val.get("created_at", "")),
+                updated_at=str(val.get("updated_at", "")),
+                metadata=val.get("metadata", {}),
+                values=val.get("values", {}),
+            )
+
+    # -----------------------------------------------------------------------
+    # Phase 2: Checkpointer supplement
+    # Discovers threads not yet in the Store (e.g. created by LangGraph
+    # Server) and lazily migrates them so future searches skip this phase.
+    # -----------------------------------------------------------------------
+    try:
+        async for checkpoint_tuple in checkpointer.alist(None):
+            cfg = getattr(checkpoint_tuple, "config", {})
+            thread_id = cfg.get("configurable", {}).get("thread_id")
+            if not thread_id or thread_id in merged:
+                continue
+
+            # Skip sub-graph checkpoints (checkpoint_ns is non-empty for those)
+            if cfg.get("configurable", {}).get("checkpoint_ns", ""):
+                continue
+
+            ckpt_meta = getattr(checkpoint_tuple, "metadata", {}) or {}
+            # Strip LangGraph internal keys from the user-visible metadata dict
+            user_meta = {k: v for k, v in ckpt_meta.items() if k not in ("created_at", "updated_at", "step", "source", "writes", "parents")}
+
+            # Extract state values (title) from the checkpoint's channel_values
+            checkpoint_data = getattr(checkpoint_tuple, "checkpoint", {}) or {}
+            channel_values = checkpoint_data.get("channel_values", {})
+            ckpt_values = {}
+            if title := channel_values.get("title"):
+                ckpt_values["title"] = title
+
+            thread_resp = ThreadResponse(
+                thread_id=thread_id,
+                status=_derive_thread_status(checkpoint_tuple),
+                created_at=str(ckpt_meta.get("created_at", "")),
+                updated_at=str(ckpt_meta.get("updated_at", ckpt_meta.get("created_at", ""))),
+                metadata=user_meta,
+                values=ckpt_values,
+            )
+            merged[thread_id] = thread_resp
+
+            # Lazy migration — write to Store so the next search finds it there
+            if store is not None:
+                try:
+                    await _store_upsert(store, thread_id, metadata=user_meta, values=ckpt_values or None)
+                except Exception:
+                    logger.debug("Failed to migrate thread %s to store (non-fatal)", thread_id)
+    except Exception:
+        logger.exception("Checkpointer scan failed during thread search")
+        # Don't raise — return whatever was collected from Store + partial scan
+
+    # -----------------------------------------------------------------------
+    # Phase 3: Filter → sort → paginate
+    # -----------------------------------------------------------------------
+    results = list(merged.values())
+
+    if body.metadata:
+        results = [r for r in results if all(r.metadata.get(k) == v for k, v in body.metadata.items())]
+
+    if body.status:
+        results = [r for r in results if r.status == body.status]
+
+    results.sort(key=lambda r: r.updated_at, reverse=True)
+    return results[body.offset : body.offset + body.limit]
 
 
 @router.patch("/{thread_id}", response_model=ThreadResponse)
-@require_permission("threads", "write", owner_check=True, require_existing=True)
 async def patch_thread(thread_id: str, body: ThreadPatchRequest, request: Request) -> ThreadResponse:
     """Merge metadata into a thread record."""
-    from app.gateway.deps import get_thread_meta_repo
+    store = get_store(request)
+    if store is None:
+        raise HTTPException(status_code=503, detail="Store not available")
 
-    thread_meta_repo = get_thread_meta_repo(request)
+    record = await _store_get(store, thread_id)
-    record = await thread_meta_repo.get(thread_id)
     if record is None:
         raise HTTPException(status_code=404, detail=f"Thread {thread_id} not found")
 
-    # ``body.metadata`` already stripped by ``ThreadPatchRequest._strip_reserved``.
+    now = time.time()
+    updated = dict(record)
+    updated.setdefault("metadata", {}).update(body.metadata)
+    updated["updated_at"] = now
+
     try:
-        await thread_meta_repo.update_metadata(thread_id, body.metadata)
+        await _store_put(store, updated)
     except Exception:
-        logger.exception("Failed to patch thread %s", sanitize_log_param(thread_id))
+        logger.exception("Failed to patch thread %s", thread_id)
         raise HTTPException(status_code=500, detail="Failed to update thread")
 
-    # Re-read to get the merged metadata + refreshed updated_at
-    record = await thread_meta_repo.get(thread_id) or record
     return ThreadResponse(
         thread_id=thread_id,
-        status=record.get("status", "idle"),
+        status=updated.get("status", "idle"),
-        created_at=str(record.get("created_at", "")),
+        created_at=str(updated.get("created_at", "")),
-        updated_at=str(record.get("updated_at", "")),
+        updated_at=str(now),
-        metadata=record.get("metadata", {}),
+        metadata=updated.get("metadata", {}),
     )
 
 
 @router.get("/{thread_id}", response_model=ThreadResponse)
-@require_permission("threads", "read", owner_check=True)
 async def get_thread(thread_id: str, request: Request) -> ThreadResponse:
     """Get thread info.
 
-    Reads metadata from the ThreadMetaStore and derives the accurate
+    Reads metadata from the Store and derives the accurate execution
-    execution status from the checkpointer.  Falls back to the checkpointer
+    status from the checkpointer.  Falls back to the checkpointer alone
-    alone for threads that pre-date ThreadMetaStore adoption (backward compat).
+    for threads that pre-date Store adoption (backward compat).
     """
-    from app.gateway.deps import get_thread_meta_repo
+    store = get_store(request)
-
-    thread_meta_repo = get_thread_meta_repo(request)
     checkpointer = get_checkpointer(request)
 
-    record: dict | None = await thread_meta_repo.get(thread_id)
+    record: dict | None = None
+    if store is not None:
+        record = await _store_get(store, thread_id)
 
     # Derive accurate status from the checkpointer
     config = {"configurable": {"thread_id": thread_id, "checkpoint_ns": ""}}
     try:
         checkpoint_tuple = await checkpointer.aget_tuple(config)
     except Exception:
-        logger.exception("Failed to get checkpoint for thread %s", sanitize_log_param(thread_id))
+        logger.exception("Failed to get checkpoint for thread %s", thread_id)
         raise HTTPException(status_code=500, detail="Failed to get thread")
 
     if record is None and checkpoint_tuple is None:
         raise HTTPException(status_code=404, detail=f"Thread {thread_id} not found")
 
-    # If the thread exists in the checkpointer but not in thread_meta (e.g.
+    # If the thread exists in the checkpointer but not the store (e.g. legacy
-    # legacy data created before thread_meta adoption), synthesize a minimal
+    # data), synthesize a minimal store record from the checkpoint metadata.
-    # record from the checkpoint metadata.
     if record is None and checkpoint_tuple is not None:
         ckpt_meta = getattr(checkpoint_tuple, "metadata", {}) or {}
         record = {
@@ -403,7 +506,6 @@ async def get_thread(thread_id: str, request: Request) -> ThreadResponse:
 
 
 @router.get("/{thread_id}/state", response_model=ThreadStateResponse)
-@require_permission("threads", "read", owner_check=True)
 async def get_thread_state(thread_id: str, request: Request) -> ThreadStateResponse:
     """Get the latest state snapshot for a thread.
 
@@ -416,7 +518,7 @@ async def get_thread_state(thread_id: str, request: Request) -> ThreadStateRespo
     try:
         checkpoint_tuple = await checkpointer.aget_tuple(config)
     except Exception:
-        logger.exception("Failed to get state for thread %s", sanitize_log_param(thread_id))
+        logger.exception("Failed to get state for thread %s", thread_id)
         raise HTTPException(status_code=500, detail="Failed to get thread state")
 
     if checkpoint_tuple is None:
@@ -453,19 +555,15 @@ async def get_thread_state(thread_id: str, request: Request) -> ThreadStateRespo
 
 
 @router.post("/{thread_id}/state", response_model=ThreadStateResponse)
-@require_permission("threads", "write", owner_check=True, require_existing=True)
 async def update_thread_state(thread_id: str, body: ThreadStateUpdateRequest, request: Request) -> ThreadStateResponse:
     """Update thread state (e.g. for human-in-the-loop resume or title rename).
 
     Writes a new checkpoint that merges *body.values* into the latest
-    channel values, then syncs any updated ``title`` field through the
+    channel values, then syncs any updated ``title`` field back to the Store
-    ThreadMetaStore abstraction so that ``/threads/search`` reflects the
+    so that ``/threads/search`` reflects the change immediately.
-    change immediately in both sqlite and memory backends.
     """
-    from app.gateway.deps import get_thread_meta_repo
-
     checkpointer = get_checkpointer(request)
-    thread_meta_repo = get_thread_meta_repo(request)
+    store = get_store(request)
 
     # checkpoint_ns must be present in the config for aput — default to ""
     # (the root graph namespace).  checkpoint_id is optional; omitting it
@@ -482,7 +580,7 @@ async def update_thread_state(thread_id: str, body: ThreadStateUpdateRequest, re
     try:
         checkpoint_tuple = await checkpointer.aget_tuple(read_config)
     except Exception:
-        logger.exception("Failed to get state for thread %s", sanitize_log_param(thread_id))
+        logger.exception("Failed to get state for thread %s", thread_id)
         raise HTTPException(status_code=500, detail="Failed to get thread state")
 
     if checkpoint_tuple is None:
@@ -516,22 +614,19 @@ async def update_thread_state(thread_id: str, body: ThreadStateUpdateRequest, re
     try:
         new_config = await checkpointer.aput(write_config, checkpoint, metadata, {})
     except Exception:
-        logger.exception("Failed to update state for thread %s", sanitize_log_param(thread_id))
+        logger.exception("Failed to update state for thread %s", thread_id)
         raise HTTPException(status_code=500, detail="Failed to update thread state")
 
     new_checkpoint_id: str | None = None
     if isinstance(new_config, dict):
         new_checkpoint_id = new_config.get("configurable", {}).get("checkpoint_id")
 
-    # Sync title changes through the ThreadMetaStore abstraction so /threads/search
+    # Sync title changes to the Store so /threads/search reflects them immediately.
-    # reflects them immediately in both sqlite and memory backends.
+    if store is not None and body.values and "title" in body.values:
-    if body.values and "title" in body.values:
+        try:
-        new_title = body.values["title"]
+            await _store_upsert(store, thread_id, values={"title": body.values["title"]})
-        if new_title:  # Skip empty strings and None
+        except Exception:
-            try:
+            logger.debug("Failed to sync title to store for thread %s (non-fatal)", thread_id)
-                await thread_meta_repo.update_display_name(thread_id, new_title)
-            except Exception:
-                logger.debug("Failed to sync title to thread_meta for %s (non-fatal)", sanitize_log_param(thread_id))
 
     return ThreadStateResponse(
         values=serialize_channel_values(channel_values),
@@ -543,16 +638,8 @@ async def update_thread_state(thread_id: str, body: ThreadStateUpdateRequest, re
 
 
 @router.post("/{thread_id}/history", response_model=list[HistoryEntry])
-@require_permission("threads", "read", owner_check=True)
 async def get_thread_history(thread_id: str, body: ThreadHistoryRequest, request: Request) -> list[HistoryEntry]:
-    """Get checkpoint history for a thread.
+    """Get checkpoint history for a thread."""
-
-    Messages are read from the checkpointer's channel values (the
-    authoritative source) and serialized via
-    :func:`~deerflow.runtime.serialization.serialize_channel_values`.
-    Only the latest (first) checkpoint carries the ``messages`` key to
-    avoid duplicating them across every entry.
-    """
     checkpointer = get_checkpointer(request)
 
     config: dict[str, Any] = {"configurable": {"thread_id": thread_id}}
@@ -560,7 +647,6 @@ async def get_thread_history(thread_id: str, body: ThreadHistoryRequest, request
         config["configurable"]["checkpoint_id"] = body.before
 
     entries: list[HistoryEntry] = []
-    is_latest_checkpoint = True
     try:
         async for checkpoint_tuple in checkpointer.alist(config, limit=body.limit):
             ckpt_config = getattr(checkpoint_tuple, "config", {})
@@ -575,42 +661,22 @@ async def get_thread_history(thread_id: str, body: ThreadHistoryRequest, request
 
             channel_values = checkpoint.get("channel_values", {})
 
-            # Build values from checkpoint channel_values
-            values: dict[str, Any] = {}
-            if title := channel_values.get("title"):
-                values["title"] = title
-            if thread_data := channel_values.get("thread_data"):
-                values["thread_data"] = thread_data
-
-            # Attach messages from checkpointer only for the latest checkpoint
-            if is_latest_checkpoint:
-                messages = channel_values.get("messages")
-                if messages:
-                    values["messages"] = serialize_channel_values({"messages": messages}).get("messages", [])
-            is_latest_checkpoint = False
-
             # Derive next tasks
             tasks_raw = getattr(checkpoint_tuple, "tasks", []) or []
             next_tasks = [t.name for t in tasks_raw if hasattr(t, "name")]
 
-            # Strip LangGraph internal keys from metadata
-            user_meta = {k: v for k, v in metadata.items() if k not in ("created_at", "updated_at", "step", "source", "writes", "parents")}
-            # Keep step for ordering context
-            if "step" in metadata:
-                user_meta["step"] = metadata["step"]
-
             entries.append(
                 HistoryEntry(
                     checkpoint_id=checkpoint_id,
                     parent_checkpoint_id=parent_id,
-                    metadata=user_meta,
+                    metadata=metadata,
-                    values=values,
+                    values=serialize_channel_values(channel_values),
                     created_at=str(metadata.get("created_at", "")),
                     next=next_tasks,
                 )
             )
     except Exception:
-        logger.exception("Failed to get history for thread %s", sanitize_log_param(thread_id))
+        logger.exception("Failed to get history for thread %s", thread_id)
         raise HTTPException(status_code=500, detail="Failed to get thread history")
 
     return entries

backend/app/gateway/routers/uploads.py

@@ -4,10 +4,9 @@ import logging
 import os
 import stat
 
-from fastapi import APIRouter, File, HTTPException, Request, UploadFile
+from fastapi import APIRouter, File, HTTPException, UploadFile
 from pydantic import BaseModel
 
-from app.gateway.authz import require_permission
 from deerflow.config.paths import get_paths
 from deerflow.sandbox.sandbox_provider import get_sandbox_provider
 from deerflow.uploads.manager import (
@@ -55,10 +54,8 @@ def _make_file_sandbox_writable(file_path: os.PathLike[str] | str) -> None:
 
 
 @router.post("", response_model=UploadResponse)
-@require_permission("threads", "write", owner_check=True, require_existing=True)
 async def upload_files(
     thread_id: str,
-    request: Request,
     files: list[UploadFile] = File(...),
 ) -> UploadResponse:
     """Upload multiple files to a thread's uploads directory."""
@@ -136,8 +133,7 @@ async def upload_files(
 
 
 @router.get("/list", response_model=dict)
-@require_permission("threads", "read", owner_check=True)
+async def list_uploaded_files(thread_id: str) -> dict:
-async def list_uploaded_files(thread_id: str, request: Request) -> dict:
     """List all files in a thread's uploads directory."""
     try:
         uploads_dir = get_uploads_dir(thread_id)
@@ -155,8 +151,7 @@ async def list_uploaded_files(thread_id: str, request: Request) -> dict:
 
 
 @router.delete("/{filename}")
-@require_permission("threads", "delete", owner_check=True, require_existing=True)
+async def delete_uploaded_file(thread_id: str, filename: str) -> dict:
-async def delete_uploaded_file(thread_id: str, filename: str, request: Request) -> dict:
     """Delete a file from a thread's uploads directory."""
     try:
         uploads_dir = get_uploads_dir(thread_id)

backend/app/gateway/services.py

@@ -8,17 +8,16 @@ frames, and consuming stream bridge events.  Router modules
 from __future__ import annotations
 
 import asyncio
-import dataclasses
 import json
 import logging
 import re
+import time
 from typing import Any
 
 from fastapi import HTTPException, Request
 from langchain_core.messages import HumanMessage
 
-from app.gateway.deps import get_run_context, get_run_manager, get_run_store, get_stream_bridge
+from app.gateway.deps import get_checkpointer, get_run_manager, get_store, get_stream_bridge
-from app.gateway.utils import sanitize_log_param
 from deerflow.runtime import (
     END_SENTINEL,
     HEARTBEAT_SENTINEL,
@@ -172,6 +171,71 @@ def build_run_config(
 # ---------------------------------------------------------------------------
 
 
+async def _upsert_thread_in_store(store, thread_id: str, metadata: dict | None) -> None:
+    """Create or refresh the thread record in the Store.
+
+    Called from :func:`start_run` so that threads created via the stateless
+    ``/runs/stream`` endpoint (which never calls ``POST /threads``) still
+    appear in ``/threads/search`` results.
+    """
+    # Deferred import to avoid circular import with the threads router module.
+    from app.gateway.routers.threads import _store_upsert
+
+    try:
+        await _store_upsert(store, thread_id, metadata=metadata)
+    except Exception:
+        logger.warning("Failed to upsert thread %s in store (non-fatal)", thread_id)
+
+
+async def _sync_thread_title_after_run(
+    run_task: asyncio.Task,
+    thread_id: str,
+    checkpointer: Any,
+    store: Any,
+) -> None:
+    """Wait for *run_task* to finish, then persist the generated title to the Store.
+
+    TitleMiddleware writes the generated title to the LangGraph agent state
+    (checkpointer) but the Gateway's Store record is not updated automatically.
+    This coroutine closes that gap by reading the final checkpoint after the
+    run completes and syncing ``values.title`` into the Store record so that
+    subsequent ``/threads/search`` responses include the correct title.
+
+    Runs as a fire-and-forget :func:`asyncio.create_task`; failures are
+    logged at DEBUG level and never propagate.
+    """
+    # Wait for the background run task to complete (any outcome).
+    # asyncio.wait does not propagate task exceptions — it just returns
+    # when the task is done, cancelled, or failed.
+    await asyncio.wait({run_task})
+
+    # Deferred import to avoid circular import with the threads router module.
+    from app.gateway.routers.threads import _store_get, _store_put
+
+    try:
+        ckpt_config = {"configurable": {"thread_id": thread_id, "checkpoint_ns": ""}}
+        ckpt_tuple = await checkpointer.aget_tuple(ckpt_config)
+        if ckpt_tuple is None:
+            return
+
+        channel_values = ckpt_tuple.checkpoint.get("channel_values", {})
+        title = channel_values.get("title")
+        if not title:
+            return
+
+        existing = await _store_get(store, thread_id)
+        if existing is None:
+            return
+
+        updated = dict(existing)
+        updated.setdefault("values", {})["title"] = title
+        updated["updated_at"] = time.time()
+        await _store_put(store, updated)
+        logger.debug("Synced title %r for thread %s", title, thread_id)
+    except Exception:
+        logger.debug("Failed to sync title for thread %s (non-fatal)", thread_id, exc_info=True)
+
+
 async def start_run(
     body: Any,
     thread_id: str,
@@ -191,25 +255,11 @@ async def start_run(
     """
     bridge = get_stream_bridge(request)
     run_mgr = get_run_manager(request)
-    run_ctx = get_run_context(request)
+    checkpointer = get_checkpointer(request)
+    store = get_store(request)
 
     disconnect = DisconnectMode.cancel if body.on_disconnect == "cancel" else DisconnectMode.continue_
 
-    # Resolve follow_up_to_run_id: explicit from request, or auto-detect from latest successful run
-    follow_up_to_run_id = getattr(body, "follow_up_to_run_id", None)
-    if follow_up_to_run_id is None:
-        run_store = get_run_store(request)
-        try:
-            recent_runs = await run_store.list_by_thread(thread_id, limit=1)
-            if recent_runs and recent_runs[0].get("status") == "success":
-                follow_up_to_run_id = recent_runs[0]["run_id"]
-        except Exception:
-            pass  # Don't block run creation
-
-    # Enrich base context with per-run field
-    if follow_up_to_run_id:
-        run_ctx = dataclasses.replace(run_ctx, follow_up_to_run_id=follow_up_to_run_id)
-
     try:
         record = await run_mgr.create_or_reject(
             thread_id,
@@ -218,28 +268,17 @@ async def start_run(
             metadata=body.metadata or {},
             kwargs={"input": body.input, "config": body.config},
             multitask_strategy=body.multitask_strategy,
-            follow_up_to_run_id=follow_up_to_run_id,
         )
     except ConflictError as exc:
         raise HTTPException(status_code=409, detail=str(exc)) from exc
     except UnsupportedStrategyError as exc:
         raise HTTPException(status_code=501, detail=str(exc)) from exc
 
-    # Upsert thread metadata so the thread appears in /threads/search,
+    # Ensure the thread is visible in /threads/search, even for threads that
-    # even for threads that were never explicitly created via POST /threads
+    # were never explicitly created via POST /threads (e.g. stateless runs).
-    # (e.g. stateless runs).
+    store = get_store(request)
-    try:
+    if store is not None:
-        existing = await run_ctx.thread_meta_repo.get(thread_id)
+        await _upsert_thread_in_store(store, thread_id, body.metadata)
-        if existing is None:
-            await run_ctx.thread_meta_repo.create(
-                thread_id,
-                assistant_id=body.assistant_id,
-                metadata=body.metadata,
-            )
-        else:
-            await run_ctx.thread_meta_repo.update_status(thread_id, "running")
-    except Exception:
-        logger.warning("Failed to upsert thread_meta for %s (non-fatal)", sanitize_log_param(thread_id))
 
     agent_factory = resolve_agent_factory(body.assistant_id)
     graph_input = normalize_input(body.input)
@@ -272,7 +311,8 @@ async def start_run(
             bridge,
             run_mgr,
             record,
-            ctx=run_ctx,
+            checkpointer=checkpointer,
+            store=store,
             agent_factory=agent_factory,
             graph_input=graph_input,
             config=config,
@@ -284,9 +324,11 @@ async def start_run(
     )
     record.task = task
 
-    # Title sync is handled by worker.py's finally block which reads the
+    # After the run completes, sync the title generated by TitleMiddleware from
-    # title from the checkpoint and calls thread_meta_repo.update_display_name
+    # the checkpointer into the Store record so that /threads/search returns the
-    # after the run completes.
+    # correct title instead of an empty values dict.
+    if store is not None:
+        asyncio.create_task(_sync_thread_title_after_run(task, thread_id, checkpointer, store))
 
     return record
 

backend/app/gateway/utils.py

@@ -1,6 +0,0 @@
-"""Shared utility helpers for the Gateway layer."""
-
-
-def sanitize_log_param(value: str) -> str:
-    """Strip control characters to prevent log injection."""
-    return value.replace("\n", "").replace("\r", "").replace("\x00", "")

backend/docs/AUTH_TEST_DOCKER_GAP.md

@@ -1,77 +0,0 @@
-# Docker Test Gap (Section 七 7.4)
-
-This file documents the only **un-executed** test cases from
-`backend/docs/AUTH_TEST_PLAN.md` after the full release validation pass.
-
-## Why this gap exists
-
-The release validation environment (sg_dev: `10.251.229.92`) **does not have
-a Docker daemon installed**. The TC-DOCKER cases are container-runtime
-behavior tests that need an actual Docker engine to spin up
-`docker/docker-compose.yaml` services.
-
-```bash
-$ ssh sg_dev "which docker; docker --version"
-# (empty)
-# bash: docker: command not found
-```
-
-All other test plan sections were executed against either:
-- The local dev box (Mac, all services running locally), or
-- The deployed sg_dev instance (gateway + frontend + nginx via SSH tunnel)
-
-## Cases not executed
-
-| Case | Title | What it covers | Why not run |
-|---|---|---|---|
-| TC-DOCKER-01 | `users.db` volume persistence | Verify the `DEER_FLOW_HOME` bind mount survives container restart | needs `docker compose up` |
-| TC-DOCKER-02 | Session persistence across container restart | `AUTH_JWT_SECRET` env var keeps cookies valid after `docker compose down && up` | needs `docker compose down/up` |
-| TC-DOCKER-03 | Per-worker rate limiter divergence | Confirms in-process `_login_attempts` dict doesn't share state across `gunicorn` workers (4 by default in the compose file); known limitation, documented | needs multi-worker container |
-| TC-DOCKER-04 | IM channels skip AuthMiddleware | Verify Feishu/Slack/Telegram dispatchers run in-container against `http://langgraph:2024` without going through nginx | needs `docker logs` |
-| TC-DOCKER-05 | Admin credentials surfacing | **Updated post-simplify** — was "log scrape", now "0600 credential file in `DEER_FLOW_HOME`". The file-based behavior is already validated by TC-1.1 + TC-UPG-13 on sg_dev (non-Docker), so the only Docker-specific gap is verifying the volume mount carries the file out to the host | needs container + host volume |
-| TC-DOCKER-06 | Gateway-mode Docker deploy | `./scripts/deploy.sh --gateway` produces a 3-container topology (no `langgraph` container); same auth flow as standard mode | needs `docker compose --profile gateway` |
-
-## Coverage already provided by non-Docker tests
-
-The **auth-relevant** behavior in each Docker case is already exercised by
-the test cases that ran on sg_dev or local:
-
-| Docker case | Auth behavior covered by |
-|---|---|
-| TC-DOCKER-01 (volume persistence) | TC-REENT-01 on sg_dev (admin row survives gateway restart) — same SQLite file, just no container layer between |
-| TC-DOCKER-02 (session persistence) | TC-API-02/03/06 (cookie roundtrip), plus TC-REENT-04 (multi-cookie) — JWT verification is process-state-free, container restart is equivalent to `pkill uvicorn && uv run uvicorn` |
-| TC-DOCKER-03 (per-worker rate limit) | TC-GW-04 + TC-REENT-09 (single-worker rate limit + 5min expiry). The cross-worker divergence is an architectural property of the in-memory dict; no auth code path differs |
-| TC-DOCKER-04 (IM channels skip auth) | Code-level only: `app/channels/manager.py` uses `langgraph_sdk` directly with no cookie handling. The langgraph_auth handler is bypassed by going through SDK, not HTTP |
-| TC-DOCKER-05 (credential surfacing) | TC-1.1 on sg_dev (file at `~/deer-flow/backend/.deer-flow/admin_initial_credentials.txt`, mode 0600, password 22 chars) — the only Docker-unique step is whether the bind mount projects this path onto the host, which is a `docker compose` config check, not a runtime behavior change |
-| TC-DOCKER-06 (gateway-mode container) | Section 七 7.2 covered by TC-GW-01..05 + Section 二 (gateway-mode auth flow on sg_dev) — same Gateway code, container is just a packaging change |
-
-## Reproduction steps when Docker becomes available
-
-Anyone with `docker` + `docker compose` installed can reproduce the gap by
-running the test plan section verbatim. Pre-flight:
-
-```bash
-# Required on the host
-docker --version           # >=24.x
-docker compose version     # plugin >=2.x
-
-# Required env var (otherwise sessions reset on every container restart)
-echo "AUTH_JWT_SECRET=$(python3 -c 'import secrets; print(secrets.token_urlsafe(32))')" \
-  >> .env
-
-# Optional: pin DEER_FLOW_HOME to a stable host path
-echo "DEER_FLOW_HOME=$HOME/deer-flow-data" >> .env
-```
-
-Then run TC-DOCKER-01..06 from the test plan as written.
-
-## Decision log
-
-- **Not blocking the release.** The auth-relevant behavior in every Docker
-  case has an already-validated equivalent on bare metal. The gap is purely
-  about *container packaging* details (bind mounts, multi-worker, log
-  collection), not about whether the auth code paths work.
-- **TC-DOCKER-05 was updated in place** in `AUTH_TEST_PLAN.md` to reflect
-  the post-simplify reality (credentials file → 0600 file, no log leak).
-  The old "grep 'Password:' in docker logs" expectation would have failed
-  silently and given a false sense of coverage.

backend/docs/AUTH_UPGRADE.md

@@ -1,129 +0,0 @@
-# Authentication Upgrade Guide
-
-DeerFlow 内置了认证模块。本文档面向从无认证版本升级的用户。
-
-## 核心概念
-
-认证模块采用**始终强制**策略：
-
-- 首次启动时自动创建 admin 账号，随机密码打印到控制台日志
-- 认证从一开始就是强制的，无竞争窗口
-- 历史对话（升级前创建的 thread）自动迁移到 admin 名下
-
-## 升级步骤
-
-### 1. 更新代码
-
-```bash
-git pull origin main
-cd backend && make install
-```
-
-### 2. 首次启动
-
-```bash
-make dev
-```
-
-控制台会输出：
-
-```
-============================================================
-  Admin account created on first boot
-  Email:    admin@deerflow.dev
-  Password: aB3xK9mN_pQ7rT2w
-  Change it after login: Settings → Account
-============================================================
-```
-
-如果未登录就重启了服务，不用担心——只要 setup 未完成，每次启动都会重置密码并重新打印到控制台。
-
-### 3. 登录
-
-访问 `http://localhost:2026/login`，使用控制台输出的邮箱和密码登录。
-
-### 4. 修改密码
-
-登录后进入 Settings → Account → Change Password。
-
-### 5. 添加用户（可选）
-
-其他用户通过 `/login` 页面注册，自动获得 **user** 角色。每个用户只能看到自己的对话。
-
-## 安全机制
-
-| 机制 | 说明 |
-|------|------|
-| JWT HttpOnly Cookie | Token 不暴露给 JavaScript，防止 XSS 窃取 |
-| CSRF Double Submit Cookie | 所有 POST/PUT/DELETE 请求需携带 `X-CSRF-Token` |
-| bcrypt 密码哈希 | 密码不以明文存储 |
-| 多租户隔离 | 用户只能访问自己的 thread |
-| HTTPS 自适应 | 检测 `x-forwarded-proto`，自动设置 `Secure` cookie 标志 |
-
-## 常见操作
-
-### 忘记密码
-
-```bash
-cd backend
-
-# 重置 admin 密码
-python -m app.gateway.auth.reset_admin
-
-# 重置指定用户密码
-python -m app.gateway.auth.reset_admin --email user@example.com
-```
-
-会输出新的随机密码。
-
-### 完全重置
-
-删除用户数据库，重启后自动创建新 admin：
-
-```bash
-rm -f backend/.deer-flow/users.db
-# 重启服务，控制台输出新密码
-```
-
-## 数据存储
-
-| 文件 | 内容 |
-|------|------|
-| `.deer-flow/users.db` | SQLite 用户数据库（密码哈希、角色） |
-| `.env` 中的 `AUTH_JWT_SECRET` | JWT 签名密钥（未设置时自动生成临时密钥，重启后 session 失效） |
-
-### 生产环境建议
-
-```bash
-# 生成持久化 JWT 密钥，避免重启后所有用户需重新登录
-python -c "import secrets; print(secrets.token_urlsafe(32))"
-# 将输出添加到 .env：
-# AUTH_JWT_SECRET=<生成的密钥>
-```
-
-## API 端点
-
-| 端点 | 方法 | 说明 |
-|------|------|------|
-| `/api/v1/auth/login/local` | POST | 邮箱密码登录（OAuth2 form） |
-| `/api/v1/auth/register` | POST | 注册新用户（user 角色） |
-| `/api/v1/auth/logout` | POST | 登出（清除 cookie） |
-| `/api/v1/auth/me` | GET | 获取当前用户信息 |
-| `/api/v1/auth/change-password` | POST | 修改密码 |
-| `/api/v1/auth/setup-status` | GET | 检查 admin 是否存在 |
-
-## 兼容性
-
-- **标准模式**（`make dev`）：完全兼容，admin 自动创建
-- **Gateway 模式**（`make dev-pro`）：完全兼容
-- **Docker 部署**：完全兼容，`.deer-flow/users.db` 需持久化卷挂载
-- **IM 渠道**（Feishu/Slack/Telegram）：通过 LangGraph SDK 通信，不经过认证层
-- **DeerFlowClient**（嵌入式）：不经过 HTTP，不受认证影响
-
-## 故障排查
-
-| 症状 | 原因 | 解决 |
-|------|------|------|
-| 启动后没看到密码 | admin 已存在（非首次启动） | 用 `reset_admin` 重置，或删 `users.db` |
-| 登录后 POST 返回 403 | CSRF token 缺失 | 确认前端已更新 |
-| 重启后需要重新登录 | `AUTH_JWT_SECRET` 未持久化 | 在 `.env` 中设置固定密钥 |

backend/langgraph.json

@@ -8,9 +8,6 @@
   "graphs": {
     "lead_agent": "deerflow.agents:make_lead_agent"
   },
-  "auth": {
-    "path": "./app/gateway/langgraph_auth.py:auth"
-  },
   "checkpointer": {
     "path": "./packages/harness/deerflow/agents/checkpointer/async_provider.py:make_checkpointer"
   }

backend/packages/harness/deerflow/agents/checkpointer/async_provider.py

@@ -29,7 +29,7 @@ from deerflow.agents.checkpointer.provider import (
     POSTGRES_INSTALL,
     SQLITE_INSTALL,
 )
-from deerflow.config.app_config import get_app_config
+from deerflow.config.app_config import AppConfig
 from deerflow.runtime.store._sqlite_utils import ensure_sqlite_parent_dir, resolve_sqlite_conn_str
 
 logger = logging.getLogger(__name__)
@@ -83,77 +83,24 @@ async def _async_checkpointer(config) -> AsyncIterator[Checkpointer]:
 # ---------------------------------------------------------------------------
 
 
-@contextlib.asynccontextmanager
-async def _async_checkpointer_from_database(db_config) -> AsyncIterator[Checkpointer]:
-    """Async context manager that constructs a checkpointer from unified DatabaseConfig."""
-    if db_config.backend == "memory":
-        from langgraph.checkpoint.memory import InMemorySaver
-
-        yield InMemorySaver()
-        return
-
-    if db_config.backend == "sqlite":
-        try:
-            from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
-        except ImportError as exc:
-            raise ImportError(SQLITE_INSTALL) from exc
-
-        conn_str = db_config.checkpointer_sqlite_path
-        ensure_sqlite_parent_dir(conn_str)
-        async with AsyncSqliteSaver.from_conn_string(conn_str) as saver:
-            await saver.setup()
-            yield saver
-        return
-
-    if db_config.backend == "postgres":
-        try:
-            from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
-        except ImportError as exc:
-            raise ImportError(POSTGRES_INSTALL) from exc
-
-        if not db_config.postgres_url:
-            raise ValueError("database.postgres_url is required for the postgres backend")
-
-        async with AsyncPostgresSaver.from_conn_string(db_config.postgres_url) as saver:
-            await saver.setup()
-            yield saver
-        return
-
-    raise ValueError(f"Unknown database backend: {db_config.backend!r}")
-
-
 @contextlib.asynccontextmanager
 async def make_checkpointer() -> AsyncIterator[Checkpointer]:
     """Async context manager that yields a checkpointer for the caller's lifetime.
-    Resources are opened on enter and closed on exit -- no global state::
+    Resources are opened on enter and closed on exit — no global state::
 
         async with make_checkpointer() as checkpointer:
             app.state.checkpointer = checkpointer
 
     Yields an ``InMemorySaver`` when no checkpointer is configured in *config.yaml*.
-
-    Priority:
-    1. Legacy ``checkpointer:`` config section (backward compatible)
-    2. Unified ``database:`` config section
-    3. Default InMemorySaver
     """
 
-    config = get_app_config()
+    config = AppConfig.current()
 
-    # Legacy: standalone checkpointer config takes precedence
+    if config.checkpointer is None:
-    if config.checkpointer is not None:
+        from langgraph.checkpoint.memory import InMemorySaver
-        async with _async_checkpointer(config.checkpointer) as saver:
-            yield saver
-            return
 
-    # Unified database config
+        yield InMemorySaver()
-    db_config = getattr(config, "database", None)
+        return
-    if db_config is not None and db_config.backend != "memory":
-        async with _async_checkpointer_from_database(db_config) as saver:
-            yield saver
-            return
 
-    # Default: in-memory
+    async with _async_checkpointer(config.checkpointer) as saver:
-    from langgraph.checkpoint.memory import InMemorySaver
+        yield saver
-
-    yield InMemorySaver()

backend/packages/harness/deerflow/agents/checkpointer/provider.py

@@ -25,7 +25,7 @@ from collections.abc import Iterator
 
 from langgraph.types import Checkpointer
 
-from deerflow.config.app_config import get_app_config
+from deerflow.config.app_config import AppConfig
 from deerflow.config.checkpointer_config import CheckpointerConfig
 from deerflow.runtime.store._sqlite_utils import resolve_sqlite_conn_str
 
@@ -113,25 +113,10 @@ def get_checkpointer() -> Checkpointer:
     if _checkpointer is not None:
         return _checkpointer
 
-    # Ensure app config is loaded before checking checkpointer config
+    try:
-    # This prevents returning InMemorySaver when config.yaml actually has a checkpointer section
+        config = AppConfig.current().checkpointer
-    # but hasn't been loaded yet
+    except (LookupError, FileNotFoundError):
-    from deerflow.config.app_config import _app_config
+        config = None
-    from deerflow.config.checkpointer_config import get_checkpointer_config
-
-    config = get_checkpointer_config()
-
-    if config is None and _app_config is None:
-        # Only load app config lazily when neither the app config nor an explicit
-        # checkpointer config has been initialized yet. This keeps tests that
-        # intentionally set the global checkpointer config isolated from any
-        # ambient config.yaml on disk.
-        try:
-            get_app_config()
-        except FileNotFoundError:
-            # In test environments without config.yaml, this is expected.
-            pass
-        config = get_checkpointer_config()
     if config is None:
         from langgraph.checkpoint.memory import InMemorySaver
 
@@ -180,7 +165,7 @@ def checkpointer_context() -> Iterator[Checkpointer]:
     Yields an ``InMemorySaver`` when no checkpointer is configured in *config.yaml*.
     """
 
-    config = get_app_config()
+    config = AppConfig.current()
     if config.checkpointer is None:
         from langgraph.checkpoint.memory import InMemorySaver
 

backend/packages/harness/deerflow/agents/lead_agent/agent.py

@@ -3,6 +3,7 @@ import logging
 from langchain.agents import create_agent
 from langchain.agents.middleware import AgentMiddleware, SummarizationMiddleware
 from langchain_core.runnables import RunnableConfig
+from langgraph.graph.state import CompiledStateGraph
 
 from deerflow.agents.lead_agent.prompt import apply_prompt_template
 from deerflow.agents.middlewares.clarification_middleware import ClarificationMiddleware
@@ -16,8 +17,8 @@ from deerflow.agents.middlewares.tool_error_handling_middleware import build_lea
 from deerflow.agents.middlewares.view_image_middleware import ViewImageMiddleware
 from deerflow.agents.thread_state import ThreadState
 from deerflow.config.agents_config import load_agent_config
-from deerflow.config.app_config import get_app_config
+from deerflow.config.app_config import AppConfig
-from deerflow.config.summarization_config import get_summarization_config
+from deerflow.config.deer_flow_context import DeerFlowContext
 from deerflow.models import create_chat_model
 
 logger = logging.getLogger(__name__)
@@ -25,7 +26,7 @@ logger = logging.getLogger(__name__)
 
 def _resolve_model_name(requested_model_name: str | None = None) -> str:
     """Resolve a runtime model name safely, falling back to default if invalid. Returns None if no models are configured."""
-    app_config = get_app_config()
+    app_config = AppConfig.current()
     default_model_name = app_config.models[0].name if app_config.models else None
     if default_model_name is None:
         raise ValueError("No chat models are configured. Please configure at least one model in config.yaml.")
@@ -40,7 +41,7 @@ def _resolve_model_name(requested_model_name: str | None = None) -> str:
 
 def _create_summarization_middleware() -> SummarizationMiddleware | None:
     """Create and configure the summarization middleware from config."""
-    config = get_summarization_config()
+    config = AppConfig.current().summarization
 
     if not config.enabled:
         return None
@@ -56,15 +57,13 @@ def _create_summarization_middleware() -> SummarizationMiddleware | None:
     # Prepare keep parameter
     keep = config.keep.to_tuple()
 
-    # Prepare model parameter.
+    # Prepare model parameter
-    # Bind "middleware:summarize" tag so RunJournal identifies these LLM calls
-    # as middleware rather than lead_agent (SummarizationMiddleware is a
-    # LangChain built-in, so we tag the model at creation time).
     if config.model_name:
         model = create_chat_model(name=config.model_name, thinking_enabled=False)
     else:
+        # Use a lightweight model for summarization to save costs
+        # Falls back to default model if not explicitly specified
         model = create_chat_model(thinking_enabled=False)
-    model = model.with_config(tags=["middleware:summarize"])
 
     # Prepare kwargs
     kwargs = {
@@ -232,7 +231,7 @@ def _build_middlewares(config: RunnableConfig, model_name: str | None, agent_nam
         middlewares.append(todo_list_middleware)
 
     # Add TokenUsageMiddleware when token_usage tracking is enabled
-    if get_app_config().token_usage.enabled:
+    if AppConfig.current().token_usage.enabled:
         middlewares.append(TokenUsageMiddleware())
 
     # Add TitleMiddleware
@@ -243,7 +242,7 @@ def _build_middlewares(config: RunnableConfig, model_name: str | None, agent_nam
 
     # Add ViewImageMiddleware only if the current model supports vision.
     # Use the resolved runtime model_name from make_lead_agent to avoid stale config values.
-    app_config = get_app_config()
+    app_config = AppConfig.current()
     model_config = app_config.get_model_config(model_name) if model_name else None
     if model_config is not None and model_config.supports_vision:
         middlewares.append(ViewImageMiddleware())
@@ -272,7 +271,7 @@ def _build_middlewares(config: RunnableConfig, model_name: str | None, agent_nam
     return middlewares
 
 
-def make_lead_agent(config: RunnableConfig):
+def make_lead_agent(config: RunnableConfig) -> CompiledStateGraph:
     # Lazy import to avoid circular dependency
     from deerflow.tools import get_available_tools
     from deerflow.tools.builtins import setup_agent
@@ -295,7 +294,7 @@ def make_lead_agent(config: RunnableConfig):
     # Final model name resolution: request → agent config → global default, with fallback for unknown names
     model_name = _resolve_model_name(requested_model_name or agent_model_name)
 
-    app_config = get_app_config()
+    app_config = AppConfig.current()
     model_config = app_config.get_model_config(model_name)
 
     if model_config is None:
@@ -338,6 +337,7 @@ def make_lead_agent(config: RunnableConfig):
             middleware=_build_middlewares(config, model_name=model_name),
             system_prompt=apply_prompt_template(subagent_enabled=subagent_enabled, max_concurrent_subagents=max_concurrent_subagents, available_skills=set(["bootstrap"])),
             state_schema=ThreadState,
+            context_schema=DeerFlowContext,
         )
 
     # Default lead agent (unchanged behavior)
@@ -349,4 +349,5 @@ def make_lead_agent(config: RunnableConfig):
             subagent_enabled=subagent_enabled, max_concurrent_subagents=max_concurrent_subagents, agent_name=agent_name, available_skills=set(agent_config.skills) if agent_config and agent_config.skills is not None else None
         ),
         state_schema=ThreadState,
+        context_schema=DeerFlowContext,
     )

backend/packages/harness/deerflow/agents/lead_agent/prompt.py

@@ -5,6 +5,7 @@ from datetime import datetime
 from functools import lru_cache
 
 from deerflow.config.agents_config import load_agent_soul
+from deerflow.config.app_config import AppConfig
 from deerflow.skills import load_skills
 from deerflow.skills.types import Skill
 from deerflow.subagents import get_available_subagent_names
@@ -164,30 +165,6 @@ Skip simple one-off tasks.
 """
 
 
-def _skill_mutability_label(category: str) -> str:
-    return "[custom, editable]" if category == "custom" else "[built-in]"
-
-
-def clear_skills_system_prompt_cache() -> None:
-    _get_cached_skills_prompt_section.cache_clear()
-
-
-def _build_skill_evolution_section(skill_evolution_enabled: bool) -> str:
-    if not skill_evolution_enabled:
-        return ""
-    return """
-## Skill Self-Evolution
-After completing a task, consider creating or updating a skill when:
-- The task required 5+ tool calls to resolve
-- You overcame non-obvious errors or pitfalls
-- The user corrected your approach and the corrected version worked
-- You discovered a non-trivial, recurring workflow
-If you used a skill and encountered issues not covered by it, patch it immediately.
-Prefer patch over edit. Before creating a new skill, confirm with the user first.
-Skip simple one-off tasks.
-"""
-
-
 def _build_subagent_section(max_concurrent: int) -> str:
     """Build the subagent system prompt section with dynamic concurrency limit.
 
@@ -542,9 +519,8 @@ def _get_memory_context(agent_name: str | None = None) -> str:
     """
     try:
         from deerflow.agents.memory import format_memory_for_injection, get_memory_data
-        from deerflow.config.memory_config import get_memory_config
 
-        config = get_memory_config()
+        config = AppConfig.current().memory
         if not config.enabled or not config.injection_enabled:
             return ""
 
@@ -600,9 +576,7 @@ def get_skills_prompt_section(available_skills: set[str] | None = None) -> str:
     skills = _get_enabled_skills()
 
     try:
-        from deerflow.config import get_app_config
+        config = AppConfig.current()
-
-        config = get_app_config()
         container_base_path = config.skills.container_path
         skill_evolution_enabled = config.skill_evolution.enabled
     except Exception:
@@ -641,9 +615,7 @@ def get_deferred_tools_prompt_section() -> str:
     from deerflow.tools.builtins.tool_search import get_deferred_registry
 
     try:
-        from deerflow.config import get_app_config
+        if not AppConfig.current().tool_search.enabled:
-
-        if not get_app_config().tool_search.enabled:
             return ""
     except Exception:
         return ""
@@ -659,9 +631,7 @@ def get_deferred_tools_prompt_section() -> str:
 def _build_acp_section() -> str:
     """Build the ACP agent prompt section, only if ACP agents are configured."""
     try:
-        from deerflow.config.acp_config import get_acp_agents
+        agents = AppConfig.current().acp_agents
-
-        agents = get_acp_agents()
         if not agents:
             return ""
     except Exception:
@@ -679,9 +649,7 @@ def _build_acp_section() -> str:
 def _build_custom_mounts_section() -> str:
     """Build a prompt section for explicitly configured sandbox mounts."""
     try:
-        from deerflow.config import get_app_config
+        mounts = AppConfig.current().sandbox.mounts or []
-
-        mounts = get_app_config().sandbox.mounts or []
     except Exception:
         logger.exception("Failed to load configured sandbox mounts for the lead-agent prompt")
         return ""

backend/packages/harness/deerflow/agents/memory/queue.py

@@ -7,7 +7,7 @@ from dataclasses import dataclass, field
 from datetime import UTC, datetime
 from typing import Any
 
-from deerflow.config.memory_config import get_memory_config
+from deerflow.config.app_config import AppConfig
 
 logger = logging.getLogger(__name__)
 
@@ -56,7 +56,7 @@ class MemoryUpdateQueue:
             correction_detected: Whether recent turns include an explicit correction signal.
             reinforcement_detected: Whether recent turns include a positive reinforcement signal.
         """
-        config = get_memory_config()
+        config = AppConfig.current().memory
         if not config.enabled:
             return
 
@@ -87,7 +87,7 @@ class MemoryUpdateQueue:
 
     def _reset_timer(self) -> None:
         """Reset the debounce timer."""
-        config = get_memory_config()
+        config = AppConfig.current().memory
 
         # Cancel existing timer if any
         if self._timer is not None:

backend/packages/harness/deerflow/agents/memory/storage.py

@@ -9,7 +9,7 @@ from pathlib import Path
 from typing import Any
 
 from deerflow.config.agents_config import AGENT_NAME_PATTERN
-from deerflow.config.memory_config import get_memory_config
+from deerflow.config.app_config import AppConfig
 from deerflow.config.paths import get_paths
 
 logger = logging.getLogger(__name__)
@@ -84,7 +84,7 @@ class FileMemoryStorage(MemoryStorage):
             self._validate_agent_name(agent_name)
             return get_paths().agent_memory_file(agent_name)
 
-        config = get_memory_config()
+        config = AppConfig.current().memory
         if config.storage_path:
             p = Path(config.storage_path)
             return p if p.is_absolute() else get_paths().base_dir / p
@@ -177,7 +177,7 @@ def get_memory_storage() -> MemoryStorage:
         if _storage_instance is not None:
             return _storage_instance
 
-        config = get_memory_config()
+        config = AppConfig.current().memory
         storage_class_path = config.storage_class
 
         try:

backend/packages/harness/deerflow/agents/memory/updater.py

@@ -16,7 +16,7 @@ from deerflow.agents.memory.storage import (
     get_memory_storage,
     utc_now_iso_z,
 )
-from deerflow.config.memory_config import get_memory_config
+from deerflow.config.app_config import AppConfig
 from deerflow.models import create_chat_model
 
 logger = logging.getLogger(__name__)
@@ -265,7 +265,7 @@ class MemoryUpdater:
 
     def _get_model(self):
         """Get the model for memory updates."""
-        config = get_memory_config()
+        config = AppConfig.current().memory
         model_name = self._model_name or config.model_name
         return create_chat_model(name=model_name, thinking_enabled=False)
 
@@ -289,7 +289,7 @@ class MemoryUpdater:
         Returns:
             True if update was successful, False otherwise.
         """
-        config = get_memory_config()
+        config = AppConfig.current().memory
         if not config.enabled:
             return False
 
@@ -378,7 +378,7 @@ class MemoryUpdater:
         Returns:
             Updated memory data.
         """
-        config = get_memory_config()
+        config = AppConfig.current().memory
         now = utc_now_iso_z()
 
         # Update user sections

backend/packages/harness/deerflow/agents/middlewares/loop_detection_middleware.py

@@ -24,6 +24,8 @@ from langchain.agents.middleware import AgentMiddleware
 from langchain_core.messages import HumanMessage
 from langgraph.runtime import Runtime
 
+from deerflow.config.deer_flow_context import DeerFlowContext
+
 logger = logging.getLogger(__name__)
 
 # Defaults — can be overridden via constructor
@@ -31,6 +33,8 @@ _DEFAULT_WARN_THRESHOLD = 3  # inject warning after 3 identical calls
 _DEFAULT_HARD_LIMIT = 5  # force-stop after 5 identical calls
 _DEFAULT_WINDOW_SIZE = 20  # track last N tool calls
 _DEFAULT_MAX_TRACKED_THREADS = 100  # LRU eviction limit
+_DEFAULT_TOOL_FREQ_WARN = 30  # warn after 30 calls to the same tool type
+_DEFAULT_TOOL_FREQ_HARD_LIMIT = 50  # force-stop after 50 calls to the same tool type
 
 
 def _normalize_tool_call_args(raw_args: object) -> tuple[dict, str | None]:
@@ -125,8 +129,14 @@ def _hash_tool_calls(tool_calls: list[dict]) -> str:
 
 _WARNING_MSG = "[LOOP DETECTED] You are repeating the same tool calls. Stop calling tools and produce your final answer now. If you cannot complete the task, summarize what you accomplished so far."
 
+_TOOL_FREQ_WARNING_MSG = (
+    "[LOOP DETECTED] You have called {tool_name} {count} times without producing a final answer. Stop calling tools and produce your final answer now. If you cannot complete the task, summarize what you accomplished so far."
+)
+
 _HARD_STOP_MSG = "[FORCED STOP] Repeated tool calls exceeded the safety limit. Producing final answer with results collected so far."
 
+_TOOL_FREQ_HARD_STOP_MSG = "[FORCED STOP] Tool {tool_name} called {count} times — exceeded the per-tool safety limit. Producing final answer with results collected so far."
+
 
 class LoopDetectionMiddleware(AgentMiddleware[AgentState]):
     """Detects and breaks repetitive tool call loops.
@@ -140,6 +150,12 @@ class LoopDetectionMiddleware(AgentMiddleware[AgentState]):
             Default: 20.
         max_tracked_threads: Maximum number of threads to track before
             evicting the least recently used. Default: 100.
+        tool_freq_warn: Number of calls to the same tool *type* (regardless
+            of arguments) before injecting a frequency warning. Catches
+            cross-file read loops that hash-based detection misses.
+            Default: 30.
+        tool_freq_hard_limit: Number of calls to the same tool type before
+            forcing a stop. Default: 50.
     """
 
     def __init__(
@@ -148,23 +164,27 @@ class LoopDetectionMiddleware(AgentMiddleware[AgentState]):
         hard_limit: int = _DEFAULT_HARD_LIMIT,
         window_size: int = _DEFAULT_WINDOW_SIZE,
         max_tracked_threads: int = _DEFAULT_MAX_TRACKED_THREADS,
+        tool_freq_warn: int = _DEFAULT_TOOL_FREQ_WARN,
+        tool_freq_hard_limit: int = _DEFAULT_TOOL_FREQ_HARD_LIMIT,
     ):
         super().__init__()
         self.warn_threshold = warn_threshold
         self.hard_limit = hard_limit
         self.window_size = window_size
         self.max_tracked_threads = max_tracked_threads
+        self.tool_freq_warn = tool_freq_warn
+        self.tool_freq_hard_limit = tool_freq_hard_limit
         self._lock = threading.Lock()
         # Per-thread tracking using OrderedDict for LRU eviction
         self._history: OrderedDict[str, list[str]] = OrderedDict()
         self._warned: dict[str, set[str]] = defaultdict(set)
+        # Per-thread, per-tool-type cumulative call counts
+        self._tool_freq: dict[str, dict[str, int]] = defaultdict(lambda: defaultdict(int))
+        self._tool_freq_warned: dict[str, set[str]] = defaultdict(set)
 
-    def _get_thread_id(self, runtime: Runtime) -> str:
+    def _get_thread_id(self, runtime: Runtime[DeerFlowContext]) -> str:
         """Extract thread_id from runtime context for per-thread tracking."""
-        thread_id = runtime.context.get("thread_id") if runtime.context else None
+        return runtime.context.thread_id or "default"
-        if thread_id:
-            return thread_id
-        return "default"
 
     def _evict_if_needed(self) -> None:
         """Evict least recently used threads if over the limit.
@@ -174,11 +194,19 @@ class LoopDetectionMiddleware(AgentMiddleware[AgentState]):
         while len(self._history) > self.max_tracked_threads:
             evicted_id, _ = self._history.popitem(last=False)
             self._warned.pop(evicted_id, None)
+            self._tool_freq.pop(evicted_id, None)
+            self._tool_freq_warned.pop(evicted_id, None)
             logger.debug("Evicted loop tracking for thread %s (LRU)", evicted_id)
 
     def _track_and_check(self, state: AgentState, runtime: Runtime) -> tuple[str | None, bool]:
         """Track tool calls and check for loops.
 
+        Two detection layers:
+          1. **Hash-based** (existing): catches identical tool call sets.
+          2. **Frequency-based** (new): catches the same *tool type* being
+             called many times with varying arguments (e.g. ``read_file``
+             on 40 different files).
+
         Returns:
             (warning_message_or_none, should_hard_stop)
         """
@@ -213,6 +241,7 @@ class LoopDetectionMiddleware(AgentMiddleware[AgentState]):
             count = history.count(call_hash)
             tool_names = [tc.get("name", "?") for tc in tool_calls]
 
+            # --- Layer 1: hash-based (identical call sets) ---
             if count >= self.hard_limit:
                 logger.error(
                     "Loop hard limit reached — forcing stop",
@@ -239,8 +268,40 @@ class LoopDetectionMiddleware(AgentMiddleware[AgentState]):
                         },
                     )
                     return _WARNING_MSG, False
-                # Warning already injected for this hash — suppress
+
-                return None, False
+            # --- Layer 2: per-tool-type frequency ---
+            freq = self._tool_freq[thread_id]
+            for tc in tool_calls:
+                name = tc.get("name", "")
+                if not name:
+                    continue
+                freq[name] += 1
+                tc_count = freq[name]
+
+                if tc_count >= self.tool_freq_hard_limit:
+                    logger.error(
+                        "Tool frequency hard limit reached — forcing stop",
+                        extra={
+                            "thread_id": thread_id,
+                            "tool_name": name,
+                            "count": tc_count,
+                        },
+                    )
+                    return _TOOL_FREQ_HARD_STOP_MSG.format(tool_name=name, count=tc_count), True
+
+                if tc_count >= self.tool_freq_warn:
+                    warned = self._tool_freq_warned[thread_id]
+                    if name not in warned:
+                        warned.add(name)
+                        logger.warning(
+                            "Tool frequency warning — too many calls to same tool type",
+                            extra={
+                                "thread_id": thread_id,
+                                "tool_name": name,
+                                "count": tc_count,
+                            },
+                        )
+                        return _TOOL_FREQ_WARNING_MSG.format(tool_name=name, count=tc_count), False
 
         return None, False
 
@@ -271,7 +332,7 @@ class LoopDetectionMiddleware(AgentMiddleware[AgentState]):
             stripped_msg = last_msg.model_copy(
                 update={
                     "tool_calls": [],
-                    "content": self._append_text(last_msg.content, _HARD_STOP_MSG),
+                    "content": self._append_text(last_msg.content, warning),
                 }
             )
             return {"messages": [stripped_msg]}
@@ -288,11 +349,11 @@ class LoopDetectionMiddleware(AgentMiddleware[AgentState]):
         return None
 
     @override
-    def after_model(self, state: AgentState, runtime: Runtime) -> dict | None:
+    def after_model(self, state: AgentState, runtime: Runtime[DeerFlowContext]) -> dict | None:
         return self._apply(state, runtime)
 
     @override
-    async def aafter_model(self, state: AgentState, runtime: Runtime) -> dict | None:
+    async def aafter_model(self, state: AgentState, runtime: Runtime[DeerFlowContext]) -> dict | None:
         return self._apply(state, runtime)
 
     def reset(self, thread_id: str | None = None) -> None:
@@ -301,6 +362,10 @@ class LoopDetectionMiddleware(AgentMiddleware[AgentState]):
             if thread_id:
                 self._history.pop(thread_id, None)
                 self._warned.pop(thread_id, None)
+                self._tool_freq.pop(thread_id, None)
+                self._tool_freq_warned.pop(thread_id, None)
             else:
                 self._history.clear()
                 self._warned.clear()
+                self._tool_freq.clear()
+                self._tool_freq_warned.clear()

backend/packages/harness/deerflow/agents/middlewares/memory_middleware.py

@@ -6,11 +6,10 @@ from typing import Any, override
 
 from langchain.agents import AgentState
 from langchain.agents.middleware import AgentMiddleware
-from langgraph.config import get_config
 from langgraph.runtime import Runtime
 
 from deerflow.agents.memory.queue import get_memory_queue
-from deerflow.config.memory_config import get_memory_config
+from deerflow.config.deer_flow_context import DeerFlowContext
 
 logger = logging.getLogger(__name__)
 
@@ -193,7 +192,7 @@ class MemoryMiddleware(AgentMiddleware[MemoryMiddlewareState]):
         self._agent_name = agent_name
 
     @override
-    def after_agent(self, state: MemoryMiddlewareState, runtime: Runtime) -> dict | None:
+    def after_agent(self, state: MemoryMiddlewareState, runtime: Runtime[DeerFlowContext]) -> dict | None:
         """Queue conversation for memory update after agent completes.
 
         Args:
@@ -203,15 +202,11 @@ class MemoryMiddleware(AgentMiddleware[MemoryMiddlewareState]):
         Returns:
             None (no state changes needed from this middleware).
         """
-        config = get_memory_config()
+        memory_config = runtime.context.app_config.memory
-        if not config.enabled:
+        if not memory_config.enabled:
             return None
 
-        # Get thread ID from runtime context first, then fall back to LangGraph's configurable metadata
+        thread_id = runtime.context.thread_id
-        thread_id = runtime.context.get("thread_id") if runtime.context else None
-        if thread_id is None:
-            config_data = get_config()
-            thread_id = config_data.get("configurable", {}).get("thread_id")
         if not thread_id:
             logger.debug("No thread_id in context, skipping memory update")
             return None

backend/packages/harness/deerflow/agents/middlewares/thread_data_middleware.py

@@ -3,10 +3,10 @@ from typing import NotRequired, override
 
 from langchain.agents import AgentState
 from langchain.agents.middleware import AgentMiddleware
-from langgraph.config import get_config
 from langgraph.runtime import Runtime
 
 from deerflow.agents.thread_state import ThreadDataState
+from deerflow.config.deer_flow_context import DeerFlowContext
 from deerflow.config.paths import Paths, get_paths
 
 logger = logging.getLogger(__name__)
@@ -74,14 +74,10 @@ class ThreadDataMiddleware(AgentMiddleware[ThreadDataMiddlewareState]):
         return self._get_thread_paths(thread_id)
 
     @override
-    def before_agent(self, state: ThreadDataMiddlewareState, runtime: Runtime) -> dict | None:
+    def before_agent(self, state: ThreadDataMiddlewareState, runtime: Runtime[DeerFlowContext]) -> dict | None:
-        context = runtime.context or {}
+        thread_id = runtime.context.thread_id
-        thread_id = context.get("thread_id")
-        if thread_id is None:
-            config = get_config()
-            thread_id = config.get("configurable", {}).get("thread_id")
 
-        if thread_id is None:
+        if not thread_id:
             raise ValueError("Thread ID is required in runtime context or config.configurable")
 
         if self._lazy_init:

backend/packages/harness/deerflow/agents/middlewares/title_middleware.py

@@ -1,14 +1,13 @@
 """Middleware for automatic thread title generation."""
 
 import logging
-from typing import Any, NotRequired, override
+from typing import NotRequired, override
 
 from langchain.agents import AgentState
 from langchain.agents.middleware import AgentMiddleware
-from langgraph.config import get_config
 from langgraph.runtime import Runtime
 
-from deerflow.config.title_config import get_title_config
+from deerflow.config.app_config import AppConfig
 from deerflow.models import create_chat_model
 
 logger = logging.getLogger(__name__)
@@ -46,7 +45,7 @@ class TitleMiddleware(AgentMiddleware[TitleMiddlewareState]):
 
     def _should_generate_title(self, state: TitleMiddlewareState) -> bool:
         """Check if we should generate a title for this thread."""
-        config = get_title_config()
+        config = AppConfig.current().title
         if not config.enabled:
             return False
 
@@ -71,7 +70,7 @@ class TitleMiddleware(AgentMiddleware[TitleMiddlewareState]):
 
         Returns (prompt_string, user_msg) so callers can use user_msg as fallback.
         """
-        config = get_title_config()
+        config = AppConfig.current().title
         messages = state.get("messages", [])
 
         user_msg_content = next((m.content for m in messages if m.type == "human"), "")
@@ -89,32 +88,18 @@ class TitleMiddleware(AgentMiddleware[TitleMiddlewareState]):
 
     def _parse_title(self, content: object) -> str:
         """Normalize model output into a clean title string."""
-        config = get_title_config()
+        config = AppConfig.current().title
         title_content = self._normalize_content(content)
         title = title_content.strip().strip('"').strip("'")
         return title[: config.max_chars] if len(title) > config.max_chars else title
 
     def _fallback_title(self, user_msg: str) -> str:
-        config = get_title_config()
+        config = AppConfig.current().title
         fallback_chars = min(config.max_chars, 50)
         if len(user_msg) > fallback_chars:
             return user_msg[:fallback_chars].rstrip() + "..."
         return user_msg if user_msg else "New Conversation"
 
-    def _get_runnable_config(self) -> dict[str, Any]:
-        """Inherit the parent RunnableConfig and add middleware tag.
-
-        This ensures RunJournal identifies LLM calls from this middleware
-        as ``middleware:title`` instead of ``lead_agent``.
-        """
-        try:
-            parent = get_config()
-        except Exception:
-            parent = {}
-        config = {**parent}
-        config["tags"] = [*(config.get("tags") or []), "middleware:title"]
-        return config
-
     def _generate_title_result(self, state: TitleMiddlewareState) -> dict | None:
         """Generate a local fallback title without blocking on an LLM call."""
         if not self._should_generate_title(state):
@@ -128,7 +113,7 @@ class TitleMiddleware(AgentMiddleware[TitleMiddlewareState]):
         if not self._should_generate_title(state):
             return None
 
-        config = get_title_config()
+        config = AppConfig.current().title
         prompt, user_msg = self._build_title_prompt(state)
 
         try:
@@ -136,7 +121,7 @@ class TitleMiddleware(AgentMiddleware[TitleMiddlewareState]):
                 model = create_chat_model(name=config.model_name, thinking_enabled=False)
             else:
                 model = create_chat_model(thinking_enabled=False)
-            response = await model.ainvoke(prompt, config=self._get_runnable_config())
+            response = await model.ainvoke(prompt)
             title = self._parse_title(response.content)
             if title:
                 return {"title": title}

backend/packages/harness/deerflow/agents/middlewares/tool_error_handling_middleware.py

@@ -94,9 +94,9 @@ def _build_runtime_middlewares(
     middlewares.append(LLMErrorHandlingMiddleware())
 
     # Guardrail middleware (if configured)
-    from deerflow.config.guardrails_config import get_guardrails_config
+    from deerflow.config.app_config import AppConfig
 
-    guardrails_config = get_guardrails_config()
+    guardrails_config = AppConfig.current().guardrails
     if guardrails_config.enabled and guardrails_config.provider:
         import inspect
 

backend/packages/harness/deerflow/agents/middlewares/uploads_middleware.py

@@ -9,6 +9,7 @@ from langchain.agents.middleware import AgentMiddleware
 from langchain_core.messages import HumanMessage
 from langgraph.runtime import Runtime
 
+from deerflow.config.deer_flow_context import DeerFlowContext
 from deerflow.config.paths import Paths, get_paths
 from deerflow.utils.file_conversion import extract_outline
 
@@ -184,7 +185,7 @@ class UploadsMiddleware(AgentMiddleware[UploadsMiddlewareState]):
         return files if files else None
 
     @override
-    def before_agent(self, state: UploadsMiddlewareState, runtime: Runtime) -> dict | None:
+    def before_agent(self, state: UploadsMiddlewareState, runtime: Runtime[DeerFlowContext]) -> dict | None:
         """Inject uploaded files information before agent execution.
 
         New files come from the current message's additional_kwargs.files.
@@ -213,14 +214,7 @@ class UploadsMiddleware(AgentMiddleware[UploadsMiddlewareState]):
             return None
 
         # Resolve uploads directory for existence checks
-        thread_id = (runtime.context or {}).get("thread_id")
+        thread_id = runtime.context.thread_id
-        if thread_id is None:
-            try:
-                from langgraph.config import get_config
-
-                thread_id = get_config().get("configurable", {}).get("thread_id")
-            except RuntimeError:
-                pass  # get_config() raises outside a runnable context (e.g. unit tests)
         uploads_dir = self._paths.sandbox_uploads_dir(thread_id) if thread_id else None
 
         # Get newly uploaded files from the current message's additional_kwargs.files
@@ -262,21 +256,25 @@ class UploadsMiddleware(AgentMiddleware[UploadsMiddlewareState]):
         files_message = self._create_files_message(new_files, historical_files)
 
         # Extract original content - handle both string and list formats
-        original_content = ""
+        original_content = last_message.content
-        if isinstance(last_message.content, str):
+        if isinstance(original_content, str):
-            original_content = last_message.content
+            # Simple case: string content, just prepend files message
-        elif isinstance(last_message.content, list):
+            updated_content = f"{files_message}\n\n{original_content}"
-            text_parts = []
+        elif isinstance(original_content, list):
-            for block in last_message.content:
+            # Complex case: list content (multimodal), preserve all blocks
-                if isinstance(block, dict) and block.get("type") == "text":
+            # Prepend files message as the first text block
-                    text_parts.append(block.get("text", ""))
+            files_block = {"type": "text", "text": f"{files_message}\n\n"}
-            original_content = "\n".join(text_parts)
+            # Keep all original blocks (including images)
+            updated_content = [files_block, *original_content]
+        else:
+            # Other types, preserve as-is
+            updated_content = original_content
 
         # Create new message with combined content.
         # Preserve additional_kwargs (including files metadata) so the frontend
         # can read structured file info from the streamed message.
         updated_message = HumanMessage(
-            content=f"{files_message}\n\n{original_content}",
+            content=updated_content,
             id=last_message.id,
             additional_kwargs=last_message.additional_kwargs,
         )

backend/packages/harness/deerflow/client.py

@@ -36,8 +36,9 @@ from deerflow.agents.lead_agent.agent import _build_middlewares
 from deerflow.agents.lead_agent.prompt import apply_prompt_template
 from deerflow.agents.thread_state import ThreadState
 from deerflow.config.agents_config import AGENT_NAME_PATTERN
-from deerflow.config.app_config import get_app_config, reload_app_config
+from deerflow.config.app_config import AppConfig
-from deerflow.config.extensions_config import ExtensionsConfig, SkillStateConfig, get_extensions_config, reload_extensions_config
+from deerflow.config.deer_flow_context import DeerFlowContext
+from deerflow.config.extensions_config import ExtensionsConfig, SkillStateConfig
 from deerflow.config.paths import get_paths
 from deerflow.models import create_chat_model
 from deerflow.skills.installer import install_skill_from_archive
@@ -141,8 +142,8 @@ class DeerFlowClient:
             middlewares: Optional list of custom middlewares to inject into the agent.
         """
         if config_path is not None:
-            reload_app_config(config_path)
+            AppConfig.init(AppConfig.from_file(config_path))
-        self._app_config = get_app_config()
+        self._app_config = AppConfig.current()
 
         if agent_name is not None and not AGENT_NAME_PATTERN.match(agent_name):
             raise ValueError(f"Invalid agent name '{agent_name}'. Must match pattern: {AGENT_NAME_PATTERN.pattern}")
@@ -551,9 +552,7 @@ class DeerFlowClient:
         self._ensure_agent(config)
 
         state: dict[str, Any] = {"messages": [HumanMessage(content=message)]}
-        context = {"thread_id": thread_id}
+        context = DeerFlowContext(app_config=self._app_config, thread_id=thread_id, agent_name=self._agent_name)
-        if self._agent_name:
-            context["agent_name"] = self._agent_name
 
         seen_ids: set[str] = set()
         # Cross-mode handoff: ids already streamed via LangGraph ``messages``
@@ -816,8 +815,8 @@ class DeerFlowClient:
             Dict with "mcp_servers" key mapping server name to config,
             matching the Gateway API ``McpConfigResponse`` schema.
         """
-        config = get_extensions_config()
+        ext = AppConfig.current().extensions
-        return {"mcp_servers": {name: server.model_dump() for name, server in config.mcp_servers.items()}}
+        return {"mcp_servers": {name: server.model_dump() for name, server in ext.mcp_servers.items()}}
 
     def update_mcp_config(self, mcp_servers: dict[str, dict]) -> dict:
         """Update MCP server configurations.
@@ -839,18 +838,19 @@ class DeerFlowClient:
         if config_path is None:
             raise FileNotFoundError("Cannot locate extensions_config.json. Set DEER_FLOW_EXTENSIONS_CONFIG_PATH or ensure it exists in the project root.")
 
-        current_config = get_extensions_config()
+        current_ext = AppConfig.current().extensions
 
         config_data = {
             "mcpServers": mcp_servers,
-            "skills": {name: {"enabled": skill.enabled} for name, skill in current_config.skills.items()},
+            "skills": {name: {"enabled": skill.enabled} for name, skill in current_ext.skills.items()},
         }
 
         self._atomic_write_json(config_path, config_data)
 
         self._agent = None
         self._agent_config_key = None
-        reloaded = reload_extensions_config()
+        AppConfig.init(AppConfig.from_file())
+        reloaded = AppConfig.current().extensions
         return {"mcp_servers": {name: server.model_dump() for name, server in reloaded.mcp_servers.items()}}
 
     # ------------------------------------------------------------------
@@ -904,19 +904,19 @@ class DeerFlowClient:
         if config_path is None:
             raise FileNotFoundError("Cannot locate extensions_config.json. Set DEER_FLOW_EXTENSIONS_CONFIG_PATH or ensure it exists in the project root.")
 
-        extensions_config = get_extensions_config()
+        ext = AppConfig.current().extensions
-        extensions_config.skills[name] = SkillStateConfig(enabled=enabled)
+        ext.skills[name] = SkillStateConfig(enabled=enabled)
 
         config_data = {
-            "mcpServers": {n: s.model_dump() for n, s in extensions_config.mcp_servers.items()},
+            "mcpServers": {n: s.model_dump() for n, s in ext.mcp_servers.items()},
-            "skills": {n: {"enabled": sc.enabled} for n, sc in extensions_config.skills.items()},
+            "skills": {n: {"enabled": sc.enabled} for n, sc in ext.skills.items()},
         }
 
         self._atomic_write_json(config_path, config_data)
 
         self._agent = None
         self._agent_config_key = None
-        reload_extensions_config()
+        AppConfig.init(AppConfig.from_file())
 
         updated = next((s for s in load_skills(enabled_only=False) if s.name == name), None)
         if updated is None:
@@ -999,9 +999,7 @@ class DeerFlowClient:
         Returns:
             Memory config dict.
         """
-        from deerflow.config.memory_config import get_memory_config
+        config = AppConfig.current().memory
-
-        config = get_memory_config()
         return {
             "enabled": config.enabled,
             "storage_path": config.storage_path,

backend/packages/harness/deerflow/community/aio_sandbox/aio_sandbox_provider.py

@@ -25,7 +25,7 @@ except ImportError:  # pragma: no cover - Windows fallback
     fcntl = None  # type: ignore[assignment]
     import msvcrt
 
-from deerflow.config import get_app_config
+from deerflow.config.app_config import AppConfig
 from deerflow.config.paths import VIRTUAL_PATH_PREFIX, get_paths
 from deerflow.sandbox.sandbox import Sandbox
 from deerflow.sandbox.sandbox_provider import SandboxProvider
@@ -148,7 +148,7 @@ class AioSandboxProvider(SandboxProvider):
 
     def _load_config(self) -> dict:
         """Load sandbox configuration from app config."""
-        config = get_app_config()
+        config = AppConfig.current()
         sandbox_config = config.sandbox
 
         idle_timeout = getattr(sandbox_config, "idle_timeout", None)
@@ -279,7 +279,7 @@ class AioSandboxProvider(SandboxProvider):
         so the host Docker daemon can resolve the path.
         """
         try:
-            config = get_app_config()
+            config = AppConfig.current()
             skills_path = config.skills.get_skills_path()
             container_path = config.skills.container_path
 

backend/packages/harness/deerflow/community/ddg_search/tools.py

@@ -7,7 +7,7 @@ import logging
 
 from langchain.tools import tool
 
-from deerflow.config import get_app_config
+from deerflow.config.app_config import AppConfig
 
 logger = logging.getLogger(__name__)
 
@@ -63,7 +63,7 @@ def web_search_tool(
         query: Search keywords describing what you want to find. Be specific for better results.
         max_results: Maximum number of results to return. Default is 5.
     """
-    config = get_app_config().get_tool_config("web_search")
+    config = AppConfig.current().get_tool_config("web_search")
 
     # Override max_results from config if set
     if config is not None and "max_results" in config.model_extra:

backend/packages/harness/deerflow/community/exa/tools.py

@@ -3,11 +3,11 @@ import json
 from exa_py import Exa
 from langchain.tools import tool
 
-from deerflow.config import get_app_config
+from deerflow.config.app_config import AppConfig
 
 
 def _get_exa_client(tool_name: str = "web_search") -> Exa:
-    config = get_app_config().get_tool_config(tool_name)
+    config = AppConfig.current().get_tool_config(tool_name)
     api_key = None
     if config is not None and "api_key" in config.model_extra:
         api_key = config.model_extra.get("api_key")
@@ -22,7 +22,7 @@ def web_search_tool(query: str) -> str:
         query: The query to search for.
     """
     try:
-        config = get_app_config().get_tool_config("web_search")
+        config = AppConfig.current().get_tool_config("web_search")
         max_results = 5
         search_type = "auto"
         contents_max_characters = 1000

backend/packages/harness/deerflow/community/firecrawl/tools.py

@@ -3,11 +3,11 @@ import json
 from firecrawl import FirecrawlApp
 from langchain.tools import tool
 
-from deerflow.config import get_app_config
+from deerflow.config.app_config import AppConfig
 
 
 def _get_firecrawl_client(tool_name: str = "web_search") -> FirecrawlApp:
-    config = get_app_config().get_tool_config(tool_name)
+    config = AppConfig.current().get_tool_config(tool_name)
     api_key = None
     if config is not None and "api_key" in config.model_extra:
         api_key = config.model_extra.get("api_key")
@@ -22,7 +22,7 @@ def web_search_tool(query: str) -> str:
         query: The query to search for.
     """
     try:
-        config = get_app_config().get_tool_config("web_search")
+        config = AppConfig.current().get_tool_config("web_search")
         max_results = 5
         if config is not None:
             max_results = config.model_extra.get("max_results", max_results)

backend/packages/harness/deerflow/community/image_search/tools.py

@@ -7,7 +7,7 @@ import logging
 
 from langchain.tools import tool
 
-from deerflow.config import get_app_config
+from deerflow.config.app_config import AppConfig
 
 logger = logging.getLogger(__name__)
 
@@ -99,7 +99,7 @@ def image_search_tool(
         type_image: Image type filter. Options: "photo", "clipart", "gif", "transparent", "line". Use "photo" for realistic references.
         layout: Layout filter. Options: "Square", "Tall", "Wide". Choose based on your generation needs.
     """
-    config = get_app_config().get_tool_config("image_search")
+    config = AppConfig.current().get_tool_config("image_search")
 
     # Override max_results from config if set
     if config is not None and "max_results" in config.model_extra:

backend/packages/harness/deerflow/community/infoquest/tools.py

@@ -1,6 +1,6 @@
 from langchain.tools import tool
 
-from deerflow.config import get_app_config
+from deerflow.config.app_config import AppConfig
 from deerflow.utils.readability import ReadabilityExtractor
 
 from .infoquest_client import InfoQuestClient
@@ -9,12 +9,12 @@ readability_extractor = ReadabilityExtractor()
 
 
 def _get_infoquest_client() -> InfoQuestClient:
-    search_config = get_app_config().get_tool_config("web_search")
+    search_config = AppConfig.current().get_tool_config("web_search")
     search_time_range = -1
     if search_config is not None and "search_time_range" in search_config.model_extra:
         search_time_range = search_config.model_extra.get("search_time_range")
 
-    fetch_config = get_app_config().get_tool_config("web_fetch")
+    fetch_config = AppConfig.current().get_tool_config("web_fetch")
     fetch_time = -1
     if fetch_config is not None and "fetch_time" in fetch_config.model_extra:
         fetch_time = fetch_config.model_extra.get("fetch_time")
@@ -25,7 +25,7 @@ def _get_infoquest_client() -> InfoQuestClient:
     if fetch_config is not None and "navigation_timeout" in fetch_config.model_extra:
         navigation_timeout = fetch_config.model_extra.get("navigation_timeout")
 
-    image_search_config = get_app_config().get_tool_config("image_search")
+    image_search_config = AppConfig.current().get_tool_config("image_search")
     image_search_time_range = -1
     if image_search_config is not None and "image_search_time_range" in image_search_config.model_extra:
         image_search_time_range = image_search_config.model_extra.get("image_search_time_range")

backend/packages/harness/deerflow/community/jina_ai/tools.py

@@ -1,7 +1,7 @@
 from langchain.tools import tool
 
 from deerflow.community.jina_ai.jina_client import JinaClient
-from deerflow.config import get_app_config
+from deerflow.config.app_config import AppConfig
 from deerflow.utils.readability import ReadabilityExtractor
 
 readability_extractor = ReadabilityExtractor()
@@ -20,7 +20,7 @@ async def web_fetch_tool(url: str) -> str:
     """
     jina_client = JinaClient()
     timeout = 10
-    config = get_app_config().get_tool_config("web_fetch")
+    config = AppConfig.current().get_tool_config("web_fetch")
     if config is not None and "timeout" in config.model_extra:
         timeout = config.model_extra.get("timeout")
     html_content = await jina_client.crawl(url, return_format="html", timeout=timeout)

backend/packages/harness/deerflow/community/tavily/tools.py

@@ -3,11 +3,11 @@ import json
 from langchain.tools import tool
 from tavily import TavilyClient
 
-from deerflow.config import get_app_config
+from deerflow.config.app_config import AppConfig
 
 
 def _get_tavily_client() -> TavilyClient:
-    config = get_app_config().get_tool_config("web_search")
+    config = AppConfig.current().get_tool_config("web_search")
     api_key = None
     if config is not None and "api_key" in config.model_extra:
         api_key = config.model_extra.get("api_key")
@@ -21,7 +21,7 @@ def web_search_tool(query: str) -> str:
     Args:
         query: The query to search for.
     """
-    config = get_app_config().get_tool_config("web_search")
+    config = AppConfig.current().get_tool_config("web_search")
     max_results = 5
     if config is not None and "max_results" in config.model_extra:
         max_results = config.model_extra.get("max_results")

backend/packages/harness/deerflow/config/__init__.py

@@ -1,6 +1,6 @@
-from .app_config import get_app_config
+from .app_config import AppConfig
-from .extensions_config import ExtensionsConfig, get_extensions_config
+from .extensions_config import ExtensionsConfig
-from .memory_config import MemoryConfig, get_memory_config
+from .memory_config import MemoryConfig
 from .paths import Paths, get_paths
 from .skill_evolution_config import SkillEvolutionConfig
 from .skills_config import SkillsConfig
@@ -13,18 +13,16 @@ from .tracing_config import (
 )
 
 __all__ = [
-    "get_app_config",
+    "AppConfig",
-    "SkillEvolutionConfig",
-    "Paths",
-    "get_paths",
-    "SkillsConfig",
     "ExtensionsConfig",
-    "get_extensions_config",
     "MemoryConfig",
-    "get_memory_config",
+    "Paths",
-    "get_tracing_config",
+    "SkillEvolutionConfig",
-    "get_explicitly_enabled_tracing_providers",
+    "SkillsConfig",
     "get_enabled_tracing_providers",
+    "get_explicitly_enabled_tracing_providers",
+    "get_paths",
+    "get_tracing_config",
     "is_tracing_enabled",
     "validate_enabled_tracing_providers",
 ]

backend/packages/harness/deerflow/config/acp_config.py

@@ -1,16 +1,13 @@
 """ACP (Agent Client Protocol) agent configuration loaded from config.yaml."""
 
-import logging
+from pydantic import BaseModel, ConfigDict, Field
-from collections.abc import Mapping
-
-from pydantic import BaseModel, Field
-
-logger = logging.getLogger(__name__)
 
 
 class ACPAgentConfig(BaseModel):
     """Configuration for a single ACP-compatible agent."""
 
+    model_config = ConfigDict(frozen=True)
+
     command: str = Field(description="Command to launch the ACP agent subprocess")
     args: list[str] = Field(default_factory=list, description="Additional command arguments")
     env: dict[str, str] = Field(default_factory=dict, description="Environment variables to inject into the agent subprocess. Values starting with $ are resolved from host environment variables.")
@@ -24,28 +21,3 @@ class ACPAgentConfig(BaseModel):
             "are denied — the agent must be configured to operate without requesting permissions."
         ),
     )
-
-
-_acp_agents: dict[str, ACPAgentConfig] = {}
-
-
-def get_acp_agents() -> dict[str, ACPAgentConfig]:
-    """Get the currently configured ACP agents.
-
-    Returns:
-        Mapping of agent name -> ACPAgentConfig.  Empty dict if no ACP agents are configured.
-    """
-    return _acp_agents
-
-
-def load_acp_config_from_dict(config_dict: Mapping[str, Mapping[str, object]] | None) -> None:
-    """Load ACP agent configuration from a dictionary (typically from config.yaml).
-
-    Args:
-        config_dict: Mapping of agent name -> config fields.
-    """
-    global _acp_agents
-    if config_dict is None:
-        config_dict = {}
-    _acp_agents = {name: ACPAgentConfig(**cfg) for name, cfg in config_dict.items()}
-    logger.info("ACP config loaded: %d agent(s): %s", len(_acp_agents), list(_acp_agents.keys()))

backend/packages/harness/deerflow/config/agents_config.py

@@ -5,7 +5,7 @@ import re
 from typing import Any
 
 import yaml
-from pydantic import BaseModel
+from pydantic import BaseModel, ConfigDict
 
 from deerflow.config.paths import get_paths
 
@@ -18,6 +18,8 @@ AGENT_NAME_PATTERN = re.compile(r"^[A-Za-z0-9-]+$")
 class AgentConfig(BaseModel):
     """Configuration for a custom agent."""
 
+    model_config = ConfigDict(frozen=True)
+
     name: str
     description: str = ""
     model: str | None = None

backend/packages/harness/deerflow/config/app_config.py

@@ -1,31 +1,31 @@
+from __future__ import annotations
+
 import logging
 import os
 from contextvars import ContextVar
 from pathlib import Path
-from typing import Any, Self
+from typing import Any, ClassVar, Self
 
 import yaml
 from dotenv import load_dotenv
 from pydantic import BaseModel, ConfigDict, Field
 
-from deerflow.config.acp_config import load_acp_config_from_dict
+from deerflow.config.acp_config import ACPAgentConfig
-from deerflow.config.checkpointer_config import CheckpointerConfig, load_checkpointer_config_from_dict
+from deerflow.config.checkpointer_config import CheckpointerConfig
-from deerflow.config.database_config import DatabaseConfig
 from deerflow.config.extensions_config import ExtensionsConfig
-from deerflow.config.guardrails_config import GuardrailsConfig, load_guardrails_config_from_dict
+from deerflow.config.guardrails_config import GuardrailsConfig
-from deerflow.config.memory_config import MemoryConfig, load_memory_config_from_dict
+from deerflow.config.memory_config import MemoryConfig
 from deerflow.config.model_config import ModelConfig
-from deerflow.config.run_events_config import RunEventsConfig
 from deerflow.config.sandbox_config import SandboxConfig
 from deerflow.config.skill_evolution_config import SkillEvolutionConfig
 from deerflow.config.skills_config import SkillsConfig
-from deerflow.config.stream_bridge_config import StreamBridgeConfig, load_stream_bridge_config_from_dict
+from deerflow.config.stream_bridge_config import StreamBridgeConfig
-from deerflow.config.subagents_config import SubagentsAppConfig, load_subagents_config_from_dict
+from deerflow.config.subagents_config import SubagentsAppConfig
-from deerflow.config.summarization_config import SummarizationConfig, load_summarization_config_from_dict
+from deerflow.config.summarization_config import SummarizationConfig
-from deerflow.config.title_config import TitleConfig, load_title_config_from_dict
+from deerflow.config.title_config import TitleConfig
 from deerflow.config.token_usage_config import TokenUsageConfig
 from deerflow.config.tool_config import ToolConfig, ToolGroupConfig
-from deerflow.config.tool_search_config import ToolSearchConfig, load_tool_search_config_from_dict
+from deerflow.config.tool_search_config import ToolSearchConfig
 
 load_dotenv()
 
@@ -57,11 +57,10 @@ class AppConfig(BaseModel):
     memory: MemoryConfig = Field(default_factory=MemoryConfig, description="Memory subsystem configuration")
     subagents: SubagentsAppConfig = Field(default_factory=SubagentsAppConfig, description="Subagent runtime configuration")
     guardrails: GuardrailsConfig = Field(default_factory=GuardrailsConfig, description="Guardrail middleware configuration")
-    model_config = ConfigDict(extra="allow", frozen=False)
+    model_config = ConfigDict(extra="allow", frozen=True)
-    database: DatabaseConfig = Field(default_factory=DatabaseConfig, description="Unified database backend configuration")
-    run_events: RunEventsConfig = Field(default_factory=RunEventsConfig, description="Run event storage configuration")
     checkpointer: CheckpointerConfig | None = Field(default=None, description="Checkpointer configuration")
     stream_bridge: StreamBridgeConfig | None = Field(default=None, description="Stream bridge configuration")
+    acp_agents: dict[str, ACPAgentConfig] = Field(default_factory=dict, description="ACP agent configurations keyed by agent name")
 
     @classmethod
     def resolve_config_path(cls, config_path: str | None = None) -> Path:
@@ -109,41 +108,6 @@ class AppConfig(BaseModel):
 
         config_data = cls.resolve_env_variables(config_data)
 
-        # Load title config if present
-        if "title" in config_data:
-            load_title_config_from_dict(config_data["title"])
-
-        # Load summarization config if present
-        if "summarization" in config_data:
-            load_summarization_config_from_dict(config_data["summarization"])
-
-        # Load memory config if present
-        if "memory" in config_data:
-            load_memory_config_from_dict(config_data["memory"])
-
-        # Load subagents config if present
-        if "subagents" in config_data:
-            load_subagents_config_from_dict(config_data["subagents"])
-
-        # Load tool_search config if present
-        if "tool_search" in config_data:
-            load_tool_search_config_from_dict(config_data["tool_search"])
-
-        # Load guardrails config if present
-        if "guardrails" in config_data:
-            load_guardrails_config_from_dict(config_data["guardrails"])
-
-        # Load checkpointer config if present
-        if "checkpointer" in config_data:
-            load_checkpointer_config_from_dict(config_data["checkpointer"])
-
-        # Load stream bridge config if present
-        if "stream_bridge" in config_data:
-            load_stream_bridge_config_from_dict(config_data["stream_bridge"])
-
-        # Always refresh ACP agent config so removed entries do not linger across reloads.
-        load_acp_config_from_dict(config_data.get("acp_agents", {}))
-
         # Load extensions config separately (it's in a different file)
         extensions_config = ExtensionsConfig.from_file()
         config_data["extensions"] = extensions_config.model_dump()
@@ -254,130 +218,26 @@ class AppConfig(BaseModel):
         """
         return next((group for group in self.tool_groups if group.name == name), None)
 
+    # -- Lifecycle (class-level singleton via ContextVar) --
 
-_app_config: AppConfig | None = None
+    _current: ClassVar[ContextVar[AppConfig]] = ContextVar("deerflow_app_config")
-_app_config_path: Path | None = None
-_app_config_mtime: float | None = None
-_app_config_is_custom = False
-_current_app_config: ContextVar[AppConfig | None] = ContextVar("deerflow_current_app_config", default=None)
-_current_app_config_stack: ContextVar[tuple[AppConfig | None, ...]] = ContextVar("deerflow_current_app_config_stack", default=())
 
+    @classmethod
+    def init(cls, config: AppConfig) -> None:
+        """Set the AppConfig for the current context. Call once at process startup."""
+        cls._current.set(config)
 
-def _get_config_mtime(config_path: Path) -> float | None:
+    @classmethod
-    """Get the modification time of a config file if it exists."""
+    def current(cls) -> AppConfig:
-    try:
+        """Get the current AppConfig.
-        return config_path.stat().st_mtime
-    except OSError:
-        return None
 
-
+        Auto-initializes from config file on first access for backward compatibility.
-def _load_and_cache_app_config(config_path: str | None = None) -> AppConfig:
+        Prefer calling AppConfig.init() explicitly at process startup.
-    """Load config from disk and refresh cache metadata."""
+        """
-    global _app_config, _app_config_path, _app_config_mtime, _app_config_is_custom
+        try:
-
+            return cls._current.get()
-    resolved_path = AppConfig.resolve_config_path(config_path)
+        except LookupError:
-    _app_config = AppConfig.from_file(str(resolved_path))
+            logger.debug("AppConfig not initialized, auto-loading from file")
-    _app_config_path = resolved_path
+            config = cls.from_file()
-    _app_config_mtime = _get_config_mtime(resolved_path)
+            cls._current.set(config)
-    _app_config_is_custom = False
+            return config
-    return _app_config
-
-
-def get_app_config() -> AppConfig:
-    """Get the DeerFlow config instance.
-
-    Returns a cached singleton instance and automatically reloads it when the
-    underlying config file path or modification time changes. Use
-    `reload_app_config()` to force a reload, or `reset_app_config()` to clear
-    the cache.
-    """
-    global _app_config, _app_config_path, _app_config_mtime
-
-    runtime_override = _current_app_config.get()
-    if runtime_override is not None:
-        return runtime_override
-
-    if _app_config is not None and _app_config_is_custom:
-        return _app_config
-
-    resolved_path = AppConfig.resolve_config_path()
-    current_mtime = _get_config_mtime(resolved_path)
-
-    should_reload = _app_config is None or _app_config_path != resolved_path or _app_config_mtime != current_mtime
-    if should_reload:
-        if _app_config_path == resolved_path and _app_config_mtime is not None and current_mtime is not None and _app_config_mtime != current_mtime:
-            logger.info(
-                "Config file has been modified (mtime: %s -> %s), reloading AppConfig",
-                _app_config_mtime,
-                current_mtime,
-            )
-        _load_and_cache_app_config(str(resolved_path))
-    return _app_config
-
-
-def reload_app_config(config_path: str | None = None) -> AppConfig:
-    """Reload the config from file and update the cached instance.
-
-    This is useful when the config file has been modified and you want
-    to pick up the changes without restarting the application.
-
-    Args:
-        config_path: Optional path to config file. If not provided,
-                     uses the default resolution strategy.
-
-    Returns:
-        The newly loaded AppConfig instance.
-    """
-    return _load_and_cache_app_config(config_path)
-
-
-def reset_app_config() -> None:
-    """Reset the cached config instance.
-
-    This clears the singleton cache, causing the next call to
-    `get_app_config()` to reload from file. Useful for testing
-    or when switching between different configurations.
-    """
-    global _app_config, _app_config_path, _app_config_mtime, _app_config_is_custom
-    _app_config = None
-    _app_config_path = None
-    _app_config_mtime = None
-    _app_config_is_custom = False
-
-
-def set_app_config(config: AppConfig) -> None:
-    """Set a custom config instance.
-
-    This allows injecting a custom or mock config for testing purposes.
-
-    Args:
-        config: The AppConfig instance to use.
-    """
-    global _app_config, _app_config_path, _app_config_mtime, _app_config_is_custom
-    _app_config = config
-    _app_config_path = None
-    _app_config_mtime = None
-    _app_config_is_custom = True
-
-
-def peek_current_app_config() -> AppConfig | None:
-    """Return the runtime-scoped AppConfig override, if one is active."""
-    return _current_app_config.get()
-
-
-def push_current_app_config(config: AppConfig) -> None:
-    """Push a runtime-scoped AppConfig override for the current execution context."""
-    stack = _current_app_config_stack.get()
-    _current_app_config_stack.set(stack + (_current_app_config.get(),))
-    _current_app_config.set(config)
-
-
-def pop_current_app_config() -> None:
-    """Pop the latest runtime-scoped AppConfig override for the current execution context."""
-    stack = _current_app_config_stack.get()
-    if not stack:
-        _current_app_config.set(None)
-        return
-    previous = stack[-1]
-    _current_app_config_stack.set(stack[:-1])
-    _current_app_config.set(previous)

backend/packages/harness/deerflow/config/checkpointer_config.py

@@ -2,7 +2,7 @@
 
 from typing import Literal
 
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, ConfigDict, Field
 
 CheckpointerType = Literal["memory", "sqlite", "postgres"]
 
@@ -10,6 +10,8 @@ CheckpointerType = Literal["memory", "sqlite", "postgres"]
 class CheckpointerConfig(BaseModel):
     """Configuration for LangGraph state persistence checkpointer."""
 
+    model_config = ConfigDict(frozen=True)
+
     type: CheckpointerType = Field(
         description="Checkpointer backend type. "
         "'memory' is in-process only (lost on restart). "
@@ -23,24 +25,3 @@ class CheckpointerConfig(BaseModel):
         "For sqlite, use a file path like '.deer-flow/checkpoints.db' or ':memory:' for in-memory. "
         "For postgres, use a DSN like 'postgresql://user:pass@localhost:5432/db'.",
     )
-
-
-# Global configuration instance — None means no checkpointer is configured.
-_checkpointer_config: CheckpointerConfig | None = None
-
-
-def get_checkpointer_config() -> CheckpointerConfig | None:
-    """Get the current checkpointer configuration, or None if not configured."""
-    return _checkpointer_config
-
-
-def set_checkpointer_config(config: CheckpointerConfig | None) -> None:
-    """Set the checkpointer configuration."""
-    global _checkpointer_config
-    _checkpointer_config = config
-
-
-def load_checkpointer_config_from_dict(config_dict: dict) -> None:
-    """Load checkpointer configuration from a dictionary."""
-    global _checkpointer_config
-    _checkpointer_config = CheckpointerConfig(**config_dict)

backend/packages/harness/deerflow/config/database_config.py

@@ -1,92 +0,0 @@
-"""Unified database backend configuration.
-
-Controls BOTH the LangGraph checkpointer and the DeerFlow application
-persistence layer (runs, threads metadata, users, etc.). The user
-configures one backend; the system handles physical separation details.
-
-SQLite mode: checkpointer and app use different .db files in the same
-directory to avoid write-lock contention. This is automatic.
-
-Postgres mode: both use the same database URL but maintain independent
-connection pools with different lifecycles.
-
-Memory mode: checkpointer uses MemorySaver, app uses in-memory stores.
-No database is initialized.
-
-Sensitive values (postgres_url) should use $VAR syntax in config.yaml
-to reference environment variables from .env:
-
-    database:
-      backend: postgres
-      postgres_url: $DATABASE_URL
-
-The $VAR resolution is handled by AppConfig.resolve_env_variables()
-before this config is instantiated -- DatabaseConfig itself does not
-need to do any environment variable processing.
-"""
-
-from __future__ import annotations
-
-import os
-from typing import Literal
-
-from pydantic import BaseModel, Field
-
-
-class DatabaseConfig(BaseModel):
-    backend: Literal["memory", "sqlite", "postgres"] = Field(
-        default="memory",
-        description=("Storage backend for both checkpointer and application data. 'memory' for development (no persistence across restarts), 'sqlite' for single-node deployment, 'postgres' for production multi-node deployment."),
-    )
-    sqlite_dir: str = Field(
-        default=".deer-flow/data",
-        description=("Directory for SQLite database files. Checkpointer uses {sqlite_dir}/checkpoints.db, application data uses {sqlite_dir}/app.db."),
-    )
-    postgres_url: str = Field(
-        default="",
-        description=(
-            "PostgreSQL connection URL, shared by checkpointer and app. "
-            "Use $DATABASE_URL in config.yaml to reference .env. "
-            "Example: postgresql://user:pass@host:5432/deerflow "
-            "(the +asyncpg driver suffix is added automatically where needed)."
-        ),
-    )
-    echo_sql: bool = Field(
-        default=False,
-        description="Echo all SQL statements to log (debug only).",
-    )
-    pool_size: int = Field(
-        default=5,
-        description="Connection pool size for the app ORM engine (postgres only).",
-    )
-
-    # -- Derived helpers (not user-configured) --
-
-    @property
-    def _resolved_sqlite_dir(self) -> str:
-        """Resolve sqlite_dir to an absolute path (relative to CWD)."""
-        from pathlib import Path
-
-        return str(Path(self.sqlite_dir).resolve())
-
-    @property
-    def checkpointer_sqlite_path(self) -> str:
-        """SQLite file path for the LangGraph checkpointer."""
-        return os.path.join(self._resolved_sqlite_dir, "checkpoints.db")
-
-    @property
-    def app_sqlite_path(self) -> str:
-        """SQLite file path for application ORM data."""
-        return os.path.join(self._resolved_sqlite_dir, "app.db")
-
-    @property
-    def app_sqlalchemy_url(self) -> str:
-        """SQLAlchemy async URL for the application ORM engine."""
-        if self.backend == "sqlite":
-            return f"sqlite+aiosqlite:///{self.app_sqlite_path}"
-        if self.backend == "postgres":
-            url = self.postgres_url
-            if url.startswith("postgresql://"):
-                url = url.replace("postgresql://", "postgresql+asyncpg://", 1)
-            return url
-        raise ValueError(f"No SQLAlchemy URL for backend={self.backend!r}")

backend/packages/harness/deerflow/config/deer_flow_context.py

@@ -0,0 +1,59 @@
+"""Per-invocation context for DeerFlow agent execution.
+
+Injected via LangGraph Runtime. Middleware and tools access this
+via Runtime[DeerFlowContext] parameters, through resolve_context().
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any
+
+
+@dataclass(frozen=True)
+class DeerFlowContext:
+    """Typed, immutable, per-invocation context injected via LangGraph Runtime.
+
+    Fields are all known at run start and never change during execution.
+    Mutable runtime state (e.g. sandbox_id) flows through ThreadState, not here.
+    """
+
+    app_config: Any  # AppConfig — typed as Any to avoid circular import at module level
+    thread_id: str
+    agent_name: str | None = None
+
+
+def resolve_context(runtime: Any) -> DeerFlowContext:
+    """Extract or construct DeerFlowContext from runtime.
+
+    Gateway/Client paths: runtime.context is already DeerFlowContext → return directly.
+    LangGraph Server / legacy dict path: construct from dict context or configurable fallback.
+    """
+    ctx = getattr(runtime, "context", None)
+    if isinstance(ctx, DeerFlowContext):
+        return ctx
+
+    from deerflow.config.app_config import AppConfig
+
+    # Try dict context first (legacy path, tests), then configurable
+    if isinstance(ctx, dict):
+        return DeerFlowContext(
+            app_config=AppConfig.current(),
+            thread_id=ctx.get("thread_id", ""),
+            agent_name=ctx.get("agent_name"),
+        )
+
+    # No context at all — fall back to LangGraph configurable
+    try:
+        from langgraph.config import get_config
+
+        cfg = get_config().get("configurable", {})
+    except RuntimeError:
+        # Outside runnable context (e.g. unit tests)
+        cfg = {}
+
+    return DeerFlowContext(
+        app_config=AppConfig.current(),
+        thread_id=cfg.get("thread_id", ""),
+        agent_name=cfg.get("agent_name"),
+    )

backend/packages/harness/deerflow/config/extensions_config.py

@@ -11,6 +11,8 @@ from pydantic import BaseModel, ConfigDict, Field
 class McpOAuthConfig(BaseModel):
     """OAuth configuration for an MCP server (HTTP/SSE transports)."""
 
+    model_config = ConfigDict(extra="allow", frozen=True)
+
     enabled: bool = Field(default=True, description="Whether OAuth token injection is enabled")
     token_url: str = Field(description="OAuth token endpoint URL")
     grant_type: Literal["client_credentials", "refresh_token"] = Field(
@@ -28,12 +30,13 @@ class McpOAuthConfig(BaseModel):
     default_token_type: str = Field(default="Bearer", description="Default token type when missing in token response")
     refresh_skew_seconds: int = Field(default=60, description="Refresh token this many seconds before expiry")
     extra_token_params: dict[str, str] = Field(default_factory=dict, description="Additional form params sent to token endpoint")
-    model_config = ConfigDict(extra="allow")
 
 
 class McpServerConfig(BaseModel):
     """Configuration for a single MCP server."""
 
+    model_config = ConfigDict(extra="allow", frozen=True)
+
     enabled: bool = Field(default=True, description="Whether this MCP server is enabled")
     type: str = Field(default="stdio", description="Transport type: 'stdio', 'sse', or 'http'")
     command: str | None = Field(default=None, description="Command to execute to start the MCP server (for stdio type)")
@@ -43,12 +46,13 @@ class McpServerConfig(BaseModel):
     headers: dict[str, str] = Field(default_factory=dict, description="HTTP headers to send (for sse or http type)")
     oauth: McpOAuthConfig | None = Field(default=None, description="OAuth configuration (for sse or http type)")
     description: str = Field(default="", description="Human-readable description of what this MCP server provides")
-    model_config = ConfigDict(extra="allow")
 
 
 class SkillStateConfig(BaseModel):
     """Configuration for a single skill's state."""
 
+    model_config = ConfigDict(frozen=True)
+
     enabled: bool = Field(default=True, description="Whether this skill is enabled")
 
 
@@ -64,7 +68,7 @@ class ExtensionsConfig(BaseModel):
         default_factory=dict,
         description="Map of skill name to state configuration",
     )
-    model_config = ConfigDict(extra="allow", populate_by_name=True)
+    model_config = ConfigDict(extra="allow", frozen=True, populate_by_name=True)
 
     @classmethod
     def resolve_config_path(cls, config_path: str | None = None) -> Path | None:
@@ -195,62 +199,3 @@ class ExtensionsConfig(BaseModel):
             # Default to enable for public & custom skill
             return skill_category in ("public", "custom")
         return skill_config.enabled
-
-
-_extensions_config: ExtensionsConfig | None = None
-
-
-def get_extensions_config() -> ExtensionsConfig:
-    """Get the extensions config instance.
-
-    Returns a cached singleton instance. Use `reload_extensions_config()` to reload
-    from file, or `reset_extensions_config()` to clear the cache.
-
-    Returns:
-        The cached ExtensionsConfig instance.
-    """
-    global _extensions_config
-    if _extensions_config is None:
-        _extensions_config = ExtensionsConfig.from_file()
-    return _extensions_config
-
-
-def reload_extensions_config(config_path: str | None = None) -> ExtensionsConfig:
-    """Reload the extensions config from file and update the cached instance.
-
-    This is useful when the config file has been modified and you want
-    to pick up the changes without restarting the application.
-
-    Args:
-        config_path: Optional path to extensions config file. If not provided,
-                     uses the default resolution strategy.
-
-    Returns:
-        The newly loaded ExtensionsConfig instance.
-    """
-    global _extensions_config
-    _extensions_config = ExtensionsConfig.from_file(config_path)
-    return _extensions_config
-
-
-def reset_extensions_config() -> None:
-    """Reset the cached extensions config instance.
-
-    This clears the singleton cache, causing the next call to
-    `get_extensions_config()` to reload from file. Useful for testing
-    or when switching between different configurations.
-    """
-    global _extensions_config
-    _extensions_config = None
-
-
-def set_extensions_config(config: ExtensionsConfig) -> None:
-    """Set a custom extensions config instance.
-
-    This allows injecting a custom or mock config for testing purposes.
-
-    Args:
-        config: The ExtensionsConfig instance to use.
-    """
-    global _extensions_config
-    _extensions_config = config

backend/packages/harness/deerflow/config/guardrails_config.py

@@ -1,11 +1,13 @@
 """Configuration for pre-tool-call authorization."""
 
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, ConfigDict, Field
 
 
 class GuardrailProviderConfig(BaseModel):
     """Configuration for a guardrail provider."""
 
+    model_config = ConfigDict(frozen=True)
+
     use: str = Field(description="Class path (e.g. 'deerflow.guardrails.builtin:AllowlistProvider')")
     config: dict = Field(default_factory=dict, description="Provider-specific settings passed as kwargs")
 
@@ -18,31 +20,9 @@ class GuardrailsConfig(BaseModel):
     agent's passport reference, and returns an allow/deny decision.
     """
 
+    model_config = ConfigDict(frozen=True)
+
     enabled: bool = Field(default=False, description="Enable guardrail middleware")
     fail_closed: bool = Field(default=True, description="Block tool calls if provider errors")
     passport: str | None = Field(default=None, description="OAP passport path or hosted agent ID")
     provider: GuardrailProviderConfig | None = Field(default=None, description="Guardrail provider configuration")
-
-
-_guardrails_config: GuardrailsConfig | None = None
-
-
-def get_guardrails_config() -> GuardrailsConfig:
-    """Get the guardrails config, returning defaults if not loaded."""
-    global _guardrails_config
-    if _guardrails_config is None:
-        _guardrails_config = GuardrailsConfig()
-    return _guardrails_config
-
-
-def load_guardrails_config_from_dict(data: dict) -> GuardrailsConfig:
-    """Load guardrails config from a dict (called during AppConfig loading)."""
-    global _guardrails_config
-    _guardrails_config = GuardrailsConfig.model_validate(data)
-    return _guardrails_config
-
-
-def reset_guardrails_config() -> None:
-    """Reset the cached config instance. Used in tests to prevent singleton leaks."""
-    global _guardrails_config
-    _guardrails_config = None

backend/packages/harness/deerflow/config/memory_config.py

@@ -1,11 +1,13 @@
 """Configuration for memory mechanism."""
 
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, ConfigDict, Field
 
 
 class MemoryConfig(BaseModel):
     """Configuration for global memory mechanism."""
 
+    model_config = ConfigDict(frozen=True)
+
     enabled: bool = Field(
         default=True,
         description="Whether to enable memory mechanism",
@@ -59,24 +61,3 @@ class MemoryConfig(BaseModel):
         le=8000,
         description="Maximum tokens to use for memory injection",
     )
-
-
-# Global configuration instance
-_memory_config: MemoryConfig = MemoryConfig()
-
-
-def get_memory_config() -> MemoryConfig:
-    """Get the current memory configuration."""
-    return _memory_config
-
-
-def set_memory_config(config: MemoryConfig) -> None:
-    """Set the memory configuration."""
-    global _memory_config
-    _memory_config = config
-
-
-def load_memory_config_from_dict(config_dict: dict) -> None:
-    """Load memory configuration from a dictionary."""
-    global _memory_config
-    _memory_config = MemoryConfig(**config_dict)

backend/packages/harness/deerflow/config/model_config.py

@@ -12,7 +12,7 @@ class ModelConfig(BaseModel):
         description="Class path of the model provider(e.g. langchain_openai.ChatOpenAI)",
     )
     model: str = Field(..., description="Model name")
-    model_config = ConfigDict(extra="allow")
+    model_config = ConfigDict(extra="allow", frozen=True)
     use_responses_api: bool | None = Field(
         default=None,
         description="Whether to route OpenAI ChatOpenAI calls through the /v1/responses API",

backend/packages/harness/deerflow/config/run_events_config.py

@@ -1,33 +0,0 @@
-"""Run event storage configuration.
-
-Controls where run events (messages + execution traces) are persisted.
-
-Backends:
-- memory: In-memory storage, data lost on restart. Suitable for
-  development and testing.
-- db: SQL database via SQLAlchemy ORM. Provides full query capability.
-  Suitable for production deployments.
-- jsonl: Append-only JSONL files. Lightweight alternative for
-  single-node deployments that need persistence without a database.
-"""
-
-from __future__ import annotations
-
-from typing import Literal
-
-from pydantic import BaseModel, Field
-
-
-class RunEventsConfig(BaseModel):
-    backend: Literal["memory", "db", "jsonl"] = Field(
-        default="memory",
-        description="Storage backend for run events. 'memory' for development (no persistence), 'db' for production (SQL queries), 'jsonl' for lightweight single-node persistence.",
-    )
-    max_trace_content: int = Field(
-        default=10240,
-        description="Maximum trace content size in bytes before truncation (db backend only).",
-    )
-    track_token_usage: bool = Field(
-        default=True,
-        description="Whether RunJournal should accumulate token counts to RunRow.",
-    )

backend/packages/harness/deerflow/config/sandbox_config.py

@@ -4,6 +4,8 @@ from pydantic import BaseModel, ConfigDict, Field
 class VolumeMountConfig(BaseModel):
     """Configuration for a volume mount."""
 
+    model_config = ConfigDict(frozen=True)
+
     host_path: str = Field(..., description="Path on the host machine")
     container_path: str = Field(..., description="Path inside the container")
     read_only: bool = Field(default=False, description="Whether the mount is read-only")
@@ -80,4 +82,4 @@ class SandboxConfig(BaseModel):
         description="Maximum characters to keep from ls tool output. Output exceeding this limit is head-truncated. Set to 0 to disable truncation.",
     )
 
-    model_config = ConfigDict(extra="allow")
+    model_config = ConfigDict(extra="allow", frozen=True)

backend/packages/harness/deerflow/config/skill_evolution_config.py

@@ -1,9 +1,11 @@
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, ConfigDict, Field
 
 
 class SkillEvolutionConfig(BaseModel):
     """Configuration for agent-managed skill evolution."""
 
+    model_config = ConfigDict(frozen=True)
+
     enabled: bool = Field(
         default=False,
         description="Whether the agent can create and modify skills under skills/custom.",

backend/packages/harness/deerflow/config/skills_config.py

@@ -1,6 +1,6 @@
 from pathlib import Path
 
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, ConfigDict, Field
 
 
 def _default_repo_root() -> Path:
@@ -11,6 +11,8 @@ def _default_repo_root() -> Path:
 class SkillsConfig(BaseModel):
     """Configuration for skills system"""
 
+    model_config = ConfigDict(frozen=True)
+
     path: str | None = Field(
         default=None,
         description="Path to skills directory. If not specified, defaults to ../skills relative to backend directory",

backend/packages/harness/deerflow/config/stream_bridge_config.py

@@ -2,7 +2,7 @@
 
 from typing import Literal
 
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, ConfigDict, Field
 
 StreamBridgeType = Literal["memory", "redis"]
 
@@ -10,6 +10,8 @@ StreamBridgeType = Literal["memory", "redis"]
 class StreamBridgeConfig(BaseModel):
     """Configuration for the stream bridge that connects agent workers to SSE endpoints."""
 
+    model_config = ConfigDict(frozen=True)
+
     type: StreamBridgeType = Field(
         default="memory",
         description="Stream bridge backend type. 'memory' uses in-process asyncio.Queue (single-process only). 'redis' uses Redis Streams (planned for Phase 2, not yet implemented).",
@@ -22,25 +24,3 @@ class StreamBridgeConfig(BaseModel):
         default=256,
         description="Maximum number of events buffered per run in the memory bridge.",
     )
-
-
-# Global configuration instance — None means no stream bridge is configured
-# (falls back to memory with defaults).
-_stream_bridge_config: StreamBridgeConfig | None = None
-
-
-def get_stream_bridge_config() -> StreamBridgeConfig | None:
-    """Get the current stream bridge configuration, or None if not configured."""
-    return _stream_bridge_config
-
-
-def set_stream_bridge_config(config: StreamBridgeConfig | None) -> None:
-    """Set the stream bridge configuration."""
-    global _stream_bridge_config
-    _stream_bridge_config = config
-
-
-def load_stream_bridge_config_from_dict(config_dict: dict) -> None:
-    """Load stream bridge configuration from a dictionary."""
-    global _stream_bridge_config
-    _stream_bridge_config = StreamBridgeConfig(**config_dict)

backend/packages/harness/deerflow/config/subagents_config.py

@@ -1,15 +1,13 @@
 """Configuration for the subagent system loaded from config.yaml."""
 
-import logging
+from pydantic import BaseModel, ConfigDict, Field
-
-from pydantic import BaseModel, Field
-
-logger = logging.getLogger(__name__)
 
 
 class SubagentOverrideConfig(BaseModel):
     """Per-agent configuration overrides."""
 
+    model_config = ConfigDict(frozen=True)
+
     timeout_seconds: int | None = Field(
         default=None,
         ge=1,
@@ -25,6 +23,8 @@ class SubagentOverrideConfig(BaseModel):
 class SubagentsAppConfig(BaseModel):
     """Configuration for the subagent system."""
 
+    model_config = ConfigDict(frozen=True)
+
     timeout_seconds: int = Field(
         default=900,
         ge=1,
@@ -62,41 +62,3 @@ class SubagentsAppConfig(BaseModel):
         if self.max_turns is not None:
             return self.max_turns
         return builtin_default
-
-
-_subagents_config: SubagentsAppConfig = SubagentsAppConfig()
-
-
-def get_subagents_app_config() -> SubagentsAppConfig:
-    """Get the current subagents configuration."""
-    return _subagents_config
-
-
-def load_subagents_config_from_dict(config_dict: dict) -> None:
-    """Load subagents configuration from a dictionary."""
-    global _subagents_config
-    _subagents_config = SubagentsAppConfig(**config_dict)
-
-    overrides_summary = {}
-    for name, override in _subagents_config.agents.items():
-        parts = []
-        if override.timeout_seconds is not None:
-            parts.append(f"timeout={override.timeout_seconds}s")
-        if override.max_turns is not None:
-            parts.append(f"max_turns={override.max_turns}")
-        if parts:
-            overrides_summary[name] = ", ".join(parts)
-
-    if overrides_summary:
-        logger.info(
-            "Subagents config loaded: default timeout=%ss, default max_turns=%s, per-agent overrides=%s",
-            _subagents_config.timeout_seconds,
-            _subagents_config.max_turns,
-            overrides_summary,
-        )
-    else:
-        logger.info(
-            "Subagents config loaded: default timeout=%ss, default max_turns=%s, no per-agent overrides",
-            _subagents_config.timeout_seconds,
-            _subagents_config.max_turns,
-        )

backend/packages/harness/deerflow/config/summarization_config.py

@@ -2,7 +2,7 @@
 
 from typing import Literal
 
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, ConfigDict, Field
 
 ContextSizeType = Literal["fraction", "tokens", "messages"]
 
@@ -10,6 +10,8 @@ ContextSizeType = Literal["fraction", "tokens", "messages"]
 class ContextSize(BaseModel):
     """Context size specification for trigger or keep parameters."""
 
+    model_config = ConfigDict(frozen=True)
+
     type: ContextSizeType = Field(description="Type of context size specification")
     value: int | float = Field(description="Value for the context size specification")
 
@@ -21,6 +23,8 @@ class ContextSize(BaseModel):
 class SummarizationConfig(BaseModel):
     """Configuration for automatic conversation summarization."""
 
+    model_config = ConfigDict(frozen=True)
+
     enabled: bool = Field(
         default=False,
         description="Whether to enable automatic conversation summarization",
@@ -51,24 +55,3 @@ class SummarizationConfig(BaseModel):
         default=None,
         description="Custom prompt template for generating summaries. If not provided, uses the default LangChain prompt.",
     )
-
-
-# Global configuration instance
-_summarization_config: SummarizationConfig = SummarizationConfig()
-
-
-def get_summarization_config() -> SummarizationConfig:
-    """Get the current summarization configuration."""
-    return _summarization_config
-
-
-def set_summarization_config(config: SummarizationConfig) -> None:
-    """Set the summarization configuration."""
-    global _summarization_config
-    _summarization_config = config
-
-
-def load_summarization_config_from_dict(config_dict: dict) -> None:
-    """Load summarization configuration from a dictionary."""
-    global _summarization_config
-    _summarization_config = SummarizationConfig(**config_dict)

backend/packages/harness/deerflow/config/title_config.py

@@ -1,11 +1,13 @@
 """Configuration for automatic thread title generation."""
 
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, ConfigDict, Field
 
 
 class TitleConfig(BaseModel):
     """Configuration for automatic thread title generation."""
 
+    model_config = ConfigDict(frozen=True)
+
     enabled: bool = Field(
         default=True,
         description="Whether to enable automatic title generation",
@@ -30,24 +32,3 @@ class TitleConfig(BaseModel):
         default=("Generate a concise title (max {max_words} words) for this conversation.\nUser: {user_msg}\nAssistant: {assistant_msg}\n\nReturn ONLY the title, no quotes, no explanation."),
         description="Prompt template for title generation",
     )
-
-
-# Global configuration instance
-_title_config: TitleConfig = TitleConfig()
-
-
-def get_title_config() -> TitleConfig:
-    """Get the current title configuration."""
-    return _title_config
-
-
-def set_title_config(config: TitleConfig) -> None:
-    """Set the title configuration."""
-    global _title_config
-    _title_config = config
-
-
-def load_title_config_from_dict(config_dict: dict) -> None:
-    """Load title configuration from a dictionary."""
-    global _title_config
-    _title_config = TitleConfig(**config_dict)

backend/packages/harness/deerflow/config/token_usage_config.py

@@ -1,7 +1,9 @@
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, ConfigDict, Field
 
 
 class TokenUsageConfig(BaseModel):
     """Configuration for token usage tracking."""
 
+    model_config = ConfigDict(frozen=True)
+
     enabled: bool = Field(default=False, description="Enable token usage tracking middleware")

backend/packages/harness/deerflow/config/tool_config.py

@@ -5,7 +5,7 @@ class ToolGroupConfig(BaseModel):
     """Config section for a tool group"""
 
     name: str = Field(..., description="Unique name for the tool group")
-    model_config = ConfigDict(extra="allow")
+    model_config = ConfigDict(extra="allow", frozen=True)
 
 
 class ToolConfig(BaseModel):
@@ -17,4 +17,4 @@ class ToolConfig(BaseModel):
         ...,
         description="Variable name of the tool provider(e.g. deerflow.sandbox.tools:bash_tool)",
     )
-    model_config = ConfigDict(extra="allow")
+    model_config = ConfigDict(extra="allow", frozen=True)

backend/packages/harness/deerflow/config/tool_search_config.py

@@ -1,6 +1,6 @@
 """Configuration for deferred tool loading via tool_search."""
 
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, ConfigDict, Field
 
 
 class ToolSearchConfig(BaseModel):
@@ -11,25 +11,9 @@ class ToolSearchConfig(BaseModel):
     via the tool_search tool at runtime.
     """
 
+    model_config = ConfigDict(frozen=True)
+
     enabled: bool = Field(
         default=False,
         description="Defer tools and enable tool_search",
     )
-
-
-_tool_search_config: ToolSearchConfig | None = None
-
-
-def get_tool_search_config() -> ToolSearchConfig:
-    """Get the tool search config, loading from AppConfig if needed."""
-    global _tool_search_config
-    if _tool_search_config is None:
-        _tool_search_config = ToolSearchConfig()
-    return _tool_search_config
-
-
-def load_tool_search_config_from_dict(data: dict) -> ToolSearchConfig:
-    """Load tool search config from a dict (called during AppConfig loading)."""
-    global _tool_search_config
-    _tool_search_config = ToolSearchConfig.model_validate(data)
-    return _tool_search_config

backend/packages/harness/deerflow/config/tracing_config.py

@@ -1,7 +1,7 @@
 import os
 import threading
 
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, ConfigDict, Field
 
 _config_lock = threading.Lock()
 
@@ -9,6 +9,8 @@ _config_lock = threading.Lock()
 class LangSmithTracingConfig(BaseModel):
     """Configuration for LangSmith tracing."""
 
+    model_config = ConfigDict(frozen=True)
+
     enabled: bool = Field(...)
     api_key: str | None = Field(...)
     project: str = Field(...)
@@ -26,6 +28,8 @@ class LangSmithTracingConfig(BaseModel):
 class LangfuseTracingConfig(BaseModel):
     """Configuration for Langfuse tracing."""
 
+    model_config = ConfigDict(frozen=True)
+
     enabled: bool = Field(...)
     public_key: str | None = Field(...)
     secret_key: str | None = Field(...)
@@ -50,6 +54,8 @@ class LangfuseTracingConfig(BaseModel):
 class TracingConfig(BaseModel):
     """Tracing configuration for supported providers."""
 
+    model_config = ConfigDict(frozen=True)
+
     langsmith: LangSmithTracingConfig = Field(...)
     langfuse: LangfuseTracingConfig = Field(...)
 

backend/packages/harness/deerflow/models/factory.py

@@ -2,7 +2,7 @@ import logging
 
 from langchain.chat_models import BaseChatModel
 
-from deerflow.config import get_app_config
+from deerflow.config.app_config import AppConfig
 from deerflow.reflection import resolve_class
 from deerflow.tracing import build_tracing_callbacks
 
@@ -39,7 +39,7 @@ def create_chat_model(name: str | None = None, thinking_enabled: bool = False, *
     Returns:
         A chat model instance.
     """
-    config = get_app_config()
+    config = AppConfig.current()
     if name is None:
         name = config.models[0].name
     model_config = config.get_model_config(name)
@@ -113,16 +113,7 @@ def create_chat_model(name: str | None = None, thinking_enabled: bool = False, *
         elif "reasoning_effort" not in model_settings_from_config:
             model_settings_from_config["reasoning_effort"] = "medium"
 
-    # Ensure stream_usage is enabled so that token usage metadata is available
+    model_instance = model_class(**{**model_settings_from_config, **kwargs})
-    # in streaming responses.  LangChain's BaseChatOpenAI only defaults
-    # stream_usage=True when no custom base_url/api_base is set, so models
-    # hitting third-party endpoints (e.g. doubao, deepseek) silently lose
-    # usage data.  We default it to True unless explicitly configured.
-    if "stream_usage" not in model_settings_from_config and "stream_usage" not in kwargs:
-        if "stream_usage" in getattr(model_class, "model_fields", {}):
-            model_settings_from_config["stream_usage"] = True
-
-    model_instance = model_class(**kwargs, **model_settings_from_config)
 
     callbacks = build_tracing_callbacks()
     if callbacks:

backend/packages/harness/deerflow/persistence/__init__.py

@@ -1,13 +0,0 @@
-"""DeerFlow application persistence layer (SQLAlchemy 2.0 async ORM).
-
-This module manages DeerFlow's own application data -- runs metadata,
-thread ownership, cron jobs, users. It is completely separate from
-LangGraph's checkpointer, which manages graph execution state.
-
-Usage:
-    from deerflow.persistence import init_engine, close_engine, get_session_factory
-"""
-
-from deerflow.persistence.engine import close_engine, get_engine, get_session_factory, init_engine
-
-__all__ = ["close_engine", "get_engine", "get_session_factory", "init_engine"]

backend/packages/harness/deerflow/persistence/base.py

@@ -1,40 +0,0 @@
-"""SQLAlchemy declarative base with automatic to_dict support.
-
-All DeerFlow ORM models inherit from this Base. It provides a generic
-to_dict() method via SQLAlchemy's inspect() so individual models don't
-need to write their own serialization logic.
-
-LangGraph's checkpointer tables are NOT managed by this Base.
-"""
-
-from __future__ import annotations
-
-from sqlalchemy import inspect as sa_inspect
-from sqlalchemy.orm import DeclarativeBase
-
-
-class Base(DeclarativeBase):
-    """Base class for all DeerFlow ORM models.
-
-    Provides:
-    - Automatic to_dict() via SQLAlchemy column inspection.
-    - Standard __repr__() showing all column values.
-    """
-
-    def to_dict(self, *, exclude: set[str] | None = None) -> dict:
-        """Convert ORM instance to plain dict.
-
-        Uses SQLAlchemy's inspect() to iterate mapped column attributes.
-
-        Args:
-            exclude: Optional set of column keys to omit.
-
-        Returns:
-            Dict of {column_key: value} for all mapped columns.
-        """
-        exclude = exclude or set()
-        return {c.key: getattr(self, c.key) for c in sa_inspect(type(self)).mapper.column_attrs if c.key not in exclude}
-
-    def __repr__(self) -> str:
-        cols = ", ".join(f"{c.key}={getattr(self, c.key)!r}" for c in sa_inspect(type(self)).mapper.column_attrs)
-        return f"{type(self).__name__}({cols})"

backend/packages/harness/deerflow/persistence/engine.py

@@ -1,185 +0,0 @@
-"""Async SQLAlchemy engine lifecycle management.
-
-Initializes at Gateway startup, provides session factory for
-repositories, disposes at shutdown.
-
-When database.backend="memory", init_engine is a no-op and
-get_session_factory() returns None. Repositories must check for
-None and fall back to in-memory implementations.
-"""
-
-from __future__ import annotations
-
-import json
-import logging
-
-from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker, create_async_engine
-
-
-def _json_serializer(obj: object) -> str:
-    """JSON serializer with ensure_ascii=False for Chinese character support."""
-    return json.dumps(obj, ensure_ascii=False)
-
-
-logger = logging.getLogger(__name__)
-
-_engine: AsyncEngine | None = None
-_session_factory: async_sessionmaker[AsyncSession] | None = None
-
-
-async def _auto_create_postgres_db(url: str) -> None:
-    """Connect to the ``postgres`` maintenance DB and CREATE DATABASE.
-
-    The target database name is extracted from *url*.  The connection is
-    made to the default ``postgres`` database on the same server using
-    ``AUTOCOMMIT`` isolation (CREATE DATABASE cannot run inside a
-    transaction).
-    """
-    from sqlalchemy import text
-    from sqlalchemy.engine.url import make_url
-
-    parsed = make_url(url)
-    db_name = parsed.database
-    if not db_name:
-        raise ValueError("Cannot auto-create database: no database name in URL")
-
-    # Connect to the default 'postgres' database to issue CREATE DATABASE
-    maint_url = parsed.set(database="postgres")
-    maint_engine = create_async_engine(maint_url, isolation_level="AUTOCOMMIT")
-    try:
-        async with maint_engine.connect() as conn:
-            await conn.execute(text(f'CREATE DATABASE "{db_name}"'))
-        logger.info("Auto-created PostgreSQL database: %s", db_name)
-    finally:
-        await maint_engine.dispose()
-
-
-async def init_engine(
-    backend: str,
-    *,
-    url: str = "",
-    echo: bool = False,
-    pool_size: int = 5,
-    sqlite_dir: str = "",
-) -> None:
-    """Create the async engine and session factory, then auto-create tables.
-
-    Args:
-        backend: "memory", "sqlite", or "postgres".
-        url: SQLAlchemy async URL (for sqlite/postgres).
-        echo: Echo SQL to log.
-        pool_size: Postgres connection pool size.
-        sqlite_dir: Directory to create for SQLite (ensured to exist).
-    """
-    global _engine, _session_factory
-
-    if backend == "memory":
-        logger.info("Persistence backend=memory -- ORM engine not initialized")
-        return
-
-    if backend == "postgres":
-        try:
-            import asyncpg  # noqa: F401
-        except ImportError:
-            raise ImportError("database.backend is set to 'postgres' but asyncpg is not installed.\nInstall it with:\n    uv sync --extra postgres\nOr switch to backend: sqlite in config.yaml for single-node deployment.") from None
-
-    if backend == "sqlite":
-        import os
-
-        from sqlalchemy import event
-
-        os.makedirs(sqlite_dir or ".", exist_ok=True)
-        _engine = create_async_engine(url, echo=echo, json_serializer=_json_serializer)
-
-        # Enable WAL on every new connection. SQLite PRAGMA settings are
-        # per-connection, so we wire the listener instead of running PRAGMA
-        # once at startup. WAL gives concurrent reads + writers without
-        # blocking and is the standard recommendation for any production
-        # SQLite deployment (TC-UPG-06 in AUTH_TEST_PLAN.md). The companion
-        # ``synchronous=NORMAL`` is the safe-and-fast pairing — fsync only
-        # at WAL checkpoint boundaries instead of every commit.
-        @event.listens_for(_engine.sync_engine, "connect")
-        def _enable_sqlite_wal(dbapi_conn, _record):  # noqa: ARG001 — SQLAlchemy contract
-            cursor = dbapi_conn.cursor()
-            try:
-                cursor.execute("PRAGMA journal_mode=WAL;")
-                cursor.execute("PRAGMA synchronous=NORMAL;")
-                cursor.execute("PRAGMA foreign_keys=ON;")
-            finally:
-                cursor.close()
-    elif backend == "postgres":
-        _engine = create_async_engine(
-            url,
-            echo=echo,
-            pool_size=pool_size,
-            pool_pre_ping=True,
-            json_serializer=_json_serializer,
-        )
-    else:
-        raise ValueError(f"Unknown persistence backend: {backend!r}")
-
-    _session_factory = async_sessionmaker(_engine, expire_on_commit=False)
-
-    # Auto-create tables (dev convenience). Production should use Alembic.
-    from deerflow.persistence.base import Base
-
-    # Import all models so Base.metadata discovers them.
-    # When no models exist yet (scaffolding phase), this is a no-op.
-    try:
-        import deerflow.persistence.models  # noqa: F401
-    except ImportError:
-        # Models package not yet available — tables won't be auto-created.
-        # This is expected during initial scaffolding or minimal installs.
-        logger.debug("deerflow.persistence.models not found; skipping auto-create tables")
-
-    try:
-        async with _engine.begin() as conn:
-            await conn.run_sync(Base.metadata.create_all)
-    except Exception as exc:
-        if backend == "postgres" and "does not exist" in str(exc):
-            # Database not yet created — attempt to auto-create it, then retry.
-            await _auto_create_postgres_db(url)
-            # Rebuild engine against the now-existing database
-            await _engine.dispose()
-            _engine = create_async_engine(url, echo=echo, pool_size=pool_size, pool_pre_ping=True, json_serializer=_json_serializer)
-            _session_factory = async_sessionmaker(_engine, expire_on_commit=False)
-            async with _engine.begin() as conn:
-                await conn.run_sync(Base.metadata.create_all)
-        else:
-            raise
-
-    logger.info("Persistence engine initialized: backend=%s", backend)
-
-
-async def init_engine_from_config(config) -> None:
-    """Convenience: init engine from a DatabaseConfig object."""
-    if config.backend == "memory":
-        await init_engine("memory")
-        return
-    await init_engine(
-        backend=config.backend,
-        url=config.app_sqlalchemy_url,
-        echo=config.echo_sql,
-        pool_size=config.pool_size,
-        sqlite_dir=config.sqlite_dir if config.backend == "sqlite" else "",
-    )
-
-
-def get_session_factory() -> async_sessionmaker[AsyncSession] | None:
-    """Return the async session factory, or None if backend=memory."""
-    return _session_factory
-
-
-def get_engine() -> AsyncEngine | None:
-    """Return the async engine, or None if not initialized."""
-    return _engine
-
-
-async def close_engine() -> None:
-    """Dispose the engine, release all connections."""
-    global _engine, _session_factory
-    if _engine is not None:
-        await _engine.dispose()
-        logger.info("Persistence engine closed")
-    _engine = None
-    _session_factory = None

backend/packages/harness/deerflow/persistence/feedback/__init__.py

@@ -1,6 +0,0 @@
-"""Feedback persistence — ORM and SQL repository."""
-
-from deerflow.persistence.feedback.model import FeedbackRow
-from deerflow.persistence.feedback.sql import FeedbackRepository
-
-__all__ = ["FeedbackRepository", "FeedbackRow"]

backend/packages/harness/deerflow/persistence/feedback/model.py

@@ -1,30 +0,0 @@
-"""ORM model for user feedback on runs."""
-
-from __future__ import annotations
-
-from datetime import UTC, datetime
-
-from sqlalchemy import DateTime, String, Text
-from sqlalchemy.orm import Mapped, mapped_column
-
-from deerflow.persistence.base import Base
-
-
-class FeedbackRow(Base):
-    __tablename__ = "feedback"
-
-    feedback_id: Mapped[str] = mapped_column(String(64), primary_key=True)
-    run_id: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
-    thread_id: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
-    owner_id: Mapped[str | None] = mapped_column(String(64), index=True)
-    message_id: Mapped[str | None] = mapped_column(String(64))
-    # message_id is an optional RunEventStore event identifier —
-    # allows feedback to target a specific message or the entire run
-
-    rating: Mapped[int] = mapped_column(nullable=False)
-    # +1 (thumbs-up) or -1 (thumbs-down)
-
-    comment: Mapped[str | None] = mapped_column(Text)
-    # Optional text feedback from the user
-
-    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=lambda: datetime.now(UTC))

backend/packages/harness/deerflow/persistence/feedback/sql.py

@@ -1,139 +0,0 @@
-"""SQLAlchemy-backed feedback storage.
-
-Each method acquires its own short-lived session.
-"""
-
-from __future__ import annotations
-
-import uuid
-from datetime import UTC, datetime
-
-from sqlalchemy import case, func, select
-from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
-
-from deerflow.persistence.feedback.model import FeedbackRow
-from deerflow.runtime.user_context import AUTO, _AutoSentinel, resolve_owner_id
-
-
-class FeedbackRepository:
-    def __init__(self, session_factory: async_sessionmaker[AsyncSession]) -> None:
-        self._sf = session_factory
-
-    @staticmethod
-    def _row_to_dict(row: FeedbackRow) -> dict:
-        d = row.to_dict()
-        val = d.get("created_at")
-        if isinstance(val, datetime):
-            d["created_at"] = val.isoformat()
-        return d
-
-    async def create(
-        self,
-        *,
-        run_id: str,
-        thread_id: str,
-        rating: int,
-        owner_id: str | None | _AutoSentinel = AUTO,
-        message_id: str | None = None,
-        comment: str | None = None,
-    ) -> dict:
-        """Create a feedback record. rating must be +1 or -1."""
-        if rating not in (1, -1):
-            raise ValueError(f"rating must be +1 or -1, got {rating}")
-        resolved_owner_id = resolve_owner_id(owner_id, method_name="FeedbackRepository.create")
-        row = FeedbackRow(
-            feedback_id=str(uuid.uuid4()),
-            run_id=run_id,
-            thread_id=thread_id,
-            owner_id=resolved_owner_id,
-            message_id=message_id,
-            rating=rating,
-            comment=comment,
-            created_at=datetime.now(UTC),
-        )
-        async with self._sf() as session:
-            session.add(row)
-            await session.commit()
-            await session.refresh(row)
-            return self._row_to_dict(row)
-
-    async def get(
-        self,
-        feedback_id: str,
-        *,
-        owner_id: str | None | _AutoSentinel = AUTO,
-    ) -> dict | None:
-        resolved_owner_id = resolve_owner_id(owner_id, method_name="FeedbackRepository.get")
-        async with self._sf() as session:
-            row = await session.get(FeedbackRow, feedback_id)
-            if row is None:
-                return None
-            if resolved_owner_id is not None and row.owner_id != resolved_owner_id:
-                return None
-            return self._row_to_dict(row)
-
-    async def list_by_run(
-        self,
-        thread_id: str,
-        run_id: str,
-        *,
-        limit: int = 100,
-        owner_id: str | None | _AutoSentinel = AUTO,
-    ) -> list[dict]:
-        resolved_owner_id = resolve_owner_id(owner_id, method_name="FeedbackRepository.list_by_run")
-        stmt = select(FeedbackRow).where(FeedbackRow.thread_id == thread_id, FeedbackRow.run_id == run_id)
-        if resolved_owner_id is not None:
-            stmt = stmt.where(FeedbackRow.owner_id == resolved_owner_id)
-        stmt = stmt.order_by(FeedbackRow.created_at.asc()).limit(limit)
-        async with self._sf() as session:
-            result = await session.execute(stmt)
-            return [self._row_to_dict(r) for r in result.scalars()]
-
-    async def list_by_thread(
-        self,
-        thread_id: str,
-        *,
-        limit: int = 100,
-        owner_id: str | None | _AutoSentinel = AUTO,
-    ) -> list[dict]:
-        resolved_owner_id = resolve_owner_id(owner_id, method_name="FeedbackRepository.list_by_thread")
-        stmt = select(FeedbackRow).where(FeedbackRow.thread_id == thread_id)
-        if resolved_owner_id is not None:
-            stmt = stmt.where(FeedbackRow.owner_id == resolved_owner_id)
-        stmt = stmt.order_by(FeedbackRow.created_at.asc()).limit(limit)
-        async with self._sf() as session:
-            result = await session.execute(stmt)
-            return [self._row_to_dict(r) for r in result.scalars()]
-
-    async def delete(
-        self,
-        feedback_id: str,
-        *,
-        owner_id: str | None | _AutoSentinel = AUTO,
-    ) -> bool:
-        resolved_owner_id = resolve_owner_id(owner_id, method_name="FeedbackRepository.delete")
-        async with self._sf() as session:
-            row = await session.get(FeedbackRow, feedback_id)
-            if row is None:
-                return False
-            if resolved_owner_id is not None and row.owner_id != resolved_owner_id:
-                return False
-            await session.delete(row)
-            await session.commit()
-            return True
-
-    async def aggregate_by_run(self, thread_id: str, run_id: str) -> dict:
-        """Aggregate feedback stats for a run using database-side counting."""
-        stmt = select(
-            func.count().label("total"),
-            func.coalesce(func.sum(case((FeedbackRow.rating == 1, 1), else_=0)), 0).label("positive"),
-            func.coalesce(func.sum(case((FeedbackRow.rating == -1, 1), else_=0)), 0).label("negative"),
-        ).where(FeedbackRow.thread_id == thread_id, FeedbackRow.run_id == run_id)
-        async with self._sf() as session:
-            row = (await session.execute(stmt)).one()
-            return {
-                "run_id": run_id,
-                "total": row.total,
-                "positive": row.positive,
-                "negative": row.negative,
-            }

backend/packages/harness/deerflow/persistence/migrations/alembic.ini

@@ -1,38 +0,0 @@
-[alembic]
-script_location = %(here)s
-# Default URL for offline mode / autogenerate.
-# Runtime uses engine from DeerFlow config.
-sqlalchemy.url = sqlite+aiosqlite:///./data/app.db
-
-[loggers]
-keys = root,sqlalchemy,alembic
-
-[handlers]
-keys = console
-
-[formatters]
-keys = generic
-
-[logger_root]
-level = WARN
-handlers = console
-
-[logger_sqlalchemy]
-level = WARN
-handlers =
-qualname = sqlalchemy.engine
-
-[logger_alembic]
-level = INFO
-handlers =
-qualname = alembic
-
-[handler_console]
-class = StreamHandler
-args = (sys.stderr,)
-level = NOTSET
-formatter = generic
-
-[formatter_generic]
-format = %(levelname)-5.5s [%(name)s] %(message)s
-datefmt = %H:%M:%S

backend/packages/harness/deerflow/persistence/migrations/env.py

@@ -1,65 +0,0 @@
-"""Alembic environment for DeerFlow application tables.
-
-ONLY manages DeerFlow's tables (runs, threads_meta, cron_jobs, users).
-LangGraph's checkpointer tables are managed by LangGraph itself -- they
-have their own schema lifecycle and must not be touched by Alembic.
-"""
-
-from __future__ import annotations
-
-import asyncio
-import logging
-from logging.config import fileConfig
-
-from alembic import context
-from sqlalchemy.ext.asyncio import create_async_engine
-
-from deerflow.persistence.base import Base
-
-# Import all models so metadata is populated.
-try:
-    import deerflow.persistence.models  # noqa: F401 — register ORM models with Base.metadata
-except ImportError:
-    # Models not available — migration will work with existing metadata only.
-    logging.getLogger(__name__).warning("Could not import deerflow.persistence.models; Alembic may not detect all tables")
-
-config = context.config
-if config.config_file_name is not None:
-    fileConfig(config.config_file_name)
-
-target_metadata = Base.metadata
-
-
-def run_migrations_offline() -> None:
-    url = config.get_main_option("sqlalchemy.url")
-    context.configure(
-        url=url,
-        target_metadata=target_metadata,
-        literal_binds=True,
-        render_as_batch=True,
-    )
-    with context.begin_transaction():
-        context.run_migrations()
-
-
-def do_run_migrations(connection):
-    context.configure(
-        connection=connection,
-        target_metadata=target_metadata,
-        render_as_batch=True,  # Required for SQLite ALTER TABLE support
-    )
-    with context.begin_transaction():
-        context.run_migrations()
-
-
-async def run_migrations_online() -> None:
-    connectable = create_async_engine(config.get_main_option("sqlalchemy.url"))
-    async with connectable.connect() as connection:
-        await connection.run_sync(do_run_migrations)
-    await connectable.dispose()
-
-
-if context.is_offline_mode():
-    run_migrations_offline()
-else:
-    asyncio.run(run_migrations_online())

backend/packages/harness/deerflow/persistence/models/__init__.py

@@ -1,23 +0,0 @@
-"""ORM model registration entry point.
-
-Importing this module ensures all ORM models are registered with
-``Base.metadata`` so Alembic autogenerate detects every table.
-
-The actual ORM classes have moved to entity-specific subpackages:
-- ``deerflow.persistence.thread_meta``
-- ``deerflow.persistence.run``
-- ``deerflow.persistence.feedback``
-- ``deerflow.persistence.user``
-
-``RunEventRow`` remains in ``deerflow.persistence.models.run_event`` because
-its storage implementation lives in ``deerflow.runtime.events.store.db`` and
-there is no matching entity directory.
-"""
-
-from deerflow.persistence.feedback.model import FeedbackRow
-from deerflow.persistence.models.run_event import RunEventRow
-from deerflow.persistence.run.model import RunRow
-from deerflow.persistence.thread_meta.model import ThreadMetaRow
-from deerflow.persistence.user.model import UserRow
-
-__all__ = ["FeedbackRow", "RunEventRow", "RunRow", "ThreadMetaRow", "UserRow"]

backend/packages/harness/deerflow/persistence/models/run_event.py

@@ -1,35 +0,0 @@
-"""ORM model for run events."""
-
-from __future__ import annotations
-
-from datetime import UTC, datetime
-
-from sqlalchemy import JSON, DateTime, Index, String, Text, UniqueConstraint
-from sqlalchemy.orm import Mapped, mapped_column
-
-from deerflow.persistence.base import Base
-
-
-class RunEventRow(Base):
-    __tablename__ = "run_events"
-
-    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
-    thread_id: Mapped[str] = mapped_column(String(64), nullable=False)
-    run_id: Mapped[str] = mapped_column(String(64), nullable=False)
-    # Owner of the conversation this event belongs to. Nullable for data
-    # created before auth was introduced; populated by auth middleware on
-    # new writes and by the boot-time orphan migration on existing rows.
-    owner_id: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
-    event_type: Mapped[str] = mapped_column(String(32), nullable=False)
-    category: Mapped[str] = mapped_column(String(16), nullable=False)
-    # "message" | "trace" | "lifecycle"
-    content: Mapped[str] = mapped_column(Text, default="")
-    event_metadata: Mapped[dict] = mapped_column(JSON, default=dict)
-    seq: Mapped[int] = mapped_column(nullable=False)
-    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=lambda: datetime.now(UTC))
-
-    __table_args__ = (
-        UniqueConstraint("thread_id", "seq", name="uq_events_thread_seq"),
-        Index("ix_events_thread_cat_seq", "thread_id", "category", "seq"),
-        Index("ix_events_run", "thread_id", "run_id", "seq"),
-    )

backend/packages/harness/deerflow/persistence/run/__init__.py

@@ -1,6 +0,0 @@
-"""Run metadata persistence — ORM and SQL repository."""
-
-from deerflow.persistence.run.model import RunRow
-from deerflow.persistence.run.sql import RunRepository
-
-__all__ = ["RunRepository", "RunRow"]