From d3333ea4731eaff91dd2243a8c7efbc276af146b Mon Sep 17 00:00:00 2001 From: Austin Jian Date: Mon, 19 Oct 2026 16:55:01 -0400 Subject: [PATCH 1/2] feat: persist a finite automatic upload allowance --- admin/components/run-overview.tsx | 6 +- admin/lib/run-progress.ts | 11 + admin/lib/types.ts | 1 + admin/tests/run-progress.test.ts | 23 + docs/automatic-uploads.md | 41 +- orchestrator/api/app.py | 9 +- orchestrator/supervisor.py | 102 +++- orchestrator/upload_admission.py | 397 +++++++++++++++ orchestrator/worker_presence.py | 88 +++- .../tests/test_orchestrator_api_submission.py | 81 +++ scripts/tests/test_upload_admission.py | 477 ++++++++++++++++++ scripts/tests/test_worker_presence.py | 30 +- 12 files changed, 1217 insertions(+), 49 deletions(-) create mode 100644 orchestrator/upload_admission.py create mode 100644 scripts/tests/test_upload_admission.py diff --git a/admin/components/run-overview.tsx b/admin/components/run-overview.tsx index 311c17a..07d0829 100644 --- a/admin/components/run-overview.tsx +++ b/admin/components/run-overview.tsx @@ -143,7 +143,11 @@ export function RunOverview({ ))} - ) : ready && !progress.finished && !progress.paused && !progress.question ? ( + ) : ready && + !progress.finished && + !progress.paused && + !progress.question && + !progress.worker?.reason ? (

Next: {ready.name}. {ready.nextAction}

diff --git a/admin/lib/run-progress.ts b/admin/lib/run-progress.ts index 9e640a0..8f7eddd 100644 --- a/admin/lib/run-progress.ts +++ b/admin/lib/run-progress.ts @@ -12,6 +12,11 @@ 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'; } @@ -19,6 +24,12 @@ 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.'; diff --git a/admin/lib/types.ts b/admin/lib/types.ts index 74a97eb..b4261e8 100644 --- a/admin/lib/types.ts +++ b/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; diff --git a/admin/tests/run-progress.test.ts b/admin/tests/run-progress.test.ts index 6711e61..890e93b 100644 --- a/admin/tests/run-progress.test.ts +++ b/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); + } +}); diff --git a/docs/automatic-uploads.md b/docs/automatic-uploads.md index 8270e44..39cbd4d 100644 --- a/docs/automatic-uploads.md +++ b/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 diff --git a/orchestrator/api/app.py b/orchestrator/api/app.py index c380259..7cf017d 100644 --- a/orchestrator/api/app.py +++ b/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: diff --git a/orchestrator/supervisor.py b/orchestrator/supervisor.py index 1734b7b..d119878 100644 --- a/orchestrator/supervisor.py +++ b/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 diff --git a/orchestrator/upload_admission.py b/orchestrator/upload_admission.py new file mode 100644 index 0000000..26d92b8 --- /dev/null +++ b/orchestrator/upload_admission.py @@ -0,0 +1,397 @@ +"""Durable, root-wide admission of a finite number of automatic uploads. + +This counts uploads, not provider requests or dollars. The supervisor dispatch lock is +independent of per-run stage locks: detached children never acquire this lock. +""" + +from __future__ import annotations + +import fcntl +import hashlib +import json +import os +import re +import stat +import uuid +from contextlib import contextmanager +from pathlib import Path + +from orchestrator.journal import KINDS +from orchestrator.run_policy import hands_off + +LEDGER_FILE = ".upload-admission.json" +DISPATCH_LOCK_FILE = ".upload-dispatch.lock" +OWNER_LOCK_FILE = ".upload-owner.lock" +SCHEMA = "wander.upload-admission/1" +MAX_BYTES = 16 * 1024 * 1024 + + +class AdmissionError(RuntimeError): + """Uncertain admission state must never authorize more work.""" + + +def _decode(raw: bytes): + def pairs(items): + result = {} + for key, value in items: + if key in result: + raise AdmissionError("Duplicate admission or journal field") + result[key] = value + return result + + try: + return json.loads(raw, object_pairs_hook=pairs, parse_constant=lambda _: _invalid()) + except (ValueError, RecursionError) as error: + raise AdmissionError("Malformed upload admission state") from error + + +def _invalid(): + raise AdmissionError("Non-finite upload admission state") + + +def _read(directory: int, name: str, *, missing: bool = False) -> bytes | None: + try: + descriptor = os.open(name, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK, dir_fd=directory) + except FileNotFoundError: + if missing: + return None + raise AdmissionError(f"Missing upload admission evidence: {name}") from None + try: + metadata = os.fstat(descriptor) + if not stat.S_ISREG(metadata.st_mode) or metadata.st_size > MAX_BYTES: + raise AdmissionError(f"Invalid upload admission evidence: {name}") + with os.fdopen(descriptor, "rb", closefd=False) as source: + raw = source.read(MAX_BYTES + 1) + if len(raw) > MAX_BYTES: + raise AdmissionError(f"Oversized upload admission evidence: {name}") + return raw + finally: + os.close(descriptor) + + +def _open_lock(directory: int, name: str) -> int: + descriptor = os.open( + name, os.O_RDWR | os.O_CREAT | os.O_NOFOLLOW | os.O_NONBLOCK, 0o600, dir_fd=directory + ) + metadata = os.fstat(descriptor) + if not stat.S_ISREG(metadata.st_mode) or metadata.st_nlink != 1: + os.close(descriptor) + raise AdmissionError(f"Invalid upload admission lock: {name}") + return descriptor + + +def _identity(directory: str, manifest: dict) -> dict: + digest = hashlib.sha256( + json.dumps(manifest, sort_keys=True, separators=(",", ":"), allow_nan=False).encode() + ).hexdigest() + return {"directory": directory, "runId": manifest["runId"], "manifestSha256": digest} + + +def _history(directory: int, *, required: bool) -> tuple[bool, bool, bool]: + """Return (terminal, unresolved, started), refusing every unreadable journal line.""" + raw = _read(directory, "journal.jsonl", missing=not required) + if raw is None: + return False, False, False + terminal = started = False + active = set() + for line in raw.splitlines(): + entry = _decode(line) + if ( + not isinstance(entry, dict) + or set(entry) != {"kind", "at", "data"} + or entry["kind"] not in KINDS + or not isinstance(entry["at"], str) + or not isinstance(entry["data"], dict) + ): + raise AdmissionError("Malformed upload journal") + kind, data = entry["kind"], entry["data"] + if kind.startswith("step."): + started = True + step, attempt = data.get("step"), data.get("attempt", "") + if not isinstance(step, str) or not step or not isinstance(attempt, str): + raise AdmissionError("Malformed upload attempt identity") + key = (step, attempt) + if kind in {"step.started", "step.executing"}: + active.add(key) + started = True + else: + if data.get("status") not in ( + "succeeded", + "failed", + "canceled", + "blocked", + "skipped", + ): + raise AdmissionError("Malformed upload attempt outcome") + active.discard(key) + elif kind == "run.finished": + if data.get("status") not in {"succeeded", "failed", "canceled"}: + raise AdmissionError("Malformed upload terminal outcome") + terminal = True + elif kind == "run.resumed": + terminal = False + return terminal, bool(active), started + + +class UploadAdmission: + """One lifetime owner for a bound; all supervisors serialize automatic dispatch here.""" + + def __init__(self, root: Path, limit: int | None, *, uploads_only: bool = False): + if limit is not None and (type(limit) is not int or limit <= 0 or not uploads_only): + raise AdmissionError("--max-uploads must be positive and requires --uploads-only") + self.root = root.resolve() + self.limit = limit + self.uploads_only = uploads_only + self.directory = self.lock = self.owner = None + self.state = None + + def __enter__(self): + try: + self.directory = os.open(self.root, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW) + self.lock = _open_lock(self.directory, DISPATCH_LOCK_FILE) + if self.limit is not None: + self.owner = _open_lock(self.directory, OWNER_LOCK_FILE) + try: + fcntl.flock(self.owner, fcntl.LOCK_EX | fcntl.LOCK_NB) + except BlockingIOError as error: + raise AdmissionError( + "A bounded upload supervisor already owns this root" + ) from error + with self.dispatch(): + if self.limit is not None and self.state is None: + self._check_unadmitted([]) + self.state = { + "schema": SCHEMA, + "root": str(self.root), + "maxUploads": self.limit, + "admissions": [], + "active": None, + } + # The stable lock contains an immutable policy anchor. A crash here + # leaves a missing ledger error, never an unbounded or reset policy. + anchor = json.dumps(self._anchor()).encode() + os.pwrite(self.lock, anchor, 0) + os.fsync(self.lock) + os.fsync(self.directory) + self._save() + if self.state is not None and self.limit is None and self.uploads_only: + raise AdmissionError("This root requires its original --max-uploads limit") + if self.limit is not None: + self.refresh() + return self + except BaseException: + self.__exit__(None, None, None) + raise + + def __exit__(self, *_): + for name in ("owner", "lock", "directory"): + descriptor = getattr(self, name) + if descriptor is not None: + os.close(descriptor) + setattr(self, name, None) + + def _anchor(self): + return {key: self.state[key] for key in ("schema", "root", "maxUploads")} + + @contextmanager + def dispatch(self): + fcntl.flock(self.lock, fcntl.LOCK_EX) + try: + metadata = os.stat(DISPATCH_LOCK_FILE, dir_fd=self.directory, follow_symlinks=False) + held = os.fstat(self.lock) + if (metadata.st_dev, metadata.st_ino) != ( + held.st_dev, + held.st_ino, + ) or held.st_nlink != 1: + raise AdmissionError("Upload dispatch lock was replaced") + if self.owner is not None: + owner = os.stat(OWNER_LOCK_FILE, dir_fd=self.directory, follow_symlinks=False) + held_owner = os.fstat(self.owner) + if (owner.st_dev, owner.st_ino) != ( + held_owner.st_dev, + held_owner.st_ino, + ) or held_owner.st_nlink != 1: + raise AdmissionError("Upload owner lock was replaced") + self._load() + yield + finally: + fcntl.flock(self.lock, fcntl.LOCK_UN) + + def _load(self): + anchor_raw = os.pread(self.lock, MAX_BYTES + 1, 0) + raw = _read(self.directory, LEDGER_FILE, missing=True) + if not anchor_raw and raw is None: + if self.state is not None: + raise AdmissionError("Upload admission policy disappeared") + return + if not anchor_raw or raw is None: + raise AdmissionError("Missing upload admission ledger or policy anchor") + anchor, state = _decode(anchor_raw), _decode(raw) + if ( + not isinstance(state, dict) + or set(state) != {"schema", "root", "maxUploads", "admissions", "active"} + or state["schema"] != SCHEMA + or state["root"] != str(self.root) + or type(state["maxUploads"]) is not int + or state["maxUploads"] <= 0 + or not isinstance(state["admissions"], list) + or len(state["admissions"]) > state["maxUploads"] + or not isinstance(anchor, dict) + or type(anchor.get("maxUploads")) is not int + or anchor != {key: state[key] for key in ("schema", "root", "maxUploads")} + or (self.limit is not None and state["maxUploads"] != self.limit) + ): + raise AdmissionError("Invalid or mismatched upload admission policy") + directories, ids = set(), set() + for entry in state["admissions"]: + if ( + not isinstance(entry, dict) + or set(entry) != {"directory", "runId", "manifestSha256"} + or not isinstance(entry["directory"], str) + or entry["directory"] in {"", ".", ".."} + or "/" in entry["directory"] + or not isinstance(entry["runId"], str) + or not 0 < len(entry["runId"]) <= 512 + or not isinstance(entry["manifestSha256"], str) + or re.fullmatch(r"[0-9a-f]{64}", entry["manifestSha256"]) is None + or entry["directory"] in directories + or entry["runId"] in ids + ): + raise AdmissionError("Invalid or duplicate upload admission identity") + directories.add(entry["directory"]) + ids.add(entry["runId"]) + if state["active"] is not None and ( + not isinstance(state["active"], str) or state["active"] not in directories + ): + raise AdmissionError("Invalid active upload admission") + self.state = state + + def _save(self): + temporary = f".{uuid.uuid4()}.upload-admission.tmp" + descriptor = os.open( + temporary, + os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, + 0o600, + dir_fd=self.directory, + ) + try: + with os.fdopen(descriptor, "w") as target: + json.dump(self.state, target, sort_keys=True, allow_nan=False) + target.write("\n") + target.flush() + os.fsync(target.fileno()) + os.replace(temporary, LEDGER_FILE, src_dir_fd=self.directory, dst_dir_fd=self.directory) + os.fsync(self.directory) + finally: + try: + os.unlink(temporary, dir_fd=self.directory) + except FileNotFoundError: + pass + + def _inspect(self, name, *, identity=None): + from orchestrator.supervisor import run_manifest + + directory = os.open( + name, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=self.directory + ) + try: + manifest = run_manifest(self.root / name) + if manifest is None: + raise AdmissionError(f"Unreadable upload manifest: {name}") + if identity is not None and _identity(name, manifest) != identity: + raise AdmissionError(f"Admitted upload identity changed: {name}") + terminal, unresolved, started = _history(directory, required=identity is not None) + for evidence in ("BRIEF.md", "session.id", "transcript.jsonl"): + try: + os.stat(evidence, dir_fd=directory, follow_symlinks=False) + started = True + except FileNotFoundError: + pass + return manifest, (terminal, unresolved, started) + finally: + os.close(directory) + + def _check_unadmitted(self, admissions): + from orchestrator.supervisor import run_manifest + + admitted = {item["directory"] for item in admissions} + for described in sorted(self.root.glob("*/run.json")): + if described.parent.name in admitted: + continue + manifest = run_manifest(described.parent) + if manifest is None or not hands_off(manifest.get("options", {})): + continue + _, (terminal, unresolved, started) = self._inspect(described.parent.name) + if unresolved or (started and not terminal): + raise AdmissionError( + f"Unadmitted automatic work already started: {described.parent.name}" + ) + + def refresh(self): + """Reclaim prior admissions before considering a fresh upload; caller holds dispatch.""" + if self.limit is None: + return + unfinished = [] + for identity in self.state["admissions"]: + _, (terminal, unresolved, _) = self._inspect(identity["directory"], identity=identity) + if not terminal or unresolved: + unfinished.append(identity["directory"]) + self._check_unadmitted(self.state["admissions"]) + active = self.state["active"] + next_active = active if active in unfinished else next(iter(unfinished), None) + if active != next_active: + self.state["active"] = next_active + self._save() + + @property + def active(self): + return self.state["active"] if self.state else None + + @property + def remaining(self): + return self.limit - len(self.state["admissions"]) if self.limit is not None else None + + def claim(self, run_dir: Path, manifest: dict) -> bool: + """Called immediately before a harness dispatch, under the common root lock.""" + if manifest is None: + raise AdmissionError("Unreadable upload manifest before dispatch") + if not hands_off(manifest.get("options", {})): + return self.limit is None + if self.limit is None: + return self.state is None + self.refresh() + name = run_dir.name + identity = _identity(name, manifest) + admitted = self.state["admissions"] + if any(item["directory"] == name for item in admitted): + return self.active == name + if self.active is not None or not self.remaining: + return False + if any(item["runId"] == manifest["runId"] for item in admitted): + raise AdmissionError("Upload runId already belongs to an admitted directory") + _, (terminal, unresolved, started) = self._inspect(name) + if terminal: + return False + if unresolved or started: + raise AdmissionError("Cannot adopt previously started automatic work") + directory = os.open( + name, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=self.directory + ) + try: + descriptor = os.open( + "journal.jsonl", + os.O_WRONLY | os.O_CREAT | os.O_APPEND | os.O_NOFOLLOW | os.O_NONBLOCK, + 0o600, + dir_fd=directory, + ) + try: + os.fsync(descriptor) + finally: + os.close(descriptor) + os.fsync(directory) + finally: + os.close(directory) + admitted.append(identity) + self.state["active"] = name + self._save() + return True diff --git a/orchestrator/worker_presence.py b/orchestrator/worker_presence.py index 0e1fa03..b23fc07 100644 --- a/orchestrator/worker_presence.py +++ b/orchestrator/worker_presence.py @@ -13,7 +13,7 @@ from pathlib import Path SCHEMA = "wander.worker-presence" -VERSION = 2 +VERSION = 3 FRESHNESS_SECONDS = 45.0 REFRESH_SECONDS = 5.0 MAX_RECORD_BYTES = 16_384 @@ -31,6 +31,7 @@ def _open_directory(root: Path) -> int: 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 + admission_reason = None if root is not None: directory = None try: @@ -57,10 +58,14 @@ def worker_status(root: Path | None, run_id: str | None = None, *, hands_off: bo if ( record.get("schema") != SCHEMA or type(record.get("version")) is not int - or record["version"] not in (1, VERSION) + or record["version"] not in (1, 2, VERSION) + or (record["version"] >= 2 and type(record.get("uploads_only")) is not bool) or ( record["version"] == VERSION - and type(record.get("uploads_only")) is not bool + and ( + type(record.get("manual_only")) is not bool + or (record["manual_only"] and record["uploads_only"]) + ) ) or record.get("runs_root") != str(resolved) or type(record.get("watcher")) is not bool @@ -68,18 +73,17 @@ def worker_status(root: Path | None, run_id: str | None = None, *, hands_off: bo or not _run_id(record["run_id"]) or "active_run_id" not in record or not _run_id(record["active_run_id"]) + or record.get("admission") not in (None, "occupied", "exhausted") or type(seen) not in (int, float) or not math.isfinite(seen) 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 - ): + # Old readers ignore versions whose capabilities they cannot represent. + # Version 1 is unrestricted; version 2 adds uploads-only scope. + if record["version"] >= 2 and record["uploads_only"] and hands_off is not True: + continue + if record["version"] == VERSION and record["manual_only"] and hands_off is True: continue selected, active = record["run_id"], record["active_run_id"] if selected is not None and active not in (None, selected): @@ -90,6 +94,11 @@ def worker_status(root: Path | None, run_id: str | None = None, *, hands_off: bo matches = selected in (None, run_id) else: matches = active == run_id + if hands_off is True and record.get("uploads_only") is True: + if record.get("admission") == "exhausted": + admission_reason = "admission_exhausted" + elif record.get("admission") == "occupied" and admission_reason is None: + admission_reason = "admission_busy" if matches: matching += 1 busy += active is not None @@ -100,14 +109,22 @@ def worker_status(root: Path | None, run_id: str | None = None, *, hands_off: bo finally: if directory is not None: os.close(directory) + reason = admission_reason if not matching else None return { + **({"reason": reason} if reason else {}), "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." + else ( + "The upload admission allowance is exhausted; this upload remains queued." + if reason == "admission_exhausted" + else "The upload supervisor is occupied with an admitted run; this upload remains queued." + if reason == "admission_busy" + else "No matching supervisor has a recent heartbeat." + ) ), } @@ -122,10 +139,17 @@ def __init__( run_id: str | None = None, watcher: bool = True, uploads_only: bool = False, + manual_only: bool = False, heartbeat: Path | None = None, refresh_interval: float = REFRESH_SECONDS, ): - if not _run_id(run_id) or type(watcher) is not bool or type(uploads_only) is not bool: + if ( + not _run_id(run_id) + or type(watcher) is not bool + or type(uploads_only) is not bool + or type(manual_only) is not bool + or (uploads_only and manual_only) + ): 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") @@ -133,10 +157,12 @@ def __init__( self.run_id = run_id self.watcher = watcher self.uploads_only = uploads_only + self.manual_only = manual_only self.heartbeat = heartbeat self.refresh_interval = refresh_interval self.name = f"{uuid.uuid4()}.json" self._active: str | None = None + self._admission: str | None = None self._directory: int | None = None self._thread: threading.Thread | None = None self._stop = threading.Event() @@ -167,6 +193,42 @@ def __exit__(self, exc_type, exc_value, traceback): if self._error is not None and exc_type is None: raise RuntimeError("Supervisor heartbeat stopped refreshing") from self._error + def set_scope( + self, + *, + run_id: str | None, + watcher: bool, + active_run_id: str | None = None, + admission: str | None = None, + manual_only: bool = False, + ) -> None: + """Publish admission scope and activity together under the heartbeat lock.""" + if ( + not _run_id(run_id) + or type(watcher) is not bool + or not _run_id(active_run_id) + or admission not in (None, "occupied", "exhausted") + or type(manual_only) is not bool + or (manual_only and self.uploads_only) + or (run_id is not None and active_run_id not in (None, run_id)) + ): + raise ValueError("Invalid supervisor scope") + with self._lock: + if self._error is not None: + raise RuntimeError("Supervisor heartbeat stopped refreshing") + self.run_id = run_id + self.watcher = watcher + self._active = active_run_id + self._admission = admission + self.manual_only = manual_only + try: + self._write() + except OSError as error: + self._error = error + self._stop.set() + self._remove() + raise RuntimeError("Supervisor heartbeat stopped refreshing") from 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") @@ -192,9 +254,11 @@ def _write(self) -> None: "runs_root": str(self.root), "watcher": self.watcher, "uploads_only": self.uploads_only, + "manual_only": self.manual_only, "run_id": self.run_id, "active_run_id": self._active, "last_seen": now, + **({"admission": self._admission} if self._admission else {}), } # Preserve the legacy timestamp format and refresh it on the same background timer. if self.heartbeat is not None: diff --git a/scripts/tests/test_orchestrator_api_submission.py b/scripts/tests/test_orchestrator_api_submission.py index 7ef6736..0201077 100644 --- a/scripts/tests/test_orchestrator_api_submission.py +++ b/scripts/tests/test_orchestrator_api_submission.py @@ -239,6 +239,87 @@ def finish_without_model(_task, workspace, _prompt, **_kwargs): self.assertIsNone(manual["finished"]) self.assertFalse(manual["currentStages"]) + def test_upload_admission_survives_restart_and_leaves_next_upload_queued(self): + from orchestrator import supervisor + from orchestrator.agent.contracts import AgentResult + from orchestrator.agent.harness import HarnessAgent, HarnessOutcome, HarnessPolicy + from orchestrator.journal import Journal + + processed = [] + passes = 0 + + class StopRehearsal(Exception): + pass + + def availability(): + with patch( + "orchestrator.doctor.inspect_setup", + return_value={"ready": True, "status": "okay", "checks": []}, + ): + response = self.client.post( + "/api/pipeline/doctor", headers=AUTH, json={"hands_off": True} + ) + self.assertEqual(response.status_code, 200, response.text) + report = response.json() + if report["worker"].get("reason") == "admission_exhausted": + check = next(item for item in report["checks"] if item["name"] == "worker.watch") + self.assertIn("operator", check["action"]) + self.assertNotIn("Connect", check["action"]) + return report["worker"] + + def next_pass(_): + nonlocal passes + passes += 1 + if passes == 1: + self.assertEqual(availability()["status"], "available") + response = self.upload("first", options='{"hands_off":true}') + self.assertEqual(response.status_code, 201, response.text) + else: + self.assertEqual(processed, ["first"]) + self.assertEqual(availability()["status"], "unavailable") + self.assertEqual(availability()["reason"], "admission_exhausted") + response = self.upload("second", options='{"hands_off":true}') + self.assertEqual(response.status_code, 201, response.text) + raise StopRehearsal + + def finish_without_model(_task, workspace, _prompt, **_kwargs): + manifest = json.loads((workspace / "run.json").read_text()) + processed.append(manifest["runId"]) + self.assertEqual(availability()["status"], "unavailable") + progress = self.client.get("/api/pipeline/runs/first/progress", headers=AUTH).json() + self.assertEqual(progress["worker"]["status"], "available") + Journal(workspace).append( + "run.finished", status="failed", summary="Fixture ended without inference." + ) + return HarnessOutcome(AgentResult(status="completed", transcript_path="inert")) + + arguments = ["--runs", str(self.root / "runs"), "--uploads-only", "--max-uploads", "1"] + with ( + patch.dict(os.environ, {}, clear=True), + patch.object(supervisor, "repository_from_environment", return_value=self.repository), + patch.object( + HarnessPolicy, + "from_environment", + return_value=HarnessPolicy(kind="command", command=("unused",)), + ), + patch.object(HarnessAgent, "run", side_effect=finish_without_model), + ): + with ( + patch.object(supervisor.time, "sleep", side_effect=next_pass), + self.assertRaises(StopRehearsal), + ): + supervisor.main([*arguments, "--watch"]) + self.assertEqual(supervisor.main([*arguments, "--once"]), 0) + + self.assertEqual(processed, ["first"]) + self.assertEqual(self.repository.run_summary("first")["attempts"], []) + second = self.client.get("/api/pipeline/runs/second/progress", headers=AUTH).json() + self.assertIsNone(second["finished"]) + self.assertFalse(second["currentStages"]) + self.assertIsNone(second["question"]) + self.assertEqual(second["worker"]["status"], "unavailable") + self.assertEqual(self.repository.run_summary("second")["attempts"], []) + 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_upload_admission.py b/scripts/tests/test_upload_admission.py new file mode 100644 index 0000000..5300757 --- /dev/null +++ b/scripts/tests/test_upload_admission.py @@ -0,0 +1,477 @@ +"""Finite upload admissions persist without invoking a model or provider.""" + +import json + +import pytest + +from orchestrator import supervisor +from orchestrator.journal import Journal +from orchestrator.upload_admission import ( + AdmissionError, + DISPATCH_LOCK_FILE, + LEDGER_FILE, + OWNER_LOCK_FILE, + UploadAdmission, +) +from orchestrator.worker_presence import worker_status + + +def upload(root, name="one"): + return supervisor.open_run( + root, run_id=name, name=name, source=None, options={"hands_off": True} + ) + + +def claim(policy, run): + with policy.dispatch(): + return policy.claim(run, supervisor.run_manifest(run)) + + +def ledger(root): + return json.loads((root / LEDGER_FILE).read_text()) + + +@pytest.fixture(autouse=True) +def no_providers(monkeypatch): + monkeypatch.delenv("WANDER_STAGE_AUTHORIZATION", raising=False) + monkeypatch.setattr(supervisor, "repository_from_environment", lambda: None) + + def forbidden(*_args, **_kwargs): + pytest.fail("Admission tests must not invoke a real harness") + + monkeypatch.setattr(supervisor.HarnessAgent, "run", forbidden) + + +def test_restart_keeps_failed_admission_consumed(tmp_path): + first, second = upload(tmp_path), upload(tmp_path, "two") + with UploadAdmission(tmp_path, 1, uploads_only=True) as policy: + assert claim(policy, first) + Journal(first).append("run.finished", status="failed") + with UploadAdmission(tmp_path, 1, uploads_only=True) as policy: + assert policy.remaining == 0 + assert policy.active is None + assert not claim(policy, second) + assert len(ledger(tmp_path)["admissions"]) == 1 + + +@pytest.mark.parametrize("status", ["failed", "succeeded", "canceled"]) +def test_terminal_slot_waits_for_every_detached_attempt(tmp_path, status): + first, second = upload(tmp_path), upload(tmp_path, "two") + with UploadAdmission(tmp_path, 2, uploads_only=True) as policy: + assert claim(policy, first) + journal = Journal(first) + for attempt in ("old", "new"): + journal.append("step.started", step="world", attempt=attempt) + journal.append("run.finished", status=status) + assert not claim(policy, second) + journal.append("step.finished", step="world", attempt="new", status="failed") + assert not claim(policy, second) + journal.append("step.finished", step="world", attempt="old", status="succeeded") + assert claim(policy, second) + assert policy.active == "two" + assert policy.remaining == 0 + + +def test_paused_run_keeps_slot_and_restart_reclaims_it(tmp_path): + first, second = upload(tmp_path), upload(tmp_path, "two") + with UploadAdmission(tmp_path, 2, uploads_only=True) as policy: + assert claim(policy, first) + (first / "paused").touch() + with UploadAdmission(tmp_path, 2, uploads_only=True) as policy: + assert policy.active == first.name + assert not claim(policy, second) + (first / "paused").unlink() + assert claim(policy, first) + assert len(ledger(tmp_path)["admissions"]) == 1 + + +def test_resume_reclaims_existing_admission_without_refund(tmp_path): + first, second = upload(tmp_path), upload(tmp_path, "two") + with UploadAdmission(tmp_path, 1, uploads_only=True) as policy: + assert claim(policy, first) + Journal(first).append("run.finished", status="failed") + assert not claim(policy, second) + Journal(first).append("run.resumed", reason="operator") + assert claim(policy, first) + assert policy.remaining == 0 + + +@pytest.mark.parametrize("mutation", ["manifest", "directory", "runId", "journal", "ledger"]) +def test_missing_or_modified_admitted_evidence_fails_closed(tmp_path, mutation): + first = upload(tmp_path) + with UploadAdmission(tmp_path, 1, uploads_only=True) as policy: + assert claim(policy, first) + if mutation == "manifest": + (first / "run.json").unlink() + elif mutation == "directory": + first.rename(tmp_path / "renamed") + elif mutation == "runId": + manifest = json.loads((first / "run.json").read_text()) + manifest["runId"] = "replacement" + (first / "run.json").write_text(json.dumps(manifest)) + elif mutation == "journal": + (first / "journal.jsonl").unlink() + else: + (tmp_path / LEDGER_FILE).unlink() + with pytest.raises((AdmissionError, OSError)): + with UploadAdmission(tmp_path, 1, uploads_only=True): + pytest.fail("Changed evidence must not authorize work") + + +def test_manifest_hash_is_canonical_but_option_change_is_rejected(tmp_path): + first = upload(tmp_path) + with UploadAdmission(tmp_path, 1, uploads_only=True) as policy: + assert claim(policy, first) + manifest = json.loads((first / "run.json").read_text()) + (first / "run.json").write_text(json.dumps(manifest, sort_keys=True)) + assert claim(policy, first) + manifest["options"]["people"] = 2 + (first / "run.json").write_text(json.dumps(manifest)) + with pytest.raises(AdmissionError, match="identity changed"): + claim(policy, first) + + +@pytest.mark.parametrize("contents", ["{}", "[]", "{bad", '{"x":1,"x":2}', "NaN"]) +@pytest.mark.parametrize("target", [LEDGER_FILE, "journal.jsonl"]) +def test_corrupt_ledger_or_journal_never_releases_slot(tmp_path, contents, target): + first = upload(tmp_path) + with UploadAdmission(tmp_path, 1, uploads_only=True) as policy: + assert claim(policy, first) + path = tmp_path / target if target == LEDGER_FILE else first / target + path.write_text(contents) + with pytest.raises(AdmissionError): + with UploadAdmission(tmp_path, 1, uploads_only=True): + pytest.fail("Corrupt evidence must fail closed") + + +@pytest.mark.parametrize( + "evidence", ["started", "terminal_unresolved", "transcript", "finished_step"] +) +def test_new_policy_refuses_unadmitted_existing_work(tmp_path, evidence): + first = upload(tmp_path) + if evidence == "transcript": + (first / "transcript.jsonl").touch() + elif evidence == "finished_step": + Journal(first).append("step.finished", step="world", attempt="one", status="succeeded") + else: + Journal(first).append("step.started", step="world", attempt="one") + if evidence == "terminal_unresolved": + Journal(first).append("run.finished", status="failed") + with pytest.raises(AdmissionError, match="already started"): + with UploadAdmission(tmp_path, 1, uploads_only=True): + pytest.fail("Existing work cannot be made free") + assert not (tmp_path / LEDGER_FILE).exists() + + +def test_owner_lock_excludes_second_worker_and_releases_on_exit(tmp_path): + with UploadAdmission(tmp_path, 2, uploads_only=True): + with pytest.raises(AdmissionError, match="already owns"): + with UploadAdmission(tmp_path, 2, uploads_only=True): + pytest.fail("Only one bounded owner is allowed") + with UploadAdmission(tmp_path, 2, uploads_only=True): + pass + + +def test_unrestricted_existing_worker_rechecks_new_policy(tmp_path): + first = upload(tmp_path) + with UploadAdmission(tmp_path, None) as unrestricted: + assert claim(unrestricted, first) + with UploadAdmission(tmp_path, 1, uploads_only=True): + assert not claim(unrestricted, first) + assert not claim(unrestricted, first) + manual = supervisor.open_run(tmp_path, run_id="manual", name="manual", source=None) + assert claim(unrestricted, manual) + + +@pytest.mark.parametrize("target", [LEDGER_FILE, DISPATCH_LOCK_FILE, OWNER_LOCK_FILE]) +def test_state_and_lock_symlinks_are_never_followed(tmp_path, target): + root = tmp_path / "runs" + root.mkdir() + outside = tmp_path / "outside" + outside.write_text("preserve") + (root / target).symlink_to(outside) + with pytest.raises((AdmissionError, OSError)): + with UploadAdmission(root, 1, uploads_only=True): + pytest.fail("Must not follow symlinks") + assert outside.read_text() == "preserve" + + +@pytest.mark.parametrize("target", ["run.json", "journal.jsonl"]) +def test_admitted_manifest_and_journal_symlinks_fail_closed(tmp_path, target): + first = upload(tmp_path) + with UploadAdmission(tmp_path, 1, uploads_only=True) as policy: + assert claim(policy, first) + path = first / target + outside = tmp_path / "outside" + path.rename(outside) + path.symlink_to(outside) + with pytest.raises((AdmissionError, OSError)): + claim(policy, first) + + +@pytest.mark.parametrize( + "flags", + [ + ["--max-uploads", "0"], + ["--max-uploads", "-1"], + ["--max-uploads", "1"], + ["--uploads-only", "--max-uploads", "1.5"], + ], +) +def test_invalid_flags_fail_before_presence_or_harness(tmp_path, monkeypatch, flags): + monkeypatch.setattr(supervisor, "WorkerPresence", lambda *_a, **_kw: pytest.fail("no presence")) + with pytest.raises(SystemExit) as error: + supervisor.main(["--runs", str(tmp_path), "--once", *flags]) + assert error.value.code == 2 + + +@pytest.mark.parametrize("flags", [["--uploads-only"], ["--uploads-only", "--max-uploads", "2"]]) +def test_existing_policy_requires_same_limit_before_presence(tmp_path, monkeypatch, flags): + with UploadAdmission(tmp_path, 1, uploads_only=True): + pass + monkeypatch.setattr(supervisor, "WorkerPresence", lambda *_a, **_kw: pytest.fail("no presence")) + with pytest.raises(SystemExit) as error: + supervisor.main(["--runs", str(tmp_path), "--once", *flags]) + assert error.value.code == 2 + + +def test_watch_presence_scopes_active_pause_drain_and_exhaustion(tmp_path, monkeypatch): + first = upload(tmp_path) + second = upload(tmp_path, "two") + calls = [] + passes = 0 + + class Done(Exception): + pass + + def work(path, *_args): + calls.append(path) + assert worker_status(tmp_path, hands_off=True)["status"] == "unavailable" + assert worker_status(tmp_path, path.name, hands_off=True)["busyWorkers"] == 1 + Journal(path).append("step.started", step="world", attempt="one") + (path / "paused").touch() + return supervisor.WorkOutcome("paused", 1) + + def tick(_): + nonlocal passes + passes += 1 + assert calls == [first] + assert worker_status(tmp_path, hands_off=True)["status"] == "unavailable" + if passes == 1: + assert worker_status(tmp_path, "one", hands_off=True)["status"] == "available" + Journal(first).append("run.finished", status="canceled") + elif passes == 2: + assert worker_status(tmp_path, "one", hands_off=True)["status"] == "available" + Journal(first).append("step.finished", step="world", attempt="one", status="failed") + else: + assert worker_status(tmp_path, "one", hands_off=True)["status"] == "unavailable" + assert worker_status(tmp_path, "two", hands_off=True)["status"] == "unavailable" + raise Done + + monkeypatch.setattr(supervisor, "work_on", work) + monkeypatch.setattr(supervisor.time, "sleep", tick) + with pytest.raises(Done): + supervisor.main( + ["--runs", str(tmp_path), "--uploads-only", "--max-uploads", "1", "--watch"] + ) + assert not (second / "journal.jsonl").exists() + + +def test_idle_bounded_watcher_accepts_new_uploads(tmp_path, monkeypatch): + class Done(Exception): + pass + + def tick(_): + assert worker_status(tmp_path, hands_off=True)["status"] == "available" + raise Done + + monkeypatch.setattr(supervisor.time, "sleep", tick) + with pytest.raises(Done): + supervisor.main( + ["--runs", str(tmp_path), "--uploads-only", "--max-uploads", "1", "--watch"] + ) + + +@pytest.mark.parametrize("status", ["blocked", "skipped"]) +def test_normal_nonsuccess_step_outcomes_settle_admission(tmp_path, status): + first, second = upload(tmp_path), upload(tmp_path, "two") + with UploadAdmission(tmp_path, 2, uploads_only=True) as policy: + assert claim(policy, first) + journal = Journal(first) + journal.append("step.started", step="world", attempt="one") + journal.append("step.executing", step="world", attempt="one") + journal.append("step.finished", step="world", attempt="one", status=status) + journal.append("run.finished", status="failed") + assert claim(policy, second) + + +@pytest.mark.parametrize( + "admission,reason", [("occupied", "admission_busy"), ("exhausted", "admission_exhausted")] +) +def test_presence_admission_reason_requires_automatic_unmatched_query(tmp_path, admission, reason): + from orchestrator.worker_presence import WorkerPresence + + with WorkerPresence(tmp_path, uploads_only=True) as presence: + presence.set_scope( + run_id="one" if admission == "occupied" else None, + watcher=admission == "occupied", + active_run_id="one" if admission == "occupied" else None, + admission=admission, + ) + status = worker_status(tmp_path, hands_off=True) + assert status["status"] == "unavailable" + assert status["reason"] == reason + assert "queued" in status["detail"] + assert "reason" not in worker_status(tmp_path) + assert "reason" not in worker_status(tmp_path, "two") + if admission == "occupied": + assert "reason" not in worker_status(tmp_path, "one", hands_off=True) + with WorkerPresence(tmp_path, uploads_only=True): + assert worker_status(tmp_path, hands_off=True)["status"] == "available" + assert "reason" not in worker_status(tmp_path, hands_off=True) + + +def test_dispatch_serializes_policy_installation_with_unrestricted_work(tmp_path): + import threading + + first = upload(tmp_path) + attempting, completed = threading.Event(), threading.Event() + errors = [] + + def install(): + attempting.set() + try: + with UploadAdmission(tmp_path, 1, uploads_only=True): + errors.append("unexpected policy installation") + except AdmissionError as error: + errors.append(str(error)) + finally: + completed.set() + + with UploadAdmission(tmp_path, None) as unrestricted: + with unrestricted.dispatch(): + assert unrestricted.claim(first, supervisor.run_manifest(first)) + thread = threading.Thread(target=install) + thread.start() + assert attempting.wait(2) + assert not completed.wait(0.05) + Journal(first).append("step.started", step="world", attempt="existing") + assert completed.wait(2) + thread.join() + assert len(errors) == 1 and "already started" in errors[0] + assert not (tmp_path / LEDGER_FILE).exists() + + +@pytest.mark.parametrize("target", [DISPATCH_LOCK_FILE, OWNER_LOCK_FILE]) +def test_replaced_lock_stops_live_owner(tmp_path, target): + first = upload(tmp_path) + with UploadAdmission(tmp_path, 1, uploads_only=True) as policy: + assert claim(policy, first) + path = tmp_path / target + contents = path.read_bytes() + path.unlink() + path.write_bytes(contents) + with pytest.raises(AdmissionError, match="lock was replaced"): + claim(policy, first) + + +def test_conflicting_run_id_fails_before_presence(tmp_path, monkeypatch): + first = upload(tmp_path) + with UploadAdmission(tmp_path, 2, uploads_only=True) as policy: + assert claim(policy, first) + monkeypatch.setattr(supervisor, "WorkerPresence", lambda *_a, **_kw: pytest.fail("no presence")) + with pytest.raises(SystemExit) as error: + supervisor.main( + ["--runs", str(tmp_path), "--uploads-only", "--max-uploads", "2", "--run-id", "other"] + ) + assert error.value.code == 2 + + +@pytest.mark.parametrize( + "field,value", + [ + ("maxUploads", True), + ("root", "/different"), + ("active", "unclaimed"), + ("admissions", {}), + ("extra", 1), + ], +) +def test_ledger_schema_and_root_are_strict(tmp_path, field, value): + with UploadAdmission(tmp_path, 1, uploads_only=True): + pass + state = ledger(tmp_path) + state[field] = value + (tmp_path / LEDGER_FILE).write_text(json.dumps(state)) + with pytest.raises(AdmissionError): + with UploadAdmission(tmp_path, 1, uploads_only=True): + pytest.fail("Invalid policy must not be used") + + +def test_unrestricted_selected_manual_run_still_dispatches_under_active_policy( + tmp_path, monkeypatch +): + first = upload(tmp_path) + manual = supervisor.open_run(tmp_path, run_id="manual", name="manual", source=None) + with UploadAdmission(tmp_path, 1, uploads_only=True) as policy: + assert claim(policy, first) + calls = [] + monkeypatch.setattr( + supervisor, + "work_on", + lambda path, *_: calls.append(path) or supervisor.WorkOutcome("manual", 0), + ) + assert supervisor.main(["--runs", str(tmp_path), "--run-id", "manual"]) == 0 + assert calls == [manual] + + +def test_unrestricted_watcher_on_policy_root_advertises_manual_only(tmp_path, monkeypatch): + with UploadAdmission(tmp_path, 1, uploads_only=True): + pass + + class Done(Exception): + pass + + def tick(_): + assert worker_status(tmp_path)["status"] == "available" + assert worker_status(tmp_path, "manual")["status"] == "available" + assert worker_status(tmp_path, hands_off=True)["status"] == "unavailable" + assert worker_status(tmp_path, "automatic", hands_off=True)["status"] == "unavailable" + raise Done + + monkeypatch.setattr(supervisor.time, "sleep", tick) + with pytest.raises(Done): + supervisor.main(["--runs", str(tmp_path), "--watch"]) + + +def test_selected_watcher_stops_if_another_admitted_run_is_resumed(tmp_path, monkeypatch): + first, second = upload(tmp_path), upload(tmp_path, "two") + with UploadAdmission(tmp_path, 2, uploads_only=True) as policy: + assert claim(policy, first) + Journal(first).append("run.finished", status="failed") + assert claim(policy, second) + Journal(second).append("run.finished", status="failed") + passes = 0 + + def tick(_): + nonlocal passes + passes += 1 + assert passes == 1 + Journal(first).append("run.resumed", reason="operator") + + monkeypatch.setattr(supervisor.time, "sleep", tick) + with pytest.raises(SystemExit) as error: + supervisor.main( + [ + "--runs", + str(tmp_path), + "--uploads-only", + "--max-uploads", + "2", + "--run-id", + "two", + "--watch", + ] + ) + assert error.value.code == 2 + assert worker_status(tmp_path, "one", hands_off=True)["status"] == "unavailable" diff --git a/scripts/tests/test_worker_presence.py b/scripts/tests/test_worker_presence.py index 18a20fd..f3a51aa 100644 --- a/scripts/tests/test_worker_presence.py +++ b/scripts/tests/test_worker_presence.py @@ -68,7 +68,7 @@ def test_selected_watcher_never_advertises_new_uploads(tmp_path): {"last_seen": float("inf")}, {"last_seen": True}, {"last_seen": "now"}, - {"version": 3}, + {"version": 4}, {"uploads_only": "true"}, {"uploads_only": None}, {"version": True}, @@ -285,7 +285,7 @@ def test_uploads_only_capability_matches_automatic_requests_only(tmp_path, selec ) record = json.loads((tmp_path / ".workers" / presence.name).read_text()) assert record["run_id"] == selected - assert record["version"] == 2 # Version-1 readers reject instead of overstating capability. + assert record["version"] == 3 # Older readers reject rather than overstate capability. assert record["uploads_only"] is True @@ -318,3 +318,29 @@ def test_version_two_requires_explicit_capability(tmp_path): def test_presence_rejects_non_boolean_capability(tmp_path, value): with pytest.raises(ValueError, match="scope"): WorkerPresence(tmp_path, uploads_only=value) + + +@pytest.mark.parametrize("uploads_only", [False, True]) +def test_historical_version_two_worker_preserves_upload_capability(tmp_path, uploads_only): + with WorkerPresence(tmp_path, uploads_only=uploads_only) as presence: + path = tmp_path / ".workers" / presence.name + record = json.loads(path.read_text()) + record["version"] = 2 + del record["manual_only"] + path.write_text(json.dumps(record)) + assert worker_status(tmp_path, hands_off=True)["status"] == "available" + assert (worker_status(tmp_path)["status"] == "available") == (not uploads_only) + + +@pytest.mark.parametrize( + "mutation", + [{"manual_only": None}, {"manual_only": "true"}, {"manual_only": True, "uploads_only": True}], +) +def test_version_three_requires_consistent_manual_capability(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" + assert worker_status(tmp_path, hands_off=True)["status"] == "unavailable" From c3e23158bf9e6a6b4c163c02b45da36ba553a5df Mon Sep 17 00:00:00 2001 From: Austin Jian Date: Mon, 19 Oct 2026 17:01:18 -0400 Subject: [PATCH 2/2] test: await detached stage before finishing session --- scripts/tests/test_orchestrator_session.py | 21 +++++++++++++-------- 1 file changed, 13 insertions(+), 8 deletions(-) diff --git a/scripts/tests/test_orchestrator_session.py b/scripts/tests/test_orchestrator_session.py index 80cdccc..ee94729 100644 --- a/scripts/tests/test_orchestrator_session.py +++ b/scripts/tests/test_orchestrator_session.py @@ -150,15 +150,11 @@ def test_the_agent_runs_a_step_and_the_run_records_it(self): session = RunSession( "run-1", run_dir, - HarnessAgent( - self.agent( - 'subprocess.run(["wander", "step", "clean"])\n' - 'subprocess.run(["wander", "finish", "blocked", "source is not a video"])' - ) - ), + HarnessAgent(self.agent('subprocess.run(["wander", "step", "clean"], check=True)')), ) - outcome = session.work(turns=2) - self.assertEqual(outcome.finished, "blocked") + outcome = session.turn() + self.assertIsNone(outcome.finished) + self.assertEqual(outcome.steps_run, 1) journal = Journal(run_dir) # The step runs on its own, so its outcome lands after the agent's turn is over. for _ in range(200): @@ -169,6 +165,15 @@ def test_the_agent_runs_a_step_and_the_run_records_it(self): self.assertEqual(journal.last_status("clean"), "failed") finished = [e for e in journal.entries() if e.kind == "step.finished"][0] self.assertTrue((run_dir / finished.data["log"]).is_file(), "the step's log is kept") + # Finish only after the detached stage settles. Finishing in the launch turn + # races the child's admission and correctly blocks it on a slower runner. + # Rewrite the stand-in script while retaining the session's configured environment. + self.agent( + 'subprocess.run(["wander", "finish", "blocked", "source is not a video"], check=True)' + ) + outcome = session.work(turns=1) + self.assertEqual(outcome.finished, "blocked") + self.assertEqual(journal.last_status("clean"), "failed") def test_a_question_ends_the_turn_and_holds_the_run(self): run_dir = self.open()