factoryscope.py

"""Single-call FactoryScope end-to-end acceptance execution."""

from __future__ import annotations

import csv
import io
import json
import platform
from dataclasses import asdict
from datetime import datetime
from hashlib import sha256
from importlib.metadata import version
from pathlib import Path
from zipfile import ZIP_DEFLATED, ZipFile, ZipInfo

from sklearn import __version__ as sklearn_version

from src.contracts import project_contract_snapshot
from src.decisions import load_numeric_decisions, validate_behavior_values, validate_numeric_decisions
from src.detect import (
    EvidenceStatus,
    FeatureContract,
    Window,
    calibrate_alerts,
    create_split_manifest,
    fit_isolation_forest,
)
from src.evaluate import Event, EventContract, evaluate_events
from src.ingest import ingest_members
from src.provenance import (
    LabelState,
    SamplingState,
    canonical_json,
    content_hash,
    create_label_manifest,
    sha256_file,
    verify_provenance_chain,
)
from src.report_agent import Claim, ReportStatus, app as report_app
from src.reproduction import create_reproduction_manifest, record_negative_result
from src.validate import QAStatus, validate_records


def _write_immutable(path: Path, data: bytes) -> None:
    if path.exists() and path.read_bytes() != data:
        raise ValueError(f"refusing to combine or overwrite an independent artifact: {path}")
    path.parent.mkdir(parents=True, exist_ok=True)
    if not path.exists():
        path.write_bytes(data)


def _fixture_archive(path: Path, values: dict[str, object]) -> None:
    timestamps = values["timestamps"]
    members = {
        "train.csv": f"ts,machine_id,channel,value\n{timestamps['train']},bearing-fixture,ch-1,{values['train_value']}\n",
        "dev.csv": f"ts,machine_id,channel,value\n{timestamps['dev']},bearing-fixture,ch-1,{values['dev_value']}\n",
        "blind.csv": f"ts,machine_id,channel,value\n{timestamps['blind']},bearing-fixture,ch-1,{values['blind_value']}\n",
    }
    buffer = io.BytesIO()
    with ZipFile(buffer, "w") as archive:
        for name, text in members.items():
            info = ZipInfo(name, tuple(values["zip_timestamp"]))
            info.compress_type = ZIP_DEFLATED
            archive.writestr(info, text)
    _write_immutable(path, buffer.getvalue())


def _code_revision() -> str:
    project_root = Path(__file__).parent
    code_files = (Path(__file__), project_root / "scripts/run_acceptance.py",
                  *sorted((project_root / "src").rglob("*.py")))
    return content_hash({
        str(path.relative_to(project_root)): sha256_file(path) for path in code_files
    })


