itdove/lore#83: Refactor: replace level magic numbers with LevelPolicy object

src/lore/cli.py

@@ -6,6 +6,8 @@
 import sys
 from pathlib import Path
 
+from lore.config.models import IMPLICIT_LEVELS
+
 _RECALL_PREAMBLE = "IMPORTANT: The following team knowledge is relevant to this prompt."
 
 
@@ -501,7 +503,7 @@ def _build_hierarchy_filters(
 ) -> tuple[list[int] | None, list[tuple[str, str]] | None]:
     from lore.config.utils import get_project_remote
 
-    levels = [1]
+    levels: list[int] = []
     repos: list[tuple[str, str]] = []
     remote = get_project_remote()
     if remote[0]:
@@ -662,7 +664,7 @@ def _cmd_config_add_level(args: argparse.Namespace) -> int:
         )
         return 1
 
-    if args.level < 2:
+    if args.level in IMPLICIT_LEVELS:
         print(
             "Levels 0 and 1 are reserved (individual and project). Use 2+.",
             file=sys.stderr,

src/lore/config/__init__.py

@@ -1,9 +1,13 @@
 from lore.config.loaders import _clear_config_cache
 from lore.config.manager import get_global_config, get_project_config
 from lore.config.models import (
+    IMPLICIT_LEVELS,
+    INDIVIDUAL_LEVEL,
+    PROJECT_LEVEL,
     GitConfig,
     GlobalConfig,
     HierarchyLevel,
+    LevelPolicy,
     LLMConfig,
     ProjectConfig,
     SearchConfig,
@@ -24,10 +28,14 @@
 )
 
 __all__ = [
+    "IMPLICIT_LEVELS",
+    "INDIVIDUAL_LEVEL",
+    "PROJECT_LEVEL",
     "GitConfig",
     "GlobalConfig",
     "HierarchyLevel",
     "LLMConfig",
+    "LevelPolicy",
     "ProjectConfig",
     "SearchConfig",
     "StoreConfig",

src/lore/config/manager.py

@@ -11,6 +11,7 @@
     load_project_config,
 )
 from lore.config.models import (
+    IMPLICIT_LEVELS,
     CaptureConfig,
     GitConfig,
     GlobalConfig,
@@ -118,7 +119,7 @@ def get_project_config(project_dir: str | Path | None = None) -> ProjectConfig:
     for entry in hierarchy_raw:
         if "level" not in entry or "repo" not in entry:
             continue
-        if entry["level"] < 2:
+        if entry["level"] in IMPLICIT_LEVELS:
             logger.warning(
                 "Skipping hierarchy entry with level %d (repo=%s): "
                 "levels 0-1 reserved for implicit individual/project",

src/lore/config/models.py

@@ -2,6 +2,23 @@
 
 from dataclasses import dataclass, field
 
+INDIVIDUAL_LEVEL = 0
+PROJECT_LEVEL = 1
+IMPLICIT_LEVELS: frozenset[int] = frozenset({INDIVIDUAL_LEVEL, PROJECT_LEVEL})
+
+
+@dataclass
+class LevelPolicy:
+    level: int
+    name: str
+    repo_url: str | None
+    repo_branch: str | None
+    writable: bool
+    stores_locally: bool
+    creates_pr: bool
+    locally_deletable: bool
+    pr_path_prefix: str | None
+
 
 @dataclass
 class HierarchyLevel:

src/lore/mcp/server.py

@@ -7,6 +7,12 @@
 from mcp.server.fastmcp import FastMCP
 
 from lore.config.manager import get_global_config, get_project_config
+from lore.config.models import (
+    IMPLICIT_LEVELS,
+    INDIVIDUAL_LEVEL,
+    PROJECT_LEVEL,
+    LevelPolicy,
+)
 from lore.config.utils import get_project_remote
 from lore.embedding import EmbeddingProvider
 from lore.embedding import get_embedding_provider as _create_embedding_provider
@@ -103,18 +109,71 @@ def _create_shared_pr(
         return None
 
 
-def _resolve_level(
-    level_name: str,
-) -> tuple[int, str, str | None, str | None, bool]:
+def _submit_pr(
+    policy: LevelPolicy,
+    key: str,
+    content: str,
+    frontmatter: dict,
+    title: str,
+    body: str,
+) -> str | None:
+    if not (policy.creates_pr and policy.repo_url):
+        return None
+    override = (
+        f"{policy.pr_path_prefix}{key_to_path(key)}" if policy.pr_path_prefix else None
+    )
+    return _create_shared_pr(
+        key=key,
+        content=content,
+        frontmatter=frontmatter,
+        title=title,
+        body=body,
+        repo_url=policy.repo_url,
+        repo_branch=policy.repo_branch,
+        file_path_override=override,
+    )
+
+
+def _resolve_level(level_name: str) -> LevelPolicy:
     if level_name == "individual":
-        return 0, "individual", None, None, True
+        return LevelPolicy(
+            level=INDIVIDUAL_LEVEL,
+            name="individual",
+            repo_url=None,
+            repo_branch=None,
+            writable=True,
+            stores_locally=True,
+            creates_pr=False,
+            locally_deletable=True,
+            pr_path_prefix=None,
+        )
     if level_name == "project":
         repo_url, repo_branch = get_project_remote()
-        return 1, "project", repo_url, repo_branch, True
+        return LevelPolicy(
+            level=PROJECT_LEVEL,
+            name="project",
+            repo_url=repo_url,
+            repo_branch=repo_branch,
+            writable=True,
+            stores_locally=True,
+            creates_pr=True,
+            locally_deletable=True,
+            pr_path_prefix=".lore/knowledge/",
+        )
     cfg = get_project_config()
     for h in cfg.hierarchy:
         if h.name == level_name:
-            return h.level, level_name, h.repo, h.branch, h.writable
+            return LevelPolicy(
+                level=h.level,
+                name=level_name,
+                repo_url=h.repo,
+                repo_branch=h.branch,
+                writable=h.writable,
+                stores_locally=False,
+                creates_pr=True,
+                locally_deletable=False,
+                pr_path_prefix=None,
+            )
     available = ["individual", "project"] + [h.name for h in cfg.hierarchy if h.name]
     raise ValueError(f"Unknown level '{level_name}'. Available: {available}")
 
@@ -293,8 +352,8 @@ def store_knowledge(
             PR URL for project and shared levels via GitInterface).
         """
         _validate_key(key)
-        level_int, level_name, repo_url, repo_branch, writable = _resolve_level(level)
-        _assert_writable(level, writable)
+        policy = _resolve_level(level)
+        _assert_writable(level, policy.writable)
         store = _get_store()
         cfg = get_global_config()
 
@@ -305,13 +364,13 @@ def store_knowledge(
                 if emb_vec:
                     embedding_blob = embed_to_blob(emb_vec)
                     dupes = store.query_vector(
-                        emb_vec, limit=1, filter_levels=[level_int]
+                        emb_vec, limit=1, filter_levels=[policy.level]
                     )
                     if dupes:
                         closest, dist = dupes[0]
                         if (
                             dist < cfg.search.dedup_threshold
-                            and closest.level == level_int
+                            and closest.level == policy.level
                         ):
                             store.update(
                                 closest.key,
@@ -334,50 +393,44 @@ def store_knowledge(
         tags_list = [t.strip() for t in (tags or "").split(",") if t.strip()]
         fm = {"tags": tags_list, "created_by": "lore-agent"}
 
-        pr_url = None
-        if level_int >= 1 and repo_url:
-            override = f".lore/knowledge/{key_to_path(key)}" if level_int == 1 else None
-            pr_url = _create_shared_pr(
-                key=key,
-                content=value,
-                frontmatter=fm,
-                title=f"lore: add {key}",
-                body=f"Auto-generated by lore store_knowledge"
-                f" at level '{level_name}'",
-                repo_url=repo_url,
-                repo_branch=repo_branch,
-                file_path_override=override,
-            )
-            if level_int > 1:
-                return {
-                    "key": key,
-                    "level": level_int,
-                    "pr_url": pr_url,
-                }
-
-        existing = store.get_by_key_and_level(key, level_int)
+        pr_url = _submit_pr(
+            policy,
+            key=key,
+            content=value,
+            frontmatter=fm,
+            title=f"lore: add {key}",
+            body=f"Auto-generated by lore store_knowledge" f" at level '{policy.name}'",
+        )
+        if pr_url is not None and not policy.stores_locally:
+            return {
+                "key": key,
+                "level": policy.level,
+                "pr_url": pr_url,
+            }
+
+        existing = store.get_by_key_and_level(key, policy.level)
         if existing:
             store.update(
                 key,
                 value,
                 reason="updated via store_knowledge",
                 actor="mcp",
                 tags=tags,
-                level=level_int,
+                level=policy.level,
             )
             entry_id = existing.id
         else:
             entry = KnowledgeEntry(
                 key=key,
                 value=value,
                 tags=tags or "",
-                level=level_int,
-                level_name=level_name,
+                level=policy.level,
+                level_name=policy.name,
                 embedding=embedding_blob,
             )
             entry_id = store.store(entry)
 
-        return {"id": entry_id, "key": key, "level": level_int, "pr_url": pr_url}
+        return {"id": entry_id, "key": key, "level": policy.level, "pr_url": pr_url}
 
     @server.tool()
     def negate_knowledge(
@@ -400,34 +453,29 @@ def negate_knowledge(
             PR URL for project and shared levels).
         """
         _validate_key(key)
-        level_int, level_name, repo_url, repo_branch, writable = _resolve_level(level)
-        _assert_writable(level, writable)
+        policy = _resolve_level(level)
+        _assert_writable(level, policy.writable)
         store = _get_store()
-        existing = store.get_by_key_and_level(key, level_int)
+        existing = store.get_by_key_and_level(key, policy.level)
         if existing is None:
             return {"error": f"Key not found: '{key}' at level '{level}'"}
 
         negation_content = f"[NEGATED] {reason}\n\n" f"Previous value: {existing.value}"
         fm = {"created_by": "lore-agent", "negated": True}
 
-        pr_url = None
-        if level_int >= 1 and repo_url:
-            override = f".lore/knowledge/{key_to_path(key)}" if level_int == 1 else None
-            pr_url = _create_shared_pr(
-                key=key,
-                content=negation_content,
-                frontmatter=fm,
-                title=f"lore: negate {key}",
-                body=f"Auto-generated by lore negate_knowledge: {reason}",
-                repo_url=repo_url,
-                repo_branch=repo_branch,
-                file_path_override=override,
-            )
-            if level_int > 1:
-                return {"key": key, "negated": True, "pr_url": pr_url}
+        pr_url = _submit_pr(
+            policy,
+            key=key,
+            content=negation_content,
+            frontmatter=fm,
+            title=f"lore: negate {key}",
+            body=f"Auto-generated by lore negate_knowledge: {reason}",
+        )
+        if pr_url is not None and not policy.stores_locally:
+            return {"key": key, "negated": True, "pr_url": pr_url}
 
         try:
-            store.negate(key, reason, level=level_int)
+            store.negate(key, reason, level=policy.level)
         except ValueError as exc:
             return {"error": str(exc)}
 
@@ -451,20 +499,20 @@ def delete_knowledge(
         Returns:
             Confirmation of deletion, or error for shared entries.
         """
-        level_int, _, _, _, writable = _resolve_level(level)
-        _assert_writable(level, writable)
+        policy = _resolve_level(level)
+        _assert_writable(level, policy.writable)
         store = _get_store()
-        existing = store.get_by_key_and_level(key, level_int)
+        existing = store.get_by_key_and_level(key, policy.level)
         if existing is None:
             other = store.get(key)
-            if other is not None and other.level > 1:
+            if other is not None and other.level not in IMPLICIT_LEVELS:
                 return {"error": _SHARED_DELETE_ERR}
             return {"error": f"Key not found: '{key}' at level '{level}'"}
 
-        if existing.level > 1:
+        if not policy.locally_deletable:
             return {"error": _SHARED_DELETE_ERR}
 
-        store.delete(key, reason="deleted via MCP", actor="mcp", level=level_int)
+        store.delete(key, reason="deleted via MCP", actor="mcp", level=policy.level)
         return {"key": key, "deleted": True}
 
     return server

src/lore/store/sqlite.py

@@ -7,6 +7,7 @@
 
 import sqlite_vec
 
+from lore.config.models import IMPLICIT_LEVELS
 from lore.embedding.base import blob_to_embed, cosine_distance, embed_to_blob
 from lore.store.base import HistoryRecord, KnowledgeEntry, StoreBackend
 
@@ -195,6 +196,7 @@ def _build_filter(
         filter_repos: list[tuple[str, str]] | None,
         include_negated: bool,
         col_prefix: str = "",
+        implicit_levels: frozenset[int] = IMPLICIT_LEVELS,
     ) -> tuple[list[str], list]:
         p = f"{col_prefix}." if col_prefix else ""
         conditions = [
@@ -205,13 +207,16 @@ def _build_filter(
         params: list = []
 
         if filter_levels is not None:
-            levels = sorted(set(filter_levels) | {0, 1})
+            levels = sorted(set(filter_levels) | implicit_levels)
             placeholders = ", ".join("?" for _ in levels)
             conditions.append(f"{p}level IN ({placeholders})")
             params.extend(levels)
 
         if filter_repos is not None:
-            repo_clauses = [f"{p}level = 0", f"{p}level = 1"]
+            repo_clauses = []
+            for lv in sorted(implicit_levels):
+                repo_clauses.append(f"{p}level = ?")
+                params.append(lv)
             for repo_url, repo_branch in filter_repos:
                 repo_clauses.append(f"({p}repo_url = ? AND {p}repo_branch = ?)")
                 params.extend([repo_url, repo_branch])