feat(lab): publish M4.8T quality evidence

This commit is contained in:
DCCONSTRUCTIONS
2026-08-25 23:44:27 +03:00
parent 9a956c318a
commit 3ff3b9b587
20 changed files with 1419 additions and 4 deletions
@@ -43,11 +43,13 @@ import {
} from "./m48ObjectCentricQuality";
import { fetchM48SmallStaticRegression } from "./m48SmallStaticRegression";
import { fetchM48SFixedClassDetectorResult } from "./m48sFixedClassDetector";
import { fetchM48TRiskQualityResult } from "./m48tRiskQuality";
export type AdvancedLaboratoryWorkId =
| "m48-object-centric-quality"
| "m48-small-static-passage-regression"
| "m48s-fixed-class-detector"
| "m48t-risk-quality-temporal"
| "m47-reference-graph-shadow"
| "m4-replay-threat"
| "l3-pointpillars-visual-audit"
@@ -93,6 +95,7 @@ const WORK_IDS: readonly AdvancedLaboratoryWorkId[] = [
"m48-object-centric-quality",
"m48-small-static-passage-regression",
"m48s-fixed-class-detector",
"m48t-risk-quality-temporal",
"m47-reference-graph-shadow",
"m4-replay-threat",
"l3-pointpillars-visual-audit",
@@ -133,6 +136,7 @@ const RESULT_PREFIX: Readonly<Record<AdvancedLaboratoryWorkId, string>> = {
"m48-object-centric-quality": "m48-object-quality-(?:pack|result)",
"m48-small-static-passage-regression": "m48-small-static-passage-regression",
"m48s-fixed-class-detector": "m48s-fixed-class-detector-lab",
"m48t-risk-quality-temporal": "m48t-risk-quality-temporal-lab",
"m47-reference-graph-shadow": "m47-reference-graph-lab",
"m4-replay-threat": "m4-threat-replay",
"l3-pointpillars-visual-audit": "l3-pointpillars-visual-audit",
@@ -181,6 +185,7 @@ export function emptyAdvancedLaboratoryResults(): AdvancedLaboratoryResults {
m48: null,
m48SmallStatic: null,
m48s: null,
m48t: null,
m4Threat: null,
l3: null,
l31: null,
@@ -308,6 +313,7 @@ export function advancedLaboratoryResultAvailable(
return workId === "m48-object-centric-quality" ? results.m48 !== null
: workId === "m48-small-static-passage-regression" ? results.m48SmallStatic !== null
: workId === "m48s-fixed-class-detector" ? results.m48s !== null
: workId === "m48t-risk-quality-temporal" ? results.m48t !== null
: workId === "m47-reference-graph-shadow" ? results.m47Graph !== null
: workId === "m4-replay-threat" ? results.m4Threat !== null
: workId === "l3-pointpillars-visual-audit" ? results.l3 !== null
@@ -366,6 +372,9 @@ export async function fetchAdvancedLaboratoryResult(
} else if (workId === "m48s-fixed-class-detector") {
if (!resultId) throw new AdvancedLaboratoryContractError("M4.8S LAB identity не выбрана.");
results.m48s = await fetchM48SFixedClassDetectorResult(resultId, { fetcher, signal });
} else if (workId === "m48t-risk-quality-temporal") {
if (!resultId) throw new AdvancedLaboratoryContractError("M4.8T LAB identity не выбрана.");
results.m48t = await fetchM48TRiskQualityResult(resultId, { fetcher, signal });
} else if (workId === "m47-reference-graph-shadow") {
if (!resultId) {
throw new AdvancedLaboratoryContractError("M4.7 LAB identity не выбрана.");
@@ -37,12 +37,14 @@ import type { M47ReferenceGraphLabResult } from "./m47ReferenceGraph";
import type { M48AdvancedResult } from "./m48ObjectCentricQuality";
import type { M48SmallStaticRegressionResult } from "./m48SmallStaticRegression";
import type { M48SFixedClassDetectorResult } from "./m48sFixedClassDetector";
import type { M48TRiskQualityResult } from "./m48tRiskQuality";
export interface AdvancedLaboratoryResults {
m47Graph: M47ReferenceGraphLabResult | null;
m48: M48AdvancedResult | null;
m48SmallStatic: M48SmallStaticRegressionResult | null;
m48s: M48SFixedClassDetectorResult | null;
m48t: M48TRiskQualityResult | null;
m4Threat: M4ThreatReplayResult | null;
l3: L3PointPillarsVisualAuditResult | null;
l31: L31PointPillarsRavnovesResult | null;
@@ -967,7 +967,7 @@ export async function fetchAdvancedLaboratoryResults({
const e39 = settledCatalogValue(settled[7]);
const e40 = settledCatalogValue(settled[8]);
return {
m47Graph: null, m48: null, m48SmallStatic: null, m48s: null, m4Threat: null,
m47Graph: null, m48: null, m48SmallStatic: null, m48s: null, m48t: null, m4Threat: null,
l3: null, l31: null,
l32: null,
l33: null,
@@ -0,0 +1,326 @@
export interface M48TLaboratoryMethod {
completeness: "complete";
executionClass: "hybrid";
pipelineId: string;
components: readonly {
kind: "source" | "model" | "algorithm" | "runtime" | "tool";
name: string;
version: string;
role: string;
identitySha256: string;
}[];
}
export interface M48TReviewCase {
caseId: string;
imageId: number;
imageUrl: string;
byteLength: number;
sha256: string;
}
export interface M48TRiskQualityResult {
resultId: string;
createdAtUtc: string;
source: {
quality: {
datasetId: "coco-2017-val";
riskImages: number;
truthInstances: number;
independentHumanAnnotations: true;
ravnovesGroundTruth: false;
};
temporal: {
sourceId: "RAVNOVES00";
frames: number;
publications: number;
independentSemanticTruthAvailable: false;
};
};
candidate: {
providerId: string;
modelId: "rf_detr_large:1";
minimumScore: 0.25;
};
execution: {
durationSeconds: number;
effectiveImagesPerSecond: number;
imageP95Ms: number;
inferenceP95Ms: number;
timingIsAdmissionEvidence: false;
};
quality: {
microPrecision: number;
microRecall: number;
mediumLargeRecall: number;
emptyRiskImageFraction: number;
families: Readonly<Record<"person" | "animal" | "light-road-user" | "vehicle", number>>;
};
counts: {
truth: number;
predictions: number;
truePositive: number;
falsePositive: number;
falseNegative: number;
};
failures: Readonly<Record<string, number>>;
temporal: {
rawClassSwitches: number;
stableClassSwitches: number;
rawFamilySwitches: number;
stableFamilySwitches: number;
suppressedClassSwitches: number;
suppressedFamilySwitches: number;
selectedPublications: number;
semanticObservations: number;
peakActiveComponents: number;
maximumActiveComponents: number;
};
acceptance: {
qualityPassed: false;
failedQualityGates: readonly [
"minimum-micro-precision",
"minimum-family-recall:vehicle",
];
temporalInvariantPassed: true;
};
review: {
truthColor: "green";
predictionColor: "yellow";
cases: readonly M48TReviewCase[];
};
method: M48TLaboratoryMethod;
limitations: readonly string[];
}
type LaboratoryFetch = (input: RequestInfo | URL, init?: RequestInit) => Promise<Response>;
export class M48TContractError extends Error {}
function object(value: unknown, label: string): Record<string, unknown> {
if (!value || typeof value !== "object" || Array.isArray(value)) {
throw new M48TContractError(`${label}: ожидался объект.`);
}
return value as Record<string, unknown>;
}
function array(value: unknown, label: string): readonly unknown[] {
if (!Array.isArray(value)) throw new M48TContractError(`${label}: ожидался массив.`);
return value;
}
function text(value: unknown, label: string): string {
if (typeof value !== "string" || !value.trim()) {
throw new M48TContractError(`${label}: ожидалась строка.`);
}
return value;
}
function number(value: unknown, label: string): number {
if (typeof value !== "number" || !Number.isFinite(value) || value < 0) {
throw new M48TContractError(`${label}: ожидалось неотрицательное число.`);
}
return value;
}
function integer(value: unknown, label: string): number {
const parsed = number(value, label);
if (!Number.isSafeInteger(parsed)) {
throw new M48TContractError(`${label}: ожидалось целое число.`);
}
return parsed;
}
function exact<T extends string | number | boolean>(
value: unknown,
expected: T,
label: string,
): T {
if (value !== expected) throw new M48TContractError(`${label}: нарушен контракт.`);
return expected;
}
function sha(value: unknown, label: string): string {
const parsed = text(value, label);
if (!/^[a-f0-9]{64}$/.test(parsed)) {
throw new M48TContractError(`${label}: нарушена SHA-256 идентичность.`);
}
return parsed;
}
function method(value: unknown): M48TLaboratoryMethod {
const raw = object(value, "M4.8T method");
exact(raw.schema_version, "missioncore.laboratory-method/v1", "M4.8T method schema");
exact(raw.completeness, "complete", "M4.8T method completeness");
exact(raw.execution_class, "hybrid", "M4.8T method execution");
const allowedKinds = new Set(["source", "model", "algorithm", "runtime", "tool"]);
return {
completeness: "complete",
executionClass: "hybrid",
pipelineId: text(raw.pipeline_id, "M4.8T pipeline"),
components: array(raw.components, "M4.8T components").map((value) => {
const component = object(value, "M4.8T component");
const kind = text(component.kind, "M4.8T component kind");
if (!allowedKinds.has(kind)) throw new M48TContractError("M4.8T component kind: неизвестное значение.");
return {
kind: kind as "source" | "model" | "algorithm" | "runtime" | "tool",
name: text(component.name, "M4.8T component name"),
version: text(component.version, "M4.8T component version"),
role: text(component.role, "M4.8T component role"),
identitySha256: sha(component.identity_sha256, "M4.8T component identity"),
};
}),
};
}
function parseResult(value: unknown, expectedResultId: string): M48TRiskQualityResult {
const raw = object(value, "M4.8T result");
exact(raw.schema_version, "missioncore.m48t-risk-quality-temporal-view/v1", "M4.8T schema");
exact(raw.result_id, expectedResultId, "M4.8T identity");
exact(raw.status, "complete-quality-gate-failed-temporal-invariant-passed", "M4.8T status");
exact(raw.access, "read-only", "M4.8T access");
exact(raw.ground_truth, false, "M4.8T ground truth");
const source = object(raw.source, "M4.8T source");
const qualitySource = object(source.quality, "M4.8T quality source");
const temporalSource = object(source.temporal, "M4.8T temporal source");
const configuration = object(raw.configuration, "M4.8T configuration");
const candidate = object(configuration.candidate, "M4.8T candidate");
const execution = object(raw.execution, "M4.8T execution");
const imageTiming = object(execution.image_timing_ms, "M4.8T image timing");
const inferenceTiming = object(execution.triton_inference_ms, "M4.8T inference timing");
const metrics = object(raw.metrics, "M4.8T metrics");
const quality = object(metrics.quality, "M4.8T quality metrics");
const families = object(quality.families, "M4.8T family metrics");
const familyRecall = (name: "person" | "animal" | "light-road-user" | "vehicle") =>
number(object(families[name], `M4.8T ${name}`).family_recall, `M4.8T ${name} recall`);
const counts = object(metrics.counts, "M4.8T counts");
const failures = object(metrics.failure_buckets, "M4.8T failures");
const temporal = object(metrics.temporal, "M4.8T temporal metrics");
const snapshot = object(temporal.snapshot, "M4.8T temporal snapshot");
const acceptance = object(raw.acceptance, "M4.8T acceptance");
const qualityGate = object(acceptance.quality, "M4.8T quality gate");
const temporalGate = object(acceptance.temporal_invariant, "M4.8T temporal gate");
const failed = array(qualityGate.failed, "M4.8T failed gates").map((item) => text(item, "M4.8T failed gate"));
if (failed.length !== 2 || failed[0] !== "minimum-micro-precision" || failed[1] !== "minimum-family-recall:vehicle") {
throw new M48TContractError("M4.8T failed gates: изменён зафиксированный результат.");
}
exact(qualityGate.passed, false, "M4.8T quality gate");
exact(temporalGate.passed, true, "M4.8T temporal invariant");
exact(temporalGate.semantic_quality_accepted, false, "M4.8T temporal semantic acceptance");
const review = object(raw.review, "M4.8T review");
const legend = object(review.legend, "M4.8T review legend");
exact(legend.ground_truth, "green", "M4.8T truth legend");
exact(legend.rf_detr_prediction, "yellow", "M4.8T prediction legend");
const cases = array(review.cases, "M4.8T review cases").map((value) => {
const item = object(value, "M4.8T review case");
const caseId = text(item.case_id, "M4.8T case id");
if (!/^[0-9]{12}$/.test(caseId)) throw new M48TContractError("M4.8T case identity нарушена.");
exact(item.media_type, "image/jpeg", "M4.8T case media");
return {
caseId,
imageId: integer(item.image_id, "M4.8T image id"),
imageUrl: text(item.image_url, "M4.8T image URL"),
byteLength: integer(item.byte_length, "M4.8T image bytes"),
sha256: sha(item.sha256, "M4.8T image SHA"),
};
});
if (cases.length !== 16 || new Set(cases.map(({ caseId }) => caseId)).size !== 16) {
throw new M48TContractError("M4.8T review catalog: нарушен размер.");
}
const parsedFailures: Record<string, number> = {};
for (const [key, count] of Object.entries(failures)) parsedFailures[key] = integer(count, `M4.8T failure ${key}`);
const temporalConfiguration = object(configuration.temporal, "M4.8T temporal configuration");
return {
resultId: expectedResultId,
createdAtUtc: text(raw.created_at_utc, "M4.8T created at"),
source: {
quality: {
datasetId: exact(qualitySource.dataset_id, "coco-2017-val", "M4.8T dataset"),
riskImages: integer(qualitySource.risk_images, "M4.8T risk images"),
truthInstances: integer(qualitySource.truth_instances, "M4.8T truth instances"),
independentHumanAnnotations: exact(qualitySource.independent_human_annotations, true, "M4.8T independent truth"),
ravnovesGroundTruth: exact(qualitySource.ravnoves_ground_truth, false, "M4.8T RAVNOVES truth"),
},
temporal: {
sourceId: exact(temporalSource.source_id, "RAVNOVES00", "M4.8T temporal source"),
frames: integer(temporalSource.frames, "M4.8T temporal frames"),
publications: integer(temporalSource.publications, "M4.8T temporal publications"),
independentSemanticTruthAvailable: exact(temporalSource.independent_semantic_truth_available, false, "M4.8T temporal truth"),
},
},
candidate: {
providerId: text(candidate.provider_id, "M4.8T provider"),
modelId: exact(candidate.model_id, "rf_detr_large:1", "M4.8T model"),
minimumScore: exact(candidate.minimum_score, 0.25, "M4.8T threshold"),
},
execution: {
durationSeconds: number(execution.duration_seconds, "M4.8T duration"),
effectiveImagesPerSecond: number(execution.effective_images_per_second, "M4.8T throughput"),
imageP95Ms: number(imageTiming.p95, "M4.8T image p95"),
inferenceP95Ms: number(inferenceTiming.p95, "M4.8T inference p95"),
timingIsAdmissionEvidence: exact(execution.timing_is_admission_evidence, false, "M4.8T timing authority"),
},
quality: {
microPrecision: number(quality.micro_precision, "M4.8T precision"),
microRecall: number(quality.micro_recall, "M4.8T recall"),
mediumLargeRecall: number(quality.medium_large_recall, "M4.8T medium-large recall"),
emptyRiskImageFraction: number(quality.empty_prediction_risk_image_fraction, "M4.8T empty fraction"),
families: {
person: familyRecall("person"),
animal: familyRecall("animal"),
"light-road-user": familyRecall("light-road-user"),
vehicle: familyRecall("vehicle"),
},
},
counts: {
truth: integer(qualitySource.truth_instances, "M4.8T truth count"),
predictions: integer(counts.predictions, "M4.8T predictions"),
truePositive: integer(counts.true_positive, "M4.8T TP"),
falsePositive: integer(counts.false_positive, "M4.8T FP"),
falseNegative: integer(counts.false_negative, "M4.8T FN"),
},
failures: parsedFailures,
temporal: {
rawClassSwitches: integer(temporal.raw_class_switches, "M4.8T raw class switches"),
stableClassSwitches: integer(temporal.stable_class_switches, "M4.8T stable class switches"),
rawFamilySwitches: integer(temporal.raw_family_switches, "M4.8T raw family switches"),
stableFamilySwitches: integer(temporal.stable_family_switches, "M4.8T stable family switches"),
suppressedClassSwitches: integer(temporal.suppressed_or_deferred_class_switches, "M4.8T suppressed class switches"),
suppressedFamilySwitches: integer(temporal.suppressed_or_deferred_family_switches, "M4.8T suppressed family switches"),
selectedPublications: integer(temporal.selected_publications, "M4.8T selected publications"),
semanticObservations: integer(temporal.semantic_current_observations, "M4.8T semantic observations"),
peakActiveComponents: integer(snapshot.peak_active_components, "M4.8T active peak"),
maximumActiveComponents: integer(temporalConfiguration.maximum_active_components, "M4.8T active bound"),
},
acceptance: {
qualityPassed: false,
failedQualityGates: ["minimum-micro-precision", "minimum-family-recall:vehicle"],
temporalInvariantPassed: true,
},
review: { truthColor: "green", predictionColor: "yellow", cases },
method: method(raw.method),
limitations: array(raw.limitations, "M4.8T limitations").map((item) => text(item, "M4.8T limitation")),
};
}
export async function fetchM48TRiskQualityResult(
resultId: string,
{
fetcher = fetch,
signal,
}: { fetcher?: LaboratoryFetch; signal?: AbortSignal } = {},
): Promise<M48TRiskQualityResult> {
if (!/^m48t-risk-quality-temporal-lab-[a-f0-9]{64}$/.test(resultId)) {
throw new M48TContractError("M4.8T result identity недопустима.");
}
const response = await fetcher(`/api/v1/laboratory/m48t/risk-quality/results/${resultId}`, {
method: "GET",
headers: { Accept: "application/json" },
signal,
});
if (!response.ok) throw new M48TContractError(`M4.8T недоступен: HTTP ${response.status}.`);
return parseResult(await response.json(), resultId);
}
@@ -45,6 +45,7 @@ import { M47ReferenceGraphResultView } from "./M47ReferenceGraphResult";
import { M48ObjectCentricQualityResultView } from "./M48ObjectCentricQualityResult";
import { M48SmallStaticPassageRegressionResultView } from "./M48SmallStaticPassageRegressionResult";
import { M48SFixedClassDetectorResultView } from "./M48SFixedClassDetectorResult";
import { M48TRiskQualityResultView } from "./M48TRiskQualityResult";
export { isAdvancedLaboratoryWorkId };
export type { AdvancedLaboratoryWorkId };
@@ -96,6 +97,9 @@ export function AdvancedLaboratoryResult({
if (workId === "m48s-fixed-class-detector" && results.m48s) {
return <M48SFixedClassDetectorResultView rigLabel={rigLabel} result={results.m48s} />;
}
if (workId === "m48t-risk-quality-temporal" && results.m48t) {
return <M48TRiskQualityResultView rigLabel={rigLabel} result={results.m48t} />;
}
if (workId === "m47-reference-graph-shadow" && results.m47Graph) {
return <M47ReferenceGraphResultView rigLabel={rigLabel} result={results.m47Graph} />;
}
@@ -0,0 +1,79 @@
import {
LaboratoryEvidence,
LaboratoryResultSummary,
LaboratorySummary,
LaboratoryWorkTemplate,
} from "../../components/laboratory/LaboratoryPresentation";
import type { M48TRiskQualityResult } from "../../core/laboratory/m48tRiskQuality";
import { M48TRiskQualityVisual } from "./M48TRiskQualityVisual";
function percent(value: number, digits = 1): string {
return `${(value * 100).toLocaleString("ru-RU", { maximumFractionDigits: digits })}%`;
}
function decimal(value: number, digits = 1): string {
return value.toLocaleString("ru-RU", { maximumFractionDigits: digits });
}
export function M48TRiskQualityResultView({
rigLabel,
result,
}: {
rigLabel: string;
result: M48TRiskQualityResult;
}) {
return (
<LaboratoryWorkTemplate
summary={(
<LaboratorySummary
title="M4.8T · semantic quality + temporal identity"
description="RF-DETR-L проверен на полном независимом COCO val2017 по заранее записанным risk-family гейтам. Отдельно тот же advisory-класс стабилизирован на geometry-owned ID полного RAVNOVES00 replay; семантика не участвует в association или occupancy."
status="Quality gate failed · temporal invariant passed"
statusTone="warning"
facts={[
{ label: "Quality truth", value: `COCO val2017 · ${result.source.quality.truthInstances.toLocaleString("ru-RU")} risk instances` },
{ label: "Candidate", value: `${result.candidate.modelId} · threshold ${decimal(result.candidate.minimumScore, 2)}` },
{ label: "Temporal source", value: `${rigLabel} · ${result.source.temporal.frames.toLocaleString("ru-RU")} world states` },
{ label: "Authority", value: "SHADOW ONLY · geometry-owned occupancy · production NO" },
]}
brief={{
question: "Достаточно ли качественна текущая risk-семантика RF-DETR и можно ли убрать кадровое мерцание класса без влияния на геометрию?",
approach: `Все ${result.source.quality.riskImages.toLocaleString("ru-RU")} COCO-изображений с risk-классами оценены при неизменном score ${decimal(result.candidate.minimumScore, 2)}. Затем bounded history 5 / confirm 2 / switch 3 применена только к advisory-классу уже существующих component ID.`,
principalResult: `Recall прошёл: ${percent(result.quality.microRecall)} overall и ${percent(result.quality.mediumLargeRecall)} medium+large. Precision ${percent(result.quality.microPrecision)} и vehicle recall ${percent(result.quality.families.vehicle)} не прошли гейты. Temporal shadow сократил class switches ${result.temporal.rawClassSwitches}${result.temporal.stableClassSwitches} и family switches ${result.temporal.rawFamilySwitches}${result.temporal.stableFamilySwitches}.`,
limitation: "COCO не является truth городского маршрута RAVNOVES00, а temporal replay не имеет независимой track/class truth. Batch timing не принимается как realtime-гейт.",
}}
method={result.method}
/>
)}
evidence={(
<LaboratoryEvidence
eyebrow="M4.8T VISUAL EVIDENCE · INDEPENDENT COCO TRUTH"
title="16 hash-bound review cases: human truth против RF-DETR"
kind="diagnostic-model"
resizable
>
<M48TRiskQualityVisual result={result} />
</LaboratoryEvidence>
)}
result={(
<LaboratoryResultSummary
title="Temporal anti-flicker принят как инвариант; detector quality не принят"
status="2/10 predeclared quality checks failed"
statusTone="warning"
metrics={[
{ label: "Precision / recall", value: `${percent(result.quality.microPrecision)} / ${percent(result.quality.microRecall)}`, hint: "gates ≥ 80% / ≥ 75%" },
{ label: "Vehicle family", value: percent(result.quality.families.vehicle), hint: "gate ≥ 85% · failed" },
{ label: "TP / FP / FN", value: `${result.counts.truePositive.toLocaleString("ru-RU")} / ${result.counts.falsePositive.toLocaleString("ru-RU")} / ${result.counts.falseNegative.toLocaleString("ru-RU")}`, hint: `${result.counts.predictions.toLocaleString("ru-RU")} predictions` },
{ label: "Class / family switches", value: `${result.temporal.rawClassSwitches}${result.temporal.stableClassSwitches} / ${result.temporal.rawFamilySwitches}${result.temporal.stableFamilySwitches}`, hint: `peak active ${result.temporal.peakActiveComponents}/${result.temporal.maximumActiveComponents}` },
{ label: "Batch throughput", value: `${decimal(result.execution.effectiveImagesPerSecond, 2)} image/s`, hint: `image p95 ${decimal(result.execution.imageP95Ms, 2)} ms · non-admission` },
]}
conclusion={{
proved: `На независимой COCO truth текущий RF-DETR сохраняет высокий recall: ${percent(result.quality.microRecall)} overall и ${percent(result.quality.mediumLargeRecall)} medium+large. Bounded temporal state подавил или отложил ${result.temporal.suppressedClassSwitches} class-switch и ${result.temporal.suppressedFamilySwitches} family-switch, сохранив association и occupancy class-independent; active state ${result.temporal.peakActiveComponents}/${result.temporal.maximumActiveComponents}, eviction 0.`,
notProved: `Semantic candidate не принят: precision ${percent(result.quality.microPrecision)} при gate 80%, vehicle recall ${percent(result.quality.families.vehicle)} при gate 85%. Не доказаны RAVNOVES class truth, physical track identity, child/adult, unknown moving hazards, risk policy и realtime admission этой batch-командой.`,
decision: "Не менять зафиксированные гейты и не добавлять второй detector. RF-DETR остаётся shadow-кандидатом; temporal stabilizer допустим только как advisory-слой поверх geometry-owned ID.",
}}
/>
)}
/>
);
}
@@ -0,0 +1,63 @@
import { useState } from "react";
import { Icon, IconButton } from "@nodedc/ui-react";
import { LaboratoryEvidenceViewer } from "../../components/laboratory/LaboratoryEvidenceViewer";
import type { M48TRiskQualityResult } from "../../core/laboratory/m48tRiskQuality";
export function M48TRiskQualityVisual({ result }: { result: M48TRiskQualityResult }) {
const [index, setIndex] = useState(0);
const [expanded, setExpanded] = useState(false);
const item = result.review.cases[index] ?? null;
const navigate = (offset: -1 | 1) => {
setIndex((current) => (
current + offset + result.review.cases.length
) % result.review.cases.length);
};
return (
<div className="l3-visual-audit">
<LaboratoryEvidenceViewer
label="M4.8T independent COCO quality review"
mode="quality"
modes={[{ value: "quality", label: "COCO TRUTH" }]}
expanded={expanded}
onModeChange={() => undefined}
onExpandedChange={setExpanded}
actions={(
<div className="l3-visual-audit__pagination">
<IconButton label="Предыдущий M4.8T review case" onClick={() => navigate(-1)}>
<Icon name="chevron-left" size={16} />
</IconButton>
<IconButton label="Следующий M4.8T review case" onClick={() => navigate(1)}>
<Icon name="chevron-right" size={16} />
</IconButton>
</div>
)}
overlay={item ? (
<div className="l3-visual-audit__overlay">
<div>
<span>COCO val2017 · independent human truth</span>
<strong>case {index + 1}/{result.review.cases.length} · image {item.imageId}</strong>
<small>Зелёный truth · жёлтый RF-DETR prediction · score 0,25</small>
</div>
</div>
) : null}
>
{item ? (
<div className="l32-camera-scene">
<img
src={item.imageUrl}
alt={`M4.8T COCO review image ${item.imageId}: truth and RF-DETR boxes`}
draggable={false}
/>
</div>
) : (
<div className="l3-visual-audit__state" role="alert">
<Icon name="alert" size={18} />
M4.8T visual evidence недоступно.
</div>
)}
</LaboratoryEvidenceViewer>
</div>
);
}
@@ -84,6 +84,13 @@ const KNOWN_WORKS: Readonly<Record<Exclude<LaboratoryWorkId, `session:${string}`
experimentName: "RAVNOVES00 fixed-class risk detector",
variantName: "M4.8S · RF-DETR-L TensorRT/Triton shadow",
},
"m48t-risk-quality-temporal": {
profileId: "rig-ravnoves-perception-gate-v1",
profileName: (rigLabel) => `${rig(rigLabel)} RIGHT · COCO quality + bounded temporal identity`,
experimentId: "m48t-risk-quality-temporal",
experimentName: "RF-DETR independent semantic quality and temporal identity",
variantName: "M4.8T · COCO val2017 truth + RAVNOVES00 temporal shadow",
},
"m47-reference-graph-shadow": {
profileId: "rig-dual-evidence-virtual-corridor-v1",
profileName: (rigLabel) => `${rig(rigLabel)} RIGHT · Camera + LiDAR dual evidence`,
@@ -22,6 +22,7 @@ function mergeResults(
m48: next.m48 ?? current.m48,
m48SmallStatic: next.m48SmallStatic ?? current.m48SmallStatic,
m48s: next.m48s ?? current.m48s,
m48t: next.m48t ?? current.m48t,
m4Threat: next.m4Threat ?? current.m4Threat,
l3: next.l3 ?? current.l3,
l31: next.l31 ?? current.l31,
@@ -0,0 +1,10 @@
{
"schema_version": "missioncore.laboratory-evidence-definition/v1",
"work_id": "m48t-risk-quality-temporal",
"evidence": {
"runtime_relative_root": "m48t-risk-quality/lab-results",
"result_id_prefix": "m48t-risk-quality-temporal-lab",
"document_name": "manifest.json",
"schema_version": "missioncore.m48t-risk-quality-temporal-lab/v1"
}
}
+14
View File
@@ -128,6 +128,20 @@
"run": "missioncore.laboratory-run/v1",
"evidence": "missioncore.m48s-fixed-class-detector-lab/v1"
}
},
{
"work_id": "m48t-risk-quality-temporal",
"lifecycle": "experimental",
"isolation": "bounded-adapter",
"adapter_id": "experimental.m48t-risk-quality-temporal/v1",
"input_roles": ["repository_root"],
"contracts": {
"source": "missioncore.m48t-sealed-quality-temporal-source-set/v1",
"provider": "missioncore.rf-detr-risk-quality-provider/v1",
"graph": "missioncore.m48t-risk-quality-temporal-lab-graph/v1",
"run": "missioncore.laboratory-run/v1",
"evidence": "missioncore.m48t-risk-quality-temporal-lab/v1"
}
}
],
"legacy_work_ids": [
+7
View File
@@ -253,6 +253,13 @@
"signal": "progress",
"lifecycle": "current",
"visual_evidence": "available"
},
{
"catalog_id": "m48t-risk-quality-temporal",
"evidence_id": "m48t-risk-quality-temporal-lab-ed5355fe0adb9b18942d75aff2362d79190b9e864f070ebe7a1c15fccbc0fbfb",
"signal": "progress",
"lifecycle": "current",
"visual_evidence": "available"
}
]
}
+16
View File
@@ -322,6 +322,7 @@ def canonical_laboratory_adapters() -> dict[str, LaboratoryAdapter]:
"canonical.e46j-raw-fisheye-realtime/v1": _run_e46j,
"experimental.e47-semantic-slam-shadow/v1": _run_e47,
"experimental.m48s-fixed-class-detector/v1": _run_m48s_fixed_class_detector,
"experimental.m48t-risk-quality-temporal/v1": _run_m48t_risk_quality_temporal,
}
@@ -342,6 +343,21 @@ def _run_m48s_fixed_class_detector(
)
def _run_m48t_risk_quality_temporal(
request: LaboratoryRunRequest,
) -> LaboratoryAdapterResult:
from k1link.laboratory.m48t_risk_quality_lab import build_m48t_risk_quality_lab
result = build_m48t_risk_quality_lab(
repository_root=request.inputs["repository_root"],
output_root=request.output_root,
)
return LaboratoryAdapterResult(
result_root=result.result_root,
result_id=result.result_id,
)
def _run_m48_small_static_passage_regression(
request: LaboratoryRunRequest,
) -> LaboratoryAdapterResult:
@@ -0,0 +1,400 @@
"""Seal the M4.8T semantic-quality and temporal-identity evidence as a LAB result."""
from __future__ import annotations
import hashlib
import json
import shutil
import tempfile
from dataclasses import dataclass
from datetime import UTC, datetime
from pathlib import Path
from typing import Any, Final
from k1link.perception.fixed_class_detector_tournament import (
canonical_json,
false_authority,
sha256_path,
)
LAB_SCHEMA: Final = "missioncore.m48t-risk-quality-temporal-lab/v1"
REPORT_SCHEMA: Final = "missioncore.m48t-risk-quality-temporal-report/v1"
CATALOG_SCHEMA: Final = "missioncore.m48t-risk-quality-review-catalog/v1"
RESULT_PREFIX: Final = "m48t-risk-quality-temporal-lab-"
PROFILE_RELATIVE_PATH: Final = Path(
"config/perception/m48t-risk-quality-temporal-v1.json"
)
WORKER_RELATIVE_ROOT: Final = Path(
".runtime/worker-results/m48t-risk-quality-coco2017-val-full-v1"
)
TEMPORAL_LEDGER_RELATIVE_PATH: Final = Path(
".runtime/m48s-reference-graph-shadow/full-replay-d85983f1/frames.jsonl"
)
class M48TRiskQualityLabError(RuntimeError):
"""Raised when the M4.8T evidence cannot be sealed honestly."""
@dataclass(frozen=True, slots=True)
class M48TRiskQualityLabResult:
result_root: Path
result_id: str
manifest: dict[str, Any]
def build_m48t_risk_quality_lab(
*,
repository_root: Path,
output_root: Path,
) -> M48TRiskQualityLabResult:
repository = repository_root.expanduser().resolve(strict=True)
profile_path = repository / PROFILE_RELATIVE_PATH
worker_root = repository / WORKER_RELATIVE_ROOT
temporal_ledger_path = repository / TEMPORAL_LEDGER_RELATIVE_PATH
quality_path = worker_root / "result.json"
temporal_path = worker_root / "temporal-semantic-shadow.json"
predictions_path = worker_root / "predictions.jsonl"
failures_path = worker_root / "failures.jsonl"
review_root = worker_root / "review"
for path in (
profile_path,
quality_path,
temporal_path,
predictions_path,
failures_path,
temporal_ledger_path,
):
if path.is_symlink() or not path.is_file():
raise M48TRiskQualityLabError(f"required M4.8T evidence is missing: {path.name}")
if review_root.is_symlink() or not review_root.is_dir():
raise M48TRiskQualityLabError("M4.8T review evidence is missing")
profile = _read_object(profile_path)
quality = _read_object(quality_path)
temporal = _read_object(temporal_path)
review_paths = sorted(review_root.glob("review-*.jpg"))
_validate_inputs(
profile=profile,
profile_path=profile_path,
quality=quality,
predictions_path=predictions_path,
failures_path=failures_path,
temporal=temporal,
temporal_ledger_path=temporal_ledger_path,
review_paths=review_paths,
)
method = _method(profile, quality)
identity = {
"schema_version": LAB_SCHEMA,
"profile": {
"profile_id": profile["profile_id"],
"sha256": sha256_path(profile_path),
},
"source": {
"quality_dataset_id": quality["dataset"]["dataset_id"],
"quality_report_identity_sha256": quality["report_identity_sha256"],
"quality_report_sha256": sha256_path(quality_path),
"predictions_sha256": sha256_path(predictions_path),
"failures_sha256": sha256_path(failures_path),
"temporal_source_id": temporal["source"]["source_id"],
"temporal_ledger_sha256": sha256_path(temporal_ledger_path),
"temporal_shadow_sha256": sha256_path(temporal_path),
},
"method": method,
"authority": false_authority(),
}
identity_sha256 = hashlib.sha256(canonical_json(identity)).hexdigest()
result_id = RESULT_PREFIX + identity_sha256
completed_ns = quality["execution"]["completed_at_unix_ns"]
created_at_utc = (
datetime.fromtimestamp(completed_ns / 1_000_000_000, UTC)
.isoformat(timespec="microseconds")
.replace("+00:00", "Z")
)
root = output_root.expanduser().absolute()
root.mkdir(mode=0o700, parents=True, exist_ok=True)
destination = root / result_id
if destination.exists():
manifest = _read_object(destination / "manifest.json")
if (
manifest.get("schema_version") != LAB_SCHEMA
or manifest.get("identity_sha256") != identity_sha256
or manifest.get("result_id") != result_id
):
raise M48TRiskQualityLabError("existing M4.8T LAB identity conflicts")
return M48TRiskQualityLabResult(destination, result_id, manifest)
temporary = Path(tempfile.mkdtemp(prefix=".m48t-risk-quality-lab-", dir=root))
try:
(temporary / "review").mkdir(mode=0o700)
shutil.copyfile(profile_path, temporary / "profile.json")
shutil.copyfile(quality_path, temporary / "worker-quality-result.json")
shutil.copyfile(predictions_path, temporary / "predictions.jsonl")
shutil.copyfile(failures_path, temporary / "failures.jsonl")
shutil.copyfile(temporal_path, temporary / "temporal-semantic-shadow.json")
review_items: list[dict[str, object]] = []
for source in review_paths:
destination_image = temporary / "review" / source.name
shutil.copyfile(source, destination_image)
case_id = source.stem.removeprefix("review-")
review_items.append(
{
"case_id": case_id,
"image_id": int(case_id),
"path": f"review/{source.name}",
"media_type": "image/jpeg",
"byte_length": destination_image.stat().st_size,
"sha256": sha256_path(destination_image),
}
)
catalog = {
"schema_version": CATALOG_SCHEMA,
"result_id": result_id,
"case_count": len(review_items),
"legend": {
"ground_truth": "green",
"rf_detr_prediction": "yellow",
},
"cases": review_items,
}
catalog_path = temporary / "catalog.json"
catalog_path.write_bytes(canonical_json(catalog) + b"\n")
decision = {
"quality_evaluated": True,
"quality_accepted": False,
"failed_quality_gates": quality["quality_gates"]["failed"],
"temporal_invariant_evaluated": True,
"temporal_invariant_passed": True,
"candidate_accepted": False,
"production_accepted": False,
}
limitations = [
"COCO val2017 is independent class truth, but it is not RAVNOVES00 domain truth.",
"The temporal replay has no independent semantic or physical track-identity truth.",
(
"Worker timing is throughput evidence for this batch run, not a realtime "
"admission gate."
),
(
"Child/adult distinction, unknown moving objects and behavior risk policy "
"were not evaluated."
),
"Semantic labels do not own geometry association, occupancy, navigation or actuation.",
]
report = {
"schema_version": REPORT_SCHEMA,
"result_id": result_id,
"source": {
"quality": quality["dataset"],
"temporal": temporal["source"],
},
"configuration": {
"candidate": quality["candidate"],
"matching": profile["matching"],
"quality_gates": profile["quality_gates"],
"temporal": profile["temporal"],
},
"method": method,
"execution": quality["execution"],
"metrics": {
"quality": quality["metrics"],
"counts": quality["counts"],
"failure_buckets": quality["failure_buckets"],
"detector_rejections": quality["detector_rejections"],
"temporal": temporal["metrics"],
},
"acceptance": {
"quality": quality["quality_gates"],
"temporal_invariant": temporal["temporal_invariant_gate"],
},
"decision": decision,
"limitations": limitations,
"authority": false_authority(),
"visual_evidence": {
"kind": "independent-coco-human-truth-review",
"case_count": len(review_items),
"catalog_schema_version": CATALOG_SCHEMA,
"ground_truth_for_ravnoves00": False,
},
}
report_path = temporary / "report.json"
report_path.write_bytes(canonical_json(report) + b"\n")
artifacts = _artifact_manifest(temporary)
manifest = {
"schema_version": LAB_SCHEMA,
"result_id": result_id,
"identity_sha256": identity_sha256,
"identity": identity,
"created_at_utc": created_at_utc,
"status": "complete-quality-gate-failed-temporal-invariant-passed",
"completed": True,
"bounded_question_accepted": False,
"ground_truth": False,
"catalog": {
"path": "catalog.json",
"sha256": sha256_path(catalog_path),
"byte_length": catalog_path.stat().st_size,
},
"method": method,
"metrics": report["metrics"],
"decision": decision,
"limitations": limitations,
"authority": false_authority(),
"artifacts": artifacts,
}
(temporary / "manifest.json").write_bytes(canonical_json(manifest) + b"\n")
temporary.replace(destination)
except BaseException:
shutil.rmtree(temporary, ignore_errors=True)
raise
return M48TRiskQualityLabResult(destination, result_id, manifest)
def _validate_inputs(
*,
profile: dict[str, Any],
profile_path: Path,
quality: dict[str, Any],
predictions_path: Path,
failures_path: Path,
temporal: dict[str, Any],
temporal_ledger_path: Path,
review_paths: list[Path],
) -> None:
expected_failed = ["minimum-micro-precision", "minimum-family-recall:vehicle"]
if (
profile.get("schema_version")
!= "missioncore.m48t-risk-quality-temporal-profile/v1"
or quality.get("schema_version") != "missioncore.m48t-risk-quality-report/v1"
or temporal.get("schema_version")
!= "missioncore.m48t-temporal-semantic-shadow/v1"
or quality.get("profile_sha256") != sha256_path(profile_path)
or temporal.get("profile_sha256") != sha256_path(profile_path)
or quality.get("profile_id") != profile.get("profile_id")
or temporal.get("profile_id") != profile.get("profile_id")
or quality.get("provenance", {}).get("predictions_sha256")
!= sha256_path(predictions_path)
or quality.get("provenance", {}).get("failures_sha256")
!= sha256_path(failures_path)
or temporal.get("source", {}).get("frame_ledger_sha256")
!= sha256_path(temporal_ledger_path)
or quality.get("dataset", {}).get("truth_instances") != 16060
or quality.get("quality_gates", {}).get("passed") is not False
or quality.get("quality_gates", {}).get("failed") != expected_failed
or temporal.get("temporal_invariant_gate", {}).get("passed") is not True
or temporal.get("temporal_invariant_gate", {}).get("semantic_quality_accepted")
is not False
or quality.get("authority", {}).get("candidate_accepted") is not False
or len(review_paths) != 16
or any(path.is_symlink() or not path.is_file() for path in review_paths)
):
raise M48TRiskQualityLabError("M4.8T evidence contract changed")
def _method(profile: dict[str, Any], quality: dict[str, Any]) -> dict[str, object]:
provenance = quality["provenance"]
return {
"schema_version": "missioncore.laboratory-method/v1",
"completeness": "complete",
"execution_class": "hybrid",
"pipeline_id": "m48t-coco-quality-plus-bounded-temporal-identity/v1",
"components": [
{
"kind": "source",
"name": "COCO 2017 validation",
"version": "val2017 independent human annotations",
"role": "semantic quality truth",
"identity_sha256": provenance["annotations_document_sha256"],
},
{
"kind": "model",
"name": "RF-DETR-L COCO",
"version": quality["candidate"]["model_id"],
"role": "fixed-class risk detector under evaluation",
"identity_sha256": provenance["runtime_artifact_sha256"],
},
{
"kind": "algorithm",
"name": "COCO risk-family scorer",
"version": profile["matching"]["method"],
"role": "predeclared exact-class and family quality gates",
"identity_sha256": provenance["runner_sha256"],
},
{
"kind": "source",
"name": "RAVNOVES00 reference-graph replay",
"version": "4481-frame immutable publication ledger",
"role": "temporal semantic anti-flicker shadow",
"identity_sha256": (
"badfa2f5f4f33fea7d5ad0e14fe5bbe38e1d490fc661d637789c354b8576d533"
),
},
{
"kind": "algorithm",
"name": "bounded temporal semantic identity",
"version": "history-5-confirm-2-switch-3/v1",
"role": "stabilize advisory class on geometry-owned component IDs",
"identity_sha256": hashlib.sha256(
canonical_json(profile["temporal"])
).hexdigest(),
},
],
}
def _artifact_manifest(root: Path) -> list[dict[str, object]]:
artifacts: list[dict[str, object]] = []
for path in sorted(item for item in root.rglob("*") if item.is_file()):
relative = path.relative_to(root).as_posix()
if relative == "manifest.json":
continue
media_type = "application/json"
schema_version: str | None = None
role = "supporting-evidence"
if relative == "report.json":
role = "laboratory-report"
schema_version = REPORT_SCHEMA
elif relative == "catalog.json":
role = "visual-evidence-catalog"
schema_version = CATALOG_SCHEMA
elif relative == "profile.json":
role = "predeclared-quality-temporal-profile"
schema_version = "missioncore.m48t-risk-quality-temporal-profile/v1"
elif relative == "worker-quality-result.json":
role = "upstream-worker-quality-evidence"
schema_version = "missioncore.m48t-risk-quality-report/v1"
elif relative == "temporal-semantic-shadow.json":
role = "upstream-temporal-shadow-evidence"
schema_version = "missioncore.m48t-temporal-semantic-shadow/v1"
elif relative.endswith(".jsonl"):
media_type = "application/x-ndjson"
role = "upstream-quality-ledger"
elif relative.endswith(".jpg"):
media_type = "image/jpeg"
role = "visual-evidence-independent-truth-review"
artifacts.append(
{
"role": role,
"path": relative,
"byte_length": path.stat().st_size,
"sha256": sha256_path(path),
"media_type": media_type,
"schema_version": schema_version,
}
)
return artifacts
def _read_object(path: Path) -> dict[str, Any]:
try:
value = json.loads(path.read_text("utf-8"))
except (OSError, json.JSONDecodeError) as exc:
raise M48TRiskQualityLabError(f"invalid JSON evidence: {path.name}") from exc
if not isinstance(value, dict):
raise M48TRiskQualityLabError(f"JSON evidence must be an object: {path.name}")
return value
+14
View File
@@ -128,6 +128,7 @@ from k1link.web.m48_object_quality_api import build_m48_object_quality_router
from k1link.web.m48s_fixed_class_detector_lab_api import (
build_m48s_fixed_class_detector_lab_router,
)
from k1link.web.m48t_risk_quality_lab_api import build_m48t_risk_quality_lab_router
from k1link.web.map_api import (
MapGatewayConfiguration,
MapGatewayProxy,
@@ -151,6 +152,7 @@ from k1link.web.runtime_readiness import (
build_runtime_readiness,
)
from k1link.web.session_api import build_session_router
from k1link.web.simulation_world_provider_api import build_simulation_world_provider_router
from k1link.web.system_telemetry_api import build_system_telemetry_router
from k1link.web.viewer_diagnostics_api import build_viewer_diagnostics_router
@@ -964,6 +966,17 @@ app.include_router(
),
)
)
app.include_router(
build_m48t_risk_quality_lab_router(
root_provider=lambda: (
REPOSITORY_ROOT
/ ".runtime"
/ "compute-experiments"
/ "m48t-risk-quality"
/ "lab-results"
),
)
)
app.include_router(
build_e47_semantic_slam_router(
root_provider=lambda: (
@@ -1296,6 +1309,7 @@ app.include_router(
root_provider=lambda: REPOSITORY_ROOT / ".runtime" / "system",
)
)
app.include_router(build_simulation_world_provider_router())
frontend_dist = REPOSITORY_ROOT / "apps" / "control-station" / "dist"
app.include_router(
build_viewer_diagnostics_router(
+301
View File
@@ -0,0 +1,301 @@
"""Read-only API for the sealed M4.8T quality and temporal LAB."""
from __future__ import annotations
import copy
import hashlib
import json
import re
from collections.abc import Callable
from pathlib import Path, PurePosixPath
from typing import Any, Final
from fastapi import APIRouter, HTTPException, Query
from fastapi.responses import FileResponse
from k1link.laboratory.evidence_registry import LaboratoryEvidenceDefinition
from k1link.laboratory.evidence_report import (
LaboratoryEvidenceReportError,
verify_laboratory_evidence_result,
)
from k1link.laboratory.m48t_risk_quality_lab import (
CATALOG_SCHEMA,
LAB_SCHEMA,
REPORT_SCHEMA,
RESULT_PREFIX,
)
from k1link.perception.fixed_class_detector_tournament import false_authority
RootProvider = Callable[[], Path | None]
RESULT_ID: Final = re.compile(rf"^{re.escape(RESULT_PREFIX)}[a-f0-9]{{64}}$")
CASE_ID: Final = re.compile(r"^[0-9]{12}$")
VIEW_SCHEMA: Final = "missioncore.m48t-risk-quality-temporal-view/v1"
CATALOG_VIEW_SCHEMA: Final = "missioncore.m48t-risk-quality-temporal-catalog/v1"
_DEFINITION: Final = LaboratoryEvidenceDefinition(
work_id="m48t-risk-quality-temporal",
runtime_relative_root=PurePosixPath("m48t-risk-quality/lab-results"),
result_id_prefix="m48t-risk-quality-temporal-lab",
document_name="manifest.json",
result_schema_version=LAB_SCHEMA,
)
def build_m48t_risk_quality_lab_router(
*, root_provider: RootProvider = lambda: None
) -> APIRouter:
router = APIRouter(
prefix="/api/v1/laboratory/m48t/risk-quality",
tags=["laboratory"],
)
@router.get("/results")
def list_results(limit: int = Query(default=1, ge=1, le=10)) -> dict[str, object]:
root = _configured_root(root_provider)
if root is None:
return _empty_catalog(False)
candidates = _candidates(root)
items: list[dict[str, object]] = []
invalid_total = 0
for candidate in candidates:
try:
items.append(_project_result(candidate))
except RuntimeError:
invalid_total += 1
items.sort(
key=lambda item: (str(item["created_at_utc"]), str(item["result_id"])),
reverse=True,
)
return {
"schema_version": CATALOG_VIEW_SCHEMA,
"configured": True,
"items": items[:limit],
"candidate_total": len(candidates),
"invalid_total": invalid_total,
"access": "read-only",
}
@router.get("/results/{result_id}")
def get_result(result_id: str) -> dict[str, object]:
try:
return _project_result(_resolve_candidate(root_provider, result_id))
except RuntimeError:
raise HTTPException(status_code=404, detail="M4.8T result not found") from None
@router.get("/results/{result_id}/review/{case_id}.jpg")
def get_review_image(result_id: str, case_id: str) -> FileResponse:
if CASE_ID.fullmatch(case_id) is None:
raise HTTPException(status_code=404, detail="M4.8T review case not found")
try:
loaded = _load_result(_resolve_candidate(root_provider, result_id))
except RuntimeError:
raise HTTPException(status_code=404, detail="M4.8T result not found") from None
descriptor = next(
(
item
for item in loaded["catalog"]["cases"]
if isinstance(item, dict) and item.get("case_id") == case_id
),
None,
)
if not isinstance(descriptor, dict):
raise HTTPException(status_code=404, detail="M4.8T review case not found")
candidate = loaded["root"]
path = (candidate / str(descriptor["path"])).resolve()
if (
not path.is_relative_to(candidate)
or path.is_symlink()
or not path.is_file()
or descriptor.get("byte_length") != path.stat().st_size
or descriptor.get("sha256") != _sha256(path)
):
raise HTTPException(status_code=404, detail="M4.8T review case not found")
return FileResponse(
path,
media_type="image/jpeg",
headers={
"Cache-Control": "public, max-age=31536000, immutable",
"ETag": f'"{descriptor["sha256"]}"',
"X-Content-Type-Options": "nosniff",
},
)
return router
def _project_result(candidate: Path) -> dict[str, object]:
loaded = _load_result(candidate)
manifest = loaded["manifest"]
report = loaded["report"]
catalog = loaded["catalog"]
result_id = str(manifest["result_id"])
review_cases = [
{
"case_id": item["case_id"],
"image_id": item["image_id"],
"image_url": (
f"/api/v1/laboratory/m48t/risk-quality/results/{result_id}"
f"/review/{item['case_id']}.jpg"
),
"media_type": item["media_type"],
"byte_length": item["byte_length"],
"sha256": item["sha256"],
}
for item in catalog["cases"]
]
return {
"schema_version": VIEW_SCHEMA,
"result_id": result_id,
"created_at_utc": manifest["created_at_utc"],
"status": manifest["status"],
"source": copy.deepcopy(report["source"]),
"configuration": copy.deepcopy(report["configuration"]),
"method": copy.deepcopy(report["method"]),
"execution": copy.deepcopy(report["execution"]),
"metrics": copy.deepcopy(report["metrics"]),
"acceptance": copy.deepcopy(report["acceptance"]),
"decision": copy.deepcopy(report["decision"]),
"limitations": copy.deepcopy(report["limitations"]),
"review": {
"legend": copy.deepcopy(catalog["legend"]),
"cases": review_cases,
},
"ground_truth": False,
"authority": copy.deepcopy(report["authority"]),
"access": "read-only",
}
def _load_result(candidate: Path) -> dict[str, Any]:
if (
not candidate.is_dir()
or candidate.is_symlink()
or RESULT_ID.fullmatch(candidate.name) is None
):
raise RuntimeError("M4.8T result candidate is invalid")
try:
verify_laboratory_evidence_result(_DEFINITION, candidate)
manifest = _read_object(candidate / "manifest.json")
report = _read_object(candidate / "report.json")
catalog = _read_object(candidate / "catalog.json")
except (LaboratoryEvidenceReportError, OSError, ValueError) as exc:
raise RuntimeError("M4.8T result integrity failed") from exc
identity = manifest.get("identity")
if (
manifest.get("schema_version") != LAB_SCHEMA
or manifest.get("result_id") != candidate.name
or manifest.get("status")
!= "complete-quality-gate-failed-temporal-invariant-passed"
or manifest.get("completed") is not True
or manifest.get("bounded_question_accepted") is not False
or manifest.get("ground_truth") is not False
or not isinstance(manifest.get("created_at_utc"), str)
or not isinstance(identity, dict)
or manifest.get("identity_sha256") != _canonical_sha256(identity)
or not candidate.name.endswith(str(manifest.get("identity_sha256")))
or identity.get("authority") != false_authority()
or manifest.get("authority") != false_authority()
or report.get("schema_version") != REPORT_SCHEMA
or report.get("result_id") != candidate.name
or report.get("authority") != false_authority()
or report.get("decision", {}).get("quality_accepted") is not False
or report.get("decision", {}).get("temporal_invariant_passed") is not True
or catalog.get("schema_version") != CATALOG_SCHEMA
or catalog.get("result_id") != candidate.name
or catalog.get("case_count") != 16
or not isinstance(catalog.get("cases"), list)
or len(catalog["cases"]) != 16
or any(not _valid_case(item) for item in catalog["cases"])
):
raise RuntimeError("M4.8T result contract changed")
return {"root": candidate, "manifest": manifest, "report": report, "catalog": catalog}
def _valid_case(value: object) -> bool:
return (
isinstance(value, dict)
and isinstance(value.get("case_id"), str)
and CASE_ID.fullmatch(value["case_id"]) is not None
and value.get("image_id") == int(value["case_id"])
and value.get("path") == f"review/review-{value['case_id']}.jpg"
and value.get("media_type") == "image/jpeg"
and isinstance(value.get("byte_length"), int)
and value["byte_length"] > 0
and isinstance(value.get("sha256"), str)
and re.fullmatch(r"[a-f0-9]{64}", value["sha256"]) is not None
)
def _resolve_candidate(provider: RootProvider, result_id: str) -> Path:
if RESULT_ID.fullmatch(result_id) is None:
raise HTTPException(status_code=404, detail="M4.8T result not found")
root = _configured_root(provider)
if root is None:
raise HTTPException(status_code=404, detail="M4.8T result not found")
candidate = (root / result_id).resolve()
if candidate.parent != root or candidate.is_symlink() or not candidate.is_dir():
raise HTTPException(status_code=404, detail="M4.8T result not found")
return candidate
def _configured_root(provider: RootProvider) -> Path | None:
value = provider()
if value is None:
return None
candidate = value.expanduser().absolute()
if candidate.is_symlink():
return None
try:
root = candidate.resolve(strict=True)
except OSError:
return None
return root if root.is_dir() else None
def _candidates(root: Path) -> list[Path]:
return sorted(
(
item
for item in root.iterdir()
if item.is_dir() and not item.is_symlink() and RESULT_ID.fullmatch(item.name)
),
key=lambda item: item.stat().st_mtime_ns,
reverse=True,
)
def _empty_catalog(configured: bool) -> dict[str, object]:
return {
"schema_version": CATALOG_VIEW_SCHEMA,
"configured": configured,
"items": [],
"candidate_total": 0,
"invalid_total": 0,
"access": "read-only",
}
def _read_object(path: Path) -> dict[str, Any]:
value = json.loads(path.read_text("utf-8"))
if not isinstance(value, dict):
raise ValueError("JSON evidence must be an object")
return value
def _canonical_sha256(value: object) -> str:
return hashlib.sha256(
json.dumps(
value,
ensure_ascii=False,
sort_keys=True,
separators=(",", ":"),
allow_nan=False,
).encode("utf-8")
).hexdigest()
def _sha256(path: Path) -> str:
digest = hashlib.sha256()
with path.open("rb") as stream:
for chunk in iter(lambda: stream.read(1024 * 1024), b""):
digest.update(chunk)
return digest.hexdigest()
+2 -1
View File
@@ -127,7 +127,7 @@ def test_product_registry_declares_every_advanced_evidence_source() -> None:
repository_root / "config" / "laboratories"
)
assert len(registry.definitions) == 37
assert len(registry.definitions) == 38
assert {item.work_id for item in registry.definitions} >= {
"e31-source-binding",
"e46j-raw-fisheye-realtime",
@@ -142,6 +142,7 @@ def test_product_registry_declares_every_advanced_evidence_source() -> None:
"m48-object-centric-quality",
"m48-small-static-passage-regression",
"m48s-fixed-class-detector",
"m48t-risk-quality-temporal",
}
m48 = next(
item for item in registry.definitions
+9 -1
View File
@@ -99,6 +99,7 @@ def test_repository_registry_classifies_every_evidence_definition() -> None:
"e46j-raw-fisheye-realtime",
"e47-semantic-slam-shadow",
"m48s-fixed-class-detector",
"m48t-risk-quality-temporal",
}
by_work_id = {row.work_id: row for row in execution.definitions}
assert by_work_id["m48-small-static-passage-regression"].evidence_contract == (
@@ -111,10 +112,17 @@ def test_repository_registry_classifies_every_evidence_definition() -> None:
assert by_work_id["e47-semantic-slam-shadow"].isolation == "bounded-adapter"
assert by_work_id["m48s-fixed-class-detector"].lifecycle == "experimental"
assert by_work_id["m48s-fixed-class-detector"].isolation == "bounded-adapter"
assert by_work_id["m48t-risk-quality-temporal"].lifecycle == "experimental"
assert by_work_id["m48t-risk-quality-temporal"].isolation == "bounded-adapter"
assert all(
row.lifecycle == "canonical"
for row in execution.definitions
if row.work_id not in {"e47-semantic-slam-shadow", "m48s-fixed-class-detector"}
if row.work_id
not in {
"e47-semantic-slam-shadow",
"m48s-fixed-class-detector",
"m48t-risk-quality-temporal",
}
)
assert len(execution.definitions) + len(execution.legacy_work_ids) == len(
evidence.definitions
@@ -80,7 +80,7 @@ def test_product_value_review_registry_covers_reviewed_laboratory_families() ->
root / "config" / "laboratory-value-review.json"
)
assert len(registry.entries) == 36
assert len(registry.entries) == 37
assert {entry.catalog_id for entry in registry.entries} >= {
"e28-local-surface",
"e46d-temporal-failure-audit",
@@ -89,4 +89,5 @@ def test_product_value_review_registry_covers_reviewed_laboratory_families() ->
"l34f-adjudicated-reference",
"m4-replay-threat",
"m48s-fixed-class-detector",
"m48t-risk-quality-temporal",
}
+152
View File
@@ -0,0 +1,152 @@
from __future__ import annotations
import hashlib
from pathlib import Path
from fastapi import FastAPI
from fastapi.testclient import TestClient
from k1link.laboratory.m48t_risk_quality_lab import (
CATALOG_SCHEMA,
LAB_SCHEMA,
REPORT_SCHEMA,
RESULT_PREFIX,
)
from k1link.perception.fixed_class_detector_tournament import canonical_json, false_authority
from k1link.web.m48t_risk_quality_lab_api import build_m48t_risk_quality_lab_router
def _write_json(path: Path, value: object) -> None:
path.write_bytes(canonical_json(value) + b"\n")
def _descriptor(path: Path, root: Path, role: str, media_type: str) -> dict[str, object]:
return {
"role": role,
"path": path.relative_to(root).as_posix(),
"byte_length": path.stat().st_size,
"sha256": hashlib.sha256(path.read_bytes()).hexdigest(),
"media_type": media_type,
"schema_version": None,
}
def _fixture(tmp_path: Path) -> tuple[TestClient, Path, str]:
root = tmp_path / "results"
root.mkdir()
identity = {"schema_version": LAB_SCHEMA, "authority": false_authority()}
identity_sha256 = hashlib.sha256(canonical_json(identity)).hexdigest()
result_id = RESULT_PREFIX + identity_sha256
result_root = root / result_id
review_root = result_root / "review"
review_root.mkdir(parents=True)
cases = []
image_paths = []
for index in range(16):
case_id = f"{index + 1:012d}"
image_path = review_root / f"review-{case_id}.jpg"
image_path.write_bytes(b"jpeg" + bytes([index]))
image_paths.append(image_path)
cases.append(
{
"case_id": case_id,
"image_id": index + 1,
"path": f"review/{image_path.name}",
"media_type": "image/jpeg",
"byte_length": image_path.stat().st_size,
"sha256": hashlib.sha256(image_path.read_bytes()).hexdigest(),
}
)
catalog = {
"schema_version": CATALOG_SCHEMA,
"result_id": result_id,
"case_count": 16,
"legend": {"ground_truth": "green", "rf_detr_prediction": "yellow"},
"cases": cases,
}
report = {
"schema_version": REPORT_SCHEMA,
"result_id": result_id,
"source": {
"quality": {
"dataset_id": "coco-2017-val",
"risk_images": 3348,
"truth_instances": 16060,
},
"temporal": {"source_id": "RAVNOVES00", "frames": 4481},
},
"configuration": {},
"method": {"schema_version": "missioncore.laboratory-method/v1"},
"execution": {},
"metrics": {},
"acceptance": {},
"decision": {
"quality_accepted": False,
"temporal_invariant_passed": True,
},
"limitations": [],
"authority": false_authority(),
}
catalog_path = result_root / "catalog.json"
report_path = result_root / "report.json"
_write_json(catalog_path, catalog)
_write_json(report_path, report)
artifacts = [
_descriptor(catalog_path, result_root, "visual-evidence-catalog", "application/json"),
_descriptor(report_path, result_root, "laboratory-report", "application/json"),
*[
_descriptor(path, result_root, "visual-evidence-independent-truth-review", "image/jpeg")
for path in image_paths
],
]
manifest = {
"schema_version": LAB_SCHEMA,
"result_id": result_id,
"identity_sha256": identity_sha256,
"identity": identity,
"created_at_utc": "2026-08-25T17:38:01.505739Z",
"status": "complete-quality-gate-failed-temporal-invariant-passed",
"completed": True,
"bounded_question_accepted": False,
"ground_truth": False,
"authority": false_authority(),
"artifacts": artifacts,
}
_write_json(result_root / "manifest.json", manifest)
app = FastAPI()
app.include_router(build_m48t_risk_quality_lab_router(root_provider=lambda: root))
return TestClient(app), result_root, result_id
def test_m48t_lab_api_projects_failed_quality_and_verified_review(tmp_path: Path) -> None:
client, result_root, result_id = _fixture(tmp_path)
catalog = client.get("/api/v1/laboratory/m48t/risk-quality/results")
assert catalog.status_code == 200
assert catalog.json()["items"][0]["result_id"] == result_id
assert catalog.json()["invalid_total"] == 0
result = client.get(f"/api/v1/laboratory/m48t/risk-quality/results/{result_id}")
assert result.status_code == 200
assert result.json()["decision"]["quality_accepted"] is False
assert result.json()["decision"]["temporal_invariant_passed"] is True
assert len(result.json()["review"]["cases"]) == 16
case_id = "000000000001"
image = client.get(
f"/api/v1/laboratory/m48t/risk-quality/results/{result_id}/review/{case_id}.jpg"
)
assert image.status_code == 200
assert image.headers["content-type"] == "image/jpeg"
assert image.content == (result_root / f"review/review-{case_id}.jpg").read_bytes()
def test_m48t_lab_api_fails_closed_after_visual_tamper(tmp_path: Path) -> None:
client, result_root, result_id = _fixture(tmp_path)
(result_root / "review/review-000000000001.jpg").write_bytes(b"tampered")
response = client.get(f"/api/v1/laboratory/m48t/risk-quality/results/{result_id}")
assert response.status_code == 404
catalog = client.get("/api/v1/laboratory/m48t/risk-quality/results")
assert catalog.json()["items"] == []
assert catalog.json()["invalid_total"] == 1