fix(maintenance): keep titled and history-referenced prompts when pruning (#239)

* fix(maintenance): keep titled and history-referenced prompts when pruning

prune_orphaned_prompts now also keeps prompts with a prompt_library_state
row (titled or explicit Library parents, whose state would cascade away)
and prompts that are the currently valid navigation hint of a recorded
run (execution_run_contexts positive/negative_prompt_id in the current
storage epoch whose exact raw snapshot text still matches). Stale hints
that history reads already ignore do not block pruning.

Fixes #237

* fix(maintenance): mirror unique-text hint rule and compute live hints once

The prune exempted a hinted prompt even when another prompt shared its
exact raw text, although history reads treat such a hint as dead. The
live-hint set now applies the reader's rule (current storage epoch, exact
raw text matching exactly one prompt, which is the hinted one) and lives
beside _prompt_hint in execution_contexts as LIVE_HINT_CTES.

The set is materialized once per statement and excluded with NOT IN, so
the prune no longer scans execution_run_contexts once per prompt. No
schema or index change.

Refs #237

database/execution_contexts.py

@@ -132,6 +132,41 @@ def _insert_exact(
         conn.execute(f"INSERT INTO {table} ({names}) VALUES ({placeholders})", values)
 
 
+def _role_hints(role: str) -> str:
+    return (
+        f"SELECT r.{role}_prompt_id, tx.text_value "
+        "FROM execution_run_contexts r "
+        "JOIN execution_contexts c ON c.context_id = r.context_id "
+        f"JOIN execution_prompt_snapshots s ON s.snapshot_id = c.{role}_snapshot_id "
+        "JOIN execution_context_texts tx ON tx.text_id = s.raw_text_id "
+        f"WHERE r.{role}_prompt_id IS NOT NULL AND r.reference_epoch = ("
+        "SELECT storage_epoch FROM execution_history_metadata WHERE singleton_id = 1)"
+    )
+
+
+# Set form of the read-time prompt-hint rule in get_execution_contexts():
+# a stored role hint is live only in the current storage epoch, and only when
+# its exact raw snapshot text matches exactly one prompt, which is the hinted
+# one (_prompt_hint). Evaluated once per statement, never per prompt row.
+# Defines the CTE ``live_hint_prompts(prompt_id)``; prefix it with WITH.
+LIVE_HINT_CTES = (
+    "stored_hints(prompt_id, raw) AS ("
+    + _role_hints("positive")
+    + " UNION "
+    + _role_hints("negative")
+    + "), unique_hint_texts(prompt_id, raw) AS ("
+    "SELECT MIN(id), text COLLATE BINARY FROM prompts "
+    "WHERE text COLLATE BINARY IN (SELECT raw FROM stored_hints) "
+    "GROUP BY text COLLATE BINARY HAVING COUNT(*) = 1"
+    "), live_hint_prompts(prompt_id) AS ("
+    "SELECT h.prompt_id FROM stored_hints h JOIN unique_hint_texts u "
+    "ON u.prompt_id = h.prompt_id AND u.raw = h.raw COLLATE BINARY)"
+)
+LIVE_HINT_PROMPT_IDS_SQL = (
+    "WITH " + LIVE_HINT_CTES + " SELECT DISTINCT prompt_id FROM live_hint_prompts"
+)
+
+
 def _prompt_hint(conn: sqlite3.Connection, raw: str | None) -> int | None:
     if raw is None:
         return None

database/operations.py

@@ -13,7 +13,7 @@ from .pairs import PairDependencyError, PairOperations
 from .library import LibraryOperations
 from .library_commands import LibraryCommandOperations
 from .execution_history import ExecutionHistoryOperations
-from .execution_contexts import ExecutionContextOperations
+from .execution_contexts import ExecutionContextOperations, LIVE_HINT_CTES
 from .execution_outputs import ExecutionOutputOperations
 from .history_reads import HistoryReadOperations
 
@@ -156,6 +156,28 @@ def _coerce_file_info(metadata: Any) -> tuple:
     )
 
 
