#!/usr/bin/env python3
"""Generate the synthetic AI-pipeline handoff fixtures and evidence manifest.

The two tiny Parquet files are deliberately healthy in isolation but inconsistent
when reconciled across stages. They are teaching data, not model output or a
representative production pipeline.

For a downloaded copy, keep ``ai-pipeline-handoff-audit.sql`` beside this script,
then run::

    python -m pip install duckdb==1.5.3
    python generate_ai_pipeline_handoff_fixtures.py

The fixtures and manifest are written beside the downloaded script. In this
repository, the source script writes to ``public/data/seo`` instead.
"""

from __future__ import annotations

import hashlib
import json
from pathlib import Path

import duckdb


SCRIPT_PATH = Path(__file__).resolve()
SQL_BUNDLE_NAME = "ai-pipeline-handoff-audit.sql"


def resolve_output_dir() -> Path:
    """Use the published artifact directory in-repo and the script directory elsewhere."""

    script_dir = SCRIPT_PATH.parent
    repository_root = script_dir.parent
    if script_dir.name == "scripts" and (repository_root / "package.json").is_file():
        return repository_root / "public" / "data" / "seo"
    return script_dir


OUTPUT_DIR = resolve_output_dir()
GENERATED_ON = "2026-08-24"

DOCUMENTS_FIXTURE = OUTPUT_DIR / "ai-pipeline-documents.parquet"
EMBEDDINGS_FIXTURE = OUTPUT_DIR / "ai-pipeline-embeddings.parquet"
MANIFEST = OUTPUT_DIR / "ai-pipeline-handoff-manifest.json"
SQL_BUNDLE = OUTPUT_DIR / SQL_BUNDLE_NAME
PUBLIC_GENERATOR = OUTPUT_DIR / "generate_ai_pipeline_handoff_fixtures.py"

DOCUMENTS_LOCAL_AUDIT_SQL = r"""
SELECT
  count(*) AS rows,
  count_if(document_id IS NULL OR trim(document_id) = '') AS missing_document_ids,
  count(*) FILTER (WHERE document_id IS NOT NULL AND trim(document_id) <> '')
    - count(DISTINCT document_id)
      FILTER (WHERE document_id IS NOT NULL AND trim(document_id) <> '')
    AS duplicate_document_ids,
  count_if(normalized_text IS NULL OR trim(normalized_text) = '') AS blank_text_rows,
  count_if(text_revision IS NULL OR trim(text_revision) = '') AS missing_text_revision_rows,
  count_if(normalization_run_id IS NULL OR trim(normalization_run_id) = '')
    AS missing_normalization_run_rows,
  count_if(source_uri IS NULL OR trim(source_uri) = '') AS missing_source_uri_rows,
  count_if(source_revision IS NULL OR trim(source_revision) = '')
    AS missing_source_revision_rows,
  count_if(license IS NULL OR trim(license) = '') AS missing_license_rows,
  count_if(is_synthetic IS DISTINCT FROM true) AS non_synthetic_rows
FROM read_parquet(?);
"""

EMBEDDINGS_LOCAL_AUDIT_SQL = r"""
WITH numbered AS (
  SELECT row_number() OVER () AS audit_row_number, *
  FROM read_parquet(?)
), vector_stats AS (
  SELECT
    audit_row_number,
    count_if(value IS NULL OR NOT isfinite(value)) AS non_finite_values,
    sum(value * value) AS squared_norm
  FROM numbered, UNNEST(embedding) AS vector(value)
  GROUP BY audit_row_number
)
SELECT
  count(*) AS rows,
  count_if(e.document_id IS NULL OR trim(e.document_id) = '') AS missing_document_ids,
  count(*) FILTER (WHERE e.document_id IS NOT NULL AND trim(e.document_id) <> '')
    - count(DISTINCT e.document_id)
      FILTER (WHERE e.document_id IS NOT NULL AND trim(e.document_id) <> '')
    AS duplicate_document_ids,
  count_if(e.source_normalization_run_id IS NULL OR trim(e.source_normalization_run_id) = '')
    AS missing_source_run_rows,
  count_if(e.text_revision IS NULL OR trim(e.text_revision) = '')
    AS missing_text_revision_rows,
  count_if(e.embedding_run_id IS NULL OR trim(e.embedding_run_id) = '')
    AS missing_embedding_run_rows,
  count_if(e.embedding_model_revision IS NULL OR trim(e.embedding_model_revision) = '')
    AS missing_model_revision_rows,
  count_if(e.embedding_model_revision IS DISTINCT FROM 'synthetic-3d-demo-v1')
    AS unexpected_model_revision_rows,
  count(DISTINCT e.embedding_model_revision)
    FILTER (WHERE e.embedding_model_revision IS NOT NULL)
    AS observed_model_revisions,
  count_if(e.source_uri IS NULL OR trim(e.source_uri) = '') AS missing_source_uri_rows,
  count_if(e.source_revision IS NULL OR trim(e.source_revision) = '')
    AS missing_source_revision_rows,
  count_if(e.license IS NULL OR trim(e.license) = '') AS missing_license_rows,
  count_if(e.embedding IS NULL) AS missing_embedding_rows,
  count_if(e.embedding IS NOT NULL AND array_length(e.embedding) <> 3)
    AS wrong_dimension_rows,
  count_if(coalesce(v.non_finite_values, 0) > 0) AS non_finite_rows,
  count_if(
    array_length(e.embedding) = 3
    AND coalesce(v.non_finite_values, 0) = 0
    AND v.squared_norm = 0
  ) AS zero_vector_rows,
  count_if(e.is_synthetic IS DISTINCT FROM true) AS non_synthetic_rows
FROM numbered AS e
LEFT JOIN vector_stats AS v USING (audit_row_number);
"""

