Upload 10 files

src/bucket.py

@@ -162,10 +162,125 @@ def append_run_event(run_id: str, *, bucket_source: str, step: str, status: str,
     return event
 
 
-def delete_run_folder(run_id: str, *, bucket_source: str, token: str | None = None) -> None:
-    """Delete a run folder from the private bucket."""
+def _bucket_path_kind(entry: Any) -> tuple[str, str | None]:
+    """Return ``(path, type)`` for HfFileSystem listing entries.
+
+    HfFileSystem may return plain strings or fsspec-style dictionaries, and
+    bucket directory objects are virtual.  Centralise the normalisation so run
+    deletion can remove concrete files without depending on a single listing
+    shape.
+    """
+    if isinstance(entry, dict):
+        return str(entry.get("name") or entry.get("path") or ""), entry.get("type")
+    return str(entry or ""), None
+
+
+def _collect_run_folder_files(fs: HfFileSystem, root: str) -> list[str]:
+    """Collect concrete bucket objects under a run prefix.
+
+    A single ``rm(root, recursive=True)`` can be unreliable on HF Storage
+    Buckets because the run folder is a virtual prefix.  Prefer explicit
+    recursive discovery, with several fallbacks for older/partial filesystem
+    behaviours.
+    """
+    root = root.rstrip("/")
+    found: set[str] = set()
+
+    # Best source when available: recursive find with fresh listing.
+    try:
+        entries = fs.find(root, withdirs=False, detail=False, refresh=True)
+        for entry in entries or []:
+            path, kind = _bucket_path_kind(entry)
+            if path and path != root and not path.endswith("/") and kind != "directory":
+                found.add(path)
+    except Exception:
+        pass
+
+    # Glob fallback catches buckets where find is incomplete or disabled.
+    for pattern in (f"{root}/**", f"{root}/**/*", f"{root}/*"):
+        try:
+            for entry in fs.glob(pattern) or []:
+                path, kind = _bucket_path_kind(entry)
+                if path and path != root and not path.endswith("/") and kind != "directory":
+                    found.add(path)
+        except Exception:
+            pass
+
+    # Manual recursive ls fallback.  Keep it iterative to avoid recursion loops
+    # if the remote returns the same virtual prefix multiple times.
+    pending = [root]
+    seen_dirs: set[str] = set()
+    while pending:
+        prefix = pending.pop().rstrip("/")
+        if prefix in seen_dirs:
+            continue
+        seen_dirs.add(prefix)
+        try:
+            entries = fs.ls(prefix, detail=True, refresh=True) or []
+        except Exception:
+            continue
+        for entry in entries:
+            path, kind = _bucket_path_kind(entry)
+            if not path or path == prefix:
+                continue
+            if kind == "directory" or path.endswith("/"):
+                pending.append(path.rstrip("/"))
+            else:
+                found.add(path)
+
+    return sorted(found, key=lambda value: value.count("/"), reverse=True)
+
+
+def delete_run_folder(run_id: str, *, bucket_source: str, token: str | None = None) -> dict[str, Any]:
+    """Delete every concrete object under a run prefix from the private bucket.
+
+    Returns a small report so the API and tests can distinguish a real bucket
+    cleanup from a UI-only tombstone.  The generated/tested Space is never
+    deleted here; only ``runs/<run_id>/`` bucket artifacts are removed.
+    """
     paths = RunPaths(run_id, bucket_source=bucket_source)
-    _fs(token).rm(paths.root, recursive=True)
+    root = paths.root.rstrip("/")
+    fs = _fs(token)
+    files = _collect_run_folder_files(fs, root)
+    deleted: list[str] = []
+    errors: list[str] = []
+
+    for path in files:
+        try:
+            fs.rm(path, recursive=False)
+            deleted.append(path)
+        except FileNotFoundError:
+            deleted.append(path)
+        except Exception as exc:  # noqa: BLE001
+            errors.append(f"{path}: {redact(str(exc))}")
+
+    # Keep the old recursive call as a final cleanup for any filesystem that
+    # does support virtual directory removal, but never rely on it as the only
+    # deletion mechanism.
+    try:
+        fs.rm(root, recursive=True)
+    except FileNotFoundError:
+        pass
+    except Exception as exc:  # noqa: BLE001
+        # If explicit file deletion succeeded, a virtual-directory cleanup error
+        # should not make the whole operation fail.  It is useful diagnostic
+        # metadata though, so report it when nothing was deleted.
+        if not deleted and not files:
+            errors.append(f"{root}: {redact(str(exc))}")
+
+    remaining = _collect_run_folder_files(fs, root)
+    if remaining:
+        errors.append(f"{len(remaining)} object(s) still present under runs/{run_id}")
+
+    if errors:
+        raise RuntimeError("; ".join(errors))
+
+    return {
+        "root": root,
+        "matched_count": len(files),
+        "deleted_count": len(deleted),
+        "remaining_count": len(remaining),
+    }
 
 
 def write_launch_metadata(run_id: str, *, bucket_source: str, payload: dict[str, Any], token: str | None = None) -> None:
@@ -359,8 +474,29 @@ def _bucket_tree_url(bucket_source: str, run_id: str, rel_path: str = "") -> str
 
 
 def _path_exists(path: str, token: str | None = None) -> bool:
+    """Best-effort existence check for Bucket objects.
+
+    HfFileSystem.exists() can lag or return false for recently written Bucket
+    objects depending on the backend view used by a running Job.  For UI affordances
+    such as the Run traces dock, a false negative is worse than a slightly slower
+    check, so fall back to parent listings and glob probes.
+    """
+    fs = _fs(token)
+    try:
+        if bool(fs.exists(path)):
+            return True
+    except Exception:
+        pass
+    try:
+        parent, name = path.rsplit("/", 1)
+        for entry in fs.ls(parent, detail=True):
+            entry_name = entry.get("name") if isinstance(entry, dict) else str(entry)
+            if str(entry_name).rstrip("/") == path.rstrip("/") or str(entry_name).rstrip("/").endswith("/" + name):
+                return True
+    except Exception:
+        pass
     try:
-        return bool(_fs(token).exists(path))
+        return bool(list(fs.glob(path))[:1])
     except Exception:
         return False
 
@@ -372,6 +508,25 @@ def _has_glob(pattern: str, token: str | None = None) -> bool:
         return False
 
 
+def _folder_has_content(path: str, token: str | None = None) -> bool:
+    """Return true when a Bucket folder/prefix contains at least one object."""
+    fs = _fs(token)
+    prefix = path.rstrip("/")
+    for probe in (f"{prefix}/agent_trace.jsonl", f"{prefix}/events.jsonl"):
+        if _path_exists(probe, token=token):
+            return True
+    try:
+        entries = fs.ls(prefix, detail=True)
+        if entries:
+            return True
+    except Exception:
+        pass
+    for pattern in (f"{prefix}/*", f"{prefix}/**/*"):
+        if _has_glob(pattern, token=token):
+            return True
+    return False
+
+
 def _run_document_links(run_id: str, *, bucket_source: str, bundle: dict[str, Any], token: str | None = None) -> list[dict[str, Any]]:
     """Return the small document dock shown under Active Run events.
 
@@ -391,31 +546,51 @@ def _run_document_links(run_id: str, *, bucket_source: str, bundle: dict[str, An
     paths = RunPaths(run_id, bucket_source=bucket_source)
     files = [f for f in (bundle.get("files") or []) if isinstance(f, dict)]
     file_paths = {str(f.get("path") or "") for f in files}
+    manifest = bundle.get("artifact_manifest") or {}
+    manifest_artifacts = [a for a in (manifest.get("artifacts") or []) if isinstance(a, dict)] if isinstance(manifest, dict) else []
+    manifest_paths = {str(a.get("path") or "") for a in manifest_artifacts if a.get("present")}
+
+    def manifest_has(path: str) -> bool:
+        path = path.strip("/")
+        prefix = path.rstrip("/") + "/"
+        return path in manifest_paths or any(p.startswith(prefix) for p in manifest_paths)
+
+    def first_existing_path(*candidates: str) -> str:
+        for candidate in candidates:
+            clean = candidate.strip("/")
+            if clean in file_paths or clean in manifest_paths:
+                return clean
+        if token:
+            for candidate in candidates:
+                clean = candidate.strip("/")
+                if _path_exists(f"{paths.root}/{clean}", token=token):
+                    return clean
+        return candidates[0].strip("/") if candidates else ""
 
     def trace_folder_has_content(prefix: str) -> bool:
         rel_prefix = prefix.rstrip("/") + "/"
         if any(path.startswith(rel_prefix) and not path.endswith("/") for path in file_paths):
             return True
-        root_prefix = f"{paths.root}/{rel_prefix}"
-        for pattern in (f"{root_prefix}*", f"{root_prefix}**/*"):
-            try:
-                found = [str(path) for path in _fs(token).glob(pattern)]
-            except Exception:
-                found = []
-            if any(item.replace(paths.root + "/", "", 1).startswith(rel_prefix) for item in found):
-                return True
-        return False
+        if manifest_has(prefix):
+            return True
+        if not token:
+            return False
+        root_prefix = f"{paths.root}/{prefix.rstrip('/')}"
+        # _folder_has_content probes direct children and recursive **/* patterns via _fs(token).glob(pattern).
+        return _folder_has_content(root_prefix, token=token)
 
     raw_present = trace_folder_has_content("traces/raw")
     redacted_present = trace_folder_has_content("traces/redacted")
     raw_trace_url = _bucket_tree_url(bucket_source, run_id, "traces/raw")
     redacted_trace_url = _bucket_tree_url(bucket_source, run_id, "traces/redacted")
-    smoke_present = bool(bundle.get("generation_smoke")) or "tests/generation_smoke.json" in file_paths or _path_exists(f"{paths.root}/tests/generation_smoke.json", token=token)
-    report_present = bool(bundle.get("report")) or _path_exists(paths.report, token=token)
-    blockers_present = bool(bundle.get("technical_blockers")) or "generated/TECHNICAL_BLOCKERS.json" in file_paths or _path_exists(f"{paths.root}/generated/TECHNICAL_BLOCKERS.json", token=token) or _path_exists(f"{paths.root}/TECHNICAL_BLOCKERS.json", token=token)
-    repair_decision_present = "repair/REPAIR_DECISION.json" in file_paths or _path_exists(f"{paths.root}/repair/REPAIR_DECISION.json", token=token)
-    blockage_present = "repair/BLOCKAGE.json" in file_paths or _path_exists(f"{paths.root}/repair/BLOCKAGE.json", token=token)
-    repair_present = any(path.startswith("repair/") for path in file_paths) or _path_exists(f"{paths.root}/repair/REPAIR_SUMMARY.md", token=token)
+    smoke_path = first_existing_path("tests/generation_smoke.json", "generation_smoke.json")
+    blockers_path = first_existing_path("generated/TECHNICAL_BLOCKERS.json", "TECHNICAL_BLOCKERS.json")
+    smoke_present = bool(bundle.get("generation_smoke")) or manifest_has("tests/generation_smoke.json") or manifest_has("generation_smoke.json") or "tests/generation_smoke.json" in file_paths or "generation_smoke.json" in file_paths or smoke_path in file_paths
+    report_present = bool(bundle.get("report")) or manifest_has("report.md") or "report.md" in file_paths or (bool(token) and _path_exists(paths.report, token=token))
+    blockers_present = bool(bundle.get("technical_blockers")) or manifest_has("generated/TECHNICAL_BLOCKERS.json") or manifest_has("TECHNICAL_BLOCKERS.json") or "generated/TECHNICAL_BLOCKERS.json" in file_paths or "TECHNICAL_BLOCKERS.json" in file_paths or blockers_path in file_paths
+    repair_decision_present = manifest_has("repair/REPAIR_DECISION.json") or "repair/REPAIR_DECISION.json" in file_paths or (bool(token) and _path_exists(f"{paths.root}/repair/REPAIR_DECISION.json", token=token))
+    blockage_present = manifest_has("repair/BLOCKAGE.json") or "repair/BLOCKAGE.json" in file_paths or (bool(token) and _path_exists(f"{paths.root}/repair/BLOCKAGE.json", token=token))
+    repair_present = manifest_has("repair") or any(path.startswith("repair/") for path in file_paths) or (bool(token) and _path_exists(f"{paths.root}/repair/REPAIR_SUMMARY.md", token=token))
     status = str((bundle.get("summary") or {}).get("status") or (bundle.get("state") or {}).get("status") or "").lower()
     blockers_relevant = blockers_present or any(marker in status for marker in ("manual", "blocker", "failed", "error"))
 
@@ -452,7 +627,7 @@ def _run_document_links(run_id: str, *, bucket_source: str, bundle: dict[str, An
             "subtitle": "Validation + latency",
             "icon": "⚡",
             "present": smoke_present,
-            "url": _bucket_file_url(bucket_source, run_id, "tests/generation_smoke.json"),
+            "url": _bucket_file_url(bucket_source, run_id, smoke_path),
         },
     ]
     if repair_decision_present:
@@ -499,7 +674,7 @@ def _run_document_links(run_id: str, *, bucket_source: str, bundle: dict[str, An
                 "subtitle": "Why inference is blocked",
                 "icon": "⚠️",
                 "present": blockers_present,
-                "url": _bucket_file_url(bucket_source, run_id, "generated/TECHNICAL_BLOCKERS.json"),
+                "url": _bucket_file_url(bucket_source, run_id, blockers_path),
                 "tone": "warn",
             }
         )
@@ -507,7 +682,10 @@ def _run_document_links(run_id: str, *, bucket_source: str, bundle: dict[str, An
 
 
 def _normalize_model_name(value: str | None) -> str:
-    return re.sub(r"[^a-z0-9]+", "", (value or "").lower())
+    raw = (value or "").strip().lower()
+    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]:
@@ -516,11 +694,14 @@ def _extract_pi_models_from_text(text: str) -> list[str]:
     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"\bMoonshotAI/[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"\bDeepSeek[-_/A-Za-z0-9.]+",
         r"\b[Mm]odel(?:\\s+used|\\s+selected|\\s*:)?\\s*[:=]?\\s*([A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+)",
     ]
     found: list[str] = []
@@ -603,6 +784,7 @@ def summarize_run_bundle(run_id: str, bundle: dict[str, Any], *, bucket_source:
         "kind": state.get("kind") or launch.get("kind") or summary_file.get("kind") or "unknown",
         "status": status,
         "model_id": state.get("model_id") or launch.get("model_id") or summary_file.get("model_id") or state.get("model") or "",
+        "pi_model": state.get("pi_model") or launch.get("pi_model") or summary_file.get("pi_model") or (bundle.get("pi_model_resolution") or {}).get("requested_model") or "",
         "target_space": target_space,
         "target_space_url": state.get("target_space_url") or launch.get("target_space_url") or summary_file.get("target_space_url") or (f"https://huggingface.co/spaces/{target_space}" if target_space else ""),
         "job_id": state.get("job_id") or launch.get("job_id") or summary_file.get("job_id") or "",
@@ -649,6 +831,7 @@ def read_run_bundle(run_id: str, *, bucket_source: str, token: str | None = None
         "api_schema": _safe_read_json(f"{paths.root}/tests/api_schema.json", token=token) or _safe_read_json(f"{paths.root}/tests/generation_api_schema.json", token=token),
         "repair_decision": _safe_read_json(f"{paths.root}/repair/REPAIR_DECISION.json", token=token),
         "blockage": _safe_read_json(f"{paths.root}/repair/BLOCKAGE.json", token=token),
+        "artifact_manifest": _safe_read_json(f"{paths.root}/artifact_manifest.json", token=token),
         "files": _list_run_files(run_id, bucket_source=bucket_source, token=token) if include_heavy else [],
     }
     bundle["run_documents"] = _run_document_links(run_id, bucket_source=bucket_source, bundle=bundle, token=token)

src/worker_payload.py

@@ -88,6 +88,89 @@ def append_event(path: Path, step: str, status: str, message: str, data: dict |
     print(line, flush=True)
 
 
+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, "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 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 ""
@@ -249,25 +332,38 @@ def exception_payload(exc: Exception) -> dict:
 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)
-    write_json(run_dir / "state.json", {
+    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": "universal_model_card_builder",
+        "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")
     raise SystemExit(1)
 
 
@@ -437,7 +533,13 @@ def configure_pi(events_path: Path, model: str):
 
 
 def normalize_pi_model_name(value: str | None) -> str:
-    return re.sub(r"[^a-z0-9]+", "", (value or "").lower())
+    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]:
@@ -447,11 +549,14 @@ def extract_pi_models_from_text(text: str) -> list[str]:
     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"\bMoonshotAI/[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"\bDeepSeek[-_/A-Za-z0-9.]+",
         r"\b[Mm]odel(?:\\s+used|\\s+selected|\\s*:)?\\s*[:=]?\\s*([A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+)",
     ]
     found: list[str] = []
@@ -539,6 +644,7 @@ 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
 
 
@@ -943,6 +1049,10 @@ def safe_same_code_reupload(api, workspace: Path, target_space_id: str, token: s
 
 
 def collect_space_logs(target_space_id: str, token: str, run_dir: Path, events_path: Path):
+    # Import locally because main() imports HfApi inside its own scope after
+    # installing dependencies. Recovery/log collection can run from helper
+    # functions where that local name is otherwise unavailable.
+    from huggingface_hub import HfApi
     logs_dir = run_dir / "logs"
     logs_dir.mkdir(parents=True, exist_ok=True)
     written = []
@@ -976,6 +1086,7 @@ def collect_space_logs(target_space_id: str, token: str, run_dir: Path, events_p
             written.append({"file": filename, "source": "HfApi.get_space_logs", "error": msg[-1000:]})
 
     append_event(events_path, "space_logs", "success", "Collected best-effort Space logs", {"files": written})
+    write_artifact_manifest(run_dir, reason="space_logs_collected")
     return written
 
 
@@ -1394,7 +1505,7 @@ sdk: gradio
 app_file: app.py
 python_version: "3.10"
 suggested_hardware: {preferred_hardware or fallback_hardware or "cpu-basic"}
-short_description: "Agent-built model demo"
+short_description: "Generated model demo"
 ---
 
 # Generated Model Space — Agentic Space Factory
@@ -1456,33 +1567,62 @@ Deliverables:
 
 
 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():
-        return
+        readme_path.write_text("# Generated Model Space\n", encoding="utf-8")
     text = readme_path.read_text(encoding="utf-8", errors="ignore")
-    if not text.startswith("---"):
-        return
-    parts = text.split("---", 2)
-    if len(parts) < 3:
-        return
-    _, frontmatter, body = parts
+    body = text
+    metadata = {}
     changed = False
-    sanitized_lines = []
-    for line in frontmatter.splitlines():
-        if line.strip().startswith("short_description:"):
-            value = "Generated model demo"
-            sanitized_lines.append(f"short_description: {value}")
-            changed = True
+    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:
-            sanitized_lines.append(line)
-    # If Pi added other unexpectedly long one-line metadata values, leave them alone:
-    # the known Hub validation blocker for this run was short_description > 60 chars.
-    if changed:
-        new_text = "---\n" + "\n".join(sanitized_lines).strip() + "\n---" + body
-        readme_path.write_text(new_text, encoding="utf-8")
-        append_event(events_path, "metadata_sanitize", "success", "Sanitized README metadata", {"short_description": "Generated model demo"})
-
+            body = text
+            changed = True
+    else:
+        changed = True
 
+    required = {
+        "title": metadata.get("title") or "Generated Model Space",
+        "sdk": "gradio",
+        "app_file": "app.py",
+        "python_version": metadata.get("python_version") or "3.10",
+    }
+    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"} 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]})
 
 def normalize_requirements_for_modern_hub(workspace: Path, events_path: Path):
     """Normalize only broad known-dangerous base dependencies before upload.
@@ -2576,7 +2716,7 @@ def main():
     pi_model_resolution = {"requested_model": pi_model, "configured_model": pi_model, "effective_model": pi_model, "provider": "huggingface", "mismatch": False}
 
     append_event(events_path, "bootstrap", "started", "Universal model-card builder worker started", {"model_id": model_id, "target_space_id": target_space_id})
-    write_json(state_path, {"run_id": run_id, "kind": "universal_model_card_builder", "status": "running", "message": "Attempting Universal model-card builderd Space creation", "model_id": model_id, "target_space": target_space_id, "created_by": hf_username, "bucket_source": bucket_source, "created_at": now(), "updated_at": now()})
+    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):
@@ -2649,12 +2789,29 @@ def main():
         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})
@@ -2716,6 +2873,7 @@ def main():
             "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,
@@ -2790,6 +2948,7 @@ The wrapper validated the live Space using HTTP `/health` first, with Gradio Cli
 """
         (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")
         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
@@ -3144,6 +3303,10 @@ def safe_same_code_reupload(api, workspace: Path, target_space_id: str, token: s
 
 
 def collect_space_logs(target_space_id: str, token: str, run_dir: Path, events_path: Path):
+    # Import locally because main() imports HfApi inside its own scope after
+    # installing dependencies. Recovery/log collection can run from helper
+    # functions where that local name is otherwise unavailable.
+    from huggingface_hub import HfApi
     logs_dir = run_dir / "logs"
     logs_dir.mkdir(parents=True, exist_ok=True)
     written = []
@@ -3177,6 +3340,7 @@ def collect_space_logs(target_space_id: str, token: str, run_dir: Path, events_p
             written.append({"file": filename, "source": "HfApi.get_space_logs", "error": msg[-1000:]})
 
     append_event(events_path, "space_logs", "success", "Collected best-effort Space logs", {"files": written})
+    write_artifact_manifest(run_dir, reason="space_logs_collected")
     return written