feat(pointcloud): complete annotation and result lifecycle workflows

shuishen

22 hours ago 385be2eca72eb3833efa4be0a0088b34e764788a

 tests/test_serve_workbench_console.py

@@ -1,7 +1,17 @@
from __future__ import annotations

import importlib.util
import io
import subprocess
import sys
import tempfile
import unittest
import json
import struct
import zipfile
from unittest import mock
from http import HTTPStatus
from urllib.parse import quote
from pathlib import Path


@@ -13,7 +23,64 @@
SPEC.loader.exec_module(MODULE)


def write_test_npz(path: Path, point_count: int) -> None:
    """Create just enough valid NPY headers for the console's dependency-free validator."""
    def npy_header(shape: tuple[int, int]) -> bytes:
        text = repr({"descr": "<f8", "fortran_order": False, "shape": shape}).encode("ascii")
        padding = (-((10 + len(text) + 1) % 16)) % 16
        header = text + (b" " * padding) + b"\n"
        return b"\x93NUMPY\x01\x00" + struct.pack("<H", len(header)) + header

    with zipfile.ZipFile(path, "w", compression=zipfile.ZIP_DEFLATED) as archive:
        archive.writestr("xyz.npy", npy_header((point_count, 3)))
        archive.writestr("las_rgb.npy", npy_header((point_count, 3)))


