Transition analysis API to asynchronous inference orchestrator

apps/desktop/src-tauri/src/main.rs

@@ -10,13 +10,12 @@ use std::{
     process::{Command, Stdio},
     sync::{
         atomic::{AtomicU64, AtomicUsize, Ordering},
-        mpsc,
-        Arc, Mutex,
+        mpsc, Arc, Mutex,
     },
     thread,
     time::{Duration, Instant},
 };
-use tauri::{Manager, Runtime};
+use tauri::{Emitter, Manager, Runtime};
 use time::{format_description::well_known::Rfc3339, OffsetDateTime};
 
 #[derive(Clone)]
@@ -602,12 +601,21 @@ fn failed_status(
     }
 }
 
-fn store_status(state: &AppState, status: AnalysisJobStatus) {
+fn store_status(state: &AppState, status: &AnalysisJobStatus) {
     if let Ok(mut jobs) = state.0.jobs.lock() {
-        jobs.insert(status.job_id.clone(), status);
+        jobs.insert(status.job_id.clone(), status.clone());
     }
 }
 
+fn store_status_and_emit<R: Runtime>(
+    state: &AppState,
+    app: &tauri::AppHandle<R>,
+    status: &AnalysisJobStatus,
+) {
+    store_status(state, status);
+    let _ = app.emit("analysis-job-updated", status);
+}
+
 fn store_bootstrap_source(state: &AppState, summary: ProjectBootstrapSummaryPayload) {
     if let Ok(mut sources) = state.0.bootstrap_sources.lock() {
         sources.insert(summary.project_id.clone(), summary);
@@ -629,17 +637,19 @@ fn lookup_bootstrap_source(
 
 fn drain_analysis_status_updates(
     state: &AppState,
+    app: &tauri::AppHandle<impl Runtime>,
     status_rx: &mpsc::Receiver<AnalysisJobStatus>,
     last_status: &mut Option<AnalysisJobStatus>,
 ) {
     while let Ok(status) = status_rx.try_recv() {
-        store_status(state, status.clone());
+        store_status_and_emit(state, app, &status);
         *last_status = Some(status);
     }
 }
 
 fn run_analysis_engine(
     state: AppState,
+    app: tauri::AppHandle<impl Runtime>,
     job_id: String,
     request: AnalysisJobRequest,
     requested_at: String,
@@ -749,7 +759,7 @@ fn run_analysis_engine(
     let mut last_status = None;
     let exit_status;
     loop {
-        drain_analysis_status_updates(&state, &status_rx, &mut last_status);
+        drain_analysis_status_updates(&state, &app, &status_rx, &mut last_status);
         match process.try_wait() {
             Ok(Some(status)) => {
                 exit_status = status;
@@ -792,7 +802,7 @@ fn run_analysis_engine(
     }
     let reader_last_status = stdout_reader.join().unwrap_or(None);
     let _ = stderr_reader.join();
-    drain_analysis_status_updates(&state, &status_rx, &mut last_status);
+    drain_analysis_status_updates(&state, &app, &status_rx, &mut last_status);
     if last_status.is_none() {
         last_status = reader_last_status;
     }
@@ -823,7 +833,11 @@ fn run_analysis_engine(
 }
 
 #[tauri::command]
-fn start_analysis_job(request: Value, state: tauri::State<'_, AppState>) -> AnalysisJobStatus {
+fn start_analysis_job(
+    request: Value,
+    app: tauri::AppHandle<impl Runtime>,
+    state: tauri::State<'_, AppState>,
+) -> AnalysisJobStatus {
     let requested_at = iso_timestamp_now();
     let mut parsed_request = match parse_request_payload(request) {
         Ok(parsed_request) => parsed_request,
@@ -884,13 +898,15 @@ fn start_analysis_job(request: Value, state: tauri::State<'_, AppState>) -> Anal
         result: None,
         error: None,
     };
-    store_status(&state, queued.clone());
+    store_status_and_emit(&state, &app, &queued);
 
     let app_state = state.inner().clone();
+    let worker_app_handle = app.clone();
     std::thread::spawn(move || {
-        store_status(
+        store_status_and_emit(
             &app_state,
-            AnalysisJobStatus {
+            &worker_app_handle,
+            &AnalysisJobStatus {
                 job_id: job_id.clone(),
                 state: AnalysisJobState::Running,
                 requested_at: requested_at.clone(),
@@ -903,8 +919,14 @@ fn start_analysis_job(request: Value, state: tauri::State<'_, AppState>) -> Anal
                 error: None,
             },
         );
-        let finished = run_analysis_engine(app_state.clone(), job_id, parsed_request, requested_at);
-        store_status(&app_state, finished);
+        let finished = run_analysis_engine(
+            app_state.clone(),
+            worker_app_handle.clone(),
+            job_id,
+            parsed_request,
+            requested_at,
+        );
+        store_status_and_emit(&app_state, &worker_app_handle, &finished);
         release_job_slot(&app_state);
     });
 
@@ -1385,7 +1407,10 @@ mod tests {
             "YouTube import timed out.",
         );
 
-        assert_eq!(result.expect_err("slow child should time out"), "YouTube import timed out.");
+        assert_eq!(
+            result.expect_err("slow child should time out"),
+            "YouTube import timed out."
+        );
     }
 
     #[test]

apps/desktop/src/App.test.tsx

@@ -5,7 +5,9 @@ import { App } from "./App";
 const tauriInvoke = vi.fn();
 const mockLoadProject = vi.fn();
 const mockSaveProject = vi.fn();
+const mockSubscribeToAnalysisJobUpdates = vi.fn();
 let mockImportYoutubeUrlError = false;
+let latestStatusSubscription: ((payload: Record<string, unknown>) => void) | null = null;
 
 type TauriWindow = Window & {
   __TAURI_INTERNALS__?: unknown;
@@ -30,6 +32,8 @@ vi.mock("./lib/analysis", async (importActual) => {
       sourceLabel: "Late Night Set",
       roleFocus: ["bass-guitar", "keys-right", "lead-vocal"]
     }),
+    subscribeToAnalysisJobUpdates: (...args: Parameters<typeof mockSubscribeToAnalysisJobUpdates>) =>
+      mockSubscribeToAnalysisJobUpdates(...args),
     loadProject: () => mockLoadProject(),
     saveProject: (song: unknown) => mockSaveProject(song)
   };
@@ -171,7 +175,17 @@ describe("App", () => {
     tauriInvoke.mockReset();
     mockLoadProject.mockReset();
     mockSaveProject.mockReset();
+    mockSubscribeToAnalysisJobUpdates.mockReset();
     mockImportYoutubeUrlError = false;
+    latestStatusSubscription = null;
+    mockSubscribeToAnalysisJobUpdates.mockImplementation(
+      async (_jobId: string, onUpdate: (status: Record<string, unknown>) => void) => {
+        latestStatusSubscription = onUpdate;
+        return () => {
+          latestStatusSubscription = null;
+        };
+      }
+    );
     delete tauriWindow.__TAURI_INTERNALS__;
     tauriWindow.__TAURI_INVOKE__ = tauriInvoke;
   });
@@ -407,6 +421,52 @@ describe("App", () => {
     );
   });
 
+  it("applies pushed analysis status updates over the IPC event bridge", async () => {
+    tauriInvoke
+      .mockResolvedValueOnce(bootstrapResponse())
+      .mockResolvedValueOnce(jobStatusResponse({
+        jobId: "job-push-1",
+        state: "queued",
+        progressLabel: "Queued for analysis"
+      }));
+
+    render(<App />);
+
+    fireEvent.click(screen.getByRole("button", { name: /choose local audio/i }));
+    await waitFor(() => {
+      expect(screen.getByText(/late-night-set\.wav/i)).toBeTruthy();
+    });
+
+    fireEvent.click(screen.getByRole("button", { name: /start analysis/i }));
+    await waitFor(() => {
+      expect(screen.getByText(/queued for analysis/i)).toBeTruthy();
+    });
+    await waitFor(() => {
+      expect(mockSubscribeToAnalysisJobUpdates).toHaveBeenCalledWith(
+        "job-push-1",
+        expect.any(Function)
+      );
+    });
+
+    latestStatusSubscription?.(
+      jobStatusResponse({
+        jobId: "job-push-1",
+        state: "running",
+        progressLabel: "Separating stems... (45%)",
+        progressStage: "separate",
+        progressPercent: 45
+      })
+    );
+    await waitFor(() => {
+      expect(screen.getByText(/separating stems/i)).toBeTruthy();
+    });
+
+    latestStatusSubscription?.(succeededResult());
+    await waitFor(() => {
+      expect(screen.getByRole("heading", { name: /Late Night Set/i })).toBeTruthy();
+    });
+  });
+
   it("keeps handoff metadata tied to the source that produced the current result", async () => {
     const originalCreateObjectUrl = URL.createObjectURL;
     const originalRevokeObjectUrl = URL.revokeObjectURL;

apps/desktop/src/App.tsx

@@ -1,4 +1,4 @@
-import { useEffect, useMemo, useState, type ReactNode } from "react";
+import { useCallback, useEffect, useMemo, useState, type ReactNode } from "react";
 import {
   AudioWaveform,
   CircleHelp,
@@ -36,6 +36,7 @@ import {
   isSupportedYoutubeUrl,
   loadProject,
   saveProject,
+  subscribeToAnalysisJobUpdates,
   selectLocalAudioSource,
   startAnalysisJob
 } from "./lib/analysis";
@@ -177,6 +178,7 @@ export function App() {
   const [jobResult, setJobResult] = useState<RehearsalSong | null>(null);
   const [jobResultBootstrap, setJobResultBootstrap] = useState<ProjectBootstrapSummary | null>(null);
   const [jobError, setJobError] = useState<string | null>(null);
+  const [renderedProgressPercent, setRenderedProgressPercent] = useState<number | undefined>(undefined);
   const [isStarting, setIsStarting] = useState(false);
   const [selectedBootstrap, setSelectedBootstrap] = useState<ProjectBootstrapSummary | null>(null);
   const [activeAnalysisBootstrap, setActiveAnalysisBootstrap] = useState<ProjectBootstrapSummary | null>(null);
@@ -194,6 +196,71 @@ export function App() {
       }
     : defaultRequest;
 
+  /** Documented. */
+  const applyJobStatus = useCallback((nextStatus: AnalysisJobStatus) => {
+    setJobStatus(nextStatus);
+    if (nextStatus.state === "succeeded" && nextStatus.result) {
+      setJobResult(nextStatus.result);
+      setJobResultBootstrap(activeAnalysisBootstrap);
+      setActiveAnalysisBootstrap(null);
+      setJobError(null);
+    }
+    if (nextStatus.state === "failed") {
+      setActiveAnalysisBootstrap(null);
+      setJobError(nextStatus.error?.message ?? t("analysisCouldNotStart"));
+    }
+  }, [activeAnalysisBootstrap, t]);
+
+  useEffect(() => {
+    const targetPercent = jobStatus?.progressPercent;
+    if (targetPercent === undefined) {
+      setRenderedProgressPercent(undefined);
+      return;
+    }
+    if (jobStatus?.state === "failed" || jobStatus?.state === "succeeded") {
+      setRenderedProgressPercent(targetPercent);
+      return;
+    }
+
+    const currentPercent = renderedProgressPercent ?? 0;
+    if (currentPercent >= targetPercent) {
+      setRenderedProgressPercent(targetPercent);
+      return;
+    }
+
+    const timer = window.setTimeout(() => {
+      setRenderedProgressPercent((current) => {
+        const base = current ?? 0;
+        return Math.min(targetPercent, base + 1);
+      });
+    }, 20);
+    return () => window.clearTimeout(timer);
+  }, [jobStatus?.progressPercent, jobStatus?.state, renderedProgressPercent]);
+
+  useEffect(() => {
+    if (!jobStatus || (jobStatus.state !== "queued" && jobStatus.state !== "running")) {
+      return;
+    }
+
+    let disposed = false;
+    let unsubscribe: () => void = Function.prototype as () => void;
+    void subscribeToAnalysisJobUpdates(jobStatus.jobId, (nextStatus) => {
+      if (!disposed) {
+        applyJobStatus(nextStatus);
+      }
+    }).then((cleanup) => {
+      if (disposed) {
+        cleanup();
+        return;
+      }
+      unsubscribe = cleanup;
+    });
+    return () => {
+      disposed = true;
+      unsubscribe();
+    };
+  }, [applyJobStatus, jobStatus?.jobId, jobStatus?.state]);
+
   useEffect(() => {
     if (!jobStatus || (jobStatus.state !== "queued" && jobStatus.state !== "running")) {
       return;
@@ -202,17 +269,7 @@ export function App() {
     const timer = window.setTimeout(async () => {
       try {
         const nextStatus = await getAnalysisJobStatus(jobStatus.jobId);
-        setJobStatus(nextStatus);
-        if (nextStatus.state === "succeeded" && nextStatus.result) {
-          setJobResult(nextStatus.result);
-          setJobResultBootstrap(activeAnalysisBootstrap);
-          setActiveAnalysisBootstrap(null);
-          setJobError(null);
-        }
-        if (nextStatus.state === "failed") {
-          setActiveAnalysisBootstrap(null);
-          setJobError(nextStatus.error?.message ?? t("analysisCouldNotStart"));
-        }
+        applyJobStatus(nextStatus);
       } catch (error) {
         if (error instanceof Error && error.message === "Invalid analysis job status response") {
           const fallbackMessage = t("analysisCouldNotStart");
@@ -242,7 +299,7 @@ export function App() {
     }, ANALYSIS_POLL_INTERVAL_MS);
 
     return () => window.clearTimeout(timer);
-  }, [activeAnalysisBootstrap, jobStatus, t]);
+  }, [applyJobStatus, jobStatus, t]);
 
   /** Documented. */
   const handleStartAnalysis = async () => {
@@ -255,15 +312,13 @@ export function App() {
     setIsStarting(true);
     try {
       const nextStatus = await startAnalysisJob(selectedRequest);
-      setJobStatus(nextStatus);
       if (nextStatus.state === "succeeded" && nextStatus.result) {
+        setJobStatus(nextStatus);
         setJobResult(nextStatus.result);
         setJobResultBootstrap(submittedBootstrap);
         setActiveAnalysisBootstrap(null);
-      }
-      if (nextStatus.state === "failed") {
-        setActiveAnalysisBootstrap(null);
-        setJobError(nextStatus.error?.message ?? t("analysisCouldNotStart"));
+      } else {
+        applyJobStatus(nextStatus);
       }
     } catch {
       setJobStatus(null);
@@ -585,13 +640,15 @@ export function App() {
                       {jobStatus.state === "running" && <span className="inline-block size-4 shrink-0 animate-spin rounded-full border-2 border-cyan-100/30 border-t-cyan-200" />}
                       <span className="min-w-0 flex-1 truncate">{progressMessage(t, jobStatus)}</span>
                       {jobStatus.progressPercent !== undefined && (
-                        <span className="shrink-0 tabular-nums text-cyan-50/80">{jobStatus.progressPercent}%</span>
+                        <span className="shrink-0 tabular-nums text-cyan-50/80">
+                          {(renderedProgressPercent ?? jobStatus.progressPercent)}%
+                        </span>
                       )}
                     </div>
                     {jobStatus.progressPercent !== undefined && (
                       <Progress
                         aria-label="Analysis progress"
-                        value={jobStatus.progressPercent}
+                        value={renderedProgressPercent ?? jobStatus.progressPercent}
                         className="mt-2"
                       />
                     )}