Upload 14 files

src/auth.py

@@ -46,6 +46,64 @@ class OAuthContext:
         return self.expires_at <= datetime.now(timezone.utc)
 
 
+def oauth_lifetime_summary(ctx: OAuthContext, *, now: datetime | None = None) -> dict[str, Any]:
+    """Return a safe, UI-ready OAuth lifetime summary without exposing tokens."""
+    current = now or datetime.now(timezone.utc)
+    if current.tzinfo is None:
+        current = current.replace(tzinfo=timezone.utc)
+    if ctx.expires_at is None:
+        return {
+            "status": "unknown",
+            "severity": "neutral",
+            "label": "HF Auth: expiry unknown",
+            "expires_at": None,
+            "seconds_until_expiry": None,
+            "recommendation": "OAuth expiry is not exposed by this runtime. Refresh sign-in before long builds if unsure.",
+        }
+    seconds = int((ctx.expires_at - current).total_seconds())
+    if seconds <= 0:
+        status = "expired"
+        severity = "error"
+        label = "HF Auth: expired"
+        recommendation = "Sign in again before launching or validating Jobs."
+    elif seconds < 30 * 60:
+        status = "critical"
+        severity = "error"
+        label = f"HF Auth: {format_duration_compact(seconds)} left"
+        recommendation = "Refresh sign-in before launching a build, repair, or linked Space Test."
+    elif seconds < 90 * 60:
+        status = "warning"
+        severity = "warn"
+        label = f"HF Auth: {format_duration_compact(seconds)} left"
+        recommendation = "Refresh sign-in before long or high-risk builds."
+    else:
+        status = "ok"
+        severity = "success"
+        label = f"HF Auth: {format_duration_compact(seconds)} left"
+        recommendation = "OAuth session looks safe for normal builds."
+    return {
+        "status": status,
+        "severity": severity,
+        "label": label,
+        "expires_at": ctx.expires_at.isoformat(),
+        "seconds_until_expiry": seconds,
+        "recommendation": recommendation,
+    }
+
+
+def format_duration_compact(seconds: int | float | None) -> str:
+    if seconds is None:
+        return "unknown"
+    total = max(0, int(seconds))
+    hours, rem = divmod(total, 3600)
+    minutes = rem // 60
+    if hours:
+        return f"{hours}h {minutes}m"
+    if minutes:
+        return f"{minutes}m"
+    return f"{total}s"
+
+
 def _parse_scope(scope: Any) -> set[str]:
     if not scope:
         return set()
@@ -167,6 +225,7 @@ def public_oauth_context(ctx: OAuthContext) -> dict[str, Any]:
         "scopes": sorted(ctx.scopes),
         "missing_scopes": ctx.missing_scopes,
         "expires_at": ctx.expires_at.isoformat() if ctx.expires_at else None,
+        "auth_lifetime": oauth_lifetime_summary(ctx),
         "authenticated": True,
     }
 

src/bucket.py

@@ -1073,6 +1073,8 @@ def read_run_bundle(run_id: str, *, bucket_source: str, token: str | None = None
         "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),
+        "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),

src/eval_archive.py

@@ -9,7 +9,7 @@ from .bucket import RunPaths, read_json, read_text, write_json
 from .eval_config import effective_eval_config
 from .security import redact
 
-_TERMINAL = {"success", "done", "completed", "failed", "failure", "error", "cancelled", "canceled", "manual", "stale", "partial", "partial_validation", "completed_with_warnings", "success_with_warnings", "validated", "validated_after_manual_space_test", "recovered_by_manual_validation", "validated_after_stale_run", "manual_validation_passed", "full_inference_success", "full_inference_candidate_health_passed", "health_only", "technical_blocker", "manual_hardware_required"}
+_TERMINAL = {"success", "done", "completed", "failed", "failure", "error", "cancelled", "canceled", "manual", "stale", "partial", "partial_validation", "completed_with_warnings", "success_with_warnings", "validated", "validated_after_manual_space_test", "recovered_by_manual_validation", "validated_after_stale_run", "manual_validation_passed", "full_inference_success", "full_inference_candidate_health_passed", "health_only", "technical_blocker", "technical_blocker_boot_only", "manual_hardware_required"}
 
 
 def _now() -> str:

src/model_scan.py

@@ -188,6 +188,192 @@ def _extract_model_card_signals(readme: str, model_index: dict[str, Any]) -> dic
     }
 
 
+
+
+
+
+NATIVE_KERNEL_PATTERNS: tuple[tuple[str, tuple[str, ...], str], ...] = (
+    ("flash_attn", ("flash-attn", "flash_attn", "flash attention", "flashattention", "flash-attention", "enable_flashattn", "flash_attention_2"), "Flash Attention dependency or runtime flag"),
+    ("xformers", ("xformers", "memory_efficient_attention"), "xFormers attention dependency"),
+    ("triton", ("triton", "triton kernel", "@triton", "triton.jit"), "Triton/fused kernel dependency"),
+    ("custom_cuda", ("cuda extension", "custom cuda", "cpp_extension", "setup.py build_ext", "fused kernel", "fused ops", "custom kernel"), "Custom native/CUDA extension"),
+    ("attention_interface", ("attentioninterface", "attention interface", "attn_implementation"), "Transformers attention backend hook"),
+    ("hf_kernels", ("hf kernels", "kernel hub", "hugging face kernels", "kernels-community", "from kernels import", "pip install kernels"), "HF Kernels / Kernel Hub mention"),
+)
+
+
+def build_kernel_strategy(*, readme: str | None = None, files: list[str] | None = None) -> dict[str, Any]:
+    """Return a visibility-only mitigation plan for native kernel dependencies.
+
+    This is metadata/readme based. It must not inject packages or mark a model
+    terminally blocked by itself; it gives Pi and the UI a safer strategy than
+    blindly adding source-built CUDA packages to requirements.txt.
+    """
+    text = (readme or "").lower()
+    file_list = [str(f).lower() for f in (files or [])]
+    detected: list[str] = []
+    signals: list[str] = []
+    for key, patterns, label in NATIVE_KERNEL_PATTERNS:
+        if any(pattern in text for pattern in patterns):
+            detected.append(key)
+            signals.append(label)
+    if any(f.endswith((".cu", ".cuh")) or "/csrc/" in f or f.startswith("csrc/") for f in file_list):
+        if "custom_cuda" not in detected:
+            detected.append("custom_cuda")
+            signals.append("Native CUDA/C++ source files in repository")
+    if any(f.endswith((".cpp", ".cc")) and ("cuda" in f or "/csrc/" in f or f.startswith("csrc/")) for f in file_list):
+        if "custom_cuda" not in detected:
+            detected.append("custom_cuda")
+            signals.append("Native C++ extension source files in repository")
+
+    native_risk = any(k in detected for k in {"flash_attn", "xformers", "triton", "custom_cuda"})
+    candidate_backends: list[str] = []
+    if native_risk:
+        candidate_backends.extend(["torch_sdpa", "hf_kernels", "xformers_wheel"])
+    if "attention_interface" in detected:
+        candidate_backends.append("transformers_attention_interface")
+    if "hf_kernels" in detected and "hf_kernels" not in candidate_backends:
+        candidate_backends.append("hf_kernels")
+
+    rejected_actions = []
+    if native_risk:
+        rejected_actions.append({
+            "action": "blind_pip_install_native_cuda_package",
+            "reason": "Source-built CUDA/native packages are fragile in Spaces and can mismatch the managed PyTorch/CUDA runtime.",
+        })
+    if "flash_attn" in detected:
+        rejected_actions.append({
+            "action": "pip_install_flash_attn_without_fallback",
+            "reason": "Prefer PyTorch SDPA, Transformers AttentionInterface, HF Kernels/Kernel Hub, or a compatible wheel before forcing flash-attn source builds.",
+        })
+
+    selected = "none"
+    if native_risk:
+        selected = "prefer_runtime_backends_before_source_builds"
+    elif "hf_kernels" in detected:
+        selected = "hf_kernels_available_if_model_uses_supported_kernel"
+
+    return {
+        "schema_version": "kernel_strategy.v1",
+        "native_kernel_risk": bool(native_risk),
+        "detected_dependencies": detected,
+        "signals": signals[:10],
+        "selected_strategy": selected,
+        "candidate_backends": candidate_backends[:8],
+        "rejected_actions": rejected_actions,
+        "requires_manual_review": bool("custom_cuda" in detected),
+        "pi_instruction": (
+            "Do not compile native CUDA packages blindly. Prefer PyTorch SDPA, compatible wheels, HF Kernels/Kernel Hub, "
+            "Transformers AttentionInterface, or Diffusers attention processors when they match the required operation. "
+            "Declare a technical blocker if a strict native extension has no plausible fallback."
+            if native_risk else
+            "No native kernel dependency was detected by the metadata scan."
+        ),
+        "visibility_only": True,
+    }
+
+
+def _has_any(text: str, patterns: tuple[str, ...]) -> bool:
+    lower = text.lower()
+    return any(pattern in lower for pattern in patterns)
+
+
+def _build_complexity_assessment(
+    *,
+    pipeline_tag: str | None,
+    library_name: str | None,
+    tags: list[str],
+    files: list[str],
+    readme: str,
+    expected_output_type: str | None,
+    custom_code: bool,
+    gated: Any,
+) -> dict[str, Any]:
+    """Estimate build/runtime risk without changing the existing pre-scan verdict.
+
+    This is deliberately a lightweight metadata/model-card heuristic. It helps
+    the UI warn users before long Jobs; it is not a hard launch gate.
+    """
+    lower_readme = (readme or "").lower()
+    lower_files = [str(f).lower() for f in files]
+    tag_set = {str(t).lower() for t in tags}
+    task = (pipeline_tag or "").lower()
+    library = (library_name or "").lower()
+
+    points = 0
+    signals: list[str] = []
+    mitigations: list[str] = []
+
+    def add(points_delta: int, signal: str, mitigation: str | None = None) -> None:
+        nonlocal points
+        points += points_delta
+        if signal not in signals:
+            signals.append(signal)
+        if mitigation and mitigation not in mitigations:
+            mitigations.append(mitigation)
+
+    if expected_output_type == "video" or "video" in task or {"text-to-video", "image-to-video"} & tag_set:
+        add(3, "video or image-to-video output", "Expect a long build/repair cycle and prefer strong fallback hardware.")
+    if any(word in lower_readme for word in ("avatar", "audio-driven", "audio driven", "talking head", "lip sync", "lip-sync")):
+        add(2, "audio/video avatar workflow", "Refresh sign-in before launch and expect larger validation payloads.")
+    if custom_code or "trust_remote_code" in tag_set or "custom_code" in tag_set:
+        add(2, "custom code or trust_remote_code required", "Review generated code and blockers before paid hardware attempts.")
+    if gated in {True, "auto", "manual"} or str(gated).lower() in {"true", "auto", "manual"}:
+        add(1, "gated model access", "Confirm the same HF account has accepted access terms.")
+
+    if _has_any(lower_readme, ("torchrun", "nproc_per_node", "distributed", "init_process_group", "nccl", "context_parallel", "tensor parallel", "pipeline parallel", "multi-gpu", "multi gpu")):
+        add(4, "multi-GPU / distributed runtime hints", "Treat ZeroGPU as unlikely unless Pi can prove a single-GPU refactor.")
+    if _has_any(lower_readme, ("flash-attn", "flash_attn", "flash attention", "flashattention", "flash-attention")):
+        add(2, "flash-attn or custom attention dependency", "Prefer PyTorch SDPA, xformers wheels, or HF Kernels before source builds.")
+    if _has_any(lower_readme, ("xformers", "triton", "cuda extension", "fused kernel", "fused ops", "custom kernel")) or any(f.endswith((".cu", ".cpp")) or "/csrc/" in f for f in lower_files):
+        add(2, "native CUDA/kernel dependency risk", "Prefer runtime-compatible wheels or HF Kernels over compiling during Space build.")
+    if "ffmpeg" in lower_readme or any("ffmpeg" in f for f in lower_files):
+        add(1, "ffmpeg or system media dependency", "Ensure generated app declares media/system requirements clearly.")
+    if _has_any(lower_readme, ("conda ", "mamba ", "apt-get", "sudo apt", "pip install -e", "git clone")):
+        add(1, "non-standard install instructions", "Pi should vendor required code or simplify requirements for Spaces.")
+
+    weight_files = [f for f in lower_files if f.endswith(WEIGHT_EXTENSIONS)]
+    safetensor_shards = [f for f in lower_files if f.endswith(".safetensors")]
+    if len(weight_files) >= 20 or len(safetensor_shards) >= 12:
+        add(2, "many weight shards", "Expect longer cold start and validation windows.")
+    elif len(weight_files) >= 8 or len(safetensor_shards) >= 6:
+        add(1, "multiple weight shards", "Allow extra boot time before validation.")
+
+    vram_match = re.search(r"\b(?:vram|gpu memory|memory)\D{0,16}([3-9]\d|1\d{2})\s*(?:gb|gib)\b|\b([3-9]\d|1\d{2})\s*(?:gb|gib)\s*(?:vram|gpu)", lower_readme)
+    if vram_match:
+        add(3, "high VRAM mentioned in model card", "Prefer fixed GPU fallback and refresh sign-in before launch.")
+
+    # Keep the assessment independent from the existing pass/fail scan score.
+    if points >= 9:
+        level = "very_high"
+        label = "Very high risk"
+        recommended_seconds = 180 * 60
+    elif points >= 6:
+        level = "high"
+        label = "High risk"
+        recommended_seconds = 120 * 60
+    elif points >= 3:
+        level = "medium"
+        label = "Medium risk"
+        recommended_seconds = 60 * 60
+    else:
+        level = "low"
+        label = "Low risk"
+        recommended_seconds = 30 * 60
+
+    return {
+        "schema_version": "model_build_risk.v1",
+        "level": level,
+        "label": label,
+        "score": max(0, points),
+        "recommended_session_seconds": recommended_seconds,
+        "recommended_session_minutes": recommended_seconds // 60,
+        "signals": signals[:10],
+        "mitigations": mitigations[:6],
+        "summary": f"{label}; recommended HF session remaining: {recommended_seconds // 60}m+.",
+        "visibility_only": True,
+    }
+
 def _classify(score: int, *, blocking: bool = False) -> str:
     if blocking:
         return "unsupported"
