feat: archive, import_entry and Denkarium migration (PB-11, #3994)

CHANGELOG.md

@@ -2,6 +2,10 @@
 
 ## [Unreleased]
 
+- PB-11 (BACH #3994): `NoteStore` can now replace the BACH Denkarium. Additive columns `is_archived`, `archived_reason`, `archived_at`, `legacy_id` are migrated when an existing database is opened; new `archive()`, `unarchive()`, `delete()`, `import_entry()` (keeps foreign timestamps, idempotent per `legacy_id`); `list_entries()`/`search()` hide archived entries unless `include_archived=True`; entry types accept the aliases `denkarium` (note) and `logbuch` (logbook); `Entry` and `stats()` expose the new fields; `promote()` can record `promoted_id`.
+- Added the one-off migration `python -m llm_note.migrate_denkarium SOURCE_DB TARGET_DB` (dry run by default; `--apply --snapshot-dir` snapshots the source table, imports, aborts on a count mismatch; the source table is only read).
+- The package version stays frozen at `1.0.5` (see RELEASE_GATE.md); a `1.1.0` bump needs a separate release decision.
+
 - Added vector SVG interface preview for Denkarium Local Web GUI (`assets/denkarium-preview.svg`) and integrated it into Section 8 across all six README localizations.
 - Corrected REST API endpoint documentation in Section 8 and sequence diagrams in Section 4 across all six README files to reference `/api/entries`.
 - Hardened `pytest.ini` with `norecursedirs` to exclude `references/` from recursive test collection.

llm_note/migrate_denkarium.py

@@ -0,0 +1,131 @@
+"""One-off, idempotent migration of BACH ``denkarium_entries`` into llm-note.
+
+Dry run is the default. ``--apply`` first snapshots the source table, then imports
+via :meth:`NoteStore.import_entry` (keeps timestamps, remembers the old id as
+``legacy_id``) and aborts on a count mismatch. The source table is only read.
+
+    python -m llm_note.migrate_denkarium SOURCE_DB TARGET_DB [--apply --snapshot-dir DIR]
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import sqlite3
+import sys
+from datetime import datetime, timezone
+from pathlib import Path
+
+from .store import ENTRY_TYPE_ALIASES, NoteStore
+
+_TABLE = "denkarium_entries"
+_VALID_TYPES = set(ENTRY_TYPE_ALIASES) | set(ENTRY_TYPE_ALIASES.values())
+
+
+class MigrationError(RuntimeError):
+    """Raised when the source cannot be mapped losslessly."""
+
+
+def _read_source(source_db: Path) -> list[dict]:
+    conn = sqlite3.connect(f"file:{source_db.as_posix()}?mode=ro", uri=True)
+    conn.row_factory = sqlite3.Row
+    try:
+        return [dict(r) for r in conn.execute(f"SELECT * FROM {_TABLE} ORDER BY id")]
+    finally:
+        conn.close()
+
+
+def _snapshot(source_db: Path, snapshot_dir: Path) -> Path:
+    snapshot_dir.mkdir(parents=True, exist_ok=True)
+    out = snapshot_dir / f"{_TABLE}_{datetime.now(timezone.utc):%Y%m%d-%H%M%S-%f}.db"
+    src = sqlite3.connect(f"file:{source_db.as_posix()}?mode=ro", uri=True)
+    dst = sqlite3.connect(out)
+    try:
+        rows = src.execute(f"SELECT * FROM {_TABLE}")
+        cols = [d[0] for d in rows.description]
+        dst.execute(f"CREATE TABLE {_TABLE} ({', '.join(cols)})")
+        dst.executemany(
+            f"INSERT INTO {_TABLE} VALUES ({', '.join('?' * len(cols))})", rows.fetchall()
+        )
+        dst.commit()
+    finally:
+        src.close()
+        dst.close()
+    return out
+
+
+def _kwargs(row: dict) -> dict:
+    if row.get("entry_type") not in _VALID_TYPES:
+        raise MigrationError(f"id {row['id']}: unknown entry_type {row.get('entry_type')!r}")
+    if not row.get("content"):
+        raise MigrationError(f"id {row['id']}: empty content")
+    mood = row.get("mood")
+    return {
+        "entry_type": row["entry_type"],
+        "category": row.get("category") or "note",
+        "title": row.get("title"),
+        "mood": mood if isinstance(mood, int) and 1 <= mood <= 5 else None,  # 0 -> NULL
+        "source": row.get("source") or "user",
+        "tags": row.get("tags"),
+        "promoted_to": row.get("promoted_to"),
+        "promoted_id": row.get("promoted_id"),
+        "created_at": row.get("created_at"),
+        "updated_at": row.get("updated_at"),
+        "legacy_id": row["id"],
+        "is_archived": bool(row.get("is_archived")),
+        "archived_reason": row.get("archived_reason"),
+        "archived_at": row.get("archived_at"),
+    }
+
+
+def migrate(
+    source_db: str | Path,
+    target_db: str | Path,
+    *,
+    apply: bool = False,
+    snapshot_dir: str | Path | None = None,
+) -> dict:
+    source_db, target_db = Path(source_db), Path(target_db)
+    rows = _read_source(source_db)
+    prepared = [(r["content"], _kwargs(r)) for r in rows]  # preflight: all or nothing
+    report: dict = {"dry_run": not apply, "source_count": len(rows), "imported": 0}
+    if not apply:
+        return report
+    if snapshot_dir is None:
+        raise MigrationError("apply requires a snapshot_dir (snapshot of the source first)")
+    report["snapshot"] = str(_snapshot(source_db, Path(snapshot_dir)))
+
+    store = NoteStore(target_db)
+    # ponytail: list limit 1000 (MAX_QUERY_LIMIT); Denkarium has ~14 rows, add paging if ever larger
+    known = {e.legacy_id for e in store.list_entries(include_archived=True, limit=1000)}
+    mapping: dict[str, int] = {}
+    for content, kw in prepared:
+        entry = store.import_entry(content, **kw)
+        mapping[str(kw["legacy_id"])] = entry.id
+        report["imported"] += kw["legacy_id"] not in known
+    mig_ids = {e.legacy_id for e in store.list_entries(include_archived=True, limit=1000)}
+    report["mapping"] = mapping
+    report["target_count"] = len(mig_ids & {r["id"] for r in rows})
+    if report["target_count"] != len(rows):
+        raise MigrationError(f"count mismatch: source {len(rows)} != target {report['target_count']}")
+    return report
+
+
+def main(argv: list[str] | None = None) -> int:
+    ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
+    ap.add_argument("source_db")
+    ap.add_argument("target_db")
+    ap.add_argument("--apply", action="store_true", help="really write (default: dry run)")
+    ap.add_argument("--snapshot-dir")
+    args = ap.parse_args(argv)
+    try:
+        report = migrate(args.source_db, args.target_db, apply=args.apply, snapshot_dir=args.snapshot_dir)
+    except MigrationError as exc:
+        print(f"ABORT: {exc}", file=sys.stderr)
+        return 2
+    print(json.dumps(report, indent=2, ensure_ascii=False))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

llm_note/store.py

@@ -16,6 +16,14 @@
 from uuid import uuid4
 
 MAX_QUERY_LIMIT = 1000
+# BACH Denkarium vocabulary -> llm-note vocabulary
+ENTRY_TYPE_ALIASES = {"denkarium": "note", "logbuch": "logbook"}
+_ARCHIVE_COLUMNS = {
+    "is_archived": "INTEGER DEFAULT 0",
+    "archived_reason": "TEXT",
+    "archived_at": "TEXT",
+    "legacy_id": "INTEGER",
+}
 _WINDOWS_RESERVED_NAMES = {
     "CON",
     "PRN",
@@ -30,6 +38,14 @@
 )
 
 
+def _norm_type(entry_type: str | None) -> str | None:
+    return ENTRY_TYPE_ALIASES.get(entry_type, entry_type) if entry_type else entry_type
+
+
+def _now() -> str:
+    return datetime.now().astimezone().strftime("%Y-%m-%d %H:%M")
+
+
 def _validate_limit(limit: int) -> int:
     if isinstance(limit, bool) or not isinstance(limit, int):
         raise TypeError("limit must be an integer")
@@ -114,6 +130,14 @@ class Entry:
     mood: int | None = None
     promoted_to: str | None = None
     created_at: str | None = None
+    tags: str | None = None
+    source: str | None = None
+    updated_at: str | None = None
+    is_archived: bool = False
+    archived_reason: str | None = None
+    archived_at: str | None = None
+    legacy_id: int | None = None
+    promoted_id: int | None = None
 
 
 class NoteStore:
@@ -158,6 +182,10 @@ def _init_schema(self) -> None:
                 )
                 """
             )
+            have = {r["name"] for r in conn.execute("PRAGMA table_info(note_entries)")}
+            for name, ddl in _ARCHIVE_COLUMNS.items():
+                if name not in have:
+                    conn.execute(f"ALTER TABLE note_entries ADD COLUMN {name} {ddl}")
             conn.execute(
                 "CREATE INDEX IF NOT EXISTS idx_note_type ON note_entries(entry_type)"
             )
@@ -178,7 +206,8 @@ def write(
         mood: int | None = None,
         source: str = "user",
     ) -> Entry:
-        now = datetime.now().astimezone().strftime("%Y-%m-%d %H:%M")
+        now = _now()
+        entry_type = _norm_type(entry_type)
         if entry_type == "logbook" and not title:
             title = f"Log {now}"
 
@@ -209,14 +238,17 @@ def list_entries(
         entry_type: str | None = None,
         category: str | None = None,
         limit: int = 10,
+        include_archived: bool = False,
     ) -> list[Entry]:
         limit = _validate_limit(limit)
         query = "SELECT * FROM note_entries"
         where: list[str] = []
         params: list[object] = []
+        if not include_archived:
+            where.append("COALESCE(is_archived, 0) = 0")
         if entry_type:
             where.append("entry_type = ?")
-            params.append(entry_type)
+            params.append(_norm_type(entry_type))
         if category:
             where.append("category = ?")
             params.append(category)
@@ -235,6 +267,7 @@ def search(
         entry_type: str | None = None,
         category: str | None = None,
         limit: int = 20,
+        include_archived: bool = False,
     ) -> list[Entry]:
         limit = _validate_limit(limit)
         escaped_term = (
@@ -250,9 +283,11 @@ def search(
             )
         ]
         params: list[object] = [pattern, pattern, pattern, pattern]
+        if not include_archived:
+            where.append("COALESCE(is_archived, 0) = 0")
         if entry_type:
             where.append("entry_type = ?")
-            params.append(entry_type)
+            params.append(_norm_type(entry_type))
         if category:
             where.append("category = ?")
             params.append(category)
@@ -266,6 +301,79 @@ def search(
             ).fetchall()
         return [self._row_to_entry(row) for row in rows]
 
+    def import_entry(
+        self,
+        content: str,
+        *,
+        entry_type: str = "note",
+        category: str = "note",
+        title: str | None = None,
+        mood: int | None = None,
+        source: str = "user",
+        tags: str | None = None,
+        promoted_to: str | None = None,
+        promoted_id: int | None = None,
+        created_at: str | None = None,
+        updated_at: str | None = None,
+        legacy_id: int | None = None,
+        is_archived: bool = False,
+        archived_reason: str | None = None,
+        archived_at: str | None = None,
+    ) -> Entry:
+        """Insert an entry keeping foreign timestamps; idempotent per ``legacy_id``."""
+
+        with self._connect() as conn:
+            if legacy_id is not None:
+                row = conn.execute(
+                    "SELECT id FROM note_entries WHERE legacy_id = ?", (legacy_id,)
+                ).fetchone()
+                if row is not None:
+                    return self.get(int(row["id"]))
+            cur = conn.execute(
+                """
+                INSERT INTO note_entries
+                    (entry_type, title, content, category, tags, source, mood,
+                     promoted_to, promoted_id, created_at, updated_at, legacy_id,
+                     is_archived, archived_reason, archived_at)
+                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+                """,
+                (
+                    _norm_type(entry_type), title, content, category, tags, source,
+                    mood, promoted_to, promoted_id, created_at or _now(), updated_at,
+                    legacy_id, int(is_archived), archived_reason, archived_at,
+                ),
+            )
+            entry_id = int(cur.lastrowid)
+        return self.get(entry_id)
+
+    def archive(self, entry_id: int, reason: str | None = None) -> Entry:
+        self.get(entry_id)  # KeyError if unknown
+        with self._connect() as conn:
+            conn.execute(
+                "UPDATE note_entries SET is_archived = 1, archived_reason = ?,"
+                " archived_at = ? WHERE id = ?",
+                (reason, _now(), entry_id),
+            )
+        return self.get(entry_id)
+
+    def unarchive(self, entry_id: int) -> Entry:
+        self.get(entry_id)
+        with self._connect() as conn:
+            conn.execute(
+                "UPDATE note_entries SET is_archived = 0, archived_reason = NULL,"
+                " archived_at = NULL WHERE id = ?",
+                (entry_id,),
+            )
+        return self.get(entry_id)
+
+    def delete(self, entry_id: int) -> None:
+        with self._connect() as conn:
+            deleted = conn.execute(
+                "DELETE FROM note_entries WHERE id = ?", (entry_id,)
+            ).rowcount
+        if deleted == 0:
+            raise KeyError(f"Entry not found: {entry_id}")
+
     def brainstorm(self, topic: str) -> Entry:
         return self.write(
             f"Topic: {topic}\n\nIdeas:\n- ",
@@ -274,14 +382,17 @@ def brainstorm(self, topic: str) -> Entry:
             title=f"Brainstorm: {topic}",
         )
 
-    def promote(self, entry_id: int, target: str) -> Entry:
+    def promote(
+        self, entry_id: int, target: str, promoted_id: int | None = None
+    ) -> Entry:
         if target not in {"task", "wiki", "issue"}:
             raise ValueError("target must be task, wiki, or issue")
         now = datetime.now().astimezone().isoformat(timespec="seconds")
         with self._connect() as conn:
             conn.execute(
-                "UPDATE note_entries SET promoted_to = ?, updated_at = ? WHERE id = ?",
-                (target, now, entry_id),
+                "UPDATE note_entries SET promoted_to = ?,"
+                " promoted_id = COALESCE(?, promoted_id), updated_at = ? WHERE id = ?",
+                (target, promoted_id, now, entry_id),
             )
         return self.get(entry_id)
 
@@ -294,7 +405,15 @@ def stats(self) -> dict[str, int]:
             logbook = conn.execute(
                 "SELECT COUNT(*) FROM note_entries WHERE entry_type = 'logbook'"
             ).fetchone()[0]
-        return {"total": int(total), "promoted": int(promoted), "logbook": int(logbook)}
+            archived = conn.execute(
+                "SELECT COUNT(*) FROM note_entries WHERE COALESCE(is_archived, 0) = 1"
+            ).fetchone()[0]
+        return {
+            "total": int(total),
+            "promoted": int(promoted),
+            "logbook": int(logbook),
+            "archived": int(archived),
+        }
 
     @staticmethod
     def _row_to_entry(row: sqlite3.Row) -> Entry:
@@ -307,6 +426,14 @@ def _row_to_entry(row: sqlite3.Row) -> Entry:
             mood=row["mood"],
             promoted_to=row["promoted_to"],
             created_at=row["created_at"],
+            tags=row["tags"],
+            source=row["source"],
+            updated_at=row["updated_at"],
+            is_archived=bool(row["is_archived"]),
+            archived_reason=row["archived_reason"],
+            archived_at=row["archived_at"],
+            legacy_id=row["legacy_id"],
+            promoted_id=row["promoted_id"],
         )