| from __future__ import annotations |
|
|
| from typing import Any |
|
|
| from .progress import progress_from_events |
|
|
| PHASE_ORDER = [ |
| "start", |
| "model", |
| "agent", |
| "hardware", |
| "deploy", |
| "live_validation", |
| "recovery", |
| "archive", |
| "done", |
| ] |
|
|
| PHASE_LABELS = { |
| "start": "Start", |
| "model": "Model", |
| "agent": "Agent", |
| "hardware": "GPU", |
| "deploy": "Deploy", |
| "live_validation": "Live test", |
| "recovery": "Recovery", |
| "archive": "Archive", |
| "done": "Done", |
| } |
|
|
| PHASE_STEPS = { |
| "start": {"token_context", "bootstrap", "dependencies", "auth", "workspace", "node", "pi_install"}, |
| "model": {"model_analysis", "model_prescan"}, |
| "agent": {"pi_config", "pi_run", "traces", "pi_model_resolution", "pi_verification"}, |
| "hardware": {"hardware_strategy", "create_space_hardware", "create_space"}, |
| "deploy": {"metadata_sanitize", "requirements_sanitize", "upload_files", "space_logs", "space_runtime", "live_wait"}, |
| "live_validation": {"endpoint_discovery", "api_validation", "generation_smoke", "inference_gate"}, |
| "recovery": { |
| "failure_detected", |
| "failure_diagnosis", |
| "pi_diagnosis", |
| "repair_decision", |
| "wait_for_logs", |
| "factory_rebuild", |
| "repair", |
| "repair_diagnosis", |
| "repair_brief", |
| "repair_plan", |
| "repair_patch", |
| "repair_upload", |
| "repair_validation", |
| }, |
| "archive": {"report_write", "artifact_manifest", "anonymous_eval", "eval_publish", "eval_archive"}, |
| "done": {"done", "failure", "technical_blocker", "technical_blocker_boot_only", "manual_hardware_required"}, |
| } |
|
|
| STEP_TO_PHASE = {step: phase for phase, steps in PHASE_STEPS.items() for step in steps} |
|
|
| SUCCESS_STATUSES = {"success", "done", "passed", "completed", "full_inference_success"} |
| PARTIAL_STATUSES = {"full_inference_candidate_health_passed", "health_only", "partial", "partial_validation", "completed_with_warnings"} |
| FAILED_STATUSES = {"failed", "failure", "error"} |
| RUNNING_STATUSES = {"started", "running", "waiting", "pending", "scheduled"} |
| CANCELLED_STATUSES = {"cancelled", "canceled"} |
| MANUAL_STATUSES = {"manual_hardware_required", "manual_action_required", "generated_needs_manual_hardware", "waiting_manual_hardware"} |
| BLOCKED_STATUSES = {"technical_blocker", "technical_blocker_boot_only", "blocked"} |
| AUTH_STATUSES = {"auth_refresh_required", "oauth_expired", "auth_expired", "repair_validation_inconclusive_auth", "inconclusive_auth_expired"} |
|
|
| TERMINAL_SOURCE_KEYS = ("status", "final_status", "effective_status", "effective_verdict", "display_status", "verdict", "result_status", "validation_status") |
|
|
|
|
| def _lower(value: Any) -> str: |
| return str(value or "").strip().lower() |
|
|
|
|
| def _events(bundle: dict[str, Any]) -> list[dict[str, Any]]: |
| return [event for event in (bundle.get("events") or []) if isinstance(event, dict)] |
|
|
|
|
| def _event_steps(bundle: dict[str, Any]) -> set[str]: |
| return {_lower(event.get("step")) for event in _events(bundle) if event.get("step")} |
|
|
|
|
| def _events_for_phase(bundle: dict[str, Any], phase: str) -> list[dict[str, Any]]: |
| steps = PHASE_STEPS.get(phase, set()) |
| return [event for event in _events(bundle) if _lower(event.get("step")) in steps] |
|
|
|
|
| def _latest_event_for_phase(bundle: dict[str, Any], phase: str) -> dict[str, Any] | None: |
| events = _events_for_phase(bundle, phase) |
| return events[-1] if events else None |
|
|
|
|
| def _latest_events_by_step(bundle: dict[str, Any], phase: str) -> dict[str, dict[str, Any]]: |
| latest: dict[str, dict[str, Any]] = {} |
| for event in _events_for_phase(bundle, phase): |
| step = _lower(event.get("step")) |
| if step: |
| latest[step] = event |
| return latest |
|
|
|
|
| def _has_downstream_proof(bundle: dict[str, Any], phase: str) -> bool: |
| steps = _event_steps(bundle) |
| phase_index = PHASE_ORDER.index(phase) |
| return any(steps.intersection(PHASE_STEPS[p]) for p in PHASE_ORDER[phase_index + 1 :]) |
|
|
|
|
| def _has_successful_phase_event(bundle: dict[str, Any], phase: str) -> bool: |
| return any(_lower(e.get("status")) in SUCCESS_STATUSES for e in _events_for_phase(bundle, phase)) |
|
|
|
|
| def _has_status(bundle: dict[str, Any], statuses: set[str]) -> bool: |
| state = bundle.get("state") or {} |
| summary = bundle.get("summary") or {} |
| gate = bundle.get("inference_gate") or {} |
| smoke = bundle.get("generation_smoke") or {} |
| values = { |
| _lower(state.get("status")), |
| _lower(state.get("gate_status")), |
| _lower(summary.get("status")), |
| _lower(gate.get("status")), |
| _lower(smoke.get("status")), |
| } |
| values.update(_lower(event.get("status")) for event in _events(bundle)) |
| return bool(values.intersection(statuses)) |
|
|
|
|
| def _gate(bundle: dict[str, Any]) -> dict[str, Any]: |
| return bundle.get("inference_gate") or {} |
|
|
|
|
| def _signals(bundle: dict[str, Any]) -> dict[str, Any]: |
| gate = _gate(bundle) |
| return gate.get("implementation_signals") if isinstance(gate.get("implementation_signals"), dict) else {} |
|
|
|
|
| def _smoke(bundle: dict[str, Any]) -> dict[str, Any]: |
| smoke = bundle.get("generation_smoke") or {} |
| if smoke: |
| return smoke |
| gate_smoke = (_gate(bundle).get("generation_smoke") or {}) |
| return gate_smoke if isinstance(gate_smoke, dict) else {} |
|
|
|
|
| def _runtime(bundle: dict[str, Any]) -> dict[str, Any]: |
| runtime = bundle.get("space_runtime") or {} |
| return runtime if isinstance(runtime, dict) else {} |
|
|
|
|
| def _live_status(bundle: dict[str, Any]) -> dict[str, Any]: |
| live = bundle.get("live_status") or {} |
| return live if isinstance(live, dict) else {} |
|
|
|
|
| def _latest_event_by_step(bundle: dict[str, Any], step_name: str) -> dict[str, Any]: |
| target = _lower(step_name) |
| for event in reversed(_events(bundle)): |
| if _lower(event.get("step")) == target: |
| return event |
| return {} |
|
|
|
|
|
|
| def _runtime_history(bundle: dict[str, Any], *, live: dict[str, Any], runtime: dict[str, Any], smoke: dict[str, Any], gate: dict[str, Any]) -> list[dict[str, Any]]: |
| """Return a compact build→runtime→API→smoke history for Live test. |
| |
| v198.26.9 keeps this curated: it restores the building/runtime feedback |
| users need without dumping raw events into Active Run. |
| """ |
| history: list[dict[str, Any]] = [] |
| seen: set[str] = set() |
|
|
| def add(stage: str, label: str, status: str = "info", ts: Any = "", detail: Any = "") -> None: |
| key = f"{stage}:{label}:{status}" |
| if key in seen: |
| return |
| seen.add(key) |
| history.append({"stage": stage, "label": label, "status": status, "ts": ts or "", "detail": str(detail or "")[:500]}) |
|
|
| for event in _events(bundle): |
| step = _lower(event.get("step")) |
| status = _lower(event.get("status")) or "info" |
| ts = event.get("ts") or event.get("created_at") or "" |
| msg = event.get("message") or event.get("step") or "" |
| if step in {"create_space", "create_space_hardware"}: |
| add("space_created", "Space created" if status in SUCCESS_STATUSES else "Creating Space", status, ts, msg) |
| elif step == "upload_files": |
| add("runtime_uploaded", "Runtime uploaded" if status in SUCCESS_STATUSES else "Uploading runtime", status, ts, msg) |
| elif step in {"space_runtime", "live_wait"}: |
| lower_msg = _lower(msg) |
| if "building" in lower_msg or "build" in lower_msg: |
| add("space_building", "Space is building", status, ts, msg) |
| elif "running" in lower_msg or "runtime" in lower_msg or step == "space_runtime": |
| add("space_running", "Space runtime observed", status, ts, msg) |
| else: |
| add("space_runtime", "Space runtime check", status, ts, msg) |
| elif step in {"endpoint_discovery", "api_validation"}: |
| add("api_schema", "Gradio API checked" if status in SUCCESS_STATUSES else "Checking Gradio API", status, ts, msg) |
| elif step == "generation_smoke": |
| add("generation_smoke", "Generation smoke passed" if status in SUCCESS_STATUSES else "Generation smoke running" if status in RUNNING_STATUSES else "Generation smoke checked", status, ts, msg) |
| elif step == "inference_gate": |
| add("inference_gate", "Inference gate resolved", status, ts, msg) |
|
|
| runtime_stage = str(runtime.get("stage") or runtime.get("status") or runtime.get("runtime_status") or "").strip() |
| upper_stage = runtime_stage.upper() |
| if upper_stage: |
| if "BUILD" in upper_stage and not any(h["stage"] == "space_building" for h in history): |
| add("space_building", "Space is building", "running", runtime.get("updated_at") or "", runtime_stage) |
| if "RUNNING" in upper_stage and not any(h["stage"] == "space_running" for h in history): |
| add("space_running", "Space runtime is running", "success", runtime.get("updated_at") or "", runtime_stage) |
| if "ERROR" in upper_stage: |
| add("space_runtime_error", "Space runtime error", "failed", runtime.get("updated_at") or "", runtime.get("error") or runtime_stage) |
| if live.get("stage") and not history: |
| add(str(live.get("stage")), str(live.get("message") or live.get("stage")), str(live.get("status") or "info"), live.get("updated_at") or "") |
| endpoints = gate.get("endpoints") or gate.get("api_endpoints") or gate.get("named_endpoints") or [] |
| selected = str(smoke.get("api_name") or gate.get("api_name") or gate.get("selected_endpoint") or gate.get("selected_api_name") or "").strip() |
| if (selected or (isinstance(endpoints, list) and endpoints)) and not any(h["stage"] == "api_schema" for h in history): |
| add("api_schema", f"Gradio API detected{': ' + selected if selected else ''}", "success") |
| if (_lower(smoke.get("status")) == "success" or smoke.get("ok") is True or (gate.get("implementation_signals") or {}).get("generation_smoke_passed") is True) and not any(h["stage"] == "generation_smoke" for h in history): |
| add("generation_smoke", "Generation smoke passed", "success", smoke.get("updated_at") or "", smoke.get("api_name") or "") |
| return history[-10:] |
|
|
| def build_live_validation_model(bundle: dict[str, Any]) -> dict[str, Any]: |
| """Compact live validation telemetry for the UI. |
| |
| The timeline phase remains the coarse product step. This model gives the |
| Live Validation card concrete runtime facts while the Space is building, |
| starting, exposing Gradio endpoints, or running generation smoke tests. |
| """ |
| runtime = _runtime(bundle) |
| smoke = _smoke(bundle) |
| gate = _gate(bundle) |
| signals = _signals(bundle) |
| api_event = _latest_event_by_step(bundle, "api_validation") |
| smoke_event = _latest_event_by_step(bundle, "generation_smoke") |
| live_status = _live_status(bundle) |
| live_stage = str(live_status.get("stage") or "").strip() |
| live_message = str(live_status.get("message") or "").strip() |
| runtime_stage = str(runtime.get("stage") or runtime.get("status") or "").strip() |
| runtime_error = str(runtime.get("error") or "").strip() |
| health_passed = signals.get("health_passed") is True or _lower(api_event.get("status")) in SUCCESS_STATUSES |
| health_failed = signals.get("health_passed") is False or _lower(api_event.get("status")) in FAILED_STATUSES |
| smoke_status = _lower(smoke.get("status") or smoke_event.get("status")) |
| endpoints = gate.get("endpoints") or gate.get("api_endpoints") or gate.get("named_endpoints") or [] |
| endpoint_count = len(endpoints) if isinstance(endpoints, list) else 0 |
| selected_endpoint = str( |
| smoke.get("api_name") |
| or gate.get("api_name") |
| or gate.get("selected_endpoint") |
| or gate.get("selected_api_name") |
| or "" |
| ).strip() |
| endpoint_known = bool(endpoint_count or selected_endpoint or smoke_status == "success" or signals.get("generation_smoke_passed") is True) |
|
|
| severity = "info" |
| stage = "waiting_for_space" |
| label = "Waiting for Space runtime" |
| next_action = "Waiting for the generated Space to build and start." |
|
|
| upper_stage = runtime_stage.upper() |
| if "BUILD_ERROR" in upper_stage: |
| severity = "error" |
| stage = "space_build_error" |
| label = "Space build error" |
| next_action = "Open Space logs or let the recovery protocol inspect the build failure." |
| elif "RUNTIME_ERROR" in upper_stage: |
| severity = "error" |
| stage = "space_runtime_error" |
| label = "Space runtime error" |
| next_action = "Open runtime logs or retry after the recovery protocol finishes." |
| elif smoke_status == "success" or signals.get("generation_smoke_passed") is True: |
| severity = "success" |
| stage = "generation_verified" |
| label = "Generation verified" |
| next_action = "The generated Space passed the automatic generation smoke test." |
| elif smoke and smoke_status in FAILED_STATUSES.union({"timeout", "exception"}): |
| severity = "warning" |
| stage = "generation_smoke_failed" |
| label = "Generation not verified" |
| if smoke.get("auto_retry", {}).get("retried"): |
| next_action = "The factory retried once with schema-corrected values. Use Prefill Space Test to retry manually if needed." |
| elif smoke.get("retryable_with_schema_payload") or smoke.get("auto_retry_supported"): |
| next_action = "Use Prefill Space Test after the run finishes to retry with schema-adjusted args." |
| else: |
| next_action = "Use Prefill Space Test after the run finishes to retry with adjusted args." |
| elif live_stage == "generation_smoke_retry": |
| severity = "info" |
| stage = "generation_smoke_retry" |
| label = "Retrying generation smoke" |
| next_action = "Retrying once with schema-corrected Gradio choice values." |
| elif live_stage == "generation_smoke": |
| severity = "info" |
| stage = "generation_smoke_running" |
| label = "Generation smoke running" |
| next_action = "Waiting for the generated endpoint response." |
| elif live_stage == "endpoint_discovery": |
| severity = "info" |
| stage = "endpoint_discovery" |
| label = "Discovering Gradio endpoints" |
| next_action = "Inspecting the Space API schema before validation." |
| elif _latest_event_by_step(bundle, "endpoint_discovery"): |
| event = _latest_event_by_step(bundle, "endpoint_discovery") |
| severity = "success" if _lower(event.get("status")) in SUCCESS_STATUSES else "info" |
| stage = "endpoint_discovery" |
| label = "Gradio endpoint discovery" |
| next_action = str(event.get("message") or "Endpoint discovery completed.") |
| elif health_passed: |
| severity = "info" |
| stage = "health_passed" |
| label = "Health passed; waiting for generation smoke" |
| next_action = "Waiting for Gradio endpoints and automatic generation validation." |
| elif health_failed: |
| severity = "warning" |
| stage = "health_not_confirmed" |
| label = "Health not confirmed" |
| next_action = "Waiting for the Space API to become reachable or fail conclusively." |
| elif "BUILD" in upper_stage: |
| stage = "space_building" |
| label = "Space is building" |
| elif "RUNNING" in upper_stage or "START" in upper_stage: |
| stage = "space_starting" |
| label = "Space runtime is starting" |
| next_action = "Waiting for the live API and Gradio schema." |
| elif endpoint_count: |
| stage = "gradio_schema_detected" |
| label = "Gradio endpoints detected" |
| next_action = "Waiting for automatic generation smoke validation." |
|
|
| return { |
| "stage": stage, |
| "label": label, |
| "severity": severity, |
| "runtime_stage": runtime_stage, |
| "runtime_error": runtime_error, |
| "health": "passed" if health_passed else "failed" if health_failed else "pending", |
| "gradio_schema": "detected" if endpoint_known else "pending", |
| "endpoint_count": endpoint_count, |
| "selected_endpoint": selected_endpoint, |
| "generation_smoke": "passed" if smoke_status == "success" else "failed" if smoke and smoke_status in FAILED_STATUSES.union({"timeout", "exception"}) else "pending", |
| "live_status_stage": live_stage, |
| "live_status_updated_at": live_status.get("updated_at") or "", |
| "message": smoke.get("error") or smoke.get("message") or live_message or api_event.get("message") or runtime_error or "", |
| "next_action": next_action, |
| "runtime_history": _runtime_history(bundle, live=live_status, runtime=runtime, smoke=smoke, gate=gate), |
| } |
|
|
|
|
| def _eval_publish(bundle: dict[str, Any]) -> dict[str, Any]: |
| status = bundle.get("eval_publish_status") or bundle.get("eval_publish") or {} |
| return status if isinstance(status, dict) else {} |
|
|
|
|
| def _build_error_observation(bundle: dict[str, Any]) -> dict[str, Any]: |
| observation = bundle.get("build_error_observation") or {} |
| return observation if isinstance(observation, dict) else {} |
|
|
|
|
| def _final_blocker_label(bundle: dict[str, Any]) -> str: |
| observation = _build_error_observation(bundle) |
| reason = _lower(observation.get("reason")) |
| tail = str(observation.get("tail") or observation.get("first_error") or "") |
| if reason or "pyenv install 3.1" in tail or "BUILD FAILED" in tail: |
| if "pyenv install 3.1" in tail or "python3.1" in tail.replace(" ", ""): |
| return "Build failed: invalid Python version requested pyenv install 3.1" |
| return "Build failed after repair" |
| blockage = bundle.get("blockage") or {} |
| if isinstance(blockage, dict): |
| status = str(blockage.get("status") or "").replace("_", " ").strip() |
| decision = blockage.get("decision") if isinstance(blockage.get("decision"), dict) else {} |
| reason_text = str(decision.get("reason") or blockage.get("reason") or "").strip() |
| if reason_text: |
| return reason_text[:220] |
| if status: |
| return status.title() |
| return "" |
|
|
|
|
| def _is_record_not_ready(status: dict[str, Any]) -> bool: |
| return _lower(status.get("reason")) == "record_not_ready" |
|
|
|
|
| def _has_terminal_event(bundle: dict[str, Any]) -> bool: |
| return any(_lower(event.get("step")) in {"done", "failure", "technical_blocker", "technical_blocker_boot_only", "manual_hardware_required"} for event in _events(bundle)) |
|
|
|
|
| def _latest_recovery_event(bundle: dict[str, Any]) -> dict[str, Any] | None: |
| events = _events_for_phase(bundle, "recovery") |
| return events[-1] if events else None |
|
|
|
|
| def _phase_status_from_events(bundle: dict[str, Any], phase: str) -> str: |
| latest_by_step = _latest_events_by_step(bundle, phase) |
| if not latest_by_step: |
| return "pending" |
| statuses = {_lower(event.get("status")) for event in latest_by_step.values()} |
| if statuses.intersection(FAILED_STATUSES): |
| return "failed" |
| if statuses.intersection({"warning"}) or statuses.intersection(PARTIAL_STATUSES): |
| return "warning" |
| if statuses.intersection(RUNNING_STATUSES): |
| |
| |
| |
| |
| |
| return "running" |
| if statuses.intersection(SUCCESS_STATUSES): |
| return "complete" |
| return "complete" |
|
|
|
|
| def _normalize_verdict_token(value: Any) -> str: |
| normalized = _lower(value) |
| if not normalized: |
| return "" |
| if normalized in AUTH_STATUSES: |
| return "auth_refresh_required" |
| if normalized in {"succeeded", "success", "done", "completed", "passed", "repair_success"}: |
| return "full_inference_success" |
| if normalized in SUCCESS_STATUSES: |
| return "full_inference_success" if normalized == "full_inference_success" else "success" |
| if normalized in PARTIAL_STATUSES or normalized in {"partial", "health_only", "full_inference_candidate_health_passed", "completed_with_warnings"}: |
| return "partial_validation" |
| if normalized in MANUAL_STATUSES: |
| return "manual_action_required" |
| if normalized in BLOCKED_STATUSES: |
| return "technical_blocker_boot_only" if normalized == "technical_blocker_boot_only" else "technical_blocker" |
| if normalized in FAILED_STATUSES: |
| return "failed" |
| if normalized in CANCELLED_STATUSES or normalized == "stopped": |
| return "cancelled" |
| if normalized == "stale": |
| return "stale" |
| if normalized in {"running", "pending", "queued", "started", "waiting", "scheduled"}: |
| return "running" |
| return normalized |
|
|
|
|
| def _first_terminal_verdict_from_sources(bundle: dict[str, Any]) -> str: |
| """Return authoritative terminal verdict for canonical timeline convergence. |
| |
| v198.26.7: final reconciliation and summary must win over stale state, |
| live_status, or repair artifacts. This keeps Active Run and Timeline aligned |
| with Run Explorer once the backend has written the final success/partial/fail. |
| """ |
| sources: list[dict[str, Any]] = [] |
| for src in ( |
| bundle.get("final_status_reconciliation") or {}, |
| bundle.get("summary") or {}, |
| bundle.get("summary_file") or {}, |
| bundle.get("effective_run_status") or {}, |
| (bundle.get("summary") or {}).get("effective_run_status") if isinstance((bundle.get("summary") or {}).get("effective_run_status"), dict) else {}, |
| bundle.get("eval_record") or {}, |
| ): |
| if isinstance(src, dict): |
| sources.append(src) |
| for source in sources: |
| for key in TERMINAL_SOURCE_KEYS: |
| verdict = _normalize_verdict_token(source.get(key)) |
| if verdict and verdict != "running": |
| return verdict |
| if source.get("ok") is True or source.get("generation_smoke_passed") is True: |
| return "full_inference_success" |
| return "" |
|
|
|
|
| def _verdict(bundle: dict[str, Any]) -> str: |
| terminal = _first_terminal_verdict_from_sources(bundle) |
| if terminal: |
| return terminal |
| gate = _gate(bundle) |
| state = bundle.get("state") or {} |
| repair = bundle.get("repair_outcome") or {} |
| for value in (gate.get("status"), state.get("gate_status"), state.get("status"), repair.get("post_repair_validation"), repair.get("failure_type")): |
| normalized = _normalize_verdict_token(value) |
| if normalized: |
| return normalized |
| return "running" if _events(bundle) else "unknown" |
|
|
|
|
| def _final_visual_status(verdict: str) -> str: |
| if verdict in {"full_inference_success", "success", "passed"}: |
| return "success" |
| if verdict in {"partial_validation", "partial", "health_only", "completed_with_warnings"}: |
| return "warn" |
| if verdict in {"failed", "failure", "technical_blocker", "technical_blocker_boot_only"}: |
| return "error" |
| if verdict == "auth_refresh_required": |
| return "warn" |
| if verdict in {"manual_action_required", "cancelled"}: |
| return "stopped" |
| return "running" |
|
|
|
|
| def _status_label(verdict: str) -> tuple[str, str]: |
| if verdict == "full_inference_success": |
| return "Full inference success", "Space boots, generation passed, and latency was measured." |
| if verdict == "partial_validation": |
| return "Completed with partial validation", "The run finished, but full generation was not verified." |
| if verdict == "technical_blocker_boot_only": |
| return "Technical blocker", "Health may pass, but no generation endpoint exists for full inference." |
| if verdict == "technical_blocker": |
| return "Technical blocker", "The run found a technical blocker." |
| if verdict == "auth_refresh_required": |
| return "Auth refresh required", "HF OAuth expired or is too close to expiry; sign in again before retrying validation." |
| if verdict == "manual_action_required": |
| return "Manual action required", "The run needs user action before validation can continue." |
| if verdict == "failed": |
| return "Failed", "The run did not complete successfully." |
| if verdict == "cancelled": |
| return "Cancelled", "The run was cancelled." |
| if verdict in {"running", "pending", "unknown"}: |
| return "Running", "" |
| return verdict.replace("_", " ").title(), "" |
|
|
|
|
| def _hardware_warning(bundle: dict[str, Any]) -> dict[str, Any] | None: |
| events = _events_for_phase(bundle, "hardware") |
| preferred_failed = any(_lower(e.get("step")) == "create_space_hardware" and _lower(e.get("status")) in FAILED_STATUSES for e in events) |
| fallback_ok = any(_lower(e.get("step")) in {"create_space", "create_space_hardware"} and _lower(e.get("status")) in SUCCESS_STATUSES and _lower((e.get("data") or {}).get("hardware")) for e in events) |
| if preferred_failed and fallback_ok: |
| return {"code": "hardware_fallback_used", "label": "Fallback GPU used", "detail": "Preferred hardware was unavailable; the factory continued on fallback hardware."} |
| return None |
|
|
|
|
| def _pi_model_warning(bundle: dict[str, Any]) -> dict[str, Any] | None: |
| for event in _events(bundle): |
| if _lower(event.get("step")) == "pi_model_resolution" and _lower(event.get("status")) == "warning": |
| data = event.get("data") or {} |
| requested = data.get("requested_model") or data.get("configured_model") or "requested model" |
| effective = data.get("effective_model") or "observed model" |
| return {"code": "pi_model_changed", "label": "Pi assistant model changed", "detail": f"{requested} → {effective}"} |
| return None |
|
|
|
|
| def _logs_warning(bundle: dict[str, Any]) -> dict[str, Any] | None: |
| logs = bundle.get("space_logs_index") or bundle.get("space_logs") or {} |
| if not isinstance(logs, dict): |
| return None |
| quality = _lower(logs.get("log_quality") or logs.get("quality")) |
| if quality and quality not in {"full", "complete"}: |
| return {"code": "logs_partial", "label": "Space logs partial", "detail": f"Log quality: {quality}."} |
| return None |
|
|
|
|
| def _archive_warning(bundle: dict[str, Any], *, terminal: bool) -> dict[str, Any] | None: |
| status = _eval_publish(bundle) |
| if not status or status.get("published") is True: |
| return None |
| |
| |
| if _is_record_not_ready(status): |
| return None |
| if status.get("attempted"): |
| return {"code": "eval_archive_not_published", "label": "Eval archive not published", "detail": str(status.get("reason") or "Backend archive copy did not complete.")} |
| return None |
|
|
|
|
| def _build_phase_details(bundle: dict[str, Any], phase: str) -> list[dict[str, Any]]: |
| details: list[dict[str, Any]] = [] |
| if phase == "deploy": |
| blocker = _final_blocker_label(bundle) |
| if blocker and _build_error_observation(bundle): |
| details.append({"label": blocker, "status": "failed"}) |
| else: |
| latest = _latest_event_for_phase(bundle, phase) |
| if latest: |
| details.append({"label": str(latest.get("message") or latest.get("step") or PHASE_LABELS[phase]), "status": _lower(latest.get("status")) or "info"}) |
| elif phase == "live_validation": |
| live = build_live_validation_model(bundle) |
| runtime_stage = live.get("runtime_stage") or "" |
| runtime_history = live.get("runtime_history") if isinstance(live.get("runtime_history"), list) else [] |
| for item in runtime_history: |
| if isinstance(item, dict) and item.get("label"): |
| details.append({"label": str(item.get("label")), "status": str(item.get("status") or "info")}) |
| if runtime_stage and not runtime_history: |
| details.append({"label": f"Space runtime: {runtime_stage}", "status": "info"}) |
| if live.get("health") == "passed": |
| details.append({"label": "Health passed", "status": "complete"}) |
| elif live.get("health") == "failed": |
| details.append({"label": "Health not confirmed", "status": "warning"}) |
| smoke = _smoke(bundle) |
| generation_passed = live.get("generation_smoke") == "passed" |
| if live.get("gradio_schema") == "detected": |
| count = live.get("endpoint_count") or 0 |
| selected = str(live.get("selected_endpoint") or smoke.get("api_name") or "").strip() |
| if selected: |
| details.append({"label": f"Endpoint selected: {selected}", "status": "complete"}) |
| elif count: |
| details.append({"label": f"Gradio endpoints detected: {count}", "status": "complete" if generation_passed else "info"}) |
| elif not generation_passed: |
| details.append({"label": "Waiting for Gradio endpoints", "status": "info"}) |
| if generation_passed: |
| pass |
| elif live.get("generation_smoke") == "failed": |
| reason = smoke.get("failure_type") or smoke.get("error") or live.get("message") or "Generation was not verified" |
| details.append({"label": f"Generation not verified: {reason}", "status": "warning"}) |
| auto_retry = smoke.get("auto_retry") if isinstance(smoke.get("auto_retry"), dict) else {} |
| if auto_retry.get("retried"): |
| details.append({"label": "Smoke test retried once with schema-corrected choices", "status": "info"}) |
| elif smoke.get("retryable_with_schema_payload") or smoke.get("auto_retry_supported"): |
| details.append({"label": "Schema-adjusted retry available in Space Test", "status": "info"}) |
| else: |
| details.append({"label": "Generation smoke pending", "status": "info"}) |
| blocker = _final_blocker_label(bundle) |
| if blocker: |
| details.append({"label": blocker, "status": "failed"}) |
| recommendation = smoke.get("recommended_zero_gpu_duration_seconds") or (_gate(bundle).get("zero_gpu_duration_recommendation") or {}).get("recommended_zero_gpu_duration_seconds") |
| if recommendation: |
| source_hw = str(_smoke(bundle).get("hardware") or _gate(bundle).get("hardware") or (bundle.get("summary") or {}).get("selected_hardware") or "").strip() |
| suffix = f" · measured on {source_hw}" if source_hw else "" |
| details.append({"label": f"ZeroGPU estimate: {recommendation}s{suffix}", "status": "complete"}) |
| elif phase == "archive": |
| eval_status = _eval_publish(bundle) |
| if eval_status: |
| if eval_status.get("published") is True: |
| details.append({"label": "Eval archive published", "status": "complete"}) |
| if eval_status.get("archive_relative_path"): |
| details.append({"label": str(eval_status.get("archive_relative_path")), "status": "info"}) |
| elif eval_status.get("attempted"): |
| if _is_record_not_ready(eval_status): |
| details.append({"label": "Eval archive finalizing", "status": "info"}) |
| else: |
| details.append({"label": f"Eval archive not published: {eval_status.get('reason') or 'unknown'}", "status": "warning"}) |
| else: |
| details.append({"label": "Eval archive pending", "status": "pending"}) |
| elif phase == "recovery": |
| steps = _event_steps(bundle) |
| if not steps.intersection(PHASE_STEPS["recovery"]): |
| details.append({"label": "Recovery not needed", "status": "skipped"}) |
| else: |
| failed = any(_lower(e.get("step")) == "repair_validation" and _lower(e.get("status")) in FAILED_STATUSES for e in _events(bundle)) |
| if failed: |
| details.append({"label": "Patch attempted; final blocker remains", "status": "failed"}) |
| blocker = _final_blocker_label(bundle) |
| if blocker: |
| details.append({"label": blocker, "status": "failed"}) |
| elif any(_lower(e.get("step")) == "repair_validation" and _lower(e.get("status")) in SUCCESS_STATUSES for e in _events(bundle)): |
| details.append({"label": "Repair revalidated", "status": "complete"}) |
| else: |
| details.append({"label": "Recovery in progress", "status": "running"}) |
| else: |
| latest = _latest_event_for_phase(bundle, phase) |
| if latest: |
| details.append({"label": str(latest.get("message") or latest.get("step") or PHASE_LABELS[phase]), "status": _lower(latest.get("status")) or "info"}) |
| return details |
|
|
|
|
| def _phase_summary(bundle: dict[str, Any], phase: str, status: str) -> str: |
| if phase == "agent": |
| |
| |
| |
| |
| pass |
| if phase == "hardware": |
| warning = _hardware_warning(bundle) |
| if warning: |
| return warning["label"] |
| if phase == "deploy": |
| blocker = _final_blocker_label(bundle) |
| if blocker and _build_error_observation(bundle): |
| return blocker |
| if phase == "live_validation": |
| blocker = _final_blocker_label(bundle) |
| if blocker and _final_visual_status(_verdict(bundle)) == "error": |
| return "Live validation incomplete" |
| signals = _signals(bundle) |
| smoke = _smoke(bundle) |
| if signals.get("generation_smoke_passed") is True or _lower(smoke.get("status")) == "success": |
| latency = smoke.get("latency_seconds") or smoke.get("observed_latency_seconds") |
| return f"Generation passed in {latency}s" if latency is not None else "Generation passed" |
| if signals.get("health_passed") is True: |
| return "Health passed; generation not verified" |
| live = build_live_validation_model(bundle) |
| return str(live.get("label") or ("Live validation pending" if status in {"pending", "running"} else "Live validation incomplete")) |
| if phase == "recovery": |
| steps = _event_steps(bundle) |
| if not steps.intersection(PHASE_STEPS["recovery"]): |
| return "Not needed" |
| if any(_lower(e.get("step")) == "repair_validation" and _lower(e.get("status")) in FAILED_STATUSES for e in _events(bundle)): |
| return "Patch attempted; final blocker remains" |
| if any(_lower(e.get("step")) == "repair_validation" and _lower(e.get("status")) in SUCCESS_STATUSES for e in _events(bundle)): |
| return "Repair revalidated" |
| latest = _latest_recovery_event(bundle) or {} |
| latest_step = _lower(latest.get("step")) |
| if latest_step in {"repair_validation"}: |
| return "Revalidating repair" |
| if latest_step in {"repair", "repair_plan", "repair_patch", "repair_upload"}: |
| return "Repairing" |
| return "Diagnosing repair" |
| if phase == "archive": |
| eval_status = _eval_publish(bundle) |
| if eval_status.get("published") is True: |
| return "Eval archive published" |
| if eval_status.get("attempted") and not _is_record_not_ready(eval_status): |
| return "Eval archive attempted" |
| if eval_status.get("attempted") and _is_record_not_ready(eval_status): |
| return "Finalizing archive" if status in {"pending", "running"} else "Final artifacts written" |
| return "Archive pending" if status in {"pending", "running"} else "Final artifacts written" |
| if phase == "done": |
| label, _ = _status_label(_verdict(bundle)) |
| return label |
| latest = _latest_event_for_phase(bundle, phase) |
| if latest: |
| return str(latest.get("message") or PHASE_LABELS[phase]) |
| if status == "skipped": |
| return "Not needed" |
| return "Pending" if status == "pending" else PHASE_LABELS[phase] |
|
|
|
|
| def _phase_status(bundle: dict[str, Any], phase: str, verdict: str) -> str: |
| steps = _event_steps(bundle) |
| terminal = _final_visual_status(verdict) in {"success", "warn", "error", "stopped"} |
|
|
| if phase == "done": |
| if verdict in {"full_inference_success", "success", "passed"}: |
| return "complete" |
| if verdict in {"partial_validation", "partial", "completed_with_warnings", "health_only"}: |
| return "warning" |
| if verdict in {"failed", "technical_blocker", "technical_blocker_boot_only"}: |
| return "failed" |
| if verdict in {"manual_action_required", "cancelled"}: |
| return "stopped" |
| return "pending" |
|
|
| if phase == "recovery": |
| if not steps.intersection(PHASE_STEPS["recovery"]): |
| return "skipped" if terminal else "pending" |
| if _final_visual_status(verdict) == "success": |
| return "complete" |
| if any(_lower(e.get("step")) == "repair_validation" and _lower(e.get("status")) in FAILED_STATUSES for e in _events(bundle)): |
| return "failed" |
| if any(_lower(e.get("step")) == "repair_validation" and _lower(e.get("status")) in SUCCESS_STATUSES for e in _events(bundle)): |
| return "complete" |
| latest_recovery = _latest_recovery_event(bundle) or {} |
| latest_status = _lower(latest_recovery.get("status")) |
| if latest_status in FAILED_STATUSES: |
| return "failed" |
| return "warning" if terminal else "running" |
|
|
| if phase == "agent": |
| latest_by_step = _latest_events_by_step(bundle, phase) |
| pi_run = latest_by_step.get("pi_run") or {} |
| |
| |
| |
| if terminal and _final_visual_status(verdict) == "success" and _has_downstream_proof(bundle, phase): |
| return "complete" |
| if _lower(pi_run.get("status")) in RUNNING_STATUSES and not terminal: |
| return "running" |
| if terminal: |
| material_events = [e for e in _events_for_phase(bundle, phase) if _lower(e.get("step")) != "pi_model_resolution"] |
| material_statuses = {_lower(e.get("status")) for e in material_events} |
| if material_statuses.intersection(FAILED_STATUSES) and _final_visual_status(verdict) != "success": |
| return "failed" |
| if material_statuses.intersection(RUNNING_STATUSES) and not _has_downstream_proof(bundle, phase): |
| return "running" |
| if material_statuses.intersection(SUCCESS_STATUSES) or _has_downstream_proof(bundle, phase): |
| return "complete" |
|
|
| if phase == "deploy": |
| blocker = _final_blocker_label(bundle) |
| if blocker and _build_error_observation(bundle): |
| return "failed" |
| if phase == "live_validation": |
| blocker = _final_blocker_label(bundle) |
| if blocker and _final_visual_status(_verdict(bundle)) == "error": |
| return "failed" |
| signals = _signals(bundle) |
| smoke = _smoke(bundle) |
| if _final_visual_status(verdict) == "success": |
| return "complete" |
| if signals.get("generation_smoke_passed") is True or _lower(smoke.get("status")) == "success": |
| return "complete" |
| if signals.get("health_passed") is True or smoke: |
| return "warning" |
| return _phase_status_from_events(bundle, phase) |
|
|
| if phase == "archive": |
| eval_status = _eval_publish(bundle) |
| if eval_status.get("published") is True: |
| return "complete" |
| if eval_status.get("attempted"): |
| if _is_record_not_ready(eval_status): |
| return "pending" if not terminal else "complete" |
| return "warning" |
| if terminal: |
| return "complete" |
| return "pending" |
|
|
| if phase == "deploy" and _build_error_observation(bundle): |
| return "failed" |
|
|
| |
| if phase == "hardware" and _hardware_warning(bundle): |
| return "warning" |
|
|
| event_status = _phase_status_from_events(bundle, phase) |
| if event_status != "pending": |
| if event_status == "running" and terminal and (_has_downstream_proof(bundle, phase) or _has_successful_phase_event(bundle, phase)): |
| return "complete" |
| return event_status |
|
|
| |
| |
| if terminal and _has_downstream_proof(bundle, phase): |
| return "complete" |
| return "pending" |
|
|
|
|
| def build_run_timeline_model(bundle: dict[str, Any]) -> dict[str, Any]: |
| """Return the canonical product progress/timeline projection for a run. |
| |
| This model is deliberately derived from persisted run artifacts and events, |
| not from transient DOM state. It separates process completion from the final |
| result verdict so a completed run can be success, partial, failed, blocked, |
| or manual-action-required without the progress bar lying. |
| """ |
| bundle = bundle or {} |
| state = bundle.get("state") or {} |
| progress = progress_from_events(_events(bundle), state=state) |
| verdict = _verdict(bundle) |
| visual_status = _final_visual_status(verdict) |
| label, subtitle = _status_label(verdict) |
| terminal = visual_status in {"success", "warn", "error", "stopped"} |
| percent = 100 if terminal else int(progress.get("progress") or 0) |
|
|
| warnings = [w for w in (_pi_model_warning(bundle), _hardware_warning(bundle), _logs_warning(bundle), _archive_warning(bundle, terminal=terminal)) if w] |
| phases: list[dict[str, Any]] = [] |
| for phase in PHASE_ORDER: |
| status = _phase_status(bundle, phase, verdict) |
| phases.append( |
| { |
| "id": phase, |
| "label": PHASE_LABELS[phase], |
| "status": status, |
| "summary": _phase_summary(bundle, phase, status), |
| "details": _build_phase_details(bundle, phase), |
| } |
| ) |
|
|
| return { |
| "schema_version": "run_timeline_model.v1", |
| "progress": { |
| "percent": max(0, min(100, percent)), |
| "visual_status": visual_status, |
| "label": label, |
| "subtitle": subtitle, |
| "process_status": progress.get("status") or "unknown", |
| "verdict": verdict, |
| "terminal": terminal, |
| }, |
| "phases": phases, |
| "warnings": warnings, |
| "live_validation": build_live_validation_model(bundle), |
| } |
|
|