cleveragents-core/robot/helper_schema_parity_migration.py

"""Robot helper for schema-parity migration verification.

Checks the migration-produced SQLite schema for:

1. ``resource_links.link_type`` defaulting to ``contains``.
2. ``checkpoint_metadata`` foreign keys to ``decisions`` and ``resources``.
3. ``idx_decisions_superseded`` partial index with
   ``WHERE superseded_by IS NOT NULL``.
4. Runtime SQLite foreign key enforcement for ``checkpoint_metadata``.

Subcommands (each independent for isolated failure reporting):

- ``schema-parity-link-type``: Verifies resource_links.link_type column
  and default.
- ``schema-parity-fks``: Verifies checkpoint_metadata FK constraints and
  runtime enforcement (orphan rejection + positive path).
- ``schema-parity-index``: Verifies idx_decisions_superseded partial index.
"""

from __future__ import annotations

import sys
import tempfile
from collections.abc import Callable
from pathlib import Path
from typing import Any

from sqlalchemy import create_engine, inspect, text
from sqlalchemy.exc import IntegrityError

from cleveragents.infrastructure.database.migration_runner import MigrationRunner


def _cleanup_db_files(path: str) -> None:
    Path(path).unlink(missing_ok=True)
    Path(f"{path}-wal").unlink(missing_ok=True)
    Path(f"{path}-shm").unlink(missing_ok=True)


def _setup_migrated_db() -> tuple[Any, Any, str]:
    """Create a temp DB, run migrations, return (engine, inspector, db_path)."""
    with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as tmp_file:
        db_path = tmp_file.name
    db_url = f"sqlite:///{db_path}"

    MigrationRunner(db_url).init_or_upgrade()
    engine = create_engine(db_url, connect_args={"check_same_thread": False})
    inspector = inspect(engine)
    return engine, inspector, db_path


def _teardown_db(engine: Any, db_path: str) -> None:
    if engine is not None:
        engine.dispose()
    _cleanup_db_files(db_path)


def _ensure_test_action_and_plan(conn: Any) -> None:
    """Insert prerequisite action and plan rows for FK tests.

    .. note::
        This function is **not idempotent** — it performs blind INSERTs
        without existence checks.  It is safe only when called against a
        freshly created database (as ``_setup_migrated_db`` provides).
        If idempotent behaviour is needed, see the Behave counterpart in
        ``features/steps/db_schema_parity_steps.py``.
    """
    conn.execute(
        text(
            """
            INSERT INTO actions (
                namespaced_name, namespace, name, description,
                definition_of_done, strategy_actor, execution_actor,
                created_at, updated_at
            ) VALUES (
                :namespaced_name, :namespace, :name, :description,
                :definition_of_done, :strategy_actor, :execution_actor,
                :created_at, :updated_at
            )
            """
        ),
        {
            "namespaced_name": "local/test-action",
            "namespace": "local",
            "name": "test-action",
            "description": "test action",
            "definition_of_done": "test dod",
            "strategy_actor": "local/strategy",
            "execution_actor": "local/execution",
            "created_at": "2026-01-01T00:00:00",
            "updated_at": "2026-01-01T00:00:00",
        },
    )
    conn.execute(
        text(
            """
            INSERT INTO v3_plans (
                plan_id, root_plan_id, action_name,
                namespaced_name, namespace,
                description, created_at, updated_at
            ) VALUES (
                :plan_id, :root_plan_id, :action_name,
                :namespaced_name, :namespace,
                :description, :created_at, :updated_at
            )
            """
        ),
        {
            "plan_id": "01ARZ3NDEKTSV4RRFFQ69G5FAV",
            "root_plan_id": "01ARZ3NDEKTSV4RRFFQ69G5FAV",
            "action_name": "local/test-action",
            "namespaced_name": "local/test-plan",
            "namespace": "local",
            "description": "test plan",
            "created_at": "2026-01-01T00:00:00",
            "updated_at": "2026-01-01T00:00:00",
        },
    )


# ---------------------------------------------------------------------------
# Subcommand: schema-parity-link-type
# ---------------------------------------------------------------------------


def _schema_parity_link_type() -> None:
    engine, inspector, db_path = _setup_migrated_db()
    try:
        resource_link_columns = inspector.get_columns("resource_links")
        link_type = next(
            (
                column
                for column in resource_link_columns
                if column["name"] == "link_type"
            ),
            None,
        )
        assert link_type is not None, "resource_links.link_type column is missing"
        default = str(link_type.get("default") or "").lower()
        assert "contains" in default, (
            "resource_links.link_type default must include 'contains', "
            f"got {link_type.get('default')!r}"
        )

        print("schema-parity-link-type-ok")
    finally:
        _teardown_db(engine, db_path)


# ---------------------------------------------------------------------------
# Subcommand: schema-parity-fks
# ---------------------------------------------------------------------------


