| from __future__ import annotations |
|
|
| from datetime import datetime, timedelta, timezone |
| import re |
| 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 |
| from .effective_status import compute_effective_run_status |
|
|
| 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", "auth_refresh_required"} |
| 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", |
| "demo_usable_full_promise_not_verified", |
| "interactive_app_available_smoke_failed", |
| "manual_test_required_smoke_failed", |
| } |
| 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", "technical_blocker_boot_only", "blocked"} |
| RUNNING_RAW_STATUSES = {"started", "running", "waiting", "queued", "pending", "scheduled"} |
| AUTH_RAW_STATUSES = {"auth_refresh_required", "oauth_expired", "auth_expired", "repair_validation_inconclusive_auth", "inconclusive_auth_expired"} |
|
|
| AUTHORITATIVE_TERMINAL_SOURCE_KEYS = ("status", "final_status", "effective_status", "effective_verdict", "display_status", "verdict", "result_status", "validation_status") |
| TERMINAL_DISPLAY_STATUSES = SUCCESS_RAW_STATUSES | PARTIAL_RAW_STATUSES | MANUAL_RAW_STATUSES | FAILED_RAW_STATUSES | CANCELLED_RAW_STATUSES | BLOCKED_RAW_STATUSES | AUTH_RAW_STATUSES | {"stale", "stopped", "succeeded", "validated_after_space_test", "validated_after_manual_space_test", "recovered_by_space_test", "recovered_by_manual_validation"} |
|
|
| 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")), |
| _lower((bundle.get("repair_outcome") or {}).get("post_repair_validation")), |
| _lower((bundle.get("repair_outcome") or {}).get("failure_type")), |
| } |
| |
| |
| |
| return {v for v in values if v} |
|
|
|
|
| def _normalize_terminal_status(value: Any) -> str: |
| status = _lower(value) |
| if not status: |
| return "" |
| if status in {"succeeded", "success", "done", "completed", "passed", "repair_success"}: |
| return "full_inference_success" |
| if status in {"cancelled", "canceled"}: |
| return "stopped" |
| if status in {"partial", "health_only", "full_inference_candidate_health_passed", "completed_with_warnings"}: |
| return "partial_validation" |
| if status in TERMINAL_DISPLAY_STATUSES: |
| return status |
| return "" |
|
|
|
|
| def _first_authoritative_terminal_status(bundle: dict[str, Any]) -> tuple[str, str]: |
| """Return strongest terminal display status and source for Active Run. |
| |
| v198.26.7 invariant: final reconciliation and summary are terminal sources |
| of truth. They must not be downgraded by stale state/live/progress artifacts |
| left behind by a repaired run. |
| """ |
| ordered_sources: list[tuple[str, dict[str, Any]]] = [ |
| ("final_status_reconciliation", bundle.get("final_status_reconciliation") or {}), |
| ("summary", bundle.get("summary") or {}), |
| ("summary_file", bundle.get("summary_file") or {}), |
| ("effective_run_status", bundle.get("effective_run_status") or {}), |
| ] |
| summary = bundle.get("summary") or {} |
| if isinstance(summary.get("effective_run_status"), dict): |
| ordered_sources.append(("summary.effective_run_status", summary.get("effective_run_status") or {})) |
| |
| |
| ordered_sources.extend([ |
| ("inference_gate", bundle.get("inference_gate") or {}), |
| ("generation_smoke", bundle.get("generation_smoke") or {}), |
| ("live_status", bundle.get("live_status") or {}), |
| ]) |
| blocking_seen = False |
| for source_name, source in ordered_sources: |
| if not isinstance(source, dict): |
| continue |
| for key in AUTHORITATIVE_TERMINAL_SOURCE_KEYS: |
| normalized = _normalize_terminal_status(source.get(key)) |
| if normalized: |
| if normalized in BLOCKED_RAW_STATUSES | FAILED_RAW_STATUSES | MANUAL_RAW_STATUSES | AUTH_RAW_STATUSES | CANCELLED_RAW_STATUSES | {"stopped", "stale"}: |
| return normalized, f"{source_name}.{key}" |
| if normalized == "full_inference_success" and blocking_seen: |
| continue |
| return normalized, f"{source_name}.{key}" |
| raw_text = " ".join(_lower(source.get(k)) for k in AUTHORITATIVE_TERMINAL_SOURCE_KEYS if isinstance(source.get(k), (str, int, float, bool))) |
| if any(token in raw_text for token in ("technical_blocker", "technical blocker", "blocked", "failed", "manual_hardware_required", "auth_refresh_required", "stopped")): |
| blocking_seen = True |
| |
| |
| if not blocking_seen and (source.get("ok") is True or source.get("generation_smoke_passed") is True): |
| return "full_inference_success", f"{source_name}.generation_smoke_passed" |
| signals = source.get("implementation_signals") if isinstance(source.get("implementation_signals"), dict) else {} |
| if not blocking_seen and signals.get("generation_smoke_passed") is True: |
| return "full_inference_success", f"{source_name}.implementation_signals.generation_smoke_passed" |
| return "", "" |
|
|
|
|
| def _status_model_from_terminal_status(status: str, *, source: str = "") -> dict[str, str | bool]: |
| normalized = _normalize_terminal_status(status) or "unknown" |
| if normalized == "full_inference_success": |
| global_status, verdict = "succeeded", "full_inference_success" |
| elif normalized in PARTIAL_RAW_STATUSES or normalized == "partial_validation": |
| global_status, verdict = "partial", normalized if normalized not in {"partial", "health_only", "full_inference_candidate_health_passed", "completed_with_warnings"} else "partial_validation" |
| elif normalized in MANUAL_RAW_STATUSES: |
| global_status, verdict = "waiting_manual_action", "manual_action_required" |
| elif normalized in BLOCKED_RAW_STATUSES: |
| global_status, verdict = "blocked", "technical_blocker_boot_only" if normalized == "technical_blocker_boot_only" else "technical_blocker" |
| elif normalized in FAILED_RAW_STATUSES: |
| global_status, verdict = "failed", "failed" |
| elif normalized in AUTH_RAW_STATUSES: |
| global_status, verdict = "auth_refresh_required", "auth_refresh_required" |
| elif normalized in CANCELLED_RAW_STATUSES or normalized == "stopped": |
| global_status, verdict = "cancelled", "cancelled" |
| elif normalized == "stale": |
| global_status, verdict = "stale", "stale" |
| else: |
| global_status, verdict = normalized, normalized |
| return {"global_status": global_status, "verdict": verdict, "terminal_source": source} |
|
|
|
|
| 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) |
| authoritative_status, terminal_source = _first_authoritative_terminal_status(bundle) |
|
|
| if authoritative_status: |
| terminal_model = _status_model_from_terminal_status(authoritative_status, source=terminal_source) |
| global_status = str(terminal_model["global_status"]) |
| verdict = str(terminal_model["verdict"]) |
| elif statuses.intersection(AUTH_RAW_STATUSES): |
| global_status = "auth_refresh_required" |
| verdict = "auth_refresh_required" |
| elif statuses.intersection(CANCELLED_RAW_STATUSES): |
| global_status = "cancelled" |
| verdict = "cancelled" |
| elif manual: |
| global_status = "waiting_manual_action" |
| verdict = "manual_action_required" |
| elif blocker: |
| global_status = "blocked" |
| verdict = "technical_blocker_boot_only" if "technical_blocker_boot_only" in statuses else "technical_blocker" |
| elif statuses.intersection(FAILED_RAW_STATUSES): |
| global_status = "failed" |
| verdict = "failed" |
| elif statuses.intersection(PARTIAL_RAW_STATUSES): |
| global_status = "partial" |
| verdict = "partial_validation" |
| elif statuses.intersection(SUCCESS_RAW_STATUSES): |
| global_status = "succeeded" |
| verdict = "passed" |
| 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", "auth_refresh_required"}, |
| "is_pollable": global_status in {"queued", "running", "validating", "unknown"}, |
| "is_stale": is_stale, |
| "latest_activity_at": latest.isoformat() if latest else "", |
| "terminal_source": terminal_source if 'terminal_source' in locals() 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": |
| link_state = build_space_link_state(bundle) |
| if link_state.get("links_ready"): |
| issue_title = "Validation failed after Space upload" |
| issue_detail = "The generated Space exists and remains available, but automatic validation or recovery ended in failure." |
| else: |
| 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 _normalize_target_space_id(value: Any) -> str: |
| text = str(value or "").strip() |
| if not text: |
| return "" |
| text = text.replace("https://huggingface.co/spaces/", "").strip("/") |
| text = text.split("/settings", 1)[0].strip("/") |
| if ".hf.space" in text and "/" not in text: |
| return "" |
| return text if re.match(r"^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$", text) else "" |
|
|
|
|
| def _space_target_from_events(events: list[dict[str, Any]] | None, *, require_success: bool = False) -> str: |
| for event in reversed(events or []): |
| if not isinstance(event, dict): |
| continue |
| step = str(event.get("step") or "") |
| status = str(event.get("status") or "").lower() |
| if step not in {"create_space", "create_space_hardware", "upload_files", "runtime_upload_epoch"}: |
| continue |
| if require_success and status != "success": |
| continue |
| data = event.get("data") if isinstance(event.get("data"), dict) else {} |
| target = _normalize_target_space_id(data.get("target_space") or data.get("target_space_id")) |
| if target: |
| return target |
| return "" |
|
|
|
|
| def _space_target_from_bundle(bundle: dict[str, Any]) -> str: |
| summary = bundle.get("summary") or {} |
| state = bundle.get("state") or {} |
| launch = bundle.get("launch") or {} |
| runtime_upload_epoch = bundle.get("runtime_upload_epoch") or {} |
| identity = bundle.get("space_identity") or {} |
| return _normalize_target_space_id(_first_nonempty( |
| identity.get("target_space"), |
| summary.get("target_space"), |
| state.get("target_space"), |
| launch.get("target_space"), |
| summary.get("target_space_id"), |
| state.get("target_space_id"), |
| launch.get("target_space_id"), |
| runtime_upload_epoch.get("target_space_id"), |
| _space_target_from_events(bundle.get("events") or []), |
| )) |
|
|
|
|
| def _space_target_url_from_bundle(bundle: dict[str, Any], target: str = "") -> str: |
| summary = bundle.get("summary") or {} |
| state = bundle.get("state") or {} |
| launch = bundle.get("launch") or {} |
| identity = bundle.get("space_identity") or {} |
| explicit = _first_nonempty(identity.get("target_space_url"), summary.get("target_space_url"), state.get("target_space_url"), launch.get("target_space_url")) |
| if explicit: |
| return explicit |
| target = _normalize_target_space_id(target) |
| return f"https://huggingface.co/spaces/{target}" if target else "" |
|
|
|
|
| def build_space_link_state(bundle: dict[str, Any], *, target: str = "", target_url: str = "") -> dict[str, Any]: |
| """Return durable Space link readiness for Active Run. |
| |
| v198.26.9 invariant: Space links are a persisted run fact. They must not |
| depend on the frontend receiving the create/upload event inside a truncated |
| recent-events window. If a runtime upload epoch or Space runtime probe is |
| present, a terminal failed run still gets Open Space / Settings links. |
| """ |
| identity = bundle.get("space_identity") or {} |
| runtime_upload_epoch = bundle.get("runtime_upload_epoch") or {} |
| runtime_history = runtime_upload_epoch.get("history") if isinstance(runtime_upload_epoch.get("history"), list) else [] |
| space_runtime = bundle.get("space_runtime") or {} |
| target = _space_target_from_bundle(bundle) if not target else _normalize_target_space_id(target) |
| target_url = target_url or _space_target_url_from_bundle(bundle, target) |
| create_event_ok = False |
| upload_event_ok = False |
| for event in bundle.get("events") or []: |
| if not isinstance(event, dict): |
| continue |
| step = str(event.get("step") or "") |
| status = str(event.get("status") or "").lower() |
| if status == "success" and step in {"create_space", "create_space_hardware"}: |
| create_event_ok = True |
| if status == "success" and step in {"upload_files", "runtime_upload_epoch"}: |
| upload_event_ok = True |
| runtime_uploaded = bool(identity.get("runtime_uploaded") or identity.get("space_uploaded") or runtime_upload_epoch.get("last_upload_completed_at") or runtime_upload_epoch.get("upload_sequence") or runtime_history or upload_event_ok) |
| space_runtime_known = bool(identity.get("space_runtime_known") or (isinstance(space_runtime, dict) and (space_runtime.get("stage") or space_runtime.get("status") or space_runtime.get("runtime_status") or space_runtime.get("updated_at") or space_runtime.get("url")))) |
| space_created = bool(identity.get("space_created") or (target and (create_event_ok or runtime_uploaded or space_runtime_known))) |
| links_ready = bool(target_url and (identity.get("links_ready") or space_created or runtime_uploaded or space_runtime_known)) |
| sources: list[str] = [] |
| if identity.get("source"): |
| sources.append(str(identity.get("source"))) |
| if create_event_ok: |
| sources.append("create_space_event") |
| if upload_event_ok: |
| sources.append("upload_event") |
| if runtime_uploaded: |
| sources.append("runtime_upload_epoch") |
| if space_runtime_known: |
| sources.append("space_runtime") |
| return { |
| "schema_version": "space_link_state.v198_26_9", |
| "target_space": target, |
| "target_space_id": target, |
| "target_space_known": bool(target), |
| "target_space_url": target_url, |
| "target_space_settings_url": f"{target_url}/settings" if target_url else "", |
| "space_created": bool(space_created), |
| "runtime_uploaded": bool(runtime_uploaded), |
| "space_uploaded": bool(runtime_uploaded), |
| "space_runtime_known": bool(space_runtime_known), |
| "links_ready": bool(links_ready), |
| "can_open_space": bool(links_ready), |
| "source": "+".join(dict.fromkeys([s for s in sources if s])) or "resolved_from_persisted_artifacts", |
| } |
|
|
|
|
| def _space_links_ready_from_bundle(bundle: dict[str, Any]) -> bool: |
| return bool(build_space_link_state(bundle).get("links_ready")) |
|
|
|
|
|
|
| def _number_from_value(value: Any) -> float | None: |
| if value is None or value == "": |
| return None |
| if isinstance(value, (int, float)) and not isinstance(value, bool): |
| number = float(value) |
| return number if number > 0 else None |
| if isinstance(value, dict): |
| |
| for key in ( |
| "seconds", |
| "duration_seconds", |
| "recommended_zero_gpu_duration_seconds", |
| "recommended_zerogpu_duration_seconds", |
| "recommended_duration_seconds", |
| "latency_seconds", |
| "observed_latency_seconds", |
| ): |
| found = _number_from_value(value.get(key)) |
| if found is not None: |
| return found |
| return None |
| text = str(value).strip().lower().replace("seconds", "").replace("second", "").replace("sec", "").replace("s", "") |
| try: |
| number = float(text) |
| return number if number > 0 else None |
| except Exception: |
| return None |
|
|
|
|
| def _first_number(*values: Any) -> float | None: |
| for value in values: |
| number = _number_from_value(value) |
| if number is not None: |
| return number |
| return None |
|
|
|
|
| def _latest_successful_linked_validation(bundle: dict[str, Any]) -> dict[str, Any]: |
| linked = bundle.get("linked_validations") or {} |
| rows = linked.get("validations") if isinstance(linked, dict) else [] |
| if not isinstance(rows, list): |
| return {} |
| successful = [row for row in rows if isinstance(row, dict) and _lower(row.get("status") or row.get("effective_status")) in {"success", "passed", "succeeded", "full_inference_success", "validated_after_space_test", "validated_after_manual_space_test"}] |
| if not successful: |
| return {} |
| successful.sort(key=lambda row: str(row.get("validated_at") or row.get("updated_at") or row.get("created_at") or ""), reverse=True) |
| return successful[0] |
|
|
|
|
| def build_validation_metrics(bundle: dict[str, Any]) -> dict[str, Any]: |
| """Durable validation/latency model for Active Run. |
| |
| v198.26.9 invariant: latency and ZeroGPU duration are post-run facts, not |
| incidental event text. They are resolved once from persisted automatic |
| smoke artifacts and linked Space Test artifacts, then exposed by /view and |
| /progress so the frontend does not have to guess. |
| """ |
| bundle = bundle or {} |
| summary = bundle.get("summary") or {} |
| state = bundle.get("state") or {} |
| smoke = bundle.get("generation_smoke") or {} |
| gate = bundle.get("inference_gate") or {} |
| gate_smoke = gate.get("generation_smoke") if isinstance(gate.get("generation_smoke"), dict) else {} |
| post_build = bundle.get("post_build_validation_status") or {} |
| manual = bundle.get("manual_validation_status") or {} |
| linked_success = _latest_successful_linked_validation(bundle) |
| gate_recommendation = gate.get("zero_gpu_duration_recommendation") if isinstance(gate, dict) else None |
|
|
| |
| |
| source_order = [linked_success, post_build if _lower(post_build.get("status")) == "success" else {}, manual if _lower(manual.get("status")) == "success" else {}, smoke, gate_smoke, gate, summary, state] |
| latency = _first_number(*(src.get("latency_seconds") for src in source_order if isinstance(src, dict)), *(src.get("observed_latency_seconds") for src in source_order if isinstance(src, dict))) |
| duration = _first_number( |
| *(src.get("recommended_zero_gpu_duration_seconds") for src in source_order if isinstance(src, dict)), |
| *(src.get("recommended_zerogpu_duration_seconds") for src in source_order if isinstance(src, dict)), |
| *(src.get("recommended_duration_seconds") for src in source_order if isinstance(src, dict)), |
| gate_recommendation, |
| ) |
| api_name = _first_nonempty(*(src.get("api_name") for src in source_order if isinstance(src, dict)), gate.get("selected_api_name"), gate.get("selected_endpoint")) |
| hardware = _first_nonempty( |
| *(src.get("hardware_used_for_validation") for src in source_order if isinstance(src, dict)), |
| *(src.get("recommendation_hardware") for src in source_order if isinstance(src, dict)), |
| smoke.get("hardware"), |
| gate.get("hardware"), |
| summary.get("selected_hardware"), |
| state.get("selected_hardware"), |
| ) |
| output_artifact = _first_nonempty(*(src.get("output_artifact") or src.get("artifact_url") for src in source_order if isinstance(src, dict))) |
| source = "" |
| if linked_success or _lower(post_build.get("status")) == "success" or _lower(manual.get("status")) == "success": |
| source = "linked_space_test" |
| elif smoke.get("status") == "success" or smoke.get("ok") is True: |
| source = "automatic_generation_smoke" |
| elif gate_smoke or gate: |
| source = "inference_gate" |
| source = _first_nonempty(*(src.get("recommendation_source") for src in source_order if isinstance(src, dict)), source) |
| health_passed = bool((gate.get("implementation_signals") or {}).get("health_passed") or smoke.get("health_passed") or gate_smoke.get("health_passed")) |
| smoke_passed = bool(smoke.get("ok") or _lower(smoke.get("status")) == "success" or _lower(gate_smoke.get("status")) == "success" or (gate.get("implementation_signals") or {}).get("generation_smoke_passed") is True) |
| validation_run_id = _first_nonempty(linked_success.get("validation_run_id"), linked_success.get("run_id"), post_build.get("validation_run_id"), manual.get("validation_run_id")) |
| metrics = { |
| "schema_version": "validation_metrics.v198_26_9", |
| "available": bool(latency or duration or smoke_passed or health_passed or validation_run_id), |
| "source": source, |
| "latency_seconds": latency, |
| "observed_latency_seconds": latency, |
| "recommended_zero_gpu_duration_seconds": duration, |
| "recommended_zerogpu_duration_seconds": duration, |
| "hardware_used_for_validation": hardware, |
| "recommendation_hardware": hardware, |
| "api_name": api_name, |
| "output_artifact": output_artifact, |
| "health_passed": health_passed, |
| "generation_smoke_passed": smoke_passed, |
| "automatic_smoke_passed": bool(smoke.get("ok") or _lower(smoke.get("status")) == "success"), |
| "linked_space_test_passed": bool(linked_success or _lower(post_build.get("status")) == "success" or _lower(manual.get("status")) == "success"), |
| "validation_run_id": validation_run_id, |
| "updated_at": _first_nonempty(linked_success.get("validated_at"), linked_success.get("updated_at"), post_build.get("validated_at"), post_build.get("updated_at"), manual.get("validated_at"), manual.get("updated_at"), summary.get("updated_at"), state.get("updated_at")), |
| } |
| return metrics |
|
|
| def _known_gradio_endpoint_info(bundle: dict[str, Any]) -> dict[str, Any]: |
| smoke = bundle.get("generation_smoke") or {} |
| gate = bundle.get("inference_gate") or {} |
| live = bundle.get("live_status") or {} |
| schema = bundle.get("api_schema") or bundle.get("gradio_schema") or {} |
| endpoints: list[Any] = [] |
| for source in (smoke, gate, schema, live): |
| if not isinstance(source, dict): |
| continue |
| for key in ("discovered_api_names", "api_names", "endpoints", "api_endpoints", "named_endpoints"): |
| value = source.get(key) |
| if isinstance(value, list): |
| endpoints.extend(value) |
| endpoint = _first_nonempty(smoke.get("api_name"), smoke.get("endpoint"), gate.get("api_name"), schema.get("selected_api_name")) |
| endpoint_count = len([e for e in endpoints if e]) |
| endpoint_known = bool(endpoint and endpoint != "/generate") or endpoint_count > 0 or bool(smoke.get("status") == "success") |
| return {"endpoint": endpoint or "/generate", "endpoint_count": endpoint_count, "endpoint_known": endpoint_known, "endpoints": endpoints[:20]} |
|
|
|
|
| def _contract_declares_no_full_inference(bundle: dict[str, Any]) -> bool: |
| contract = bundle.get("inference_contract") or bundle.get("INFERENCE_CONTRACT") or {} |
| if not isinstance(contract, dict): |
| contract = {} |
| smoke = bundle.get("generation_smoke") or {} |
| blockers = bundle.get("technical_blockers") or {} |
| statuses = _raw_statuses(bundle) |
| return bool( |
| "technical_blocker_boot_only" in statuses |
| or contract.get("full_inference_implemented") is False |
| or (contract.get("primary_api_name") in {None, "", False} and _lower(contract.get("validation_level")) in {"boot-only", "boot_only", "health-only", "health_only"}) |
| or (_lower(smoke.get("status")) == "skipped" and _lower(smoke.get("skip_reason") or smoke.get("reason")) in {"contract_declared_no_full_inference", "full_inference_not_implemented"}) |
| or (_lower(blockers.get("status")) in {"technical_blocker_boot_only", "technical_blocker"} and contract.get("full_inference_implemented") is False) |
| ) |
|
|
|
|
| def _repair_auth_context(bundle: dict[str, Any], status_model: dict[str, Any]) -> bool: |
| repair = bundle.get("repair_outcome") or {} |
| summary = bundle.get("summary") or {} |
| state = bundle.get("state") or {} |
| values = { |
| _lower(status_model.get("global_status")), |
| _lower(status_model.get("verdict")), |
| _lower(summary.get("status")), |
| _lower(state.get("status")), |
| _lower(summary.get("failure_type")), |
| _lower(state.get("failure_type")), |
| _lower(repair.get("post_repair_validation")), |
| _lower(repair.get("failure_type")), |
| _lower(repair.get("status")), |
| } |
| return bool(values.intersection(AUTH_RAW_STATUSES)) |
|
|
|
|
| def build_space_test_policy(bundle: dict[str, Any], status_model: dict[str, Any]) -> dict[str, Any]: |
| """Return the canonical linked Space Test policy for a Build Run. |
| |
| Space Test is linked-only: parent Build Run and target Space are mandatory, |
| but a known endpoint is optional because the validation worker can discover |
| Gradio endpoints before generating a smoke-test payload. |
| """ |
| summary = bundle.get("summary") or {} |
| state = bundle.get("state") or {} |
| smoke = bundle.get("generation_smoke") or {} |
| runtime = bundle.get("space_runtime") or {} |
| target = _space_target_from_bundle(bundle) |
| target_url = _space_target_url_from_bundle(bundle, target) |
| raw_status = _lower(status_model.get("raw_status") or summary.get("status") or state.get("status")) |
| global_status = _lower(status_model.get("global_status")) |
| verdict = _lower(status_model.get("verdict")) |
| endpoint_info = _known_gradio_endpoint_info(bundle) |
| requires_discovery = bool(target and not endpoint_info.get("endpoint_known")) |
| runtime_stage = _lower(runtime.get("stage") or runtime.get("status")) if isinstance(runtime, dict) else "" |
| health_passed = bool((bundle.get("inference_gate") or {}).get("implementation_signals", {}).get("health_passed") or smoke.get("health_passed")) |
|
|
| def policy(mode: str, enabled: bool, reason: str, label: str, message: str, *, on_success: str = "unchanged", on_failure: str = "unchanged", allow_replay: bool = False, allow_recovery: bool = False) -> dict[str, Any]: |
| return { |
| "schema_version": "space_test_policy.v1", |
| "mode": mode, |
| "enabled": bool(enabled), |
| "reason": reason, |
| "label": label, |
| "message": message, |
| "target_space": target, |
| "target_space_url": target_url, |
| "parent_build_run_id": summary.get("run_id") or state.get("run_id") or "", |
| "endpoint": endpoint_info.get("endpoint") or "/generate", |
| "endpoint_known": bool(endpoint_info.get("endpoint_known")), |
| "endpoint_count": endpoint_info.get("endpoint_count") or 0, |
| "requires_endpoint_discovery": bool(requires_discovery and enabled), |
| "can_update_parent_effective_status": on_success != "unchanged", |
| "effective_status_on_success": on_success, |
| "effective_status_on_failure": on_failure, |
| "allow_replay": bool(allow_replay), |
| "allow_recovery": bool(allow_recovery), |
| } |
|
|
| if not target: |
| return policy("unavailable", False, "no_target_space", "Space Test unavailable", "No generated Space is available to validate.") |
| if _repair_auth_context(bundle, status_model): |
| return policy("recover", True, "auth_refresh_required", "Retry after sign-in refresh", "Repair patch may have been uploaded, but validation could not continue because HF OAuth expired. Sign in again, then retry linked validation.", on_success="recovered_by_manual_validation", allow_recovery=True) |
| if _contract_declares_no_full_inference(bundle): |
| blocked = policy("blocked", False, "no_generation_endpoint_by_contract", "Blocked — no generation endpoint", "Health may pass, but Pi declared full inference unavailable and no generation endpoint exists. Review TECHNICAL_BLOCKERS.json / PI_SUMMARY.md or provide a dedicated implementation and hardware plan.") |
| blocked["endpoint"] = "" |
| blocked["requires_endpoint_discovery"] = False |
| blocked["no_generation_endpoint"] = True |
| blocked["can_retry_schema"] = False |
| return blocked |
| if global_status in {"running", "unknown"}: |
| return policy("blocked", False, "build_not_terminal", "Available after final Space state", "Space Test becomes available after the Build Run reaches a final Space state.") |
| if global_status == "cancelled" or raw_status in {"stopped", "cancelled", "canceled"}: |
| return policy("unavailable", False, "run_stopped", "Space Test unavailable", "This run was stopped. Start a new build before validation.") |
| if status_model.get("manual_validation_passed") or _lower(status_model.get("effective_status")) in {"validated_after_manual_space_test", "manual_validation_passed", "validated"}: |
| return policy("replay", True, "manual_validation_already_passed", "Replay linked validation", "This Build Run already has a successful linked Space Test. You can replay the validation.", allow_replay=True) |
| if raw_status in {"completed_with_warnings", "success_with_warnings"}: |
| return policy("replay", True, "completed_with_warnings", "Replay validation", "Build completed with warnings. Space Test can replay or confirm the validated endpoint.", allow_replay=True) |
| if global_status == "succeeded" or verdict in {"passed", "success"}: |
| return policy("replay", True, "automatic_smoke_passed", "Replay automatic smoke", "Automatic smoke test passed. You can replay the validated request.", allow_replay=True) |
| if global_status == "partial" or verdict == "partial_validation" or raw_status in PARTIAL_RAW_STATUSES: |
| return policy("complete", True, "partial_generation_not_verified", "Complete validation", "Automatic smoke did not verify generation. Run linked Space Test to complete validation.", on_success="validated_after_manual_space_test") |
| if global_status == "waiting_manual_action" or status_model.get("requires_manual_action"): |
| reachable = "running" in runtime_stage or health_passed or endpoint_info.get("endpoint_known") |
| if reachable: |
| return policy("complete", True, "manual_hardware_action_possible", "Validate after manual action", "Manual hardware action may be required. If the Space is now running, run linked validation.", on_success="validated_after_manual_space_test") |
| return policy("blocked", False, "manual_hardware_required", "Manual hardware action required", "Complete the hardware action in Space Settings before validation.") |
| if global_status == "failed": |
| return policy("recover", True, "failed_after_space_creation", "Recover validation", "The generated Space previously failed validation. If it is now reachable, run linked Space Test to check recovery.", on_success="recovered_by_manual_validation", allow_recovery=True) |
| if global_status == "stale": |
| return policy("recover", True, "stale_with_target_space", "Recover stale run", "The original run became stale. If the Space is now available, run linked validation to recover the result.", on_success="recovered_by_manual_validation", allow_recovery=True) |
| return policy("blocked", False, "unsupported_build_state", "Space Test unavailable", "This Build Run state is not eligible for linked Space Test yet.") |
|
|
|
|
| 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 {} |
| metrics = build_validation_metrics(bundle) |
| policy = build_space_test_policy(bundle, status_model) |
| |
| |
| |
| |
| |
| if policy.get("no_generation_endpoint") is True or policy.get("reason") == "no_generation_endpoint_by_contract": |
| endpoint = "" |
| else: |
| endpoint = policy.get("endpoint") or smoke.get("api_name") or smoke.get("endpoint") or "/generate" |
| parent_blocked = _lower(status_model.get("global_status")) in {"blocked", "failed", "auth_refresh_required", "cancelled"} or _lower(status_model.get("verdict")) in BLOCKED_RAW_STATUSES | FAILED_RAW_STATUSES | AUTH_RAW_STATUSES | {"cancelled", "stopped"} |
| smoke_evidence_passed = bool(metrics.get("generation_smoke_passed") or smoke.get("ok") or smoke.get("status") == "success") |
| preview_status = "blocked" if parent_blocked else "passed" if smoke_evidence_passed else "pending" |
| return { |
| "target_space": policy.get("target_space") or summary.get("target_space") or "", |
| "target_space_url": policy.get("target_space_url") or summary.get("target_space_url") or "", |
| "endpoint": endpoint, |
| "status": preview_status, |
| "smoke_evidence_passed": smoke_evidence_passed, |
| "latency_seconds": metrics.get("latency_seconds") or smoke.get("latency_seconds") or summary.get("latency_seconds"), |
| "observed_latency_seconds": metrics.get("observed_latency_seconds"), |
| "recommended_zero_gpu_duration_seconds": metrics.get("recommended_zero_gpu_duration_seconds"), |
| "recommended_zerogpu_duration_seconds": metrics.get("recommended_zerogpu_duration_seconds"), |
| "validation_metrics": metrics, |
| "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 "", |
| "policy": policy, |
| } |
|
|
|
|
| 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 {} |
| manual_validation = bundle.get("manual_validation_status") or summary.get("manual_validation_status") or {} |
| status_model = normalize_run_status(bundle, now=now) |
| effective_run_status = bundle.get("effective_run_status") if isinstance(bundle.get("effective_run_status"), dict) else compute_effective_run_status(bundle, build_status=status_model.get("verdict") or status_model.get("global_status"), build_verdict=status_model.get("verdict")) |
| if effective_run_status.get("post_build_status") not in {"", "none", None}: |
| legacy_status = effective_run_status.get("legacy_effective_status") or manual_validation.get("effective_status") or "validated_after_manual_space_test" |
| status_model = { |
| **status_model, |
| "manual_validation_passed": True, |
| "post_build_validation_passed": True, |
| "post_build_status": effective_run_status.get("post_build_status"), |
| "effective_status": effective_run_status.get("effective_status") or legacy_status, |
| "effective_verdict": effective_run_status.get("effective_verdict") or legacy_status, |
| "legacy_effective_status": legacy_status, |
| } |
| phase = derive_product_phase(bundle, status_model) |
| space_test_policy = build_space_test_policy(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()) |
| target = _space_target_from_bundle(bundle) |
| target_url = _space_target_url_from_bundle(bundle, target) |
| space_link_state = build_space_link_state(bundle, target=target, target_url=target_url) |
| validation_metrics = build_validation_metrics({**bundle, "summary": summary}) |
| links = { |
| "job_url": _job_url_from_view_sources(bucket_source=bucket_source, summary=summary, state=state, launch=bundle.get("launch") or {}), |
| "target_space_url": target_url, |
| "target_space_settings_url": f"{target_url}/settings" if target_url else "", |
| "artifacts_url": summary.get("artifacts_url") or f"https://huggingface.co/buckets/{bucket_source}/tree/{_runs_prefix()}/{run_id}", |
| "links_ready": bool(space_link_state.get("links_ready")), |
| "space_created": bool(space_link_state.get("space_created")), |
| "space_uploaded": bool(space_link_state.get("runtime_uploaded")), |
| "runtime_uploaded": bool(space_link_state.get("runtime_uploaded")), |
| "space_runtime_known": bool(space_link_state.get("space_runtime_known")), |
| "source": space_link_state.get("source") or "", |
| } |
| 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(), |
| "display_status": effective_run_status.get("display_status") or status_model.get("global_status"), |
| "display_label": effective_run_status.get("display_label") or status_model["global_status"].replace("_", " ").title(), |
| "raw_status": status_model["raw_status"], |
| "space": target or summary.get("target_space") or "", |
| "space_url": target_url, |
| "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, |
| "effective_run_status": effective_run_status, |
| "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), "linked_validation": manual_validation, "mode": "linked_build_run"}, |
| "space_test_policy": space_test_policy, |
| "space_link_state": space_link_state, |
| "validation_metrics": validation_metrics, |
| "run_documents": bundle.get("run_documents") or [], |
| "manual_validation": manual_validation, |
| "post_build_validation": effective_run_status.get("post_build_validation") or {}, |
| "display_status": effective_run_status.get("display_status") or status_model.get("global_status"), |
| "effective_status": effective_run_status.get("effective_status") or status_model.get("effective_status") or status_model["verdict"], |
| "effective_verdict": effective_run_status.get("effective_verdict") or status_model.get("effective_verdict") or status_model["verdict"], |
| "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(space_link_state.get("can_open_space")), |
| "can_validate": bool(space_test_policy.get("enabled")), |
| "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] |
|
|