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>
45 lines
1.7 KiB
Python
45 lines
1.7 KiB
Python
"""Detect entrypoint tests (T07): end-to-end digests -> patterns, persisted."""
|
|
|
|
import os
|
|
import sys
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
from session_memory.core.store import Store # noqa: E402
|
|
from session_memory.detect.__main__ import run_detect # noqa: E402
|
|
|
|
|
|
def _digest(uid, flavor, repo, **markers):
|
|
return {
|
|
"session_uid": uid, "flavor": flavor, "repo": repo, "outcome": "fail",
|
|
"cost": {"input_tokens": 10, "output_tokens": 1},
|
|
"markers": {"errors": markers.get("errors", 0), "retries": markers.get("retries", 0),
|
|
"test_runs": 0, "edits": 0, "human_interventions": 0},
|
|
}
|
|
|
|
|
|
def _config(tmp_path):
|
|
return {"store": {"db_path": str(tmp_path / ".store/m.db"),
|
|
"blob_dir": str(tmp_path / ".store/blobs"),
|
|
"cursor": str(tmp_path / ".store/c.json")}}
|
|
|
|
|
|
def test_run_detect_persists_cross_flavor_pattern(tmp_path):
|
|
cfg = _config(tmp_path)
|
|
st = Store(cfg["store"]["db_path"], cfg["store"]["blob_dir"])
|
|
# same problem (retry_storm) across two flavors -> cross-flavor candidate
|
|
st.write_digest("claude:a", _digest("claude:a", "claude", "r1", retries=5))
|
|
st.write_digest("codex:b", _digest("codex:b", "codex", "r2", retries=4))
|
|
st.close()
|
|
|
|
patterns = run_detect(cfg, min_frequency=2)
|
|
assert len(patterns) == 1
|
|
assert patterns[0]["cross_flavor"] is True
|
|
assert patterns[0]["signal_type"] == "retry_storm"
|
|
|
|
# persisted to the Tier 2 patterns table
|
|
st2 = Store(cfg["store"]["db_path"], cfg["store"]["blob_dir"])
|
|
rows = st2.db.execute("SELECT key FROM patterns").fetchall()
|
|
assert len(rows) == 1
|
|
st2.close()
|