def _schema_parity_fks() -> None:
    engine, inspector, db_path = _setup_migrated_db()
    try:
        checkpoint_fks = inspector.get_foreign_keys("checkpoint_metadata")
        signatures = {
            (
                tuple(fk.get("constrained_columns") or []),
                fk.get("referred_table"),
                tuple(fk.get("referred_columns") or []),
            )
            for fk in checkpoint_fks
        }

        assert (("decision_id",), "decisions", ("decision_id",)) in signatures, (
            "Missing checkpoint_metadata FK decision_id -> decisions.decision_id"
        )
        assert (("resource_id",), "resources", ("resource_id",)) in signatures, (
            "Missing checkpoint_metadata FK resource_id -> resources.resource_id"
        )

        with engine.begin() as conn:
            _ensure_test_action_and_plan(conn)

            try:
                conn.execute(
                    text(
                        """
                        INSERT INTO checkpoint_metadata (
                            checkpoint_id, plan_id, decision_id,
                            checkpoint_type, resource_id, sandbox_ref,
                            filesystem_path, created_at
                        ) VALUES (
                            :checkpoint_id, :plan_id, :decision_id,
                            :checkpoint_type, :resource_id, :sandbox_ref,
                            :filesystem_path, :created_at
                        )
                        """
                    ),
                    {
                        "checkpoint_id": "01ARZ3NDEKTSV4RRFFQ69G5FAW",
                        "plan_id": "01ARZ3NDEKTSV4RRFFQ69G5FAV",
                        "decision_id": "01ARZ3NDEKTSV4RRFFQ69G5FAX",
                        "checkpoint_type": "manual",
                        "resource_id": "01ARZ3NDEKTSV4RRFFQ69G5FAY",
                        "sandbox_ref": "test-ref",
                        "filesystem_path": "",
                        "created_at": "2026-01-01T00:00:00",
                    },
                )
            except IntegrityError:
                pass
            else:
                raise AssertionError(
                    "checkpoint_metadata accepted orphan decision/resource references"
                )

        # Verify each FK independently: orphan decision_id only
        with engine.begin() as conn:
            try:
                conn.execute(
                    text(
                        """
                        INSERT INTO checkpoint_metadata (
                            checkpoint_id, plan_id, decision_id,
                            checkpoint_type, sandbox_ref,
                            filesystem_path, created_at
                        ) VALUES (
                            :checkpoint_id, :plan_id, :decision_id,
                            :checkpoint_type, :sandbox_ref,
                            :filesystem_path, :created_at
                        )
                        """
                    ),
                    {
                        "checkpoint_id": "01ARZ3NDEKTSV4RRFFQ69G5FC0",
                        "plan_id": "01ARZ3NDEKTSV4RRFFQ69G5FAV",
                        "decision_id": "01ARZ3NDEKTSV4RRFFQ69G5FC1",
                        "checkpoint_type": "manual",
                        "sandbox_ref": "test-ref",
                        "filesystem_path": "",
                        "created_at": "2026-01-01T00:00:00",
                    },
                )
            except IntegrityError:
                pass
            else:
                raise AssertionError(
                    "checkpoint_metadata accepted orphan decision_id independently"
                )

        # Verify each FK independently: orphan resource_id only
        with engine.begin() as conn:
            try:
                conn.execute(
                    text(
                        """
                        INSERT INTO checkpoint_metadata (
                            checkpoint_id, plan_id, resource_id,
                            checkpoint_type, sandbox_ref,
                            filesystem_path, created_at
                        ) VALUES (
                            :checkpoint_id, :plan_id, :resource_id,
                            :checkpoint_type, :sandbox_ref,
                            :filesystem_path, :created_at
                        )
                        """
                    ),
                    {
                        "checkpoint_id": "01ARZ3NDEKTSV4RRFFQ69G5FC2",
                        "plan_id": "01ARZ3NDEKTSV4RRFFQ69G5FAV",
                        "resource_id": "01ARZ3NDEKTSV4RRFFQ69G5FC3",
                        "checkpoint_type": "manual",
                        "sandbox_ref": "test-ref",
                        "filesystem_path": "",
                        "created_at": "2026-01-01T00:00:00",
                    },
                )
            except IntegrityError:
                pass
            else:
                raise AssertionError(
                    "checkpoint_metadata accepted orphan resource_id independently"
                )

        # Positive test: valid FK references should be accepted
        with engine.begin() as conn:
            decision_id = "01ARZ3NDEKTSV4RRFFQ69G5FC4"
            resource_id = "01ARZ3NDEKTSV4RRFFQ69G5FC5"

            conn.execute(
                text(
                    """
                    INSERT INTO decisions (
                        decision_id, plan_id, decision_type, question,
                        chosen_option, context_snapshot_json, sequence_number,
                        created_at
                    ) VALUES (
                        :decision_id, :plan_id, :decision_type, :question,
                        :chosen_option, :context_snapshot_json, :sequence_number,
                        :created_at
                    )
                    """
                ),
                {
                    "decision_id": decision_id,
                    "plan_id": "01ARZ3NDEKTSV4RRFFQ69G5FAV",
                    "decision_type": "strategy_choice",
                    "question": "test question",
                    "chosen_option": "test option",
                    "context_snapshot_json": "{}",
                    "sequence_number": 1,
                    "created_at": "2026-01-01T00:00:00",
                },
            )

            resource_columns = {
                col["name"] for col in inspect(engine).get_columns("resources")
            }
            cols = "resource_id, type_name, resource_kind, created_at, updated_at"
            vals = ":resource_id, :type_name, :resource_kind, :created_at, :updated_at"
            res_params: dict[str, str] = {
                "resource_id": resource_id,
                "type_name": "git-checkout",
                "resource_kind": "physical",
                "created_at": "2026-01-01T00:00:00",
                "updated_at": "2026-01-01T00:00:00",
            }
            if "namespaced_name" in resource_columns:
                cols = (
                    "resource_id, namespaced_name, type_name,"
                    " resource_kind, created_at, updated_at"
                )
                vals = (
                    ":resource_id, :namespaced_name, :type_name,"
                    " :resource_kind, :created_at, :updated_at"
                )
                res_params["namespaced_name"] = f"local/{resource_id}"
            conn.execute(
                text(f"INSERT INTO resources ({cols}) VALUES ({vals})"),
                res_params,
            )

            conn.execute(
                text(
                    """
                    INSERT INTO checkpoint_metadata (
                        checkpoint_id, plan_id, decision_id,
                        checkpoint_type, resource_id, sandbox_ref,
                        filesystem_path, created_at
                    ) VALUES (
                        :checkpoint_id, :plan_id, :decision_id,
                        :checkpoint_type, :resource_id, :sandbox_ref,
                        :filesystem_path, :created_at
                    )
                    """
                ),
                {
                    "checkpoint_id": "01ARZ3NDEKTSV4RRFFQ69G5FC6",
                    "plan_id": "01ARZ3NDEKTSV4RRFFQ69G5FAV",
                    "decision_id": decision_id,
                    "checkpoint_type": "manual",
                    "resource_id": resource_id,
                    "sandbox_ref": "test-ref",
                    "filesystem_path": "",
                    "created_at": "2026-01-01T00:00:00",
                },
            )

        print("schema-parity-fks-ok")
    finally:
        _teardown_db(engine, db_path)


