94 lines
2.9 KiB
Python
94 lines
2.9 KiB
Python
"""The latency diagnosis agent stays evidence-first and does not overwrite edits."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import json
|
|
from pathlib import Path
|
|
|
|
|
|
def _load_reader():
|
|
path = (
|
|
Path(__file__).resolve().parents[1]
|
|
/ "app"
|
|
/ "gateway"
|
|
/ "routers"
|
|
/ "_latency_diagnosis_seed_assets"
|
|
/ "skills"
|
|
/ "latency-diagnosis"
|
|
/ "scripts"
|
|
/ "read_records.py"
|
|
)
|
|
spec = importlib.util.spec_from_file_location("latency_read_records", path)
|
|
assert spec and spec.loader
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
def test_soul_checks_command_and_disk_before_the_model() -> None:
|
|
from app.gateway.routers import _latency_diagnosis_seed as seed
|
|
|
|
soul = (seed._ASSETS_DIR / seed.LATENCY_DIAGNOSIS_AGENT_ID / "SOUL.md").read_text(encoding="utf-8")
|
|
command_rule = soul.index("整次接近这段空等，检查只有几十毫秒")
|
|
model_rule = soul.index("这时才查模型日志")
|
|
assert command_rule < model_rule
|
|
assert "这段不要写成模型问题" in soul
|
|
assert "剩余空间" in soul
|
|
assert "first_chunk_ms" in soul
|
|
|
|
|
|
def test_seed_is_idempotent_and_preserves_local_edits(tmp_path, monkeypatch) -> None:
|
|
from app.gateway.routers import _latency_diagnosis_seed as seed
|
|
|
|
class FakePaths:
|
|
def agent_dir(self, agent_id: str) -> Path:
|
|
return tmp_path / "agents" / agent_id
|
|
|
|
monkeypatch.setattr(seed, "get_paths", lambda: FakePaths())
|
|
monkeypatch.setattr(
|
|
seed,
|
|
"_ensure_skill",
|
|
lambda: None,
|
|
)
|
|
|
|
assert seed.ensure_latency_diagnosis_agent() is True
|
|
soul = tmp_path / "agents" / seed.LATENCY_DIAGNOSIS_AGENT_ID / "SOUL.md"
|
|
soul.write_text("现场调整", encoding="utf-8")
|
|
|
|
assert seed.ensure_latency_diagnosis_agent() is False
|
|
assert soul.read_text(encoding="utf-8") == "现场调整"
|
|
|
|
|
|
def test_reader_prints_check_and_total_without_inventing_rows(tmp_path, capsys) -> None:
|
|
reader = _load_reader()
|
|
missing = tmp_path / "missing"
|
|
reader._candidates = lambda: [missing]
|
|
reader.main()
|
|
assert "没有找到" in capsys.readouterr().out
|
|
|
|
path = tmp_path / "skill-view-diagnostics.jsonl"
|
|
path.write_text(
|
|
json.dumps(
|
|
{
|
|
"at": "2026-10-02T10:00:00+00:00",
|
|
"tool": "bash",
|
|
"status": "narrow",
|
|
"total_ms": 30,
|
|
"tool_ms": 40000,
|
|
"lock_wait_ms": 0,
|
|
"skill_names": ["demo"],
|
|
"command": "echo hi",
|
|
},
|
|
ensure_ascii=False,
|
|
)
|
|
+ "\n",
|
|
encoding="utf-8",
|
|
)
|
|
reader._candidates = lambda: [path]
|
|
reader.main()
|
|
printed = capsys.readouterr().out
|
|
assert "total_ms=30" in printed
|
|
assert "tool_ms=40000" in printed
|
|
assert "echo hi" in printed
|