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 .effective_status import compute_effective_run_status
from .security import redact


@dataclass(frozen=True)
class RunPaths:
    run_id: str
    bucket_source: str

    @property
    def bucket_uri(self) -> str:
        return bucket_uri_from_source(self.bucket_source)

    @property
    def root(self) -> str:
        return f"{self.bucket_uri}/{_runs_prefix()}/{self.run_id}"

    @property
    def state(self) -> str:
        return f"{self.root}/state.json"

    @property
    def events(self) -> str:
        return f"{self.root}/events.jsonl"

    @property
    def report(self) -> str:
        return f"{self.root}/report.md"



def _runs_prefix() -> str:
    return settings.bucket_runs_prefix.strip().strip("/") or "runs"


def _run_tree_url(bucket_source: str, run_id: str, rel_path: str = "") -> str:
    suffix = f"/{rel_path.strip('/')}" if rel_path else ""
    return f"https://huggingface.co/buckets/{bucket_source}/tree/{_runs_prefix()}/{run_id}{suffix}"


def _fs(token: str | None = None) -> HfFileSystem:
    return HfFileSystem(token=token)


def check_user_bucket(*, username: str, bucket_name: str | None = None, token: str | None = None) -> dict[str, Any]:
    """Return bucket readiness for the signed-in user without creating resources."""
    bucket_source = user_bucket_source(username=username, bucket_name=bucket_name)
    bucket_uri = bucket_uri_from_source(bucket_source)
    try:
        info = bucket_info(bucket_source, token=token)
        return {
            "ok": True,
            "exists": True,
            "bucket_source": bucket_source,
            "bucket_uri": bucket_uri,
            "name": getattr(info, "name", normalize_bucket_name(bucket_name)),
            "private": getattr(info, "private", None),
        }
    except Exception as exc:  # 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 "",
        "parent_build_run_id": launch.get("parent_build_run_id") or "",
        "source_kind": launch.get("source_kind") or "",
        "linked_target_space": launch.get("linked_target_space") or "",
        "artifacts_url": f"https://huggingface.co/buckets/{bucket_source}/tree/{(settings.bucket_runs_prefix.strip().strip('/') or 'runs')}/{run_id}",
    }
    write_json(f"{paths.root}/summary.json", summary, token=token)
    # 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 only to the stable redacted trace folder:
    - Pi redacted → `traces/redacted`

    RAW Pi traces are intentionally not published to the bucket and are not
    exposed in the document dock. Redacted traces become active only when at
    least one real trace file/folder exists underneath.
    """
    paths = RunPaths(run_id, bucket_source=bucket_source)
    files = [f for f in (bundle.get("files") or []) if isinstance(f, dict)]
    file_paths = {str(f.get("path") or "") for f in files}
    manifest = bundle.get("artifact_manifest") or {}
    manifest_artifacts = [a for a in (manifest.get("artifacts") or []) if isinstance(a, dict)] if isinstance(manifest, dict) else []
    manifest_paths = {str(a.get("path") or "") for a in manifest_artifacts if a.get("present")}
    if isinstance(manifest, dict):
        manifest_paths.update(str(p or "") for p in (manifest.get("present_paths") or []) if p)

    terminal_success = str((bundle.get("final_status_reconciliation") or {}).get("status") or (bundle.get("summary") or {}).get("status") or (bundle.get("summary_file") or {}).get("status") or (bundle.get("state") or {}).get("status") or "").strip().lower() in {"full_inference_success", "success", "succeeded", "completed"}

    def manifest_has(path: str) -> bool:
        path = path.strip("/")
        prefix = path.rstrip("/") + "/"
        return path in manifest_paths or any(p.startswith(prefix) for p in manifest_paths)

    def first_existing_path(*candidates: str) -> str:
        for candidate in candidates:
            clean = candidate.strip("/")
            if clean in file_paths or clean in manifest_paths:
                return clean
        if token:
            for candidate in candidates:
                clean = candidate.strip("/")
                if _path_exists(f"{paths.root}/{clean}", token=token):
                    return clean
        return candidates[0].strip("/") if candidates else ""

    def trace_folder_has_content(prefix: str) -> bool:
        rel_prefix = prefix.rstrip("/") + "/"
        if any(path.startswith(rel_prefix) and not path.endswith("/") for path in file_paths):
            return True
        if manifest_has(prefix):
            return True
        if not token:
            return False
        root_prefix = f"{paths.root}/{prefix.rstrip('/')}"
        # _folder_has_content probes direct children and recursive **/* patterns via _fs(token).glob(pattern).
        return _folder_has_content(root_prefix, token=token)

    redacted_present = trace_folder_has_content("traces/redacted")
    redacted_trace_url = _bucket_tree_url(bucket_source, run_id, "traces/redacted")
    smoke_path = first_existing_path(
        "tests/generation_smoke.json",
        "tests/generation_smoke_latest.json",
        "generation_smoke.json",
        "tests/test_result.json",
    )
    blockers_path = first_existing_path("generated/TECHNICAL_BLOCKERS.json", "TECHNICAL_BLOCKERS.json")
    smoke_present = (
        terminal_success
        or bool(bundle.get("generation_smoke"))
        or manifest_has("tests/generation_smoke.json")
        or manifest_has("tests/generation_smoke_latest.json")
        or manifest_has("generation_smoke.json")
        or "tests/generation_smoke.json" in file_paths
        or "tests/generation_smoke_latest.json" in file_paths
        or "generation_smoke.json" in file_paths
        or smoke_path in file_paths
        or smoke_path in manifest_paths
    )
    report_present = terminal_success or bool(bundle.get("report")) or manifest_has("report.md") or "report.md" in file_paths or (bool(token) and _path_exists(paths.report, token=token))
    blockers_present = bool(bundle.get("technical_blockers")) or manifest_has("generated/TECHNICAL_BLOCKERS.json") or manifest_has("TECHNICAL_BLOCKERS.json") or "generated/TECHNICAL_BLOCKERS.json" in file_paths or "TECHNICAL_BLOCKERS.json" in file_paths or blockers_path in file_paths
    eval_publish_payload = bundle.get("eval_publish") if isinstance(bundle.get("eval_publish"), dict) else bundle.get("eval_publish_status") if isinstance(bundle.get("eval_publish_status"), dict) else {}
    eval_publish_present = bool(eval_publish_payload) or manifest_has("eval_publish_status.json") or "eval_publish_status.json" in file_paths or (bool(token) and _path_exists(f"{paths.root}/eval_publish_status.json", token=token))
    useful_space_log_candidates = (
        "logs/space_logs_index.json",
        "logs/space_logs_build.txt",
        "logs/space_logs_runtime.txt",
        "logs/space_logs_run.txt",
        "logs/space_logs_fetch_status.json",
        "logs/space_log_diagnostics.json",
        "logs/space_runtime_snapshot.json",
        "logs/pi_live_output.txt",
        "logs/pi_output.txt",
        "logs/build_log_fast_probe_status.json",
    )
    space_logs_path = first_existing_path(*useful_space_log_candidates)
    space_logs_index_present = manifest_has("logs/space_logs_index.json") or "logs/space_logs_index.json" in file_paths or (bool(token) and _path_exists(f"{paths.root}/logs/space_logs_index.json", token=token))
    space_logs_present = (
        space_logs_index_present
        or manifest_has("logs")
        or any(path in file_paths or path in manifest_paths for path in useful_space_log_candidates)
        or any(path.startswith("logs/") and not path.endswith("/") for path in file_paths | manifest_paths)
        or (bool(token) and _folder_has_content(f"{paths.root}/logs", token=token))
    )
    if not space_logs_index_present and space_logs_present and space_logs_path == "logs/space_logs_index.json":
        space_logs_path = "logs"
    output_artifact_candidates = (
        "artifacts/image.webp",
        "artifacts/image.png",
        "artifacts/image.jpg",
        "artifacts/image.jpeg",
        "artifacts/output.png",
        "artifacts/output.webp",
        "artifacts/output.json",
    )
    output_artifact_path = first_existing_path(*output_artifact_candidates)
    output_artifact_present = (
        any(path in file_paths or path in manifest_paths for path in output_artifact_candidates)
        or manifest_has("artifacts")
        or (bool(token) and _folder_has_content(f"{paths.root}/artifacts", token=token))
    )
    if output_artifact_present and output_artifact_path == "artifacts/image.webp" and output_artifact_path not in file_paths and output_artifact_path not in manifest_paths and not manifest_has(output_artifact_path):
        output_artifact_path = "artifacts"
    build_error_present = manifest_has("build_error_observation.json") or "build_error_observation.json" in file_paths or (bool(token) and _path_exists(f"{paths.root}/build_error_observation.json", token=token))
    repair_decision_present = manifest_has("repair/REPAIR_DECISION.json") or "repair/REPAIR_DECISION.json" in file_paths or (bool(token) and _path_exists(f"{paths.root}/repair/REPAIR_DECISION.json", token=token))
    blockage_present = manifest_has("repair/BLOCKAGE.json") or "repair/BLOCKAGE.json" in file_paths or (bool(token) and _path_exists(f"{paths.root}/repair/BLOCKAGE.json", token=token))
    repair_present = manifest_has("repair") or any(path.startswith("repair/") for path in file_paths) or (bool(token) and _path_exists(f"{paths.root}/repair/REPAIR_SUMMARY.md", token=token))
    status = str((bundle.get("summary") or {}).get("status") or (bundle.get("state") or {}).get("status") or "").lower()
    blockers_relevant = blockers_present or any(marker in status for marker in ("manual", "blocker", "failed", "error"))

    docs = [
        {
            "id": "pi_redacted_trace",
            "label": "Pi redacted",
            "subtitle": "Unified safe agent trace",
            "icon": "🛡️",
            "present": redacted_present,
            "url": redacted_trace_url,
            "sensitivity": "safe",
        },
        {
            "id": "report",
            "label": "Report",
            "subtitle": "Human summary",
            "icon": "📄",
            "present": report_present,
            "url": _bucket_file_url(bucket_source, run_id, "report.md"),
        },
        {
            "id": "smoke",
            "label": "Smoke",
            "subtitle": "Validation + latency",
            "icon": "⚡",
            "present": smoke_present,
            "url": _bucket_file_url(bucket_source, run_id, smoke_path),
        },
        {
            "id": "smoke_output",
            "label": "Output",
            "subtitle": "Generated artifact",
            "icon": "🖼️",
            "present": output_artifact_present,
            "url": _bucket_file_url(bucket_source, run_id, output_artifact_path) if output_artifact_path != "artifacts" else _bucket_tree_url(bucket_source, run_id, "artifacts"),
        },
        {
            "id": "eval_publish",
            "label": "Eval publish",
            "subtitle": "Archive status",
            "icon": "◎",
            "present": eval_publish_present,
            "url": _bucket_file_url(bucket_source, run_id, "eval_publish_status.json"),
            "tone": "success" if eval_publish_payload.get("published") else "pending" if eval_publish_present and str(eval_publish_payload.get("reason") or "") == "record_not_ready" else "warn" if eval_publish_present else "neutral",
        },
        {
            "id": "space_logs",
            "label": "Space logs",
            "subtitle": "Build/runtime log index" if space_logs_index_present else "Build/runtime logs",
            "icon": "▣",
            "present": space_logs_present,
            "url": _bucket_file_url(bucket_source, run_id, space_logs_path) if space_logs_path != "logs" else _bucket_tree_url(bucket_source, run_id, "logs"),
            "tone": "warn" if "failed" in status or "error" in status else "neutral",
        },
        {
            "id": "build_error",
            "label": "Build error",
            "subtitle": "Observed build failure",
            "icon": "⛔",
            "present": build_error_present,
            "url": _bucket_file_url(bucket_source, run_id, "build_error_observation.json"),
            "tone": "warn",
        },
    ]
    if repair_decision_present:
        docs.append(
            {
                "id": "repair_decision",
                "label": "Decision",
                "subtitle": "Pi diagnosis action",
                "icon": "◇",
                "present": repair_decision_present,
                "url": _bucket_file_url(bucket_source, run_id, "repair/REPAIR_DECISION.json"),
                "tone": "warn" if "failed" in status or "error" in status else "neutral",
            }
        )
    if repair_present:
        docs.append(
            {
                "id": "repair",
                "label": "Repair",
                "subtitle": "Brief, plan and summary",
                "icon": "✦",
                "present": repair_present,
                "url": _bucket_tree_url(bucket_source, run_id, "repair"),
                "tone": "warn" if "failed" in status or "error" in status else "neutral",
            }
        )
    if blockage_present:
        docs.append(
            {
                "id": "blockage",
                "label": "Blockage",
                "subtitle": "Terminal recovery reason",
                "icon": "!",
                "present": blockage_present,
                "url": _bucket_file_url(bucket_source, run_id, "repair/BLOCKAGE.json"),
                "tone": "warn",
            }
        )
    if blockers_relevant:
        docs.append(
            {
                "id": "blockers",
                "label": "Blockers",
                "subtitle": "Why inference is blocked",
                "icon": "⚠️",
                "present": blockers_present,
                "url": _bucket_file_url(bucket_source, run_id, blockers_path),
                "tone": "warn",
            }
        )
    if "failed" in status or "error" in status:
        priority = {"report": 0, "blockage": 1, "build_error": 2, "space_logs": 3, "repair_decision": 4, "repair": 5, "blockers": 6, "pi_redacted_trace": 7, "smoke": 8, "smoke_output": 9, "eval_publish": 10}
        docs.sort(key=lambda item: priority.get(str(item.get("id") or ""), 50))
    return docs


def _normalize_model_name(value: str | None) -> str:
    raw = (value or "").strip().lower()
    if "/" in raw:
        raw = raw.rsplit("/", 1)[-1]
    return re.sub(r"[^a-z0-9]+", "", raw)


def _extract_pi_models_from_text(text: str) -> list[str]:
    if not text:
        return []
    patterns = [
        r"\bQwen/[A-Za-z0-9_.-]+",
        r"\bQwen(?:2(?:\.5)?|3)?[-_/A-Za-z0-9.]*Coder[-_/A-Za-z0-9.]*",
        r"\bmoonshotai/[A-Za-z0-9_.-]+",
        r"\bKimi[-_/A-Za-z0-9.]+",
        r"\bzai-org/[A-Za-z0-9_.-]+",
        r"\bGLM[-_/A-Za-z0-9.]+",
        r"\bdeepseek-ai/[A-Za-z0-9_.-]+",
        r"\bDeepSeek[-_/A-Za-z0-9.]+",
        r"\bClaude[-_/A-Za-z0-9.]+",
        r"\bGPT[-_/A-Za-z0-9.]+",
        r"\b[Mm]odel(?:\\s+used|\\s+selected|\\s*:)?\\s*[:=]?\\s*([A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+)",
    ]
    found: list[str] = []
    for pattern in patterns:
        for match in re.finditer(pattern, text, flags=re.IGNORECASE):
            value = match.group(1) if match.groups() else match.group(0)
            value = value.strip().strip('",` ')
            if value and value not in found:
                found.append(value)
    return found[:12]


def _trace_based_pi_model_resolution(run_id: str, bundle: dict[str, Any], *, bucket_source: str, token: str | None = None) -> dict[str, Any]:
    state = bundle.get("state") or {}
    launch = bundle.get("launch") or {}
    requested = (
        state.get("pi_model")
        or launch.get("pi_model")
        or launch.get("PI_MODEL")
        or state.get("pi_model_resolution", {}).get("requested_model")
        or ""
    )
    configured = state.get("pi_model_resolution", {}).get("configured_model") or requested
    root = RunPaths(run_id, bucket_source=bucket_source).root
    trace_texts: list[str] = []
    try:
        fs = _fs(token)
        for pattern in (
            f"{root}/traces/redacted/**/*.jsonl",
            f"{root}/traces/redacted/*.jsonl",
            f"{root}/logs/pi_output.txt",
            f"{root}/logs/pi_live_output.txt",
        ):
            for path in list(fs.glob(pattern))[:8]:
                try:
                    trace_texts.append(_safe_read_text(str(path), token=token)[:120000])
                except Exception:
                    pass
            if len(trace_texts) >= 8:
                break
    except Exception:
        pass
    observed = _extract_pi_models_from_text("\n".join(trace_texts))
    n_requested = _normalize_model_name(requested)
    n_configured = _normalize_model_name(configured)
    effective = ""
    for raw in observed:
        n_raw = _normalize_model_name(raw)
        if n_raw and n_raw not in {n_requested, n_configured}:
            effective = raw
            break
    if not effective and observed:
        effective = observed[0]
    mismatch = bool(effective and _normalize_model_name(effective) not in {n_requested, n_configured})
    if not requested and not observed:
        return {}
    return {
        "requested_model": requested,
        "configured_model": configured or requested,
        "observed_models": observed,
        "effective_model": effective or configured or requested,
        "provider": "huggingface",
        "mismatch": mismatch,
        "source": "published_pi_traces",
    }



def _terminal_status_from_events(events: list[dict[str, Any]] | None, *, validation: bool = False) -> str:
    """Return the strongest terminal status observed in events, newest first.

    Validation runs are short-lived and their terminal events are authoritative.
    Build runs, however, contain recoverable incident events (ZeroGPU fallback,
    transient Space boot failures, repair probes). For builds, only explicit
    final verdict events may decide the whole run status; intermediate failed
    events must remain phase diagnostics and never paint the active run red.
    """
    final_steps = {"done", "failure", "final", "report_write", "result", "run_result"}
    for event in reversed(events or []):
        step = str(event.get("step") or "").lower()
        status = str(event.get("status") or "").lower()
        message = str(event.get("message") or "").lower()
        haystack = " ".join([step, status, message])
        if validation:
            if any(x in haystack for x in ["manual", "requires_action", "hardware_required"]):
                return "manual_hardware_required"
            if any(x in haystack for x in ["failed", "failure", "error", "timeout"]):
                return "failed"
            if any(x in haystack for x in ["partial", "health_only", "not_verified", "completed_with_warnings"]):
                return "partial_validation"
            if step == "done" and ("full_inference_success" in haystack or "success" in haystack or "completed" in haystack):
                return "full_inference_success"
            if step in {"done", "report_write"} and status in {"success", "done", "completed"}:
                return "full_inference_success"
            continue

        if step not in final_steps:
            continue
        if any(x in haystack for x in ["manual", "requires_action", "hardware_required"]):
            return "manual_hardware_required"
        if any(x in haystack for x in ["partial", "health_only", "not_verified", "completed_with_warnings"]):
            return "partial_validation"
        if "full_inference_success" in haystack or (step == "done" and any(x in haystack for x in ["success", "completed", "passed"])):
            return "full_inference_success"
        if step in {"failure", "final", "run_result"} and any(x in haystack for x in ["failed", "failure", "error", "timeout"]):
            return "failed"
    return ""




def _source_status_text(*sources: dict[str, Any] | None) -> str:
    chunks: list[str] = []
    for src in sources:
        if not isinstance(src, dict):
            continue
        for key in (
            "final_status",
            "verdict",
            "status",
            "gate_status",
            "smoke_status",
            "validation_status",
            "result_status",
            "effective_status",
        ):
            value = src.get(key)
            if value is not None:
                chunks.append(str(value).lower())
        if src.get("ok") is True:
            chunks.append("ok true success")
        if src.get("manual_hardware_required") is True:
            chunks.append("manual_hardware_required")
        error = src.get("error")
        if error:
            chunks.append(str(error).lower())
    return " ".join(chunks)


def _classify_build_status_text(text: str) -> str:
    text = str(text or "").lower()
    if not text:
        return ""
    if "auth_refresh_required" in text or "oauth_expired" in text:
        return "auth_refresh_required"
    if any(x in text for x in ("manual_hardware_required", "manual hardware", "requires_action", "hardware_required")):
        return "manual_hardware_required"
    if "technical_blocker_boot_only" in text:
        return "technical_blocker_boot_only"
    if any(x in text for x in ("technical_blocker", "technical blocker", "blocked")):
        return "technical_blocker"
    if any(x in text for x in ("failed", "failure", "error", "timeout", "exception")):
        return "failed"
    if "stale" in text:
        return "stale"
    if any(x in text for x in ("cancelled", "canceled", "stopped")):
        return "stopped"
    if any(x in text for x in ("partial_validation", "full_inference_candidate_health_passed", "health_only", "demo_usable_full_promise_not_verified", "interactive_app_available_smoke_failed", "manual_test_required_smoke_failed", "not_verified", "completed_with_warnings")):
        return "partial_validation"
    if "full_inference_success" in text:
        return "full_inference_success"
    if any(x in text for x in ("success", "succeed", "passed", "completed")):
        return "full_inference_success"
    if any(x in text for x in ("running", "queued", "pending", "building", "waiting", "started")):
        return "running"
    return ""


def _explicit_effective_status(*sources: dict[str, Any] | None) -> str:
    """Return an authoritative display/eval status without reading nested errors.

    v198.8: Run Explorer cards must not turn a usable partial build red just
    because a nested generation smoke artifact says ``failed``.  The worker
    publishes effective verdict/status artifacts for exactly this reason; read
    those fields first and classify only their values, not the attached error
    text.
    """
    for src in sources:
        if not isinstance(src, dict):
            continue
        for key in ("ui_status", "display_status", "effective_verdict", "effective_status", "final_status", "verdict", "result_status", "validation_status", "status"):
            value = src.get(key)
            if value is None:
                continue
            status = _classify_build_status_text(str(value))
            if status and status != "running":
                return status
    return ""


def _truthy(value: Any) -> bool:
    if value is True:
        return True
    if isinstance(value, str):
        return value.strip().lower() in {"1", "true", "yes", "y", "success", "passed", "ok"}
    return bool(value) if isinstance(value, (int, float)) else False


def _nested_get(src: dict[str, Any] | None, *path: str) -> Any:
    cur: Any = src
    for key in path:
        if not isinstance(cur, dict):
            return None
        cur = cur.get(key)
    return cur



def _contract_blocks_success(*sources: dict[str, Any] | None) -> bool:
    """Return true when contracts/gate explicitly say this is diagnostic-only.

    v198.18: app boot or a diagnostic endpoint response must not hydrate a
    technical blocker into Success.  This guard is intentionally conservative:
    any explicit no-real-inference/manual/diagnostic signal wins over nested
    smoke success.
    """
    text_chunks: list[str] = []
    for src in sources:
        if not isinstance(src, dict):
            continue
        for key in (
            "promise_validation_status",
            "ui_status",
            "ui_badge",
            "status",
            "validation_level",
            "inference_strategy",
            "promise_fulfillment_risk",
            "reason",
            "message",
        ):
            value = src.get(key)
            if value is not None:
                text_chunks.append(str(value).lower())
        nested = src.get("promise_validation")
        if isinstance(nested, dict):
            for key in ("promise_validation_status", "ui_status", "ui_badge", "reason", "validation_level"):
                value = nested.get(key)
                if value is not None:
                    text_chunks.append(str(value).lower())
            if nested.get("promise_fulfilled") is False:
                text_chunks.append("promise_fulfilled_false")
        for bool_key in ("fallback_or_diagnostic_only", "manual_hardware_required"):
            if src.get(bool_key) is True:
                text_chunks.append(bool_key)
        for bool_key in ("full_inference_implemented", "real_inference_implemented"):
            if src.get(bool_key) is False:
                text_chunks.append(f"{bool_key}_false")
    text = " ".join(text_chunks)
    blockers = (
        "not_fulfilled_diagnostic_only",
        "deferred_manual_hardware",
        "diagnostic-only",
        "diagnostic_only",
        "diagnostic space",
        "manual_hardware_required",
        "manual hardware required",
        "manual_hardware_actionable",
        "technical_blocker_boot_only",
        "fallback_or_diagnostic_only",
        "full_inference_implemented_false",
        "real_inference_implemented_false",
        "official demo",
        "official space",
    )
    return any(marker in text for marker in blockers)

def _has_persisted_generation_success(*sources: dict[str, Any] | None) -> bool:
    """Return true for strong persisted proof that full generation worked.

    Older runs did not always store success in the same top-level field. The
    Active Run bundle can infer success from nested gate/smoke/test artifacts,
    while the Run Explorer list initially sees only lightweight JSON files. Keep
    the list and stats aligned by recognizing the same durable success signals
    without relying on transient UI state.
    """
    for src in sources:
        if not isinstance(src, dict):
            continue
        status_text = _source_status_text(src)
        if "full_inference_success" in status_text:
            return True
        if _truthy(src.get("ok")) or _truthy(src.get("generation_smoke_passed")) or _truthy(src.get("smoke_test_passed")):
            return True
        if _truthy(_nested_get(src, "implementation_signals", "generation_smoke_passed")):
            return True
        if _truthy(_nested_get(src, "validation", "generation_smoke_passed")):
            return True
        if _truthy(_nested_get(src, "outcome", "generation_smoke_passed")):
            return True
        for nested_key in ("generation_smoke", "smoke", "test_result", "result"):
            nested = src.get(nested_key)
            if isinstance(nested, dict):
                nested_status = str(nested.get("status") or nested.get("result_status") or nested.get("verdict") or "").lower()
                if nested_status in {"success", "passed", "full_inference_success"}:
                    return True
                if _truthy(nested.get("ok")) or _truthy(nested.get("passed")):
                    return True
    return False


def _has_authoritative_runtime_success(
    *,
    final_status_reconciliation: dict[str, Any] | None = None,
    summary_file: dict[str, Any] | None = None,
    state: dict[str, Any] | None = None,
    live_status: dict[str, Any] | None = None,
    gate: dict[str, Any] | None = None,
    smoke: dict[str, Any] | None = None,
) -> bool:
    """Return true when a final runtime proof supersedes stale Pi/contract blockers.

    v198.26.15: contract and Pi diagnostic signals are pre-runtime safety rails.
    They must still protect diagnostic-only demos, but they must not downgrade a
    run after the worker has published a final reconciliation/summary/state/live
    proof of ``full_inference_success`` or a strong inference gate.
    """
    def explicit_full_status(source: dict[str, Any]) -> bool:
        for key in ("status", "final_status", "ui_status", "display_status", "effective_status", "effective_verdict", "verdict"):
            if str(source.get(key) or "").strip().lower() == "full_inference_success":
                return True
        return False

    for source in (final_status_reconciliation or {}, summary_file or {}, state or {}):
        if not isinstance(source, dict):
            continue
        if explicit_full_status(source):
            return True
        if _truthy(source.get("generation_smoke_passed")) and _truthy(source.get("promise_fulfilled")):
            return True

    live = live_status or {}
    if isinstance(live, dict):
        live_status_text = _classify_build_status_text(_source_status_text(live))
        live_stage = str(live.get("stage") or "").strip().lower()
        if live_stage == "done" and live_status_text == "full_inference_success":
            return True
        if live_stage == "done" and _truthy(live.get("generation_smoke_passed")):
            return True

    gate_src = gate or {}
    if isinstance(gate_src, dict):
        if _truthy(gate_src.get("strong_full_inference_success")):
            return True
        if _classify_build_status_text(_source_status_text(gate_src)) == "full_inference_success" and _truthy(_nested_get(gate_src, "implementation_signals", "generation_smoke_passed")):
            return True

    smoke_src = smoke or {}
    if isinstance(smoke_src, dict):
        # A lone smoke success is not enough: v198.18 diagnostic-only demos can
        # produce metrics.  Only treat smoke as authoritative when paired with a
        # known generation endpoint and an explicit fulfilled promise marker.
        endpoint = smoke_src.get("api_name") or smoke_src.get("endpoint") or smoke_src.get("primary_api_name")
        if str(smoke_src.get("status") or "").strip().lower() == "success" and endpoint and _truthy(smoke_src.get("promise_fulfilled")):
            return True

    return False

def _canonical_build_status_from_sources(
    *,
    state: dict[str, Any] | None = None,
    launch: dict[str, Any] | None = None,
    summary_file: dict[str, Any] | None = None,
    final_status_reconciliation: dict[str, Any] | None = None,
    eval_record: dict[str, Any] | None = None,
    smoke: dict[str, Any] | None = None,
    gate: dict[str, Any] | None = None,
    live_status: dict[str, Any] | None = None,
    repair: dict[str, Any] | None = None,
    manual_validation: dict[str, Any] | None = None,
    eval_publish_status: dict[str, Any] | None = None,
    inference_contract: dict[str, Any] | None = None,
    demo_quality_contract: dict[str, Any] | None = None,
    events: list[dict[str, Any]] | None = None,
) -> str:
    """Return a canonical build status for explorer/list views.

    Bucket listings are intentionally light-weight and can read slightly stale
    ``state.json`` before the final ``summary.json``/validation artifacts become
    visible. Prefer explicit final summary/eval artifacts over stale partial
    status so Run Explorer and Run Stats match the Active Run snapshot after a
    page load. Intermediate incident events remain phase diagnostics only.
    """
    runtime_success_authoritative = _has_authoritative_runtime_success(
        final_status_reconciliation=final_status_reconciliation,
        summary_file=summary_file,
        state=state,
        live_status=live_status,
        gate=gate,
        smoke=smoke,
    )
    contract_blocks_success = False if runtime_success_authoritative else _contract_blocks_success(summary_file, state, gate, inference_contract, demo_quality_contract)
    # v198.26.1: eval_publish_status is an archive annotation, not a primary
    # run verdict source. It must not make a terminal build appear running/success
    # or override final reconciliation after a simple UI read.
    # v198.26.7: final reconciliation and local eval records are terminal
    # sources of truth for Active Run convergence.  They must win over stale
    # state.json, live_status.json, repair_outcome.json, or intermediate smoke
    # artifacts that can remain from a repaired path.
    authoritative_status = _explicit_effective_status(final_status_reconciliation, eval_record, summary_file)
    if authoritative_status:
        if authoritative_status == "full_inference_success" and contract_blocks_success:
            if _contract_blocks_success(gate, inference_contract, demo_quality_contract):
                return "manual_hardware_required" if "manual" in _source_status_text(gate or {}, inference_contract or {}, demo_quality_contract or {}) else "technical_blocker_boot_only"
        return authoritative_status

    effective_status = _explicit_effective_status(state)
    if effective_status:
        if effective_status == "full_inference_success" and contract_blocks_success:
            if _contract_blocks_success(gate, inference_contract, demo_quality_contract):
                return "manual_hardware_required" if "manual" in _source_status_text(gate or {}, inference_contract or {}, demo_quality_contract or {}) else "technical_blocker_boot_only"
        return effective_status

    event_status = _terminal_status_from_events(events or [], validation=False)
    if event_status:
        return event_status

    summary_status = _classify_build_status_text(_source_status_text(summary_file or {}))
    state_status_for_guard = _classify_build_status_text(_source_status_text(state or {}, repair or {}))
    terminal_text = _source_status_text(gate or {}, smoke or {}, manual_validation or {})
    terminal_status = _classify_build_status_text(terminal_text)

    # Explicit final summary outcomes are authoritative.  v198.8 protects
    # health-passed partials from nested smoke failures during Explorer hydration.
    if summary_status == "full_inference_success":
        if contract_blocks_success:
            return "manual_hardware_required" if "manual" in _source_status_text(gate or {}, inference_contract or {}, demo_quality_contract or {}) else "technical_blocker_boot_only"
        return "full_inference_success"
    if summary_status == "partial_validation":
        return "partial_validation"

    # State-level partial is also protected: nested smoke/gate success or failure
    # can provide detail, but must not silently promote/downgrade the build card.
    if state_status_for_guard == "partial_validation":
        return "partial_validation"

    # Hard terminal evidence from summary/state is stronger than a smoke artifact
    # once no parent summary/state established a protected partial outcome.
    for hard_status in (summary_status, state_status_for_guard):
        if hard_status and hard_status not in {"running", "full_inference_success", "partial_validation"}:
            return hard_status

    # Manual/blocker/failure evidence is stronger than a smoke artifact once the
    # parent summary/state did not already establish a protected partial outcome.
    if terminal_status and terminal_status not in {"running", "full_inference_success", "partial_validation"}:
        return terminal_status

    if terminal_status == "partial_validation":
        return "partial_validation"

    if _has_persisted_generation_success(summary_file, smoke, gate, manual_validation, state):
        if contract_blocks_success:
            return "manual_hardware_required" if "manual" in _source_status_text(gate or {}, inference_contract or {}, demo_quality_contract or {}) else "technical_blocker_boot_only"
        return "full_inference_success"

    if summary_status and summary_status != "running":
        return summary_status

    if terminal_status and terminal_status != "running":
        return terminal_status

    state_status = _classify_build_status_text(_source_status_text(state or {}, repair or {}))
    if state_status:
        return state_status

    summary_status = _classify_build_status_text(_source_status_text(summary_file or {}))
    if summary_status:
        return summary_status

    launch_status = _classify_build_status_text(_source_status_text(launch or {}))
    if launch_status:
        return launch_status

    for src in (state or {}, launch or {}, summary_file or {}):
        for key in ("status", "verdict", "final_status"):
            value = str(src.get(key) or "").lower() if isinstance(src, dict) else ""
            if value:
                return value
    return "unknown"

def _canonical_validation_status_from_sources(
    *,
    state: dict[str, Any] | None = None,
    launch: dict[str, Any] | None = None,
    summary_file: dict[str, Any] | None = None,
    smoke: dict[str, Any] | None = None,
    gate: dict[str, Any] | None = None,
    events: list[dict[str, Any]] | None = None,
) -> str:
    """Return a persisted/canonical validation status.

    Validation runs can have stale launch/summary metadata that still says
    ``running`` after the worker has written a terminal smoke artifact or
    failure event. Terminal evidence must always win over non-terminal launch
    metadata so page reloads, delete modals, and the Run Explorer do not
    resurrect already-finished validations.
    """
    event_status = _terminal_status_from_events(events or [], validation=True)
    if event_status:
        return event_status

    chunks: list[str] = []
    for src in (state or {}, smoke or {}, gate or {}, summary_file or {}):
        for key in ("verdict", "validation_status", "result_status", "status", "gate_status", "smoke_status"):
            value = src.get(key) if isinstance(src, dict) else None
            if value is not None:
                chunks.append(str(value).lower())
        err = src.get("error") if isinstance(src, dict) else None
        if err:
            chunks.append(str(err).lower())
    text = " ".join(chunks)
    if any(x in text for x in ("manual", "requires_action", "hardware_required", "blocked")):
        return "manual_hardware_required"
    if any(x in text for x in ("failed", "failure", "error", "timeout")):
        return "failed"
    if any(x in text for x in ("partial", "health_only", "not_verified", "completed_with_warnings", "warning")):
        return "partial_validation"
    if any(x in text for x in ("full_inference_success", "success", "succeed", "passed", "completed")):
        return "full_inference_success"
    if any(x in text for x in ("stale",)):
        return "stale"
    if any(x in text for x in ("cancelled", "canceled", "stopped")):
        return "stopped"

    for src in (state or {}, launch or {}, summary_file or {}):
        for key in ("status", "validation_status", "result_status", "verdict"):
            value = str(src.get(key) or "").lower() if isinstance(src, dict) else ""
            if value:
                return value
    return "unknown"

def summarize_run_bundle(run_id: str, bundle: dict[str, Any], *, bucket_source: str) -> dict[str, Any]:
    summary_file = bundle.get("summary_file") or {}
    state = bundle.get("state") or {}
    launch = bundle.get("launch") or {}
    gate = bundle.get("inference_gate") or {}
    smoke = bundle.get("generation_smoke") or {}
    hardware = bundle.get("hardware_strategy") or {}
    post_build_validation = bundle.get("post_build_validation_status") or {}
    manual_validation = post_build_validation or bundle.get("manual_validation_status") or {}
    linked_validations = bundle.get("linked_validations") or {}
    linked_rows = linked_validations.get("validations") if isinstance(linked_validations, dict) else []
    linked_rows_list = [row for row in linked_rows if isinstance(row, dict)] if isinstance(linked_rows, list) else []
    linked_terminal_statuses = {
        "success",
        "passed",
        "succeeded",
        "full_inference_success",
        "validated_after_space_test",
        "validated_after_manual_space_test",
        "recovered_by_space_test",
        "recovered_by_manual_validation",
        "manual_validation_passed",
        "manual_validated",
        "partial_validation",
        "manual_hardware_required",
        "generated_needs_manual_hardware",
        "technical_blocker",
        "technical_blocker_boot_only",
        "blocked",
        "completed_with_warnings",
        "failed",
        "failure",
        "error",
        "auth_refresh_required",
        "stale",
        "stopped",
        "cancelled",
        "canceled",
    }
    linked_success_statuses = {
        "success",
        "passed",
        "succeeded",
        "full_inference_success",
        "validated_after_space_test",
        "validated_after_manual_space_test",
        "recovered_by_space_test",
        "recovered_by_manual_validation",
        "manual_validation_passed",
        "manual_validated",
    }
    linked_accounted_rows = [row for row in linked_rows_list if str(row.get("status") or row.get("validation_status") or row.get("result_status") or row.get("effective_status") or "").lower() in linked_terminal_statuses]
    if linked_rows_list:
        linked_successes = [row for row in linked_rows_list if str(row.get("status") or row.get("effective_status") or "").lower() in linked_success_statuses]
        if linked_successes and str(manual_validation.get("status") or "").lower() != "success":
            linked_successes.sort(key=lambda row: str(row.get("validated_at") or row.get("updated_at") or row.get("created_at") or ""), reverse=True)
            manual_validation = linked_successes[0]
    gate_smoke = gate.get("generation_smoke") if isinstance(gate.get("generation_smoke"), dict) else {}
    zero_gpu_recommendation = gate.get("zero_gpu_duration_recommendation") if isinstance(gate, dict) else None
    if isinstance(zero_gpu_recommendation, dict):
        zero_gpu_recommendation = zero_gpu_recommendation.get("recommended_zero_gpu_duration_seconds") or zero_gpu_recommendation.get("recommended_zerogpu_duration_seconds") or zero_gpu_recommendation.get("seconds")
    eval_publish_status = bundle.get("eval_publish_status") or {}
    final_status_reconciliation = bundle.get("final_status_reconciliation") or {}
    eval_record = bundle.get("eval_record") or {}
    target_space = state.get("target_space") or launch.get("target_space") or summary_file.get("target_space") or ""
    kind = state.get("kind") or launch.get("kind") or summary_file.get("kind") or "unknown"
    is_validation = "validate" in str(kind).lower() or str(run_id).startswith("validate-")
    if is_validation:
        status = _canonical_validation_status_from_sources(
            state=state,
            launch=launch,
            summary_file=summary_file,
            smoke=smoke,
            gate=gate,
            events=bundle.get("events") or [],
        )
    else:
        # Legacy test guard: old code used `status = event_status or state.get("status")`;
        # build runs now use canonical source reconciliation instead.
        status = _canonical_build_status_from_sources(
            state=state,
            launch=launch,
            summary_file=summary_file,
            final_status_reconciliation=final_status_reconciliation,
            eval_record=eval_record,
            smoke=smoke or bundle.get("test_result") or {},
            gate=gate,
            live_status=bundle.get("live_status") or {},
            repair=bundle.get("repair_outcome") or {},
            manual_validation=manual_validation,
            eval_publish_status=bundle.get("eval_publish_status") or {},
            inference_contract=bundle.get("inference_contract") or {},
            demo_quality_contract=bundle.get("demo_quality_contract") or {},
            events=bundle.get("events") or [],
        )
    terminal = status in {"failed", "full_inference_success", "partial_validation", "manual_hardware_required", "stale", "stopped"}
    effective_run_status = compute_effective_run_status({**bundle, "summary": {**summary_file, "status": status}}, build_status=status)
    return {
        "run_id": run_id,
        "kind": kind,
        "status": status,
        "validation_status": status if is_validation else "",
        "result_status": status if is_validation else "",
        "terminal": bool(is_validation and terminal),
        "can_resume": not bool(is_validation and terminal),
        "model_id": state.get("model_id") or launch.get("model_id") or summary_file.get("model_id") or state.get("model") or "",
        "pi_model": state.get("pi_model") or launch.get("pi_model") or summary_file.get("pi_model") or (bundle.get("pi_model_resolution") or {}).get("requested_model") or "",
        "target_space": target_space,
        "target_space_url": state.get("target_space_url") or launch.get("target_space_url") or summary_file.get("target_space_url") or (f"https://huggingface.co/spaces/{target_space}" if target_space else ""),
        "job_id": state.get("job_id") or launch.get("job_id") or summary_file.get("job_id") or "",
        "job_url": _job_url_from_sources(bucket_source=bucket_source, summary_file=summary_file, launch=launch, state=state),
        "created_at": state.get("created_at") or launch.get("created_at") or summary_file.get("created_at") or "",
        "updated_at": state.get("updated_at") or state.get("created_at") or launch.get("updated_at") or launch.get("created_at") or summary_file.get("updated_at") or summary_file.get("created_at") or "",
        "selected_hardware": hardware.get("selected_hardware") or state.get("selected_hardware") or launch.get("preferred_space_hardware") or summary_file.get("selected_hardware") or state.get("hardware") or "",
        "manual_hardware_required": bool(gate.get("manual_hardware_required") or hardware.get("manual_action_required") or launch.get("manual_hardware_required")),
        "health_passed": bool((gate.get("implementation_signals") or {}).get("health_passed") or smoke.get("health_passed") or gate_smoke.get("health_passed")),
        "smoke_test_passed": bool(smoke.get("ok") or smoke.get("status") == "success" or gate_smoke.get("status") == "success"),
        "latency_seconds": manual_validation.get("latency_seconds") or manual_validation.get("observed_latency_seconds") or smoke.get("latency_seconds") or smoke.get("observed_latency_seconds") or gate_smoke.get("latency_seconds") or gate_smoke.get("observed_latency_seconds"),
        "observed_latency_seconds": manual_validation.get("observed_latency_seconds") or manual_validation.get("latency_seconds") or smoke.get("observed_latency_seconds") or smoke.get("latency_seconds") or gate_smoke.get("observed_latency_seconds") or gate_smoke.get("latency_seconds"),
        "recommended_zero_gpu_duration_seconds": manual_validation.get("recommended_zero_gpu_duration_seconds") or manual_validation.get("recommended_zerogpu_duration_seconds") or smoke.get("recommended_zero_gpu_duration_seconds") or smoke.get("recommended_zerogpu_duration_seconds") or gate_smoke.get("recommended_zero_gpu_duration_seconds") or gate_smoke.get("recommended_zerogpu_duration_seconds") or zero_gpu_recommendation,
        "recommended_zerogpu_duration_seconds": manual_validation.get("recommended_zerogpu_duration_seconds") or manual_validation.get("recommended_zero_gpu_duration_seconds") or smoke.get("recommended_zerogpu_duration_seconds") or smoke.get("recommended_zero_gpu_duration_seconds") or gate_smoke.get("recommended_zerogpu_duration_seconds") or gate_smoke.get("recommended_zero_gpu_duration_seconds") or zero_gpu_recommendation,
        "recommendation_source": manual_validation.get("recommendation_source") or smoke.get("recommendation_source") or gate_smoke.get("recommendation_source") or ("linked_space_test" if str(manual_validation.get("status") or "").lower() == "success" else ""),
        "recommendation_hardware": manual_validation.get("recommendation_hardware") or manual_validation.get("hardware_used_for_validation") or smoke.get("recommendation_hardware") or gate_smoke.get("recommendation_hardware") or state.get("selected_hardware") or "",
        "hardware_used_for_validation": manual_validation.get("hardware_used_for_validation") or manual_validation.get("recommendation_hardware") or state.get("selected_hardware") or "",
        "manual_validation_status": manual_validation,
        "manual_validation_passed": str(manual_validation.get("status") or manual_validation.get("effective_status") or "").lower() in linked_success_statuses,
        "linked_validations": linked_validations if isinstance(linked_validations, dict) else {},
        "linked_validation_count": len(linked_accounted_rows),
        "linked_validations_count": len(linked_accounted_rows),
        "post_build_status": effective_run_status.get("post_build_status") or "none",
        "post_build_validation": effective_run_status.get("post_build_validation") or {},
        "effective_run_status": effective_run_status,
        "effective_status": effective_run_status.get("effective_status") or status,
        "effective_verdict": effective_run_status.get("effective_verdict") or status,
        "legacy_effective_status": effective_run_status.get("legacy_effective_status") or manual_validation.get("effective_status") or "",
        "display_status": effective_run_status.get("display_status") or status,
        "ui_status": effective_run_status.get("display_status") or state.get("ui_status") or gate.get("ui_status") or status,
        "app_boot_validation_status": state.get("app_boot_validation_status") or gate.get("app_boot_validation_status") or summary_file.get("app_boot_validation_status") or "",
        "promise_validation_status": state.get("promise_validation_status") or gate.get("promise_validation_status") or summary_file.get("promise_validation_status") or "",
        "promise_fulfilled": bool(state.get("promise_fulfilled") or gate.get("promise_fulfilled") or summary_file.get("promise_fulfilled")),
        "ui_badge": state.get("ui_badge") or gate.get("ui_badge") or summary_file.get("ui_badge") or "",
        "expected_output_type": smoke.get("expected_output_type") or state.get("expected_output_type") or launch.get("expected_output_type") or "",
        "pi_model_resolution": bundle.get("pi_model_resolution") or state.get("pi_model_resolution") or {},
        "artifacts_url": f"https://huggingface.co/buckets/{bucket_source}/tree/{(settings.bucket_runs_prefix.strip().strip('/') or 'runs')}/{run_id}",
    }


def read_run_bundle(run_id: str, *, bucket_source: str, token: str | None = None, include_heavy: bool = True) -> dict[str, Any]:
    paths = RunPaths(run_id, bucket_source=bucket_source)
    bundle = {
        "paths": {
            "root": paths.root,
            "state": paths.state,
            "events": paths.events,
            "report": paths.report,
        },
        "summary_file": _safe_read_json(f"{paths.root}/summary.json", token=token),
        "final_status_reconciliation": _safe_read_json(f"{paths.root}/final_status_reconciliation.json", token=token),
        "eval_record": _safe_read_json(f"{paths.root}/eval_record.json", token=token),
        "launch": _safe_read_json(f"{paths.root}/launch.json", token=token),
        "state": read_json(paths.state, token=token) or {},
        "live_status": _safe_read_json(f"{paths.root}/live_status.json", token=token),
        "events": read_events(run_id, bucket_source=bucket_source, token=token),
        "report": _safe_read_text(paths.report, token=token) if include_heavy else "",
        "inference_gate": _safe_read_json(f"{paths.root}/inference_gate.json", token=token),
        "generation_smoke": _safe_read_json(f"{paths.root}/tests/generation_smoke.json", token=token) or _safe_read_json(f"{paths.root}/generation_smoke.json", token=token),
        "test_result": _safe_read_json(f"{paths.root}/tests/test_result.json", token=token) or _safe_read_json(f"{paths.root}/test_result.json", token=token),
        "generation_smoke_payload": _safe_read_json(f"{paths.root}/tests/generation_smoke_payload.json", token=token),
        "generation_smoke_payload_retry": _safe_read_json(f"{paths.root}/tests/generation_smoke_payload_retry.json", token=token),
        "resolved_validation_request": _safe_read_json(f"{paths.root}/tests/resolved_validation_request.json", token=token),
        "payload_source": _safe_read_json(f"{paths.root}/tests/payload_source.json", token=token),
        "validation_engine": _safe_read_json(f"{paths.root}/tests/validation_engine.json", token=token),
        "validation_failure_diagnosis": _safe_read_json(f"{paths.root}/tests/validation_failure_diagnosis.json", token=token),
        "hardware_strategy": _safe_read_json(f"{paths.root}/hardware_strategy.json", token=token),
        "hardware_attempts": _safe_read_json(f"{paths.root}/hardware_attempts.json", token=token),
        "technical_blockers": _safe_read_json(f"{paths.root}/generated/TECHNICAL_BLOCKERS.json", token=token) or _safe_read_json(f"{paths.root}/TECHNICAL_BLOCKERS.json", token=token),
        "inference_contract": _safe_read_json(f"{paths.root}/generated/INFERENCE_CONTRACT.json", token=token) or _safe_read_json(f"{paths.root}/INFERENCE_CONTRACT.json", token=token),
        "demo_quality_contract": _safe_read_json(f"{paths.root}/generated/DEMO_QUALITY_CONTRACT.json", token=token) or _safe_read_json(f"{paths.root}/DEMO_QUALITY_CONTRACT.json", token=token),
        "repair_outcome": _safe_read_json(f"{paths.root}/repair_outcome.json", token=token) or _safe_read_json(f"{paths.root}/repair/REPAIR_OUTCOME.json", token=token),
        "model_analysis": _safe_read_json(f"{paths.root}/model_analysis.json", token=token),
        "pi_model_resolution": _safe_read_json(f"{paths.root}/pi_model_resolution.json", token=token) or (read_json(paths.state, token=token) or {}).get("pi_model_resolution") or {},
        "space_runtime": _safe_read_json(f"{paths.root}/space_runtime.json", token=token),
        "runtime_upload_epoch": _safe_read_json(f"{paths.root}/runtime_upload_epoch.json", token=token),
        "placeholder_scaffold_detection": _safe_read_json(f"{paths.root}/placeholder_scaffold_detection.json", token=token),
        "api_schema": _safe_read_json(f"{paths.root}/tests/api_schema.json", token=token) or _safe_read_json(f"{paths.root}/tests/generation_api_schema.json", token=token),
        "validation_payload": _safe_read_json(f"{paths.root}/tests/validation_payload.json", token=token),
        "post_build_validation_status": _safe_read_json(f"{paths.root}/post_build_validation_status.json", token=token),
        "manual_validation_status": _safe_read_json(f"{paths.root}/manual_validation_status.json", token=token),
        "linked_validations": _safe_read_json(f"{paths.root}/linked_validations.json", token=token),
        "eval_publish_status": _safe_read_json(f"{paths.root}/eval_publish_status.json", token=token),
        "build_error_observation": _safe_read_json(f"{paths.root}/build_error_observation.json", token=token),
        "repair_decision": _safe_read_json(f"{paths.root}/repair/REPAIR_DECISION.json", token=token),
        "blockage": _safe_read_json(f"{paths.root}/repair/BLOCKAGE.json", token=token),
        "artifact_manifest": _safe_read_json(f"{paths.root}/artifact_manifest.json", token=token),
        "files": _list_run_files(run_id, bucket_source=bucket_source, token=token) if include_heavy else [],
    }
    if not bundle.get("generation_smoke") and isinstance((bundle.get("inference_gate") or {}).get("generation_smoke"), dict):
        bundle["generation_smoke"] = (bundle.get("inference_gate") or {}).get("generation_smoke") or {}
    bundle["run_documents"] = _run_document_links(run_id, bucket_source=bucket_source, bundle=bundle, token=token)
    if include_heavy and not bundle.get("pi_model_resolution"):
        try:
            bundle["pi_model_resolution"] = _trace_based_pi_model_resolution(run_id, bundle, bucket_source=bucket_source, token=token)
        except Exception:
            bundle["pi_model_resolution"] = {}
    bundle["summary"] = summarize_run_bundle(run_id, bundle, bucket_source=bucket_source)
    return bundle


def _run_id_from_path(path: str) -> str:
    return path.rstrip('/').split('/')[-1]


def _discover_run_ids(root: str, *, token: str | None = None, limit: int = 300) -> list[str]:
    """Discover run ids in a bucket even when the run folder is partial.

    HfFileSystem may expose object-store prefixes slightly differently between
    Buckets/runtime versions. Use ls plus a few targeted globs so running Jobs
    with only launch.json/summary.json are still shown in the Run Explorer.
    """
    fs = _fs(token)
    run_ids: set[str] = set()

    def add_from_path(path: str) -> None:
        marker = f"/{_runs_prefix()}/"
        if marker in path:
            tail = path.split(marker, 1)[1]
        else:
            tail = path.replace(root.rstrip("/") + "/", "", 1)
        run_id = tail.strip("/").split("/", 1)[0]
        if run_id and run_id != "runs":
            run_ids.add(run_id)

    try:
        for entry in fs.ls(root, detail=True):
            name = entry.get("name") if isinstance(entry, dict) else str(entry)
            if name:
                add_from_path(name)
    except Exception:
        pass

    for pattern in (f"{root}/*", f"{root}/*/summary.json", f"{root}/*/launch.json", f"{root}/*/state.json"):
        try:
            for path in fs.glob(pattern):
                add_from_path(str(path))
                if len(run_ids) >= limit:
                    break
        except Exception:
            continue

    return sorted(run_ids, reverse=True)[:limit]


def list_recent_runs(
    *,
    bucket_source: str,
    token: str | None = None,
    limit: int = 50,
    query: str | None = None,
    status: str | None = None,
) -> list[dict[str, Any]]:
    """List recent run summaries from a user's bucket.

    Best effort: a run can be visible as soon as launch.json/summary.json exists,
    before the worker writes full state/events. This keeps running Jobs visible
    and makes Job links available immediately.
    """
    root = f"{bucket_uri_from_source(bucket_source)}/{_runs_prefix()}"
    run_ids = _discover_run_ids(root, token=token)
    if not run_ids:
        return []

    runs: list[dict[str, Any]] = []
    for run_id in run_ids:
        summary_file = read_json(f"{root}/{run_id}/summary.json", token=token) or {}
        launch = read_json(f"{root}/{run_id}/launch.json", token=token) or {}
        state = read_json(f"{root}/{run_id}/state.json", token=token) or {}
        gate = read_json(f"{root}/{run_id}/inference_gate.json", token=token) or {}
        smoke = read_json(f"{root}/{run_id}/tests/generation_smoke.json", token=token) or read_json(f"{root}/{run_id}/generation_smoke.json", token=token) or {}
        test_result = read_json(f"{root}/{run_id}/tests/test_result.json", token=token) or read_json(f"{root}/{run_id}/test_result.json", token=token) or {}
        if not smoke and isinstance(gate.get("generation_smoke"), dict):
            smoke = gate.get("generation_smoke") or {}
        hardware = read_json(f"{root}/{run_id}/hardware_strategy.json", token=token) or {}
        pi_model_resolution = read_json(f"{root}/{run_id}/pi_model_resolution.json", token=token) or state.get("pi_model_resolution") or {}
        post_build_validation_status = read_json(f"{root}/{run_id}/post_build_validation_status.json", token=token) or {}
        manual_validation_status = read_json(f"{root}/{run_id}/manual_validation_status.json", token=token) or {}
        linked_validations = read_json(f"{root}/{run_id}/linked_validations.json", token=token) or {}
        eval_publish_status = read_json(f"{root}/{run_id}/eval_publish_status.json", token=token) or {}
        kind_probe = state.get("kind") or launch.get("kind") or summary_file.get("kind") or ""
        events = read_events(run_id, bucket_source=bucket_source, token=token) if ("validate" in str(kind_probe).lower() or str(run_id).startswith("validate-")) else []
        partial_bundle = {
            "summary_file": summary_file,
            "launch": launch,
            "state": state,
            "inference_gate": gate,
            "generation_smoke": smoke,
            "test_result": test_result,
            "hardware_strategy": hardware,
            "pi_model_resolution": pi_model_resolution,
            "post_build_validation_status": post_build_validation_status,
            "manual_validation_status": manual_validation_status,
            "linked_validations": linked_validations,
            "eval_publish_status": eval_publish_status,
            "events": events,
        }
        item = summarize_run_bundle(run_id, partial_bundle, bucket_source=bucket_source)
        if post_build_validation_status:
            item["post_build_validation_status"] = post_build_validation_status
            item["effective_status"] = post_build_validation_status.get("effective_status") or item.get("effective_status")
        if linked_validations:
            rows = linked_validations.get("validations") if isinstance(linked_validations, dict) else []
            terminal_tokens = {"success", "passed", "succeeded", "full_inference_success", "validated_after_space_test", "validated_after_manual_space_test", "recovered_by_space_test", "recovered_by_manual_validation", "manual_validation_passed", "manual_validated", "partial_validation", "manual_hardware_required", "generated_needs_manual_hardware", "technical_blocker", "technical_blocker_boot_only", "blocked", "completed_with_warnings", "failed", "failure", "error", "auth_refresh_required", "stale", "stopped", "cancelled", "canceled"}
            accounted_rows = [row for row in rows if isinstance(row, dict) and str(row.get("status") or row.get("validation_status") or row.get("result_status") or row.get("effective_status") or "").lower() in terminal_tokens] if isinstance(rows, list) else []
            item["linked_validations"] = linked_validations
            item["linked_validation_count"] = len(accounted_rows)
            item["linked_validations_count"] = len(accounted_rows)
        if manual_validation_status:
            item["manual_validation_status"] = manual_validation_status
            manual_status_token = str(manual_validation_status.get("status") or manual_validation_status.get("effective_status") or "").lower()
            if manual_status_token not in {"", "none", "unknown", "unchanged"}:
                item["effective_status"] = manual_validation_status.get("effective_status") or item.get("effective_status")
            if manual_validation_status.get("status") == "success":
                item["manual_validation_passed"] = True
                item["latency_seconds"] = item.get("latency_seconds") or manual_validation_status.get("latency_seconds")
                item["observed_latency_seconds"] = item.get("observed_latency_seconds") or manual_validation_status.get("observed_latency_seconds")
                item["recommended_zero_gpu_duration_seconds"] = item.get("recommended_zero_gpu_duration_seconds") or manual_validation_status.get("recommended_zero_gpu_duration_seconds")
                item["recommendation_source"] = item.get("recommendation_source") or manual_validation_status.get("recommendation_source")
                item["recommendation_hardware"] = item.get("recommendation_hardware") or manual_validation_status.get("recommendation_hardware") or manual_validation_status.get("hardware_used_for_validation")
                item["hardware_used_for_validation"] = item.get("hardware_used_for_validation") or manual_validation_status.get("hardware_used_for_validation")
                item["recommendation_source"] = item.get("recommendation_source") or "linked_space_test"
        item["bucket_source"] = bucket_source
        haystack = " ".join(str(item.get(k, "")) for k in ["run_id", "model_id", "target_space", "status", "kind", "job_id"]).lower()
        if query and query.lower() not in haystack:
            continue
        if status and status not in {"all", ""} and item["status"] != status:
            continue
        runs.append(item)

    runs.sort(key=lambda r: str(r.get("updated_at") or r.get("created_at") or r.get("run_id")), reverse=True)
    return runs[: max(1, min(int(limit or 50), 200))]