55 lines
2.1 KiB
Python
55 lines
2.1 KiB
Python
|
"""Clusterer + evidence + cross-flavor tests (T05/T06)."""
|
||
|
|
|
||
|
|
import os
|
||
|
|
import sys
|
||
|
|
|
||
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
|
|
|
||
|
|
from session_memory.detect.cluster import cluster # noqa: E402
|
||
|
|
from session_memory.detect.signals import PROBLEM, SUCCESS, Signal # noqa: E402
|
||
|
|
|
||
|
|
|
||
|
|
def _sig(uid, flavor, repo, type_, polarity, locus, mag=1.0):
|
||
|
|
return Signal(session_uid=uid, flavor=flavor, repo=repo, type=type_,
|
||
|
|
polarity=polarity, locus=locus, magnitude=mag)
|
||
|
|
|
||
|
|
|
||
|
|
def test_min_frequency_filters_singletons():
|
||
|
|
sigs = [_sig("claude:a", "claude", "r1", "retry_storm", PROBLEM, "retries")]
|
||
|
|
assert cluster(sigs, min_frequency=2) == []
|
||
|
|
|
||
|
|
|
||
|
|
def test_clusters_recurring_signal_with_evidence():
|
||
|
|
sigs = [
|
||
|
|
_sig("claude:a", "claude", "r1", "retry_storm", PROBLEM, "retries", 5),
|
||
|
|
_sig("claude:b", "claude", "r2", "retry_storm", PROBLEM, "retries", 3),
|
||
|
|
]
|
||
|
|
pats = cluster(sigs, min_frequency=2)
|
||
|
|
assert len(pats) == 1
|
||
|
|
p = pats[0]
|
||
|
|
assert p.frequency == 2
|
||
|
|
assert p.sessions == ["claude:a", "claude:b"]
|
||
|
|
assert sorted(p.repos) == ["r1", "r2"]
|
||
|
|
assert p.flavors == ["claude"]
|
||
|
|
assert p.cross_flavor is False
|
||
|
|
assert p.cost_impact == 8.0
|
||
|
|
|
||
|
|
|
||
|
|
def test_cross_flavor_flagged_and_ranked_first():
|
||
|
|
sigs = [
|
||
|
|
# cross-flavor problem (claude + codex)
|
||
|
|
_sig("claude:a", "claude", "r1", "repeated_errors", PROBLEM, "errors", 3),
|
||
|
|
_sig("codex:b", "codex", "r2", "repeated_errors", PROBLEM, "errors", 3),
|
||
|
|
# single-flavor success cluster with higher raw impact
|
||
|
|
_sig("grok:c", "grok", "r3", "clean_pass", SUCCESS, "outcome", 5),
|
||
|
|
_sig("grok:d", "grok", "r4", "clean_pass", SUCCESS, "outcome", 5),
|
||
|
|
]
|
||
|
|
pats = cluster(sigs, min_frequency=2)
|
||
|
|
assert len(pats) == 2
|
||
|
|
xf = next(p for p in pats if p.signal_type == "repeated_errors")
|
||
|
|
assert xf.cross_flavor is True
|
||
|
|
assert sorted(xf.flavors) == ["claude", "codex"]
|
||
|
|
# cross-flavor pattern is ranked first even if another has higher raw impact
|
||
|
|
assert pats[0].cross_flavor is True
|
||
|
|
assert "cross-flavor" in pats[0].title
|