@@ -356,6 +542,24 @@ def analyze_model_metadata(
 
     score = max(0, min(100, score))
     verdict = _classify(score, blocking=unsupported)
+    build_risk = _build_complexity_assessment(
+        pipeline_tag=pipeline_tag,
+        library_name=library_name,
+        tags=tags,
+        files=files,
+        readme=readme,
+        expected_output_type=expected_output_type,
+        custom_code=custom_code,
+        gated=gated,
+    )
+    kernel_strategy = build_kernel_strategy(readme=readme, files=files)
+    if kernel_strategy.get("native_kernel_risk"):
+        recommendations.append("Native kernel risk detected; prefer PyTorch SDPA, HF Kernels/Kernel Hub, compatible wheels, or model-specific fallbacks before source builds.")
+    if build_risk["level"] in {"high", "very_high"}:
+        recommendations.append(
+            f"Refresh HF sign-in before launch; this scan recommends {build_risk['recommended_session_minutes']}m+ remaining for this model."
+        )
+
     if not recommendations:
         if verdict == "safe":
             recommendations.append("Good candidate for Strict inference.")
@@ -381,6 +585,8 @@ def analyze_model_metadata(
         "risk_signals": risk[:10],
         "recommendations": recommendations[:6],
         "expected_output_type": expected_output_type or "",
+        "build_risk": build_risk,
+        "kernel_strategy": kernel_strategy,
         "metadata": {
             "pipeline_tag": pipeline_tag or "",
             "library_name": library_name or "",
@@ -397,6 +603,10 @@ def analyze_model_metadata(
             "runtime_hints": runtime_hints,
             "expected_output_type": expected_output_type or "",
             "diffusers_standard": bool(is_diffusers and has_model_index and has_safetensors),
+            "build_risk_level": build_risk["level"],
+            "recommended_session_minutes": build_risk["recommended_session_minutes"],
+            "native_kernel_risk": bool(kernel_strategy.get("native_kernel_risk")),
+            "kernel_strategy": kernel_strategy,
         },
     }
 

src/progress.py

@@ -36,6 +36,7 @@ STEP_ORDER = [
     "repair_upload",
     "repair_validation",
     "technical_blocker",
+    "technical_blocker_boot_only",
     "manual_hardware_required",
     "upload_files",
     "space_runtime",
@@ -80,6 +81,7 @@ STEP_LABELS = {
     "repair_upload": "Repair upload",
     "repair_validation": "Repair validation",
     "technical_blocker": "Technical blocker",
+    "technical_blocker_boot_only": "Boot-only blocker",
     "manual_hardware_required": "Manual hardware",
     "upload_files": "Upload files",
     "space_runtime": "Space runtime",
@@ -116,6 +118,7 @@ DONE_STATUSES = {
     "full_inference_success",
     "manual_hardware_required",
     "technical_blocker",
+    "technical_blocker_boot_only",
 }
 PARTIAL_STATUSES = {
     "full_inference_candidate_health_passed",
@@ -341,7 +344,7 @@ def progress_from_events(events: list[dict[str, Any]] | None, *, state: dict[str
         "current_step": current_step,
         "current_step_label": STEP_LABELS.get(current_step, current_step),
         "last_event": last_message or "No events yet",
-        "visual_status": "error" if overall_status in FAILED_STATUSES else ("warn" if overall_status in PARTIAL_STATUSES else ("stopped" if overall_status in CANCELLED_STATUSES else ("success" if overall_status in DONE_STATUSES and overall_status not in {"manual_hardware_required", "technical_blocker"} else ("stopped" if overall_status in {"manual_hardware_required", "technical_blocker"} else ("running" if overall_status == "running" else "neutral"))))),
+        "visual_status": "error" if overall_status in FAILED_STATUSES else ("warn" if overall_status in PARTIAL_STATUSES else ("stopped" if overall_status in CANCELLED_STATUSES else ("success" if overall_status in DONE_STATUSES and overall_status not in {"manual_hardware_required", "technical_blocker", "technical_blocker_boot_only"} else ("stopped" if overall_status in {"manual_hardware_required", "technical_blocker", "technical_blocker_boot_only"} else ("running" if overall_status == "running" else "neutral"))))),
         "elapsed_seconds": max(0, elapsed),
         "eta_seconds": None,
         "timeline": timeline,

src/timeline_model.py

@@ -51,7 +51,7 @@ PHASE_STEPS = {
         "repair_validation",
     },
     "archive": {"report_write", "artifact_manifest", "anonymous_eval", "eval_publish", "eval_archive"},
-    "done": {"done", "failure", "technical_blocker", "manual_hardware_required"},
+    "done": {"done", "failure", "technical_blocker", "technical_blocker_boot_only", "manual_hardware_required"},
 }
 
 STEP_TO_PHASE = {step: phase for phase, steps in PHASE_STEPS.items() for step in steps}
@@ -62,7 +62,8 @@ FAILED_STATUSES = {"failed", "failure", "error"}
 RUNNING_STATUSES = {"started", "running", "waiting", "pending", "scheduled"}
 CANCELLED_STATUSES = {"cancelled", "canceled"}
 MANUAL_STATUSES = {"manual_hardware_required", "manual_action_required", "generated_needs_manual_hardware", "waiting_manual_hardware"}
-BLOCKED_STATUSES = {"technical_blocker", "blocked"}
+BLOCKED_STATUSES = {"technical_blocker", "technical_blocker_boot_only", "blocked"}
+AUTH_STATUSES = {"auth_refresh_required", "oauth_expired", "auth_expired", "repair_validation_inconclusive_auth", "inconclusive_auth_expired"}
 
 
 def _lower(value: Any) -> str:
@@ -316,7 +317,7 @@ def _is_record_not_ready(status: dict[str, Any]) -> bool:
 
 
 def _has_terminal_event(bundle: dict[str, Any]) -> bool:
-    return any(_lower(event.get("step")) in {"done", "failure", "technical_blocker", "manual_hardware_required"} for event in _events(bundle))
+    return any(_lower(event.get("step")) in {"done", "failure", "technical_blocker", "technical_blocker_boot_only", "manual_hardware_required"} for event in _events(bundle))
 
 
 def _latest_recovery_event(bundle: dict[str, Any]) -> dict[str, Any] | None:
@@ -351,9 +352,12 @@ def _verdict(bundle: dict[str, Any]) -> str:
         return _lower(eval_record.get("verdict"))
     gate = _gate(bundle)
     state = bundle.get("state") or {}
-    for value in (gate.get("status"), state.get("gate_status"), state.get("status"), (bundle.get("summary") or {}).get("status")):
+    repair = bundle.get("repair_outcome") or {}
+    for value in (gate.get("status"), state.get("gate_status"), state.get("status"), (bundle.get("summary") or {}).get("status"), repair.get("post_repair_validation"), repair.get("failure_type")):
         normalized = _lower(value)
         if normalized:
+            if normalized in AUTH_STATUSES:
+                return "auth_refresh_required"
             if normalized in SUCCESS_STATUSES:
                 return "full_inference_success" if normalized == "full_inference_success" else "success"
             if normalized in PARTIAL_STATUSES:
@@ -361,7 +365,7 @@ def _verdict(bundle: dict[str, Any]) -> str:
             if normalized in MANUAL_STATUSES:
                 return "manual_action_required"
             if normalized in BLOCKED_STATUSES:
-                return "technical_blocker"
+                return "technical_blocker_boot_only" if normalized == "technical_blocker_boot_only" else "technical_blocker"
             if normalized in FAILED_STATUSES:
                 return "failed"
             if normalized in CANCELLED_STATUSES:
@@ -375,8 +379,10 @@ def _final_visual_status(verdict: str) -> str:
         return "success"
     if verdict in {"partial_validation", "partial", "health_only", "completed_with_warnings"}:
         return "warn"
-    if verdict in {"failed", "failure", "technical_blocker"}:
+    if verdict in {"failed", "failure", "technical_blocker", "technical_blocker_boot_only"}:
         return "error"
+    if verdict == "auth_refresh_required":
+        return "warn"
     if verdict in {"manual_action_required", "cancelled"}:
         return "stopped"
     return "running"
@@ -387,8 +393,12 @@ def _status_label(verdict: str) -> tuple[str, str]:
         return "Full inference success", "Space boots, generation passed, and latency was measured."
     if verdict == "partial_validation":
         return "Completed with partial validation", "The run finished, but full generation was not verified."
+    if verdict == "technical_blocker_boot_only":
+        return "Technical blocker", "Health may pass, but no generation endpoint exists for full inference."
     if verdict == "technical_blocker":
         return "Technical blocker", "The run found a technical blocker."
+    if verdict == "auth_refresh_required":
+        return "Auth refresh required", "HF OAuth expired or is too close to expiry; sign in again before retrying validation."
     if verdict == "manual_action_required":
         return "Manual action required", "The run needs user action before validation can continue."
     if verdict == "failed":
@@ -598,7 +608,7 @@ def _phase_status(bundle: dict[str, Any], phase: str, verdict: str) -> str:
             return "complete"
         if verdict in {"partial_validation", "partial", "completed_with_warnings", "health_only"}:
             return "warning"
-        if verdict in {"failed", "technical_blocker"}:
+        if verdict in {"failed", "technical_blocker", "technical_blocker_boot_only"}:
             return "failed"
         if verdict in {"manual_action_required", "cancelled"}:
             return "stopped"

src/version.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
-ASF_APP_VERSION = "v190.32"
-ASF_RELEASE_NAME = "Agentic Space Factory v190.32"
+ASF_APP_VERSION = "v190.33"
+ASF_RELEASE_NAME = "Agentic Space Factory v190.33"
 
 
 def resolve_app_version(value: str | None = None) -> str:

src/view_models.py

@@ -13,7 +13,7 @@ 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"}
+TERMINAL_GLOBAL_STATUSES = {"succeeded", "partial", "failed", "cancelled", "blocked", "waiting_manual_action", "auth_refresh_required"}
 SUCCESS_RAW_STATUSES = {
     "success",
     "done",
@@ -37,8 +37,9 @@ MANUAL_RAW_STATUSES = {
 }
 FAILED_RAW_STATUSES = {"failed", "failure", "error", "repair_failed"}
 CANCELLED_RAW_STATUSES = {"cancelled", "canceled"}
-BLOCKED_RAW_STATUSES = {"technical_blocker", "blocked", "health_only"}
+BLOCKED_RAW_STATUSES = {"technical_blocker", "technical_blocker_boot_only", "blocked", "health_only"}
 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"}
 
 STEP_TO_PHASE = {step: step for step in STEP_ORDER} | {alias: canonical for alias, canonical in STEP_ALIASES.items()}
 
@@ -133,6 +134,8 @@ def _raw_statuses(bundle: dict[str, Any]) -> set[str]:
         _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
@@ -174,12 +177,18 @@ def normalize_run_status(
     manual = requires_manual_action(bundle)
     blocker = has_technical_blocker(bundle)
 
-    if statuses.intersection(CANCELLED_RAW_STATUSES):
+    if 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(SUCCESS_RAW_STATUSES):
         global_status = "succeeded"
         verdict = "passed"
@@ -189,9 +198,6 @@ def normalize_run_status(
     elif statuses.intersection(FAILED_RAW_STATUSES):
         global_status = "failed"
         verdict = "failed"
-    elif blocker:
-        global_status = "blocked"
-        verdict = "technical_blocker"
     elif statuses.intersection(RUNNING_RAW_STATUSES) or bundle.get("events") or launch:
         global_status = "running"
         verdict = "pending"
@@ -219,7 +225,7 @@ def normalize_run_status(
         "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"},
+        "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 "",
@@ -422,6 +428,40 @@ def _known_gradio_endpoint_info(bundle: dict[str, Any]) -> dict[str, Any]:
     return {"endpoint": endpoint or "/generate", "endpoint_count": endpoint_count, "endpoint_known": endpoint_known, "endpoints": endpoints[:20]}
 
 
+def _contract_declares_no_full_inference(bundle: dict[str, Any]) -> bool:
+    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.
 
@@ -467,6 +507,15 @@ def build_space_test_policy(bundle: dict[str, Any], status_model: dict[str, Any]
 
     if not target:
         return policy("unavailable", False, "no_target_space", "Space Test unavailable", "No generated Space is available to validate.")
+    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"}:
@@ -495,10 +544,19 @@ def build_space_test_model(bundle: dict[str, Any], status_model: dict[str, Any])
     smoke = bundle.get("generation_smoke") or {}
     summary = bundle.get("summary") or {}
     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"
     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": policy.get("endpoint") or smoke.get("api_name") or smoke.get("endpoint") or "/generate",
+        "endpoint": endpoint,
         "status": "passed" if smoke.get("ok") or smoke.get("status") == "success" else "pending",
         "latency_seconds": smoke.get("latency_seconds") or summary.get("latency_seconds"),
         "expected_output_type": summary.get("expected_output_type") or smoke.get("expected_output_type") or "",

src/worker_payload.py

@@ -10,6 +10,7 @@ def _encode(script: str) -> str:
 
 UNIVERSAL_MODEL_CARD_WORKER_SCRIPT = r'''
 
+import base64
 import hashlib
 import hmac
 import hashlib
@@ -48,6 +49,8 @@ INTERNAL_WORKSPACE_ARTIFACT_NAMES = {
     "REPAIR_SUMMARY.md",
     "PI_SUMMARY.md",
     "TECHNICAL_BLOCKERS.json",
+    "pi_feasibility_brief.json",
+    "pi_implementation_plan.json",
 }
 
 
@@ -278,7 +281,8 @@ def write_final_summary(run_dir: Path, final_state: dict, inference_gate: dict |
         "bucket_source": final_state.get("bucket_source") or os.environ.get("BUCKET_SOURCE", ""),
         "health_passed": bool(((gate.get("implementation_signals") or {}).get("health_passed") is True) or smoke.get("health_passed")),
         "generation_smoke_passed": bool(((gate.get("implementation_signals") or {}).get("generation_smoke_passed") is True) or smoke.get("status") == "success"),
-        "failure_type": smoke.get("failure_type") or gate.get("failure_type") or "",
+        "failure_type": smoke.get("failure_type") or gate.get("failure_type") or (final_state.get("details") or {}).get("failure_type") or (final_state.get("details") or {}).get("repair_failure_type") or (final_state.get("repair_outcome") or {}).get("failure_type") or final_state.get("failure_type") or "",
+        "repair_outcome_status": (final_state.get("repair_outcome") or {}).get("post_repair_validation") or (final_state.get("details") or {}).get("repair_outcome_status") or "",
         "updated_at": now(),
     }
     write_json(run_dir / "summary.json", payload)
@@ -308,7 +312,7 @@ def requirements_has_package(lines: list[str], package: str) -> bool:
         stripped = line.strip()
         if not stripped or stripped.startswith("#") or stripped.startswith("-") or "://" in stripped:
             continue
-        name = re.split(r"[<>=!~;\[]", stripped, 1)[0].strip().lower().replace("_", "-")
+        name = re.split(r"[<>=!~;\[]", stripped, maxsplit=1)[0].strip().lower().replace("_", "-")
         if name == wanted:
             return True
     return False
@@ -353,9 +357,13 @@ def write_artifact_manifest(run_dir: Path, *, events_path: Path | None = None, r
         _artifact_entry(run_dir, "live_status.json"),
         _artifact_entry(run_dir, "report.md"),
         _artifact_entry(run_dir, "model_analysis.json"),
+        _artifact_entry(run_dir, "planning/pi_feasibility_brief.json"),
+        _artifact_entry(run_dir, "planning/pi_implementation_plan.json"),
+        _artifact_entry(run_dir, "planning/worker_plan_review.json"),
         _artifact_entry(run_dir, "hardware_strategy.json"),
         _artifact_entry(run_dir, "hardware_attempts.json"),
         _artifact_entry(run_dir, "inference_gate.json"),
+        _artifact_entry(run_dir, "repair_outcome.json"),
         _artifact_entry(run_dir, "space_runtime.json"),
         _artifact_entry(run_dir, "tests/generation_smoke.json"),
         _artifact_entry(run_dir, "tests/api_schema.json"),
@@ -367,6 +375,7 @@ def write_artifact_manifest(run_dir: Path, *, events_path: Path | None = None, r
         _artifact_entry(run_dir, "repair/REPAIR_BRIEF.md"),
         _artifact_entry(run_dir, "repair/REPAIR_PLAN.md"),
         _artifact_entry(run_dir, "repair/REPAIR_SUMMARY.md"),
+        _artifact_entry(run_dir, "repair/REPAIR_OUTCOME.json"),
         _artifact_entry(run_dir, "repair/BLOCKAGE.json"),
         _artifact_entry(run_dir, "logs/pi_live_output.txt"),
         _artifact_entry(run_dir, "logs/pi_output.txt"),
@@ -429,6 +438,198 @@ def redact_text(text: str | None) -> str:
 
 
 
+
+
+class AuthRefreshRequired(RuntimeError):
+    """Raised when an OAuth/JWT token is expired or too close to expiry."""
+
+
+def _decode_jwt_claims_unverified(token: str) -> dict:
+    """Decode JWT claims without verification, only to inspect non-sensitive expiry metadata.
+
+    This never returns or writes the raw token. Opaque PAT-style tokens are supported by
+    returning an empty dict so they remain usable with an `unknown` expiry state.
+    """
+    try:
+        parts = (token or "").split(".")
+        if len(parts) < 2:
+            return {}
+        payload = parts[1]
+        payload += "=" * ((4 - len(payload) % 4) % 4)
+        raw = base64.urlsafe_b64decode(payload.encode("utf-8"))
+        data = json.loads(raw.decode("utf-8"))
+        return data if isinstance(data, dict) else {}
+    except Exception:
+        return {}
+
+
+def token_expiry_status(token: str, *, minimum_required_seconds: int = 0) -> dict:
+    issued_at = now()
+    if not token:
+        return {
+            "schema_version": "auth_status.v1",
+            "checked_at": issued_at,
+            "token_present": False,
+            "token_kind": "missing",
+            "expiry_known": False,
+            "status": "missing",
+            "safe_for_phase": False,
+            "minimum_required_seconds": minimum_required_seconds,
+        }
+    claims = _decode_jwt_claims_unverified(token)
+    exp = claims.get("exp") if isinstance(claims, dict) else None
+    iat = claims.get("iat") if isinstance(claims, dict) else None
+    token_kind = "oauth_jwt" if exp is not None else ("jwt_without_exp" if claims else "opaque_or_unknown")
+    payload = {
+        "schema_version": "auth_status.v1",
+        "checked_at": issued_at,
+        "token_present": True,
+        "token_kind": token_kind,
+        "expiry_known": exp is not None,
+        "minimum_required_seconds": int(minimum_required_seconds or 0),
+        "token_value": "[REDACTED]",
+    }
+    if iat is not None:
+        try:
+            payload["issued_at"] = datetime.fromtimestamp(int(iat), tz=timezone.utc).isoformat()
+        except Exception:
+            pass
+    if exp is None:
+        payload.update({"status": "unknown", "safe_for_phase": True, "auth_risk": "unknown_expiry"})
+        return payload
+    try:
+        exp_int = int(exp)
+        seconds_left = exp_int - int(time.time())
+        payload.update({
+            "expires_at": datetime.fromtimestamp(exp_int, tz=timezone.utc).isoformat(),
+            "seconds_until_expiry": seconds_left,
+        })
+    except Exception:
+        payload.update({"status": "unknown", "safe_for_phase": True, "auth_risk": "invalid_exp_claim"})
+        return payload
+    if seconds_left <= 0:
+        payload.update({"status": "expired", "safe_for_phase": False, "auth_risk": "expired"})
+    elif minimum_required_seconds and seconds_left < minimum_required_seconds:
+        payload.update({"status": "expires_soon", "safe_for_phase": False, "auth_risk": "expires_before_phase_budget"})
+    else:
+        payload.update({"status": "ok", "safe_for_phase": True, "auth_risk": "ok"})
+    return payload
+
+
+def write_auth_probe(run_dir: Path, events_path: Path | None, phase: str, token: str, *, minimum_required_seconds: int = 0, raise_on_unsafe: bool = False) -> dict:
+    payload = token_expiry_status(token, minimum_required_seconds=minimum_required_seconds)
+    payload["phase"] = phase
+    safe_payload = {k: v for k, v in payload.items() if k != "token_value"}
+    try:
+        probes_dir = run_dir / "auth_probes"
+        probes_dir.mkdir(parents=True, exist_ok=True)
+        write_json(probes_dir / f"{phase}.json", safe_payload)
+        write_json(run_dir / "auth_status.json", safe_payload)
+    except Exception:
+        pass
+    status = str(payload.get("status") or "unknown")
+    if events_path:
+        event_status = "success" if payload.get("safe_for_phase") else "failed"
+        if status == "unknown":
+            event_status = "warning"
+        append_event(
+            events_path,
+            "auth_probe",
+            event_status,
+            f"HF OAuth/token expiry check for {phase}: {status}",
+            safe_payload,
+        )
+    if raise_on_unsafe and not payload.get("safe_for_phase"):
+        raise AuthRefreshRequired(f"HF auth token is {status} before {phase}; refresh sign-in before continuing.")
+    return payload
+
+
+def is_auth_expired_error(error: Exception | str) -> bool:
+    text = str(error or "").lower()
+    return any(marker in text for marker in [
+        "oauth token has expired",
+        "exp claim timestamp check failed",
+        "token has expired",
+        "jwt expired",
+    ])
+
+
+
+def classify_repair_validation_error(error: Exception | str) -> dict:
+    """Classify post-repair validation failures without losing the repair result.
+
+    A repair can be correctly diagnosed, patched, and uploaded while the final
+    validation is inconclusive because OAuth expired. Keep that distinct from a
+    model/runtime repair failure so the UI and audit trail do not imply that Pi's
+    patch necessarily failed.
+    """
+    text = str(error or "")
+    if isinstance(error, AuthRefreshRequired) or is_auth_expired_error(text):
+        return {
+            "post_repair_validation": "inconclusive_auth_expired",
+            "failure_type": "repair_validation_inconclusive_auth",
+            "terminal_status": "auth_refresh_required",
+            "message": "Repair progress could not be validated because HF auth expired or is too close to expiry.",
+        }
+    return {
+        "post_repair_validation": "failed",
+        "failure_type": "repair_validation_failed",
+        "terminal_status": "failed",
+        "message": "Repair was applied, but the repaired Space did not pass live validation.",
+    }
+
+
+def write_repair_outcome(run_dir: Path, events_path: Path | None = None, **updates) -> dict:
+    """Write a cumulative repair outcome artifact.
+
+    Keep it at the run root for list-view consumers and mirror it inside
+    repair/ for manual audits. The payload is intentionally compact and
+    redacted; full logs remain in logs/ and repair/ artifacts.
+    """
+    root_path = run_dir / "repair_outcome.json"
+    existing = load_json_if_exists(root_path) if root_path.exists() else {}
+    if not isinstance(existing, dict):
+        existing = {}
+    payload = {
+        "schema_version": "repair_outcome.v1",
+        **existing,
+        **{k: v for k, v in updates.items() if v is not None},
+        "updated_at": now(),
+    }
+    write_json(root_path, payload)
+    try:
+        write_json(run_dir / "repair" / "REPAIR_OUTCOME.json", payload)
+    except Exception:
+        pass
+    if events_path:
+        append_event(
+            events_path,
+            "repair_outcome",
+            str(payload.get("post_repair_validation") or payload.get("repair_status") or payload.get("repair_decision") or "updated"),
+            "Repair outcome artifact updated",
+            payload,
+        )
+    return payload
+
+def minimum_auth_seconds_for_phase(phase: str) -> int:
+    env_key = "ASF_AUTH_MIN_SECONDS_" + re.sub(r"[^A-Z0-9]+", "_", phase.upper()).strip("_")
+    raw = os.environ.get(env_key) or os.environ.get("ASF_AUTH_MIN_SECONDS", "")
+    if raw:
+        try:
+            return max(0, int(raw))
+        except Exception:
+            pass
+    defaults = {
+        "before_space_create": 900,
+        "before_upload": 900,
+        "before_initial_validation": 1800,
+        "before_repair": 1800,
+        "before_repair_upload": 1800,
+        "before_repair_validation": 1800,
+        "before_linked_space_test": 1800,
+    }
+    return defaults.get(phase, 0)
+
 def ensure_hf_token_context(run_dir: Path, events_path: Path | None = None) -> dict:
     """Ensure Pi and HF tooling see the same private HF token safely.
 
@@ -442,6 +643,7 @@ def ensure_hf_token_context(run_dir: Path, events_path: Path | None = None) -> d
         os.environ.setdefault("HF_TOKEN", token)
         os.environ.setdefault("HUGGING_FACE_HUB_TOKEN", token)
         os.environ.setdefault("HF_HUB_ENABLE_HF_TRANSFER", "1")
+    expiry = token_expiry_status(token, minimum_required_seconds=minimum_auth_seconds_for_phase("before_initial_validation"))
     payload = {
         "hf_token_present": bool(token),
         "hf_token_length": len(token) if token else 0,
@@ -450,6 +652,13 @@ def ensure_hf_token_context(run_dir: Path, events_path: Path | None = None) -> d
         "bucket_source": os.environ.get("BUCKET_SOURCE") or "",
         "target_space_id": os.environ.get("TARGET_SPACE_ID") or "",
         "token_value": "[REDACTED]" if token else "",
+        "token_kind": expiry.get("token_kind"),
+        "expiry_known": expiry.get("expiry_known"),
+        "expires_at": expiry.get("expires_at"),
+        "seconds_until_expiry_at_job_start": expiry.get("seconds_until_expiry"),
+        "auth_risk": expiry.get("auth_risk"),
+        "safe_for_long_build": bool(expiry.get("safe_for_phase")),
+        "minimum_required_seconds": expiry.get("minimum_required_seconds"),
     }
     run_dir.mkdir(parents=True, exist_ok=True)
     write_json(run_dir / "token_context.json", payload)
@@ -734,7 +943,7 @@ def eval_verdict(status: str, phase: str, health_passed: bool, smoke_passed: boo
         return "cancelled"
     if status == "manual_hardware_required" or inference_gate.get("manual_hardware_required"):
         return "manual_action_required"
-    if status == "technical_blocker":
+    if status in {"technical_blocker", "technical_blocker_boot_only"}:
         return "technical_blocker"
     if full_inference_verified:
         return "success"
@@ -951,6 +1160,10 @@ def fail(run_dir: Path, events_path: Path, message: str, details: dict | None =
     existing_state = load_json_if_exists(run_dir / "state.json") if (run_dir / "state.json").exists() else {}
     if not isinstance(existing_state, dict):
         existing_state = {}
+    repair_outcome = load_json_if_exists(run_dir / "repair_outcome.json") if (run_dir / "repair_outcome.json").exists() else {}
+    if isinstance(repair_outcome, dict) and repair_outcome:
+        safe.setdefault("repair_outcome_status", repair_outcome.get("post_repair_validation") or repair_outcome.get("repair_status") or "")
+        safe.setdefault("repair_failure_type", repair_outcome.get("failure_type") or "")
     failure_state = {
         **existing_state,
         "run_id": os.environ.get("RUN_ID"),
@@ -959,6 +1172,7 @@ def fail(run_dir: Path, events_path: Path, message: str, details: dict | None =
         "message": message,
         "updated_at": now(),
         "details": safe,
+        "repair_outcome": repair_outcome if isinstance(repair_outcome, dict) else {},
     }
     # Preserve the target Space when the worker fails after repository creation.
     target_space = failure_state.get("target_space") or os.environ.get("TARGET_SPACE_ID") or ""
@@ -1594,6 +1808,85 @@ def read_inference_contract(workspace: Path | None) -> dict:
         return {}
 
 
+def _as_bool_false(value) -> bool:
+    if value is False:
+        return True
+    if isinstance(value, str):
+        return value.strip().lower() in {"false", "no", "0", "off"}
+    return False
+
+
+def _read_workspace_json(workspace: Path | None, filename: str) -> dict:
+    if not workspace:
+        return {}
+    path = workspace / filename
+    if not path.exists():
+        return {}
+    try:
+        data = json.loads(path.read_text(encoding="utf-8", errors="replace"))
+        return data if isinstance(data, dict) else {}
+    except Exception:
+        return {}
+
+
+def contract_declares_no_full_inference(workspace: Path | None) -> dict:
+    """Return a structured reason when Pi explicitly declared boot-only/no inference.
+
+    v191.2: A machine-readable no-inference contract must be respected by the
+    worker. In that case automatic generation smoke is skipped instead of
+    inventing /generate fallback arguments that cannot exist.
+    """
+    contract = read_inference_contract(workspace)
+    blockers = _read_workspace_json(workspace, "TECHNICAL_BLOCKERS.json")
+    validation_level = str(contract.get("validation_level") or "").strip().lower().replace("_", "-")
+    primary = normalize_api_name(str(contract.get("primary_api_name") or "")) if contract else ""
+    contract_full_false = _as_bool_false(contract.get("full_inference_implemented"))
+    blocker_full_false = _as_bool_false(blockers.get("full_inference_implemented"))
+    validation_boot_only = validation_level in {"boot-only", "health-only", "diagnostic-only", "info-only"}
+    primary_absent = not primary or primary == "/health"
+    blockers_count = blockers.get("blockers_count")
+    if not isinstance(blockers_count, int):
+        raw_blockers = blockers.get("blockers")
+        blockers_count = len(raw_blockers) if isinstance(raw_blockers, list) else 0
+
+    declared = bool(contract_full_false or blocker_full_false or (validation_boot_only and primary_absent))
+    return {
+        "declared": declared,
+        "source": "inference_contract" if contract_full_false or validation_boot_only else "technical_blockers" if blocker_full_false else "",
+        "full_inference_implemented": False if (contract_full_false or blocker_full_false) else contract.get("full_inference_implemented"),
+        "validation_level": contract.get("validation_level"),
+        "primary_api_name": contract.get("primary_api_name"),
+        "blockers_count": blockers_count,
+        "contract_present": bool(contract),
+        "technical_blockers_present": bool(blockers),
+    }
+
+
+def write_contract_skipped_generation_smoke(run_dir: Path, events_path: Path, expected_output_type: str, target_space_id: str, reason: dict | None = None) -> dict:
+    reason = reason or {}
+    payload = {
+        "status": "skipped",
+        "skip_reason": "contract_declared_no_full_inference",
+        "reason": "Pi declared that full inference is not implemented; automatic generation smoke was skipped.",
+        "target_space": target_space_id,
+        "api_name": None,
+        "expected_output_type": expected_output_type,
+        "validation_mode": "automatic_smoke",
+        "payload_source": "contract_declared_no_full_inference",
+        "contract_aware_skip": True,
+        "contract_reason": reason,
+        "next_action": "Full inference is blocked. No generation endpoint exists. Review TECHNICAL_BLOCKERS.json / PI_SUMMARY.md or provide a dedicated implementation and hardware plan.",
+        **measured_zero_gpu_recommendation(None),
+    }
+    write_json(run_dir / "tests" / "generation_smoke.json", payload)
+    write_json(run_dir / "tests" / "payload_source.json", {"schema_version": "1.0", "validation_engine": "unified_gradio_validation_harness", "validation_mode": "automatic_smoke", "payload_source": "contract_declared_no_full_inference", "selected_api_name": None, "parent_smoke_payload_used": False})
+    write_json(run_dir / "tests" / "validation_engine.json", {"schema_version": "1.0", "validation_engine": "unified_gradio_validation_harness", "validation_mode": "automatic_smoke", "resolved_request_required_before_predict": False, "skipped_by_contract": True})
+    write_json(run_dir / "tests" / "resolved_validation_request.json", {"api_name": None, "test_args": [], "test_kwargs": {}, "expected_output_type": expected_output_type, "validation_mode": "automatic_smoke", "payload_source": "contract_declared_no_full_inference", "resolved_request_required_before_predict": False})
+    write_live_status(run_dir, stage="generation_smoke", status="skipped", message="Generation smoke skipped because Pi declared full inference unavailable", data=payload)
+    append_event(events_path, "generation_smoke", "skipped", "Generation smoke skipped because Pi declared no full inference endpoint", payload)
+    return payload
+
+
 def contract_smoke_test_payload(contract: dict, api_name: str, endpoint, expected_output_type: str) -> dict | None:
     smoke = contract.get("smoke_test") if isinstance(contract, dict) else None
     if not isinstance(smoke, dict):
@@ -2583,6 +2876,100 @@ def create_space_with_hardware_strategy(api, target_space_id: str, token: str, p
             raise
 
 
+
+
+def _safe_json_object_from_workspace(path: Path) -> dict:
+    payload = read_json(path, {})
+    return payload if isinstance(payload, dict) else {}
+
+
+def _first_truthy_string(*values) -> str | None:
+    for value in values:
+        if isinstance(value, str) and value.strip():
+            return value.strip()
+    return None
+
+
+def write_pi_planning_review(workspace: Path, run_dir: Path, events_path: Path, model_analysis: dict | None = None, implementation_mode: str = "") -> dict:
+    """Persist Pi planning artifacts and a non-blocking worker review.
+
+    v191.9 intentionally makes planning visibility-only. Pi is asked to produce
+    a feasibility brief and implementation plan, but older/partial Pi behavior
+    must not fail a run that would otherwise work. The worker records whether
+    those artifacts exist, mirrors them under planning/, and keeps the final
+    authority in the existing deterministic gates.
+    """
+    model_analysis = model_analysis or {}
+    planning_dir = run_dir / "planning"
+    planning_dir.mkdir(parents=True, exist_ok=True)
+    brief_path = workspace / "pi_feasibility_brief.json"
+    plan_path = workspace / "pi_implementation_plan.json"
+    brief = _safe_json_object_from_workspace(brief_path)
+    plan = _safe_json_object_from_workspace(plan_path)
+    brief_present = bool(brief)
+    plan_present = bool(plan)
+    if brief_present:
+        write_json(planning_dir / "pi_feasibility_brief.json", brief)
+    else:
+        write_json(planning_dir / "pi_feasibility_brief.json", {
+            "schema_version": "pi_feasibility_brief.v1",
+            "status": "missing",
+            "non_blocking": True,
+            "message": "Pi did not produce pi_feasibility_brief.json; continuing with existing deterministic worker gates.",
+        })
+    if plan_present:
+        write_json(planning_dir / "pi_implementation_plan.json", plan)
+    else:
+        write_json(planning_dir / "pi_implementation_plan.json", {
+            "schema_version": "pi_implementation_plan.v1",
+            "status": "missing",
+            "non_blocking": True,
+            "message": "Pi did not produce pi_implementation_plan.json; continuing with existing deterministic worker gates.",
+        })
+
+    declared_strategy = _first_truthy_string(
+        brief.get("recommended_strategy"),
+        brief.get("strategy"),
+        plan.get("strategy"),
+        plan.get("implementation_strategy"),
+    ) or "unspecified"
+    should_attempt_full = brief.get("should_attempt_full_inference")
+    if should_attempt_full is None:
+        should_attempt_full = plan.get("should_attempt_full_inference")
+    worker_recommendation = "continue_existing_flow"
+    warnings: list[str] = []
+    if not brief_present:
+        warnings.append("missing_pi_feasibility_brief")
+    if not plan_present:
+        warnings.append("missing_pi_implementation_plan")
+    if declared_strategy in {"boot_only_blocker", "boot_only_blocker_or_manual_refactor", "manual_hardware_required"} or should_attempt_full is False:
+        worker_recommendation = "respect_plan_if_contract_declares_blocker"
+    elif declared_strategy in {"full_inference", "proceed_full_inference", "attempt_full_inference"} or should_attempt_full is True:
+        worker_recommendation = "proceed_full_inference_with_existing_gates"
+
+    review = {
+        "schema_version": "worker_plan_review.v1",
+        "status": "ok" if brief_present and plan_present else "partial",
+        "non_blocking": True,
+        "implementation_mode": implementation_mode,
+        "model_id": model_analysis.get("model_id"),
+        "pipeline_tag": model_analysis.get("pipeline_tag"),
+        "library_name": model_analysis.get("library_name"),
+        "planning_artifacts_present": {
+            "pi_feasibility_brief": brief_present,
+            "pi_implementation_plan": plan_present,
+        },
+        "declared_strategy": declared_strategy,
+        "should_attempt_full_inference": should_attempt_full,
+        "worker_recommendation": worker_recommendation,
+        "warnings": warnings,
+        "authority_note": "Pi planning is advisory in v191.9; deterministic worker gates, INFERENCE_CONTRACT.json, TECHNICAL_BLOCKERS.json, and live validation remain authoritative.",
+        "created_at": now(),
+    }
+    write_json(planning_dir / "worker_plan_review.json", review)
+    append_event(events_path, "pi_planning_review", review["status"], "Pi feasibility and implementation planning reviewed", review)
+    return review
+
 def create_initial_workspace(workspace: Path, model_id: str, target_space_id: str, preferred_hardware: str, fallback_hardware: str, allow_fallback: bool, implementation_mode: str, model_analysis: dict | None = None):
     workspace.mkdir(parents=True, exist_ok=True)
     model_analysis = model_analysis or {}
@@ -2630,7 +3017,6 @@ diffusers
 accelerate
 safetensors
 torch
-kernels
 pillow
 numpy
 requests
@@ -2677,9 +3063,21 @@ Non-negotiable safety and product constraints:
 - Work only inside the current workspace.
 - The wrapper will create the private Space, request allowed hardware best-effort, upload files, and validate the live app. Do not create/delete repos yourself in this builder worker.
 - Preserve a cheap health endpoint named `health` with `api_name="health"`. It must not load weights, run GPU work, or download large files.
-- Do not pin huggingface_hub below 1.0. Use huggingface_hub>=0.34.0,<2.0.0 unless the model card requires a narrower compatible range. If transformers>=5 is used, keep huggingface_hub compatible with it, for example huggingface_hub>=1.5.0,<2.0.0.
+- Platform dependency policy: Gradio, Gradio Client, Hugging Face Hub, Spaces, and hf_xet are owned by Agentic Space Factory / the Hugging Face Spaces runtime. Do not pin or downgrade them. Never write `gradio==...`, `gradio<=...`, `gradio~=...`, `gradio-client==...`, `gradio_client==...`, or stale exact pins for `huggingface_hub`, `spaces`, or `hf_xet`.
+- If generated code needs an older Gradio API, update the code to modern Gradio instead of downgrading Gradio.
+- Follow the gist dependency method for model-specific packages: pin only when useful to reduce pip resolver backtracking or when the model card/build logs explicitly require a version; do not cargo-cult every version from examples.
+- Do not pin torch or torchaudio unless unavoidable and explicitly justified; Spaces provide the PyTorch stack. If torchvision is needed, prefer leaving it unpinned so it can resolve against the managed torch runtime.
+- If model code is not pip-installable, vendor the necessary code into the Space repo instead of referencing local paths, editable installs, or clone-time side effects in requirements.txt.
+- Do not run local venv installs, broad `pip install --dry-run`, or dependency checks as proof of compatibility; only the live Space build/logs are authoritative. Use `python -m py_compile app.py` only as a cheap syntax check.
+- Use huggingface_hub>=0.34.0,<2.0.0 unless a newer Transformers path requires a compatible newer range. The worker will normalize platform-owned dependency lines before upload.
 - README.md frontmatter must remain valid; if it uses short_description, it must be 60 characters or fewer.
 
+Planning contract:
+- Before committing to a full implementation strategy, produce `pi_feasibility_brief.json` and `pi_implementation_plan.json`. These files are advisory artifacts for the worker and must not contain secrets.
+- `pi_feasibility_brief.json` should include: complexity, expected_runtime, zerogpu_feasible, requires_custom_code, multi_gpu_risk, dependency_risk, recommended_strategy, minimum_hardware, should_attempt_full_inference, and evidence.
+- `pi_implementation_plan.json` should include: selected_strategy, intended_files, endpoints, dependency_strategy, hardware_strategy, smoke_test_plan, fallback_plan, and blocker_policy.
+- If the model appears high-risk or not portable, say so in the plan rather than silently attempting a fragile fake success. The worker will still use INFERENCE_CONTRACT.json, TECHNICAL_BLOCKERS.json, and live validation as the final authority.
+
 Implementation contract:
 - If IMPLEMENTATION_MODE is `full-inference-gated`, you are not allowed to silently replace generation with a placeholder and call it success.
 - Try to implement the closest real inference path for the model card using evidence from README, model metadata, config files, and repo files.
@@ -2687,10 +3085,13 @@ Implementation contract:
 - If the model is standard and feasible, implement a real generate/predict function and expose it as a Gradio endpoint.
 - If the model requires GPU, add ZeroGPU-compatible `@spaces.GPU(...)` only around the inference function. Do not decorate health. Do not assume dedicated A100/H200 hardware is available automatically; if such hardware is needed, document it as a manual requirement.
 - If the model requires special dependencies, include them only when needed and document risks.
-- Investigate compatibility fallbacks before declaring a blocker: PyTorch SDPA, xformers, HF Kernels where relevant, CPU/offload/lazy loading, smaller resolution/steps, safe smoke-test inputs.
+- Investigate compatibility fallbacks before declaring a blocker: PyTorch SDPA, xformers, HF Kernels/Kernel Hub where relevant, Transformers AttentionInterface, Diffusers attention processors, CPU/offload/lazy loading, smaller resolution/steps, safe smoke-test inputs.
+- Native kernel policy: if the model card or logs mention flash-attn, custom attention kernels, xformers, Triton kernels, fused ops, or CUDA/C++ extensions, do not blindly add source-built native packages to requirements.txt. Prefer runtime-compatible drop-in backends or compatible wheels first. Use `kernels` only when it directly matches the needed operation and the target hardware/runtime is plausible; document the selected backend in PI_SUMMARY.md and INFERENCE_CONTRACT.json.
 - If real inference is impossible or unsafe in a Space, write TECHNICAL_BLOCKERS.json with concrete evidence for every blocker.
 
 Deliverables:
+- pi_feasibility_brief.json must summarize feasibility and risk before/alongside implementation.
+- pi_implementation_plan.json must summarize the selected implementation plan, endpoints, dependencies, and validation approach.
 - app.py must boot on Hugging Face Spaces.
 - app.py must expose health/api_name="health".
 - If real generation is implemented, generate/predict must attempt a real model call, not only return a textual diagnostic.
@@ -2778,13 +3179,13 @@ def sanitize_readme_metadata(workspace: Path, events_path: Path):
         )
 
 def normalize_requirements_for_modern_hub(workspace: Path, events_path: Path):
-    """Normalize only broad known-dangerous base dependencies before upload.
+    """Normalize broad known-dangerous base and platform dependencies before upload.
 
-    Do not try to solve every dependency conflict here. Pi is responsible for
-    reading concrete build logs and patching requirements when a Space build
-    fails. The Factory only prevents obviously unsafe broad ranges that would
-    make the initial build non-deterministic, then observes the Space status and
-    hands real build errors back to Pi with evidence.
+    Do not try to solve every dependency conflict here. Do not try to solve every model-specific dependency conflict here. Pi is
+    responsible for reading concrete build logs and patching requirements when
+    a Space build fails. The Factory only prevents obviously unsafe broad ranges
+    and owns the Gradio/HF platform stack so old generated pins cannot conflict
+    with the current Spaces runtime; concrete pip conflicts remain Pi repair work.
     """
     req_path = workspace / "requirements.txt"
     if not req_path.exists():
@@ -2805,27 +3206,62 @@ def normalize_requirements_for_modern_hub(workspace: Path, events_path: Path):
 
     # Minimal base pins only. Do not globally pin diffusers here: newer model
     # cards may legitimately need a recent Diffusers release. If diffusers causes
-    # a pip conflict, Pi must repair from the concrete build log.
+    # a pip conflict, Pi is responsible for concrete build-log repairs. Pi must repair from the concrete build log.
     policy: dict[str, str] = {
         "huggingface-hub": "huggingface_hub>=0.34.0,<2.0.0",
         "transformers": "transformers>=4.51.0,<5.0.0",
     }
+    # Platform-owned dependencies are normalized more aggressively than model
+    # dependencies. Gradio/HF runtime packages must stay current with Spaces and
+    # Agentic Space Factory validation; Pi should patch app.py for modern Gradio
+    # rather than pinning or downgrading this stack.
+    platform_policy: dict[str, str] = {
+        "gradio": "gradio",
+        "gradio-client": "gradio-client",
+        "huggingface-hub": policy["huggingface-hub"],
+        "spaces": "spaces>=0.30",
+        "hf-xet": "hf_xet",
+    }
     aliases = {
         "huggingface_hub": "huggingface-hub",
         "huggingface-hub": "huggingface-hub",
         "transformers": "transformers",
+        "gradio": "gradio",
+        "gradio-client": "gradio-client",
+        "gradio_client": "gradio-client",
+        "spaces": "spaces",
+        "hf-xet": "hf-xet",
+        "hf_xet": "hf-xet",
     }
+    platform_owned = set(platform_policy)
     seen_policy: set[str] = set()
+    seen_platform: set[str] = set()
     filtered: list[str] = []
     changed = False
+    removed_platform_pins: list[str] = []
+    normalized_platform_lines: list[dict] = []
 
     for line in package_lines:
         stripped = line.strip()
         if stripped.startswith("#") or "://" in stripped or stripped.startswith((".", "/")):
             filtered.append(line)
             continue
-        name = re.split(r"[<>=!~;\[]", stripped, 1)[0].strip().lower().replace("_", "-")
+        name = re.split(r"[<>=!~;\[]", stripped, maxsplit=1)[0].strip().lower().replace("_", "-")
         canonical = aliases.get(name)
+        if canonical in platform_owned:
+            normalized = platform_policy[canonical]
+            if stripped != normalized:
+                changed = True
+                removed_platform_pins.append(stripped)
+                normalized_platform_lines.append({"from": stripped, "to": normalized, "package": canonical})
+            if canonical not in seen_platform:
+                filtered.append(normalized)
+                seen_platform.add(canonical)
+                if canonical in policy:
+                    seen_policy.add(canonical)
+            else:
+                changed = True
+            continue
         if canonical in policy:
             if stripped != policy[canonical]:
                 changed = True
@@ -2843,34 +3279,53 @@ def normalize_requirements_for_modern_hub(workspace: Path, events_path: Path):
             stable_policy_lines.append(policy[canonical])
             changed = True
 
+    injected_platform_lines: list[str] = []
+    app_text = (workspace / "app.py").read_text(encoding="utf-8", errors="ignore").lower() if (workspace / "app.py").exists() else ""
+    if ("import gradio" in app_text or "from gradio" in app_text or "gr." in app_text) and "gradio" not in seen_platform:
+        injected_platform_lines.append(platform_policy["gradio"])
+        seen_platform.add("gradio")
+        changed = True
+    if ("gradio_client" in app_text or "from gradio_client" in app_text) and "gradio-client" not in seen_platform:
+        injected_platform_lines.append(platform_policy["gradio-client"])
+        seen_platform.add("gradio-client")
+        changed = True
+
     torch_added = False
-    if workspace_app_imports_torch(workspace) and not requirements_has_package(filtered + stable_policy_lines, "torch"):
+    if workspace_app_imports_torch(workspace) and not requirements_has_package(filtered + stable_policy_lines + injected_platform_lines, "torch"):
         stable_policy_lines.append("torch>=2.0.0")
         torch_added = True
         changed = True
 
-    new_lines = prefix_lines + stable_policy_lines + filtered
+    new_lines = prefix_lines + stable_policy_lines + injected_platform_lines + filtered
     new = "\n".join(line for line in new_lines if line.strip()) + "\n"
     if new != raw:
         changed = True
     if changed:
         req_path.write_text(new, encoding="utf-8")
+        policy_payload = {
+            "schema_version": "1.0",
+            "status": "normalized",
+            "platform_owned_dependencies": ["gradio", "gradio-client", "huggingface_hub", "spaces", "hf_xet"],
+            "removed_pins": removed_platform_pins,
+            "normalized_platform_lines": normalized_platform_lines,
+            "injected_platform_lines": injected_platform_lines,
+            "base_policy": {"huggingface_hub": policy["huggingface-hub"], "transformers": policy["transformers"]},
+            "base_policy_reason": "Avoid uncontrolled Transformers 5.x while preserving model-specific dependency choices for Pi to repair from build logs; require torch when app.py imports torch.",
+            "torch_added": torch_added,
+            "torch_policy": "torch>=2.0.0",
+            "torch_reason": "app_imports_torch" if torch_added else "not_needed_or_already_present",
+            "model_dependency_policy": "Model-specific pins are preserved unless they hit a known deterministic guardrail; concrete conflicts remain Pi repair work from build logs.",
+            "reason": "Gradio/HF runtime dependencies are owned by Agentic Space Factory / Spaces runtime and must not be downgraded by generated requirements.",
+        }
+        write_json(workspace / "requirements_policy.json", policy_payload)
         append_event(
             events_path,
             "requirements_sanitize",
             "success",
-            "Normalized broad base dependencies; concrete pip conflicts remain Pi repair work",
-            {
-                "huggingface_hub": policy["huggingface-hub"],
-                "transformers": policy["transformers"],
-                "reason": "Avoid uncontrolled Transformers 5.x while preserving model-specific dependency choices for Pi to repair from build logs; require torch when app.py imports torch.",
-                "torch_added": torch_added,
-                "torch_policy": "torch>=2.0.0",
-                "torch_reason": "app_imports_torch" if torch_added else "not_needed_or_already_present",
-            },
+            "Normalized platform-owned Gradio/HF dependencies and broad base dependencies; concrete pip conflicts remain Pi repair work",
+            policy_payload,
         )
 
-
 def useful_log_signals(text: str) -> list[str]:
     low = (text or "").lower()
     signals = []
@@ -3354,6 +3809,7 @@ First read `INCIDENT_BRIEF.md` and the HF Spaces gist operational rules: {GIST_U
 
 You are not allowed to edit code in this diagnosis step. Your task is to decide the next action for the Factory.
 Use the gist method: read logs first, identify the first actionable error, use the cheapest useful iteration rung, and require a live Gradio/API validation before success.
+When the first error is dependency-related, classify it with the ASF dependency contract in mind: Gradio/HF runtime packages are platform-owned, model-specific pins need evidence, torch/torchaudio are managed by Spaces, and non-pip-installable model code should be vendored.
 
 Write `REPAIR_DECISION.json` exactly as requested in INCIDENT_BRIEF.md. Do not patch files during this step.
 """
@@ -3471,7 +3927,11 @@ A separate Pi diagnosis already decided that `patch_code` is justified. Do not r
 - Preserve or restore a cheap `health` endpoint.
 - Preserve the expected Gradio API endpoint when possible.
 - Keep README metadata valid and short_description <= 60 chars.
-- Keep Hugging Face Hub requirements modern; do not pin below 1.0 unless unavoidable and explained.
+- Platform dependency policy: keep Gradio / Gradio Client / Hugging Face Hub / Spaces modern and worker-owned. Do not downgrade Gradio or add stale exact pins. If the failure involves a Gradio dependency conflict, remove the stale Gradio pin and patch app.py for modern Gradio instead of downgrading Gradio.
+- For model-specific dependencies, follow the gist method: patch the first concrete resolver error minimally, preserve justified model-card pins, and do not pin torch/torchaudio unless unavoidable and explicitly justified. If torchvision is needed, prefer leaving it unpinned so it resolves against the managed torch runtime.
+- If model code is not pip-installable, vendor the necessary code into the Space repo instead of relying on local paths, editable installs, or clone-time side effects in requirements.txt.
+- Do not prove dependency fixes with local venv installs or broad pip dry-runs; Space build logs are authoritative. Keep local checks cheap, such as `python -m py_compile app.py`.
+- Native kernel policy: for flash-attn/custom attention/xformers/Triton/fused CUDA errors, do not blindly add source-built packages. First consider PyTorch SDPA, compatible prebuilt wheels, HF Kernels/Kernel Hub, Transformers AttentionInterface, or Diffusers attention processors. If no compatible fallback exists, declare a blocker rather than forcing a fragile build.
 
 ## Required repair artifacts
 Before modifying files, write `REPAIR_PLAN.md` with:
@@ -3552,6 +4012,8 @@ First read `REPAIR_BRIEF.md`, `INCIDENT_BRIEF.md`, `DEPENDENCY_ERROR_BRIEF.md` i
 You are continuing the same build run, not starting a separate project.
 This patch is allowed only because the diagnosis decision selected `patch_code`.
 If `DEPENDENCY_ERROR_BRIEF.md` exists, treat it as evidence for the gist method: identify the first pip error, patch dependency pins minimally, and do not modify inference code unless the dependency fix alone cannot address that first error.
+Follow the ASF/gist dependency contract during repair: Gradio/HF runtime packages are platform-owned, model-specific pins must be evidence-backed, torch/torchaudio should remain managed by Spaces, and non-pip-installable model code should be vendored instead of referenced through fragile local paths.
+For native kernel failures such as flash-attn/custom attention/xformers/Triton/fused CUDA ops, do not blindly add source-built native packages. Prefer PyTorch SDPA, compatible wheels, HF Kernels/Kernel Hub, Transformers AttentionInterface, or Diffusers attention processors when they match the required operation. Use `kernels` only when directly justified and document the backend in REPAIR_SUMMARY.md / INFERENCE_CONTRACT.json.
 Use the available HF token context to inspect private Hub resources when needed, but never print or persist token values.
 
 Critical method:
@@ -3627,18 +4089,53 @@ def recover_after_live_validation_failure(api, workspace: Path, run_dir: Path, e
     dependency_issue = extract_pip_dependency_issue(f"{current_error}\n{build_log}\n{runtime_log}")
     if dependency_issue:
         write_dependency_error_brief(workspace, run_dir, events_path, dependency_issue, build_log, runtime_log, current_error)
+        write_repair_outcome(
+            run_dir,
+            events_path,
+            repair_trigger="dependency_resolution_error",
+            root_cause="dependency_resolution_error",
+            initial_error=str(current_error)[:4000],
+            patch_applied=False,
+            upload_success=False,
+            post_repair_validation="not_started",
+        )
     if apply_dependency_guardrail_repair(workspace, run_dir, events_path, current_error, build_log, runtime_log):
+        write_repair_outcome(
+            run_dir,
+            events_path,
+            repair_trigger="dependency_resolution_error",
+            repair_decision="deterministic_dependency_guardrail",
+            patch_applied=True,
+            upload_success=False,
+            post_repair_validation="not_started",
+        )
         append_event(events_path, "factory_rebuild", "started", "Re-uploading dependency-guardrailed workspace after pip resolver build error")
         if not safe_same_code_reupload(api, workspace, target_space_id, token, run_dir, events_path, reason="dependency_guardrail_repair"):
+            write_repair_outcome(run_dir, events_path, upload_success=False, post_repair_validation="not_started", failure_type="repair_upload_blocked_by_guardrail")
             raise RuntimeError("Dependency guardrail rebuild skipped by restart guardrails")
+        write_repair_outcome(run_dir, events_path, upload_success=True, post_repair_validation="pending")
         append_event(events_path, "factory_rebuild", "success", "Dependency-guardrailed workspace uploaded; revalidating live Space")
         append_event(events_path, "repair_validation", "started", "Revalidating after deterministic dependency repair")
         write_live_status(run_dir, stage="live_validation", status="running", message="Waiting for Space runtime and health endpoint", data={"target_space": target_space_id})
         try:
+            write_auth_probe(run_dir, events_path, "before_initial_validation", token, minimum_required_seconds=minimum_auth_seconds_for_phase("before_initial_validation"), raise_on_unsafe=True)
             validation = validate_live_api(api, target_space_id, token, run_dir, events_path, timeout_s=1200)
+            write_repair_outcome(run_dir, events_path, post_repair_validation="success", failure_type="", final_user_message="Deterministic dependency repair resolved the build blockage.")
             append_event(events_path, "repair_validation", "success", "Deterministic dependency repair resolved the build blockage")
             return validation
         except Exception as exc:
+            repair_class = classify_repair_validation_error(exc)
+            write_repair_outcome(
+                run_dir,
+                events_path,
+                post_repair_validation=repair_class["post_repair_validation"],
+                failure_type=repair_class["failure_type"],
+                final_user_message=repair_class["message"],
+                validation_error=str(exc)[:4000],
+            )
+            if repair_class["terminal_status"] == "auth_refresh_required":
+                append_event(events_path, "repair_validation", "auth_refresh_required", repair_class["message"], {"error": str(exc)[:4000]})
+                raise AuthRefreshRequired(repair_class["message"])
             current_error = f"{current_error}\n\nDependency guardrail rebuild did not resolve validation: {str(exc)[:4000]}"
             collect_space_logs(target_space_id, token, run_dir, events_path)
             append_event(events_path, "repair_validation", "failed", "Dependency guardrail rebuild did not resolve validation; falling back to Pi diagnosis", {"error": str(exc)[:4000]})
@@ -3676,10 +4173,27 @@ def recover_after_live_validation_failure(api, workspace: Path, run_dir: Path, e
             append_event(events_path, "factory_rebuild", "success", "Same-code workspace re-uploaded; revalidating live Space")
             append_event(events_path, "repair_validation", "started", "Revalidating after same-code factory rebuild")
             try:
+                write_auth_probe(run_dir, events_path, "before_repair_validation", token, minimum_required_seconds=minimum_auth_seconds_for_phase("before_repair_validation"), raise_on_unsafe=True)
                 validation = validate_live_api(api, target_space_id, token, run_dir, events_path, timeout_s=1200)
+                write_repair_outcome(run_dir, events_path, repair_decision="factory_rebuild_same_code", patch_applied=False, upload_success=True, post_repair_validation="success", failure_type="", final_user_message="Same-code factory rebuild resolved the blockage.")
                 append_event(events_path, "repair_validation", "success", "Same-code factory rebuild resolved the blockage")
                 return validation
             except Exception as exc:
+                repair_class = classify_repair_validation_error(exc)
+                write_repair_outcome(
+                    run_dir,
+                    events_path,
+                    repair_decision="factory_rebuild_same_code",
+                    patch_applied=False,
+                    upload_success=True,
+                    post_repair_validation=repair_class["post_repair_validation"],
+                    failure_type=repair_class["failure_type"],
+                    final_user_message=repair_class["message"],
+                    validation_error=str(exc)[:4000],
+                )
+                if repair_class["terminal_status"] == "auth_refresh_required":
+                    append_event(events_path, "repair_validation", "auth_refresh_required", repair_class["message"], {"error": str(exc)[:4000]})
+                    raise AuthRefreshRequired(repair_class["message"])
                 current_error = f"{current_error}\n\nSame-code factory rebuild did not resolve validation: {str(exc)[:4000]}"
                 collect_space_logs(target_space_id, token, run_dir, events_path)
                 append_event(events_path, "repair_validation", "failed", "Same-code factory rebuild did not resolve validation; re-diagnosing", {"error": str(exc)[:4000]})
@@ -3692,20 +4206,49 @@ def recover_after_live_validation_failure(api, workspace: Path, run_dir: Path, e
                 raise RuntimeError("Automated recovery stopped after patch budget was exhausted")
             budgets[action] -= 1
             append_event(events_path, "repair", "started", "Pi diagnosis allows a minimal code patch", {"decision": decision})
+            write_repair_outcome(
+                run_dir,
+                events_path,
+                repair_trigger=(decision.get("classification") or {}).get("category") or "live_validation_failure",
+                root_cause=(decision.get("classification") or {}).get("category") or "live_validation_failure",
+                repair_decision="patch_code",
+                decision=decision,
+                patch_applied=False,
+                upload_success=False,
+                post_repair_validation="not_started",
+            )
             repaired = repair_workspace_with_pi(workspace, run_dir, events_path, pi_model, target_space_id, model_id, current_error, implementation_mode, expected_output_type, decision=decision)
             if not repaired:
+                write_repair_outcome(run_dir, events_path, patch_applied=False, upload_success=False, post_repair_validation="not_started", failure_type="repair_patch_failed", final_user_message="Structured patch repair failed before redeploy.")
                 write_blockage_artifact(workspace, run_dir, events_path, decision, current_error, status="failed_after_repair")
                 append_event(events_path, "failure", "failed", "Structured patch repair failed before redeploy", {"decision": decision})
                 raise RuntimeError("Structured patch repair failed before redeploy")
+            write_repair_outcome(run_dir, events_path, patch_applied=True, post_repair_validation="not_started")
             append_event(events_path, "repair_upload", "started", "Uploading repaired workspace")
+            write_auth_probe(run_dir, events_path, "before_repair_upload", token, minimum_required_seconds=minimum_auth_seconds_for_phase("before_repair_upload"), raise_on_unsafe=True)
             upload_workspace(api, workspace, target_space_id, token, run_dir, events_path)
+            write_repair_outcome(run_dir, events_path, upload_success=True, post_repair_validation="pending")
             append_event(events_path, "repair_upload", "success", "Repaired workspace uploaded")
             append_event(events_path, "repair_validation", "started", "Revalidating repaired Space")
             try:
+                write_auth_probe(run_dir, events_path, "before_repair_validation", token, minimum_required_seconds=minimum_auth_seconds_for_phase("before_repair_validation"), raise_on_unsafe=True)
                 validation = validate_live_api(api, target_space_id, token, run_dir, events_path, timeout_s=1200)
+                write_repair_outcome(run_dir, events_path, post_repair_validation="success", failure_type="", final_user_message="Repaired Space passed live API validation.")
                 append_event(events_path, "repair_validation", "success", "Repaired Space passed live API validation")
                 return validation
             except Exception as exc:
+                repair_class = classify_repair_validation_error(exc)
+                write_repair_outcome(
+                    run_dir,
+                    events_path,
+                    post_repair_validation=repair_class["post_repair_validation"],
+                    failure_type=repair_class["failure_type"],
+                    final_user_message=repair_class["message"],
+                    validation_error=str(exc)[:4000],
+                )
+                if repair_class["terminal_status"] == "auth_refresh_required":
+                    append_event(events_path, "repair_validation", "auth_refresh_required", repair_class["message"], {"error": str(exc)[:4000]})
+                    raise AuthRefreshRequired(repair_class["message"])
                 current_error = f"{current_error}\n\nPatch repair did not resolve validation: {str(exc)[:4000]}"
                 collect_space_logs(target_space_id, token, run_dir, events_path)
                 append_event(events_path, "repair_validation", "failed", "Repair attempted, but validation still failed", {"error": str(exc)[:4000]})
@@ -3799,8 +4342,12 @@ def infer_generation_gate(workspace: Path, implementation_mode: str, validation:
     }
 
     if blocker_detected:
-        status = "technical_blocker"
-        message = "Space boots, but full model inference was not implemented. See TECHNICAL_BLOCKERS.json / PI_SUMMARY.md."
+        if isinstance(generation_smoke, dict) and generation_smoke.get("skip_reason") == "contract_declared_no_full_inference":
+            status = "technical_blocker_boot_only"
+            message = "Space boots, but Pi declared full inference unavailable; generation smoke was skipped by contract. See TECHNICAL_BLOCKERS.json / PI_SUMMARY.md."
+        else:
+            status = "technical_blocker"
+            message = "Space boots, but full model inference was not implemented. See TECHNICAL_BLOCKERS.json / PI_SUMMARY.md."
     elif implementation_mode in {"full-inference-gated", "full-inference-attempt"} and smoke_ok:
         status = "full_inference_success"
         message = "Space boots and a live generation smoke test passed. ZeroGPU duration recommendation was measured from real inference."
@@ -3891,6 +4438,7 @@ def main():
         api = HfApi(token=token)
         whoami = api.whoami(token=token)
         append_event(events_path, "auth", "success", "Authenticated inside Job", {"whoami_name": whoami.get("name")})
+        write_auth_probe(run_dir, events_path, "before_space_create", token, minimum_required_seconds=minimum_auth_seconds_for_phase("before_space_create"), raise_on_unsafe=True)
 
         append_event(events_path, "model_analysis", "started", "Fetching model metadata", {"model_id": model_id})
         info = api.model_info(model_id, token=token, files_metadata=True)
@@ -3929,6 +4477,7 @@ def main():
         emit_pi_model_resolution(events_path, pi_model_resolution)
         if not (workspace / "PI_SUMMARY.md").exists():
             (workspace / "PI_SUMMARY.md").write_text("# Pi Summary\n\nPi did not create a PI_SUMMARY.md. See logs/pi_output.txt.\n", encoding="utf-8")
+        write_pi_planning_review(workspace, run_dir, events_path, analysis, implementation_mode)
 
         app_text = (workspace / "app.py").read_text(encoding="utf-8", errors="ignore")
         if "/health" not in app_text and "api_name=\"health\"" not in app_text and "api_name='health'" not in app_text:
@@ -3973,6 +4522,7 @@ def main():
         # starts directly on the requested hardware. If it fell back to CPU, the run
         # remains valid but will be marked manual_hardware_required when inference
         # signals indicate GPU is needed.
+        write_auth_probe(run_dir, events_path, "before_upload", token, minimum_required_seconds=minimum_auth_seconds_for_phase("before_upload"), raise_on_unsafe=True)
         upload_workspace(api, workspace, target_space_id, token, run_dir, events_path)
         write_artifact_manifest(run_dir, reason="workspace_uploaded")
 
@@ -3993,21 +4543,25 @@ def main():
             )
         generation_smoke = None
         if implementation_mode in {"full-inference-gated", "full-inference-attempt"}:
-            try:
-                generation_smoke = run_generation_smoke(target_space_id, token, run_dir, events_path, expected_output_type, workspace=workspace)
-            except Exception as smoke_error:
-                write_live_status(run_dir, stage="generation_smoke", status="failed", message="Live generation smoke test failed", data={"error": str(smoke_error)[:2000]})
-                generation_smoke = {
-                    "status": "failed",
-                    "target_space": target_space_id,
-                    "expected_output_type": expected_output_type,
-                    "error": str(smoke_error)[:4000],
-                    "next_action": "Use Prefill Space Test after the run finishes to retry with schema-adjusted arguments.",
-                    **measured_zero_gpu_recommendation(None),
-                }
-                generation_smoke.update(classify_generation_smoke_error(smoke_error))
-                write_json(run_dir / "tests" / "generation_smoke.json", generation_smoke)
-                append_event(events_path, "generation_smoke", "failed", "Live generation smoke test failed; ZeroGPU duration was not measured", generation_smoke)
+            contract_skip_reason = contract_declares_no_full_inference(workspace)
+            if contract_skip_reason.get("declared"):
+                generation_smoke = write_contract_skipped_generation_smoke(run_dir, events_path, expected_output_type, target_space_id, contract_skip_reason)
+            else:
+                try:
+                    generation_smoke = run_generation_smoke(target_space_id, token, run_dir, events_path, expected_output_type, workspace=workspace)
+                except Exception as smoke_error:
+                    write_live_status(run_dir, stage="generation_smoke", status="failed", message="Live generation smoke test failed", data={"error": str(smoke_error)[:2000]})
+                    generation_smoke = {
+                        "status": "failed",
+                        "target_space": target_space_id,
+                        "expected_output_type": expected_output_type,
+                        "error": str(smoke_error)[:4000],
+                        "next_action": "Use Prefill Space Test after the run finishes to retry with schema-adjusted arguments.",
+                        **measured_zero_gpu_recommendation(None),
+                    }
+                    generation_smoke.update(classify_generation_smoke_error(smoke_error))
+                    write_json(run_dir / "tests" / "generation_smoke.json", generation_smoke)
+                    append_event(events_path, "generation_smoke", "failed", "Live generation smoke test failed; ZeroGPU duration was not measured", generation_smoke)
         else:
             generation_smoke = measured_zero_gpu_recommendation(None) | {"status": "skipped", "expected_output_type": expected_output_type}
             write_json(run_dir / "tests" / "generation_smoke.json", generation_smoke)
@@ -4017,7 +4571,7 @@ def main():
         # hardware requests failed, classify the run honestly as needing manual
         # hardware instead of pretending CPU/default hardware is enough. the existing-Space validation workflow
         # can then smoke-test generation after the user sets a GPU manually.
-        manual_hw_required = selected_hardware == "default-cpu-or-existing" and inference_gate.get("status") not in {"technical_blocker", "health_only"} and (
+        manual_hw_required = selected_hardware == "default-cpu-or-existing" and inference_gate.get("status") not in {"technical_blocker", "technical_blocker_boot_only", "health_only"} and (
             inference_gate.get("implementation_signals", {}).get("has_spaces_gpu")
             or inference_gate.get("implementation_signals", {}).get("has_torch")
             or any((a.get("manual_action_required") for a in hardware_attempts if isinstance(a, dict)))
@@ -4134,7 +4688,15 @@ The wrapper validated the live Space using HTTP `/health` first, with Gradio Cli
             collect_pi_traces(run_dir, events_path)
         except Exception:
             pass
-        fail(run_dir, events_path, "Universal model-card builder worker failed", {"error": str(exc)})
+        terminal_status = "auth_refresh_required" if isinstance(exc, AuthRefreshRequired) or is_auth_expired_error(exc) else "failed"
+        details = {"error": str(exc)}
+        if terminal_status == "auth_refresh_required":
+            details["failure_type"] = "auth_refresh_required"
+        repair_outcome = load_json_if_exists(run_dir / "repair_outcome.json") if (run_dir / "repair_outcome.json").exists() else {}
+        if isinstance(repair_outcome, dict) and repair_outcome:
+            details["repair_outcome"] = repair_outcome
+            details["failure_type"] = repair_outcome.get("failure_type") or details.get("failure_type", "")
+        fail(run_dir, events_path, "Universal model-card builder worker failed", details, status=terminal_status)
 
 
 if __name__ == "__main__":
@@ -4144,6 +4706,7 @@ if __name__ == "__main__":
 
 
 VALIDATE_EXISTING_SPACE_WORKER_SCRIPT = r'''
+import base64
 import json
 import os
 import re
@@ -4361,7 +4924,7 @@ def requirements_has_package(lines: list[str], package: str) -> bool:
         stripped = line.strip()
         if not stripped or stripped.startswith("#") or stripped.startswith("-") or "://" in stripped:
             continue
-        name = re.split(r"[<>=!~;\[]", stripped, 1)[0].strip().lower().replace("_", "-")
+        name = re.split(r"[<>=!~;\[]", stripped, maxsplit=1)[0].strip().lower().replace("_", "-")
         if name == wanted:
             return True
     return False
@@ -4547,7 +5110,7 @@ def eval_verdict(status: str, phase: str, health_passed: bool, smoke_passed: boo
         return "cancelled"
     if status == "manual_hardware_required" or inference_gate.get("manual_hardware_required"):
         return "manual_action_required"
-    if status == "technical_blocker":
+    if status in {"technical_blocker", "technical_blocker_boot_only"}:
         return "technical_blocker"
     if full_inference_verified:
         return "success"
@@ -4758,6 +5321,141 @@ def publish_eval_record(run_dir: Path, *, phase: str, events_path: Path | None =
     return {"enabled": True, "written": True, "path": str(dest), "publish_mode": "backend"}
 
 
+
+
+class AuthRefreshRequired(RuntimeError):
+    """Raised when an OAuth/JWT token is expired or too close to expiry."""
+
+
+def _decode_jwt_claims_unverified(token: str) -> dict:
+    """Decode JWT claims without verification, only to inspect non-sensitive expiry metadata.
+
+    This never returns or writes the raw token. Opaque PAT-style tokens are supported by
+    returning an empty dict so they remain usable with an `unknown` expiry state.
+    """
+    try:
+        parts = (token or "").split(".")
+        if len(parts) < 2:
+            return {}
+        payload = parts[1]
+        payload += "=" * ((4 - len(payload) % 4) % 4)
+        raw = base64.urlsafe_b64decode(payload.encode("utf-8"))
+        data = json.loads(raw.decode("utf-8"))
+        return data if isinstance(data, dict) else {}
+    except Exception:
+        return {}
+
+
+def token_expiry_status(token: str, *, minimum_required_seconds: int = 0) -> dict:
+    issued_at = now()
+    if not token:
+        return {
+            "schema_version": "auth_status.v1",
+            "checked_at": issued_at,
+            "token_present": False,
+            "token_kind": "missing",
+            "expiry_known": False,
+            "status": "missing",
+            "safe_for_phase": False,
+            "minimum_required_seconds": minimum_required_seconds,
+        }
+    claims = _decode_jwt_claims_unverified(token)
+    exp = claims.get("exp") if isinstance(claims, dict) else None
+    iat = claims.get("iat") if isinstance(claims, dict) else None
+    token_kind = "oauth_jwt" if exp is not None else ("jwt_without_exp" if claims else "opaque_or_unknown")
+    payload = {
+        "schema_version": "auth_status.v1",
+        "checked_at": issued_at,
+        "token_present": True,
+        "token_kind": token_kind,
+        "expiry_known": exp is not None,
+        "minimum_required_seconds": int(minimum_required_seconds or 0),
+        "token_value": "[REDACTED]",
+    }
+    if iat is not None:
+        try:
+            payload["issued_at"] = datetime.fromtimestamp(int(iat), tz=timezone.utc).isoformat()
+        except Exception:
+            pass
+    if exp is None:
+        payload.update({"status": "unknown", "safe_for_phase": True, "auth_risk": "unknown_expiry"})
+        return payload
+    try:
+        exp_int = int(exp)
+        seconds_left = exp_int - int(time.time())
+        payload.update({
+            "expires_at": datetime.fromtimestamp(exp_int, tz=timezone.utc).isoformat(),
+            "seconds_until_expiry": seconds_left,
+        })
+    except Exception:
+        payload.update({"status": "unknown", "safe_for_phase": True, "auth_risk": "invalid_exp_claim"})
+        return payload
+    if seconds_left <= 0:
+        payload.update({"status": "expired", "safe_for_phase": False, "auth_risk": "expired"})
+    elif minimum_required_seconds and seconds_left < minimum_required_seconds:
+        payload.update({"status": "expires_soon", "safe_for_phase": False, "auth_risk": "expires_before_phase_budget"})
+    else:
+        payload.update({"status": "ok", "safe_for_phase": True, "auth_risk": "ok"})
+    return payload
+
+
+def write_auth_probe(run_dir: Path, events_path: Path | None, phase: str, token: str, *, minimum_required_seconds: int = 0, raise_on_unsafe: bool = False) -> dict:
+    payload = token_expiry_status(token, minimum_required_seconds=minimum_required_seconds)
+    payload["phase"] = phase
+    safe_payload = {k: v for k, v in payload.items() if k != "token_value"}
+    try:
+        probes_dir = run_dir / "auth_probes"
+        probes_dir.mkdir(parents=True, exist_ok=True)
+        write_json(probes_dir / f"{phase}.json", safe_payload)
+        write_json(run_dir / "auth_status.json", safe_payload)
+    except Exception:
+        pass
+    status = str(payload.get("status") or "unknown")
+    if events_path:
+        event_status = "success" if payload.get("safe_for_phase") else "failed"
+        if status == "unknown":
+            event_status = "warning"
+        append_event(
+            events_path,
+            "auth_probe",
+            event_status,
+            f"HF OAuth/token expiry check for {phase}: {status}",
+            safe_payload,
+        )
+    if raise_on_unsafe and not payload.get("safe_for_phase"):
+        raise AuthRefreshRequired(f"HF auth token is {status} before {phase}; refresh sign-in before continuing.")
+    return payload
+
+
+def is_auth_expired_error(error: Exception | str) -> bool:
+    text = str(error or "").lower()
+    return any(marker in text for marker in [
+        "oauth token has expired",
+        "exp claim timestamp check failed",
+        "token has expired",
+        "jwt expired",
+    ])
+
+
+def minimum_auth_seconds_for_phase(phase: str) -> int:
+    env_key = "ASF_AUTH_MIN_SECONDS_" + re.sub(r"[^A-Z0-9]+", "_", phase.upper()).strip("_")
+    raw = os.environ.get(env_key) or os.environ.get("ASF_AUTH_MIN_SECONDS", "")
+    if raw:
+        try:
+            return max(0, int(raw))
+        except Exception:
+            pass
+    defaults = {
+        "before_space_create": 900,
+        "before_upload": 900,
+        "before_initial_validation": 1800,
+        "before_repair": 1800,
+        "before_repair_upload": 1800,
+        "before_repair_validation": 1800,
+        "before_linked_space_test": 1800,
+    }
+    return defaults.get(phase, 0)
+
 def ensure_hf_token_context(run_dir: Path, events_path: Path | None = None) -> dict:
     """Ensure Pi and HF tooling see the same private HF token safely.
 
@@ -4771,6 +5469,7 @@ def ensure_hf_token_context(run_dir: Path, events_path: Path | None = None) -> d
         os.environ.setdefault("HF_TOKEN", token)
         os.environ.setdefault("HUGGING_FACE_HUB_TOKEN", token)
         os.environ.setdefault("HF_HUB_ENABLE_HF_TRANSFER", "1")
+    expiry = token_expiry_status(token, minimum_required_seconds=minimum_auth_seconds_for_phase("before_initial_validation"))
     payload = {
         "hf_token_present": bool(token),
         "hf_token_length": len(token) if token else 0,
@@ -4779,6 +5478,13 @@ def ensure_hf_token_context(run_dir: Path, events_path: Path | None = None) -> d
         "bucket_source": os.environ.get("BUCKET_SOURCE") or "",
         "target_space_id": os.environ.get("TARGET_SPACE_ID") or "",
         "token_value": "[REDACTED]" if token else "",
+        "token_kind": expiry.get("token_kind"),
+        "expiry_known": expiry.get("expiry_known"),
+        "expires_at": expiry.get("expires_at"),
+        "seconds_until_expiry_at_job_start": expiry.get("seconds_until_expiry"),
+        "auth_risk": expiry.get("auth_risk"),
+        "safe_for_long_build": bool(expiry.get("safe_for_phase")),
+        "minimum_required_seconds": expiry.get("minimum_required_seconds"),
     }
     run_dir.mkdir(parents=True, exist_ok=True)
     write_json(run_dir / "token_context.json", payload)
@@ -6086,6 +6792,8 @@ def main():
     write_json(state_path, {"run_id": run_id, "kind": "linked_space_validation", "status": "running", "target_space": target_space_id, "parent_build_run_id": parent_build_run_id, "source_kind": "build_run_prefill", "created_by": username, "updated_at": now()})
     if not token:
         raise RuntimeError("HF_TOKEN is missing")
+    ensure_hf_token_context(run_dir, events_path)
+    write_auth_probe(run_dir, events_path, "before_linked_space_test", token, minimum_required_seconds=minimum_auth_seconds_for_phase("before_linked_space_test"), raise_on_unsafe=True)
     if not TARGET_RE.match(target_space_id):
         raise ValueError("TARGET_SPACE_ID must look like owner/space-name")
     if not parent_build_run_id:
@@ -6186,14 +6894,19 @@ Target Space: [`{target_space_id}`](https://huggingface.co/spaces/{target_space_
         append_event(events_path, "done", "full_inference_success", "Existing Space validation completed", {"latency_seconds": smoke.get("latency_seconds")})
     except Exception as exc:
         collect_space_logs(target_space_id, token or "", run_dir, events_path)
+        auth_blocked = isinstance(exc, AuthRefreshRequired) or is_auth_expired_error(exc)
         details = {"error": str(exc)[:4000]}
-        failure_state = {"run_id": run_id, "kind": "linked_space_validation", "status": "failed", "target_space": target_space_id, "parent_build_run_id": parent_build_run_id, "source_kind": "build_run_prefill", "details": details, "updated_at": now()}
+        if auth_blocked:
+            details["status_reason"] = "auth_refresh_required"
+            details["failure_type"] = "auth_refresh_required"
+        terminal_status = "auth_refresh_required" if auth_blocked else "failed"
+        failure_state = {"run_id": run_id, "kind": "linked_space_validation", "status": terminal_status, "message": "HF OAuth token expired or will expire too soon; refresh sign-in before continuing" if auth_blocked else str(exc), "target_space": target_space_id, "parent_build_run_id": parent_build_run_id, "source_kind": "build_run_prefill", "details": details, "updated_at": now()}
         write_json(state_path, failure_state)
-        write_final_summary(run_dir, failure_state, {}, {"status": "failed", "failure_type": details.get("failure_type", "linked_validation_failed") if isinstance(details, dict) else "linked_validation_failed"}, status="failed", message=str(details.get("error", "Existing Space validation failed")) if isinstance(details, dict) else "Existing Space validation failed")
+        write_final_summary(run_dir, failure_state, {}, {"status": terminal_status, "failure_type": details.get("failure_type", "linked_validation_failed") if isinstance(details, dict) else "linked_validation_failed"}, status=terminal_status, message=str(failure_state.get("message") or details.get("error", "Existing Space validation failed")) if isinstance(details, dict) else "Existing Space validation failed")
         if parent_build_run_id:
             parent_dir = output_root / "runs" / parent_build_run_id
             parent_dir.mkdir(parents=True, exist_ok=True)
-            failed_status = {"status": "failed", "validation_run_id": run_id, "parent_build_run_id": parent_build_run_id, "target_space": target_space_id, "details": details, "updated_at": now(), "effective_status": "unchanged"}
+            failed_status = {"status": terminal_status, "validation_run_id": run_id, "parent_build_run_id": parent_build_run_id, "target_space": target_space_id, "details": details, "updated_at": now(), "effective_status": "unchanged"}
             linked_path = parent_dir / "linked_validations.json"
             linked = read_json(linked_path, {"parent_build_run_id": parent_build_run_id, "validations": []}) or {"parent_build_run_id": parent_build_run_id, "validations": []}
             validations = linked.get("validations") if isinstance(linked, dict) else []
@@ -6208,7 +6921,7 @@ Target Space: [`{target_space_id}`](https://huggingface.co/spaces/{target_space_
                 write_json(parent_dir / "manual_validation_status.json", failed_status)
             write_json(linked_path, linked)
         (run_dir / "report.md").write_text(f"# Existing Space Validation Failed\n\n```json\n{json.dumps(details, indent=2, ensure_ascii=False)}\n```\n", encoding="utf-8")
-        append_event(events_path, "failure", "failed", "Existing Space validation failed", details)
+        append_event(events_path, "failure", "failed", "HF OAuth token expired or will expire too soon; refresh sign-in before continuing" if auth_blocked else "Existing Space validation failed", details)
         try:
             publish_eval_record(run_dir, phase="failure", events_path=events_path)
             append_event(events_path, "anonymous_eval", "success", "Published anonymized failed validation evaluation record locally for backend archive publishing", {"enabled": True, "publish_mode": "backend", "archive_publish_confirmed": False})
@@ -6217,8 +6930,8 @@ Target Space: [`{target_space_id}`](https://huggingface.co/spaces/{target_space_
         # v190.33: also rewrite failure summary as the final visible file update
         # for linked validations, so a failed validation cannot remain listed
         # as running after the worker has already terminalized state.json.
-        write_final_summary(run_dir, failure_state, {}, {"status": "failed", "failure_type": details.get("failure_type", "linked_validation_failed") if isinstance(details, dict) else "linked_validation_failed"}, status="failed", message=str(details.get("error", "Existing Space validation failed")) if isinstance(details, dict) else "Existing Space validation failed")
-        append_event(events_path, "summary_write", "success", "Wrote terminal failed validation summary", {"status": "failed"})
+        write_final_summary(run_dir, failure_state, {}, {"status": terminal_status, "failure_type": details.get("failure_type", "linked_validation_failed") if isinstance(details, dict) else "linked_validation_failed"}, status=terminal_status, message=str(failure_state.get("message") or details.get("error", "Existing Space validation failed")) if isinstance(details, dict) else "Existing Space validation failed")
+        append_event(events_path, "summary_write", "success", "Wrote terminal failed validation summary", {"status": terminal_status})
         raise SystemExit(1)