wuyang
67 lines
1.9 KiB
Python
67 lines
1.9 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":"暂时独立保留",'
|
|
'"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
|
|
]
|