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", "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.3", "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 _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.3" 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.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, }) 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//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.3"): 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.3", "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)