Stream live worker progress via async build dispatch

The agent's /upload and /checkout_git were synchronous, blocking until a
build finished, so the controller could not relay live per-step progress.
Add ?async=1 support: the agent saves/clones the source, starts the build in
a background thread (holding build_lock), and returns 202 + {"id": ...}
immediately. The controller dispatches async, then polls /progress/<job_id>,
appending each step/log line to the job log and SSE stream, and treats the
build as done when progress clears (404) and the worker reports ready.

ctrl/scheduler.py

@@ -1,6 +1,7 @@
 import datetime
 import os
 import threading
+import time
 
 import requests
 
@@ -97,21 +98,22 @@ class Scheduler:
 
         try:
             if job["source_type"] == "git":
-                worker_job = self._request_git(job, worker)
+                worker_job_id = self._request_git(job, worker)
             else:
-                worker_job = self._request_upload(job, worker)
+                worker_job_id = self._request_upload(job, worker)
         except Exception as e:
             db.update_job(job_id, status="failed", finished_at=datetime.datetime.now().isoformat(),
                           error=str(e))
             self._append(job_id, f"[scheduler] FAILED: {e}\n")
             return
 
-        worker_job_id = worker_job.get("id")
         worker_url = worker["url"]
         db.update_job(job_id, status="building", worker_job_id=worker_job_id, worker_url=worker_url)
+        self._append(job_id, f"[scheduler] Building on {worker_url} (worker job {worker_job_id})\n")
 
-        # download artifacts
+        # Poll progress until the worker reports the build finished.
         try:
+            self._wait_for_build(job, worker, worker_job_id)
             self._fetch_artifacts(job, worker, worker_job_id)
             db.update_job(job_id, status="done", finished_at=datetime.datetime.now().isoformat())
             self._append(job_id, "[scheduler] Build complete, artifacts saved.\n")
@@ -123,28 +125,65 @@ class Scheduler:
             self._publish(job["id"], "job", '"done"')
 
     def _request_git(self, job, worker):
-        self._append(job["id"], f"[scheduler] Cloning {job['source']} on {worker['url']}\n")
-        resp = requests.post(f"{worker['url']}/checkout_git",
+        job_id = job["id"]
+        self._append(job_id, f"[scheduler] Cloning {job['source']} on {worker['url']}\n")
+        resp = requests.post(f"{worker['url']}/checkout_git?async=1",
                              json={"repo_url": job["source"]},
-                             timeout=settings.WORKER_BUILD_TIMEOUT,
+                             timeout=settings.WORKER_STATUS_TIMEOUT,
                              stream=False)
-        return self._worker_response(resp, "checkout")
+        data = self._worker_response(resp, "checkout")
+        if not data.get("id"):
+            raise RuntimeError("checkout did not return a job id")
+        return data["id"]
 
     def _request_upload(self, job, worker):
+        job_id = job["id"]
         path = self._source_path(job)
         if not path or not os.path.exists(path):
             raise FileNotFoundError(f"Source file not found: {path}")
-        self._append(job["id"], f"[scheduler] Uploading {os.path.basename(path)} to {worker['url']}\n")
+        self._append(job_id, f"[scheduler] Uploading {os.path.basename(path)} to {worker['url']}\n")
         with open(path, "rb") as f:
-            resp = requests.post(f"{worker['url']}/upload",
+            resp = requests.post(f"{worker['url']}/upload?async=1",
                                  files={"file": (os.path.basename(path), f, "application/zip")},
-                                 timeout=settings.WORKER_BUILD_TIMEOUT,
+                                 timeout=settings.WORKER_STATUS_TIMEOUT,
                                  stream=False)
-        worker_job = self._worker_response(resp, "upload")
-        if not worker_job.get("id"):
-            # Agent may have started building before returning; try progress fallback.
-            pass
-        return worker_job
+        data = self._worker_response(resp, "upload")
+        if not data.get("id"):
+            raise RuntimeError("upload did not return a job id")
+        return data["id"]
+
+    def _wait_for_build(self, job, worker, worker_job_id):
+        """Poll worker progress until the build completes (progress cleared + worker ready)."""
+        job_id = job["id"]
+        deadline = datetime.datetime.now() + datetime.timedelta(seconds=settings.WORKER_BUILD_TIMEOUT)
+        while datetime.datetime.now() < deadline:
+            progress = self._probe_progress(job, worker["url"], worker_job_id)
+            if progress is None:
+                status = wrk.probe_worker(worker["url"])
+                if status and status.get("status") == "ready":
+                    return
+                # progress gone but worker not ready: brief window, keep waiting
+                time.sleep(settings.SCHEDULER_INTERVAL)
+                continue
+            line = progress.get("last_log_line") or progress.get("step") or ""
+            if line:
+                self._append(job_id, f"[worker] {line}\n")
+            time.sleep(settings.SCHEDULER_INTERVAL)
+        raise RuntimeError(f"Worker did not finish in {settings.WORKER_BUILD_TIMEOUT}s")
+
+    @staticmethod
+    def _probe_progress(job, worker_url, worker_job_id):
+        try:
+            resp = requests.get(f"{worker_url}/progress/{worker_job_id}",
+                                timeout=settings.WORKER_STATUS_TIMEOUT)
+        except requests.RequestException:
+            return None
+        if resp.status_code == 404:
+            return None
+        try:
+            return resp.json()
+        except ValueError:
+            return None
 
     @staticmethod
     def _worker_response(resp, action):