geoai/geoai-workbench.git

			@@ -1,44 +1,1173 @@
			"""Serve the local GeoAI Workbench console from the repository root."""
			"""Serve the local GeoAI Workbench console and its narrow local-run APIs."""

			from __future__ import annotations

			import argparse
			import base64
			import binascii
			import hashlib
			import json
			import os
			from pathlib import PurePosixPath
			import re
			import shutil
			import subprocess
			import threading
			from datetime import UTC, datetime
			from http import HTTPStatus
			from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer
			from pathlib import Path
			from urllib.parse import unquote, urlsplit
			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_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",
			)
			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]] = {}


			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:
			return path.relative_to(root).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 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"
			records.append(
			{
			"id": run_id,
			"label": "既有基线结果" if run_id == "baseline" else run_id,
			"note": "CPU 基线：人员与常见车辆；树木不在当前模型有效类别内。",
			"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
			run_id = artifact.name
			records.append(
			{
			"id": run_id,
			"label": run_id,
			"note": "ChangeStar CPU 变化栅格与 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 ""),
			}
			)
			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 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)


			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 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):
			"""Read-only static handler rooted at the workbench repository."""
			"""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
			if urlsplit(self.path).path == "/":
			path = urlsplit(self.path).path
			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/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/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_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/"):
			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)
			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, 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"}, "change-detection")

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

			def receive_binary_upload(
			self,
			path: str,
			capability: str,
			allowed_roles: 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, {".jpg", ".jpeg", ".png", ".tif", ".tiff"})
			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_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")
			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.")
			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
			python = self.root / ".venvs" / "01-object-detection" / "Scripts" / "python.exe"
			if not python.is_file():
			raise ApiError("Object-detection virtual environment is unavailable. Run the capability setup first.")
			with RUN_LOCK:
			self.run_command([str(python), str(self.root / "capabilities" / "01-object-detection" / "run_detection.py"), "--input", str(raw_root), "--output", str(output)], 1200)
			if not (output / "run_metadata.json").is_file():
			raise ApiError("Detection script finished without the expected result metadata.")
			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")
			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 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
			python = self.root / ".venvs" / "00-change-detection" / "Scripts" / "python.exe"
			if not python.is_file():
			raise ApiError("Change-detection virtual environment is unavailable. Run the capability setup first.")
			with RUN_LOCK:
			self.run_command(
			[
			str(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,
			"--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"),
			("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 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),
			"artifacts": {
			"probability_raster": "change_probability.tif",
			"raw_mask_raster": "change_mask.tif",
			"mask_raster": "change_mask.tif",
			"overlay": "change_overlay.jpg",
			"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")
			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.")
			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,
			}
			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),
			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,
			) -> None:
			python = self.root / ".venvs" / "00-change-detection" / "Scripts" / "python.exe"
			try:
			if not python.is_file():
			raise ApiError("Change-detection virtual environment is unavailable. Run the capability setup first.")
			self._update_scan_job(run_id, status="running", phase="inference")
			with RUN_LOCK:
			self.run_command(
			[
			str(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,
			"--processed-output", str(processed_root),
			"--output", str(inference_output),
			],
			SCAN_JOB_TIMEOUT,
			)
			inference_metadata_path = inference_output / "run_metadata.json"
			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 = [
			str(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 = [
			str(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,
			"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 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
			)
			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/"):
			@@ -71,13 +1200,10 @@
			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( # noqa: E731
			handler_args, directory=os.fspath(root), *handler_kwargs
			)
			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(f"Serving built console and read-only artifacts from: {root}")
			print("Local runs use fixed capability scripts and create a new run directory.")
			try:
			server.serve_forever()
			except KeyboardInterrupt: