src/view_models.py

from __future__ import annotations

from datetime import datetime, timedelta, timezone
import re
from typing import Any

from .progress import STEP_ALIASES, STEP_LABELS, STEP_ORDER
from .timeline_model import build_run_timeline_model
from .config import settings
from .effective_status import compute_effective_run_status

PRODUCT_STEPS = [{"id": step, "label": STEP_LABELS[step]} for step in STEP_ORDER]


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

TERMINAL_GLOBAL_STATUSES = {"succeeded", "partial", "failed", "cancelled", "blocked", "waiting_manual_action", "auth_refresh_required"}
SUCCESS_RAW_STATUSES = {
    "success",
    "done",
    "completed",
    "passed",
    "full_inference_success",
    "repair_success",
}
PARTIAL_RAW_STATUSES = {
    "full_inference_candidate_health_passed",
    "health_only",
    "partial",
    "partial_validation",
    "completed_with_warnings",
    "demo_usable_full_promise_not_verified",
    "interactive_app_available_smoke_failed",
    "manual_test_required_smoke_failed",
}
MANUAL_RAW_STATUSES = {
    "manual_hardware_required",
    "generated_needs_manual_hardware",
    "waiting_manual_hardware",
    "manual_action_required",
}
FAILED_RAW_STATUSES = {"failed", "failure", "error", "repair_failed"}
CANCELLED_RAW_STATUSES = {"cancelled", "canceled"}
BLOCKED_RAW_STATUSES = {"technical_blocker", "technical_blocker_boot_only", "blocked"}
RUNNING_RAW_STATUSES = {"started", "running", "waiting", "queued", "pending", "scheduled"}
AUTH_RAW_STATUSES = {"auth_refresh_required", "oauth_expired", "auth_expired", "repair_validation_inconclusive_auth", "inconclusive_auth_expired"}

AUTHORITATIVE_TERMINAL_SOURCE_KEYS = ("status", "final_status", "effective_status", "effective_verdict", "display_status", "verdict", "result_status", "validation_status")
TERMINAL_DISPLAY_STATUSES = SUCCESS_RAW_STATUSES | PARTIAL_RAW_STATUSES | MANUAL_RAW_STATUSES | FAILED_RAW_STATUSES | CANCELLED_RAW_STATUSES | BLOCKED_RAW_STATUSES | AUTH_RAW_STATUSES | {"stale", "stopped", "succeeded", "validated_after_space_test", "validated_after_manual_space_test", "recovered_by_space_test", "recovered_by_manual_validation"}

STEP_TO_PHASE = {step: step for step in STEP_ORDER} | {alias: canonical for alias, canonical in STEP_ALIASES.items()}


def parse_ts(value: Any) -> datetime | None:
    if not value:
        return None
    if isinstance(value, datetime):
        return value if value.tzinfo else value.replace(tzinfo=timezone.utc)
    try:
        parsed = datetime.fromisoformat(str(value).replace("Z", "+00:00"))
        return parsed if parsed.tzinfo else parsed.replace(tzinfo=timezone.utc)
    except Exception:
        return None


def _lower(value: Any) -> str:
    return str(value or "").strip().lower()


