diff --git a/apps/control-station/src/core/observation/labReplayCapability.ts b/apps/control-station/src/core/observation/labReplayCapability.ts index 3ff4c2c..68b68b8 100644 --- a/apps/control-station/src/core/observation/labReplayCapability.ts +++ b/apps/control-station/src/core/observation/labReplayCapability.ts @@ -1,4 +1,4 @@ -export interface ObservationLabReplayCapability { +export interface ObservationCanonicalLabReplayCapability { schemaVersion: "missioncore.observation-lab-replay-capability/v1"; kind: "canonical-recorded-rerun"; viewerProfile: "recorded-session"; @@ -7,6 +7,19 @@ export interface ObservationLabReplayCapability { commandsEnabled: false; } +export interface ObservationPortableLabReplayCapability { + schemaVersion: "missioncore.observation-lab-replay-capability/v2"; + kind: "portable-result-review"; + viewerProfile: "portable-result"; + timeline: "result-defined"; + activation: "explicit"; + commandsEnabled: false; +} + +export type ObservationLabReplayCapability = + | ObservationCanonicalLabReplayCapability + | ObservationPortableLabReplayCapability; + export class ObservationLabReplayCapabilityContractError extends Error { constructor(message: string) { super(message); @@ -31,6 +44,19 @@ const CANONICAL_PROVENANCE_KEYS = new Set([ "authority", "method", ]); +const PORTABLE_PROVENANCE_KEYS = new Set([ + "schema_version", + "authority", + "calculation_profile", + "calculation_profile_sha256", + "job", + "source", + "run_definition", + "result_package", + "replay_capability", + "storage", + "method", +]); const AUTHORITY_KEYS = new Set([ "commands_enabled", "navigation_or_safety_accepted", @@ -80,32 +106,53 @@ export function decodeLabReplayCapability( } assertExactKeys(value, CAPABILITY_KEYS, `Replay-возможность LAB-сессии ${sessionId}`); if ( - value.schema_version !== "missioncore.observation-lab-replay-capability/v1" - || value.kind !== "canonical-recorded-rerun" - || value.viewer_profile !== "recorded-session" - || value.timeline !== "session_time" - || value.activation !== "explicit" + value.activation !== "explicit" || value.commands_enabled !== false ) { throw new ObservationLabReplayCapabilityContractError( `Replay-возможность LAB-сессии ${sessionId} нарушает observation-only контракт.`, ); } - return { - schemaVersion: "missioncore.observation-lab-replay-capability/v1", - kind: "canonical-recorded-rerun", - viewerProfile: "recorded-session", - timeline: "session_time", - activation: "explicit", - commandsEnabled: false, - }; + if ( + value.schema_version === "missioncore.observation-lab-replay-capability/v1" + && value.kind === "canonical-recorded-rerun" + && value.viewer_profile === "recorded-session" + && value.timeline === "session_time" + ) { + return { + schemaVersion: "missioncore.observation-lab-replay-capability/v1", + kind: "canonical-recorded-rerun", + viewerProfile: "recorded-session", + timeline: "session_time", + activation: "explicit", + commandsEnabled: false, + }; + } + if ( + value.schema_version === "missioncore.observation-lab-replay-capability/v2" + && value.kind === "portable-result-review" + && value.viewer_profile === "portable-result" + && value.timeline === "result-defined" + ) { + return { + schemaVersion: "missioncore.observation-lab-replay-capability/v2", + kind: "portable-result-review", + viewerProfile: "portable-result", + timeline: "result-defined", + activation: "explicit", + commandsEnabled: false, + }; + } + throw new ObservationLabReplayCapabilityContractError( + `Replay-возможность LAB-сессии ${sessionId} содержит неизвестный viewer capability.`, + ); } export function decodeCanonicalLabReplayCapabilityProvenance( provenance: Readonly>, sessionId: string, resultId: string, -): ObservationLabReplayCapability { +): ObservationCanonicalLabReplayCapability { assertExactKeys(provenance, CANONICAL_PROVENANCE_KEYS, `Canonical provenance LAB-сессии ${sessionId}`); if ( provenance.schema_version !== "missioncore.canonical-recorded-lab-projection/v1" @@ -176,7 +223,7 @@ export function decodeCanonicalLabReplayCapabilityProvenance( ); } const capability = decodeLabReplayCapability(provenance.replay_capability, sessionId); - if (capability === null) { + if (capability === null || capability.kind !== "canonical-recorded-rerun") { throw new ObservationLabReplayCapabilityContractError( `Canonical provenance LAB-сессии ${sessionId} не содержит replay capability.`, ); @@ -184,6 +231,44 @@ export function decodeCanonicalLabReplayCapabilityProvenance( return capability; } +export function decodePortableLabReplayCapabilityProvenance( + provenance: Readonly>, + sessionId: string, + resultId: string, + sourceSessionId: string, + definitionSha256: string, +): ObservationPortableLabReplayCapability { + assertExactKeys( + provenance, + PORTABLE_PROVENANCE_KEYS, + `Portable provenance LAB-сессии ${sessionId}`, + ); + if ( + provenance.schema_version !== "missioncore.observatory-portable-result-publication/v1" + || !isRecord(provenance.source) + || provenance.source.session_id !== sourceSessionId + || !isRecord(provenance.run_definition) + || provenance.run_definition.definition_sha256 !== definitionSha256 + || !isRecord(provenance.result_package) + || typeof provenance.result_package.manifest_sha256 !== "string" + || !SHA256.test(provenance.result_package.manifest_sha256) + || typeof provenance.result_package.artifact_manifest_id !== "string" + || !SHA256.test(provenance.result_package.artifact_manifest_id) + || sessionId !== resultId + ) { + throw new ObservationLabReplayCapabilityContractError( + `Portable provenance LAB-сессии ${sessionId} потеряла immutable identity.`, + ); + } + const capability = decodeLabReplayCapability(provenance.replay_capability, sessionId); + if (capability === null || capability.kind !== "portable-result-review") { + throw new ObservationLabReplayCapabilityContractError( + `Portable provenance LAB-сессии ${sessionId} не содержит viewer capability.`, + ); + } + return capability; +} + /** Rolling bridge for a new frontend against the pre-v2 catalog endpoint. */ export function decodeRollingCanonicalLabReplayCapability( provenance: Readonly>, diff --git a/apps/control-station/src/core/observatory/catalogMutations.ts b/apps/control-station/src/core/observatory/catalogMutations.ts index 2d9ac64..c7d30ef 100644 --- a/apps/control-station/src/core/observatory/catalogMutations.ts +++ b/apps/control-station/src/core/observatory/catalogMutations.ts @@ -115,11 +115,18 @@ export async function deleteObservatoryLabProjection( function admittedProjectionId(binding: ObservatoryRecordedRunBinding): string { const sessionId = binding.evidenceSessionId; + const admittedViewer = ( + binding.kind === "canonical-recorded-rerun" + && binding.viewerProfile === "recorded-session" + && binding.timeline === "session_time" + ) || ( + binding.kind === "portable-result-review" + && binding.viewerProfile === "portable-result" + && binding.timeline === "result-defined" + ); if ( - binding.kind !== "canonical-recorded-rerun" + !admittedViewer || binding.activation !== "explicit" - || binding.viewerProfile !== "recorded-session" - || binding.timeline !== "session_time" || binding.resultId !== sessionId || binding.sourceSessionId === sessionId || !SAFE_SESSION_ID.test(sessionId) diff --git a/apps/control-station/src/core/observatory/recordedJobs.ts b/apps/control-station/src/core/observatory/recordedJobs.ts index 550e2f2..c8bb5e5 100644 --- a/apps/control-station/src/core/observatory/recordedJobs.ts +++ b/apps/control-station/src/core/observatory/recordedJobs.ts @@ -14,6 +14,12 @@ export type ObservatoryRecordedJobState = | "failed" | "reconciliation-required"; +export type ObservatoryRecordedJobPublicationState = + | "not-required" + | "pending" + | "failed" + | "published"; + export interface ObservatoryRecordedJob { readonly jobId: string; readonly idempotencyKey: string; @@ -24,6 +30,12 @@ export interface ObservatoryRecordedJob { readonly restartFromZero: boolean; readonly resultId: string | null; readonly terminalMessage: string | null; + readonly publication: { + readonly state: ObservatoryRecordedJobPublicationState; + readonly attempts: number; + readonly error: string | null; + readonly publishedAtUtc: string | null; + }; readonly createdAtUtc: string; readonly updatedAtUtc: string; } @@ -117,12 +129,42 @@ export async function submitObservatoryRecordedJob( return job; } +export async function retryObservatoryRecordedJobPublication( + jobId: string, + { + signal, + fetcher = globalThis.fetch, + }: { + signal?: AbortSignal; + fetcher?: ObservatoryRecordedJobFetch; + } = {}, +): Promise { + const response = await request( + fetcher, + `/api/v1/observatory/runs/${encodeURIComponent(jobId)}/publication/retry`, + { + method: "POST", + headers: { Accept: "application/json" }, + signal, + }, + ); + const body = await responseBody(response); + if (!response.ok) throw apiError(body, response.status); + const job = decodeJob(body); + if (job.jobId !== jobId) { + throw new ObservatoryRecordedJobContractError( + "Повтор публикации вернул другой расчёт.", + ); + } + return job; +} + function decodeJob(value: unknown): ObservatoryRecordedJob { const row = record(value, "расчёт"); exactKeys(row, [ "authority", "checkpoint_policy", "claim_generation", "claim_lease", "created_at_utc", "executor", "idempotency_key", "identity_sha256", "job_id", "preemption_receipt_sha256", - "preemption_requested", "priority", "request_sha256", "restart_from_zero", "result", + "preemption_requested", "priority", "publication", "request_sha256", "restart_from_zero", "result", "schema_version", "setup", "source", "state", "submission_receipt_sha256", "terminal", "updated_at_utc", ], "расчёт"); @@ -161,6 +203,21 @@ function decodeJob(value: unknown): ObservatoryRecordedJob { if (result !== null) exactKeys(result, ["result_id", "sha256"], "result"); const terminal = row.terminal === null ? null : record(row.terminal, "terminal"); if (terminal !== null) exactKeys(terminal, ["code", "message"], "terminal"); + const publication = record(row.publication, "publication"); + exactKeys( + publication, + ["attempts", "error", "published_at_utc", "state"], + "publication", + ); + const publicationState = oneOf(publication.state, [ + "not-required", "pending", "failed", "published", + ] as const, "publication.state"); + const publicationError = publication.error === null + ? null + : text(publication.error, "publication.error"); + const publishedAtUtc = publication.published_at_utc === null + ? null + : text(publication.published_at_utc, "publication.published_at_utc"); return { jobId: text(row.job_id, "job_id"), idempotencyKey: text(row.idempotency_key, "idempotency_key"), @@ -173,6 +230,12 @@ function decodeJob(value: unknown): ObservatoryRecordedJob { terminalMessage: terminal === null || terminal.message === null ? null : text(terminal.message, "terminal.message"), + publication: { + state: publicationState, + attempts: nonNegativeInteger(publication.attempts, "publication.attempts"), + error: publicationError, + publishedAtUtc, + }, createdAtUtc: text(row.created_at_utc, "created_at_utc"), updatedAtUtc: text(row.updated_at_utc, "updated_at_utc"), }; diff --git a/apps/control-station/src/core/observatory/recordedRun.ts b/apps/control-station/src/core/observatory/recordedRun.ts index e93fa1e..0dde8ac 100644 --- a/apps/control-station/src/core/observatory/recordedRun.ts +++ b/apps/control-station/src/core/observatory/recordedRun.ts @@ -6,23 +6,62 @@ import { } from "../laboratory/vegetationShadow"; import type { ObservationLabInstance } from "../observation/sessionArchive"; import { + decodeLabReplayCapability, decodeCanonicalLabReplayCapabilityProvenance, + decodePortableLabReplayCapabilityProvenance, ObservationLabReplayCapabilityContractError, } from "../observation/labReplayCapability"; const CANONICAL_RESULT_ID = /^lab-v1-vegetation-shadow-[a-f0-9]{64}$/; const CANONICAL_SOURCE_SESSION_ID = "20260828T130511Z_viewer_live"; -export interface ObservatoryRecordedRunBinding { - readonly kind: "canonical-recorded-rerun"; +interface ObservatoryRecordedRunBindingBase { readonly evidenceSessionId: string; readonly sourceSessionId: string; readonly resultId: string; - readonly viewerProfile: "recorded-session"; - readonly timeline: "session_time"; readonly activation: "explicit"; } +export interface ObservatoryCanonicalRecordedRunBinding + extends ObservatoryRecordedRunBindingBase { + readonly kind: "canonical-recorded-rerun"; + readonly viewerProfile: "recorded-session"; + readonly timeline: "session_time"; +} + +export interface ObservatoryPortableResultBinding + extends ObservatoryRecordedRunBindingBase { + readonly kind: "portable-result-review"; + readonly viewerProfile: "portable-result"; + readonly timeline: "result-defined"; + readonly definitionSha256: string; +} + +export type ObservatoryRecordedRunBinding = + | ObservatoryCanonicalRecordedRunBinding + | ObservatoryPortableResultBinding; + +export interface ObservatoryPortableResultReview { + readonly kind: "portable-result"; + readonly resultId: string; + readonly sourceSessionId: string; + readonly resultKind: string; + readonly definitionSha256: string; + readonly calculationProfile: Readonly> | null; + readonly artifactManifestId: string; + readonly artifacts: readonly { + readonly role: string; + readonly mediaType: string; + readonly sha256: string; + readonly byteLength: number; + }[]; + readonly resultDocument: Readonly>; +} + +export type ObservatoryRecordedRunReview = + | { readonly kind: "canonical-recorded-rerun"; readonly review: VegetationFullRouteReview } + | ObservatoryPortableResultReview; + export class ObservatoryRecordedRunContractError extends Error { constructor(message: string) { super(message); @@ -36,6 +75,44 @@ export function observatoryRecordedRunBinding( ): ObservatoryRecordedRunBinding | null { const capability = lab.replayCapability; if (!capability) return null; + if (capability.kind === "portable-result-review") { + if ( + evidenceSessionId !== lab.resultId + || lab.sourceSessionId === evidenceSessionId + || lab.configSha256 === null + || !/^[a-f0-9]{64}$/.test(lab.configSha256) + || capability.viewerProfile !== "portable-result" + || capability.timeline !== "result-defined" + || capability.activation !== "explicit" + || capability.commandsEnabled !== false + ) { + throw new ObservatoryRecordedRunContractError( + `Portable-результат ${lab.resultId} не допущен к универсальному viewer.`, + ); + } + try { + decodePortableLabReplayCapabilityProvenance( + lab.provenance, + evidenceSessionId, + lab.resultId, + lab.sourceSessionId, + lab.configSha256, + ); + } catch (error) { + if (!(error instanceof ObservationLabReplayCapabilityContractError)) throw error; + throw new ObservatoryRecordedRunContractError(error.message); + } + return { + kind: "portable-result-review", + evidenceSessionId, + sourceSessionId: lab.sourceSessionId, + resultId: lab.resultId, + viewerProfile: "portable-result", + timeline: "result-defined", + activation: "explicit", + definitionSha256: lab.configSha256, + }; + } try { decodeCanonicalLabReplayCapabilityProvenance( lab.provenance, @@ -87,25 +164,31 @@ export async function fetchObservatoryRecordedRunReview( fetcher?: LaboratoryFetch; signal?: AbortSignal; }, -): Promise { +): Promise { if (binding.sourceSessionId !== selectedSourceSessionId) { throw new ObservatoryRecordedRunContractError( "Запуск не связан с выбранной исходной сессией.", ); } + if (binding.kind === "portable-result-review") { + return fetchPortableResultReview(binding, { fetcher, signal }); + } const result = await fetchVegetationShadowResultMetadata(binding.resultId, { fetcher, signal, }); - return admitObservatoryRecordedRunReview( - binding, - selectedSourceSessionId, - result, - ); + return { + kind: "canonical-recorded-rerun", + review: admitObservatoryRecordedRunReview( + binding, + selectedSourceSessionId, + result, + ), + }; } export function admitObservatoryRecordedRunReview( - binding: ObservatoryRecordedRunBinding, + binding: ObservatoryCanonicalRecordedRunBinding, selectedSourceSessionId: string, result: VegetationShadowResult, ): VegetationFullRouteReview { @@ -127,3 +210,126 @@ export function admitObservatoryRecordedRunReview( } return review; } + +async function fetchPortableResultReview( + binding: ObservatoryPortableResultBinding, + { + fetcher, + signal, + }: { + fetcher: LaboratoryFetch; + signal?: AbortSignal; + }, +): Promise { + const response = await fetcher( + `/api/v1/observatory/portable-results/${encodeURIComponent(binding.resultId)}`, + { headers: { Accept: "application/json" }, signal }, + ); + let body: unknown; + try { + body = await response.json(); + } catch { + throw new ObservatoryRecordedRunContractError( + "Portable viewer получил некорректный ответ сервера.", + ); + } + if (!response.ok) { + throw new ObservatoryRecordedRunContractError( + isRecord(body) && typeof body.detail === "string" + ? body.detail + : `Portable viewer вернул HTTP ${response.status}.`, + ); + } + return decodePortableResultReview(body, binding); +} + +function decodePortableResultReview( + value: unknown, + binding: ObservatoryPortableResultBinding, +): ObservatoryPortableResultReview { + if (!isRecord(value)) { + throw new ObservatoryRecordedRunContractError("Portable viewer вернул не объект."); + } + const expected = new Set([ + "schema_version", + "result_id", + "source_session_id", + "result_kind", + "definition_sha256", + "viewer_capability", + "calculation_profile", + "artifact_manifest_id", + "artifacts", + "result_document", + ]); + if (Object.keys(value).some((key) => !expected.has(key)) || Object.keys(value).length !== expected.size) { + throw new ObservatoryRecordedRunContractError("Portable viewer изменил контракт ответа."); + } + let viewerCapability; + try { + viewerCapability = decodeLabReplayCapability( + value.viewer_capability, + binding.evidenceSessionId, + ); + } catch (error) { + if (!(error instanceof ObservationLabReplayCapabilityContractError)) throw error; + throw new ObservatoryRecordedRunContractError(error.message); + } + if ( + value.schema_version !== "missioncore.observatory-portable-result-view/v1" + || value.result_id !== binding.resultId + || value.source_session_id !== binding.sourceSessionId + || value.definition_sha256 !== binding.definitionSha256 + || typeof value.result_kind !== "string" + || typeof value.artifact_manifest_id !== "string" + || !/^[a-f0-9]{64}$/.test(value.artifact_manifest_id) + || !Array.isArray(value.artifacts) + || !isRecord(value.result_document) + || !isRecord(value.calculation_profile) + || viewerCapability === null + || viewerCapability.kind !== "portable-result-review" + || viewerCapability.viewerProfile !== binding.viewerProfile + || viewerCapability.timeline !== binding.timeline + ) { + throw new ObservatoryRecordedRunContractError("Portable viewer потерял identity результата."); + } + const artifacts = value.artifacts.map((item) => { + if ( + !isRecord(item) + || Object.keys(item).length !== 4 + || !["role", "media_type", "sha256", "byte_length"].every( + (key) => Object.prototype.hasOwnProperty.call(item, key), + ) + || typeof item.role !== "string" + || typeof item.media_type !== "string" + || typeof item.sha256 !== "string" + || !/^[a-f0-9]{64}$/.test(item.sha256) + || typeof item.byte_length !== "number" + || !Number.isSafeInteger(item.byte_length) + || item.byte_length < 0 + ) { + throw new ObservatoryRecordedRunContractError("Portable viewer получил неверный artifact."); + } + return { + role: item.role, + mediaType: item.media_type, + sha256: item.sha256, + byteLength: item.byte_length, + }; + }); + return { + kind: "portable-result", + resultId: binding.resultId, + sourceSessionId: binding.sourceSessionId, + resultKind: value.result_kind, + definitionSha256: binding.definitionSha256, + calculationProfile: value.calculation_profile, + artifactManifestId: value.artifact_manifest_id, + artifacts, + resultDocument: value.result_document, + }; +} + +function isRecord(value: unknown): value is Record { + return value !== null && typeof value === "object" && !Array.isArray(value); +} diff --git a/apps/control-station/src/core/observatory/useObservatoryLaboratorySetups.ts b/apps/control-station/src/core/observatory/useObservatoryLaboratorySetups.ts index 3e470fb..65dfd51 100644 --- a/apps/control-station/src/core/observatory/useObservatoryLaboratorySetups.ts +++ b/apps/control-station/src/core/observatory/useObservatoryLaboratorySetups.ts @@ -42,54 +42,64 @@ 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 }), + const legacyResult = settled( + fetchObservatoryLaboratorySetups(sourceSessionId, { signal: request.signal }), ); - void fetchObservatoryLaboratorySetups(sourceSessionId, { signal: request.signal }) - .then(async (legacyCatalog) => { + const portableResult = settled( + fetchObservatoryPortableLaboratorySetups(sourceSessionId, { signal: request.signal }), + ); + void Promise.all([legacyResult, portableResult]) + .then(([legacy, portable]) => { if (request.signal.aborted || requestSequence.current !== sequence) return; - publishSetupCatalog( - legacyCatalog, - setCatalog, - setSelectedSetupId, - { preserveUnknownSelection: true }, - ); - setState("ready"); - const optionalPortable = await portableResult; - if (request.signal.aborted || requestSequence.current !== sequence) return; - if (optionalPortable.status === "fulfilled") { + if (legacy.status === "fulfilled" && portable.status === "fulfilled") { try { publishSetupCatalog( - mergeSetupCatalogs(legacyCatalog, optionalPortable.value), + mergeSetupCatalogs(legacy.value, portable.value), setCatalog, setSelectedSetupId, ); setError(null); } catch (caught) { - publishSetupCatalog(legacyCatalog, setCatalog, setSelectedSetupId); + publishSetupCatalog(legacy.value, setCatalog, setSelectedSetupId); setError(catalogErrorMessage( caught, "Portable-каталог профилей нарушил локальный контракт.", )); } + } else if ( + legacy.status === "rejected" + && portable.status === "fulfilled" + ) { + publishSetupCatalog(portable.value, setCatalog, setSelectedSetupId); + setError(catalogErrorMessage( + legacy.reason, + "Архивный каталог сетапов недоступен.", + )); + } else if ( + legacy.status === "fulfilled" + && portable.status === "rejected" + ) { + publishSetupCatalog(legacy.value, setCatalog, setSelectedSetupId); + setError(catalogErrorMessage( + portable.reason, + "Portable-каталог профилей недоступен.", + )); + } else if ( + legacy.status === "rejected" + && portable.status === "rejected" + ) { + setState("error"); + setError(catalogErrorMessage( + portable.reason, + catalogErrorMessage(legacy.reason, "Каталог сетапов недоступен."), + )); + return; + } else { + setState("error"); + setError("Каталог сетапов вернул неизвестное состояние."); return; } - publishSetupCatalog(legacyCatalog, setCatalog, setSelectedSetupId); - setError(catalogErrorMessage( - optionalPortable.reason, - "Portable-каталог профилей недоступен.", - )); - }) - .catch((caught: unknown) => { - if (request.signal.aborted || requestSequence.current !== sequence) return; - setState("error"); - setError(caught instanceof Error && caught.message.trim() - ? caught.message - : "Каталог сетапов недоступен."); + setState("ready"); }); return () => request.abort(); }, [revision, sourceSessionId]); @@ -188,6 +198,13 @@ function catalogErrorMessage(caught: unknown, fallback: string): string { : fallback; } +function settled(promise: Promise): Promise> { + return promise.then( + (value) => ({ status: "fulfilled", value }), + (reason: unknown) => ({ status: "rejected", reason }), + ); +} + function mergeSetupCatalogs( legacy: ObservatoryLaboratorySetupCatalog, portable: ObservatoryLaboratorySetupCatalog, @@ -195,13 +212,9 @@ function mergeSetupCatalogs( if (legacy.sourceSessionId !== portable.sourceSessionId) { throw new Error("Каталоги сетапов относятся к разным исходным сессиям."); } - const portableProfileNames = new Set( - portable.setups.map((setup) => setup.displayName), - ); + const portableSetupIds = new Set(portable.setups.map((setup) => setup.setupId)); const setups = [ - ...legacy.setups.filter( - (setup) => !portableProfileNames.has(setup.displayName), - ), + ...legacy.setups.filter((setup) => !portableSetupIds.has(setup.setupId)), ...portable.setups, ]; if (new Set(setups.map((setup) => setup.setupId)).size !== setups.length) { diff --git a/apps/control-station/src/core/observatory/useObservatoryRecordedJobs.ts b/apps/control-station/src/core/observatory/useObservatoryRecordedJobs.ts index 82f8dda..6a1a542 100644 --- a/apps/control-station/src/core/observatory/useObservatoryRecordedJobs.ts +++ b/apps/control-station/src/core/observatory/useObservatoryRecordedJobs.ts @@ -2,16 +2,25 @@ import { useCallback, useEffect, useMemo, useRef, useState } from "react"; import { fetchObservatoryRecordedJobs, + retryObservatoryRecordedJobPublication, submitObservatoryRecordedJob, type ObservatoryRecordedJob, } from "./recordedJobs"; -type RecordedJobsState = "idle" | "loading" | "ready" | "refreshing" | "submitting" | "error"; +type RecordedJobsState = + | "idle" + | "loading" + | "ready" + | "refreshing" + | "submitting" + | "retrying-publication" + | "error"; const OPEN_STATES = new Set([ "accepted", "queued", "claimed", "running", "paused", "preemption-pending", "reconciliation-required", ]); +const POLL_INTERVAL_MS = 1_500; export function useObservatoryRecordedJobs(sourceSessionId: string, setupId: string) { const [jobs, setJobs] = useState([]); @@ -60,6 +69,16 @@ export function useObservatoryRecordedJobs(sourceSessionId: string, setupId: str () => jobs.find((job) => OPEN_STATES.has(job.state)) ?? null, [jobs], ); + const publicationPending = jobs.some((job) => job.publication.state === "pending"); + + useEffect(() => { + if (state !== "ready" || (!activeJob && !publicationPending)) return; + const timer = globalThis.setTimeout( + () => setRevision((value) => value + 1), + POLL_INTERVAL_MS, + ); + return () => globalThis.clearTimeout(timer); + }, [activeJob, publicationPending, revision, state]); useEffect(() => { if (!latestJob || activeJob) return; @@ -108,7 +127,47 @@ export function useObservatoryRecordedJobs(sourceSessionId: string, setupId: str } }, [activeJob, selectionKey, setupId, sourceSessionId, state]); - return { jobs, latestJob, activeJob, state, error, refresh, submit }; + const retryPublication = useCallback(async (): Promise => { + if (!latestJob || latestJob.publication.state !== "failed") return null; + if (state === "retrying-publication") return null; + requestRef.current?.abort(); + const request = new AbortController(); + requestRef.current = request; + setState("retrying-publication"); + setError(null); + try { + const job = await retryObservatoryRecordedJobPublication(latestJob.jobId, { + signal: request.signal, + }); + if (request.signal.aborted || requestRef.current !== request) return null; + setJobs((current) => [ + job, + ...current.filter((candidate) => candidate.jobId !== job.jobId), + ]); + setState("ready"); + return job; + } catch (caught) { + if (request.signal.aborted || requestRef.current !== request) return null; + setState("error"); + setError(caught instanceof Error && caught.message.trim() + ? caught.message + : "Не удалось повторить публикацию результата."); + return null; + } finally { + if (requestRef.current === request) requestRef.current = null; + } + }, [latestJob, state]); + + return { + jobs, + latestJob, + activeJob, + state, + error, + refresh, + submit, + retryPublication, + }; } function createIdempotencyKey(): string { diff --git a/apps/control-station/src/workspaces/observatory/ObservatoryWorkspace.tsx b/apps/control-station/src/workspaces/observatory/ObservatoryWorkspace.tsx index 8f37bfd..8ea7ea7 100644 --- a/apps/control-station/src/workspaces/observatory/ObservatoryWorkspace.tsx +++ b/apps/control-station/src/workspaces/observatory/ObservatoryWorkspace.tsx @@ -178,6 +178,7 @@ export function ObservatoryWorkspace({ >(null); const overviewRef = useRef(null); const replayCoordinatorRef = useRef(createObservationReplayCoordinator()); + const refreshedPublishedJobRef = useRef(null); const items = controller.catalog?.items ?? EMPTY_OBSERVATORY_ITEMS; const closeReplay = useCallback(() => { @@ -244,14 +245,23 @@ export function ObservatoryWorkspace({ const presentedJob = recordedJobsController.activeJob ?? recordedJobsController.latestJob; const presentedJobStatus = presentedJob - ? recordedJobStatus[presentedJob.state] + ? presentedJob.state === "succeeded" + ? presentedJob.publication.state === "published" + ? { label: "Результат опубликован", tone: "success" as const } + : presentedJob.publication.state === "failed" + ? { label: "Ошибка публикации", tone: "danger" as const } + : presentedJob.publication.state === "not-required" + ? { label: "Расчёт завершён", tone: "success" as const } + : recordedJobStatus.succeeded + : recordedJobStatus[presentedJob.state] : null; const queueStatusError = setupController.preflight.kind === "error" ? setupController.preflight.message : recordedJobsController.error; const queueStateBusy = recordedJobsController.state === "loading" || recordedJobsController.state === "refreshing" - || recordedJobsController.state === "submitting"; + || recordedJobsController.state === "submitting" + || recordedJobsController.state === "retrying-publication"; const canSubmitRecordedJob = queueSubmissionAllowed && recordedJobsController.state === "ready" && recordedJobsController.activeJob === null; @@ -267,6 +277,16 @@ export function ObservatoryWorkspace({ (evidence) => evidence.sessionId === replayEvidenceId, ) ?? null; + useEffect(() => { + const publishedJob = recordedJobsController.jobs.find( + (job) => job.publication.state === "published" && job.resultId !== null, + ); + if (!publishedJob || refreshedPublishedJobRef.current === publishedJob.jobId) return; + refreshedPublishedJobRef.current = publishedJob.jobId; + void controller.refresh(); + setupController.refresh(); + }, [controller.refresh, recordedJobsController.jobs, setupController.refresh]); + useEffect(() => { if ( replayEvidenceId === null @@ -496,10 +516,14 @@ export function ObservatoryWorkspace({ ) : recordedJobsController.state === "submitting" ? ( Ставим в очередь + ) : recordedJobsController.state === "retrying-publication" ? ( + Повторяем публикацию ) : presentedJobStatus ? ( {presentedJobStatus.label} @@ -542,6 +566,16 @@ export function ObservatoryWorkspace({ Рассчитать ) : null} + {presentedJob?.publication.state === "failed" + && recordedJobsController.state !== "retrying-publication" ? ( + + ) : null} @@ -610,6 +644,20 @@ export function ObservatoryWorkspace({
Чтение
{selectedSession.source.replayable ? "Доступна" : "Не подготовлена"}
+
+
Оборудование
+
+ {selectedSession.source.captureAttestation?.equipmentDisplayName + ?? "Не аттестовано"} +
+
+
+
Профиль записи
+
+ {selectedSession.source.captureAttestation?.captureProfileId + ?? "Не определён"} +
+
@@ -738,12 +786,20 @@ export function ObservatoryWorkspace({
) : replay.kind === "ready" ? ( -
+
- ВИЗУАЛЬНЫЙ РАЗБОР / ЗАПИСАННАЯ СЕССИЯ + + {replay.review.kind === "canonical-recorded-rerun" + ? "ВИЗУАЛЬНЫЙ РАЗБОР / ЗАПИСАННАЯ СЕССИЯ" + : "РЕЗУЛЬТАТ / ПРОВЕРЕННЫЙ ДОКУМЕНТ"} +

{replayEvidence?.label ?? replay.binding.resultId}

-

Записанный маршрут синхронизирован по общей временной шкале.

+

+ {replay.review.kind === "canonical-recorded-rerun" + ? "Записанный маршрут синхронизирован по общей временной шкале." + : "Показан проверенный документ результата и связанные с ним артефакты."} +

- + {replay.review.kind === "canonical-recorded-rerun" ? ( + + ) : ( + +
+ Проверено +

{replay.review.resultKind}

+

Связанных артефактов: {replay.review.artifacts.length}

+
+ Документ результата + 
{JSON.stringify(replay.review.resultDocument, null, 2)}
+
+
+ + )}
) : null} diff --git a/apps/control-station/test/observatoryRecordedJobs.test.mjs b/apps/control-station/test/observatoryRecordedJobs.test.mjs index 9150eb6..7e36df2 100644 --- a/apps/control-station/test/observatoryRecordedJobs.test.mjs +++ b/apps/control-station/test/observatoryRecordedJobs.test.mjs @@ -5,6 +5,7 @@ import { createServer } from "vite"; let server; let fetchObservatoryRecordedJobs; +let retryObservatoryRecordedJobPublication; let submitObservatoryRecordedJob; let ObservatoryRecordedJobContractError; @@ -65,6 +66,12 @@ function job(state = "queued") { result: state === "succeeded" ? { result_id: "m49-result", sha256: "d".repeat(64) } : null, + publication: { + state: "not-required", + attempts: 0, + error: null, + published_at_utc: null, + }, terminal: state === "failed" ? { code: "worker-failed", message: "Worker завершил расчёт с ошибкой." } : null, @@ -82,6 +89,7 @@ before(async () => { }); ({ fetchObservatoryRecordedJobs, + retryObservatoryRecordedJobPublication, submitObservatoryRecordedJob, ObservatoryRecordedJobContractError, } = await server.ssrLoadModule("/src/core/observatory/recordedJobs.ts")); @@ -212,3 +220,28 @@ test("portable submission carries the exact definition/check fence", async () => check_sha256: "f".repeat(64), }); }); + +test("publication retry never submits a second compute request", async () => { + let request; + const response = job("succeeded"); + response.publication = { + state: "published", + attempts: 2, + error: null, + published_at_utc: "2026-09-01T12:00:00Z", + }; + const retried = await retryObservatoryRecordedJobPublication(response.job_id, { + fetcher: async (input, init) => { + request = { input: String(input), init }; + return new Response(JSON.stringify(response), { status: 200 }); + }, + }); + + assert.equal( + request.input, + `/api/v1/observatory/runs/${response.job_id}/publication/retry`, + ); + assert.equal(request.init.method, "POST"); + assert.equal(request.init.body, undefined); + assert.equal(retried.publication.state, "published"); +}); diff --git a/apps/control-station/test/observatoryRecordedRun.test.mjs b/apps/control-station/test/observatoryRecordedRun.test.mjs index 60f80a7..173d4a2 100644 --- a/apps/control-station/test/observatoryRecordedRun.test.mjs +++ b/apps/control-station/test/observatoryRecordedRun.test.mjs @@ -6,6 +6,7 @@ import { createServer } from "vite"; let server; let admitObservatoryRecordedRunReview; +let fetchObservatoryRecordedRunReview; let observatoryRecordedRunBinding; let ObservatoryRecordedRunContractError; @@ -17,6 +18,7 @@ before(async () => { }); ({ admitObservatoryRecordedRunReview, + fetchObservatoryRecordedRunReview, observatoryRecordedRunBinding, ObservatoryRecordedRunContractError, } = await server.ssrLoadModule("/src/core/observatory/recordedRun.ts")); @@ -87,6 +89,55 @@ function lab(overrides = {}) { }; } +function portableCapability() { + return { + schemaVersion: "missioncore.observation-lab-replay-capability/v2", + kind: "portable-result-review", + viewerProfile: "portable-result", + timeline: "result-defined", + activation: "explicit", + commandsEnabled: false, + }; +} + +function portableCapabilityDocument() { + return { + schema_version: "missioncore.observation-lab-replay-capability/v2", + kind: "portable-result-review", + viewer_profile: "portable-result", + timeline: "result-defined", + activation: "explicit", + commands_enabled: false, + }; +} + +function portableLab(portableResultId, definitionSha256) { + return lab({ + labId: "M4.9", + resultId: portableResultId, + resultKind: "recorded-route-analysis", + sourceResultId: null, + configSha256: definitionSha256, + replayCapability: portableCapability(), + provenance: { + schema_version: "missioncore.observatory-portable-result-publication/v1", + authority: {}, + calculation_profile: {}, + calculation_profile_sha256: "b".repeat(64), + job: {}, + source: { session_id: sourceSessionId }, + run_definition: { definition_sha256: definitionSha256 }, + result_package: { + manifest_sha256: "c".repeat(64), + artifact_manifest_id: "d".repeat(64), + }, + replay_capability: portableCapabilityDocument(), + storage: {}, + method: {}, + }, + }); +} + test("Observatory admits only the exact typed canonical recorded run", () => { const binding = observatoryRecordedRunBinding(resultId, lab()); assert.deepEqual(binding, { @@ -167,6 +218,67 @@ test("Observatory rechecks the selected source and sealed review before mounting ); }); +test("Observatory selects and strictly decodes a portable result viewer by capability", async () => { + const portableResultId = "portable-result-a"; + const definitionSha256 = "e".repeat(64); + const binding = observatoryRecordedRunBinding( + portableResultId, + portableLab(portableResultId, definitionSha256), + ); + assert.deepEqual(binding, { + kind: "portable-result-review", + evidenceSessionId: portableResultId, + sourceSessionId, + resultId: portableResultId, + viewerProfile: "portable-result", + timeline: "result-defined", + activation: "explicit", + definitionSha256, + }); + + let requestedUrl = null; + const review = await fetchObservatoryRecordedRunReview(binding, { + selectedSourceSessionId: sourceSessionId, + fetcher: async (url) => { + requestedUrl = url; + return { + ok: true, + status: 200, + json: async () => ({ + schema_version: "missioncore.observatory-portable-result-view/v1", + result_id: portableResultId, + source_session_id: sourceSessionId, + result_kind: "recorded-route-analysis", + definition_sha256: definitionSha256, + viewer_capability: portableCapabilityDocument(), + calculation_profile: { name: "M4.9" }, + artifact_manifest_id: "d".repeat(64), + artifacts: [{ + role: "result-document", + media_type: "application/json", + sha256: "f".repeat(64), + byte_length: 42, + }], + result_document: { verdict: "reviewed" }, + }), + }; + }, + }); + + assert.equal( + requestedUrl, + `/api/v1/observatory/portable-results/${portableResultId}`, + ); + assert.equal(review.kind, "portable-result"); + assert.deepEqual(review.resultDocument, { verdict: "reviewed" }); + assert.deepEqual(review.artifacts, [{ + role: "result-document", + mediaType: "application/json", + sha256: "f".repeat(64), + byteLength: 42, + }]); +}); + test("Observatory run admission remains metadata-only until explicit UI activation", async () => { const source = await readFile( new URL("../src/core/observatory/recordedRun.ts", import.meta.url), diff --git a/apps/control-station/test/observatoryWorkspace.test.mjs b/apps/control-station/test/observatoryWorkspace.test.mjs index b138596..c547d6c 100644 --- a/apps/control-station/test/observatoryWorkspace.test.mjs +++ b/apps/control-station/test/observatoryWorkspace.test.mjs @@ -99,6 +99,10 @@ test("Observatory mounts the one shared canonical replay only after explicit adm assert.doesNotMatch(workspace, /observatory-evidence-card[^>]*tone="soft"/); assert.match(workspace, /Открыть визуальный разбор/); assert.match(workspace, /replay\.kind === "ready"[\s\S]*Документ результата<\/summary>/); + assert.doesNotMatch(workspace, /UNIVERSAL VIEWER|content-addressed artifacts/); assert.match(workspace, /Проверяем точную связь результата/); assert.match(workspace, /role="alert"/); assert.match(workspace, /Повторить/); @@ -247,12 +251,18 @@ test("Observatory keeps one compact selector axis without the obsolete setup det workspace, /"preemption-pending": \{[\s\S]*label: "Ждём подтверждения остановки"/, ); - assert.doesNotMatch(setupHook, /Promise\.allSettled/); - assert.match(setupHook, /publishSetupCatalog\(\s*legacyCatalog/); - assert.match(setupHook, /preserveUnknownSelection: true/); + assert.match(setupHook, /Promise\.all\(\[legacyResult, portableResult\]\)/); assert.match( setupHook, - /portableProfileNames[\s\S]*legacy\.setups\.filter\([\s\S]*!portableProfileNames\.has\(setup\.displayName\)[\s\S]*\.\.\.portable\.setups/, + /legacy\.status === "rejected"[\s\S]*portable\.status === "fulfilled"[\s\S]*publishSetupCatalog\(portable\.value/, + ); + assert.match( + setupHook, + /legacy\.status === "fulfilled"[\s\S]*portable\.status === "rejected"[\s\S]*publishSetupCatalog\(legacy\.value/, + ); + assert.match( + setupHook, + /portableSetupIds[\s\S]*legacy\.setups\.filter\([\s\S]*!portableSetupIds\.has\(setup\.setupId\)[\s\S]*\.\.\.portable\.setups/, ); assert.match(setupHook, /selectedSetupId, sourceSessionId/); assert.match(workspace, /Worker готов к проверке/); @@ -270,7 +280,11 @@ test("Observatory keeps one compact selector axis without the obsolete setup det assert.match(jobsHook, /return `observatory-ui:\$\{entropy\}`;/); assert.doesNotMatch(jobsHook, /observatory-ui:[^`]*sourceSessionId|\.slice\(0, 160\)/); assert.match(workspace, /recordedJobsController\.refresh\(\)/); - assert.doesNotMatch(`${workspace}\n${jobsHook}`, /setInterval|setTimeout/); + assert.match(jobsHook, /POLL_INTERVAL_MS = 1_500/); + assert.match(jobsHook, /globalThis\.setTimeout/); + assert.doesNotMatch(jobsHook, /setInterval/); + assert.match(workspace, /job\.publication\.state === "published"/); + assert.match(workspace, /void controller\.refresh\(\);[\s\S]*setupController\.refresh\(\);/); assert.match( styles, /\.observatory-catalog-bar__controls \{[\s\S]*min-width: 0;[\s\S]*flex: 1 1 auto;[\s\S]*flex-wrap: nowrap;[\s\S]*justify-content: flex-end;/,