Upload 8 files

src/bucket.py

@@ -133,11 +133,56 @@ def write_json(path: str, payload: dict[str, Any], token: str | None = None) ->
 
 
 def write_launch_metadata(run_id: str, *, bucket_source: str, payload: dict[str, Any], token: str | None = None) -> None:
+    """Persist enough metadata immediately for the Run Explorer.
+
+    Workers can take time before writing state.json/events.jsonl. The custom UI
+    should still be able to show a just-launched run, open its Job, and resume
+    polling after a page refresh. Store launch.json plus a lightweight
+    summary.json and minimal state.json fallback. The worker may overwrite
+    state.json later with richer information.
+    """
     paths = RunPaths(run_id, bucket_source=bucket_source)
     launch = dict(payload)
     launch.setdefault("run_id", run_id)
     launch.setdefault("bucket_source", bucket_source)
+    launch.setdefault("status", "running")
     write_json(f"{paths.root}/launch.json", launch, token=token)
+    summary = {
+        "run_id": run_id,
+        "bucket_source": bucket_source,
+        "kind": launch.get("kind") or "unknown",
+        "status": launch.get("status") or "running",
+        "model_id": launch.get("model_id") or "",
+        "target_space": launch.get("target_space") or "",
+        "target_space_url": launch.get("target_space_url") or (f"https://huggingface.co/spaces/{launch.get('target_space')}" if launch.get("target_space") else ""),
+        "job_id": launch.get("job_id") or "",
+        "job_url": launch.get("job_url") or "",
+        "created_by": launch.get("created_by") or launch.get("username") or "",
+        "created_at": launch.get("created_at") or "",
+        "updated_at": launch.get("updated_at") or launch.get("created_at") or "",
+        "selected_hardware": launch.get("preferred_space_hardware") or launch.get("selected_hardware") or "",
+        "artifacts_url": f"https://huggingface.co/buckets/{bucket_source}/tree/main/runs/{run_id}",
+    }
+    write_json(f"{paths.root}/summary.json", summary, token=token)
+    # Minimal state fallback: progress polling can treat it as running until
+    # the worker writes a definitive state.json.
+    write_json(
+        f"{paths.root}/state.json",
+        {
+            "run_id": run_id,
+            "kind": summary["kind"],
+            "status": summary["status"],
+            "model_id": summary["model_id"],
+            "target_space": summary["target_space"],
+            "target_space_url": summary["target_space_url"],
+            "job_id": summary["job_id"],
+            "job_url": summary["job_url"],
+            "created_by": summary["created_by"],
+            "created_at": summary["created_at"],
+            "updated_at": summary["updated_at"],
+        },
+        token=token,
+    )
 
 
 def read_events(run_id: str, *, bucket_source: str, token: str | None = None) -> list[dict[str, Any]]:
