ss-tools/scripts/migrate_catalog_native_result.py

# #region Tooling.CatalogNativeResultFixtureMigration [C:3] [TYPE Module] [SEMANTICS test,fixture,catalog,native,result,migration]
# @BRIEF Bring the published-catalog fixture's table entry up to the runtime DTO + native_result contract.
# @INVARIANT Only fields the contract requires are added and only contract-forbidden shapes are removed;
#            every value the entry already carried is preserved or re-derived by its production owner.
# @INVARIANT The fixture's outer identity tokens (entry_digest, coordinate_hash, catalog_digest,
#            baseline_family) are placeholders compared by equality, so they are proven unchanged
#            rather than recomputed.
# @RATIONALE require_entry_native_result refuses a table entry without native_result
#            (native_result_catalog.py:30-32) and require_runtime_value_policy refuses the canonical
#            YAML shape (native_result_catalog.py:83-87), which together produced every
#            NATIVE_RESULT_CATALOG_INVALID in the execution and registry runs.
# @REJECTED Adding only a native_result block was rejected: the entry also carried an outlawed
#           `policy` key, no `comparison_policy`, expected.canonical instead of canonical_value,
#           non-string canonical cells and provenance fields the contract does not declare.
"""Bring the table entry in the published-catalog fixture up to the runtime DTO contract.

Why: require_entry_native_result refuses an entry whose expected.kind is 'table' but which carries
no native_result block (native_result_catalog.py:30-32).  That refusal is the NATIVE_RESULT_CATALOG_INVALID
seen in the execution and registry runs.  Adding the block alone is not enough: the entry is in the
canonical YAML shape, and require_runtime_value_policy (native_result_catalog.py:83-87) refuses an
entry that has 'policy', lacks 'comparison_policy', lacks expected.canonical_value, or still carries
expected.canonical.  BaselineEntry also requires label, dataset_id and updated_at, and forbids extras.

Every derived value comes from the production helpers that define it, following the existing owner
tests/services/dashboard_testing/execution/native_pin_literal_fixture.py.  The fixture's outer identity
tokens (entry_digest, coordinate_hash, catalog_digest, baseline_family) are placeholders compared by
equality, not recomputed from the entry, so they are deliberately left alone - the round-trip check
below proves that claim rather than assuming it.

Run with --write to apply; without it the script only reports.
"""

from __future__ import annotations

import hashlib
import json
from pathlib import Path
import sys

BACKEND = Path(__file__).resolve().parents[0] / ".." / "backend"
sys.path.insert(0, str(BACKEND))

from src.schemas.dashboard_testing import NormalizedValue  # noqa: E402
from src.schemas.dashboard_testing.catalog import BaselineEntry  # noqa: E402
from src.services.dashboard_testing.native_reference_policy import native_policy_snapshot  # noqa: E402
from src.services.dashboard_testing.native_result_catalog import (  # noqa: E402
    require_entry_native_result,
    require_native_revision,
)
from src.services.dashboard_testing.result_values import (  # noqa: E402
    canonical_json,
    descriptor_sha256,
    require_result_descriptor,
    result_identity_sha256,
    validate_result_value,
)

PATH = Path(__file__).resolve().parents[0] / ".." / "specs" / "044-dashboard-scenario-execution" / "fixtures" / "production-contract-refresh.json"
TABLE_INDEX = 1


def sync_pin_native_result(data: dict) -> int:
    """The pin declares what resolution returns; the metric entry now carries a native_result."""
    by_baseline = {wrapper["baseline_id"]: wrapper["entry"]
                   for wrapper in data["catalog_revision"]["entry_revisions"]}
    synced = 0
    for pin_entry in data["baseline_pin"].get("entries", []):
        entry = by_baseline.get(pin_entry.get("baseline_id"))
        block = (entry or {}).get("native_result")
        if block is not None and pin_entry.get("native_result") != block:
            pin_entry["native_result"] = json.loads(json.dumps(block))
            synced += 1
    return synced


def canonical_holders(data: dict, entry: dict):
    """Every canonical table the fixture repeats: the entry's own and both comparison sides."""
    yield entry["expected"]["canonical"]
    for comparison in data.get("result", {}).get("comparisons", []):
        for side in ("actual", "expected"):
            holder = comparison.get(side, {}).get("canonical")
            if isinstance(holder, dict):
                yield holder


def build_descriptor(entry: dict, dataset_id: int, run_id: str, key_columns: list[str]):
    """Descriptor derived from the entry's own canonical table, not invented.

    `selected_metric` must appear among the descriptor's column names
    (result_descriptor.py:77), and the fixture's canonical data is the only place that named the
    metric column `revenue` while the entry's result_key, the evaluation criterion and the
    comparisons all call it `revenue_by_region`; the caller renames it before this runs.
    """
    canonical = entry["expected"]["canonical"]
    columns = canonical["columns"]
    codes = {name: (1 if name in key_columns else 0) for name in columns}
    request_sha = hashlib.sha256(canonical_json({"entry": entry["baseline_id"]})).hexdigest()
    return require_result_descriptor({
        "schema_version": 1,
        "canonicalization_version": 1,
        "kind": "table",
        "scope": "native_query_result",
        "chart_id": entry["chart_id"],
        "dataset_id": dataset_id,
        "selected_metric": entry["result_key"],
        "query_semantics_sha256": "c" * 64,
        "columns": [{"name": name, "generic_type_code": codes[name], "nullable": False} for name in columns],
        "key_columns": list(key_columns),
        "native_request": {"content_ref": f"draft:{run_id}:{request_sha}", "sha256": request_sha},
    })


