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 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), }, ] 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", } ) 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 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 "" status = state.get("status") or launch.get("status") or summary_file.get("status") or gate.get("status") or smoke.get("status") or "unknown" return { "run_id": run_id, "kind": state.get("kind") or launch.get("kind") or summary_file.get("kind") or "unknown", "status": status, "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), "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 {} 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, } 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))]