"""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()