Upload 9 files

src/jobs.py

@@ -3,7 +3,7 @@ from __future__ import annotations
 import re
 from typing import Any
 
-from huggingface_hub import Volume, fetch_job_logs, inspect_job, run_job
+from huggingface_hub import Volume, cancel_job, fetch_job_logs, inspect_job, run_job
 
 from .config import bucket_uri_from_source, user_bucket_source, settings
 from .bucket import assert_user_bucket_ready
@@ -241,3 +241,14 @@ def fetch_recent_logs_safe(job_id: str, token: str | None = None, max_lines: int
         return "\n".join(str(line).rstrip("\n") for line in logs[-max_lines:])
     except Exception as exc:  # noqa: BLE001
         return f"Could not fetch job logs: {exc}"
+
+
+def cancel_job_safe(job_id: str, *, namespace: str | None = None, token: str | None = None) -> dict[str, Any]:
+    """Cancel a running HF Job. Returns a JSON-safe status payload."""
+    if not job_id:
+        return {"ok": False, "error": "Missing job_id"}
+    try:
+        cancel_job(job_id=job_id, namespace=namespace, token=token)
+        return {"ok": True, "job_id": job_id, "status": "cancelled"}
+    except Exception as exc:  # noqa: BLE001
+        return {"ok": False, "job_id": job_id, "error": str(exc)}

src/progress.py

@@ -1,12 +1,12 @@
 from __future__ import annotations
 
-from dataclasses import dataclass
 from datetime import datetime, timezone
 from typing import Any
 
+# Exact worker-facing steps emitted by src/worker_payload.py.
+# Keep this list close to the Job logs/events rather than a marketing pipeline.
 STEP_ORDER = [
-    "bucket_ready",
-    "job_launched",
+    "bootstrap",
     "dependencies",
     "auth",
     "model_analysis",
@@ -15,80 +15,80 @@ STEP_ORDER = [
     "pi_install",
     "pi_config",
     "pi_run",
+    "pi_verification",
+    "metadata_sanitize",
+    "requirements_sanitize",
+    "hardware_strategy",
+    "create_space_hardware",
     "create_space",
+    "repair",
     "upload_files",
-    "hardware",
+    "space_runtime",
+    "space_logs",
     "api_validation",
+    "live_wait",
+    "generation_smoke",
     "inference_gate",
     "report_write",
     "done",
+    "failure",
 ]
 
 STEP_LABELS = {
-    "bucket_ready": "Bucket ready",
-    "job_launched": "Job launched",
+    "bootstrap": "Bootstrap",
     "dependencies": "Dependencies",
-    "auth": "Authenticated",
+    "auth": "Auth",
     "model_analysis": "Model analysis",
     "workspace": "Workspace",
     "node": "Node/npm",
-    "pi_install": "Pi installed",
-    "pi_config": "Pi configured",
-    "pi_run": "Pi running",
+    "pi_install": "Pi install",
+    "pi_config": "Pi config",
+    "pi_run": "Pi run",
+    "pi_verification": "Pi verification",
+    "metadata_sanitize": "Metadata sanitize",
+    "requirements_sanitize": "Requirements sanitize",
+    "hardware_strategy": "Hardware strategy",
+    "create_space_hardware": "Create with hardware",
     "create_space": "Create Space",
+    "repair": "Repair pass",
     "upload_files": "Upload files",
-    "hardware": "Hardware request",
+    "space_runtime": "Space runtime",
+    "space_logs": "Space logs",
     "api_validation": "API validation",
-    "inference_gate": "Gate",
+    "live_wait": "Live wait",
+    "generation_smoke": "Generation smoke",
+    "inference_gate": "Inference gate",
     "report_write": "Report",
     "done": "Done",
-}
-
-STEP_PROGRESS = {
-    "bucket_ready": 5,
-    "job_launched": 8,
-    "dependencies": 12,
-    "auth": 16,
-    "model_analysis": 22,
-    "workspace": 28,
-    "node": 34,
-    "pi_install": 40,
-    "pi_config": 44,
-    "pi_run": 62,
-    "create_space": 72,
-    "upload_files": 80,
-    "hardware": 86,
-    "api_validation": 92,
-    "inference_gate": 96,
-    "report_write": 98,
-    "done": 100,
+    "failure": "Failure",
 }
 
 STEP_ALIASES = {
-    "bootstrap": "job_launched",
-    "dependencies": "dependencies",
-    "auth": "auth",
-    "model_analysis": "model_analysis",
-    "workspace": "workspace",
-    "node": "node",
-    "pi_install": "pi_install",
-    "pi_config": "pi_config",
-    "pi_run": "pi_run",
-    "create_space": "create_space",
-    "upload_files": "upload_files",
-    "hardware_preferred": "hardware",
-    "hardware_fallback": "hardware",
-    "hardware": "hardware",
-    "api_validation": "api_validation",
-    "inference_gate": "inference_gate",
-    "report_write": "report_write",
-    "done": "done",
-    "failure": "done",
+    "bucket_ready": "bootstrap",
+    "job_launched": "bootstrap",
+    "hardware_preferred": "hardware_strategy",
+    "hardware_fallback": "hardware_strategy",
+    "hardware": "hardware_strategy",
+    "build": "space_runtime",
+    "diagnose": "repair",
+    "diagnosis": "repair",
+    "patch": "repair",
 }
 
-DONE_STATUSES = {"success", "done", "completed", "passed", "full_inference_success", "full_inference_candidate_health_passed", "manual_hardware_required", "technical_blocker"}
-RUNNING_STATUSES = {"started", "running", "waiting"}
-FAILED_STATUSES = {"failed", "error"}
+DONE_STATUSES = {
+    "success",
+    "done",
+    "completed",
+    "passed",
+    "full_inference_success",
+    "full_inference_candidate_health_passed",
+    "manual_hardware_required",
+    "technical_blocker",
+    "health_only",
+}
+RUNNING_STATUSES = {"started", "running", "waiting", "pending", "scheduled"}
+FAILED_STATUSES = {"failed", "error", "failure"}
+CANCELLED_STATUSES = {"cancelled", "canceled"}
 
 
 def _parse_ts(ts: str | None) -> datetime | None:
@@ -103,19 +103,38 @@ def _parse_ts(ts: str | None) -> datetime | None:
 def _canonical_step(step: str | None) -> str | None:
     if not step:
         return None
-    return STEP_ALIASES.get(step, step if step in STEP_PROGRESS else None)
+    value = str(step)
+    return STEP_ALIASES.get(value, value if value in STEP_ORDER else None)
+
+
+def _is_terminal_status(status: str | None) -> bool:
+    s = str(status or "").lower()
+    return s in DONE_STATUSES or s in FAILED_STATUSES or s in CANCELLED_STATUSES
+
+
+def _progress_for_index(index: int) -> int:
+    if index <= 0:
+        return 3
+    if len(STEP_ORDER) <= 1:
+        return 0
+    return int(round((index / (len(STEP_ORDER) - 1)) * 100))
 
 
 def progress_from_events(events: list[dict[str, Any]] | None, *, state: dict[str, Any] | None = None) -> dict[str, Any]:
-    """Build a stable UI progress model from worker events.jsonl + optional state.json."""
+    """Build a live progress model from the actual worker events.
+
+    Unlike earlier product pipelines, this keeps the worker's real step names.
+    Missing steps stay pending; they are not automatically marked done.
+    """
     events = events or []
     state = state or {}
     step_status = {step: "pending" for step in STEP_ORDER}
-    last_event = None
-    current_step = "job_launched"
+    last_event: dict[str, Any] | None = None
+    current_step = "bootstrap"
     terminal_status = None
     first_ts = None
     last_ts = None
+    observed_steps: list[str] = []
 
     for event in events:
         if not isinstance(event, dict):
@@ -130,43 +149,57 @@ def progress_from_events(events: list[dict[str, Any]] | None, *, state: dict[str
             last_ts = ts
         if not step:
             continue
+
         current_step = step
+        if step not in observed_steps:
+            observed_steps.append(step)
+
         if status in FAILED_STATUSES:
             step_status[step] = "failed"
             terminal_status = "failed"
+        elif status in CANCELLED_STATUSES:
+            step_status[step] = "blocked"
+            terminal_status = "cancelled"
         elif status in RUNNING_STATUSES:
-            if step_status.get(step) != "done":
+            if step_status.get(step) not in {"done", "failed"}:
                 step_status[step] = "running"
         elif status in DONE_STATUSES or status:
             step_status[step] = "done"
 
-        if step == "done" and status:
+        if step in {"done", "failure"} and status:
             terminal_status = status
 
-    # Mark all previous steps as done up to the current running/done step.
-    current_index = STEP_ORDER.index(current_step) if current_step in STEP_ORDER else 0
-    for step in STEP_ORDER[:current_index]:
-        if step_status[step] == "pending":
-            step_status[step] = "done"
+    status_from_state = str(state.get("status") or state.get("gate_status") or "").lower()
+    if status_from_state in CANCELLED_STATUSES:
+        terminal_status = "cancelled"
+    elif status_from_state in FAILED_STATUSES:
+        terminal_status = "failed"
+    elif status_from_state in DONE_STATUSES:
+        terminal_status = status_from_state
 
     if terminal_status and terminal_status != "failed":
-        current_step = "done"
-        for step in STEP_ORDER:
-            if step_status[step] != "failed":
-                step_status[step] = "done"
+        if terminal_status in CANCELLED_STATUSES:
+            current_step = current_step or "failure"
+        elif current_step != "failure":
+            current_step = "done"
+
+    if observed_steps:
+        current_index = max(STEP_ORDER.index(s) for s in observed_steps if s in STEP_ORDER)
+    else:
+        current_index = 0
+    if current_step in STEP_ORDER:
+        current_index = max(current_index, STEP_ORDER.index(current_step))
 
-    status_from_state = state.get("status") or state.get("gate_status")
     overall_status = terminal_status or status_from_state or ("running" if events else "not_started")
 
-    if overall_status in {"failed", "error"}:
-        progress = max(STEP_PROGRESS.get(current_step, 8), 8)
+    if overall_status in FAILED_STATUSES:
+        progress = max(_progress_for_index(current_index), 3)
+    elif overall_status in CANCELLED_STATUSES:
+        progress = max(_progress_for_index(current_index), 3)
     elif current_step == "done" or overall_status in DONE_STATUSES:
         progress = 100
     else:
-        progress = STEP_PROGRESS.get(current_step, 8)
-        if step_status.get(current_step) == "running":
-            previous = STEP_ORDER[max(0, current_index - 1)] if current_index > 0 else current_step
-            progress = max(STEP_PROGRESS.get(previous, 0) + 2, progress - 8)
+        progress = max(3, min(99, _progress_for_index(current_index)))
 
     now = datetime.now(timezone.utc)
     elapsed = int(((last_ts or now) - first_ts).total_seconds()) if first_ts else 0

src/view_models.py

@@ -3,17 +3,9 @@ from __future__ import annotations
 from datetime import datetime, timedelta, timezone
 from typing import Any
 
-PRODUCT_STEPS = [
-    {"id": "bucket_ready", "label": "Bucket ready"},
-    {"id": "job_launched", "label": "Job launched"},
-    {"id": "model_analysis", "label": "Model analysis"},
-    {"id": "generate_app", "label": "Generate app"},
-    {"id": "create_space", "label": "Create Space"},
-    {"id": "upload_files", "label": "Upload files"},
-    {"id": "hardware", "label": "Hardware"},
-    {"id": "live_validation", "label": "Live validation"},
-    {"id": "report", "label": "Report"},
-]
+from .progress import STEP_ALIASES, STEP_LABELS, STEP_ORDER
+
+PRODUCT_STEPS = [{"id": step, "label": STEP_LABELS[step]} for step in STEP_ORDER]
 
 TERMINAL_GLOBAL_STATUSES = {"succeeded", "failed", "cancelled", "blocked", "waiting_manual_action"}
 SUCCESS_RAW_STATUSES = {
@@ -32,40 +24,11 @@ MANUAL_RAW_STATUSES = {
     "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 = {
-    "bucket_ready": "bucket_ready",
-    "job_launched": "job_launched",
-    "bootstrap": "job_launched",
-    "dependencies": "job_launched",
-    "auth": "job_launched",
-    "model_analysis": "model_analysis",
-    "workspace": "generate_app",
-    "node": "generate_app",
-    "pi_install": "generate_app",
-    "pi_config": "generate_app",
-    "pi_run": "generate_app",
-    "generate_app": "generate_app",
-    "create_space": "create_space",
-    "upload_files": "upload_files",
-    "hardware_preferred": "hardware",
-    "hardware_fallback": "hardware",
-    "hardware": "hardware",
-    "build": "hardware",
-    "space_build": "hardware",
-    "diagnose": "hardware",
-    "diagnosis": "hardware",
-    "repair": "hardware",
-    "patch": "hardware",
-    "api_validation": "live_validation",
-    "generation_smoke": "live_validation",
-    "inference_gate": "live_validation",
-    "report_write": "report",
-    "done": "report",
-    "failure": "report",
-}
+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:
@@ -199,7 +162,10 @@ def normalize_run_status(
     manual = requires_manual_action(bundle)
     blocker = has_technical_blocker(bundle)
 
-    if manual:
+    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):
@@ -246,10 +212,12 @@ def normalize_run_status(
 
 
 def derive_product_phase(bundle: dict[str, Any], status_model: dict[str, Any]) -> str:
-    if status_model["global_status"] in {"succeeded", "failed", "blocked", "stale"}:
-        return "report"
+    if status_model["global_status"] == "succeeded":
+        return "done"
+    if status_model["global_status"] in {"failed", "blocked", "stale", "cancelled"}:
+        return "failure"
     if status_model["requires_manual_action"]:
-        return "hardware"
+        return "inference_gate"
 
     events = [e for e in (bundle.get("events") or []) if isinstance(e, dict)]
     for event in reversed(events):
@@ -257,19 +225,19 @@ def derive_product_phase(bundle: dict[str, Any], status_model: dict[str, Any]) -
         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"
+            return "hardware_strategy"
         if "missing" in message or "error" in message or status in FAILED_RAW_STATUSES:
-            return "hardware"
+            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"
+        return "hardware_strategy"
     if bundle.get("launch"):
-        return "job_launched"
-    return "job_launched"
+        return "bootstrap"
+    return "bootstrap"
 
 
 def build_pipeline(phase: str, status_model: dict[str, Any]) -> list[dict[str, Any]]:
@@ -379,7 +347,7 @@ def build_diagnostics(bundle: dict[str, Any], status_model: dict[str, Any], phas
 
     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 {"live_validation", "report"} or status_model["global_status"] == "succeeded"
+    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"),