feat(maintenance): always back up before storing workflows once

Final review fixes I1, I2 and I3 for the workflow storage dedupe:

- I1: the migration always takes a verified backup, whatever
  maintenance.safety_backup says; the plan reports backup.policy
  "required" and the disk check counts it.
- I2: that backup is prompts.db.pre-workflow-storage_<stamp>, outside the
  pruned prompts.db.backup_* pool and kept until the user removes it; the
  job's last line and the preview name it, and the legacy restore accepts
  it. README's changelog warns that an older version opens and restores a
  rewritten database but shows no workflows, and says how to go back.
- I3: tools/workflow_text_digests.py writes the sha256 of every record's
  stored texts (read-only; inline before the rewrite, through the accessor
  after), so a diff before and after proves byte identity.
- A row the migration could not rewrite is listed in the job's errors and
  named in its last line, and Reclaim space still runs.

README.md

@@ -970,6 +970,11 @@ MIT License - see LICENSE file for details.
 
 ## Changelog
 
+### Unreleased (Workflows stored once)
+
+- **🗜️ Smaller database**: each distinct workflow and prompt graph is stored once, compressed, instead of again for every image. New images use it at once; the maintenance job "Store each workflow once" rewrites existing records and then reclaims the space. A 20,000-image test library went from 1.4 GB to 60 MB. The job always takes a verified backup first, `prompts.db.pre-workflow-storage_<date time>` beside the database, which is kept until you delete it. No image or video file is opened or changed.
+- **⚠️ Downgrading after the rewrite**: an older Prompt Manager version still opens and even restores a rewritten database, but shows its images without workflows or prompt graphs. To go back to an older version, stop ComfyUI, delete `prompts.db-wal` and `prompts.db-shm`, and copy `prompts.db.pre-workflow-storage_<date time>` over `prompts.db` (records added since that backup are lost), or stay on this version.
+
 ### v3.3.1 (Proxy Routing Fix)
 
 - **🌐 Proxy support**: the dashboard, gallery, tag page and metadata viewer now work when ComfyUI is reached through a proxy mount such as SwarmUI's `/ComfyBackendDirect/`. Scripts, images, API calls, progress streams and gallery folder filters stay under the mount, and the canvas node's buttons use ComfyUI's `fileURL()`. Direct ComfyUI URLs are unchanged. Thanks to @ghosttigerllc-bit ([#164](https://github.com/ComfyAssets/ComfyUI_PromptManager/pull/164))

TESTING.md

@@ -2721,10 +2721,25 @@ table definition, the id columns, their foreign keys and indexes, the stored val
 dangling ids (`TestSchemaErrors`); a pre-dedupe file is accepted and migrated on open
 and on restore. The `workflow_storage` maintenance plan counts the old-form records
 (one pass over record headers) and its apply starts "Store each workflow once"
-(`records`, cancellable): the verified safety backup, then batches of at most 200 rows or
-16 MB of text, each one `BEGIN IMMEDIATE` transaction found by a rowid range search and
-read one row at a time; each text is read back through the accessor before its row
-changes, and a mismatch or any error rolls the batch back whole. A crash or Cancel leaves
+(`records`, cancellable): first, always and whatever `maintenance.safety_backup` says,
+a verified pre-migration backup named `<db>.pre-workflow-storage_<YYYYmmdd_HHMMSS_ffffff>`,
+outside the pruned `<db>.backup_*` pool and kept until the user removes it (the preview
+reports `backup.policy: "required"`; `test_the_backup_is_taken_even_with_the_policy_off`,
+`test_the_backup_is_outside_the_pool_never_pruned_and_restores`); then batches of at most
+200 rows, 16 MB of text or 0.5 s, each one `BEGIN IMMEDIATE` transaction found by a rowid
+range search and read one row at a time, followed by 0.15 s with the write lock free so
+a writer on another connection gets in (`test_a_writer_on_another_connection_gets_in_between_batches`).
+Each text is read back through the accessor before its row changes. Each row has its own
+savepoint: a row that cannot be rewritten (undecodable text, a text that would not come
+back) stays inline, is listed in `errors` and named in the job's last line, and the rest
+go on, Reclaim space included (`TestRowErrors`); any other failure rolls the batch back
+whole. A reused document is decompressed and checked first and repaired from the text
+when damaged, and `verify_database_file` (restore and backup verification) reads every
+document and refuses a damaged one (`TestDamagedDocuments`). The legacy table rebuilds
+refuse a file that already has the document id columns (`TestRebuildGuards`).
+`tools/workflow_text_digests.py <db> <out>` writes the sha256 of every record's stored
+texts (read-only), so a `diff` of its output before and after the rewrite proves byte
+identity. A crash or Cancel leaves
 committed batches new form and the rest old form, and the next apply continues
 (`test_a_crash_mid_migration_leaves_every_row_readable_and_resumes`,
 `test_cancel_stops_after_the_current_batch_and_a_rerun_continues`). Values that are not
