perf(scan): cache unchanged objective scan results

benchmarks/benchmark_incremental_cache.py

@@ -0,0 +1,58 @@
+"""Microbenchmark the cold and warm objective-scan manifest path."""
+
+from __future__ import annotations
+
+import tempfile
+from pathlib import Path
+from time import perf_counter
+
+from core.runtime_state import make_runtime_context, runtime_scope
+from engine.planning.incremental_cache import prepare_scan_cache, store_scan_cache
+from languages._framework.base.types import DetectorPhase, LangConfig
+from languages._framework.runtime import make_lang_run
+
+
+def main(file_count: int = 10_000) -> None:
+    with tempfile.TemporaryDirectory() as directory:
+        root = Path(directory)
+        for index in range(file_count):
+            (root / f"module_{index:05d}.py").write_text(f"VALUE = {index}\n")
+        language = make_lang_run(
+            LangConfig(
+                name="benchmark",
+                extensions=[".py"],
+                exclusions=[],
+                default_src=".",
+                build_dep_graph=lambda _: {},
+                entry_patterns=[],
+                barrel_names=set(),
+                file_finder=lambda _: [str(path) for path in root.glob("*.py")],
+            )
+        )
+        phase = DetectorPhase("benchmark", lambda *_: ([], {}))
+        runtime = make_runtime_context()
+        runtime.project_root = root
+        with runtime_scope(runtime):
+            started = perf_counter()
+            cold = prepare_scan_cache(
+                root, language, [phase], profile="ci", include_slow=False,
+                zone_overrides=None,
+            )
+            cold_seconds = perf_counter() - started
+            store_scan_cache(cold, [], {})
+            started = perf_counter()
+            warm = prepare_scan_cache(
+                root, language, [phase], profile="ci", include_slow=False,
+                zone_overrides=None,
+            )
+            warm_seconds = perf_counter() - started
+        print({
+            "files": file_count,
+            "cold_seconds": round(cold_seconds, 4),
+            "warm_seconds": round(warm_seconds, 4),
+            "hit": warm.hit,
+        })
+
+
+if __name__ == "__main__":
+    main()

docs/incremental-scans.md

@@ -0,0 +1,23 @@
+# Incremental objective scans
+
+`structorium scan --profile ci` and `--profile objective` persist a
+content-addressed result under `.structorium/cache/objective-scan-v1.json`.
+An unchanged scan reuses normalized objective findings. Full scans are not
+cached because subjective review may depend on time or external state.
+
+The key covers the Structorium source hash, selected phases, language/runtime
+configuration, zone overrides, and every scanned source digest. The manifest
+reuses a digest only when both file size and nanosecond modification time are
+unchanged. Writes are atomic and corrupt cache documents are ignored.
+
+Disable reuse for parity/debugging with:
+
+```bash
+STRUCTORIUM_DISABLE_INCREMENTAL_CACHE=1 structorium scan --profile ci
+```
+
+Run the 10,000-file manifest benchmark with:
+
+```bash
+PYTHONPATH=. python benchmarks/benchmark_incremental_cache.py
+```

engine/planning/incremental_cache.py

