diff --git a/src/k1link/compute/__init__.py b/src/k1link/compute/__init__.py new file mode 100644 index 0000000..e39194d --- /dev/null +++ b/src/k1link/compute/__init__.py @@ -0,0 +1,15 @@ +"""Host-owned compute handoff contracts.""" + +from .jobs import ( + COMPUTE_JOB_SCHEMA, + CameraComputeJob, + prepare_camera_compute_job, + validate_camera_compute_job, +) + +__all__ = [ + "COMPUTE_JOB_SCHEMA", + "CameraComputeJob", + "prepare_camera_compute_job", + "validate_camera_compute_job", +] diff --git a/src/k1link/compute/jobs.py b/src/k1link/compute/jobs.py new file mode 100644 index 0000000..6c647a8 --- /dev/null +++ b/src/k1link/compute/jobs.py @@ -0,0 +1,414 @@ +from __future__ import annotations + +import hashlib +import json +import os +import re +import secrets +import shutil +import stat +from dataclasses import dataclass +from pathlib import Path, PurePosixPath +from typing import Any + +from k1link.artifacts import write_json_atomic +from k1link.sessions import SessionIntegrityError, inspect_recorded_media_epoch + +COMPUTE_JOB_SCHEMA = "missioncore.compute-job/v1" +COMPUTE_RESULT_SCHEMA = "missioncore.compute-result/v1" +COMPUTE_PROFILE = "recorded-camera-perception/v1" +MAX_JOB_MANIFEST_BYTES = 32 * 1024 * 1024 +MAX_JOB_FILES = 500_003 + +_SAFE_COMPONENT = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$") +_SAFE_JOB_ID = re.compile(r"^recorded-camera-[a-f0-9]{24}$") +_SHA256 = re.compile(r"^[a-f0-9]{64}$") + + +@dataclass(frozen=True, slots=True) +class CameraComputeJob: + job_id: str + job_root: Path + manifest_path: Path + input_sha256: str + input_byte_length: int + segment_count: int + timeline_start_seconds: float + timeline_end_seconds: float + + +def prepare_camera_compute_job( + *, + session_root: Path, + source_id: str, + codec_epoch: int, + origin_epoch_ns: int, + origin_monotonic_ns: int, + output_root: Path, +) -> CameraComputeJob: + """Publish one immutable, bounded camera epoch as a compute job. + + The source evidence is read and validated but never rewritten. Publication + copies the canonical epoch into a private staging directory, revalidates the + copy, writes a path-free content-addressed manifest, then atomically renames + the complete job into place. + """ + + if _SAFE_COMPONENT.fullmatch(source_id) is None: + raise ValueError("camera source id is not a safe component") + if codec_epoch < 1: + raise ValueError("camera codec epoch must be positive") + if origin_epoch_ns < 0 or origin_monotonic_ns < 0: + raise ValueError("session timeline origin must be non-negative") + + session = session_root.expanduser().resolve(strict=True) + if not session.is_dir() or _SAFE_COMPONENT.fullmatch(session.name) is None: + raise ValueError("observation session root is invalid") + source = (session / "media" / source_id).resolve(strict=True) + if not source.is_dir() or not source.is_relative_to(session): + raise SessionIntegrityError("camera source escapes its observation session") + epoch = (source / f"epoch-{codec_epoch}").resolve(strict=True) + if not epoch.is_dir() or epoch.parent != source: + raise SessionIntegrityError("camera epoch escapes its source") + + inspected = inspect_recorded_media_epoch( + epoch, + expected_source_name=source_id, + origin_epoch_ns=origin_epoch_ns, + origin_monotonic_ns=origin_monotonic_ns, + ) + file_records = _epoch_file_records(epoch, inspected.segments) + input_document = _input_document( + session_id=session.name, + source_id=source_id, + codec_epoch=codec_epoch, + inspected=inspected, + file_records=file_records, + ) + input_sha256 = hashlib.sha256(_canonical_json(input_document)).hexdigest() + job_id = f"recorded-camera-{input_sha256[:24]}" + manifest = _job_manifest(job_id, input_sha256, input_document) + + root = _prepare_private_directory(output_root) + final = root / job_id + if final.exists(): + existing = validate_camera_compute_job(final) + if existing.input_sha256 != input_sha256: + raise SessionIntegrityError("compute job id collides with different input") + return existing + + staging = root / f".tmp-{secrets.token_hex(16)}" + staging.mkdir(mode=0o700) + published = False + try: + destination_epoch = staging / "input" / "camera" / source_id / f"epoch-{codec_epoch}" + destination_epoch.mkdir(mode=0o700, parents=True) + (destination_epoch / "segments").mkdir(mode=0o700) + for record in file_records: + relative = PurePosixPath(str(record["path"])) + source_file = epoch / Path(*relative.parts[4:]) + destination = staging / Path(*relative.parts) + _copy_regular_file(source_file, destination) + + copied = inspect_recorded_media_epoch( + destination_epoch, + expected_source_name=source_id, + origin_epoch_ns=origin_epoch_ns, + origin_monotonic_ns=origin_monotonic_ns, + ) + copied_records = _epoch_file_records(destination_epoch, copied.segments) + copied_input = _input_document( + session_id=session.name, + source_id=source_id, + codec_epoch=codec_epoch, + inspected=copied, + file_records=copied_records, + ) + if _canonical_json(copied_input) != _canonical_json(input_document): + raise SessionIntegrityError("camera compute staging copy changed") + + write_json_atomic(staging / "job.json", manifest) + _fsync_tree(staging) + os.replace(staging, final) + _fsync_directory(root) + published = True + finally: + if not published and staging.exists(): + shutil.rmtree(staging) + + return validate_camera_compute_job(final) + + +def validate_camera_compute_job(job_root: Path) -> CameraComputeJob: + """Validate a published compute job without trusting local paths in JSON.""" + + root = job_root.expanduser().resolve(strict=True) + if not root.is_dir() or _SAFE_JOB_ID.fullmatch(root.name) is None: + raise SessionIntegrityError("compute job root is invalid") + manifest_path = root / "job.json" + manifest = _read_json_object(manifest_path, root) + if ( + manifest.get("schema_version") != COMPUTE_JOB_SCHEMA + or manifest.get("job_id") != root.name + or manifest.get("profile") != COMPUTE_PROFILE + or manifest.get("result_contract") + != { + "schema_version": COMPUTE_RESULT_SCHEMA, + "timestamp_basis": "session-time-seconds", + } + ): + raise SessionIntegrityError("compute job manifest is incompatible") + input_sha256 = manifest.get("input_sha256") + input_document = manifest.get("input") + if ( + not isinstance(input_sha256, str) + or _SHA256.fullmatch(input_sha256) is None + or not isinstance(input_document, dict) + or hashlib.sha256(_canonical_json(input_document)).hexdigest() != input_sha256 + or root.name != f"recorded-camera-{input_sha256[:24]}" + ): + raise SessionIntegrityError("compute job input generation is inconsistent") + + source_id = input_document.get("source_id") + codec_epoch = input_document.get("codec_epoch") + timeline = input_document.get("timeline") + files = input_document.get("files") + if ( + not isinstance(source_id, str) + or _SAFE_COMPONENT.fullmatch(source_id) is None + or not isinstance(codec_epoch, int) + or isinstance(codec_epoch, bool) + or codec_epoch < 1 + or not isinstance(timeline, dict) + or not isinstance(files, list) + or not 4 <= len(files) <= MAX_JOB_FILES + ): + raise SessionIntegrityError("compute job input descriptor is invalid") + + expected_prefix = PurePosixPath("input") / "camera" / source_id / f"epoch-{codec_epoch}" + seen: set[str] = set() + total_bytes = 0 + for record in files: + if not isinstance(record, dict): + raise SessionIntegrityError("compute job file descriptor is invalid") + relative = _safe_relative_path(record.get("path"), expected_prefix) + encoded = relative.as_posix() + if encoded in seen: + raise SessionIntegrityError("compute job file descriptor is duplicated") + seen.add(encoded) + expected_bytes = record.get("byte_length") + expected_sha256 = record.get("sha256") + if ( + not isinstance(expected_bytes, int) + or isinstance(expected_bytes, bool) + or expected_bytes < 1 + or not isinstance(expected_sha256, str) + or _SHA256.fullmatch(expected_sha256) is None + ): + raise SessionIntegrityError("compute job file identity is invalid") + payload_path = root / Path(*relative.parts) + metadata = _confined_regular_file(payload_path, root) + if metadata.st_size != expected_bytes or _sha256_file(payload_path) != expected_sha256: + raise SessionIntegrityError("compute job payload identity changed") + total_bytes += expected_bytes + + required = { + (expected_prefix / "summary.json").as_posix(), + (expected_prefix / "index.jsonl").as_posix(), + (expected_prefix / "init.mp4").as_posix(), + } + segment_count = input_document.get("segment_count") + if not isinstance(segment_count, int) or isinstance(segment_count, bool) or segment_count < 1: + raise SessionIntegrityError("compute job segment count is invalid") + required.update( + (expected_prefix / "segments" / f"{sequence}.m4s").as_posix() + for sequence in range(1, segment_count + 1) + ) + if seen != required or input_document.get("byte_length") != total_bytes: + raise SessionIntegrityError("compute job payload set is inconsistent") + + start = timeline.get("start_seconds") + end = timeline.get("end_seconds") + if not isinstance(start, (int, float)) or not isinstance(end, (int, float)): + raise SessionIntegrityError("compute job timeline is invalid") + return CameraComputeJob( + job_id=root.name, + job_root=root, + manifest_path=manifest_path, + input_sha256=input_sha256, + input_byte_length=total_bytes, + segment_count=segment_count, + timeline_start_seconds=float(start), + timeline_end_seconds=float(end), + ) + + +def _input_document( + *, + session_id: str, + source_id: str, + codec_epoch: int, + inspected: Any, + file_records: list[dict[str, Any]], +) -> dict[str, Any]: + summary_record = next( + record for record in file_records if record["path"].endswith("summary.json") + ) + index_record = next(record for record in file_records if record["path"].endswith("index.jsonl")) + return { + "kind": "canonical-camera-epoch", + "session_id": session_id, + "source_id": source_id, + "codec_epoch": codec_epoch, + "synchronization": "host-arrival-best-effort", + "media_type": inspected.media_type, + "timeline": { + "basis": "session-time-seconds", + "start_seconds": inspected.timeline_start_seconds, + "end_seconds": inspected.timeline_end_seconds, + }, + "segment_count": len(inspected.segments), + "byte_length": sum(int(record["byte_length"]) for record in file_records), + "archive_summary_sha256": summary_record["sha256"], + "archive_index_sha256": index_record["sha256"], + "files": file_records, + } + + +def _job_manifest(job_id: str, input_sha256: str, input_document: dict[str, Any]) -> dict[str, Any]: + return { + "schema_version": COMPUTE_JOB_SCHEMA, + "job_id": job_id, + "profile": COMPUTE_PROFILE, + "input_sha256": input_sha256, + "input": input_document, + "result_contract": { + "schema_version": COMPUTE_RESULT_SCHEMA, + "timestamp_basis": "session-time-seconds", + }, + } + + +def _epoch_file_records(epoch: Path, segments: tuple[Any, ...]) -> list[dict[str, Any]]: + prefix = PurePosixPath("input") / "camera" / epoch.parent.name / epoch.name + paths = [epoch / "summary.json", epoch / "index.jsonl", epoch / "init.mp4"] + paths.extend(segment.path for segment in segments) + records: list[dict[str, Any]] = [] + for path in paths: + metadata = _confined_regular_file(path, epoch) + relative_inside_epoch = path.relative_to(epoch) + records.append( + { + "path": (prefix / relative_inside_epoch.as_posix()).as_posix(), + "byte_length": metadata.st_size, + "sha256": _sha256_file(path), + } + ) + return records + + +def _prepare_private_directory(path: Path) -> Path: + candidate = path.expanduser() + candidate.mkdir(mode=0o700, parents=True, exist_ok=True) + metadata = candidate.lstat() + if stat.S_ISLNK(metadata.st_mode) or not stat.S_ISDIR(metadata.st_mode): + raise ValueError("compute output root must be a real directory") + root = candidate.resolve(strict=True) + os.chmod(root, 0o700) + return root + + +def _copy_regular_file(source: Path, destination: Path) -> None: + confinement_root = source.parents[1] if source.parent.name == "segments" else source.parent + _confined_regular_file(source, confinement_root) + destination.parent.mkdir(mode=0o700, parents=True, exist_ok=True) + with source.open("rb") as reader, destination.open("xb") as writer: + shutil.copyfileobj(reader, writer, length=1024 * 1024) + writer.flush() + os.fsync(writer.fileno()) + os.chmod(destination, 0o600) + + +def _read_json_object(path: Path, root: Path) -> dict[str, Any]: + metadata = _confined_regular_file(path, root) + if not 0 < metadata.st_size <= MAX_JOB_MANIFEST_BYTES: + raise SessionIntegrityError("compute job manifest is outside bounds") + try: + value = json.loads(path.read_text(encoding="utf-8")) + except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc: + raise SessionIntegrityError("compute job manifest is unavailable") from exc + if not isinstance(value, dict): + raise SessionIntegrityError("compute job manifest is not an object") + return value + + +def _safe_relative_path(value: object, expected_prefix: PurePosixPath) -> PurePosixPath: + if not isinstance(value, str) or not value or "\\" in value: + raise SessionIntegrityError("compute job payload path is invalid") + relative = PurePosixPath(value) + if ( + relative.is_absolute() + or ".." in relative.parts + or relative.parts[: len(expected_prefix.parts)] != expected_prefix.parts + ): + raise SessionIntegrityError("compute job payload path escapes its root") + return relative + + +def _confined_regular_file(path: Path, root: Path) -> os.stat_result: + try: + resolved_root = root.resolve(strict=True) + resolved = path.resolve(strict=True) + metadata = path.lstat() + except OSError as exc: + raise SessionIntegrityError("compute job payload is unavailable") from exc + if ( + stat.S_ISLNK(metadata.st_mode) + or not stat.S_ISREG(metadata.st_mode) + or not resolved.is_relative_to(resolved_root) + ): + raise SessionIntegrityError("compute job payload is not a confined regular file") + return metadata + + +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 _canonical_json(value: object) -> bytes: + try: + payload = json.dumps( + value, + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + allow_nan=False, + ).encode("utf-8") + except (TypeError, ValueError) as exc: + raise SessionIntegrityError("compute job descriptor cannot be encoded") from exc + if not 0 < len(payload) <= MAX_JOB_MANIFEST_BYTES: + raise SessionIntegrityError("compute job descriptor is outside bounds") + return payload + + +def _fsync_tree(root: Path) -> None: + directories = sorted( + (path for path in root.rglob("*") if path.is_dir()), + key=lambda path: len(path.parts), + reverse=True, + ) + for directory in directories: + _fsync_directory(directory) + _fsync_directory(root) + + +def _fsync_directory(path: Path) -> None: + descriptor = os.open(path, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0)) + try: + os.fsync(descriptor) + finally: + os.close(descriptor) diff --git a/src/k1link/device_plugins/xgrids_k1/cli.py b/src/k1link/device_plugins/xgrids_k1/cli.py index c3ff939..c386f8f 100644 --- a/src/k1link/device_plugins/xgrids_k1/cli.py +++ b/src/k1link/device_plugins/xgrids_k1/cli.py @@ -17,11 +17,13 @@ from rich.table import Table from k1link import __version__ from k1link.artifacts import write_json_atomic +from k1link.compute import prepare_camera_compute_job from k1link.device_plugins.xgrids_k1.analyze import ( DEFAULT_STREAM_SUMMARY_MAX_PAYLOAD_BYTES, MAX_STREAM_SUMMARY_PAYLOAD_BYTES, summarize_mqtt_streams, ) +from k1link.device_plugins.xgrids_k1.archive import discover_legacy_viewer_sessions from k1link.device_plugins.xgrids_k1.ble.gatt import dump_metadata from k1link.device_plugins.xgrids_k1.ble.scanner import scan from k1link.device_plugins.xgrids_k1.ble.wifi_provisioning import ( @@ -42,6 +44,7 @@ from k1link.device_plugins.xgrids_k1.protocol.application_authority import ( ) from k1link.device_plugins.xgrids_k1.usb.snapshot import snapshot as usb_snapshot from k1link.macos_credentials import CredentialDialogError, prompt_wifi_credentials +from k1link.sessions import SessionIntegrityError app = typer.Typer( name="k1link", @@ -57,11 +60,16 @@ authority_app = typer.Typer( help="One-time local authority administration; never sends a K1 command.", no_args_is_help=True, ) +compute_app = typer.Typer( + help="Bounded, content-addressed handoffs to external compute workers.", + no_args_is_help=True, +) app.add_typer(ble_app, name="ble") app.add_typer(net_app, name="net") app.add_typer(usb_app, name="usb") app.add_typer(analyze_app, name="analyze") app.add_typer(authority_app, name="authority") +app.add_typer(compute_app, name="compute") class ToolStatus(TypedDict): @@ -579,5 +587,71 @@ def analyze_mqtt_streams( console.print(f"Saved: {out}") +@compute_app.command("prepare-camera-job") +def prepare_camera_job( + session_root: Annotated[ + Path, + typer.Option( + "--session-root", + exists=True, + file_okay=False, + dir_okay=True, + readable=True, + help="Sealed observation session containing the canonical camera epoch.", + ), + ], + source_id: Annotated[ + str, + typer.Option("--source", help="Canonical archived camera source id."), + ], + codec_epoch: Annotated[ + int, + typer.Option("--epoch", min=1, help="Physical camera codec epoch number."), + ], + out_root: Annotated[ + Path, + typer.Option( + "--out-root", + help="Ignored private root where the immutable job directory is published.", + ), + ], +) -> None: + """Validate and package one archived camera epoch without touching the K1.""" + + session = session_root.expanduser().resolve() + candidates = { + candidate.session_id: candidate + for candidate in discover_legacy_viewer_sessions(session.parent) + } + candidate = candidates.get(session.name) + if candidate is None or candidate.session_root != session: + console.print("[red]Compute job failed:[/red] session is not a sealed archive") + raise typer.Exit(code=2) + if all(media.source_id != source_id for media in candidate.media_sources): + console.print("[red]Compute job failed:[/red] camera source is not cataloged") + raise typer.Exit(code=2) + try: + job = prepare_camera_compute_job( + session_root=session, + source_id=source_id, + codec_epoch=codec_epoch, + origin_epoch_ns=candidate.timeline_origin_epoch_ns, + origin_monotonic_ns=candidate.timeline_origin_monotonic_ns, + output_root=out_root, + ) + except (OSError, SessionIntegrityError, ValueError) as exc: + console.print(f"[red]Compute job failed:[/red] {type(exc).__name__}: {exc}") + raise typer.Exit(code=2) from exc + + console.print(f"[green]Compute job ready:[/green] {job.job_id}") + console.print( + f"Segments: {job.segment_count}; bytes: {job.input_byte_length}; " + f"session time: {job.timeline_start_seconds:.6f}–" + f"{job.timeline_end_seconds:.6f} s" + ) + console.print(f"Input SHA-256: {job.input_sha256}") + console.print(f"Saved: {job.job_root}") + + if __name__ == "__main__": app() diff --git a/src/k1link/sessions/__init__.py b/src/k1link/sessions/__init__.py index 4191857..d756b83 100644 --- a/src/k1link/sessions/__init__.py +++ b/src/k1link/sessions/__init__.py @@ -10,6 +10,7 @@ from .media import ( RecordedMediaFile, RecordedMediaInspector, RecordedMediaManifest, + inspect_recorded_media_epoch, validate_recorded_media_timeline, ) from .models import ( @@ -64,6 +65,7 @@ __all__ = [ "RecordedMediaFile", "RecordedMediaInspector", "RecordedMediaManifest", + "inspect_recorded_media_epoch", "ReplayCommand", "ReplayArtifact", "RecordingExporter", diff --git a/src/k1link/sessions/media.py b/src/k1link/sessions/media.py index 4b4301f..db96cc1 100644 --- a/src/k1link/sessions/media.py +++ b/src/k1link/sessions/media.py @@ -331,6 +331,40 @@ class RecordedMediaInspector: ) +def inspect_recorded_media_epoch( + epoch: Path, + *, + expected_source_name: str, + origin_epoch_ns: int, + origin_monotonic_ns: int, +) -> RecordedMediaEpoch: + """Validate one canonical camera epoch for a bounded downstream handoff. + + This is the single-epoch form of :class:`RecordedMediaInspector`. It keeps + compute-job packaging on the same digest and ISO-BMFF timing contract as + saved-session replay without requiring the worker to see the rest of the + observation session. + """ + + match = _EPOCH_PATTERN.fullmatch(epoch.name) + if match is None: + raise SessionIntegrityError("recorded media epoch name is invalid") + try: + resolved = epoch.resolve(strict=True) + source = resolved.parent.resolve(strict=True) + except OSError as exc: + raise SessionIntegrityError("recorded media epoch is unavailable") from exc + if source.name != expected_source_name or resolved.parent != source: + raise SessionIntegrityError("recorded media epoch source is inconsistent") + return _read_epoch( + resolved, + ordinal=int(match.group(1)), + expected_source_name=expected_source_name, + origin_epoch_ns=origin_epoch_ns, + origin_monotonic_ns=origin_monotonic_ns, + ) + + def validate_recorded_media_timeline( manifests: tuple[RecordedMediaManifest, ...], *, @@ -863,19 +897,37 @@ def _read_epoch( segment_count = summary.get("segment_count") if not _non_negative_int(segment_count) or not 1 <= int(segment_count) <= MAX_MEDIA_SEGMENTS: raise SessionIntegrityError("recorded media segment count is invalid") + match = _EPOCH_PATTERN.fullmatch(epoch.name) + if ( + match is None + or summary.get("codec_epoch") != int(match.group(1)) + or summary.get("entry_count") != int(segment_count) + or summary.get("media_segment_count") != int(segment_count) + or summary.get("commit_policy") != "per-segment-fsync" + or summary.get("artifacts") + != { + "init": "init.mp4", + "segments": "segments", + "index": "index.jsonl", + } + ): + raise SessionIntegrityError("recorded media summary aggregate is inconsistent") try: - raw_lines = ( - _read_confined_file( - epoch / "index.jsonl", - epoch, - MAX_MEDIA_INDEX_BYTES, - ) - .decode("utf-8") - .splitlines() + raw_index = _read_confined_file( + epoch / "index.jsonl", + epoch, + MAX_MEDIA_INDEX_BYTES, ) + raw_lines = raw_index.decode("utf-8").splitlines() except UnicodeDecodeError as exc: raise SessionIntegrityError("recorded media index is unavailable") from exc + expected_index_sha256 = summary.get("index_sha256") + if ( + not isinstance(expected_index_sha256, str) + or hashlib.sha256(raw_index).hexdigest() != expected_index_sha256 + ): + raise SessionIntegrityError("recorded media index digest changed") if len(raw_lines) != int(segment_count): raise SessionIntegrityError("recorded media index length does not match its summary") entries: list[dict[str, Any]] = [] @@ -962,6 +1014,8 @@ def _read_epoch( media_type = _mp4_media_type(init_payload) timing = _mp4_video_timing(init_payload, _Mp4ParseBudget()) fragment_timings: list[_Mp4VideoFragmentTiming] = [] + stream_sha256 = hashlib.sha256(init_payload) + valid_bytes = len(init_payload) for segment in segments: payload = _read_confined_file( segment.path, @@ -970,6 +1024,8 @@ def _read_epoch( ) if hashlib.sha256(payload).hexdigest() != segment.sha256: raise SessionIntegrityError("recorded media segment digest changed") + stream_sha256.update(payload) + valid_bytes += len(payload) fragment_timings.append( _mp4_video_fragment_timing( payload, @@ -977,6 +1033,11 @@ def _read_epoch( _Mp4ParseBudget(), ) ) + if ( + summary.get("valid_bytes") != valid_bytes + or summary.get("stream_sha256") != stream_sha256.hexdigest() + ): + raise SessionIntegrityError("recorded media stream aggregate changed") for previous, current in zip( fragment_timings, fragment_timings[1:], diff --git a/tests/test_compute_jobs.py b/tests/test_compute_jobs.py new file mode 100644 index 0000000..bc75417 --- /dev/null +++ b/tests/test_compute_jobs.py @@ -0,0 +1,141 @@ +from __future__ import annotations + +import hashlib +import json +from pathlib import Path + +import pytest + +from k1link.compute import prepare_camera_compute_job, validate_camera_compute_job +from k1link.sessions import SessionIntegrityError, inspect_recorded_media_epoch +from k1link.web.camera_archive import CameraArchiveWriter + + +def _recorded_h264_fixture() -> tuple[bytes, bytes]: + def box(box_type: bytes, payload: bytes = b"") -> bytes: + return (8 + len(payload)).to_bytes(4, "big") + box_type + payload + + def full_box(box_type: bytes, payload: bytes = b"", *, flags: int = 0) -> bytes: + return box(box_type, bytes([0]) + flags.to_bytes(3, "big") + payload) + + track_id = 1 + sample_duration = 500 + tkhd = full_box( + b"tkhd", + b"\x00" * 8 + track_id.to_bytes(4, "big") + b"\x00" * 4, + ) + mdhd = full_box( + b"mdhd", + b"\x00" * 8 + (1_000).to_bytes(4, "big") + b"\x00" * 4, + ) + hdlr = full_box(b"hdlr", b"\x00" * 4 + b"vide") + trak = box(b"trak", tkhd + box(b"mdia", mdhd + hdlr)) + trex = full_box( + b"trex", + track_id.to_bytes(4, "big") + + (1).to_bytes(4, "big") + + sample_duration.to_bytes(4, "big") + + b"\x00" * 8, + ) + init = box(b"ftyp", b"isom") + box( + b"moov", + trak + box(b"mvex", trex) + box(b"avcC", b"\x01\x64\x00\x28"), + ) + tfhd = full_box(b"tfhd", track_id.to_bytes(4, "big"), flags=0x020000) + tfdt = full_box(b"tfdt", (0).to_bytes(4, "big")) + trun = full_box(b"trun", (1).to_bytes(4, "big")) + fragment = box(b"moof", box(b"traf", tfhd + tfdt + trun)) + box(b"mdat", b"frame") + return init, fragment + + +def _camera_epoch(tmp_path: Path) -> tuple[Path, bytes]: + session = tmp_path / "session-1" + session.mkdir() + init, fragment = _recorded_h264_fixture() + writer = CameraArchiveWriter(session, "sensor.camera.left", 1) + writer.append("init", init, host_epoch_ns=1_000_000_000, host_monotonic_ns=2_000_000_000) + writer.append( + "media", + fragment, + host_epoch_ns=1_500_000_000, + host_monotonic_ns=2_500_000_000, + ) + writer.close() + return session, fragment + + +def test_camera_compute_job_is_content_addressed_and_idempotent(tmp_path: Path) -> None: + session, fragment = _camera_epoch(tmp_path) + source_epoch = session / "media" / "sensor.camera.left" / "epoch-1" + source_identity = { + path.relative_to(source_epoch).as_posix(): hashlib.sha256(path.read_bytes()).hexdigest() + for path in source_epoch.rglob("*") + if path.is_file() + } + + job = prepare_camera_compute_job( + session_root=session, + source_id="sensor.camera.left", + codec_epoch=1, + origin_epoch_ns=1_000_000_000, + origin_monotonic_ns=2_000_000_000, + output_root=tmp_path / "jobs", + ) + repeated = prepare_camera_compute_job( + session_root=session, + source_id="sensor.camera.left", + codec_epoch=1, + origin_epoch_ns=1_000_000_000, + origin_monotonic_ns=2_000_000_000, + output_root=tmp_path / "jobs", + ) + + assert repeated == job + assert job.job_id == f"recorded-camera-{job.input_sha256[:24]}" + assert job.segment_count == 1 + assert job.timeline_start_seconds == 0.0 + assert job.timeline_end_seconds == 0.5 + copied = ( + job.job_root / "input" / "camera" / "sensor.camera.left" / "epoch-1" / "segments" / "1.m4s" + ) + assert copied.read_bytes() == fragment + assert validate_camera_compute_job(job.job_root) == job + assert source_identity == { + path.relative_to(source_epoch).as_posix(): hashlib.sha256(path.read_bytes()).hexdigest() + for path in source_epoch.rglob("*") + if path.is_file() + } + + +def test_camera_compute_job_rejects_changed_published_payload(tmp_path: Path) -> None: + session, _ = _camera_epoch(tmp_path) + job = prepare_camera_compute_job( + session_root=session, + source_id="sensor.camera.left", + codec_epoch=1, + origin_epoch_ns=1_000_000_000, + origin_monotonic_ns=2_000_000_000, + output_root=tmp_path / "jobs", + ) + segment = next(job.job_root.glob("input/camera/*/epoch-1/segments/1.m4s")) + segment.write_bytes(b"changed") + + with pytest.raises(SessionIntegrityError, match="payload identity changed"): + validate_camera_compute_job(job.job_root) + + +def test_epoch_inspection_checks_summary_aggregate_digests(tmp_path: Path) -> None: + session, _ = _camera_epoch(tmp_path) + epoch = session / "media" / "sensor.camera.left" / "epoch-1" + summary_path = epoch / "summary.json" + summary = json.loads(summary_path.read_text(encoding="utf-8")) + summary["index_sha256"] = "0" * 64 + summary_path.write_text(json.dumps(summary), encoding="utf-8") + + with pytest.raises(SessionIntegrityError, match="index digest changed"): + inspect_recorded_media_epoch( + epoch, + expected_source_name="sensor.camera.left", + origin_epoch_ns=1_000_000_000, + origin_monotonic_ns=2_000_000_000, + )