diff --git a/apps/control-station/src/core/data/artifactHealth.ts b/apps/control-station/src/core/data/artifactHealth.ts new file mode 100644 index 0000000..71d13d6 --- /dev/null +++ b/apps/control-station/src/core/data/artifactHealth.ts @@ -0,0 +1,73 @@ +export interface ArtifactScope { + scope_id: string; + file_count: number; + logical_bytes: number; + modified_at_utc: string | null; +} + +export interface ArtifactHealth { + schema_version: "missioncore.artifact-health/v1"; + generated_at_utc: string; + overall_state: "live" | "attention" | "unavailable"; + inventory: { + state: "live" | "degraded" | "unavailable"; + scanned_at_utc: string; + scan_duration_ms: number; + file_count: number; + logical_bytes: number; + unreadable_entries: number; + files_delta_since_previous: number | null; + bytes_delta_since_previous: number | null; + scopes: ArtifactScope[]; + }; + exact_audit: { + state: "exact-current" | "exact-stale" | "unavailable"; + result_id: string | null; + audited_at_utc: string | null; + root_count: number; + file_count: number; + logical_bytes: number; + unique_content_bytes: number; + duplicate_bytes: number; + duplicate_fraction: number; + amplification_ratio: number; + duplicate_content_groups: number; + changed_roots: string[]; + limitations: string[]; + }; + content_references: { + state: "not-indexed"; + storage_model: "materialized-copies"; + storage_migration_authorized: false; + next_gate: string; + }; +} + +function isRecord(value: unknown): value is Record { + return Boolean(value) && typeof value === "object" && !Array.isArray(value); +} + +export async function fetchArtifactHealth( + signal?: AbortSignal, +): Promise { + const response = await fetch("/api/v1/data/artifact-health", { + method: "GET", + headers: { Accept: "application/json" }, + signal, + }); + if (!response.ok) { + throw new Error(`API здоровья данных вернул HTTP ${response.status}.`); + } + const document: unknown = await response.json(); + if ( + !isRecord(document) + || document.schema_version !== "missioncore.artifact-health/v1" + || !isRecord(document.inventory) + || !Array.isArray(document.inventory.scopes) + || !isRecord(document.exact_audit) + || !isRecord(document.content_references) + ) { + throw new Error("Ответ здоровья данных не соответствует контракту."); + } + return document as unknown as ArtifactHealth; +} diff --git a/apps/control-station/src/productModel.ts b/apps/control-station/src/productModel.ts index 46e01b6..16ec44c 100644 --- a/apps/control-station/src/productModel.ts +++ b/apps/control-station/src/productModel.ts @@ -21,6 +21,7 @@ export type WorkspaceKind = | "compute-modules" | "network-monitor" | "datasets" + | "artifact-health" | "lab-archive"; export type CapabilityStatus = "active" | "ready" | "contract" | "later"; @@ -495,6 +496,17 @@ export const workspaces: WorkspaceDefinition[] = [ kind: "datasets", groups: [], }, + { + id: "artifact-health", + root: "data", + label: "Здоровье данных", + title: "Здоровье данных", + eyebrow: "ДАННЫЕ / АРТЕФАКТЫ", + description: "Живой объём runtime, рост и точный статус дублирования контента.", + icon: "database", + kind: "artifact-health", + groups: [], + }, { id: "streams", root: "data", diff --git a/apps/control-station/src/styles.css b/apps/control-station/src/styles.css index fd665ac..46d2bee 100644 --- a/apps/control-station/src/styles.css +++ b/apps/control-station/src/styles.css @@ -12,3 +12,4 @@ @import "./styles/observation.css"; @import "./styles/environment-settings.css"; @import "./styles/system-telemetry.css"; +@import "./styles/artifact-health.css"; diff --git a/apps/control-station/src/styles/artifact-health.css b/apps/control-station/src/styles/artifact-health.css new file mode 100644 index 0000000..e69a780 --- /dev/null +++ b/apps/control-station/src/styles/artifact-health.css @@ -0,0 +1,237 @@ +.artifact-health-workspace { + display: grid; + min-width: 0; + gap: 1rem; + padding-bottom: 1rem; +} + +.artifact-health-lead, +.artifact-scope-panel > header, +.artifact-health-details header { + display: flex; + min-width: 0; + align-items: flex-start; + justify-content: space-between; + gap: 1.2rem; +} + +.artifact-health-lead { + padding: 0.25rem 0 0.35rem; +} + +.artifact-health-lead h2, +.artifact-health-lead p, +.artifact-scope-panel h3, +.artifact-scope-panel p, +.artifact-health-details h3, +.artifact-health-details p, +.artifact-health-state h3, +.artifact-health-state p { + margin: 0; +} + +.artifact-health-lead h2 { + margin-top: 0.4rem; + color: var(--nodedc-text-primary); + font-size: 1.42rem; + letter-spacing: -0.035em; +} + +.artifact-health-lead p, +.artifact-scope-panel p, +.artifact-health-details p, +.artifact-health-state p { + color: var(--nodedc-text-muted); + font-size: 0.66rem; + line-height: 1.5; +} + +.artifact-health-lead p { + max-width: 48rem; + margin-top: 0.42rem; +} + +.artifact-health-state { + display: grid; + justify-items: start; + gap: 0.65rem; +} + +.artifact-health-state h3 { + color: var(--nodedc-text-primary); + font-size: 1rem; +} + +.artifact-health-notice { + display: flex; + align-items: center; + gap: 0.7rem; + color: var(--nodedc-text-secondary); + font-size: 0.64rem; +} + +.artifact-health-metrics { + display: grid; + grid-template-columns: repeat(4, minmax(0, 1fr)); + gap: 0.65rem; +} + +.artifact-health-metrics > div { + display: grid; + min-width: 0; + gap: 0.28rem; +} + +.artifact-health-metrics span, +.artifact-health-metrics small, +.artifact-scope-list span, +.artifact-health-details dt { + color: var(--nodedc-text-muted); + font-size: 0.6rem; +} + +.artifact-health-metrics strong { + color: var(--nodedc-text-primary); + font-size: 1.18rem; + font-weight: 720; + letter-spacing: -0.035em; +} + +.artifact-scope-panel, +.artifact-health-details > div { + display: grid; + min-width: 0; + gap: 1rem; +} + +.artifact-scope-panel h3, +.artifact-health-details h3 { + margin-top: 0.36rem; + color: var(--nodedc-text-primary); + font-size: 1rem; + letter-spacing: -0.025em; +} + +.artifact-scope-panel p { + margin-top: 0.36rem; +} + +.artifact-scope-list { + display: grid; + gap: 0.22rem; + margin: 0; + padding: 0; + list-style: none; +} + +.artifact-scope-list li { + display: grid; + grid-template-columns: minmax(10rem, 0.7fr) minmax(12rem, 1.4fr) 6rem; + align-items: center; + gap: 0.8rem; + min-width: 0; + padding: 0.58rem 0.66rem; + border-radius: var(--nodedc-radius-control); + background: var(--station-panel-soft); +} + +.artifact-scope-list li > div:first-child { + display: grid; + min-width: 0; + gap: 0.16rem; +} + +.artifact-scope-list strong { + overflow: hidden; + color: var(--nodedc-text-primary); + font-size: 0.66rem; + font-weight: 680; + text-overflow: ellipsis; + white-space: nowrap; +} + +.artifact-scope-list li > strong { + text-align: right; +} + +.artifact-scope-list__measure { + overflow: hidden; + height: 0.34rem; + border-radius: 999px; + background: var(--station-panel-deep); +} + +.artifact-scope-list__measure span { + display: block; + min-width: 0.2rem; + height: 100%; + border-radius: inherit; + background: rgb(var(--nodedc-accent-rgb) / 0.78); +} + +.artifact-health-details { + display: grid; + grid-template-columns: repeat(2, minmax(0, 1fr)); + gap: 0.65rem; +} + +.artifact-health-details dl { + display: grid; + grid-template-columns: repeat(2, minmax(0, 1fr)); + gap: 0.45rem; + margin: 0; +} + +.artifact-health-details dl > div { + display: grid; + min-width: 0; + gap: 0.18rem; + padding: 0.66rem; + border-radius: var(--nodedc-radius-control); + background: var(--station-panel-soft); +} + +.artifact-health-details dd { + overflow: hidden; + margin: 0; + color: var(--nodedc-text-primary); + font-size: 0.66rem; + font-weight: 680; + text-overflow: ellipsis; + white-space: nowrap; +} + +@media (max-width: 980px) { + .artifact-health-metrics { + grid-template-columns: repeat(2, minmax(0, 1fr)); + } + + .artifact-health-details { + grid-template-columns: 1fr; + } +} + +@media (max-width: 680px) { + .artifact-health-lead, + .artifact-scope-panel > header, + .artifact-health-details header { + display: grid; + } + + .artifact-health-metrics { + grid-template-columns: 1fr; + } + + .artifact-scope-list li { + grid-template-columns: minmax(0, 1fr) 5.2rem; + } + + .artifact-scope-list__measure { + grid-column: 1 / -1; + grid-row: 2; + } + + .artifact-health-details dl { + grid-template-columns: 1fr; + } +} diff --git a/apps/control-station/src/workspaces/Workspaces.tsx b/apps/control-station/src/workspaces/Workspaces.tsx index b885dfc..ba6ffb8 100644 --- a/apps/control-station/src/workspaces/Workspaces.tsx +++ b/apps/control-station/src/workspaces/Workspaces.tsx @@ -11,7 +11,6 @@ import { Icon, StatusBadge, } from "@nodedc/ui-react"; - import { ObservationMedia, ObservationSourcePicker, @@ -44,11 +43,11 @@ import { finiteMetric, formatNumber, pipelineLatency, sourceModeLabel } from ".. import type { SceneSettings } from "../sceneSettings"; import type { WorkspaceRendererProps } from "./contracts"; import { DatasetGatewayWorkspace } from "./DatasetGatewayWorkspace"; +import { ArtifactHealthWorkspace } from "./data/ArtifactHealthWorkspace"; import { ContourHealthWorkspace } from "./ContourHealthWorkspace"; import { LaboratoryArchiveWorkspace } from "./laboratory/LaboratoryArchiveWorkspace"; import { ComputeModulesWorkspace } from "./system/ComputeModulesWorkspace"; import { NetworkWorkspace } from "./system/NetworkWorkspace"; - function statusTone(status: CapabilityStatus): "success" | "accent" | "warning" | "neutral" { if (status === "active") return "success"; if (status === "ready") return "accent"; @@ -1156,7 +1155,6 @@ function RecordingsWorkspace(props: WorkspaceRendererProps) { ); } - export function WorkspaceRenderer(props: WorkspaceRendererProps) { switch (props.definition.kind) { case "spatial": @@ -1185,6 +1183,8 @@ export function WorkspaceRenderer(props: WorkspaceRendererProps) { return ; case "datasets": return ; + case "artifact-health": + return ; case "lab-archive": return ( 0 ? "+" : "−"; + const absolute = Math.abs(value); + return unit === "files" + ? `${prefix}${formatInteger(absolute)} файлов` + : `${prefix}${formatBytes(absolute)}`; +} + +function errorLabel(error: unknown): string { + return error instanceof Error && error.message.trim() + ? error.message + : "Срез здоровья данных недоступен."; +} + +export function ArtifactHealthWorkspace() { + const [health, setHealth] = useState(null); + const [error, setError] = useState(null); + const [generation, setGeneration] = useState(0); + + useEffect(() => { + const controller = new AbortController(); + let timer: number | null = null; + const poll = () => { + void fetchArtifactHealth(controller.signal) + .then((document) => { + if (controller.signal.aborted) return; + setHealth(document); + setError(null); + }) + .catch((loadError: unknown) => { + if (!controller.signal.aborted) setError(errorLabel(loadError)); + }) + .finally(() => { + if (!controller.signal.aborted) { + timer = window.setTimeout(poll, POLL_MILLISECONDS); + } + }); + }; + poll(); + return () => { + controller.abort(); + if (timer !== null) window.clearTimeout(timer); + }; + }, [generation]); + + if (!health) { + return ( +
+
+
+ ДАННЫЕ / ЗДОРОВЬЕ АРТЕФАКТОВ +

Живой контур хранения

+

+ Метаданные локальных артефактов и последний точный content-аудит + без изменения или автоматического удаления данных. +

+
+
+ + + {error ? "Срез недоступен" : "Сканируем метаданные"} + +

{error ?? "Собираем файловый инвентарь"}

+

Содержимое больших файлов не читается и повторно не хешируется.

+ {error ? ( + + ) : null} + +
+ ); + } + + const largestScope = Math.max( + ...health.inventory.scopes.map((scope) => scope.logical_bytes), + 1, + ); + const exactCurrent = health.exact_audit.state === "exact-current"; + return ( +
+
+
+ ДАННЫЕ / ЗДОРОВЬЕ АРТЕФАКТОВ +

Живой контур хранения

+

+ Текущий объём runtime, рост между замерами и доказанное дублирование. + Контур остаётся только наблюдающим. +

+
+ + {health.overall_state === "live" ? "Стабильно" : "Требует оптимизации"} + +
+ + {error ? ( + + {error} + Показан последний успешный срез. + + ) : null} + +
+ + Runtime всего + {formatBytes(health.inventory.logical_bytes)} + {deltaLabel(health.inventory.bytes_delta_since_previous, "bytes")} + + + Файлов + {formatInteger(health.inventory.file_count)} + {deltaLabel(health.inventory.files_delta_since_previous, "files")} + + + Доказанные дубли + {formatBytes(health.exact_audit.duplicate_bytes)} + {formatPercent(health.exact_audit.duplicate_fraction)} audit-выборки + + + Content-групп + {formatInteger(health.exact_audit.duplicate_content_groups)} + Коэффициент {health.exact_audit.amplification_ratio.toFixed(2)}× + +
+ + +
+
+ LIVE INVENTORY +

Распределение runtime

+

+ Метаданные обновлены {formatTimestamp(health.inventory.scanned_at_utc)} + {" · "}{Math.round(health.inventory.scan_duration_ms)} мс +

+
+ + {health.inventory.state === "live" ? "Live" : "Частичный срез"} + +
+
    + {health.inventory.scopes.slice(0, 8).map((scope) => ( +
  1. +
    + {scope.scope_id} + {formatInteger(scope.file_count)} файлов +
    +
    +
    + {formatBytes(scope.logical_bytes)} +
    2. + ))} +
+ + +
+ +
+
+ E44 / EXACT SHA-256 AUDIT +

Доказанное дублирование

+
+ + {exactCurrent ? "Точный срез актуален" : "Нужен повторный аудит"} + +
+
+
Охвачено корней
{health.exact_audit.root_count}
+
Логический объём
{formatBytes(health.exact_audit.logical_bytes)}
+
Уникальный content
{formatBytes(health.exact_audit.unique_content_bytes)}
+
Дата аудита
{formatTimestamp(health.exact_audit.audited_at_utc)}
+
+ {health.exact_audit.changed_roots.length ? ( +

Изменены корни: {health.exact_audit.changed_roots.join(", ")}.

+ ) : ( +

Все 14 измеренных корней совпадают с точным аудитом по числу файлов и байтам.

+ )} + + + +
+
+ CONTENT REFERENCES +

Материализованные копии

+
+ Не индексировано +
+

+ Content-addressed ссылки ещё не включены. Миграция хранения не + авторизована: сначала выбираются классы дублей с максимальным + измеренным эффектом. +

+
+
Режим
Наблюдение
+
Удаление
Запрещено
+
Автомиграция
Запрещена
+
Следующий gate
Content refs / chunking
+
+ +
+
+ ); +} diff --git a/apps/control-station/test/artifactHealth.test.mjs b/apps/control-station/test/artifactHealth.test.mjs new file mode 100644 index 0000000..ba34c35 --- /dev/null +++ b/apps/control-station/test/artifactHealth.test.mjs @@ -0,0 +1,41 @@ +import assert from "node:assert/strict"; +import { readFile } from "node:fs/promises"; +import { test } from "node:test"; + +const sourceRoot = new URL("../src/", import.meta.url); + +async function read(relativePath) { + return readFile(new URL(relativePath, sourceRoot), "utf8"); +} + +test("artifact health remains a bounded read-only data feature", async () => { + const [ + productModel, + workspaceHub, + core, + workspace, + featureStyles, + styles, + ] = await Promise.all([ + read("productModel.ts"), + read("workspaces/Workspaces.tsx"), + read("core/data/artifactHealth.ts"), + read("workspaces/data/ArtifactHealthWorkspace.tsx"), + read("styles/artifact-health.css"), + read("styles.css"), + ]); + + assert.match(productModel, /kind: "artifact-health"/); + assert.match(workspaceHub, //); + assert.match(core, /\/api\/v1\/data\/artifact-health/); + assert.doesNotMatch(core, /@nodedc\/ui-react/); + assert.match(workspace, /fetchArtifactHealth/); + assert.match(workspace, /POLL_MILLISECONDS\s*=\s*30_000/); + assert.doesNotMatch(workspace, /delete|deduplicat|migration.*fetch/i); + assert.doesNotMatch( + featureStyles, + /\bborder:\s*1px\s+solid/, + "Artifact health must not add decorative perimeter outlines", + ); + assert.match(styles, /artifact-health\.css/); +}); diff --git a/src/k1link/web/app.py b/src/k1link/web/app.py index e39e9c0..cef5a91 100644 --- a/src/k1link/web/app.py +++ b/src/k1link/web/app.py @@ -33,6 +33,7 @@ from k1link.sessions import ( SessionStore, ) from k1link.web.advanced_laboratory_api import build_advanced_laboratory_router +from k1link.web.artifact_health_api import build_artifact_health_router from k1link.web.compute_contour_api import build_compute_contour_router from k1link.web.device_plugin_composition import load_installed_device_plugins from k1link.web.e30_engineering_api import build_e30_engineering_router @@ -653,6 +654,18 @@ app.include_router( root_provider=lambda: REPOSITORY_ROOT / ".runtime" / "system", ) ) +app.include_router( + build_artifact_health_router( + runtime_root_provider=lambda: REPOSITORY_ROOT / ".runtime", + e44_results_root_provider=lambda: ( + REPOSITORY_ROOT + / ".runtime" + / "compute-experiments" + / "e44" + / "results" + ), + ) +) app.include_router( build_compute_contour_router( root_provider=lambda: REPOSITORY_ROOT / ".runtime" / "system", diff --git a/src/k1link/web/artifact_health_api.py b/src/k1link/web/artifact_health_api.py new file mode 100644 index 0000000..00d7b47 --- /dev/null +++ b/src/k1link/web/artifact_health_api.py @@ -0,0 +1,383 @@ +"""Read-only runtime artifact inventory and exact amplification audit status.""" + +from __future__ import annotations + +import json +import os +import threading +import time +from collections.abc import Callable +from datetime import UTC, datetime +from pathlib import Path +from typing import Any, Final, Literal + +from fastapi import APIRouter +from pydantic import BaseModel, ConfigDict, Field + +from k1link.compute.e44_data_amplification_audit import ( + E44DataAmplificationAuditError, + read_e44_data_amplification_audit, +) + +ARTIFACT_HEALTH_SCHEMA: Final = "missioncore.artifact-health/v1" +DEFAULT_CACHE_SECONDS: Final = 30.0 + +RootProvider = Callable[[], Path] + + +def _utc_now() -> str: + return datetime.now(UTC).isoformat().replace("+00:00", "Z") + + +def _utc_from_timestamp(timestamp: float) -> str: + return datetime.fromtimestamp(timestamp, UTC).isoformat().replace("+00:00", "Z") + + +class ArtifactScope(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + + scope_id: str + file_count: int = Field(ge=0) + logical_bytes: int = Field(ge=0) + modified_at_utc: str | None = None + + +class RuntimeArtifactInventory(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + + state: Literal["live", "degraded", "unavailable"] + scanned_at_utc: str + scan_duration_ms: float = Field(ge=0) + file_count: int = Field(ge=0) + logical_bytes: int = Field(ge=0) + unreadable_entries: int = Field(ge=0) + files_delta_since_previous: int | None = None + bytes_delta_since_previous: int | None = None + scopes: tuple[ArtifactScope, ...] + + +class ExactAmplificationAudit(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + + state: Literal["exact-current", "exact-stale", "unavailable"] + result_id: str | None = None + audited_at_utc: str | None = None + root_count: int = Field(default=0, ge=0) + file_count: int = Field(default=0, ge=0) + logical_bytes: int = Field(default=0, ge=0) + unique_content_bytes: int = Field(default=0, ge=0) + duplicate_bytes: int = Field(default=0, ge=0) + duplicate_fraction: float = Field(default=0, ge=0, le=1) + amplification_ratio: float = Field(default=1, ge=1) + duplicate_content_groups: int = Field(default=0, ge=0) + changed_roots: tuple[str, ...] = () + limitations: tuple[str, ...] = () + + +class ContentReferenceStatus(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + + state: Literal["not-indexed"] = "not-indexed" + storage_model: Literal["materialized-copies"] = "materialized-copies" + storage_migration_authorized: Literal[False] = False + next_gate: str + + +class ArtifactHealthDocument(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + + schema_version: Literal["missioncore.artifact-health/v1"] = ARTIFACT_HEALTH_SCHEMA + generated_at_utc: str + overall_state: Literal["live", "attention", "unavailable"] + inventory: RuntimeArtifactInventory + exact_audit: ExactAmplificationAudit + content_references: ContentReferenceStatus + + +class ArtifactHealthService: + """Cache metadata-only scans; never hash, mutate, deduplicate, or follow links.""" + + def __init__( + self, + runtime_root_provider: RootProvider, + e44_results_root_provider: RootProvider, + *, + cache_seconds: float = DEFAULT_CACHE_SECONDS, + ) -> None: + self._runtime_root_provider = runtime_root_provider + self._e44_results_root_provider = e44_results_root_provider + self._cache_seconds = max(cache_seconds, 0) + self._lock = threading.Lock() + self._cached_at = 0.0 + self._cached: ArtifactHealthDocument | None = None + + def snapshot(self) -> ArtifactHealthDocument: + now = time.monotonic() + with self._lock: + if ( + self._cached is not None + and now - self._cached_at < self._cache_seconds + ): + return self._cached + previous = self._cached.inventory if self._cached is not None else None + audit_source = self._read_latest_audit() + expected_roots = audit_source[2] if audit_source is not None else {} + inventory, audited_roots = self._scan_runtime(expected_roots, previous) + exact_audit = self._audit_document(audit_source, audited_roots) + overall_state: Literal["live", "attention", "unavailable"] + if inventory.state == "unavailable": + overall_state = "unavailable" + elif ( + inventory.state == "degraded" + or exact_audit.state != "exact-current" + or exact_audit.duplicate_bytes > 0 + ): + overall_state = "attention" + else: + overall_state = "live" + document = ArtifactHealthDocument( + generated_at_utc=_utc_now(), + overall_state=overall_state, + inventory=inventory, + exact_audit=exact_audit, + content_references=ContentReferenceStatus( + next_gate=( + "Select content-addressed references or chunking only from " + "measured dominant duplicate classes." + ) + ), + ) + self._cached = document + self._cached_at = now + return document + + def _read_latest_audit( + self, + ) -> tuple[dict[str, Any], dict[str, Any], dict[str, dict[str, int]]] | None: + results_root = self._e44_results_root_provider() + if not results_root.is_dir(): + return None + candidates = sorted( + ( + candidate + for candidate in results_root.glob("e44-data-amplification-*") + if candidate.is_dir() + ), + key=lambda candidate: candidate.stat().st_mtime_ns, + reverse=True, + ) + for candidate in candidates: + try: + audit = read_e44_data_amplification_audit(candidate) + identity = audit.manifest["identity"] + analysis = audit.report["analysis"] + expected: dict[str, dict[str, int]] = {} + for row in identity["artifact_roots"]: + metrics = analysis["roots"][row["label"]] + expected[str(row["root_name"])] = { + "files": int(metrics["files"]), + "logical_bytes": int(metrics["logical_bytes"]), + } + return audit.manifest, audit.report, expected + except ( + E44DataAmplificationAuditError, + KeyError, + OSError, + TypeError, + ValueError, + json.JSONDecodeError, + ): + continue + return None + + def _scan_runtime( + self, + expected_roots: dict[str, dict[str, int]], + previous: RuntimeArtifactInventory | None, + ) -> tuple[RuntimeArtifactInventory, dict[str, list[dict[str, int]]]]: + started = time.monotonic() + runtime_root = self._runtime_root_provider() + scanned_at = _utc_now() + if not runtime_root.is_dir(): + return ( + RuntimeArtifactInventory( + state="unavailable", + scanned_at_utc=scanned_at, + scan_duration_ms=(time.monotonic() - started) * 1000, + file_count=0, + logical_bytes=0, + unreadable_entries=0, + scopes=(), + ), + {}, + ) + + scopes: dict[str, dict[str, int | float]] = {} + audited: dict[str, list[dict[str, int]]] = { + root_name: [] for root_name in expected_roots + } + total_files = 0 + total_bytes = 0 + unreadable = 0 + + def visit( + directory: Path, + scope_id: str, + active_audits: tuple[dict[str, int], ...], + ) -> None: + nonlocal total_files, total_bytes, unreadable + audits = active_audits + if directory.name in expected_roots: + metrics = {"files": 0, "logical_bytes": 0} + audited[directory.name].append(metrics) + audits = (*audits, metrics) + try: + entries = tuple(os.scandir(directory)) + except OSError: + unreadable += 1 + return + for entry in entries: + try: + if entry.is_symlink(): + continue + if entry.is_dir(follow_symlinks=False): + visit(Path(entry.path), scope_id, audits) + continue + if not entry.is_file(follow_symlinks=False): + continue + stat_result = entry.stat(follow_symlinks=False) + except OSError: + unreadable += 1 + continue + byte_length = max(int(stat_result.st_size), 0) + modified = float(stat_result.st_mtime) + total_files += 1 + total_bytes += byte_length + scope = scopes.setdefault( + scope_id, + {"files": 0, "logical_bytes": 0, "modified": 0.0}, + ) + scope["files"] += 1 + scope["logical_bytes"] += byte_length + scope["modified"] = max(float(scope["modified"]), modified) + for metrics in audits: + metrics["files"] += 1 + metrics["logical_bytes"] += byte_length + + try: + for entry in os.scandir(runtime_root): + try: + if entry.is_symlink(): + continue + if entry.is_dir(follow_symlinks=False): + visit(Path(entry.path), entry.name, ()) + elif entry.is_file(follow_symlinks=False): + stat_result = entry.stat(follow_symlinks=False) + byte_length = max(int(stat_result.st_size), 0) + total_files += 1 + total_bytes += byte_length + scopes.setdefault( + "_root", + {"files": 0, "logical_bytes": 0, "modified": 0.0}, + ) + scopes["_root"]["files"] += 1 + scopes["_root"]["logical_bytes"] += byte_length + scopes["_root"]["modified"] = max( + float(scopes["_root"]["modified"]), + float(stat_result.st_mtime), + ) + except OSError: + unreadable += 1 + except OSError: + unreadable += 1 + + scope_documents = tuple( + ArtifactScope( + scope_id=scope_id, + file_count=int(metrics["files"]), + logical_bytes=int(metrics["logical_bytes"]), + modified_at_utc=( + _utc_from_timestamp(float(metrics["modified"])) + if float(metrics["modified"]) > 0 + else None + ), + ) + for scope_id, metrics in sorted( + scopes.items(), + key=lambda row: (-int(row[1]["logical_bytes"]), row[0]), + ) + ) + inventory = RuntimeArtifactInventory( + state="degraded" if unreadable else "live", + scanned_at_utc=scanned_at, + scan_duration_ms=(time.monotonic() - started) * 1000, + file_count=total_files, + logical_bytes=total_bytes, + unreadable_entries=unreadable, + files_delta_since_previous=( + total_files - previous.file_count if previous is not None else None + ), + bytes_delta_since_previous=( + total_bytes - previous.logical_bytes if previous is not None else None + ), + scopes=scope_documents, + ) + return inventory, audited + + def _audit_document( + self, + audit_source: tuple[ + dict[str, Any], dict[str, Any], dict[str, dict[str, int]] + ] + | None, + audited_roots: dict[str, list[dict[str, int]]], + ) -> ExactAmplificationAudit: + if audit_source is None: + return ExactAmplificationAudit(state="unavailable") + manifest, report, expected_roots = audit_source + changed_roots = tuple( + sorted( + root_name + for root_name, expected in expected_roots.items() + if expected not in audited_roots.get(root_name, ()) + ) + ) + analysis = report["analysis"] + return ExactAmplificationAudit( + state="exact-stale" if changed_roots else "exact-current", + result_id=str(report["result_id"]), + audited_at_utc=str(manifest["created_at_utc"]), + root_count=int(analysis["root_count"]), + file_count=int(analysis["file_count"]), + logical_bytes=int(analysis["logical_bytes"]), + unique_content_bytes=int(analysis["unique_content_bytes"]), + duplicate_bytes=int(analysis["duplicate_bytes"]), + duplicate_fraction=float(analysis["duplicate_fraction"]), + amplification_ratio=float(analysis["amplification_ratio"]), + duplicate_content_groups=int(analysis["duplicate_content_groups"]), + changed_roots=changed_roots, + limitations=tuple(str(item) for item in report.get("limitations", ())), + ) + + +def build_artifact_health_router( + *, + runtime_root_provider: RootProvider, + e44_results_root_provider: RootProvider, + cache_seconds: float = DEFAULT_CACHE_SECONDS, +) -> APIRouter: + router = APIRouter() + service = ArtifactHealthService( + runtime_root_provider, + e44_results_root_provider, + cache_seconds=cache_seconds, + ) + + @router.get( + "/api/v1/data/artifact-health", + response_model=ArtifactHealthDocument, + ) + def artifact_health() -> ArtifactHealthDocument: + return service.snapshot() + + return router diff --git a/tests/test_artifact_health_api.py b/tests/test_artifact_health_api.py new file mode 100644 index 0000000..35e716c --- /dev/null +++ b/tests/test_artifact_health_api.py @@ -0,0 +1,83 @@ +from __future__ import annotations + +import json +from pathlib import Path + +from k1link.compute.e44_data_amplification_audit import ( + build_e44_data_amplification_audit, +) +from k1link.web.artifact_health_api import ArtifactHealthService + + +def _write(path: Path, payload: bytes) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_bytes(payload) + + +def test_artifact_health_reports_live_inventory_and_exact_audit(tmp_path: Path) -> None: + runtime = tmp_path / ".runtime" + first = runtime / "compute-experiments" / "e30" / "result-a" + second = runtime / "compute-experiments" / "e40" / "result-b" + _write(first / "shared.bin", b"shared") + _write(first / "only-a.bin", b"a") + _write(second / "shared.bin", b"shared") + _write(second / "only-b.bin", b"bb") + e44_results = runtime / "compute-experiments" / "e44" / "results" + build_e44_data_amplification_audit( + artifact_roots={"e30": first, "e40": second}, + output_root=e44_results, + ) + + service = ArtifactHealthService( + lambda: runtime, + lambda: e44_results, + cache_seconds=0, + ) + document = service.snapshot() + + assert document.inventory.state == "live" + assert document.inventory.file_count >= 6 + assert document.inventory.logical_bytes > 0 + assert document.exact_audit.state == "exact-current" + assert document.exact_audit.duplicate_bytes == len(b"shared") + assert document.content_references.state == "not-indexed" + assert document.content_references.storage_migration_authorized is False + + +def test_artifact_health_marks_exact_audit_stale_after_source_change( + tmp_path: Path, +) -> None: + runtime = tmp_path / ".runtime" + first = runtime / "e30" / "result-a" + second = runtime / "e40" / "result-b" + _write(first / "a.bin", b"a") + _write(second / "b.bin", b"b") + e44_results = runtime / "e44" / "results" + audit = build_e44_data_amplification_audit( + artifact_roots={"e30": first, "e40": second}, + output_root=e44_results, + ) + manifest = json.loads((audit.result_root / "manifest.json").read_text()) + first_name = manifest["identity"]["artifact_roots"][0]["root_name"] + _write(first / "new.bin", b"new") + + document = ArtifactHealthService( + lambda: runtime, + lambda: e44_results, + cache_seconds=0, + ).snapshot() + + assert document.exact_audit.state == "exact-stale" + assert first_name in document.exact_audit.changed_roots + + +def test_artifact_health_reports_unavailable_runtime(tmp_path: Path) -> None: + document = ArtifactHealthService( + lambda: tmp_path / "missing", + lambda: tmp_path / "missing-e44", + cache_seconds=0, + ).snapshot() + + assert document.overall_state == "unavailable" + assert document.inventory.state == "unavailable" + assert document.exact_audit.state == "unavailable"