toki
de4d8f4ff8
세 Agent의 direct route를 동일한 fail-closed preflight와 격리 실행 경계에서 비교하고, 관측되지 않은 preset 셀이 실행되는 것을 막기 위해 연결 계약과 증거 수집 흐름을 고정한다.
679 lines
25 KiB
Python
679 lines
25 KiB
Python
"""Closed, secret-safe connectivity preflight contracts for benchmark cells.
|
|
|
|
This module deliberately has no caller, provider, or network dependency. Caller
|
|
adapters provide typed observations, while this boundary proves that those
|
|
observations exactly match one immutable manifest cell before they can be stored.
|
|
A blocked caller may report that it observed no effective binding at all, but it
|
|
can never report a partial or manifest-derived synthetic one.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import re
|
|
import stat
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Any, Protocol
|
|
|
|
from scripts.agent_benchmark.manifest import (
|
|
CALLER_ENUM,
|
|
ROUTE_KIND_ENUM,
|
|
STAGE_ENUM,
|
|
STAGE_RANK,
|
|
ExpectedBinding,
|
|
MatrixCell,
|
|
TOKEN_RE,
|
|
)
|
|
|
|
|
|
SCHEMA_VERSION = "1"
|
|
IDENTITY_RE = re.compile(r"^sha256:[0-9a-f]{64}$")
|
|
MAX_EVIDENCE_BYTES = 65536
|
|
|
|
# Canonical issue order: registration blockers first, then implementation gaps.
|
|
ISSUE_CODE_ORDER = (
|
|
"credential_missing",
|
|
"model_missing",
|
|
"route_missing",
|
|
"effort_unsupported",
|
|
"endpoint_incompatible",
|
|
"auth_incompatible",
|
|
"protocol_incompatible",
|
|
"stream_incompatible",
|
|
)
|
|
REGISTRATION_ISSUE_CODES = frozenset(ISSUE_CODE_ORDER[:4])
|
|
IMPLEMENTATION_ISSUE_CODES = frozenset(ISSUE_CODE_ORDER[4:])
|
|
ISSUE_CODES = REGISTRATION_ISSUE_CODES | IMPLEMENTATION_ISSUE_CODES
|
|
ISSUE_RANK: dict[str, int] = {code: rank for rank, code in enumerate(ISSUE_CODE_ORDER)}
|
|
# The only resume vocabulary; callers can never attach their own text.
|
|
ISSUE_RESUME_CODES: dict[str, str] = {
|
|
"credential_missing": "register_credential",
|
|
"model_missing": "register_model",
|
|
"route_missing": "register_route",
|
|
"effort_unsupported": "register_effort_support",
|
|
"endpoint_incompatible": "implement_endpoint_adapter",
|
|
"auth_incompatible": "implement_auth_adapter",
|
|
"protocol_incompatible": "implement_protocol_adapter",
|
|
"stream_incompatible": "implement_stream_adapter",
|
|
}
|
|
RESUME_CODES = frozenset(ISSUE_RESUME_CODES.values())
|
|
RESULT_STATUSES = ("ready", "registration_required", "implementation_gap")
|
|
EFFECTIVE_FIELDS = (
|
|
"effective_route_kind",
|
|
"effective_route_id",
|
|
"effective_model",
|
|
"effective_effort",
|
|
)
|
|
|
|
|
|
class ConnectivityError(Exception):
|
|
"""Base class for closed connectivity contract failures."""
|
|
|
|
|
|
class ConnectivityValidationError(ConnectivityError):
|
|
"""Raised when a caller observation or evidence value is not admissible."""
|
|
|
|
|
|
class ConnectivityEvidenceError(ConnectivityError):
|
|
"""Raised when durable evidence cannot be safely written or read."""
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class CallerCapability:
|
|
"""Static public capability claimed by one benchmark caller adapter."""
|
|
|
|
caller: str
|
|
route_kinds: tuple[str, ...]
|
|
efforts: tuple[str, ...]
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class EffectiveBinding:
|
|
"""One observed stage binding, including its exact effective effort."""
|
|
|
|
stage: str
|
|
model: str
|
|
effort: str | None = None
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class RequestedEffectiveBinding:
|
|
"""Requested identity plus one all-or-none observed effective binding.
|
|
|
|
The requested fields are always required. The effective group is either
|
|
fully absent, which only a blocked result may report, or complete and exact.
|
|
"""
|
|
|
|
cell_id: str
|
|
caller: str
|
|
requested_route_kind: str
|
|
requested_route_id: str
|
|
requested_model: str
|
|
requested_effort: str
|
|
effective_route_kind: str | None = None
|
|
effective_route_id: str | None = None
|
|
effective_model: str | None = None
|
|
effective_effort: str | None = None
|
|
effective_bindings: tuple[EffectiveBinding, ...] = ()
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ConnectivityIssue:
|
|
"""Closed blocker code paired with its one permitted resume code."""
|
|
|
|
code: str
|
|
resume_code: str
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ConnectivityResult:
|
|
"""Validated immutable result for exactly one caller/cell preflight."""
|
|
|
|
capability: CallerCapability
|
|
binding: RequestedEffectiveBinding
|
|
issues: tuple[ConnectivityIssue, ...]
|
|
status: str
|
|
|
|
|
|
class CallerPreflight(Protocol):
|
|
"""Adapter boundary; implementations must not return raw caller output."""
|
|
|
|
capability: CallerCapability
|
|
|
|
def preflight(
|
|
self, cell: MatrixCell
|
|
) -> tuple[RequestedEffectiveBinding, tuple[ConnectivityIssue, ...]]:
|
|
"""Return typed, secret-free observation for ``cell``."""
|
|
|
|
|
|
def _fail(message: str) -> None:
|
|
# Messages contain only fixed contract field names, never input values.
|
|
raise ConnectivityValidationError(message)
|
|
|
|
|
|
def _require_identifier(value: Any, field_name: str) -> str:
|
|
if not isinstance(value, str) or not TOKEN_RE.fullmatch(value):
|
|
_fail(f"invalid {field_name}")
|
|
return value
|
|
|
|
|
|
def _require_identity(value: Any, field_name: str) -> str:
|
|
if not isinstance(value, str) or not IDENTITY_RE.fullmatch(value):
|
|
raise ConnectivityEvidenceError(f"invalid {field_name}")
|
|
return value
|
|
|
|
|
|
def _validate_closed_tuple(
|
|
values: Any,
|
|
field_name: str,
|
|
*,
|
|
allowed: tuple[str, ...] | None = None,
|
|
) -> None:
|
|
"""Check shape, item type, membership and uniqueness before canonical order.
|
|
|
|
Ordering is checked last so unknown or unhashable entries can never reach
|
|
``set`` or an enum index lookup and escape as a built-in exception.
|
|
"""
|
|
if not isinstance(values, tuple) or not values:
|
|
_fail(f"invalid {field_name}")
|
|
for item in values:
|
|
if not isinstance(item, str):
|
|
_fail(f"invalid {field_name} item")
|
|
if allowed is None:
|
|
_require_identifier(item, f"{field_name} item")
|
|
elif item not in allowed:
|
|
_fail(f"invalid {field_name} item")
|
|
if len(set(values)) != len(values):
|
|
_fail(f"duplicate {field_name} item")
|
|
if tuple(sorted(values, key=allowed.index if allowed is not None else None)) != values:
|
|
_fail(f"non-canonical {field_name}")
|
|
|
|
|
|
def _validate_capability(capability: CallerCapability) -> None:
|
|
if not isinstance(capability, CallerCapability):
|
|
_fail("invalid caller capability")
|
|
if capability.caller not in CALLER_ENUM:
|
|
_fail("invalid capability caller")
|
|
_validate_closed_tuple(
|
|
capability.route_kinds, "capability route_kinds", allowed=ROUTE_KIND_ENUM
|
|
)
|
|
_validate_closed_tuple(capability.efforts, "capability efforts")
|
|
|
|
|
|
def _validate_binding_shape(binding: RequestedEffectiveBinding) -> bool:
|
|
"""Validate requested fields, then the all-or-none effective group.
|
|
|
|
Returns whether the caller reported an effective observation at all.
|
|
"""
|
|
if not isinstance(binding, RequestedEffectiveBinding):
|
|
_fail("invalid requested/effective binding")
|
|
_require_identifier(binding.cell_id, "binding cell_id")
|
|
if binding.caller not in CALLER_ENUM:
|
|
_fail("invalid binding caller")
|
|
if binding.requested_route_kind not in ROUTE_KIND_ENUM:
|
|
_fail("invalid requested route_kind")
|
|
for name in ("requested_route_id", "requested_model", "requested_effort"):
|
|
_require_identifier(getattr(binding, name), f"binding {name}")
|
|
|
|
if not isinstance(binding.effective_bindings, tuple):
|
|
_fail("invalid effective_bindings")
|
|
observed = tuple(getattr(binding, name) is not None for name in EFFECTIVE_FIELDS)
|
|
if any(observed) != all(observed):
|
|
_fail("partial effective observation")
|
|
if not any(observed):
|
|
if binding.effective_bindings:
|
|
_fail("partial effective observation")
|
|
return False
|
|
|
|
if binding.effective_route_kind not in ROUTE_KIND_ENUM:
|
|
_fail("invalid effective route_kind")
|
|
for name in ("effective_route_id", "effective_model", "effective_effort"):
|
|
_require_identifier(getattr(binding, name), f"binding {name}")
|
|
if not binding.effective_bindings:
|
|
_fail("partial effective observation")
|
|
stages: list[str] = []
|
|
for item in binding.effective_bindings:
|
|
if not isinstance(item, EffectiveBinding):
|
|
_fail("invalid effective binding item")
|
|
if item.stage not in STAGE_ENUM:
|
|
_fail("invalid effective binding stage")
|
|
_require_identifier(item.model, "effective binding model")
|
|
if item.effort is not None:
|
|
_require_identifier(item.effort, "effective binding effort")
|
|
stages.append(item.stage)
|
|
if len(set(stages)) != len(stages):
|
|
_fail("duplicate effective binding stage")
|
|
if stages != sorted(stages, key=STAGE_RANK.__getitem__):
|
|
_fail("non-canonical effective binding order")
|
|
return True
|
|
|
|
|
|
def _expected_as_effective(expected: ExpectedBinding) -> EffectiveBinding:
|
|
return EffectiveBinding(expected.stage, expected.model, expected.effort)
|
|
|
|
|
|
def _cell_binding(cell: MatrixCell) -> tuple[str, str, str, str]:
|
|
return (
|
|
cell.iop.route_kind,
|
|
cell.iop.route_id,
|
|
cell.iop.request_model,
|
|
cell.iop.requested_effort,
|
|
)
|
|
|
|
|
|
def validate_requested_binding(
|
|
cell: MatrixCell,
|
|
capability: CallerCapability,
|
|
binding: RequestedEffectiveBinding,
|
|
) -> None:
|
|
"""Fail closed unless caller identity and requested route/model/effort are exact."""
|
|
if not isinstance(cell, MatrixCell):
|
|
_fail("invalid matrix cell")
|
|
_validate_capability(capability)
|
|
_validate_binding_shape(binding)
|
|
|
|
if capability.caller != cell.caller or binding.caller != cell.caller:
|
|
_fail("caller identity mismatch")
|
|
if binding.cell_id != cell.id:
|
|
_fail("cell identity mismatch")
|
|
if cell.iop.route_kind not in capability.route_kinds:
|
|
_fail("unsupported route_kind")
|
|
if cell.iop.requested_effort not in capability.efforts:
|
|
_fail("unsupported effort")
|
|
|
|
requested = (
|
|
binding.requested_route_kind,
|
|
binding.requested_route_id,
|
|
binding.requested_model,
|
|
binding.requested_effort,
|
|
)
|
|
if requested != _cell_binding(cell):
|
|
_fail("requested binding mismatch")
|
|
|
|
|
|
def validate_effective_binding(
|
|
cell: MatrixCell,
|
|
binding: RequestedEffectiveBinding,
|
|
*,
|
|
required: bool,
|
|
) -> None:
|
|
"""Require the observation only for ``ready``; any present group must be exact."""
|
|
if not isinstance(cell, MatrixCell):
|
|
_fail("invalid matrix cell")
|
|
if not _validate_binding_shape(binding):
|
|
if required:
|
|
_fail("missing effective observation")
|
|
return
|
|
|
|
effective = (
|
|
binding.effective_route_kind,
|
|
binding.effective_route_id,
|
|
binding.effective_model,
|
|
binding.effective_effort,
|
|
)
|
|
if effective != _cell_binding(cell):
|
|
_fail("effective binding substitution")
|
|
expected_stages = tuple(_expected_as_effective(item) for item in cell.iop.expected_bindings)
|
|
if binding.effective_bindings != expected_stages:
|
|
_fail("effective stage binding mismatch")
|
|
|
|
|
|
def validate_binding(
|
|
cell: MatrixCell,
|
|
capability: CallerCapability,
|
|
binding: RequestedEffectiveBinding,
|
|
*,
|
|
require_effective: bool = True,
|
|
) -> None:
|
|
"""Validate the requested phase and then the effective phase of one binding."""
|
|
validate_requested_binding(cell, capability, binding)
|
|
validate_effective_binding(cell, binding, required=require_effective)
|
|
|
|
|
|
def _validate_issue(issue: ConnectivityIssue) -> None:
|
|
if not isinstance(issue, ConnectivityIssue):
|
|
_fail("invalid issue")
|
|
if not isinstance(issue.code, str) or issue.code not in ISSUE_CODES:
|
|
_fail("invalid issue code")
|
|
if not isinstance(issue.resume_code, str) or issue.resume_code != ISSUE_RESUME_CODES[issue.code]:
|
|
_fail("invalid issue resume_code")
|
|
|
|
|
|
def classify_issues(issues: tuple[ConnectivityIssue, ...]) -> str:
|
|
"""Return the only permitted status, with implementation gaps taking precedence."""
|
|
if not isinstance(issues, tuple):
|
|
_fail("issues must be a tuple")
|
|
codes: list[str] = []
|
|
for issue in issues:
|
|
_validate_issue(issue)
|
|
codes.append(issue.code)
|
|
if len(set(codes)) != len(codes):
|
|
_fail("duplicate issue code")
|
|
if codes != sorted(codes, key=ISSUE_RANK.__getitem__):
|
|
_fail("non-canonical issue order")
|
|
if any(code in IMPLEMENTATION_ISSUE_CODES for code in codes):
|
|
return "implementation_gap"
|
|
if any(code in REGISTRATION_ISSUE_CODES for code in codes):
|
|
return "registration_required"
|
|
return "ready"
|
|
|
|
|
|
def make_result(
|
|
cell: MatrixCell,
|
|
capability: CallerCapability,
|
|
binding: RequestedEffectiveBinding,
|
|
issues: tuple[ConnectivityIssue, ...] = (),
|
|
) -> ConnectivityResult:
|
|
"""Construct a result only after phase-exact no-substitution validation."""
|
|
status = classify_issues(issues)
|
|
validate_requested_binding(cell, capability, binding)
|
|
validate_effective_binding(cell, binding, required=status == "ready")
|
|
return ConnectivityResult(capability, binding, issues, status)
|
|
|
|
|
|
def validate_result(cell: MatrixCell, result: ConnectivityResult) -> None:
|
|
"""Revalidate a received result before it is consumed or persisted."""
|
|
if not isinstance(result, ConnectivityResult):
|
|
_fail("invalid connectivity result")
|
|
status = classify_issues(result.issues)
|
|
if result.status not in RESULT_STATUSES or result.status != status:
|
|
_fail("result status mismatch")
|
|
validate_requested_binding(cell, result.capability, result.binding)
|
|
validate_effective_binding(cell, result.binding, required=status == "ready")
|
|
|
|
|
|
def _binding_dict(binding: RequestedEffectiveBinding) -> dict[str, Any]:
|
|
return {
|
|
"cell_id": binding.cell_id,
|
|
"caller": binding.caller,
|
|
"requested_route_kind": binding.requested_route_kind,
|
|
"requested_route_id": binding.requested_route_id,
|
|
"requested_model": binding.requested_model,
|
|
"requested_effort": binding.requested_effort,
|
|
"effective_route_kind": binding.effective_route_kind,
|
|
"effective_route_id": binding.effective_route_id,
|
|
"effective_model": binding.effective_model,
|
|
"effective_effort": binding.effective_effort,
|
|
"effective_bindings": [
|
|
{"stage": item.stage, "model": item.model, "effort": item.effort}
|
|
for item in binding.effective_bindings
|
|
],
|
|
}
|
|
|
|
|
|
def canonical_evidence_bytes(
|
|
cell: MatrixCell,
|
|
result: ConnectivityResult,
|
|
endpoint_identity: str,
|
|
config_identity: str,
|
|
) -> bytes:
|
|
"""Return deterministic schema-closed evidence containing no raw endpoint data."""
|
|
validate_result(cell, result)
|
|
_require_identity(endpoint_identity, "endpoint_identity")
|
|
_require_identity(config_identity, "config_identity")
|
|
payload = {
|
|
"schema_version": SCHEMA_VERSION,
|
|
"cell": {"id": cell.id, "caller": cell.caller},
|
|
"status": result.status,
|
|
"binding": _binding_dict(result.binding),
|
|
"issues": [
|
|
{"code": issue.code, "resume_code": issue.resume_code}
|
|
for issue in result.issues
|
|
],
|
|
"endpoint_identity": endpoint_identity,
|
|
"config_identity": config_identity,
|
|
}
|
|
return json.dumps(payload, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode("ascii") + b"\n"
|
|
|
|
|
|
def _safe_relative_path(relative_path: str | Path) -> Path:
|
|
if not isinstance(relative_path, (str, Path)):
|
|
raise ConnectivityEvidenceError("invalid evidence relative path")
|
|
path = Path(relative_path)
|
|
if (
|
|
path.is_absolute()
|
|
or not path.parts
|
|
or any(part in ("", ".", "..") for part in path.parts)
|
|
or path.suffix != ".json"
|
|
):
|
|
raise ConnectivityEvidenceError("invalid evidence relative path")
|
|
return path
|
|
|
|
|
|
def _require_nofollow_support() -> None:
|
|
if (
|
|
not hasattr(os, "O_NOFOLLOW")
|
|
or not hasattr(os, "O_DIRECTORY")
|
|
or os.open not in os.supports_dir_fd
|
|
or os.mkdir not in os.supports_dir_fd
|
|
):
|
|
raise ConnectivityEvidenceError("evidence no-follow traversal unsupported")
|
|
|
|
|
|
def _open_directory(name: str, dir_fd: int | None, message: str) -> int:
|
|
flags = os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW
|
|
try:
|
|
return os.open(name, flags, dir_fd=dir_fd)
|
|
except OSError as exc:
|
|
raise ConnectivityEvidenceError(message) from exc
|
|
|
|
|
|
def _open_root(root: str | Path) -> int:
|
|
"""Open the evidence root by descending one no-follow component at a time."""
|
|
if not isinstance(root, (str, Path)):
|
|
raise ConnectivityEvidenceError("invalid evidence root")
|
|
path = Path(root)
|
|
parts = list(path.parts)
|
|
if path.is_absolute():
|
|
descriptor = _open_directory(parts[0], None, "invalid evidence root")
|
|
parts = parts[1:]
|
|
else:
|
|
descriptor = _open_directory(".", None, "invalid evidence root")
|
|
try:
|
|
for part in parts:
|
|
if part in ("", ".", ".."):
|
|
raise ConnectivityEvidenceError("invalid evidence root")
|
|
child = _open_directory(part, descriptor, "invalid evidence root")
|
|
os.close(descriptor)
|
|
descriptor = child
|
|
except BaseException:
|
|
os.close(descriptor)
|
|
raise
|
|
return descriptor
|
|
|
|
|
|
def _open_evidence_parent(
|
|
root: str | Path, relative_path: str | Path, *, create: bool
|
|
) -> tuple[int, str]:
|
|
"""Return a descriptor for the verified parent directory and the final name."""
|
|
relative = _safe_relative_path(relative_path)
|
|
_require_nofollow_support()
|
|
descriptor = _open_root(root)
|
|
try:
|
|
for part in relative.parts[:-1]:
|
|
if create:
|
|
try:
|
|
os.mkdir(part, 0o700, dir_fd=descriptor)
|
|
except FileExistsError:
|
|
pass
|
|
except OSError as exc:
|
|
raise ConnectivityEvidenceError("unsafe evidence parent") from exc
|
|
child = _open_directory(part, descriptor, "unsafe evidence parent")
|
|
os.close(descriptor)
|
|
descriptor = child
|
|
except BaseException:
|
|
os.close(descriptor)
|
|
raise
|
|
return descriptor, relative.parts[-1]
|
|
|
|
|
|
def write_evidence(
|
|
root: str | Path,
|
|
relative_path: str | Path,
|
|
cell: MatrixCell,
|
|
result: ConnectivityResult,
|
|
endpoint_identity: str,
|
|
config_identity: str,
|
|
) -> None:
|
|
"""Atomically create evidence once; any existing or symlinked target is rejected."""
|
|
payload = canonical_evidence_bytes(cell, result, endpoint_identity, config_identity)
|
|
parent, name = _open_evidence_parent(root, relative_path, create=True)
|
|
try:
|
|
flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW
|
|
try:
|
|
descriptor = os.open(name, flags, 0o600, dir_fd=parent)
|
|
except FileExistsError as exc:
|
|
raise ConnectivityEvidenceError("evidence target already exists") from exc
|
|
except OSError as exc:
|
|
raise ConnectivityEvidenceError("evidence write rejected") from exc
|
|
try:
|
|
with os.fdopen(descriptor, "wb") as stream:
|
|
stream.write(payload)
|
|
stream.flush()
|
|
os.fsync(stream.fileno())
|
|
except OSError as exc:
|
|
raise ConnectivityEvidenceError("evidence write rejected") from exc
|
|
finally:
|
|
os.close(parent)
|
|
|
|
|
|
def _read_bounded(descriptor: int) -> bytes:
|
|
"""Read at most one byte past the cap so oversized input fails closed."""
|
|
chunks: list[bytes] = []
|
|
remaining = MAX_EVIDENCE_BYTES + 1
|
|
try:
|
|
while remaining > 0:
|
|
chunk = os.read(descriptor, remaining)
|
|
if not chunk:
|
|
break
|
|
chunks.append(chunk)
|
|
remaining -= len(chunk)
|
|
except OSError as exc:
|
|
raise ConnectivityEvidenceError("invalid evidence bytes") from exc
|
|
raw = b"".join(chunks)
|
|
if len(raw) > MAX_EVIDENCE_BYTES:
|
|
raise ConnectivityEvidenceError("evidence target too large")
|
|
return raw
|
|
|
|
|
|
def _read_evidence_bytes(root: str | Path, relative_path: str | Path) -> bytes:
|
|
parent, name = _open_evidence_parent(root, relative_path, create=False)
|
|
try:
|
|
try:
|
|
descriptor = os.open(
|
|
name, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK, dir_fd=parent
|
|
)
|
|
except OSError as exc:
|
|
raise ConnectivityEvidenceError("unsafe evidence target") from exc
|
|
try:
|
|
info = os.fstat(descriptor)
|
|
if not stat.S_ISREG(info.st_mode):
|
|
raise ConnectivityEvidenceError("unsafe evidence target")
|
|
if info.st_size > MAX_EVIDENCE_BYTES:
|
|
raise ConnectivityEvidenceError("evidence target too large")
|
|
return _read_bounded(descriptor)
|
|
finally:
|
|
os.close(descriptor)
|
|
finally:
|
|
os.close(parent)
|
|
|
|
|
|
def _no_duplicate_object(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
|
|
result: dict[str, Any] = {}
|
|
for key, value in pairs:
|
|
if key in result:
|
|
raise ConnectivityEvidenceError("duplicate evidence field")
|
|
result[key] = value
|
|
return result
|
|
|
|
|
|
def _read_binding(data: Any) -> RequestedEffectiveBinding:
|
|
if not isinstance(data, dict) or set(data) != {
|
|
"cell_id", "caller", "requested_route_kind", "requested_route_id",
|
|
"requested_model", "requested_effort", "effective_route_kind",
|
|
"effective_route_id", "effective_model", "effective_effort", "effective_bindings",
|
|
}:
|
|
raise ConnectivityEvidenceError("invalid evidence binding")
|
|
stages = data["effective_bindings"]
|
|
if not isinstance(stages, list):
|
|
raise ConnectivityEvidenceError("invalid evidence binding")
|
|
bindings: list[EffectiveBinding] = []
|
|
for item in stages:
|
|
if not isinstance(item, dict) or set(item) != {"stage", "model", "effort"}:
|
|
raise ConnectivityEvidenceError("invalid evidence binding")
|
|
bindings.append(EffectiveBinding(item["stage"], item["model"], item["effort"]))
|
|
try:
|
|
binding = RequestedEffectiveBinding(
|
|
data["cell_id"], data["caller"], data["requested_route_kind"],
|
|
data["requested_route_id"], data["requested_model"], data["requested_effort"],
|
|
data["effective_route_kind"], data["effective_route_id"],
|
|
data["effective_model"], data["effective_effort"], tuple(bindings),
|
|
)
|
|
_validate_binding_shape(binding)
|
|
except ConnectivityValidationError as exc:
|
|
raise ConnectivityEvidenceError("invalid evidence binding") from exc
|
|
return binding
|
|
|
|
|
|
def _validate_evidence_payload(payload: dict[str, Any], cell: MatrixCell) -> None:
|
|
if not isinstance(cell, MatrixCell):
|
|
raise ConnectivityEvidenceError("invalid matrix cell")
|
|
if not isinstance(payload.get("cell"), dict) or set(payload["cell"]) != {"id", "caller"}:
|
|
raise ConnectivityEvidenceError("invalid evidence cell")
|
|
binding = _read_binding(payload["binding"])
|
|
if payload["cell"] != {"id": binding.cell_id, "caller": binding.caller}:
|
|
raise ConnectivityEvidenceError("evidence cell mismatch")
|
|
if binding.cell_id != cell.id or binding.caller != cell.caller:
|
|
raise ConnectivityEvidenceError("evidence cell mismatch")
|
|
requested = (
|
|
binding.requested_route_kind,
|
|
binding.requested_route_id,
|
|
binding.requested_model,
|
|
binding.requested_effort,
|
|
)
|
|
if requested != _cell_binding(cell):
|
|
raise ConnectivityEvidenceError("requested binding mismatch")
|
|
try:
|
|
validate_effective_binding(cell, binding, required=payload["status"] == "ready")
|
|
except ConnectivityValidationError as exc:
|
|
raise ConnectivityEvidenceError("invalid evidence binding") from exc
|
|
issues_raw = payload["issues"]
|
|
if not isinstance(issues_raw, list):
|
|
raise ConnectivityEvidenceError("invalid evidence issues")
|
|
try:
|
|
issues = tuple(
|
|
ConnectivityIssue(item["code"], item["resume_code"])
|
|
for item in issues_raw
|
|
if isinstance(item, dict) and set(item) == {"code", "resume_code"}
|
|
)
|
|
if len(issues) != len(issues_raw) or classify_issues(issues) != payload["status"]:
|
|
raise ConnectivityValidationError("result status mismatch")
|
|
except (ConnectivityValidationError, KeyError, TypeError) as exc:
|
|
raise ConnectivityEvidenceError("invalid evidence issues") from exc
|
|
|
|
|
|
def read_evidence(
|
|
root: str | Path, relative_path: str | Path, cell: MatrixCell
|
|
) -> dict[str, Any]:
|
|
"""Read only canonical, schema-closed evidence; corruption fails closed."""
|
|
raw = _read_evidence_bytes(root, relative_path)
|
|
try:
|
|
parsed = json.loads(raw.decode("ascii"), object_pairs_hook=_no_duplicate_object)
|
|
except (UnicodeDecodeError, json.JSONDecodeError) as exc:
|
|
raise ConnectivityEvidenceError("invalid evidence bytes") from exc
|
|
if not isinstance(parsed, dict) or set(parsed) != {
|
|
"schema_version", "cell", "status", "binding", "issues",
|
|
"endpoint_identity", "config_identity",
|
|
}:
|
|
raise ConnectivityEvidenceError("invalid evidence schema")
|
|
if parsed.get("schema_version") != SCHEMA_VERSION or parsed.get("status") not in RESULT_STATUSES:
|
|
raise ConnectivityEvidenceError("invalid evidence schema")
|
|
_require_identity(parsed.get("endpoint_identity"), "endpoint_identity")
|
|
_require_identity(parsed.get("config_identity"), "config_identity")
|
|
_validate_evidence_payload(parsed, cell)
|
|
canonical = json.dumps(parsed, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode("ascii") + b"\n"
|
|
if raw != canonical:
|
|
raise ConnectivityEvidenceError("non-canonical evidence")
|
|
return parsed
|