diff --git a/database/models.py b/database/models.py index 3a25ea4..461c418 100644 --- a/database/models.py +++ b/database/models.py @@ -34,6 +34,16 @@ def utc_now_iso() -> str: ) +def normalize_image_path(path: str) -> str: + """Canonical form of an image path for generated_images.file_path. + + Absolute, with '..' and '.' collapsed and the platform's case folding + applied, so two spellings of one file (or the same basename in two date + folders) compare the way the filesystem would. + """ + return os.path.normcase(os.path.normpath(os.path.abspath(path))) + + SQLITE_HEADER = b"SQLite format 3\x00" REQUIRED_PROMPT_COLUMNS = ("id", "text", "created_at") @@ -173,8 +183,9 @@ class PromptModel: workflow_data TEXT, prompt_metadata TEXT, parameters TEXT, + file_path TEXT, FOREIGN KEY (prompt_id) REFERENCES prompts(id) ON DELETE CASCADE, - UNIQUE(prompt_id, filename) + UNIQUE(prompt_id, file_path) ) """) @@ -195,9 +206,6 @@ class PromptModel: ) """) - # Add unique constraint to existing databases (migration) - self._migrate_add_unique_constraint(conn) - # Check if we need to migrate from old schema with workflow_name self._migrate_workflow_name_removal(conn) @@ -207,9 +215,12 @@ class PromptModel: # Migrate JSON tags to normalized junction tables self._migrate_json_tags_to_junction(conn) - # Usage tracking columns (last, after any migration that rebuilds prompts) + # Usage tracking columns (after any migration that rebuilds prompts) self._migrate_add_usage_columns(conn) + # Image uniqueness by full path instead of basename (3.2.4) + self._migrate_image_uniqueness_by_path(conn) + def _create_indexes(self, conn: sqlite3.Connection) -> None: """ Create indexes for better query performance. @@ -233,6 +244,7 @@ class PromptModel: "CREATE INDEX IF NOT EXISTS idx_prompts_run_count ON prompts(run_count)", "CREATE INDEX IF NOT EXISTS idx_prompt_images ON generated_images(prompt_id)", "CREATE INDEX IF NOT EXISTS idx_image_path ON generated_images(image_path)", + "CREATE INDEX IF NOT EXISTS idx_image_file_path ON generated_images(file_path)", "CREATE INDEX IF NOT EXISTS idx_generation_time ON generated_images(generation_time)", "CREATE INDEX IF NOT EXISTS idx_prompt_tags_tag ON prompt_tags(tag_id)", "CREATE INDEX IF NOT EXISTS idx_tags_name ON tags(name)", @@ -412,100 +424,6 @@ class PromptModel: self.logger.error(f"Migration error (foreign key types): {e}") raise - def _migrate_add_unique_constraint(self, conn: sqlite3.Connection) -> None: - """ - Add UNIQUE constraint on (prompt_id, filename) to prevent duplicate image entries. - - This migration: - 1. Checks if the constraint already exists - 2. Removes duplicate entries (keeping the most recent) - 3. Recreates the table with the UNIQUE constraint - - Args: - conn: Active database connection - - Raises: - sqlite3.Error: If the rebuild fails (logged, nothing half-applied) - """ - try: - cursor = conn.execute("PRAGMA index_list(generated_images)") - indexes = cursor.fetchall() - - has_unique_constraint = False - for idx in indexes: - if idx[2] == 1: # unique flag - cursor = conn.execute(f"PRAGMA index_info({idx[1]})") - columns = [col[2] for col in cursor.fetchall()] - if "prompt_id" in columns and "filename" in columns: - has_unique_constraint = True - break - - if has_unique_constraint: - return # Already migrated - - self.logger.info( - "Migrating database: adding UNIQUE constraint on (prompt_id, filename)" - ) - with self._table_rebuild(conn): - # Remove duplicates keeping only the most recent (highest id) - cursor = conn.execute(""" - DELETE FROM generated_images - WHERE id NOT IN ( - SELECT MAX(id) FROM generated_images - GROUP BY prompt_id, filename - ) - """) - if cursor.rowcount > 0: - self.logger.info( - f"Removed {cursor.rowcount} duplicate image entries" - ) - - conn.execute("DROP TABLE IF EXISTS generated_images_new") - conn.execute(""" - CREATE TABLE generated_images_new ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - prompt_id INTEGER NOT NULL, - image_path TEXT NOT NULL, - filename TEXT NOT NULL, - generation_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP, - file_size INTEGER, - width INTEGER, - height INTEGER, - format TEXT, - workflow_data TEXT, - prompt_metadata TEXT, - parameters TEXT, - FOREIGN KEY (prompt_id) REFERENCES prompts(id) ON DELETE CASCADE, - UNIQUE(prompt_id, filename) - ) - """) - conn.execute(""" - INSERT INTO generated_images_new - (id, prompt_id, image_path, filename, generation_time, file_size, - width, height, format, workflow_data, prompt_metadata, parameters) - SELECT id, prompt_id, image_path, filename, generation_time, file_size, - width, height, format, workflow_data, prompt_metadata, parameters - FROM generated_images - """) - conn.execute("DROP TABLE generated_images") - conn.execute( - "ALTER TABLE generated_images_new RENAME TO generated_images" - ) - conn.execute( - "CREATE INDEX IF NOT EXISTS idx_prompt_images ON generated_images(prompt_id)" - ) - conn.execute( - "CREATE INDEX IF NOT EXISTS idx_image_path ON generated_images(image_path)" - ) - conn.execute( - "CREATE INDEX IF NOT EXISTS idx_generation_time ON generated_images(generation_time)" - ) - self.logger.info("UNIQUE constraint migration completed successfully") - - except sqlite3.Error as e: - self.logger.error(f"Migration error (image UNIQUE constraint): {e}") - raise - def _migrate_json_tags_to_junction(self, conn: sqlite3.Connection) -> None: """ Populate tags and prompt_tags tables from legacy JSON tags column. @@ -558,11 +476,12 @@ class PromptModel: def _migrate_add_usage_columns(self, conn: sqlite3.Connection) -> None: """ - Add last_used_at and run_count (3.2.4) and fill in any prompt without usage data. + Add last_used_at and run_count (3.2.4) and estimate usage for existing rows. - The backfill runs on every start but only touches rows whose last_used_at - is NULL, so it is a no-op once healed. That also repairs rows written by an - older version after a downgrade, or left unfilled by an interrupted upgrade. + The backfill runs only when the columns are added, i.e. on the upgrade + from a version without usage tracking. Afterwards a NULL last_used_at + means "never run": saving a prompt does not stamp it, only + record_prompt_use does, and "Recently Used" relies on that. The estimate comes from linked images: run_count is the image count (at least 1) and last_used_at the newest image time, falling back to created_at. @@ -571,14 +490,19 @@ class PromptModel: """ try: columns = {row[1] for row in conn.execute("PRAGMA table_info(prompts)")} + added = False if "last_used_at" not in columns: self.logger.info("Adding prompt usage column last_used_at") conn.execute("ALTER TABLE prompts ADD COLUMN last_used_at TIMESTAMP") + added = True if "run_count" not in columns: self.logger.info("Adding prompt usage column run_count") conn.execute( "ALTER TABLE prompts ADD COLUMN run_count INTEGER NOT NULL DEFAULT 0" ) + added = True + if not added: + return cursor = conn.execute(f""" UPDATE prompts SET @@ -601,6 +525,80 @@ class PromptModel: self.logger.error(f"Migration error (usage columns): {e}") raise + def _migrate_image_uniqueness_by_path(self, conn: sqlite3.Connection) -> None: + """ + Rebuild generated_images so an image is unique per (prompt_id, file_path). + + Older schemas were unique per (prompt_id, filename), i.e. the basename, + so ComfyUI's same-named files in different date folders were silently + dropped. file_path holds normalize_image_path(image_path); rows whose + paths normalise to the same file collapse to the newest one. + + Raises: + sqlite3.Error: If the rebuild fails (logged, nothing half-applied) + """ + try: + columns = { + row[1] for row in conn.execute("PRAGMA table_info(generated_images)") + } + if "file_path" in columns: + return + + self.logger.info("Migrating database: image uniqueness by full path") + with self._table_rebuild(conn): + rows = conn.execute( + "SELECT id, prompt_id, image_path FROM generated_images ORDER BY id" + ).fetchall() + keep = {} + for row_id, prompt_id, image_path in rows: + keep[(prompt_id, normalize_image_path(image_path))] = row_id + + conn.execute("DROP TABLE IF EXISTS generated_images_new") + conn.execute(""" + CREATE TABLE generated_images_new ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + prompt_id INTEGER NOT NULL, + image_path TEXT NOT NULL, + filename TEXT NOT NULL, + generation_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + file_size INTEGER, + width INTEGER, + height INTEGER, + format TEXT, + workflow_data TEXT, + prompt_metadata TEXT, + parameters TEXT, + file_path TEXT, + FOREIGN KEY (prompt_id) REFERENCES prompts(id) ON DELETE CASCADE, + UNIQUE(prompt_id, file_path) + ) + """) + conn.executemany( + """ + INSERT INTO generated_images_new + (id, prompt_id, image_path, filename, generation_time, file_size, + width, height, format, workflow_data, prompt_metadata, parameters, + file_path) + SELECT id, prompt_id, image_path, filename, generation_time, file_size, + width, height, format, workflow_data, prompt_metadata, parameters, + ? + FROM generated_images WHERE id = ? + """, + [(file_path, row_id) for (_, file_path), row_id in keep.items()], + ) + conn.execute("DROP TABLE generated_images") + conn.execute( + "ALTER TABLE generated_images_new RENAME TO generated_images" + ) + dropped = len(rows) - len(keep) + if dropped: + self.logger.info(f"Collapsed {dropped} duplicate image rows") + self.logger.info("Image path uniqueness migration completed") + + except sqlite3.Error as e: + self.logger.error(f"Migration error (image path uniqueness): {e}") + raise + def migrate_database(self) -> None: """ Apply any pending database migrations. diff --git a/database/operations.py b/database/operations.py index 918e2a9..e1ad96d 100644 --- a/database/operations.py +++ b/database/operations.py @@ -8,7 +8,7 @@ import datetime import os from typing import Optional, List, Dict, Any, Union -from .models import PromptModel, utc_now_iso +from .models import PromptModel, normalize_image_path, utc_now_iso # Import logging system try: @@ -30,11 +30,15 @@ TAG_SUBQUERY = ( # Server-side sort orders, keyed by the value the UI sends. Only these strings ever # reach ORDER BY; unknown keys fall back to DEFAULT_SORT. +# last_used_at is NULL until the prompt has run, so "IS NULL" first keeps +# never-run prompts after every run one (SQLite sorts NULL first in DESC). SORT_ORDERS = { - "last_used_desc": "last_used_at DESC, id DESC", + "last_used_desc": "last_used_at IS NULL, last_used_at DESC, id DESC", "created_desc": "created_at DESC, id DESC", "created_asc": "created_at ASC, id ASC", - "run_count_desc": "run_count DESC, last_used_at DESC, id DESC", + "run_count_desc": ( + "run_count DESC, last_used_at IS NULL, last_used_at DESC, id DESC" + ), "rating_desc": "rating IS NULL, rating DESC, id DESC", "rating_asc": "rating IS NULL, rating ASC, id DESC", "text_asc": "text COLLATE NOCASE ASC, id ASC", @@ -48,6 +52,18 @@ def order_by_clause(sort: Optional[str]) -> str: return SORT_ORDERS.get(sort or DEFAULT_SORT, SORT_ORDERS[DEFAULT_SORT]) +LIKE_ESCAPE = "\\" + + +def escape_like(term: str) -> str: + """Escape LIKE wildcards in user text; use with ``LIKE ? ESCAPE '\\'``.""" + return ( + term.replace(LIKE_ESCAPE, LIKE_ESCAPE + LIKE_ESCAPE) + .replace("%", LIKE_ESCAPE + "%") + .replace("_", LIKE_ESCAPE + "_") + ) + + def _resolve_db_path(db_path: Optional[str] = None) -> str: """Resolve the database path from config, falling back to defaults. @@ -147,8 +163,8 @@ class PromptDatabase: """ INSERT INTO prompts ( text, category, tags, rating, notes, hash, created_at, - updated_at, last_used_at - ) VALUES (?, ?, NULL, ?, ?, ?, ?, ?, ?) + updated_at + ) VALUES (?, ?, NULL, ?, ?, ?, ?, ?) """, ( text.strip(), @@ -158,7 +174,6 @@ class PromptDatabase: prompt_hash, datetime.datetime.now(datetime.timezone.utc).isoformat(), datetime.datetime.now(datetime.timezone.utc).isoformat(), - utc_now_iso(), ), ) prompt_id = cursor.lastrowid @@ -274,8 +289,8 @@ class PromptDatabase: params = [] if text: - query_parts.append("AND text LIKE ?") - params.append(f"%{text}%") + query_parts.append("AND text LIKE ? ESCAPE '\\'") + params.append(f"%{escape_like(text)}%") if category: query_parts.append("AND category = ?") @@ -287,9 +302,10 @@ class PromptDatabase: query_parts.append( "AND prompts.id IN (" " SELECT pt.prompt_id FROM prompt_tags pt" - " JOIN tags t ON pt.tag_id = t.id WHERE t.name LIKE ?)" + " JOIN tags t ON pt.tag_id = t.id" + " WHERE t.name LIKE ? ESCAPE '\\')" ) - params.append(f"%{tag}%") + params.append(f"%{escape_like(tag)}%") else: query_parts.append( "AND prompts.id IN (" @@ -659,8 +675,8 @@ class PromptDatabase: search_clause = "" params: list = [] if search: - search_clause = "HAVING t.name LIKE ?" - params.append(f"%{search}%") + search_clause = "HAVING t.name LIKE ? ESCAPE '\\'" + params.append(f"%{escape_like(search)}%") sort_map = { "alpha_desc": "tag COLLATE NOCASE DESC", @@ -1370,20 +1386,23 @@ class PromptDatabase: # Proceed with linking filename = os.path.basename(image_path) + file_path = normalize_image_path(image_path) file_info = metadata.get("file_info", {}) if metadata else {} - # Use INSERT OR IGNORE to skip duplicates (same prompt_id + filename) + # INSERT OR IGNORE skips a file already linked to this prompt + # (same prompt_id + normalised full path) cursor = conn.execute( """ INSERT OR IGNORE INTO generated_images - (prompt_id, image_path, filename, file_size, width, height, format, - workflow_data, prompt_metadata, parameters) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + (prompt_id, image_path, filename, file_path, file_size, width, + height, format, workflow_data, prompt_metadata, parameters) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( prompt_id_int, image_path, filename, + file_path, file_info.get("size"), ( file_info.get("dimensions", [None, None])[0] @@ -1403,9 +1422,9 @@ class PromptDatabase: ) conn.commit() - if cursor.lastrowid == 0: + if cursor.rowcount == 0: self.logger.debug( - f"Image {filename} already linked to prompt {prompt_id_int}" + f"Image {file_path} already linked to prompt {prompt_id_int}" ) return 0 @@ -1913,8 +1932,13 @@ class PromptDatabase: gi.workflow_data, gi.prompt_metadata, gi.generation_time FROM generated_images gi JOIN prompts p ON gi.prompt_id = p.id - WHERE gi.image_path = ? OR gi.image_path LIKE ?""", - (image_path, f"%{os.path.basename(image_path)}"), + WHERE gi.image_path = ? OR gi.file_path = ? + OR gi.image_path LIKE ? ESCAPE '\\'""", + ( + image_path, + normalize_image_path(image_path), + f"%{escape_like(os.path.basename(image_path))}", + ), ) row = cursor.fetchone() if not row: @@ -1944,8 +1968,9 @@ class PromptDatabase: """Return the prompt_id linked to an image, or None.""" with self.model.get_connection() as conn: cursor = conn.execute( - "SELECT prompt_id FROM generated_images WHERE image_path = ?", - (image_path,), + "SELECT prompt_id FROM generated_images" + " WHERE image_path = ? OR file_path = ?", + (image_path, normalize_image_path(image_path)), ) row = cursor.fetchone() return row[0] if row else None diff --git a/tests/test_database.py b/tests/test_database.py index b03daff..6e2d994 100644 --- a/tests/test_database.py +++ b/tests/test_database.py @@ -455,6 +455,72 @@ class TestImageOperations(DatabaseTestCase): self.assertEqual(len(images), 0) +class TestImageUniquenessByPath(DatabaseTestCase): + """ComfyUI writes same-named files into different date folders.""" + + def _link(self, prompt_id, path): + return self.db.link_image_to_prompt(prompt_id=prompt_id, image_path=path) + + def test_same_filename_in_different_folders_both_link(self): + pid = self._save("two folders") + first = self._link(pid, os.path.join("out", "2026-01-01", "a.png")) + second = self._link(pid, os.path.join("out", "2026-01-02", "a.png")) + + self.assertTrue(first) + self.assertTrue(second) + self.assertNotEqual(first, second) + self.assertEqual(len(self.db.get_prompt_images(pid)), 2) + + def test_same_path_spelled_differently_links_once(self): + pid = self._save("one file") + first = self._link(pid, os.path.join("out", "x", "..", "a.png")) + second = self._link(pid, os.path.join(".", "out", "a.png")) + + self.assertTrue(first) + self.assertEqual(second, 0) + images = self.db.get_prompt_images(pid) + self.assertEqual(len(images), 1) + expected = os.path.normcase( + os.path.normpath(os.path.abspath(os.path.join("out", "a.png"))) + ) + self.assertEqual(images[0]["file_path"], expected) + + def test_same_file_can_link_to_two_prompts(self): + a = self._save("prompt a") + b = self._save("prompt b") + path = os.path.join("out", "shared.png") + self.assertTrue(self._link(a, path)) + self.assertTrue(self._link(b, path)) + + +class TestSearchEscapesLikeWildcards(DatabaseTestCase): + def test_percent_is_literal(self): + self._save("100% sure") + self._save("100 percent sure") + texts = [p["text"] for p in self.db.search_prompts(text="100%")] + self.assertEqual(texts, ["100% sure"]) + + def test_underscore_is_literal(self): + self._save("snake_case name") + self._save("snakeXcase name") + texts = [p["text"] for p in self.db.search_prompts(text="snake_case")] + self.assertEqual(texts, ["snake_case name"]) + + def test_backslash_is_literal(self): + self._save("path C:\\out\\img") + self._save("path C:out img") + texts = [p["text"] for p in self.db.search_prompts(text="C:\\out")] + self.assertEqual(texts, ["path C:\\out\\img"]) + + def test_partial_tag_filter_is_literal(self): + self._save("tagged a", tags=["50%_off"]) + self._save("tagged b", tags=["50x_off"]) + texts = [ + p["text"] for p in self.db.search_prompts(tags=["50%"], tag_partial=True) + ] + self.assertEqual(texts, ["tagged a"]) + + class TestEdgeCases(DatabaseTestCase): """Test edge cases and boundary conditions.""" diff --git a/tests/test_database_migrations.py b/tests/test_database_migrations.py index 57286a6..ae6e164 100644 --- a/tests/test_database_migrations.py +++ b/tests/test_database_migrations.py @@ -194,6 +194,71 @@ class TestIdempotentMigrations(MigrationTestCase): db = self._open() self.assertEqual(db.get_prompt_by_id(1)["text"], "a") + def test_image_table_is_rebuilt_with_uniqueness_by_full_path(self): + self._legacy_db( + [ + "INSERT INTO prompts (text, hash, workflow_name) VALUES ('a', 'h1', 'wf')", + # same basename in two folders: both must survive + "INSERT INTO generated_images (prompt_id, image_path, filename)" + " VALUES (1, '/out/2026-01-01/a.png', 'a.png')", + "INSERT INTO generated_images (prompt_id, image_path, filename)" + " VALUES (1, '/out/2026-01-02/a.png', 'a.png')", + # two spellings of one file: only the newest row survives + "INSERT INTO generated_images (prompt_id, image_path, filename)" + " VALUES (1, '/out/x/../b.png', 'b.png')", + "INSERT INTO generated_images (prompt_id, image_path, filename)" + " VALUES (1, '/out/b.png', 'b.png')", + ] + ) + + db = self._open() + + images = db.get_prompt_images(1) + self.assertEqual(len(images), 3) + self.assertIn("file_path", self._columns("generated_images")) + self.assertEqual( + sorted(i["file_path"] for i in images), + sorted( + os.path.normcase(os.path.normpath(p)) + for p in ( + "/out/2026-01-01/a.png", + "/out/2026-01-02/a.png", + "/out/b.png", + ) + ), + ) + self.assertEqual([i["id"] for i in images if i["filename"] == "b.png"], [4]) + self.assertEqual( + self._unique_columns("generated_images"), ["file_path", "prompt_id"] + ) + + def test_image_uniqueness_migration_is_not_undone_on_the_next_start(self): + self._legacy_db( + ["INSERT INTO prompts (text, hash, workflow_name) VALUES ('a', 'h1', 'wf')"] + ) + self._open().close() + PromptModel.reset_schema_cache() + db = self._open() + self.assertEqual( + self._unique_columns("generated_images"), ["file_path", "prompt_id"] + ) + self.assertTrue(db.link_image_to_prompt(1, "/out/1/a.png")) + self.assertTrue(db.link_image_to_prompt(1, "/out/2/a.png")) + self.assertEqual(len(db.get_prompt_images(1)), 2) + + def _unique_columns(self, table): + conn = sqlite3.connect(self.path) + try: + columns = [] + for idx in conn.execute(f"PRAGMA index_list({table})"): + if idx[2] == 1: + columns.extend( + c[2] for c in conn.execute(f"PRAGMA index_info({idx[1]})") + ) + return sorted(columns) + finally: + conn.close() + def _table_names(self): conn = sqlite3.connect(self.path) try: diff --git a/tests/test_prompt_usage.py b/tests/test_prompt_usage.py index 08da96c..b7f29c0 100644 --- a/tests/test_prompt_usage.py +++ b/tests/test_prompt_usage.py @@ -44,10 +44,32 @@ class UsageTestCase(unittest.TestCase): class TestRecordPromptUse(UsageTestCase): - def test_new_prompt_starts_unused_but_timestamped(self): + def test_new_prompt_starts_unused_and_unstamped(self): + # Saving is not using: only record_prompt_use stamps last_used_at prompt = self.db.get_prompt_by_id(self._save("fresh")) self.assertEqual(prompt["run_count"], 0) - self.assertTrue(prompt["last_used_at"]) + self.assertIsNone(prompt["last_used_at"]) + + def test_never_run_prompts_sort_after_run_ones(self): + ran = self._save("run once") + later = self._save("saved later, never run") + self.db.record_prompt_use(ran) + last = self._save("saved last, never run") + + result = self.db.get_recent_prompts(limit=10, sort="last_used_desc") + # The run prompt leads; never-run prompts follow, newest first + self.assertEqual(self._ids(result["prompts"]), [ran, last, later]) + + searched = self.db.search_prompts(text="run", sort="last_used_desc") + self.assertEqual(self._ids(searched)[0], ran) + + def test_most_used_sort_breaks_ties_with_unstamped_last(self): + a = self._save("a") + b = self._save("b") + self._set(a, run_count=2, last_used_at="2026-01-01T00:00:00.000+00:00") + self._set(b, run_count=2) # same count, never stamped + result = self.db.get_recent_prompts(limit=10, sort="run_count_desc") + self.assertEqual(self._ids(result["prompts"]), [a, b]) def test_record_use_increments_and_touches_last_used(self): pid = self._save("used") @@ -182,9 +204,11 @@ class TestUsageMigration(unittest.TestCase): self.assertEqual(unused["run_count"], 1) self.assertTrue(unused["last_used_at"].startswith("2026-01-01T10:00:00")) - def test_rows_written_while_downgraded_are_healed_on_next_start(self): + def test_backfill_runs_only_when_the_columns_are_added(self): db = PromptDatabase(self.path) - # 3.2.3 does not know the columns: its inserts leave last_used_at NULL + # A row whose usage is unknown (e.g. written by 3.2.3 after a downgrade) + # stays "never run": a restart must not invent a run for it, otherwise + # Recently Used could never tell saved-but-unused prompts from run ones. with sqlite3.connect(self.path) as conn: conn.execute( "INSERT INTO prompts (text, hash, created_at)" @@ -196,9 +220,9 @@ class TestUsageMigration(unittest.TestCase): self.assertIsNone(db.get_prompt_by_id(legacy_id)["last_used_at"]) PromptModel.reset_schema_cache() - healed = PromptDatabase(self.path).get_prompt_by_id(legacy_id) - self.assertTrue(healed["last_used_at"].startswith("2026-04-01T00:00:00")) - self.assertEqual(healed["run_count"], 1) + later = PromptDatabase(self.path).get_prompt_by_id(legacy_id) + self.assertIsNone(later["last_used_at"]) + self.assertEqual(later["run_count"], 0) def test_migration_runs_once(self): db = PromptDatabase(self.path)