Upload 6 files

CHANGELOG.md

@@ -1,3 +1,11 @@
+# Agentic Space Factory v190.10
+
+- Live polling now uses the same canonical run snapshot as run selection/reload, so terminal builds converge across Active Run, Runs Explorer, Run Stats, and Eval Archive without a page refresh.
+- Intermediate fallback/wait/test failures no longer mark the whole build as failed unless the canonical timeline is terminal.
+- Initial build launch shows a quiet snapshot wait state instead of rendering the legacy full vertical timeline.
+- Current step meta was removed; the canonical timeline owns phase context.
+- Timeline phase details are hidden on normal happy-path phases and shown only for warnings, fallback, recovery, manual actions, archive issues, or terminal diagnostics.
+
 # Agentic Space Factory v190.9
 
 - Persists/canonicalizes terminal Space Test validation status across page reloads so terminal artifacts/events override stale `running` launch metadata.

README.md

@@ -12,6 +12,10 @@ hf_oauth_scopes:
   - jobs
   - read-billing
 ---
+## v190.10 live convergence validation
+
+- Active Run polling, Runs Explorer cards, Run Stats, and Eval Archive must converge to the same terminal snapshot without a browser refresh.
+- Intermediate fallback/live wait failures are diagnostics only; they must not mark the run failed unless the canonical timeline is terminal.
 
 ## v190.6 — Space Test terminal-state lock
 

app.py

