generated from coulomb/repo-seed
- detect/signals.py: pure extractors over digests (retry storm, repeated errors, budget overrun vs corpus p90, abandoned, clean pass, recovery) - detect/cluster.py: deterministic clustering into candidate Patterns with evidence (sessions/repos/flavors/cost impact) + cross-flavor flagging - detect/__main__.py: python -m session_memory.detect, ranked report (cross-flavor first) + --json; persists candidates to Tier 2 patterns table - core/store.py: list_digests + save_patterns - tests for signals, cluster, detect entrypoint Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
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
|