From 1b3e0b3406ed6f9d336db7fd56ccf9690134a55c Mon Sep 17 00:00:00 2001 From: DCCONSTRUCTIONS Date: Wed, 5 Aug 2026 12:29:06 +0300 Subject: [PATCH] refactor(platform): freeze laboratory and telemetry boundaries --- .../l3-pointpillars-visual-audit.json | 10 + .../l31-pointpillars-ravnoves.json | 10 + .../l32-pointpillars-camera-review.json | 10 + .../l33-camera-first-detector-review.json | 10 + config/laboratory-execution.json | 83 +++ deploy/telemetry-plane/.env.example | 3 - deploy/telemetry-plane/README.md | 44 +- deploy/telemetry-plane/prepare.py | 486 +++++++++++++--- .../telegraf/mission-core-windows.conf.tmpl | 2 +- ...16_ARCHITECTURE_AUDIT_EXECUTION_ROADMAP.md | 17 +- ...1-local-compute-contour-telemetry-plane.md | 24 + ...y-canonicalization-and-intentional-debt.md | 30 +- src/k1link/compute/pipeline_telemetry.py | 52 +- src/k1link/laboratory/__init__.py | 22 + src/k1link/laboratory/evidence_report.py | 37 ++ src/k1link/laboratory/execution.py | 544 ++++++++++++++++++ src/k1link/web/advanced_laboratory_api.py | 8 +- src/k1link/web/app.py | 5 + tests/test_laboratory_evidence_registry.py | 6 +- tests/test_laboratory_execution.py | 193 +++++++ tests/test_pipeline_telemetry.py | 36 ++ tests/test_telemetry_plane_prepare.py | 173 +++++- 22 files changed, 1657 insertions(+), 148 deletions(-) create mode 100644 config/laboratories/l3-pointpillars-visual-audit.json create mode 100644 config/laboratories/l31-pointpillars-ravnoves.json create mode 100644 config/laboratories/l32-pointpillars-camera-review.json create mode 100644 config/laboratories/l33-camera-first-detector-review.json create mode 100644 config/laboratory-execution.json create mode 100644 src/k1link/laboratory/execution.py create mode 100644 tests/test_laboratory_execution.py diff --git a/config/laboratories/l3-pointpillars-visual-audit.json b/config/laboratories/l3-pointpillars-visual-audit.json new file mode 100644 index 0000000..131740f --- /dev/null +++ b/config/laboratories/l3-pointpillars-visual-audit.json @@ -0,0 +1,10 @@ +{ + "schema_version": "missioncore.laboratory-evidence-definition/v1", + "work_id": "l3-pointpillars-visual-audit", + "evidence": { + "runtime_relative_root": "l3/visual-audits", + "result_id_prefix": "l3-pointpillars-visual-audit", + "document_name": "manifest.json", + "schema_version": "missioncore.l3-pointpillars-visual-audit/v1" + } +} diff --git a/config/laboratories/l31-pointpillars-ravnoves.json b/config/laboratories/l31-pointpillars-ravnoves.json new file mode 100644 index 0000000..3f2d00f --- /dev/null +++ b/config/laboratories/l31-pointpillars-ravnoves.json @@ -0,0 +1,10 @@ +{ + "schema_version": "missioncore.laboratory-evidence-definition/v1", + "work_id": "l31-pointpillars-ravnoves", + "evidence": { + "runtime_relative_root": "l3/pointpillars-ravnoves", + "result_id_prefix": "l31-pointpillars-ravnoves", + "document_name": "manifest.json", + "schema_version": "missioncore.l31-pointpillars-ravnoves/v1" + } +} diff --git a/config/laboratories/l32-pointpillars-camera-review.json b/config/laboratories/l32-pointpillars-camera-review.json new file mode 100644 index 0000000..65478ed --- /dev/null +++ b/config/laboratories/l32-pointpillars-camera-review.json @@ -0,0 +1,10 @@ +{ + "schema_version": "missioncore.laboratory-evidence-definition/v1", + "work_id": "l32-pointpillars-camera-review", + "evidence": { + "runtime_relative_root": "l3/pointpillars-camera-review", + "result_id_prefix": "l32-pointpillars-camera-review", + "document_name": "manifest.json", + "schema_version": "missioncore.l32-pointpillars-camera-review/v1" + } +} diff --git a/config/laboratories/l33-camera-first-detector-review.json b/config/laboratories/l33-camera-first-detector-review.json new file mode 100644 index 0000000..9e0827a --- /dev/null +++ b/config/laboratories/l33-camera-first-detector-review.json @@ -0,0 +1,10 @@ +{ + "schema_version": "missioncore.laboratory-evidence-definition/v1", + "work_id": "l33-camera-first-detector-review", + "evidence": { + "runtime_relative_root": "l3/camera-first-detector-review", + "result_id_prefix": "l33-camera-first-detector-review", + "document_name": "manifest.json", + "schema_version": "missioncore.l33-camera-first-detector-review/v1" + } +} diff --git a/config/laboratory-execution.json b/config/laboratory-execution.json new file mode 100644 index 0000000..4ca162d --- /dev/null +++ b/config/laboratory-execution.json @@ -0,0 +1,83 @@ +{ + "schema_version": "missioncore.laboratory-execution-registry/v1", + "definitions": [ + { + "work_id": "e33-worker-shadow", + "lifecycle": "canonical", + "isolation": "core-adapter", + "adapter_id": "canonical.e33-worker-shadow/v1", + "input_roles": ["package_root"], + "contracts": { + "source": "missioncore.e32-track-geometry-replay/v1", + "provider": "missioncore.e33-recorded-worker-provider/v1", + "graph": "missioncore.e33-worker-shadow-graph/v1", + "run": "missioncore.laboratory-run/v1", + "evidence": "missioncore.e33-worker-shadow-result/v1" + } + }, + { + "work_id": "e35-degradation-recovery", + "lifecycle": "canonical", + "isolation": "core-adapter", + "adapter_id": "canonical.e35-degradation-recovery/v1", + "input_roles": [ + "e32_result_root", + "e33_result_root", + "e34_result_root", + "e34_profile_path", + "profile_path" + ], + "contracts": { + "source": "missioncore.e35-upstream-result-set/v1", + "provider": "missioncore.e35-degradation-provider/v1", + "graph": "missioncore.e35-degradation-recovery-graph/v1", + "run": "missioncore.laboratory-run/v1", + "evidence": "missioncore.e35-degradation-result/v1" + } + }, + { + "work_id": "e46j-raw-fisheye-realtime", + "lifecycle": "canonical", + "isolation": "core-adapter", + "adapter_id": "canonical.e46j-raw-fisheye-realtime/v1", + "input_roles": ["raw_root", "profile_path"], + "contracts": { + "source": "missioncore.e46j-raw-fisheye-source/v1", + "provider": "missioncore.e46j-yolox-provider/v1", + "graph": "missioncore.e46j-raw-fisheye-graph/v1", + "run": "missioncore.laboratory-run/v1", + "evidence": "missioncore.e46j-raw-fisheye-realtime-result/v1" + } + } + ], + "legacy_work_ids": [ + "e31-source-binding", + "e32-track-geometry", + "e34-temporal-layer", + "e37-ravnoves-acceptance", + "e38-perception-baseline", + "e39-perception-refinement", + "e40-perception-product-gate", + "e46-detector-truth-island", + "e46a-ai-engineering-preannotation", + "e46b-temporal-motion", + "e46c-full-replay-world-tracks", + "e46d-temporal-failure-audit", + "e46e-ready-stack", + "e46f-dashcam-bakeoff", + "e46g-rectified-detector-bakeoff", + "e46h-full-rectified-front-replay", + "e46i-grounding-dino-full-replay", + "l3-pointpillars-visual-audit", + "l31-pointpillars-ravnoves", + "l32-pointpillars-camera-review", + "l33-camera-first-detector-review", + "l34-right-yolox-truth-island-freeze", + "l34a-assisted-yolox-error-audit", + "l34b-nested-box-consolidation-shadow", + "l34c-tile-seam-stitch-shadow", + "l34d-cumulative-postprocessing-candidate", + "l34e-self-review-diagnostic", + "l34f-adjudicated-reference" + ] +} diff --git a/deploy/telemetry-plane/.env.example b/deploy/telemetry-plane/.env.example index 9729f36..f0c807d 100644 --- a/deploy/telemetry-plane/.env.example +++ b/deploy/telemetry-plane/.env.example @@ -7,6 +7,3 @@ MISSIONCORE_DB_PASSWORD=replace-with-a-random-local-secret MISSIONCORE_DB_INGEST_PASSWORD=replace-with-a-separate-ingest-secret MISSIONCORE_MQTT_INGEST_USER=missioncore-ingest MISSIONCORE_MQTT_INGEST_PASSWORD=replace-with-a-random-local-secret -MISSIONCORE_MQTT_WORKER_006_USER=worker-006 -MISSIONCORE_MQTT_WORKER_006_CONTOUR=worker-006 -MISSIONCORE_MQTT_WORKER_006_PASSWORD=replace-with-a-different-random-local-secret diff --git a/deploy/telemetry-plane/README.md b/deploy/telemetry-plane/README.md index 1c01af1..1beada1 100644 --- a/deploy/telemetry-plane/README.md +++ b/deploy/telemetry-plane/README.md @@ -43,9 +43,43 @@ unique passwords and build the ACL/password files before starting the stack. ```bash uv run python prepare.py --initialize --mqtt-bind-address +uv run python prepare.py --enroll-agent worker-006 --contour-id worker-006 +uv run python prepare.py docker compose up -d --build ``` +`--enroll-agent` is generic: repeat it with a new globally unique `agent-id` and +the contour it owns. Agent credentials live only in the private +`runtime/agents.json` registry with mode `0600`. A normal `prepare.py` run rebuilds +the Mosquitto password file from that exact registry and generates one non-wildcard +writer ACL per agent, so removed or renamed identities cannot survive in the broker +password file by accident. Existing installations can use `--migrate` to import the +old Worker 006 variables once without replacing their credential. + +Build the reviewed Windows installer bundle and export the selected agent's private +stdin payload as separate files: + +```bash +uv run python prepare.py --build-agent-bundle windows \ + --output runtime/missioncore-telemetry-agent-windows.zip +uv run python prepare.py --export-agent-payload worker-006 \ + --node-id DESKTOP-OPJ8J04 \ + --output runtime/worker-006.private.json +``` + +The ZIP is deterministic, content-addressed and contains only the pinned installer, +updater, collector and configuration template. It contains no credential. Transfer +the ZIP and private payload separately; on the Worker, unpack the ZIP and pass the +payload through stdin: + +```powershell +Get-Content .\worker-006.private.json -Raw | + .\Install-NdcMissionCoreTelegraf.ps1 +``` + +Delete the transferred payload after the service has been accepted. The installer +persists the scoped credential only in the ACL-restricted Windows service environment. + Expected Docker object names: ```text @@ -97,8 +131,8 @@ replacement for MQTT TLS. Each agent credential is bound to one exact `contours//agents//+` prefix. Adding another contour requires -issuing another password entry and explicit ACL row; the wildcard contour writer is -not permitted. +`--enroll-agent`; the wildcard contour writer is not permitted. An `agent-id` is +globally unique because Mosquitto ACL ownership is username-based. ## Worker agent @@ -134,6 +168,12 @@ frame boundary and aggregate activation count without producing one MQTT row per frame activation. Telegraf's `inputs.tail` owns the saved file offset, keeps at most 1000 undelivered lines in flight and publishes the records through the same QoS 1 pipeline output. +The source outbox rotates at 64 MiB into content-addressed +`pipeline-telemetry..jsonl` segments. Telegraf tails both the active file and +segments. At eight retained segments the writer fails telemetry publication +observably instead of deleting evidence which may not yet have been acknowledged. +Segment reclamation therefore remains an explicit operator action after normalized +storage is verified; inference control flow remains fail-open. The normalizer verifies the topic-bound record and restores the original native document before storage. A broker outage therefore stays inside the existing Telegraf buffer; the perception container receives neither MQTT credentials nor a diff --git a/deploy/telemetry-plane/prepare.py b/deploy/telemetry-plane/prepare.py index 3d62ad3..38cdc99 100644 --- a/deploy/telemetry-plane/prepare.py +++ b/deploy/telemetry-plane/prepare.py @@ -1,22 +1,56 @@ from __future__ import annotations import argparse +import hashlib import ipaddress +import json import os import re import secrets import subprocess +import uuid +import zipfile +from dataclasses import dataclass from pathlib import Path from typing import Final ROOT: Final = Path(__file__).resolve().parent ENV_PATH: Final = ROOT / ".env" -RUNTIME: Final = ROOT / "runtime" / "mosquitto" +RUNTIME_ROOT: Final = ROOT / "runtime" +RUNTIME: Final = RUNTIME_ROOT / "mosquitto" +AGENT_REGISTRY_PATH: Final = RUNTIME_ROOT / "agents.json" IMAGE: Final = "eclipse-mosquitto:2.1.2-alpine" PLACEHOLDER: Final = "replace-with-" +AGENT_REGISTRY_SCHEMA: Final = "missioncore.telemetry-agent-credential-registry/v1" +AGENT_PAYLOAD_SCHEMA: Final = "missioncore.telemetry-agent-provisioning-payload/v1" +AGENT_BUNDLE_SCHEMA: Final = "missioncore.telemetry-agent-bundle/v1" SAFE_IDENTIFIER: Final = re.compile( r"^[a-z0-9](?:[a-z0-9-]{0,62}[a-z0-9])?$" ) +SAFE_INTERVAL: Final = re.compile(r"^[1-9][0-9]{0,2}s$") +WINDOWS_BUNDLE_FILES: Final = ( + "Get-NdcMissionCorePipelineTelemetry.ps1", + "Install-NdcMissionCoreTelegraf.ps1", + "Update-NdcMissionCoreTelegraf.ps1", + "mission-core-windows.conf.tmpl", +) + + +@dataclass(frozen=True, slots=True) +class AgentCredential: + contour_id: str + agent_id: str + password: str + + def __post_init__(self) -> None: + _safe_identifier(self.contour_id, "contour_id") + _safe_identifier(self.agent_id, "agent_id") + if ( + not 32 <= len(self.password) <= 512 + or self.password.startswith(PLACEHOLDER) + or any(ord(character) < 33 for character in self.password) + ): + raise RuntimeError("agent password must contain a non-placeholder value") def _initialize_environment(bind_address: str, *, overwrite: bool = False) -> None: @@ -33,20 +67,17 @@ def _initialize_environment(bind_address: str, *, overwrite: bool = False) -> No "MISSIONCORE_DB_INGEST_PASSWORD": secrets.token_urlsafe(36), "MISSIONCORE_MQTT_INGEST_USER": "missioncore-ingest", "MISSIONCORE_MQTT_INGEST_PASSWORD": secrets.token_urlsafe(36), - "MISSIONCORE_MQTT_WORKER_006_USER": "worker-006", - "MISSIONCORE_MQTT_WORKER_006_CONTOUR": "worker-006", - "MISSIONCORE_MQTT_WORKER_006_PASSWORD": secrets.token_urlsafe(36), } - ENV_PATH.write_text( - "".join(f"{name}={value}\n" for name, value in values.items()), - encoding="utf-8", + _write_private( + ENV_PATH, + "".join(f"{name}={value}\n" for name, value in values.items()).encode(), + replace=overwrite, ) - os.chmod(ENV_PATH, 0o600) def _environment() -> dict[str, str]: if not ENV_PATH.is_file(): - raise RuntimeError("copy .env.example to .env and set unique secrets first") + raise RuntimeError("run prepare.py --initialize and set unique secrets first") values: dict[str, str] = {} for raw_line in ENV_PATH.read_text(encoding="utf-8").splitlines(): line = raw_line.strip() @@ -62,16 +93,190 @@ def _migrate_environment() -> None: additions: dict[str, str] = {} if "MISSIONCORE_DB_INGEST_PASSWORD" not in values: additions["MISSIONCORE_DB_INGEST_PASSWORD"] = secrets.token_urlsafe(36) - if "MISSIONCORE_MQTT_WORKER_006_CONTOUR" not in values: - additions["MISSIONCORE_MQTT_WORKER_006_CONTOUR"] = "worker-006" - if not additions: + if additions: + with ENV_PATH.open("a", encoding="utf-8", newline="\n") as stream: + for name, value in additions.items(): + stream.write(f"{name}={value}\n") + stream.flush() + os.fsync(stream.fileno()) + os.chmod(ENV_PATH, 0o600) + values.update(additions) + _migrate_legacy_worker(values) + + +def _migrate_legacy_worker(values: dict[str, str]) -> None: + legacy_names = ( + "MISSIONCORE_MQTT_WORKER_006_USER", + "MISSIONCORE_MQTT_WORKER_006_CONTOUR", + "MISSIONCORE_MQTT_WORKER_006_PASSWORD", + ) + if AGENT_REGISTRY_PATH.exists() or not all(values.get(name) for name in legacy_names): return - with ENV_PATH.open("a", encoding="utf-8", newline="\n") as stream: - for name, value in additions.items(): - stream.write(f"{name}={value}\n") - stream.flush() - os.fsync(stream.fileno()) - os.chmod(ENV_PATH, 0o600) + credential = AgentCredential( + contour_id=_identifier(values, legacy_names[1]), + agent_id=_identifier(values, legacy_names[0]), + password=_required(values, legacy_names[2]), + ) + _write_agent_registry((credential,)) + + +def _read_agent_registry(*, missing_ok: bool = False) -> tuple[AgentCredential, ...]: + if not AGENT_REGISTRY_PATH.exists(): + if missing_ok: + return () + raise RuntimeError("no telemetry agents enrolled; use --enroll-agent first") + if AGENT_REGISTRY_PATH.is_symlink() or not AGENT_REGISTRY_PATH.is_file(): + raise RuntimeError("telemetry agent registry must be a regular file") + if AGENT_REGISTRY_PATH.stat().st_size > 1024 * 1024: + raise RuntimeError("telemetry agent registry is too large") + try: + payload: object = json.loads(AGENT_REGISTRY_PATH.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as exc: + raise RuntimeError("telemetry agent registry is unreadable") from exc + if not isinstance(payload, dict) or set(payload) != {"schema_version", "agents"}: + raise RuntimeError("telemetry agent registry shape is invalid") + if payload["schema_version"] != AGENT_REGISTRY_SCHEMA: + raise RuntimeError("telemetry agent registry schema is invalid") + rows = payload["agents"] + if not isinstance(rows, list): + raise RuntimeError("telemetry agent registry agents must be an array") + credentials: list[AgentCredential] = [] + for row in rows: + if not isinstance(row, dict) or set(row) != {"contour_id", "agent_id", "password"}: + raise RuntimeError("telemetry agent registry row is invalid") + if not all(isinstance(value, str) for value in row.values()): + raise RuntimeError("telemetry agent registry values must be strings") + credentials.append( + AgentCredential( + contour_id=row["contour_id"], + agent_id=row["agent_id"], + password=row["password"], + ) + ) + _reject_agent_duplicates(tuple(credentials)) + os.chmod(AGENT_REGISTRY_PATH, 0o600) + return tuple(credentials) + + +def _write_agent_registry(credentials: tuple[AgentCredential, ...]) -> None: + _reject_agent_duplicates(credentials) + document = { + "schema_version": AGENT_REGISTRY_SCHEMA, + "agents": [ + { + "contour_id": credential.contour_id, + "agent_id": credential.agent_id, + "password": credential.password, + } + for credential in sorted(credentials, key=lambda item: item.agent_id) + ], + } + _write_private( + AGENT_REGISTRY_PATH, + _canonical_json(document) + b"\n", + replace=True, + ) + + +def _enroll_agent(contour_id: str, agent_id: str) -> AgentCredential: + credential = AgentCredential( + contour_id=contour_id, + agent_id=agent_id, + password=secrets.token_urlsafe(36), + ) + existing = _read_agent_registry(missing_ok=True) + if any(row.agent_id == agent_id for row in existing): + raise RuntimeError(f"telemetry agent is already enrolled: {agent_id}") + _write_agent_registry((*existing, credential)) + return credential + + +def _export_agent_payload( + *, + agent_id: str, + node_id: str, + output: Path, + mqtt_host: str | None = None, + telemetry_interval: str = "2s", +) -> None: + if ( + not node_id + or node_id != node_id.strip() + or len(node_id) > 256 + or any(ord(character) < 32 for character in node_id) + ): + raise RuntimeError("node_id is invalid") + if SAFE_INTERVAL.fullmatch(telemetry_interval) is None: + raise RuntimeError("telemetry interval must be 1-999 seconds") + matches = [row for row in _read_agent_registry() if row.agent_id == agent_id] + if len(matches) != 1: + raise RuntimeError(f"telemetry agent is not enrolled: {agent_id}") + credential = matches[0] + values = _environment() + resolved_host = mqtt_host or _required(values, "MISSIONCORE_MQTT_BIND_ADDRESS") + _private_mqtt_host(resolved_host) + port = _required(values, "MISSIONCORE_MQTT_PORT") + if not port.isdigit() or not 1 <= int(port) <= 65535: + raise RuntimeError("MISSIONCORE_MQTT_PORT is invalid") + payload = { + "schema_version": AGENT_PAYLOAD_SCHEMA, + "MISSIONCORE_CONTOUR_ID": credential.contour_id, + "MISSIONCORE_AGENT_ID": credential.agent_id, + "MISSIONCORE_NODE_ID": node_id, + "MISSIONCORE_MQTT_HOST": resolved_host, + "MISSIONCORE_MQTT_PORT": port, + "MISSIONCORE_MQTT_USERNAME": credential.agent_id, + "MISSIONCORE_MQTT_PASSWORD": credential.password, + "MISSIONCORE_TELEMETRY_INTERVAL": telemetry_interval, + } + _write_private(output, _canonical_json(payload) + b"\n", replace=False) + + +def _build_agent_bundle(*, platform_name: str, output: Path) -> str: + if platform_name != "windows": + raise RuntimeError("only the reviewed windows telemetry agent bundle is available") + candidate = output.expanduser().absolute() + if candidate.exists() or candidate.is_symlink(): + raise RuntimeError("agent bundle output already exists") + sources = [(name, ROOT / "telegraf" / name) for name in WINDOWS_BUNDLE_FILES] + inventory = [] + for name, path in sources: + if path.is_symlink() or not path.is_file(): + raise RuntimeError(f"agent bundle source is unavailable: {name}") + content = path.read_bytes() + inventory.append( + { + "path": name, + "byte_length": len(content), + "sha256": hashlib.sha256(content).hexdigest(), + } + ) + identity = { + "schema_version": AGENT_BUNDLE_SCHEMA, + "platform": platform_name, + "credential_embedded": False, + "files": inventory, + } + identity_sha256 = hashlib.sha256(_canonical_json(identity)).hexdigest() + manifest = { + **identity, + "bundle_id": f"missioncore-telemetry-agent-{platform_name}-{identity_sha256}", + "identity_sha256": identity_sha256, + } + candidate.parent.mkdir(mode=0o700, parents=True, exist_ok=True) + staging = candidate.parent / f".{candidate.name}.{uuid.uuid4().hex}.tmp" + with zipfile.ZipFile( + staging, + "w", + compression=zipfile.ZIP_DEFLATED, + compresslevel=9, + ) as archive: + _zip_entry(archive, "manifest.json", _canonical_json(manifest) + b"\n") + for name, path in sources: + _zip_entry(archive, name, path.read_bytes()) + os.chmod(staging, 0o600) + os.replace(staging, candidate) + return str(manifest["bundle_id"]) def _required(values: dict[str, str], name: str) -> str: @@ -82,12 +287,36 @@ def _required(values: dict[str, str], name: str) -> str: def _identifier(values: dict[str, str], name: str) -> str: - value = _required(values, name) + return _safe_identifier(_required(values, name), name) + + +def _safe_identifier(value: str, label: str) -> str: if SAFE_IDENTIFIER.fullmatch(value) is None: - raise RuntimeError(f"{name} must contain a DNS-safe lowercase identifier") + raise RuntimeError(f"{label} must contain a DNS-safe lowercase identifier") return value +def _private_mqtt_host(value: str) -> None: + try: + address = ipaddress.ip_address(value) + except ValueError: + if not value.endswith(".local") or len(value) > 253 or any( + SAFE_IDENTIFIER.fullmatch(label) is None for label in value[:-6].split(".") + ): + raise RuntimeError( + "MQTT host must be a private IP or stable .local hostname" + ) from None + else: + if not address.is_private or address.is_unspecified or address.is_multicast: + raise RuntimeError("MQTT host must be a private IP or stable .local hostname") + + +def _reject_agent_duplicates(credentials: tuple[AgentCredential, ...]) -> None: + agent_ids = [credential.agent_id for credential in credentials] + if len(agent_ids) != len(set(agent_ids)): + raise RuntimeError("telemetry agent_id must be globally unique") + + def _password_entry(path: Path, username: str, password: str, *, create: bool) -> None: command = [ "docker", @@ -114,82 +343,169 @@ def _prepare_password_entries( path: Path, ingest_user: str, ingest_password: str, - worker_user: str, - worker_password: str, + agents: tuple[AgentCredential, ...], ) -> None: - _password_entry( - path, - ingest_user, - ingest_password, - create=not path.exists(), + staging = path.parent / f".{path.name}.{uuid.uuid4().hex}.tmp" + try: + _password_entry(staging, ingest_user, ingest_password, create=True) + for credential in sorted(agents, key=lambda item: item.agent_id): + _password_entry( + staging, + credential.agent_id, + credential.password, + create=False, + ) + os.chmod(staging, 0o600) + os.replace(staging, path) + finally: + if staging.exists(): + staging.unlink() + + +def _acl_document(ingest_user: str, agents: tuple[AgentCredential, ...]) -> str: + lines = [ + f"user {ingest_user}", + "topic read mission-core/v1/contours/+/agents/+/+", + "topic read $SYS/broker/uptime", + "", + "# Every credential is scoped to one contour and one stable agent.", + ] + for credential in sorted(agents, key=lambda item: item.agent_id): + lines.extend( + [ + f"user {credential.agent_id}", + ( + "topic write mission-core/v1/contours/" + f"{credential.contour_id}/agents/{credential.agent_id}/+" + ), + "", + ] + ) + return "\n".join(lines) + + +def _prepare_plane() -> None: + values = _environment() + _migrate_legacy_worker(values) + agents = _read_agent_registry() + ingest_user = _required(values, "MISSIONCORE_MQTT_INGEST_USER") + ingest_password = _required(values, "MISSIONCORE_MQTT_INGEST_PASSWORD") + _required(values, "MISSIONCORE_DB_PASSWORD") + RUNTIME.mkdir(mode=0o700, parents=True, exist_ok=True) + password_path = RUNTIME / "passwords" + _prepare_password_entries(password_path, ingest_user, ingest_password, agents) + acl_path = RUNTIME / "acl" + _write_private( + acl_path, + _acl_document(ingest_user, agents).encode("utf-8"), + replace=True, ) - _password_entry(path, worker_user, worker_password, create=False) + + +def _write_private(path: Path, content: bytes, *, replace: bool) -> None: + candidate = path.expanduser().absolute() + if candidate.is_symlink() or (candidate.exists() and not replace): + raise RuntimeError(f"refusing to replace private file: {candidate.name}") + candidate.parent.mkdir(mode=0o700, parents=True, exist_ok=True) + staging = candidate.parent / f".{candidate.name}.{uuid.uuid4().hex}.tmp" + descriptor = os.open(staging, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600) + try: + os.write(descriptor, content) + os.fsync(descriptor) + finally: + os.close(descriptor) + try: + os.replace(staging, candidate) + os.chmod(candidate, 0o600) + finally: + if staging.exists(): + staging.unlink() + + +def _zip_entry(archive: zipfile.ZipFile, name: str, content: bytes) -> None: + info = zipfile.ZipInfo(name, date_time=(2026, 1, 1, 0, 0, 0)) + info.compress_type = zipfile.ZIP_DEFLATED + info.external_attr = 0o600 << 16 + info.create_system = 3 + archive.writestr(info, content, compress_type=zipfile.ZIP_DEFLATED, compresslevel=9) + + +def _canonical_json(value: object) -> bytes: + return json.dumps( + value, + sort_keys=True, + separators=(",", ":"), + ensure_ascii=False, + ).encode("utf-8") def main() -> None: parser = argparse.ArgumentParser() - parser.add_argument( - "--initialize", - action="store_true", - help="create a private .env with generated local credentials", - ) - parser.add_argument( - "--migrate", - action="store_true", - help="add newly required generated secrets without replacing existing values", - ) - parser.add_argument( - "--mqtt-bind-address", - default="127.0.0.1", - help="host IP exposed to telemetry agents", - ) + parser.add_argument("--initialize", action="store_true") + parser.add_argument("--migrate", action="store_true") + parser.add_argument("--mqtt-bind-address", default="127.0.0.1") + parser.add_argument("--enroll-agent", metavar="AGENT_ID") + parser.add_argument("--contour-id") + parser.add_argument("--export-agent-payload", metavar="AGENT_ID") + parser.add_argument("--node-id") + parser.add_argument("--output", type=Path) + parser.add_argument("--mqtt-host") + parser.add_argument("--telemetry-interval", default="2s") + parser.add_argument("--build-agent-bundle", choices=("windows",)) arguments = parser.parse_args() + + action_count = sum( + bool(value) + for value in ( + arguments.initialize, + arguments.migrate, + arguments.enroll_agent, + arguments.export_agent_payload, + arguments.build_agent_bundle, + ) + ) + if action_count > 1: + raise RuntimeError("select exactly one provisioning action") if arguments.initialize: _initialize_environment(arguments.mqtt_bind_address) + print("Private telemetry-plane environment initialized.", flush=True) + return if arguments.migrate: _migrate_environment() - values = _environment() - ingest_user = _required(values, "MISSIONCORE_MQTT_INGEST_USER") - ingest_password = _required(values, "MISSIONCORE_MQTT_INGEST_PASSWORD") - worker_user = _identifier(values, "MISSIONCORE_MQTT_WORKER_006_USER") - worker_contour = _identifier( - values, - "MISSIONCORE_MQTT_WORKER_006_CONTOUR", - ) - worker_password = _required(values, "MISSIONCORE_MQTT_WORKER_006_PASSWORD") - _required(values, "MISSIONCORE_DB_PASSWORD") - - RUNTIME.mkdir(parents=True, exist_ok=True) - password_path = RUNTIME / "passwords" - _prepare_password_entries( - password_path, - ingest_user, - ingest_password, - worker_user, - worker_password, - ) - acl_path = RUNTIME / "acl" - acl_path.write_text( - "\n".join( - [ - f"user {ingest_user}", - "topic read mission-core/v1/contours/+/agents/+/+", - "topic read $SYS/broker/uptime", - "", - "# Every credential is scoped to one contour and one stable agent.", - f"user {worker_user}", - ( - "topic write mission-core/v1/contours/" - f"{worker_contour}/agents/{worker_user}/+" - ), - "", - ] - ), - encoding="utf-8", - ) - os.chmod(password_path, 0o600) - os.chmod(acl_path, 0o600) - print("Mosquitto password and ACL files prepared.", flush=True) + print("Private telemetry-plane state migrated.", flush=True) + return + if arguments.enroll_agent: + if not arguments.contour_id: + raise RuntimeError("--contour-id is required with --enroll-agent") + credential = _enroll_agent(arguments.contour_id, arguments.enroll_agent) + print( + f"Telemetry agent enrolled: {credential.agent_id} ({credential.contour_id}).", + flush=True, + ) + return + if arguments.export_agent_payload: + if not arguments.node_id or arguments.output is None: + raise RuntimeError("--node-id and --output are required for payload export") + _export_agent_payload( + agent_id=arguments.export_agent_payload, + node_id=arguments.node_id, + output=arguments.output, + mqtt_host=arguments.mqtt_host, + telemetry_interval=arguments.telemetry_interval, + ) + print(f"Private agent payload written: {arguments.output.name}.", flush=True) + return + if arguments.build_agent_bundle: + if arguments.output is None: + raise RuntimeError("--output is required for agent bundle build") + bundle_id = _build_agent_bundle( + platform_name=arguments.build_agent_bundle, + output=arguments.output, + ) + print(f"Secret-free agent bundle built: {bundle_id}.", flush=True) + return + _prepare_plane() + print("Mosquitto password and ACL files prepared for all enrolled agents.", flush=True) if __name__ == "__main__": diff --git a/deploy/telemetry-plane/telegraf/mission-core-windows.conf.tmpl b/deploy/telemetry-plane/telegraf/mission-core-windows.conf.tmpl index fab2e09..13ba712 100644 --- a/deploy/telemetry-plane/telegraf/mission-core-windows.conf.tmpl +++ b/deploy/telemetry-plane/telegraf/mission-core-windows.conf.tmpl @@ -62,7 +62,7 @@ ] [[inputs.tail]] - files = ["D:\\NDC_MISSIONCORE\\runtime\\derived\\.perception-persistent-publish\\pipeline-telemetry.jsonl"] + files = ["D:\\NDC_MISSIONCORE\\runtime\\derived\\.perception-persistent-publish\\pipeline-telemetry*.jsonl"] initial_read_offset = "saved-or-beginning" watch_method = "poll" max_undelivered_lines = 1000 diff --git a/docs/16_ARCHITECTURE_AUDIT_EXECUTION_ROADMAP.md b/docs/16_ARCHITECTURE_AUDIT_EXECUTION_ROADMAP.md index 3073b91..4149699 100644 --- a/docs/16_ARCHITECTURE_AUDIT_EXECUTION_ROADMAP.md +++ b/docs/16_ARCHITECTURE_AUDIT_EXECUTION_ROADMAP.md @@ -108,18 +108,19 @@ The evidence/report layer is now generic and proof-bound, but historical LAB execution remains heterogeneous. ADR 0039 limits the remaining stabilization work instead of turning it into a bulk archive migration. -The gate is: +The gate is implemented for the current architecture freeze: -1. Preserve all existing historical LABs as immutable `legacy` evidence. -2. Admit genuinely novel stacks as bounded `experimental` adapters rather than +1. [x] Preserve all existing historical LABs as immutable `legacy` evidence. +2. [x] Admit genuinely novel stacks as bounded `experimental` adapters rather than discarding them or importing them directly into core. -3. Prove one source/provider/graph/run path on E33, E35 and E46J. -4. Ensure a new ordinary `canonical` LAB does not add a new branch to the +3. [x] Prove one source/provider/graph/run path on E33, E35 and E46J. +4. [x] Ensure a new ordinary `canonical` LAB does not add a new branch to the application shell, advanced dispatcher or generic workspace. -5. Close local telemetry operational debt: multi-contour credential enrollment, +5. [x] Close local architecture-level telemetry debt: multi-contour credential enrollment, executable agent bundle, journal rotation, restart/reconnect/duplicate/stale/ - identity-mismatch acceptance and run correlation. -6. Record production/fleet, second-source, Quick Connect, physical mount and + identity-mismatch contracts and run correlation. Second-agent physical acceptance + remains an intentional deployment debt, not a CV-development blocker. +6. [x] Record production/fleet, second-source, Quick Connect, physical mount and live-authority gaps as intentional non-blocking debt with explicit triggers. A9/E36 remains a later scientific transfer gate. It is not a precondition for diff --git a/docs/adr/0031-local-compute-contour-telemetry-plane.md b/docs/adr/0031-local-compute-contour-telemetry-plane.md index 3dfa67b..3e78d86 100644 --- a/docs/adr/0031-local-compute-contour-telemetry-plane.md +++ b/docs/adr/0031-local-compute-contour-telemetry-plane.md @@ -102,6 +102,30 @@ the failure counter after one day and applying recovery to non-crash failures. T does not make an invalid endpoint valid; it keeps transient broker or network loss from silently leaving the telemetry agent stopped after connectivity returns. +## 2026-08-05 generic enrollment and bounded outbox + +Worker 006 is no longer encoded in the broker bootstrap contract. The private +`missioncore.telemetry-agent-credential-registry/v1` registry admits any globally +unique agent identity, binds it to one contour and rebuilds the broker password file +plus exact writer ACLs from the registry. The old Worker 006 environment variables +are accepted only by a one-time private migration and are not part of the current +example configuration. + +The reviewed Windows host-agent package is generated as a deterministic, +content-addressed and secret-free ZIP. A selected agent's credential is exported in a +separate non-overwriting `0600` stdin payload. This closes repeatable agent +installation without pretending to automate GPU drivers, Docker/WSL, model stores or +the perception runtime. + +The native run journal is a bounded outbox. It rotates active JSONL at 64 MiB into a +content-addressed segment and retains at most eight segments. Telegraf tails the +active path and segment glob using tracking metrics and a saved offset. Reaching the +bound raises an observable telemetry transport error rather than deleting potentially +unacknowledged evidence; inference remains fail-open. Segment deletion is not +automatic and requires a separate proof that normalized storage contains the rows. +QoS 1 duplicates remain safe because the normalizer upserts the complete normalized +sample identity. + ## Why these components - **Telegraf** is the universal host agent. It already supports Windows and Linux host diff --git a/docs/adr/0039-bounded-laboratory-canonicalization-and-intentional-debt.md b/docs/adr/0039-bounded-laboratory-canonicalization-and-intentional-debt.md index f0c7dab..f99d4ff 100644 --- a/docs/adr/0039-bounded-laboratory-canonicalization-and-intentional-debt.md +++ b/docs/adr/0039-bounded-laboratory-canonicalization-and-intentional-debt.md @@ -1,7 +1,7 @@ # ADR 0039: Bounded laboratory canonicalization and intentional debt Date: 2026-08-05 -Status: accepted; executable canary migration remains open +Status: accepted and implemented for the architecture freeze ## Context @@ -44,6 +44,15 @@ The first executable canaries are deliberately limited to: Passing those three canaries closes the laboratory canonicalization slice. It does not authorize a bulk migration of historical runs. +The implementation is `missioncore.laboratory-execution-registry/v1` plus the +common `LaboratoryRunner`. The registry classifies every published evidence +definition exactly once, admits only declared input roles and routes E33, E35 and +E46J through bounded adapters. The runner emits native run/stage telemetry, verifies +the immutable evidence document and all artifact hashes, and publishes one +content-addressed `missioncore.laboratory-run-receipt/v1`. Existing canary evidence +was reverified without mutation: E33 has four verified artifacts, E35 four and E46J +seven. All other current evidence definitions are explicitly read-only `legacy`. + ### Experimental An experimental LAB may use a stack or execution shape which the current @@ -91,18 +100,18 @@ its stack cannot yet satisfy the canonical graph. ## Worker and telemetry boundary The compute-contour catalog, selector, generic MQTT topic namespace and telemetry -query path already support more than one contour. They do not constitute worker -provisioning. Credential enrollment, a digest-bound agent bundle and an accepted -second agent remain open. Installing GPU drivers, Docker/WSL, Triton, models and +query path support more than one contour. Generic scoped credential enrollment and a +digest-bound secret-free Windows agent bundle are implemented. A second physical +agent acceptance remains open. Installing GPU drivers, Docker/WSL, Triton, models and the perception runtime is a separate deployment-profile problem and is not folded into a universal one-click installer. -Local telemetry stabilization is part of the architecture freeze: multi-contour -credential issuance, executable agent installation, bounded journal rotation, -restart/reconnect/duplicate handling, stale and identity-mismatch behavior, and -run correlation must be accepted. Fleet TLS, Linux acceptance, HA and production -storage/deployment remain explicit intentional debt until their deployment trigger -exists. +Local telemetry stabilization now includes multi-contour credential issuance, +executable agent installation, bounded fail-closed journal rotation, the accepted +Windows recovery policy, QoS 1/idempotent duplicate handling, stale and +identity-mismatch behavior, and run correlation. Physical acceptance on a second +agent, fleet TLS, Linux acceptance, HA and production storage/deployment remain +explicit intentional debt until their deployment trigger exists. ## Consequences @@ -113,4 +122,3 @@ exists. - The product core grows only when repeated value justifies a reusable contract. - Debt is not treated as a binary pass/fail list: every deferred item names the claim it blocks and the event which returns it to the critical path. - diff --git a/src/k1link/compute/pipeline_telemetry.py b/src/k1link/compute/pipeline_telemetry.py index 40d0b0f..3e64f33 100644 --- a/src/k1link/compute/pipeline_telemetry.py +++ b/src/k1link/compute/pipeline_telemetry.py @@ -8,6 +8,7 @@ No sink is created implicitly and telemetry never grants command authority. from __future__ import annotations +import hashlib import json import os import re @@ -30,6 +31,8 @@ SAFE_TOPIC_IDENTIFIER: Final = re.compile( ) MAX_TEXT_LENGTH: Final = 256 MAX_PAYLOAD_BYTES: Final = 1024 * 1024 +DEFAULT_JOURNAL_MAX_BYTES: Final = 64 * 1024 * 1024 +DEFAULT_JOURNAL_MAX_SEGMENTS: Final = 8 STAGE_STATES: Final = frozenset({"started", "completed", "failed"}) RUN_STATES: Final = STAGE_STATES _AUTHORITY: Final = { @@ -243,10 +246,27 @@ class PipelineTelemetryEmitter: class JsonlPipelineTelemetrySink: - """Append topic-bound telemetry records for local, auditable execution evidence.""" + """Append to a bounded, fail-closed local outbox for auditable execution evidence. - def __init__(self, path: Path) -> None: + Completed segments are content-addressed and never pruned implicitly. When the + segment bound is reached, telemetry publication fails observably instead of + deleting an event that Telegraf may not have acknowledged yet. + """ + + def __init__( + self, + path: Path, + *, + max_bytes: int = DEFAULT_JOURNAL_MAX_BYTES, + max_segments: int = DEFAULT_JOURNAL_MAX_SEGMENTS, + ) -> None: + if max_bytes < MAX_PAYLOAD_BYTES + 4096: + raise PipelineTelemetryError("pipeline journal max_bytes is too small") + if not 1 <= max_segments <= 64: + raise PipelineTelemetryError("pipeline journal max_segments is invalid") self.path = path.expanduser().absolute() + self.max_bytes = max_bytes + self.max_segments = max_segments self._lock = threading.Lock() def publish(self, topic: str, payload: bytes) -> None: @@ -261,6 +281,10 @@ class JsonlPipelineTelemetrySink: encoded = _canonical_json(record) + b"\n" with self._lock: self.path.parent.mkdir(mode=0o700, parents=True, exist_ok=True) + if self.path.is_symlink(): + raise PipelineTelemetryError("pipeline journal must not be a symlink") + if self.path.exists() and self.path.stat().st_size + len(encoded) > self.max_bytes: + self._rotate() descriptor = os.open( self.path, os.O_APPEND | os.O_CREAT | os.O_WRONLY, @@ -272,6 +296,30 @@ class JsonlPipelineTelemetrySink: finally: os.close(descriptor) + def _rotate(self) -> None: + if not self.path.is_file() or self.path.stat().st_size == 0: + return + segments = tuple(self.path.parent.glob(f"{self.path.stem}.*{self.path.suffix}")) + if len(segments) >= self.max_segments: + raise PipelineTelemetryError( + "pipeline journal segment bound reached; acknowledged segments require review" + ) + digest = _file_sha256(self.path) + destination = self.path.with_name( + f"{self.path.stem}.{digest}{self.path.suffix}" + ) + if destination.exists() or destination.is_symlink(): + raise PipelineTelemetryError("pipeline journal segment identity already exists") + os.replace(self.path, destination) + + +def _file_sha256(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as stream: + for block in iter(lambda: stream.read(1024 * 1024), b""): + digest.update(block) + return digest.hexdigest() + class MqttPipelineTelemetrySink: """Publish through a worker-owned, already-connected Paho-compatible client.""" diff --git a/src/k1link/laboratory/__init__.py b/src/k1link/laboratory/__init__.py index e74da78..f2d88aa 100644 --- a/src/k1link/laboratory/__init__.py +++ b/src/k1link/laboratory/__init__.py @@ -11,6 +11,18 @@ from k1link.laboratory.evidence_report import ( LaboratoryEvidenceReportError, LaboratoryEvidenceReportNotFound, LaboratoryEvidenceReportService, + verify_laboratory_evidence_result, +) +from k1link.laboratory.execution import ( + LABORATORY_EXECUTION_REGISTRY_SCHEMA, + LABORATORY_RUN_RECEIPT_SCHEMA, + LaboratoryAdapterResult, + LaboratoryExecutionDefinition, + LaboratoryExecutionError, + LaboratoryExecutionRegistry, + LaboratoryRunner, + LaboratoryRunRequest, + LaboratoryRunResult, ) from k1link.laboratory.value_review_registry import ( LABORATORY_VALUE_REVIEW_INDEX_SCHEMA, @@ -28,7 +40,17 @@ __all__ = [ "LaboratoryEvidenceReportError", "LaboratoryEvidenceReportNotFound", "LaboratoryEvidenceReportService", + "verify_laboratory_evidence_result", "LaboratoryRegistryError", + "LABORATORY_EXECUTION_REGISTRY_SCHEMA", + "LABORATORY_RUN_RECEIPT_SCHEMA", + "LaboratoryAdapterResult", + "LaboratoryExecutionDefinition", + "LaboratoryExecutionError", + "LaboratoryExecutionRegistry", + "LaboratoryRunRequest", + "LaboratoryRunResult", + "LaboratoryRunner", "LABORATORY_VALUE_REVIEW_INDEX_SCHEMA", "LABORATORY_VALUE_REVIEW_REGISTRY_SCHEMA", "LaboratoryValueReviewEntry", diff --git a/src/k1link/laboratory/evidence_report.py b/src/k1link/laboratory/evidence_report.py index 19f6d64..f3167e5 100644 --- a/src/k1link/laboratory/evidence_report.py +++ b/src/k1link/laboratory/evidence_report.py @@ -27,6 +27,43 @@ class LaboratoryEvidenceReportNotFound(LaboratoryEvidenceReportError): """Raised when the requested evidence identity is not available.""" +def verify_laboratory_evidence_result( + definition: LaboratoryEvidenceDefinition, + result_root: Path, +) -> dict[str, object]: + """Verify one immutable result without projecting it into a UI report.""" + + candidate = result_root.expanduser().absolute() + if candidate.is_symlink(): + raise LaboratoryEvidenceReportError("LAB result must not be a symlink") + try: + resolved = candidate.resolve(strict=True) + except OSError as exc: + raise LaboratoryEvidenceReportError("LAB evidence result is unavailable") from exc + if not resolved.is_dir() or definition.result_id_pattern.fullmatch(resolved.name) is None: + raise LaboratoryEvidenceReportError("LAB evidence result path is invalid") + document_path = _safe_file(resolved, definition.document_name) + document_bytes = _read_bounded(document_path, _DOCUMENT_MAX_BYTES, "LAB document") + document = _json_object(document_bytes, "LAB document") + _validate_document(document, definition, resolved.name) + identity = _object_or_none(document.get("identity")) + identity_sha256 = document.get("identity_sha256") + if identity is None or not isinstance(identity_sha256, str): + raise LaboratoryEvidenceReportError("LAB identity proof is missing") + if ( + _canonical_sha256(identity) != identity_sha256 + or not resolved.name.endswith(identity_sha256) + ): + raise LaboratoryEvidenceReportError("LAB identity proof is invalid") + artifacts = _verified_artifacts(resolved, document.get("artifacts")) + return { + "result_id": resolved.name, + "identity_sha256": identity_sha256, + "document_sha256": hashlib.sha256(document_bytes).hexdigest(), + "artifact_count": len(artifacts), + } + + class LaboratoryEvidenceReportService: def __init__( self, diff --git a/src/k1link/laboratory/execution.py b/src/k1link/laboratory/execution.py new file mode 100644 index 0000000..ae0417d --- /dev/null +++ b/src/k1link/laboratory/execution.py @@ -0,0 +1,544 @@ +"""Admission and execution boundary for reproducible Mission Core laboratories. + +Legacy evidence remains readable but is not executable through this boundary. A +canonical or experimental laboratory must declare its inputs and contracts before +an adapter can run it. The boundary adds telemetry and one immutable receipt; it +does not duplicate the laboratory algorithm. +""" + +from __future__ import annotations + +import hashlib +import json +import os +import re +import time +import uuid +from collections.abc import Callable, Mapping +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Final, Literal, cast + +from k1link.compute.pipeline_telemetry import ( + PipelineTelemetryEmitter, + PipelineTelemetryIdentity, + PipelineTelemetrySink, +) +from k1link.laboratory.evidence_registry import ( + LaboratoryEvidenceRegistry, +) +from k1link.laboratory.evidence_report import verify_laboratory_evidence_result + +LABORATORY_EXECUTION_REGISTRY_SCHEMA: Final = ( + "missioncore.laboratory-execution-registry/v1" +) +LABORATORY_RUN_RECEIPT_SCHEMA: Final = "missioncore.laboratory-run-receipt/v1" +LABORATORY_RUN_CONTRACT: Final = "missioncore.laboratory-run/v1" + +_MAX_REGISTRY_BYTES: Final = 128 * 1024 +_IDENTIFIER: Final = re.compile(r"^[a-z][a-z0-9-]{2,95}$") +_INPUT_ROLE: Final = re.compile(r"^[a-z][a-z0-9_]{2,63}$") +_RUN_IDENTIFIER: Final = re.compile(r"^[a-z0-9][a-z0-9._-]{2,95}$") +_ADAPTER_IDENTIFIER: Final = re.compile( + r"^(?:canonical|experimental)\.[a-z][a-z0-9-]{2,95}/v[1-9][0-9]*$" +) +_CONTRACT: Final = re.compile(r"^missioncore\.[a-z0-9.-]+/v[1-9][0-9]*$") +_DEFINITION_KEYS: Final = frozenset( + {"work_id", "lifecycle", "isolation", "adapter_id", "input_roles", "contracts"} +) +_CONTRACT_KEYS: Final = frozenset({"source", "provider", "graph", "run", "evidence"}) +_AUTHORITY: Final = { + "commands_enabled": False, + "navigation_or_safety_accepted": False, +} + +LaboratoryLifecycle = Literal["canonical", "experimental"] + + +class LaboratoryExecutionError(RuntimeError): + """The execution registry, admission request, or immutable receipt is invalid.""" + + +@dataclass(frozen=True, slots=True) +class LaboratoryExecutionDefinition: + work_id: str + lifecycle: LaboratoryLifecycle + isolation: str + adapter_id: str + input_roles: tuple[str, ...] + source_contract: str + provider_contract: str + graph_contract: str + run_contract: str + evidence_contract: str + + def __post_init__(self) -> None: + _identifier(self.work_id, "work_id") + if self.lifecycle not in {"canonical", "experimental"}: + raise LaboratoryExecutionError("laboratory lifecycle is invalid") + expected_isolation = ( + "core-adapter" if self.lifecycle == "canonical" else "bounded-adapter" + ) + if self.isolation != expected_isolation: + raise LaboratoryExecutionError( + f"{self.lifecycle} laboratory isolation must be {expected_isolation}" + ) + if _ADAPTER_IDENTIFIER.fullmatch(self.adapter_id) is None or not self.adapter_id.startswith( + f"{self.lifecycle}." + ): + raise LaboratoryExecutionError("laboratory adapter_id is invalid") + if not self.input_roles or len(self.input_roles) != len(set(self.input_roles)): + raise LaboratoryExecutionError("laboratory input_roles must be unique and non-empty") + for role in self.input_roles: + if _INPUT_ROLE.fullmatch(role) is None: + raise LaboratoryExecutionError("input role is invalid") + for value in ( + self.source_contract, + self.provider_contract, + self.graph_contract, + self.run_contract, + self.evidence_contract, + ): + if _CONTRACT.fullmatch(value) is None: + raise LaboratoryExecutionError("laboratory contract identity is invalid") + if self.run_contract != LABORATORY_RUN_CONTRACT: + raise LaboratoryExecutionError("laboratory run contract is unsupported") + + +@dataclass(frozen=True, slots=True) +class LaboratoryExecutionRegistry: + definitions: tuple[LaboratoryExecutionDefinition, ...] + legacy_work_ids: tuple[str, ...] + + def __post_init__(self) -> None: + work_ids = [definition.work_id for definition in self.definitions] + if len(work_ids) != len(set(work_ids)): + raise LaboratoryExecutionError("duplicate executable laboratory work_id") + if len(self.legacy_work_ids) != len(set(self.legacy_work_ids)): + raise LaboratoryExecutionError("duplicate legacy laboratory work_id") + for work_id in self.legacy_work_ids: + _identifier(work_id, "legacy work_id") + if set(work_ids).intersection(self.legacy_work_ids): + raise LaboratoryExecutionError("a laboratory cannot be executable and legacy") + + @classmethod + def from_file( + cls, + path: Path, + evidence_registry: LaboratoryEvidenceRegistry, + ) -> LaboratoryExecutionRegistry: + candidate = path.expanduser().absolute() + if candidate.is_symlink() or not candidate.is_file(): + raise LaboratoryExecutionError("laboratory execution registry must be a regular file") + if candidate.stat().st_size > _MAX_REGISTRY_BYTES: + raise LaboratoryExecutionError("laboratory execution registry is too large") + try: + payload: object = json.loads(candidate.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as exc: + raise LaboratoryExecutionError("laboratory execution registry is unreadable") from exc + document = _object(payload, "laboratory execution registry") + _exact_keys( + document, + frozenset({"schema_version", "definitions", "legacy_work_ids"}), + "laboratory execution registry", + ) + if document["schema_version"] != LABORATORY_EXECUTION_REGISTRY_SCHEMA: + raise LaboratoryExecutionError("laboratory execution registry schema is invalid") + rows = document["definitions"] + legacy_rows = document["legacy_work_ids"] + if not isinstance(rows, list) or not isinstance(legacy_rows, list): + raise LaboratoryExecutionError("laboratory registry rows must be arrays") + definitions = tuple(_definition(row) for row in rows) + if not all(isinstance(item, str) for item in legacy_rows): + raise LaboratoryExecutionError("legacy work IDs must be strings") + registry = cls(definitions=definitions, legacy_work_ids=tuple(legacy_rows)) + registry.validate_evidence_registry(evidence_registry) + return registry + + def validate_evidence_registry(self, evidence_registry: LaboratoryEvidenceRegistry) -> None: + evidence_by_work_id = { + definition.work_id: definition for definition in evidence_registry.definitions + } + classified = {definition.work_id for definition in self.definitions}.union( + self.legacy_work_ids + ) + if classified != set(evidence_by_work_id): + missing = sorted(set(evidence_by_work_id) - classified) + unknown = sorted(classified - set(evidence_by_work_id)) + raise LaboratoryExecutionError( + f"laboratory classification is incomplete; missing={missing}, unknown={unknown}" + ) + for definition in self.definitions: + if ( + evidence_by_work_id[definition.work_id].result_schema_version + != definition.evidence_contract + ): + raise LaboratoryExecutionError( + f"laboratory evidence contract mismatch: {definition.work_id}" + ) + + def executable(self, work_id: str) -> LaboratoryExecutionDefinition: + for definition in self.definitions: + if definition.work_id == work_id: + return definition + if work_id in self.legacy_work_ids: + raise LaboratoryExecutionError( + f"legacy laboratory is read-only and cannot be executed: {work_id}" + ) + raise LaboratoryExecutionError(f"laboratory work_id is not classified: {work_id}") + + +@dataclass(frozen=True, slots=True) +class LaboratoryRunRequest: + work_id: str + run_id: str + request_id: str + contour_id: str + agent_id: str + node_id: str + source_id: str + source_package_id: str + method_id: str + inputs: Mapping[str, Path] + output_root: Path + receipt_root: Path + + +@dataclass(frozen=True, slots=True) +class LaboratoryAdapterResult: + result_root: Path + result_id: str + + +@dataclass(frozen=True, slots=True) +class LaboratoryRunResult: + result_root: Path + result_id: str + receipt_root: Path + receipt_id: str + receipt: dict[str, Any] + + +LaboratoryAdapter = Callable[[LaboratoryRunRequest], LaboratoryAdapterResult] + + +class LaboratoryRunner: + """Execute an admitted adapter and publish one common proof and telemetry shape.""" + + def __init__( + self, + *, + registry: LaboratoryExecutionRegistry, + evidence_registry: LaboratoryEvidenceRegistry, + sink: PipelineTelemetrySink, + adapters: Mapping[str, LaboratoryAdapter] | None = None, + clock_ns: Callable[[], int] = time.monotonic_ns, + ) -> None: + self._registry = registry + self._evidence = { + definition.work_id: definition for definition in evidence_registry.definitions + } + self._sink = sink + self._adapters = dict(adapters or canonical_laboratory_adapters()) + self._clock_ns = clock_ns + + def run(self, request: LaboratoryRunRequest) -> LaboratoryRunResult: + definition = self._registry.executable(request.work_id) + _validate_request(request, definition) + adapter = self._adapters.get(definition.adapter_id) + if adapter is None: + raise LaboratoryExecutionError( + f"laboratory adapter is not installed: {definition.adapter_id}" + ) + telemetry = PipelineTelemetryEmitter( + identity=PipelineTelemetryIdentity( + contour_id=request.contour_id, + agent_id=request.agent_id, + node_id=request.node_id, + lab_id=request.work_id, + run_id=request.run_id, + request_id=request.request_id, + source_id=request.source_id, + source_package_id=request.source_package_id, + method_id=request.method_id, + ), + sink=self._sink, + clock_ns=self._clock_ns, + ) + started_ns = self._clock_ns() + telemetry.run("started") + try: + with telemetry.stage("execute-adapter"): + result = adapter(request) + with telemetry.stage("verify-evidence") as outcome: + proof = verify_laboratory_evidence_result( + self._evidence[request.work_id], + result.result_root, + ) + if result.result_id != proof["result_id"]: + raise LaboratoryExecutionError("adapter result_id does not match evidence") + artifact_count = proof["artifact_count"] + if not isinstance(artifact_count, int) or isinstance(artifact_count, bool): + raise LaboratoryExecutionError("evidence artifact count is invalid") + outcome.output_count = artifact_count + with telemetry.stage("publish-receipt"): + receipt_root, receipt_id, receipt = _publish_receipt( + request=request, + definition=definition, + proof=proof, + ) + except BaseException as exc: + telemetry.run( + "failed", + duration_ms=max(0.0, (self._clock_ns() - started_ns) / 1_000_000), + exit_code=1, + error_type=type(exc).__name__, + ) + raise + telemetry.run( + "completed", + duration_ms=max(0.0, (self._clock_ns() - started_ns) / 1_000_000), + exit_code=0, + ) + return LaboratoryRunResult( + result_root=result.result_root, + result_id=result.result_id, + receipt_root=receipt_root, + receipt_id=receipt_id, + receipt=receipt, + ) + + +def canonical_laboratory_adapters() -> dict[str, LaboratoryAdapter]: + return { + "canonical.e33-worker-shadow/v1": _run_e33, + "canonical.e35-degradation-recovery/v1": _run_e35, + "canonical.e46j-raw-fisheye-realtime/v1": _run_e46j, + } + + +def _run_e33(request: LaboratoryRunRequest) -> LaboratoryAdapterResult: + from k1link.compute.e33_worker_shadow import run_e33_worker_shadow + + result = run_e33_worker_shadow( + request.inputs["package_root"], + request.output_root, + execution_id=request.run_id, + ) + return LaboratoryAdapterResult(result_root=result.result_root, result_id=result.result_id) + + +def _run_e35(request: LaboratoryRunRequest) -> LaboratoryAdapterResult: + from k1link.compute.e35_degradation_replay import build_e35_degradation_replay + + result = build_e35_degradation_replay( + e32_result_root=request.inputs["e32_result_root"], + e33_result_root=request.inputs["e33_result_root"], + e34_result_root=request.inputs["e34_result_root"], + e34_profile_path=request.inputs["e34_profile_path"], + profile_path=request.inputs["profile_path"], + output_root=request.output_root, + ) + return LaboratoryAdapterResult(result_root=result.result_root, result_id=result.result_id) + + +def _run_e46j(request: LaboratoryRunRequest) -> LaboratoryAdapterResult: + from k1link.compute.e46j_raw_fisheye_realtime import ( + build_e46j_raw_fisheye_realtime, + ) + + result = build_e46j_raw_fisheye_realtime( + raw_root=request.inputs["raw_root"], + profile_path=request.inputs["profile_path"], + output_root=request.output_root, + ) + return LaboratoryAdapterResult( + result_root=Path(result["result_root"]), + result_id=str(result["result_id"]), + ) + + +def _validate_request( + request: LaboratoryRunRequest, + definition: LaboratoryExecutionDefinition, +) -> None: + if _RUN_IDENTIFIER.fullmatch(request.run_id) is None: + raise LaboratoryExecutionError("laboratory run_id is invalid") + for value, label in ( + (request.request_id, "request_id"), + (request.node_id, "node_id"), + (request.source_id, "source_id"), + (request.source_package_id, "source_package_id"), + (request.method_id, "method_id"), + ): + _bounded_text(value, label) + actual_roles = set(request.inputs) + expected_roles = set(definition.input_roles) + if actual_roles != expected_roles: + raise LaboratoryExecutionError( + "laboratory input roles are invalid; " + f"missing={sorted(expected_roles - actual_roles)}, " + f"unexpected={sorted(actual_roles - expected_roles)}" + ) + for role, path in request.inputs.items(): + if not isinstance(path, Path): + raise LaboratoryExecutionError(f"laboratory input must be a Path: {role}") + candidate = path.expanduser().absolute() + if candidate.is_symlink(): + raise LaboratoryExecutionError(f"laboratory input must not be a symlink: {role}") + try: + candidate.resolve(strict=True) + except OSError as exc: + raise LaboratoryExecutionError(f"laboratory input is unavailable: {role}") from exc + for path, label in ( + (request.output_root, "output_root"), + (request.receipt_root, "receipt_root"), + ): + if not isinstance(path, Path): + raise LaboratoryExecutionError(f"{label} must be a Path") + candidate = path.expanduser().absolute() + if candidate.is_symlink(): + raise LaboratoryExecutionError(f"{label} must not be a symlink") + + +def _publish_receipt( + *, + request: LaboratoryRunRequest, + definition: LaboratoryExecutionDefinition, + proof: dict[str, object], +) -> tuple[Path, str, dict[str, Any]]: + identity = { + "schema_version": LABORATORY_RUN_RECEIPT_SCHEMA, + "work_id": request.work_id, + "lifecycle": definition.lifecycle, + "adapter_id": definition.adapter_id, + "run_id": request.run_id, + "request_id": request.request_id, + "contour_id": request.contour_id, + "agent_id": request.agent_id, + "node_id": request.node_id, + "source_id": request.source_id, + "source_package_id": request.source_package_id, + "method_id": request.method_id, + "contracts": { + "source": definition.source_contract, + "provider": definition.provider_contract, + "graph": definition.graph_contract, + "run": definition.run_contract, + "evidence": definition.evidence_contract, + }, + "result_id": proof["result_id"], + "result_identity_sha256": proof["identity_sha256"], + "evidence_document_sha256": proof["document_sha256"], + "artifact_count": proof["artifact_count"], + "authority": _AUTHORITY, + } + identity_sha256 = hashlib.sha256(_canonical_json(identity)).hexdigest() + receipt_id = f"laboratory-run-receipt-{identity_sha256}" + receipt = { + **identity, + "receipt_id": receipt_id, + "identity_sha256": identity_sha256, + } + destination = request.receipt_root.expanduser().absolute() / receipt_id + document_path = destination / "receipt.json" + encoded = _canonical_json(receipt) + b"\n" + if destination.exists(): + if ( + destination.is_symlink() + or document_path.is_symlink() + or not document_path.is_file() + ): + raise LaboratoryExecutionError("existing laboratory receipt is invalid") + if document_path.read_bytes() != encoded: + raise LaboratoryExecutionError("immutable laboratory receipt changed") + return destination, receipt_id, receipt + destination.parent.mkdir(mode=0o700, parents=True, exist_ok=True) + staging = destination.parent / f".{receipt_id}.{uuid.uuid4().hex}.tmp" + staging.mkdir(mode=0o700, exist_ok=False) + try: + descriptor = os.open(staging / "receipt.json", os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600) + try: + os.write(descriptor, encoded) + os.fsync(descriptor) + finally: + os.close(descriptor) + os.replace(staging, destination) + except BaseException: + if staging.exists(): + for child in staging.iterdir(): + child.unlink() + staging.rmdir() + raise + return destination, receipt_id, receipt + + +def _definition(value: object) -> LaboratoryExecutionDefinition: + row = _object(value, "laboratory execution definition") + _exact_keys(row, _DEFINITION_KEYS, "laboratory execution definition") + contracts = _object(row["contracts"], "laboratory contracts") + _exact_keys(contracts, _CONTRACT_KEYS, "laboratory contracts") + input_roles = row["input_roles"] + if not isinstance(input_roles, list) or not all( + isinstance(item, str) for item in input_roles + ): + raise LaboratoryExecutionError("laboratory input_roles must be strings") + lifecycle = row["lifecycle"] + if lifecycle not in {"canonical", "experimental"}: + raise LaboratoryExecutionError("laboratory lifecycle is invalid") + return LaboratoryExecutionDefinition( + work_id=_text(row["work_id"], "work_id"), + lifecycle=cast(LaboratoryLifecycle, lifecycle), + isolation=_text(row["isolation"], "isolation"), + adapter_id=_text(row["adapter_id"], "adapter_id"), + input_roles=tuple(input_roles), + source_contract=_text(contracts["source"], "source contract"), + provider_contract=_text(contracts["provider"], "provider contract"), + graph_contract=_text(contracts["graph"], "graph contract"), + run_contract=_text(contracts["run"], "run contract"), + evidence_contract=_text(contracts["evidence"], "evidence contract"), + ) + + +def _object(value: object, label: str) -> dict[str, object]: + if not isinstance(value, dict) or not all(isinstance(key, str) for key in value): + raise LaboratoryExecutionError(f"{label} must be an object") + return value + + +def _exact_keys(document: dict[str, object], expected: frozenset[str], label: str) -> None: + actual = frozenset(document) + if actual != expected: + raise LaboratoryExecutionError( + f"{label} keys are invalid; missing={sorted(expected - actual)}, " + f"unexpected={sorted(actual - expected)}" + ) + + +def _text(value: object, label: str) -> str: + if not isinstance(value, str) or not value or value != value.strip(): + raise LaboratoryExecutionError(f"{label} must be a non-empty trimmed string") + return value + + +def _identifier(value: object, label: str) -> str: + text = _text(value, label) + if _IDENTIFIER.fullmatch(text) is None: + raise LaboratoryExecutionError(f"{label} is invalid") + return text + + +def _bounded_text(value: object, label: str) -> str: + text = _text(value, label) + if len(text) > 256 or any(ord(character) < 32 for character in text): + raise LaboratoryExecutionError(f"{label} is invalid") + return text + + +def _canonical_json(value: object) -> bytes: + return json.dumps( + value, + sort_keys=True, + separators=(",", ":"), + ensure_ascii=False, + ).encode("utf-8") diff --git a/src/k1link/web/advanced_laboratory_api.py b/src/k1link/web/advanced_laboratory_api.py index 921b4bd..249e3e4 100644 --- a/src/k1link/web/advanced_laboratory_api.py +++ b/src/k1link/web/advanced_laboratory_api.py @@ -910,7 +910,7 @@ def build_advanced_laboratory_router( if not isinstance(items, list): raise RuntimeError("advanced LAB index items are invalid") l3_identity = latest_l3_visual_identity(l3_visual_root_provider) - if l3_identity is not None: + if evidence_registry is None and l3_identity is not None: items.append( { "work_id": "l3-pointpillars-visual-audit", @@ -919,7 +919,7 @@ def build_advanced_laboratory_router( } ) l31_identity = latest_l31_identity(l31_ravnoves_root_provider) - if l31_identity is not None: + if evidence_registry is None and l31_identity is not None: items.append( { "work_id": "l31-pointpillars-ravnoves", @@ -928,7 +928,7 @@ def build_advanced_laboratory_router( } ) l32_identity = latest_l32_identity(l32_camera_review_root_provider) - if l32_identity is not None: + if evidence_registry is None and l32_identity is not None: items.append( { "work_id": "l32-pointpillars-camera-review", @@ -937,7 +937,7 @@ def build_advanced_laboratory_router( } ) l33_identity = latest_l33_identity(l33_camera_first_review_root_provider) - if l33_identity is not None: + if evidence_registry is None and l33_identity is not None: items.append( { "work_id": "l33-camera-first-detector-review", diff --git a/src/k1link/web/app.py b/src/k1link/web/app.py index 9969e8d..b58f2c5 100644 --- a/src/k1link/web/app.py +++ b/src/k1link/web/app.py @@ -26,6 +26,7 @@ from k1link.compute import ( from k1link.laboratory import ( LaboratoryEvidenceRegistry, LaboratoryEvidenceReportService, + LaboratoryExecutionRegistry, LaboratoryValueReviewRegistry, ) from k1link.sessions import ( @@ -142,6 +143,10 @@ INVALID_REQUEST_DETAIL = "Некорректные параметры запро LABORATORY_EVIDENCE_REGISTRY = LaboratoryEvidenceRegistry.from_directory( REPOSITORY_ROOT / "config" / "laboratories" ) +LABORATORY_EXECUTION_REGISTRY = LaboratoryExecutionRegistry.from_file( + REPOSITORY_ROOT / "config" / "laboratory-execution.json", + LABORATORY_EVIDENCE_REGISTRY, +) LABORATORY_VALUE_REVIEW_REGISTRY = LaboratoryValueReviewRegistry.from_file( REPOSITORY_ROOT / "config" / "laboratory-value-review.json" ) diff --git a/tests/test_laboratory_evidence_registry.py b/tests/test_laboratory_evidence_registry.py index abc0b54..5b1a06d 100644 --- a/tests/test_laboratory_evidence_registry.py +++ b/tests/test_laboratory_evidence_registry.py @@ -127,9 +127,13 @@ def test_product_registry_declares_every_advanced_evidence_source() -> None: repository_root / "config" / "laboratories" ) - assert len(registry.definitions) == 27 + assert len(registry.definitions) == 31 assert {item.work_id for item in registry.definitions} >= { "e31-source-binding", "e46j-raw-fisheye-realtime", + "l3-pointpillars-visual-audit", + "l31-pointpillars-ravnoves", + "l32-pointpillars-camera-review", + "l33-camera-first-detector-review", "l34f-adjudicated-reference", } diff --git a/tests/test_laboratory_execution.py b/tests/test_laboratory_execution.py new file mode 100644 index 0000000..e90684f --- /dev/null +++ b/tests/test_laboratory_execution.py @@ -0,0 +1,193 @@ +from __future__ import annotations + +import hashlib +import json +from dataclasses import replace +from pathlib import Path + +import pytest + +from k1link.compute.pipeline_telemetry import JsonlPipelineTelemetrySink +from k1link.laboratory import ( + LaboratoryAdapterResult, + LaboratoryEvidenceRegistry, + LaboratoryExecutionDefinition, + LaboratoryExecutionError, + LaboratoryExecutionRegistry, + LaboratoryRunner, + LaboratoryRunRequest, +) + +REPOSITORY_ROOT = Path(__file__).resolve().parents[1] + + +def _registries() -> tuple[LaboratoryEvidenceRegistry, LaboratoryExecutionRegistry]: + evidence = LaboratoryEvidenceRegistry.from_directory( + REPOSITORY_ROOT / "config" / "laboratories" + ) + execution = LaboratoryExecutionRegistry.from_file( + REPOSITORY_ROOT / "config" / "laboratory-execution.json", + evidence, + ) + return evidence, execution + + +def _evidence_result(root: Path, *, work_id: str) -> LaboratoryAdapterResult: + evidence, _ = _registries() + definition = next(row for row in evidence.definitions if row.work_id == work_id) + artifact = b"canonical evidence\n" + identity = { + "schema_version": definition.result_schema_version, + "source": {"id": "fixture-source"}, + "authority": { + "commands_enabled": False, + "navigation_or_safety_accepted": False, + }, + } + identity_sha256 = hashlib.sha256(_canonical_json(identity)).hexdigest() + result_id = f"{definition.result_id_prefix}-{identity_sha256}" + result_root = root / result_id + result_root.mkdir(parents=True, exist_ok=True) + (result_root / "artifact.txt").write_bytes(artifact) + document = { + "schema_version": definition.result_schema_version, + "result_id": result_id, + "identity_sha256": identity_sha256, + "identity": identity, + "artifacts": [ + { + "role": "fixture", + "path": "artifact.txt", + "byte_length": len(artifact), + "sha256": hashlib.sha256(artifact).hexdigest(), + } + ], + } + (result_root / definition.document_name).write_bytes(_canonical_json(document) + b"\n") + return LaboratoryAdapterResult(result_root=result_root, result_id=result_id) + + +def _request(tmp_path: Path) -> LaboratoryRunRequest: + package = tmp_path / "input-package" + package.mkdir() + return LaboratoryRunRequest( + work_id="e33-worker-shadow", + run_id="run-001", + request_id="request-001", + contour_id="worker-006", + agent_id="worker-006", + node_id="DESKTOP-OPJ8J04", + source_id="e32-fixture", + source_package_id="e32-package-fixture", + method_id="e33-worker-shadow/v1", + inputs={"package_root": package}, + output_root=tmp_path / "results", + receipt_root=tmp_path / "receipts", + ) + + +def test_repository_registry_classifies_every_evidence_definition() -> None: + evidence, execution = _registries() + + assert {row.work_id for row in execution.definitions} == { + "e33-worker-shadow", + "e35-degradation-recovery", + "e46j-raw-fisheye-realtime", + } + assert all(row.lifecycle == "canonical" for row in execution.definitions) + assert len(execution.definitions) + len(execution.legacy_work_ids) == len( + evidence.definitions + ) + + +def test_legacy_is_read_only_and_experimental_requires_bounded_adapter() -> None: + _, execution = _registries() + + with pytest.raises(LaboratoryExecutionError, match="read-only"): + execution.executable("e31-source-binding") + + with pytest.raises(LaboratoryExecutionError, match="bounded-adapter"): + LaboratoryExecutionDefinition( + work_id="new-experiment", + lifecycle="experimental", + isolation="core-adapter", + adapter_id="experimental.new-experiment/v1", + input_roles=("source_root",), + source_contract="missioncore.experimental-source/v1", + provider_contract="missioncore.experimental-provider/v1", + graph_contract="missioncore.experimental-graph/v1", + run_contract="missioncore.laboratory-run/v1", + evidence_contract="missioncore.experimental-result/v1", + ) + + +def test_runner_uses_common_telemetry_evidence_and_immutable_receipt( + tmp_path: Path, +) -> None: + evidence, execution = _registries() + request = _request(tmp_path) + + def adapter(run_request: LaboratoryRunRequest) -> LaboratoryAdapterResult: + assert run_request == request + return _evidence_result(run_request.output_root, work_id=run_request.work_id) + + telemetry_path = tmp_path / "pipeline.jsonl" + runner = LaboratoryRunner( + registry=execution, + evidence_registry=evidence, + sink=JsonlPipelineTelemetrySink(telemetry_path), + adapters={"canonical.e33-worker-shadow/v1": adapter}, + ) + + first = runner.run(request) + second = runner.run(request) + + assert first.receipt_id == second.receipt_id + assert first.receipt_root == second.receipt_root + assert first.receipt["result_id"] == first.result_id + assert first.receipt["authority"]["commands_enabled"] is False + receipt = json.loads((first.receipt_root / "receipt.json").read_text()) + assert receipt == first.receipt + records = [json.loads(line) for line in telemetry_path.read_text().splitlines()] + states = [row["payload"]["payload"]["event"]["state"] for row in records] + assert states.count("started") == 8 + assert states.count("completed") == 8 + assert all(row["payload"]["lab_id"] == request.work_id for row in records) + + +def test_runner_rejects_undeclared_input_before_adapter(tmp_path: Path) -> None: + evidence, execution = _registries() + request = _request(tmp_path) + request = replace( + request, + inputs={ + **request.inputs, + "hidden_profile": request.inputs["package_root"], + }, + ) + called = False + + def adapter(_: LaboratoryRunRequest) -> LaboratoryAdapterResult: + nonlocal called + called = True + raise AssertionError("must not run") + + runner = LaboratoryRunner( + registry=execution, + evidence_registry=evidence, + sink=JsonlPipelineTelemetrySink(tmp_path / "pipeline.jsonl"), + adapters={"canonical.e33-worker-shadow/v1": adapter}, + ) + + with pytest.raises(LaboratoryExecutionError, match="unexpected=.*hidden_profile"): + runner.run(request) + assert called is False + + +def _canonical_json(value: object) -> bytes: + return json.dumps( + value, + sort_keys=True, + separators=(",", ":"), + ensure_ascii=False, + ).encode("utf-8") diff --git a/tests/test_pipeline_telemetry.py b/tests/test_pipeline_telemetry.py index 1963dfc..c2b9e84 100644 --- a/tests/test_pipeline_telemetry.py +++ b/tests/test_pipeline_telemetry.py @@ -1,5 +1,6 @@ from __future__ import annotations +import hashlib import importlib.util import json from pathlib import Path @@ -8,6 +9,7 @@ from types import ModuleType, SimpleNamespace import pytest from k1link.compute.pipeline_telemetry import ( + MAX_PAYLOAD_BYTES, JsonlPipelineTelemetrySink, MqttPipelineTelemetrySink, PipelineTelemetryEmitter, @@ -231,6 +233,40 @@ def test_jsonl_sink_records_topic_bound_documents(tmp_path: Path) -> None: assert path.stat().st_mode & 0o077 == 0 +def test_jsonl_sink_rotates_to_content_addressed_segment(tmp_path: Path) -> None: + path = tmp_path / "pipeline-telemetry.jsonl" + maximum = MAX_PAYLOAD_BYTES + 4096 + existing = (b"{}\n" * (maximum // 3))[: maximum - 32] + path.write_bytes(existing) + sink = JsonlPipelineTelemetrySink(path, max_bytes=maximum, max_segments=2) + + sink.publish(_identity().topic, b"{}") + + segments = tuple(tmp_path.glob("pipeline-telemetry.*.jsonl")) + assert len(segments) == 1 + assert segments[0].read_bytes() == existing + assert segments[0].stem.split(".")[-1] == hashlib.sha256(existing).hexdigest() + assert json.loads(path.read_text())["payload"] == {} + + +def test_jsonl_sink_refuses_to_delete_unacknowledged_segment_at_bound( + tmp_path: Path, +) -> None: + path = tmp_path / "pipeline-telemetry.jsonl" + maximum = MAX_PAYLOAD_BYTES + 4096 + active = b"x" * maximum + path.write_bytes(active) + retained = tmp_path / f"pipeline-telemetry.{'a' * 64}.jsonl" + retained.write_bytes(b"retained") + sink = JsonlPipelineTelemetrySink(path, max_bytes=maximum, max_segments=1) + + with pytest.raises(PipelineTelemetryError, match="segment bound reached"): + sink.publish(_identity().topic, b"{}") + + assert path.read_bytes() == active + assert retained.read_bytes() == b"retained" + + def test_mqtt_sink_uses_qos_one_without_retention() -> None: calls: list[tuple[str, bytes, int, bool]] = [] diff --git a/tests/test_telemetry_plane_prepare.py b/tests/test_telemetry_plane_prepare.py index e06cadd..da9e4b3 100644 --- a/tests/test_telemetry_plane_prepare.py +++ b/tests/test_telemetry_plane_prepare.py @@ -1,7 +1,10 @@ from __future__ import annotations import importlib.util +import json import stat +import sys +import zipfile from pathlib import Path from types import ModuleType @@ -22,6 +25,7 @@ def _prepare_module() -> ModuleType: assert specification is not None assert specification.loader is not None module = importlib.util.module_from_spec(specification) + sys.modules[specification.name] = module specification.loader.exec_module(module) return module @@ -29,74 +33,122 @@ def _prepare_module() -> ModuleType: prepare = _prepare_module() -def test_initialize_environment_generates_private_unique_secrets( +def _private_paths(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(prepare, "ENV_PATH", tmp_path / ".env") + monkeypatch.setattr(prepare, "RUNTIME_ROOT", tmp_path / "runtime") + monkeypatch.setattr(prepare, "RUNTIME", tmp_path / "runtime" / "mosquitto") + monkeypatch.setattr( + prepare, + "AGENT_REGISTRY_PATH", + tmp_path / "runtime" / "agents.json", + ) + + +def _write_environment(path: Path) -> None: + path.write_text( + "MISSIONCORE_MQTT_BIND_ADDRESS=192.168.68.52\n" + "MISSIONCORE_MQTT_PORT=1883\n" + "MISSIONCORE_DB_PASSWORD=db-secret\n" + "MISSIONCORE_DB_INGEST_PASSWORD=db-ingest-secret\n" + "MISSIONCORE_MQTT_INGEST_USER=missioncore-ingest\n" + "MISSIONCORE_MQTT_INGEST_PASSWORD=mqtt-ingest-secret\n", + encoding="utf-8", + ) + + +def test_initialize_environment_generates_only_plane_secrets( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: - env_path = tmp_path / ".env" - monkeypatch.setattr(prepare, "ENV_PATH", env_path) + _private_paths(tmp_path, monkeypatch) prepare._initialize_environment("192.0.2.15") values = dict( line.split("=", 1) - for line in env_path.read_text(encoding="utf-8").splitlines() + for line in prepare.ENV_PATH.read_text(encoding="utf-8").splitlines() ) assert values["MISSIONCORE_MQTT_BIND_ADDRESS"] == "192.0.2.15" - secrets = { + generated = { values["MISSIONCORE_DB_PASSWORD"], values["MISSIONCORE_DB_INGEST_PASSWORD"], values["MISSIONCORE_MQTT_INGEST_PASSWORD"], - values["MISSIONCORE_MQTT_WORKER_006_PASSWORD"], } - assert len(secrets) == 4 - assert all(len(secret) >= 40 for secret in secrets) - assert stat.S_IMODE(env_path.stat().st_mode) == 0o600 + assert len(generated) == 3 + assert all(len(secret) >= 40 for secret in generated) + assert not any("WORKER_006" in name for name in values) + assert stat.S_IMODE(prepare.ENV_PATH.stat().st_mode) == 0o600 def test_initialize_environment_refuses_to_replace_credentials( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: - env_path = tmp_path / ".env" - env_path.write_text("existing=true\n", encoding="utf-8") - monkeypatch.setattr(prepare, "ENV_PATH", env_path) + _private_paths(tmp_path, monkeypatch) + prepare.ENV_PATH.write_text("existing=true\n", encoding="utf-8") with pytest.raises(RuntimeError, match="refusing to overwrite"): prepare._initialize_environment("127.0.0.1") - assert env_path.read_text(encoding="utf-8") == "existing=true\n" + assert prepare.ENV_PATH.read_text(encoding="utf-8") == "existing=true\n" -def test_environment_migration_adds_only_new_private_values( +def test_migration_moves_legacy_worker_to_private_registry( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: - env_path = tmp_path / ".env" - env_path.write_text( + _private_paths(tmp_path, monkeypatch) + prepare.ENV_PATH.write_text( "MISSIONCORE_DB_PASSWORD=keep-me\n" - "MISSIONCORE_MQTT_WORKER_006_USER=worker-006\n", + "MISSIONCORE_MQTT_WORKER_006_USER=worker-006\n" + "MISSIONCORE_MQTT_WORKER_006_CONTOUR=worker-006\n" + f"MISSIONCORE_MQTT_WORKER_006_PASSWORD={'l' * 40}\n", encoding="utf-8", ) - monkeypatch.setattr(prepare, "ENV_PATH", env_path) prepare._migrate_environment() - first = env_path.read_text(encoding="utf-8") + first_env = prepare.ENV_PATH.read_text(encoding="utf-8") + first_registry = prepare.AGENT_REGISTRY_PATH.read_bytes() prepare._migrate_environment() - assert "MISSIONCORE_DB_PASSWORD=keep-me" in first - assert "MISSIONCORE_DB_INGEST_PASSWORD=" in first - assert "MISSIONCORE_MQTT_WORKER_006_CONTOUR=worker-006" in first - assert env_path.read_text(encoding="utf-8") == first - assert stat.S_IMODE(env_path.stat().st_mode) == 0o600 + assert "MISSIONCORE_DB_PASSWORD=keep-me" in first_env + assert "MISSIONCORE_DB_INGEST_PASSWORD=" in first_env + assert prepare.ENV_PATH.read_text(encoding="utf-8") == first_env + assert prepare.AGENT_REGISTRY_PATH.read_bytes() == first_registry + assert prepare._read_agent_registry() == ( + prepare.AgentCredential("worker-006", "worker-006", "l" * 40), + ) + assert stat.S_IMODE(prepare.AGENT_REGISTRY_PATH.stat().st_mode) == 0o600 -def test_existing_password_file_is_updated_without_recreation( +def test_enrollment_is_generic_private_and_rejects_duplicate_agent_id( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: - password_path = tmp_path / "passwords" - password_path.write_text("existing", encoding="utf-8") + _private_paths(tmp_path, monkeypatch) + + first = prepare._enroll_agent("compute-east", "worker-006") + second = prepare._enroll_agent("compute-west", "worker-007") + + assert first.password != second.password + assert prepare._read_agent_registry() == (first, second) + assert stat.S_IMODE(prepare.AGENT_REGISTRY_PATH.stat().st_mode) == 0o600 + with pytest.raises(RuntimeError, match="already enrolled"): + prepare._enroll_agent("another-contour", "worker-006") + + +def test_password_file_is_rebuilt_and_acl_is_exactly_scoped( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + _private_paths(tmp_path, monkeypatch) + prepare.RUNTIME.mkdir(parents=True) + password_path = prepare.RUNTIME / "passwords" + password_path.write_text("stale-user", encoding="utf-8") + agents = ( + prepare.AgentCredential("compute-west", "worker-007", "w" * 40), + prepare.AgentCredential("compute-east", "worker-006", "e" * 40), + ) calls: list[tuple[str, bool]] = [] def capture( @@ -106,9 +158,9 @@ def test_existing_password_file_is_updated_without_recreation( *, create: bool, ) -> None: - assert path == password_path assert password calls.append((username, create)) + path.write_text("new-password-file", encoding="utf-8") monkeypatch.setattr(prepare, "_password_entry", capture) @@ -116,11 +168,70 @@ def test_existing_password_file_is_updated_without_recreation( password_path, "missioncore-ingest", "ingest-secret", - "worker-006", - "worker-secret", + agents, ) assert calls == [ - ("missioncore-ingest", False), + ("missioncore-ingest", True), ("worker-006", False), + ("worker-007", False), ] + assert password_path.read_text() == "new-password-file" + acl = prepare._acl_document("missioncore-ingest", agents) + assert "topic write mission-core/v1/contours/compute-east/agents/worker-006/+" in acl + assert "topic write mission-core/v1/contours/compute-west/agents/worker-007/+" in acl + assert "topic write mission-core/v1/contours/+/agents/+/+" not in acl + + +def test_payload_export_is_private_separate_and_non_overwriting( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + _private_paths(tmp_path, monkeypatch) + _write_environment(prepare.ENV_PATH) + credential = prepare._enroll_agent("compute-east", "worker-007") + output = tmp_path / "worker-007.private.json" + + prepare._export_agent_payload( + agent_id="worker-007", + node_id="WORKSTATION-007", + output=output, + ) + + payload = json.loads(output.read_text()) + assert payload["schema_version"] == prepare.AGENT_PAYLOAD_SCHEMA + assert payload["MISSIONCORE_CONTOUR_ID"] == "compute-east" + assert payload["MISSIONCORE_AGENT_ID"] == "worker-007" + assert payload["MISSIONCORE_MQTT_PASSWORD"] == credential.password + assert stat.S_IMODE(output.stat().st_mode) == 0o600 + with pytest.raises(RuntimeError, match="refusing to replace"): + prepare._export_agent_payload( + agent_id="worker-007", + node_id="WORKSTATION-007", + output=output, + ) + + +def test_windows_agent_bundle_is_deterministic_and_contains_no_credentials( + tmp_path: Path, +) -> None: + first = tmp_path / "agent-a.zip" + second = tmp_path / "agent-b.zip" + + first_id = prepare._build_agent_bundle(platform_name="windows", output=first) + second_id = prepare._build_agent_bundle(platform_name="windows", output=second) + + assert first_id == second_id + assert first.read_bytes() == second.read_bytes() + with zipfile.ZipFile(first) as archive: + assert set(archive.namelist()) == { + "manifest.json", + *prepare.WINDOWS_BUNDLE_FILES, + } + manifest = json.loads(archive.read("manifest.json")) + assert manifest["bundle_id"] == first_id + assert manifest["credential_embedded"] is False + archive_bytes = b"".join(archive.read(name) for name in archive.namelist()) + assert b"replace-with-a-random-local-secret" not in archive_bytes + assert b"legacy-worker-secret" not in archive_bytes + assert stat.S_IMODE(first.stat().st_mode) == 0o600