Vito
4ef236aaf7
* feat(history): define bounded output reference contracts * feat(database): migrate execution output evidence to schema v3 * feat(history): persist immutable run output references * test(history): isolate mocked node gallery monitor * feat(history): capture bounded output descriptors with shared leases * fix(history): authenticate payload owners and release failed captures * feat(history): attach output evidence to exact finalized attempts * test(history): verify run output evidence on real ComfyUI * fix(history): release idle worker payload references * docs(history): clarify defensive lease cleanup * test(history): align output fixtures with native path styles * test(history): separate restore completion from writer drain * test(history): bound batch drains by progress and total time
434 lines
18 KiB
Python
434 lines
18 KiB
Python
"""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(),
|
|
(history.SCHEMA_VERSION, 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],
|
|
history.SCHEMA_VERSION,
|
|
)
|
|
|
|
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_owned_identifier_constraints_reject_nul_suffixed_bytes(self):
|
|
columns = {
|
|
"execution_context_texts": ("text_id",),
|
|
"execution_prompt_snapshots": (
|
|
"snapshot_id",
|
|
"raw_text_id",
|
|
"effective_text_id",
|
|
"prepend_text_id",
|
|
"append_text_id",
|
|
),
|
|
"execution_contexts": (
|
|
"context_id",
|
|
"positive_snapshot_id",
|
|
"negative_snapshot_id",
|
|
),
|
|
"execution_run_contexts": (
|
|
"run_id",
|
|
"anchor_id",
|
|
"context_id",
|
|
"reference_epoch",
|
|
),
|
|
"execution_context_capture": ("run_id",),
|
|
}
|
|
for table, names in columns.items():
|
|
for column in names:
|
|
with self.subTest(table=table, column=column):
|
|
conn = self.connection()
|
|
self.populate(conn)
|
|
prefix = (
|
|
RUN
|
|
if column in ("run_id", "reference_epoch")
|
|
else "node" if column == "anchor_id" else "a" * 64
|
|
)
|
|
with self.assertRaises(sqlite3.IntegrityError):
|
|
conn.execute(
|
|
f"UPDATE {table} SET {column}=?",
|
|
(prefix + "\0" + "x" * 1024,),
|
|
)
|
|
|
|
def test_context_validation_rejects_oversized_keys_before_fetch(self):
|
|
for table, column, prefix in (
|
|
("execution_context_texts", "text_id", "a" * 64),
|
|
("execution_run_contexts", "anchor_id", "node"),
|
|
):
|
|
with self.subTest(table=table, column=column):
|
|
conn = self.connection()
|
|
self.populate(conn)
|
|
malicious = prefix + "\0" + "x" * 1024
|
|
if table == "execution_context_texts":
|
|
for dependent in (
|
|
"execution_context_capture",
|
|
"execution_run_contexts",
|
|
"execution_contexts",
|
|
"execution_prompt_snapshots",
|
|
):
|
|
conn.execute(f"DELETE FROM {dependent}")
|
|
conn.execute("PRAGMA ignore_check_constraints=ON")
|
|
conn.execute(f"UPDATE {table} SET {column}=?", (malicious,))
|
|
|
|
def bounded_text(value):
|
|
self.assertNotEqual(value, malicious.encode())
|
|
return value.decode()
|
|
|
|
conn.text_factory = bounded_text
|
|
self.assertIsNotNone(history.history_schema_error(conn))
|
|
|
|
def test_unicode_anchor_boundary_and_prompt_nuls_preserve_exact_text(self):
|
|
conn = self.connection()
|
|
self.populate(conn)
|
|
text = "hello\0world"
|
|
text_id = content_digest("text", (text,))
|
|
snapshot_id = content_digest("snapshot", (text, None, None, None))
|
|
context_id = content_digest("context", (snapshot_id, None))
|
|
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=?",
|
|
(snapshot_id, text_id),
|
|
)
|
|
conn.execute(
|
|
"UPDATE execution_contexts SET context_id=?,positive_snapshot_id=?",
|
|
(context_id, snapshot_id),
|
|
)
|
|
conn.execute(
|
|
"UPDATE execution_run_contexts SET context_id=?,anchor_id=?",
|
|
(context_id, "😀" * 128),
|
|
)
|
|
self.assertIsNone(history.history_schema_error(conn))
|
|
self.assertEqual(
|
|
conn.execute("SELECT text_value FROM execution_context_texts").fetchone(),
|
|
(text,),
|
|
)
|
|
|
|
def test_link_without_capture_header_is_malformed_v2(self):
|
|
from database.execution_context_schema import context_schema_error
|
|
|
|
conn = self.connection()
|
|
self.populate(conn)
|
|
conn.execute("DELETE FROM execution_context_capture")
|
|
before = "\n".join(conn.iterdump())
|
|
self.assertIsNotNone(context_schema_error(conn))
|
|
self.assertIsNotNone(history.history_schema_error(conn, allow_missing=False))
|
|
self.assertEqual("\n".join(conn.iterdump()), before)
|
|
|
|
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)
|