wuyang
136 lines
4.3 KiB
Python
136 lines
4.3 KiB
Python
import asyncio
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
from agents.exceptions import MaxTurnsExceeded
|
|
|
|
from app.config import Settings, UserSeed
|
|
from app.curator import Curator, Runner
|
|
from app.db import Database
|
|
|
|
|
|
def test_tool_loop_uses_no_tool_convergence_repair(
|
|
tmp_path: Path, monkeypatch
|
|
):
|
|
admin = UserSeed(
|
|
id="admin",
|
|
label="管理员",
|
|
role="admin",
|
|
access_key_hash="unused",
|
|
)
|
|
database = Database(tmp_path / "repair.sqlite3")
|
|
database.initialize([admin])
|
|
fragment = database.create_fragment("admin", "一个需要收敛的念头")
|
|
settings = Settings(
|
|
data_dir=tmp_path,
|
|
database_path=tmp_path / "repair.sqlite3",
|
|
users=(admin,),
|
|
session_secret="unused",
|
|
deepseek_api_key="test-key",
|
|
deepseek_base_url="https://api.deepseek.com",
|
|
deepseek_model="deepseek-v4-pro",
|
|
cookie_secure=False,
|
|
auth_disabled=True,
|
|
)
|
|
calls = 0
|
|
|
|
async def fake_run(*args, **kwargs):
|
|
nonlocal calls
|
|
calls += 1
|
|
if calls == 1:
|
|
raise MaxTurnsExceeded("Max turns (6) exceeded")
|
|
assert kwargs["max_turns"] == 2
|
|
assert args[0].tools == []
|
|
return SimpleNamespace(
|
|
final_output=(
|
|
'{"standalone":true,"reasoning_note":"暂时独立保留",'
|
|
'"direction_signal":"none","assessments":[]}'
|
|
)
|
|
)
|
|
|
|
monkeypatch.setattr(Runner, "run", fake_run)
|
|
curator = Curator(database, settings)
|
|
|
|
asyncio.run(curator.analyze(fragment["id"]))
|
|
|
|
assert calls == 2
|
|
assert database.get_fragment(fragment["id"], "admin")[
|
|
"analysis_status"
|
|
] == "done"
|
|
run = database.list_agent_runs(1)[0]
|
|
events = database.list_agent_events(run["id"])
|
|
assert run["status"] == "success"
|
|
assert run["attempt_count"] == 2
|
|
assert "convergence_repair_started" in [
|
|
event["event_type"] for event in events
|
|
]
|
|
|
|
|
|
def test_curator_recovers_visible_prior_fragment(
|
|
tmp_path: Path, monkeypatch
|
|
):
|
|
admin = UserSeed(
|
|
id="admin",
|
|
label="管理员",
|
|
role="admin",
|
|
access_key_hash="unused",
|
|
)
|
|
database = Database(tmp_path / "recovery.sqlite3")
|
|
database.initialize([admin])
|
|
earlier = database.create_fragment(
|
|
"admin", "有个新工作可能要做小爱架构。"
|
|
)
|
|
database.apply_assessments("admin", earlier["id"], [])
|
|
current = database.create_fragment(
|
|
"admin", "确定要做，先熟悉 harness 5.0。"
|
|
)
|
|
settings = Settings(
|
|
data_dir=tmp_path,
|
|
database_path=tmp_path / "recovery.sqlite3",
|
|
users=(admin,),
|
|
session_secret="unused",
|
|
deepseek_api_key="test-key",
|
|
deepseek_base_url="https://api.deepseek.com",
|
|
deepseek_model="deepseek-v4-pro",
|
|
cookie_secure=False,
|
|
auth_disabled=True,
|
|
)
|
|
|
|
async def fake_run(*args, **kwargs):
|
|
return SimpleNamespace(
|
|
final_output=(
|
|
"{"
|
|
'"direction_signal":"explicit",'
|
|
'"standalone":false,'
|
|
'"reasoning_note":"明确的新工作，应形成可跟踪脉络",'
|
|
'"assessments":[{'
|
|
'"idea_id":null,'
|
|
'"title":"小爱新架构开发",'
|
|
'"summary":"熟悉现有方案并推进新架构开发。",'
|
|
'"maturity":10,'
|
|
'"confidence":0.9,'
|
|
'"motion":"启动中",'
|
|
'"position":"先熟悉 harness 5.0 核心方案。",'
|
|
'"tension":"任务已明确，但范围与现有方案尚未理解。",'
|
|
'"trajectory":"从可能插入的工作变成明确承诺。",'
|
|
'"possible_moves":["阅读核心方案"],'
|
|
'"relevance":1,'
|
|
f'"supporting_fragment_ids":["{earlier["id"]}"]'
|
|
"}]}"
|
|
)
|
|
)
|
|
|
|
monkeypatch.setattr(Runner, "run", fake_run)
|
|
curator = Curator(database, settings)
|
|
|
|
asyncio.run(curator.analyze(current["id"]))
|
|
|
|
idea = database.get_idea(
|
|
"admin", database.list_ideas("admin")[0]["id"]
|
|
)
|
|
assert idea["title"] == "小爱新架构开发"
|
|
assert [item["id"] for item in idea["fragments"]] == [
|
|
earlier["id"],
|
|
current["id"],
|
|
]
|