@@ -2735,8 +2750,9 @@ does not start and the finished migration names Optimize database. No image or v
 file is opened (`test_the_jobs_never_open_a_media_file` tracks every Python open under
 the output folder and compares bytes and mtimes). Memory stays bounded by one row plus
 the batch's distinct documents (`test_memory_stays_bounded_by_one_row`, tracemalloc).
-Restoring a deduped backup on a version before this one is not supported (that version
-reads rewritten images without their workflow).
+Restoring a deduped backup on a version before this one is not supported: that
+version's restore accepts the file and then reads rewritten images without their
+workflow (README changelog).
 
 ## Workspace data fields for the concept pages (#249)
 

py/api/maintenance.py

@@ -69,6 +69,8 @@ PREVIEW_BODY_MAX = 4096
 DUPLICATE_PREVIEW_BODY_MAX = 1024 * 1024
 DUPLICATE_FILES = "duplicate_files"
 WORKFLOW_STORAGE = "workflow_storage"  # applied by py/api/workflow_storage.py
+# Its backup's name, outside the pruned ``<db>.backup_*`` pool.
+PRE_MIGRATION_LABEL = "pre-workflow-storage"
 PREVIEW_OPERATIONS = MAINTENANCE_OPERATIONS + (DUPLICATE_FILES,)
 # Room left over beyond what a step writes, so a disk is never filled to 0.
 DISK_MARGIN_BYTES = 16 * 2**20
@@ -245,9 +247,9 @@ def _remove_quietly(*paths):
             pass
 
 
-def _reserve_backup_name(db_path):
-    """Create and return an unused ``<db>.backup_<stamp>[_n]`` (O_EXCL)."""
-    for candidate in _backup_names(db_path):
+def _reserve_backup_name(db_path, label="backup"):
+    """Create and return an unused ``<db>.<label>_<stamp>[_n]`` (O_EXCL)."""
+    for candidate in _backup_names(db_path, label):
         try:
             fd = os.open(candidate, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600)
         except FileExistsError:
@@ -257,13 +259,13 @@ def _reserve_backup_name(db_path):
     raise SafetyBackupFailed("No free backup name")
 
 
-def _backup_names(db_path):
-    base = f"{db_path}.backup_{_now().strftime('%Y%m%d_%H%M%S_%f')}"
+def _backup_names(db_path, label="backup"):
+    base = f"{db_path}.{label}_{_now().strftime('%Y%m%d_%H%M%S_%f')}"
     for attempt in range(_NAME_ATTEMPTS):
         yield base if attempt == 0 else f"{base}_{attempt}"
 
 
-def _place_backup(temp, db_path):
+def _place_backup(temp, db_path, label="backup"):
     """Give the verified ``temp`` its unique backup name; return that path.
 
     ``os.link`` is atomic and refuses an existing name, so no empty or half
@@ -271,7 +273,7 @@ def _place_backup(temp, db_path):
     without hard links falls back to reserving the name (``O_EXCL``) and
     ``os.replace``. The temp file is removed by the caller.
     """
-    for candidate in _backup_names(db_path):
+    for candidate in _backup_names(db_path, label):
         try:
             os.link(temp, candidate)
         except FileExistsError:
@@ -282,7 +284,7 @@ def _place_backup(temp, db_path):
             return candidate
     else:
         raise SafetyBackupFailed("No free backup name")
-    final = _reserve_backup_name(db_path)
+    final = _reserve_backup_name(db_path, label)
     try:
         os.replace(temp, final)
     except OSError:
@@ -336,12 +338,15 @@ def _fsync(path):
         os.fsync(handle.fileno())
 
 
-def take_safety_backup(model):
+def take_safety_backup(model, label="backup"):
     """Write and verify a safety backup; return its file name (blocking).
 
     The copy is made in a temp file beside the database and only renamed
-    onto its final, unique name once it verified, so a backup file on disk
-    is always a verified one. Raises SafetyBackupFailed, leaving no file.
+    onto its final, unique name ``<db>.<label>_<stamp>`` once it verified,
+    so a backup file on disk is always a verified one. ``label`` "backup"
+    joins the pruned pool; another label (the workflow storage migration's
+    ``pre-workflow-storage``) is kept until the user removes it. Raises
+    SafetyBackupFailed, leaving no file.
     """
     db_path = os.path.abspath(model.db_path)
     fd, temp = tempfile.mkstemp(
@@ -359,7 +364,7 @@ def take_safety_backup(model):
         _fsync(temp)
         if os.name != "nt":
             os.chmod(temp, 0o600)
-        final = _place_backup(temp, db_path)
+        final = _place_backup(temp, db_path, label)
         return os.path.basename(final)
     except OSError as exc:
         _remove_quietly(final)
@@ -493,11 +498,15 @@ class MaintenanceRoutesMixin:
                 "storage-changed",
             )
         policy = self._safety_backup_policy()
+        destination = "backup"
+        if operation == WORKFLOW_STORAGE:
+            # Always backed up first, whatever the policy (the way back).
+            policy, destination = "required", PRE_MIGRATION_LABEL
         try:
             disk, _short, size = await self._run_in_executor(
                 disk_plan,
                 db.model.db_path,
-                backup=policy == "on",
+                backup=policy != "off",
                 vacuum=operation == "optimize",
             )
         except (OSError, sqlite3.Error):
