"""Serve the local GeoAI Workbench console and its narrow local-run APIs."""

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
from pathlib import Path, PurePosixPath
from typing import Any
from urllib.parse import parse_qs, unquote, urlsplit
from uuid import uuid4


DEFAULT_HOST = "127.0.0.1"
DEFAULT_PORT = 6173
# Uploads are sent as Base64 JSON. Keep the request limit above two 1 GiB
# files after encoding while retaining a per-file bound for local experiments.
MAX_REQUEST_BYTES = 3072 * 1024 * 1024
MAX_FILE_BYTES = 1024 * 1024 * 1024
MAX_IMAGES_PER_RUN = 12
MAX_SEGMENTATION_IMAGES_PER_RUN = 6
MAX_MEASUREMENT_RASTERS_PER_RUN = 4
MAX_POINTCLOUDS_PER_RUN = 2
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
CHANGE_THRESHOLD_MIN = 0.01
CHANGE_THRESHOLD_MAX = 0.99
CHANGE_MAX_DIMENSION_DEFAULT = 1024
CHANGE_MAX_DIMENSION_AUTO = 0
CHANGE_MAX_DIMENSION_MIN = 512
CHANGE_MAX_DIMENSION_MAX = 4096
CHANGE_PROCESSING_MODE_DEFAULT = "auto"
CHANGE_PROCESSING_MODES = {"auto", "image", "geotiff"}
SCAN_DEFAULT_THRESHOLDS = [0.3, 0.4, 0.5]
SCAN_DEFAULT_AREAS = [64, 256, 686]
SCAN_MAX_THRESHOLDS = 6
SCAN_MAX_AREAS = 6
SCAN_MAX_COMBINATIONS = 24
SCAN_JOB_TIMEOUT = 1800
ALLOWED_PATH_PREFIXES = (
    "apps/workbench-console",
    "shared/outputs",
    "shared/data/raw/00-change-detection",
    "shared/data/raw/01-object-detection",
    "shared/data/raw/02-semantic-mapping",
    "shared/data/raw/09-anomaly-detection",
    "shared/data/raw/05-3d-pointcloud",
)
SAFE_FILE_NAME = re.compile(r"[^\w.-]+", re.UNICODE)
SAFE_UPLOAD_ID = re.compile(r"^[0-9a-f]{32}$")
SAFE_SCAN_ID = re.compile(r"^[A-Za-z0-9._-]{1,100}$")
SAFE_SCAN_RESULT_ID = re.compile(r"^threshold-\d+(?:\.\d+)?_area-\d+$")
RUN_LOCK = threading.Lock()
SCAN_JOBS: dict[str, dict[str, Any]] = {}
SCAN_JOBS_LOCK = threading.Lock()
ANOMALY_JOB_LOCK = threading.Lock()
ANOMALY_JOBS: dict[str, dict[str, Any]] = {}
PHOTO_RECONSTRUCTION_JOBS: dict[str, dict[str, Any]] = {}
PHOTO_RECONSTRUCTION_JOBS_LOCK = threading.Lock()
POINTCLOUD_TRAINING_JOBS: dict[str, dict[str, Any]] = {}
POINTCLOUD_TRAINING_JOBS_LOCK = threading.Lock()
POINTCLOUD_INFERENCE_JOBS: dict[str, dict[str, Any]] = {}
POINTCLOUD_INFERENCE_JOBS_LOCK = threading.Lock()
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):
    """A request error that can be shown to the local console user."""


def safe_file_name(value: str, expected_suffixes: set[str]) -> str:
    name = Path(value).name
    suffix = Path(name).suffix.lower()
    if suffix not in expected_suffixes:
        raise ApiError(f"Unsupported file type: {suffix or '(none)'}.")
    stem = SAFE_FILE_NAME.sub("_", Path(name).stem).strip("._") or "upload"
    return f"{stem[:80]}{suffix}"


def decode_upload(payload: dict[str, Any], expected_suffixes: set[str]) -> tuple[str, bytes]:
    if not isinstance(payload, dict) or not isinstance(payload.get("name"), str) or not isinstance(payload.get("content"), str):
        raise ApiError("Each uploaded file must include name and Base64 content.")
    name = safe_file_name(payload["name"], expected_suffixes)
    try:
        content = base64.b64decode(payload["content"], validate=True)
    except (binascii.Error, ValueError) as exc:
        raise ApiError(f"Invalid Base64 file content for {name}.") from exc
    if not content:
        raise ApiError(f"Uploaded file is empty: {name}.")
    if len(content) > MAX_FILE_BYTES:
        raise ApiError(f"Uploaded file exceeds {MAX_FILE_BYTES // (1024 * 1024)} MB: {name}.")
    return name, content


def make_run_id(prefix: str) -> str:
    return f"{prefix}-{datetime.now(UTC):%Y%m%d-%H%M%S}-{uuid4().hex[:6]}"


def relative_path(root: Path, path: Path) -> str:
    # 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]:
    try:
        payload = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError):
        return {}
    return payload if isinstance(payload, dict) else {}


def file_sha256(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as stream:
        for chunk in iter(lambda: stream.read(8 * 1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


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}

    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__}))"],
                cwd=root,
                capture_output=True,
                text=True,
                timeout=20,
                check=False,
            )
            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):
            probe_reason = "CUDA probe could not return a valid result."
    if requested_device == "cuda":
        raise ApiError(f"CUDA was requested for {label}, but it is unavailable: {probe_reason}")
    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": "auto", "fallbackUsed": True, "fallbackReason": probe_reason}


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 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]]:
    output_root = root / "shared" / "outputs" / "15-trajectory-analysis"
    records: list[dict[str, Any]] = []
    for metadata_path in output_root.rglob("run_metadata.json"):
        artifact = metadata_path.parent
        if not (artifact / "trajectory_summary.csv").is_file() or not (artifact / "events.json").is_file():
            continue
        metadata = load_json(metadata_path)
        case_id = str(metadata.get("case_id") or artifact.name)
        is_real = case_id == "tian-dun-flight-19578"
        records.append(
            {
                "id": case_id,
                "label": "田墩实飞" if is_real else case_id,
                "note": "区域相交是来源数据的空间结果，不是违规结论。" if is_real else "本地实验运行结果，可继续查看结构化输出。",
                "artifactRoot": relative_path(root, artifact),
                "showSpatialContext": (artifact / "zones.geojson").is_file() and (artifact / "reference_routes.geojson").is_file(),
                "showFlyableZones": (artifact / "flyable_zones.geojson").is_file(),
                "createdAt": str(metadata.get("created_at") or ""),
            }
        )
    return sorted(records, key=lambda item: (item["createdAt"], item["id"]), reverse=True)


def detection_runs(root: Path) -> list[dict[str, Any]]:
    output_root = root / "shared" / "outputs" / "01-object-detection"
    records: list[dict[str, Any]] = []
    for metadata_path in output_root.rglob("run_metadata.json"):
        artifact = metadata_path.parent
        if not (artifact / "detections.json").is_file():
            continue
        metadata = load_json(metadata_path)
        input_value = str(metadata.get("input_dir") or "")
        input_dir = Path(input_value) if input_value else root / "shared" / "data" / "raw" / "01-object-detection"
        try:
            input_root = relative_path(root, input_dir.resolve())
        except ValueError:
            continue
        run_id = artifact.name if artifact != output_root else "baseline"
        device = str(metadata.get("device") or "cpu").lower()
        execution = "GPU" if device.startswith("cuda") else "CPU"
        records.append(
            {
                "id": run_id,
                "label": "既有基线结果" if run_id == "baseline" else run_id,
                "note": f"{execution} 基线：人员与常见车辆；树木不在当前模型有效类别内。",
                "artifactRoot": relative_path(root, artifact),
                "inputRoot": input_root,
                "createdAt": str(metadata.get("created_at") or ""),
            }
        )
    return sorted(records, key=lambda item: (item["createdAt"], item["id"]), reverse=True)


def change_runs(root: Path) -> list[dict[str, Any]]:
    output_root = root / "shared" / "outputs" / "00-change-detection"
    records: list[dict[str, Any]] = []
    for metadata_path in output_root.rglob("run_metadata.json"):
        artifact = metadata_path.parent
        metadata = load_json(metadata_path)
        artifacts = metadata.get("artifacts")
        if metadata.get("capability") != "00-change-detection" or metadata.get("schema_version") != 1 or metadata.get("kind") == "parameter-scan-inference" or not isinstance(artifacts, dict):
            continue
        if not (artifact / str(artifacts.get("overlay") or "")).is_file() or not (artifact / str(artifacts.get("vector") or "")).is_file():
            continue
        raw_root_value = str(metadata.get("raw_input_dir") or "shared/data/raw/00-change-detection/validation-20260817")
        raw_root = root / Path(raw_root_value)
        input_files = metadata.get("input_files")
        if not isinstance(input_files, list) or len(input_files) != 2:
            continue
        before_value = str(metadata.get("raw_before") or (Path(raw_root_value) / str(input_files[0])).as_posix())
        after_value = str(metadata.get("raw_after") or (Path(raw_root_value) / str(input_files[1])).as_posix())
        try:
            before_path = (root / before_value).resolve()
            after_path = (root / after_value).resolve()
            allowed_raw = (root / "shared" / "data" / "raw" / "00-change-detection").resolve()
            before_path.relative_to(allowed_raw)
            after_path.relative_to(allowed_raw)
        except ValueError:
            continue
        if not before_path.is_file() or not after_path.is_file():
            continue
        registered_before_name = str(artifacts.get("before_processed_preview") or "")
        registered_after_name = str(artifacts.get("after_registered_preview") or "")
        registered_before_path = artifact / registered_before_name if registered_before_name else None
        registered_after_path = artifact / registered_after_name if registered_after_name else None
        run_id = artifact.name
        device = str(metadata.get("device") or "cpu").lower()
        execution = "GPU" if device.startswith("cuda") else "CPU"
        record = {
                "id": run_id,
                "label": run_id,
                "note": f"ChangeStar {execution} 变化栅格与 GeoAI 像素坐标图斑；结果需人工复核。",
                "artifactRoot": relative_path(root, artifact),
                "beforeImage": relative_path(root, before_path),
                "afterImage": relative_path(root, after_path),
                "createdAt": str(metadata.get("created_at") or ""),
            }
        if registered_before_path and registered_after_path and registered_before_path.is_file() and registered_after_path.is_file():
            record["registeredBeforeImage"] = relative_path(root, registered_before_path)
            record["registeredAfterImage"] = relative_path(root, registered_after_path)
            # Existing console consumers use afterImage as the vector-overlay
            # base. Point it at the registered grid so polygons and highlights
            # share the same pixel coordinates as the model output.
            record["rawAfterImage"] = record["afterImage"]
            record["afterImage"] = record["registeredAfterImage"]
        records.append(record)
    return sorted(records, key=lambda item: (item["createdAt"], item["id"]), reverse=True)


