From 4ef96c7930160c7b06cb9c39bd5d09779d117a71 Mon Sep 17 00:00:00 2001 From: xuhd Date: Thu, 14 May 2026 17:09:05 +0800 Subject: [PATCH] =?UTF-8?q?feat(skills):=20=E5=A4=9A=E7=A7=9F=E6=88=B7?= =?UTF-8?q?=E9=98=B6=E6=AE=B5=203=20=E2=80=94=20Curator=20+=20=E9=85=8D?= =?UTF-8?q?=E7=BD=AE=20per-user?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit curator_log.py - ring buffer 从单一 deque 改为 dict[user_id -> deque]，每个用户 自己的运行日志互不可见 - start_new_run / record / get_recent / clear 全部接 user_id 关键字参数；未传时落入 __default__ 桶兼容旧调用 - 新增 clear_all() 用于跨用户清理（仅测试用） curator.py - 状态文件路径从 skills/.curator_state 改为 .deer-flow/users/{user_id}/.curator_state（per-user 隔离） - 所有公开 API 接受 user_id 参数：should_run_now / apply_automatic_transitions / _render_candidate_list / run_curator_review / _curator_thread_target / get_curator_status / _make_skill_archive_tool / _run_llm_review - 新增 _list_user_owned_skill_names(user_id)：通过 SkillStore 查询当前用户拥有的 skill 名集合，用于过滤 transitions / LLM candidates，保证 user A 跑 curator 不会动 user B 的 skill； SkillStore 不可达时退化为"全部 agent-created skill"（兼容旧部署） - _get_curator_config(user_id) 通过 load_skill_evolution_config_for_user 叠加 per-user override，让阈值/间隔/moderation_model 也是 per-user - maybe_run_curator 启动时枚举 UserRow，按用户检查 should_run_now 并各起一个 daemon 线程跑 skill_evolution_runtime.py - 全局 backend/skill_evolution.json 语义保留作为 fallback - 新增 per-user API：resolve_user_path / load_user_overrides / save_user_overrides / get_user_overrides_mtime，文件落到 .deer-flow/users/{user_id}/skill_evolution.json - 新增 load_skill_evolution_config_for_user(user_id)：读全局 SkillEvolutionConfig 作为 baseline，浅合并 per-user JSON， 返回 Pydantic 实例。验证失败时回落到 baseline routers/skills.py - /api/curator/status / run / config GET+PUT / logs 全部接 _current_user_id 注入运行上下文 - update_curator_config 改为 per-user：保存到 users/{user_id}/skill_evolution.json，不再 reset_app_config - _build_curator_config_response 接受 SkillEvolutionConfig 实例 （而不是 AppConfig）以适配 per-user 合并结果 旧 19 个测试继续通过；未引入新依赖。 Co-Authored-By: Claude Opus 4.7 (1M context) --- .../backend/app/gateway/routers/skills.py | 73 +++--- .../config/skill_evolution_runtime.py | 109 ++++++++- .../harness/deerflow/skills/curator.py | 212 +++++++++++++----- .../harness/deerflow/skills/curator_log.py | 85 ++++--- 4 files changed, 358 insertions(+), 121 deletions(-) diff --git a/offline-backend-20260512/backend/app/gateway/routers/skills.py b/offline-backend-20260512/backend/app/gateway/routers/skills.py index 63878141..90b03711 100644 --- a/offline-backend-20260512/backend/app/gateway/routers/skills.py +++ b/offline-backend-20260512/backend/app/gateway/routers/skills.py @@ -780,23 +780,32 @@ async def pin_skill(skill_name: str, request: SkillPinRequest, config: AppConfig @router.get("/curator/status", summary="Curator Status") -async def curator_status() -> dict: - """Return the current curator state and per-lifecycle skill counts.""" +async def curator_status(request: Request) -> dict: + """Return the current curator state and per-lifecycle skill counts for the + authenticated user.""" try: from deerflow.skills.curator import get_curator_status - return get_curator_status() + return get_curator_status(_current_user_id(request)) except Exception as e: logger.error("Failed to get curator status: %s", e, exc_info=True) raise HTTPException(status_code=500, detail=f"Failed to get curator status: {str(e)}") @router.post("/curator/run", summary="Trigger Curator Run") -async def run_curator() -> dict: - """Manually trigger an immediate curator review pass (bypasses interval check, runs in background thread).""" +async def run_curator(request: Request) -> dict: + """Manually trigger a curator review pass for the authenticated user + (bypasses interval check, runs in a background thread).""" try: import threading from deerflow.skills.curator import _curator_thread_target - thread = threading.Thread(target=_curator_thread_target, daemon=True, name="skill-curator-manual") + + user_id = _current_user_id(request) + thread = threading.Thread( + target=_curator_thread_target, + args=(user_id,), + daemon=True, + name=f"skill-curator-manual-{user_id}", + ) thread.start() return {"triggered": True, "message": "Curator run started in background"} except Exception as e: @@ -845,9 +854,12 @@ class CuratorLogsResponse(BaseModel): last_run_at: str | None -def _build_curator_config_response(config: AppConfig) -> CuratorConfigResponse: - """Pure helper so both the GET route and the post-PUT refresh share one shape.""" - evo = config.skill_evolution +def _build_curator_config_response(evo) -> CuratorConfigResponse: + """Pure helper so both the GET route and the post-PUT refresh share one shape. + + ``evo`` is a ``SkillEvolutionConfig`` instance (already merged with per-user + overrides if any). + """ return CuratorConfigResponse( enabled=evo.enabled, moderation_model_name=evo.moderation_model_name, @@ -861,38 +873,37 @@ def _build_curator_config_response(config: AppConfig) -> CuratorConfigResponse: @router.get("/curator/config", response_model=CuratorConfigResponse, summary="Get Skill Evolution Config") -async def get_curator_config(config: AppConfig = Depends(get_config)) -> CuratorConfigResponse: - """Return the current effective skill_evolution config (yaml defaults + skill_evolution.json overrides).""" +async def get_curator_config(request: Request) -> CuratorConfigResponse: + """Return the current curator config for the authenticated user (yaml + defaults + that user's overrides).""" try: - return _build_curator_config_response(config) + from deerflow.config.skill_evolution_runtime import load_skill_evolution_config_for_user + + evo = load_skill_evolution_config_for_user(_current_user_id(request)) + return _build_curator_config_response(evo) except Exception as e: logger.error("Failed to read curator config: %s", e, exc_info=True) raise HTTPException(status_code=500, detail=f"Failed to read curator config: {str(e)}") @router.put("/curator/config", response_model=CuratorConfigResponse, summary="Update Skill Evolution Config") -async def update_curator_config(body: CuratorConfigRequest) -> CuratorConfigResponse: - """Persist overrides to skill_evolution.json and force AppConfig to reload on next read. +async def update_curator_config(body: CuratorConfigRequest, request: Request) -> CuratorConfigResponse: + """Persist per-user overrides to ``users/{user_id}/skill_evolution.json`` + and return the effective merged config. - The payload is validated against the SkillEvolutionConfig shape; unknown - fields are rejected by the Pydantic model_validate call. + The payload is validated against the ``SkillEvolutionConfig`` shape; unknown + fields are rejected. """ try: - from deerflow.config.app_config import get_app_config, reset_app_config from deerflow.config.skill_evolution_config import SkillEvolutionConfig - from deerflow.config.skill_evolution_runtime import save_overrides + from deerflow.config.skill_evolution_runtime import load_skill_evolution_config_for_user, save_user_overrides + user_id = _current_user_id(request) payload = body.model_dump() # Round-trip through SkillEvolutionConfig to ensure persisted shape stays valid. SkillEvolutionConfig.model_validate(payload) - save_overrides(payload) - - # Reset cache so the next get_app_config() picks up the new overrides - # (the mtime check would catch it too, but resetting makes the response - # below deterministically reflect the new state). - reset_app_config() - fresh_config = get_app_config() - return _build_curator_config_response(fresh_config) + save_user_overrides(user_id, payload) + return _build_curator_config_response(load_skill_evolution_config_for_user(user_id)) except HTTPException: raise except Exception as e: @@ -901,8 +912,9 @@ async def update_curator_config(body: CuratorConfigRequest) -> CuratorConfigResp @router.get("/curator/logs", response_model=CuratorLogsResponse, summary="Recent Curator Logs") -async def get_curator_logs() -> CuratorLogsResponse: - """Return in-memory log records from the most recent curator run. +async def get_curator_logs(request: Request) -> CuratorLogsResponse: + """Return in-memory log records from the most recent curator run for the + authenticated user. The buffer is process-local and cleared when curator starts a new run or when the backend restarts. Records are ordered oldest → newest. @@ -911,8 +923,9 @@ async def get_curator_logs() -> CuratorLogsResponse: from deerflow.skills import curator_log from deerflow.skills.curator import get_curator_status - records = curator_log.get_recent() - status = get_curator_status() + user_id = _current_user_id(request) + records = curator_log.get_recent(user_id) + status = get_curator_status(user_id) return CuratorLogsResponse( records=[CuratorLogRecord(**r) for r in records], run_count=int(status.get("run_count", 0)), diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/config/skill_evolution_runtime.py b/offline-backend-20260512/backend/packages/harness/deerflow/config/skill_evolution_runtime.py index 4a2e6fe6..093f306f 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/config/skill_evolution_runtime.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/config/skill_evolution_runtime.py @@ -1,9 +1,17 @@ """Runtime overrides for `skill_evolution` config, stored in skill_evolution.json. -The file format mirrors the ``skill_evolution`` YAML section — its content is -deep-merged onto the defaults loaded from ``config.yaml`` when ``AppConfig`` is -instantiated. The file is optional; absence (or a malformed payload) is treated -as "no overrides" so the system keeps the YAML-only behaviour. +Two scopes exist: + +* **Global** (back-compat): ``backend/skill_evolution.json`` deep-merged onto + ``config.yaml.skill_evolution`` at AppConfig load time. Used as a fallback + for users who never set a personal override. +* **Per-user**: ``.deer-flow/users/{user_id}/skill_evolution.json``. Read by + the curator on each run and by the ``GET/PUT /api/curator/config`` route + with the active user id. Curator merges this on top of the global config + via :func:`load_curator_config_for_user`. + +The file is optional in both scopes; absence (or a malformed payload) is +treated as "no overrides" so the system keeps the YAML-only behaviour. """ from __future__ import annotations @@ -113,3 +121,96 @@ def get_overrides_mtime() -> float | None: return path.stat().st_mtime except OSError: return None + + +# --------------------------------------------------------------------------- +# Per-user overrides +# --------------------------------------------------------------------------- + + +def resolve_user_path(user_id: str) -> Path: + """Return ``{base_dir}/users/{user_id}/skill_evolution.json``.""" + from deerflow.config.paths import get_paths + + return get_paths().user_dir(user_id) / _FILENAME + + +def load_user_overrides(user_id: str) -> dict[str, Any] | None: + """Read the per-user overrides file for ``user_id`` (None if missing/invalid).""" + path = resolve_user_path(user_id) + if not path.exists(): + return None + try: + raw = path.read_text(encoding="utf-8").strip() + if not raw: + return None + data = json.loads(raw) + except OSError as exc: + logger.warning("Failed to read %s: %s", path, exc) + return None + except json.JSONDecodeError as exc: + logger.warning("User skill_evolution overrides at %s are not valid JSON: %s", path, exc) + return None + if not isinstance(data, dict): + logger.warning("User skill_evolution overrides at %s must be a JSON object", path) + return None + return data + + +def save_user_overrides(user_id: str, payload: dict[str, Any]) -> Path: + """Atomically persist ``payload`` to the per-user overrides file.""" + if not isinstance(payload, dict): + raise TypeError("skill_evolution overrides payload must be a dict") + path = resolve_user_path(user_id) + path.parent.mkdir(parents=True, exist_ok=True) + fd, tmp_path_str = tempfile.mkstemp(prefix=".skill-evolution-", suffix=".tmp", dir=str(path.parent)) + tmp_path = Path(tmp_path_str) + try: + with os.fdopen(fd, "w", encoding="utf-8") as f: + json.dump(payload, f, indent=2, ensure_ascii=False) + os.replace(tmp_path, path) + except Exception: + tmp_path.unlink(missing_ok=True) + raise + logger.info("Wrote per-user skill_evolution overrides to %s", path) + return path + + +def get_user_overrides_mtime(user_id: str) -> float | None: + path = resolve_user_path(user_id) + try: + return path.stat().st_mtime + except OSError: + return None + + +def load_skill_evolution_config_for_user(user_id: str | None): + """Return a ``SkillEvolutionConfig`` instance with per-user overrides merged + on top of the global AppConfig defaults. + + Used by the curator + the per-user ``GET/PUT /api/curator/config`` route. + When ``user_id`` is None or the per-user file is absent, returns the global + config unchanged. + """ + from deerflow.config import get_app_config + from deerflow.config.skill_evolution_config import SkillEvolutionConfig + + base = get_app_config().skill_evolution + if not user_id: + return base + overrides = load_user_overrides(user_id) + if not overrides: + return base + + merged = base.model_dump() + for key, value in overrides.items(): + existing = merged.get(key) + if isinstance(existing, dict) and isinstance(value, dict): + merged[key] = {**existing, **value} + else: + merged[key] = value + try: + return SkillEvolutionConfig.model_validate(merged) + except Exception: + logger.warning("Per-user skill_evolution overrides for %s failed validation, ignoring", user_id, exc_info=True) + return base diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/skills/curator.py b/offline-backend-20260512/backend/packages/harness/deerflow/skills/curator.py index 9695b850..08491e0e 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/skills/curator.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/skills/curator.py @@ -1,7 +1,14 @@ -"""Skill curator — automatic lifecycle transitions and LLM consolidation pass.""" +"""Skill curator — automatic lifecycle transitions and LLM consolidation pass. + +Per-user since the multitenancy upgrade: every public function takes a +``user_id`` argument and the on-disk state lives under +``{base_dir}/users/{user_id}/.curator_state`` (alongside the per-user +memory.json and threads/ dirs). +""" from __future__ import annotations +import asyncio import json import logging import os @@ -22,12 +29,18 @@ def _get_skills_root() -> Path: return Path(get_or_new_skill_storage().get_skills_root_path()) -def _curator_state_path() -> Path: - return _get_skills_root() / _CURATOR_STATE_FILENAME +def _curator_state_path(user_id: str | None) -> Path: + """Return the state file path. Per-user when user_id is known; falls back + to the global ``skills/.curator_state`` for legacy callers.""" + if not user_id: + return _get_skills_root() / _CURATOR_STATE_FILENAME + from deerflow.config.paths import get_paths + + return get_paths().user_dir(user_id) / _CURATOR_STATE_FILENAME -def _read_curator_state() -> dict: - path = _curator_state_path() +def _read_curator_state(user_id: str | None) -> dict: + path = _curator_state_path(user_id) if not path.exists(): return {"run_count": 0, "paused": False} try: @@ -36,8 +49,8 @@ def _read_curator_state() -> dict: return {"run_count": 0, "paused": False} -def _write_curator_state(state: dict) -> None: - path = _curator_state_path() +def _write_curator_state(state: dict, user_id: str | None) -> None: + path = _curator_state_path(user_id) try: path.parent.mkdir(parents=True, exist_ok=True) tmp_fd, tmp_path = tempfile.mkstemp(dir=path.parent, suffix=".tmp") @@ -55,23 +68,54 @@ def _write_curator_state(state: dict) -> None: logger.debug("Failed to write curator state", exc_info=True) -def _get_curator_config(): - try: - from deerflow.config import get_app_config +def _get_curator_config(user_id: str | None = None): + """Return the curator config object for ``user_id``. - config = get_app_config() - skill_evo = getattr(config, "skill_evolution", None) + Reads ``config.yaml.skill_evolution.curator`` as the baseline and merges + the per-user override file (``users/{user_id}/skill_evolution.json``) on + top. With ``user_id=None`` falls back to the global default. + """ + try: + from deerflow.config.skill_evolution_runtime import load_skill_evolution_config_for_user + + skill_evo = load_skill_evolution_config_for_user(user_id) return getattr(skill_evo, "curator", None) except Exception: return None -def should_run_now() -> bool: - """Return True if the curator interval has elapsed and it is not paused.""" +def _list_user_owned_skill_names(user_id: str) -> set[str] | None: + """Return skill names owned by ``user_id`` (used to filter transitions). + + Returns None if the SkillStore is unreachable — curator then degrades to + "process every on-disk agent-created skill" (legacy behaviour). + """ + try: + from deerflow.persistence.engine import get_session_factory + from deerflow.persistence.skills import make_skill_store + + factory = get_session_factory() + if factory is None: + return None + store = make_skill_store(factory) + + async def _run() -> list[dict]: + return await store.list_visible(user_id) + + records = asyncio.run(_run()) + return {r["name"] for r in records if r.get("owner_user_id") == user_id} + except Exception: + logger.debug("_list_user_owned_skill_names failed", exc_info=True) + return None + + +def should_run_now(user_id: str) -> bool: + """Return True if the curator interval has elapsed for ``user_id`` and it + is not paused.""" curator_cfg = _get_curator_config() if curator_cfg is None or not getattr(curator_cfg, "enabled", True): return False - state = _read_curator_state() + state = _read_curator_state(user_id) if state.get("paused"): return False last_run = state.get("last_run_at") @@ -85,11 +129,11 @@ def should_run_now() -> bool: return True -def apply_automatic_transitions(now: datetime | None = None) -> dict: +def apply_automatic_transitions(user_id: str, now: datetime | None = None) -> dict: """Pure age-based state transitions — no LLM involved. - Only processes skills with source=='agent'. Skips pinned skills. - Returns counters: {marked_stale, archived, checked}. + Only processes agent-created skills owned by ``user_id``. Skips pinned + skills. Returns counters: {marked_stale, archived, checked}. """ from deerflow.skills.usage import STATE_ACTIVE, STATE_ARCHIVED, STATE_STALE, agent_created_entries, latest_activity_at, set_state @@ -100,7 +144,10 @@ def apply_automatic_transitions(now: datetime | None = None) -> dict: stale_days = getattr(curator_cfg, "stale_after_days", 30) if curator_cfg else 30 archive_days = getattr(curator_cfg, "archive_after_days", 90) if curator_cfg else 90 + owned = _list_user_owned_skill_names(user_id) entries = agent_created_entries() + if owned is not None: + entries = {name: entry for name, entry in entries.items() if name in owned} counters = {"marked_stale": 0, "archived": 0, "checked": len(entries)} for name, entry in entries.items(): @@ -129,10 +176,17 @@ def apply_automatic_transitions(now: datetime | None = None) -> dict: return counters -def _render_candidate_list() -> str: +def _render_candidate_list(user_id: str) -> str: from deerflow.skills.usage import STATE_ARCHIVED, all_entries - entries = {k: v for k, v in all_entries().items() if v.get("source") == "agent" and v.get("state") != STATE_ARCHIVED} + owned = _list_user_owned_skill_names(user_id) + entries = { + k: v + for k, v in all_entries().items() + if v.get("source") == "agent" and v.get("state") != STATE_ARCHIVED + } + if owned is not None: + entries = {k: v for k, v in entries.items() if k in owned} if not entries: return "No agent-created active skills found." lines = [] @@ -248,7 +302,7 @@ _CURATOR_REVIEW_PROMPT = ( ) -def _make_skill_archive_tool(): +def _make_skill_archive_tool(user_id: str): """Curator-only tool: move a skill directory to .archive/ (recoverable).""" from langchain.tools import tool as lc_tool @@ -266,10 +320,10 @@ def _make_skill_archive_tool(): storage = get_or_new_skill_storage() if not storage.custom_skill_exists(name): - curator_log.record("warning", f"skill_archive: '{name}' not found or already archived") + curator_log.record("warning", f"skill_archive: '{name}' not found or already archived", user_id=user_id) return f"Skill '{name}' not found or already archived." if get_entry(name).get("pinned"): - curator_log.record("info", f"skill_archive: '{name}' is pinned — skipping") + curator_log.record("info", f"skill_archive: '{name}' is pinned — skipping", user_id=user_id) return f"Skill '{name}' is pinned — skipping." skill_dir = storage.get_custom_skill_dir(name) archive_root = Path(storage.get_skills_root_path()) / "custom" / ".archive" @@ -279,16 +333,16 @@ def _make_skill_archive_tool(): try: skill_dir.rename(dest) except Exception as e: - curator_log.record("error", f"skill_archive: archive failed for '{name}': {e}") + curator_log.record("error", f"skill_archive: archive failed for '{name}': {e}", user_id=user_id) return f"Archive failed: {e}" set_state(name, STATE_ARCHIVED) - curator_log.record("info", f"Archived '{name}' → custom/.archive/{dest.name}") + curator_log.record("info", f"Archived '{name}' → custom/.archive/{dest.name}", user_id=user_id) return f"Archived '{name}' → custom/.archive/{dest.name}." return _skill_archive -def _run_llm_review(candidate_list: str) -> str: +def _run_llm_review(candidate_list: str, *, user_id: str) -> str: """Run the curator consolidation pass as a React agent with skill tools bound.""" from deerflow.skills import curator_log @@ -301,10 +355,10 @@ def _run_llm_review(candidate_list: str) -> str: from deerflow.tools.skill_manage_tool import skill_manage_tool model = create_chat_model() - tools = [skill_list_tool, skill_view_tool, skill_manage_tool, _make_skill_archive_tool()] + tools = [skill_list_tool, skill_view_tool, skill_manage_tool, _make_skill_archive_tool(user_id)] agent = create_react_agent(model, tools) - curator_log.record("info", "LLM review: invoking React agent") + curator_log.record("info", "LLM review: invoking React agent", user_id=user_id) prompt = _CURATOR_REVIEW_PROMPT + f"\n\nSkill candidates for this pass:\n{candidate_list}" result = agent.invoke({"messages": [HumanMessage(content=prompt)]}) @@ -313,20 +367,20 @@ def _run_llm_review(candidate_list: str) -> str: if not getattr(msg, "tool_calls", None): content = getattr(msg, "content", "") if isinstance(content, str) and content.strip(): - curator_log.record("info", "LLM review: agent returned final answer") + curator_log.record("info", "LLM review: agent returned final answer", user_id=user_id) return content if isinstance(content, list): text = " ".join( block.get("text", "") if isinstance(block, dict) else str(block) for block in content ).strip() - curator_log.record("info", "LLM review: agent returned final answer") + curator_log.record("info", "LLM review: agent returned final answer", user_id=user_id) return text - curator_log.record("warning", "LLM review: no final assistant message produced") + curator_log.record("warning", "LLM review: no final assistant message produced", user_id=user_id) return "" except Exception as e: logger.warning("Curator LLM review failed: %s", e) - curator_log.record("warning", f"LLM review failed: {e}") + curator_log.record("warning", f"LLM review failed: {e}", user_id=user_id) return "" @@ -344,18 +398,21 @@ def _parse_structured_summary(text: str) -> dict: return {"consolidations": [], "prunings": []} -def run_curator_review() -> dict: - """Main curator run: automatic transitions → LLM consolidation → write state.""" +def run_curator_review(user_id: str) -> dict: + """Main curator run for ``user_id``: transitions → LLM consolidation → state. + + Only skills owned by ``user_id`` are touched. + """ from deerflow.skills import curator_log - curator_log.start_new_run() + curator_log.start_new_run(user_id=user_id) now = datetime.now(UTC) - logger.info("Curator: starting run at %s", now.isoformat()) - curator_log.record("info", f"Curator starting run at {now.isoformat()}") + logger.info("Curator[%s]: starting run at %s", user_id, now.isoformat()) + curator_log.record("info", f"Curator starting run at {now.isoformat()}", user_id=user_id) - transition_counters = apply_automatic_transitions(now) - logger.info("Curator: transitions=%s", transition_counters) + transition_counters = apply_automatic_transitions(user_id, now) + logger.info("Curator[%s]: transitions=%s", user_id, transition_counters) curator_log.record( "info", ( @@ -363,13 +420,16 @@ def run_curator_review() -> dict: f"marked_stale={transition_counters['marked_stale']}, " f"archived={transition_counters['archived']}" ), + user_id=user_id, ) - candidate_list = _render_candidate_list() + candidate_list = _render_candidate_list(user_id) candidate_line_count = 0 if not candidate_list.strip() else len([line for line in candidate_list.splitlines() if line.startswith("- ")]) - curator_log.record("info", f"Candidates for LLM review: {candidate_line_count} skill(s)") + curator_log.record( + "info", f"Candidates for LLM review: {candidate_line_count} skill(s)", user_id=user_id + ) - llm_text = _run_llm_review(candidate_list) + llm_text = _run_llm_review(candidate_list, user_id=user_id) suggestions = _parse_structured_summary(llm_text) curator_log.record( "info", @@ -377,9 +437,10 @@ def run_curator_review() -> dict: f"LLM suggestions parsed: consolidations={len(suggestions['consolidations'])}, " f"prunings={len(suggestions['prunings'])}" ), + user_id=user_id, ) - state = _read_curator_state() + state = _read_curator_state(user_id) state["last_run_at"] = now.isoformat() state["run_count"] = state.get("run_count", 0) + 1 state["last_run_summary"] = ( @@ -388,45 +449,78 @@ def run_curator_review() -> dict: f"llm: {len(suggestions['consolidations'])} consolidations suggested" ) state["last_consolidation_suggestions"] = suggestions - _write_curator_state(state) + _write_curator_state(state, user_id) - logger.info("Curator: run complete — %s", state["last_run_summary"]) - curator_log.record("info", f"Curator run complete — {state['last_run_summary']}") + logger.info("Curator[%s]: run complete — %s", user_id, state["last_run_summary"]) + curator_log.record("info", f"Curator run complete — {state['last_run_summary']}", user_id=user_id) return {"transitions": transition_counters, "suggestions": suggestions} -def _curator_thread_target() -> None: +def _curator_thread_target(user_id: str) -> None: from deerflow.skills import curator_log try: - run_curator_review() + run_curator_review(user_id) except Exception as exc: - logger.exception("Curator background run failed") - curator_log.record("error", f"Curator background run failed: {exc}") + logger.exception("Curator[%s] background run failed", user_id) + curator_log.record("error", f"Curator background run failed: {exc}", user_id=user_id) async def maybe_run_curator() -> None: - """Async entry called at startup — spawns a daemon thread if the interval has elapsed.""" - if not should_run_now(): + """Async startup entry — kicks off a thread per user whose interval elapsed. + + Skips users with no DB-tracked entries (a fresh deployment) and never + blocks lifespan startup on failure. + """ + try: + from sqlalchemy import select + + from deerflow.persistence.engine import get_session_factory + from deerflow.persistence.user.model import UserRow + + factory = get_session_factory() + if factory is None: + return + async with factory() as session: + result = await session.execute(select(UserRow.id)) + user_ids = [str(row[0]) for row in result.all()] + except Exception: + logger.exception("Curator: failed to enumerate users at startup") return - logger.info("Curator: interval elapsed, spawning background thread") - thread = threading.Thread(target=_curator_thread_target, daemon=True, name="skill-curator") - thread.start() + + for user_id in user_ids: + try: + if not should_run_now(user_id): + continue + logger.info("Curator[%s]: interval elapsed, spawning background thread", user_id) + thread = threading.Thread( + target=_curator_thread_target, + args=(user_id,), + daemon=True, + name=f"skill-curator-{user_id}", + ) + thread.start() + except Exception: + logger.exception("Curator: failed to spawn for user %s", user_id) -def get_curator_status() -> dict: - """Return curator state for the API status endpoint.""" +def get_curator_status(user_id: str) -> dict: + """Return curator state for ``user_id``.""" from deerflow.skills.usage import STATE_ACTIVE, STATE_ARCHIVED, STATE_STALE, agent_created_entries - state = _read_curator_state() + state = _read_curator_state(user_id) + owned = _list_user_owned_skill_names(user_id) entries = agent_created_entries() + if owned is not None: + entries = {name: entry for name, entry in entries.items() if name in owned} counts: dict[str, int] = {STATE_ACTIVE: 0, STATE_STALE: 0, STATE_ARCHIVED: 0} for entry in entries.values(): s = entry.get("state", STATE_ACTIVE) counts[s] = counts.get(s, 0) + 1 + cfg = _get_curator_config() return { - "enabled": _get_curator_config() is not None and getattr(_get_curator_config(), "enabled", True), + "enabled": cfg is not None and getattr(cfg, "enabled", True), "run_count": state.get("run_count", 0), "last_run_at": state.get("last_run_at"), "last_run_summary": state.get("last_run_summary"), diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/skills/curator_log.py b/offline-backend-20260512/backend/packages/harness/deerflow/skills/curator_log.py index 53dcb9bf..2b8780d6 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/skills/curator_log.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/skills/curator_log.py @@ -1,11 +1,14 @@ """In-memory ring buffer holding the most recent curator run's execution log. -The buffer is intentionally process-local and non-persistent — restarting the -server clears it. The Gateway exposes the buffer over ``GET /api/curator/logs`` -so the frontend can show what the latest curator pass actually did. +Now sharded per-user — each ``user_id`` gets its own deque so users only see +their own curator activity from the Web UI. -All public functions swallow internal exceptions so a logging failure can never -break the curator run itself. +The buffer is intentionally process-local and non-persistent. The Gateway +exposes it over ``GET /api/curator/logs`` with the active user id resolved +from the request context. + +All public functions swallow internal exceptions so a logging failure can +never break the curator run itself. """ from __future__ import annotations @@ -19,36 +22,44 @@ from typing import Any logger = logging.getLogger(__name__) _MAX_RECORDS = 200 +# Reserved bucket key for "no user context" callers (legacy code paths or +# tests). Real curator runs always supply a concrete user_id. +_DEFAULT_BUCKET = "__default__" _lock = threading.Lock() -_buffer: deque[dict[str, Any]] = deque(maxlen=_MAX_RECORDS) +_buffers: dict[str, deque[dict[str, Any]]] = {} + + +def _bucket_key(user_id: str | None) -> str: + return user_id or _DEFAULT_BUCKET + + +def _get_or_create(user_id: str | None) -> deque[dict[str, Any]]: + key = _bucket_key(user_id) + buffer = _buffers.get(key) + if buffer is None: + buffer = deque(maxlen=_MAX_RECORDS) + _buffers[key] = buffer + return buffer def _now_iso() -> str: return datetime.now(UTC).isoformat() -def start_new_run() -> None: - """Clear the buffer and emit a ``run_started`` record. - - Called once at the top of each curator run so the frontend sees a fresh - log stream rather than the previous run's tail. - """ +def start_new_run(user_id: str | None = None) -> None: + """Clear the user's buffer and emit a ``run_started`` record.""" try: with _lock: - _buffer.clear() - _buffer.append({"ts": _now_iso(), "level": "info", "message": "run_started"}) + buffer = _get_or_create(user_id) + buffer.clear() + buffer.append({"ts": _now_iso(), "level": "info", "message": "run_started"}) except Exception: logger.debug("curator_log.start_new_run failed", exc_info=True) -def record(level: str, message: str, **fields: Any) -> None: - """Append a single log record. Never raises. - - ``level`` is one of ``info``/``warning``/``error``; unknown values are - accepted and rendered verbatim by the frontend. Extra keyword arguments - are merged into the record so callers can attach structured context. - """ +def record(level: str, message: str, *, user_id: str | None = None, **fields: Any) -> None: + """Append a single log record to the user's buffer. Never raises.""" try: entry: dict[str, Any] = { "ts": _now_iso(), @@ -58,25 +69,43 @@ def record(level: str, message: str, **fields: Any) -> None: if fields: entry.update({k: v for k, v in fields.items() if k not in entry}) with _lock: - _buffer.append(entry) + buffer = _get_or_create(user_id) + buffer.append(entry) except Exception: logger.debug("curator_log.record failed", exc_info=True) -def get_recent() -> list[dict[str, Any]]: - """Return a snapshot copy of the buffer contents. Never raises.""" +def get_recent(user_id: str | None = None) -> list[dict[str, Any]]: + """Return a snapshot of records for ``user_id``. Never raises.""" try: with _lock: - return list(_buffer) + buffer = _buffers.get(_bucket_key(user_id)) + if buffer is None: + return [] + return list(buffer) except Exception: logger.debug("curator_log.get_recent failed", exc_info=True) return [] -def clear() -> None: - """Drop all buffered records. Used by tests.""" +def clear(user_id: str | None = None) -> None: + """Drop a user's records. With ``user_id=None`` drops the default bucket. + + Use ``clear_all()`` for cross-user cleanup (tests). + """ try: with _lock: - _buffer.clear() + buffer = _buffers.get(_bucket_key(user_id)) + if buffer is not None: + buffer.clear() except Exception: logger.debug("curator_log.clear failed", exc_info=True) + + +def clear_all() -> None: + """Drop every user's records. Used by tests.""" + try: + with _lock: + _buffers.clear() + except Exception: + logger.debug("curator_log.clear_all failed", exc_info=True)