@@ -522,8 +531,8 @@ class MaintenanceRoutesMixin:
                 **public,
                 "backup": {
                     "policy": policy,
-                    "destination_label": f"{name}.backup_<date and time>",
-                    "expected_bytes": size if policy == "on" else None,
+                    "destination_label": f"{name}.{destination}_<date and time>",
+                    "expected_bytes": size if policy != "off" else None,
                 },
                 "disk": disk,
             },
@@ -620,6 +629,8 @@ class MaintenanceRoutesMixin:
         Decided once, when the apply starts (the disk check measured it), and
         handed to the job.
         """
+        if plan.operation == WORKFLOW_STORAGE:
+            return True  # never without its pre-migration backup
         return plan.payload["backup"] == "on" or self._safety_backup_policy() == "on"
 
     async def _safety_backup_step(self, job, db, backup_on):

py/api/workflow_storage.py

@@ -6,13 +6,17 @@ records to rewrite, the backup and the disk check) and applied with its token
 (``POST /prompt_manager/maintenance/apply``). The apply starts:
 
 1. **Store each workflow once**: kind ``maintenance``, admission ``records``,
-   cancellable, groups ``changed``/``protected``/``errors``. It takes the
-   promised verified safety backup (``backup-failed`` ends it with nothing
-   rewritten), then rewrites old-form rows batch by batch
-   (``database/workflow_storage.py``): one transaction per batch, progress
-   after each, Cancel honoured between batches. A failed batch rolls back
-   whole and fails the job; every record still reads as before. Stopped or
-   failed, the next apply continues from the first old-form row.
+   cancellable, groups ``changed``/``protected``/``errors``. It always takes
+   a verified pre-migration backup, whatever the safety backup policy, named
+   ``<db>.pre-workflow-storage_<stamp>`` outside the pruned ``<db>.backup_*``
+   pool and kept until the user removes it (``backup-failed`` ends the job
+   with nothing rewritten). It then rewrites old-form rows batch by batch
+   (``database/workflow_storage.py``): one transaction per batch of at most
+   ``BATCH_SECONDS``, then ``BATCH_PAUSE`` with the write lock free, progress
+   after each, Cancel honoured between batches. A row that cannot be
+   rewritten stays inline and is listed in ``errors``; any other failure
+   rolls its batch back whole and fails the job. Every record still reads as
+   before, and the next apply continues from the first old-form row.
 2. **Reclaim space**, once every record was looked at: kind ``maintenance``,
    admission ``exclusive`` (it runs alone; VACUUM holds the write lock), not
    cancellable. It checks the disk for VACUUM's copies of the live pages
@@ -23,16 +27,20 @@ records to rewrite, the backup and the disk check) and applied with its token
 Only the database changes: no image or video file is opened.
 """
 
+import asyncio
 import sqlite3
 
 from .jobs import MAINTENANCE_JOB_KIND, JobBusy
 from .maintenance import (
+    BACKUP_FAILED,
     MAINTENANCE_GROUPS,
+    PRE_MIGRATION_LABEL,
+    SafetyBackupFailed,
     _fail_job,
     _megabytes,
-    backup_note,
     disk_plan,
     short_space_message,
+    take_safety_backup,
 )
 
 try:
@@ -50,6 +58,11 @@ REWRITE_FAILED = (
     "rolled back; every record reads as before. Run it again to continue."
 )
 RECLAIM_FAILED = "Reclaim space failed; the database is unchanged."
+ROW_KEPT = "could not be rewritten; it keeps its inline text and reads as before"
+
+
+def _backup_part(backup):
+    return f"Pre-migration backup {backup}"
 
 
 def _plural(count, one, many=None):
@@ -83,26 +96,43 @@ class WorkflowStorageRoutesMixin:
             runner=lambda job: self._run_workflow_storage(job, db, backup),
         )
 
-    async def _run_workflow_storage(self, job, db, backup_on):
-        ok, backup = await self._safety_backup_step(job, db, backup_on)
-        if not ok:
+    async def _run_workflow_storage(self, job, db, _backup_on):
+        # Always backed up first, whatever the policy: the way back.
+        backup = await self._pre_migration_backup(job, db)
+        if backup is None:
             return
-        try:
-            done = await self._rewrite_workflows(job, db, backup)
-        finally:
-            if backup:
-                await self._prune_backup_pool(job, db)
-        if done:
+        if await self._rewrite_workflows(job, db, backup):
             job.finish("finished")
             self._start_reclaim_space(job, db)
 
