src/eval_archive.py

from __future__ import annotations

import json
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

from .bucket import RunPaths, read_json, read_text, write_json
from .eval_config import effective_eval_config
from .security import redact

_TERMINAL = {"success", "done", "completed", "failed", "failure", "error", "cancelled", "canceled", "manual", "stale", "partial", "partial_validation", "completed_with_warnings", "success_with_warnings", "validated", "validated_after_manual_space_test", "recovered_by_manual_validation", "validated_after_stale_run", "manual_validation_passed", "full_inference_success", "full_inference_candidate_health_passed", "health_only", "technical_blocker", "technical_blocker_boot_only", "manual_hardware_required"}


def _now() -> str:
    return datetime.now(timezone.utc).isoformat()


def _safe_segment(value: Any, default: str) -> str:
    cleaned = "".join(ch if ch.isalnum() or ch in {"-", "_"} else "-" for ch in str(value or "").strip())
    return cleaned.strip("-_") or default


def _eval_mount(cfg: dict[str, Any] | None = None) -> Path | None:
    cfg = cfg or effective_eval_config()
    if not cfg.get("enabled"):
        return None
    mount = Path(str(cfg.get("job_mount_path") or "/evals"))
    return mount


def _publish_status_path(paths: RunPaths) -> str:
    return f"{paths.root}/eval_publish_status.json"


def _publish_result(paths: RunPaths, result: dict[str, Any], *, token: str | None = None) -> dict[str, Any]:
    """Persist a best-effort publish status next to the run artifacts.

    This status is deliberately separate from the worker-local ``eval_record.json``:
    it tells the UI whether the backend archive copy was actually attempted and
    whether a file was written to the operator eval bucket.
    """
    payload = {
        "schema_version": "1.4",
        "checked_at": _now(),
        "publish_mode": "backend",
        "attempted": bool(result.get("attempted")),
        **result,
    }
    try:
        write_json(_publish_status_path(paths), payload, token=token)
    except Exception as exc:  # noqa: BLE001
        payload["status_write_warning"] = redact(str(exc))[:500]
    return payload



def _first_successful_linked_validation(payload: dict[str, Any] | None) -> dict[str, Any]:
    if not isinstance(payload, dict):
        return {}
    rows = payload.get("validations") if isinstance(payload.get("validations"), list) else []
    success_tokens = {"success", "passed", "full_inference_success", "validated_after_manual_space_test", "recovered_by_manual_validation"}
    for row in reversed(rows):
        if isinstance(row, dict) and str(row.get("status") or row.get("effective_status") or "").lower() in success_tokens:
            return row
    return {}


def _safe_hash_from_record(record: dict[str, Any], value: Any, prefix: str) -> str:
    # Use the run's existing anonymous hash salt context when possible. The
    # worker already hashed run/user/model identifiers; backend archive metadata
    # must not introduce raw linked run IDs or target Space IDs.
    import hashlib
    seed = str(record.get("anonymous_run_id") or record.get("run_id_hash") or "eval")
    raw = f"{prefix}:{seed}:{value}".encode("utf-8", "ignore")
    return f"{prefix}-" + hashlib.sha256(raw).hexdigest()[:16]


def _len_if_list(value: Any) -> int:
    return len(value) if isinstance(value, list) else 0


def _seconds_bucket(value: Any) -> str:
    try:
        seconds = int(value)
    except Exception:
        return "unknown"
    if seconds < 0:
        return "expired"
    if seconds < 15 * 60:
        return "lt_15m"
    if seconds < 60 * 60:
        return "lt_1h"
    if seconds < 4 * 60 * 60:
        return "lt_4h"
    if seconds < 8 * 60 * 60:
        return "lt_8h"
    return "gte_8h"


