feat(history): migrate execution contexts to schema v2

This commit is contained in:
Vito Sansevero
2026-10-17 17:39:20 -07:00
parent 523c136455
commit cc6e7f9ec0
6 changed files with 903 additions and 28 deletions
+380
View File
@@ -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
+75 -14
View File
@@ -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:
+327
View File
@@ -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)
+6 -2
View File
@@ -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",
+95 -12
View File
@@ -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"):
+20
View File
@@ -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"}
)