Upload 10 files

src/bucket.py

@@ -380,11 +380,16 @@ def _run_document_links(run_id: str, *, bucket_source: str, bundle: dict[str, An
         prefix = prefix.rstrip("/") + "/"
         return any(path.startswith(prefix) for path in file_paths)
 
-    raw_present = has_prefix("traces/raw") or _has_glob(f"{paths.root}/traces/raw/**/*", token=token) or _has_glob(f"{paths.root}/traces/raw/*", token=token)
-    redacted_present = has_prefix("traces/redacted") or _has_glob(f"{paths.root}/traces/redacted/**/*", token=token) or _has_glob(f"{paths.root}/traces/redacted/*", token=token)
+    raw_agent_trace_present = "traces/raw/agent_trace.jsonl" in file_paths or _path_exists(f"{paths.root}/traces/raw/agent_trace.jsonl", token=token)
+    redacted_agent_trace_present = "traces/redacted/agent_trace.jsonl" in file_paths or _path_exists(f"{paths.root}/traces/redacted/agent_trace.jsonl", token=token)
+    raw_present = raw_agent_trace_present or has_prefix("traces/raw") or _has_glob(f"{paths.root}/traces/raw/**/*", token=token) or _has_glob(f"{paths.root}/traces/raw/*", token=token)
+    redacted_present = redacted_agent_trace_present or has_prefix("traces/redacted") or _has_glob(f"{paths.root}/traces/redacted/**/*", token=token) or _has_glob(f"{paths.root}/traces/redacted/*", token=token)
     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)
     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"))
 
@@ -392,19 +397,19 @@ def _run_document_links(run_id: str, *, bucket_source: str, bundle: dict[str, An
         {
             "id": "pi_raw_trace",
             "label": "Pi RAW",
-            "subtitle": "Full internal trace",
+            "subtitle": "Unified build/diagnosis/repair trace",
             "icon": "🧾",
             "present": raw_present,
-            "url": _bucket_tree_url(bucket_source, run_id, "traces/raw"),
+            "url": _bucket_file_url(bucket_source, run_id, "traces/raw/agent_trace.jsonl") if raw_agent_trace_present else _bucket_tree_url(bucket_source, run_id, "traces/raw"),
             "sensitivity": "raw",
         },
         {
             "id": "pi_redacted_trace",
             "label": "Pi redacted",
-            "subtitle": "Safe to share",
+            "subtitle": "Unified safe agent trace",
             "icon": "🛡️",
             "present": redacted_present,
-            "url": _bucket_tree_url(bucket_source, run_id, "traces/redacted"),
+            "url": _bucket_file_url(bucket_source, run_id, "traces/redacted/agent_trace.jsonl") if redacted_agent_trace_present else _bucket_tree_url(bucket_source, run_id, "traces/redacted"),
             "sensitivity": "safe",
         },
         {
@@ -424,6 +429,42 @@ def _run_document_links(run_id: str, *, bucket_source: str, bundle: dict[str, An
             "url": _bucket_file_url(bucket_source, run_id, "tests/generation_smoke.json"),
         },
     ]
+    if repair_decision_present:
+        docs.append(
+            {
+                "id": "repair_decision",
+                "label": "Decision",
+                "subtitle": "Pi diagnosis action",
+                "icon": "◇",
+                "present": repair_decision_present,
+                "url": _bucket_file_url(bucket_source, run_id, "repair/REPAIR_DECISION.json"),
+                "tone": "warn" if "failed" in status or "error" in status else "neutral",
+            }
+        )
+    if repair_present:
+        docs.append(
+            {
+                "id": "repair",
+                "label": "Repair",
+                "subtitle": "Brief, plan and summary",
+                "icon": "✦",
+                "present": repair_present,
+                "url": _bucket_tree_url(bucket_source, run_id, "repair"),
+                "tone": "warn" if "failed" in status or "error" in status else "neutral",
+            }
+        )
+    if blockage_present:
+        docs.append(
+            {
+                "id": "blockage",
+                "label": "Blockage",
+                "subtitle": "Terminal recovery reason",
+                "icon": "!",
+                "present": blockage_present,
+                "url": _bucket_file_url(bucket_source, run_id, "repair/BLOCKAGE.json"),
+                "tone": "warn",
+            }
+        )
     if blockers_relevant:
         docs.append(
             {
@@ -576,6 +617,8 @@ def read_run_bundle(run_id: str, *, bucket_source: str, token: str | None = None
         "pi_model_resolution": _safe_read_json(f"{paths.root}/pi_model_resolution.json", token=token) or (read_json(paths.state, token=token) or {}).get("pi_model_resolution") or {},
         "space_runtime": _safe_read_json(f"{paths.root}/space_runtime.json", token=token),
         "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),
         "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/progress.py

@@ -15,13 +15,28 @@ STEP_ORDER = [
     "pi_install",
     "pi_config",
     "pi_run",
+    "pi_model_resolution",
     "pi_verification",
     "metadata_sanitize",
     "requirements_sanitize",
     "hardware_strategy",
     "create_space_hardware",
     "create_space",
+    "failure_detected",
+    "failure_diagnosis",
+    "pi_diagnosis",
+    "repair_decision",
+    "wait_for_logs",
+    "factory_rebuild",
     "repair",
+    "repair_diagnosis",
+    "repair_brief",
+    "repair_plan",
+    "repair_patch",
+    "repair_upload",
+    "repair_validation",
+    "technical_blocker",
+    "manual_hardware_required",
     "upload_files",
     "space_runtime",
     "space_logs",
@@ -44,13 +59,28 @@ STEP_LABELS = {
     "pi_install": "Pi install",
     "pi_config": "Pi config",
     "pi_run": "Pi run",
+    "pi_model_resolution": "Pi model",
     "pi_verification": "Pi verification",
     "metadata_sanitize": "Metadata sanitize",
     "requirements_sanitize": "Requirements sanitize",
     "hardware_strategy": "Hardware strategy",
     "create_space_hardware": "Create with hardware",
     "create_space": "Create Space",
-    "repair": "Repair pass",
+    "failure_detected": "Failure detected",
+    "failure_diagnosis": "Failure diagnosis",
+    "pi_diagnosis": "Pi diagnosis",
+    "repair_decision": "Repair decision",
+    "wait_for_logs": "Wait for logs",
+    "factory_rebuild": "Factory rebuild",
+    "repair": "Repair start",
+    "repair_diagnosis": "Patch allowed",
+    "repair_brief": "Repair brief",
+    "repair_plan": "Repair plan",
+    "repair_patch": "Repair patch",
+    "repair_upload": "Repair upload",
+    "repair_validation": "Repair validation",
+    "technical_blocker": "Technical blocker",
+    "manual_hardware_required": "Manual hardware",
     "upload_files": "Upload files",
     "space_runtime": "Space runtime",
     "space_logs": "Space logs",
@@ -70,9 +100,12 @@ STEP_ALIASES = {
     "hardware_fallback": "hardware_strategy",
     "hardware": "hardware_strategy",
     "build": "space_runtime",
-    "diagnose": "repair",
-    "diagnosis": "repair",
-    "patch": "repair",
+    "diagnose": "failure_diagnosis",
+    "diagnosis": "failure_diagnosis",
+    "pi_decision": "repair_decision",
+    "decision": "repair_decision",
+    "patch": "repair_patch",
+    "repair_failed": "repair_validation",
 }
 
 DONE_STATUSES = {
@@ -196,18 +229,21 @@ def progress_from_events(events: list[dict[str, Any]] | None, *, state: dict[str
         terminal_status = "cancelled"
     elif status_from_state in DONE_STATUSES:
         terminal_status = status_from_state
-    elif status_from_state in FAILED_STATUSES and (not events or current_status in FAILED_STATUSES or current_step == "failure"):
+    elif status_from_state in FAILED_STATUSES:
         terminal_status = "failed"
+        if current_status not in FAILED_STATUSES and current_step != "failure":
+            current_step = "failure"
+            current_status = "failed"
+            furthest_index = STEP_ORDER.index("failure")
     elif job_stage in CANCELLED_STATUSES:
         terminal_status = "cancelled"
     elif job_stage in FAILED_STATUSES:
-        # HF Job failed is authoritative only when the worker did not emit a
-        # later waiting/running validation/manual state. This avoids showing a
-        # recovered run as failed, but still turns genuinely stopped jobs red.
-        if not events or current_status in FAILED_STATUSES or current_step == "failure":
-            terminal_status = "failed"
-        elif current_status not in RUNNING_STATUSES:
-            terminal_status = "failed"
+        # A terminal HF Job failure must produce an explicit red timeline point.
+        terminal_status = "failed"
+        if current_status not in FAILED_STATUSES and current_step != "failure":
+            current_step = "failure"
+            current_status = "failed"
+            furthest_index = STEP_ORDER.index("failure")
 
     if terminal_status:
         overall_status = terminal_status
@@ -230,6 +266,24 @@ def progress_from_events(events: list[dict[str, Any]] | None, *, state: dict[str
         for step in STEP_ORDER:
             if step != "failure":
                 step_status[step] = "done"
+    elif overall_status in FAILED_STATUSES:
+        failure_idx = STEP_ORDER.index("failure")
+        if current_step in STEP_ORDER and current_step != "failure":
+            # Preserve the failed worker step when it is known; otherwise expose
+            # the explicit Failure stage so the UI always has a red dot.
+            failed_idx = STEP_ORDER.index(current_step)
+        else:
+            failed_idx = failure_idx
+            current_step = "failure"
+        furthest_index = max(furthest_index, failed_idx)
+        for idx, step in enumerate(STEP_ORDER):
+            if idx < failed_idx:
+                step_status[step] = "done"
+            elif idx == failed_idx:
+                step_status[step] = "failed"
+            else:
+                step_status[step] = "pending"
+        step_status["failure"] = "failed"
     else:
         for idx, step in enumerate(STEP_ORDER):
             if idx < furthest_index:

src/worker_payload.py

@@ -60,6 +60,66 @@ def redact_text(text: str | None) -> str:
     return value
 
 
+
+
+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 {}
@@ -170,6 +230,7 @@ def run_cmd_streaming(
     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.
 
@@ -178,12 +239,15 @@ def run_cmd_streaming(
     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,
@@ -214,7 +278,10 @@ def run_cmd_streaming(
                         f.write(clean + "\n")
                 now_m = time.monotonic()
                 if events_path and now_m - last_event > 8:
-                    append_event(events_path, step, "running", "Pi is still working", {"output_tail": "\n".join(lines[-20:])[-2000:]})
+                    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)
@@ -235,7 +302,10 @@ def run_cmd_streaming(
     finally:
         if run_dir:
             sync_pi_traces(run_dir, emit_event=False)
-    return int(proc.returncode or 0), redact_text("\n".join(lines))
+    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):
@@ -383,7 +453,8 @@ def emit_pi_model_resolution(events_path: Path, resolution: dict):
 
 def collect_pi_traces(run_dir: Path, events_path: Path):
     count = sync_pi_traces(run_dir, emit_event=False)
-    append_event(events_path, "traces", "success", "Collected Pi traces", {"count": count})
+    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"})
     return count
 
 
@@ -1079,66 +1150,633 @@ def normalize_requirements_for_modern_hub(workspace: Path, events_path: Path):
         append_event(events_path, "requirements_sanitize", "success", "Normalized huggingface_hub requirement for modern dependency resolution", {"huggingface_hub": hub_line})
 
 
-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):
-    """Ask Pi for one minimal build/runtime repair pass based on collected logs."""
+
+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."
+    return {
+        "category": category,
+        "failure_phase": phase,
+        "logs_quality": quality,
+        "signals": useful_log_signals(text),
+        "recommendation": recommendation,
+    }
+
+
+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", "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)}
+```
+
+## 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 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 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"
-    before_dir = repair_dir / "before"
-    after_dir = repair_dir / "after"
-    if before_dir.exists():
-        shutil.rmtree(before_dir)
-    shutil.copytree(workspace, before_dir, ignore=shutil.ignore_patterns(".git", "node_modules", "__pycache__", "*.pyc"))
-    goal = f"""You are Pi repairing a Hugging Face Space generated by Agentic Space Factory.
+    repair_dir.mkdir(parents=True, exist_ok=True)
+    classification = classify_repair_failure(failure_reason, build_log, runtime_log)
+    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.
 
-MODEL_ID: {model_id}
-TARGET_SPACE_ID: {target_space_id}
+First read `INCIDENT_BRIEF.md` and the HF Spaces gist operational rules: {GIST_URL}
+
+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.
 
-The first build/runtime validation failed.
+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)}
+```
 
-Failure summary:
-{failure_reason[:4000]}
+## 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}`
 
-Build log tail:
+## 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', '')}
+
+## Observed failure
+```text
+{failure_reason[:5000]}
+```
+
+## Build log tail
 ```text
 {build_log[-12000:]}
 ```
 
-Runtime log tail:
+## Runtime log tail
 ```text
 {runtime_log[-12000:]}
 ```
 
-Repair contract:
-- Make the smallest patch possible.
-- Prefer fixing dependency resolver conflicts, missing imports, invalid metadata, Gradio endpoint bugs, and import-order issues.
-- Do not replace real inference with a placeholder unless TECHNICAL_BLOCKERS.json clearly explains why full inference is impossible.
-- Preserve a cheap health endpoint with api_name="health".
-- Keep README frontmatter valid, short_description <= 60 chars.
-- Do not pin huggingface_hub below 1.0. For modern generated Spaces use huggingface_hub>=0.34.0,<2.0.0. If transformers>=5 is present, use huggingface_hub>=1.5.0,<2.0.0.
-- Do not delete the app. Do not publish anything. Work only in the current workspace.
+## Workspace files
+```text
+{chr(10).join(inventory)}
+```
 
-Deliverables:
-- patched app.py / requirements.txt / README.md as needed
-- REPAIR_SUMMARY.md explaining the patch
-- keep or update INFERENCE_CONTRACT.json if the inference contract changed
+## 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()
+        suspicious = ["placeholder", "dummy image", "fake inference", "todo: replace", "not implemented", "canned response"]
+        if any(marker in app_text for marker in suspicious):
+            return False, "Strict inference repair appears to introduce placeholder/fake inference markers."
+    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`, and the HF Spaces gist operational rules: {GIST_URL}
+
+This is not a new build. This patch is allowed only because the diagnosis decision selected `patch_code`.
+
+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")
-    append_event(events_path, "repair", "started", "Running Pi repair pass using build/runtime logs", {"model": pi_model})
+    (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", "failed", "Pi repair returned a non-zero exit code", {"returncode": code, "output_tail": out[-3000:]})
+        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", "Pi repair pass completed", {"output_tail": out[-3000:]})
+    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)
+    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})
+            upload_workspace(api, workspace, target_space_id, token, run_dir, events_path)
+            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)
@@ -1155,7 +1793,7 @@ def upload_workspace(api, workspace: Path, target_space_id: str, token: str, run
         repo_id=target_space_id,
         repo_type="space",
         token=token,
-        ignore_patterns=[".git/*", "node_modules/*", "__pycache__/*", "*.pyc", "GOAL.md"],
+        ignore_patterns=[".git/*", "node_modules/*", "__pycache__/*", "*.pyc", "GOAL.md", "REPAIR_GOAL.md", "REPAIR_BRIEF.md", "REPAIR_PLAN.md", "REPAIR_SUMMARY.md"],
     )
     uploaded_files = sorted(str(p.relative_to(workspace)) for p in workspace.rglob("*") if p.is_file() and "node_modules" not in p.parts and "__pycache__" not in p.parts)
     append_event(events_path, "upload_files", "success", "Uploaded generated workspace folder", {"file_count": len(uploaded_files), "files_sample": uploaded_files[:50]})
@@ -1323,6 +1961,7 @@ def main():
             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:
@@ -1373,13 +2012,12 @@ def main():
         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, "repair", "started", "Initial live validation failed; attempting one repair pass", {"error": str(validation_error)[:2000]})
-            collect_space_logs(target_space_id, token, run_dir, events_path)
-            repaired = repair_workspace_with_pi(workspace, run_dir, events_path, pi_model, target_space_id, model_id, str(validation_error))
-            if not repaired:
-                raise
-            upload_workspace(api, workspace, target_space_id, token, run_dir, events_path)
-            validation = validate_live_api(api, target_space_id, token, run_dir, events_path, timeout_s=1200)
+            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: