fix(orchestrator): stop hands-off paid work once the package is unreachable

docs/automatic-uploads.md

@@ -97,6 +97,21 @@ change the submitted run through extra stage flags or automatically retry a paid
 Existing provider ledgers and unresolved-operation guards continue to apply. A known provider
 operation must be recovered separately rather than submitted again.
 
+Once a paid stage that the package depends on fails or is blocked, the run cannot produce a
+preview, because hands-off mode never repeats a paid stage. The next paid launch is then
+refused and the run finishes `failed`, so it does not spend on a world generation or avatars
+that can no longer be packaged. Stages already running finish and keep their outputs. While an
+unpaid package prerequisite (such as `person_prep`) is failed, paid launches are held instead:
+rerunning that step is still admitted, and paid work resumes only once it succeeds. Unpaid steps
+use the same fixed defaults on a rerun, so a failure caused by the footage itself, such as no
+fully visible person, repeats and the run ends without spending on generation.
+
+The shared stage ledger counts executions per source SHA-256 across all runs, including
+historical imports. A second execution needs an explicit new hypothesis, which a hands-off run
+cannot supply. Uploading a video that was already processed therefore blocks at its first
+previously executed paid stage. Use a source that has not been processed before. An upload
+does not reuse earlier outputs or worlds.
+
 When the agent needs a decision, exhausts its useful work, or stalls, the run records a final
 failure with its reason instead of waiting for a question in the UI. Existing outputs and
 evidence remain available. A readable packaged scene may be delivered as an **unaccepted

orchestrator/cli.py

@@ -31,6 +31,7 @@
     delivery_metadata,
     finish_hands_off,
     hands_off,
