SILVER/NODEDC_MISSION_CORE
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

feat(data): add read-only artifact health monitor

Browse Source
This commit is contained in:
DCCONSTRUCTIONS
2026-07-29 01:54:41 +03:00
parent 79eb4b46f7
commit 37b8930527
10 changed files with 1110 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/k1link/web/app.py
+13
View File
@@ -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",
src/k1link/web/artifact_health_api.py
+383
View File
@@ -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