diff --git a/admin/components/create-run.tsx b/admin/components/create-run.tsx
index 10f3eae..0f1f726 100644
--- a/admin/components/create-run.tsx
+++ b/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>
   );
diff --git a/admin/components/run-overview.tsx b/admin/components/run-overview.tsx
index 1febf41..311c17a 100644
--- a/admin/components/run-overview.tsx
+++ b/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">
diff --git a/admin/lib/run-progress.ts b/admin/lib/run-progress.ts
index 1444977..9e640a0 100644
--- a/admin/lib/run-progress.ts
+++ b/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;
+}
diff --git a/admin/lib/types.ts b/admin/lib/types.ts
index 7e8eb6f..74a97eb 100644
--- a/admin/lib/types.ts
+++ b/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;
diff --git a/admin/tests/run-progress.test.ts b/admin/tests/run-progress.test.ts
index 9898805..6711e61 100644
--- a/admin/tests/run-progress.test.ts
+++ b/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.');
+});
diff --git a/docs/automatic-uploads.md b/docs/automatic-uploads.md
index 055fb82..4c4cd8c 100644
--- a/docs/automatic-uploads.md
+++ b/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.
diff --git a/orchestrator/api/app.py b/orchestrator/api/app.py
index d9be46f..eca54cb 100644
--- a/orchestrator/api/app.py
+++ b/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:
diff --git a/orchestrator/api/progress.py b/orchestrator/api/progress.py
index d573f1a..586afe2 100644
--- a/orchestrator/api/progress.py
+++ b/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(),
diff --git a/orchestrator/supervisor.py b/orchestrator/supervisor.py
index 40ba36e..5bb69ef 100644
--- a/orchestrator/supervisor.py
+++ b/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:
diff --git a/orchestrator/worker_presence.py b/orchestrator/worker_presence.py
new file mode 100644
index 0000000..79fc20e
--- /dev/null
+++ b/orchestrator/worker_presence.py
@@ -0,0 +1,226 @@
+"""Bounded, local supervisor liveness; presence does not promise successful work."""
+
+from __future__ import annotations
+
+import json
+import logging
+import math
+import os
+import stat
+import threading
+import time
+import uuid
+from pathlib import Path
+
+SCHEMA = "wander.worker-presence"
+VERSION = 1
+FRESHNESS_SECONDS = 45.0
+REFRESH_SECONDS = 5.0
+MAX_RECORD_BYTES = 16_384
+
+
+def _run_id(value: object) -> bool:
+    return value is None or (isinstance(value, str) and 0 < len(value) <= 512)
+
+
+def _open_directory(root: Path) -> int:
+    # A symlink must not move registration or discovery outside this runs root.
+    return os.open(root / ".workers", os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW)
+
+
+def worker_status(root: Path | None, run_id: str | None = None) -> dict:
+    """Count fresh matching supervisors without disclosing local process information."""
+    matching = busy = 0
+    if root is not None:
+        directory = None
+        try:
+            resolved = root.resolve()
+            directory = _open_directory(resolved)
+            for name in os.listdir(directory):
+                try:
+                    if not name.endswith(".json") or str(uuid.UUID(name[:-5])) != name[:-5]:
+                        continue
+                    descriptor = os.open(
+                        name, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK, dir_fd=directory
+                    )
+                    with os.fdopen(descriptor, "rb") as record_file:
+                        metadata = os.fstat(record_file.fileno())
+                        if (
+                            not stat.S_ISREG(metadata.st_mode)
+                            or metadata.st_size > MAX_RECORD_BYTES
+                        ):
+                            continue
+                        record = json.loads(record_file.read(MAX_RECORD_BYTES + 1))
+                    if not isinstance(record, dict):
+                        continue
+                    seen = record.get("last_seen")
+                    if (
+                        record.get("schema") != SCHEMA
+                        or type(record.get("version")) is not int
+                        or record["version"] != VERSION
+                        or record.get("runs_root") != str(resolved)
+                        or type(record.get("watcher")) is not bool
+                        or "run_id" not in record
+                        or not _run_id(record["run_id"])
+                        or "active_run_id" not in record
+                        or not _run_id(record["active_run_id"])
+                        or type(seen) not in (int, float)
+                        or not math.isfinite(seen)
+                        or not 0 <= time.time() - seen <= FRESHNESS_SECONDS
+                    ):
+                        continue
+                    selected, active = record["run_id"], record["active_run_id"]
+                    if selected is not None and active not in (None, selected):
+                        continue
+                    if run_id is None:
+                        matches = record["watcher"] and selected is None
+                    elif record["watcher"]:
+                        matches = selected in (None, run_id)
+                    else:
+                        matches = active == run_id
+                    if matches:
+                        matching += 1
+                        busy += active is not None
+                except (OSError, ValueError, TypeError, OverflowError, RecursionError):
+                    continue
+        except (OSError, ValueError, RuntimeError):
+            pass
+        finally:
+            if directory is not None:
+                os.close(directory)
+    return {
+        "status": "available" if matching else "unavailable",
+        "matchingWorkers": matching,
+        "busyWorkers": busy,
+        "detail": (
+            "A matching supervisor has a recent heartbeat."
+            if matching
+            else "No matching supervisor has a recent heartbeat."
+        ),
+    }
+
+
+class WorkerPresence:
+    """Keep one independent record fresh even while a supervisor's agent blocks."""
+
+    def __init__(
+        self,
+        root: Path,
+        *,
+        run_id: str | None = None,
+        watcher: bool = True,
+        heartbeat: Path | None = None,
+        refresh_interval: float = REFRESH_SECONDS,
+    ):
+        if not _run_id(run_id) or type(watcher) is not bool:
+            raise ValueError("Invalid supervisor scope")
+        if not math.isfinite(refresh_interval) or not 0 < refresh_interval < FRESHNESS_SECONDS:
+            raise ValueError("Heartbeat interval must be positive and below the freshness limit")
+        self.root = root.resolve()
+        self.run_id = run_id
+        self.watcher = watcher
+        self.heartbeat = heartbeat
+        self.refresh_interval = refresh_interval
+        self.name = f"{uuid.uuid4()}.json"
+        self._active: str | None = None
+        self._directory: int | None = None
+        self._thread: threading.Thread | None = None
+        self._stop = threading.Event()
+        self._lock = threading.Lock()
+        self._error: OSError | None = None
+
+    def __enter__(self) -> WorkerPresence:
+        try:
+            (self.root / ".workers").mkdir(exist_ok=True)
+            self._directory = _open_directory(self.root)
+            self._write()
+            self._thread = threading.Thread(
+                target=self._refresh, name="worker-heartbeat", daemon=True
+            )
+            self._thread.start()
+        except (OSError, RuntimeError) as error:
+            self._remove()
+            self._close_directory()
+            raise RuntimeError("Could not register supervisor heartbeat") from error
+        return self
+
+    def __exit__(self, exc_type, exc_value, traceback):
+        self._stop.set()
+        if self._thread is not None:
+            self._thread.join()
+        self._remove()
+        self._close_directory()
+        if self._error is not None and exc_type is None:
+            raise RuntimeError("Supervisor heartbeat stopped refreshing") from self._error
+
+    def set_active(self, run_id: str | None) -> None:
+        if not _run_id(run_id) or (self.run_id is not None and run_id not in (None, self.run_id)):
+            raise ValueError("Active run does not match supervisor scope")
+        with self._lock:
+            if self._error is not None:
+                raise RuntimeError("Supervisor heartbeat stopped refreshing") from self._error
+            self._active = run_id
+            try:
+                self._write()
+            except OSError as error:
+                self._error = error
+                self._stop.set()
+                self._remove()
+                raise RuntimeError("Supervisor heartbeat stopped refreshing") from error
+
+    def _write(self) -> None:
+        if self._directory is None:
+            raise RuntimeError("Supervisor heartbeat is not registered")
+        now = time.time()
+        record = {
+            "schema": SCHEMA,
+            "version": VERSION,
+            "runs_root": str(self.root),
+            "watcher": self.watcher,
+            "run_id": self.run_id,
+            "active_run_id": self._active,
+            "last_seen": now,
+        }
+        # Preserve the legacy timestamp format and refresh it on the same background timer.
+        if self.heartbeat is not None:
+            self.heartbeat.write_text(f"{now}\n")
+        temporary = f".{self.name}.{uuid.uuid4()}.tmp"
+        try:
+            descriptor = os.open(
+                temporary,
+                os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW,
+                0o600,
+                dir_fd=self._directory,
+            )
+            with os.fdopen(descriptor, "w") as record_file:
+                json.dump(record, record_file)
+                record_file.write("\n")
+            os.replace(temporary, self.name, src_dir_fd=self._directory, dst_dir_fd=self._directory)
+        finally:
+            try:
+                os.unlink(temporary, dir_fd=self._directory)
+            except FileNotFoundError:
+                pass
+
+    def _refresh(self) -> None:
+        while not self._stop.wait(self.refresh_interval):
+            with self._lock:
+                try:
+                    self._write()
+                except OSError as error:
+                    self._error = error
+                    self._remove()
+                    logging.getLogger(__name__).error("Supervisor heartbeat stopped refreshing")
+                    return
+
+    def _remove(self) -> None:
+        if self._directory is not None:
+            try:
+                os.unlink(self.name, dir_fd=self._directory)
+            except OSError:
+                pass  # Expiry bounds stale presence if cleanup is unavailable.
+
+    def _close_directory(self) -> None:
+        if self._directory is not None:
+            os.close(self._directory)
+            self._directory = None
diff --git a/scripts/tests/test_orchestrator_api_submission.py b/scripts/tests/test_orchestrator_api_submission.py
index bc3504f..0313326 100644
--- a/scripts/tests/test_orchestrator_api_submission.py
+++ b/scripts/tests/test_orchestrator_api_submission.py
@@ -23,6 +23,7 @@
 from orchestrator.database import ArtifactRecord, Base, EventRecord, NodeRecord, RunRecord
 from orchestrator.repository import PipelineRepository
 from orchestrator.supervisor import runs_needing_work
