| from __future__ import annotations |
|
|
| import json |
| import tempfile |
| from datetime import datetime, timezone |
| from dataclasses import dataclass |
| from pathlib import Path |
| from typing import Any |
|
|
| from huggingface_hub import HfFileSystem, bucket_info, create_bucket, sync_bucket |
|
|
| from .config import bucket_uri_from_source, normalize_bucket_name, settings, user_bucket_source |
| from .effective_status import compute_effective_run_status |
| from .security import redact |
|
|
|
|
| @dataclass(frozen=True) |
| class RunPaths: |
| run_id: str |
| bucket_source: str |
|
|
| @property |
| def bucket_uri(self) -> str: |
| return bucket_uri_from_source(self.bucket_source) |
|
|
| @property |
| def root(self) -> str: |
| return f"{self.bucket_uri}/{_runs_prefix()}/{self.run_id}" |
|
|
| @property |
| def state(self) -> str: |
| return f"{self.root}/state.json" |
|
|
| @property |
| def events(self) -> str: |
| return f"{self.root}/events.jsonl" |
|
|
| @property |
| def report(self) -> str: |
| return f"{self.root}/report.md" |
|
|
|
|
|
|
| def _runs_prefix() -> str: |
| return settings.bucket_runs_prefix.strip().strip("/") or "runs" |
|
|
|
|
| def _run_tree_url(bucket_source: str, run_id: str, rel_path: str = "") -> str: |
| suffix = f"/{rel_path.strip('/')}" if rel_path else "" |
| return f"https://huggingface.co/buckets/{bucket_source}/tree/{_runs_prefix()}/{run_id}{suffix}" |
|
|
|
|
| def _fs(token: str | None = None) -> HfFileSystem: |
| return HfFileSystem(token=token) |
|
|
|
|
| def check_user_bucket(*, username: str, bucket_name: str | None = None, token: str | None = None) -> dict[str, Any]: |
| """Return bucket readiness for the signed-in user without creating resources.""" |
| bucket_source = user_bucket_source(username=username, bucket_name=bucket_name) |
| bucket_uri = bucket_uri_from_source(bucket_source) |
| try: |
| info = bucket_info(bucket_source, token=token) |
| return { |
| "ok": True, |
| "exists": True, |
| "bucket_source": bucket_source, |
| "bucket_uri": bucket_uri, |
| "name": getattr(info, "name", normalize_bucket_name(bucket_name)), |
| "private": getattr(info, "private", None), |
| } |
| except Exception as exc: |
| error = str(exc) |
| not_found = any(marker in error.lower() for marker in ["404", "not found", "repository not found", "bucket not found"]) |
| return { |
| "ok": False, |
| "exists": False if not_found else None, |
| "bucket_source": bucket_source, |
| "bucket_uri": bucket_uri, |
| "error": error, |
| } |
|
|
|
|
| def create_user_bucket(*, username: str, bucket_name: str | None = None, token: str | None = None) -> dict[str, Any]: |
| """Create the signed-in user's private run bucket, then return readiness.""" |
| bucket_source = user_bucket_source(username=username, bucket_name=bucket_name) |
| try: |
| url = create_bucket(bucket_source, private=True, exist_ok=True, token=token) |
| except Exception as exc: |
| return { |
| "ok": False, |
| "exists": None, |
| "bucket_source": bucket_source, |
| "bucket_uri": bucket_uri_from_source(bucket_source), |
| "error": str(exc), |
| } |
| status = check_user_bucket(username=username, bucket_name=bucket_name, token=token) |
| status["created_or_existing"] = True |
| status["create_url"] = str(url) |
| return status |
|
|
|
|
| def assert_user_bucket_ready(*, username: str, bucket_name: str | None = None, token: str | None = None) -> dict[str, Any]: |
| """Raise a clear error if the user's run bucket cannot be used.""" |
| status = check_user_bucket(username=username, bucket_name=bucket_name, token=token) |
| if not status.get("ok"): |
| source = status.get("bucket_source") or user_bucket_source(username=username, bucket_name=bucket_name) |
| error = status.get("error") or "Bucket does not exist or is not accessible." |
| raise ValueError( |
| f"Run bucket `{source}` is not ready. Click 'Create private run bucket' first, " |
| f"or create it manually in Hugging Face Storage Buckets. Details: {error}" |
| ) |
| return status |
|
|
|
|
| def read_text(path: str, token: str | None = None) -> str | None: |
| fs = _fs(token) |
| try: |
| with fs.open(path, "r") as f: |
| return f.read() |
| except FileNotFoundError: |
| return None |
| except Exception as exc: |
| return f"[Could not read {path}: {exc}]" |
|
|
|
|
| def read_json(path: str, token: str | None = None) -> dict[str, Any] | None: |
| content = read_text(path, token=token) |
| if not content or content.startswith("[Could not read"): |
| return None |
| try: |
| return json.loads(content) |
| except json.JSONDecodeError: |
| return {"_error": "Invalid JSON", "raw": redact(content)} |
|
|
|
|
|
|
|
|
| def _is_bucket_uri(path: str) -> bool: |
| """Return True for canonical HF Storage Bucket URIs.""" |
| return str(path or "").startswith("hf://buckets/") |
|
|
|
|
| def _bucket_uri_without_scheme(path: str) -> str: |
| """Return the HfFileSystem-native path form without the optional hf:// prefix.""" |
| return str(path or "").removeprefix("hf://") |
|
|
|
|
| def _split_bucket_uri(path: str) -> tuple[str, str]: |
| """Return ``(bucket_source, relative_path)`` for an ``hf://buckets/...`` URI.""" |
| value = str(path or "") |
| if not value.startswith("hf://buckets/"): |
| raise ValueError(f"Expected an hf://buckets/ URI, got: {redact(value)}") |
| rest = value.removeprefix("hf://buckets/").strip("/") |
| parts = rest.split("/", 2) |
| if len(parts) < 3 or not parts[0] or not parts[1] or not parts[2]: |
| raise ValueError(f"Bucket URI must include namespace, bucket and object path: {redact(value)}") |
| return f"{parts[0]}/{parts[1]}", parts[2] |
|
|
|
|
| def _sync_bucket_single_text(path: str, content: str, *, token: str | None = None) -> None: |
| """Fallback writer using the official bucket sync API. |
| |
| Some HfFileSystem/fsspec versions can route fresh bucket writes through the |
| repository resolver and fail with messages like "repository and revision |
| exist". ``sync_bucket`` is the same path used by ``hf buckets sync`` and is |
| purpose-built for mutable bucket objects. A tiny temporary folder mirrors |
| the object path so only the requested file is uploaded. |
| """ |
| bucket_source, rel_path = _split_bucket_uri(path) |
| with tempfile.TemporaryDirectory(prefix="asf-bucket-write-") as tmp: |
| root = Path(tmp) |
| local_path = root / rel_path |
| local_path.parent.mkdir(parents=True, exist_ok=True) |
| local_path.write_text(content, encoding="utf-8") |
| sync_bucket(source=str(root), dest=bucket_uri_from_source(bucket_source), token=token, quiet=True) |
|
|
|
|
| def _write_bucket_text(fs: HfFileSystem, path: str, content: str, *, token: str | None = None) -> None: |
| """Write text to an HF Bucket object without creating virtual directories. |
| |
| API-side writes use remote ``hf://buckets/<namespace>/<bucket>/...`` object |
| URIs. Job-side writes use the mounted local path. For bucket URIs, try the |
| native fsspec write helpers first, then fall back to ``sync_bucket``. Never |
| call ``makedirs`` for virtual bucket prefixes such as ``runs/<run_id>/``. |
| """ |
| attempts: list[str] = [] |
| if _is_bucket_uri(path): |
| data = content.encode("utf-8") |
| candidates = [path, _bucket_uri_without_scheme(path)] |
| for candidate in candidates: |
| try: |
| fs.write_text(candidate, content, encoding="utf-8") |
| return |
| except Exception as exc: |
| attempts.append(f"write_text({candidate!r}): {exc}") |
| try: |
| fs.pipe_file(candidate, data, mode="overwrite") |
| return |
| except Exception as exc: |
| attempts.append(f"pipe_file({candidate!r}): {exc}") |
| try: |
| with fs.open(candidate, "wb") as f: |
| f.write(data) |
| return |
| except Exception as exc: |
| attempts.append(f"open-wb({candidate!r}): {exc}") |
| try: |
| _sync_bucket_single_text(path, content, token=token) |
| return |
| except Exception as exc: |
| attempts.append(f"sync_bucket({path!r}): {exc}") |
| raise RuntimeError("Could not write to HF Storage Bucket object. " + " | ".join(attempts[-4:])) from exc |
| with fs.open(path, "w") as f: |
| f.write(content) |
|
|
|
|
| def write_text(path: str, content: str, token: str | None = None) -> None: |
| """Write a text document to a bucket path using HfFileSystem or bucket sync.""" |
| fs = _fs(token) |
| _write_bucket_text(fs, path, content, token=token) |
|
|
|
|
| def write_json(path: str, payload: dict[str, Any], token: str | None = None) -> None: |
| """Write a JSON document to a bucket path using HfFileSystem. |
| |
| Used by API routes to persist launch metadata immediately after a Job is |
| created, before the worker has had a chance to write state.json. |
| """ |
| write_text(path, json.dumps(payload, indent=2, ensure_ascii=False), token=token) |
|
|
|
|
| def append_run_event(run_id: str, *, bucket_source: str, step: str, status: str, message: str, details: dict[str, Any] | None = None, token: str | None = None) -> dict[str, Any]: |
| """Append one JSONL event to a run's events.jsonl in the bucket.""" |
| paths = RunPaths(run_id, bucket_source=bucket_source) |
| event = { |
| "ts": datetime.now(timezone.utc).isoformat(), |
| "step": step, |
| "status": status, |
| "message": message, |
| "details": details or {}, |
| } |
| try: |
| existing = read_text(paths.events, token=token) or "" |
| if existing and not existing.endswith("\n"): |
| existing += "\n" |
| except Exception: |
| existing = "" |
| write_text(paths.events, existing + json.dumps(event, ensure_ascii=False) + "\n", token=token) |
| return event |
|
|
|
|
| def _bucket_path_kind(entry: Any) -> tuple[str, str | None]: |
| """Return ``(path, type)`` for HfFileSystem listing entries. |
| |
| HfFileSystem may return plain strings or fsspec-style dictionaries, and |
| bucket directory objects are virtual. Centralise the normalisation so run |
| deletion can remove concrete files without depending on a single listing |
| shape. |
| """ |
| if isinstance(entry, dict): |
| return str(entry.get("name") or entry.get("path") or ""), entry.get("type") |
| return str(entry or ""), None |
|
|
|
|
| def _collect_run_folder_files(fs: HfFileSystem, root: str) -> list[str]: |
| """Collect concrete bucket objects under a run prefix. |
| |
| A single ``rm(root, recursive=True)`` can be unreliable on HF Storage |
| Buckets because the run folder is a virtual prefix. Prefer explicit |
| recursive discovery, with several fallbacks for older/partial filesystem |
| behaviours. |
| """ |
| root = root.rstrip("/") |
| found: set[str] = set() |
|
|
| |
| try: |
| entries = fs.find(root, withdirs=False, detail=False, refresh=True) |
| for entry in entries or []: |
| path, kind = _bucket_path_kind(entry) |
| if path and path != root and not path.endswith("/") and kind != "directory": |
| found.add(path) |
| except Exception: |
| pass |
|
|
| |
| for pattern in (f"{root}/**", f"{root}/**/*", f"{root}/*"): |
| try: |
| for entry in fs.glob(pattern) or []: |
| path, kind = _bucket_path_kind(entry) |
| if path and path != root and not path.endswith("/") and kind != "directory": |
| found.add(path) |
| except Exception: |
| pass |
|
|
| |
| |
| pending = [root] |
| seen_dirs: set[str] = set() |
| while pending: |
| prefix = pending.pop().rstrip("/") |
| if prefix in seen_dirs: |
| continue |
| seen_dirs.add(prefix) |
| try: |
| entries = fs.ls(prefix, detail=True, refresh=True) or [] |
| except Exception: |
| continue |
| for entry in entries: |
| path, kind = _bucket_path_kind(entry) |
| if not path or path == prefix: |
| continue |
| if kind == "directory" or path.endswith("/"): |
| pending.append(path.rstrip("/")) |
| else: |
| found.add(path) |
|
|
| return sorted(found, key=lambda value: value.count("/"), reverse=True) |
|
|
|
|
| def delete_run_folder(run_id: str, *, bucket_source: str, token: str | None = None) -> dict[str, Any]: |
| """Delete every concrete object under a run prefix from the private bucket. |
| |
| Returns a small report so the API and tests can distinguish a real bucket |
| cleanup from a UI-only tombstone. The generated/tested Space is never |
| deleted here; only ``runs/<run_id>/`` bucket artifacts are removed. |
| """ |
| paths = RunPaths(run_id, bucket_source=bucket_source) |
| root = paths.root.rstrip("/") |
| fs = _fs(token) |
| files = _collect_run_folder_files(fs, root) |
| deleted: list[str] = [] |
| errors: list[str] = [] |
|
|
| for path in files: |
| try: |
| fs.rm(path, recursive=False) |
| deleted.append(path) |
| except FileNotFoundError: |
| deleted.append(path) |
| except Exception as exc: |
| errors.append(f"{path}: {redact(str(exc))}") |
|
|
| |
| |
| |
| try: |
| fs.rm(root, recursive=True) |
| except FileNotFoundError: |
| pass |
| except Exception as exc: |
| |
| |
| |
| if not deleted and not files: |
| errors.append(f"{root}: {redact(str(exc))}") |
|
|
| remaining = _collect_run_folder_files(fs, root) |
| if remaining: |
| errors.append(f"{len(remaining)} object(s) still present under runs/{run_id}") |
|
|
| if errors: |
| raise RuntimeError("; ".join(errors)) |
|
|
| return { |
| "root": root, |
| "matched_count": len(files), |
| "deleted_count": len(deleted), |
| "remaining_count": len(remaining), |
| } |
|
|
|
|
| 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 "", |
| "parent_build_run_id": launch.get("parent_build_run_id") or "", |
| "source_kind": launch.get("source_kind") or "", |
| "linked_target_space": launch.get("linked_target_space") or "", |
| "artifacts_url": f"https://huggingface.co/buckets/{bucket_source}/tree/{(settings.bucket_runs_prefix.strip().strip('/') or 'runs')}/{run_id}", |
| } |
| write_json(f"{paths.root}/summary.json", summary, token=token) |
| |
| |
| 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]]: |
| paths = RunPaths(run_id, bucket_source=bucket_source) |
| content = read_text(paths.events, token=token) |
| if not content: |
| return [] |
| events: list[dict[str, Any]] = [] |
| for line in content.splitlines(): |
| line = line.strip() |
| if not line: |
| continue |
| try: |
| events.append(json.loads(line)) |
| except json.JSONDecodeError: |
| events.append({"step": "parse_events", "status": "warning", "message": redact(line)}) |
| return events |
|
|
|
|
| def _safe_read_json(path: str, token: str | None = None) -> dict[str, Any]: |
| return read_json(path, token=token) or {} |
|
|
|
|
| def _safe_read_text(path: str, token: str | None = None) -> str: |
| return redact(read_text(path, token=token) or "") |
|
|
|
|
|
|
|
|
| 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_sources(*, bucket_source: str, summary_file: dict[str, Any], launch: dict[str, Any], state: dict[str, Any]) -> str: |
| """Return a stable HF Job URL from any metadata shape we have. |
| |
| Older run bundles may have job_id in summary.json, created_by in launch.json, |
| or only the bucket owner available. The UI should still expose the Job link |
| whenever a job_id is known. |
| """ |
| explicit = summary_file.get("job_url") or launch.get("job_url") or state.get("job_url") |
| if explicit: |
| return str(explicit) |
| job_id = summary_file.get("job_id") or launch.get("job_id") or state.get("job_id") |
| owner = ( |
| summary_file.get("created_by") |
| or summary_file.get("username") |
| or summary_file.get("owner") |
| or launch.get("created_by") |
| or launch.get("username") |
| or launch.get("owner") |
| or state.get("created_by") |
| or state.get("username") |
| or state.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 _job_url_from_launch_or_state(*, launch: dict[str, Any], state: dict[str, Any]) -> str: |
| |
| return _job_url_from_sources(bucket_source="", summary_file={}, launch=launch, state=state) |
|
|
|
|
| def _job_id_from_launch_or_state(*, launch: dict[str, Any], state: dict[str, Any]) -> str: |
| return str(launch.get("job_id") or state.get("job_id") or "") |
|
|
| def _list_run_files( |
| run_id: str, |
| *, |
| bucket_source: str, |
| token: str | None = None, |
| prefixes: tuple[str, ...] = ("generated", "tests", "logs", "artifacts", "traces", "repair"), |
| max_files: int = 120, |
| ) -> list[dict[str, Any]]: |
| """Return a compact, best-effort file index for a run. |
| |
| The Run Explorer should not fail when a Bucket contains partial or old runs. |
| This function therefore treats every listing error as non-fatal and limits |
| recursion so the UI stays responsive even when traces are large. |
| """ |
| fs = _fs(token) |
| base = f"{bucket_uri_from_source(bucket_source)}/{_runs_prefix()}/{run_id}" |
| files: list[dict[str, Any]] = [] |
|
|
| def walk(prefix: str, depth: int = 0) -> None: |
| if len(files) >= max_files or depth > 3: |
| return |
| path = f"{base}/{prefix}" |
| try: |
| entries = fs.ls(path, detail=True) |
| except Exception: |
| return |
| for entry in entries: |
| if len(files) >= max_files: |
| break |
| name = entry.get("name") if isinstance(entry, dict) else str(entry) |
| if not name: |
| continue |
| typ = str(entry.get("type") or "") if isinstance(entry, dict) else "" |
| size = entry.get("size") if isinstance(entry, dict) else None |
| rel = name.replace(base + "/", "", 1) |
| if typ == "directory": |
| walk(rel, depth + 1) |
| continue |
| files.append( |
| { |
| "path": rel, |
| "name": rel.split("/")[-1], |
| "size": size, |
| "url": _run_tree_url(bucket_source, run_id, rel), |
| } |
| ) |
|
|
| for prefix in prefixes: |
| walk(prefix) |
| return files |
|
|
|
|
|
|
|
|
| def _bucket_file_url(bucket_source: str, run_id: str, rel_path: str) -> str: |
| """Return the Hugging Face Bucket UI URL for a run file. |
| |
| Bucket file pages use `/tree/...`, not `/blob/...`. Keeping this helper |
| named after the UI concept avoids producing broken document links. |
| """ |
| return _bucket_tree_url(bucket_source, run_id, rel_path) |
|
|
|
|
| def _bucket_tree_url(bucket_source: str, run_id: str, rel_path: str = "") -> str: |
| suffix = f"/{rel_path.strip('/')}" if rel_path else "" |
| prefix = settings.bucket_runs_prefix.strip().strip("/") or "runs" |
| return f"https://huggingface.co/buckets/{bucket_source}/tree/{prefix}/{run_id}{suffix}" |
|
|
|
|
| def _path_exists(path: str, token: str | None = None) -> bool: |
| """Best-effort existence check for Bucket objects. |
| |
| HfFileSystem.exists() can lag or return false for recently written Bucket |
| objects depending on the backend view used by a running Job. For UI affordances |
| such as the Run traces dock, a false negative is worse than a slightly slower |
| check, so fall back to parent listings and glob probes. |
| """ |
| fs = _fs(token) |
| try: |
| if bool(fs.exists(path)): |
| return True |
| except Exception: |
| pass |
| try: |
| parent, name = path.rsplit("/", 1) |
| for entry in fs.ls(parent, detail=True): |
| entry_name = entry.get("name") if isinstance(entry, dict) else str(entry) |
| if str(entry_name).rstrip("/") == path.rstrip("/") or str(entry_name).rstrip("/").endswith("/" + name): |
| return True |
| except Exception: |
| pass |
| try: |
| return bool(list(fs.glob(path))[:1]) |
| except Exception: |
| return False |
|
|
|
|
| def _has_glob(pattern: str, token: str | None = None) -> bool: |
| try: |
| return bool(list(_fs(token).glob(pattern))[:1]) |
| except Exception: |
| return False |
|
|
|
|
| def _folder_has_content(path: str, token: str | None = None) -> bool: |
| """Return true when a Bucket folder/prefix contains at least one object.""" |
| fs = _fs(token) |
| prefix = path.rstrip("/") |
| for probe in (f"{prefix}/agent_trace.jsonl", f"{prefix}/events.jsonl"): |
| if _path_exists(probe, token=token): |
| return True |
| try: |
| entries = fs.ls(prefix, detail=True) |
| if entries: |
| return True |
| except Exception: |
| pass |
| for pattern in (f"{prefix}/*", f"{prefix}/**/*"): |
| if _has_glob(pattern, token=token): |
| return True |
| return False |
|
|
|
|
| def _run_document_links(run_id: str, *, bucket_source: str, bundle: dict[str, Any], token: str | None = None) -> list[dict[str, Any]]: |
| """Return the small document dock shown under Active Run events. |
| |
| Keep this intentionally narrow: the full bucket folder is already available |
| through the Artifacts button. These are only the high-signal files users |
| need most often while inspecting a build run. |
| |
| Trace buttons link only to the stable redacted trace folder: |
| - Pi redacted → `traces/redacted` |
| |
| RAW Pi traces are intentionally not published to the bucket and are not |
| exposed in the document dock. Redacted traces become active only when at |
| least one real trace file/folder exists underneath. |
| """ |
| paths = RunPaths(run_id, bucket_source=bucket_source) |
| files = [f for f in (bundle.get("files") or []) if isinstance(f, dict)] |
| file_paths = {str(f.get("path") or "") for f in files} |
| manifest = bundle.get("artifact_manifest") or {} |
| manifest_artifacts = [a for a in (manifest.get("artifacts") or []) if isinstance(a, dict)] if isinstance(manifest, dict) else [] |
| manifest_paths = {str(a.get("path") or "") for a in manifest_artifacts if a.get("present")} |
| if isinstance(manifest, dict): |
| manifest_paths.update(str(p or "") for p in (manifest.get("present_paths") or []) if p) |
|
|
| terminal_success = str((bundle.get("final_status_reconciliation") or {}).get("status") or (bundle.get("summary") or {}).get("status") or (bundle.get("summary_file") or {}).get("status") or (bundle.get("state") or {}).get("status") or "").strip().lower() in {"full_inference_success", "success", "succeeded", "completed"} |
|
|
| def manifest_has(path: str) -> bool: |
| path = path.strip("/") |
| prefix = path.rstrip("/") + "/" |
| return path in manifest_paths or any(p.startswith(prefix) for p in manifest_paths) |
|
|
| def first_existing_path(*candidates: str) -> str: |
| for candidate in candidates: |
| clean = candidate.strip("/") |
| if clean in file_paths or clean in manifest_paths: |
| return clean |
| if token: |
| for candidate in candidates: |
| clean = candidate.strip("/") |
| if _path_exists(f"{paths.root}/{clean}", token=token): |
| return clean |
| return candidates[0].strip("/") if candidates else "" |
|
|
| def trace_folder_has_content(prefix: str) -> bool: |
| rel_prefix = prefix.rstrip("/") + "/" |
| if any(path.startswith(rel_prefix) and not path.endswith("/") for path in file_paths): |
| return True |
| if manifest_has(prefix): |
| return True |
| if not token: |
| return False |
| root_prefix = f"{paths.root}/{prefix.rstrip('/')}" |
| |
| return _folder_has_content(root_prefix, token=token) |
|
|
| redacted_present = trace_folder_has_content("traces/redacted") |
| redacted_trace_url = _bucket_tree_url(bucket_source, run_id, "traces/redacted") |
| smoke_path = first_existing_path( |
| "tests/generation_smoke.json", |
| "tests/generation_smoke_latest.json", |
| "generation_smoke.json", |
| "tests/test_result.json", |
| ) |
| blockers_path = first_existing_path("generated/TECHNICAL_BLOCKERS.json", "TECHNICAL_BLOCKERS.json") |
| smoke_present = ( |
| terminal_success |
| or bool(bundle.get("generation_smoke")) |
| or manifest_has("tests/generation_smoke.json") |
| or manifest_has("tests/generation_smoke_latest.json") |
| or manifest_has("generation_smoke.json") |
| or "tests/generation_smoke.json" in file_paths |
| or "tests/generation_smoke_latest.json" in file_paths |
| or "generation_smoke.json" in file_paths |
| or smoke_path in file_paths |
| or smoke_path in manifest_paths |
| ) |
| report_present = terminal_success or bool(bundle.get("report")) or manifest_has("report.md") or "report.md" in file_paths or (bool(token) and _path_exists(paths.report, token=token)) |
| blockers_present = bool(bundle.get("technical_blockers")) or manifest_has("generated/TECHNICAL_BLOCKERS.json") or manifest_has("TECHNICAL_BLOCKERS.json") or "generated/TECHNICAL_BLOCKERS.json" in file_paths or "TECHNICAL_BLOCKERS.json" in file_paths or blockers_path in file_paths |
| eval_publish_payload = bundle.get("eval_publish") if isinstance(bundle.get("eval_publish"), dict) else bundle.get("eval_publish_status") if isinstance(bundle.get("eval_publish_status"), dict) else {} |
| eval_publish_present = bool(eval_publish_payload) or manifest_has("eval_publish_status.json") or "eval_publish_status.json" in file_paths or (bool(token) and _path_exists(f"{paths.root}/eval_publish_status.json", token=token)) |
| useful_space_log_candidates = ( |
| "logs/space_logs_index.json", |
| "logs/space_logs_build.txt", |
| "logs/space_logs_runtime.txt", |
| "logs/space_logs_run.txt", |
| "logs/space_logs_fetch_status.json", |
| "logs/space_log_diagnostics.json", |
| "logs/space_runtime_snapshot.json", |
| "logs/pi_live_output.txt", |
| "logs/pi_output.txt", |
| "logs/build_log_fast_probe_status.json", |
| ) |
| space_logs_path = first_existing_path(*useful_space_log_candidates) |
| space_logs_index_present = manifest_has("logs/space_logs_index.json") or "logs/space_logs_index.json" in file_paths or (bool(token) and _path_exists(f"{paths.root}/logs/space_logs_index.json", token=token)) |
| space_logs_present = ( |
| space_logs_index_present |
| or manifest_has("logs") |
| or any(path in file_paths or path in manifest_paths for path in useful_space_log_candidates) |
| or any(path.startswith("logs/") and not path.endswith("/") for path in file_paths | manifest_paths) |
| or (bool(token) and _folder_has_content(f"{paths.root}/logs", token=token)) |
| ) |
| if not space_logs_index_present and space_logs_present and space_logs_path == "logs/space_logs_index.json": |
| space_logs_path = "logs" |
| output_artifact_candidates = ( |
| "artifacts/image.webp", |
| "artifacts/image.png", |
| "artifacts/image.jpg", |
| "artifacts/image.jpeg", |
| "artifacts/output.png", |
| "artifacts/output.webp", |
| "artifacts/output.json", |
| ) |
| output_artifact_path = first_existing_path(*output_artifact_candidates) |
| output_artifact_present = ( |
| any(path in file_paths or path in manifest_paths for path in output_artifact_candidates) |
| or manifest_has("artifacts") |
| or (bool(token) and _folder_has_content(f"{paths.root}/artifacts", token=token)) |
| ) |
| if output_artifact_present and output_artifact_path == "artifacts/image.webp" and output_artifact_path not in file_paths and output_artifact_path not in manifest_paths and not manifest_has(output_artifact_path): |
| output_artifact_path = "artifacts" |
| build_error_present = manifest_has("build_error_observation.json") or "build_error_observation.json" in file_paths or (bool(token) and _path_exists(f"{paths.root}/build_error_observation.json", token=token)) |
| repair_decision_present = manifest_has("repair/REPAIR_DECISION.json") or "repair/REPAIR_DECISION.json" in file_paths or (bool(token) and _path_exists(f"{paths.root}/repair/REPAIR_DECISION.json", token=token)) |
| blockage_present = manifest_has("repair/BLOCKAGE.json") or "repair/BLOCKAGE.json" in file_paths or (bool(token) and _path_exists(f"{paths.root}/repair/BLOCKAGE.json", token=token)) |
| repair_present = manifest_has("repair") or any(path.startswith("repair/") for path in file_paths) or (bool(token) and _path_exists(f"{paths.root}/repair/REPAIR_SUMMARY.md", token=token)) |
| status = str((bundle.get("summary") or {}).get("status") or (bundle.get("state") or {}).get("status") or "").lower() |
| blockers_relevant = blockers_present or any(marker in status for marker in ("manual", "blocker", "failed", "error")) |
|
|
| docs = [ |
| { |
| "id": "pi_redacted_trace", |
| "label": "Pi redacted", |
| "subtitle": "Unified safe agent trace", |
| "icon": "🛡️", |
| "present": redacted_present, |
| "url": redacted_trace_url, |
| "sensitivity": "safe", |
| }, |
| { |
| "id": "report", |
| "label": "Report", |
| "subtitle": "Human summary", |
| "icon": "📄", |
| "present": report_present, |
| "url": _bucket_file_url(bucket_source, run_id, "report.md"), |
| }, |
| { |
| "id": "smoke", |
| "label": "Smoke", |
| "subtitle": "Validation + latency", |
| "icon": "⚡", |
| "present": smoke_present, |
| "url": _bucket_file_url(bucket_source, run_id, smoke_path), |
| }, |
| { |
| "id": "smoke_output", |
| "label": "Output", |
| "subtitle": "Generated artifact", |
| "icon": "🖼️", |
| "present": output_artifact_present, |
| "url": _bucket_file_url(bucket_source, run_id, output_artifact_path) if output_artifact_path != "artifacts" else _bucket_tree_url(bucket_source, run_id, "artifacts"), |
| }, |
| { |
| "id": "eval_publish", |
| "label": "Eval publish", |
| "subtitle": "Archive status", |
| "icon": "◎", |
| "present": eval_publish_present, |
| "url": _bucket_file_url(bucket_source, run_id, "eval_publish_status.json"), |
| "tone": "success" if eval_publish_payload.get("published") else "pending" if eval_publish_present and str(eval_publish_payload.get("reason") or "") == "record_not_ready" else "warn" if eval_publish_present else "neutral", |
| }, |
| { |
| "id": "space_logs", |
| "label": "Space logs", |
| "subtitle": "Build/runtime log index" if space_logs_index_present else "Build/runtime logs", |
| "icon": "▣", |
| "present": space_logs_present, |
| "url": _bucket_file_url(bucket_source, run_id, space_logs_path) if space_logs_path != "logs" else _bucket_tree_url(bucket_source, run_id, "logs"), |
| "tone": "warn" if "failed" in status or "error" in status else "neutral", |
| }, |
| { |
| "id": "build_error", |
| "label": "Build error", |
| "subtitle": "Observed build failure", |
| "icon": "⛔", |
| "present": build_error_present, |
| "url": _bucket_file_url(bucket_source, run_id, "build_error_observation.json"), |
| "tone": "warn", |
| }, |
| ] |
| if repair_decision_present: |
| docs.append( |
| { |
| "id": "repair_decision", |
| "label": "Decision", |
| "subtitle": "Pi diagnosis action", |
| "icon": "◇", |
| "present": repair_decision_present, |
| "url": _bucket_file_url(bucket_source, run_id, "repair/REPAIR_DECISION.json"), |
| "tone": "warn" if "failed" in status or "error" in status else "neutral", |
| } |
| ) |
| if repair_present: |
| docs.append( |
| { |
| "id": "repair", |
| "label": "Repair", |
| "subtitle": "Brief, plan and summary", |
| "icon": "✦", |
| "present": repair_present, |
| "url": _bucket_tree_url(bucket_source, run_id, "repair"), |
| "tone": "warn" if "failed" in status or "error" in status else "neutral", |
| } |
| ) |
| if blockage_present: |
| docs.append( |
| { |
| "id": "blockage", |
| "label": "Blockage", |
| "subtitle": "Terminal recovery reason", |
| "icon": "!", |
| "present": blockage_present, |
| "url": _bucket_file_url(bucket_source, run_id, "repair/BLOCKAGE.json"), |
| "tone": "warn", |
| } |
| ) |
| if blockers_relevant: |
| docs.append( |
| { |
| "id": "blockers", |
| "label": "Blockers", |
| "subtitle": "Why inference is blocked", |
| "icon": "⚠️", |
| "present": blockers_present, |
| "url": _bucket_file_url(bucket_source, run_id, blockers_path), |
| "tone": "warn", |
| } |
| ) |
| if "failed" in status or "error" in status: |
| priority = {"report": 0, "blockage": 1, "build_error": 2, "space_logs": 3, "repair_decision": 4, "repair": 5, "blockers": 6, "pi_redacted_trace": 7, "smoke": 8, "smoke_output": 9, "eval_publish": 10} |
| docs.sort(key=lambda item: priority.get(str(item.get("id") or ""), 50)) |
| return docs |
|
|
|
|
| def _normalize_model_name(value: str | None) -> str: |
| raw = (value or "").strip().lower() |
| if "/" in raw: |
| raw = raw.rsplit("/", 1)[-1] |
| return re.sub(r"[^a-z0-9]+", "", raw) |
|
|
|
|
| def _extract_pi_models_from_text(text: str) -> list[str]: |
| if not text: |
| return [] |
| patterns = [ |
| r"\bQwen/[A-Za-z0-9_.-]+", |
| r"\bQwen(?:2(?:\.5)?|3)?[-_/A-Za-z0-9.]*Coder[-_/A-Za-z0-9.]*", |
| r"\bmoonshotai/[A-Za-z0-9_.-]+", |
| r"\bKimi[-_/A-Za-z0-9.]+", |
| r"\bzai-org/[A-Za-z0-9_.-]+", |
| r"\bGLM[-_/A-Za-z0-9.]+", |
| r"\bdeepseek-ai/[A-Za-z0-9_.-]+", |
| r"\bDeepSeek[-_/A-Za-z0-9.]+", |
| r"\bClaude[-_/A-Za-z0-9.]+", |
| r"\bGPT[-_/A-Za-z0-9.]+", |
| r"\b[Mm]odel(?:\\s+used|\\s+selected|\\s*:)?\\s*[:=]?\\s*([A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+)", |
| ] |
| found: list[str] = [] |
| for pattern in patterns: |
| for match in re.finditer(pattern, text, flags=re.IGNORECASE): |
| value = match.group(1) if match.groups() else match.group(0) |
| value = value.strip().strip('",` ') |
| if value and value not in found: |
| found.append(value) |
| return found[:12] |
|
|
|
|
| def _trace_based_pi_model_resolution(run_id: str, bundle: dict[str, Any], *, bucket_source: str, token: str | None = None) -> dict[str, Any]: |
| state = bundle.get("state") or {} |
| launch = bundle.get("launch") or {} |
| requested = ( |
| state.get("pi_model") |
| or launch.get("pi_model") |
| or launch.get("PI_MODEL") |
| or state.get("pi_model_resolution", {}).get("requested_model") |
| or "" |
| ) |
| configured = state.get("pi_model_resolution", {}).get("configured_model") or requested |
| root = RunPaths(run_id, bucket_source=bucket_source).root |
| trace_texts: list[str] = [] |
| try: |
| fs = _fs(token) |
| for pattern in ( |
| f"{root}/traces/redacted/**/*.jsonl", |
| f"{root}/traces/redacted/*.jsonl", |
| f"{root}/logs/pi_output.txt", |
| f"{root}/logs/pi_live_output.txt", |
| ): |
| for path in list(fs.glob(pattern))[:8]: |
| try: |
| trace_texts.append(_safe_read_text(str(path), token=token)[:120000]) |
| except Exception: |
| pass |
| if len(trace_texts) >= 8: |
| break |
| except Exception: |
| pass |
| observed = _extract_pi_models_from_text("\n".join(trace_texts)) |
| n_requested = _normalize_model_name(requested) |
| n_configured = _normalize_model_name(configured) |
| effective = "" |
| for raw in observed: |
| n_raw = _normalize_model_name(raw) |
| if n_raw and n_raw not in {n_requested, n_configured}: |
| effective = raw |
| break |
| if not effective and observed: |
| effective = observed[0] |
| mismatch = bool(effective and _normalize_model_name(effective) not in {n_requested, n_configured}) |
| if not requested and not observed: |
| return {} |
| return { |
| "requested_model": requested, |
| "configured_model": configured or requested, |
| "observed_models": observed, |
| "effective_model": effective or configured or requested, |
| "provider": "huggingface", |
| "mismatch": mismatch, |
| "source": "published_pi_traces", |
| } |
|
|
|
|
|
|
| def _terminal_status_from_events(events: list[dict[str, Any]] | None, *, validation: bool = False) -> str: |
| """Return the strongest terminal status observed in events, newest first. |
| |
| Validation runs are short-lived and their terminal events are authoritative. |
| Build runs, however, contain recoverable incident events (ZeroGPU fallback, |
| transient Space boot failures, repair probes). For builds, only explicit |
| final verdict events may decide the whole run status; intermediate failed |
| events must remain phase diagnostics and never paint the active run red. |
| """ |
| final_steps = {"done", "failure", "final", "report_write", "result", "run_result"} |
| for event in reversed(events or []): |
| step = str(event.get("step") or "").lower() |
| status = str(event.get("status") or "").lower() |
| message = str(event.get("message") or "").lower() |
| haystack = " ".join([step, status, message]) |
| if validation: |
| if any(x in haystack for x in ["manual", "requires_action", "hardware_required"]): |
| return "manual_hardware_required" |
| if any(x in haystack for x in ["failed", "failure", "error", "timeout"]): |
| return "failed" |
| if any(x in haystack for x in ["partial", "health_only", "not_verified", "completed_with_warnings"]): |
| return "partial_validation" |
| if step == "done" and ("full_inference_success" in haystack or "success" in haystack or "completed" in haystack): |
| return "full_inference_success" |
| if step in {"done", "report_write"} and status in {"success", "done", "completed"}: |
| return "full_inference_success" |
| continue |
|
|
| if step not in final_steps: |
| continue |
| if any(x in haystack for x in ["manual", "requires_action", "hardware_required"]): |
| return "manual_hardware_required" |
| if any(x in haystack for x in ["partial", "health_only", "not_verified", "completed_with_warnings"]): |
| return "partial_validation" |
| if "full_inference_success" in haystack or (step == "done" and any(x in haystack for x in ["success", "completed", "passed"])): |
| return "full_inference_success" |
| if step in {"failure", "final", "run_result"} and any(x in haystack for x in ["failed", "failure", "error", "timeout"]): |
| return "failed" |
| return "" |
|
|
|
|
|
|
|
|
| def _source_status_text(*sources: dict[str, Any] | None) -> str: |
| chunks: list[str] = [] |
| for src in sources: |
| if not isinstance(src, dict): |
| continue |
| for key in ( |
| "final_status", |
| "verdict", |
| "status", |
| "gate_status", |
| "smoke_status", |
| "validation_status", |
| "result_status", |
| "effective_status", |
| ): |
| value = src.get(key) |
| if value is not None: |
| chunks.append(str(value).lower()) |
| if src.get("ok") is True: |
| chunks.append("ok true success") |
| if src.get("manual_hardware_required") is True: |
| chunks.append("manual_hardware_required") |
| error = src.get("error") |
| if error: |
| chunks.append(str(error).lower()) |
| return " ".join(chunks) |
|
|
|
|
| def _classify_build_status_text(text: str) -> str: |
| text = str(text or "").lower() |
| if not text: |
| return "" |
| if "auth_refresh_required" in text or "oauth_expired" in text: |
| return "auth_refresh_required" |
| if any(x in text for x in ("manual_hardware_required", "manual hardware", "requires_action", "hardware_required")): |
| return "manual_hardware_required" |
| if "technical_blocker_boot_only" in text: |
| return "technical_blocker_boot_only" |
| if any(x in text for x in ("technical_blocker", "technical blocker", "blocked")): |
| return "technical_blocker" |
| if any(x in text for x in ("failed", "failure", "error", "timeout", "exception")): |
| return "failed" |
| if "stale" in text: |
| return "stale" |
| if any(x in text for x in ("cancelled", "canceled", "stopped")): |
| return "stopped" |
| if any(x in text for x in ("partial_validation", "full_inference_candidate_health_passed", "health_only", "demo_usable_full_promise_not_verified", "interactive_app_available_smoke_failed", "manual_test_required_smoke_failed", "not_verified", "completed_with_warnings")): |
| return "partial_validation" |
| if "full_inference_success" in text: |
| return "full_inference_success" |
| if any(x in text for x in ("success", "succeed", "passed", "completed")): |
| return "full_inference_success" |
| if any(x in text for x in ("running", "queued", "pending", "building", "waiting", "started")): |
| return "running" |
| return "" |
|
|
|
|
| def _explicit_effective_status(*sources: dict[str, Any] | None) -> str: |
| """Return an authoritative display/eval status without reading nested errors. |
| |
| v198.8: Run Explorer cards must not turn a usable partial build red just |
| because a nested generation smoke artifact says ``failed``. The worker |
| publishes effective verdict/status artifacts for exactly this reason; read |
| those fields first and classify only their values, not the attached error |
| text. |
| """ |
| for src in sources: |
| if not isinstance(src, dict): |
| continue |
| for key in ("ui_status", "display_status", "effective_verdict", "effective_status", "final_status", "verdict", "result_status", "validation_status", "status"): |
| value = src.get(key) |
| if value is None: |
| continue |
| status = _classify_build_status_text(str(value)) |
| if status and status != "running": |
| return status |
| return "" |
|
|
|
|
| def _truthy(value: Any) -> bool: |
| if value is True: |
| return True |
| if isinstance(value, str): |
| return value.strip().lower() in {"1", "true", "yes", "y", "success", "passed", "ok"} |
| return bool(value) if isinstance(value, (int, float)) else False |
|
|
|
|
| def _nested_get(src: dict[str, Any] | None, *path: str) -> Any: |
| cur: Any = src |
| for key in path: |
| if not isinstance(cur, dict): |
| return None |
| cur = cur.get(key) |
| return cur |
|
|
|
|
|
|
| def _contract_blocks_success(*sources: dict[str, Any] | None) -> bool: |
| """Return true when contracts/gate explicitly say this is diagnostic-only. |
| |
| v198.18: app boot or a diagnostic endpoint response must not hydrate a |
| technical blocker into Success. This guard is intentionally conservative: |
| any explicit no-real-inference/manual/diagnostic signal wins over nested |
| smoke success. |
| """ |
| text_chunks: list[str] = [] |
| for src in sources: |
| if not isinstance(src, dict): |
| continue |
| for key in ( |
| "promise_validation_status", |
| "ui_status", |
| "ui_badge", |
| "status", |
| "validation_level", |
| "inference_strategy", |
| "promise_fulfillment_risk", |
| "reason", |
| "message", |
| ): |
| value = src.get(key) |
| if value is not None: |
| text_chunks.append(str(value).lower()) |
| nested = src.get("promise_validation") |
| if isinstance(nested, dict): |
| for key in ("promise_validation_status", "ui_status", "ui_badge", "reason", "validation_level"): |
| value = nested.get(key) |
| if value is not None: |
| text_chunks.append(str(value).lower()) |
| if nested.get("promise_fulfilled") is False: |
| text_chunks.append("promise_fulfilled_false") |
| for bool_key in ("fallback_or_diagnostic_only", "manual_hardware_required"): |
| if src.get(bool_key) is True: |
| text_chunks.append(bool_key) |
| for bool_key in ("full_inference_implemented", "real_inference_implemented"): |
| if src.get(bool_key) is False: |
| text_chunks.append(f"{bool_key}_false") |
| text = " ".join(text_chunks) |
| blockers = ( |
| "not_fulfilled_diagnostic_only", |
| "deferred_manual_hardware", |
| "diagnostic-only", |
| "diagnostic_only", |
| "diagnostic space", |
| "manual_hardware_required", |
| "manual hardware required", |
| "manual_hardware_actionable", |
| "technical_blocker_boot_only", |
| "fallback_or_diagnostic_only", |
| "full_inference_implemented_false", |
| "real_inference_implemented_false", |
| "official demo", |
| "official space", |
| ) |
| return any(marker in text for marker in blockers) |
|
|
| def _has_persisted_generation_success(*sources: dict[str, Any] | None) -> bool: |
| """Return true for strong persisted proof that full generation worked. |
| |
| Older runs did not always store success in the same top-level field. The |
| Active Run bundle can infer success from nested gate/smoke/test artifacts, |
| while the Run Explorer list initially sees only lightweight JSON files. Keep |
| the list and stats aligned by recognizing the same durable success signals |
| without relying on transient UI state. |
| """ |
| for src in sources: |
| if not isinstance(src, dict): |
| continue |
| status_text = _source_status_text(src) |
| if "full_inference_success" in status_text: |
| return True |
| if _truthy(src.get("ok")) or _truthy(src.get("generation_smoke_passed")) or _truthy(src.get("smoke_test_passed")): |
| return True |
| if _truthy(_nested_get(src, "implementation_signals", "generation_smoke_passed")): |
| return True |
| if _truthy(_nested_get(src, "validation", "generation_smoke_passed")): |
| return True |
| if _truthy(_nested_get(src, "outcome", "generation_smoke_passed")): |
| return True |
| for nested_key in ("generation_smoke", "smoke", "test_result", "result"): |
| nested = src.get(nested_key) |
| if isinstance(nested, dict): |
| nested_status = str(nested.get("status") or nested.get("result_status") or nested.get("verdict") or "").lower() |
| if nested_status in {"success", "passed", "full_inference_success"}: |
| return True |
| if _truthy(nested.get("ok")) or _truthy(nested.get("passed")): |
| return True |
| return False |
|
|
|
|
| def _has_authoritative_runtime_success( |
| *, |
| final_status_reconciliation: dict[str, Any] | None = None, |
| summary_file: dict[str, Any] | None = None, |
| state: dict[str, Any] | None = None, |
| live_status: dict[str, Any] | None = None, |
| gate: dict[str, Any] | None = None, |
| smoke: dict[str, Any] | None = None, |
| ) -> bool: |
| """Return true when a final runtime proof supersedes stale Pi/contract blockers. |
| |
| v198.26.15: contract and Pi diagnostic signals are pre-runtime safety rails. |
| They must still protect diagnostic-only demos, but they must not downgrade a |
| run after the worker has published a final reconciliation/summary/state/live |
| proof of ``full_inference_success`` or a strong inference gate. |
| """ |
| def explicit_full_status(source: dict[str, Any]) -> bool: |
| for key in ("status", "final_status", "ui_status", "display_status", "effective_status", "effective_verdict", "verdict"): |
| if str(source.get(key) or "").strip().lower() == "full_inference_success": |
| return True |
| return False |
|
|
| for source in (final_status_reconciliation or {}, summary_file or {}, state or {}): |
| if not isinstance(source, dict): |
| continue |
| if explicit_full_status(source): |
| return True |
| if _truthy(source.get("generation_smoke_passed")) and _truthy(source.get("promise_fulfilled")): |
| return True |
|
|
| live = live_status or {} |
| if isinstance(live, dict): |
| live_status_text = _classify_build_status_text(_source_status_text(live)) |
| live_stage = str(live.get("stage") or "").strip().lower() |
| if live_stage == "done" and live_status_text == "full_inference_success": |
| return True |
| if live_stage == "done" and _truthy(live.get("generation_smoke_passed")): |
| return True |
|
|
| gate_src = gate or {} |
| if isinstance(gate_src, dict): |
| if _truthy(gate_src.get("strong_full_inference_success")): |
| return True |
| if _classify_build_status_text(_source_status_text(gate_src)) == "full_inference_success" and _truthy(_nested_get(gate_src, "implementation_signals", "generation_smoke_passed")): |
| return True |
|
|
| smoke_src = smoke or {} |
| if isinstance(smoke_src, dict): |
| |
| |
| |
| endpoint = smoke_src.get("api_name") or smoke_src.get("endpoint") or smoke_src.get("primary_api_name") |
| if str(smoke_src.get("status") or "").strip().lower() == "success" and endpoint and _truthy(smoke_src.get("promise_fulfilled")): |
| return True |
|
|
| return False |
|
|
| def _canonical_build_status_from_sources( |
| *, |
| state: dict[str, Any] | None = None, |
| launch: dict[str, Any] | None = None, |
| summary_file: dict[str, Any] | None = None, |
| final_status_reconciliation: dict[str, Any] | None = None, |
| eval_record: dict[str, Any] | None = None, |
| smoke: dict[str, Any] | None = None, |
| gate: dict[str, Any] | None = None, |
| live_status: dict[str, Any] | None = None, |
| repair: dict[str, Any] | None = None, |
| manual_validation: dict[str, Any] | None = None, |
| eval_publish_status: dict[str, Any] | None = None, |
| inference_contract: dict[str, Any] | None = None, |
| demo_quality_contract: dict[str, Any] | None = None, |
| events: list[dict[str, Any]] | None = None, |
| ) -> str: |
| """Return a canonical build status for explorer/list views. |
| |
| Bucket listings are intentionally light-weight and can read slightly stale |
| ``state.json`` before the final ``summary.json``/validation artifacts become |
| visible. Prefer explicit final summary/eval artifacts over stale partial |
| status so Run Explorer and Run Stats match the Active Run snapshot after a |
| page load. Intermediate incident events remain phase diagnostics only. |
| """ |
| runtime_success_authoritative = _has_authoritative_runtime_success( |
| final_status_reconciliation=final_status_reconciliation, |
| summary_file=summary_file, |
| state=state, |
| live_status=live_status, |
| gate=gate, |
| smoke=smoke, |
| ) |
| contract_blocks_success = False if runtime_success_authoritative else _contract_blocks_success(summary_file, state, gate, inference_contract, demo_quality_contract) |
| |
| |
| |
| |
| |
| |
| |
| authoritative_status = _explicit_effective_status(final_status_reconciliation, eval_record, summary_file) |
| if authoritative_status: |
| if authoritative_status == "full_inference_success" and contract_blocks_success: |
| if _contract_blocks_success(gate, inference_contract, demo_quality_contract): |
| return "manual_hardware_required" if "manual" in _source_status_text(gate or {}, inference_contract or {}, demo_quality_contract or {}) else "technical_blocker_boot_only" |
| return authoritative_status |
|
|
| effective_status = _explicit_effective_status(state) |
| if effective_status: |
| if effective_status == "full_inference_success" and contract_blocks_success: |
| if _contract_blocks_success(gate, inference_contract, demo_quality_contract): |
| return "manual_hardware_required" if "manual" in _source_status_text(gate or {}, inference_contract or {}, demo_quality_contract or {}) else "technical_blocker_boot_only" |
| return effective_status |
|
|
| event_status = _terminal_status_from_events(events or [], validation=False) |
| if event_status: |
| return event_status |
|
|
| summary_status = _classify_build_status_text(_source_status_text(summary_file or {})) |
| state_status_for_guard = _classify_build_status_text(_source_status_text(state or {}, repair or {})) |
| terminal_text = _source_status_text(gate or {}, smoke or {}, manual_validation or {}) |
| terminal_status = _classify_build_status_text(terminal_text) |
|
|
| |
| |
| if summary_status == "full_inference_success": |
| if contract_blocks_success: |
| return "manual_hardware_required" if "manual" in _source_status_text(gate or {}, inference_contract or {}, demo_quality_contract or {}) else "technical_blocker_boot_only" |
| return "full_inference_success" |
| if summary_status == "partial_validation": |
| return "partial_validation" |
|
|
| |
| |
| if state_status_for_guard == "partial_validation": |
| return "partial_validation" |
|
|
| |
| |
| for hard_status in (summary_status, state_status_for_guard): |
| if hard_status and hard_status not in {"running", "full_inference_success", "partial_validation"}: |
| return hard_status |
|
|
| |
| |
| if terminal_status and terminal_status not in {"running", "full_inference_success", "partial_validation"}: |
| return terminal_status |
|
|
| if terminal_status == "partial_validation": |
| return "partial_validation" |
|
|
| if _has_persisted_generation_success(summary_file, smoke, gate, manual_validation, state): |
| if contract_blocks_success: |
| return "manual_hardware_required" if "manual" in _source_status_text(gate or {}, inference_contract or {}, demo_quality_contract or {}) else "technical_blocker_boot_only" |
| return "full_inference_success" |
|
|
| if summary_status and summary_status != "running": |
| return summary_status |
|
|
| if terminal_status and terminal_status != "running": |
| return terminal_status |
|
|
| state_status = _classify_build_status_text(_source_status_text(state or {}, repair or {})) |
| if state_status: |
| return state_status |
|
|
| summary_status = _classify_build_status_text(_source_status_text(summary_file or {})) |
| if summary_status: |
| return summary_status |
|
|
| launch_status = _classify_build_status_text(_source_status_text(launch or {})) |
| if launch_status: |
| return launch_status |
|
|
| for src in (state or {}, launch or {}, summary_file or {}): |
| for key in ("status", "verdict", "final_status"): |
| value = str(src.get(key) or "").lower() if isinstance(src, dict) else "" |
| if value: |
| return value |
| return "unknown" |
|
|
| def _canonical_validation_status_from_sources( |
| *, |
| state: dict[str, Any] | None = None, |
| launch: dict[str, Any] | None = None, |
| summary_file: dict[str, Any] | None = None, |
| smoke: dict[str, Any] | None = None, |
| gate: dict[str, Any] | None = None, |
| events: list[dict[str, Any]] | None = None, |
| ) -> str: |
| """Return a persisted/canonical validation status. |
| |
| Validation runs can have stale launch/summary metadata that still says |
| ``running`` after the worker has written a terminal smoke artifact or |
| failure event. Terminal evidence must always win over non-terminal launch |
| metadata so page reloads, delete modals, and the Run Explorer do not |
| resurrect already-finished validations. |
| """ |
| event_status = _terminal_status_from_events(events or [], validation=True) |
| if event_status: |
| return event_status |
|
|
| chunks: list[str] = [] |
| for src in (state or {}, smoke or {}, gate or {}, summary_file or {}): |
| for key in ("verdict", "validation_status", "result_status", "status", "gate_status", "smoke_status"): |
| value = src.get(key) if isinstance(src, dict) else None |
| if value is not None: |
| chunks.append(str(value).lower()) |
| err = src.get("error") if isinstance(src, dict) else None |
| if err: |
| chunks.append(str(err).lower()) |
| text = " ".join(chunks) |
| if any(x in text for x in ("manual", "requires_action", "hardware_required", "blocked")): |
| return "manual_hardware_required" |
| if any(x in text for x in ("failed", "failure", "error", "timeout")): |
| return "failed" |
| if any(x in text for x in ("partial", "health_only", "not_verified", "completed_with_warnings", "warning")): |
| return "partial_validation" |
| if any(x in text for x in ("full_inference_success", "success", "succeed", "passed", "completed")): |
| return "full_inference_success" |
| if any(x in text for x in ("stale",)): |
| return "stale" |
| if any(x in text for x in ("cancelled", "canceled", "stopped")): |
| return "stopped" |
|
|
| for src in (state or {}, launch or {}, summary_file or {}): |
| for key in ("status", "validation_status", "result_status", "verdict"): |
| value = str(src.get(key) or "").lower() if isinstance(src, dict) else "" |
| if value: |
| return value |
| return "unknown" |
|
|
| 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 {} |
| post_build_validation = bundle.get("post_build_validation_status") or {} |
| manual_validation = post_build_validation or bundle.get("manual_validation_status") or {} |
| linked_validations = bundle.get("linked_validations") or {} |
| linked_rows = linked_validations.get("validations") if isinstance(linked_validations, dict) else [] |
| linked_rows_list = [row for row in linked_rows if isinstance(row, dict)] if isinstance(linked_rows, list) else [] |
| linked_terminal_statuses = { |
| "success", |
| "passed", |
| "succeeded", |
| "full_inference_success", |
| "validated_after_space_test", |
| "validated_after_manual_space_test", |
| "recovered_by_space_test", |
| "recovered_by_manual_validation", |
| "manual_validation_passed", |
| "manual_validated", |
| "partial_validation", |
| "manual_hardware_required", |
| "generated_needs_manual_hardware", |
| "technical_blocker", |
| "technical_blocker_boot_only", |
| "blocked", |
| "completed_with_warnings", |
| "failed", |
| "failure", |
| "error", |
| "auth_refresh_required", |
| "stale", |
| "stopped", |
| "cancelled", |
| "canceled", |
| } |
| linked_success_statuses = { |
| "success", |
| "passed", |
| "succeeded", |
| "full_inference_success", |
| "validated_after_space_test", |
| "validated_after_manual_space_test", |
| "recovered_by_space_test", |
| "recovered_by_manual_validation", |
| "manual_validation_passed", |
| "manual_validated", |
| } |
| linked_accounted_rows = [row for row in linked_rows_list if str(row.get("status") or row.get("validation_status") or row.get("result_status") or row.get("effective_status") or "").lower() in linked_terminal_statuses] |
| if linked_rows_list: |
| linked_successes = [row for row in linked_rows_list if str(row.get("status") or row.get("effective_status") or "").lower() in linked_success_statuses] |
| if linked_successes and str(manual_validation.get("status") or "").lower() != "success": |
| linked_successes.sort(key=lambda row: str(row.get("validated_at") or row.get("updated_at") or row.get("created_at") or ""), reverse=True) |
| manual_validation = linked_successes[0] |
| gate_smoke = gate.get("generation_smoke") if isinstance(gate.get("generation_smoke"), dict) else {} |
| zero_gpu_recommendation = gate.get("zero_gpu_duration_recommendation") if isinstance(gate, dict) else None |
| if isinstance(zero_gpu_recommendation, dict): |
| zero_gpu_recommendation = zero_gpu_recommendation.get("recommended_zero_gpu_duration_seconds") or zero_gpu_recommendation.get("recommended_zerogpu_duration_seconds") or zero_gpu_recommendation.get("seconds") |
| eval_publish_status = bundle.get("eval_publish_status") or {} |
| final_status_reconciliation = bundle.get("final_status_reconciliation") or {} |
| eval_record = bundle.get("eval_record") or {} |
| target_space = state.get("target_space") or launch.get("target_space") or summary_file.get("target_space") or "" |
| kind = state.get("kind") or launch.get("kind") or summary_file.get("kind") or "unknown" |
| is_validation = "validate" in str(kind).lower() or str(run_id).startswith("validate-") |
| if is_validation: |
| status = _canonical_validation_status_from_sources( |
| state=state, |
| launch=launch, |
| summary_file=summary_file, |
| smoke=smoke, |
| gate=gate, |
| events=bundle.get("events") or [], |
| ) |
| else: |
| |
| |
| status = _canonical_build_status_from_sources( |
| state=state, |
| launch=launch, |
| summary_file=summary_file, |
| final_status_reconciliation=final_status_reconciliation, |
| eval_record=eval_record, |
| smoke=smoke or bundle.get("test_result") or {}, |
| gate=gate, |
| live_status=bundle.get("live_status") or {}, |
| repair=bundle.get("repair_outcome") or {}, |
| manual_validation=manual_validation, |
| eval_publish_status=bundle.get("eval_publish_status") or {}, |
| inference_contract=bundle.get("inference_contract") or {}, |
| demo_quality_contract=bundle.get("demo_quality_contract") or {}, |
| events=bundle.get("events") or [], |
| ) |
| terminal = status in {"failed", "full_inference_success", "partial_validation", "manual_hardware_required", "stale", "stopped"} |
| effective_run_status = compute_effective_run_status({**bundle, "summary": {**summary_file, "status": status}}, build_status=status) |
| return { |
| "run_id": run_id, |
| "kind": kind, |
| "status": status, |
| "validation_status": status if is_validation else "", |
| "result_status": status if is_validation else "", |
| "terminal": bool(is_validation and terminal), |
| "can_resume": not bool(is_validation and terminal), |
| "model_id": state.get("model_id") or launch.get("model_id") or summary_file.get("model_id") or state.get("model") or "", |
| "pi_model": state.get("pi_model") or launch.get("pi_model") or summary_file.get("pi_model") or (bundle.get("pi_model_resolution") or {}).get("requested_model") or "", |
| "target_space": target_space, |
| "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": _job_url_from_sources(bucket_source=bucket_source, summary_file=summary_file, launch=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") or gate_smoke.get("health_passed")), |
| "smoke_test_passed": bool(smoke.get("ok") or smoke.get("status") == "success" or gate_smoke.get("status") == "success"), |
| "latency_seconds": manual_validation.get("latency_seconds") or manual_validation.get("observed_latency_seconds") or smoke.get("latency_seconds") or smoke.get("observed_latency_seconds") or gate_smoke.get("latency_seconds") or gate_smoke.get("observed_latency_seconds"), |
| "observed_latency_seconds": manual_validation.get("observed_latency_seconds") or manual_validation.get("latency_seconds") or smoke.get("observed_latency_seconds") or smoke.get("latency_seconds") or gate_smoke.get("observed_latency_seconds") or gate_smoke.get("latency_seconds"), |
| "recommended_zero_gpu_duration_seconds": manual_validation.get("recommended_zero_gpu_duration_seconds") or manual_validation.get("recommended_zerogpu_duration_seconds") or smoke.get("recommended_zero_gpu_duration_seconds") or smoke.get("recommended_zerogpu_duration_seconds") or gate_smoke.get("recommended_zero_gpu_duration_seconds") or gate_smoke.get("recommended_zerogpu_duration_seconds") or zero_gpu_recommendation, |
| "recommended_zerogpu_duration_seconds": manual_validation.get("recommended_zerogpu_duration_seconds") or manual_validation.get("recommended_zero_gpu_duration_seconds") or smoke.get("recommended_zerogpu_duration_seconds") or smoke.get("recommended_zero_gpu_duration_seconds") or gate_smoke.get("recommended_zerogpu_duration_seconds") or gate_smoke.get("recommended_zero_gpu_duration_seconds") or zero_gpu_recommendation, |
| "recommendation_source": manual_validation.get("recommendation_source") or smoke.get("recommendation_source") or gate_smoke.get("recommendation_source") or ("linked_space_test" if str(manual_validation.get("status") or "").lower() == "success" else ""), |
| "recommendation_hardware": manual_validation.get("recommendation_hardware") or manual_validation.get("hardware_used_for_validation") or smoke.get("recommendation_hardware") or gate_smoke.get("recommendation_hardware") or state.get("selected_hardware") or "", |
| "hardware_used_for_validation": manual_validation.get("hardware_used_for_validation") or manual_validation.get("recommendation_hardware") or state.get("selected_hardware") or "", |
| "manual_validation_status": manual_validation, |
| "manual_validation_passed": str(manual_validation.get("status") or manual_validation.get("effective_status") or "").lower() in linked_success_statuses, |
| "linked_validations": linked_validations if isinstance(linked_validations, dict) else {}, |
| "linked_validation_count": len(linked_accounted_rows), |
| "linked_validations_count": len(linked_accounted_rows), |
| "post_build_status": effective_run_status.get("post_build_status") or "none", |
| "post_build_validation": effective_run_status.get("post_build_validation") or {}, |
| "effective_run_status": effective_run_status, |
| "effective_status": effective_run_status.get("effective_status") or status, |
| "effective_verdict": effective_run_status.get("effective_verdict") or status, |
| "legacy_effective_status": effective_run_status.get("legacy_effective_status") or manual_validation.get("effective_status") or "", |
| "display_status": effective_run_status.get("display_status") or status, |
| "ui_status": effective_run_status.get("display_status") or state.get("ui_status") or gate.get("ui_status") or status, |
| "app_boot_validation_status": state.get("app_boot_validation_status") or gate.get("app_boot_validation_status") or summary_file.get("app_boot_validation_status") or "", |
| "promise_validation_status": state.get("promise_validation_status") or gate.get("promise_validation_status") or summary_file.get("promise_validation_status") or "", |
| "promise_fulfilled": bool(state.get("promise_fulfilled") or gate.get("promise_fulfilled") or summary_file.get("promise_fulfilled")), |
| "ui_badge": state.get("ui_badge") or gate.get("ui_badge") or summary_file.get("ui_badge") or "", |
| "expected_output_type": smoke.get("expected_output_type") or state.get("expected_output_type") or launch.get("expected_output_type") or "", |
| "pi_model_resolution": bundle.get("pi_model_resolution") or state.get("pi_model_resolution") or {}, |
| "artifacts_url": f"https://huggingface.co/buckets/{bucket_source}/tree/{(settings.bucket_runs_prefix.strip().strip('/') or 'runs')}/{run_id}", |
| } |
|
|
|
|
| def read_run_bundle(run_id: str, *, bucket_source: str, token: str | None = None, include_heavy: bool = True) -> dict[str, Any]: |
| paths = RunPaths(run_id, bucket_source=bucket_source) |
| bundle = { |
| "paths": { |
| "root": paths.root, |
| "state": paths.state, |
| "events": paths.events, |
| "report": paths.report, |
| }, |
| "summary_file": _safe_read_json(f"{paths.root}/summary.json", token=token), |
| "final_status_reconciliation": _safe_read_json(f"{paths.root}/final_status_reconciliation.json", token=token), |
| "eval_record": _safe_read_json(f"{paths.root}/eval_record.json", token=token), |
| "launch": _safe_read_json(f"{paths.root}/launch.json", token=token), |
| "state": read_json(paths.state, token=token) or {}, |
| "live_status": _safe_read_json(f"{paths.root}/live_status.json", token=token), |
| "events": read_events(run_id, bucket_source=bucket_source, token=token), |
| "report": _safe_read_text(paths.report, token=token) if include_heavy else "", |
| "inference_gate": _safe_read_json(f"{paths.root}/inference_gate.json", token=token), |
| "generation_smoke": _safe_read_json(f"{paths.root}/tests/generation_smoke.json", token=token) or _safe_read_json(f"{paths.root}/generation_smoke.json", token=token), |
| "test_result": _safe_read_json(f"{paths.root}/tests/test_result.json", token=token) or _safe_read_json(f"{paths.root}/test_result.json", token=token), |
| "generation_smoke_payload": _safe_read_json(f"{paths.root}/tests/generation_smoke_payload.json", token=token), |
| "generation_smoke_payload_retry": _safe_read_json(f"{paths.root}/tests/generation_smoke_payload_retry.json", token=token), |
| "resolved_validation_request": _safe_read_json(f"{paths.root}/tests/resolved_validation_request.json", token=token), |
| "payload_source": _safe_read_json(f"{paths.root}/tests/payload_source.json", token=token), |
| "validation_engine": _safe_read_json(f"{paths.root}/tests/validation_engine.json", token=token), |
| "validation_failure_diagnosis": _safe_read_json(f"{paths.root}/tests/validation_failure_diagnosis.json", token=token), |
| "hardware_strategy": _safe_read_json(f"{paths.root}/hardware_strategy.json", token=token), |
| "hardware_attempts": _safe_read_json(f"{paths.root}/hardware_attempts.json", token=token), |
| "technical_blockers": _safe_read_json(f"{paths.root}/generated/TECHNICAL_BLOCKERS.json", token=token) or _safe_read_json(f"{paths.root}/TECHNICAL_BLOCKERS.json", token=token), |
| "inference_contract": _safe_read_json(f"{paths.root}/generated/INFERENCE_CONTRACT.json", token=token) or _safe_read_json(f"{paths.root}/INFERENCE_CONTRACT.json", token=token), |
| "demo_quality_contract": _safe_read_json(f"{paths.root}/generated/DEMO_QUALITY_CONTRACT.json", token=token) or _safe_read_json(f"{paths.root}/DEMO_QUALITY_CONTRACT.json", token=token), |
| "repair_outcome": _safe_read_json(f"{paths.root}/repair_outcome.json", token=token) or _safe_read_json(f"{paths.root}/repair/REPAIR_OUTCOME.json", token=token), |
| "model_analysis": _safe_read_json(f"{paths.root}/model_analysis.json", token=token), |
| "pi_model_resolution": _safe_read_json(f"{paths.root}/pi_model_resolution.json", token=token) or (read_json(paths.state, token=token) or {}).get("pi_model_resolution") or {}, |
| "space_runtime": _safe_read_json(f"{paths.root}/space_runtime.json", token=token), |
| "runtime_upload_epoch": _safe_read_json(f"{paths.root}/runtime_upload_epoch.json", token=token), |
| "placeholder_scaffold_detection": _safe_read_json(f"{paths.root}/placeholder_scaffold_detection.json", token=token), |
| "api_schema": _safe_read_json(f"{paths.root}/tests/api_schema.json", token=token) or _safe_read_json(f"{paths.root}/tests/generation_api_schema.json", token=token), |
| "validation_payload": _safe_read_json(f"{paths.root}/tests/validation_payload.json", token=token), |
| "post_build_validation_status": _safe_read_json(f"{paths.root}/post_build_validation_status.json", token=token), |
| "manual_validation_status": _safe_read_json(f"{paths.root}/manual_validation_status.json", token=token), |
| "linked_validations": _safe_read_json(f"{paths.root}/linked_validations.json", token=token), |
| "eval_publish_status": _safe_read_json(f"{paths.root}/eval_publish_status.json", token=token), |
| "build_error_observation": _safe_read_json(f"{paths.root}/build_error_observation.json", token=token), |
| "repair_decision": _safe_read_json(f"{paths.root}/repair/REPAIR_DECISION.json", token=token), |
| "blockage": _safe_read_json(f"{paths.root}/repair/BLOCKAGE.json", token=token), |
| "artifact_manifest": _safe_read_json(f"{paths.root}/artifact_manifest.json", token=token), |
| "files": _list_run_files(run_id, bucket_source=bucket_source, token=token) if include_heavy else [], |
| } |
| if not bundle.get("generation_smoke") and isinstance((bundle.get("inference_gate") or {}).get("generation_smoke"), dict): |
| bundle["generation_smoke"] = (bundle.get("inference_gate") or {}).get("generation_smoke") or {} |
| bundle["run_documents"] = _run_document_links(run_id, bucket_source=bucket_source, bundle=bundle, token=token) |
| if include_heavy and not bundle.get("pi_model_resolution"): |
| try: |
| bundle["pi_model_resolution"] = _trace_based_pi_model_resolution(run_id, bundle, bucket_source=bucket_source, token=token) |
| except Exception: |
| bundle["pi_model_resolution"] = {} |
| bundle["summary"] = summarize_run_bundle(run_id, bundle, bucket_source=bucket_source) |
| return bundle |
|
|
|
|
| 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: |
| marker = f"/{_runs_prefix()}/" |
| if marker in path: |
| tail = path.split(marker, 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, |
| token: str | None = None, |
| limit: int = 50, |
| query: str | None = None, |
| status: str | None = None, |
| ) -> list[dict[str, Any]]: |
| """List recent run summaries from a user's bucket. |
| |
| 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. |
| """ |
| root = f"{bucket_uri_from_source(bucket_source)}/{_runs_prefix()}" |
| run_ids = _discover_run_ids(root, token=token) |
| if not run_ids: |
| return [] |
|
|
| runs: list[dict[str, Any]] = [] |
| 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 {} |
| test_result = read_json(f"{root}/{run_id}/tests/test_result.json", token=token) or read_json(f"{root}/{run_id}/test_result.json", token=token) or {} |
| if not smoke and isinstance(gate.get("generation_smoke"), dict): |
| smoke = gate.get("generation_smoke") or {} |
| hardware = read_json(f"{root}/{run_id}/hardware_strategy.json", token=token) or {} |
| pi_model_resolution = read_json(f"{root}/{run_id}/pi_model_resolution.json", token=token) or state.get("pi_model_resolution") or {} |
| post_build_validation_status = read_json(f"{root}/{run_id}/post_build_validation_status.json", token=token) or {} |
| manual_validation_status = read_json(f"{root}/{run_id}/manual_validation_status.json", token=token) or {} |
| linked_validations = read_json(f"{root}/{run_id}/linked_validations.json", token=token) or {} |
| eval_publish_status = read_json(f"{root}/{run_id}/eval_publish_status.json", token=token) or {} |
| kind_probe = state.get("kind") or launch.get("kind") or summary_file.get("kind") or "" |
| events = read_events(run_id, bucket_source=bucket_source, token=token) if ("validate" in str(kind_probe).lower() or str(run_id).startswith("validate-")) else [] |
| partial_bundle = { |
| "summary_file": summary_file, |
| "launch": launch, |
| "state": state, |
| "inference_gate": gate, |
| "generation_smoke": smoke, |
| "test_result": test_result, |
| "hardware_strategy": hardware, |
| "pi_model_resolution": pi_model_resolution, |
| "post_build_validation_status": post_build_validation_status, |
| "manual_validation_status": manual_validation_status, |
| "linked_validations": linked_validations, |
| "eval_publish_status": eval_publish_status, |
| "events": events, |
| } |
| item = summarize_run_bundle(run_id, partial_bundle, bucket_source=bucket_source) |
| if post_build_validation_status: |
| item["post_build_validation_status"] = post_build_validation_status |
| item["effective_status"] = post_build_validation_status.get("effective_status") or item.get("effective_status") |
| if linked_validations: |
| rows = linked_validations.get("validations") if isinstance(linked_validations, dict) else [] |
| terminal_tokens = {"success", "passed", "succeeded", "full_inference_success", "validated_after_space_test", "validated_after_manual_space_test", "recovered_by_space_test", "recovered_by_manual_validation", "manual_validation_passed", "manual_validated", "partial_validation", "manual_hardware_required", "generated_needs_manual_hardware", "technical_blocker", "technical_blocker_boot_only", "blocked", "completed_with_warnings", "failed", "failure", "error", "auth_refresh_required", "stale", "stopped", "cancelled", "canceled"} |
| accounted_rows = [row for row in rows if isinstance(row, dict) and str(row.get("status") or row.get("validation_status") or row.get("result_status") or row.get("effective_status") or "").lower() in terminal_tokens] if isinstance(rows, list) else [] |
| item["linked_validations"] = linked_validations |
| item["linked_validation_count"] = len(accounted_rows) |
| item["linked_validations_count"] = len(accounted_rows) |
| if manual_validation_status: |
| item["manual_validation_status"] = manual_validation_status |
| manual_status_token = str(manual_validation_status.get("status") or manual_validation_status.get("effective_status") or "").lower() |
| if manual_status_token not in {"", "none", "unknown", "unchanged"}: |
| item["effective_status"] = manual_validation_status.get("effective_status") or item.get("effective_status") |
| if manual_validation_status.get("status") == "success": |
| item["manual_validation_passed"] = True |
| item["latency_seconds"] = item.get("latency_seconds") or manual_validation_status.get("latency_seconds") |
| item["observed_latency_seconds"] = item.get("observed_latency_seconds") or manual_validation_status.get("observed_latency_seconds") |
| item["recommended_zero_gpu_duration_seconds"] = item.get("recommended_zero_gpu_duration_seconds") or manual_validation_status.get("recommended_zero_gpu_duration_seconds") |
| item["recommendation_source"] = item.get("recommendation_source") or manual_validation_status.get("recommendation_source") |
| item["recommendation_hardware"] = item.get("recommendation_hardware") or manual_validation_status.get("recommendation_hardware") or manual_validation_status.get("hardware_used_for_validation") |
| item["hardware_used_for_validation"] = item.get("hardware_used_for_validation") or manual_validation_status.get("hardware_used_for_validation") |
| item["recommendation_source"] = item.get("recommendation_source") or "linked_space_test" |
| item["bucket_source"] = bucket_source |
| 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: |
| continue |
| runs.append(item) |
|
|
| runs.sort(key=lambda r: str(r.get("updated_at") or r.get("created_at") or r.get("run_id")), reverse=True) |
| return runs[: max(1, min(int(limit or 50), 200))] |
|
|