@@ -0,0 +1,160 @@
+"""Persistent content-addressed cache for deterministic objective scans."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+
+from core._internal.text_utils import get_project_root
+from languages._framework.base.types import DetectorPhase
+from languages._framework.runtime import LangRun
+from versioning import compute_tool_hash
+
+CACHE_SCHEMA = 1
+
+
+@dataclass(frozen=True)
+class ScanCacheLookup:
+    """Prepared cache key plus an optional normalized result."""
+
+    cache_path: Path
+    key: str
+    files: dict[str, dict[str, int | str]]
+    findings: list[dict[str, Any]] | None = None
+    potentials: dict[str, int] | None = None
+
+    @property
+    def hit(self) -> bool:
+        return self.findings is not None and self.potentials is not None
+
+
+def _read_document(cache_path: Path) -> dict[str, Any]:
+    try:
+        payload = json.loads(cache_path.read_text(encoding="utf-8"))
+    except (OSError, ValueError, TypeError):
+        return {}
+    return payload if isinstance(payload, dict) else {}
+
+
+def _source_files(scan_path: Path, lang: LangRun) -> list[Path]:
+    if lang.file_finder is None:
+        return []
+    root = get_project_root().resolve()
+    candidates = lang.file_finder(root)
+    resolved: set[Path] = set()
+    scan_root = scan_path.resolve()
+    for candidate in candidates:
+        path = Path(candidate)
+        if not path.is_absolute():
+            path = root / path
+        path = path.resolve()
+        if path.is_file() and (path.is_relative_to(scan_root) or scan_root == root):
+            resolved.add(path)
+    return sorted(resolved)
+
+
+def _build_manifest(
+    paths: list[Path], previous: dict[str, Any]
+) -> dict[str, dict[str, int | str]]:
+    root = get_project_root().resolve()
+    manifest: dict[str, dict[str, int | str]] = {}
+    previous_files = previous.get("files", {})
+    if not isinstance(previous_files, dict):
+        previous_files = {}
+
+    for path in paths:
+        try:
+            stat = path.stat()
+        except OSError:
+            continue
+        try:
+            relative = path.relative_to(root).as_posix()
+        except ValueError:
+            relative = path.as_posix()
+        old = previous_files.get(relative, {})
+        unchanged = (
+            isinstance(old, dict)
+            and old.get("size") == stat.st_size
+            and old.get("mtime_ns") == stat.st_mtime_ns
+            and isinstance(old.get("sha256"), str)
+        )
+        if unchanged:
+            digest = str(old["sha256"])
+        else:
+            try:
+                digest = hashlib.sha256(path.read_bytes()).hexdigest()
+            except OSError:
+                continue
+        manifest[relative] = {
+            "size": stat.st_size,
+            "mtime_ns": stat.st_mtime_ns,
+            "sha256": digest,
+        }
+    return manifest
+
+
+def prepare_scan_cache(
+    scan_path: Path,
+    lang: LangRun,
+    phases: list[DetectorPhase],
+    *,
+    profile: str,
+    include_slow: bool,
+    zone_overrides: dict[str, str] | None,
+) -> ScanCacheLookup:
+    """Return a cache lookup, hashing only files whose stat fingerprint changed."""
+    root = get_project_root().resolve()
+    cache_path = root / ".structorium" / "cache" / "objective-scan-v1.json"
+    previous = _read_document(cache_path)
+    manifest = _build_manifest(_source_files(scan_path, lang), previous)
+    key_payload = {
+        "schema": CACHE_SCHEMA,
+        "tool": compute_tool_hash(),
+        "language": lang.name,
+        "profile": profile,
+        "include_slow": include_slow,
+        "phases": [(phase.label, phase.slow) for phase in phases],
+        "settings": lang.runtime_settings,
+        "options": lang.runtime_options,
+        "zone_overrides": zone_overrides or {},
+        "files": manifest,
+    }
+    key = hashlib.sha256(
+        json.dumps(key_payload, sort_keys=True, separators=(",", ":"), default=str).encode()
+    ).hexdigest()
+    result = previous.get("result", {})
+    if previous.get("key") == key and isinstance(result, dict):
+        findings = result.get("findings")
+        potentials = result.get("potentials")
+        if isinstance(findings, list) and isinstance(potentials, dict):
+            return ScanCacheLookup(
+                cache_path, key, manifest, findings, potentials
+            )
+    return ScanCacheLookup(cache_path, key, manifest)
+
+
+def store_scan_cache(
+    lookup: ScanCacheLookup,
+    findings: list[dict[str, Any]],
+    potentials: dict[str, int],
+) -> None:
+    """Atomically persist a successful objective-scan result."""
+    document = {
+        "schema": CACHE_SCHEMA,
+        "key": lookup.key,
+        "files": lookup.files,
+        "result": {"findings": findings, "potentials": potentials},
+    }
+    lookup.cache_path.parent.mkdir(parents=True, exist_ok=True)
+    temporary = lookup.cache_path.with_suffix(".tmp")
+    temporary.write_text(
+        json.dumps(document, sort_keys=True, separators=(",", ":"), default=str),
+        encoding="utf-8",
+    )
+    temporary.replace(lookup.cache_path)
+
+
+__all__ = ["ScanCacheLookup", "prepare_scan_cache", "store_scan_cache"]

engine/planning/scan.py

@@ -2,19 +2,20 @@
 
 from __future__ import annotations
 
+import os
 import sys
 from dataclasses import dataclass
 from pathlib import Path
 
 from core._internal.text_utils import PROJECT_ROOT
+from core.file_paths import rel
+from core.output_api import colorize
 from engine.planning.common import is_subjective_phase
 from engine.policy.zones import ZONE_POLICIES, FileZoneMap
-from core.file_paths import rel
 from languages import auto_detect_lang, available_langs, get_lang
 from languages._framework.base.types import DetectorPhase, LangConfig
 from languages._framework.runtime import LangRun, make_lang_run
 from state import Finding
-from core.output_api import colorize
 
 
 @dataclass
@@ -125,8 +126,33 @@ def _generate_findings_from_lang(
     """Run detector phases from a LangRun."""
     _build_zone_map(path, lang, zone_overrides)
     phases = _select_phases(lang, include_slow=include_slow, profile=profile)
-    findings, all_potentials = _run_phases(path, lang, phases)
+    cache_lookup = None
+    cache_allowed = (
+        profile in {"objective", "ci"}
+        and os.environ.get("STRUCTORIUM_DISABLE_INCREMENTAL_CACHE") != "1"
+    )
+    if cache_allowed:
+        from engine.planning.incremental_cache import prepare_scan_cache
+
+        cache_lookup = prepare_scan_cache(
+            path,
+            lang,
+            phases,
+            profile=profile,
+            include_slow=include_slow,
+            zone_overrides=zone_overrides,
+        )
+    if cache_lookup is not None and cache_lookup.hit:
+        _stderr("  Incremental cache: objective findings reused")
+        findings = list(cache_lookup.findings or [])
+        all_potentials = dict(cache_lookup.potentials or {})
+    else:
+        findings, all_potentials = _run_phases(path, lang, phases)
     _stamp_finding_context(findings, lang)
+    if cache_lookup is not None and not cache_lookup.hit:
+        from engine.planning.incremental_cache import store_scan_cache
+
+        store_scan_cache(cache_lookup, findings, all_potentials)
     _stderr(f"\n  Total: {len(findings)} findings")
     return findings, all_potentials
 

tests/scan/test_incremental_cache.py

@@ -0,0 +1,76 @@
+from pathlib import Path
+
+from engine.planning.incremental_cache import prepare_scan_cache, store_scan_cache
+from languages._framework.base.types import DetectorPhase, LangConfig
+from languages._framework.runtime import make_lang_run
+
+
+def _language(root: Path):
+    def files(_: Path) -> list[str]:
+        return [str(path) for path in sorted(root.glob("*.py"))]
+
+    return make_lang_run(
+        LangConfig(
+            name="fixture",
+            extensions=[".py"],
+            exclusions=[],
+            default_src=".",
+            build_dep_graph=lambda _: {},
+            entry_patterns=[],
+            barrel_names=set(),
+            file_finder=files,
+        )
+    )
+
+
+def _phase() -> DetectorPhase:
+    return DetectorPhase("fixture", lambda path, lang: ([], {}))
+
+
+def _lookup(root: Path, monkeypatch):
+    monkeypatch.setenv("STRUCTORIUM_ROOT", str(root))
+    from core.runtime_state import make_runtime_context, runtime_scope
+
+    runtime = make_runtime_context()
+    runtime.project_root = root
+    with runtime_scope(runtime):
+        return prepare_scan_cache(
+            root,
+            _language(root),
+            [_phase()],
+            profile="ci",
+            include_slow=False,
+            zone_overrides=None,
+        )
+
+
+def test_round_trip_cache_hit(tmp_path: Path, monkeypatch) -> None:
+    (tmp_path / "app.py").write_text("value = 1\n")
+    first = _lookup(tmp_path, monkeypatch)
+    assert not first.hit
+    store_scan_cache(first, [{"detector": "fixture"}], {"fixture": 1})
+
+    second = _lookup(tmp_path, monkeypatch)
+    assert second.hit
+    assert second.findings == [{"detector": "fixture"}]
+
+
+def test_source_change_invalidates_result(tmp_path: Path, monkeypatch) -> None:
+    source = tmp_path / "app.py"
+    source.write_text("value = 1\n")
+    first = _lookup(tmp_path, monkeypatch)
+    store_scan_cache(first, [], {})
+
+    source.write_text("value = 200\n")
+    second = _lookup(tmp_path, monkeypatch)
+    assert not second.hit
+    assert second.key != first.key
+
+
+def test_corrupt_cache_fails_open(tmp_path: Path, monkeypatch) -> None:
+    (tmp_path / "app.py").write_text("value = 1\n")
+    cache = tmp_path / ".structorium" / "cache" / "objective-scan-v1.json"
+    cache.parent.mkdir(parents=True)
+    cache.write_text("not-json")
+
+    assert not _lookup(tmp_path, monkeypatch).hit