+from orchestrator.worker_presence import WorkerPresence
 
 TOKEN = "test-token-with-at-least-24-characters"
 AUTH = {"Authorization": f"Bearer {TOKEN}"}
@@ -99,6 +100,70 @@ def test_hands_off_upload_is_persisted_and_immediately_discoverable(self):
         self.assertIs(summary["configuration"]["options"]["hands_off"], True)
         self.assertFalse((manifest.parent / "paused").exists())
 
+    def test_worker_presence_tracks_upload_scope_and_disconnect_without_starting_work(self):
+        self.assertEqual(self.upload(options='{"marble":"none","hands_off":true}').status_code, 201)
+
+        def progress():
+            response = self.client.get("/api/pipeline/runs/run-upload/progress", headers=AUTH)
+            self.assertEqual(response.status_code, 200, response.text)
+            return response.json()
+
+        self.assertEqual(progress()["worker"]["status"], "unavailable")
+        runs_root = self.root / "runs"
+        with WorkerPresence(runs_root, watcher=True, run_id="another-run"):
+            self.assertEqual(progress()["worker"]["status"], "unavailable")
+            with WorkerPresence(runs_root, watcher=True, run_id="run-upload") as worker:
+                worker.set_active("run-upload")
+                self.assertEqual(progress()["worker"]["matchingWorkers"], 1)
+                self.assertEqual(progress()["worker"]["busyWorkers"], 1)
+        disconnected = progress()
+        self.assertEqual(disconnected["worker"]["status"], "unavailable")
+        self.assertTrue(disconnected["handsOff"])
+        self.assertIsNone(disconnected["finished"])
+        self.assertFalse(disconnected["currentStages"])
+        # Presence reporting itself does not execute, mutate, fail, or retry the queued run.
+        self.assertIn(self.manifests()[0].parent, runs_needing_work(runs_root))
+        self.assertEqual(self.repository.run_summary("run-upload")["attempts"], [])
+
+    def test_upload_setup_requires_a_watcher_for_this_root_and_new_uploads(self):
+        def check():
+            with patch(
+                "orchestrator.doctor.inspect_setup",
+                return_value={"ready": True, "status": "okay", "checks": []},
+            ):
+                response = self.client.post("/api/pipeline/doctor", headers=AUTH, json={})
+            self.assertEqual(response.status_code, 200, response.text)
+            return response.json()
+
+        self.assertEqual(check()["status"], "blocked")
+        runs_root = self.root / "runs"
+        runs_root.mkdir()
+        (self.root / "other-runs").mkdir()
+        with WorkerPresence(runs_root, watcher=True, run_id="existing-run"):
+            self.assertFalse(check()["ready"])
+        with WorkerPresence(self.root / "other-runs", watcher=True):
+            self.assertFalse(check()["ready"])
+        with WorkerPresence(runs_root, watcher=False):
+            self.assertFalse(check()["ready"])
+        with WorkerPresence(runs_root, watcher=True):
+            report = check()
+            self.assertTrue(report["ready"])
+            self.assertEqual(report["checks"][-1]["name"], "worker.watch")
+            self.assertEqual(report["worker"]["status"], "available")
+            # A worker cannot turn failed dependency checks into a ready setup.
+            with patch(
+                "orchestrator.doctor.inspect_setup",
+                return_value={
+                    "ready": False,
+                    "status": "blocked",
+                    "checks": [{"name": "ffmpeg", "status": "blocked", "detail": "Missing."}],
+                },
+            ):
+                response = self.client.post("/api/pipeline/doctor", headers=AUTH, json={})
+            self.assertFalse(response.json()["ready"])
+            self.assertEqual(response.json()["status"], "blocked")
+        self.assertEqual(check()["worker"]["status"], "unavailable")
+
     def test_clean_candidates_are_previewable_without_exposing_other_candidate_files(self):
         response = self.upload("clean-recovery")
         self.assertEqual(response.status_code, 201, response.text)
diff --git a/scripts/tests/test_worker_presence.py b/scripts/tests/test_worker_presence.py
new file mode 100644
index 0000000..d5ac92a
--- /dev/null
+++ b/scripts/tests/test_worker_presence.py
@@ -0,0 +1,258 @@
+"""Supervisor liveness is fresh, scoped, and independent of blocking agent work."""
+
+import json
+import os
+import threading
+import time
+import uuid
+
+import pytest
+
+from orchestrator import supervisor
+from orchestrator.agent.harness import HarnessPolicy
+from orchestrator.worker_presence import FRESHNESS_SECONDS, WorkerPresence, worker_status
+
+
+def test_missing_root_and_empty_directory_are_unavailable(tmp_path):
+    for root in (None, tmp_path, tmp_path / "missing"):
+        assert worker_status(root) == {
+            "status": "unavailable",
+            "matchingWorkers": 0,
+            "busyWorkers": 0,
+            "detail": "No matching supervisor has a recent heartbeat.",
+        }
+
+
+def test_independent_workers_count_matching_and_busy_and_cleanup(tmp_path):
+    with WorkerPresence(tmp_path) as broad:
+        with WorkerPresence(tmp_path, run_id="selected") as selected:
+            broad.set_active("other")
+            selected.set_active("selected")
+            assert worker_status(tmp_path)["matchingWorkers"] == 1
+            assert worker_status(tmp_path)["busyWorkers"] == 1
+            assert worker_status(tmp_path, "selected")["matchingWorkers"] == 2
+            assert worker_status(tmp_path, "selected")["busyWorkers"] == 2
+            selected.set_active(None)
+            assert worker_status(tmp_path, "selected")["busyWorkers"] == 1
+        assert worker_status(tmp_path, "selected")["matchingWorkers"] == 1
+    assert worker_status(tmp_path)["status"] == "unavailable"
+    assert list((tmp_path / ".workers").iterdir()) == []
+
+
+@pytest.mark.parametrize("selected", [None, "selected"])
+def test_one_pass_matches_only_its_active_run(tmp_path, selected):
+    with WorkerPresence(tmp_path, run_id=selected, watcher=False) as presence:
+        assert worker_status(tmp_path)["status"] == "unavailable"
+        assert worker_status(tmp_path, "selected")["status"] == "unavailable"
+        presence.set_active("selected")
+        assert worker_status(tmp_path)["status"] == "unavailable"
+        assert worker_status(tmp_path, "other")["status"] == "unavailable"
+        assert worker_status(tmp_path, "selected")["matchingWorkers"] == 1
+        presence.set_active(None)
+        assert worker_status(tmp_path, "selected")["status"] == "unavailable"
+
+
+def test_selected_watcher_never_advertises_new_uploads(tmp_path):
+    with WorkerPresence(tmp_path, run_id="selected"):
+        assert worker_status(tmp_path)["status"] == "unavailable"
+        assert worker_status(tmp_path, "other")["status"] == "unavailable"
+        assert worker_status(tmp_path, "selected")["status"] == "available"
+
+
+@pytest.mark.parametrize(
+    "mutation",
+    [
+        {"last_seen": time.time() - FRESHNESS_SECONDS - 5},
+        {"last_seen": time.time() + 3600},
+        {"last_seen": float("nan")},
+        {"last_seen": float("inf")},
+        {"last_seen": True},
+        {"last_seen": "now"},
+        {"version": 2},
+        {"version": True},
+        {"schema": "other"},
+        {"watcher": "true"},
+        {"runs_root": "/elsewhere"},
+        {"run_id": []},
+        {"active_run_id": {}},
+        {"run_id": "selected", "active_run_id": "different"},
+    ],
+)
+def test_invalid_records_are_ignored(tmp_path, mutation):
+    with WorkerPresence(tmp_path) as presence:
+        path = tmp_path / ".workers" / presence.name
+        record = json.loads(path.read_text())
+        record.update(mutation)
+        path.write_text(json.dumps(record))
+        assert worker_status(tmp_path)["status"] == "unavailable"
+
+
+@pytest.mark.parametrize("contents", ["{bad", "[]", "null", "{}", '"text"', "[" * 1500])
+def test_malformed_records_are_ignored(tmp_path, contents):
+    with WorkerPresence(tmp_path) as presence:
+        (tmp_path / ".workers" / presence.name).write_text(contents)
+        assert worker_status(tmp_path)["status"] == "unavailable"
+
+
+def test_escaping_and_symlink_records_are_ignored(tmp_path):
+    root = tmp_path / "runs"
+    root.mkdir()
+    with WorkerPresence(root) as presence:
+        path = root / ".workers" / presence.name
+        original = path.read_text()
+        path.unlink()
+        outside = tmp_path / "outside.json"
+        outside.write_text(original)
+        path.symlink_to(outside)
+        assert worker_status(root)["status"] == "unavailable"
+        path.unlink()
+        record = json.loads(original)
+        record["runs_root"] = str(root / ".." / "runs")
+        path.write_text(json.dumps(record))
+        assert worker_status(root)["status"] == "unavailable"
+    (root / ".workers").rmdir()
+    outside_directory = tmp_path / "outside"
+    outside_directory.mkdir()
+    (outside_directory / f"{uuid.uuid4()}.json").write_text(original)
+    (root / ".workers").symlink_to(outside_directory, target_is_directory=True)
+    assert worker_status(root)["status"] == "unavailable"
+    with pytest.raises(RuntimeError, match="Could not register"):
+        with WorkerPresence(root):
+            pytest.fail("registration through a symlink must fail")
+
+
+def test_nonregular_oversized_and_temporary_records_are_ignored(tmp_path):
+    with WorkerPresence(tmp_path) as presence:
+        directory = tmp_path / ".workers"
+        path = directory / presence.name
+        original = path.read_text()
+        path.unlink()
+        (directory / "invalid-name.json").write_text(original)
+        (directory / f".{uuid.uuid4()}.tmp").write_text(original)
+        (directory / f"{uuid.uuid4()}.json").write_text(" " * 17000 + original)
+        (directory / f"{uuid.uuid4()}.json").mkdir()
+        os.mkfifo(directory / f"{uuid.uuid4()}.json")
+        assert worker_status(tmp_path)["status"] == "unavailable"
+
+
+def test_exception_removes_presence(tmp_path):
+    with pytest.raises(ValueError, match="work failed"):
+        with WorkerPresence(tmp_path):
+            raise ValueError("work failed")
+    assert list((tmp_path / ".workers").iterdir()) == []
+
+
+def test_registration_failure_does_not_advertise_worker(tmp_path):
+    with pytest.raises(RuntimeError, match="Could not register"):
+        with WorkerPresence(tmp_path, heartbeat=tmp_path / "missing" / "heartbeat"):
+            pytest.fail("failed heartbeat must not enter work")
+    assert worker_status(tmp_path)["status"] == "unavailable"
+    assert list((tmp_path / ".workers").iterdir()) == []
+
+
+def test_background_failure_removes_presence_and_surfaces_error(tmp_path, monkeypatch):
+    failed = threading.Event()
+    presence = WorkerPresence(tmp_path, refresh_interval=0.01)
+    with pytest.raises(RuntimeError, match="stopped refreshing"):
+        with presence:
+
+            def fail_write():
+                failed.set()
+                raise OSError("disk unavailable")
+
+            monkeypatch.setattr(presence, "_write", fail_write)
+            assert failed.wait(2)
+            # Synchronize with the thread so cleanup has completed after the failing write.
+            with presence._lock:
+                assert worker_status(tmp_path)["status"] == "unavailable"
+    assert list((tmp_path / ".workers").iterdir()) == []
+
+
+@pytest.mark.parametrize(
+    ("flags", "generic_available", "run_available"),
+    [
+        ([], True, True),
+        (["--once"], False, True),
+        (["--run-id", "selected"], False, True),
+        (["--run-id", "selected", "--watch"], False, True),
+    ],
+)
+def test_supervisor_refreshes_during_blocking_work_and_cleans_up(
+    tmp_path, monkeypatch, flags, generic_available, run_available
+):
+    supervisor.open_run(tmp_path, run_id="selected", name="clip", source=None)
+    heartbeat = tmp_path / "legacy-heartbeat"
+    monkeypatch.setenv("WANDER_HEARTBEAT", str(heartbeat))
+    monkeypatch.setattr(supervisor, "repository_from_environment", lambda: None)
+    monkeypatch.setattr(
+        HarnessPolicy,
+        "from_environment",
+        lambda: HarnessPolicy(kind="command", command=("unused",)),
+    )
+    refreshed = threading.Event()
+    baseline = None
+    instances = []
+
+    class ObservedPresence(WorkerPresence):
+        def __init__(self, *args, **kwargs):
+            super().__init__(*args, **kwargs, refresh_interval=0.01)
+            instances.append(self)
+
+        def _write(self):
+            super()._write()
+            if baseline is not None and threading.current_thread() is self._thread:
+                refreshed.set()
+
+    monkeypatch.setattr(supervisor, "WorkerPresence", ObservedPresence)
+
+    def work(*_):
+        nonlocal baseline
+        assert (worker_status(tmp_path)["status"] == "available") == generic_available
+        assert (worker_status(tmp_path, "selected")["status"] == "available") == run_available
+        assert worker_status(tmp_path, "selected")["busyWorkers"] == 1
+        path = next((tmp_path / ".workers").glob("*.json"))
+        baseline = json.loads(path.read_text())["last_seen"]
+        # The main supervisor is blocked here; only its background heartbeat can advance.
+        assert refreshed.wait(2)
+        with instances[0]._lock:
+            current = json.loads(path.read_text())["last_seen"]
+            assert current > baseline
+            assert float(heartbeat.read_text()) == current
+            assert heartbeat.read_text().endswith("\n")
+        return supervisor.WorkOutcome("done", 0)
+
+    class StopWatch(Exception):
+        pass
+
+    def stop_watch(_):
+        assert worker_status(tmp_path, "selected")["busyWorkers"] == 0
+        raise StopWatch
+
+    monkeypatch.setattr(supervisor, "work_on", work)
+    monkeypatch.setattr(supervisor.time, "sleep", stop_watch)
+    once = "--once" in flags or ("--run-id" in flags and "--watch" not in flags)
+    if once:
+        assert supervisor.main(["--runs", str(tmp_path), *flags]) == 0
+    else:
+        with pytest.raises(StopWatch):
+            supervisor.main(["--runs", str(tmp_path), *flags])
+    assert list((tmp_path / ".workers").iterdir()) == []
+
+
+def test_supervisor_work_failure_cleans_up(tmp_path, monkeypatch):
+    supervisor.open_run(tmp_path, run_id="failed", name="clip", source=None)
+    monkeypatch.setattr(supervisor, "repository_from_environment", lambda: None)
+    monkeypatch.setattr(
+        HarnessPolicy,
+        "from_environment",
+        lambda: HarnessPolicy(kind="command", command=("unused",)),
+    )
+
+    def fail_work(*_):
+        assert worker_status(tmp_path, "failed")["busyWorkers"] == 1
+        raise ValueError("work failed")
+
+    monkeypatch.setattr(supervisor, "work_on", fail_work)
+    with pytest.raises(ValueError, match="work failed"):
+        supervisor.main(["--runs", str(tmp_path), "--once"])
+    assert list((tmp_path / ".workers").iterdir()) == []