+# Exempt references: images, pairs, the __protected__ tag, Library parent
+# state, and live recorded-run hints (computed once, see LIVE_HINT_CTES).
+ORPHANED_PROMPTS_SQL = "WITH " + LIVE_HINT_CTES + """
+SELECT p.id FROM prompts p
+LEFT JOIN generated_images gi ON p.id = gi.prompt_id
+WHERE gi.prompt_id IS NULL
+AND NOT EXISTS (
+    SELECT 1 FROM prompt_pairs pp
+    WHERE pp.positive_prompt_id = p.id OR pp.negative_prompt_id = p.id
+)
+AND NOT EXISTS (
+    SELECT 1 FROM prompt_tags pt
+    JOIN tags t ON pt.tag_id = t.id
+    WHERE pt.prompt_id = p.id AND t.name = '__protected__'
+)
+AND NOT EXISTS (
+    SELECT 1 FROM prompt_library_state ls WHERE ls.prompt_id = p.id
+)
+AND p.id NOT IN (SELECT prompt_id FROM live_hint_prompts)
+"""
+
+
 class PromptDatabase(
     PairOperations,
     LibraryOperations,
@@ -2162,25 +2184,18 @@ class PromptDatabase(
             return [dict(row) for row in cursor.fetchall()]
 
     def prune_orphaned_prompts(self) -> int:
-        """Delete prompts with no linked images and no __protected__ tag.
+        """Delete prompts nothing references.
+
+        A prompt is kept when it has a linked image, belongs to a pair, carries
+        the ``__protected__`` tag, has Library parent state (a titled or
+        explicit parent), or is the currently valid navigation hint of a
+        recorded run: same storage epoch and exact raw snapshot text matching
+        only that prompt, the conditions under which history reads resolve it.
 
         Returns number of prompts removed.
         """
         with self._pair_transaction() as conn:
-            cursor = conn.execute("""
+            cursor = conn.execute(ORPHANED_PROMPTS_SQL)
-                SELECT p.id FROM prompts p
-                LEFT JOIN generated_images gi ON p.id = gi.prompt_id
-                WHERE gi.prompt_id IS NULL
-                AND NOT EXISTS (
-                    SELECT 1 FROM prompt_pairs pp
-                    WHERE pp.positive_prompt_id = p.id OR pp.negative_prompt_id = p.id
-                )
-                AND NOT EXISTS (
-                    SELECT 1 FROM prompt_tags pt
-                    JOIN tags t ON pt.tag_id = t.id
-                    WHERE pt.prompt_id = p.id AND t.name = '__protected__'
-                )
-            """)
             orphaned = [row["id"] for row in cursor.fetchall()]
             if not orphaned:
                 return 0

tests/test_database.py

@@ -25,7 +25,11 @@ try:
 except ImportError:  # discovered with tests/ as the top-level directory
     from open_handles import assert_closed  # noqa: E402
 
-from database.operations import PromptDatabase, _resolve_db_path
+from database.operations import (
+    ORPHANED_PROMPTS_SQL,
+    PromptDatabase,
+    _resolve_db_path,
+)
 from utils.hashing import generate_prompt_hash
 
 
@@ -1094,6 +1098,50 @@ class TestMaintenanceOperations(DatabaseTestCase):
         self.assertIsNotNone(self.db.get_prompt_by_id(illustrated))
         self.assertEqual(self.db.prune_orphaned_prompts(), 0)
 
+    def test_prune_orphaned_prompts_spares_library_parents(self):
+        titled = self._save("titled parent")
+        untitled = self._save("untitled parent")
+        orphan = self._save("orphan")
+        with self.db.model.get_connection() as conn:
+            conn.execute(
+                "INSERT INTO prompt_library_state(prompt_id,title) "
+                "VALUES(?,?),(?,NULL)",
+                (titled, "My recipe", untitled),
+            )
+            conn.commit()
+
+        self.assertEqual(self.db.prune_orphaned_prompts(), 1)
+        self.assertIsNone(self.db.get_prompt_by_id(orphan))
+        self.assertIsNotNone(self.db.get_prompt_by_id(titled))
+        self.assertIsNotNone(self.db.get_prompt_by_id(untitled))
+        with self.db.model.get_connection() as conn:
+            titles = conn.execute(
+                "SELECT prompt_id,title FROM prompt_library_state ORDER BY prompt_id"
+            ).fetchall()
+        self.assertEqual(
+            [tuple(row) for row in titles], [(titled, "My recipe"), (untitled, None)]
+        )
+        self.assertEqual(self.db.prune_orphaned_prompts(), 0)
+
+    def test_prune_query_has_no_scan_inside_correlated_subquery(self):
+        conn = self.db.model.get_connection()
+        plan = conn.execute("EXPLAIN QUERY PLAN " + ORPHANED_PROMPTS_SQL).fetchall()
+        parents = {row[0]: row[1] for row in plan}
+        details = {row[0]: row[3] for row in plan}
+
+        def correlated(node):
+            while node in parents:
+                node = parents[node]
+                if details.get(node, "").startswith("CORRELATED"):
+                    return True
+            return False
+
+        scans = [
+            row[3] for row in plan if row[3].startswith("SCAN ") and correlated(row[0])
+        ]
+        self.assertEqual(scans, [])
+        self.assertIn("execution_run_contexts", ORPHANED_PROMPTS_SQL)
+
     def test_consistency_check_reports_orphaned_rows(self):
         self.assertEqual(self.db.check_consistency(), [])
         pid = self._save("to orphan", tags=["t"])

tests/test_execution_context_database.py

@@ -13,6 +13,8 @@ import unittest
 
 from database.execution_history import HistoryEpochChanged
 from database.execution_context_schema import context_schema_error
+from database.execution_contexts import LIVE_HINT_PROMPT_IDS_SQL
+from database.execution_history_schema import rotate_history_epoch
 from utils.execution_contexts import (
     ContextBundle,
     PromptSnapshot,
@@ -365,6 +367,101 @@ class ContextDatabaseTests(unittest.TestCase):
         self.assertEqual(conn.execute("SELECT COUNT(*) FROM prompts").fetchone()[0], 0)
         self.assertIsNone(context_schema_error(conn))
 
+    def test_prune_keeps_prompts_referenced_by_recorded_runs(self):
+        db = self.database()
+        conn = db.model.get_connection()
+        with conn:
+            conn.execute(
+                "INSERT INTO prompts(id,text) VALUES(1,'positive'),(2,'negative'),"
+                "(3,'unreferenced')"
+            )
+        start = self.capture(db)
+        original = db.get_execution_contexts(start.run_id)["contexts"][0]
+        self.assertEqual(
+            (original["positive_prompt_id"], original["negative_prompt_id"]), (1, 2)
+        )
+        self.assertEqual(db.prune_orphaned_prompts(), 1)
+        self.assertEqual(
+            [row[0] for row in conn.execute("SELECT id FROM prompts ORDER BY id")],
+            [1, 2],
+        )
+        self.assertEqual(
+            db.get_execution_contexts(start.run_id)["contexts"][0], original
+        )
+        self.assertEqual(db.prune_orphaned_prompts(), 0)
+        db.update_prompt_text(2, "edited")
+        self.assertEqual(db.prune_orphaned_prompts(), 1)
+        self.assertEqual(
+            [row[0] for row in conn.execute("SELECT id FROM prompts")], [1]
+        )
+        self.assertIsNone(context_schema_error(conn))
+
+    def test_prune_ignores_hints_made_ambiguous_by_identical_text(self):
+        db = self.database()
+        conn = db.model.get_connection()
+        with conn:
+            conn.execute(
+                "INSERT INTO prompts(id,text) VALUES(1,'positive'),(2,'negative')"
+            )
+        start = self.capture(db)
+        context = db.get_execution_contexts(start.run_id)["contexts"][0]
+        self.assertEqual(context["negative_prompt_id"], 2)
+        with conn:
+            conn.execute("INSERT INTO prompts(id,text) VALUES(3,'negative')")
+        db.link_image_to_prompt(3, "/out/three.png")
+        context = db.get_execution_contexts(start.run_id)["contexts"][0]
+        self.assertEqual(
+            (context["positive_prompt_id"], context["negative_prompt_id"]), (1, None)
+        )
+        self.assertEqual(db.prune_orphaned_prompts(), 1)
+        self.assertEqual(
+            [row[0] for row in conn.execute("SELECT id FROM prompts ORDER BY id")],
+            [1, 3],
+        )
+        after = db.get_execution_contexts(start.run_id)["contexts"][0]
+        self.assertEqual(
+            (after["positive_prompt_id"], after["negative_prompt_id"]), (1, None)
+        )
+        self.assertEqual(db.prune_orphaned_prompts(), 0)
+
+    def test_live_hint_set_matches_reader_hints(self):
+        db = self.database()
+        conn = db.model.get_connection()
+        with conn:
+            conn.execute(
+                "INSERT INTO prompts(id,text) VALUES(1,'positive'),(2,'negative'),"
+                "(4,'other')"
+            )
+        runs = [self.capture(db, 0)]
+        runs.append(self.capture(db, 1, bundle(positive="other", negative="none")))
+
+        def assert_same():
+            reader = {
+                context[key]
+                for run in runs
+                for context in db.get_execution_contexts(run.run_id)["contexts"]
+                for key in ("positive_prompt_id", "negative_prompt_id")
+                if context[key] is not None
+            }
+            live = {row[0] for row in conn.execute(LIVE_HINT_PROMPT_IDS_SQL)}
+            self.assertEqual(live, reader)
+            return live
+
+        self.assertEqual(assert_same(), {1, 2, 4})
+        with conn:
+            conn.execute("INSERT INTO prompts(id,text) VALUES(3,'negative')")
+        self.assertEqual(assert_same(), {1, 4})
+        with conn:
+            conn.execute("DELETE FROM prompts WHERE id=3")
+        self.assertEqual(assert_same(), {1, 2, 4})
+        db.update_prompt_text(4, "Other")
+        self.assertEqual(assert_same(), {1, 2})
+        with conn:
+            conn.execute("INSERT INTO prompts(id,text) VALUES(5,'none')")
+        self.assertEqual(assert_same(), {1, 2})
+        rotate_history_epoch(conn)
+        self.assertEqual(assert_same(), set())
+
     def test_hints_require_exact_unique_raw_text_and_never_refresh(self):
         db = self.database()
         conn = db.model.get_connection()