feat: persist a finite automatic upload allowance

admin/components/run-overview.tsx

@@ -143,7 +143,11 @@ export function RunOverview({
             </article>
           ))}
         </div>
-      ) : ready && !progress.finished && !progress.paused && !progress.question ? (
+      ) : ready &&
+        !progress.finished &&
+        !progress.paused &&
+        !progress.question &&
+        !progress.worker?.reason ? (
         <p className="field__hint">
           Next: {ready.name}. {ready.nextAction}
         </p>

admin/lib/run-progress.ts

@@ -12,13 +12,24 @@ export function progressTitle(progress: RunProgress): string {
   if (progress.question) return 'Waiting for your answer';
   if (progress.currentStages.length) return `Running ${progress.currentStages.join(', ')}`;
   if (progress.stopped.length) return 'Processing stopped';
+  if (progress.worker?.status === 'unavailable') {
+    if (progress.worker.reason === 'admission_exhausted')
+      return 'Upload saved — processing limit reached';
+    if (progress.worker.reason === 'admission_busy') return 'Upload queued';
+  }
   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') {
+    if (progress.worker.reason === 'admission_exhausted') {
+      return 'Your upload is saved. The configured upload limit has been reached; an operator must arrange more processing capacity.';
+    }
+    if (progress.worker.reason === 'admission_busy') {
+      return 'Your upload is saved. The worker is handling another upload; this one remains queued.';
+    }
     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.';

admin/lib/types.ts

@@ -231,6 +231,7 @@ export interface SetupReport {
 
 export interface WorkerStatus {
   status: 'available' | 'unavailable';
+  reason?: 'admission_busy' | 'admission_exhausted';
   matchingWorkers: number;
   busyWorkers: number;
   detail: string;

admin/tests/run-progress.test.ts

@@ -69,3 +69,26 @@ test('worker reconnect updates the message without masking terminal or manual ho
   assert.equal(workerNotice({ ...connected, handsOff: false }), null);
   assert.equal(workerNotice(progress), 'This server has not reported worker availability.');
 });
+
+test('admission limits keep queued uploads distinct from disconnected workers', () => {
+  const exhausted: RunProgress = {
+    ...offline,
+    worker: { ...offline.worker!, reason: 'admission_exhausted' },
+  };
+  const busy: RunProgress = {
+    ...offline,
+    worker: { ...offline.worker!, reason: 'admission_busy' },
+  };
+  assert.equal(progressTitle(exhausted), 'Upload saved — processing limit reached');
+  assert.match(workerNotice(exhausted)!, /upload is saved.*limit has been reached.*operator/);
+  assert.doesNotMatch(workerNotice(exhausted)!, /connect|Processing automatically/);
+  assert.equal(progressTitle(busy), 'Upload queued');
+  assert.match(workerNotice(busy)!, /worker is handling another upload.*remains queued/);
+  for (const limited of [exhausted, busy]) {
+    assert.equal(progressTitle({ ...limited, currentStages: ['clean'] }), 'Running clean');
+    assert.equal(progressTitle({ ...limited, paused: true }), 'Run paused');
+    assert.equal(progressTitle({ ...limited, finished: 'canceled' }), 'Run canceled');
+    assert.equal(workerNotice({ ...limited, finished: 'failed' }), null);
+    assert.equal(workerNotice({ ...limited, paused: true }), null);
+  }
+});

docs/automatic-uploads.md

@@ -34,11 +34,44 @@ that setting grants a source-specific exception, which cannot authorize unrelate
 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.
 
+### Limit the number of uploads admitted
+
+For a finite trial, add `--max-uploads 1`:
+
+```sh
+uv run --locked --group inference python -m orchestrator.supervisor \
+  --runs /absolute/path/to/shared/runs --uploads-only --watch --max-uploads 1
+```
+
+This persistently limits how many distinct uploads this runs root can admit. An admission is
+recorded before starting its agent. Restarting the supervisor uses the same history; failures
+and cancellation do not refund a slot. A different limit is refused. Keep the runs root and its
+admission history together; deleting or replacing that history is not a supported reset.
+
+The bounded worker processes one admitted upload at a time. A paused upload retains its place.
+A finished upload retains its place while detached stages remain unresolved. The next upload
+can start only after that work settles and if unused admission slots remain. Missing or changed
+run identity and malformed history stop admission instead of freeing a slot. Existing automatic
+work that already started cannot silently be adopted into a new allowance.
+
+While occupied, the worker reports availability only for its admitted run. Once its allowance
+is exhausted, it no longer advertises availability for new uploads. The API can still store new
+uploads with **Queue for later**; storing a video does not consume an admission or start inference.
+Updated supervisors sharing this root respect its persisted policy even if they omit the limit;
+they cannot process automatic uploads outside the bounded worker. Update all supervisors before
+using this control. Manual work remains separate.
+
+This is an upload admission count, not a dollar ceiling, provider-operation allowance, or total
+runtime limit. It covers these supervisors; direct runner, `wander`, and provider invocations
+remain outside it. Provider spending controls, the shared stage ledger, and the intended VLM
+request allowance remain required. A one-upload allowance does not authorize a new paid run.
+
 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.
+`hands_off: true` submissions and runs, and unavailable for manual ones. On a bounded root, a
+generic worker continues serving manual runs and reports that restricted capability. Update the
+API with the worker so it understands that scope; older APIs ignore version-three records instead
+of reporting a worker that cannot process their request. The updated reader still accepts
+version-one unrestricted workers and version-two workers with automatic-upload scope.
 
 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

orchestrator/api/app.py

@@ -243,15 +243,18 @@ def doctor(options: RunOptions, user: str = Depends(actor)):
         report = inspect_setup(options.model_dump(mode="json"))
         worker = worker_status(review_workspace_root, hands_off=options.hands_off)
         available = worker["status"] == "available"
+        action = "Connect a worker for new uploads, or queue for later."
+        if worker.get("reason") == "admission_exhausted":
+            action = "Queue for later; an operator must arrange more processing capacity."
+        elif worker.get("reason") == "admission_busy":
+            action = "Queue for later while the worker handles its admitted upload."
         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.",
+                "action": "" if available else action,
             }
         )
         if not available:

orchestrator/supervisor.py

@@ -25,6 +25,7 @@
 from orchestrator.session import RunSession
 from orchestrator.config import REPOSITORY_ROOT, resolve_runs_root
 from orchestrator.worker_presence import WorkerPresence
+from orchestrator.upload_admission import AdmissionError, UploadAdmission
 from orchestrator.steps import (
     STEPS,
     RunOptions,
@@ -205,6 +206,11 @@ def main(argv: list[str] | None = None) -> int:
         action="store_true",
         help="process only runs explicitly submitted with options.hands_off true",
     )
+    parser.add_argument(
+        "--max-uploads",
+        type=int,
+        help="persist a cumulative positive upload admission limit for this runs root",
+    )
     parser.add_argument("--turns", type=int, default=12, help="turns per run before moving on")
     mode = parser.add_mutually_exclusive_group()
     mode.add_argument(
@@ -228,58 +234,100 @@ def main(argv: list[str] | None = None) -> int:
             "--uploads-only cannot inherit WANDER_STAGE_AUTHORIZATION: "
             "a source-specific paid-stage permit must not be reused across uploads"
         )
+    if args.max_uploads is not None and (args.max_uploads <= 0 or not args.uploads_only):
+        parser.error("--max-uploads must be positive and requires --uploads-only")
     single_pass = args.once or (args.run_id and not args.watch)
 
     root = resolve_runs_root(args.runs)
     root.mkdir(parents=True, exist_ok=True)
+    try:
+        with UploadAdmission(root, args.max_uploads, uploads_only=args.uploads_only) as admission:
+            return supervise(args, root, bool(single_pass), admission)
+    except (AdmissionError, OSError) as error:
+        parser.error(f"Upload admission stopped: {error}")
+
+
+def supervise(args, root: Path, single_pass: bool, admission: UploadAdmission) -> int:
     repository = repository_from_environment()
     policy = HarnessPolicy.from_environment()
     print(f"supervising {root} with {policy.kind}:{policy.model or 'default'}", flush=True)
-
     heartbeat = os.environ.get("WANDER_HEARTBEAT")
-    # When a run was last given an agent, and how long its journal was then. `due` reads this
-    # to tell a run that is quiet because its work is still running from one that is quiet
-    # because its agent stopped.
     pushed: dict[str, tuple[float, int]] = {}
 
+    def scope():
+        if args.max_uploads is not None:
+            active = next(
+                (
+                    item["runId"]
+                    for item in admission.state["admissions"]
+                    if item["directory"] == admission.active
+                ),
+                None,
+            )
+            if active is not None and args.run_id is not None and active != args.run_id:
+                raise AdmissionError("--run-id conflicts with the active upload admission")
+            return {
+                "run_id": active or args.run_id,
+                "watcher": not single_pass and (active is not None or bool(admission.remaining)),
+                "active_run_id": active,
+                "admission": "occupied"
+                if active is not None
+                else ("exhausted" if not admission.remaining else None),
+            }
+        return {
+            "run_id": args.run_id,
+            "watcher": not single_pass,
+            "manual_only": admission.state is not None,
+            "active_run_id": None,
+        }
+
+    initial = scope()
     with WorkerPresence(
         root,
-        run_id=args.run_id,
-        watcher=not single_pass,
+        run_id=initial["run_id"],
+        watcher=initial["watcher"],
         uploads_only=args.uploads_only,
+        manual_only=initial.get("manual_only", False),
         heartbeat=Path(heartbeat) if heartbeat else None,
     ) as presence:
+        presence.set_scope(**initial)
         while True:
+            with admission.dispatch():
+                admission.refresh()
+                presence.set_scope(**scope())
             pending = runs_needing_work(root, uploads_only=args.uploads_only)
+            if admission.active is not None:
+                pending.sort(key=lambda path: path.name != admission.active)
             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
-                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:
+                # Policy installation, claims, and automatic harness dispatch share this
+                # root lock even for unrestricted workers. Detached stages never need it.
+                with admission.dispatch():
+                    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
+                    if not admission.claim(run_dir, manifest):
+                        presence.set_scope(**scope())
                         continue
-                    outcome = work_on(run_dir, repository, policy, args.turns)
-                finally:
-                    presence.set_active(None)
+                    presence.set_scope(**scope())
+                    run_id = manifest["runId"]
+                    presence.set_active(run_id)
+                    try:
+                        current = run_manifest(run_dir, uploads_only=args.uploads_only)
+                        if current != manifest:
+                            if args.max_uploads is not None:
+                                raise AdmissionError("Run manifest changed before dispatch")
+                            continue
+                        # A bounded claim is durable before the harness can start any work.
+                        outcome = work_on(run_dir, repository, policy, args.turns)
+                    finally:
+                        admission.refresh()
+                        presence.set_scope(**scope())
                 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