geoai/geoai-workbench.git

			@@ -3,15 +3,18 @@
			from __future__ import annotations

			import argparse
			import ast
			import base64
			import binascii
			import hashlib
			import json
			import math
			import os
			import re
			import shutil
			import subprocess
			import threading
			import zipfile
			from datetime import UTC, datetime
			from http import HTTPStatus
			from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer
			@@ -31,8 +34,8 @@
			MAX_SEGMENTATION_IMAGES_PER_RUN = 6
			MAX_MEASUREMENT_RASTERS_PER_RUN = 4
			MAX_POINTCLOUDS_PER_RUN = 2
			MAX_PHOTO_RECONSTRUCTION_IMAGES_PER_RUN = 30
			PHOTO_RECONSTRUCTION_TIMEOUT = 7200
			MAX_PHOTO_RECONSTRUCTION_IMAGES_PER_RUN = 1_000
			PHOTO_RECONSTRUCTION_TIMEOUT = 86_400
			RISK_RULE_REQUIRED_FILES = {"observations", "zones", "rules"}
			MAX_ANOMALY_IMAGES_PER_ROLE = 6
			CHANGE_THRESHOLD_DEFAULT = 0.5
			@@ -74,14 +77,27 @@
			POINTCLOUD_TRAINING_JOBS_LOCK = threading.Lock()
			POINTCLOUD_INFERENCE_JOBS: dict[str, dict[str, Any]] = {}
			POINTCLOUD_INFERENCE_JOBS_LOCK = threading.Lock()
			MAX_ANNOTATION_LABELS = 400_000
			POINTCLOUD_CLASS_CODES = {1, 2, 5, 6, 15, 16}
			POINTCLOUD_ANNOTATION_SOURCE_JOBS: dict[str, dict[str, Any]] = {}
			POINTCLOUD_ANNOTATION_SOURCE_JOBS_LOCK = threading.Lock()
			POINTCLOUD_DETAIL_REQUEST_LOCK = threading.Lock()
			MAX_POINTCLOUD_DETAIL_RADIUS = 10_000_000.0
			DEFAULT_POINTCLOUD_ANNOTATION_CLASSES = (
			{"code": 1, "key": "other_unknown", "label": "其他/未知", "color": [128, 128, 128], "builtIn": True},
			{"code": 2, "key": "ground", "label": "地面", "color": [151, 111, 51], "builtIn": True},
			{"code": 5, "key": "vegetation", "label": "植被", "color": [59, 163, 87], "builtIn": True},
			{"code": 6, "key": "building_structure", "label": "建筑物", "color": [224, 115, 55], "builtIn": True},
			{"code": 15, "key": "pole_tower", "label": "杆塔", "color": [149, 89, 210], "builtIn": True},
			{"code": 16, "key": "power_line", "label": "电线", "color": [231, 196, 61], "builtIn": True},
			)
			POINTCLOUD_CLASS_CODES = {item["code"] for item in DEFAULT_POINTCLOUD_ANNOTATION_CLASSES}
			ANNOTATION_CLASS_KEY = re.compile(r"^[a-z][a-z0-9_]{0,47}$")
			POINTCLOUD_CPU_ENVIRONMENT = "05-3d-pointcloud"
			POINTCLOUD_GPU_ENVIRONMENT = "05-3d-pointcloud-gpu"
			OBJECT_DETECTION_CPU_ENVIRONMENT = "01-object-detection"
			OBJECT_DETECTION_GPU_ENVIRONMENT = "01-object-detection-cuda"
			CHANGE_DETECTION_CPU_ENVIRONMENT = "00-change-detection"
			CHANGE_DETECTION_GPU_ENVIRONMENT = "00-change-detection-cuda"
			COMPUTE_DEVICES = {"auto", "cpu", "cuda"}


			class ApiError(ValueError):
			@@ -117,7 +133,9 @@


			def relative_path(root: Path, path: Path) -> str:
			return path.relative_to(root).as_posix()
			# Windows may present a temporary root with its 8.3 spelling while an
			# input directory has already been resolved to its long spelling.
			return path.resolve().relative_to(root.resolve()).as_posix()


			def load_json(path: Path) -> dict[str, Any]:
			@@ -136,18 +154,19 @@
			return digest.hexdigest()


			def pointcloud_execution_environment(root: Path, requested_device: str = "auto") -> dict[str, str]:
			"""Select only a fixed point-cloud interpreter after a short CUDA probe.
			def select_execution_environment(root: Path, requested_device: str, cpu_environment: str, gpu_environment: str, label: str) -> dict[str, Any]:
			"""Choose an allowlisted CPU/CUDA interpreter and retain the selection evidence."""
			if requested_device not in COMPUTE_DEVICES:
			raise ApiError(f"{label} device must be auto, cpu, or cuda.")
			cpu_python = root / ".venvs" / cpu_environment / "Scripts" / "python.exe"
			gpu_python = root / ".venvs" / gpu_environment / "Scripts" / "python.exe"
			if requested_device == "cpu":
			if not cpu_python.is_file():
			raise ApiError(f"{label} CPU virtual environment is unavailable. Run the capability setup first.")
			return {"python": str(cpu_python), "device": "cpu", "environment": cpu_environment, "torchVersion": "unknown", "requestedDevice": "cpu", "fallbackUsed": False, "fallbackReason": None}

			The console never accepts a browser-supplied Python path. A failed or
			unavailable GPU environment is an expected condition for ``auto`` and
			falls back to the retained CPU environment.
			"""
			if requested_device not in {"auto", "cpu", "cuda"}:
			raise ApiError("Point-cloud device must be auto, cpu, or cuda.")
			cpu_python = root / ".venvs" / POINTCLOUD_CPU_ENVIRONMENT / "Scripts" / "python.exe"
			gpu_python = root / ".venvs" / POINTCLOUD_GPU_ENVIRONMENT / "Scripts" / "python.exe"
			if requested_device != "cpu" and gpu_python.is_file():
			probe_reason = "Fixed CUDA virtual environment is unavailable."
			if gpu_python.is_file():
			try:
			probe = subprocess.run(
			[str(gpu_python), "-c", "import json, torch; print(json.dumps({'cuda': bool(torch.cuda.is_available()), 'torch': torch.__version__}))"],
			@@ -157,64 +176,48 @@
			timeout=20,
			check=False,
			)
			payload = json.loads(probe.stdout.strip().splitlines()[-1]) if probe.returncode == 0 and probe.stdout.strip() else {}
			if payload.get("cuda") is True and isinstance(payload.get("torch"), str):
			return {"python": str(gpu_python), "device": "cuda", "environment": POINTCLOUD_GPU_ENVIRONMENT, "torchVersion": payload["torch"]}
			if probe.returncode == 0 and probe.stdout.strip():
			payload = json.loads(probe.stdout.strip().splitlines()[-1])
			if payload.get("cuda") is True and isinstance(payload.get("torch"), str):
			return {"python": str(gpu_python), "device": "cuda", "environment": gpu_environment, "torchVersion": payload["torch"], "requestedDevice": requested_device, "fallbackUsed": False, "fallbackReason": None}
			probe_reason = "CUDA probe reports that PyTorch cannot use CUDA."
			else:
			probe_reason = "CUDA probe process did not complete successfully."
			except (OSError, subprocess.SubprocessError, json.JSONDecodeError, IndexError):
			pass
			probe_reason = "CUDA probe could not return a valid result."
			if requested_device == "cuda":
			raise ApiError("CUDA was requested, but the fixed point-cloud GPU environment is unavailable.")
			raise ApiError(f"CUDA was requested for {label}, but it is unavailable: {probe_reason}")
			if not cpu_python.is_file():
			raise ApiError("3D point-cloud CPU virtual environment is unavailable. Run the capability setup first.")
			return {"python": str(cpu_python), "device": "cpu", "environment": POINTCLOUD_CPU_ENVIRONMENT, "torchVersion": "unknown"}
			raise ApiError(f"{label} CPU virtual environment is unavailable. Run the capability setup first.")
			return {"python": str(cpu_python), "device": "cpu", "environment": cpu_environment, "torchVersion": "unknown", "requestedDevice": "auto", "fallbackUsed": True, "fallbackReason": probe_reason}


			def object_detection_execution_environment(root: Path) -> dict[str, str]:
			"""Choose the fixed object-detection CUDA environment only after probing it."""
			cpu_python = root / ".venvs" / OBJECT_DETECTION_CPU_ENVIRONMENT / "Scripts" / "python.exe"
			gpu_python = root / ".venvs" / OBJECT_DETECTION_GPU_ENVIRONMENT / "Scripts" / "python.exe"
			if gpu_python.is_file():
			try:
			probe = subprocess.run(
			[str(gpu_python), "-c", "import json, torch; print(json.dumps({'cuda': bool(torch.cuda.is_available()), 'torch': torch.__version__}))"],
			cwd=root,
			capture_output=True,
			text=True,
			timeout=20,
			check=False,
			)
			payload = json.loads(probe.stdout.strip().splitlines()[-1]) if probe.returncode == 0 and probe.stdout.strip() else {}
			if payload.get("cuda") is True and isinstance(payload.get("torch"), str):
			return {"python": str(gpu_python), "device": "cuda", "environment": OBJECT_DETECTION_GPU_ENVIRONMENT, "torchVersion": payload["torch"]}
			except (OSError, subprocess.SubprocessError, json.JSONDecodeError, IndexError):
			pass
			if not cpu_python.is_file():
			raise ApiError("Object-detection CPU virtual environment is unavailable. Run the capability setup first.")
			return {"python": str(cpu_python), "device": "cpu", "environment": OBJECT_DETECTION_CPU_ENVIRONMENT, "torchVersion": "unknown"}
			def pointcloud_execution_environment(root: Path, requested_device: str = "auto") -> dict[str, Any]:
			return select_execution_environment(root, requested_device, POINTCLOUD_CPU_ENVIRONMENT, POINTCLOUD_GPU_ENVIRONMENT, "Point-cloud")


			def change_detection_execution_environment(root: Path) -> dict[str, str]:
			"""Choose the fixed ChangeStar CUDA environment only after probing it."""
			cpu_python = root / ".venvs" / CHANGE_DETECTION_CPU_ENVIRONMENT / "Scripts" / "python.exe"
			gpu_python = root / ".venvs" / CHANGE_DETECTION_GPU_ENVIRONMENT / "Scripts" / "python.exe"
			if gpu_python.is_file():
			try:
			probe = subprocess.run(
			[str(gpu_python), "-c", "import json, torch; print(json.dumps({'cuda': bool(torch.cuda.is_available()), 'torch': torch.__version__}))"],
			cwd=root,
			capture_output=True,
			text=True,
			timeout=20,
			check=False,
			)
			payload = json.loads(probe.stdout.strip().splitlines()[-1]) if probe.returncode == 0 and probe.stdout.strip() else {}
			if payload.get("cuda") is True and isinstance(payload.get("torch"), str):
			return {"python": str(gpu_python), "device": "cuda", "environment": CHANGE_DETECTION_GPU_ENVIRONMENT, "torchVersion": payload["torch"]}
			except (OSError, subprocess.SubprocessError, json.JSONDecodeError, IndexError):
			pass
			if not cpu_python.is_file():
			raise ApiError("Change-detection CPU virtual environment is unavailable. Run the capability setup first.")
			return {"python": str(cpu_python), "device": "cpu", "environment": CHANGE_DETECTION_CPU_ENVIRONMENT, "torchVersion": "unknown"}
			def object_detection_execution_environment(root: Path, requested_device: str = "auto") -> dict[str, Any]:
			return select_execution_environment(root, requested_device, OBJECT_DETECTION_CPU_ENVIRONMENT, OBJECT_DETECTION_GPU_ENVIRONMENT, "Object-detection")


			def change_detection_execution_environment(root: Path, requested_device: str = "auto") -> dict[str, Any]:
			return select_execution_environment(root, requested_device, CHANGE_DETECTION_CPU_ENVIRONMENT, CHANGE_DETECTION_GPU_ENVIRONMENT, "Change-detection")


			def record_execution_metadata(path: Path, execution: dict[str, Any]) -> None:
			"""Preserve requested device, actual device and an automatic CPU fallback reason."""
			metadata = load_json(path)
			metadata["requested_device"] = execution["requestedDevice"]
			metadata["device"] = execution["device"]
			metadata["execution"] = {
			"requested_device": execution["requestedDevice"],
			"actual_device": execution["device"],
			"environment": execution["environment"],
			"torch_version": execution["torchVersion"],
			"fallback_used": execution["fallbackUsed"],
			"fallback_reason": execution["fallbackReason"],
			}
			path.write_text(json.dumps(metadata, ensure_ascii=False, indent=2), encoding="utf-8")


			def trajectory_runs(root: Path) -> list[dict[str, Any]]:
			@@ -511,8 +514,262 @@
			"artifactRoot": case["artifactRoot"], "file": name, "url": f"/{case['artifactRoot']}/{name}",
			"sha256": file_sha256(path), "pointCount": int(cloud.get("semantic_preview_points") or 0),
			"sourceKind": str(cloud.get("semantic_annotation_source_kind") or "generated point-cloud preview"),
			"detailAvailable": False, "detailFile": None,
			})
			sources.extend(standalone_pointcloud_annotation_sources(root))
			sources.extend(multiview_pointcloud_annotation_sources(root))
			return sources


			def standalone_pointcloud_annotation_sources(root: Path) -> list[dict[str, Any]]:
			"""Discover preview-only annotation uploads without pretending they are geometry runs."""
			output_root = root / "shared" / "outputs" / "05-3d-pointcloud"
			records: list[dict[str, Any]] = []
			metadata_paths = [
			output_root.glob("runs/annotation-source-/run_metadata.json"),
			output_root.glob("texture-baked-/run_metadata.json"),
			]
			for metadata_path in metadata_paths:
			artifact = metadata_path.parent
			metadata = load_json(metadata_path)
			contract = metadata.get("annotation_source")
			if metadata.get("capability") != "05-3d-pointcloud" or not metadata.get("annotation_source_job") or not isinstance(contract, dict):
			continue
			name, count, checksum = contract.get("file"), contract.get("point_count"), contract.get("sha256")
			if not isinstance(name, str) or Path(name).name != name or not isinstance(count, int) or count < 1 or not isinstance(checksum, str):
			continue
			path = artifact / name
			if not path.is_file() or path.suffix.lower() != ".ply" or file_sha256(path) != checksum or ply_vertex_count(path) != count:
			continue
			input_data = metadata.get("input") if isinstance(metadata.get("input"), dict) else {}
			input_name = str(input_data.get("file") or name)
			records.append({
			"id": f"{artifact.name}:{name}", "runId": artifact.name, "label": f"{artifact.name} / {input_name}",
			"artifactRoot": relative_path(root, artifact), "file": name, "url": f"/{relative_path(root, artifact)}/{name}",
			"sha256": checksum, "pointCount": count,
			"sourceKind": str(contract.get("kind") or "generated RGB/XYZ annotation preview"),
			"sourceHasRgb": bool(input_data.get("has_rgb")),
			"detailAvailable": False, "detailFile": None,
			})
			return sorted(records, key=lambda item: (item["runId"], item["id"]), reverse=True)


			def pointcloud_annotation_source(root: Path, source_id: str) -> dict[str, Any]:
			if not isinstance(source_id, str) or not source_id or "/" in source_id or len(source_id) > 300:
			raise ApiError("Invalid annotation source id.")
			source = next((item for item in pointcloud_annotation_sources(root) if item["id"] == source_id), None)
			if not source:
			raise ApiError("The selected generated annotation source is unavailable.")
			return source


			def pointcloud_annotation_detail(
			root: Path,
			source_id: str,
			center: tuple[float, float, float],
			radius: float,
			) -> bytes:
			"""Read one bounded, server-selected RGB detail window as a binary PLY."""
			if not all(math.isfinite(value) and abs(value) <= 1_000_000_000.0 for value in center):
			raise ApiError("Detail centre must contain finite local coordinates.")
			if not math.isfinite(radius) or not 0 < radius <= MAX_POINTCLOUD_DETAIL_RADIUS:
			raise ApiError(f"Detail radius must be between 0 and {MAX_POINTCLOUD_DETAIL_RADIUS:g}.")
			source = pointcloud_annotation_source(root, source_id)
			detail_name = source.get("detailFile")
			if not source.get("detailAvailable") or not isinstance(detail_name, str) or Path(detail_name).name != detail_name:
			raise ApiError("This annotation source has no local RGB detail layer.")
			artifact = (root / str(source["artifactRoot"])).resolve()
			output_root = (root / "shared" / "outputs" / "05-3d-pointcloud").resolve()
			detail_path = (artifact / detail_name).resolve()
			try:
			artifact.relative_to(output_root)
			detail_path.relative_to(artifact)
			except ValueError as exc:
			raise ApiError("The requested detail layer is outside the local point-cloud outputs.") from exc
			if not detail_path.is_file() or detail_path.suffix.lower() not in {".las", ".laz", ".ply"}:
			raise ApiError("The local RGB detail layer is unavailable.")
			python = root / ".venvs" / POINTCLOUD_CPU_ENVIRONMENT / "Scripts" / "python.exe"
			exporter = root / "capabilities" / "05-3d-pointcloud" / "export_pointcloud_detail.py"
			if not python.is_file() or not exporter.is_file():
			raise ApiError("Point-cloud detail exporter is unavailable. Run the capability setup first.")
			command = [
			str(python), str(exporter), "--input", str(detail_path), "--center",
			*(f"{value:.12g}" for value in center), "--radius", f"{radius:.12g}",
			"--max-points", "1200000",
			]
			if not POINTCLOUD_DETAIL_REQUEST_LOCK.acquire(timeout=1):
			raise ApiError("A point-cloud detail request is already running. Stop moving briefly and retry.")
			try:
			completed = subprocess.run(command, capture_output=True, timeout=120, check=False)
			except subprocess.TimeoutExpired as exc:
			raise ApiError("The local point-cloud detail request exceeded 120 seconds.") from exc
			finally:
			POINTCLOUD_DETAIL_REQUEST_LOCK.release()
			if completed.returncode != 0:
			message = completed.stderr.decode("utf-8", errors="replace").strip()
			raise ApiError(message[:400] or "The local point-cloud detail exporter failed.")
			payload = completed.stdout
			if not payload.startswith(b"ply\nformat binary_little_endian 1.0\n") or len(payload) > 24 * 1024 * 1024:
			raise ApiError("The local point-cloud detail response is invalid or exceeds its size limit.")
			return payload


			def ply_vertex_count(path: Path) -> int \| None:
			"""Read only the bounded PLY header; the point body can be hundreds of MB."""
			try:
			with path.open("rb") as stream:
			header = bytearray()
			while len(header) < 65_536:
			line = stream.readline(4_096)
			if not line:
			return None
			header.extend(line)
			if line.rstrip(b"\r\n") == b"end_header":
			break
			else:
			return None
			vertex_count: int \| None = None
			for line in header.decode("ascii").splitlines():
			fields = line.split()
			if len(fields) == 3 and fields[:2] == ["element", "vertex"] and fields[2].isdigit():
			vertex_count = int(fields[2])
			return vertex_count
			except (OSError, UnicodeDecodeError):
			return None


			def npz_array_row_count(path: Path, array_name: str) -> int \| None:
			"""Validate an NPZ array shape without importing a ML environment in the console."""
			try:
			with zipfile.ZipFile(path) as archive:
			with archive.open(f"{array_name}.npy") as stream:
			magic = stream.read(6)
			version = stream.read(2)
			if magic != b"\x93NUMPY" or len(version) != 2:
			return None
			header_size = 2 if version[0] == 1 else 4
			header_length = int.from_bytes(stream.read(header_size), "little")
			if header_length < 1 or header_length > 16_384:
			return None
			header = ast.literal_eval(stream.read(header_length).decode("latin1"))
			shape = header.get("shape") if isinstance(header, dict) else None
			if not isinstance(shape, tuple) or len(shape) != 2 or not all(isinstance(value, int) and value >= 0 for value in shape):
			return None
			return int(shape[0])
			except (OSError, KeyError, ValueError, SyntaxError, zipfile.BadZipFile):
			return None


			def multiview_pointcloud_annotation_sources(root: Path) -> list[dict[str, Any]]:
			"""Discover only complete, order-verified multi-view fusion annotation sources."""
			output_root = root / "shared" / "outputs" / "05-3d-pointcloud"
			records: list[dict[str, Any]] = []
			for metadata_path in output_root.glob("multiview-feature-*/run_metadata.json"):
			artifact = metadata_path.parent
			metadata = load_json(metadata_path)
			contract = metadata.get("annotation_source")
			artifacts = metadata.get("artifacts")
			if metadata.get("capability") != "05-3d-pointcloud" or not isinstance(contract, dict) or not isinstance(artifacts, dict):
			continue
			if contract.get("schema_version") != 1 or contract.get("kind") != "multiview_photo_feature_fusion":
			continue
			point_cloud = contract.get("point_cloud")
			feature_dataset = contract.get("feature_dataset")
			point_count = contract.get("point_count")
			if (
			not isinstance(point_cloud, str)
			or not isinstance(feature_dataset, str)
			or Path(point_cloud).name != point_cloud
			or Path(feature_dataset).name != feature_dataset
			or point_cloud != "multiview-annotation-source.ply"
			or feature_dataset != "multiview-point-features.npz"
			or artifacts.get("annotation_source") != point_cloud
			or artifacts.get("feature_dataset") != feature_dataset
			or not isinstance(point_count, int)
			or point_count < 1
			):
			continue
			cloud_path = artifact / point_cloud
			dataset_path = artifact / feature_dataset
			if not cloud_path.is_file() or not dataset_path.is_file():
			continue
			if contract.get("point_cloud_sha256") != file_sha256(cloud_path) or contract.get("feature_dataset_sha256") != file_sha256(dataset_path):
			continue
			if ply_vertex_count(cloud_path) != point_count or npz_array_row_count(dataset_path, "xyz") != point_count or npz_array_row_count(dataset_path, "las_rgb") != point_count:
			continue
			records.append({
			"id": f"{artifact.name}:{point_cloud}",
			"runId": artifact.name,
			"label": f"多视角照片特征融合样本 / {artifact.name}",
			"artifactRoot": relative_path(root, artifact),
			"file": point_cloud,
			"url": f"/{relative_path(root, artifact)}/{point_cloud}",
			"sha256": str(contract["point_cloud_sha256"]),
			"pointCount": point_count,
			"sourceKind": "多视角照片特征融合样本（原始 LAS RGB / XYZ；与特征数据同序）",
			})
			return sorted(records, key=lambda item: (item["runId"], item["id"]), reverse=True)


			def annotation_classes_path(root: Path) -> Path:
			return root / "shared" / "outputs" / "05-3d-pointcloud" / "annotation-classes.json"


			def annotation_classes(root: Path) -> list[dict[str, Any]]:
			"""Return the local editable taxonomy, with stable defaults for old workspaces."""
			saved = load_json(annotation_classes_path(root)).get("classes")
			if not isinstance(saved, list):
			return [dict(item) for item in DEFAULT_POINTCLOUD_ANNOTATION_CLASSES]
			try:
			return validate_annotation_classes(saved, allow_builtin=True)
			except ApiError:
			# A corrupt local taxonomy must not prevent existing annotations from opening.
			return [dict(item) for item in DEFAULT_POINTCLOUD_ANNOTATION_CLASSES]


			def validate_annotation_classes(values: list[Any], *, allow_builtin: bool) -> list[dict[str, Any]]:
			if not values or len(values) > 64:
			raise ApiError("The annotation taxonomy must contain 1 to 64 classes.")
			normalized: list[dict[str, Any]] = []
			codes: set[int] = set()
			keys: set[str] = set()
			labels: set[str] = set()
			built_in_codes = {item["code"] for item in DEFAULT_POINTCLOUD_ANNOTATION_CLASSES}
			for item in values:
			if not isinstance(item, dict):
			raise ApiError("Each annotation class must be an object.")
			code, key, label, color = item.get("code"), item.get("key"), item.get("label"), item.get("color")
			if not isinstance(code, int) or not 1 <= code <= 255:
			raise ApiError("Annotation class codes must be integers from 1 to 255 for LAS compatibility.")
			if not isinstance(key, str) or not ANNOTATION_CLASS_KEY.fullmatch(key):
			raise ApiError("Annotation class keys must use lowercase English letters, numbers, and underscores.")
			if not isinstance(label, str) or not 1 <= len(label.strip()) <= 40:
			raise ApiError("Annotation class labels must contain 1 to 40 characters.")
			if not isinstance(color, list) or len(color) != 3 or not all(isinstance(value, int) and 0 <= value <= 255 for value in color):
			raise ApiError("Annotation class colors must be three RGB integers from 0 to 255.")
			if code in codes or key in keys or label.strip() in labels:
			raise ApiError("Annotation class code, key, and label must each be unique.")
			if code in built_in_codes and not allow_builtin:
			raise ApiError("Built-in annotation classes cannot be replaced.")
			codes.add(code); keys.add(key); labels.add(label.strip())
			normalized.append({"code": code, "key": key, "label": label.strip(), "color": color, "builtIn": code in built_in_codes})
			return sorted(normalized, key=lambda item: item["code"])


			def write_annotation_classes(root: Path, classes: list[dict[str, Any]]) -> None:
			path = annotation_classes_path(root)
			path.parent.mkdir(parents=True, exist_ok=True)
			path.write_text(json.dumps({"schema_version": 1, "classes": classes}, ensure_ascii=False, indent=2), encoding="utf-8")


			def annotation_class_codes_in_use(root: Path) -> set[int]:
			used: set[int] = set()
			for path in (root / "shared" / "outputs" / "05-3d-pointcloud" / "annotations").glob("*/annotation.json"):
			record = load_json(path)
			for label in record.get("labels", []):
			if isinstance(label, list) and len(label) == 2 and isinstance(label[1], int):
			used.add(label[1])
			return used


			def pointcloud_annotations(root: Path) -> list[dict[str, Any]]:
			@@ -526,16 +783,170 @@
			return sorted(records, key=lambda item: (str(item["createdAt"]), str(item["id"])), reverse=True)


			def pointcloud_annotation_source_deletion_plan(root: Path, source_id: str) -> dict[str, Any]:
			"""Describe every generated artifact that will be removed for one source.

			Source paths are discovered server-side. Files outside the established
			point-cloud output/raw/processed layouts, including ``baseData``, are never
			part of this plan.
			"""
			source = next((item for item in pointcloud_annotation_sources(root) if item["id"] == source_id), None)
			if not source:
			raise ApiError("The selected annotation source is unavailable.")
			output_root = (root / "shared" / "outputs" / "05-3d-pointcloud").resolve()
			artifact = (root / str(source["artifactRoot"])).resolve()
			try:
			artifact.relative_to(output_root)
			except ValueError as exc:
			raise ApiError("The selected annotation source is outside the allowed output directory.") from exc
			siblings = [item for item in pointcloud_annotation_sources(root) if item["runId"] == str(source["runId"])]
			source_ids = {item["id"] for item in siblings}
			annotation_root = output_root / "annotations"
			annotations: list[Path] = []
			for path in annotation_root.glob("*/annotation.json"):
			if load_json(path).get("source_id") in source_ids:
			annotations.append(path.parent)
			annotation_paths = {path / "annotation.json" for path in annotations}
			training_root = output_root / "training-runs"
			training: list[Path] = []
			for metrics_path in training_root.glob("*/metrics.json"):
			value = load_json(metrics_path).get("annotation")
			if not isinstance(value, str):
			continue
			try:
			if Path(value).resolve() in annotation_paths:
			training.append(metrics_path.parent)
			except OSError:
			continue
			training_models = {path / "model.pt" for path in training}
			inference_root = output_root / "model-inference-runs"
			inference: list[Path] = []
			for metadata_path in inference_root.glob("*/run_metadata.json"):
			model = load_json(metadata_path).get("model")
			model_path = model.get("path") if isinstance(model, dict) else None
			if not isinstance(model_path, str):
			continue
			try:
			if Path(model_path).resolve() in training_models:
			inference.append(metadata_path.parent)
			except OSError:
			continue
			run_id = str(source["runId"])
			raw_root = root / "shared" / "data" / "raw" / "05-3d-pointcloud"
			processed_root = root / "shared" / "data" / "processed" / "05-3d-pointcloud"
			raw_candidates = [raw_root / "annotation-source-runs" / run_id, raw_root / "runs" / run_id]
			processed_candidates = [processed_root / "annotation-source-runs" / run_id, processed_root / "runs" / run_id]
			raw = [path for path in raw_candidates if path.is_dir()]
			processed = [path for path in processed_candidates if path.is_dir()]
			metadata = load_json(artifact / "run_metadata.json")
			return {
			"sourceId": source_id,
			"label": source["label"],
			"sourceKind": source["sourceKind"],
			"runId": run_id,
			"artifactRoot": str(source["artifactRoot"]),
			"outputDirectories": 1,
			"rawDirectories": len(raw),
			"processedDirectories": len(processed),
			"annotationRevisions": len(annotations),
			"trainingRuns": len(training),
			"inferenceRuns": len(inference),
			"siblingSources": len(siblings),
			"removesOriginalUpload": bool(raw),
			"preservesExternalInputs": not bool(raw) and not bool(metadata.get("annotation_source_job")),
			}


			def assert_removable_pointcloud_directory(root: Path, path: Path, allowed_root: Path) -> None:
			resolved = path.resolve()
			allowed = allowed_root.resolve()
			try:
			relative = resolved.relative_to(allowed)
			except ValueError as exc:
			raise ApiError("A deletion target is outside the allowed point-cloud workspace.") from exc
			if not relative.parts or not resolved.is_dir():
			raise ApiError("A deletion target is invalid.")


			def active_pointcloud_source_dependencies(run_id: str, annotation_ids: set[str], training_ids: set[str]) -> bool:
			active = {"queued", "running"}
			with POINTCLOUD_ANNOTATION_SOURCE_JOBS_LOCK:
			if any(job.get("runId") == run_id and job.get("status") in active for job in POINTCLOUD_ANNOTATION_SOURCE_JOBS.values()):
			return True
			with POINTCLOUD_TRAINING_JOBS_LOCK:
			if any(job.get("annotationId") in annotation_ids for job in POINTCLOUD_TRAINING_JOBS.values() if job.get("status") in active):
			return True
			with POINTCLOUD_INFERENCE_JOBS_LOCK:
			if any(job.get("modelId") in training_ids and job.get("status") in active for job in POINTCLOUD_INFERENCE_JOBS.values()):
			return True
			return False


			def pointcloud_annotation_source_job(job_id: str) -> dict[str, Any] \| None:
			with POINTCLOUD_ANNOTATION_SOURCE_JOBS_LOCK:
			value = POINTCLOUD_ANNOTATION_SOURCE_JOBS.get(job_id)
			return dict(value) if value else None


			def execute_pointcloud_annotation_source_job(
			root: Path,
			job_id: str,
			processed_path: Path,
			output: Path,
			source_sha256: str,
			source_bytes: int,
			) -> None:
			with POINTCLOUD_ANNOTATION_SOURCE_JOBS_LOCK:
			POINTCLOUD_ANNOTATION_SOURCE_JOBS[job_id].update({"status": "running", "stage": "preparing_annotation_preview", "startedAt": datetime.now(UTC).isoformat()})
			command = [
			str(root / ".venvs" / POINTCLOUD_CPU_ENVIRONMENT / "Scripts" / "python.exe"),
			str(root / "capabilities" / "05-3d-pointcloud" / "prepare_annotation_source.py"),
			"--input", str(processed_path), "--output", str(output),
			]
			try:
			with RUN_LOCK:
			completed = subprocess.run(command, cwd=root, capture_output=True, text=True, timeout=900, check=False)
			if completed.returncode:
			message = (completed.stderr or completed.stdout or "Unknown script error.").strip().splitlines()[-1]
			raise RuntimeError(f"Processing failed: {message[:600]}")
			metadata_path = output / "run_metadata.json"
			if not metadata_path.is_file():
			raise RuntimeError("Point-cloud processing finished without result metadata.")
			metadata = load_json(metadata_path)
			run_id = pointcloud_annotation_source_job(job_id)["runId"]
			metadata["input_dir"] = relative_path(root, processed_path.parent)
			metadata["raw_input_dir"] = relative_path(root, root / "shared" / "data" / "raw" / "05-3d-pointcloud" / "annotation-source-runs" / run_id)
			metadata["source_sha256"] = {processed_path.name: source_sha256}
			metadata["source_bytes"] = {processed_path.name: source_bytes}
			metadata["annotation_source_job"] = True
			metadata_path.write_text(json.dumps(metadata, ensure_ascii=False, indent=2), encoding="utf-8")
			source = next((item for item in standalone_pointcloud_annotation_sources(root) if item["runId"] == run_id), None)
			if not source:
			raise RuntimeError("Annotation preview finished without a discoverable annotation source.")
			with POINTCLOUD_ANNOTATION_SOURCE_JOBS_LOCK:
			POINTCLOUD_ANNOTATION_SOURCE_JOBS[job_id].update({"status": "complete", "stage": "complete", "completedAt": datetime.now(UTC).isoformat(), "source": source})
			except Exception as exc:
			with POINTCLOUD_ANNOTATION_SOURCE_JOBS_LOCK:
			POINTCLOUD_ANNOTATION_SOURCE_JOBS[job_id].update({"status": "failed", "stage": "failed", "completedAt": datetime.now(UTC).isoformat(), "error": str(exc)[:700]})


			def pointcloud_training_job(job_id: str) -> dict[str, Any] \| None:
			with POINTCLOUD_TRAINING_JOBS_LOCK:
			value = POINTCLOUD_TRAINING_JOBS.get(job_id)
			return dict(value) if value else None


			def execute_pointcloud_training_job(root: Path, job_id: str, annotation: Path, output: Path, execution: dict[str, str]) -> None:
			def execute_pointcloud_training_job(root: Path, job_id: str, annotation: Path, output: Path, execution: dict[str, str], trainer: str = "rgb_xyz_baseline") -> None:
			with POINTCLOUD_TRAINING_JOBS_LOCK:
			POINTCLOUD_TRAINING_JOBS[job_id].update({"status": "running", "stage": "training", "startedAt": datetime.now(UTC).isoformat()})
			command = [execution["python"], str(root / "capabilities" / "05-3d-pointcloud" / "train_pointcloud_semantic_model.py"), "--annotation", str(annotation), "--output", str(output), "--device", execution["device"]]
			scripts = {
			"rgb_xyz_baseline": "train_pointcloud_semantic_model.py",
			"multiview_local_attention_baseline": "train_multiview_point_transformer.py",
			}
			script = scripts.get(trainer)
			if not script:
			raise ApiError("Unsupported point-cloud training workflow.")
			command = [execution["python"], str(root / "capabilities" / "05-3d-pointcloud" / script), "--annotation", str(annotation), "--output", str(output), "--device", execution["device"]]
			try:
			with RUN_LOCK:
			completed = subprocess.run(command, cwd=root, capture_output=True, text=True, timeout=14_400, check=False)
			@@ -547,8 +958,11 @@
			preview = output / "predicted-semantic-preview.ply"
			if not all(path.is_file() for path in (metrics, model, preview)):
			raise ApiError("Training finished without model, metrics, and predicted preview artifacts.")
			metadata = output / "run_metadata.json"
			if metadata.is_file():
			record_execution_metadata(metadata, execution)
			with POINTCLOUD_TRAINING_JOBS_LOCK:
			POINTCLOUD_TRAINING_JOBS[job_id].update({"status": "complete", "stage": "complete", "completedAt": datetime.now(UTC).isoformat(), "artifactRoot": relative_path(root, output), "metrics": relative_path(root, metrics), "model": relative_path(root, model), "preview": relative_path(root, preview)})
			POINTCLOUD_TRAINING_JOBS[job_id].update({"status": "complete", "stage": "complete", "completedAt": datetime.now(UTC).isoformat(), "artifactRoot": relative_path(root, output), "metrics": relative_path(root, metrics), "model": relative_path(root, model), "preview": relative_path(root, preview), "trainer": trainer})
			except Exception as exc:
			with POINTCLOUD_TRAINING_JOBS_LOCK:
			POINTCLOUD_TRAINING_JOBS[job_id].update({"status": "failed", "stage": "failed", "completedAt": datetime.now(UTC).isoformat(), "error": str(exc)[:700]})
			@@ -562,7 +976,7 @@
			metrics_path = model_path.with_name("metrics.json")
			metrics = load_json(metrics_path)
			classes = metrics.get("classes")
			if metrics.get("capability") != "05-3d-pointcloud" or metrics.get("classification") != "B" or not isinstance(classes, dict):
			if metrics.get("capability") != "05-3d-pointcloud" or metrics.get("classification") != "B" or metrics.get("model_input_kind", "rgb_xyz") != "rgb_xyz" or not isinstance(classes, dict):
			continue
			class_codes = sorted(str(code) for code in classes if str(code).isdigit())
			if len(class_codes) < 2:
			@@ -584,6 +998,69 @@
			with POINTCLOUD_INFERENCE_JOBS_LOCK:
			value = POINTCLOUD_INFERENCE_JOBS.get(job_id)
			return dict(value) if value else None


			def latest_pointcloud_auto_annotation_job(root: Path, source_id: str, model_id: str) -> dict[str, Any] \| None:
			"""Recover a completed local automatic-annotation run after a server restart."""
			source = next((item for item in pointcloud_annotation_sources(root) if item["id"] == source_id), None)
			model = next((item for item in pointcloud_semantic_models(root) if item["id"] == model_id), None)
			if not source or not model:
			raise ApiError("The selected annotation source or trained model is unavailable.")
			output_root = root / "shared" / "outputs" / "05-3d-pointcloud" / "auto-annotation-runs"
			records: list[dict[str, Any]] = []
			model_sha256 = file_sha256(root / str(model["model"]))
			for metadata_path in output_root.glob("*/run_metadata.json"):
			artifact = metadata_path.parent
			metadata = load_json(metadata_path)
			prediction = metadata.get("prediction")
			input_record = metadata.get("input")
			model_record = metadata.get("model")
			processing = metadata.get("processing")
			automatic = prediction.get("automatic_annotation") if isinstance(prediction, dict) else None
			preview = artifact / "predicted-semantic-preview.ply"
			classified_las = artifact / "predicted-semantic-classified.las"
			class_counts = artifact / "class-counts.csv"
			summary = artifact / "prediction-summary.json"
			candidates = artifact / "automatic-annotation-candidates.json"
			if (
			metadata.get("capability") != "05-3d-pointcloud"
			or not isinstance(input_record, dict)
			or not isinstance(model_record, dict)
			or not isinstance(processing, dict)
			or not isinstance(automatic, dict)
			or input_record.get("sha256") != source.get("sha256")
			or model_record.get("sha256") != model_sha256
			or not all(path.is_file() for path in (preview, classified_las, class_counts, summary, candidates))
			):
			continue
			confidence = automatic.get("candidate_confidence")
			if not isinstance(confidence, (int, float)):
			continue
			records.append({
			"id": f"recovered-{artifact.name}",
			"runId": artifact.name,
			"modelId": model_id,
			"sourceId": source_id,
			"inputName": str(source["file"]),
			"candidateConfidence": float(confidence),
			"status": "complete",
			"stage": "complete",
			"requestedDevice": processing.get("requested_device"),
			"device": processing.get("device", "cpu"),
			"environment": processing.get("environment"),
			"torchVersion": metadata.get("versions", {}).get("torch") if isinstance(metadata.get("versions"), dict) else None,
			"fallbackUsed": processing.get("fallback_used", False),
			"fallbackReason": processing.get("fallback_reason"),
			"createdAt": str(metadata.get("created_at") or ""),
			"artifactRoot": relative_path(root, artifact),
			"metadata": relative_path(root, metadata_path),
			"preview": relative_path(root, preview),
			"classifiedLas": relative_path(root, classified_las),
			"classCounts": relative_path(root, class_counts),
			"summary": relative_path(root, summary),
			"candidateFile": relative_path(root, candidates),
			})
			return max(records, key=lambda item: (item["createdAt"], item["runId"])) if records else None


			def validate_pointcloud_model_input(path: Path) -> None:
			@@ -611,10 +1088,12 @@
			raise ApiError("LAS/LAZ 文件没有点记录。请选择包含实际 RGB 点位的完整点云文件，而不是空分块。")


			def execute_pointcloud_inference_job(root: Path, job_id: str, model: Path, source: Path, output: Path, execution: dict[str, str]) -> None:
			def execute_pointcloud_inference_job(root: Path, job_id: str, model: Path, source: Path, output: Path, execution: dict[str, str], annotation_source_id: str \| None = None, candidate_confidence: float \| None = None) -> None:
			with POINTCLOUD_INFERENCE_JOBS_LOCK:
			POINTCLOUD_INFERENCE_JOBS[job_id].update({"status": "running", "stage": "inference", "startedAt": datetime.now(UTC).isoformat()})
			command = [execution["python"], str(root / "capabilities" / "05-3d-pointcloud" / "apply_pointcloud_semantic_model.py"), "--model", str(model), "--input", str(source), "--output", str(output), "--device", execution["device"]]
			if annotation_source_id:
			command.extend(["--annotation-source-id", annotation_source_id, "--candidate-confidence", f"{candidate_confidence or 0.95:.6f}"])
			try:
			with RUN_LOCK:
			completed = subprocess.run(command, cwd=root, capture_output=True, text=True, timeout=14_400, check=False)
			@@ -628,8 +1107,12 @@
			summary = output / "prediction-summary.json"
			if not all(path.is_file() for path in (metadata, preview, classified_las, counts, summary)):
			raise ApiError("Model inference finished without all expected prediction artifacts.")
			candidates = output / "automatic-annotation-candidates.json"
			if annotation_source_id and not candidates.is_file():
			raise ApiError("Automatic annotation finished without the expected candidate artifact.")
			record_execution_metadata(metadata, execution)
			with POINTCLOUD_INFERENCE_JOBS_LOCK:
			POINTCLOUD_INFERENCE_JOBS[job_id].update({"status": "complete", "stage": "complete", "completedAt": datetime.now(UTC).isoformat(), "artifactRoot": relative_path(root, output), "metadata": relative_path(root, metadata), "preview": relative_path(root, preview), "classifiedLas": relative_path(root, classified_las), "classCounts": relative_path(root, counts), "summary": relative_path(root, summary)})
			POINTCLOUD_INFERENCE_JOBS[job_id].update({"status": "complete", "stage": "complete", "completedAt": datetime.now(UTC).isoformat(), "artifactRoot": relative_path(root, output), "metadata": relative_path(root, metadata), "preview": relative_path(root, preview), "classifiedLas": relative_path(root, classified_las), "classCounts": relative_path(root, counts), "summary": relative_path(root, summary), "candidateFile": relative_path(root, candidates) if annotation_source_id else None})
			except Exception as exc:
			with POINTCLOUD_INFERENCE_JOBS_LOCK:
			POINTCLOUD_INFERENCE_JOBS[job_id].update({"status": "failed", "stage": "failed", "completedAt": datetime.now(UTC).isoformat(), "error": str(exc)[:700]})
			@@ -700,6 +1183,161 @@
			return sorted(records, key=lambda item: (item["createdAt"], item["id"]), reverse=True)


			# A console run is removable only when it was created in the fixed ``runs``
			# layout. Discovery also exposes baseline and validation artifacts, but those
			# are project evidence rather than disposable console-owned copies.
			RUN_DELETION_CAPABILITIES: dict[str, tuple[str, Any]] = {
			"00-change-detection": ("00-change-detection", change_runs),
			"01-object-detection": ("01-object-detection", detection_runs),
			"02-semantic-mapping": ("02-semantic-mapping", semantic_runs),
			"04-spatial-measurement": ("04-spatial-measurement", measurement_runs),
			"05-3d-pointcloud": ("05-3d-pointcloud", pointcloud_runs),
			"07-risk-rule-engine": ("07-risk-rule-engine", risk_rule_runs),
			"09-anomaly-detection": ("09-anomaly-detection", anomaly_runs),
			"15-trajectory-analysis": ("15-trajectory-analysis", trajectory_runs),
			}


			def valid_run_deletion_id(run_id: str) -> bool:
			return bool(run_id) and len(run_id) <= 160 and "/" not in run_id and "\\" not in run_id and not SAFE_FILE_NAME.search(run_id)


			def existing_directory(path: Path, parent: Path) -> Path \| None:
			"""Return an existing direct child directory, never a caller-supplied path."""
			try:
			resolved_parent = parent.resolve()
			resolved = path.resolve()
			resolved.relative_to(resolved_parent)
			except (OSError, ValueError):
			return None
			return resolved if resolved.is_dir() else None


			def run_deletion_plan(root: Path, capability: str, run_id: str) -> dict[str, Any]:
			if capability not in RUN_DELETION_CAPABILITIES:
			raise ApiError("This capability does not expose removable console runs.")
			if not valid_run_deletion_id(run_id):
			raise ApiError("Invalid run id.")
			capability_dir, discover = RUN_DELETION_CAPABILITIES[capability]
			record = next((item for item in discover(root) if item.get("id") == run_id), None)
			if not record:
			raise ApiError("The selected result is unavailable.")

			output_root = (root / "shared" / "outputs" / capability_dir).resolve()
			expected_output = (output_root / "runs" / run_id).resolve()
			artifact_value = record.get("artifactRoot")
			try:
			artifact = (root / str(artifact_value)).resolve()
			except OSError as exc:
			raise ApiError("The selected result has an invalid artifact location.") from exc
			if artifact != expected_output:
			return {
			"capability": capability,
			"runId": run_id,
			"label": str(record.get("label") or run_id),
			"removable": False,
			"reason": "This is a built-in baseline, validation artifact, or external result. It was not created in the console-owned run layout.",
			"outputDirectories": [],
			"rawDirectories": [],
			"processedDirectories": [],
			"dependentDirectories": [],
			"preservesExternalInputs": True,
			}

			output_directories = [expected_output] if existing_directory(expected_output, output_root / "runs") else []
			raw_root = (root / "shared" / "data" / "raw" / capability_dir).resolve()
			processed_root = (root / "shared" / "data" / "processed" / capability_dir).resolve()
			raw_candidates = [raw_root / "runs" / run_id]
			processed_candidates = [processed_root / run_id, processed_root / "runs" / run_id]
			raw_directories = [path for candidate in raw_candidates if (path := existing_directory(candidate, raw_root))]
			processed_directories = [path for candidate in processed_candidates if (path := existing_directory(candidate, processed_root))]
			dependent_directories: list[Path] = []
			if capability == "00-change-detection":
			scan_root = output_root / "parameter-scans" / run_id
			if path := existing_directory(scan_root, output_root / "parameter-scans"):
			dependent_directories.append(path)
			return {
			"capability": capability,
			"runId": run_id,
			"label": str(record.get("label") or run_id),
			"removable": bool(output_directories),
			"reason": None if output_directories else "The console-owned output directory is missing, so no deletion is performed.",
			"outputDirectories": [relative_path(root, path) for path in output_directories],
			"rawDirectories": [relative_path(root, path) for path in raw_directories],
			"processedDirectories": [relative_path(root, path) for path in processed_directories],
			"dependentDirectories": [relative_path(root, path) for path in dependent_directories],
			"preservesExternalInputs": True,
			}


			def delete_console_run(root: Path, capability: str, run_id: str) -> dict[str, Any]:
			plan = run_deletion_plan(root, capability, run_id)
			if not plan["removable"]:
			raise ApiError(str(plan["reason"] or "The selected result cannot be removed."))
			directories = [plan["dependentDirectories"], plan["processedDirectories"], plan["rawDirectories"], plan["outputDirectories"]]
			for relative in directories:
			location = (root / relative).resolve()
			# Every entry was created by run_deletion_plan from fixed roots above.
			if location.is_dir():
			shutil.rmtree(location)
			return {"capability": capability, "runId": run_id, "removedDirectories": directories, "preservesExternalInputs": True}


			def pointcloud_semantic_model_deletion_plan(root: Path, model_id: str) -> dict[str, Any]:
			if not valid_run_deletion_id(model_id):
			raise ApiError("Invalid semantic model id.")
			model = next((item for item in pointcloud_semantic_models(root) if item["id"] == model_id), None)
			if not model:
			raise ApiError("The selected trained model is unavailable.")
			output_root = (root / "shared" / "outputs" / "05-3d-pointcloud").resolve()
			model_root = (output_root / "training-runs" / model_id).resolve()
			model_path = (root / str(model["model"])).resolve()
			if model_path != model_root / "model.pt" or not existing_directory(model_root, output_root / "training-runs"):
			raise ApiError("The selected trained model is outside the console-owned training layout.")
			model_sha256 = file_sha256(model_path)
			inference_directories: list[Path] = []
			for metadata_path in (output_root / "model-inference-runs").glob("*/run_metadata.json"):
			metadata = load_json(metadata_path)
			value = metadata.get("model")
			candidate = value.get("path") if isinstance(value, dict) else None
			try:
			if isinstance(candidate, str) and Path(candidate).resolve() == model_path:
			inference_directories.append(metadata_path.parent.resolve())
			except OSError:
			continue
			auto_directories: list[Path] = []
			for metadata_path in (output_root / "auto-annotation-runs").glob("*/run_metadata.json"):
			value = load_json(metadata_path).get("model")
			if isinstance(value, dict) and value.get("sha256") == model_sha256:
			auto_directories.append(metadata_path.parent.resolve())
			raw_root = (root / "shared" / "data" / "raw" / "05-3d-pointcloud" / "model-inference-runs").resolve()
			processed_root = (root / "shared" / "data" / "processed" / "05-3d-pointcloud" / "model-inference-runs").resolve()
			raw_directories = [path for directory in inference_directories if (path := existing_directory(raw_root / directory.name, raw_root))]
			processed_directories = [path for directory in inference_directories if (path := existing_directory(processed_root / directory.name, processed_root))]
			return {
			"modelId": model_id,
			"label": str(model["label"]),
			"removable": True,
			"trainingDirectories": [relative_path(root, model_root)],
			"inferenceDirectories": [relative_path(root, path) for path in inference_directories],
			"autoAnnotationDirectories": [relative_path(root, path) for path in auto_directories],
			"rawDirectories": [relative_path(root, path) for path in raw_directories],
			"processedDirectories": [relative_path(root, path) for path in processed_directories],
			"preservesAnnotationRevisions": True,
			"preservesExternalInputs": True,
			}


			def delete_pointcloud_semantic_model(root: Path, model_id: str) -> dict[str, Any]:
			plan = pointcloud_semantic_model_deletion_plan(root, model_id)
			directories = [plan["autoAnnotationDirectories"], plan["inferenceDirectories"], plan["processedDirectories"], plan["rawDirectories"], *plan["trainingDirectories"]]
			for relative in directories:
			location = (root / relative).resolve()
			if location.is_dir():
			shutil.rmtree(location)
			return {"modelId": model_id, "removedDirectories": directories, "preservesAnnotationRevisions": True, "preservesExternalInputs": True}


			def anomaly_job(job_id: str) -> dict[str, Any] \| None:
			with ANOMALY_JOB_LOCK:
			value = ANOMALY_JOBS.get(job_id)
			@@ -709,7 +1347,21 @@
			def photo_reconstruction_job(job_id: str) -> dict[str, Any] \| None:
			with PHOTO_RECONSTRUCTION_JOBS_LOCK:
			value = PHOTO_RECONSTRUCTION_JOBS.get(job_id)
			return dict(value) if value else None
			job = dict(value) if value else None
			if not job:
			return None
			progress_path = job.pop("progressPath", None)
			if job["status"] == "complete":
			job["progress"] = {"percent": 100, "stage": "complete", "message": "照片重建已完成，结果已加入案例库。", "inputImages": job["inputImages"], "estimate": True}
			return job
			if job["status"] == "failed":
			job["progress"] = {"percent": 0, "stage": "failed", "message": job.get("error") or "照片重建失败。", "inputImages": job["inputImages"], "estimate": True}
			return job
			if isinstance(progress_path, str):
			progress = load_json(Path(progress_path))
			if progress:
			job["progress"] = progress
			return job


			def run_background_command(command: list[str], root: Path, timeout: int) -> None:
			@@ -737,6 +1389,7 @@
			"--input", str(processed_root), "--output", str(sparse_output),
			"--max-image-size", "2000", "--max-features", "18000",
			"--camera-model", "OPENCV",
			"--progress-file", str(processed_root / "photo_reconstruction_progress.json"),
			]
			if use_position_priors:
			sparse_command.extend(["--matching-mode", "spatial", "--matching-neighbors", "4", "--use-position-priors", "--prior-position-loss-scale-m", "0.05"])
			@@ -749,6 +1402,7 @@
			"--openmvs-bin", str(root / "shared" / "tools" / "openmvs-2.4.0" / "vc17" / "x64" / "Release"),
			"--threads", "12", "--max-resolution", "2400", "--dense-resolution-level", "0",
			"--dense-number-views", "8", "--dense-number-views-fuse", "2", "--target-faces", "800000",
			"--progress-file", str(processed_root / "photo_reconstruction_progress.json"),
			]
			try:
			with PHOTO_RECONSTRUCTION_JOBS_LOCK:
			@@ -869,6 +1523,17 @@

			def do_GET(self) -> None: # noqa: N802 - inherited standard-library method name
			path = urlsplit(self.path).path
			run_deletion_prefix = "/api/runs/"
			if path.startswith(run_deletion_prefix) and path.endswith("/deletion-plan"):
			try:
			parts = path[len(run_deletion_prefix):].split("/")
			if len(parts) != 3 or parts[2] != "deletion-plan":
			raise ApiError("Invalid result deletion-plan endpoint.")
			capability, run_id = (unquote(parts[0]), unquote(parts[1]))
			self.send_json(HTTPStatus.OK, {"plan": run_deletion_plan(self.root, capability, run_id)})
			except ApiError as exc:
			self.send_json(HTTPStatus.BAD_REQUEST, {"error": str(exc)})
			return
			if path == "/api/change-detection/runs":
			self.send_json(HTTPStatus.OK, {"runs": change_runs(self.root)})
			return
			@@ -905,11 +1570,73 @@
			if path == "/api/3d-pointcloud/annotation-sources":
			self.send_json(HTTPStatus.OK, {"sources": pointcloud_annotation_sources(self.root)})
			return
			if path == "/api/3d-pointcloud/annotation-detail":
			try:
			query = parse_qs(urlsplit(self.path).query)
			source_id = query.get("sourceId", [None])[0]
			values = [query.get(axis, [None])[0] for axis in ("x", "y", "z", "radius")]
			if not isinstance(source_id, str) or any(value is None for value in values):
			raise ApiError("Detail request must include sourceId, x, y, z, and radius.")
			try:
			x, y, z, radius = (float(value) for value in values)
			except (TypeError, ValueError) as exc:
			raise ApiError("Detail coordinates and radius must be numbers.") from exc
			payload = pointcloud_annotation_detail(self.root, source_id, (x, y, z), radius)
			self.send_response(HTTPStatus.OK)
			self.send_header("Content-Type", "application/octet-stream")
			self.send_header("Content-Length", str(len(payload)))
			self.send_header("Cache-Control", "no-store")
			self.end_headers()
			self.wfile.write(payload)
			except ApiError as exc:
			self.send_json(HTTPStatus.BAD_REQUEST, {"error": str(exc)})
			return
			deletion_plan_prefix = "/api/3d-pointcloud/annotation-source-deletion-plans/"
			if path.startswith(deletion_plan_prefix):
			try:
			source_id = unquote(path[len(deletion_plan_prefix):])
			self.send_json(HTTPStatus.OK, {"plan": pointcloud_annotation_source_deletion_plan(self.root, source_id)})
			except ApiError as exc:
			self.send_json(HTTPStatus.BAD_REQUEST, {"error": str(exc)})
			return
			if path == "/api/3d-pointcloud/annotation-classes":
			self.send_json(HTTPStatus.OK, {"classes": annotation_classes(self.root)})
			return
			if path == "/api/3d-pointcloud/annotations":
			self.send_json(HTTPStatus.OK, {"annotations": pointcloud_annotations(self.root)})
			return
			if path == "/api/3d-pointcloud/semantic-models":
			self.send_json(HTTPStatus.OK, {"models": pointcloud_semantic_models(self.root)})
			return
			semantic_model_plan_prefix = "/api/3d-pointcloud/semantic-model-deletion-plans/"
			if path.startswith(semantic_model_plan_prefix):
			try:
			model_id = unquote(path[len(semantic_model_plan_prefix):])
			self.send_json(HTTPStatus.OK, {"plan": pointcloud_semantic_model_deletion_plan(self.root, model_id)})
			except ApiError as exc:
			self.send_json(HTTPStatus.BAD_REQUEST, {"error": str(exc)})
			return
			if path == "/api/3d-pointcloud/auto-annotation-runs/latest":
			try:
			query = parse_qs(urlsplit(self.path).query)
			source_id = query.get("sourceId", [None])[0]
			model_id = query.get("modelId", [None])[0]
			if not isinstance(source_id, str) or not isinstance(model_id, str):
			raise ApiError("Automatic-annotation recovery needs sourceId and modelId.")
			self.send_json(HTTPStatus.OK, {"job": latest_pointcloud_auto_annotation_job(self.root, source_id, model_id)})
			except ApiError as exc:
			self.send_json(HTTPStatus.BAD_REQUEST, {"error": str(exc)})
			return
			review_prefix = "/api/3d-pointcloud/auto-annotation-review-drafts/"
			if path.startswith(review_prefix):
			try:
			run_id = unquote(path[len(review_prefix):])
			if not run_id or SAFE_FILE_NAME.search(run_id):
			raise ApiError("Invalid automatic-annotation run id.")
			review_path = self.root / "shared" / "outputs" / "05-3d-pointcloud" / "auto-annotation-runs" / run_id / "review-corrections.json"
			self.send_json(HTTPStatus.OK, {"review": load_json(review_path) if review_path.is_file() else None})
			except ApiError as exc:
			self.send_json(HTTPStatus.BAD_REQUEST, {"error": str(exc)})
			return
			if path.startswith("/api/3d-pointcloud/model-inference-jobs/"):
			job_id = path.rstrip("/").rsplit("/", 1)[-1]
			@@ -920,6 +1647,11 @@
			job_id = path.rstrip("/").rsplit("/", 1)[-1]
			job = pointcloud_training_job(job_id)
			self.send_json(HTTPStatus.OK if job else HTTPStatus.NOT_FOUND, {"job": job} if job else {"error": "Unknown point-cloud training job."})
			return
			if path.startswith("/api/3d-pointcloud/annotation-source-jobs/"):
			job_id = path.rstrip("/").rsplit("/", 1)[-1]
			job = pointcloud_annotation_source_job(job_id)
			self.send_json(HTTPStatus.OK if job else HTTPStatus.NOT_FOUND, {"job": job} if job else {"error": "Unknown point-cloud annotation source job."})
			return
			if path.startswith("/api/3d-pointcloud/photo-reconstruction-jobs/"):
			job_id = path.rstrip("/").rsplit("/", 1)[-1]
			@@ -976,11 +1708,26 @@
			if path == "/api/3d-pointcloud/annotations":
			self.send_json(HTTPStatus.CREATED, {"annotation": self.create_pointcloud_annotation(payload)})
			return
			if path == "/api/3d-pointcloud/annotation-classes":
			self.send_json(HTTPStatus.CREATED, {"class": self.create_pointcloud_annotation_class(payload)})
			return
			if path == "/api/3d-pointcloud/annotation-source-runs":
			self.send_json(HTTPStatus.ACCEPTED, {"job": self.create_pointcloud_annotation_source_run(payload)})
			return
			if path == "/api/3d-pointcloud/training-runs":
			self.send_json(HTTPStatus.ACCEPTED, {"job": self.create_pointcloud_training_run(payload)})
			return
			if path == "/api/3d-pointcloud/model-inference-runs":
			self.send_json(HTTPStatus.ACCEPTED, {"job": self.create_pointcloud_model_inference_run(payload)})
			return
			if path == "/api/3d-pointcloud/auto-annotation-runs":
			self.send_json(HTTPStatus.ACCEPTED, {"job": self.create_pointcloud_auto_annotation_run(payload)})
			return
			if path == "/api/3d-pointcloud/auto-annotation-acceptances":
			self.send_json(HTTPStatus.CREATED, {"annotation": self.accept_pointcloud_auto_annotation(payload)})
			return
			if path == "/api/3d-pointcloud/auto-annotation-review-drafts":
			self.send_json(HTTPStatus.CREATED, {"review": self.save_pointcloud_auto_annotation_review(payload)})
			return
			if path == "/api/3d-pointcloud/photo-reconstruction-runs":
			self.send_json(HTTPStatus.ACCEPTED, {"job": self.create_photo_reconstruction_run(payload)})
			@@ -1002,6 +1749,55 @@

			def do_DELETE(self) -> None: # noqa: N802 - annotation revisions are explicitly user-removable
			path = urlsplit(self.path).path
			run_prefix = "/api/runs/"
			if path.startswith(run_prefix):
			try:
			parts = path[len(run_prefix):].split("/")
			if len(parts) != 2:
			raise ApiError("Invalid result deletion endpoint.")
			capability, run_id = (unquote(parts[0]), unquote(parts[1]))
			self.send_json(HTTPStatus.OK, {"removed": delete_console_run(self.root, capability, run_id)})
			except ApiError as exc:
			self.send_json(HTTPStatus.BAD_REQUEST, {"error": str(exc)})
			except Exception as exc: # pragma: no cover - defensive deletion boundary
			self.log_error("console run deletion failed: %s", exc)
			self.send_json(HTTPStatus.INTERNAL_SERVER_ERROR, {"error": "Result deletion failed. Check the console terminal for details."})
			return
			semantic_model_prefix = "/api/3d-pointcloud/semantic-models/"
			if path.startswith(semantic_model_prefix):
			try:
			model_id = unquote(path[len(semantic_model_prefix):])
			self.send_json(HTTPStatus.OK, {"removed": delete_pointcloud_semantic_model(self.root, model_id)})
			except ApiError as exc:
			self.send_json(HTTPStatus.BAD_REQUEST, {"error": str(exc)})
			except Exception as exc: # pragma: no cover - defensive deletion boundary
			self.log_error("semantic model deletion failed: %s", exc)
			self.send_json(HTTPStatus.INTERNAL_SERVER_ERROR, {"error": "Semantic model deletion failed. Check the console terminal for details."})
			return
			source_prefix = "/api/3d-pointcloud/annotation-sources/"
			if path.startswith(source_prefix):
			try:
			source_id = unquote(path[len(source_prefix):])
			if not source_id or "/" in source_id or len(source_id) > 300:
			raise ApiError("Invalid annotation source id.")
			self.send_json(HTTPStatus.OK, {"removed": self.delete_pointcloud_annotation_source(source_id)})
			except ApiError as exc:
			self.send_json(HTTPStatus.BAD_REQUEST, {"error": str(exc)})
			except Exception as exc: # pragma: no cover - defensive deletion boundary
			self.log_error("annotation source deletion failed: %s", exc)
			self.send_json(HTTPStatus.INTERNAL_SERVER_ERROR, {"error": "Annotation source deletion failed. Check the console terminal for details."})
			return
			class_prefix = "/api/3d-pointcloud/annotation-classes/"
			if path.startswith(class_prefix):
			try:
			code_value = path[len(class_prefix):]
			if not code_value.isdigit():
			raise ApiError("Invalid annotation class code.")
			deleted = self.delete_pointcloud_annotation_class(int(code_value))
			self.send_json(HTTPStatus.OK, {"deletedCode": deleted})
			except ApiError as exc:
			self.send_json(HTTPStatus.BAD_REQUEST, {"error": str(exc)})
			return
			prefix = "/api/3d-pointcloud/annotations/"
			if not path.startswith(prefix):
			self.send_json(HTTPStatus.NOT_FOUND, {"error": "Unknown local API endpoint."})
			@@ -1208,10 +2004,14 @@

			def create_detection_run(self, payload: dict[str, Any]) -> dict[str, Any]:
			uploads = payload.get("images")
			requested_device = payload.get("device", "auto")
			if not isinstance(uploads, list) or not uploads:
			raise ApiError("Object-detection request must include at least one image.")
			if len(uploads) > MAX_IMAGES_PER_RUN:
			raise ApiError(f"A local run accepts at most {MAX_IMAGES_PER_RUN} images.")
			if not isinstance(requested_device, str):
			raise ApiError("Object-detection device must be auto, cpu, or cuda.")
			execution = object_detection_execution_environment(self.root, requested_device)
			decoded = [decode_upload(item, {".jpg", ".jpeg", ".png"}) for item in uploads]
			if len({name.casefold() for name, _ in decoded}) != len(decoded):
			raise ApiError("Uploaded image names must be unique within one run.")
			@@ -1221,16 +2021,18 @@
			for name, content in decoded:
			(raw_root / name).write_bytes(content)
			output = self.root / "shared" / "outputs" / "01-object-detection" / "runs" / run_id
			execution = object_detection_execution_environment(self.root)
			with RUN_LOCK:
			self.run_command([execution["python"], str(self.root / "capabilities" / "01-object-detection" / "run_detection.py"), "--input", str(raw_root), "--output", str(output), "--device", execution["device"]], 1200)
			if not (output / "run_metadata.json").is_file():
			metadata_path = output / "run_metadata.json"
			if not metadata_path.is_file():
			raise ApiError("Detection script finished without the expected result metadata.")
			record_execution_metadata(metadata_path, execution)
			return next(item for item in detection_runs(self.root) if item["id"] == run_id)

			def create_change_run(self, payload: dict[str, Any]) -> dict[str, Any]:
			files = payload.get("files")
			uploads = payload.get("uploads")
			requested_device = payload.get("device", "auto")
			staged: dict[str, tuple[str, Path]] = {}
			if isinstance(uploads, dict):
			staged["before"] = self.resolve_change_upload(uploads.get("before"), "before")
			@@ -1255,6 +2057,9 @@
			max_dimension = int(max_dimension_value)
			if max_dimension != CHANGE_MAX_DIMENSION_AUTO and not CHANGE_MAX_DIMENSION_MIN <= max_dimension <= CHANGE_MAX_DIMENSION_MAX:
			raise ApiError("Change-detection resolution must be 0 or between 512 and 4096.")
			if not isinstance(requested_device, str):
			raise ApiError("Change-detection device must be auto, cpu, or cuda.")
			execution = change_detection_execution_environment(self.root, requested_device)
			if staged:
			before_name, after_name = staged["before"][0], staged["after"][0]
			else:
			@@ -1273,7 +2078,6 @@
			after_path.write_bytes(decoded_after[1])
			processed_root = self.root / "shared" / "data" / "processed" / "00-change-detection" / run_id
			output = self.root / "shared" / "outputs" / "00-change-detection" / "runs" / run_id
			execution = change_detection_execution_environment(self.root)
			with RUN_LOCK:
			self.run_command(
			[
			@@ -1417,6 +2221,7 @@

			def create_change_scan(self, payload: dict[str, Any]) -> dict[str, Any]:
			uploads = payload.get("uploads")
			requested_device = payload.get("device", "auto")
			if not isinstance(uploads, dict):
			raise ApiError("Parameter scan must contain staged before and after uploads.")
			staged = {
			@@ -1433,6 +2238,9 @@
			max_dimension = int(max_dimension_value)
			if max_dimension != CHANGE_MAX_DIMENSION_AUTO and not CHANGE_MAX_DIMENSION_MIN <= max_dimension <= CHANGE_MAX_DIMENSION_MAX:
			raise ApiError("Change-detection resolution must be 0 or between 512 and 4096.")
			if not isinstance(requested_device, str):
			raise ApiError("Change-detection device must be auto, cpu, or cuda.")
			execution = change_detection_execution_environment(self.root, requested_device)
			run_id = make_run_id("scan")
			raw_root = self.root / "shared" / "data" / "raw" / "00-change-detection" / "runs" / run_id
			before_path = raw_root / "before" / staged["before"][0]
			@@ -1453,10 +2261,16 @@
			"minimumAreas": areas,
			"processingMode": processing_mode,
			"maxDimension": max_dimension,
			"requestedDevice": execution["requestedDevice"],
			"device": execution["device"],
			"environment": execution["environment"],
			"torchVersion": execution["torchVersion"],
			"fallbackUsed": execution["fallbackUsed"],
			"fallbackReason": execution["fallbackReason"],
			}
			thread = threading.Thread(
			target=self._run_change_scan,
			args=(run_id, before_path, after_path, processed_root, inference_output, scan_output, thresholds, areas, processing_mode, max_dimension),
			args=(run_id, before_path, after_path, processed_root, inference_output, scan_output, thresholds, areas, processing_mode, max_dimension, execution),
			daemon=True,
			name=f"change-scan-{run_id}",
			)
			@@ -1480,11 +2294,11 @@
			areas: list[int],
			processing_mode: str,
			max_dimension: int,
			execution: dict[str, Any],
			) -> None:
			execution = change_detection_execution_environment(self.root)
			python = execution["python"]
			try:
			self._update_scan_job(run_id, status="running", phase="inference", device=execution["device"], environment=execution["environment"], torchVersion=execution["torchVersion"])
			self._update_scan_job(run_id, status="running", phase="inference")
			with RUN_LOCK:
			self.run_command(
			[
			@@ -1502,6 +2316,7 @@
			SCAN_JOB_TIMEOUT,
			)
			inference_metadata_path = inference_output / "run_metadata.json"
			record_execution_metadata(inference_metadata_path, execution)
			inference_metadata = load_json(inference_metadata_path)
			inference_metadata["kind"] = "parameter-scan-inference"
			inference_metadata["scan_job_id"] = run_id
			@@ -1540,6 +2355,9 @@
			"device": execution["device"],
			"environment": execution["environment"],
			"torch_version": execution["torchVersion"],
			"requested_device": execution["requestedDevice"],
			"fallback_used": execution["fallbackUsed"],
			"fallback_reason": execution["fallbackReason"],
			"raw_input_dir": relative_path(self.root, before_path.parent.parent),
			"processed_input_dir": relative_path(self.root, processed_root),
			}
			@@ -1741,6 +2559,7 @@
			metadata_path = output / "run_metadata.json"
			if not metadata_path.is_file():
			raise ApiError("3D point-cloud script finished without the expected result metadata.")
			record_execution_metadata(metadata_path, execution)
			metadata = load_json(metadata_path)
			metadata["input_dir"] = relative_path(self.root, processed_root)
			metadata["raw_input_dir"] = relative_path(self.root, raw_root)
			@@ -1759,14 +2578,14 @@
			source = next((item for item in pointcloud_annotation_sources(self.root) if item["id"] == source_id), None)
			if not source:
			raise ApiError("The selected generated annotation source is unavailable.")
			if len(labels) > MAX_ANNOTATION_LABELS:
			raise ApiError(f"An annotation revision accepts at most {MAX_ANNOTATION_LABELS} labelled points.")
			active_classes = annotation_classes(self.root)
			class_by_code = {item["code"]: item for item in active_classes}
			compact: dict[int, int] = {}
			for item in labels:
			if not isinstance(item, list) or len(item) != 2 or not all(isinstance(value, int) for value in item):
			raise ApiError("Each annotation label must be [pointIndex, classCode].")
			index, code = item
			if index < 0 or index >= int(source["pointCount"]) or code not in POINTCLOUD_CLASS_CODES:
			if index < 0 or index >= int(source["pointCount"]) or code not in class_by_code:
			raise ApiError("Annotation contains an out-of-range point index or unsupported class code.")
			compact[index] = code
			if not compact:
			@@ -1775,18 +2594,128 @@
			location = self.root / "shared" / "outputs" / "05-3d-pointcloud" / "annotations" / annotation_id
			location.mkdir(parents=True, exist_ok=False)
			source_path = self.root / str(source["artifactRoot"]) / str(source["file"])
			class_counts = {str(code): sum(value == code for value in compact.values()) for code in sorted(POINTCLOUD_CLASS_CODES)}
			class_counts = {str(code): sum(value == code for value in compact.values()) for code in sorted(class_by_code)}
			document = {
			"schema_version": 1, "id": annotation_id, "created_at": datetime.now(UTC).isoformat(),
			"source_id": source_id, "source_path": str(source_path.resolve()), "source_sha256": source["sha256"],
			"source_run_id": source["runId"], "point_count": int(source["pointCount"]),
			"labels": [[index, code] for index, code in sorted(compact.items())], "class_counts": class_counts,
			"class_schema": {str(code): {"code": code} for code in sorted(POINTCLOUD_CLASS_CODES)},
			"class_schema": {str(code): item for code, item in sorted(class_by_code.items())},
			"provenance": "human_confirmed_point_labels_only",
			}
			path = location / "annotation.json"
			path.write_text(json.dumps(document, ensure_ascii=False, indent=2), encoding="utf-8")
			return {"id": annotation_id, "sourceId": source_id, "path": relative_path(self.root, path), "labelCount": len(compact), "classCounts": class_counts, "createdAt": document["created_at"]}

			def create_pointcloud_annotation_class(self, payload: dict[str, Any]) -> dict[str, Any]:
			key, label, color = payload.get("key"), payload.get("label"), payload.get("color")
			if not isinstance(key, str) or not isinstance(label, str) or not isinstance(color, list):
			raise ApiError("A class key, label, and RGB color are required.")
			current = annotation_classes(self.root)
			used_codes = {item["code"] for item in current}
			code = next((value for value in range(17, 256) if value not in used_codes), None)
			if code is None:
			raise ApiError("All LAS-compatible annotation class codes are already in use.")
			candidate = {"code": code, "key": key, "label": label, "color": color, "builtIn": False}
			updated = validate_annotation_classes([*current, candidate], allow_builtin=True)
			write_annotation_classes(self.root, updated)
			return next(item for item in updated if item["code"] == code)

			def delete_pointcloud_annotation_class(self, code: int) -> int:
			current = annotation_classes(self.root)
			item = next((value for value in current if value["code"] == code), None)
			if not item:
			raise ApiError("The annotation class is unavailable.")
			if item["builtIn"]:
			raise ApiError("Built-in annotation classes cannot be deleted.")
			if code in annotation_class_codes_in_use(self.root):
			raise ApiError("This annotation class is used by a saved annotation revision and cannot be deleted.")
			write_annotation_classes(self.root, [value for value in current if value["code"] != code])
			return code

			def create_pointcloud_annotation_source_run(self, payload: dict[str, Any]) -> dict[str, Any]:
			name, staged_path, expected_sha256 = self.resolve_pointcloud_upload(payload.get("pointCloud"))
			run_id = make_run_id("annotation-source")
			raw_root = self.root / "shared" / "data" / "raw" / "05-3d-pointcloud" / "annotation-source-runs" / run_id
			processed_root = self.root / "shared" / "data" / "processed" / "05-3d-pointcloud" / "annotation-source-runs" / run_id
			output = self.root / "shared" / "outputs" / "05-3d-pointcloud" / "runs" / run_id
			raw_root.mkdir(parents=True, exist_ok=False)
			processed_root.mkdir(parents=True, exist_ok=False)
			raw_path = raw_root / name
			shutil.copyfile(staged_path, raw_path)
			source_sha256 = file_sha256(raw_path)
			if expected_sha256 and source_sha256 != expected_sha256:
			raise ApiError("Uploaded point-cloud checksum changed while staging.")
			processed_path = processed_root / name
			shutil.copyfile(raw_path, processed_path)
			shutil.rmtree(staged_path.parent)
			python = self.root / ".venvs" / POINTCLOUD_CPU_ENVIRONMENT / "Scripts" / "python.exe"
			if not python.is_file():
			raise ApiError("3D point-cloud CPU virtual environment is unavailable. Run the capability setup first.")
			job_id = uuid4().hex
			job = {"id": job_id, "runId": run_id, "inputName": name, "status": "queued", "stage": "queued", "createdAt": datetime.now(UTC).isoformat(), "sourceSha256": source_sha256}
			with POINTCLOUD_ANNOTATION_SOURCE_JOBS_LOCK:
			POINTCLOUD_ANNOTATION_SOURCE_JOBS[job_id] = job
			thread = threading.Thread(target=execute_pointcloud_annotation_source_job, args=(self.root, job_id, processed_path, output, source_sha256, raw_path.stat().st_size), daemon=True, name=f"annotation-source-{job_id[:8]}")
			thread.start()
			return dict(job)

			def delete_pointcloud_annotation_source(self, source_id: str) -> dict[str, Any]:
			plan = pointcloud_annotation_source_deletion_plan(self.root, source_id)
			source = next(item for item in pointcloud_annotation_sources(self.root) if item["id"] == source_id)
			run_id = str(plan["runId"])
			sibling_ids = {item["id"] for item in pointcloud_annotation_sources(self.root) if item["runId"] == run_id}
			output_root = (self.root / "shared" / "outputs" / "05-3d-pointcloud").resolve()
			annotation_root = output_root / "annotations"
			annotations = [path.parent for path in annotation_root.glob("*/annotation.json") if load_json(path).get("source_id") in sibling_ids]
			annotation_paths = {path / "annotation.json" for path in annotations}
			training_root = output_root / "training-runs"
			training = []
			for metrics_path in training_root.glob("*/metrics.json"):
			value = load_json(metrics_path).get("annotation")
			try:
			if isinstance(value, str) and Path(value).resolve() in annotation_paths:
			training.append(metrics_path.parent)
			except OSError:
			continue
			training_models = {path / "model.pt" for path in training}
			inference_root = output_root / "model-inference-runs"
			inference = []
			for metadata_path in inference_root.glob("*/run_metadata.json"):
			model = load_json(metadata_path).get("model")
			value = model.get("path") if isinstance(model, dict) else None
			try:
			if isinstance(value, str) and Path(value).resolve() in training_models:
			inference.append(metadata_path.parent)
			except OSError:
			continue
			if active_pointcloud_source_dependencies(run_id, {path.name for path in annotations}, {path.name for path in training}):
			raise ApiError("This data source has a queued or running dependent task. Wait for it to finish before removing the full data chain.")
			artifact = (self.root / str(source["artifactRoot"])).resolve()
			raw_root = self.root / "shared" / "data" / "raw" / "05-3d-pointcloud"
			processed_root = self.root / "shared" / "data" / "processed" / "05-3d-pointcloud"
			raw = [path for path in (raw_root / "annotation-source-runs" / run_id, raw_root / "runs" / run_id) if path.is_dir()]
			processed = [path for path in (processed_root / "annotation-source-runs" / run_id, processed_root / "runs" / run_id) if path.is_dir()]
			targets = [inference, training, annotations, artifact, raw, *processed]
			unique: list[tuple[Path, Path]] = []
			seen: set[Path] = set()
			for target in targets:
			resolved = target.resolve()
			if resolved in seen:
			continue
			seen.add(resolved)
			if target in raw:
			allowed = raw_root
			elif target in processed:
			allowed = processed_root
			else:
			allowed = output_root
			assert_removable_pointcloud_directory(self.root, target, allowed)
			unique.append((target, allowed))
			# Dependents first; every target was resolved against a fixed local root.
			for target, _ in unique:
			shutil.rmtree(target)
			return {"sourceId": source_id, "runId": run_id, "removed": {"outputDirectories": int(plan["outputDirectories"]), "rawDirectories": int(plan["rawDirectories"]), "processedDirectories": int(plan["processedDirectories"]), "annotationRevisions": len(annotations), "trainingRuns": len(training), "inferenceRuns": len(inference), "siblingSources": int(plan["siblingSources"])}, "preservedExternalInputs": bool(plan["preservesExternalInputs"])}

			def create_pointcloud_training_run(self, payload: dict[str, Any]) -> dict[str, Any]:
			annotation_id = payload.get("annotationId")
			@@ -1799,19 +2728,27 @@
			record = load_json(annotation)
			if record.get("schema_version") != 1:
			raise ApiError("The selected annotation revision is unavailable.")
			source_id = record.get("source_id")
			source = next((item for item in pointcloud_annotation_sources(self.root) if item["id"] == source_id), None)
			if not source:
			raise ApiError("The annotation source is unavailable or its verification contract no longer passes.")
			if source.get("sourceHasRgb") is False:
			raise ApiError("This annotation source has no readable RGB values. It can be reviewed visually but cannot train the current RGB semantic model.")
			trainer = "multiview_local_attention_baseline" if str(source.get("sourceKind", "")).startswith("多视角照片特征融合") else "rgb_xyz_baseline"
			execution = pointcloud_execution_environment(self.root, device)
			job_id = uuid4().hex
			output = self.root / "shared" / "outputs" / "05-3d-pointcloud" / "training-runs" / make_run_id("semantic-model")
			job = {"id": job_id, "annotationId": annotation_id, "status": "queued", "stage": "queued", "requestedDevice": device, "device": execution["device"], "environment": execution["environment"], "torchVersion": execution["torchVersion"], "createdAt": datetime.now(UTC).isoformat()}
			output = self.root / "shared" / "outputs" / "05-3d-pointcloud" / "training-runs" / make_run_id("multiview-point-attention" if trainer == "multiview_local_attention_baseline" else "semantic-model")
			job = {"id": job_id, "annotationId": annotation_id, "trainer": trainer, "status": "queued", "stage": "queued", "requestedDevice": execution["requestedDevice"], "device": execution["device"], "environment": execution["environment"], "torchVersion": execution["torchVersion"], "fallbackUsed": execution["fallbackUsed"], "fallbackReason": execution["fallbackReason"], "createdAt": datetime.now(UTC).isoformat()}
			with POINTCLOUD_TRAINING_JOBS_LOCK:
			POINTCLOUD_TRAINING_JOBS[job_id] = job
			thread = threading.Thread(target=execute_pointcloud_training_job, args=(self.root, job_id, annotation, output, execution), daemon=True, name=f"pointcloud-training-{job_id[:8]}")
			thread = threading.Thread(target=execute_pointcloud_training_job, args=(self.root, job_id, annotation, output, execution, trainer), daemon=True, name=f"pointcloud-training-{job_id[:8]}")
			thread.start()
			return dict(job)

			def create_pointcloud_model_inference_run(self, payload: dict[str, Any]) -> dict[str, Any]:
			model_id = payload.get("modelId")
			upload = payload.get("pointCloud")
			device = payload.get("device", "auto")
			if not isinstance(model_id, str) or SAFE_FILE_NAME.search(model_id) or len(model_id) > 120:
			raise ApiError("Invalid trained model id.")
			model_record = next((item for item in pointcloud_semantic_models(self.root) if item["id"] == model_id), None)
			@@ -1822,7 +2759,9 @@
			if Path(name).suffix.lower() not in suffixes:
			raise ApiError("Model inference requires a PLY, PCD, XYZ, LAS, or LAZ point cloud.")
			validate_pointcloud_model_input(staged_path)
			execution = pointcloud_execution_environment(self.root)
			if not isinstance(device, str):
			raise ApiError("Point-cloud device must be auto, cpu, or cuda.")
			execution = pointcloud_execution_environment(self.root, device)
			run_id = make_run_id("semantic-inference")
			raw_root = self.root / "shared" / "data" / "raw" / "05-3d-pointcloud" / "model-inference-runs" / run_id
			processed_root = self.root / "shared" / "data" / "processed" / "05-3d-pointcloud" / "model-inference-runs" / run_id
			@@ -1845,12 +2784,149 @@
			except ValueError as exc:
			raise ApiError("Selected model is outside the allowed training output directory.") from exc
			job_id = uuid4().hex
			job = {"id": job_id, "runId": run_id, "modelId": model_id, "inputName": name, "status": "queued", "stage": "queued", "requestedDevice": "auto", "device": execution["device"], "environment": execution["environment"], "torchVersion": execution["torchVersion"], "createdAt": datetime.now(UTC).isoformat(), "sourceSha256": actual_sha256, "rawInput": relative_path(self.root, raw_path), "processedInput": relative_path(self.root, processed_path)}
			job = {"id": job_id, "runId": run_id, "modelId": model_id, "inputName": name, "status": "queued", "stage": "queued", "requestedDevice": execution["requestedDevice"], "device": execution["device"], "environment": execution["environment"], "torchVersion": execution["torchVersion"], "fallbackUsed": execution["fallbackUsed"], "fallbackReason": execution["fallbackReason"], "createdAt": datetime.now(UTC).isoformat(), "sourceSha256": actual_sha256, "rawInput": relative_path(self.root, raw_path), "processedInput": relative_path(self.root, processed_path)}
			with POINTCLOUD_INFERENCE_JOBS_LOCK:
			POINTCLOUD_INFERENCE_JOBS[job_id] = job
			thread = threading.Thread(target=execute_pointcloud_inference_job, args=(self.root, job_id, model_path, processed_path, output, execution), daemon=True, name=f"pointcloud-inference-{job_id[:8]}")
			thread.start()
			return dict(job)

			def create_pointcloud_auto_annotation_run(self, payload: dict[str, Any]) -> dict[str, Any]:
			model_id, source_id, device, confidence = payload.get("modelId"), payload.get("sourceId"), payload.get("device", "auto"), payload.get("candidateConfidence", 0.95)
			if not isinstance(model_id, str) or SAFE_FILE_NAME.search(model_id) or len(model_id) > 120:
			raise ApiError("Invalid trained model id.")
			if not isinstance(source_id, str):
			raise ApiError("Automatic annotation needs a selected local annotation source.")
			if not isinstance(device, str) or device not in COMPUTE_DEVICES:
			raise ApiError("Point-cloud device must be auto, cpu, or cuda.")
			if not isinstance(confidence, (int, float)) or isinstance(confidence, bool) or not 0.5 <= float(confidence) < 1.0:
			raise ApiError("Candidate confidence must be between 0.5 and 1.0.")
			model_record = next((item for item in pointcloud_semantic_models(self.root) if item["id"] == model_id), None)
			source = next((item for item in pointcloud_annotation_sources(self.root) if item["id"] == source_id), None)
			if not model_record or not source:
			raise ApiError("The selected model or annotation source is unavailable.")
			if source.get("sourceHasRgb") is False:
			raise ApiError("Automatic annotation needs observed RGB point features.")
			model_path = (self.root / str(model_record["model"])).resolve()
			source_path = (self.root / str(source["artifactRoot"]) / str(source["file"])).resolve()
			training_root = (self.root / "shared" / "outputs" / "05-3d-pointcloud" / "training-runs").resolve()
			output_root = (self.root / "shared" / "outputs" / "05-3d-pointcloud").resolve()
			try:
			model_path.relative_to(training_root)
			source_path.relative_to(output_root)
			except ValueError as exc:
			raise ApiError("Selected automatic-annotation inputs are outside local workbench outputs.") from exc
			if not model_path.is_file() or not source_path.is_file() or file_sha256(source_path) != source["sha256"]:
			raise ApiError("Selected automatic-annotation input no longer passes its verification contract.")
			execution = pointcloud_execution_environment(self.root, device)
			run_id = make_run_id("auto-annotation")
			output = output_root / "auto-annotation-runs" / run_id
			job_id = uuid4().hex
			job = {"id": job_id, "runId": run_id, "modelId": model_id, "sourceId": source_id, "inputName": str(source["file"]), "candidateConfidence": float(confidence), "status": "queued", "stage": "queued", "requestedDevice": execution["requestedDevice"], "device": execution["device"], "environment": execution["environment"], "torchVersion": execution["torchVersion"], "fallbackUsed": execution["fallbackUsed"], "fallbackReason": execution["fallbackReason"], "createdAt": datetime.now(UTC).isoformat(), "sourceSha256": source["sha256"]}
			with POINTCLOUD_INFERENCE_JOBS_LOCK:
			POINTCLOUD_INFERENCE_JOBS[job_id] = job
			thread = threading.Thread(target=execute_pointcloud_inference_job, args=(self.root, job_id, model_path, source_path, output, execution, source_id, float(confidence)), daemon=True, name=f"auto-annotation-{job_id[:8]}")
			thread.start()
			return dict(job)

			def save_pointcloud_auto_annotation_review(self, payload: dict[str, Any]) -> dict[str, Any]:
			run_id, source_id, corrections = payload.get("runId"), payload.get("sourceId"), payload.get("corrections")
			if not isinstance(run_id, str) or SAFE_FILE_NAME.search(run_id) or len(run_id) > 120:
			raise ApiError("Invalid automatic-annotation run id.")
			if not isinstance(source_id, str) or not isinstance(corrections, list):
			raise ApiError("Automatic candidate review needs a source id and corrections array.")
			source = pointcloud_annotation_source(self.root, source_id)
			output_root = (self.root / "shared" / "outputs" / "05-3d-pointcloud" / "auto-annotation-runs").resolve()
			artifact = (output_root / run_id).resolve()
			candidate_path = artifact / "automatic-annotation-candidates.json"
			try:
			artifact.relative_to(output_root)
			except ValueError as exc:
			raise ApiError("Automatic candidate review path is outside local workbench outputs.") from exc
			candidate = load_json(candidate_path)
			if candidate.get("schema_version") != 1 or candidate.get("source_id") != source_id or candidate.get("source_sha256") != source["sha256"]:
			raise ApiError("Automatic candidate provenance no longer matches the selected source.")
			active_codes = {item["code"] for item in annotation_classes(self.root)}
			compact: dict[int, int] = {}
			for item in corrections:
			if not isinstance(item, list) or len(item) != 2 or not all(isinstance(value, int) for value in item):
			raise ApiError("Each candidate correction must be [pointIndex, classCodeOrZero].")
			index, code = item
			if index < 0 or index >= int(source["pointCount"]) or (code != 0 and code not in active_codes):
			raise ApiError("Candidate review contains an out-of-range point index or unsupported class code.")
			compact[index] = code
			review = {
			"schema_version": 1,
			"run_id": run_id,
			"source_id": source_id,
			"source_sha256": source["sha256"],
			"candidate_confidence": candidate.get("candidate_confidence"),
			"corrections": [[index, code] for index, code in sorted(compact.items())],
			"saved_at": datetime.now(UTC).isoformat(),
			}
			(artifact / "review-corrections.json").write_text(json.dumps(review, ensure_ascii=False, indent=2), encoding="utf-8")
			return {"runId": run_id, "sourceId": source_id, "correctionCount": len(compact), "savedAt": review["saved_at"]}

			def accept_pointcloud_auto_annotation(self, payload: dict[str, Any]) -> dict[str, Any]:
			run_id, source_id, base_annotation_id = payload.get("runId"), payload.get("sourceId"), payload.get("baseAnnotationId")
			if not isinstance(run_id, str) or SAFE_FILE_NAME.search(run_id) or len(run_id) > 120:
			raise ApiError("Invalid automatic-annotation run id.")
			if not isinstance(source_id, str):
			raise ApiError("Automatic-annotation acceptance needs a source id.")
			source = pointcloud_annotation_source(self.root, source_id)
			output_root = (self.root / "shared" / "outputs" / "05-3d-pointcloud" / "auto-annotation-runs").resolve()
			candidate_path = (output_root / run_id / "automatic-annotation-candidates.json").resolve()
			try:
			candidate_path.relative_to(output_root)
			except ValueError as exc:
			raise ApiError("Automatic candidate path is outside local workbench outputs.") from exc
			candidate = load_json(candidate_path)
			if candidate.get("schema_version") != 1 or candidate.get("source_id") != source_id or candidate.get("source_sha256") != source["sha256"]:
			raise ApiError("Automatic candidate provenance no longer matches the selected source.")
			raw_labels = candidate.get("labels")
			if not isinstance(raw_labels, list):
			raise ApiError("Automatic candidate labels are unavailable.")
			merged: dict[int, int] = {}
			for item in raw_labels:
			if not isinstance(item, list) or len(item) != 3 or not isinstance(item[0], int) or not isinstance(item[1], int):
			raise ApiError("Automatic candidate labels are invalid.")
			merged[item[0]] = item[1]
			review_path = candidate_path.parent / "review-corrections.json"
			review_correction_count = 0
			if review_path.is_file():
			review = load_json(review_path)
			if review.get("schema_version") != 1 or review.get("run_id") != run_id or review.get("source_id") != source_id or review.get("source_sha256") != source["sha256"]:
			raise ApiError("Candidate review draft provenance no longer matches the selected source.")
			corrections = review.get("corrections")
			if not isinstance(corrections, list):
			raise ApiError("Candidate review corrections are invalid.")
			valid_codes = {item["code"] for item in annotation_classes(self.root)}
			for item in corrections:
			if not isinstance(item, list) or len(item) != 2 or not all(isinstance(value, int) for value in item):
			raise ApiError("Candidate review corrections are invalid.")
			index, code = item
			if index < 0 or index >= int(source["pointCount"]) or (code != 0 and code not in valid_codes):
			raise ApiError("Candidate review contains an out-of-range point index or unsupported class code.")
			if code == 0:
			merged.pop(index, None)
			else:
			merged[index] = code
			review_correction_count += 1
			if isinstance(base_annotation_id, str) and base_annotation_id:
			base_path = self.root / "shared" / "outputs" / "05-3d-pointcloud" / "annotations" / base_annotation_id / "annotation.json"
			base = load_json(base_path)
			if base.get("source_id") != source_id:
			raise ApiError("The base human annotation belongs to another source.")
			for item in base.get("labels", []):
			if isinstance(item, list) and len(item) == 2 and all(isinstance(value, int) for value in item):
			merged[item[0]] = item[1]
			annotation = self.create_pointcloud_annotation({"sourceId": source_id, "labels": [[index, code] for index, code in merged.items()]})
			annotation_path = self.root / str(annotation["path"])
			document = load_json(annotation_path)
			document["provenance"] = "user_confirmed_high_confidence_model_candidates_with_human_labels_preferred"
			document["automatic_annotation"] = {"run_id": run_id, "candidate_confidence": candidate.get("candidate_confidence"), "candidate_count": candidate.get("candidate_count"), "review_correction_count": review_correction_count, "base_annotation_id": base_annotation_id or None}
			annotation_path.write_text(json.dumps(document, ensure_ascii=False, indent=2), encoding="utf-8")
			return annotation

			def create_photo_reconstruction_run(self, payload: dict[str, Any]) -> dict[str, Any]:
			uploads = payload.get("photos")
			@@ -1891,7 +2967,9 @@
			for _, staged_path, _ in photos:
			shutil.rmtree(staged_path.parent)

			job = {"id": job_id, "runId": run_id, "status": "queued", "stage": "queued", "createdAt": datetime.now(UTC).isoformat(), "inputImages": len(photos), "usePositionPriors": use_position_priors}
			progress_path = processed_root / "photo_reconstruction_progress.json"
			progress_path.write_text(json.dumps({"percent": 0, "stage": "queued", "message": "照片已保存，正在等待 CPU 重建资源。", "inputImages": len(photos), "updatedAt": datetime.now(UTC).isoformat(), "estimate": True}, ensure_ascii=False), encoding="utf-8")
			job = {"id": job_id, "runId": run_id, "status": "queued", "stage": "queued", "createdAt": datetime.now(UTC).isoformat(), "inputImages": len(photos), "usePositionPriors": use_position_priors, "progressPath": str(progress_path)}
			with PHOTO_RECONSTRUCTION_JOBS_LOCK:
			PHOTO_RECONSTRUCTION_JOBS[job_id] = job
			thread = threading.Thread(
			@@ -1901,7 +2979,7 @@
			name=f"photo-reconstruction-{run_id}",
			)
			thread.start()
			return dict(job)
			return photo_reconstruction_job(job_id) or {key: value for key, value in job.items() if key != "progressPath"}

			def create_risk_rule_run(self, payload: dict[str, Any]) -> dict[str, Any]:
			files = payload.get("files")