From 600a479f0b1d11eb08ffdc7bbddea04663f7cda3 Mon Sep 17 00:00:00 2001 From: Vito Sansevero Date: Sat, 17 Oct 2026 17:47:27 -0700 Subject: [PATCH] feat(history): persist immutable execution context snapshots --- database/execution_contexts.py | 329 ++++++++++++++++ database/operations.py | 8 +- tests/test_execution_context_database.py | 468 +++++++++++++++++++++++ tests/test_execution_history_restore.py | 45 +++ 4 files changed, 849 insertions(+), 1 deletion(-) create mode 100644 database/execution_contexts.py create mode 100644 tests/test_execution_context_database.py diff --git a/database/execution_contexts.py b/database/execution_contexts.py new file mode 100644 index 0000000..b21313a --- /dev/null +++ b/database/execution_contexts.py @@ -0,0 +1,329 @@ +"""Immutable execution capture and exact, epoch-fenced library navigation hints.""" + +import json +import sqlite3 +from typing import Iterator + +from utils.execution_contexts import ( + CAPTURE_REASONS, + ContextBundle, + content_digest, + validate_bundle, +) + +FIELDS = ("raw", "effective", "prepend", "append") +ROLES = ("positive", "negative") + + +def _prepare(bundle: ContextBundle) -> list[dict]: + validate_bundle(bundle) + if bundle.reason not in CAPTURE_REASONS: + raise ValueError("Invalid capture reason") + prepared = [] + complete = bool(bundle.contexts) + for context in bundle.contexts: + evidence = {"format_version": 1, "role_provenance": context.role_provenance} + snapshots = [] + for role in ROLES: + snapshot = getattr(context, role) + if snapshot is None: + snapshots.append(None) + evidence[role] = None + complete = False + continue + values = tuple(getattr(snapshot, field).value for field in FIELDS) + complete = complete and all(value is not None for value in values) + snapshots.append((content_digest("snapshot", values), values)) + evidence[role] = { + field: { + "provenance": getattr(snapshot, field).provenance, + "source": getattr(snapshot, field).source, + "reason": getattr(snapshot, field).reason, + } + for field in FIELDS + } + if all(snapshot is None for snapshot in snapshots): + raise ValueError("Context must contain a snapshot") + context_id = content_digest( + "context", tuple(s[0] if s else None for s in snapshots) + ) + encoded = json.dumps( + evidence, + ensure_ascii=False, + allow_nan=False, + separators=(",", ":"), + sort_keys=True, + ) + if len(encoded.encode("utf-8")) > 8192: + raise ValueError("Oversized context evidence") + prepared.append( + { + "anchor_id": context.anchor_id, + "context_id": context_id, + "snapshots": snapshots, + "evidence_json": encoded, + } + ) + if bundle.state == "complete" and not complete: + raise ValueError("Incomplete scalar capture") + return sorted(prepared, key=lambda item: item["anchor_id"]) + + +def _existing_row( + conn: sqlite3.Connection, table: str, columns: tuple[str, ...], values: tuple +) -> bool: + row = conn.execute( + f"SELECT {','.join(columns[1:])} FROM {table} WHERE {columns[0]}=?", + (values[0],), + ).fetchone() + if row is None: + return False + if tuple(row) != values[1:]: + raise ValueError("Execution context digest collision") + return True + + +def _content_rows(item: dict) -> Iterator[tuple[str, tuple[str, ...], tuple]]: + snapshot_ids = [] + for snapshot in item["snapshots"]: + if snapshot is None: + snapshot_ids.append(None) + continue + snapshot_id, values = snapshot + text_ids = [] + for value in values: + text_id = None if value is None else content_digest("text", (value,)) + if value is not None: + yield "execution_context_texts", ("text_id", "text_value"), ( + text_id, + value, + ) + text_ids.append(text_id) + yield "execution_prompt_snapshots", ( + "snapshot_id", + *(f"{field}_text_id" for field in FIELDS), + ), (snapshot_id, *text_ids) + snapshot_ids.append(snapshot_id) + yield "execution_contexts", ( + "context_id", + "positive_snapshot_id", + "negative_snapshot_id", + ), (item["context_id"], *snapshot_ids) + + +def _insert_exact( + conn: sqlite3.Connection, table: str, columns: tuple[str, ...], values: tuple +) -> None: + if not _existing_row(conn, table, columns, values): + names = ",".join(columns) + placeholders = ",".join("?" for _ in columns) + conn.execute(f"INSERT INTO {table} ({names}) VALUES ({placeholders})", values) + + +def _prompt_hint(conn: sqlite3.Connection, raw: str | None) -> int | None: + if raw is None: + return None + rows = conn.execute( + "SELECT id FROM prompts WHERE text=? COLLATE BINARY LIMIT 2", (raw,) + ).fetchall() + return rows[0][0] if len(rows) == 1 else None + + +def _hints( + conn: sqlite3.Connection, positive: str | None, negative: str | None +) -> tuple: + pos_id, neg_id = _prompt_hint(conn, positive), _prompt_hint(conn, negative) + pair_id = None + negative_ambiguous = ( + negative is not None + and neg_id is None + and conn.execute( + "SELECT 1 FROM prompts WHERE text=? COLLATE BINARY LIMIT 1", (negative,) + ).fetchone() + is not None + ) + if pos_id is not None and negative is not None and not negative_ambiguous: + rows = conn.execute( + "SELECT pp.id FROM prompt_pairs pp LEFT JOIN prompts n ON " + "n.id=pp.negative_prompt_id " + "WHERE pp.positive_prompt_id=? AND ((pp.negative_prompt_id IS NULL AND " + "?='') " + "OR (pp.negative_prompt_id=? AND n.text=? COLLATE BINARY)) LIMIT 2", + (pos_id, negative, neg_id, negative), + ).fetchall() + if len(rows) == 1: + pair_id = rows[0][0] + return pos_id, neg_id, pair_id + + +def _values(conn: sqlite3.Connection, snapshot_id: str | None) -> dict | None: + if snapshot_id is None: + return None + row = conn.execute( + "SELECT r.text_value,e.text_value,p.text_value,a.text_value " + "FROM execution_prompt_snapshots s " + "LEFT JOIN execution_context_texts r ON r.text_id=s.raw_text_id " + "LEFT JOIN execution_context_texts e ON e.text_id=s.effective_text_id " + "LEFT JOIN execution_context_texts p ON p.text_id=s.prepend_text_id " + "LEFT JOIN execution_context_texts a ON a.text_id=s.append_text_id " + "WHERE s.snapshot_id=?", + (snapshot_id,), + ).fetchone() + if row is None: + raise sqlite3.DatabaseError("Missing execution context snapshot") + return dict(zip(FIELDS, row)) + + +class ExecutionContextOperations: + """Compose with ExecutionHistoryOperations on PromptDatabase.""" + + def write_execution_contexts( + self, run_id: str, expected_epoch: str, bundle: ContextBundle + ) -> bool: + prepared = _prepare(bundle) + with self._execution_transaction(expected_epoch) as conn: + if ( + conn.execute( + "SELECT 1 FROM execution_runs WHERE run_id=? AND storage_epoch=?", + (run_id, expected_epoch), + ).fetchone() + is None + ): + return False + capture = conn.execute( + "SELECT state,reason,format_version FROM execution_context_capture " + "WHERE run_id=?", + (run_id,), + ).fetchone() + if capture is not None: + for item in prepared: + for table, columns, values in _content_rows(item): + _existing_row(conn, table, columns, values) + links = conn.execute( + "SELECT anchor_id,context_id,evidence_json FROM " + "execution_run_contexts WHERE run_id=? ORDER BY anchor_id", + (run_id,), + ).fetchall() + expected = [ + (item["anchor_id"], item["context_id"], item["evidence_json"]) + for item in prepared + ] + # Compare JSON objects so serialization order never creates a conflict. + original = [(row[0], row[1], json.loads(row[2])) for row in links] + proposed = [ + (anchor, context, json.loads(evidence)) + for anchor, context, evidence in expected + ] + if capture[0] != "conflicting" and ( + tuple(capture) + != (bundle.state, bundle.reason, bundle.format_version) + or original != proposed + ): + conn.execute( + "UPDATE execution_context_capture SET " + "state='conflicting',reason='conflicting' WHERE run_id=?", + (run_id,), + ) + return True + for item in prepared: + for table, columns, values in _content_rows(item): + _insert_exact(conn, table, columns, values) + raw = [ + snapshot[1][0] if snapshot else None + for snapshot in item["snapshots"] + ] + positive_id, negative_id, pair_id = _hints(conn, *raw) + conn.execute( + "INSERT INTO " + "execution_run_contexts(run_id,anchor_id,context_id,evidence_json," + "positive_prompt_id,negative_prompt_id,pair_id,reference_epoch," + "capture_version) VALUES(?,?,?,?,?,?,?,?,1)", + ( + run_id, + item["anchor_id"], + item["context_id"], + item["evidence_json"], + positive_id, + negative_id, + pair_id, + expected_epoch, + ), + ) + conn.execute( + "INSERT INTO " + "execution_context_capture(run_id,state,reason,format_version) " + "VALUES(?,?,?,?)", + (run_id, bundle.state, bundle.reason, bundle.format_version), + ) + return True + + def get_execution_contexts(self, run_id: str) -> dict: + """Return first-capture scalars; hints are None unless currently valid. + + reference_epoch remains historical evidence even when hints are invalidated. + Reads never rebind a first-capture reference or alter stored capture data. + """ + conn = self.model.get_connection() + own_transaction = not conn.in_transaction + if own_transaction: + conn.execute("BEGIN") + try: + capture = conn.execute( + "SELECT state,reason,format_version FROM execution_context_capture " + "WHERE run_id=?", + (run_id,), + ).fetchone() + if capture is None: + return { + "state": "not_captured", + "reason": "missing", + "format_version": None, + "contexts": [], + } + epoch = self.get_execution_epoch() + contexts = [] + rows = conn.execute( + "SELECT r.*,c.positive_snapshot_id,c.negative_snapshot_id FROM " + "execution_run_contexts r JOIN execution_contexts c ON " + "c.context_id=r.context_id WHERE r.run_id=? ORDER BY r.anchor_id", + (run_id,), + ) + for row in rows: + positive = _values(conn, row["positive_snapshot_id"]) + negative = _values(conn, row["negative_snapshot_id"]) + hints = (None, None, None) + if row["reference_epoch"] == epoch: + current = _hints( + conn, + positive["raw"] if positive else None, + negative["raw"] if negative else None, + ) + hints = tuple( + row[key] if row[key] == value else None + for key, value in zip( + ("positive_prompt_id", "negative_prompt_id", "pair_id"), + current, + ) + ) + contexts.append( + { + "anchor_id": row["anchor_id"], + "context_id": row["context_id"], + "positive": positive, + "negative": negative, + "evidence": json.loads(row["evidence_json"]), + "positive_prompt_id": hints[0], + "negative_prompt_id": hints[1], + "pair_id": hints[2], + "reference_epoch": row["reference_epoch"], + } + ) + return { + "state": capture[0], + "reason": capture[1], + "format_version": capture[2], + "contexts": contexts, + } + finally: + if own_transaction: + conn.rollback() diff --git a/database/operations.py b/database/operations.py index f927196..92c1ead 100644 --- a/database/operations.py +++ b/database/operations.py @@ -12,6 +12,7 @@ from .models import PromptModel, normalize_image_path, utc_now_iso from .pairs import PairDependencyError, PairOperations from .library import LibraryOperations from .execution_history import ExecutionHistoryOperations +from .execution_contexts import ExecutionContextOperations # Import logging system try: @@ -152,7 +153,12 @@ def _coerce_file_info(metadata: Any) -> tuple: ) -class PromptDatabase(PairOperations, LibraryOperations, ExecutionHistoryOperations): +class PromptDatabase( + PairOperations, + LibraryOperations, + ExecutionHistoryOperations, + ExecutionContextOperations, +): """Database operations class for managing prompts.""" def __init__(self, db_path: Optional[str] = None): diff --git a/tests/test_execution_context_database.py b/tests/test_execution_context_database.py new file mode 100644 index 0000000..bdc0f47 --- /dev/null +++ b/tests/test_execution_context_database.py @@ -0,0 +1,468 @@ +"""Immutable scalar capture, deduplication, and conservative navigation hints.""" + +from concurrent.futures import ThreadPoolExecutor +from dataclasses import replace +from pathlib import Path +from unittest.mock import patch +import unittest + +from database.execution_history import HistoryEpochChanged +from database.execution_context_schema import context_schema_error +from utils.execution_contexts import ( + ContextBundle, + PromptSnapshot, + RecipeContext, + TextEvidence, +) + +try: + from tests import test_execution_history_database as history_tests +except ImportError: + import test_execution_history_database as history_tests + + +def known(value): + return TextEvidence(value, "recorded", "runtime_return", None) + + +def snapshot(raw): + return PromptSnapshot(known(raw), known(raw), known(""), known("")) + + +def bundle(negative="negative", positive="positive", anchor="4"): + return ContextBundle( + (RecipeContext(anchor, snapshot(positive), snapshot(negative), "inferred"),), + "complete", + "captured", + ) + + +class ContextDatabaseTests(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, item=0, capture=None): + start = self.start(db, item) + db.write_execution_start(start) + db.write_execution_finish(self.finish(start)) + self.assertTrue( + db.write_execution_contexts( + start.run_id, start.storage_epoch, capture or bundle() + ) + ) + return start + + def counts(self, db): + conn = db.model.get_connection() + return tuple( + conn.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0] + for table in ( + "execution_runs", + "execution_contexts", + "execution_prompt_snapshots", + "execution_context_capture", + ) + ) + + def test_a_a_b_cardinality(self): + db = self.database() + self.assertTrue( + callable(getattr(db, "write_execution_contexts", None)), + "PromptDatabase must expose immutable context persistence", + ) + starts = [self.capture(db, i, bundle(n)) for i, n in enumerate(("A", "A", "B"))] + self.assertEqual(self.counts(db), (3, 2, 3, 3)) + reads = [db.get_execution_contexts(s.run_id) for s in starts] + self.assertEqual( + reads[0]["contexts"][0]["context_id"], reads[1]["contexts"][0]["context_id"] + ) + self.assertNotEqual( + reads[0]["contexts"][0]["context_id"], reads[2]["contexts"][0]["context_id"] + ) + self.assertIsNone(context_schema_error(db.model.get_connection())) + + def test_parallel_identical_inserts_deduplicate(self): + db = self.database() + starts = [self.start(db, i) for i in range(8)] + for start in starts: + db.write_execution_start(start) + + def write(start): + return db.write_execution_contexts( + start.run_id, start.storage_epoch, bundle() + ) + + with ThreadPoolExecutor(max_workers=4) as pool: + self.assertEqual(list(pool.map(write, starts)), [True] * 8) + self.assertEqual(self.counts(db), (8, 1, 2, 8)) + + def test_replay_freezes_first_capture(self): + db = self.database() + start = self.capture(db) + original = db.get_execution_contexts(start.run_id) + counts = self.counts(db) + self.assertTrue( + db.write_execution_contexts(start.run_id, start.storage_epoch, bundle()) + ) + db.write_execution_finish(self.finish(start)) + self.assertEqual(db.get_execution_contexts(start.run_id), original) + self.assertTrue( + db.write_execution_contexts( + start.run_id, start.storage_epoch, bundle("changed") + ) + ) + read = db.get_execution_contexts(start.run_id) + self.assertEqual( + (read["state"], read["reason"]), ("conflicting", "conflicting") + ) + self.assertEqual(read["contexts"], original["contexts"]) + self.assertEqual(self.counts(db), counts) + db.write_execution_contexts(start.run_id, start.storage_epoch, bundle()) + self.assertEqual(db.get_execution_contexts(start.run_id), read) + + def test_evidence_replay_and_anchor_changes_conflict_without_inserting(self): + for change in ("evidence", "anchor", "status"): + with self.subTest(change=change): + db = self.database(str(Path(self.temp.name) / f"{change}.db")) + start = self.capture(db) + original = db.get_execution_contexts(start.run_id)["contexts"] + candidate = bundle() + if change == "evidence": + ctx = candidate.contexts[0] + candidate = replace( + candidate, + contexts=( + replace( + ctx, + positive=replace( + ctx.positive, + raw=TextEvidence( + "positive", "inferred", "graph_literal", None + ), + ), + ), + ), + ) + elif change == "anchor": + candidate = bundle(anchor="5") + else: + candidate = replace(candidate, state="partial", reason="partial") + db.write_execution_contexts( + start.run_id, start.storage_epoch, candidate + ) + read = db.get_execution_contexts(start.run_id) + self.assertEqual(read["state"], "conflicting") + self.assertEqual(read["contexts"], original) + self.assertEqual(self.counts(db), (1, 1, 2, 1)) + + def test_hash_collision_rejected_without_partial_insert(self): + db = self.database() + start = self.capture(db) + original = db.get_execution_contexts(start.run_id) + second = self.start(db, 1) + db.write_execution_start(second) + # A collision must compare stored scalar content, then roll back all rows. + from utils.execution_contexts import content_digest + + collided = content_digest("text", ("negative",)) + + def digest(kind, fields): + return ( + collided + if kind == "text" and fields == ("changed",) + else content_digest(kind, fields) + ) + + before = tuple(db.model.get_connection().iterdump()) + with patch("database.execution_contexts.content_digest", side_effect=digest): + with self.assertRaises(ValueError): + db.write_execution_contexts( + second.run_id, + second.storage_epoch, + bundle("changed", "new-positive"), + ) + self.assertEqual(tuple(db.model.get_connection().iterdump()), before) + self.assertEqual(db.get_execution_contexts(start.run_id), original) + + def test_missing_parent_and_epoch_mismatch_insert_nothing(self): + db = self.database() + start = self.start(db) + self.assertEqual( + db.get_execution_contexts(start.run_id)["state"], "not_captured" + ) + self.assertFalse( + db.write_execution_contexts(start.run_id, start.storage_epoch, bundle()) + ) + self.assertEqual(self.counts(db), (0, 0, 0, 0)) + db.write_execution_start(start) + with self.assertRaises(HistoryEpochChanged): + db.write_execution_contexts( + start.run_id, "11111111-1111-4111-8111-111111111111", bundle() + ) + with db.model.get_connection() as conn: + conn.execute( + "UPDATE execution_runs SET storage_epoch=? WHERE run_id=?", + ("11111111-1111-4111-8111-111111111111", start.run_id), + ) + self.assertFalse( + db.write_execution_contexts(start.run_id, start.storage_epoch, bundle()) + ) + self.assertEqual(self.counts(db), (1, 0, 0, 0)) + + def test_invalid_bundle_has_no_partial_rows(self): + db = self.database() + start = self.start(db) + db.write_execution_start(start) + for candidate in ( + replace(bundle(), reason="arbitrary"), + replace(bundle(), contexts=()), + replace(bundle(), contexts=(RecipeContext("4", None, None, "inferred"),)), + ): + with self.subTest(candidate=candidate), self.assertRaises(ValueError): + db.write_execution_contexts( + start.run_id, start.storage_epoch, candidate + ) + self.assertEqual(self.counts(db), (1, 0, 0, 0)) + + def test_unknown_empty_unicode_and_evidence_round_trip(self): + db = self.database() + unknown = TextEvidence(None, "unknown", "unknown", "missing") + ctx = RecipeContext( + "3", + PromptSnapshot(known(" e\u0301 \n"), unknown, known(""), known("é")), + None, + "inferred", + ) + start = self.capture(db, capture=ContextBundle((ctx,), "partial", "missing")) + read = db.get_execution_contexts(start.run_id)["contexts"][0] + self.assertEqual( + read["positive"], + {"raw": " e\u0301 \n", "effective": None, "prepend": "", "append": "é"}, + ) + self.assertIsNone(read["negative"]) + self.assertEqual( + read["evidence"]["positive"]["effective"], + {"provenance": "unknown", "source": "unknown", "reason": "missing"}, + ) + self.assertEqual( + set(read["evidence"]), + {"format_version", "role_provenance", "positive", "negative"}, + ) + self.assertIsNone(context_schema_error(db.model.get_connection())) + + def test_library_mutations_and_id_reuse_do_not_rewrite_history(self): + db = self.database() + conn = db.model.get_connection() + with conn: + conn.execute( + "INSERT INTO prompts(id,text) VALUES(1,'positive'),(2,'negative')" + ) + conn.execute( + "INSERT INTO prompt_pairs(id,positive_prompt_id,negative_prompt_id) " + "VALUES(7,1,2)" + ) + start = self.capture(db) + original = db.get_execution_contexts(start.run_id)["contexts"][0] + self.assertEqual( + ( + original["positive_prompt_id"], + original["negative_prompt_id"], + original["pair_id"], + ), + (1, 2, 7), + ) + db.archive_prompt_pair(7) + self.assertEqual( + db.get_execution_contexts(start.run_id)["contexts"][0], original + ) + db.delete_prompt_pair(7) + db.update_prompt_text(1, "edited") + read = db.get_execution_contexts(start.run_id)["contexts"][0] + self.assertIsNone(read["positive_prompt_id"]) + self.assertIsNone(read["pair_id"]) + db.delete_prompt(2) + with conn: + conn.execute("INSERT INTO prompts(id,text) VALUES(2,'reused')") + self.assertIsNone( + db.get_execution_contexts(start.run_id)["contexts"][0]["negative_prompt_id"] + ) + self.assertEqual(db.prune_orphaned_prompts(), 2) + after = db.get_execution_contexts(start.run_id)["contexts"][0] + for key in ("positive", "negative", "evidence", "context_id"): + self.assertEqual(after[key], original[key]) + self.assertEqual(conn.execute("SELECT COUNT(*) FROM prompts").fetchone()[0], 0) + self.assertIsNone(context_schema_error(conn)) + + def test_hints_require_exact_unique_raw_text_and_never_refresh(self): + db = self.database() + conn = db.model.get_connection() + with conn: + conn.execute( + "INSERT INTO prompts(id,text) VALUES(1,' positive " + "'),(2,'negative'),(3,'negative')" + ) + start = self.capture(db) + read = db.get_execution_contexts(start.run_id)["contexts"][0] + self.assertIsNone(read["positive_prompt_id"]) + self.assertIsNone(read["negative_prompt_id"]) + with conn: + conn.execute("DELETE FROM prompts WHERE id=3") + conn.execute("UPDATE prompts SET text='positive' WHERE id=1") + db.write_execution_contexts(start.run_id, start.storage_epoch, bundle()) + self.assertEqual(db.get_execution_contexts(start.run_id)["contexts"], [read]) + fresh = self.capture(db, 1) + self.assertEqual( + db.get_execution_contexts(fresh.run_id)["contexts"][0][ + "positive_prompt_id" + ], + 1, + ) + with conn: + conn.execute("INSERT INTO prompts(id,text) VALUES(3,'positive')") + self.assertIsNone( + db.get_execution_contexts(fresh.run_id)["contexts"][0]["positive_prompt_id"] + ) + + def test_null_negative_pair_only_matches_known_exact_empty(self): + db = self.database() + with db.model.get_connection() as conn: + conn.execute("INSERT INTO prompts(id,text) VALUES(1,'positive')") + conn.execute("INSERT INTO prompt_pairs(id,positive_prompt_id) VALUES(7,1)") + for item, raw, expected in ((0, "", 7), (1, " ", None), (2, None, None)): + candidate = bundle(negative=raw or "") + if raw is None: + ctx = candidate.contexts[0] + candidate = replace( + candidate, + contexts=( + replace( + ctx, + negative=replace( + ctx.negative, + raw=TextEvidence(None, "unknown", "unknown", "missing"), + ), + ), + ), + state="partial", + reason="partial", + ) + start = self.capture(db, item, candidate) + self.assertEqual( + db.get_execution_contexts(start.run_id)["contexts"][0]["pair_id"], + expected, + ) + + def test_snapshot_and_context_collisions_roll_back_all_new_rows(self): + from utils.execution_contexts import content_digest + + for kind in ("snapshot", "context"): + with self.subTest(kind=kind): + db = self.database(str(Path(self.temp.name) / f"{kind}.db")) + first = self.capture(db) + old_context = db.get_execution_contexts(first.run_id)["contexts"][0][ + "context_id" + ] + old_snapshot = ( + db.model.get_connection() + .execute( + "SELECT negative_snapshot_id FROM execution_contexts WHERE " + "context_id=?", + (old_context,), + ) + .fetchone()[0] + ) + second = self.start(db, 1) + db.write_execution_start(second) + before = tuple(db.model.get_connection().iterdump()) + + def digest(current_kind, fields): + if current_kind == kind: + return old_snapshot if kind == "snapshot" else old_context + return content_digest(current_kind, fields) + + with ( + patch( + "database.execution_contexts.content_digest", side_effect=digest + ), + self.assertRaises(ValueError), + ): + db.write_execution_contexts( + second.run_id, + second.storage_epoch, + bundle("changed", "new-positive"), + ) + self.assertEqual(tuple(db.model.get_connection().iterdump()), before) + + def test_parallel_contradictory_replay_keeps_one_first_capture(self): + db = self.database() + start = self.start(db) + db.write_execution_start(start) + + def write(negative): + return db.write_execution_contexts( + start.run_id, start.storage_epoch, bundle(negative) + ) + + with ThreadPoolExecutor(max_workers=2) as pool: + self.assertEqual(list(pool.map(write, ("A", "B"))), [True, True]) + read = db.get_execution_contexts(start.run_id) + self.assertEqual(read["state"], "conflicting") + self.assertIn(read["contexts"][0]["negative"]["raw"], ("A", "B")) + self.assertEqual(self.counts(db), (1, 1, 2, 1)) + + def test_merge_never_rebinds_a_hint_to_normalized_equal_primary(self): + db = self.database() + with db.model.get_connection() as conn: + conn.execute( + "INSERT INTO prompts(id,text,created_at) VALUES(1,' POSITIVE " + "','2020-01-01'),(2,'positive','2021-01-01')" + ) + start = self.capture(db) + before = db.get_execution_contexts(start.run_id)["contexts"][0] + self.assertEqual(before["positive_prompt_id"], 2) + self.assertEqual(db.cleanup_duplicates(), 1) + after = db.get_execution_contexts(start.run_id)["contexts"][0] + self.assertIsNone(after["positive_prompt_id"]) + self.assertEqual(after["positive"], before["positive"]) + self.assertEqual(after["evidence"], before["evidence"]) + self.assertEqual( + db.model.get_connection().execute("SELECT text FROM prompts").fetchone()[0], + " POSITIVE ", + ) + + def test_ambiguous_empty_negative_omits_pair_hint(self): + db = self.database() + with db.model.get_connection() as conn: + conn.execute( + "INSERT INTO prompts(id,text) VALUES(1,'positive'),(2,''),(3,'')" + ) + conn.execute("INSERT INTO prompt_pairs(id,positive_prompt_id) VALUES(7,1)") + start = self.capture(db, capture=bundle(negative="")) + read = db.get_execution_contexts(start.run_id)["contexts"][0] + self.assertIsNone(read["negative_prompt_id"]) + self.assertIsNone(read["pair_id"]) + + def test_replay_checks_scalar_content_even_if_digests_collide(self): + db = self.database() + start = self.capture(db) + before = tuple(db.model.get_connection().iterdump()) + from utils.execution_contexts import content_digest + + original_snapshot = content_digest("snapshot", ("negative", "negative", "", "")) + + def digest(kind, fields): + if kind == "snapshot" and fields == ("changed", "changed", "", ""): + return original_snapshot + return content_digest(kind, fields) + + with ( + patch("database.execution_contexts.content_digest", side_effect=digest), + self.assertRaises(ValueError), + ): + db.write_execution_contexts( + start.run_id, start.storage_epoch, bundle("changed") + ) + self.assertEqual(tuple(db.model.get_connection().iterdump()), before) diff --git a/tests/test_execution_history_restore.py b/tests/test_execution_history_restore.py index 3ec3ac5..0eba0e6 100644 --- a/tests/test_execution_history_restore.py +++ b/tests/test_execution_history_restore.py @@ -324,6 +324,51 @@ class HistoryRestoreTests(unittest.TestCase): self.assertNotEqual(db.get_execution_epoch(), original_epoch) self.assertEqual(notices, [True]) + def test_restore_preserves_context_scalars_but_never_reepochs_hints(self): + try: + from tests.test_execution_context_database import bundle + except ImportError: + from test_execution_context_database import bundle + from database.execution_history import HistoryEpochChanged + + db = self.database() + with db.model.get_connection() as conn: + conn.execute( + "INSERT INTO prompts(id,text) VALUES(1,'positive'),(2,'negative')" + ) + conn.execute( + "INSERT INTO prompt_pairs(id,positive_prompt_id,negative_prompt_id) " + "VALUES(7,1,2)" + ) + start = self.start(db) + db.write_execution_start(start) + db.write_execution_contexts(start.run_id, start.storage_epoch, bundle()) + original = db.get_execution_contexts(start.run_id) + backup = str(Path(self.temp.name) / "contexts.db") + self.assertTrue(db.model.backup_database(backup)) + source_bytes = Path(backup).read_bytes() + for _ in range(2): + db.model.restore_from_file(backup) + read = db.get_execution_contexts(start.run_id) + expected = original["contexts"][0].copy() + expected.update( + positive_prompt_id=None, negative_prompt_id=None, pair_id=None + ) + self.assertEqual(read["contexts"], [expected]) + self.assertEqual(read["state"], "complete") + self.assertEqual( + read["contexts"][0]["reference_epoch"], start.storage_epoch + ) + with self.assertRaises(HistoryEpochChanged): + db.write_execution_contexts(start.run_id, start.storage_epoch, bundle()) + self.assertFalse( + db.write_execution_contexts( + start.run_id, db.get_execution_epoch(), bundle() + ) + ) + self.assertEqual(db.get_execution_contexts(start.run_id), read) + self.assertEqual(Path(backup).read_bytes(), source_bytes) + if __name__ == "__main__": unittest.main()