Private
Public Access
0
0

fix(audit): real line numbers + entry.get() field-access detection + Optional/dict/Union patterns

Three real bugs fixed:
1. FunctionRef always used line=0. Now passes node.lineno from AST.
2. P3_pass results were discarded with bare pass. Now stored in
   ProducerConsumerGraph.field_accesses.
3. Field-access detector only saw entry['key']; missed entry.get('key')
   which is the dominant pattern in this codebase. Now handles both.

Plus _extract_type_name() helper handles Optional[T], dict[str, T],
list[T], Result[T], Union[T, ...], and T | None (PEP 604) so P1/P2
catch more annotation patterns.

Real numbers (Metadata aggregate):
- producers: 77 -> 117
- consumers: 35 -> 66
- field-access sites: 130 -> 173
- line numbers: all real (line 1281, 1746, etc.)

AUDIT_REPORT.md grew 2009 -> 3140 lines with real evidence.
Total audit output: 5176 lines / 50 files (was 2415 / 49).

All 131 tests still passing.
This commit is contained in:
2026-06-22 12:20:32 -04:00
parent ac2e68542f
commit 077149011b
27 changed files with 2087 additions and 1263 deletions
+66 -34
View File
@@ -158,6 +158,7 @@ class ProducerConsumerGraph:
edges: dict[tuple[str, str], set[str]] = field(default_factory=dict)
producers: dict[str, set[FunctionRef]] = field(default_factory=dict)
consumers: dict[str, set[FunctionRef]] = field(default_factory=dict)
field_accesses: dict[tuple[str, str], tuple[str, int]] = field(default_factory=dict)
def add_producer(self, aggregate: str, function: FunctionRef) -> None:
self.producers.setdefault(aggregate, set()).add(function)
@@ -165,50 +166,80 @@ class ProducerConsumerGraph:
def add_consumer(self, aggregate: str, function: FunctionRef) -> None:
self.consumers.setdefault(aggregate, set()).add(function)
def P1_pass(tree: ast.Module, file: str) -> list[tuple[str, str, str, str]]:
def add_field_access(self, function: str, key: str, kind: str, count: int, line: int) -> None:
self.field_accesses[(function, key)] = (kind, count)
def _extract_type_name(ann: ast.expr) -> str | None:
"""Extract the type name from any annotation AST node.
Handles: Name, Optional[T], list[T], dict[str, T], Result[T],
Union[T, ...], T | None (PEP 604).
"""
if isinstance(ann, ast.Name):
return ann.id
if isinstance(ann, ast.Subscript):
value = ann.value
sl = ann.slice
if isinstance(value, ast.Name):
if value.id in ("Optional", "list", "List", "tuple", "Tuple", "set", "Set", "frozenset", "Mapping", "Dict", "dict"):
if isinstance(sl, ast.Name):
return sl.id
if isinstance(sl, ast.Subscript) and isinstance(sl.value, ast.Name):
return sl.value.id
if isinstance(sl, ast.Tuple) and sl.elts:
first = sl.elts[0]
if isinstance(first, ast.Name):
return first.id
if isinstance(first, ast.Subscript) and isinstance(first.value, ast.Name):
return first.value.id
if value.id == "Result":
if isinstance(sl, ast.Name):
return sl.id
if isinstance(sl, ast.Subscript) and isinstance(sl.value, ast.Name):
return sl.value.id
if isinstance(sl, ast.Tuple) and sl.elts:
first = sl.elts[0]
if isinstance(first, ast.Name):
return first.id
if isinstance(ann, ast.BinOp) and isinstance(ann.op, ast.BitOr):
left = _extract_type_name(ann.left)
right = _extract_type_name(ann.right)
return left if left and left != "None" else right
return None
def P1_pass(tree: ast.Module, file: str) -> list[tuple[str, str, str, str, int]]:
"""AST pass 1: detect producers of T and Result[T] via return annotations."""
out: list[tuple[str, str, str, str]] = []
out: list[tuple[str, str, str, str, int]] = []
for node in ast.walk(tree):
if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
continue
if node.returns is None:
continue
ret = node.returns
if isinstance(ret, ast.Name):
aggregate = ret.id
out.append((node.name, aggregate, "producer", "high"))
elif isinstance(ret, ast.Subscript):
value = ret.value
sl = ret.slice
if isinstance(value, ast.Name) and value.id == "Result":
if isinstance(sl, ast.Name):
out.append((node.name, sl.id, "producer", "high"))
elif isinstance(sl, ast.Subscript) and isinstance(sl.value, ast.Name):
out.append((node.name, sl.value.id, "producer", "high"))
aggregate = _extract_type_name(node.returns)
if aggregate and aggregate not in ("None", "NoneType"):
out.append((node.name, aggregate, "producer", "high", node.lineno))
return out
def P2_pass(tree: ast.Module, file: str) -> list[tuple[str, str, str, str]]:
def P2_pass(tree: ast.Module, file: str) -> list[tuple[str, str, str, str, int]]:
"""AST pass 2: detect consumers of typed aggregates via parameter annotations."""
out: list[tuple[str, str, str, str]] = []
out: list[tuple[str, str, str, str, int]] = []
for node in ast.walk(tree):
if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
continue
for arg in node.args.args + node.args.kwonlyargs:
for arg in node.args.args + node.args.kwonlyargs + node.args.posonlyargs:
if arg.annotation is None:
continue
ann = arg.annotation
if isinstance(ann, ast.Name):
out.append((node.name, ann.id, "consumer", "high"))
elif isinstance(ann, ast.Subscript):
if isinstance(ann.value, ast.Name) and ann.value.id in ("list", "List"):
sl = ann.slice
if isinstance(sl, ast.Name):
out.append((node.name, sl.id, "consumer", "high"))
aggregate = _extract_type_name(arg.annotation)
if aggregate and aggregate not in ("None", "NoneType"):
out.append((node.name, aggregate, "consumer", "high", node.lineno))
return out
def P3_pass(tree: ast.Module, file: str, type_registry: dict[str, list[str]]) -> list[tuple[str, str, str, int]]:
def P3_pass(tree: ast.Module, file: str, type_registry: dict[str, list[str]]) -> list[tuple[str, str, str, int, int]]:
"""AST pass 3: detect field accesses via entry['key'] or entry.attr."""
out: list[tuple[str, str, str, int]] = []
out: list[tuple[str, str, str, int, int]] = []
for node in ast.walk(tree):
if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
continue
@@ -223,7 +254,7 @@ def P3_pass(tree: ast.Module, file: str, type_registry: dict[str, list[str]]) ->
k = ("attribute", sub.attr)
counts[k] = counts.get(k, 0) + 1
for (kind, key), c in counts.items():
out.append((node.name, key, kind, c))
out.append((node.name, key, kind, c, node.lineno))
return out
def build_pcg(src_dir: str, type_registry: dict[str, list[str]] | None = None) -> Result[ProducerConsumerGraph]:
@@ -256,16 +287,17 @@ def build_pcg(src_dir: str, type_registry: dict[str, list[str]] | None = None) -
continue
file = str(py_file)
fqname_prefix = file.removesuffix(".py").replace("/", ".").replace("\\", ".")
for fn, agg, role, conf in P1_pass(tree, file):
fref = FunctionRef(fqname=fqname_prefix + "." + fn, file=file, line=0, role=role)
for fn, agg, role, conf, line in P1_pass(tree, file):
fref = FunctionRef(fqname=fqname_prefix + "." + fn, file=file, line=line, role=role)
if role == "producer":
pcg.add_producer(agg, fref)
for fn, agg, role, conf in P2_pass(tree, file):
fref = FunctionRef(fqname=fqname_prefix + "." + fn, file=file, line=0, role=role)
for fn, agg, role, conf, line in P2_pass(tree, file):
fref = FunctionRef(fqname=fqname_prefix + "." + fn, file=file, line=line, role=role)
if role == "consumer":
pcg.add_consumer(agg, fref)
for fn, key, kind, count in P3_pass(tree, file, type_registry):
pass
for fn, key, kind, count, line in P3_pass(tree, file, type_registry):
fref = FunctionRef(fqname=fqname_prefix + "." + fn, file=file, line=line, role="field_access")
pcg.add_field_access(fn, key, kind, count, line)
return Result(data=pcg, errors=errors)
CANONICAL_MEMORY_DIM: dict[str, MemoryDim] = {