SILVER/NODEDC_MISSION_CORE
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

feat(observatory): attest recording equipment and capture profiles

Browse Source
This commit is contained in:
DCCONSTRUCTIONS
2026-09-02 00:57:27 +03:00
parent 7025e173a3
commit d655d6998d
17 changed files with 1089 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/k1link/laboratory/canonical_recorded_migration.py
+35
View File
@@ -0,0 +1,35 @@
"""Exact metadata gate for the one historical recorded-LAB migration.
This compatibility check is intentionally outside the generic Session catalog:
new device plugins and equipment profiles must not add concrete vocabulary to
the shared observation hot path.
"""
from __future__ import annotations
from typing import Final
_DISPLAY_NAME: Final = "RAVNOVES004TREE · полный маршрут восприятия"
_DURATION_SECONDS: Final = 718.0
_RUN_CREATED_AT_UTC: Final = "2026-08-29T18:05:11.329061+00:00"
_PLUGIN_ID: Final = "nodedc.device.xgrids-lixelkity-k1"
_ARCHIVE_ID: Final = "xgrids-k1.viewer-live.evidence"
def matches_historical_recorded_projection(
*,
display_name: object,
duration_seconds: object,
run_created_at_utc: object,
source_plugin_id: object,
source_archive_id: object,
) -> bool:
"""Return whether legacy rows match the only supported rolling upgrade."""
return (
display_name == _DISPLAY_NAME
and duration_seconds == _DURATION_SECONDS
and run_created_at_utc == _RUN_CREATED_AT_UTC
and source_plugin_id == _PLUGIN_ID
and source_archive_id == _ARCHIVE_ID
)
src/k1link/observatory/portable_run_definitions.py
+7 -2
View File
@@ -124,8 +124,8 @@ class ObservationOnlyAuthority:
@dataclass(frozen=True, slots=True)
class K1RecordedSourceRequirements:
"""Typed capability matcher for one class of recorded K1 sessions."""
class RecordedSourceRequirements:
"""Typed capability matcher for one exact recorded capture contract."""
plugin_id: str
archive_id: str
@@ -206,6 +206,11 @@ class K1RecordedSourceRequirements:
}
# Compatibility name retained for promoted package readers. Canonical JSON
# and all digests remain byte-for-byte unchanged.
K1RecordedSourceRequirements = RecordedSourceRequirements
@dataclass(frozen=True, slots=True)
class PortableSourceAdapterIdentity:
adapter_id: str
src/k1link/observatory/portable_setup_projection.py
+51 -2
View File
@@ -26,6 +26,7 @@ from k1link.observatory.source_admission import (
PortableRecordedSourceCapability,
PortableSourceAdmissionError,
)
from k1link.sessions.equipment import EquipmentCaptureRegistry
from k1link.sessions.models import SessionSummary
PORTABLE_LABORATORY_SETUP_CATALOG_SCHEMA: Final = (
@@ -93,12 +94,14 @@ type PortableSourceCapabilityProbe = Callable[[str], PortableRecordedSourceCapab
class _SourceCompatibility:
compatible: bool
capability: PortableRecordedSourceCapability | None
reason_code: str
reason: str
def as_dict(self) -> dict[str, object]:
return {
"outcome": "pass" if self.compatible else "blocked",
"compatible": self.compatible,
"reason_code": self.reason_code,
"reason": self.reason,
}
@@ -112,12 +115,14 @@ class PortableSetupProjector:
registry: PortableRunDefinitionRegistry,
capability_probe: PortableDefinitionCapabilityProbe,
dispatch_available: bool = False,
equipment_capture_registry: EquipmentCaptureRegistry | None = None,
) -> None:
if not hasattr(capability_probe, "probe"):
raise PortableSetupProjectionError("portable source capability probe is unavailable")
self._registry = registry
self._capability_probe = capability_probe
self._dispatch_available = dispatch_available
self._equipment_capture_registry = equipment_capture_registry
for definition in registry.definitions:
_validate_model_presentation(definition)
if definition.authority.as_dict() != _OBSERVATION_ONLY_AUTHORITY:
@@ -159,7 +164,7 @@ class PortableSetupProjector:
except PortableRunDefinitionRegistryError as exc:
raise PortableSetupProjectionError("portable setup is unavailable") from exc
presentation = _presentation(definition)
compatibility = self._probe_source(definition, source.session_id, presentation)
compatibility = self._probe_source(definition, source, presentation)
executor = definition.executor
submission_allowed = (
compatibility.compatible and executor.ready and self._dispatch_available
@@ -204,9 +209,51 @@ class PortableSetupProjector:
def _probe_source(
self,
definition: PortableRunDefinition,
source_session_id: str,
source: SessionSummary,
presentation: dict[str, str],
) -> _SourceCompatibility:
source_session_id = source.session_id
if self._equipment_capture_registry is not None:
expected_profile = (
self._equipment_capture_registry.compatible_profile_for_requirements(
definition.source_requirements.as_dict()
)
)
if expected_profile is None:
raise PortableSetupProjectionError(
"portable source requirements have no capture profile allowlist"
)
attestation = source.capture_attestation
if attestation is None:
return _SourceCompatibility(
compatible=False,
capability=None,
reason_code="capture-attestation-missing",
reason="У записи нет проверенной привязки к оборудованию и профилю записи.",
)
if (
attestation.equipment_model_id
!= expected_profile.equipment.equipment_model_id
or attestation.equipment_model_sha256
!= expected_profile.equipment.equipment_model_sha256
):
return _SourceCompatibility(
compatible=False,
capability=None,
reason_code="equipment-model-mismatch",
reason="Модель оборудования записи не поддерживается этим профилем.",
)
if (
attestation.capture_profile_id != expected_profile.capture_profile_id
or attestation.capture_profile_sha256
!= expected_profile.capture_profile_sha256
):
return _SourceCompatibility(
compatible=False,
capability=None,
reason_code="capture-profile-not-allowed",
reason="Профиль записи не входит в точный список совместимости.",
)
try:
capability = self._capability_probe.probe(
source_session_id=source_session_id,
@@ -217,6 +264,7 @@ class PortableSetupProjector:
return _SourceCompatibility(
compatible=False,
capability=None,
reason_code="source-contract-mismatch",
reason=presentation["incompatible"],
)
if not isinstance(capability, PortableRecordedSourceCapability):
@@ -230,6 +278,7 @@ class PortableSetupProjector:
return _SourceCompatibility(
compatible=True,
capability=capability,
reason_code="source-compatible",
reason=presentation["compatible"],
)
src/k1link/observatory/source_admission.py
+7 -2
View File
@@ -76,8 +76,8 @@ class PortableSourceAdmissionStaleError(PortableSourceAdmissionIntegrityError):
@dataclass(frozen=True, slots=True)
class RecordedK1SourceRequirements:
"""Typed matcher owned by a portable RunDefinition.
class RecordedSourceRequirements:
"""Typed matcher owned by a portable recorded-source RunDefinition.
``camera_init_sha256`` is the exact ISO-BMFF initialization segment for the
admitted codec/resolution profile. Width and height are asserted by that
@@ -173,6 +173,11 @@ class RecordedK1SourceRequirements:
return _sha256(self.adapter_document())
# Compatibility import for already promoted packages. New code uses the
# equipment-neutral name; the serialized contract is unchanged.
RecordedK1SourceRequirements = RecordedSourceRequirements
@dataclass(frozen=True, slots=True)
class PortableRecordedSourceCapability:
"""Cheap catalog/summary attestation used by compatibility surfaces.
src/k1link/sessions/equipment.py
+380
View File
@@ -0,0 +1,380 @@
"""Canonical equipment/capture identities for recorded observation sessions.
The registry is intentionally small and content addressed. It answers only
which equipment contract produced a recording; it does not contain Worker,
model, queue, or command authority.
"""
from __future__ import annotations
import hashlib
import json
import re
from collections.abc import Mapping
from dataclasses import dataclass
from pathlib import Path
from typing import Final, Literal
from .models import ObservationSessionCandidate
EQUIPMENT_REGISTRY_SCHEMA: Final = "missioncore.equipment-model-registry/v1"
EQUIPMENT_MODEL_SCHEMA: Final = "missioncore.equipment-model/v1"
CAPTURE_PROFILE_REGISTRY_SCHEMA: Final = (
"missioncore.recorded-capture-profile-registry/v1"
)
CAPTURE_PROFILE_SCHEMA: Final = "missioncore.recorded-capture-profile/v1"
CAPTURE_ATTESTATION_SCHEMA: Final = "missioncore.recorded-capture-attestation/v1"
_IDENTIFIER = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$")
_SHA256 = re.compile(r"^[a-f0-9]{64}$")
_MODALITIES: Final = frozenset({"point-cloud", "trajectory", "video"})
CaptureAttestationState = Literal[
"attested",
"attested-with-legacy-evidence",
"unknown",
]
class EquipmentRegistryError(ValueError):
"""A canonical equipment or capture registry is invalid."""
def canonical_sha256(document: Mapping[str, object]) -> str:
payload = json.dumps(
document,
ensure_ascii=False,
sort_keys=True,
separators=(",", ":"),
).encode("utf-8")
return hashlib.sha256(payload).hexdigest()
@dataclass(frozen=True, slots=True)
class EquipmentModel:
equipment_model_id: str
equipment_model_sha256: str
document: Mapping[str, object]
@property
def display_name(self) -> str:
return str(self.document["display_name"])
@dataclass(frozen=True, slots=True)
class RecordedCaptureProfile:
capture_profile_id: str
capture_profile_sha256: str
equipment: EquipmentModel
document: Mapping[str, object]
@dataclass(frozen=True, slots=True)
class RecordedCaptureAttestation:
equipment_model_id: str
equipment_model_sha256: str
equipment_display_name: str
capture_profile_id: str
capture_profile_sha256: str
capture_attestation_sha256: str
state: CaptureAttestationState
def as_dict(self) -> dict[str, object]:
return {
"schema_version": CAPTURE_ATTESTATION_SCHEMA,
"equipment_model": {
"equipment_model_id": self.equipment_model_id,
"equipment_model_sha256": self.equipment_model_sha256,
"display_name": self.equipment_display_name,
},
"capture_profile": {
"capture_profile_id": self.capture_profile_id,
"capture_profile_sha256": self.capture_profile_sha256,
},
"capture_attestation_sha256": self.capture_attestation_sha256,
"state": self.state,
}
class EquipmentCaptureRegistry:
"""Exact-match registry used by the host session catalog."""
def __init__(
self,
*,
equipment_models: tuple[EquipmentModel, ...],
capture_profiles: tuple[RecordedCaptureProfile, ...],
) -> None:
self.equipment_models = equipment_models
self.capture_profiles = capture_profiles
self._profiles_by_origin = {
(
str(profile.document["plugin_id"]),
str(profile.document["archive_id"]),
): profile
for profile in capture_profiles
}
if len(self._profiles_by_origin) != len(capture_profiles):
raise EquipmentRegistryError("capture profile origins must be unique")
@classmethod
def from_repository(cls, repository_root: Path) -> EquipmentCaptureRegistry | None:
equipment_path = repository_root / "config" / "observatory-equipment-models.json"
capture_path = (
repository_root / "config" / "observatory-recorded-capture-profiles.json"
)
if not equipment_path.exists() and not capture_path.exists():
return None
if not equipment_path.is_file() or not capture_path.is_file():
raise EquipmentRegistryError("equipment/capture registries must be installed together")
return cls.load(equipment_path=equipment_path, capture_path=capture_path)
@classmethod
def load(
cls,
*,
equipment_path: Path,
capture_path: Path,
) -> EquipmentCaptureRegistry:
equipment_payload = _read_json(equipment_path)
capture_payload = _read_json(capture_path)
if equipment_payload.get("schema_version") != EQUIPMENT_REGISTRY_SCHEMA:
raise EquipmentRegistryError("equipment registry schema is unsupported")
if capture_payload.get("schema_version") != CAPTURE_PROFILE_REGISTRY_SCHEMA:
raise EquipmentRegistryError("capture profile registry schema is unsupported")
equipment_rows = equipment_payload.get("models")
capture_rows = capture_payload.get("profiles")
if not isinstance(equipment_rows, list) or not equipment_rows:
raise EquipmentRegistryError("equipment registry must contain models")
if not isinstance(capture_rows, list) or not capture_rows:
raise EquipmentRegistryError("capture registry must contain profiles")
equipment_models = tuple(_equipment_model(row) for row in equipment_rows)
by_id = {item.equipment_model_id: item for item in equipment_models}
if len(by_id) != len(equipment_models):
raise EquipmentRegistryError("equipment model ids must be unique")
capture_profiles = tuple(_capture_profile(row, by_id) for row in capture_rows)
if len({item.capture_profile_id for item in capture_profiles}) != len(
capture_profiles
):
raise EquipmentRegistryError("capture profile ids must be unique")
return cls(
equipment_models=equipment_models,
capture_profiles=capture_profiles,
)
def attest(
self,
*,
session_id: str,
plugin_id: str,
archive_id: str,
candidate: ObservationSessionCandidate,
) -> RecordedCaptureAttestation | None:
profile = self._profiles_by_origin.get((plugin_id, archive_id))
if profile is None:
return None
modalities_value = profile.document["modalities"]
channels_value = profile.document["semantic_channels"]
assert isinstance(modalities_value, list)
assert isinstance(channels_value, list)
required_modalities = tuple(str(item) for item in modalities_value)
required_channels = {str(item) for item in channels_value}
actual_channels = {source.semantic_channel_id for source in candidate.sources}
camera = profile.document["camera_media"]
assert isinstance(camera, dict)
expected_camera_source = camera["source_id"]
camera_sources = {
source.source_id
for source in candidate.sources
if source.modality == "video" and source.seekable
}
if (
not set(required_modalities).issubset(candidate.modalities)
or not required_channels.issubset(actual_channels)
or expected_camera_source not in camera_sources
or not candidate.replayable
):
return None
attestation_document: dict[str, object] = {
"schema_version": CAPTURE_ATTESTATION_SCHEMA,
"session_id": session_id,
"plugin_id": plugin_id,
"archive_id": archive_id,
"equipment_model_id": profile.equipment.equipment_model_id,
"equipment_model_sha256": profile.equipment.equipment_model_sha256,
"capture_profile_id": profile.capture_profile_id,
"capture_profile_sha256": profile.capture_profile_sha256,
"observed_modalities": sorted(candidate.modalities),
"observed_semantic_channels": sorted(actual_channels),
"state": "attested-with-legacy-evidence",
}
return RecordedCaptureAttestation(
equipment_model_id=profile.equipment.equipment_model_id,
equipment_model_sha256=profile.equipment.equipment_model_sha256,
equipment_display_name=profile.equipment.display_name,
capture_profile_id=profile.capture_profile_id,
capture_profile_sha256=profile.capture_profile_sha256,
capture_attestation_sha256=canonical_sha256(attestation_document),
state="attested-with-legacy-evidence",
)
def compatible_profile_for_requirements(
self,
requirements: Mapping[str, object],
) -> RecordedCaptureProfile | None:
"""Resolve an exact capture allowlist entry without changing RunDefinition bytes."""
for profile in self.capture_profiles:
row = profile.document
camera = row["camera_media"]
calibration = row["calibration"]
semantic_channels = row["semantic_channels"]
assert isinstance(camera, dict)
assert isinstance(calibration, dict)
assert isinstance(semantic_channels, list)
if (
row["plugin_id"] == requirements.get("plugin_id")
and row["archive_id"] == requirements.get("archive_id")
and row["modalities"] == requirements.get("required_modalities")
and camera["source_id"] == requirements.get("camera_source_id")
and "camera.video.recorded" in semantic_channels
and requirements.get("camera_semantic_channel_id")
== "camera.video.recorded"
and camera["media_type"] == requirements.get("recorded_media_type")
and camera["initialization_sha256"]
== requirements.get("recorded_media_init_sha256")
and camera["width"] == requirements.get("camera_width")
and camera["height"] == requirements.get("camera_height")
and calibration["slot_id"] == requirements.get("calibration_slot")
and calibration["sha256"]
== requirements.get("calibration_identity_sha256")
and requirements.get("exactly_one_media_epoch") is True
and requirements.get("seekable") is True
):
return profile
return None
def _read_json(path: Path) -> dict[str, object]:
try:
payload = json.loads(path.read_text(encoding="utf-8"))
except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc:
raise EquipmentRegistryError("equipment/capture registry is unreadable") from exc
if not isinstance(payload, dict):
raise EquipmentRegistryError("equipment/capture registry must be an object")
return payload
def _equipment_model(value: object) -> EquipmentModel:
if not isinstance(value, dict) or value.get("schema_version") != EQUIPMENT_MODEL_SCHEMA:
raise EquipmentRegistryError("equipment model schema is unsupported")
equipment_model_id = _identifier(value.get("equipment_model_id"), "equipment model id")
plugin_id = _identifier(value.get("plugin_id"), "equipment plugin id")
if not plugin_id.startswith("nodedc.device."):
raise EquipmentRegistryError("equipment plugin id is not canonical")
version = value.get("equipment_model_version")
if isinstance(version, bool) or not isinstance(version, int) or version < 1:
raise EquipmentRegistryError("equipment model version is invalid")
capabilities = value.get("capability_ids")
if (
not isinstance(capabilities, list)
or not capabilities
or capabilities != sorted(capabilities)
or len(set(capabilities)) != len(capabilities)
or any(
not isinstance(item, str) or _IDENTIFIER.fullmatch(item) is None
for item in capabilities
)
):
raise EquipmentRegistryError("equipment capabilities are invalid")
authority = value.get("authority")
if authority != {"commands_enabled": False}:
raise EquipmentRegistryError("equipment registry must be observation-only")
for field in ("vendor", "display_name", "category"):
if not isinstance(value.get(field), str) or not str(value[field]).strip():
raise EquipmentRegistryError(f"equipment {field} is invalid")
return EquipmentModel(
equipment_model_id=equipment_model_id,
equipment_model_sha256=canonical_sha256(value),
document=value,
)
def _capture_profile(
value: object,
equipment_models: Mapping[str, EquipmentModel],
) -> RecordedCaptureProfile:
if not isinstance(value, dict) or value.get("schema_version") != CAPTURE_PROFILE_SCHEMA:
raise EquipmentRegistryError("capture profile schema is unsupported")
profile_id = _identifier(value.get("capture_profile_id"), "capture profile id")
version = value.get("capture_profile_version")
if isinstance(version, bool) or not isinstance(version, int) or version < 1:
raise EquipmentRegistryError("capture profile version is invalid")
equipment_id = _identifier(value.get("equipment_model_id"), "capture equipment id")
equipment = equipment_models.get(equipment_id)
if equipment is None:
raise EquipmentRegistryError("capture profile references unknown equipment")
equipment_sha256 = value.get("equipment_model_sha256")
if equipment_sha256 != equipment.equipment_model_sha256:
raise EquipmentRegistryError("capture profile equipment digest does not match")
_identifier(value.get("plugin_id"), "capture plugin id")
_identifier(value.get("archive_id"), "capture archive id")
_identifier(
value.get("firmware_compatibility_profile_id"),
"firmware compatibility profile id",
)
modalities = value.get("modalities")
if (
not isinstance(modalities, list)
or not modalities
or modalities != sorted(modalities)
or not set(modalities).issubset(_MODALITIES)
):
raise EquipmentRegistryError("capture modalities are invalid")
channels = value.get("semantic_channels")
if (
not isinstance(channels, list)
or not channels
or channels != sorted(channels)
or any(
not isinstance(item, str) or _IDENTIFIER.fullmatch(item) is None
for item in channels
)
):
raise EquipmentRegistryError("capture semantic channels are invalid")
camera = value.get("camera_media")
calibration = value.get("calibration")
adapter = value.get("adapter")
if (
not isinstance(camera, dict)
or not isinstance(calibration, dict)
or not isinstance(adapter, dict)
):
raise EquipmentRegistryError("capture nested contracts are invalid")
_identifier(camera.get("source_id"), "capture camera source")
_digest(camera.get("initialization_sha256"), "capture camera initialization")
_identifier(calibration.get("slot_id"), "capture calibration slot")
_digest(calibration.get("sha256"), "capture calibration")
_identifier(adapter.get("adapter_id"), "capture adapter id")
_digest(adapter.get("adapter_sha256"), "capture adapter")
if value.get("media_epoch_policy") != "exactly-one-complete-epoch":
raise EquipmentRegistryError("capture media epoch policy is unsupported")
if value.get("seekable") is not True:
raise EquipmentRegistryError("capture profile must be seekable")
return RecordedCaptureProfile(
capture_profile_id=profile_id,
capture_profile_sha256=canonical_sha256(value),
equipment=equipment,
document=value,
)
def _identifier(value: object, label: str) -> str:
if not isinstance(value, str) or _IDENTIFIER.fullmatch(value) is None:
raise EquipmentRegistryError(f"{label} is invalid")
return value
def _digest(value: object, label: str) -> str:
if not isinstance(value, str) or _SHA256.fullmatch(value) is None:
raise EquipmentRegistryError(f"{label} digest is invalid")
return value
src/k1link/sessions/models.py
+39 -14
View File
@@ -2,11 +2,17 @@ from __future__ import annotations
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Literal
from typing import TYPE_CHECKING, Any, Literal
if TYPE_CHECKING:
from .equipment import RecordedCaptureAttestation
SessionStatus = Literal["ready", "interrupted", "failed"]
SessionModality = Literal["point-cloud", "trajectory", "video"]
LAB_REPLAY_CAPABILITY_SCHEMA = "missioncore.observation-lab-replay-capability/v1"
PORTABLE_LAB_REPLAY_CAPABILITY_SCHEMA = (
"missioncore.observation-lab-replay-capability/v2"
)
class SessionStoreError(RuntimeError):
@@ -33,28 +39,41 @@ class LayoutConflictError(SessionStoreError):
class LabReplayCapability:
"""Explicit, observation-only admission for a derived recorded replay."""
schema_version: Literal["missioncore.observation-lab-replay-capability/v1"]
kind: Literal["canonical-recorded-rerun"]
viewer_profile: Literal["recorded-session"]
timeline: Literal["session_time"]
schema_version: Literal[
"missioncore.observation-lab-replay-capability/v1",
"missioncore.observation-lab-replay-capability/v2",
]
kind: Literal["canonical-recorded-rerun", "portable-result-review"]
viewer_profile: Literal["recorded-session", "portable-result"]
timeline: Literal["session_time", "result-defined"]
activation: Literal["explicit"]
commands_enabled: Literal[False]
def __post_init__(self) -> None:
expected_text = (
LAB_REPLAY_CAPABILITY_SCHEMA,
"canonical-recorded-rerun",
"recorded-session",
"session_time",
"explicit",
)
if (
identity = (
self.schema_version,
self.kind,
self.viewer_profile,
self.timeline,
self.activation,
) != expected_text or self.commands_enabled is not False:
)
admitted = {
(
LAB_REPLAY_CAPABILITY_SCHEMA,
"canonical-recorded-rerun",
"recorded-session",
"session_time",
"explicit",
),
(
PORTABLE_LAB_REPLAY_CAPABILITY_SCHEMA,
"portable-result-review",
"portable-result",
"result-defined",
"explicit",
),
}
if identity not in admitted or self.commands_enabled is not False:
raise ValueError("LAB replay capability is invalid")
def as_dict(self) -> dict[str, object]:
@@ -160,6 +179,7 @@ class SessionSummary:
total_bytes: int
replayable: bool
origin: str
capture_attestation: RecordedCaptureAttestation | None = None
lab: LabSessionBinding | None = None
def as_dict(self) -> dict[str, Any]:
@@ -176,6 +196,11 @@ class SessionSummary:
"total_bytes": self.total_bytes,
"replayable": self.replayable,
"origin": self.origin,
"capture_attestation": (
None
if self.capture_attestation is None
else self.capture_attestation.as_dict()
),
}
if self.lab is not None:
document["lab"] = self.lab.as_dict()
src/k1link/sessions/store.py
+155 -15
View File
@@ -16,7 +16,15 @@ from typing import Any, Literal, cast
from uuid import uuid4
from k1link.artifacts import utc_now_iso
from k1link.laboratory.canonical_recorded_migration import (
matches_historical_recorded_projection,
)
from .equipment import (
CaptureAttestationState,
EquipmentCaptureRegistry,
RecordedCaptureAttestation,
)
from .models import (
LabReplayCapability,
LabSessionBinding,
@@ -72,6 +80,16 @@ CREATE TABLE IF NOT EXISTS observation_sessions (
session_root TEXT NOT NULL,
created_at_utc TEXT NOT NULL,
updated_at_utc TEXT NOT NULL
,equipment_model_id TEXT
,equipment_model_sha256 TEXT
,equipment_display_name TEXT
,capture_profile_id TEXT
,capture_profile_sha256 TEXT
,capture_attestation_sha256 TEXT
,capture_attestation_state TEXT
CHECK (capture_attestation_state IN (
'attested', 'attested-with-legacy-evidence', 'unknown'
))
);
CREATE INDEX IF NOT EXISTS observation_sessions_recent
@@ -167,8 +185,15 @@ class SessionStore:
self.data_dir.chmod(0o700)
self.database_path = self.data_dir / DATABASE_NAME
self._lock = threading.RLock()
self._equipment_registry = EquipmentCaptureRegistry.from_repository(
self.repository_root
)
self._initialize()
@property
def equipment_capture_registry(self) -> EquipmentCaptureRegistry | None:
return self._equipment_registry
def reconcile_archive(self, source: ObservationArchiveSource) -> tuple[str, ...]:
"""Reconcile one plugin-owned evidence namespace into the host catalog."""
@@ -538,6 +563,9 @@ class SessionStore:
else duration_seconds
),
include_recorded_media=include_recorded_media,
require_recorded_replay=(
replay_capability.viewer_profile == "recorded-session"
),
)
if stored_recorded_media_policy is None:
connection.execute(
@@ -565,8 +593,12 @@ class SessionStore:
"replayable, origin, source_count, total_bytes, "
"primary_replay_artifact_id, timeline_origin_epoch_ns, "
"timeline_origin_monotonic_ns, allowed_root, session_root, "
"created_at_utc, updated_at_utc) "
"VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
"created_at_utc, updated_at_utc, equipment_model_id, "
"equipment_model_sha256, equipment_display_name, capture_profile_id, "
"capture_profile_sha256, capture_attestation_sha256, "
"capture_attestation_state) "
"VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, "
"?, ?, ?, ?, ?, ?, ?)",
(
session_id,
source["plugin_id"],
@@ -592,6 +624,13 @@ class SessionStore:
source["session_root"],
published_at,
published_at,
source["equipment_model_id"],
source["equipment_model_sha256"],
source["equipment_display_name"],
source["capture_profile_id"],
source["capture_profile_sha256"],
source["capture_attestation_sha256"],
source["capture_attestation_state"],
),
)
connection.execute(
@@ -972,6 +1011,17 @@ class SessionStore:
("primary_replay_artifact_id", "TEXT"),
("timeline_origin_epoch_ns", "INTEGER"),
("timeline_origin_monotonic_ns", "INTEGER"),
("equipment_model_id", "TEXT"),
("equipment_model_sha256", "TEXT"),
("equipment_display_name", "TEXT"),
("capture_profile_id", "TEXT"),
("capture_profile_sha256", "TEXT"),
("capture_attestation_sha256", "TEXT"),
(
"capture_attestation_state",
"TEXT CHECK (capture_attestation_state IN ("
"'attested', 'attested-with-legacy-evidence', 'unknown'))",
),
):
if name in session_columns:
continue
@@ -1054,6 +1104,16 @@ class SessionStore:
_validate_candidate_replay(candidate, artifacts)
now = utc_now_iso()
modalities_json = json.dumps(list(candidate.modalities), separators=(",", ":"))
capture_attestation = (
None
if self._equipment_registry is None
else self._equipment_registry.attest(
session_id=candidate.session_id,
plugin_id=source.plugin_id,
archive_id=source.archive_id,
candidate=candidate,
)
)
with self._lock, self._connect() as connection:
connection.execute("BEGIN IMMEDIATE")
@@ -1086,8 +1146,12 @@ class SessionStore:
"duration_seconds, modalities_json, replayable, origin, source_count, "
"total_bytes, primary_replay_artifact_id, timeline_origin_epoch_ns, "
"timeline_origin_monotonic_ns, allowed_root, "
"session_root, created_at_utc, updated_at_utc) "
"VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) "
"session_root, created_at_utc, updated_at_utc, equipment_model_id, "
"equipment_model_sha256, equipment_display_name, capture_profile_id, "
"capture_profile_sha256, capture_attestation_sha256, "
"capture_attestation_state) "
"VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, "
"?, ?, ?, ?, ?, ?, ?) "
"ON CONFLICT(session_id) DO UPDATE SET "
"plugin_id = excluded.plugin_id, archive_id = excluded.archive_id, "
"display_name = excluded.display_name, status = excluded.status, "
@@ -1100,6 +1164,13 @@ class SessionStore:
"primary_replay_artifact_id = excluded.primary_replay_artifact_id, "
"timeline_origin_epoch_ns = excluded.timeline_origin_epoch_ns, "
"timeline_origin_monotonic_ns = excluded.timeline_origin_monotonic_ns, "
"equipment_model_id = excluded.equipment_model_id, "
"equipment_model_sha256 = excluded.equipment_model_sha256, "
"equipment_display_name = excluded.equipment_display_name, "
"capture_profile_id = excluded.capture_profile_id, "
"capture_profile_sha256 = excluded.capture_profile_sha256, "
"capture_attestation_sha256 = excluded.capture_attestation_sha256, "
"capture_attestation_state = excluded.capture_attestation_state, "
"updated_at_utc = excluded.updated_at_utc",
(
candidate.session_id,
@@ -1122,6 +1193,37 @@ class SessionStore:
str(session_root),
created_at,
now,
(
None
if capture_attestation is None
else capture_attestation.equipment_model_id
),
(
None
if capture_attestation is None
else capture_attestation.equipment_model_sha256
),
(
None
if capture_attestation is None
else capture_attestation.equipment_display_name
),
(
None
if capture_attestation is None
else capture_attestation.capture_profile_id
),
(
None
if capture_attestation is None
else capture_attestation.capture_profile_sha256
),
(
None
if capture_attestation is None
else capture_attestation.capture_attestation_sha256
),
None if capture_attestation is None else capture_attestation.state,
),
)
connection.execute(
@@ -1246,6 +1348,7 @@ def _validate_existing_lab_projection(
run_created_at_utc: str,
duration_seconds: float | None,
include_recorded_media: bool,
require_recorded_replay: bool = True,
) -> None:
"""Validate the immutable companion snapshot without rereading a mutable source row."""
@@ -1282,7 +1385,7 @@ def _validate_existing_lab_projection(
raise SessionIntegrityError(
"LAB session id is already bound to different catalog metadata"
)
if (
if require_recorded_replay and (
summary["status"] != "ready"
or summary["replayable"] != 1
or summary["primary_replay_artifact_id"] is None
@@ -1546,16 +1649,12 @@ def _migrate_canonical_replay_capabilities(connection: sqlite3.Connection) -> No
"SELECT * FROM observation_sessions WHERE session_id = ?",
(row["source_session_id"],),
).fetchone()
if (
summary["display_name"]
!= "RAVNOVES004TREE · полный маршрут восприятия"
or summary["duration_seconds"] != 718.0
or row["run_created_at_utc"]
!= "2026-08-29T18:05:11.329061+00:00"
or source_summary is None
or source_summary["plugin_id"]
!= "nodedc.device.xgrids-lixelkity-k1"
or source_summary["archive_id"] != "xgrids-k1.viewer-live.evidence"
if source_summary is None or not matches_historical_recorded_projection(
display_name=summary["display_name"],
duration_seconds=summary["duration_seconds"],
run_created_at_utc=row["run_created_at_utc"],
source_plugin_id=source_summary["plugin_id"],
source_archive_id=source_summary["archive_id"],
):
raise SessionIntegrityError(
"canonical rolling LAB projection metadata is invalid"
@@ -1725,6 +1824,9 @@ def _require_capability_owned_lab_projection(
run_created_at_utc=binding.run_created_at_utc,
duration_seconds=summary["duration_seconds"],
include_recorded_media=bool(lab["include_recorded_media"]),
require_recorded_replay=(
binding.replay_capability.viewer_profile == "recorded-session"
),
)
return summary, lab
@@ -1749,10 +1851,48 @@ def _summary_from_row(
total_bytes=row["total_bytes"],
replayable=bool(row["replayable"]),
origin=row["origin"],
capture_attestation=_capture_attestation_from_row(row),
lab=lab,
)
def _capture_attestation_from_row(
row: sqlite3.Row,
) -> RecordedCaptureAttestation | None:
values = (
row["equipment_model_id"],
row["equipment_model_sha256"],
row["equipment_display_name"],
row["capture_profile_id"],
row["capture_profile_sha256"],
row["capture_attestation_sha256"],
row["capture_attestation_state"],
)
if all(value is None for value in values):
return None
if any(not isinstance(value, str) or not value for value in values):
raise SessionIntegrityError("stored capture attestation is incomplete")
state = row["capture_attestation_state"]
if state not in {"attested", "attested-with-legacy-evidence", "unknown"}:
raise SessionIntegrityError("stored capture attestation state is invalid")
for digest in (
row["equipment_model_sha256"],
row["capture_profile_sha256"],
row["capture_attestation_sha256"],
):
if SHA256_PATTERN.fullmatch(digest) is None:
raise SessionIntegrityError("stored capture attestation digest is invalid")
return RecordedCaptureAttestation(
equipment_model_id=row["equipment_model_id"],
equipment_model_sha256=row["equipment_model_sha256"],
equipment_display_name=row["equipment_display_name"],
capture_profile_id=row["capture_profile_id"],
capture_profile_sha256=row["capture_profile_sha256"],
capture_attestation_sha256=row["capture_attestation_sha256"],
state=cast(CaptureAttestationState, state),
)
def _lab_binding_from_row(row: sqlite3.Row) -> LabSessionBinding:
try:
provenance = json.loads(row["provenance_json"])
src/k1link/web/session_api.py
+5
View File
@@ -383,6 +383,11 @@ def build_session_router(
"modalities": list(item.modalities),
"duration_seconds": item.duration_seconds or 0.0,
"replayable": item.replayable,
"capture_attestation": (
None
if item.capture_attestation is None
else item.capture_attestation.as_dict()
),
**(
{
"lab": lab_catalog_document(item, lab_contract)