feat: add Gaussian simulation workspace

src/k1link/simulation/projects.py

@@ -0,0 +1,635 @@
+"""Durable Mission Core catalog and orchestration for portable Gaussian worlds."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import os
+import re
+import shutil
+import threading
+import time
+from contextlib import suppress
+from pathlib import Path
+from typing import Any, Final
+from urllib.parse import quote
+from uuid import uuid4
+
+from k1link.artifacts import utc_now_iso
+from k1link.simulation.gaussian_pipeline_gateway import (
+    BUILD_REQUEST_SCHEMA,
+    GaussianPipelineGateway,
+    GaussianPipelineGatewayError,
+    configured_gaussian_pipeline_gateway,
+    discover_gaussian_source_bundle,
+)
+
+PROJECT_SCHEMA: Final = "missioncore.simulation-project/v1"
+WORLD_MANIFEST_SCHEMA: Final = "missioncore.simulation-world-manifest/v1"
+PROJECT_ID_PATTERN: Final = re.compile(r"^sim-[a-f0-9]{32}$")
+SOURCE_FILE_ID_PATTERN: Final = re.compile(r"^src-[0-9]{5}-[a-f0-9]{8}$")
+PROVIDER_JOB_ID_PATTERN: Final = re.compile(r"^gsp-[0-9]{14}-[a-f0-9]{8}$")
+MAX_SOURCE_FILES: Final = 10_000
+MAX_SOURCE_BYTES: Final = 16 * 1024 * 1024 * 1024
+MAX_UPLOAD_CHUNK_BYTES: Final = 16 * 1024 * 1024
+TERMINAL_STATES: Final = {"ready", "failed"}
+ACTIVE_STATES: Final = {"queued", "processing", "importing"}
+PROVIDER_JOB_STATES: Final = {
+    "queued",
+    "verifying_source",
+    "inspecting",
+    "building_preview",
+    "building_streamed_sog",
+    "building_collision",
+    "ready",
+    "failed",
+}
+PROVIDER_POLL_TIMEOUT_SECONDS: Final = 2 * 60 * 60 + 5 * 60
+
+
+class SimulationProjectError(RuntimeError):
+    """The project request violates its durable catalog contract."""
+
+
+class SimulationProjectNotFoundError(SimulationProjectError):
+    """The requested project or source member does not exist."""
+
+
+class SimulationProjectConflictError(SimulationProjectError):
+    """The requested transition conflicts with current project state."""
+
+
+class SimulationProjectStore:
+    def __init__(self, data_dir: Path) -> None:
+        self.root = data_dir.expanduser().resolve() / "simulation-worlds"
+        self.projects_root = self.root / "projects"
+        self.projects_root.mkdir(mode=0o700, parents=True, exist_ok=True)
+        self._lock = threading.RLock()
+
+    def create(
+        self,
+        *,
+        name: str,
+        scene_type: str,
+        source_kind: str,
+        files: list[dict[str, object]],
+    ) -> dict[str, Any]:
+        display_name = _project_name(name)
+        if scene_type not in {"interior", "outdoor", "object"}:
+            raise SimulationProjectError("simulation scene type is invalid")
+        if source_kind not in {"archive", "folder"}:
+            raise SimulationProjectError("simulation source kind is invalid")
+        if not files or len(files) > MAX_SOURCE_FILES:
+            raise SimulationProjectError("simulation source file count is outside limits")
+        normalized: list[dict[str, object]] = []
+        logical_paths: set[str] = set()
+        total_bytes = 0
+        for index, raw in enumerate(files):
+            if set(raw) != {"logical_path", "byte_length"}:
+                raise SimulationProjectError("simulation source file contract is invalid")
+            logical_path = _logical_path(raw.get("logical_path"))
+            byte_length = raw.get("byte_length")
+            if (
+                not isinstance(byte_length, int)
+                or isinstance(byte_length, bool)
+                or byte_length <= 0
+                or byte_length > MAX_SOURCE_BYTES
+            ):
+                raise SimulationProjectError("simulation source file size is invalid")
+            if logical_path in logical_paths:
+                raise SimulationProjectError("simulation source paths must be unique")
+            logical_paths.add(logical_path)
+            total_bytes += byte_length
+            normalized.append({
+                "file_id": f"src-{index:05d}-{uuid4().hex[:8]}",
+                "logical_path": logical_path,
+                "byte_length": byte_length,
+                "uploaded_bytes": 0,
+                "sha256": None,
+            })
+        if total_bytes > MAX_SOURCE_BYTES:
+            raise SimulationProjectError("simulation source exceeds byte admission")
+        if source_kind == "archive":
+            if len(normalized) != 1 or "/" in str(normalized[0]["logical_path"]):
+                raise SimulationProjectError("archive source must be one top-level file")
+            if Path(str(normalized[0]["logical_path"])).suffix.lower() not in {
+                ".zip",
+                ".rar",
+                ".7z",
+            }:
+                raise SimulationProjectError("archive source must be zip, rar or 7z")
+        project_id = f"sim-{uuid4().hex}"
+        now = utc_now_iso()
+        document: dict[str, Any] = {
+            "schema_version": PROJECT_SCHEMA,
+            "project_id": project_id,
+            "name": display_name,
+            "scene_type": scene_type,
+            "status": "uploading",
+            "source": {
+                "kind": source_kind,
+                "total_byte_length": total_bytes,
+                "uploaded_byte_length": 0,
+                "files": normalized,
+                "bundle_sha256": None,
+            },
+            "provider": {
+                "provider_id": "gaussian-pipeline",
+                "job_id": None,
+                "state": None,
+                "progress": None,
+                "runtime": None,
+            },
+            "artifacts": [],
+            "world_manifest": None,
+            "error": None,
+            "created_at_utc": now,
+            "updated_at_utc": now,
+        }
+        with self._lock:
+            project_root = self._project_root(project_id)
+            project_root.mkdir(mode=0o700, parents=False, exist_ok=False)
+            (project_root / "source").mkdir(mode=0o700)
+            (project_root / "artifacts").mkdir(mode=0o700)
+            self._write(document)
+        return document
+
+    def list(self) -> list[dict[str, Any]]:
+        with self._lock:
+            projects: list[dict[str, Any]] = []
+            for entry in self.projects_root.iterdir():
+                if not entry.is_dir() or PROJECT_ID_PATTERN.fullmatch(entry.name) is None:
+                    continue
+                projects.append(self._read(entry.name))
+            projects.sort(
+                key=lambda item: (str(item["created_at_utc"]), str(item["project_id"])),
+                reverse=True,
+            )
+            return projects
+
+    def get(self, project_id: str) -> dict[str, Any]:
+        with self._lock:
+            return self._read(project_id)
+
+    def update_metadata(
+        self,
+        project_id: str,
+        *,
+        name: str,
+        scene_type: str,
+    ) -> dict[str, Any]:
+        display_name = _project_name(name)
+        if scene_type not in {"interior", "outdoor", "object"}:
+            raise SimulationProjectError("simulation scene type is invalid")
+        with self._lock:
+            document = self._read(project_id)
+            document["name"] = display_name
+            document["scene_type"] = scene_type
+            document["updated_at_utc"] = utc_now_iso()
+            self._write(document)
+            return document
+
+    def upload_state(self, project_id: str, file_id: str) -> tuple[int, int]:
+        with self._lock:
+            document = self._read(project_id)
+            source_file = _source_file(document, file_id)
+            return int(source_file["uploaded_bytes"]), int(source_file["byte_length"])
+
+    def append_upload(
+        self,
+        project_id: str,
+        file_id: str,
+        *,
+        offset: int,
+        payload: bytes,
+    ) -> dict[str, Any]:
+        if not payload or len(payload) > MAX_UPLOAD_CHUNK_BYTES:
+            raise SimulationProjectError("simulation upload chunk is outside limits")
+        with self._lock:
+            document = self._read(project_id)
+            if document["status"] != "uploading":
+                raise SimulationProjectConflictError("simulation source upload is closed")
+            source_file = _source_file(document, file_id)
+            uploaded = int(source_file["uploaded_bytes"])
+            byte_length = int(source_file["byte_length"])
+            if offset != uploaded:
+                raise SimulationProjectConflictError("simulation upload offset does not match")
+            if uploaded + len(payload) > byte_length:
+                raise SimulationProjectError("simulation upload exceeds declared file size")
+            target = self._source_path(project_id, str(source_file["logical_path"]))
+            target.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
+            mode = "xb" if uploaded == 0 else "r+b"
+            with target.open(mode) as stream:
+                if uploaded:
+                    stream.seek(uploaded)
+                stream.write(payload)
+                stream.flush()
+                os.fsync(stream.fileno())
+            uploaded += len(payload)
+            source_file["uploaded_bytes"] = uploaded
+            if uploaded == byte_length:
+                source_file["sha256"] = _sha256(target)
+            document["source"]["uploaded_byte_length"] = sum(
+                int(item["uploaded_bytes"]) for item in document["source"]["files"]
+            )
+            document["updated_at_utc"] = utc_now_iso()
+            self._write(document)
+            return document
+
+    def begin_build(self, project_id: str) -> dict[str, Any]:
+        with self._lock:
+            document = self._read(project_id)
+            if document["status"] != "uploading":
+                raise SimulationProjectConflictError("simulation project cannot start a build")
+            if any(
+                int(item["uploaded_bytes"]) != int(item["byte_length"])
+                for item in document["source"]["files"]
+            ):
+                raise SimulationProjectConflictError("simulation source upload is incomplete")
+            document["status"] = "queued"
+            document["error"] = None
+            document["updated_at_utc"] = utc_now_iso()
+            self._write(document)
+            return document
+
+    def update_processing(
+        self,
+        project_id: str,
+        *,
+        status: str,
+        provider_job_id: str | None = None,
+        provider_state: str | None = None,
+        progress: object = None,
+        bundle_sha256: str | None = None,
+    ) -> dict[str, Any]:
+        if status not in {"queued", "processing", "importing"}:
+            raise SimulationProjectError("simulation processing status is invalid")
+        with self._lock:
+            document = self._read(project_id)
+            document["status"] = status
+            if provider_job_id is not None:
+                if PROVIDER_JOB_ID_PATTERN.fullmatch(provider_job_id) is None:
+                    raise SimulationProjectError("simulation provider job id is invalid")
+                document["provider"]["job_id"] = provider_job_id
+            if provider_state is not None:
+                document["provider"]["state"] = provider_state
+            if progress is not None:
+                document["provider"]["progress"] = progress
+            if bundle_sha256 is not None:
+                document["source"]["bundle_sha256"] = bundle_sha256
+            document["updated_at_utc"] = utc_now_iso()
+            self._write(document)
+            return document
+
+    def complete(
+        self,
+        project_id: str,
+        *,
+        result: dict[str, Any],
+        artifacts: list[dict[str, Any]],
+        world_manifest: dict[str, Any],
+    ) -> dict[str, Any]:
+        with self._lock:
+            document = self._read(project_id)
+            document["status"] = "ready"
+            document["provider"]["state"] = "ready"
+            document["provider"]["runtime"] = result.get("runtime")
+            document["artifacts"] = artifacts
+            document["world_manifest"] = world_manifest
+            document["error"] = None
+            document["updated_at_utc"] = utc_now_iso()
+            self._write(document)
+            return document
+
+    def fail(self, project_id: str, message: str) -> dict[str, Any]:
+        with self._lock:
+            document = self._read(project_id)
+            document["status"] = "failed"
+            document["error"] = message[:2048]
+            document["updated_at_utc"] = utc_now_iso()
+            self._write(document)
+            return document
+
+    def artifact_path(self, project_id: str, logical_path: str) -> tuple[Path, dict[str, Any]]:
+        safe = _logical_path(logical_path)
+        with self._lock:
+            document = self._read(project_id)
+            descriptor = next(
+                (item for item in document["artifacts"] if item.get("logical_path") == safe),
+                None,
+            )
+            if descriptor is None:
+                raise SimulationProjectNotFoundError("simulation artifact is unavailable")
+            target = self._artifact_path(project_id, safe)
+            if not target.is_file() or target.is_symlink():
+                raise SimulationProjectNotFoundError("simulation artifact is unavailable")
+            return target, descriptor
+
+    def delete(self, project_id: str) -> None:
+        with self._lock:
+            document = self._read(project_id)
+            if document["status"] not in TERMINAL_STATES and document["status"] != "uploading":
+                raise SimulationProjectConflictError(
+                    "active simulation project cannot be deleted"
+                )
+            shutil.rmtree(self._project_root(project_id))
+
+    def source_root(self, project_id: str) -> Path:
+        self.get(project_id)
+        return self._project_root(project_id) / "source"
+
+    def artifacts_root(self, project_id: str) -> Path:
+        self.get(project_id)
+        return self._project_root(project_id) / "artifacts"
+
+    def _read(self, project_id: str) -> dict[str, Any]:
+        project_root = self._project_root(project_id)
+        try:
+            document = json.loads((project_root / "project.json").read_text(encoding="utf-8"))
+        except (OSError, json.JSONDecodeError) as exc:
+            raise SimulationProjectNotFoundError("simulation project is unavailable") from exc
+        if (
+            not isinstance(document, dict)
+            or document.get("schema_version") != PROJECT_SCHEMA
+            or document.get("project_id") != project_id
+        ):
+            raise SimulationProjectError("persisted simulation project identity is invalid")
+        return document
+
+    def _write(self, document: dict[str, Any]) -> None:
+        project_id = str(document["project_id"])
+        destination = self._project_root(project_id) / "project.json"
+        temporary = destination.with_name(f".project-{uuid4().hex}.tmp")
+        with temporary.open("x", encoding="utf-8") as stream:
+            json.dump(document, stream, ensure_ascii=False, indent=2)
+            stream.write("\n")
+            stream.flush()
+            os.fsync(stream.fileno())
+        temporary.replace(destination)
+
+    def _project_root(self, project_id: str) -> Path:
+        if PROJECT_ID_PATTERN.fullmatch(project_id) is None:
+            raise SimulationProjectNotFoundError("simulation project is unavailable")
+        return self.projects_root / project_id
+
+    def _source_path(self, project_id: str, logical_path: str) -> Path:
+        return _confined_path(self._project_root(project_id) / "source", logical_path)
+
+    def _artifact_path(self, project_id: str, logical_path: str) -> Path:
+        return _confined_path(self._project_root(project_id) / "artifacts", logical_path)
+
+
+class SimulationProjectService:
+    def __init__(
+        self,
+        store: SimulationProjectStore,
+        provider_factory=configured_gaussian_pipeline_gateway,
+    ) -> None:
+        self.store = store
+        self.provider_factory = provider_factory
+
+    def recover_pending(self) -> int:
+        pending = [
+            project for project in self.store.list()
+            if project.get("status") in ACTIVE_STATES
+        ]
+        for project in pending:
+            threading.Thread(
+                target=self.process,
+                args=(str(project["project_id"]),),
+                name=f"simulation-recovery-{str(project['project_id'])[-8:]}",
+                daemon=True,
+            ).start()
+        return len(pending)
+
+    def process(self, project_id: str) -> None:
+        provider: GaussianPipelineGateway | None = None
+        try:
+            project = self.store.get(project_id)
+            provider = self.provider_factory()
+            if provider is None:
+                raise SimulationProjectError("Gaussian Pipeline не настроен.")
+            provider.capabilities()
+            existing_job_id = project["provider"].get("job_id")
+            if isinstance(existing_job_id, str):
+                job_id = existing_job_id
+            else:
+                if project["status"] != "queued":
+                    raise SimulationProjectError(
+                        "Активная Gaussian-сборка потеряла provider job id после перезапуска."
+                    )
+                source_root = self.store.source_root(project_id)
+                if project["source"]["kind"] == "archive":
+                    source_file = project["source"]["files"][0]
+                    archive = provider.upload_source_archive(
+                        _confined_path(source_root, str(source_file["logical_path"]))
+                    )
+                    source = provider.normalize_archive(archive)
+                else:
+                    entrypoint, source_format = discover_gaussian_source_bundle(source_root)
+                    source = provider.upload_source_bundle(
+                        source_root,
+                        entrypoint=entrypoint,
+                        source_format=source_format,
+                    )
+                request = {
+                    "schema_version": BUILD_REQUEST_SCHEMA,
+                    "idempotency_key": f"missioncore-{project_id}",
+                    "source": source.to_dict(),
+                    "outputs": {
+                        "preview_sog": True,
+                        "streamed_sog": True,
+                        "collision": False,
+                    },
+                    "preview_lod": "coarsest",
+                    "collision_profile": None,
+                }
+                submitted = provider.submit_build(request)
+                job_id = submitted.get("job_id")
+                if not isinstance(job_id, str):
+                    raise SimulationProjectError("Gaussian Pipeline не вернул job id.")
+                self.store.update_processing(
+                    project_id,
+                    status="processing",
+                    provider_job_id=job_id,
+                    provider_state=str(submitted.get("state") or "queued"),
+                    progress=submitted.get("progress"),
+                    bundle_sha256=source.bundle_sha256,
+                )
+            deadline = time.monotonic() + PROVIDER_POLL_TIMEOUT_SECONDS
+            while True:
+                if time.monotonic() >= deadline:
+                    raise SimulationProjectError(
+                        "Gaussian Pipeline превысил лимит ожидания сборки."
+                    )
+                job = provider.get_job(job_id)
+                state = job.get("state")
+                if not isinstance(state, str) or state not in PROVIDER_JOB_STATES:
+                    raise SimulationProjectError(
+                        "Gaussian Pipeline вернул неизвестное состояние сборки."
+                    )
+                self.store.update_processing(
+                    project_id,
+                    status="processing",
+                    provider_state=str(state or "unknown"),
+                    progress=job.get("progress"),
+                )
+                if state == "failed":
+                    error = job.get("error")
+                    message = error.get("message") if isinstance(error, dict) else None
+                    raise SimulationProjectError(
+                        str(message or "Gaussian Pipeline завершил сборку с ошибкой.")
+                    )
+                if state == "ready":
+                    break
+                time.sleep(2.0)
+            self.store.update_processing(
+                project_id,
+                status="importing",
+                provider_state="ready",
+            )
+            result = provider.get_result(job_id)
+            artifacts = _artifact_descriptors(result.get("artifacts"))
+            artifacts_root = self.store.artifacts_root(project_id)
+            for descriptor in artifacts:
+                provider.download_artifact(
+                    job_id,
+                    descriptor,
+                    _confined_path(artifacts_root, str(descriptor["logical_path"])),
+                )
+            world_manifest = _world_manifest(project_id, artifacts)
+            self.store.complete(
+                project_id,
+                result=result,
+                artifacts=artifacts,
+                world_manifest=world_manifest,
+            )
+        except (GaussianPipelineGatewayError, SimulationProjectError, OSError) as exc:
+            with suppress(SimulationProjectError):
+                self.store.fail(project_id, str(exc))
+        finally:
+            if provider is not None:
+                provider.close()
+
+    def delete(self, project_id: str) -> None:
+        project = self.store.get(project_id)
+        job_id = project["provider"].get("job_id")
+        if isinstance(job_id, str):
+            provider = self.provider_factory()
+            if provider is None:
+                raise SimulationProjectConflictError(
+                    "Gaussian Pipeline недоступен для удаления серверных артефактов."
+                )
+            try:
+                provider.delete_job(job_id)
+            finally:
+                provider.close()
+        self.store.delete(project_id)
+
+
+def _artifact_descriptors(value: object) -> list[dict[str, Any]]:
+    if not isinstance(value, list) or not value:
+        raise SimulationProjectError("Gaussian Pipeline не вернул артефакты сцены.")
+    descriptors: list[dict[str, Any]] = []
+    for item in value:
+        if not isinstance(item, dict):
+            raise SimulationProjectError("Gaussian artifact descriptor is invalid")
+        logical_path = _logical_path(item.get("logical_path"))
+        sha256 = item.get("sha256")
+        byte_length = item.get("byte_length")
+        media_type = item.get("media_type")
+        role = item.get("role")
+        if (
+            not isinstance(sha256, str)
+            or re.fullmatch(r"[a-f0-9]{64}", sha256) is None
+            or not isinstance(byte_length, int)
+            or isinstance(byte_length, bool)
+            or byte_length < 0
+            or not isinstance(media_type, str)
+            or not isinstance(role, str)
+        ):
+            raise SimulationProjectError("Gaussian artifact fields are invalid")
+        descriptors.append({
+            "role": role,
+            "logical_path": logical_path,
+            "media_type": media_type,
+            "sha256": sha256,
+            "byte_length": byte_length,
+        })
+    return descriptors
+
+
+def _world_manifest(project_id: str, artifacts: list[dict[str, Any]]) -> dict[str, Any]:
+    def url_for(role: str) -> str | None:
+        descriptor = next((item for item in artifacts if item["role"] == role), None)
+        if descriptor is None:
+            return None
+        encoded = "/".join(
+            quote(part, safe="")
+            for part in str(descriptor["logical_path"]).split("/")
+        )
+        return f"/api/v1/simulation-worlds/projects/{project_id}/artifacts/{encoded}"
+
+    return {
+        "schema_version": WORLD_MANIFEST_SCHEMA,
+        "project_id": project_id,
+        "visual": {
+            "preview_sog_url": url_for("preview"),
+            "streamed_sog_url": url_for("stream-manifest"),
+        },
+        "collision": {
+            "mesh_url": url_for("collision-mesh"),
+            "available": url_for("collision-mesh") is not None,
+        },
+        "transforms": {
+            "world_from_visual": [1, 0, 0, 0, 0, 1, 0, 0, 0, 0, 1, 0, 0, 0, 0, 1],
+            "world_from_collision": [1, 0, 0, 0, 0, 1, 0, 0, 0, 0, 1, 0, 0, 0, 0, 1],
+        },
+    }
+
+
+def _project_name(value: str) -> str:
+    normalized = " ".join(value.split())
+    if not 1 <= len(normalized) <= 120:
+        raise SimulationProjectError("simulation project name is invalid")
+    return normalized
+
+
+def _logical_path(value: object) -> str:
+    if not isinstance(value, str) or not value or len(value) > 1024:
+        raise SimulationProjectError("simulation logical path is invalid")
+    if value.startswith("/") or "\\" in value or "\x00" in value:
+        raise SimulationProjectError("simulation logical path is unsafe")
+    parts = value.split("/")
+    if any(part in {"", ".", ".."} for part in parts):
+        raise SimulationProjectError("simulation logical path is unsafe")
+    return value
+
+
+def _source_file(document: dict[str, Any], file_id: str) -> dict[str, Any]:
+    if SOURCE_FILE_ID_PATTERN.fullmatch(file_id) is None:
+        raise SimulationProjectNotFoundError("simulation source file is unavailable")
+    source_file = next(
+        (item for item in document["source"]["files"] if item.get("file_id") == file_id),
+        None,
+    )
+    if source_file is None:
+        raise SimulationProjectNotFoundError("simulation source file is unavailable")
+    return source_file
+
+
+def _confined_path(root: Path, logical_path: str) -> Path:
+    safe = _logical_path(logical_path)
+    candidate = root.joinpath(*safe.split("/")).resolve()
+    resolved_root = root.resolve()
+    if not candidate.is_relative_to(resolved_root):
+        raise SimulationProjectError("simulation path escaped its project root")
+    return candidate
+
+
+def _sha256(path: Path) -> str:
+    digest = hashlib.sha256()
+    with path.open("rb") as stream:
+        for chunk in iter(lambda: stream.read(1024 * 1024), b""):
+            digest.update(chunk)
+    return digest.hexdigest()