feat(placement): solve each capture's merged scene instead of a reference

backend/server/contracts/jobs.py

@@ -68,9 +68,9 @@
     "coverage": ("dense",),
     "scene": ("coverage", "objects"),
     "objects_3d": ("scene", "objects"),
-    "placement": ("scene",),
+    "placement": ("scene", "objects_3d"),
     "criteria": ("scene", "reads", "objects_3d"),
-    "result": ("criteria",),
+    "result": ("criteria", "placement"),
 }
 OPTIONAL_STAGES: frozenset[str] = frozenset({"objects", "reads", "objects_3d"})
 

backend/server/orchestrator/graph.py

@@ -154,13 +154,17 @@ def _gpu(
             entry="run_3d",
         ),
         # The real placement stage (server/placement/): the repo's unchanged top-level
-        # solver behind `POST /v1/placements`, run on a wire-format scene. Optional: it
-        # adds a decision to the run and degrades (`degraded:placement`, capture
-        # `manual_review`) instead of sinking it. It is exempt from stage stubs — the
-        # only real stage on the dev path — because it is deterministic CPU work.
+        # solver behind `POST /v1/placements`, run on the wire-format scene the bridge
+        # converts from the run's merged scene. Placement waits for `objects_3d` to be
+        # terminal (not just `scene` done) so it only ever solves the scene the
+        # coordinator has already merged — a missing or failed objects_3d stage refuses
+        # placement instead of implying no hazards. Optional: it adds a decision to the
+        # run and degrades (`degraded:placement`, capture `manual_review`) instead of
+        # sinking it. It is exempt from stage stubs — the only real stage on the dev
+        # path — because it is deterministic CPU work.
         StageSpec(
             name="placement",
-            deps=(("scene", _D),),
+            deps=(("scene", _D), ("objects_3d", _T)),
             where="cpu",
             timeout_s=120,
             max_attempts=2,
@@ -185,7 +189,7 @@ def _gpu(
         ),
         _cpu(
             "result",
-            (("criteria", _D),),
+            (("criteria", _D), ("placement", _T)),
             60,
             ("result/result.json", "result/result.internal.json", "result/review.html", "result/preview.ply"),
         ),
@@ -238,10 +242,13 @@ def _gpu(
         "poses/registration_report.json",
     ),
     "placement": (
-        # A wire-format scene, the same bytes POST /v1/placements accepts. Nothing
-        # publishes this key yet (the bridge from Hunter's scene.json is future
-        # work), so the stage falls back to its committed reference fixture.
-        "placement/scene.json",
+        # The run's merged source scene (metres) and the objects_3d artifact. The
+        # coordinator's `_merge_if_ready` merges detected objects into scene/scene.json
+        # before `_spawn_ready` can start placement, and placement's deps require
+        # objects_3d terminal, so hydration always carries the merged bytes here. The
+        # objects_3d artifact is readiness evidence only — placement never merges it.
+        "scene/scene.json",
+        "objects/objects_3d.json",
     ),
     "criteria": (
         "scene/",

backend/server/placement/adapter.py

@@ -12,21 +12,36 @@
 import cannot silently succeed — every path is pinned under the repository root.
 
 Stage contract (`run`):
-- Input: `placement/scene.json` in the run artifacts — a wire-format scene, the same
-  bytes `POST /v1/placements` accepts. Nothing publishes that key yet (the bridge from
-  Hunter's scene output is future work), so the stage falls back to the committed
-  reference fixture `reference_scene.json`, a byte-identical copy of the top-level
-  server's `tests/fixtures/example-scene.json`; the fallback is recorded as a stage
-  warning.
+- Inputs (from the run's artifacts, hydrated under the work directory): the merged
+  source scene `scene/scene.json` (backend `base-scene/v0` metres — the coordinator's
+  `_merge_if_ready` merges the detected objects into it before placement can start)
+  and `objects/objects_3d.json` as readiness evidence only. There is no reference
+  fallback: a run without these bytes refuses placement rather than solving a scene
+  that is not this capture's.
+- Readiness gate: the `objects_3d` stage record must be `done` and its artifact must
+  carry `format = hss-objects-3d/v0`. A missing, failed or skipped `objects_3d` stage
+  refuses placement (`placement_objects_3d_unavailable`, not retryable) — a run whose
+  hazard pass did not finish must degrade to manual review, never imply no hazards.
+  A successful explicit empty object list is valid.
+- Conversion: `placement.scene_bridge.convert_scene` validates the merged `Scene`
+  (duplicate ids, unsupported or generated evidence, self-contradictory geometry) and
+  converts it to the wire scene exactly once — the adapter never merges objects into
+  the scene again. The converted document is serialized deterministically (sorted
+  keys, non-finite values forbidden) and those exact bytes are what the unchanged
+  solver parses, what `stats.input_sha256` binds to, and what `placement/scene.json`
+  publishes. The source scene artifact is never written back.
 - Output: `placement/placement.json` — the solver's result verbatim (sorted keys), valid
   against the top-level `schemas/result.schema.json`; `placement/scene.json` — the
-  exact input bytes. Only `stats.elapsed_ms` varies between runs: the decision is
-  deterministic for fixed scene bytes and rules.
-- Failure: a malformed or unplaceable scene raises `PlacementSceneError` (code
-  `placement_scene_invalid`, not retryable — the bytes do not heal on a retry). A
-  missing solver tree raises `PlacementBoundaryError` (code `placement_boundary_missing`).
-  Both are recorded stage data; the optional stage degrades the run instead of failing
-  it.
+  exact converted input bytes. Only `stats.elapsed_ms` varies between runs: the decision
+  is deterministic for fixed scene bytes and rules.
+- Failure: a missing input artifact raises `PlacementInputError` (code
+  `placement_scene_input_missing`); a malformed or unplaceable scene raises
+  `PlacementSceneError` (code `placement_scene_invalid`); a scene the bridge cannot
+  support raises `SceneBridgeError` carrying `placement_scene_unsupported` or
+  `placement_scene_invalid`; a missing solver tree raises `PlacementBoundaryError`
+  (code `placement_boundary_missing`). All are not retryable (the bytes do not heal on
+  a retry) and are recorded stage data; the optional stage degrades the run instead of
+  failing it.
 """
 
 from __future__ import annotations
@@ -41,6 +56,7 @@
 from typing import Any
 
 from server.contracts.jobs import StageContext, StageResult
+from server.placement.scene_bridge import convert_scene
 
 # backend/server/placement/adapter.py -> backend/server/placement -> backend/server
 # -> backend -> repository root. The top-level placement server lives beside backend/.
@@ -77,6 +93,24 @@ def __init__(self, message: str) -> None:
         self.code = "placement_scene_invalid"
 
 
+class PlacementInputError(Exception):
+    """A required source artifact is absent from the run. Not retryable: the stage
+    refuses rather than solving a scene that is not this capture's."""
+
+    def __init__(self, message: str) -> None:
+        super().__init__(message)
+        self.code = "placement_scene_input_missing"
+
+
+class PlacementObjectsError(Exception):
+    """The run's objects_3d pass did not finish, so the scene's hazards are not known
+    to be complete. Not retryable: placement refuses instead of implying no hazards."""
+
+    def __init__(self, message: str) -> None:
+        super().__init__(message)
+        self.code = "placement_objects_3d_unavailable"
+
+
 def _top_level_server_dir() -> Path:
     return _REPO_ROOT / "server"
 
@@ -164,51 +198,92 @@ def _loaded_rules(modules: dict[str, ModuleType]) -> Any:
     return _RULES
 
 
-def _scene_input(ctx: StageContext) -> tuple[bytes, bool]:
-    """The scene bytes and whether they came from the run artifact (as opposed to the
-    committed reference fixture)."""
+_SOURCE_SCENE = "scene/scene.json"
+_OBJECTS_3D = "objects/objects_3d.json"
+
+
+def _read_input(ctx: StageContext, key: str) -> bytes:
+    """The hydrated run artifact's bytes. Missing means refusal, never a fallback."""
+
+    path = Path(ctx.work_dir) / key
+    if not path.is_file():
+        raise PlacementInputError(
+            f"{key} is missing from this run's artifacts: placement only solves the "
+            "scene this capture produced (STAGE_INPUTS['placement'] hydrates it)"
+        )
+    return path.read_bytes()
 
-    # stage_context hydrates the stage's selected run artifacts as files under the
-    # work directory (STAGE_INPUTS["placement"] admits "placement/scene.json");
-    # ctx.inputs itself only carries packet_dir and run_control.
-    artifact = Path(ctx.work_dir) / "placement" / "scene.json"
-    if artifact.is_file():
-        return artifact.read_bytes(), True
-    fixture = Path(__file__).resolve().parent / "reference_scene.json"
-    return fixture.read_bytes(), False
+
+def _objects_3d_evidence(ctx: StageContext) -> None:
+    """The objects_3d readiness gate: stage done, artifact the declared format.
+
+    Readiness evidence only — the coordinator's `_merge_if_ready` has already merged
+    the detected objects into the source scene, and this stage never merges them again.
+    """
+
+    record = ctx.state.get_stage(ctx.run_id, "objects_3d")
+    status = getattr(record, "status", None)
+    if status != "done":
+        raise PlacementObjectsError(
+            f"objects_3d stage status is {status or 'missing'!r}, not done: the scene's "
+            "hazards may be incomplete, so placement refuses rather than implying no hazards"
+        )
+    raw = _read_input(ctx, _OBJECTS_3D)
+    try:
+        document = json.loads(raw.decode("utf-8"))
+    except (UnicodeDecodeError, ValueError) as exc:
+        raise PlacementObjectsError(f"{_OBJECTS_3D} is not readable JSON: {exc}") from exc
+    fmt = document.get("format") if isinstance(document, dict) else None
+    if fmt != "hss-objects-3d/v0":
+        raise PlacementObjectsError(f'{_OBJECTS_3D} must carry format "hss-objects-3d/v0", got {fmt!r}')
+
+
+def _source_scene_bytes(ctx: StageContext) -> bytes:
+    """The merged source scene's bytes, JSON-parsed and re-serialized deterministically
+    after conversion — those converted bytes are the solver's whole world."""
+
+    raw = _read_input(ctx, _SOURCE_SCENE)
+    try:
+        document = json.loads(raw.decode("utf-8"))
+    except (UnicodeDecodeError, ValueError) as exc:
+        raise PlacementSceneError(f"source scene {_SOURCE_SCENE} is not readable JSON: {exc}") from exc
+    # Validates the merged Scene (duplicate ids, generated or unsupported evidence,
+    # self-contradictory geometry) and converts to the wire scene. SceneBridgeError
+    # carries its own typed code and JSON pointer; it is a ValueError, so it must not
+    # be caught by the solver's ValueError handler below.
+    converted = convert_scene(document)
+    # Sorted keys, no NaN/Inf: deterministic bytes, and a non-finite value fails here
+    # instead of reaching the solver.
+    return json.dumps(converted, sort_keys=True, allow_nan=False).encode("utf-8")
 
 
 def run(ctx: StageContext) -> StageResult:
-    """Orchestrator stage entry: solve the scene, publish the artifact bytes verbatim."""
+    """Orchestrator stage entry: convert this capture's merged scene, solve it,
+    publish the artifact bytes verbatim."""
 
     modules = load_solver()
     scene_mod, solver_mod = modules["scene"], modules["solver"]
-    raw, from_artifact = _scene_input(ctx)
+    _objects_3d_evidence(ctx)
+    scene_bytes = _source_scene_bytes(ctx)
     try:
-        document = json.loads(raw.decode("utf-8"))
         rules = _loaded_rules(modules)
-        parsed = scene_mod.parse_scene(document, rules.rules, raw)
+        parsed = scene_mod.parse_scene(json.loads(scene_bytes.decode("utf-8")), rules.rules, scene_bytes)
         result = solver_mod.solve(parsed, rules)
         text = json.dumps(result, sort_keys=True, allow_nan=False)
     except scene_mod.SceneError as exc:  # schema and geometry refusals from the parser
         raise PlacementSceneError(str(exc)) from exc
-    except ValueError as exc:  # malformed JSON, SceneTooComplex, non-finite lengths
+    except ValueError as exc:  # SceneTooComplex, non-finite lengths
         raise PlacementSceneError(str(exc)) from exc
 
     out_dir = Path(ctx.work_dir) / "placement"
     out_dir.mkdir(parents=True, exist_ok=True)
     (out_dir / "placement.json").write_bytes(text.encode("utf-8"))
-    (out_dir / "scene.json").write_bytes(raw)
+    (out_dir / "scene.json").write_bytes(scene_bytes)
     return StageResult(
         artifacts={
             "placement/placement.json": "placement/placement.json",
             "placement/scene.json": "placement/scene.json",
         },
         metrics={"elapsed_ms": float(result["stats"]["elapsed_ms"])},
-        warnings=[]
-        if from_artifact
-        else [
-            "scene_input: committed reference fixture (no placement/scene.json run "
-            "artifact published yet; the bridge from Hunter's scene output is future work)"
-        ],
+        warnings=[],
     )

backend/tests/test_backend_w6_gate.py

@@ -244,9 +244,12 @@ def test_staged_placement_result_is_schema_valid_and_pinned_to_the_fixture() ->
     assert PLACEMENT["stats"]["elapsed_ms"] > 0
 
 
-def test_a_fresh_front_door_journey_matches_the_staged_receipt(monkeypatch, tmp_path) -> None:
-    """One fresh journey (the same helpers PR #269's tests run) must reproduce
-    the staged decision bytes — everything but the solver's wall-clock timing."""
+def test_a_fresh_front_door_journey_solves_its_own_converted_scene(monkeypatch, tmp_path) -> None:
+    """One fresh journey (the same helpers PR #269's tests run) solves this capture's
+    merged scene, converted by the scene bridge: the solved bytes, the echoed input
+    hash and the direct endpoint's decision all bind to those converted bytes, and
+    the terminal manual_review journey is preserved. The staged receipt stays a
+    historical record of the reference-fixture stage and pins nothing here."""
 
     spec = importlib.util.spec_from_file_location(
         "w6_frontdoor_reference", BACKEND / "tests" / "test_capture_placement_frontdoor.py"
@@ -284,15 +287,35 @@ def test_a_fresh_front_door_journey_matches_the_staged_receipt(monkeypatch, tmp_
     client = TestClient(app)
     client.headers["Authorization"] = f"Bearer {'d' * 40}"
 
-    _, run_id, _ = fd._walk_front_door(root, packet, client, app, packet["packetId"])
-    artifacts = fd._run_artifacts(app.state.deps, run_id)
-    fresh = json.loads(app.state.deps.storage.get_bytes(artifacts["placement/placement.json"]))
+    _, run_id, events = fd._walk_front_door(root, packet, client, app, packet["packetId"])
+    deps = app.state.deps
+    artifacts = fd._run_artifacts(deps, run_id)
+    fresh = json.loads(deps.storage.get_bytes(artifacts["placement/placement.json"]))
+
+    # The solved bytes are this journey's merged scene, converted by the bridge —
+    # not a fixture.
+    source = json.loads(deps.storage.get_bytes(artifacts["scene/scene.json"]))
+    scene_bytes = json.dumps(fd.convert_scene(source), sort_keys=True, allow_nan=False).encode()
+    assert deps.storage.get_bytes(artifacts["placement/scene.json"]) == scene_bytes
+    assert fresh["stats"]["input_sha256"] == hashlib.sha256(scene_bytes).hexdigest()
+
+    # The terminal manual_review journey is preserved.
+    run = deps.state.get_run(run_id)
+    assert run is not None and run.status == "done" and run.outcome_kind == fresh["decision"]
+    assert "placement" not in run.degraded
+    terminals = [event for event in events if event["type"] in fd.TERMINAL_EVENTS]
+    assert [event["type"] for event in terminals] == ["verdict_ready"]
+    assert terminals[0]["data"]["kind"] == fresh["decision"]
 
     def projection(payload: dict) -> bytes:
         decision = json.loads(json.dumps(payload))
         decision["stats"].pop("elapsed_ms")
         return json.dumps(decision, sort_keys=True).encode()
 
-    assert projection(fresh) == projection(PLACEMENT), (
-        "a fresh journey's decision differs from the staged placement result"
+    # The direct endpoint and the capture front door agree on the same converted bytes.
+    with TestClient(fd.adapter.load_api()["api"].app) as direct:
+        response = direct.post("/v1/placements", content=scene_bytes, headers={"Content-Type": "application/json"})
+    assert response.status_code == 200, response.text
+    assert projection(response.json()) == projection(fresh), (
+        "the direct endpoint's decision differs from the fresh journey's placement decision"
     )