def change_parameter_scans(root: Path) -> list[dict[str, Any]]:
    """Discover read-only parameter scans produced from an existing change run."""
    output_root = root / "shared" / "outputs" / "00-change-detection"
    records: list[dict[str, Any]] = []
    for summary_path in output_root.rglob("scan_summary.json"):
        scan_root = summary_path.parent
        summary = load_json(summary_path)
        results: list[dict[str, Any]] = []
        for item in summary.get("results", []):
            if not isinstance(item, dict) or not isinstance(item.get("directory"), str):
                continue
            directory = scan_root / item["directory"]
            overlay = directory / "overlay_preview.jpg"
            mask = directory / "change_mask.tif"
            regions = directory / "regions.json"
            if not overlay.is_file() or not mask.is_file() or not regions.is_file():
                continue
            results.append(
                {
                    "id": item["directory"],
                    "label": f"T={float(item.get('threshold', 0.5)):.2f} / 面积={int(item.get('minimum_area_pixels', 0))} px",
                    "threshold": item.get("threshold"),
                    "minimumAreaPixels": item.get("minimum_area_pixels"),
                    "cleanedComponents": item.get("cleaned_components"),
                    "changedPixels": item.get("changed_pixels"),
                    "changedPixelRatio": item.get("changed_pixel_ratio"),
                    "vectorFeatureCount": item.get("vector_feature_count"),
                    "fullVectorFeatureCount": (load_json(directory / "full_result.json").get("full_vector_feature_count") if (directory / "full_result.json").is_file() else None),
                    "rectangleFeatureCount": (
                        load_json(directory / "full_result.json").get("rectangle_vector_feature_count")
                        if (directory / "full_result.json").is_file() and load_json(directory / "full_result.json").get("rectangle_vector_feature_count") is not None
                        else len(load_json(directory / "changes_rectangles.geojson").get("features", [])) if (directory / "changes_rectangles.geojson").is_file() else None
                    ),
                    "overlay": relative_path(root, overlay),
                    "mask": relative_path(root, mask),
                    "regions": relative_path(root, regions),
                    "vector": (relative_path(root, directory / "changes.geojson") if (directory / "changes.geojson").is_file() else None),
                    "rectangleVector": (relative_path(root, directory / "changes_rectangles.geojson") if (directory / "changes_rectangles.geojson").is_file() else None),
                    "rectangleVectorWgs84": (relative_path(root, directory / "changes_rectangles_wgs84.geojson") if (directory / "changes_rectangles_wgs84.geojson").is_file() else None),
                }
            )
        # A failed job may have been repaired or materialized later. Keep it
        # discoverable whenever at least one complete candidate exists; only
        # hide scans that still have no usable result.
        if not results:
            continue
        scan_metadata = load_json(scan_root / "scan_metadata.json")
        contact_sheet = scan_root / "parameter_scan_contact_sheet.jpg"
        records.append(
            {
                "id": scan_root.name,
                "label": f"低成本参数扫描 · {scan_root.name}",
                "note": f"复用已有变化概率结果，不重新运行 ChangeStar；源运行：{summary.get('source_run', '未知')}",
                "artifactRoot": relative_path(root, scan_root),
                "sourceRun": summary.get("source_run"),
                "userSubmitted": bool(scan_metadata),
                "contactSheet": relative_path(root, contact_sheet) if contact_sheet.is_file() else None,
                "results": results,
            }
        )
    return sorted(records, key=lambda item: item["id"], reverse=True)


def semantic_runs(root: Path) -> list[dict[str, Any]]:
    output_root = root / "shared" / "outputs" / "02-semantic-mapping"
    records: list[dict[str, Any]] = []
    for metadata_path in output_root.rglob("run_metadata.json"):
        artifact = metadata_path.parent
        metadata = load_json(metadata_path)
        if metadata.get("capability") != "02-semantic-mapping" or not isinstance(metadata.get("images"), list):
            continue
        run_id = artifact.name if artifact != output_root else "baseline"
        records.append(
            {
                "id": run_id,
                "label": "语义分割基线" if run_id == "baseline" else run_id,
                "note": f"{metadata.get('task_name') or '通用颜色规则基线'}，输出栅格掩膜与 GeoAI 矢量结果。",
                "artifactRoot": relative_path(root, artifact),
                "inputRoot": str(metadata.get("input_dir") or "shared/data/processed/02-semantic-mapping"),
                "rawInputRoot": str(metadata.get("raw_input_dir") or "shared/data/raw/02-semantic-mapping"),
                "createdAt": str(metadata.get("created_at") or ""),
            }
        )
    return sorted(records, key=lambda item: (item["createdAt"], item["id"]), reverse=True)


def measurement_runs(root: Path) -> list[dict[str, Any]]:
    output_root = root / "shared" / "outputs" / "04-spatial-measurement"
    records: list[dict[str, Any]] = []
    for metadata_path in output_root.rglob("run_metadata.json"):
        artifact = metadata_path.parent
        metadata = load_json(metadata_path)
        if metadata.get("capability") != "04-spatial-measurement" or not isinstance(metadata.get("images"), list):
            continue
        run_id = artifact.name
        records.append(
            {
                "id": run_id,
                "label": run_id,
                "note": "GeoAI 栅格转矢量后进行对象计数、面积和周长测量。",
                "artifactRoot": relative_path(root, artifact),
                "createdAt": str(metadata.get("created_at") or ""),
            }
        )
    return sorted(records, key=lambda item: (item["createdAt"], item["id"]), reverse=True)


def pointcloud_runs(root: Path) -> list[dict[str, Any]]:
    output_root = root / "shared" / "outputs" / "05-3d-pointcloud"
    records: list[dict[str, Any]] = []
    for metadata_path in output_root.rglob("run_metadata.json"):
        artifact = metadata_path.parent
        metadata = load_json(metadata_path)
        dense_photo_reconstruction = metadata.get("dense_photo_reconstruction")
        if metadata.get("capability") == "05-3d-pointcloud" and isinstance(dense_photo_reconstruction, dict):
            textured_model = dense_photo_reconstruction.get("textured_model_file")
            dense_point_cloud = dense_photo_reconstruction.get("dense_point_cloud_file")
            mesh = dense_photo_reconstruction.get("mesh_file")
            if all(isinstance(item, str) and (artifact / item).is_file() for item in (textured_model, dense_point_cloud, mesh)):
                run_id = artifact.name
                records.append(
                    {
                        "id": run_id,
                        "label": str(metadata.get("display_name") or run_id),
                        "note": "CPU 稠密 MVS：显示经过深度融合、网格化和纹理化的局部模型；不是测绘级坐标、DSM、正射图或语义识别结论。",
                        "artifactRoot": relative_path(root, artifact),
                        "createdAt": str(metadata.get("created_at") or ""),
                    }
                )
                continue
        photo_reconstruction = metadata.get("photo_reconstruction")
        if metadata.get("capability") == "05-3d-pointcloud" and isinstance(photo_reconstruction, dict):
            preview = photo_reconstruction.get("preview_file")
            point_cloud = photo_reconstruction.get("point_cloud_file")
            if isinstance(preview, str) and isinstance(point_cloud, str) and (artifact / preview).is_file() and (artifact / point_cloud).is_file():
                run_id = artifact.name
                records.append(
                    {
                        "id": run_id,
                        "label": str(metadata.get("display_name") or run_id),
                        "note": "CPU 稀疏 SfM：显示可复核点云、相机位姿与误差；不是稠密重建、DSM、语义识别或测绘精度结论。",
                        "artifactRoot": relative_path(root, artifact),
                        "createdAt": str(metadata.get("created_at") or ""),
                    }
                )
                continue
        point_clouds = metadata.get("point_clouds")
        if metadata.get("capability") != "05-3d-pointcloud" or not isinstance(point_clouds, list) or not point_clouds:
            continue
        if any(not isinstance(item, dict) or not (artifact / str(item.get("preview_file") or "")).is_file() or not (artifact / str(item.get("vector_file") or "")).is_file() for item in point_clouds):
            continue
        run_id = artifact.name
        records.append(
            {
                "id": run_id,
                "label": str(metadata.get("display_name") or run_id),
                "note": "CPU 语义规则基线：地面、植被、构筑物以及电线/杆塔候选，需要人工复核；不提供测绘精度或资产台账结论。" if any(isinstance(item, dict) and item.get("semantic_summary_file") for item in point_clouds) else "CPU 几何基线：地面/高出地物分离、DSM、近似网格和 GeoAI 足迹；不提供语义类别或测绘精度结论。",
                "artifactRoot": relative_path(root, artifact),
                "createdAt": str(metadata.get("created_at") or ""),
            }
        )
    return sorted(records, key=lambda item: (item["createdAt"], item["id"]), reverse=True)


def pointcloud_annotation_sources(root: Path) -> list[dict[str, Any]]:
    """Expose only generated, fixed preview PLYs suitable for manual labels."""
    sources: list[dict[str, Any]] = []
    for case in pointcloud_runs(root):
        artifact = root / str(case["artifactRoot"])
        metadata = load_json(artifact / "run_metadata.json")
        for cloud in metadata.get("point_clouds", []):
            if not isinstance(cloud, dict):
                continue
            name = cloud.get("semantic_annotation_source_point_cloud")
            if not isinstance(name, str) or Path(name).name != name:
                continue
            path = artifact / name
            if not path.is_file() or path.suffix.lower() != ".ply":
                continue
            sources.append({
                "id": f"{case['id']}:{name}", "runId": case["id"], "label": f"{case['label']} / {cloud.get('file', name)}",
                "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]]:
    output = root / "shared" / "outputs" / "05-3d-pointcloud" / "annotations"
    records: list[dict[str, Any]] = []
    for path in output.glob("*/annotation.json"):
        data = load_json(path)
        if data.get("schema_version") != 1 or not isinstance(data.get("id"), str):
            continue
        records.append({"id": data["id"], "sourceId": data.get("source_id"), "createdAt": data.get("created_at"), "labelCount": len(data.get("labels", [])), "classCounts": data.get("class_counts", {}), "path": relative_path(root, path)})
    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], 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()})
    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)
        if completed.returncode:
            message = (completed.stderr or completed.stdout or "Unknown training error.").strip().splitlines()[-1]
            raise ApiError(message[:600])
        metrics = output / "metrics.json"
        model = output / "model.pt"
        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), "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]})


def pointcloud_semantic_models(root: Path) -> list[dict[str, Any]]:
    """Expose only complete locally trained models, never arbitrary model paths."""
    output_root = root / "shared" / "outputs" / "05-3d-pointcloud" / "training-runs"
    records: list[dict[str, Any]] = []
    for model_path in output_root.glob("*/model.pt"):
        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 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:
            continue
        test = metrics.get("test") if isinstance(metrics.get("test"), dict) else {}
        report = test.get("report") if isinstance(test.get("report"), dict) else {}
        summary: dict[str, float] = {}
        for code in class_codes:
            definition = classes.get(code)
            key = definition.get("key") if isinstance(definition, dict) else None
            score = report.get(key) if isinstance(key, str) else None
            if isinstance(score, dict) and isinstance(score.get("f1-score"), (int, float)):
                summary[key] = round(float(score["f1-score"]), 3)
        records.append({"id": model_path.parent.name, "label": model_path.parent.name, "artifactRoot": relative_path(root, model_path.parent), "model": relative_path(root, model_path), "metrics": relative_path(root, metrics_path), "createdAt": str(metrics.get("created_at") or ""), "classes": classes, "testF1": summary})
    return sorted(records, key=lambda item: (item["createdAt"], item["id"]), reverse=True)