+    unpaid_prerequisite_problem,
 )
 from orchestrator.steps import (
     SAME_AS,
@@ -295,6 +296,10 @@ def admit_step(
         finish_hands_off(context, journal, problem)
         print(problem, file=sys.stderr)
         return 2
+    if not args.attempt and (held := unpaid_prerequisite_problem(context, args.step, journal)):
+        # Not terminal: the unpaid prerequisite may still succeed when rerun.
+        print(held, file=sys.stderr)
+        return 2
     if context.retained_delivery is not None:
         problem = retained_delivery_problem(context, args, journal)
         if problem:

orchestrator/run_policy.py

@@ -11,7 +11,7 @@
 from typing import Any
 
 from orchestrator.journal import Journal
-from orchestrator.steps import STEPS, base_step
+from orchestrator.steps import STEPS, base_step, stage_graph
 
 
 def hands_off(options: dict[str, Any]) -> bool:
@@ -53,6 +53,59 @@ def admission_problem(context, step: str, extra: list[str], journal: Journal, at
                 f"Hands-off paid step {step} already has an attempt or retained result. "
                 "Automatic paid retries are disabled; retain its outputs and provider receipt."
             )
+        if lost := failed_package_prerequisite(context, journal, paid=True):
+            return (
+                f"Hands-off paid step {step} was not started: required paid step {lost} cannot "
+                "run again, so no packaged preview can follow and further paid work is wasted."
+            )
+    return None
+
+
+def unpaid_prerequisite_problem(context, step: str, journal: Journal) -> str | None:
+    """Hold paid work while an unpaid prerequisite of the package is failed.
+
+    Unlike a paid failure this is not final: rerunning the unpaid step is admitted, and paid
+    work resumes once it succeeds. A failure that repeats leaves the run without runnable work.
+    """
+    if not hands_off(context.options) or not STEPS[base_step(step)].paid:
+        return None
+    if failed := failed_package_prerequisite(context, journal, paid=False):
+        return (
+            f"Hands-off paid step {step} was not started: required step {failed} has not "
+            "succeeded, so no packaged preview can follow yet. Rerun it first."
+        )
+    return None
+
+
+def failed_package_prerequisite(context, journal: Journal, *, paid: bool) -> str | None:
+    """A paid (or unpaid) prerequisite of the package whose latest outcome is a failure.
+
+    Hands-off runs never repeat a paid step, so a paid failure is final: the package, and with
+    it any preview, can no longer be produced. An unpaid step can still be rerun.
+    """
+    stages, deps = stage_graph(context.options)
+    package = "package_people" if "package_people" in stages else "package"
+    required, pending = set(), [package]
+    while pending:
+        name = pending.pop()
+        if name not in required:
+            required.add(name)
+            pending.extend(deps.get(name, []))
+    outcomes = {
+        entry.data.get("step"): entry.data.get("status")
+        for entry in journal.entries()
+        if entry.kind == "step.finished"
+    }
+    for name, record in stage_records(context.run_dir).items():
+        if isinstance(record, dict) and record.get("status") == "failed":
+            outcomes.setdefault(name, "failed")
+    for name in stages:
+        if (
+            name in required
+            and STEPS[base_step(name)].paid == paid
+            and outcomes.get(name) in {"failed", "blocked"}
+        ):
+            return f"{name} ({outcomes[name]})"
     return None
 
 

scripts/tests/test_hands_off_runs.py

@@ -14,7 +14,11 @@
 from orchestrator.agent.contracts import AgentResult
 from orchestrator.agent.harness import HarnessAgent, HarnessOutcome, HarnessPolicy
 from orchestrator.journal import Journal
-from orchestrator.run_policy import delivery_metadata
+from orchestrator.run_policy import (
+    admission_problem,
+    delivery_metadata,
+    unpaid_prerequisite_problem,
+)
 from orchestrator.session import RunSession
 from orchestrator.steps import RunOptions
 from orchestrator.supervisor import runs_needing_work, work_on
@@ -158,6 +162,46 @@ def test_paid_retry_is_refused_after_restart_and_resume(run, monkeypatch, record
     popen.assert_not_called()
 
 
+@pytest.mark.parametrize("status", ["blocked", "failed"])
+def test_lost_package_prerequisite_stops_further_paid_work(run, monkeypatch, status):
+    journal = run.journal()
+    journal.append("step.started", step="pi3x", attempt="one", number=1)
+    journal.append("step.finished", step="pi3x", attempt="one", status=status)
+    popen = Mock(
+        side_effect=AssertionError("paid work launched after the package became impossible")
+    )
+    monkeypatch.setattr(cli.subprocess, "Popen", popen)
+    assert cli.main(["step", "--wait", "lhm_frozen"]) == 2
+    finished = Journal(run.run_dir).entries()[-1]
+    assert finished.kind == "run.finished" and finished.data["status"] == "failed"
+    assert "pi3x" in finished.data["summary"]
+    popen.assert_not_called()
+
+
+def test_lost_prerequisite_check_ignores_unpaid_and_unrelated_failures(run):
+    journal = run.journal()
+    # Unpaid steps can be rerun; clean does not feed the package of a run without a world.
+    journal.append("step.finished", step="frame_align", attempt="one", status="failed")
+    journal.append("step.finished", step="clean", attempt="one", status="failed")
+    assert admission_problem(run, "lhm_frozen", [], journal) is None
+    (run.run_dir / "state.json").write_text('{"stages":{"lhm_frozen":{"status":"failed"}}}')
+    assert "lhm_frozen" in admission_problem(run, "pi3x", [], journal)
+
+
+def test_failed_unpaid_prerequisite_holds_paid_work_until_it_succeeds(run, monkeypatch):
+    journal = run.journal()
+    journal.append("step.finished", step="person_prep", attempt="one", status="failed")
+    popen = Mock(side_effect=AssertionError("paid work launched without its prerequisite"))
+    monkeypatch.setattr(cli.subprocess, "Popen", popen)
+    assert cli.main(["step", "--wait", "lhm_frozen"]) == 2
+    # Held, not finished: the unpaid step may still be rerun.
+    assert Journal(run.run_dir).finished() is None
+    popen.assert_not_called()
+    assert unpaid_prerequisite_problem(run, "person_prep", journal) is None
+    journal.append("step.finished", step="person_prep", attempt="two", status="succeeded")
+    assert unpaid_prerequisite_problem(run, "lhm_frozen", journal) is None
+
+
 def test_existing_detached_attempt_can_claim_execution_once(run):
     journal = run.journal()
     journal.append(