| """Offline v2 asset loop and the external WorldEdge transaction contract. |
| |
| Asset.canonical_transform maps raw asset coordinates (including its pivot, |
| source scale/axis conventions) to a metre, Z-up, bbox-bottom-centred frame. |
| The world transform is T(bottom_center) @ Rz(yaw) @ canonical_transform. |
| No mesh rescaling or deletion is performed to satisfy validation. |
| |
| Callbacks must implement their own I/O timeout: max_seconds is checked at |
| callback boundaries, rather than claiming to interrupt arbitrary host code. |
| """ |
| from copy import deepcopy |
| from dataclasses import dataclass |
| import hashlib |
| import json |
| import math |
| from threading import RLock |
| from time import monotonic |
| from typing import Protocol |
|
|
| from .validation import _capability_names, _polygon, _validate_support_bbox, _vector, effective_support_requests, validate_required_levels, validate_scene |
|
|
|
|
| IDENTITY = ((1., 0., 0., 0.), (0., 1., 0., 0.), (0., 0., 1., 0.), (0., 0., 0., 1.)) |
|
|
|
|
| @dataclass(frozen=True) |
| class SupportSurface: |
| """Verified horizontal surface in the asset canonical frame, not bbox top.""" |
| surface_id: str |
| local_polygon_xy_m: tuple |
| local_z_m: float |
|
|
| def __post_init__(self): |
| if not isinstance(self.surface_id, str) or not self.surface_id: |
| raise ValueError("support surface needs a stable ID") |
| _polygon(self.local_polygon_xy_m) |
| _vector((self.local_z_m,), 1) |
| object.__setattr__(self, "local_polygon_xy_m", tuple(tuple(p) for p in self.local_polygon_xy_m)) |
|
|
| def as_dict(self): |
| return {"surface_id": self.surface_id, "local_polygon_xy_m": [list(p) for p in self.local_polygon_xy_m], |
| "local_z_m": self.local_z_m} |
|
|
|
|
| @dataclass(frozen=True) |
| class Asset: |
| ref: str |
| category: str |
| actual_size_local_m: tuple |
| description: str = "" |
| canonical_transform: tuple = IDENTITY |
| semantic_front_local: tuple | None = None |
| capabilities: tuple | None = None |
| support_surfaces: tuple = () |
| provenance: tuple = () |
|
|
| def __post_init__(self): |
| if not isinstance(self.ref, str) or not self.ref.strip() or not isinstance(self.category, str) or not self.category.strip(): |
| raise ValueError("asset needs a reference and category") |
| if not isinstance(self.description, str): |
| raise ValueError("asset description must be a string") |
| object.__setattr__(self, "actual_size_local_m", _vector(self.actual_size_local_m, 3, positive=True)) |
| rows = tuple(_vector(row, 4) for row in self.canonical_transform) |
| if len(rows) != 4 or rows[3] != (0., 0., 0., 1.): |
| raise ValueError("canonical_transform must be a finite affine 4x4 transform") |
| determinant = (rows[0][0]*(rows[1][1]*rows[2][2]-rows[1][2]*rows[2][1]) - |
| rows[0][1]*(rows[1][0]*rows[2][2]-rows[1][2]*rows[2][0]) + |
| rows[0][2]*(rows[1][0]*rows[2][1]-rows[1][1]*rows[2][0])) |
| if abs(determinant) < 1e-12: |
| raise ValueError("canonical_transform is singular") |
| object.__setattr__(self, "canonical_transform", rows) |
| if self.semantic_front_local is not None: |
| front = _vector(self.semantic_front_local, 3) |
| if math.hypot(front[0], front[1]) < 1e-12 or abs(front[2]) > 1e-6: |
| raise ValueError("semantic front must be a nonzero horizontal direction") |
| if front[0] <= 0 or abs(front[1] / math.hypot(front[0], front[1])) > 1e-6: |
| raise ValueError("known semantic front must already be canonical +X; normalize asset size and transform upstream") |
| object.__setattr__(self, "semantic_front_local", (1., 0., 0.)) |
| if self.capabilities is not None: |
| object.__setattr__(self, "capabilities", _capability_names(self.capabilities)) |
| if any(not isinstance(surface, SupportSurface) for surface in self.support_surfaces): |
| raise ValueError("invalid verified support surface metadata") |
| _validate_support_bbox([surface.as_dict() for surface in self.support_surfaces], self.actual_size_local_m) |
| object.__setattr__(self, "support_surfaces", tuple(self.support_surfaces)) |
| if not isinstance(self.provenance, (tuple, list)) or any(not isinstance(x, str) for x in self.provenance): |
| raise ValueError("asset provenance must be a list/tuple of source references") |
| object.__setattr__(self, "provenance", tuple(self.provenance)) |
|
|
|
|
| def _eligible(asset, request, prediction): |
| |
| |
| if request.get("attributes"): |
| return False |
| if asset.category.casefold().strip() != request["category"].casefold().strip(): |
| return False |
| required = set(request.get("required_capabilities", ())) |
| if required and (asset.capabilities is None or not required.issubset(asset.capabilities)): |
| return False |
| if request.get("semantic_front_required") and asset.semantic_front_local is None: |
| return False |
| size = asset.actual_size_local_m |
| fixed = request.get("fixed_size_local_m") |
| if fixed is not None and any(b is not None and abs(a-b) > 1e-6 for a, b in zip(size, fixed)): |
| return False |
| bounds = request.get("size_bounds_local_m") |
| if bounds is not None and any(x < lo-1e-6 or x > hi+1e-6 for x, lo, hi in zip(size, bounds["min"], bounds["max"])): |
| return False |
| tolerance = request.get("retrieval_tolerance_log", request.get("retrieval_tolerance")) |
| if tolerance is not None: |
| limits = (tolerance,) * 3 if isinstance(tolerance, (int, float)) else tuple(tolerance) |
| if len(limits) != 3 or any(not math.isfinite(x) or x < 0 for x in limits): |
| raise ValueError("retrieval_tolerance is a scalar or three max absolute log ratios") |
| if any(abs(math.log(a/b)) > limit+1e-12 for a, b, limit in |
| zip(size, prediction["target_size_local_m"], limits)): |
| return False |
| return True |
|
|
|
|
| class Resolver(Protocol): |
| def resolve(self, request_object, prediction, *, excluded_refs=()) -> Asset | None: ... |
|
|
|
|
| @dataclass(frozen=True) |
| class CatalogResolver: |
| """Category/capability reference catalog; unverified typed attributes reject. |
| |
| Natural-language descriptions are not a verified attribute evidence source. |
| Attribute-bearing requests need an explicit external evidence contract before |
| this reference loop can admit them. |
| """ |
| assets: tuple |
|
|
| def __post_init__(self): |
| if any(not isinstance(asset, Asset) for asset in self.assets): |
| raise ValueError("catalog entries must be Asset instances") |
| if len({a.ref for a in self.assets}) != len(self.assets): |
| raise ValueError("duplicate asset references") |
| object.__setattr__(self, "assets", tuple(self.assets)) |
|
|
| def resolve(self, request_object, prediction, *, excluded_refs=()): |
| options = tuple(a for a in self.assets if a.ref not in excluded_refs and _eligible(a, request_object, prediction)) |
| return min(options, key=lambda a: (sum(abs(math.log(x/y)) for x, y in |
| zip(a.actual_size_local_m, prediction["target_size_local_m"])), a.ref), default=None) |
|
|
|
|
| def _transform(asset, position, yaw): |
| c, s = math.cos(yaw), math.sin(yaw) |
| world = ((c, -s, 0, position[0]), (s, c, 0, position[1]), (0, 0, 1, position[2]), (0, 0, 0, 1)) |
| return [[sum(world[i][k]*asset.canonical_transform[k][j] for k in range(4)) for j in range(4)] for i in range(4)] |
|
|
|
|
| def reconcile(assets, condition, prediction): |
| """Resolve actual geometry and propagate verified support heights without changing targets.""" |
| requests = {obj["id"]: obj for obj in effective_support_requests(condition)} |
| records = {obj["id"]: {**deepcopy(obj), "target_bottom_center_m": list(obj["bottom_center_m"]), |
| "target_yaw_rad": obj["yaw_rad"], "actual_size_local_m": list(assets[obj["id"]].actual_size_local_m), |
| "asset_ref": assets[obj["id"]].ref, |
| "capabilities": list(assets[obj["id"]].capabilities) if assets[obj["id"]].capabilities is not None else None, |
| "semantic_front_local": list(assets[obj["id"]].semantic_front_local) if |
| assets[obj["id"]].semantic_front_local is not None else None, |
| "support_parent": requests[obj["id"]].get("support_parent"), |
| "support_surfaces": [surface.as_dict() for surface in assets[obj["id"]].support_surfaces], |
| "provenance": list(assets[obj["id"]].provenance)} for obj in prediction["objects"]} |
| done = frozenset() |
| fixed = {obj["id"]: obj for obj in condition["room"].get("fixed_objects", ())} |
| while len(done) < len(records): |
| ready = tuple(ident for ident in records if ident not in done and |
| (requests[ident].get("support_parent") not in records or requests[ident].get("support_parent") in done)) |
| if not ready: |
| raise ValueError("support graph contains a cycle") |
| for ident in ready: |
| parent = requests[ident].get("support_parent") |
| parent_obj = records.get(parent, fixed.get(parent)) |
| surfaces = parent_obj.get("support_surfaces", ()) if parent_obj else () |
| surface_id = requests[ident].get("support_surface_id") |
| surfaces = tuple(sf for sf in surfaces if surface_id is None or sf["surface_id"] == surface_id) |
| |
| if len(surfaces) == 1: |
| p = records[ident]["bottom_center_m"] |
| records = {**records, ident: {**records[ident], "bottom_center_m": |
| [p[0], p[1], parent_obj["bottom_center_m"][2]+surfaces[0]["local_z_m"]], |
| "support_surface_id": surfaces[0]["surface_id"]}} |
| done = done.union(ready) |
| return [{**records[obj["id"]], "asset_transform": _transform(assets[obj["id"]], |
| records[obj["id"]]["bottom_center_m"], records[obj["id"]]["yaw_rad"])} for obj in prediction["objects"]] |
|
|
|
|
| @dataclass(frozen=True) |
| class RuntimeBudget: |
| max_asset_retries: int = 2 |
| max_repair_calls: int = 0 |
| max_seconds: float = 10.0 |
|
|
| def __post_init__(self): |
| for count in (self.max_asset_retries, self.max_repair_calls): |
| if isinstance(count, bool) or not isinstance(count, int) or count < 0: |
| raise ValueError("retry and repair budgets must be nonnegative integers") |
| if isinstance(self.max_seconds, bool) or not math.isfinite(self.max_seconds) or self.max_seconds < 0: |
| raise ValueError("time budget must be finite and nonnegative") |
|
|
|
|
| @dataclass(frozen=True) |
| class BoundedTranslationRepair: |
| """Small deterministic XY move for boundary/box penetration; no geometry changes. |
| |
| Moves one offending object and its support descendants per call. Other |
| constraints may still fail, so every returned scene requires revalidation. |
| This is a reference local repair, not a proof of feasible re-layout. |
| """ |
| max_step_m: float = .25 |
|
|
| def __post_init__(self): |
| if isinstance(self.max_step_m, bool) or not math.isfinite(self.max_step_m) or self.max_step_m <= 0: |
| raise ValueError("translation step must be finite and positive") |
|
|
| def __call__(self, condition, objects, report): |
| index = {obj["id"]: obj for obj in objects} |
| failure = next((check for check in report["checks"] if check["status"] == "violation" and |
| check["code"] in ("boundary", "collision", "fixed_collision")), None) |
| if failure is None: |
| return deepcopy(objects) |
| ident = next((ident for ident in failure["object_ids"] if ident in index), None) |
| if ident is None: |
| return deepcopy(objects) |
| obj = index[ident] |
| if failure["code"] == "boundary": |
| if condition["room"].get("boundary_known") is False: |
| return deepcopy(objects) |
| point = _polygon(condition["room"]["floor_polygon_xy_m"]).representative_point() |
| delta = (point.x - obj["bottom_center_m"][0], point.y - obj["bottom_center_m"][1]) |
| else: |
| fixed = {fixed["id"]: fixed for fixed in condition["room"].get("fixed_objects", ())} |
| other = next((index.get(i, fixed.get(i)) for i in failure["object_ids"] if i != ident), None) |
| if other is None: |
| return deepcopy(objects) |
| delta = tuple(obj["bottom_center_m"][q]-other["bottom_center_m"][q] for q in (0, 1)) |
| norm = math.hypot(*delta) |
| delta = tuple(x * min(self.max_step_m, norm)/norm for x in delta) if norm > 1e-12 else (self.max_step_m, 0.) |
| parents = {req["id"]: req.get("support_parent") for req in effective_support_requests(condition)} |
| affected = frozenset((ident,)) |
| while True: |
| expanded = affected.union(child for child, parent in parents.items() if parent in affected) |
| if expanded == affected: |
| break |
| affected = expanded |
| return [{**obj, "bottom_center_m": [obj["bottom_center_m"][0]+delta[0], |
| obj["bottom_center_m"][1]+delta[1], obj["bottom_center_m"][2]]} |
| if obj["id"] in affected else deepcopy(obj) for obj in objects] |
|
|
|
|
| class Host(Protocol): |
| """A real adapter must atomically check version, deduplicate, and commit all objects.""" |
| def commit(self, objects, *, idempotency_key, expected_world_version) -> dict: ... |
|
|
|
|
| @dataclass(frozen=True) |
| class _HostState: |
| version: int = 0 |
| objects: tuple = () |
| requests: tuple = () |
|
|
|
|
| class AtomicMemoryHost: |
| """Offline reference transaction sink; never changes an external world.""" |
| def __init__(self): |
| self._state = _HostState() |
| self._lock = RLock() |
|
|
| def snapshot(self): |
| with self._lock: |
| return {"version": self._state.version, "objects": deepcopy(list(self._state.objects)), |
| "commits": len(self._state.requests)} |
|
|
| def commit(self, objects, *, idempotency_key, expected_world_version): |
| if not isinstance(idempotency_key, str) or not idempotency_key: |
| raise ValueError("commit requires a stable idempotency key") |
| if isinstance(expected_world_version, bool) or not isinstance(expected_world_version, int): |
| raise ValueError("commit requires an integer world version") |
| payload = deepcopy(list(objects)) |
| fingerprint = hashlib.sha256(json.dumps(payload, sort_keys=True, allow_nan=False).encode()).hexdigest() |
| with self._lock: |
| previous = next((x for x in self._state.requests if x[0] == idempotency_key), None) |
| if previous: |
| if previous[1] != fingerprint: |
| raise ValueError("idempotency key already used for a different scene") |
| return {"committed": True, "world_version": previous[2], "idempotent": True} |
| if self._state.version != expected_world_version: |
| raise ValueError("world version conflict") |
| ids = [obj.get("id") for obj in payload] |
| existing = {obj["id"] for obj in self._state.objects} |
| if any(not isinstance(ident, str) or not ident for ident in ids) or len(ids) != len(set(ids)) or existing.intersection(ids): |
| raise ValueError("commit object IDs must be new, nonempty and unique") |
| version = self._state.version + 1 |
| self._state = _HostState(version, self._state.objects + tuple(payload), |
| self._state.requests + ((idempotency_key, fingerprint, version),)) |
| return {"committed": True, "world_version": version, "idempotent": False} |
|
|
|
|
| def _repair_is_pose_only(before, after): |
| if not isinstance(after, (list, tuple)) or len(after) != len(before): |
| return False |
| index = {obj.get("id"): obj for obj in after if isinstance(obj, dict)} |
| if len(index) != len(before) or set(index) != {obj["id"] for obj in before}: |
| return False |
| allowed = {"bottom_center_m", "yaw_rad", "asset_transform"} |
| return all({k: v for k, v in obj.items() if k not in allowed} == |
| {k: v for k, v in index[obj["id"]].items() if k not in allowed} for obj in before) |
|
|
|
|
| def _result(condition, raw, target, first, final, first_objects, objects, diagnostics, counts, start, commit=None): |
| n = len(condition.get("objects", ())) if isinstance(condition, dict) else 0 |
| mismatch = [sum(abs(math.log(a/b)) for a, b in zip(obj["actual_size_local_m"], obj["target_size_local_m"])) |
| for obj in objects] |
| return {"ok": bool(final and final["ok"] and not counts.get("fatal")), "committed": bool(commit and commit.get("committed")), |
| "raw_model_output": deepcopy(raw), "actual_first_pass_objects": deepcopy(first_objects), |
| "final_objects": deepcopy(objects), "validation": {"target": target, "actual_first_pass": first, "final": final}, |
| "diagnostics": list(diagnostics), "commit": commit, |
| "attempts": list(counts.get("attempts", ())), |
| "metrics": {"model": {"schema_success": counts.get("schema_success", False), |
| "requested_ids_exactly_once": counts.get("schema_success", False), |
| "positive_valid_size": counts.get("schema_success", False), |
| "target_geometry_valid": bool(target and target["ok"])}, |
| "asset": {"retrieval_coverage": counts.get("resolved", 0)/n if n else 1., |
| "resolver_calls": counts.get("resolver_calls", 0), |
| "target_actual_log_size_l1_mean": sum(mismatch)/len(mismatch) if mismatch else None, |
| "capability_satisfaction": bool(objects) and not counts.get("asset_filter_failure", False), |
| "first_pass_actual_geometry_validation": bool(first and first["ok"])}, |
| "system": {"repaired_success": bool(final and final["ok"] and |
| (counts.get("repair_calls", 0) or counts.get("asset_retries", 0))), |
| "fallback_rate": 0., "final_commit": bool(commit and commit.get("committed")), |
| "commit_attempted": counts.get("commit_attempted", False), |
| "solver_success": None, "fastfill_latency_ms": None, |
| "end_to_end_latency_ms": (monotonic()-start)*1000, |
| "asset_retries": counts.get("asset_retries", 0), "repair_calls": counts.get("repair_calls", 0)}}} |
|
|
|
|
| def run_pipeline(condition, prediction, resolver, *, host=None, budget=RuntimeBudget(), repair=None, |
| expected_world_version=None, idempotency_key=None, required_levels=("bbox",)): |
| """Resolve, reconcile, validate, bounded retry/pose repair, then one atomic commit. |
| |
| repair(condition, actual_objects, report) receives copies and may change |
| positions/yaw only. It must explicitly account for support dependencies. |
| A real host adapter carries responsibility for an atomic external transaction; |
| this function never calls it for a failed or unknown required validation. |
| """ |
| start, diagnostics, counts = monotonic(), (), {} |
| raw, objects, first_objects, target, first, final = deepcopy(prediction), [], [], None, None, None |
| def finish(commit=None): |
| return _result(condition, raw, target, first, final, first_objects, objects, diagnostics, counts, start, commit) |
| def expired(): |
| return monotonic()-start >= budget.max_seconds |
| if expired(): |
| diagnostics = ({"code": "budget_exhausted", "phase": "start"},) |
| return finish() |
| try: |
| from .schema import validate_condition, validate_layout |
| required_levels = validate_required_levels(required_levels) |
| validate_condition(condition) |
| validate_layout(prediction, condition) |
| requests = {obj["id"]: obj for obj in effective_support_requests(condition)} |
| counts = {**counts, "schema_success": True} |
| target = validate_scene(condition, prediction["objects"], required_levels=required_levels) |
| predictions = {o["id"]: o for o in prediction["objects"]} |
| assets, excluded = {}, {ident: () for ident in requests} |
| for ident in requests: |
| if expired(): |
| raise TimeoutError("asset_resolution") |
| counts = {**counts, "resolver_calls": counts.get("resolver_calls", 0)+1} |
| asset = resolver.resolve(deepcopy(requests[ident]), deepcopy(predictions[ident]), excluded_refs=()) |
| counts = {**counts, "attempts": counts.get("attempts", ())+ |
| ({"type": "asset_resolution", "object_id": ident, "asset_ref": asset.ref if isinstance(asset, Asset) else None},)} |
| if asset is None or not isinstance(asset, Asset) or not _eligible(asset, requests[ident], predictions[ident]): |
| diagnostics += ({"code": "asset_unavailable", "object_id": ident, |
| "requirements": deepcopy(requests[ident]), |
| "target_size_local_m": list(predictions[ident]["target_size_local_m"])},) |
| counts = {**counts, "asset_filter_failure": True} |
| return finish() |
| assets = {**assets, ident: asset} |
| counts = {**counts, "resolved": len(assets)} |
| objects = reconcile(assets, condition, prediction) |
| first_objects = deepcopy(objects) |
| first = final = validate_scene(condition, objects, stage="actual", required_levels=required_levels) |
| while not final["ok"] and counts.get("asset_retries", 0) < budget.max_asset_retries: |
| if expired(): |
| raise TimeoutError("asset_retry") |
| failed = tuple(ident for c in final["checks"] if c["hard"] and c["status"] != "pass" for ident in c["object_ids"]) |
| candidates = tuple(ident for ident in requests if ident in failed) or tuple(requests) |
| changed = False |
| for ident in candidates: |
| if counts.get("asset_retries", 0) >= budget.max_asset_retries: |
| break |
| if expired(): |
| raise TimeoutError("asset_retry") |
| excluded = {**excluded, ident: excluded[ident]+(assets[ident].ref,)} |
| counts = {**counts, "asset_retries": counts.get("asset_retries", 0)+1, |
| "resolver_calls": counts.get("resolver_calls", 0)+1} |
| asset = resolver.resolve(deepcopy(requests[ident]), deepcopy(predictions[ident]), excluded_refs=excluded[ident]) |
| counts = {**counts, "attempts": counts.get("attempts", ())+ |
| ({"type": "asset_retry", "object_id": ident, "asset_ref": asset.ref if isinstance(asset, Asset) else None, |
| "excluded_refs": list(excluded[ident])},)} |
| if asset is not None and isinstance(asset, Asset) and asset.ref not in excluded[ident] and _eligible(asset, requests[ident], predictions[ident]): |
| assets, changed = {**assets, ident: asset}, True |
| diagnostics += ({"code": "asset_reselected", "object_id": ident, "asset_ref": asset.ref},) |
| break |
| if not changed: |
| break |
| objects = reconcile(assets, condition, prediction) |
| final = validate_scene(condition, objects, stage="actual", required_levels=required_levels) |
| while not final["ok"] and repair is not None and counts.get("repair_calls", 0) < budget.max_repair_calls: |
| if expired(): |
| raise TimeoutError("repair") |
| counts = {**counts, "repair_calls": counts.get("repair_calls", 0)+1} |
| repaired = repair(deepcopy(condition), deepcopy(objects), deepcopy(final)) |
| counts = {**counts, "attempts": counts.get("attempts", ())+ |
| ({"type": "pose_repair", "call": counts["repair_calls"]},)} |
| if not _repair_is_pose_only(objects, repaired): |
| diagnostics += ({"code": "invalid_repair", "reason": "repair must preserve IDs, targets, assets and actual sizes"},) |
| break |
| objects = [{**obj, "asset_transform": _transform(assets[obj["id"]], obj["bottom_center_m"], obj["yaw_rad"])} for obj in repaired] |
| final = validate_scene(condition, objects, stage="actual", required_levels=required_levels) |
| if expired(): |
| raise TimeoutError("before_commit") |
| if not final["ok"]: |
| diagnostics += ({"code": "validation_failed", "required_failures": |
| [c for c in final["checks"] if c["hard"] and c["status"] != "pass"]},) |
| return finish() |
| if host is not None: |
| try: |
| counts = {**counts, "commit_attempted": True} |
| commit = host.commit(deepcopy(objects), idempotency_key=idempotency_key, |
| expected_world_version=expected_world_version) |
| if not isinstance(commit, dict) or commit.get("committed") is not True: |
| raise ValueError("host did not acknowledge an atomic commit") |
| return finish(commit) |
| except (ValueError, RuntimeError, OSError) as exc: |
| diagnostics += ({"code": "commit_failed", "message": str(exc)},) |
| counts = {**counts, "fatal": True} |
| return finish() |
| return finish() |
| except TimeoutError as exc: |
| diagnostics += ({"code": "budget_exhausted", "phase": str(exc)},) |
| except (ValueError, KeyError, TypeError) as exc: |
| diagnostics += ({"code": "invalid_input_or_geometry", "message": str(exc)},) |
| except (RuntimeError, OSError) as exc: |
| diagnostics += ({"code": "runtime_failure", "message": str(exc)},) |
| counts = {**counts, "fatal": True} |
| return finish() |
|
|
