fix: report worker availability for automatic uploads

admin/components/create-run.tsx

@@ -24,8 +24,8 @@ export function SetupChecks({ report }: { report: SetupReport }) {
     <section className="setup-checks" aria-label="Pipeline setup checks" aria-live="polite">
       <h3>{report.ready ? 'Local setup checks passed' : 'Setup needs attention'}</h3>
       <p className="field__hint">
-        Checks describe this server. Provider access and worker readiness may still need
-        verification.
+        Checks include whether a worker is connected for new uploads. Provider access and processing
+        quality still need verification.
       </p>
       <ul>
         {checks.map((check) => (
@@ -310,8 +310,8 @@ export function CreateRun({ onCreated }: { onCreated: (runId: string) => void })
         </p>
       ) : null}
       <p className="field__hint">
-        Generated worlds, tracking and fine-tuning can use paid providers. Inspect the cleaned
-        frames when the run requests review. Original recorded audio is preserved.
+        Generated worlds, tracking and fine-tuning can use paid providers. Original recorded audio
+        is preserved.
       </p>
     </form>
   );

admin/components/run-overview.tsx

@@ -5,7 +5,7 @@ import { Progress } from '@/components/progress';
 import { Status } from '@/components/status';
 import { loadRunProgress } from '@/lib/client';
 import { formatBytes } from '@/lib/format';
-import { progressTitle } from '@/lib/run-progress';
+import { progressTitle, workerNotice } from '@/lib/run-progress';
 import type { RunProgress } from '@/lib/types';
 
 export function RunOverview({
@@ -63,6 +63,7 @@ export function RunOverview({
     (stage) => stage.status === 'pending' && !stage.dependencies.length,
   );
   const title = progressTitle(progress);
+  const workerMessage = workerNotice(progress);
 
   return (
     <section className="run-overview" aria-labelledby="run-progress-title">
@@ -94,8 +95,15 @@ export function RunOverview({
           {progress.result.summary}
         </p>
       ) : null}
-      {progress.handsOff && !progress.finished && !progress.paused ? (
-        <p className="field__hint">Processing automatically. You can leave this page.</p>
+      {workerMessage ? (
+        <p
+          className={
+            progress.worker?.status === 'unavailable' ? 'notice notice--warn' : 'field__hint'
+          }
+          role="status"
+        >
+          {workerMessage}
+        </p>
       ) : null}
       {reused.length ? (
         <p className="field__hint">

admin/lib/run-progress.ts

@@ -14,3 +14,16 @@ export function progressTitle(progress: RunProgress): string {
   if (progress.stopped.length) return 'Processing stopped';
   return 'Waiting for the worker';
 }
+
+export function workerNotice(progress: RunProgress): string | null {
+  if (progress.finished || progress.paused || progress.question) return null;
+  if (!progress.worker) return 'This server has not reported worker availability.';
+  if (progress.worker.status === 'unavailable') {
+    return progress.currentStages.length
+      ? 'No worker is connected for this run. Started steps may still finish; processing will continue when a worker reconnects.'
+      : 'Waiting for a worker. Your upload is saved and will be picked up when a worker connects.';
+  }
+  return progress.handsOff
+    ? 'Worker connected. Processing automatically; you can leave this page.'
+    : null;
+}

admin/lib/types.ts

@@ -226,6 +226,14 @@ export interface SetupReport {
   ready: boolean;
   status: 'okay' | 'blocked' | 'unverified';
   checks: SetupCheck[];
+  worker?: WorkerStatus;
+}
+
+export interface WorkerStatus {
+  status: 'available' | 'unavailable';
+  matchingWorkers: number;
+  busyWorkers: number;
+  detail: string;
 }
 
 export interface StageProgress {
@@ -249,6 +257,7 @@ export interface RunPreview {
 }
 
 export interface RunProgress {
+  worker?: WorkerStatus;
   viewerPath?: string | null;
   runId: string;
   finished: string | null;

admin/tests/run-progress.test.ts

@@ -1,6 +1,6 @@
 import assert from 'node:assert/strict';
 import { test } from 'node:test';
-import { progressTitle } from '../lib/run-progress';
+import { progressTitle, workerNotice } from '../lib/run-progress';
 import type { RunProgress } from '../lib/types';
 
 const progress: RunProgress = {
@@ -39,3 +39,33 @@ test('working and explicitly paused uploads have distinct progress', () => {
   assert.equal(progressTitle({ ...progress, currentStages: ['clean'] }), 'Running clean');
   assert.equal(progressTitle({ ...progress, paused: true }), 'Run paused');
 });
+
+const offline: RunProgress = {
+  ...progress,
+  handsOff: true,
+  worker: { status: 'unavailable', matchingWorkers: 0, busyWorkers: 0, detail: 'No live worker.' },
+};
+
+test('a saved upload with no worker never claims it is processing automatically', () => {
+  assert.match(workerNotice(offline)!, /Waiting for a worker.*upload is saved/);
+  assert.doesNotMatch(workerNotice(offline)!, /Processing automatically/);
+  assert.match(
+    workerNotice({ ...offline, currentStages: ['clean'] })!,
+    /Started steps may still finish/,
+  );
+  assert.equal(progressTitle({ ...offline, currentStages: ['clean'] }), 'Running clean');
+});
+
+test('worker reconnect updates the message without masking terminal or manual holds', () => {
+  const connected: RunProgress = {
+    ...offline,
+    worker: { status: 'available', matchingWorkers: 1, busyWorkers: 1, detail: 'Connected.' },
+  };
+  assert.match(workerNotice(connected)!, /Worker connected.*Processing automatically/);
+  for (const state of [{ finished: 'failed' }, { paused: true }, { question: 'Review output' }]) {
+    assert.equal(workerNotice({ ...offline, ...state }), null);
+    assert.equal(workerNotice({ ...connected, ...state }), null);
+  }
+  assert.equal(workerNotice({ ...connected, handsOff: false }), null);
+  assert.equal(workerNotice(progress), 'This server has not reported worker availability.');
+});

docs/automatic-uploads.md

@@ -13,6 +13,24 @@ its configured runs root. Do not carry a source-specific paid-stage permit into
 unrelated uploads. This change does not provision a hosted worker or turn a local API into a
 public multi-user service.
 
+The upload setup check verifies that a recent supervisor heartbeat exists for the API's runs
+root and accepts new uploads. A worker scoped with `--run-id`, a one-pass worker, or a worker
+watching another directory does not satisfy that check. When no matching worker is connected,
+the form offers **Queue for later** and retains the upload for a worker to pick up. API clients
+may still submit while workers are offline; submission stores a run, not a promise of immediate
+execution.
+
+The run page updates worker availability while processing. Heartbeats refresh every five seconds,
+including during long agent turns, and expire after 45 seconds without an update. A disconnect
+does not fail the run or repeat a provider call; detached stages may still finish. Restarting a
+matching supervisor resumes the existing directory under the ordinary admission checks. Paused,
+canceled and finished runs retain their existing state.
+
+Worker status is a liveness signal, not proof of provider access, spending authorization,
+available GPU capacity, or successful reconstruction. It requires the API and worker to share
+the same runs directory. Older workers without these heartbeat records appear unavailable until
+updated. The legacy `WANDER_HEARTBEAT` timestamp remains supported for process monitoring.
+
 Automatic processing skips the human cleaned-frame pause with the existing `--no-gate`
 option. The agent still inspects outputs and runs the existing reconstruction checks. It cannot
 change the submitted run through extra stage flags or automatically retry a paid stage.

orchestrator/api/app.py

@@ -40,6 +40,7 @@
 from orchestrator.journal import Journal
 from orchestrator.repository import PipelineRepository, TERMINAL_RUN_STATUSES, live_run_status
 from orchestrator.steps import STEPS, RunOptions, base_step
+from orchestrator.worker_presence import worker_status
 
 GRAPH_VERSION = "wander.agent-run/1"
 
@@ -239,7 +240,24 @@ def doctor(options: RunOptions, user: str = Depends(actor)):
         del user
         from orchestrator.doctor import inspect_setup
 
-        return inspect_setup(options.model_dump(mode="json"))
+        report = inspect_setup(options.model_dump(mode="json"))
+        worker = worker_status(review_workspace_root)
+        available = worker["status"] == "available"
+        report["worker"] = worker
+        report["checks"].append(
+            {
+                "name": "worker.watch",
+                "status": "okay" if available else "blocked",
+                "detail": worker["detail"],
+                "action": ""
+                if available
+                else "Connect a worker for new uploads, or queue for later.",
+            }
+        )
+        if not available:
+            report["ready"] = False
+            report["status"] = "blocked"
+        return report
 
     def require_supported_budget(budget: BudgetPolicy) -> None:
         if budget.maximum_attempts_by_stage or budget.maximum_provider_operations:

orchestrator/api/progress.py

@@ -14,6 +14,7 @@
 from orchestrator.journal import Journal
 from orchestrator.steps import STEPS, base_step, needs_of
 from orchestrator.api.viewer import register_viewer_routes, scene_manifest
+from orchestrator.worker_presence import worker_status
 
 
 def read_json(path: Path) -> dict:
@@ -235,6 +236,7 @@ def describe_progress(root: Path) -> dict:
                 }
     return {
         "runId": snapshot.run_id,
+        "worker": worker_status(root.parent, snapshot.run_id),
         "finished": journal.finished(),
         "paused": (root / "paused").exists(),
         "question": journal.unanswered(),

orchestrator/supervisor.py

@@ -23,6 +23,7 @@
 from orchestrator.run_policy import hands_off
 from orchestrator.session import RunSession
 from orchestrator.config import REPOSITORY_ROOT, resolve_runs_root
+from orchestrator.worker_presence import WorkerPresence
 from orchestrator.steps import (
     STEPS,
     base_step,
@@ -178,36 +179,45 @@ def main(argv: list[str] | None = None) -> int:
     # because its agent stopped.
     pushed: dict[str, tuple[float, int]] = {}
 
-    while True:
-        if heartbeat:
-            Path(heartbeat).write_text(f"{time.time()}\n")
-        pending = runs_needing_work(root)
-        if args.run_id:
-            import json
-
-            pending = [
-                d for d in pending if json.loads((d / RUN_FILE).read_text())["runId"] == args.run_id
-            ]
-        for run_dir in pending:
-            # One pass per run at a time. A step runs detached and outlives the turn that
-            # started it, so a run whose journal has not moved is a run whose agent is
-            # waiting on something -- handing it a second agent is how a 1600-credit world
-            # gets submitted twice.
-            if not single_pass and not due(run_dir, pushed, args.push_after):
-                continue
-            outcome = work_on(run_dir, repository, policy, args.turns)
-            print(outcome.message, flush=True)
-            if outcome.resume_next_pass:
-                # Continue useful work on the next pass, still bounded by --turns
-                # and the ordinary admission checks. Do not fabricate an unseen event.
-                pushed.pop(str(run_dir), None)
-            else:
-                # Keep the cursor the session actually observed. A reply can arrive between
-                # its final read and this assignment; reading again here would swallow it.
-                pushed[str(run_dir)] = (time.monotonic(), outcome.observed_entries)
-        if single_pass:
-            return 0
-        time.sleep(args.interval)
+    with WorkerPresence(
+        root,
+        run_id=args.run_id,
+        watcher=not single_pass,
+        heartbeat=Path(heartbeat) if heartbeat else None,
+    ) as presence:
+        while True:
+            pending = runs_needing_work(root)
+            if args.run_id:
+                pending = [
+                    d
+                    for d in pending
+                    if json.loads((d / RUN_FILE).read_text())["runId"] == args.run_id
+                ]
+            for run_dir in pending:
+                # One pass per run at a time. A step runs detached and outlives the turn that
+                # started it, so a run whose journal has not moved is a run whose agent is
+                # waiting on something -- handing it a second agent is how a 1600-credit world
+                # gets submitted twice.
+                if not single_pass and not due(run_dir, pushed, args.push_after):
+                    continue
+                run_id = json.loads((run_dir / RUN_FILE).read_text())["runId"]
+                presence.set_active(run_id)
+                try:
+                    outcome = work_on(run_dir, repository, policy, args.turns)
+                finally:
+                    presence.set_active(None)
+                print(outcome.message, flush=True)
+                if outcome.resume_next_pass:
+                    # Continue useful work on the next pass, still bounded by --turns
+                    # and the ordinary admission checks. Do not fabricate an unseen event.
+                    pushed.pop(str(run_dir), None)
+                else:
+                    # Keep the cursor the session actually observed. A reply can arrive between
+                    # its final read and this assignment; reading again here would swallow it.
+                    pushed[str(run_dir)] = (time.monotonic(), outcome.observed_entries)
+            if single_pass:
+                return 0
+            time.sleep(args.interval)
 
 
 def owner_of(run_dir: Path) -> str | None: