feat(orchestrator): dev-only scripted local outcomes for stage stubs

backend/server/orchestrator/README.md

@@ -33,4 +33,18 @@ make ci
 
 ## Stubs
 
-`HSS_ORCH_STAGE_STUBS=1` runs `server/orchestrator/stubs/<stage>.py` instead of `server.<stage>.run`. Stub verdicts are `manual_review` with empty `views_needed`.
+`HSS_ORCH_STAGE_STUBS=1` runs `server/orchestrator/stubs/<stage>.py` instead of `server.<stage>.run`. Stub verdicts are `manual_review` with empty `views_needed`. The stub review page says every stage is a stand-in.
+
+### Scripted local outcomes (dev only)
+
+With stage stubs on and `SMP_ENVIRONMENT=dev`, `HSS_LOCAL_OUTCOME` scripts the terminal state a stubbed local run reaches — the app can be walked to each state it renders, end to end, without Modal or GPU:
+
+| `HSS_LOCAL_OUTCOME` | Capture status | Notes |
+| --- | --- | --- |
+| unset / `manual_review` (default) | `manual_review` | Today's stub run, byte-identical. |
+| `eligible` | `complete` | Result carries a proposed placement (`boxArkitWorld` + `arkitEpoch` when the packet has one), worded as proposed. |
+| `needs_more_photos` | `needs_views` | One scripted band view; the app sends a retake request. |
+| `not_eligible` | `complete` | Needs `SHOW_REJECT_TO_HOMEOWNER=1`; without it the criteria stub refuses and the run fails loudly, leaving the capture on `manual_review`. |
+| `failed` | `failed` | The validate stub fails the run onto outcome row 1 (validation stand-in `imagesUsable: false`). |
+
+Scripted messages and review pages always say it is a local test outcome, not a check of a wall. `HSS_LOCAL_STAGE_DELAY_S` (float, 0–30, default 0) makes each stub sleep so the app shows processing events. Both variables are refused loudly — with an error naming the variable, at config load — when stage stubs are off, the environment is not dev, the value is unknown, or the number is out of range. A run keeps its scripted behavior across restart while the variables stay set.

backend/server/orchestrator/config.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import math
 import os
 
 from pydantic import BaseModel, ConfigDict
@@ -11,6 +12,57 @@ def _flag(value: str) -> bool:
     return value.strip().lower() in {"1", "true", "yes", "on"}
 
 
+_LOCAL_OUTCOMES = (
+    "manual_review",
+    "eligible",
+    "needs_more_photos",
+    "not_eligible",
+    "failed",
+)
+
+
+def _dev_stubs(env: dict[str, str]) -> bool:
+    """Scripted local outcomes are honored only under dev stage stubs."""
+    return _flag(env.get("HSS_ORCH_STAGE_STUBS", "0")) and env.get("SMP_ENVIRONMENT", "prod") == "dev"
+
+
+def _require_dev_stubs(name: str, env: dict[str, str]) -> None:
+    if _dev_stubs(env):
+        return
+    raise ValueError(
+        f"{name} is only honored with HSS_ORCH_STAGE_STUBS=1 and SMP_ENVIRONMENT=dev "
+        f"(got HSS_ORCH_STAGE_STUBS={env.get('HSS_ORCH_STAGE_STUBS', '0')!r}, "
+        f"SMP_ENVIRONMENT={env.get('SMP_ENVIRONMENT', '(unset)')!r}); "
+        "refusing to run with it silently ignored"
+    )
+
+
+def _local_outcome(env: dict[str, str]) -> str:
+    raw = env.get("HSS_LOCAL_OUTCOME", "")
+    if not raw:
+        return "manual_review"
+    if raw not in _LOCAL_OUTCOMES:
+        raise ValueError(f"HSS_LOCAL_OUTCOME={raw!r} is not one of {', '.join(_LOCAL_OUTCOMES)}")
+    _require_dev_stubs("HSS_LOCAL_OUTCOME", env)
+    return raw
+
+
+def _local_stage_delay_s(env: dict[str, str]) -> float:
+    raw = env.get("HSS_LOCAL_STAGE_DELAY_S", "")
+    if not raw:
+        return 0.0
+    try:
+        value = float(raw)
+    except ValueError:
+        raise ValueError(f"HSS_LOCAL_STAGE_DELAY_S={raw!r} is not a number of seconds") from None
+    if not math.isfinite(value):
+        raise ValueError(f"HSS_LOCAL_STAGE_DELAY_S={raw!r} is not a finite number of seconds")
+    if not 0 <= value <= 30:
+        raise ValueError(f"HSS_LOCAL_STAGE_DELAY_S={raw!r} is outside the range 0 to 30 seconds")
+    _require_dev_stubs("HSS_LOCAL_STAGE_DELAY_S", env)
+    return value
+
+
 class OrchConfig(BaseModel):
     model_config = ConfigDict(extra="ignore")
 
@@ -35,6 +87,10 @@ class OrchConfig(BaseModel):
     # cancels a live worker within 15s of the deadline.
     coordinator_loop_s: float = 1
     modal_remote: bool = False