def _v191_plus_metadata(record: dict[str, Any], paths: RunPaths, *, token: str | None = None) -> dict[str, Any]:
    """Return privacy-safe feature metrics introduced after the v191 line.

    This intentionally stores only aggregate/status information. It never copies
    README/model-card contents, generated requirements/code, Pi evidence text,
    endpoint schemas, prompts, tokens, bucket paths or target Space IDs.
    """
    analysis = read_json(f"{paths.root}/model_analysis.json", token=token) or {}
    if not isinstance(analysis, dict):
        analysis = {}
    requirements_policy = read_json(f"{paths.root}/generated/requirements_policy.json", token=token) or read_json(f"{paths.root}/requirements_policy.json", token=token) or {}
    if not isinstance(requirements_policy, dict):
        requirements_policy = {}
    auth_status = read_json(f"{paths.root}/auth_status.json", token=token) or {}
    if not isinstance(auth_status, dict):
        auth_status = {}
    repair_outcome = read_json(f"{paths.root}/repair_outcome.json", token=token) or read_json(f"{paths.root}/repair/REPAIR_OUTCOME.json", token=token) or {}
    if not isinstance(repair_outcome, dict):
        repair_outcome = {}
    worker_plan_review = read_json(f"{paths.root}/planning/worker_plan_review.json", token=token) or {}
    if not isinstance(worker_plan_review, dict):
        worker_plan_review = {}
    grounding = read_json(f"{paths.root}/planning/model_card_grounding_review.json", token=token) or worker_plan_review.get("model_card_grounding") or {}
    if not isinstance(grounding, dict):
        grounding = {}
    contract = read_json(f"{paths.root}/generated/INFERENCE_CONTRACT.json", token=token) or {}
    if not isinstance(contract, dict):
        contract = {}
    smoke = read_json(f"{paths.root}/tests/generation_smoke.json", token=token) or {}
    if not isinstance(smoke, dict):
        smoke = {}
    build_risk = analysis.get("build_risk") if isinstance(analysis.get("build_risk"), dict) else {}
    kernel_strategy = analysis.get("kernel_strategy") if isinstance(analysis.get("kernel_strategy"), dict) else {}
    if not kernel_strategy and isinstance(analysis.get("metadata"), dict):
        kernel_strategy = analysis["metadata"].get("kernel_strategy") if isinstance(analysis["metadata"].get("kernel_strategy"), dict) else {}

    grounding_source = grounding.get("model_card_source") if isinstance(grounding.get("model_card_source"), dict) else {}
    return {
        "schema_version": "v191_plus_eval_signals.v1",
        "platform_dependency_policy": {
            "present": bool(requirements_policy),
            "status": requirements_policy.get("status") or "",
            "removed_platform_pin_count": _len_if_list(requirements_policy.get("removed_pins")),
            "normalized_platform_line_count": _len_if_list(requirements_policy.get("normalized_platform_lines")),
            "injected_platform_line_count": _len_if_list(requirements_policy.get("injected_platform_lines")),
            "torch_added": bool(requirements_policy.get("torch_added")),
        },
        "auth_context": {
            "present": bool(auth_status),
            "status": auth_status.get("status") or "",
            "token_kind": auth_status.get("token_kind") or "",
            "expiry_known": bool(auth_status.get("expiry_known")),
            "seconds_until_expiry_bucket": _seconds_bucket(auth_status.get("seconds_until_expiry")),
            "safe_for_phase": bool(auth_status.get("safe_for_phase")),
        },
        "model_scan": {
            "build_risk_level": build_risk.get("level") or analysis.get("build_risk_level") or "",
            "build_risk_signal_count": _len_if_list(build_risk.get("signals")),
            "build_risk_visibility_only": bool(build_risk.get("visibility_only", True)) if build_risk else True,
            "recommended_session_minutes": build_risk.get("recommended_session_minutes"),
            "kernel_strategy_present": bool(kernel_strategy),
            "native_kernel_detected": bool(kernel_strategy.get("native_kernel_detected") or kernel_strategy.get("detected")),
            "kernel_signal_count": _len_if_list(kernel_strategy.get("signals")),
            "kernel_candidate_count": _len_if_list(kernel_strategy.get("candidates")),
        },
        "contract_validation": {
            "contract_present": bool(contract),
            "full_inference_implemented": bool(contract.get("full_inference_implemented")),
            "validation_level": contract.get("validation_level") or "",
            "requires_gpu": bool(contract.get("requires_gpu")),
            "blockers_count": int(contract.get("blockers_count") or 0) if str(contract.get("blockers_count") or "0").isdigit() else 0,
            "generation_smoke_status": smoke.get("status") or "",
            "generation_smoke_skipped": str(smoke.get("status") or "").lower() == "skipped",
            "generation_smoke_skip_reason": smoke.get("skip_reason") or "",
        },
        "repair_outcome": {
            "present": bool(repair_outcome),
            "repair_decision": repair_outcome.get("repair_decision") or repair_outcome.get("decision") or "",
            "patch_applied": bool(repair_outcome.get("patch_applied")),
            "upload_success": bool(repair_outcome.get("upload_success")),
            "post_repair_validation": repair_outcome.get("post_repair_validation") or "",
            "failure_type": repair_outcome.get("failure_type") or "",
        },
        "planning": {
            "worker_plan_review_present": bool(worker_plan_review),
            "status": worker_plan_review.get("status") or "",
            "declared_strategy": worker_plan_review.get("declared_strategy") or "",
            "worker_recommendation": worker_plan_review.get("worker_recommendation") or "",
            "warning_count": _len_if_list(worker_plan_review.get("warnings")),
        },
        "model_card_grounding": {
            "present": bool(grounding),
            "status": grounding.get("status") or "",
            "source_available": bool(grounding.get("source_available")),
            "model_card_present": bool(grounding.get("model_card_present")),
            "source": grounding_source.get("source") or "",
            "resolved_card_file": grounding_source.get("resolved_card_file") or "",
            "fallback_used": bool(grounding_source.get("fallback_used")),
            "pi_evidence_present": bool(grounding.get("pi_evidence_present")),
            "pi_evidence_count": int(grounding.get("pi_evidence_count") or 0),
            "warning_count": _len_if_list(grounding.get("warnings")),
            "warnings": [str(w)[:120] for w in (grounding.get("warnings") if isinstance(grounding.get("warnings"), list) else [])[:8]],
        },
    }