def verify_e2e(*, output_dir: str | Path | None = None) -> dict[str, str]:
    """Create and verify one causally closed local-fixture acceptance run."""
    project_root = Path(__file__).parent
    code_revision = _code_revision()
    decisions = load_numeric_decisions()
    snapshot = project_contract_snapshot()
    ndr_version = sha256_file(project_root / "docs/research/numeric-decisions.json")
    environment = {
        "python": platform.python_version(), "platform": platform.platform(),
        "scikit-learn": sklearn_version, "numpy": version("numpy"), "scipy": version("scipy"),
        "uv.lock": sha256_file(project_root / "uv.lock"), "verification_mode": "fixture",
    }
    run_context_hash = content_hash({
        "code_revision": code_revision,
        "contract_snapshot": snapshot.manifest_sha256,
        "numeric_decision_register": ndr_version,
        "environment": environment,
    })
    output = Path(output_dir or Path("artifacts/acceptance") / run_context_hash).resolve()
    by_decision = {item.decision_id: item for item in decisions}
    data_decision = by_decision["NDR-DEV-ACCEPTANCE-DATA-001"]
    if not isinstance(data_decision.value, dict):
        raise ValueError("acceptance fixture data decision must be a mapping")
    data_values = data_decision.value
    registered_data = validate_numeric_decisions(
        (data_decision,), (data_decision.decision_id,), consumer="QA"
    )
    validate_behavior_values(registered_data, data_values)
    archive_path = output / "factoryscope-fixture.zip"
    _fixture_archive(archive_path, data_values)
    archive_digest = sha256_file(archive_path)
    member_metadata = {
        f"{split}.csv": {
            "test_id": "offline-e2e-fixture", "run_id": f"fixture-{split}",
            "bearing_id": "bearing-fixture", "channel": "ch-1", "sensor_axis": "x",
            "sensor_location": "fixture", "unit": "fixture-value",
            "record_length": data_values["record_length"],
            "timestamp_semantics": "UTC fixture timestamp", "sampling_hz": None,
            "sampling_state": SamplingState.UNKNOWN,
            "measurement_time": data_values["timestamps"][split],
        }
        for split in ("train", "dev", "blind")
    }
    source, measurements = ingest_members(
        archive_path=str(archive_path), archive_url=archive_path.as_uri(),
        expected_sha256=archive_digest, members=member_metadata, verification_mode="fixture",
        collected_at=datetime.fromisoformat(data_values["collected_at"].replace("Z", "+00:00")),
    )
    labels = tuple(create_label_manifest(item, state=LabelState.UNKNOWN, label=None, evidence=None)
                   for item in measurements)
    verify_provenance_chain(source, measurements, labels)

    records = []
    with ZipFile(archive_path) as archive:
        by_member = {item.member_path: item for item in measurements}
        for name in sorted(archive.namelist()):
            for row in csv.DictReader(io.TextIOWrapper(archive.open(name), encoding="utf-8")):
                records.append({**row, "value": float(row["value"]),
                                "measurement_id": by_member[name].measurement_id})
    qa = validate_records(records, contract_snapshot_id=snapshot.contract_snapshot_id)
    if qa.status is not QAStatus.SUCCESS:
        raise ValueError("fixture QA did not succeed")

    split_decision = by_decision["NDR-DEV-ACCEPTANCE-SPLIT-001"]
    model_decision = by_decision["NDR-DEV-IF-001"]
    serialization_decision = by_decision["NDR-MODEL-SERIALIZATION-001"]
    alert_decision = by_decision["NDR-DEV-ACCEPTANCE-ALERT-001"]
    event_decision = by_decision["NDR-DEV-ACCEPTANCE-EVENT-001"]
    uncertainty_decision = by_decision["NDR-DEV-ACCEPTANCE-UNCERTAINTY-001"]
    citation_decision = by_decision["NDR-CITATION-001"]
    windows = tuple(Window.create(item.measurement_id, item.run_id, item.measurement_time or "", item.measurement_time or "")
                    for item in measurements)
    partition_by_member = {"train.csv": "train", "dev.csv": "dev", "blind.csv": "blind"}
    split = create_split_manifest(
        windows, {item.run_id: partition_by_member[item.member_path] for item in measurements},
        version="fixture-split-v1", minimum_independent_runs=split_decision.value,
        decision_id=split_decision.decision_id, decisions=(split_decision,),
        contract_snapshot_id=snapshot.contract_snapshot_id,
    )
    feature_contract = FeatureContract.create(
        "fixture-raw-value-v1", ("raw fixture value",), (model_decision.decision_id,),
        contract_snapshot_id=snapshot.contract_snapshot_id,
    )
    value_by_measurement = {row["measurement_id"]: row["value"] for row in qa.accepted}
    feature_rows = {window.window_id: (value_by_measurement[window.measurement_id],) for window in windows}
    fitted = fit_isolation_forest(
        feature_contract=feature_contract, split_manifest=split, feature_rows=feature_rows,
        configuration=model_decision.value, decision_ids=(model_decision.decision_id,),
        decisions=(model_decision, serialization_decision),
    )
    if fitted["status"] is not EvidenceStatus.READY:
        raise ValueError(f"fixture model was not fit: {fitted['reason']}")
    raw_scores = fitted["artifact"]
    model_path = output / "isolation-forest.pkl"
    _write_immutable(model_path, fitted["model_bytes"])
    if sha256_file(model_path) != raw_scores.model_sha256:
        raise ValueError("preserved model bytes do not match raw-score evidence")

    dev_window_ids = tuple(window_id for window_id, part in split.assignments if part == "dev")
    if not isinstance(alert_decision.value, dict):
        raise ValueError("acceptance alert decision must be a mapping")
    alert_rule = alert_decision.value
    if (
        alert_rule.get("method") != "single_dev_raw_score"
        or alert_rule.get("comparison") != "score <= threshold"
    ):
        raise ValueError("unsupported acceptance alert comparison")
    calibration = calibrate_alerts(
        raw_scores, version="fixture-dev-calibration-v1", dev_lineage=dev_window_ids,
        threshold=None, scope=alert_decision.scope,
        configuration={"threshold_derivation": alert_rule["method"]},
        decision_ids=(alert_decision.decision_id,), decisions=(alert_decision,),
        split_manifest=split,
    )
    run_by_window = {window.window_id: window.run_id for window in windows}
    assignment_by_window = dict(split.assignments)
    blind_scores = tuple(
        (window_id, score)
        for window_id, score in raw_scores.scores
        if assignment_by_window[window_id] == "blind"
    )
    alert_window_ids = tuple(
        window_id for window_id, score in blind_scores if score <= calibration.threshold
    )
    if (
        not isinstance(event_decision.value, dict)
        or event_decision.value.get("timeline") != "zero_based_blind_alert_ordinal"
    ):
        raise ValueError("unsupported acceptance event timeline")
    alerts = tuple(
        Event.create(run_by_window[window_id], float(index), float(index),
                     (raw_scores.raw_score_artifact_id, calibration.calibration_artifact_id, window_id),
                     contract_snapshot_id=snapshot.contract_snapshot_id)
        for index, window_id in enumerate(alert_window_ids)
    )
    event_contract = EventContract.create(
        version="fixture-event-v1", merge_gap=event_decision.value["merge_gap"],
        match_tolerance=event_decision.value["match_tolerance"], exposure_unit="fixture-row",
        decision_ids=(event_decision.decision_id,), decisions=(event_decision,),
        contract_snapshot_id=snapshot.contract_snapshot_id,
    )
    if not isinstance(uncertainty_decision.value, dict):
        raise ValueError("acceptance uncertainty decision must be a mapping")
    eligible_normal_exposure = uncertainty_decision.value["eligible_normal_exposure"]
    validate_behavior_values(
        {uncertainty_decision.decision_id: uncertainty_decision},
        {"eligible_normal_exposure": eligible_normal_exposure},
    )
    evaluation = evaluate_events(
        alerts, (), contract=event_contract, eligible_normal_exposure=eligible_normal_exposure,
        exposure_lineage=(calibration.calibration_artifact_id,),
        uncertainty_decision_ids=(uncertainty_decision.decision_id,),
        decisions=(uncertainty_decision,), split_manifest=split,
    )
    upstream = (source, *measurements, *labels, split, raw_scores, calibration, *alerts,
                evaluation.event_match_set, evaluation.eligible_exposure_set, evaluation)
    reproduction = create_reproduction_manifest(
        command=("python", "scripts/run_acceptance.py"),
        code_revision=code_revision,
        source_hashes=(source.archive_sha256, source.manifest_sha256),
        manifest_versions=("source-v1", "measurement-v1", "label-v1", split.version),
        configuration={"verification_mode": "fixture", "alert_comparison": alert_rule["comparison"]},
        numeric_decision_register_version=ndr_version,
        artifact_hashes=(qa.artifact_sha256, raw_scores.model_sha256, raw_scores.output_sha256,
                         calibration.output_sha256, evaluation.output_sha256),
        environment=environment, contract_snapshot=snapshot,
        contract_artifacts=upstream,
        source_archive_paths={source.source_record_id: archive_path},
    )
    claim = Claim.create("The local fixture establishes detector performance.", "official NASA archive", (),
                         required=True, claim_key="detector-performance")
    negatives = (
        record_negative_result(reproduction, status="INSUFFICIENT_EVIDENCE",
                               failure_path="official event evaluation sample is unavailable",
                               cause_artifact_id=evaluation.evaluation_id),
        record_negative_result(reproduction, status="ABSTAIN",
                               failure_path="performance claim has no official archive citation",
                               cause_artifact_id=claim.claim_id),
    )
    report = report_app.invoke({
        "reproduction": reproduction,
        "claims": (claim,),
        "evidence": (),
        "negative_results": negatives,
        "citation_decision_ids": (citation_decision.decision_id,),
        "decisions": (citation_decision,),
        "evaluation": evaluation,
    })["report"]
    if evaluation.status is not EvidenceStatus.INSUFFICIENT_EVIDENCE or report.report_status is not ReportStatus.ABSTAIN:
        raise ValueError("fixture evidence boundaries were promoted to a performance claim")
    if report.negative_results != negatives:
        raise ValueError("report did not preserve the independent acceptance negative results")

    negative_payload = tuple(asdict(item) for item in report.negative_results)
    negative_bytes = canonical_json(negative_payload)
    negative_path = output / "negative-results.json"
    _write_immutable(negative_path, negative_bytes)
    negative_hash = sha256(negative_bytes).hexdigest()
    manifest_payload = {
        "execution_id": reproduction.execution_id,
        "pipeline": ("source", "measurement_label", "qa", "split", "isolation_forest_raw_score",
                     "dev_calibration", "event_metrics", "report"),
        "source_lineage": {"verification_mode": "fixture", "source_record_id": source.source_record_id,
                           "manifest_sha256": source.manifest_sha256,
                           "official_archive_validation": "NOT_EVALUATED"},
        "measurements": tuple((item.measurement_id, item.manifest_sha256) for item in measurements),
        "labels": tuple((item.label_id, item.manifest_sha256) for item in labels),
        "qa_result": {"status": qa.status, "artifact_sha256": qa.artifact_sha256},
        "split_lineage": {"split_manifest_id": split.split_id, "content_sha256": split.content_sha256,
                          "assignments": split.assignments},
        "feature_contract": asdict(feature_contract),
        "numeric_decisions": {"decision_ids": tuple(item.decision_id for item in
            (data_decision, split_decision, model_decision, serialization_decision,
             alert_decision, event_decision, uncertainty_decision, citation_decision)),
            "snapshot_sha256": reproduction.numeric_decision_register_version},
        "model_evidence": {"artifact": model_path.name, "sha256": raw_scores.model_sha256,
                           "scikit_learn_version": fitted["sklearn_version"],
                           "parameters": fitted["model_parameters"],
                           "numeric_decision_snapshot_hash": fitted["numeric_decision_snapshot_hash"]},
        "raw_scores": {"artifact_id": raw_scores.raw_score_artifact_id, "rows": raw_scores.scores,
                       "input_window_set_hash": raw_scores.input_window_set_hash,
                       "output_sha256": raw_scores.output_sha256},
        "dev_calibration": {"artifact_id": calibration.calibration_artifact_id,
                            "raw_score_artifact_id": calibration.raw_score_artifact_id,
                            "threshold": calibration.threshold,
                            "threshold_derivation": alert_rule["method"],
                            "comparison": alert_rule["comparison"],
                            "output_sha256": calibration.output_sha256},
        "event_metrics": {"evaluation_id": evaluation.evaluation_id, "status": evaluation.status,
                          "alert_window_ids": alert_window_ids,
                          "evaluation_population": "blind",
                          "timeline": event_decision.value["timeline"],
                          "output_sha256": evaluation.output_sha256},
        "report": {"report_id": report.report_id, "status": report.report_status,
                   "artifact_sha256": report.artifact_sha256,
                   "reproduction_manifest_sha256": report.reproduction_manifest_sha256},
        "reproduction_manifest": {"manifest_sha256": reproduction.manifest_sha256,
                                  "contract_artifact_records": reproduction.contract_artifact_records,
                                  "source_verification_records": reproduction.source_verification_records},
        "negative_results": {"artifact": negative_path.name, "sha256": negative_hash,
                             "count": len(negatives)},
        "performance_evaluation": "NOT_EVALUATED",
    }
    if set(dict(split.assignments)) != {window.window_id for window in windows} or set(dict(raw_scores.scores)) != set(dict(split.assignments)):
        raise ValueError("raw score row identity is not reconstructable from split lineage")
    if calibration.raw_score_artifact_id != raw_scores.raw_score_artifact_id or report.reproduction_manifest_sha256 != reproduction.manifest_sha256:
        raise ValueError("downstream artifact lineage is broken")
    if any(item.execution_id != reproduction.execution_id for item in negatives):
        raise ValueError("negative result belongs to another execution")
    if sha256(negative_path.read_bytes()).hexdigest() != negative_hash:
        raise ValueError("negative result reference hash failed verification")
    manifest_bytes = canonical_json(manifest_payload)
    manifest_path = output / "acceptance-manifest.json"
    _write_immutable(manifest_path, manifest_bytes)
    manifest_hash = sha256(manifest_path.read_bytes()).hexdigest()
    if manifest_hash != content_hash(json.loads(manifest_path.read_text(encoding="utf-8"))):
        raise ValueError("acceptance manifest content hash failed verification")
    return {"execution_id": reproduction.execution_id, "status": "SUCCESS",
            "manifest": str(manifest_path), "manifest_sha256": manifest_hash,
            "negative_results": str(negative_path), "negative_results_sha256": negative_hash}