+    # Dev-only scripted outcome for the stage stubs; see the README's Stubs
+    # section. Both are refused loudly outside dev stage stubs.
+    local_outcome: str = "manual_review"
+    local_stage_delay_s: float = 0.0
 
 
 def load_orch(environ: dict[str, str] | None = None) -> OrchConfig:
@@ -63,4 +119,6 @@ def pick(name: str, default: str) -> str:
         test_kill_coordinator_after=pick("HSS_ORCH_TEST_KILL_COORDINATOR_AFTER", ""),
         coordinator_loop_s=float(pick("HSS_ORCH_COORDINATOR_LOOP_S", "1")),
         modal_remote=_flag(pick("HSS_ORCH_MODAL_REMOTE", "0")),
+        local_outcome=_local_outcome(env),
+        local_stage_delay_s=_local_stage_delay_s(env),
     )

backend/server/orchestrator/stubs/_common.py

@@ -5,16 +5,23 @@
 import hashlib
 import io
 import json
+import time
 from pathlib import Path
 
 import numpy as np
 
 from server.contracts.jobs import StageContext, StageResult
+from server.orchestrator.config import load_orch
 
 _PLY = b"ply\nformat ascii 1.0\nelement vertex 1\nproperty float x\nproperty float y\nproperty float z\nend_header\n0 0 0\n"
 
 
 def write_files(ctx: StageContext, files: dict[str, bytes]) -> StageResult:
+    # The dev-only scripted delay: each stub sleeps so the app sees the stage
+    # move through processing. load_orch refuses a stray value at boot.
+    delay = load_orch().local_stage_delay_s
+    if delay > 0:
+        time.sleep(delay)
     artifacts: dict[str, str] = {}
     for key, data in files.items():
         path = ctx.work_dir / key
@@ -118,5 +125,41 @@ def sha256(data: bytes) -> str:
     return hashlib.sha256(data).hexdigest()
 
 
+def scripted_outcome() -> str:
+    """The dev-only scripted terminal state, or the default manual_review."""
+    return load_orch().local_outcome
+
+
+_SCRIPTED_VIEW = {
+    "id": "stub-wall-band",
+    "kind": "band",
+    "criteria": ["stub"],
+    "why": "Local test stand-in: this retake request is scripted, not measured.",
+    "band": "wall",
+    "wall_id": "w1",
+}
+
+
+def scripted_verdict(profile: str, scene_id: str, outcome: str) -> dict:
+    """The stub verdict for a scripted local outcome (dev + stage stubs only).
+
+    Changes only the site decision and views_needed, so the outcome table's
+    rows 11-14 land the capture on the chosen state. manual_review keeps the
+    plain stub document byte-identical, and failed never reaches criteria
+    (the validate stub ends the run on row 1).
+    """
+    verdict = verdict_document(profile, scene_id)
+    if outcome == "eligible":
+        verdict["site"]["decision"] = "pass"
+        verdict["site"]["reasons"] = ["scripted:eligible"]
+    elif outcome == "not_eligible":
+        verdict["site"]["decision"] = "reject"
+        verdict["site"]["reasons"] = ["scripted:not_eligible"]
+    elif outcome == "needs_more_photos":
+        verdict["site"]["reasons"] = ["scripted:needs_more_photos"]
+        verdict["views_needed"] = [dict(_SCRIPTED_VIEW)]
+    return verdict
+
+
 def repo_root() -> Path:
     return Path(__file__).resolve().parents[3]

backend/server/orchestrator/stubs/criteria.py

@@ -1,16 +1,37 @@
-"""Stub ``criteria`` stage. Decisions stay ``manual_review`` so a stub cannot be an unsafe pass."""
+"""Stub ``criteria`` stage. Decisions stay ``manual_review`` so a stub cannot be an unsafe pass.
+
+A scripted local outcome (``HSS_LOCAL_OUTCOME``, dev + stage stubs only) is the
+one exception: it writes the decision the outcome table's rows 11-14 need to
+land the capture on the chosen state. Nothing else changes.
+"""
 
 from __future__ import annotations
 
 from server.contracts.jobs import StageContext, StageResult
 from server.eval.sceneio import schema_errors
-from server.orchestrator.stubs._common import json_bytes, verdict_document, write_files
+from server.orchestrator.stubs._common import (
+    json_bytes,
+    scripted_outcome,
+    scripted_verdict,
+    verdict_document,
+    write_files,
+)
 
 
 def run(ctx: StageContext) -> StageResult:
+    outcome = scripted_outcome()
+    if outcome == "not_eligible" and not getattr(ctx.settings, "SHOW_REJECT_TO_HOMEOWNER", False):
+        raise RuntimeError(
+            "HSS_LOCAL_OUTCOME=not_eligible needs SHOW_REJECT_TO_HOMEOWNER=1; "
+            "with the reject hidden the capture would land manual_review instead"
+        )
     files = {}
     for profile in ("A", "B", "C", "S"):
-        verdict = verdict_document(profile, ctx.capture_id)
+        verdict = (
+            verdict_document(profile, ctx.capture_id)
+            if outcome in ("manual_review", "failed")
+            else scripted_verdict(profile, ctx.capture_id, outcome)
+        )
         errors = schema_errors(verdict)
         if errors:
             raise RuntimeError("artifact_invalid:criteria " + "; ".join(errors[:5]))