From e436fb56d4631449774c1beb4c2bbfcf08b01431 Mon Sep 17 00:00:00 2001 From: DCCONSTRUCTIONS Date: Thu, 3 Sep 2026 10:16:01 +0300 Subject: [PATCH] fix(observatory): offer only uncalculated profiles without completion badges --- .../src/core/observatory/laboratorySetups.ts | 10 + .../src/core/observatory/recordedJobs.ts | 9 +- .../useObservatoryLaboratorySetups.ts | 144 +++----------- .../observatory/useObservatoryRecordedJobs.ts | 151 +++++++------- .../src/styles/observatory.css | 24 +-- .../observatory/ObservatoryWorkspace.tsx | 185 +++++------------- .../test/observatoryHooks.test.mjs | 96 +++++++++ .../test/observatoryLaboratorySetups.test.mjs | 34 ++++ .../test/observatoryRecordedJobs.test.mjs | 30 +++ .../test/observatoryWorkspace.test.mjs | 53 ++--- src/k1link/observatory/recorded_jobs.py | 6 + src/k1link/web/observatory_api.py | 2 + tests/test_observatory_recorded_jobs.py | 25 +++ tests/test_observatory_recorded_run_api.py | 19 ++ 14 files changed, 420 insertions(+), 368 deletions(-) create mode 100644 apps/control-station/test/observatoryHooks.test.mjs diff --git a/apps/control-station/src/core/observatory/laboratorySetups.ts b/apps/control-station/src/core/observatory/laboratorySetups.ts index 6f77107..9598c91 100644 --- a/apps/control-station/src/core/observatory/laboratorySetups.ts +++ b/apps/control-station/src/core/observatory/laboratorySetups.ts @@ -74,6 +74,16 @@ export interface ObservatoryLaboratorySetupCatalog { readonly setups: readonly ObservatoryLaboratorySetup[]; } +// Cache identity is verified by the portable catalog decoder/server, never by a LAB label. +export function selectableObservatorySetups(catalog: ObservatoryLaboratorySetupCatalog | null) { + return catalog?.setups.filter((setup) => ( + setup.origin === "portable-definition" + && setup.runDefinition !== null + && setup.compatibility.compatible + && setup.preflight.outcome !== "existing" + )) ?? []; +} + export interface ObservatoryLaboratoryRunPreflight { readonly sourceSessionId: string; readonly setupId: string; diff --git a/apps/control-station/src/core/observatory/recordedJobs.ts b/apps/control-station/src/core/observatory/recordedJobs.ts index c8bb5e5..17031f2 100644 --- a/apps/control-station/src/core/observatory/recordedJobs.ts +++ b/apps/control-station/src/core/observatory/recordedJobs.ts @@ -60,9 +60,11 @@ export async function fetchObservatoryRecordedJobs( setupId: string, { signal, + definitionSha256, fetcher = globalThis.fetch, }: { signal?: AbortSignal; + definitionSha256?: string; fetcher?: ObservatoryRecordedJobFetch; } = {}, ): Promise { @@ -71,6 +73,7 @@ export async function fetchObservatoryRecordedJobs( setup_id: setupId, limit: "20", }); + if (definitionSha256 !== undefined) query.set("definition_sha256", definitionSha256); const response = await request(fetcher, `/api/v1/observatory/runs?${query}`, { method: "GET", headers: { Accept: "application/json" }, @@ -84,6 +87,7 @@ export async function fetchObservatoryRecordedJobs( observationAuthority(row.authority); return array(row.items, "items").map(decodeJob).filter((job) => ( job.sourceSessionId === sourceSessionId && job.setupId === setupId + && (definitionSha256 === undefined || job.definitionSha256 === definitionSha256) )); } @@ -121,9 +125,10 @@ export async function submitObservatoryRecordedJob( const body = await responseBody(response); if (!response.ok) throw apiError(body, response.status); const job = decodeJob(body); - if (job.sourceSessionId !== sourceSessionId || job.setupId !== setupId) { + if (job.sourceSessionId !== sourceSessionId || job.setupId !== setupId + || (portableBinding !== null && job.definitionSha256 !== portableBinding.definitionSha256)) { throw new ObservatoryRecordedJobContractError( - "Расчёт относится к другой сессии или сетапу.", + "Расчёт относится к другой сессии, сетапу или версии профиля.", ); } return job; diff --git a/apps/control-station/src/core/observatory/useObservatoryLaboratorySetups.ts b/apps/control-station/src/core/observatory/useObservatoryLaboratorySetups.ts index 65dfd51..778e1bd 100644 --- a/apps/control-station/src/core/observatory/useObservatoryLaboratorySetups.ts +++ b/apps/control-station/src/core/observatory/useObservatoryLaboratorySetups.ts @@ -1,9 +1,9 @@ import { useCallback, useEffect, useMemo, useRef, useState } from "react"; import { - fetchObservatoryLaboratorySetups, fetchObservatoryPortableLaboratorySetups, preflightObservatoryLaboratorySetup, + selectableObservatorySetups, type ObservatoryLaboratoryRunPreflight, type ObservatoryLaboratorySetupCatalog, } from "./laboratorySetups"; @@ -25,6 +25,9 @@ export function useObservatoryLaboratorySetups(sourceSessionId: string) { const requestSequence = useRef(0); const preflightRequest = useRef(null); const activeCatalog = catalog?.sourceSessionId === sourceSessionId ? catalog : null; + const selectableSetups = useMemo(() => selectableObservatorySetups(activeCatalog), [activeCatalog]); + const selectedSetup = selectableSetups.find((setup) => setup.setupId === selectedSetupId) ?? null; + const selectedDefinitionSha256 = selectedSetup?.runDefinition?.definitionSha256 ?? null; useEffect(() => { preflightRequest.current?.abort(); @@ -42,64 +45,25 @@ export function useObservatoryLaboratorySetups(sourceSessionId: string) { setState((current) => activeCatalog && current !== "idle" ? "refreshing" : "loading"); setError(null); setPreflight({ kind: "idle" }); - const legacyResult = settled( - fetchObservatoryLaboratorySetups(sourceSessionId, { signal: request.signal }), - ); - const portableResult = settled( - fetchObservatoryPortableLaboratorySetups(sourceSessionId, { signal: request.signal }), - ); - void Promise.all([legacyResult, portableResult]) - .then(([legacy, portable]) => { + // Archived definitions belong to evidence, not to the calculation selector. + void fetchObservatoryPortableLaboratorySetups(sourceSessionId, { signal: request.signal }) + .then((next) => { if (request.signal.aborted || requestSequence.current !== sequence) return; - if (legacy.status === "fulfilled" && portable.status === "fulfilled") { - try { - publishSetupCatalog( - mergeSetupCatalogs(legacy.value, portable.value), - setCatalog, - setSelectedSetupId, - ); - setError(null); - } catch (caught) { - 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; - } + setCatalog(next); + const selectable = selectableObservatorySetups(next); + setSelectedSetupId((current) => ( + selectable.some((setup) => setup.setupId === current) + ? current + : selectable[0]?.setupId ?? "" + )); setState("ready"); + }) + .catch((caught: unknown) => { + if (request.signal.aborted || requestSequence.current !== sequence) return; + setState("error"); + setError(caught instanceof Error && caught.message.trim() + ? caught.message + : "Каталог профилей недоступен."); }); return () => request.abort(); }, [revision, sourceSessionId]); @@ -110,12 +74,7 @@ export function useObservatoryLaboratorySetups(sourceSessionId: string) { preflightRequest.current?.abort(); preflightRequest.current = null; setPreflight((current) => current.kind === "idle" ? current : { kind: "idle" }); - }, [selectedSetupId, sourceSessionId]); - - const selectedSetup = useMemo( - () => activeCatalog?.setups.find((setup) => setup.setupId === selectedSetupId) ?? null, - [activeCatalog, selectedSetupId], - ); + }, [selectedSetupId, sourceSessionId, selectedDefinitionSha256]); const selectSetup = useCallback((setupId: string) => { preflightRequest.current?.abort(); @@ -160,9 +119,10 @@ export function useObservatoryLaboratorySetups(sourceSessionId: string) { return { catalog: activeCatalog, + selectableSetups, state, error, - selectedSetupId, + selectedSetupId: selectedSetup?.setupId ?? "", selectedSetup, selectSetup, refresh, @@ -170,59 +130,3 @@ export function useObservatoryLaboratorySetups(sourceSessionId: string) { check, }; } - -function publishSetupCatalog( - next: ObservatoryLaboratorySetupCatalog, - setCatalog: (catalog: ObservatoryLaboratorySetupCatalog) => void, - setSelectedSetupId: (update: (current: string) => string) => void, - { preserveUnknownSelection = false }: { preserveUnknownSelection?: boolean } = {}, -): void { - setCatalog(next); - setSelectedSetupId((current) => { - if (next.setups.some( - (setup) => setup.setupId === current && setup.compatibility.compatible, - )) return current; - if (preserveUnknownSelection && current) 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 catalogErrorMessage(caught: unknown, fallback: string): string { - return caught instanceof Error && caught.message.trim() - ? caught.message - : fallback; -} - -function settled(promise: Promise): Promise> { - return promise.then( - (value) => ({ status: "fulfilled", value }), - (reason: unknown) => ({ status: "rejected", reason }), - ); -} - -function mergeSetupCatalogs( - legacy: ObservatoryLaboratorySetupCatalog, - portable: ObservatoryLaboratorySetupCatalog, -): ObservatoryLaboratorySetupCatalog { - if (legacy.sourceSessionId !== portable.sourceSessionId) { - throw new Error("Каталоги сетапов относятся к разным исходным сессиям."); - } - const portableSetupIds = new Set(portable.setups.map((setup) => setup.setupId)); - const setups = [ - ...legacy.setups.filter((setup) => !portableSetupIds.has(setup.setupId)), - ...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/core/observatory/useObservatoryRecordedJobs.ts b/apps/control-station/src/core/observatory/useObservatoryRecordedJobs.ts index 6a1a542..df5b325 100644 --- a/apps/control-station/src/core/observatory/useObservatoryRecordedJobs.ts +++ b/apps/control-station/src/core/observatory/useObservatoryRecordedJobs.ts @@ -1,4 +1,4 @@ -import { useCallback, useEffect, useMemo, useRef, useState } from "react"; +import { useCallback, useEffect, useRef, useState } from "react"; import { fetchObservatoryRecordedJobs, @@ -16,60 +16,83 @@ type RecordedJobsState = | "retrying-publication" | "error"; +interface JobSnapshot { + readonly selectionKey: string; + readonly jobs: readonly ObservatoryRecordedJob[]; + readonly state: RecordedJobsState; + readonly error: string | null; +} +const EMPTY_JOBS = [] as const; 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([]); - const [state, setState] = useState("idle"); - const [error, setError] = useState(null); +export function useObservatoryRecordedJobs( + sourceSessionId: string, + setupId: string, + definitionSha256: string, +) { + const selectionKey = JSON.stringify([sourceSessionId, setupId, definitionSha256]); + const [snapshot, setSnapshot] = useState({ + selectionKey: "", jobs: EMPTY_JOBS, state: "idle", error: null, + }); const [revision, setRevision] = useState(0); const requestRef = useRef(null); const requestSequence = useRef(0); const idempotencyKeys = useRef(new Map()); - const selectionKey = `${sourceSessionId}\u0000${setupId}`; + const observedJobId = useRef(null); + const hasSelection = Boolean(sourceSessionId && setupId && definitionSha256); + // A source/profile change must not display even one render of the previous queue. + const current = snapshot.selectionKey === selectionKey ? snapshot : null; + const jobs = current?.jobs ?? EMPTY_JOBS; + const state = current?.state ?? (hasSelection ? "loading" : "idle"); + const error = current?.error ?? null; useEffect(() => { requestRef.current?.abort(); requestRef.current = null; - if (!sourceSessionId || !setupId) { - setJobs([]); - setState("idle"); - setError(null); + if (!hasSelection) { + setSnapshot({ selectionKey, jobs: EMPTY_JOBS, state: "idle", error: null }); return; } const sequence = ++requestSequence.current; const request = new AbortController(); requestRef.current = request; - setState((current) => jobs.length > 0 && current !== "idle" ? "refreshing" : "loading"); - setError(null); - void fetchObservatoryRecordedJobs(sourceSessionId, setupId, { signal: request.signal }) - .then((next) => { - if (request.signal.aborted || requestSequence.current !== sequence) return; - setJobs(next); - setState("ready"); - }) - .catch((caught: unknown) => { - if (request.signal.aborted || requestSequence.current !== sequence) return; - setState("error"); - setError(caught instanceof Error && caught.message.trim() + setSnapshot({ + selectionKey, jobs, + state: jobs.length > 0 ? "refreshing" : "loading", error: null, + }); + void fetchObservatoryRecordedJobs(sourceSessionId, setupId, { + definitionSha256, signal: request.signal, + }).then((next) => { + if (request.signal.aborted || requestSequence.current !== sequence) return; + const pending = next.find((job) => OPEN_STATES.has(job.state)); + if (pending) observedJobId.current = pending.jobId; + setSnapshot({ selectionKey, jobs: next, state: "ready", error: null }); + }).catch((caught: unknown) => { + if (request.signal.aborted || requestSequence.current !== sequence) return; + setSnapshot({ + selectionKey, jobs, state: "error", + error: caught instanceof Error && caught.message.trim() ? caught.message - : "Очередь расчётов недоступна."); + : "Очередь расчётов недоступна.", }); + }).finally(() => { + if (requestRef.current === request) requestRef.current = null; + }); return () => request.abort(); }, [revision, selectionKey]); useEffect(() => () => requestRef.current?.abort(), []); const latestJob = jobs[0] ?? null; - const activeJob = useMemo( - () => jobs.find((job) => OPEN_STATES.has(job.state)) ?? null, - [jobs], - ); + const activeJob = jobs.find((job) => OPEN_STATES.has(job.state)) ?? null; const publicationPending = jobs.some((job) => job.publication.state === "pending"); + // Do not promote historical terminal logs to a new operator action's error. + const computationFailed = latestJob?.state === "failed" + && latestJob.jobId === observedJobId.current; useEffect(() => { if (state !== "ready" || (!activeJob && !publicationPending)) return; @@ -93,80 +116,76 @@ export function useObservatoryRecordedJobs(sourceSessionId: string, setupId: str readonly checkSha256: string; } | null = null, ): Promise => { - if (!sourceSessionId || !setupId || state === "submitting") return null; + if (!hasSelection || state !== "ready" + || portableBinding?.definitionSha256 !== definitionSha256 + || publicationPending || latestJob?.publication.state === "failed") return null; if (activeJob) return activeJob; requestRef.current?.abort(); const request = new AbortController(); requestRef.current = request; - const key = idempotencyKeys.current.get(selectionKey) - ?? createIdempotencyKey(); + const key = idempotencyKeys.current.get(selectionKey) ?? createIdempotencyKey(); idempotencyKeys.current.set(selectionKey, key); - setState("submitting"); - setError(null); + setSnapshot({ selectionKey, jobs, state: "submitting", error: null }); try { const job = await submitObservatoryRecordedJob( - sourceSessionId, - setupId, - key, - portableBinding, - { signal: request.signal }, + sourceSessionId, setupId, key, portableBinding, { signal: request.signal }, ); if (request.signal.aborted || requestRef.current !== request) return null; - setJobs((current) => [job, ...current.filter((candidate) => candidate.jobId !== job.jobId)]); - setState("ready"); + observedJobId.current = job.jobId; + setSnapshot({ + selectionKey, jobs: [job, ...jobs.filter((candidate) => candidate.jobId !== job.jobId)], + state: "ready", error: null, + }); return job; } catch (caught) { if (request.signal.aborted || requestRef.current !== request) return null; - setState("error"); - setError(caught instanceof Error && caught.message.trim() - ? caught.message - : "Не удалось поставить расчёт в очередь."); + setSnapshot({ + selectionKey, jobs, state: "error", + error: caught instanceof Error && caught.message.trim() + ? caught.message + : "Не удалось поставить расчёт в очередь.", + }); return null; } finally { if (requestRef.current === request) requestRef.current = null; } - }, [activeJob, selectionKey, setupId, sourceSessionId, state]); + }, [activeJob, definitionSha256, hasSelection, jobs, latestJob, + publicationPending, selectionKey, setupId, sourceSessionId, state]); const retryPublication = useCallback(async (): Promise => { - if (!latestJob || latestJob.publication.state !== "failed") return null; - if (state === "retrying-publication") return null; + if (!latestJob || latestJob.publication.state !== "failed" + || state === "retrying-publication") return null; requestRef.current?.abort(); const request = new AbortController(); requestRef.current = request; - setState("retrying-publication"); - setError(null); + setSnapshot({ selectionKey, jobs, state: "retrying-publication", error: 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"); + setSnapshot({ + selectionKey, jobs: [job, ...jobs.filter((candidate) => candidate.jobId !== job.jobId)], + state: "ready", error: null, + }); return job; } catch (caught) { if (request.signal.aborted || requestRef.current !== request) return null; - setState("error"); - setError(caught instanceof Error && caught.message.trim() - ? caught.message - : "Не удалось повторить публикацию результата."); + setSnapshot({ + selectionKey, jobs, state: "error", + error: caught instanceof Error && caught.message.trim() + ? caught.message + : "Не удалось повторить публикацию результата.", + }); return null; } finally { if (requestRef.current === request) requestRef.current = null; } - }, [latestJob, state]); + }, [jobs, latestJob, selectionKey, state]); return { - jobs, - latestJob, - activeJob, - state, - error, - refresh, - submit, - retryPublication, + jobs, latestJob, activeJob, publicationPending, computationFailed, + state, error, refresh, submit, retryPublication, }; } diff --git a/apps/control-station/src/styles/observatory.css b/apps/control-station/src/styles/observatory.css index 3399438..ea39255 100644 --- a/apps/control-station/src/styles/observatory.css +++ b/apps/control-station/src/styles/observatory.css @@ -6,6 +6,8 @@ min-width: 0; min-height: 100%; padding: 1rem; + container-name: observatory-workspace; + container-type: inline-size; } .observatory-lead, @@ -306,13 +308,6 @@ text-align: center; } -.observatory-evidence__bounded-note { - margin: 0; - color: var(--nodedc-text-muted); - font-size: var(--nodedc-font-size-xs); - line-height: 1.45; -} - .observatory-notice { justify-content: flex-start; color: var(--nodedc-text-secondary); @@ -373,7 +368,7 @@ } } -@media (max-width: 920px) { +@container observatory-workspace (max-width: 920px) { .observatory-lead, .observatory-catalog-bar, .observatory-catalog-bar__controls, @@ -382,12 +377,6 @@ flex-direction: column; } - .observatory-replay__header, - .observatory-replay-state { - align-items: stretch; - flex-direction: column; - } - .observatory-catalog-bar__controls { flex-basis: auto; } @@ -401,5 +390,12 @@ .observatory-catalog-bar__run { justify-content: flex-end; } +} +@media (max-width: 920px) { + .observatory-replay__header, + .observatory-replay-state { + align-items: stretch; + flex-direction: column; + } } diff --git a/apps/control-station/src/workspaces/observatory/ObservatoryWorkspace.tsx b/apps/control-station/src/workspaces/observatory/ObservatoryWorkspace.tsx index 8ea7ea7..fb81b24 100644 --- a/apps/control-station/src/workspaces/observatory/ObservatoryWorkspace.tsx +++ b/apps/control-station/src/workspaces/observatory/ObservatoryWorkspace.tsx @@ -33,10 +33,8 @@ import { import { useObservatoryCatalog } from "../../core/observatory/useObservatoryCatalog"; import { useObservatoryLaboratorySetups } from "../../core/observatory/useObservatoryLaboratorySetups"; import { useObservatoryRecordedJobs } from "../../core/observatory/useObservatoryRecordedJobs"; -import type { ObservatoryRecordedJobState } from "../../core/observatory/recordedJobs"; import type { WorkspaceDefinition } from "../../productModel"; -const MAX_PRESENTED_EVIDENCE = 6; const EMPTY_OBSERVATORY_ITEMS = [] as const; type ObservatoryRecordedRunReview = Awaited< ReturnType @@ -88,27 +86,6 @@ const modalityLabel: Record = { telemetry: "Телеметрия", }; -const recordedJobStatus: Record< - ObservatoryRecordedJobState, - { - readonly label: string; - readonly tone: "success" | "accent" | "warning" | "danger" | "neutral"; - } -> = { - accepted: { label: "Принят", tone: "neutral" }, - queued: { label: "Ждёт Worker", tone: "neutral" }, - claimed: { label: "Назначен Worker", tone: "accent" }, - running: { label: "Выполняется", tone: "accent" }, - paused: { label: "Пауза: live-поток", tone: "warning" }, - "preemption-pending": { - label: "Ждём подтверждения остановки", - tone: "warning", - }, - succeeded: { label: "Вычислено · ждёт публикации", tone: "warning" }, - failed: { label: "Ошибка расчёта", tone: "danger" }, - "reconciliation-required": { label: "Нужна сверка", tone: "warning" }, -}; - function statusTone( status: ObservationSessionStatus, ): "success" | "accent" | "warning" | "danger" | "neutral" { @@ -135,15 +112,6 @@ function formatDuration(seconds: number): string { : `${minutes}:${String(remainingSeconds).padStart(2, "0")}`; } -function catalogStateLabel( - state: ReturnType["state"], -): string { - if (state === "ready") return "Срез актуален"; - if (state === "refreshing") return "Обновляем каталог"; - if (state === "error") return "Каталог недоступен"; - return "Читаем каталог"; -} - function mutationErrorMessage(error: unknown): string { return error instanceof Error && error.message.trim() ? error.message @@ -166,6 +134,7 @@ export function ObservatoryWorkspace({ const recordedJobsController = useObservatoryRecordedJobs( selectedSessionId, setupController.selectedSetupId, + setupController.selectedSetup?.runDefinition?.definitionSha256 ?? "", ); const [replay, setReplay] = useState({ kind: "closed" }); const [renameTarget, setRenameTarget] = useState(null); @@ -202,30 +171,19 @@ export function ObservatoryWorkspace({ const selectedSession = items.find( (item) => item.source.id === selectedSessionId, ) ?? null; - const presentedEvidence = selectedSession?.evidence.slice( - 0, - MAX_PRESENTED_EVIDENCE, - ) ?? []; + const presentedEvidence = selectedSession?.evidence ?? []; const options = useMemo(() => items.map(({ source, evidence }) => ({ value: source.id, label: source.label, description: `${formatTimestamp(source.startedAtUtc)} · ${formatDuration(source.durationSeconds)} · ${evidence.length} результатов`, })), [items]); const setupOptions = useMemo(() => ( - setupController.catalog?.setups.map((setup) => ({ + setupController.selectableSetups.map((setup) => ({ value: setup.setupId, label: setup.displayName, - description: setup.compatibility.compatible - ? setup.origin === "existing-result" - ? "Готовый результат" - : setup.origin === "portable-definition" - ? setup.executor.state === "ready" - ? "Запись совместима · Worker готов к проверке" - : "Запись совместима · Worker не установлен" - : "Совместимый архивный сетап" - : "Несовместим с выбранной сессией", - })) ?? [] - ), [setupController.catalog]); + description: setup.description, + })) + ), [setupController.selectableSetups]); const preflightCandidate = setupController.preflight.kind === "ready" ? setupController.preflight.value : null; @@ -242,29 +200,29 @@ export function ObservatoryWorkspace({ && runPreflight?.outcome === "queueable" && runPreflight.submissionAllowed, ); - const presentedJob = recordedJobsController.activeJob - ?? recordedJobsController.latestJob; - const presentedJobStatus = presentedJob - ? 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" + const presentedJob = recordedJobsController.activeJob ?? recordedJobsController.latestJob; + const publicationFailed = presentedJob?.publication.state === "failed"; + const calculationPending = recordedJobsController.activeJob !== null + || recordedJobsController.publicationPending || recordedJobsController.state === "submitting" || recordedJobsController.state === "retrying-publication"; + const showCalculate = setupController.selectedSetup !== null + && !calculationPending && !publicationFailed; const canSubmitRecordedJob = queueSubmissionAllowed + && setupController.state === "ready" && recordedJobsController.state === "ready" - && recordedJobsController.activeJob === null; + && !calculationPending && !publicationFailed; + const queueStatusError = setupController.preflight.kind === "error" + ? setupController.preflight.message + : recordedJobsController.error + ?? (publicationFailed + ? presentedJob?.publication.error ?? "Не удалось опубликовать результат." + : recordedJobsController.computationFailed + ? "Не удалось выполнить расчёт. Можно повторить запуск." + : runPreflight?.outcome === "blocked" + ? runPreflight.checks.filter((check) => check.outcome === "fail") + .map((check) => check.message).join(" ") || "Запуск профиля недоступен." + : null); const initialLoading = !controller.catalog && ["idle", "loading"].includes(controller.state); const unavailable = !controller.catalog && controller.state === "error"; @@ -316,7 +274,8 @@ export function ObservatoryWorkspace({ setDeleteTarget(null); setMutationError(null); setMutationReconciliation(null); - }, [controller.catalog, mutationPending, mutationReconciliation]); + if (mutationReconciliation.kind === "delete") setupController.refresh(); + }, [controller.catalog, mutationPending, mutationReconciliation, setupController.refresh]); const openReplay = useCallback((binding: ObservatoryRecordedRunBinding) => { const attempt = replayCoordinatorRef.current.begin(); @@ -417,6 +376,7 @@ export function ObservatoryWorkspace({ } await deleteObservatoryLabProjection(deleteTarget.recordedRun); controller.applyEvidenceDeletion(deleteTarget.sessionId); + setupController.refresh(); setDeleteTarget(null); setMutationReconciliation(null); void controller.refresh(); @@ -437,7 +397,7 @@ export function ObservatoryWorkspace({ } finally { setMutationPending(null); } - }, [closeReplay, controller, deleteTarget, mutationPending, replay]); + }, [closeReplay, controller, deleteTarget, mutationPending, replay, setupController.refresh]); return (
- - {catalogStateLabel(controller.state)} - @@ -486,7 +441,7 @@ export function ObservatoryWorkspace({ disabled={!selectedSessionId || setupOptions.length === 0} searchable searchPlaceholder="Поиск по сетапам" - emptyLabel={setupController.state === "error" ? "Каталог сетапов недоступен" : "Сетап не найден"} + emptyLabel={setupController.state === "error" ? "Каталог профилей недоступен" : "Нет профилей для расчёта"} minMenuWidth={360} menuWidth={500} onChange={setupController.selectSetup} @@ -509,47 +464,15 @@ export function ObservatoryWorkspace({ > Обновить -
- {queueStatusError ? ( - - Очередь недоступна - - ) : recordedJobsController.state === "submitting" ? ( - Ставим в очередь - ) : recordedJobsController.state === "retrying-publication" ? ( - Повторяем публикацию - ) : presentedJobStatus ? ( - - {presentedJobStatus.label} - - ) : setupController.preflight.kind === "checking" ? ( - Проверяем сетап - ) : queueStateBusy ? ( - Читаем очередь - ) : setupController.selectedSetup?.origin === "portable-definition" - && runPreflight ? ( - - {setupController.selectedSetup.compatibility.compatible - ? setupController.selectedSetup.executor.state === "not-installed" - ? "Worker-профиль не установлен" - : "Запуск профиля недоступен" - : "Запись несовместима"} - +
+ {calculationPending ? ( + ) : null} - {canSubmitRecordedJob ? ( + {showCalculate ? ( ) : null} - {presentedJob?.publication.state === "failed" - && recordedJobsController.state !== "retrying-publication" ? ( - - ) : null}
+ {queueStatusError ? ( + + {queueStatusError} + {publicationFailed ? ( + + ) : null} + + ) : null} + {controller.error && controller.catalog ? ( Показан последний срез @@ -588,9 +517,8 @@ export function ObservatoryWorkspace({ ) : null} - {setupController.error && setupController.catalog ? ( + {setupController.error ? ( - Показан последний каталог сетапов {setupController.error} @@ -700,9 +628,6 @@ export function ObservatoryWorkspace({ {evidence.lab.resultKind} · {formatTimestamp(evidence.publishedAtUtc)} - {evidence.recordedRun ? ( - Записанный разбор - ) : null}
{evidence.recordedRun ? ( <> @@ -748,12 +673,6 @@ export function ObservatoryWorkspace({

)} - {selectedSession.evidence.length > presentedEvidence.length ? ( -

- Показаны {presentedEvidence.length} последних из {selectedSession.evidence.length} - {" "}связанных результатов. Полный архив остаётся в legacy LAB. -

- ) : null} ) : null} diff --git a/apps/control-station/test/observatoryHooks.test.mjs b/apps/control-station/test/observatoryHooks.test.mjs new file mode 100644 index 0000000..cdbf627 --- /dev/null +++ b/apps/control-station/test/observatoryHooks.test.mjs @@ -0,0 +1,96 @@ +import assert from "node:assert/strict"; +import React from "react"; +import { after, before, test } from "node:test"; +import { createServer } from "vite"; + +let server; +let useObservatoryRecordedJobs; +let useObservatoryLaboratorySetups; + +before(async () => { + server = await createServer({ + appType: "custom", logLevel: "silent", server: { middlewareMode: true }, + }); + ({ useObservatoryRecordedJobs } = await server.ssrLoadModule("/src/core/observatory/useObservatoryRecordedJobs.ts")); + ({ useObservatoryLaboratorySetups } = await server.ssrLoadModule("/src/core/observatory/useObservatoryLaboratorySetups.ts")); +}); +after(async () => { await server?.close(); }); + +// Deliberately inspect the render BEFORE effects/cleanup: stale state must already be hidden. +function renderBeforeEffects(hook, args, stateSlots) { + const internals = React.__CLIENT_INTERNALS_DO_NOT_USE_OR_WARN_USERS_THEY_CANNOT_UPGRADE; + const previous = internals.H; + let stateIndex = 0; + internals.H = { + useState: (initial) => [ + stateIndex < stateSlots.length ? stateSlots[stateIndex++] + : typeof initial === "function" ? initial() : initial, + () => {}, + ], + useRef: (value) => ({ current: value }), + useMemo: (factory) => factory(), + useCallback: (fn) => fn, + useEffect: () => {}, + }; + try { return hook(...args); } finally { internals.H = previous; } +} + +test("queue data and errors cannot survive a source/setup/version change for one render", () => { + const original = ["source-a", "profile-a", "a".repeat(64)]; + const activeJob = { state: "running", publication: { state: "not-required" } }; + const snapshot = { + selectionKey: JSON.stringify(original), jobs: [activeJob], state: "ready", error: "old error", + }; + assert.equal(renderBeforeEffects(useObservatoryRecordedJobs, original, [snapshot]).activeJob, activeJob); + for (const next of [ + ["source-b", original[1], original[2]], + [original[0], "profile-b", original[2]], + [original[0], original[1], "b".repeat(64)], + ["", "", ""], + ]) { + const view = renderBeforeEffects(useObservatoryRecordedJobs, next, [snapshot]); + assert.deepEqual(view.jobs, []); + assert.equal(view.activeJob, null); + assert.equal(view.latestJob, null); + assert.equal(view.error, null); + assert.notEqual(view.state, "ready"); + } +}); + +test("published, incompatible and previous-source profiles have no effective selection", () => { + const profile = { + setupId: "profile-a", origin: "portable-definition", + compatibility: { compatible: true }, + runDefinition: { definitionSha256: "a".repeat(64) }, + preflight: { outcome: "ready" }, + }; + function render(sourceId, setups) { + return renderBeforeEffects(useObservatoryLaboratorySetups, [sourceId], [ + { sourceSessionId: "source-a", setups }, "ready", null, "profile-a", 0, { kind: "idle" }, + ]); + } + assert.equal(render("source-a", [profile]).selectedSetupId, "profile-a"); + for (const view of [ + render("source-b", [profile]), + render("source-a", [{ ...profile, preflight: { outcome: "existing" } }]), + render("source-a", [{ ...profile, compatibility: { compatible: false } }]), + render("source-a", []), + ]) { + assert.equal(view.selectedSetupId, ""); + assert.equal(view.selectedSetup, null); + assert.deepEqual(view.selectableSetups, []); + } +}); + +test("historical failures are not presented as errors of a new operator action", () => { + const selection = ["source-a", "profile-a", "a".repeat(64)]; + const snapshot = { + selectionKey: JSON.stringify(selection), + jobs: [{ jobId: "old-run", state: "failed", publication: { state: "not-required" } }], + state: "ready", error: null, + }; + const view = renderBeforeEffects(useObservatoryRecordedJobs, selection, [snapshot]); + assert.equal(view.computationFailed, false); + assert.equal(view.error, null); + assert.equal(view.activeJob, null); +}); diff --git a/apps/control-station/test/observatoryLaboratorySetups.test.mjs b/apps/control-station/test/observatoryLaboratorySetups.test.mjs index 1377ebf..48b9c8f 100644 --- a/apps/control-station/test/observatoryLaboratorySetups.test.mjs +++ b/apps/control-station/test/observatoryLaboratorySetups.test.mjs @@ -7,6 +7,7 @@ let server; let fetchObservatoryLaboratorySetups; let fetchObservatoryPortableLaboratorySetups; let preflightObservatoryLaboratorySetup; +let selectableObservatorySetups; let ObservatoryLaboratorySetupContractError; const authority = { @@ -168,6 +169,7 @@ before(async () => { fetchObservatoryLaboratorySetups, fetchObservatoryPortableLaboratorySetups, preflightObservatoryLaboratorySetup, + selectableObservatorySetups, ObservatoryLaboratorySetupContractError, } = await server.ssrLoadModule("/src/core/observatory/laboratorySetups.ts")); }); @@ -396,6 +398,38 @@ test("portable exact cached result is readable without an installed executor", a assert.equal(setup.preservedResults[0].access, "observatory"); }); +test("selector contains only compatible uncalculated portable definitions", async () => { + const uncached = (await fetchCachedPortable(portableSetup())).setups[0]; + const cached = (await fetchCachedPortable(cachedPortableSetup())).setups[0]; + const currentVersion = { + ...uncached, + runDefinition: { ...uncached.runDefinition, version: 2, definitionSha256: "f".repeat(64) }, + }; + const catalog = (setups) => ({ sourceSessionId: "source-a", setups }); + assert.deepEqual(selectableObservatorySetups(null), []); + assert.deepEqual(selectableObservatorySetups(catalog([])), []); + assert.deepEqual(selectableObservatorySetups(catalog([cached])), []); + // Same display label and setup ID, but no verified cache hit for the new definition. + assert.deepEqual(selectableObservatorySetups(catalog([currentVersion])), [currentVersion]); + assert.deepEqual(selectableObservatorySetups(catalog([ + cached, + { ...uncached, setupId: "remaining-profile" }, + ])).map((s) => s.setupId), ["remaining-profile"]); + assert.deepEqual(selectableObservatorySetups(catalog([ + { ...uncached, compatibility: { compatible: false, reasons: [] } }, + { ...uncached, origin: "archived-definition" }, + { ...uncached, origin: "existing-result" }, + { ...uncached, runDefinition: null }, + ])), []); + // Worker availability does not masquerade as completed computation. + assert.equal(uncached.executor.state, "not-installed"); + assert.deepEqual(selectableObservatorySetups(catalog([uncached])), [uncached]); + // A missing/deleted/unpublished projection is not a completed result. + assert.deepEqual(selectableObservatorySetups(catalog([{ + ...uncached, preservedResults: cached.preservedResults, + }])).map((s) => s.setupId), [uncached.setupId]); +}); + for (const [label, change] of [ ["another source", (s) => { s.existing_results[0].identity.source_session_id = "source-b"; }], ["another setup", (s) => { s.existing_results[0].identity.setup_id = "another-profile"; }], diff --git a/apps/control-station/test/observatoryRecordedJobs.test.mjs b/apps/control-station/test/observatoryRecordedJobs.test.mjs index 7e36df2..f241ce5 100644 --- a/apps/control-station/test/observatoryRecordedJobs.test.mjs +++ b/apps/control-station/test/observatoryRecordedJobs.test.mjs @@ -245,3 +245,33 @@ test("publication retry never submits a second compute request", async () => { assert.equal(request.init.body, undefined); assert.equal(retried.publication.state, "published"); }); + +test("queue query and response are fenced to the selected definition", async () => { + const oldVersion = job("succeeded"); + oldVersion.setup.definition_sha256 = "f".repeat(64); + const otherSource = job("running"); + otherSource.source.session_id = "source-b"; + const otherSetup = job("running"); + otherSetup.setup.setup_id = "other-setup"; + let url; + const jobs = await fetchObservatoryRecordedJobs("source-a", "m49-tgs", { + definitionSha256: "8".repeat(64), + fetcher: async (input) => { + url = new URL(String(input), "http://localhost"); + return new Response(JSON.stringify({ + schema_version: "missioncore.observatory-recorded-job-list/v1", + items: [oldVersion, otherSource, otherSetup, job("queued")], authority, + })); + }, + }); + assert.equal(url.searchParams.get("definition_sha256"), "8".repeat(64)); + assert.deepEqual(jobs.map((j) => j.state), ["queued"]); +}); + +test("portable submit rejects a successful response for an old definition", async () => { + await assert.rejects(submitObservatoryRecordedJob( + "source-a", "m49-tgs", "test-request", + { definitionSha256: "f".repeat(64), checkSha256: "e".repeat(64) }, + { fetcher: async () => new Response(JSON.stringify(job("succeeded"))) }, + ), ObservatoryRecordedJobContractError); +}); diff --git a/apps/control-station/test/observatoryWorkspace.test.mjs b/apps/control-station/test/observatoryWorkspace.test.mjs index c547d6c..87ee327 100644 --- a/apps/control-station/test/observatoryWorkspace.test.mjs +++ b/apps/control-station/test/observatoryWorkspace.test.mjs @@ -75,8 +75,8 @@ test("Observatory mounts the one shared canonical replay only after explicit adm assert.match(workspace, /useObservatoryCatalog/); assert.match(workspace, /Связанных результатов нет/); assert.match(workspace, /не является выводом о качестве/); - assert.match(workspace, /\.evidence\.slice\([\s\S]*MAX_PRESENTED_EVIDENCE/); - assert.match(workspace, /Полный архив остаётся в legacy LAB/); + assert.match(workspace, /presentedEvidence = selectedSession\?\.evidence \?\? \[\]/); + assert.doesNotMatch(workspace, /MAX_PRESENTED_EVIDENCE|\.evidence\.slice\(/); assert.match(workspace, /Полнота исторических/); assert.match(workspace, /вне текущего загруженного среза/); assert.match(workspace, /observatory-notice__copy/); @@ -136,7 +136,7 @@ test("Observatory mounts the one shared canonical replay only after explicit adm assert.match(viewerProfiles, /kind: "lab-recorded-evidence"/); assert.match( styles, - /@media \(max-width: 920px\)[\s\S]*\.observatory-catalog-bar__controls \.nodedc-select-anchor \{[\s\S]*flex: 0 0 auto;/, + /@container observatory-workspace \(max-width: 920px\)[\s\S]*\.observatory-catalog-bar__controls \.nodedc-select-anchor \{[\s\S]*flex: 0 0 auto;/, ); assert.match( styles, @@ -229,7 +229,7 @@ test("Observatory keeps one compact selector axis without the obsolete setup det assert.match(workspace, /className="observatory-catalog-bar" padding="sm"/); assert.match(workspace, /label="Выбрать сохранённую сессию"/); assert.match(workspace, /label="Выбрать сетап лаборатории"/); - assert.match(workspace, /Показан последний каталог сетапов/); + assert.match(workspace, /setupController\.error \? \(/); assert.doesNotMatch(workspace, /Выбор меняет только читаемую карточку/); assert.doesNotMatch(workspace, /ObservatorySetupDetail|observatory-setup-detail/); assert.doesNotMatch(styles, /observatory-setup-detail|observatory-setup-results/); @@ -240,44 +240,31 @@ test("Observatory keeps one compact selector axis without the obsolete setup det ); assert.match( workspace, - /canSubmitRecordedJob \? \([\s\S]*>\s*Рассчитать\s*<\/Button>/, + /showCalculate \? \([\s\S]*>\s*Рассчитать\s*<\/Button>/, ); - assert.match(workspace, /accepted: \{ label: "Принят"/); - assert.match(workspace, /queued: \{ label: "Ждёт Worker"/); - assert.match(workspace, /claimed: \{ label: "Назначен Worker"/); - assert.match(workspace, /running: \{ label: "Выполняется"/); - assert.match(workspace, /paused: \{ label: "Пауза: live-поток"/); - assert.match( - workspace, - /"preemption-pending": \{[\s\S]*label: "Ждём подтверждения остановки"/, - ); - assert.match(setupHook, /Promise\.all\(\[legacyResult, portableResult\]\)/); - assert.match( - setupHook, - /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 готов к проверке/); + assert.doesNotMatch(workspace, /recordedJobStatus|presentedJobStatus|Расчёт завершён|Результат опубликован|Вычислено ·/); + assert.match(workspace, /setupController\.selectableSetups\.map/); + assert.match(workspace, /showCalculate = setupController\.selectedSetup !== null/); + assert.match(workspace, /disabled=\{!canSubmitRecordedJob\}/); + assert.match(workspace, /aria-busy=\{calculationPending\}/); + const runBar = workspace.slice(workspace.indexOf('
tuple[ObservatoryRecordedJob, ...]: if source_session_id is not None: _validate_pattern(source_session_id, _SESSION_ID, "source session id") if setup_id is not None: _validate_pattern(setup_id, _IDENTIFIER, "setup id") + if definition_sha256 is not None: + _validate_pattern(definition_sha256, _SHA256, "definition sha256") if not isinstance(limit, int) or isinstance(limit, bool) or not 1 <= limit <= 500: raise ValueError("recorded-job list limit is invalid") clauses: list[str] = [] @@ -2096,6 +2099,9 @@ class ObservatoryRecordedJobQueue: if setup_id is not None: clauses.append("setup_id = ?") parameters.append(setup_id) + if definition_sha256 is not None: + clauses.append("definition_sha256 = ?") + parameters.append(definition_sha256) where = f" WHERE {' AND '.join(clauses)}" if clauses else "" parameters.append(limit) with self._read_connection() as connection: diff --git a/src/k1link/web/observatory_api.py b/src/k1link/web/observatory_api.py index e3ff871..b002218 100644 --- a/src/k1link/web/observatory_api.py +++ b/src/k1link/web/observatory_api.py @@ -1131,11 +1131,13 @@ def build_observatory_router( pattern=r"^[a-z][a-z0-9-]{2,95}$", ), limit: int = Query(default=20, ge=1, le=100), + definition_sha256: str | None = Query(default=None, pattern=r"^[a-f0-9]{64}$"), ) -> dict[str, object]: try: jobs = recorded_job_queue.list_jobs( source_session_id=source_session_id, setup_id=setup_id, + definition_sha256=definition_sha256, limit=limit, ) except (ObservatoryRecordedQueueError, ValueError) as exc: diff --git a/tests/test_observatory_recorded_jobs.py b/tests/test_observatory_recorded_jobs.py index 43a8783..75726a4 100644 --- a/tests/test_observatory_recorded_jobs.py +++ b/tests/test_observatory_recorded_jobs.py @@ -202,6 +202,31 @@ def _reconciliation_request( ) +def test_list_filters_definition_before_page_limit(tmp_path: Path) -> None: + previous = _definitions().definitions[0] + current = replace(previous, definition_version=2, definition_sha256="0" * 64) + now = NOW + queue = ObservatoryRecordedJobQueue( + tmp_path, definitions=RecordedRunDefinitionRegistry((previous, current)), + clock=lambda: now, + ) + first, _ = queue.submit(_intent()) + now = "2026-08-30T21:01:00.000Z" + queue.submit( + replace( + _intent(idempotency_key="new-definition"), + definition_sha256=current.definition_sha256, + ), + ) + assert queue.list_jobs(limit=1)[0].definition_sha256 == current.definition_sha256 + assert queue.list_jobs( + source_session_id=first.source_session_id, setup_id=previous.setup_id, + definition_sha256=previous.definition_sha256, limit=1, + ) == (first,) + with pytest.raises(ValueError): + queue.list_jobs(definition_sha256="not-a-digest") + + @pytest.mark.parametrize("enqueue", [False, True]) def test_portable_duplicate_guard_preserves_original_request( tmp_path: Path, diff --git a/tests/test_observatory_recorded_run_api.py b/tests/test_observatory_recorded_run_api.py index d514e38..67f8c2e 100644 --- a/tests/test_observatory_recorded_run_api.py +++ b/tests/test_observatory_recorded_run_api.py @@ -228,6 +228,25 @@ def test_exact_m49_preflight_is_queueable_and_submit_is_idempotently_queued( ] assert queue.admission_gate().blocked is False + for digest, count in ((definition.definition_sha256, 1), ("f" * 64, 0)): + exact = client.get( + "/api/v1/observatory/runs", + params={ + "source_session_id": RAV00_SESSION_ID, "setup_id": SETUP_ID, + "definition_sha256": digest, + }, + ) + assert exact.status_code == 200 + assert len(exact.json()["items"]) == count + malformed = client.get( + "/api/v1/observatory/runs", + params={ + "source_session_id": RAV00_SESSION_ID, "setup_id": SETUP_ID, + "definition_sha256": "not-a-digest", + }, + ) + assert malformed.status_code == 422 + def test_recorded_run_routes_fail_closed_when_queue_initialization_failed() -> None: registry = LaboratorySetupRegistry.from_file(