refactor: migrate entity storage from JSON to flat markdown files

plugins/kaizen/README.md

@@ -30,7 +30,7 @@ claude --plugin-dir /path/to/kaizen/repo/plugins/kaizen
 ### Entity Retrieval (Automatic)
 
 When you submit a prompt, the plugin automatically:
-1. Loads all stored entities from `.kaizen/entities.json`
+1. Loads all stored entities from `.kaizen/entities/` (one markdown file per entity)
 2. Formats and injects them into the conversation context
 3. Claude applies relevant entities to the current task
 
@@ -41,7 +41,7 @@ By default, you must manually invoke the `/kaizen:learn` skill to extract entiti
 2. Invoke `/kaizen:learn`
 3. The plugin analyzes the conversation trajectory
 4. Extracts actionable entities from what worked/failed
-5. Saves new entities to `.kaizen/entities.json`
+5. Saves new entities as markdown files in `.kaizen/entities/{type}/`
 
 ## Example Walkthrough
 
@@ -87,24 +87,33 @@ Manually invoke to export the current conversation as a trajectory JSON file:
 
 ## Entities Storage
 
-Entities are stored in `.kaizen/entities.json`:
-
-```json
-{
-  "entities": [
-    {
-      "content": "Use Python PIL/Pillow for image metadata extraction in sandboxed environments",
-      "rationale": "System tools like exiftool may not be available",
-      "category": "strategy",
-      "trigger": "When extracting image metadata in containerized environments"
-    }
-  ]
-}
+Entities are stored as individual markdown files in `.kaizen/entities/`, nested by type:
+
+```
+.kaizen/entities/
+  guideline/
+    use-python-pil-for-image-metadata-extraction.md
+    cache-api-responses-locally.md
+```
+
+Each file uses markdown with YAML frontmatter:
+
+```markdown
+---
+type: guideline
+trigger: When extracting image metadata in containerized environments
+---
+
+Use Python PIL/Pillow for image metadata extraction in sandboxed environments
+
+## Rationale
+
+System tools like exiftool may not be available
 ```
 
 ## Environment Variables
 
-- `KAIZEN_ENTITIES_FILE`: Override the default entities storage location
+- `KAIZEN_ENTITIES_DIR`: Override the default entities directory location
 - `CLAUDE_PROJECT_ROOT`: Set by Claude Code, used to locate project-level entities
 
 ## Verification

plugins/kaizen/lib/entity_io.py

@@ -0,0 +1,290 @@
+"""Shared entity I/O utilities for the Kaizen plugin.
+
+Handles reading and writing entities as flat markdown files with YAML
+frontmatter, organized in type-nested directories.
+"""
+
+import datetime
+import getpass
+import os
+import re
+import tempfile
+from pathlib import Path
+
+
+# ---------------------------------------------------------------------------
+# Logging
+# ---------------------------------------------------------------------------
+
+
+def _get_log_dir():
+    """Get user-scoped log directory with restrictive permissions."""
+    try:
+        uid = os.getuid()
+    except AttributeError:
+        uid = getpass.getuser()
+    log_dir = os.path.join(tempfile.gettempdir(), f"kaizen-{uid}")
+    os.makedirs(log_dir, mode=0o700, exist_ok=True)
+    return log_dir
+
+
+_LOG_FILE = os.path.join(_get_log_dir(), "kaizen-plugin.log")
+
+
+def log(component, message):
+    """Append a timestamped message to the shared log file.
+
+    Args:
+        component: Short label like "retrieve" or "save".
+        message: The log line.
+    """
+    if not os.environ.get("KAIZEN_DEBUG"):
+        return
+    try:
+        timestamp = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+        with open(_LOG_FILE, "a", encoding="utf-8") as f:
+            f.write(f"[{timestamp}] [{component}] {message}\n")
+    except OSError:
+        pass
+
+
+# ---------------------------------------------------------------------------
+# Directory discovery
+# ---------------------------------------------------------------------------
+
+
+def find_entities_dir():
+    """Locate the entities directory.
+
+    Search order:
+      1. ``KAIZEN_ENTITIES_DIR`` env var (authoritative, no fallback)
+      2. ``{CLAUDE_PROJECT_ROOT}/.kaizen/entities/``
+      3. ``.kaizen/entities/`` (cwd)
+
+    Returns:
+        Path to the directory if it exists, else ``None``.
+    """
+    env_dir = os.environ.get("KAIZEN_ENTITIES_DIR")
+    if env_dir:
+        p = Path(env_dir)
+        return p if p.is_dir() else None
+
+    project_root = os.environ.get("CLAUDE_PROJECT_ROOT")
+    candidates = []
+    if project_root:
+        candidates.append(Path(project_root) / ".kaizen" / "entities")
+    candidates.append(Path(".kaizen") / "entities")
+
+    for c in candidates:
+        if c.is_dir():
+            return c
+    return None
+
+
+def get_default_entities_dir():
+    """Return (and create) the default entities directory.
+
+    Prefers ``{CLAUDE_PROJECT_ROOT}/.kaizen/entities/``, falls back to
+    ``.kaizen/entities/``.
+    """
+    project_root = os.environ.get("CLAUDE_PROJECT_ROOT", "")
+    if project_root:
+        base = Path(project_root) / ".kaizen" / "entities"
+    else:
+        base = Path(".kaizen") / "entities"
+    base.mkdir(parents=True, exist_ok=True)
+    return base.resolve()
+
+
+# ---------------------------------------------------------------------------
+# Slugify / filename helpers
+# ---------------------------------------------------------------------------
+
+
+def slugify(text, max_length=60):
+    """Convert *text* to a filesystem-safe slug.
+
+    >>> slugify("Use temp files for JSON transfer!")
+    'use-temp-files-for-json-transfer'
+    """
+    text = text.lower()
+    text = re.sub(r"[^a-z0-9]+", "-", text)
+    text = text.strip("-")
+    # Truncate at max_length, but don't break in the middle of a word
+    if len(text) > max_length:
+        text = text[:max_length].rsplit("-", 1)[0]
+    return text or "entity"
+
+
+def unique_filename(directory, slug):
+    """Return a Path that doesn't collide with existing files in *directory*.
+
+    Tries ``slug.md``, then ``slug-2.md``, ``slug-3.md``, etc.
+    """
+    directory = Path(directory)
+    candidate = directory / f"{slug}.md"
+    if not candidate.exists():
+        return candidate
+    n = 2
+    while True:
+        candidate = directory / f"{slug}-{n}.md"
+        if not candidate.exists():
+            return candidate
+        n += 1
+
+
+# ---------------------------------------------------------------------------
+# Markdown <-> dict conversion
+# ---------------------------------------------------------------------------
+
+_FRONTMATTER_KEYS = ("type", "trigger")
+
+
+def entity_to_markdown(entity):
+    """Serialize an entity dict to markdown with YAML frontmatter.
+
+    Args:
+        entity: dict with keys ``content``, and optionally ``type``,
+                ``trigger``, ``rationale``.
+
+    Returns:
+        A string suitable for writing to a ``.md`` file.
+    """
+    lines = ["---"]
+    for key in _FRONTMATTER_KEYS:
+        val = entity.get(key)
+        if val:
+            lines.append(f"{key}: {val}")
+    lines.append("---")
+    lines.append("")
+
+    content = entity.get("content", "")
+    lines.append(content)
+
+    rationale = entity.get("rationale")
+    if rationale:
+        lines.append("")
+        lines.append("## Rationale")
+        lines.append("")
+        lines.append(rationale)
+
+    lines.append("")
+    return "\n".join(lines)
+
+
+def markdown_to_entity(path):
+    """Parse a markdown entity file back into a dict.
+
+    Handles YAML frontmatter with simple ``key: value`` lines (no nested
+    structures, no PyYAML dependency).
+
+    Returns:
+        dict with ``content``, ``type``, ``trigger``, ``rationale`` keys.
+    """
+    path = Path(path)
+    text = path.read_text(encoding="utf-8")
+
+    entity = {}
+
+    # Split frontmatter
+    if text.startswith("---"):
+        parts = text.split("---", 2)
+        if len(parts) >= 3:
+            frontmatter = parts[1].strip()
+            body = parts[2]
+            for line in frontmatter.splitlines():
+                line = line.strip()
+                if not line:
+                    continue
+                key, _, value = line.partition(":")
+                key = key.strip()
+                value = value.strip()
+                if key and value:
+                    entity[key] = value
+        else:
+            body = text
+    else:
+        body = text
+
+    # Split body into content and rationale
+    body = body.strip()
+    m = re.search(r"^## Rationale", body, re.MULTILINE)
+    if m:
+        content = body[: m.start()].strip()
+        rationale = body[m.end() :].strip()
+        if rationale:
+            entity["rationale"] = rationale
+    else:
+        content = body
+
+    if content:
+        entity["content"] = content
+
+    return entity
+
+
+# ---------------------------------------------------------------------------
+# Bulk load / write
+# ---------------------------------------------------------------------------
+
+
+def load_all_entities(entities_dir):
+    """Glob ``**/*.md`` under *entities_dir* and parse each file.
+
+    Returns:
+        list of entity dicts.
+    """
+    entities_dir = Path(entities_dir)
+    entities = []
+    for md in sorted(entities_dir.glob("**/*.md")):
+        try:
+            entity = markdown_to_entity(md)
+            if entity.get("content"):
+                entities.append(entity)
+        except OSError:
+            pass
+    return entities
+
+
+def write_entity_file(directory, entity):
+    """Write a single entity as a markdown file under *directory*.
+
+    The file is placed in a ``{type}/`` subdirectory.  Uses atomic
+    write (write to ``.tmp``, then ``os.rename``).
+
+    Returns:
+        Path to the written file.
+    """
+    entity_type = entity.get("type", "general")
+    if not re.fullmatch(r"[a-z0-9][a-z0-9_-]*", entity_type):
+        entity_type = "general"
+    type_dir = Path(directory) / entity_type
+    type_dir.mkdir(parents=True, exist_ok=True)
+
+    slug = slugify(entity.get("content", "entity"))
+    content = entity_to_markdown(entity)
+
+    # Write to a unique temp file first (avoids predictable .tmp collisions)
+    fd, tmp_path = tempfile.mkstemp(dir=type_dir, suffix=".tmp", prefix=slug)
+    try:
+        os.write(fd, content.encode("utf-8"))
+        os.close(fd)
+        fd = None
+
+        # Atomically claim the target using O_EXCL to detect races
+        target = unique_filename(type_dir, slug)
+        try:
+            claim_fd = os.open(str(target), os.O_CREAT | os.O_EXCL | os.O_WRONLY)
+            os.close(claim_fd)
+        except FileExistsError:
+            # Another writer beat us — re-discover a free name
+            target = unique_filename(type_dir, slug)
+
+        os.rename(tmp_path, target)
+        return target
+    except BaseException:
+        if fd is not None:
+            os.close(fd)
+        if os.path.exists(tmp_path):
+            os.unlink(tmp_path)
+        raise