feat: isolate automatic uploads in a dedicated worker

docs/automatic-uploads.md

@@ -13,6 +13,33 @@ 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.
 
+## Worker for automatic uploads
+
+Use a dedicated supervisor for the upload form:
+
+```sh
+uv run --locked --group inference python -m orchestrator.supervisor \
+  --runs /absolute/path/to/shared/runs --uploads-only --watch
+```
+
+Keep that process running alongside the API. It discovers new automatic uploads on each pass,
+including uploads submitted after startup. Manual and legacy runs remain with their existing
+workers; paused and finished runs remain untouched. `--uploads-only --once` makes a single pass
+over eligible uploads and exits. Adding `--run-id` further restricts the worker to that run.
+
+Start the upload worker from its own configuration with the intended provider credentials,
+spending controls, shared `WANDER_STAGE_LEDGER`, and any separately prepared VLM allowance.
+The worker refuses `--uploads-only` when `WANDER_STAGE_AUTHORIZATION` is present, even if empty:
+that setting grants a source-specific exception, which cannot authorize unrelated uploads.
+Do not reuse an exhausted allowance or reset the shared ledger to enable a service. Provider
+spending authorization is still required before enabling paid processing.
+
+Worker presence records include their processing mode. An uploads-only worker is available for
+`hands_off: true` submissions and runs, and unavailable for manual ones. Update the API with
+the worker so it understands that scope; older APIs ignore the new record version instead of
+reporting a worker that cannot process their request. Existing version-one worker records remain
+readable as unrestricted workers by the updated API.
+
 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,

docs/pipeline-setup.md

@@ -179,11 +179,18 @@ The API exposes `GET /api/pipeline/runs/{id}/saved-inputs` for availability and
 ### Submit new work
 
 The admin reads its option defaults and supported world modes from `GET /api/pipeline/options`.
-`POST /api/pipeline/doctor` inspects the API server locally; it does not establish that a separate
-worker is ready. Uploads can be queued while worker setup is repaired. The API and supervisor
+`POST /api/pipeline/doctor` inspects the API server locally and checks for a live worker matching
+the requested automatic/manual mode. Its local dependency and credential checks describe the API
+process, not a separate worker's environment. A heartbeat does not establish provider access or
+spending authorization. Uploads can be queued while worker setup is repaired. The API and supervisor
 must share `WANDER_RUNS_ROOT`. `create_service()` accepts local artifact storage through
 `WANDER_ARTIFACT_ROOT`; importing it no longer requires cloud configuration.
 