# ---------------------------------------------------------------------------
# Subcommand: schema-parity-index
# ---------------------------------------------------------------------------


def _schema_parity_index() -> None:
    engine, inspector, db_path = _setup_migrated_db()
    try:
        decision_indexes = inspector.get_indexes("decisions")
        partial_index = next(
            (
                index
                for index in decision_indexes
                if index.get("name") == "idx_decisions_superseded"
            ),
            None,
        )
        assert partial_index is not None, "idx_decisions_superseded is missing"
        assert list(partial_index.get("column_names") or []) == ["superseded_by"], (
            "idx_decisions_superseded must index decisions.superseded_by"
        )

        with engine.connect() as conn:
            row = conn.execute(
                text(
                    "SELECT sql FROM sqlite_master "
                    "WHERE type = 'index' AND name = :index_name"
                ),
                {"index_name": "idx_decisions_superseded"},
            ).fetchone()

        assert row is not None, "sqlite_master is missing idx_decisions_superseded"
        sql = str(row[0] or "").lower()
        assert "where superseded_by is not null" in sql, (
            "idx_decisions_superseded is not a partial index"
        )

        print("schema-parity-index-ok")
    finally:
        _teardown_db(engine, db_path)


# ---------------------------------------------------------------------------
# Legacy combined subcommand (kept for backward compatibility)
# ---------------------------------------------------------------------------


def _schema_parity() -> None:
    _schema_parity_link_type()
    _schema_parity_fks()
    _schema_parity_index()
    print("schema-parity-ok")


_COMMANDS: dict[str, Callable[[], None]] = {
    "schema-parity": _schema_parity,
    "schema-parity-link-type": _schema_parity_link_type,
    "schema-parity-fks": _schema_parity_fks,
    "schema-parity-index": _schema_parity_index,
}


def main() -> None:
    if len(sys.argv) < 2:
        raise SystemExit(f"Expected command argument. Valid: {', '.join(_COMMANDS)}")

    command = sys.argv[1]
    handler = _COMMANDS.get(command)
    if handler is None:
        raise SystemExit(f"Unknown command: {command}. Valid: {', '.join(_COMMANDS)}")

    handler()


if __name__ == "__main__":
    main()