diff --git a/apps/control-station/src/core/observatory/laboratorySetups.ts b/apps/control-station/src/core/observatory/laboratorySetups.ts index e7149ca..8c26266 100644 --- a/apps/control-station/src/core/observatory/laboratorySetups.ts +++ b/apps/control-station/src/core/observatory/laboratorySetups.ts @@ -1,16 +1,34 @@ +import { preflightObservatoryPortableLaboratorySetup } from "./portableLaboratorySetups"; + +export { fetchObservatoryPortableLaboratorySetups } from "./portableLaboratorySetups"; + const CATALOG_SCHEMA = "missioncore.observatory-laboratory-setup-catalog/v1"; const PREFLIGHT_REQUEST_SCHEMA = "missioncore.observatory-run-preflight-request/v1"; const PREFLIGHT_SCHEMA = "missioncore.observatory-run-preflight/v1"; const SHA256 = /^[a-f0-9]{64}$/; -export type ObservatoryLaboratorySetupOrigin = "archived-definition" | "existing-result"; -export type ObservatoryLaboratorySetupAction = "open-existing" | "open-legacy" | "blocked"; +export type ObservatoryLaboratorySetupOrigin = + | "archived-definition" + | "existing-result" + | "portable-definition"; +export type ObservatoryLaboratorySetupAction = + | "open-existing" + | "open-legacy" + | "blocked"; export interface ObservatoryLaboratoryRunDefinition { readonly definitionId: string; readonly version: number; - readonly workId: string; + readonly workId: string | null; readonly definitionSha256: string; + readonly resultSchema: string | null; + readonly resultKind: string | null; + readonly models: readonly { + readonly name: string; + readonly releaseId: string; + readonly modelId: string; + readonly architecture: string; + }[]; readonly configuration: readonly { readonly role: string; readonly sha256: string; @@ -38,7 +56,7 @@ export interface ObservatoryLaboratorySetup { }; readonly executor: { readonly contourId: string; - readonly state: "not-installed"; + readonly state: "not-installed" | "ready"; readonly reasonCode: string; readonly reason: string; }; @@ -47,7 +65,7 @@ export interface ObservatoryLaboratorySetup { readonly outcome: "existing" | "blocked"; readonly action: ObservatoryLaboratorySetupAction; readonly reason: string; - readonly submissionAllowed: false; + readonly submissionAllowed: boolean; readonly existingResultIds: readonly string[]; }; } @@ -124,6 +142,9 @@ export async function preflightObservatoryLaboratorySetup( fetcher?: ObservatoryLaboratorySetupFetch; } = {}, ): Promise { + if (setup.origin === "portable-definition") { + return preflightObservatoryPortableLaboratorySetup(sourceSessionId, setup); + } const response = await request( fetcher, "/api/v1/observatory/run-preflights", @@ -245,6 +266,9 @@ function decodeRunDefinition(value: unknown): ObservatoryLaboratoryRunDefinition version: positiveInteger(row.version, "definition version"), workId: text(row.work_id, "work_id"), definitionSha256: digest, + resultSchema: null, + resultKind: null, + models: [], configuration: array(row.configuration, "configuration").map((item) => { const reference = record(item, "configuration reference"); exactKeys(reference, ["role", "sha256"], "configuration reference"); diff --git a/apps/control-station/src/core/observatory/portableLaboratorySetupDecoder.ts b/apps/control-station/src/core/observatory/portableLaboratorySetupDecoder.ts new file mode 100644 index 0000000..1d3b52c --- /dev/null +++ b/apps/control-station/src/core/observatory/portableLaboratorySetupDecoder.ts @@ -0,0 +1,307 @@ +import type { + ObservatoryLaboratoryRunDefinition, + ObservatoryLaboratorySetup, + ObservatoryLaboratorySetupCatalog, +} from "./laboratorySetups"; + +const PORTABLE_CATALOG_SCHEMA = "missioncore.observatory-portable-setup-catalog/v2"; +const SHA256 = /^[a-f0-9]{64}$/; + +export class ObservatoryPortableSetupDecodeError extends Error { + constructor(message: string) { + super(message); + this.name = "ObservatoryPortableSetupDecodeError"; + } +} + +export function decodePortableCatalog(value: unknown): ObservatoryLaboratorySetupCatalog { + const row = record(value, "portable-каталог сетапов"); + exactKeys( + row, + ["authority", "schema_version", "setups", "source_session_id"], + "portable-каталог сетапов", + ); + exact(row.schema_version, PORTABLE_CATALOG_SCHEMA, "schema_version portable-каталога"); + observationAuthority(row.authority); + return { + sourceSessionId: text(row.source_session_id, "portable source_session_id"), + setups: array(row.setups, "portable setups").map(decodePortableSetup), + }; +} + +function decodePortableSetup(value: unknown): ObservatoryLaboratorySetup { + const row = record(value, "portable-сетап"); + exactKeys(row, [ + "authority", "description", "display_name", "executor", "existing_results", + "origin", "preflight", "run_definition", "setup_id", "source_compatibility", + "source_requirements", + ], "portable-сетап"); + exact(row.origin, "portable-definition", "portable origin"); + observationAuthority(row.authority); + decodePortableSourceRequirements(row.source_requirements); + + const compatibility = record(row.source_compatibility, "portable source_compatibility"); + const compatibilityKeys = compatibility.evidence === undefined + ? ["compatible", "outcome", "reason"] + : ["compatible", "evidence", "outcome", "reason"]; + exactKeys(compatibility, compatibilityKeys, "portable source_compatibility"); + const compatible = boolean(compatibility.compatible, "portable compatible"); + exact( + compatibility.outcome, + compatible ? "pass" : "blocked", + "portable compatibility outcome", + ); + const compatibilityReason = text(compatibility.reason, "portable compatibility reason"); + if (compatibility.evidence !== undefined) { + const evidence = record(compatibility.evidence, "portable compatibility evidence"); + exactKeys(evidence, [ + "frame_count", "timeline_end_seconds", "timeline_start_seconds", + ], "portable compatibility evidence"); + positiveInteger(evidence.frame_count, "portable frame_count"); + finiteNumber(evidence.timeline_start_seconds, "portable timeline_start_seconds"); + finiteNumber(evidence.timeline_end_seconds, "portable timeline_end_seconds"); + } + + const executor = record(row.executor, "portable executor"); + exactKeys(executor, ["contour_id", "ready", "reason", "state"], "portable executor"); + const executorState = oneOf( + executor.state, + ["not-installed", "ready"] as const, + "portable executor state", + ); + const executorReady = boolean(executor.ready, "portable executor ready"); + if (executorReady !== (executorState === "ready")) { + throw new ObservatoryPortableSetupDecodeError( + "Portable executor: состояние готовности противоречиво.", + ); + } + const executorReason = executor.reason === null + ? "Исполнитель установлен." + : text(executor.reason, "portable executor reason"); + + const preflight = record(row.preflight, "portable preflight"); + exactKeys(preflight, [ + "action", "existing_result_ids", "outcome", "reason", "submission_allowed", + ], "portable preflight"); + exact(preflight.outcome, "blocked", "portable preflight outcome"); + exact(preflight.action, "blocked", "portable preflight action"); + const submissionAllowed = boolean( + preflight.submission_allowed, + "portable preflight submission_allowed", + ); + if (submissionAllowed) { + throw new ObservatoryPortableSetupDecodeError( + "Portable preflight: постановка в очередь ещё не поддерживается.", + ); + } + const existingResults = array(row.existing_results, "portable existing_results"); + const existingResultIds = array( + preflight.existing_result_ids, + "portable existing_result_ids", + ); + if (existingResults.length > 0 || existingResultIds.length > 0) { + throw new ObservatoryPortableSetupDecodeError( + "Portable result: проверяемая привязка результата к RunDefinition ещё не поддерживается.", + ); + } + + return { + setupId: text(row.setup_id, "portable setup_id"), + displayName: text(row.display_name, "portable display_name"), + description: text(row.description, "portable description"), + origin: "portable-definition", + runDefinition: decodePortableRunDefinition(row.run_definition), + compatibility: { + compatible, + reasons: compatible + ? [] + : [{ code: "source-capability-blocked", message: compatibilityReason }], + }, + executor: { + contourId: text(executor.contour_id, "portable executor contour_id"), + state: executorState, + reasonCode: executorReady + ? "portable-executor-ready" + : "portable-executor-not-installed", + reason: executorReason, + }, + preservedResults: [], + preflight: { + outcome: "blocked", + action: "blocked", + reason: text(preflight.reason, "portable preflight reason"), + submissionAllowed, + existingResultIds: [], + }, + }; +} + +function decodePortableSourceRequirements(value: unknown): void { + const row = record(value, "portable source_requirements"); + exactKeys(row, [ + "archive_id", "calibration_identity_sha256", "calibration_slot", "camera_height", + "camera_semantic_channel_id", "camera_source_id", "camera_width", + "exactly_one_media_epoch", "plugin_id", "recorded_media_init_sha256", + "recorded_media_type", "required_modalities", "seekable", + ], "portable source_requirements"); + for (const key of [ + "archive_id", "calibration_slot", "camera_semantic_channel_id", "camera_source_id", + "plugin_id", "recorded_media_type", + ]) text(row[key], `portable source_requirements.${key}`); + for (const key of ["calibration_identity_sha256", "recorded_media_init_sha256"]) { + const digest = text(row[key], `portable source_requirements.${key}`); + if (!SHA256.test(digest)) { + throw new ObservatoryPortableSetupDecodeError( + `portable source_requirements.${key}: некорректный digest.`, + ); + } + } + positiveInteger(row.camera_width, "portable camera_width"); + positiveInteger(row.camera_height, "portable camera_height"); + exact(row.exactly_one_media_epoch, true, "portable exactly_one_media_epoch"); + exact(row.seekable, true, "portable seekable"); + array(row.required_modalities, "portable required_modalities") + .forEach((item) => text(item, "portable required modality")); +} + +function decodePortableRunDefinition(value: unknown): ObservatoryLaboratoryRunDefinition { + const row = record(value, "portable RunDefinition"); + exactKeys(row, [ + "definition_id", "definition_sha256", "models", "result_kind", "result_schema", + "version", + ], "portable RunDefinition"); + const digest = text(row.definition_sha256, "portable definition_sha256"); + if (!SHA256.test(digest)) { + throw new ObservatoryPortableSetupDecodeError( + "Portable RunDefinition: некорректный digest.", + ); + } + return { + definitionId: text(row.definition_id, "portable definition_id"), + version: positiveInteger(row.version, "portable definition version"), + workId: null, + definitionSha256: digest, + resultSchema: text(row.result_schema, "portable result_schema"), + resultKind: text(row.result_kind, "portable result_kind"), + models: array(row.models, "portable models").map((item) => { + const model = record(item, "portable model"); + exactKeys( + model, + ["architecture", "model_id", "name", "release_id"], + "portable model", + ); + return { + name: text(model.name, "portable model name"), + releaseId: text(model.release_id, "portable model release_id"), + modelId: text(model.model_id, "portable model model_id"), + architecture: text(model.architecture, "portable model architecture"), + }; + }), + configuration: [], + }; +} + +function observationAuthority(value: unknown): void { + const row = record(value, "authority"); + const keys = [ + "commands_enabled", + "actuation_allowed", + "navigation_or_safety_accepted", + "production_accepted", + ]; + exactKeys(row, keys, "authority"); + for (const key of keys) exact(row[key], false, `authority.${key}`); +} + +function record(value: unknown, label: string): Record { + if (!value || typeof value !== "object" || Array.isArray(value)) { + throw new ObservatoryPortableSetupDecodeError( + `${label}: ожидался объект.`, + ); + } + return value as Record; +} + +function array(value: unknown, label: string): unknown[] { + if (!Array.isArray(value)) { + throw new ObservatoryPortableSetupDecodeError( + `${label}: ожидался массив.`, + ); + } + return value; +} + +function text(value: unknown, label: string): string { + if (typeof value !== "string" || !value.trim()) { + throw new ObservatoryPortableSetupDecodeError( + `${label}: ожидался текст.`, + ); + } + return value; +} + +function boolean(value: unknown, label: string): boolean { + if (typeof value !== "boolean") { + throw new ObservatoryPortableSetupDecodeError( + `${label}: ожидался boolean.`, + ); + } + return value; +} + +function positiveInteger(value: unknown, label: string): number { + if (!Number.isInteger(value) || Number(value) < 1) { + throw new ObservatoryPortableSetupDecodeError( + `${label}: ожидалось положительное целое.`, + ); + } + return Number(value); +} + +function finiteNumber(value: unknown, label: string): number { + if (typeof value !== "number" || !Number.isFinite(value)) { + throw new ObservatoryPortableSetupDecodeError( + `${label}: ожидалось конечное число.`, + ); + } + return value; +} + +function exact(value: unknown, expected: T, label: string): T { + if (value !== expected) { + throw new ObservatoryPortableSetupDecodeError( + `${label}: значение изменилось.`, + ); + } + return expected; +} + +function oneOf( + value: unknown, + allowed: T, + label: string, +): T[number] { + if (typeof value !== "string" || !allowed.includes(value)) { + throw new ObservatoryPortableSetupDecodeError( + `${label}: значение не поддерживается.`, + ); + } + return value as T[number]; +} + +function exactKeys( + value: Record, + expected: readonly string[], + label: string, +): void { + const actual = Object.keys(value).sort(); + const sortedExpected = [...expected].sort(); + if ( + actual.length !== sortedExpected.length + || actual.some((key, index) => key !== sortedExpected[index]) + ) { + throw new ObservatoryPortableSetupDecodeError( + `${label}: обнаружены неизвестные поля.`, + ); + } +} diff --git a/apps/control-station/src/core/observatory/portableLaboratorySetups.ts b/apps/control-station/src/core/observatory/portableLaboratorySetups.ts new file mode 100644 index 0000000..c5b14c6 --- /dev/null +++ b/apps/control-station/src/core/observatory/portableLaboratorySetups.ts @@ -0,0 +1,142 @@ +import { decodePortableCatalog } from "./portableLaboratorySetupDecoder"; +import type { + ObservatoryLaboratoryRunPreflight, + ObservatoryLaboratorySetup, + ObservatoryLaboratorySetupCatalog, +} from "./laboratorySetups"; + +export type ObservatoryPortableLaboratorySetupFetch = ( + input: RequestInfo | URL, + init?: RequestInit, +) => Promise; + +class ObservatoryPortableLaboratorySetupContractError extends Error { + readonly status: number | null; + + constructor(message: string, status: number | null = null) { + super(message); + this.name = "ObservatoryPortableLaboratorySetupContractError"; + this.status = status; + } +} + +export async function fetchObservatoryPortableLaboratorySetups( + sourceSessionId: string, + { + signal, + fetcher = globalThis.fetch, + }: { + signal?: AbortSignal; + fetcher?: ObservatoryPortableLaboratorySetupFetch; + } = {}, +): Promise { + const response = await request( + fetcher, + `/api/v1/observatory/portable-laboratory-setups?source_session_id=${encodeURIComponent(sourceSessionId)}`, + { method: "GET", headers: { Accept: "application/json" }, signal }, + ); + const body = await responseBody(response); + if (!response.ok) throw apiError(body, response.status); + const catalog = decodePortableCatalog(body); + if (catalog.sourceSessionId !== sourceSessionId) { + throw new ObservatoryPortableLaboratorySetupContractError( + "Portable-каталог сетапов относится к другой исходной сессии.", + ); + } + return catalog; +} + +export function preflightObservatoryPortableLaboratorySetup( + sourceSessionId: string, + setup: ObservatoryLaboratorySetup, +): ObservatoryLaboratoryRunPreflight { + const definitionSha256 = setup.runDefinition?.definitionSha256 ?? null; + if (setup.origin !== "portable-definition" || definitionSha256 === null) { + throw new ObservatoryPortableLaboratorySetupContractError( + "Portable-сетап не содержит RunDefinition.", + ); + } + if ( + setup.preflight.outcome !== "blocked" + || setup.preflight.action !== "blocked" + || setup.preflight.existingResultIds.length > 0 + || setup.preservedResults.length > 0 + ) { + throw new ObservatoryPortableLaboratorySetupContractError( + "Portable-result ещё не имеет проверяемой привязки к RunDefinition.", + ); + } + return { + sourceSessionId, + setupId: setup.setupId, + definitionSha256, + outcome: "blocked", + submissionAllowed: false, + checks: [ + { + checkId: "source-compatibility", + outcome: setup.compatibility.compatible ? "pass" : "fail", + reasonCode: setup.compatibility.compatible + ? "source-capability-admitted" + : "source-capability-blocked", + message: setup.compatibility.compatible + ? "Запись соответствует portable-профилю LAB V1." + : setup.compatibility.reasons[0]?.message + ?? "Запись не соответствует portable-профилю LAB V1.", + }, + { + checkId: "executor", + outcome: setup.executor.state === "ready" ? "pass" : "fail", + reasonCode: setup.executor.reasonCode, + message: setup.executor.reason, + }, + { + checkId: "durable-queue", + outcome: "fail", + reasonCode: "portable-dispatch-unavailable", + message: setup.preflight.reason, + }, + ], + existingResultIds: [], + }; +} + +async function request( + fetcher: ObservatoryPortableLaboratorySetupFetch, + input: string, + init: RequestInit, +): Promise { + try { + return await fetcher(input, init); + } catch (error) { + if (error instanceof DOMException && error.name === "AbortError") throw error; + throw new ObservatoryPortableLaboratorySetupContractError( + "Portable-каталог сетапов недоступен.", + ); + } +} + +async function responseBody(response: Response): Promise { + const textBody = await response.text(); + if (!textBody) return undefined; + try { + return JSON.parse(textBody) as unknown; + } catch { + return textBody; + } +} + +function apiError( + body: unknown, + status: number, +): ObservatoryPortableLaboratorySetupContractError { + const detail = body && typeof body === "object" && !Array.isArray(body) + ? (body as Record).detail + : null; + return new ObservatoryPortableLaboratorySetupContractError( + typeof detail === "string" && detail.trim() + ? detail + : `Observatory API вернул HTTP ${status}.`, + status, + ); +} diff --git a/apps/control-station/src/core/observatory/useObservatoryLaboratorySetups.ts b/apps/control-station/src/core/observatory/useObservatoryLaboratorySetups.ts index cb7d575..3d08826 100644 --- a/apps/control-station/src/core/observatory/useObservatoryLaboratorySetups.ts +++ b/apps/control-station/src/core/observatory/useObservatoryLaboratorySetups.ts @@ -2,6 +2,7 @@ import { useCallback, useEffect, useMemo, useRef, useState } from "react"; import { fetchObservatoryLaboratorySetups, + fetchObservatoryPortableLaboratorySetups, preflightObservatoryLaboratorySetup, type ObservatoryLaboratoryRunPreflight, type ObservatoryLaboratorySetupCatalog, @@ -41,22 +42,35 @@ export function useObservatoryLaboratorySetups(sourceSessionId: string) { setState((current) => activeCatalog && current !== "idle" ? "refreshing" : "loading"); setError(null); setPreflight({ kind: "idle" }); + const portableResult = fetchObservatoryPortableLaboratorySetups( + sourceSessionId, + { signal: request.signal }, + ).then( + (value) => ({ status: "fulfilled" as const, value }), + (reason: unknown) => ({ status: "rejected" as const, reason }), + ); void fetchObservatoryLaboratorySetups(sourceSessionId, { signal: request.signal }) - .then((next) => { + .then(async (legacyCatalog) => { if (request.signal.aborted || requestSequence.current !== sequence) return; - setCatalog(next); + publishSetupCatalog(legacyCatalog, setCatalog, setSelectedSetupId); setState("ready"); - setSelectedSetupId((current) => { - if (next.setups.some( - (setup) => setup.setupId === current && setup.compatibility.compatible, - )) return current; - return next.setups.find( - (setup) => setup.compatibility.compatible && setup.preflight.outcome === "existing", - )?.setupId - ?? next.setups.find((setup) => setup.compatibility.compatible)?.setupId - ?? next.setups[0]?.setupId - ?? ""; - }); + const optionalPortable = await portableResult; + if (request.signal.aborted || requestSequence.current !== sequence) return; + if (optionalPortable.status === "fulfilled") { + publishSetupCatalog( + mergeSetupCatalogs(legacyCatalog, optionalPortable.value), + setCatalog, + setSelectedSetupId, + ); + setError(null); + return; + } + setError( + optionalPortable.reason instanceof Error + && optionalPortable.reason.message.trim() + ? optionalPortable.reason.message + : "Portable-каталог LAB V1 недоступен.", + ); }) .catch((caught: unknown) => { if (request.signal.aborted || requestSequence.current !== sequence) return; @@ -129,6 +143,39 @@ export function useObservatoryLaboratorySetups(sourceSessionId: string) { }; } +function publishSetupCatalog( + next: ObservatoryLaboratorySetupCatalog, + setCatalog: (catalog: ObservatoryLaboratorySetupCatalog) => void, + setSelectedSetupId: (update: (current: string) => string) => void, +): void { + setCatalog(next); + setSelectedSetupId((current) => { + if (next.setups.some( + (setup) => setup.setupId === current && setup.compatibility.compatible, + )) return current; + return next.setups.find( + (setup) => setup.compatibility.compatible && setup.preflight.outcome === "existing", + )?.setupId + ?? next.setups.find((setup) => setup.compatibility.compatible)?.setupId + ?? next.setups[0]?.setupId + ?? ""; + }); +} + +function mergeSetupCatalogs( + legacy: ObservatoryLaboratorySetupCatalog, + portable: ObservatoryLaboratorySetupCatalog, +): ObservatoryLaboratorySetupCatalog { + if (legacy.sourceSessionId !== portable.sourceSessionId) { + throw new Error("Каталоги сетапов относятся к разным исходным сессиям."); + } + const setups = [...legacy.setups, ...portable.setups]; + if (new Set(setups.map((setup) => setup.setupId)).size !== setups.length) { + throw new Error("Каталоги сетапов содержат повторяющиеся идентификаторы."); + } + return { sourceSessionId: legacy.sourceSessionId, setups }; +} + export type ObservatoryLaboratorySetupsController = ReturnType< typeof useObservatoryLaboratorySetups >; diff --git a/apps/control-station/src/workspaces/observatory/ObservatoryWorkspace.tsx b/apps/control-station/src/workspaces/observatory/ObservatoryWorkspace.tsx index c0634f3..b331da4 100644 --- a/apps/control-station/src/workspaces/observatory/ObservatoryWorkspace.tsx +++ b/apps/control-station/src/workspaces/observatory/ObservatoryWorkspace.tsx @@ -104,7 +104,7 @@ const recordedJobStatus: Record< label: "Ждём подтверждения остановки", tone: "warning", }, - succeeded: { label: "Готово", tone: "success" }, + succeeded: { label: "Вычислено · ждёт публикации", tone: "warning" }, failed: { label: "Ошибка расчёта", tone: "danger" }, "reconciliation-required": { label: "Нужна сверка", tone: "warning" }, }; @@ -210,7 +210,13 @@ export function ObservatoryWorkspace({ value: setup.setupId, label: setup.displayName, description: setup.compatibility.compatible - ? setup.origin === "existing-result" ? "Готовый результат" : "Совместимый архивный сетап" + ? setup.origin === "existing-result" + ? "Готовый результат" + : setup.origin === "portable-definition" + ? setup.executor.state === "ready" + ? "Запись совместима · Worker установлен, запуск закрыт" + : "Запись совместима · Worker не установлен" + : "Совместимый архивный сетап" : "Несовместим с выбранной сессией", })) ?? [] ), [setupController.catalog]); @@ -489,6 +495,20 @@ export function ObservatoryWorkspace({ Проверяем сетап ) : queueStateBusy ? ( Читаем очередь + ) : setupController.selectedSetup?.origin === "portable-definition" + && runPreflight ? ( + + {setupController.selectedSetup.compatibility.compatible + ? setupController.selectedSetup.executor.state === "not-installed" + ? "Worker LAB V1 не установлен" + : "Запуск LAB V1 недоступен" + : "Запись несовместима"} + ) : null} {canSubmitRecordedJob ? (