Convert pipeline /api/run to background job + polling

server.py

@@ -230,6 +230,58 @@ async def get_screening_history():
 
 
 # ── Pipeline ─────────────────────────────────────────────────────────────────
+# The pipeline can take 2-15 minutes per run depending on resume count, LLM
+# rate limits, and whether JobBERT needs warming. Holding the HTTP request
+# open that long doesn't work behind HF Spaces' ~90s proxy timeout — the
+# browser sees a disconnect while the subprocess keeps running silently.
+#
+# So we run main.py in a background thread, key it by a job UUID, and let
+# the UI poll /api/run/status/{job_id} until the job reports completed.
+
+import threading as _threading
+import uuid as _uuid_pipeline
+import time as _time
+
+_PIPELINE_JOBS: dict = {}                  # job_id -> {status, result, started_at, finished_at, cmd}
+_PIPELINE_LOCK = _threading.Lock()
+
+def _strip_ansi(s: str) -> str:
+    return re.sub(r'\x1b\[[0-9;]*m', '', s or '')
+
+
+def _run_pipeline_worker(job_id: str, cmd: list[str]) -> None:
+    """Background-thread worker that runs main.py and stores the result."""
+    with _PIPELINE_LOCK:
+        _PIPELINE_JOBS[job_id]["status"] = "running"
+    try:
+        result = subprocess.run(
+            cmd, capture_output=True, text=True,
+            cwd=str(BASE),
+            timeout=1500,   # 25 min — large headroom for first-run JobBERT warm-up + LLM retries
+        )
+        with _PIPELINE_LOCK:
+            _PIPELINE_JOBS[job_id].update({
+                "status":      "completed" if result.returncode == 0 else "failed",
+                "returncode":  result.returncode,
+                "stdout":      _strip_ansi(result.stdout[-3000:] if result.stdout else ""),
+                "stderr":      _strip_ansi(result.stderr[-2000:] if result.stderr else ""),
+                "finished_at": _time.time(),
+            })
+    except subprocess.TimeoutExpired:
+        with _PIPELINE_LOCK:
+            _PIPELINE_JOBS[job_id].update({
+                "status":      "timeout",
+                "stderr":      "Pipeline exceeded 25-minute internal timeout.",
+                "finished_at": _time.time(),
+            })
+    except Exception as exc:
+        with _PIPELINE_LOCK:
+            _PIPELINE_JOBS[job_id].update({
+                "status":      "error",
+                "stderr":      f"Worker crashed: {type(exc).__name__}: {exc}",
+                "finished_at": _time.time(),
+            })
+
 
 @app.post("/api/run")
 async def run_pipeline(
@@ -238,14 +290,15 @@ async def run_pipeline(
     jd: str = Query(default=""),
     weights: str = Query(default=""),
 ):
+    """
+    Kick off the pipeline as a background job and return a job_id immediately
+    (within ~10ms). The UI polls /api/run/status/{job_id} for completion.
+    """
     if not MAIN_PY.exists():
         raise HTTPException(status_code=404, detail="main.py not found")
 
     cmd = [sys.executable, str(MAIN_PY), "--top", str(top_n), "--explain", str(explain_n)]
     if jd:
-        # Mirror jd_parser's title derivation so the substring filter in main.py
-        # matches the parsed jd_title. e.g. "Jr. Accountant - JD.docx" must
-        # become "Jr. Accountant" — not the raw "Jr. Accountant - JD" stem.
         stem = jd.rsplit(".", 1)[0]
         stem = stem.replace("_", " ").replace("-", " ")
         stem = re.sub(r"\s+[Jj][Dd]$", "", stem).strip()
@@ -254,29 +307,65 @@ async def run_pipeline(
         stem = re.sub(r"\s+", " ", stem).strip()
         cmd += ["--jd", stem]
     if weights:
-        # Pass the weights JSON through verbatim; main.py validates it.
         cmd += ["--weights", weights]
 
-    try:
-        result = subprocess.run(
-            cmd,
-            capture_output=True,
-            text=True,
-            cwd=str(BASE),
-            timeout=900,   # 15 min — accommodates LLM rate-limit retries
-        )
-        import re as _re
-        def _strip_ansi(s: str) -> str:
-            return _re.sub(r'\x1b\[[0-9;]*m', '', s or '')
+    # Reject concurrent runs — main.py writes to a shared SQLite + CSVs;
+    # parallel runs would corrupt them.
+    with _PIPELINE_LOCK:
+        running = [jid for jid, j in _PIPELINE_JOBS.items()
+                   if j.get("status") in ("queued", "running")]
+        if running:
+            raise HTTPException(status_code=409,
+                                detail=f"Another pipeline is still running: {running[0]}")
+        job_id = str(_uuid_pipeline.uuid4())
+        _PIPELINE_JOBS[job_id] = {
+            "status":     "queued",
+            "cmd":        cmd,
+            "started_at": _time.time(),
+        }
+
+    _threading.Thread(target=_run_pipeline_worker,
+                      args=(job_id, cmd), daemon=True).start()
+    return {"job_id": job_id, "status": "queued"}
 
+
+@app.get("/api/run/status/{job_id}")
+async def run_status(job_id: str):
+    """
+    Return the current status of a pipeline job. UI polls this every few
+    seconds. Returns 404 only for truly unknown IDs — otherwise always
+    returns a dict with at least `status`.
+    """
+    with _PIPELINE_LOCK:
+        job = _PIPELINE_JOBS.get(job_id)
+    if not job:
+        raise HTTPException(status_code=404, detail="Unknown job_id")
+
+    elapsed = _time.time() - job["started_at"]
+    return {
+        "job_id":     job_id,
+        "status":     job["status"],
+        "elapsed_s":  round(elapsed, 1),
+        "success":    job["status"] == "completed",
+        "returncode": job.get("returncode"),
+        "stdout":     job.get("stdout", ""),
+        "stderr":     job.get("stderr", ""),
+    }
+
+
+@app.get("/api/run/jobs")
+async def run_jobs():
+    """Debug: list every job and its current status. Useful when a UI lost its job_id."""
+    with _PIPELINE_LOCK:
         return {
-            "success": result.returncode == 0,
-            "returncode": result.returncode,
-            "stdout": _strip_ansi(result.stdout[-3000:] if result.stdout else ""),
-            "stderr": _strip_ansi(result.stderr[-2000:] if result.stderr else ""),
+            "jobs": [
+                {"job_id": jid, "status": j["status"],
+                 "elapsed_s": round(_time.time() - j["started_at"], 1)}
+                for jid, j in sorted(_PIPELINE_JOBS.items(),
+                                     key=lambda kv: kv[1].get("started_at", 0),
+                                     reverse=True)
+            ][:20]
         }
-    except subprocess.TimeoutExpired:
-        raise HTTPException(status_code=408, detail="Pipeline timed out after 15 minutes")
 
 
 # ── Results & Candidates ─────────────────────────────────────────────────────

static/app.js

@@ -532,30 +532,65 @@ async function runPipeline() {
 
     btn.disabled = true;
     btn.innerHTML = '<span class="spinner"></span> Running…';
-    showAlert('pipeline-status', 'info', '<span class="spinner"></span> Pipeline is running — please wait…');
+    showAlert('pipeline-status', 'info', '<span class="spinner"></span> Submitting pipeline…');
     document.getElementById('pipeline-log-wrap').classList.add('hidden');
 
     let url = `/api/run?top_n=${topN}&explain_n=${expN}` +
               `&jd=${encodeURIComponent(jd)}` +
               `&weights=${encodeURIComponent(JSON.stringify(weightsPayload))}`;
 
-    const data = await api(url, { method:'POST' });
+    // Phase 1: enqueue the job (returns in ~10 ms with a job_id).
+    const start = await api(url, { method: 'POST' });
+
+    if (!start || !start.job_id) {
+        btn.disabled = false;
+        btn.innerHTML = '▶&nbsp; Run Pipeline';
+        showAlert('pipeline-status', 'error',
+            '✗ Could not start pipeline — check server logs.');
+        return;
+    }
+
+    const jobId = start.job_id;
+    showAlert('pipeline-status', 'info',
+        `<span class="spinner"></span> Pipeline running… (job ${jobId.slice(0, 8)}) ` +
+        `<span id="pipe-elapsed">0s</span>`);
+
+    // Phase 2: poll status every 3 seconds until it finishes or errors.
+    // This survives HF Spaces' ~90s HTTP proxy timeout because each poll is
+    // a fresh quick request.
+    let data = null;
+    while (true) {
+        await new Promise(r => setTimeout(r, 3000));
+        const status = await api(`/api/run/status/${jobId}`);
+        if (!status) continue;   // transient network blip — try again
+
+        const el = document.getElementById('pipe-elapsed');
+        if (el) el.textContent = `${Math.round(status.elapsed_s)}s`;
+
+        if (status.status === 'completed' || status.status === 'failed' ||
+            status.status === 'error'     || status.status === 'timeout') {
+            data = status;
+            break;
+        }
+    }
 
     btn.disabled = false;
     btn.innerHTML = '▶&nbsp; Run Pipeline';
 
-    if (!data) {
-        showAlert('pipeline-status', 'error', '✗ Request failed — check server is running.');
-        return;
-    }
     if (data.success) {
         STATE.lastPipelineJd = jd ? jd.replace(/\.[^.]+$/, '') : '';
         showAlert('pipeline-status', 'success',
-            '✓ Pipeline completed successfully. <a href="#" onclick="switchPage(\'results\');return false;" ' +
+            `✓ Pipeline completed in ${Math.round(data.elapsed_s)}s. ` +
+            '<a href="#" onclick="switchPage(\'results\');return false;" ' +
             'style="color:var(--green);font-weight:600;text-decoration:underline">→ View Results</a>');
         setTimeout(() => switchPage('results'), 1500);
     } else {
-        showAlert('pipeline-status', 'error', `✗ Pipeline failed (exit code ${data.returncode})`);
+        const reason = data.status === 'timeout'
+            ? 'timed out'
+            : data.returncode != null
+                ? `failed (exit ${data.returncode})`
+                : `errored (${data.status})`;
+        showAlert('pipeline-status', 'error', `✗ Pipeline ${reason}`);
     }
     if (data.stdout || data.stderr) {
         document.getElementById('pipeline-log-wrap').classList.remove('hidden');