Repository navigation
Expand file tree
/
Copy pathstructural_check.py
More file actions
221 lines (197 loc) · 9.06 KB
/
Copy pathstructural_check.py
File metadata and controls
221 lines (197 loc) · 9.06 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
"""`cgr check`: the structural delta of the working tree against a git ref.
The graph is assumed to reflect `--base` (index at the base, then edit);
the files that differ between the base and the working tree are re-ingested
and the delta reported the same way the MCP write tools report it after
each write (issue #1525). Exit status 1 with `--fail-on-found` makes it a
CI or pre-commit gate.
The check measures the graph against the working tree, and the re-ingest
brings the graph up to the tree, so a second run on an unchanged tree
reports nothing: the delta was already applied. Rebuild the graph at the
base (or index at the base before editing) to measure the same edit again.
The CLI holds no lock against a concurrently running MCP server; like every
other command that writes the graph, run it when no other writer is active.
"""
from __future__ import annotations
import subprocess
from collections.abc import Mapping
from pathlib import Path
from tree_sitter import Parser
from . import constants as cs
from .config import load_ignore_patterns
from .graph_updater import GraphUpdater, _load_exclusion_state
from .structural_delta import StructuralDelta, normalise_paths, observe
from .types_defs import LanguageQueries
from .utils.path_utils import derive_project_name
_GIT_DELETED = "D"
_CGR_STATE_PREFIX = ".cgr-"
class CheckError(ValueError):
"""The working tree cannot be compared against the requested base."""
def changed_since(repo_root: Path, base: str) -> tuple[list[str], list[str]]:
"""Files that differ from `base`: (changed or added, deleted).
Untracked files count as added; a rename shows as one deletion and one
addition so the delta reports the old symbols as removed or renamed.
"""
if base.startswith("-"):
# Placed before git's own `--`, a dash-prefixed value would be read
# as a diff option (`--cached` compares the index) rather than a
# revision, and the check would measure the wrong files.
raise CheckError(cs.CHECK_BAD_BASE.format(base=base))
# A range (`HEAD~1..HEAD`, `main...HEAD`) makes `git diff` compare its
# endpoints instead of a commit with the working tree, silently leaving
# the current edits out; only a value naming one commit is a base.
verified = subprocess.run(
[cs.SHELL_CMD_GIT, "rev-parse", "--verify", "--quiet", f"{base}^{{commit}}"],
cwd=repo_root,
capture_output=True,
encoding=cs.ENCODING_UTF8,
check=False,
)
if verified.returncode != 0:
raise CheckError(cs.CHECK_BASE_NOT_A_COMMIT.format(base=base))
try:
status = subprocess.run(
# `--relative`: paths relative to `repo_root`, which may sit below
# the git toplevel; `ls-files --others` is already cwd-relative.
# `-z`: NUL-delimited, unquoted paths, so a name holding a tab or
# a newline is not C-quoted into something that does not exist.
[
cs.SHELL_CMD_GIT,
"diff",
"--name-status",
"--no-renames",
"--relative",
"-z",
base,
"--",
],
cwd=repo_root,
capture_output=True,
encoding=cs.ENCODING_UTF8,
check=True,
).stdout
untracked = subprocess.run(
[cs.SHELL_CMD_GIT, "ls-files", "--others", "--exclude-standard", "-z"],
cwd=repo_root,
capture_output=True,
encoding=cs.ENCODING_UTF8,
check=True,
).stdout
except (OSError, subprocess.CalledProcessError) as error:
detail = getattr(error, "stderr", "") or str(error)
raise CheckError(
cs.CHECK_GIT_FAILED.format(base=base, error=detail.strip())
) from error
changed: set[str] = set()
deleted: set[str] = set()
# `-z` output alternates status and path fields, each NUL-terminated.
fields = [f for f in status.split("\0") if f]
for code, path in zip(fields[0::2], fields[1::2], strict=False):
(deleted if code.startswith(_GIT_DELETED) else changed).add(path)
# cgr's own untracked state files (hash cache, directory mtimes, ...)
# are not source and must not be re-ingested or reported as reparsed.
changed.update(
entry
for entry in untracked.split("\0")
if entry and not Path(entry).name.startswith(_CGR_STATE_PREFIX)
)
return sorted(changed), sorted(deleted)
def _stamp_is_named(stored: dict[str, list[str] | str]) -> bool:
"""Whether the run that wrote this stamp was given an explicit --project.
Stamps written before the field existed have no `named` key; they read
as unnamed, which is what the overwhelming majority of them were.
"""
return bool(stored.get("named"))
def indexed_scope(
repo_root: Path, project_name: str, *, explicit: bool = False
) -> tuple[frozenset[str] | None, frozenset[str] | None]:
"""The exclusion scope `project_name`'s graph was last indexed under.
The last completed run stamps its effective `--exclude` and unignore
sets (CLI flags included, not only `.cgrignore`) in the exclusion state
file; the check must re-ingest under that same scope or a file the
index deliberately left out would enter the graph. The stamp is per
repository, so one written by ANOTHER project indexed from this tree
refuses the check rather than lending that project's scope. Without a
stamp the `.cgrignore` file is the only scope there is.
Ownership comes from the stamp, which records whether the run that
wrote it was given a `--project`. The stamp is per repository and the
last run of ANY project overwrites it, so the check must establish that
it belongs to the project asked for and refuse otherwise: a scope from
the wrong project silently re-ingests files the index left out, or drops
files it deliberately kept.
"""
stored = _load_exclusion_state(repo_root / cs.EXCLUSION_STATE_FILENAME)
if stored is not None:
owner = stored.get("project")
# `cgr index` without --project stamps the bare directory name
# where `cgr check` derives the digest-suffixed one, so an UNNAMED
# run's stamp answers to either spelling of this tree's identity.
# A NAMED run's answers only to the name it was given.
#
# The two cannot be told apart by the string alone -- a project
# deliberately called `myrepo` in a directory called `myrepo` writes
# the same owner an unnamed run does -- which is why the writer
# records `named` and this reads it rather than guessing (#1525).
#
# `explicit` is what the CALLER asked for. Both sides must agree:
# an unnamed stamp does not serve `--project myrepo`, because the
# last unnamed run of this tree overwrote whatever the named project
# had stamped, and its scope is simply gone rather than inferable.
default_names = {derive_project_name(repo_root), repo_root.resolve().name}
stamp_named = _stamp_is_named(stored)
if stamp_named:
mine = owner == project_name
else:
mine = (
not explicit
and owner in default_names
and project_name in default_names
)
if isinstance(owner, str) and not mine:
raise CheckError(
cs.CHECK_SCOPE_OF_OTHER_PROJECT.format(
project=project_name, other=owner
)
)
exclude = stored.get("exclude") or []
unignore = stored.get("unignore") or []
return (
frozenset(exclude) or None, # type: ignore[arg-type]
frozenset(unignore) or None, # type: ignore[arg-type]
)
cgrignore = load_ignore_patterns(repo_root)
return cgrignore.exclude or None, cgrignore.unignore or None
def run_check(
repo_root: Path,
base: str,
project_name: str,
ingestor: object,
parsers: Mapping[cs.SupportedLanguage, Parser],
queries: Mapping[cs.SupportedLanguage, LanguageQueries],
exclude_paths: frozenset[str] | None = None,
unignore_paths: frozenset[str] | None = None,
) -> StructuralDelta:
"""Re-ingest what changed since `base` and return the structural delta.
`exclude_paths` and `unignore_paths` are the project's indexing scope
(the `.cgrignore` file plus any CLI excludes): a changed file outside
that scope must not enter the graph through the check.
"""
changed, deleted = changed_since(repo_root, base)
changed = normalise_paths(changed, repo_root)
deleted = normalise_paths(deleted, repo_root)
updater = GraphUpdater(
ingestor=ingestor, # type: ignore[arg-type]
repo_path=repo_root,
parsers=parsers,
queries=queries,
project_name=project_name,
exclude_paths=exclude_paths,
unignore_paths=unignore_paths,
)
fetch_all = getattr(ingestor, "fetch_all")
return observe(
fetch_all,
project_name,
[*changed, *deleted],
lambda: updater.reingest(changed, deleted=deleted),
repo_root=repo_root,
)