fix: bugs and code quality in analyses system

- Fix MCP save_analysis always reporting "updated" (checked existence after upsert)
- Fix SPA analysis section using global DB instead of db parameter
- Fix parser crash on non-dict YAML frontmatter
- Extract _save_tags/_fetch_tags helpers in db.py (DRY notes + analyses)
- Move json import to module level in server.py
- Modernize test fixtures: tmp_path instead of tempfile + manual cleanup
- Add edge case tests: unclosed frontmatter, non-dict YAML, empty body

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: queelius

src/chartfold/analysis_parser.py

@@ -46,7 +46,8 @@ def parse_analysis_file(path: str | Path) -> dict:
         if len(parts) >= 3:
             yaml_text = parts[1]
             content = parts[2].lstrip("\n")
-            frontmatter = yaml.safe_load(yaml_text) or {}
+            parsed_yaml = yaml.safe_load(yaml_text)
+            frontmatter = parsed_yaml if isinstance(parsed_yaml, dict) else {}
 
     title = frontmatter.pop("title", None)
     if not title:

src/chartfold/db.py

@@ -231,6 +231,27 @@ def last_load_counts(self, source: str) -> dict[str, int] | None:
             "source_assets": row.get("source_assets_count", 0),
         }
 
+    # --- Tag helpers (shared by notes and analyses) ---
+
+    def _save_tags(self, table: str, fk_col: str, fk_id: int, tags: list[str]) -> None:
+        """Replace all tags for a record: delete existing, insert new."""
+        self.conn.execute(f"DELETE FROM {table} WHERE {fk_col}=?", (fk_id,))
+        for tag in tags:
+            clean = tag.strip()
+            if clean:
+                self.conn.execute(
+                    f"INSERT OR IGNORE INTO {table} ({fk_col}, tag) VALUES (?, ?)",
+                    (fk_id, clean),
+                )
+
+    def _fetch_tags(self, table: str, fk_col: str, fk_id: int) -> list[str]:
+        """Fetch sorted tags for a record."""
+        rows = self.query(
+            f"SELECT tag FROM {table} WHERE {fk_col} = ? ORDER BY tag",
+            (fk_id,),
+        )
+        return [r["tag"] for r in rows]
+
     # --- Personal notes CRUD ---
 
     def save_note(
@@ -254,7 +275,6 @@ def save_note(
                     "WHERE id=?",
                     (title, content, now, ref_table, ref_id, note_id),
                 )
-                self.conn.execute("DELETE FROM note_tags WHERE note_id=?", (note_id,))
             else:
                 # Create new note
                 cursor = self.conn.execute(
@@ -265,13 +285,7 @@ def save_note(
                 # lastrowid is always int after INSERT
                 note_id = cursor.lastrowid or 0
 
-            for tag in tags:
-                clean_tag = tag.strip()
-                if clean_tag:
-                    self.conn.execute(
-                        "INSERT OR IGNORE INTO note_tags (note_id, tag) VALUES (?, ?)",
-                        (note_id, clean_tag),
-                    )
+            self._save_tags("note_tags", "note_id", note_id, tags)
 
         return note_id
 
@@ -285,11 +299,7 @@ def get_note(self, note_id: int) -> dict | None:
         if not rows:
             return None
         note = rows[0]
-        tag_rows = self.query(
-            "SELECT tag FROM note_tags WHERE note_id = ? ORDER BY tag",
-            (note_id,),
-        )
-        note["tags"] = [r["tag"] for r in tag_rows]
+        note["tags"] = self._fetch_tags("note_tags", "note_id", note_id)
         return note
 
     def search_notes_personal(
@@ -333,13 +343,8 @@ def search_notes_personal(
             tuple(params),
         )
 
-        # Attach tags to each result
         for row in rows:
-            tag_rows = self.query(
-                "SELECT tag FROM note_tags WHERE note_id = ? ORDER BY tag",
-                (row["id"],),
-            )
-            row["tags"] = [r["tag"] for r in tag_rows]
+            row["tags"] = self._fetch_tags("note_tags", "note_id", row["id"])
 
         return rows
 
@@ -379,7 +384,6 @@ def save_analysis(
                     "category=?, summary=?, source=?, updated_at=? WHERE id=?",
                     (title, content, frontmatter_json, category, summary, source, now, analysis_id),
                 )
-                self.conn.execute("DELETE FROM analysis_tags WHERE analysis_id=?", (analysis_id,))
             else:
                 cursor = self.conn.execute(
                     "INSERT INTO analyses (slug, title, content, frontmatter, "
@@ -389,13 +393,7 @@ def save_analysis(
                 )
                 analysis_id = cursor.lastrowid or 0
 
-            for tag in tags:
-                clean_tag = tag.strip()
-                if clean_tag:
-                    self.conn.execute(
-                        "INSERT OR IGNORE INTO analysis_tags (analysis_id, tag) VALUES (?, ?)",
-                        (analysis_id, clean_tag),
-                    )
+            self._save_tags("analysis_tags", "analysis_id", analysis_id, tags)
 
         return analysis_id
 
@@ -409,11 +407,7 @@ def get_analysis(self, slug_or_id: str | int) -> dict | None:
         if not rows:
             return None
         analysis = rows[0]
-        tag_rows = self.query(
-            "SELECT tag FROM analysis_tags WHERE analysis_id = ? ORDER BY tag",
-            (analysis["id"],),
-        )
-        analysis["tags"] = [r["tag"] for r in tag_rows]
+        analysis["tags"] = self._fetch_tags("analysis_tags", "analysis_id", analysis["id"])
         return analysis
 
     def search_analyses(
@@ -458,11 +452,7 @@ def search_analyses(
         )
 
         for row in rows:
-            tag_rows = self.query(
-                "SELECT tag FROM analysis_tags WHERE analysis_id = ? ORDER BY tag",
-                (row["id"],),
-            )
-            row["tags"] = [r["tag"] for r in tag_rows]
+            row["tags"] = self._fetch_tags("analysis_tags", "analysis_id", row["id"])
 
         return rows
 

src/chartfold/mcp/server.py

@@ -6,6 +6,7 @@
 
 from __future__ import annotations
 
+import json
 import os
 import re
 
@@ -674,8 +675,6 @@ def save_analysis(
         frontmatter_yaml: Optional YAML string with additional metadata fields.
             Will be parsed and stored as JSON for json_extract() queries.
     """
-    import json
-
     db = _get_db()
     try:
         tag_list = [t.strip() for t in tags.split(",") if t.strip()] if tags else []
@@ -690,6 +689,7 @@ def save_analysis(
             except Exception:
                 frontmatter_json = frontmatter_yaml  # Store as-is if not valid YAML
 
+        existing_before = db.get_analysis(slug)
         saved_id = db.save_analysis(
             slug=slug,
             title=title,
@@ -701,9 +701,7 @@ def save_analysis(
             source=source,
         )
 
-        existing = db.get_analysis(slug)
-        is_update = existing and existing["id"] == saved_id
-        return {"id": saved_id, "slug": slug, "status": "updated" if is_update else "created"}
+        return {"id": saved_id, "slug": slug, "status": "updated" if existing_before else "created"}
     finally:
         db.close()
 

src/chartfold/spa/js/sections.js

@@ -1095,7 +1095,7 @@ const Sections = {
     var data = [];
     var fromDB = false;
     try {
-      var dbRows = DB.query("SELECT slug, title, category, summary, content, source FROM analyses ORDER BY updated_at DESC");
+      var dbRows = db.query("SELECT slug, title, category, summary, content, source FROM analyses ORDER BY updated_at DESC");
       if (dbRows.length > 0) {
         data = dbRows.map(function(r) {
           return { slug: r.slug, title: r.title, body: r.content, filename: r.slug + '.md', category: r.category, summary: r.summary };