feat(lab): complete E30 evidence review gate

This commit is contained in:
DCCONSTRUCTIONS
2026-07-27 11:00:32 +03:00
parent a44d7627fd
commit 001d597a89
55 changed files with 15897 additions and 1548 deletions
+259
View File
@@ -0,0 +1,259 @@
"""Exact camera-frame evidence for an E30 materialization.
The E30 point substrate is useful only when a reviewer can see the camera
observation that caused the semantic claim. This module binds an immutable
camera compute job and decodes only the exact source-frame indices selected by
the E30 review pack. The canonical camera archive remains unchanged.
"""
from __future__ import annotations
import hashlib
import os
import shutil
import subprocess
import threading
from dataclasses import dataclass
from pathlib import Path
from typing import Final
from PIL import Image
from k1link.compute.jobs import CameraComputeJob, validate_camera_compute_job
_READ_CHUNK_BYTES: Final = 1024 * 1024
_FFMPEG_TIMEOUT_SECONDS: Final = 600.0
class E30CameraEvidenceError(ValueError):
"""The bound camera job or a decoded evidence frame is invalid."""
@dataclass(frozen=True, slots=True)
class E30CameraEvidenceSource:
job: CameraComputeJob
epoch_root: Path
ffmpeg_path: Path
encoder_identity: str
def identity(self) -> dict[str, object]:
return {
"schema_version": "missioncore.e30-camera-evidence-source/v1",
"job_id": self.job.job_id,
"input_sha256": self.job.input_sha256,
"session_id": self.job.session_id,
"source_id": self.job.source_id,
"codec_epoch": self.job.codec_epoch,
"segment_count": self.job.segment_count,
"source_frame_basis": "zero-based-decoded-video-frame",
"camera_sequence_basis": "source-frame-index-plus-one",
"decoder": {
"name": "ffmpeg",
"identity": self.encoder_identity,
"output": "jpeg-q2-yuvj420p",
},
}
def open_e30_camera_evidence_source(
*,
camera_job_root: Path,
ffmpeg_path: Path,
expected_session_id: str,
expected_source_id: str,
) -> E30CameraEvidenceSource:
"""Validate the immutable camera job and bind its decoder identity."""
try:
job = validate_camera_compute_job(camera_job_root)
except (OSError, ValueError) as exc:
raise E30CameraEvidenceError("E30 camera compute job is invalid") from exc
if (
job.session_id != expected_session_id
or job.source_id != expected_source_id
):
raise E30CameraEvidenceError(
"E30 camera job does not match the LiDAR/projection source"
)
decoder = ffmpeg_path.expanduser().resolve(strict=True)
if not decoder.is_file():
raise E30CameraEvidenceError("ffmpeg is unavailable")
try:
version = subprocess.run(
[str(decoder), "-version"],
check=True,
capture_output=True,
text=True,
timeout=10,
)
except (OSError, subprocess.SubprocessError) as exc:
raise E30CameraEvidenceError("ffmpeg identity is unavailable") from exc
version_line = version.stdout.splitlines()[0].strip()
if not version_line:
raise E30CameraEvidenceError("ffmpeg identity is empty")
epoch = (
job.job_root
/ "input"
/ "camera"
/ job.source_id
/ f"epoch-{job.codec_epoch}"
).resolve(strict=True)
if not epoch.is_dir() or not epoch.is_relative_to(job.job_root):
raise E30CameraEvidenceError("E30 camera epoch escaped its compute job")
return E30CameraEvidenceSource(
job=job,
epoch_root=epoch,
ffmpeg_path=decoder,
encoder_identity=hashlib.sha256(version_line.encode()).hexdigest(),
)
def materialize_e30_camera_frames(
*,
source: E30CameraEvidenceSource,
source_frame_indices: tuple[int, ...],
destination_root: Path,
width: int,
height: int,
) -> dict[int, dict[str, object]]:
"""Decode exact source frames once and return verified artifact metadata."""
frames = tuple(sorted(set(source_frame_indices)))
if (
not frames
or any(
isinstance(frame, bool)
or frame < 0
or frame >= source.job.segment_count
for frame in frames
)
or width < 1
or height < 1
):
raise E30CameraEvidenceError("E30 camera frame selection is invalid")
destination = destination_root.resolve()
destination.mkdir(mode=0o700, parents=True, exist_ok=False)
init_path = source.epoch_root / "init.mp4"
segments = tuple(
source.epoch_root / "segments" / f"{sequence}.m4s"
for sequence in range(1, source.job.segment_count + 1)
)
select = "+".join(f"eq(n\\,{frame})" for frame in frames)
temporary_pattern = destination / ".decoded-%06d.jpg"
read_fd, write_fd = os.pipe()
process = subprocess.Popen(
[
str(source.ffmpeg_path),
"-hide_banner",
"-loglevel",
"error",
"-i",
"pipe:0",
"-vf",
f"select={select}",
"-fps_mode",
"passthrough",
"-frames:v",
str(len(frames)),
"-q:v",
"2",
"-pix_fmt",
"yuvj420p",
"-start_number",
"0",
str(temporary_pattern),
],
stdin=read_fd,
stdout=subprocess.DEVNULL,
stderr=subprocess.PIPE,
shell=False,
)
os.close(read_fd)
feeder_errors: list[BaseException] = []
def feed_archive() -> None:
try:
with os.fdopen(write_fd, "wb", buffering=0) as sink:
for path in (init_path, *segments):
with path.open("rb") as stream:
shutil.copyfileobj(
stream,
sink,
length=_READ_CHUNK_BYTES,
)
except BrokenPipeError:
return
except BaseException as exc: # pragma: no cover - child pipe boundary
feeder_errors.append(exc)
feeder = threading.Thread(
target=feed_archive,
name="e30-camera-evidence-ffmpeg",
daemon=True,
)
feeder.start()
try:
_, stderr = process.communicate(timeout=_FFMPEG_TIMEOUT_SECONDS)
except subprocess.TimeoutExpired as exc:
process.kill()
process.communicate()
raise E30CameraEvidenceError("E30 camera extraction timed out") from exc
feeder.join(timeout=10.0)
if feeder.is_alive():
raise E30CameraEvidenceError("E30 camera archive feeder did not stop")
if feeder_errors:
raise E30CameraEvidenceError(
"E30 camera archive streaming failed"
) from feeder_errors[0]
if process.returncode != 0:
message = (stderr or b"").decode(errors="replace").strip()[-1000:]
raise E30CameraEvidenceError(
f"ffmpeg rejected the E30 camera archive: {message}"
)
artifacts: dict[int, dict[str, object]] = {}
for ordinal, frame_index in enumerate(frames):
decoded = destination / f".decoded-{ordinal:06d}.jpg"
final = destination / f"frame-{frame_index:06d}.jpg"
if decoded.is_symlink() or not decoded.is_file():
raise E30CameraEvidenceError(
"ffmpeg returned an incomplete E30 camera frame set"
)
try:
with Image.open(decoded) as image:
image.verify()
with Image.open(decoded) as image:
decoded_size = image.size
decoded_format = image.format
except (OSError, ValueError) as exc:
raise E30CameraEvidenceError(
"an E30 camera frame is not a valid image"
) from exc
if decoded_size != (width, height) or decoded_format != "JPEG":
raise E30CameraEvidenceError(
"an E30 camera frame has an unexpected format"
)
os.replace(decoded, final)
artifacts[frame_index] = {
"role": "camera-frame",
"path": f"frames/{final.name}",
"media_type": "image/jpeg",
"byte_length": final.stat().st_size,
"sha256": _sha256_file(final),
"width": width,
"height": height,
"source_frame_index": frame_index,
"camera_sequence": frame_index + 1,
"exact_source_frame": True,
}
if any(destination.glob(".decoded-*.jpg")):
raise E30CameraEvidenceError("ffmpeg returned undeclared E30 frames")
return artifacts
def _sha256_file(path: Path) -> str:
digest = hashlib.sha256()
with path.open("rb") as stream:
while chunk := stream.read(_READ_CHUNK_BYTES):
digest.update(chunk)
return digest.hexdigest()
@@ -0,0 +1,645 @@
"""Immutable AI-assisted engineering generations over E30 evidence.
This contract is intentionally separate from ``e30_human_review``. It records
an explicitly AI-assisted engineering assessment, retains uncertainty, and
routes only bounded exceptions to a human. It never claims human ground truth
or navigation/safety acceptance.
"""
from __future__ import annotations
import hashlib
import json
import math
import os
import re
import shutil
import tempfile
from collections import Counter
from pathlib import Path
from typing import Any, Final
from k1link.artifacts import utc_now_iso, write_json_atomic
E30_ENGINEERING_DECISION_SCHEMA_V1: Final = (
"missioncore.e30-engineering-decision/v1"
)
E30_ENGINEERING_DECISION_SCHEMA: Final = (
"missioncore.e30-engineering-decision/v2"
)
E30_ENGINEERING_DECISION_SCHEMAS: Final = (
E30_ENGINEERING_DECISION_SCHEMA_V1,
E30_ENGINEERING_DECISION_SCHEMA,
)
E30_ENGINEERING_GENERATION_SCHEMA: Final = (
"missioncore.e30-engineering-generation/v1"
)
E30_ENGINEERING_SUMMARY_SCHEMA: Final = (
"missioncore.e30-engineering-summary/v1"
)
E30_ENGINEERING_CAUSES_SCHEMA: Final = (
"missioncore.e30-engineering-causes/v1"
)
E30_ENGINEERING_EXCEPTION_SCHEMA_V1: Final = (
"missioncore.e30-engineering-exception/v1"
)
E30_ENGINEERING_EXCEPTION_SCHEMA: Final = (
"missioncore.e30-engineering-exception/v2"
)
E30_ENGINEERING_EXCEPTION_SCHEMAS: Final = (
E30_ENGINEERING_EXCEPTION_SCHEMA_V1,
E30_ENGINEERING_EXCEPTION_SCHEMA,
)
E30_ENGINEERING_DECISIONS_NAME: Final = "engineering-decisions.jsonl"
E30_ENGINEERING_EXCEPTIONS_NAME: Final = "human-exceptions.jsonl"
E30_ENGINEERING_CAUSES_NAME: Final = "cause-distribution.json"
E30_ENGINEERING_SUMMARY_NAME: Final = "summary.json"
E30_ENGINEERING_MANIFEST_NAME: Final = "manifest.json"
STRATA: Final = (
"conflict",
"agree",
"camera-only",
"unknown",
"geometry-only",
)
VERDICTS: Final = ("confirmed", "corrected", "insufficient-evidence")
DETECTOR_ASSESSMENTS: Final = (
"valid",
"class-mismatch",
"false-positive",
"missed-object",
"not-applicable",
"insufficient-evidence",
)
PROJECTION_ASSESSMENTS: Final = ("aligned", "misaligned", "not-assessable")
OWNERSHIP_ASSESSMENTS: Final = (
"object",
"surface-or-background",
"static-environment",
"self",
"insufficient-support",
"not-applicable",
"mixed",
"insufficient-evidence",
)
EXCEPTION_DISPOSITIONS: Final = (
"object-present",
"background-or-noise",
"insufficient-evidence",
)
_MATERIALIZATION_ID = re.compile(r"^e30-materialization-[a-f0-9]{64}$")
_GENERATION_ID = re.compile(r"^e30-engineering-generation-[a-f0-9]{64}$")
_ITEM_ID = re.compile(r"^e30-review-item-[a-f0-9]{64}$")
_SHA256 = re.compile(r"^[a-f0-9]{64}$")
_SAFE_PRODUCER = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:@/-]{0,159}$")
class E30EngineeringGenerationError(RuntimeError):
"""An engineering generation or its immutable source is invalid."""
def _canonical_json(value: object) -> bytes:
return json.dumps(
value,
ensure_ascii=False,
sort_keys=True,
separators=(",", ":"),
allow_nan=False,
).encode("utf-8")
def _sha256(path: Path) -> str:
digest = hashlib.sha256()
with path.open("rb") as stream:
while chunk := stream.read(1024 * 1024):
digest.update(chunk)
return digest.hexdigest()
def _regular_file(root: Path, relative: object) -> Path:
if not isinstance(relative, str) or not relative or relative.startswith("/"):
raise E30EngineeringGenerationError("artifact path is invalid")
path = root / relative
if (
path.is_symlink()
or not path.is_file()
or root.resolve() not in path.resolve().parents
):
raise E30EngineeringGenerationError("artifact is unavailable")
return path
def _read_json(path: Path) -> dict[str, Any]:
if path.is_symlink() or not path.is_file():
raise E30EngineeringGenerationError("required JSON is unavailable")
try:
value = json.loads(path.read_text(encoding="utf-8"))
except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc:
raise E30EngineeringGenerationError("required JSON is invalid") from exc
if not isinstance(value, dict):
raise E30EngineeringGenerationError("required JSON must be an object")
return value
def _read_jsonl(path: Path) -> list[dict[str, Any]]:
if path.is_symlink() or not path.is_file():
raise E30EngineeringGenerationError("required JSONL is unavailable")
values: list[dict[str, Any]] = []
with path.open("r", encoding="utf-8") as stream:
for line_number, line in enumerate(stream, start=1):
try:
value = json.loads(line)
except json.JSONDecodeError as exc:
raise E30EngineeringGenerationError(
f"invalid JSONL line {line_number}"
) from exc
if not isinstance(value, dict):
raise E30EngineeringGenerationError(
f"JSONL line {line_number} must be an object"
)
values.append(value)
return values
def _validate_artifact(root: Path, value: object) -> Path:
if not isinstance(value, dict):
raise E30EngineeringGenerationError("artifact metadata is invalid")
path = _regular_file(root, value.get("path"))
digest = value.get("sha256")
byte_length = value.get("byte_length")
if (
not isinstance(digest, str)
or _SHA256.fullmatch(digest) is None
or not isinstance(byte_length, int)
or isinstance(byte_length, bool)
or byte_length <= 0
or path.stat().st_size != byte_length
or _sha256(path) != digest
):
raise E30EngineeringGenerationError("artifact content changed")
return path
def _materialization(
root: Path,
) -> tuple[dict[str, Any], list[dict[str, Any]]]:
root = root.resolve()
if (
root.is_symlink()
or not root.is_dir()
or _MATERIALIZATION_ID.fullmatch(root.name) is None
):
raise E30EngineeringGenerationError("materialization root is invalid")
manifest_path = root / "manifest.json"
index_path = root / "materialized-items.jsonl"
manifest = _read_json(manifest_path)
identity = manifest.get("identity")
identity_sha256 = manifest.get("identity_sha256")
if (
manifest.get("schema_version")
!= "missioncore.e30-evidence-materialization/v2"
or manifest.get("result_id") != root.name
or not isinstance(identity, dict)
or not isinstance(identity_sha256, str)
or _SHA256.fullmatch(identity_sha256) is None
or hashlib.sha256(_canonical_json(identity)).hexdigest()
!= identity_sha256
or root.name != f"e30-materialization-{identity_sha256}"
or manifest.get("camera_evidence_available") is not True
or manifest.get("human_review_complete") is not False
or manifest.get("lab_published") is not False
):
raise E30EngineeringGenerationError("materialization identity changed")
authority = manifest.get("authority")
if (
not isinstance(authority, dict)
or authority.get("commands_enabled") is not False
or authority.get("navigation_or_safety_accepted") is not False
):
raise E30EngineeringGenerationError("materialization authority differs")
items = _read_jsonl(index_path)
if (
len(items) != manifest.get("item_count")
or len(items) != 486
or len({item.get("item_id") for item in items}) != len(items)
):
raise E30EngineeringGenerationError("materialization item count differs")
for expected_sequence, item in enumerate(items):
if (
item.get("schema_version")
!= "missioncore.e30-evidence-materialization-item/v2"
or item.get("sequence") != expected_sequence
or _ITEM_ID.fullmatch(str(item.get("item_id"))) is None
or item.get("stratum") not in STRATA
):
raise E30EngineeringGenerationError("materialization item changed")
_validate_artifact(root, item.get("artifact"))
_validate_artifact(root, item.get("camera_frame"))
return manifest, items
def _reason_taxonomy(
root: Path,
*,
expected_binding: object,
) -> tuple[str, ...]:
root = root.resolve()
if root.is_symlink() or not root.is_dir():
raise E30EngineeringGenerationError("review pack root is invalid")
manifest = _read_json(root / "manifest.json")
identity = manifest.get("identity")
if not isinstance(expected_binding, dict) or not isinstance(identity, dict):
raise E30EngineeringGenerationError("review pack binding is invalid")
if (
manifest.get("schema_version")
!= "missioncore.e30-evidence-review-pack/v1"
or manifest.get("result_id") != expected_binding.get("result_id")
or root.name != expected_binding.get("result_id")
or manifest.get("identity_sha256")
!= expected_binding.get("identity_sha256")
or manifest.get("selected_item_count")
!= expected_binding.get("item_count")
):
raise E30EngineeringGenerationError("review pack identity changed")
reason_taxonomy = identity.get("reason_taxonomy")
if (
not isinstance(reason_taxonomy, list)
or not reason_taxonomy
or not all(isinstance(value, str) and value for value in reason_taxonomy)
):
raise E30EngineeringGenerationError("reason taxonomy is unavailable")
return tuple(reason_taxonomy)
def _sheets(
root: Path,
*,
materialization_root: Path,
) -> tuple[dict[str, Any], dict[str, dict[str, object]]]:
root = root.resolve()
if root.is_symlink() or not root.is_dir():
raise E30EngineeringGenerationError("review sheet root is invalid")
manifest = _read_json(root / "manifest.json")
identity_sha256 = manifest.get("identity_sha256")
identity = {
key: value
for key, value in manifest.items()
if key
not in {
"identity_sha256",
"sheet_count",
"item_count",
"authority",
}
}
if (
manifest.get("schema_version")
!= "missioncore.e30-engineering-review-sheets/v1"
or manifest.get("materialization_id") != materialization_root.name
or not isinstance(identity_sha256, str)
or _SHA256.fullmatch(identity_sha256) is None
or hashlib.sha256(_canonical_json(identity)).hexdigest()
!= identity_sha256
or manifest.get("item_count") != 486
or manifest.get("sheet_count") != len(manifest.get("sheets", []))
):
raise E30EngineeringGenerationError("review sheet identity changed")
by_item: dict[str, dict[str, object]] = {}
sheets = manifest.get("sheets")
if not isinstance(sheets, list):
raise E30EngineeringGenerationError("review sheet list is invalid")
for sheet in sheets:
if not isinstance(sheet, dict):
raise E30EngineeringGenerationError("review sheet metadata is invalid")
path = _regular_file(root, sheet.get("path"))
if (
path.stat().st_size != sheet.get("byte_length")
or _sha256(path) != sheet.get("sha256")
or not isinstance(sheet.get("entries"), list)
):
raise E30EngineeringGenerationError("review sheet changed")
for entry in sheet["entries"]:
if (
not isinstance(entry, dict)
or _ITEM_ID.fullmatch(str(entry.get("item_id"))) is None
or entry["item_id"] in by_item
):
raise E30EngineeringGenerationError("review sheet entry is invalid")
by_item[str(entry["item_id"])] = {
"path": sheet["path"],
"sha256": sheet["sha256"],
"ordinal": entry.get("ordinal"),
}
if len(by_item) != 486:
raise E30EngineeringGenerationError("review sheet coverage differs")
return manifest, by_item
def _validate_decision(
*,
value: dict[str, Any],
item: dict[str, Any],
reason_taxonomy: tuple[str, ...],
expected_sheet: dict[str, object],
) -> dict[str, Any]:
confidence = value.get("confidence")
verdict = value.get("verdict")
effective_stratum = value.get("effective_stratum")
cause_code = value.get("cause_code")
exception_required = value.get("human_exception_required")
exception_reason = value.get("exception_reason")
note = value.get("evidence_note")
review_prompt = value.get("review_prompt")
if (
value.get("schema_version") != E30_ENGINEERING_DECISION_SCHEMA
or value.get("sequence") != item["sequence"]
or value.get("item_id") != item["item_id"]
or value.get("review_key") != item["review_key"]
or value.get("source_stratum") != item["stratum"]
or verdict not in VERDICTS
or effective_stratum not in (*STRATA, None)
or value.get("detector_assessment") not in DETECTOR_ASSESSMENTS
or value.get("projection_assessment") not in PROJECTION_ASSESSMENTS
or value.get("point_ownership") not in OWNERSHIP_ASSESSMENTS
or not isinstance(confidence, (int, float))
or isinstance(confidence, bool)
or not math.isfinite(float(confidence))
or not 0.0 <= float(confidence) <= 1.0
or not isinstance(exception_required, bool)
or exception_reason not in (None, "ambiguity", "high-impact")
or exception_required != (exception_reason is not None)
or not isinstance(note, str)
or not note.strip()
or len(note) > 1_000
or value.get("review_sheet") != expected_sheet
):
raise E30EngineeringGenerationError("engineering decision is invalid")
normalized_review_prompt: dict[str, object] | None = None
if exception_required:
if (
not isinstance(review_prompt, dict)
or set(review_prompt) != {"question", "focus", "effects"}
or not isinstance(review_prompt.get("question"), str)
or not review_prompt["question"].strip()
or len(review_prompt["question"]) > 240
or not isinstance(review_prompt.get("focus"), str)
or not review_prompt["focus"].strip()
or len(review_prompt["focus"]) > 500
or not isinstance(review_prompt.get("effects"), dict)
or set(review_prompt["effects"]) != set(EXCEPTION_DISPOSITIONS)
or any(
not isinstance(review_prompt["effects"].get(disposition), str)
or not review_prompt["effects"][disposition].strip()
or len(review_prompt["effects"][disposition]) > 500
for disposition in EXCEPTION_DISPOSITIONS
)
):
raise E30EngineeringGenerationError(
"engineering exception review prompt is invalid"
)
normalized_review_prompt = {
"question": review_prompt["question"].strip(),
"focus": review_prompt["focus"].strip(),
"effects": {
disposition: review_prompt["effects"][disposition].strip()
for disposition in EXCEPTION_DISPOSITIONS
},
}
elif review_prompt is not None:
raise E30EngineeringGenerationError(
"non-exception decision carries a review prompt"
)
if cause_code is not None and cause_code not in reason_taxonomy:
raise E30EngineeringGenerationError("decision cause is outside taxonomy")
if verdict == "confirmed" and effective_stratum != item["stratum"]:
raise E30EngineeringGenerationError("confirmed decision changes stratum")
if verdict == "corrected" and effective_stratum == item["stratum"]:
raise E30EngineeringGenerationError("corrected decision keeps stratum")
if verdict == "insufficient-evidence" and (
effective_stratum is not None or not exception_required
):
raise E30EngineeringGenerationError(
"insufficient decision must route an exception"
)
if verdict != "insufficient-evidence" and confidence < 0.5:
raise E30EngineeringGenerationError("issued verdict confidence is too low")
return {
**value,
"confidence": round(float(confidence), 4),
"evidence_note": note.strip(),
"review_prompt": normalized_review_prompt,
}
def build_e30_engineering_generation(
*,
materialization_root: Path,
review_pack_root: Path,
review_sheets_root: Path,
decisions_path: Path,
output_root: Path,
producer_id: str,
method_id: str,
) -> dict[str, Any]:
"""Validate 486 decisions and publish one immutable engineering generation."""
if (
_SAFE_PRODUCER.fullmatch(producer_id) is None
or _SAFE_PRODUCER.fullmatch(method_id) is None
):
raise E30EngineeringGenerationError("producer identity is invalid")
materialization_root = materialization_root.resolve()
materialization_manifest, items = _materialization(
materialization_root
)
taxonomy = _reason_taxonomy(
review_pack_root,
expected_binding=materialization_manifest["identity"].get("review_pack"),
)
sheet_manifest, sheets_by_item = _sheets(
review_sheets_root,
materialization_root=materialization_root,
)
raw_decisions = _read_jsonl(decisions_path)
if len(raw_decisions) != len(items):
raise E30EngineeringGenerationError("engineering coverage is incomplete")
decisions = [
_validate_decision(
value=value,
item=item,
reason_taxonomy=taxonomy,
expected_sheet=sheets_by_item[str(item["item_id"])],
)
for item, value in zip(items, raw_decisions, strict=True)
]
if len({decision["item_id"] for decision in decisions}) != len(decisions):
raise E30EngineeringGenerationError("engineering decision ids repeat")
source = {
"materialization_id": materialization_root.name,
"materialization_identity_sha256": materialization_manifest[
"identity_sha256"
],
"materialization_manifest_sha256": _sha256(
materialization_root / "manifest.json"
),
"materialization_index_sha256": _sha256(
materialization_root / "materialized-items.jsonl"
),
"review_pack_id": materialization_manifest["identity"]["review_pack"][
"result_id"
],
"item_count": len(items),
}
producer = {
"kind": "ai-assisted-engineering-review",
"producer_id": producer_id,
"method_id": method_id,
"review_sheet_identity_sha256": sheet_manifest["identity_sha256"],
"claims_human_ground_truth": False,
}
decisions_input_sha256 = _sha256(decisions_path)
identity = {
"schema_version": E30_ENGINEERING_GENERATION_SCHEMA,
"source": source,
"producer": producer,
"decisions_input_sha256": decisions_input_sha256,
"decision_content_sha256": hashlib.sha256(
b"".join(_canonical_json(value) + b"\n" for value in decisions)
).hexdigest(),
}
identity_sha256 = hashlib.sha256(_canonical_json(identity)).hexdigest()
generation_id = f"e30-engineering-generation-{identity_sha256}"
if _GENERATION_ID.fullmatch(generation_id) is None:
raise AssertionError("generated E30 identity is invalid")
output_root.mkdir(parents=True, exist_ok=True)
output_root = output_root.resolve()
destination = output_root / generation_id
if destination.exists():
existing = _read_json(destination / E30_ENGINEERING_MANIFEST_NAME)
if (
destination.is_symlink()
or not destination.is_dir()
or existing.get("identity") != identity
or existing.get("identity_sha256") != identity_sha256
):
raise E30EngineeringGenerationError("generation identity collision")
return existing
verdicts = Counter(str(value["verdict"]) for value in decisions)
causes = Counter(
str(value["cause_code"])
for value in decisions
if value.get("cause_code") is not None
)
detector = Counter(str(value["detector_assessment"]) for value in decisions)
projection = Counter(str(value["projection_assessment"]) for value in decisions)
ownership = Counter(str(value["point_ownership"]) for value in decisions)
exceptions = [value for value in decisions if value["human_exception_required"]]
summary = {
"schema_version": E30_ENGINEERING_SUMMARY_SCHEMA,
"item_count": len(decisions),
"reviewed_item_count": len(decisions),
"verdict_distribution": dict(sorted(verdicts.items())),
"detector_distribution": dict(sorted(detector.items())),
"projection_distribution": dict(sorted(projection.items())),
"point_ownership_distribution": dict(sorted(ownership.items())),
"human_exception_count": len(exceptions),
"mean_confidence": round(
sum(float(value["confidence"]) for value in decisions) / len(decisions),
4,
),
}
cause_document = {
"schema_version": E30_ENGINEERING_CAUSES_SCHEMA,
"item_count_with_cause": sum(causes.values()),
"reasons": [
{"reason_code": reason, "count": count}
for reason, count in sorted(causes.items())
],
}
staging = Path(tempfile.mkdtemp(prefix=f".{generation_id}.", dir=output_root))
try:
decisions_output = staging / E30_ENGINEERING_DECISIONS_NAME
with decisions_output.open("wb") as stream:
for decision in decisions:
stream.write(_canonical_json(decision) + b"\n")
stream.flush()
os.fsync(stream.fileno())
exception_output = staging / E30_ENGINEERING_EXCEPTIONS_NAME
with exception_output.open("wb") as stream:
for decision in exceptions:
exception = {
"schema_version": E30_ENGINEERING_EXCEPTION_SCHEMA,
"sequence": decision["sequence"],
"item_id": decision["item_id"],
"review_key": decision["review_key"],
"source_stratum": decision["source_stratum"],
"exception_reason": decision["exception_reason"],
"confidence": decision["confidence"],
"review_sheet": decision["review_sheet"],
"evidence_note": decision["evidence_note"],
"review_prompt": decision["review_prompt"],
}
stream.write(_canonical_json(exception) + b"\n")
stream.flush()
os.fsync(stream.fileno())
write_json_atomic(staging / E30_ENGINEERING_SUMMARY_NAME, summary)
write_json_atomic(staging / E30_ENGINEERING_CAUSES_NAME, cause_document)
artifacts = [
{
"role": "engineering-decisions",
"path": E30_ENGINEERING_DECISIONS_NAME,
},
{
"role": "human-exceptions",
"path": E30_ENGINEERING_EXCEPTIONS_NAME,
},
{
"role": "engineering-summary",
"path": E30_ENGINEERING_SUMMARY_NAME,
},
{
"role": "cause-distribution",
"path": E30_ENGINEERING_CAUSES_NAME,
},
]
artifact_documents = []
for artifact in artifacts:
path = staging / artifact["path"]
artifact_documents.append(
{
**artifact,
"sha256": _sha256(path),
"byte_length": path.stat().st_size,
}
)
created_at_utc = utc_now_iso()
manifest = {
"schema_version": E30_ENGINEERING_GENERATION_SCHEMA,
"result_id": generation_id,
"identity": identity,
"identity_sha256": identity_sha256,
"created_at_utc": created_at_utc,
"summary": summary,
"cause_distribution": cause_document,
"artifacts": artifact_documents,
"ai_review_complete": True,
"human_exception_complete": len(exceptions) == 0,
"human_review_complete": False,
"lab_published": False,
"authority": {
"commands_enabled": False,
"navigation_or_safety_accepted": False,
},
}
write_json_atomic(staging / E30_ENGINEERING_MANIFEST_NAME, manifest)
staging.rename(destination)
return manifest
except Exception:
shutil.rmtree(staging, ignore_errors=True)
raise
+759
View File
@@ -0,0 +1,759 @@
from __future__ import annotations
import hashlib
import json
import os
import re
import shutil
import tempfile
from collections import Counter
from collections.abc import Iterator
from contextlib import contextmanager
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Final, Literal
from k1link.artifacts import utc_now_iso, write_json_atomic
E30_HUMAN_REVIEW_DRAFT_SCHEMA: Final = "missioncore.e30-human-review-draft/v2"
E30_HUMAN_REVIEW_EVENT_SCHEMA: Final = "missioncore.e30-human-review-event/v2"
E30_HUMAN_REVIEW_DECISION_SCHEMA: Final = (
"missioncore.e30-human-review-decision/v2"
)
E30_HUMAN_REVIEW_GENERATION_SCHEMA: Final = (
"missioncore.e30-human-review-generation/v2"
)
E30_HUMAN_REVIEW_FINALIZATION_SCHEMA: Final = (
"missioncore.e30-human-review-finalization/v2"
)
E30_HUMAN_REVIEW_DECISIONS_NAME: Final = "review-decisions.jsonl"
E30_HUMAN_REVIEW_DISTRIBUTION_NAME: Final = "disposition-distribution.json"
E30_HUMAN_REVIEW_MANIFEST_NAME: Final = "manifest.json"
E30_STRATA: Final = (
"conflict",
"agree",
"camera-only",
"unknown",
"geometry-only",
)
E30_DISPOSITIONS: Final = (
"object-present",
"background-or-noise",
"insufficient-evidence",
)
E30Stratum = Literal[
"conflict",
"agree",
"camera-only",
"unknown",
"geometry-only",
]
E30ExceptionDisposition = Literal[
"object-present",
"background-or-noise",
"insufficient-evidence",
]
_DRAFT_ID = re.compile(r"^e30-human-draft-[a-f0-9]{64}$")
_GENERATION_ID = re.compile(r"^e30-review-generation-[a-f0-9]{64}$")
_EVENT_ID = re.compile(r"^e30-review-event-[a-f0-9]{64}$")
_CONTENT_ID = re.compile(r"^[a-z0-9][a-z0-9-]*-[a-f0-9]{64}$")
_SHA256 = re.compile(r"^[a-f0-9]{64}$")
_REVIEWER_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:@-]{0,127}$")
_IDEMPOTENCY_KEY = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$")
_MAX_JSON_BYTES: Final = 512 * 1024
_MAX_EVENT_LOG_BYTES: Final = 32 * 1024 * 1024
_MAX_NOTES_LENGTH: Final = 2_000
class E30HumanReviewError(RuntimeError):
"""Base error for the exception-review lifecycle."""
class E30HumanReviewNotFoundError(E30HumanReviewError):
"""The requested draft or generation does not exist."""
class E30HumanReviewConflictError(E30HumanReviewError):
"""The request conflicts with the current append-only revision."""
class E30HumanReviewValidationError(E30HumanReviewError):
"""A reviewer decision violates the frozen review protocol."""
class E30HumanReviewIntegrityError(E30HumanReviewError):
"""Stored reviewer evidence is incomplete, changed or incompatible."""
@dataclass(frozen=True)
class E30ReviewSubject:
item_id: str
sequence: int
source_stratum: E30Stratum
@dataclass(frozen=True)
class E30ReviewSubstrate:
materialization_id: str
materialization_identity_sha256: str
review_pack_id: str
review_items_sha256: str
reason_taxonomy: tuple[str, ...]
subjects: tuple[E30ReviewSubject, ...]
engineering_generation_id: str | None = None
def __post_init__(self) -> None:
if (
_CONTENT_ID.fullmatch(self.materialization_id) is None
or _CONTENT_ID.fullmatch(self.review_pack_id) is None
or _SHA256.fullmatch(self.materialization_identity_sha256) is None
or _SHA256.fullmatch(self.review_items_sha256) is None
or (
self.engineering_generation_id is not None
and _CONTENT_ID.fullmatch(self.engineering_generation_id) is None
)
):
raise E30HumanReviewValidationError("E30 source binding is invalid")
if not self.subjects:
raise E30HumanReviewValidationError("E30 review substrate is empty")
if tuple(subject.sequence for subject in self.subjects) != tuple(
range(len(self.subjects))
):
raise E30HumanReviewValidationError("E30 subjects are out of sequence")
if len({subject.item_id for subject in self.subjects}) != len(self.subjects):
raise E30HumanReviewValidationError("E30 subject ids are not unique")
if any(
subject.source_stratum not in E30_STRATA
or _CONTENT_ID.fullmatch(subject.item_id) is None
for subject in self.subjects
):
raise E30HumanReviewValidationError("E30 subject is invalid")
@property
def item_set_sha256(self) -> str:
return hashlib.sha256(
_canonical_json(
[
{
"item_id": subject.item_id,
"sequence": subject.sequence,
"source_stratum": subject.source_stratum,
}
for subject in self.subjects
]
)
).hexdigest()
def binding(self) -> dict[str, object]:
return {
"materialization_id": self.materialization_id,
"materialization_identity_sha256": self.materialization_identity_sha256,
"review_pack_id": self.review_pack_id,
"review_items_sha256": self.review_items_sha256,
"engineering_generation_id": self.engineering_generation_id,
"item_count": len(self.subjects),
"item_set_sha256": self.item_set_sha256,
}
def _canonical_json(value: object) -> bytes:
return json.dumps(
value,
ensure_ascii=False,
sort_keys=True,
separators=(",", ":"),
allow_nan=False,
).encode("utf-8")
def _sha256(path: Path) -> str:
digest = hashlib.sha256()
with path.open("rb") as stream:
while chunk := stream.read(1024 * 1024):
digest.update(chunk)
return digest.hexdigest()
def _read_json(path: Path) -> dict[str, Any]:
if (
path.is_symlink()
or not path.is_file()
or not 0 < path.stat().st_size <= _MAX_JSON_BYTES
):
raise E30HumanReviewIntegrityError("E30 human-review JSON is unavailable")
try:
value = json.loads(path.read_text(encoding="utf-8"))
except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc:
raise E30HumanReviewIntegrityError("E30 human-review JSON is invalid") from exc
if not isinstance(value, dict):
raise E30HumanReviewIntegrityError("E30 human-review JSON must be an object")
return value
def _safe_root(root: Path) -> Path:
root.mkdir(parents=True, exist_ok=True)
if root.is_symlink() or not root.is_dir():
raise E30HumanReviewIntegrityError("E30 human-review root is invalid")
return root.resolve()
def _normalized_reviewer_id(value: str) -> str:
value = value.strip()
if _REVIEWER_ID.fullmatch(value) is None:
raise E30HumanReviewValidationError("reviewer_id is invalid")
return value
def _normalized_notes(value: str | None) -> str | None:
if value is None:
return None
value = value.strip()
if len(value) > _MAX_NOTES_LENGTH:
raise E30HumanReviewValidationError("review notes are too long")
return value or None
@contextmanager
def _exclusive_lock(path: Path) -> Iterator[None]:
path.parent.mkdir(parents=True, exist_ok=True)
descriptor = os.open(path, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600)
try:
yield
finally:
os.close(descriptor)
path.unlink(missing_ok=True)
class E30HumanReviewStore:
"""Append-only human decisions over AI-selected exception items."""
def __init__(self, *, draft_root: Path, generation_root: Path) -> None:
self.draft_root = _safe_root(draft_root)
self.generation_root = _safe_root(generation_root)
def create_or_resume(
self,
*,
substrate: E30ReviewSubstrate,
reviewer_id: str,
) -> dict[str, object]:
if substrate.engineering_generation_id is None:
raise E30HumanReviewValidationError(
"engineering generation binding is required"
)
reviewer_id = _normalized_reviewer_id(reviewer_id)
identity = {
"schema_version": E30_HUMAN_REVIEW_DRAFT_SCHEMA,
"protocol": "exception-disposition-review/v1",
"source": substrate.binding(),
"reviewer_id": reviewer_id,
}
identity_sha256 = hashlib.sha256(_canonical_json(identity)).hexdigest()
draft_id = f"e30-human-draft-{identity_sha256}"
root = self.draft_root / draft_id
with _exclusive_lock(self.draft_root / ".locks" / f"{draft_id}.lock"):
if root.is_symlink():
raise E30HumanReviewIntegrityError("E30 draft must not be a symlink")
root.mkdir(mode=0o700, parents=False, exist_ok=True)
manifest_path = root / "manifest.json"
if not manifest_path.exists():
write_json_atomic(
manifest_path,
{
"schema_version": E30_HUMAN_REVIEW_DRAFT_SCHEMA,
"draft_id": draft_id,
"identity": identity,
"identity_sha256": identity_sha256,
"created_at_utc": utc_now_iso(),
"source": substrate.binding(),
"authority": {
"commands_enabled": False,
"navigation_or_safety_accepted": False,
},
},
)
return self._summary(root, substrate)
def get(
self,
*,
draft_id: str,
substrate: E30ReviewSubstrate,
) -> dict[str, object]:
return self._summary(self._draft_root(draft_id), substrate)
def record_decision(
self,
*,
draft_id: str,
substrate: E30ReviewSubstrate,
item_id: str,
expected_revision: int,
idempotency_key: str,
disposition: E30ExceptionDisposition,
notes: str | None,
) -> dict[str, object]:
if _IDEMPOTENCY_KEY.fullmatch(idempotency_key) is None:
raise E30HumanReviewValidationError("idempotency_key is invalid")
if expected_revision < 0:
raise E30HumanReviewValidationError("expected_revision is invalid")
subject = next(
(value for value in substrate.subjects if value.item_id == item_id),
None,
)
if subject is None:
raise E30HumanReviewNotFoundError("E30 review item was not found")
decision = self._validate_decision(disposition=disposition, notes=notes)
fingerprint = hashlib.sha256(
_canonical_json({"item_id": item_id, **decision})
).hexdigest()
root = self._draft_root(draft_id)
with _exclusive_lock(self.draft_root / ".locks" / f"{draft_id}.lock"):
manifest, events, current = self._load(root, substrate)
if self._finalization(root) is not None:
raise E30HumanReviewConflictError("E30 review draft is finalized")
prior = next(
(
event
for event in events
if event.get("idempotency_key") == idempotency_key
),
None,
)
if prior is not None:
if prior.get("request_fingerprint") != fingerprint:
raise E30HumanReviewConflictError(
"idempotency_key is already bound to another decision"
)
return self._project_summary(
manifest, substrate, events, current, None
)
if len(events) != expected_revision:
raise E30HumanReviewConflictError(
f"E30 review revision is {len(events)}, not {expected_revision}"
)
superseded = current.get(item_id)
payload: dict[str, object] = {
"schema_version": E30_HUMAN_REVIEW_EVENT_SCHEMA,
"draft_id": draft_id,
"sequence": len(events) + 1,
"item_id": item_id,
"source_stratum": subject.source_stratum,
**decision,
"reviewer_id": manifest["identity"]["reviewer_id"],
"decided_at_utc": utc_now_iso(),
"idempotency_key": idempotency_key,
"request_fingerprint": fingerprint,
"supersedes_event_id": (
superseded.get("event_id") if superseded else None
),
"previous_event_sha256": (
events[-1]["event_sha256"] if events else None
),
}
event_sha256 = hashlib.sha256(_canonical_json(payload)).hexdigest()
event = {
**payload,
"event_id": f"e30-review-event-{event_sha256}",
"event_sha256": event_sha256,
}
with (root / "events.jsonl").open("ab", buffering=0) as stream:
stream.write(_canonical_json(event) + b"\n")
os.fsync(stream.fileno())
events.append(event)
current[item_id] = event
return self._project_summary(manifest, substrate, events, current, None)
def finalize(
self,
*,
draft_id: str,
substrate: E30ReviewSubstrate,
expected_revision: int,
) -> dict[str, object]:
root = self._draft_root(draft_id)
with _exclusive_lock(self.draft_root / ".locks" / f"{draft_id}.lock"):
manifest, events, current = self._load(root, substrate)
existing = self._finalization(root)
if existing is not None:
return self._generation_summary(existing, substrate)
if len(events) != expected_revision:
raise E30HumanReviewConflictError(
f"E30 review revision is {len(events)}, not {expected_revision}"
)
if len(current) != len(substrate.subjects):
raise E30HumanReviewConflictError(
"E30 review coverage is incomplete: "
f"{len(substrate.subjects) - len(current)} decisions remain"
)
generation = self._write_generation(manifest, substrate, current)
finalization = {
"schema_version": E30_HUMAN_REVIEW_FINALIZATION_SCHEMA,
"draft_id": draft_id,
"generation_id": generation["result_id"],
"generation_identity_sha256": generation["identity_sha256"],
"final_revision": len(events),
"finalized_at_utc": generation["created_at_utc"],
}
write_json_atomic(root / "finalized.json", finalization)
return self._generation_summary(finalization, substrate)
@staticmethod
def _validate_decision(
*,
disposition: object,
notes: object,
) -> dict[str, object]:
if disposition not in E30_DISPOSITIONS:
raise E30HumanReviewValidationError("review disposition is invalid")
if notes is not None and not isinstance(notes, str):
raise E30HumanReviewValidationError("review notes are invalid")
return {
"disposition": disposition,
"notes": _normalized_notes(notes),
}
def _draft_root(self, draft_id: str) -> Path:
root = self.draft_root / draft_id
if (
_DRAFT_ID.fullmatch(draft_id) is None
or root.is_symlink()
or not root.is_dir()
):
raise E30HumanReviewNotFoundError("E30 human-review draft was not found")
return root
def _summary(
self,
root: Path,
substrate: E30ReviewSubstrate,
) -> dict[str, object]:
manifest, events, current = self._load(root, substrate)
return self._project_summary(
manifest,
substrate,
events,
current,
self._finalization(root),
)
def _load(
self,
root: Path,
substrate: E30ReviewSubstrate,
) -> tuple[dict[str, Any], list[dict[str, Any]], dict[str, dict[str, Any]]]:
manifest = _read_json(root / "manifest.json")
identity = manifest.get("identity")
identity_sha256 = manifest.get("identity_sha256")
if (
manifest.get("schema_version") != E30_HUMAN_REVIEW_DRAFT_SCHEMA
or manifest.get("draft_id") != root.name
or not isinstance(identity, dict)
or not isinstance(identity_sha256, str)
or hashlib.sha256(_canonical_json(identity)).hexdigest()
!= identity_sha256
or root.name != f"e30-human-draft-{identity_sha256}"
or manifest.get("source") != substrate.binding()
or not isinstance(manifest.get("created_at_utc"), str)
or not self._diagnostic_authority(manifest.get("authority"))
):
raise E30HumanReviewIntegrityError("E30 human-review draft changed")
events_path = root / "events.jsonl"
if not events_path.exists():
return manifest, [], {}
if (
events_path.is_symlink()
or not events_path.is_file()
or events_path.stat().st_size > _MAX_EVENT_LOG_BYTES
):
raise E30HumanReviewIntegrityError("E30 event log is invalid")
subject_by_id = {value.item_id: value for value in substrate.subjects}
events: list[dict[str, Any]] = []
current: dict[str, dict[str, Any]] = {}
previous_sha256: str | None = None
with events_path.open("r", encoding="utf-8") as stream:
for expected_sequence, line in enumerate(stream, start=1):
try:
event = json.loads(line)
except json.JSONDecodeError as exc:
raise E30HumanReviewIntegrityError(
"E30 event is invalid"
) from exc
subject = subject_by_id.get(event.get("item_id"))
event_sha256 = event.get("event_sha256")
payload = {
key: value
for key, value in event.items()
if key not in {"event_id", "event_sha256"}
}
if (
event.get("schema_version") != E30_HUMAN_REVIEW_EVENT_SCHEMA
or event.get("draft_id") != root.name
or event.get("sequence") != expected_sequence
or subject is None
or event.get("source_stratum") != subject.source_stratum
or event.get("reviewer_id") != identity.get("reviewer_id")
or event.get("previous_event_sha256") != previous_sha256
or not isinstance(event_sha256, str)
or hashlib.sha256(_canonical_json(payload)).hexdigest()
!= event_sha256
or event.get("event_id")
!= f"e30-review-event-{event_sha256}"
):
raise E30HumanReviewIntegrityError("E30 event chain changed")
self._validate_decision(
disposition=event.get("disposition"),
notes=event.get("notes"),
)
superseded = current.get(subject.item_id)
if event.get("supersedes_event_id") != (
superseded.get("event_id") if superseded else None
):
raise E30HumanReviewIntegrityError(
"E30 supersession chain changed"
)
events.append(event)
current[subject.item_id] = event
previous_sha256 = event_sha256
return manifest, events, current
def _project_summary(
self,
manifest: dict[str, Any],
substrate: E30ReviewSubstrate,
events: list[dict[str, Any]],
current: dict[str, dict[str, Any]],
finalization: dict[str, Any] | None,
) -> dict[str, object]:
distribution = Counter(
str(event["disposition"]) for event in current.values()
)
decisions = [
{
"item_id": subject.item_id,
"source_stratum": subject.source_stratum,
"disposition": event["disposition"],
"notes": event["notes"],
"event_id": event["event_id"],
"decided_at_utc": event["decided_at_utc"],
}
for subject in substrate.subjects
if (event := current.get(subject.item_id)) is not None
]
return {
"schema_version": E30_HUMAN_REVIEW_DRAFT_SCHEMA,
"draft_id": manifest["draft_id"],
"materialization_id": substrate.materialization_id,
"engineering_generation_id": substrate.engineering_generation_id,
"reviewer_id": manifest["identity"]["reviewer_id"],
"created_at_utc": manifest["created_at_utc"],
"state": "finalized" if finalization else "active",
"revision": len(events),
"item_count": len(substrate.subjects),
"reviewed_item_count": len(decisions),
"remaining_item_count": len(substrate.subjects) - len(decisions),
"disposition_distribution": dict(sorted(distribution.items())),
"generation_id": (
finalization["generation_id"] if finalization else None
),
"decisions": decisions,
"lab_published": False,
"access": "read-only" if finalization else "review-write",
}
def _write_generation(
self,
manifest: dict[str, Any],
substrate: E30ReviewSubstrate,
current: dict[str, dict[str, Any]],
) -> dict[str, Any]:
decisions = [
{
"schema_version": E30_HUMAN_REVIEW_DECISION_SCHEMA,
"sequence": subject.sequence,
"item_id": subject.item_id,
"source_stratum": subject.source_stratum,
"disposition": current[subject.item_id]["disposition"],
"notes": current[subject.item_id]["notes"],
"reviewer_id": manifest["identity"]["reviewer_id"],
"decision_event_id": current[subject.item_id]["event_id"],
"decided_at_utc": current[subject.item_id]["decided_at_utc"],
}
for subject in substrate.subjects
]
identity = {
"schema_version": E30_HUMAN_REVIEW_GENERATION_SCHEMA,
"protocol": "exception-disposition-review/v1",
"source": substrate.binding(),
"reviewer_id": manifest["identity"]["reviewer_id"],
"decision_event_ids": [
decision["decision_event_id"] for decision in decisions
],
}
identity_sha256 = hashlib.sha256(_canonical_json(identity)).hexdigest()
generation_id = f"e30-review-generation-{identity_sha256}"
destination = self.generation_root / generation_id
if destination.exists():
existing = _read_json(destination / E30_HUMAN_REVIEW_MANIFEST_NAME)
if existing.get("identity") != identity:
raise E30HumanReviewIntegrityError(
"E30 reviewer generation identity collision"
)
return existing
staging = Path(
tempfile.mkdtemp(prefix=f".{generation_id}.", dir=self.generation_root)
)
try:
decisions_path = staging / E30_HUMAN_REVIEW_DECISIONS_NAME
with decisions_path.open("wb") as stream:
for decision in decisions:
stream.write(_canonical_json(decision) + b"\n")
stream.flush()
os.fsync(stream.fileno())
distribution = {
"schema_version": "missioncore.e30-human-review-dispositions/v1",
"dispositions": dict(
sorted(Counter(
str(decision["disposition"]) for decision in decisions
).items())
),
}
distribution_path = staging / E30_HUMAN_REVIEW_DISTRIBUTION_NAME
write_json_atomic(distribution_path, distribution)
created_at_utc = utc_now_iso()
generation_manifest: dict[str, Any] = {
"schema_version": E30_HUMAN_REVIEW_GENERATION_SCHEMA,
"result_id": generation_id,
"identity": identity,
"identity_sha256": identity_sha256,
"created_at_utc": created_at_utc,
"reviewer_id": manifest["identity"]["reviewer_id"],
"item_count": len(decisions),
"disposition_distribution": distribution["dispositions"],
"coverage": {
"expected_item_count": len(substrate.subjects),
"reviewed_item_count": len(decisions),
"complete": True,
},
"human_review_complete": True,
"lab_published": False,
"artifacts": [
self._artifact("review-decisions", decisions_path),
self._artifact("disposition-distribution", distribution_path),
],
"authority": {
"commands_enabled": False,
"navigation_or_safety_accepted": False,
},
}
write_json_atomic(
staging / E30_HUMAN_REVIEW_MANIFEST_NAME,
generation_manifest,
)
os.replace(staging, destination)
return generation_manifest
finally:
if staging.exists():
shutil.rmtree(staging)
@staticmethod
def _artifact(role: str, path: Path) -> dict[str, object]:
return {
"role": role,
"path": path.name,
"byte_length": path.stat().st_size,
"sha256": _sha256(path),
}
def _finalization(self, root: Path) -> dict[str, Any] | None:
path = root / "finalized.json"
if not path.exists():
return None
value = _read_json(path)
if (
value.get("schema_version") != E30_HUMAN_REVIEW_FINALIZATION_SCHEMA
or value.get("draft_id") != root.name
or _GENERATION_ID.fullmatch(str(value.get("generation_id"))) is None
or _SHA256.fullmatch(
str(value.get("generation_identity_sha256"))
) is None
):
raise E30HumanReviewIntegrityError("E30 finalization marker changed")
return value
def _generation_summary(
self,
finalization: dict[str, Any],
substrate: E30ReviewSubstrate,
) -> dict[str, object]:
generation_id = str(finalization["generation_id"])
root = self.generation_root / generation_id
if root.is_symlink() or not root.is_dir():
raise E30HumanReviewIntegrityError(
"E30 reviewer generation is unavailable"
)
manifest = _read_json(root / E30_HUMAN_REVIEW_MANIFEST_NAME)
identity = manifest.get("identity")
identity_sha256 = manifest.get("identity_sha256")
if (
manifest.get("schema_version") != E30_HUMAN_REVIEW_GENERATION_SCHEMA
or manifest.get("result_id") != generation_id
or not isinstance(identity, dict)
or not isinstance(identity_sha256, str)
or hashlib.sha256(_canonical_json(identity)).hexdigest()
!= identity_sha256
or generation_id != f"e30-review-generation-{identity_sha256}"
or identity.get("source") != substrate.binding()
or manifest.get("human_review_complete") is not True
or manifest.get("lab_published") is not False
or not self._diagnostic_authority(manifest.get("authority"))
):
raise E30HumanReviewIntegrityError("E30 reviewer generation changed")
expected = {
"review-decisions": E30_HUMAN_REVIEW_DECISIONS_NAME,
"disposition-distribution": E30_HUMAN_REVIEW_DISTRIBUTION_NAME,
}
artifacts = manifest.get("artifacts")
if not isinstance(artifacts, list) or len(artifacts) != len(expected):
raise E30HumanReviewIntegrityError("E30 generation artifacts are invalid")
for artifact in artifacts:
if not isinstance(artifact, dict):
raise E30HumanReviewIntegrityError(
"E30 generation artifact metadata is invalid"
)
path = root / str(artifact.get("path"))
if (
expected.get(str(artifact.get("role"))) != path.name
or path.is_symlink()
or not path.is_file()
or path.stat().st_size != artifact.get("byte_length")
or _sha256(path) != artifact.get("sha256")
):
raise E30HumanReviewIntegrityError(
"E30 generation artifact content changed"
)
return {
"schema_version": E30_HUMAN_REVIEW_GENERATION_SCHEMA,
"generation_id": generation_id,
"materialization_id": substrate.materialization_id,
"engineering_generation_id": substrate.engineering_generation_id,
"reviewer_id": manifest["reviewer_id"],
"created_at_utc": manifest["created_at_utc"],
"item_count": manifest["item_count"],
"disposition_distribution": manifest["disposition_distribution"],
"coverage": manifest["coverage"],
"human_review_complete": True,
"lab_published": False,
"access": "read-only",
}
@staticmethod
def _diagnostic_authority(value: object) -> bool:
return (
isinstance(value, dict)
and value.get("commands_enabled") is False
and value.get("navigation_or_safety_accepted") is False
)
File diff suppressed because it is too large Load Diff
+734
View File
@@ -0,0 +1,734 @@
"""Deterministic E30 review selection over one immutable E29 result.
The pack is a review substrate, not a human-reviewed LAB result. It binds each
selected item to the exact E29 frame and its camera/LiDAR/pose/surface source
identities, preserves the E29 observation or geometry-cluster snapshot and
leaves the reviewer decision empty. No perception threshold is changed.
"""
from __future__ import annotations
import hashlib
import json
import math
import os
import shutil
from collections import Counter
from dataclasses import asdict, dataclass
from datetime import UTC, datetime
from pathlib import Path
from typing import Any, Final
from .semantic_geometry_fusion import (
CAMERA_GEOMETRY_FRAME_SCHEMA,
CAMERA_GEOMETRY_FRAMES_NAME,
CAMERA_GEOMETRY_FUSION_SCHEMA,
CAMERA_GEOMETRY_MANIFEST_NAME,
CAMERA_GEOMETRY_REPORT_NAME,
CAMERA_GEOMETRY_REPORT_SCHEMA,
)
E30_REVIEW_PACK_SCHEMA: Final = "missioncore.e30-evidence-review-pack/v1"
E30_REVIEW_ITEM_SCHEMA: Final = "missioncore.e30-evidence-review-item/v1"
E30_REVIEW_ITEMS_NAME: Final = "review-items.jsonl"
E30_REVIEW_MANIFEST_NAME: Final = "manifest.json"
E30_REASON_TAXONOMY: Final = (
"no_lidar_observation",
"outside_lidar_support",
"outside_camera_fov",
"time_mismatch",
"pose_age",
"calibration_residual",
"surface_rejection",
"ground_leakage",
"foreground_occlusion",
"background_leakage",
"component_split",
"component_merge",
"self_points",
"sparse_support",
"semantic_mask_error",
"detector_error",
"track_identity_error",
"post_lio_motion_artifact",
"unknown",
)
_STATUS_TO_STRATUM: Final = {
"agree": "agree",
"single-source-camera": "camera-only",
"conflict": "conflict",
"unknown": "unknown",
}
class E30ReviewPackError(ValueError):
"""An E29 source or E30 review-pack derivative violates the contract."""
@dataclass(frozen=True, slots=True)
class E30ReviewSelectionProfile:
"""Bounded deterministic sample sizes for non-conflict E29 strata."""
profile_id: str = "e30-e29-stratified-review/v1"
agree_maximum: int = 96
camera_only_maximum: int = 128
unknown_maximum: int = 96
geometry_only_maximum: int = 128
temporal_bins: int = 12
def __post_init__(self) -> None:
if (
not self.profile_id
or len(self.profile_id) > 160
or not 1 <= self.agree_maximum <= 4096
or not 1 <= self.camera_only_maximum <= 4096
or not 1 <= self.unknown_maximum <= 4096
or not 1 <= self.geometry_only_maximum <= 4096
or not 2 <= self.temporal_bins <= 128
):
raise E30ReviewPackError("E30 review selection profile is invalid")
def to_dict(self) -> dict[str, object]:
return {
"schema_version": "missioncore.e30-review-selection-profile/v1",
**asdict(self),
"conflict_selection": "all",
"sample_selection": "round-robin-class-range-time-strata-sha256",
}
def maximum_for(self, stratum: str) -> int:
values = {
"agree": self.agree_maximum,
"camera-only": self.camera_only_maximum,
"unknown": self.unknown_maximum,
"geometry-only": self.geometry_only_maximum,
}
try:
return values[stratum]
except KeyError as exc:
raise E30ReviewPackError("E30 review stratum is unknown") from exc
DEFAULT_E30_REVIEW_SELECTION_PROFILE: Final = E30ReviewSelectionProfile()
@dataclass(frozen=True, slots=True)
class E30ReviewPack:
result_root: Path
result_id: str
manifest: dict[str, Any]
@dataclass(frozen=True, slots=True)
class _ReviewCandidate:
review_key: str
stratum: str
group_key: tuple[str, ...]
document: dict[str, object]
def build_e30_review_pack(
*,
e29_result_root: Path,
output_root: Path,
profile: E30ReviewSelectionProfile = DEFAULT_E30_REVIEW_SELECTION_PROFILE,
) -> E30ReviewPack:
"""Build one immutable, bounded selection pack over a complete E29 result."""
source_root = e29_result_root.expanduser()
if source_root.is_symlink():
raise E30ReviewPackError("E29 result root must not be a symlink")
source_root = source_root.resolve(strict=True)
if not source_root.is_dir():
raise E30ReviewPackError("E29 result root must be a directory")
source_manifest_path = _regular_file(
source_root,
CAMERA_GEOMETRY_MANIFEST_NAME,
)
source_manifest = _read_json(source_manifest_path, "E29 manifest")
if source_manifest.get("schema_version") != CAMERA_GEOMETRY_FUSION_SCHEMA:
raise E30ReviewPackError("E29 manifest schema is incompatible")
result_id = _required_string(source_manifest, "result_id")
if not result_id.startswith("e29-camera-geometry-"):
raise E30ReviewPackError("E29 result id is incompatible")
if source_root.name != result_id:
raise E30ReviewPackError("E29 result directory and manifest id disagree")
if source_manifest.get("ground_truth") is not False:
raise E30ReviewPackError("E29 diagnostic result must not claim ground truth")
identity = _required_object(source_manifest, "identity")
source_identity_sha256 = _required_string(source_manifest, "identity_sha256")
if (
hashlib.sha256(_canonical_json(identity)).hexdigest()
!= source_identity_sha256
or result_id != f"e29-camera-geometry-{source_identity_sha256}"
):
raise E30ReviewPackError("E29 manifest identity digest is invalid")
_reject_authority(_required_object(identity, "authority"), "E29 identity")
frame_record = _artifact_record(source_manifest, "camera-geometry-frames")
report_record = _artifact_record(source_manifest, "camera-geometry-report")
frames_path = _verified_artifact(
source_root,
frame_record,
CAMERA_GEOMETRY_FRAMES_NAME,
)
report_path = _verified_artifact(
source_root,
report_record,
CAMERA_GEOMETRY_REPORT_NAME,
)
report = _read_json(report_path, "E29 report")
_validate_report(report, result_id, identity)
frame_count = _required_int(identity, "frame_count")
timeline_start = _required_float(identity, "timeline_start_seconds")
timeline_end = _required_float(identity, "timeline_end_seconds")
if frame_count < 1 or timeline_end < timeline_start:
raise E30ReviewPackError("E29 identity timeline is invalid")
source_bindings: dict[str, object] = {
"e29_result_id": result_id,
"e29_identity_sha256": source_identity_sha256,
"e29_frames_sha256": _required_string(frame_record, "sha256"),
"e29_report_sha256": _required_string(report_record, "sha256"),
"camera_result_id": _required_string(identity, "source_result_id"),
"lidar_pack_id": _required_string(identity, "source_pack_id"),
"local_surface_model_id": _required_string(
identity,
"local_surface_model_id",
),
"fusion_profile_id": _required_string(
_required_object(identity, "profile"),
"profile_id",
),
}
candidates, observed_counts = _read_candidates(
frames_path,
source_bindings=source_bindings,
expected_frame_count=frame_count,
timeline_start=timeline_start,
timeline_end=timeline_end,
temporal_bins=profile.temporal_bins,
)
_validate_observed_counts(report, observed_counts)
profile_document = profile.to_dict()
identity_document = {
"schema_version": E30_REVIEW_PACK_SCHEMA,
"source": source_bindings,
"selection_profile": profile_document,
"reason_taxonomy": list(E30_REASON_TAXONOMY),
"human_review_complete": False,
"producer_sha256": _sha256_file(Path(__file__).resolve(strict=True)),
"authority": {
"commands_enabled": False,
"navigation_or_safety_accepted": False,
},
}
identity_sha256 = hashlib.sha256(_canonical_json(identity_document)).hexdigest()
review_result_id = f"e30-review-pack-{identity_sha256}"
selected = _select_candidates(candidates, profile, identity_sha256)
selected_counts = Counter(candidate.stratum for candidate in selected)
output = output_root.expanduser().absolute()
output.mkdir(mode=0o700, parents=True, exist_ok=True)
result_root = output / review_result_id
if result_root.exists():
manifest = _read_existing_pack(result_root, identity_document)
return E30ReviewPack(result_root, review_result_id, manifest)
staging = output / f".{review_result_id}.{os.getpid()}.incomplete"
staging.mkdir(mode=0o700, exist_ok=False)
items_path = staging / E30_REVIEW_ITEMS_NAME
try:
with items_path.open("x", encoding="utf-8") as stream:
for sequence, candidate in enumerate(selected):
document = {
"schema_version": E30_REVIEW_ITEM_SCHEMA,
"sequence": sequence,
"item_id": _item_id(identity_sha256, candidate.review_key),
**candidate.document,
"review": {
"state": "unreviewed",
"reason_code": None,
"notes": None,
},
"authority": {
"commands_enabled": False,
"navigation_or_safety_accepted": False,
},
}
stream.write(_canonical_json(document).decode("utf-8") + "\n")
items_artifact = _artifact("review-items", items_path, "application/x-ndjson")
manifest = {
"schema_version": E30_REVIEW_PACK_SCHEMA,
"result_id": review_result_id,
"identity_sha256": identity_sha256,
"identity": identity_document,
"created_at_utc": datetime.now(UTC).isoformat(),
"classification": "private-derived-perception-review-substrate",
"human_review_complete": False,
"lab_published": False,
"source_counts": dict(sorted(observed_counts.items())),
"selected_counts": dict(sorted(selected_counts.items())),
"selected_item_count": len(selected),
"artifacts": [items_artifact],
"limitations": [
"the pack contains no human decisions",
"E29 stores support counts but not point-index ownership",
"selected and rejected point visualization requires source-bound reprojection",
"selection is not object-detection or association accuracy",
],
"authority": {
"commands_enabled": False,
"navigation_or_safety_accepted": False,
},
}
_write_json(staging / E30_REVIEW_MANIFEST_NAME, manifest)
os.replace(staging, result_root)
except BaseException:
shutil.rmtree(staging, ignore_errors=True)
raise
return E30ReviewPack(result_root, review_result_id, manifest)
def _read_candidates(
frames_path: Path,
*,
source_bindings: dict[str, object],
expected_frame_count: int,
timeline_start: float,
timeline_end: float,
temporal_bins: int,
) -> tuple[list[_ReviewCandidate], Counter[str]]:
candidates: list[_ReviewCandidate] = []
counts: Counter[str] = Counter()
observed_frame_count = 0
with frames_path.open("r", encoding="utf-8") as stream:
for expected_frame_index, line in enumerate(stream):
observed_frame_count += 1
try:
value = json.loads(line)
except json.JSONDecodeError as exc:
raise E30ReviewPackError("E29 frame JSON is invalid") from exc
frame = _object(value, "E29 frame")
if frame.get("schema_version") != CAMERA_GEOMETRY_FRAME_SCHEMA:
raise E30ReviewPackError("E29 frame schema is incompatible")
frame_index = _required_int(frame, "frame_index")
if frame_index != expected_frame_index:
raise E30ReviewPackError("E29 frame order is not contiguous")
source_frame_index = _required_int(frame, "source_frame_index")
if source_frame_index < 0:
raise E30ReviewPackError("E29 source frame index is invalid")
session_seconds = _required_float(frame, "session_seconds")
temporal_bin = _temporal_bin(
session_seconds,
timeline_start,
timeline_end,
temporal_bins,
)
binding = {
**source_bindings,
"frame_index": frame_index,
"source_frame_index": source_frame_index,
"session_seconds": session_seconds,
}
observations = _required_array(frame, "semantic_observations")
for observation_index, raw_observation in enumerate(observations):
observation = _object(raw_observation, "E29 semantic observation")
status = _required_string(observation, "geometry_status")
try:
stratum = _STATUS_TO_STRATUM[status]
except KeyError as exc:
raise E30ReviewPackError(
"E29 semantic geometry status is incompatible"
) from exc
counts[stratum] += 1
label = _required_string(observation, "label")
association_group = _required_string(
observation,
"association_group",
)
range_bucket = _range_bucket(
_optional_float(observation.get("range_m"))
)
review_key = f"semantic:{frame_index}:{observation_index}"
candidates.append(
_ReviewCandidate(
review_key=review_key,
stratum=stratum,
group_key=(
association_group,
label,
range_bucket,
f"time-{temporal_bin:03d}",
),
document={
"review_key": review_key,
"stratum": stratum,
"range_bucket": range_bucket,
"evidence_binding": binding,
"e29_locator": {
"kind": "semantic-observation",
"observation_index": observation_index,
},
"e29_snapshot": observation,
"materialization": {
"camera_frame_bound": True,
"lidar_frame_bound": True,
"pose_frame_bound": True,
"surface_frame_bound": True,
"point_indices_available_in_e29": False,
"source_reprojection_required": True,
},
},
)
)
geometry_clusters = _required_array(frame, "geometry_only_occupied")
for cluster_index, raw_cluster in enumerate(geometry_clusters):
cluster = _object(raw_cluster, "E29 geometry-only cluster")
if cluster.get("geometry_status") != "single-source-geometry":
raise E30ReviewPackError(
"E29 geometry-only status is incompatible"
)
counts["geometry-only"] += 1
range_bucket = _range_bucket(
_optional_float(cluster.get("nearest_range_m"))
)
review_key = f"geometry:{frame_index}:{cluster_index}"
candidates.append(
_ReviewCandidate(
review_key=review_key,
stratum="geometry-only",
group_key=(
range_bucket,
f"time-{temporal_bin:03d}",
),
document={
"review_key": review_key,
"stratum": "geometry-only",
"range_bucket": range_bucket,
"evidence_binding": binding,
"e29_locator": {
"kind": "geometry-only-cluster",
"cluster_index": cluster_index,
},
"e29_snapshot": cluster,
"materialization": {
"camera_frame_bound": True,
"lidar_frame_bound": True,
"pose_frame_bound": True,
"surface_frame_bound": True,
"point_indices_available_in_e29": False,
"source_reprojection_required": True,
},
},
)
)
if observed_frame_count != expected_frame_count:
raise E30ReviewPackError("E29 frame artifact is incomplete")
return candidates, counts
def _select_candidates(
candidates: list[_ReviewCandidate],
profile: E30ReviewSelectionProfile,
seed: str,
) -> list[_ReviewCandidate]:
by_stratum: dict[str, list[_ReviewCandidate]] = {}
for candidate in candidates:
by_stratum.setdefault(candidate.stratum, []).append(candidate)
admitted_strata = {
"agree",
"camera-only",
"conflict",
"unknown",
"geometry-only",
}
if not set(by_stratum).issubset(admitted_strata):
raise E30ReviewPackError("E29 result contains an unknown E30 review stratum")
selected = list(by_stratum.get("conflict", []))
for stratum in ("agree", "camera-only", "unknown", "geometry-only"):
selected.extend(
_round_robin_sample(
by_stratum.get(stratum, []),
profile.maximum_for(stratum),
seed,
)
)
selected.sort(key=lambda candidate: candidate.review_key)
if len({candidate.review_key for candidate in selected}) != len(selected):
raise E30ReviewPackError("E30 selection contains duplicate review items")
return selected
def _round_robin_sample(
candidates: list[_ReviewCandidate],
maximum: int,
seed: str,
) -> list[_ReviewCandidate]:
if len(candidates) <= maximum:
return list(candidates)
groups: dict[tuple[str, ...], list[_ReviewCandidate]] = {}
for candidate in candidates:
groups.setdefault(candidate.group_key, []).append(candidate)
for group in groups.values():
group.sort(key=lambda candidate: _selection_rank(seed, candidate.review_key))
selected: list[_ReviewCandidate] = []
active = sorted(groups)
while active and len(selected) < maximum:
next_active: list[tuple[str, ...]] = []
for key in active:
group = groups[key]
if group and len(selected) < maximum:
selected.append(group.pop(0))
if group:
next_active.append(key)
active = next_active
if len(selected) != maximum:
raise E30ReviewPackError("E30 stratified selector did not fill its quota")
return selected
def _selection_rank(seed: str, review_key: str) -> str:
return hashlib.sha256(f"{seed}:{review_key}".encode()).hexdigest()
def _item_id(identity_sha256: str, review_key: str) -> str:
digest = hashlib.sha256(f"{identity_sha256}:{review_key}".encode()).hexdigest()
return f"e30-review-item-{digest}"
def _temporal_bin(
session_seconds: float,
start: float,
end: float,
bin_count: int,
) -> int:
if not start <= session_seconds <= end:
raise E30ReviewPackError("E29 frame escaped the identity timeline")
if end == start:
return 0
fraction = (session_seconds - start) / (end - start)
return min(bin_count - 1, max(0, int(fraction * bin_count)))
def _range_bucket(value: float | None) -> str:
if value is None:
return "unavailable"
if value < 3.0:
return "near"
if value < 7.0:
return "middle"
return "far"
def _validate_report(
report: dict[str, Any],
result_id: str,
identity: dict[str, Any],
) -> None:
if (
report.get("schema_version") != CAMERA_GEOMETRY_REPORT_SCHEMA
or report.get("result_id") != result_id
or report.get("status") != "diagnostic-replay-complete"
or report.get("ground_truth") is not False
or report.get("identity") != identity
):
raise E30ReviewPackError("E29 report is not a complete diagnostic replay")
_reject_authority(_required_object(report, "authority"), "E29 report")
def _validate_observed_counts(
report: dict[str, Any],
observed: Counter[str],
) -> None:
metrics = _required_object(report, "metrics")
semantic = _required_object(metrics, "semantic_observations")
statuses = _required_object(semantic, "geometry_status")
expected = {
"agree": _required_int(statuses, "agree"),
"camera-only": _required_int(statuses, "single-source-camera"),
"conflict": _required_int(statuses, "conflict"),
"unknown": _required_int(statuses, "unknown"),
"geometry-only": _required_int(
_required_object(metrics, "geometry_only_occupied"),
"cluster_count",
),
}
actual = {stratum: observed[stratum] for stratum in expected}
if actual != expected:
raise E30ReviewPackError("E29 frame and report stratum counts disagree")
def _artifact_record(
manifest: dict[str, Any],
role: str,
) -> dict[str, Any]:
artifacts = _required_array(manifest, "artifacts")
matches = [
_object(item, "E29 artifact")
for item in artifacts
if isinstance(item, dict) and item.get("role") == role
]
if len(matches) != 1:
raise E30ReviewPackError(f"E29 manifest has no unique {role} artifact")
return matches[0]
def _verified_artifact(
root: Path,
record: dict[str, Any],
expected_name: str,
) -> Path:
relative = _required_string(record, "path")
if Path(relative).is_absolute() or relative != expected_name:
raise E30ReviewPackError("E29 artifact path is incompatible")
path = _regular_file(root, relative)
if path.stat().st_size != _required_int(record, "byte_length"):
raise E30ReviewPackError("E29 artifact byte length changed")
if _sha256_file(path) != _required_string(record, "sha256"):
raise E30ReviewPackError("E29 artifact digest changed")
return path
def _regular_file(root: Path, relative: str) -> Path:
raw = root / relative
if raw.is_symlink():
raise E30ReviewPackError("evidence artifact must not be a symlink")
path = raw.resolve(strict=True)
try:
path.relative_to(root)
except ValueError as exc:
raise E30ReviewPackError("evidence artifact escaped its result root") from exc
if not path.is_file():
raise E30ReviewPackError("evidence artifact must be a regular file")
return path
def _artifact(role: str, path: Path, media_type: str) -> dict[str, object]:
return {
"role": role,
"path": path.name,
"media_type": media_type,
"byte_length": path.stat().st_size,
"sha256": _sha256_file(path),
}
def _read_existing_pack(
result_root: Path,
expected_identity: dict[str, object],
) -> dict[str, Any]:
manifest_path = _regular_file(result_root, E30_REVIEW_MANIFEST_NAME)
manifest = _read_json(manifest_path, "E30 review manifest")
if (
manifest.get("schema_version") != E30_REVIEW_PACK_SCHEMA
or manifest.get("identity") != expected_identity
or manifest.get("human_review_complete") is not False
or manifest.get("lab_published") is not False
):
raise E30ReviewPackError("existing E30 review pack identity differs")
_reject_authority(_required_object(manifest, "authority"), "E30 review pack")
artifact = _artifact_record(manifest, "review-items")
_verified_artifact(result_root, artifact, E30_REVIEW_ITEMS_NAME)
return manifest
def _reject_authority(authority: dict[str, Any], label: str) -> None:
if (
authority.get("commands_enabled") is not False
or authority.get("navigation_or_safety_accepted") is not False
):
raise E30ReviewPackError(f"{label} must remain diagnostic-only")
def _read_json(path: Path, label: str) -> dict[str, Any]:
try:
return _object(json.loads(path.read_text(encoding="utf-8")), label)
except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc:
raise E30ReviewPackError(f"{label} is invalid") from exc
def _write_json(path: Path, value: dict[str, object] | dict[str, Any]) -> None:
with path.open("x", encoding="utf-8") as stream:
stream.write(_canonical_json(value).decode("utf-8"))
stream.write("\n")
def _canonical_json(value: object) -> bytes:
return json.dumps(
value,
ensure_ascii=False,
sort_keys=True,
separators=(",", ":"),
allow_nan=False,
).encode()
def _sha256_file(path: Path) -> str:
digest = hashlib.sha256()
with path.open("rb") as stream:
while chunk := stream.read(1024 * 1024):
digest.update(chunk)
return digest.hexdigest()
def _object(value: object, label: str) -> dict[str, Any]:
if not isinstance(value, dict) or any(not isinstance(key, str) for key in value):
raise E30ReviewPackError(f"{label} must be an object")
return value
def _required_object(document: dict[str, Any], key: str) -> dict[str, Any]:
return _object(document.get(key), key)
def _required_array(document: dict[str, Any], key: str) -> list[object]:
value = document.get(key)
if not isinstance(value, list):
raise E30ReviewPackError(f"{key} must be an array")
return value
def _required_string(document: dict[str, Any], key: str) -> str:
value = document.get(key)
if not isinstance(value, str) or not value:
raise E30ReviewPackError(f"{key} must be a nonempty string")
return value
def _required_int(document: dict[str, Any], key: str) -> int:
value = document.get(key)
if not isinstance(value, int) or isinstance(value, bool):
raise E30ReviewPackError(f"{key} must be an integer")
return value
def _required_float(document: dict[str, Any], key: str) -> float:
value = document.get(key)
if not isinstance(value, int | float) or isinstance(value, bool):
raise E30ReviewPackError(f"{key} must be numeric")
parsed = float(value)
if not math.isfinite(parsed):
raise E30ReviewPackError(f"{key} must be finite")
return parsed
def _optional_float(value: object) -> float | None:
if value is None:
return None
if not isinstance(value, int | float) or isinstance(value, bool):
raise E30ReviewPackError("optional range must be numeric")
parsed = float(value)
if not math.isfinite(parsed):
raise E30ReviewPackError("optional range must be finite")
return parsed
+58 -25
View File
@@ -132,6 +132,12 @@ class _SemanticSupport:
occupied_source_indices: IntArray
@dataclass(frozen=True, slots=True)
class _GeometryClusterSupport:
document: dict[str, object]
occupied_source_indices: IntArray
def build_camera_geometry_fusion(
*,
fusion_frames_path: Path,
@@ -552,6 +558,28 @@ def _geometry_clusters(
claimed_source_indices: set[int],
profile: CameraGeometryFusionProfile,
) -> list[dict[str, object]]:
return [
support.document
for support in _geometry_cluster_supports(
points_map=points_map,
point_class=point_class,
point_height_m=point_height_m,
sensor_position_map=sensor_position_map,
claimed_source_indices=claimed_source_indices,
profile=profile,
)
]
def _geometry_cluster_supports(
*,
points_map: FloatArray,
point_class: npt.NDArray[np.uint8],
point_height_m: npt.NDArray[np.float32],
sensor_position_map: FloatArray,
claimed_source_indices: set[int],
profile: CameraGeometryFusionProfile,
) -> list[_GeometryClusterSupport]:
occupied = np.flatnonzero(point_class == POINT_OCCUPIED).astype(np.int64)
if occupied.size == 0:
return []
@@ -564,7 +592,7 @@ def _geometry_clusters(
occupied,
profile.geometry_voxel_size_m,
)
documents: list[dict[str, object]] = []
supports: list[_GeometryClusterSupport] = []
for indices, voxel_count in components:
if (
indices.size < profile.geometry_minimum_cluster_points
@@ -574,33 +602,38 @@ def _geometry_clusters(
continue
values = points_map[indices]
distances = np.linalg.norm(values - sensor_position_map, axis=1)
documents.append(
{
"geometry_status": "single-source-geometry",
"semantic_class": None,
"point_count": int(indices.size),
"voxel_count": voxel_count,
"centroid_map_xyz_m": np.median(values, axis=0).astype(np.float64).tolist(),
"bounds_map_xyz_m": [
np.min(values, axis=0).astype(np.float64).tolist(),
np.max(values, axis=0).astype(np.float64).tolist(),
],
"height_range_m": [
float(np.min(point_height_m[indices])),
float(np.max(point_height_m[indices])),
],
"nearest_range_m": float(np.min(distances)),
"unknown_is_occupied": True,
"navigation_or_safety_accepted": False,
}
supports.append(
_GeometryClusterSupport(
document={
"geometry_status": "single-source-geometry",
"semantic_class": None,
"point_count": int(indices.size),
"voxel_count": voxel_count,
"centroid_map_xyz_m": np.median(values, axis=0)
.astype(np.float64)
.tolist(),
"bounds_map_xyz_m": [
np.min(values, axis=0).astype(np.float64).tolist(),
np.max(values, axis=0).astype(np.float64).tolist(),
],
"height_range_m": [
float(np.min(point_height_m[indices])),
float(np.max(point_height_m[indices])),
],
"nearest_range_m": float(np.min(distances)),
"unknown_is_occupied": True,
"navigation_or_safety_accepted": False,
},
occupied_source_indices=indices.astype(np.int64, copy=False),
)
)
documents.sort(
key=lambda item: (
_required_float(item["nearest_range_m"]),
-_required_int(item["point_count"]),
supports.sort(
key=lambda support: (
_required_float(support.document["nearest_range_m"]),
-_required_int(support.document["point_count"]),
)
)
return documents[: profile.maximum_geometry_clusters_per_frame]
return supports[: profile.maximum_geometry_clusters_per_frame]
def _voxel_components(
+484
View File
@@ -0,0 +1,484 @@
"""Provider-neutral sensor representation capabilities and algorithm admission.
This module does not reinterpret source evidence. It records which physical
properties a representation actually provides and rejects algorithms whose
declared requirements are not satisfied. Compatibility never grants command,
navigation or safety authority.
"""
from __future__ import annotations
import re
from dataclasses import dataclass
from enum import StrEnum
from typing import Final
SENSOR_REPRESENTATION_CAPABILITIES_SCHEMA: Final = (
"missioncore.sensor-representation-capabilities/v1"
)
SENSOR_ALGORITHM_REQUIREMENTS_SCHEMA: Final = (
"missioncore.sensor-algorithm-requirements/v1"
)
SENSOR_ALGORITHM_ADMISSION_SCHEMA: Final = "missioncore.sensor-algorithm-admission/v1"
_IDENTIFIER = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:/-]{0,159}$")
class SensorRepresentationContractError(ValueError):
"""A representation, requirement or admission violates the v1 contract."""
class SensorRepresentationKind(StrEnum):
NATIVE_SENSOR_SCAN = "native-sensor-scan"
REGISTERED_MAP_INCREMENT = "registered-map-increment"
ACCUMULATED_MAP = "accumulated-map"
DEPTH_IMAGE = "depth-image"
class SensorSourceCurrentness(StrEnum):
CURRENT_OBSERVATION = "current-observation"
FRAME_INCREMENT = "frame-increment"
PERSISTENT_RECONSTRUCTION = "persistent-reconstruction"
class SensorCapability(StrEnum):
METRIC_XYZ = "metric_xyz"
METRIC_INTENSITY = "metric_intensity"
MAP_REGISTERED = "map_registered"
SENSOR_POSE_AVAILABLE = "sensor_pose_available"
PER_POINT_TIME = "per_point_time"
RING_OR_CHANNEL = "ring_or_channel"
SEPARATE_IMU = "separate_imu"
SHARED_HARDWARE_CLOCK = "shared_hardware_clock"
NATIVE_RAY_MODEL = "native_ray_model"
SENSOR_ORIGIN_PER_POINT = "sensor_origin_per_point"
RAY_CLEARING_VALID = "ray_clearing_valid"
MOTION_COMPENSATION_VALID = "motion_compensation_valid"
FREE_SPACE_EVIDENCE_VALID = "free_space_evidence_valid"
PERSISTENT_RECONSTRUCTION = "persistent_reconstruction"
@dataclass(frozen=True, slots=True)
class SensorRepresentationCapabilities:
"""State the admitted physical meaning of one versioned sensor product."""
profile_id: str
source_profile_id: str
representation_kind: SensorRepresentationKind
coordinate_frame: str
source_currentness: SensorSourceCurrentness
capabilities: frozenset[SensorCapability]
def __post_init__(self) -> None:
_safe_identifier(self.profile_id, "representation profile id")
_safe_identifier(self.source_profile_id, "source profile id")
_safe_identifier(self.coordinate_frame, "coordinate frame")
if not self.capabilities:
raise SensorRepresentationContractError(
"representation must declare at least one admitted capability"
)
capabilities = self.capabilities
point_capabilities = {
SensorCapability.METRIC_INTENSITY,
SensorCapability.MAP_REGISTERED,
SensorCapability.PER_POINT_TIME,
SensorCapability.RING_OR_CHANNEL,
SensorCapability.NATIVE_RAY_MODEL,
SensorCapability.SENSOR_ORIGIN_PER_POINT,
SensorCapability.RAY_CLEARING_VALID,
SensorCapability.MOTION_COMPENSATION_VALID,
SensorCapability.FREE_SPACE_EVIDENCE_VALID,
}
if (
capabilities.intersection(point_capabilities)
and SensorCapability.METRIC_XYZ not in capabilities
):
raise SensorRepresentationContractError(
"point-level capabilities require metric xyz"
)
if (
SensorCapability.RAY_CLEARING_VALID in capabilities
and (
SensorCapability.NATIVE_RAY_MODEL not in capabilities
or SensorCapability.SENSOR_ORIGIN_PER_POINT not in capabilities
)
):
raise SensorRepresentationContractError(
"ray clearing requires a native ray model and per-point sensor origin"
)
if (
SensorCapability.MOTION_COMPENSATION_VALID in capabilities
and (
SensorCapability.PER_POINT_TIME not in capabilities
or SensorCapability.SENSOR_POSE_AVAILABLE not in capabilities
)
):
raise SensorRepresentationContractError(
"motion compensation requires point time and sensor pose"
)
if (
SensorCapability.FREE_SPACE_EVIDENCE_VALID in capabilities
and SensorCapability.RAY_CLEARING_VALID not in capabilities
):
raise SensorRepresentationContractError(
"free-space evidence requires admitted ray clearing"
)
persistent = SensorCapability.PERSISTENT_RECONSTRUCTION in capabilities
currentness_is_persistent = (
self.source_currentness
is SensorSourceCurrentness.PERSISTENT_RECONSTRUCTION
)
if persistent is not currentness_is_persistent:
raise SensorRepresentationContractError(
"persistent reconstruction capability and currentness disagree"
)
if self.representation_kind is SensorRepresentationKind.REGISTERED_MAP_INCREMENT:
required = {
SensorCapability.METRIC_XYZ,
SensorCapability.MAP_REGISTERED,
}
if not required.issubset(capabilities):
raise SensorRepresentationContractError(
"registered map increments require metric map-registered xyz"
)
if self.source_currentness is not SensorSourceCurrentness.FRAME_INCREMENT:
raise SensorRepresentationContractError(
"registered map increments must declare frame-increment currentness"
)
elif self.representation_kind is SensorRepresentationKind.ACCUMULATED_MAP:
required = {
SensorCapability.METRIC_XYZ,
SensorCapability.MAP_REGISTERED,
SensorCapability.PERSISTENT_RECONSTRUCTION,
}
if not required.issubset(capabilities):
raise SensorRepresentationContractError(
"accumulated maps require persistent map-registered metric xyz"
)
def supports(self, capability: SensorCapability) -> bool:
return capability in self.capabilities
def to_dict(self) -> dict[str, object]:
return {
"schema_version": SENSOR_REPRESENTATION_CAPABILITIES_SCHEMA,
"profile_id": self.profile_id,
"source_profile_id": self.source_profile_id,
"representation_kind": self.representation_kind.value,
"coordinate_frame": self.coordinate_frame,
"source_currentness": self.source_currentness.value,
"capabilities": {
capability.value: capability in self.capabilities
for capability in SensorCapability
},
"semantics": {
"absence_of_endpoints_means_free": False,
"unknown_remains_unknown": True,
},
"authority": {
"compatibility_only": True,
"commands_enabled": False,
"navigation_or_safety_accepted": False,
},
}
@classmethod
def from_dict(cls, value: object) -> SensorRepresentationCapabilities:
document = _object(value, "sensor representation capabilities")
_exact_keys(
document,
{
"schema_version",
"profile_id",
"source_profile_id",
"representation_kind",
"coordinate_frame",
"source_currentness",
"capabilities",
"semantics",
"authority",
},
"sensor representation capabilities",
)
if (
document.get("schema_version")
!= SENSOR_REPRESENTATION_CAPABILITIES_SCHEMA
):
raise SensorRepresentationContractError(
"sensor representation capabilities schema is incompatible"
)
capability_document = _object(
document.get("capabilities"),
"sensor capabilities",
)
expected_capability_keys = {capability.value for capability in SensorCapability}
_exact_keys(
capability_document,
expected_capability_keys,
"sensor capabilities",
)
capabilities = frozenset(
capability
for capability in SensorCapability
if _bool(capability_document, capability.value)
)
semantics = _object(document.get("semantics"), "sensor semantics")
_exact_keys(
semantics,
{"absence_of_endpoints_means_free", "unknown_remains_unknown"},
"sensor semantics",
)
if (
_bool(semantics, "absence_of_endpoints_means_free")
or not _bool(semantics, "unknown_remains_unknown")
):
raise SensorRepresentationContractError(
"v1 sensor semantics cannot infer free space from missing endpoints"
)
authority = _object(document.get("authority"), "sensor authority")
_exact_keys(
authority,
{
"compatibility_only",
"commands_enabled",
"navigation_or_safety_accepted",
},
"sensor authority",
)
if (
not _bool(authority, "compatibility_only")
or _bool(authority, "commands_enabled")
or _bool(authority, "navigation_or_safety_accepted")
):
raise SensorRepresentationContractError(
"v1 sensor capability admission cannot grant authority"
)
try:
return cls(
profile_id=_string(document, "profile_id"),
source_profile_id=_string(document, "source_profile_id"),
representation_kind=SensorRepresentationKind(
_string(document, "representation_kind")
),
coordinate_frame=_string(document, "coordinate_frame"),
source_currentness=SensorSourceCurrentness(
_string(document, "source_currentness")
),
capabilities=capabilities,
)
except ValueError as exc:
raise SensorRepresentationContractError(
"sensor representation enum is unknown"
) from exc
@dataclass(frozen=True, slots=True)
class SensorAlgorithmRequirements:
"""Declare the minimum representation properties required by an algorithm."""
algorithm_id: str
accepted_representations: tuple[SensorRepresentationKind, ...]
required_capabilities: frozenset[SensorCapability]
def __post_init__(self) -> None:
_safe_identifier(self.algorithm_id, "algorithm id")
if not self.accepted_representations:
raise SensorRepresentationContractError(
"algorithm must declare at least one accepted representation"
)
if len(self.accepted_representations) != len(
set(self.accepted_representations)
):
raise SensorRepresentationContractError(
"accepted algorithm representations must be unique"
)
if not self.required_capabilities:
raise SensorRepresentationContractError(
"algorithm must declare at least one required capability"
)
def to_dict(self) -> dict[str, object]:
return {
"schema_version": SENSOR_ALGORITHM_REQUIREMENTS_SCHEMA,
"algorithm_id": self.algorithm_id,
"accepted_representations": [
representation.value for representation in self.accepted_representations
],
"required_capabilities": sorted(
capability.value for capability in self.required_capabilities
),
"on_capability_mismatch": "reject",
}
@classmethod
def from_dict(cls, value: object) -> SensorAlgorithmRequirements:
document = _object(value, "sensor algorithm requirements")
_exact_keys(
document,
{
"schema_version",
"algorithm_id",
"accepted_representations",
"required_capabilities",
"on_capability_mismatch",
},
"sensor algorithm requirements",
)
if document.get("schema_version") != SENSOR_ALGORITHM_REQUIREMENTS_SCHEMA:
raise SensorRepresentationContractError(
"sensor algorithm requirements schema is incompatible"
)
if document.get("on_capability_mismatch") != "reject":
raise SensorRepresentationContractError(
"sensor algorithm capability mismatch must reject"
)
try:
accepted_representations = tuple(
SensorRepresentationKind(_string_value(item, "representation kind"))
for item in _array(document, "accepted_representations")
)
required_capabilities = frozenset(
SensorCapability(_string_value(item, "sensor capability"))
for item in _array(document, "required_capabilities")
)
return cls(
algorithm_id=_string(document, "algorithm_id"),
accepted_representations=accepted_representations,
required_capabilities=required_capabilities,
)
except ValueError as exc:
raise SensorRepresentationContractError(
"sensor algorithm requirement enum is unknown"
) from exc
@dataclass(frozen=True, slots=True)
class SensorAlgorithmAdmission:
profile_id: str
algorithm_id: str
admitted: bool
reasons: tuple[str, ...]
def to_dict(self) -> dict[str, object]:
return {
"schema_version": SENSOR_ALGORITHM_ADMISSION_SCHEMA,
"profile_id": self.profile_id,
"algorithm_id": self.algorithm_id,
"admitted": self.admitted,
"reasons": list(self.reasons),
"authority": {
"compatibility_only": True,
"commands_enabled": False,
"navigation_or_safety_accepted": False,
},
}
def assess_sensor_algorithm(
profile: SensorRepresentationCapabilities,
requirements: SensorAlgorithmRequirements,
) -> SensorAlgorithmAdmission:
"""Return a deterministic binary admission without inferring capabilities."""
reasons: list[str] = []
if profile.representation_kind not in requirements.accepted_representations:
reasons.append(
f"representation-not-accepted:{profile.representation_kind.value}"
)
missing_capabilities = requirements.required_capabilities.difference(
profile.capabilities
)
reasons.extend(
f"missing-capability:{capability.value}"
for capability in sorted(missing_capabilities, key=lambda item: item.value)
)
return SensorAlgorithmAdmission(
profile_id=profile.profile_id,
algorithm_id=requirements.algorithm_id,
admitted=not reasons,
reasons=tuple(reasons),
)
def require_sensor_algorithm_admission(
profile: SensorRepresentationCapabilities,
requirements: SensorAlgorithmRequirements,
) -> SensorAlgorithmAdmission:
"""Return admission or reject the algorithm/representation pairing."""
admission = assess_sensor_algorithm(profile, requirements)
if not admission.admitted:
raise SensorRepresentationContractError(
f"{requirements.algorithm_id} rejected for {profile.profile_id}: "
+ ", ".join(admission.reasons)
)
return admission
def _safe_identifier(value: str, label: str) -> str:
if not isinstance(value, str) or _IDENTIFIER.fullmatch(value) is None:
raise SensorRepresentationContractError(f"{label} is not a safe identifier")
return value
def _object(value: object, label: str) -> dict[str, object]:
if not isinstance(value, dict) or any(not isinstance(key, str) for key in value):
raise SensorRepresentationContractError(f"{label} must be an object")
return value
def _exact_keys(
document: dict[str, object],
expected: set[str],
label: str,
) -> None:
if set(document) != expected:
raise SensorRepresentationContractError(f"{label} fields are incompatible")
def _array(document: dict[str, object], key: str) -> list[object]:
value = document.get(key)
if not isinstance(value, list):
raise SensorRepresentationContractError(f"{key} must be an array")
return value
def _string(document: dict[str, object], key: str) -> str:
return _string_value(document.get(key), key)
def _string_value(value: object, label: str) -> str:
if not isinstance(value, str) or not value:
raise SensorRepresentationContractError(f"{label} must be a nonempty string")
return value
def _bool(document: dict[str, object], key: str) -> bool:
value = document.get(key)
if not isinstance(value, bool):
raise SensorRepresentationContractError(f"{key} must be a boolean")
return value
K1_LIO_PCL_CAPABILITIES: Final = SensorRepresentationCapabilities(
profile_id="xgrids-k1-live-lio-pcl-capabilities/v1",
source_profile_id="xgrids-k1-live-lio-pcl/v1",
representation_kind=SensorRepresentationKind.REGISTERED_MAP_INCREMENT,
coordinate_frame="map",
source_currentness=SensorSourceCurrentness.FRAME_INCREMENT,
capabilities=frozenset(
{
SensorCapability.METRIC_XYZ,
SensorCapability.METRIC_INTENSITY,
SensorCapability.MAP_REGISTERED,
SensorCapability.SENSOR_POSE_AVAILABLE,
}
),
)
+85
View File
@@ -33,6 +33,9 @@ from k1link.sessions import (
SessionStore,
)
from k1link.web.device_plugin_composition import load_installed_device_plugins
from k1link.web.e30_engineering_api import build_e30_engineering_router
from k1link.web.e30_human_review_api import build_e30_human_review_router
from k1link.web.e30_review_api import build_e30_review_router
from k1link.web.environment_api import build_environment_router
from k1link.web.laboratory_api import build_laboratory_router
from k1link.web.lidar_api import build_lidar_router
@@ -493,6 +496,88 @@ app.include_router(
),
)
)
app.include_router(
build_e30_review_router(
materialization_root_provider=lambda: (
REPOSITORY_ROOT
/ ".runtime"
/ "compute-experiments"
/ "e30"
/ "materializations"
),
review_pack_root_provider=lambda: (
REPOSITORY_ROOT
/ ".runtime"
/ "compute-experiments"
/ "e30"
/ "review-packs"
),
)
)
app.include_router(
build_e30_engineering_router(
generation_root_provider=lambda: (
REPOSITORY_ROOT
/ ".runtime"
/ "compute-experiments"
/ "e30"
/ "engineering-generations"
),
materialization_root_provider=lambda: (
REPOSITORY_ROOT
/ ".runtime"
/ "compute-experiments"
/ "e30"
/ "materializations"
),
review_pack_root_provider=lambda: (
REPOSITORY_ROOT
/ ".runtime"
/ "compute-experiments"
/ "e30"
/ "review-packs"
),
)
)
app.include_router(
build_e30_human_review_router(
materialization_root_provider=lambda: (
REPOSITORY_ROOT
/ ".runtime"
/ "compute-experiments"
/ "e30"
/ "materializations"
),
review_pack_root_provider=lambda: (
REPOSITORY_ROOT
/ ".runtime"
/ "compute-experiments"
/ "e30"
/ "review-packs"
),
engineering_generation_root_provider=lambda: (
REPOSITORY_ROOT
/ ".runtime"
/ "compute-experiments"
/ "e30"
/ "engineering-generations"
),
draft_root_provider=lambda: (
REPOSITORY_ROOT
/ ".runtime"
/ "compute-experiments"
/ "e30"
/ "human-review-drafts"
),
generation_root_provider=lambda: (
REPOSITORY_ROOT
/ ".runtime"
/ "compute-experiments"
/ "e30"
/ "human-review-generations"
),
)
)
frontend_dist = REPOSITORY_ROOT / "apps" / "control-station" / "dist"
+690
View File
@@ -0,0 +1,690 @@
from __future__ import annotations
import copy
import hashlib
import json
import math
import re
from collections import Counter
from collections.abc import Callable
from functools import lru_cache
from pathlib import Path
from typing import Any, Final
from fastapi import APIRouter, HTTPException, Query
from k1link.compute.e30_engineering_generation import (
E30_ENGINEERING_CAUSES_SCHEMA,
E30_ENGINEERING_DECISION_SCHEMA,
E30_ENGINEERING_DECISION_SCHEMAS,
E30_ENGINEERING_EXCEPTION_SCHEMA,
E30_ENGINEERING_EXCEPTION_SCHEMAS,
E30_ENGINEERING_GENERATION_SCHEMA,
E30_ENGINEERING_SUMMARY_SCHEMA,
EXCEPTION_DISPOSITIONS,
)
from k1link.web.e30_review_api import (
E30ReviewEvidenceError,
e30_review_item_summary,
load_verified_e30_review,
)
LABORATORY_E30_ENGINEERING_CATALOG_SCHEMA: Final = (
"missioncore.laboratory-e30-engineering-generations/v1"
)
LABORATORY_E30_ENGINEERING_DECISION_SCHEMA: Final = (
"missioncore.laboratory-e30-engineering-decision/v1"
)
LABORATORY_E30_ENGINEERING_EXCEPTIONS_SCHEMA: Final = (
"missioncore.laboratory-e30-engineering-exceptions/v1"
)
_GENERATION_ID = re.compile(r"^e30-engineering-generation-[a-f0-9]{64}$")
_MATERIALIZATION_ID = re.compile(r"^e30-materialization-[a-f0-9]{64}$")
_ITEM_ID = re.compile(r"^e30-review-item-[a-f0-9]{64}$")
_SHA256 = re.compile(r"^[a-f0-9]{64}$")
_MAX_MANIFEST_BYTES: Final = 1024 * 1024
_MAX_DECISIONS_BYTES: Final = 8 * 1024 * 1024
_MAX_AUXILIARY_BYTES: Final = 2 * 1024 * 1024
RootProvider = Callable[[], Path | None]
class E30EngineeringEvidenceError(ValueError):
"""An engineering generation is incomplete, changed, or incompatible."""
def _canonical_json(value: object) -> bytes:
return json.dumps(
value,
ensure_ascii=False,
sort_keys=True,
separators=(",", ":"),
allow_nan=False,
).encode("utf-8")
def _sha256(path: Path) -> str:
digest = hashlib.sha256()
with path.open("rb") as stream:
while chunk := stream.read(1024 * 1024):
digest.update(chunk)
return digest.hexdigest()
def _read_json(path: Path, maximum_bytes: int) -> dict[str, Any]:
if path.is_symlink() or not path.is_file():
raise E30EngineeringEvidenceError("engineering JSON is unavailable")
size = path.stat().st_size
if not 0 < size <= maximum_bytes:
raise E30EngineeringEvidenceError("engineering JSON is out of bounds")
try:
value = json.loads(path.read_text(encoding="utf-8"))
except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc:
raise E30EngineeringEvidenceError("engineering JSON is invalid") from exc
if not isinstance(value, dict):
raise E30EngineeringEvidenceError("engineering JSON must be an object")
return value
def _safe_file(root: Path, value: object) -> Path:
if not isinstance(value, str) or not value or Path(value).is_absolute():
raise E30EngineeringEvidenceError("engineering artifact path is invalid")
if ".." in Path(value).parts:
raise E30EngineeringEvidenceError("engineering artifact escaped its root")
candidate = root / value
if candidate.is_symlink():
raise E30EngineeringEvidenceError("engineering artifact is a symlink")
candidate = candidate.resolve(strict=True)
try:
candidate.relative_to(root)
except ValueError as exc:
raise E30EngineeringEvidenceError(
"engineering artifact escaped its root"
) from exc
if not candidate.is_file():
raise E30EngineeringEvidenceError("engineering artifact is unavailable")
return candidate
def _artifact(
root: Path,
value: object,
*,
allow_empty: bool = False,
) -> tuple[Path, dict[str, Any]]:
if not isinstance(value, dict):
raise E30EngineeringEvidenceError("engineering artifact metadata is invalid")
byte_length = value.get("byte_length")
digest = value.get("sha256")
if (
not isinstance(byte_length, int)
or isinstance(byte_length, bool)
or byte_length < (0 if allow_empty else 1)
or not isinstance(digest, str)
or _SHA256.fullmatch(digest) is None
):
raise E30EngineeringEvidenceError("engineering artifact metadata is invalid")
path = _safe_file(root, value.get("path"))
if path.stat().st_size != byte_length or _sha256(path) != digest:
raise E30EngineeringEvidenceError("engineering artifact content changed")
return path, value
def _read_jsonl(path: Path, maximum_bytes: int) -> list[dict[str, Any]]:
if path.stat().st_size > maximum_bytes:
raise E30EngineeringEvidenceError("engineering JSONL is out of bounds")
values: list[dict[str, Any]] = []
with path.open("r", encoding="utf-8") as stream:
for line in stream:
try:
value = json.loads(line)
except json.JSONDecodeError as exc:
raise E30EngineeringEvidenceError(
"engineering JSONL is invalid"
) from exc
if not isinstance(value, dict):
raise E30EngineeringEvidenceError(
"engineering JSONL row must be an object"
)
values.append(value)
return values
def _signature(root: Path) -> tuple[int, ...]:
paths = [
root / "manifest.json",
root / "engineering-decisions.jsonl",
root / "human-exceptions.jsonl",
root / "summary.json",
root / "cause-distribution.json",
]
signature: list[int] = []
for path in paths:
stat = path.stat()
signature.extend((stat.st_size, stat.st_mtime_ns))
return tuple(signature)
def _authority(value: object) -> None:
if (
not isinstance(value, dict)
or value.get("commands_enabled") is not False
or value.get("navigation_or_safety_accepted") is not False
):
raise E30EngineeringEvidenceError(
"engineering generation must remain diagnostic-only"
)
def _verify_materialization_binding(
*,
materialization_root: Path,
source: dict[str, Any],
) -> None:
result_id = source.get("materialization_id")
if (
not isinstance(result_id, str)
or _MATERIALIZATION_ID.fullmatch(result_id) is None
):
raise E30EngineeringEvidenceError("materialization binding is invalid")
candidate = materialization_root / result_id
manifest_path = candidate / "manifest.json"
if candidate.is_symlink() or not candidate.is_dir():
raise E30EngineeringEvidenceError("materialization binding is unavailable")
manifest = _read_json(manifest_path, _MAX_MANIFEST_BYTES)
if (
manifest.get("result_id") != result_id
or manifest.get("identity_sha256")
!= source.get("materialization_identity_sha256")
or _sha256(manifest_path) != source.get("materialization_manifest_sha256")
or _sha256(candidate / "materialized-items.jsonl")
!= source.get("materialization_index_sha256")
):
raise E30EngineeringEvidenceError("materialization binding changed")
def _validate_decisions(
values: list[dict[str, Any]],
) -> tuple[dict[str, object], dict[str, object]]:
if (
len(values) != 486
or len({value.get("item_id") for value in values}) != len(values)
):
raise E30EngineeringEvidenceError("engineering coverage is incomplete")
verdicts: Counter[str] = Counter()
detector: Counter[str] = Counter()
projection: Counter[str] = Counter()
ownership: Counter[str] = Counter()
causes: Counter[str] = Counter()
exceptions = 0
confidence_total = 0.0
for sequence, value in enumerate(values):
confidence = value.get("confidence")
schema_version = value.get("schema_version")
exception_required = value.get("human_exception_required")
review_prompt = value.get("review_prompt")
if (
schema_version not in E30_ENGINEERING_DECISION_SCHEMAS
or value.get("sequence") != sequence
or _ITEM_ID.fullmatch(str(value.get("item_id"))) is None
or not isinstance(value.get("verdict"), str)
or not isinstance(value.get("detector_assessment"), str)
or not isinstance(value.get("projection_assessment"), str)
or not isinstance(value.get("point_ownership"), str)
or not isinstance(confidence, (int, float))
or isinstance(confidence, bool)
or not math.isfinite(float(confidence))
or not isinstance(exception_required, bool)
):
raise E30EngineeringEvidenceError("engineering decision is invalid")
if schema_version == E30_ENGINEERING_DECISION_SCHEMA and (
exception_required != _valid_review_prompt(review_prompt)
):
raise E30EngineeringEvidenceError(
"engineering exception prompt is invalid"
)
verdicts[value["verdict"]] += 1
detector[value["detector_assessment"]] += 1
projection[value["projection_assessment"]] += 1
ownership[value["point_ownership"]] += 1
if isinstance(value.get("cause_code"), str):
causes[value["cause_code"]] += 1
if value["human_exception_required"]:
exceptions += 1
confidence_total += float(confidence)
summary: dict[str, object] = {
"schema_version": E30_ENGINEERING_SUMMARY_SCHEMA,
"item_count": len(values),
"reviewed_item_count": len(values),
"verdict_distribution": dict(sorted(verdicts.items())),
"detector_distribution": dict(sorted(detector.items())),
"projection_distribution": dict(sorted(projection.items())),
"point_ownership_distribution": dict(sorted(ownership.items())),
"human_exception_count": exceptions,
"mean_confidence": round(confidence_total / len(values), 4),
}
cause_document: dict[str, object] = {
"schema_version": E30_ENGINEERING_CAUSES_SCHEMA,
"item_count_with_cause": sum(causes.values()),
"reasons": [
{"reason_code": reason, "count": count}
for reason, count in sorted(causes.items())
],
}
return summary, cause_document
def _valid_review_prompt(value: object) -> bool:
if not isinstance(value, dict) or set(value) != {"question", "focus", "effects"}:
return False
effects = value.get("effects")
return (
isinstance(value.get("question"), str)
and bool(value["question"].strip())
and isinstance(value.get("focus"), str)
and bool(value["focus"].strip())
and isinstance(effects, dict)
and set(effects) == set(EXCEPTION_DISPOSITIONS)
and all(
isinstance(effects.get(disposition), str)
and bool(effects[disposition].strip())
for disposition in EXCEPTION_DISPOSITIONS
)
)
@lru_cache(maxsize=8)
def _generation_cached(
root_text: str,
materialization_root_text: str,
signature: tuple[int, ...],
) -> tuple[dict[str, object], tuple[dict[str, Any], ...]]:
del signature
root = Path(root_text)
materialization_root = Path(materialization_root_text)
if (
root.is_symlink()
or not root.is_dir()
or _GENERATION_ID.fullmatch(root.name) is None
):
raise E30EngineeringEvidenceError("engineering generation id is invalid")
manifest = _read_json(root / "manifest.json", _MAX_MANIFEST_BYTES)
identity = manifest.get("identity")
identity_sha256 = manifest.get("identity_sha256")
if (
manifest.get("schema_version") != E30_ENGINEERING_GENERATION_SCHEMA
or manifest.get("result_id") != root.name
or not isinstance(identity, dict)
or not isinstance(identity_sha256, str)
or _SHA256.fullmatch(identity_sha256) is None
or hashlib.sha256(_canonical_json(identity)).hexdigest() != identity_sha256
or root.name != f"e30-engineering-generation-{identity_sha256}"
or manifest.get("ai_review_complete") is not True
or manifest.get("human_review_complete") is not False
or manifest.get("lab_published") is not False
):
raise E30EngineeringEvidenceError(
"engineering generation identity is invalid"
)
_authority(manifest.get("authority"))
source = identity.get("source")
producer = identity.get("producer")
if (
not isinstance(source, dict)
or not isinstance(producer, dict)
or producer.get("kind") != "ai-assisted-engineering-review"
or producer.get("claims_human_ground_truth") is not False
):
raise E30EngineeringEvidenceError("engineering provenance is invalid")
_verify_materialization_binding(
materialization_root=materialization_root,
source=source,
)
artifacts = manifest.get("artifacts")
if not isinstance(artifacts, list):
raise E30EngineeringEvidenceError("engineering artifacts are invalid")
by_role = {
str(value.get("role")): value
for value in artifacts
if isinstance(value, dict)
}
required_roles = {
"engineering-decisions",
"human-exceptions",
"engineering-summary",
"cause-distribution",
}
if set(by_role) != required_roles or len(artifacts) != len(required_roles):
raise E30EngineeringEvidenceError("engineering artifacts are incomplete")
decisions_path, decisions_artifact = _artifact(
root,
by_role["engineering-decisions"],
)
exceptions_path, _ = _artifact(
root,
by_role["human-exceptions"],
allow_empty=True,
)
summary_path, _ = _artifact(root, by_role["engineering-summary"])
causes_path, _ = _artifact(root, by_role["cause-distribution"])
decisions = _read_jsonl(decisions_path, _MAX_DECISIONS_BYTES)
exceptions = _read_jsonl(exceptions_path, _MAX_AUXILIARY_BYTES)
for exception in exceptions:
exception_schema = exception.get("schema_version")
if (
exception_schema not in E30_ENGINEERING_EXCEPTION_SCHEMAS
or (
exception_schema == E30_ENGINEERING_EXCEPTION_SCHEMA
and not _valid_review_prompt(exception.get("review_prompt"))
)
):
raise E30EngineeringEvidenceError(
"engineering exception contract is invalid"
)
calculated_summary, calculated_causes = _validate_decisions(decisions)
summary = _read_json(summary_path, _MAX_AUXILIARY_BYTES)
causes = _read_json(causes_path, _MAX_AUXILIARY_BYTES)
if (
summary != calculated_summary
or causes != calculated_causes
or manifest.get("summary") != summary
or manifest.get("cause_distribution") != causes
or len(exceptions) != summary["human_exception_count"]
or identity.get("decision_content_sha256")
!= decisions_artifact.get("sha256")
or source.get("item_count") != len(decisions)
):
raise E30EngineeringEvidenceError("engineering summary changed")
exception_ids = {value.get("item_id") for value in exceptions}
expected_exception_ids = {
value["item_id"]
for value in decisions
if value["human_exception_required"]
}
if exception_ids != expected_exception_ids:
raise E30EngineeringEvidenceError("engineering exception queue changed")
catalog_item: dict[str, object] = {
"generation_id": root.name,
"created_at_utc": manifest.get("created_at_utc"),
"materialization_id": source["materialization_id"],
"producer": copy.deepcopy(producer),
"summary": copy.deepcopy(summary),
"cause_distribution": copy.deepcopy(causes),
"human_exceptions": [
{
"item_id": value["item_id"],
"review_key": value["review_key"],
"source_stratum": value["source_stratum"],
"confidence": value["confidence"],
"review_prompt": copy.deepcopy(value.get("review_prompt")),
}
for value in exceptions
],
"ai_review_complete": True,
"human_exception_complete": (
manifest.get("human_exception_complete") is True
),
"human_review_complete": False,
"lab_published": False,
"access": "read-only",
"authority": copy.deepcopy(manifest["authority"]),
}
return catalog_item, tuple(decisions)
def load_verified_e30_engineering_generation(
*,
generation_root: Path,
materialization_root: Path,
result_id: str,
generation_id: str,
) -> tuple[dict[str, object], tuple[dict[str, Any], ...]]:
generation_root = generation_root.resolve()
materialization_root = materialization_root.resolve()
if (
not generation_root.is_dir()
or not materialization_root.is_dir()
or _MATERIALIZATION_ID.fullmatch(result_id) is None
or _GENERATION_ID.fullmatch(generation_id) is None
):
raise E30EngineeringEvidenceError(
"engineering generation is unavailable"
)
candidate = generation_root / generation_id
if candidate.is_symlink() or not candidate.is_dir():
raise E30EngineeringEvidenceError(
"engineering generation is unavailable"
)
catalog_item, decisions = _generation_cached(
str(candidate.resolve()),
str(materialization_root),
_signature(candidate),
)
if catalog_item["materialization_id"] != result_id:
raise E30EngineeringEvidenceError(
"engineering generation materialization differs"
)
return copy.deepcopy(catalog_item), decisions
def build_e30_engineering_router(
*,
generation_root_provider: RootProvider = lambda: None,
materialization_root_provider: RootProvider = lambda: None,
review_pack_root_provider: RootProvider = lambda: None,
) -> APIRouter:
router = APIRouter(prefix="/api/v1/laboratory/e30", tags=["laboratory"])
def roots() -> tuple[Path, Path] | None:
generation_root = generation_root_provider()
materialization_root = materialization_root_provider()
if generation_root is None or materialization_root is None:
return None
generation_root = generation_root.resolve()
materialization_root = materialization_root.resolve()
if not generation_root.is_dir() or not materialization_root.is_dir():
return None
return generation_root, materialization_root
@router.get("/reviews/{result_id}/engineering-generations")
def list_engineering_generations(
result_id: str,
limit: int = Query(default=1, ge=1, le=10),
) -> dict[str, object]:
configured_roots = roots()
if (
configured_roots is None
or _MATERIALIZATION_ID.fullmatch(result_id) is None
):
return {
"schema_version": LABORATORY_E30_ENGINEERING_CATALOG_SCHEMA,
"configured": configured_roots is not None,
"items": [],
"candidate_total": 0,
"invalid_total": 0,
"access": "read-only",
}
generation_root, materialization_root = configured_roots
candidates = sorted(
(
candidate
for candidate in generation_root.iterdir()
if candidate.is_dir()
and _GENERATION_ID.fullmatch(candidate.name) is not None
),
key=lambda candidate: candidate.stat().st_mtime_ns,
reverse=True,
)
items: list[dict[str, object]] = []
invalid_total = 0
matching_total = 0
for candidate in candidates:
try:
item, _ = _generation_cached(
str(candidate.resolve()),
str(materialization_root),
_signature(candidate),
)
if item["materialization_id"] != result_id:
continue
matching_total += 1
if len(items) < limit:
items.append(copy.deepcopy(item))
except (E30EngineeringEvidenceError, OSError):
invalid_total += 1
return {
"schema_version": LABORATORY_E30_ENGINEERING_CATALOG_SCHEMA,
"configured": True,
"items": items,
"candidate_total": matching_total,
"invalid_total": invalid_total,
"access": "read-only",
}
@router.get(
"/reviews/{result_id}/engineering-generations/"
"{generation_id}/exceptions"
)
def list_engineering_exceptions(
result_id: str,
generation_id: str,
limit: int = Query(default=48, ge=1, le=128),
cursor: int = Query(default=0, ge=0),
) -> dict[str, object]:
configured_roots = roots()
review_pack_root = review_pack_root_provider()
if (
configured_roots is None
or review_pack_root is None
or not review_pack_root.is_dir()
or _MATERIALIZATION_ID.fullmatch(result_id) is None
or _GENERATION_ID.fullmatch(generation_id) is None
):
raise HTTPException(
status_code=404,
detail="A3 engineering exception queue не найдена",
)
generation_root, materialization_root = configured_roots
candidate = generation_root / generation_id
if candidate.is_symlink() or not candidate.is_dir():
raise HTTPException(
status_code=404,
detail="A3 engineering exception queue не найдена",
)
try:
catalog_item, _ = _generation_cached(
str(candidate.resolve()),
str(materialization_root),
_signature(candidate),
)
if catalog_item["materialization_id"] != result_id:
raise HTTPException(
status_code=404,
detail="A3 engineering exception queue не найдена",
)
_, rows, _ = load_verified_e30_review(
materialization_root=materialization_root,
review_pack_root=review_pack_root,
result_id=result_id,
)
rows_by_id = {row["item_id"]: row for row in rows}
exception_ids = [
value["item_id"]
for value in catalog_item["human_exceptions"]
]
if any(item_id not in rows_by_id for item_id in exception_ids):
raise E30EngineeringEvidenceError(
"engineering exception source item is unavailable"
)
page_ids = exception_ids[cursor : cursor + limit]
next_cursor = cursor + len(page_ids)
return {
"schema_version": LABORATORY_E30_ENGINEERING_EXCEPTIONS_SCHEMA,
"result_id": result_id,
"generation_id": generation_id,
"items": [
e30_review_item_summary(rows_by_id[item_id])
for item_id in page_ids
],
"total": len(exception_ids),
"next_cursor": (
next_cursor if next_cursor < len(exception_ids) else None
),
"access": "read-only",
}
except HTTPException:
raise
except (
E30EngineeringEvidenceError,
E30ReviewEvidenceError,
OSError,
) as exc:
raise HTTPException(
status_code=409,
detail="A3 engineering exception queue не прошла проверку целостности",
) from exc
@router.get(
"/reviews/{result_id}/engineering-generations/"
"{generation_id}/items/{item_id}"
)
def get_engineering_decision(
result_id: str,
generation_id: str,
item_id: str,
) -> dict[str, object]:
configured_roots = roots()
if (
configured_roots is None
or _MATERIALIZATION_ID.fullmatch(result_id) is None
or _GENERATION_ID.fullmatch(generation_id) is None
or _ITEM_ID.fullmatch(item_id) is None
):
raise HTTPException(
status_code=404,
detail="A3 engineering decision не найден",
)
generation_root, materialization_root = configured_roots
candidate = generation_root / generation_id
if candidate.is_symlink() or not candidate.is_dir():
raise HTTPException(
status_code=404,
detail="A3 engineering decision не найден",
)
try:
catalog_item, decisions = _generation_cached(
str(candidate.resolve()),
str(materialization_root),
_signature(candidate),
)
if catalog_item["materialization_id"] != result_id:
raise HTTPException(
status_code=404,
detail="A3 engineering decision не найден",
)
decision = next(
(value for value in decisions if value["item_id"] == item_id),
None,
)
if decision is None:
raise HTTPException(
status_code=404,
detail="A3 engineering decision не найден",
)
return {
"schema_version": LABORATORY_E30_ENGINEERING_DECISION_SCHEMA,
"generation_id": generation_id,
"decision": copy.deepcopy(decision),
"access": "read-only",
}
except HTTPException:
raise
except (E30EngineeringEvidenceError, OSError) as exc:
raise HTTPException(
status_code=409,
detail="A3 engineering generation не прошла проверку целостности",
) from exc
return router
+263
View File
@@ -0,0 +1,263 @@
from __future__ import annotations
from collections.abc import Callable
from pathlib import Path
from typing import Annotated, Literal
from fastapi import APIRouter, HTTPException
from fastapi import Path as ApiPath
from pydantic import BaseModel, ConfigDict, Field
from k1link.compute.e30_human_review import (
E30HumanReviewConflictError,
E30HumanReviewIntegrityError,
E30HumanReviewNotFoundError,
E30HumanReviewStore,
E30HumanReviewValidationError,
E30ExceptionDisposition,
E30ReviewSubject,
E30ReviewSubstrate,
)
from k1link.web.e30_engineering_api import (
E30EngineeringEvidenceError,
load_verified_e30_engineering_generation,
)
from k1link.web.e30_review_api import (
E30ReviewEvidenceError,
load_verified_e30_review,
)
RootProvider = Callable[[], Path | None]
class E30HumanReviewCreateRequest(BaseModel):
model_config = ConfigDict(extra="forbid")
reviewer_id: str = Field(min_length=1, max_length=128)
engineering_generation_id: str = Field(
pattern=r"^e30-engineering-generation-[a-f0-9]{64}$"
)
class E30HumanReviewDecisionRequest(BaseModel):
model_config = ConfigDict(extra="forbid")
expected_revision: int = Field(ge=0)
idempotency_key: str = Field(min_length=1, max_length=128)
disposition: E30ExceptionDisposition
notes: str | None = Field(default=None, max_length=2_000)
class E30HumanReviewFinalizeRequest(BaseModel):
model_config = ConfigDict(extra="forbid")
expected_revision: int = Field(ge=0)
confirm_generation: Literal[True]
def build_e30_human_review_router(
*,
materialization_root_provider: RootProvider = lambda: None,
review_pack_root_provider: RootProvider = lambda: None,
engineering_generation_root_provider: RootProvider = lambda: None,
draft_root_provider: RootProvider = lambda: None,
generation_root_provider: RootProvider = lambda: None,
) -> APIRouter:
router = APIRouter(prefix="/api/v1/laboratory/e30", tags=["laboratory"])
def source(
result_id: str,
engineering_generation_id: str,
) -> E30ReviewSubstrate:
materialization_root = materialization_root_provider()
review_pack_root = review_pack_root_provider()
engineering_generation_root = engineering_generation_root_provider()
if (
materialization_root is None
or review_pack_root is None
or engineering_generation_root is None
):
raise HTTPException(status_code=404, detail="E30 review не найден")
try:
_, _, source_substrate = load_verified_e30_review(
materialization_root=materialization_root,
review_pack_root=review_pack_root,
result_id=result_id,
)
generation, _ = load_verified_e30_engineering_generation(
generation_root=engineering_generation_root,
materialization_root=materialization_root,
result_id=result_id,
generation_id=engineering_generation_id,
)
subjects_by_id = {
subject.item_id: subject
for subject in source_substrate.subjects
}
exception_ids = [
value["item_id"]
for value in generation["human_exceptions"]
]
if (
not exception_ids
or any(item_id not in subjects_by_id for item_id in exception_ids)
):
raise E30HumanReviewValidationError(
"engineering exception substrate is invalid"
)
return E30ReviewSubstrate(
materialization_id=source_substrate.materialization_id,
materialization_identity_sha256=(
source_substrate.materialization_identity_sha256
),
review_pack_id=source_substrate.review_pack_id,
review_items_sha256=source_substrate.review_items_sha256,
reason_taxonomy=(),
subjects=tuple(
E30ReviewSubject(
item_id=item_id,
sequence=sequence,
source_stratum=subjects_by_id[item_id].source_stratum,
)
for sequence, item_id in enumerate(exception_ids)
),
engineering_generation_id=engineering_generation_id,
)
except (
E30EngineeringEvidenceError,
E30ReviewEvidenceError,
E30HumanReviewValidationError,
OSError,
) as exc:
raise HTTPException(
status_code=409,
detail="E30 source evidence не прошло проверку целостности",
) from exc
def store() -> E30HumanReviewStore:
draft_root = draft_root_provider()
generation_root = generation_root_provider()
if draft_root is None or generation_root is None:
raise HTTPException(
status_code=503,
detail="A3 human review не настроен",
)
try:
return E30HumanReviewStore(
draft_root=draft_root,
generation_root=generation_root,
)
except (E30HumanReviewIntegrityError, OSError) as exc:
raise HTTPException(
status_code=503,
detail="A3 human review storage недоступен",
) from exc
def invoke(
operation: Callable[[], dict[str, object]],
) -> dict[str, object]:
try:
return operation()
except E30HumanReviewNotFoundError as exc:
raise HTTPException(
status_code=404,
detail="A3 human review draft не найден",
) from exc
except E30HumanReviewValidationError as exc:
raise HTTPException(status_code=422, detail=str(exc)) from exc
except E30HumanReviewConflictError as exc:
raise HTTPException(status_code=409, detail=str(exc)) from exc
except (E30HumanReviewIntegrityError, OSError) as exc:
raise HTTPException(
status_code=409,
detail="A3 human review не прошло проверку целостности",
) from exc
@router.post("/reviews/{result_id}/human-review")
def create_or_resume_human_review(
result_id: str,
request: E30HumanReviewCreateRequest,
) -> dict[str, object]:
substrate = source(result_id, request.engineering_generation_id)
review_store = store()
return invoke(
lambda: review_store.create_or_resume(
substrate=substrate,
reviewer_id=request.reviewer_id,
)
)
@router.get("/reviews/{result_id}/human-review/{draft_id}")
def get_human_review(
result_id: str,
draft_id: Annotated[
str,
ApiPath(pattern=r"^e30-human-draft-[a-f0-9]{64}$"),
],
engineering_generation_id: str,
) -> dict[str, object]:
substrate = source(result_id, engineering_generation_id)
review_store = store()
return invoke(
lambda: review_store.get(draft_id=draft_id, substrate=substrate)
)
@router.put(
"/reviews/{result_id}/human-review/{draft_id}/decisions/{item_id}"
)
def record_human_review_decision(
result_id: str,
draft_id: Annotated[
str,
ApiPath(pattern=r"^e30-human-draft-[a-f0-9]{64}$"),
],
item_id: Annotated[
str,
ApiPath(pattern=r"^e30-review-item-[a-f0-9]{64}$"),
],
request: E30HumanReviewDecisionRequest,
engineering_generation_id: str,
) -> dict[str, object]:
substrate = source(result_id, engineering_generation_id)
review_store = store()
return invoke(
lambda: review_store.record_decision(
draft_id=draft_id,
substrate=substrate,
item_id=item_id,
expected_revision=request.expected_revision,
idempotency_key=request.idempotency_key,
disposition=request.disposition,
notes=request.notes,
)
)
@router.post("/reviews/{result_id}/human-review/{draft_id}/finalize")
def finalize_human_review(
result_id: str,
draft_id: Annotated[
str,
ApiPath(pattern=r"^e30-human-draft-[a-f0-9]{64}$"),
],
request: E30HumanReviewFinalizeRequest,
engineering_generation_id: str,
) -> dict[str, object]:
substrate = source(result_id, engineering_generation_id)
review_store = store()
generation = invoke(
lambda: review_store.finalize(
draft_id=draft_id,
substrate=substrate,
expected_revision=request.expected_revision,
)
)
draft = invoke(
lambda: review_store.get(draft_id=draft_id, substrate=substrate)
)
return {
"schema_version": "missioncore.laboratory-e30-human-review-finalized/v2",
"draft": draft,
"generation": generation,
}
return router
+818
View File
@@ -0,0 +1,818 @@
from __future__ import annotations
import copy
import hashlib
import json
import re
from collections import Counter
from collections.abc import Callable
from functools import lru_cache
from pathlib import Path
from typing import Annotated, Any, Final, Literal, cast
import numpy as np
import numpy.typing as npt
from fastapi import APIRouter, HTTPException, Query
from fastapi.responses import FileResponse
from k1link.compute.e30_human_review import (
E30ReviewSubject,
E30ReviewSubstrate,
E30Stratum,
)
from k1link.compute.e30_materialization import (
E30_MATERIALIZATION_INDEX_NAME,
E30_MATERIALIZATION_ITEM_SCHEMA,
E30_MATERIALIZATION_MANIFEST_NAME,
E30_MATERIALIZATION_SCHEMA,
)
LABORATORY_E30_CATALOG_SCHEMA: Final = "missioncore.laboratory-e30-catalog/v1"
LABORATORY_E30_ITEMS_SCHEMA: Final = "missioncore.laboratory-e30-items/v1"
LABORATORY_E30_ITEM_DETAIL_SCHEMA: Final = (
"missioncore.laboratory-e30-item-detail/v1"
)
_MATERIALIZATION_ID = re.compile(r"^e30-materialization-[a-f0-9]{64}$")
_REVIEW_PACK_ID = re.compile(r"^e30-review-pack-[a-f0-9]{64}$")
_REVIEW_ITEM_ID = re.compile(r"^e30-review-item-[a-f0-9]{64}$")
_SHA256 = re.compile(r"^[a-f0-9]{64}$")
_STRATA: Final = ("conflict", "agree", "camera-only", "unknown", "geometry-only")
_MAX_MANIFEST_BYTES: Final = 512 * 1024
_MAX_INDEX_BYTES: Final = 8 * 1024 * 1024
_MAX_ITEM_BYTES: Final = 8 * 1024 * 1024
_MAX_CAMERA_FRAME_BYTES: Final = 8 * 1024 * 1024
RootProvider = Callable[[], Path | None]
Stratum = Literal["conflict", "agree", "camera-only", "unknown", "geometry-only"]
class E30ReviewEvidenceError(ValueError):
"""E30 review evidence is incomplete, changed or incompatible."""
def _canonical_json(value: object) -> bytes:
return json.dumps(
value,
ensure_ascii=False,
sort_keys=True,
separators=(",", ":"),
allow_nan=False,
).encode()
def _sha256(path: Path) -> str:
digest = hashlib.sha256()
with path.open("rb") as stream:
while chunk := stream.read(1024 * 1024):
digest.update(chunk)
return digest.hexdigest()
def _read_json(path: Path, maximum_bytes: int) -> dict[str, Any]:
if path.is_symlink() or not path.is_file():
raise E30ReviewEvidenceError("E30 JSON artifact is unavailable")
size = path.stat().st_size
if not 0 < size <= maximum_bytes:
raise E30ReviewEvidenceError("E30 JSON artifact is out of bounds")
try:
value = json.loads(path.read_text(encoding="utf-8"))
except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc:
raise E30ReviewEvidenceError("E30 JSON artifact is invalid") from exc
if not isinstance(value, dict):
raise E30ReviewEvidenceError("E30 JSON artifact must be an object")
return value
def _safe_relative_file(root: Path, value: object) -> Path:
if not isinstance(value, str) or not value or Path(value).is_absolute():
raise E30ReviewEvidenceError("E30 artifact path is invalid")
if ".." in Path(value).parts:
raise E30ReviewEvidenceError("E30 artifact path escaped its root")
candidate = root / value
if candidate.is_symlink():
raise E30ReviewEvidenceError("E30 artifact must not be a symlink")
candidate = candidate.resolve(strict=True)
try:
candidate.relative_to(root)
except ValueError as exc:
raise E30ReviewEvidenceError("E30 artifact escaped its root") from exc
if not candidate.is_file():
raise E30ReviewEvidenceError("E30 artifact is unavailable")
return candidate
def _artifact(
root: Path,
value: object,
*,
expected_role: str | None = None,
) -> tuple[Path, dict[str, Any]]:
if not isinstance(value, dict):
raise E30ReviewEvidenceError("E30 artifact metadata is invalid")
role = value.get("role")
byte_length = value.get("byte_length")
digest = value.get("sha256")
if (
(expected_role is not None and role != expected_role)
or not isinstance(byte_length, int)
or byte_length <= 0
or not isinstance(digest, str)
or _SHA256.fullmatch(digest) is None
):
raise E30ReviewEvidenceError("E30 artifact metadata is invalid")
path = _safe_relative_file(root, value.get("path"))
if path.stat().st_size != byte_length or _sha256(path) != digest:
raise E30ReviewEvidenceError("E30 artifact content changed")
return path, value
def _authority(value: object) -> None:
if (
not isinstance(value, dict)
or value.get("commands_enabled") is not False
or value.get("navigation_or_safety_accepted") is not False
):
raise E30ReviewEvidenceError("E30 evidence must remain diagnostic-only")
def _signature(root: Path) -> tuple[int, ...]:
paths = [
root / E30_MATERIALIZATION_MANIFEST_NAME,
root / E30_MATERIALIZATION_INDEX_NAME,
*sorted((root / "items").glob("*.npz")),
*sorted((root / "frames").glob("*.jpg")),
]
signature: list[int] = []
for path in paths:
stat = path.stat()
signature.extend((stat.st_size, stat.st_mtime_ns))
return tuple(signature)
def _review_pack(
root: Path,
*,
expected_id: str,
expected_items_sha256: str,
) -> dict[str, Any]:
candidate = root / expected_id
if (
_REVIEW_PACK_ID.fullmatch(expected_id) is None
or candidate.is_symlink()
or not candidate.is_dir()
):
raise E30ReviewEvidenceError("linked E30 review pack is unavailable")
manifest = _read_json(candidate / "manifest.json", _MAX_MANIFEST_BYTES)
identity = manifest.get("identity")
identity_sha256 = manifest.get("identity_sha256")
if (
manifest.get("schema_version") != "missioncore.e30-evidence-review-pack/v1"
or manifest.get("result_id") != expected_id
or not isinstance(identity, dict)
or not isinstance(identity_sha256, str)
or _SHA256.fullmatch(identity_sha256) is None
or hashlib.sha256(_canonical_json(identity)).hexdigest() != identity_sha256
or expected_id != f"e30-review-pack-{identity_sha256}"
or manifest.get("human_review_complete") is not False
or manifest.get("lab_published") is not False
):
raise E30ReviewEvidenceError("linked E30 review pack identity is invalid")
_authority(manifest.get("authority"))
artifacts = manifest.get("artifacts")
if not isinstance(artifacts, list) or len(artifacts) != 1:
raise E30ReviewEvidenceError("linked E30 review artifacts are invalid")
_, artifact = _artifact(
candidate,
artifacts[0],
expected_role="review-items",
)
if artifact.get("sha256") != expected_items_sha256:
raise E30ReviewEvidenceError("linked E30 review item digest differs")
return manifest
@lru_cache(maxsize=8)
def _read_materialization_cached(
root_text: str,
review_root_text: str,
signature: tuple[int, ...],
) -> tuple[dict[str, Any], tuple[dict[str, Any], ...]]:
del signature
root = Path(root_text)
review_root = Path(review_root_text)
if (
root.is_symlink()
or not root.is_dir()
or _MATERIALIZATION_ID.fullmatch(root.name) is None
):
raise E30ReviewEvidenceError("E30 materialization id is invalid")
manifest = _read_json(
root / E30_MATERIALIZATION_MANIFEST_NAME,
_MAX_MANIFEST_BYTES,
)
identity = manifest.get("identity")
identity_sha256 = manifest.get("identity_sha256")
if (
manifest.get("schema_version") != E30_MATERIALIZATION_SCHEMA
or manifest.get("result_id") != root.name
or not isinstance(identity, dict)
or not isinstance(identity_sha256, str)
or _SHA256.fullmatch(identity_sha256) is None
or hashlib.sha256(_canonical_json(identity)).hexdigest() != identity_sha256
or root.name != f"e30-materialization-{identity_sha256}"
or manifest.get("human_review_complete") is not False
or manifest.get("lab_published") is not False
):
raise E30ReviewEvidenceError("E30 materialization identity is invalid")
_authority(manifest.get("authority"))
artifacts = manifest.get("artifacts")
if not isinstance(artifacts, list) or len(artifacts) != 1:
raise E30ReviewEvidenceError("E30 materialization artifacts are invalid")
index_path, _ = _artifact(
root,
artifacts[0],
expected_role="materialized-items",
)
if index_path.stat().st_size > _MAX_INDEX_BYTES:
raise E30ReviewEvidenceError("E30 materialization index is out of bounds")
review_binding = identity.get("review_pack")
if not isinstance(review_binding, dict):
raise E30ReviewEvidenceError("E30 review binding is invalid")
review_manifest = _review_pack(
review_root,
expected_id=_string(review_binding, "result_id"),
expected_items_sha256=_sha_value(review_binding, "items_sha256"),
)
rows: list[dict[str, Any]] = []
counts: Counter[str] = Counter()
with index_path.open("r", encoding="utf-8") as stream:
for expected_sequence, line in enumerate(stream):
try:
value = json.loads(line)
except json.JSONDecodeError as exc:
raise E30ReviewEvidenceError(
"E30 materialization item is invalid"
) from exc
if not isinstance(value, dict):
raise E30ReviewEvidenceError(
"E30 materialization item must be an object"
)
item_id = value.get("item_id")
stratum = value.get("stratum")
if (
value.get("schema_version") != E30_MATERIALIZATION_ITEM_SCHEMA
or value.get("sequence") != expected_sequence
or not isinstance(item_id, str)
or _REVIEW_ITEM_ID.fullmatch(item_id) is None
or stratum not in _STRATA
or not isinstance(value.get("review"), dict)
or value["review"].get("state") != "unreviewed"
):
raise E30ReviewEvidenceError(
"E30 materialization item contract is invalid"
)
_authority(value.get("authority"))
_, artifact = _artifact(root, value.get("artifact"))
if artifact.get("path") != f"items/{item_id}.npz":
raise E30ReviewEvidenceError("E30 item artifact path differs")
camera_frame = value.get("camera_frame")
if camera_frame is not None:
camera_path, camera_artifact = _artifact(
root,
camera_frame,
expected_role="camera-frame",
)
expected_frame = value.get("evidence_binding", {}).get(
"source_frame_index"
)
if (
not isinstance(expected_frame, int)
or isinstance(expected_frame, bool)
or expected_frame < 0
or camera_path.stat().st_size > _MAX_CAMERA_FRAME_BYTES
or camera_artifact.get("media_type") != "image/jpeg"
or camera_artifact.get("source_frame_index")
!= expected_frame
or camera_artifact.get("exact_source_frame") is not True
or camera_artifact.get("path")
!= f"frames/frame-{expected_frame:06d}.jpg"
):
raise E30ReviewEvidenceError(
"E30 camera frame binding is invalid"
)
triage = value.get("engineering_triage")
if (
not isinstance(triage, dict)
or triage.get("schema_version")
!= "missioncore.e30-engineering-triage/v1"
or triage.get("semantic_verdict") is not None
or triage.get("human_exception_required") is not None
):
raise E30ReviewEvidenceError(
"E30 engineering triage contract is invalid"
)
rows.append(value)
counts[stratum] += 1
if (
len(rows) != manifest.get("item_count")
or len(rows) != review_binding.get("item_count")
or len({row["item_id"] for row in rows}) != len(rows)
):
raise E30ReviewEvidenceError("E30 materialization item count differs")
review_identity = review_manifest.get("identity")
reason_taxonomy = (
review_identity.get("reason_taxonomy")
if isinstance(review_identity, dict)
else None
)
source = identity.get("source")
projection = identity.get("projection")
if (
not isinstance(reason_taxonomy, list)
or not all(isinstance(reason, str) for reason in reason_taxonomy)
or not isinstance(source, dict)
or not isinstance(projection, dict)
):
raise E30ReviewEvidenceError("E30 catalog metadata is incomplete")
catalog_item = {
"result_id": root.name,
"created_at_utc": manifest.get("created_at_utc"),
"review_pack_id": review_binding["result_id"],
"e29_result_id": source.get("e29_result_id"),
"source_session_id": source.get("source_session_id"),
"item_count": len(rows),
"stratum_counts": {stratum: counts[stratum] for stratum in _STRATA},
"reason_taxonomy": reason_taxonomy,
"projection": projection,
"camera_evidence_available": (
manifest.get("camera_evidence_available") is True
),
"human_review_complete": False,
"lab_published": False,
"access": "read-only",
"authority": manifest.get("authority"),
}
return catalog_item, tuple(rows)
def _roots(
*,
materialization_root_provider: RootProvider,
review_pack_root_provider: RootProvider,
) -> tuple[Path, Path] | None:
materialization_root = materialization_root_provider()
review_root = review_pack_root_provider()
if materialization_root is None or review_root is None:
return None
materialization_root = materialization_root.resolve()
review_root = review_root.resolve()
if not materialization_root.is_dir() or not review_root.is_dir():
return None
return materialization_root, review_root
def _candidate(
materialization_root: Path,
result_id: str,
) -> Path:
if _MATERIALIZATION_ID.fullmatch(result_id) is None:
raise HTTPException(status_code=404, detail="E30 review не найден")
candidate = materialization_root / result_id
if candidate.is_symlink() or not candidate.is_dir():
raise HTTPException(status_code=404, detail="E30 review не найден")
return candidate
def e30_review_item_summary(row: dict[str, Any]) -> dict[str, object]:
binding = row.get("evidence_binding")
locator = row.get("e29_locator")
snapshot = row.get("e29_snapshot")
materialization = row.get("materialization")
if (
not isinstance(binding, dict)
or not isinstance(locator, dict)
or not isinstance(snapshot, dict)
or not isinstance(materialization, dict)
):
raise E30ReviewEvidenceError("E30 item summary is incomplete")
return {
"item_id": row["item_id"],
"sequence": row["sequence"],
"review_key": row["review_key"],
"stratum": row["stratum"],
"range_bucket": row["range_bucket"],
"frame_index": binding.get("frame_index"),
"source_frame_index": binding.get("source_frame_index"),
"session_seconds": binding.get("session_seconds"),
"locator": locator,
"snapshot": snapshot,
"materialization": materialization,
"camera_frame_available": isinstance(row.get("camera_frame"), dict),
"engineering_triage": row.get("engineering_triage"),
"review": row["review"],
"access": "read-only",
}
def _logical_arrays_sha256(arrays: dict[str, npt.NDArray[Any]]) -> str:
digest = hashlib.sha256()
for name in sorted(arrays):
array = np.ascontiguousarray(arrays[name])
digest.update(name.encode())
digest.update(array.dtype.str.encode())
digest.update(_canonical_json(list(array.shape)))
digest.update(array.tobytes(order="C"))
return digest.hexdigest()
def _item_detail(root: Path, row: dict[str, Any]) -> dict[str, object]:
path, artifact = _artifact(root, row.get("artifact"))
if path.stat().st_size > _MAX_ITEM_BYTES:
raise E30ReviewEvidenceError("E30 item artifact is out of bounds")
expected_files = {
"selected_source_indices",
"selected_points_map_xyz_m",
"candidate_source_indices",
"candidate_points_map_xyz_m",
"projected_source_indices",
"projected_points_map_xyz_m",
"projected_pixels_xy",
"projected_depth_m",
"projected_point_class",
"projected_point_height_m",
"projected_candidate_mask",
"projected_selected_mask",
"sensor_position_map_xyz_m",
"sensor_orientation_map_from_lidar_xyzw",
}
with np.load(path, allow_pickle=False) as archive:
if set(archive.files) != expected_files:
raise E30ReviewEvidenceError("E30 item arrays are incompatible")
arrays = {
name: np.ascontiguousarray(archive[name])
for name in expected_files
}
if _logical_arrays_sha256(arrays) != artifact.get("logical_sha256"):
raise E30ReviewEvidenceError("E30 item logical content changed")
projected_count = arrays["projected_source_indices"].shape[0]
selected_count = arrays["selected_source_indices"].shape[0]
candidate_count = arrays["candidate_source_indices"].shape[0]
if (
arrays["selected_source_indices"].shape != (selected_count,)
or arrays["selected_points_map_xyz_m"].shape != (selected_count, 3)
or arrays["candidate_source_indices"].shape != (candidate_count,)
or arrays["candidate_points_map_xyz_m"].shape != (candidate_count, 3)
or arrays["projected_points_map_xyz_m"].shape != (projected_count, 3)
or arrays["projected_pixels_xy"].shape != (projected_count, 2)
or arrays["projected_depth_m"].shape != (projected_count,)
or arrays["projected_point_class"].shape != (projected_count,)
or arrays["projected_point_height_m"].shape != (projected_count,)
or arrays["projected_candidate_mask"].shape != (projected_count,)
or arrays["projected_selected_mask"].shape != (projected_count,)
or arrays["sensor_position_map_xyz_m"].shape != (3,)
or arrays["sensor_orientation_map_from_lidar_xyzw"].shape != (4,)
or not all(np.isfinite(value).all() for value in arrays.values())
):
raise E30ReviewEvidenceError("E30 item array shape is invalid")
camera_frame = row.get("camera_frame")
camera_document: dict[str, object] | None = None
if camera_frame is not None:
_, camera_artifact = _artifact(
root,
camera_frame,
expected_role="camera-frame",
)
camera_document = {
"available": True,
"url": (
f"/api/v1/laboratory/e30/reviews/{root.name}/items/"
f"{row['item_id']}/camera-frame"
f"?generation={camera_artifact['sha256']}"
),
"sha256": camera_artifact["sha256"],
"width": camera_artifact.get("width"),
"height": camera_artifact.get("height"),
"source_frame_index": camera_artifact.get("source_frame_index"),
"exact_source_frame": True,
}
return {
**e30_review_item_summary(row),
"camera_frame": camera_document,
"selected": {
"source_indices": arrays["selected_source_indices"].astype(
np.int64
).tolist(),
"points_map_xyz_m": arrays["selected_points_map_xyz_m"].astype(
np.float64
).tolist(),
},
"candidate": {
"source_indices": arrays["candidate_source_indices"].astype(
np.int64
).tolist(),
"points_map_xyz_m": arrays["candidate_points_map_xyz_m"].astype(
np.float64
).tolist(),
},
"projection": {
"source_indices": arrays["projected_source_indices"].astype(
np.int64
).tolist(),
"points_map_xyz_m": arrays["projected_points_map_xyz_m"].astype(
np.float64
).tolist(),
"pixels_xy": arrays["projected_pixels_xy"].astype(np.float64).tolist(),
"depth_m": arrays["projected_depth_m"].astype(np.float64).tolist(),
"point_class": arrays["projected_point_class"].astype(np.int64).tolist(),
"point_height_m": arrays["projected_point_height_m"].astype(
np.float64
).tolist(),
"candidate_mask": arrays["projected_candidate_mask"].astype(
np.int64
).tolist(),
"selected_mask": arrays["projected_selected_mask"].astype(
np.int64
).tolist(),
},
"pose": {
"position_map_xyz_m": arrays["sensor_position_map_xyz_m"].astype(
np.float64
).tolist(),
"orientation_map_from_lidar_xyzw": arrays[
"sensor_orientation_map_from_lidar_xyzw"
]
.astype(np.float64)
.tolist(),
},
}
def _materialization(
candidate: Path,
review_root: Path,
) -> tuple[dict[str, Any], tuple[dict[str, Any], ...]]:
return _read_materialization_cached(
str(candidate.resolve()),
str(review_root.resolve()),
_signature(candidate),
)
def load_verified_e30_review(
*,
materialization_root: Path,
review_pack_root: Path,
result_id: str,
) -> tuple[
dict[str, Any],
tuple[dict[str, Any], ...],
E30ReviewSubstrate,
]:
"""Load the verified A2 evidence and its exact A3 source binding."""
materialization_root = materialization_root.resolve()
review_pack_root = review_pack_root.resolve()
if (
not materialization_root.is_dir()
or not review_pack_root.is_dir()
or _MATERIALIZATION_ID.fullmatch(result_id) is None
):
raise E30ReviewEvidenceError("E30 materialization is unavailable")
candidate = materialization_root / result_id
if candidate.is_symlink() or not candidate.is_dir():
raise E30ReviewEvidenceError("E30 materialization is unavailable")
catalog, rows = _materialization(candidate, review_pack_root)
manifest = _read_json(
candidate / E30_MATERIALIZATION_MANIFEST_NAME,
_MAX_MANIFEST_BYTES,
)
identity = manifest.get("identity")
review_binding = identity.get("review_pack") if isinstance(identity, dict) else None
if (
not isinstance(identity, dict)
or not isinstance(review_binding, dict)
or not isinstance(manifest.get("identity_sha256"), str)
):
raise E30ReviewEvidenceError("E30 reviewer source binding is incomplete")
substrate = E30ReviewSubstrate(
materialization_id=result_id,
materialization_identity_sha256=cast(str, manifest["identity_sha256"]),
review_pack_id=_string(review_binding, "result_id"),
review_items_sha256=_sha_value(review_binding, "items_sha256"),
reason_taxonomy=tuple(cast(list[str], catalog["reason_taxonomy"])),
subjects=tuple(
E30ReviewSubject(
item_id=cast(str, row["item_id"]),
sequence=cast(int, row["sequence"]),
source_stratum=cast(E30Stratum, row["stratum"]),
)
for row in rows
),
)
return catalog, rows, substrate
def _string(document: dict[str, Any], key: str) -> str:
value = document.get(key)
if not isinstance(value, str) or not value:
raise E30ReviewEvidenceError(f"{key} must be a nonempty string")
return value
def _sha_value(document: dict[str, Any], key: str) -> str:
value = _string(document, key)
if _SHA256.fullmatch(value) is None:
raise E30ReviewEvidenceError(f"{key} must be a SHA-256 digest")
return value
def build_e30_review_router(
*,
materialization_root_provider: RootProvider = lambda: None,
review_pack_root_provider: RootProvider = lambda: None,
) -> APIRouter:
router = APIRouter(prefix="/api/v1/laboratory/e30", tags=["laboratory"])
@router.get("/reviews")
def list_reviews(
limit: int = Query(default=1, ge=1, le=10),
) -> dict[str, object]:
roots = _roots(
materialization_root_provider=materialization_root_provider,
review_pack_root_provider=review_pack_root_provider,
)
if roots is None:
return {
"schema_version": LABORATORY_E30_CATALOG_SCHEMA,
"configured": False,
"items": [],
"candidate_total": 0,
"invalid_total": 0,
"access": "read-only",
}
materialization_root, review_root = roots
candidates = sorted(
(
candidate
for candidate in materialization_root.iterdir()
if candidate.is_dir()
and _MATERIALIZATION_ID.fullmatch(candidate.name) is not None
),
key=lambda candidate: candidate.stat().st_mtime_ns,
reverse=True,
)
items: list[dict[str, object]] = []
invalid_total = 0
for candidate in candidates:
try:
item, _ = _materialization(candidate, review_root)
if len(items) < limit:
items.append(copy.deepcopy(item))
except (E30ReviewEvidenceError, OSError):
invalid_total += 1
return {
"schema_version": LABORATORY_E30_CATALOG_SCHEMA,
"configured": True,
"items": items,
"candidate_total": len(candidates),
"invalid_total": invalid_total,
"access": "read-only",
}
@router.get("/reviews/{result_id}/items")
def list_review_items(
result_id: str,
stratum: Annotated[Stratum, Query()] = "conflict",
limit: Annotated[int, Query(ge=1, le=128)] = 48,
cursor: Annotated[int, Query(ge=0)] = 0,
) -> dict[str, object]:
roots = _roots(
materialization_root_provider=materialization_root_provider,
review_pack_root_provider=review_pack_root_provider,
)
if roots is None:
raise HTTPException(status_code=404, detail="E30 review не найден")
materialization_root, review_root = roots
candidate = _candidate(materialization_root, result_id)
try:
catalog, rows = _materialization(candidate, review_root)
filtered = [row for row in rows if row.get("stratum") == stratum]
page = filtered[cursor : cursor + limit]
next_cursor = cursor + len(page)
return {
"schema_version": LABORATORY_E30_ITEMS_SCHEMA,
"result_id": result_id,
"stratum": stratum,
"items": [e30_review_item_summary(row) for row in page],
"total": len(filtered),
"next_cursor": next_cursor if next_cursor < len(filtered) else None,
"reason_taxonomy": catalog["reason_taxonomy"],
"access": "read-only",
}
except (E30ReviewEvidenceError, OSError) as exc:
raise HTTPException(
status_code=409,
detail="E30 review evidence не прошло проверку целостности",
) from exc
@router.get("/reviews/{result_id}/items/{item_id}")
def get_review_item(
result_id: str,
item_id: str,
) -> dict[str, object]:
roots = _roots(
materialization_root_provider=materialization_root_provider,
review_pack_root_provider=review_pack_root_provider,
)
if roots is None or _REVIEW_ITEM_ID.fullmatch(item_id) is None:
raise HTTPException(status_code=404, detail="E30 review item не найден")
materialization_root, review_root = roots
candidate = _candidate(materialization_root, result_id)
try:
_, rows = _materialization(candidate, review_root)
row = next((value for value in rows if value.get("item_id") == item_id), None)
if row is None:
raise HTTPException(
status_code=404,
detail="E30 review item не найден",
)
return {
"schema_version": LABORATORY_E30_ITEM_DETAIL_SCHEMA,
"result_id": result_id,
"item": _item_detail(candidate, row),
"access": "read-only",
}
except HTTPException:
raise
except (E30ReviewEvidenceError, OSError, ValueError) as exc:
raise HTTPException(
status_code=409,
detail="E30 review item не прошёл проверку целостности",
) from exc
@router.get("/reviews/{result_id}/items/{item_id}/camera-frame")
def get_review_camera_frame(
result_id: str,
item_id: str,
generation: Annotated[str, Query(min_length=64, max_length=64)],
) -> FileResponse:
roots = _roots(
materialization_root_provider=materialization_root_provider,
review_pack_root_provider=review_pack_root_provider,
)
if (
roots is None
or _REVIEW_ITEM_ID.fullmatch(item_id) is None
or _SHA256.fullmatch(generation) is None
):
raise HTTPException(
status_code=404,
detail="E30 camera frame не найден",
)
materialization_root, review_root = roots
candidate = _candidate(materialization_root, result_id)
try:
_, rows = _materialization(candidate, review_root)
row = next(
(value for value in rows if value.get("item_id") == item_id),
None,
)
if row is None or row.get("camera_frame") is None:
raise HTTPException(
status_code=404,
detail="E30 camera frame не найден",
)
path, artifact = _artifact(
candidate,
row["camera_frame"],
expected_role="camera-frame",
)
if artifact.get("sha256") != generation:
raise HTTPException(
status_code=409,
detail="E30 camera generation изменилась",
)
return FileResponse(
path,
media_type="image/jpeg",
filename=f"e30-frame-{artifact['source_frame_index']:06d}.jpg",
headers={
"ETag": f'"{generation}"',
"Cache-Control": "private, max-age=31536000, immutable",
"X-Content-Type-Options": "nosniff",
},
)
except HTTPException:
raise
except (E30ReviewEvidenceError, OSError, ValueError) as exc:
raise HTTPException(
status_code=409,
detail="E30 camera frame не прошёл проверку целостности",
) from exc
return router