Upload 16 files

src/bucket.py

@@ -1820,6 +1820,7 @@ def read_run_bundle(run_id: str, *, bucket_source: str, token: str | None = None
         "build_error_observation": _safe_read_json(f"{paths.root}/build_error_observation.json", token=token),
         "repair_decision": _safe_read_json(f"{paths.root}/repair/REPAIR_DECISION.json", token=token),
         "blockage": _safe_read_json(f"{paths.root}/repair/BLOCKAGE.json", token=token),
+        "provider_quota_blocker": _safe_read_json(f"{paths.root}/PROVIDER_QUOTA_BLOCKER.json", token=token) or _safe_read_json(f"{paths.root}/provider_quota_blocker.json", token=token),
         "repair_attempts": _safe_read_json(f"{paths.root}/repair/repair_attempts.json", token=token),
         "artifact_manifest": _safe_read_json(f"{paths.root}/artifact_manifest.json", token=token),
         "files": _list_run_files(run_id, bucket_source=bucket_source, token=token) if include_heavy else [],
@@ -1906,6 +1907,7 @@ def _run_list_summary_needs_enrichment(run_id: str, summary_file: dict[str, Any]
         "technical_blocker",
         "technical_blocker_boot_only",
         "blocked",
+        "provider_quota_blocked",
         "failed",
         "failure",
         "error",
@@ -1981,7 +1983,7 @@ def _build_run_list_item_from_files(run_id: str, *, root: str, bucket_source: st
         item["effective_status"] = post_build_validation_status.get("effective_status") or item.get("effective_status")
     if linked_validations:
         rows = linked_validations.get("validations") if isinstance(linked_validations, dict) else []
-        terminal_tokens = {"success", "passed", "succeeded", "full_inference_success", "validated_after_space_test", "validated_after_manual_space_test", "success_with_expert_intervention", "expert_intervention_validation_passed", "recovered_by_space_test", "recovered_by_manual_validation", "manual_validation_passed", "manual_validated", "partial_validation", "manual_hardware_required", "generated_needs_manual_hardware", "technical_blocker", "technical_blocker_boot_only", "blocked", "completed_with_warnings", "failed", "failure", "error", "auth_refresh_required", "stale", "stopped", "cancelled", "canceled"}
+        terminal_tokens = {"success", "passed", "succeeded", "full_inference_success", "validated_after_space_test", "validated_after_manual_space_test", "success_with_expert_intervention", "expert_intervention_validation_passed", "recovered_by_space_test", "recovered_by_manual_validation", "manual_validation_passed", "manual_validated", "partial_validation", "manual_hardware_required", "generated_needs_manual_hardware", "technical_blocker", "technical_blocker_boot_only", "blocked", "completed_with_warnings", "failed", "failure", "error", "auth_refresh_required", "provider_quota_blocked", "stale", "stopped", "cancelled", "canceled"}
         accounted_rows = [row for row in rows if isinstance(row, dict) and str(row.get("status") or row.get("validation_status") or row.get("result_status") or row.get("effective_status") or "").lower() in terminal_tokens] if isinstance(rows, list) else []
         item["linked_validations"] = linked_validations
         item["linked_validation_count"] = len(accounted_rows)

src/effective_status.py

@@ -43,7 +43,8 @@ MANUAL_BUILD_TOKENS = {
     "manual_action_required",
     "waiting_manual_action",
 }
-BLOCKED_BUILD_TOKENS = {"technical_blocker", "technical_blocker_boot_only", "blocked"}
+BLOCKED_BUILD_TOKENS = {"technical_blocker", "technical_blocker_boot_only", "blocked", "provider_quota_blocked"}
+PROVIDER_QUOTA_TOKENS = {"provider_quota_blocked"}
 
 
 def _lower(value: Any) -> str:
@@ -193,6 +194,8 @@ def _normalize_build_status(value: Any) -> str:
         return "full_inference_success"
     if status in {"repair_validation_failed", "upload_failed", "smoke_still_failed", "repair_failed_same_error", "repair_exhausted", "no_patch_produced_by_pi", "no_relevant_patch_produced_by_pi", "pi_patch_rejected_by_guard"}:
         return "failed"
+    if status in {"provider_quota_blocked"}:
+        return "provider_quota_blocked"
     if status in {"cancelled", "canceled"}:
         return "stopped"
     if status in {"full_inference_candidate_health_passed", "health_only", "partial", "completed_with_warnings"}:
@@ -295,6 +298,8 @@ def compute_effective_run_status(
         display_label = "Partial validation"
     elif display_status == "manual_hardware_required":
         display_label = "Manual hardware required"
+    elif display_status == "provider_quota_blocked":
+        display_label = "Provider quota blocked"
     elif display_status in BLOCKED_BUILD_TOKENS:
         display_label = "Diagnostic Space"
     elif display_status in FAILED_BUILD_TOKENS:

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", "success_with_expert_intervention", "expert_intervention_validation_passed", "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"}
+_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", "success_with_expert_intervention", "expert_intervention_validation_passed", "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", "provider_quota_blocked"}
 
 
 def _now() -> str:

src/progress.py

@@ -96,6 +96,7 @@ STEP_LABELS = {
 }
 
 STEP_ALIASES = {
+    "provider_quota": "pi_run",
     "bucket_ready": "bootstrap",
     "job_launched": "bootstrap",
     "hardware_preferred": "hardware_strategy",
@@ -129,6 +130,7 @@ PARTIAL_STATUSES = {
 }
 RUNNING_STATUSES = {"started", "running", "waiting", "pending", "scheduled"}
 FAILED_STATUSES = {"failed", "error", "failure", "repair_validation_failed", "upload_failed", "smoke_still_failed", "repair_failed_same_error", "repair_exhausted", "no_patch_produced_by_pi", "no_relevant_patch_produced_by_pi", "pi_patch_rejected_by_guard"}
+BLOCKED_STATUSES = {"provider_quota_blocked"}
 CANCELLED_STATUSES = {"cancelled", "canceled"}
 
 
@@ -150,7 +152,7 @@ def _canonical_step(step: str | None) -> str | None:
 
 def _is_terminal_status(status: str | None) -> bool:
     s = str(status or "").lower()
-    return s in DONE_STATUSES or s in PARTIAL_STATUSES or s in FAILED_STATUSES or s in CANCELLED_STATUSES
+    return s in DONE_STATUSES or s in PARTIAL_STATUSES or s in FAILED_STATUSES or s in BLOCKED_STATUSES or s in CANCELLED_STATUSES
 
 
 def _progress_for_index(index: int) -> int:
@@ -243,6 +245,8 @@ def progress_from_events(events: list[dict[str, Any]] | None, *, state: dict[str
         terminal_status = "cancelled"
     elif status_from_state in PARTIAL_STATUSES:
         terminal_status = status_from_state
+    elif status_from_state in BLOCKED_STATUSES:
+        terminal_status = status_from_state
     elif status_from_state in DONE_STATUSES:
         terminal_status = status_from_state
     elif status_from_state in FAILED_STATUSES:
@@ -318,7 +322,7 @@ def progress_from_events(events: list[dict[str, Any]] | None, *, state: dict[str
 
     if overall_status in FAILED_STATUSES:
         progress = max(_progress_for_index(furthest_index), 3)
-    elif overall_status in CANCELLED_STATUSES:
+    elif overall_status in CANCELLED_STATUSES or overall_status in BLOCKED_STATUSES:
         progress = max(_progress_for_index(furthest_index), 3)
     elif overall_status in DONE_STATUSES or overall_status in PARTIAL_STATUSES or current_step == "done":
         progress = 100
@@ -350,7 +354,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", "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"))))),
+        "visual_status": "error" if overall_status in FAILED_STATUSES else ("warn" if overall_status in PARTIAL_STATUSES or overall_status in BLOCKED_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

@@ -31,7 +31,7 @@ PHASE_LABELS = {
 PHASE_STEPS = {
     "start": {"token_context", "bootstrap", "dependencies", "auth", "workspace", "node", "pi_install"},
     "model": {"model_analysis", "model_prescan"},
-    "agent": {"pi_config", "pi_run", "traces", "pi_model_resolution", "pi_verification"},
+    "agent": {"pi_config", "pi_run", "provider_quota", "traces", "pi_model_resolution", "pi_verification"},
     "hardware": {"hardware_strategy", "create_space_hardware", "create_space"},
     "deploy": {"metadata_sanitize", "requirements_sanitize", "upload_files", "space_logs", "space_runtime", "live_wait"},
     "live_validation": {"endpoint_discovery", "api_validation", "generation_smoke", "inference_gate"},
@@ -51,7 +51,7 @@ PHASE_STEPS = {
         "repair_validation",
     },
     "archive": {"report_write", "artifact_manifest", "anonymous_eval", "eval_publish", "eval_archive"},
-    "done": {"done", "failure", "technical_blocker", "technical_blocker_boot_only", "manual_hardware_required"},
+    "done": {"done", "failure", "technical_blocker", "technical_blocker_boot_only", "provider_quota", "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", "technical_blocker_boot_only", "blocked"}
+BLOCKED_STATUSES = {"technical_blocker", "technical_blocker_boot_only", "blocked", "provider_quota_blocked"}
+PROVIDER_QUOTA_STATUSES = {"provider_quota_blocked"}
 AUTH_STATUSES = {"auth_refresh_required", "oauth_expired", "auth_expired", "repair_validation_inconclusive_auth", "inconclusive_auth_expired"}
 
 TERMINAL_SOURCE_KEYS = ("status", "final_status", "effective_status", "effective_verdict", "display_status", "verdict", "result_status", "validation_status")
@@ -783,6 +784,8 @@ def _normalize_verdict_token(value: Any) -> str:
         return "partial_validation"
     if normalized in MANUAL_STATUSES:
         return "manual_action_required"
+    if normalized in PROVIDER_QUOTA_STATUSES:
+        return "provider_quota_blocked"
     if normalized in BLOCKED_STATUSES:
         return "technical_blocker_boot_only" if normalized == "technical_blocker_boot_only" else "technical_blocker"
     if normalized in FAILED_STATUSES:
@@ -846,10 +849,14 @@ 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 {"provider_quota_blocked"}:
+        return "warn"
     if verdict in {"failed", "failure", "technical_blocker", "technical_blocker_boot_only"}:
         return "error"
     if verdict == "auth_refresh_required":
         return "warn"
+    if verdict == "provider_quota_blocked":
+        return "warn"
     if verdict in {"manual_action_required", "cancelled"}:
         return "stopped"
     return "running"
@@ -866,6 +873,8 @@ def _status_label(verdict: str) -> tuple[str, str]:
         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 == "provider_quota_blocked":
+        return "Provider quota blocked", "HF Inference Providers quota or monthly spending limit blocked Pi before demo generation."
     if verdict == "manual_action_required":
         return "Manual action required", "The run needs user action before validation can continue."
     if verdict == "failed":

src/version.py

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

src/view_models.py

@@ -44,11 +44,12 @@ MANUAL_RAW_STATUSES = {
 FAILED_RAW_STATUSES = {"failed", "failure", "error", "repair_failed", "repair_validation_failed", "upload_failed", "smoke_still_failed", "repair_failed_same_error", "repair_exhausted", "no_patch_produced_by_pi", "no_relevant_patch_produced_by_pi", "pi_patch_rejected_by_guard"}
 CANCELLED_RAW_STATUSES = {"cancelled", "canceled"}
 BLOCKED_RAW_STATUSES = {"technical_blocker", "technical_blocker_boot_only", "blocked"}
+PROVIDER_QUOTA_RAW_STATUSES = {"provider_quota_blocked"}
 RUNNING_RAW_STATUSES = {"started", "running", "waiting", "queued", "pending", "scheduled"}
 AUTH_RAW_STATUSES = {"auth_refresh_required", "oauth_expired", "auth_expired", "repair_validation_inconclusive_auth", "inconclusive_auth_expired"}
 
 AUTHORITATIVE_TERMINAL_SOURCE_KEYS = ("status", "final_status", "effective_status", "effective_verdict", "display_status", "verdict", "result_status", "validation_status", "post_repair_validation", "failure_type")
-TERMINAL_DISPLAY_STATUSES = SUCCESS_RAW_STATUSES | PARTIAL_RAW_STATUSES | MANUAL_RAW_STATUSES | FAILED_RAW_STATUSES | CANCELLED_RAW_STATUSES | BLOCKED_RAW_STATUSES | AUTH_RAW_STATUSES | {"stale", "stopped", "succeeded", "validated_after_space_test", "validated_after_manual_space_test", "success_with_expert_intervention", "expert_intervention_validation_passed", "recovered_by_space_test", "recovered_by_manual_validation"}
+TERMINAL_DISPLAY_STATUSES = SUCCESS_RAW_STATUSES | PARTIAL_RAW_STATUSES | MANUAL_RAW_STATUSES | FAILED_RAW_STATUSES | CANCELLED_RAW_STATUSES | BLOCKED_RAW_STATUSES | PROVIDER_QUOTA_RAW_STATUSES | AUTH_RAW_STATUSES | {"stale", "stopped", "succeeded", "validated_after_space_test", "validated_after_manual_space_test", "success_with_expert_intervention", "expert_intervention_validation_passed", "recovered_by_space_test", "recovered_by_manual_validation"}
 
 STEP_TO_PHASE = {step: step for step in STEP_ORDER} | {alias: canonical for alias, canonical in STEP_ALIASES.items()}
 
@@ -203,7 +204,7 @@ def _first_authoritative_terminal_status(bundle: dict[str, Any]) -> tuple[str, s
         for key in AUTHORITATIVE_TERMINAL_SOURCE_KEYS:
             normalized = _normalize_terminal_status(source.get(key))
             if normalized:
-                if normalized in BLOCKED_RAW_STATUSES | FAILED_RAW_STATUSES | MANUAL_RAW_STATUSES | AUTH_RAW_STATUSES | CANCELLED_RAW_STATUSES | {"stopped", "stale"}:
+                if normalized in BLOCKED_RAW_STATUSES | PROVIDER_QUOTA_RAW_STATUSES | FAILED_RAW_STATUSES | MANUAL_RAW_STATUSES | AUTH_RAW_STATUSES | CANCELLED_RAW_STATUSES | {"stopped", "stale"}:
                     return normalized, f"{source_name}.{key}"
                 if normalized == "full_inference_success" and blocking_seen:
                     continue
@@ -245,6 +246,8 @@ def _status_model_from_terminal_status(status: str, *, source: str = "") -> dict
         global_status, verdict = "partial", normalized if normalized not in {"partial", "health_only", "full_inference_candidate_health_passed", "completed_with_warnings"} else "partial_validation"
     elif normalized in MANUAL_RAW_STATUSES:
         global_status, verdict = "waiting_manual_action", "manual_action_required"
+    elif normalized in PROVIDER_QUOTA_RAW_STATUSES:
+        global_status, verdict = "blocked", "provider_quota_blocked"
     elif normalized in BLOCKED_RAW_STATUSES:
         global_status, verdict = "blocked", "technical_blocker_boot_only" if normalized == "technical_blocker_boot_only" else "technical_blocker"
     elif normalized in FAILED_RAW_STATUSES:
@@ -310,6 +313,9 @@ def normalize_run_status(
     elif manual:
         global_status = "waiting_manual_action"
         verdict = "manual_action_required"
+    elif statuses.intersection(PROVIDER_QUOTA_RAW_STATUSES):
+        global_status = "blocked"
+        verdict = "provider_quota_blocked"
     elif blocker:
         global_status = "blocked"
         verdict = "technical_blocker_boot_only" if "technical_blocker_boot_only" in statuses else "technical_blocker"
@@ -479,7 +485,12 @@ def build_diagnostics(bundle: dict[str, Any], status_model: dict[str, Any], phas
     issue_title = ""
     issue_detail = ""
     issue_status = ""
-    if blocker_items:
+    provider_quota = bundle.get("provider_quota_blocker") or (bundle.get("state") or {}).get("provider_quota_blocker") or ((bundle.get("state") or {}).get("details") or {}).get("provider_quota_blocker")
+    if isinstance(provider_quota, dict) and provider_quota:
+        issue_title = provider_quota.get("title") or "Provider quota blocked"
+        issue_detail = provider_quota.get("user_message") or provider_quota.get("summary") or "Hugging Face Inference Providers quota or billing blocked Pi before ASF could generate the demo."
+        issue_status = "action_required"
+    elif blocker_items:
         first = blocker_items[0]
         issue_title = _first_nonempty(first.get("type"), first.get("name"), "Technical blocker") if isinstance(first, dict) else "Technical blocker"
         issue_detail = _first_nonempty(first.get("claim"), first.get("reason"), first.get("message")) if isinstance(first, dict) else str(first)

src/worker_payload.py

@@ -40,8 +40,8 @@ DEFAULT_FALLBACK_SPACE_HARDWARE = "a10g-large"
 DEFAULT_MODEL_ID = "sshleifer/tiny-gpt2"
 MAX_PI_REPAIR_ATTEMPTS = 3
 DEFAULT_FACTORY_UPLOAD_RETRY_BUDGET = 1
-APP_VERSION = "v198.26.127"
-app_version = "v198.26.127"
+APP_VERSION = "v198.26.130"
+app_version = "v198.26.130"
 
 # Internal agent/recovery files may be needed inside the transient Pi
 # workspace, but they should not be published to the generated Space or shown
@@ -630,6 +630,7 @@ def build_runtime_upload_payload(workspace: Path, run_dir: Path, events_path: Pa
     # candidate and stamp the app so /health can prove which candidate is
     # actually being served. No further runtime mutation is allowed after this.
     freeze_release_candidate(payload_dir, workspace, run_dir, events_path, manifest, target_space_id=os.environ.get("TARGET_SPACE_ID", ""))
+    release_candidate_payload_compile_guard(payload_dir, run_dir, events_path, reason="runtime_upload_payload_post_release_stamp")
     manifest_path = run_dir / "runtime_upload_payload_manifest.json"
     write_json(manifest_path, manifest)
 
@@ -739,6 +740,46 @@ def _current_repair_attempt_for_release(run_dir: Path) -> int | None:
     return None
 
 
+def _release_identity_insert_index(app_text: str) -> int:
+    """Return a safe top-level insertion point for the ASF runtime stamp.
+
+    v198.26.127 inserted after any line whose stripped text looked like an
+    import. That accidentally matched imports inside top-level try/except
+    blocks, producing invalid Python such as `try: import torch; ASF_STAMP;
+    except ...`.  Only top-level module body statements are safe anchors.
+    """
+    try:
+        tree = ast.parse(app_text)
+        lines = app_text.splitlines(True)
+        insert_after = 0
+        for idx, node in enumerate(tree.body):
+            if idx == 0 and isinstance(node, ast.Expr) and isinstance(getattr(node, "value", None), ast.Constant) and isinstance(node.value.value, str):
+                insert_after = int(getattr(node, "end_lineno", node.lineno) or node.lineno)
+                continue
+            if isinstance(node, ast.ImportFrom) and getattr(node, "module", "") == "__future__":
+                insert_after = int(getattr(node, "end_lineno", node.lineno) or node.lineno)
+                continue
+            if isinstance(node, (ast.Import, ast.ImportFrom)) and int(getattr(node, "col_offset", 0) or 0) == 0:
+                insert_after = int(getattr(node, "end_lineno", node.lineno) or node.lineno)
+                continue
+            break
+        return max(0, min(insert_after, len(lines)))
+    except Exception:
+        lines = app_text.splitlines(True)
+        insert_after = 0
+        # Conservative fallback: only unindented imports/future imports count.
+        for i, line in enumerate(lines[:120]):
+            if line.startswith("import ") or line.startswith("from "):
+                insert_after = i + 1
+                continue
+            stripped = line.strip()
+            if not stripped or stripped.startswith("#"):
+                continue
+            if insert_after:
+                break
+        return insert_after
+
+
 def _insert_asf_release_constant(app_text: str, release: dict) -> str:
     marker = "# ASF runtime release identity stamp"
     if marker in app_text:
@@ -772,48 +813,90 @@ def _insert_asf_release_constant(app_text: str, release: dict) -> str:
         "\n"
     )
     lines = app_text.splitlines(True)
-    insert_at = 0
-    # Preserve shebang/encoding/comments and place after the import block where possible.
-    for i, line in enumerate(lines[:80]):
-        stripped = line.strip()
-        if stripped.startswith("import ") or stripped.startswith("from "):
-            insert_at = i + 1
+    insert_at = _release_identity_insert_index(app_text)
     lines.insert(insert_at, block)
     return "".join(lines)
 
 
-def _wrap_health_functions_with_release_identity(app_text: str, run_dir: Path | None = None) -> tuple[str, list[str]]:
+def _top_level_health_functions(app_text: str) -> tuple[list[tuple[int, str, bool]], list[str]]:
     try:
         tree = ast.parse(app_text)
     except Exception:
-        return app_text, []
-    wrappers: list[tuple[int, str]] = []
-    names = []
-    for node in ast.walk(tree):
+        return [], []
+    wrappers: list[tuple[int, str, bool]] = []
+    names: list[str] = []
+    for node in tree.body:
         if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
             name = str(node.name or "")
             lname = name.lower()
             if lname in {"health", "health_check", "runtime_health", "get_health"} or ("health" in lname and not lname.startswith("_asf")):
                 if getattr(node, "end_lineno", None):
                     names.append(name)
-                    wrappers.append((int(node.end_lineno), name))
+                    wrappers.append((int(node.end_lineno), name, isinstance(node, ast.AsyncFunctionDef)))
+    return wrappers, names
+
+
+def _wrap_health_functions_with_release_identity(app_text: str, run_dir: Path | None = None) -> tuple[str, list[str]]:
+    # Only wrap top-level health functions. Nested functions or class methods
+    # must not receive top-level wrapper assignments based on their end_lineno,
+    # because that can inject code into an unrelated block and break generated apps.
+    wrappers, names = _top_level_health_functions(app_text)
     if not wrappers:
         return app_text, []
     lines = app_text.splitlines(True)
-    for end_lineno, name in sorted(wrappers, reverse=True):
+    for end_lineno, name, is_async in sorted(wrappers, reverse=True):
         alias = f"_asf_original_{name}"
-        wrapper = (
-            f"\ntry:\n"
-            f"    {alias} = {name}\n"
-            f"    def {name}(*args, **kwargs):\n"
-            f"        return _asf_stamp_health_payload({alias}(*args, **kwargs))\n"
-            f"except Exception:\n"
-            f"    pass\n\n"
-        )
+        if is_async:
+            wrapper = (
+                f"\ntry:\n"
+                f"    {alias} = {name}\n"
+                f"    async def {name}(*args, **kwargs):\n"
+                f"        _asf_result = {alias}(*args, **kwargs)\n"
+                f"        if hasattr(_asf_result, '__await__'):\n"
+                f"            _asf_result = await _asf_result\n"
+                f"        return _asf_stamp_health_payload(_asf_result)\n"
+                f"except Exception:\n"
+                f"    pass\n\n"
+            )
+        else:
+            wrapper = (
+                f"\ntry:\n"
+                f"    {alias} = {name}\n"
+                f"    def {name}(*args, **kwargs):\n"
+                f"        return _asf_stamp_health_payload({alias}(*args, **kwargs))\n"
+                f"except Exception:\n"
+                f"    pass\n\n"
+            )
         lines.insert(end_lineno, wrapper)
     return "".join(lines), names
 
 
+def release_candidate_payload_compile_guard(payload_dir: Path, run_dir: Path, events_path: Path | None, *, reason: str = "release_candidate_post_stamp") -> dict:
+    app_path = payload_dir / "app.py"
+    result = {
+        "schema_version": "release_candidate_payload_compile_guard.v198_26_129",
+        "reason": reason,
+        "checked_file": "runtime_upload_payload/app.py",
+        "passed": False,
+        "error": "",
+        "updated_at": now(),
+    }
+    if not app_path.exists():
+        result["error"] = "app.py missing from runtime upload payload"
+    else:
+        try:
+            compile(app_path.read_text(encoding="utf-8", errors="ignore"), str(app_path), "exec")
+            result["passed"] = True
+        except Exception as exc:
+            result["error"] = f"{type(exc).__name__}: {exc}"
+    write_json(run_dir / "release" / "RELEASE_CANDIDATE_COMPILE_GUARD.json", result)
+    if events_path is not None:
+        append_event(events_path, "release_candidate_compile_guard", "success" if result.get("passed") else "failed", "Checked stamped runtime payload app.py before upload", result)
+    if not result.get("passed"):
+        raise RuntimeError("release_candidate_payload_compile_failed: " + str(result.get("error") or "unknown"))
+    return result
+
+
 def patch_runtime_payload_with_release_identity(payload_dir: Path, release: dict, run_dir: Path, events_path: Path) -> dict:
     payload_dir.mkdir(parents=True, exist_ok=True)
     write_json(payload_dir / "ASF_RELEASE.json", release)
@@ -2487,6 +2570,7 @@ def eval_terminal_status(status: str, phase: str = "") -> bool:
         "partial_validation",
         "technical_blocker",
         "manual_hardware_required",
+        "provider_quota_blocked",
     }
 
 
@@ -2541,6 +2625,8 @@ 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 == "provider_quota_blocked":
+        return "provider_quota_blocked"
     if status in {"technical_blocker", "technical_blocker_boot_only"}:
         return "technical_blocker"
     if full_inference_verified:
@@ -2925,7 +3011,10 @@ def run_final_cleanup_if_needed(run_dir: Path, events_path: Path, final_state: d
     health_semantic_passed = signals.get("health_semantic_passed")
     if health_semantic_passed is None:
         health_semantic_passed = signals.get("health_passed")
-    if runtime_recovery.get("triggered") and runtime_recovery.get("recovery_attempted") and not runtime_recovery.get("recovery_exhausted"):
+    space_creation = generated_space_creation_evidence(run_dir, events_path)
+    if target_space and not space_creation.get("space_created"):
+        decision = {"should_pause": False, "reason": "target_space_not_created", "final_status": final_status, "selected_hardware": selected_hardware, "target_space_planned": target_space, "space_creation_evidence": space_creation}
+    elif runtime_recovery.get("triggered") and runtime_recovery.get("recovery_attempted") and not runtime_recovery.get("recovery_exhausted"):
         decision = {"should_pause": False, "reason": "runtime_recovery_pending", "final_status": final_status, "selected_hardware": selected_hardware}
     else:
         decision = should_pause_generated_space(final_status, target_space, selected_hardware, health_semantic_passed=health_semantic_passed if isinstance(health_semantic_passed, bool) else None)
@@ -2949,9 +3038,140 @@ def run_final_cleanup_if_needed(run_dir: Path, events_path: Path, final_state: d
     write_json(run_dir / "state.json", state)
     return cleanup
 
+
+PROVIDER_QUOTA_BLOCKED_STATUS = "provider_quota_blocked"
+PROVIDER_QUOTA_MARKERS = (
+    "monthly spending limit for inference providers",
+    "exceeded your monthly spending limit",
+    "spending limit for inference providers",
+    "inference providers quota",
+    "provider quota",
+    "quota exceeded",
+    "insufficient credits",
+    "billing limit",
+    "payment required",
+)
+
+
+def classify_provider_quota_blocker(text: str | None, *, phase: str = "pi_run", returncode: int | None = None) -> dict:
+    """Return a user-facing blocker when HF Inference Providers quota/billing stopped Pi.
+
+    This is not a generated Space failure. It happens before ASF can produce or
+    validate a demo, so the UI must show an external provider/quota blocker
+    instead of a generic model/build failure.
+    """
+    raw = str(text or "")
+    low = raw.lower()
+    if not raw:
+        return {}
+    matched = next((marker for marker in PROVIDER_QUOTA_MARKERS if marker in low), "")
+    # HF provider errors often include a structured 403 and the exact spending
+    # limit phrase.  Keep this conservative so unrelated model errors still go
+    # through the normal repair path.
+    if not matched and not ("403" in low and "inference provider" in low and any(k in low for k in ("quota", "billing", "spending", "credits", "limit"))):
+        return {}
+    excerpt = redact_text(raw[-2200:])
+    return {
+        "schema_version": "provider_quota_blocker.v198_26_130",
+        "status": PROVIDER_QUOTA_BLOCKED_STATUS,
+        "category": "hf_inference_provider_quota",
+        "phase": phase,
+        "returncode": int(returncode) if returncode is not None else None,
+        "matched_marker": matched or "403 inference provider quota/billing marker",
+        "title": "Provider quota blocked",
+        "summary": "Hugging Face Inference Providers quota or monthly spending limit blocked Pi before ASF could generate the demo.",
+        "user_message": "ASF could not start the Pi coding pass because the Hugging Face Inference Providers quota or monthly spending limit was reached. No Space was uploaded and no model validation was attempted.",
+        "suggested_action": "Check Hugging Face billing/quota settings or switch the Pi provider/model, then retry the build.",
+        "retryable_after_user_action": True,
+        "model_failure": False,
+        "space_failure": False,
+        "space_upload_attempted": False,
+        "output_tail": excerpt,
+        "updated_at": now(),
+    }
+
+
+def write_provider_quota_blocker(run_dir: Path, events_path: Path, blocker: dict) -> dict:
+    payload = {**(blocker or {}), "updated_at": now()}
+    write_json(run_dir / "PROVIDER_QUOTA_BLOCKER.json", payload)
+    write_json(run_dir / "provider_quota_blocker.json", payload)
+    append_event(
+        events_path,
+        "provider_quota",
+        "blocked",
+        payload.get("user_message") or "Hugging Face Inference Providers quota blocked Pi before Space upload",
+        {k: payload.get(k) for k in ("category", "phase", "returncode", "matched_marker", "suggested_action")},
+    )
+    write_live_status(run_dir, stage="provider_quota", status=PROVIDER_QUOTA_BLOCKED_STATUS, message=payload.get("user_message") or payload.get("summary") or "Provider quota blocked", data=payload)
+    return payload
+
+
+def generated_space_creation_evidence(run_dir: Path, events_path: Path | None = None) -> dict:
+    """Return whether ASF has evidence that the target Space repo actually exists.
+
+    target_space_id may be planned before Pi runs. Cleanup/autopause must not act
+    on that planned id unless create/upload/runtime evidence proves the repo was
+    actually created or reused by this run.
+    """
+    evidence = {"space_created": False, "source": "none", "reason": "no_create_space_or_upload_evidence"}
+    try:
+        runtime_epoch = load_json_if_exists(run_dir / "runtime_upload_epoch.json") if (run_dir / "runtime_upload_epoch.json").exists() else {}
+        if isinstance(runtime_epoch, dict) and (runtime_epoch.get("last_upload_completed_at") or runtime_epoch.get("upload_sequence") or runtime_epoch.get("history")):
+            return {"space_created": True, "source": "runtime_upload_epoch", "reason": "runtime_upload_recorded"}
+    except Exception:
+        pass
+    for artifact, source in (("space_runtime.json", "space_runtime"), ("hardware_strategy.json", "hardware_strategy")):
+        try:
+            payload = load_json_if_exists(run_dir / artifact) if (run_dir / artifact).exists() else {}
+            if not isinstance(payload, dict) or not payload:
+                continue
+            if source == "space_runtime" and (payload.get("stage") or payload.get("runtime_status") or payload.get("status") or payload.get("url")):
+                return {"space_created": True, "source": source, "reason": "runtime_artifact_present"}
+            if source == "hardware_strategy" and payload.get("created") is True:
+                return {"space_created": True, "source": source, "reason": "hardware_strategy_created_true"}
+        except Exception:
+            pass
+    try:
+        path = events_path or (run_dir / "events.jsonl")
+        if path and path.exists():
+            for line in path.read_text(encoding="utf-8", errors="ignore").splitlines():
+                if not line.strip():
+                    continue
+                try:
+                    event = json.loads(line)
+                except Exception:
+                    continue
+                if not isinstance(event, dict):
+                    continue
+                step = str(event.get("step") or "").lower()
+                status = str(event.get("status") or "").lower()
+                data = event.get("data") if isinstance(event.get("data"), dict) else {}
+                if step == "create_space" and status in {"success", "warning"}:
+                    return {"space_created": True, "source": "events.create_space", "reason": status}
+                if step in {"upload_files", "repair_upload"} and status == "success":
+                    return {"space_created": True, "source": f"events.{step}", "reason": "upload_success"}
+                if data.get("created") is True or data.get("ok") is True and step == "create_space":
+                    return {"space_created": True, "source": "events.create_space.data", "reason": "created_or_ok_true"}
+    except Exception:
+        pass
+    return evidence
+
+
 def fail(run_dir: Path, events_path: Path, message: str, details: dict | None = None, status: str = "failed"):
     safe = safe_details(details)
-    append_event(events_path, "failure", "failed", message, safe)
+    quota_source = "\n".join([str(message or ""), str(safe.get("output_tail") or ""), str(safe.get("error") or "")])
+    quota_blocker = safe.get("provider_quota_blocker") if isinstance(safe.get("provider_quota_blocker"), dict) else classify_provider_quota_blocker(quota_source, phase=str(safe.get("phase") or safe.get("step") or "failure"), returncode=safe.get("returncode") if isinstance(safe.get("returncode"), int) else None)
+    if quota_blocker:
+        status = PROVIDER_QUOTA_BLOCKED_STATUS
+        safe["provider_quota_blocker"] = quota_blocker
+        safe.setdefault("failure_type", PROVIDER_QUOTA_BLOCKED_STATUS)
+        safe.setdefault("failure_owner", "hf_inference_provider")
+        message = quota_blocker.get("user_message") or message
+        try:
+            write_provider_quota_blocker(run_dir, events_path, quota_blocker)
+        except Exception:
+            pass
+    append_event(events_path, "failure", "blocked" if status == PROVIDER_QUOTA_BLOCKED_STATUS else "failed", message, safe)
     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 = {}
@@ -2969,11 +3189,21 @@ def fail(run_dir: Path, events_path: Path, message: str, details: dict | None =
         "details": safe,
         "repair_outcome": repair_outcome if isinstance(repair_outcome, dict) else {},
     }
-    # Preserve the target Space when the worker fails after repository creation.
+    # Preserve target Space links only when the worker has evidence that the repo
+    # was actually created/reused.  Before Pi finishes, TARGET_SPACE_ID is only a
+    # plan and cleanup/UI links must not treat it as a real Space.
     target_space = failure_state.get("target_space") or os.environ.get("TARGET_SPACE_ID") or ""
-    if target_space:
+    space_creation = generated_space_creation_evidence(run_dir, events_path)
+    if target_space and space_creation.get("space_created"):
         failure_state["target_space"] = target_space
         failure_state["target_space_url"] = f"https://huggingface.co/spaces/{target_space}"
+        failure_state["space_creation_evidence"] = space_creation
+    elif target_space:
+        failure_state.pop("target_space", None)
+        failure_state.pop("target_space_id", None)
+        failure_state.pop("target_space_url", None)
+        failure_state["target_space_planned"] = target_space
+        failure_state["space_creation_evidence"] = space_creation
     write_json(run_dir / "state.json", failure_state)
     cleanup_status = run_final_cleanup_if_needed(run_dir, events_path, failure_state)
     failure_state["cleanup"] = cleanup_status
@@ -9360,9 +9590,11 @@ def write_terminal_status_reconciliation(run_dir: Path, events_path: Path | None
     runtime_upload_epoch = load_json_if_exists(run_dir / "runtime_upload_epoch.json") if (run_dir / "runtime_upload_epoch.json").exists() else {}
     space_runtime = load_json_if_exists(run_dir / "space_runtime.json") if (run_dir / "space_runtime.json").exists() else {}
     runtime_history = runtime_upload_epoch.get("history") if isinstance(runtime_upload_epoch.get("history"), list) else []
-    target_space = state.get("target_space") or state.get("target_space_id") or runtime_upload_epoch.get("target_space_id") or os.environ.get("TARGET_SPACE_ID", "")
+    planned_target_space = state.get("target_space") or state.get("target_space_id") or runtime_upload_epoch.get("target_space_id") or os.environ.get("TARGET_SPACE_ID", "")
     runtime_uploaded = bool(runtime_upload_epoch.get("last_upload_completed_at") or runtime_upload_epoch.get("upload_sequence") or runtime_history)
     space_runtime_known = bool(isinstance(space_runtime, dict) and (space_runtime.get("stage") or space_runtime.get("status") or space_runtime.get("runtime_status") or space_runtime.get("updated_at") or space_runtime.get("url")))
+    space_creation = generated_space_creation_evidence(run_dir, events_path)
+    target_space = planned_target_space if space_creation.get("space_created") else ""
     payload = {
         "schema_version": "final_status_reconciliation.v198_26_9",
         "status": status,
@@ -9370,8 +9602,10 @@ def write_terminal_status_reconciliation(run_dir: Path, events_path: Path | None
         "message": message or state.get("message") or "",
         "job_exit_code": int(job_exit_code if job_exit_code is not None else (0 if status == "full_inference_success" else 1)),
         "target_space": target_space,
+        "target_space_planned": planned_target_space if not target_space else "",
         "target_space_url": f"https://huggingface.co/spaces/{target_space}" if target_space else "",
-        "space_created": bool(target_space and (runtime_uploaded or space_runtime_known)),
+        "space_created": bool(target_space and space_creation.get("space_created")),
+        "space_creation_evidence": space_creation,
         "runtime_uploaded": bool(runtime_uploaded),
         "space_uploaded": bool(runtime_uploaded),
         "space_runtime_known": bool(space_runtime_known),
@@ -15488,6 +15722,12 @@ def main():
         )
         (run_dir / "logs" / "pi_output.txt").write_text(pi_out, encoding="utf-8")
         if code != 0:
+            quota_blocker = classify_provider_quota_blocker(pi_out, phase="pi_run", returncode=code)
+            if quota_blocker:
+                append_event(events_path, "pi_run", "blocked", "Pi was blocked by Hugging Face Inference Providers quota/billing", {"returncode": code, "provider_quota_blocked": True, "output_tail": pi_out[-4000:]})
+                write_provider_quota_blocker(run_dir, events_path, quota_blocker)
+                collect_pi_traces(run_dir, events_path)
+                fail(run_dir, events_path, quota_blocker.get("user_message") or "Pi provider quota blocked before Space upload", {"returncode": code, "output_tail": pi_out[-4000:], "provider_quota_blocker": quota_blocker, "phase": "pi_run"}, status=PROVIDER_QUOTA_BLOCKED_STATUS)
             append_event(events_path, "pi_run", "failed", "Pi returned a non-zero exit code", {"returncode": code, "output_tail": pi_out[-4000:]})
             collect_pi_traces(run_dir, events_path)
             fail(run_dir, events_path, "Pi failed before Space upload", {"returncode": code, "output_tail": pi_out[-4000:]})
@@ -16688,6 +16928,8 @@ 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 == "provider_quota_blocked":
+        return "provider_quota_blocked"
     if status in {"technical_blocker", "technical_blocker_boot_only"}:
         return "technical_blocker"
     if full_inference_verified:
@@ -20307,7 +20549,7 @@ def smoke_generate(target_space_id: str, token: str, run_dir: Path, events_path:
     # write_json(run_dir / "tests" / "payload_source.json", payload_source_record)
     # write_json(run_dir / "tests" / "replay_source.json", replay_source)
     # write_json(run_dir / "tests" / "validation_preflight.json", validation_preflight)
-    app_version = "v198.26.127"
+    app_version = "v198.26.130"
     engine_version = "unified_gradio_validation_harness_v198_25_3"
     parent_build_run_id = os.environ.get("PARENT_BUILD_RUN_ID", "").strip()
     validation_mode = os.environ.get("VALIDATION_MODE") or os.environ.get("SPACE_TEST_POLICY_MODE") or "linked"