src/bucket.py

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/<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:  # 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/<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:  # 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 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),
        "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 {}
        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))]