412 lines
16 KiB
Python
412 lines
16 KiB
Python
"""Tests for progress_detect.py — progress detection signals.
|
|
|
|
Covers target_of extraction, window stats, coverage, evaluate
|
|
logic (normal, read-only, landed suppression, slow repeat),
|
|
and the full fixture-driven label verification.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
|
|
from progress_detect import (
|
|
DetectConfig,
|
|
_normalize_bash,
|
|
coverage,
|
|
evaluate,
|
|
target_of,
|
|
window_stats,
|
|
)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _call(t, tool, args, landed=False):
|
|
"""Build a call tuple as progress_detect.evaluate expects."""
|
|
return (t, tool, json.dumps(args), landed)
|
|
|
|
|
|
def _calls_from_fixture(session):
|
|
"""Convert a fixture session dict to call tuples."""
|
|
return [
|
|
(c["t"], c["tool"], json.dumps(c["args"]), c["landed"])
|
|
for c in session["calls"]
|
|
]
|
|
|
|
|
|
def _file_lines_fn(file_lines_dict):
|
|
"""Return a callable for coverage() from a file-path→line-count dict."""
|
|
def fn(fp):
|
|
if fp is None:
|
|
return 0
|
|
return file_lines_dict.get(fp, 0) or 0
|
|
return fn
|
|
|
|
|
|
_DEFAULT_CFG = DetectConfig()
|
|
_FIXTURE_PATH = os.path.join(
|
|
os.path.dirname(__file__), "fixtures", "progress", "fixture.json"
|
|
)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# target_of
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestTargetOf:
|
|
"""target_of(tool, args_json) -> tuple key for counting."""
|
|
|
|
def test_read_with_offset_limit(self):
|
|
"""read with filePath, offset, limit -> file key with offset/limit."""
|
|
args = json.dumps({"filePath": "/home/user/proj/main.py", "offset": 10, "limit": 50})
|
|
result = target_of("read", args)
|
|
assert result[0] == "file"
|
|
assert result[1] == "main.py"
|
|
assert result[2] == 10
|
|
assert result[3] == 50
|
|
|
|
def test_read_without_offset_limit(self):
|
|
"""read with filePath only -> file key, offset/limit remain None."""
|
|
args = json.dumps({"filePath": "/home/user/proj/data.json"})
|
|
result = target_of("read", args)
|
|
assert result[0] == "file"
|
|
assert result[1] == "data.json"
|
|
assert result[2] is None
|
|
assert result[3] is None
|
|
|
|
def test_bash_with_git_files(self):
|
|
"""bash with git commands referencing files -> bash-files key."""
|
|
args = json.dumps({"command": "git add main.py && git commit -m 'fix' && git push origin main"})
|
|
result = target_of("bash", args)
|
|
assert result[0] == "bash-files"
|
|
assert "main.py" in result[1]
|
|
|
|
def test_bash_no_files(self):
|
|
"""bash with no file-ish args -> bash key with first words."""
|
|
args = json.dumps({"command": "echo hello world"})
|
|
result = target_of("bash", args)
|
|
assert result[0] == "bash"
|
|
assert "echo" in result[1]
|
|
|
|
def test_edit_with_filepath(self):
|
|
"""edit with filePath -> file key only."""
|
|
args = json.dumps({"filePath": "/home/user/proj/app.py", "oldString": "foo", "newString": "bar"})
|
|
result = target_of("edit", args)
|
|
assert result[0] == "file"
|
|
assert result[1] == "app.py"
|
|
|
|
def test_grep_with_pattern(self):
|
|
"""grep -> (grep, pattern[:60])."""
|
|
args = json.dumps({"pattern": "def test_"})
|
|
result = target_of("grep", args)
|
|
assert result[0] == "grep"
|
|
assert "def test_" in result[1]
|
|
|
|
def test_glob_with_pattern(self):
|
|
"""glob -> (glob, pattern[:60])."""
|
|
args = json.dumps({"pattern": "**/*.py"})
|
|
result = target_of("glob", args)
|
|
assert result[0] == "glob"
|
|
assert "**/*.py" in result[1]
|
|
|
|
def test_invalid_json_fallback(self):
|
|
"""Invalid JSON -> returns (tool, truncated args)."""
|
|
result = target_of("read", "{{{bad json}}}")
|
|
assert result[0] == "read"
|
|
assert len(result[1]) <= 80
|
|
|
|
def test_unknown_tool_fallback(self):
|
|
"""Unknown tool with valid JSON -> (tool, truncated JSON)."""
|
|
result = target_of("some_tool", json.dumps({"k": "v"}))
|
|
assert result[0] == "some_tool"
|
|
|
|
def test_normalize_bash_comment_lines(self):
|
|
"""20 distinct bash-comment commands produce 20 distinct keys."""
|
|
tokens = set()
|
|
for i in range(20):
|
|
cmd = f"# Check file{i}: do something{i}"
|
|
tokens.add(_normalize_bash(cmd))
|
|
assert len(tokens) == 20
|
|
|
|
def test_normalize_bash_cd_prefix(self):
|
|
"""cd <dir> && ... prefix is stripped so different commands differ."""
|
|
tokens = set()
|
|
for i in range(20):
|
|
cmd = f"cd /x && echo file{i}"
|
|
tokens.add(_normalize_bash(cmd))
|
|
assert len(tokens) == 20
|
|
|
|
def test_normalize_bash_env_prefix(self):
|
|
"""VAR=val... prefix is stripped so different commands differ."""
|
|
r1 = _normalize_bash("PYTHONPATH=src python -m a")
|
|
r2 = _normalize_bash("PYTHONPATH=src python -m b")
|
|
assert r1 != r2
|
|
assert "python -m a" == r1
|
|
assert "python -m b" == r2
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# window_stats
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestWindowStats:
|
|
"""window_stats(calls, n) -> (dup, top_n, slow_count)."""
|
|
|
|
def test_duplicate_calls_have_positive_dup(self):
|
|
"""Repeated (tool, args) pairs -> dup > 0."""
|
|
calls = [_call(0, "read", {"filePath": "/x.py"})] * 10 + \
|
|
[_call(1, "read", {"filePath": "/y.py"})] * 10
|
|
dup, _top_n, _slow_count = window_stats(calls, 20)
|
|
assert dup > 0
|
|
|
|
def test_all_unique_calls_have_zero_dup(self):
|
|
"""Every call has distinct (tool, args) -> dup = 0."""
|
|
calls = [_call(i, "read", {"filePath": f"/file_{i}.py"}) for i in range(20)]
|
|
dup, _top_n, _slow_count = window_stats(calls, 20)
|
|
assert dup == 0.0
|
|
|
|
def test_window_shorter_than_calls(self):
|
|
"""Only last n calls contribute to window stats."""
|
|
calls = [_call(i, "read", {"filePath": "/x.py"}) for i in range(100)] + \
|
|
[_call(i, "bash", {"command": f"echo {i}"}) for i in range(100, 200)]
|
|
dup, _top_n, _slow_count = window_stats(calls, 10)
|
|
# Last 10 are bash with distinct commands -> all unique -> dup = 0
|
|
assert dup == 0.0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# coverage
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestCoverage:
|
|
"""coverage(calls, n, file_lines) -> (ratio, filename)."""
|
|
|
|
def test_reads_from_single_file_produce_coverage(self):
|
|
"""Repeated reads of one file produce coverage > 0."""
|
|
calls = [_call(i, "read", {"filePath": "/project/main.py", "limit": 100}) for i in range(10)]
|
|
fl = {"/project/main.py": 200}
|
|
cov, fname = coverage(calls, 20, _file_lines_fn(fl))
|
|
assert cov > 0
|
|
assert fname == "main.py"
|
|
|
|
def test_no_read_returns_zero(self):
|
|
"""No 'read' calls -> (0.0, '')."""
|
|
calls = [_call(i, "bash", {"command": "ls"}) for i in range(10)]
|
|
cov, fname = coverage(calls, 20, _file_lines_fn({}))
|
|
assert cov == 0.0
|
|
assert fname == ""
|
|
|
|
def test_empty_window_returns_zero(self):
|
|
"""Empty call list -> (0.0, '')."""
|
|
cov, fname = coverage([], 0, _file_lines_fn({}))
|
|
assert cov == 0.0
|
|
assert fname == ""
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# evaluate — pure logic
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestEvaluate:
|
|
"""evaluate() — signal detection."""
|
|
|
|
def test_not_enough_calls(self):
|
|
"""< 40 calls -> (False, None)."""
|
|
calls = [_call(i, "read", {"filePath": "/x.py"}) for i in range(30)]
|
|
flag, reason = evaluate(calls, _DEFAULT_CFG, _file_lines_fn({}), "test")
|
|
assert flag is False
|
|
assert reason is None
|
|
|
|
def test_flags_normal_high_dup(self):
|
|
"""Normal agent, no landed, high dup -> flagged."""
|
|
calls = [_call(i, "read", {"filePath": "/x.py"}) for i in range(100)]
|
|
flag, reason = evaluate(calls, _DEFAULT_CFG, _file_lines_fn({}), "normal agent")
|
|
assert flag is True
|
|
assert reason["dup"] >= _DEFAULT_CFG.dup_min
|
|
assert reason["landed"] is False
|
|
assert reason["ro"] is False
|
|
|
|
def test_flags_ro_high_top(self):
|
|
"""Read-only agent, high top -> flagged (lower threshold)."""
|
|
calls = [_call(i, "read", {"filePath": "/x.py"}) for i in range(100)]
|
|
ro_cfg = DetectConfig(read_only_agent_keywords=("explore", "librarian", "oracle"))
|
|
flag, reason = evaluate(calls, ro_cfg, _file_lines_fn({}), "explore subagent")
|
|
assert flag is True
|
|
assert reason["ro"] is True
|
|
|
|
def test_skips_with_landed(self):
|
|
"""Landed call within window's time span -> not flagged."""
|
|
# 60 repeat reads; position 55 has landed=True within the time span
|
|
calls = [_call(i, "read", {"filePath": "/x.py"}) for i in range(60)]
|
|
calls[55] = _call(55, "read", {"filePath": "/x.py"}, landed=True)
|
|
flag, _reason = evaluate(calls, _DEFAULT_CFG, _file_lines_fn({}), "test")
|
|
assert flag is False
|
|
|
|
def test_slow_signal(self):
|
|
"""Single exact (tool, args) pair repeated 15+ times across all calls -> slow flagged."""
|
|
calls = [_call(i, "read", {"filePath": "/x.py"}) for i in range(100)]
|
|
# window must be >= min_calls to pass the early-exit guard
|
|
cfg = DetectConfig(window=50, min_calls=40, cum_min=15)
|
|
flag, reason = evaluate(calls, cfg, _file_lines_fn({}), "test")
|
|
assert flag is True
|
|
# The slow signal fires when slow_count >= cum_min; the exact-total
|
|
# cosmetic relabel only happens when top_n is lower than top_min,
|
|
# which is not the case when every call targets the same file.
|
|
assert reason["top"] >= cfg.cum_min
|
|
|
|
def test_evaluate_flag_reason_has_expected_keys(self):
|
|
"""Flagged evaluation returns a reason dict with all expected keys."""
|
|
calls = [_call(i, "read", {"filePath": "/x.py"}) for i in range(100)]
|
|
flag, reason = evaluate(calls, _DEFAULT_CFG, _file_lines_fn({}), "test")
|
|
assert flag is True
|
|
for key in ("flag", "dup", "top", "top_what", "landed", "ro", "n", "t1"):
|
|
assert key in reason
|
|
|
|
def test_evaluate_exact_min_calls_quiet(self):
|
|
"""Exactly min_calls (40) with diverse commands -> not flagged."""
|
|
calls = [_call(i, "bash", {"command": f"echo {i}"}) for i in range(40)]
|
|
flag, reason = evaluate(calls, _DEFAULT_CFG, _file_lines_fn({}), "test")
|
|
assert flag is False
|
|
assert reason is None
|
|
|
|
def test_evaluate_extra_landed_times(self):
|
|
"""extra_landed_times inside window gates dup/top signals -> not flagged."""
|
|
calls = [_call(i, "read", {"filePath": "/x.py"}) for i in range(80)]
|
|
# No own landed, but extra_landed_times has a time inside [t0, t1]
|
|
extra = [40.0] # well within the 80-call window's time span
|
|
flag, reason = evaluate(calls, _DEFAULT_CFG, _file_lines_fn({}),
|
|
"test", extra_landed_times=extra)
|
|
assert flag is False, "Should not flag when extra landed times are inside window"
|
|
if reason:
|
|
assert reason["landed"] is True
|
|
|
|
def test_extra_landed_times_outside_window(self):
|
|
"""extra_landed_times outside window -> still flagged (not landed)."""
|
|
calls = [_call(i, "read", {"filePath": "/x.py"}) for i in range(80)]
|
|
# extra time far outside window's time span
|
|
extra = [999999.0]
|
|
flag, reason = evaluate(calls, _DEFAULT_CFG, _file_lines_fn({}),
|
|
"test", extra_landed_times=extra)
|
|
assert flag is True, "Should still flag when extra landed times are outside window"
|
|
assert reason["landed"] is False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# fixture labels — end-to-end validation
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestFixtureLabels:
|
|
"""Load the fixture and verify every session's label against the detector.
|
|
|
|
Uses a sliding window (step 5) from min_calls to len(calls).
|
|
"""
|
|
|
|
@staticmethod
|
|
def _load_fixture():
|
|
with open(_FIXTURE_PATH) as f:
|
|
return json.load(f)
|
|
|
|
def test_all(self):
|
|
sessions = self._load_fixture()
|
|
cfg = _DEFAULT_CFG
|
|
|
|
must_flag = [s for s in sessions if s["label"] == "must_flag"]
|
|
must_not = [s for s in sessions if s["label"] == "must_not_flag"]
|
|
|
|
# --- must_not_flag sessions: never flag ----------------------------
|
|
for s in must_not:
|
|
sid = s["session_id"]
|
|
calls = _calls_from_fixture(s)
|
|
fl = s.get("file_lines", {})
|
|
fl_fn = _file_lines_fn(fl)
|
|
flagged = False
|
|
for i in range(cfg.min_calls, len(calls) + 1, 5):
|
|
r = evaluate(calls, cfg, fl_fn, s["title"], end_idx=i)
|
|
if r and r[0]:
|
|
flagged = True
|
|
break
|
|
assert not flagged, (
|
|
f"Session {sid} (must_not_flag) should not flag but did at end_idx={i}"
|
|
)
|
|
|
|
# --- must_flag sessions: at least one window flags ----------------
|
|
for s in must_flag:
|
|
sid = s["session_id"]
|
|
calls = _calls_from_fixture(s)
|
|
fl = s.get("file_lines", {})
|
|
fl_fn = _file_lines_fn(fl)
|
|
flagged = False
|
|
for i in range(cfg.min_calls, len(calls) + 1, 5):
|
|
r = evaluate(calls, cfg, fl_fn, s["title"], end_idx=i)
|
|
if r and r[0]:
|
|
flagged = True
|
|
break
|
|
assert flagged, (
|
|
f"Session {sid} (must_flag) should flag but did not"
|
|
)
|
|
|
|
def test_atlas_207_boundary(self):
|
|
"""Atlas session: no flag before 02:07; at least one flag after.
|
|
|
|
02:07 (02:07:36 UTC on Sep 26) is the wall-clock time of the first
|
|
flagged window's last call. Before that timestamp there are fewer
|
|
than min_calls (40), so evaluation cannot produce a signal.
|
|
"""
|
|
sessions = self._load_fixture()
|
|
atlas = next(s for s in sessions if s["session_id"] == "ses_f2506ac70ffebLUxBsq0aauOn9")
|
|
calls = _calls_from_fixture(atlas)
|
|
fl_fn = _file_lines_fn(atlas.get("file_lines", {}))
|
|
cfg = _DEFAULT_CFG
|
|
|
|
# 02:07 wall-clock cutoff — first flagged window's last-call timestamp
|
|
cutoff = 1790402856356
|
|
pre_flagged = []
|
|
post_flagged = []
|
|
for i in range(cfg.min_calls, len(calls) + 1, 5):
|
|
subset = calls[:i]
|
|
if len(subset) < cfg.min_calls:
|
|
continue
|
|
win = subset[-cfg.window:]
|
|
if len(win) < cfg.min_calls:
|
|
continue
|
|
win_end = win[-1][0]
|
|
r = evaluate(calls, cfg, fl_fn, atlas["title"], end_idx=i)
|
|
if r and r[0]:
|
|
if win_end < cutoff:
|
|
pre_flagged.append(i)
|
|
else:
|
|
post_flagged.append(i)
|
|
|
|
assert len(pre_flagged) == 0, (
|
|
f"Expected no flags before 02:07, got {len(pre_flagged)} at indices {pre_flagged}"
|
|
)
|
|
assert len(post_flagged) >= 1, (
|
|
"Expected at least one flag after 02:07, got none"
|
|
)
|
|
|
|
def test_specific_sessions_flag(self):
|
|
"""Two explicitly required sessions must flag."""
|
|
sessions = self._load_fixture()
|
|
cfg = _DEFAULT_CFG
|
|
|
|
for sid in (
|
|
"ses_f24fcf651ffeOm9O4E14QSZeJ4", # item 2 worker 1st attempt, 307 calls
|
|
"ses_f235d755bffeo0uWtD1g2HwQUg", # helper printed spec 12x, 81 calls
|
|
):
|
|
s = next(x for x in sessions if x["session_id"] == sid)
|
|
calls = _calls_from_fixture(s)
|
|
fl_fn = _file_lines_fn(s.get("file_lines", {}))
|
|
flagged = False
|
|
for i in range(cfg.min_calls, len(calls) + 1, 5):
|
|
r = evaluate(calls, cfg, fl_fn, s["title"], end_idx=i)
|
|
if r and r[0]:
|
|
flagged = True
|
|
break
|
|
assert flagged, (
|
|
f"Session {sid} must flag but did not"
|
|
) |