From 1db50500c97fa50fb94425b2fbc274432915440e Mon Sep 17 00:00:00 2001 From: Vito Sansevero Date: Sat, 17 Oct 2026 20:47:22 -0700 Subject: [PATCH] feat(history): persist immutable run output references --- database/execution_outputs.py | 153 +++++++++ database/operations.py | 2 + tests/test_execution_history_restore.py | 22 ++ tests/test_execution_output_database.py | 426 ++++++++++++++++++++++++ tests/test_execution_output_gallery.py | 244 ++++++++++++++ 5 files changed, 847 insertions(+) create mode 100644 database/execution_outputs.py create mode 100644 tests/test_execution_output_database.py create mode 100644 tests/test_execution_output_gallery.py diff --git a/database/execution_outputs.py b/database/execution_outputs.py new file mode 100644 index 0000000..1c3ed8c --- /dev/null +++ b/database/execution_outputs.py @@ -0,0 +1,153 @@ +"""Immutable run-owned output references, independent of mutable galleries.""" + +import sqlite3 +from contextlib import ExitStack +from typing import Callable, ContextManager + +try: + from ..utils.execution_history_events import RunFinish, encode_command + from ..utils.execution_outputs import OutputBundle, validate_output_bundle +except ImportError: + from utils.execution_history_events import RunFinish, encode_command + from utils.execution_outputs import OutputBundle, validate_output_bundle + + +_REFERENCE_FIELDS = ( + "output_node_id", + "image_ordinal", + "folder_type", + "subfolder", + "filename", + "path_style", + "root_key", +) +_REFERENCE_COLUMNS = ",".join(_REFERENCE_FIELDS) + + +def _capture(conn: sqlite3.Connection, run_id: str) -> sqlite3.Row | None: + capture = conn.execute( + "SELECT state,reason,format_version,source FROM execution_output_capture " + "WHERE run_id=?", + (run_id,), + ).fetchone() + # Legacy callers may disable FK enforcement; never adopt orphan evidence. + if capture is None: + if conn.execute( + "SELECT 1 FROM execution_run_outputs WHERE run_id=? LIMIT 1", (run_id,) + ).fetchone(): + raise sqlite3.DatabaseError("Output references have no capture header") + elif ( + conn.execute( + "SELECT 1 FROM execution_runs WHERE run_id=?", (run_id,) + ).fetchone() + is None + ): + raise sqlite3.DatabaseError("Output capture has no execution parent") + return capture + + +def _references(conn: sqlite3.Connection, run_id: str) -> list[tuple]: + return [ + tuple(row) + for row in conn.execute( + f"SELECT {_REFERENCE_COLUMNS} FROM execution_run_outputs " + "WHERE run_id=? ORDER BY output_node_id,image_ordinal", + (run_id,), + ) + ] + + +class ExecutionOutputOperations: + """Compose with ExecutionHistoryOperations on PromptDatabase.""" + + def write_execution_outputs_for_finish( + self, + command: RunFinish, + bundle: OutputBundle, + *, + generation_guard: Callable[[], ContextManager[bool]] | None = None, + ) -> bool: + encode_command(command) + if type(command) is not RunFinish: + raise ValueError("Expected execution finish command") + validate_output_bundle(bundle) + header = (bundle.state, bundle.reason, bundle.format_version, bundle.source) + prepared = [ + tuple(getattr(reference, field) for field in _REFERENCE_FIELDS) + for reference in bundle.references + ] + # Enter after BEGIN and retain the guard until COMMIT/ROLLBACK finishes. + with ExitStack() as guards: + with self._execution_transaction(command.storage_epoch) as conn: + if generation_guard is not None and not guards.enter_context( + generation_guard() + ): + return False + parent = conn.execute( + "SELECT run_id FROM execution_runs WHERE session_id=? " + "AND queue_item_id=? AND storage_epoch=?", + (command.session_id, command.queue_item_id, command.storage_epoch), + ).fetchone() + if parent is None: + return False + run_id = parent[0] + capture = _capture(conn, run_id) + if capture is not None: + if capture[0] != "conflicting" and ( + tuple(capture) != header + or _references(conn, run_id) != prepared + ): + conn.execute( + "UPDATE execution_output_capture SET " + "state='conflicting',reason='conflicting' WHERE run_id=?", + (run_id,), + ) + return True + conn.execute( + "INSERT INTO execution_output_capture " + "(run_id,state,reason,format_version,source) VALUES(?,?,?,?,?)", + (run_id, *header), + ) + conn.executemany( + f"INSERT INTO execution_run_outputs (run_id,{_REFERENCE_COLUMNS}) " + "VALUES(?,?,?,?,?,?,?,?)", + [(run_id, *reference) for reference in prepared], + ) + return True + + def get_execution_outputs(self, run_id: str) -> dict: + """Read frozen evidence in one snapshot without resolving its locators.""" + conn = self.model.get_connection() + own_transaction = not conn.in_transaction + if own_transaction: + conn.execute("BEGIN") + try: + capture = _capture(conn, run_id) + if capture is None: + return { + "state": "not_captured", + "reason": None, + "format_version": None, + "source": None, + "references": [], + } + references = [] + for row in _references(conn, run_id): + reference = dict(zip(_REFERENCE_FIELDS, row)) + reference.update( + file_identity="unverified", + newly_generated="unknown", + context_membership="unknown", + transient=reference["folder_type"] == "temp", + ) + references.append(reference) + return { + "state": capture[0], + "reason": capture[1], + "format_version": capture[2], + "source": capture[3], + "references": references, + } + finally: + if own_transaction: + conn.rollback() diff --git a/database/operations.py b/database/operations.py index 92c1ead..3df8f5f 100644 --- a/database/operations.py +++ b/database/operations.py @@ -13,6 +13,7 @@ from .pairs import PairDependencyError, PairOperations from .library import LibraryOperations from .execution_history import ExecutionHistoryOperations from .execution_contexts import ExecutionContextOperations +from .execution_outputs import ExecutionOutputOperations # Import logging system try: @@ -158,6 +159,7 @@ class PromptDatabase( LibraryOperations, ExecutionHistoryOperations, ExecutionContextOperations, + ExecutionOutputOperations, ): """Database operations class for managing prompts.""" diff --git a/tests/test_execution_history_restore.py b/tests/test_execution_history_restore.py index be68a42..f6d5326 100644 --- a/tests/test_execution_history_restore.py +++ b/tests/test_execution_history_restore.py @@ -22,6 +22,28 @@ class HistoryRestoreTests(unittest.TestCase): finish = history_tests.HistoryDatabaseTests.finish count = history_tests.HistoryDatabaseTests.count + def test_output_operations_restore_preserves_evidence_and_rejects_stale_finish( + self, + ): + from database.execution_history import HistoryEpochChanged + from tests.test_execution_output_database import bundle + + db = self.database() + start = self.start(db) + db.write_execution_start(start) + finish = self.finish(start) + db.write_execution_outputs_for_finish(finish, bundle()) + evidence = db.get_execution_outputs(start.run_id) + backup = str(Path(self.temp.name) / "output-operations.db") + self.assertTrue(db.model.backup_database(backup)) + source = Path(backup).read_bytes() + db.model.restore_from_file(backup) + self.assertEqual(db.get_execution_outputs(start.run_id), evidence) + with self.assertRaises(HistoryEpochChanged): + db.write_execution_outputs_for_finish(finish, bundle("late.png")) + self.assertEqual(db.get_execution_outputs(start.run_id), evidence) + self.assertEqual(Path(backup).read_bytes(), source) + def test_output_wal_backup_restore_preserves_rows_and_historical_epoch(self): try: from tests.test_execution_output_schema import capture diff --git a/tests/test_execution_output_database.py b/tests/test_execution_output_database.py new file mode 100644 index 0000000..ee494a6 --- /dev/null +++ b/tests/test_execution_output_database.py @@ -0,0 +1,426 @@ +"""Frozen output occurrences belong to an exact stored execution attempt.""" + +import sqlite3 +import unittest +from concurrent.futures import ThreadPoolExecutor +from contextlib import contextmanager +from dataclasses import replace +from threading import Barrier +from unittest.mock import patch + +from database.execution_history import HistoryEpochChanged +from utils.execution_outputs import OutputBundle, OutputReference + +try: + from tests import test_execution_history_database as history_tests +except ImportError: + import test_execution_history_database as history_tests + + +def bundle(filename="first.png", *, root="a" * 64, node="9", ordinal=0): + return OutputBundle( + (OutputReference(node, ordinal, "output", "", filename, "posix", root),), + "complete" if root else "partial", + "reported" if root else "root_unavailable", + ) + + +class OutputDatabaseTests(unittest.TestCase): + setUp = history_tests.HistoryDatabaseTests.setUp + close_databases = history_tests.HistoryDatabaseTests.close_databases + database = history_tests.HistoryDatabaseTests.database + start = history_tests.HistoryDatabaseTests.start + finish = history_tests.HistoryDatabaseTests.finish + + def capture(self, db, value=None, item=0): + start = self.start(db, item) + db.write_execution_start(start) + self.assertTrue( + db.write_execution_outputs_for_finish(self.finish(start), value or bundle()) + ) + return start + + def rows(self, db): + return [ + tuple(row) + for row in db.model.get_connection().execute( + "SELECT * FROM execution_run_outputs ORDER BY run_id," + "output_node_id,image_ordinal" + ) + ] + + def test_exact_attempt_parent_not_recomputed_uuid(self): + db = self.database() + stored_id = "55555555-5555-4555-8555-555555555555" + start = replace(self.start(db), run_id=stored_id) + db.write_execution_start(start) + self.assertTrue( + db.write_execution_outputs_for_finish(self.finish(start), bundle()) + ) + self.assertEqual(self.rows(db)[0][0], stored_id) + self.assertEqual(db.get_execution_outputs(stored_id)["state"], "complete") + other = self.capture(db, item=1) + self.assertEqual(len(self.rows(db)), 2) + self.assertEqual( + db.get_execution_outputs(other.run_id)["references"][0]["filename"], + "first.png", + ) + self.assertEqual(db.get_execution_run(stored_id)["outcome"], "observed") + + def test_exact_replay_noop_and_conflict_sticky(self): + db = self.database() + first = bundle() + refs = (replace(first.references[0], output_node_id="2"),) + ( + replace(first.references[0], image_ordinal=0), + replace(first.references[0], image_ordinal=1), + ) + first = replace(first, references=refs) + start = self.capture(db, first) + finish = self.finish(start) + original = self.rows(db) + # Construct identical DTOs from dictionaries with different insertion orders. + reordered = tuple( + OutputReference(**dict(reversed(list(vars(ref).items())))) for ref in refs + ) + self.assertTrue( + db.write_execution_outputs_for_finish( + finish, replace(first, references=reordered) + ) + ) + self.assertEqual(db.get_execution_outputs(start.run_id)["state"], "complete") + self.assertEqual(self.rows(db), original) + self.assertTrue( + db.write_execution_outputs_for_finish(finish, bundle("other.png")) + ) + capture = db.get_execution_outputs(start.run_id) + self.assertEqual( + (capture["state"], capture["reason"]), ("conflicting", "conflicting") + ) + self.assertEqual( + [r["filename"] for r in capture["references"]], ["first.png"] * 3 + ) + self.assertEqual(capture["references"][0]["context_membership"], "unknown") + for replay in (first, bundle("third.png")): + db.write_execution_outputs_for_finish(finish, replay) + self.assertEqual(db.get_execution_outputs(start.run_id), capture) + self.assertEqual(self.rows(db), original) + + def test_header_and_every_reference_scalar_participate_in_replay(self): + variants = ( + replace(bundle(), state="partial", reason="unsupported"), + bundle(root="b" * 64), + bundle(node="8"), + bundle(ordinal=1), + replace( + bundle(), + references=(replace(bundle().references[0], folder_type="temp"),), + ), + replace( + bundle(), + references=(replace(bundle().references[0], subfolder="nested"),), + ), + replace( + bundle(), + references=(replace(bundle().references[0], path_style="windows"),), + ), + OutputBundle((), "complete", "none_reported"), + ) + db = self.database() + for item, variant in enumerate(variants): + with self.subTest(item=item): + start = self.capture(db, item=item) + db.write_execution_outputs_for_finish(self.finish(start), variant) + capture = db.get_execution_outputs(start.run_id) + self.assertEqual(capture["state"], "conflicting") + self.assertEqual(capture["references"][0]["root_key"], "a" * 64) + self.assertEqual(capture["references"][0]["output_node_id"], "9") + + def test_concurrent_replay_keeps_first_rows(self): + db = self.database() + start = self.start(db) + db.write_execution_start(start) + finish = self.finish(start) + peers = [self.database() for _ in range(4)] + barrier = Barrier(len(peers)) + + def write(index): + barrier.wait(timeout=5) + return peers[index].write_execution_outputs_for_finish( + finish, bundle("first.png" if index % 2 else "second.png") + ) + + with ThreadPoolExecutor(max_workers=4) as executor: + self.assertEqual(list(executor.map(write, range(4))), [True] * 4) + capture = db.get_execution_outputs(start.run_id) + self.assertEqual( + (capture["state"], capture["reason"]), ("conflicting", "conflicting") + ) + self.assertEqual(len(capture["references"]), 1) + frozen = self.rows(db) + for value in (bundle(), bundle("second.png")): + db.write_execution_outputs_for_finish(finish, value) + self.assertEqual(self.rows(db), frozen) + + def test_epoch_and_generation_guard_cover_commit(self): + db = self.database() + start = self.start(db) + db.write_execution_start(start) + observations = [] + conn = db.model.get_connection() + active = False + + @contextmanager + def guard(): + nonlocal active + self.assertTrue(conn.in_transaction) + active = True + try: + yield True + finally: + observations.append(("exit", conn.in_transaction)) + active = False + + def trace(sql): + if sql == "COMMIT": + observations.append(("commit", active)) + + conn.set_trace_callback(trace) + self.addCleanup(conn.set_trace_callback, None) + self.assertTrue( + db.write_execution_outputs_for_finish( + self.finish(start), bundle(), generation_guard=guard + ) + ) + self.assertEqual(observations, [("commit", True), ("exit", False)]) + self.assertFalse(active) + + @contextmanager + def stale(): + yield False + + other = self.start(db, item=1) + db.write_execution_start(other) + self.assertFalse( + db.write_execution_outputs_for_finish( + self.finish(other), bundle(), generation_guard=stale + ) + ) + self.assertEqual( + db.get_execution_outputs(other.run_id)["state"], "not_captured" + ) + old_epoch = start.storage_epoch + with conn: + conn.execute( + "UPDATE execution_history_metadata SET storage_epoch=?", + ("66666666-6666-4666-8666-666666666666",), + ) + with self.assertRaises(HistoryEpochChanged): + db.write_execution_outputs_for_finish( + self.finish(start), bundle("late.png") + ) + self.assertEqual(db.get_execution_run(start.run_id)["storage_epoch"], old_epoch) + self.assertEqual( + db.get_execution_outputs(start.run_id)["references"][0]["filename"], + "first.png", + ) + + def test_failure_rolls_back_header_and_keeps_guard_until_rollback(self): + db = self.database() + start = self.start(db) + db.write_execution_start(start) + conn = db.model.get_connection() + with conn: + conn.execute( + "CREATE TRIGGER reject_output BEFORE INSERT ON execution_run_outputs " + "BEGIN SELECT RAISE(ABORT,'injected'); END" + ) + states = [] + + @contextmanager + def guard(): + states.append(conn.in_transaction) + try: + yield True + finally: + states.append(conn.in_transaction) + + with self.assertRaises(sqlite3.IntegrityError): + db.write_execution_outputs_for_finish( + self.finish(start), bundle(), generation_guard=guard + ) + self.assertEqual(states, [True, False]) + self.assertEqual( + db.get_execution_outputs(start.run_id)["state"], "not_captured" + ) + self.assertEqual(self.rows(db), []) + + def test_missing_header_or_parent_rejected(self): + db = self.database() + start = self.start(db) + self.assertFalse( + db.write_execution_outputs_for_finish(self.finish(start), bundle()) + ) + self.assertEqual( + db.get_execution_outputs(start.run_id), + { + "state": "not_captured", + "reason": None, + "format_version": None, + "source": None, + "references": [], + }, + ) + db.write_execution_start(start) + conn = db.model.get_connection() + conn.execute("PRAGMA foreign_keys=OFF") + self.assertEqual(conn.execute("PRAGMA foreign_keys").fetchone()[0], 0) + with conn: + conn.execute( + "INSERT INTO execution_run_outputs " + "VALUES(?, '9',0,'output','','orphan.png','posix',?)", + (start.run_id, "a" * 64), + ) + with self.assertRaises(sqlite3.DatabaseError): + db.get_execution_outputs(start.run_id) + with self.assertRaises(sqlite3.DatabaseError): + db.write_execution_outputs_for_finish(self.finish(start), bundle()) + self.assertEqual( + conn.execute("SELECT count(*) FROM execution_output_capture").fetchone()[0], + 0, + ) + with conn: + conn.execute("DELETE FROM execution_run_outputs") + db.write_execution_outputs_for_finish(self.finish(start), bundle()) + with conn: + conn.execute("DELETE FROM execution_runs") + with self.assertRaises(sqlite3.DatabaseError): + db.get_execution_outputs(start.run_id) + + def test_reads_never_access_files_or_rebind(self): + db = self.database() + value = bundle(root=None) + value = replace( + value, + references=( + replace( + value.references[0], + folder_type="temp", + subfolder=" Exact Ω ", + filename="Keep Case.png", + ), + ), + ) + start = self.capture(db, value) + conn = db.model.get_connection() + before = list(conn.iterdump()) + with ( + patch("builtins.open", side_effect=AssertionError("file read")), + patch("os.stat", side_effect=AssertionError("stat")), + patch("os.path.realpath", side_effect=AssertionError("resolve")), + ): + capture = db.get_execution_outputs(start.run_id) + self.assertEqual( + capture, + { + "state": "partial", + "reason": "root_unavailable", + "format_version": 1, + "source": "host_final_history", + "references": [ + { + "output_node_id": "9", + "image_ordinal": 0, + "folder_type": "temp", + "subfolder": " Exact Ω ", + "filename": "Keep Case.png", + "path_style": "posix", + "root_key": None, + "file_identity": "unverified", + "newly_generated": "unknown", + "context_membership": "unknown", + "transient": True, + } + ], + }, + ) + self.assertEqual(list(conn.iterdump()), before) + for table in ( + "prompts", + "prompt_pairs", + "generated_images", + "execution_run_contexts", + ): + self.assertEqual( + conn.execute(f"SELECT count(*) FROM {table}").fetchone()[0], 0 + ) + + def test_read_uses_consistent_snapshot_and_preserves_callers_transaction(self): + db = self.database() + start = self.capture(db) + peer = self.database() + conn = db.model.get_connection() + changed = False + + def trace(sql): + nonlocal changed + if "FROM execution_run_outputs" in sql and not changed: + changed = True + with peer.model.get_connection() as writer: + writer.execute( + "UPDATE execution_run_outputs SET filename='later.png'" + ) + + conn.set_trace_callback(trace) + self.addCleanup(conn.set_trace_callback, None) + self.assertEqual( + db.get_execution_outputs(start.run_id)["references"][0]["filename"], + "first.png", + ) + conn.set_trace_callback(None) + self.assertTrue(changed) + self.assertFalse(conn.in_transaction) + self.assertEqual( + db.get_execution_outputs(start.run_id)["references"][0]["filename"], + "later.png", + ) + conn.execute("BEGIN") + db.get_execution_outputs(start.run_id) + self.assertTrue(conn.in_transaction) + conn.rollback() + + def test_empty_unavailable_and_partial_headers_remain_distinct(self): + db = self.database() + for item, (state, reason) in enumerate( + ( + ("complete", "none_reported"), + ("unavailable", "unavailable"), + ("partial", "unsupported"), + ) + ): + start = self.capture(db, OutputBundle((), state, reason), item) + capture = db.get_execution_outputs(start.run_id) + self.assertEqual( + (capture["state"], capture["reason"], capture["references"]), + (state, reason, []), + ) + + def test_invalid_scalars_rejected_before_transaction(self): + db = self.database() + start = self.start(db) + db.write_execution_start(start) + trace = [] + conn = db.model.get_connection() + conn.set_trace_callback(trace.append) + self.addCleanup(conn.set_trace_callback, None) + for invalid in ( + replace( + bundle(), + references=(replace(bundle().references[0], filename="../unsafe.png"),), + ), + replace(bundle(), source="filesystem_scan"), + ): + with self.assertRaises(ValueError): + db.write_execution_outputs_for_finish(self.finish(start), invalid) + with self.assertRaises(ValueError): + db.write_execution_outputs_for_finish(start, bundle()) + self.assertEqual(trace, []) diff --git a/tests/test_execution_output_gallery.py b/tests/test_execution_output_gallery.py new file mode 100644 index 0000000..661e3a8 --- /dev/null +++ b/tests/test_execution_output_gallery.py @@ -0,0 +1,244 @@ +"""Mutable metadata galleries never own or retarget frozen output references.""" + +from PIL import Image +from PIL.PngImagePlugin import PngInfo + +from database.pairs import PairDependencyError +from tests import test_execution_history_database as history_tests +from tests.test_api_admin import AdminAPITestCase +from tests.test_execution_output_database import bundle +from utils.execution_outputs import snapshot_output_roots + + +class OutputGalleryTests(AdminAPITestCase): + start = history_tests.HistoryDatabaseTests.start + finish = history_tests.HistoryDatabaseTests.finish + + def capture(self, filename="first.png", item=0, root=None): + db = self.api.db + start = self.start(db, item) + db.write_execution_start(start) + key = snapshot_output_roots( + "posix", str(root or self.output_dir), None + ).output_key + db.write_execution_outputs_for_finish( + self.finish(start), bundle(filename, root=key) + ) + return start.run_id, db.get_execution_outputs(start.run_id) + + def assert_evidence(self, run, expected): + self.assertEqual(self.api.db.get_execution_outputs(run), expected) + reference = expected["references"][0] + self.assertEqual(reference["context_membership"], "unknown") + self.assertNotIn("image_id", reference) + self.assertNotIn("pair_id", reference) + self.assertNotIn("prompt_id", reference) + + def gallery(self): + conn = self.api.db.model.get_connection() + return { + table: [ + tuple(row) for row in conn.execute(f"SELECT * FROM {table} ORDER BY id") + ] + for table in ( + "prompts", + "prompt_pairs", + "generated_images", + "image_prompt_pairs", + ) + } + + def test_metadata_disabled_or_malformed_still_captures_without_gallery_rows(self): + for item, metadata in enumerate(({}, {"prompt": "{malformed"})): + with self.subTest(metadata=metadata): + path = self.output_dir / f"metadata-{item}.png" + info = PngInfo() + for key, value in metadata.items(): + info.add_text(key, value) + Image.new("RGB", (2, 2)).save(path, pnginfo=info) + extracted = self.api._extract_comfyui_metadata(str(path)) + self.assertEqual( + self.api._ingest_scanned_file(path, extracted), + None if not metadata else "unresolved", + ) + before = self.gallery() + run, evidence = self.capture(path.name, item) + self.assertEqual(evidence["state"], "complete") + self.assertEqual(self.gallery(), before) + self.assertEqual(self.api.db.get_recent_prompts()["total"], 0) + self.assert_evidence(run, evidence) + + def test_scans_before_after_delayed_repeated_and_overwrite_keep_exact_counters( + self, + ): + metadata = { + "parameters": "cat\nNegative prompt: blur\nSteps: 20, Sampler: Euler" + } + db = self.api.db + captures = [] + for item, before_capture in enumerate((True, False)): + path = self.output_dir / f"scan-{item}.png" + path.write_bytes(b"first contents") + if before_capture: + first = self.api._ingest_scan_batch_sync([(path, metadata)]) + before = self.gallery() + run, evidence = self.capture(path.name, item) + captures.append((run, evidence)) + self.assertEqual(self.gallery(), before) + if not before_capture: + first = self.api._ingest_scan_batch_sync([(path, metadata)]) + self.assertEqual( + first, + { + "processed": 1, + "found": 1, + "added": 1, + "linked": 0, + "enriched": 0, + "unchanged": 0, + "ambiguous": 0, + "unresolved": 0, + "conflict": 0, + "failed": 0, + "pairs_added": int(before_capture), + "pair_links_added": 1, + }, + ) + for _ in range(2): + repeated = self.api._ingest_scan_batch_sync([(path, metadata)]) + self.assertEqual( + repeated, + { + "processed": 1, + "found": 1, + "added": 0, + "linked": 0, + "enriched": 0, + "unchanged": 1, + "ambiguous": 0, + "unresolved": 0, + "conflict": 0, + "failed": 0, + "pairs_added": 0, + "pair_links_added": 0, + }, + ) + path.write_bytes(b"replacement bytes at the same relative path") + conflict = self.api._ingest_scan_batch_sync( + [ + ( + path, + {"parameters": metadata["parameters"].replace("blur", "noise")}, + ) + ] + ) + self.assertEqual(conflict["conflict"], 1) + self.assertEqual(conflict["added"], 0) + self.assertEqual(db.list_prompt_pairs()["total"], 1) + self.assert_evidence(run, evidence) + pair = db.list_prompt_pairs()["pairs"][0] + self.assertEqual((pair["run_count"], pair["image_count"]), (0, 2)) + self.assertEqual( + [p["run_count"] for p in db.get_recent_prompts()["prompts"]], [0, 0] + ) + for run, evidence in captures: + self.assert_evidence(run, evidence) + + def test_same_relative_path_across_roots_never_retargets(self): + second_root = self.output_dir / "second-root" + second_root.mkdir() + metadata = { + "parameters": "cat\nNegative prompt: blur\nSteps: 20, Sampler: Euler" + } + records = [] + for item, root in enumerate((self.output_dir, second_root)): + path = root / "same.png" + path.write_bytes(b"image") + result = self.api._ingest_scanned_file(path, metadata, detailed=True) + self.assertEqual(result["status"], "added") + records.append(self.capture("same.png", item, root)) + self.assertNotEqual( + records[0][1]["references"][0]["root_key"], + records[1][1]["references"][0]["root_key"], + ) + self.assertEqual(self.api.db.list_prompt_pairs()["pairs"][0]["image_count"], 2) + for root in (self.output_dir, second_root): + (root / "same.png").unlink() + second_root.rmdir() + self.assertEqual(self.api.db.cleanup_missing_images(), 2) + for run, expected in records: + self.assert_evidence(run, expected) + + def test_archive_delete_prune_cleanup_and_numeric_id_reuse_keep_evidence(self): + db = self.api.db + path = self.output_dir / "first.png" + path.write_bytes(b"image") + result = self.api._ingest_scanned_file( + path, + {"parameters": "cat\nNegative prompt: blur\nSteps: 20, Sampler: Euler"}, + detailed=True, + ) + self.assertEqual( + (result["status"], result["pairs_added"], result["links_added"]), + ("added", 1, 1), + ) + image_id, pair_id = result["image_id"], result["pair_id"] + pair = db.get_prompt_pair(pair_id) + positive_id = pair["positive_prompt_id"] + run, evidence = self.capture() + with self.assertRaises(PairDependencyError): + db.delete_prompt(positive_id) + with self.assertRaises(PairDependencyError): + db.delete_prompt_pair(pair_id) + self.assertTrue(db.archive_prompt_pair(pair_id)) + self.assertEqual(db.list_prompt_pairs()["total"], 0) + self.assertEqual(db.list_prompt_pairs(include_archived=True)["total"], 1) + self.assertEqual(db.get_prompt_pair(pair_id)["run_count"], 0) + self.assert_evidence(run, evidence) + self.assertTrue(db.delete_image(image_id)) + self.assertFalse(db.delete_image(image_id)) + self.assertEqual(db.get_image_prompt_pairs(image_id), []) + self.assertTrue(db.delete_prompt_pair(pair_id)) + self.assertTrue(db.delete_prompt(positive_id)) + self.assertEqual(db.prune_orphaned_prompts(), 1) + self.assertEqual(db.prune_orphaned_prompts(), 0) + self.assertEqual(db.cleanup_missing_images(), 0) + self.assert_evidence(run, evidence) + + # Explicit IDs reproduce restored/imported numeric-ID reuse without assuming + # SQLite's allocator reuses AUTOINCREMENT identities. + conn = db.model.get_connection() + with conn: + conn.execute( + "INSERT INTO prompts(id,text,run_count) " + "VALUES(?, 'replacement prompt', 17)", + (positive_id,), + ) + conn.execute( + "INSERT INTO prompt_pairs(id,positive_prompt_id,run_count) " + "VALUES(?,?,23)", + (pair_id, positive_id), + ) + conn.execute( + "INSERT INTO generated_images " + "(id,prompt_id,image_path,filename,file_path) " + "VALUES(?,?,?,'first.png',?)", + (image_id, positive_id, str(path), str(path)), + ) + conn.execute( + "INSERT INTO image_prompt_pairs(image_id,pair_id,provenance) " + "VALUES(?,?,'{}')", + (image_id, pair_id), + ) + self.assert_evidence(run, evidence) + self.assertTrue(db.archive_prompt_pair(pair_id)) + path.unlink() + self.assertEqual(db.cleanup_missing_images(), 1) + self.assertEqual(db.get_image_prompt_pairs(image_id), []) + self.assertTrue(db.delete_prompt_pair(pair_id)) + # Standalone prompt deletion retains its original gallery cascade. + recreated = db.link_image_to_prompt(positive_id, str(path)) + self.assertGreater(recreated, 0) + self.assertTrue(db.delete_prompt(positive_id)) + self.assertIsNone(db.get_image_by_id(recreated)) + self.assert_evidence(run, evidence)