Upload 15 files

src/bucket.py

@@ -1384,6 +1384,8 @@ def read_run_bundle(run_id: str, *, bucket_source: str, token: str | None = None
         "model_analysis": _safe_read_json(f"{paths.root}/model_analysis.json", token=token),
         "pi_model_resolution": _safe_read_json(f"{paths.root}/pi_model_resolution.json", token=token) or (read_json(paths.state, token=token) or {}).get("pi_model_resolution") or {},
         "space_runtime": _safe_read_json(f"{paths.root}/space_runtime.json", token=token),
+        "runtime_upload_epoch": _safe_read_json(f"{paths.root}/runtime_upload_epoch.json", token=token),
+        "placeholder_scaffold_detection": _safe_read_json(f"{paths.root}/placeholder_scaffold_detection.json", token=token),
         "api_schema": _safe_read_json(f"{paths.root}/tests/api_schema.json", token=token) or _safe_read_json(f"{paths.root}/tests/generation_api_schema.json", token=token),
         "validation_payload": _safe_read_json(f"{paths.root}/tests/validation_payload.json", token=token),
         "post_build_validation_status": _safe_read_json(f"{paths.root}/post_build_validation_status.json", token=token),

src/effective_status.py

@@ -23,6 +23,7 @@ PARTIAL_BUILD_TOKENS = {
     "health_only",
     "full_inference_candidate_health_passed",
     "demo_usable_full_promise_not_verified",
+    "placeholder_scaffold_deployed",
     "interactive_app_available_smoke_failed",
     "manual_test_required_smoke_failed",
     "completed_with_warnings",
@@ -150,7 +151,7 @@ def _normalize_build_status(value: Any) -> str:
         return "stopped"
     if status in {"full_inference_candidate_health_passed", "health_only", "partial", "completed_with_warnings"}:
         return "partial_validation"
-    if status in {"demo_usable_full_promise_not_verified", "interactive_app_available_smoke_failed", "manual_test_required_smoke_failed"}:
+    if status in {"demo_usable_full_promise_not_verified", "placeholder_scaffold_deployed", "interactive_app_available_smoke_failed", "manual_test_required_smoke_failed"}:
         return status
     return status or "unknown"
 
@@ -211,6 +212,8 @@ def compute_effective_run_status(
         display_label = "Full inference success"
     elif display_status == "demo_usable_full_promise_not_verified":
         display_label = "Demo usable — full promise not verified"
+    elif display_status == "placeholder_scaffold_deployed":
+        display_label = "Placeholder demo — runtime not implemented"
     elif display_status == "interactive_app_available_smoke_failed":
         display_label = "Smoke input issue — Space reachable"
     elif display_status == "manual_test_required_smoke_failed":
@@ -229,7 +232,7 @@ def compute_effective_run_status(
         display_label = display_status.replace("_", " ").title() if display_status else "Unknown"
 
     return {
-        "schema_version": "effective_run_status.v198_26_3",
+        "schema_version": "effective_run_status.v198_26_5",
         "build_status": normalized_build,
         "build_verdict": normalized_verdict,
         "build_verdict_preserved": True,

src/version.py

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

src/view_models.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 from datetime import datetime, timedelta, timezone
+import re
 from typing import Any
 
 from .progress import STEP_ALIASES, STEP_LABELS, STEP_ORDER
@@ -391,28 +392,82 @@ def build_diagnostics(bundle: dict[str, Any], status_model: dict[str, Any], phas
     }
 
 
+def _normalize_target_space_id(value: Any) -> str:
+    text = str(value or "").strip()
+    if not text:
+        return ""
+    text = text.replace("https://huggingface.co/spaces/", "").strip("/")
+    text = text.split("/settings", 1)[0].strip("/")
+    if ".hf.space" in text and "/" not in text:
+        return ""
+    return text if re.match(r"^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$", text) else ""
+
+
+def _space_target_from_events(events: list[dict[str, Any]] | None, *, require_success: bool = False) -> str:
+    for event in reversed(events or []):
+        if not isinstance(event, dict):
+            continue
+        step = str(event.get("step") or "")
+        status = str(event.get("status") or "").lower()
+        if step not in {"create_space", "create_space_hardware", "upload_files", "runtime_upload_epoch"}:
+            continue
+        if require_success and status != "success":
+            continue
+        data = event.get("data") if isinstance(event.get("data"), dict) else {}
+        target = _normalize_target_space_id(data.get("target_space") or data.get("target_space_id"))
+        if target:
+            return target
+    return ""
+
+
 def _space_target_from_bundle(bundle: dict[str, Any]) -> str:
     summary = bundle.get("summary") or {}
     state = bundle.get("state") or {}
     launch = bundle.get("launch") or {}
-    return _first_nonempty(
+    runtime_upload_epoch = bundle.get("runtime_upload_epoch") or {}
+    identity = bundle.get("space_identity") or {}
+    return _normalize_target_space_id(_first_nonempty(
+        identity.get("target_space"),
         summary.get("target_space"),
         state.get("target_space"),
         launch.get("target_space"),
         summary.get("target_space_id"),
-    )
+        state.get("target_space_id"),
+        launch.get("target_space_id"),
+        runtime_upload_epoch.get("target_space_id"),
+        _space_target_from_events(bundle.get("events") or []),
+    ))
 
 
 def _space_target_url_from_bundle(bundle: dict[str, Any], target: str = "") -> str:
     summary = bundle.get("summary") or {}
     state = bundle.get("state") or {}
     launch = bundle.get("launch") or {}
-    explicit = _first_nonempty(summary.get("target_space_url"), state.get("target_space_url"), launch.get("target_space_url"))
+    identity = bundle.get("space_identity") or {}
+    explicit = _first_nonempty(identity.get("target_space_url"), summary.get("target_space_url"), state.get("target_space_url"), launch.get("target_space_url"))
     if explicit:
         return explicit
+    target = _normalize_target_space_id(target)
     return f"https://huggingface.co/spaces/{target}" if target else ""
 
 
+def _space_links_ready_from_bundle(bundle: dict[str, Any]) -> bool:
+    identity = bundle.get("space_identity") or {}
+    if identity.get("links_ready") or identity.get("space_created") or identity.get("space_uploaded"):
+        return True
+    runtime_upload_epoch = bundle.get("runtime_upload_epoch") or {}
+    if runtime_upload_epoch.get("last_upload_completed_at"):
+        return True
+    for event in bundle.get("events") or []:
+        if not isinstance(event, dict):
+            continue
+        step = str(event.get("step") or "")
+        status = str(event.get("status") or "").lower()
+        if status == "success" and step in {"create_space", "create_space_hardware", "upload_files", "runtime_upload_epoch"}:
+            return True
+    return False
+
+
 def _known_gradio_endpoint_info(bundle: dict[str, Any]) -> dict[str, Any]:
     smoke = bundle.get("generation_smoke") or {}
     gate = bundle.get("inference_gate") or {}
@@ -602,10 +657,12 @@ def build_run_view_model(
     if created:
         end = latest if status_model.get("is_terminal") and latest else current_now
         elapsed_seconds = int((end - created).total_seconds())
+    target = _space_target_from_bundle(bundle)
+    target_url = _space_target_url_from_bundle(bundle, target)
     links = {
         "job_url": _job_url_from_view_sources(bucket_source=bucket_source, summary=summary, state=state, launch=bundle.get("launch") or {}),
-        "target_space_url": summary.get("target_space_url") or "",
-        "target_space_settings_url": f"{summary.get('target_space_url')}/settings" if summary.get("target_space_url") else "",
+        "target_space_url": target_url,
+        "target_space_settings_url": f"{target_url}/settings" if target_url else "",
         "artifacts_url": summary.get("artifacts_url") or f"https://huggingface.co/buckets/{bucket_source}/tree/{_runs_prefix()}/{run_id}",
     }
     return {
@@ -619,8 +676,8 @@ def build_run_view_model(
             "display_status": effective_run_status.get("display_status") or status_model.get("global_status"),
             "display_label": effective_run_status.get("display_label") or status_model["global_status"].replace("_", " ").title(),
             "raw_status": status_model["raw_status"],
-            "space": summary.get("target_space") or "",
-            "space_url": summary.get("target_space_url") or "",
+            "space": target or summary.get("target_space") or "",
+            "space_url": target_url,
             "current_phase": phase,
             "current_phase_label": next((s["label"] for s in PRODUCT_STEPS if s["id"] == phase), phase),
             "elapsed_seconds": max(0, elapsed_seconds or 0),
@@ -643,7 +700,7 @@ def build_run_view_model(
         "actions": {
             "can_resume": status_model["global_status"] in {"running", "stale", "unknown"},
             "can_stop": status_model["global_status"] in {"running", "queued", "unknown"},
-            "can_open_space": bool(summary.get("target_space_url")),
+            "can_open_space": bool(target_url and _space_links_ready_from_bundle(bundle)),
             "can_validate": bool(space_test_policy.get("enabled")),
             "requires_manual_action": status_model["requires_manual_action"],
         },

src/worker_payload.py

@@ -35,8 +35,8 @@ DEFAULT_PREFERRED_SPACE_HARDWARE = "zero-a10g"
 DEFAULT_FALLBACK_SPACE_HARDWARE = "a10g-large"
 DEFAULT_MODEL_ID = "sshleifer/tiny-gpt2"
 MAX_PI_REPAIR_ATTEMPTS = 3
-APP_VERSION = "v198.26.3"
-app_version = "v198.26.3"
+APP_VERSION = "v198.26.5"
+app_version = "v198.26.5"
 
 # Internal agent/recovery files may be needed inside the transient Pi
 # workspace, but they should not be published to the generated Space or shown
@@ -1888,7 +1888,7 @@ def _gpu_partial_autopause_mode() -> str:
 def should_pause_generated_space(final_status: str | None, target_space: str | None, selected_hardware: str | None = None, *, keep_failed_spaces_running: bool | None = None, health_semantic_passed: bool | None = None) -> dict:
     """Decide whether a generated Space should be paused after a terminal outcome.
 
-    v198.26.3 keeps GPU partial cleanup non-aggressive by default: a reachable
+    v198.26.5 keeps GPU partial cleanup non-aggressive by default: a reachable
     partial can still be useful and should be validated via Space Test/minimal
     smoke before pausing.  `ASF_GPU_PARTIAL_AUTOPAUSE_MODE=strict` restores
     automatic pause for GPU partials; `recommended` emits a cost warning only.
@@ -2365,6 +2365,75 @@ def sanitize_model_id(model_id: str) -> str:
     return model_id
 
 
+
+
+def expected_space_subdomain(target_space_id: str) -> str:
+    owner, _, slug = str(target_space_id or "").partition("/")
+    if not owner or not slug:
+        return ""
+    return re.sub(r"[^a-z0-9-]+", "-", f"{owner}-{slug}".lower()).strip("-")
+
+
+def gradio_client_identity_payload(client, target_space_id: str) -> dict:
+    """Best-effort identity check for gradio_client target resolution.
+
+    v198.26.5: inspect every observed `.hf.space` URL independently.  A
+    correct repo id in `client.src` must not hide a mismatched app URL such as
+    `...-3a319105` expected but `...-2b5b161.hf.space` loaded by the client.
+    """
+    from urllib.parse import urlparse
+    observed_values = []
+    for attr in ("src", "space_id", "space_name", "app_url", "root_url", "src_url", "api_url"):
+        try:
+            value = getattr(client, attr, None)
+        except Exception:
+            value = None
+        if value:
+            observed_values.append({"attr": attr, "value": str(value)})
+    expected_repo = str(target_space_id or "")
+    expected_subdomain = expected_space_subdomain(expected_repo)
+    observed_text = " ".join(item["value"] for item in observed_values).lower()
+    observed_hf_space_urls = []
+    mismatched_hf_space_urls = []
+    for item in observed_values:
+        value = item["value"]
+        if ".hf.space" not in value:
+            continue
+        parsed = urlparse(value if value.startswith(("http://", "https://")) else "https://" + value.lstrip("/"))
+        host = (parsed.netloc or parsed.path.split("/", 1)[0]).lower()
+        if not host.endswith(".hf.space"):
+            continue
+        subdomain = host.rsplit(".hf.space", 1)[0]
+        record = {"attr": item["attr"], "value": value, "host": host, "subdomain": subdomain}
+        observed_hf_space_urls.append(record)
+        if expected_subdomain and subdomain != expected_subdomain:
+            mismatched_hf_space_urls.append(record)
+    mismatch = False
+    if mismatched_hf_space_urls:
+        mismatch = True
+    elif observed_text and expected_repo and expected_repo.lower() not in observed_text and not observed_hf_space_urls:
+        mismatch = True
+    return {
+        "schema_version": "gradio_client_identity.v198_26_5",
+        "target_space_id": expected_repo,
+        "expected_subdomain": expected_subdomain,
+        "observed": observed_values,
+        "observed_hf_space_urls": observed_hf_space_urls,
+        "mismatched_hf_space_urls": mismatched_hf_space_urls,
+        "mismatch": bool(mismatch),
+        "failure_owner": "factory_validation_client" if mismatch else "",
+        "failure_class": "space_identity_mismatch" if mismatch else "",
+    }
+
+
+def write_gradio_client_identity(client, target_space_id: str, run_dir: Path, events_path: Path, *, phase: str) -> dict:
+    payload = gradio_client_identity_payload(client, target_space_id)
+    payload["phase"] = phase
+    write_json(run_dir / "tests" / f"gradio_client_identity_{phase}.json", payload)
+    if payload.get("mismatch"):
+        append_event(events_path, "space_identity", "failed", "Gradio client resolved a different Space than the run target", payload)
+    return payload
+
 def make_gradio_client(target_space_id: str, token: str, timeout_s: int | float | None = None):
     import inspect
     from gradio_client import Client
@@ -2920,7 +2989,7 @@ def _smoke_param_name(param: dict) -> str:
 def _minimal_numeric_value(name: str, value, *, expected_output_type: str = ""):
     """Return a safer minimal smoke value for expensive generation parameters.
 
-    v198.26.3 introduces a second validation level: a minimal demo smoke can
+    v198.26.5 introduces a second validation level: a minimal demo smoke can
     prove the Space is usable after a canonical/promise smoke OOM, but it must
     not be promoted to full inference success.
     """
@@ -2986,7 +3055,7 @@ def build_minimal_demo_smoke_args(args: list, params: list[dict], expected_outpu
 def minimal_smoke_status_payload(payload: dict) -> dict:
     """Mark a successful minimal smoke without claiming full promise success."""
     out = dict(payload or {})
-    out["schema_version"] = "generation_smoke_result.v198_26_3"
+    out["schema_version"] = "generation_smoke_result.v198_26_5"
     out["status"] = "demo_usable_smoke_passed"
     out["demo_usable_smoke_passed"] = True
     out["canonical_promise_smoke_passed"] = False
@@ -3222,6 +3291,106 @@ def write_contract_skipped_generation_smoke(run_dir: Path, events_path: Path, ex
     return payload
 
 
+
+
+def _param_semantic_name(param: dict, index: int, expected_output_type: str = "") -> str:
+    """Best-effort semantic name for anonymous Gradio `param_N` schemas.
+
+    v198.26.5: gradio_client can expose Textbox/Number inputs as param_0,
+    param_1... while Pi's canonical smoke example uses semantic keys like
+    prompt/width/height/steps/seed.  A positional anonymous fallback prevents a
+    valid prompt from being replaced by the schema-generated empty string.
+    """
+    raw = str((param or {}).get("name") or "").strip().lower().replace("-", "_").replace(" ", "_")
+    component = str((param or {}).get("component") or "").strip().lower()
+    if raw and not re.fullmatch(r"param_\d+|arg\d+", raw):
+        return raw
+    expected = str(expected_output_type or "").strip().lower()
+    if index == 0 and any(token in component for token in ("textbox", "text", "str")):
+        return "prompt"
+    if index == 0 and expected in {"text", "audio", "video", "image"}:
+        return "prompt"
+    common = ["prompt", "width", "height", "num_inference_steps", "seed", "guidance_scale"]
+    if expected == "audio":
+        common = ["prompt", "duration", "steps", "seed", "guidance_scale"]
+    if expected == "video":
+        common = ["prompt", "width", "height", "num_frames", "num_inference_steps", "seed"]
+    if index < len(common):
+        return common[index]
+    return raw or f"arg{index}"
+
+
+def _canonical_key_aliases(name: str) -> list[str]:
+    name = str(name or "").strip().lower().replace("-", "_").replace(" ", "_")
+    aliases = {
+        "prompt": ["prompt", "text", "query", "input", "instruction", "caption"],
+        "negative_prompt": ["negative_prompt", "negative", "negative_text"],
+        "width": ["width", "image_width", "w"],
+        "height": ["height", "image_height", "h"],
+        "num_inference_steps": ["num_inference_steps", "inference_steps", "steps", "num_steps", "sampling_steps"],
+        "steps": ["steps", "num_inference_steps", "inference_steps", "num_steps", "sampling_steps"],
+        "seed": ["seed", "random_seed"],
+        "guidance_scale": ["guidance_scale", "cfg_scale", "scale", "guidance"],
+        "duration": ["duration", "seconds", "length", "audio_length"],
+        "num_frames": ["num_frames", "frames", "frame_count"],
+    }
+    return aliases.get(name, [name, name.replace("_", " "), name.replace("_", "-")])
+
+
+def _canonical_dict_value_for_param(raw_args: dict, param: dict, index: int, generated: list, expected_output_type: str) -> tuple[object, str]:
+    semantic = _param_semantic_name(param, index, expected_output_type)
+    raw_lower = {str(k).strip().lower().replace("-", "_").replace(" ", "_"): k for k in raw_args.keys()}
+    for candidate in _canonical_key_aliases(semantic):
+        key = raw_lower.get(candidate.replace("-", "_").replace(" ", "_"))
+        if key is not None:
+            return raw_args[key], f"semantic:{candidate}"
+    name = str((param or {}).get("name") or f"arg{index}")
+    for candidate in (name, name.replace("_", " "), name.replace("_", "-"), name.lower(), name.lower().replace("_", " "), name.lower().replace("_", "-")):
+        if candidate in raw_args:
+            return raw_args[candidate], f"schema_name:{candidate}"
+    values = list(raw_args.values())
+    if index < len(values) and re.fullmatch(r"param_\d+|arg\d+", str(name).strip().lower()):
+        return values[index], "anonymous_positional_dict_order"
+    return (generated[index] if index < len(generated) else smoke_value_for_parameter(param, expected_output_type)), "schema_generated_fallback"
+
+
+def detect_required_text_payload_resolution_issue(args: list, params: list[dict], canonical_example: dict | None = None) -> dict:
+    """Detect when ASF resolved a required text prompt to empty before predict()."""
+    non_empty_canonical = []
+    if isinstance(canonical_example, dict):
+        raw_inputs, _ = _raw_inputs_from_canonical_smoke(canonical_example)
+        if isinstance(raw_inputs, dict):
+            for key, value in raw_inputs.items():
+                if isinstance(value, str) and value.strip():
+                    non_empty_canonical.append({"key": str(key), "value_preview": value[:120]})
+        elif isinstance(raw_inputs, list):
+            for i, value in enumerate(raw_inputs):
+                if isinstance(value, str) and value.strip():
+                    non_empty_canonical.append({"index": i, "value_preview": value[:120]})
+    for index, param in enumerate(params or []):
+        if index >= len(args or []):
+            continue
+        name = str((param or {}).get("name") or f"arg{index}").strip()
+        semantic = _param_semantic_name(param, index)
+        component = str((param or {}).get("component") or "").lower()
+        required = bool((param or {}).get("required", False))
+        value = args[index]
+        is_text = any(token in component for token in ("textbox", "text", "str")) or semantic in {"prompt", "text", "query", "instruction"}
+        if required and is_text and isinstance(value, str) and not value.strip() and non_empty_canonical:
+            return {
+                "schema_version": "smoke_payload_resolution_guard.v198_26_5",
+                "detected": True,
+                "failure_owner": "factory_validation_client",
+                "failure_class": "smoke_payload_resolution_failed",
+                "failure_type": "required_text_resolved_empty",
+                "parameter_index": index,
+                "parameter_name": name,
+                "semantic_name": semantic,
+                "canonical_non_empty_inputs": non_empty_canonical[:5],
+                "recommended_action": "Fix canonical smoke dict-to-args resolution before calling gradio_client.predict; do not call the Space with an empty required prompt.",
+            }
+    return {"schema_version": "smoke_payload_resolution_guard.v198_26_5", "detected": False}
+
 def _canonical_smoke_dict_from_contracts(inference_contract: dict | None, demo_quality_contract: dict | None) -> tuple[dict | None, str]:
     """Return the preferred canonical smoke example and its source.
 
@@ -3282,24 +3451,16 @@ def canonical_smoke_example_payload(inference_contract: dict, demo_quality_contr
 
     if isinstance(raw_args, dict):
         args = []
+        resolution = []
         for i, param in enumerate(params):
-            name = str(param.get("name") or f"arg{i}")
-            candidates = [
-                name,
-                name.replace("_", " "),
-                name.replace("_", "-"),
-                name.lower(),
-                name.lower().replace("_", " "),
-                name.lower().replace("_", "-"),
-            ]
-            found = False
-            for candidate in candidates:
-                if candidate in raw_args:
-                    args.append(coerce_smoke_value(raw_args[candidate], param))
-                    found = True
-                    break
-            if not found:
-                args.append(generated[i] if i < len(generated) else smoke_value_for_parameter(param, expected_output_type))
+            value, source_key = _canonical_dict_value_for_param(raw_args, param, i, generated, expected_output_type)
+            args.append(coerce_smoke_value(value, param))
+            resolution.append({
+                "index": i,
+                "name": str(param.get("name") or f"arg{i}"),
+                "semantic_name": _param_semantic_name(param, i, expected_output_type),
+                "source": source_key,
+            })
         if not params:
             args = list(raw_args.values())
         return {
@@ -3310,6 +3471,7 @@ def canonical_smoke_example_payload(inference_contract: dict, demo_quality_contr
             "source": source,
             "canonical_smoke_example_present": True,
             "canonical_smoke_reason": smoke.get("reason") or smoke.get("why_representative") or smoke.get("description") or "",
+            "dict_resolution": resolution,
         }
 
     # Empty/unsupported inputs: canonical exists but cannot be transformed safely.
@@ -3369,6 +3531,7 @@ def validation_health_semantics(validation: dict | None) -> dict:
 
     negative_statuses = {"unhealthy", "failed", "failure", "error", "not_ready", "not ready", "runtime_error"}
     positive_statuses = {"healthy", "ok", "ready", "success", "passed"}
+    scaffold_markers = {"initial-scaffold", "initial_scaffold", "placeholder", "scaffold", "template"}
     for payload in _semantic_health_payloads(validation):
         for key in ("status", "health", "state"):
             if key in payload:
@@ -3379,6 +3542,12 @@ def validation_health_semantics(validation: dict | None) -> dict:
                 elif value in positive_statuses and key != "state":
                     details["semantic_checked"] = True
                     details["positive_markers"].append(f"{key}={value}")
+        for key in ("stage", "runtime_stage", "phase"):
+            if key in payload:
+                value = str(payload.get(key) or "").strip().lower()
+                if value in scaffold_markers or any(marker in value for marker in ("initial-scaffold", "initial_scaffold", "placeholder")):
+                    details["semantic_checked"] = True
+                    details["negative_markers"].append(f"{key}={value}")
         for key in ("pipeline_ready", "pipeline_loaded", "model_ready", "model_loaded"):
             if key in payload:
                 details["semantic_checked"] = True
@@ -3428,6 +3597,45 @@ def is_cuda_oom_text(text: str) -> bool:
     return ("cuda" in lowered or "gpu" in lowered or "outofmemory" in lowered) and any(marker in lowered for marker in markers)
 
 
+
+
+def is_placeholder_scaffold_text(text: str) -> bool:
+    lowered = str(text or "").lower()
+    markers = [
+        "initial scaffold",
+        "initial-scaffold",
+        "initial_scaffold",
+        "pi should replace this",
+        "model-specific inference path",
+        "placeholder demo",
+        "placeholder scaffold",
+        "stage: initial-scaffold",
+    ]
+    return any(marker in lowered for marker in markers)
+
+
+def detect_placeholder_scaffold(workspace: Path) -> dict:
+    """Detect a final runtime that still contains ASF/Pi placeholder scaffold text."""
+    app_path = workspace / "app.py"
+    text = app_path.read_text(encoding="utf-8", errors="ignore") if app_path.exists() else ""
+    detected = is_placeholder_scaffold_text(text)
+    markers = []
+    lowered = text.lower()
+    for marker in ["initial scaffold", "initial-scaffold", "pi should replace this", "model-specific inference path", "placeholder scaffold"]:
+        if marker in lowered:
+            markers.append(marker)
+    return {
+        "schema_version": "placeholder_scaffold_detection.v198_26_5",
+        "detected": bool(detected),
+        "failure_owner": "pi_generation" if detected else "",
+        "failure_class": "placeholder_scaffold_deployed" if detected else "",
+        "failure_type": "model_specific_runtime_missing" if detected else "",
+        "repair_candidate": bool(detected),
+        "markers": markers,
+        "checked_file": "app.py",
+        "recommended_action": "Replace the initial scaffold with a model-specific inference app before treating the Space as healthy." if detected else "",
+    }
+
 def _extract_missing_executable(text: str) -> str:
     text = str(text or "")
     patterns = [
@@ -3481,7 +3689,25 @@ def diagnose_generation_smoke_failure(error_or_result, *, inference_strategy: st
     marker = "no value provided for required argument:"
     missing_executable = _extract_missing_executable(text)
 
-    if "asf smoke input materialization failed" in lowered or "does not exist on local filesystem" in lowered:
+    if "smoke_payload_resolution_failed" in lowered or "required_text_resolved_empty" in lowered or "prompt cannot be empty" in lowered:
+        base.update({
+            "failure_owner": "factory_validation_client",
+            "failure_class": "smoke_payload_resolution_failed",
+            "failure_type": "required_text_resolved_empty",
+            "actionability": "factory_fix_required",
+            "repair_candidate": False,
+            "recommended_action": "Fix ASF canonical smoke payload resolution so required prompt/text inputs are non-empty before calling gradio_client; do not repair the generated Space.",
+        })
+    elif is_placeholder_scaffold_text(text):
+        base.update({
+            "failure_owner": "pi_generation",
+            "failure_class": "placeholder_scaffold_deployed",
+            "failure_type": "model_specific_runtime_missing",
+            "actionability": "targeted_pi_repair",
+            "repair_candidate": True,
+            "recommended_action": "Run a targeted Pi repair to replace the initial scaffold with a model-specific inference app that returns the promised artifact.",
+        })
+    elif "asf smoke input materialization failed" in lowered or "does not exist on local filesystem" in lowered:
         base.update({
             "failure_owner": "factory_validation_client",
             "failure_class": "smoke_input_materialization_failed",
@@ -3905,6 +4131,14 @@ def run_generation_smoke(target_space_id: str, token: str, run_dir: Path, events
     timeout_s = smoke_timeout_seconds(expected_output_type)
     append_event(events_path, "generation_smoke", "started", "Calling live generation endpoint for ZeroGPU timing", {"api_name": api_name, "expected_output_type": expected_output_type, "smoke_timeout_seconds": timeout_s})
     client = make_gradio_client(target_space_id, token, timeout_s=timeout_s)
+    identity = write_gradio_client_identity(client, target_space_id, run_dir, events_path, phase="generation_smoke")
+    if identity.get("mismatch"):
+        diagnosis = {"schema_version": "generation_smoke_diagnosis.v198_26_5", "smoke_passed": False, "phase": "space_identity", "expected_output_type": expected_output_type or "", "failure_owner": "factory_validation_client", "failure_class": "space_identity_mismatch", "failure_type": "space_identity_mismatch", "actionability": "factory_fix_required", "repair_candidate": False, "recommended_action": "Do not validate a Space whose gradio_client URL does not match the target_space_id; fix Space identity propagation before retrying.", "evidence": [json.dumps(identity, ensure_ascii=False)[:1200]]}
+        payload = {"status": "failed", "target_space": target_space_id, "api_name": api_name, "expected_output_type": expected_output_type, "error": "space_identity_mismatch", "failure_type": "space_identity_mismatch", "failure_class": "space_identity_mismatch", "failure_owner": "factory_validation_client", "actionability": "factory_fix_required", "repair_candidate": False, "space_identity": identity, **measured_zero_gpu_recommendation(None)}
+        write_json(run_dir / "tests" / "generation_smoke_diagnosis.json", diagnosis)
+        record_generation_smoke_result(run_dir, payload, phase="generation_smoke")
+        append_event(events_path, "generation_smoke", "failed", "Space identity mismatch before generation smoke", payload)
+        return payload
     schema = client.view_api(return_format="dict")
     discovered = api_names_from_schema(schema)
     if api_name not in discovered and discovered:
@@ -3939,14 +4173,67 @@ def run_generation_smoke(target_space_id: str, token: str, run_dir: Path, events
     if contract_payload and contract_payload.get("canonical_smoke_example_present"):
         canonical_example, canonical_source = _canonical_smoke_dict_from_contracts(contract, demo_quality_contract)
     smoke_payload = {"api_name": api_name, "test_args": test_args, "raw_test_args": raw_test_args, "test_kwargs": test_kwargs, "parameters": smoke_parameters, "source": smoke_source, "contract_present": bool(contract), "demo_quality_contract_present": bool(demo_quality_contract), "canonical_smoke_example_present": bool(canonical_example), "canonical_smoke_example_source": canonical_source, "choice_corrections": initial_choice_changes, "file_input_conversions": file_input_conversions, "smoke_timeout_seconds": timeout_s}
+    payload_resolution_guard = detect_required_text_payload_resolution_issue(test_args, smoke_parameters, canonical_example)
+    smoke_payload["payload_resolution_guard"] = payload_resolution_guard
     if canonical_example:
-        write_json(run_dir / "tests" / "canonical_smoke_example.json", {"source": canonical_source, "example": canonical_example, "resolved_api_name": api_name, "raw_resolved_args": raw_test_args, "resolved_args": test_args, "resolved_kwargs": test_kwargs, "expected_output_type": expected_output_type, "file_input_conversions": file_input_conversions})
+        write_json(run_dir / "tests" / "canonical_smoke_example.json", {"source": canonical_source, "example": canonical_example, "resolved_api_name": api_name, "raw_resolved_args": raw_test_args, "resolved_args": test_args, "resolved_kwargs": test_kwargs, "expected_output_type": expected_output_type, "file_input_conversions": file_input_conversions, "dict_resolution": contract_payload.get("dict_resolution") if isinstance(contract_payload, dict) else [], "payload_resolution_guard": payload_resolution_guard})
     write_json(run_dir / "tests" / "generation_smoke_payload.json", smoke_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": smoke_source, "selected_api_name": api_name, "parent_smoke_payload_used": False, "canonical_smoke_example_used": bool(canonical_example), "canonical_smoke_example_source": canonical_source})
     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": True})
-    write_json(run_dir / "tests" / "resolved_validation_request.json", {"api_name": api_name, "test_args": test_args, "raw_test_args": raw_test_args, "test_kwargs": test_kwargs, "expected_output_type": expected_output_type, "validation_mode": "automatic_smoke", "payload_source": smoke_source, "canonical_smoke_example_used": bool(canonical_example), "canonical_smoke_example_source": canonical_source, "schema_choice_corrections": initial_choice_changes, "file_input_conversions": file_input_conversions, "smoke_timeout_seconds": timeout_s})
-    write_json(run_dir / "tests" / "schema_coercion.json", {"api_name": api_name, "changes": initial_choice_changes, "original_args": raw_test_args, "resolved_args": test_args, "file_input_conversions": file_input_conversions})
+    write_json(run_dir / "tests" / "resolved_validation_request.json", {"api_name": api_name, "test_args": test_args, "raw_test_args": raw_test_args, "test_kwargs": test_kwargs, "expected_output_type": expected_output_type, "validation_mode": "automatic_smoke", "payload_source": smoke_source, "canonical_smoke_example_used": bool(canonical_example), "canonical_smoke_example_source": canonical_source, "schema_choice_corrections": initial_choice_changes, "file_input_conversions": file_input_conversions, "smoke_timeout_seconds": timeout_s, "payload_resolution_guard": payload_resolution_guard})
+    write_json(run_dir / "tests" / "schema_coercion.json", {"api_name": api_name, "changes": initial_choice_changes, "original_args": raw_test_args, "resolved_args": test_args, "file_input_conversions": file_input_conversions, "payload_resolution_guard": payload_resolution_guard})
     write_json(run_dir / "tests" / "generation_api_schema.json", {"schema": schema, "api_names": discovered, "selected_api_name": api_name, "endpoint_parameters": endpoint_parameters, "smoke_parameters": smoke_parameters, "smoke_payload_source": smoke_source, "canonical_smoke_example_used": bool(canonical_example)})
+    if payload_resolution_guard.get("detected"):
+        diagnosis = {
+            "schema_version": "generation_smoke_diagnosis.v198_26_5",
+            "smoke_passed": False,
+            "phase": "payload_resolution",
+            "inference_strategy": str(contract.get("inference_strategy") or ""),
+            "expected_output_type": expected_output_type or "",
+            "failure_owner": "factory_validation_client",
+            "failure_class": "smoke_payload_resolution_failed",
+            "failure_type": "required_text_resolved_empty",
+            "actionability": "factory_fix_required",
+            "repair_candidate": False,
+            "recommended_action": payload_resolution_guard.get("recommended_action"),
+            "evidence": [json.dumps(payload_resolution_guard, ensure_ascii=False)[:1200]],
+        }
+        payload = {
+            "status": "failed",
+            "target_space": target_space_id,
+            "api_name": api_name,
+            "discovered_api_names": discovered,
+            "endpoint_parameters": endpoint_parameters,
+            "test_args": test_args,
+            "test_kwargs": test_kwargs,
+            "smoke_payload_source": smoke_source,
+            "validation_level": "canonical_promise_smoke",
+            "demo_usable_smoke_passed": False,
+            "canonical_promise_smoke_passed": False,
+            "full_promise_smoke_passed": False,
+            "canonical_smoke_example_used": bool(canonical_example),
+            "canonical_smoke_example_source": canonical_source,
+            "payload_resolution_guard": payload_resolution_guard,
+            "file_input_conversions": file_input_conversions,
+            "smoke_timeout_seconds": timeout_s,
+            "next_action": diagnosis["recommended_action"],
+            "expected_output_type": expected_output_type,
+            "latency_seconds": None,
+            "result_info": {},
+            "copied_artifacts": [],
+            "validated_at": now(),
+            "failure_type": diagnosis["failure_type"],
+            "failure_class": diagnosis["failure_class"],
+            "failure_owner": diagnosis["failure_owner"],
+            "actionability": diagnosis["actionability"],
+            "repair_candidate": False,
+            **measured_zero_gpu_recommendation(None),
+        }
+        write_json(run_dir / "tests" / "generation_smoke_diagnosis.json", diagnosis)
+        record_generation_smoke_result(run_dir, payload, phase="generation_smoke")
+        write_live_status(run_dir, stage="generation_smoke", status="failed", message="Automatic smoke payload resolution failed before calling Gradio", data=payload)
+        append_event(events_path, "generation_smoke", "failed", "Automatic smoke payload resolution failed before calling Gradio", payload)
+        return payload
     write_live_status(run_dir, stage="generation_smoke", status="running", message="Calling live generation endpoint", data={"api_name": api_name, "source": smoke_source, "canonical_smoke_example_used": bool(canonical_example)})
     started = time.time()
     retry_info = {"attempts": 1, "choice_corrections": initial_choice_changes, "retried": False}
@@ -3994,7 +4281,7 @@ def run_generation_smoke(target_space_id: str, token: str, run_dir: Path, events
             retry_info = {"attempts": 2, "choice_corrections": initial_choice_changes, "file_input_conversions": file_input_conversions, "retried": True, "first_error": str(first_error)[:2000], "retry_reason": "canonical_cuda_oom_minimal_demo_smoke", "minimal_changes": minimal_changes}
             minimal_artifact = {"api_name": api_name, "test_args": minimal_args, "raw_test_args": minimal_raw_args, "test_kwargs": test_kwargs, "parameters": minimal_params, "validation_level": "minimal_demo_smoke", **retry_info}
             write_json(run_dir / "tests" / "generation_smoke_payload_minimal.json", minimal_artifact)
-            write_json(run_dir / "tests" / "minimal_demo_smoke_retry.json", {"schema_version": "minimal_demo_smoke_retry.v198_26_3", "triggered": True, "failure_owner": "hardware", "failure_type": "canonical_cuda_oom", "canonical_error": str(first_error)[:2000], "minimal_changes": minimal_changes, "retry_payload_path": "tests/generation_smoke_payload_minimal.json"})
+            write_json(run_dir / "tests" / "minimal_demo_smoke_retry.json", {"schema_version": "minimal_demo_smoke_retry.v198_26_5", "triggered": True, "failure_owner": "hardware", "failure_type": "canonical_cuda_oom", "canonical_error": str(first_error)[:2000], "minimal_changes": minimal_changes, "retry_payload_path": "tests/generation_smoke_payload_minimal.json"})
             try:
                 result = client.predict(*minimal_args, api_name=api_name, **test_kwargs)
                 raw_test_args = minimal_raw_args
@@ -4657,6 +4944,9 @@ def validate_http_health(target_space_id: str, token: str, run_dir: Path, events
 
 def validate_gradio_api(target_space_id: str, token: str, run_dir: Path, events_path: Path, attempt: int):
     client = make_gradio_client(target_space_id, token)
+    identity = write_gradio_client_identity(client, target_space_id, run_dir, events_path, phase="api_validation")
+    if identity.get("mismatch"):
+        raise RuntimeError(f"space_identity_mismatch: expected {target_space_id}, observed {identity.get('observed')}")
     schema = client.view_api(return_format="dict")
     write_json(run_dir / "tests" / "api_schema.json", schema if isinstance(schema, dict) else {"schema": str(schema)})
     discovered = api_names_from_schema(schema)
@@ -9278,6 +9568,10 @@ def upload_workspace(api, workspace: Path, target_space_id: str, token: str, run
     if gen_dir.exists():
         shutil.rmtree(gen_dir)
     shutil.copytree(workspace, gen_dir, ignore=internal_workspace_copy_ignore)
+    scaffold_detection = detect_placeholder_scaffold(workspace)
+    write_json(run_dir / "placeholder_scaffold_detection.json", scaffold_detection)
+    if scaffold_detection.get("detected"):
+        append_event(events_path, "placeholder_scaffold", "warning", "Generated runtime still contains initial scaffold markers", scaffold_detection)
     payload_dir, manifest = build_runtime_upload_payload(workspace, run_dir, events_path)
     try:
         api.upload_folder(
@@ -9329,6 +9623,7 @@ def infer_generation_gate(workspace: Path, implementation_mode: str, validation:
     req_text = (workspace / "requirements.txt").read_text(encoding="utf-8", errors="ignore") if (workspace / "requirements.txt").exists() else ""
     blockers_path = workspace / "TECHNICAL_BLOCKERS.json"
     blockers = load_json_if_exists(blockers_path)
+    scaffold_detection = detect_placeholder_scaffold(workspace)
 
     combined = (app_text + "\n" + summary_text).lower()
     blocked_markers = [
@@ -9359,11 +9654,12 @@ def infer_generation_gate(workspace: Path, implementation_mode: str, validation:
     full_inference_requested = implementation_mode in {"full-inference-gated", "full-inference-attempt"}
     promise_fulfilled = bool(promise_validation.get("promise_fulfilled"))
     heuristic_marker_detected = any(m in combined for m in blocked_markers)
+    scaffold_detected = bool(scaffold_detection.get("detected"))
     blocker_source = str(blockers.get("source") or "") if isinstance(blockers, dict) else ""
     heuristic_blocker_json = bool(blockers) and blocker_source == "worker_heuristic_from_PI_SUMMARY_or_app.py"
     heuristic_blocker_detected = bool(heuristic_marker_detected or heuristic_blocker_json)
     contract_blocker_detected = bool(contract_no_full or diagnostic_contract)
-    hard_blocker_detected = bool((bool(blockers) and not heuristic_blocker_json) or contract_blocker_detected)
+    hard_blocker_detected = bool((bool(blockers) and not heuristic_blocker_json) or contract_blocker_detected or scaffold_detected)
     blocker_detected = bool(hard_blocker_detected or heuristic_blocker_detected)
     strong_full_inference_success = bool(full_inference_requested and smoke_ok and promise_fulfilled)
     recommendation = generation_smoke if isinstance(generation_smoke, dict) else measured_zero_gpu_recommendation(None)
@@ -9379,6 +9675,7 @@ def infer_generation_gate(workspace: Path, implementation_mode: str, validation:
         "health_semantic_passed": health_semantics.get("semantic_passed"),
         "health_semantic_negative_markers": health_semantics.get("negative_markers") or [],
         "health_load_error": health_semantics.get("load_error") or "",
+        "placeholder_scaffold_detected": scaffold_detected,
         "generation_smoke_passed": smoke_ok,
         "demo_usable_smoke_passed": minimal_smoke_ok,
         "canonical_promise_smoke_passed": smoke_ok,
@@ -9393,10 +9690,17 @@ def infer_generation_gate(workspace: Path, implementation_mode: str, validation:
             "source": "TECHNICAL_BLOCKERS.json" if heuristic_blocker_json else "PI_SUMMARY.md_or_app.py",
             "non_blocking": True,
         })
+    if scaffold_detected:
+        blocking_evidence.append({
+            "type": "placeholder_scaffold_deployed",
+            "source": "app.py",
+            "overridable_by_live_smoke": True,
+            "details": scaffold_detection,
+        })
     if hard_blocker_detected:
         blocking_evidence.append({
             "type": "contract_or_explicit_blocker",
-            "source": "INFERENCE_CONTRACT.json_or_TECHNICAL_BLOCKERS.json",
+            "source": "INFERENCE_CONTRACT.json_or_TECHNICAL_BLOCKERS.json_or_app.py",
             "overridable_by_live_smoke": True,
         })
 
@@ -9421,6 +9725,9 @@ def infer_generation_gate(workspace: Path, implementation_mode: str, validation:
                 message = "Generation smoke was skipped by contract because full inference was declared unavailable; this is not a full inference success."
             else:
                 message = "Space boots/responds as a diagnostic app, but real/full model inference is not implemented; this is not a full inference success."
+        elif scaffold_detected:
+            status = "placeholder_scaffold_deployed"
+            message = "Space boots, but the published app still contains the initial scaffold instead of model-specific inference."
         elif hard_blocker_detected:
             status = "technical_blocker"
             message = "Space boots, but full model inference was not implemented. See TECHNICAL_BLOCKERS.json / PI_SUMMARY.md."
@@ -9518,6 +9825,7 @@ def infer_generation_gate(workspace: Path, implementation_mode: str, validation:
             "recommendation_confidence": recommendation.get("recommendation_confidence"),
             "measurement_note": recommendation.get("measurement_note"),
         },
+        "placeholder_scaffold_detection": scaffold_detection,
         "blockers": blockers,
     }
     write_json(run_dir / "inference_gate.json", gate)
@@ -12096,7 +12404,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.3"
+    app_version = "v198.26.5"
     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"
@@ -12481,7 +12789,7 @@ def main():
             effective_status_on_success = os.environ.get("EFFECTIVE_STATUS_ON_SUCCESS") or "validated_after_space_test"
             space_test_policy_mode = os.environ.get("SPACE_TEST_POLICY_MODE") or "complete"
             manual_status = {
-                "schema_version": "post_build_validation.v198_26_3",
+                "schema_version": "post_build_validation.v198_26_5",
                 "status": "success",
                 "effective_status": effective_status_on_success,
                 "legacy_effective_status": "validated_after_manual_space_test",
@@ -12559,7 +12867,7 @@ Target Space: [`{target_space_id}`](https://huggingface.co/spaces/{target_space_
         if parent_build_run_id:
             parent_dir = output_root / "runs" / parent_build_run_id
             parent_dir.mkdir(parents=True, exist_ok=True)
-            failed_status = {"schema_version": "post_build_validation.v198_26_3", "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"}
+            failed_status = {"schema_version": "post_build_validation.v198_26_5", "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 []