diff --git a/README.md b/README.md index 9dfb63b5f..eafa47cac 100644 --- a/README.md +++ b/README.md @@ -276,6 +276,7 @@ with Repo() as repo: | `add` | Track file(s) with optional provenance | | `status` | Show freshness of tracked files (data & deps) | | `diff` | Content diff with preprocessing support | +| `audit` | Blob classification, lineage, and cache analysis | | `cache` | Inspect cache (path, md5, dir) | | `cat` | View cached file contents | | `push` | Upload data to remote storage | @@ -299,6 +300,7 @@ with Repo() as repo: ### Added in DVX - `dvx run` - Parallel pipeline execution with per-file provenance - `dvx diff` preprocessing - Pipe through commands before diffing (with `{}` placeholder) +- `dvx audit` - Blob classification (input/generated/foreign), lineage, orphan detection - `dvx cache path/md5` - Cache introspection - `dvx cat` - View cached files directly - `dvx status --yaml` - Detailed status with hashes diff --git a/specs/blob-audit-lineage.md b/specs/blob-audit-lineage.md new file mode 100644 index 000000000..a5cc289db --- /dev/null +++ b/specs/blob-audit-lineage.md @@ -0,0 +1,158 @@ +# Blob Audit and Lineage Tracking + +## Problem + +DVX stores provenance per-artifact (`meta.computation` in `.dvc` files), but there's no way to query across artifacts: which blobs are used where, which are still necessary, which are orphaned, and what the full dependency graph looks like. DVC's `gc` operates at the hash level (keep referenced hashes, delete unreferenced), but doesn't reason about the *relationships* between blobs across commits. + +## Use Cases + +### 1. "What blobs does this commit need?" +Given a commit SHA, enumerate all `.dvc` files, their output hashes, and transitively all input hashes. Answer: "to fully reproduce this commit's state, you need these N blobs totaling X MB." + +### 2. "Where is this blob used?" +Given a blob hash (or path), find all commits/branches/tags that reference it — either as a direct output or as a transitive dependency. Answer: "this blob is referenced by 3 commits on `main` and 1 tag." + +### 3. "Which blobs are generated vs. input?" +Classify all blobs in the cache (or remote) by their provenance: +- **Input**: no computation, added directly via `dvx add` or `dvx import-url` +- **Generated**: has computation, output of `dvx run` +- **Foreign**: imported via `--no-download`, tracked by ETag but not cached locally +- **Orphaned**: in cache but not referenced by any `.dvc` file in any branch/tag/commit + +### 4. "What's the minimal cache for this branch?" +Given a branch, compute the minimal set of blobs needed: +- All input blobs (not reproducible) +- Generated blobs only if their inputs are unavailable +- Total size of irreducible inputs + +### 5. "Can I safely delete this remote blob?" +Before deleting a blob from S3, check: +- Is it an input blob? (If so, it's irreplaceable — don't delete unless another copy exists) +- Is it generated? (Safe to delete if inputs are available) +- Is it referenced by any commit? (If not, it's orphaned — safe to delete) + +## Implementation + +### Module structure + +``` +src/dvx/audit/ + __init__.py # Exports: scan_workspace, audit_artifact, find_orphans + model.py # BlobKind, Reproducibility, BlobInfo, AuditSummary + scan.py # Scanning and classification logic +src/dvx/cli/audit.py # Click command +``` + +### Data model (`model.py`) + +- **`BlobKind`** enum: `INPUT`, `GENERATED`, `FOREIGN`, `ORPHANED` +- **`Reproducibility`** enum: `REPRODUCIBLE`, `NOT_REPRODUCIBLE`, `UNKNOWN` +- **`BlobInfo`** dataclass: path, md5, size, kind, reproducible, cmd, deps, git_deps, in_local_cache, in_remote_cache, is_dir, nfiles +- **`AuditSummary`** dataclass: blobs list + computed aggregates (counts/sizes by kind, cache stats) + - `to_dict()` for JSON serialization — this is the data contract for the web UI + +### Classification logic (`scan.py`) + +`classify_blob(DVCFileInfo) → (BlobKind, Reproducibility)`: +- No md5 on outs → `FOREIGN` +- Has `cmd` → `GENERATED` + - `meta.reproducible: false` → `NOT_REPRODUCIBLE` + - `meta.reproducible: true` or absent → `REPRODUCIBLE` (positive default) +- No cmd → `INPUT` + +Reuses from existing code: +- `find_dvc_files()` from `cache.py` — discovers `.dvc` files +- `read_dvc_file()` from `dvc_files.py` — parses `.dvc` YAML +- `check_local_cache()` from `cache.py` — checks local cache existence +- `read_dir_manifest()` from `dvc_files.py` — reads directory manifest entries for orphan detection + +### Orphan detection (`find_orphans`) + +1. Collect all md5 hashes referenced by `.dvc` files (output hashes + dep hashes + dir manifest entries) +2. Walk `.dvc/cache/files/md5/` to enumerate all cached blobs +3. Return blobs not in the referenced set, with their sizes + +### CLI (`cli/audit.py`) + +``` +dvx audit # workspace summary +dvx audit # per-artifact lineage +dvx audit -o/--orphans # list unreferenced cache blobs +dvx audit -g/--graph # DOT dependency graph (colored by kind) +dvx audit --json # machine-readable (any mode) +dvx audit -r/--remote # also check remote cache +dvx audit -j/--jobs N # parallel workers for remote checks +``` + +### DVCFileInfo extension + +Added `reproducible: bool | None = None` field to `DVCFileInfo` in `dvc_files.py`, read from `meta.reproducible` in `read_dvc_file()`. + +### Output formats + +**Summary** (no args): +``` +Blobs in workspace: 34 + Input: 18 (810 MB) + Generated: 16 (80 MB, 14 reproducible) + Foreign: 0 + +Local cache: 30 of 34 (870 MB) + Missing: 4 (20 MB) +``` + +**Per-artifact** (`dvx audit `): +``` +Path: www/public/taxes-2025-lots.geojson +MD5: abc123... +Size: 22.8 MB +Type: Generated (reproducible) +Command: python -m jc_taxes.geojson_yearly --year 2025 --agg lot + +Dependencies (2 data + 2 code): + [data] data/taxrecords_enriched.parquet (def456...) + [code] src/jc_taxes/geojson_yearly.py (git: aabbcc) + +Cache: local=yes remote=not checked +``` + +**Orphans** (`dvx audit --orphans`): +``` +4 orphaned blob(s) (12 MB): + a433cf78... (8.2 MB) + b782ee41... (3.8 MB) +``` + +**JSON** (`dvx audit --json`): full `AuditSummary.to_dict()` serialized. + +**Graph** (`dvx audit --graph`): Graphviz DOT with kind-based node coloring: +- Input = palegreen, Generated = lightblue (lighter if reproducible), Foreign = gray dashed + +### Integration with `dvx dag` + +The graph output reuses the same conceptual DAG structure as `dvx dag` but colors nodes by `BlobKind` rather than by position (root/leaf/middle). For full graph features (clustering, Mermaid, HTML), use `dvx dag`. + +## ML Pipeline Considerations + +Large-scale ML training pipelines amplify the audit problem. A single training run may produce: +- **Checkpoints**: dozens of multi-GB model snapshots at different training steps +- **Evaluation artifacts**: metrics, predictions, confusion matrices at each checkpoint +- **Intermediate data**: preprocessed datasets, tokenized corpora, embedding caches + +The `meta.reproducible: false` opt-out is particularly important here — expensive training outputs should be explicitly marked non-reproducible to prevent accidental eviction. + +## Future (not this PR) + +- Cross-commit scanning (which commits reference which blobs) +- `dvx gc --evict-reproducible` (uses audit classification) — see `evictable-generated-blobs.md` +- Remote cache size analysis +- SQLite index for large repos +- UI extension: audit view tab in `ui/` (Vite + React + @xyflow/react) + +## Open Questions + +- How expensive is scanning `.dvc` files across all commits? For repos with thousands of commits and hundreds of `.dvc` files, this could be slow. The SQLite index amortizes this but adds maintenance burden. +- Should `dvx audit` also check dep *availability* (can this blob actually be regenerated right now)? This requires checking that all transitive inputs exist in cache or remote. +- Should lineage be queryable in the other direction? ("What outputs does this input produce?" — useful for impact analysis when an input changes.) +- For ML pipelines: should DVX integrate with experiment trackers (W&B, MLflow) to correlate blob lineage with training metrics? Or should it stay purely at the data layer and let users join the two? +- Should `dvx audit` support a `--cost` flag that estimates regeneration cost from historical run times logged in `meta.computation`? diff --git a/specs/evictable-generated-blobs.md b/specs/evictable-generated-blobs.md new file mode 100644 index 000000000..4429193de --- /dev/null +++ b/specs/evictable-generated-blobs.md @@ -0,0 +1,141 @@ +# Reproducible Generated Blobs + +## Problem + +DVX tracks both **generated** artifacts (have `meta.computation` with `cmd` + `deps`) and **input** artifacts (no computation, or foreign imports). Currently GC treats them identically — keep or evict based on git scope. But generated artifacts are fundamentally different: they can be **regenerated** from their inputs + code at any commit, making them safe to evict from cache even when referenced. + +This matters for projects like jc-taxes where ~450MB of GeoJSON is regenerated from deterministic code + parquet inputs. Every regen creates 16 new cache blobs; old versions accumulate even though they're reproducible from the corresponding code. + +## Concepts + +### Blob taxonomy + +| | **Has computation** | **No computation** | +|--|--------------------|--------------------| +| **DVX-tracked** (`.dvc` file) | Generated: `dvx run` output | Input: `dvx add`'d raw data | +| **Foreign** (`import-url --no-download`) | n/a | External: tracked by ETag, not cached | + +Generated blobs are the only ones that can be safely evicted without data loss, because they satisfy: +1. Their `.dvc` file records the exact `cmd` + `deps` + `git_deps` +2. At any git commit, the code (`git_deps`) and data inputs (`deps`) are pinned +3. Re-running the command with those inputs reproduces the output + +### Reproducibility spectrum + +Not all generated blobs are equally reproducible. The `reproducible` flag encodes a **confidence level about reproducibility**: + +| Level | Example | Reproducible? | Notes | +|-------|---------|--------------|-------| +| **Bit-reproducible** | jc-taxes GeoJSON from deterministic Python | Yes (default) | Same inputs + code → byte-identical output | +| **Semantically reproducible** | ML inference, deterministic but float-sensitive | Mostly | May differ at last bits due to hardware/library versions | +| **Reproducible with seed** | Single-GPU training with fixed seed | Cautiously | Reproduce given same hardware + library versions | +| **Non-reproducible** | Distributed training with weight-update races | No | Even with same inputs, output differs per run | + +Note: even "non-reproducible" training is becoming tractable — Google's Marin 8B (JAX + TPU) achieved bit-reproducibility for large-scale training, specifically to enable debugging loss spikes in expensive runs. But this is the exception; most distributed training has inherent nondeterminism from collective ops ordering. + +**Key insight**: reproducibility is the positive default. Blobs with `computation` are assumed reproducible unless explicitly marked `meta.reproducible: false`. A `dvx run` output that took 10,000 GPU-hours to produce should be explicitly marked non-reproducible by the user. + +### Reproducibility + +A generated blob is **reproducible** (and thus safe to evict) when: +- It has a `computation` block in its `.dvc` file +- It is NOT marked `meta.reproducible: false` (opt-out) +- All its `deps` are available (either cached or themselves reproducible) +- All its `git_deps` are available (always true — they're in git) + +An input blob is **never evictable** unless backed up to a remote (the existing `--safe` GC behavior). + +## Proposed Changes + +### 1. `.dvc` file: `reproducible` flag + +The `meta.reproducible` field is opt-out — present only when a generated blob is NOT reproducible: + +```yaml +outs: + - md5: abc123 + size: 22851069 + path: model-checkpoint.pt + +meta: + reproducible: false + computation: + cmd: python train.py --steps=50000 --seed=42 + deps: + data/training_set.parquet: def456 + git_deps: + train.py: aabbcc +``` + +- Default (absent): blobs with `computation` are assumed reproducible +- `reproducible: false` — blob cannot be reliably regenerated, keep in cache +- Classification is surfaced by `dvx audit` (see blob-audit-lineage spec) + +### 2. `dvx gc --evict-reproducible` + +New GC mode: evict blobs that are reproducible (have `computation` and not marked `reproducible: false`), keeping only input blobs and non-reproducible outputs. + +```bash +# Evict reproducible blobs not used in current workspace +dvx gc -w --evict-reproducible + +# Evict reproducible blobs from all commits (keep only inputs in cache) +dvx gc -A --evict-reproducible + +# Dry run +dvx gc -w --evict-reproducible --dry +``` + +Logic: +1. Identify all `.dvc` files in scope +2. For each, check if it has `computation` and is NOT `meta.reproducible: false` +3. If reproducible: remove from cache (local and/or remote per `--cloud`) +4. If not reproducible: keep (standard GC behavior) + +### 3. Non-reproducible marking + +For expensive or non-deterministic outputs, mark them explicitly: + +```yaml +# In the .dvc file, set meta.reproducible: false +meta: + reproducible: false + computation: + cmd: ... +``` + +This can be done manually in the YAML, or a future `dvx mark --not-reproducible ` command. + +### 4. `dvx regen` (optional, future) + +Regenerate evicted blobs on demand: + +```bash +# Regenerate a specific evicted blob +dvx regen www/public/taxes-2025-lots.geojson + +# Regenerate all evicted blobs needed for current workspace +dvx regen --workspace +``` + +Uses `meta.computation.cmd` with the pinned deps. This is essentially `dvx run` but specifically targeting blobs that were evicted. + +## Interaction with `dvx push` + +When pushing, reproducible blobs could optionally be skipped: + +```bash +# Push only non-reproducible (input) blobs +dvx push --skip-reproducible + +# Push everything (default, current behavior) +dvx push +``` + +This saves remote storage for blobs that are reproducible. The tradeoff: re-generating is slower than pulling from cache, but for deterministic pipelines the remote copy is pure redundancy. + +## Open Questions + +- Should there be a project-level default in `.dvc/config`? E.g., `core.assume_reproducible = true` (already the default behavior) or `false` to require explicit opt-in. +- Should there be a "confidence" level? E.g., `reproducible: true` (fully reproducible) vs `reproducible: expensive` (reproducible but costly)? +- How does this interact with remote storage billing? If reproducible blobs are skipped on push, the remote is smaller but regen requires local compute. diff --git a/src/dvx/audit/__init__.py b/src/dvx/audit/__init__.py new file mode 100644 index 000000000..935ebf130 --- /dev/null +++ b/src/dvx/audit/__init__.py @@ -0,0 +1,17 @@ +"""DVX blob audit — classification, lineage, and cache analysis.""" + +from dvx.audit.model import AuditSummary, BlobInfo, BlobKind, Reproducibility +from dvx.audit.repo_view import FilesystemRepoView, SnapshotRepoView +from dvx.audit.scan import audit_artifact, find_orphans, scan_workspace + +__all__ = [ + "AuditSummary", + "BlobInfo", + "BlobKind", + "FilesystemRepoView", + "Reproducibility", + "SnapshotRepoView", + "audit_artifact", + "find_orphans", + "scan_workspace", +] diff --git a/src/dvx/audit/model.py b/src/dvx/audit/model.py new file mode 100644 index 000000000..1032a3a7e --- /dev/null +++ b/src/dvx/audit/model.py @@ -0,0 +1,135 @@ +"""Data model for DVX blob audit and classification.""" + +from dataclasses import dataclass, field +from enum import Enum + + +class BlobKind(str, Enum): + """Classification of a DVC-tracked blob.""" + INPUT = "input" # dvx add'd, no computation + GENERATED = "generated" # dvx run output, has computation + FOREIGN = "foreign" # import-url --no-download, no md5 + ORPHANED = "orphaned" # in cache but unreferenced + + +class Reproducibility(str, Enum): + """Whether a generated blob can be reproduced from its inputs.""" + REPRODUCIBLE = "reproducible" + NOT_REPRODUCIBLE = "not_reproducible" + UNKNOWN = "unknown" + + +@dataclass +class BlobInfo: + """Audit information for a single DVC-tracked blob.""" + path: str + md5: str | None + size: int + kind: BlobKind + reproducible: Reproducibility + # Provenance + cmd: str | None = None + deps: dict[str, str] = field(default_factory=dict) + git_deps: dict[str, str] = field(default_factory=dict) + # Cache status + in_local_cache: bool = False + in_remote_cache: bool | None = None # None = not checked + # Directory metadata + is_dir: bool = False + nfiles: int | None = None + + def to_dict(self) -> dict: + """Serialize to dict for JSON output.""" + d = { + "path": self.path, + "md5": self.md5, + "size": self.size, + "kind": self.kind.value, + "reproducible": self.reproducible.value, + } + if self.cmd: + d["cmd"] = self.cmd + if self.deps: + d["deps"] = self.deps + if self.git_deps: + d["git_deps"] = self.git_deps + d["in_local_cache"] = self.in_local_cache + if self.in_remote_cache is not None: + d["in_remote_cache"] = self.in_remote_cache + d["is_dir"] = self.is_dir + if self.nfiles is not None: + d["nfiles"] = self.nfiles + return d + + +@dataclass +class AuditSummary: + """Aggregated audit results for a workspace.""" + blobs: list[BlobInfo] = field(default_factory=list) + + @property + def by_kind(self) -> dict[BlobKind, list[BlobInfo]]: + result: dict[BlobKind, list[BlobInfo]] = {} + for blob in self.blobs: + result.setdefault(blob.kind, []).append(blob) + return result + + @property + def total_count(self) -> int: + return len(self.blobs) + + @property + def total_size(self) -> int: + return sum(b.size for b in self.blobs) + + def count_by_kind(self, kind: BlobKind) -> int: + return sum(1 for b in self.blobs if b.kind == kind) + + def size_by_kind(self, kind: BlobKind) -> int: + return sum(b.size for b in self.blobs if b.kind == kind) + + @property + def reproducible_count(self) -> int: + return sum( + 1 for b in self.blobs + if b.kind == BlobKind.GENERATED + and b.reproducible == Reproducibility.REPRODUCIBLE + ) + + @property + def cached_count(self) -> int: + return sum(1 for b in self.blobs if b.in_local_cache) + + @property + def cached_size(self) -> int: + return sum(b.size for b in self.blobs if b.in_local_cache) + + @property + def missing_count(self) -> int: + return sum(1 for b in self.blobs if not b.in_local_cache) + + @property + def missing_size(self) -> int: + return sum(b.size for b in self.blobs if not b.in_local_cache) + + def to_dict(self) -> dict: + """Serialize to dict for JSON output.""" + return { + "blobs": [b.to_dict() for b in self.blobs], + "summary": { + "total": {"count": self.total_count, "size": self.total_size}, + "by_kind": { + kind.value: { + "count": self.count_by_kind(kind), + "size": self.size_by_kind(kind), + } + for kind in BlobKind + if self.count_by_kind(kind) > 0 + }, + "reproducible_count": self.reproducible_count, + "cache": { + "cached": {"count": self.cached_count, "size": self.cached_size}, + "missing": {"count": self.missing_count, "size": self.missing_size}, + }, + }, + } diff --git a/src/dvx/audit/repo_view.py b/src/dvx/audit/repo_view.py new file mode 100644 index 000000000..f0d8ff85f --- /dev/null +++ b/src/dvx/audit/repo_view.py @@ -0,0 +1,192 @@ +"""Abstraction over repo I/O for testable audit logic. + +A RepoView provides the minimal interface that audit/scan needs: +listing .dvc files, reading their contents, and checking cache status. + +Two implementations: +- FilesystemRepoView: reads from an actual repo on disk (production) +- SnapshotRepoView: reads from a JSON snapshot (testing) +""" + +from __future__ import annotations + +import json +from dataclasses import dataclass, field +from pathlib import Path +from typing import Protocol + +from dvx.run.dvc_files import DVCFileInfo + + +class RepoView(Protocol): + """Minimal repo interface for audit operations.""" + + def dvc_files(self, targets: list[str] | None = None) -> list[str]: + """List .dvc file paths (relative to repo root).""" + ... + + def read_dvc(self, path: str) -> DVCFileInfo | None: + """Read and parse a .dvc file.""" + ... + + def is_cached(self, md5: str) -> bool: + """Check if md5 is in local cache.""" + ... + + def is_remote_cached(self, md5: str, remote: str | None = None) -> bool: + """Check if md5 is in remote cache.""" + ... + + def cache_entries(self) -> list[tuple[str, int]]: + """List all (md5, size) in local cache.""" + ... + + def dir_manifest(self, md5: str) -> dict[str, str]: + """Read directory manifest: {relative_path: md5}.""" + ... + + +class FilesystemRepoView: + """RepoView backed by an actual filesystem repo.""" + + def __init__(self, root: Path | None = None): + if root is None: + from dvc.repo import Repo as DVCRepo + try: + root = Path(DVCRepo.find_root()) + except Exception: + root = Path.cwd() + self.root = root + + def dvc_files(self, targets: list[str] | None = None) -> list[str]: + from dvx.cache import find_dvc_files + return find_dvc_files(targets) + + def read_dvc(self, path: str) -> DVCFileInfo | None: + from dvx.run.dvc_files import read_dvc_file + return read_dvc_file(Path(path)) + + def is_cached(self, md5: str) -> bool: + from dvx.cache import check_local_cache + return check_local_cache(md5) + + def is_remote_cached(self, md5: str, remote: str | None = None) -> bool: + from dvx.cache import check_remote_cache + return check_remote_cache(md5, remote) + + def cache_entries(self) -> list[tuple[str, int]]: + cache_dir = self.root / ".dvc" / "cache" / "files" / "md5" + if not cache_dir.exists(): + return [] + entries = [] + for prefix_dir in sorted(cache_dir.iterdir()): + if not prefix_dir.is_dir() or len(prefix_dir.name) != 2: + continue + for cache_file in sorted(prefix_dir.iterdir()): + if not cache_file.is_file(): + continue + name = cache_file.name + is_dir_manifest = name.endswith(".dir") + if is_dir_manifest: + name = name[:-4] + md5 = prefix_dir.name + name + entries.append((md5, cache_file.stat().st_size)) + return entries + + def dir_manifest(self, md5: str) -> dict[str, str]: + from dvx.run.dvc_files import read_dir_manifest + cache_dir = self.root / ".dvc" / "cache" / "files" / "md5" + return read_dir_manifest(md5, cache_dir) + + +@dataclass +class SnapshotRepoView: + """RepoView backed by a JSON snapshot for testing. + + Load from the snapshot files captured by audit tooling: + - dvc-contents.json: array of {file, md5, size, path, deps, meta, ...} + - cache-files.txt: "SIZE .dvc/cache/files/md5/XX/YYYYYY" per line + + Usage: + view = SnapshotRepoView.load(Path("tmp/crashes-snapshot")) + summary = scan_workspace(view=view) + """ + + entries: list[dict] = field(default_factory=list) + cache: dict[str, int] = field(default_factory=dict) # {md5: size} + remote_cache: set[str] = field(default_factory=set) + manifests: dict[str, dict[str, str]] = field(default_factory=dict) # {dir_md5: {path: md5}} + + @classmethod + def load(cls, snapshot_dir: Path) -> SnapshotRepoView: + """Load from a snapshot directory.""" + entries = [] + contents_path = snapshot_dir / "dvc-contents.json" + if contents_path.exists(): + entries = json.loads(contents_path.read_text()) + + cache: dict[str, int] = {} + cache_path = snapshot_dir / "cache-files.txt" + if cache_path.exists(): + for line in cache_path.read_text().splitlines(): + parts = line.strip().split() + if len(parts) != 2: + continue + size_str, path = parts + segs = path.split("/") + if len(segs) >= 6: + md5 = segs[4] + segs[5] + cache[md5] = int(size_str) + + return cls(entries=entries, cache=cache) + + def dvc_files(self, targets: list[str] | None = None) -> list[str]: + files = [e["file"] for e in self.entries if "file" in e] + if targets: + target_set = set(targets) + # Support matching by .dvc path or by output path + files = [ + f for f in files + if f in target_set or f.removesuffix(".dvc") in target_set + ] + return files + + def read_dvc(self, path: str) -> DVCFileInfo | None: + for e in self.entries: + if e.get("file") == path: + return self._entry_to_info(e) + return None + + def is_cached(self, md5: str) -> bool: + clean = md5.removesuffix(".dir") + return clean in self.cache + + def is_remote_cached(self, md5: str, remote: str | None = None) -> bool: + clean = md5.removesuffix(".dir") + return clean in self.remote_cache + + def cache_entries(self) -> list[tuple[str, int]]: + return list(self.cache.items()) + + def dir_manifest(self, md5: str) -> dict[str, str]: + return self.manifests.get(md5, {}) + + @staticmethod + def _entry_to_info(e: dict) -> DVCFileInfo: + meta = e.get("meta", {}) + computation = meta.get("computation", {}) + md5_raw = e.get("md5", "") + is_dir = md5_raw.endswith(".dir") + md5 = md5_raw[:-4] if is_dir else md5_raw + return DVCFileInfo( + path=e.get("path", ""), + md5=md5, + size=e.get("size", 0), + cmd=computation.get("cmd"), + deps=computation.get("deps") or {}, + git_deps=computation.get("git_deps") or {}, + nfiles=e.get("nfiles"), + is_dir=is_dir, + reproducible=meta.get("reproducible"), + git_tracked=bool(meta.get("git_tracked")), + ) diff --git a/src/dvx/audit/scan.py b/src/dvx/audit/scan.py new file mode 100644 index 000000000..bcc64d7fb --- /dev/null +++ b/src/dvx/audit/scan.py @@ -0,0 +1,182 @@ +"""Scanning and classification logic for DVX blob audit.""" + +from __future__ import annotations + +from pathlib import Path +from typing import TYPE_CHECKING + +from dvx.audit.model import ( + AuditSummary, + BlobInfo, + BlobKind, + Reproducibility, +) +from dvx.run.dvc_files import DVCFileInfo + +if TYPE_CHECKING: + from dvx.audit.repo_view import RepoView + + +def classify_blob(info: DVCFileInfo) -> tuple[BlobKind, Reproducibility]: + """Classify a blob by its DVCFileInfo. + + Returns: + (kind, reproducibility) tuple + """ + if not info.md5: + return BlobKind.FOREIGN, Reproducibility.UNKNOWN + if info.cmd: + # Generated blob — check reproducibility + if info.reproducible is False: + repro = Reproducibility.NOT_REPRODUCIBLE + elif info.reproducible is True: + repro = Reproducibility.REPRODUCIBLE + else: + # Default: blobs with computation are assumed reproducible + repro = Reproducibility.REPRODUCIBLE + return BlobKind.GENERATED, repro + return BlobKind.INPUT, Reproducibility.UNKNOWN + + +def _default_view() -> RepoView: + from dvx.audit.repo_view import FilesystemRepoView + return FilesystemRepoView() + + +def _blob_info_from_dvc( + info: DVCFileInfo, + view: RepoView, + check_remote: bool = False, + remote: str | None = None, +) -> BlobInfo: + """Build a BlobInfo from a parsed DVCFileInfo.""" + kind, repro = classify_blob(info) + + in_local = False + if info.md5: + md5_for_cache = f"{info.md5}.dir" if info.is_dir else info.md5 + in_local = view.is_cached(md5_for_cache) + + in_remote: bool | None = None + if check_remote and info.md5: + md5_for_cache = f"{info.md5}.dir" if info.is_dir else info.md5 + in_remote = view.is_remote_cached(md5_for_cache, remote) + + return BlobInfo( + path=info.path, + md5=info.md5 or None, + size=info.size, + kind=kind, + reproducible=repro, + cmd=info.cmd, + deps=info.deps, + git_deps=info.git_deps, + in_local_cache=in_local, + in_remote_cache=in_remote, + is_dir=info.is_dir, + nfiles=info.nfiles, + ) + + +def scan_workspace( + targets: list[str] | None = None, + remote: str | None = None, + check_remote: bool = False, + view: RepoView | None = None, +) -> AuditSummary: + """Scan workspace and classify all tracked blobs. + + Args: + targets: Specific targets to scan (None = all .dvc files) + remote: Remote name for cache checks + check_remote: Whether to check remote cache status + view: RepoView to use (default: FilesystemRepoView) + + Returns: + AuditSummary with classified blobs + """ + if view is None: + view = _default_view() + + dvc_files = view.dvc_files(targets) + blobs: list[BlobInfo] = [] + + for dvc_file in dvc_files: + info = view.read_dvc(dvc_file) + if info is None: + continue + blob = _blob_info_from_dvc(info, view=view, check_remote=check_remote, remote=remote) + blobs.append(blob) + + return AuditSummary(blobs=blobs) + + +def audit_artifact( + path: str, + remote: str | None = None, + check_remote: bool = False, + view: RepoView | None = None, +) -> BlobInfo | None: + """Get detailed audit info for a single artifact. + + Args: + path: Path to the artifact (or its .dvc file) + remote: Remote name for cache checks + check_remote: Whether to check remote cache + view: RepoView to use (default: FilesystemRepoView) + + Returns: + BlobInfo or None if not found + """ + if view is None: + view = _default_view() + + info = view.read_dvc(path) + if info is None: + return None + return _blob_info_from_dvc(info, view=view, check_remote=check_remote, remote=remote) + + +def find_orphans( + root: Path | None = None, + view: RepoView | None = None, +) -> list[tuple[str, int]]: + """Find cache blobs not referenced by any .dvc file. + + Walks the cache and diffs against all hashes referenced + by .dvc files (including directory manifest entries). + + Args: + root: Repository root (ignored if view is provided) + view: RepoView to use (default: FilesystemRepoView) + + Returns: + List of (md5, size) for orphaned blobs + """ + if view is None: + view = _default_view() + + # Collect all hashes referenced by .dvc files + referenced: set[str] = set() + dvc_files = view.dvc_files() + for dvc_file in dvc_files: + info = view.read_dvc(dvc_file) + if info is None or not info.md5: + continue + referenced.add(info.md5) + # Also add hashes from deps (they may be cached too) + for dep_md5 in info.deps.values(): + referenced.add(dep_md5) + # For directories, add all file hashes from manifest + if info.is_dir: + manifest = view.dir_manifest(info.md5) + for file_md5 in manifest.values(): + referenced.add(file_md5) + + # Diff cache against referenced set + orphans: list[tuple[str, int]] = [] + for md5, size in view.cache_entries(): + if md5 not in referenced: + orphans.append((md5, size)) + + return orphans diff --git a/src/dvx/cli/audit.py b/src/dvx/cli/audit.py new file mode 100644 index 000000000..9925d790b --- /dev/null +++ b/src/dvx/cli/audit.py @@ -0,0 +1,246 @@ +"""DVX audit command — blob classification, lineage, and cache analysis.""" + +import json + +import click + + +def _format_size(size_bytes: int) -> str: + """Format bytes as human-readable size.""" + if size_bytes >= 1_000_000_000: + return f"{size_bytes / 1_000_000_000:.1f} GB" + if size_bytes >= 1_000_000: + return f"{size_bytes / 1_000_000:.1f} MB" + if size_bytes >= 1_000: + return f"{size_bytes / 1_000:.1f} KB" + return f"{size_bytes} B" + + +def _print_summary(summary): + """Print workspace audit summary.""" + from dvx.audit.model import BlobKind, Reproducibility + + click.echo(f"Blobs in workspace: {summary.total_count}") + + input_count = summary.count_by_kind(BlobKind.INPUT) + input_size = summary.size_by_kind(BlobKind.INPUT) + if input_count: + click.echo(f" Input: {input_count} ({_format_size(input_size)})") + + gen_count = summary.count_by_kind(BlobKind.GENERATED) + gen_size = summary.size_by_kind(BlobKind.GENERATED) + if gen_count: + repro = summary.reproducible_count + repro_str = f", {repro} reproducible" if repro else "" + click.echo(f" Generated: {gen_count} ({_format_size(gen_size)}{repro_str})") + + foreign_count = summary.count_by_kind(BlobKind.FOREIGN) + click.echo(f" Foreign: {foreign_count}") + + click.echo() + click.echo( + f"Local cache: {summary.cached_count} of {summary.total_count} " + f"({_format_size(summary.cached_size)})" + ) + if summary.missing_count: + click.echo( + f" Missing: {summary.missing_count} " + f"({_format_size(summary.missing_size)})" + ) + + +def _print_artifact(blob): + """Print per-artifact lineage.""" + from dvx.audit.model import BlobKind, Reproducibility + + click.echo(f"Path: {blob.path}") + if blob.md5: + click.echo(f"MD5: {blob.md5}") + click.echo(f"Size: {_format_size(blob.size)}") + + kind_str = blob.kind.value.capitalize() + if blob.kind == BlobKind.GENERATED: + repro_str = blob.reproducible.value.replace("_", " ") + kind_str = f"{kind_str} ({repro_str})" + click.echo(f"Type: {kind_str}") + + if blob.cmd: + click.echo(f"Command: {blob.cmd}") + + data_deps = {k: v for k, v in blob.deps.items()} + code_deps = {k: v for k, v in blob.git_deps.items()} + total_deps = len(data_deps) + len(code_deps) + if total_deps: + click.echo() + click.echo(f"Dependencies ({len(data_deps)} data + {len(code_deps)} code):") + for dep_path, dep_md5 in sorted(data_deps.items()): + click.echo(f" [data] {dep_path} ({dep_md5[:8]}...)") + for dep_path, dep_sha in sorted(code_deps.items()): + click.echo(f" [code] {dep_path} (git: {dep_sha[:6]})") + + click.echo() + local_str = "yes" if blob.in_local_cache else "no" + remote_str = "not checked" + if blob.in_remote_cache is True: + remote_str = "yes" + elif blob.in_remote_cache is False: + remote_str = "no" + click.echo(f"Cache: local={local_str} remote={remote_str}") + + +def _print_orphans(orphans): + """Print orphaned cache blobs.""" + if not orphans: + click.echo("No orphaned blobs found.") + return + + total_size = sum(size for _, size in orphans) + click.echo(f"{len(orphans)} orphaned blob(s) ({_format_size(total_size)}):") + for md5, size in orphans: + click.echo(f" {md5} ({_format_size(size)})") + + +def _format_graph(summary): + """Format audit as Graphviz DOT with kind-based coloring.""" + from dvx.audit.model import BlobKind, Reproducibility + + lines = ["digraph DVX_Audit {"] + lines.append(" rankdir=TB;") + lines.append(" node [shape=box, style=filled];") + lines.append("") + + for blob in summary.blobs: + escaped = blob.path.replace('"', '\\"') + label = escaped + if blob.is_dir: + label += "/" + + if blob.kind == BlobKind.INPUT: + color = "palegreen" + elif blob.kind == BlobKind.GENERATED: + if blob.reproducible == Reproducibility.REPRODUCIBLE: + color = "lightblue" + else: + color = "steelblue1" + elif blob.kind == BlobKind.FOREIGN: + color = "lightgray" + else: + color = "lightyellow" + + style = 'style="filled,dashed"' if blob.kind == BlobKind.FOREIGN else "style=filled" + lines.append(f' "{escaped}" [label="{label}", fillcolor={color}, {style}];') + + lines.append("") + + # Add edges from deps + blob_paths = {b.path for b in summary.blobs} + for blob in summary.blobs: + escaped_path = blob.path.replace('"', '\\"') + for dep_path in sorted(blob.deps.keys()): + escaped_dep = dep_path.replace('"', '\\"') + if dep_path not in blob_paths: + # External dep node (not a tracked blob) + lines.append( + f' "{escaped_dep}" [label="{escaped_dep}", ' + f'style=dashed, color=gray, fontcolor=gray];' + ) + blob_paths.add(dep_path) + lines.append(f' "{escaped_dep}" -> "{escaped_path}";') + + lines.append("") + lines.append(" // Legend") + lines.append(' subgraph cluster_legend {') + lines.append(' label="Legend"; style=rounded;') + lines.append(' legend_input [label="Input", fillcolor=palegreen, style=filled];') + lines.append(' legend_gen [label="Generated\\n(reproducible)", fillcolor=lightblue, style=filled];') + lines.append(' legend_foreign [label="Foreign", fillcolor=lightgray, style="filled,dashed"];') + lines.append(' legend_input -> legend_gen -> legend_foreign [style=invis];') + lines.append(" }") + lines.append("}") + return "\n".join(lines) + + +@click.command("audit") +@click.argument("targets", nargs=-1) +@click.option("--json", "output_json", is_flag=True, help="Machine-readable JSON output.") +@click.option("-g", "--graph", is_flag=True, help="Output DOT dependency graph colored by kind.") +@click.option("-j", "--jobs", type=int, default=None, help="Parallel workers for remote checks.") +@click.option("-o", "--orphans", is_flag=True, help="List unreferenced cache blobs.") +@click.option("-r", "--remote", default=None, help="Also check remote cache.") +@click.option("-S", "--snapshot", type=click.Path(exists=True), help="Load from snapshot directory (testing).") +def audit(targets, output_json, graph, jobs, orphans, remote, snapshot): + """Audit workspace blobs: classification, lineage, and cache analysis. + + With no arguments, prints a workspace summary. + With a path argument, shows per-artifact lineage. + With --orphans, lists unreferenced cache blobs. + With --graph, outputs a DOT dependency graph colored by blob kind. + + \b + Examples: + dvx audit # workspace summary + dvx audit some/artifact # per-artifact lineage + dvx audit --orphans # find orphans + dvx audit --json # JSON output + dvx audit --graph | dot -Tsvg # colored DAG + dvx audit -r myremote # also check remote cache + dvx audit -S tmp/crashes-snapshot # from snapshot + """ + from pathlib import Path + + from dvx.audit.scan import audit_artifact, find_orphans, scan_workspace + + view = None + if snapshot: + from dvx.audit.repo_view import SnapshotRepoView + view = SnapshotRepoView.load(Path(snapshot)) + + check_remote = remote is not None + + if orphans: + orphan_list = find_orphans(view=view) + if output_json: + click.echo(json.dumps( + [{"md5": md5, "size": size} for md5, size in orphan_list], + indent=2, + )) + else: + _print_orphans(orphan_list) + return + + target_list = list(targets) if targets else None + + # Per-artifact mode: single target + if target_list and len(target_list) == 1 and not graph: + blob = audit_artifact( + target_list[0], + remote=remote, + check_remote=check_remote, + view=view, + ) + if blob is None: + raise click.ClickException(f"No .dvc file found for: {target_list[0]}") + if output_json: + click.echo(json.dumps(blob.to_dict(), indent=2)) + else: + _print_artifact(blob) + return + + # Workspace summary / graph mode + summary = scan_workspace( + targets=target_list, + remote=remote, + check_remote=check_remote, + view=view, + ) + + if not summary.blobs: + click.echo("No .dvc files found in workspace.", err=True) + return + + if output_json: + click.echo(json.dumps(summary.to_dict(), indent=2)) + elif graph: + click.echo(_format_graph(summary)) + else: + _print_summary(summary) diff --git a/src/dvx/cli/main.py b/src/dvx/cli/main.py index bcace7304..11e8005e6 100644 --- a/src/dvx/cli/main.py +++ b/src/dvx/cli/main.py @@ -436,6 +436,15 @@ def remote(ctx, args): cli.add_command(dag) +# ============================================================================= +# Audit - from audit module +# ============================================================================= + +from .audit import audit + +cli.add_command(audit) + + # ============================================================================= # Shell Integration # ============================================================================= diff --git a/src/dvx/run/dvc_files.py b/src/dvx/run/dvc_files.py index 372df3246..9c02c047c 100644 --- a/src/dvx/run/dvc_files.py +++ b/src/dvx/run/dvc_files.py @@ -196,6 +196,8 @@ class DVCFileInfo: # Directory metadata nfiles: int | None = None is_dir: bool = False + # Reproducibility (from meta.reproducible; None = unset) + reproducible: bool | None = None # Git-tracked import (file in Git, not DVC cache) git_tracked: bool = False # Legacy field for backward compatibility @@ -253,6 +255,8 @@ def read_dvc_file(output_path: Path) -> DVCFileInfo | None: # Directory metadata nfiles=out.get("nfiles"), is_dir=is_dir, + # Reproducibility flag + reproducible=meta.get("reproducible"), # Git-tracked import git_tracked=bool(meta.get("git_tracked")), stage=meta.get("stage"), # Legacy only