CROSS_STAGE_AUDIT_SQL = r"""
SELECT
  count_if(d.document_id IS NULL) AS orphan_embeddings,
  count_if(e.document_id IS NULL) AS documents_without_embeddings,
  count_if(
    d.document_id IS NOT NULL
    AND e.document_id IS NOT NULL
    AND e.source_normalization_run_id IS DISTINCT FROM d.normalization_run_id
  ) AS mismatched_source_run_rows,
  count_if(
    d.document_id IS NOT NULL
    AND e.document_id IS NOT NULL
    AND e.text_revision IS DISTINCT FROM d.text_revision
  ) AS mismatched_text_revision_rows
FROM read_parquet(?) AS d
FULL OUTER JOIN read_parquet(?) AS e
  ON d.document_id = e.document_id;
"""

FAILING_IDS_SQL = r"""
WITH reconciled AS (
  SELECT
    coalesce(d.document_id, e.document_id) AS document_id,
    d.document_id AS documents_id,
    e.document_id AS embeddings_id,
    d.normalization_run_id,
    e.source_normalization_run_id,
    d.text_revision AS documents_text_revision,
    e.text_revision AS embeddings_text_revision
  FROM read_parquet(?) AS d
  FULL OUTER JOIN read_parquet(?) AS e
    ON d.document_id = e.document_id
), failures AS (
  SELECT document_id, 'orphan_embedding' AS issue
  FROM reconciled
  WHERE documents_id IS NULL
  UNION ALL
  SELECT document_id, 'document_without_embedding' AS issue
  FROM reconciled
  WHERE embeddings_id IS NULL
  UNION ALL
  SELECT document_id, 'source_normalization_run_mismatch' AS issue
  FROM reconciled
  WHERE documents_id IS NOT NULL
    AND embeddings_id IS NOT NULL
    AND source_normalization_run_id IS DISTINCT FROM normalization_run_id
  UNION ALL
  SELECT document_id, 'text_revision_mismatch' AS issue
  FROM reconciled
  WHERE documents_id IS NOT NULL
    AND embeddings_id IS NOT NULL
    AND embeddings_text_revision IS DISTINCT FROM documents_text_revision
)
SELECT document_id, issue
FROM failures
ORDER BY document_id, issue;
"""

EXPECTED_DOCUMENTS_FINDINGS = {
    "rows": 3,
    "missing_document_ids": 0,
    "duplicate_document_ids": 0,
    "blank_text_rows": 0,
    "missing_text_revision_rows": 0,
    "missing_normalization_run_rows": 0,
    "missing_source_uri_rows": 0,
    "missing_source_revision_rows": 0,
    "missing_license_rows": 0,
    "non_synthetic_rows": 0,
}

EXPECTED_EMBEDDINGS_FINDINGS = {
    "rows": 3,
    "missing_document_ids": 0,
    "duplicate_document_ids": 0,
    "missing_source_run_rows": 0,
    "missing_text_revision_rows": 0,
    "missing_embedding_run_rows": 0,
    "missing_model_revision_rows": 0,
    "unexpected_model_revision_rows": 0,
    "observed_model_revisions": 1,
    "missing_source_uri_rows": 0,
    "missing_source_revision_rows": 0,
    "missing_license_rows": 0,
    "missing_embedding_rows": 0,
    "wrong_dimension_rows": 0,
    "non_finite_rows": 0,
    "zero_vector_rows": 0,
    "non_synthetic_rows": 0,
}

