Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
68 changes: 68 additions & 0 deletions scripts/codelens.py
Original file line number Diff line number Diff line change
Expand Up @@ -998,6 +998,69 @@ def compute_confidence_distribution_flat(result: Dict[str, Any]) -> Dict[str, in
return dist


# ─── Always-Warm Registry (issue #237) ─────────────────────────

_AUTO_RESCAN_FILE_THRESHOLD = 20


def _check_staleness(workspace: str, args) -> Optional[Dict[str, Any]]:
"""Check registry staleness vs git HEAD and auto-rescan small diffs.

Registry staleness vs git HEAD was previously only surfaced passively via
`history --check git-status`'s "re-scan recommendation" field — nothing
else read it. Every other analysis command could silently answer from a
stale graph. Note: deliberately NOT gated on the same _REGISTRY_COMMANDS
set as the auto-setup block in main() — that set still lists pre-#195
leaf command names and is missing several current umbrella commands
(audit, security, deps, doctor — see issue #244); "every command except
scan itself" is simpler and correct without depending on that list
being fixed.

Returns:
None if not stale (or staleness can't be determined — e.g. not a
git repo, git unavailable, no registry yet). Otherwise a dict with
``was_stale``, ``auto_rescanned``, ``changed_files_count``, and
(when not auto-rescanned) a ``hint`` field.
"""
if args.command == "scan" or not _registry_exists(workspace):
return None
try:
import git_aware
from utils import default_db_path
db_path = getattr(args, "db_path", None) or default_db_path(workspace)
if not git_aware.rescan_recommended(workspace, db_path):
return None
last_sha = git_aware.get_last_indexed_sha(workspace, db_path)
changed_files = (
git_aware.get_changed_files(workspace, since_sha=last_sha)
if last_sha else []
)
changed_count = len(changed_files)
if 0 < changed_count <= _AUTO_RESCAN_FILE_THRESHOLD:
from commands.scan import cmd_scan
cmd_scan(workspace, incremental=True)
return {
"was_stale": True,
"auto_rescanned": True,
"changed_files_count": changed_count,
}
return {
"was_stale": True,
"auto_rescanned": False,
"changed_files_count": changed_count,
"hint": (
"Registry may be stale vs current git HEAD "
f"(>{_AUTO_RESCAN_FILE_THRESHOLD} files changed or "
"branch switch detected). Run 'scan --incremental' "
"to refresh before trusting these results."
),
}
except Exception:
# Best-effort — never block the actual command on staleness
# detection failing (e.g. not a git repo, git binary missing).
return None


# ─── CLI Entry Point ──────────────────────────────────────────

def main():
Expand Down Expand Up @@ -1413,11 +1476,16 @@ def main():
"message": f"Auto-setup failed at {auto_setup_result.get('stage')}: {auto_setup_result.get('error')}",
}

staleness_info = _check_staleness(workspace, args)

try:
cmd_info = registry[args.command]

result = cmd_info["execute"](args, workspace)

if staleness_info and isinstance(result, dict):
result["_staleness"] = staleness_info

# ─── Dispatch enrichment (scan-specific) ──────
if args.command == "scan":
# Auto-save snapshot after scan
Expand Down
106 changes: 106 additions & 0 deletions tests/test_staleness_check.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,106 @@
"""Tests for _check_staleness() — always-warm registry (issue #237).

Registry staleness vs git HEAD was previously only surfaced passively via
`history --check git-status`'s "re-scan recommendation" field — no other
command read it, so every analysis command could silently answer from a
stale graph. `_check_staleness()` is called for every command except
`scan` itself: small diffs (<= threshold files) trigger a transparent
incremental re-scan; larger diffs just attach a hint instead of forcing
a possibly-slow rescan on every call.
"""

import argparse
import os
import sys
from unittest import mock

import pytest

SCRIPT_DIR = os.path.join(
os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "scripts"
)
if SCRIPT_DIR not in sys.path:
sys.path.insert(0, SCRIPT_DIR)

from codelens import _check_staleness, _AUTO_RESCAN_FILE_THRESHOLD # noqa: E402


def _args(command="context", db_path=None):
return argparse.Namespace(command=command, db_path=db_path)


class TestCheckStaleness:
def test_scan_command_never_checked(self):
"""scan itself must never trigger staleness checking — it's the
command that WOULD update the staleness, checking before running
it would be circular."""
with mock.patch("codelens._registry_exists", return_value=True):
result = _check_staleness(".", _args(command="scan"))
assert result is None

def test_no_registry_returns_none(self):
with mock.patch("codelens._registry_exists", return_value=False):
result = _check_staleness(".", _args())
assert result is None

def test_not_stale_returns_none(self):
with mock.patch("codelens._registry_exists", return_value=True), \
mock.patch("git_aware.rescan_recommended", return_value=False):
result = _check_staleness(".", _args())
assert result is None

def test_small_diff_auto_rescans(self):
"""A diff at or below the threshold triggers a transparent
incremental rescan and reports auto_rescanned=True."""
changed = [f"file{i}.py" for i in range(_AUTO_RESCAN_FILE_THRESHOLD)]
with mock.patch("codelens._registry_exists", return_value=True), \
mock.patch("git_aware.rescan_recommended", return_value=True), \
mock.patch("git_aware.get_last_indexed_sha", return_value="abc123"), \
mock.patch("git_aware.get_changed_files", return_value=changed), \
mock.patch("commands.scan.cmd_scan") as mock_scan:
result = _check_staleness(".", _args())

mock_scan.assert_called_once_with(".", incremental=True)
assert result["was_stale"] is True
assert result["auto_rescanned"] is True
assert result["changed_files_count"] == _AUTO_RESCAN_FILE_THRESHOLD
assert "hint" not in result

def test_large_diff_hints_without_rescanning(self):
"""A diff above the threshold must NOT trigger an automatic
rescan (could be slow/expensive) — just a hint."""
changed = [f"file{i}.py" for i in range(_AUTO_RESCAN_FILE_THRESHOLD + 1)]
with mock.patch("codelens._registry_exists", return_value=True), \
mock.patch("git_aware.rescan_recommended", return_value=True), \
mock.patch("git_aware.get_last_indexed_sha", return_value="abc123"), \
mock.patch("git_aware.get_changed_files", return_value=changed), \
mock.patch("commands.scan.cmd_scan") as mock_scan:
result = _check_staleness(".", _args())

mock_scan.assert_not_called()
assert result["was_stale"] is True
assert result["auto_rescanned"] is False
assert result["changed_files_count"] == _AUTO_RESCAN_FILE_THRESHOLD + 1
assert "hint" in result

def test_branch_switch_no_last_sha_treated_as_large_diff(self):
"""detect_branch_switch (via rescan_recommended) can fire with no
resolvable last_sha — must not crash trying to diff against None,
and must not silently auto-rescan an unknown-size change."""
with mock.patch("codelens._registry_exists", return_value=True), \
mock.patch("git_aware.rescan_recommended", return_value=True), \
mock.patch("git_aware.get_last_indexed_sha", return_value=None), \
mock.patch("commands.scan.cmd_scan") as mock_scan:
result = _check_staleness(".", _args())

mock_scan.assert_not_called()
assert result["changed_files_count"] == 0
assert result["auto_rescanned"] is False

def test_exception_in_git_aware_never_propagates(self):
"""Not a git repo / git binary missing must degrade to None, not
crash the command that was actually requested."""
with mock.patch("codelens._registry_exists", return_value=True), \
mock.patch("git_aware.rescan_recommended", side_effect=RuntimeError("no git")):
result = _check_staleness(".", _args())
assert result is None
Loading