feat(history): persist immutable run output references
This commit is contained in:
Vito Sansevero
@@ -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()
|
||||
@@ -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."""
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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, [])
|
||||
@@ -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)
|
||||
Reference in New Issue
Block a user