@@ -234,25 +279,26 @@ def _list_run_files(
 
 
 def summarize_run_bundle(run_id: str, bundle: dict[str, Any], *, bucket_source: str) -> dict[str, Any]:
+    summary_file = bundle.get("summary_file") 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 {}
-    target_space = state.get("target_space") or launch.get("target_space") or ""
-    status = state.get("status") or launch.get("status") or gate.get("status") or smoke.get("status") or "unknown"
+    target_space = state.get("target_space") or launch.get("target_space") or summary_file.get("target_space") or ""
+    status = state.get("status") or launch.get("status") or summary_file.get("status") or gate.get("status") or smoke.get("status") or "unknown"
     return {
         "run_id": run_id,
-        "kind": state.get("kind") or launch.get("kind") or "unknown",
+        "kind": state.get("kind") or launch.get("kind") or summary_file.get("kind") or "unknown",
         "status": status,
-        "model_id": state.get("model_id") or launch.get("model_id") or state.get("model") or "",
+        "model_id": state.get("model_id") or launch.get("model_id") or summary_file.get("model_id") or state.get("model") or "",
         "target_space": target_space,
-        "target_space_url": state.get("target_space_url") or launch.get("target_space_url") or (f"https://huggingface.co/spaces/{target_space}" if target_space else ""),
-        "job_id": _job_id_from_launch_or_state(launch=launch, state=state),
-        "job_url": _job_url_from_launch_or_state(launch=launch, state=state),
-        "created_at": state.get("created_at") or launch.get("created_at") or "",
-        "updated_at": state.get("updated_at") or state.get("created_at") or launch.get("updated_at") or launch.get("created_at") or "",
-        "selected_hardware": hardware.get("selected_hardware") or state.get("selected_hardware") or launch.get("preferred_space_hardware") or state.get("hardware") or "",
+        "target_space_url": state.get("target_space_url") or launch.get("target_space_url") or summary_file.get("target_space_url") or (f"https://huggingface.co/spaces/{target_space}" if target_space else ""),
+        "job_id": state.get("job_id") or launch.get("job_id") or summary_file.get("job_id") or "",
+        "job_url": state.get("job_url") or launch.get("job_url") or summary_file.get("job_url") or _job_url_from_launch_or_state(launch={**summary_file, **launch}, state=state),
+        "created_at": state.get("created_at") or launch.get("created_at") or summary_file.get("created_at") or "",
+        "updated_at": state.get("updated_at") or state.get("created_at") or launch.get("updated_at") or launch.get("created_at") or summary_file.get("updated_at") or summary_file.get("created_at") or "",
+        "selected_hardware": hardware.get("selected_hardware") or state.get("selected_hardware") or launch.get("preferred_space_hardware") or summary_file.get("selected_hardware") or state.get("hardware") or "",
         "manual_hardware_required": bool(gate.get("manual_hardware_required") or hardware.get("manual_action_required") or launch.get("manual_hardware_required")),
         "health_passed": bool((gate.get("implementation_signals") or {}).get("health_passed") or smoke.get("health_passed")),
         "smoke_test_passed": bool(smoke.get("ok") or smoke.get("status") == "success"),
@@ -271,6 +317,7 @@ def read_run_bundle(run_id: str, *, bucket_source: str, token: str | None = None
             "events": paths.events,
             "report": paths.report,
         },
+        "summary_file": _safe_read_json(f"{paths.root}/summary.json", token=token),
         "launch": _safe_read_json(f"{paths.root}/launch.json", token=token),
         "state": read_json(paths.state, token=token) or {},
         "events": read_events(run_id, bucket_source=bucket_source, token=token),
@@ -292,6 +339,45 @@ def _run_id_from_path(path: str) -> str:
     return path.rstrip('/').split('/')[-1]
 
 
+def _discover_run_ids(root: str, *, token: str | None = None, limit: int = 300) -> list[str]:
+    """Discover run ids in a bucket even when the run folder is partial.
+
+    HfFileSystem may expose object-store prefixes slightly differently between
+    Buckets/runtime versions. Use ls plus a few targeted globs so running Jobs
+    with only launch.json/summary.json are still shown in the Run Explorer.
+    """
+    fs = _fs(token)
+    run_ids: set[str] = set()
+
+    def add_from_path(path: str) -> None:
+        if "/runs/" in path:
+            tail = path.split("/runs/", 1)[1]
+        else:
+            tail = path.replace(root.rstrip("/") + "/", "", 1)
+        run_id = tail.strip("/").split("/", 1)[0]
+        if run_id and run_id != "runs":
+            run_ids.add(run_id)
+
+    try:
+        for entry in fs.ls(root, detail=True):
+            name = entry.get("name") if isinstance(entry, dict) else str(entry)
+            if name:
+                add_from_path(name)
+    except Exception:
+        pass
+
+    for pattern in (f"{root}/*", f"{root}/*/summary.json", f"{root}/*/launch.json", f"{root}/*/state.json"):
+        try:
+            for path in fs.glob(pattern):
+                add_from_path(str(path))
+                if len(run_ids) >= limit:
+                    break
+        except Exception:
+            continue
+
+    return sorted(run_ids, reverse=True)[:limit]
+
+
 def list_recent_runs(
     *,
     bucket_source: str,
@@ -302,35 +388,34 @@ def list_recent_runs(
 ) -> list[dict[str, Any]]:
     """List recent run summaries from a user's bucket.
 
-    Best effort: if a run folder is incomplete, return the information that is
-    available instead of failing the entire explorer.
+    Best effort: a run can be visible as soon as launch.json/summary.json exists,
+    before the worker writes full state/events. This keeps running Jobs visible
+    and makes Job links available immediately.
     """
-    fs = _fs(token)
     root = f"{bucket_uri_from_source(bucket_source)}/runs"
-    try:
-        entries = fs.ls(root, detail=True)
-    except FileNotFoundError:
-        return []
-    except Exception:
+    run_ids = _discover_run_ids(root, token=token)
+    if not run_ids:
         return []
 
     runs: list[dict[str, Any]] = []
-    for entry in entries:
-        name = entry.get("name") if isinstance(entry, dict) else str(entry)
-        if not name:
-            continue
-        run_id = _run_id_from_path(name)
-        if run_id in {"runs", ""}:
-            continue
+    for run_id in run_ids:
+        summary_file = read_json(f"{root}/{run_id}/summary.json", token=token) or {}
         launch = read_json(f"{root}/{run_id}/launch.json", token=token) or {}
         state = read_json(f"{root}/{run_id}/state.json", token=token) or {}
         gate = read_json(f"{root}/{run_id}/inference_gate.json", token=token) or {}
         smoke = read_json(f"{root}/{run_id}/tests/generation_smoke.json", token=token) or read_json(f"{root}/{run_id}/generation_smoke.json", token=token) or {}
         hardware = read_json(f"{root}/{run_id}/hardware_strategy.json", token=token) or {}
-        partial_bundle = {"launch": launch, "state": state, "inference_gate": gate, "generation_smoke": smoke, "hardware_strategy": hardware}
+        partial_bundle = {
+            "summary_file": summary_file,
+            "launch": launch,
+            "state": state,
+            "inference_gate": gate,
+            "generation_smoke": smoke,
+            "hardware_strategy": hardware,
+        }
         item = summarize_run_bundle(run_id, partial_bundle, bucket_source=bucket_source)
         item["bucket_source"] = bucket_source
-        haystack = " ".join(str(item.get(k, "")) for k in ["run_id", "model_id", "target_space", "status", "kind"]).lower()
+        haystack = " ".join(str(item.get(k, "")) for k in ["run_id", "model_id", "target_space", "status", "kind", "job_id"]).lower()
         if query and query.lower() not in haystack:
             continue
         if status and status not in {"all", ""} and item["status"] != status: