Upload 15 files

src/bucket.py

@@ -670,14 +670,68 @@ def _run_document_links(run_id: str, *, bucket_source: str, bundle: dict[str, An
 
     redacted_present = trace_folder_has_content("traces/redacted")
     redacted_trace_url = _bucket_tree_url(bucket_source, run_id, "traces/redacted")
-    smoke_path = first_existing_path("tests/generation_smoke.json", "generation_smoke.json")
+    smoke_path = first_existing_path(
+        "tests/generation_smoke.json",
+        "tests/generation_smoke_latest.json",
+        "generation_smoke.json",
+        "tests/test_result.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
+    smoke_present = (
+        bool(bundle.get("generation_smoke"))
+        or manifest_has("tests/generation_smoke.json")
+        or manifest_has("tests/generation_smoke_latest.json")
+        or manifest_has("generation_smoke.json")
+        or "tests/generation_smoke.json" in file_paths
+        or "tests/generation_smoke_latest.json" in file_paths
+        or "generation_smoke.json" in file_paths
+        or smoke_path in file_paths
+        or smoke_path in manifest_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
     eval_publish_payload = bundle.get("eval_publish") if isinstance(bundle.get("eval_publish"), dict) else bundle.get("eval_publish_status") if isinstance(bundle.get("eval_publish_status"), dict) else {}
     eval_publish_present = bool(eval_publish_payload) or manifest_has("eval_publish_status.json") or "eval_publish_status.json" in file_paths or (bool(token) and _path_exists(f"{paths.root}/eval_publish_status.json", token=token))
+    useful_space_log_candidates = (
+        "logs/space_logs_index.json",
+        "logs/space_logs_build.txt",
+        "logs/space_logs_runtime.txt",
+        "logs/space_logs_run.txt",
+        "logs/space_logs_fetch_status.json",
+        "logs/space_log_diagnostics.json",
+        "logs/space_runtime_snapshot.json",
+        "logs/pi_live_output.txt",
+        "logs/pi_output.txt",
+        "logs/build_log_fast_probe_status.json",
+    )
+    space_logs_path = first_existing_path(*useful_space_log_candidates)
     space_logs_index_present = manifest_has("logs/space_logs_index.json") or "logs/space_logs_index.json" in file_paths or (bool(token) and _path_exists(f"{paths.root}/logs/space_logs_index.json", token=token))
+    space_logs_present = (
+        space_logs_index_present
+        or manifest_has("logs")
+        or any(path in file_paths or path in manifest_paths for path in useful_space_log_candidates)
+        or any(path.startswith("logs/") and not path.endswith("/") for path in file_paths | manifest_paths)
+        or (bool(token) and _folder_has_content(f"{paths.root}/logs", token=token))
+    )
+    if not space_logs_index_present and space_logs_present and space_logs_path == "logs/space_logs_index.json":
+        space_logs_path = "logs"
+    output_artifact_candidates = (
+        "artifacts/image.webp",
+        "artifacts/image.png",
+        "artifacts/image.jpg",
+        "artifacts/image.jpeg",
+        "artifacts/output.png",
+        "artifacts/output.webp",
+        "artifacts/output.json",
+    )
+    output_artifact_path = first_existing_path(*output_artifact_candidates)
+    output_artifact_present = (
+        any(path in file_paths or path in manifest_paths for path in output_artifact_candidates)
+        or manifest_has("artifacts")
+        or (bool(token) and _folder_has_content(f"{paths.root}/artifacts", token=token))
+    )
+    if output_artifact_present and output_artifact_path == "artifacts/image.webp" and output_artifact_path not in file_paths and output_artifact_path not in manifest_paths and not manifest_has(output_artifact_path):
+        output_artifact_path = "artifacts"
     build_error_present = manifest_has("build_error_observation.json") or "build_error_observation.json" in file_paths or (bool(token) and _path_exists(f"{paths.root}/build_error_observation.json", token=token))
     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))
@@ -711,6 +765,14 @@ def _run_document_links(run_id: str, *, bucket_source: str, bundle: dict[str, An
             "present": smoke_present,
             "url": _bucket_file_url(bucket_source, run_id, smoke_path),
         },
+        {
+            "id": "smoke_output",
+            "label": "Output",
+            "subtitle": "Generated artifact",
+            "icon": "🖼️",
+            "present": output_artifact_present,
+            "url": _bucket_file_url(bucket_source, run_id, output_artifact_path) if output_artifact_path != "artifacts" else _bucket_tree_url(bucket_source, run_id, "artifacts"),
+        },
         {
             "id": "eval_publish",
             "label": "Eval publish",
@@ -723,10 +785,10 @@ def _run_document_links(run_id: str, *, bucket_source: str, bundle: dict[str, An
         {
             "id": "space_logs",
             "label": "Space logs",
-            "subtitle": "Build/runtime log index",
+            "subtitle": "Build/runtime log index" if space_logs_index_present else "Build/runtime logs",
             "icon": "▣",
-            "present": space_logs_index_present,
-            "url": _bucket_file_url(bucket_source, run_id, "logs/space_logs_index.json"),
+            "present": space_logs_present,
+            "url": _bucket_file_url(bucket_source, run_id, space_logs_path) if space_logs_path != "logs" else _bucket_tree_url(bucket_source, run_id, "logs"),
             "tone": "warn" if "failed" in status or "error" in status else "neutral",
         },
         {
@@ -788,7 +850,7 @@ def _run_document_links(run_id: str, *, bucket_source: str, bundle: dict[str, An
             }
         )
     if "failed" in status or "error" in status:
-        priority = {"report": 0, "blockage": 1, "build_error": 2, "space_logs": 3, "repair_decision": 4, "repair": 5, "blockers": 6, "pi_redacted_trace": 7, "smoke": 8, "eval_publish": 10}
+        priority = {"report": 0, "blockage": 1, "build_error": 2, "space_logs": 3, "repair_decision": 4, "repair": 5, "blockers": 6, "pi_redacted_trace": 7, "smoke": 8, "smoke_output": 9, "eval_publish": 10}
         docs.sort(key=lambda item: priority.get(str(item.get("id") or ""), 50))
     return docs
 
@@ -1114,6 +1176,64 @@ def _has_persisted_generation_success(*sources: dict[str, Any] | None) -> bool:
                     return True
     return False
 
+
+def _has_authoritative_runtime_success(
+    *,
+    final_status_reconciliation: dict[str, Any] | None = None,
+    summary_file: dict[str, Any] | None = None,
+    state: dict[str, Any] | None = None,
+    live_status: dict[str, Any] | None = None,
+    gate: dict[str, Any] | None = None,
+    smoke: dict[str, Any] | None = None,
+) -> bool:
+    """Return true when a final runtime proof supersedes stale Pi/contract blockers.
+
+    v198.26.15: contract and Pi diagnostic signals are pre-runtime safety rails.
+    They must still protect diagnostic-only demos, but they must not downgrade a
+    run after the worker has published a final reconciliation/summary/state/live
+    proof of ``full_inference_success`` or a strong inference gate.
+    """
+    def explicit_full_status(source: dict[str, Any]) -> bool:
+        for key in ("status", "final_status", "ui_status", "display_status", "effective_status", "effective_verdict", "verdict"):
+            if str(source.get(key) or "").strip().lower() == "full_inference_success":
+                return True
+        return False
+
+    for source in (final_status_reconciliation or {}, summary_file or {}, state or {}):
+        if not isinstance(source, dict):
+            continue
+        if explicit_full_status(source):
+            return True
+        if _truthy(source.get("generation_smoke_passed")) and _truthy(source.get("promise_fulfilled")):
+            return True
+
+    live = live_status or {}
+    if isinstance(live, dict):
+        live_status_text = _classify_build_status_text(_source_status_text(live))
+        live_stage = str(live.get("stage") or "").strip().lower()
+        if live_stage == "done" and live_status_text == "full_inference_success":
+            return True
+        if live_stage == "done" and _truthy(live.get("generation_smoke_passed")):
+            return True
+
+    gate_src = gate or {}
+    if isinstance(gate_src, dict):
+        if _truthy(gate_src.get("strong_full_inference_success")):
+            return True
+        if _classify_build_status_text(_source_status_text(gate_src)) == "full_inference_success" and _truthy(_nested_get(gate_src, "implementation_signals", "generation_smoke_passed")):
+            return True
+
+    smoke_src = smoke or {}
+    if isinstance(smoke_src, dict):
+        # A lone smoke success is not enough: v198.18 diagnostic-only demos can
+        # produce metrics.  Only treat smoke as authoritative when paired with a
+        # known generation endpoint and an explicit fulfilled promise marker.
+        endpoint = smoke_src.get("api_name") or smoke_src.get("endpoint") or smoke_src.get("primary_api_name")
+        if str(smoke_src.get("status") or "").strip().lower() == "success" and endpoint and _truthy(smoke_src.get("promise_fulfilled")):
+            return True
+
+    return False
+
 def _canonical_build_status_from_sources(
     *,
     state: dict[str, Any] | None = None,
@@ -1123,6 +1243,7 @@ def _canonical_build_status_from_sources(
     eval_record: dict[str, Any] | None = None,
     smoke: dict[str, Any] | None = None,
     gate: dict[str, Any] | None = None,
+    live_status: dict[str, Any] | None = None,
     repair: dict[str, Any] | None = None,
     manual_validation: dict[str, Any] | None = None,
     eval_publish_status: dict[str, Any] | None = None,
@@ -1138,7 +1259,15 @@ def _canonical_build_status_from_sources(
     status so Run Explorer and Run Stats match the Active Run snapshot after a
     page load. Intermediate incident events remain phase diagnostics only.
     """
-    contract_blocks_success = _contract_blocks_success(summary_file, state, gate, inference_contract, demo_quality_contract)
+    runtime_success_authoritative = _has_authoritative_runtime_success(
+        final_status_reconciliation=final_status_reconciliation,
+        summary_file=summary_file,
+        state=state,
+        live_status=live_status,
+        gate=gate,
+        smoke=smoke,
+    )
+    contract_blocks_success = False if runtime_success_authoritative else _contract_blocks_success(summary_file, state, gate, inference_contract, demo_quality_contract)
     # v198.26.1: eval_publish_status is an archive annotation, not a primary
     # run verdict source. It must not make a terminal build appear running/success
     # or override final reconciliation after a simple UI read.
@@ -1365,6 +1494,7 @@ def summarize_run_bundle(run_id: str, bundle: dict[str, Any], *, bucket_source:
             eval_record=eval_record,
             smoke=smoke or bundle.get("test_result") or {},
             gate=gate,
+            live_status=bundle.get("live_status") or {},
             repair=bundle.get("repair_outcome") or {},
             manual_validation=manual_validation,
             eval_publish_status=bundle.get("eval_publish_status") or {},

src/effective_status.py

@@ -68,6 +68,8 @@ def _target_space_from_bundle(bundle: dict[str, Any]) -> str:
 
 
 def _contract_declares_no_full_inference(bundle: dict[str, Any]) -> bool:
+    if _has_authoritative_runtime_success(bundle):
+        return False
     contract = bundle.get("inference_contract") or {}
     quality = bundle.get("demo_quality_contract") or {}
     gate = bundle.get("inference_gate") or {}
@@ -94,6 +96,39 @@ def _contract_declares_no_full_inference(bundle: dict[str, Any]) -> bool:
     return False
 
 
+def _has_authoritative_runtime_success(bundle: dict[str, Any]) -> bool:
+    """Return true when final runtime proof invalidates stale contract blockers."""
+    def explicit_full_status(source: dict[str, Any]) -> bool:
+        for key in ("status", "final_status", "ui_status", "display_status", "effective_status", "effective_verdict", "verdict"):
+            if _lower(source.get(key)) == "full_inference_success":
+                return True
+        return False
+
+    for source in (
+        bundle.get("final_status_reconciliation") or {},
+        bundle.get("summary") or bundle.get("summary_file") or {},
+        bundle.get("state") or {},
+    ):
+        if not isinstance(source, dict):
+            continue
+        if explicit_full_status(source):
+            return True
+        if source.get("generation_smoke_passed") is True and source.get("promise_fulfilled") is True:
+            return True
+
+    live = bundle.get("live_status") or {}
+    if isinstance(live, dict):
+        if _lower(live.get("stage")) == "done" and _normalize_build_status(live.get("status")) == "full_inference_success":
+            return True
+        if _lower(live.get("stage")) == "done" and live.get("generation_smoke_passed") is True:
+            return True
+
+    gate = bundle.get("inference_gate") or {}
+    if isinstance(gate, dict) and gate.get("strong_full_inference_success") is True:
+        return True
+    return False
+
+
 def _linked_rows(bundle: dict[str, Any]) -> list[dict[str, Any]]:
     linked = bundle.get("linked_validations") or {}
     rows = linked.get("validations") if isinstance(linked, dict) else linked
@@ -178,6 +213,8 @@ def compute_effective_run_status(
         launch.get("status"),
         "unknown",
     )
+    if _has_authoritative_runtime_success(bundle):
+        raw_build_status = "full_inference_success"
     normalized_build = _normalize_build_status(raw_build_status)
     normalized_verdict = _normalize_build_status(build_verdict or normalized_build)
     post = _successful_post_build_validation(bundle)

src/timeline_model.py

@@ -759,6 +759,8 @@ def _first_terminal_verdict_from_sources(bundle: dict[str, Any]) -> str:
         bundle.get("effective_run_status") or {},
         (bundle.get("summary") or {}).get("effective_run_status") if isinstance((bundle.get("summary") or {}).get("effective_run_status"), dict) else {},
         bundle.get("eval_record") or {},
+        bundle.get("state") or {},
+        bundle.get("live_status") or {},
     ):
         if isinstance(src, dict):
             sources.append(src)
@@ -778,8 +780,9 @@ def _verdict(bundle: dict[str, Any]) -> str:
         return terminal
     gate = _gate(bundle)
     state = bundle.get("state") or {}
+    live = bundle.get("live_status") or {}
     repair = bundle.get("repair_outcome") or {}
-    for value in (gate.get("status"), state.get("gate_status"), state.get("status"), repair.get("post_repair_validation"), repair.get("failure_type")):
+    for value in (state.get("status"), live.get("status") if _lower(live.get("stage")) == "done" else "", gate.get("status"), state.get("gate_status"), repair.get("post_repair_validation"), repair.get("failure_type")):
         normalized = _normalize_verdict_token(value)
         if normalized:
             return normalized

src/version.py

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

src/view_models.py

@@ -182,12 +182,15 @@ def _first_authoritative_terminal_status(bundle: dict[str, Any]) -> tuple[str, s
     summary = bundle.get("summary") or {}
     if isinstance(summary.get("effective_run_status"), dict):
         ordered_sources.append(("summary.effective_run_status", summary.get("effective_run_status") or {}))
-    # Gate/smoke are lower priority: they can confirm terminal success/partial
-    # only when no final/summary verdict has arrived yet.
+    # State/live_status are runtime artifacts written after the worker has seen
+    # the generated Space.  They are stronger than stale Pi/contract/gate
+    # diagnostic blockers, but still lower priority than final reconciliation and
+    # summary.
     ordered_sources.extend([
+        ("state", bundle.get("state") or {}),
+        ("live_status", bundle.get("live_status") or {}),
         ("inference_gate", bundle.get("inference_gate") or {}),
         ("generation_smoke", bundle.get("generation_smoke") or {}),
-        ("live_status", bundle.get("live_status") or {}),
     ])
     blocking_seen = False
     for source_name, source in ordered_sources:
@@ -214,6 +217,22 @@ def _first_authoritative_terminal_status(bundle: dict[str, Any]) -> tuple[str, s
     return "", ""
 
 
+def _has_authoritative_runtime_success(bundle: dict[str, Any]) -> bool:
+    status, _source = _first_authoritative_terminal_status(bundle)
+    if status == "full_inference_success":
+        return True
+    live = bundle.get("live_status") or {}
+    if isinstance(live, dict):
+        if _lower(live.get("stage")) == "done" and _normalize_terminal_status(live.get("status")) == "full_inference_success":
+            return True
+        if _lower(live.get("stage")) == "done" and live.get("generation_smoke_passed") is True:
+            return True
+    gate = bundle.get("inference_gate") or {}
+    if isinstance(gate, dict) and gate.get("strong_full_inference_success") is True:
+        return True
+    return False
+
+
 def _status_model_from_terminal_status(status: str, *, source: str = "") -> dict[str, str | bool]:
     normalized = _normalize_terminal_status(status) or "unknown"
     if normalized == "full_inference_success":
@@ -251,6 +270,8 @@ def requires_manual_action(bundle: dict[str, Any]) -> bool:
 
 
 def has_technical_blocker(bundle: dict[str, Any]) -> bool:
+    if _has_authoritative_runtime_success(bundle):
+        return False
     blockers = bundle.get("technical_blockers") or {}
     blocker_items = blockers.get("blockers") if isinstance(blockers, dict) else None
     return bool(blocker_items or _raw_statuses(bundle).intersection(BLOCKED_RAW_STATUSES))
@@ -733,26 +754,104 @@ def build_validation_metrics(bundle: dict[str, Any]) -> dict[str, Any]:
     }
     return metrics
 
+def _normalize_gradio_api_name(value: Any) -> str:
+    text = str(value or "").strip()
+    if not text:
+        return ""
+    if text.lower() in {"none", "null", "false"}:
+        return ""
+    if text.startswith("/"):
+        return text
+    return f"/{text}"
+
+
+def _automatic_smoke_passed(bundle: dict[str, Any]) -> bool:
+    smoke = bundle.get("generation_smoke") or {}
+    state = bundle.get("state") or {}
+    summary = bundle.get("summary") or {}
+    summary_file = bundle.get("summary_file") or {}
+    gate = bundle.get("inference_gate") or {}
+    gate_smoke = gate.get("generation_smoke") if isinstance(gate.get("generation_smoke"), dict) else {}
+    final = bundle.get("final_status_reconciliation") or {}
+    test_result = bundle.get("test_result") or {}
+    sources = (smoke, gate_smoke, final, summary, summary_file, state, test_result)
+    for source in sources:
+        if not isinstance(source, dict):
+            continue
+        if source.get("ok") is True:
+            return True
+        if _lower(source.get("status")) == "success" and (
+            source is smoke
+            or source is gate_smoke
+            or source.get("latency_seconds") is not None
+            or source.get("output_artifact")
+            or source.get("artifact_url")
+        ):
+            return True
+        if source.get("generation_smoke_passed") is True or source.get("smoke_test_passed") is True:
+            return True
+    signals = gate.get("implementation_signals") if isinstance(gate.get("implementation_signals"), dict) else {}
+    return bool(signals.get("generation_smoke_passed") is True or gate.get("strong_full_inference_success") is True)
+
+
+def _smoke_payload_available(bundle: dict[str, Any]) -> bool:
+    for key in ("generation_smoke_payload", "generation_smoke_payload_retry", "resolved_validation_request", "validation_payload"):
+        payload = bundle.get(key)
+        if not isinstance(payload, dict):
+            continue
+        for payload_key in ("effective_args", "test_args", "args"):
+            if isinstance(payload.get(payload_key), list):
+                return True
+        for payload_key in ("effective_kwargs", "test_kwargs", "kwargs"):
+            if isinstance(payload.get(payload_key), dict):
+                return True
+        if payload.get("api_name") or payload.get("endpoint"):
+            return True
+    return False
+
+
 def _known_gradio_endpoint_info(bundle: dict[str, Any]) -> dict[str, Any]:
     smoke = bundle.get("generation_smoke") or {}
     gate = bundle.get("inference_gate") or {}
     live = bundle.get("live_status") or {}
     schema = bundle.get("api_schema") or bundle.get("gradio_schema") or {}
-    endpoints: list[Any] = []
-    for source in (smoke, gate, schema, live):
+    state = bundle.get("state") or {}
+    summary = bundle.get("summary") or {}
+    summary_file = bundle.get("summary_file") or {}
+    final = bundle.get("final_status_reconciliation") or {}
+    launch = bundle.get("launch") or {}
+    payload = bundle.get("generation_smoke_payload") or {}
+    payload_retry = bundle.get("generation_smoke_payload_retry") or {}
+    resolved = bundle.get("resolved_validation_request") or {}
+    promise = state.get("promise_validation") if isinstance(state.get("promise_validation"), dict) else {}
+    gate_smoke = gate.get("generation_smoke") if isinstance(gate.get("generation_smoke"), dict) else {}
+    state_smoke = state.get("generation_smoke") if isinstance(state.get("generation_smoke"), dict) else {}
+    summary_smoke = summary.get("generation_smoke") if isinstance(summary.get("generation_smoke"), dict) else {}
+
+    endpoints: list[str] = []
+    for source in (smoke, gate_smoke, state_smoke, summary_smoke, gate, schema, live, state, summary, summary_file, final, launch, payload, payload_retry, resolved, promise):
         if not isinstance(source, dict):
             continue
-        for key in ("discovered_api_names", "api_names", "endpoints", "api_endpoints", "named_endpoints"):
+        for key in ("discovered_api_names", "api_names", "endpoints", "api_endpoints", "named_endpoints", "available_api_names"):
             value = source.get(key)
             if isinstance(value, list):
-                endpoints.extend(value)
-    endpoint = _first_nonempty(smoke.get("api_name"), smoke.get("endpoint"), gate.get("api_name"), schema.get("selected_api_name"))
-    endpoint_count = len([e for e in endpoints if e])
-    endpoint_known = bool(endpoint and endpoint != "/generate") or endpoint_count > 0 or bool(smoke.get("status") == "success")
-    return {"endpoint": endpoint or "/generate", "endpoint_count": endpoint_count, "endpoint_known": endpoint_known, "endpoints": endpoints[:20]}
+                endpoints.extend(_normalize_gradio_api_name(item) for item in value if _normalize_gradio_api_name(item))
+        for key in ("api_name", "endpoint", "selected_api_name", "selected_endpoint", "primary_api_name", "api_endpoint", "generation_endpoint"):
+            endpoint_candidate = _normalize_gradio_api_name(source.get(key))
+            if endpoint_candidate:
+                endpoints.append(endpoint_candidate)
+    # Deduplicate while preserving discovery order.
+    endpoints = list(dict.fromkeys(e for e in endpoints if e))
+    endpoint = _first_nonempty(*endpoints)
+    runtime_success = _has_authoritative_runtime_success(bundle)
+    smoke_passed = _automatic_smoke_passed(bundle)
+    endpoint_known = bool(endpoint) or smoke_passed or runtime_success or _smoke_payload_available(bundle)
+    return {"endpoint": endpoint or "/generate", "endpoint_count": len(endpoints), "endpoint_known": endpoint_known, "endpoints": endpoints[:20]}
 
 
 def _contract_declares_no_full_inference(bundle: dict[str, Any]) -> bool:
+    if _has_authoritative_runtime_success(bundle):
+        return False
     contract = bundle.get("inference_contract") or bundle.get("INFERENCE_CONTRACT") or {}
     if not isinstance(contract, dict):
         contract = {}
@@ -803,6 +902,8 @@ def build_space_test_policy(bundle: dict[str, Any], status_model: dict[str, Any]
     global_status = _lower(status_model.get("global_status"))
     verdict = _lower(status_model.get("verdict"))
     endpoint_info = _known_gradio_endpoint_info(bundle)
+    runtime_success = _has_authoritative_runtime_success(bundle)
+    smoke_passed = _automatic_smoke_passed(bundle)
     requires_discovery = bool(target and not endpoint_info.get("endpoint_known"))
     runtime_stage = _lower(runtime.get("stage") or runtime.get("status")) if isinstance(runtime, dict) else ""
     health_passed = bool((bundle.get("inference_gate") or {}).get("implementation_signals", {}).get("health_passed") or smoke.get("health_passed"))
@@ -831,6 +932,17 @@ def build_space_test_policy(bundle: dict[str, Any], status_model: dict[str, Any]
 
     if not target:
         return policy("unavailable", False, "no_target_space", "Space Test unavailable", "No generated Space is available to validate.")
+    # v198.26.16: runtime proof beats stale Pi/contract policy blockers.  If the
+    # build has a terminal full-inference proof or a successful automatic smoke
+    # replay source, the Space Test panel must expose a replayable validation
+    # context instead of resurrecting "no generation endpoint".
+    if runtime_success or (smoke_passed and (global_status == "succeeded" or verdict in {"passed", "success", "full_inference_success"})):
+        replay = policy("replay", True, "automatic_smoke_passed", "Replay automatic smoke", "Automatic smoke test passed. You can replay the validated request.", allow_replay=True)
+        replay["runtime_proof_authoritative"] = bool(runtime_success)
+        replay["automatic_smoke_passed"] = bool(smoke_passed)
+        replay["no_generation_endpoint"] = False
+        replay["can_retry_schema"] = True
+        return replay
     if _repair_auth_context(bundle, status_model):
         return policy("recover", True, "auth_refresh_required", "Retry after sign-in refresh", "Repair patch may have been uploaded, but validation could not continue because HF OAuth expired. Sign in again, then retry linked validation.", on_success="recovered_by_manual_validation", allow_recovery=True)
     if _contract_declares_no_full_inference(bundle):
@@ -848,8 +960,12 @@ def build_space_test_policy(bundle: dict[str, Any], status_model: dict[str, Any]
         return policy("replay", True, "manual_validation_already_passed", "Replay linked validation", "This Build Run already has a successful linked Space Test. You can replay the validation.", allow_replay=True)
     if raw_status in {"completed_with_warnings", "success_with_warnings"}:
         return policy("replay", True, "completed_with_warnings", "Replay validation", "Build completed with warnings. Space Test can replay or confirm the validated endpoint.", allow_replay=True)
-    if global_status == "succeeded" or verdict in {"passed", "success"}:
-        return policy("replay", True, "automatic_smoke_passed", "Replay automatic smoke", "Automatic smoke test passed. You can replay the validated request.", allow_replay=True)
+    if global_status == "succeeded" or verdict in {"passed", "success", "full_inference_success"}:
+        replay = policy("replay", True, "automatic_smoke_passed", "Replay automatic smoke", "Automatic smoke test passed. You can replay the validated request.", allow_replay=True)
+        replay["automatic_smoke_passed"] = bool(smoke_passed)
+        replay["no_generation_endpoint"] = False
+        replay["can_retry_schema"] = True
+        return replay
     if global_status == "partial" or verdict == "partial_validation" or raw_status in PARTIAL_RAW_STATUSES:
         return policy("complete", True, "partial_generation_not_verified", "Complete validation", "Automatic smoke did not verify generation. Run linked Space Test to complete validation.", on_success="validated_after_manual_space_test")
     if global_status == "waiting_manual_action" or status_model.get("requires_manual_action"):
@@ -878,9 +994,10 @@ def build_space_test_model(bundle: dict[str, Any], status_model: dict[str, Any])
         endpoint = ""
     else:
         endpoint = policy.get("endpoint") or smoke.get("api_name") or smoke.get("endpoint") or "/generate"
-    parent_blocked = _lower(status_model.get("global_status")) in {"blocked", "failed", "auth_refresh_required", "cancelled"} or _lower(status_model.get("verdict")) in BLOCKED_RAW_STATUSES | FAILED_RAW_STATUSES | AUTH_RAW_STATUSES | {"cancelled", "stopped"}
-    smoke_evidence_passed = bool(metrics.get("generation_smoke_passed") or smoke.get("ok") or smoke.get("status") == "success")
-    preview_status = "blocked" if parent_blocked else "passed" if smoke_evidence_passed else "pending"
+    policy_blocks = policy.get("enabled") is False and policy.get("mode") in {"blocked", "unavailable"}
+    parent_blocked = bool(policy_blocks and (_lower(status_model.get("global_status")) in {"blocked", "failed", "auth_refresh_required", "cancelled"} or _lower(status_model.get("verdict")) in BLOCKED_RAW_STATUSES | FAILED_RAW_STATUSES | AUTH_RAW_STATUSES | {"cancelled", "stopped"}))
+    smoke_evidence_passed = bool(metrics.get("generation_smoke_passed") or smoke.get("ok") or smoke.get("status") == "success" or policy.get("automatic_smoke_passed") is True)
+    preview_status = "blocked" if parent_blocked else "passed" if smoke_evidence_passed or policy.get("mode") == "replay" else "pending"
     return {
         "target_space": policy.get("target_space") or summary.get("target_space") or "",
         "target_space_url": policy.get("target_space_url") or summary.get("target_space_url") or "",

src/worker_payload.py

@@ -35,8 +35,8 @@ DEFAULT_PREFERRED_SPACE_HARDWARE = "zero-a10g"
 DEFAULT_FALLBACK_SPACE_HARDWARE = "a10g-large"
 DEFAULT_MODEL_ID = "sshleifer/tiny-gpt2"
 MAX_PI_REPAIR_ATTEMPTS = 3
-APP_VERSION = "v198.26.14"
-app_version = "v198.26.14"
+APP_VERSION = "v198.26.17"
+app_version = "v198.26.17"
 
 # Internal agent/recovery files may be needed inside the transient Pi
 # workspace, but they should not be published to the generated Space or shown
@@ -12563,7 +12563,7 @@ def smoke_generate(target_space_id: str, token: str, run_dir: Path, events_path:
     # write_json(run_dir / "tests" / "payload_source.json", payload_source_record)
     # write_json(run_dir / "tests" / "replay_source.json", replay_source)
     # write_json(run_dir / "tests" / "validation_preflight.json", validation_preflight)
-    app_version = "v198.26.14"
+    app_version = "v198.26.17"
     engine_version = "unified_gradio_validation_harness_v198_25_3"
     parent_build_run_id = os.environ.get("PARENT_BUILD_RUN_ID", "").strip()
     validation_mode = os.environ.get("VALIDATION_MODE") or os.environ.get("SPACE_TEST_POLICY_MODE") or "linked"