+For a dedicated automatic-upload worker, use `orchestrator.supervisor --uploads-only --watch`.
+This leaves manual/legacy runs with their current workers and rejects source-specific exception
+permits. See [automatic uploads](automatic-uploads.md#worker-for-automatic-uploads) for configuration
+and worker-version compatibility.
+
 New runs retain a source SHA-256 in `run.json`. Stages refuse missing or changed source digests
 before launching; existing outputs stay on disk. Legacy manifests without a digest need a new
 submission instead of blindly reusing cached state. Orchestrated runs reject cuts instead of

orchestrator/api/app.py

@@ -241,7 +241,7 @@ def doctor(options: RunOptions, user: str = Depends(actor)):
         from orchestrator.doctor import inspect_setup
 
         report = inspect_setup(options.model_dump(mode="json"))
-        worker = worker_status(review_workspace_root)
+        worker = worker_status(review_workspace_root, hands_off=options.hands_off)
         available = worker["status"] == "available"
         report["worker"] = worker
         report["checks"].append(

orchestrator/api/progress.py

@@ -213,6 +213,7 @@ def describe_progress(root: Path) -> dict:
         item["url"] = prefix + quote(item["path"], safe="/")
     stopped = [row for row in rows if row["status"] in {"failed", "blocked"}]
     current = [row["name"] for row in rows if row["status"] == "running"]
+    hands_off = (read_json(root / "run.json").get("options") or {}).get("hands_off") is True
     viewer_available = scene_manifest(root) is not None
     result = None
     if journal.finished():
@@ -236,15 +237,15 @@ def describe_progress(root: Path) -> dict:
                 }
     return {
         "runId": snapshot.run_id,
-        "worker": worker_status(root.parent, snapshot.run_id),
+        "worker": worker_status(root.parent, snapshot.run_id, hands_off=hands_off),
         "finished": journal.finished(),
         "paused": (root / "paused").exists(),
         "question": journal.unanswered(),
         "currentStages": current,
         "stopped": stopped,
         "stages": rows,
         "previews": inputs,
-        "handsOff": (read_json(root / "run.json").get("options") or {}).get("hands_off") is True,
+        "handsOff": hands_off,
         "result": result,
         "viewerPath": (
             f"/fourd.html?scene={quote(f'/api/pipeline/runs/{snapshot.run_id}/viewer/scene.json', safe='')}"

orchestrator/supervisor.py

@@ -14,6 +14,7 @@
 import argparse
 import json
 import os
+import stat
 import time
 from dataclasses import dataclass
 from pathlib import Path
@@ -26,6 +27,7 @@
 from orchestrator.worker_presence import WorkerPresence
 from orchestrator.steps import (
     STEPS,
+    RunOptions,
     base_step,
     dependencies_within,
     people_steps,
@@ -48,21 +50,72 @@ def repository_from_environment():
     return PipelineRepository(sessionmaker(create_engine(url), expire_on_commit=False))
 
 
-def runs_needing_work(root: Path) -> list[Path]:
-    """Run directories that are neither finished nor waiting on an answer."""
+def run_manifest(run_dir: Path, *, uploads_only: bool = False) -> dict | None:
+    """Read a well-formed run without following a run or manifest symlink."""
+    directory = None
+    try:
+        directory = os.open(run_dir, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW)
+        descriptor = os.open(
+            RUN_FILE, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK, dir_fd=directory
+        )
+        with os.fdopen(descriptor) as manifest_file:
+            if not stat.S_ISREG(os.fstat(manifest_file.fileno()).st_mode):
+                return None
+            manifest = json.load(manifest_file)
+        if not isinstance(manifest, dict):
+            return None
+        run_id = manifest.get("runId")
+        options = manifest.get("options", {})
+        if (
+            not isinstance(run_id, str)
+            or not 0 < len(run_id) <= 512
+            or not isinstance(manifest.get("name"), str)
+            or not manifest["name"]
+            or not isinstance(manifest.get("source"), str)
+            or (manifest.get("goal") is not None and not isinstance(manifest["goal"], str))
+            or (
+                manifest.get("retainedDelivery") is not None
+                and not isinstance(manifest["retainedDelivery"], dict)
+            )
+            or not isinstance(options, dict)
+            or (uploads_only and not hands_off(options))
+        ):
+            return None
+        automatic = hands_off(options)
+        repository = manifest.get("repository")
+        if automatic or manifest.get("retainedDelivery") is not None:
+            if not isinstance(repository, str) or not repository:
+                return None
+        elif repository is not None and not isinstance(repository, str):
+            return None
+        # Ordinary legacy sessions fall back to the current repository and accept
+        # integer strings for people. Automatic manifests require exact typed options.
+        # Check the raw conversion used by RunSession too: Pydantic accepts "2.0",
+        # but int("2.0") cannot be consumed by session planning.
+        RunOptions.model_validate(options, strict=automatic)
+        int(options.get("people") or 1)
+        # Missing source files (including open_run's empty source) remain stage preflight.
+        return manifest
+    except (OSError, ValueError, RecursionError):
+        return None
+    finally:
+        if directory is not None:
+            os.close(directory)
+
+
+def runs_needing_work(root: Path, *, uploads_only: bool = False) -> list[Path]:
+    """Eligible run directories that are neither finished nor waiting on an answer."""
     pending = []
     for described in sorted(root.glob(f"*/{RUN_FILE}")):
         run_dir = described.parent
+        manifest = run_manifest(run_dir, uploads_only=uploads_only)
+        if manifest is None:
+            continue
         journal = Journal(run_dir)
         if journal.finished():
             continue
-        if journal.unanswered():
-            try:
-                unattended = hands_off(json.loads(described.read_text()).get("options") or {})
-            except (OSError, ValueError, AttributeError):
-                unattended = False
-            if not unattended:
-                continue
+        if journal.unanswered() and not hands_off(manifest.get("options", {})):
+            continue
         if journal.control_status():
             continue
         pending.append(run_dir)
@@ -147,6 +200,11 @@ def main(argv: list[str] | None = None) -> int:
     parser.add_argument(
         "--run-id", help="select one run; stop after one pass unless --watch is set"
     )
+    parser.add_argument(
+        "--uploads-only",
+        action="store_true",
+        help="process only runs explicitly submitted with options.hands_off true",
+    )
     parser.add_argument("--turns", type=int, default=12, help="turns per run before moving on")
     mode = parser.add_mutually_exclusive_group()
     mode.add_argument(
@@ -165,6 +223,11 @@ def main(argv: list[str] | None = None) -> int:
         help="seconds before a run that went nowhere is given a fresh agent",
     )
     args = parser.parse_args(argv)
+    if args.uploads_only and "WANDER_STAGE_AUTHORIZATION" in os.environ:
+        parser.error(
+            "--uploads-only cannot inherit WANDER_STAGE_AUTHORIZATION: "
+            "a source-specific paid-stage permit must not be reused across uploads"
+        )
     single_pass = args.once or (args.run_id and not args.watch)
 
     root = resolve_runs_root(args.runs)
@@ -183,26 +246,29 @@ def main(argv: list[str] | None = None) -> int:
         root,
         run_id=args.run_id,
         watcher=not single_pass,
+        uploads_only=args.uploads_only,
         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
-                ]
+            pending = runs_needing_work(root, uploads_only=args.uploads_only)
             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"]
+                manifest = run_manifest(run_dir, uploads_only=args.uploads_only)
+                if manifest is None or (args.run_id and manifest["runId"] != args.run_id):
+                    continue
+                run_id = manifest["runId"]
                 presence.set_active(run_id)
                 try:
+                    # Discovery may precede another run's long turn. Recheck the scope
+                    # immediately before handing this directory to an agent.
+                    current = run_manifest(run_dir, uploads_only=args.uploads_only)
+                    if current is None or current["runId"] != run_id:
+                        continue
                     outcome = work_on(run_dir, repository, policy, args.turns)
                 finally:
                     presence.set_active(None)

orchestrator/worker_presence.py

@@ -13,7 +13,7 @@
 from pathlib import Path
 
 SCHEMA = "wander.worker-presence"
-VERSION = 1
+VERSION = 2
 FRESHNESS_SECONDS = 45.0
 REFRESH_SECONDS = 5.0
 MAX_RECORD_BYTES = 16_384
@@ -28,7 +28,7 @@ def _open_directory(root: Path) -> int:
     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:
+def worker_status(root: Path | None, run_id: str | None = None, *, hands_off: bool = False) -> dict:
     """Count fresh matching supervisors without disclosing local process information."""
     matching = busy = 0
     if root is not None:
@@ -57,7 +57,11 @@ def worker_status(root: Path | None, run_id: str | None = None) -> dict:
                     if (
                         record.get("schema") != SCHEMA
                         or type(record.get("version")) is not int
-                        or record["version"] != VERSION
+                        or record["version"] not in (1, VERSION)
+                        or (
+                            record["version"] == VERSION
+                            and type(record.get("uploads_only")) is not bool
+                        )
                         or record.get("runs_root") != str(resolved)
                         or type(record.get("watcher")) is not bool
                         or "run_id" not in record
@@ -69,6 +73,14 @@ def worker_status(root: Path | None, run_id: str | None = None) -> dict:
                         or not 0 <= time.time() - seen <= FRESHNESS_SECONDS
                     ):
                         continue
+                    # Version 1 predates restricted workers. Version 2 is deliberately
+                    # incompatible with old readers, which only accepted version 1.
+                    if (
+                        record["version"] == VERSION
+                        and record["uploads_only"]
+                        and hands_off is not True
+                    ):
+                        continue
                     selected, active = record["run_id"], record["active_run_id"]
                     if selected is not None and active not in (None, selected):
                         continue
@@ -109,16 +121,18 @@ def __init__(
         *,
         run_id: str | None = None,
         watcher: bool = True,
+        uploads_only: bool = False,
         heartbeat: Path | None = None,
         refresh_interval: float = REFRESH_SECONDS,
     ):
-        if not _run_id(run_id) or type(watcher) is not bool:
+        if not _run_id(run_id) or type(watcher) is not bool or type(uploads_only) 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.uploads_only = uploads_only
         self.heartbeat = heartbeat
         self.refresh_interval = refresh_interval
         self.name = f"{uuid.uuid4()}.json"
@@ -177,6 +191,7 @@ def _write(self) -> None:
             "version": VERSION,
             "runs_root": str(self.root),
             "watcher": self.watcher,
+            "uploads_only": self.uploads_only,
             "run_id": self.run_id,
             "active_run_id": self._active,
             "last_seen": now,