generated from coulomb/repo-seed
session-memory Phase 1: Detect pipeline (T04-T07)
- 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>
This commit is contained in:
54
tests/test_cluster.py
Normal file
54
tests/test_cluster.py
Normal file
@@ -0,0 +1,54 @@
|
||||
"""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
|
||||
Reference in New Issue
Block a user