feat(lab): add lazy E40 evidence review
This commit is contained in:
DCCONSTRUCTIONS
@@ -0,0 +1,436 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
from collections.abc import Iterator
|
||||
from contextlib import contextmanager
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Final, Literal
|
||||
|
||||
from k1link.artifacts import utc_now_iso, write_json_atomic
|
||||
|
||||
E40_OPERATOR_REVIEW_SCHEMA: Final = "missioncore.e40-operator-review/v1"
|
||||
E40_OPERATOR_REVIEW_PROTOCOL: Final = "sealed-error-adjudication/v1"
|
||||
|
||||
E40OperatorVerdict = Literal["confirmed-error", "rejected-error"]
|
||||
|
||||
_RESULT_ID = re.compile(r"^e40-perception-product-gate-[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}$")
|
||||
_REVIEW_ID = re.compile(r"^e40-operator-review-[a-f0-9]{64}$")
|
||||
_EVENT_ID = re.compile(r"^e40-operator-verdict-[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}$")
|
||||
_VERDICTS: Final = {"confirmed-error", "rejected-error"}
|
||||
_MAX_JSON_BYTES: Final = 2 * 1024 * 1024
|
||||
|
||||
|
||||
class E40OperatorReviewError(RuntimeError):
|
||||
"""Base error for the E40 operator-review lifecycle."""
|
||||
|
||||
|
||||
class E40OperatorReviewConflictError(E40OperatorReviewError):
|
||||
"""The caller attempted to update a stale review revision."""
|
||||
|
||||
|
||||
class E40OperatorReviewValidationError(E40OperatorReviewError):
|
||||
"""The requested review operation violates the frozen protocol."""
|
||||
|
||||
|
||||
class E40OperatorReviewIntegrityError(E40OperatorReviewError):
|
||||
"""The stored review no longer matches its sealed E40 substrate."""
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class E40OperatorReviewSubject:
|
||||
item_id: str
|
||||
sequence: int
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class E40OperatorReviewSubstrate:
|
||||
result_id: str
|
||||
materialization_id: str
|
||||
case_catalog_sha256: str
|
||||
subjects: tuple[E40OperatorReviewSubject, ...]
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
if (
|
||||
_RESULT_ID.fullmatch(self.result_id) is None
|
||||
or _MATERIALIZATION_ID.fullmatch(self.materialization_id) is None
|
||||
or _SHA256.fullmatch(self.case_catalog_sha256) is None
|
||||
or not self.subjects
|
||||
or len(self.subjects) > 64
|
||||
):
|
||||
raise E40OperatorReviewValidationError(
|
||||
"E40 operator-review source binding is invalid"
|
||||
)
|
||||
if len({subject.item_id for subject in self.subjects}) != len(self.subjects):
|
||||
raise E40OperatorReviewValidationError(
|
||||
"E40 operator-review subjects are not unique"
|
||||
)
|
||||
if any(
|
||||
_ITEM_ID.fullmatch(subject.item_id) is None or subject.sequence < 0
|
||||
for subject in self.subjects
|
||||
):
|
||||
raise E40OperatorReviewValidationError(
|
||||
"E40 operator-review subject is invalid"
|
||||
)
|
||||
|
||||
@property
|
||||
def item_set_sha256(self) -> str:
|
||||
return hashlib.sha256(
|
||||
_canonical_json(
|
||||
[
|
||||
{"item_id": subject.item_id, "sequence": subject.sequence}
|
||||
for subject in self.subjects
|
||||
]
|
||||
)
|
||||
).hexdigest()
|
||||
|
||||
def binding(self) -> dict[str, object]:
|
||||
return {
|
||||
"result_id": self.result_id,
|
||||
"materialization_id": self.materialization_id,
|
||||
"case_catalog_sha256": self.case_catalog_sha256,
|
||||
"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 _safe_root(root: Path) -> Path:
|
||||
root.mkdir(parents=True, exist_ok=True)
|
||||
if root.is_symlink() or not root.is_dir():
|
||||
raise E40OperatorReviewIntegrityError(
|
||||
"E40 operator-review root is invalid"
|
||||
)
|
||||
return root.resolve()
|
||||
|
||||
|
||||
def _reviewer_id(value: str) -> str:
|
||||
value = value.strip()
|
||||
if _REVIEWER_ID.fullmatch(value) is None:
|
||||
raise E40OperatorReviewValidationError("reviewer_id is invalid")
|
||||
return value
|
||||
|
||||
|
||||
def _review_id(
|
||||
substrate: E40OperatorReviewSubstrate,
|
||||
reviewer_id: str,
|
||||
) -> str:
|
||||
identity = {
|
||||
"protocol": E40_OPERATOR_REVIEW_PROTOCOL,
|
||||
"source": substrate.binding(),
|
||||
"reviewer_id": reviewer_id,
|
||||
}
|
||||
return f"e40-operator-review-{hashlib.sha256(_canonical_json(identity)).hexdigest()}"
|
||||
|
||||
|
||||
@contextmanager
|
||||
def _exclusive_lock(path: Path) -> Iterator[None]:
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
try:
|
||||
descriptor = os.open(path, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600)
|
||||
except FileExistsError as exc:
|
||||
raise E40OperatorReviewConflictError(
|
||||
"E40 operator review is being updated"
|
||||
) from exc
|
||||
try:
|
||||
yield
|
||||
finally:
|
||||
os.close(descriptor)
|
||||
path.unlink(missing_ok=True)
|
||||
|
||||
|
||||
class E40OperatorReviewStore:
|
||||
"""Mutable operator verdicts kept separate from immutable E40 evidence."""
|
||||
|
||||
def __init__(self, *, root: Path) -> None:
|
||||
self.root = _safe_root(root)
|
||||
|
||||
def get(
|
||||
self,
|
||||
*,
|
||||
substrate: E40OperatorReviewSubstrate,
|
||||
reviewer_id: str,
|
||||
) -> dict[str, object]:
|
||||
reviewer_id = _reviewer_id(reviewer_id)
|
||||
review_id = _review_id(substrate, reviewer_id)
|
||||
path = self.root / f"{review_id}.json"
|
||||
if path.is_symlink():
|
||||
raise E40OperatorReviewIntegrityError(
|
||||
"E40 operator review must not be a symlink"
|
||||
)
|
||||
if not path.exists():
|
||||
return self._empty(
|
||||
review_id=review_id,
|
||||
substrate=substrate,
|
||||
reviewer_id=reviewer_id,
|
||||
)
|
||||
return self._read(
|
||||
path=path,
|
||||
review_id=review_id,
|
||||
substrate=substrate,
|
||||
reviewer_id=reviewer_id,
|
||||
)
|
||||
|
||||
def record_verdict(
|
||||
self,
|
||||
*,
|
||||
substrate: E40OperatorReviewSubstrate,
|
||||
reviewer_id: str,
|
||||
item_id: str,
|
||||
expected_revision: int,
|
||||
idempotency_key: str,
|
||||
verdict: E40OperatorVerdict,
|
||||
) -> dict[str, object]:
|
||||
reviewer_id = _reviewer_id(reviewer_id)
|
||||
if (
|
||||
_ITEM_ID.fullmatch(item_id) is None
|
||||
or item_id not in {subject.item_id for subject in substrate.subjects}
|
||||
):
|
||||
raise E40OperatorReviewValidationError(
|
||||
"item_id is outside the E40 error catalog"
|
||||
)
|
||||
if (
|
||||
_IDEMPOTENCY_KEY.fullmatch(idempotency_key) is None
|
||||
or verdict not in _VERDICTS
|
||||
):
|
||||
raise E40OperatorReviewValidationError(
|
||||
"E40 operator verdict request is invalid"
|
||||
)
|
||||
if expected_revision < 0:
|
||||
raise E40OperatorReviewValidationError(
|
||||
"expected_revision is invalid"
|
||||
)
|
||||
|
||||
review_id = _review_id(substrate, reviewer_id)
|
||||
path = self.root / f"{review_id}.json"
|
||||
lock_path = self.root / ".locks" / f"{review_id}.lock"
|
||||
with _exclusive_lock(lock_path):
|
||||
current = self.get(
|
||||
substrate=substrate,
|
||||
reviewer_id=reviewer_id,
|
||||
)
|
||||
decisions = list(current["decisions"]) # type: ignore[arg-type]
|
||||
replay = next(
|
||||
(
|
||||
decision
|
||||
for decision in decisions
|
||||
if decision["idempotency_key"] == idempotency_key
|
||||
),
|
||||
None,
|
||||
)
|
||||
if replay is not None:
|
||||
if (
|
||||
replay["item_id"] != item_id
|
||||
or replay["verdict"] != verdict
|
||||
):
|
||||
raise E40OperatorReviewConflictError(
|
||||
"idempotency key was already used for another verdict"
|
||||
)
|
||||
return current
|
||||
if current["revision"] != expected_revision:
|
||||
raise E40OperatorReviewConflictError(
|
||||
"E40 operator-review revision changed"
|
||||
)
|
||||
|
||||
next_revision = expected_revision + 1
|
||||
decided_at = utc_now_iso()
|
||||
event_identity = {
|
||||
"review_id": review_id,
|
||||
"item_id": item_id,
|
||||
"verdict": verdict,
|
||||
"revision": next_revision,
|
||||
"idempotency_key": idempotency_key,
|
||||
"decided_at_utc": decided_at,
|
||||
}
|
||||
decision = {
|
||||
"item_id": item_id,
|
||||
"verdict": verdict,
|
||||
"revision": next_revision,
|
||||
"idempotency_key": idempotency_key,
|
||||
"decided_at_utc": decided_at,
|
||||
"event_id": (
|
||||
"e40-operator-verdict-"
|
||||
f"{hashlib.sha256(_canonical_json(event_identity)).hexdigest()}"
|
||||
),
|
||||
}
|
||||
decisions = [
|
||||
previous
|
||||
for previous in decisions
|
||||
if previous["item_id"] != item_id
|
||||
]
|
||||
decisions.append(decision)
|
||||
subject_order = {
|
||||
subject.item_id: index
|
||||
for index, subject in enumerate(substrate.subjects)
|
||||
}
|
||||
decisions.sort(key=lambda value: subject_order[str(value["item_id"])])
|
||||
payload = self._payload(
|
||||
review_id=review_id,
|
||||
substrate=substrate,
|
||||
reviewer_id=reviewer_id,
|
||||
revision=next_revision,
|
||||
decisions=decisions,
|
||||
updated_at_utc=decided_at,
|
||||
)
|
||||
write_json_atomic(path, payload)
|
||||
path.chmod(0o600)
|
||||
return payload
|
||||
|
||||
def _empty(
|
||||
self,
|
||||
*,
|
||||
review_id: str,
|
||||
substrate: E40OperatorReviewSubstrate,
|
||||
reviewer_id: str,
|
||||
) -> dict[str, object]:
|
||||
return self._payload(
|
||||
review_id=review_id,
|
||||
substrate=substrate,
|
||||
reviewer_id=reviewer_id,
|
||||
revision=0,
|
||||
decisions=[],
|
||||
updated_at_utc=None,
|
||||
)
|
||||
|
||||
def _payload(
|
||||
self,
|
||||
*,
|
||||
review_id: str,
|
||||
substrate: E40OperatorReviewSubstrate,
|
||||
reviewer_id: str,
|
||||
revision: int,
|
||||
decisions: list[dict[str, object]],
|
||||
updated_at_utc: str | None,
|
||||
) -> dict[str, object]:
|
||||
return {
|
||||
"schema_version": E40_OPERATOR_REVIEW_SCHEMA,
|
||||
"review_id": review_id,
|
||||
"protocol": E40_OPERATOR_REVIEW_PROTOCOL,
|
||||
"source": substrate.binding(),
|
||||
"reviewer_id": reviewer_id,
|
||||
"revision": revision,
|
||||
"decisions": decisions,
|
||||
"reviewed_item_count": len(decisions),
|
||||
"remaining_item_count": len(substrate.subjects) - len(decisions),
|
||||
"updated_at_utc": updated_at_utc,
|
||||
"access": "review-write",
|
||||
}
|
||||
|
||||
def _read(
|
||||
self,
|
||||
*,
|
||||
path: Path,
|
||||
review_id: str,
|
||||
substrate: E40OperatorReviewSubstrate,
|
||||
reviewer_id: str,
|
||||
) -> dict[str, object]:
|
||||
if (
|
||||
path.is_symlink()
|
||||
or not path.is_file()
|
||||
or not 0 < path.stat().st_size <= _MAX_JSON_BYTES
|
||||
):
|
||||
raise E40OperatorReviewIntegrityError(
|
||||
"E40 operator review is unavailable"
|
||||
)
|
||||
try:
|
||||
value = json.loads(path.read_text(encoding="utf-8"))
|
||||
except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc:
|
||||
raise E40OperatorReviewIntegrityError(
|
||||
"E40 operator review is invalid"
|
||||
) from exc
|
||||
if not isinstance(value, dict):
|
||||
raise E40OperatorReviewIntegrityError(
|
||||
"E40 operator review must be an object"
|
||||
)
|
||||
decisions = value.get("decisions")
|
||||
revision = value.get("revision")
|
||||
if (
|
||||
value.get("schema_version") != E40_OPERATOR_REVIEW_SCHEMA
|
||||
or value.get("review_id") != review_id
|
||||
or value.get("protocol") != E40_OPERATOR_REVIEW_PROTOCOL
|
||||
or value.get("source") != substrate.binding()
|
||||
or value.get("reviewer_id") != reviewer_id
|
||||
or value.get("access") != "review-write"
|
||||
or isinstance(revision, bool)
|
||||
or not isinstance(revision, int)
|
||||
or revision < 0
|
||||
or not isinstance(decisions, list)
|
||||
or value.get("reviewed_item_count") != len(decisions)
|
||||
or value.get("remaining_item_count")
|
||||
!= len(substrate.subjects) - len(decisions)
|
||||
):
|
||||
raise E40OperatorReviewIntegrityError(
|
||||
"E40 operator review identity is invalid"
|
||||
)
|
||||
known_items = {subject.item_id for subject in substrate.subjects}
|
||||
item_ids: set[str] = set()
|
||||
idempotency_keys: set[str] = set()
|
||||
for decision in decisions:
|
||||
decided_at = decision.get("decided_at_utc") if isinstance(
|
||||
decision, dict
|
||||
) else None
|
||||
event_identity = {
|
||||
"review_id": review_id,
|
||||
"item_id": decision.get("item_id") if isinstance(
|
||||
decision, dict
|
||||
) else None,
|
||||
"verdict": decision.get("verdict") if isinstance(
|
||||
decision, dict
|
||||
) else None,
|
||||
"revision": decision.get("revision") if isinstance(
|
||||
decision, dict
|
||||
) else None,
|
||||
"idempotency_key": decision.get("idempotency_key") if isinstance(
|
||||
decision, dict
|
||||
) else None,
|
||||
"decided_at_utc": decided_at,
|
||||
}
|
||||
expected_event_id = (
|
||||
"e40-operator-verdict-"
|
||||
f"{hashlib.sha256(_canonical_json(event_identity)).hexdigest()}"
|
||||
)
|
||||
if (
|
||||
not isinstance(decision, dict)
|
||||
or decision.get("item_id") not in known_items
|
||||
or decision.get("item_id") in item_ids
|
||||
or decision.get("verdict") not in _VERDICTS
|
||||
or not isinstance(decision.get("revision"), int)
|
||||
or not 1 <= int(decision["revision"]) <= revision
|
||||
or not isinstance(decision.get("decided_at_utc"), str)
|
||||
or _EVENT_ID.fullmatch(str(decision.get("event_id"))) is None
|
||||
or decision.get("event_id") != expected_event_id
|
||||
or _IDEMPOTENCY_KEY.fullmatch(
|
||||
str(decision.get("idempotency_key"))
|
||||
)
|
||||
is None
|
||||
or decision.get("idempotency_key") in idempotency_keys
|
||||
):
|
||||
raise E40OperatorReviewIntegrityError(
|
||||
"E40 operator verdict is invalid"
|
||||
)
|
||||
item_ids.add(str(decision["item_id"]))
|
||||
idempotency_keys.add(str(decision["idempotency_key"]))
|
||||
if decisions and max(
|
||||
int(decision["revision"]) for decision in decisions
|
||||
) != revision:
|
||||
raise E40OperatorReviewIntegrityError(
|
||||
"E40 operator-review revision is invalid"
|
||||
)
|
||||
return value
|
||||
Reference in New Issue
Block a user