xuhdandClaude Opus 4.7
4ef96c7930
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>
112 lines
3.4 KiB
Python
112 lines
3.4 KiB
Python
"""In-memory ring buffer holding the most recent curator run's execution log.
|
|
|
|
Now sharded per-user — each ``user_id`` gets its own deque so users only see
|
|
their own curator activity from the Web UI.
|
|
|
|
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
|
|
|
|
import logging
|
|
import threading
|
|
from collections import deque
|
|
from datetime import UTC, datetime
|
|
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()
|
|
_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(user_id: str | None = None) -> None:
|
|
"""Clear the user's buffer and emit a ``run_started`` record."""
|
|
try:
|
|
with _lock:
|
|
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, *, 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(),
|
|
"level": str(level or "info"),
|
|
"message": str(message),
|
|
}
|
|
if fields:
|
|
entry.update({k: v for k, v in fields.items() if k not in entry})
|
|
with _lock:
|
|
buffer = _get_or_create(user_id)
|
|
buffer.append(entry)
|
|
except Exception:
|
|
logger.debug("curator_log.record failed", exc_info=True)
|
|
|
|
|
|
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:
|
|
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(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 = _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)
|