+    async def _pre_migration_backup(self, job, db):
+        """The verified ``<db>.pre-workflow-storage_<stamp>`` copy, or None.
+
+        Outside the pruned ``<db>.backup_*`` pool: it stays until the user
+        removes it. A failure ends the job ``backup-failed``, nothing changed.
+        """
+        job.update(status_line="Taking the pre-migration backup")
+        job.log_line("Taking the pre-migration backup of the database")
+        try:
+            backup = await self._run_in_executor(
+                take_safety_backup, db.model, PRE_MIGRATION_LABEL
+            )
+        except SafetyBackupFailed:
+            self.logger.error("Pre-migration backup failed", exc_info=True)
+            _fail_job(job, "backup-failed", BACKUP_FAILED)
+            return None
+        job.log_line(
+            f"Pre-migration backup {backup} verified; it is kept until you remove it"
+        )
+        return backup
+
     async def _rewrite_workflows(self, job, db, backup):
         """Rewrite every old-form row; True when every row was looked at."""
         counts = await self._run_in_executor(db.workflow_storage_status)
         job.update(total=counts["rows"], processed=0)
         job.update(status_line="Rewriting image records")
         job.log_line(f"{_plural(counts['rows'], 'image record')} to rewrite")
-        rows = protected = 0
+        rows = protected = errors = 0
         last_id = 0
         while True:
             if job.cancelled:
@@ -123,8 +153,15 @@ class WorkflowStorageRoutesMixin:
             last_id = batch["last_id"]
             rows += batch["rows"]
             protected += batch["protected"]
+            errors += len(batch["errors"])
             job.record_counted("changed", batch["rows"])
             job.record_counted("protected", batch["protected"])
+            for image_id in batch["errors"]:
+                job.record(
+                    "errors",
+                    {"path": "", "detail": f"Image record {image_id}: {ROW_KEPT}"},
+                    f"! Image record {image_id}: {ROW_KEPT}",
+                )
             total = max(job.total, rows)
             job.update(
                 processed=rows,
@@ -133,14 +170,22 @@ class WorkflowStorageRoutesMixin:
                 status_line=f"Rewrote {_plural(rows, 'image record')}",
             )
             job.publish({"type": "progress", "processed": rows, "total": total})
+            # Leave the write lock free between batches (monitor, run counts).
+            await asyncio.sleep(workflow_storage.BATCH_PAUSE)
         removed = await self._run_in_executor(db.sweep_workflow_documents)
         line = f"Rewrote {_plural(rows, 'image record')}"
         if protected:
             line += f"; {_plural(protected, 'stored value')} not text, kept as is"
+        if errors:
+            kept = "keeps its" if errors == 1 else "keep their"
+            line += (
+                f"; {_plural(errors, 'image record')} could not be rewritten and "
+                f"{kept} inline text"
+            )
         job.log_line(line)
         if removed:
             job.log_line(f"Removed {_plural(removed, 'unused workflow document')}")
-        job.update(progress=100, status_line=f"{line} · {backup_note(backup)}")
+        job.update(progress=100, status_line=f"{line} · {_backup_part(backup)}")
         return True
 
     def _stopped(self, job, rows, backup):
@@ -149,7 +194,7 @@ class WorkflowStorageRoutesMixin:
             "run it again to continue"
         )
         job.log_line(line)
-        job.update(status_line=f"{line} · {backup_note(backup)}")
+        job.update(status_line=f"{line} · {_backup_part(backup)}")
         job.finish("cancelled")
         return False
 

tests/test_api_workflow_storage.py

@@ -19,7 +19,7 @@ from unittest.mock import patch
 from PIL import Image, PngImagePlugin
 
 from database import workflow_storage as ws
-from database.workflow_documents import DocumentReader, resolve_image_row
+from database.workflow_documents import DocumentReader, resolve_image_row, store_text
 from py.api import maintenance
 from py.api.jobs import MAINTENANCE_JOB_KIND
 
@@ -118,7 +118,11 @@ class TestWorkflowStorageJob(WorkflowStorageTestCase):
             plan["media"]["label"],
             "No image or video file is opened or changed; only the database.",
         )
-        self.assertEqual(plan["backup"]["policy"], "on")
+        self.assertEqual(plan["backup"]["policy"], "required")
+        self.assertEqual(
+            plan["backup"]["destination_label"],
+            "prompts.db.pre-workflow-storage_<date and time>",
+        )
         self.assertGreater(plan["backup"]["expected_bytes"], 0)
         self.assertTrue(plan["disk"]["sufficient"])
 
@@ -143,9 +147,17 @@ class TestWorkflowStorageJob(WorkflowStorageTestCase):
         self.assertEqual(job.processed, 40)
         self.assertEqual(self.inline_count(), 0)
         self.assertEqual(self.texts(), before)
-        [backup] = self.backups()
-        self.assertIn(f"Safety backup {backup} verified", "\n".join(job.log))
-        self.assertTrue(job.status_line.startswith("Rewrote 40 image records"))
+        self.assertEqual(self.backups(), [])  # never in the pruned pool
+        [backup] = [
+            name
+            for name in os.listdir(self.tmpdir)
+            if name.startswith("prompts.db.pre-workflow-storage_")
+        ]
+        self.assertIn(f"Pre-migration backup {backup} verified", "\n".join(job.log))
+        self.assertEqual(
+            job.status_line,
+            f"Rewrote 40 image records · Pre-migration backup {backup}",
+        )
         # The safety backup holds the old form, verified.
         conn = sqlite3.connect(os.path.join(self.tmpdir, backup))
         try:
