Extract Rule Cache Catalog (#23)

Author: trek-e

CONTEXT.md

@@ -100,6 +100,15 @@ is only the adapter for request validation, dependency handoff, HTTP exception m
 serialization.
 _Avoid_: rules route, rules engine route, rule API implementation.
 
+**Rule Cache Catalog**:
+The read-only Pathfinder module behavior behind cached ruling browse operations: listing rulings
+for a topic, recent ruling history across topics, topic activity summaries, malformed-cache skip
+policy, and cache ordering by `last_reused_at`. The code module lives at
+`modules/pathfinder/app/rule_cache_catalog.py`. The HTTP routes in
+`modules/pathfinder/app/routes/rule.py` only validate request shape, enforce initialization, and
+serialize the catalog result.
+_Avoid_: rule list route logic, cached ruling helper, rule history adapter.
+
 **PF Archive Import Plan**:
 The side-effect-free Pathfinder module behavior that turns an on-disk archive root into planned
 Pathfinder Vault writes: markdown walk, known NPC slug discovery, route decisions, large-import
@@ -200,6 +209,7 @@ Adapters should do only translation/auth/delegation.
 - PF2e Foundry NeDB chat import: `modules/pathfinder/app/foundry_chat_import.py`
 - PF2e Rule Query: `modules/pathfinder/app/rule_query.py`
 - Pathfinder Player Interaction: `modules/pathfinder/app/player_interaction_orchestrator.py`
+- Rule Cache Catalog: `modules/pathfinder/app/rule_cache_catalog.py`
 
 ### Authoritative flows
 - Message flow:

modules/pathfinder/app/routes/rule.py

@@ -16,7 +16,6 @@
 """
 from __future__ import annotations
 
-import logging
 import re
 
 from fastapi import APIRouter, HTTPException
@@ -31,6 +30,7 @@
     generate_ruling_from_passages,
 )
 from app.resolve_model import resolve
+from app.rule_cache_catalog import RuleCacheCatalog
 from app.rule_query import (
     RuleQueryCompositionError,
     RuleQueryDependencies,
@@ -40,15 +40,10 @@
 )
 from app.rules import (
     MAX_QUERY_CHARS,
-    RULING_CACHE_PATH_PREFIX,
     RulesIndex,
-    _parse_ruling_cache,
-    coerce_topic,
     keyword_classify_topic,
 )
 
-logger = logging.getLogger(__name__)
-
 router = APIRouter(prefix="/rule", tags=["rule"])
 
 # Module-level singletons — set by main.py lifespan, patchable in tests.
@@ -204,58 +199,14 @@ async def rule_query(req: RuleQueryRequest) -> JSONResponse:
 # --- Enumeration endpoints — no LLM, no embedding, pure Obsidian directory walks ---
 
 
-def _build_ruling_index_entry(path: str, parsed: dict, topic: str | None = None) -> dict:
-    """Extract the summary fields a UI / bot would list from a cached ruling."""
-    # hash is the last path component without .md.
-    hash_part = path.rsplit("/", 1)[-1].removesuffix(".md")
-    return {
-        "hash": hash_part,
-        "topic": topic or parsed.get("topic"),
-        "question": parsed.get("question", ""),
-        "composed_at": parsed.get("composed_at", ""),
-        "last_reused_at": parsed.get("last_reused_at", parsed.get("composed_at", "")),
-        "marker": parsed.get("marker", ""),
-        "source": parsed.get("source"),
-    }
-
-
-async def _collect_rulings_under(prefix: str) -> list[tuple[str, dict]]:
-    """Walk a topic-prefix (or root rulings prefix), return list of (path, parsed_frontmatter)."""
-    if obsidian is None:
-        return []
-    try:
-        paths = await obsidian.list_directory(prefix)
-    except Exception as exc:
-        logger.warning("_collect_rulings_under: list_directory %s failed: %s", prefix, exc)
-        return []
-    out: list[tuple[str, dict]] = []
-    for p in paths:
-        if not p.endswith(".md"):
-            continue
-        text = await obsidian.get_note(p)
-        if not text:
-            continue
-        parsed = _parse_ruling_cache(text)
-        if parsed is None:
-            logger.warning("_collect_rulings_under: malformed cache at %s — skipping", p)
-            continue
-        out.append((p, parsed))
-    return out
-
-
 @router.post("/show")
 async def rule_show(req: RuleShowRequest) -> JSONResponse:
     """List rulings under a given topic folder. Sorted by last_reused_at desc."""
     if obsidian is None:
         raise HTTPException(
             status_code=503, detail={"error": "rule subsystem not initialised"}
         )
-    topic = coerce_topic(req.topic)
-    prefix = f"{RULING_CACHE_PATH_PREFIX}/{topic}/"
-    collected = await _collect_rulings_under(prefix)
-    entries = [_build_ruling_index_entry(p, parsed, topic=topic) for p, parsed in collected]
-    entries.sort(key=lambda e: e.get("last_reused_at", ""), reverse=True)
-    return JSONResponse({"topic": topic, "count": len(entries), "rulings": entries})
+    return JSONResponse(await RuleCacheCatalog(obsidian).show_topic(req.topic))
 
 
 @router.post("/history")
@@ -265,29 +216,7 @@ async def rule_history(req: RuleHistoryRequest) -> JSONResponse:
         raise HTTPException(
             status_code=503, detail={"error": "rule subsystem not initialised"}
         )
-    n = req.n  # already clamped to [1, 100] by field_validator
-    root_prefix = f"{RULING_CACHE_PATH_PREFIX}/"
-    try:
-        all_paths = await obsidian.list_directory(root_prefix)
-    except Exception as exc:
-        logger.warning("rule_history: root list_directory failed: %s", exc)
-        all_paths = []
-    all_entries: list[dict] = []
-    for p in all_paths:
-        if not p.endswith(".md"):
-            continue
-        text = await obsidian.get_note(p)
-        if not text:
-            continue
-        parsed = _parse_ruling_cache(text)
-        if parsed is None:
-            continue
-        # topic = path segment between the root prefix and the final filename.
-        stripped = p.removeprefix(root_prefix)
-        topic = stripped.split("/", 1)[0] if "/" in stripped else "misc"
-        all_entries.append(_build_ruling_index_entry(p, parsed, topic=topic))
-    all_entries.sort(key=lambda e: e.get("last_reused_at", ""), reverse=True)
-    return JSONResponse({"n": n, "rulings": all_entries[:n]})
+    return JSONResponse(await RuleCacheCatalog(obsidian).history(req.n))
 
 
 @router.post("/list")
@@ -297,32 +226,4 @@ async def rule_list() -> JSONResponse:
         raise HTTPException(
             status_code=503, detail={"error": "rule subsystem not initialised"}
         )
-    root_prefix = f"{RULING_CACHE_PATH_PREFIX}/"
-    try:
-        all_paths = await obsidian.list_directory(root_prefix)
-    except Exception as exc:
-        logger.warning("rule_list: root list_directory failed: %s", exc)
-        all_paths = []
-    # Group by topic slug (first segment after root_prefix).
-    per_topic: dict[str, dict] = {}
-    for p in all_paths:
-        if not p.endswith(".md"):
-            continue
-        stripped = p.removeprefix(root_prefix)
-        if "/" not in stripped:
-            continue
-        topic = stripped.split("/", 1)[0]
-        text = await obsidian.get_note(p)
-        if not text:
-            continue
-        parsed = _parse_ruling_cache(text)
-        if parsed is None:
-            continue
-        bucket = per_topic.setdefault(topic, {"slug": topic, "count": 0, "last_activity": ""})
-        bucket["count"] += 1
-        last_act = parsed.get("last_reused_at", parsed.get("composed_at", ""))
-        if last_act > bucket["last_activity"]:
-            bucket["last_activity"] = last_act
-    topics = list(per_topic.values())
-    topics.sort(key=lambda t: t.get("last_activity", ""), reverse=True)
-    return JSONResponse({"topics": topics})
+    return JSONResponse(await RuleCacheCatalog(obsidian).topics())

modules/pathfinder/app/rule_cache_catalog.py

@@ -0,0 +1,103 @@
+"""Cached Pathfinder ruling catalog."""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from app.rules import RULING_CACHE_PATH_PREFIX, _parse_ruling_cache, coerce_topic
+
+logger = logging.getLogger(__name__)
+
+
+class RuleCacheCatalog:
+    """Read-only catalog for cached Pathfinder rulings."""
+
+    def __init__(self, obsidian: Any) -> None:
+        self._obsidian = obsidian
+
+    async def show_topic(self, topic: str) -> dict:
+        """List rulings under one topic, newest reuse first."""
+        coerced = coerce_topic(topic)
+        prefix = f"{RULING_CACHE_PATH_PREFIX}/{coerced}/"
+        collected = await self._collect_rulings_under(prefix)
+        entries = [
+            self._build_ruling_index_entry(path, parsed, topic=coerced)
+            for path, parsed in collected
+        ]
+        entries.sort(key=lambda entry: entry.get("last_reused_at", ""), reverse=True)
+        return {"topic": coerced, "count": len(entries), "rulings": entries}
+
+    async def history(self, n: int) -> dict:
+        """Return the N most recent rulings across all topics."""
+        root_prefix = f"{RULING_CACHE_PATH_PREFIX}/"
+        entries: list[dict] = []
+        for path, parsed in await self._collect_rulings_under(root_prefix):
+            stripped = path.removeprefix(root_prefix)
+            topic = stripped.split("/", 1)[0] if "/" in stripped else "misc"
+            entries.append(self._build_ruling_index_entry(path, parsed, topic=topic))
+        entries.sort(key=lambda entry: entry.get("last_reused_at", ""), reverse=True)
+        return {"n": n, "rulings": entries[:n]}
+
+    async def topics(self) -> dict:
+        """Enumerate topic folders and their cache activity."""
+        root_prefix = f"{RULING_CACHE_PATH_PREFIX}/"
+        per_topic: dict[str, dict] = {}
+        for path, parsed in await self._collect_rulings_under(root_prefix):
+            stripped = path.removeprefix(root_prefix)
+            if "/" not in stripped:
+                continue
+            topic = stripped.split("/", 1)[0]
+            bucket = per_topic.setdefault(
+                topic, {"slug": topic, "count": 0, "last_activity": ""}
+            )
+            bucket["count"] += 1
+            last_activity = parsed.get("last_reused_at", parsed.get("composed_at", ""))
+            if last_activity > bucket["last_activity"]:
+                bucket["last_activity"] = last_activity
+        topics = list(per_topic.values())
+        topics.sort(key=lambda item: item.get("last_activity", ""), reverse=True)
+        return {"topics": topics}
+
+    async def _collect_rulings_under(self, prefix: str) -> list[tuple[str, dict]]:
+        """Walk a ruling prefix and return parsed cache notes."""
+        try:
+            paths = await self._obsidian.list_directory(prefix)
+        except Exception as exc:
+            logger.warning("rule catalog: list_directory %s failed: %s", prefix, exc)
+            return []
+
+        collected: list[tuple[str, dict]] = []
+        for path in paths:
+            if not path.endswith(".md"):
+                continue
+            text = await self._obsidian.get_note(path)
+            if not text:
+                continue
+            parsed = _parse_ruling_cache(text)
+            if parsed is None:
+                logger.warning("rule catalog: malformed cache at %s, skipping", path)
+                continue
+            collected.append((path, parsed))
+        return collected
+
+    @staticmethod
+    def _build_ruling_index_entry(
+        path: str, parsed: dict, topic: str | None = None
+    ) -> dict:
+        """Extract the summary fields a UI or Discord adapter would list."""
+        hash_part = path.rsplit("/", 1)[-1].removesuffix(".md")
+        return {
+            "hash": hash_part,
+            "topic": topic or parsed.get("topic"),
+            "question": parsed.get("question", ""),
+            "composed_at": parsed.get("composed_at", ""),
+            "last_reused_at": parsed.get(
+                "last_reused_at", parsed.get("composed_at", "")
+            ),
+            "marker": parsed.get("marker", ""),
+            "source": parsed.get("source"),
+        }
+
+
+__all__ = ["RuleCacheCatalog"]