From a0532df17d82b38502de143d3cfb828a5eb887b8 Mon Sep 17 00:00:00 2001 From: Vito Sansevero Date: Sat, 17 Oct 2026 20:37:18 -0700 Subject: [PATCH] feat(database): migrate execution output evidence to schema v3 --- database/execution_history_schema.py | 53 ++- database/execution_output_schema.py | 218 +++++++++++ tests/test_execution_context_schema.py | 4 +- tests/test_execution_history_database.py | 17 +- tests/test_execution_history_restore.py | 129 ++++++- tests/test_execution_output_schema.py | 439 +++++++++++++++++++++++ 6 files changed, 839 insertions(+), 21 deletions(-) create mode 100644 database/execution_output_schema.py create mode 100644 tests/test_execution_output_schema.py diff --git a/database/execution_history_schema.py b/database/execution_history_schema.py index 81e3dc6..fd678e6 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 = 2 +SCHEMA_VERSION = 3 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 = 2", + "schema_version = 3", _uuid_check("storage_epoch"), ) _RUN_CHECKS = ( @@ -140,6 +140,17 @@ def history_schema_error( } from .execution_context_schema import TABLE_COLUMNS, context_schema_error + from .execution_output_schema import ( + TABLE_COLUMNS as OUTPUT_COLUMNS, + output_schema_error, + ) + + output_present = any( + row[0] in OUTPUT_COLUMNS + for row in conn.execute("SELECT name FROM sqlite_master") + ) + if not tables and output_present: + return "Execution outputs require execution history" context_present = any( row[0] in TABLE_COLUMNS for row in conn.execute("SELECT name FROM sqlite_master WHERE type='table'") @@ -154,9 +165,11 @@ def history_schema_error( "SELECT CASE WHEN typeof(schema_version) = 'integer' " "THEN schema_version ELSE NULL END FROM execution_history_metadata LIMIT 2" ).fetchall() - if len(metadata) != 1 or len(metadata[0]) != 1 or metadata[0][0] not in (1, 2): + if len(metadata) != 1 or len(metadata[0]) != 1 or metadata[0][0] not in (1, 2, 3): return "Unsupported execution history metadata/version" version = metadata[0][0] + if version < 3 and output_present: + return "Unexpected execution outputs for history v1/v2" if version == 1 and context_present: return "Unexpected execution contexts for history v1" if not allow_missing and version != SCHEMA_VERSION: @@ -170,9 +183,9 @@ 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: + if table == "execution_history_metadata": checks = tuple( - check.replace("schema_version = 2", "schema_version = 1") + check.replace("schema_version = 3", f"schema_version = {version}") for check in checks ) if sql.count("check(") != len(checks) or any( @@ -222,9 +235,11 @@ 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 + if version >= 2: + error = context_schema_error(conn) + if error: + return error + return output_schema_error(conn) if version == 3 else None def install_context_schema(conn: sqlite3.Connection) -> None: @@ -233,8 +248,14 @@ def install_context_schema(conn: sqlite3.Connection) -> None: install(conn) +def install_output_schema(conn: sqlite3.Connection) -> None: + from .execution_output_schema import install_output_schema as install + + install(conn) + + def migrate_execution_history(conn: sqlite3.Connection) -> None: - """Install versioned history and context tables atomically, without backfill.""" + """Install history, contexts and outputs atomically, without backfill.""" conn.execute("SAVEPOINT execution_history_schema") try: error = history_schema_error(conn) @@ -263,14 +284,15 @@ def migrate_execution_history(conn: sqlite3.Connection) -> None: (SCHEMA_VERSION, str(uuid4())), ) install_context_schema(conn) + install_output_schema(conn) else: version, epoch = conn.execute( "SELECT schema_version,storage_epoch FROM execution_history_metadata" ).fetchone() - if version == 1: + if version < SCHEMA_VERSION: conn.execute( "ALTER TABLE execution_history_metadata RENAME TO " - "execution_history_metadata_v1" + "execution_history_metadata_old" ) definitions = [ f"{name} {kind}{' NOT NULL' if required else ''}" @@ -285,10 +307,13 @@ def migrate_execution_history(conn: sqlite3.Connection) -> None: f"({', '.join(definitions)})" ) conn.execute( - "INSERT INTO execution_history_metadata VALUES(1,2,?)", (epoch,) + "INSERT INTO execution_history_metadata VALUES(1,?,?)", + (SCHEMA_VERSION, epoch), ) - conn.execute("DROP TABLE execution_history_metadata_v1") - install_context_schema(conn) + conn.execute("DROP TABLE execution_history_metadata_old") + if version == 1: + install_context_schema(conn) + install_output_schema(conn) error = history_schema_error(conn, allow_missing=False) if error: raise sqlite3.DatabaseError(error) diff --git a/database/execution_output_schema.py b/database/execution_output_schema.py new file mode 100644 index 0000000..c39867e --- /dev/null +++ b/database/execution_output_schema.py @@ -0,0 +1,218 @@ +"""Owned schema and bounded restore validation for frozen output locators.""" + +import sqlite3 + +from .execution_history_schema import _compact, _uuid_check + +try: + from ..utils.execution_outputs import ( + BUNDLE_OVERHEAD, + MAX_BUNDLE_BYTES, + MAX_IMAGE_POSITIONS, + MAX_REFERENCE_BYTES, + REFERENCE_OVERHEAD, + STATE_REASONS, + OutputBundle, + OutputReference, + validate_output_bundle, + ) +except ImportError: + from utils.execution_outputs import ( + BUNDLE_OVERHEAD, + MAX_BUNDLE_BYTES, + MAX_IMAGE_POSITIONS, + MAX_REFERENCE_BYTES, + REFERENCE_OVERHEAD, + STATE_REASONS, + OutputBundle, + OutputReference, + validate_output_bundle, + ) + + +TABLE_COLUMNS = { + "execution_output_capture": ( + ("run_id", "TEXT", 1, 1), + ("state", "TEXT", 1, 0), + ("reason", "TEXT", 1, 0), + ("format_version", "INTEGER", 1, 0), + ("source", "TEXT", 1, 0), + ), + "execution_run_outputs": ( + ("run_id", "TEXT", 1, 1), + ("output_node_id", "TEXT", 1, 2), + ("image_ordinal", "INTEGER", 1, 3), + ("folder_type", "TEXT", 1, 0), + ("subfolder", "TEXT", 1, 0), + ("filename", "TEXT", 1, 0), + ("path_style", "TEXT", 1, 0), + ("root_key", "TEXT", 0, 0), + ), +} +TABLE_FOREIGN_KEYS = { + "execution_output_capture": (("run_id", "execution_runs", "run_id"),), + "execution_run_outputs": (("run_id", "execution_output_capture", "run_id"),), +} + + +def _text_check(column: str, maximum: int, *, characters: int | None = None) -> str: + return ( + f"typeof({column}) = 'text' AND length({column}) <= " + f"{maximum if characters is None else characters} " + f"AND length(CAST({column} AS BLOB)) <= {maximum} " + f"AND instr(CAST({column} AS BLOB), x'00') = 0" + ) + + +def _enum_check(column: str, values: tuple | frozenset) -> str: + return ( + f"{_text_check(column, 32)} AND {column} IN (" + + ",".join("'" + value + "'" for value in sorted(values)) + + ")" + ) + + +_RUN_ID_CHECK = f"{_uuid_check('run_id')} AND {_text_check('run_id', 36)}" +TABLE_CHECKS = { + "execution_output_capture": ( + _RUN_ID_CHECK, + _enum_check("state", tuple(STATE_REASONS)), + _enum_check("reason", frozenset.union(*STATE_REASONS.values())), + "typeof(format_version) = 'integer' AND format_version = 1", + f"{_text_check('source', 18)} AND source = 'host_final_history'", + ), + "execution_run_outputs": ( + _RUN_ID_CHECK, + f"{_text_check('output_node_id', 512, characters=128)} " + "AND length(output_node_id) >= 1", + "typeof(image_ordinal) = 'integer' AND image_ordinal BETWEEN 0 AND 255", + _enum_check("folder_type", ("output", "temp")), + _text_check("subfolder", 2048), + f"{_text_check('filename', 1024)} AND length(filename) >= 1", + _enum_check("path_style", ("posix", "windows")), + f"root_key IS NULL OR ({_text_check('root_key', 64)} " + "AND length(root_key) = 64 AND root_key NOT GLOB '*[^0-9a-f]*')", + ), +} + + +def _table_sql(table: str) -> str: + columns = TABLE_COLUMNS[table] + 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] + ) + return f"CREATE TABLE {table} ({', '.join(definitions)})" + + +def install_output_schema(conn: sqlite3.Connection) -> None: + """Install the two owned tables within the caller's migration transaction.""" + for table in TABLE_COLUMNS: + conn.execute(_table_sql(table)) + + +def output_schema_error(conn: sqlite3.Connection) -> str | None: + """Reject altered schemas and invalid rows before returning bounded DTOs.""" + if conn.execute( + "SELECT 1 FROM sqlite_master WHERE type IN ('trigger','view') LIMIT 1" + ).fetchone(): + return "Unsupported execution output 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 output 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 or _compact(tables[table]) != _compact( + _table_sql(table) + ): + return f"{table} has invalid execution output schema" + 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})") + ) + ) + if foreign != tuple( + (*key, "NO ACTION", "NO ACTION", "NONE") + for key in TABLE_FOREIGN_KEYS[table] + ): + return f"{table} has invalid execution output foreign keys" + for index in conn.execute(f"PRAGMA index_list({table})"): + name = index[1].replace('"', '""') + keys = tuple( + row for row in conn.execute(f'PRAGMA index_xinfo("{name}")') if row[5] + ) + if ( + not index[2] + or index[3] != "pk" + or index[4] + or tuple(key[2] for key in keys) + != tuple(column[0] for column in expected if column[3]) + or any(key[3] or key[4] != "BINARY" for key in keys) + ): + return f"{table} has unexpected execution output index/uniqueness" + # Run every scalar guard in SQLite before any evidence text is fetched. + predicate = " AND ".join(f"({check})" for check in TABLE_CHECKS[table]) + if conn.execute( + f"SELECT 1 FROM {table} WHERE NOT ({predicate}) LIMIT 1" + ).fetchone(): + return f"{table} contains invalid execution output records" + if conn.execute(f"PRAGMA foreign_key_check({table})").fetchone(): + return f"{table} contains orphan execution output records" + reference_bytes = f"{REFERENCE_OVERHEAD} + " + " + ".join( + f"coalesce(length(CAST({column} AS BLOB)), 0)" + for column in ( + "output_node_id", + "folder_type", + "subfolder", + "filename", + "path_style", + "root_key", + ) + ) + if conn.execute( + "SELECT 1 FROM execution_run_outputs " + f"WHERE ({reference_bytes}) > {MAX_REFERENCE_BYTES} LIMIT 1" + ).fetchone(): + return "Oversized execution output reference" + if conn.execute( + "SELECT 1 FROM execution_run_outputs GROUP BY run_id " + f"HAVING COUNT(*) > {MAX_IMAGE_POSITIONS} OR " + f"{BUNDLE_OVERHEAD} + SUM({reference_bytes}) > {MAX_BUNDLE_BYTES} LIMIT 1" + ).fetchone(): + return "Oversized execution output bundle" + try: + for run_id, state, reason, version, source in conn.execute( + "SELECT run_id,state,reason,format_version,source " + "FROM execution_output_capture" + ): + references = tuple( + OutputReference(*row) + for row in conn.execute( + "SELECT output_node_id,image_ordinal,folder_type,subfolder," + "filename,path_style,root_key FROM execution_run_outputs " + "WHERE run_id=? ORDER BY output_node_id,image_ordinal", + (run_id,), + ) + ) + validate_output_bundle( + OutputBundle(references, state, reason, version, source) + ) + except (ValueError, TypeError, OverflowError, sqlite3.Error) as exc: + return f"Invalid execution output content: {type(exc).__name__}" + return None diff --git a/tests/test_execution_context_schema.py b/tests/test_execution_context_schema.py index de31eec..759ad76 100644 --- a/tests/test_execution_context_schema.py +++ b/tests/test_execution_context_schema.py @@ -65,7 +65,7 @@ class ContextSchemaTests(unittest.TestCase): conn.execute( "SELECT schema_version,storage_epoch FROM execution_history_metadata" ).fetchone(), - (2, EPOCH), + (history.SCHEMA_VERSION, EPOCH), ) self.assertEqual( conn.execute("SELECT * FROM execution_runs").fetchall(), before @@ -94,7 +94,7 @@ class ContextSchemaTests(unittest.TestCase): conn.execute( "SELECT schema_version FROM execution_history_metadata" ).fetchone()[0], - 2, + history.SCHEMA_VERSION, ) def test_migration_failure_rolls_back_metadata_and_all_tables(self): diff --git a/tests/test_execution_history_database.py b/tests/test_execution_history_database.py index e06cfcb..f308701 100644 --- a/tests/test_execution_history_database.py +++ b/tests/test_execution_history_database.py @@ -97,6 +97,12 @@ class HistoryDatabaseTests(unittest.TestCase): with conn: from database.execution_context_schema import TABLE_COLUMNS + from database.execution_output_schema import ( + TABLE_COLUMNS as OUTPUT_COLUMNS, + ) + + for table in reversed(OUTPUT_COLUMNS): + conn.execute(f"DROP TABLE {table}") for table in reversed(TABLE_COLUMNS): conn.execute(f"DROP TABLE {table}") conn.execute("DROP TABLE execution_runs") @@ -198,7 +204,10 @@ class HistoryDatabaseTests(unittest.TestCase): self.assertEqual(self.count(db), 1) def test_schema_rejects_extra_uniqueness_and_allows_nonunique_indexes(self): - from database.execution_history_schema import history_schema_error + from database.execution_history_schema import ( + SCHEMA_VERSION, + history_schema_error, + ) for table, constraint in ( ("execution_runs", "job_id"), @@ -236,8 +245,8 @@ class HistoryDatabaseTests(unittest.TestCase): else: conn.execute( "INSERT INTO execution_history_metadata " - "VALUES(1,2,?)", - (epoch,), + "VALUES(1,?,?)", + (SCHEMA_VERSION, epoch), ) else: conn.execute( @@ -436,7 +445,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=3", + "UPDATE execution_history_metadata SET schema_version=4", "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 5b379e1..be68a42 100644 --- a/tests/test_execution_history_restore.py +++ b/tests/test_execution_history_restore.py @@ -22,6 +22,111 @@ class HistoryRestoreTests(unittest.TestCase): finish = history_tests.HistoryDatabaseTests.finish count = history_tests.HistoryDatabaseTests.count + def test_output_wal_backup_restore_preserves_rows_and_historical_epoch(self): + try: + from tests.test_execution_output_schema import capture + except ImportError: + from test_execution_output_schema import capture + db = self.database() + start = self.start(db) + db.write_execution_start(start) + conn = db.model.get_connection() + conn.execute("PRAGMA wal_autocheckpoint=0") + with conn: + capture(conn, run=start.run_id) + expected = [ + tuple(row) for row in conn.execute("SELECT * FROM execution_run_outputs") + ] + source = str(Path(self.temp.name) / "outputs-wal.db") + self.assertGreater(Path(self.path + "-wal").stat().st_size, 0) + self.assertTrue(db.model.backup_database(source)) + original = Path(source).read_bytes() + for _ in range(2): + previous = db.get_execution_epoch() + db.model.restore_from_file(source) + self.assertNotEqual(db.get_execution_epoch(), previous) + self.assertNotEqual(db.get_execution_epoch(), start.storage_epoch) + conn = db.model.get_connection() + self.assertEqual( + [ + tuple(row) + for row in conn.execute("SELECT * FROM execution_run_outputs") + ], + expected, + ) + self.assertEqual( + db.get_execution_run(start.run_id)["storage_epoch"], start.storage_epoch + ) + self.assertEqual( + tuple( + conn.execute( + "SELECT state,reason,format_version,source FROM " + "execution_output_capture" + ).fetchone() + ), + ("complete", "reported", 1, "host_final_history"), + ) + self.assertEqual(Path(source).read_bytes(), original) + + def test_malformed_outputs_preserve_source_live_and_existing_safety_backup(self): + try: + from tests.test_execution_output_schema import capture + except ImportError: + from test_execution_output_schema import capture + live = self.database() + start = self.start(live) + live.write_execution_start(start) + with live.model.get_connection() as conn: + capture(conn, run=start.run_id) + before = list(live.model.get_connection().iterdump()) + safety = Path(self.path + ".backup_existing") + safety.write_bytes(b"existing safety backup") + mutations = ( + "DROP TABLE execution_run_outputs", + "DROP TABLE execution_output_capture", + "DELETE FROM execution_output_capture", + "DELETE FROM execution_runs", + "CREATE UNIQUE INDEX bad_output ON execution_run_outputs(filename)", + "CREATE INDEX bad_output ON execution_run_outputs(filename)", + "CREATE VIEW bad_output AS SELECT * FROM execution_run_outputs", + "CREATE TRIGGER bad_output AFTER INSERT ON execution_run_outputs " + "BEGIN SELECT 1; END", + "PRAGMA ignore_check_constraints=ON; UPDATE execution_output_capture " + "SET source='filesystem_scan'", + "PRAGMA ignore_check_constraints=ON; UPDATE execution_output_capture " + "SET format_version=2", + "PRAGMA ignore_check_constraints=ON; UPDATE execution_output_capture " + "SET state='unavailable',reason='unavailable'", + "PRAGMA ignore_check_constraints=ON; UPDATE execution_run_outputs " + "SET filename='a' || char(0) || 'b'", + "PRAGMA ignore_check_constraints=ON; UPDATE execution_run_outputs " + "SET filename=replace(hex(zeroblob(1025)),'0','x')", + "DROP TABLE execution_runs; DROP TABLE execution_history_metadata", + ) + for index, mutation in enumerate(mutations): + with self.subTest(mutation=mutation): + source = str(Path(self.temp.name) / f"bad-output-{index}.db") + self.assertTrue(live.model.backup_database(source)) + with closing(sqlite3.connect(source)) as conn, conn: + conn.executescript(mutation) + original = Path(source).read_bytes() + backups = { + p.name: p.read_bytes() + for p in Path(self.temp.name).glob("live.db.backup_*") + } + self.assertFalse(PromptModel.verify_database_file(source)[0]) + with self.assertRaises(ValueError): + live.model.restore_from_file(source) + self.assertEqual(list(live.model.get_connection().iterdump()), before) + self.assertEqual(Path(source).read_bytes(), original) + self.assertEqual( + { + p.name: p.read_bytes() + for p in Path(self.temp.name).glob("live.db.backup_*") + }, + backups, + ) + def test_legacy_identifier_and_diagnostic_bytes_rejected_before_fetch(self): try: from tests.test_execution_context_schema import legacy, RUN @@ -200,6 +305,7 @@ class HistoryRestoreTests(unittest.TestCase): def test_wal_and_old_backup_staging_leave_source_unchanged(self): from database.execution_context_schema import TABLE_COLUMNS + from database.execution_output_schema import TABLE_COLUMNS as OUTPUT_COLUMNS source = str(Path(self.temp.name) / "v1.db") old = self.database(source) @@ -207,6 +313,8 @@ class HistoryRestoreTests(unittest.TestCase): 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(OUTPUT_COLUMNS): + conn.execute(f"DROP TABLE {table}") for table in reversed(TABLE_COLUMNS): conn.execute(f"DROP TABLE {table}") sql = ( @@ -215,7 +323,7 @@ class HistoryRestoreTests(unittest.TestCase): "name='execution_history_metadata'" ) .fetchone()[0] - .replace("schema_version = 2", "schema_version = 1") + .replace("schema_version = 3", "schema_version = 1") ) conn.execute("DROP TABLE execution_history_metadata") conn.execute(sql) @@ -272,6 +380,25 @@ class HistoryRestoreTests(unittest.TestCase): source = str(Path(self.temp.name) / "bad-context.db") self.assertTrue(live.model.backup_database(source)) with closing(sqlite3.connect(source)) as conn, conn: + # Manufacture v2 explicitly so this still tests old restores. + conn.execute("DROP TABLE execution_run_outputs") + conn.execute("DROP TABLE execution_output_capture") + sql = ( + conn.execute( + "SELECT sql FROM sqlite_master WHERE " + "name='execution_history_metadata'" + ) + .fetchone()[0] + .replace("schema_version = 3", "schema_version = 2") + ) + epoch = conn.execute( + "SELECT storage_epoch FROM execution_history_metadata" + ).fetchone()[0] + conn.execute("DROP TABLE execution_history_metadata") + conn.execute(sql) + conn.execute( + "INSERT INTO execution_history_metadata VALUES(1,2,?)", (epoch,) + ) conn.execute(mutation) original = Path(source).read_bytes() with self.assertRaises(ValueError): diff --git a/tests/test_execution_output_schema.py b/tests/test_execution_output_schema.py new file mode 100644 index 0000000..479d027 --- /dev/null +++ b/tests/test_execution_output_schema.py @@ -0,0 +1,439 @@ +"""Atomic output schema upgrades and bounded validation of untrusted evidence.""" + +import sqlite3 +import unittest +from unittest.mock import patch + +from database import execution_history_schema as history +from database.execution_context_schema import install_context_schema + +try: + from tests.test_execution_context_schema import EPOCH, RUN, legacy +except ImportError: + from test_execution_context_schema import EPOCH, RUN, legacy + +OUTPUT_TABLES = ("execution_output_capture", "execution_run_outputs") +ROOT = "a" * 64 + + +def v2(conn): + """Manufacture the supported pre-output metadata CHECK and context schema.""" + legacy(conn) + sql = ( + conn.execute( + "SELECT sql FROM sqlite_master WHERE name='execution_history_metadata'" + ) + .fetchone()[0] + .replace("schema_version = 1", "schema_version = 2") + ) + conn.execute("DROP TABLE execution_history_metadata") + conn.execute(sql) + conn.execute("INSERT INTO execution_history_metadata VALUES(1,2,?)", (EPOCH,)) + install_context_schema(conn) + conn.commit() + + +def capture(conn, run=RUN, state="complete", reason="reported", rows=True): + conn.execute( + "INSERT INTO execution_output_capture VALUES(?,?,?,1,'host_final_history')", + (run, state, reason), + ) + if rows: + conn.execute( + "INSERT INTO execution_run_outputs VALUES(?, '保存😀', 0, 'output', " + "'Batch Å', ' image 😀.png ', 'posix', ?)", + (run, ROOT), + ) + + +class OutputSchemaTests(unittest.TestCase): + def connection(self): + conn = sqlite3.connect(":memory:") + self.addCleanup(conn.close) + return conn + + def populated(self): + conn = self.connection() + legacy(conn) + history.migrate_execution_history(conn) + capture(conn) + conn.commit() + return conn + + def test_legacy_v1_v2_upgrade_preserves_rows_ddl_epoch_and_counters(self): + for fixture in (None, legacy, v2): + with self.subTest(fixture=fixture): + conn = self.connection() + if fixture: + fixture(conn) + before = list(conn.iterdump()) + ddl = dict(conn.execute("SELECT name,sql FROM sqlite_master")) + self.assertIsNone(history.history_schema_error(conn)) + self.assertIsNotNone( + history.history_schema_error(conn, allow_missing=False) + ) + history.migrate_execution_history(conn) + version, epoch = conn.execute( + "SELECT schema_version,storage_epoch FROM " + "execution_history_metadata" + ).fetchone() + self.assertEqual(version, 3) + if fixture: + self.assertEqual(epoch, EPOCH) + self.assertEqual( + conn.execute("SELECT * FROM prompts").fetchall(), + [(7, "kept", 19)], + ) + self.assertEqual( + conn.execute( + "SELECT run_id,storage_epoch FROM execution_runs" + ).fetchall(), + [(RUN, EPOCH)], + ) + after = list(conn.iterdump()) + self.assertTrue( + all(line in after for line in before if "metadata" not in line) + ) + for name, sql in ddl.items(): + if name != "execution_history_metadata": + self.assertEqual( + conn.execute( + "SELECT sql FROM sqlite_master WHERE name=?", + (name,), + ).fetchone()[0], + sql, + ) + for table in OUTPUT_TABLES: + self.assertEqual( + conn.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0], 0 + ) + self.assertIsNone( + history.history_schema_error(conn, allow_missing=False) + ) + snapshot = list(conn.iterdump()) + history.migrate_execution_history(conn) + self.assertEqual(list(conn.iterdump()), snapshot) + + def test_v2_context_ids_and_content_survive_upgrade(self): + conn = self.connection() + try: + from tests.test_execution_context_schema import ContextSchemaTests + except ImportError: + from test_execution_context_schema import ContextSchemaTests + ContextSchemaTests().populate(conn) + # Explicit supported v2: remove every output-owned table and rebuild metadata. + for table in reversed(OUTPUT_TABLES): + conn.execute(f"DROP TABLE IF EXISTS {table}") + sql = ( + conn.execute( + "SELECT sql FROM sqlite_master WHERE name='execution_history_metadata'" + ) + .fetchone()[0] + .replace("schema_version = 3", "schema_version = 2") + ) + conn.execute("DROP TABLE execution_history_metadata") + conn.execute(sql) + conn.execute("INSERT INTO execution_history_metadata VALUES(1,2,?)", (EPOCH,)) + from database.execution_context_schema import TABLE_COLUMNS + + before = { + table: conn.execute(f"SELECT * FROM {table}").fetchall() + for table in TABLE_COLUMNS + } + history.migrate_execution_history(conn) + for table, rows in before.items(): + self.assertEqual(conn.execute(f"SELECT * FROM {table}").fetchall(), rows) + self.assertIsNone(history.history_schema_error(conn, allow_missing=False)) + + def test_creation_and_final_validation_failures_roll_back_and_retry(self): + for fixture in (None, legacy, v2): + for failure in ("creation", "validation"): + with self.subTest(fixture=fixture, failure=failure): + conn = self.connection() + if fixture: + fixture(conn) + before = list(conn.iterdump()) + if failure == "creation": + + def fail(connection): + connection.execute( + "CREATE TABLE execution_output_capture(x)" + ) + raise sqlite3.OperationalError( + "injected output DDL failure" + ) + + replacement = patch.object( + history, "install_output_schema", fail + ) + else: + original = history.history_schema_error + + def fail(connection, *, allow_missing=True): + return ( + original(connection) + if allow_missing + else "injected final output validation failure" + ) + + replacement = patch.object( + history, "history_schema_error", fail + ) + with ( + replacement, + self.assertRaisesRegex(sqlite3.DatabaseError, "injected"), + ): + history.migrate_execution_history(conn) + self.assertEqual(list(conn.iterdump()), before) + history.migrate_execution_history(conn) + self.assertIsNone( + history.history_schema_error(conn, allow_missing=False) + ) + + def test_outputs_without_history_or_under_old_versions_are_rejected(self): + for fixture in (None, legacy, v2): + for table in OUTPUT_TABLES: + with self.subTest(fixture=fixture, table=table): + conn = self.connection() + if fixture: + fixture(conn) + conn.execute(f"CREATE TABLE {table}(x TEXT)") + before = list(conn.iterdump()) + self.assertIsNotNone(history.history_schema_error(conn)) + with self.assertRaises(sqlite3.DatabaseError): + history.migrate_execution_history(conn) + self.assertEqual(list(conn.iterdump()), before) + + def test_missing_output_table_rejected_in_both_validation_modes(self): + for table in OUTPUT_TABLES: + conn = self.populated() + conn.execute(f"DROP TABLE {table}") + for allow in (True, False): + self.assertIsNotNone( + history.history_schema_error(conn, allow_missing=allow) + ) + + def test_exact_owned_indexes_checks_columns_foreign_keys_and_objects(self): + mutations = ( + "CREATE UNIQUE INDEX extra ON execution_run_outputs(filename)", + "CREATE INDEX extra ON execution_run_outputs(filename)", + "CREATE TRIGGER extra AFTER INSERT ON execution_run_outputs " + "BEGIN SELECT 1; END", + "CREATE VIEW extra AS SELECT * FROM execution_run_outputs", + ) + for sql in mutations: + with self.subTest(sql=sql): + conn = self.populated() + conn.execute(sql) + self.assertIsNotNone(history.history_schema_error(conn)) + replacements = ( + ( + "execution_run_outputs", + "filename TEXT NOT NULL", + "filename TEXT NOT NULL UNIQUE", + ), + ( + "execution_run_outputs", + "filename TEXT NOT NULL", + "filename TEXT NOT NULL COLLATE NOCASE", + ), + ( + "execution_run_outputs", + "PRIMARY KEY (run_id, output_node_id, image_ordinal)", + "PRIMARY KEY (run_id, image_ordinal, output_node_id)", + ), + ( + "execution_output_capture", + "format_version = 1", + "format_version IN (1, 2)", + ), + ( + "execution_output_capture", + "REFERENCES execution_runs(run_id)", + "REFERENCES execution_runs(job_id)", + ), + ("execution_run_outputs", "ON DELETE NO ACTION", "ON DELETE CASCADE"), + ("execution_run_outputs", "ON UPDATE NO ACTION", "ON UPDATE CASCADE"), + ( + "execution_run_outputs", + "REFERENCES execution_output_capture(run_id)", + "REFERENCES execution_runs(run_id)", + ), + ("execution_run_outputs", "root_key TEXT", "root_key BLOB"), + ) + for table, old, new in replacements: + with self.subTest(table=table, replacement=new): + conn = self.populated() + sql = conn.execute( + "SELECT sql FROM sqlite_master WHERE name=?", (table,) + ).fetchone()[0] + self.assertIn(old, sql) + conn.execute(f"DROP TABLE {table}") + conn.execute(sql.replace(old, new)) + self.assertIsNotNone(history.history_schema_error(conn)) + + def test_orphan_run_header_and_rows_rejected_without_fk_enforcement(self): + for mutation in ( + "DELETE FROM execution_runs", + "DELETE FROM execution_output_capture", + "UPDATE execution_run_outputs " + "SET run_id='33333333-3333-4333-8333-333333333333'", + ): + with self.subTest(mutation=mutation): + conn = self.populated() + conn.execute(mutation) + self.assertIsNotNone(history.history_schema_error(conn)) + + def test_malformed_scalars_rejected_before_text_or_row_fetch(self): + cases = ( + ("execution_output_capture", "run_id", RUN + "\0"), + ("execution_output_capture", "state", "complete\0"), + ("execution_output_capture", "reason", "reported\0"), + ("execution_output_capture", "source", "host_final_history\0"), + ("execution_output_capture", "format_version", "x" * 4096), + ("execution_run_outputs", "run_id", RUN + "\0"), + ("execution_run_outputs", "output_node_id", "x" * 129), + ("execution_run_outputs", "output_node_id", "😀" * 129), + ("execution_run_outputs", "output_node_id", "n\0"), + ("execution_run_outputs", "image_ordinal", "x" * 4096), + ("execution_run_outputs", "folder_type", "output\0"), + ("execution_run_outputs", "subfolder", "x" * 2049), + ("execution_run_outputs", "subfolder", "a\0b"), + ("execution_run_outputs", "filename", "x" * 1025), + ("execution_run_outputs", "filename", "a\0b"), + ("execution_run_outputs", "path_style", "posix\0"), + ("execution_run_outputs", "root_key", ROOT + "\0"), + ("execution_run_outputs", "filename", b"\xff"), + ) + for table, column, malicious in cases: + with self.subTest(table=table, column=column, value=repr(malicious)[:30]): + conn = self.populated() + conn.execute("PRAGMA ignore_check_constraints=ON") + conn.execute(f"UPDATE {table} SET {column}=?", (malicious,)) + + def bounded_text(value): + self.assertNotEqual( + value, + malicious.encode() if type(malicious) is str else malicious, + ) + return value.decode() + + def bounded_row(cursor, row): + self.assertFalse(any(value == malicious for value in row)) + return row + + conn.text_factory = bounded_text + conn.row_factory = bounded_row + self.assertIsNotNone(history.history_schema_error(conn)) + + def test_state_version_source_and_locator_corruption_rejected(self): + cases = ( + ("execution_output_capture", "source", "scan"), + ("execution_output_capture", "format_version", 2), + ("execution_output_capture", "state", "not_captured"), + ("execution_output_capture", "reason", "none_reported"), + ("execution_run_outputs", "filename", "../image.png"), + ("execution_run_outputs", "filename", ""), + ("execution_run_outputs", "subfolder", "a/../b"), + ("execution_run_outputs", "folder_type", "input"), + ("execution_run_outputs", "path_style", "native"), + ("execution_run_outputs", "image_ordinal", 256), + ("execution_run_outputs", "image_ordinal", -1), + ("execution_run_outputs", "root_key", None), + ) + for table, column, value in cases: + with self.subTest(table=table, column=column, value=value): + conn = self.populated() + conn.execute("PRAGMA ignore_check_constraints=ON") + conn.execute(f"UPDATE {table} SET {column}=?", (value,)) + self.assertIsNotNone(history.history_schema_error(conn)) + + def test_state_row_consistency_and_conflicting_first_rows(self): + cases = ( + ("complete", "reported", False, False), + ("complete", "none_reported", False, True), + ("complete", "none_reported", True, False), + ("partial", "invalid_evidence", False, True), + ("partial", "root_unavailable", False, False), + ("partial", "root_unavailable", True, True), + ("unavailable", "unavailable", True, False), + ("unavailable", "invalid_evidence", False, True), + ("partial", "reported", True, False), + ("conflicting", "conflicting", True, True), + ) + for state, reason, rows, valid in cases: + with self.subTest(state=state, reason=reason, rows=rows): + conn = self.connection() + legacy(conn) + history.migrate_execution_history(conn) + conn.execute("PRAGMA ignore_check_constraints=ON") + capture(conn, state=state, reason=reason, rows=rows) + self.assertEqual(history.history_schema_error(conn) is None, valid) + + def test_bundle_count_and_exact_payload_bounds(self): + for count, filename, valid in ( + (256, "a.png", True), + (257, "a.png", False), + (128, "x" * 1024, False), + ): + with self.subTest(count=count, filename_bytes=len(filename)): + conn = self.populated() + conn.execute("DELETE FROM execution_run_outputs") + conn.executemany( + "INSERT INTO execution_run_outputs " + "VALUES(?,?,0,'temp','',?,'posix',?)", + ((RUN, str(index), filename, ROOT) for index in range(count)), + ) + self.assertEqual(history.history_schema_error(conn) is None, valid) + + def test_exact_128_kib_charge_is_valid_but_one_extra_byte_is_rejected(self): + conn = self.populated() + conn.execute("DELETE FROM execution_run_outputs") + # 128-byte bundle header plus 128 references charged 1023 bytes each. + # Each row: 128 overhead + 1 node + 4 folder + 821 name + 5 style + 64 root. + conn.executemany( + "INSERT INTO execution_run_outputs VALUES(?,'n',?,'temp','',?,'posix',?)", + ((RUN, ordinal, "x" * 821, ROOT) for ordinal in range(128)), + ) + self.assertIsNone(history.history_schema_error(conn)) + conn.execute( + "UPDATE execution_run_outputs SET filename=? WHERE image_ordinal=0", + ("x" * 822,), + ) + self.assertIsNotNone(history.history_schema_error(conn)) + + def test_oversized_bundle_is_rejected_before_fetching_its_scalar_payload(self): + conn = self.populated() + conn.execute("DELETE FROM execution_run_outputs") + filename = "x" * 1024 + conn.executemany( + "INSERT INTO execution_run_outputs VALUES(?,?,0,'temp','',?,'posix',?)", + ((RUN, str(index), filename, ROOT) for index in range(128)), + ) + + def bounded_text(value): + self.assertNotEqual(value, filename.encode()) + return value.decode() + + conn.text_factory = bounded_text + self.assertIsNotNone(history.history_schema_error(conn)) + + def test_duplicate_locators_and_unicode_survive_validation_unchanged(self): + conn = self.populated() + conn.execute( + "INSERT INTO execution_run_outputs SELECT run_id,output_node_id," + "1,folder_type,subfolder,filename,path_style,root_key " + "FROM execution_run_outputs" + ) + conn.execute( + "INSERT INTO execution_run_outputs SELECT run_id,'other'," + "image_ordinal,folder_type,subfolder,filename,path_style,root_key " + "FROM execution_run_outputs" + ) + before = conn.execute("SELECT * FROM execution_run_outputs").fetchall() + self.assertEqual(len(before), 4) + self.assertIsNone(history.history_schema_error(conn, allow_missing=False)) + history.migrate_execution_history(conn) + self.assertEqual( + conn.execute("SELECT * FROM execution_run_outputs").fetchall(), before + )