xuhdandClaude Opus 4.7
39def1c20b
调度健壮性:
- 技能进化总开关 skill_evolution.enabled 真正生效;CuratorConfig
与 CuratorSubConfig 默认 enabled 改为 false(显式 opt-in)
- 新增 per-user 运行锁(.curator_state 的 running/started_at),
防定时器与手动触发并发重复运行;6h stale 检测防进程残留死锁
- maybe_run_curator 改用 ThreadPoolExecutor 限并发(上限 5),
防首次部署所有用户同时到期的惊群
- 首次运行持续失败时按 6h 退避重试,不再每个 tick 烧 LLM 配额
- should_run_now 时间解析容忍 naive 时间串;损坏的 last_run_at
触发重跑而非永久死锁
- SkillStore 新增 list_published_names 批量接口,curator 一轮
只查一次 published,取代逐技能查库
LLM 评审超时:
- _run_llm_review 给 agent.invoke 套整体超时
(curator.review_timeout_seconds,默认 600s),超时中止本次评审
并在 curator 日志输出,curator 正常收尾、释放运行锁
- copy_context 把 user 上下文传入工作线程
模型可配置:
- 新增 curator.model_name;curator 评审与技能安全审核共用该模型,
未配置则回退主模型
- 移除独立的 moderation_model_name 配置
- 前端进化面板新增模型下拉选择器
手动触发体验:
- /api/curator/run 派发后台线程前先预检,返回 {triggered, reason},
不再无论开关状态都提示"已开始"
- 前端:总开关/curator 开关未开时禁用"立即执行"按钮,
被预检拦下时按原因(开关未开 / 已有任务在跑)给出准确提示
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
119 lines
3.5 KiB
Python
119 lines
3.5 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import threading
|
|
|
|
import pytest
|
|
|
|
from deerflow.config import skill_evolution_runtime
|
|
from deerflow.skills import curator_log
|
|
|
|
|
|
@pytest.fixture
|
|
def overrides_path(tmp_path, monkeypatch):
|
|
"""Redirect skill_evolution_runtime.resolve_path at a fresh temp file."""
|
|
target = tmp_path / "skill_evolution.json"
|
|
monkeypatch.setenv("DEER_FLOW_SKILL_EVOLUTION_CONFIG_PATH", str(target))
|
|
yield target
|
|
|
|
|
|
def test_load_overrides_returns_none_when_missing(overrides_path):
|
|
assert not overrides_path.exists()
|
|
assert skill_evolution_runtime.load_overrides() is None
|
|
|
|
|
|
def test_save_and_load_overrides_roundtrip(overrides_path):
|
|
payload = {
|
|
"enabled": True,
|
|
"curator": {
|
|
"enabled": True,
|
|
"interval_hours": 12,
|
|
"stale_after_days": 7,
|
|
"archive_after_days": 30,
|
|
"model_name": "curator-mini",
|
|
},
|
|
}
|
|
written = skill_evolution_runtime.save_overrides(payload)
|
|
assert written == overrides_path
|
|
assert overrides_path.exists()
|
|
assert skill_evolution_runtime.load_overrides() == payload
|
|
|
|
|
|
def test_load_overrides_handles_invalid_json(overrides_path):
|
|
overrides_path.write_text("{not valid json", encoding="utf-8")
|
|
assert skill_evolution_runtime.load_overrides() is None
|
|
|
|
|
|
def test_load_overrides_rejects_non_object_root(overrides_path):
|
|
overrides_path.write_text(json.dumps([1, 2, 3]), encoding="utf-8")
|
|
assert skill_evolution_runtime.load_overrides() is None
|
|
|
|
|
|
def test_app_config_merges_overrides_on_from_file(overrides_path, monkeypatch, tmp_path):
|
|
yaml_cfg = tmp_path / "config.yaml"
|
|
yaml_cfg.write_text(
|
|
"""
|
|
sandbox:
|
|
use: deerflow.sandbox.local:LocalSandboxProvider
|
|
skill_evolution:
|
|
enabled: false
|
|
curator:
|
|
interval_hours: 168
|
|
stale_after_days: 30
|
|
""".strip(),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
# Overrides flip enabled and change interval_hours.
|
|
skill_evolution_runtime.save_overrides(
|
|
{
|
|
"enabled": True,
|
|
"curator": {"interval_hours": 24},
|
|
}
|
|
)
|
|
|
|
from deerflow.config.app_config import AppConfig
|
|
|
|
cfg = AppConfig.from_file(str(yaml_cfg))
|
|
assert cfg.skill_evolution.enabled is True
|
|
assert cfg.skill_evolution.curator.interval_hours == 24
|
|
# Untouched yaml values remain.
|
|
assert cfg.skill_evolution.curator.stale_after_days == 30
|
|
|
|
|
|
def test_curator_log_start_new_run_resets_and_seeds_record():
|
|
curator_log.clear()
|
|
curator_log.record("info", "old record")
|
|
assert len(curator_log.get_recent()) == 1
|
|
|
|
curator_log.start_new_run()
|
|
records = curator_log.get_recent()
|
|
assert len(records) == 1
|
|
assert records[0]["message"] == "run_started"
|
|
assert records[0]["level"] == "info"
|
|
|
|
|
|
def test_curator_log_record_is_thread_safe():
|
|
curator_log.clear()
|
|
|
|
def writer(idx: int) -> None:
|
|
for i in range(20):
|
|
curator_log.record("info", f"thread-{idx}-{i}")
|
|
|
|
threads = [threading.Thread(target=writer, args=(i,)) for i in range(5)]
|
|
for t in threads:
|
|
t.start()
|
|
for t in threads:
|
|
t.join()
|
|
|
|
records = curator_log.get_recent()
|
|
# 5 threads * 20 records each = 100, well under the 200 ring-buffer cap.
|
|
assert len(records) == 100
|
|
assert all("message" in r and "ts" in r for r in records)
|
|
|
|
|
|
def test_curator_log_record_ignores_internal_errors(monkeypatch):
|
|
# Even if the timestamp helper blows up, record() must not raise.
|
|
monkeypatch.setattr(curator_log, "_now_iso", lambda: (_ for _ in ()).throw(RuntimeError("boom")))
|
|
curator_log.record("info", "should not raise")
|