script-build-and-framework-tmp/visualization/backend/scripts/validate_host_contracts.py

"""Validate fake records against the live parent Host and Agent contracts.

Run from visualization/backend with both parent source roots and Host dependencies
on PYTHONPATH.  This remains a verification command, never a runtime dependency.
"""

from __future__ import annotations

from app.contracts import SCHEMA_CATALOG
from app.fake_data import ARTIFACT_DIGESTS, GRAPH
from agent.orchestration.wire import (
    AttemptView,
    OperationView,
    PlannerDecisionView,
    TaskSpecView,
    TaskView,
    ValidationView,
)
from script_build_host.domain.phase_three_artifacts import (
    canonicalize_legacy_projection,
    hydrate_root_delivery_manifest,
    root_delivery_input_closure_digest,
)
from script_build_host.domain.phase_two_artifacts import hydrate_phase_two_artifact
from script_build_host.domain.task_contracts import ScriptTaskContractV1
from script_build_host.infrastructure.legacy_tables import (
    script_build_element,
    script_build_paragraph,
    script_build_paragraph_element,
)
from script_build_host.infrastructure.tables import (
    artifact_version_table,
    input_snapshot_table,
)
from script_build_host.repositories.sqlalchemy import _snapshot_from_row

WIRE_MODELS = {
    "TaskView": TaskView,
    "TaskSpecView": TaskSpecView,
    "OperationView": OperationView,
    "AttemptView": AttemptView,
    "ValidationView": ValidationView,
    "PlannerDecisionView": PlannerDecisionView,
}

PHASE_TWO_MODELS = {
    "StructureArtifactV1",
    "ParagraphArtifactV1",
    "ElementSetArtifactV1",
    "ComparisonArtifactV1",
    "StructuredScriptArtifactV1",
    "CandidatePortfolioArtifactV1",
}


def _assert_table_fields(model_name: str, table: object) -> None:
    columns = tuple(column.name for column in table.columns)  # type: ignore[attr-defined]
    assert SCHEMA_CATALOG[model_name] == columns, (model_name, columns)


def main() -> None:
    for node in GRAPH.nodes:
        if node.record.model_name in WIRE_MODELS:
            WIRE_MODELS[node.record.model_name].model_validate(node.record.payload)
        if node.record.model_name == "ScriptTaskContractV1":
            contract = ScriptTaskContractV1.from_payload(node.record.payload)
            if contract.task_kind.value in {"compose", "candidate-portfolio"}:
                contract.require_execution_ready()
        if node.record.model_name in PHASE_TWO_MODELS:
            hydrate_phase_two_artifact(node.record.payload)

    records = {node.record.model_name: node.record.payload for node in GRAPH.nodes}
    manifest = hydrate_root_delivery_manifest(records["RootDeliveryManifestV1"])
    structured = hydrate_phase_two_artifact(records["StructuredScriptArtifactV1"])
    canonical = canonicalize_legacy_projection(
        structured,
        direction=records["LegacyProjectionCanonicalV1"]["direction"],
        summary=records["LegacyProjectionCanonicalV1"]["summary"],
    )
    assert canonical.content_payload() == records["LegacyProjectionCanonicalV1"]
    assert manifest.legacy_projection_digest == canonical.canonical_sha256

    refs = records["RootDeliveryManifestV1"]
    assert manifest.input_closure_digest == root_delivery_input_closure_digest(
        direction_ref=refs["direction_ref"],
        direction_digest=ARTIFACT_DIGESTS[110],
        candidate_portfolio_ref=refs["candidate_portfolio_ref"],
        candidate_portfolio_digest=ARTIFACT_DIGESTS[150],
        structured_script_ref=refs["structured_script_ref"],
        structured_script_digest=ARTIFACT_DIGESTS[140],
    )

    snapshot_row = next(
        node.record.payload
        for node in GRAPH.nodes
        if node.record.model_name == "InputSnapshotRow"
    )
    hydrated_snapshot = _snapshot_from_row(snapshot_row)
    logical_snapshot = next(
        node.record.payload
        for node in GRAPH.nodes
        if node.record.model_name == "ScriptBuildInputSnapshotV1"
    )
    assert hydrated_snapshot.snapshot_id == logical_snapshot["snapshot_id"]
    assert hydrated_snapshot.canonical_sha256 == logical_snapshot["canonical_sha256"]

    _assert_table_fields("InputSnapshotRow", input_snapshot_table)
    _assert_table_fields("ArtifactVersionRow", artifact_version_table)
    _assert_table_fields("ParagraphRow", script_build_paragraph)
    _assert_table_fields("ElementRow", script_build_element)
    _assert_table_fields("ParagraphElementLinkRow", script_build_paragraph_element)
    print(f"validated {len(GRAPH.nodes)} fake nodes against live source contracts")


if __name__ == "__main__":
    main()