85 lines
3.2 KiB
Python
85 lines
3.2 KiB
Python
"""Tests for transparent filtering rules."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from swe_data_processing.heuristics import classify_metrics, length_bucket
|
|
|
|
|
|
def _metrics(*, failed: int = 0, calls: int = 10, streak: int = 0, tokens=50_000):
|
|
return {
|
|
"sample_id": "sample-1",
|
|
"instance_id": "repo-1",
|
|
"source_group": "example_group",
|
|
"source_parquet": "/data/example.parquet",
|
|
"resolved": 1,
|
|
"length": {"turn_count": 30, "token_count": tokens},
|
|
"tools": {
|
|
"tool_call_count": calls,
|
|
"failed_tool_call_count": failed,
|
|
"failed_tool_call_rate": failed / calls if calls else 0.0,
|
|
"longest_consecutive_failure_run": streak,
|
|
"error_positions": {
|
|
"early_count": failed,
|
|
"early_fraction": 1.0 if failed else 0.0,
|
|
"occupied_bins_5": min(failed, 5),
|
|
},
|
|
"error_type_counts": {},
|
|
"error_tool_counts": {},
|
|
},
|
|
"structure": {
|
|
"invalid_turn_count": 0,
|
|
"malformed_tool_definition_count": 0,
|
|
"malformed_tool_call_count": 0,
|
|
"unknown_tool_call_count": 0,
|
|
"missing_tool_result_count": 0,
|
|
"orphan_tool_result_count": 0,
|
|
},
|
|
}
|
|
|
|
|
|
def test_length_buckets_match_training_limits() -> None:
|
|
assert length_bucket(None) == "TOKENIZER_REQUIRED"
|
|
assert length_bucket(81_920) == "LE_81920"
|
|
assert length_bucket(81_921) == "81921_TO_131072"
|
|
assert length_bucket(131_073) == "131073_TO_262144"
|
|
assert length_bucket(262_145) == "GT_262144"
|
|
|
|
|
|
def test_five_consecutive_failures_is_hard_reject() -> None:
|
|
decision = classify_metrics(_metrics(failed=5, calls=20, streak=5), 10)
|
|
assert "FIVE_CONSECUTIVE_TOOL_FAILURES" in decision["hard_reject_reasons"]
|
|
assert decision["recommended_action"] == "DROP_DEFINITE_TOOL_PROBLEM"
|
|
|
|
|
|
def test_early_cluster_is_review_only() -> None:
|
|
decision = classify_metrics(_metrics(failed=3, calls=20, streak=2), 10)
|
|
assert decision["hard_reject_reasons"] == []
|
|
assert decision["review_flags"] == ["EARLY_FAILURE_CLUSTER"]
|
|
assert decision["recommended_action"] == "REVIEW_HEURISTIC_HIT"
|
|
|
|
|
|
def test_relative_count_outlier_is_not_automatic_drop() -> None:
|
|
decision = classify_metrics(_metrics(failed=10, calls=100, streak=2), 10)
|
|
assert decision["hard_reject_reasons"] == []
|
|
assert "EXTREME_ERROR_COUNT" in decision["review_flags"]
|
|
|
|
|
|
def test_distributed_failures_require_review_instead_of_automatic_drop() -> None:
|
|
decision = classify_metrics(_metrics(failed=8, calls=40, streak=2), 20)
|
|
assert decision["hard_reject_reasons"] == []
|
|
assert "PERSISTENT_DISTRIBUTED_FAILURES" in decision["review_flags"]
|
|
|
|
|
|
def test_failed_outcome_is_never_recommended_for_training() -> None:
|
|
metrics = _metrics()
|
|
metrics["resolved"] = 0
|
|
decision = classify_metrics(metrics, 10)
|
|
assert decision["outcome_use"] == "EXCLUDE_FROM_SUCCESS_SFT"
|
|
assert decision["recommended_action"] == "EXCLUDE_FAILED_OUTCOME"
|
|
|
|
|
|
def test_decision_keeps_source_provenance_for_targeted_review() -> None:
|
|
decision = classify_metrics(_metrics(), 10)
|
|
assert decision["source_group"] == "example_group"
|
|
assert decision["source_parquet"] == "/data/example.parquet"
|