EXPECTED_CROSS_STAGE_FINDINGS = {
    "orphan_embeddings": 1,
    "documents_without_embeddings": 1,
    "mismatched_source_run_rows": 2,
    "mismatched_text_revision_rows": 1,
}

EXPECTED_FAILING_IDS = [
    {"document_id": "doc-001", "issue": "source_normalization_run_mismatch"},
    {"document_id": "doc-001", "issue": "text_revision_mismatch"},
    {"document_id": "doc-002", "issue": "source_normalization_run_mismatch"},
    {"document_id": "doc-003", "issue": "orphan_embedding"},
    {"document_id": "doc-004", "issue": "document_without_embedding"},
]


def sql_path(path: Path) -> str:
    return path.as_posix().replace("'", "''")


def sha256(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def result_dict(
    connection: duckdb.DuckDBPyConnection,
    sql: str,
    params: list[str],
) -> dict[str, int]:
    cursor = connection.execute(sql, params)
    row = cursor.fetchone()
    if row is None:
        raise RuntimeError("Audit query returned no result")
    return {description[0]: int(value) for description, value in zip(cursor.description, row)}


def schema(connection: duckdb.DuckDBPyConnection, path: Path) -> list[dict[str, str]]:
    rows = connection.execute(
        "DESCRIBE SELECT * FROM read_parquet(?)",
        [str(path)],
    ).fetchall()
    return [{"name": row[0], "type": row[1], "nullable": row[2]} for row in rows]


def assert_findings(label: str, actual: object, expected: object) -> None:
    if actual != expected:
        raise RuntimeError(f"{label} changed: expected {expected!r}, got {actual!r}")


def main() -> None:
    OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
    if not SQL_BUNDLE.exists():
        raise FileNotFoundError(
            f"Missing companion SQL bundle: {SQL_BUNDLE.name}. "
            "Keep it beside the downloaded generator script."
        )

    connection = duckdb.connect()
    connection.execute(
        """
        CREATE OR REPLACE TABLE documents_fixture AS
        SELECT *
        FROM (VALUES
          ('doc-001',
           'Columnar files keep related values together for analytical scans.',
           'text-r2', 'norm-042',
           'https://example.test/synthetic/doc-001', 'source-r1', 'CC0-1.0',
           TIMESTAMP '2026-08-24 09:00:00', true),
          ('doc-002',
           'Lineage records connect each derived artifact to its exact source run.',
           'text-r7', 'norm-042',
           'https://example.test/synthetic/doc-002', 'source-r3', 'CC0-1.0',
           TIMESTAMP '2026-08-24 09:00:01', true),
          ('doc-004',
           'Release gates should reconcile documents and embeddings before promotion.',
           'text-r1', 'norm-042',
           'https://example.test/synthetic/doc-004', 'source-r1', 'CC0-1.0',
           TIMESTAMP '2026-08-24 09:00:02', true)
        ) AS rows(
          document_id, normalized_text, text_revision, normalization_run_id,
          source_uri, source_revision, license, normalized_at, is_synthetic
        );
        """
    )
    connection.execute(
        f"COPY documents_fixture TO '{sql_path(DOCUMENTS_FIXTURE)}' "
        "(FORMAT PARQUET, COMPRESSION ZSTD);"
    )

    connection.execute(
        """
        CREATE OR REPLACE TABLE embeddings_fixture AS
        SELECT *
        FROM (VALUES
          ('doc-001', 'text-r1', 'norm-041', 'embed-007',
           'synthetic-3d-demo-v1', [0.10::DOUBLE, 0.20, 0.30],
           'https://example.test/synthetic/doc-001', 'source-r1', 'CC0-1.0',
           TIMESTAMP '2026-08-24 09:05:00', true),
          ('doc-002', 'text-r7', 'norm-041', 'embed-007',
           'synthetic-3d-demo-v1', [0.40::DOUBLE, 0.50, 0.60],
           'https://example.test/synthetic/doc-002', 'source-r3', 'CC0-1.0',
           TIMESTAMP '2026-08-24 09:05:01', true),
          ('doc-003', 'text-r3', 'norm-041', 'embed-007',
           'synthetic-3d-demo-v1', [0.70::DOUBLE, 0.80, 0.90],
           'https://example.test/synthetic/doc-003', 'source-r2', 'CC0-1.0',
           TIMESTAMP '2026-08-24 09:05:02', true)
        ) AS rows(
          document_id, text_revision, source_normalization_run_id, embedding_run_id,
          embedding_model_revision, embedding, source_uri, source_revision, license,
          embedded_at, is_synthetic
        );
        """
    )
    connection.execute(
        f"COPY embeddings_fixture TO '{sql_path(EMBEDDINGS_FIXTURE)}' "
        "(FORMAT PARQUET, COMPRESSION ZSTD);"
    )

    documents_findings = result_dict(
        connection,
        DOCUMENTS_LOCAL_AUDIT_SQL,
        [str(DOCUMENTS_FIXTURE)],
    )
    embeddings_findings = result_dict(
        connection,
        EMBEDDINGS_LOCAL_AUDIT_SQL,
        [str(EMBEDDINGS_FIXTURE)],
    )
    cross_stage_findings = result_dict(
        connection,
        CROSS_STAGE_AUDIT_SQL,
        [str(DOCUMENTS_FIXTURE), str(EMBEDDINGS_FIXTURE)],
    )
    failing_cursor = connection.execute(
        FAILING_IDS_SQL,
        [str(DOCUMENTS_FIXTURE), str(EMBEDDINGS_FIXTURE)],
    )
    failing_ids = [
        {"document_id": row[0], "issue": row[1]}
        for row in failing_cursor.fetchall()
    ]

    assert_findings("documents local findings", documents_findings, EXPECTED_DOCUMENTS_FINDINGS)
    assert_findings("embeddings local findings", embeddings_findings, EXPECTED_EMBEDDINGS_FINDINGS)
    assert_findings(
        "cross-stage findings",
        cross_stage_findings,
        EXPECTED_CROSS_STAGE_FINDINGS,
    )
    assert_findings("failing IDs", failing_ids, EXPECTED_FAILING_IDS)

    PUBLIC_GENERATOR.write_text(SCRIPT_PATH.read_text(encoding="utf-8"), encoding="utf-8")

    manifest = {
        "purpose": (
            "Synthetic teaching fixtures for reconciling an AI document-to-embedding "
            "pipeline handoff"
        ),
        "generated_on": GENERATED_ON,
        "generator": "scripts/generate_ai_pipeline_handoff_fixtures.py",
        "duckdb_version": connection.execute("SELECT version()").fetchone()[0],
        "scenario": {
            "documents_normalization_run": "norm-042",
            "embeddings_declared_source_run": "norm-041",
            "documents_ids": ["doc-001", "doc-002", "doc-004"],
            "embeddings_ids": ["doc-001", "doc-002", "doc-003"],
            "shared_matching_text_revision_ids": ["doc-002"],
            "shared_stale_text_revision_ids": ["doc-001"],
            "embedding_model_revision": "synthetic-3d-demo-v1",
            "embedding_dimensions": 3,
        },
        "evidence_artifacts": [
            {
                "name": PUBLIC_GENERATOR.name,
                "role": "fixture_generator",
                "bytes": PUBLIC_GENERATOR.stat().st_size,
                "sha256": sha256(PUBLIC_GENERATOR),
            },
            {
                "name": SQL_BUNDLE.name,
                "role": "audit_sql_bundle",
                "bytes": SQL_BUNDLE.stat().st_size,
                "sha256": sha256(SQL_BUNDLE),
            },
        ],
        "local_identity_findings": {
            "documents": documents_findings,
            "embeddings": embeddings_findings,
        },
        "cross_stage_findings": cross_stage_findings,
        "failing_ids": failing_ids,
        "limitations": [
            "Tiny synthetic teaching data; not representative of production volume or distributions",
            "The three-dimensional vectors are invented and were not produced by an embedding model",
            "The example run IDs, timestamps, source URLs, revisions, and license labels are illustrative",
            "These checks cover identity and declared lineage, not semantic quality, consent, privacy, legal fitness, or model suitability",
            "Native DuckDB generated the evidence; the repository test separately executes every published SQL statement in the shipped DuckDB-WASM build",
        ],
        "files": [
            {
                "name": DOCUMENTS_FIXTURE.name,
                "role": "normalized_documents_fixture",
                "bytes": DOCUMENTS_FIXTURE.stat().st_size,
                "sha256": sha256(DOCUMENTS_FIXTURE),
                "schema": schema(connection, DOCUMENTS_FIXTURE),
                "expected_local_findings": documents_findings,
            },
            {
                "name": EMBEDDINGS_FIXTURE.name,
                "role": "embeddings_fixture",
                "bytes": EMBEDDINGS_FIXTURE.stat().st_size,
                "sha256": sha256(EMBEDDINGS_FIXTURE),
                "schema": schema(connection, EMBEDDINGS_FIXTURE),
                "expected_local_findings": embeddings_findings,
            },
        ],
    }
    MANIFEST.write_text(json.dumps(manifest, indent=2) + "\n", encoding="utf-8")
    print(json.dumps(manifest, indent=2))


if __name__ == "__main__":
    main()