def main() -> int:
    data = json.loads(PATH.read_text(encoding="utf-8"))
    revision = data["catalog_revision"]
    wrapper = revision["entry_revisions"][TABLE_INDEX]
    entry = wrapper["entry"]
    already = "native_result" in entry and "canonical" not in entry.get("expected", {})
    if already:
        # Idempotent second half: the declared pin must return what the resolver now produces from
        # this entry, which includes the native_result block the pin oracle did not carry before.
        synced = sync_pin_native_result(data)
        print(f"entry already migrated; pin entries synced: {synced}")
        if "--write" in sys.argv:
            PATH.write_text(json.dumps(data, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")
            print(f"written: {PATH}")
        return 0
    entry.setdefault("dataset_id", 4)
    canonical = entry["expected"]["canonical"]

    report: dict[str, object] = {
        "baseline_id": entry["baseline_id"],
        "chart_id": entry["chart_id"],
        "dataset_id": entry["dataset_id"],
        "result_key": entry["result_key"],
        "columns": canonical["columns"],
        "rows": len(canonical["rows"]),
        "had_policy_key": "policy" in entry,
        "had_canonical_key": "canonical" in entry["expected"],
        "outer_entry_digest": wrapper["entry_digest"],
        "outer_catalog_digest": revision["catalog_digest"],
    }

    run_id = entry["provenance"].get("source_artifact_id") or wrapper["capture_artifact_id"]
    key_columns = list(entry.get("policy", {}).get("key_columns", []))
    # The metric column is the one name the fixture got wrong: result_key, the evaluation criterion
    # and both comparison sides call it revenue_by_region, only the canonical data says "revenue".
    # A descriptor requires selected_metric to be a column, so the canonical data follows its own
    # declared metric everywhere it repeats the column list.
    renamed = 0
    for holder in canonical_holders(data, entry):
        if holder.get("columns") and entry["result_key"] not in holder["columns"]:
            holder["columns"] = [entry["result_key"] if name == "revenue" else name
                                 for name in holder["columns"]]
            renamed += 1
        # Canonical table cells are strings by contract: require_cell refuses a non-str cell
        # (result_values.py:71-72), and numeric columns must be decimal notation.
        holder["rows"] = [[None if cell is None else str(cell) for cell in row]
                          for row in holder.get("rows", [])]
    report["canonical_column_lists_renamed"] = renamed

    descriptor = build_descriptor(entry, entry["dataset_id"], run_id, key_columns)
    value = {"kind": "table", "canonical_value": json.dumps(canonical, separators=(",", ":"))}
    response_sha = entry["source_response_hash"]
    record = {
        "schema_version": 1,
        "descriptor": descriptor.model_dump(mode="json"),
        "descriptor_sha256": descriptor_sha256(descriptor),
        "result_identity_sha256": result_identity_sha256(descriptor),
        "capture_artifact_id": wrapper["capture_artifact_id"],
        "capture_content_ref": f"draft:{run_id}:{response_sha}",
        "source_response_hash": response_sha,
        "canonical_value_digest": validate_result_value(descriptor, NormalizedValue.model_validate(value)),
        "context_sha256": "b" * 64,
    }
    policy = native_policy_snapshot(descriptor)
    block = {
        "schema_version": 1,
        "authority_record": record,
        "authority_record_sha256": hashlib.sha256(canonical_json(record)).hexdigest(),
        "policy_snapshot": policy,
        "policy_sha256": hashlib.sha256(canonical_json(policy)).hexdigest(),
    }

    # Rewrite the entry into the runtime DTO shape the catalog contract declares.
    entry.pop("policy", None)
    entry["expected"] = value
    entry["comparison_policy"] = policy["comparison_policy"]
    entry["native_result"] = block
    entry.setdefault("label", "Revenue by region")
    entry.setdefault("updated_at", entry["created_at"])
    # Provenance allows exactly environment/actor/agent_run_id; the captured capture-binding fields
    # the fixture carried there are not part of that contract, and the block now owns the capture
    # identity (capture_artifact_id, capture_content_ref, source_response_hash).
    dropped = sorted(set(entry["provenance"]) - {"environment", "actor", "agent_run_id"})
    entry["provenance"] = {k: v for k, v in entry["provenance"].items()
                           if k in {"environment", "actor", "agent_run_id"}}
    report["provenance_fields_dropped"] = dropped

    # Proof, not diff review: the same validators the production path runs.
    require_entry_native_result(entry)
    BaselineEntry.model_validate_json(canonical_json(entry), strict=True)
    require_native_revision(revision)
    report["require_entry_native_result"] = "PASS"
    report["baseline_entry_strict"] = "PASS"
    report["require_native_revision"] = "PASS"

    # Prove the outer placeholder identities were not silently recomputed.
    report["entry_digest_unchanged"] = wrapper["entry_digest"] == report["outer_entry_digest"]
    report["catalog_digest_unchanged"] = revision["catalog_digest"] == report["outer_catalog_digest"]
    report["pin_unchanged"] = data["baseline_pin"]["catalog_digest"] == report["outer_catalog_digest"]

    for key, value in report.items():
        print(f"{key}: {value}")

    must_hold = ("entry_digest_unchanged", "catalog_digest_unchanged", "pin_unchanged")
    failed = [k for k in must_hold if report[k] is not True]
    if failed:
        raise SystemExit(f"verification failed: {failed}")

    if "--write" in sys.argv:
        PATH.write_text(json.dumps(data, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")
        print(f"written: {PATH}")
    else:
        print("dry run (pass --write to apply)")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())
# #endregion Tooling.CatalogNativeResultFixtureMigration