def _effective_record_metadata(record: dict[str, Any], paths: RunPaths, *, token: str | None = None, state: dict[str, Any] | None = None) -> tuple[dict[str, Any], dict[str, Any]]:
    """Return an archive-safe eval record enriched with effective outcome data.

    The local worker eval_record.json is intentionally automatic-run centric. The
    backend can see parent linked-validation status and safely add aggregate
    metadata without storing prompts, args, target Space IDs, bucket paths or raw
    generated code in the operator archive.
    """
    out = dict(record)
    outcome = record.get("outcome") if isinstance(record.get("outcome"), dict) else {}
    validation = record.get("validation") if isinstance(record.get("validation"), dict) else {}
    automatic_verdict = str(record.get("verdict") or outcome.get("verdict") or "")
    automatic_status = str(outcome.get("status") or (state or {}).get("status") or record.get("phase") or "")
    manual = read_json(f"{paths.root}/manual_validation_status.json", token=token) or {}
    linked = read_json(f"{paths.root}/linked_validations.json", token=token) or {}
    if not isinstance(manual, dict):
        manual = {}
    if not isinstance(linked, dict):
        linked = {}
    success = manual if str(manual.get("status") or "").lower() == "success" else _first_successful_linked_validation(linked)
    effective_status = str((state or {}).get("effective_status") or outcome.get("effective_status") or "")
    manual_applied = bool(success)
    if manual_applied:
        effective_status = str(success.get("effective_status") or "validated_after_manual_space_test")
    elif not effective_status:
        effective_status = automatic_verdict or automatic_status
    effective_verdict = effective_status or automatic_verdict
    linked_rows = linked.get("validations") if isinstance(linked.get("validations"), list) else []
    linked_success_count = sum(1 for row in linked_rows if isinstance(row, dict) and str(row.get("status") or "").lower() == "success")
    linked_failure_count = sum(1 for row in linked_rows if isinstance(row, dict) and str(row.get("status") or "").lower() in {"failed", "failure", "error"})
    linked_summary: dict[str, Any] = {
        "present": bool(linked_rows or manual_applied),
        "count": len([row for row in linked_rows if isinstance(row, dict)]),
        "success_count": linked_success_count,
        "failure_count": linked_failure_count,
        "latest_success": bool(success),
    }
    if success:
        linked_summary.update({
            "status": "success",
            "mode": success.get("space_test_policy_mode") or success.get("mode") or "complete",
            "api_name": success.get("api_name") or "",
            "latency_seconds": success.get("latency_seconds") or success.get("observed_latency_seconds"),
            "hardware_used_for_validation": success.get("hardware_used_for_validation") or success.get("recommendation_hardware") or "",
            "validation_run_hash": _safe_hash_from_record(record, success.get("validation_run_id") or "linked", "validation"),
        })
    retry = read_json(f"{paths.root}/tests/generation_smoke_payload_retry.json", token=token) or {}
    smoke = read_json(f"{paths.root}/tests/generation_smoke.json", token=token) or {}
    discovery = read_json(f"{paths.root}/tests/gradio_endpoint_discovery.json", token=token) or {}
    if not isinstance(retry, dict):
        retry = {}
    if not isinstance(smoke, dict):
        smoke = {}
    if not isinstance(discovery, dict):
        discovery = {}
    auto_retry = smoke.get("auto_retry") if isinstance(smoke.get("auto_retry"), dict) else {}
    smoke_retry = {
        "retried": bool(retry) or bool(auto_retry.get("retried")),
        "reason": retry.get("retry_reason") or auto_retry.get("retry_reason") or auto_retry.get("reason") or "",
        "attempts": retry.get("attempts") or auto_retry.get("attempts") or (2 if retry else 1),
        "passed_after_retry": bool((smoke.get("status") == "success") and (retry or auto_retry.get("retried"))),
    }
    endpoint_discovery = {
        "required": bool(discovery.get("endpoint_discovery_required") or discovery.get("requested_api_name") in {"", None}),
        "succeeded": bool(discovery.get("selected_api_name") or discovery.get("selected_endpoint")),
        "selected_endpoint": discovery.get("selected_api_name") or discovery.get("selected_endpoint") or "",
        "candidate_count": len(discovery.get("discovered_api_names") or discovery.get("candidates") or []),
        "excluded_health_endpoint": True,
    } if discovery else {"required": False, "succeeded": False, "selected_endpoint": "", "candidate_count": 0, "excluded_health_endpoint": True}
    out["schema_version"] = "1.4"
    out["automatic_outcome"] = {
        "status": automatic_status,
        "verdict": automatic_verdict,
        "health_passed": bool(outcome.get("health_passed") or validation.get("health_passed")),
        "generation_smoke_passed": bool(outcome.get("generation_smoke_passed") or validation.get("generation_smoke_passed")),
        "full_inference_verified": bool(outcome.get("full_inference_verified") or validation.get("full_inference_verified")),
    }
    out["effective_outcome"] = {
        "automatic_verdict": automatic_verdict,
        "automatic_status": automatic_status,
        "effective_verdict": effective_verdict,
        "effective_status": effective_status,
        "manual_validation_applied": manual_applied,
        "source": "linked_space_test" if manual_applied else "automatic_run",
    }
    out["linked_validation"] = linked_summary
    out["endpoint_discovery"] = endpoint_discovery
    out["smoke_retry"] = smoke_retry
    out["v191_plus"] = _v191_plus_metadata(record, paths, token=token)
    out.setdefault("privacy", {})
    if isinstance(out["privacy"], dict):
        out["privacy"].update({
            "linked_run_ids_redacted": True,
            "endpoint_schema_stored": False,
            "validation_args_stored": False,
            "raw_prompts_stored": False,
            "generated_code_stored": False,
            "tokens_stored": False,
            "model_card_raw_stored": False,
            "pi_evidence_text_stored": False,
            "requirements_txt_stored": False,
            "auth_token_stored": False,
        })
    publish_hint = {
        "effective_status": effective_status,
        "effective_verdict": effective_verdict,
        "manual_validation_applied": manual_applied,
        "linked_validation_count": linked_summary["count"],
    }
    return out, publish_hint


