diff --git a/hooks/lib/learning_db_v2.py b/hooks/lib/learning_db_v2.py index 9a2a5604..3929cbb9 100755 --- a/hooks/lib/learning_db_v2.py +++ b/hooks/lib/learning_db_v2.py @@ -22,7 +22,6 @@ import re import sqlite3 import sys -from collections import namedtuple from contextlib import contextmanager from datetime import datetime, timedelta, timezone from pathlib import Path @@ -65,134 +64,6 @@ # Enforced by hooks/tests/test_injection_floor.py. INJECTION_MIN_CONFIDENCE = 0.5 -# Carried over from learning_db.py for backward compatibility -ERROR_TYPES = { - "missing_file": [ - r"no such file", - r"file not found", - r"cannot find", - r"does not exist", - ], - "permissions": [r"permission denied", r"access denied", r"not permitted"], - "syntax_error": [r"syntax ?error", r"unexpected token", r"parse error"], - "type_error": [r"type error", r"cannot convert", r"incompatible type"], - "import_error": [r"import error", r"module not found", r"no module named"], - "timeout": [r"timeout", r"timed out", r"deadline exceeded"], - "connection": [r"connection refused", r"network error", r"unreachable"], - "memory": [r"out of memory", r"memory error", r"heap"], - "multiple_matches": [r"multiple matches", r"found \d+ matches", r"replace_all"], -} - -DEFAULT_FIX_ACTIONS = { - "missing_file": {"fix_type": "auto", "fix_action": "create_file"}, - "permissions": {"fix_type": "manual", "fix_action": "check_permissions"}, - # systematic-debugging is a workflow pipeline, not an invocable skill. - "syntax_error": {"fix_type": "skill", "fix_action": "workflow"}, - "type_error": {"fix_type": "skill", "fix_action": "workflow"}, - "import_error": {"fix_type": "auto", "fix_action": "install_module"}, - "timeout": {"fix_type": "auto", "fix_action": "retry_with_timeout"}, - "connection": {"fix_type": "auto", "fix_action": "retry"}, - "memory": {"fix_type": "manual", "fix_action": "reduce_memory"}, - "multiple_matches": {"fix_type": "auto", "fix_action": "use_replace_all"}, -} - -# error-learner.py writes this as the solution half whenever it cannot map an -# error onto a real fix. The result restates the error type and the tool and -# carries no instruction, so an injector must not spend context on it. -DEFAULT_FIX_SOLUTION_TEMPLATE = "Fix {error_type} error in {tool_name}: {error}" - - -# ─── Contentless Hint Detection ──────────────────────────────── - -# error-learner stores a learning as " " and re-records -# nest it (" "), so the LAST arrow marks the -# solution. Unicode is the arrow in use (762 rows); 7 older rows use ASCII. -_SOLUTION_SEPARATORS = (" → ", " -> ") - -# Per-placeholder matchers for the stub template. error_type and tool_name are -# single tokens; the error snippet is free text and is absent from rows written -# before the snippet was appended to the template. -_STUB_FIELD_PATTERNS = {"error_type": r"\S+", "tool_name": r"\S+", "error": r".*"} -_STUB_OPTIONAL_FIELDS = frozenset({"error"}) - - -def _stub_literal(text: str, loose: bool) -> str: - """Escape a template literal, turning each whitespace run into a matcher.""" - whitespace = r"\s*" if loose else r"\s+" - out: list[str] = [] - in_run = False - for char in text: - if char.isspace(): - if not in_run: - out.append(whitespace) - in_run = True - else: - in_run = False - out.append(re.escape(char)) - return "".join(out) - - -def _build_stub_solution_pattern(template: str = DEFAULT_FIX_SOLUTION_TEMPLATE) -> "re.Pattern[str]": - """Compile the stub matcher from the template that writes stubs. - - Deriving the matcher from DEFAULT_FIX_SOLUTION_TEMPLATE keeps one source of - truth: renaming the template updates the matcher in the same edit instead of - stranding a hardcoded regex that silently stops matching. - """ - parts = re.split(r"\{(\w+)\}", template) - segments: list[str] = [] - for index, part in enumerate(parts): - if index % 2 == 0: - segments.append(_stub_literal(part, loose=False)) - continue - field = _STUB_FIELD_PATTERNS.get(part, r"\S+") - if part in _STUB_OPTIONAL_FIELDS: - # Fold the preceding literal into the optional group: the older rows - # stop at the tool name, with neither separator nor snippet. - literal = _stub_literal(parts[index - 1], loose=True) - segments[-1] = f"(?:{literal}{field})?" - else: - segments.append(field) - return re.compile("^" + "".join(segments) + "$") - - -_STUB_SOLUTION_RE = _build_stub_solution_pattern() - - -def solution_summary(value: object) -> str: - """Return the one-line solution half of a stored learning value. - - Returns "" for anything that is not a string: a malformed row carries no - solution, and callers must not have to pre-check the type. - - Searches the whole value, not just its first line: a captured error message - is usually multi-line, so the arrow sits on the last line. Reading only the - first line surfaced the raw capture instead -- nginx configs, ssh debug - output, diff hunks -- as the hint. Values with no arrow (prose gotchas) fall - back to their first line. - """ - if not isinstance(value, str): - return "" - for separator in _SOLUTION_SEPARATORS: - if separator in value: - value = value.rsplit(separator, 1)[1] - break - line = value.split("\n")[0] - # Drop control characters (BEL, ANSI escapes) captured from tool output. - return "".join(ch for ch in line if ch >= " " or ch == "\t").strip()[:120] - - -def hint_has_solution(value: object) -> bool: - """Report whether a stored learning carries an injectable solution. - - False for an empty or malformed value and for a generic stub, both of which - cost context and return no instruction. The row itself stays in the database: - its recurrence and frequency signal still feeds error classification and the - auto-feedback loop, it is only unfit to inject. - """ - summary = solution_summary(value) - return bool(summary) and not _STUB_SOLUTION_RE.match(summary) - # ─── Database Connection ─────────────────────────────────────── @@ -980,48 +851,6 @@ def sanitize_for_context(text: str) -> str: return text -def sanitize_fts_query(term: str) -> str: - """Strip FTS5 operators from a search term to prevent query injection. - - FTS5 operators (NOT, NEAR, AND, OR, *, quotes, parens, minus, colon) are removed - to ensure terms are treated as plain text matches. - """ - import re as _re - - # Remove FTS5 keyword operators FIRST (before special-char removal strips - # adjacent parens and breaks word boundaries, e.g. "NEAR(a b)" -> "NEARa b"). - term = _re.sub(r"\b(NOT|NEAR|AND|OR)\b", "", term, flags=_re.IGNORECASE) - # Remove FTS5 special characters - term = _re.sub(r'["\(\)\*:\-\^\+]', "", term) - return term.strip() - - -# ─── Error Classification (from learning_db.py) ─────────────── - - -def classify_error(message: str) -> str: - message_lower = message.lower() - for error_type, patterns in ERROR_TYPES.items(): - if any(re.search(p, message_lower) for p in patterns): - return error_type - return "unknown" - - -def normalize_error(message: str) -> str: - normalized = message.lower().strip() - normalized = re.sub(r"[/\\][\w./\\-]+[/\\]", "", normalized) - normalized = re.sub(r"line \d+", "line N", normalized) - normalized = re.sub(r"0x[0-9a-f]+", "0xADDR", normalized) - normalized = re.sub(r"\d{4}-\d{2}-\d{2}[Tt ]\d{2}:\d{2}:\d{2}", "TIMESTAMP", normalized) - return normalized - - -def generate_signature(error_message: str, error_type: str) -> str: - normalized = normalize_error(error_message) - content = f"{error_type}:{normalized}" - return hashlib.md5(content.encode()).hexdigest()[:16] - - # ─── Core API ────────────────────────────────────────────────── @@ -2317,127 +2146,6 @@ def query_learnings( return [dict(row) for row in rows] -def search_learnings( - query: str, - *, - min_confidence: float = 0.0, - exclude_graduated: bool = True, - categories: list[str] | None = None, - project_path: str | None = None, - exclude_test_sources: bool = True, - limit: int = 50, -) -> list[dict]: - """Full-text search across learnings with BM25 ranking. - - Unlike query_learnings() which matches exact tag substrings, - this uses FTS5 with porter stemming for fuzzy, ranked retrieval. - - Args: - query: FTS5 query string. Supports OR/AND operators and prefix - matching (e.g. "goroutine OR channel", "circuit*"). - min_confidence: Minimum confidence threshold for results. - exclude_graduated: If True, omit entries that have graduated. - categories: If given, restrict results to rows whose category is in - this list (matches ANY, same semantics as query_learnings()'s - `tags` filter). Use this to keep cross-domain categories (e.g. - "voice", "review", "design") out of results meant for a - different domain (e.g. tool-error hints). - project_path: If given, restrict results to rows with a NULL - project_path (global knowledge) or an exact match — same - semantics as query_learnings()'s `project_path` filter. - exclude_test_sources: If True (default), omit entries where source - starts with 'test' — same default and rationale as - query_learnings() (ADR-191): keeps test fixtures out of - production injection. Pass False to include them (e.g. auditing). - limit: Maximum number of results to return. - - Returns: - List of learning dicts ordered by BM25 relevance (best first), - each with an additional 'rank' key containing the BM25 score. - """ - init_db() - - if not query or not query.strip(): - return [] - - # Sanitize FTS query terms before matching - query_str = query - if query_str: - terms = query_str.split(" OR ") - terms = [sanitize_fts_query(t.strip()) for t in terms if t.strip()] - terms = [t for t in terms if t] # Remove empty after sanitization - if terms: - query_str = " OR ".join(terms) - else: - return [] # All terms were FTS operators — no valid query - - conditions = ["l.confidence >= ?"] - params: list = [min_confidence] - - if exclude_graduated: - conditions.append("l.graduated_to IS NULL") - - if exclude_test_sources: - conditions.append("l.source NOT LIKE 'test%'") - - if categories: - category_clauses = [] - for category in categories: - category_clauses.append("l.category = ?") - params.append(category) - conditions.append(f"({' OR '.join(category_clauses)})") - - if project_path: - conditions.append("(l.project_path IS NULL OR l.project_path = ?)") - params.append(project_path) - - where = " AND ".join(conditions) - - with get_connection() as conn: - try: - rows = conn.execute( - f""" - SELECT l.*, bm25(learnings_fts) AS rank - FROM learnings_fts fts - JOIN learnings l ON l.id = fts.rowid - WHERE learnings_fts MATCH ? - AND {where} - ORDER BY rank - LIMIT ? - """, - (query_str, *params, limit), - ).fetchall() - return [dict(row) for row in rows] - except sqlite3.OperationalError: - # Invalid FTS5 query syntax — fall back to empty results - return [] - - -def lookup_error_solution( - error_message: str, - min_confidence: float = 0.7, -) -> dict | None: - """Look up a solution for an error pattern. Backward-compatible with error-learner.""" - init_db() - - error_type = classify_error(error_message) - signature = generate_signature(error_message, error_type) - - with get_connection() as conn: - row = conn.execute( - """ - SELECT topic, key, value, confidence, fix_type, fix_action, error_signature - FROM learnings - WHERE error_signature = ? AND confidence >= ? AND category = 'error' - """, - (signature, min_confidence), - ).fetchone() - - if row: - return dict(row) - return None - - def record_instruction_compliance( instruction_id: str, compliant: bool, @@ -2578,158 +2286,6 @@ def decay_confidence(topic: str, key: str, delta: float = 0.10) -> float: return new_conf -# ─── Graduation Targets ─────────────────────────────────────── -# -# A graduated learning is excluded from injection forever (exclude_graduated is -# the default in query_learnings/search_learnings). So `graduated_to` must name -# a durable artifact in the repo. Ephemeral values -- "session-artifact" and the -# "pruned:" family -- suppressed 98 rows permanently while naming nothing a -# reader could open. - -_GRADUATION_SENTINELS = frozenset({"session-artifact", "environment-artifact"}) -_GRADUATION_SENTINEL_PREFIXES = ("pruned:",) - -_AGENT_PREFIX = "agent:" -_SKILL_PREFIX = "skill:" -_TARGET_PREFIX = "target:" - -# agent:/skill: names index into a fixed layout, so keep them to plain names. -_SAFE_TARGET_NAME = re.compile(r"^[A-Za-z0-9._-]+$") - - -# Result of resolving a `graduated_to` value against the repo tree. -# raw: the stored value, stripped -# path: repo-relative path it normalizes to, or None when not path-shaped -# durable: True when `path` exists inside the repo -# reason: resolved | missing | outside-repo | sentinel | empty -# -# collections.namedtuple, not typing.NamedTuple: `typing` costs ~4ms to import -# and this module loads on every Bash and Edit tool call. -GraduationTarget = namedtuple("GraduationTarget", "raw path durable reason") - - -def default_repo_root() -> Path: - """Return the repo root: CLAUDE_PROJECT_DIR, else nearest .git ancestor, else cwd.""" - env_dir = os.environ.get("CLAUDE_PROJECT_DIR") - if env_dir: - return Path(env_dir) - cwd = Path.cwd() - for candidate in (cwd, *cwd.parents): - if (candidate / ".git").exists(): - return candidate - return cwd - - -def _resolve_repo_path(raw: str, value: str, root: Path) -> GraduationTarget: - """Resolve a path-shaped target under `root`. Anything escaping it is out of repo.""" - try: - root_real = root.expanduser().resolve() - expanded = Path(value).expanduser() - candidate = expanded if expanded.is_absolute() else root_real / expanded - rel = candidate.resolve().relative_to(root_real) - except (ValueError, OSError, RuntimeError): - return GraduationTarget(raw, None, False, "outside-repo") - - rel_str = rel.as_posix() - if not rel_str or rel_str == ".": - return GraduationTarget(raw, None, False, "outside-repo") - if (root_real / rel_str).exists(): - return GraduationTarget(raw, rel_str, True, "resolved") - return GraduationTarget(raw, rel_str, False, "missing") - - -def resolve_graduation_target(target: object, *, repo_root: Path | str | None = None) -> GraduationTarget: - """Resolve a `graduated_to` value to a durable repo artifact. - - Normalizes every notation the database carries: - `agent:X` -> `agents/X.md`, `skill:X` -> that skill's SKILL.md in any - group, `target:PATH` -> `PATH`, and a bare path as-is. Values naming no - file -- sentinels, deleted paths, machine-local paths outside the repo -- - come back with durable=False. - - Args: - target: The stored `graduated_to` value. - repo_root: Root to resolve against. Defaults to default_repo_root(). - """ - raw = target.strip() if isinstance(target, str) else "" - if not raw: - return GraduationTarget(raw, None, False, "empty") - - root = Path(repo_root) if repo_root is not None else default_repo_root() - - value = raw - if value.startswith(_TARGET_PREFIX): - value = value[len(_TARGET_PREFIX) :].strip() - if not value: - return GraduationTarget(raw, None, False, "empty") - - lowered = value.lower() - if lowered in _GRADUATION_SENTINELS or lowered.startswith(_GRADUATION_SENTINEL_PREFIXES): - return GraduationTarget(raw, None, False, "sentinel") - - if value.startswith(_AGENT_PREFIX): - name = value[len(_AGENT_PREFIX) :].strip() - if not _SAFE_TARGET_NAME.match(name): - return GraduationTarget(raw, None, False, "missing") - return _resolve_repo_path(raw, f"agents/{name}.md", root) - - if value.startswith(_SKILL_PREFIX): - name = value[len(_SKILL_PREFIX) :].strip() - if not _SAFE_TARGET_NAME.match(name): - return GraduationTarget(raw, None, False, "missing") - for match in sorted(root.glob(f"skills/*/{name}/SKILL.md")): - return _resolve_repo_path(raw, match.relative_to(root).as_posix(), root) - return GraduationTarget(raw, f"skills/{name}", False, "missing") - - return _resolve_repo_path(raw, value, root) - - -def mark_graduated(topic: str, key: str, target: str, *, repo_root: Path | str | None = None) -> bool: - """Mark entry as graduated to a permanent location. - - Refuses ephemeral targets (`session-artifact`, `pruned:*`, empty): a - graduated row never injects again, and a session artifact is not a durable - home for a learning. A path-shaped target that does not currently resolve - is written with a warning, because the file may exist in another checkout. - - Args: - topic: Learning topic. - key: Learning key. - target: Durable artifact the knowledge moved into. - repo_root: Root used to resolve `target`. Defaults to default_repo_root(). - - Returns: - True if an entry was updated, False if the target was refused or no - matching entry was found. - """ - resolved = resolve_graduation_target(target, repo_root=repo_root) - if resolved.reason in ("sentinel", "empty"): - print( - f"WARNING: mark_graduated refused non-durable target {target!r} for " - f"{topic}/{key} — graduation needs a durable file in the repo", - file=sys.stderr, - ) - return False - if not resolved.durable: - print( - f"WARNING: mark_graduated target {target!r} does not resolve to a repo " - f"file ({resolved.reason}) — recording anyway for {topic}/{key}", - file=sys.stderr, - ) - - init_db() - with get_connection() as conn: - cursor = conn.execute( - "UPDATE learnings SET graduated_to = ? WHERE topic = ? AND key = ?", - (target, topic, key), - ) - conn.commit() - if cursor.rowcount == 0: - print(f"WARNING: mark_graduated found no entry for topic={topic!r} key={key!r}", file=sys.stderr) - return False - return True - - VALID_EVENT_TYPES = { "secret_detected", "approval_requested", @@ -3061,93 +2617,6 @@ def prune(min_confidence: float = 0.3, older_than_days: int = 90) -> int: # ─── Import / Export ─────────────────────────────────────────── -def import_from_retro(retro_dir: str) -> dict: - """Import existing retro L2 markdown files into learning.db.""" - retro_path = Path(retro_dir) - l2_dir = retro_path / "L2" - if not l2_dir.is_dir(): - return {"imported": 0, "skipped": 0, "errors": ["L2 directory not found"]} - - imported = 0 - skipped = 0 - errors = [] - - for md_file in sorted(l2_dir.glob("*.md")): - try: - content = md_file.read_text() - topic = md_file.stem - - # Extract metadata from header - conf_match = re.search(r"\*\*Confidence\*\*:\s*(\w+)", content) - confidence_str = conf_match.group(1) if conf_match else "MEDIUM" - conf_map = {"HIGH": 0.85, "MEDIUM": 0.65, "LOW": 0.45} - base_confidence = conf_map.get(confidence_str.upper(), 0.65) - - tags_match = re.search(r"\*\*Tags\*\*:\s*(.+)", content) - tags = [t.strip() for t in tags_match.group(1).split(",")] if tags_match else [] - - source_match = re.search(r"\*\*Source\*\*:\s*(.+)", content) - source_str = f"migrated:{source_match.group(1).strip()}" if source_match else "migrated:retro" - - # Parse entries by ### heading - parts = re.split(r"(?=^### )", content, flags=re.MULTILINE) - for part in parts[1:] if len(parts) > 1 else []: - heading_match = re.match(r"### (.+?)(?:\n|$)", part) - if not heading_match: - continue - - raw_key = heading_match.group(1).strip() - - # Check for graduation marker - graduated_to = None - grad_match = re.search(r"\[GRADUATED\s*→\s*(.+?)\]", raw_key) - if grad_match: - graduated_to = grad_match.group(1).strip() - raw_key = re.sub(r"\s*\[GRADUATED\s*→\s*.+?\]", "", raw_key).strip() - - # Check for observation count - obs_match = re.search(r"\[(\d+)x\]", raw_key) - obs_count = int(obs_match.group(1)) if obs_match else 1 - raw_key = re.sub(r"\s*\[\d+x\]", "", raw_key).strip() - - key = raw_key.lower().replace(" ", "-") - value = part[len(heading_match.group(0)) :].strip() - - if not value: - skipped += 1 - continue - - result = record_learning( - topic=topic, - key=key, - value=value, - category="design", # Best guess for retro entries - confidence=base_confidence, - tags=tags if tags else None, - source=source_str, - ) - - # Apply graduated_to if present - if graduated_to: - mark_graduated(topic, key, graduated_to) - - # Set observation count directly if > 1 - if obs_count > 1: - with get_connection() as conn: - conn.execute( - "UPDATE learnings SET observation_count = ? WHERE topic = ? AND key = ?", - (obs_count, topic, key), - ) - conn.commit() - - imported += 1 - - except Exception as e: - errors.append(f"{md_file.name}: {e}") - - return {"imported": imported, "skipped": skipped, "errors": errors} - - def import_from_patterns_db(db_path: str) -> dict: """Import existing patterns.db into learning.db.""" patterns_path = Path(db_path) diff --git a/hooks/tests/test_fts5_search.py b/hooks/tests/test_fts5_search.py index 7a9ae08a..c050ac80 100755 --- a/hooks/tests/test_fts5_search.py +++ b/hooks/tests/test_fts5_search.py @@ -1,16 +1,11 @@ #!/usr/bin/env python3 """ -Tests for FTS5 full-text search in learning_db_v2. +Schema contract tests for the learnings_fts FTS5 index in learning_db_v2. Verifies: -- FTS5 virtual table creation and trigger-based sync -- search_learnings() with BM25 ranking -- Porter stemming (morphological matching) -- OR/AND query syntax -- Prefix queries -- Migration backfill for pre-existing rows -- Backward compatibility with query_learnings() -- Edge cases: empty query, invalid syntax, no matches +- trigger-based sync of learnings_fts on UPDATE and DELETE +- migration backfill of rows written before the FTS table existed +- query_learnings() lookup by topic """ import sqlite3 @@ -33,25 +28,30 @@ def isolated_db(tmp_path, monkeypatch): return tmp_path -def _record(topic: str, key: str, value: str, tags: list[str] | None = None, confidence: float = 0.7) -> dict: - """Helper to record a learning with defaults. - - source="manual" (not "test*") so these fixture rows survive - search_learnings()'s default exclude_test_sources=True -- the dedicated - TestExcludeTestSources class below uses an explicit "test*" source to - exercise that filter directly. - """ +def _record(topic: str, key: str, value: str, tags: list[str] | None = None) -> dict: + """Record a learning with defaults.""" return db.record_learning( topic=topic, key=key, value=value, category="design", - confidence=confidence, + confidence=0.7, tags=tags, source="manual", ) +def _fts_match(term: str) -> list[dict]: + """Query learnings_fts directly and join back to the learnings row.""" + with db.get_connection() as conn: + rows = conn.execute( + "SELECT l.topic, l.value FROM learnings_fts JOIN learnings l ON l.id = learnings_fts.rowid " + "WHERE learnings_fts MATCH ?", + (term,), + ).fetchall() + return [dict(row) for row in rows] + + class TestTriggerSync: """Verify FTS index stays in sync with learnings table.""" @@ -65,7 +65,7 @@ def test_update_syncs_to_fts(self): tags=["go"], ) - results = db.search_learnings("goroutines") + results = _fts_match("goroutines") assert len(results) == 1 assert "goroutines" in results[0]["value"] @@ -73,7 +73,7 @@ def test_delete_syncs_to_fts(self): _record("temp-topic", "temp-key", "Temporary value for deletion test", tags=["temp"]) # Verify it's searchable - assert len(db.search_learnings("temporary")) >= 1 + assert len(_fts_match("temporary")) >= 1 # Delete directly with db.get_connection() as conn: @@ -81,222 +81,7 @@ def test_delete_syncs_to_fts(self): conn.commit() # FTS should no longer find it - results = db.search_learnings("temporary") - assert len(results) == 0 - - -class TestSearchLearnings: - """Test the search_learnings() function.""" - - def test_basic_search(self): - _record("go-patterns", "mutex-usage", "Use sync.Mutex for shared state", tags=["go", "concurrency"]) - _record("python-patterns", "dataclass-usage", "Use dataclasses for structured data", tags=["python"]) - - results = db.search_learnings("mutex") - assert len(results) == 1 - assert results[0]["topic"] == "go-patterns" - - def test_bm25_ranking(self): - """More relevant results should rank higher (more negative BM25).""" - _record("topic-a", "key-a", "goroutine goroutine goroutine channel", tags=["go"]) - _record("topic-b", "key-b", "some other content with goroutine once", tags=["misc"]) - - results = db.search_learnings("goroutine") - assert len(results) == 2 - # First result should have more negative rank (= more relevant) - assert results[0]["rank"] <= results[1]["rank"] - assert results[0]["topic"] == "topic-a" - - def test_porter_stemming(self): - """Porter stemmer should match morphological variants.""" - _record("config", "config-patterns", "Configuring the application requires validation", tags=["config"]) - - # All these forms should match via stemming - for query in ["configuring", "configured", "configuration", "configure"]: - results = db.search_learnings(query) - assert len(results) >= 1, f"Stemming failed for '{query}'" - - def test_or_query(self): - _record("topic-a", "key-a", "Worker pool implementation", tags=["go"]) - _record("topic-b", "key-b", "Circuit breaker pattern", tags=["resilience"]) - _record("topic-c", "key-c", "Unrelated topic about databases", tags=["sql"]) - - results = db.search_learnings("worker OR circuit") - assert len(results) == 2 - topics = {r["topic"] for r in results} - assert topics == {"topic-a", "topic-b"} - - def test_and_query(self): - _record("topic-a", "key-a", "State machine in Go", tags=["go", "state-machine"]) - _record("topic-b", "key-b", "State management in React", tags=["react"]) - _record("topic-c", "key-c", "Machine learning basics", tags=["ml"]) - - results = db.search_learnings("state AND machine") - assert len(results) == 1 - assert results[0]["topic"] == "topic-a" - - def test_prefix_query(self): - _record("topic-a", "key-a", "Circuit breaker for resilience", tags=["circuit-breaker"]) - - results = db.search_learnings("circuit*") - assert len(results) >= 1 - assert results[0]["topic"] == "topic-a" - - def test_searches_across_all_columns(self): - """FTS5 indexes topic, key, value, and tags — all should be searchable.""" - _record("goroutine-patterns", "pool-design", "Describes worker pool", tags=["concurrency"]) - - # Match in topic column - assert len(db.search_learnings("goroutine")) >= 1 - # Match in key column - assert len(db.search_learnings("pool")) >= 1 - # Match in value column - assert len(db.search_learnings("worker")) >= 1 - # Match in tags column - assert len(db.search_learnings("concurrency")) >= 1 - - def test_min_confidence_filter(self): - _record("topic-a", "key-a", "Low confidence result", tags=["test"], confidence=0.3) - _record("topic-b", "key-b", "High confidence result", tags=["test"], confidence=0.9) - - results = db.search_learnings("confidence result", min_confidence=0.5) - assert len(results) == 1 - assert results[0]["topic"] == "topic-b" - - def test_exclude_graduated(self): - _record("topic-a", "key-a", "Graduated entry about testing", tags=["test"]) - db.mark_graduated("topic-a", "key-a", "agent:test-agent") - _record("topic-b", "key-b", "Active entry about testing", tags=["test"]) - - # Default: exclude graduated - results = db.search_learnings("testing") - assert len(results) == 1 - assert results[0]["topic"] == "topic-b" - - # Include graduated - results = db.search_learnings("testing", exclude_graduated=False) - assert len(results) == 2 - - def test_limit(self): - for i in range(10): - _record(f"topic-{i}", f"key-{i}", f"Entry number {i} about testing", tags=["test"]) - - results = db.search_learnings("testing", limit=3) - assert len(results) == 3 - - def test_empty_query_returns_empty(self): - _record("topic-a", "key-a", "Some content", tags=["test"]) - assert db.search_learnings("") == [] - assert db.search_learnings(" ") == [] - - def test_invalid_fts_syntax_returns_empty(self): - """Invalid FTS5 query syntax should not raise, just return empty.""" - _record("topic-a", "key-a", "Some content", tags=["test"]) - # Unbalanced quotes and other invalid FTS5 syntax - assert db.search_learnings('"unclosed quote') == [] - - def test_categories_filter_matches_any(self): - """ADR: pretool-injector-scoping -- categories restricts to an allowlist.""" - _record("topic-a", "key-a", "Error content about deadlocks", tags=["go"]) # category=design (default) - db.record_learning( - topic="topic-b", - key="key-b", - value="Deadlock error pattern", - category="error", - confidence=0.9, - tags=["go"], - source="manual", - ) - db.record_learning( - topic="topic-c", - key="key-c", - value="Deadlock gotcha note", - category="gotcha", - confidence=0.9, - tags=["go"], - source="manual", - ) - - results = db.search_learnings("deadlock", categories=["error", "gotcha"]) - topics = {r["topic"] for r in results} - assert topics == {"topic-b", "topic-c"} - - def test_categories_filter_excludes_others(self): - _record("topic-a", "key-a", "Voice content about deadlocks", tags=["go"]) # category=design - results = db.search_learnings("deadlock", categories=["error", "gotcha", "debug"]) - assert results == [] - - def test_project_path_filter_matches_global_and_exact(self): - db.record_learning( - topic="topic-global", - key="key-a", - value="Global content about circuit breakers", - category="error", - confidence=0.9, - source="manual", - project_path=None, - ) - db.record_learning( - topic="topic-same-project", - key="key-b", - value="Same-project content about circuit breakers", - category="error", - confidence=0.9, - source="manual", - project_path="/home/user/project-a", - ) - db.record_learning( - topic="topic-other-project", - key="key-c", - value="Other-project content about circuit breakers", - category="error", - confidence=0.9, - source="manual", - project_path="/home/user/project-b", - ) - - results = db.search_learnings("circuit breakers", project_path="/home/user/project-a") - topics = {r["topic"] for r in results} - assert topics == {"topic-global", "topic-same-project"} - assert "topic-other-project" not in topics - - def test_project_path_none_is_no_op(self): - db.record_learning( - topic="topic-other-project", - key="key-c", - value="Other-project content about circuit breakers", - category="error", - confidence=0.9, - source="manual", - project_path="/home/user/project-b", - ) - results = db.search_learnings("circuit breakers") - assert len(results) == 1 - - def test_exclude_test_sources_default_excludes(self): - """ADR: pretool-injector-scoping Concern 1 -- parity with query_learnings().""" - db.record_learning( - topic="topic-fixture", - key="key-a", - value="Test fixture content about retries", - category="error", - confidence=0.9, - source="test-fixture", - ) - results = db.search_learnings("retries") - assert results == [] - - def test_exclude_test_sources_false_includes(self): - db.record_learning( - topic="topic-fixture", - key="key-a", - value="Test fixture content about retries", - category="error", - confidence=0.9, - source="test-fixture", - ) - results = db.search_learnings("retries", exclude_test_sources=False) - assert len(results) == 1 + assert _fts_match("temporary") == [] class TestMigrationBackfill: @@ -369,7 +154,7 @@ def test_backfill_existing_rows(self, isolated_db): db.init_db() # The pre-existing row should be searchable via FTS - results = db.search_learnings("goroutines") + results = _fts_match("goroutines") assert len(results) == 1 assert results[0]["topic"] == "pre-existing" @@ -382,15 +167,3 @@ def test_query_learnings_by_topic(self): results = db.query_learnings(topic="go-patterns", exclude_test_sources=False) assert len(results) == 1 - - def test_both_apis_find_same_entry(self): - _record("shared", "shared-key", "Shared content about concurrency", tags=["go", "concurrency"]) - - query_results = db.query_learnings(tags=["concurrency"], exclude_test_sources=False) - search_results = db.search_learnings("concurrency") - - assert len(query_results) >= 1 - assert len(search_results) >= 1 - # Both should find the same entry - assert query_results[0]["topic"] == search_results[0]["topic"] - assert query_results[0]["key"] == search_results[0]["key"] diff --git a/hooks/tests/test_graduation_targets.py b/hooks/tests/test_graduation_targets.py deleted file mode 100644 index 7edf2e40..00000000 --- a/hooks/tests/test_graduation_targets.py +++ /dev/null @@ -1,189 +0,0 @@ -#!/usr/bin/env python3 -"""Tests for graduation-target resolution and the mark_graduated write guard. - -A graduated learning is suppressed from injection forever, so `graduated_to` -must name a durable artifact in the repo. Ephemeral sentinels such as -`session-artifact` suppressed 97 rows permanently. - -Covers: -- every notation the database carries: bare path, `agent:X`, `skill:X`, - `target:PATH`, sentinels, out-of-repo paths, traversal, empty; -- mark_graduated refuses a sentinel target and records nothing; -- mark_graduated warns but still writes a path-shaped target that is missing. - -Run with: python3 -m pytest hooks/tests/test_graduation_targets.py -v -""" - -import sys -from pathlib import Path - -import pytest - -_LIB_DIR = Path(__file__).resolve().parent.parent / "lib" -if str(_LIB_DIR) not in sys.path: - sys.path.insert(0, str(_LIB_DIR)) - -import learning_db_v2 as db - - -@pytest.fixture(autouse=True) -def isolated_db(tmp_path, monkeypatch): - """Use a fresh temp learning.db for each test -- never the real one.""" - monkeypatch.setenv("CLAUDE_LEARNING_DIR", str(tmp_path / "learning")) - db._initialized = False - yield - db._initialized = False - - -@pytest.fixture -def repo(tmp_path): - """A tmp tree shaped like the toolkit repo.""" - root = tmp_path / "repo" - for rel in ( - "agents/golang-general-engineer.md", - "skills/meta/install/SKILL.md", - "skills/process/worktree-agent/SKILL.md", - "skills/meta/retro/SKILL.md", - "docs/what-didnt-work.md", - "hooks/session-context.py", - ): - path = root / rel - path.parent.mkdir(parents=True, exist_ok=True) - path.write_text("stub\n") - (root / "scripts").mkdir(parents=True, exist_ok=True) - return root - - -class TestResolveGraduationTarget: - def test_repo_relative_path_that_exists_is_durable(self, repo): - result = db.resolve_graduation_target("docs/what-didnt-work.md", repo_root=repo) - assert result.durable is True - assert result.reason == "resolved" - assert result.path == "docs/what-didnt-work.md" - - def test_repo_relative_directory_that_exists_is_durable(self, repo): - result = db.resolve_graduation_target("scripts/", repo_root=repo) - assert result.durable is True - - def test_repo_relative_path_that_is_missing_is_not_durable(self, repo): - result = db.resolve_graduation_target("agents/general-purpose.md", repo_root=repo) - assert result.durable is False - assert result.reason == "missing" - assert result.path == "agents/general-purpose.md" - - def test_agent_prefix_normalizes_to_agents_markdown(self, repo): - result = db.resolve_graduation_target("agent:golang-general-engineer", repo_root=repo) - assert result.durable is True - assert result.path == "agents/golang-general-engineer.md" - - def test_agent_prefix_for_unknown_agent_is_not_durable(self, repo): - result = db.resolve_graduation_target("agent:general-purpose", repo_root=repo) - assert result.durable is False - assert result.reason == "missing" - - def test_skill_prefix_normalizes_to_skill_directory(self, repo): - result = db.resolve_graduation_target("skill:install", repo_root=repo) - assert result.durable is True - assert result.path == "skills/meta/install/SKILL.md" - - def test_skill_prefix_finds_skill_in_any_group(self, repo): - result = db.resolve_graduation_target("skill:worktree-agent", repo_root=repo) - assert result.durable is True - assert result.path == "skills/process/worktree-agent/SKILL.md" - - def test_skill_prefix_for_unknown_skill_is_not_durable(self, repo): - result = db.resolve_graduation_target("skill:does-not-exist", repo_root=repo) - assert result.durable is False - - def test_target_prefix_strips_to_the_path(self, repo): - result = db.resolve_graduation_target("target:skills/meta/retro/SKILL.md", repo_root=repo) - assert result.durable is True - assert result.path == "skills/meta/retro/SKILL.md" - - def test_target_prefix_composes_with_agent_prefix(self, repo): - result = db.resolve_graduation_target("target:agents/golang-general-engineer.md", repo_root=repo) - assert result.durable is True - - @pytest.mark.parametrize("sentinel", ["session-artifact", "pruned:environment-artifact", "pruned:anything"]) - def test_sentinels_are_not_durable(self, sentinel, repo): - result = db.resolve_graduation_target(sentinel, repo_root=repo) - assert result.durable is False - assert result.reason == "sentinel" - assert result.path is None - - @pytest.mark.parametrize("value", ["", " ", None]) - def test_empty_target_is_not_durable(self, value, repo): - result = db.resolve_graduation_target(value, repo_root=repo) - assert result.durable is False - assert result.reason == "empty" - - def test_absolute_path_outside_the_repo_is_not_durable(self, repo, tmp_path): - outside = tmp_path / "elsewhere" / "SKILL.md" - outside.parent.mkdir(parents=True, exist_ok=True) - outside.write_text("stub\n") - result = db.resolve_graduation_target(str(outside), repo_root=repo) - assert result.durable is False - assert result.reason == "outside-repo" - - def test_home_relative_path_is_not_durable(self, repo): - result = db.resolve_graduation_target("~/.claude/hooks/scanner.py", repo_root=repo) - assert result.durable is False - assert result.reason == "outside-repo" - - def test_traversal_escaping_the_repo_is_not_durable(self, repo): - result = db.resolve_graduation_target("../../etc/passwd", repo_root=repo) - assert result.durable is False - assert result.reason == "outside-repo" - - def test_absolute_path_inside_the_repo_resolves_relative(self, repo): - result = db.resolve_graduation_target(str(repo / "docs/what-didnt-work.md"), repo_root=repo) - assert result.durable is True - assert result.path == "docs/what-didnt-work.md" - - def test_skill_name_with_a_slash_is_rejected(self, repo): - result = db.resolve_graduation_target("skill:../../etc", repo_root=repo) - assert result.durable is False - - -class TestMarkGraduatedGuard: - def _seed(self): - db.record_learning( - topic="import_error", - key="sig-1", - value="module missing -> pip install it", - category="error", - source="hook:error-learner", - ) - - def _graduated_to(self): - with db.get_connection() as conn: - row = conn.execute( - "SELECT graduated_to FROM learnings WHERE topic = 'import_error' AND key = 'sig-1'" - ).fetchone() - return row["graduated_to"] - - def test_sentinel_target_is_refused_and_nothing_is_written(self, repo, capsys): - self._seed() - assert db.mark_graduated("import_error", "sig-1", "session-artifact", repo_root=repo) is False - assert self._graduated_to() is None - assert "session-artifact" in capsys.readouterr().err - - def test_pruned_sentinel_is_refused(self, repo): - self._seed() - assert db.mark_graduated("import_error", "sig-1", "pruned:environment-artifact", repo_root=repo) is False - assert self._graduated_to() is None - - def test_durable_target_is_written_without_warning(self, repo, capsys): - self._seed() - assert db.mark_graduated("import_error", "sig-1", "agent:golang-general-engineer", repo_root=repo) is True - assert self._graduated_to() == "agent:golang-general-engineer" - assert capsys.readouterr().err == "" - - def test_missing_path_target_warns_but_still_writes(self, repo, capsys): - self._seed() - assert db.mark_graduated("import_error", "sig-1", "agents/not-yet-created.md", repo_root=repo) is True - assert self._graduated_to() == "agents/not-yet-created.md" - assert "does not resolve" in capsys.readouterr().err - - def test_missing_row_still_returns_false(self, repo): - assert db.mark_graduated("nope", "nope", "agent:golang-general-engineer", repo_root=repo) is False diff --git a/hooks/tests/test_stub_hint_filter.py b/hooks/tests/test_stub_hint_filter.py deleted file mode 100644 index ae083943..00000000 --- a/hooks/tests/test_stub_hint_filter.py +++ /dev/null @@ -1,113 +0,0 @@ -#!/usr/bin/env python3 -"""Tests for the contentless-hint filter in learning_db_v2. - -The retired error-learner wrote a generic stub solution whenever it could not -classify an error ("Fix error in : "). Those rows are -still in the database: the solution half carries no instruction, so any reader -that summarizes rows must drop them. hint_has_solution() is that predicate. - -The learning loop that consumed it is retired; the predicate stays because the -stub rows it recognizes are still stored. - -Covers: -- hint_has_solution on a real solution, on stubs built from representative - DEFAULT_FIX_ACTIONS entries, on multi-line and Unicode-arrow values, and on - empty or malformed input. -- The matcher is derived from DEFAULT_FIX_SOLUTION_TEMPLATE, so renaming the - template cannot leave a stale regex behind. - -Uses a throwaway learning.db via CLAUDE_LEARNING_DIR — never the real DB. - -Run with: python3 -m pytest hooks/tests/test_stub_hint_filter.py -v -""" - -import sys -from pathlib import Path - -import pytest - -REPO_ROOT = Path(__file__).resolve().parents[2] -LIB_DIR = REPO_ROOT / "hooks" / "lib" -if str(LIB_DIR) not in sys.path: - sys.path.insert(0, str(LIB_DIR)) - -import learning_db_v2 as db - -ARROW = "→" - - -@pytest.fixture(autouse=True) -def isolated_db(tmp_path, monkeypatch): - """Point every test at a throwaway learning.db.""" - monkeypatch.setenv("CLAUDE_LEARNING_DIR", str(tmp_path)) - db._initialized = False - yield tmp_path - db._initialized = False - - -def _record(topic: str, key: str, value: str, category: str = "error") -> None: - db.record_learning( - topic=topic, - key=key, - value=value, - category=category, - confidence=0.9, - source="manual", - project_path=None, - ) - - -def _stub(error_type: str, tool: str = "Bash", snippet: str = "exit status 1") -> str: - return db.DEFAULT_FIX_SOLUTION_TEMPLATE.format(error_type=error_type, tool_name=tool, error=snippet) - - -# ── hint_has_solution ───────────────────────────────────────────── - - -class TestHintHasSolution: - def test_real_solution_is_kept(self): - value = f"config.yaml: No such file or directory {ARROW} copy config.yaml.example first" - assert db.hint_has_solution(value) is True - - # The {error_type} slot is one wildcard: a bare word, an underscored word, and the - # fallback type cover it. - @pytest.mark.parametrize("error_type", ["timeout", "multiple_matches", "unknown"]) - def test_stub_for_fix_action_is_dropped(self, error_type): - value = f"boom {ARROW} {_stub(error_type)}" - assert db.hint_has_solution(value) is False - - def test_bare_stub_without_a_snippet_is_dropped(self): - """Rows written before the snippet was appended end at the tool name.""" - assert db.hint_has_solution(f"boom {ARROW} Fix timeout error in Bash") is False - - def test_stub_with_an_empty_snippet_is_dropped(self): - assert db.hint_has_solution(f"boom {ARROW} {_stub('unknown', snippet='')}") is False - - def test_multiline_error_half_does_not_hide_the_stub(self): - value = f"server {{\n listen 80;\n}}\nnginx: test failed {ARROW} {_stub('unknown', snippet='nginx: t')}" - assert db.hint_has_solution(value) is False - - def test_multiline_error_half_keeps_a_real_solution(self): - value = f"server {{\n listen 80;\n}}\nnginx: test failed {ARROW} run nginx -t and fix the block" - assert db.hint_has_solution(value) is True - - def test_ascii_arrow_value_is_read_the_same_way(self): - assert db.hint_has_solution("Found 3 matches -> pass replace_all=True") is True - assert db.hint_has_solution("Found 3 matches -> Fix multiple_matches error in Edit") is False - - def test_nested_arrows_read_the_last_solution(self): - assert db.hint_has_solution(f"exit 1 {ARROW} timeout {ARROW} Fix timeout error in Bash") is False - assert db.hint_has_solution(f"exit 1 {ARROW} timeout {ARROW} retry with --timeout 300") is True - - @pytest.mark.parametrize("value", ["", f"boom {ARROW} ", f"boom {ARROW} \n\n", None]) - def test_empty_or_malformed_values_carry_no_solution(self, value): - assert db.hint_has_solution(value) is False - - def test_prose_gotcha_without_an_arrow_is_kept(self): - assert db.hint_has_solution("Prefer rg over grep; grep misses .gitignored paths") is True - - def test_matcher_tracks_the_template(self): - """The matcher is built from the template, so a rename cannot strand it.""" - rebuilt = db._build_stub_solution_pattern("Repair {error_type} fault in {tool_name}: {error}") - assert rebuilt.match("Repair timeout fault in Bash: exit 1") - assert not rebuilt.match("Fix timeout error in Bash: exit 1") diff --git a/tests/test_learning_db_and_routing_outcomes.py b/tests/test_learning_db_and_routing_outcomes.py index 3318d750..180f5f8a 100644 --- a/tests/test_learning_db_and_routing_outcomes.py +++ b/tests/test_learning_db_and_routing_outcomes.py @@ -89,29 +89,6 @@ def test_empty_and_none(self): assert sanitize_for_context(None) is None -class TestSanitizeFtsQuery: - """sanitize_fts_query must strip FTS5 operators.""" - - @pytest.mark.parametrize( - "input_term,expected_absent", - [ - ('"quoted"', '"'), - ("term*", "*"), - ("NOT term", "NOT"), - ("col:value", ":"), - ("a AND b", "AND"), - ("a OR b", "OR"), - ("NEAR(a b)", "NEAR"), - ("(grouped)", "("), - ], - ) - def test_operators_stripped(self, input_term, expected_absent): - from learning_db_v2 import sanitize_fts_query - - result = sanitize_fts_query(input_term) - assert expected_absent not in result - - # =========================================================================== # Routing-outcome finalization: every pending dispatch reaches a terminal state # ===========================================================================