feat(database): migrate execution output evidence to schema v3
This commit is contained in:
Vito Sansevero
@@ -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)
|
||||
|
||||
@@ -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
|
||||
@@ -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):
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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):
|
||||
|
||||
@@ -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
|
||||
)
|
||||
Reference in New Issue
Block a user