From d56230e04dc7ad2111e9fcc6edb687746409a29c Mon Sep 17 00:00:00 2001 From: xuhd Date: Thu, 21 May 2026 17:34:05 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E8=AE=B0=E5=BF=86=E9=85=8D=E7=BD=AE?= =?UTF-8?q?=E6=94=B9=E4=B8=BA=E6=95=B0=E6=8D=AE=E5=BA=93=E5=AD=98=E5=82=A8?= =?UTF-8?q?=EF=BC=8C=E7=94=A8=E6=88=B7=E5=8F=AA=E5=8F=AF=E9=85=8D=E7=BD=AE?= =?UTF-8?q?=E5=8F=AC=E5=9B=9E=E6=A8=A1=E5=BC=8F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新增 persistence/memory_config/ 模块：UserMemoryConfigRow ORM 模型 + SqlUserMemoryConfigRepository + MemoryUserMemoryConfigStore（内存兼容） - 新增 Alembic 迁移脚本 20260521_01 创建 user_memory_configs 表 - memory_config.py：USER_WRITABLE_FIELDS 从 3 个缩减为 1 个 (hindsight.recall_budget)，移除文件 IO，改为进程内 cache + async DB 对接函数 - Gateway lifespan 启动时批量预热 recall_budget cache - memory router PATCH/DELETE 改用 DB 存储，支持 400 校验非法 budget 值 Co-Authored-By: Claude Sonnet 4.6 --- .../backend/app/gateway/app.py | 12 ++ .../backend/app/gateway/deps.py | 12 ++ .../backend/app/gateway/routers/memory.py | 30 +-- .../harness/deerflow/config/memory_config.py | 178 +++++++++--------- .../persistence/memory_config/__init__.py | 23 +++ .../persistence/memory_config/base.py | 32 ++++ .../persistence/memory_config/memory.py | 23 +++ .../persistence/memory_config/model.py | 29 +++ .../deerflow/persistence/memory_config/sql.py | 50 +++++ .../20260521_01_user_memory_configs.py | 44 +++++ .../deerflow/persistence/models/__init__.py | 2 + 11 files changed, 329 insertions(+), 106 deletions(-) create mode 100644 offline-backend-20260512/backend/packages/harness/deerflow/persistence/memory_config/__init__.py create mode 100644 offline-backend-20260512/backend/packages/harness/deerflow/persistence/memory_config/base.py create mode 100644 offline-backend-20260512/backend/packages/harness/deerflow/persistence/memory_config/memory.py create mode 100644 offline-backend-20260512/backend/packages/harness/deerflow/persistence/memory_config/model.py create mode 100644 offline-backend-20260512/backend/packages/harness/deerflow/persistence/memory_config/sql.py create mode 100644 offline-backend-20260512/backend/packages/harness/deerflow/persistence/migrations/versions/20260521_01_user_memory_configs.py diff --git a/offline-backend-20260512/backend/app/gateway/app.py b/offline-backend-20260512/backend/app/gateway/app.py index 2e161eb2..1abe12e6 100644 --- a/offline-backend-20260512/backend/app/gateway/app.py +++ b/offline-backend-20260512/backend/app/gateway/app.py @@ -449,6 +449,18 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]: # Run Alembic migrations before anything else touches the schema. await _run_db_migrations() + # 预热 recall_budget cache（从 DB 批量加载,使 get_effective_memory_config 同步可读） + try: + from deerflow.config.memory_config import load_recall_budget_cache + + _mc_store = getattr(app.state, "memory_config_store", None) + if _mc_store is not None: + _recall_budgets = await _mc_store.load_all() + load_recall_budget_cache(_recall_budgets) + logger.info("Memory config cache warmed: %d user override(s)", len(_recall_budgets)) + except Exception: + logger.exception("Memory config cache warm-up failed (non-fatal)") + # Ensure admin user exists (auto-create on first boot) # Must run AFTER langgraph_runtime so app.state.store is available for thread migration await _ensure_admin_user(app) diff --git a/offline-backend-20260512/backend/app/gateway/deps.py b/offline-backend-20260512/backend/app/gateway/deps.py index 2d630d1b..1ed32a5f 100644 --- a/offline-backend-20260512/backend/app/gateway/deps.py +++ b/offline-backend-20260512/backend/app/gateway/deps.py @@ -138,6 +138,10 @@ async def langgraph_runtime(app: FastAPI) -> AsyncGenerator[None, None]: app.state.article_type_store = make_article_type_store(sf) + from deerflow.persistence.memory_config import make_user_memory_config_store + + app.state.memory_config_store = make_user_memory_config_store(sf) + # 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) @@ -261,6 +265,14 @@ def get_article_type_store(request: Request): return getattr(request.app.state, "article_type_store", None) +def get_memory_config_store(request: Request): + """Return the per-user memory config store.""" + val = getattr(request.app.state, "memory_config_store", None) + if val is None: + raise HTTPException(status_code=503, detail="Memory config store not available") + return val + + def get_run_context(request: Request) -> RunContext: """Build a :class:`RunContext` from ``app.state`` singletons. diff --git a/offline-backend-20260512/backend/app/gateway/routers/memory.py b/offline-backend-20260512/backend/app/gateway/routers/memory.py index deffd141..bf507fa9 100644 --- a/offline-backend-20260512/backend/app/gateway/routers/memory.py +++ b/offline-backend-20260512/backend/app/gateway/routers/memory.py @@ -21,15 +21,16 @@ from __future__ import annotations from typing import Any -from fastapi import APIRouter, HTTPException, Query +from fastapi import APIRouter, HTTPException, Query, Request from pydantic import BaseModel, Field +from app.gateway.deps import get_memory_config_store from deerflow.config.agents_config import AGENT_NAME_PATTERN from deerflow.config.memory_config import ( USER_WRITABLE_FIELDS, - delete_user_memory_override, + delete_user_memory_override_db, get_effective_memory_config, - save_user_memory_override, + save_user_memory_override_db, ) from deerflow.config.paths import get_paths from deerflow.runtime.user_context import get_effective_user_id @@ -295,14 +296,17 @@ async def get_memory_v2_config() -> MemoryV2ConfigResponse: "/memory/v2/config", response_model=MemoryV2ConfigResponse, summary="Patch Memory V2 Config", - description="把白名单字段写入当前用户的 per-user 配置覆盖。非白名单字段被忽略。", + description="把 hindsight.recall_budget 写入当前用户的 per-user 配置覆盖 (DB 存储)。", ) -async def patch_memory_v2_config(request: MemoryConfigPatchRequest) -> MemoryV2ConfigResponse: - """更新当前用户的 per-user 记忆配置覆盖。""" +async def patch_memory_v2_config(request: MemoryConfigPatchRequest, req: Request) -> MemoryV2ConfigResponse: + """更新当前用户的 per-user 记忆配置覆盖 (仅 hindsight.recall_budget)。""" user_id = get_effective_user_id() + store = get_memory_config_store(req) try: - save_user_memory_override(user_id, request.overrides) - except OSError as exc: + await save_user_memory_override_db(user_id, request.overrides, store) + except ValueError as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc + except Exception as exc: # noqa: BLE001 raise HTTPException(status_code=500, detail="Failed to save memory config override.") from exc config = get_effective_memory_config(user_id) @@ -317,18 +321,20 @@ async def patch_memory_v2_config(request: MemoryConfigPatchRequest) -> MemoryV2C response_model=MemoryV2ConfigResponse, summary="Delete Memory V2 Config Override", description=( - "撤销当前用户的 per-user 配置覆盖。传 ?field=injection.enabled 只撤销该字段;" - "不传 field 则清空整个 per-user 覆盖。" + "撤销当前用户的 per-user 配置覆盖。传 ?field=hindsight.recall_budget 只撤销该字段;" + "不传 field 则清空全部覆盖。" ), ) async def delete_memory_v2_config( + req: Request, field: str | None = Query(None, description="要撤销的点号路径字段;省略则清空全部覆盖"), ) -> MemoryV2ConfigResponse: """撤销当前用户的 per-user 记忆配置覆盖。""" user_id = get_effective_user_id() + store = get_memory_config_store(req) try: - delete_user_memory_override(user_id, field) - except OSError as exc: + await delete_user_memory_override_db(user_id, field, store) + except Exception as exc: # noqa: BLE001 raise HTTPException(status_code=500, detail="Failed to delete memory config override.") from exc config = get_effective_memory_config(user_id) diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/config/memory_config.py b/offline-backend-20260512/backend/packages/harness/deerflow/config/memory_config.py index 0dfd804e..fba6d105 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/config/memory_config.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/config/memory_config.py @@ -12,21 +12,21 @@ injection: {...} # 系统提示词注入 security: {...} # 内容安全扫描 -三层配置覆盖 +两层配置覆盖 ============ ``get_effective_memory_config(user_id)`` 按优先级合并 (高 -> 低): - 1. per-user 覆盖 ``{base_dir}/users/{user_id}/memory_config.json`` - 2. 全局覆盖 ``{base_dir}/memory_user_overrides.json`` - 3. config.yaml ``memory:`` 段 (即 ``get_memory_config()``) - 4. Pydantic 默认值 + 1. per-user 覆盖 数据库 ``user_memory_configs`` 表 (经进程内 cache 读取) + (有 per-user 记录时自动强制 enabled=True) + 2. config.yaml ``memory:`` 段 (即 ``get_memory_config()``) + 3. Pydantic 默认值 -第 1、2 层只允许覆盖 :data:`USER_WRITABLE_FIELDS` 白名单内的字段。 +第 1 层只允许覆盖 :data:`USER_WRITABLE_FIELDS` 白名单内的字段。 +cache 由 Gateway lifespan 启动时预热,写入时同步更新,保持同步读取路径。 """ from __future__ import annotations import copy -import json import logging from typing import Any @@ -129,31 +129,26 @@ class MemoryConfig(BaseModel): # =========================================================================== -# per-user 运行时覆盖 +# per-user 运行时覆盖 — 白名单 # =========================================================================== -# 用户可在运行时覆盖的字段白名单 (点号路径)。 -# 不在此表中的字段必须改 config.yaml 并重启 (例如 provider / api_key / 字符配额)。 +# 用户唯一可覆盖的字段:召回深度。 +# injection.* / builtin.* / enabled 始终由 config.yaml 控制,用户不可关闭。 USER_WRITABLE_FIELDS: frozenset[str] = frozenset( { - # injection.enabled and injection.include_builtin are always ON; - # removing them from the whitelist makes _filter_to_whitelist drop - # any per-user override for these fields, so stored "false" values - # in existing memory_config.json files are silently ignored. - "injection.max_tokens", - "injection.include_hindsight", - # hindsight.memory_mode is fixed to "hybrid"; not user-writable. - # hindsight.bank_mission is fixed; not user-writable. "hindsight.recall_budget", } ) +_VALID_RECALL_BUDGETS: frozenset[str] = frozenset({"low", "mid", "high"}) + def _filter_to_whitelist(overrides: dict[str, Any]) -> dict[str, Any]: """把覆盖 dict 过滤为只含白名单字段的嵌套 dict。 - 输入既支持嵌套形式 ``{"injection": {"enabled": false}}``,也支持 - 点号扁平形式 ``{"injection.enabled": false}``。非白名单字段被丢弃并 log。 + 输入支持嵌套形式 ``{"hindsight": {"recall_budget": "high"}}`` 或 + 点号扁平形式 ``{"hindsight.recall_budget": "high"}``。 + 非白名单字段被丢弃并 log。 """ result: dict[str, Any] = {} @@ -184,21 +179,6 @@ def _deep_merge(base: dict[str, Any], override: dict[str, Any]) -> dict[str, Any return merged -def _load_json_file(path: Any) -> dict[str, Any]: - """读取一个 JSON 覆盖文件;不存在或解析失败时返回空 dict。""" - try: - from pathlib import Path - - p = Path(path) - if not p.exists(): - return {} - data = json.loads(p.read_text(encoding="utf-8")) - return data if isinstance(data, dict) else {} - except (OSError, json.JSONDecodeError) as exc: - logger.warning("读取记忆配置覆盖文件失败 %s: %s", path, exc) - return {} - - # =========================================================================== # 全局单例 + 加载入口 # =========================================================================== @@ -223,82 +203,88 @@ def load_memory_config_from_dict(config_dict: dict) -> None: _memory_config = MemoryConfig(**config_dict) +# =========================================================================== +# per-user recall_budget 进程内 cache +# +# Gateway lifespan 启动时调用 load_recall_budget_cache() 从 DB 批量加载。 +# API 写入时同步更新 cache,使 get_effective_memory_config() 保持同步读取。 +# =========================================================================== + +_recall_budget_cache: dict[str, str] = {} + + +def load_recall_budget_cache(data: dict[str, str]) -> None: + """用 DB 批量加载结果替换整个 cache (启动时调用一次)。""" + global _recall_budget_cache + _recall_budget_cache = dict(data) + logger.info("recall_budget cache loaded: %d user(s)", len(_recall_budget_cache)) + + +def set_recall_budget_cache(user_id: str, value: str) -> None: + """写入单个用户的 recall_budget cache。""" + _recall_budget_cache[user_id] = value + + +def delete_recall_budget_cache(user_id: str) -> None: + """从 cache 中移除某用户的 recall_budget。""" + _recall_budget_cache.pop(user_id, None) + + +def get_recall_budget_from_cache(user_id: str) -> str | None: + """返回 cache 中某用户的 recall_budget,未设置时返回 None。""" + return _recall_budget_cache.get(user_id) + + def get_effective_memory_config(user_id: str) -> MemoryConfig: - """按三层覆盖优先级,返回某用户的生效记忆配置。 + """返回某用户的生效记忆配置 (同步,从进程内 cache 读取 per-user 覆盖)。 - 优先级 (高 -> 低):per-user 覆盖 > 全局覆盖 > config.yaml > 默认值。 - 第 1、2 层只应用 :data:`USER_WRITABLE_FIELDS` 白名单内的字段。 - - 覆盖文件读取失败不会抛异常 —— 退化为只用 config.yaml 级别配置。 + 优先级 (高 -> 低):per-user DB 覆盖 (经 cache) > config.yaml > 默认值。 + 有 per-user 记录时自动强制 enabled=True。 """ base = _memory_config.model_dump() - + recall_budget = _recall_budget_cache.get(user_id) + if recall_budget is not None: + base.setdefault("hindsight", {})["recall_budget"] = recall_budget + base["enabled"] = True try: - from deerflow.config.paths import get_paths - - paths = get_paths() - global_overrides = _filter_to_whitelist( - _load_json_file(paths.memory_global_overrides_file) - ) - user_overrides = _filter_to_whitelist( - _load_json_file(paths.user_memory_config_file(user_id)) - ) - except Exception as exc: # noqa: BLE001 - 配置覆盖失败必须降级,不能阻断对话 - logger.warning("加载 per-user 记忆配置覆盖失败,退回 config.yaml: %s", exc) - return _memory_config - - merged = _deep_merge(_deep_merge(base, global_overrides), user_overrides) - try: - return MemoryConfig(**merged) + return MemoryConfig(**base) except Exception as exc: # noqa: BLE001 - 覆盖产生非法配置时降级 logger.warning("合并后的记忆配置非法,退回 config.yaml: %s", exc) return _memory_config -def save_user_memory_override(user_id: str, patch: dict[str, Any]) -> dict[str, Any]: - """把白名单字段写入某用户的 per-user 覆盖文件,返回写入后的完整覆盖 dict。 +# =========================================================================== +# Gateway 调用的异步 DB 对接函数 +# =========================================================================== + + +async def save_user_memory_override_db(user_id: str, patch: dict[str, Any], store: Any) -> dict[str, Any]: + """把白名单字段写入 DB store 并更新 cache,返回生效配置 dict。 ``patch`` 支持嵌套或点号扁平形式;非白名单字段被忽略。 + ``store`` 为 ``UserMemoryConfigStore`` 实例 (避免循环 import,用 Any)。 """ - from deerflow.config.paths import get_paths - - path = get_paths().user_memory_config_file(user_id) - current = _load_json_file(path) filtered = _filter_to_whitelist(patch) - merged = _deep_merge(current, filtered) - - path.parent.mkdir(parents=True, exist_ok=True) - path.write_text(json.dumps(merged, ensure_ascii=False, indent=2), encoding="utf-8") - return merged + recall_budget = (filtered.get("hindsight") or {}).get("recall_budget") + if recall_budget is not None: + if recall_budget not in _VALID_RECALL_BUDGETS: + raise ValueError(f"recall_budget 必须为 {sorted(_VALID_RECALL_BUDGETS)} 之一,收到:{recall_budget!r}") + await store.set_recall_budget(user_id, recall_budget) + set_recall_budget_cache(user_id, recall_budget) + return get_effective_memory_config(user_id).model_dump() -def delete_user_memory_override(user_id: str, field: str | None = None) -> dict[str, Any]: - """撤销某用户的 per-user 覆盖。 +async def delete_user_memory_override_db(user_id: str, field: str | None, store: Any) -> dict[str, Any]: + """撤销某用户的 per-user 覆盖并更新 cache,返回生效配置 dict。 - ``field`` 为点号路径 (如 ``injection.enabled``) 时只删该字段; - 为 ``None`` 时清空整个 per-user 覆盖文件。返回剩余的覆盖 dict。 + ``field=None`` 清空全部;``field="hindsight.recall_budget"`` 只清该字段。 """ - from deerflow.config.paths import get_paths - - path = get_paths().user_memory_config_file(user_id) - if field is None: - if path.exists(): - path.unlink() - return {} - - current = _load_json_file(path) - if "." in field: - section, sub_field = field.split(".", 1) - if isinstance(current.get(section), dict): - current[section].pop(sub_field, None) - if not current[section]: - current.pop(section, None) + if field is None or field == "hindsight.recall_budget": + await store.delete(user_id) + delete_recall_budget_cache(user_id) else: - current.pop(field, None) - - path.parent.mkdir(parents=True, exist_ok=True) - path.write_text(json.dumps(current, ensure_ascii=False, indent=2), encoding="utf-8") - return current + logger.warning("忽略删除非白名单的记忆配置覆盖字段: %s", field) + return get_effective_memory_config(user_id).model_dump() __all__ = [ @@ -312,6 +298,10 @@ __all__ = [ "set_memory_config", "load_memory_config_from_dict", "get_effective_memory_config", - "save_user_memory_override", - "delete_user_memory_override", + "load_recall_budget_cache", + "set_recall_budget_cache", + "delete_recall_budget_cache", + "get_recall_budget_from_cache", + "save_user_memory_override_db", + "delete_user_memory_override_db", ] diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/memory_config/__init__.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/memory_config/__init__.py new file mode 100644 index 00000000..4936d59a --- /dev/null +++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/memory_config/__init__.py @@ -0,0 +1,23 @@ +"""Per-user memory configuration override persistence.""" + +from __future__ import annotations + +from deerflow.persistence.memory_config.base import UserMemoryConfigStore +from deerflow.persistence.memory_config.memory import MemoryUserMemoryConfigStore +from deerflow.persistence.memory_config.model import UserMemoryConfigRow +from deerflow.persistence.memory_config.sql import SqlUserMemoryConfigRepository + + +def make_user_memory_config_store(session_factory) -> UserMemoryConfigStore: + if session_factory is None: + return MemoryUserMemoryConfigStore() + return SqlUserMemoryConfigRepository(session_factory) + + +__all__ = [ + "UserMemoryConfigStore", + "UserMemoryConfigRow", + "MemoryUserMemoryConfigStore", + "SqlUserMemoryConfigRepository", + "make_user_memory_config_store", +] diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/memory_config/base.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/memory_config/base.py new file mode 100644 index 00000000..ff47a853 --- /dev/null +++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/memory_config/base.py @@ -0,0 +1,32 @@ +"""Abstract persistence API for per-user memory config overrides.""" + +from __future__ import annotations + +from abc import ABC, abstractmethod + + +class UserMemoryConfigStore(ABC): + + @abstractmethod + async def get_recall_budget(self, user_id: str) -> str | None: + """Return the user's recall_budget override, or None if not set.""" + raise NotImplementedError + + @abstractmethod + async def set_recall_budget(self, user_id: str, value: str) -> None: + """Upsert the user's recall_budget override.""" + raise NotImplementedError + + @abstractmethod + async def delete(self, user_id: str) -> None: + """Remove all per-user overrides for the given user.""" + raise NotImplementedError + + @abstractmethod + async def load_all(self) -> dict[str, str]: + """Bulk-load all user recall_budget overrides. + + Returns {user_id: recall_budget} for all rows where recall_budget IS NOT NULL. + Called once at startup to warm the in-memory cache. + """ + raise NotImplementedError diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/memory_config/memory.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/memory_config/memory.py new file mode 100644 index 00000000..68630dbf --- /dev/null +++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/memory_config/memory.py @@ -0,0 +1,23 @@ +"""In-memory user memory config store (used when database.backend=memory).""" + +from __future__ import annotations + +from deerflow.persistence.memory_config.base import UserMemoryConfigStore + + +class MemoryUserMemoryConfigStore(UserMemoryConfigStore): + + def __init__(self) -> None: + self._data: dict[str, str] = {} + + async def get_recall_budget(self, user_id: str) -> str | None: + return self._data.get(user_id) + + async def set_recall_budget(self, user_id: str, value: str) -> None: + self._data[user_id] = value + + async def delete(self, user_id: str) -> None: + self._data.pop(user_id, None) + + async def load_all(self) -> dict[str, str]: + return dict(self._data) diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/memory_config/model.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/memory_config/model.py new file mode 100644 index 00000000..2e0310f1 --- /dev/null +++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/memory_config/model.py @@ -0,0 +1,29 @@ +"""ORM model for per-user memory configuration overrides.""" + +from __future__ import annotations + +from datetime import UTC, datetime + +from sqlalchemy import String +from sqlalchemy.orm import Mapped, mapped_column + +from deerflow.persistence.base import Base +from deerflow.persistence.types import BeijingDateTime + + +class UserMemoryConfigRow(Base): + __tablename__ = "user_memory_configs" + + # One row per user; upserted on write. + user_id: Mapped[str] = mapped_column(String(64), primary_key=True) + + # The only user-writable field. NULL means "use global default from config.yaml". + # Allowed values: "low" | "mid" | "high". + recall_budget: Mapped[str | None] = mapped_column(String(16), nullable=True) + + updated_at: Mapped[datetime] = mapped_column( + BeijingDateTime(), + nullable=False, + default=lambda: datetime.now(UTC), + onupdate=lambda: datetime.now(UTC), + ) diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/memory_config/sql.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/memory_config/sql.py new file mode 100644 index 00000000..0969b39a --- /dev/null +++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/memory_config/sql.py @@ -0,0 +1,50 @@ +"""SQLAlchemy-backed user memory config repository.""" + +from __future__ import annotations + +from datetime import UTC, datetime + +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker + +from deerflow.persistence.memory_config.base import UserMemoryConfigStore +from deerflow.persistence.memory_config.model import UserMemoryConfigRow + + +class SqlUserMemoryConfigRepository(UserMemoryConfigStore): + + def __init__(self, session_factory: async_sessionmaker[AsyncSession]) -> None: + self._sf = session_factory + + async def get_recall_budget(self, user_id: str) -> str | None: + async with self._sf() as session: + row = await session.get(UserMemoryConfigRow, user_id) + return row.recall_budget if row is not None else None + + async def set_recall_budget(self, user_id: str, value: str) -> None: + async with self._sf() as session: + row = await session.get(UserMemoryConfigRow, user_id) + if row is None: + row = UserMemoryConfigRow( + user_id=user_id, + recall_budget=value, + updated_at=datetime.now(UTC), + ) + session.add(row) + else: + row.recall_budget = value + row.updated_at = datetime.now(UTC) + await session.commit() + + async def delete(self, user_id: str) -> None: + async with self._sf() as session: + row = await session.get(UserMemoryConfigRow, user_id) + if row is not None: + await session.delete(row) + await session.commit() + + async def load_all(self) -> dict[str, str]: + stmt = select(UserMemoryConfigRow).where(UserMemoryConfigRow.recall_budget.is_not(None)) + async with self._sf() as session: + result = await session.execute(stmt) + return {row.user_id: row.recall_budget for row in result.scalars() if row.recall_budget is not None} diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/migrations/versions/20260521_01_user_memory_configs.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/migrations/versions/20260521_01_user_memory_configs.py new file mode 100644 index 00000000..d949154a --- /dev/null +++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/migrations/versions/20260521_01_user_memory_configs.py @@ -0,0 +1,44 @@ +"""Create user_memory_configs table. + +Replaces the per-user file-based memory config override +({base_dir}/users/{user_id}/memory_config.json) with a DB row. +Stores only hindsight.recall_budget — the single user-writable field +after the memory config simplification. + +Revision ID: 20260521_01 +Revises: 20260519_02 +Create Date: 2026-05-21 +""" + +from __future__ import annotations + +import sqlalchemy as sa +from alembic import op +from sqlalchemy import inspect + + +revision = "20260521_01" +down_revision = "20260519_02" +branch_labels = None +depends_on = None + + +def _has_table(table_name: str) -> bool: + inspector = inspect(op.get_bind()) + return table_name in inspector.get_table_names() + + +def upgrade() -> None: + if _has_table("user_memory_configs"): + return + op.create_table( + "user_memory_configs", + sa.Column("user_id", sa.String(64), primary_key=True), + sa.Column("recall_budget", sa.String(16), nullable=True), + sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False), + ) + + +def downgrade() -> None: + if _has_table("user_memory_configs"): + op.drop_table("user_memory_configs") diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/models/__init__.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/models/__init__.py index 53b77024..3b19bb8d 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/models/__init__.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/models/__init__.py @@ -16,6 +16,7 @@ there is no matching entity directory. """ from deerflow.persistence.agents.model import AgentRow +from deerflow.persistence.memory_config.model import UserMemoryConfigRow from deerflow.persistence.feedback.model import FeedbackRow from deerflow.persistence.llm_metrics.model import LlmCallMetricRow from deerflow.persistence.models.run_event import RunEventRow @@ -55,5 +56,6 @@ __all__ = [ "ThreadMetaRow", "ThreadShareRow", "ToolCallMetricRow", + "UserMemoryConfigRow", "UserRow", ]