157 lines
6.5 KiB
Python
157 lines
6.5 KiB
Python
"""The admin timing page explains whether a wait is the skill check or the command."""
|
|
|
|
from datetime import UTC, datetime, timedelta
|
|
|
|
from deerflow.sandbox.skill_view import (
|
|
_EVENT_LOCK,
|
|
_SKILL_VIEW_EVENTS,
|
|
begin_skill_view_trace,
|
|
commit_skill_view_trace,
|
|
diagnose_skill_view,
|
|
end_skill_view_trace,
|
|
explain_skill_view_event,
|
|
note_skill_view_check,
|
|
query_skill_view_events,
|
|
recent_skill_view_events,
|
|
record_skill_view_event,
|
|
remember_invoked_skill,
|
|
reset_skill_view_events,
|
|
)
|
|
|
|
|
|
def _event(status: str, total_ms: float, **extra) -> dict:
|
|
return {
|
|
"status": status,
|
|
"total_ms": total_ms,
|
|
"lock_wait_ms": extra.get("lock_wait_ms", 0),
|
|
"source_entries": extra.get("source_entries", 0),
|
|
"path_count": extra.get("path_count", 0),
|
|
}
|
|
|
|
|
|
def test_switch_card_is_omitted():
|
|
report = diagnose_skill_view(cache_enabled=False, reuse_enabled=False, events=[])
|
|
assert all(item["id"] != "switches" for item in report["checks"])
|
|
|
|
|
|
def test_fast_skip_points_at_the_command_itself():
|
|
report = diagnose_skill_view(
|
|
cache_enabled=True,
|
|
reuse_enabled=True,
|
|
events=[_event("skipped", 12), _event("narrow", 40, path_count=1)],
|
|
)
|
|
assert report["level"] == "ok"
|
|
latest = next(item for item in report["checks"] if item["id"] == "latest")
|
|
assert "命令" in latest["look_at"] or "脚本" in latest["cause"]
|
|
repeat = next(item for item in report["checks"] if item["id"] == "repeat")
|
|
assert repeat["level"] == "ok"
|
|
|
|
|
|
def test_repeated_full_refresh_is_called_out():
|
|
events = [_event("refreshed", 18000, source_entries=4000), _event("hit", 9000, source_entries=4000)]
|
|
report = diagnose_skill_view(cache_enabled=True, reuse_enabled=True, events=events)
|
|
assert report["level"] == "bad"
|
|
assert "4000" in report["conclusion"]
|
|
repeat = next(item for item in report["checks"] if item["id"] == "repeat")
|
|
assert "2 条" in repeat["result"]
|
|
|
|
|
|
def test_lock_wait_is_separated_from_a_file_scan():
|
|
explained = explain_skill_view_event(_event("joined", 12000, lock_wait_ms=11000))
|
|
assert explained["level"] == "warn"
|
|
assert "等" in explained["headline"]
|
|
|
|
|
|
def test_command_and_skill_are_saved_only_after_the_call_returns():
|
|
reset_skill_view_events()
|
|
try:
|
|
note_skill_view_check(status="narrow", total_ms=30, path_count=1)
|
|
remember_invoked_skill("task-report-build")
|
|
assert recent_skill_view_events() == []
|
|
token = begin_skill_view_trace()
|
|
try:
|
|
note_skill_view_check(status="narrow", total_ms=30, path_count=1)
|
|
remember_invoked_skill("task-report-build")
|
|
assert recent_skill_view_events() == []
|
|
commit_skill_view_trace(
|
|
tool="bash",
|
|
command="python /mnt/skills/public/task-report-build/scripts/run.py\necho done",
|
|
skill_names=["directory-name"],
|
|
tool_ms=8000,
|
|
allow_empty=True,
|
|
)
|
|
finally:
|
|
end_skill_view_trace(token)
|
|
event = recent_skill_view_events()[0]
|
|
assert event["status"] == "narrow"
|
|
assert event["tool"] == "bash"
|
|
assert event["command"] == "python /mnt/skills/public/task-report-build/scripts/run.py echo done"
|
|
assert event["skill_names"] == ["task-report-build"]
|
|
assert event["tool_ms"] == 8000
|
|
report = diagnose_skill_view(cache_enabled=True, reuse_enabled=True)
|
|
latest = next(item for item in report["checks"] if item["id"] == "latest")
|
|
assert latest["skill_names"] == ["task-report-build"]
|
|
assert "task-report-build" in report["events"][0]["explanation"]["headline"]
|
|
finally:
|
|
reset_skill_view_events()
|
|
|
|
|
|
def test_diagnostic_write_cannot_escape_into_the_tool(monkeypatch):
|
|
def boom(**kwargs):
|
|
raise RuntimeError("diagnostic store failed")
|
|
|
|
monkeypatch.setattr("deerflow.sandbox.skill_view.record_skill_view_event", boom)
|
|
token = begin_skill_view_trace()
|
|
try:
|
|
note_skill_view_check(status="skipped")
|
|
commit_skill_view_trace(tool="bash", command="echo hello", allow_empty=True)
|
|
finally:
|
|
end_skill_view_trace(token)
|
|
|
|
|
|
def test_event_buffer_keeps_the_newest_sample_first():
|
|
reset_skill_view_events()
|
|
try:
|
|
record_skill_view_event(status="skipped", total_ms=5)
|
|
record_skill_view_event(status="narrow", total_ms=20, path_count=1)
|
|
report = diagnose_skill_view(cache_enabled=True, reuse_enabled=True)
|
|
assert report["events"][0]["status"] == "narrow"
|
|
assert report["events"][1]["status"] == "skipped"
|
|
assert report["events"][0]["explanation"]["level"] == "ok"
|
|
finally:
|
|
reset_skill_view_events()
|
|
|
|
|
|
def test_events_older_than_one_day_are_dropped():
|
|
reset_skill_view_events()
|
|
try:
|
|
record_skill_view_event(status="narrow", total_ms=10, tool="bash", command="echo now", tool_ms=100)
|
|
record_skill_view_event(status="skipped", total_ms=5, tool="bash", command="echo old", tool_ms=9000)
|
|
old = (datetime.now(UTC) - timedelta(days=2)).isoformat(timespec="seconds")
|
|
with _EVENT_LOCK:
|
|
_SKILL_VIEW_EVENTS[1]["at"] = old
|
|
page = query_skill_view_events(page=1, page_size=20, sort="time", order="desc")
|
|
assert page["total"] == 1
|
|
assert [item["command"] for item in page["events"]] == ["echo now"]
|
|
finally:
|
|
reset_skill_view_events()
|
|
|
|
|
|
def test_events_can_be_paged_and_sorted_by_time_or_duration():
|
|
reset_skill_view_events()
|
|
try:
|
|
record_skill_view_event(status="skipped", total_ms=10, tool="bash", command="first", tool_ms=100)
|
|
record_skill_view_event(status="narrow", total_ms=4000, tool="bash", command="second", tool_ms=50)
|
|
record_skill_view_event(status="skipped", total_ms=20, tool="read_file", command="third", tool_ms=8000)
|
|
by_duration = query_skill_view_events(page=1, page_size=2, sort="duration", order="desc")
|
|
assert by_duration["total"] == 3
|
|
assert by_duration["page"] == 1
|
|
assert [item["command"] for item in by_duration["events"]] == ["third", "first"]
|
|
by_check = query_skill_view_events(page=1, page_size=3, sort="check", order="desc")
|
|
assert [item["command"] for item in by_check["events"]] == ["second", "third", "first"]
|
|
by_time = query_skill_view_events(page=2, page_size=2, sort="time", order="asc")
|
|
assert by_time["page"] == 2
|
|
assert [item["command"] for item in by_time["events"]] == ["third"]
|
|
finally:
|
|
reset_skill_view_events()
|