"""Rows-present forward-copy coverage for the recreate migrations. The V011–V016 migrations rebuild ``meta_skill_runs`` / ``meta_skill_run_steps`` with a drop-and-copy. Full-directory tests only ever exercise those copies against EMPTY tables (the second ``apply_pending`` filters to nothing), so these tests build the schema as of just-before the target migration, seed synthetic rows, then migrate to head and assert row-for-row survival. Also covers the re-run guards (yoyo's apply-then-mark crash window / operator reapply, simulated by deleting the ledger row) and the fail-loud guard against live foreign-key enforcement on the migration connection. All data below is synthetic public-dummy content. """ from __future__ import annotations import sqlite3 from datetime import datetime from pathlib import Path import pytest from yoyo import get_backend, read_migrations from opensquilla.persistence.migrator import apply_pending MIGRATIONS_DIR = Path(__file__).resolve().parents[2] / "migrations" V010_RUNS = "V010__meta_skill_runs" V011 = "V011__meta_skill_runs_triggered_by_auto" V012 = "V012__meta_skill_run_steps_allow_llm_chat" V013 = "V013__meta_skill_runs_clarify" V014 = "V014__meta_skill_run_steps_allow_user_input" V016 = "V016__meta_skill_runs_triggered_by_manual_command" RUN_COLUMNS_V010 = ( "run_id", "meta_skill_name", "meta_skill_digest", "plan_snapshot_json", "triggered_by", "session_key", "turn_id", "owner_pid", "status", "started_at_ms", "ended_at_ms", "inputs_json", "final_text", "failed_step_id", "error", "truncated_fields", ) STEP_COLUMNS_V010 = ( "run_id", "step_id", "step_kind", "declared_skill", "effective_skill", "status", "started_at_ms", "ended_at_ms", "rendered_inputs_json", "output_text", "error", "substitute_step_id", "truncated_fields", ) CLARIFY_COLUMNS = ( "awaiting_step_id", "awaiting_schema_json", "awaiting_since", "awaiting_filled_json", "step_outputs_json", "parse_failure_count", ) def _open_conn(db_path: str) -> sqlite3.Connection: conn = sqlite3.connect(db_path) conn.execute("PRAGMA foreign_keys = ON") return conn def _apply_through(db: str, last_id: str) -> None: """Apply *last_id* and its transitive __depends__ closure only.""" # Mirror the migrator's Python 3.12 datetime adapter registration; the # yoyo ledger writes applied_at datetimes on this direct-backend path. sqlite3.register_adapter(datetime, lambda value: value.isoformat(" ")) backend = get_backend("sqlite:///" + db) try: migrations = read_migrations(str(MIGRATIONS_DIR)) by_id = {migration.id: migration for migration in migrations} wanted: set[str] = set() todo = [last_id] while todo: migration_id = todo.pop() if migration_id in wanted: continue wanted.add(migration_id) todo.extend(dep.id for dep in by_id[migration_id].depends) subset = migrations.filter(lambda m: m.id in wanted) with backend.lock(): backend.apply_migrations(backend.to_apply(subset)) finally: backend.connection.close() def _unmark(db: str, migration_id: str) -> None: """Delete *migration_id* from the yoyo ledger, simulating the apply-then-mark crash window (or an operator forcing a reapply).""" conn = sqlite3.connect(db) try: tables = [ row[0] for row in conn.execute( "SELECT name FROM sqlite_master WHERE type='table' " "AND name LIKE '%yoyo_migration'" ).fetchall() ] assert tables, "yoyo ledger table not found" removed = 0 for table in tables: cur = conn.execute( f'DELETE FROM "{table}" WHERE migration_id = ?', (migration_id,) ) removed += cur.rowcount conn.commit() assert removed, f"{migration_id} was not recorded in the ledger" finally: conn.close() def _fk_check_clean(conn: sqlite3.Connection) -> None: orphans = conn.execute("PRAGMA foreign_key_check").fetchall() assert orphans == [], f"foreign_key_check found orphans: {orphans}" def _seed_run( conn: sqlite3.Connection, run_id: str, *, triggered_by: str = "hard_takeover", status: str = "ok", session_key: str | None = None, ) -> dict[str, object]: values = { "run_id": run_id, "meta_skill_name": "synth-meta", "meta_skill_digest": "cafef00d", "plan_snapshot_json": '{"steps": ["collect", "draft"]}', "triggered_by": triggered_by, "session_key": session_key, "turn_id": f"turn-{run_id}", "owner_pid": 4242, "status": status, "started_at_ms": 1_000_000, "ended_at_ms": 1_000_500, "inputs_json": '{"topic": "demo"}', "final_text": f"final text for {run_id}", "failed_step_id": None, "error": None, "truncated_fields": "", } columns = ", ".join(RUN_COLUMNS_V010) marks = ", ".join("?" for _ in RUN_COLUMNS_V010) conn.execute( f"INSERT INTO meta_skill_runs ({columns}) VALUES ({marks})", tuple(values[column] for column in RUN_COLUMNS_V010), ) conn.commit() return values def _seed_step( conn: sqlite3.Connection, run_id: str, step_id: str, *, step_kind: str = "agent", status: str = "ok", ) -> dict[str, object]: values = { "run_id": run_id, "step_id": step_id, "step_kind": step_kind, "declared_skill": "note_taker", "effective_skill": "note_taker", "status": status, "started_at_ms": 1_000_100, "ended_at_ms": 1_000_200, "rendered_inputs_json": '{"prompt": "demo"}', "output_text": f"output for {step_id}", "error": None, "substitute_step_id": None, "truncated_fields": "", } columns = ", ".join(STEP_COLUMNS_V010) marks = ", ".join("?" for _ in STEP_COLUMNS_V010) conn.execute( f"INSERT INTO meta_skill_run_steps ({columns}) VALUES ({marks})", tuple(values[column] for column in STEP_COLUMNS_V010), ) conn.commit() return values def _fetch_row( conn: sqlite3.Connection, table: str, columns: tuple[str, ...], where: str, args: tuple ) -> tuple: return conn.execute( f"SELECT {', '.join(columns)} FROM {table} WHERE {where}", args ).fetchone() def _assert_row( conn: sqlite3.Connection, table: str, columns: tuple[str, ...], where: str, args: tuple, expected: dict[str, object], ) -> None: row = _fetch_row(conn, table, columns, where, args) assert row is not None, f"row {args} missing from {table}" for column, actual in zip(columns, row): assert actual == expected[column], ( f"{table}.{column} for {args}: expected {expected[column]!r}, " f"got {actual!r}" ) # ── forward copies with rows present ───────────────────────────────── def test_v011_forward_copy_preserves_seeded_rows(tmp_path: Path) -> None: db = str(tmp_path / "v011_fwd.db") _apply_through(db, V010_RUNS) conn = _open_conn(db) try: run_a = _seed_run(conn, "rA", triggered_by="hard_takeover", status="ok") run_b = _seed_run(conn, "rB", triggered_by="soft_meta_invoke", status="failed") step_a = _seed_step(conn, "rA", "s1", step_kind="agent") step_b = _seed_step(conn, "rA", "s2", step_kind="tool_call", status="failed") finally: conn.close() applied = apply_pending(db, MIGRATIONS_DIR) assert V011 in applied conn = _open_conn(db) try: _assert_row(conn, "meta_skill_runs", RUN_COLUMNS_V010, "run_id=?", ("rA",), run_a) _assert_row(conn, "meta_skill_runs", RUN_COLUMNS_V010, "run_id=?", ("rB",), run_b) _assert_row( conn, "meta_skill_run_steps", STEP_COLUMNS_V010, "step_id=?", ("s1",), step_a ) _assert_row( conn, "meta_skill_run_steps", STEP_COLUMNS_V010, "step_id=?", ("s2",), step_b ) # New columns picked up defaults across the chain. clarify = _fetch_row( conn, "meta_skill_runs", CLARIFY_COLUMNS, "run_id=?", ("rA",) ) assert clarify == (None, None, None, None, None, 0) usage = conn.execute( "SELECT usage_json FROM meta_skill_run_steps WHERE step_id='s1'" ).fetchone() assert usage == ("{}",) # Relaxed CHECK accepts the new trigger values. _seed_run(conn, "rC", triggered_by="auto_cron") _fk_check_clean(conn) finally: conn.close() def test_v012_forward_copy_preserves_seeded_steps(tmp_path: Path) -> None: db = str(tmp_path / "v012_fwd.db") _apply_through(db, V011) conn = _open_conn(db) try: _seed_run(conn, "r1", triggered_by="auto_dream", status="running") step_a = _seed_step(conn, "r1", "s1", step_kind="llm_classify") step_b = _seed_step(conn, "r1", "s2", step_kind="skill_exec", status="substituted") finally: conn.close() applied = apply_pending(db, MIGRATIONS_DIR) assert V012 in applied conn = _open_conn(db) try: _assert_row( conn, "meta_skill_run_steps", STEP_COLUMNS_V010, "step_id=?", ("s1",), step_a ) _assert_row( conn, "meta_skill_run_steps", STEP_COLUMNS_V010, "step_id=?", ("s2",), step_b ) # Widened CHECK accepts llm_chat after the copy. _seed_step(conn, "r1", "s3", step_kind="llm_chat") _fk_check_clean(conn) finally: conn.close() def test_v013_forward_copy_preserves_seeded_rows(tmp_path: Path) -> None: db = str(tmp_path / "v013_fwd.db") _apply_through(db, V012) conn = _open_conn(db) try: run_a = _seed_run(conn, "r1", status="running", session_key="agent:demo:one") run_b = _seed_run(conn, "r2", status="cancelled") step_a = _seed_step(conn, "r1", "s1", step_kind="llm_chat") finally: conn.close() applied = apply_pending(db, MIGRATIONS_DIR) assert V013 in applied conn = _open_conn(db) try: _assert_row(conn, "meta_skill_runs", RUN_COLUMNS_V010, "run_id=?", ("r1",), run_a) _assert_row(conn, "meta_skill_runs", RUN_COLUMNS_V010, "run_id=?", ("r2",), run_b) _assert_row( conn, "meta_skill_run_steps", STEP_COLUMNS_V010, "step_id=?", ("s1",), step_a ) clarify = _fetch_row( conn, "meta_skill_runs", CLARIFY_COLUMNS, "run_id=?", ("r1",) ) assert clarify == (None, None, None, None, None, 0) _fk_check_clean(conn) finally: conn.close() def test_v014_forward_preserves_awaiting_user_run(tmp_path: Path) -> None: """V013→head path: a parked awaiting_user run with populated clarify state must survive the V014 step rebuild and the V016 run rebuild.""" db = str(tmp_path / "v014_fwd.db") _apply_through(db, V013) clarify_values = { "awaiting_step_id": "collect", "awaiting_schema_json": '{"fields": [{"name": "city"}]}', "awaiting_since": 1_000_000.5, "awaiting_filled_json": '{"city": "Exampleville"}', "step_outputs_json": '{"collect": {"city": "Exampleville"}}', "parse_failure_count": 2, } conn = _open_conn(db) try: run = _seed_run( conn, "r1", status="running", session_key="agent:demo:awaiting" ) step = _seed_step(conn, "r1", "s1", step_kind="agent", status="running") assignments = ", ".join(f"{column} = ?" for column in CLARIFY_COLUMNS) conn.execute( f"UPDATE meta_skill_runs SET status='awaiting_user', {assignments} " "WHERE run_id='r1'", tuple(clarify_values[column] for column in CLARIFY_COLUMNS), ) conn.commit() finally: conn.close() applied = apply_pending(db, MIGRATIONS_DIR) assert V014 in applied assert V016 in applied conn = _open_conn(db) try: expected = dict(run) expected["status"] = "awaiting_user" _assert_row( conn, "meta_skill_runs", RUN_COLUMNS_V010, "run_id=?", ("r1",), expected ) _assert_row( conn, "meta_skill_runs", CLARIFY_COLUMNS, "run_id=?", ("r1",), clarify_values ) _assert_row( conn, "meta_skill_run_steps", STEP_COLUMNS_V010, "step_id=?", ("s1",), step ) # Widened CHECK accepts user_input after the copy. _seed_step(conn, "r1", "s2", step_kind="user_input", status="running") _fk_check_clean(conn) finally: conn.close() # ── re-run guards (apply-then-mark crash window / operator reapply) ── def test_v013_reapply_keeps_clarify_state(tmp_path: Path) -> None: """Re-running V013 against an already-migrated schema must not NULL a parked awaiting_user run's clarify columns (that would break resume).""" db = str(tmp_path / "v013_reapply.db") apply_pending(db, MIGRATIONS_DIR) clarify_values = { "awaiting_step_id": "collect", "awaiting_schema_json": '{"fields": []}', "awaiting_since": 2_000_000.25, "awaiting_filled_json": "{}", "step_outputs_json": '{"collect": {}}', "parse_failure_count": 1, } conn = _open_conn(db) try: _seed_run(conn, "r1", status="running", session_key="agent:demo:parked") assignments = ", ".join(f"{column} = ?" for column in CLARIFY_COLUMNS) conn.execute( f"UPDATE meta_skill_runs SET status='awaiting_user', {assignments} " "WHERE run_id='r1'", tuple(clarify_values[column] for column in CLARIFY_COLUMNS), ) conn.commit() finally: conn.close() _unmark(db, V013) applied = apply_pending(db, MIGRATIONS_DIR) assert applied == [V013] conn = _open_conn(db) try: _assert_row( conn, "meta_skill_runs", CLARIFY_COLUMNS, "run_id=?", ("r1",), clarify_values ) status = conn.execute( "SELECT status FROM meta_skill_runs WHERE run_id='r1'" ).fetchone() assert status == ("awaiting_user",) finally: conn.close() def test_v014_reapply_keeps_usage_json(tmp_path: Path) -> None: """Re-running V014 post-V015 must not rebuild the step table with the pre-usage column list (that would silently drop usage_json data).""" db = str(tmp_path / "v014_reapply.db") apply_pending(db, MIGRATIONS_DIR) conn = _open_conn(db) try: _seed_run(conn, "r1") _seed_step(conn, "r1", "s1") conn.execute( "UPDATE meta_skill_run_steps SET usage_json=? WHERE step_id='s1'", ('{"input_tokens": 5}',), ) conn.commit() finally: conn.close() _unmark(db, V014) applied = apply_pending(db, MIGRATIONS_DIR) assert applied == [V014] conn = _open_conn(db) try: usage = conn.execute( "SELECT usage_json FROM meta_skill_run_steps WHERE step_id='s1'" ).fetchone() assert usage == ('{"input_tokens": 5}',) finally: conn.close() @pytest.mark.parametrize("migration_id", [V011, V012, V016]) def test_recreate_reapply_is_noop_on_head_schema( tmp_path: Path, migration_id: str ) -> None: """Re-running the CHECK-widening recreates against the head schema must neither fail on column-count drift nor lose any data.""" db = str(tmp_path / "reapply.db") apply_pending(db, MIGRATIONS_DIR) conn = _open_conn(db) try: run = _seed_run(conn, "r1") step = _seed_step(conn, "r1", "s1") finally: conn.close() _unmark(db, migration_id) applied = apply_pending(db, MIGRATIONS_DIR) assert applied == [migration_id] conn = _open_conn(db) try: _assert_row(conn, "meta_skill_runs", RUN_COLUMNS_V010, "run_id=?", ("r1",), run) _assert_row( conn, "meta_skill_run_steps", STEP_COLUMNS_V010, "step_id=?", ("s1",), step ) _fk_check_clean(conn) finally: conn.close() # ── fail-loud guard against live FK enforcement ────────────────────── def _load_module(migration_id: str): migrations = read_migrations(str(MIGRATIONS_DIR)) migration = next(m for m in migrations if m.id == migration_id) migration.load() return migration.module @pytest.mark.parametrize( "migration_id", [V011, V012, V013, V014, "V015__meta_skill_step_usage", V016], ) def test_fk_guard_raises_when_enforcement_is_live( tmp_path: Path, migration_id: str ) -> None: """`PRAGMA foreign_keys` is a no-op inside a transaction, so a recreate running on an enforcement-enabled connection cannot switch it off; every recreate migration must refuse instead of cascade-deleting child rows.""" db = str(tmp_path / "fk_guard.db") apply_pending(db, MIGRATIONS_DIR) module = _load_module(migration_id) conn = sqlite3.connect(db) conn.isolation_level = None # autocommit; transactions managed manually try: conn.execute("PRAGMA foreign_keys = ON") conn.execute("BEGIN") with pytest.raises(RuntimeError, match="foreign_keys"): module._assert_fk_enforcement_off(conn) conn.execute("ROLLBACK") finally: conn.close() def test_v011_recreate_invokes_fk_guard(tmp_path: Path) -> None: """End-to-end: the V011 apply path itself refuses on a live-FK connection inside a transaction (pre-V011 schema, rows present).""" db = str(tmp_path / "fk_guard_apply.db") _apply_through(db, V010_RUNS) conn = sqlite3.connect(db) conn.isolation_level = None try: conn.execute("BEGIN") inserted = conn.execute( "INSERT INTO meta_skill_runs (" "run_id, meta_skill_name, meta_skill_digest, plan_snapshot_json, " "triggered_by, status, started_at_ms, inputs_json" ") VALUES ('r1', 'synth', 'cafef00d', '{}', 'hard_takeover', " "'ok', 1000000, '{}')" ) assert inserted.rowcount == 1 conn.execute("COMMIT") conn.execute("PRAGMA foreign_keys = ON") conn.execute("BEGIN") module = _load_module(V011) with pytest.raises(RuntimeError, match="cascade-delete"): module.apply_step(conn) conn.execute("ROLLBACK") # Rows untouched by the refused rebuild. row = conn.execute( "SELECT run_id FROM meta_skill_runs WHERE run_id='r1'" ).fetchone() assert row == ("r1",) finally: conn.close()