def _should_attempt_publish(record: dict[str, Any], bundle_state: dict[str, Any] | None = None) -> bool:
    status = str(record.get("outcome", {}).get("status") or record.get("phase") or "").lower()
    state_status = str((bundle_state or {}).get("status") or "").lower()
    phase = str(record.get("phase") or "").lower()
    return status in _TERMINAL or state_status in _TERMINAL or phase in {"final", "failure"}


def maybe_publish_eval_record(
    run_id: str,
    *,
    bucket_source: str,
    token: str | None = None,
    state: dict[str, Any] | None = None,
    force: bool = False,
) -> dict[str, Any]:
    """Publish a run-local anonymized eval record to the operator archive.

    User Jobs write only ``runs/<run_id>/eval_record.json`` in the user's bucket.
    The ASF Space backend owns the private operator eval bucket mounted at
    ``/evals`` and copies the already-anonymized record there. This lets evals
    work for all users without giving their Jobs write access to the private
    operator bucket.
    """
    cfg = effective_eval_config()
    paths = RunPaths(run_id, bucket_source=bucket_source)
    if not cfg.get("enabled"):
        return _publish_result(paths, {"enabled": False, "published": False, "attempted": False, "reason": "eval_archive_disabled"}, token=token)
    mount = _eval_mount(cfg)
    if mount is None:
        return _publish_result(paths, {"enabled": False, "published": False, "attempted": False, "reason": "eval_archive_disabled"}, token=token)
    if not mount.exists():
        return _publish_result(paths, {"enabled": True, "published": False, "attempted": True, "reason": "eval_mount_missing", "mount": str(mount)}, token=token)
    if not mount.is_dir():
        return _publish_result(paths, {"enabled": True, "published": False, "attempted": True, "reason": "eval_mount_not_directory", "mount": str(mount)}, token=token)

    status_path = _publish_status_path(paths)
    prev = read_json(status_path, token=token) or {}

    record = read_json(f"{paths.root}/eval_record.json", token=token)
    if not record:
        return _publish_result(paths, {"enabled": True, "published": False, "attempted": True, "reason": "record_not_ready", "local_record_found": False, "mount": str(mount)}, token=token)
    if not isinstance(record, dict):
        return _publish_result(paths, {"enabled": True, "published": False, "attempted": True, "reason": "invalid_record", "local_record_found": True, "mount": str(mount)}, token=token)
    if not force and not _should_attempt_publish(record, state):
        return _publish_result(paths, {"enabled": True, "published": False, "attempted": True, "reason": "run_not_terminal", "local_record_found": True, "mount": str(mount)}, token=token)

    privacy = record.get("privacy") if isinstance(record.get("privacy"), dict) else {}
    if privacy.get("generated_code_stored") or privacy.get("raw_prompts_stored") or privacy.get("tokens_stored"):
        return _publish_result(paths, {"enabled": True, "published": False, "attempted": True, "reason": "privacy_flags_rejected", "local_record_found": True, "mount": str(mount)}, token=token)

    record, publish_hint = _effective_record_metadata(record, paths, token=token, state=state)
    if not force and isinstance(prev, dict) and prev.get("published") and str(prev.get("schema_version") or "").startswith("1.4"):
        if prev.get("effective_status") == publish_hint.get("effective_status") and prev.get("linked_validation_count") == publish_hint.get("linked_validation_count"):
            return {**prev, "enabled": True, "skipped": True, "reason": "already_published"}

    finished = str(record.get("finished_at") or record.get("started_at") or _now())
    day = finished[:10] if len(finished) >= 10 else _now()[:10]
    try:
        yyyy, mm, dd = day.split("-")
    except Exception:
        yyyy, mm, dd = _now()[:10].split("-")
    anon_run_id = _safe_segment(record.get("anonymous_run_id") or record.get("run_id_hash"), "unknown-run")
    dest_dir = mount / yyyy / mm / dd / anon_run_id
    dest_dir.mkdir(parents=True, exist_ok=True)
    dest = dest_dir / "eval_record.json"
    dest.write_text(json.dumps(record, indent=2, ensure_ascii=False, sort_keys=True) + "\n", encoding="utf-8")

    compact = read_text(f"{paths.root}/events_compact.jsonl", token=token)
    if not compact:
        timeline = record.get("timeline") if isinstance(record.get("timeline"), list) else []
        compact = "".join(json.dumps(item, ensure_ascii=False) + "\n" for item in timeline)
    if compact:
        (dest_dir / "events_compact.jsonl").write_text(compact, encoding="utf-8")

    report = {
        "schema_version": "1.4",
        "enabled": True,
        "attempted": True,
        "published": True,
        "published_at": _now(),
        "publish_mode": "backend",
        "archive_path": str(dest),
        "archive_relative_path": str(dest.relative_to(mount)),
        "eval_bucket_source": cfg.get("bucket_source") or "",
        "eval_bucket_path": cfg.get("bucket_path") or "evals",
        "mount": str(mount),
        "local_record_found": True,
        "post_write_file_exists": dest.exists(),
        "anonymous_run_id": anon_run_id,
        **publish_hint,
    }
    return _publish_result(paths, report, token=token)