feat(skills): 多租户阶段 3 — Curator + 配置 per-user

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) <noreply@anthropic.com>

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:
+async def curator_status(request: Request) -> dict:
-    """Return the current curator state and per-lifecycle skill counts."""
+    """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:
+async def run_curator(request: Request) -> dict:
-    """Manually trigger an immediate curator review pass (bypasses interval check, runs in background thread)."""
+    """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:
+def _build_curator_config_response(evo) -> CuratorConfigResponse:
-    """Pure helper so both the GET route and the post-PUT refresh share one shape."""
+    """Pure helper so both the GET route and the post-PUT refresh share one shape.
-    evo = config.skill_evolution
+
+    ``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:
+async def get_curator_config(request: Request) -> CuratorConfigResponse:
-    """Return the current effective skill_evolution config (yaml defaults + skill_evolution.json overrides)."""
+    """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:
+async def update_curator_config(body: CuratorConfigRequest, request: Request) -> CuratorConfigResponse:
-    """Persist overrides to skill_evolution.json and force AppConfig to reload on next read.
+    """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
+    The payload is validated against the ``SkillEvolutionConfig`` shape; unknown
-    fields are rejected by the Pydantic model_validate call.
+    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)
+        save_user_overrides(user_id, payload)
-
+        return _build_curator_config_response(load_skill_evolution_config_for_user(user_id))
-        # 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)
     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:
+async def get_curator_logs(request: Request) -> CuratorLogsResponse:
-    """Return in-memory log records from the most recent curator run.
+    """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()
+        user_id = _current_user_id(request)
-        status = get_curator_status()
+        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)),

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
+Two scopes exist:
-deep-merged onto the defaults loaded from ``config.yaml`` when ``AppConfig`` is
+
-instantiated. The file is optional; absence (or a malformed payload) is treated
+* **Global** (back-compat): ``backend/skill_evolution.json`` deep-merged onto
-as "no overrides" so the system keeps the YAML-only behaviour.
+  ``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

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:
+def _curator_state_path(user_id: str | None) -> Path:
-    return _get_skills_root() / _CURATOR_STATE_FILENAME
+    """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:
+def _read_curator_state(user_id: str | None) -> dict:
-    path = _curator_state_path()
+    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:
+def _write_curator_state(state: dict, user_id: str | None) -> None:
-    path = _curator_state_path()
+    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():
+def _get_curator_config(user_id: str | None = None):
-    try:
+    """Return the curator config object for ``user_id``.
-        from deerflow.config import get_app_config
 
-        config = get_app_config()
+    Reads ``config.yaml.skill_evolution.curator`` as the baseline and merges
-        skill_evo = getattr(config, "skill_evolution", None)
+    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:
+def _list_user_owned_skill_names(user_id: str) -> set[str] | None:
-    """Return True if the curator interval has elapsed and it is not paused."""
+    """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.
+    Only processes agent-created skills owned by ``user_id``. Skips pinned
-    Returns counters: {marked_stale, archived, checked}.
+    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:
+def run_curator_review(user_id: str) -> dict:
-    """Main curator run: automatic transitions → LLM consolidation → write state."""
+    """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())
+    logger.info("Curator[%s]: starting run at %s", user_id, now.isoformat())
-    curator_log.record("info", f"Curator starting run at {now.isoformat()}")
+    curator_log.record("info", f"Curator starting run at {now.isoformat()}", user_id=user_id)
 
-    transition_counters = apply_automatic_transitions(now)
+    transition_counters = apply_automatic_transitions(user_id, now)
-    logger.info("Curator: transitions=%s", transition_counters)
+    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"])
+    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']}")
+    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")
+        logger.exception("Curator[%s] background run failed", user_id)
-        curator_log.record("error", f"Curator background run failed: {exc}")
+        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."""
+    """Async startup entry — kicks off a thread per user whose interval elapsed.
-    if not should_run_now():
+
+    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")
+    for user_id in user_ids:
-    thread.start()
+        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:
+def get_curator_status(user_id: str) -> dict:
-    """Return curator state for the API status endpoint."""
+    """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"),

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
+Now sharded per-user — each ``user_id`` gets its own deque so users only see
-server clears it. The Gateway exposes the buffer over ``GET /api/curator/logs``
+their own curator activity from the Web UI.
-so the frontend can show what the latest curator pass actually did.
 
-All public functions swallow internal exceptions so a logging failure can never
+The buffer is intentionally process-local and non-persistent. The Gateway
-break the curator run itself.
+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:
+def start_new_run(user_id: str | None = None) -> None:
-    """Clear the buffer and emit a ``run_started`` record.
+    """Clear the user's 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.
-    """
     try:
         with _lock:
-            _buffer.clear()
+            buffer = _get_or_create(user_id)
-            _buffer.append({"ts": _now_iso(), "level": "info", "message": "run_started"})
+            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:
+def record(level: str, message: str, *, user_id: str | None = None, **fields: Any) -> None:
-    """Append a single log record. Never raises.
+    """Append a single log record to the user's buffer. 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.
-    """
     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]]:
+def get_recent(user_id: str | None = None) -> list[dict[str, Any]]:
-    """Return a snapshot copy of the buffer contents. Never raises."""
+    """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:
+def clear(user_id: str | None = None) -> None:
-    """Drop all buffered records. Used by tests."""
+    """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)