from __future__ import annotations import base64 import textwrap def _encode(script: str) -> str: return base64.b64encode(script.encode("utf-8")).decode("ascii") UNIVERSAL_MODEL_CARD_WORKER_SCRIPT = r''' import hashlib import hmac import hashlib import hmac import json import os import re import shutil import subprocess import sys import time from datetime import datetime, timezone from pathlib import Path from textwrap import dedent TARGET_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{1,95}/[A-Za-z0-9][A-Za-z0-9._-]{1,95}$") GIST_URL = "https://gist.github.com/gary149/2aba2962375fa9ca56bb9ef53f00b73d" AUTO_SPACE_HARDWARE_CHOICES = {"zero-a10g", "cpu-basic", "t4-small", "t4-medium", "a10g-small", "a10g-large", "l4x1", "l40sx1"} DEFAULT_PREFERRED_SPACE_HARDWARE = "zero-a10g" DEFAULT_FALLBACK_SPACE_HARDWARE = "a10g-large" DEFAULT_MODEL_ID = "sshleifer/tiny-gpt2" # Internal agent/recovery files may be needed inside the transient Pi # workspace, but they should not be published to the generated Space or shown # twice as generated application files. Canonical copies live in run_dir/logs, # run_dir/repair, and run_dir/traces. INTERNAL_WORKSPACE_ARTIFACT_NAMES = { "GOAL.md", "INCIDENT_BRIEF.md", "PI_DIAGNOSIS_GOAL.md", "DEPENDENCY_ERROR_BRIEF.md", "REPAIR_DECISION.json", "REPAIR_GOAL.md", "REPAIR_BRIEF.md", "REPAIR_PLAN.md", "REPAIR_SUMMARY.md", "PI_SUMMARY.md", "TECHNICAL_BLOCKERS.json", } def internal_workspace_upload_ignore_patterns() -> list[str]: patterns = [".git/*", ".cache/*", "**/.cache/*", "node_modules/*", "__pycache__/*", "*.pyc", ".pi/*", "**/.pi/*", "auth.json", "**/auth.json"] for name in sorted(INTERNAL_WORKSPACE_ARTIFACT_NAMES): patterns.extend([name, f"**/{name}"]) return patterns def internal_workspace_copy_ignore(_dir: str, names: list[str]) -> set[str]: ignored = {".git", ".cache", "node_modules", "__pycache__", ".pi"} ignored.update(name for name in names if name in INTERNAL_WORKSPACE_ARTIFACT_NAMES or name.endswith(".pyc")) return ignored def is_publishable_workspace_file(path: Path) -> bool: if path.name in INTERNAL_WORKSPACE_ARTIFACT_NAMES or path.name == "auth.json": return False if any(part in {".git", ".cache", "node_modules", "__pycache__"} for part in path.parts): return False if path.suffix == ".pyc": return False return True def now(): return datetime.now(timezone.utc).isoformat() def write_json(path: Path, payload: dict): path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(payload, indent=2, ensure_ascii=False) + "\n", encoding="utf-8") def append_event(path: Path, step: str, status: str, message: str, data: dict | None = None): path.parent.mkdir(parents=True, exist_ok=True) event = {"ts": now(), "step": step, "status": status, "message": message, "data": data or {}} line = json.dumps(event, ensure_ascii=False) with path.open("a", encoding="utf-8") as f: f.write(line + "\n") print(line, flush=True) SAFE_SPACE_PYTHON_VERSIONS = {"3.10", "3.11", "3.12"} DEFAULT_SPACE_PYTHON_VERSION = "3.10" def normalize_space_python_version(value) -> tuple[str, bool, str]: raw = str(value or "").strip().strip("\"'") lowered = raw.lower().replace("python", "").strip() match = re.search(r"(3)\.(\d+)(?:\.\d+)?", lowered) if match: normalized = f"{match.group(1)}.{match.group(2)}" else: normalized = lowered if normalized in SAFE_SPACE_PYTHON_VERSIONS: return normalized, normalized != raw, "allowed" return DEFAULT_SPACE_PYTHON_VERSION, True, f"unsupported_or_ambiguous:{raw or 'missing'}" def requirements_has_package(lines: list[str], package: str) -> bool: wanted = package.lower().replace("_", "-") for line in lines: stripped = line.strip() if not stripped or stripped.startswith("#") or stripped.startswith("-") or "://" in stripped: continue name = re.split(r"[<>=!~;\[]", stripped, 1)[0].strip().lower().replace("_", "-") if name == wanted: return True return False def workspace_app_imports_torch(workspace: Path) -> bool: app_path = workspace / "app.py" if not app_path.exists(): return False text = app_path.read_text(encoding="utf-8", errors="ignore") return bool(re.search(r"(?m)^\s*(import\s+torch\b|from\s+torch\b)", text)) def _artifact_entry(run_dir: Path, rel_path: str, *, kind: str = "file") -> dict: path = run_dir / rel_path if kind == "folder": present = path.exists() and any(child.is_file() for child in path.rglob("*")) else: present = path.exists() and path.is_file() payload = {"path": rel_path, "kind": kind, "present": bool(present)} if present and kind == "file": try: payload["size"] = path.stat().st_size except Exception: pass return payload def write_artifact_manifest(run_dir: Path, *, events_path: Path | None = None, reason: str = "snapshot") -> dict: """Snapshot the artifacts that the worker actually wrote to the mounted bucket. The web UI should prefer this manifest over guessed paths. It is written during failures as well as success so a run that created a Space and then crashed can still expose traces, logs and reports in the dock. """ run_id = os.environ.get("RUN_ID", "") bucket_source = os.environ.get("BUCKET_SOURCE", "") artifacts = [ _artifact_entry(run_dir, "events.jsonl"), _artifact_entry(run_dir, "state.json"), _artifact_entry(run_dir, "report.md"), _artifact_entry(run_dir, "model_analysis.json"), _artifact_entry(run_dir, "hardware_strategy.json"), _artifact_entry(run_dir, "hardware_attempts.json"), _artifact_entry(run_dir, "inference_gate.json"), _artifact_entry(run_dir, "space_runtime.json"), _artifact_entry(run_dir, "tests/generation_smoke.json"), _artifact_entry(run_dir, "tests/api_schema.json"), _artifact_entry(run_dir, "generated/TECHNICAL_BLOCKERS.json"), _artifact_entry(run_dir, "repair/INCIDENT_BRIEF.md"), _artifact_entry(run_dir, "repair/DEPENDENCY_ERROR_BRIEF.md"), _artifact_entry(run_dir, "repair/PI_DIAGNOSIS_GOAL.md"), _artifact_entry(run_dir, "repair/REPAIR_DECISION.json"), _artifact_entry(run_dir, "repair/REPAIR_BRIEF.md"), _artifact_entry(run_dir, "repair/REPAIR_PLAN.md"), _artifact_entry(run_dir, "repair/REPAIR_SUMMARY.md"), _artifact_entry(run_dir, "repair/BLOCKAGE.json"), _artifact_entry(run_dir, "logs/pi_live_output.txt"), _artifact_entry(run_dir, "logs/pi_output.txt"), _artifact_entry(run_dir, "logs/pi_diagnosis_output.txt"), _artifact_entry(run_dir, "logs/pi_repair_output.txt"), _artifact_entry(run_dir, "logs/space_logs_build.txt"), _artifact_entry(run_dir, "logs/space_logs_runtime.txt"), _artifact_entry(run_dir, "logs/space_logs_index.json"), _artifact_entry(run_dir, "logs/space_runtime_snapshot.json"), _artifact_entry(run_dir, "logs/space_log_diagnostics.json"), _artifact_entry(run_dir, "traces/raw", kind="folder"), _artifact_entry(run_dir, "traces/redacted", kind="folder"), _artifact_entry(run_dir, "generated", kind="folder"), _artifact_entry(run_dir, "repair", kind="folder"), _artifact_entry(run_dir, "logs", kind="folder"), ] payload = { "run_id": run_id, "bucket_source": bucket_source, "updated_at": now(), "reason": reason, "artifacts": artifacts, "present_paths": [a["path"] for a in artifacts if a.get("present")], } write_json(run_dir / "artifact_manifest.json", payload) if os.environ.get("ASF_EVAL_ENABLED", "").strip().lower() in {"1", "true", "yes", "on"}: try: publish_eval_record(run_dir, phase=f"manifest:{reason}", events_path=events_path) except Exception: pass if events_path is not None: try: append_event(events_path, "artifact_manifest", "success", "Updated run artifact manifest", {"reason": reason, "present_count": len(payload["present_paths"])}) except Exception: pass return payload def update_state(run_dir: Path, patch: dict) -> dict: path = run_dir / "state.json" current = load_json_if_exists(path) if path.exists() else {} if not isinstance(current, dict): current = {} merged = {**current, **patch, "updated_at": now()} write_json(path, merged) return merged def redact_text(text: str | None) -> str: if not text: return "" value = text for secret_name in ["HF_TOKEN", "HUGGING_FACE_HUB_TOKEN"]: secret = os.environ.get(secret_name) if secret: value = value.replace(secret, "[REDACTED]") value = re.sub(r"Bearer\s+[A-Za-z0-9_\-.=]+", "Bearer [REDACTED]", value) value = re.sub(r"hf_[A-Za-z0-9_\-]{10,}", "hf_[REDACTED]", value) return value def ensure_hf_token_context(run_dir: Path, events_path: Path | None = None) -> dict: """Ensure Pi and HF tooling see the same private HF token safely. HF Jobs receive HF_TOKEN as a secret. Pi, huggingface_hub, gradio_client, and the hf CLI can use either HF_TOKEN or HUGGING_FACE_HUB_TOKEN depending on the tool/version, so expose both aliases inside the isolated Job process. Never write the token value to artifacts. """ token = os.environ.get("HF_TOKEN") or os.environ.get("HUGGING_FACE_HUB_TOKEN") or "" if token: os.environ.setdefault("HF_TOKEN", token) os.environ.setdefault("HUGGING_FACE_HUB_TOKEN", token) os.environ.setdefault("HF_HUB_ENABLE_HF_TRANSFER", "1") payload = { "hf_token_present": bool(token), "hf_token_length": len(token) if token else 0, "hugging_face_hub_token_alias_present": bool(os.environ.get("HUGGING_FACE_HUB_TOKEN")), "hf_username": os.environ.get("HF_USERNAME") or "", "bucket_source": os.environ.get("BUCKET_SOURCE") or "", "target_space_id": os.environ.get("TARGET_SPACE_ID") or "", "token_value": "[REDACTED]" if token else "", } run_dir.mkdir(parents=True, exist_ok=True) write_json(run_dir / "token_context.json", payload) if events_path: append_event(events_path, "token_context", "success" if token else "warning", "Verified HF token context for Pi/HF operations", {k: v for k, v in payload.items() if k != "token_value"}) return payload def pi_tooling_context_note() -> str: return """HF tooling context: - You are running inside an HF Job with an HF_TOKEN secret configured. - The Factory exposes both HF_TOKEN and HUGGING_FACE_HUB_TOKEN aliases for huggingface_hub, gradio_client, hf CLI, and Pi provider access. - You may use the token through normal tools when needed to inspect the model/Space or validate private resources. - Never print, write, echo, or commit token values. Do not include Authorization headers in artifacts. - If a Hub operation fails with 401/403/quota/billing, report it as evidence instead of retrying blindly. """ def _agent_trace_json_safe(payload): try: return json.loads(redact_text(json.dumps(payload, ensure_ascii=False))) except Exception: return {"value": redact_text(str(payload))[:8000]} def write_agent_trace_record( run_dir: Path, *, phase: str, event: str, status: str = "info", message: str = "", data: dict | None = None, text: str | None = None, artifacts: list[str] | None = None, ): """Append one run-level agent trace record to RAW and redacted journals. Pi can be invoked several times during a run: initial build, blockage diagnosis, and optional repair patch. The individual Pi session files and stdout logs remain useful, but this journal is the canonical narrative tying those phases together into one auditable agent run. """ try: payload = { "ts": now(), "phase": phase, "event": event, "status": status, "message": message, "data": data or {}, "artifacts": artifacts or [], } if text is not None: payload["text"] = text raw_dir = run_dir / "traces" / "raw" redacted_dir = run_dir / "traces" / "redacted" raw_dir.mkdir(parents=True, exist_ok=True) redacted_dir.mkdir(parents=True, exist_ok=True) raw_line = json.dumps(payload, ensure_ascii=False) redacted_payload = _agent_trace_json_safe(payload) redacted_line = json.dumps(redacted_payload, ensure_ascii=False) with (raw_dir / "agent_trace.jsonl").open("a", encoding="utf-8") as f: f.write(raw_line + "\n") with (redacted_dir / "agent_trace.jsonl").open("a", encoding="utf-8") as f: f.write(redacted_line + "\n") except Exception: # Trace journaling must never break the build/repair path. pass def append_agent_trace_artifact(run_dir: Path, *, phase: str, event: str, artifact: str, text: str | None = None, status: str = "success", data: dict | None = None): tail = None if text is None else text[-20000:] write_agent_trace_record(run_dir, phase=phase, event=event, status=status, message=f"Agent phase wrote {artifact}", data=data or {}, text=tail, artifacts=[artifact]) def safe_details(details: dict | None) -> dict: if not details: return {} try: return json.loads(redact_text(json.dumps(details, ensure_ascii=False))) except Exception: return {"redacted_details": redact_text(str(details))[-4000:]} def exception_payload(exc: Exception) -> dict: """Extract the most useful redacted details from HF Hub / HTTP exceptions.""" payload = { "error": redact_text(str(exc))[:4000], "exception_type": type(exc).__name__, } response = getattr(exc, "response", None) if response is not None: status_code = getattr(response, "status_code", None) if status_code is not None: payload["status_code"] = status_code try: text_body = response.text except Exception: text_body = "" if text_body: payload["response_text"] = redact_text(text_body)[:4000] payload["hf_error_detail"] = redact_text(text_body)[:4000] try: json_body = response.json() except Exception: json_body = None if json_body is not None: payload["response_json"] = safe_details(json_body) if isinstance(json_body, dict): for key in ("error", "message", "detail", "details"): value = json_body.get(key) if value: payload["hf_error_message"] = redact_text(value if isinstance(value, str) else json.dumps(value, ensure_ascii=False))[:4000] break for attr in ("server_message", "request_id"): value = getattr(exc, attr, None) if value: payload[attr] = redact_text(str(value))[:1000] return payload def eval_enabled() -> bool: return os.environ.get("ASF_EVAL_RECORD_ENABLED", "").strip().lower() in {"1", "true", "yes", "on"} def eval_safe_hash(value: str | None, prefix: str = "h") -> str: raw = (value or "").strip() salt = os.environ.get("ASF_EVAL_SALT", "").strip() if not raw: return "" if salt: digest = hmac.new(salt.encode("utf-8"), raw.encode("utf-8"), hashlib.sha256).hexdigest()[:24] else: digest = hashlib.sha256(raw.encode("utf-8")).hexdigest()[:24] return f"{prefix}_{digest}" def eval_load_json(path: Path) -> dict: if not path.exists(): return {} try: return json.loads(path.read_text(encoding="utf-8", errors="replace")) except Exception as exc: return {"_parse_error": str(exc)} def eval_load_events(path: Path) -> list[dict]: if not path.exists(): return [] events = [] for line in path.read_text(encoding="utf-8", errors="replace").splitlines(): try: evt = json.loads(line) if isinstance(evt, dict): events.append(evt) except Exception: pass return events def eval_parse_ts(value: str | None) -> float | None: if not value: return None try: return datetime.fromisoformat(str(value).replace("Z", "+00:00")).timestamp() except Exception: return None def eval_compact_timeline(events: list[dict]) -> list[dict]: stages: dict[str, dict] = {} order: list[str] = [] for evt in events: step = str(evt.get("step") or "unknown")[:80] status = str(evt.get("status") or "")[:80] ts = str(evt.get("ts") or "") tsv = eval_parse_ts(ts) if step not in stages: stages[step] = {"stage": step, "first_ts": ts, "last_ts": ts, "status": status, "event_count": 0} order.append(step) item = stages[step] item["last_ts"] = ts or item.get("last_ts") item["status"] = status or item.get("status") item["event_count"] = int(item.get("event_count") or 0) + 1 if tsv is not None: if "_first" not in item: item["_first"] = tsv item["_last"] = tsv timeline = [] for step in order: item = dict(stages[step]) first = item.pop("_first", None) last = item.pop("_last", None) if first is not None and last is not None and last >= first: item["duration_ms"] = int((last - first) * 1000) timeline.append(item) return timeline[-80:] def eval_redacted_tail(path: Path, limit: int = 2400) -> str: if os.environ.get("ASF_EVAL_INCLUDE_REDACTED_TAILS", "").strip().lower() not in {"1", "true", "yes", "on"}: return "" try: return redact_text(path.read_text(encoding="utf-8", errors="replace"))[-limit:] except Exception: return "" def eval_terminal_status(status: str, phase: str = "") -> bool: status = str(status or "").strip().lower() phase = str(phase or "").strip().lower() if phase in {"final", "failure"}: return True return status in { "done", "success", "failed", "cancelled", "canceled", "full_inference_success", "full_inference_candidate_health_passed", "health_only", "technical_blocker", "manual_hardware_required", } def eval_event_seen(events: list[dict], step: str, status: str | None = None) -> bool: for event in events or []: if str(event.get("step") or "") != step: continue if status is None or str(event.get("status") or "") == status: return True return False def eval_health_passed(inference_gate: dict, generation_smoke: dict, events: list[dict], status: str) -> bool: signals = inference_gate.get("implementation_signals") if isinstance(inference_gate.get("implementation_signals"), dict) else {} if signals.get("health_passed") is True: return True if str(status or "") in {"full_inference_success", "full_inference_candidate_health_passed", "health_only"}: return True if eval_event_seen(events, "api_validation", "success") or eval_event_seen(events, "repair_validation", "success"): return True return False def eval_generation_smoke_passed(inference_gate: dict, generation_smoke: dict) -> bool: signals = inference_gate.get("implementation_signals") if isinstance(inference_gate.get("implementation_signals"), dict) else {} return bool(signals.get("generation_smoke_passed") is True or generation_smoke.get("status") == "success") def eval_failure_summary(status: str, state: dict, inference_gate: dict, generation_smoke: dict, health_passed: bool, smoke_passed: bool) -> dict: failure_type = generation_smoke.get("failure_type") or inference_gate.get("failure_type") or "" failure_owner = generation_smoke.get("failure_owner") or inference_gate.get("failure_owner") or "" missing_argument = generation_smoke.get("missing_argument") or inference_gate.get("missing_argument") or "" reason = missing_argument or failure_type or str(state.get("message") or inference_gate.get("message") or "")[:240] if not failure_owner: if failure_type == "validator_request_error" or missing_argument: failure_owner = "factory_validator" elif smoke_passed: failure_owner = "" elif health_passed and str(generation_smoke.get("status") or "") == "failed": failure_owner = "unknown" return { "failure_type": failure_type, "failure_owner": failure_owner, "failure_reason": reason, "missing_argument": missing_argument, } def eval_verdict(status: str, phase: str, health_passed: bool, smoke_passed: bool, full_inference_verified: bool, inference_gate: dict) -> str: status = str(status or "").strip() if status in {"cancelled", "canceled"}: return "cancelled" if status == "manual_hardware_required" or inference_gate.get("manual_hardware_required"): return "manual_action_required" if status == "technical_blocker": return "technical_blocker" if full_inference_verified: return "success" if status in {"full_inference_candidate_health_passed", "health_only"} or (health_passed and not smoke_passed): return "partial_validation" if phase == "failure" or status in {"failed", "error"}: return "failed" if eval_terminal_status(status, phase): return "partial_validation" if health_passed else "failed" return "running" def publish_eval_record(run_dir: Path, *, phase: str, events_path: Path | None = None) -> dict: """Write an anonymized cross-run evaluation record to the optional eval bucket. This intentionally stores structured metrics and hashed identifiers, not raw generated code, prompts, tokens, user bucket paths, or target Space IDs. """ if not eval_enabled(): return {"enabled": False} # Jobs only write the anonymized eval record back into the user's run bucket. # The ASF backend later publishes this record to the operator eval bucket # mounted on the ASF Space. User Jobs must never mount or write the private # operator archive directly. run_id = os.environ.get("RUN_ID", run_dir.name) events = eval_load_events(events_path or (run_dir / "events.jsonl")) state = eval_load_json(run_dir / "state.json") analysis = eval_load_json(run_dir / "model_analysis.json") hardware_strategy = eval_load_json(run_dir / "hardware_strategy.json") hardware_attempts = eval_load_json(run_dir / "hardware_attempts.json") inference_gate = eval_load_json(run_dir / "inference_gate.json") generation_smoke = eval_load_json(run_dir / "tests" / "generation_smoke.json") space_runtime = eval_load_json(run_dir / "space_runtime.json") manifest = eval_load_json(run_dir / "artifact_manifest.json") repair_decision = eval_load_json(run_dir / "repair" / "REPAIR_DECISION.json") blockage = eval_load_json(run_dir / "repair" / "BLOCKAGE.json") pi_resolution = state.get("pi_model_resolution") if isinstance(state.get("pi_model_resolution"), dict) else {} status = str(state.get("status") or inference_gate.get("status") or phase or "unknown") failure_message = str(state.get("message") or (state.get("details") or {}).get("error") or "")[:1200] started_at = events[0].get("ts") if events else (state.get("created_at") or os.environ.get("LAUNCHED_AT") or "") finished_at = state.get("updated_at") or (events[-1].get("ts") if events else now()) model_id = os.environ.get("MODEL_ID") or state.get("model_id") or analysis.get("model_id") or "" target_space = state.get("target_space") or os.environ.get("TARGET_SPACE_ID") or "" include_model_id = os.environ.get("ASF_EVAL_INCLUDE_MODEL_ID", "").strip().lower() in {"1", "true", "yes", "on"} anon_run_id = eval_safe_hash(run_id, "run") health_passed = eval_health_passed(inference_gate, generation_smoke, events, status) generation_smoke_passed = eval_generation_smoke_passed(inference_gate, generation_smoke) full_inference_verified = bool(status == "full_inference_success" or (health_passed and generation_smoke_passed)) process_completed = eval_terminal_status(status, phase) failure_summary = eval_failure_summary(status, state, inference_gate, generation_smoke, health_passed, generation_smoke_passed) verdict = eval_verdict(status, phase, health_passed, generation_smoke_passed, full_inference_verified, inference_gate) record = { "process_completed": process_completed, "verdict": verdict, "failure_owner": failure_summary.get("failure_owner") or "", "failure_reason": failure_summary.get("failure_reason") or "", "schema_version": "1.2", "app_version": os.environ.get("ASF_VERSION", "unknown"), "phase": phase, "anonymous_run_id": anon_run_id, "run_id_hash": eval_safe_hash(run_id, "run"), "anonymous_user_hash": eval_safe_hash(os.environ.get("HF_USERNAME"), "user"), "salt_configured": bool(os.environ.get("ASF_EVAL_SALT")), "run_kind": state.get("kind") or ("validate_existing_space" if os.environ.get("API_NAME") else "universal_model_card_builder"), "started_at": started_at, "finished_at": finished_at, "input_model": { "model_id_redacted": not include_model_id, "model_id": model_id if include_model_id else "", "model_hash": eval_safe_hash(model_id, "model"), "pipeline_tag": analysis.get("pipeline_tag"), "library_name": analysis.get("library_name"), "default_model_target": bool(analysis.get("default_model_target")), }, "config": { "implementation_mode": os.environ.get("IMPLEMENTATION_MODE", ""), "expected_output_type": os.environ.get("EXPECTED_OUTPUT_TYPE", ""), "pi_model_requested": os.environ.get("PI_MODEL") or state.get("pi_model") or "", "pi_model_effective": pi_resolution.get("effective_model") or pi_resolution.get("observed_model") or "", "try_zero_gpu_first": os.environ.get("TRY_ZERO_GPU_FIRST", ""), "allow_fixed_gpu_fallback": os.environ.get("ALLOW_FIXED_GPU_FALLBACK", ""), "preferred_gpu": os.environ.get("PREFERRED_SPACE_HARDWARE", ""), "fallback_gpu": os.environ.get("FALLBACK_SPACE_HARDWARE", ""), }, "outcome": { "status": status, "verdict": verdict, "process_completed": process_completed, "health_passed": health_passed, "generation_smoke_passed": generation_smoke_passed, "full_inference_verified": full_inference_verified, "failure_owner": failure_summary.get("failure_owner") or "", "failure_reason": failure_summary.get("failure_reason") or "", "message_redacted": redact_text(failure_message)[:1200], "target_space_hash": eval_safe_hash(target_space, "space"), "space_created": bool(target_space), "space_runtime_stage": space_runtime.get("stage"), "space_runtime_hardware": space_runtime.get("hardware") or space_runtime.get("requested_hardware"), }, "timeline": eval_compact_timeline(events), "hardware": { "selected": hardware_strategy.get("selected_hardware") or state.get("selected_hardware") or "", "strategy": hardware_strategy.get("strategy") or hardware_attempts.get("strategy") or "", "attempts": hardware_strategy.get("attempts") or hardware_attempts.get("attempts") or state.get("hardware_attempts") or [], "single_space_per_run": True, }, "pi": { "requested_model": os.environ.get("PI_MODEL") or state.get("pi_model") or "", "effective_model": pi_resolution.get("effective_model") or pi_resolution.get("observed_model") or "", "model_mismatch": bool(pi_resolution.get("mismatch")), "build_log_tail_redacted": eval_redacted_tail(run_dir / "logs" / "pi_output.txt"), "diagnosis_log_tail_redacted": eval_redacted_tail(run_dir / "logs" / "pi_diagnosis_output.txt"), }, "recovery": { "entered": any(str(e.get("step", "")).startswith(("repair", "failure_diagnosis", "pi_diagnosis", "technical_blocker")) for e in events), "decision": repair_decision.get("action") or repair_decision.get("decision") or "", "confidence": repair_decision.get("confidence") or "", "blockage": bool(blockage), "blockage_category": blockage.get("category") or blockage.get("reason") or "", }, "validation": { "health_passed": health_passed, "generation_smoke_passed": generation_smoke_passed, "full_inference_verified": full_inference_verified, "generation_smoke_status": generation_smoke.get("status") or "", "latency_seconds": generation_smoke.get("latency_seconds") or generation_smoke.get("observed_latency_seconds"), "expected_output_type": generation_smoke.get("expected_output_type") or os.environ.get("EXPECTED_OUTPUT_TYPE", ""), "inference_gate_status": inference_gate.get("status") or "", "manual_hardware_required": bool(inference_gate.get("manual_hardware_required")), "failure_type": failure_summary.get("failure_type") or "", "failure_owner": failure_summary.get("failure_owner") or "", "failure_reason": failure_summary.get("failure_reason") or "", "missing_argument": failure_summary.get("missing_argument") or "", }, "artifacts": { "present_count": len(manifest.get("present_paths") or []), "present_paths": sorted(str(p) for p in (manifest.get("present_paths") or []))[:120], }, "privacy": { "tokens_stored": False, "raw_prompts_stored": False, "generated_code_stored": False, "target_space_redacted": True, "user_bucket_redacted": True, "redacted_tails_enabled": os.environ.get("ASF_EVAL_INCLUDE_REDACTED_TAILS", "").strip().lower() in {"1", "true", "yes", "on"}, }, } day = (finished_at or now())[:10] try: yyyy, mm, dd = day.split("-") except Exception: yyyy, mm, dd = now()[:10].split("-") dest = run_dir / "eval_record.json" write_json(dest, record) events_dest = run_dir / "events_compact.jsonl" events_dest.write_text("".join(json.dumps(item, ensure_ascii=False) + "\n" for item in record["timeline"]), encoding="utf-8") return {"enabled": True, "written": True, "path": str(dest), "publish_mode": "backend"} 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) 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 = {} failure_state = { **existing_state, "run_id": os.environ.get("RUN_ID"), "kind": existing_state.get("kind") or "universal_model_card_builder", "status": status, "message": message, "updated_at": now(), "details": safe, } # Preserve the target Space when the worker fails after repository creation. target_space = failure_state.get("target_space") or os.environ.get("TARGET_SPACE_ID") or "" if target_space: failure_state["target_space"] = target_space failure_state["target_space_url"] = f"https://huggingface.co/spaces/{target_space}" write_json(run_dir / "state.json", failure_state) report = f"""# Agentic Space Factory — model Article Reproduction Report Status: **{status}** {message} Target Space: {failure_state.get('target_space_url') or failure_state.get('target_space') or 'not created / unknown'} ```json {json.dumps(safe, indent=2, ensure_ascii=False)} ``` """ (run_dir / "report.md").write_text(report, encoding="utf-8") write_artifact_manifest(run_dir, events_path=events_path, reason="failure") try: publish_eval_record(run_dir, phase="failure", events_path=events_path) except Exception: pass raise SystemExit(1) def run_cmd(cmd: list[str], *, cwd: Path | None = None, env: dict | None = None, timeout: int = 600): result = subprocess.run(cmd, cwd=str(cwd) if cwd else None, env=env, text=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, timeout=timeout) return result.returncode, redact_text(result.stdout) def sync_pi_traces(run_dir: Path, *, emit_event: bool = False, events_path: Path | None = None): """Best-effort incremental copy of Pi session JSONL traces into the run bucket.""" traces_dir = Path.home() / ".pi" / "agent" / "sessions" raw_dir = run_dir / "traces" / "raw" redacted_dir = run_dir / "traces" / "redacted" raw_dir.mkdir(parents=True, exist_ok=True) redacted_dir.mkdir(parents=True, exist_ok=True) count = 0 if traces_dir.exists(): for path in traces_dir.rglob("*.jsonl"): rel = path.relative_to(traces_dir) target_raw = raw_dir / rel target_raw.parent.mkdir(parents=True, exist_ok=True) text = path.read_text(encoding="utf-8", errors="ignore") target_raw.write_text(text, encoding="utf-8") target_redacted = redacted_dir / rel target_redacted.parent.mkdir(parents=True, exist_ok=True) target_redacted.write_text(redact_text(text), encoding="utf-8") count += 1 if emit_event and events_path: append_event(events_path, "pi_traces", "running", "Synchronized Pi traces while Pi is running", {"count": count}) return count def run_cmd_streaming( cmd: list[str], *, cwd: Path | None = None, env: dict | None = None, timeout: int = 600, live_log_path: Path | None = None, events_path: Path | None = None, run_dir: Path | None = None, step: str = "pi_run", trace_phase: str | None = None, ): """Run a command while incrementally writing stdout to the bucket. subprocess.run() only returns output at the end. Pi can work for many minutes, so stream stdout to logs/pi_live_output.txt and periodically copy Pi session JSONL traces into traces/redacted while the process is still alive. """ started = time.monotonic() trace_phase = trace_phase or step last_event = 0.0 last_trace_sync = 0.0 lines: list[str] = [] if live_log_path: live_log_path.parent.mkdir(parents=True, exist_ok=True) live_log_path.write_text("", encoding="utf-8") if run_dir: write_agent_trace_record(run_dir, phase=trace_phase, event="command_started", status="started", message="Pi command started", data={"step": step, "cwd": str(cwd) if cwd else ""}) proc = subprocess.Popen( cmd, cwd=str(cwd) if cwd else None, env=env, text=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, bufsize=1, ) try: assert proc.stdout is not None while True: if timeout and time.monotonic() - started > timeout: proc.kill() message = f"Command timed out after {timeout}s" lines.append(message) if live_log_path: with live_log_path.open("a", encoding="utf-8") as f: f.write(message + "\n") return 124, redact_text("\n".join(lines)) line = proc.stdout.readline() if line: clean = redact_text(line.rstrip("\n")) lines.append(clean) if live_log_path: with live_log_path.open("a", encoding="utf-8") as f: f.write(clean + "\n") now_m = time.monotonic() if events_path and now_m - last_event > 8: output_tail = "\n".join(lines[-20:])[-2000:] append_event(events_path, step, "running", "Pi is still working", {"output_tail": output_tail}) if run_dir: write_agent_trace_record(run_dir, phase=trace_phase, event="stdout_tail", status="running", message="Pi is still working", text=output_tail, data={"step": step}) last_event = now_m if run_dir and now_m - last_trace_sync > 15: sync_pi_traces(run_dir, emit_event=False) last_trace_sync = now_m continue if proc.poll() is not None: remainder = proc.stdout.read() if proc.stdout else "" if remainder: for raw in remainder.splitlines(): clean = redact_text(raw) lines.append(clean) if live_log_path: with live_log_path.open("a", encoding="utf-8") as f: f.write(clean + "\n") break time.sleep(0.2) finally: if run_dir: sync_pi_traces(run_dir, emit_event=False) output = redact_text("\n".join(lines)) if run_dir: write_agent_trace_record(run_dir, phase=trace_phase, event="command_finished", status="success" if int(proc.returncode or 0) == 0 else "failed", message="Pi command finished", text=output[-20000:], data={"step": step, "returncode": int(proc.returncode or 0)}, artifacts=[str(live_log_path.relative_to(run_dir)) if live_log_path and str(live_log_path).startswith(str(run_dir)) else ""]) return int(proc.returncode or 0), output def install_python_deps(events_path: Path): append_event(events_path, "dependencies", "started", "Installing Python worker dependencies") code, out = run_cmd([sys.executable, "-m", "pip", "install", "-q", "--upgrade", "huggingface_hub>=1.0.0", "gradio_client>=2.0.0", "requests>=2.31.0"], timeout=600) if code != 0: append_event(events_path, "dependencies", "failed", "Python dependency installation failed", {"output_tail": out[-4000:]}) raise RuntimeError(out) append_event(events_path, "dependencies", "success", "Python worker dependencies installed") def ensure_node(events_path: Path): node = shutil.which("node") npm = shutil.which("npm") if node and npm: _, node_v = run_cmd([node, "--version"], timeout=30) _, npm_v = run_cmd([npm, "--version"], timeout=30) append_event(events_path, "node", "success", "Node/npm already available", {"node": node_v.strip(), "npm": npm_v.strip()}) return append_event(events_path, "node", "started", "Installing nodejs/npm through apt-get") code, out = run_cmd(["bash", "-lc", "apt-get update -qq && apt-get install -y -qq nodejs npm"], timeout=600) if code != 0: append_event(events_path, "node", "failed", "Could not install nodejs/npm", {"output_tail": out[-4000:]}) raise RuntimeError(out) append_event(events_path, "node", "success", "Installed nodejs/npm") def install_pi(events_path: Path): ensure_node(events_path) append_event(events_path, "pi_install", "started", "Installing Pi coding agent from npm") code, out = run_cmd(["npm", "install", "-g", "@mariozechner/pi-coding-agent"], timeout=900) if code != 0: append_event(events_path, "pi_install", "failed", "Pi npm installation failed", {"output_tail": out[-4000:]}) raise RuntimeError(out) code, version = run_cmd(["pi", "--version"], timeout=60) append_event(events_path, "pi_install", "success", "Pi installed", {"version_output": version.strip()[-300:]}) def configure_pi(events_path: Path, model: str): pi_dir = Path.home() / ".pi" / "agent" pi_dir.mkdir(parents=True, exist_ok=True) token = os.environ.get("HF_TOKEN") or os.environ.get("HUGGING_FACE_HUB_TOKEN") or "" if token: os.environ.setdefault("HF_TOKEN", token) os.environ.setdefault("HUGGING_FACE_HUB_TOKEN", token) (pi_dir / "auth.json").write_text(json.dumps({"huggingface": {"type": "api_key", "key": token}}, indent=2), encoding="utf-8") (pi_dir / "settings.json").write_text(json.dumps({"model": model, "provider": "huggingface", "autoRun": True, "autoApply": True}, indent=2), encoding="utf-8") append_event(events_path, "pi_config", "success" if token else "warning", "Configured Pi with Hugging Face provider token context", {"model": model, "hf_token_present": bool(token), "token_value": "[REDACTED]" if token else ""}) def normalize_pi_model_name(value: str | None) -> str: raw = (value or "").strip().lower() # Pi/provider traces may report either a full Hub id (moonshotai/Kimi-K2-...) # or only the served model name (Kimi-K2-...). Compare on the model leaf so # provider fallbacks are detected without false positives from missing owners. if "/" in raw: raw = raw.rsplit("/", 1)[-1] return re.sub(r"[^a-z0-9]+", "", raw) 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"\bmoonshotai/[A-Za-z0-9_.-]+", r"\bKimi[-_/A-Za-z0-9.]+", r"\bzai-org/[A-Za-z0-9_.-]+", r"\bGLM[-_/A-Za-z0-9.]+", r"\bdeepseek-ai/[A-Za-z0-9_.-]+", r"\bDeepSeek[-_/A-Za-z0-9.]+", r"\bClaude[-_/A-Za-z0-9.]+", r"\bGPT[-_/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) write_agent_trace_record(run_dir, phase="initial_build", event="pi_sessions_collected", status="success", message="Synchronized Pi session traces into the run-level trace folder", data={"count": count}, artifacts=["traces/raw/agent_trace.jsonl", "traces/redacted/agent_trace.jsonl"]) append_event(events_path, "traces", "success", "Collected Pi traces", {"count": count, "agent_trace": "traces/redacted/agent_trace.jsonl"}) write_artifact_manifest(run_dir, reason="pi_traces_collected") return count def sanitize_model_id(model_id: str) -> str: model_id = (model_id or DEFAULT_MODEL_ID).strip().replace("https://huggingface.co/", "") model_id = model_id.split("?", 1)[0].strip("/") if not re.match(r"^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$", model_id): raise ValueError("MODEL_ID must look like owner/model-name") return model_id def make_gradio_client(target_space_id: str, token: str): import inspect from gradio_client import Client params = inspect.signature(Client).parameters if "token" in params: return Client(target_space_id, token=token) if "hf_token" in params: return Client(target_space_id, hf_token=token) if "api_key" in params: return Client(target_space_id, api_key=token) if "headers" in params: return Client(target_space_id, headers={"Authorization": f"Bearer {token}"}) return Client(target_space_id) def api_names_from_schema(schema) -> list[str]: names: list[str] = [] if isinstance(schema, dict): endpoints = schema.get("named_endpoints") or schema.get("endpoints") or {} if isinstance(endpoints, dict): for key, value in endpoints.items(): if isinstance(key, str) and key.startswith("/"): names.append(key) if isinstance(value, dict): api_name = value.get("api_name") if isinstance(api_name, str) and api_name.startswith("/"): names.append(api_name) if isinstance(schema.get("dependencies"), list): for dep in schema["dependencies"]: if isinstance(dep, dict): api_name = dep.get("api_name") if isinstance(api_name, str): names.append(api_name if api_name.startswith("/") else f"/{api_name}") 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 endpoint_parameters_for_smoke(endpoint) -> list[dict]: """Return Gradio endpoint parameters in call order with names/defaults. gradio_client schema formats vary across versions. Keep this permissive and preserve ordering because client.predict uses positional arguments. """ params = None def find_params(obj): nonlocal params if params is not None: return if isinstance(obj, dict): for key in ("parameters", "inputs"): value = obj.get(key) if isinstance(value, list) and value: params = value return for value in obj.values(): if isinstance(value, (dict, list)): find_params(value) elif isinstance(obj, list): for item in obj: find_params(item) find_params(endpoint) out = [] for i, param in enumerate(params or []): if isinstance(param, dict): component = param.get("component") or param.get("component_type") or {} if not isinstance(component, dict): component = {} name = ( param.get("parameter_name") or param.get("parameterName") or param.get("name") or param.get("label") or component.get("label") or component.get("name") or f"arg{i}" ) default = None has_default = False for key in ("parameter_default", "default", "value"): if key in param: default = param.get(key) has_default = True break if not has_default: for key in ("value", "default"): if key in component: default = component.get(key) has_default = True break if isinstance(default, str) and default.strip().lower() in {"none", "null", ""} and not param.get("parameter_has_default"): has_default = False default = None if "parameter_has_default" in param: has_default = bool(param.get("parameter_has_default")) required = bool(param.get("required", not has_default)) out.append({ "name": str(name).strip(), "required": required, "has_default": has_default, "default": default, "component": component.get("type") or component.get("component") or param.get("component_type") or param.get("type") or "", }) elif isinstance(param, str): out.append({"name": param.strip() or f"arg{i}", "required": True, "has_default": False, "default": None, "component": ""}) return out def smoke_value_for_parameter(param: dict, expected_output_type: str): name = str(param.get("name") or "").strip().lower().replace(" ", "_").replace("-", "_") component = str(param.get("component") or "").lower() if param.get("has_default"): return param.get("default") if name in {"prompt", "text", "query", "input", "instruction"} or "prompt" in name and "negative" not in name: return default_smoke_args(expected_output_type)[0] if name in {"negative_prompt", "negative", "negative_text"} or "negative" in name: return "" if name in {"seed", "random_seed"} or name.endswith("_seed"): return 42 if name in {"height", "width", "image_height", "image_width"}: return 512 if name in {"num_inference_steps", "inference_steps", "steps", "num_steps"} or "step" in name: return 4 if name in {"guidance_scale", "cfg_scale", "scale"} or "guidance" in name: return 0.0 if "bool" in component or name.startswith("enable_") or name.startswith("use_"): return False if "number" in component or "slider" in component: return 0 return "" def coerce_smoke_value(value, param: dict): name = str(param.get("name") or "").strip().lower().replace(" ", "_").replace("-", "_") component = str(param.get("component") or "").lower() if value is None: return value if isinstance(value, str): stripped = value.strip() if name in {"height", "width", "image_height", "image_width", "seed", "random_seed", "num_inference_steps", "inference_steps", "steps", "num_steps"} or name.endswith("_seed") or "step" in name: try: return int(float(stripped)) except Exception: return value if name in {"guidance_scale", "cfg_scale", "scale"} or "guidance" in name or "slider" in component or "number" in component: try: number = float(stripped) return int(number) if number.is_integer() and name not in {"guidance_scale", "cfg_scale", "scale"} else number except Exception: return value return value def build_generation_smoke_args(endpoint, expected_output_type: str) -> tuple[list, list[dict]]: params = endpoint_parameters_for_smoke(endpoint) if not params: return default_smoke_args(expected_output_type), [] return [coerce_smoke_value(smoke_value_for_parameter(param, expected_output_type), param) for param in params], params def read_inference_contract(workspace: Path | None) -> dict: if not workspace: return {} path = workspace / "INFERENCE_CONTRACT.json" if not path.exists(): return {} try: data = json.loads(path.read_text(encoding="utf-8")) return data if isinstance(data, dict) else {} except Exception: return {} def contract_smoke_test_payload(contract: dict, api_name: str, endpoint, expected_output_type: str) -> dict | None: smoke = contract.get("smoke_test") if isinstance(contract, dict) else None if not isinstance(smoke, dict): return None params = endpoint_parameters_for_smoke(endpoint) smoke_api = normalize_api_name(str(smoke.get("api_name") or smoke.get("endpoint") or api_name)) raw_args = smoke.get("args", []) raw_kwargs = smoke.get("kwargs", {}) if raw_kwargs is None: raw_kwargs = {} if not isinstance(raw_kwargs, dict): raw_kwargs = {} if isinstance(raw_args, list): args = [coerce_smoke_value(v, params[i] if i < len(params) else {}) for i, v in enumerate(raw_args)] if params and len(args) < len(params): generated, _ = build_generation_smoke_args(endpoint, expected_output_type) args.extend(generated[len(args):]) return {"api_name": smoke_api, "test_args": args, "test_kwargs": raw_kwargs, "parameters": params, "source": "inference_contract_smoke_test"} if isinstance(raw_args, dict): generated, _ = build_generation_smoke_args(endpoint, expected_output_type) args = [] 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)) if not params: # No schema: keep deterministic insertion order from the contract. args = list(raw_args.values()) return {"api_name": smoke_api, "test_args": args, "test_kwargs": raw_kwargs, "parameters": params, "source": "inference_contract_smoke_test"} return None def validation_health_passed(validation: dict | None) -> bool: if not isinstance(validation, dict) or validation.get("status") != "success": return False method = str(validation.get("method") or "").lower() validator = str(validation.get("validator") or "").lower() api_name = normalize_api_name(str(validation.get("api_name") or "")) if validation.get("api_name") else "" return method in {"http_health", "gradio"} or validator in {"http_get_health", "gradio_client"} or api_name == "/health" def classify_generation_smoke_error(error: Exception | str) -> dict: text = str(error or "") lowered = text.lower() payload = {"failure_type": "generation_smoke_error", "failure_owner": "unknown"} marker = "no value provided for required argument:" if marker in lowered: missing = text.split(":", 1)[-1].strip().strip("'\"") payload.update({ "failure_type": "validator_request_error", "failure_owner": "factory_validator", "retryable_with_schema_payload": True, "missing_argument": missing, }) elif any(m in lowered for m in ["timeout", "timed out", "read operation timed out"]): payload.update({"failure_type": "timeout", "failure_owner": "infra_or_generated_space", "retryable_with_longer_timeout": True}) elif any(m in lowered for m in ["traceback", "runtimeerror", "attributeerror", "valueerror", "cuda", "out of memory", "exception"]): payload.update({"failure_type": "app_runtime_error", "failure_owner": "generated_space", "repair_candidate": True}) elif any(m in lowered for m in ["invalid state", "runtime_error", "build_error"]): payload.update({"failure_type": "space_runtime_state", "failure_owner": "generated_space", "repair_candidate": True}) return payload 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 = [] type_hints = [] text_values = [] def visit(obj): type_hints.append(type(obj).__name__.lower()) if isinstance(obj, (str, Path)): value = str(obj) if value: paths.append(value) text_values.append(value) elif isinstance(obj, dict): for key, value in obj.items(): key_l = str(key).lower() if key_l in {"path", "name", "url", "file", "filepath"}: visit(value) elif key_l in {"mime_type", "mime", "type", "format"} and value: type_hints.append(str(value).lower()) elif key_l in {"image", "video", "audio", "text"}: type_hints.append(key_l) visit(value) elif isinstance(value, (dict, list, tuple)): visit(value) elif isinstance(obj, (list, tuple)): for item in obj: visit(item) else: # PIL Images, numpy arrays returned through Gradio objects, etc. module = getattr(type(obj), "__module__", "") if module: type_hints.append(module.lower()) visit(result) lower_paths = [str(p).lower() for p in paths] hints = " ".join(type_hints + lower_paths) info["detected_paths"] = paths[:20] info["type_hints"] = type_hints[:30] if expected == "any": return result is not None, info image_ext = (".png", ".jpg", ".jpeg", ".webp", ".gif", ".bmp", ".tiff") video_ext = (".mp4", ".mov", ".webm", ".avi", ".mkv") audio_ext = (".wav", ".mp3", ".flac", ".ogg", ".m4a") if expected == "text": if isinstance(result, str) and bool(result.strip()): return True, info return any(isinstance(x, str) and x.strip() and not Path(x).suffix.lower() in image_ext + video_ext + audio_ext for x in text_values), info if expected == "image": return any(p.endswith(image_ext) for p in lower_paths) or any(h in hints for h in ["image", "pil.", "png", "jpeg", "jpg", "webp"]), info if expected == "video": return any(p.endswith(video_ext) for p in lower_paths) or any(h in hints for h in ["video", "mp4", "webm", "moviepy"]), info if expected == "audio": return any(p.endswith(audio_ext) for p in lower_paths) or any(h in hints for h in ["audio", "wav", "soundfile"]), 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, workspace: Path | None = None): api_name = "/generate" 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) contract = read_inference_contract(workspace) contract_primary = normalize_api_name(str(contract.get("primary_api_name") or "")) if contract else "" if contract_primary and contract_primary in discovered and contract_primary != "/health": api_name = contract_primary endpoint = endpoint_schema_for_api(schema, api_name) contract_payload = contract_smoke_test_payload(contract, api_name, endpoint, expected_output_type) if contract_payload and contract_payload.get("api_name") in discovered: api_name = contract_payload["api_name"] endpoint = endpoint_schema_for_api(schema, api_name) contract_payload = contract_smoke_test_payload(contract, api_name, endpoint, expected_output_type) endpoint_parameters = endpoint_parameter_names(endpoint) if contract_payload: test_args = contract_payload.get("test_args") or [] test_kwargs = contract_payload.get("test_kwargs") or {} smoke_parameters = contract_payload.get("parameters") or endpoint_parameters_for_smoke(endpoint) smoke_source = contract_payload.get("source") or "inference_contract_smoke_test" else: test_args, smoke_parameters = build_generation_smoke_args(endpoint, expected_output_type) smoke_source = "gradio_schema" if smoke_parameters else "default_fallback" smoke_payload = {"api_name": api_name, "test_args": test_args, "test_kwargs": test_kwargs, "parameters": smoke_parameters, "source": smoke_source, "contract_present": bool(contract)} write_json(run_dir / "tests" / "generation_smoke_payload.json", smoke_payload) 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}) 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, "smoke_payload_source": smoke_source, "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"], "recommended_zerogpu_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 # generated slugs are ASCII and hyphen-friendly. return f"https://{owner}-{name}.hf.space".replace("_", "-").lower() def runtime_to_dict(runtime) -> dict: payload = {} for attr in ["stage", "hardware", "requested_hardware", "sleep_time", "storage", "gc_timeout"]: value = getattr(runtime, attr, None) payload[attr] = getattr(value, "value", value) return {k: str(v) if v is not None else None for k, v in payload.items()} def write_space_runtime(api, target_space_id: str, token: str, run_dir: Path, events_path: Path, attempt: int | None = None) -> dict: try: runtime = api.get_space_runtime(repo_id=target_space_id, token=token) payload = runtime_to_dict(runtime) payload["attempt"] = attempt write_json(run_dir / "space_runtime.json", payload) return payload except Exception as exc: payload = {"error": str(exc)[:2000], "attempt": attempt} write_json(run_dir / "space_runtime.json", payload) append_event(events_path, "space_runtime", "warning", "Could not fetch Space runtime", payload) return payload RESTART_GUARDRAIL_STATE = { "restart": {"count": 0, "last_ts": 0.0}, "factory_reboot": {"count": 0, "last_ts": 0.0}, "same_code_upload": {"count": 0, "last_ts": 0.0}, } RESTART_GUARDRAIL_LIMITS = { "restart": 1, "factory_reboot": 1, "same_code_upload": 1, } RESTART_GUARDRAIL_COOLDOWN_SECONDS = int(os.environ.get("SPACE_FACTORY_RESTART_COOLDOWN_SECONDS", "240")) def runtime_stage_value(runtime_payload: dict | None) -> str: return str((runtime_payload or {}).get("stage") or (runtime_payload or {}).get("runtime_stage") or "").lower() def restart_guardrail_allows(action: str, events_path: Path, *, reason: str = "") -> bool: state = RESTART_GUARDRAIL_STATE.setdefault(action, {"count": 0, "last_ts": 0.0}) limit = RESTART_GUARDRAIL_LIMITS.get(action, 1) now = time.time() if state["count"] >= limit: append_event(events_path, "factory_rebuild", "warning", f"Blocked {action}: restart budget exhausted", {"action": action, "limit": limit, "reason": reason}) return False if state["last_ts"] and now - state["last_ts"] < RESTART_GUARDRAIL_COOLDOWN_SECONDS: append_event(events_path, "factory_rebuild", "warning", f"Blocked {action}: restart cooldown active", {"action": action, "cooldown_seconds": RESTART_GUARDRAIL_COOLDOWN_SECONDS, "elapsed_seconds": round(now - state["last_ts"], 1), "reason": reason}) return False state["count"] += 1 state["last_ts"] = now return True def safe_restart_space(api, target_space_id: str, token: str, run_dir: Path, events_path: Path, *, factory_reboot: bool = False, reason: str = "", require_logs_checked: bool = True) -> bool: """Restart a Space with strict guardrails. Hugging Face exposes HfApi.restart_space(repo_id=..., factory_reboot=True) for a factory rebuild and restart_space(..., factory_reboot=False) for a normal restart. The Factory only calls it through this wrapper so we do not create restart loops or mask actionable build errors. """ action = "factory_reboot" if factory_reboot else "restart" if require_logs_checked: logs_dir = run_dir / "logs" if not ((logs_dir / "space_logs_build.txt").exists() or (logs_dir / "space_logs_runtime.txt").exists()): append_event(events_path, "factory_rebuild", "warning", f"Skipped {action}: logs have not been collected yet", {"reason": reason}) return False runtime_before = write_space_runtime(api, target_space_id, token, run_dir, events_path) stage = runtime_stage_value(runtime_before) if stage in {"building", "buildqueued", "build_queued", "starting", "pending"}: append_event(events_path, "factory_rebuild", "warning", f"Skipped {action}: Space is already busy", {"runtime": runtime_before, "reason": reason}) return False if not restart_guardrail_allows(action, events_path, reason=reason): return False try: append_event(events_path, "factory_rebuild", "started", f"Requesting {'factory reboot' if factory_reboot else 'Space restart'} via Hugging Face API", {"runtime_before": runtime_before, "reason": reason}) result = api.restart_space(repo_id=target_space_id, token=token, factory_reboot=factory_reboot) payload = {"factory_reboot": factory_reboot, "reason": reason, "result": str(result)[:1000]} write_json(run_dir / ("space_factory_reboot.json" if factory_reboot else "space_restart.json"), payload) append_event(events_path, "factory_rebuild", "success", f"{'Factory reboot' if factory_reboot else 'Space restart'} requested", payload) return True except TypeError: # Older huggingface_hub versions may not expose factory_reboot. Do not # loop; fall back to a plain restart only for non-factory requests. if factory_reboot: append_event(events_path, "factory_rebuild", "warning", "Installed huggingface_hub does not support restart_space(factory_reboot=True)", {"reason": reason}) return False try: result = api.restart_space(repo_id=target_space_id, token=token) append_event(events_path, "factory_rebuild", "success", "Space restart requested", {"reason": reason, "result": str(result)[:1000]}) return True except Exception as exc: append_event(events_path, "factory_rebuild", "failed", "Space restart API call failed", {"reason": reason, **exception_payload(exc)}) return False except Exception as exc: append_event(events_path, "factory_rebuild", "failed", f"{'Factory reboot' if factory_reboot else 'Space restart'} API call failed", {"reason": reason, **exception_payload(exc)}) return False def safe_same_code_reupload(api, workspace: Path, target_space_id: str, token: str, run_dir: Path, events_path: Path, *, reason: str = "") -> bool: """Re-upload the current workspace once, with restart-loop guardrails.""" if not restart_guardrail_allows("same_code_upload", events_path, reason=reason): return False runtime_before = write_space_runtime(api, target_space_id, token, run_dir, events_path) stage = runtime_stage_value(runtime_before) if stage in {"building", "buildqueued", "build_queued", "starting", "pending"}: append_event(events_path, "factory_rebuild", "warning", "Skipped same-code upload: Space is already building/starting", {"runtime": runtime_before, "reason": reason}) return False upload_workspace(api, workspace, target_space_id, token, run_dir, events_path) append_event(events_path, "factory_rebuild", "success", "Same-code workspace uploaded under restart guardrails", {"reason": reason, "runtime_before": runtime_before}) return True def collect_space_logs(target_space_id: str, token: str, run_dir: Path, events_path: Path): """Collect Space build/runtime logs with explicit capability detection and an index. This is intentionally more structured than the old best-effort collector: - runtime state is always captured when possible; - runtime/build log collection first uses HfApi.get_space_logs when available; - if the installed huggingface_hub lacks that method, REST fallbacks are attempted; - legacy text files are still written, but availability/quality lives in space_logs_index.json. """ from huggingface_hub import HfApi import json as _json import sys as _sys try: import huggingface_hub as _hf_hub hub_version = getattr(_hf_hub, "__version__", None) except Exception: hub_version = None logs_dir = run_dir / "logs" logs_dir.mkdir(parents=True, exist_ok=True) api = HfApi(token=token) collected_at = now() written = [] index = { "schema_version": "1.0", "target_space_id": target_space_id, "collected_at": collected_at, "huggingface_hub_version": hub_version, "python_version": _sys.version.split()[0], "capabilities": { "hfapi_get_space_runtime": hasattr(api, "get_space_runtime"), "hfapi_get_space_logs": hasattr(api, "get_space_logs"), "rest_fallback": True, }, "entries": {}, } def _tail(text: str, n: int = 1200) -> str: return (text or "")[-n:] def _stringify_log_response(out) -> str: if out is None: return "" if isinstance(out, (list, tuple)): return "\n".join(str(x) for x in out) if isinstance(out, (dict, list)): return _json.dumps(out, ensure_ascii=False, indent=2) return str(out) def _write_text(filename: str, text: str) -> int: path = logs_dir / filename path.write_text(text or "", encoding="utf-8") return path.stat().st_size if path.exists() else 0 def _entry(key: str, *, filename: str, available: bool, source: str, quality: str, reason: str = "", error: str = "", text: str = ""): size = _write_text(filename, text) payload = { "available": bool(available), "path": f"logs/{filename}", "source": source, "quality": quality, "reason": reason, "error": error[:1500] if error else "", "size_bytes": size, "tail": _tail(text), } index["entries"][key] = payload written.append({ "file": filename, "source": source, "available": bool(available), "quality": quality, "reason": reason, "error": error[:1000] if error else "", "returncode": 0 if available else 1, "tail": _tail(text, 1000), }) return payload def _unavailable_text(kind: str, reason: str, error: str = "") -> str: return ( f"[ASF_LOG_UNAVAILABLE]\n" f"kind={kind}\n" f"reason={reason}\n" f"target_space_id={target_space_id}\n" f"huggingface_hub_version={hub_version}\n" f"error={error[:1500] if error else ''}\n" "See logs/space_logs_index.json for structured availability metadata.\n" ) # Runtime state / snapshot. This is not a log stream, but it is critical context for # BUILD_ERROR/RUNTIME_ERROR triage and is always tracked separately from log streams. try: runtime_obj = api.get_space_runtime(repo_id=target_space_id, token=token) runtime_payload = runtime_to_dict(runtime_obj) runtime_payload.update({"target_space_id": target_space_id, "collected_at": collected_at}) runtime_text = _json.dumps(runtime_payload, indent=2, ensure_ascii=False) _entry("runtime_snapshot", filename="space_runtime_snapshot.json", available=True, source="HfApi.get_space_runtime", quality="snapshot", text=runtime_text) # Legacy path retained because older reports/tests point to this file. _entry("runtime_state_legacy", filename="space_runtime_state.json", available=True, source="HfApi.get_space_runtime", quality="snapshot", text=runtime_text) except Exception as exc: err = f"{type(exc).__name__}: {str(exc)[:1500]}" _entry("runtime_snapshot", filename="space_runtime_snapshot.json", available=False, source="HfApi.get_space_runtime", quality="unavailable", reason="runtime_snapshot_error", error=err, text=_unavailable_text("runtime_snapshot", "runtime_snapshot_error", err)) _entry("runtime_state_legacy", filename="space_runtime_state.json", available=False, source="HfApi.get_space_runtime", quality="unavailable", reason="runtime_snapshot_error", error=err, text=_unavailable_text("runtime_state", "runtime_snapshot_error", err)) def _collect_via_hfapi(kind: str): if not hasattr(api, "get_space_logs"): return None, "hub_method_unavailable", "" try: out = api.get_space_logs(repo_id=target_space_id, token=token, build=(kind == "build")) text = _stringify_log_response(out) if text.strip(): return text, "", "" return "", "empty_response", "" except Exception as exc: return None, "hfapi_error", f"{type(exc).__name__}: {str(exc)[:1500]}" def _collect_via_rest(kind: str): # Private/unstable endpoints vary across hub deployments, so this is best-effort. # We still try because it can recover logs when the Python SDK lacks get_space_logs. try: import requests except Exception as exc: return None, "requests_unavailable", f"{type(exc).__name__}: {str(exc)[:1500]}" headers = {"Authorization": f"Bearer {token}", "Accept": "text/plain,application/json,*/*"} if token else {"Accept": "text/plain,application/json,*/*"} encoded = target_space_id candidates = [ f"https://huggingface.co/api/spaces/{encoded}/logs?build={'true' if kind == 'build' else 'false'}", f"https://huggingface.co/api/spaces/{encoded}/logs/{kind}", ] last_error = "" for url in candidates: try: response = requests.get(url, headers=headers, timeout=20) body = response.text or "" if response.ok and body.strip(): ctype = response.headers.get("content-type", "") return body, "", f"{url} ({response.status_code}, {ctype})" last_error = f"{url} returned {response.status_code}: {body[:500]}" except Exception as exc: last_error = f"{url} failed: {type(exc).__name__}: {str(exc)[:500]}" return None, "rest_logs_unavailable", last_error def _collect_stream(kind: str, filename: str): attempted = [] text, reason, error = _collect_via_hfapi(kind) attempted.append({"method": "HfApi.get_space_logs", "reason": reason, "error": error[:500] if error else ""}) if text is not None and text.strip(): return _entry(kind, filename=filename, available=True, source="HfApi.get_space_logs", quality="full", text=text) rest_text, rest_reason, rest_error = _collect_via_rest(kind) attempted.append({"method": "hf_rest_logs", "reason": rest_reason, "error": rest_error[:500] if rest_error else ""}) if rest_text is not None and rest_text.strip(): source = "hf_rest_logs" if rest_error and rest_error.startswith("https://"): source = rest_error return _entry(kind, filename=filename, available=True, source=source, quality="full", text=rest_text) # Keep legacy files present, but explicitly mark them as unavailable rather than # pretending an SDK AttributeError is a Space runtime/build log. This also avoids # the obsolete CLI path that emitted "invalid choice: spaces". final_reason = rest_reason or reason or "logs_unavailable" final_error = rest_error or error or "" entry = _entry(kind, filename=filename, available=False, source="log_collection", quality="unavailable", reason=final_reason, error=final_error, text=_unavailable_text(kind, final_reason, final_error)) entry["attempted_methods"] = attempted return entry runtime_entry = _collect_stream("runtime", "space_logs_runtime.txt") build_entry = _collect_stream("build", "space_logs_build.txt") def _first_error_from(text: str) -> str: if not text: return "" lines = text.splitlines() lowered = [line.lower() for line in lines] markers = ("traceback", "exception", "error", "exit code", "failed", "runtimeerror", "attributeerror", "modulenotfounderror") for idx, line in enumerate(lowered): if any(marker in line for marker in markers): start = max(0, idx - 3) end = min(len(lines), idx + 12) return "\n".join(lines[start:end])[:4000] return "" runtime_text = (logs_dir / "space_logs_runtime.txt").read_text(encoding="utf-8", errors="ignore") if runtime_entry.get("available") and (logs_dir / "space_logs_runtime.txt").exists() else "" build_text = (logs_dir / "space_logs_build.txt").read_text(encoding="utf-8", errors="ignore") if build_entry.get("available") and (logs_dir / "space_logs_build.txt").exists() else "" first_error = _first_error_from(runtime_text) or _first_error_from(build_text) log_quality = "full" if runtime_entry.get("available") or build_entry.get("available") else "snapshot_only" if index["entries"].get("runtime_snapshot", {}).get("available") else "unavailable" diagnostics = { "schema_version": "1.0", "target_space_id": target_space_id, "collected_at": collected_at, "log_quality": log_quality, "runtime_logs_available": bool(runtime_entry.get("available")), "build_logs_available": bool(build_entry.get("available")), "first_error": first_error, "diagnosis_log_source": "runtime_or_build_logs" if first_error else ("runtime_snapshot" if log_quality == "snapshot_only" else "none"), } diagnostics_text = _json.dumps(diagnostics, ensure_ascii=False, indent=2) _entry("diagnostics", filename="space_log_diagnostics.json", available=True, source="asf_log_collector", quality=log_quality, text=diagnostics_text) index["log_quality"] = log_quality index["runtime_logs_available"] = bool(runtime_entry.get("available")) index["build_logs_available"] = bool(build_entry.get("available")) index["first_error"] = first_error[:1000] index_text = _json.dumps(index, ensure_ascii=False, indent=2) (logs_dir / "space_logs_index.json").write_text(index_text, encoding="utf-8") written.append({"file": "space_logs_index.json", "source": "asf_log_collector", "available": True, "quality": log_quality, "returncode": 0, "tail": _tail(index_text, 1000)}) event_status = "success" if runtime_entry.get("available") or build_entry.get("available") else "warning" event_message = "Collected Space logs and runtime snapshot" if event_status == "success" else "Space log streams unavailable; runtime snapshot/index written" append_event(events_path, "space_logs", event_status, event_message, {"files": written, "log_quality": log_quality, "index": "logs/space_logs_index.json"}) write_artifact_manifest(run_dir, reason="space_logs_collected") return written def validate_http_health(target_space_id: str, token: str, run_dir: Path, events_path: Path, attempt: int): import requests base_url = space_subdomain_url(target_space_id) url = base_url.rstrip("/") + "/health" headers = {"Authorization": f"Bearer {token}", "Accept": "application/json,text/plain,*/*"} response = requests.get(url, headers=headers, timeout=20) payload = { "status": "success" if response.ok else "failed", "attempt": attempt, "url": url, "status_code": response.status_code, "content_type": response.headers.get("content-type"), "text": response.text[:2000], } if response.ok: try: payload["json"] = response.json() except Exception: pass write_json(run_dir / "tests" / "http_health.json", payload) write_json(run_dir / "tests" / "test_result.json", payload | {"validator": "http_get_health"}) append_event(events_path, "api_validation", "success", "HTTP /health validation passed", {"attempt": attempt, "url": url, "status_code": response.status_code}) return payload | {"validator": "http_get_health"} raise RuntimeError(f"HTTP /health returned {response.status_code}: {response.text[:500]}") 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) 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) candidates = [] for name in ["/health", "/predict", "/greet"] + discovered: if name not in candidates: candidates.append(name) errors = [] for api_name in candidates: try: if api_name == "/greet": result = client.predict("Agentic Space Factory", api_name=api_name) else: result = client.predict(api_name=api_name) payload = {"status": "success", "attempt": attempt, "api_name": api_name, "discovered_api_names": discovered, "result_repr": repr(result)[:2000], "validator": "gradio_client"} write_json(run_dir / "tests" / "test_result.json", payload) append_event(events_path, "api_validation", "success", "Gradio API validation passed", {"attempt": attempt, "api_name": api_name, "discovered_api_names": discovered}) return payload except Exception as exc: errors.append({"api_name": api_name, "error": str(exc)[:1000]}) raise RuntimeError("; ".join(f"{e['api_name']}: {e['error']}" for e in errors[:5]) or "No callable API endpoints found") def latest_collected_build_log(run_dir: Path) -> str: path = run_dir / "logs" / "space_logs_build.txt" return path.read_text(encoding="utf-8", errors="ignore") if path.exists() else "" def runtime_stage_is_build_error(stage: str) -> bool: value = (stage or "").upper() return "BUILD_ERROR" in value or value in {"BUILDING_ERROR", "ERROR", "FAILED", "BUILD_FAILED"} def runtime_stage_is_busy(stage: str) -> bool: value = (stage or "").upper() return any(marker in value for marker in ["BUILDING", "BUILDQUEUED", "BUILD_QUEUED", "STARTING", "PENDING"]) def build_log_has_terminal_error(text: str) -> bool: low = (text or "").lower() terminal_markers = [ "--> error:", "error: resolutionimpossible", "resolutionimpossible", "no matching distribution found", "could not find a version that satisfies the requirement", "did not complete successfully", "process \"/bin/sh -c pip install", "cannot install", "conflicting dependencies", "metadata-generation-failed", "exit code: 1", ] return any(marker in low for marker in terminal_markers) def raise_if_collected_build_error(target_space_id: str, token: str, run_dir: Path, events_path: Path, *, attempt: int, reason: str = "", runtime_payload: dict | None = None): collect_space_logs(target_space_id, token, run_dir, events_path) build_log = latest_collected_build_log(run_dir) stage = str((runtime_payload or {}).get("stage") or "") if runtime_stage_is_build_error(stage) or build_log_has_terminal_error(build_log): issue = extract_pip_dependency_issue(build_log) payload = { "attempt": attempt, "reason": reason, "runtime": runtime_payload or {}, "dependency_issue": issue, "tail": build_log[-4000:], } append_event(events_path, "api_validation", "failed", "Space is in build error; stopping wait and entering recovery", payload) write_json(run_dir / "build_error_observation.json", payload) raise RuntimeError(f"Space build failed according to runtime/logs. {json.dumps(issue, ensure_ascii=False)[:1000]} See build_error_observation.json and logs/space_logs_build.txt") def validate_live_api(api, target_space_id: str, token: str, run_dir: Path, events_path: Path, timeout_s: int = 900): append_event(events_path, "api_validation", "started", "Waiting for live HTTP /health or Gradio API to become available") deadline = time.time() + timeout_s attempt = 0 last_error = None runtime_error_count = 0 while time.time() < deadline: attempt += 1 runtime_payload = write_space_runtime(api, target_space_id, token, run_dir, events_path, attempt) stage = str(runtime_payload.get("stage") or "").upper() # Never keep waiting when the Hub already reports a BUILD_ERROR. This # is the critical observer loop: manual refresh should not be required. if runtime_stage_is_build_error(stage): raise_if_collected_build_error(target_space_id, token, run_dir, events_path, attempt=attempt, reason=f"runtime_stage={stage}", runtime_payload=runtime_payload) # While the Space is building/starting, periodically inspect build logs. # A terminal pip/build error in logs is enough to enter recovery even if # runtime state has not propagated yet. if attempt == 1 or attempt % 2 == 0 or any(marker in stage for marker in ["BUILD", "ERROR", "RUNTIME"]): raise_if_collected_build_error(target_space_id, token, run_dir, events_path, attempt=attempt, reason=f"runtime_stage={stage}", runtime_payload=runtime_payload) if "RUNTIME_ERROR" in stage: runtime_error_count += 1 last_error = f"Space runtime stage is {stage}" if runtime_error_count >= 2: raise RuntimeError(f"Space is in RUNTIME_ERROR. See logs/space_logs_runtime.txt and logs/space_logs_build.txt. Last runtime: {runtime_payload}") try: return validate_http_health(target_space_id, token, run_dir, events_path, attempt) except Exception as exc: last_error = f"HTTP /health failed: {exc}" try: return validate_gradio_api(target_space_id, token, run_dir, events_path, attempt) except Exception as exc: last_error = (last_error or "") + f"; Gradio API failed: {exc}" append_event(events_path, "api_validation", "waiting", "Live health/API not ready yet", {"attempt": attempt, "runtime": runtime_payload, "error": last_error[-1500:] if last_error else None}) time.sleep(30) collect_space_logs(target_space_id, token, run_dir, events_path) raise RuntimeError(f"Live health/API validation did not pass before timeout: {last_error}") def is_auth_or_billing_like_error(error: str | None) -> bool: value = error or "" markers = [ "401", "402", "403", "Invalid username or password", "Unauthorized", "Repository Not Found", "payment", "billing", "quota", "grant", ] return any(marker.lower() in value.lower() for marker in markers) def request_hardware(api, target_space_id: str, hardware: str, token: str, events_path: Path, step: str, retries: int = 2): """Best-effort hardware request after Space creation. V23 tries hardware at create_repo time first. This function remains as a fallback for cases where a Space was created on CPU and the Hub later accepts a hardware switch. Auth/billing/quota errors are not retried. """ if not hardware: return {"phase": "post_create_request", "requested": False, "hardware": hardware, "ok": False, "error": "empty hardware"} last_error = None for attempt in range(1, retries + 1): try: runtime = api.request_space_hardware(repo_id=target_space_id, hardware=hardware, token=token) payload = { "phase": "post_create_request", "requested": True, "hardware": hardware, "ok": True, "attempt": attempt, "runtime_stage": getattr(getattr(runtime, "stage", None), "value", str(getattr(runtime, "stage", None))), "requested_hardware": getattr(runtime, "requested_hardware", None), "hardware_current": getattr(runtime, "hardware", None), } append_event(events_path, step, "success", f"Requested Space hardware {hardware}", payload) return payload except Exception as exc: exc_payload = exception_payload(exc) last_error = str(exc_payload.get("hf_error_message") or exc_payload.get("hf_error_detail") or exc_payload.get("error") or "")[:4000] auth_like = is_auth_or_billing_like_error(last_error) payload = {"phase": "post_create_request", "attempt": attempt, "hardware": hardware, "error": last_error, "manual_action_required": auth_like, **exc_payload} append_event(events_path, step, "failed" if auth_like or attempt == retries else "waiting", f"Could not request Space hardware {hardware}", payload) if auth_like: return {"phase": "post_create_request", "requested": True, "hardware": hardware, "ok": False, "attempts": attempt, "error": last_error, "manual_action_required": True, **exc_payload} if attempt < retries: time.sleep(8 * attempt) return {"phase": "post_create_request", "requested": True, "hardware": hardware, "ok": False, "attempts": retries, "error": last_error, "manual_action_required": False} def normalize_auto_space_hardware(value: str | None, default: str) -> str: candidate = (value or default).strip() return candidate if candidate in AUTO_SPACE_HARDWARE_CHOICES else default def build_hardware_sequence(preferred_hardware: str, fallback_hardware: str, allow_fixed_gpu_fallback: bool, try_zero_gpu_first: bool = True) -> list[str]: sequence = [] preferred_hardware = normalize_auto_space_hardware(preferred_hardware, DEFAULT_PREFERRED_SPACE_HARDWARE) fallback_hardware = normalize_auto_space_hardware(fallback_hardware, DEFAULT_FALLBACK_SPACE_HARDWARE) planned = [] if try_zero_gpu_first: planned.append("zero-a10g") planned.append(preferred_hardware) if allow_fixed_gpu_fallback: planned.append(fallback_hardware) for hw in planned: value = (hw or "").strip() if value and value not in sequence: sequence.append(value) return sequence def is_repo_already_exists_error(error: str) -> bool: text = str(error or "").lower() return ("already" in text and ("created this space repo" in text or "already exists" in text or "repo already" in text or "name already" in text)) def collision_safe_target_space_id(target_space_id: str, attempt: int) -> str: owner, slug = target_space_id.split("/", 1) base = re.sub(r"[^A-Za-z0-9._-]+", "-", slug).strip(".-_") or "space-factory" suffix = f"-r{int(time.time()) % 100000}-{attempt}" max_slug_len = min(95, 96 - len(owner) - 1) return f"{owner}/{base[:max_slug_len - len(suffix)].rstrip('.-_')}{suffix}" def create_space_with_hardware_strategy(api, target_space_id: str, token: str, preferred_hardware: str, fallback_hardware: str, allow_fixed_gpu_fallback: bool, events_path: Path, allow_auto_rename: bool = False, try_zero_gpu_first: bool = True): """Create a private Space and request hardware as early as possible. HF supports `space_hardware` directly on create_repo. This is the cleanest moment to request hardware because the Space does not need a second restart. If OAuth/billing/quota prevents automatic hardware selection, fall back to a normal private CPU Space and mark manual hardware as required. When the target Space name was generated automatically, a stale duplicate run id or a double-submit should not kill the build. In that one case, retry creation with a fresh suffix and return the final target Space id so the rest of the run, report and UI stay coherent. User-provided names still fail loudly if they already exist. """ sequence = build_hardware_sequence(preferred_hardware, fallback_hardware, allow_fixed_gpu_fallback, try_zero_gpu_first) attempts = [] current_target_space_id = target_space_id rename_attempt = 0 def maybe_rename_after_collision(error: str, phase: str, hardware: str | None = None) -> bool: nonlocal current_target_space_id, rename_attempt if not (allow_auto_rename and is_repo_already_exists_error(error) and rename_attempt < 5): return False previous = current_target_space_id rename_attempt += 1 current_target_space_id = collision_safe_target_space_id(previous, rename_attempt) append_event( events_path, "target_space_collision", "warning", "Auto-generated target Space already existed; retrying with a fresh generated name", {"phase": phase, "hardware": hardware, "previous_target_space": previous, "target_space": current_target_space_id}, ) return True def previous_create_attempt_for_current_target() -> bool: return any( str(item.get("target_space") or "") == current_target_space_id and str(item.get("phase") or "").startswith("create_repo_") for item in attempts if isinstance(item, dict) ) def adopt_existing_after_ambiguous_create(error: str, phase: str, hardware: str | None = None): """Keep one Space per run when a previous create may have succeeded. HF Hub calls can occasionally create the repo but still return an error to the client, or a hardware-at-creation ZeroGPU attempt can leave a CPU Space behind. In that situation the next hardware fallback sees "already exists" for the same target. That must not trigger an auto-rename, otherwise one run could create a second Space. """ if not (is_repo_already_exists_error(error) and previous_create_attempt_for_current_target()): return None payload = { "phase": "create_repo_existing_after_ambiguous_create", "hardware": hardware or "cpu-basic", "ok": True, "target_space": current_target_space_id, "source_error": error, "single_space_per_run": True, } append_event( events_path, "create_space", "warning", "Target Space already exists after an earlier create attempt; reusing it instead of generating another Space", payload, ) if hardware: hw_payload = request_hardware(api, current_target_space_id, hardware, token, events_path, "hardware_post_create_reuse") attempts.append({**payload, "post_create_hardware_request": hw_payload}) selected = hardware if hw_payload.get("ok") else "default-cpu-or-existing" manual = bool(hw_payload.get("manual_action_required") or not hw_payload.get("ok")) return {"created": True, "target_space_id": current_target_space_id, "selected_hardware": selected, "requested_sequence": sequence, "attempts": attempts, "manual_action_required": manual} attempts.append(payload) return {"created": True, "target_space_id": current_target_space_id, "selected_hardware": "default-cpu-or-existing", "requested_sequence": sequence, "attempts": attempts, "manual_action_required": True} for hardware in sequence: while True: try: append_event(events_path, "create_space_hardware", "started", f"Creating private Space with requested hardware {hardware}", {"target_space": current_target_space_id, "hardware": hardware}) api.create_repo( repo_id=current_target_space_id, repo_type="space", space_sdk="gradio", private=True, exist_ok=False, space_hardware=hardware, token=token, ) payload = {"phase": "create_repo_space_hardware", "hardware": hardware, "ok": True, "target_space": current_target_space_id} append_event(events_path, "create_space", "success", f"Private target Space created with requested hardware {hardware}", payload) return {"created": True, "target_space_id": current_target_space_id, "selected_hardware": hardware, "requested_sequence": sequence, "attempts": attempts + [payload], "manual_action_required": False} except Exception as exc: exc_payload = exception_payload(exc) error = str(exc_payload.get("hf_error_message") or exc_payload.get("hf_error_detail") or exc_payload.get("error") or "")[:4000] reused = adopt_existing_after_ambiguous_create(error, "create_repo_space_hardware", hardware) if reused: return reused if maybe_rename_after_collision(error, "create_repo_space_hardware", hardware): continue manual = is_auth_or_billing_like_error(error) payload = {"phase": "create_repo_space_hardware", "hardware": hardware, "ok": False, "error": error, "manual_action_required": manual, "target_space": current_target_space_id, **exc_payload} attempts.append(payload) append_event(events_path, "create_space_hardware", "failed", f"Could not create Space with requested hardware {hardware}", payload) break # Continue through the sequence: ZeroGPU quota/auth can fail while a fixed GPU # may still be worth trying. If fixed GPU also fails, we'll create CPU below. append_event(events_path, "create_space", "started", "Creating private target Space on default CPU after hardware-at-creation attempts failed", {"target_space": current_target_space_id}) while True: try: api.create_repo(repo_id=current_target_space_id, repo_type="space", space_sdk="gradio", private=True, exist_ok=False, token=token) cpu_payload = {"phase": "create_repo_default_cpu", "hardware": "cpu-basic", "ok": True, "target_space": current_target_space_id, "manual_action_required": True} append_event(events_path, "create_space", "success", "Private target Space created on default CPU; manual hardware selection may be required", cpu_payload) return {"created": True, "target_space_id": current_target_space_id, "selected_hardware": "default-cpu-or-existing", "requested_sequence": sequence, "attempts": attempts + [cpu_payload], "manual_action_required": True} except Exception as exc: exc_payload = exception_payload(exc) error = str(exc_payload.get("hf_error_message") or exc_payload.get("hf_error_detail") or exc_payload.get("error") or "")[:4000] reused = adopt_existing_after_ambiguous_create(error, "create_repo_default_cpu") if reused: return reused if maybe_rename_after_collision(error, "create_repo_default_cpu"): continue raise def create_initial_workspace(workspace: Path, model_id: str, target_space_id: str, preferred_hardware: str, fallback_hardware: str, allow_fallback: bool, implementation_mode: str, model_analysis: dict | None = None): workspace.mkdir(parents=True, exist_ok=True) model_analysis = model_analysis or {} pipeline_tag = model_analysis.get("pipeline_tag") library_name = model_analysis.get("library_name") tags = model_analysis.get("tags", [])[:40] siblings = model_analysis.get("siblings", [])[:60] app_py = f"""import gradio as gr from huggingface_hub import model_info, list_repo_files MODEL_ID = {model_id!r} TARGET_SPACE_ID = {target_space_id!r} def health(): return {{ "status": "booted", "model_id": MODEL_ID, "target_space_id": TARGET_SPACE_ID, "stage": "initial-scaffold", "note": "Pi should replace this scaffold with a model-specific demo while preserving a cheap health endpoint.", }} def placeholder(*args): return "Initial scaffold. Pi should replace this with a model-specific inference path, or write TECHNICAL_BLOCKERS.json." with gr.Blocks(title="Generated Model Space — Agentic Space Factory") as demo: gr.Markdown("# Generated Model Space — Agentic Space Factory") gr.Markdown(f"Private generated Space for `{{MODEL_ID}}`.") gr.JSON(label="Health", value=health(), every=None) gr.Button("Health check").click(fn=health, inputs=None, outputs=gr.JSON(), api_name="health") gr.Textbox(label="Input", value="Hello from Agentic Space Factory").submit(fn=placeholder, inputs=None, outputs=gr.Textbox(), api_name="predict") gr.Button("Run placeholder").click(fn=placeholder, inputs=None, outputs=gr.Textbox(), api_name="predict") if __name__ == "__main__": demo.launch() """ (workspace / "app.py").write_text(app_py, encoding="utf-8") req = """gradio>=6.0.0 huggingface_hub>=0.34.0,<2.0.0 spaces transformers>=4.45.0,<6.0.0 diffusers accelerate safetensors torch kernels pillow numpy requests """ (workspace / "requirements.txt").write_text(req, encoding="utf-8") readme = f"""--- title: Generated Model Space sdk: gradio app_file: app.py python_version: "3.10" suggested_hardware: {preferred_hardware or fallback_hardware or "cpu-basic"} short_description: "Generated model demo" --- # Generated Model Space — Agentic Space Factory Private generated Space for `{model_id}`. This Space is created by Agentic Space Factory. It should remain private until manually reviewed. """ (workspace / "README.md").write_text(readme, encoding="utf-8") analysis_json = json.dumps({"pipeline_tag": pipeline_tag, "library_name": library_name, "tags": tags, "siblings": siblings}, indent=2, ensure_ascii=False) goal = f"""You are Pi running inside a Hugging Face Job for Agentic Space Factory. Goal: build the best possible private Hugging Face Space demo for an arbitrary model card. MODEL_ID: {model_id} TARGET_SPACE_ID: {target_space_id} IMPLEMENTATION_MODE: {implementation_mode} MODEL_METADATA: ```json {analysis_json} ``` First read and follow the operational rules from this gist: {GIST_URL} {pi_tooling_context_note()} Non-negotiable safety and product constraints: - The target Space must remain private. - Do not delete any user resources. - Do not print secrets or tokens. - Work only inside the current workspace. - The wrapper will create the private Space, request allowed hardware best-effort, upload files, and validate the live app. Do not create/delete repos yourself in this builder worker. - Preserve a cheap health endpoint named `health` with `api_name="health"`. It must not load weights, run GPU work, or download large files. - Do not pin huggingface_hub below 1.0. Use huggingface_hub>=0.34.0,<2.0.0 unless the model card requires a narrower compatible range. If transformers>=5 is used, keep huggingface_hub compatible with it, for example huggingface_hub>=1.5.0,<2.0.0. - README.md frontmatter must remain valid; if it uses short_description, it must be 60 characters or fewer. Implementation contract: - If IMPLEMENTATION_MODE is `full-inference-gated`, you are not allowed to silently replace generation with a placeholder and call it success. - Try to implement the closest real inference path for the model card using evidence from README, model metadata, config files, and repo files. - You may choose an appropriate Gradio UI for the task: text, image, audio, video, multimodal, embeddings, classification, etc. - If the model is standard and feasible, implement a real generate/predict function and expose it as a Gradio endpoint. - If the model requires GPU, add ZeroGPU-compatible `@spaces.GPU(...)` only around the inference function. Do not decorate health. Do not assume dedicated A100/H200 hardware is available automatically; if such hardware is needed, document it as a manual requirement. - If the model requires special dependencies, include them only when needed and document risks. - Investigate compatibility fallbacks before declaring a blocker: PyTorch SDPA, xformers, HF Kernels where relevant, CPU/offload/lazy loading, smaller resolution/steps, safe smoke-test inputs. - If real inference is impossible or unsafe in a Space, write TECHNICAL_BLOCKERS.json with concrete evidence for every blocker. Deliverables: - app.py must boot on Hugging Face Spaces. - app.py must expose health/api_name="health". - If real generation is implemented, generate/predict must attempt a real model call, not only return a textual diagnostic. - If real generation is not implemented, write TECHNICAL_BLOCKERS.json with: full_inference_implemented=false, blockers[], evidence[], minimum_runtime, and suggested_next_step. - Write INFERENCE_CONTRACT.json with: full_inference_implemented, health_endpoint, primary_api_name, expected_output_type, validation_level, requires_gpu, estimated_vram, and blockers_count. - README.md must explain the runtime strategy, task, limitations, and how to test. - Write a concise PI_SUMMARY.md with what you changed and whether full inference is implemented. """ (workspace / "GOAL.md").write_text(goal, encoding="utf-8") return ["app.py", "requirements.txt", "README.md", "GOAL.md"] def sanitize_readme_metadata(workspace: Path, events_path: Path): """Ensure the generated Space README always keeps valid HF metadata. Pi is allowed to rewrite README.md, but Spaces require a valid YAML frontmatter block for sdk/app_file. If Pi drops it, the Space can fail with a Hub configuration error before the app even builds. Treat this as a factory invariant, not a Pi preference. """ readme_path = workspace / "README.md" if not readme_path.exists(): readme_path.write_text("# Generated Model Space\n", encoding="utf-8") text = readme_path.read_text(encoding="utf-8", errors="ignore") body = text metadata = {} changed = False if text.startswith("---"): parts = text.split("---", 2) if len(parts) >= 3: _, frontmatter, body = parts for line in frontmatter.splitlines(): if ":" not in line: continue key, value = line.split(":", 1) metadata[key.strip()] = value.strip() else: body = text changed = True else: changed = True raw_python_version = metadata.get("python_version") or DEFAULT_SPACE_PYTHON_VERSION python_version, python_version_changed, python_version_reason = normalize_space_python_version(raw_python_version) required = { "title": metadata.get("title") or "Generated Model Space", "sdk": "gradio", "app_file": "app.py", "python_version": python_version, } if python_version_changed: changed = True suggested = metadata.get("suggested_hardware") if suggested: required["suggested_hardware"] = suggested short = metadata.get("short_description") or "Generated model demo" if len(short.strip('"\'')) > 60: short = "Generated model demo" required["short_description"] = short ordered_keys = ["title", "sdk", "app_file", "python_version", "suggested_hardware", "short_description"] lines = [] for key in ordered_keys: if key in required and required[key]: value = str(required[key]).strip() if key in {"title", "short_description", "python_version"} and not (value.startswith('"') or value.startswith("'")): value = json.dumps(value, ensure_ascii=False) lines.append(f"{key}: {value}") normalized_body = body.lstrip("\n") or "# Generated Model Space\n" new_text = "---\n" + "\n".join(lines) + "\n---\n\n" + normalized_body if new_text != text: readme_path.write_text(new_text, encoding="utf-8") append_event( events_path, "metadata_sanitize", "success", "Ensured README Space metadata", { "metadata_keys": [k for k in ordered_keys if k in required], "python_version_before": str(raw_python_version).strip().strip("\"'"), "python_version_after": python_version, "python_version_normalized": bool(python_version_changed), "python_version_reason": python_version_reason, }, ) def normalize_requirements_for_modern_hub(workspace: Path, events_path: Path): """Normalize only broad known-dangerous base dependencies before upload. Do not try to solve every dependency conflict here. Pi is responsible for reading concrete build logs and patching requirements when a Space build fails. The Factory only prevents obviously unsafe broad ranges that would make the initial build non-deterministic, then observes the Space status and hands real build errors back to Pi with evidence. """ req_path = workspace / "requirements.txt" if not req_path.exists(): return raw = req_path.read_text(encoding="utf-8", errors="ignore") lines = [line.rstrip() for line in raw.splitlines()] prefix_lines: list[str] = [] package_lines: list[str] = [] for line in lines: stripped = line.strip() if not stripped: continue if stripped.startswith("--") or stripped.startswith("-f "): prefix_lines.append(line) else: package_lines.append(line) # Minimal base pins only. Do not globally pin diffusers here: newer model # cards may legitimately need a recent Diffusers release. If diffusers causes # a pip conflict, Pi must repair from the concrete build log. policy: dict[str, str] = { "huggingface-hub": "huggingface_hub>=0.34.0,<2.0.0", "transformers": "transformers>=4.51.0,<5.0.0", } aliases = { "huggingface_hub": "huggingface-hub", "huggingface-hub": "huggingface-hub", "transformers": "transformers", } seen_policy: set[str] = set() filtered: list[str] = [] changed = False for line in package_lines: stripped = line.strip() if stripped.startswith("#") or "://" in stripped or stripped.startswith((".", "/")): filtered.append(line) continue name = re.split(r"[<>=!~;\[]", stripped, 1)[0].strip().lower().replace("_", "-") canonical = aliases.get(name) if canonical in policy: if stripped != policy[canonical]: changed = True if canonical not in seen_policy: filtered.append(policy[canonical]) seen_policy.add(canonical) else: changed = True continue filtered.append(line) stable_policy_lines: list[str] = [] for canonical in ["huggingface-hub", "transformers"]: if canonical not in seen_policy: stable_policy_lines.append(policy[canonical]) changed = True torch_added = False if workspace_app_imports_torch(workspace) and not requirements_has_package(filtered + stable_policy_lines, "torch"): stable_policy_lines.append("torch>=2.0.0") torch_added = True changed = True new_lines = prefix_lines + stable_policy_lines + filtered new = "\n".join(line for line in new_lines if line.strip()) + "\n" if new != raw: changed = True if changed: req_path.write_text(new, encoding="utf-8") append_event( events_path, "requirements_sanitize", "success", "Normalized broad base dependencies; concrete pip conflicts remain Pi repair work", { "huggingface_hub": policy["huggingface-hub"], "transformers": policy["transformers"], "reason": "Avoid uncontrolled Transformers 5.x while preserving model-specific dependency choices for Pi to repair from build logs; require torch when app.py imports torch.", "torch_added": torch_added, "torch_policy": "torch>=2.0.0", "torch_reason": "app_imports_torch" if torch_added else "not_needed_or_already_present", }, ) def useful_log_signals(text: str) -> list[str]: low = (text or "").lower() signals = [] patterns = { "python_exception": ["traceback", "modulenotfounderror", "importerror", "runtimeerror", "valueerror", "typeerror"], "dependency_resolution": ["resolutionimpossible", "no matching distribution", "could not find a version", "pip subprocess", "metadata-generation-failed"], "gradio_api": ["api_name", "endpoint", "unexpected keyword", "too many arguments", "not enough arguments", "no api found"], "model_loading": ["from_pretrained", "model_index", "safetensors", "does not appear to have", "failed to load", "pipeline"], "hardware_memory": ["cuda out of memory", "outofmemoryerror", "oom", "not enough memory"], "auth_access": ["401", "403", "gated", "unauthorized", "forbidden", "repository not found"], "zerogpu": ["zerogpu", "spaces.gpu", "duration"], } for name, needles in patterns.items(): if any(n in low for n in needles): signals.append(name) return signals def log_quality(build_log: str = "", runtime_log: str = "", failure_reason: str = "") -> str: combined = f"{build_log}\n{runtime_log}\n{failure_reason}".strip() low = combined.lower() if not combined or len(combined) < 80: return "empty" no_reason_markers = [ "no logs", "logs are empty", "no reason", "unknown error", "runtime error", "build error", "error: none", "status=error", ] if any(m in low for m in no_reason_markers) and not useful_log_signals(combined): return "no_reason" if useful_log_signals(combined): return "useful" return "partial" def classify_repair_failure(failure_reason: str, build_log: str = "", runtime_log: str = "") -> dict: """Classify a failed live validation before asking Pi to choose an action. This classifier is deliberately conservative. It gives Pi and the Factory evidence quality and a suggested class, but the diagnosis step must still make a bounded decision before any patch is allowed. """ text = f"{failure_reason}\n{build_log[-6000:]}\n{runtime_log[-6000:]}".lower() quality = log_quality(build_log, runtime_log, failure_reason) phase = "unknown" if "build" in text or "pip" in text or "container" in text: phase = "space_build" if "runtime" in text or "health" in text or "client.predict" in text or "api" in text: phase = "space_runtime" if "generation" in text or "smoke" in text or "expected output" in text: phase = "api_validation" checks = [ ("cuda_oom", ["cuda out of memory", "outofmemoryerror", "oom", "not enough memory"], "Prefer hardware/manual action or memory reduction; do not fake inference."), ("dependency_error", ["resolutionimpossible", "could not find a version", "no matching distribution", "dependency conflict", "pip subprocess", "metadata-generation-failed"], "Fix requirements with compatible, modern pins."), ("import_error", ["modulenotfounderror", "importerror", "cannot import name"], "Fix imports or requirements without hiding the error."), ("gradio_api_mismatch", ["no api found", "api_name", "endpoint", "unexpected keyword", "too many arguments", "not enough arguments"], "Preserve the expected Gradio endpoint and align input/output schema."), ("model_loading_error", ["from_pretrained", "safetensors", "model_index", "pipeline", "failed to load", "does not appear to have"], "Re-read the model card usage and fix the pipeline loading path."), ("hf_auth_error", ["401", "403", "gated", "unauthorized", "forbidden", "repository not found"], "Do not bypass auth; report gated/private access if needed."), ("space_boot_timeout", ["timed out", "timeout", "space did not become ready", "sleeping"], "Inspect logs first; if logs are empty, wait/rebuild same code before patching."), ("wrong_output_type", ["wrong output", "expected output", "invalid output", "not an image", "not text"], "Return the expected output type from real inference."), ("zero_gpu_duration_error", ["duration", "spaces.gpu", "zerogpu", "zero gpu"], "Tune @spaces.GPU(duration=...) from observed smoke-test latency when available."), ] category = "unknown_runtime_error" recommendation = "Use the Space logs, Gradio API schema and smoke error to choose the smallest truthful action." for code, needles, rec in checks: if any(n in text for n in needles): category = code recommendation = rec break if quality in {"empty", "no_reason"} and category == "unknown_runtime_error": category = "hf_runtime_flake" if phase != "space_build" else "hf_build_flake" recommendation = "Logs are not actionable. Prefer wait/inspect or a same-code factory rebuild before any code patch." dependency_issue = extract_pip_dependency_issue(f"{failure_reason}\n{build_log}\n{runtime_log}") if dependency_issue: category = "dependency_error" phase = "space_build" quality = "useful" recommendation = "Patch requirements.txt or dependency pins from the first pip resolver error; do not rebuild same code first." return { "category": category, "failure_phase": phase, "logs_quality": quality, "signals": useful_log_signals(text), "recommendation": recommendation, "dependency_issue": dependency_issue, "gist_alignment": "read first actionable pip error, patch minimally, rebuild, validate live" if dependency_issue else "standard blockage diagnosis", } def workspace_file_inventory(workspace: Path, max_files: int = 80) -> list[str]: files = [] for path in sorted(workspace.rglob("*")): if not path.is_file(): continue if any(part in {".git", ".cache", "node_modules", "__pycache__"} for part in path.parts): continue try: files.append(str(path.relative_to(workspace))) except Exception: files.append(path.name) if len(files) >= max_files: break return files def write_incident_brief(workspace: Path, run_dir: Path, *, target_space_id: str, model_id: str, pi_model: str, failure_reason: str, build_log: str, runtime_log: str, classification: dict, implementation_mode: str, expected_output_type: str, iteration: int, budgets: dict | None = None) -> str: repair_dir = run_dir / "repair" repair_dir.mkdir(parents=True, exist_ok=True) inventory = workspace_file_inventory(workspace) budgets = budgets or {} brief = f"""# Agentic Space Factory incident brief This is a blockage diagnosis pass for an existing generated Hugging Face Space. Pi must use the HF Spaces gist method: read logs first, act once, choose the cheapest useful iteration rung, and verify on the live Space. ## Target - Model ID: `{model_id}` - Target Space: `{target_space_id}` - Pi model: `{pi_model}` - Implementation mode: `{implementation_mode}` - Expected output type: `{expected_output_type}` - Diagnosis iteration: `{iteration}` ## Current classifier hints ```json {json.dumps(classification, indent=2, ensure_ascii=False)} ``` ## Dependency build error policy If `dependency_issue` is non-empty, this is an actionable pip build failure. Prefer `patch_code` with a minimal `requirements.txt` change. Do not choose `factory_rebuild_same_code` until a dependency patch has been attempted. ## HF tooling/token context ```json {json.dumps(load_json_if_exists(run_dir / "token_context.json"), indent=2, ensure_ascii=False)} ``` ## Remaining action budget ```json {json.dumps(budgets, indent=2, ensure_ascii=False)} ``` ## Observed failure ```text {failure_reason[:5000]} ``` ## Build log tail ```text {build_log[-12000:]} ``` ## Runtime log tail ```text {runtime_log[-12000:]} ``` ## Workspace files ```text {chr(10).join(inventory)} ``` ## Allowed decisions You must choose exactly one action: - `wait_for_logs`: logs are empty/late; wait and collect again. - `inspect_more_logs`: enough uncertainty remains; collect logs/API status again without changing code. - `factory_rebuild_same_code`: likely HF build/runtime flake; re-upload the same workspace to force a rebuild, then revalidate. - `patch_code`: logs are actionable and point to a code/dependency/API/model-loading issue. - `request_manual_hardware`: issue is hardware/quota/restricted GPU/memory and should not be patched away. - `declare_technical_blocker`: no safe automated action remains. ## Hard rules - Do not choose `patch_code` when logs are empty or no-reason unless you cite concrete evidence from files/API schema. - Do not fake inference or replace model output with placeholders. - Do not patch hardware/quota/authorization failures as code bugs. - If logs are actionable, identify the first real error, not only the last line. - If using `factory_rebuild_same_code`, do not change files. ## Required output Write `REPAIR_DECISION.json` in the workspace root with this schema: ```json {{ "action": "wait_for_logs|inspect_more_logs|factory_rebuild_same_code|patch_code|request_manual_hardware|declare_technical_blocker", "confidence": "low|medium|high", "reason": "short explanation", "evidence": ["specific evidence"], "patch_allowed": false, "requires_manual_hardware": false }} ``` """ (repair_dir / "INCIDENT_BRIEF.md").write_text(brief, encoding="utf-8") (workspace / "INCIDENT_BRIEF.md").write_text(brief, encoding="utf-8") write_json(repair_dir / "classification.json", classification) return brief def extract_json_object(text: str) -> dict: if not text: return {} candidates = [] fenced = re.findall(r"```(?:json)?\s*(\{.*?\})\s*```", text, flags=re.S) candidates.extend(fenced) first = text.find("{") last = text.rfind("}") if first != -1 and last != -1 and last > first: candidates.append(text[first:last + 1]) for candidate in candidates: try: value = json.loads(candidate) if isinstance(value, dict): return value except Exception: continue return {} def extract_pip_dependency_issue(text: str = "") -> dict: """Extract the first actionable pip dependency resolver/build error. The point is not to solve every dependency automatically. It is to make dependency build failures first-class evidence so Pi gets a precise repair brief and the Factory does not choose wait/rebuild for a deterministic requirements problem. """ raw = text or "" low = raw.lower() issue: dict = {} patterns = [ ("no_matching_distribution", r"Could not find a version that satisfies the requirement\s+([^\s]+)(?:\s+\(from[^\n]*\))?"), ("no_matching_distribution", r"No matching distribution found for\s+([^\s]+)"), ("resolution_impossible", r"ResolutionImpossible[^\n]*"), ("metadata_generation_failed", r"metadata-generation-failed[^\n]*"), ("subprocess_exited", r"pip subprocess.*?did not run successfully[^\n]*"), ] for kind, pattern in patterns: match = re.search(pattern, raw, flags=re.I | re.S) if match: requirement = match.group(1).strip(" ,.;:") if match.groups() else "" package = re.split(r"[<>=!~;\[]", requirement, 1)[0].strip().lower().replace("_", "-") issue = { "kind": kind, "requirement": requirement, "package": package, "line": match.group(0)[:1000], } break if not issue and any(marker in low for marker in ["could not find a version", "no matching distribution", "resolutionimpossible", "metadata-generation-failed"]): issue = {"kind": "dependency_resolution", "requirement": "", "package": "", "line": raw[-2000:]} if issue: issue["category"] = "dependency_error" issue["actionable"] = True issue["recommended_action"] = "patch_code" issue["recommended_patch_scope"] = "requirements.txt or dependency pins only, unless the first error proves otherwise" return issue def write_dependency_error_brief(workspace: Path, run_dir: Path, events_path: Path, issue: dict, build_log: str = "", runtime_log: str = "", failure_reason: str = "") -> str: repair_dir = run_dir / "repair" repair_dir.mkdir(parents=True, exist_ok=True) brief = f"""# Dependency build error brief The Space build failed during dependency installation. This brief does **not** replace the HF Spaces gist. It is a focused evidence appendix for applying the gist method: read logs first, find the first actionable build error, make one minimal requirements patch, rebuild, then validate live. ## First dependency error ```json {json.dumps(issue, indent=2, ensure_ascii=False)} ``` ## Gist-aligned repair guidance - Read the first real pip resolver error, not only the last log line. - Use the cheapest useful iteration rung from the gist. - For actionable pip errors, the cheapest useful action is normally a minimal `requirements.txt` / dependency pin patch, followed by rebuild and live validation. - Do not use `factory_rebuild_same_code` before a dependency patch when the build log already contains a concrete pip resolver error. - Do not change inference behavior to work around dependency resolution. - Keep the patch minimal and explain why the chosen versions are compatible. ## Build log tail ```text {build_log[-12000:]} ``` ## Runtime log tail ```text {runtime_log[-4000:]} ``` ## Failure reason ```text {failure_reason[:4000]} ``` """ (repair_dir / "DEPENDENCY_ERROR_BRIEF.md").write_text(brief, encoding="utf-8") (workspace / "DEPENDENCY_ERROR_BRIEF.md").write_text(brief, encoding="utf-8") append_event(events_path, "failure_diagnosis", "warning", "Detected pip dependency build error; applying gist log-first method", {"dependency_issue": issue, "artifact": "repair/DEPENDENCY_ERROR_BRIEF.md"}) write_agent_trace_record(run_dir, phase="diagnosis", event="dependency_error_detected", status="warning", message="Detected pip dependency build error; applying gist log-first method", data={"dependency_issue": issue}, artifacts=["repair/DEPENDENCY_ERROR_BRIEF.md"]) return brief def known_dependency_guardrail_needed(workspace: Path, failure_reason: str = "", build_log: str = "", runtime_log: str = "") -> bool: text = f"{failure_reason}\n{build_log}\n{runtime_log}" low = text.lower() issue = extract_pip_dependency_issue(text) # Keep automatic deterministic fixes extremely narrow. General pip conflicts # must be handed to Pi with DEPENDENCY_ERROR_BRIEF.md. if issue and issue.get("package") == "safetensors" and "transformers-5" in low and "safetensors>=0.8.0" in low: return True return False def apply_dependency_guardrail_repair(workspace: Path, run_dir: Path, events_path: Path, failure_reason: str = "", build_log: str = "", runtime_log: str = "") -> bool: """Apply deterministic requirements repair for known pip resolver failures. This handles build failures before Pi can produce a useful runtime patch. """ if not known_dependency_guardrail_needed(workspace, failure_reason, build_log, runtime_log): return False before = (workspace / "requirements.txt").read_text(encoding="utf-8", errors="ignore") if (workspace / "requirements.txt").exists() else "" normalize_requirements_for_modern_hub(workspace, events_path) after = (workspace / "requirements.txt").read_text(encoding="utf-8", errors="ignore") if (workspace / "requirements.txt").exists() else "" repair_dir = run_dir / "repair" repair_dir.mkdir(parents=True, exist_ok=True) summary = """# Deterministic dependency repair The Space build failed during `pip install` because the resolver selected a known Transformers 5.x path requiring unavailable `safetensors>=0.8.0`. The Factory applied only the narrow known trap guardrail: - `transformers>=4.51.0,<5.0.0` - `huggingface_hub>=0.34.0,<2.0.0` Other dependency conflicts are passed to Pi through `DEPENDENCY_ERROR_BRIEF.md`. No inference code was changed. """ (repair_dir / "REPAIR_SUMMARY.md").write_text(summary, encoding="utf-8") (repair_dir / "REPAIR_PLAN.md").write_text("# Deterministic dependency repair plan\n\nNormalize requirements and re-upload the same workspace for a clean Space rebuild.\n", encoding="utf-8") write_agent_trace_record( run_dir, phase="repair_patch", event="dependency_guardrail_repair", status="success", message="Factory applied deterministic requirements repair for safetensors>=0.8.0 resolver failure", data={"before": before[-2000:], "after": after[-2000:]}, artifacts=["repair/REPAIR_SUMMARY.md", "repair/REPAIR_PLAN.md"], ) append_event( events_path, "repair_patch", "success", "Applied deterministic dependency guardrail for pip resolver build error", { "category": "dependency_error", "transformers": "transformers>=4.51.0,<5.0.0", "note": "General diffusers/safetensors conflicts are Pi repair work from DEPENDENCY_ERROR_BRIEF.md.", }, ) return before != after def fallback_decision_from_classifier(classification: dict, budgets: dict | None = None) -> dict: budgets = budgets or {} category = classification.get("category") or "unknown_runtime_error" quality = classification.get("logs_quality") or "partial" if category in {"hf_auth_error"}: action = "declare_technical_blocker" elif category == "cuda_oom": action = "request_manual_hardware" elif quality in {"empty", "no_reason"}: if budgets.get("wait_for_logs", 0) > 0: action = "wait_for_logs" elif budgets.get("factory_rebuild_same_code", 0) > 0: action = "factory_rebuild_same_code" else: action = "declare_technical_blocker" elif category in {"dependency_error", "import_error", "gradio_api_mismatch", "model_loading_error", "wrong_output_type", "zero_gpu_duration_error", "space_boot_timeout"}: action = "patch_code" if budgets.get("patch_code", 0) > 0 else "declare_technical_blocker" elif budgets.get("factory_rebuild_same_code", 0) > 0: action = "factory_rebuild_same_code" elif budgets.get("patch_code", 0) > 0 and quality == "useful": action = "patch_code" else: action = "declare_technical_blocker" return { "action": action, "confidence": "low", "reason": f"Factory fallback decision from classifier category={category}, logs_quality={quality}.", "evidence": classification.get("signals") or [], "patch_allowed": action == "patch_code", "requires_manual_hardware": action == "request_manual_hardware", "source": "factory_fallback", } def normalize_repair_decision(decision: dict, classification: dict, budgets: dict | None = None) -> dict: budgets = budgets or {} allowed = {"wait_for_logs", "inspect_more_logs", "factory_rebuild_same_code", "patch_code", "request_manual_hardware", "declare_technical_blocker"} action = str(decision.get("action") or "").strip().lower().replace("-", "_") if action not in allowed: decision = fallback_decision_from_classifier(classification, budgets) action = decision["action"] quality = classification.get("logs_quality") or "partial" category = classification.get("category") or "unknown_runtime_error" overrides = [] if action == "patch_code" and quality in {"empty", "no_reason"} and category in {"hf_runtime_flake", "hf_build_flake", "unknown_runtime_error"}: overrides.append("patch_code_blocked_without_actionable_logs") if budgets.get("wait_for_logs", 0) > 0: action = "wait_for_logs" elif budgets.get("factory_rebuild_same_code", 0) > 0: action = "factory_rebuild_same_code" else: action = "declare_technical_blocker" if category == "dependency_error" and quality == "useful" and action in {"wait_for_logs", "inspect_more_logs", "factory_rebuild_same_code"}: overrides.append("gist_log_first_dependency_error_needs_minimal_requirements_patch") # The gist says read the first actionable error and act once surgically. # For a concrete pip resolver error, a same-code rebuild/wait is not the # cheapest useful rung. Ask Pi to patch requirements minimally instead. action = "patch_code" if budgets.get("patch_code", 0) > 0 else "declare_technical_blocker" if action in budgets and budgets.get(action, 0) <= 0: overrides.append(f"budget_exhausted:{action}") action = "declare_technical_blocker" if action == "patch_code" and category in {"hf_auth_error"}: overrides.append("auth_failure_not_patchable") action = "declare_technical_blocker" if action == "patch_code" and category == "cuda_oom": # The agent may sometimes reduce memory pressure, but default to manual # hardware unless the logs are useful and it explicitly gave evidence. evidence = " ".join(str(x).lower() for x in decision.get("evidence") or []) if "dtype" not in evidence and "offload" not in evidence and "steps" not in evidence: overrides.append("oom_needs_hardware_or_explicit_memory_plan") action = "request_manual_hardware" normalized = { "action": action, "confidence": str(decision.get("confidence") or "low"), "reason": str(decision.get("reason") or "No reason provided.")[:2000], "evidence": list(decision.get("evidence") or [])[:20], "patch_allowed": action == "patch_code", "requires_manual_hardware": action == "request_manual_hardware", "factory_overrides": overrides, "classification": classification, } return normalized def diagnose_failure_with_pi(workspace: Path, run_dir: Path, events_path: Path, pi_model: str, target_space_id: str, model_id: str, failure_reason: str, implementation_mode: str, expected_output_type: str, iteration: int, budgets: dict) -> dict: logs_dir = run_dir / "logs" build_log = (logs_dir / "space_logs_build.txt").read_text(encoding="utf-8", errors="ignore") if (logs_dir / "space_logs_build.txt").exists() else "" runtime_log = (logs_dir / "space_logs_runtime.txt").read_text(encoding="utf-8", errors="ignore") if (logs_dir / "space_logs_runtime.txt").exists() else "" repair_dir = run_dir / "repair" repair_dir.mkdir(parents=True, exist_ok=True) classification = classify_repair_failure(failure_reason, build_log, runtime_log) dependency_issue = classification.get("dependency_issue") or {} if dependency_issue: write_dependency_error_brief(workspace, run_dir, events_path, dependency_issue, build_log, runtime_log, failure_reason) append_event(events_path, "failure_diagnosis", "started", "Preparing Pi blockage diagnosis from Space status and logs", {"iteration": iteration, **classification}) brief = write_incident_brief( workspace, run_dir, target_space_id=target_space_id, model_id=model_id, pi_model=pi_model, failure_reason=failure_reason, build_log=build_log, runtime_log=runtime_log, classification=classification, implementation_mode=implementation_mode, expected_output_type=expected_output_type, iteration=iteration, budgets=budgets, ) goal = f"""You are Pi in BLOCKAGE DIAGNOSIS MODE for Agentic Space Factory. First read `INCIDENT_BRIEF.md` and the HF Spaces gist operational rules: {GIST_URL} {pi_tooling_context_note()} You are not allowed to edit code in this diagnosis step. Your task is to decide the next action for the Factory. Use the gist method: read logs first, identify the first actionable error, use the cheapest useful iteration rung, and require a live Gradio/API validation before success. Write `REPAIR_DECISION.json` exactly as requested in INCIDENT_BRIEF.md. Do not patch files during this step. """ (workspace / "PI_DIAGNOSIS_GOAL.md").write_text(goal, encoding="utf-8") (repair_dir / "PI_DIAGNOSIS_GOAL.md").write_text(goal, encoding="utf-8") append_event(events_path, "pi_diagnosis", "started", "Running Pi diagnosis decision before any repair action", {"model": pi_model, "iteration": iteration}) write_agent_trace_record(run_dir, phase="diagnosis", event="command_started", status="started", message="Pi blockage diagnosis started", data={"model": pi_model, "iteration": iteration}, artifacts=["repair/INCIDENT_BRIEF.md", "repair/PI_DIAGNOSIS_GOAL.md"]) code, out = run_cmd(["pi", "-p", goal], cwd=workspace, timeout=900) logs_dir.mkdir(parents=True, exist_ok=True) (logs_dir / "pi_diagnosis_output.txt").write_text(out, encoding="utf-8") append_agent_trace_artifact(run_dir, phase="diagnosis", event="diagnosis_output", artifact="logs/pi_diagnosis_output.txt", text=out, status="success" if code == 0 else "failed", data={"returncode": code}) raw_decision = {} decision_path = workspace / "REPAIR_DECISION.json" if decision_path.exists(): raw_decision = load_json_if_exists(decision_path) if not raw_decision: raw_decision = extract_json_object(out) if code != 0: append_event(events_path, "pi_diagnosis", "warning", "Pi diagnosis returned non-zero; using conservative factory fallback", {"returncode": code, "output_tail": out[-3000:]}) if not raw_decision: raw_decision = fallback_decision_from_classifier(classification, budgets) decision = normalize_repair_decision(raw_decision, classification, budgets) write_json(repair_dir / "REPAIR_DECISION.json", decision) (workspace / "REPAIR_DECISION.json").write_text(json.dumps(decision, indent=2, ensure_ascii=False) + "\n", encoding="utf-8") write_agent_trace_record(run_dir, phase="diagnosis", event="repair_decision", status="success", message=f"Pi diagnosis selected action: {decision.get('action')}", data={"decision": decision}, artifacts=["repair/REPAIR_DECISION.json"]) append_event(events_path, "repair_decision", "success", f"Pi diagnosis selected action: {decision.get('action')}", {"decision": decision}) return decision def write_blockage_artifact(workspace: Path, run_dir: Path, events_path: Path, decision: dict, failure_reason: str, *, status: str = "technical_blocker"): blocker = { "full_inference_implemented": False, "source": "agentic_space_factory_blockage_protocol", "status": status, "decision": decision, "failure_reason": failure_reason[:4000], "blockers": [ { "type": decision.get("action") or status, "claim": decision.get("reason") or "Automated recovery could not safely continue.", "evidence": decision.get("evidence") or [], "severity": "blocking", } ], "suggested_next_step": "Inspect repair/INCIDENT_BRIEF.md and repair/REPAIR_DECISION.json, then retry manually or adjust hardware/settings if appropriate.", } (workspace / "TECHNICAL_BLOCKERS.json").write_text(json.dumps(blocker, indent=2, ensure_ascii=False) + "\n", encoding="utf-8") gen_dir = run_dir / "generated" gen_dir.mkdir(parents=True, exist_ok=True) (gen_dir / "TECHNICAL_BLOCKERS.json").write_text(json.dumps(blocker, indent=2, ensure_ascii=False) + "\n", encoding="utf-8") write_json(run_dir / "repair" / "BLOCKAGE.json", blocker) append_event(events_path, "technical_blocker", "failed", "Automated recovery stopped with an auditable blocker", {"decision": decision}) def write_repair_brief(workspace: Path, run_dir: Path, *, target_space_id: str, model_id: str, pi_model: str, failure_reason: str, build_log: str, runtime_log: str, classification: dict, implementation_mode: str, expected_output_type: str, decision: dict | None = None) -> str: repair_dir = run_dir / "repair" repair_dir.mkdir(parents=True, exist_ok=True) inventory = workspace_file_inventory(workspace) decision = decision or {} brief = f"""# Agentic Space Factory repair brief ## Scope This is a structured patch pass for an existing generated Hugging Face Space. A separate Pi diagnosis already decided that `patch_code` is justified. Do not rebuild from scratch unless the current architecture is impossible to boot. ## Diagnosis decision ```json {json.dumps(decision, indent=2, ensure_ascii=False)} ``` ## Original target - Model ID: `{model_id}` - Target Space: `{target_space_id}` - Pi model: `{pi_model}` - Implementation mode: `{implementation_mode}` - Expected output type: `{expected_output_type}` ## Failure classification - Category: `{classification.get('category', 'unknown_runtime_error')}` - Logs quality: `{classification.get('logs_quality', 'partial')}` - Failure phase: `{classification.get('failure_phase', 'unknown')}` - Recommended strategy: {classification.get('recommendation', '')} ## HF tooling/token context ```json {json.dumps(load_json_if_exists(run_dir / "token_context.json"), indent=2, ensure_ascii=False)} ``` ## Observed failure ```text {failure_reason[:5000]} ``` ## Build log tail ```text {build_log[-12000:]} ``` ## Runtime log tail ```text {runtime_log[-12000:]} ``` ## Workspace files ```text {chr(10).join(inventory)} ``` ## Repair constraints - Patch the current workspace; do not change the product goal. - Make the smallest patch that fixes the classified failure. - Preserve real inference for Strict inference mode. - Do not replace inference with static placeholders, fake images, canned text, or swallowed exceptions. - Do not change the model ID unless you write a technical blocker explaining why. - Preserve or restore a cheap `health` endpoint. - Preserve the expected Gradio API endpoint when possible. - Keep README metadata valid and short_description <= 60 chars. - Keep Hugging Face Hub requirements modern; do not pin below 1.0 unless unavoidable and explained. ## Required repair artifacts Before modifying files, write `REPAIR_PLAN.md` with: - root cause - strategy - files to change - risk level - whether hardware/manual action is required After modifying files, write `REPAIR_SUMMARY.md` with: - files changed - why the patch is minimal - how it preserves real inference - how to validate it """ (repair_dir / "REPAIR_BRIEF.md").write_text(brief, encoding="utf-8") (workspace / "REPAIR_BRIEF.md").write_text(brief, encoding="utf-8") write_json(repair_dir / "classification.json", classification) return brief def sanity_check_repair_workspace(workspace: Path, implementation_mode: str) -> tuple[bool, str]: app_path = workspace / "app.py" req_path = workspace / "requirements.txt" readme_path = workspace / "README.md" for required in [app_path, req_path, readme_path]: if not required.exists() or not required.read_text(encoding="utf-8", errors="ignore").strip(): return False, f"Repair removed or emptied required file: {required.name}" if implementation_mode == "full-inference-gated": app_text = app_path.read_text(encoding="utf-8", errors="ignore").lower() # Only block obvious fake-return implementations. Do not fail a valid # targeted runtime patch just because a comment/README still contains # words like "placeholder". The previous broad string scan blocked the # exact kind of repair Pi should be allowed to do. hard_fake_patterns = [ r"return\s+[\"'](?:placeholder|dummy image|fake inference|not implemented|canned response)", r"def\s+placeholder\s*\(", r"gr\.button\([^\n]*placeholder", ] if any(re.search(pattern, app_text) for pattern in hard_fake_patterns): return False, "Strict inference repair appears to introduce executable placeholder/fake inference markers." real_inference_markers = ["from_pretrained", "diffusionpipeline", "zimagepipeline", "pipe(", ".to(\"cuda\")", "@spaces.gpu"] if not any(marker in app_text for marker in real_inference_markers): return False, "Strict inference repair no longer shows a real model inference path." return True, "Repair workspace sanity checks passed." def repair_workspace_with_pi(workspace: Path, run_dir: Path, events_path: Path, pi_model: str, target_space_id: str, model_id: str, failure_reason: str, implementation_mode: str = "full-inference-gated", expected_output_type: str = "any", decision: dict | None = None): """Structured one-shot patch pass, only after Pi diagnosis chooses patch_code.""" logs_dir = run_dir / "logs" build_log = (logs_dir / "space_logs_build.txt").read_text(encoding="utf-8", errors="ignore") if (logs_dir / "space_logs_build.txt").exists() else "" runtime_log = (logs_dir / "space_logs_runtime.txt").read_text(encoding="utf-8", errors="ignore") if (logs_dir / "space_logs_runtime.txt").exists() else "" repair_dir = run_dir / "repair" before_dir = repair_dir / "before" after_dir = repair_dir / "after" if before_dir.exists(): shutil.rmtree(before_dir) repair_dir.mkdir(parents=True, exist_ok=True) shutil.copytree(workspace, before_dir, ignore=shutil.ignore_patterns(".git", "node_modules", "__pycache__", "*.pyc")) classification = (decision or {}).get("classification") or classify_repair_failure(failure_reason, build_log, runtime_log) append_event(events_path, "repair_diagnosis", "success", "Patch repair is allowed by Pi diagnosis", {"category": classification.get("category"), "decision": decision or {}}) brief = write_repair_brief( workspace, run_dir, target_space_id=target_space_id, model_id=model_id, pi_model=pi_model, failure_reason=failure_reason, build_log=build_log, runtime_log=runtime_log, classification=classification, implementation_mode=implementation_mode, expected_output_type=expected_output_type, decision=decision, ) append_event(events_path, "repair_brief", "success", "Repair brief generated for minimal patch", {"category": classification.get("category"), "brief": "repair/REPAIR_BRIEF.md"}) goal = f"""You are Pi in STRUCTURED REPAIR MODE / STRUCTURED PATCH REPAIR MODE for Agentic Space Factory. First read `REPAIR_BRIEF.md`, `INCIDENT_BRIEF.md`, `DEPENDENCY_ERROR_BRIEF.md` if present, and the HF Spaces gist operational rules: {GIST_URL} {pi_tooling_context_note()} You are continuing the same build run, not starting a separate project. This patch is allowed only because the diagnosis decision selected `patch_code`. If `DEPENDENCY_ERROR_BRIEF.md` exists, treat it as evidence for the gist method: identify the first pip error, patch dependency pins minimally, and do not modify inference code unless the dependency fix alone cannot address that first error. Use the available HF token context to inspect private Hub resources when needed, but never print or persist token values. Critical method: 1. Diagnose the root cause from the brief and current files. 2. Write `REPAIR_PLAN.md` before editing files. 3. Apply the smallest patch possible. 4. Write `REPAIR_SUMMARY.md` after editing. 5. Preserve real inference. Never fake outputs just to satisfy validation. Failure category: {classification.get('category')} Recommended strategy: {classification.get('recommendation')} Hard constraints: - Do not rebuild from scratch unless you clearly justify it in REPAIR_PLAN.md. - Do not change MODEL_ID `{model_id}` unless you write a technical blocker. - Do not remove the expected Gradio/API contract. - For Strict inference, do not replace inference with placeholders, static sample files, canned text, or broad try/except blocks that hide failures. - Keep or restore a cheap health endpoint with api_name="health". - Keep README metadata valid. - Do not publish anything. Work only in the current workspace. Required deliverables: REPAIR_PLAN.md, patched files, REPAIR_SUMMARY.md. """ (workspace / "REPAIR_GOAL.md").write_text(goal, encoding="utf-8") (repair_dir / "REPAIR_GOAL.md").write_text(goal, encoding="utf-8") append_event(events_path, "repair_plan", "started", "Running Pi minimal patch repair", {"model": pi_model, "category": classification.get("category")}) write_agent_trace_record(run_dir, phase="repair_patch", event="command_started", status="started", message="Pi structured patch repair started", data={"model": pi_model, "category": classification.get("category"), "decision": decision or {}}, artifacts=["repair/REPAIR_BRIEF.md", "repair/REPAIR_GOAL.md", "repair/REPAIR_DECISION.json"]) code, out = run_cmd(["pi", "-p", goal], cwd=workspace, timeout=1500) logs_dir.mkdir(parents=True, exist_ok=True) (logs_dir / "pi_repair_output.txt").write_text(out, encoding="utf-8") append_agent_trace_artifact(run_dir, phase="repair_patch", event="repair_output", artifact="logs/pi_repair_output.txt", text=out, status="success" if code == 0 else "failed", data={"returncode": code}) if code != 0: append_event(events_path, "repair_patch", "failed", "Pi patch repair returned a non-zero exit code", {"returncode": code, "output_tail": out[-3000:]}) return False plan_path = workspace / "REPAIR_PLAN.md" summary_path = workspace / "REPAIR_SUMMARY.md" if not plan_path.exists(): plan_path.write_text(f"# Repair plan\n\nPi did not create a separate plan file. Classified category: `{classification.get('category')}`.\n\nStrategy: {classification.get('recommendation')}\n", encoding="utf-8") if not summary_path.exists(): summary_path.write_text("# Repair summary\n\nPi completed a structured repair pass but did not write a separate summary file. See logs/pi_repair_output.txt.\n", encoding="utf-8") shutil.copy2(plan_path, repair_dir / "REPAIR_PLAN.md") shutil.copy2(summary_path, repair_dir / "REPAIR_SUMMARY.md") write_agent_trace_record(run_dir, phase="repair_patch", event="repair_plan_summary", status="success", message="Repair plan and summary artifacts are available", data={}, artifacts=["repair/REPAIR_PLAN.md", "repair/REPAIR_SUMMARY.md"]) append_event(events_path, "repair_plan", "success", "Repair plan and summary artifacts are available", {"plan": "repair/REPAIR_PLAN.md", "summary": "repair/REPAIR_SUMMARY.md"}) normalize_requirements_for_modern_hub(workspace, events_path) ok, sanity_message = sanity_check_repair_workspace(workspace, implementation_mode) if not ok: append_event(events_path, "repair_patch", "failed", sanity_message, {"category": classification.get("category")}) return False append_event(events_path, "repair_patch", "success", sanity_message, {"category": classification.get("category")}) if after_dir.exists(): shutil.rmtree(after_dir) shutil.copytree(workspace, after_dir, ignore=shutil.ignore_patterns(".git", "node_modules", "__pycache__", "*.pyc")) append_event(events_path, "repair", "success", "Structured repair patch completed; ready to re-upload and revalidate", {"output_tail": out[-3000:], "category": classification.get("category")}) return True def recover_after_live_validation_failure(api, workspace: Path, run_dir: Path, events_path: Path, *, pi_model: str, target_space_id: str, model_id: str, token: str, failure_reason: str, implementation_mode: str, expected_output_type: str): """Let Pi diagnose the blockage, then execute one bounded Factory action at a time. This is the core blockage protocol. Pi decides among allowed actions, but the Factory validates budgets and refuses unsafe patches without actionable logs. """ budgets = {"wait_for_logs": 1, "inspect_more_logs": 1, "factory_rebuild_same_code": 1, "patch_code": 1} current_error = failure_reason collect_space_logs(target_space_id, token, run_dir, events_path) logs_dir = run_dir / "logs" build_log = (logs_dir / "space_logs_build.txt").read_text(encoding="utf-8", errors="ignore") if (logs_dir / "space_logs_build.txt").exists() else "" runtime_log = (logs_dir / "space_logs_runtime.txt").read_text(encoding="utf-8", errors="ignore") if (logs_dir / "space_logs_runtime.txt").exists() else "" dependency_issue = extract_pip_dependency_issue(f"{current_error}\n{build_log}\n{runtime_log}") if dependency_issue: write_dependency_error_brief(workspace, run_dir, events_path, dependency_issue, build_log, runtime_log, current_error) if apply_dependency_guardrail_repair(workspace, run_dir, events_path, current_error, build_log, runtime_log): append_event(events_path, "factory_rebuild", "started", "Re-uploading dependency-guardrailed workspace after pip resolver build error") if not safe_same_code_reupload(api, workspace, target_space_id, token, run_dir, events_path, reason="dependency_guardrail_repair"): raise RuntimeError("Dependency guardrail rebuild skipped by restart guardrails") append_event(events_path, "factory_rebuild", "success", "Dependency-guardrailed workspace uploaded; revalidating live Space") append_event(events_path, "repair_validation", "started", "Revalidating after deterministic dependency repair") try: validation = validate_live_api(api, target_space_id, token, run_dir, events_path, timeout_s=1200) append_event(events_path, "repair_validation", "success", "Deterministic dependency repair resolved the build blockage") return validation except Exception as exc: current_error = f"{current_error}\n\nDependency guardrail rebuild did not resolve validation: {str(exc)[:4000]}" collect_space_logs(target_space_id, token, run_dir, events_path) append_event(events_path, "repair_validation", "failed", "Dependency guardrail rebuild did not resolve validation; falling back to Pi diagnosis", {"error": str(exc)[:4000]}) for iteration in range(1, 5): decision = diagnose_failure_with_pi( workspace, run_dir, events_path, pi_model, target_space_id, model_id, current_error, implementation_mode, expected_output_type, iteration, budgets, ) action = decision.get("action") if action in {"wait_for_logs", "inspect_more_logs"}: if budgets.get(action, 0) <= 0: decision["factory_overrides"] = list(decision.get("factory_overrides") or []) + [f"budget_exhausted:{action}"] write_blockage_artifact(workspace, run_dir, events_path, decision, current_error) raise RuntimeError(f"Automated recovery stopped after {action} budget was exhausted") budgets[action] -= 1 wait_seconds = int(os.environ.get("SPACE_FACTORY_FAILURE_LOG_WAIT_SECONDS", "45")) append_event(events_path, "wait_for_logs", "started", "Waiting for delayed HF Space logs before changing code", {"seconds": wait_seconds, "decision": decision}) time.sleep(max(1, wait_seconds)) collect_space_logs(target_space_id, token, run_dir, events_path) append_event(events_path, "wait_for_logs", "success", "Collected Space logs after wait", {"remaining_budget": budgets}) current_error = f"{current_error}\n\nAfter wait_for_logs/inspect_more_logs, validation is still considered failed; re-diagnose with refreshed logs." continue if action == "factory_rebuild_same_code": if budgets.get(action, 0) <= 0: decision["factory_overrides"] = list(decision.get("factory_overrides") or []) + ["budget_exhausted:factory_rebuild_same_code"] write_blockage_artifact(workspace, run_dir, events_path, decision, current_error) raise RuntimeError("Automated recovery stopped after factory rebuild budget was exhausted") budgets[action] -= 1 append_event(events_path, "factory_rebuild", "started", "Re-uploading the same workspace to force a same-code Space rebuild", {"decision": decision}) if not safe_same_code_reupload(api, workspace, target_space_id, token, run_dir, events_path, reason="pi_decision_factory_rebuild_same_code"): decision["factory_overrides"] = list(decision.get("factory_overrides") or []) + ["same_code_upload_blocked_by_restart_guardrail"] write_blockage_artifact(workspace, run_dir, events_path, decision, current_error) raise RuntimeError("Same-code rebuild skipped by restart guardrails") append_event(events_path, "factory_rebuild", "success", "Same-code workspace re-uploaded; revalidating live Space") append_event(events_path, "repair_validation", "started", "Revalidating after same-code factory rebuild") try: validation = validate_live_api(api, target_space_id, token, run_dir, events_path, timeout_s=1200) append_event(events_path, "repair_validation", "success", "Same-code factory rebuild resolved the blockage") return validation except Exception as exc: current_error = f"{current_error}\n\nSame-code factory rebuild did not resolve validation: {str(exc)[:4000]}" collect_space_logs(target_space_id, token, run_dir, events_path) append_event(events_path, "repair_validation", "failed", "Same-code factory rebuild did not resolve validation; re-diagnosing", {"error": str(exc)[:4000]}) continue if action == "patch_code": if budgets.get(action, 0) <= 0: decision["factory_overrides"] = list(decision.get("factory_overrides") or []) + ["budget_exhausted:patch_code"] write_blockage_artifact(workspace, run_dir, events_path, decision, current_error) raise RuntimeError("Automated recovery stopped after patch budget was exhausted") budgets[action] -= 1 append_event(events_path, "repair", "started", "Pi diagnosis allows a minimal code patch", {"decision": decision}) repaired = repair_workspace_with_pi(workspace, run_dir, events_path, pi_model, target_space_id, model_id, current_error, implementation_mode, expected_output_type, decision=decision) if not repaired: write_blockage_artifact(workspace, run_dir, events_path, decision, current_error, status="failed_after_repair") append_event(events_path, "failure", "failed", "Structured patch repair failed before redeploy", {"decision": decision}) raise RuntimeError("Structured patch repair failed before redeploy") append_event(events_path, "repair_upload", "started", "Uploading repaired workspace") upload_workspace(api, workspace, target_space_id, token, run_dir, events_path) append_event(events_path, "repair_upload", "success", "Repaired workspace uploaded") append_event(events_path, "repair_validation", "started", "Revalidating repaired Space") try: validation = validate_live_api(api, target_space_id, token, run_dir, events_path, timeout_s=1200) append_event(events_path, "repair_validation", "success", "Repaired Space passed live API validation") return validation except Exception as exc: current_error = f"{current_error}\n\nPatch repair did not resolve validation: {str(exc)[:4000]}" collect_space_logs(target_space_id, token, run_dir, events_path) append_event(events_path, "repair_validation", "failed", "Repair attempted, but validation still failed", {"error": str(exc)[:4000]}) write_blockage_artifact(workspace, run_dir, events_path, decision, current_error, status="failed_after_repair") append_event(events_path, "failure", "failed", "Run failed after structured repair attempt", {"repair_error": str(exc)[:4000]}) raise if action == "request_manual_hardware": write_blockage_artifact(workspace, run_dir, events_path, decision, current_error, status="manual_hardware_required") append_event(events_path, "manual_hardware_required", "failed", "Pi diagnosis requested manual hardware or quota action", {"decision": decision}) raise RuntimeError("Pi diagnosis requested manual hardware or quota action") write_blockage_artifact(workspace, run_dir, events_path, decision, current_error) append_event(events_path, "failure", "failed", "Pi diagnosis declared a technical blocker", {"decision": decision}) raise RuntimeError("Pi diagnosis declared a technical blocker") final_decision = {"action": "declare_technical_blocker", "reason": "Blockage protocol exhausted all bounded iterations.", "evidence": [], "patch_allowed": False} write_blockage_artifact(workspace, run_dir, events_path, final_decision, current_error) append_event(events_path, "failure", "failed", "Blockage protocol exhausted all bounded iterations") raise RuntimeError("Blockage protocol exhausted all bounded iterations") def upload_workspace(api, workspace: Path, target_space_id: str, token: str, run_dir: Path, events_path: Path): sanitize_readme_metadata(workspace, events_path) normalize_requirements_for_modern_hub(workspace, events_path) append_event(events_path, "upload_files", "started", "Uploading generated universal model-card workspace recursively") gen_dir = run_dir / "generated" if gen_dir.exists(): shutil.rmtree(gen_dir) shutil.copytree(workspace, gen_dir, ignore=internal_workspace_copy_ignore) for filename in ["app.py", "README.md", "requirements.txt"]: if not (workspace / filename).exists(): raise RuntimeError(f"Missing required generated file: {filename}") api.upload_folder( folder_path=str(workspace), repo_id=target_space_id, repo_type="space", token=token, ignore_patterns=internal_workspace_upload_ignore_patterns(), ) uploaded_files = sorted(str(p.relative_to(workspace)) for p in workspace.rglob("*") if p.is_file() and is_publishable_workspace_file(p.relative_to(workspace))) append_event(events_path, "upload_files", "success", "Uploaded generated workspace folder", {"file_count": len(uploaded_files), "files_sample": uploaded_files[:50]}) def load_json_if_exists(path: Path) -> dict: if not path.exists(): return {} try: return json.loads(path.read_text(encoding="utf-8", errors="replace")) except Exception as exc: 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, 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 performs model inference. In full-inference-gated mode we require either an actual implementation signal or a machine-readable blocker report. """ app_text = (workspace / "app.py").read_text(encoding="utf-8", errors="ignore") if (workspace / "app.py").exists() else "" summary_text = (workspace / "PI_SUMMARY.md").read_text(encoding="utf-8", errors="ignore") if (workspace / "PI_SUMMARY.md").exists() else "" 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) combined = (app_text + "\n" + summary_text).lower() blocked_markers = [ "full generation is not implemented", "full generation is intentionally not wired", "full inference is blocked", "returns a detailed diagnostic", "diagnostic report instead", "placeholder generator", "placeholder generation", "info-only", "not implemented", "cannot run in this environment", "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_health_passed(validation), "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"}: status = "full_inference_candidate_health_passed" 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." if blocker_detected and not blockers: blockers = { "full_inference_implemented": False, "source": "worker_heuristic_from_PI_SUMMARY_or_app.py", "blockers": [ { "type": "agent_declared_or_detected_blocker", "claim": "Pi-generated artifacts state that full inference is blocked/not implemented or generation returns diagnostics/placeholders.", "evidence": "See PI_SUMMARY.md and app.py in generated artifacts.", "severity": "blocking", } ], "required_investigations_for_next_run": [ "Check whether PyTorch SDPA can replace flash-attn calls.", "Check whether HF Kernels flash-attn2/3/4 can replace required flash-attn APIs.", "Verify whether 2-GPU context parallelism is strictly required or can be reduced to a single-GPU smoke test.", ], } (workspace / "TECHNICAL_BLOCKERS.json").write_text(json.dumps(blockers, indent=2, ensure_ascii=False) + "\n", encoding="utf-8") (run_dir / "generated" / "TECHNICAL_BLOCKERS.json").write_text(json.dumps(blockers, indent=2, ensure_ascii=False) + "\n", encoding="utf-8") gate = { "status": status, "message": message, "implementation_mode": implementation_mode, "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) append_event(events_path, "inference_gate", status, message, gate) return gate def main(): run_id = os.environ["RUN_ID"] hf_username = os.environ.get("HF_USERNAME", "unknown") bucket_source = os.environ.get("BUCKET_SOURCE", "unknown") output_root = Path(os.environ.get("OUTPUT_ROOT", "/output")) target_space_id = os.environ.get("TARGET_SPACE_ID", "") target_space_auto_generated = os.environ.get("TARGET_SPACE_AUTO_GENERATED", "false").lower() in {"1", "true", "yes", "on"} model_id = sanitize_model_id(os.environ.get("MODEL_ID", DEFAULT_MODEL_ID)) pi_model = os.environ.get("PI_MODEL", "Qwen/Qwen3-Coder-Next") preferred_hardware = normalize_auto_space_hardware(os.environ.get("PREFERRED_SPACE_HARDWARE"), DEFAULT_PREFERRED_SPACE_HARDWARE) fallback_hardware = normalize_auto_space_hardware(os.environ.get("FALLBACK_SPACE_HARDWARE"), DEFAULT_FALLBACK_SPACE_HARDWARE) try_zero_gpu_first = os.environ.get("TRY_ZERO_GPU_FIRST", "true").lower() in {"1", "true", "yes", "on"} 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") ensure_hf_token_context(run_dir, events_path) 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 builder Space creation", "model_id": model_id, "pi_model": pi_model, "pi_model_resolution": pi_model_resolution, "target_space": target_space_id, "created_by": hf_username, "bucket_source": bucket_source, "created_at": now(), "updated_at": now()}) if not token: fail(run_dir, events_path, "HF_TOKEN is missing from Job secrets") if not TARGET_RE.match(target_space_id): fail(run_dir, events_path, "Invalid TARGET_SPACE_ID", {"target_space_id": target_space_id}) try: install_python_deps(events_path) from huggingface_hub import HfApi api = HfApi(token=token) whoami = api.whoami(token=token) append_event(events_path, "auth", "success", "Authenticated inside Job", {"whoami_name": whoami.get("name")}) append_event(events_path, "model_analysis", "started", "Fetching model metadata", {"model_id": model_id}) info = api.model_info(model_id, token=token, files_metadata=True) siblings = [getattr(s, "rfilename", "") for s in (info.siblings or [])] analysis = {"model_id": model_id, "pipeline_tag": getattr(info, "pipeline_tag", None), "library_name": getattr(info, "library_name", None), "tags": list(getattr(info, "tags", []) or [])[:100], "siblings": siblings[:160], "default_model_target": model_id == DEFAULT_MODEL_ID, "preferred_hardware": preferred_hardware, "fallback_hardware": fallback_hardware, "allow_fixed_gpu_fallback": allow_fixed_gpu_fallback, "implementation_mode": implementation_mode} write_json(run_dir / "model_analysis.json", analysis) append_event(events_path, "model_analysis", "success", "Model metadata fetched", {"pipeline_tag": analysis["pipeline_tag"], "library_name": analysis["library_name"]}) create_initial_workspace(workspace, model_id, target_space_id, preferred_hardware, fallback_hardware, allow_fixed_gpu_fallback, implementation_mode, analysis) append_event(events_path, "workspace", "success", "Prepared universal model-card workspace", {"files": sorted(p.name for p in workspace.iterdir())}) install_pi(events_path) configure_pi(events_path, pi_model) append_event(events_path, "pi_run", "started", "Running Pi on universal model-card workspace", {"model": pi_model}) (run_dir / "logs").mkdir(parents=True, exist_ok=True) code, pi_out = run_cmd_streaming( ["pi", "-p", (workspace / "GOAL.md").read_text(encoding="utf-8")], cwd=workspace, timeout=2400, live_log_path=run_dir / "logs" / "pi_live_output.txt", events_path=events_path, run_dir=run_dir, step="pi_run", trace_phase="initial_build", ) (run_dir / "logs" / "pi_output.txt").write_text(pi_out, encoding="utf-8") if code != 0: 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:]}) 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") app_text = (workspace / "app.py").read_text(encoding="utf-8", errors="ignore") if "/health" not in app_text and "api_name=\"health\"" not in app_text and "api_name='health'" not in app_text: append_event(events_path, "pi_verification", "failed", "app.py does not appear to expose /health; injecting safe health endpoint is not implemented") fail(run_dir, events_path, "Pi output did not preserve a /health endpoint") append_event(events_path, "pi_verification", "success", "Pi output preserved health validation endpoint") append_event(events_path, "hardware_strategy", "started", "Creating Space with hardware-at-creation strategy", {"preferred_hardware": preferred_hardware, "fallback_hardware": fallback_hardware, "allow_fixed_gpu_fallback": allow_fixed_gpu_fallback}) hardware_strategy = create_space_with_hardware_strategy( api, target_space_id, token, preferred_hardware, fallback_hardware, allow_fixed_gpu_fallback, events_path, allow_auto_rename=target_space_auto_generated, try_zero_gpu_first=try_zero_gpu_first, ) target_space_id = hardware_strategy.get("target_space_id") or target_space_id selected_hardware = hardware_strategy.get("selected_hardware") or "default-cpu-or-existing" hardware_attempts = list(hardware_strategy.get("attempts") or []) requested_hardware_sequence = list(hardware_strategy.get("requested_sequence") or []) update_state(run_dir, { "run_id": run_id, "kind": "universal_model_card_builder", "status": "running", "message": "Private Space created; uploading generated workspace", "model_id": model_id, "pi_model": pi_model, "target_space": target_space_id, "target_space_url": f"https://huggingface.co/spaces/{target_space_id}", "selected_hardware": selected_hardware, "hardware_attempts": hardware_attempts, "requested_hardware_sequence": requested_hardware_sequence, "created_by": hf_username, "bucket_source": bucket_source, }) write_artifact_manifest(run_dir, reason="space_created") # Upload after create. If create_repo(space_hardware=...) succeeded, the build # starts directly on the requested hardware. If it fell back to CPU, the run # remains valid but will be marked manual_hardware_required when inference # signals indicate GPU is needed. upload_workspace(api, workspace, target_space_id, token, run_dir, events_path) write_artifact_manifest(run_dir, reason="workspace_uploaded") if selected_hardware == "default-cpu-or-existing": append_event(events_path, "hardware", "warning", "Automatic hardware-at-creation failed; Space is on default CPU unless user changes it manually", {"attempts": hardware_attempts}) write_json(run_dir / "hardware_attempts.json", {"selected_hardware": selected_hardware, "requested_sequence": requested_hardware_sequence, "attempts": hardware_attempts, "strategy": "create_repo_space_hardware_first", "try_zero_gpu_first": try_zero_gpu_first}) write_json(run_dir / "hardware_strategy.json", {"selected_hardware": selected_hardware, "requested_sequence": requested_hardware_sequence, "attempts": hardware_attempts, "manual_action_required": selected_hardware == "default-cpu-or-existing", "strategy": "create_repo_space_hardware_first", "try_zero_gpu_first": try_zero_gpu_first}) try: validation = validate_live_api(api, target_space_id, token, run_dir, events_path, timeout_s=1200) except Exception as validation_error: append_event(events_path, "failure_detected", "warning", "Initial live validation failed; entering Pi blockage diagnosis protocol", {"error": str(validation_error)[:2000]}) validation = recover_after_live_validation_failure( api, workspace, run_dir, events_path, pi_model=pi_model, target_space_id=target_space_id, model_id=model_id, token=token, failure_reason=str(validation_error), implementation_mode=implementation_mode, expected_output_type=expected_output_type, ) generation_smoke = None if implementation_mode in {"full-inference-gated", "full-inference-attempt"}: try: generation_smoke = run_generation_smoke(target_space_id, token, run_dir, events_path, expected_output_type, workspace=workspace) except Exception as smoke_error: 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), } generation_smoke.update(classify_generation_smoke_error(smoke_error)) write_json(run_dir / "tests" / "generation_smoke.json", generation_smoke) append_event(events_path, "generation_smoke", "failed", "Live generation smoke test failed; ZeroGPU duration was not measured", generation_smoke) else: generation_smoke = measured_zero_gpu_recommendation(None) | {"status": "skipped", "expected_output_type": expected_output_type} write_json(run_dir / "tests" / "generation_smoke.json", generation_smoke) 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 # hardware instead of pretending CPU/default hardware is enough. the existing-Space validation workflow # can then smoke-test generation after the user sets a GPU manually. manual_hw_required = selected_hardware == "default-cpu-or-existing" and inference_gate.get("status") not in {"technical_blocker", "health_only"} and ( inference_gate.get("implementation_signals", {}).get("has_spaces_gpu") or inference_gate.get("implementation_signals", {}).get("has_torch") or any((a.get("manual_action_required") for a in hardware_attempts if isinstance(a, dict))) ) if manual_hw_required: inference_gate = dict(inference_gate) inference_gate["status"] = "manual_hardware_required" inference_gate["message"] = "Space was generated and boots, but automatic ZeroGPU/fixed-GPU assignment failed. Set hardware manually, then run the existing-Space validation workflow." inference_gate["manual_hardware_required"] = True inference_gate["hardware_attempts"] = hardware_attempts write_json(run_dir / "inference_gate.json", inference_gate) append_event(events_path, "inference_gate", "manual_hardware_required", inference_gate["message"], inference_gate) collect_pi_traces(run_dir, events_path) final_state = { "run_id": run_id, "kind": "universal_model_card_builder", "status": inference_gate["status"], "message": inference_gate["message"], "model_id": model_id, "pi_model": pi_model, "target_space": target_space_id, "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, "bucket_source": bucket_source, } write_json(state_path, final_state) report = f"""# Agentic Space Factory — Universal Model-Card Builder Report Run ID: `{run_id}` Status: **{inference_gate['status']}** {inference_gate['message']} 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}` Hardware changes are best-effort with OAuth. If requests fail with 401/auth/billing errors, set the Space hardware manually and rerun validation. ```json {json.dumps(hardware_attempts, indent=2, ensure_ascii=False)} ``` ## Health validation The wrapper validated the live Space using HTTP `/health` first, with Gradio Client as fallback. This only proves bootability. ```json {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 {json.dumps(inference_gate, indent=2, ensure_ascii=False)} ``` ## Pi summary {(workspace / 'PI_SUMMARY.md').read_text(encoding='utf-8', errors='ignore') if (workspace / 'PI_SUMMARY.md').exists() else 'No PI_SUMMARY.md was produced.'} ## Safety - The target Space was created private. - No public publication was attempted. - Raw traces should remain private; redacted traces are stored separately. - If fallback fixed GPU was used or selected manually, review billing/hardware settings manually after the run. """ (run_dir / "report.md").write_text(report, encoding="utf-8") append_event(events_path, "report_write", "success", "Wrote report.md") write_artifact_manifest(run_dir, events_path=events_path, reason="final") try: publish_eval_record(run_dir, phase="final", events_path=events_path) append_event(events_path, "anonymous_eval", "success", "Published anonymized evaluation record locally for backend archive publishing", {"enabled": True, "publish_mode": "backend", "archive_publish_confirmed": False}) except Exception as eval_exc: append_event(events_path, "anonymous_eval", "warning", "Could not publish anonymized evaluation record", {"error": str(eval_exc)[:1000]}) append_event(events_path, "done", inference_gate["status"], "Universal model-card builder completed", {"target_space": target_space_id, "selected_hardware": selected_hardware, "gate_status": inference_gate["status"]}) except SystemExit: raise except Exception as exc: try: collect_pi_traces(run_dir, events_path) except Exception: pass fail(run_dir, events_path, "Universal model-card builder worker failed", {"error": str(exc)}) if __name__ == "__main__": main() ''' VALIDATE_EXISTING_SPACE_WORKER_SCRIPT = r''' import json import os import re import shutil import subprocess import sys import time from datetime import datetime, timezone from pathlib import Path TARGET_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{1,95}/[A-Za-z0-9][A-Za-z0-9._-]{1,95}$") def now(): return datetime.now(timezone.utc).isoformat() def write_json(path: Path, payload: dict): path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(payload, indent=2, ensure_ascii=False) + "\n", encoding="utf-8") def append_event(path: Path, step: str, status: str, message: str, data: dict | None = None): path.parent.mkdir(parents=True, exist_ok=True) event = {"ts": now(), "step": step, "status": status, "message": message, "data": data or {}} line = json.dumps(event, ensure_ascii=False) with path.open("a", encoding="utf-8") as f: f.write(line + "\n") print(line, flush=True) SAFE_SPACE_PYTHON_VERSIONS = {"3.10", "3.11", "3.12"} DEFAULT_SPACE_PYTHON_VERSION = "3.10" def normalize_space_python_version(value) -> tuple[str, bool, str]: raw = str(value or "").strip().strip("\"'") lowered = raw.lower().replace("python", "").strip() match = re.search(r"(3)\.(\d+)(?:\.\d+)?", lowered) if match: normalized = f"{match.group(1)}.{match.group(2)}" else: normalized = lowered if normalized in SAFE_SPACE_PYTHON_VERSIONS: return normalized, normalized != raw, "allowed" return DEFAULT_SPACE_PYTHON_VERSION, True, f"unsupported_or_ambiguous:{raw or 'missing'}" def requirements_has_package(lines: list[str], package: str) -> bool: wanted = package.lower().replace("_", "-") for line in lines: stripped = line.strip() if not stripped or stripped.startswith("#") or stripped.startswith("-") or "://" in stripped: continue name = re.split(r"[<>=!~;\[]", stripped, 1)[0].strip().lower().replace("_", "-") if name == wanted: return True return False def workspace_app_imports_torch(workspace: Path) -> bool: app_path = workspace / "app.py" if not app_path.exists(): return False text = app_path.read_text(encoding="utf-8", errors="ignore") return bool(re.search(r"(?m)^\s*(import\s+torch\b|from\s+torch\b)", text)) def redact_text(text: str | None) -> str: if not text: return "" value = text for secret_name in ["HF_TOKEN", "HUGGING_FACE_HUB_TOKEN"]: secret = os.environ.get(secret_name) if secret: value = value.replace(secret, "[REDACTED]") value = re.sub(r"Bearer\s+[A-Za-z0-9_\-.=]+", "Bearer [REDACTED]", value) value = re.sub(r"hf_[A-Za-z0-9_\-]{10,}", "hf_[REDACTED]", value) return value def eval_enabled() -> bool: return os.environ.get("ASF_EVAL_RECORD_ENABLED", "").strip().lower() in {"1", "true", "yes", "on"} def eval_safe_hash(value: str | None, prefix: str = "h") -> str: raw = (value or "").strip() salt = os.environ.get("ASF_EVAL_SALT", "").strip() if not raw: return "" if salt: digest = hmac.new(salt.encode("utf-8"), raw.encode("utf-8"), hashlib.sha256).hexdigest()[:24] else: digest = hashlib.sha256(raw.encode("utf-8")).hexdigest()[:24] return f"{prefix}_{digest}" def eval_load_json(path: Path) -> dict: if not path.exists(): return {} try: return json.loads(path.read_text(encoding="utf-8", errors="replace")) except Exception as exc: return {"_parse_error": str(exc)} def eval_load_events(path: Path) -> list[dict]: if not path.exists(): return [] events = [] for line in path.read_text(encoding="utf-8", errors="replace").splitlines(): try: evt = json.loads(line) if isinstance(evt, dict): events.append(evt) except Exception: pass return events def eval_parse_ts(value: str | None) -> float | None: if not value: return None try: return datetime.fromisoformat(str(value).replace("Z", "+00:00")).timestamp() except Exception: return None def eval_compact_timeline(events: list[dict]) -> list[dict]: stages: dict[str, dict] = {} order: list[str] = [] for evt in events: step = str(evt.get("step") or "unknown")[:80] status = str(evt.get("status") or "")[:80] ts = str(evt.get("ts") or "") tsv = eval_parse_ts(ts) if step not in stages: stages[step] = {"stage": step, "first_ts": ts, "last_ts": ts, "status": status, "event_count": 0} order.append(step) item = stages[step] item["last_ts"] = ts or item.get("last_ts") item["status"] = status or item.get("status") item["event_count"] = int(item.get("event_count") or 0) + 1 if tsv is not None: if "_first" not in item: item["_first"] = tsv item["_last"] = tsv timeline = [] for step in order: item = dict(stages[step]) first = item.pop("_first", None) last = item.pop("_last", None) if first is not None and last is not None and last >= first: item["duration_ms"] = int((last - first) * 1000) timeline.append(item) return timeline[-80:] def eval_redacted_tail(path: Path, limit: int = 2400) -> str: if os.environ.get("ASF_EVAL_INCLUDE_REDACTED_TAILS", "").strip().lower() not in {"1", "true", "yes", "on"}: return "" try: return redact_text(path.read_text(encoding="utf-8", errors="replace"))[-limit:] except Exception: return "" def eval_terminal_status(status: str, phase: str = "") -> bool: status = str(status or "").strip().lower() phase = str(phase or "").strip().lower() if phase in {"final", "failure"}: return True return status in { "done", "success", "failed", "cancelled", "canceled", "full_inference_success", "full_inference_candidate_health_passed", "health_only", "technical_blocker", "manual_hardware_required", } def eval_event_seen(events: list[dict], step: str, status: str | None = None) -> bool: for event in events or []: if str(event.get("step") or "") != step: continue if status is None or str(event.get("status") or "") == status: return True return False def eval_health_passed(inference_gate: dict, generation_smoke: dict, events: list[dict], status: str) -> bool: signals = inference_gate.get("implementation_signals") if isinstance(inference_gate.get("implementation_signals"), dict) else {} if signals.get("health_passed") is True: return True if str(status or "") in {"full_inference_success", "full_inference_candidate_health_passed", "health_only"}: return True if eval_event_seen(events, "api_validation", "success") or eval_event_seen(events, "repair_validation", "success"): return True return False def eval_generation_smoke_passed(inference_gate: dict, generation_smoke: dict) -> bool: signals = inference_gate.get("implementation_signals") if isinstance(inference_gate.get("implementation_signals"), dict) else {} return bool(signals.get("generation_smoke_passed") is True or generation_smoke.get("status") == "success") def eval_failure_summary(status: str, state: dict, inference_gate: dict, generation_smoke: dict, health_passed: bool, smoke_passed: bool) -> dict: failure_type = generation_smoke.get("failure_type") or inference_gate.get("failure_type") or "" failure_owner = generation_smoke.get("failure_owner") or inference_gate.get("failure_owner") or "" missing_argument = generation_smoke.get("missing_argument") or inference_gate.get("missing_argument") or "" reason = missing_argument or failure_type or str(state.get("message") or inference_gate.get("message") or "")[:240] if not failure_owner: if failure_type == "validator_request_error" or missing_argument: failure_owner = "factory_validator" elif smoke_passed: failure_owner = "" elif health_passed and str(generation_smoke.get("status") or "") == "failed": failure_owner = "unknown" return { "failure_type": failure_type, "failure_owner": failure_owner, "failure_reason": reason, "missing_argument": missing_argument, } def eval_verdict(status: str, phase: str, health_passed: bool, smoke_passed: bool, full_inference_verified: bool, inference_gate: dict) -> str: status = str(status or "").strip() if status in {"cancelled", "canceled"}: return "cancelled" if status == "manual_hardware_required" or inference_gate.get("manual_hardware_required"): return "manual_action_required" if status == "technical_blocker": return "technical_blocker" if full_inference_verified: return "success" if status in {"full_inference_candidate_health_passed", "health_only"} or (health_passed and not smoke_passed): return "partial_validation" if phase == "failure" or status in {"failed", "error"}: return "failed" if eval_terminal_status(status, phase): return "partial_validation" if health_passed else "failed" return "running" def publish_eval_record(run_dir: Path, *, phase: str, events_path: Path | None = None) -> dict: """Write an anonymized cross-run evaluation record to the optional eval bucket. This intentionally stores structured metrics and hashed identifiers, not raw generated code, prompts, tokens, user bucket paths, or target Space IDs. """ if not eval_enabled(): return {"enabled": False} # Jobs only write the anonymized eval record back into the user's run bucket. # The ASF backend later publishes this record to the operator eval bucket # mounted on the ASF Space. User Jobs must never mount or write the private # operator archive directly. run_id = os.environ.get("RUN_ID", run_dir.name) events = eval_load_events(events_path or (run_dir / "events.jsonl")) state = eval_load_json(run_dir / "state.json") analysis = eval_load_json(run_dir / "model_analysis.json") hardware_strategy = eval_load_json(run_dir / "hardware_strategy.json") hardware_attempts = eval_load_json(run_dir / "hardware_attempts.json") inference_gate = eval_load_json(run_dir / "inference_gate.json") generation_smoke = eval_load_json(run_dir / "tests" / "generation_smoke.json") space_runtime = eval_load_json(run_dir / "space_runtime.json") manifest = eval_load_json(run_dir / "artifact_manifest.json") repair_decision = eval_load_json(run_dir / "repair" / "REPAIR_DECISION.json") blockage = eval_load_json(run_dir / "repair" / "BLOCKAGE.json") pi_resolution = state.get("pi_model_resolution") if isinstance(state.get("pi_model_resolution"), dict) else {} status = str(state.get("status") or inference_gate.get("status") or phase or "unknown") failure_message = str(state.get("message") or (state.get("details") or {}).get("error") or "")[:1200] started_at = events[0].get("ts") if events else (state.get("created_at") or os.environ.get("LAUNCHED_AT") or "") finished_at = state.get("updated_at") or (events[-1].get("ts") if events else now()) model_id = os.environ.get("MODEL_ID") or state.get("model_id") or analysis.get("model_id") or "" target_space = state.get("target_space") or os.environ.get("TARGET_SPACE_ID") or "" include_model_id = os.environ.get("ASF_EVAL_INCLUDE_MODEL_ID", "").strip().lower() in {"1", "true", "yes", "on"} anon_run_id = eval_safe_hash(run_id, "run") health_passed = eval_health_passed(inference_gate, generation_smoke, events, status) generation_smoke_passed = eval_generation_smoke_passed(inference_gate, generation_smoke) full_inference_verified = bool(status == "full_inference_success" or (health_passed and generation_smoke_passed)) process_completed = eval_terminal_status(status, phase) failure_summary = eval_failure_summary(status, state, inference_gate, generation_smoke, health_passed, generation_smoke_passed) verdict = eval_verdict(status, phase, health_passed, generation_smoke_passed, full_inference_verified, inference_gate) record = { "process_completed": process_completed, "verdict": verdict, "failure_owner": failure_summary.get("failure_owner") or "", "failure_reason": failure_summary.get("failure_reason") or "", "schema_version": "1.2", "app_version": os.environ.get("ASF_VERSION", "unknown"), "phase": phase, "anonymous_run_id": anon_run_id, "run_id_hash": eval_safe_hash(run_id, "run"), "anonymous_user_hash": eval_safe_hash(os.environ.get("HF_USERNAME"), "user"), "salt_configured": bool(os.environ.get("ASF_EVAL_SALT")), "run_kind": state.get("kind") or ("validate_existing_space" if os.environ.get("API_NAME") else "validate_existing_space"), "started_at": started_at, "finished_at": finished_at, "input_model": { "model_id_redacted": not include_model_id, "model_id": model_id if include_model_id else "", "model_hash": eval_safe_hash(model_id, "model"), "pipeline_tag": analysis.get("pipeline_tag"), "library_name": analysis.get("library_name"), "default_model_target": bool(analysis.get("default_model_target")), }, "config": { "implementation_mode": os.environ.get("IMPLEMENTATION_MODE", ""), "expected_output_type": os.environ.get("EXPECTED_OUTPUT_TYPE", ""), "pi_model_requested": os.environ.get("PI_MODEL") or state.get("pi_model") or "", "pi_model_effective": pi_resolution.get("effective_model") or pi_resolution.get("observed_model") or "", "try_zero_gpu_first": os.environ.get("TRY_ZERO_GPU_FIRST", ""), "allow_fixed_gpu_fallback": os.environ.get("ALLOW_FIXED_GPU_FALLBACK", ""), "preferred_gpu": os.environ.get("PREFERRED_SPACE_HARDWARE", ""), "fallback_gpu": os.environ.get("FALLBACK_SPACE_HARDWARE", ""), }, "outcome": { "status": status, "verdict": verdict, "process_completed": process_completed, "health_passed": health_passed, "generation_smoke_passed": generation_smoke_passed, "full_inference_verified": full_inference_verified, "failure_owner": failure_summary.get("failure_owner") or "", "failure_reason": failure_summary.get("failure_reason") or "", "message_redacted": redact_text(failure_message)[:1200], "target_space_hash": eval_safe_hash(target_space, "space"), "space_created": bool(target_space), "space_runtime_stage": space_runtime.get("stage"), "space_runtime_hardware": space_runtime.get("hardware") or space_runtime.get("requested_hardware"), }, "timeline": eval_compact_timeline(events), "hardware": { "selected": hardware_strategy.get("selected_hardware") or state.get("selected_hardware") or "", "strategy": hardware_strategy.get("strategy") or hardware_attempts.get("strategy") or "", "attempts": hardware_strategy.get("attempts") or hardware_attempts.get("attempts") or state.get("hardware_attempts") or [], "single_space_per_run": True, }, "pi": { "requested_model": os.environ.get("PI_MODEL") or state.get("pi_model") or "", "effective_model": pi_resolution.get("effective_model") or pi_resolution.get("observed_model") or "", "model_mismatch": bool(pi_resolution.get("mismatch")), "build_log_tail_redacted": eval_redacted_tail(run_dir / "logs" / "pi_output.txt"), "diagnosis_log_tail_redacted": eval_redacted_tail(run_dir / "logs" / "pi_diagnosis_output.txt"), }, "recovery": { "entered": any(str(e.get("step", "")).startswith(("repair", "failure_diagnosis", "pi_diagnosis", "technical_blocker")) for e in events), "decision": repair_decision.get("action") or repair_decision.get("decision") or "", "confidence": repair_decision.get("confidence") or "", "blockage": bool(blockage), "blockage_category": blockage.get("category") or blockage.get("reason") or "", }, "validation": { "health_passed": health_passed, "generation_smoke_passed": generation_smoke_passed, "full_inference_verified": full_inference_verified, "generation_smoke_status": generation_smoke.get("status") or "", "latency_seconds": generation_smoke.get("latency_seconds") or generation_smoke.get("observed_latency_seconds"), "expected_output_type": generation_smoke.get("expected_output_type") or os.environ.get("EXPECTED_OUTPUT_TYPE", ""), "inference_gate_status": inference_gate.get("status") or "", "manual_hardware_required": bool(inference_gate.get("manual_hardware_required")), "failure_type": failure_summary.get("failure_type") or "", "failure_owner": failure_summary.get("failure_owner") or "", "failure_reason": failure_summary.get("failure_reason") or "", "missing_argument": failure_summary.get("missing_argument") or "", }, "artifacts": { "present_count": len(manifest.get("present_paths") or []), "present_paths": sorted(str(p) for p in (manifest.get("present_paths") or []))[:120], }, "privacy": { "tokens_stored": False, "raw_prompts_stored": False, "generated_code_stored": False, "target_space_redacted": True, "user_bucket_redacted": True, "redacted_tails_enabled": os.environ.get("ASF_EVAL_INCLUDE_REDACTED_TAILS", "").strip().lower() in {"1", "true", "yes", "on"}, }, } day = (finished_at or now())[:10] try: yyyy, mm, dd = day.split("-") except Exception: yyyy, mm, dd = now()[:10].split("-") dest = run_dir / "eval_record.json" write_json(dest, record) events_dest = run_dir / "events_compact.jsonl" events_dest.write_text("".join(json.dumps(item, ensure_ascii=False) + "\n" for item in record["timeline"]), encoding="utf-8") return {"enabled": True, "written": True, "path": str(dest), "publish_mode": "backend"} def ensure_hf_token_context(run_dir: Path, events_path: Path | None = None) -> dict: """Ensure Pi and HF tooling see the same private HF token safely. HF Jobs receive HF_TOKEN as a secret. Pi, huggingface_hub, gradio_client, and the hf CLI can use either HF_TOKEN or HUGGING_FACE_HUB_TOKEN depending on the tool/version, so expose both aliases inside the isolated Job process. Never write the token value to artifacts. """ token = os.environ.get("HF_TOKEN") or os.environ.get("HUGGING_FACE_HUB_TOKEN") or "" if token: os.environ.setdefault("HF_TOKEN", token) os.environ.setdefault("HUGGING_FACE_HUB_TOKEN", token) os.environ.setdefault("HF_HUB_ENABLE_HF_TRANSFER", "1") payload = { "hf_token_present": bool(token), "hf_token_length": len(token) if token else 0, "hugging_face_hub_token_alias_present": bool(os.environ.get("HUGGING_FACE_HUB_TOKEN")), "hf_username": os.environ.get("HF_USERNAME") or "", "bucket_source": os.environ.get("BUCKET_SOURCE") or "", "target_space_id": os.environ.get("TARGET_SPACE_ID") or "", "token_value": "[REDACTED]" if token else "", } run_dir.mkdir(parents=True, exist_ok=True) write_json(run_dir / "token_context.json", payload) if events_path: append_event(events_path, "token_context", "success" if token else "warning", "Verified HF token context for Pi/HF operations", {k: v for k, v in payload.items() if k != "token_value"}) return payload def pi_tooling_context_note() -> str: return """HF tooling context: - You are running inside an HF Job with an HF_TOKEN secret configured. - The Factory exposes both HF_TOKEN and HUGGING_FACE_HUB_TOKEN aliases for huggingface_hub, gradio_client, hf CLI, and Pi provider access. - You may use the token through normal tools when needed to inspect the model/Space or validate private resources. - Never print, write, echo, or commit token values. Do not include Authorization headers in artifacts. - If a Hub operation fails with 401/403/quota/billing, report it as evidence instead of retrying blindly. """ def run_cmd(cmd: list[str], *, env: dict | None = None, timeout: int = 120): result = subprocess.run(cmd, env=env, text=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, timeout=timeout) return result.returncode, redact_text(result.stdout) def install_deps(events_path: Path): append_event(events_path, "dependencies", "started", "Installing validation dependencies") code, out = run_cmd([sys.executable, "-m", "pip", "install", "-q", "--upgrade", "huggingface_hub>=1.0.0", "gradio_client>=2.0.0", "requests>=2.31.0"], timeout=600) if code != 0: append_event(events_path, "dependencies", "failed", "Dependency installation failed", {"output_tail": out[-4000:]}) raise RuntimeError(out) append_event(events_path, "dependencies", "success", "Validation dependencies installed") def make_gradio_client(target_space_id: str, token: str): import inspect from gradio_client import Client params = inspect.signature(Client).parameters if "token" in params: return Client(target_space_id, token=token) if "hf_token" in params: return Client(target_space_id, hf_token=token) if "api_key" in params: return Client(target_space_id, api_key=token) if "headers" in params: return Client(target_space_id, headers={"Authorization": f"Bearer {token}"}) return Client(target_space_id) def api_names_from_schema(schema) -> list[str]: names: list[str] = [] def add(name): if not isinstance(name, str) or not name: return if not name.startswith("/"): name = "/" + name if name not in names: names.append(name) def walk(obj): if isinstance(obj, dict): for k, v in obj.items(): if k in {"api_name", "apiName"}: add(v) if isinstance(k, str) and k.startswith("/"): add(k) walk(v) elif isinstance(obj, list): for item in obj: walk(item) walk(schema) 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 endpoint_parameter_objects(endpoint) -> list: params: list = [] def add_param(param): if isinstance(param, (dict, str)): params.append(param) def walk(obj): if isinstance(obj, dict): direct = obj.get("parameters") or obj.get("inputs") or obj.get("input_components") or obj.get("components") if isinstance(direct, list): for param in direct: add_param(param) return 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 params def _schema_text(value) -> str: if value is None: return "" if isinstance(value, (str, int, float, bool)): return str(value) if isinstance(value, dict): for key in ["label", "name", "type", "component", "component_type", "type_name", "class_name", "__type__", "datatype", "python_type", "value"]: text = _schema_text(value.get(key)) if text: return text api_info = value.get("api_info") if isinstance(value.get("api_info"), dict) else {} for key in ["type", "component", "label"]: text = _schema_text(api_info.get(key)) if text: return text return "" def _schema_name(param, index: int) -> str: if isinstance(param, str): return param if isinstance(param, dict): for key in ["parameter_name", "parameterName", "name", "label", "id"]: text = _schema_text(param.get(key)) if text: return text component = param.get("component") if isinstance(param.get("component"), dict) else {} for key in ["label", "name"]: text = _schema_text(component.get(key)) if text: return text return f"arg{index}" def _schema_component(param) -> str: if isinstance(param, str): return "" if isinstance(param, dict): return " ".join(filter(None, [ _schema_text(param.get("component")), _schema_text(param.get("component_type")), _schema_text(param.get("type")), _schema_text(param.get("datatype")), _schema_text(param.get("api_info")), ])).lower() return "" def _schema_choices(param) -> list: if not isinstance(param, dict): return [] raw = param.get("choices") or param.get("options") or (param.get("api_info") or {}).get("choices") or (param.get("component") or {}).get("choices") or [] if not isinstance(raw, list): return [] out = [] for choice in raw: text = _schema_text(choice) if text: out.append(text) return out def _schema_default(param): if not isinstance(param, dict): return None sentinel = object() for key in ["default", "value", "example_input"]: value = param.get(key, sentinel) if value is not sentinel: return value api_info = param.get("api_info") if isinstance(param.get("api_info"), dict) else {} if "default" in api_info: return api_info.get("default") component = param.get("component") if isinstance(param.get("component"), dict) else {} if "value" in component: return component.get("value") return None def _coerce_number(value, fallback): try: if value is None or value == "": return fallback number = float(value) return int(number) if number.is_integer() else number except Exception: return fallback def autofill_value_for_schema_param(param, index: int): name = _schema_name(param, index).lower().replace(" ", "_") component = _schema_component(param) default = _schema_default(param) choices = _schema_choices(param) if default is not None and not isinstance(default, (dict, list)): return default if choices: return choices[0] if any(token in component for token in ["file", "upload", "audio", "video"]): raise ValueError(f"Cannot auto-fill required file/media input: {_schema_name(param, index)}") if "image" in component and not any(token in name for token in ["prompt", "negative"]): raise ValueError(f"Cannot auto-fill required image input: {_schema_name(param, index)}") if "negative" in name and "prompt" in name: return "" if "prompt" in name or "text" in component or "textbox" in component: return "a cinematic robot cat astronaut, detailed, studio lighting" if "random_seed" in name: return True if "checkbox" in component or "bool" in component: return False if "seed" in name: return 42 if "height" in name or "width" in name or "resolution" in name: return 1024 if "step" in name or "num_inference" in name: return 9 if "guidance" in name or "cfg" in name: return 0.0 if "slider" in component or "number" in component or "int" in component or "float" in component: return _coerce_number(default, 0) return "test" def build_args_from_gradio_schema(api_name: str, schema, current_args: list) -> tuple[list, dict]: endpoint = endpoint_schema_for_api(schema, api_name) params = endpoint_parameter_objects(endpoint) if not params: return current_args, {"args_were_autofilled": False, "args_source": "provided_or_default", "endpoint_parameters": endpoint_parameter_names(endpoint)} if isinstance(current_args, list) and len(current_args) >= len(params): return current_args, {"args_were_autofilled": False, "args_source": "provided", "endpoint_parameters": [_schema_name(p, i) for i, p in enumerate(params)]} generated = [] for index, param in enumerate(params): if isinstance(current_args, list) and index < len(current_args): generated.append(current_args[index]) else: generated.append(autofill_value_for_schema_param(param, index)) return generated, {"args_were_autofilled": True, "args_source": "gradio_schema_auto_fill", "endpoint_parameters": [_schema_name(p, i) for i, p in enumerate(params)]} def runtime_to_dict(runtime) -> dict: payload = {} for attr in ["stage", "hardware", "requested_hardware", "sleep_time", "storage", "gc_timeout"]: value = getattr(runtime, attr, None) payload[attr] = getattr(value, "value", value) return {k: str(v) if v is not None else None for k, v in payload.items()} def write_space_runtime(api, target_space_id: str, token: str, run_dir: Path, events_path: Path, attempt: int | None = None) -> dict: try: runtime = api.get_space_runtime(repo_id=target_space_id, token=token) payload = runtime_to_dict(runtime) payload["attempt"] = attempt write_json(run_dir / "space_runtime.json", payload) return payload except Exception as exc: payload = {"error": str(exc)[:2000], "attempt": attempt} write_json(run_dir / "space_runtime.json", payload) append_event(events_path, "space_runtime", "warning", "Could not fetch Space runtime", payload) return payload RESTART_GUARDRAIL_STATE = { "restart": {"count": 0, "last_ts": 0.0}, "factory_reboot": {"count": 0, "last_ts": 0.0}, "same_code_upload": {"count": 0, "last_ts": 0.0}, } RESTART_GUARDRAIL_LIMITS = { "restart": 1, "factory_reboot": 1, "same_code_upload": 1, } RESTART_GUARDRAIL_COOLDOWN_SECONDS = int(os.environ.get("SPACE_FACTORY_RESTART_COOLDOWN_SECONDS", "240")) def runtime_stage_value(runtime_payload: dict | None) -> str: return str((runtime_payload or {}).get("stage") or (runtime_payload or {}).get("runtime_stage") or "").lower() def restart_guardrail_allows(action: str, events_path: Path, *, reason: str = "") -> bool: state = RESTART_GUARDRAIL_STATE.setdefault(action, {"count": 0, "last_ts": 0.0}) limit = RESTART_GUARDRAIL_LIMITS.get(action, 1) now = time.time() if state["count"] >= limit: append_event(events_path, "factory_rebuild", "warning", f"Blocked {action}: restart budget exhausted", {"action": action, "limit": limit, "reason": reason}) return False if state["last_ts"] and now - state["last_ts"] < RESTART_GUARDRAIL_COOLDOWN_SECONDS: append_event(events_path, "factory_rebuild", "warning", f"Blocked {action}: restart cooldown active", {"action": action, "cooldown_seconds": RESTART_GUARDRAIL_COOLDOWN_SECONDS, "elapsed_seconds": round(now - state["last_ts"], 1), "reason": reason}) return False state["count"] += 1 state["last_ts"] = now return True def safe_restart_space(api, target_space_id: str, token: str, run_dir: Path, events_path: Path, *, factory_reboot: bool = False, reason: str = "", require_logs_checked: bool = True) -> bool: """Restart a Space with strict guardrails. Hugging Face exposes HfApi.restart_space(repo_id=..., factory_reboot=True) for a factory rebuild and restart_space(..., factory_reboot=False) for a normal restart. The Factory only calls it through this wrapper so we do not create restart loops or mask actionable build errors. """ action = "factory_reboot" if factory_reboot else "restart" if require_logs_checked: logs_dir = run_dir / "logs" if not ((logs_dir / "space_logs_build.txt").exists() or (logs_dir / "space_logs_runtime.txt").exists()): append_event(events_path, "factory_rebuild", "warning", f"Skipped {action}: logs have not been collected yet", {"reason": reason}) return False runtime_before = write_space_runtime(api, target_space_id, token, run_dir, events_path) stage = runtime_stage_value(runtime_before) if stage in {"building", "buildqueued", "build_queued", "starting", "pending"}: append_event(events_path, "factory_rebuild", "warning", f"Skipped {action}: Space is already busy", {"runtime": runtime_before, "reason": reason}) return False if not restart_guardrail_allows(action, events_path, reason=reason): return False try: append_event(events_path, "factory_rebuild", "started", f"Requesting {'factory reboot' if factory_reboot else 'Space restart'} via Hugging Face API", {"runtime_before": runtime_before, "reason": reason}) result = api.restart_space(repo_id=target_space_id, token=token, factory_reboot=factory_reboot) payload = {"factory_reboot": factory_reboot, "reason": reason, "result": str(result)[:1000]} write_json(run_dir / ("space_factory_reboot.json" if factory_reboot else "space_restart.json"), payload) append_event(events_path, "factory_rebuild", "success", f"{'Factory reboot' if factory_reboot else 'Space restart'} requested", payload) return True except TypeError: # Older huggingface_hub versions may not expose factory_reboot. Do not # loop; fall back to a plain restart only for non-factory requests. if factory_reboot: append_event(events_path, "factory_rebuild", "warning", "Installed huggingface_hub does not support restart_space(factory_reboot=True)", {"reason": reason}) return False try: result = api.restart_space(repo_id=target_space_id, token=token) append_event(events_path, "factory_rebuild", "success", "Space restart requested", {"reason": reason, "result": str(result)[:1000]}) return True except Exception as exc: append_event(events_path, "factory_rebuild", "failed", "Space restart API call failed", {"reason": reason, **exception_payload(exc)}) return False except Exception as exc: append_event(events_path, "factory_rebuild", "failed", f"{'Factory reboot' if factory_reboot else 'Space restart'} API call failed", {"reason": reason, **exception_payload(exc)}) return False def safe_same_code_reupload(api, workspace: Path, target_space_id: str, token: str, run_dir: Path, events_path: Path, *, reason: str = "") -> bool: """Re-upload the current workspace once, with restart-loop guardrails.""" if not restart_guardrail_allows("same_code_upload", events_path, reason=reason): return False runtime_before = write_space_runtime(api, target_space_id, token, run_dir, events_path) stage = runtime_stage_value(runtime_before) if stage in {"building", "buildqueued", "build_queued", "starting", "pending"}: append_event(events_path, "factory_rebuild", "warning", "Skipped same-code upload: Space is already building/starting", {"runtime": runtime_before, "reason": reason}) return False upload_workspace(api, workspace, target_space_id, token, run_dir, events_path) append_event(events_path, "factory_rebuild", "success", "Same-code workspace uploaded under restart guardrails", {"reason": reason, "runtime_before": runtime_before}) return True def collect_space_logs(target_space_id: str, token: str, run_dir: Path, events_path: Path): """Collect Space build/runtime logs with explicit capability detection and an index. This is intentionally more structured than the old best-effort collector: - runtime state is always captured when possible; - runtime/build log collection first uses HfApi.get_space_logs when available; - if the installed huggingface_hub lacks that method, REST fallbacks are attempted; - legacy text files are still written, but availability/quality lives in space_logs_index.json. """ from huggingface_hub import HfApi import json as _json import sys as _sys try: import huggingface_hub as _hf_hub hub_version = getattr(_hf_hub, "__version__", None) except Exception: hub_version = None logs_dir = run_dir / "logs" logs_dir.mkdir(parents=True, exist_ok=True) api = HfApi(token=token) collected_at = now() written = [] index = { "schema_version": "1.0", "target_space_id": target_space_id, "collected_at": collected_at, "huggingface_hub_version": hub_version, "python_version": _sys.version.split()[0], "capabilities": { "hfapi_get_space_runtime": hasattr(api, "get_space_runtime"), "hfapi_get_space_logs": hasattr(api, "get_space_logs"), "rest_fallback": True, }, "entries": {}, } def _tail(text: str, n: int = 1200) -> str: return (text or "")[-n:] def _stringify_log_response(out) -> str: if out is None: return "" if isinstance(out, (list, tuple)): return "\n".join(str(x) for x in out) if isinstance(out, (dict, list)): return _json.dumps(out, ensure_ascii=False, indent=2) return str(out) def _write_text(filename: str, text: str) -> int: path = logs_dir / filename path.write_text(text or "", encoding="utf-8") return path.stat().st_size if path.exists() else 0 def _entry(key: str, *, filename: str, available: bool, source: str, quality: str, reason: str = "", error: str = "", text: str = ""): size = _write_text(filename, text) payload = { "available": bool(available), "path": f"logs/{filename}", "source": source, "quality": quality, "reason": reason, "error": error[:1500] if error else "", "size_bytes": size, "tail": _tail(text), } index["entries"][key] = payload written.append({ "file": filename, "source": source, "available": bool(available), "quality": quality, "reason": reason, "error": error[:1000] if error else "", "returncode": 0 if available else 1, "tail": _tail(text, 1000), }) return payload def _unavailable_text(kind: str, reason: str, error: str = "") -> str: return ( f"[ASF_LOG_UNAVAILABLE]\n" f"kind={kind}\n" f"reason={reason}\n" f"target_space_id={target_space_id}\n" f"huggingface_hub_version={hub_version}\n" f"error={error[:1500] if error else ''}\n" "See logs/space_logs_index.json for structured availability metadata.\n" ) # Runtime state / snapshot. This is not a log stream, but it is critical context for # BUILD_ERROR/RUNTIME_ERROR triage and is always tracked separately from log streams. try: runtime_obj = api.get_space_runtime(repo_id=target_space_id, token=token) runtime_payload = runtime_to_dict(runtime_obj) runtime_payload.update({"target_space_id": target_space_id, "collected_at": collected_at}) runtime_text = _json.dumps(runtime_payload, indent=2, ensure_ascii=False) _entry("runtime_snapshot", filename="space_runtime_snapshot.json", available=True, source="HfApi.get_space_runtime", quality="snapshot", text=runtime_text) # Legacy path retained because older reports/tests point to this file. _entry("runtime_state_legacy", filename="space_runtime_state.json", available=True, source="HfApi.get_space_runtime", quality="snapshot", text=runtime_text) except Exception as exc: err = f"{type(exc).__name__}: {str(exc)[:1500]}" _entry("runtime_snapshot", filename="space_runtime_snapshot.json", available=False, source="HfApi.get_space_runtime", quality="unavailable", reason="runtime_snapshot_error", error=err, text=_unavailable_text("runtime_snapshot", "runtime_snapshot_error", err)) _entry("runtime_state_legacy", filename="space_runtime_state.json", available=False, source="HfApi.get_space_runtime", quality="unavailable", reason="runtime_snapshot_error", error=err, text=_unavailable_text("runtime_state", "runtime_snapshot_error", err)) def _collect_via_hfapi(kind: str): if not hasattr(api, "get_space_logs"): return None, "hub_method_unavailable", "" try: out = api.get_space_logs(repo_id=target_space_id, token=token, build=(kind == "build")) text = _stringify_log_response(out) if text.strip(): return text, "", "" return "", "empty_response", "" except Exception as exc: return None, "hfapi_error", f"{type(exc).__name__}: {str(exc)[:1500]}" def _collect_via_rest(kind: str): # Private/unstable endpoints vary across hub deployments, so this is best-effort. # We still try because it can recover logs when the Python SDK lacks get_space_logs. try: import requests except Exception as exc: return None, "requests_unavailable", f"{type(exc).__name__}: {str(exc)[:1500]}" headers = {"Authorization": f"Bearer {token}", "Accept": "text/plain,application/json,*/*"} if token else {"Accept": "text/plain,application/json,*/*"} encoded = target_space_id candidates = [ f"https://huggingface.co/api/spaces/{encoded}/logs?build={'true' if kind == 'build' else 'false'}", f"https://huggingface.co/api/spaces/{encoded}/logs/{kind}", ] last_error = "" for url in candidates: try: response = requests.get(url, headers=headers, timeout=20) body = response.text or "" if response.ok and body.strip(): ctype = response.headers.get("content-type", "") return body, "", f"{url} ({response.status_code}, {ctype})" last_error = f"{url} returned {response.status_code}: {body[:500]}" except Exception as exc: last_error = f"{url} failed: {type(exc).__name__}: {str(exc)[:500]}" return None, "rest_logs_unavailable", last_error def _collect_stream(kind: str, filename: str): attempted = [] text, reason, error = _collect_via_hfapi(kind) attempted.append({"method": "HfApi.get_space_logs", "reason": reason, "error": error[:500] if error else ""}) if text is not None and text.strip(): return _entry(kind, filename=filename, available=True, source="HfApi.get_space_logs", quality="full", text=text) rest_text, rest_reason, rest_error = _collect_via_rest(kind) attempted.append({"method": "hf_rest_logs", "reason": rest_reason, "error": rest_error[:500] if rest_error else ""}) if rest_text is not None and rest_text.strip(): source = "hf_rest_logs" if rest_error and rest_error.startswith("https://"): source = rest_error return _entry(kind, filename=filename, available=True, source=source, quality="full", text=rest_text) # Keep legacy files present, but explicitly mark them as unavailable rather than # pretending an SDK AttributeError is a Space runtime/build log. This also avoids # the obsolete CLI path that emitted "invalid choice: spaces". final_reason = rest_reason or reason or "logs_unavailable" final_error = rest_error or error or "" entry = _entry(kind, filename=filename, available=False, source="log_collection", quality="unavailable", reason=final_reason, error=final_error, text=_unavailable_text(kind, final_reason, final_error)) entry["attempted_methods"] = attempted return entry runtime_entry = _collect_stream("runtime", "space_logs_runtime.txt") build_entry = _collect_stream("build", "space_logs_build.txt") def _first_error_from(text: str) -> str: if not text: return "" lines = text.splitlines() lowered = [line.lower() for line in lines] markers = ("traceback", "exception", "error", "exit code", "failed", "runtimeerror", "attributeerror", "modulenotfounderror") for idx, line in enumerate(lowered): if any(marker in line for marker in markers): start = max(0, idx - 3) end = min(len(lines), idx + 12) return "\n".join(lines[start:end])[:4000] return "" runtime_text = (logs_dir / "space_logs_runtime.txt").read_text(encoding="utf-8", errors="ignore") if runtime_entry.get("available") and (logs_dir / "space_logs_runtime.txt").exists() else "" build_text = (logs_dir / "space_logs_build.txt").read_text(encoding="utf-8", errors="ignore") if build_entry.get("available") and (logs_dir / "space_logs_build.txt").exists() else "" first_error = _first_error_from(runtime_text) or _first_error_from(build_text) log_quality = "full" if runtime_entry.get("available") or build_entry.get("available") else "snapshot_only" if index["entries"].get("runtime_snapshot", {}).get("available") else "unavailable" diagnostics = { "schema_version": "1.0", "target_space_id": target_space_id, "collected_at": collected_at, "log_quality": log_quality, "runtime_logs_available": bool(runtime_entry.get("available")), "build_logs_available": bool(build_entry.get("available")), "first_error": first_error, "diagnosis_log_source": "runtime_or_build_logs" if first_error else ("runtime_snapshot" if log_quality == "snapshot_only" else "none"), } diagnostics_text = _json.dumps(diagnostics, ensure_ascii=False, indent=2) _entry("diagnostics", filename="space_log_diagnostics.json", available=True, source="asf_log_collector", quality=log_quality, text=diagnostics_text) index["log_quality"] = log_quality index["runtime_logs_available"] = bool(runtime_entry.get("available")) index["build_logs_available"] = bool(build_entry.get("available")) index["first_error"] = first_error[:1000] index_text = _json.dumps(index, ensure_ascii=False, indent=2) (logs_dir / "space_logs_index.json").write_text(index_text, encoding="utf-8") written.append({"file": "space_logs_index.json", "source": "asf_log_collector", "available": True, "quality": log_quality, "returncode": 0, "tail": _tail(index_text, 1000)}) event_status = "success" if runtime_entry.get("available") or build_entry.get("available") else "warning" event_message = "Collected Space logs and runtime snapshot" if event_status == "success" else "Space log streams unavailable; runtime snapshot/index written" append_event(events_path, "space_logs", event_status, event_message, {"files": written, "log_quality": log_quality, "index": "logs/space_logs_index.json"}) write_artifact_manifest(run_dir, reason="space_logs_collected") return written def space_subdomain_url(target_space_id: str) -> str: owner, name = target_space_id.split("/", 1) return f"https://{owner}-{name}.hf.space".replace("_", "-").lower() def validate_http_health(target_space_id: str, token: str, run_dir: Path, attempt: int): import requests url = space_subdomain_url(target_space_id).rstrip("/") + "/health" headers = {"Authorization": f"Bearer {token}", "Accept": "application/json,text/plain,*/*"} response = requests.get(url, headers=headers, timeout=20) payload = { "status": "success" if response.ok else "failed", "attempt": attempt, "url": url, "status_code": response.status_code, "content_type": response.headers.get("content-type"), "text": response.text[:2000], } if response.ok: try: payload["json"] = response.json() except Exception: pass write_json(run_dir / "tests" / "http_health.json", payload) return payload raise RuntimeError(f"HTTP /health returned {response.status_code}: {response.text[:500]}") def wait_until_live(api, target_space_id: str, token: str, run_dir: Path, events_path: Path, timeout_s: int = 1800): append_event(events_path, "live_wait", "started", "Waiting for existing Space to become live") deadline = time.time() + timeout_s attempt = 0 last_error = None while time.time() < deadline: attempt += 1 runtime_payload = write_space_runtime(api, target_space_id, token, run_dir, events_path, attempt) stage = str(runtime_payload.get("stage") or "").upper() if "RUNTIME_ERROR" in stage: collect_space_logs(target_space_id, token, run_dir, events_path) last_error = f"Space is in RUNTIME_ERROR: {runtime_payload}" append_event(events_path, "live_wait", "waiting", "Space is in runtime error; still waiting in case hardware was changed manually", {"attempt": attempt, "runtime": runtime_payload}) time.sleep(30) continue try: health = validate_http_health(target_space_id, token, run_dir, attempt) append_event(events_path, "live_wait", "success", "HTTP /health is live", {"attempt": attempt}) return {"validator": "http_health", "health": health, "runtime": runtime_payload} except Exception as http_exc: last_error = f"HTTP health failed: {http_exc}" try: client = make_gradio_client(target_space_id, token) schema = client.view_api(return_format="dict") names = api_names_from_schema(schema) write_json(run_dir / "tests" / "api_schema.json", {"schema": schema, "api_names": names}) if names: append_event(events_path, "live_wait", "success", "Gradio API schema is live", {"attempt": attempt, "api_names": names}) return {"validator": "gradio_schema", "api_names": names, "runtime": runtime_payload} except Exception as gr_exc: last_error = (last_error or "") + f"; Gradio schema failed: {gr_exc}" append_event(events_path, "live_wait", "waiting", "Space not live yet", {"attempt": attempt, "runtime": runtime_payload, "error": last_error[-1500:] if last_error else None}) time.sleep(30) collect_space_logs(target_space_id, token, run_dir, events_path) raise RuntimeError(f"Space did not become live before timeout: {last_error}") def parse_json_env(name: str, default): value = os.environ.get(name) if not value: return default try: return json.loads(value) except Exception as exc: raise ValueError(f"Invalid JSON for {name}: {exc}") def result_contains_expected_output(result, expected_output_type: str) -> tuple[bool, dict]: expected = (expected_output_type or "any").lower().strip() info = {"expected_output_type": expected, "result_type": type(result).__name__, "result_repr": repr(result)[:2000]} paths = [] type_hints = [] text_values = [] def visit(obj): type_hints.append(type(obj).__name__.lower()) if isinstance(obj, (str, Path)): value = str(obj) if value: paths.append(value) text_values.append(value) elif isinstance(obj, dict): for key, value in obj.items(): key_l = str(key).lower() if key_l in {"path", "name", "url", "file", "filepath"}: visit(value) elif key_l in {"mime_type", "mime", "type", "format"} and value: type_hints.append(str(value).lower()) elif key_l in {"image", "video", "audio", "text"}: type_hints.append(key_l) visit(value) elif isinstance(value, (dict, list, tuple)): visit(value) elif isinstance(obj, (list, tuple)): for item in obj: visit(item) else: module = getattr(type(obj), "__module__", "") if module: type_hints.append(module.lower()) visit(result) lower_paths = [str(p).lower() for p in paths] hints = " ".join(type_hints + lower_paths) info["detected_paths"] = paths[:20] info["type_hints"] = type_hints[:30] if expected == "any": return result is not None, info image_ext = (".png", ".jpg", ".jpeg", ".webp", ".gif", ".bmp", ".tiff") video_ext = (".mp4", ".mov", ".webm", ".avi", ".mkv") audio_ext = (".wav", ".mp3", ".flac", ".ogg", ".m4a") if expected == "text": if isinstance(result, str) and bool(result.strip()): return True, info return any(isinstance(x, str) and x.strip() and Path(x).suffix.lower() not in image_ext + video_ext + audio_ext for x in text_values), info if expected == "image": return any(p.endswith(image_ext) for p in lower_paths) or any(h in hints for h in ["image", "pil.", "png", "jpeg", "jpg", "webp", "gradio.data_classes.filedata"]), info if expected == "video": return any(p.endswith(video_ext) for p in lower_paths) or any(h in hints for h in ["video", "mp4", "webm", "moviepy"]), info if expected == "audio": return any(p.endswith(audio_ext) for p in lower_paths) or any(h in hints for h in ["audio", "wav", "soundfile"]), 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 smoke_generate(target_space_id: str, token: str, run_dir: Path, events_path: Path): 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", {}) if not isinstance(test_args, list): raise ValueError("TEST_ARGS_JSON must be a JSON list") if not isinstance(test_kwargs, dict): raise ValueError("TEST_KWARGS_JSON must be a JSON object") append_event(events_path, "generation_smoke", "started", "Calling live generation endpoint", {"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) safe_kwargs, dropped_kwargs, endpoint_parameters = sanitize_kwargs_for_schema(api_name, schema, test_kwargs) test_args, autofill_meta = build_args_from_gradio_schema(api_name, schema, test_args) if autofill_meta.get("endpoint_parameters"): endpoint_parameters = autofill_meta.get("endpoint_parameters") validation_payload = {"api_name": api_name, "test_args": test_args, "test_kwargs": safe_kwargs, "effective_args": test_args, "effective_kwargs": safe_kwargs, **autofill_meta} write_json(run_dir / "tests" / "validation_payload.json", validation_payload) write_json(run_dir / "tests" / "api_schema.json", {"schema": schema, "api_names": discovered, "selected_api_name": api_name, "endpoint_parameters": endpoint_parameters}) if autofill_meta.get("args_were_autofilled"): append_event(events_path, "payload", "success", "Filled validation arguments from discovered Gradio schema", {"api_name": api_name, "args_source": autofill_meta.get("args_source"), "endpoint_parameters": endpoint_parameters, "effective_args": test_args}) 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() 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: failure_payload = {"status": "failed", "target_space": target_space_id, "api_name": api_name, "discovered_api_names": discovered, "test_args": test_args, "test_kwargs": safe_kwargs, "effective_args": test_args, "effective_kwargs": safe_kwargs, "args_source": autofill_meta.get("args_source"), "args_were_autofilled": bool(autofill_meta.get("args_were_autofilled")), "ignored_test_kwargs": dropped_kwargs, "endpoint_parameters": endpoint_parameters, "expected_output_type": expected_output_type, "error": message[:2000], "validated_at": now()} write_json(run_dir / "tests" / "generation_smoke.json", failure_payload) write_json(run_dir / "tests" / "test_result.json", failure_payload) append_event(events_path, "generation_smoke", "failed", "Live generation smoke test failed", failure_payload) raise latency = time.time() - started ok, info = result_contains_expected_output(result, expected_output_type) copied = copy_result_artifacts(result, run_dir) payload = { "status": "success" if ok else "failed", "target_space": target_space_id, "api_name": api_name, "discovered_api_names": discovered, "test_args": test_args, "test_kwargs": safe_kwargs, "effective_args": test_args, "effective_kwargs": safe_kwargs, "args_source": autofill_meta.get("args_source"), "args_were_autofilled": bool(autofill_meta.get("args_were_autofilled")), "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) write_json(run_dir / "tests" / "test_result.json", payload) if ok: append_event(events_path, "generation_smoke", "success", "Live generation smoke test passed", {"latency_seconds": payload["latency_seconds"], "recommended_zero_gpu_duration_seconds": payload["recommended_zero_gpu_duration_seconds"], "recommended_zerogpu_duration_seconds": payload["recommended_zero_gpu_duration_seconds"], "copied_artifacts": copied[:5]}) return payload append_event(events_path, "generation_smoke", "failed", "Live generation returned an unexpected output type", payload) raise RuntimeError("Generation smoke test failed: unexpected output type") def main(): run_id = os.environ["RUN_ID"] username = os.environ.get("HF_USERNAME", "unknown") output_root = Path(os.environ.get("OUTPUT_ROOT", "/output")) target_space_id = os.environ["TARGET_SPACE_ID"].strip() 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" append_event(events_path, "bootstrap", "started", "Existing Space validation worker started", {"target_space_id": target_space_id}) write_json(state_path, {"run_id": run_id, "kind": "validate_existing_space", "status": "running", "target_space": target_space_id, "created_by": username, "updated_at": now()}) if not token: raise RuntimeError("HF_TOKEN is missing") if not TARGET_RE.match(target_space_id): raise ValueError("TARGET_SPACE_ID must look like owner/space-name") try: install_deps(events_path) from huggingface_hub import HfApi api = HfApi(token=token) whoami = api.whoami(token=token) append_event(events_path, "auth", "success", "Authenticated inside validation Job", {"whoami_name": whoami.get("name")}) live = wait_until_live(api, target_space_id, token, run_dir, events_path, timeout_s=int(os.environ.get("LIVE_TIMEOUT_SECONDS", "1800"))) smoke = smoke_generate(target_space_id, token, run_dir, events_path) final_state = { "run_id": run_id, "kind": "validate_existing_space", "status": "full_inference_success", "message": "Existing Space passed live health/schema validation and generation smoke test.", "target_space": target_space_id, "target_space_url": f"https://huggingface.co/spaces/{target_space_id}", "live_validation": live, "generation_smoke": smoke, "updated_at": now(), } write_json(state_path, final_state) report = f"""# Agentic Space Factory — Existing Space Validation Report Status: **full_inference_success** Target Space: [`{target_space_id}`](https://huggingface.co/spaces/{target_space_id}) ## Generation smoke test ```json {json.dumps(smoke, indent=2, ensure_ascii=False)} ``` ## Notes - This validation is intended for Spaces whose hardware was set manually after generation. - Latency is measured from the live Gradio endpoint call. - The recommended ZeroGPU duration is a rough estimate from this live run, not a guarantee. """ (run_dir / "report.md").write_text(report, encoding="utf-8") append_event(events_path, "report_write", "success", "Wrote report.md") try: publish_eval_record(run_dir, phase="final", events_path=events_path) append_event(events_path, "anonymous_eval", "success", "Published anonymized validation evaluation record locally for backend archive publishing", {"enabled": True, "publish_mode": "backend", "archive_publish_confirmed": False}) except Exception as eval_exc: append_event(events_path, "anonymous_eval", "warning", "Could not publish anonymized validation evaluation record", {"error": str(eval_exc)[:1000]}) append_event(events_path, "done", "full_inference_success", "Existing Space validation completed", {"latency_seconds": smoke.get("latency_seconds")}) except Exception as exc: collect_space_logs(target_space_id, token or "", run_dir, events_path) details = {"error": str(exc)[:4000]} write_json(state_path, {"run_id": run_id, "kind": "validate_existing_space", "status": "failed", "target_space": target_space_id, "details": details, "updated_at": now()}) (run_dir / "report.md").write_text(f"# Existing Space Validation Failed\n\n```json\n{json.dumps(details, indent=2, ensure_ascii=False)}\n```\n", encoding="utf-8") append_event(events_path, "failure", "failed", "Existing Space validation failed", details) try: publish_eval_record(run_dir, phase="failure", events_path=events_path) append_event(events_path, "anonymous_eval", "success", "Published anonymized failed validation evaluation record locally for backend archive publishing", {"enabled": True, "publish_mode": "backend", "archive_publish_confirmed": False}) except Exception as eval_exc: append_event(events_path, "anonymous_eval", "warning", "Could not publish anonymized failed validation evaluation record", {"error": str(eval_exc)[:1000]}) raise SystemExit(1) if __name__ == "__main__": main() ''' def universal_model_card_worker_script() -> str: """Return the universal model-card builder worker script source.""" return UNIVERSAL_MODEL_CARD_WORKER_SCRIPT def validate_existing_space_worker_script() -> str: """Return the existing-Space validation worker script source.""" return VALIDATE_EXISTING_SPACE_WORKER_SCRIPT def encoded_universal_model_card_worker_script() -> str: """Return the base64-encoded universal model-card builder worker script. Kept for backwards compatibility in tests/imports. New Jobs read the worker from WORKER_SCRIPT_PATH to avoid oversized env/argv payloads. """ return _encode(UNIVERSAL_MODEL_CARD_WORKER_SCRIPT) def encoded_validate_existing_space_worker_script() -> str: """Return the base64-encoded existing-Space validation worker script. Kept for backwards compatibility in tests/imports. New Jobs read the worker from WORKER_SCRIPT_PATH to avoid oversized env/argv payloads. """ return _encode(VALIDATE_EXISTING_SPACE_WORKER_SCRIPT) def python_decode_and_run_command() -> list[str]: """Small command list for `run_job`. The worker script is stored in the run bucket and exposed through the Job volume as WORKER_SCRIPT_PATH. This avoids putting the large worker source in argv/env, which can make HF Jobs fail before Python starts with `argument list too long`. WORKER_SCRIPT_B64 remains as a tiny compatibility fallback for older tests/manual launch paths. """ runner = textwrap.dedent( """ import base64, os, pathlib, subprocess, sys script_path = os.environ.get('WORKER_SCRIPT_PATH') if script_path: path = pathlib.Path(script_path) if not path.exists(): raise SystemExit(f'Worker script not found at WORKER_SCRIPT_PATH={script_path}') else: script_b64 = os.environ.get('WORKER_SCRIPT_B64') if not script_b64: raise SystemExit('Missing WORKER_SCRIPT_PATH') path = pathlib.Path('/tmp/space_factory_worker.py') path.write_text(base64.b64decode(script_b64).decode('utf-8'), encoding='utf-8') raise SystemExit(subprocess.call([sys.executable, str(path)])) """ ).strip() return ["python", "-c", runner]