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]