Upload 9 files

src/bucket.py

@@ -335,6 +335,87 @@ def _list_run_files(
     return files
 
 
+def _normalize_model_name(value: str | None) -> str:
+    return re.sub(r"[^a-z0-9]+", "", (value or "").lower())
+
+
+def _extract_pi_models_from_text(text: str) -> list[str]:
+    if not text:
+        return []
+    patterns = [
+        r"\bQwen/[A-Za-z0-9_.-]+",
+        r"\bQwen(?:2(?:\.5)?|3)?[-_/A-Za-z0-9.]*Coder[-_/A-Za-z0-9.]*",
+        r"\bKimi[-_/A-Za-z0-9.]+",
+        r"\bMoonshotAI/[A-Za-z0-9_.-]+",
+        r"\bClaude[-_/A-Za-z0-9.]+",
+        r"\bGPT[-_/A-Za-z0-9.]+",
+        r"\bDeepSeek[-_/A-Za-z0-9.]+",
+        r"\b[Mm]odel(?:\\s+used|\\s+selected|\\s*:)?\\s*[:=]?\\s*([A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+)",
+    ]
+    found: list[str] = []
+    for pattern in patterns:
+        for match in re.finditer(pattern, text, flags=re.IGNORECASE):
+            value = match.group(1) if match.groups() else match.group(0)
+            value = value.strip().strip('",` ')
+            if value and value not in found:
+                found.append(value)
+    return found[:12]
+
+
+def _trace_based_pi_model_resolution(run_id: str, bundle: dict[str, Any], *, bucket_source: str, token: str | None = None) -> dict[str, Any]:
+    state = bundle.get("state") or {}
+    launch = bundle.get("launch") or {}
+    requested = (
+        state.get("pi_model")
+        or launch.get("pi_model")
+        or launch.get("PI_MODEL")
+        or state.get("pi_model_resolution", {}).get("requested_model")
+        or ""
+    )
+    configured = state.get("pi_model_resolution", {}).get("configured_model") or requested
+    root = RunPaths(run_id, bucket_source=bucket_source).root
+    trace_texts: list[str] = []
+    try:
+        fs = _fs(token)
+        for pattern in (
+            f"{root}/traces/redacted/**/*.jsonl",
+            f"{root}/traces/redacted/*.jsonl",
+            f"{root}/traces/raw/**/*.jsonl",
+            f"{root}/logs/pi_output.txt",
+            f"{root}/logs/pi_live_output.txt",
+        ):
+            for path in fs.glob(pattern):
+                try:
+                    trace_texts.append(_safe_read_text(str(path), token=token)[:250000])
+                except Exception:
+                    pass
+    except Exception:
+        pass
+    observed = _extract_pi_models_from_text("\n".join(trace_texts))
+    n_requested = _normalize_model_name(requested)
+    n_configured = _normalize_model_name(configured)
+    effective = ""
+    for raw in observed:
+        n_raw = _normalize_model_name(raw)
+        if n_raw and n_raw not in {n_requested, n_configured}:
+            effective = raw
+            break
+    if not effective and observed:
+        effective = observed[0]
+    mismatch = bool(effective and _normalize_model_name(effective) not in {n_requested, n_configured})
+    if not requested and not observed:
+        return {}
+    return {
+        "requested_model": requested,
+        "configured_model": configured or requested,
+        "observed_models": observed,
+        "effective_model": effective or configured or requested,
+        "provider": "huggingface",
+        "mismatch": mismatch,
+        "source": "published_pi_traces",
+    }
+
+
 def summarize_run_bundle(run_id: str, bundle: dict[str, Any], *, bucket_source: str) -> dict[str, Any]:
     summary_file = bundle.get("summary_file") or {}
     state = bundle.get("state") or {}
@@ -361,6 +442,7 @@ def summarize_run_bundle(run_id: str, bundle: dict[str, Any], *, bucket_source:
         "smoke_test_passed": bool(smoke.get("ok") or smoke.get("status") == "success"),
         "latency_seconds": smoke.get("latency_seconds"),
         "expected_output_type": smoke.get("expected_output_type") or state.get("expected_output_type") or launch.get("expected_output_type") or "",
+        "pi_model_resolution": bundle.get("pi_model_resolution") or state.get("pi_model_resolution") or {},
         "artifacts_url": f"https://huggingface.co/buckets/{bucket_source}/tree/runs/{run_id}",
     }
 
@@ -385,9 +467,12 @@ def read_run_bundle(run_id: str, *, bucket_source: str, token: str | None = None
         "hardware_attempts": _safe_read_json(f"{paths.root}/hardware_attempts.json", token=token),
         "technical_blockers": _safe_read_json(f"{paths.root}/TECHNICAL_BLOCKERS.json", token=token),
         "model_analysis": _safe_read_json(f"{paths.root}/model_analysis.json", token=token),
+        "pi_model_resolution": _safe_read_json(f"{paths.root}/pi_model_resolution.json", token=token) or (read_json(paths.state, token=token) or {}).get("pi_model_resolution") or {},
         "space_runtime": _safe_read_json(f"{paths.root}/space_runtime.json", token=token),
         "files": _list_run_files(run_id, bucket_source=bucket_source, token=token) if include_heavy else [],
     }
+    if not bundle.get("pi_model_resolution"):
+        bundle["pi_model_resolution"] = _trace_based_pi_model_resolution(run_id, bundle, bucket_source=bucket_source, token=token)
     bundle["summary"] = summarize_run_bundle(run_id, bundle, bucket_source=bucket_source)
     return bundle
 
@@ -462,6 +547,7 @@ def list_recent_runs(
         gate = read_json(f"{root}/{run_id}/inference_gate.json", token=token) or {}
         smoke = read_json(f"{root}/{run_id}/tests/generation_smoke.json", token=token) or read_json(f"{root}/{run_id}/generation_smoke.json", token=token) or {}
         hardware = read_json(f"{root}/{run_id}/hardware_strategy.json", token=token) or {}
+        pi_model_resolution = read_json(f"{root}/{run_id}/pi_model_resolution.json", token=token) or state.get("pi_model_resolution") or {}
         partial_bundle = {
             "summary_file": summary_file,
             "launch": launch,
@@ -469,6 +555,7 @@ def list_recent_runs(
             "inference_gate": gate,
             "generation_smoke": smoke,
             "hardware_strategy": hardware,
+            "pi_model_resolution": pi_model_resolution,
         }
         item = summarize_run_bundle(run_id, partial_bundle, bucket_source=bucket_source)
         item["bucket_source"] = bucket_source

src/jobs.py

@@ -119,6 +119,7 @@ def launch_universal_model_card_job(
     fallback_space_hardware: str | None = None,
     allow_fixed_gpu_fallback: bool = True,
     implementation_mode: str | None = None,
+    expected_output_type: str | None = None,
     run_id: str | None = None,
     bucket_name: str | None = None,
 ) -> dict[str, Any]:
@@ -144,6 +145,7 @@ def launch_universal_model_card_job(
     env["FALLBACK_SPACE_HARDWARE"] = (fallback_space_hardware or "l40sx1").strip()
     env["ALLOW_FIXED_GPU_FALLBACK"] = "true" if allow_fixed_gpu_fallback else "false"
     env["IMPLEMENTATION_MODE"] = (implementation_mode or "full-inference-gated").strip()
+    env["EXPECTED_OUTPUT_TYPE"] = (expected_output_type or "any").strip()
     job = _launch_job(token=token, env=env, bucket_source=bucket_source, timeout="60m")
     return _job_result(
         job,
@@ -159,6 +161,7 @@ def launch_universal_model_card_job(
             "fallback_space_hardware": env["FALLBACK_SPACE_HARDWARE"],
             "allow_fixed_gpu_fallback": allow_fixed_gpu_fallback,
             "implementation_mode": env["IMPLEMENTATION_MODE"],
+            "expected_output_type": env["EXPECTED_OUTPUT_TYPE"],
         },
     )
 

src/worker_payload.py

@@ -279,6 +279,105 @@ def configure_pi(events_path: Path, model: str):
     append_event(events_path, "pi_config", "success", "Configured Pi", {"model": model})
 
 
+def normalize_pi_model_name(value: str | None) -> str:
+    return re.sub(r"[^a-z0-9]+", "", (value or "").lower())
+
+
+def extract_pi_models_from_text(text: str) -> list[str]:
+    """Best-effort extraction of assistant/model names from Pi stdout/session traces."""
+    if not text:
+        return []
+    patterns = [
+        r"\bQwen/[A-Za-z0-9_.-]+",
+        r"\bQwen(?:2(?:\.5)?|3)?[-_/A-Za-z0-9.]*Coder[-_/A-Za-z0-9.]*",
+        r"\bKimi[-_/A-Za-z0-9.]+",
+        r"\bMoonshotAI/[A-Za-z0-9_.-]+",
+        r"\bClaude[-_/A-Za-z0-9.]+",
+        r"\bGPT[-_/A-Za-z0-9.]+",
+        r"\bDeepSeek[-_/A-Za-z0-9.]+",
+        r"\b[Mm]odel(?:\\s+used|\\s+selected|\\s*:)?\\s*[:=]?\\s*([A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+)",
+    ]
+    found: list[str] = []
+    for pattern in patterns:
+        for match in re.finditer(pattern, text, flags=re.IGNORECASE):
+            value = match.group(1) if match.groups() else match.group(0)
+            value = value.strip().strip('",` ')
+            if value and value not in found:
+                found.append(value)
+    return found[:12]
+
+
+def detect_pi_model_resolution(requested_model: str, run_dir: Path, pi_output: str) -> dict:
+    settings_path = Path.home() / ".pi" / "agent" / "settings.json"
+    configured_model = requested_model
+    try:
+        settings = json.loads(settings_path.read_text(encoding="utf-8"))
+        configured_model = settings.get("model") or requested_model
+    except Exception:
+        settings = {"model": requested_model, "read_error": "could_not_read_settings"}
+
+    corpus = [pi_output or ""]
+    for trace in (run_dir / "traces" / "redacted").rglob("*.jsonl") if (run_dir / "traces" / "redacted").exists() else []:
+        try:
+            corpus.append(trace.read_text(encoding="utf-8", errors="ignore")[:200000])
+        except Exception:
+            pass
+    observed = extract_pi_models_from_text("\n".join(corpus))
+    normalized_requested = normalize_pi_model_name(requested_model)
+    normalized_configured = normalize_pi_model_name(configured_model)
+    normalized_observed = [normalize_pi_model_name(x) for x in observed]
+    effective_model = ""
+    for raw, normalized in zip(observed, normalized_observed):
+        if normalized and normalized != normalized_requested and normalized != normalized_configured:
+            effective_model = raw
+            break
+    if not effective_model and observed:
+        effective_model = observed[0]
+    mismatch = bool(effective_model and normalize_pi_model_name(effective_model) not in {normalized_requested, normalized_configured})
+    payload = {
+        "requested_model": requested_model,
+        "configured_model": configured_model,
+        "observed_models": observed,
+        "effective_model": effective_model or configured_model,
+        "provider": "huggingface",
+        "mismatch": mismatch,
+        "source": "published_pi_traces_after_sync",
+        "settings": settings,
+    }
+    write_json(run_dir / "pi_model_resolution.json", payload)
+    return payload
+
+
+def emit_pi_model_resolution(events_path: Path, resolution: dict):
+    if resolution.get("mismatch"):
+        append_event(
+            events_path,
+            "pi_model_resolution",
+            "warning",
+            "Pi assistant model appears to differ from the requested model",
+            {
+                "requested_model": resolution.get("requested_model"),
+                "configured_model": resolution.get("configured_model"),
+                "effective_model": resolution.get("effective_model"),
+                "observed_models": resolution.get("observed_models", [])[:8],
+                "provider": resolution.get("provider"),
+            },
+        )
+    else:
+        append_event(
+            events_path,
+            "pi_model_resolution",
+            "success",
+            "Pi assistant model matched the requested configuration",
+            {
+                "requested_model": resolution.get("requested_model"),
+                "configured_model": resolution.get("configured_model"),
+                "effective_model": resolution.get("effective_model"),
+                "provider": resolution.get("provider"),
+            },
+        )
+
+
 def collect_pi_traces(run_dir: Path, events_path: Path):
     count = sync_pi_traces(run_dir, emit_event=False)
     append_event(events_path, "traces", "success", "Collected Pi traces", {"count": count})
@@ -329,6 +428,221 @@ def api_names_from_schema(schema) -> list[str]:
     return list(dict.fromkeys(names))
 
 
+def normalize_api_name(name: str | None) -> str:
+    value = (name or "").strip()
+    if not value:
+        return "/generate"
+    return value if value.startswith("/") else "/" + value
+
+
+def default_smoke_args(expected_output_type: str) -> list:
+    expected = (expected_output_type or "any").lower()
+    if expected == "image":
+        return ["a cinematic robot cat astronaut, detailed, studio lighting"]
+    if expected == "video":
+        return ["a short cinematic shot of a robot cat astronaut walking on the moon"]
+    if expected == "audio":
+        return ["A calm voice saying hello from Agentic Space Factory."]
+    return ["Explain the concept of agentic AI in one paragraph."]
+
+
+def endpoint_schema_for_api(schema, api_name: str):
+    target = normalize_api_name(api_name)
+    alternatives = {target, target.lstrip("/")}
+    found = None
+
+    def walk(obj):
+        nonlocal found
+        if found is not None:
+            return
+        if isinstance(obj, dict):
+            for key, value in obj.items():
+                if isinstance(key, str) and key in alternatives and isinstance(value, (dict, list)):
+                    found = value
+                    return
+            value = obj.get("api_name") or obj.get("apiName")
+            if isinstance(value, str) and normalize_api_name(value) == target:
+                found = obj
+                return
+            for value in obj.values():
+                walk(value)
+        elif isinstance(obj, list):
+            for item in obj:
+                walk(item)
+
+    walk(schema)
+    return found
+
+
+def endpoint_parameter_names(endpoint) -> list[str]:
+    names: list[str] = []
+
+    def add(value):
+        if not isinstance(value, str):
+            return
+        cleaned = value.strip()
+        if cleaned and cleaned not in names:
+            names.append(cleaned)
+
+    def from_parameter(param):
+        if isinstance(param, dict):
+            for key in ["parameter_name", "parameterName", "name", "label"]:
+                add(param.get(key))
+            component = param.get("component") or param.get("component_type")
+            if isinstance(component, dict):
+                for key in ["label", "name"]:
+                    add(component.get(key))
+        elif isinstance(param, str):
+            add(param)
+
+    def walk(obj):
+        if isinstance(obj, dict):
+            params = obj.get("parameters") or obj.get("inputs")
+            if isinstance(params, list):
+                for param in params:
+                    from_parameter(param)
+            for value in obj.values():
+                if isinstance(value, (dict, list)):
+                    walk(value)
+        elif isinstance(obj, list):
+            for item in obj:
+                walk(item)
+
+    walk(endpoint)
+    return names
+
+
+def result_contains_expected_output(result, expected: str):
+    expected = (expected or "any").lower()
+    info = {"result_type": type(result).__name__, "result_repr": repr(result)[:2000]}
+    paths = []
+
+    def visit(obj):
+        if isinstance(obj, (str, Path)):
+            value = str(obj)
+            if value:
+                paths.append(value)
+        elif isinstance(obj, dict):
+            for key in ["path", "name", "url"]:
+                if key in obj:
+                    visit(obj[key])
+            for value in obj.values():
+                if isinstance(value, (dict, list, tuple)):
+                    visit(value)
+        elif isinstance(obj, (list, tuple)):
+            for item in obj:
+                visit(item)
+
+    visit(result)
+    info["detected_paths"] = paths[:20]
+    if expected == "any":
+        return result is not None, info
+    image_ext = (".png", ".jpg", ".jpeg", ".webp", ".gif")
+    video_ext = (".mp4", ".mov", ".webm")
+    audio_ext = (".wav", ".mp3", ".flac", ".ogg")
+    if expected == "text":
+        return isinstance(result, str) and bool(result.strip()), info
+    if expected == "image":
+        return any(str(p).lower().endswith(image_ext) for p in paths), info
+    if expected == "video":
+        return any(str(p).lower().endswith(video_ext) for p in paths), info
+    if expected == "audio":
+        return any(str(p).lower().endswith(audio_ext) for p in paths), info
+    return result is not None, info
+
+
+def copy_result_artifacts(result, run_dir: Path):
+    artifacts = run_dir / "artifacts"
+    artifacts.mkdir(parents=True, exist_ok=True)
+    copied = []
+
+    def maybe_copy(obj):
+        if isinstance(obj, (str, Path)):
+            path = Path(str(obj))
+            if path.exists() and path.is_file():
+                target = artifacts / path.name
+                try:
+                    shutil.copy2(path, target)
+                    copied.append(str(target))
+                except Exception:
+                    pass
+        elif isinstance(obj, dict):
+            for key in ["path", "name"]:
+                if key in obj:
+                    maybe_copy(obj[key])
+            for value in obj.values():
+                if isinstance(value, (dict, list, tuple)):
+                    maybe_copy(value)
+        elif isinstance(obj, (list, tuple)):
+            for item in obj:
+                maybe_copy(item)
+
+    maybe_copy(result)
+    return copied
+
+
+def measured_zero_gpu_recommendation(latency_seconds: float | None):
+    if latency_seconds is None:
+        return {
+            "observed_latency_seconds": None,
+            "recommended_zero_gpu_duration_seconds": None,
+            "recommendation_source": "not_measured",
+            "recommendation_confidence": "none",
+        }
+    recommended = int(max(30, min(300, latency_seconds * 2 + 15)))
+    return {
+        "observed_latency_seconds": round(latency_seconds, 3),
+        "recommended_zero_gpu_duration_seconds": recommended,
+        "recommendation_source": "live_gradio_predict",
+        "recommendation_confidence": "measured",
+        "measurement_note": "Measured from a live gradio_client.predict call, including Gradio/API/network/result serialization overhead.",
+    }
+
+
+def run_generation_smoke(target_space_id: str, token: str, run_dir: Path, events_path: Path, expected_output_type: str):
+    api_name = "/generate"
+    test_args = default_smoke_args(expected_output_type)
+    test_kwargs = {}
+    append_event(events_path, "generation_smoke", "started", "Calling live generation endpoint for ZeroGPU timing", {"api_name": api_name, "expected_output_type": expected_output_type})
+    client = make_gradio_client(target_space_id, token)
+    schema = client.view_api(return_format="dict")
+    discovered = api_names_from_schema(schema)
+    if api_name not in discovered and discovered:
+        non_health = [name for name in discovered if name != "/health"]
+        if non_health:
+            api_name = non_health[0]
+    endpoint = endpoint_schema_for_api(schema, api_name)
+    endpoint_parameters = endpoint_parameter_names(endpoint)
+    write_json(run_dir / "tests" / "generation_api_schema.json", {"schema": schema, "api_names": discovered, "selected_api_name": api_name, "endpoint_parameters": endpoint_parameters})
+    started = time.time()
+    result = client.predict(*test_args, api_name=api_name, **test_kwargs)
+    latency = time.time() - started
+    ok, info = result_contains_expected_output(result, expected_output_type)
+    copied = copy_result_artifacts(result, run_dir)
+    recommendation = measured_zero_gpu_recommendation(latency)
+    payload = {
+        "status": "success" if ok else "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,
+        "expected_output_type": expected_output_type,
+        "latency_seconds": round(latency, 3),
+        "result_info": info,
+        "copied_artifacts": copied,
+        "validated_at": now(),
+        **recommendation,
+    }
+    write_json(run_dir / "tests" / "generation_smoke.json", payload)
+    if ok:
+        append_event(events_path, "generation_smoke", "success", "Live generation smoke test passed and ZeroGPU timing was measured", {"latency_seconds": payload["latency_seconds"], "recommended_zero_gpu_duration_seconds": payload["recommended_zero_gpu_duration_seconds"], "api_name": api_name})
+    else:
+        append_event(events_path, "generation_smoke", "failed", "Live generation returned an unexpected output type", payload)
+    return payload
+
+
 def space_subdomain_url(target_space_id: str) -> str:
     owner, name = target_space_id.split("/", 1)
     # This matches the common Spaces app URL pattern. Keep conservative: our
@@ -846,7 +1160,7 @@ def load_json_if_exists(path: Path) -> dict:
         return {"parse_error": str(exc), "raw_tail": path.read_text(encoding="utf-8", errors="replace")[-2000:]}
 
 
-def infer_generation_gate(workspace: Path, implementation_mode: str, validation: dict, run_dir: Path, events_path: Path) -> dict:
+def infer_generation_gate(workspace: Path, implementation_mode: str, validation: dict, generation_smoke: dict | None, run_dir: Path, events_path: Path) -> dict:
     """Classify the run separately from process success.
 
     /health passing means the Space boots. It does not mean the generated Space
@@ -874,21 +1188,27 @@ def infer_generation_gate(workspace: Path, implementation_mode: str, validation:
         "out of scope",
     ]
     blocker_detected = bool(blockers) or any(m in combined for m in blocked_markers)
+    smoke_ok = isinstance(generation_smoke, dict) and generation_smoke.get("status") == "success"
+    recommendation = generation_smoke if isinstance(generation_smoke, dict) else measured_zero_gpu_recommendation(None)
     implementation_signals = {
         "has_spaces_gpu": "@spaces.GPU" in app_text,
         "has_torch": "torch" in req_text or "import torch" in app_text,
         "has_diffusers": "diffusers" in req_text or "diffusers" in app_text,
         "has_video_output_hint": any(x in app_text.lower() for x in ["gr.video", "video", ".mp4", "ffmpeg"]),
         "health_passed": validation.get("method") in {"http_health", "gradio"},
+        "generation_smoke_passed": smoke_ok,
+        "zero_gpu_duration_measured": recommendation.get("recommendation_confidence") == "measured",
     }
 
     if blocker_detected:
         status = "technical_blocker"
         message = "Space boots, but full model inference was not implemented. See TECHNICAL_BLOCKERS.json / PI_SUMMARY.md."
+    elif implementation_mode in {"full-inference-gated", "full-inference-attempt"} and smoke_ok:
+        status = "full_inference_success"
+        message = "Space boots and a live generation smoke test passed. ZeroGPU duration recommendation was measured from real inference."
     elif implementation_mode in {"full-inference-gated", "full-inference-attempt"}:
-        # Without a video smoke test, do not claim real inference success.
         status = "full_inference_candidate_health_passed"
-        message = "Space boots and contains inference signals, but no generation smoke test has validated a real video output."
+        message = "Space boots, but live generation smoke test did not produce a verified output. ZeroGPU duration recommendation was not measured."
     else:
         status = "health_only"
         message = "Safe scaffold health validation passed. Full inference was not requested."
@@ -921,6 +1241,14 @@ def infer_generation_gate(workspace: Path, implementation_mode: str, validation:
         "blocker_detected": blocker_detected,
         "implementation_signals": implementation_signals,
         "validation_method": validation.get("method"),
+        "generation_smoke": generation_smoke,
+        "zero_gpu_duration_recommendation": {
+            "observed_latency_seconds": recommendation.get("observed_latency_seconds"),
+            "recommended_zero_gpu_duration_seconds": recommendation.get("recommended_zero_gpu_duration_seconds"),
+            "recommendation_source": recommendation.get("recommendation_source"),
+            "recommendation_confidence": recommendation.get("recommendation_confidence"),
+            "measurement_note": recommendation.get("measurement_note"),
+        },
         "blockers": blockers,
     }
     write_json(run_dir / "inference_gate.json", gate)
@@ -940,12 +1268,14 @@ def main():
     fallback_hardware = os.environ.get("FALLBACK_SPACE_HARDWARE", "l40sx1")
     allow_fixed_gpu_fallback = os.environ.get("ALLOW_FIXED_GPU_FALLBACK", "true").lower() in {"1", "true", "yes", "on"}
     implementation_mode = os.environ.get("IMPLEMENTATION_MODE", "full-inference-attempt")
+    expected_output_type = os.environ.get("EXPECTED_OUTPUT_TYPE", "any")
     token = os.environ.get("HF_TOKEN")
 
     run_dir = output_root / "runs" / run_id
     events_path = run_dir / "events.jsonl"
     state_path = run_dir / "state.json"
     workspace = Path("/tmp/universal_workspace")
+    pi_model_resolution = {"requested_model": pi_model, "configured_model": pi_model, "effective_model": pi_model, "provider": "huggingface", "mismatch": False}
 
     append_event(events_path, "bootstrap", "started", "Universal model-card builder worker started", {"model_id": model_id, "target_space_id": target_space_id})
     write_json(state_path, {"run_id": run_id, "kind": "universal_model_card_builder", "status": "running", "message": "Attempting Universal model-card builderd Space creation", "model_id": model_id, "target_space": target_space_id, "created_by": hf_username, "bucket_source": bucket_source, "created_at": now(), "updated_at": now()})
@@ -990,6 +1320,11 @@ def main():
             collect_pi_traces(run_dir, events_path)
             fail(run_dir, events_path, "Pi failed before Space upload", {"returncode": code, "output_tail": pi_out[-4000:]})
         append_event(events_path, "pi_run", "success", "Pi completed universal model-card workspace pass", {"output_tail": pi_out[-2000:]})
+        # Pi sessions are the source of truth for the assistant/model actually used.
+        # Sync them first, then resolve requested/configured/observed model identity.
+        collect_pi_traces(run_dir, events_path)
+        pi_model_resolution = detect_pi_model_resolution(pi_model, run_dir, pi_out)
+        emit_pi_model_resolution(events_path, pi_model_resolution)
         if not (workspace / "PI_SUMMARY.md").exists():
             (workspace / "PI_SUMMARY.md").write_text("# Pi Summary\n\nPi did not create a PI_SUMMARY.md. See logs/pi_output.txt.\n", encoding="utf-8")
 
@@ -1035,7 +1370,24 @@ def main():
                 raise
             upload_workspace(api, workspace, target_space_id, token, run_dir, events_path)
             validation = validate_live_api(api, target_space_id, token, run_dir, events_path, timeout_s=1200)
-        inference_gate = infer_generation_gate(workspace, implementation_mode, validation, run_dir, events_path)
+        generation_smoke = None
+        if implementation_mode in {"full-inference-gated", "full-inference-attempt"}:
+            try:
+                generation_smoke = run_generation_smoke(target_space_id, token, run_dir, events_path, expected_output_type)
+            except Exception as smoke_error:
+                generation_smoke = {
+                    "status": "failed",
+                    "target_space": target_space_id,
+                    "expected_output_type": expected_output_type,
+                    "error": str(smoke_error)[:4000],
+                    **measured_zero_gpu_recommendation(None),
+                }
+                write_json(run_dir / "tests" / "generation_smoke.json", generation_smoke)
+                append_event(events_path, "generation_smoke", "failed", "Live generation smoke test failed; ZeroGPU duration was not measured", generation_smoke)
+        else:
+            generation_smoke = measured_zero_gpu_recommendation(None) | {"status": "skipped", "expected_output_type": expected_output_type}
+            write_json(run_dir / "tests" / "generation_smoke.json", generation_smoke)
+        inference_gate = infer_generation_gate(workspace, implementation_mode, validation, generation_smoke, run_dir, events_path)
 
         # If the generated app looks like real GPU inference but automatic
         # hardware requests failed, classify the run honestly as needing manual
@@ -1067,7 +1419,9 @@ def main():
             "target_space_url": f"https://huggingface.co/spaces/{target_space_id}",
             "selected_hardware": selected_hardware,
             "hardware_attempts": hardware_attempts,
+            "pi_model_resolution": pi_model_resolution,
             "validation": validation,
+            "generation_smoke": generation_smoke,
             "inference_gate": inference_gate,
             "updated_at": now(),
             "created_by": hf_username,
@@ -1086,6 +1440,12 @@ Target Space: https://huggingface.co/spaces/{target_space_id}
 
 Model: `{model_id}`
 
+## Pi assistant model
+
+```json
+{json.dumps(pi_model_resolution, indent=2, ensure_ascii=False)}
+```
+
 ## Hardware
 
 Selected/requested hardware: `{selected_hardware}`
@@ -1104,6 +1464,12 @@ The wrapper validated the live Space using HTTP `/health` first, with Gradio Cli
 {json.dumps(validation, indent=2, ensure_ascii=False)}
 ```
 
+## Live generation smoke / ZeroGPU duration
+
+```json
+{json.dumps(generation_smoke, indent=2, ensure_ascii=False)}
+```
+
 ## Full-inference gate
 
 ```json
@@ -1238,6 +1604,90 @@ def api_names_from_schema(schema) -> list[str]:
     return names
 
 
+def normalize_api_name(name: str | None) -> str:
+    value = (name or "").strip()
+    if not value:
+        return "/generate"
+    return value if value.startswith("/") else "/" + value
+
+
+def endpoint_schema_for_api(schema, api_name: str):
+    target = normalize_api_name(api_name)
+    alternatives = {target, target.lstrip("/")}
+    found = None
+
+    def walk(obj):
+        nonlocal found
+        if found is not None:
+            return
+        if isinstance(obj, dict):
+            for key, value in obj.items():
+                if isinstance(key, str) and key in alternatives and isinstance(value, (dict, list)):
+                    found = value
+                    return
+            value = obj.get("api_name") or obj.get("apiName")
+            if isinstance(value, str) and normalize_api_name(value) == target:
+                found = obj
+                return
+            for value in obj.values():
+                walk(value)
+        elif isinstance(obj, list):
+            for item in obj:
+                walk(item)
+
+    walk(schema)
+    return found
+
+
+def endpoint_parameter_names(endpoint) -> list[str]:
+    names: list[str] = []
+
+    def add(value):
+        if not isinstance(value, str):
+            return
+        cleaned = value.strip()
+        if cleaned and cleaned not in names:
+            names.append(cleaned)
+
+    def from_parameter(param):
+        if isinstance(param, dict):
+            for key in ["parameter_name", "parameterName", "name", "label"]:
+                add(param.get(key))
+            component = param.get("component") or param.get("component_type")
+            if isinstance(component, dict):
+                for key in ["label", "name"]:
+                    add(component.get(key))
+        elif isinstance(param, str):
+            add(param)
+
+    def walk(obj):
+        if isinstance(obj, dict):
+            params = obj.get("parameters") or obj.get("inputs")
+            if isinstance(params, list):
+                for param in params:
+                    from_parameter(param)
+            for value in obj.values():
+                if isinstance(value, (dict, list)):
+                    walk(value)
+        elif isinstance(obj, list):
+            for item in obj:
+                walk(item)
+
+    walk(endpoint)
+    return names
+
+
+def sanitize_kwargs_for_schema(api_name: str, schema, kwargs: dict):
+    endpoint = endpoint_schema_for_api(schema, api_name)
+    parameter_names = endpoint_parameter_names(endpoint)
+    if not parameter_names:
+        return kwargs, {}, []
+    allowed = set(parameter_names)
+    sanitized = {key: value for key, value in kwargs.items() if key in allowed}
+    dropped = {key: value for key, value in kwargs.items() if key not in allowed}
+    return sanitized, dropped, parameter_names
+
+
 def runtime_to_dict(runtime) -> dict:
     payload = {}
     for attr in ["stage", "hardware", "requested_hardware", "sleep_time", "storage", "gc_timeout"]:
@@ -1422,7 +1872,7 @@ def copy_result_artifacts(result, run_dir: Path):
 
 
 def smoke_generate(target_space_id: str, token: str, run_dir: Path, events_path: Path):
-    api_name = (os.environ.get("API_NAME") or "/generate").strip()
+    api_name = normalize_api_name(os.environ.get("API_NAME") or "/generate")
     expected_output_type = (os.environ.get("EXPECTED_OUTPUT_TYPE") or "any").strip()
     test_args = parse_json_env("TEST_ARGS_JSON", ["a cinematic robot cat astronaut, detailed, studio lighting"])
     test_kwargs = parse_json_env("TEST_KWARGS_JSON", {})
@@ -1434,9 +1884,33 @@ def smoke_generate(target_space_id: str, token: str, run_dir: Path, events_path:
     client = make_gradio_client(target_space_id, token)
     schema = client.view_api(return_format="dict")
     discovered = api_names_from_schema(schema)
-    write_json(run_dir / "tests" / "api_schema.json", {"schema": schema, "api_names": discovered})
+    safe_kwargs, dropped_kwargs, endpoint_parameters = sanitize_kwargs_for_schema(api_name, schema, test_kwargs)
+    write_json(run_dir / "tests" / "api_schema.json", {"schema": schema, "api_names": discovered, "selected_api_name": api_name, "endpoint_parameters": endpoint_parameters})
+    if dropped_kwargs:
+        append_event(
+            events_path,
+            "generation_smoke",
+            "warning",
+            "Ignored unsupported validation kwargs for this Gradio endpoint",
+            {"api_name": api_name, "ignored_kwargs": sorted(dropped_kwargs), "endpoint_parameters": endpoint_parameters},
+        )
     started = time.time()
-    result = client.predict(*test_args, api_name=api_name, **test_kwargs)
+    try:
+        result = client.predict(*test_args, api_name=api_name, **safe_kwargs)
+    except Exception as exc:
+        message = str(exc)
+        if safe_kwargs and "not a valid key-word argument" in message:
+            append_event(
+                events_path,
+                "generation_smoke",
+                "warning",
+                "Retrying validation without keyword arguments after Gradio rejected kwargs",
+                {"api_name": api_name, "error": message[:1500], "dropped_kwargs": sorted(safe_kwargs)},
+            )
+            safe_kwargs = {}
+            result = client.predict(*test_args, api_name=api_name)
+        else:
+            raise
     latency = time.time() - started
     ok, info = result_contains_expected_output(result, expected_output_type)
     copied = copy_result_artifacts(result, run_dir)
@@ -1446,12 +1920,19 @@ def smoke_generate(target_space_id: str, token: str, run_dir: Path, events_path:
         "api_name": api_name,
         "discovered_api_names": discovered,
         "test_args": test_args,
-        "test_kwargs": test_kwargs,
+        "test_kwargs": safe_kwargs,
+        "original_test_kwargs": test_kwargs,
+        "ignored_test_kwargs": dropped_kwargs,
+        "endpoint_parameters": endpoint_parameters,
         "expected_output_type": expected_output_type,
         "latency_seconds": round(latency, 3),
+        "observed_latency_seconds": round(latency, 3),
         "result_info": info,
         "copied_artifacts": copied,
         "recommended_zero_gpu_duration_seconds": int(max(30, min(300, latency * 2 + 15))),
+        "recommendation_source": "live_gradio_predict",
+        "recommendation_confidence": "measured",
+        "measurement_note": "Measured from a live gradio_client.predict call, including Gradio/API/network/result serialization overhead.",
         "validated_at": now(),
     }
     write_json(run_dir / "tests" / "generation_smoke.json", payload)