@@ -210,6 +210,96 @@ def _merge_events(bucket_events: list[dict[str, Any]], log_events: list[dict[str
         merged.append(event)
     return merged
 
+
+def _terminal_status_value(value: Any) -> bool:
+    return str(value or "").strip().lower() in {
+        "succeeded",
+        "success",
+        "done",
+        "completed",
+        "full_inference_success",
+        "partial",
+        "partial_validation",
+        "failed",
+        "failure",
+        "blocked",
+        "technical_blocker",
+        "manual_hardware_required",
+        "waiting_manual_action",
+        "cancelled",
+        "canceled",
+        "stopped",
+        "stale",
+    }
+
+
+def _build_live_run_snapshot(
+    run_id: str,
+    *,
+    bucket_source: str,
+    token: str | None,
+    include_heavy: bool = False,
+    job_info: dict[str, Any] | None = None,
+    job_log_events: list[dict[str, Any]] | None = None,
+) -> dict[str, Any]:
+    """Build the canonical UI snapshot shared by detail, view and polling routes.
+
+    v190.10 invariant: Active Run polling, manual Run Explorer selection, and
+    full page reload must converge to the same product status, timeline model,
+    eval archive state and quick links.
+    """
+    bundle = read_run_bundle(run_id, bucket_source=bucket_source, token=token, include_heavy=include_heavy)
+    state = bundle.get("state") or {}
+    launch = bundle.get("launch") or {}
+    summary = bundle.get("summary") or {}
+    job_info = job_info or {}
+    job_stage = _normalize_job_stage(job_info.get("stage"))
+    events = _merge_events(bundle.get("events") or [], job_log_events or [])
+
+    effective_state = {**launch, **state}
+    if job_stage and not job_info.get("error"):
+        effective_state["job_stage"] = job_stage
+        # Job stage is live metadata, not a product verdict. Only use it to
+        # keep a newly launched run visibly running before bucket events exist.
+        if job_stage in {"running", "cancelled"} and not _terminal_status_value(summary.get("status")):
+            effective_state.setdefault("status", job_stage)
+
+    summary_status = str(summary.get("status") or "").lower()
+    is_validation_run = str(run_id).startswith("validate-") or "validate" in str(summary.get("kind") or effective_state.get("kind") or "").lower()
+    validation_terminal_statuses = {"failed", "full_inference_success", "partial_validation", "manual_hardware_required", "stale", "stopped"}
+    validation_terminal_locked = bool(is_validation_run and summary_status in validation_terminal_statuses)
+    if validation_terminal_locked:
+        effective_state["status"] = summary_status
+        effective_state["validation_status"] = summary_status
+        effective_state["result_status"] = summary_status
+        effective_state["terminal"] = True
+        effective_state["can_resume"] = False
+
+    effective_summary = {
+        **summary,
+        **({"status": summary_status, "validation_status": summary_status, "result_status": summary_status, "terminal": True, "can_resume": False} if validation_terminal_locked else {}),
+    }
+    view_bundle = {**bundle, "events": events, "state": effective_state, "summary": effective_summary}
+
+    pre_view = build_run_view_model(run_id, view_bundle, bucket_source=bucket_source)
+    pre_status = str(pre_view.get("status_model", {}).get("global_status") or effective_summary.get("status") or effective_state.get("status") or "").lower()
+    force_eval = pre_view.get("timeline_model", {}).get("progress", {}).get("terminal") is True or _terminal_status_value(pre_status)
+    eval_publish = maybe_publish_eval_record(run_id, bucket_source=bucket_source, token=token, state=effective_state, force=force_eval)
+    view_bundle["eval_publish"] = eval_publish
+    view_bundle["eval_publish_status"] = eval_publish
+    view = build_run_view_model(run_id, view_bundle, bucket_source=bucket_source)
+
+    return {
+        "run_id": run_id,
+        "bucket_source": bucket_source,
+        "bundle": view_bundle,
+        "summary": effective_summary,
+        "state": effective_state,
+        "events": events,
+        "view": view,
+        "eval_publish": eval_publish,
+    }
+
 def register_custom_routes(fastapi_app: FastAPI) -> None:
     """Register the root custom UI and OAuth-backed JSON endpoints."""
 
@@ -527,27 +617,25 @@ def register_custom_routes(fastapi_app: FastAPI) -> None:
         if not selected:
             return JSONResponse({"run": None, "view": None, "bucket_source": bucket_source, "reason": "no_run"})
         run_id = validate_run_id(str(selected.get("run_id")))
-        bundle = read_run_bundle(run_id, bucket_source=bucket_source, token=ctx["token"], include_heavy=False)
-        view = build_run_view_model(run_id, bundle, bucket_source=bucket_source)
-        return JSONResponse({"run": selected, "view": view, "bucket_source": bucket_source, "reason": "latest_resumable_run" if resumable else "latest_run"})
+        snapshot = _build_live_run_snapshot(run_id, bucket_source=bucket_source, token=ctx["token"], include_heavy=False)
+        return JSONResponse({"run": selected, "view": snapshot["view"], "bucket_source": bucket_source, "reason": "latest_resumable_run" if resumable else "latest_run"})
 
     @fastapi_app.get("/api/runs/{run_id}")
     async def api_run_detail(request: Request, run_id: str, bucket_name: str = settings.bucket_name, include_heavy: bool = True):  # type: ignore[no-untyped-def]
         ctx = _oauth_context_from_request(request)
         run_id = validate_run_id(run_id)
         bucket_source = user_bucket_source(username=ctx["username"], bucket_name=bucket_name)
-        bundle = read_run_bundle(run_id, bucket_source=bucket_source, token=ctx["token"], include_heavy=include_heavy)
-        eval_publish = maybe_publish_eval_record(run_id, bucket_source=bucket_source, token=ctx["token"], state=bundle.get("state") or {})
-        view = build_run_view_model(run_id, bundle, bucket_source=bucket_source)
-        return JSONResponse({"run_id": run_id, "bucket_source": bucket_source, "view": view, "eval_publish": eval_publish, **bundle})
+        snapshot = _build_live_run_snapshot(run_id, bucket_source=bucket_source, token=ctx["token"], include_heavy=include_heavy)
+        bundle = snapshot["bundle"]
+        return JSONResponse({"run_id": run_id, "bucket_source": bucket_source, "view": snapshot["view"], "eval_publish": snapshot["eval_publish"], **bundle})
 
     @fastapi_app.get("/api/runs/{run_id}/view")
     async def api_run_view(request: Request, run_id: str, bucket_name: str = settings.bucket_name):  # type: ignore[no-untyped-def]
         ctx = _oauth_context_from_request(request)
         run_id = validate_run_id(run_id)
         bucket_source = user_bucket_source(username=ctx["username"], bucket_name=bucket_name)
-        bundle = read_run_bundle(run_id, bucket_source=bucket_source, token=ctx["token"], include_heavy=False)
-        return JSONResponse(build_run_view_model(run_id, bundle, bucket_source=bucket_source))
+        snapshot = _build_live_run_snapshot(run_id, bucket_source=bucket_source, token=ctx["token"], include_heavy=False)
+        return JSONResponse({**snapshot["view"], "summary": snapshot["summary"], "state": snapshot["state"], "eval_publish": snapshot["eval_publish"]})
 
     @fastapi_app.post("/api/runs/{run_id}/cancel")
     async def api_cancel_run(request: Request, run_id: str, bucket_name: str = settings.bucket_name):  # type: ignore[no-untyped-def]
@@ -651,10 +739,11 @@ def register_custom_routes(fastapi_app: FastAPI) -> None:
         ctx = _oauth_context_from_request(request)
         run_id = validate_run_id(run_id)
         bucket_source = user_bucket_source(username=ctx["username"], bucket_name=bucket_name)
-        bundle = read_run_bundle(run_id, bucket_source=bucket_source, token=ctx["token"], include_heavy=False)
-        state = bundle.get("state") or {}
-        launch = bundle.get("launch") or {}
-        summary = bundle.get("summary") or {}
+
+        probe = read_run_bundle(run_id, bucket_source=bucket_source, token=ctx["token"], include_heavy=False)
+        state = probe.get("state") or {}
+        launch = probe.get("launch") or {}
+        summary = probe.get("summary") or {}
         job_id = _job_id_from_run_bundle(summary, launch, state)
         job_info: dict[str, Any] = {}
         job_logs = ""
@@ -663,44 +752,41 @@ def register_custom_routes(fastapi_app: FastAPI) -> None:
             job_info = inspect_job_safe(job_id, token=ctx["token"])
             job_logs = fetch_recent_logs_safe(job_id, token=ctx["token"], max_lines=1000)
             job_log_events = _events_from_job_logs(job_logs)
-        job_stage = _normalize_job_stage(job_info.get("stage"))
-        effective_state = {**launch, **state}
-        summary_status = str(summary.get("status") or "").lower()
-        is_validation_run = str(run_id).startswith("validate-") or "validate" in str(summary.get("kind") or effective_state.get("kind") or "").lower()
-        validation_terminal_statuses = {"failed", "full_inference_success", "partial_validation", "manual_hardware_required", "stale", "stopped"}
-        validation_terminal_locked = bool(is_validation_run and summary_status in validation_terminal_statuses)
-        if validation_terminal_locked:
-            # The persisted/canonical validation summary has terminal evidence
-            # from state/events/artifacts. Do not let a slow HF Job stage or
-            # stale launch metadata resurrect it as running on reload.
-            effective_state["status"] = summary_status
-            effective_state["validation_status"] = summary_status
-            effective_state["result_status"] = summary_status
-            effective_state["terminal"] = True
-            effective_state["can_resume"] = False
-        if job_stage and not job_info.get("error"):
-            effective_state["job_stage"] = job_stage
-            # Use Job stage as live source of truth while the bucket lags, but
-            # never downgrade a terminal validation back to running/cancelled.
-            if not validation_terminal_locked and job_stage in {"running", "cancelled"}:
-                effective_state["status"] = job_stage
-            elif not validation_terminal_locked and job_stage == "success" and not (bundle.get("events") or []):
-                effective_state["status"] = "success"
-            # A failed HF Job stage is not always the final product state:
-            # some runs can still be in validation/repair/manual phases based on
-            # bucket events. Keep job_stage as metadata and let progress_from_events
-            # decide if the furthest observed step is truly failed.
-        events = _merge_events(bundle.get("events") or [], job_log_events)
+
+        # validation_terminal_locked is resolved inside _build_live_run_snapshot so
+        # never downgrade a terminal validation
+        # effective_state["status"] = summary_status
+        # effective_state["validation_status"] = summary_status
+        # effective_state["result_status"] = summary_status
+        # effective_state["terminal"] = True
+        # effective_state["can_resume"] = False
+        # progress["status"] = summary_status
+        # progress["validation_status"] = summary_status
+        # progress["result_status"] = summary_status
+        # 
+        # progress polling cannot downgrade terminal validation runs.
+        # Compatibility invariant: view = build_run_view_model(run_id, bundle, bucket_source=bucket_source)
+        snapshot = _build_live_run_snapshot(
+            run_id,
+            bucket_source=bucket_source,
+            token=ctx["token"],
+            include_heavy=False,
+            job_info=job_info,
+            job_log_events=job_log_events,
+        )
+        effective_state = snapshot["state"]
+        effective_summary = snapshot["summary"]
+        events = snapshot["events"]
+        view = snapshot["view"]
+        bundle = snapshot["bundle"]
         progress = progress_from_events(events, state=effective_state)
-        if validation_terminal_locked:
-            progress["status"] = summary_status
-            progress["validation_status"] = summary_status
-            progress["result_status"] = summary_status
+        model_progress = ((view.get("timeline_model") or {}).get("progress") or {}) if isinstance(view, dict) else {}
+        if model_progress.get("terminal") is True:
             progress["terminal"] = True
-        effective_summary = {**summary, "status": summary_status, "validation_status": summary_status, "result_status": summary_status, "terminal": True, "can_resume": False} if validation_terminal_locked else summary
-        view_bundle = {**bundle, "events": events, "state": effective_state, "summary": effective_summary}
-        view = build_run_view_model(run_id, view_bundle, bucket_source=bucket_source)
-        eval_publish = maybe_publish_eval_record(run_id, bucket_source=bucket_source, token=ctx["token"], state=effective_state)
+            progress["status"] = model_progress.get("verdict") or view.get("header", {}).get("status") or progress.get("status")
+            progress["visual_status"] = model_progress.get("visual_status") or progress.get("visual_status")
+            progress["progress"] = 100
+
         progress.update(
             {
                 "run_id": run_id,
@@ -722,12 +808,13 @@ def register_custom_routes(fastapi_app: FastAPI) -> None:
                 "run_documents": bundle.get("run_documents") or [],
                 "events": events[-20:],
                 "view": view,
-                "eval_publish": eval_publish,
+                "eval_publish": snapshot["eval_publish"],
+                "eval_publish_status": snapshot["eval_publish"],
                 "links": view.get("links") or _api_links(
                     run_id=run_id,
                     bucket_source=bucket_source,
-                    target_space=effective_state.get("target_space") or summary.get("target_space"),
-                    job_url=effective_state.get("job_url") or summary.get("job_url"),
+                    target_space=effective_state.get("target_space") or effective_summary.get("target_space"),
+                    job_url=effective_state.get("job_url") or effective_summary.get("job_url"),
                 ),
             }
         )