diff --git a/apps/control-station/src/productModel.ts b/apps/control-station/src/productModel.ts index 3b6575e..554a759 100644 --- a/apps/control-station/src/productModel.ts +++ b/apps/control-station/src/productModel.ts @@ -171,7 +171,7 @@ export const workspaces: WorkspaceDefinition[] = [ id: "observatory", root: "polygon", label: "Обсерватория", - title: "Обсерватория восприятия", + title: "Проверка компьютерного зрения", eyebrow: "ТЕСТОВЫЙ КОНТУР / ОБСЕРВАТОРИЯ", description: "Сессии и квалификация компьютерного зрения без доступа к управлению.", icon: "eye", diff --git a/apps/control-station/src/styles/observatory.css b/apps/control-station/src/styles/observatory.css index ef1156d..2aac577 100644 --- a/apps/control-station/src/styles/observatory.css +++ b/apps/control-station/src/styles/observatory.css @@ -1,5 +1,7 @@ .observatory-workspace { display: grid; + grid-auto-rows: max-content; + align-content: start; gap: 1rem; min-width: 0; min-height: 100%; @@ -37,6 +39,14 @@ margin: 0.3rem 0 0; } +.observatory-catalog-bar h3, +.observatory-session-summary h3, +.observatory-evidence h3 { + font-size: var(--nodedc-font-size-lg); + letter-spacing: -0.02em; + line-height: 1.15; +} + .observatory-lead p, .observatory-catalog-bar p, .observatory-state p, @@ -44,7 +54,14 @@ max-width: 52rem; margin: 0.35rem 0 0; color: var(--nodedc-text-muted); - line-height: 1.5; + font-size: var(--nodedc-font-size-sm); + line-height: 1.35; +} + +.observatory-catalog-bar, +.observatory-session-summary { + align-self: start; + min-height: 0; } .observatory-catalog-bar__copy { @@ -77,6 +94,8 @@ .observatory-session-stack { display: grid; + grid-auto-rows: max-content; + align-content: start; gap: 0.9rem; container-name: observatory-session; container-type: inline-size; @@ -130,6 +149,8 @@ margin: 0; overflow-wrap: anywhere; color: var(--nodedc-text-secondary); + font-size: var(--nodedc-font-size-sm); + line-height: 1.35; } .observatory-session-summary__end { @@ -168,7 +189,8 @@ max-width: 68rem; margin: 0.3rem 0 0; color: var(--nodedc-text-muted); - line-height: 1.45; + font-size: var(--nodedc-font-size-sm); + line-height: 1.35; } .observatory-setup-detail__badges, @@ -212,6 +234,8 @@ margin: 0; overflow-wrap: anywhere; color: var(--nodedc-text-secondary); + font-size: var(--nodedc-font-size-sm); + line-height: 1.35; } .observatory-setup-results { @@ -250,6 +274,8 @@ .observatory-setup-results small { color: var(--nodedc-text-muted); + font-size: var(--nodedc-font-size-sm); + line-height: 1.35; } .observatory-setup-detail__footer { diff --git a/apps/control-station/src/workspaces/observatory/ObservatoryWorkspace.tsx b/apps/control-station/src/workspaces/observatory/ObservatoryWorkspace.tsx index 35b6f17..35cc30d 100644 --- a/apps/control-station/src/workspaces/observatory/ObservatoryWorkspace.tsx +++ b/apps/control-station/src/workspaces/observatory/ObservatoryWorkspace.tsx @@ -145,6 +145,7 @@ export function ObservatoryWorkspace({ const [mutationReconciliation, setMutationReconciliation] = useState< ObservatoryMutationReconciliation | null >(null); + const overviewRef = useRef(null); const replayCoordinatorRef = useRef(createObservationReplayCoordinator()); const items = controller.catalog?.items ?? EMPTY_OBSERVATORY_ITEMS; @@ -153,6 +154,11 @@ export function ObservatoryWorkspace({ setReplay((current) => current.kind === "closed" ? current : { kind: "closed" }); }, []); + const returnToOverview = useCallback(() => { + closeReplay(); + overviewRef.current?.scrollIntoView({ block: "start" }); + }, [closeReplay]); + useEffect(() => { if (items.some((item) => item.source.id === selectedSessionId)) return; closeReplay(); @@ -352,7 +358,7 @@ export function ObservatoryWorkspace({ data-observatory-authority="observation-only" data-observatory-viewer={replay.kind === "ready" ? "attached" : "detached"} > -
+
{definition.eyebrow}

{definition.title}

@@ -368,7 +374,7 @@ export function ObservatoryWorkspace({
- +
ИСТОЧНИК ДОКАЗАТЕЛЬСТВ

Сохранённая сессия

@@ -606,7 +612,7 @@ export function ObservatoryWorkspace({

Проверяем точную связь результата с исходной сессией

Визуализатор и данные маршрута ещё не запущены.

- + ) : replay.kind === "error" ? ( @@ -623,7 +629,7 @@ export function ObservatoryWorkspace({ > Повторить - + ) : replay.kind === "ready" ? ( @@ -638,7 +644,7 @@ export function ObservatoryWorkspace({ size="compact" variant="secondary" icon={} - onClick={closeReplay} + onClick={returnToOverview} > Закрыть разбор diff --git a/apps/control-station/test/observatoryWorkspace.test.mjs b/apps/control-station/test/observatoryWorkspace.test.mjs index 010cb6f..2c485b3 100644 --- a/apps/control-station/test/observatoryWorkspace.test.mjs +++ b/apps/control-station/test/observatoryWorkspace.test.mjs @@ -35,7 +35,7 @@ test("Observatory is the third independent Polygon workspace", () => { id: "observatory", root: "polygon", label: "Обсерватория", - title: "Обсерватория восприятия", + title: "Проверка компьютерного зрения", eyebrow: "ТЕСТОВЫЙ КОНТУР / ОБСЕРВАТОРИЯ", description: "Сессии и квалификация компьютерного зрения без доступа к управлению.", icon: "eye", @@ -95,6 +95,10 @@ test("Observatory mounts the one shared canonical replay only after explicit adm assert.match(workspace, /role="alert"/); assert.match(workspace, /Повторить/); assert.match(workspace, /Закрыть разбор/); + assert.match( + workspace, + /const returnToOverview[\s\S]*closeReplay\(\);[\s\S]*scrollIntoView\(\{ block: "start" \}\)/, + ); assert.match(workspace, /

\{replayEvidence\?\.label \?\? replay\.binding\.resultId\}<\/h3>/); assert.doesNotMatch(workspace, /RAVNOVES004TREE · полный маршрут восприятия/); assert.match(workspace, /const selectSession[\s\S]*closeReplay\(\);[\s\S]*setSelectedSessionId/); @@ -130,6 +134,14 @@ test("Observatory mounts the one shared canonical replay only after explicit adm styles, /\.observatory-session-stack \{[\s\S]*container-name: observatory-session;[\s\S]*container-type: inline-size;/, ); + assert.match( + styles, + /\.observatory-workspace \{[\s\S]*grid-auto-rows: max-content;[\s\S]*align-content: start;/, + ); + assert.match( + styles, + /\.observatory-session-summary__facts dd \{[\s\S]*font-size: var\(--nodedc-font-size-sm\);/, + ); assert.match( styles, /@container observatory-session \(max-width: 56rem\) \{[\s\S]*\.observatory-session-summary \{[\s\S]*grid-template-columns: minmax\(0, 1fr\);/, diff --git a/docs/15_LABORATORY_RUN_CANON.md b/docs/15_LABORATORY_RUN_CANON.md index f2d274e..c0abade 100644 --- a/docs/15_LABORATORY_RUN_CANON.md +++ b/docs/15_LABORATORY_RUN_CANON.md @@ -275,6 +275,34 @@ execution canaries. Once they prove one automatic run-to-evidence path, the LAB canonicalization slice closes and remaining historical work stays `legacy` unless an integrity or product need justifies a targeted migration. +## Observatory preparation and Worker dispatch + +Selecting a source and a laboratory setup in Observatory is not itself a run. +Before a remote executor is installed, Mission Core may persist only an immutable +`missioncore.observatory-run-preparation/v1` receipt. That receipt binds the exact +source-catalog snapshot, setup id, RunDefinition version and configuration digest +to one idempotency key. Its only admitted terminal state is `blocked`; it has no +`run_id`, dispatch receipt or execution authority. + +A real Worker run begins only after a separate durable dispatcher has atomically +published an immutable dispatch receipt. The executor must be an additive, +allowlisted service owned outside the K1 acquisition path. It may accept bounded +versioned data, but never an arbitrary command, filesystem path or PowerShell +fragment from the UI. Exact retries return the original receipt; a reused key with +different identity fails closed. Transport uncertainty produces an explicit +reconciliation state and never an automatic duplicate run. + +The first executable Observatory setup remains the exact recorded +RAVNOVES00/M4.9T5 definition. Its full and integrated successors are different +definitions and results. The current RAVNOVES004TREE LAB V1 projection remains +replay-only until a new, independently versioned RunDefinition is reconstructed; +Mission Core does not invent a configuration for an already published result. + +Worker telemetry is secondary observation evidence. It does not replace the +authoritative run ledger, dispatch receipt, result validation or common laboratory +receipt. K1 control, Simulation/Gaussian runtimes and legacy LAB projections are +outside this executor boundary and are not restarted or migrated by it. + ## Planning discipline A planned LAB number is not a placeholder page. Before execution it must name: diff --git a/scripts/build_m49_tgs_full_shadow_worker_artifact.py b/scripts/build_m49_tgs_full_shadow_worker_artifact.py index df9dfaa..3d1102c 100644 --- a/scripts/build_m49_tgs_full_shadow_worker_artifact.py +++ b/scripts/build_m49_tgs_full_shadow_worker_artifact.py @@ -43,9 +43,78 @@ def sha256_file(path: Path) -> str: def git_revision() -> str: result = subprocess.run( - ["git", "rev-parse", "HEAD"], cwd=REPOSITORY_ROOT, check=True, capture_output=True, text=True + ["git", "rev-parse", "HEAD"], + cwd=REPOSITORY_ROOT, + check=True, + capture_output=True, + text=True, ) - return result.stdout.strip() + revision = result.stdout.strip() + if re.fullmatch(r"[a-f0-9]{40}", revision) is None: + raise ArtifactBuildError("Git revision is not a full SHA-1") + return revision + + +def materialize_revision(revision: str, destination: Path) -> None: + """Materialize only the declared release inputs from one exact Git commit.""" + + if re.fullmatch(r"[a-f0-9]{40}", revision) is None: + raise ArtifactBuildError("artifact revision is invalid") + if destination.exists(): + raise ArtifactBuildError("revision destination already exists") + + verified = subprocess.run( + ["git", "rev-parse", "--verify", f"{revision}^{{commit}}"], + cwd=REPOSITORY_ROOT, + check=False, + capture_output=True, + text=True, + ) + if verified.returncode != 0 or verified.stdout.strip() != revision: + raise ArtifactBuildError("artifact revision does not identify an existing commit") + + archive_path = destination.parent / "source.tar" + archived = subprocess.run( + [ + "git", + "archive", + "--format=tar", + "--output", + str(archive_path), + revision, + "--", + *(path.as_posix() for path in SOURCES), + ], + cwd=REPOSITORY_ROOT, + check=False, + capture_output=True, + text=True, + ) + if archived.returncode != 0: + raise ArtifactBuildError("declared release inputs are not present in artifact revision") + + destination.mkdir() + root = destination.resolve() + with tarfile.open(archive_path, "r:") as archive: + members = archive.getmembers() + for member in members: + target = (destination / member.name).resolve() + if target != root and root not in target.parents: + raise ArtifactBuildError("Git archive contains an unsafe path") + if not member.isdir() and not member.isreg(): + raise ArtifactBuildError("Git archive contains a non-regular release input") + for member in members: + target = destination / member.name + if member.isdir(): + target.mkdir(parents=True, exist_ok=True) + continue + stream = archive.extractfile(member) + if stream is None: + raise ArtifactBuildError("Git archive release input cannot be read") + target.parent.mkdir(parents=True, exist_ok=True) + with stream, target.open("wb") as output: + while chunk := stream.read(1024 * 1024): + output.write(chunk) def tar_info(path: Path, arcname: str) -> tarfile.TarInfo: @@ -81,24 +150,34 @@ def write_archive(stage: Path, target: Path) -> None: archive.addfile(info, io.BytesIO()) -def build(patch_id: str, output_directory: Path, *, revision: str | None = None) -> dict[str, object]: +def build( + patch_id: str, + output_directory: Path, + *, + revision: str | None = None, +) -> dict[str, object]: if PATCH_ID.fullmatch(patch_id) is None: raise ArtifactBuildError("patch id is invalid") - sources = tuple(REPOSITORY_ROOT / source for source in SOURCES) - if any(path.is_symlink() or not path.is_file() for path in sources): - raise ArtifactBuildError("release input is not a regular file") selected_revision = revision or git_revision() if re.fullmatch(r"[a-f0-9]{40}", selected_revision) is None: raise ArtifactBuildError("artifact revision is invalid") with tempfile.TemporaryDirectory(prefix="mission-core-m49-tgs-full-") as directory: stage = Path(directory) + snapshot = stage / "source" + materialize_revision(selected_revision, snapshot) + sources = tuple(snapshot / source for source in SOURCES) + if any(path.is_symlink() or not path.is_file() for path in sources): + raise ArtifactBuildError("release input is not a regular file") payload = stage / "payload" payload.mkdir() files: dict[str, dict[str, object]] = {} for source in sources: destination = payload / source.name destination.write_bytes(source.read_bytes()) - files[destination.name] = {"bytes": destination.stat().st_size, "sha256": sha256_file(destination)} + files[destination.name] = { + "bytes": destination.stat().st_size, + "sha256": sha256_file(destination), + } release = { "schema_version": "missioncore.m49-tgs-full-shadow-worker-release/v1", "patch_id": patch_id, @@ -106,7 +185,9 @@ def build(patch_id: str, output_directory: Path, *, revision: str | None = None) "worker_id": "worker-006", "candidate_id": "travel-tgs-full-shadow", "license": "GPL-3.0-or-later", - "source_pack_sha256": "0685d24219d8236caf8b7f1685e93f6d6b59e7fd015a768d88a92bbe8b154944", + "source_pack_sha256": ( + "0685d24219d8236caf8b7f1685e93f6d6b59e7fd015a768d88a92bbe8b154944" + ), "images": { "travel": "sha256:7b412020f4d8392d1d1ed1b33beadc44140f0ea8f781e62dd69796042334300f", "parity": "sha256:ceb13548617e4bd3f619766bfdff00af3fa5160946b367828da6d2233dcdcba0", @@ -121,10 +202,14 @@ def build(patch_id: str, output_directory: Path, *, revision: str | None = None) "files": files, } release_path = payload / "release.json" - release_path.write_text(json.dumps(release, indent=2, sort_keys=True) + "\n", encoding="utf-8") + release_path.write_text( + json.dumps(release, indent=2, sort_keys=True) + "\n", + encoding="utf-8", + ) payload_names = sorted((*files, release_path.name)) (stage / "manifest.env").write_text( - f"id={patch_id}\ncomponent=mission-core-worker\ntype=qualification-release\n", encoding="utf-8" + f"id={patch_id}\ncomponent=mission-core-worker\ntype=qualification-release\n", + encoding="utf-8", ) (stage / "files.txt").write_text("\n".join(payload_names) + "\n", encoding="utf-8") target = output_directory.resolve() / f"nodedc-{patch_id}.tgz" @@ -142,10 +227,15 @@ def build(patch_id: str, output_directory: Path, *, revision: str | None = None) def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("patch_id") - parser.add_argument("--output-directory", type=Path, default=REPOSITORY_ROOT / ".runtime/worker-artifacts") + parser.add_argument( + "--output-directory", + type=Path, + default=REPOSITORY_ROOT / ".runtime/worker-artifacts", + ) + parser.add_argument("--revision") arguments = parser.parse_args() try: - result = build(arguments.patch_id, arguments.output_directory) + result = build(arguments.patch_id, arguments.output_directory, revision=arguments.revision) except (ArtifactBuildError, OSError, subprocess.SubprocessError) as exc: parser.error(str(exc)) print(json.dumps(result, indent=2, sort_keys=True)) diff --git a/src/k1link/observatory/__init__.py b/src/k1link/observatory/__init__.py index 4e76e29..965836e 100644 --- a/src/k1link/observatory/__init__.py +++ b/src/k1link/observatory/__init__.py @@ -3,6 +3,24 @@ from k1link.observatory.canonical_result import ( is_admitted_observatory_recorded_result, ) +from k1link.observatory.run_preparations import ( + MAX_RUN_PREPARATION_RECORDS, + MAX_RUN_PREPARATION_STORAGE_BYTES, + OBSERVATORY_RUN_PREPARATION_REQUEST_SCHEMA, + OBSERVATORY_RUN_PREPARATION_SCHEMA, + RUN_PREPARATION_DATABASE_NAME, + RUN_PREPARATION_SQLITE_LOCK_TIMEOUT_SECONDS, + ObservatoryRunPreparation, + ObservatoryRunPreparationCapacityError, + ObservatoryRunPreparationConflictError, + ObservatoryRunPreparationError, + ObservatoryRunPreparationIntegrityError, + ObservatoryRunPreparationIntent, + ObservatoryRunPreparationLedger, + ObservatoryRunPreparationNotFoundError, + load_observatory_run_preparation_ledger, + observatory_run_preparation_request_sha256, +) from k1link.observatory.setups import ( LABORATORY_SETUP_CATALOG_SCHEMA, LABORATORY_SETUP_REGISTRY_SCHEMA, @@ -13,7 +31,23 @@ from k1link.observatory.setups import ( __all__ = [ "LABORATORY_SETUP_CATALOG_SCHEMA", "LABORATORY_SETUP_REGISTRY_SCHEMA", + "MAX_RUN_PREPARATION_RECORDS", + "MAX_RUN_PREPARATION_STORAGE_BYTES", + "OBSERVATORY_RUN_PREPARATION_REQUEST_SCHEMA", + "OBSERVATORY_RUN_PREPARATION_SCHEMA", + "RUN_PREPARATION_DATABASE_NAME", + "RUN_PREPARATION_SQLITE_LOCK_TIMEOUT_SECONDS", "LaboratorySetupRegistry", "LaboratorySetupRegistryError", + "ObservatoryRunPreparation", + "ObservatoryRunPreparationCapacityError", + "ObservatoryRunPreparationConflictError", + "ObservatoryRunPreparationError", + "ObservatoryRunPreparationIntegrityError", + "ObservatoryRunPreparationIntent", + "ObservatoryRunPreparationLedger", + "ObservatoryRunPreparationNotFoundError", "is_admitted_observatory_recorded_result", + "load_observatory_run_preparation_ledger", + "observatory_run_preparation_request_sha256", ] diff --git a/src/k1link/observatory/run_preparations.py b/src/k1link/observatory/run_preparations.py new file mode 100644 index 0000000..4a1d6c0 --- /dev/null +++ b/src/k1link/observatory/run_preparations.py @@ -0,0 +1,712 @@ +"""Durable, observation-only preparation receipts for Observatory runs. + +This ledger deliberately does not model a run lifecycle. Mission Core does +not currently have a durable dispatcher for the archived Observatory +RunDefinition, so a preparation can only record the exact admitted input and +the fail-closed blocker. A real run ledger must be introduced together with +an atomic durable dispatch receipt; until then ``run_id`` remains absent. +""" + +from __future__ import annotations + +import hashlib +import json +import os +import re +import sqlite3 +import threading +from collections.abc import Callable, Iterator +from contextlib import contextmanager, suppress +from dataclasses import dataclass +from datetime import datetime, timedelta +from pathlib import Path +from typing import Final, Literal +from uuid import uuid4 + +from k1link.artifacts import utc_now_iso + +OBSERVATORY_RUN_PREPARATION_SCHEMA: Final = ( + "missioncore.observatory-run-preparation/v1" +) +OBSERVATORY_RUN_PREPARATION_REQUEST_SCHEMA: Final = ( + "missioncore.observatory-run-preparation-request/v1" +) +RUN_PREPARATION_DATABASE_NAME: Final = "observatory-run-preparations.sqlite3" +MAX_RUN_PREPARATION_RECORDS: Final = 10_000 +MAX_RUN_PREPARATION_STORAGE_BYTES: Final = 64 * 1024 * 1024 +RUN_PREPARATION_SQLITE_LOCK_TIMEOUT_SECONDS: Final = 0.1 +_SQLITE_BUSY_TIMEOUT_MILLISECONDS: Final = 100 + +_PREPARATION_ID = re.compile(r"^observatory-prep-[a-f0-9]{32}$") +_IDEMPOTENCY_KEY = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,159}$") +_SESSION_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$") +_IDENTIFIER = re.compile(r"^[a-z][a-z0-9-]{2,95}$") +_SHA256 = re.compile(r"^[a-f0-9]{64}$") +_STATE: Literal["blocked"] = "blocked" +_AUTHORITY: Final = { + "commands_enabled": False, + "actuation_allowed": False, + "navigation_or_safety_accepted": False, + "production_accepted": False, +} + +_TABLE_SQL = """ +CREATE TABLE observatory_run_preparations ( + preparation_id TEXT PRIMARY KEY, + idempotency_key TEXT NOT NULL UNIQUE, + request_sha256 TEXT NOT NULL, + receipt_sha256 TEXT NOT NULL, + source_session_id TEXT NOT NULL, + source_catalog_sha256 TEXT NOT NULL, + setup_id TEXT NOT NULL, + definition_id TEXT NOT NULL, + definition_version INTEGER NOT NULL CHECK (definition_version > 0), + definition_sha256 TEXT NOT NULL, + state TEXT NOT NULL CHECK (state = 'blocked'), + blocker_code TEXT NOT NULL, + blocker_message TEXT NOT NULL, + created_at_utc TEXT NOT NULL +); +""" +_SCHEMA_SQL = _TABLE_SQL.replace("CREATE TABLE ", "CREATE TABLE IF NOT EXISTS ", 1) + +_EXPECTED_COLUMNS: Final = ( + ("preparation_id", "TEXT", 0, 1), + ("idempotency_key", "TEXT", 1, 0), + ("request_sha256", "TEXT", 1, 0), + ("receipt_sha256", "TEXT", 1, 0), + ("source_session_id", "TEXT", 1, 0), + ("source_catalog_sha256", "TEXT", 1, 0), + ("setup_id", "TEXT", 1, 0), + ("definition_id", "TEXT", 1, 0), + ("definition_version", "INTEGER", 1, 0), + ("definition_sha256", "TEXT", 1, 0), + ("state", "TEXT", 1, 0), + ("blocker_code", "TEXT", 1, 0), + ("blocker_message", "TEXT", 1, 0), + ("created_at_utc", "TEXT", 1, 0), +) + + +class ObservatoryRunPreparationError(RuntimeError): + """Base error for the isolated Observatory preparation ledger.""" + + +class ObservatoryRunPreparationConflictError(ObservatoryRunPreparationError): + """An idempotency key is already bound to another preparation request.""" + + +class ObservatoryRunPreparationNotFoundError(ObservatoryRunPreparationError): + """A preparation receipt is not present.""" + + +class ObservatoryRunPreparationIntegrityError(ObservatoryRunPreparationError): + """The durable ledger violates its fail-closed immutable contract.""" + + +class ObservatoryRunPreparationCapacityError(ObservatoryRunPreparationError): + """The bounded preparation ledger cannot admit another receipt.""" + + +@dataclass(frozen=True, slots=True) +class ObservatoryRunPreparationIntent: + """Validated, path-free identity captured at the preparation boundary.""" + + idempotency_key: str + source_session_id: str + source_catalog_sha256: str + setup_id: str + definition_id: str + definition_version: int + definition_sha256: str + blocker_code: str + blocker_message: str + + def __post_init__(self) -> None: + _validate_idempotency_key(self.idempotency_key) + _validate_pattern(self.source_session_id, _SESSION_ID, "source session id") + _validate_digest(self.source_catalog_sha256, "source catalog sha256") + _validate_pattern(self.setup_id, _IDENTIFIER, "setup id") + _validate_pattern(self.definition_id, _IDENTIFIER, "definition id") + if ( + not isinstance(self.definition_version, int) + or isinstance(self.definition_version, bool) + or self.definition_version < 1 + ): + raise ValueError("definition version must be positive") + _validate_digest(self.definition_sha256, "definition sha256") + _validate_pattern(self.blocker_code, _IDENTIFIER, "blocker code") + _validate_text(self.blocker_message, "blocker message", max_length=1_000) + + @property + def request_sha256(self) -> str: + """Fingerprint only the exact client payload used for idempotency.""" + + return observatory_run_preparation_request_sha256( + idempotency_key=self.idempotency_key, + source_session_id=self.source_session_id, + setup_id=self.setup_id, + definition_sha256=self.definition_sha256, + ) + + +@dataclass(frozen=True, slots=True) +class ObservatoryRunPreparation: + preparation_id: str + idempotency_key: str + request_sha256: str + receipt_sha256: str + source_session_id: str + source_catalog_sha256: str + setup_id: str + definition_id: str + definition_version: int + definition_sha256: str + blocker_code: str + blocker_message: str + created_at_utc: str + state: Literal["blocked"] = _STATE + + @classmethod + def from_intent( + cls, + intent: ObservatoryRunPreparationIntent, + *, + preparation_id: str, + created_at_utc: str, + ) -> ObservatoryRunPreparation: + receipt_sha256 = _receipt_sha256( + preparation_id=preparation_id, + idempotency_key=intent.idempotency_key, + request_sha256=intent.request_sha256, + source_session_id=intent.source_session_id, + source_catalog_sha256=intent.source_catalog_sha256, + setup_id=intent.setup_id, + definition_id=intent.definition_id, + definition_version=intent.definition_version, + definition_sha256=intent.definition_sha256, + blocker_code=intent.blocker_code, + blocker_message=intent.blocker_message, + created_at_utc=created_at_utc, + ) + return cls( + preparation_id=preparation_id, + idempotency_key=intent.idempotency_key, + request_sha256=intent.request_sha256, + receipt_sha256=receipt_sha256, + source_session_id=intent.source_session_id, + source_catalog_sha256=intent.source_catalog_sha256, + setup_id=intent.setup_id, + definition_id=intent.definition_id, + definition_version=intent.definition_version, + definition_sha256=intent.definition_sha256, + blocker_code=intent.blocker_code, + blocker_message=intent.blocker_message, + created_at_utc=created_at_utc, + ) + + def __post_init__(self) -> None: + _validate_pattern(self.preparation_id, _PREPARATION_ID, "preparation id") + intent = ObservatoryRunPreparationIntent( + idempotency_key=self.idempotency_key, + source_session_id=self.source_session_id, + source_catalog_sha256=self.source_catalog_sha256, + setup_id=self.setup_id, + definition_id=self.definition_id, + definition_version=self.definition_version, + definition_sha256=self.definition_sha256, + blocker_code=self.blocker_code, + blocker_message=self.blocker_message, + ) + _validate_digest(self.request_sha256, "request sha256") + if self.request_sha256 != intent.request_sha256: + raise ObservatoryRunPreparationIntegrityError( + "stored preparation request fingerprint changed" + ) + _validate_digest(self.receipt_sha256, "receipt sha256") + if self.state != _STATE: + raise ObservatoryRunPreparationIntegrityError( + "preparation cannot claim a run lifecycle state" + ) + _validate_timestamp(self.created_at_utc) + if self.receipt_sha256 != _receipt_sha256( + preparation_id=self.preparation_id, + idempotency_key=self.idempotency_key, + request_sha256=self.request_sha256, + source_session_id=self.source_session_id, + source_catalog_sha256=self.source_catalog_sha256, + setup_id=self.setup_id, + definition_id=self.definition_id, + definition_version=self.definition_version, + definition_sha256=self.definition_sha256, + blocker_code=self.blocker_code, + blocker_message=self.blocker_message, + created_at_utc=self.created_at_utc, + ): + raise ObservatoryRunPreparationIntegrityError( + "stored preparation receipt fingerprint changed" + ) + + def as_dict(self) -> dict[str, object]: + return { + "schema_version": OBSERVATORY_RUN_PREPARATION_SCHEMA, + "preparation_id": self.preparation_id, + "run_id": None, + "idempotency_key": self.idempotency_key, + "request_sha256": self.request_sha256, + "receipt_sha256": self.receipt_sha256, + "source": { + "session_id": self.source_session_id, + "catalog_sha256": self.source_catalog_sha256, + }, + "setup": { + "setup_id": self.setup_id, + "definition": { + "definition_id": self.definition_id, + "version": self.definition_version, + "definition_sha256": self.definition_sha256, + }, + }, + "state": self.state, + "blocker": { + "reason_code": self.blocker_code, + "message": self.blocker_message, + }, + "preconditions_passed": True, + "submission_allowed": False, + "dispatch_receipt": None, + "created_at_utc": self.created_at_utc, + "authority": dict(_AUTHORITY), + } + + +class ObservatoryRunPreparationLedger: + """SQLite-backed immutable preparation receipts with durable idempotency.""" + + def __init__( + self, + data_dir: Path, + *, + clock: Callable[[], str] = utc_now_iso, + max_records: int = MAX_RUN_PREPARATION_RECORDS, + ) -> None: + if ( + not isinstance(max_records, int) + or isinstance(max_records, bool) + or not 1 <= max_records <= MAX_RUN_PREPARATION_RECORDS + ): + raise ValueError("preparation ledger record quota is invalid") + self.data_dir = data_dir.expanduser().resolve() + self.database_path = self.data_dir / RUN_PREPARATION_DATABASE_NAME + self._clock = clock + self._max_records = max_records + self._lock = threading.RLock() + self._initialize() + + def prepare( + self, + intent: ObservatoryRunPreparationIntent, + ) -> tuple[ObservatoryRunPreparation, bool]: + """Persist once, returning the original receipt for an exact retry.""" + + try: + with self._lock, self._connect() as connection: + connection.execute("BEGIN IMMEDIATE") + existing = connection.execute( + "SELECT * FROM observatory_run_preparations " + "WHERE idempotency_key = ?", + (intent.idempotency_key,), + ).fetchone() + if existing is not None: + record = _record_from_row(existing) + if record.request_sha256 != intent.request_sha256: + raise ObservatoryRunPreparationConflictError( + "idempotency key is bound to another preparation request" + ) + connection.commit() + return record, False + + if self._bounded_record_count(connection) >= self._max_records: + raise ObservatoryRunPreparationCapacityError( + "preparation ledger record quota is exhausted" + ) + record = ObservatoryRunPreparation.from_intent( + intent, + preparation_id=f"observatory-prep-{uuid4().hex}", + created_at_utc=self._clock(), + ) + connection.execute( + "INSERT INTO observatory_run_preparations " + "(preparation_id, idempotency_key, request_sha256, receipt_sha256, " + "source_session_id, source_catalog_sha256, setup_id, " + "definition_id, definition_version, definition_sha256, " + "state, blocker_code, blocker_message, created_at_utc) " + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", + ( + record.preparation_id, + record.idempotency_key, + record.request_sha256, + record.receipt_sha256, + record.source_session_id, + record.source_catalog_sha256, + record.setup_id, + record.definition_id, + record.definition_version, + record.definition_sha256, + record.state, + record.blocker_code, + record.blocker_message, + record.created_at_utc, + ), + ) + connection.commit() + return record, True + except ObservatoryRunPreparationError: + raise + except (OSError, sqlite3.Error) as exc: + raise ObservatoryRunPreparationIntegrityError( + "preparation ledger write failed" + ) from exc + + def get(self, preparation_id: str) -> ObservatoryRunPreparation: + _validate_pattern(preparation_id, _PREPARATION_ID, "preparation id") + try: + with self._lock, self._connect() as connection: + row = connection.execute( + "SELECT * FROM observatory_run_preparations " + "WHERE preparation_id = ?", + (preparation_id,), + ).fetchone() + except (OSError, sqlite3.Error) as exc: + raise ObservatoryRunPreparationIntegrityError( + "preparation ledger read failed" + ) from exc + if row is None: + raise ObservatoryRunPreparationNotFoundError(preparation_id) + return _record_from_row(row) + + def get_by_idempotency_key( + self, + idempotency_key: str, + ) -> ObservatoryRunPreparation | None: + """Return a sealed receipt without consulting mutable source/catalog state.""" + + _validate_idempotency_key(idempotency_key) + try: + with self._lock, self._connect() as connection: + row = connection.execute( + "SELECT * FROM observatory_run_preparations " + "WHERE idempotency_key = ?", + (idempotency_key,), + ).fetchone() + except (OSError, sqlite3.Error) as exc: + raise ObservatoryRunPreparationIntegrityError( + "preparation ledger read failed" + ) from exc + return None if row is None else _record_from_row(row) + + def _initialize(self) -> None: + try: + self.data_dir.mkdir(mode=0o700, parents=True, exist_ok=True) + self._validate_storage_paths() + with suppress(OSError): + self.data_dir.chmod(0o700) + with self._connect() as connection: + connection.executescript(_SCHEMA_SQL) + self._validate_schema(connection) + if self._bounded_record_count(connection) > self._max_records: + raise ObservatoryRunPreparationCapacityError( + "preparation ledger record quota is exceeded" + ) + connection.commit() + self._validate_storage_paths(require_database=True) + with suppress(OSError): + self.database_path.chmod(0o600) + _fsync_directory(self.data_dir) + except ObservatoryRunPreparationError: + raise + except (OSError, sqlite3.Error) as exc: + raise ObservatoryRunPreparationIntegrityError( + "preparation ledger initialization failed" + ) from exc + + def _validate_schema(self, connection: sqlite3.Connection) -> None: + columns = connection.execute( + "SELECT name, type, \"notnull\", pk FROM pragma_table_info(?) " + "ORDER BY cid LIMIT ?", + ("observatory_run_preparations", len(_EXPECTED_COLUMNS) + 1), + ).fetchall() + actual_columns = tuple( + (row["name"], row["type"], row["notnull"], row["pk"]) + for row in columns + ) + if actual_columns != _EXPECTED_COLUMNS: + raise ObservatoryRunPreparationIntegrityError( + "preparation ledger schema is incompatible" + ) + indexes = connection.execute( + "SELECT name, \"unique\", partial FROM pragma_index_list(?) " + "ORDER BY seq LIMIT 9", + ("observatory_run_preparations",), + ).fetchall() + if len(indexes) > 8: + raise ObservatoryRunPreparationIntegrityError( + "preparation ledger has an unsafe index set" + ) + if not _has_full_binary_unique_index( + connection, + indexes, + "preparation_id", + ): + raise ObservatoryRunPreparationIntegrityError( + "preparation ledger lost receipt identity uniqueness" + ) + if not _has_full_binary_unique_index( + connection, + indexes, + "idempotency_key", + ): + raise ObservatoryRunPreparationIntegrityError( + "preparation ledger lost idempotency uniqueness" + ) + table = connection.execute( + "SELECT sql FROM sqlite_master " + "WHERE type = 'table' AND name = 'observatory_run_preparations' " + "LIMIT 1" + ).fetchone() + if ( + table is None + or not isinstance(table["sql"], str) + or _normalized_sql(table["sql"]) != _normalized_sql(_TABLE_SQL) + ): + raise ObservatoryRunPreparationIntegrityError( + "preparation ledger table contract is incompatible" + ) + + def _bounded_record_count(self, connection: sqlite3.Connection) -> int: + rows = connection.execute( + "SELECT 1 FROM observatory_run_preparations LIMIT ?", + (self._max_records + 1,), + ).fetchall() + return len(rows) + + def _validate_storage_paths(self, *, require_database: bool = False) -> None: + paths = ( + self.database_path, + Path(f"{self.database_path}-wal"), + Path(f"{self.database_path}-shm"), + ) + total_bytes = 0 + for path in paths: + if path.is_symlink() or (path.exists() and not path.is_file()): + raise ObservatoryRunPreparationIntegrityError( + "preparation ledger path is unsafe" + ) + if path.exists(): + total_bytes += path.stat().st_size + if require_database and not self.database_path.is_file(): + raise ObservatoryRunPreparationIntegrityError( + "preparation ledger database is unavailable" + ) + if total_bytes > MAX_RUN_PREPARATION_STORAGE_BYTES: + raise ObservatoryRunPreparationCapacityError( + "preparation ledger storage quota is exceeded" + ) + + @contextmanager + def _connect(self) -> Iterator[sqlite3.Connection]: + self._validate_storage_paths() + connection = sqlite3.connect( + self.database_path, + timeout=RUN_PREPARATION_SQLITE_LOCK_TIMEOUT_SECONDS, + ) + connection.row_factory = sqlite3.Row + try: + connection.execute("PRAGMA foreign_keys = ON") + connection.execute("PRAGMA synchronous = FULL") + connection.execute( + f"PRAGMA busy_timeout = {_SQLITE_BUSY_TIMEOUT_MILLISECONDS}" + ) + connection.execute("PRAGMA journal_mode = WAL") + yield connection + finally: + connection.close() + + +def _record_from_row(row: sqlite3.Row) -> ObservatoryRunPreparation: + try: + return ObservatoryRunPreparation( + preparation_id=row["preparation_id"], + idempotency_key=row["idempotency_key"], + request_sha256=row["request_sha256"], + receipt_sha256=row["receipt_sha256"], + source_session_id=row["source_session_id"], + source_catalog_sha256=row["source_catalog_sha256"], + setup_id=row["setup_id"], + definition_id=row["definition_id"], + definition_version=row["definition_version"], + definition_sha256=row["definition_sha256"], + state=row["state"], + blocker_code=row["blocker_code"], + blocker_message=row["blocker_message"], + created_at_utc=row["created_at_utc"], + ) + except (IndexError, KeyError, TypeError, ValueError) as exc: + raise ObservatoryRunPreparationIntegrityError( + "stored preparation receipt is invalid" + ) from exc + + +def load_observatory_run_preparation_ledger( + data_dir: Path, +) -> tuple[ObservatoryRunPreparationLedger | None, str | None]: + """Open the optional ledger within a short, fixed startup lock budget.""" + + try: + return ObservatoryRunPreparationLedger(data_dir), None + except (ObservatoryRunPreparationError, OSError) as exc: + return None, str(exc) + + +def observatory_run_preparation_request_sha256( + *, + idempotency_key: str, + source_session_id: str, + setup_id: str, + definition_sha256: str, +) -> str: + """Hash the exact immutable client payload, without mutable resolution state.""" + + _validate_idempotency_key(idempotency_key) + _validate_pattern(source_session_id, _SESSION_ID, "source session id") + _validate_pattern(setup_id, _IDENTIFIER, "setup id") + _validate_digest(definition_sha256, "definition sha256") + document = { + "schema_version": OBSERVATORY_RUN_PREPARATION_REQUEST_SCHEMA, + "idempotency_key": idempotency_key, + "source_session_id": source_session_id, + "setup_id": setup_id, + "definition_sha256": definition_sha256, + } + return hashlib.sha256(_canonical_json(document)).hexdigest() + + +def _receipt_sha256( + *, + preparation_id: str, + idempotency_key: str, + request_sha256: str, + source_session_id: str, + source_catalog_sha256: str, + setup_id: str, + definition_id: str, + definition_version: int, + definition_sha256: str, + blocker_code: str, + blocker_message: str, + created_at_utc: str, +) -> str: + document = { + "schema_version": OBSERVATORY_RUN_PREPARATION_SCHEMA, + "preparation_id": preparation_id, + "idempotency_key": idempotency_key, + "request_sha256": request_sha256, + "source_session_id": source_session_id, + "source_catalog_sha256": source_catalog_sha256, + "setup_id": setup_id, + "definition_id": definition_id, + "definition_version": definition_version, + "definition_sha256": definition_sha256, + "state": _STATE, + "blocker_code": blocker_code, + "blocker_message": blocker_message, + "created_at_utc": created_at_utc, + } + return hashlib.sha256(_canonical_json(document)).hexdigest() + + +def _has_full_binary_unique_index( + connection: sqlite3.Connection, + indexes: list[sqlite3.Row], + column: str, +) -> bool: + for index in indexes: + if not bool(index["unique"]) or bool(index["partial"]): + continue + fields = connection.execute( + "SELECT name, coll, desc, key FROM pragma_index_xinfo(?) " + "ORDER BY seqno LIMIT 4", + (index["name"],), + ).fetchall() + if len(fields) > 3: + continue + key_fields = [field for field in fields if bool(field["key"])] + if ( + len(key_fields) == 1 + and key_fields[0]["name"] == column + and key_fields[0]["coll"] == "BINARY" + and not bool(key_fields[0]["desc"]) + ): + return True + return False + + +def _normalized_sql(value: str) -> str: + return " ".join(value.split()).removesuffix(";") + + +def _validate_idempotency_key(value: object) -> None: + _validate_pattern(value, _IDEMPOTENCY_KEY, "idempotency key") + + +def _validate_digest(value: object, label: str) -> None: + _validate_pattern(value, _SHA256, label) + + +def _validate_pattern(value: object, pattern: re.Pattern[str], label: str) -> None: + if not isinstance(value, str) or pattern.fullmatch(value) is None: + raise ValueError(f"{label} is invalid") + + +def _validate_text(value: object, label: str, *, max_length: int) -> None: + if ( + not isinstance(value, str) + or not value.strip() + or value != value.strip() + or len(value) > max_length + ): + raise ValueError(f"{label} is invalid") + + +def _validate_timestamp(value: object) -> None: + _validate_text(value, "created timestamp", max_length=64) + assert isinstance(value, str) + try: + parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) + except ValueError as exc: + raise ValueError("created timestamp is invalid") from exc + if ( + parsed.tzinfo is None + or parsed.utcoffset() != timedelta(0) + or not value.endswith("Z") + ): + raise ValueError("created timestamp must use UTC") + + +def _canonical_json(value: object) -> bytes: + return json.dumps( + value, + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + ).encode("utf-8") + + +def _fsync_directory(path: Path) -> None: + flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0) + flags |= getattr(os, "O_DIRECTORY", 0) + descriptor = os.open(path, flags) + try: + os.fsync(descriptor) + finally: + os.close(descriptor) diff --git a/src/k1link/web/app.py b/src/k1link/web/app.py index f657d7e..41f71f0 100644 --- a/src/k1link/web/app.py +++ b/src/k1link/web/app.py @@ -36,7 +36,12 @@ from k1link.laboratory.m48_raw_evidence import ( M48RawEvidenceError, M48RawEvidenceReader, ) -from k1link.observatory import LaboratorySetupRegistry, LaboratorySetupRegistryError +from k1link.observatory import ( + LaboratorySetupRegistry, + LaboratorySetupRegistryError, + ObservatoryRunPreparationLedger, + load_observatory_run_preparation_ledger, +) from k1link.sessions import ( MaterializedRecording, RecordedCameraFrameService, @@ -191,6 +196,8 @@ LABORATORY_RUNNER = LaboratoryRunner( LABORATORY_VALUE_REVIEW_REGISTRY = LaboratoryValueReviewRegistry.from_file( REPOSITORY_ROOT / "config" / "laboratory-value-review.json" ) +OBSERVATORY_LABORATORY_SETUP_REGISTRY: LaboratorySetupRegistry | None +OBSERVATORY_LABORATORY_SETUP_REGISTRY_ERROR: str | None try: OBSERVATORY_LABORATORY_SETUP_REGISTRY = LaboratorySetupRegistry.from_file( REPOSITORY_ROOT / "config" / "observatory-laboratory-setups.json", @@ -229,6 +236,12 @@ plugin_environment = load_installed_device_plugins(REPOSITORY_ROOT) plugin_catalog: DevicePluginCatalog = plugin_environment.catalog plugin_dispatcher: DevicePluginDispatcher = plugin_environment.dispatcher session_store = SessionStore(REPOSITORY_ROOT) +OBSERVATORY_RUN_PREPARATION_LEDGER: ObservatoryRunPreparationLedger | None +OBSERVATORY_RUN_PREPARATION_LEDGER_ERROR: str | None +( + OBSERVATORY_RUN_PREPARATION_LEDGER, + OBSERVATORY_RUN_PREPARATION_LEDGER_ERROR, +) = load_observatory_run_preparation_ledger(session_store.data_dir) simulation_project_store = SimulationProjectStore(session_store.data_dir) simulation_project_service = SimulationProjectService(simulation_project_store) session_artifact_gateway = configured_artifact_gateway(session_store.data_dir) @@ -710,6 +723,8 @@ app.include_router( session_store, setup_registry=OBSERVATORY_LABORATORY_SETUP_REGISTRY, setup_registry_error=OBSERVATORY_LABORATORY_SETUP_REGISTRY_ERROR, + run_preparation_ledger=OBSERVATORY_RUN_PREPARATION_LEDGER, + run_preparation_ledger_error=OBSERVATORY_RUN_PREPARATION_LEDGER_ERROR, ) ) app.include_router( diff --git a/src/k1link/web/observatory_api.py b/src/k1link/web/observatory_api.py index acbe7ef..9ef1ba6 100644 --- a/src/k1link/web/observatory_api.py +++ b/src/k1link/web/observatory_api.py @@ -3,13 +3,22 @@ from __future__ import annotations from typing import Any, Literal from fastapi import APIRouter, HTTPException, Query, Response +from fastapi import Path as ApiPath from pydantic import BaseModel, ConfigDict, Field from k1link.observatory import ( LaboratorySetupRegistry, + ObservatoryRunPreparationCapacityError, + ObservatoryRunPreparationConflictError, + ObservatoryRunPreparationIntegrityError, + ObservatoryRunPreparationIntent, + ObservatoryRunPreparationLedger, + ObservatoryRunPreparationNotFoundError, is_admitted_observatory_recorded_result, + observatory_run_preparation_request_sha256, ) from k1link.sessions import SessionIntegrityError, SessionNotFoundError, SessionStore +from k1link.sessions.models import SessionSummary OBSERVATORY_PROJECTION_SCHEMA: Literal[ "missioncore.observatory-lab-projection/v1" @@ -57,17 +66,41 @@ class ObservatoryRunPreflightRequest(_StrictApiModel): definition_sha256: str | None = Field(default=None, pattern=r"^[a-f0-9]{64}$") +class ObservatoryRunPreparationRequest(_StrictApiModel): + schema_version: Literal[ + "missioncore.observatory-run-preparation-request/v1" + ] + idempotency_key: str = Field( + min_length=1, + max_length=160, + pattern=r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,159}$", + ) + source_session_id: str = Field( + min_length=1, + max_length=128, + pattern=r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$", + ) + setup_id: str = Field( + min_length=3, + max_length=96, + pattern=r"^[a-z][a-z0-9-]{2,95}$", + ) + definition_sha256: str = Field(pattern=r"^[a-f0-9]{64}$") + + def build_observatory_router( store: SessionStore, *, setup_registry: LaboratorySetupRegistry | None = None, setup_registry_error: str | None = None, + run_preparation_ledger: ObservatoryRunPreparationLedger | None = None, + run_preparation_ledger_error: str | None = None, ) -> APIRouter: """Build bounded catalog-only mutations for typed Observatory projections.""" router = APIRouter(tags=["observatory"]) - def source_summary(session_id: str): + def source_summary(session_id: str) -> SessionSummary: try: summary = store.get_session(session_id).summary except SessionNotFoundError as exc: @@ -89,6 +122,30 @@ def build_observatory_router( ) return summary + def source_summary_with_catalog_snapshot( + session_id: str, + ) -> tuple[SessionSummary, str]: + try: + detail, snapshot_sha256 = store.get_session_with_catalog_snapshot(session_id) + except SessionNotFoundError as exc: + raise HTTPException(status_code=404, detail="Исходная сессия не найдена.") from exc + except ValueError as exc: + raise HTTPException( + status_code=422, + detail="Некорректный идентификатор исходной сессии.", + ) from exc + except SessionIntegrityError as exc: + raise HTTPException( + status_code=409, + detail="Каталог исходной сессии нарушил контракт целостности.", + ) from exc + if detail.summary.lab is not None: + raise HTTPException( + status_code=409, + detail="Для подготовки нужна исходная, а не лабораторная сессия.", + ) + return detail.summary, snapshot_sha256 + def available_observatory_results(source_session_id: str) -> frozenset[str]: if setup_registry is None: return frozenset() @@ -242,6 +299,215 @@ def build_observatory_router( detail="Каталог сетапов Обсерватории не прошёл проверку целостности.", ) + if setup_registry is not None and run_preparation_ledger is not None: + + @router.post("/api/v1/observatory/run-preparations") + def prepare_observatory_run( + request: ObservatoryRunPreparationRequest, + ) -> dict[str, object]: + try: + existing = run_preparation_ledger.get_by_idempotency_key( + request.idempotency_key + ) + request_sha256 = observatory_run_preparation_request_sha256( + idempotency_key=request.idempotency_key, + source_session_id=request.source_session_id, + setup_id=request.setup_id, + definition_sha256=request.definition_sha256, + ) + except ObservatoryRunPreparationCapacityError as exc: + raise HTTPException( + status_code=503, + detail="Квота журнала подготовки расчётов исчерпана.", + ) from exc + except (ObservatoryRunPreparationIntegrityError, ValueError) as exc: + raise HTTPException( + status_code=503, + detail="Журнал подготовки расчётов недоступен.", + ) from exc + if existing is not None: + if existing.request_sha256 != request_sha256: + raise HTTPException( + status_code=409, + detail="Ключ идемпотентности уже связан с другой подготовкой.", + ) + return existing.as_dict() + + source, source_catalog_sha256 = source_summary_with_catalog_snapshot( + request.source_session_id + ) + try: + setup_registry.setup(request.setup_id) + except KeyError as exc: + raise HTTPException( + status_code=404, + detail="Сетап лаборатории не найден.", + ) from exc + catalog = setup_registry.catalog( + source, + available_observatory_result_ids=available_observatory_results( + request.source_session_id + ), + ) + setups = catalog.get("setups") + if not isinstance(setups, list): + raise HTTPException( + status_code=503, + detail="Каталог сетапов Обсерватории нарушил контракт.", + ) + projected = next( + ( + item + for item in setups + if isinstance(item, dict) and item.get("setup_id") == request.setup_id + ), + None, + ) + if projected is None: + raise HTTPException( + status_code=503, + detail="Каталог сетапов Обсерватории нарушил контракт.", + ) + definition = projected.get("run_definition") + if not isinstance(definition, dict): + raise HTTPException( + status_code=409, + detail="Для сетапа нет запускаемой RunDefinition.", + ) + expected_digest = definition.get("definition_sha256") + if request.definition_sha256 != expected_digest: + raise HTTPException( + status_code=409, + detail="Идентичность RunDefinition изменилась; обновите каталог.", + ) + compatibility = projected.get("compatibility") + preflight = projected.get("preflight") + executor = projected.get("executor") + if ( + not isinstance(compatibility, dict) + or not isinstance(preflight, dict) + or not isinstance(executor, dict) + ): + raise HTTPException( + status_code=503, + detail="Каталог сетапов Обсерватории нарушил контракт.", + ) + if compatibility.get("compatible") is not True: + raise HTTPException( + status_code=409, + detail="Исходная сессия несовместима с выбранным сетапом.", + ) + if preflight.get("outcome") == "existing": + raise HTTPException( + status_code=409, + detail="Точный результат уже существует; новый расчёт не подготовлен.", + ) + if ( + preflight.get("submission_allowed") is not False + or executor.get("state") != "not-installed" + ): + raise HTTPException( + status_code=503, + detail="Dispatch-контракт Обсерватории не установлен.", + ) + definition_id = definition.get("definition_id") + definition_version = definition.get("version") + blocker_code = executor.get("reason_code") + blocker_message = executor.get("reason") + if ( + not isinstance(definition_id, str) + or not isinstance(definition_version, int) + or isinstance(definition_version, bool) + or not isinstance(expected_digest, str) + or not isinstance(blocker_code, str) + or not isinstance(blocker_message, str) + ): + raise HTTPException( + status_code=503, + detail="Каталог сетапов Обсерватории нарушил контракт.", + ) + try: + intent = ObservatoryRunPreparationIntent( + idempotency_key=request.idempotency_key, + source_session_id=request.source_session_id, + source_catalog_sha256=source_catalog_sha256, + setup_id=request.setup_id, + definition_id=definition_id, + definition_version=definition_version, + definition_sha256=expected_digest, + blocker_code=blocker_code, + blocker_message=blocker_message, + ) + preparation, _created = run_preparation_ledger.prepare(intent) + except ObservatoryRunPreparationConflictError as exc: + raise HTTPException( + status_code=409, + detail="Ключ идемпотентности уже связан с другой подготовкой.", + ) from exc + except ObservatoryRunPreparationCapacityError as exc: + raise HTTPException( + status_code=503, + detail="Квота журнала подготовки расчётов исчерпана.", + ) from exc + except (ObservatoryRunPreparationIntegrityError, ValueError) as exc: + raise HTTPException( + status_code=503, + detail="Журнал подготовки расчётов недоступен.", + ) from exc + return preparation.as_dict() + + @router.get("/api/v1/observatory/run-preparations/{preparation_id}") + def get_observatory_run_preparation( + preparation_id: str = ApiPath( + min_length=49, + max_length=49, + pattern=r"^observatory-prep-[a-f0-9]{32}$", + ), + ) -> dict[str, object]: + try: + return run_preparation_ledger.get(preparation_id).as_dict() + except ObservatoryRunPreparationNotFoundError as exc: + raise HTTPException( + status_code=404, + detail="Подготовка расчёта не найдена.", + ) from exc + except ObservatoryRunPreparationCapacityError as exc: + raise HTTPException( + status_code=503, + detail="Квота журнала подготовки расчётов исчерпана.", + ) from exc + except (ObservatoryRunPreparationIntegrityError, ValueError) as exc: + raise HTTPException( + status_code=503, + detail="Журнал подготовки расчётов недоступен.", + ) from exc + + elif setup_registry_error is not None or run_preparation_ledger_error is not None: + + @router.post("/api/v1/observatory/run-preparations") + def unavailable_observatory_run_preparation( + request: ObservatoryRunPreparationRequest, + ) -> None: + del request + raise HTTPException( + status_code=503, + detail="Подготовка расчётов Обсерватории недоступна.", + ) + + @router.get("/api/v1/observatory/run-preparations/{preparation_id}") + def unavailable_observatory_run_preparation_receipt( + preparation_id: str = ApiPath( + min_length=49, + max_length=49, + pattern=r"^observatory-prep-[a-f0-9]{32}$", + ), + ) -> None: + del preparation_id + raise HTTPException( + status_code=503, + detail="Подготовка расчётов Обсерватории недоступна.", + ) + @router.patch( "/api/v1/observatory/lab-projections/{session_id}", response_model=ObservatoryProjectionDocument, diff --git a/tests/test_m49_tgs_full_shadow.py b/tests/test_m49_tgs_full_shadow.py index 6a52303..fcf4a2c 100644 --- a/tests/test_m49_tgs_full_shadow.py +++ b/tests/test_m49_tgs_full_shadow.py @@ -3,10 +3,13 @@ from __future__ import annotations import hashlib import importlib.util import json +import subprocess import sys +import tarfile from pathlib import Path import numpy as np +import pytest from k1link.laboratory.m49_tgs_full_shadow import ( M49TgsFullShadowResult, @@ -34,6 +37,49 @@ ARTIFACT = importlib.util.module_from_spec(ARTIFACT_SPEC) ARTIFACT_SPEC.loader.exec_module(ARTIFACT) +def _create_artifact_git_repository(root: Path) -> tuple[str, dict[Path, bytes]]: + subprocess.run(["git", "init", "-q", str(root)], check=True, capture_output=True) + subprocess.run( + ["git", "config", "user.email", "mission-core-test@example.invalid"], + cwd=root, + check=True, + capture_output=True, + ) + subprocess.run( + ["git", "config", "user.name", "Mission Core Test"], + cwd=root, + check=True, + capture_output=True, + ) + committed: dict[Path, bytes] = {} + for index, relative in enumerate(ARTIFACT.SOURCES): + payload = f"committed source {index}: {relative.as_posix()}\n".encode() + path = root / relative + path.parent.mkdir(parents=True, exist_ok=True) + path.write_bytes(payload) + committed[relative] = payload + subprocess.run( + ["git", "add", "--", *ARTIFACT.SOURCES], + cwd=root, + check=True, + capture_output=True, + ) + subprocess.run( + ["git", "commit", "-q", "-m", "fixture"], + cwd=root, + check=True, + capture_output=True, + ) + revision = subprocess.run( + ["git", "rev-parse", "HEAD"], + cwd=root, + check=True, + capture_output=True, + text=True, + ).stdout.strip() + return revision, committed + + def test_full_shadow_exact_multiset_and_costmap_priority() -> None: native = np.asarray( [[2.0, 0.0, 0.0, 0.0], [2.0, 0.0, 0.0, 0.0], [3.0, 0.0, 1.0, 0.0], [4.0, 0.0, 2.0, 0.0]], @@ -47,11 +93,10 @@ def test_full_shadow_exact_multiset_and_costmap_priority() -> None: def test_full_shadow_worker_artifact_is_deterministic_and_cpu_only(tmp_path: Path) -> None: - revision = "a" * 40 + revision = ARTIFACT.git_revision() first = ARTIFACT.build("m49-tgs-full-test", tmp_path / "one", revision=revision) second = ARTIFACT.build("m49-tgs-full-test", tmp_path / "two", revision=revision) assert first["sha256"] == second["sha256"] - import tarfile with tarfile.open(first["artifact"], "r:gz") as archive: release = archive.extractfile("payload/release.json") @@ -64,6 +109,67 @@ def test_full_shadow_worker_artifact_is_deterministic_and_cpu_only(tmp_path: Pat assert "gpu_requested = $false" in runner_text +def test_full_shadow_explicit_revision_ignores_current_file_drift( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + repository = tmp_path / "repository" + revision, committed = _create_artifact_git_repository(repository) + drifted_source = ARTIFACT.SOURCES[0] + (repository / drifted_source).write_bytes(b"dirty current-worktree replacement\n") + monkeypatch.setattr(ARTIFACT, "REPOSITORY_ROOT", repository) + + explicit = ARTIFACT.build( + "m49-tgs-full-clean-revision", + tmp_path / "explicit", + revision=revision, + ) + default_head = ARTIFACT.build("m49-tgs-full-clean-revision", tmp_path / "default") + + assert explicit["code_revision"] == revision + assert default_head["code_revision"] == revision + assert explicit["sha256"] == default_head["sha256"] + with tarfile.open(explicit["artifact"], "r:gz") as archive: + source = archive.extractfile(f"payload/{drifted_source.name}") + release_stream = archive.extractfile("payload/release.json") + assert source is not None and release_stream is not None + assert source.read() == committed[drifted_source] + release = json.loads(release_stream.read()) + assert release["code_revision"] == revision + assert release["files"][drifted_source.name]["sha256"] == hashlib.sha256( + committed[drifted_source] + ).hexdigest() + + +@pytest.mark.parametrize("revision", ["HEAD", "../HEAD", "--all", "A" * 40]) +def test_full_shadow_artifact_rejects_unsafe_or_symbolic_revision( + revision: str, + tmp_path: Path, +) -> None: + with pytest.raises(ARTIFACT.ArtifactBuildError, match="artifact revision is invalid"): + ARTIFACT.build("m49-tgs-full-invalid-revision", tmp_path, revision=revision) + + +def test_full_shadow_artifact_rejects_missing_revision( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + repository = tmp_path / "repository" + revision, _ = _create_artifact_git_repository(repository) + missing_revision = "f" * 40 if revision != "f" * 40 else "e" * 40 + monkeypatch.setattr(ARTIFACT, "REPOSITORY_ROOT", repository) + + with pytest.raises( + ARTIFACT.ArtifactBuildError, + match="artifact revision does not identify an existing commit", + ): + ARTIFACT.build( + "m49-tgs-full-missing-revision", + tmp_path / "artifact", + revision=missing_revision, + ) + + def test_full_shadow_seal_binds_visual_and_semantic_timelines(tmp_path: Path) -> None: source = tmp_path / "worker" source.mkdir() diff --git a/tests/test_observatory_run_preparations.py b/tests/test_observatory_run_preparations.py new file mode 100644 index 0000000..00ff89e --- /dev/null +++ b/tests/test_observatory_run_preparations.py @@ -0,0 +1,615 @@ +from __future__ import annotations + +import sqlite3 +from concurrent.futures import ThreadPoolExecutor +from dataclasses import replace +from pathlib import Path +from threading import Barrier +from time import monotonic +from types import SimpleNamespace + +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient + +from k1link.observatory import ( + MAX_RUN_PREPARATION_RECORDS, + RUN_PREPARATION_DATABASE_NAME, + LaboratorySetupRegistry, + ObservatoryRunPreparationCapacityError, + ObservatoryRunPreparationConflictError, + ObservatoryRunPreparationIntegrityError, + ObservatoryRunPreparationIntent, + ObservatoryRunPreparationLedger, + load_observatory_run_preparation_ledger, +) +from k1link.observatory import run_preparations as run_preparations_module +from k1link.sessions import SessionNotFoundError +from k1link.sessions.models import SessionSummary +from k1link.web.observatory_api import build_observatory_router + +REPOSITORY_ROOT = Path(__file__).resolve().parents[1] +REGISTRY_PATH = REPOSITORY_ROOT / "config" / "observatory-laboratory-setups.json" +SOURCE_SESSION_ID = "20260720T065719Z_viewer_live" +SOURCE_LABEL = "RAVNOVES00" +CURRENT_SOURCE_SESSION_ID = "20260828T130511Z_viewer_live" +SNAPSHOT_A = "a" * 64 +SNAPSHOT_B = "b" * 64 + + +def _registry() -> LaboratorySetupRegistry: + return LaboratorySetupRegistry.from_file( + REGISTRY_PATH, + repository_root=REPOSITORY_ROOT, + ) + + +def _source( + session_id: str = SOURCE_SESSION_ID, + label: str = SOURCE_LABEL, +) -> SessionSummary: + return SessionSummary( + session_id=session_id, + display_name=label, + status="ready", + started_at_utc="2026-07-20T06:57:19Z", + completed_at_utc="2026-07-20T07:10:47Z", + duration_seconds=808.0, + modalities=("point-cloud", "trajectory", "video"), + source_count=3, + total_bytes=1, + replayable=True, + origin="recorded", + ) + + +def _definition() -> dict[str, object]: + setup = _registry().catalog(_source())["setups"][0] + assert isinstance(setup, dict) + definition = setup["run_definition"] + assert isinstance(definition, dict) + return definition + + +def _intent( + *, + idempotency_key: str = "operator-request-001", + source_catalog_sha256: str = SNAPSHOT_A, +) -> ObservatoryRunPreparationIntent: + definition = _definition() + return ObservatoryRunPreparationIntent( + idempotency_key=idempotency_key, + source_session_id=SOURCE_SESSION_ID, + source_catalog_sha256=source_catalog_sha256, + setup_id="m49-tgs-full-shadow-v1", + definition_id=str(definition["definition_id"]), + definition_version=int(definition["version"]), + definition_sha256=str(definition["definition_sha256"]), + blocker_code="laboratory-runner-adapter-not-installed", + blocker_message=( + "Повторный запуск этого сетапа ещё не подключён к общему контуру расчёта." + ), + ) + + +class _ReadOnlySourceStore: + def __init__(self, source: SessionSummary | None = None) -> None: + resolved = source or _source() + self.sessions = {resolved.session_id: resolved} + self.snapshot_sha256 = SNAPSHOT_A + self.snapshot_reads = 0 + + def get_session(self, session_id: str): + try: + summary = self.sessions[session_id] + except KeyError as exc: + raise SessionNotFoundError(session_id) from exc + return SimpleNamespace(summary=summary) + + def get_session_with_catalog_snapshot(self, session_id: str): + self.snapshot_reads += 1 + return self.get_session(session_id), self.snapshot_sha256 + + +def _client( + tmp_path: Path, + *, + store: _ReadOnlySourceStore | None = None, +) -> tuple[TestClient, ObservatoryRunPreparationLedger, _ReadOnlySourceStore]: + source_store = store or _ReadOnlySourceStore() + ledger = ObservatoryRunPreparationLedger( + tmp_path / "mission-core", + clock=lambda: "2026-08-30T20:00:00.000Z", + ) + app = FastAPI() + app.include_router( + build_observatory_router( + source_store, # type: ignore[arg-type] + setup_registry=_registry(), + run_preparation_ledger=ledger, + ) + ) + return TestClient(app), ledger, source_store + + +def _request( + *, + idempotency_key: str = "operator-request-001", + definition_sha256: str | None = None, +) -> dict[str, object]: + definition = _definition() + return { + "schema_version": "missioncore.observatory-run-preparation-request/v1", + "idempotency_key": idempotency_key, + "source_session_id": SOURCE_SESSION_ID, + "setup_id": "m49-tgs-full-shadow-v1", + "definition_sha256": definition_sha256 or definition["definition_sha256"], + } + + +def _create_custom_ledger_schema( + data_dir: Path, + *, + idempotency_declaration: str, + extra_sql: str = "", +) -> None: + data_dir.mkdir(exist_ok=True) + with sqlite3.connect(data_dir / RUN_PREPARATION_DATABASE_NAME) as connection: + connection.executescript( + f""" + CREATE TABLE observatory_run_preparations ( + preparation_id TEXT PRIMARY KEY, + idempotency_key {idempotency_declaration}, + request_sha256 TEXT NOT NULL, + receipt_sha256 TEXT NOT NULL, + source_session_id TEXT NOT NULL, + source_catalog_sha256 TEXT NOT NULL, + setup_id TEXT NOT NULL, + definition_id TEXT NOT NULL, + definition_version INTEGER NOT NULL CHECK (definition_version > 0), + definition_sha256 TEXT NOT NULL, + state TEXT NOT NULL CHECK (state = 'blocked'), + blocker_code TEXT NOT NULL, + blocker_message TEXT NOT NULL, + created_at_utc TEXT NOT NULL + ); + {extra_sql} + """ + ) + + +def test_preparation_ledger_is_durable_and_exactly_idempotent(tmp_path: Path) -> None: + data_dir = tmp_path / "mission-core" + ledger = ObservatoryRunPreparationLedger( + data_dir, + clock=lambda: "2026-08-30T20:00:00.000Z", + ) + + first, first_created = ledger.prepare(_intent()) + second, second_created = ledger.prepare(_intent()) + restored = ObservatoryRunPreparationLedger(data_dir).get(first.preparation_id) + restored_by_key = ObservatoryRunPreparationLedger(data_dir).get_by_idempotency_key( + first.idempotency_key + ) + + assert first_created is True + assert second_created is False + assert second == first + assert restored == first + assert restored_by_key == first + assert first.state == "blocked" + assert first.as_dict()["run_id"] is None + assert first.as_dict()["dispatch_receipt"] is None + assert first.as_dict()["submission_allowed"] is False + assert first.as_dict()["receipt_sha256"] == first.receipt_sha256 + with sqlite3.connect(data_dir / RUN_PREPARATION_DATABASE_NAME) as connection: + count = connection.execute( + "SELECT COUNT(*) FROM observatory_run_preparations" + ).fetchone()[0] + assert count == 1 + assert (data_dir / RUN_PREPARATION_DATABASE_NAME).stat().st_mode & 0o777 == 0o600 + + +def test_preparation_idempotency_key_replays_original_resolved_source_identity( + tmp_path: Path, +) -> None: + ledger = ObservatoryRunPreparationLedger(tmp_path) + first, _created = ledger.prepare(_intent()) + + replayed, replay_created = ledger.prepare( + _intent(source_catalog_sha256=SNAPSHOT_B) + ) + + assert replay_created is False + assert replayed == first + assert replayed.source_catalog_sha256 == SNAPSHOT_A + + with pytest.raises(ObservatoryRunPreparationConflictError, match="idempotency"): + ledger.prepare(replace(_intent(), definition_sha256="c" * 64)) + + +def test_preparation_ledger_detects_stored_identity_drift(tmp_path: Path) -> None: + ledger = ObservatoryRunPreparationLedger(tmp_path) + record, _created = ledger.prepare(_intent()) + with sqlite3.connect(ledger.database_path) as connection: + connection.execute( + "UPDATE observatory_run_preparations SET source_catalog_sha256 = ?", + (SNAPSHOT_B,), + ) + connection.commit() + + with pytest.raises( + ObservatoryRunPreparationIntegrityError, + match="receipt fingerprint", + ): + ledger.get(record.preparation_id) + + +def test_preparation_ledger_rejects_an_incompatible_existing_schema( + tmp_path: Path, +) -> None: + tmp_path.mkdir(exist_ok=True) + with sqlite3.connect(tmp_path / RUN_PREPARATION_DATABASE_NAME) as connection: + connection.execute( + "CREATE TABLE observatory_run_preparations (preparation_id TEXT)" + ) + + with pytest.raises( + ObservatoryRunPreparationIntegrityError, + match="initialization|schema", + ): + ObservatoryRunPreparationLedger(tmp_path) + + +def test_preparation_ledger_rejects_partial_idempotency_uniqueness( + tmp_path: Path, +) -> None: + _create_custom_ledger_schema( + tmp_path, + idempotency_declaration="TEXT NOT NULL", + extra_sql=( + "CREATE UNIQUE INDEX partial_idempotency " + "ON observatory_run_preparations(idempotency_key) " + "WHERE state = 'blocked';" + ), + ) + + with pytest.raises(ObservatoryRunPreparationIntegrityError, match="uniqueness"): + ObservatoryRunPreparationLedger(tmp_path) + + +def test_preparation_ledger_rejects_non_binary_idempotency_collation( + tmp_path: Path, +) -> None: + _create_custom_ledger_schema( + tmp_path, + idempotency_declaration="TEXT COLLATE NOCASE NOT NULL UNIQUE", + ) + + with pytest.raises(ObservatoryRunPreparationIntegrityError, match="uniqueness"): + ObservatoryRunPreparationLedger(tmp_path) + + +def test_preparation_ledger_enforces_record_quota_but_keeps_retries_available( + tmp_path: Path, +) -> None: + ledger = ObservatoryRunPreparationLedger(tmp_path, max_records=1) + first, _created = ledger.prepare(_intent()) + + replayed, replay_created = ledger.prepare(_intent()) + assert replay_created is False + assert replayed == first + with pytest.raises(ObservatoryRunPreparationCapacityError, match="quota"): + ledger.prepare(_intent(idempotency_key="operator-request-002")) + + +def test_preparation_ledger_concurrent_retry_commits_one_receipt( + tmp_path: Path, +) -> None: + first_ledger = ObservatoryRunPreparationLedger(tmp_path) + second_ledger = ObservatoryRunPreparationLedger(tmp_path) + barrier = Barrier(2) + + def prepare( + ledger: ObservatoryRunPreparationLedger, + snapshot_sha256: str, + ) -> tuple[str, str, bool]: + barrier.wait() + record, created = ledger.prepare( + _intent(source_catalog_sha256=snapshot_sha256) + ) + return record.preparation_id, record.source_catalog_sha256, created + + with ThreadPoolExecutor(max_workers=2) as executor: + first_future = executor.submit(prepare, first_ledger, SNAPSHOT_A) + second_future = executor.submit(prepare, second_ledger, SNAPSHOT_B) + first = first_future.result(timeout=10) + second = second_future.result(timeout=10) + + assert first[0] == second[0] + assert first[1] == second[1] + assert sorted((first[2], second[2])) == [False, True] + with sqlite3.connect(first_ledger.database_path) as connection: + assert connection.execute( + "SELECT COUNT(*) FROM observatory_run_preparations" + ).fetchone()[0] == 1 + + +def test_preparation_ledger_startup_uses_only_bounded_admission_queries( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + statements: list[str] = [] + connect = sqlite3.connect + + def traced_connect(*args: object, **kwargs: object) -> sqlite3.Connection: + connection = connect(*args, **kwargs) + connection.set_trace_callback(statements.append) + return connection + + monkeypatch.setattr(run_preparations_module.sqlite3, "connect", traced_connect) + + ObservatoryRunPreparationLedger(tmp_path) + + normalized = [statement.lower() for statement in statements] + assert not any("quick_check" in statement for statement in normalized) + quota_queries = [ + statement + for statement in normalized + if "select 1 from observatory_run_preparations limit" in statement + ] + assert len(quota_queries) == 1 + assert str(MAX_RUN_PREPARATION_RECORDS + 1) in quota_queries[0] + + +def test_optional_ledger_loader_isolates_startup_from_ledger_corruption( + tmp_path: Path, +) -> None: + _create_custom_ledger_schema( + tmp_path, + idempotency_declaration="TEXT NOT NULL", + ) + + ledger, error = load_observatory_run_preparation_ledger(tmp_path) + + assert ledger is None + assert error is not None + + +def test_optional_ledger_loader_has_a_short_database_lock_budget( + tmp_path: Path, +) -> None: + ledger = ObservatoryRunPreparationLedger(tmp_path) + locked = sqlite3.connect(ledger.database_path, timeout=0) + locked.execute("PRAGMA journal_mode = DELETE") + locked.execute("BEGIN EXCLUSIVE") + try: + started = monotonic() + loaded, error = load_observatory_run_preparation_ledger(tmp_path) + elapsed = monotonic() - started + finally: + locked.rollback() + locked.close() + + assert loaded is None + assert error is not None + assert elapsed < 0.5 + + +def test_preparation_endpoint_records_only_a_blocked_non_run_receipt( + tmp_path: Path, +) -> None: + client, ledger, source_store = _client(tmp_path) + + response = client.post( + "/api/v1/observatory/run-preparations", + json=_request(), + ) + + assert response.status_code == 200 + document = response.json() + assert document == ledger.get(document["preparation_id"]).as_dict() + assert document["schema_version"] == "missioncore.observatory-run-preparation/v1" + assert document["run_id"] is None + assert document["state"] == "blocked" + assert document["preconditions_passed"] is True + assert document["submission_allowed"] is False + assert document["dispatch_receipt"] is None + assert document["source"] == { + "session_id": SOURCE_SESSION_ID, + "catalog_sha256": SNAPSHOT_A, + } + assert document["setup"]["definition"]["definition_sha256"] == ( + _definition()["definition_sha256"] + ) + assert document["blocker"]["reason_code"] == ( + "laboratory-runner-adapter-not-installed" + ) + assert document["authority"] == { + "commands_enabled": False, + "actuation_allowed": False, + "navigation_or_safety_accepted": False, + "production_accepted": False, + } + assert source_store.snapshot_reads == 1 + assert client.post("/api/v1/observatory/runs", json={}).status_code == 404 + + +def test_preparation_endpoint_duplicate_returns_the_same_durable_receipt( + tmp_path: Path, +) -> None: + client, _ledger, _source_store = _client(tmp_path) + + first = client.post("/api/v1/observatory/run-preparations", json=_request()) + second = client.post("/api/v1/observatory/run-preparations", json=_request()) + restored = client.get( + f"/api/v1/observatory/run-preparations/{first.json()['preparation_id']}" + ) + + assert first.status_code == second.status_code == restored.status_code == 200 + assert second.json() == first.json() + assert restored.json() == first.json() + + +def test_preparation_endpoint_exact_retry_precedes_mutable_source_and_catalog_checks( + tmp_path: Path, +) -> None: + client, _ledger, source_store = _client(tmp_path) + first = client.post( + "/api/v1/observatory/run-preparations", + json=_request(), + ) + assert first.status_code == 200 + + source_store.snapshot_sha256 = SNAPSHOT_B + source_store.sessions.clear() + replayed = client.post( + "/api/v1/observatory/run-preparations", + json=_request(), + ) + + assert replayed.status_code == 200 + assert replayed.json() == first.json() + assert replayed.json()["source"]["catalog_sha256"] == SNAPSHOT_A + assert source_store.snapshot_reads == 1 + + +def test_preparation_endpoint_new_key_captures_a_new_source_snapshot( + tmp_path: Path, +) -> None: + client, _ledger, source_store = _client(tmp_path) + first = client.post( + "/api/v1/observatory/run-preparations", + json=_request(), + ) + assert first.status_code == 200 + + source_store.snapshot_sha256 = SNAPSHOT_B + second = client.post( + "/api/v1/observatory/run-preparations", + json=_request(idempotency_key="operator-request-002"), + ) + + assert second.status_code == 200 + assert second.json()["preparation_id"] != first.json()["preparation_id"] + assert second.json()["source"]["catalog_sha256"] == SNAPSHOT_B + + +def test_preparation_endpoint_same_key_rejects_a_changed_client_payload_early( + tmp_path: Path, +) -> None: + client, _ledger, source_store = _client(tmp_path) + assert client.post( + "/api/v1/observatory/run-preparations", + json=_request(), + ).status_code == 200 + source_store.sessions.clear() + + conflict = client.post( + "/api/v1/observatory/run-preparations", + json=_request(definition_sha256="f" * 64), + ) + + assert conflict.status_code == 409 + assert source_store.snapshot_reads == 1 + + +def test_preparation_endpoint_maps_early_ledger_capacity_failure_to_503( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + client, _ledger, source_store = _client(tmp_path) + monkeypatch.setattr( + run_preparations_module, + "MAX_RUN_PREPARATION_STORAGE_BYTES", + 0, + ) + + response = client.post( + "/api/v1/observatory/run-preparations", + json=_request(), + ) + + assert response.status_code == 503 + assert source_store.snapshot_reads == 0 + + +def test_preparation_endpoint_rejects_stale_incompatible_and_non_executable_inputs( + tmp_path: Path, +) -> None: + client, _ledger, source_store = _client(tmp_path) + + stale = client.post( + "/api/v1/observatory/run-preparations", + json=_request(definition_sha256="f" * 64), + ) + assert stale.status_code == 409 + + source_store.sessions[SOURCE_SESSION_ID] = replace( + _source(), + replayable=False, + ) + incompatible = client.post( + "/api/v1/observatory/run-preparations", + json=_request(idempotency_key="operator-request-002"), + ) + assert incompatible.status_code == 409 + + current_store = _ReadOnlySourceStore( + _source(CURRENT_SOURCE_SESSION_ID, "RAVNOVES004TREE") + ) + current_client, _current_ledger, _ = _client(tmp_path / "current", store=current_store) + no_definition = current_client.post( + "/api/v1/observatory/run-preparations", + json={ + "schema_version": "missioncore.observatory-run-preparation-request/v1", + "idempotency_key": "operator-request-003", + "source_session_id": CURRENT_SOURCE_SESSION_ID, + "setup_id": "lab-v1-ravnoves004tree-final", + "definition_sha256": "f" * 64, + }, + ) + assert no_definition.status_code == 409 + + +@pytest.mark.parametrize( + "request_patch", + [ + {"idempotency_key": "../escape"}, + {"source_session_id": "../source"}, + {"setup_id": "../setup"}, + {"definition_sha256": "not-a-digest"}, + ], +) +def test_preparation_endpoint_rejects_unsafe_request_fields( + tmp_path: Path, + request_patch: dict[str, object], +) -> None: + client, _ledger, _source_store = _client(tmp_path) + document = {**_request(), **request_patch} + + assert client.post( + "/api/v1/observatory/run-preparations", + json=document, + ).status_code == 422 + + +def test_preparation_endpoint_fails_closed_when_ledger_is_unavailable() -> None: + app = FastAPI() + app.include_router( + build_observatory_router( + _ReadOnlySourceStore(), # type: ignore[arg-type] + setup_registry=_registry(), + run_preparation_ledger_error="database corrupt", + ) + ) + client = TestClient(app) + + assert client.post( + "/api/v1/observatory/run-preparations", + json=_request(), + ).status_code == 503 + assert client.get( + "/api/v1/observatory/run-preparations/" + "observatory-prep-00000000000000000000000000000000" + ).status_code == 503