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 .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: # noqa: BLE001 - report readable status in UI 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: # noqa: BLE001 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: # noqa: BLE001 - surface readable error in UI 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///...`` 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//``. """ 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: # noqa: BLE001 - keep trying supported bucket writers attempts.append(f"write_text({candidate!r}): {exc}") try: fs.pipe_file(candidate, data, mode="overwrite") return except Exception as exc: # noqa: BLE001 attempts.append(f"pipe_file({candidate!r}): {exc}") try: with fs.open(candidate, "wb") as f: f.write(data) return except Exception as exc: # noqa: BLE001 attempts.append(f"open-wb({candidate!r}): {exc}") try: _sync_bucket_single_text(path, content, token=token) return except Exception as exc: # noqa: BLE001 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() # Best source when available: recursive find with fresh listing. 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 # Glob fallback catches buckets where find is incomplete or disabled. 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 # Manual recursive ls fallback. Keep it iterative to avoid recursion loops # if the remote returns the same virtual prefix multiple times. 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//`` 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: # noqa: BLE001 errors.append(f"{path}: {redact(str(exc))}") # Keep the old recursive call as a final cleanup for any filesystem that # does support virtual directory removal, but never rely on it as the only # deletion mechanism. try: fs.rm(root, recursive=True) except FileNotFoundError: pass except Exception as exc: # noqa: BLE001 # If explicit file deletion succeeded, a virtual-directory cleanup error # should not make the whole operation fail. It is useful diagnostic # metadata though, so report it when nothing was deleted. 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 "", "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) # 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]]: 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: # Backward-compatible wrapper for older call sites/tests. 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 to the stable trace folders: - Pi RAW โ†’ `traces/raw` - Pi redacted โ†’ `traces/redacted` They become active only when at least one real trace file/folder exists underneath. Pi commonly writes `traces/raw/--tmp-universal_workspace--/...` and `traces/redacted/--tmp-universal_workspace--/...`; the stable folder URL lets the user browse the exact session folder without us guessing the file. """ 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")} 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('/')}" # _folder_has_content probes direct children and recursive **/* patterns via _fs(token).glob(pattern). return _folder_has_content(root_prefix, token=token) raw_present = trace_folder_has_content("traces/raw") redacted_present = trace_folder_has_content("traces/redacted") raw_trace_url = _bucket_tree_url(bucket_source, run_id, "traces/raw") redacted_trace_url = _bucket_tree_url(bucket_source, run_id, "traces/redacted") smoke_path = first_existing_path("tests/generation_smoke.json", "generation_smoke.json") blockers_path = first_existing_path("generated/TECHNICAL_BLOCKERS.json", "TECHNICAL_BLOCKERS.json") smoke_present = bool(bundle.get("generation_smoke")) or manifest_has("tests/generation_smoke.json") or manifest_has("generation_smoke.json") or "tests/generation_smoke.json" in file_paths or "generation_smoke.json" in file_paths or smoke_path in file_paths report_present = 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)) 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)) 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_raw_trace", "label": "Pi RAW", "subtitle": "Unified build/diagnosis/repair trace", "icon": "๐Ÿงพ", "present": raw_present, "url": raw_trace_url, "sensitivity": "raw", }, { "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": "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", "icon": "โ–ฃ", "present": space_logs_index_present, "url": _bucket_file_url(bucket_source, run_id, "logs/space_logs_index.json"), "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_raw_trace": 7, "pi_redacted_trace": 8, "smoke": 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}/traces/raw/**/*.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 _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 {} 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: event_status = _terminal_status_from_events(bundle.get("events") or [], validation=False) status = event_status or state.get("status") or launch.get("status") or summary_file.get("status") or gate.get("status") or smoke.get("status") or "unknown" terminal = status in {"failed", "full_inference_success", "partial_validation", "manual_hardware_required", "stale", "stopped"} 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")), "smoke_test_passed": bool(smoke.get("ok") or smoke.get("status") == "success"), "latency_seconds": smoke.get("latency_seconds") or smoke.get("observed_latency_seconds"), "observed_latency_seconds": smoke.get("observed_latency_seconds") or smoke.get("latency_seconds"), "recommended_zero_gpu_duration_seconds": smoke.get("recommended_zero_gpu_duration_seconds") or smoke.get("recommended_zerogpu_duration_seconds"), "recommended_zerogpu_duration_seconds": smoke.get("recommended_zerogpu_duration_seconds") or smoke.get("recommended_zero_gpu_duration_seconds"), "recommendation_source": smoke.get("recommendation_source") 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), "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), "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), "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), "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), "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), "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 [], } 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 {} 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 {} 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, "hardware_strategy": hardware, "pi_model_resolution": pi_model_resolution, "events": events, } 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", "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))]