@@ -169,13 +181,15 @@ class TestWorkflowStorageJob(WorkflowStorageTestCase):
             "GET", f"/prompt_manager/prompts/{prompt_id}/images"
         )
         self.assertEqual(await listed.json(), listed_before)
-        self.assertEqual(self.strays(), [])
+        self.assertEqual(self.strays(), [backup])  # only the kept backup
         self.assertNotIn(self.tmpdir, json.dumps(job.report()))
 
     async def test_a_backup_failure_rewrites_nothing(self):
         self.old_library()
+        from py.api import workflow_storage as api_storage
+
         with patch.object(
-            maintenance,
+            api_storage,
             "take_safety_backup",
             side_effect=maintenance.SafetyBackupFailed("no"),
         ):
@@ -354,10 +368,7 @@ class TestWorkflowStorageOutcomes(WorkflowStorageTestCase):
                 "file_path, workflow_data) VALUES (?, '/x/b', 'b', '/x/b', x'00')",
                 (prompt_id,),
             )
-            conn.execute(
-                "INSERT INTO workflow_documents (sha256, byte_length, data) "
-                "VALUES (zeroblob(32), 0, x'789c030000000001')"
-            )
+            store_text(conn, "a document no record references")
         _plan, _summary, job = await self.start()
         await self.finish(job)
         log = "\n".join(job.log)

tests/test_workflow_storage_review.py

