diff --git a/database/execution_context_schema.py b/database/execution_context_schema.py new file mode 100644 index 0000000..fe0fa09 --- /dev/null +++ b/database/execution_context_schema.py @@ -0,0 +1,380 @@ +"""Owned execution-context v1 tables within execution-history schema v2. + +Content IDs hash exact scalar values; evidence JSON contains metadata only. +All validation scans use cursors and bounded per-row payloads. +""" + +import json +import sqlite3 + +from database.execution_history_schema import _compact, _uuid_check +from utils.execution_contexts import ( + BUNDLE_STATES, + CAPTURE_REASONS, + EVIDENCE_SOURCES, + UNKNOWN_REASONS, + MAX_CONTEXT_BYTES, + ContextBundle, + PromptSnapshot, + RecipeContext, + TextEvidence, + validate_bundle, + content_digest, + text_size, +) + +FIELDS = ("raw", "effective", "prepend", "append") +TABLE_COLUMNS = { + "execution_context_texts": ( + ("text_id", "TEXT", 1, 1), + ("text_value", "TEXT", 1, 0), + ), + "execution_prompt_snapshots": ( + ("snapshot_id", "TEXT", 1, 1), + *((f"{field}_text_id", "TEXT", 0, 0) for field in FIELDS), + ), + "execution_contexts": ( + ("context_id", "TEXT", 1, 1), + ("positive_snapshot_id", "TEXT", 0, 0), + ("negative_snapshot_id", "TEXT", 0, 0), + ), + "execution_run_contexts": ( + ("run_id", "TEXT", 1, 1), + ("anchor_id", "TEXT", 1, 2), + ("context_id", "TEXT", 1, 0), + ("evidence_json", "TEXT", 1, 0), + ("positive_prompt_id", "INTEGER", 0, 0), + ("negative_prompt_id", "INTEGER", 0, 0), + ("pair_id", "INTEGER", 0, 0), + ("reference_epoch", "TEXT", 0, 0), + ("capture_version", "INTEGER", 1, 0), + ), + "execution_context_capture": ( + ("run_id", "TEXT", 1, 1), + ("state", "TEXT", 1, 0), + ("reason", "TEXT", 1, 0), + ("format_version", "INTEGER", 1, 0), + ), +} + + +def _digest_check(column: str, nullable: bool = False) -> str: + check = ( + f"typeof({column}) = 'text' AND length({column}) = 64 " + f"AND {column} NOT GLOB '*[^0-9a-f]*'" + ) + return f"{column} IS NULL OR ({check})" if nullable else check + + +def _enum_check(column: str, values: frozenset[str]) -> str: + return ( + f"typeof({column}) = 'text' AND {column} IN (" + + ",".join("'" + value + "'" for value in sorted(values)) + + ")" + ) + + +TABLE_CHECKS = { + "execution_context_texts": ( + _digest_check("text_id"), + "typeof(text_value) = 'text' AND length(CAST(text_value AS BLOB)) <= 65536", + ), + "execution_prompt_snapshots": ( + _digest_check("snapshot_id"), + *(_digest_check(f"{field}_text_id", True) for field in FIELDS), + ), + "execution_contexts": ( + _digest_check("context_id"), + _digest_check("positive_snapshot_id", True), + _digest_check("negative_snapshot_id", True), + "positive_snapshot_id IS NOT NULL OR negative_snapshot_id IS NOT NULL", + ), + "execution_run_contexts": ( + _uuid_check("run_id"), + "typeof(anchor_id) = 'text' AND length(anchor_id) BETWEEN 1 AND 128", + _digest_check("context_id"), + "typeof(evidence_json) = 'text' AND length(CAST(evidence_json AS " + "BLOB)) <= 8192", + *( + f"{column} IS NULL OR (typeof({column}) = 'integer' AND {column} > 0)" + for column in ("positive_prompt_id", "negative_prompt_id", "pair_id") + ), + f"reference_epoch IS NULL OR ({_uuid_check('reference_epoch')})", + "typeof(capture_version) = 'integer' AND capture_version = 1", + ), + "execution_context_capture": ( + _uuid_check("run_id"), + _enum_check("state", BUNDLE_STATES), + _enum_check("reason", CAPTURE_REASONS), + "length(CAST(reason AS BLOB)) <= 64", + "typeof(format_version) = 'integer' AND format_version = 1", + ), +} +# (source column, destination table, destination column); actions always NO ACTION. +TABLE_FOREIGN_KEYS = { + "execution_context_texts": (), + "execution_prompt_snapshots": tuple( + (f"{field}_text_id", "execution_context_texts", "text_id") for field in FIELDS + ), + "execution_contexts": tuple( + (f"{role}_snapshot_id", "execution_prompt_snapshots", "snapshot_id") + for role in ("positive", "negative") + ), + "execution_run_contexts": ( + ("run_id", "execution_runs", "run_id"), + ("context_id", "execution_contexts", "context_id"), + ), + "execution_context_capture": (("run_id", "execution_runs", "run_id"),), +} +# Only SQLite-generated primary-key indexes are unique. No secondary indexes required. + + +def install_context_schema(conn: sqlite3.Connection) -> None: + """Install tables inside the caller's transaction, without committing.""" + for table, columns in TABLE_COLUMNS.items(): + definitions = [ + f"{name} {kind}{' NOT NULL' if required else ''}" + for name, kind, required, _ in columns + ] + primary = [name for name, _, _, order in columns if order] + definitions.append(f"PRIMARY KEY ({', '.join(primary)})") + definitions.extend(f"CHECK({check})" for check in TABLE_CHECKS[table]) + definitions.extend( + f"FOREIGN KEY ({column}) REFERENCES {target}({key}) " + "ON UPDATE NO ACTION ON DELETE NO ACTION" + for column, target, key in TABLE_FOREIGN_KEYS[table] + ) + conn.execute(f"CREATE TABLE {table} ({', '.join(definitions)})") + + +def _unique_object(pairs: list[tuple[str, object]]) -> dict: + result = {} + for key, value in pairs: + if key in result: + raise ValueError("Duplicate evidence key") + result[key] = value + return result + + +def _snapshot_values(conn: sqlite3.Connection, snapshot_id: str) -> tuple: + row = conn.execute( + "SELECT raw_text_id,effective_text_id,prepend_text_id,append_text_id " + "FROM execution_prompt_snapshots WHERE snapshot_id=?", + (snapshot_id,), + ).fetchone() + if row is None: + raise ValueError("Missing snapshot") + values = [] + for text_id in row: + if text_id is None: + values.append(None) + else: + text = conn.execute( + "SELECT text_value FROM execution_context_texts WHERE text_id=?", + (text_id,), + ).fetchone() + if text is None: + raise ValueError("Missing context text") + text_size(text[0]) + values.append(text[0]) + return tuple(values) + + +def _validate_evidence( + conn: sqlite3.Connection, anchor_id: str, context_id: str, encoded: str +) -> tuple[RecipeContext, bool]: + text_size(encoded, 8192) + evidence = json.loads(encoded, object_pairs_hook=_unique_object) + if ( + type(evidence) is not dict + or set(evidence) + != {"format_version", "role_provenance", "positive", "negative"} + or type(evidence["format_version"]) is not int + or evidence["format_version"] != 1 + or evidence["role_provenance"] != "inferred" + ): + raise ValueError("Invalid context evidence") + row = conn.execute( + "SELECT positive_snapshot_id,negative_snapshot_id FROM " + "execution_contexts WHERE context_id=?", + (context_id,), + ).fetchone() + if row is None: + raise ValueError("Missing context") + complete = True + snapshots = [] + for role, snapshot_id in zip(("positive", "negative"), row): + metadata = evidence[role] + if snapshot_id is None: + if metadata is not None: + raise ValueError("Unexpected role evidence") + complete = False + snapshots.append(None) + continue + values = _snapshot_values(conn, snapshot_id) + if type(metadata) is not dict or set(metadata) != set(FIELDS): + raise ValueError("Invalid snapshot evidence") + fields = [] + for field, value in zip(FIELDS, values): + item = metadata[field] + if type(item) is not dict or set(item) != { + "provenance", + "source", + "reason", + }: + raise ValueError("Invalid field evidence") + provenance, source, reason = ( + item["provenance"], + item["source"], + item["reason"], + ) + if ( + type(provenance) is not str + or type(source) is not str + or source not in EVIDENCE_SOURCES.get(provenance, ()) + ): + raise ValueError("Invalid evidence source") + if value is None: + complete = False + if ( + provenance != "unknown" + or type(reason) is not str + or reason not in UNKNOWN_REASONS + ): + raise ValueError("Unknown evidence requires reason") + elif provenance == "unknown" or reason is not None: + raise ValueError("Known evidence metadata mismatch") + fields.append(TextEvidence(value, provenance, source, reason)) + snapshots.append(PromptSnapshot(*fields)) + return RecipeContext(anchor_id, *snapshots, evidence["role_provenance"]), complete + + +def context_schema_error(conn: sqlite3.Connection) -> str | None: + """Validate exact owned schema, bounded evidence, fingerprints and links.""" + if conn.execute( + "SELECT 1 FROM sqlite_master WHERE type IN ('trigger','view') LIMIT 1" + ).fetchone(): + return "Unsupported execution context trigger/view" + tables = { + row[0]: row[1] + for row in conn.execute("SELECT name,sql FROM sqlite_master WHERE type='table'") + if row[0] in TABLE_COLUMNS + } + if tables.keys() != TABLE_COLUMNS.keys(): + return "Incomplete execution context schema" + for table, expected in TABLE_COLUMNS.items(): + columns = tuple( + (row[1], row[2].upper(), row[3], row[5]) + for row in conn.execute(f"PRAGMA table_xinfo({table})") + ) + if columns != expected: + return f"{table} has invalid execution context columns" + sql = _compact(tables[table]) + checks = TABLE_CHECKS[table] + if sql.count("check(") != len(checks) or any( + f"check({_compact(check)})" not in sql for check in checks + ): + return f"{table} has invalid execution context checks" + foreign = tuple( + sorted( + (row[3], row[2], row[4], row[5], row[6], row[7]) + for row in conn.execute(f"PRAGMA foreign_key_list({table})") + ) + ) + expected_foreign = tuple( + sorted( + (*key, "NO ACTION", "NO ACTION", "NONE") + for key in TABLE_FOREIGN_KEYS[table] + ) + ) + if foreign != expected_foreign: + return f"{table} has invalid execution context foreign keys" + for index in conn.execute(f"PRAGMA index_list({table})"): + if not index[2]: + continue + name = index[1].replace('"', '""') + keys = tuple( + row for row in conn.execute(f'PRAGMA index_xinfo("{name}")') if row[5] + ) + primary = tuple(column[0] for column in expected if column[3]) + if ( + index[3] != "pk" + or index[4] + or tuple(key[2] for key in keys) != primary + or any(key[3] or key[4] != "BINARY" for key in keys) + ): + return f"{table} has unexpected execution context uniqueness" + predicate = " AND ".join(f"({check})" for check in checks) + if conn.execute( + f"SELECT 1 FROM {table} WHERE NOT ({predicate}) LIMIT 1" + ).fetchone(): + return f"{table} contains invalid execution context records" + if conn.execute(f"PRAGMA foreign_key_check({table})").fetchone(): + return f"{table} contains orphan execution context records" + if conn.execute( + "SELECT 1 FROM execution_run_contexts GROUP BY run_id HAVING " + "COUNT(*) > 16 LIMIT 1" + ).fetchone(): + return "Too many execution contexts per run" + try: + for text_id, text in conn.execute( + "SELECT text_id,text_value FROM execution_context_texts" + ): + text_size(text) + if content_digest("text", (text,)) != text_id: + raise ValueError("Invalid text fingerprint") + for (snapshot_id,) in conn.execute( + "SELECT snapshot_id FROM execution_prompt_snapshots" + ): + if ( + content_digest("snapshot", _snapshot_values(conn, snapshot_id)) + != snapshot_id + ): + raise ValueError("Invalid snapshot fingerprint") + for context_id, positive, negative in conn.execute( + "SELECT context_id,positive_snapshot_id,negative_snapshot_id FROM " + "execution_contexts" + ): + if content_digest("context", (positive, negative)) != context_id: + raise ValueError("Invalid context fingerprint") + payload_size = 0 + for snapshot_id in (positive, negative): + if snapshot_id is not None: + payload_size += sum( + text_size(value) + for value in _snapshot_values(conn, snapshot_id) + if value is not None + ) + if payload_size > MAX_CONTEXT_BYTES: + raise ValueError("Oversized detached context") + previous_run = None + contexts = () + for run_id, anchor_id, context_id, evidence in conn.execute( + "SELECT run_id,anchor_id,context_id,evidence_json FROM " + "execution_run_contexts ORDER BY run_id,anchor_id" + ): + if run_id != previous_run: + contexts = () + previous_run = run_id + context, complete = _validate_evidence( + conn, anchor_id, context_id, evidence + ) + status = conn.execute( + "SELECT state,reason FROM execution_context_capture WHERE run_id=?", + (run_id,), + ).fetchone() + state, reason = status if status else ("partial", "") + if state == "complete" and not complete: + raise ValueError("Incomplete evidence marked complete") + candidate = (*contexts, context) + validate_bundle(ContextBundle(candidate, state, reason)) + contexts = candidate + if conn.execute( + "SELECT 1 FROM execution_context_capture c WHERE state='complete' " + "AND NOT EXISTS (SELECT 1 FROM execution_run_contexts r WHERE " + "r.run_id=c.run_id) LIMIT 1" + ).fetchone(): + raise ValueError("Complete capture has no contexts") + except (ValueError, TypeError, RecursionError, OverflowError) as exc: + return f"Invalid execution context content: {type(exc).__name__}" + return None diff --git a/database/execution_history_schema.py b/database/execution_history_schema.py index 562648f..10a8d86 100644 --- a/database/execution_history_schema.py +++ b/database/execution_history_schema.py @@ -4,7 +4,7 @@ import re import sqlite3 from uuid import uuid4 -SCHEMA_VERSION = 1 +SCHEMA_VERSION = 2 def _uuid_check(column: str) -> str: @@ -24,7 +24,7 @@ def _integer_check(column: str, *, nullable: bool = False) -> str: _METADATA_CHECKS = ( "singleton_id = 1", - "schema_version = 1", + "schema_version = 2", _uuid_check("storage_epoch"), ) _RUN_CHECKS = ( @@ -44,13 +44,13 @@ _RUN_CHECKS = ( ) ), "outcome IN ('observed', 'succeeded', 'failed', 'interrupted', 'unknown')", - "(outcome = 'observed' AND reason IS NULL AND observed_callback_at IS NULL) OR " - "(outcome != 'observed' AND reason IS NOT NULL " - "AND observed_callback_at IS NOT NULL AND " + "(outcome = 'observed' AND reason IS NULL AND " + "observed_callback_at IS NULL) OR (outcome != 'observed' AND " + "reason IS NOT NULL AND observed_callback_at IS NOT NULL AND " "((outcome = 'succeeded' AND reason = 'terminal_success') OR " - "(outcome = 'failed' AND reason = 'terminal_error') OR " - "(outcome = 'interrupted' AND reason = 'terminal_interrupted') OR " - "(outcome = 'unknown' AND reason IN ('missing_terminal', " + "(outcome = 'failed' AND reason = 'terminal_error') OR (outcome = " + "'interrupted' AND reason = 'terminal_interrupted') OR (outcome = " + "'unknown' AND reason IN ('missing_terminal', " "'conflicting_terminal', 'invalid_evidence', 'evidence_limit'))))", "diagnostics_omitted IN (0, 1)", *( @@ -114,10 +114,28 @@ def history_schema_error( for row in conn.execute("SELECT name,sql FROM sqlite_master WHERE type='table'") if row[0] in _TABLE_COLUMNS } + from database.execution_context_schema import TABLE_COLUMNS, context_schema_error + + context_present = any( + row[0] in TABLE_COLUMNS + for row in conn.execute("SELECT name FROM sqlite_master WHERE type='table'") + ) + if not tables and context_present: + return "Execution contexts require execution history" if not tables: return None if allow_missing else "Missing execution history schema" if tables.keys() != _TABLE_COLUMNS.keys(): return "Incomplete execution history schema" + metadata = conn.execute( + "SELECT * FROM execution_history_metadata LIMIT 2" + ).fetchall() + if len(metadata) != 1 or len(metadata[0]) != 3 or metadata[0][1] not in (1, 2): + return "Unsupported execution history metadata/version" + version = metadata[0][1] + if version == 1 and context_present: + return "Unexpected execution contexts for history v1" + if not allow_missing and version != SCHEMA_VERSION: + return "Execution history migration required" for table, expected in _TABLE_COLUMNS.items(): columns = tuple( (row[1], row[2].upper(), row[3], row[5]) @@ -127,6 +145,11 @@ def history_schema_error( return f"{table} has invalid execution history columns" sql = _compact(tables[table]) checks = _TABLE_CHECKS[table] + if table == "execution_history_metadata" and version == 1: + checks = tuple( + check.replace("schema_version = 2", "schema_version = 1") + for check in checks + ) if sql.count("check(") != len(checks) or any( f"check({_compact(check)})" not in sql for check in checks ): @@ -156,10 +179,10 @@ def history_schema_error( ).fetchone(): return f"{table} contains invalid execution history records" metadata = conn.execute( - "SELECT singleton_id,schema_version,storage_epoch " - "FROM execution_history_metadata" + "SELECT singleton_id,schema_version,storage_epoch FROM " + "execution_history_metadata" ).fetchall() - if len(metadata) != 1 or metadata[0][0] != 1 or metadata[0][1] != SCHEMA_VERSION: + if len(metadata) != 1 or metadata[0][0] != 1 or metadata[0][1] != version: return "Unsupported execution history metadata/version" indexes = {row[1]: row for row in conn.execute("PRAGMA index_list(execution_runs)")} for name, (unique, columns) in _INDEXES.items(): @@ -172,13 +195,24 @@ def history_schema_error( item[3] or item[4] != "BINARY" for item in keys ): return f"Invalid execution history index {name}" + if version == 2: + return context_schema_error(conn) return None +def install_context_schema(conn: sqlite3.Connection) -> None: + from database.execution_context_schema import install_context_schema as install + + install(conn) + + def migrate_execution_history(conn: sqlite3.Connection) -> None: - """Install both tables and indexes atomically, without backfilling runs.""" + """Install versioned history and context tables atomically, without backfill.""" conn.execute("SAVEPOINT execution_history_schema") try: + error = history_schema_error(conn) + if error: + raise sqlite3.DatabaseError(error) present = conn.execute( "SELECT 1 FROM sqlite_master WHERE name IN " "('execution_runs','execution_history_metadata') LIMIT 1" @@ -201,6 +235,33 @@ def migrate_execution_history(conn: sqlite3.Connection) -> None: "INSERT INTO execution_history_metadata VALUES(1,?,?)", (SCHEMA_VERSION, str(uuid4())), ) + install_context_schema(conn) + else: + version, epoch = conn.execute( + "SELECT schema_version,storage_epoch FROM execution_history_metadata" + ).fetchone() + if version == 1: + conn.execute( + "ALTER TABLE execution_history_metadata RENAME TO " + "execution_history_metadata_v1" + ) + definitions = [ + f"{name} {kind}{' NOT NULL' if required else ''}" + f"{' PRIMARY KEY' if primary else ''}" + for name, kind, required, primary in _TABLE_COLUMNS[ + "execution_history_metadata" + ] + ] + definitions.extend(f"CHECK({check})" for check in _METADATA_CHECKS) + conn.execute( + "CREATE TABLE execution_history_metadata " + f"({', '.join(definitions)})" + ) + conn.execute( + "INSERT INTO execution_history_metadata VALUES(1,2,?)", (epoch,) + ) + conn.execute("DROP TABLE execution_history_metadata_v1") + install_context_schema(conn) error = history_schema_error(conn, allow_missing=False) if error: raise sqlite3.DatabaseError(error) @@ -216,8 +277,8 @@ def rotate_history_epoch(conn: sqlite3.Connection) -> str: epoch = str(uuid4()) with conn: cursor = conn.execute( - "UPDATE execution_history_metadata SET storage_epoch=? " - "WHERE singleton_id=1", + "UPDATE execution_history_metadata SET storage_epoch=? WHERE " + "singleton_id=1", (epoch,), ) if cursor.rowcount != 1: diff --git a/tests/test_execution_context_schema.py b/tests/test_execution_context_schema.py new file mode 100644 index 0000000..4211267 --- /dev/null +++ b/tests/test_execution_context_schema.py @@ -0,0 +1,327 @@ +"""Versioned migration and hostile context restore boundaries.""" + +import json +import sqlite3 +import unittest +from unittest.mock import patch + +from database import execution_history_schema as history +from utils.execution_contexts import content_digest + +EPOCH = "11111111-1111-4111-8111-111111111111" +RUN = "22222222-2222-4222-8222-222222222222" + + +def legacy(conn): + for table, columns in history._TABLE_COLUMNS.items(): + definitions = [ + f"{name} {kind}{' NOT NULL' if required else ''}" + f"{' PRIMARY KEY' if primary else ''}" + for name, kind, required, primary in columns + ] + checks = ( + ( + "singleton_id = 1", + "schema_version = 1", + history._uuid_check("storage_epoch"), + ) + if table == "execution_history_metadata" + else history._RUN_CHECKS + ) + definitions.extend(f"CHECK({check})" for check in checks) + conn.execute(f"CREATE TABLE {table} ({', '.join(definitions)})") + for name, (unique, columns) in history._INDEXES.items(): + conn.execute( + f"CREATE {'UNIQUE ' if unique else ''}INDEX {name} " + f"ON execution_runs({', '.join(columns)})" + ) + conn.execute("INSERT INTO execution_history_metadata VALUES(1,1,?)", (EPOCH,)) + conn.execute( + "INSERT INTO execution_runs " + "VALUES(?,?,?,?,?,1,100,NULL,NULL,NULL,'observed',NULL,0,NULL,NULL" + ",NULL)", + (RUN, EPOCH, 0, RUN, EPOCH), + ) + conn.execute( + "CREATE TABLE prompts(id INTEGER PRIMARY KEY, text TEXT, run_count INTEGER)" + ) + conn.execute("INSERT INTO prompts VALUES(7,'kept',19)") + conn.commit() + + +class ContextSchemaTests(unittest.TestCase): + def connection(self): + conn = sqlite3.connect(":memory:") + self.addCleanup(conn.close) + return conn + + def test_v1_upgrade_preserves_epoch_and_records(self): + conn = self.connection() + legacy(conn) + before = conn.execute("SELECT * FROM execution_runs").fetchall() + self.assertIsNone(history.history_schema_error(conn)) + history.migrate_execution_history(conn) + self.assertEqual( + conn.execute( + "SELECT schema_version,storage_epoch FROM execution_history_metadata" + ).fetchone(), + (2, EPOCH), + ) + self.assertEqual( + conn.execute("SELECT * FROM execution_runs").fetchall(), before + ) + self.assertEqual( + conn.execute("SELECT * FROM prompts").fetchall(), [(7, "kept", 19)] + ) + for table in ( + "execution_context_texts", + "execution_prompt_snapshots", + "execution_contexts", + "execution_run_contexts", + "execution_context_capture", + ): + self.assertEqual( + conn.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0], 0 + ) + + def test_migration_idempotent(self): + conn = self.connection() + history.migrate_execution_history(conn) + first = "\n".join(conn.iterdump()) + history.migrate_execution_history(conn) + self.assertEqual("\n".join(conn.iterdump()), first) + self.assertEqual( + conn.execute( + "SELECT schema_version FROM execution_history_metadata" + ).fetchone()[0], + 2, + ) + + def test_migration_failure_rolls_back_metadata_and_all_tables(self): + conn = self.connection() + legacy(conn) + before = "\n".join(conn.iterdump()) + + def fail(connection): + connection.execute("CREATE TABLE execution_context_texts(x TEXT)") + raise sqlite3.OperationalError("injected context DDL failure") + + with patch( + "database.execution_history_schema.install_context_schema", + fail, + create=True, + ): + with self.assertRaisesRegex(sqlite3.OperationalError, "injected"): + history.migrate_execution_history(conn) + self.assertEqual("\n".join(conn.iterdump()), before) + + def test_partial_context_schema_rejected_even_for_v1_or_absent(self): + for old in (False, True): + conn = self.connection() + if old: + legacy(conn) + conn.execute("CREATE TABLE execution_context_texts(text_id TEXT)") + self.assertIsNotNone(history.history_schema_error(conn)) + with self.assertRaises(sqlite3.DatabaseError): + history.migrate_execution_history(conn) + + def populate(self, conn): + legacy(conn) + history.migrate_execution_history(conn) + text = content_digest("text", ("hello",)) + snapshot = content_digest("snapshot", ("hello", None, None, None)) + context = content_digest("context", (snapshot, None)) + conn.execute("INSERT INTO execution_context_texts VALUES(?,?)", (text, "hello")) + conn.execute( + "INSERT INTO execution_prompt_snapshots VALUES(?,?,NULL,NULL,NULL)", + (snapshot, text), + ) + conn.execute( + "INSERT INTO execution_contexts VALUES(?,?,NULL)", (context, snapshot) + ) + unknown = dict(provenance="unknown", source="unknown", reason="missing") + evidence = dict( + format_version=1, + role_provenance="inferred", + positive=dict( + raw=dict(provenance="recorded", source="runtime_return", reason=None), + effective=unknown, + prepend=unknown, + append=unknown, + ), + negative=None, + ) + conn.execute( + "INSERT INTO execution_run_contexts VALUES(?,?,?,?,NULL,NULL,NULL,NULL,1)", + (RUN, "node", context, json.dumps(evidence)), + ) + conn.execute( + "INSERT INTO execution_context_capture VALUES(?,'partial','missing',1)", + (RUN,), + ) + return evidence + + def test_valid_context_and_fingerprints(self): + conn = self.connection() + self.populate(conn) + self.assertIsNone(history.history_schema_error(conn, allow_missing=False)) + + def test_v2_rejects_partial_schema_extra_uniqueness_bad_fingerprint_or_orphan(self): + mutations = ( + "DROP TABLE execution_context_capture", + "CREATE UNIQUE INDEX extra_text ON execution_context_texts(text_value)", + "UPDATE execution_context_texts SET text_value='changed'", + "UPDATE execution_prompt_snapshots SET raw_text_id='" + "0" * 64 + "'", + "UPDATE execution_contexts SET context_id='" + "0" * 64 + "'", + "UPDATE execution_run_contexts SET " + "run_id='33333333-3333-4333-8333-333333333333'", + "UPDATE execution_run_contexts SET evidence_json='{}'", + "UPDATE execution_context_capture SET " + "run_id='33333333-3333-4333-8333-333333333333'", + ) + for mutation in mutations: + with self.subTest(mutation=mutation): + conn = self.connection() + self.populate(conn) + conn.execute(mutation) + self.assertIsNotNone(history.history_schema_error(conn)) + + def test_evidence_rejects_extra_metadata_and_false_unknown(self): + for change in ("extra", "known", "role", "version"): + conn = self.connection() + evidence = self.populate(conn) + if change == "extra": + evidence["graph"] = {} + elif change == "known": + evidence["positive"]["effective"] = dict( + provenance="inferred", source="graph_literal", reason=None + ) + elif change == "role": + evidence["negative"] = evidence["positive"] + else: + evidence["format_version"] = True + conn.execute( + "UPDATE execution_run_contexts SET evidence_json=?", + (json.dumps(evidence),), + ) + self.assertIsNotNone(history.history_schema_error(conn)) + + def test_more_than_sixteen_context_links_rejected(self): + conn = self.connection() + self.populate(conn) + for i in range(16): + conn.execute( + "INSERT INTO execution_run_contexts SELECT " + "run_id,?,context_id,evidence_json,positive_prompt_id,negative_pro" + "mpt_id,pair_id,reference_epoch,capture_version FROM " + "execution_run_contexts LIMIT 1", + (str(i),), + ) + self.assertIsNotNone(history.history_schema_error(conn)) + + def test_owned_schema_rejects_triggers_views_and_generated_columns(self): + for mutation in ( + "CREATE TRIGGER hidden_write AFTER INSERT ON " + "execution_context_texts BEGIN DELETE FROM " + "execution_context_texts; END", + "CREATE VIEW context_view AS SELECT * FROM execution_context_texts", + "ALTER TABLE execution_context_texts ADD COLUMN hidden TEXT " + "GENERATED ALWAYS AS (text_value) VIRTUAL", + ): + with self.subTest(mutation=mutation): + conn = self.connection() + self.populate(conn) + conn.execute(mutation) + self.assertIsNotNone(history.history_schema_error(conn)) + + def test_duplicate_json_keys_and_unbounded_or_false_complete_rows_rejected(self): + for change in ( + "duplicate", + "oversized", + "reason", + "complete", + "empty_complete", + ): + with self.subTest(change=change): + conn = self.connection() + self.populate(conn) + conn.execute("PRAGMA ignore_check_constraints=ON") + if change == "duplicate": + conn.execute( + "UPDATE execution_run_contexts SET " + "evidence_json=replace(evidence_json, '{', " + "'{\"format_version\":1,')" + ) + elif change == "oversized": + conn.execute( + "UPDATE execution_context_texts SET text_value=?", + ("é" * 32769,), + ) + elif change == "reason": + conn.execute( + "UPDATE execution_context_capture SET reason='arbitrary' " + ) + else: + conn.execute( + "UPDATE execution_context_capture SET state='complete'" + ) + if change == "empty_complete": + conn.execute("DELETE FROM execution_run_contexts") + self.assertIsNotNone(history.history_schema_error(conn)) + + def test_restore_enforces_context_and_run_aggregate_payload_bounds(self): + for size, links, valid in ( + (40000, 0, False), + (40000, 1, False), + (25000, 5, True), + (25000, 6, False), + ): + with self.subTest(size=size, links=links): + conn = self.connection() + evidence = self.populate(conn) + text = "x" * size + text_id = content_digest("text", (text,)) + snapshot_id = content_digest("snapshot", (text,) * 4) + context_id = content_digest("context", (snapshot_id, snapshot_id)) + conn.execute( + "UPDATE execution_context_texts SET text_id=?,text_value=?", + (text_id, text), + ) + conn.execute( + "UPDATE execution_prompt_snapshots SET " + "snapshot_id=?,raw_text_id=?,effective_text_id=?,prepend_text_id=?" + ",append_text_id=?", + (snapshot_id,) + (text_id,) * 4, + ) + conn.execute( + "UPDATE execution_contexts SET " + "context_id=?,positive_snapshot_id=?,negative_snapshot_id=?", + (context_id, snapshot_id, snapshot_id), + ) + known = dict( + provenance="recorded", source="runtime_return", reason=None + ) + evidence["positive"] = dict.fromkeys( + ("raw", "effective", "prepend", "append"), known + ) + evidence["negative"] = evidence["positive"] + conn.execute( + "UPDATE execution_run_contexts SET context_id=?,evidence_json=?", + (context_id, json.dumps(evidence)), + ) + for i in range(links - 1): + conn.execute( + "INSERT INTO execution_run_contexts SELECT " + "run_id,?,context_id,evidence_json,positive_prompt_id," + "negative_prompt_id,pair_id,reference_epoch," + "capture_version FROM " + "execution_run_contexts LIMIT 1", + (str(i),), + ) + if not links: + conn.execute("DELETE FROM execution_run_contexts") + error = history.history_schema_error(conn) + if valid: + self.assertIsNone(error) + else: + self.assertIsNotNone(error) diff --git a/tests/test_execution_history_database.py b/tests/test_execution_history_database.py index 4c9f37e..e06cfcb 100644 --- a/tests/test_execution_history_database.py +++ b/tests/test_execution_history_database.py @@ -95,6 +95,10 @@ class HistoryDatabaseTests(unittest.TestCase): for pre_pair in (False, True): with self.subTest(pre_pair=pre_pair): with conn: + from database.execution_context_schema import TABLE_COLUMNS + + for table in reversed(TABLE_COLUMNS): + conn.execute(f"DROP TABLE {table}") conn.execute("DROP TABLE execution_runs") conn.execute("DROP TABLE execution_history_metadata") if pre_pair: @@ -232,7 +236,7 @@ class HistoryDatabaseTests(unittest.TestCase): else: conn.execute( "INSERT INTO execution_history_metadata " - "VALUES(1,1,?)", + "VALUES(1,2,?)", (epoch,), ) else: @@ -432,7 +436,7 @@ class HistoryDatabaseTests(unittest.TestCase): "sql=replace(sql, 'queue_item_id >= 0', 'queue_item_id >= -1') " "WHERE name='execution_runs'; PRAGMA writable_schema=OFF", "PRAGMA ignore_check_constraints=ON; " - "UPDATE execution_history_metadata SET schema_version=2", + "UPDATE execution_history_metadata SET schema_version=3", "PRAGMA writable_schema=ON; UPDATE sqlite_master SET " "sql=replace(sql, \"'observed'\", \"'OBSERVED'\") " "WHERE name='execution_runs'; PRAGMA writable_schema=OFF", diff --git a/tests/test_execution_history_restore.py b/tests/test_execution_history_restore.py index 601287c..3ec3ac5 100644 --- a/tests/test_execution_history_restore.py +++ b/tests/test_execution_history_restore.py @@ -87,14 +87,14 @@ class HistoryRestoreTests(unittest.TestCase): source = str(Path(self.temp.name) / "old.db") with closing(sqlite3.connect(source)) as conn, conn: conn.execute( - "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, workflow_name TEXT)" + "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, workflow_name TEXT)" ) conn.execute( - "INSERT INTO prompts(id,text,created_at) " - "VALUES(7,'old prompt','2020-01-01')" + "INSERT INTO prompts(id,text,created_at) VALUES(7,'old " + "prompt','2020-01-01')" ) original = Path(source).read_bytes() self.assertEqual(PromptModel.verify_database_file(source), (True, "ok")) @@ -113,6 +113,89 @@ class HistoryRestoreTests(unittest.TestCase): (7, "old prompt"), ) + def test_wal_and_old_backup_staging_leave_source_unchanged(self): + from database.execution_context_schema import TABLE_COLUMNS + + source = str(Path(self.temp.name) / "v1.db") + old = self.database(source) + start = self.start(old) + old.write_execution_start(start) + with old.model.get_connection() as conn: + conn.execute("INSERT INTO prompts(id,text,run_count) VALUES(7,'old',19)") + for table in reversed(TABLE_COLUMNS): + conn.execute(f"DROP TABLE {table}") + sql = ( + conn.execute( + "SELECT sql FROM sqlite_master WHERE " + "name='execution_history_metadata'" + ) + .fetchone()[0] + .replace("schema_version = 2", "schema_version = 1") + ) + conn.execute("DROP TABLE execution_history_metadata") + conn.execute(sql) + conn.execute( + "INSERT INTO execution_history_metadata VALUES(1,1,?)", + (start.storage_epoch,), + ) + conn.execute("PRAGMA wal_autocheckpoint=0") + with conn: + conn.execute("UPDATE prompts SET run_count=31 WHERE id=7") + source_bytes = Path(source).read_bytes() + wal_bytes = Path(source + "-wal").read_bytes() + self.assertEqual(PromptModel.verify_database_file(source), (True, "ok")) + live = self.database() + live.model.restore_from_file(source) + self.assertEqual(Path(source).read_bytes(), source_bytes) + self.assertEqual(Path(source + "-wal").read_bytes(), wal_bytes) + self.assertEqual(live.get_execution_run(start.run_id)["record_version"], 1) + self.assertEqual( + live.model.get_connection() + .execute("SELECT run_count FROM prompts WHERE id=7") + .fetchone()[0], + 31, + ) + self.assertEqual( + live.model.get_connection() + .execute("SELECT COUNT(*) FROM execution_run_contexts") + .fetchone()[0], + 0, + ) + self.assertEqual( + conn.execute( + "SELECT schema_version FROM execution_history_metadata" + ).fetchone()[0], + 1, + ) + + def test_malformed_v2_context_restore_keeps_live_database_unchanged(self): + live = self.database() + start = self.start(live) + live.write_execution_start(start) + with live.model.get_connection() as conn: + conn.execute("INSERT INTO prompts(text,run_count) VALUES('kept',19)") + before = "\n".join(live.model.get_connection().iterdump()) + for mutation in ( + "DROP TABLE execution_context_capture", + "CREATE UNIQUE INDEX bad_context_unique ON " + "execution_context_texts(text_value)", + "INSERT INTO execution_context_texts VALUES('" + + "0" * 64 + + "','wrong fingerprint')", + ): + with self.subTest(mutation=mutation): + source = str(Path(self.temp.name) / "bad-context.db") + self.assertTrue(live.model.backup_database(source)) + with closing(sqlite3.connect(source)) as conn, conn: + conn.execute(mutation) + original = Path(source).read_bytes() + with self.assertRaises(ValueError): + live.model.restore_from_file(source) + self.assertEqual( + "\n".join(live.model.get_connection().iterdump()), before + ) + self.assertEqual(Path(source).read_bytes(), original) + def test_failed_staging_rotation_keeps_live_epoch_and_never_notifies(self): from database.execution_history import subscribe_history_restore @@ -189,17 +272,17 @@ class HistoryRestoreTests(unittest.TestCase): ) conn.execute("DROP TABLE old_runs") conn.execute( - "CREATE UNIQUE INDEX idx_execution_runs_identity " - "ON execution_runs(session_id, queue_item_id)" + "CREATE UNIQUE INDEX idx_execution_runs_identity ON " + "execution_runs(session_id, queue_item_id)" ) conn.execute( - "CREATE INDEX idx_execution_runs_observed " - "ON execution_runs(observed_dequeue_at, run_id)" + "CREATE INDEX idx_execution_runs_observed ON " + "execution_runs(observed_dequeue_at, run_id)" ) else: conn.execute( - "CREATE UNIQUE INDEX unexpected_job_uniqueness " - "ON execution_runs(job_id)" + "CREATE UNIQUE INDEX unexpected_job_uniqueness ON " + "execution_runs(job_id)" ) original = Path(source).read_bytes() with self.assertRaisesRegex(ValueError, "uniqu"): diff --git a/utils/execution_contexts.py b/utils/execution_contexts.py index 37eb35c..1a1a492 100644 --- a/utils/execution_contexts.py +++ b/utils/execution_contexts.py @@ -98,6 +98,26 @@ BUNDLE_STATES = frozenset( {"complete", "partial", "ambiguous", "unavailable", "conflicting"} ) +CAPTURE_REASONS = frozenset( + { + "", + "captured", + "partial", + "unavailable", + "unsupported", + "ambiguous", + "limit_reached", + "missing", + "conflicting", + "no_supported_context", + "budget_exhausted", + "invalid_evidence", + "multiple_samples", + "source_transformed", + "field_limit", + "graph_limit", + } +) UNKNOWN_REASONS = frozenset( {"unavailable", "unsupported", "ambiguous", "limit_reached", "missing"} )