def pointcloud_inference_job(job_id: str) -> dict[str, Any] | None:
    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:
    """Reject obviously incomplete uploads before consuming a CPU inference job."""
    size = path.stat().st_size
    if size < 1:
        raise ApiError("点云文件为空。请选择包含 RGB 点位的完整点云导出文件。")
    if path.suffix.lower() not in {".las", ".laz"}:
        return
    if size < 227:
        raise ApiError("LAS/LAZ 文件不完整（小于有效 LAS 文件头）。请选择完整点云文件，不要上传空白或未完成下载的分块。")
    with path.open("rb") as stream:
        header = stream.read(375)
    if len(header) < 227 or header[:4] != b"LASF":
        raise ApiError("LAS/LAZ 文件头无效。请选择完整的 LAS/LAZ 点云导出文件。")
    header_size = int.from_bytes(header[94:96], "little")
    if header_size < 227 or header_size > size:
        raise ApiError("LAS/LAZ 文件头不完整。请选择完整的 LAS/LAZ 点云导出文件。")
    version_minor = header[25]
    point_count_offset, point_count_size = (247, 8) if version_minor >= 4 else (107, 4)
    if len(header) < point_count_offset + point_count_size:
        raise ApiError("LAS/LAZ 文件头不完整。请选择完整的 LAS/LAZ 点云导出文件。")
    point_count = int.from_bytes(header[point_count_offset:point_count_offset + point_count_size], "little")
    if point_count < 1:
        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], 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)
        if completed.returncode:
            message = (completed.stderr or completed.stdout or "Unknown model inference error.").strip().splitlines()[-1]
            raise ApiError(message[:600])
        metadata = output / "run_metadata.json"
        preview = output / "predicted-semantic-preview.ply"
        classified_las = output / "predicted-semantic-classified.las"
        counts = output / "class-counts.csv"
        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), "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]})


def risk_rule_runs(root: Path) -> list[dict[str, Any]]:
    output_root = root / "shared" / "outputs" / "07-risk-rule-engine"
    records: list[dict[str, Any]] = []
    for metadata_path in output_root.rglob("run_metadata.json"):
        artifact = metadata_path.parent
        metadata = load_json(metadata_path)
        artifacts = metadata.get("artifacts")
        if metadata.get("capability") != "07-risk-rule-engine" or not isinstance(artifacts, dict):
            continue
        required = ("risk_raster", "risk_preview", "risk_vector", "risk_scores_csv", "summary")
        if any(not isinstance(artifacts.get(key), str) or not (artifact / artifacts[key]).is_file() for key in required):
            continue
        run_id = artifact.name
        records.append(
            {
                "id": run_id,
                "label": str(metadata.get("display_name") or run_id),
                "note": "可审计空间规则评分，仅供人工复核，不构成事件或处置结论。",
                "artifactRoot": relative_path(root, artifact),
                "createdAt": str(metadata.get("created_at") or ""),
            }
        )
    return sorted(records, key=lambda item: (item["createdAt"], item["id"]), reverse=True)


def anomaly_runs(root: Path) -> list[dict[str, Any]]:
    output_root = root / "shared" / "outputs" / "09-anomaly-detection"
    allowed_raw = (root / "shared" / "data" / "raw" / "09-anomaly-detection").resolve()
    records: list[dict[str, Any]] = []
    for metadata_path in output_root.rglob("run_metadata.json"):
        artifact = metadata_path.parent
        metadata = load_json(metadata_path)
        images = metadata.get("images")
        if metadata.get("capability") != "09-anomaly-detection" or not isinstance(images, list):
            continue
        raw_input_value = str(metadata.get("raw_input_dir") or "")
        raw_reference_value = str(metadata.get("raw_reference_dir") or "")
        if not raw_input_value or not raw_reference_value:
            continue
        try:
            raw_input = (root / raw_input_value).resolve()
            raw_reference = (root / raw_reference_value).resolve()
            raw_input.relative_to(allowed_raw)
            raw_reference.relative_to(allowed_raw)
        except ValueError:
            continue
        if not raw_input.is_dir() or not raw_reference.is_dir():
            continue
        if any(not (artifact / str(item.get("overlay_file") or "")).is_file() for item in images if isinstance(item, dict)):
            continue
        run_id = artifact.name
        records.append(
            {
                "id": run_id,
                "label": str(metadata.get("display_name") or run_id),
                "note": str(metadata.get("case_note") or "规则基线与 Isolation Forest 的视觉离群候选，只供人工复核。"),
                "artifactRoot": relative_path(root, artifact),
                "inputRoot": relative_path(root, raw_input),
                "referenceRoot": relative_path(root, raw_reference),
                "createdAt": str(metadata.get("created_at") or ""),
            }
        )
    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)
        return dict(value) if value else None


def photo_reconstruction_job(job_id: str) -> dict[str, Any] | None:
    with PHOTO_RECONSTRUCTION_JOBS_LOCK:
        value = PHOTO_RECONSTRUCTION_JOBS.get(job_id)
        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:
    completed = subprocess.run(command, cwd=root, capture_output=True, text=True, timeout=timeout, check=False)
    if completed.returncode:
        message = (completed.stderr or completed.stdout or "Unknown script error.").strip().splitlines()[-1]
        raise ApiError(f"Processing failed: {message[:600]}")