def _first_nonempty(*values: Any) -> str:
    for value in values:
        if value is not None and str(value).strip():
            return str(value)
    return ""

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_view_sources(*, bucket_source: str, summary: dict[str, Any], state: dict[str, Any], launch: dict[str, Any]) -> str:
    explicit = summary.get("job_url") or state.get("job_url") or launch.get("job_url")
    if explicit:
        return str(explicit)
    job_id = summary.get("job_id") or state.get("job_id") or launch.get("job_id")
    owner = (
        summary.get("created_by")
        or summary.get("username")
        or summary.get("owner")
        or state.get("created_by")
        or state.get("username")
        or state.get("owner")
        or launch.get("created_by")
        or launch.get("username")
        or launch.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 _latest_activity_at(bundle: dict[str, Any], summary: dict[str, Any], state: dict[str, Any], launch: dict[str, Any]) -> datetime | None:
    candidates: list[datetime] = []
    for event in bundle.get("events") or []:
        ts = parse_ts(event.get("ts") if isinstance(event, dict) else None)
        if ts:
            candidates.append(ts)
    for source in (summary, state, launch, bundle.get("summary_file") or {}):
        for key in ("updated_at", "finished_at", "created_at", "started_at"):
            ts = parse_ts(source.get(key) if isinstance(source, dict) else None)
            if ts:
                candidates.append(ts)
    return max(candidates) if candidates else None


def _created_at(summary: dict[str, Any], state: dict[str, Any], launch: dict[str, Any]) -> datetime | None:
    for source in (summary, state, launch):
        ts = parse_ts(source.get("created_at") if isinstance(source, dict) else None)
        if ts:
            return ts
    return None


def _raw_statuses(bundle: dict[str, Any]) -> set[str]:
    summary = bundle.get("summary") 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 {}
    blockers = bundle.get("technical_blockers") or {}

    values = {
        _lower(summary.get("status")),
        _lower(state.get("status")),
        _lower(state.get("gate_status")),
        _lower(launch.get("status")),
        _lower(gate.get("status")),
        _lower(smoke.get("status")),
        _lower(hardware.get("status")),
        _lower(blockers.get("status")),
        _lower((bundle.get("repair_outcome") or {}).get("post_repair_validation")),
        _lower((bundle.get("repair_outcome") or {}).get("failure_type")),
    }
    # Event-level statuses describe individual steps and must not by themselves
    # turn a whole run into success/failure. The product phase is derived from
    # events separately in derive_product_phase().
    return {v for v in values if v}


def _normalize_terminal_status(value: Any) -> str:
    status = _lower(value)
    if not status:
        return ""
    if status in {"succeeded", "success", "done", "completed", "passed", "repair_success"}:
        return "full_inference_success"
    if status in {"cancelled", "canceled"}:
        return "stopped"
    if status in {"partial", "health_only", "full_inference_candidate_health_passed", "completed_with_warnings"}:
        return "partial_validation"
    if status in TERMINAL_DISPLAY_STATUSES:
        return status
    return ""


def _first_authoritative_terminal_status(bundle: dict[str, Any]) -> tuple[str, str]:
    """Return strongest terminal display status and source for Active Run.

    v198.26.7 invariant: final reconciliation and summary are terminal sources
    of truth.  They must not be downgraded by stale state/live/progress artifacts
    left behind by a repaired run.
    """
    ordered_sources: list[tuple[str, dict[str, Any]]] = [
        ("final_status_reconciliation", bundle.get("final_status_reconciliation") or {}),
        ("summary", bundle.get("summary") or {}),
        ("summary_file", bundle.get("summary_file") or {}),
        ("effective_run_status", bundle.get("effective_run_status") or {}),
    ]
    summary = bundle.get("summary") or {}
    if isinstance(summary.get("effective_run_status"), dict):
        ordered_sources.append(("summary.effective_run_status", summary.get("effective_run_status") or {}))
    # State/live_status are runtime artifacts written after the worker has seen
    # the generated Space.  They are stronger than stale Pi/contract/gate
    # diagnostic blockers, but still lower priority than final reconciliation and
    # summary.
    ordered_sources.extend([
        ("state", bundle.get("state") or {}),
        ("live_status", bundle.get("live_status") or {}),
        ("inference_gate", bundle.get("inference_gate") or {}),
        ("generation_smoke", bundle.get("generation_smoke") or {}),
    ])
    blocking_seen = False
    for source_name, source in ordered_sources:
        if not isinstance(source, dict):
            continue
        for key in AUTHORITATIVE_TERMINAL_SOURCE_KEYS:
            normalized = _normalize_terminal_status(source.get(key))
            if normalized:
                if normalized in BLOCKED_RAW_STATUSES | FAILED_RAW_STATUSES | MANUAL_RAW_STATUSES | AUTH_RAW_STATUSES | CANCELLED_RAW_STATUSES | {"stopped", "stale"}:
                    return normalized, f"{source_name}.{key}"
                if normalized == "full_inference_success" and blocking_seen:
                    continue
                return normalized, f"{source_name}.{key}"
        raw_text = " ".join(_lower(source.get(k)) for k in AUTHORITATIVE_TERMINAL_SOURCE_KEYS if isinstance(source.get(k), (str, int, float, bool)))
        if any(token in raw_text for token in ("technical_blocker", "technical blocker", "blocked", "failed", "manual_hardware_required", "auth_refresh_required", "stopped")):
            blocking_seen = True
        # Metrics/smoke are evidence, not a verdict. They may confirm success
        # only when no authoritative blocking source has appeared earlier.
        if not blocking_seen and (source.get("ok") is True or source.get("generation_smoke_passed") is True):
            return "full_inference_success", f"{source_name}.generation_smoke_passed"
        signals = source.get("implementation_signals") if isinstance(source.get("implementation_signals"), dict) else {}
        if not blocking_seen and signals.get("generation_smoke_passed") is True:
            return "full_inference_success", f"{source_name}.implementation_signals.generation_smoke_passed"
    return "", ""


def _has_authoritative_runtime_success(bundle: dict[str, Any]) -> bool:
    status, _source = _first_authoritative_terminal_status(bundle)
    if status == "full_inference_success":
        return True
    live = bundle.get("live_status") or {}
    if isinstance(live, dict):
        if _lower(live.get("stage")) == "done" and _normalize_terminal_status(live.get("status")) == "full_inference_success":
            return True
        if _lower(live.get("stage")) == "done" and live.get("generation_smoke_passed") is True:
            return True
    gate = bundle.get("inference_gate") or {}
    if isinstance(gate, dict) and gate.get("strong_full_inference_success") is True:
        return True
    return False


def _status_model_from_terminal_status(status: str, *, source: str = "") -> dict[str, str | bool]:
    normalized = _normalize_terminal_status(status) or "unknown"
    if normalized == "full_inference_success":
        global_status, verdict = "succeeded", "full_inference_success"
    elif normalized in PARTIAL_RAW_STATUSES or normalized == "partial_validation":
        global_status, verdict = "partial", normalized if normalized not in {"partial", "health_only", "full_inference_candidate_health_passed", "completed_with_warnings"} else "partial_validation"
    elif normalized in MANUAL_RAW_STATUSES:
        global_status, verdict = "waiting_manual_action", "manual_action_required"
    elif normalized in BLOCKED_RAW_STATUSES:
        global_status, verdict = "blocked", "technical_blocker_boot_only" if normalized == "technical_blocker_boot_only" else "technical_blocker"
    elif normalized in FAILED_RAW_STATUSES:
        global_status, verdict = "failed", "failed"
    elif normalized in AUTH_RAW_STATUSES:
        global_status, verdict = "auth_refresh_required", "auth_refresh_required"
    elif normalized in CANCELLED_RAW_STATUSES or normalized == "stopped":
        global_status, verdict = "cancelled", "cancelled"
    elif normalized == "stale":
        global_status, verdict = "stale", "stale"
    else:
        global_status, verdict = normalized, normalized
    return {"global_status": global_status, "verdict": verdict, "terminal_source": source}


def requires_manual_action(bundle: dict[str, Any]) -> bool:
    summary = bundle.get("summary") or {}
    gate = bundle.get("inference_gate") or {}
    hardware = bundle.get("hardware_strategy") or {}
    statuses = _raw_statuses(bundle)
    return bool(
        summary.get("manual_hardware_required")
        or gate.get("manual_hardware_required")
        or hardware.get("manual_action_required")
        or statuses.intersection(MANUAL_RAW_STATUSES)
    )


def has_technical_blocker(bundle: dict[str, Any]) -> bool:
    if _has_authoritative_runtime_success(bundle):
        return False
    blockers = bundle.get("technical_blockers") or {}
    blocker_items = blockers.get("blockers") if isinstance(blockers, dict) else None
    return bool(blocker_items or _raw_statuses(bundle).intersection(BLOCKED_RAW_STATUSES))


def normalize_run_status(
    bundle: dict[str, Any],
    *,
    now: datetime | None = None,
    stale_after: timedelta = timedelta(hours=6),
) -> dict[str, Any]:
    """Return canonical status flags for the product UI."""
    now = now or datetime.now(timezone.utc)
    summary = bundle.get("summary") or {}
    state = bundle.get("state") or {}
    launch = bundle.get("launch") or {}
    statuses = _raw_statuses(bundle)
    manual = requires_manual_action(bundle)
    blocker = has_technical_blocker(bundle)
    authoritative_status, terminal_source = _first_authoritative_terminal_status(bundle)

    if authoritative_status:
        terminal_model = _status_model_from_terminal_status(authoritative_status, source=terminal_source)
        global_status = str(terminal_model["global_status"])
        verdict = str(terminal_model["verdict"])
    elif statuses.intersection(AUTH_RAW_STATUSES):
        global_status = "auth_refresh_required"
        verdict = "auth_refresh_required"
    elif statuses.intersection(CANCELLED_RAW_STATUSES):
        global_status = "cancelled"
        verdict = "cancelled"
    elif manual:
        global_status = "waiting_manual_action"
        verdict = "manual_action_required"
    elif blocker:
        global_status = "blocked"
        verdict = "technical_blocker_boot_only" if "technical_blocker_boot_only" in statuses else "technical_blocker"
    elif statuses.intersection(FAILED_RAW_STATUSES):
        global_status = "failed"
        verdict = "failed"
    elif statuses.intersection(PARTIAL_RAW_STATUSES):
        global_status = "partial"
        verdict = "partial_validation"
    elif statuses.intersection(SUCCESS_RAW_STATUSES):
        global_status = "succeeded"
        verdict = "passed"
    elif statuses.intersection(RUNNING_RAW_STATUSES) or bundle.get("events") or launch:
        global_status = "running"
        verdict = "pending"
    else:
        global_status = "unknown"
        verdict = "unknown"

    latest = _latest_activity_at(bundle, summary, state, launch)
    created = _created_at(summary, state, launch)
    reference = latest or created
    is_terminal = global_status in TERMINAL_GLOBAL_STATUSES or global_status in {"succeeded"}
    is_stale = False
    if reference and not is_terminal and now - reference > stale_after:
        global_status = "stale"
        verdict = "stale"
        is_stale = True

    return {
        "global_status": global_status,
        "raw_status": _first_nonempty(summary.get("status"), state.get("status"), launch.get("status"), "unknown"),
        "verdict": verdict,
        "requires_manual_action": manual,
        "manual_action_type": "hardware" if manual else "",
        "has_technical_blocker": blocker,
        "has_target_space": bool(summary.get("target_space") or state.get("target_space") or launch.get("target_space")),
        "has_job_url": bool(summary.get("job_url") or state.get("job_url") or launch.get("job_url") or summary.get("job_id") or state.get("job_id") or launch.get("job_id")),
        "has_live_api_result": bool((bundle.get("generation_smoke") or {}).get("ok") or (bundle.get("generation_smoke") or {}).get("status") == "success"),
        "is_terminal": global_status in {"succeeded", "partial", "failed", "blocked", "waiting_manual_action", "cancelled", "auth_refresh_required"},
        "is_pollable": global_status in {"queued", "running", "validating", "unknown"},
        "is_stale": is_stale,
        "latest_activity_at": latest.isoformat() if latest else "",
        "terminal_source": terminal_source if 'terminal_source' in locals() else "",
    }


def derive_product_phase(bundle: dict[str, Any], status_model: dict[str, Any]) -> str:
    if status_model["global_status"] in {"succeeded", "partial"}:
        return "done"
    if status_model["global_status"] in {"failed", "blocked", "stale", "cancelled"}:
        return "failure"
    if status_model["requires_manual_action"]:
        return "inference_gate"

    events = [e for e in (bundle.get("events") or []) if isinstance(e, dict)]
    for event in reversed(events):
        step = _lower(event.get("step"))
        message = _lower(event.get("message"))
        status = _lower(event.get("status"))
        if "patch" in step or "repair" in step or "patched" in message or "restart" in message:
            return "hardware_strategy"
        if "missing" in message or "error" in message or status in FAILED_RAW_STATUSES:
            return "hardware_strategy"
        phase = STEP_TO_PHASE.get(step)
        if phase:
            return phase

    summary = bundle.get("summary") or {}
    if summary.get("target_space"):
        return "hardware_strategy"
    if bundle.get("launch"):
        return "bootstrap"
    return "bootstrap"


def build_pipeline(phase: str, status_model: dict[str, Any]) -> list[dict[str, Any]]:
    phase_ids = [s["id"] for s in PRODUCT_STEPS]
    current_index = phase_ids.index(phase) if phase in phase_ids else 0
    global_status = status_model["global_status"]

    pipeline: list[dict[str, Any]] = []
    for index, step in enumerate(PRODUCT_STEPS):
        if global_status == "succeeded":
            step_status = "completed"
        elif index < current_index:
            step_status = "completed"
        elif index == current_index:
            if global_status == "failed":
                step_status = "failed"
            elif global_status in {"blocked", "waiting_manual_action", "stale"}:
                step_status = "blocked"
            else:
                step_status = "running"
        else:
            step_status = "pending"
        pipeline.append({**step, "status": step_status})
    return pipeline


def _agent_for_event(event: dict[str, Any]) -> str:
    step = _lower(event.get("step"))
    if step in {"model_analysis", "bucket_ready", "job_launched", "bootstrap"}:
        return "Planner Agent"
    if step in {"workspace", "node", "pi_install", "pi_config", "pi_run", "repair", "patch"}:
        return "Coder Agent"
    if step in {"create_space", "upload_files", "hardware", "hardware_preferred", "hardware_fallback"}:
        return "Hub Agent"
    if step in {"api_validation", "generation_smoke", "inference_gate"}:
        return "Tester Agent"
    if _lower(event.get("status")) in FAILED_RAW_STATUSES:
        return "Diagnostician Agent"
    return "Factory Agent"


def _severity_for_event(event: dict[str, Any]) -> str:
    status = _lower(event.get("status"))
    message = _lower(event.get("message"))
    if status in FAILED_RAW_STATUSES or "traceback" in message or "modulenotfound" in message:
        return "error"
    if status in {"warning", "manual_hardware_required"} or "manual" in message:
        return "warning"
    if status in SUCCESS_RAW_STATUSES:
        return "success"
    if status in PARTIAL_RAW_STATUSES:
        return "warning"
    if status in RUNNING_RAW_STATUSES:
        return "running"
    return "info"


def build_activity_feed(bundle: dict[str, Any], *, limit: int = 40) -> list[dict[str, Any]]:
    events = [e for e in (bundle.get("events") or []) if isinstance(e, dict)]
    if not events and bundle.get("launch"):
        launch = bundle.get("launch") or {}
        events = [
            {
                "ts": launch.get("created_at") or "",
                "step": "job_launched",
                "status": launch.get("status") or "running",
                "message": f"Build job launched for {launch.get('target_space') or 'target Space'}",
            }
        ]
    feed: list[dict[str, Any]] = []
    for event in events[-limit:]:
        feed.append(
            {
                "ts": event.get("ts") or event.get("created_at") or "",
                "step": event.get("step") or "",
                "status": event.get("status") or "",
                "message": event.get("message") or event.get("step") or "Event received",
                "severity": _severity_for_event(event),
                "agent": _agent_for_event(event),
            }
        )
    return feed


def build_diagnostics(bundle: dict[str, Any], status_model: dict[str, Any], phase: str) -> dict[str, Any]:
    summary = bundle.get("summary") or {}
    smoke = bundle.get("generation_smoke") or {}
    gate = bundle.get("inference_gate") or {}
    hardware = bundle.get("hardware_strategy") or {}
    blockers = bundle.get("technical_blockers") or {}
    blocker_items = blockers.get("blockers") if isinstance(blockers, dict) else []

    issue_title = ""
    issue_detail = ""
    issue_status = ""
    if blocker_items:
        first = blocker_items[0]
        issue_title = _first_nonempty(first.get("type"), first.get("name"), "Technical blocker") if isinstance(first, dict) else "Technical blocker"
        issue_detail = _first_nonempty(first.get("claim"), first.get("reason"), first.get("message")) if isinstance(first, dict) else str(first)
        issue_status = "open"
    elif status_model["requires_manual_action"]:
        issue_title = "Manual hardware required"
        issue_detail = "Automatic hardware selection was not available. Choose hardware in Space settings, then run Space Test."
        issue_status = "action_required"
    elif status_model["global_status"] == "failed":
        link_state = build_space_link_state(bundle)
        if link_state.get("links_ready"):
            issue_title = "Validation failed after Space upload"
            issue_detail = "The generated Space exists and remains available, but automatic validation or recovery ended in failure."
        else:
            issue_title = "Run failed"
            issue_detail = "Inspect logs and run artifacts for the failing step."
        issue_status = "open"

    health_passed = bool((gate.get("implementation_signals") or {}).get("health_passed") or smoke.get("health_passed"))
    smoke_ok = bool(smoke.get("ok") or smoke.get("status") == "success")
    build_ok = phase in {"api_validation", "live_wait", "generation_smoke", "inference_gate", "report_write", "done"} or status_model["global_status"] == "succeeded"

    return {
        "build_status": "passed" if build_ok else ("blocked" if status_model["global_status"] in {"blocked", "waiting_manual_action", "failed"} else "building"),
        "api_status": "passed" if smoke_ok else ("blocked" if status_model["global_status"] in {"blocked", "failed"} else "pending"),
        "tests_status": "passed" if smoke_ok or health_passed else ("blocked" if status_model["global_status"] in {"blocked", "failed"} else "pending"),
        "verdict_status": status_model["verdict"],
        "zerogpu_rules": [
            {"label": "Gradio interface", "status": "ok"},
            {"label": "Private Space", "status": "ok" if summary.get("target_space") else "pending"},
            {"label": "ZeroGPU-first strategy", "status": "ok" if "zero" in _lower(summary.get("selected_hardware") or hardware.get("preferred_space_hardware") or "zero") else "pending"},
            {"label": "Live API verification", "status": "ok" if smoke_ok else "pending"},
        ],
        "issue": {"title": issue_title, "detail": issue_detail, "status": issue_status},
    }


def _normalize_target_space_id(value: Any) -> str:
    text = str(value or "").strip()
    if not text:
        return ""
    text = text.replace("https://huggingface.co/spaces/", "").strip("/")
    text = text.split("/settings", 1)[0].strip("/")
    if ".hf.space" in text and "/" not in text:
        return ""
    return text if re.match(r"^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$", text) else ""


def _space_target_from_events(events: list[dict[str, Any]] | None, *, require_success: bool = False) -> str:
    for event in reversed(events or []):
        if not isinstance(event, dict):
            continue
        step = str(event.get("step") or "")
        status = str(event.get("status") or "").lower()
        if step not in {"create_space", "create_space_hardware", "upload_files", "runtime_upload_epoch"}:
            continue
        if require_success and status != "success":
            continue
        data = event.get("data") if isinstance(event.get("data"), dict) else {}
        target = _normalize_target_space_id(data.get("target_space") or data.get("target_space_id"))
        if target:
            return target
    return ""


def _space_target_from_bundle(bundle: dict[str, Any]) -> str:
    summary = bundle.get("summary") or {}
    state = bundle.get("state") or {}
    launch = bundle.get("launch") or {}
    runtime_upload_epoch = bundle.get("runtime_upload_epoch") or {}
    identity = bundle.get("space_identity") or {}
    return _normalize_target_space_id(_first_nonempty(
        identity.get("target_space"),
        summary.get("target_space"),
        state.get("target_space"),
        launch.get("target_space"),
        summary.get("target_space_id"),
        state.get("target_space_id"),
        launch.get("target_space_id"),
        runtime_upload_epoch.get("target_space_id"),
        _space_target_from_events(bundle.get("events") or []),
    ))


def _space_target_url_from_bundle(bundle: dict[str, Any], target: str = "") -> str:
    summary = bundle.get("summary") or {}
    state = bundle.get("state") or {}
    launch = bundle.get("launch") or {}
    identity = bundle.get("space_identity") or {}
    explicit = _first_nonempty(identity.get("target_space_url"), summary.get("target_space_url"), state.get("target_space_url"), launch.get("target_space_url"))
    if explicit:
        return explicit
    target = _normalize_target_space_id(target)
    return f"https://huggingface.co/spaces/{target}" if target else ""


def build_space_link_state(bundle: dict[str, Any], *, target: str = "", target_url: str = "") -> dict[str, Any]:
    """Return durable Space link readiness for Active Run.

    v198.26.9 invariant: Space links are a persisted run fact.  They must not
    depend on the frontend receiving the create/upload event inside a truncated
    recent-events window.  If a runtime upload epoch or Space runtime probe is
    present, a terminal failed run still gets Open Space / Settings links.
    """
    identity = bundle.get("space_identity") or {}
    runtime_upload_epoch = bundle.get("runtime_upload_epoch") or {}
    runtime_history = runtime_upload_epoch.get("history") if isinstance(runtime_upload_epoch.get("history"), list) else []
    space_runtime = bundle.get("space_runtime") or {}
    target = _space_target_from_bundle(bundle) if not target else _normalize_target_space_id(target)
    target_url = target_url or _space_target_url_from_bundle(bundle, target)
    create_event_ok = False
    upload_event_ok = False
    for event in bundle.get("events") or []:
        if not isinstance(event, dict):
            continue
        step = str(event.get("step") or "")
        status = str(event.get("status") or "").lower()
        if status == "success" and step in {"create_space", "create_space_hardware"}:
            create_event_ok = True
        if status == "success" and step in {"upload_files", "runtime_upload_epoch"}:
            upload_event_ok = True
    runtime_uploaded = bool(identity.get("runtime_uploaded") or identity.get("space_uploaded") or runtime_upload_epoch.get("last_upload_completed_at") or runtime_upload_epoch.get("upload_sequence") or runtime_history or upload_event_ok)
    space_runtime_known = bool(identity.get("space_runtime_known") or (isinstance(space_runtime, dict) and (space_runtime.get("stage") or space_runtime.get("status") or space_runtime.get("runtime_status") or space_runtime.get("updated_at") or space_runtime.get("url"))))
    space_created = bool(identity.get("space_created") or (target and (create_event_ok or runtime_uploaded or space_runtime_known)))
    links_ready = bool(target_url and (identity.get("links_ready") or space_created or runtime_uploaded or space_runtime_known))
    sources: list[str] = []
    if identity.get("source"):
        sources.append(str(identity.get("source")))
    if create_event_ok:
        sources.append("create_space_event")
    if upload_event_ok:
        sources.append("upload_event")
    if runtime_uploaded:
        sources.append("runtime_upload_epoch")
    if space_runtime_known:
        sources.append("space_runtime")
    return {
        "schema_version": "space_link_state.v198_26_9",
        "target_space": target,
        "target_space_id": target,
        "target_space_known": bool(target),
        "target_space_url": target_url,
        "target_space_settings_url": f"{target_url}/settings" if target_url else "",
        "space_created": bool(space_created),
        "runtime_uploaded": bool(runtime_uploaded),
        "space_uploaded": bool(runtime_uploaded),
        "space_runtime_known": bool(space_runtime_known),
        "links_ready": bool(links_ready),
        "can_open_space": bool(links_ready),
        "source": "+".join(dict.fromkeys([s for s in sources if s])) or "resolved_from_persisted_artifacts",
    }


def _space_links_ready_from_bundle(bundle: dict[str, Any]) -> bool:
    return bool(build_space_link_state(bundle).get("links_ready"))



def _number_from_value(value: Any) -> float | None:
    if value is None or value == "":
        return None
    if isinstance(value, (int, float)) and not isinstance(value, bool):
        number = float(value)
        return number if number > 0 else None
    if isinstance(value, dict):
        # Common shape: {"recommended_zero_gpu_duration_seconds": 40}
        for key in (
            "seconds",
            "duration_seconds",
            "recommended_zero_gpu_duration_seconds",
            "recommended_zerogpu_duration_seconds",
            "recommended_duration_seconds",
            "latency_seconds",
            "observed_latency_seconds",
        ):
            found = _number_from_value(value.get(key))
            if found is not None:
                return found
        return None
    text = str(value).strip().lower().replace("seconds", "").replace("second", "").replace("sec", "").replace("s", "")
    try:
        number = float(text)
        return number if number > 0 else None
    except Exception:
        return None


def _first_number(*values: Any) -> float | None:
    for value in values:
        number = _number_from_value(value)
        if number is not None:
            return number
    return None


def _latest_successful_linked_validation(bundle: dict[str, Any]) -> dict[str, Any]:
    linked = bundle.get("linked_validations") or {}
    rows = linked.get("validations") if isinstance(linked, dict) else []
    if not isinstance(rows, list):
        return {}
    successful = [row for row in rows if isinstance(row, dict) and _lower(row.get("status") or row.get("effective_status")) in {"success", "passed", "succeeded", "full_inference_success", "validated_after_space_test", "validated_after_manual_space_test"}]
    if not successful:
        return {}
    successful.sort(key=lambda row: str(row.get("validated_at") or row.get("updated_at") or row.get("created_at") or ""), reverse=True)
    return successful[0]


def build_validation_metrics(bundle: dict[str, Any]) -> dict[str, Any]:
    """Durable validation/latency model for Active Run.

    v198.26.9 invariant: latency and ZeroGPU duration are post-run facts, not
    incidental event text.  They are resolved once from persisted automatic
    smoke artifacts and linked Space Test artifacts, then exposed by /view and
    /progress so the frontend does not have to guess.
    """
    bundle = bundle or {}
    summary = bundle.get("summary") or {}
    state = bundle.get("state") or {}
    smoke = bundle.get("generation_smoke") or {}
    gate = bundle.get("inference_gate") or {}
    gate_smoke = gate.get("generation_smoke") if isinstance(gate.get("generation_smoke"), dict) else {}
    post_build = bundle.get("post_build_validation_status") or {}
    manual = bundle.get("manual_validation_status") or {}
    linked_success = _latest_successful_linked_validation(bundle)
    gate_recommendation = gate.get("zero_gpu_duration_recommendation") if isinstance(gate, dict) else None

    # Linked Space Test success is a stronger post-build measurement than the
    # original automatic build smoke.  It should update the parent Active Run.
    source_order = [linked_success, post_build if _lower(post_build.get("status")) == "success" else {}, manual if _lower(manual.get("status")) == "success" else {}, smoke, gate_smoke, gate, summary, state]
    latency = _first_number(*(src.get("latency_seconds") for src in source_order if isinstance(src, dict)), *(src.get("observed_latency_seconds") for src in source_order if isinstance(src, dict)))
    duration = _first_number(
        *(src.get("recommended_zero_gpu_duration_seconds") for src in source_order if isinstance(src, dict)),
        *(src.get("recommended_zerogpu_duration_seconds") for src in source_order if isinstance(src, dict)),
        *(src.get("recommended_duration_seconds") for src in source_order if isinstance(src, dict)),
        gate_recommendation,
    )
    api_name = _first_nonempty(*(src.get("api_name") for src in source_order if isinstance(src, dict)), gate.get("selected_api_name"), gate.get("selected_endpoint"))
    hardware = _first_nonempty(
        *(src.get("hardware_used_for_validation") for src in source_order if isinstance(src, dict)),
        *(src.get("recommendation_hardware") for src in source_order if isinstance(src, dict)),
        smoke.get("hardware"),
        gate.get("hardware"),
        summary.get("selected_hardware"),
        state.get("selected_hardware"),
    )
    output_artifact = _first_nonempty(*(src.get("output_artifact") or src.get("artifact_url") for src in source_order if isinstance(src, dict)))
    source = ""
    if linked_success or _lower(post_build.get("status")) == "success" or _lower(manual.get("status")) == "success":
        source = "linked_space_test"
    elif smoke.get("status") == "success" or smoke.get("ok") is True:
        source = "automatic_generation_smoke"
    elif gate_smoke or gate:
        source = "inference_gate"
    source = _first_nonempty(*(src.get("recommendation_source") for src in source_order if isinstance(src, dict)), source)
    health_passed = bool((gate.get("implementation_signals") or {}).get("health_passed") or smoke.get("health_passed") or gate_smoke.get("health_passed"))
    smoke_passed = bool(smoke.get("ok") or _lower(smoke.get("status")) == "success" or _lower(gate_smoke.get("status")) == "success" or (gate.get("implementation_signals") or {}).get("generation_smoke_passed") is True)
    validation_run_id = _first_nonempty(linked_success.get("validation_run_id"), linked_success.get("run_id"), post_build.get("validation_run_id"), manual.get("validation_run_id"))
    metrics = {
        "schema_version": "validation_metrics.v198_26_9",
        "available": bool(latency or duration or smoke_passed or health_passed or validation_run_id),
        "source": source,
        "latency_seconds": latency,
        "observed_latency_seconds": latency,
        "recommended_zero_gpu_duration_seconds": duration,
        "recommended_zerogpu_duration_seconds": duration,
        "hardware_used_for_validation": hardware,
        "recommendation_hardware": hardware,
        "api_name": api_name,
        "output_artifact": output_artifact,
        "health_passed": health_passed,
        "generation_smoke_passed": smoke_passed,
        "automatic_smoke_passed": bool(smoke.get("ok") or _lower(smoke.get("status")) == "success"),
        "linked_space_test_passed": bool(linked_success or _lower(post_build.get("status")) == "success" or _lower(manual.get("status")) == "success"),
        "validation_run_id": validation_run_id,
        "updated_at": _first_nonempty(linked_success.get("validated_at"), linked_success.get("updated_at"), post_build.get("validated_at"), post_build.get("updated_at"), manual.get("validated_at"), manual.get("updated_at"), summary.get("updated_at"), state.get("updated_at")),
    }
    return metrics

def _normalize_gradio_api_name(value: Any) -> str:
    text = str(value or "").strip()
    if not text:
        return ""
    if text.lower() in {"none", "null", "false"}:
        return ""
    if text.startswith("/"):
        return text
    return f"/{text}"


def _automatic_smoke_passed(bundle: dict[str, Any]) -> bool:
    smoke = bundle.get("generation_smoke") or {}
    state = bundle.get("state") or {}
    summary = bundle.get("summary") or {}
    summary_file = bundle.get("summary_file") or {}
    gate = bundle.get("inference_gate") or {}
    gate_smoke = gate.get("generation_smoke") if isinstance(gate.get("generation_smoke"), dict) else {}
    final = bundle.get("final_status_reconciliation") or {}
    test_result = bundle.get("test_result") or {}
    sources = (smoke, gate_smoke, final, summary, summary_file, state, test_result)
    for source in sources:
        if not isinstance(source, dict):
            continue
        if source.get("ok") is True:
            return True
        if _lower(source.get("status")) == "success" and (
            source is smoke
            or source is gate_smoke
            or source.get("latency_seconds") is not None
            or source.get("output_artifact")
            or source.get("artifact_url")
        ):
            return True
        if source.get("generation_smoke_passed") is True or source.get("smoke_test_passed") is True:
            return True
    signals = gate.get("implementation_signals") if isinstance(gate.get("implementation_signals"), dict) else {}
    return bool(signals.get("generation_smoke_passed") is True or gate.get("strong_full_inference_success") is True)


def _smoke_payload_available(bundle: dict[str, Any]) -> bool:
    for key in ("generation_smoke_payload", "generation_smoke_payload_retry", "resolved_validation_request", "validation_payload"):
        payload = bundle.get(key)
        if not isinstance(payload, dict):
            continue
        for payload_key in ("effective_args", "test_args", "args"):
            if isinstance(payload.get(payload_key), list):
                return True
        for payload_key in ("effective_kwargs", "test_kwargs", "kwargs"):
            if isinstance(payload.get(payload_key), dict):
                return True
        if payload.get("api_name") or payload.get("endpoint"):
            return True
    return False


def _known_gradio_endpoint_info(bundle: dict[str, Any]) -> dict[str, Any]:
    smoke = bundle.get("generation_smoke") or {}
    gate = bundle.get("inference_gate") or {}
    live = bundle.get("live_status") or {}
    schema = bundle.get("api_schema") or bundle.get("gradio_schema") or {}
    state = bundle.get("state") or {}
    summary = bundle.get("summary") or {}
    summary_file = bundle.get("summary_file") or {}
    final = bundle.get("final_status_reconciliation") or {}
    launch = bundle.get("launch") or {}
    payload = bundle.get("generation_smoke_payload") or {}
    payload_retry = bundle.get("generation_smoke_payload_retry") or {}
    resolved = bundle.get("resolved_validation_request") or {}
    promise = state.get("promise_validation") if isinstance(state.get("promise_validation"), dict) else {}
    gate_smoke = gate.get("generation_smoke") if isinstance(gate.get("generation_smoke"), dict) else {}
    state_smoke = state.get("generation_smoke") if isinstance(state.get("generation_smoke"), dict) else {}
    summary_smoke = summary.get("generation_smoke") if isinstance(summary.get("generation_smoke"), dict) else {}

    endpoints: list[str] = []
    for source in (smoke, gate_smoke, state_smoke, summary_smoke, gate, schema, live, state, summary, summary_file, final, launch, payload, payload_retry, resolved, promise):
        if not isinstance(source, dict):
            continue
        for key in ("discovered_api_names", "api_names", "endpoints", "api_endpoints", "named_endpoints", "available_api_names"):
            value = source.get(key)
            if isinstance(value, list):
                endpoints.extend(_normalize_gradio_api_name(item) for item in value if _normalize_gradio_api_name(item))
        for key in ("api_name", "endpoint", "selected_api_name", "selected_endpoint", "primary_api_name", "api_endpoint", "generation_endpoint"):
            endpoint_candidate = _normalize_gradio_api_name(source.get(key))
            if endpoint_candidate:
                endpoints.append(endpoint_candidate)
    # Deduplicate while preserving discovery order.
    endpoints = list(dict.fromkeys(e for e in endpoints if e))
    endpoint = _first_nonempty(*endpoints)
    runtime_success = _has_authoritative_runtime_success(bundle)
    smoke_passed = _automatic_smoke_passed(bundle)
    endpoint_known = bool(endpoint) or smoke_passed or runtime_success or _smoke_payload_available(bundle)
    return {"endpoint": endpoint or "/generate", "endpoint_count": len(endpoints), "endpoint_known": endpoint_known, "endpoints": endpoints[:20]}


def _contract_declares_no_full_inference(bundle: dict[str, Any]) -> bool:
    if _has_authoritative_runtime_success(bundle):
        return False
    contract = bundle.get("inference_contract") or bundle.get("INFERENCE_CONTRACT") or {}
    if not isinstance(contract, dict):
        contract = {}
    smoke = bundle.get("generation_smoke") or {}
    blockers = bundle.get("technical_blockers") or {}
    statuses = _raw_statuses(bundle)
    return bool(
        "technical_blocker_boot_only" in statuses
        or contract.get("full_inference_implemented") is False
        or (contract.get("primary_api_name") in {None, "", False} and _lower(contract.get("validation_level")) in {"boot-only", "boot_only", "health-only", "health_only"})
        or (_lower(smoke.get("status")) == "skipped" and _lower(smoke.get("skip_reason") or smoke.get("reason")) in {"contract_declared_no_full_inference", "full_inference_not_implemented"})
        or (_lower(blockers.get("status")) in {"technical_blocker_boot_only", "technical_blocker"} and contract.get("full_inference_implemented") is False)
    )


def _repair_auth_context(bundle: dict[str, Any], status_model: dict[str, Any]) -> bool:
    repair = bundle.get("repair_outcome") or {}
    summary = bundle.get("summary") or {}
    state = bundle.get("state") or {}
    values = {
        _lower(status_model.get("global_status")),
        _lower(status_model.get("verdict")),
        _lower(summary.get("status")),
        _lower(state.get("status")),
        _lower(summary.get("failure_type")),
        _lower(state.get("failure_type")),
        _lower(repair.get("post_repair_validation")),
        _lower(repair.get("failure_type")),
        _lower(repair.get("status")),
    }
    return bool(values.intersection(AUTH_RAW_STATUSES))


def build_space_test_policy(bundle: dict[str, Any], status_model: dict[str, Any]) -> dict[str, Any]:
    """Return the canonical linked Space Test policy for a Build Run.

    Space Test is linked-only: parent Build Run and target Space are mandatory,
    but a known endpoint is optional because the validation worker can discover
    Gradio endpoints before generating a smoke-test payload.
    """
    summary = bundle.get("summary") or {}
    state = bundle.get("state") or {}
    smoke = bundle.get("generation_smoke") or {}
    runtime = bundle.get("space_runtime") or {}
    target = _space_target_from_bundle(bundle)
    target_url = _space_target_url_from_bundle(bundle, target)
    raw_status = _lower(status_model.get("raw_status") or summary.get("status") or state.get("status"))
    global_status = _lower(status_model.get("global_status"))
    verdict = _lower(status_model.get("verdict"))
    endpoint_info = _known_gradio_endpoint_info(bundle)
    runtime_success = _has_authoritative_runtime_success(bundle)
    smoke_passed = _automatic_smoke_passed(bundle)
    requires_discovery = bool(target and not endpoint_info.get("endpoint_known"))
    runtime_stage = _lower(runtime.get("stage") or runtime.get("status")) if isinstance(runtime, dict) else ""
    health_passed = bool((bundle.get("inference_gate") or {}).get("implementation_signals", {}).get("health_passed") or smoke.get("health_passed"))

    def policy(mode: str, enabled: bool, reason: str, label: str, message: str, *, on_success: str = "unchanged", on_failure: str = "unchanged", allow_replay: bool = False, allow_recovery: bool = False) -> dict[str, Any]:
        return {
            "schema_version": "space_test_policy.v1",
            "mode": mode,
            "enabled": bool(enabled),
            "reason": reason,
            "label": label,
            "message": message,
            "target_space": target,
            "target_space_url": target_url,
            "parent_build_run_id": summary.get("run_id") or state.get("run_id") or "",
            "endpoint": endpoint_info.get("endpoint") or "/generate",
            "endpoint_known": bool(endpoint_info.get("endpoint_known")),
            "endpoint_count": endpoint_info.get("endpoint_count") or 0,
            "requires_endpoint_discovery": bool(requires_discovery and enabled),
            "can_update_parent_effective_status": on_success != "unchanged",
            "effective_status_on_success": on_success,
            "effective_status_on_failure": on_failure,
            "allow_replay": bool(allow_replay),
            "allow_recovery": bool(allow_recovery),
        }

    if not target:
        return policy("unavailable", False, "no_target_space", "Space Test unavailable", "No generated Space is available to validate.")
    # v198.26.16: runtime proof beats stale Pi/contract policy blockers.  If the
    # build has a terminal full-inference proof or a successful automatic smoke
    # replay source, the Space Test panel must expose a replayable validation
    # context instead of resurrecting "no generation endpoint".
    if runtime_success or (smoke_passed and (global_status == "succeeded" or verdict in {"passed", "success", "full_inference_success"})):
        replay = policy("replay", True, "automatic_smoke_passed", "Replay automatic smoke", "Automatic smoke test passed. You can replay the validated request.", allow_replay=True)
        replay["runtime_proof_authoritative"] = bool(runtime_success)
        replay["automatic_smoke_passed"] = bool(smoke_passed)
        replay["no_generation_endpoint"] = False
        replay["can_retry_schema"] = True
        return replay
    if _repair_auth_context(bundle, status_model):
        return policy("recover", True, "auth_refresh_required", "Retry after sign-in refresh", "Repair patch may have been uploaded, but validation could not continue because HF OAuth expired. Sign in again, then retry linked validation.", on_success="recovered_by_manual_validation", allow_recovery=True)
    if _contract_declares_no_full_inference(bundle):
        blocked = policy("blocked", False, "no_generation_endpoint_by_contract", "Blocked — no generation endpoint", "Health may pass, but Pi declared full inference unavailable and no generation endpoint exists. Review TECHNICAL_BLOCKERS.json / PI_SUMMARY.md or provide a dedicated implementation and hardware plan.")
        blocked["endpoint"] = ""
        blocked["requires_endpoint_discovery"] = False
        blocked["no_generation_endpoint"] = True
        blocked["can_retry_schema"] = False
        return blocked
    if global_status in {"running", "unknown"}:
        return policy("blocked", False, "build_not_terminal", "Available after final Space state", "Space Test becomes available after the Build Run reaches a final Space state.")
    if global_status == "cancelled" or raw_status in {"stopped", "cancelled", "canceled"}:
        return policy("unavailable", False, "run_stopped", "Space Test unavailable", "This run was stopped. Start a new build before validation.")
    if status_model.get("manual_validation_passed") or _lower(status_model.get("effective_status")) in {"validated_after_manual_space_test", "manual_validation_passed", "validated"}:
        return policy("replay", True, "manual_validation_already_passed", "Replay linked validation", "This Build Run already has a successful linked Space Test. You can replay the validation.", allow_replay=True)
    if raw_status in {"completed_with_warnings", "success_with_warnings"}:
        return policy("replay", True, "completed_with_warnings", "Replay validation", "Build completed with warnings. Space Test can replay or confirm the validated endpoint.", allow_replay=True)
    if global_status == "succeeded" or verdict in {"passed", "success", "full_inference_success"}:
        replay = policy("replay", True, "automatic_smoke_passed", "Replay automatic smoke", "Automatic smoke test passed. You can replay the validated request.", allow_replay=True)
        replay["automatic_smoke_passed"] = bool(smoke_passed)
        replay["no_generation_endpoint"] = False
        replay["can_retry_schema"] = True
        return replay
    if global_status == "partial" or verdict == "partial_validation" or raw_status in PARTIAL_RAW_STATUSES:
        return policy("complete", True, "partial_generation_not_verified", "Complete validation", "Automatic smoke did not verify generation. Run linked Space Test to complete validation.", on_success="validated_after_manual_space_test")
    if global_status == "waiting_manual_action" or status_model.get("requires_manual_action"):
        reachable = "running" in runtime_stage or health_passed or endpoint_info.get("endpoint_known")
        if reachable:
            return policy("complete", True, "manual_hardware_action_possible", "Validate after manual action", "Manual hardware action may be required. If the Space is now running, run linked validation.", on_success="validated_after_manual_space_test")
        return policy("blocked", False, "manual_hardware_required", "Manual hardware action required", "Complete the hardware action in Space Settings before validation.")
    if global_status == "failed":
        return policy("recover", True, "failed_after_space_creation", "Recover validation", "The generated Space previously failed validation. If it is now reachable, run linked Space Test to check recovery.", on_success="recovered_by_manual_validation", allow_recovery=True)
    if global_status == "stale":
        return policy("recover", True, "stale_with_target_space", "Recover stale run", "The original run became stale. If the Space is now available, run linked validation to recover the result.", on_success="recovered_by_manual_validation", allow_recovery=True)
    return policy("blocked", False, "unsupported_build_state", "Space Test unavailable", "This Build Run state is not eligible for linked Space Test yet.")


def build_space_test_model(bundle: dict[str, Any], status_model: dict[str, Any]) -> dict[str, Any]:
    smoke = bundle.get("generation_smoke") or {}
    summary = bundle.get("summary") or {}
    metrics = build_validation_metrics(bundle)
    policy = build_space_test_policy(bundle, status_model)
    # v191.12 regression guard: a contract-declared boot-only blocker must not
    # resurrect the historical /generate fallback in the display model. Keep the
    # endpoint intentionally blank when the parent policy says no generation
    # endpoint exists by contract. Other states still use the legacy fallback so
    # endpoint discovery/recovery behavior is unchanged.
    if policy.get("no_generation_endpoint") is True or policy.get("reason") == "no_generation_endpoint_by_contract":
        endpoint = ""
    else:
        endpoint = policy.get("endpoint") or smoke.get("api_name") or smoke.get("endpoint") or "/generate"
    policy_blocks = policy.get("enabled") is False and policy.get("mode") in {"blocked", "unavailable"}
    parent_blocked = bool(policy_blocks and (_lower(status_model.get("global_status")) in {"blocked", "failed", "auth_refresh_required", "cancelled"} or _lower(status_model.get("verdict")) in BLOCKED_RAW_STATUSES | FAILED_RAW_STATUSES | AUTH_RAW_STATUSES | {"cancelled", "stopped"}))
    smoke_evidence_passed = bool(metrics.get("generation_smoke_passed") or smoke.get("ok") or smoke.get("status") == "success" or policy.get("automatic_smoke_passed") is True)
    preview_status = "blocked" if parent_blocked else "passed" if smoke_evidence_passed or policy.get("mode") == "replay" else "pending"
    return {
        "target_space": policy.get("target_space") or summary.get("target_space") or "",
        "target_space_url": policy.get("target_space_url") or summary.get("target_space_url") or "",
        "endpoint": endpoint,
        "status": preview_status,
        "smoke_evidence_passed": smoke_evidence_passed,
        "latency_seconds": metrics.get("latency_seconds") or smoke.get("latency_seconds") or summary.get("latency_seconds"),
        "observed_latency_seconds": metrics.get("observed_latency_seconds"),
        "recommended_zero_gpu_duration_seconds": metrics.get("recommended_zero_gpu_duration_seconds"),
        "recommended_zerogpu_duration_seconds": metrics.get("recommended_zerogpu_duration_seconds"),
        "validation_metrics": metrics,
        "expected_output_type": summary.get("expected_output_type") or smoke.get("expected_output_type") or "",
        "verdict": status_model["verdict"],
        "output_artifact": smoke.get("output_artifact") or smoke.get("artifact_url") or "",
        "policy": policy,
    }


def build_run_view_model(
    run_id: str,
    bundle: dict[str, Any],
    *,
    bucket_source: str,
    now: datetime | None = None,
) -> dict[str, Any]:
    summary = bundle.get("summary") or {}
    state = bundle.get("state") or {}
    manual_validation = bundle.get("manual_validation_status") or summary.get("manual_validation_status") or {}
    status_model = normalize_run_status(bundle, now=now)
    effective_run_status = bundle.get("effective_run_status") if isinstance(bundle.get("effective_run_status"), dict) else compute_effective_run_status(bundle, build_status=status_model.get("verdict") or status_model.get("global_status"), build_verdict=status_model.get("verdict"))
    if effective_run_status.get("post_build_status") not in {"", "none", None}:
        legacy_status = effective_run_status.get("legacy_effective_status") or manual_validation.get("effective_status") or "validated_after_manual_space_test"
        status_model = {
            **status_model,
            "manual_validation_passed": True,
            "post_build_validation_passed": True,
            "post_build_status": effective_run_status.get("post_build_status"),
            "effective_status": effective_run_status.get("effective_status") or legacy_status,
            "effective_verdict": effective_run_status.get("effective_verdict") or legacy_status,
            "legacy_effective_status": legacy_status,
        }
    phase = derive_product_phase(bundle, status_model)
    space_test_policy = build_space_test_policy(bundle, status_model)
    elapsed_seconds = None
    current_now = now or datetime.now(timezone.utc)
    created = parse_ts(summary.get("started_at") or state.get("started_at") or summary.get("created_at") or state.get("created_at"))
    latest = parse_ts(status_model.get("latest_activity_at"))
    if created:
        end = latest if status_model.get("is_terminal") and latest else current_now
        elapsed_seconds = int((end - created).total_seconds())
    target = _space_target_from_bundle(bundle)
    target_url = _space_target_url_from_bundle(bundle, target)
    space_link_state = build_space_link_state(bundle, target=target, target_url=target_url)
    validation_metrics = build_validation_metrics({**bundle, "summary": summary})
    timeline_model = build_run_timeline_model(bundle)
    runtime_stage_history = ((timeline_model.get("live_validation") or {}).get("runtime_stage_history") or {}) if isinstance(timeline_model, dict) else {}
    links = {
        "job_url": _job_url_from_view_sources(bucket_source=bucket_source, summary=summary, state=state, launch=bundle.get("launch") or {}),
        "target_space_url": target_url,
        "target_space_settings_url": f"{target_url}/settings" if target_url else "",
        "artifacts_url": summary.get("artifacts_url") or f"https://huggingface.co/buckets/{bucket_source}/tree/{_runs_prefix()}/{run_id}",
        "links_ready": bool(space_link_state.get("links_ready")),
        "space_created": bool(space_link_state.get("space_created")),
        "space_uploaded": bool(space_link_state.get("runtime_uploaded")),
        "runtime_uploaded": bool(space_link_state.get("runtime_uploaded")),
        "space_runtime_known": bool(space_link_state.get("space_runtime_known")),
        "source": space_link_state.get("source") or "",
    }
    return {
        "schema_version": "run_view_model.v1",
        "run_id": run_id,
        "bucket_source": bucket_source,
        "header": {
            "title": _first_nonempty(summary.get("title"), f"Build Space for {summary.get('model_id')}", run_id),
            "status": status_model["global_status"],
            "status_label": status_model["global_status"].replace("_", " ").title(),
            "display_status": effective_run_status.get("display_status") or status_model.get("global_status"),
            "display_label": effective_run_status.get("display_label") or status_model["global_status"].replace("_", " ").title(),
            "raw_status": status_model["raw_status"],
            "space": target or summary.get("target_space") or "",
            "space_url": target_url,
            "current_phase": phase,
            "current_phase_label": next((s["label"] for s in PRODUCT_STEPS if s["id"] == phase), phase),
            "elapsed_seconds": max(0, elapsed_seconds or 0),
            "started_at": summary.get("created_at") or state.get("created_at") or "",
            "updated_at": status_model["latest_activity_at"],
        },
        "status_model": status_model,
        "effective_run_status": effective_run_status,
        "pipeline": build_pipeline(phase, status_model),
        "timeline_model": timeline_model,
        "runtime_stage_history": runtime_stage_history,
        "activity": build_activity_feed(bundle),
        "diagnostics": build_diagnostics(bundle, status_model, phase),
        "space_test": {**build_space_test_model(bundle, status_model), "linked_validation": manual_validation, "mode": "linked_build_run"},
        "space_test_policy": space_test_policy,
        "space_link_state": space_link_state,
        "validation_metrics": validation_metrics,
        "run_documents": bundle.get("run_documents") or [],
        "manual_validation": manual_validation,
        "post_build_validation": effective_run_status.get("post_build_validation") or {},
        "display_status": effective_run_status.get("display_status") or status_model.get("global_status"),
        "effective_status": effective_run_status.get("effective_status") or status_model.get("effective_status") or status_model["verdict"],
        "effective_verdict": effective_run_status.get("effective_verdict") or status_model.get("effective_verdict") or status_model["verdict"],
        "actions": {
            "can_resume": status_model["global_status"] in {"running", "stale", "unknown"},
            "can_stop": status_model["global_status"] in {"running", "queued", "unknown"},
            "can_open_space": bool(space_link_state.get("can_open_space")),
            "can_validate": bool(space_test_policy.get("enabled")),
            "requires_manual_action": status_model["requires_manual_action"],
        },
        "links": links,
    }


def is_resumable_summary(summary: dict[str, Any], *, now: datetime | None = None, stale_after: timedelta = timedelta(hours=6)) -> bool:
    """Best-effort resumability test for lightweight /api/runs summaries."""
    status = _lower(summary.get("status"))
    if status in SUCCESS_RAW_STATUSES or status in PARTIAL_RAW_STATUSES or status in FAILED_RAW_STATUSES or status in BLOCKED_RAW_STATUSES or status in MANUAL_RAW_STATUSES:
        return False
    if "success" in status or "failed" in status or "blocker" in status or "manual" in status:
        return False
    now = now or datetime.now(timezone.utc)
    latest = parse_ts(summary.get("updated_at") or summary.get("created_at"))
    if latest and now - latest > stale_after:
        return False
    return status in RUNNING_RAW_STATUSES or status in {"unknown", ""}


def find_latest_resumable_run(summaries: list[dict[str, Any]], *, now: datetime | None = None) -> dict[str, Any] | None:
    candidates = [s for s in summaries if isinstance(s, dict) and is_resumable_summary(s, now=now)]
    if not candidates:
        return None
    return sorted(candidates, key=lambda s: str(s.get("updated_at") or s.get("created_at") or s.get("run_id")), reverse=True)[0]