From e305fe9e523af999a59d72bd9b079a163ed66094 Mon Sep 17 00:00:00 2001 From: smurftyy Date: Wed, 8 Jul 2026 23:47:35 +0100 Subject: [PATCH 1/8] chore:bump the schema version to 5 --- src/devklean/deletion/metadata.py | 57 +++++++++++++++++++++++++++---- 1 file changed, 51 insertions(+), 6 deletions(-) diff --git a/src/devklean/deletion/metadata.py b/src/devklean/deletion/metadata.py index 50c272b..baff86b 100644 --- a/src/devklean/deletion/metadata.py +++ b/src/devklean/deletion/metadata.py @@ -14,7 +14,9 @@ # the write side and this validation can't drift; a record with any other value # was written by a removed backend and is treated as corrupt. TRASH_STRATEGY = "trash" -SCHEMA_VERSION = 4 +# Bumped for the archive dict gaining compressed/original_size/compressed_size; +# the new fields are optional on read so schema_version <= 4 records still parse. +SCHEMA_VERSION = 5 @dataclass(frozen=True) @@ -35,12 +37,25 @@ def to_dict(self) -> dict[str, object]: class DeletionArchive: path: str format: str + # True whenever this record was produced by the compress-before-trash path + # (the only path that creates DeletionArchive today). Kept as an explicit + # field rather than inferred from archive-presence so future callers can + # record an archive without implying it was compressed. + compressed: bool = True + original_size: int | None = None + compressed_size: int | None = None def to_dict(self) -> dict[str, object]: - return { + payload: dict[str, object] = { "path": self.path, "format": self.format, + "compressed": self.compressed, } + if self.original_size is not None: + payload["original_size"] = self.original_size + if self.compressed_size is not None: + payload["compressed_size"] = self.compressed_size + return payload @dataclass(frozen=True) @@ -132,7 +147,26 @@ def _parse_record(data: dict[str, object]) -> DeletionMetadataRecord: archive_format = archive_data.get("format") if not isinstance(archive_path, str) or not isinstance(archive_format, str): raise ValueError("missing or wrong-typed archive fields") - archive = DeletionArchive(path=archive_path, format=archive_format) + + # compressed/original_size/compressed_size postdate schema_version 4; + # older records simply lack them, so absence is not an error. + compressed = archive_data.get("compressed", True) + original_size = archive_data.get("original_size") + compressed_size = archive_data.get("compressed_size") + if not isinstance(compressed, bool): + raise ValueError("archive 'compressed' field must be a boolean") + if original_size is not None and not isinstance(original_size, int): + raise ValueError("archive 'original_size' field must be an integer") + if compressed_size is not None and not isinstance(compressed_size, int): + raise ValueError("archive 'compressed_size' field must be an integer") + + archive = DeletionArchive( + path=archive_path, + format=archive_format, + compressed=compressed, + original_size=original_size, + compressed_size=compressed_size, + ) if strategy != TRASH_STRATEGY: raise ValueError(f"unrecognized strategy {strategy!r}") @@ -203,7 +237,7 @@ def record_successes( items: Sequence[CleanableItem], result: DeleteResult, strategy: str, - archives: Mapping[str, DeletionArchive | Mapping[str, str]] | None = None, + archives: Mapping[str, DeletionArchive | Mapping[str, object]] | None = None, ) -> None: deleted_paths = set(result.deleted) if not deleted_paths: @@ -241,7 +275,9 @@ def record_successes( path.write_text(json.dumps(record.to_dict(), indent=2) + "\n", encoding="utf-8") -def _coerce_archive(value: DeletionArchive | Mapping[str, str] | None) -> DeletionArchive | None: +def _coerce_archive( + value: DeletionArchive | Mapping[str, object] | None, +) -> DeletionArchive | None: if value is None: return None if isinstance(value, DeletionArchive): @@ -250,4 +286,13 @@ def _coerce_archive(value: DeletionArchive | Mapping[str, str] | None) -> Deleti format_name = value.get("format") if not isinstance(path, str) or not isinstance(format_name, str): raise ValueError("archive metadata must contain string path and format") - return DeletionArchive(path=path, format=format_name) + compressed = value.get("compressed", True) + original_size = value.get("original_size") + compressed_size = value.get("compressed_size") + return DeletionArchive( + path=path, + format=format_name, + compressed=bool(compressed), + original_size=original_size if isinstance(original_size, int) else None, + compressed_size=compressed_size if isinstance(compressed_size, int) else None, + ) From e553909328bea95264c50bd0912173392962c4f4 Mon Sep 17 00:00:00 2001 From: smurftyy Date: Wed, 8 Jul 2026 23:55:28 +0100 Subject: [PATCH 2/8] add stattic refistry for development artifacts signatures --- src/devklean/signatures/__init__.py | 25 ++++ src/devklean/signatures/registry.py | 202 ++++++++++++++++++++++++++++ 2 files changed, 227 insertions(+) create mode 100644 src/devklean/signatures/__init__.py create mode 100644 src/devklean/signatures/registry.py diff --git a/src/devklean/signatures/__init__.py b/src/devklean/signatures/__init__.py new file mode 100644 index 0000000..e70c2b4 --- /dev/null +++ b/src/devklean/signatures/__init__.py @@ -0,0 +1,25 @@ +from __future__ import annotations + +from importlib import import_module + +__all__ = [ + "ArtifactSignature", + "RiskLevel", + "SIGNATURE_REGISTRY", + "lookup", + "match_signature", +] + + +def __getattr__(name: str): + if name == "ArtifactSignature": + return import_module("devklean.signatures.registry").ArtifactSignature + if name == "RiskLevel": + return import_module("devklean.signatures.registry").RiskLevel + if name == "SIGNATURE_REGISTRY": + return import_module("devklean.signatures.registry").SIGNATURE_REGISTRY + if name == "lookup": + return import_module("devklean.signatures.registry").lookup + if name == "match_signature": + return import_module("devklean.signatures.registry").match_signature + raise AttributeError(f"module {__name__!r} has no attribute {name!r}") diff --git a/src/devklean/signatures/registry.py b/src/devklean/signatures/registry.py new file mode 100644 index 0000000..de51e20 --- /dev/null +++ b/src/devklean/signatures/registry.py @@ -0,0 +1,202 @@ +"""Static, deterministic registry of known development-artifact signatures. + +This is the single source of truth ``devklean explain`` and ``devklean +analyze`` read from. Every field on every entry is fixed data written by a +maintainer — there is no model call, heuristic scoring pass, or runtime +inference anywhere in this file. ``risk`` and ``confidence`` are looked up +verbatim for a matched entry; an unmatched path gets no entry and therefore +no verdict, by construction (``lookup``/``match_signature`` return ``None`` +rather than fabricating one). + +Matching reuses the exact signal ``scan_tree`` (``devklean.scanner.scanner``) +already uses to find cleanable directories: the directory's basename against +a fixed set of known names (``devklean.config.defaults.DEFAULT_TARGETS``). +This module does not walk the filesystem itself. +""" + +from __future__ import annotations + +import os +from dataclasses import dataclass +from enum import Enum + + +class RiskLevel(Enum): + """Fixed risk tier for deleting a matched artifact directory.""" + + LOW = "low" + MEDIUM = "medium" + HIGH = "high" + + @property + def weight(self) -> float: + """Fixed reclaim-safety weight used only by the workspace-health formula. + + Not a probability and not derived from the ``confidence`` score — a + separate, hand-assigned constant so the health formula (see + ``devklean.signatures.health``) stays traceable to this table. + """ + return _RISK_WEIGHTS[self] + + +_RISK_WEIGHTS: dict[RiskLevel, float] = { + RiskLevel.LOW: 1.0, + RiskLevel.MEDIUM: 0.6, + RiskLevel.HIGH: 0.2, +} + + +@dataclass(frozen=True) +class ArtifactSignature: + """A fixed description of one artifact-directory type. + + ``confidence`` is a fixed fraction (0.0-1.0) reflecting how unambiguous + ``dir_name`` is as a signal on its own — e.g. ``__pycache__`` is a + reserved CPython name (very high confidence); ``.cache`` is reused by + many unrelated tools (lower confidence). It is maintainer-assigned, not + computed. + """ + + dir_name: str + ecosystem: str + generated_by: str + regenerate_command: str + risk: RiskLevel + confidence: float + rationale: str + + def matches(self, path: str) -> bool: + """True if ``path``'s basename is this signature's directory name.""" + return os.path.basename(os.path.normpath(path)) == self.dir_name + + +# Seeded 1:1 from devklean.config.defaults.DEFAULT_TARGETS — the directory +# names devklean's scanner already treats as cleanable. A built-in target +# without a matching entry here is not an error: it simply falls into the +# "not recognized" bucket in `analyze`/`explain` output instead of a second, +# divergent list of "things devklean knows about". +SIGNATURE_REGISTRY: dict[str, ArtifactSignature] = { + "node_modules": ArtifactSignature( + dir_name="node_modules", + ecosystem="Node.js (npm/yarn/pnpm)", + generated_by="npm install / yarn install / pnpm install, from package.json and a lockfile", + regenerate_command="npm install (or yarn install / pnpm install, matching your lockfile)", + risk=RiskLevel.LOW, + confidence=0.98, + rationale=( + "Directory name is npm/Node's own convention; contents are fully " + "reproducible from package.json plus a lockfile." + ), + ), + "venv": ArtifactSignature( + dir_name="venv", + ecosystem="Python (venv)", + generated_by="python -m venv, populated via pip from a requirements/lock file", + regenerate_command="python -m venv venv && venv/bin/pip install -r requirements.txt", + risk=RiskLevel.LOW, + confidence=0.95, + rationale=( + "Standard virtualenv layout (pyvenv.cfg, bin/lib); reproducible from a " + "requirements file, though the exact install set depends on which file " + "was originally used." + ), + ), + ".venv": ArtifactSignature( + dir_name=".venv", + ecosystem="Python (venv)", + generated_by="python -m venv, populated via pip from a requirements/lock file", + regenerate_command="python -m venv .venv && .venv/bin/pip install -r requirements.txt", + risk=RiskLevel.LOW, + confidence=0.95, + rationale=( + "Standard virtualenv layout (pyvenv.cfg, bin/lib); reproducible from a " + "requirements file, though the exact install set depends on which file " + "was originally used." + ), + ), + "env": ArtifactSignature( + dir_name="env", + ecosystem="Python (venv, legacy naming)", + generated_by="python -m venv, populated via pip from a requirements/lock file", + regenerate_command="python -m venv env && env/bin/pip install -r requirements.txt", + risk=RiskLevel.MEDIUM, + confidence=0.75, + rationale=( + "'env' is a less specific convention than venv/.venv and is sometimes " + "used for unrelated environment-variable directories, so the match is " + "less certain." + ), + ), + "__pycache__": ArtifactSignature( + dir_name="__pycache__", + ecosystem="Python interpreter (CPython bytecode cache)", + generated_by="CPython automatically, on any import", + regenerate_command=( + "none needed — recreated automatically the next time the module is imported" + ), + risk=RiskLevel.LOW, + confidence=0.99, + rationale=( + "Name is reserved by CPython itself; contents are always .pyc bytecode " + "with zero authored content." + ), + ), + ".next": ArtifactSignature( + dir_name=".next", + ecosystem="Next.js", + generated_by="next build / next dev, from the project's Next.js source", + regenerate_command="next build (or next dev, which regenerates dev artifacts on the fly)", + risk=RiskLevel.LOW, + confidence=0.95, + rationale=( + "Directory name is Next.js's own build-output convention; reproducible " + "from source via the Next.js CLI." + ), + ), + "dist": ArtifactSignature( + dir_name="dist", + ecosystem="Generic build tooling (bundler/compiler output)", + generated_by="whatever build step the project defines (webpack, tsc, vite, build.sh, ...)", + regenerate_command="run the project's documented build command (e.g. npm run build)", + risk=RiskLevel.MEDIUM, + confidence=0.7, + rationale=( + "'dist' is a widely shared convention across many toolchains rather than " + "one tool's signature — usually build output, but devklean cannot prove " + "which build command produced it." + ), + ), + ".cache": ArtifactSignature( + dir_name=".cache", + ecosystem="Generic cache directory", + generated_by=( + "varies by tool — bundlers, test runners, and linters all use .cache differently" + ), + regenerate_command=( + "no single command — recreated automatically the next time the owning tool runs" + ), + risk=RiskLevel.MEDIUM, + confidence=0.65, + rationale=( + "Extremely generic name reused by many unrelated tools; usually safe to " + "remove, but devklean cannot identify which tool owns it or how quickly " + "it regenerates." + ), + ), +} + + +def lookup(name: str) -> ArtifactSignature | None: + """Look up a signature by exact directory name. No fuzzy matching.""" + return SIGNATURE_REGISTRY.get(name) + + +def match_signature(path: str) -> ArtifactSignature | None: + """Resolve a filesystem path to a signature by directory-name match. + + Uses the same name-based signal as ``scan_tree`` (basename membership in + a fixed set) — not a second walker, not a heuristic classifier. Returns + ``None`` on no match; callers must not substitute a guessed verdict. + """ + name = os.path.basename(os.path.normpath(path)) + return SIGNATURE_REGISTRY.get(name) From 1e840c86e170b3cecded552fdea95efe68eaba1c Mon Sep 17 00:00:00 2001 From: smurftyy Date: Thu, 9 Jul 2026 00:03:22 +0100 Subject: [PATCH 3/8] implement 'explain' command --- src/devklean/cli/commands/explain.py | 32 ++++++++++++++++++++++++++++ src/devklean/cli/dispatcher.py | 2 ++ src/devklean/cli/parser.py | 13 +++++++++-- src/devklean/output/text.py | 23 ++++++++++++++++++++ 4 files changed, 68 insertions(+), 2 deletions(-) create mode 100644 src/devklean/cli/commands/explain.py diff --git a/src/devklean/cli/commands/explain.py b/src/devklean/cli/commands/explain.py new file mode 100644 index 0000000..598190e --- /dev/null +++ b/src/devklean/cli/commands/explain.py @@ -0,0 +1,32 @@ +from __future__ import annotations + +import os +from typing import TYPE_CHECKING + +from devklean.signatures import match_signature + +if TYPE_CHECKING: + from devklean.output.text import TextRenderer + + +def run_explain(args, renderer: TextRenderer, config) -> int: + """Resolve a path against the artifact-signature registry and explain it. + + Every field printed on a match comes straight from the matched + ``ArtifactSignature`` — nothing here is generated freeform. On no match, + no risk/confidence verdict is given at all; guessing about an unrecognized + path is a hard constraint, not a style choice. + + Text-only by design, like ``doctor``/``restore``: this is a direct lookup + report, not a scannable result set, so it never needs the JSON scan/history + payload shape. + """ + path = os.path.abspath(args.path) + signature = match_signature(path) + + if signature is None: + renderer.explain_no_match(path) + return 0 + + renderer.explain_match(path, signature) + return 0 diff --git a/src/devklean/cli/dispatcher.py b/src/devklean/cli/dispatcher.py index 3cfef25..5a05ee7 100644 --- a/src/devklean/cli/dispatcher.py +++ b/src/devklean/cli/dispatcher.py @@ -5,6 +5,7 @@ from devklean.cli.commands.clean import run_clean from devklean.cli.commands.doctor import run_doctor +from devklean.cli.commands.explain import run_explain from devklean.cli.commands.history import run_history from devklean.cli.commands.restore import run_restore from devklean.cli.commands.scan import run_scan @@ -23,6 +24,7 @@ "history": run_history, "doctor": run_doctor, "restore": run_restore, + "explain": run_explain, } diff --git a/src/devklean/cli/parser.py b/src/devklean/cli/parser.py index 993eb98..a228042 100644 --- a/src/devklean/cli/parser.py +++ b/src/devklean/cli/parser.py @@ -5,9 +5,9 @@ from devklean._version import __version__ COMMAND_NAMES = frozenset( - {"scan", "clean", "history", "doctor", "stats", "restore", "config", "plugins"} + {"scan", "clean", "history", "doctor", "stats", "restore", "explain", "config", "plugins"} ) -IMPLEMENTED_COMMANDS = frozenset({"scan", "clean", "history", "doctor", "restore"}) +IMPLEMENTED_COMMANDS = frozenset({"scan", "clean", "history", "doctor", "restore", "explain"}) RESERVED_COMMANDS = frozenset({"stats", "config", "plugins"}) GLOBAL_OPTIONS = frozenset({"-h", "--help", "--version"}) @@ -84,6 +84,15 @@ def _add_subparsers(parser: argparse.ArgumentParser) -> None: "restore", help="Show how to recover deleted items from your system trash" ) + explain_parser = subparsers.add_parser( + "explain", + help="Explain what a directory is, using the artifact-signature registry", + ) + explain_parser.add_argument( + "path", + help="Path to a directory to look up in the signature registry", + ) + history_parser = subparsers.add_parser( "history", help="Show previous cleanup operations", diff --git a/src/devklean/output/text.py b/src/devklean/output/text.py index 020f689..c0635c2 100644 --- a/src/devklean/output/text.py +++ b/src/devklean/output/text.py @@ -8,6 +8,7 @@ from devklean.models import CleanableItem, DeleteResult from devklean.output.console import SYM_ERROR, SYM_SUCCESS, Console from devklean.output.sorting import items_by_size_desc +from devklean.signatures import ArtifactSignature class TextRenderer: @@ -169,6 +170,28 @@ def doctor_removed(self, removed: int) -> None: def doctor_remove_error(self, name: str, error: str) -> None: self._console.error(f"could not remove {name}: {error}") + # --- explain --- + + def explain_match(self, path: str, signature: ArtifactSignature) -> None: + c = self._console + self._println() + c.success(path) + self._println(f" {c.paint('ecosystem:', 'detail')} {signature.ecosystem}") + self._println(f" {c.paint('generated by:', 'detail')} {signature.generated_by}") + self._println(f" {c.paint('regenerate:', 'detail')} {signature.regenerate_command}") + self._println(f" {c.paint('risk:', 'detail')} {signature.risk.value}") + self._println(f" {c.paint('confidence:', 'detail')} {signature.confidence:.2f}") + self._println(f" {c.paint('why:', 'detail')} {signature.rationale}") + self._println() + + def explain_no_match(self, path: str) -> None: + c = self._console + self._println() + c.warning(path) + c.detail(" not recognized — no signature in the registry for this directory name.") + c.detail(" no risk or confidence verdict is given for unrecognized paths.") + self._println() + # --- restore --- def restore_help(self) -> None: From ced2065d6d46b35f9794a885beb1dbb9f1431a33 Mon Sep 17 00:00:00 2001 From: smurftyy Date: Thu, 9 Jul 2026 00:10:53 +0100 Subject: [PATCH 4/8] Implement 'analyze' command for directory analysis and reporting --- src/devklean/cli/commands/analyze.py | 35 ++++++++ src/devklean/cli/dispatcher.py | 2 + src/devklean/cli/parser.py | 28 +++++- src/devklean/output/text.py | 62 ++++++++++++- src/devklean/signatures/__init__.py | 30 +++++++ src/devklean/signatures/analysis.py | 99 +++++++++++++++++++++ src/devklean/signatures/health.py | 56 ++++++++++++ src/devklean/signatures/staleness.py | 122 ++++++++++++++++++++++++++ src/devklean/signatures/structural.py | 40 +++++++++ 9 files changed, 471 insertions(+), 3 deletions(-) create mode 100644 src/devklean/cli/commands/analyze.py create mode 100644 src/devklean/signatures/analysis.py create mode 100644 src/devklean/signatures/health.py create mode 100644 src/devklean/signatures/staleness.py create mode 100644 src/devklean/signatures/structural.py diff --git a/src/devklean/cli/commands/analyze.py b/src/devklean/cli/commands/analyze.py new file mode 100644 index 0000000..4de7745 --- /dev/null +++ b/src/devklean/cli/commands/analyze.py @@ -0,0 +1,35 @@ +from __future__ import annotations + +import os +from typing import TYPE_CHECKING + +from devklean.config.models import AppConfig +from devklean.scanner.scanner import scan_tree +from devklean.signatures.analysis import analyze_candidates + +if TYPE_CHECKING: + from devklean.output.text import TextRenderer + + +def run_analyze(args, renderer: TextRenderer, config: AppConfig) -> int: + """Scan for cleanable directories and report a signature-backed analysis. + + Calls ``scan_tree`` directly — the exact discovery function ``clean``/ + ``scan`` use via ``scan_directory`` — rather than a second walker. + ``scan_directory`` isn't reused as-is because its "nothing found" path + renders clean/scan's own summary and stops; analyze always wants to + render its own report (recognized/unrecognized buckets, structural + checks, health score), even over an empty candidate list. + """ + root = os.path.abspath(args.path) + if not os.path.isdir(root): + renderer.invalid_directory(root) + return 1 + + scan_report = scan_tree(root, settings=config.scan_settings) + if scan_report.permission_errors: + renderer.permission_warnings(scan_report.permission_errors) + + analysis = analyze_candidates(root, scan_report.items) + renderer.analysis_report(analysis, verbose=getattr(args, "verbose", False)) + return 0 diff --git a/src/devklean/cli/dispatcher.py b/src/devklean/cli/dispatcher.py index 5a05ee7..d4ef990 100644 --- a/src/devklean/cli/dispatcher.py +++ b/src/devklean/cli/dispatcher.py @@ -3,6 +3,7 @@ import sys from typing import Callable +from devklean.cli.commands.analyze import run_analyze from devklean.cli.commands.clean import run_clean from devklean.cli.commands.doctor import run_doctor from devklean.cli.commands.explain import run_explain @@ -25,6 +26,7 @@ "doctor": run_doctor, "restore": run_restore, "explain": run_explain, + "analyze": run_analyze, } diff --git a/src/devklean/cli/parser.py b/src/devklean/cli/parser.py index a228042..c282fd8 100644 --- a/src/devklean/cli/parser.py +++ b/src/devklean/cli/parser.py @@ -5,9 +5,22 @@ from devklean._version import __version__ COMMAND_NAMES = frozenset( - {"scan", "clean", "history", "doctor", "stats", "restore", "explain", "config", "plugins"} + { + "scan", + "clean", + "history", + "doctor", + "stats", + "restore", + "explain", + "analyze", + "config", + "plugins", + } +) +IMPLEMENTED_COMMANDS = frozenset( + {"scan", "clean", "history", "doctor", "restore", "explain", "analyze"} ) -IMPLEMENTED_COMMANDS = frozenset({"scan", "clean", "history", "doctor", "restore", "explain"}) RESERVED_COMMANDS = frozenset({"stats", "config", "plugins"}) GLOBAL_OPTIONS = frozenset({"-h", "--help", "--version"}) @@ -93,6 +106,17 @@ def _add_subparsers(parser: argparse.ArgumentParser) -> None: help="Path to a directory to look up in the signature registry", ) + analyze_parser = subparsers.add_parser( + "analyze", + help="Analyze cleanable directories using the artifact-signature registry", + ) + _add_path_argument(analyze_parser) + analyze_parser.add_argument( + "--verbose", + action="store_true", + help="Show the workspace-health formula and its raw inputs", + ) + history_parser = subparsers.add_parser( "history", help="Show previous cleanup operations", diff --git a/src/devklean/output/text.py b/src/devklean/output/text.py index c0635c2..5d8c40b 100644 --- a/src/devklean/output/text.py +++ b/src/devklean/output/text.py @@ -4,11 +4,12 @@ from devklean.deletion.history import HistoryOperation from devklean.deletion.integrity import IntegrityReport -from devklean.formatting import format_size, format_timestamp +from devklean.formatting import format_size, format_timestamp, truncate from devklean.models import CleanableItem, DeleteResult from devklean.output.console import SYM_ERROR, SYM_SUCCESS, Console from devklean.output.sorting import items_by_size_desc from devklean.signatures import ArtifactSignature +from devklean.signatures.analysis import AnalysisReport class TextRenderer: @@ -192,6 +193,65 @@ def explain_no_match(self, path: str) -> None: c.detail(" no risk or confidence verdict is given for unrecognized paths.") self._println() + # --- analyze --- + + def analysis_report(self, report: AnalysisReport, *, verbose: bool = False) -> None: + c = self._console + self._println() + self._println(f"{c.paint('devklean analyze', 'bold')} {c.paint(report.root, 'detail')}") + self._println() + + self._println(c.paint("RECOGNIZED — safe to remove (per signature registry)", "bold")) + if not report.recognized: + c.detail(" none") + else: + self._println(f" {'TYPE':<18} {'SIZE':>10} {'RISK':<8} {'ECOSYSTEM':<28} STALENESS") + self._println(c.paint(" " + "─" * 100, "detail")) + ordered = sorted(report.recognized, key=lambda rc: rc.item.size, reverse=True) + for rc in ordered: + self._println( + f" {c.paint(f'{rc.item.display_label:<18}', 'detail')} " + f"{format_size(rc.item.size):>10} " + f"{rc.signature.risk.value:<8} " + f"{truncate(rc.signature.ecosystem, 28):<28} " + f"{rc.staleness.detail}" + ) + self._println() + + self._println(c.paint("NOT RECOGNIZED — skipped (no signature registry entry)", "bold")) + if not report.unrecognized: + c.detail(" none") + else: + for item in report.unrecognized: + c.detail(f" {item.display_label:<18} {item.path}") + self._println() + + self._println(c.paint("STRUCTURAL", "bold")) + if not report.lockfile_conflicts: + c.detail(" no structural issues found") + else: + for conflict in report.lockfile_conflicts: + c.warning( + f" conflicting lockfiles in {conflict.project_root}: " + f"{', '.join(conflict.lockfiles)}" + ) + self._println() + + score = report.health.score + role = "success" if score >= 70 else "warning" if score >= 40 else "error" + self._println(c.paint("WORKSPACE HEALTH: ", "bold") + c.paint(f"{score}/100", role)) + if verbose: + inputs = report.health.inputs + c.detail(f" formula: {report.health.formula}") + c.detail( + " inputs: " + f"recognized_weighted_size={inputs.recognized_weighted_size:.0f} " + f"recognized_total_size={inputs.recognized_total_size} " + f"unrecognized_total_size={inputs.unrecognized_total_size} " + f"lockfile_conflicts={inputs.lockfile_conflicts}" + ) + self._println() + # --- restore --- def restore_help(self) -> None: diff --git a/src/devklean/signatures/__init__.py b/src/devklean/signatures/__init__.py index e70c2b4..6c292ff 100644 --- a/src/devklean/signatures/__init__.py +++ b/src/devklean/signatures/__init__.py @@ -8,6 +8,16 @@ "SIGNATURE_REGISTRY", "lookup", "match_signature", + "AnalysisReport", + "RecognizedCandidate", + "analyze_candidates", + "StalenessResult", + "estimate_staleness", + "LockfileConflict", + "detect_lockfile_conflicts", + "HealthInputs", + "HealthScore", + "compute_health_score", ] @@ -22,4 +32,24 @@ def __getattr__(name: str): return import_module("devklean.signatures.registry").lookup if name == "match_signature": return import_module("devklean.signatures.registry").match_signature + if name == "AnalysisReport": + return import_module("devklean.signatures.analysis").AnalysisReport + if name == "RecognizedCandidate": + return import_module("devklean.signatures.analysis").RecognizedCandidate + if name == "analyze_candidates": + return import_module("devklean.signatures.analysis").analyze_candidates + if name == "StalenessResult": + return import_module("devklean.signatures.staleness").StalenessResult + if name == "estimate_staleness": + return import_module("devklean.signatures.staleness").estimate_staleness + if name == "LockfileConflict": + return import_module("devklean.signatures.structural").LockfileConflict + if name == "detect_lockfile_conflicts": + return import_module("devklean.signatures.structural").detect_lockfile_conflicts + if name == "HealthInputs": + return import_module("devklean.signatures.health").HealthInputs + if name == "HealthScore": + return import_module("devklean.signatures.health").HealthScore + if name == "compute_health_score": + return import_module("devklean.signatures.health").compute_health_score raise AttributeError(f"module {__name__!r} has no attribute {name!r}") diff --git a/src/devklean/signatures/analysis.py b/src/devklean/signatures/analysis.py new file mode 100644 index 0000000..2da71e2 --- /dev/null +++ b/src/devklean/signatures/analysis.py @@ -0,0 +1,99 @@ +"""Bucket already-discovered candidates via the signature registry and score them. + +This module never walks the filesystem to find candidates — it only accepts +the ``CleanableItem`` list ``scan_tree`` already produced (the same discovery +``clean``/``scan`` use) and layers registry lookups, staleness estimates, and +structural checks on top of it. +""" + +from __future__ import annotations + +import os +from dataclasses import dataclass + +from devklean.models import CleanableItem +from devklean.signatures.health import HealthInputs, HealthScore, compute_health_score +from devklean.signatures.registry import ArtifactSignature, lookup +from devklean.signatures.staleness import StalenessResult, estimate_staleness +from devklean.signatures.structural import LockfileConflict, detect_lockfile_conflicts + + +@dataclass(frozen=True) +class RecognizedCandidate: + """A scanned candidate with a matched signature and staleness estimate.""" + + item: CleanableItem + signature: ArtifactSignature + staleness: StalenessResult + + +@dataclass(frozen=True) +class AnalysisReport: + root: str + recognized: tuple[RecognizedCandidate, ...] + unrecognized: tuple[CleanableItem, ...] + lockfile_conflicts: tuple[LockfileConflict, ...] + health: HealthScore + + +def analyze_candidates(root: str, items: list[CleanableItem]) -> AnalysisReport: + """Bucket ``items`` into recognized/unrecognized and attach analysis. + + "Recognized" means the candidate's directory name has a + ``SIGNATURE_REGISTRY`` entry; everything else — e.g. a custom target + added via config with no registry entry — is "unrecognized" and gets no + risk/confidence/staleness verdict, per the same no-guessing rule + ``explain`` follows. + """ + abs_root = os.path.abspath(root) + recognized: list[RecognizedCandidate] = [] + unrecognized: list[CleanableItem] = [] + staleness_cache: dict[str, StalenessResult] = {} + project_roots: set[str] = {abs_root} + + for item in items: + project_roots.add(os.path.dirname(item.path)) + + signature = lookup(item.name) + if signature is None: + unrecognized.append(item) + continue + + project_root = os.path.dirname(item.path) + if project_root not in staleness_cache: + staleness_cache[project_root] = estimate_staleness(project_root) + + recognized.append( + RecognizedCandidate( + item=item, + signature=signature, + staleness=staleness_cache[project_root], + ) + ) + + conflicts = tuple( + conflict + for conflict in (detect_lockfile_conflicts(p) for p in sorted(project_roots)) + if conflict is not None + ) + + recognized_total = sum(c.item.size for c in recognized) + unrecognized_total = sum(i.size for i in unrecognized) + weighted = sum(c.item.size * c.signature.risk.weight for c in recognized) + + health = compute_health_score( + HealthInputs( + recognized_weighted_size=weighted, + recognized_total_size=recognized_total, + unrecognized_total_size=unrecognized_total, + lockfile_conflicts=len(conflicts), + ) + ) + + return AnalysisReport( + root=abs_root, + recognized=tuple(recognized), + unrecognized=tuple(unrecognized), + lockfile_conflicts=conflicts, + health=health, + ) diff --git a/src/devklean/signatures/health.py b/src/devklean/signatures/health.py new file mode 100644 index 0000000..6999ac7 --- /dev/null +++ b/src/devklean/signatures/health.py @@ -0,0 +1,56 @@ +"""Workspace-health scoring: one documented formula, fixed inputs. + +The score is a static function of counted bytes and structural-conflict +counts — never a model, never a hidden heuristic. ``HealthScore.formula`` and +``HealthScore.inputs`` are always populated (not just under a debug flag) so +the number is inspectable on demand, e.g. via ``devklean analyze --verbose``. +""" + +from __future__ import annotations + +from dataclasses import dataclass + +# Flat penalty subtracted per detected lockfile conflict (see +# devklean.signatures.structural). A fixed constant, not a computed weight. +LOCKFILE_CONFLICT_PENALTY = 10 + +_FORMULA = ( + "score = round(100 * (sum(recognized_size_i * risk_weight_i) / total_size)) " + f"- {LOCKFILE_CONFLICT_PENALTY} * lockfile_conflicts, clamped to [0, 100]. " + "risk_weight: low=1.0, medium=0.6, high=0.2 (devklean.signatures.registry.RiskLevel.weight). " + "unrecognized_total_size counts toward total_size but contributes 0 to the " + "weighted sum, since devklean has no risk data for it. " + "total_size = recognized_total_size + unrecognized_total_size; " + "an empty scan (total_size == 0) scores 100." +) + + +@dataclass(frozen=True) +class HealthInputs: + """Every value the formula reads — nothing else feeds the score.""" + + recognized_weighted_size: float + recognized_total_size: int + unrecognized_total_size: int + lockfile_conflicts: int + + @property + def total_size(self) -> int: + return self.recognized_total_size + self.unrecognized_total_size + + +@dataclass(frozen=True) +class HealthScore: + score: int + inputs: HealthInputs + formula: str = _FORMULA + + +def compute_health_score(inputs: HealthInputs) -> HealthScore: + if inputs.total_size == 0: + return HealthScore(score=100, inputs=inputs) + + weighted_share = inputs.recognized_weighted_size / inputs.total_size + raw = round(100 * weighted_share) - LOCKFILE_CONFLICT_PENALTY * inputs.lockfile_conflicts + score = max(0, min(100, raw)) + return HealthScore(score=score, inputs=inputs) diff --git a/src/devklean/signatures/staleness.py b/src/devklean/signatures/staleness.py new file mode 100644 index 0000000..e66c213 --- /dev/null +++ b/src/devklean/signatures/staleness.py @@ -0,0 +1,122 @@ +"""Staleness estimation for a candidate's parent project. + +The artifact directory's own mtime/atime is never used as a signal: package +managers rewrite files inside it on every install, and atime is frequently +disabled at the mount level (``noatime``/``relatime``), so neither reflects +when the *project* was actually last worked on. Instead this derives +staleness from the parent project itself, in a fixed priority order: + +1. The git repository's last commit date, if the project is inside one. +2. The newest mtime among the parent project's own files, excluding known + artifact directories (so a fresh ``npm install`` doesn't look like recent + activity). + +If neither produces a usable timestamp, ``estimate_staleness`` returns a +result with ``known=False`` — callers must surface that explicitly rather +than print a number that looks confident but means nothing. +""" + +from __future__ import annotations + +import os +import subprocess +from dataclasses import dataclass +from datetime import datetime, timezone + +from devklean.signatures.registry import SIGNATURE_REGISTRY + +# Directory names never treated as "source" when computing the fallback +# mtime signal — the same set of artifact names the registry knows about, +# plus .git (its internal mtimes aren't project activity either). +_EXCLUDED_DIR_NAMES = frozenset(SIGNATURE_REGISTRY) | {".git"} + + +@dataclass(frozen=True) +class StalenessResult: + """A staleness estimate for one project root. + + ``known=False`` means no reliable signal existed; ``detail`` always holds + a human-readable explanation regardless of ``known``, so it's always safe + to display. + """ + + known: bool + source: str | None # "git" | "source-mtime" | None + last_activity: datetime | None + days_since: int | None + detail: str + + +def estimate_staleness(project_root: str) -> StalenessResult: + """Resolve a staleness signal for ``project_root``, git first, then mtime.""" + return _from_git(project_root) or _from_source_mtime(project_root) or StalenessResult( + known=False, + source=None, + last_activity=None, + days_since=None, + detail="no reliable signal (no git repository and no source files found)", + ) + + +def _days_since(moment: datetime) -> int: + return (datetime.now(tz=timezone.utc) - moment).days + + +def _from_git(project_root: str) -> StalenessResult | None: + try: + proc = subprocess.run( + ["git", "-C", project_root, "log", "-1", "--format=%ct"], + capture_output=True, + text=True, + timeout=5, + ) + except (OSError, subprocess.TimeoutExpired): + return None + + if proc.returncode != 0: + return None + + output = proc.stdout.strip() + if not output: + return None + + try: + epoch = int(output) + except ValueError: + return None + + last_activity = datetime.fromtimestamp(epoch, tz=timezone.utc) + days = _days_since(last_activity) + return StalenessResult( + known=True, + source="git", + last_activity=last_activity, + days_since=days, + detail=f"{days} day{'s' if days != 1 else ''} since last git commit", + ) + + +def _from_source_mtime(project_root: str) -> StalenessResult | None: + latest: float | None = None + for dirpath, dirnames, filenames in os.walk(project_root): + dirnames[:] = [d for d in dirnames if d not in _EXCLUDED_DIR_NAMES] + for filename in filenames: + try: + mtime = os.stat(os.path.join(dirpath, filename)).st_mtime + except OSError: + continue + if latest is None or mtime > latest: + latest = mtime + + if latest is None: + return None + + last_activity = datetime.fromtimestamp(latest, tz=timezone.utc) + days = _days_since(last_activity) + return StalenessResult( + known=True, + source="source-mtime", + last_activity=last_activity, + days_since=days, + detail=f"{days} day{'s' if days != 1 else ''} since newest source-file change", + ) diff --git a/src/devklean/signatures/structural.py b/src/devklean/signatures/structural.py new file mode 100644 index 0000000..0c63e27 --- /dev/null +++ b/src/devklean/signatures/structural.py @@ -0,0 +1,40 @@ +"""Structural checks: deterministic file-existence checks, not judgment calls. + +Each check here only asks "do these specific files exist in this directory?" +No sizes, no content inspection, no guessing about what a directory is for. +""" + +from __future__ import annotations + +import os +from dataclasses import dataclass + +# Groups of lockfiles that indicate two different, mutually exclusive package +# managers were both used against the same project root. Presence of 2+ names +# from the same group is flagged, regardless of which ones. +_CONFLICTING_LOCKFILE_GROUPS: tuple[tuple[str, ...], ...] = ( + ("package-lock.json", "pnpm-lock.yaml", "yarn.lock", "bun.lockb"), +) + + +@dataclass(frozen=True) +class LockfileConflict: + """Two or more mutually exclusive lockfiles found in one project root.""" + + project_root: str + lockfiles: tuple[str, ...] + + +def detect_lockfile_conflicts(project_root: str) -> LockfileConflict | None: + """Flag a project root that has 2+ lockfiles from the same conflict group. + + Plain ``os.path.isfile`` checks — no inference about which lockfile is + "correct" or which package manager is actually in use. + """ + for group in _CONFLICTING_LOCKFILE_GROUPS: + present = tuple( + name for name in group if os.path.isfile(os.path.join(project_root, name)) + ) + if len(present) >= 2: + return LockfileConflict(project_root=project_root, lockfiles=present) + return None From 2835f9f9be23fec725cb095fc6c52e2efd9ac0fa Mon Sep 17 00:00:00 2001 From: smurftyy Date: Thu, 9 Jul 2026 00:18:59 +0100 Subject: [PATCH 5/8] tests: Comprehensive tests for 'analyze','explain', signatures health,registry, staleness, and structural checks --- tests/test_analyze_command.py | 146 ++++++++++++++++++++++++++++ tests/test_explain_command.py | 92 ++++++++++++++++++ tests/test_signatures_health.py | 89 +++++++++++++++++ tests/test_signatures_registry.py | 86 ++++++++++++++++ tests/test_signatures_staleness.py | 102 +++++++++++++++++++ tests/test_signatures_structural.py | 39 ++++++++ 6 files changed, 554 insertions(+) create mode 100644 tests/test_analyze_command.py create mode 100644 tests/test_explain_command.py create mode 100644 tests/test_signatures_health.py create mode 100644 tests/test_signatures_registry.py create mode 100644 tests/test_signatures_staleness.py create mode 100644 tests/test_signatures_structural.py diff --git a/tests/test_analyze_command.py b/tests/test_analyze_command.py new file mode 100644 index 0000000..16a42b1 --- /dev/null +++ b/tests/test_analyze_command.py @@ -0,0 +1,146 @@ +"""Tests for `devklean analyze`.""" + +from __future__ import annotations + +import subprocess +import sys +from argparse import Namespace +from io import StringIO +from pathlib import Path + +from devklean.cli.commands.analyze import run_analyze +from devklean.config.defaults import DEFAULT_TARGETS +from devklean.config.models import AppConfig +from devklean.output.console import Console +from devklean.output.text import TextRenderer + + +def _config() -> AppConfig: + return AppConfig(targets=dict(DEFAULT_TARGETS)) + + +def _renderer() -> tuple[TextRenderer, StringIO]: + stream = StringIO() + return TextRenderer(console=Console(stream=stream)), stream + + +def _args(path: str, *, verbose: bool = False) -> Namespace: + return Namespace(path=path, verbose=verbose) + + +def test_analyze_buckets_recognized_and_unrecognized(tmp_path: Path) -> None: + project = tmp_path / "project" + node_modules = project / "node_modules" + node_modules.mkdir(parents=True) + (node_modules / "pkg.json").write_text("{}") + + # A custom target with no signature-registry entry: recognized by + # scan_tree (it's a configured target) but must land in "unrecognized" + # here, since devklean.signatures has no data for it. + config = AppConfig(targets={**DEFAULT_TARGETS, "turbo-cache": "Turborepo cache"}) + custom = project / "turbo-cache" + custom.mkdir() + (custom / "x").write_text("x") + + renderer, stream = _renderer() + + code = run_analyze(_args(str(project)), renderer, config) + + out = stream.getvalue() + assert code == 0 + assert "RECOGNIZED" in out + assert "Node.js" in out + assert "NOT RECOGNIZED" in out + assert str(custom) in out + + +def test_analyze_reports_no_candidates_cleanly(tmp_path: Path) -> None: + empty = tmp_path / "empty" + empty.mkdir() + renderer, stream = _renderer() + + code = run_analyze(_args(str(empty)), renderer, _config()) + + out = stream.getvalue() + assert code == 0 + assert "none" in out + assert "WORKSPACE HEALTH: 100/100" in out + + +def test_analyze_invalid_directory_returns_error(tmp_path: Path) -> None: + renderer, stream = _renderer() + + code = run_analyze(_args(str(tmp_path / "does-not-exist")), renderer, _config()) + + assert code == 1 + assert "not a directory" in stream.getvalue() + + +def test_analyze_flags_lockfile_conflict(tmp_path: Path) -> None: + project = tmp_path / "project" + project.mkdir() + (project / "package-lock.json").write_text("{}") + (project / "pnpm-lock.yaml").write_text("") + node_modules = project / "node_modules" + node_modules.mkdir() + (node_modules / "pkg.json").write_text("{}") + + renderer, stream = _renderer() + + code = run_analyze(_args(str(project)), renderer, _config()) + + out = stream.getvalue() + assert code == 0 + assert "conflicting lockfiles" in out + assert "package-lock.json" in out and "pnpm-lock.yaml" in out + + +def test_analyze_verbose_shows_formula_and_inputs(tmp_path: Path) -> None: + project = tmp_path / "project" + node_modules = project / "node_modules" + node_modules.mkdir(parents=True) + (node_modules / "pkg.json").write_text("{}") + + renderer, stream = _renderer() + + code = run_analyze(_args(str(project), verbose=True), renderer, _config()) + + out = stream.getvalue() + assert code == 0 + assert "formula:" in out + assert "inputs:" in out + assert "recognized_weighted_size=" in out + + +def test_analyze_reports_unknown_staleness_when_no_signal(tmp_path: Path) -> None: + """Integration check for the staleness fallback: no git repo, and the + only file present lives inside the artifact directory itself.""" + project = tmp_path / "project" + node_modules = project / "node_modules" + node_modules.mkdir(parents=True) + (node_modules / "pkg.json").write_text("{}") + + renderer, stream = _renderer() + + code = run_analyze(_args(str(project)), renderer, _config()) + + out = stream.getvalue() + assert code == 0 + assert "no reliable signal" in out + + +def test_analyze_command_end_to_end(tmp_path: Path) -> None: + project = tmp_path / "project" + node_modules = project / "node_modules" + node_modules.mkdir(parents=True) + (node_modules / "pkg.json").write_text("{}") + + result = subprocess.run( + [sys.executable, "-m", "devklean", "analyze", str(project)], + capture_output=True, + text=True, + check=False, + ) + + assert result.returncode == 0 + assert "WORKSPACE HEALTH" in result.stdout diff --git a/tests/test_explain_command.py b/tests/test_explain_command.py new file mode 100644 index 0000000..c909871 --- /dev/null +++ b/tests/test_explain_command.py @@ -0,0 +1,92 @@ +"""Tests for `devklean explain`.""" + +from __future__ import annotations + +import re +import subprocess +import sys +from argparse import Namespace +from pathlib import Path + +import pytest + +from devklean.cli.commands.explain import run_explain +from devklean.output.text import TextRenderer +from devklean.signatures import SIGNATURE_REGISTRY, match_signature + + +def _args(path: str) -> Namespace: + return Namespace(path=path) + + +@pytest.mark.parametrize("dir_name", sorted(SIGNATURE_REGISTRY)) +def test_explain_golden_output_per_signature(tmp_path: Path, capsys, dir_name: str) -> None: + """Golden-output test: every field printed is the matched signature's own, + verbatim — nothing generated freeform.""" + candidate = tmp_path / dir_name + candidate.mkdir() + signature = SIGNATURE_REGISTRY[dir_name] + + code = run_explain(_args(str(candidate)), TextRenderer(), None) + + out = capsys.readouterr().out + assert code == 0 + assert signature.ecosystem in out + assert signature.generated_by in out + assert signature.regenerate_command in out + assert signature.risk.value in out + assert f"{signature.confidence:.2f}" in out + assert signature.rationale in out + + +def test_explain_unrecognized_path_gives_no_verdict(tmp_path: Path, capsys) -> None: + mystery = tmp_path / "mystery_dir" + mystery.mkdir() + + code = run_explain(_args(str(mystery)), TextRenderer(), None) + + out = capsys.readouterr().out + assert code == 0 + assert "not recognized" in out.lower() + + # Hard constraint: an unrecognized path must never receive a fabricated + # risk or confidence value. Check for the risk-level words and a + # confidence-shaped number as whole tokens, since tmp_path itself + # legitimately contains digits (e.g. pytest's tmp dir numbering). + assert re.search(r"\b(low|medium|high)\b", out, re.IGNORECASE) is None + assert re.search(r"\b0\.\d{2}\b", out) is None + + +def test_explain_unrecognized_path_matches_no_registry_entry(tmp_path: Path) -> None: + mystery = tmp_path / "mystery_dir" + mystery.mkdir() + + assert match_signature(str(mystery)) is None + + +def test_explain_command_end_to_end(tmp_path: Path) -> None: + node_modules = tmp_path / "node_modules" + node_modules.mkdir() + + result = subprocess.run( + [sys.executable, "-m", "devklean", "explain", str(node_modules)], + capture_output=True, + text=True, + check=False, + ) + + assert result.returncode == 0 + assert "Node.js" in result.stdout + assert "risk:" in result.stdout + assert "confidence:" in result.stdout + + +def test_explain_requires_path_argument() -> None: + result = subprocess.run( + [sys.executable, "-m", "devklean", "explain"], + capture_output=True, + text=True, + check=False, + ) + + assert result.returncode != 0 diff --git a/tests/test_signatures_health.py b/tests/test_signatures_health.py new file mode 100644 index 0000000..fc41097 --- /dev/null +++ b/tests/test_signatures_health.py @@ -0,0 +1,89 @@ +"""Tests for the documented workspace-health formula (devklean.signatures.health).""" + +from __future__ import annotations + +from devklean.signatures.health import HealthInputs, compute_health_score + + +def test_empty_scan_scores_100() -> None: + inputs = HealthInputs( + recognized_weighted_size=0, + recognized_total_size=0, + unrecognized_total_size=0, + lockfile_conflicts=0, + ) + + result = compute_health_score(inputs) + + assert result.score == 100 + assert result.formula + + +def test_all_low_risk_recognized_bytes_scores_100() -> None: + inputs = HealthInputs( + recognized_weighted_size=1000 * 1.0, + recognized_total_size=1000, + unrecognized_total_size=0, + lockfile_conflicts=0, + ) + + assert compute_health_score(inputs).score == 100 + + +def test_all_high_risk_recognized_bytes_scores_low() -> None: + inputs = HealthInputs( + recognized_weighted_size=1000 * 0.2, + recognized_total_size=1000, + unrecognized_total_size=0, + lockfile_conflicts=0, + ) + + assert compute_health_score(inputs).score == 20 + + +def test_unrecognized_bytes_drag_the_score_down() -> None: + # 500 low-risk recognized bytes, 500 unrecognized (zero-weight) bytes. + inputs = HealthInputs( + recognized_weighted_size=500 * 1.0, + recognized_total_size=500, + unrecognized_total_size=500, + lockfile_conflicts=0, + ) + + assert compute_health_score(inputs).score == 50 + + +def test_lockfile_conflict_applies_flat_penalty() -> None: + inputs = HealthInputs( + recognized_weighted_size=1000 * 1.0, + recognized_total_size=1000, + unrecognized_total_size=0, + lockfile_conflicts=1, + ) + + assert compute_health_score(inputs).score == 90 + + +def test_score_is_clamped_to_zero() -> None: + inputs = HealthInputs( + recognized_weighted_size=0, + recognized_total_size=100, + unrecognized_total_size=0, + lockfile_conflicts=5, # 5 * 10 = 50 points, more than the 0 base score + ) + + assert compute_health_score(inputs).score == 0 + + +def test_inputs_are_exposed_on_the_result() -> None: + inputs = HealthInputs( + recognized_weighted_size=10, + recognized_total_size=20, + unrecognized_total_size=5, + lockfile_conflicts=2, + ) + + result = compute_health_score(inputs) + + assert result.inputs is inputs + assert result.inputs.total_size == 25 diff --git a/tests/test_signatures_registry.py b/tests/test_signatures_registry.py new file mode 100644 index 0000000..92da523 --- /dev/null +++ b/tests/test_signatures_registry.py @@ -0,0 +1,86 @@ +"""Golden-output tests for the artifact-signature registry (devklean.signatures).""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from devklean.config.defaults import DEFAULT_TARGETS +from devklean.signatures import SIGNATURE_REGISTRY, RiskLevel, lookup, match_signature + + +def test_registry_covers_every_default_target() -> None: + """Seeded 1:1 from DEFAULT_TARGETS — devklean's actual known artifact list.""" + assert set(SIGNATURE_REGISTRY) == set(DEFAULT_TARGETS) + + +@pytest.mark.parametrize("dir_name", sorted(SIGNATURE_REGISTRY)) +def test_every_entry_has_fixed_fields(dir_name: str) -> None: + signature = SIGNATURE_REGISTRY[dir_name] + assert signature.dir_name == dir_name + assert isinstance(signature.risk, RiskLevel) + assert 0.0 <= signature.confidence <= 1.0 + assert signature.ecosystem + assert signature.generated_by + assert signature.regenerate_command + assert signature.rationale + + +# Golden values pin the actual maintainer-assigned data so a change to any of +# them is a deliberate, reviewed edit rather than an accidental drift. +_GOLDEN = { + "node_modules": ("Node.js (npm/yarn/pnpm)", RiskLevel.LOW, 0.98), + "venv": ("Python (venv)", RiskLevel.LOW, 0.95), + ".venv": ("Python (venv)", RiskLevel.LOW, 0.95), + "env": ("Python (venv, legacy naming)", RiskLevel.MEDIUM, 0.75), + "__pycache__": ("Python interpreter (CPython bytecode cache)", RiskLevel.LOW, 0.99), + ".next": ("Next.js", RiskLevel.LOW, 0.95), + "dist": ("Generic build tooling (bundler/compiler output)", RiskLevel.MEDIUM, 0.7), + ".cache": ("Generic cache directory", RiskLevel.MEDIUM, 0.65), +} + + +def test_golden_table_covers_the_whole_registry() -> None: + assert set(_GOLDEN) == set(SIGNATURE_REGISTRY) + + +@pytest.mark.parametrize("dir_name", sorted(_GOLDEN)) +def test_golden_signature_values(dir_name: str) -> None: + ecosystem, risk, confidence = _GOLDEN[dir_name] + signature = lookup(dir_name) + assert signature is not None + assert signature.ecosystem == ecosystem + assert signature.risk == risk + assert signature.confidence == confidence + + +@pytest.mark.parametrize("dir_name", sorted(SIGNATURE_REGISTRY)) +def test_match_signature_by_directory_fixture(tmp_path: Path, dir_name: str) -> None: + """Golden-output test per signature: a real fixture directory resolves + to exactly that signature and nothing else.""" + candidate = tmp_path / dir_name + candidate.mkdir() + + matched = match_signature(str(candidate)) + + assert matched is SIGNATURE_REGISTRY[dir_name] + + +def test_match_signature_returns_none_for_unrecognized_directory(tmp_path: Path) -> None: + candidate = tmp_path / "totally_unknown_dir" + candidate.mkdir() + + assert match_signature(str(candidate)) is None + + +def test_lookup_returns_none_for_unknown_name() -> None: + assert lookup("totally_unknown_dir") is None + + +def test_signature_matches_uses_basename_not_full_path(tmp_path: Path) -> None: + signature = lookup("node_modules") + assert signature is not None + nested = tmp_path / "a" / "b" / "node_modules" + assert signature.matches(str(nested)) + assert not signature.matches(str(tmp_path / "a" / "node_modules" / "b")) diff --git a/tests/test_signatures_staleness.py b/tests/test_signatures_staleness.py new file mode 100644 index 0000000..994c05f --- /dev/null +++ b/tests/test_signatures_staleness.py @@ -0,0 +1,102 @@ +"""Tests for devklean.signatures.staleness. + +Staleness must never come from the artifact directory's own mtime/atime — it +is always derived from the parent project (git last-commit date, or the +newest mtime among the project's own non-artifact files), and must say so +explicitly when neither signal is available. +""" + +from __future__ import annotations + +import os +import subprocess +import time +from pathlib import Path + +from devklean.signatures.staleness import estimate_staleness + + +def _git(*args: str, cwd: Path) -> None: + subprocess.run(["git", *args], cwd=cwd, check=True, capture_output=True) + + +def test_staleness_uses_git_last_commit_when_available(tmp_path: Path) -> None: + project = tmp_path / "project" + project.mkdir() + _git("init", "-q", cwd=project) + _git("config", "user.email", "a@example.com", cwd=project) + _git("config", "user.name", "a", cwd=project) + (project / "main.py").write_text("print('hi')") + _git("add", "-A", cwd=project) + _git("commit", "-q", "-m", "init", cwd=project) + + result = estimate_staleness(str(project)) + + assert result.known is True + assert result.source == "git" + assert result.days_since is not None + assert result.days_since >= 0 + assert "git commit" in result.detail + + +def test_staleness_falls_back_to_source_mtime_without_git(tmp_path: Path) -> None: + project = tmp_path / "project" + project.mkdir() + (project / "main.py").write_text("print('hi')") + node_modules = project / "node_modules" + node_modules.mkdir() + # Touch the artifact dir's own file *after* the source file, to prove the + # signal comes from source files, not the artifact directory. + time.sleep(0.01) + (node_modules / "pkg.json").write_text("{}") + + result = estimate_staleness(str(project)) + + assert result.known is True + assert result.source == "source-mtime" + assert result.days_since is not None + assert "source-file change" in result.detail + + +def test_staleness_ignores_known_artifact_directories_for_mtime(tmp_path: Path) -> None: + """The artifact directory's own files must never set the mtime signal.""" + project = tmp_path / "project" + project.mkdir() + old_source = project / "main.py" + old_source.write_text("print('hi')") + old_time = time.time() - 1_000_000 + os.utime(old_source, (old_time, old_time)) + + node_modules = project / "node_modules" + node_modules.mkdir() + (node_modules / "pkg.json").write_text("{}") # freshly written, i.e. "now" + + result = estimate_staleness(str(project)) + + assert result.known is True + assert result.source == "source-mtime" + # If node_modules' fresh file had counted, days_since would be ~0. + assert result.days_since is not None + assert result.days_since > 10 + + +def test_staleness_reports_no_reliable_signal_without_git_or_source_files( + tmp_path: Path, +) -> None: + """No git repo, and the only file present lives inside a known artifact + directory — so there is nothing left to derive a signal from.""" + project = tmp_path / "project" + project.mkdir() + node_modules = project / "node_modules" + node_modules.mkdir() + (node_modules / "pkg.json").write_text("{}") + + result = estimate_staleness(str(project)) + + assert result.known is False + assert result.source is None + assert result.last_activity is None + assert result.days_since is None + assert "no reliable signal" in result.detail + assert "no git repository" in result.detail + assert "no source files found" in result.detail diff --git a/tests/test_signatures_structural.py b/tests/test_signatures_structural.py new file mode 100644 index 0000000..9d199fb --- /dev/null +++ b/tests/test_signatures_structural.py @@ -0,0 +1,39 @@ +"""Tests for devklean.signatures.structural (plain file-existence checks).""" + +from __future__ import annotations + +from pathlib import Path + +from devklean.signatures.structural import detect_lockfile_conflicts + + +def test_no_conflict_with_a_single_lockfile(tmp_path: Path) -> None: + (tmp_path / "package-lock.json").write_text("{}") + + assert detect_lockfile_conflicts(str(tmp_path)) is None + + +def test_no_conflict_with_no_lockfiles(tmp_path: Path) -> None: + assert detect_lockfile_conflicts(str(tmp_path)) is None + + +def test_flags_npm_and_pnpm_lockfile_conflict(tmp_path: Path) -> None: + (tmp_path / "package-lock.json").write_text("{}") + (tmp_path / "pnpm-lock.yaml").write_text("lockfileVersion: 6") + + conflict = detect_lockfile_conflicts(str(tmp_path)) + + assert conflict is not None + assert conflict.project_root == str(tmp_path) + assert set(conflict.lockfiles) == {"package-lock.json", "pnpm-lock.yaml"} + + +def test_flags_three_way_lockfile_conflict(tmp_path: Path) -> None: + (tmp_path / "package-lock.json").write_text("{}") + (tmp_path / "yarn.lock").write_text("") + (tmp_path / "pnpm-lock.yaml").write_text("") + + conflict = detect_lockfile_conflicts(str(tmp_path)) + + assert conflict is not None + assert len(conflict.lockfiles) == 3 From 1bfa7efb4e13abd8cc890507982d6ddb93420af6 Mon Sep 17 00:00:00 2001 From: smurftyy Date: Thu, 9 Jul 2026 01:06:06 +0100 Subject: [PATCH 6/8] Implemented the compression logic --- README.md | 61 +++++-- pyproject.toml | 4 + src/devklean/cli/commands/clean.py | 21 ++- src/devklean/config/manager.py | 11 ++ src/devklean/config/models.py | 13 ++ src/devklean/deletion/compression.py | 228 +++++++++++++++++++++++++-- src/devklean/deletion/trash.py | 111 +++++++++++-- src/devklean/output/text.py | 15 +- src/devklean/tui.py | 12 +- tests/test_compression.py | 147 +++++++++++++++++ tests/test_config.py | 6 + tests/test_deletion.py | 134 ++++++++++++++-- 12 files changed, 700 insertions(+), 63 deletions(-) create mode 100644 tests/test_compression.py diff --git a/README.md b/README.md index 66bb6c0..5b5e7a4 100644 --- a/README.md +++ b/README.md @@ -25,7 +25,7 @@ pipx install devklean pip install devklean ``` -Requires Python 3.8+. Runtime dependencies are `send2trash` (used for all deletions) and `tomli` (only on Python < 3.11). +Requires Python 3.8+. Runtime dependencies are `send2trash` (used for all deletions) and `tomli` (only on Python < 3.11). `zstandard` is an optional dependency, only needed for `compress_format = "zstd"` — see [Compression](#compression). ## Quick start @@ -70,7 +70,7 @@ devklean clean --dry-run # show what *would* be deleted; delete nothing devklean clean -i # interactive: pick items in a TUI (Linux/macOS only) devklean clean --allow-symlinks # permit deleting symlinked targets (blocked by default) devklean clean -y # skip the y/N prompt (large deletions still require typing DELETE) -devklean clean --compress # zip eligible directories before sending them to trash +devklean clean --compress # compress eligible directories (gzip) before sending them to trash ``` | Flag | Meaning | @@ -79,7 +79,7 @@ devklean clean --compress # zip eligible directories before sending them t | `-i`, `--interactive` | Choose items in a terminal UI (SPACE select, A all, D none, ENTER confirm, Q quit). **Linux/macOS only** — see [Platform support](#platform-support). | | `--allow-symlinks` | Allow deleting symbolic links. Off by default (symlinks are blocked). | | `-y`, `--yes` | Skip the standard confirmation. Deletions over the size threshold still require typing `DELETE`. | -| `--compress` | Zip eligible directories into a sibling archive before trashing them, shrinking their footprint in trash. Off by default — see [Compression](#compression). | +| `--compress` | Compress eligible directories into a sibling archive before trashing them, shrinking their footprint in trash. Off by default — see [Compression](#compression). | ### `restore` @@ -95,9 +95,10 @@ devklean restore # explains how to recover from the Recycle Bin / Trash - **macOS** — open Trash in Finder and "Put Back". - **Linux** — open Trash in your file manager and restore. -If the item was deleted with `--compress`, what lands in trash is a `.zip` -archive rather than the original directory — restore the archive from trash, -then unpack it to the original path yourself. devklean does not decompress +If the item was deleted with `--compress`, what lands in trash is a `.tar.gz` +(or `.tar.zst`, if `compress_format = "zstd"`) archive rather than the original +directory — restore the archive from trash, then extract it to the original +path yourself, e.g. `tar -xf .tar.gz`. devklean does not decompress automatically (yet). Run `devklean history` to see what was removed and when. @@ -170,7 +171,9 @@ exclude = ["node_modules", ".git"] dry_run = false interactive = false default_yes = false # skip the y/N prompt (the large-deletion DELETE gate still applies) -compress = false # zip eligible directories before trashing them +compress = false # compress eligible directories before trashing them +compress_min_size = 10485760 # bytes; directories smaller than this are trashed uncompressed (default 10 MiB) +compress_format = "gzip" # "gzip" (stdlib, default) or "zstd" (needs the devklean[zstd] extra) theme = "default" # "default" or "mono" confirm_threshold = 1073741824 # bytes; deletions >= this require typing DELETE (default 1 GiB) path = "." @@ -193,18 +196,46 @@ Color follows the `theme` setting and is automatically disabled when output is p Common artifacts (`node_modules`, `.venv`, `.next`, build caches) often compress to a fraction of their on-disk size. Pass `--compress` (or set `compress = true` -in config) and devklean will zip each eligible directory into a sibling -`.zip` archive and send *that* to trash instead of the raw directory — -shrinking how much space the deletion actually occupies in trash before it's -emptied. - -- Only applies to directories (not symlinks); files are trashed as-is. -- The archive path and format are recorded in deletion metadata, so `history` - and `doctor` see compressed deletions the same as uncompressed ones. +in config) and devklean will archive each eligible directory into a sibling +`.tar.gz` (gzip, the default) before sending *that* to trash instead of the raw +directory — shrinking how much space the deletion actually occupies in trash +before it's emptied. `zstd` is available as an opt-in format (see below) for a +better compression ratio. + +Compression is always ordered for safety: devklean compresses to a temp +archive, verifies it (test-extracts every entry and cross-checks the file +count and total uncompressed size against the source), and only *after* +`send2trash` confirms the archive is in the trash does it remove the original +directory. If compression or verification fails, or `send2trash` itself fails, +the original directory is left completely untouched and the error is reported +per-item — nothing is ever partially deleted. + +- Only applies to directories (not symlinks) at or above `compress_min_size` + (default 10 MiB); smaller directories and files are trashed as-is, since the + archive/verify overhead rarely pays for itself below that size. +- The archive path, format, original size, and compressed size are recorded in + deletion metadata, so `history` and `doctor` see compressed deletions the + same as uncompressed ones. - Off by default, so existing scripts and habits aren't surprised by archives appearing in trash. - Restoring a compressed item is manual today — see [`restore`](#restore). +### zstd (optional) + +```bash +pip install 'devklean[zstd]' +``` + +```toml +[defaults] +compress = true +compress_format = "zstd" +``` + +If `compress_format = "zstd"` is set but the `zstandard` package isn't +installed, devklean logs a warning and falls back to gzip rather than +crashing. + ## Logs `devklean` writes detailed structured logs (commands, scanned/deleted paths, sizes, errors) to: diff --git a/pyproject.toml b/pyproject.toml index 1eb7699..444b12d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -36,12 +36,16 @@ dependencies = [ ] [project.optional-dependencies] +zstd = [ + "zstandard>=0.21", +] dev = [ "pytest>=7.0", "pytest-cov>=4.0", "ruff>=0.6", "mypy>=1.8", "build>=1.0", + "zstandard>=0.21", ] [project.scripts] diff --git a/src/devklean/cli/commands/clean.py b/src/devklean/cli/commands/clean.py index 384feb0..63e580c 100644 --- a/src/devklean/cli/commands/clean.py +++ b/src/devklean/cli/commands/clean.py @@ -8,7 +8,7 @@ confirm_large_deletion, exceeds_threshold, ) -from devklean.config.models import AppConfig +from devklean.config.models import DEFAULT_COMPRESS_FORMAT, DEFAULT_COMPRESS_MIN_SIZE, AppConfig from devklean.deletion.safety import SafetyValidator from devklean.models import CleanableItem from devklean.output.base import Renderer @@ -46,6 +46,8 @@ def run_standard( validator: SafetyValidator | None = None, *, compress: bool = False, + compress_min_size: int = DEFAULT_COMPRESS_MIN_SIZE, + compress_format: str = DEFAULT_COMPRESS_FORMAT, default_yes: bool = False, confirm_threshold: int = DEFAULT_LARGE_THRESHOLD, ) -> None: @@ -59,7 +61,14 @@ def run_standard( renderer.aborted() return - result = delete_items(found, total_size, validator=validator, compress=compress) + result = delete_items( + found, + total_size, + validator=validator, + compress=compress, + compress_min_size=compress_min_size, + compress_format=compress_format, + ) renderer.deletion_result(result) @@ -78,6 +87,10 @@ def run_clean( default_yes = getattr(args, "yes", False) or getattr(defaults, "default_yes", False) confirm_threshold = getattr(defaults, "confirm_threshold", DEFAULT_LARGE_THRESHOLD) compress = getattr(args, "compress", False) or getattr(defaults, "compress", False) + # Not exposed as CLI flags (per the original issue, --compress itself is + # the only opt-in surface); config-only knobs read straight from defaults. + compress_min_size = getattr(defaults, "compress_min_size", DEFAULT_COMPRESS_MIN_SIZE) + compress_format = getattr(defaults, "compress_format", DEFAULT_COMPRESS_FORMAT) if args.interactive: # Interactive mode relies on curses, which is unavailable on Windows. @@ -100,6 +113,8 @@ def run_clean( args.dry_run, validator, compress=compress, + compress_min_size=compress_min_size, + compress_format=compress_format, confirm_threshold=confirm_threshold, ) else: @@ -109,6 +124,8 @@ def run_clean( args.dry_run, validator, compress=compress, + compress_min_size=compress_min_size, + compress_format=compress_format, default_yes=default_yes, confirm_threshold=confirm_threshold, ) diff --git a/src/devklean/config/manager.py b/src/devklean/config/manager.py index 4eb8d79..5226119 100644 --- a/src/devklean/config/manager.py +++ b/src/devklean/config/manager.py @@ -21,6 +21,8 @@ "dry_run", "interactive", "compress", + "compress_min_size", + "compress_format", "path", "default_yes", "theme", @@ -153,6 +155,8 @@ def _merge_defaults(self, layers: list[dict[str, Any]]) -> DefaultsConfig: "dry_run": base.dry_run, "interactive": base.interactive, "compress": base.compress, + "compress_min_size": base.compress_min_size, + "compress_format": base.compress_format, "path": base.path, "default_yes": base.default_yes, "theme": base.theme, @@ -167,6 +171,13 @@ def _merge_defaults(self, layers: list[dict[str, Any]]) -> DefaultsConfig: merged[key] = bool(section[key]) if "compress" in section: merged["compress"] = bool(section["compress"]) + if "compress_min_size" in section: + try: + merged["compress_min_size"] = int(section["compress_min_size"]) + except (TypeError, ValueError): + pass + if "compress_format" in section: + merged["compress_format"] = str(section["compress_format"]) if "path" in section: merged["path"] = str(section["path"]) if "theme" in section: diff --git a/src/devklean/config/models.py b/src/devklean/config/models.py index 744979b..a58ef42 100644 --- a/src/devklean/config/models.py +++ b/src/devklean/config/models.py @@ -8,12 +8,25 @@ # CLI/TUI confirmation flow re-exports this as DEFAULT_LARGE_THRESHOLD. DEFAULT_CONFIRM_THRESHOLD = 1024**3 # 1 GiB +# Single source of truth for compress-before-trash defaults; deletion/trash.py +# imports these rather than defining its own copies, so config and the +# deletion pipeline can't drift apart on what "the default" is. +# Below this size, tar/gzip overhead and the extra verify-read pass rarely +# pay for themselves, so --compress skips the directory and trashes it as-is. +DEFAULT_COMPRESS_MIN_SIZE = 10 * 1024 * 1024 # 10 MiB +# "gzip" (stdlib tarfile, always available) or "zstd" (needs the +# devklean[zstd] extra; silently falls back to gzip with a logged warning +# when requested but not installed). +DEFAULT_COMPRESS_FORMAT = "gzip" + @dataclass(frozen=True) class DefaultsConfig: dry_run: bool = False interactive: bool = False compress: bool = False + compress_min_size: int = DEFAULT_COMPRESS_MIN_SIZE + compress_format: str = DEFAULT_COMPRESS_FORMAT path: str = "." default_yes: bool = False theme: str = "default" diff --git a/src/devklean/deletion/compression.py b/src/devklean/deletion/compression.py index 5d23b88..b810f00 100644 --- a/src/devklean/deletion/compression.py +++ b/src/devklean/deletion/compression.py @@ -1,27 +1,227 @@ +"""Compress-before-trash: build and verify an archive without ever touching +the source directory. + +This module only does two things — ``compress_path`` and ``verify_archive`` — +and neither one deletes or modifies the source. Ordering the source's removal +around a *verified* archive is the caller's job (``devklean.deletion.trash``); +keeping that decision out of this module is what makes the ordering testable +on its own. + +Default format is gzip via the stdlib ``tarfile`` ('w:gz') — no new hard +dependency. zstd is opt-in via the ``devklean[zstd]`` extra (the +``zstandard`` package); stdlib ``tarfile`` only gained native zstd support in +Python 3.14, so this is manual streaming through ``zstandard`` for older +versions. If zstd is requested but the extra isn't installed, this silently +(but loggedly) falls back to gzip rather than crashing. +""" + from __future__ import annotations -import shutil +import os +import tarfile +from contextlib import ExitStack, contextmanager from dataclasses import dataclass from pathlib import Path +from tempfile import mkstemp +from typing import Iterator + +from devklean.logging_setup import get_logger +from devklean.scanner import get_dir_size + +GZIP_FORMAT = "gzip" +ZSTD_FORMAT = "zstd" +SUPPORTED_FORMATS = (GZIP_FORMAT, ZSTD_FORMAT) -ARCHIVE_FORMAT = "zip" +_ARCHIVE_SUFFIX = { + GZIP_FORMAT: ".tar.gz", + ZSTD_FORMAT: ".tar.zst", +} + +# Read/write back through the decompressor in chunks during verification, +# rather than materializing whole files in memory. +_VERIFY_CHUNK_SIZE = 1024 * 1024 + + +class CompressionVerificationError(Exception): + """A freshly built archive failed verification. + + Callers must treat this the same as a compression failure: the source + directory has not been touched and must stay exactly as it was. + """ @dataclass(frozen=True) -class CompressionArchive: - path: str +class CompressionResult: + """Everything the trash pipeline needs to verify and then record an archive.""" + + archive_path: Path format: str + original_size: int + file_count: int + + @property + def compressed_size(self) -> int: + return self.archive_path.stat().st_size + + +def compress_path(source: Path, *, format: str = GZIP_FORMAT) -> CompressionResult: + """Archive ``source`` into a new temp file. Never reads-then-deletes; + ``source`` is left completely untouched, success or failure. + + The archive is written next to ``source`` (same filesystem as the + original, so a later move never has to cross devices) with a name that + cannot collide with a real scan target. On any failure while writing, + the partial temp file is removed and the exception propagates — callers + must not treat a raised exception here as "maybe partially done". + """ + resolved_format = _resolve_format(format) + archive_path = _new_temp_archive_path(source, resolved_format) + + try: + with _open_archive_for_write(archive_path, resolved_format) as tar: + file_count = _add_tree(tar, source) + except Exception: + archive_path.unlink(missing_ok=True) + raise + + original_size = get_dir_size(str(source)) + return CompressionResult( + archive_path=archive_path, + format=resolved_format, + original_size=original_size, + file_count=file_count, + ) + + +def verify_archive(result: CompressionResult) -> None: + """Test-extract every regular-file entry and cross-check count/size. + + Entries are streamed and discarded (never written to disk), so this costs + time but no extra disk space. Raises ``CompressionVerificationError`` on + any mismatch, corruption, or unreadable entry — never returns a partial + or best-effort "probably fine" result. + """ + file_count = 0 + total_size = 0 + try: + with _open_archive_for_read(result.archive_path, result.format) as tar: + for member in tar: + if not member.isreg(): + continue + extracted = tar.extractfile(member) + if extracted is None: + raise CompressionVerificationError( + f"archive entry {member.name!r} in {result.archive_path} " + "could not be read back" + ) + while extracted.read(_VERIFY_CHUNK_SIZE): + pass + file_count += 1 + total_size += member.size + except CompressionVerificationError: + raise + except Exception as exc: + # Corruption can surface as tarfile.TarError, OSError, EOFError + # (truncated gzip stream), zlib/zstd decode errors, and more, + # depending on exactly where the archive was damaged. Every one of + # them means the same thing here: the archive does not read back + # cleanly, so it must not be trusted as a stand-in for the source. + raise CompressionVerificationError( + f"archive {result.archive_path} failed verification: {exc}" + ) from exc -def compress_directory(source: Path) -> CompressionArchive: - """Archive a directory into a sibling zip file and remove the source tree.""" - archive_path = Path( - shutil.make_archive( - str(source), - ARCHIVE_FORMAT, - root_dir=source.parent, - base_dir=source.name, + if file_count != result.file_count: + raise CompressionVerificationError( + f"archive {result.archive_path} contains {file_count} files, " + f"expected {result.file_count}" ) + if total_size != result.original_size: + raise CompressionVerificationError( + f"archive {result.archive_path} contains {total_size} uncompressed bytes, " + f"expected {result.original_size}" + ) + + +# --- internals --- + + +def _resolve_format(format: str) -> str: + if format == ZSTD_FORMAT and not _zstd_available(): + get_logger().warning( + "zstd compression requested but the 'zstandard' package is not installed " + "(pip install 'devklean[zstd]'); falling back to gzip" + ) + return GZIP_FORMAT + if format not in SUPPORTED_FORMATS: + get_logger().warning("unknown compression format %r; falling back to gzip", format) + return GZIP_FORMAT + return format + + +def _zstd_available() -> bool: + try: + import zstandard # noqa: F401 + except ImportError: + return False + return True + + +def _new_temp_archive_path(source: Path, resolved_format: str) -> Path: + fd, name = mkstemp( + prefix=f".{source.name}-", + suffix=_ARCHIVE_SUFFIX[resolved_format], + dir=source.parent, ) - shutil.rmtree(source) - return CompressionArchive(path=str(archive_path), format=ARCHIVE_FORMAT) + os.close(fd) + return Path(name) + + +def _add_tree(tar: tarfile.TarFile, source: Path) -> int: + """Add ``source`` (recursively) to ``tar``. Returns the regular-file count.""" + count = 0 + + def _count_filter(tarinfo: tarfile.TarInfo) -> tarfile.TarInfo: + nonlocal count + if tarinfo.isreg(): + count += 1 + return tarinfo + + tar.add(str(source), arcname=source.name, recursive=True, filter=_count_filter) + return count + + +@contextmanager +def _open_archive_for_write(archive_path: Path, resolved_format: str) -> Iterator[tarfile.TarFile]: + with ExitStack() as stack: + if resolved_format == ZSTD_FORMAT: + zstandard = _import_zstandard() + raw = stack.enter_context(archive_path.open("wb")) + writer = stack.enter_context(zstandard.ZstdCompressor().stream_writer(raw)) + tar = stack.enter_context(tarfile.open(fileobj=writer, mode="w|")) + else: + tar = stack.enter_context(tarfile.open(archive_path, mode="w:gz")) + yield tar + + +@contextmanager +def _open_archive_for_read(archive_path: Path, resolved_format: str) -> Iterator[tarfile.TarFile]: + with ExitStack() as stack: + if resolved_format == ZSTD_FORMAT: + zstandard = _import_zstandard() + raw = stack.enter_context(archive_path.open("rb")) + reader = stack.enter_context(zstandard.ZstdDecompressor().stream_reader(raw)) + tar = stack.enter_context(tarfile.open(fileobj=reader, mode="r|")) + else: + tar = stack.enter_context(tarfile.open(archive_path, mode="r:gz")) + yield tar + + +def _import_zstandard(): + try: + import zstandard + except ImportError as exc: # pragma: no cover - resolved_format guarantees availability + raise CompressionVerificationError( + "zstd archive requires the 'zstandard' package (pip install 'devklean[zstd]')" + ) from exc + return zstandard diff --git a/src/devklean/deletion/trash.py b/src/devklean/deletion/trash.py index 7701044..6771f0e 100644 --- a/src/devklean/deletion/trash.py +++ b/src/devklean/deletion/trash.py @@ -1,11 +1,17 @@ from __future__ import annotations +import shutil from collections.abc import Sequence from pathlib import Path from send2trash import send2trash -from devklean.deletion.compression import compress_directory +from devklean.config.models import DEFAULT_COMPRESS_FORMAT, DEFAULT_COMPRESS_MIN_SIZE +from devklean.deletion.compression import ( + CompressionVerificationError, + compress_path, + verify_archive, +) from devklean.deletion.metadata import TRASH_STRATEGY, DeletionArchive, MetadataManager from devklean.deletion.safety import SafetyValidator from devklean.logging_setup import get_logger @@ -25,6 +31,8 @@ def delete_items( metadata_manager: MetadataManager | None = None, dry_run: bool = False, compress: bool = False, + compress_min_size: int = DEFAULT_COMPRESS_MIN_SIZE, + compress_format: str = DEFAULT_COMPRESS_FORMAT, ) -> DeleteResult: """Validate, then move safe items to the native OS trash via ``send2trash``. @@ -33,6 +41,12 @@ def delete_items( *before any ``send2trash`` call is reachable* — the structural guarantee that a dry run performs no filesystem operations. Successful deletions are recorded in the metadata store (used by ``history`` and ``doctor``). + + When ``compress`` is set, eligible items go through ``_send_to_trash``'s + compress -> verify -> trash -> remove-original ordering (see its + docstring); a failure at any step before the original is removed leaves + it completely untouched and is reported as a per-item failure, the same + as any other ``send2trash`` error. """ validator = validator or SafetyValidator() safe, blocked = validator.partition(items) @@ -62,17 +76,21 @@ def delete_items( archives: dict[str, DeletionArchive] = {} for item in safe: try: - trash_path = item.path - if compress and _should_compress(item): - archive = compress_directory(Path(item.path)) - archives[item.path] = DeletionArchive(path=archive.path, format=archive.format) - trash_path = archive.path - send2trash(trash_path) + archive = _send_to_trash( + item, + compress=compress, + compress_min_size=compress_min_size, + compress_format=compress_format, + ) + if archive is not None: + archives[item.path] = archive deleted.append(item.path) - except OSError as exc: + except (OSError, CompressionVerificationError) as exc: # TrashPermissionError subclasses OSError; ENOENT/EACCES and - # platform-specific failures surface here too. Report the path and - # keep going so one bad item never aborts the batch. + # platform-specific failures surface here too, alongside + # CompressionVerificationError from a failed compress/verify. + # Report the path and keep going so one bad item never aborts + # the batch. failures.append(DeleteFailure(path=item.path, error=str(exc))) result = DeleteResult( @@ -86,11 +104,12 @@ def delete_items( for failure in result.failed: logger.warning("delete failed path=%s error=%s", failure.path, failure.error) logger.info( - "deletion summary strategy=%s deleted=%d failed=%d size=%d", + "deletion summary strategy=%s deleted=%d failed=%d size=%d compressed=%d", STRATEGY_NAME, result.deleted_count, result.failed_count, result.total_size, + len(archives), ) manager = metadata_manager or MetadataManager() @@ -98,6 +117,72 @@ def delete_items( return result -def _should_compress(item: CleanableItem) -> bool: +def _send_to_trash( + item: CleanableItem, + *, + compress: bool, + compress_min_size: int, + compress_format: str, +) -> DeletionArchive | None: + """Send one item to the native OS trash, compressing first when eligible. + + The ordering here is the entire compress-before-trash safety contract: + + 1. Compress the source to a temp archive. The source is never touched. + 2. Verify the archive (test-extract + count/size cross-check). The + source is still untouched; a failure here removes only the temp + archive and raises — nothing has happened to the source. + 3. ``send2trash`` the *archive*. Only once this call returns successfully + has anything been "deleted" from the user's perspective. + 4. Only now, with a verified copy already confirmed in the trash, remove + the original directory directly. This does not also go through + ``send2trash`` — the archive already serves as the recoverable copy, + so trashing the original too would duplicate space there and defeat + the point of compressing first. + + A failure at step 4 (the original can't be removed after all) is + reported distinctly from a failure at steps 1-3: the archive is already + safe in the trash, so no data has been lost, but the original directory + is still on disk and the caller must be told that plainly. + """ + if not (compress and _should_compress(item, compress_min_size)): + send2trash(item.path) + return None + + source = Path(item.path) + result = compress_path(source, format=compress_format) + try: + verify_archive(result) + except CompressionVerificationError: + result.archive_path.unlink(missing_ok=True) + raise + + compressed_size = result.compressed_size # read now: the file moves to trash next + try: + send2trash(str(result.archive_path)) + except OSError: + result.archive_path.unlink(missing_ok=True) + raise + + try: + shutil.rmtree(source) + except OSError as exc: + raise OSError( + f"archive is safely in the trash, but removing the original directory " + f"{source} afterward failed ({exc}); the original is still on disk" + ) from exc + + return DeletionArchive( + path=str(result.archive_path), + format=result.format, + compressed=True, + original_size=result.original_size, + compressed_size=compressed_size, + ) + + +def _should_compress(item: CleanableItem, compress_min_size: int) -> bool: source = Path(item.path) - return source.is_dir() and not source.is_symlink() + if not source.is_dir() or source.is_symlink(): + return False + return item.size >= compress_min_size diff --git a/src/devklean/output/text.py b/src/devklean/output/text.py index 5d8c40b..b89df85 100644 --- a/src/devklean/output/text.py +++ b/src/devklean/output/text.py @@ -262,9 +262,18 @@ def restore_help(self) -> None: c.detail(" • Windows — open the Recycle Bin and restore the item.") c.detail(" • macOS — open Trash in Finder and 'Put Back'.") c.detail(" • Linux — open Trash in your file manager and restore.") + self._println() + c.detail( + "If it was deleted with --compress, what's sitting in the trash is a " + "compressed archive (.tar.gz, or .tar.zst if zstd was used) — not the " + "original directory." + ) c.detail( - "If compression was enabled, restore the archive from trash and unpack it " - "to the original path." + "After restoring the archive from trash, extract it yourself to get the " + "files back, e.g. `tar -xf .tar.gz` (or `tar --zstd -xf .tar.zst`)." ) self._println() - c.detail("Run `devklean history` to see what was removed and when.") + c.detail( + "Run `devklean history` to see what was removed, when, and whether it was " + "compressed." + ) diff --git a/src/devklean/tui.py b/src/devklean/tui.py index 725f7af..6f7b8b6 100644 --- a/src/devklean/tui.py +++ b/src/devklean/tui.py @@ -5,6 +5,7 @@ confirm_large_deletion, exceeds_threshold, ) +from devklean.config.models import DEFAULT_COMPRESS_FORMAT, DEFAULT_COMPRESS_MIN_SIZE from devklean.deletion.safety import SafetyValidator from devklean.formatting import format_size, truncate from devklean.models import CleanableItem @@ -101,6 +102,8 @@ def run_interactive( validator: SafetyValidator | None = None, *, compress: bool = False, + compress_min_size: int = DEFAULT_COMPRESS_MIN_SIZE, + compress_format: str = DEFAULT_COMPRESS_FORMAT, confirm_threshold: int = DEFAULT_LARGE_THRESHOLD, ) -> None: import curses # Unix-only; imported lazily so this module loads on Windows. @@ -129,5 +132,12 @@ def run_interactive( renderer.aborted() return - result = delete_items(selected, total_size, validator=validator, compress=compress) + result = delete_items( + selected, + total_size, + validator=validator, + compress=compress, + compress_min_size=compress_min_size, + compress_format=compress_format, + ) renderer.deletion_result(result) diff --git a/tests/test_compression.py b/tests/test_compression.py new file mode 100644 index 0000000..a7d3910 --- /dev/null +++ b/tests/test_compression.py @@ -0,0 +1,147 @@ +"""Real-filesystem tests for compress_path/verify_archive. + +Per the compress-before-trash safety contract, compress_path must never touch +the source directory, and verify_archive must reliably reject anything that +doesn't read back cleanly. These tests exercise real tar/gzip (and, when +available, real zstd) I/O rather than mocking the compression internals — +mocks alone can't catch a real corrupt-archive or real-filesystem-permission +failure mode. +""" + +from __future__ import annotations + +import logging +from dataclasses import replace +from pathlib import Path + +import pytest + +from devklean.deletion.compression import ( + CompressionVerificationError, + compress_path, + verify_archive, +) + + +def _make_tree(root: Path) -> int: + """Create a small real directory tree; returns its total byte size.""" + root.mkdir(parents=True) + (root / "a.txt").write_text("hello" * 100, encoding="utf-8") + nested = root / "nested" + nested.mkdir() + (nested / "b.bin").write_bytes(b"\x00" * 4096) + return sum(f.stat().st_size for f in root.rglob("*") if f.is_file()) + + +def test_compress_path_never_touches_the_source(tmp_path: Path) -> None: + source = tmp_path / "node_modules" + size = _make_tree(source) + + result = compress_path(source) + + assert source.exists() + assert (source / "a.txt").exists() + assert (source / "nested" / "b.bin").exists() + assert result.archive_path.exists() + assert result.archive_path.parent == source.parent + assert result.format == "gzip" + assert result.original_size == size + assert result.file_count == 2 + assert result.compressed_size > 0 + + result.archive_path.unlink() + + +def test_verify_archive_passes_for_a_valid_archive(tmp_path: Path) -> None: + source = tmp_path / "node_modules" + _make_tree(source) + result = compress_path(source) + + verify_archive(result) # must not raise + + result.archive_path.unlink() + + +def test_verify_archive_raises_on_truncated_archive(tmp_path: Path) -> None: + source = tmp_path / "node_modules" + _make_tree(source) + result = compress_path(source) + + data = result.archive_path.read_bytes() + result.archive_path.write_bytes(data[: len(data) // 2]) + + with pytest.raises(CompressionVerificationError): + verify_archive(result) + + assert source.exists() # verification failure must never touch the source + result.archive_path.unlink() + + +def test_verify_archive_raises_on_file_count_mismatch(tmp_path: Path) -> None: + source = tmp_path / "node_modules" + _make_tree(source) + result = compress_path(source) + tampered = replace(result, file_count=result.file_count + 1) + + with pytest.raises(CompressionVerificationError, match="files"): + verify_archive(tampered) + + result.archive_path.unlink() + + +def test_verify_archive_raises_on_size_mismatch(tmp_path: Path) -> None: + source = tmp_path / "node_modules" + _make_tree(source) + result = compress_path(source) + tampered = replace(result, original_size=result.original_size + 1) + + with pytest.raises(CompressionVerificationError, match="bytes"): + verify_archive(tampered) + + result.archive_path.unlink() + + +def test_compress_path_cleans_up_temp_file_on_failure(tmp_path: Path, monkeypatch) -> None: + source = tmp_path / "node_modules" + _make_tree(source) + + def _boom(*args, **kwargs): + raise OSError("simulated tar write failure") + + monkeypatch.setattr("tarfile.TarFile.add", _boom) + + with pytest.raises(OSError): + compress_path(source) + + assert source.exists() + assert (source / "a.txt").exists() + assert list(tmp_path.glob(".node_modules-*")) == [] + + +def test_zstd_format_round_trips_with_real_zstandard(tmp_path: Path) -> None: + pytest.importorskip("zstandard") + source = tmp_path / "dist" + _make_tree(source) + + result = compress_path(source, format="zstd") + + assert result.format == "zstd" + verify_archive(result) + assert source.exists() + result.archive_path.unlink() + + +def test_zstd_falls_back_to_gzip_when_package_missing( + tmp_path: Path, monkeypatch, caplog: pytest.LogCaptureFixture +) -> None: + source = tmp_path / "dist" + _make_tree(source) + monkeypatch.setattr("devklean.deletion.compression._zstd_available", lambda: False) + + with caplog.at_level(logging.WARNING, logger="devklean"): + result = compress_path(source, format="zstd") + + assert result.format == "gzip" + assert any("falling back to gzip" in record.message for record in caplog.records) + verify_archive(result) + result.archive_path.unlink() diff --git a/tests/test_config.py b/tests/test_config.py index 79d75f8..aeb12e3 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -23,6 +23,8 @@ def test_config_manager_uses_defaults_when_missing(tmp_path: Path) -> None: assert config.defaults.dry_run is False assert config.defaults.interactive is False assert config.defaults.compress is False + assert config.defaults.compress_min_size == 10 * 1024 * 1024 + assert config.defaults.compress_format == "gzip" @pytest.mark.skipif(sys.platform == "win32", reason="POSIX absolute path literals") @@ -34,6 +36,8 @@ def test_config_manager_merges_user_settings(tmp_path: Path) -> None: dry_run = true interactive = true compress = true +compress_min_size = 2097152 +compress_format = "zstd" path = "~/projects" [targets] @@ -61,6 +65,8 @@ def test_config_manager_merges_user_settings(tmp_path: Path) -> None: assert config.defaults.dry_run is True assert config.defaults.interactive is True assert config.defaults.compress is True + assert config.defaults.compress_min_size == 2097152 + assert config.defaults.compress_format == "zstd" assert config.defaults.path == "~/projects" diff --git a/tests/test_deletion.py b/tests/test_deletion.py index e12bc53..130cee4 100644 --- a/tests/test_deletion.py +++ b/tests/test_deletion.py @@ -8,6 +8,7 @@ from devklean.cli.commands.clean import run_standard from devklean.deletion import delete_items +from devklean.deletion.compression import CompressionVerificationError from devklean.deletion.metadata import MetadataManager from devklean.models import CleanableItem, DeleteFailure, DeleteResult @@ -29,30 +30,133 @@ def test_delete_items_delegates_to_send2trash(tmp_path: Path, fake_trash) -> Non assert result.total_size == 1024 -def test_delete_items_compresses_directory_before_trashing(tmp_path: Path, monkeypatch) -> None: +def test_delete_items_compresses_directory_before_trashing(tmp_path: Path, fake_trash) -> None: source = tmp_path / "workspace" / "node_modules" source.mkdir(parents=True) for index in range(64): (source / f"file-{index}.txt").write_text("A" * 2048, encoding="utf-8") + original_size = sum(f.stat().st_size for f in source.rglob("*") if f.is_file()) - trashed: list[str] = [] + item = CleanableItem(str(source), "node_modules", original_size, "Node.js") + manager = MetadataManager(storage_dir=tmp_path / "m") - def _record_only(path) -> None: - trashed.append(str(path)) + # compress_min_size=0 forces compression regardless of size, so the test + # doesn't need to write a real 10 MiB+ fixture to cross the default gate. + result = delete_items( + [item], item.size, metadata_manager=manager, compress=True, compress_min_size=0 + ) - monkeypatch.setattr("devklean.deletion.trash.send2trash", _record_only) + assert result.deleted == (str(source),) + assert result.failed == () + assert not source.exists() # removed only after the archive was confirmed trashed + + [trashed_path] = fake_trash + archive = Path(trashed_path) + assert archive.name.startswith(".node_modules-") + assert archive.name.endswith(".tar.gz") + assert not archive.exists() # fake_trash removed it, same as a real send2trash call would + + records = manager.load_records() + (stored,) = records.records + archive_record = stored.record.archive + assert archive_record is not None + assert archive_record.format == "gzip" + assert archive_record.compressed is True + assert archive_record.original_size == original_size + assert archive_record.compressed_size is not None + assert archive_record.compressed_size < original_size + + +def test_delete_items_skips_compression_below_min_size(tmp_path: Path, fake_trash) -> None: + source = tmp_path / "workspace" / ".cache" + source.mkdir(parents=True) + (source / "small.txt").write_text("hi", encoding="utf-8") - item = CleanableItem(str(source), "node_modules", 64 * 2048, "Node.js") + item = CleanableItem(str(source), ".cache", 2, "Cache") manager = MetadataManager(storage_dir=tmp_path / "m") - result = delete_items([item], item.size, metadata_manager=manager, compress=True) + result = delete_items( + [item], + item.size, + metadata_manager=manager, + compress=True, + compress_min_size=10 * 1024 * 1024, + ) assert result.deleted == (str(source),) - assert trashed == [str(source.with_suffix(".zip"))] - assert not source.exists() - archive = source.with_suffix(".zip") - assert archive.exists() - assert archive.stat().st_size < item.size + assert fake_trash == [str(source)] # trashed as-is, never compressed + records = manager.load_records() + (stored,) = records.records + assert stored.record.archive is None + + +def test_delete_items_leaves_original_intact_when_verification_fails( + tmp_path: Path, monkeypatch +) -> None: + """Reproduces the PR #12 failure mode: a failure partway through the + compress-before-trash sequence must never delete the source. PR #12's bug + was calling shutil.rmtree(source) unconditionally right after building the + archive, with no verification and before send2trash ran at all — so a + later send2trash failure meant the data was already gone. This asserts + the fix: verification (or send2trash) failing leaves the source directory + completely untouched, with no partial deletion.""" + source = tmp_path / "workspace" / "node_modules" + source.mkdir(parents=True) + (source / "a.txt").write_text("A" * 2048, encoding="utf-8") + (source / "b.txt").write_text("B" * 2048, encoding="utf-8") + original_size = sum(f.stat().st_size for f in source.rglob("*") if f.is_file()) + + def _boom(result) -> None: + raise CompressionVerificationError("simulated corruption detected mid-verification") + + monkeypatch.setattr("devklean.deletion.trash.verify_archive", _boom) + + item = CleanableItem(str(source), "node_modules", original_size, "Node.js") + manager = MetadataManager(storage_dir=tmp_path / "m") + + result = delete_items( + [item], item.size, metadata_manager=manager, compress=True, compress_min_size=0 + ) + + assert result.deleted == () + assert result.failed[0].path == str(source) + assert source.exists() + assert (source / "a.txt").exists() + assert (source / "b.txt").exists() + # the failed temp archive must not be left behind either + assert list((tmp_path / "workspace").glob(".node_modules-*")) == [] + assert manager.load_records().records == () # nothing recorded for a failed item + + +def test_delete_items_leaves_original_intact_when_send2trash_fails( + tmp_path: Path, monkeypatch +) -> None: + """Same failure-mode reproduction, one step later: the archive verifies + fine but send2trash itself fails (disk full, permission denied, ...). + The original directory must still be completely untouched — it is only + ever removed *after* send2trash confirms the archive made it to trash.""" + source = tmp_path / "workspace" / "node_modules" + source.mkdir(parents=True) + (source / "a.txt").write_text("A" * 2048, encoding="utf-8") + original_size = sum(f.stat().st_size for f in source.rglob("*") if f.is_file()) + + def _boom(path) -> None: + raise OSError("simulated send2trash failure") + + monkeypatch.setattr("devklean.deletion.trash.send2trash", _boom) + + item = CleanableItem(str(source), "node_modules", original_size, "Node.js") + manager = MetadataManager(storage_dir=tmp_path / "m") + + result = delete_items( + [item], item.size, metadata_manager=manager, compress=True, compress_min_size=0 + ) + + assert result.deleted == () + assert result.failed[0].path == str(source) + assert source.exists() + assert (source / "a.txt").exists() + assert list((tmp_path / "workspace").glob(".node_modules-*")) == [] def test_delete_items_does_not_call_send2trash_on_dry_run(tmp_path: Path, fake_trash) -> None: @@ -200,7 +304,7 @@ def _selective(path) -> None: payload = json.loads(records[0].read_text(encoding="utf-8")) assert result.deleted == ("/tmp/a",) - assert payload["schema_version"] == 4 + assert payload["schema_version"] == 5 assert payload["deletion"]["strategy"] == "trash" assert isinstance(payload["deletion"]["run_id"], str) and payload["deletion"]["run_id"] assert payload["item"]["original_path"] == "/tmp/a" @@ -229,8 +333,8 @@ def test_metadata_manager_records_archive_details(tmp_path: Path) -> None: records = sorted(storage_dir.glob("*.json")) payload = json.loads(records[0].read_text(encoding="utf-8")) - assert payload["schema_version"] == 4 - assert payload["archive"] == {"path": "/tmp/a.zip", "format": "zip"} + assert payload["schema_version"] == 5 + assert payload["archive"] == {"path": "/tmp/a.zip", "format": "zip", "compressed": True} def test_metadata_manager_skips_failed_deletions(tmp_path: Path) -> None: From af2a735a14a7ca60f75e67ef057e6349fca2cd77 Mon Sep 17 00:00:00 2001 From: smurftyy Date: Thu, 9 Jul 2026 01:29:53 +0100 Subject: [PATCH 7/8] style: ruff format --- src/devklean/output/text.py | 3 +-- src/devklean/signatures/staleness.py | 16 ++++++++++------ src/devklean/signatures/structural.py | 4 +--- 3 files changed, 12 insertions(+), 11 deletions(-) diff --git a/src/devklean/output/text.py b/src/devklean/output/text.py index b89df85..ae6d036 100644 --- a/src/devklean/output/text.py +++ b/src/devklean/output/text.py @@ -274,6 +274,5 @@ def restore_help(self) -> None: ) self._println() c.detail( - "Run `devklean history` to see what was removed, when, and whether it was " - "compressed." + "Run `devklean history` to see what was removed, when, and whether it was compressed." ) diff --git a/src/devklean/signatures/staleness.py b/src/devklean/signatures/staleness.py index e66c213..6736f24 100644 --- a/src/devklean/signatures/staleness.py +++ b/src/devklean/signatures/staleness.py @@ -49,12 +49,16 @@ class StalenessResult: def estimate_staleness(project_root: str) -> StalenessResult: """Resolve a staleness signal for ``project_root``, git first, then mtime.""" - return _from_git(project_root) or _from_source_mtime(project_root) or StalenessResult( - known=False, - source=None, - last_activity=None, - days_since=None, - detail="no reliable signal (no git repository and no source files found)", + return ( + _from_git(project_root) + or _from_source_mtime(project_root) + or StalenessResult( + known=False, + source=None, + last_activity=None, + days_since=None, + detail="no reliable signal (no git repository and no source files found)", + ) ) diff --git a/src/devklean/signatures/structural.py b/src/devklean/signatures/structural.py index 0c63e27..f4b1002 100644 --- a/src/devklean/signatures/structural.py +++ b/src/devklean/signatures/structural.py @@ -32,9 +32,7 @@ def detect_lockfile_conflicts(project_root: str) -> LockfileConflict | None: "correct" or which package manager is actually in use. """ for group in _CONFLICTING_LOCKFILE_GROUPS: - present = tuple( - name for name in group if os.path.isfile(os.path.join(project_root, name)) - ) + present = tuple(name for name in group if os.path.isfile(os.path.join(project_root, name))) if len(present) >= 2: return LockfileConflict(project_root=project_root, lockfiles=present) return None From 3a09f53d685615a9c81c3783434b7ea7c0b28600 Mon Sep 17 00:00:00 2001 From: smurftyy Date: Thu, 9 Jul 2026 01:32:51 +0100 Subject: [PATCH 8/8] chore:trigger CI