@@ -0,0 +1,376 @@
+"""PR #315 final review fixes: I1, I2, I3, M1, M2, M3 and M5.
+
+- I1: the migration always takes a verified backup, whatever the policy.
+- I2: that backup has its own name outside the pruned pool.
+- I3: ``tools/workflow_text_digests.py`` proves byte identity across the rewrite.
+- M1: a writer on another connection is never starved by the batches.
+- M2: a row that cannot be rewritten stays inline, is counted, and the rest go on.
+- M3: a damaged document is refused on restore and repaired before reuse.
+- M5: the legacy table rebuilds refuse a file with document id columns.
+"""
+
+import hashlib
+import os
+import re
+import sqlite3
+import subprocess  # nosec B404
+import sys
+import threading
+import zlib
+from unittest.mock import patch
+
+from database import workflow_documents as wd
+from database import workflow_storage as ws
+from database.models import PromptModel
+
+try:
+    from tests.test_api_workflow_storage import OPERATION, WorkflowStorageTestCase
+    from tests.test_workflow_migration import MigrationTestCase
+except ImportError:  # discovered with tests/ as the top-level directory
+    from test_api_workflow_storage import OPERATION, WorkflowStorageTestCase
+    from test_workflow_migration import MigrationTestCase
+
+# After the API test modules, which stand in ComfyUI's ``server`` module.
+from py.api.admin import _prune_safety_backups  # noqa: E402
+from py.config import PromptManagerConfig  # noqa: E402
+
+ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+PRE_MIGRATION = re.compile(
+    r"^prompts\.db\.pre-workflow-storage_\d{8}_\d{6}_\d{6}(_\d+)?$"
+)
+WAIT = 20
+
+
+class TestMigrationBackup(WorkflowStorageTestCase):
+    def pre_migration(self):
+        return sorted(
+            name for name in os.listdir(self.tmpdir) if PRE_MIGRATION.match(name)
+        )
+
+    async def test_the_backup_is_taken_even_with_the_policy_off(self):
+        PromptManagerConfig.SAFETY_BACKUP = False
+        self.old_library(count=4)
+        plan = await self.preview(OPERATION)
+        self.assertEqual(plan["backup"]["policy"], "required")
+        self.assertGreater(plan["backup"]["expected_bytes"], 0)
+        self.assertGreater(plan["disk"]["required_bytes"], 0)
+        body = await self.apply(plan["token"])
+        job = self.api._jobs.get(body["job"]["id"])
+        await self.finish(job)
+        self.assertEqual(job.state, "finished", job.log)
+        [backup] = self.pre_migration()
+        self.assertNotIn("No safety backup", "\n".join(job.log))
+        self.assertIn(f"Pre-migration backup {backup} verified", "\n".join(job.log))
+
+    async def test_the_backup_is_outside_the_pool_never_pruned_and_restores(self):
+        self.old_library(count=4)
+        before = self.texts()
+        plan = await self.preview(OPERATION)
+        self.assertIn(
+            "prompts.db.pre-workflow-storage_<date and time>",
+            " ".join(row["detail"] for row in plan["details"]),
+        )
+        _plan, _summary, job = await self.start()
+        await self.finish(job)
+        [backup] = self.pre_migration()
+        self.assertEqual(self.backups(), [])  # not in the pruned pool
+        self.assertIn(f"Pre-migration backup {backup}", job.status_line)
+        path = os.path.join(self.tmpdir, backup)
+        ok, reason = PromptModel.verify_database_file(path)
+        self.assertTrue(ok, reason)
+        _prune_safety_backups(self.api.db.model.db_path, keep=0)
+        self.assertEqual(self.pre_migration(), [backup])
+        # The way back: the legacy restore accepts it, old form included.
+        self.api.db.model.restore_from_file(path)
+        self.assertEqual(self.texts(), before)
+        self.assertEqual(self.inline_count(), 4)
+
+
+class TestTextDigestsTool(MigrationTestCase):
+    def digests(self, path, out):
+        result = subprocess.run(  # nosec B603
+            [sys.executable, os.path.join(ROOT, "tools", "workflow_text_digests.py")]
+            + [path, out],
+            capture_output=True,
+            text=True,
+            cwd=self.tmpdir,
+            check=False,
+        )
+        self.assertEqual(result.returncode, 0, result.stderr)
+        with open(out, encoding="utf-8") as handle:
+            return handle.read()
+
+    def test_digests_match_across_the_rewrite_and_read_only(self):
+        _prompt_id, expected = self.library(count=12)
+        PromptModel.close_all_instances()
+        before = self.digests(self.db_path, os.path.join(self.tmpdir, "a.sha"))
+        self.assertEqual(len(before.splitlines()), len(expected))
+        self.migrate()
+        PromptModel.close_all_instances()
+        with open(self.db_path, "rb") as handle:
+            frozen = handle.read()
+        after = self.digests(self.db_path, os.path.join(self.tmpdir, "b.sha"))
+        with open(self.db_path, "rb") as handle:
+            self.assertEqual(handle.read(), frozen)
+        self.assertEqual(after, before)
+
+    def test_a_changed_text_shows_in_the_diff(self):
+        prompt_id = self.prompt()
+        self.old_form_image(prompt_id, "a.png", '{"a": 1}', None)
+        PromptModel.close_all_instances()
+        before = self.digests(self.db_path, os.path.join(self.tmpdir, "a.sha"))
+        conn = self.conn()
+        with conn:
+            conn.execute("UPDATE generated_images SET workflow_data='{\"a\":1}'")
+        PromptModel.close_all_instances()
+        after = self.digests(self.db_path, os.path.join(self.tmpdir, "b.sha"))
+        self.assertNotEqual(after, before)
+
+    def test_main_in_process_hashes_undecodable_text_too(self):
+        import contextlib
+        import importlib.util
+        import io
+
+        spec = importlib.util.spec_from_file_location(
+            "workflow_text_digests",
+            os.path.join(ROOT, "tools", "workflow_text_digests.py"),
+        )
+        tool = importlib.util.module_from_spec(spec)
+        spec.loader.exec_module(tool)
+        prompt_id = self.prompt()
+        conn = self.conn()
+        with conn:
+            conn.execute(
+                "INSERT INTO generated_images (prompt_id, image_path, filename, "
+                "file_path, workflow_data) VALUES (?, '/x', 'x', '/x', "
+                "CAST(x'ff' AS TEXT))",
+                (prompt_id,),
+            )
+        out = os.path.join(self.tmpdir, "c.sha")
+        printed = io.StringIO()
+        with contextlib.redirect_stdout(printed):
+            self.assertEqual(tool.main([self.db_path, out]), 0)
+        self.assertEqual(printed.getvalue(), "1 image records\n")
+        with open(out, encoding="utf-8") as handle:
+            line = handle.read().split()
+        self.assertEqual(line[1], hashlib.sha256(b"\xff").hexdigest())
+        self.assertEqual(line[2], "-")
+
+
+class TestWriterIsNotStarved(MigrationTestCase):
+    def test_a_writer_on_another_connection_gets_in_between_batches(self):
+        prompt_id = self.prompt()
+        for index in range(200):
+            self.old_form_image(prompt_id, f"{index}.png", f'{{"n": {index}}}', None)
+        started, done = threading.Event(), threading.Event()
+        errors, written = [], []
+        real = ws.store_text
+
+        def slow(conn, text, *args, **kwargs):
+            started.set()
+            done.wait(0.01)  # a slow disk: each row takes a while
+            return real(conn, text, *args, **kwargs)
+
+        def writer():
+            started.wait(WAIT)
+            conn = self.db.model.get_connection()  # this thread's own connection
+            conn.execute("PRAGMA busy_timeout = 2000")
+            try:
+                for index in range(5):
+                    written.append(
+                        self.db.link_image_to_prompt(
+                            prompt_id,
+                            os.path.join(self.tmpdir, f"w{index}.png"),
+                            {"workflow": {"w": index}},
+                        )
+                    )
+            except sqlite3.OperationalError as error:  # pragma: no cover - RED
+                errors.append(str(error))
+            finally:
+                self.db.model.close()
+
+        thread = threading.Thread(target=writer)
+        thread.start()
+        with patch.object(ws, "store_text", side_effect=slow):
+            self.db.migrate_workflow_storage()
+        done.set()
+        thread.join(WAIT)
+        self.assertEqual(errors, [])
+        self.assertEqual(len([image for image in written if image]), 5)
+
+    def test_a_batch_has_a_time_budget(self):
+        prompt_id = self.prompt()
+        for index in range(30):
+            self.old_form_image(prompt_id, f"{index}.png", f'{{"n": {index}}}', None)
+        ticks = iter(range(10**6))
+        with patch.object(ws, "_clock", side_effect=lambda: next(ticks)):
+            batch = ws.migrate_batch(self.conn(), 0, max_seconds=5)
+        self.assertLess(batch["rows"], 30)
+        self.assertGreater(batch["rows"], 0)
+
+
+class TestRowErrors(MigrationTestCase):
+    def test_an_unreadable_row_stays_inline_and_the_rest_are_rewritten(self):
+        prompt_id, expected = self.library(count=10)
+        conn = self.conn()
+        with conn:
+            bad = conn.execute(
+                "INSERT INTO generated_images (prompt_id, image_path, filename, "
+                "file_path, workflow_data) VALUES (?, '/x/bad', 'bad', '/x/bad', "
+                "CAST(x'7b22ff227d' AS TEXT))",
+                (prompt_id,),
+            ).lastrowid
+        result = self.migrate()
+        self.assertEqual(result["errors"], [bad])
+        self.assertEqual(result["rows"], len(expected))
+        self.assertEqual(self.inline_ids_without(bad), set())
+        raw = conn.execute(
+            "SELECT CAST(workflow_data AS BLOB), workflow_document_id "
+            "FROM generated_images WHERE id=?",
+            (bad,),
+        ).fetchone()
+        self.assertEqual((bytes(raw[0]), raw[1]), (b'{"\xff"}', None))
+        self.assertEqual(result["documents_removed"], 0)
+
+    def inline_ids_without(self, image_id):
+        return {
+            row[0]
+            for row in self.conn().execute(
+                "SELECT id FROM generated_images WHERE id != ? AND "
+                "(workflow_data IS NOT NULL OR prompt_metadata IS NOT NULL)",
+                (image_id,),
+            )
+        }
+
+    def test_a_row_whose_text_will_not_come_back_is_kept_and_counted(self):
+        prompt_id = self.prompt()
+        good = self.old_form_image(prompt_id, "good.png", '{"good": 1}', None)
+        bad = self.old_form_image(prompt_id, "bad.png", '{"bad": 1}', None)
+        real = ws.read_stored_text
+
+        def lying(conn, inline, document_id):
+            text = real(conn, inline, document_id)
+            return "other" if text == '{"bad": 1}' else text
+
+        with patch.object(ws, "read_stored_text", side_effect=lying):
+            result = self.migrate()
+        self.assertEqual(result["errors"], [bad])
+        self.assertEqual(self.inline_ids(), {bad})
+        self.assertEqual(self.texts()[good][0], '{"good": 1}')
+        self.assertEqual(self.texts()[bad][0], '{"bad": 1}')
+        self.assertEqual(self.document_count(), 1)  # the bad row's went
+
+
+class TestRowErrorsInTheJob(WorkflowStorageTestCase):
+    async def test_the_job_reports_kept_rows_and_reclaim_space_still_runs(self):
+        prompt_id = self.old_library(count=4)
+        conn = self.api.db.model.get_connection()
+        with conn:
+            conn.execute(
+                "INSERT INTO generated_images (prompt_id, image_path, filename, "
+                "file_path, workflow_data) VALUES (?, '/x/bad', 'bad', '/x/bad', "
+                "CAST(x'7b22ff227d' AS TEXT))",
+                (prompt_id,),
+            )
+        _plan, _summary, job = await self.start()
+        reclaim = await self.finish(job)
+        self.assertEqual(job.state, "finished", job.log)
+        self.assertEqual(job.counters["changed"], 4)
+        self.assertEqual(job.counters["errors"], 1)
+        self.assertIn(
+            "1 image record could not be rewritten and keeps its inline text",
+            job.status_line,
+        )
+        self.assertIsNotNone(reclaim)
+        self.assertEqual(reclaim.state, "finished", reclaim.log)
+
+
+class TestDamagedDocuments(MigrationTestCase):
+    def damage(self, document_id, data=b"not zlib"):
+        conn = self.conn()
+        with conn:
+            conn.execute(
+                "UPDATE workflow_documents SET data=? WHERE id=?", (data, document_id)
+            )
+
+    def test_verify_refuses_a_damaged_document(self):
+        document_id = self.store('{"a": 1}')
+        ok, _reason = PromptModel.verify_database_file(self.db_path)
+        self.assertTrue(ok)
+        for data in (b"not zlib", zlib.compress(b'{"a": 2}')):
+            self.damage(document_id, data)
+            with self.subTest(data=data):
+                ok, reason = PromptModel.verify_database_file(self.db_path)
+                self.assertFalse(ok)
+                self.assertEqual(
+                    reason, "Workflow documents contain a damaged document"
+                )
+
+    def test_a_damaged_document_is_repaired_before_it_is_reused(self):
+        prompt_id = self.prompt()
+        first = self.db.link_image_to_prompt(
+            prompt_id, os.path.join(self.tmpdir, "a.png"), {"workflow": {"v": 1}}
+        )
+        document_id = self.raw(first)["workflow_document_id"]
+        self.damage(document_id)
+        with self.assertLogs(wd.logger, "WARNING"):
+            second = self.db.link_image_to_prompt(
+                prompt_id, os.path.join(self.tmpdir, "b.png"), {"workflow": {"v": 1}}
+            )
+        self.assertEqual(self.raw(second)["workflow_document_id"], document_id)
+        for image_id in (first, second):
+            self.assertEqual(self.texts()[image_id][0], '{"v": 1}')
+
+    def test_store_verifies_a_reused_document_once_per_given_set(self):
+        document_id = self.store("abc")
+        verified = set()
+        conn = self.conn()
+        with conn:
+            self.assertEqual(wd.store_text(conn, "abc", verified), document_id)
+        self.assertEqual(verified, {document_id})
+        self.damage(document_id)
+        with conn:  # already verified in this set: not read again
+            self.assertEqual(wd.store_text(conn, "abc", verified), document_id)
+
+
+class TestRebuildGuards(MigrationTestCase):
+    def test_the_legacy_rebuilds_refuse_document_columns(self):
+        for column, kind in (("prompt_id", "TEXT"), ("file_path", None)):
+            path = os.path.join(self.tmpdir, f"guard-{column}.db")
+            conn = sqlite3.connect(path)
+            try:
+                columns = [
+                    "id INTEGER PRIMARY KEY AUTOINCREMENT",
+                    f"prompt_id {kind or 'INTEGER'} NOT NULL",
+                    "image_path TEXT NOT NULL",
+                    "filename TEXT NOT NULL",
+                    "generation_time TIMESTAMP",
+                    "workflow_data TEXT",
+                    "prompt_metadata TEXT",
+                    "parameters TEXT",
+                    "workflow_document_id INTEGER",
+                    "prompt_document_id INTEGER",
+                ]
+                if kind is not None:
+                    columns.append("file_path TEXT")
+                conn.executescript(
+                    "CREATE TABLE prompts (id INTEGER PRIMARY KEY AUTOINCREMENT, "
+                    "text TEXT NOT NULL, created_at TIMESTAMP, updated_at TIMESTAMP, "
+                    "category TEXT, tags TEXT, rating INTEGER, notes TEXT, "
+                    "hash TEXT UNIQUE);"
+                    f"CREATE TABLE generated_images ({', '.join(columns)});"
+                )
+                conn.commit()
+            finally:
+                conn.close()
+            with self.subTest(column=column):
+                with self.assertRaisesRegex(sqlite3.DatabaseError, "workflow document"):
+                    PromptModel(path)
+                PromptModel.close_all_instances()
+                PromptModel.reset_schema_cache()
+
+
+if __name__ == "__main__":  # pragma: no cover
+    import unittest
+
+    unittest.main()

tools/workflow_text_digests.py

@@ -0,0 +1,73 @@
+"""Digests of every image record's stored workflow and prompt graph text.
+
+Proves the workflow storage rewrite (#249 S6) kept every text byte for byte.
+Run it from this repository on the database before the rewrite (or on its
+``prompts.db.pre-workflow-storage_*`` backup) and again after, then compare::
+
+    python3 tools/workflow_text_digests.py prompts.db.pre-workflow-storage_<stamp> \\
+        before.sha
+    python3 tools/workflow_text_digests.py prompts.db after.sha
+    diff before.sha after.sha        # prints nothing when every text is identical
+
+Each line is ``<image id> <sha256 of workflow_data> <sha256 of prompt_metadata>``
+(``-`` for none), hashing the exact stored bytes: the inline column before the
+rewrite, the document read through the one accessor after it. Read-only: the
+database is opened with ``mode=ro`` and nothing is written to it.
+"""
+
+import argparse
+import hashlib
+import os
+import sqlite3
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+from database.workflow_documents import (  # noqa: E402
+    DocumentReader,
+    resolve_image_row,
+)
+
+
+def _digest(value):
+    if value is None:
+        return "-"
+    raw = value if isinstance(value, bytes) else value.encode("utf-8")
+    return hashlib.sha256(raw).hexdigest()
+
+
+def write_digests(db_path, out):
+    """Write one digest line per image record of ``db_path`` to ``out``."""
+    uri = Path(os.path.abspath(db_path)).as_uri() + "?mode=ro"
+    conn = sqlite3.connect(uri, uri=True)
+    try:
+        conn.row_factory = sqlite3.Row
+        conn.text_factory = bytes  # inline text exactly as stored, even undecodable
+        reader = DocumentReader(conn)
+        count = 0
+        with open(out, "w", encoding="utf-8") as handle:
+            for row in conn.execute("SELECT * FROM generated_images ORDER BY id"):
+                texts = resolve_image_row(reader, row)
+                handle.write(
+                    f"{row['id']} {_digest(texts.get('workflow_data'))} "
+                    f"{_digest(texts.get('prompt_metadata'))}\n"
+                )
+                count += 1
+        return count
+    finally:
+        conn.close()
+
+
+def main(argv=None):
+    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
+    parser.add_argument("database", help="prompts.db or one of its backups")
+    parser.add_argument("out", help="file to write the digests to")
+    args = parser.parse_args(argv)
+    count = write_digests(args.database, args.out)
+    print(f"{count} image records")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())