generated from coulomb/repo-seed
tegwick
436a96dcd8
- 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>
54 lines
2.0 KiB
Python
54 lines
2.0 KiB
Python
"""Signal extractor tests (T04)."""
|
|
|
|
import os
|
|
import sys
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
from session_memory.detect.signals import ( # noqa: E402
|
|
PROBLEM, SUCCESS, build_context, extract_signals,
|
|
)
|
|
|
|
|
|
def _digest(uid, flavor="claude", repo="r", outcome="success", tokens=100,
|
|
errors=0, retries=0, test_runs=0):
|
|
return {
|
|
"session_uid": uid, "flavor": flavor, "repo": repo, "outcome": outcome,
|
|
"cost": {"input_tokens": tokens, "output_tokens": 0},
|
|
"markers": {"errors": errors, "retries": retries, "test_runs": test_runs,
|
|
"edits": 0, "human_interventions": 0},
|
|
}
|
|
|
|
|
|
def test_problem_signals():
|
|
digests = [
|
|
_digest("claude:a", retries=5, outcome="fail"),
|
|
_digest("claude:b", errors=4),
|
|
_digest("claude:c", outcome="abandoned"),
|
|
]
|
|
sigs = extract_signals(digests)
|
|
types = {(s.session_uid, s.type) for s in sigs}
|
|
assert ("claude:a", "retry_storm") in types
|
|
assert ("claude:b", "repeated_errors") in types
|
|
assert ("claude:c", "abandoned") in types
|
|
assert all(s.polarity == PROBLEM for s in sigs
|
|
if s.type in ("retry_storm", "repeated_errors", "abandoned"))
|
|
|
|
|
|
def test_success_signals():
|
|
sigs = extract_signals([_digest("grok:x", outcome="success", test_runs=2)])
|
|
assert any(s.type == "clean_pass" and s.polarity == SUCCESS for s in sigs)
|
|
|
|
rec = extract_signals([_digest("codex:y", outcome="success", errors=2)])
|
|
assert any(s.type == "error_then_recovery" and s.polarity == SUCCESS for s in rec)
|
|
|
|
|
|
def test_budget_overrun_uses_corpus_p90():
|
|
digests = [_digest(f"claude:{i}", tokens=100) for i in range(10)]
|
|
digests.append(_digest("claude:big", tokens=100000))
|
|
ctx = build_context(digests)
|
|
assert ctx["tokens_p90"] >= 100
|
|
sigs = extract_signals(digests, ctx)
|
|
overruns = [s for s in sigs if s.type == "budget_overrun"]
|
|
assert overruns and overruns[0].session_uid == "claude:big"
|