Upload 6 files

app.py

@@ -113,6 +113,68 @@ def _api_links(*, run_id: str | None, bucket_source: str | None, target_space: s
         "artifacts_url": _run_artifacts_url(run_id, bucket_source),
     }
 
+
+def _job_id_from_run_bundle(summary: dict[str, Any], launch: dict[str, Any], state: dict[str, Any]) -> str:
+    return str(summary.get("job_id") or launch.get("job_id") or state.get("job_id") or "").strip()
+
+
+def _normalize_job_stage(stage: Any) -> str:
+    value = str(stage or "").strip().lower()
+    if "." in value:
+        value = value.rsplit(".", 1)[-1]
+    value = value.replace("jobstage.", "").replace("_", "-")
+    if value in {"running", "queued", "pending", "scheduled", "starting"}:
+        return "running"
+    if value in {"success", "succeeded", "complete", "completed", "done"}:
+        return "success"
+    if value in {"failed", "failure", "error"}:
+        return "failed"
+    if value in {"cancelled", "canceled", "canceling", "cancelling"}:
+        return "cancelled"
+    return value
+
+
+def _event_key(event: dict[str, Any]) -> tuple[str, str, str, str]:
+    return (
+        str(event.get("ts") or ""),
+        str(event.get("step") or ""),
+        str(event.get("status") or ""),
+        str(event.get("message") or ""),
+    )
+
+
+def _events_from_job_logs(log_text: str) -> list[dict[str, Any]]:
+    """Extract worker append_event JSON lines from HF Job logs."""
+    events: list[dict[str, Any]] = []
+    for raw in (log_text or "").splitlines():
+        line = raw.strip()
+        start = line.find("{")
+        if start > 0:
+            line = line[start:]
+        if not line.startswith("{"):
+            continue
+        try:
+            payload = json.loads(line)
+        except Exception:
+            continue
+        if isinstance(payload, dict) and payload.get("step") and payload.get("status"):
+            events.append(payload)
+    return events
+
+
+def _merge_events(bucket_events: list[dict[str, Any]], log_events: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    seen: set[tuple[str, str, str, str]] = set()
+    merged: list[dict[str, Any]] = []
+    for event in [*(bucket_events or []), *(log_events or [])]:
+        if not isinstance(event, dict):
+            continue
+        key = _event_key(event)
+        if key in seen:
+            continue
+        seen.add(key)
+        merged.append(event)
+    return merged
+
 def register_custom_routes(fastapi_app: FastAPI) -> None:
     """Register the root custom UI and OAuth-backed JSON endpoints."""
 
@@ -151,7 +213,7 @@ def register_custom_routes(fastapi_app: FastAPI) -> None:
         return JSONResponse(
             {
                 "name": "Agentic Space Factory",
-                "version": "v58-stable-refresh-pi-stdout-bucket-links",
+                "version": "v60-cancel-persist-delete-run",
                 "bucket_default": settings.bucket_name,
                 "workflows": ["build_from_model_card", "validate_existing_space", "runs_explorer"],
                 "custom_ui_status": "root_custom_ui",
@@ -365,14 +427,66 @@ def register_custom_routes(fastapi_app: FastAPI) -> None:
         if not result.get("ok"):
             raise HTTPException(status_code=400, detail=redact(str(result.get("error") or "Could not cancel Job.")))
         paths = RunPaths(run_id, bucket_source=bucket_source)
-        cancelled_state = {**launch, **state, "run_id": run_id, "status": "cancelled", "job_id": str(job_id), "job_url": summary.get("job_url") or launch.get("job_url") or state.get("job_url"), "updated_at": result.get("updated_at") or ""}
-        cancelled_summary = {**summary, "run_id": run_id, "status": "cancelled", "job_id": str(job_id), "job_url": cancelled_state.get("job_url") or summary.get("job_url") or "", "updated_at": cancelled_state.get("updated_at") or summary.get("updated_at") or ""}
+        cancelled_at = utc_now_iso()
+        job_url = summary.get("job_url") or launch.get("job_url") or state.get("job_url")
+        cancel_meta = {
+            "requested_by": ctx["username"],
+            "cancelled_at": cancelled_at,
+            "job_id": str(job_id),
+            "job_url": job_url or "",
+            "hf_cancel_result": result,
+        }
+        cancelled_state = {
+            **launch,
+            **state,
+            "run_id": run_id,
+            "status": "cancelled",
+            "job_id": str(job_id),
+            "job_url": job_url,
+            "cancel_requested": True,
+            "cancelled_at": cancelled_at,
+            "updated_at": cancelled_at,
+            "cancel": cancel_meta,
+        }
+        cancelled_summary = {
+            **summary,
+            "run_id": run_id,
+            "status": "cancelled",
+            "job_id": str(job_id),
+            "job_url": job_url or summary.get("job_url") or "",
+            "cancel_requested": True,
+            "cancelled_at": cancelled_at,
+            "updated_at": cancelled_at,
+        }
         try:
             write_json(paths.state, cancelled_state, token=ctx["token"])
             write_json(f"{paths.root}/summary.json", cancelled_summary, token=ctx["token"])
+            write_json(f"{paths.root}/cancel.json", cancel_meta, token=ctx["token"])
+            append_run_event(
+                run_id,
+                bucket_source=bucket_source,
+                step="cancel",
+                status="cancelled",
+                message="Job cancellation requested from Agentic Space Factory UI",
+                details=cancel_meta,
+                token=ctx["token"],
+            )
         except Exception:
             pass
-        return JSONResponse({"ok": True, "run_id": run_id, "job_id": str(job_id), "status": "cancelled"})
+        return JSONResponse({"ok": True, "run_id": run_id, "job_id": str(job_id), "status": "cancelled", "cancelled_at": cancelled_at})
+
+    @fastapi_app.delete("/api/runs/{run_id}")
+    async def api_delete_run(request: Request, run_id: str, bucket_name: str = settings.bucket_name):  # type: ignore[no-untyped-def]
+        ctx = _oauth_context_from_request(request)
+        run_id = validate_run_id(run_id)
+        bucket_source = user_bucket_source(username=ctx["username"], bucket_name=bucket_name)
+        try:
+            delete_run_folder(run_id, bucket_source=bucket_source, token=ctx["token"])
+        except FileNotFoundError:
+            pass
+        except Exception as exc:  # noqa: BLE001
+            raise HTTPException(status_code=400, detail=redact(str(exc))) from exc
+        return JSONResponse({"ok": True, "run_id": run_id, "deleted": True})
 
     @fastapi_app.get("/api/runs/{run_id}/progress")
     async def api_run_progress(request: Request, run_id: str, bucket_name: str = settings.bucket_name):  # type: ignore[no-untyped-def]
@@ -382,22 +496,44 @@ def register_custom_routes(fastapi_app: FastAPI) -> None:
         bundle = read_run_bundle(run_id, bucket_source=bucket_source, token=ctx["token"])
         state = bundle.get("state") or {}
         launch = bundle.get("launch") or {}
+        summary = bundle.get("summary") or {}
+        job_id = _job_id_from_run_bundle(summary, launch, state)
+        job_info: dict[str, Any] = {}
+        job_logs = ""
+        job_log_events: list[dict[str, Any]] = []
+        if job_id:
+            job_info = inspect_job_safe(job_id, token=ctx["token"])
+            job_logs = fetch_recent_logs_safe(job_id, token=ctx["token"], max_lines=220)
+            job_log_events = _events_from_job_logs(job_logs)
+        job_stage = _normalize_job_stage(job_info.get("stage"))
         effective_state = {**launch, **state}
-        events = bundle.get("events") or []
+        if job_stage and not job_info.get("error"):
+            effective_state["job_stage"] = job_stage
+            # Use Job stage as live source of truth while the bucket lags.
+            if job_stage in {"running", "failed", "cancelled"}:
+                effective_state["status"] = job_stage
+            elif job_stage == "success" and not (bundle.get("events") or []):
+                effective_state["status"] = "success"
+        events = _merge_events(bundle.get("events") or [], job_log_events)
         progress = progress_from_events(events, state=effective_state)
-        summary = bundle.get("summary") or {}
-        view = build_run_view_model(run_id, bundle, bucket_source=bucket_source)
+        view_bundle = {**bundle, "events": events}
+        view = build_run_view_model(run_id, view_bundle, bucket_source=bucket_source)
+        pi_stdout = bundle.get("pi_live_log") or bundle.get("pi_output_log") or ""
         progress.update(
             {
                 "run_id": run_id,
                 "bucket_source": bucket_source,
                 "state": effective_state,
                 "summary": summary,
+                "job_id": job_id,
+                "job_info": job_info,
+                "job_logs": job_logs[-20000:],
+                "job_log_events_count": len(job_log_events),
                 "inference_gate": bundle.get("inference_gate") or {},
                 "generation_smoke": bundle.get("generation_smoke") or {},
                 "hardware_strategy": bundle.get("hardware_strategy") or {},
                 "technical_blockers": bundle.get("technical_blockers") or {},
-                "pi_live_log": (bundle.get("pi_live_log") or bundle.get("pi_output_log") or "")[-12000:],
+                "pi_live_log": pi_stdout[-12000:],
                 "pi_live_log_source": "logs/pi_live_output.txt" if bundle.get("pi_live_log") else ("logs/pi_output.txt" if bundle.get("pi_output_log") else ""),
                 "events": events[-20:],
                 "view": view,