diff --git a/docs/automatic-uploads.md b/docs/automatic-uploads.md index 4c4cd8c..8270e44 100644 --- a/docs/automatic-uploads.md +++ b/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, diff --git a/docs/pipeline-setup.md b/docs/pipeline-setup.md index 610835b..c0a88c0 100644 --- a/docs/pipeline-setup.md +++ b/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 diff --git a/orchestrator/api/app.py b/orchestrator/api/app.py index eca54cb..c380259 100644 --- a/orchestrator/api/app.py +++ b/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( diff --git a/orchestrator/api/progress.py b/orchestrator/api/progress.py index 586afe2..fbe9d27 100644 --- a/orchestrator/api/progress.py +++ b/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,7 +237,7 @@ 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(), @@ -244,7 +245,7 @@ def describe_progress(root: Path) -> dict: "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='')}" diff --git a/orchestrator/supervisor.py b/orchestrator/supervisor.py index 5bb69ef..1734b7b 100644 --- a/orchestrator/supervisor.py +++ b/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,16 +246,11 @@ 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 @@ -200,9 +258,17 @@ def main(argv: list[str] | None = None) -> int: # 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) diff --git a/orchestrator/worker_presence.py b/orchestrator/worker_presence.py index 79fc20e..0e1fa03 100644 --- a/orchestrator/worker_presence.py +++ b/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, diff --git a/scripts/tests/test_orchestrator_api_submission.py b/scripts/tests/test_orchestrator_api_submission.py index 0313326..7ef6736 100644 --- a/scripts/tests/test_orchestrator_api_submission.py +++ b/scripts/tests/test_orchestrator_api_submission.py @@ -164,6 +164,81 @@ def check(): self.assertEqual(response.json()["status"], "blocked") self.assertEqual(check()["worker"]["status"], "unavailable") + def test_upload_only_worker_matches_automatic_options_and_progress(self): + self.assertEqual(self.upload("manual").status_code, 201) + self.assertEqual(self.upload("automatic", options='{"hands_off":true}').status_code, 201) + with WorkerPresence(self.root / "runs", uploads_only=True): + for automatic in (False, True): + run_id = "automatic" if automatic else "manual" + expected = "available" if automatic else "unavailable" + response = self.client.get(f"/api/pipeline/runs/{run_id}/progress", headers=AUTH) + self.assertEqual(response.status_code, 200, response.text) + self.assertEqual(response.json()["worker"]["status"], expected) + 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": automatic} + ) + self.assertEqual(response.status_code, 200, response.text) + self.assertEqual(response.json()["worker"]["status"], expected) + self.assertEqual(response.json()["ready"], automatic) + + def test_upload_only_supervisor_picks_up_a_later_api_submission(self): + from orchestrator import supervisor + from orchestrator.agent.contracts import AgentResult + from orchestrator.agent.harness import HarnessAgent, HarnessOutcome, HarnessPolicy + from orchestrator.journal import Journal + + self.assertEqual(self.upload("manual").status_code, 201) + processed = [] + passes = 0 + + class StopRehearsal(Exception): + pass + + def next_pass(_): + nonlocal passes + passes += 1 + if passes == 1: + self.assertEqual(processed, []) + response = self.upload("automatic", options='{"hands_off":true}') + self.assertEqual(response.status_code, 201, response.text) + else: + raise StopRehearsal + + def finish_without_model(_task, workspace, _prompt, **_kwargs): + manifest = json.loads((workspace / "run.json").read_text()) + processed.append(manifest["runId"]) + Journal(workspace).append( + "run.finished", status="failed", summary="Fixture ended without inference." + ) + return HarnessOutcome(AgentResult(status="completed", transcript_path="inert")) + + 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(supervisor.time, "sleep", side_effect=next_pass), + patch.object(HarnessAgent, "run", side_effect=finish_without_model), + self.assertRaises(StopRehearsal), + ): + supervisor.main(["--runs", str(self.root / "runs"), "--uploads-only", "--watch"]) + self.assertEqual(processed, ["automatic"]) + progress = self.client.get("/api/pipeline/runs/automatic/progress", headers=AUTH).json() + self.assertEqual(progress["finished"], "failed") + self.assertIsNone(progress["question"]) + self.assertEqual(progress["worker"]["status"], "unavailable") + self.assertEqual(self.repository.run_summary("automatic")["attempts"], []) + manual = self.client.get("/api/pipeline/runs/manual/progress", headers=AUTH).json() + self.assertIsNone(manual["finished"]) + self.assertFalse(manual["currentStages"]) + 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_worker.py b/scripts/tests/test_upload_worker.py new file mode 100644 index 0000000..75c1b02 --- /dev/null +++ b/scripts/tests/test_upload_worker.py @@ -0,0 +1,276 @@ +"""The upload service is confined to explicitly automatic run manifests.""" + +import json +import os + +import pytest + +from orchestrator import supervisor +from orchestrator.agent.harness import HarnessPolicy +from orchestrator.journal import Journal +from orchestrator.worker_presence import WorkerPresence, worker_status + + +@pytest.fixture +def inert_worker(monkeypatch): + monkeypatch.delenv("WANDER_STAGE_AUTHORIZATION", raising=False) + monkeypatch.setattr(supervisor, "repository_from_environment", lambda: None) + monkeypatch.setattr( + HarnessPolicy, + "from_environment", + lambda: HarnessPolicy(kind="command", command=("unused",)), + ) + + +def automatic(root, name): + return supervisor.open_run( + root, run_id=name, name=name, source=None, options={"hands_off": True} + ) + + +@pytest.mark.parametrize( + "manifest", + [ + {"runId": "manual"}, + {"runId": "manual", "options": {}}, + {"runId": "manual", "options": {"hands_off": False}}, + {"runId": "manual", "options": {"hands_off": "true"}}, + {"runId": "manual", "options": {"hands_off": 1}}, + {"runId": "manual", "options": []}, + {"runId": "manual", "options": None}, + {"options": {"hands_off": True}}, + {"runId": [], "options": {"hands_off": True}}, + {"runId": "", "options": {"hands_off": True}}, + [], + None, + "invalid", + ], +) +def test_manual_legacy_and_malformed_runs_are_ignored( + tmp_path, monkeypatch, inert_worker, manifest +): + run = tmp_path / "clip" + run.mkdir() + (run / "run.json").write_text(json.dumps(manifest)) + malformed = tmp_path / "broken-json" + malformed.mkdir() + (malformed / "run.json").write_text("{incomplete") + monkeypatch.setattr(supervisor, "work_on", lambda *_: pytest.fail("out-of-scope work")) + assert supervisor.main(["--runs", str(tmp_path), "--uploads-only", "--once"]) == 0 + + +def test_symlink_directories_and_manifests_are_ignored(tmp_path): + root = tmp_path / "runs" + root.mkdir() + outside = automatic(tmp_path, "outside") + (root / "linked-run").symlink_to(outside, target_is_directory=True) + linked_manifest = root / "linked-manifest" + linked_manifest.mkdir() + (linked_manifest / "run.json").symlink_to(outside / "run.json") + eligible = automatic(root, "eligible") + assert supervisor.runs_needing_work(root, uploads_only=True) == [eligible] + + +@pytest.mark.parametrize("state", ["paused", "canceled", "succeeded", "failed"]) +def test_automatic_run_control_and_terminal_states_are_preserved(tmp_path, state): + run = automatic(tmp_path, "automatic") + if state == "paused": + (run / state).touch() + else: + Journal(run).append("run.finished", status=state) + assert supervisor.runs_needing_work(tmp_path, uploads_only=True) == [] + + +def test_automatic_upload_arriving_after_startup_is_picked_up_next_pass( + tmp_path, monkeypatch, inert_worker +): + supervisor.open_run(tmp_path, run_id="manual", name="manual", source=None) + calls = [] + passes = 0 + + class StopWatch(Exception): + pass + + def tick(_): + nonlocal passes + passes += 1 + assert worker_status(tmp_path, hands_off=True)["status"] == "available" + if passes == 1: + assert calls == [] + automatic(tmp_path, "new-upload") + else: + raise StopWatch + + monkeypatch.setattr(supervisor.time, "sleep", tick) + monkeypatch.setattr( + supervisor, + "work_on", + lambda path, *_: calls.append(path) or supervisor.WorkOutcome("done", 0), + ) + with pytest.raises(StopWatch): + supervisor.main(["--runs", str(tmp_path), "--uploads-only"]) + assert calls == [tmp_path / "new-upload"] + + +@pytest.mark.parametrize("mutation", ["manual", "run-id", "symlink"]) +def test_scope_is_checked_again_immediately_before_work( + tmp_path, monkeypatch, inert_worker, mutation +): + run = automatic(tmp_path, "selected") + + class ChangedPresence(WorkerPresence): + def set_active(self, run_id): + super().set_active(run_id) + if run_id is not None: + manifest = run / "run.json" + if mutation == "symlink": + held = tmp_path / "outside.json" + manifest.rename(held) + manifest.symlink_to(held) + else: + record = json.loads(manifest.read_text()) + if mutation == "manual": + record["options"]["hands_off"] = False + else: + record["runId"] = "different" + manifest.write_text(json.dumps(record)) + + monkeypatch.setattr(supervisor, "WorkerPresence", ChangedPresence) + monkeypatch.setattr(supervisor, "work_on", lambda *_: pytest.fail("scope changed before work")) + assert supervisor.main(["--runs", str(tmp_path), "--uploads-only", "--run-id", "selected"]) == 0 + + +@pytest.mark.parametrize("permit", ["", "source-specific-permit.json"]) +def test_source_scoped_permit_fails_before_presence_or_harness(tmp_path, monkeypatch, permit): + monkeypatch.setenv("WANDER_STAGE_AUTHORIZATION", permit) + monkeypatch.setattr(supervisor, "WorkerPresence", lambda *_a, **_k: pytest.fail("registered")) + monkeypatch.setattr(HarnessPolicy, "from_environment", lambda: pytest.fail("loaded harness")) + monkeypatch.setattr( + supervisor, "repository_from_environment", lambda: pytest.fail("opened database") + ) + root = tmp_path / "not-created" + with pytest.raises(SystemExit) as error: + supervisor.main(["--runs", str(root), "--uploads-only", "--once"]) + assert error.value.code == 2 + assert not root.exists() + assert os.environ["WANDER_STAGE_AUTHORIZATION"] == permit + + +@pytest.mark.parametrize( + "mutation", + [ + {"incomplete": True}, + {"name": None}, + {"name": []}, + {"name": ""}, + {"source": None}, + {"source": []}, + {"repository": None}, + {"repository": []}, + {"repository": ""}, + {"goal": ["bad goal"]}, + {"retainedDelivery": []}, + {"retainedDelivery": "bad delivery"}, + {"options": {"hands_off": True, "people": "invalid"}}, + {"options": {"hands_off": True, "people": "2.0"}}, + {"options": {"hands_off": True, "people": 0}}, + {"options": {"hands_off": True, "people": 17}}, + {"options": {"hands_off": True, "marble": "invalid"}}, + {"options": {"hands_off": True, "marble": []}}, + {"options": {"hands_off": True, "all_people": []}}, + ], +) +def test_malformed_automatic_neighbor_cannot_stop_shared_worker( + tmp_path, monkeypatch, inert_worker, mutation +): + from orchestrator.agent.contracts import AgentResult + from orchestrator.agent.harness import HarnessAgent, HarnessOutcome + + malformed = automatic(tmp_path, "a-malformed") + manifest_path = malformed / "run.json" + manifest = json.loads(manifest_path.read_text()) + if mutation.get("incomplete"): + manifest = {"runId": "a-malformed", "options": {"hands_off": True}} + else: + manifest.update(mutation) + manifest_path.write_text(json.dumps(manifest)) + healthy = automatic(tmp_path, "z-healthy") + harness_calls = [] + + def agent_turn(self, messages, run_dir, brief, **kwargs): + harness_calls.append(run_dir) + assert run_dir == healthy + Journal(run_dir).append("run.finished", status="failed", summary="Inert test complete") + return HarnessOutcome(AgentResult(status="completed", transcript_path="inert")) + + monkeypatch.setattr(HarnessAgent, "run", agent_turn) + # Exercise real work_on and RunSession planning; only the provider boundary is inert. + assert supervisor.main(["--runs", str(tmp_path), "--uploads-only", "--once"]) == 0 + assert harness_calls == [healthy] + assert Journal(healthy).finished() == "failed" + assert not (malformed / "BRIEF.md").exists() + assert not (malformed / "bin").exists() + + +def test_empty_and_missing_source_remain_eligible_for_stage_preflight(tmp_path): + run = automatic(tmp_path, "automatic") + manifest = supervisor.run_manifest(run, uploads_only=True) + assert manifest is not None and manifest["source"] == "" + manifest["source"] = str(run / "missing.mp4") + (run / "run.json").write_text(json.dumps(manifest)) + assert supervisor.runs_needing_work(tmp_path, uploads_only=True) == [run] + + +@pytest.mark.parametrize("repository", ["absent", "", None]) +@pytest.mark.parametrize("people", [1, "2"]) +def test_legacy_manual_repository_fallback_and_integer_strings_still_run( + tmp_path, monkeypatch, inert_worker, repository, people +): + from orchestrator.agent.contracts import AgentResult + from orchestrator.agent.harness import HarnessAgent, HarnessOutcome + + run = supervisor.open_run(tmp_path, run_id="legacy", name="legacy", source=None) + manifest_path = run / "run.json" + manifest = json.loads(manifest_path.read_text()) + if repository == "absent": + del manifest["repository"] + else: + manifest["repository"] = repository + manifest["options"] = {"people": people} + manifest_path.write_text(json.dumps(manifest)) + assert supervisor.runs_needing_work(tmp_path) == [run] + assert supervisor.runs_needing_work(tmp_path, uploads_only=True) == [] + harness_calls = [] + + def agent_turn(self, messages, run_dir, brief, **kwargs): + harness_calls.append(run_dir) + Journal(run_dir).append("run.finished", status="failed", summary="Inert test complete") + return HarnessOutcome(AgentResult(status="completed", transcript_path="inert")) + + monkeypatch.setattr(HarnessAgent, "run", agent_turn) + assert supervisor.main(["--runs", str(tmp_path), "--once"]) == 0 + assert harness_calls == [run] + assert Journal(run).finished() == "failed" + + +@pytest.mark.parametrize("repository", ["absent", "", None]) +def test_retained_manual_manifest_requires_repository(tmp_path, repository): + run = supervisor.open_run(tmp_path, run_id="retained", name="retained", source=None) + manifest_path = run / "run.json" + manifest = json.loads(manifest_path.read_text()) + manifest["retainedDelivery"] = {"stage": "package_people"} + if repository == "absent": + del manifest["repository"] + else: + manifest["repository"] = repository + manifest_path.write_text(json.dumps(manifest)) + assert supervisor.runs_needing_work(tmp_path) == [] + + +def test_manual_people_value_that_session_cannot_consume_is_ignored(tmp_path): + run = supervisor.open_run(tmp_path, run_id="legacy", name="legacy", source=None) + manifest_path = run / "run.json" + manifest = json.loads(manifest_path.read_text()) + manifest["options"] = {"people": "2.0"} + manifest_path.write_text(json.dumps(manifest)) + assert supervisor.runs_needing_work(tmp_path) == [] diff --git a/scripts/tests/test_worker_presence.py b/scripts/tests/test_worker_presence.py index d5ac92a..18a20fd 100644 --- a/scripts/tests/test_worker_presence.py +++ b/scripts/tests/test_worker_presence.py @@ -68,7 +68,9 @@ def test_selected_watcher_never_advertises_new_uploads(tmp_path): {"last_seen": float("inf")}, {"last_seen": True}, {"last_seen": "now"}, - {"version": 2}, + {"version": 3}, + {"uploads_only": "true"}, + {"uploads_only": None}, {"version": True}, {"schema": "other"}, {"watcher": "true"}, @@ -177,10 +179,16 @@ def fail_write(): (["--run-id", "selected", "--watch"], False, True), ], ) +@pytest.mark.parametrize("uploads_only", [False, True]) def test_supervisor_refreshes_during_blocking_work_and_cleans_up( - tmp_path, monkeypatch, flags, generic_available, run_available + tmp_path, monkeypatch, flags, generic_available, run_available, uploads_only ): - supervisor.open_run(tmp_path, run_id="selected", name="clip", source=None) + supervisor.open_run( + tmp_path, run_id="selected", name="clip", source=None, options={"hands_off": uploads_only} + ) + monkeypatch.delenv("WANDER_STAGE_AUTHORIZATION", raising=False) + if uploads_only: + flags = [*flags, "--uploads-only"] heartbeat = tmp_path / "legacy-heartbeat" monkeypatch.setenv("WANDER_HEARTBEAT", str(heartbeat)) monkeypatch.setattr(supervisor, "repository_from_environment", lambda: None) @@ -207,9 +215,16 @@ def _write(self): def work(*_): nonlocal baseline - assert (worker_status(tmp_path)["status"] == "available") == generic_available - assert (worker_status(tmp_path, "selected")["status"] == "available") == run_available - assert worker_status(tmp_path, "selected")["busyWorkers"] == 1 + assert ( + worker_status(tmp_path, hands_off=True)["status"] == "available" + ) == generic_available + if uploads_only: + assert worker_status(tmp_path)["status"] == "unavailable" + assert worker_status(tmp_path, "selected")["status"] == "unavailable" + assert ( + worker_status(tmp_path, "selected", hands_off=True)["status"] == "available" + ) == run_available + assert worker_status(tmp_path, "selected", hands_off=True)["busyWorkers"] == 1 path = next((tmp_path / ".workers").glob("*.json")) baseline = json.loads(path.read_text())["last_seen"] # The main supervisor is blocked here; only its background heartbeat can advance. @@ -256,3 +271,50 @@ def fail_work(*_): with pytest.raises(ValueError, match="work failed"): supervisor.main(["--runs", str(tmp_path), "--once"]) assert list((tmp_path / ".workers").iterdir()) == [] + + +@pytest.mark.parametrize("selected", [None, "selected"]) +def test_uploads_only_capability_matches_automatic_requests_only(tmp_path, selected): + with WorkerPresence(tmp_path, run_id=selected, uploads_only=True) as presence: + assert worker_status(tmp_path)["status"] == "unavailable" + assert worker_status(tmp_path, "selected")["status"] == "unavailable" + assert worker_status(tmp_path, hands_off=True)["matchingWorkers"] == (selected is None) + assert worker_status(tmp_path, "selected", hands_off=True)["matchingWorkers"] == 1 + assert worker_status(tmp_path, "other", hands_off=True)["matchingWorkers"] == ( + selected is None + ) + 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["uploads_only"] is True + + +@pytest.mark.parametrize("hands_off", [False, True]) +def test_historical_version_one_worker_is_unrestricted(tmp_path, hands_off): + with WorkerPresence(tmp_path) as presence: + path = tmp_path / ".workers" / presence.name + record = json.loads(path.read_text()) + record["version"] = 1 + del record["uploads_only"] + path.write_text(json.dumps(record)) + assert worker_status(tmp_path, hands_off=hands_off)["matchingWorkers"] == 1 + assert ( + worker_status(tmp_path, "manual-or-automatic", hands_off=hands_off)["matchingWorkers"] + == 1 + ) + + +def test_version_two_requires_explicit_capability(tmp_path): + with WorkerPresence(tmp_path) as presence: + path = tmp_path / ".workers" / presence.name + record = json.loads(path.read_text()) + del record["uploads_only"] + path.write_text(json.dumps(record)) + assert worker_status(tmp_path)["status"] == "unavailable" + assert worker_status(tmp_path, hands_off=True)["status"] == "unavailable" + + +@pytest.mark.parametrize("value", [None, 0, 1, "true"]) +def test_presence_rejects_non_boolean_capability(tmp_path, value): + with pytest.raises(ValueError, match="scope"): + WorkerPresence(tmp_path, uploads_only=value)