def execute_photo_reconstruction_job(
    root: Path,
    job_id: str,
    run_id: str,
    raw_root: Path,
    processed_root: Path,
    sparse_output: Path,
    output: Path,
    source_sha256: dict[str, str],
    source_bytes: dict[str, int],
    use_position_priors: bool,
) -> None:
    python = root / ".venvs" / "05-3d-pointcloud" / "Scripts" / "python.exe"
    sparse_command = [
        str(python), str(root / "capabilities" / "05-3d-pointcloud" / "run_photo_reconstruction.py"),
        "--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"])
    else:
        sparse_command.extend(["--matching-mode", "exhaustive"])
    dense_command = [
        str(python), str(root / "capabilities" / "05-3d-pointcloud" / "run_cpu_dense_reconstruction.py"),
        "--input", str(processed_root), "--sparse-model", str(sparse_output / "sparse_model" / "0"),
        "--output", str(output),
        "--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:
            PHOTO_RECONSTRUCTION_JOBS[job_id].update({"status": "running", "stage": "sparse_sfm", "startedAt": datetime.now(UTC).isoformat()})
        with RUN_LOCK:
            run_background_command(sparse_command, root, 1800)
            sparse_metadata = load_json(sparse_output / "run_metadata.json")
            if not (sparse_output / "sparse_model" / "0").is_dir():
                raise ApiError("Sparse photo reconstruction finished without the expected COLMAP model.")
            with PHOTO_RECONSTRUCTION_JOBS_LOCK:
                PHOTO_RECONSTRUCTION_JOBS[job_id].update({"stage": "dense_mvs"})
            run_background_command(dense_command, root, PHOTO_RECONSTRUCTION_TIMEOUT)
        metadata_path = output / "run_metadata.json"
        if not metadata_path.is_file():
            raise ApiError("CPU dense reconstruction finished without the expected result metadata.")
        metadata = load_json(metadata_path)
        metadata["photo_reconstruction"] = sparse_metadata.get("photo_reconstruction", {})
        metadata["input_dir"] = relative_path(root, processed_root)
        metadata["raw_input_dir"] = relative_path(root, raw_root)
        metadata["source_sha256"] = source_sha256
        metadata["source_bytes"] = source_bytes
        metadata["console_photo_reconstruction"] = {"use_position_priors": use_position_priors, "matching_mode": "spatial" if use_position_priors else "exhaustive"}
        metadata["display_name"] = f"用户照片 CPU 稠密重建（{len(source_sha256)} 图）"
        metadata["case_note"] = "用户上传的同架次 JPG/JPEG 照片经 CPU SfM/MVS 重建；需要人工检查几何与纹理质量，不是测绘级成果。"
        metadata_path.write_text(json.dumps(metadata, ensure_ascii=False, indent=2), encoding="utf-8")
        definition = next(item for item in pointcloud_runs(root) if item["id"] == run_id)
        with PHOTO_RECONSTRUCTION_JOBS_LOCK:
            PHOTO_RECONSTRUCTION_JOBS[job_id].update({"status": "complete", "stage": "complete", "run": definition, "finishedAt": datetime.now(UTC).isoformat()})
    except Exception as exc:  # pragma: no cover - background boundary
        with PHOTO_RECONSTRUCTION_JOBS_LOCK:
            PHOTO_RECONSTRUCTION_JOBS[job_id].update({"status": "failed", "stage": "failed", "error": str(exc), "finishedAt": datetime.now(UTC).isoformat()})


def validate_anomaly_parameters(payload: dict[str, Any]) -> tuple[int, int, float, int]:
    tile_size = payload.get("tileSize", 256)
    stride = payload.get("stride", 128)
    threshold_quantile = payload.get("thresholdQuantile", 0.995)
    random_state = payload.get("randomState", 42)
    if isinstance(tile_size, bool) or not isinstance(tile_size, int) or not 128 <= tile_size <= 1024:
        raise ApiError("Tile size must be an integer between 128 and 1024.")
    if isinstance(stride, bool) or not isinstance(stride, int) or not 32 <= stride <= tile_size:
        raise ApiError("Stride must be an integer between 32 and tile size.")
    if isinstance(threshold_quantile, bool) or not isinstance(threshold_quantile, (int, float)) or not 0.9 <= float(threshold_quantile) <= 0.9999:
        raise ApiError("Threshold quantile must be between 0.9 and 0.9999.")
    if isinstance(random_state, bool) or not isinstance(random_state, int) or not 0 <= random_state <= 2_147_483_647:
        raise ApiError("Random state must be a non-negative integer.")
    return tile_size, stride, float(threshold_quantile), random_state


def execute_anomaly_job(
    root: Path,
    job_id: str,
    run_id: str,
    raw_reference: Path,
    raw_input: Path,
    processed_reference: Path,
    processed_input: Path,
    output: Path,
    tile_size: int,
    stride: int,
    threshold_quantile: float,
    random_state: int,
) -> None:
    with ANOMALY_JOB_LOCK:
        ANOMALY_JOBS[job_id]["status"] = "running"
    python = root / ".venvs" / "09-anomaly-detection" / "Scripts" / "python.exe"
    command = [
        str(python),
        str(root / "capabilities" / "09-anomaly-detection" / "run_anomaly_detection.py"),
        "--reference", str(processed_reference),
        "--input", str(processed_input),
        "--output", str(output),
        "--tile-size", str(tile_size),
        "--stride", str(stride),
        "--threshold-quantile", f"{threshold_quantile:.6f}",
        "--random-state", str(random_state),
        "--spatial-mode", "auto",
    ]
    try:
        with RUN_LOCK:
            completed = subprocess.run(command, cwd=root, capture_output=True, text=True, timeout=1800, check=False)
        if completed.returncode:
            message = (completed.stderr or completed.stdout or "Unknown script error.").strip().splitlines()[-1]
            raise ApiError(f"Processing failed: {message[:600]}")
        metadata_path = output / "run_metadata.json"
        if not metadata_path.is_file():
            raise ApiError("Anomaly-detection script finished without the expected result metadata.")
        metadata = load_json(metadata_path)
        metadata["raw_input_dir"] = relative_path(root, raw_input)
        metadata["raw_reference_dir"] = relative_path(root, raw_reference)
        metadata["processed_input_dir"] = relative_path(root, processed_input)
        metadata["processed_reference_dir"] = relative_path(root, processed_reference)
        metadata_path.write_text(json.dumps(metadata, ensure_ascii=False, indent=2), encoding="utf-8")
        definition = next(item for item in anomaly_runs(root) if item["id"] == run_id)
        with ANOMALY_JOB_LOCK:
            ANOMALY_JOBS[job_id].update({"status": "complete", "run": definition, "finishedAt": datetime.now(UTC).isoformat()})
    except Exception as exc:  # pragma: no cover - background boundary
        with ANOMALY_JOB_LOCK:
            ANOMALY_JOBS[job_id].update({"status": "failed", "error": str(exc), "finishedAt": datetime.now(UTC).isoformat()})


def semantic_tasks(root: Path) -> list[dict[str, Any]]:
    catalog = load_json(root / "capabilities" / "02-semantic-mapping" / "configs" / "task-catalog.json")
    tasks = catalog.get("tasks")
    if not isinstance(tasks, list):
        return []
    return [item for item in tasks if isinstance(item, dict) and isinstance(item.get("id"), str)]


class WorkbenchConsoleHandler(SimpleHTTPRequestHandler):
    """Static UI plus fixed, local-only ingestion and experiment commands."""

    server_version = "GeoAIWorkbench/1.0"

    @property
    def root(self) -> Path:
        return Path(self.directory).resolve()

    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
        if path == "/api/change-detection/scans":
            self.send_json(HTTPStatus.OK, {"scans": change_parameter_scans(self.root)})
            return
        if path.startswith("/api/change-detection/scan-jobs/"):
            job_id = path.rstrip("/").rsplit("/", 1)[-1]
            with SCAN_JOBS_LOCK:
                job = dict(SCAN_JOBS.get(job_id, {}))
            if not job:
                self.send_json(HTTPStatus.NOT_FOUND, {"error": "Unknown change-detection scan job."})
            else:
                self.send_json(HTTPStatus.OK, {"job": job})
            return
        if path == "/api/trajectory/runs":
            self.send_json(HTTPStatus.OK, {"runs": trajectory_runs(self.root)})
            return
        if path == "/api/object-detection/runs":
            self.send_json(HTTPStatus.OK, {"runs": detection_runs(self.root)})
            return
        if path == "/api/semantic-mapping/runs":
            self.send_json(HTTPStatus.OK, {"runs": semantic_runs(self.root)})
            return
        if path == "/api/semantic-mapping/tasks":
            self.send_json(HTTPStatus.OK, {"tasks": semantic_tasks(self.root)})
            return
        if path == "/api/spatial-measurement/runs":
            self.send_json(HTTPStatus.OK, {"runs": measurement_runs(self.root)})
            return
        if path == "/api/3d-pointcloud/runs":
            self.send_json(HTTPStatus.OK, {"runs": pointcloud_runs(self.root)})
            return
        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]
            job = pointcloud_inference_job(job_id)
            self.send_json(HTTPStatus.OK if job else HTTPStatus.NOT_FOUND, {"job": job} if job else {"error": "Unknown point-cloud model inference job."})
            return
        if path.startswith("/api/3d-pointcloud/training-jobs/"):
            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]
            job = photo_reconstruction_job(job_id)
            self.send_json(HTTPStatus.OK if job else HTTPStatus.NOT_FOUND, {"job": job} if job else {"error": "Unknown photo-reconstruction job."})
            return
        if path == "/api/risk-rule-engine/runs":
            self.send_json(HTTPStatus.OK, {"runs": risk_rule_runs(self.root)})
            return
        if path == "/api/anomaly-detection/runs":
            self.send_json(HTTPStatus.OK, {"runs": anomaly_runs(self.root)})
            return
        if path.startswith("/api/anomaly-detection/jobs/"):
            job_id = path.rstrip("/").rsplit("/", 1)[-1]
            job = anomaly_job(job_id)
            self.send_json(HTTPStatus.OK if job else HTTPStatus.NOT_FOUND, {"job": job} if job else {"error": "Unknown anomaly-detection job."})
            return
        if path == "/":
            self.send_response(HTTPStatus.FOUND)
            self.send_header("Location", "/apps/workbench-console/")
            self.end_headers()
            return
        super().do_GET()

    def do_POST(self) -> None:  # noqa: N802 - inherited standard-library method name
        path = urlsplit(self.path).path
        try:
            payload = self.read_json_body()
            if path == "/api/change-detection/runs":
                self.send_json(HTTPStatus.CREATED, {"run": self.create_change_run(payload)})
                return
            if path == "/api/change-detection/scans":
                self.send_json(HTTPStatus.ACCEPTED, {"job": self.create_change_scan(payload)})
                return
            if path.startswith("/api/change-detection/scans/") and path.endswith("/promote"):
                scan_id = path.split("/")[-2]
                self.send_json(HTTPStatus.CREATED, {"run": self.promote_change_scan(scan_id, payload)})
                return
            if path == "/api/trajectory/runs":
                self.send_json(HTTPStatus.CREATED, {"run": self.create_trajectory_run(payload)})
                return
            if path == "/api/object-detection/runs":
                self.send_json(HTTPStatus.CREATED, {"run": self.create_detection_run(payload)})
                return
            if path == "/api/semantic-mapping/runs":
                self.send_json(HTTPStatus.CREATED, {"run": self.create_semantic_run(payload)})
                return
            if path == "/api/spatial-measurement/runs":
                self.send_json(HTTPStatus.CREATED, {"run": self.create_measurement_run(payload)})
                return
            if path == "/api/3d-pointcloud/runs":
                self.send_json(HTTPStatus.CREATED, {"run": self.create_pointcloud_run(payload)})
                return
            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)})
                return
            if path == "/api/risk-rule-engine/runs":
                self.send_json(HTTPStatus.CREATED, {"run": self.create_risk_rule_run(payload)})
                return
            if path == "/api/anomaly-detection/runs":
                self.send_json(HTTPStatus.ACCEPTED, {"job": self.create_anomaly_run(payload)})
                return
            self.send_json(HTTPStatus.NOT_FOUND, {"error": "Unknown local API endpoint."})
        except ApiError as exc:
            self.send_json(HTTPStatus.BAD_REQUEST, {"error": str(exc)})
        except subprocess.TimeoutExpired:
            self.send_json(HTTPStatus.GATEWAY_TIMEOUT, {"error": "The local run exceeded its time limit; no existing result was overwritten."})
        except Exception as exc:  # pragma: no cover - defensive server boundary
            self.log_error("local run failed: %s", exc)
            self.send_json(HTTPStatus.INTERNAL_SERVER_ERROR, {"error": "Local run failed. Check the console terminal for details."})

    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."})
            return
        try:
            annotation_id = path[len(prefix):]
            if not annotation_id or "/" in annotation_id or SAFE_FILE_NAME.search(annotation_id) or len(annotation_id) > 120:
                raise ApiError("Invalid annotation id.")
            record = next((item for item in pointcloud_annotations(self.root) if item["id"] == annotation_id), None)
            if not record:
                raise ApiError("The selected annotation revision is unavailable.")
            annotation_root = (self.root / "shared" / "outputs" / "05-3d-pointcloud" / "annotations").resolve()
            location = (annotation_root / annotation_id).resolve()
            location.relative_to(annotation_root)
            if not (location / "annotation.json").is_file():
                raise ApiError("The selected annotation revision is incomplete.")
            shutil.rmtree(location)
            self.send_json(HTTPStatus.OK, {"deletedId": annotation_id})
        except ApiError as exc:
            self.send_json(HTTPStatus.BAD_REQUEST, {"error": str(exc)})
        except ValueError:
            self.send_json(HTTPStatus.BAD_REQUEST, {"error": "Invalid annotation location."})
        except Exception as exc:  # pragma: no cover - defensive server boundary
            self.log_error("annotation deletion failed: %s", exc)
            self.send_json(HTTPStatus.INTERNAL_SERVER_ERROR, {"error": "Annotation deletion failed. Check the console terminal for details."})

    def do_PUT(self) -> None:  # noqa: N802 - binary upload endpoint
        path = urlsplit(self.path).path
        if not path.startswith("/api/change-detection/uploads/") and not path.startswith("/api/anomaly-detection/uploads/") and not path.startswith("/api/3d-pointcloud/photo-uploads/") and not path.startswith("/api/3d-pointcloud/pointcloud-uploads/"):
            self.send_json(HTTPStatus.NOT_FOUND, {"error": "Unknown local API endpoint."})
            return
        try:
            if path.startswith("/api/anomaly-detection/uploads/"):
                result = self.receive_anomaly_upload(path)
            elif path.startswith("/api/3d-pointcloud/photo-uploads/"):
                result = self.receive_photo_reconstruction_upload(path)
            elif path.startswith("/api/3d-pointcloud/pointcloud-uploads/"):
                result = self.receive_pointcloud_upload(path)
            else:
                result = self.receive_change_upload(path)
            self.send_json(HTTPStatus.CREATED, result)
        except ApiError as exc:
            self.send_json(HTTPStatus.BAD_REQUEST, {"error": str(exc)})
        except Exception as exc:  # pragma: no cover - defensive server boundary
            self.log_error("binary upload failed: %s", exc)
            self.send_json(HTTPStatus.INTERNAL_SERVER_ERROR, {"error": "Binary upload failed. Check the console terminal for details."})

    def do_OPTIONS(self) -> None:  # noqa: N802
        self.send_response(HTTPStatus.NO_CONTENT)
        self.send_header("Allow", "GET, POST, PUT, DELETE, OPTIONS")
        self.end_headers()

    def read_json_body(self) -> dict[str, Any]:
        content_length = self.headers.get("Content-Length")
        if content_length is None or not content_length.isdigit():
            raise ApiError("A JSON request body with Content-Length is required.")
        size = int(content_length)
        if size <= 0 or size > MAX_REQUEST_BYTES:
            raise ApiError(f"Request must be between 1 byte and {MAX_REQUEST_BYTES // (1024 * 1024)} MB.")
        if "application/json" not in self.headers.get("Content-Type", ""):
            raise ApiError("Content-Type must be application/json.")
        try:
            payload = json.loads(self.rfile.read(size).decode("utf-8"))
        except (UnicodeDecodeError, json.JSONDecodeError) as exc:
            raise ApiError("Request body is not valid UTF-8 JSON.") from exc
        if not isinstance(payload, dict):
            raise ApiError("JSON request body must be an object.")
        return payload

    def run_command(self, command: list[str], timeout: int) -> None:
        completed = subprocess.run(command, cwd=self.root, capture_output=True, text=True, timeout=timeout, check=False)
        if completed.returncode:
            message = (completed.stderr or completed.stdout or "Unknown script error.").strip().splitlines()[-1]
            raise ApiError(f"Processing failed: {message[:600]}")

    def receive_change_upload(self, path: str) -> dict[str, Any]:
        return self.receive_binary_upload(path, "00-change-detection", {"before", "after"}, {".jpg", ".jpeg", ".png", ".tif", ".tiff"}, "change-detection")

    def receive_anomaly_upload(self, path: str) -> dict[str, Any]:
        return self.receive_binary_upload(path, "09-anomaly-detection", {"reference", "input"}, {".jpg", ".jpeg", ".png", ".tif", ".tiff"}, "anomaly-detection")

    def receive_photo_reconstruction_upload(self, path: str) -> dict[str, Any]:
        return self.receive_binary_upload(path, "05-3d-pointcloud", {"photo"}, {".jpg", ".jpeg"}, "photo reconstruction")

    def receive_pointcloud_upload(self, path: str) -> dict[str, Any]:
        return self.receive_binary_upload(path, "05-3d-pointcloud", {"pointcloud"}, {".ply", ".pcd", ".xyz", ".xyzn", ".xyzrgb", ".las", ".laz"}, "point-cloud")

    def receive_binary_upload(
        self,
        path: str,
        capability: str,
        allowed_roles: set[str],
        suffixes: set[str],
        label: str,
    ) -> dict[str, Any]:
        upload_id = path.rstrip("/").rsplit("/", 1)[-1]
        if not SAFE_UPLOAD_ID.fullmatch(upload_id):
            raise ApiError(f"Invalid {label} upload id.")
        query = parse_qs(urlsplit(self.path).query)
        role = query.get("role", [""])[0]
        if role not in allowed_roles:
            raise ApiError(f"Invalid {label} upload role.")
        encoded_name = self.headers.get("X-Upload-Name", "")
        if len(encoded_name) > 2048:
            raise ApiError("Encoded upload name is too long.")
        try:
            name = unquote(encoded_name, encoding="utf-8", errors="strict")
        except UnicodeError as exc:
            raise ApiError("Upload name is not valid UTF-8 percent encoding.") from exc
        safe_name = safe_file_name(name, suffixes)
        content_length = self.headers.get("Content-Length")
        if content_length is None or not content_length.isdigit():
            raise ApiError("Binary upload requires a Content-Length header.")
        size = int(content_length)
        if size <= 0 or size > MAX_FILE_BYTES:
            raise ApiError(f"Uploaded file must be between 1 byte and {MAX_FILE_BYTES // (1024 * 1024)} MB: {safe_name}.")
        staging = self.root / "shared" / "data" / "raw" / capability / "uploads" / upload_id
        staging.mkdir(parents=True, exist_ok=False)
        part = staging / f"{role}.part"
        target = staging / f"{role}{Path(safe_name).suffix.lower()}"
        remaining = size
        digest = hashlib.sha256()
        try:
            with part.open("wb") as stream:
                while remaining:
                    chunk = self.rfile.read(min(8 * 1024 * 1024, remaining))
                    if not chunk:
                        raise ApiError("Binary upload ended before Content-Length was reached.")
                    stream.write(chunk)
                    digest.update(chunk)
                    remaining -= len(chunk)
            part.replace(target)
            (staging / f"{role}.json").write_text(json.dumps({"role": role, "name": safe_name, "size": size, "sha256": digest.hexdigest()}), encoding="utf-8")
        except Exception:
            part.unlink(missing_ok=True)
            target.unlink(missing_ok=True)
            raise
        return {"uploadId": upload_id, "role": role, "name": safe_name, "size": size, "sha256": digest.hexdigest()}

    def resolve_change_upload(self, payload: Any, role: str) -> tuple[str, Path]:
        return self.resolve_binary_upload(payload, role, "00-change-detection", "change-detection")

    def resolve_anomaly_upload(self, payload: Any, role: str) -> tuple[str, Path, str]:
        name, path = self.resolve_binary_upload(payload, role, "09-anomaly-detection", "anomaly-detection")
        manifest = load_json(path.parent / f"{role}.json")
        return name, path, str(manifest.get("sha256") or "")

    def resolve_photo_reconstruction_upload(self, payload: Any) -> tuple[str, Path, str]:
        name, path = self.resolve_binary_upload(payload, "photo", "05-3d-pointcloud", "photo reconstruction")
        manifest = load_json(path.parent / "photo.json")
        return name, path, str(manifest.get("sha256") or "")

    def resolve_pointcloud_upload(self, payload: Any) -> tuple[str, Path, str]:
        name, path = self.resolve_binary_upload(payload, "pointcloud", "05-3d-pointcloud", "point-cloud")
        manifest = load_json(path.parent / "pointcloud.json")
        return name, path, str(manifest.get("sha256") or "")

    def resolve_binary_upload(self, payload: Any, role: str, capability: str, label: str) -> tuple[str, Path]:
        if not isinstance(payload, dict) or not isinstance(payload.get("uploadId"), str):
            raise ApiError(f"{label} uploads must include a {role} uploadId.")
        upload_id = payload["uploadId"]
        if not SAFE_UPLOAD_ID.fullmatch(upload_id):
            raise ApiError(f"Invalid {label} upload id.")
        staging = self.root / "shared" / "data" / "raw" / capability / "uploads" / upload_id
        manifest = load_json(staging / f"{role}.json")
        name = str(manifest.get("name") or "")
        path = staging / f"{role}{Path(name).suffix.lower()}"
        if manifest.get("role") != role or not name or not path.is_file():
            raise ApiError(f"The staged {role} upload is unavailable or incomplete.")
        return name, path

    def create_trajectory_run(self, payload: dict[str, Any]) -> dict[str, Any]:
        files = payload.get("files")
        if not isinstance(files, dict):
            raise ApiError("Trajectory request must contain a files object.")
        required = {
            "flight": {".xlsx"},
            "route": {".kmz"},
            "restricted": {".geojson"},
        }
        decoded = {key: decode_upload(files.get(key), suffixes) for key, suffixes in required.items()}
        flyable = decode_upload(files["flyable"], {".gzip"}) if files.get("flyable") else None
        run_id = make_run_id("trajectory")
        raw_root = self.root / "shared" / "data" / "raw" / "15-trajectory-analysis" / "runs" / run_id
        paths = {"flight": raw_root / "tracks" / decoded["flight"][0], "route": raw_root / "routes" / decoded["route"][0], "restricted": raw_root / "areas" / decoded["restricted"][0]}
        for key, path in paths.items():
            path.parent.mkdir(parents=True, exist_ok=True)
            path.write_bytes(decoded[key][1])
        if flyable:
            flyable_path = raw_root / "areas" / flyable[0]
            flyable_path.write_bytes(flyable[1])
        processed = self.root / "shared" / "data" / "processed" / "15-trajectory-analysis" / run_id
        output_parent = self.root / "shared" / "outputs" / "15-trajectory-analysis" / "runs" / run_id
        python = self.root / ".venvs" / "15-trajectory-analysis" / "Scripts" / "python.exe"
        if not python.is_file():
            raise ApiError("Trajectory virtual environment is unavailable. Run the capability setup first.")
        with RUN_LOCK:
            self.run_command([str(python), str(self.root / "capabilities" / "15-trajectory-analysis" / "prepare_real_flight.py"), "--raw-dir", str(raw_root), "--output", str(processed), "--case-id", run_id], 300)
            self.run_command([str(python), str(self.root / "capabilities" / "15-trajectory-analysis" / "run_trajectory_analysis.py"), "--input", str(processed / f"{run_id}.case.json"), "--output", str(output_parent)], 300)
        artifact = output_parent / run_id
        if not (artifact / "run_metadata.json").is_file():
            raise ApiError("Trajectory script finished without the expected result metadata.")
        return next(item for item in trajectory_runs(self.root) if item["id"] == run_id)

    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.")
        run_id = make_run_id("detection")
        raw_root = self.root / "shared" / "data" / "raw" / "01-object-detection" / "runs" / run_id
        raw_root.mkdir(parents=True, exist_ok=False)
        for name, content in decoded:
            (raw_root / name).write_bytes(content)
        output = self.root / "shared" / "outputs" / "01-object-detection" / "runs" / run_id
        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)
        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")
            staged["after"] = self.resolve_change_upload(uploads.get("after"), "after")
        elif isinstance(files, dict):
            decoded_before = decode_upload(files.get("before"), {".jpg", ".jpeg", ".png", ".tif", ".tiff"})
            decoded_after = decode_upload(files.get("after"), {".jpg", ".jpeg", ".png", ".tif", ".tiff"})
        else:
            raise ApiError("Change-detection request must contain before and after files or uploads.")
        threshold_value = payload.get("threshold", CHANGE_THRESHOLD_DEFAULT)
        if isinstance(threshold_value, bool) or not isinstance(threshold_value, (int, float)):
            raise ApiError("Change-detection threshold must be a number between 0.01 and 0.99.")
        threshold = float(threshold_value)
        if not CHANGE_THRESHOLD_MIN <= threshold <= CHANGE_THRESHOLD_MAX:
            raise ApiError("Change-detection threshold must be between 0.01 and 0.99.")
        processing_mode = payload.get("processingMode", CHANGE_PROCESSING_MODE_DEFAULT)
        if not isinstance(processing_mode, str) or processing_mode not in CHANGE_PROCESSING_MODES:
            raise ApiError("Change-detection processing mode must be auto, image, or geotiff.")
        max_dimension_value = payload.get("maxDimension", CHANGE_MAX_DIMENSION_AUTO)
        if isinstance(max_dimension_value, bool) or not isinstance(max_dimension_value, int):
            raise ApiError("Change-detection resolution must be an integer: 0 or between 512 and 4096.")
        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:
            before_name, after_name = decoded_before[0], decoded_after[0]
        run_id = make_run_id("change")
        raw_root = self.root / "shared" / "data" / "raw" / "00-change-detection" / "runs" / run_id
        before_path = raw_root / "before" / before_name
        after_path = raw_root / "after" / after_name
        before_path.parent.mkdir(parents=True, exist_ok=False)
        after_path.parent.mkdir(parents=True, exist_ok=False)
        if staged:
            shutil.copyfile(staged["before"][1], before_path)
            shutil.copyfile(staged["after"][1], after_path)
        else:
            before_path.write_bytes(decoded_before[1])
            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
        with RUN_LOCK:
            self.run_command(
                [
                    execution["python"],
                    str(self.root / "capabilities" / "00-change-detection" / "run_change_detection.py"),
                    "--before", str(before_path),
                    "--after", str(after_path),
                    "--threshold", f"{threshold:.4f}",
                    "--max-dimension", str(max_dimension),
                    "--processing-mode", processing_mode,
                    "--device", execution["device"],
                    "--processed-output", str(processed_root),
                    "--output", str(output),
                ],
                1200,
            )
        metadata_path = output / "run_metadata.json"
        if not metadata_path.is_file():
            raise ApiError("Change-detection script finished without the expected result metadata.")
        metadata = load_json(metadata_path)
        metadata["raw_input_dir"] = relative_path(self.root, raw_root)
        metadata["processed_input_dir"] = relative_path(self.root, processed_root)
        metadata["raw_before"] = relative_path(self.root, before_path)
        metadata["raw_after"] = relative_path(self.root, after_path)
        metadata_path.write_text(json.dumps(metadata, ensure_ascii=False, indent=2), encoding="utf-8")
        return next(item for item in change_runs(self.root) if item["id"] == run_id)

    def _scan_parameters(self, payload: dict[str, Any]) -> tuple[list[float], list[int]]:
        raw_thresholds = payload.get("thresholds", SCAN_DEFAULT_THRESHOLDS)
        raw_areas = payload.get("minimumAreas", SCAN_DEFAULT_AREAS)
        if not isinstance(raw_thresholds, list) or not raw_thresholds or len(raw_thresholds) > SCAN_MAX_THRESHOLDS:
            raise ApiError(f"Parameter scan thresholds must contain 1-{SCAN_MAX_THRESHOLDS} values.")
        if not isinstance(raw_areas, list) or not raw_areas or len(raw_areas) > SCAN_MAX_AREAS:
            raise ApiError(f"Parameter scan minimum areas must contain 1-{SCAN_MAX_AREAS} values.")
        thresholds: list[float] = []
        for value in raw_thresholds:
            if isinstance(value, bool) or not isinstance(value, (int, float)):
                raise ApiError("Each scan threshold must be a number between 0.01 and 0.99.")
            number = round(float(value), 4)
            if not CHANGE_THRESHOLD_MIN <= number <= CHANGE_THRESHOLD_MAX:
                raise ApiError("Each scan threshold must be between 0.01 and 0.99.")
            if number not in thresholds:
                thresholds.append(number)
        areas: list[int] = []
        for value in raw_areas:
            if isinstance(value, bool) or not isinstance(value, int) or not 16 <= value <= 200000:
                raise ApiError("Each scan minimum area must be an integer between 16 and 200000 pixels.")
            if value not in areas:
                areas.append(value)
        if len(thresholds) * len(areas) > SCAN_MAX_COMBINATIONS:
            raise ApiError(f"A parameter scan accepts at most {SCAN_MAX_COMBINATIONS} combinations.")
        return thresholds, areas

    def _scan_root(self, scan_id: str) -> Path:
        if not SAFE_SCAN_ID.fullmatch(scan_id):
            raise ApiError("Invalid parameter-scan id.")
        scan_root = self.root / "shared" / "outputs" / "00-change-detection" / "parameter-scans" / scan_id
        if not scan_root.is_dir() or not (scan_root / "scan_summary.json").is_file():
            raise ApiError("The parameter-scan result is unavailable.")
        return scan_root

    def promote_change_scan(self, scan_id: str, payload: dict[str, Any]) -> dict[str, Any]:
        scan_root = self._scan_root(scan_id)
        result_id = payload.get("resultId")
        if not isinstance(result_id, str) or not SAFE_SCAN_RESULT_ID.fullmatch(result_id):
            raise ApiError("A valid parameter-scan resultId is required.")
        result_dir = scan_root / result_id
        summary = load_json(result_dir / "summary.json")
        full_result = load_json(result_dir / "full_result.json")
        inference_run_id = str(load_json(scan_root / "scan_summary.json").get("source_run") or "")
        inference_output = self.root / "shared" / "outputs" / "00-change-detection" / "runs" / inference_run_id
        inference_metadata = load_json(inference_output / "run_metadata.json")
        if not result_dir.is_dir() or not (result_dir / "changes.geojson").is_file() or not inference_metadata:
            raise ApiError("The selected scan result is incomplete and cannot be promoted.")
        run_id = make_run_id("change")
        output = self.root / "shared" / "outputs" / "00-change-detection" / "runs" / run_id
        output.mkdir(parents=True, exist_ok=False)
        for source_name, destination_name in (
            ("change_probability.tif", "change_probability.tif"),
            ("change_mask.tif", "change_mask.tif"),
            ("generic_difference_mask.tif", "generic_difference_mask.tif"),
            ("change_model_overlay.jpg", "change_model_overlay.jpg"),
            ("before_processed_preview.jpg", "before_processed_preview.jpg"),
            ("after_registered_preview.jpg", "after_registered_preview.jpg"),
            ("changes.geojson", "changes.geojson"),
            ("changes_rectangles.geojson", "changes_rectangles.geojson"),
            ("changes_rectangles_wgs84.geojson", "changes_rectangles_wgs84.geojson"),
        ):
            source = result_dir / source_name if source_name.startswith("change_mask") or source_name.startswith("changes") else inference_output / source_name
            if source.is_file():
                shutil.copyfile(source, output / destination_name)
        overlay_source = result_dir / "overlay_preview.jpg"
        if (inference_output / "change_overlay.jpg").is_file() and float(summary.get("threshold", 0.5)) == float(inference_metadata.get("thresholds", {}).get("change_probability", 0.5)):
            overlay_source = inference_output / "change_overlay.jpg"
        if not overlay_source.is_file():
            raise ApiError("The selected scan preview is unavailable.")
        shutil.copyfile(overlay_source, output / "change_overlay.jpg")
        metadata = dict(inference_metadata)
        scan_raw_root = self.root / "shared" / "data" / "raw" / "00-change-detection" / "runs" / scan_id
        before_raw = scan_raw_root / "before" / str(inference_metadata.get("input_files", ["before.tif", "after.tif"])[0])
        after_raw = scan_raw_root / "after" / str(inference_metadata.get("input_files", ["before.tif", "after.tif"])[1])
        rectangle_count = int(full_result.get("rectangle_vector_feature_count") or 0)
        if rectangle_count == 0 and (result_dir / "changes_rectangles.geojson").is_file():
            rectangle_count = len(load_json(result_dir / "changes_rectangles.geojson").get("features", []))
        metadata.update(
            {
                "kind": "formal-change-run",
                "created_at": datetime.now(UTC).isoformat(),
                "thresholds": {"change_probability": float(summary.get("threshold", 0.5)), "minimum_component_pixels": int(summary.get("minimum_area_pixels", 16))},
                "raw_changed_pixels": int(summary.get("raw_changed_pixels", 0)),
                "changed_pixels": int(summary.get("changed_pixels", 0)),
                "changed_pixel_ratio": float(summary.get("changed_pixel_ratio", 0)),
                "vector_feature_count": int(full_result.get("full_vector_feature_count", summary.get("vector_feature_count", 0))),
                "rectangle_feature_count": rectangle_count,
                "promoted_from_scan": scan_id,
                "promoted_result": result_id,
                "raw_input_dir": relative_path(self.root, scan_raw_root),
                "raw_before": relative_path(self.root, before_raw),
                "raw_after": relative_path(self.root, after_raw),
                "processed_input_dir": relative_path(self.root, self.root / "shared" / "data" / "processed" / "00-change-detection" / scan_id),
                "generic_difference": inference_metadata.get("generic_difference"),
                "artifacts": {
                    "probability_raster": "change_probability.tif",
                    "raw_mask_raster": "change_mask.tif",
                    "mask_raster": "change_mask.tif",
                    "generic_difference_mask": "generic_difference_mask.tif" if (output / "generic_difference_mask.tif").is_file() else None,
                    "overlay": "change_overlay.jpg",
                    "model_overlay": "change_model_overlay.jpg" if (output / "change_model_overlay.jpg").is_file() else None,
                    "before_processed_preview": "before_processed_preview.jpg" if (output / "before_processed_preview.jpg").is_file() else None,
                    "after_registered_preview": "after_registered_preview.jpg" if (output / "after_registered_preview.jpg").is_file() else None,
                    "vector": "changes.geojson",
                    "rectangle_vector": "changes_rectangles.geojson",
                    "rectangle_vector_wgs84": "changes_rectangles_wgs84.geojson" if (output / "changes_rectangles_wgs84.geojson").is_file() else None,
                    "features": "full_result.json",
                },
            }
        )
        (output / "full_result.json").write_text(json.dumps(full_result, ensure_ascii=False, indent=2), encoding="utf-8")
        (output / "run_metadata.json").write_text(json.dumps(metadata, ensure_ascii=False, indent=2), encoding="utf-8")
        return next(item for item in change_runs(self.root) if item["id"] == run_id)

    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 = {
            "before": self.resolve_change_upload(uploads.get("before"), "before"),
            "after": self.resolve_change_upload(uploads.get("after"), "after"),
        }
        thresholds, areas = self._scan_parameters(payload)
        processing_mode = payload.get("processingMode", CHANGE_PROCESSING_MODE_DEFAULT)
        if not isinstance(processing_mode, str) or processing_mode not in CHANGE_PROCESSING_MODES:
            raise ApiError("Change-detection processing mode must be auto, image, or geotiff.")
        max_dimension_value = payload.get("maxDimension", CHANGE_MAX_DIMENSION_AUTO)
        if isinstance(max_dimension_value, bool) or not isinstance(max_dimension_value, int):
            raise ApiError("Change-detection resolution must be an integer: 0 or between 512 and 4096.")
        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]
        after_path = raw_root / "after" / staged["after"][0]
        before_path.parent.mkdir(parents=True, exist_ok=False)
        after_path.parent.mkdir(parents=True, exist_ok=False)
        shutil.copyfile(staged["before"][1], before_path)
        shutil.copyfile(staged["after"][1], after_path)
        processed_root = self.root / "shared" / "data" / "processed" / "00-change-detection" / run_id
        inference_output = self.root / "shared" / "outputs" / "00-change-detection" / "runs" / run_id
        scan_output = self.root / "shared" / "outputs" / "00-change-detection" / "parameter-scans" / run_id
        with SCAN_JOBS_LOCK:
            SCAN_JOBS[run_id] = {
                "id": run_id,
                "status": "queued",
                "createdAt": datetime.now(UTC).isoformat(),
                "thresholds": thresholds,
                "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, execution),
            daemon=True,
            name=f"change-scan-{run_id}",
        )
        thread.start()
        return dict(SCAN_JOBS[run_id])

    def _update_scan_job(self, job_id: str, **values: Any) -> None:
        with SCAN_JOBS_LOCK:
            if job_id in SCAN_JOBS:
                SCAN_JOBS[job_id].update(values)

    def _run_change_scan(
        self,
        run_id: str,
        before_path: Path,
        after_path: Path,
        processed_root: Path,
        inference_output: Path,
        scan_output: Path,
        thresholds: list[float],
        areas: list[int],
        processing_mode: str,
        max_dimension: int,
        execution: dict[str, Any],
    ) -> None:
        python = execution["python"]
        try:
            self._update_scan_job(run_id, status="running", phase="inference")
            with RUN_LOCK:
                self.run_command(
                    [
                        python,
                        str(self.root / "capabilities" / "00-change-detection" / "run_change_detection.py"),
                        "--before", str(before_path),
                        "--after", str(after_path),
                        "--threshold", "0.5000",
                        "--max-dimension", str(max_dimension),
                        "--processing-mode", processing_mode,
                        "--device", execution["device"],
                        "--processed-output", str(processed_root),
                        "--output", str(inference_output),
                    ],
                    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
                inference_metadata_path.write_text(json.dumps(inference_metadata, ensure_ascii=False, indent=2), encoding="utf-8")
                self._update_scan_job(run_id, phase="parameter-scan")
                command = [
                    python,
                    str(self.root / "capabilities" / "00-change-detection" / "scan_change_detection_parameters.py"),
                    "--run-dir", str(inference_output),
                    "--output", str(scan_output),
                ]
                for threshold in thresholds:
                    command.extend(["--threshold", f"{threshold:.4f}"])
                for area in areas:
                    command.extend(["--minimum-area", str(area)])
                self.run_command(command, SCAN_JOB_TIMEOUT)
                self._update_scan_job(run_id, phase="vectorization")
                vector_command = [
                    python,
                    str(self.root / "capabilities" / "00-change-detection" / "materialize_parameter_scan_candidates.py"),
                    "--scan-dir", str(scan_output),
                ]
                for threshold in thresholds:
                    for area in areas:
                        vector_command.extend(["--candidate", f"threshold-{threshold:.2f}_area-{area}"])
                self.run_command(vector_command, SCAN_JOB_TIMEOUT)
            metadata = {
                "capability": "00-change-detection",
                "kind": "parameter-scan",
                "source_run": run_id,
                "created_at": datetime.now(UTC).isoformat(),
                "thresholds": thresholds,
                "minimum_areas": areas,
                "processing_mode": processing_mode,
                "max_dimension": max_dimension,
                "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),
            }
            scan_output.mkdir(parents=True, exist_ok=True)
            (scan_output / "scan_metadata.json").write_text(json.dumps(metadata, ensure_ascii=False, indent=2), encoding="utf-8")
            self._update_scan_job(run_id, status="completed", phase="done", scanId=run_id)
        except Exception as exc:  # background errors are returned through polling
            scan_output.mkdir(parents=True, exist_ok=True)
            (scan_output / "scan_failed.json").write_text(json.dumps({"job_id": run_id, "error": str(exc)[:600]}, ensure_ascii=False, indent=2), encoding="utf-8")
            self._update_scan_job(run_id, status="failed", phase="error", error=str(exc)[:600])

    def create_semantic_run(self, payload: dict[str, Any]) -> dict[str, Any]:
        task_id = str(payload.get("taskId") or "color_baseline")
        task = next((item for item in semantic_tasks(self.root) if item["id"] == task_id), None)
        if task is None:
            raise ApiError(f"Unknown semantic-mapping task: {task_id}.")
        if task.get("selectable") is not True:
            raise ApiError(f"Semantic-mapping task is not runnable yet: {task_id}.")
        uploads = payload.get("images")
        if not isinstance(uploads, list) or not uploads:
            raise ApiError("Semantic-mapping request must include at least one image.")
        if len(uploads) > MAX_SEGMENTATION_IMAGES_PER_RUN:
            raise ApiError(f"A semantic-mapping run accepts at most {MAX_SEGMENTATION_IMAGES_PER_RUN} images.")
        decoded = [decode_upload(item, {".jpg", ".jpeg", ".png", ".tif", ".tiff"}) 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.")
        run_id = make_run_id("semantic")
        raw_root = self.root / "shared" / "data" / "raw" / "02-semantic-mapping" / "runs" / run_id
        processed_root = self.root / "shared" / "data" / "processed" / "02-semantic-mapping" / run_id
        raw_root.mkdir(parents=True, exist_ok=False)
        processed_root.mkdir(parents=True, exist_ok=False)
        for name, content in decoded:
            (raw_root / name).write_bytes(content)
            (processed_root / name).write_bytes(content)
        output = self.root / "shared" / "outputs" / "02-semantic-mapping" / "runs" / run_id
        python = self.root / ".venvs" / "02-semantic-mapping" / "Scripts" / "python.exe"
        if not python.is_file():
            raise ApiError("Semantic-mapping virtual environment is unavailable. Run the capability setup first.")
        with RUN_LOCK:
            self.run_command([str(python), str(self.root / "capabilities" / "02-semantic-mapping" / "run_semantic_segmentation.py"), "--input", str(processed_root), "--output", str(output)], 900)
        metadata_path = output / "run_metadata.json"
        if not metadata_path.is_file():
            raise ApiError("Semantic-mapping script finished without the expected result metadata.")
        metadata = load_json(metadata_path)
        metadata["task_id"] = task_id
        metadata["task_name"] = str(task.get("name") or task_id)
        metadata["input_dir"] = relative_path(self.root, processed_root)
        metadata["raw_input_dir"] = relative_path(self.root, raw_root)
        metadata_path.write_text(json.dumps(metadata, ensure_ascii=False, indent=2), encoding="utf-8")
        return next(item for item in semantic_runs(self.root) if item["id"] == run_id)

    def create_anomaly_run(self, payload: dict[str, Any]) -> dict[str, Any]:
        tile_size, stride, threshold_quantile, random_state = validate_anomaly_parameters(payload)
        uploads = payload.get("uploads")
        if not isinstance(uploads, dict):
            raise ApiError("Anomaly-detection request must contain reference and input uploads.")
        reference_values = uploads.get("reference")
        input_values = uploads.get("input")
        if not isinstance(reference_values, list) or not reference_values:
            raise ApiError("Select at least one normal reference image.")
        if not isinstance(input_values, list) or not input_values:
            raise ApiError("Select at least one image to inspect.")
        if len(reference_values) > MAX_ANOMALY_IMAGES_PER_ROLE or len(input_values) > MAX_ANOMALY_IMAGES_PER_ROLE:
            raise ApiError(f"An anomaly-detection run accepts at most {MAX_ANOMALY_IMAGES_PER_ROLE} images in each group.")
        references = [self.resolve_anomaly_upload(value, "reference") for value in reference_values]
        inputs = [self.resolve_anomaly_upload(value, "input") for value in input_values]
        if len({name.casefold() for name, _, _ in references}) != len(references):
            raise ApiError("Normal reference image names must be unique within one run.")
        if len({name.casefold() for name, _, _ in inputs}) != len(inputs):
            raise ApiError("Input image names must be unique within one run.")

        python = self.root / ".venvs" / "09-anomaly-detection" / "Scripts" / "python.exe"
        if not python.is_file():
            raise ApiError("Anomaly-detection virtual environment is unavailable. Run the capability setup first.")
        run_id = make_run_id("anomaly")
        job_id = uuid4().hex
        raw_root = self.root / "shared" / "data" / "raw" / "09-anomaly-detection" / "runs" / run_id
        raw_reference = raw_root / "reference"
        raw_input = raw_root / "input"
        processed_root = self.root / "shared" / "data" / "processed" / "09-anomaly-detection" / run_id
        processed_reference = processed_root / "reference"
        processed_input = processed_root / "input"
        output = self.root / "shared" / "outputs" / "09-anomaly-detection" / "runs" / run_id
        for directory in (raw_reference, raw_input, processed_reference, processed_input):
            directory.mkdir(parents=True, exist_ok=False)
        for group, raw_dir, processed_dir in ((references, raw_reference, processed_reference), (inputs, raw_input, processed_input)):
            for name, staged_path, expected_sha256 in group:
                raw_path = raw_dir / name
                processed_path = processed_dir / name
                shutil.copyfile(staged_path, raw_path)
                if expected_sha256 and file_sha256(raw_path) != expected_sha256:
                    raise ApiError(f"Uploaded file checksum changed while staging: {name}.")
                shutil.copyfile(raw_path, processed_path)
        for _, staged_path, _ in references + inputs:
            shutil.rmtree(staged_path.parent)

        created_at = datetime.now(UTC).isoformat()
        job = {"id": job_id, "runId": run_id, "status": "queued", "createdAt": created_at}
        with ANOMALY_JOB_LOCK:
            ANOMALY_JOBS[job_id] = job
        thread = threading.Thread(
            target=execute_anomaly_job,
            args=(self.root, job_id, run_id, raw_reference, raw_input, processed_reference, processed_input, output, tile_size, stride, threshold_quantile, random_state),
            daemon=True,
            name=f"anomaly-{run_id}",
        )
        thread.start()
        return dict(job)

    def create_measurement_run(self, payload: dict[str, Any]) -> dict[str, Any]:
        uploads = payload.get("rasters")
        if not isinstance(uploads, list) or not uploads:
            raise ApiError("Spatial-measurement request must include at least one label raster.")
        if len(uploads) > MAX_MEASUREMENT_RASTERS_PER_RUN:
            raise ApiError(f"A spatial-measurement run accepts at most {MAX_MEASUREMENT_RASTERS_PER_RUN} rasters.")
        decoded = [decode_upload(item, {".png", ".tif", ".tiff"}) for item in uploads]
        if len({name.casefold() for name, _ in decoded}) != len(decoded):
            raise ApiError("Uploaded raster names must be unique within one run.")
        run_id = make_run_id("measurement")
        raw_root = self.root / "shared" / "data" / "raw" / "04-spatial-measurement" / "runs" / run_id
        processed_root = self.root / "shared" / "data" / "processed" / "04-spatial-measurement" / run_id
        raw_root.mkdir(parents=True, exist_ok=False)
        processed_root.mkdir(parents=True, exist_ok=False)
        for name, content in decoded:
            (raw_root / name).write_bytes(content)
            (processed_root / name).write_bytes(content)
        output = self.root / "shared" / "outputs" / "04-spatial-measurement" / "runs" / run_id
        python = self.root / ".venvs" / "04-spatial-measurement" / "Scripts" / "python.exe"
        if not python.is_file():
            raise ApiError("Spatial-measurement virtual environment is unavailable. Run the capability setup first.")
        with RUN_LOCK:
            self.run_command([str(python), str(self.root / "capabilities" / "04-spatial-measurement" / "run_spatial_measurement.py"), "--input", str(processed_root), "--output", str(output)], 900)
        metadata_path = output / "run_metadata.json"
        if not metadata_path.is_file():
            raise ApiError("Spatial-measurement script finished without the expected result metadata.")
        metadata = load_json(metadata_path)
        metadata["input_dir"] = relative_path(self.root, processed_root)
        metadata["raw_input_dir"] = relative_path(self.root, raw_root)
        metadata_path.write_text(json.dumps(metadata, ensure_ascii=False, indent=2), encoding="utf-8")
        return next(item for item in measurement_runs(self.root) if item["id"] == run_id)

    def create_pointcloud_run(self, payload: dict[str, Any]) -> dict[str, Any]:
        uploads = payload.get("pointClouds")
        source_dense_run_id = payload.get("sourceDenseRunId")
        if uploads is None and not isinstance(source_dense_run_id, str):
            raise ApiError("3D point-cloud request must include at least one PLY, PCD, XYZ, LAS, or LAZ file.")
        if uploads is not None and (not isinstance(uploads, list) or not uploads):
            raise ApiError("3D point-cloud request must include at least one PLY, PCD, XYZ, LAS, or LAZ file.")
        if isinstance(uploads, list) and len(uploads) > MAX_POINTCLOUDS_PER_RUN:
            raise ApiError(f"A 3D point-cloud run accepts at most {MAX_POINTCLOUDS_PER_RUN} files.")
        suffixes = {".ply", ".pcd", ".xyz", ".xyzn", ".xyzrgb", ".las", ".laz"}
        staged_upload_dirs: list[Path] = []
        if isinstance(source_dense_run_id, str):
            if SAFE_FILE_NAME.search(source_dense_run_id) or len(source_dense_run_id) > 120:
                raise ApiError("Invalid dense point-cloud source run id.")
            source_case = next((item for item in pointcloud_runs(self.root) if item["id"] == source_dense_run_id), None)
            if not source_case:
                raise ApiError("The selected dense point-cloud source is unavailable.")
            source_artifact = self.root / str(source_case["artifactRoot"])
            source_metadata = load_json(source_artifact / "run_metadata.json")
            dense = source_metadata.get("dense_photo_reconstruction")
            source_file = dense.get("dense_point_cloud_file") if isinstance(dense, dict) else None
            source_path = source_artifact / str(source_file or "")
            if not isinstance(source_file, str) or source_path.suffix.lower() != ".ply" or not source_path.is_file():
                raise ApiError("The selected run has no available dense PLY output.")
            decoded = [(f"{source_dense_run_id}-dense.ply", source_path, file_sha256(source_path))]
        elif all(isinstance(item, dict) and isinstance(item.get("content"), str) for item in uploads):
            decoded = [(name, content, "") for name, content in (decode_upload(item, suffixes) for item in uploads)]
        else:
            decoded = [self.resolve_pointcloud_upload(item) for item in uploads]
            staged_upload_dirs = [path.parent for _, path, _ in decoded]
        if len({name.casefold() for name, _, _ in decoded}) != len(decoded):
            raise ApiError("Uploaded point-cloud names must be unique within one run.")
        run_id = make_run_id("pointcloud")
        raw_root = self.root / "shared" / "data" / "raw" / "05-3d-pointcloud" / "runs" / run_id
        processed_root = self.root / "shared" / "data" / "processed" / "05-3d-pointcloud" / run_id
        raw_root.mkdir(parents=True, exist_ok=False)
        processed_root.mkdir(parents=True, exist_ok=False)
        source_sha256: dict[str, str] = {}
        source_bytes: dict[str, int] = {}
        for name, staged_or_content, expected_sha256 in decoded:
            raw_path = raw_root / name
            if isinstance(staged_or_content, bytes):
                raw_path.write_bytes(staged_or_content)
            else:
                shutil.copyfile(staged_or_content, raw_path)
            if expected_sha256 and file_sha256(raw_path) != expected_sha256:
                raise ApiError(f"Uploaded point-cloud checksum changed while staging: {name}.")
            source_sha256[name] = file_sha256(raw_path)
            source_bytes[name] = raw_path.stat().st_size
            shutil.copyfile(raw_path, processed_root / name)
        output = self.root / "shared" / "outputs" / "05-3d-pointcloud" / "runs" / run_id
        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.")
        command = [str(python), str(self.root / "capabilities" / "05-3d-pointcloud" / "run_pointcloud_understanding.py"), "--input", str(processed_root), "--output", str(output), "--ground-up-axis", "z"]
        with RUN_LOCK:
            self.run_command(command, 900)
        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)
        metadata["source_sha256"] = source_sha256
        metadata["source_bytes"] = source_bytes
        metadata_path.write_text(json.dumps(metadata, ensure_ascii=False, indent=2), encoding="utf-8")
        for staging in staged_upload_dirs:
            shutil.rmtree(staging)
        return next(item for item in pointcloud_runs(self.root) if item["id"] == run_id)

    def create_pointcloud_annotation(self, payload: dict[str, Any]) -> dict[str, Any]:
        source_id = payload.get("sourceId")
        labels = payload.get("labels")
        if not isinstance(source_id, str) or not isinstance(labels, list):
            raise ApiError("Annotation request must include a sourceId and labels array.")
        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.")
        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 class_by_code:
                raise ApiError("Annotation contains an out-of-range point index or unsupported class code.")
            compact[index] = code
        if not compact:
            raise ApiError("Save at least one user-confirmed point label.")
        annotation_id = make_run_id("annotation")
        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(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): 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")
        device = payload.get("device", "auto")
        if not isinstance(annotation_id, str) or SAFE_FILE_NAME.search(annotation_id) or len(annotation_id) > 120:
            raise ApiError("Invalid annotation id.")
        if device not in {"auto", "cpu", "cuda"}:
            raise ApiError("Training device must be auto, cpu, or cuda.")
        annotation = self.root / "shared" / "outputs" / "05-3d-pointcloud" / "annotations" / annotation_id / "annotation.json"
        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("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, 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)
        if not model_record:
            raise ApiError("The selected trained model is unavailable or incomplete.")
        name, staged_path, expected_sha256 = self.resolve_pointcloud_upload(upload)
        suffixes = {".ply", ".pcd", ".xyz", ".xyzn", ".xyzrgb", ".las", ".laz"}
        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)
        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
        output = self.root / "shared" / "outputs" / "05-3d-pointcloud" / "model-inference-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)
        actual_sha256 = file_sha256(raw_path)
        if expected_sha256 and actual_sha256 != expected_sha256:
            raise ApiError("Uploaded point-cloud checksum changed while staging.")
        processed_path = processed_root / name
        shutil.copyfile(raw_path, processed_path)
        # Only remove the staging copy after its immutable raw copy was verified.
        shutil.rmtree(staged_path.parent)
        model_path = self.root / str(model_record["model"])
        training_root = (self.root / "shared" / "outputs" / "05-3d-pointcloud" / "training-runs").resolve()
        try:
            model_path.resolve().relative_to(training_root)
        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": 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")
        if not isinstance(uploads, list) or len(uploads) < 3:
            raise ApiError("Photo reconstruction needs at least three JPG/JPEG photos from one coherent flight or camera sequence.")
        if len(uploads) > MAX_PHOTO_RECONSTRUCTION_IMAGES_PER_RUN:
            raise ApiError(f"A photo reconstruction run accepts at most {MAX_PHOTO_RECONSTRUCTION_IMAGES_PER_RUN} photos.")
        use_position_priors = payload.get("usePositionPriors", False)
        if not isinstance(use_position_priors, bool):
            raise ApiError("Photo reconstruction usePositionPriors must be true or false.")
        photos = [self.resolve_photo_reconstruction_upload(value) for value in uploads]
        if len({name.casefold() for name, _, _ in photos}) != len(photos):
            raise ApiError("Uploaded photo names must be unique within one run.")
        python = self.root / ".venvs" / "05-3d-pointcloud" / "Scripts" / "python.exe"
        openmvs = self.root / "shared" / "tools" / "openmvs-2.4.0" / "vc17" / "x64" / "Release"
        if not python.is_file() or not (openmvs / "DensifyPointCloud.exe").is_file():
            raise ApiError("Photo-reconstruction CPU environment is unavailable. Run the capability setup first.")

        run_id = make_run_id("photo-reconstruction")
        job_id = uuid4().hex
        raw_root = self.root / "shared" / "data" / "raw" / "05-3d-pointcloud" / "runs" / run_id
        processed_root = self.root / "shared" / "data" / "processed" / "05-3d-pointcloud" / run_id
        sparse_output = processed_root / "sparse_sfm"
        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)
        source_sha256: dict[str, str] = {}
        source_bytes: dict[str, int] = {}
        for name, staged_path, expected_sha256 in photos:
            raw_path = raw_root / name
            shutil.copyfile(staged_path, raw_path)
            actual_sha256 = file_sha256(raw_path)
            if expected_sha256 and actual_sha256 != expected_sha256:
                raise ApiError(f"Uploaded file checksum changed while staging: {name}.")
            shutil.copyfile(raw_path, processed_root / name)
            source_sha256[name] = actual_sha256
            source_bytes[name] = raw_path.stat().st_size
        for _, staged_path, _ in photos:
            shutil.rmtree(staged_path.parent)

        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(
            target=execute_photo_reconstruction_job,
            args=(self.root, job_id, run_id, raw_root, processed_root, sparse_output, output, source_sha256, source_bytes, use_position_priors),
            daemon=True,
            name=f"photo-reconstruction-{run_id}",
        )
        thread.start()
        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")
        if not isinstance(files, dict) or set(files) != RISK_RULE_REQUIRED_FILES:
            raise ApiError("Risk-rule request must contain observations, zones and rules files.")
        observations_name, observations_bytes = decode_upload(files["observations"], {".geojson"})
        zones_name, zones_bytes = decode_upload(files["zones"], {".geojson"})
        rules_name, rules_bytes = decode_upload(files["rules"], {".json"})
        if len({observations_name.casefold(), zones_name.casefold(), rules_name.casefold()}) != 3:
            raise ApiError("Risk-rule uploaded file names must be unique.")
        run_id = make_run_id("risk")
        raw_root = self.root / "shared" / "data" / "raw" / "07-risk-rule-engine" / "runs" / run_id
        processed_root = self.root / "shared" / "data" / "processed" / "07-risk-rule-engine" / run_id
        raw_root.mkdir(parents=True, exist_ok=False)
        processed_root.mkdir(parents=True, exist_ok=False)
        staged = ((observations_name, observations_bytes), (zones_name, zones_bytes), (rules_name, rules_bytes))
        for name, content in staged:
            (raw_root / name).write_bytes(content)
            (processed_root / name).write_bytes(content)
        output = self.root / "shared" / "outputs" / "07-risk-rule-engine" / "runs" / run_id
        python = self.root / ".venvs" / "07-risk-rule-engine" / "Scripts" / "python.exe"
        if not python.is_file():
            raise ApiError("Risk-rule virtual environment is unavailable. Run the capability setup first.")
        command = [
            str(python), str(self.root / "capabilities" / "07-risk-rule-engine" / "run_risk_rule_engine.py"),
            "--observations", str(processed_root / observations_name), "--zones", str(processed_root / zones_name),
            "--rules", str(processed_root / rules_name), "--output", str(output),
        ]
        with RUN_LOCK:
            self.run_command(command, 600)
        metadata_path = output / "run_metadata.json"
        if not metadata_path.is_file():
            raise ApiError("Risk-rule script finished without the expected result metadata.")
        metadata = load_json(metadata_path)
        metadata["input_dir"] = relative_path(self.root, processed_root)
        metadata["raw_input_dir"] = relative_path(self.root, raw_root)
        metadata["source_bytes"] = {name: len(content) for name, content in staged}
        metadata_path.write_text(json.dumps(metadata, ensure_ascii=False, indent=2), encoding="utf-8")
        return next(item for item in risk_rule_runs(self.root) if item["id"] == run_id)

    def send_json(self, status: HTTPStatus, payload: dict[str, Any]) -> None:
        body = json.dumps(payload, ensure_ascii=False).encode("utf-8")
        self.send_response(status)
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def translate_path(self, path: str) -> str:
        """Expose only the static UI and artifacts required by the local console."""
        decoded_path = unquote(urlsplit(path).path).lstrip("/")
        requested = PurePosixPath(decoded_path)
        is_allowed = any(decoded_path == prefix or decoded_path.startswith(f"{prefix}/") for prefix in ALLOWED_PATH_PREFIXES)
        if ".." in requested.parts or not is_allowed:
            return os.fspath(Path(self.directory) / ".console-forbidden")
        if decoded_path == "apps/workbench-console" or decoded_path.startswith("apps/workbench-console/"):
            console_relative = requested.parts[2:]
            return os.fspath(Path(self.directory) / "apps" / "workbench-console" / "dist" / Path(*console_relative))
        return os.fspath(Path(self.directory).joinpath(*requested.parts))

    def end_headers(self) -> None:
        self.send_header("Cache-Control", "no-store")
        self.send_header("X-Content-Type-Options", "nosniff")
        super().end_headers()


