refactor(platform): harden LAB evidence and telemetry

2026-08-05 10:03:37 +03:00
parent 67bd96868f
commit 4c763bd8aa
42 changed files with 936 additions and 272 deletions
@@ -0,0 +1,15 @@
+"""Configuration contracts for Mission Core laboratory evidence."""
+
+from k1link.laboratory.evidence_registry import (
+    LABORATORY_EVIDENCE_DEFINITION_SCHEMA,
+    LaboratoryEvidenceDefinition,
+    LaboratoryEvidenceRegistry,
+    LaboratoryRegistryError,
+)
+
+__all__ = [
+    "LABORATORY_EVIDENCE_DEFINITION_SCHEMA",
+    "LaboratoryEvidenceDefinition",
+    "LaboratoryEvidenceRegistry",
+    "LaboratoryRegistryError",
+]
@@ -0,0 +1,187 @@
+from __future__ import annotations
+
+import json
+import re
+from dataclasses import dataclass
+from pathlib import Path, PurePosixPath
+from typing import Final
+
+LABORATORY_EVIDENCE_DEFINITION_SCHEMA: Final = "missioncore.laboratory-evidence-definition/v1"
+_DEFINITION_MAX_BYTES: Final = 16 * 1024
+_IDENTIFIER = re.compile(r"^[a-z][a-z0-9-]{2,95}$")
+_SCHEMA_VERSION = re.compile(r"^missioncore\.[a-z0-9.-]+/v[1-9][0-9]*$")
+_TOP_LEVEL_KEYS: Final = frozenset({"schema_version", "work_id", "evidence"})
+_EVIDENCE_KEYS: Final = frozenset(
+    {"runtime_relative_root", "result_id_prefix", "document_name", "schema_version"}
+)
+
+
+class LaboratoryRegistryError(ValueError):
+    """Raised when a LAB evidence definition is unsafe or ambiguous."""
+
+
+@dataclass(frozen=True, slots=True)
+class LaboratoryEvidenceDefinition:
+    work_id: str
+    runtime_relative_root: PurePosixPath
+    result_id_prefix: str
+    document_name: str
+    result_schema_version: str
+
+    def __post_init__(self) -> None:
+        _identifier(self.work_id, "work_id")
+        _identifier(self.result_id_prefix, "result_id_prefix")
+        _document_name(self.document_name)
+        _schema_version(self.result_schema_version)
+        if not isinstance(self.runtime_relative_root, PurePosixPath):
+            raise LaboratoryRegistryError("runtime_relative_root must be a POSIX path")
+        _relative_root(str(self.runtime_relative_root))
+
+    @property
+    def result_id_pattern(self) -> re.Pattern[str]:
+        return re.compile(rf"^{re.escape(self.result_id_prefix)}-[a-f0-9]{{64}}$")
+
+    def result_root(self, runtime_root: Path) -> Path:
+        return runtime_root.joinpath(*self.runtime_relative_root.parts)
+
+
+@dataclass(frozen=True, slots=True)
+class LaboratoryEvidenceRegistry:
+    definitions: tuple[LaboratoryEvidenceDefinition, ...]
+
+    def __post_init__(self) -> None:
+        if not isinstance(self.definitions, tuple) or not all(
+            isinstance(definition, LaboratoryEvidenceDefinition) for definition in self.definitions
+        ):
+            raise LaboratoryRegistryError("LAB definitions must be an immutable tuple")
+        _reject_duplicates(self.definitions)
+
+    @classmethod
+    def from_directory(cls, root: Path) -> LaboratoryEvidenceRegistry:
+        definition_root = _real_directory(root, "LAB definition root")
+        definitions = tuple(
+            _read_definition(path)
+            for path in sorted(definition_root.glob("*.json"), key=lambda item: item.name)
+        )
+        return cls(definitions=definitions)
+
+
+def _real_directory(path: Path, label: str) -> Path:
+    candidate = path.expanduser().absolute()
+    if candidate.is_symlink():
+        raise LaboratoryRegistryError(f"{label} must not be a symlink")
+    try:
+        resolved = candidate.resolve(strict=True)
+    except OSError as exc:
+        raise LaboratoryRegistryError(f"{label} does not exist") from exc
+    if not resolved.is_dir():
+        raise LaboratoryRegistryError(f"{label} must be a directory")
+    return resolved
+
+
+def _read_definition(path: Path) -> LaboratoryEvidenceDefinition:
+    if path.is_symlink() or not path.is_file():
+        raise LaboratoryRegistryError(f"LAB definition must be a regular file: {path.name}")
+    if path.stat().st_size > _DEFINITION_MAX_BYTES:
+        raise LaboratoryRegistryError(f"LAB definition is too large: {path.name}")
+    try:
+        payload: object = json.loads(path.read_text(encoding="utf-8"))
+    except (json.JSONDecodeError, OSError) as exc:
+        raise LaboratoryRegistryError(f"LAB definition is unreadable: {path.name}") from exc
+    document = _object(payload, f"LAB definition {path.name}")
+    _exact_keys(document, _TOP_LEVEL_KEYS, f"LAB definition {path.name}")
+    if document["schema_version"] != LABORATORY_EVIDENCE_DEFINITION_SCHEMA:
+        raise LaboratoryRegistryError(f"LAB definition schema is invalid: {path.name}")
+    work_id = _identifier(document["work_id"], "work_id")
+    if path.name != f"{work_id}.json":
+        raise LaboratoryRegistryError(f"LAB definition filename must match work_id: {path.name}")
+    evidence = _object(document["evidence"], f"LAB evidence {work_id}")
+    _exact_keys(evidence, _EVIDENCE_KEYS, f"LAB evidence {work_id}")
+    result_id_prefix = _identifier(evidence["result_id_prefix"], "result_id_prefix")
+    document_name = _document_name(evidence["document_name"])
+    result_schema_version = _schema_version(evidence["schema_version"])
+    return LaboratoryEvidenceDefinition(
+        work_id=work_id,
+        runtime_relative_root=_relative_root(evidence["runtime_relative_root"]),
+        result_id_prefix=result_id_prefix,
+        document_name=document_name,
+        result_schema_version=result_schema_version,
+    )
+
+
+def _object(value: object, label: str) -> dict[str, object]:
+    if not isinstance(value, dict) or not all(isinstance(key, str) for key in value):
+        raise LaboratoryRegistryError(f"{label} must be an object")
+    return value
+
+
+def _exact_keys(document: dict[str, object], expected: frozenset[str], label: str) -> None:
+    actual = frozenset(document)
+    if actual != expected:
+        missing = sorted(expected - actual)
+        unexpected = sorted(actual - expected)
+        raise LaboratoryRegistryError(
+            f"{label} keys are invalid; missing={missing}, unexpected={unexpected}"
+        )
+
+
+def _text(value: object, label: str) -> str:
+    if not isinstance(value, str) or not value.strip() or value != value.strip():
+        raise LaboratoryRegistryError(f"{label} must be a non-empty trimmed string")
+    return value
+
+
+def _identifier(value: object, label: str) -> str:
+    text = _text(value, label)
+    if _IDENTIFIER.fullmatch(text) is None:
+        raise LaboratoryRegistryError(f"{label} is invalid")
+    return text
+
+
+def _schema_version(value: object) -> str:
+    text = _text(value, "evidence schema_version")
+    if _SCHEMA_VERSION.fullmatch(text) is None:
+        raise LaboratoryRegistryError("evidence schema_version is invalid")
+    return text
+
+
+def _document_name(value: object) -> str:
+    text = _text(value, "document_name")
+    candidate = PurePosixPath(text)
+    if (
+        candidate.is_absolute()
+        or len(candidate.parts) != 1
+        or candidate.name != text
+        or candidate.suffix != ".json"
+    ):
+        raise LaboratoryRegistryError("document_name must be one JSON filename")
+    return text
+
+
+def _relative_root(value: object) -> PurePosixPath:
+    text = _text(value, "runtime_relative_root")
+    if "\\" in text:
+        raise LaboratoryRegistryError("runtime_relative_root must use POSIX separators")
+    candidate = PurePosixPath(text)
+    if (
+        candidate.is_absolute()
+        or not candidate.parts
+        or text == "."
+        or str(candidate) != text
+        or any(part in {"", ".", ".."} for part in candidate.parts)
+    ):
+        raise LaboratoryRegistryError("runtime_relative_root must be a normalized relative path")
+    return candidate
+
+
+def _reject_duplicates(definitions: tuple[LaboratoryEvidenceDefinition, ...]) -> None:
+    dimensions = {
+        "work_id": [definition.work_id for definition in definitions],
+        "result_id_prefix": [definition.result_id_prefix for definition in definitions],
+        "runtime_relative_root": [
+            str(definition.runtime_relative_root) for definition in definitions
+        ],
+    }
+    for label, values in dimensions.items():
+        if len(values) != len(set(values)):
+            raise LaboratoryRegistryError(f"duplicate LAB {label}")