session-memory: map signals to catalog recommendations via covers (WP-0010 follow-up)

Closes the gap where recurring_error suggestions showed generic 'Investigate'
instead of the curated recommendation. Added a covers[] field to SolutionPattern
(lowercase substrings a pattern's recommendation also applies to) + Catalog.find_for
(exact key first, then covers match against signal key+locus). Retro now resolves
recommendations through find_for. Tagged the read-before-edit pattern with
covers=['file has not been read','modified since read','file_not_read'] (v1.0.1).
Live: file-not-read suggestions across all repos now inherit 'Read the file before
Edit/Write'. 6 new tests; suite 158/158.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
2026-06-07 21:09:44 +02:00
parent 0d05dfcc5d
commit e237dcc622
9 changed files with 143 additions and 35 deletions

View File

@@ -0,0 +1,62 @@
"""find_for / covers tests (AGENTIC-WP-0010 follow-up)."""
import os
import sys
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from session_memory.curate.catalog import Catalog # noqa: E402
from session_memory.curate.schema import ( # noqa: E402
Provenance,
Resolution,
SolutionPattern,
)
def _pattern(pid, src, covers=None, name="P"):
return SolutionPattern(
id=pid, name=name, version="1.0.0", polarity="problem", problem="p",
resolutions=[Resolution(summary="do x")],
provenance=Provenance(source_key=src), covers=covers or [])
def test_covers_round_trips(tmp_path):
cat = Catalog(str(tmp_path))
cat.upsert(_pattern("sp-a", "problem:file_not_read:edit",
covers=["file has not been read"]))
assert cat.load("sp-a").covers == ["file has not been read"]
def test_find_for_exact_key(tmp_path):
cat = Catalog(str(tmp_path))
cat.upsert(_pattern(SolutionPattern.make_id("problem:retry_storm:retries"),
"problem:retry_storm:retries"))
got = cat.find_for("problem:retry_storm:retries")
assert got is not None and got.id == "sp-problem-retry_storm-retries"
def test_find_for_covers_match(tmp_path):
cat = Catalog(str(tmp_path))
cat.upsert(_pattern("sp-rbe", "problem:file_not_read:edit",
covers=["file has not been read", "modified since read"]))
# a recurring_error signal with a different key but matching fingerprint locus
got = cat.find_for(
"problem:recurring_error:<tool_use_error>file has not been read yet...",
locus="<tool_use_error>file has not been read yet. read it first...")
assert got is not None and got.id == "sp-rbe"
def test_find_for_no_match_returns_none(tmp_path):
cat = Catalog(str(tmp_path))
cat.upsert(_pattern("sp-rbe", "problem:file_not_read:edit",
covers=["file has not been read"]))
assert cat.find_for("problem:recurring_error:some unrelated error") is None
def test_covers_change_versions(tmp_path):
cat = Catalog(str(tmp_path))
cat.upsert(_pattern("sp-a", "problem:x:y"))
p = cat.load("sp-a")
p.covers = ["new coverage"]
assert cat.upsert(p) == "versioned" # covers is substantive content
assert cat.load("sp-a").version == "1.0.1"

View File

@@ -55,6 +55,26 @@ def test_recommendation_from_catalog(tmp_path):
assert r["suggestions"][0]["recommendation"] == "Stop and diagnose before retrying"
def test_recurring_error_inherits_recommendation_via_covers(tmp_path):
cat = Catalog(str(tmp_path / "catalog"))
cat.upsert(SolutionPattern(
id="sp-rbe", name="Read before edit", version="1.0.0", polarity="problem",
problem="edit before read",
resolutions=[Resolution(summary="Read the file first before Edit/Write")],
covers=["file has not been read"]))
digs = []
for i in range(2):
d = _digest(f"claude:{i}", "r1", "2026-06-0{}T10:00:00Z".format(i + 1))
d["error_snippets"] = [{
"fingerprint": "<tool_use_error>file has not been read yet. read it first...",
"sample": "File has not been read yet", "count": 2, "tool": "Edit"}]
digs.append(d)
r = weekly_retro(digs, catalog=cat, since="2026-05-30T00:00:00Z", until="2026-06-08T00:00:00Z")
rec_err = [s for s in r["suggestions"] if s["signal_type"] == "recurring_error"]
assert rec_err, "expected a recurring_error suggestion"
assert rec_err[0]["recommendation"] == "Read the file first before Edit/Write"
def test_caps_three_per_repo():
# five distinct problem signals in one repo -> capped at 3
digs = []