def parse_args() -> argparse.Namespace:
    root = Path(__file__).resolve().parents[1]
    parser = argparse.ArgumentParser(description="Serve the local GeoAI Workbench console.")
    parser.add_argument("--host", default=DEFAULT_HOST, help="Bind address. Defaults to loopback only.")
    parser.add_argument("--port", type=int, default=DEFAULT_PORT, help="TCP port in the 6000-6999 range.")
    parser.add_argument("--root", type=Path, default=root, help="Workbench repository root to serve.")
    return parser.parse_args()


def main() -> int:
    args = parse_args()
    if not 6000 <= args.port <= 6999:
        raise SystemExit("Port must be in the 6000-6999 range.")
    if args.host not in {"127.0.0.1", "localhost", "::1"}:
        raise SystemExit("This console is local-only. Use 127.0.0.1, localhost, or ::1.")
    root = args.root.resolve()
    app_dir = root / "apps" / "workbench-console"
    if not (root / "shared").is_dir() or not (app_dir / "dist" / "index.html").is_file():
        raise SystemExit(f"Not a GeoAI Workbench root: {root}")
    handler = lambda *handler_args, **handler_kwargs: WorkbenchConsoleHandler(*handler_args, directory=os.fspath(root), **handler_kwargs)  # noqa: E731
    server = ThreadingHTTPServer((args.host, args.port), handler)
    print(f"GeoAI Workbench console: http://{args.host}:{args.port}")
    print("Local runs use fixed capability scripts and create a new run directory.")
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        print("\nConsole stopped.")
    finally:
        server.server_close()
    return 0


if __name__ == "__main__":
    raise SystemExit(main())