from __future__ import annotations from datetime import datetime, timedelta, timezone from typing import Any from .progress import STEP_ALIASES, STEP_LABELS, STEP_ORDER from .timeline_model import build_run_timeline_model from .config import settings PRODUCT_STEPS = [{"id": step, "label": STEP_LABELS[step]} for step in STEP_ORDER] def _runs_prefix() -> str: return settings.bucket_runs_prefix.strip().strip("/") or "runs" TERMINAL_GLOBAL_STATUSES = {"succeeded", "partial", "failed", "cancelled", "blocked", "waiting_manual_action"} SUCCESS_RAW_STATUSES = { "success", "done", "completed", "passed", "full_inference_success", "repair_success", } PARTIAL_RAW_STATUSES = { "full_inference_candidate_health_passed", "health_only", "partial", "partial_validation", "completed_with_warnings", } MANUAL_RAW_STATUSES = { "manual_hardware_required", "generated_needs_manual_hardware", "waiting_manual_hardware", "manual_action_required", } FAILED_RAW_STATUSES = {"failed", "failure", "error", "repair_failed"} CANCELLED_RAW_STATUSES = {"cancelled", "canceled"} BLOCKED_RAW_STATUSES = {"technical_blocker", "blocked", "health_only"} RUNNING_RAW_STATUSES = {"started", "running", "waiting", "queued", "pending", "scheduled"} STEP_TO_PHASE = {step: step for step in STEP_ORDER} | {alias: canonical for alias, canonical in STEP_ALIASES.items()} def parse_ts(value: Any) -> datetime | None: if not value: return None if isinstance(value, datetime): return value if value.tzinfo else value.replace(tzinfo=timezone.utc) try: parsed = datetime.fromisoformat(str(value).replace("Z", "+00:00")) return parsed if parsed.tzinfo else parsed.replace(tzinfo=timezone.utc) except Exception: return None def _lower(value: Any) -> str: return str(value or "").strip().lower() def _first_nonempty(*values: Any) -> str: for value in values: if value is not None and str(value).strip(): return str(value) return "" def _owner_from_bucket_source(bucket_source: str | None) -> str: if not bucket_source: return "" return str(bucket_source).split("/", 1)[0].strip() def _job_url_from_view_sources(*, bucket_source: str, summary: dict[str, Any], state: dict[str, Any], launch: dict[str, Any]) -> str: explicit = summary.get("job_url") or state.get("job_url") or launch.get("job_url") if explicit: return str(explicit) job_id = summary.get("job_id") or state.get("job_id") or launch.get("job_id") owner = ( summary.get("created_by") or summary.get("username") or summary.get("owner") or state.get("created_by") or state.get("username") or state.get("owner") or launch.get("created_by") or launch.get("username") or launch.get("owner") or _owner_from_bucket_source(bucket_source) ) if job_id and owner: return f"https://huggingface.co/jobs/{owner}/{job_id}" return "" def _latest_activity_at(bundle: dict[str, Any], summary: dict[str, Any], state: dict[str, Any], launch: dict[str, Any]) -> datetime | None: candidates: list[datetime] = [] for event in bundle.get("events") or []: ts = parse_ts(event.get("ts") if isinstance(event, dict) else None) if ts: candidates.append(ts) for source in (summary, state, launch, bundle.get("summary_file") or {}): for key in ("updated_at", "finished_at", "created_at", "started_at"): ts = parse_ts(source.get(key) if isinstance(source, dict) else None) if ts: candidates.append(ts) return max(candidates) if candidates else None def _created_at(summary: dict[str, Any], state: dict[str, Any], launch: dict[str, Any]) -> datetime | None: for source in (summary, state, launch): ts = parse_ts(source.get("created_at") if isinstance(source, dict) else None) if ts: return ts return None def _raw_statuses(bundle: dict[str, Any]) -> set[str]: summary = bundle.get("summary") or {} state = bundle.get("state") or {} launch = bundle.get("launch") or {} gate = bundle.get("inference_gate") or {} smoke = bundle.get("generation_smoke") or {} hardware = bundle.get("hardware_strategy") or {} blockers = bundle.get("technical_blockers") or {} values = { _lower(summary.get("status")), _lower(state.get("status")), _lower(state.get("gate_status")), _lower(launch.get("status")), _lower(gate.get("status")), _lower(smoke.get("status")), _lower(hardware.get("status")), _lower(blockers.get("status")), } # Event-level statuses describe individual steps and must not by themselves # turn a whole run into success/failure. The product phase is derived from # events separately in derive_product_phase(). return {v for v in values if v} def requires_manual_action(bundle: dict[str, Any]) -> bool: summary = bundle.get("summary") or {} gate = bundle.get("inference_gate") or {} hardware = bundle.get("hardware_strategy") or {} statuses = _raw_statuses(bundle) return bool( summary.get("manual_hardware_required") or gate.get("manual_hardware_required") or hardware.get("manual_action_required") or statuses.intersection(MANUAL_RAW_STATUSES) ) def has_technical_blocker(bundle: dict[str, Any]) -> bool: blockers = bundle.get("technical_blockers") or {} blocker_items = blockers.get("blockers") if isinstance(blockers, dict) else None return bool(blocker_items or _raw_statuses(bundle).intersection(BLOCKED_RAW_STATUSES)) def normalize_run_status( bundle: dict[str, Any], *, now: datetime | None = None, stale_after: timedelta = timedelta(hours=6), ) -> dict[str, Any]: """Return canonical status flags for the product UI.""" now = now or datetime.now(timezone.utc) summary = bundle.get("summary") or {} state = bundle.get("state") or {} launch = bundle.get("launch") or {} statuses = _raw_statuses(bundle) manual = requires_manual_action(bundle) blocker = has_technical_blocker(bundle) if statuses.intersection(CANCELLED_RAW_STATUSES): global_status = "cancelled" verdict = "cancelled" elif manual: global_status = "waiting_manual_action" verdict = "manual_action_required" elif statuses.intersection(SUCCESS_RAW_STATUSES): global_status = "succeeded" verdict = "passed" elif statuses.intersection(PARTIAL_RAW_STATUSES): global_status = "partial" verdict = "partial_validation" elif statuses.intersection(FAILED_RAW_STATUSES): global_status = "failed" verdict = "failed" elif blocker: global_status = "blocked" verdict = "technical_blocker" elif statuses.intersection(RUNNING_RAW_STATUSES) or bundle.get("events") or launch: global_status = "running" verdict = "pending" else: global_status = "unknown" verdict = "unknown" latest = _latest_activity_at(bundle, summary, state, launch) created = _created_at(summary, state, launch) reference = latest or created is_terminal = global_status in TERMINAL_GLOBAL_STATUSES or global_status in {"succeeded"} is_stale = False if reference and not is_terminal and now - reference > stale_after: global_status = "stale" verdict = "stale" is_stale = True return { "global_status": global_status, "raw_status": _first_nonempty(summary.get("status"), state.get("status"), launch.get("status"), "unknown"), "verdict": verdict, "requires_manual_action": manual, "manual_action_type": "hardware" if manual else "", "has_technical_blocker": blocker, "has_target_space": bool(summary.get("target_space") or state.get("target_space") or launch.get("target_space")), "has_job_url": bool(summary.get("job_url") or state.get("job_url") or launch.get("job_url") or summary.get("job_id") or state.get("job_id") or launch.get("job_id")), "has_live_api_result": bool((bundle.get("generation_smoke") or {}).get("ok") or (bundle.get("generation_smoke") or {}).get("status") == "success"), "is_terminal": global_status in {"succeeded", "partial", "failed", "blocked", "waiting_manual_action", "cancelled"}, "is_pollable": global_status in {"queued", "running", "validating", "unknown"}, "is_stale": is_stale, "latest_activity_at": latest.isoformat() if latest else "", } def derive_product_phase(bundle: dict[str, Any], status_model: dict[str, Any]) -> str: if status_model["global_status"] in {"succeeded", "partial"}: return "done" if status_model["global_status"] in {"failed", "blocked", "stale", "cancelled"}: return "failure" if status_model["requires_manual_action"]: return "inference_gate" events = [e for e in (bundle.get("events") or []) if isinstance(e, dict)] for event in reversed(events): step = _lower(event.get("step")) message = _lower(event.get("message")) status = _lower(event.get("status")) if "patch" in step or "repair" in step or "patched" in message or "restart" in message: return "hardware_strategy" if "missing" in message or "error" in message or status in FAILED_RAW_STATUSES: return "hardware_strategy" phase = STEP_TO_PHASE.get(step) if phase: return phase summary = bundle.get("summary") or {} if summary.get("target_space"): return "hardware_strategy" if bundle.get("launch"): return "bootstrap" return "bootstrap" def build_pipeline(phase: str, status_model: dict[str, Any]) -> list[dict[str, Any]]: phase_ids = [s["id"] for s in PRODUCT_STEPS] current_index = phase_ids.index(phase) if phase in phase_ids else 0 global_status = status_model["global_status"] pipeline: list[dict[str, Any]] = [] for index, step in enumerate(PRODUCT_STEPS): if global_status == "succeeded": step_status = "completed" elif index < current_index: step_status = "completed" elif index == current_index: if global_status == "failed": step_status = "failed" elif global_status in {"blocked", "waiting_manual_action", "stale"}: step_status = "blocked" else: step_status = "running" else: step_status = "pending" pipeline.append({**step, "status": step_status}) return pipeline def _agent_for_event(event: dict[str, Any]) -> str: step = _lower(event.get("step")) if step in {"model_analysis", "bucket_ready", "job_launched", "bootstrap"}: return "Planner Agent" if step in {"workspace", "node", "pi_install", "pi_config", "pi_run", "repair", "patch"}: return "Coder Agent" if step in {"create_space", "upload_files", "hardware", "hardware_preferred", "hardware_fallback"}: return "Hub Agent" if step in {"api_validation", "generation_smoke", "inference_gate"}: return "Tester Agent" if _lower(event.get("status")) in FAILED_RAW_STATUSES: return "Diagnostician Agent" return "Factory Agent" def _severity_for_event(event: dict[str, Any]) -> str: status = _lower(event.get("status")) message = _lower(event.get("message")) if status in FAILED_RAW_STATUSES or "traceback" in message or "modulenotfound" in message: return "error" if status in {"warning", "manual_hardware_required"} or "manual" in message: return "warning" if status in SUCCESS_RAW_STATUSES: return "success" if status in PARTIAL_RAW_STATUSES: return "warning" if status in RUNNING_RAW_STATUSES: return "running" return "info" def build_activity_feed(bundle: dict[str, Any], *, limit: int = 40) -> list[dict[str, Any]]: events = [e for e in (bundle.get("events") or []) if isinstance(e, dict)] if not events and bundle.get("launch"): launch = bundle.get("launch") or {} events = [ { "ts": launch.get("created_at") or "", "step": "job_launched", "status": launch.get("status") or "running", "message": f"Build job launched for {launch.get('target_space') or 'target Space'}", } ] feed: list[dict[str, Any]] = [] for event in events[-limit:]: feed.append( { "ts": event.get("ts") or event.get("created_at") or "", "step": event.get("step") or "", "status": event.get("status") or "", "message": event.get("message") or event.get("step") or "Event received", "severity": _severity_for_event(event), "agent": _agent_for_event(event), } ) return feed def build_diagnostics(bundle: dict[str, Any], status_model: dict[str, Any], phase: str) -> dict[str, Any]: summary = bundle.get("summary") or {} smoke = bundle.get("generation_smoke") or {} gate = bundle.get("inference_gate") or {} hardware = bundle.get("hardware_strategy") or {} blockers = bundle.get("technical_blockers") or {} blocker_items = blockers.get("blockers") if isinstance(blockers, dict) else [] issue_title = "" issue_detail = "" issue_status = "" if blocker_items: first = blocker_items[0] issue_title = _first_nonempty(first.get("type"), first.get("name"), "Technical blocker") if isinstance(first, dict) else "Technical blocker" issue_detail = _first_nonempty(first.get("claim"), first.get("reason"), first.get("message")) if isinstance(first, dict) else str(first) issue_status = "open" elif status_model["requires_manual_action"]: issue_title = "Manual hardware required" issue_detail = "Automatic hardware selection was not available. Choose hardware in Space settings, then run Space Test." issue_status = "action_required" elif status_model["global_status"] == "failed": issue_title = "Run failed" issue_detail = "Inspect logs and run artifacts for the failing step." issue_status = "open" health_passed = bool((gate.get("implementation_signals") or {}).get("health_passed") or smoke.get("health_passed")) smoke_ok = bool(smoke.get("ok") or smoke.get("status") == "success") build_ok = phase in {"api_validation", "live_wait", "generation_smoke", "inference_gate", "report_write", "done"} or status_model["global_status"] == "succeeded" return { "build_status": "passed" if build_ok else ("blocked" if status_model["global_status"] in {"blocked", "waiting_manual_action", "failed"} else "building"), "api_status": "passed" if smoke_ok else ("blocked" if status_model["global_status"] in {"blocked", "failed"} else "pending"), "tests_status": "passed" if smoke_ok or health_passed else ("blocked" if status_model["global_status"] in {"blocked", "failed"} else "pending"), "verdict_status": status_model["verdict"], "zerogpu_rules": [ {"label": "Gradio interface", "status": "ok"}, {"label": "Private Space", "status": "ok" if summary.get("target_space") else "pending"}, {"label": "ZeroGPU-first strategy", "status": "ok" if "zero" in _lower(summary.get("selected_hardware") or hardware.get("preferred_space_hardware") or "zero") else "pending"}, {"label": "Live API verification", "status": "ok" if smoke_ok else "pending"}, ], "issue": {"title": issue_title, "detail": issue_detail, "status": issue_status}, } def build_space_test_model(bundle: dict[str, Any], status_model: dict[str, Any]) -> dict[str, Any]: smoke = bundle.get("generation_smoke") or {} summary = bundle.get("summary") or {} return { "target_space": summary.get("target_space") or "", "target_space_url": summary.get("target_space_url") or "", "endpoint": smoke.get("api_name") or smoke.get("endpoint") or "/generate", "status": "passed" if smoke.get("ok") or smoke.get("status") == "success" else "pending", "latency_seconds": smoke.get("latency_seconds") or summary.get("latency_seconds"), "expected_output_type": summary.get("expected_output_type") or smoke.get("expected_output_type") or "", "verdict": status_model["verdict"], "output_artifact": smoke.get("output_artifact") or smoke.get("artifact_url") or "", } def build_run_view_model( run_id: str, bundle: dict[str, Any], *, bucket_source: str, now: datetime | None = None, ) -> dict[str, Any]: summary = bundle.get("summary") or {} state = bundle.get("state") or {} status_model = normalize_run_status(bundle, now=now) phase = derive_product_phase(bundle, status_model) elapsed_seconds = None current_now = now or datetime.now(timezone.utc) created = parse_ts(summary.get("started_at") or state.get("started_at") or summary.get("created_at") or state.get("created_at")) latest = parse_ts(status_model.get("latest_activity_at")) if created: end = latest if status_model.get("is_terminal") and latest else current_now elapsed_seconds = int((end - created).total_seconds()) links = { "job_url": _job_url_from_view_sources(bucket_source=bucket_source, summary=summary, state=state, launch=bundle.get("launch") or {}), "target_space_url": summary.get("target_space_url") or "", "target_space_settings_url": f"{summary.get('target_space_url')}/settings" if summary.get("target_space_url") else "", "artifacts_url": summary.get("artifacts_url") or f"https://huggingface.co/buckets/{bucket_source}/tree/{_runs_prefix()}/{run_id}", } return { "schema_version": "run_view_model.v1", "run_id": run_id, "bucket_source": bucket_source, "header": { "title": _first_nonempty(summary.get("title"), f"Build Space for {summary.get('model_id')}", run_id), "status": status_model["global_status"], "status_label": status_model["global_status"].replace("_", " ").title(), "raw_status": status_model["raw_status"], "space": summary.get("target_space") or "", "space_url": summary.get("target_space_url") or "", "current_phase": phase, "current_phase_label": next((s["label"] for s in PRODUCT_STEPS if s["id"] == phase), phase), "elapsed_seconds": max(0, elapsed_seconds or 0), "started_at": summary.get("created_at") or state.get("created_at") or "", "updated_at": status_model["latest_activity_at"], }, "status_model": status_model, "pipeline": build_pipeline(phase, status_model), "timeline_model": build_run_timeline_model(bundle), "activity": build_activity_feed(bundle), "diagnostics": build_diagnostics(bundle, status_model, phase), "space_test": build_space_test_model(bundle, status_model), "actions": { "can_resume": status_model["global_status"] in {"running", "stale", "unknown"}, "can_stop": status_model["global_status"] in {"running", "queued", "unknown"}, "can_open_space": bool(summary.get("target_space_url")), "can_validate": bool(summary.get("target_space")), "requires_manual_action": status_model["requires_manual_action"], }, "links": links, } def is_resumable_summary(summary: dict[str, Any], *, now: datetime | None = None, stale_after: timedelta = timedelta(hours=6)) -> bool: """Best-effort resumability test for lightweight /api/runs summaries.""" status = _lower(summary.get("status")) if status in SUCCESS_RAW_STATUSES or status in PARTIAL_RAW_STATUSES or status in FAILED_RAW_STATUSES or status in BLOCKED_RAW_STATUSES or status in MANUAL_RAW_STATUSES: return False if "success" in status or "failed" in status or "blocker" in status or "manual" in status: return False now = now or datetime.now(timezone.utc) latest = parse_ts(summary.get("updated_at") or summary.get("created_at")) if latest and now - latest > stale_after: return False return status in RUNNING_RAW_STATUSES or status in {"unknown", ""} def find_latest_resumable_run(summaries: list[dict[str, Any]], *, now: datetime | None = None) -> dict[str, Any] | None: candidates = [s for s in summaries if isinstance(s, dict) and is_resumable_summary(s, now=now)] if not candidates: return None return sorted(candidates, key=lambda s: str(s.get("updated_at") or s.get("created_at") or s.get("run_id")), reverse=True)[0]