nodes/ComfyAssets-ComfyUI_PromptManager
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix: image uniqueness by full path, escaped LIKE search, no usage stamp on save

Browse Source 
generated_images was UNIQUE(prompt_id, filename) on the basename, so
ComfyUI's same-named files in different date folders were silently
ignored by INSERT OR IGNORE, and the lastrowid == 0 branch after it
never ran. The table now carries file_path (normcase/normpath/abspath
of the image path) with UNIQUE(prompt_id, file_path); the migration
collapses rows that normalise to one file and supersedes the old
basename migration, which would otherwise undo it on every start.

search_prompts and the tag search escaped nothing, so 100% matched
'100 percent' and a_b matched aXb; user text now goes through
escape_like with ESCAPE. save_prompt no longer stamps last_used_at, the
usage backfill runs only on the upgrade that adds the columns, and the
Recently Used / Most Used orders put never-run prompts (NULL) after run
ones instead of first.
This commit is contained in:
Vito Sansevero
2026-10-08 18:01:34 -07:00
parent 9234dac211
commit ae11e81203
5 changed files with 310 additions and 132 deletions
Download Patch File Download Diff File Expand all files Collapse all files
database/models.py
+101 -103
View File
@@ -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.
database/operations.py
+47 -22
View File
@@ -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
tests/test_database.py
+66
View File
@@ -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."""
tests/test_database_migrations.py
+65
View File
@@ -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:
tests/test_prompt_usage.py
+31 -7
View File
@@ -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)