class WorkbenchConsoleHandlerTests(unittest.TestCase):
    def test_run_deletion_plan_rejects_unknown_capability_and_path_traversal(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            with self.assertRaises(MODULE.ApiError):
                MODULE.run_deletion_plan(root, "not-a-capability", "run-1")
            with self.assertRaises(MODULE.ApiError):
                MODULE.run_deletion_plan(root, "02-semantic-mapping", "../run-1")

    def test_run_deletion_plan_protects_non_console_baseline(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            artifact = root / "shared" / "outputs" / "02-semantic-mapping" / "validation-baseline"
            artifact.mkdir(parents=True)
            (artifact / "run_metadata.json").write_text(json.dumps({"capability": "02-semantic-mapping", "images": []}), encoding="utf-8")
            plan = MODULE.run_deletion_plan(root, "02-semantic-mapping", "validation-baseline")
            self.assertFalse(plan["removable"])
            self.assertEqual(plan["outputDirectories"], [])

    def test_console_run_deletion_removes_only_discovered_owned_directories(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            run_id = "semantic-test-run"
            output = root / "shared" / "outputs" / "02-semantic-mapping" / "runs" / run_id
            raw = root / "shared" / "data" / "raw" / "02-semantic-mapping" / "runs" / run_id
            processed = root / "shared" / "data" / "processed" / "02-semantic-mapping" / run_id
            sibling = root / "shared" / "outputs" / "02-semantic-mapping" / "runs" / "keep-me"
            external = root / "baseData" / "keep.jpg"
            for directory in (output, raw, processed, sibling, external.parent):
                directory.mkdir(parents=True, exist_ok=True)
            (output / "run_metadata.json").write_text(json.dumps({"capability": "02-semantic-mapping", "images": [], "created_at": "2026-08-28T00:00:00Z"}), encoding="utf-8")
            external.write_bytes(b"external")
            plan = MODULE.run_deletion_plan(root, "02-semantic-mapping", run_id)
            self.assertTrue(plan["removable"])
            self.assertEqual(plan["outputDirectories"], [f"shared/outputs/02-semantic-mapping/runs/{run_id}"])
            self.assertEqual(plan["rawDirectories"], [f"shared/data/raw/02-semantic-mapping/runs/{run_id}"])
            self.assertEqual(plan["processedDirectories"], [f"shared/data/processed/02-semantic-mapping/{run_id}"])
            removed = MODULE.delete_console_run(root, "02-semantic-mapping", run_id)
            self.assertEqual(removed["runId"], run_id)
            self.assertFalse(output.exists())
            self.assertFalse(raw.exists())
            self.assertFalse(processed.exists())
            self.assertTrue(sibling.is_dir())
            self.assertEqual(external.read_bytes(), b"external")

    def make_handler(self) -> MODULE.WorkbenchConsoleHandler:
        handler = object.__new__(MODULE.WorkbenchConsoleHandler)
        handler.directory = str(ROOT)
@@ -33,6 +100,598 @@
            Path(handler.translate_path("/shared/data/raw/01-object-detection/sample.jpeg")),
            ROOT / "shared" / "data" / "raw" / "01-object-detection" / "sample.jpeg",
        )
        self.assertEqual(
            Path(handler.translate_path("/shared/data/raw/02-semantic-mapping/sample.tif")),
            ROOT / "shared" / "data" / "raw" / "02-semantic-mapping" / "sample.tif",
        )
        self.assertEqual(
            Path(handler.translate_path("/shared/data/raw/00-change-detection/sample.jpg")),
            ROOT / "shared" / "data" / "raw" / "00-change-detection" / "sample.jpg",
        )
        self.assertEqual(
            Path(handler.translate_path("/shared/data/raw/09-anomaly-detection/sample.tif")),
            ROOT / "shared" / "data" / "raw" / "09-anomaly-detection" / "sample.tif",
        )
        self.assertEqual(
            Path(handler.translate_path("/shared/data/processed/09-anomaly-detection/sample.tif")),
            ROOT / ".console-forbidden",
        )

    def test_upload_name_is_sanitized_and_extension_is_allowlisted(self) -> None:
        self.assertEqual(MODULE.safe_file_name("../../unsafe name.JPG", {".jpg"}), "unsafe_name.jpg")
        self.assertEqual(MODULE.safe_file_name("../正常参考图.JPG", {".jpg"}), "正常参考图.jpg")
        with self.assertRaises(MODULE.ApiError):
            MODULE.safe_file_name("image.exe", {".jpg", ".png"})

    def test_oversized_request_is_rejected_before_reading_body(self) -> None:
        handler = self.make_handler()
        handler.headers = {"Content-Length": str(MODULE.MAX_REQUEST_BYTES + 1), "Content-Type": "application/json"}
        handler.rfile = io.BytesIO(b"")
        with self.assertRaises(MODULE.ApiError):
            handler.read_json_body()

    def test_large_tiff_upload_limits_allow_one_gibibyte_files(self) -> None:
        self.assertEqual(MODULE.MAX_FILE_BYTES, 1024 * 1024 * 1024)
        self.assertEqual(MODULE.MAX_REQUEST_BYTES, 3072 * 1024 * 1024)

    def test_photo_reconstruction_timeout_supports_long_cpu_runs(self) -> None:
        self.assertEqual(MODULE.PHOTO_RECONSTRUCTION_TIMEOUT, 86_400)

    def test_binary_change_upload_preserves_original_bytes(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            handler = object.__new__(MODULE.WorkbenchConsoleHandler)
            handler.directory = temp_dir
            handler.path = "/api/change-detection/uploads/0123456789abcdef0123456789abcdef?role=before"
            handler.headers = {"Content-Length": "13", "X-Upload-Name": "../1.tif"}
            handler.rfile = io.BytesIO(b"raw-tif-bytes")
            result = handler.receive_change_upload("/api/change-detection/uploads/0123456789abcdef0123456789abcdef")
            staged = Path(temp_dir) / "shared" / "data" / "raw" / "00-change-detection" / "uploads" / result["uploadId"] / "before.tif"
            self.assertEqual(staged.read_bytes(), b"raw-tif-bytes")
            self.assertEqual(result["name"], "1.tif")

    def test_binary_anomaly_upload_preserves_bytes_and_checksum(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            handler = object.__new__(MODULE.WorkbenchConsoleHandler)
            handler.directory = temp_dir
            handler.path = "/api/anomaly-detection/uploads/1123456789abcdef0123456789abcdef?role=reference"
            handler.headers = {"Content-Length": "15", "X-Upload-Name": "../normal.JPG"}
            handler.rfile = io.BytesIO(b"reference-bytes")
            result = handler.receive_anomaly_upload("/api/anomaly-detection/uploads/1123456789abcdef0123456789abcdef")
            staged = Path(temp_dir) / "shared" / "data" / "raw" / "09-anomaly-detection" / "uploads" / result["uploadId"] / "reference.jpg"
            self.assertEqual(staged.read_bytes(), b"reference-bytes")
            self.assertEqual(result["sha256"], MODULE.file_sha256(staged))

    def test_binary_photo_reconstruction_upload_preserves_jpeg_bytes(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            handler = object.__new__(MODULE.WorkbenchConsoleHandler)
            handler.directory = temp_dir
            handler.path = "/api/3d-pointcloud/photo-uploads/4123456789abcdef0123456789abcdef?role=photo"
            handler.headers = {"Content-Length": "10", "X-Upload-Name": "../flight.JPG"}
            handler.rfile = io.BytesIO(b"jpeg-bytes")
            result = handler.receive_photo_reconstruction_upload("/api/3d-pointcloud/photo-uploads/4123456789abcdef0123456789abcdef")
            staged = Path(temp_dir) / "shared" / "data" / "raw" / "05-3d-pointcloud" / "uploads" / result["uploadId"] / "photo.jpg"
            self.assertEqual(result["role"], "photo")
            self.assertEqual(result["name"], "flight.jpg")
            self.assertEqual(staged.read_bytes(), b"jpeg-bytes")

    def test_binary_pointcloud_upload_preserves_las_bytes(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            handler = object.__new__(MODULE.WorkbenchConsoleHandler)
            handler.directory = temp_dir
            handler.path = "/api/3d-pointcloud/pointcloud-uploads/5123456789abcdef0123456789abcdef?role=pointcloud"
            handler.headers = {"Content-Length": "9", "X-Upload-Name": "../corridor.LAS"}
            handler.rfile = io.BytesIO(b"las-bytes")
            result = handler.receive_pointcloud_upload("/api/3d-pointcloud/pointcloud-uploads/5123456789abcdef0123456789abcdef")
            staged = Path(temp_dir) / "shared" / "data" / "raw" / "05-3d-pointcloud" / "uploads" / result["uploadId"] / "pointcloud.las"
            self.assertEqual(result["role"], "pointcloud")
            self.assertEqual(result["name"], "corridor.las")
            self.assertEqual(staged.read_bytes(), b"las-bytes")

    def test_model_inference_rejects_empty_and_incomplete_las_uploads(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            empty = root / "empty.ply"
            empty.write_bytes(b"")
            with self.assertRaisesRegex(MODULE.ApiError, "文件为空"):
                MODULE.validate_pointcloud_model_input(empty)

            incomplete = root / "tile_000_000.las"
            incomplete.write_bytes(b"las-bytes")
            with self.assertRaisesRegex(MODULE.ApiError, "文件不完整"):
                MODULE.validate_pointcloud_model_input(incomplete)

            empty_las = root / "empty.las"
            header = bytearray(227)
            header[:4] = b"LASF"
            header[24:26] = bytes((1, 2))
            header[94:96] = (227).to_bytes(2, "little")
            empty_las.write_bytes(header)
            with self.assertRaisesRegex(MODULE.ApiError, "没有点记录"):
                MODULE.validate_pointcloud_model_input(empty_las)

    def test_photo_reconstruction_request_requires_three_to_one_thousand_photos(self) -> None:
        handler = self.make_handler()
        with self.assertRaisesRegex(MODULE.ApiError, "at least three"):
            handler.create_photo_reconstruction_run({"photos": []})
        with self.assertRaisesRegex(MODULE.ApiError, "at most"):
            handler.create_photo_reconstruction_run({"photos": [{}] * (MODULE.MAX_PHOTO_RECONSTRUCTION_IMAGES_PER_RUN + 1)})
        with self.assertRaisesRegex(MODULE.ApiError, "true or false"):
            handler.create_photo_reconstruction_run({"photos": [{}, {}, {}], "usePositionPriors": "yes"})

    def test_photo_reconstruction_job_reads_progress_without_exposing_path(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            progress_path = Path(temp_dir) / "progress.json"
            progress_path.write_text(json.dumps({"percent": 68, "stage": "dense_fusion", "message": "running", "inputImages": 18, "estimate": True}), encoding="utf-8")
            with MODULE.PHOTO_RECONSTRUCTION_JOBS_LOCK:
                MODULE.PHOTO_RECONSTRUCTION_JOBS["progress-test"] = {"id": "progress-test", "runId": "run", "status": "running", "stage": "dense_mvs", "inputImages": 18, "progressPath": str(progress_path)}
            try:
                job = MODULE.photo_reconstruction_job("progress-test")
            finally:
                with MODULE.PHOTO_RECONSTRUCTION_JOBS_LOCK:
                    MODULE.PHOTO_RECONSTRUCTION_JOBS.pop("progress-test", None)
            self.assertEqual(job["progress"]["percent"], 68)
            self.assertNotIn("progressPath", job)

    def test_binary_anomaly_upload_decodes_chinese_file_name(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            handler = object.__new__(MODULE.WorkbenchConsoleHandler)
            handler.directory = temp_dir
            handler.path = "/api/anomaly-detection/uploads/3123456789abcdef0123456789abcdef?role=input"
            handler.headers = {"Content-Length": "11", "X-Upload-Name": quote("异常测试图.JPG", safe="")}
            handler.rfile = io.BytesIO(b"image-bytes")
            result = handler.receive_anomaly_upload("/api/anomaly-detection/uploads/3123456789abcdef0123456789abcdef")
            staged = Path(temp_dir) / "shared" / "data" / "raw" / "09-anomaly-detection" / "uploads" / result["uploadId"] / "input.jpg"
            self.assertEqual(result["name"], "异常测试图.jpg")
            self.assertEqual(staged.read_bytes(), b"image-bytes")

    def test_interrupted_anomaly_upload_removes_partial_file(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            handler = object.__new__(MODULE.WorkbenchConsoleHandler)
            handler.directory = temp_dir
            handler.path = "/api/anomaly-detection/uploads/2123456789abcdef0123456789abcdef?role=input"
            handler.headers = {"Content-Length": "20", "X-Upload-Name": "target.tif"}
            handler.rfile = io.BytesIO(b"short")
            with self.assertRaisesRegex(MODULE.ApiError, "ended before"):
                handler.receive_anomaly_upload("/api/anomaly-detection/uploads/2123456789abcdef0123456789abcdef")
            staging = Path(temp_dir) / "shared" / "data" / "raw" / "09-anomaly-detection" / "uploads" / "2123456789abcdef0123456789abcdef"
            self.assertFalse(any(staging.glob("*.part")))

    def test_anomaly_parameters_are_bounded(self) -> None:
        self.assertEqual(MODULE.validate_anomaly_parameters({}), (256, 128, 0.995, 42))
        with self.assertRaisesRegex(MODULE.ApiError, "Stride"):
            MODULE.validate_anomaly_parameters({"tileSize": 128, "stride": 256})
        with self.assertRaisesRegex(MODULE.ApiError, "quantile"):
            MODULE.validate_anomaly_parameters({"thresholdQuantile": 1.0})

    def test_anomaly_run_discovery_requires_exposed_raw_sources(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            artifact = root / "shared" / "outputs" / "09-anomaly-detection" / "runs" / "anomaly-test"
            raw_input = root / "shared" / "data" / "raw" / "09-anomaly-detection" / "runs" / "anomaly-test" / "input"
            raw_reference = raw_input.parent / "reference"
            artifact.mkdir(parents=True)
            raw_input.mkdir(parents=True)
            raw_reference.mkdir(parents=True)
            (artifact / "target.comparison.overlay.png").write_bytes(b"png")
            metadata = {"capability": "09-anomaly-detection", "created_at": "2026-08-18", "display_name": "五参考图边界案例", "case_note": "保留真实漏检结果。", "raw_input_dir": raw_input.relative_to(root).as_posix(), "raw_reference_dir": raw_reference.relative_to(root).as_posix(), "images": [{"overlay_file": "target.comparison.overlay.png"}]}
            (artifact / "run_metadata.json").write_text(json.dumps(metadata), encoding="utf-8")
            runs = MODULE.anomaly_runs(root)
            self.assertEqual(runs[0]["id"], "anomaly-test")
            self.assertEqual(runs[0]["label"], "五参考图边界案例")
            self.assertEqual(runs[0]["note"], "保留真实漏检结果。")

    def test_script_failure_becomes_a_useful_api_error(self) -> None:
        handler = self.make_handler()
        with self.assertRaisesRegex(MODULE.ApiError, "Processing failed"):
            handler.run_command([sys.executable, "-c", "raise SystemExit(2)"], timeout=10)

    def test_pointcloud_execution_uses_fixed_gpu_environment_when_cuda_probe_succeeds(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            gpu_python = root / ".venvs" / MODULE.POINTCLOUD_GPU_ENVIRONMENT / "Scripts" / "python.exe"
            gpu_python.parent.mkdir(parents=True)
            gpu_python.write_bytes(b"fixed-interpreter")
            with mock.patch.object(MODULE.subprocess, "run", return_value=subprocess.CompletedProcess([], 0, '{"cuda": true, "torch": "2.11.0+cu128"}\n', "")):
                execution = MODULE.pointcloud_execution_environment(root)
            self.assertEqual(execution["device"], "cuda")
            self.assertEqual(execution["environment"], MODULE.POINTCLOUD_GPU_ENVIRONMENT)
            self.assertEqual(execution["torchVersion"], "2.11.0+cu128")
            self.assertEqual(Path(execution["python"]), gpu_python)

    def test_pointcloud_execution_falls_back_to_fixed_cpu_environment(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            cpu_python = root / ".venvs" / MODULE.POINTCLOUD_CPU_ENVIRONMENT / "Scripts" / "python.exe"
            gpu_python = root / ".venvs" / MODULE.POINTCLOUD_GPU_ENVIRONMENT / "Scripts" / "python.exe"
            cpu_python.parent.mkdir(parents=True)
            gpu_python.parent.mkdir(parents=True)
            cpu_python.write_bytes(b"fixed-cpu-interpreter")
            gpu_python.write_bytes(b"fixed-gpu-interpreter")
            with mock.patch.object(MODULE.subprocess, "run", return_value=subprocess.CompletedProcess([], 0, '{"cuda": false, "torch": "2.11.0+cu128"}\n', "")):
                execution = MODULE.pointcloud_execution_environment(root)
            self.assertEqual(execution["device"], "cpu")
            self.assertEqual(execution["environment"], MODULE.POINTCLOUD_CPU_ENVIRONMENT)
            with mock.patch.object(MODULE.subprocess, "run", return_value=subprocess.CompletedProcess([], 0, '{"cuda": false, "torch": "2.11.0+cu128"}\n', "")):
                with self.assertRaisesRegex(MODULE.ApiError, "CUDA was requested"):
                    MODULE.pointcloud_execution_environment(root, "cuda")

    def test_explicit_cpu_execution_never_probes_cuda(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            cpu_python = root / ".venvs" / MODULE.OBJECT_DETECTION_CPU_ENVIRONMENT / "Scripts" / "python.exe"
            cpu_python.parent.mkdir(parents=True)
            cpu_python.write_bytes(b"fixed-cpu-interpreter")
            with mock.patch.object(MODULE.subprocess, "run") as probe:
                execution = MODULE.object_detection_execution_environment(root, "cpu")
            probe.assert_not_called()
            self.assertEqual(execution["requestedDevice"], "cpu")
            self.assertEqual(execution["device"], "cpu")
            self.assertFalse(execution["fallbackUsed"])

    def test_auto_cpu_fallback_records_the_probe_reason(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            cpu_python = root / ".venvs" / MODULE.CHANGE_DETECTION_CPU_ENVIRONMENT / "Scripts" / "python.exe"
            gpu_python = root / ".venvs" / MODULE.CHANGE_DETECTION_GPU_ENVIRONMENT / "Scripts" / "python.exe"
            cpu_python.parent.mkdir(parents=True)
            gpu_python.parent.mkdir(parents=True)
            cpu_python.write_bytes(b"fixed-cpu-interpreter")
            gpu_python.write_bytes(b"fixed-gpu-interpreter")
            with mock.patch.object(MODULE.subprocess, "run", return_value=subprocess.CompletedProcess([], 0, '{"cuda": false, "torch": "2.11.0+cu128"}\n', "")):
                execution = MODULE.change_detection_execution_environment(root, "auto")
            self.assertTrue(execution["fallbackUsed"])
            self.assertIn("cannot use CUDA", execution["fallbackReason"])

    def test_object_detection_execution_uses_fixed_gpu_environment_when_probe_succeeds(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            gpu_python = root / ".venvs" / MODULE.OBJECT_DETECTION_GPU_ENVIRONMENT / "Scripts" / "python.exe"
            gpu_python.parent.mkdir(parents=True)
            gpu_python.write_bytes(b"fixed-interpreter")
            with mock.patch.object(MODULE.subprocess, "run", return_value=subprocess.CompletedProcess([], 0, '{"cuda": true, "torch": "2.11.0+cu128"}\n', "")):
                execution = MODULE.object_detection_execution_environment(root)
            self.assertEqual(execution["device"], "cuda")
            self.assertEqual(execution["environment"], MODULE.OBJECT_DETECTION_GPU_ENVIRONMENT)

    def test_run_discovery_uses_actual_device_in_case_note(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            detection = root / "shared" / "outputs" / "01-object-detection" / "gpu-case"
            raw = root / "shared" / "data" / "raw" / "01-object-detection" / "gpu-case"
            detection.mkdir(parents=True)
            raw.mkdir(parents=True)
            (detection / "detections.json").write_text("{}", encoding="utf-8")
            (detection / "run_metadata.json").write_text(json.dumps({"input_dir": str(raw), "created_at": "2026-08-25", "device": "cuda:0"}), encoding="utf-8")
            self.assertIn("GPU", MODULE.detection_runs(root)[0]["note"])

            change = root / "shared" / "outputs" / "00-change-detection" / "gpu-case"
            change_raw = root / "shared" / "data" / "raw" / "00-change-detection" / "gpu-case"
            before = change_raw / "before.jpg"
            after = change_raw / "after.jpg"
            change.mkdir(parents=True)
            change_raw.mkdir(parents=True)
            before.write_bytes(b"before")
            after.write_bytes(b"after")
            (change / "overlay.jpg").write_bytes(b"overlay")
            (change / "changes.geojson").write_text("{}", encoding="utf-8")
            (change / "run_metadata.json").write_text(json.dumps({"capability": "00-change-detection", "schema_version": 1, "created_at": "2026-08-25", "device": "cuda", "input_files": ["before.jpg", "after.jpg"], "raw_input_dir": change_raw.relative_to(root).as_posix(), "raw_before": before.relative_to(root).as_posix(), "raw_after": after.relative_to(root).as_posix(), "artifacts": {"overlay": "overlay.jpg", "vector": "changes.geojson"}}), encoding="utf-8")
            self.assertIn("GPU", MODULE.change_runs(root)[0]["note"])

    def test_object_detection_execution_falls_back_to_fixed_cpu_environment(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            cpu_python = root / ".venvs" / MODULE.OBJECT_DETECTION_CPU_ENVIRONMENT / "Scripts" / "python.exe"
            gpu_python = root / ".venvs" / MODULE.OBJECT_DETECTION_GPU_ENVIRONMENT / "Scripts" / "python.exe"
            cpu_python.parent.mkdir(parents=True)
            gpu_python.parent.mkdir(parents=True)
            cpu_python.write_bytes(b"fixed-cpu-interpreter")
            gpu_python.write_bytes(b"fixed-gpu-interpreter")
            with mock.patch.object(MODULE.subprocess, "run", return_value=subprocess.CompletedProcess([], 0, '{"cuda": false, "torch": "2.11.0+cu128"}\n', "")):
                execution = MODULE.object_detection_execution_environment(root)
            self.assertEqual(execution["device"], "cpu")
            self.assertEqual(execution["environment"], MODULE.OBJECT_DETECTION_CPU_ENVIRONMENT)

    def test_change_detection_execution_uses_fixed_gpu_environment_when_probe_succeeds(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            gpu_python = root / ".venvs" / MODULE.CHANGE_DETECTION_GPU_ENVIRONMENT / "Scripts" / "python.exe"
            gpu_python.parent.mkdir(parents=True)
            gpu_python.write_bytes(b"fixed-interpreter")
            with mock.patch.object(MODULE.subprocess, "run", return_value=subprocess.CompletedProcess([], 0, '{"cuda": true, "torch": "2.11.0+cu128"}\n', "")):
                execution = MODULE.change_detection_execution_environment(root)
            self.assertEqual(execution["device"], "cuda")
            self.assertEqual(execution["environment"], MODULE.CHANGE_DETECTION_GPU_ENVIRONMENT)

    def test_change_detection_execution_falls_back_to_fixed_cpu_environment(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            cpu_python = root / ".venvs" / MODULE.CHANGE_DETECTION_CPU_ENVIRONMENT / "Scripts" / "python.exe"
            gpu_python = root / ".venvs" / MODULE.CHANGE_DETECTION_GPU_ENVIRONMENT / "Scripts" / "python.exe"
            cpu_python.parent.mkdir(parents=True)
            gpu_python.parent.mkdir(parents=True)
            cpu_python.write_bytes(b"fixed-cpu-interpreter")
            gpu_python.write_bytes(b"fixed-gpu-interpreter")
            with mock.patch.object(MODULE.subprocess, "run", return_value=subprocess.CompletedProcess([], 0, '{"cuda": false, "torch": "2.11.0+cu128"}\n', "")):
                execution = MODULE.change_detection_execution_environment(root)
            self.assertEqual(execution["device"], "cpu")
            self.assertEqual(execution["environment"], MODULE.CHANGE_DETECTION_CPU_ENVIRONMENT)

    def test_semantic_validation_run_is_discovered(self) -> None:
        runs = MODULE.semantic_runs(ROOT)
        self.assertTrue(any(item["id"] == "validation-20260817" for item in runs))

    def test_spatial_measurement_run_is_discovered(self) -> None:
        runs = MODULE.measurement_runs(ROOT)
        validation = next(item for item in runs if item["id"] == "validation-normal-20260817-v4")
        self.assertTrue(validation["artifactRoot"].startswith("shared/outputs/04-spatial-measurement/"))

    def test_pointcloud_validation_run_is_discovered(self) -> None:
        runs = MODULE.pointcloud_runs(ROOT)
        validation = next(item for item in runs if item["id"] == "validation-normal-20260821-v2")
        self.assertTrue(validation["artifactRoot"].startswith("shared/outputs/05-3d-pointcloud/"))

    def test_multiview_annotation_source_requires_complete_ordered_contract(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            artifact = root / "shared" / "outputs" / "05-3d-pointcloud" / "multiview-feature-valid"
            artifact.mkdir(parents=True)
            cloud = artifact / "multiview-annotation-source.ply"
            cloud.write_text(
                "ply\nformat ascii 1.0\nelement vertex 2\nproperty float x\nproperty float y\nproperty float z\nend_header\n0 0 0\n1 1 1\n",
                encoding="ascii",
            )
            dataset = artifact / "multiview-point-features.npz"
            write_test_npz(dataset, 2)
            (artifact / "run_metadata.json").write_text(json.dumps({
                "capability": "05-3d-pointcloud",
                "artifacts": {"annotation_source": cloud.name, "feature_dataset": dataset.name},
                "annotation_source": {
                    "schema_version": 1,
                    "kind": "multiview_photo_feature_fusion",
                    "point_cloud": cloud.name,
                    "feature_dataset": dataset.name,
                    "point_count": 2,
                    "point_cloud_sha256": MODULE.file_sha256(cloud),
                    "feature_dataset_sha256": MODULE.file_sha256(dataset),
                },
            }), encoding="utf-8")

            sources = MODULE.pointcloud_annotation_sources(root)
            self.assertEqual(len(sources), 1)
            self.assertEqual(sources[0]["pointCount"], 2)
            self.assertIn("多视角照片特征融合", sources[0]["sourceKind"])

            handler = object.__new__(MODULE.WorkbenchConsoleHandler)
            handler.directory = str(root)
            annotation = handler.create_pointcloud_annotation({"sourceId": sources[0]["id"], "labels": [[1, 15]]})
            self.assertEqual(annotation["labelCount"], 1)
            execution = {"python": "fixed-python", "device": "cpu", "environment": "05-3d-pointcloud", "torchVersion": "2.13.0+cpu", "requestedDevice": "cpu", "fallbackUsed": False, "fallbackReason": None}
            with mock.patch.object(MODULE, "pointcloud_execution_environment", return_value=execution), mock.patch.object(MODULE.threading, "Thread") as thread:
                thread.return_value.start.return_value = None
                job = handler.create_pointcloud_training_run({"annotationId": annotation["id"], "device": "cpu"})
            self.assertEqual(job["trainer"], "multiview_local_attention_baseline")
            self.assertTrue(job["artifactRoot"] if "artifactRoot" in job else True)
            self.assertEqual(thread.call_args.kwargs["args"][-1], "multiview_local_attention_baseline")

            dataset.unlink()
            self.assertEqual(MODULE.pointcloud_annotation_sources(root), [])

    def test_pointcloud_request_requires_allowlisted_files(self) -> None:
        handler = self.make_handler()
        with self.assertRaisesRegex(MODULE.ApiError, "PLY, PCD, XYZ, LAS, or LAZ"):
            handler.create_pointcloud_run({"pointClouds": []})
        with self.assertRaisesRegex(MODULE.ApiError, "Unsupported file type"):
            handler.create_pointcloud_run({"pointClouds": [{"name": "unsafe.exe", "content": "eA=="}]})

    def test_annotation_taxonomy_adds_custom_las_class_and_preserves_used_code(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            handler = object.__new__(MODULE.WorkbenchConsoleHandler)
            handler.directory = str(root)
            defaults = MODULE.annotation_classes(root)
            self.assertEqual({item["code"] for item in defaults}, {1, 2, 5, 6, 15, 16})
            created = handler.create_pointcloud_annotation_class({"key": "transformer", "label": "变压器", "color": [30, 144, 255]})
            self.assertEqual(created["code"], 17)
            self.assertTrue(MODULE.annotation_classes_path(root).is_file())
            revision = root / "shared" / "outputs" / "05-3d-pointcloud" / "annotations" / "annotation-used"
            revision.mkdir(parents=True)
            (revision / "annotation.json").write_text(json.dumps({"schema_version": 1, "labels": [[0, 17]]}), encoding="utf-8")
            with self.assertRaisesRegex(MODULE.ApiError, "used by a saved annotation"):
                handler.delete_pointcloud_annotation_class(17)
            (revision / "annotation.json").unlink()
            self.assertEqual(handler.delete_pointcloud_annotation_class(17), 17)
            self.assertNotIn(17, {item["code"] for item in MODULE.annotation_classes(root)})

    def test_annotation_taxonomy_rejects_invalid_custom_metadata(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            handler = object.__new__(MODULE.WorkbenchConsoleHandler)
            handler.directory = temp_dir
            with self.assertRaisesRegex(MODULE.ApiError, "lowercase English"):
                handler.create_pointcloud_annotation_class({"key": "Transformer", "label": "变压器", "color": [30, 144, 255]})
            with self.assertRaisesRegex(MODULE.ApiError, "RGB"):
                handler.create_pointcloud_annotation_class({"key": "transformer", "label": "变压器", "color": [999, 1, 1]})

    def test_preview_only_annotation_source_is_discovered_without_geometry_artifacts(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            run_id = "annotation-source-preview-only"
            artifact = root / "shared" / "outputs" / "05-3d-pointcloud" / "runs" / run_id
            artifact.mkdir(parents=True)
            preview = artifact / "upload.annotation-source.ply"
            preview.write_text(
                "ply\nformat binary_little_endian 1.0\nelement vertex 2\nproperty float x\nproperty float y\nproperty float z\nproperty uchar red\nproperty uchar green\nproperty uchar blue\nend_header\n",
                encoding="ascii",
            )
            (artifact / "run_metadata.json").write_text(json.dumps({
                "capability": "05-3d-pointcloud", "annotation_source_job": True,
                "annotation_source": {"schema_version": 1, "file": preview.name, "point_count": 2, "sha256": MODULE.file_sha256(preview)},
                "input": {"file": "upload.ply", "has_rgb": False},
            }), encoding="utf-8")
            sources = MODULE.pointcloud_annotation_sources(root)
            self.assertEqual(len(sources), 1)
            self.assertEqual(sources[0]["id"], f"{run_id}:{preview.name}")
            self.assertFalse(sources[0]["sourceHasRgb"])
            self.assertEqual(MODULE.pointcloud_annotation_source_deletion_plan(root, sources[0]["id"])["outputDirectories"], 1)

    def test_texture_baked_annotation_source_is_discovered_at_capability_root(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            run_id = "texture-baked-preview"
            artifact = root / "shared" / "outputs" / "05-3d-pointcloud" / run_id
            artifact.mkdir(parents=True)
            preview = artifact / "texture-baked-annotation-source.ply"
            preview.write_text(
                "ply\nformat binary_little_endian 1.0\nelement vertex 1\nproperty float x\nproperty float y\nproperty float z\nend_header\n",
                encoding="ascii",
            )
            (artifact / "run_metadata.json").write_text(json.dumps({
                "capability": "05-3d-pointcloud", "annotation_source_job": True,
                "annotation_source": {"schema_version": 1, "file": preview.name, "point_count": 1, "sha256": MODULE.file_sha256(preview)},
                "input": {"file": "Model_0.zip", "has_rgb": True},
            }), encoding="utf-8")
            sources = MODULE.pointcloud_annotation_sources(root)
            self.assertEqual([item["id"] for item in sources], [f"{run_id}:{preview.name}"])
            self.assertTrue(sources[0]["sourceHasRgb"])

    def test_annotation_source_removal_deletes_only_its_complete_local_chain(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            run_id = "annotation-source-test"
            artifact = root / "shared" / "outputs" / "05-3d-pointcloud" / "runs" / run_id
            artifact.mkdir(parents=True)
            annotation_cloud = artifact / "block.semantic-annotation-source.ply"
            annotation_cloud.write_bytes(b"annotation-preview")
            (artifact / "preview.png").write_bytes(b"preview")
            (artifact / "vector.geojson").write_text('{"type":"FeatureCollection","features":[]}', encoding="utf-8")
            (artifact / "run_metadata.json").write_text(json.dumps({"capability": "05-3d-pointcloud", "created_at": "2026-08-28T00:00:00+00:00", "annotation_source_job": True, "point_clouds": [{"file": "block.las", "preview_file": "preview.png", "vector_file": "vector.geojson", "semantic_annotation_source_point_cloud": annotation_cloud.name, "semantic_preview_points": 1}]}), encoding="utf-8")
            source_id = f"{run_id}:{annotation_cloud.name}"
            revision = root / "shared" / "outputs" / "05-3d-pointcloud" / "annotations" / "annotation-test"
            revision.mkdir(parents=True)
            annotation_path = revision / "annotation.json"
            annotation_path.write_text(json.dumps({"schema_version": 1, "id": "annotation-test", "source_id": source_id, "labels": [[0, 5]]}), encoding="utf-8")
            training = root / "shared" / "outputs" / "05-3d-pointcloud" / "training-runs" / "model-test"
            training.mkdir(parents=True)
            model = training / "model.pt"
            model.write_bytes(b"weights")
            (training / "metrics.json").write_text(json.dumps({"annotation": str(annotation_path)}), encoding="utf-8")
            inference = root / "shared" / "outputs" / "05-3d-pointcloud" / "model-inference-runs" / "inference-test"
            inference.mkdir(parents=True)
            (inference / "run_metadata.json").write_text(json.dumps({"model": {"path": str(model)}}), encoding="utf-8")
            raw = root / "shared" / "data" / "raw" / "05-3d-pointcloud" / "annotation-source-runs" / run_id
            processed = root / "shared" / "data" / "processed" / "05-3d-pointcloud" / "annotation-source-runs" / run_id
            raw.mkdir(parents=True); processed.mkdir(parents=True)
            (raw / "block.las").write_bytes(b"raw"); (processed / "block.las").write_bytes(b"processed")
            external = root / "baseData" / "block.las"; external.parent.mkdir(); external.write_bytes(b"must-remain")

            handler = object.__new__(MODULE.WorkbenchConsoleHandler)
            handler.directory = str(root)
            plan = MODULE.pointcloud_annotation_source_deletion_plan(root, source_id)
            self.assertEqual((plan["rawDirectories"], plan["processedDirectories"], plan["annotationRevisions"], plan["trainingRuns"], plan["inferenceRuns"]), (1, 1, 1, 1, 1))
            result = handler.delete_pointcloud_annotation_source(source_id)
            self.assertEqual(result["removed"]["trainingRuns"], 1)
            self.assertFalse(artifact.exists()); self.assertFalse(raw.exists()); self.assertFalse(processed.exists())
            self.assertFalse(revision.exists()); self.assertFalse(training.exists()); self.assertFalse(inference.exists())
            self.assertEqual(external.read_bytes(), b"must-remain")

    def test_semantic_model_discovery_requires_complete_training_artifacts(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            complete = root / "shared" / "outputs" / "05-3d-pointcloud" / "training-runs" / "semantic-good"
            incomplete = complete.parent / "semantic-incomplete"
            complete.mkdir(parents=True)
            incomplete.mkdir(parents=True)
            (complete / "model.pt").write_bytes(b"weights")
            (complete / "metrics.json").write_text(json.dumps({"capability": "05-3d-pointcloud", "classification": "B", "created_at": "2026-08-24", "classes": {"5": {"key": "vegetation", "label": "Vegetation", "color": [1, 2, 3]}, "16": {"key": "power_line", "label": "Power line", "color": [4, 5, 6]}}, "test": {"report": {"vegetation": {"f1-score": 0.9}}}}), encoding="utf-8")
            (incomplete / "model.pt").write_bytes(b"weights")
            models = MODULE.pointcloud_semantic_models(root)
            self.assertEqual([item["id"] for item in models], ["semantic-good"])
            self.assertEqual(models[0]["testF1"], {"vegetation": 0.9})

    def test_model_inference_rejects_unsafe_or_unknown_model_id(self) -> None:
        handler = self.make_handler()
        with self.assertRaisesRegex(MODULE.ApiError, "Invalid trained model id"):
            handler.create_pointcloud_model_inference_run({"modelId": "../model", "pointCloud": {}})
        with self.assertRaisesRegex(MODULE.ApiError, "unavailable or incomplete"):
            handler.create_pointcloud_model_inference_run({"modelId": "does-not-exist", "pointCloud": {}})

    def test_annotation_delete_removes_only_selected_revision(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            annotation_id = "annotation-20260822-091458-b72099"
            revision = root / "shared" / "outputs" / "05-3d-pointcloud" / "annotations" / annotation_id
            revision.mkdir(parents=True)
            (revision / "annotation.json").write_text(json.dumps({
                "schema_version": 1,
                "id": annotation_id,
                "source_id": "source:preview.ply",
                "created_at": "2026-08-22T09:14:58+00:00",
                "labels": [[1, 15]],
            }), encoding="utf-8")
            source_sentinel = root / "shared" / "data" / "raw" / "05-3d-pointcloud" / "part_01.las"
            source_sentinel.parent.mkdir(parents=True)
            source_sentinel.write_bytes(b"source-must-remain")

            handler = object.__new__(MODULE.WorkbenchConsoleHandler)
            handler.directory = str(root)
            handler.path = f"/api/3d-pointcloud/annotations/{annotation_id}"
            responses: list[tuple[HTTPStatus, dict[str, object]]] = []
            handler.send_json = lambda status, body: responses.append((status, body))

            handler.do_DELETE()

            self.assertEqual(responses, [(HTTPStatus.OK, {"deletedId": annotation_id})])
            self.assertFalse(revision.exists())
            self.assertEqual(source_sentinel.read_bytes(), b"source-must-remain")

    def test_risk_rule_validation_run_is_discovered(self) -> None:
        runs = MODULE.risk_rule_runs(ROOT)
        validation = next(item for item in runs if item["id"] == "validation-normal-20260820")
        self.assertTrue(validation["artifactRoot"].startswith("shared/outputs/07-risk-rule-engine/"))

    def test_risk_rule_request_requires_fixed_three_file_contract(self) -> None:
        handler = self.make_handler()
        with self.assertRaisesRegex(MODULE.ApiError, "observations, zones and rules"):
            handler.create_risk_rule_run({"files": {"observations": {}}})

    def test_change_run_requires_both_allowlisted_images(self) -> None:
        handler = self.make_handler()
        with self.assertRaisesRegex(MODULE.ApiError, "name and Base64"):
            handler.create_change_run({"files": {"before": {"name": "before.jpg", "content": "eA=="}}})

    def test_change_threshold_is_validated(self) -> None:
        handler = self.make_handler()
        payload = {"files": {"before": {"name": "before.jpg", "content": "eA=="}, "after": {"name": "after.jpg", "content": "eA=="}}}
        with self.assertRaisesRegex(MODULE.ApiError, "between 0.01 and 0.99"):
            handler.create_change_run({**payload, "threshold": 1.0})
        with self.assertRaisesRegex(MODULE.ApiError, "must be a number"):
            handler.create_change_run({**payload, "threshold": "0.5"})

    def test_change_resolution_is_validated(self) -> None:
        handler = self.make_handler()
        payload = {"files": {"before": {"name": "before.jpg", "content": "eA=="}, "after": {"name": "after.jpg", "content": "eA=="}}}
        with self.assertRaisesRegex(MODULE.ApiError, "between 512 and 4096"):
            handler.create_change_run({**payload, "maxDimension": 256})
        with self.assertRaisesRegex(MODULE.ApiError, "must be an integer"):
            handler.create_change_run({**payload, "maxDimension": "2048"})

    def test_change_processing_mode_is_validated(self) -> None:
        handler = self.make_handler()
        payload = {"files": {"before": {"name": "before.jpg", "content": "eA=="}, "after": {"name": "after.jpg", "content": "eA=="}}}
        with self.assertRaisesRegex(MODULE.ApiError, "auto, image, or geotiff"):
            handler.create_change_run({**payload, "processingMode": "wrong"})

    def test_change_validation_run_is_discovered_without_fake_crs(self) -> None:
        runs = MODULE.change_runs(ROOT)
        self.assertTrue(any(item["id"].startswith("validation-real-") for item in runs))

    def test_semantic_task_catalog_only_enables_verified_baseline(self) -> None:
        tasks = MODULE.semantic_tasks(ROOT)
        self.assertEqual(len(tasks), 5)
        selectable = [item["id"] for item in tasks if item.get("selectable") is True]
        self.assertEqual(selectable, ["color_baseline"])
        handler = self.make_handler()
        with self.assertRaisesRegex(MODULE.ApiError, "not runnable yet"):
            handler.create_semantic_run({"taskId": "drainage_blockage", "images": []})

    def test_blocks_repository_files_and_encoded_traversal(self) -> None:
        handler = self.make_handler()