From 38a5735abf6c48982f8131b6dc6814afb90e28f3 Mon Sep 17 00:00:00 2001 From: Sam Gu <127461594+SamGu-NRX@users.noreply.github.com> Date: Thu, 15 Oct 2026 20:02:59 +0000 Subject: [PATCH] feat(placement): solve each capture's merged scene instead of a reference MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The placement stage solved one committed reference scene and only warned when the run's own artifact was absent, so every capture scored the same geometry. It now refuses anything but this capture's merged scene: `scene/scene.json` converted by the scene bridge is the solver's whole input, re-serialized deterministically (sorted keys, no NaN) so `stats.input_sha256` binds to the exact bytes solved. The committed reference fixture is deleted. Placement now also waits for `objects_3d` to be terminal, not just `scene` done, and `result` waits for a terminal placement. The stage reads `objects/objects_3d.json` only as readiness evidence — stage done, declared `hss-objects-3d/v0` format, an explicit empty object list being the one valid "no hazards" shape — because the coordinator's `_merge_if_ready` already merged detected objects into the source scene; a second merge would duplicate same-id hazards. Missing inputs or an unfinished objects_3d stage are typed non-retryable errors (`placement_scene_input_missing`, `placement_objects_3d_unavailable`) that degrade the run to manual_review instead of implying no hazards; bridge refusals keep flowing through `placement_scene_invalid` and `placement_scene_unsupported`. STAGE_DEPS gains the same placement and result entries as the runtime graph, the input-selection contract admits the two stage inputs, and the tests move from the fixture to the journey's own bytes: the stage tests pin source-derived hashing, dependency ordering, every refusal path and one-time merge precedence; the front-door and W6 fresh-journey tests compare against a direct POST of the same converted bytes, keeping the staged receipt as a historical record. Co-authored-by: obvious-autobuild[bot] --- backend/server/contracts/jobs.py | 4 +- backend/server/orchestrator/graph.py | 27 +- .../tests/test_placement_stage.py | 314 +++++++++++++++--- backend/server/placement/adapter.py | 147 ++++++-- backend/server/placement/reference_scene.json | 44 --- backend/tests/test_backend_w6_gate.py | 39 ++- .../tests/test_capture_placement_frontdoor.py | 34 +- 7 files changed, 459 insertions(+), 150 deletions(-) delete mode 100644 backend/server/placement/reference_scene.json diff --git a/backend/server/contracts/jobs.py b/backend/server/contracts/jobs.py index 1e5788aea..01165242b 100644 --- a/backend/server/contracts/jobs.py +++ b/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"}) diff --git a/backend/server/orchestrator/graph.py b/backend/server/orchestrator/graph.py index d50fecf6a..f269b53dd 100644 --- a/backend/server/orchestrator/graph.py +++ b/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/", diff --git a/backend/server/orchestrator/tests/test_placement_stage.py b/backend/server/orchestrator/tests/test_placement_stage.py index 2537299e9..770cc8f7d 100644 --- a/backend/server/orchestrator/tests/test_placement_stage.py +++ b/backend/server/orchestrator/tests/test_placement_stage.py @@ -1,17 +1,24 @@ """The placement stage: the repo's unchanged top-level solver behind the orchestrator. -Three claims, each tested against the real code and real bytes: +Five claims, each tested against the real code and real bytes: - The stage invokes the top-level `server/` solver — loaded by explicit absolute file path, never a flat import, which could resolve the eval vendor tree's same-named - modules — on a committed scene fixture and publishes a schema-valid artifact whose - decision bytes are identical across runs. Only the solver's own `stats.elapsed_ms` - may differ: the result schema requires that field, so determinism is asserted on the - decision projection without it. -- A malformed scene is recorded stage data: the stage record carries the typed - `placement_scene_invalid` code, the failed stage event is emitted, and the optional - stage degrades the run (`degraded:placement`, capture `manual_review`) instead of - failing it. + modules — on the run's own merged scene, converted by `scene_bridge.convert_scene`, + and publishes a schema-valid artifact whose decision bytes are identical across runs. + Only the solver's own `stats.elapsed_ms` may differ. The solved bytes are the + converted capture scene, never a committed fixture. +- Dependency ordering: placement waits for `objects_3d` to be terminal as well as + `scene` done, and `result` waits for placement to be terminal — the runtime graph and + the `STAGE_DEPS` contract stay aligned. +- Refusals: a missing source scene or an objects_3d stage that did not finish (`done` + with an explicit empty object list is valid) is a typed, non-retryable stage error — + the stage degrades the run instead of implying no hazards — and writes no placement + result. +- Object-merge precedence: the hazards in the merged source scene reach the wire + geometry exactly once. The adapter treats `objects/objects_3d.json` as readiness + evidence only; a same-id detector record still sitting in that artifact must not be + merged a second time. - `POST /v1/placements` still serves its existing contract from the same loaded code, and the backend's own `server` package is never shadowed by the flat names. """ @@ -31,24 +38,23 @@ import server from server.common.backends.storage_local import LocalFsStorage -from server.contracts.jobs import RunRecord, StageRecord +from server.contracts.jobs import STAGE_DEPS, RunRecord, StageContext, StageRecord, StageResult from server.orchestrator.cli import run_local from server.orchestrator.coordinator import RunCoordinator -from server.orchestrator.graph import STAGE_NAMES +from server.orchestrator.graph import STAGE_NAMES, STAGES, is_ready, ready_stages from server.orchestrator.runner import error_parts, is_retryable from server.orchestrator.services import bind, deadline_from -from server.orchestrator.stateio import put_json +from server.orchestrator.stubs._common import scene_document from server.orchestrator.tests.conftest import build_services from server.placement import adapter +from server.placement.scene_bridge import convert_scene REPO_ROOT = Path(__file__).resolve().parents[4] TOP_SERVER = REPO_ROOT / "server" -FIXTURE = TOP_SERVER / "tests/fixtures/example-scene.json" -REFERENCE = Path(adapter.__file__).resolve().parent / "reference_scene.json" RESULT_SCHEMA = json.loads((TOP_SERVER / "schemas/result.schema.json").read_text()) PACKET_DIR = REPO_ROOT / "backend/data/fixtures/packets/mini-arkit" -# The solver's own verdict for the committed fixture, observed through the boundary. +# The solver's own verdict for the committed fixtures, observed through the boundary. EXPECTED_DECISION = "manual_review" @@ -73,10 +79,51 @@ def _decision_bytes(payload: dict) -> bytes: return json.dumps(decision, sort_keys=True).encode() -def test_reference_scene_is_the_top_level_fixture() -> None: - """The vendored input is a byte-identical copy of the fixture the task names.""" +def _hazard_document() -> dict: + """A valid merged scene with one observed tap hazard beside the stub meter.""" - assert REFERENCE.read_bytes() == FIXTURE.read_bytes() + document = scene_document("cap_hazard") + document["objects"].append( + { + "id": "h1", + "class": "gas_meter", + "geometry": {"kind": "wall_rect", "wall_id": "w1", "s_m": [2.0, 2.6], "bottom_m": 0.8, "top_m": 1.4}, + "confidence": 0.9, + "plus_minus_m": 0.05, + "source": {"kind": "tap"}, + "evidence": "observed", + } + ) + return document + + +class _StageRow: + def __init__(self, status: str) -> None: + self.status = status + + +class _State: + """The one method the adapter reads: the run's objects_3d stage record.""" + + def __init__(self, status: str | None = "done") -> None: + self.status = status + + def get_stage(self, run_id: str, stage: str) -> _StageRow | None: + return _StageRow(self.status) if self.status is not None else None + + +def _ctx(work_dir: Path, state: _State) -> types.SimpleNamespace: + """The slice of StageContext the adapter body reads.""" + + return types.SimpleNamespace(work_dir=work_dir, run_id="run_unit", state=state) + + +def _seed_inputs(work_dir: Path, scene: dict, objects_3d: dict | None) -> None: + (work_dir / "scene").mkdir(parents=True, exist_ok=True) + (work_dir / "scene" / "scene.json").write_bytes(json.dumps(scene, sort_keys=True).encode()) + if objects_3d is not None: + (work_dir / "objects").mkdir(parents=True, exist_ok=True) + (work_dir / "objects" / "objects_3d.json").write_bytes(json.dumps(objects_3d, sort_keys=True).encode()) def test_boundary_loads_the_top_level_server_by_explicit_path(monkeypatch) -> None: @@ -99,6 +146,13 @@ def test_typed_codes_reach_the_runner_seam() -> None: assert not is_retryable(scene_error) # the bytes do not heal on a retry boundary_error = adapter.PlacementBoundaryError("missing tree") assert error_parts(boundary_error) == ("placement_boundary_missing", "missing tree") + assert not is_retryable(boundary_error) + missing = adapter.PlacementInputError("scene/scene.json is missing") + assert error_parts(missing) == ("placement_scene_input_missing", "scene/scene.json is missing") + assert not is_retryable(missing) + objects = adapter.PlacementObjectsError("objects_3d stage status is 'failed'") + assert error_parts(objects) == ("placement_objects_3d_unavailable", "objects_3d stage status is 'failed'") + assert not is_retryable(objects) def test_missing_solver_tree_raises_the_typed_boundary_error(tmp_path, monkeypatch) -> None: @@ -115,7 +169,29 @@ def test_missing_solver_tree_raises_the_typed_boundary_error(tmp_path, monkeypat assert not is_retryable(boundary) -def test_run_local_places_the_reference_scene_deterministically(tmp_path) -> None: +def test_dependency_ordering_is_aligned_across_graph_and_contract() -> None: + """Placement needs the merged scene (scene done, objects_3d terminal); result + needs placement terminal. The runtime graph and STAGE_DEPS say the same thing.""" + + assert STAGES["placement"].deps == (("scene", "done"), ("objects_3d", "terminal")) + assert STAGE_DEPS["placement"] == ("scene", "objects_3d") + assert STAGES["result"].deps == (("criteria", "done"), ("placement", "terminal")) + assert STAGE_DEPS["result"] == ("criteria", "placement") + + status = {name: "pending" for name in STAGE_NAMES} + status["scene"] = "done" + status["objects_3d"] = "running" + assert "placement" not in ready_stages(status), "objects_3d still running" + status["objects_3d"] = "done" + assert is_ready(STAGES["placement"], status) + status["placement"] = "running" + status["criteria"] = "done" + assert not is_ready(STAGES["result"], status), "result waits for a terminal placement" + status["placement"] = "failed" + assert is_ready(STAGES["result"], status), "a failed placement is terminal and must not block the run" + + +def test_run_local_solves_the_captures_merged_scene_deterministically(tmp_path) -> None: tar_path = _packet_tar(tmp_path) outs: list[Path] = [] for name in ("run-a", "run-b"): @@ -127,14 +203,20 @@ def test_run_local_places_the_reference_scene_deterministically(tmp_path) -> Non placement = [_read_artifact(out, run, "placement/placement.json") for out, run in zip(outs, runs, strict=True)] scenes = [_read_artifact(out, run, "placement/scene.json") for out, run in zip(outs, runs, strict=True)] payloads = [json.loads(item) for item in placement] - fixture_sha = hashlib.sha256(FIXTURE.read_bytes()).hexdigest() - for run, payload, scene_bytes in zip(runs, payloads, scenes, strict=True): + for out, run, payload, scene_bytes in zip(outs, runs, payloads, scenes, strict=True): Draft202012Validator(RESULT_SCHEMA).validate(payload) + # The solved bytes are this capture's merged scene, converted — not a fixture. + source = json.loads(_read_artifact(out, run, "scene/scene.json")) + expected = json.dumps(convert_scene(source), sort_keys=True, allow_nan=False).encode() + assert scene_bytes == expected + assert payload["stats"]["input_sha256"] == hashlib.sha256(scene_bytes).hexdigest() assert payload["decision"] == EXPECTED_DECISION - assert payload["stats"]["input_sha256"] == fixture_sha - assert scene_bytes == FIXTURE.read_bytes() # the stage solved exactly the committed fixture assert "placement" not in run["degraded"] assert run["status"] == "done" + # The source scene artifact is untouched by the stage: still the merged + # metres document, not the stage's converted feet bytes. + assert json.loads(_read_artifact(out, run, "scene/scene.json")) == source + assert scene_bytes != _read_artifact(out, run, "scene/scene.json") stages = {rec["stage"]: rec for rec in json.loads(_read_artifact(outs[0], runs[0], "run/stages.json"))} assert stages["placement"]["status"] == "done" @@ -143,7 +225,12 @@ def test_run_local_places_the_reference_scene_deterministically(tmp_path) -> Non events = [json.loads(line) for line in (outs[0] / "events.jsonl").read_text().splitlines()] marks = [(event["data"].get("stage"), event["data"].get("status")) for event in events if event["type"] == "stage"] - assert marks.index(("scene", "done")) < marks.index(("placement", "done")) < marks.index(("criteria", "done")) + assert ( + marks.index(("objects_3d", "done")) < marks.index(("placement", "running")) < marks.index(("placement", "done")) + ) + # Criteria and placement both follow the merge and run in parallel; the result + # stage is the one that must wait for a terminal placement. + assert marks.index(("scene", "done")) < marks.index(("placement", "done")) < marks.index(("result", "running")) finals = [event for event in events if event["type"] in {"verdict_ready", "retake_request", "failed"}] assert [event["type"] for event in finals] == ["verdict_ready"] @@ -151,11 +238,53 @@ def test_run_local_places_the_reference_scene_deterministically(tmp_path) -> Non assert _decision_bytes(payloads[0]) == _decision_bytes(payloads[1]) -def test_malformed_scene_is_recorded_stage_data(tmp_path) -> None: +def test_malformed_source_scene_refuses_at_the_adapter(tmp_path) -> None: + """Malformed bytes on the hydrated input are a typed refusal with no output.""" + + work = tmp_path / "work" + work.mkdir() + _seed_inputs(work, _hazard_document(), {"format": "hss-objects-3d/v0", "objects": []}) + (work / "scene" / "scene.json").write_bytes(b"{ not json") + + with pytest.raises(adapter.PlacementSceneError): + adapter.run(_ctx(work, _State())) + assert not (work / "placement" / "placement.json").exists() + assert not (work / "placement" / "scene.json").exists() + + +def test_unsupported_source_scene_is_recorded_stage_data(tmp_path, monkeypatch) -> None: + """A scene that validates as scene v0 but sits outside the bridge's bounded scope — + a hazard with generated evidence — is a typed non-retryable placement error on the + existing conservative path: the stage fails, the run degrades to manual_review, and + no placement artifact is emitted.""" + + from server.orchestrator.stubs import scene as scene_stub + from server.orchestrator.stubs._common import json_bytes, scene_document, write_files + + def unsupported_scene(ctx: StageContext) -> StageResult: + scene = scene_document(ctx.capture_id) + scene["objects"].append( + { + "id": "h1", + "class": "gas_meter", + "geometry": {"kind": "wall_rect", "wall_id": "w1", "s_m": [2.0, 2.6], "bottom_m": 0.8, "top_m": 1.4}, + "confidence": 0.9, + "plus_minus_m": 0.05, + "source": {"kind": "vlm"}, + "evidence": "generated", + } + ) + return write_files( + ctx, + { + "scene/scene.json": json_bytes(scene), + "scene/build_report.json": json.dumps({"status": "stub"}).encode(), + }, + ) + + monkeypatch.setattr(scene_stub, "run", unsupported_scene) services, uri, digest = build_services(tmp_path, "memory") with bind(services): - key = "captures/cap_test/placement/scene.json" - services.storage.put_bytes(key, b"{ not json", "application/json") run = RunRecord( run_id="run_placefail", capture_id="cap_test", @@ -167,13 +296,9 @@ def test_malformed_scene_is_recorded_stage_data(tmp_path) -> None: profile_default=services.settings.RULES_PROFILE_DEFAULT, created_at=services.clock.iso(), deadline_at=deadline_from(services.clock, services.settings.HARD_STOP_MIN), - artifacts={"placement/scene.json": key}, + artifacts={}, ) services.state.put_run(run) - # Persist the input the way the orchestrator persists stage artifacts: the - # per-stage map is what overlay_stage_artifacts recovers, because a parallel - # stub stage's put_run would drop a key that lives only on the run record. - put_json(services.state, f"stage_artifacts:{run.run_id}:placement", {"placement/scene.json": key}) now = services.clock.iso() for name in STAGE_NAMES: services.state.upsert_stage( @@ -184,13 +309,14 @@ def test_malformed_scene_is_recorded_stage_data(tmp_path) -> None: finished = services.state.get_run(run.run_id) record = services.state.get_stage(run.run_id, "placement") assert record.status == "failed" - assert record.error_code == "placement_scene_invalid" - assert record.error.startswith("placement_scene_invalid:") + assert record.error_code == "placement_scene_unsupported" + assert record.error.startswith("placement_scene_unsupported:") assert finished.degraded == ["placement"] assert finished.status == "done" assert finished.outcome_kind == "manual_review" assert finished.failure_code == "degraded:placement" assert "placement/placement.json" not in finished.artifacts + assert "placement/scene.json" not in finished.artifacts assert services.state.get_capture("cap_test").status == "manual_review" events = services.state.events_after("cap_test", 0, limit=500) @@ -199,22 +325,136 @@ def test_malformed_scene_is_recorded_stage_data(tmp_path) -> None: for event in events if event.type == "stage" and event.data.get("stage") == "placement" and event.data.get("status") == "failed" ] - assert [event.data.get("errorCode") for event in failed] == ["placement_scene_invalid"] + assert [event.data.get("errorCode") for event in failed] == ["placement_scene_unsupported"] + marks = [(event.data.get("stage"), event.data.get("status")) for event in events if event.type == "stage"] + assert marks.index(("objects_3d", "done")) < marks.index(("placement", "failed")) finals = [event for event in events if event.type in {"verdict_ready", "retake_request", "failed"}] assert len(finals) == 1 +def test_missing_source_scene_refuses_and_writes_no_result(tmp_path) -> None: + """No merged scene in the run, no fallback: a typed refusal, no output bytes.""" + + work = tmp_path / "work" + work.mkdir() + _seed_inputs(work, _hazard_document(), {"format": "hss-objects-3d/v0", "objects": []}) + (work / "scene" / "scene.json").unlink() + + with pytest.raises(adapter.PlacementInputError) as excinfo: + adapter.run(_ctx(work, _State())) + assert "scene/scene.json" in str(excinfo.value) + assert not (work / "placement" / "placement.json").exists() + assert not (work / "placement" / "scene.json").exists() + + +def test_objects_3d_not_done_refuses_instead_of_implying_no_hazards(tmp_path) -> None: + """failed, skipped, missing records and absent artifacts all refuse. A successful + explicit empty list is valid — that is the one shape that may read as no hazards.""" + + work = tmp_path / "work" + work.mkdir() + _seed_inputs(work, _hazard_document(), {"format": "hss-objects-3d/v0", "objects": []}) + + for status in ("failed", "skipped", "running"): + with pytest.raises(adapter.PlacementObjectsError, match=status): + adapter.run(_ctx(work, _State(status))) + with pytest.raises(adapter.PlacementObjectsError): + adapter.run(_ctx(work, _State(None))) # no stage record at all + + assert not (work / "placement" / "placement.json").exists() + + # Readiness evidence must be the declared format. + (work / "objects" / "objects_3d.json").write_bytes(json.dumps({"format": "hss-objects/v9", "objects": []}).encode()) + with pytest.raises(adapter.PlacementObjectsError, match="hss-objects-3d/v0"): + adapter.run(_ctx(work, _State())) + + # The valid gate: done stage, declared format, explicit empty list. + (work / "objects" / "objects_3d.json").write_bytes( + json.dumps({"format": "hss-objects-3d/v0", "objects": []}).encode() + ) + result = adapter.run(_ctx(work, _State())) + assert (work / "placement" / "placement.json").is_file() + assert result.metrics["elapsed_ms"] > 0 + + +def test_merged_hazards_convert_once_and_never_merge_again(tmp_path) -> None: + """The observed tap hazard appears once in the source scene and once in the + converted geometry — even with a same-id detector record still sitting in the + objects_3d artifact. A second merge would duplicate it.""" + + scene = _hazard_document() + assert [obj["id"] for obj in scene["objects"]].count("h1") == 1 + # What a detector replacement would have produced, still in the readiness artifact. + objects_3d = { + "format": "hss-objects-3d/v0", + "objects": [ + { + "id": "h1", + "label": "gas_meter", + "s_m": [2.0, 2.6], + "bottom_m": 0.8, + "top_m": 1.4, + "confidence": 0.7, + "source": {"kind": "vlm"}, + } + ], + } + work = tmp_path / "work" + work.mkdir() + _seed_inputs(work, scene, objects_3d) + + adapter.run(_ctx(work, _State())) + converted = json.loads((work / "placement" / "scene.json").read_bytes()) + hazards = converted.get("objects", []) + assert [obj["type"] for obj in hazards] == ["gas_meter"] + assert len(hazards) == 1, "the same-id detector object was merged a second time" + span = hazards[0]["span_ft"] + assert span == [pytest.approx(2.0 / 0.3048), pytest.approx(2.6 / 0.3048)] + assert hazards[0]["conf"] == 0.9, "the merged tap evidence wins, not the detector record" + + +def test_distinct_source_scenes_distinct_bytes_and_same_input_is_deterministic(tmp_path) -> None: + plain = tmp_path / "plain" + plain.mkdir() + _seed_inputs(plain, scene_document("cap_plain"), {"format": "hss-objects-3d/v0", "objects": []}) + result_plain = adapter.run(_ctx(plain, _State())) + payload_plain = json.loads((plain / "placement" / "placement.json").read_bytes()) + + hazard = tmp_path / "hazard" + hazard.mkdir() + _seed_inputs(hazard, _hazard_document(), {"format": "hss-objects-3d/v0", "objects": []}) + result_hazard = adapter.run(_ctx(hazard, _State())) + payload_hazard = json.loads((hazard / "placement" / "placement.json").read_bytes()) + + # Distinct source scenes → distinct converted inputs, hashes and outputs. + scene_plain = (plain / "placement" / "scene.json").read_bytes() + scene_hazard = (hazard / "placement" / "scene.json").read_bytes() + assert scene_plain != scene_hazard + assert payload_plain["stats"]["input_sha256"] == hashlib.sha256(scene_plain).hexdigest() + assert payload_hazard["stats"]["input_sha256"] == hashlib.sha256(scene_hazard).hexdigest() + assert payload_plain["stats"]["input_sha256"] != payload_hazard["stats"]["input_sha256"] + assert result_plain.artifacts.keys() == result_hazard.artifacts.keys() + assert _decision_bytes(payload_plain) != _decision_bytes(payload_hazard), "the hazard changes the checks" + + # Same input again: identical decision bytes, only the solver's timing may move. + again = adapter.run(_ctx(plain, _State())) + payload_again = json.loads((plain / "placement" / "placement.json").read_bytes()) + assert again.artifacts == result_plain.artifacts + assert _decision_bytes(payload_again) == _decision_bytes(payload_plain) + + def test_v1_placements_contract_stays_live() -> None: modules = adapter.load_api() assert Path(modules["api"].__file__) == TOP_SERVER / "api.py" + fixture = TOP_SERVER / "tests/fixtures/example-scene.json" with TestClient(modules["api"].app) as client: response = client.post( "/v1/placements", - content=FIXTURE.read_bytes(), + content=fixture.read_bytes(), headers={"Content-Type": "application/json"}, ) assert response.status_code == 200 body = response.json() Draft202012Validator(RESULT_SCHEMA).validate(body) assert body["decision"] == EXPECTED_DECISION - assert body["stats"]["input_sha256"] == hashlib.sha256(FIXTURE.read_bytes()).hexdigest() + assert body["stats"]["input_sha256"] == hashlib.sha256(fixture.read_bytes()).hexdigest() diff --git a/backend/server/placement/adapter.py b/backend/server/placement/adapter.py index 5cde5f31c..6c2f09075 100644 --- a/backend/server/placement/adapter.py +++ b/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=[], ) diff --git a/backend/server/placement/reference_scene.json b/backend/server/placement/reference_scene.json deleted file mode 100644 index 873a92674..000000000 --- a/backend/server/placement/reference_scene.json +++ /dev/null @@ -1,44 +0,0 @@ -{ - "schema_version": "1.0", - "meter": {"pos": [0.0, 5.0, 0.0], "wall_id": "side", "plus_minus_ft": 0.3}, - "walls": [ - {"id": "back", "baseline": [[-14.0, -18.0], [-14.0, 0.0]], "height_ft": 9, "plus_minus_ft": 0.3}, - {"id": "side", "baseline": [[-14.0, 0.0], [22.0, 0.0]], "height_ft": 9, "plus_minus_ft": 0.3} - ], - "objects": [ - {"type": "gas_meter", "wall_id": "side", "span_ft": [-5.0, -4.0], "bottom_ft": 0.5, "top_ft": 2.5, - "source": "tap", "plus_minus_ft": 0.3, - "footprint": [[-5.0, 0.0], [-4.0, 0.0], [-4.0, 1.0], [-5.0, 1.0]]}, - {"type": "window", "wall_id": "side", "span_ft": [3.0, 6.0], "bottom_ft": 3.0, "top_ft": 7.0, - "attrs": {"operable": true, "well": false}, "source": "vlm", "conf": 0.86}, - {"type": "door", "wall_id": "side", "span_ft": [17.0, 20.0], "bottom_ft": 0.0, "top_ft": 6.8, - "source": "tap"}, - {"type": "ac", "wall_id": "back", "span_ft": [-20.0, -17.0], "source": "tap", - "footprint": [[-17.0, -6.0], [-14.0, -6.0], [-14.0, -3.0], [-17.0, -3.0]]} - ], - "ground": [ - {"type": "lawn", "polygon": [[-28.0, -18.0], [-14.0, -18.0], [-14.0, 14.0], [-28.0, 14.0]]}, - {"type": "mulch", "polygon": [[-14.0, 0.0], [22.0, 0.0], [22.0, 4.0], [-14.0, 4.0]]}, - {"type": "lawn", "polygon": [[-14.0, 4.0], [22.0, 4.0], [22.0, 14.0], [-14.0, 14.0]]}, - {"type": "drive", "polygon": [[22.5, 0.0], [34.0, 0.0], [34.0, 30.0], [22.5, 30.0]]} - ], - "overheads": [], - "facing": [{"wall_id": "side", "span_ft": [-14.0, 22.0], "depth_ft": 14.0, "plus_minus_ft": 0.5}], - "coverage": { - "ends": {"left": {"kind": "unexplored", "note": "back wall continues past the patio"}, - "right": {"kind": "limit", "note": "front corner, driveway"}}, - "observed": [ - {"band": "wall", "span_ft": [-26.0, 22.0]}, - {"band": "ground", "span_ft": [-26.0, 26.0], "out_ft": 14.0}, - {"band": "overhead", "span_ft": [-26.0, 22.0]}, - {"band": "facing", "span_ft": [-26.0, 22.0]} - ] - }, - "keyframes": [ - {"id": "k1", "pose": [1, 0, 0, 0, 0, 1, 0, 0, 0, 0, 1, 0, 2.0, 4.5, 9.0, 1], - "intrinsics": [1450.0, 1450.0, 960.0, 720.0], "w": 1920, "h": 1440, "img": "k1.jpg"} - ], - "stills": {"meter_close": "meter_close.jpg"}, - "gps": {"lat": 30.2672, "lon": -97.7431}, - "heading": {"deg": 181.0, "accuracy_deg": 12.0} -} diff --git a/backend/tests/test_backend_w6_gate.py b/backend/tests/test_backend_w6_gate.py index e18baa955..01c0ea868 100644 --- a/backend/tests/test_backend_w6_gate.py +++ b/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" ) diff --git a/backend/tests/test_capture_placement_frontdoor.py b/backend/tests/test_capture_placement_frontdoor.py index 2b5059025..4970bba62 100644 --- a/backend/tests/test_capture_placement_frontdoor.py +++ b/backend/tests/test_capture_placement_frontdoor.py @@ -20,16 +20,17 @@ artifact map, its stage record done with positive solver time, its event between ``scene`` and ``criteria``), so the journey fails if the capture path stops invoking it. It asserts the decision is deterministic and visible in the run and -result evidence: the artifact validates against the top-level result schema, the +result evidence: the artifact validates against the top-level result schema and +solves exactly this journey's merged scene, converted by the scene bridge; the served outcome's kind and the capture's terminal status equal the solver's -``decision`` on every journey, and two journeys plus one direct -``POST /v1/placements`` call produce identical decision bytes — everything but -the solver's own ``stats.elapsed_ms``. +``decision`` on every journey; and two journeys plus one direct +``POST /v1/placements`` call on the same converted bytes produce identical +decision bytes — everything but the solver's own ``stats.elapsed_ms``. The direct check also stays here as its own test, unchanged from the contract the placement-stage PR pinned: the front-door journeys prove the stage is wired into the capture path, the direct check proves Hunter's public contract survives. Both -run on the same committed scene bytes, which is what makes the two comparable. +compare on the same converted scene bytes, which is what makes them comparable. """ from __future__ import annotations @@ -50,6 +51,7 @@ from server.contracts.jobs import run_id_for_capture from server.orchestrator.stateio import get_json from server.placement import adapter +from server.placement.scene_bridge import convert_scene REPO_ROOT = Path(__file__).resolve().parents[2] TOP_SERVER = REPO_ROOT / "server" @@ -216,7 +218,6 @@ def _decision_bytes(payload: dict) -> bytes: def test_capture_front_door_reaches_the_placement_stage(frontdoor) -> None: root, packet, client, app = frontdoor deps = app.state.deps - fixture_bytes = SCENE_FIXTURE.read_bytes() projections = [] for tag in ("a", "b"): @@ -233,8 +234,13 @@ def test_capture_front_door_reaches_the_placement_stage(frontdoor) -> None: # path stopped invoking the stage. placement = json.loads(deps.storage.get_bytes(artifacts["placement/placement.json"])) Draft202012Validator(RESULT_SCHEMA).validate(placement) - # The stage solved exactly the committed scene bytes. - assert deps.storage.get_bytes(artifacts["placement/scene.json"]) == fixture_bytes + # The stage solved exactly this journey's merged scene, converted — the + # capture's own metres document, never a fixture. + source = json.loads(deps.storage.get_bytes(artifacts["scene/scene.json"])) + scene_bytes = json.dumps(convert_scene(source), sort_keys=True, allow_nan=False).encode() + assert deps.storage.get_bytes(artifacts["placement/scene.json"]) == scene_bytes + assert placement["stats"]["input_sha256"] == hashlib.sha256(scene_bytes).hexdigest() + assert placement["decision"] == EXPECTED_DECISION record = deps.state.get_stage(run_id, "placement") assert record.status == "done" and record.error_code is None @@ -244,14 +250,16 @@ def test_capture_front_door_reaches_the_placement_stage(frontdoor) -> None: (event["data"].get("stage"), event["data"].get("status")) for event in events if event["type"] == "stage" ] assert ("placement", "done") in marks - assert marks.index(("scene", "done")) < marks.index(("placement", "done")) < marks.index(("criteria", "done")) + # Criteria and placement both follow the merge and run in parallel; the result + # stage is the one that must wait for a terminal placement. + assert marks.index(("scene", "done")) < marks.index(("placement", "done")) + assert marks.index(("placement", "done")) < marks.index(("result", "running")) terminals = [event for event in events if event["type"] in TERMINAL_EVENTS] assert [event["type"] for event in terminals] == ["verdict_ready"] assert terminals[0]["data"]["kind"] == EXPECTED_DECISION # --- the decision is visible in the run and result evidence ------------- - assert placement["decision"] == EXPECTED_DECISION status = client.get(f"/v1/captures/{capture_id}").json() served = client.get(f"/v1/captures/{capture_id}/result") assert served.status_code == 200, served.text @@ -261,12 +269,12 @@ def test_capture_front_door_reaches_the_placement_stage(frontdoor) -> None: assert result["status"] == status["status"] == "manual_review" projections.append(_decision_bytes(placement)) - # Deterministic: two journeys over the same committed scene bytes, one decision. + # Deterministic: two journeys over the same packet, one decision. assert projections[0] == projections[1] - # The direct endpoint and the capture front door agree on the same scene bytes. + # The direct endpoint and the capture front door agree on the same converted bytes. with TestClient(adapter.load_api()["api"].app) as direct: - response = direct.post("/v1/placements", content=fixture_bytes, headers={"Content-Type": "application/json"}) + response = direct.post("/v1/placements", content=scene_bytes, headers={"Content-Type": "application/json"}) assert response.status_code == 200, response.text assert _decision_bytes(response.json()) == projections[0]