toki/iop
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
iop/scripts/readability_audit.py

2079 lines
76 KiB
Python
Raw Blame History

#!/usr/bin/env python3
"""readability_audit.py
Deterministic readability audit for the IOP monorepo.
Measures file LOC, function size, and task-local read-set (ordered file LOC
totals with budget/reason) for tracked Go/Dart/Shell/Python/Kotlin/Swift files
plus project-owned Markdown skill entrypoints. Shared Agent-Ops common rules
and skills are outside this repository readability policy. Produces sorted JSON
and optional ratchet (baseline comparison) via --check.
Usage:
python3 scripts/readability_audit.py [--check] [--output PATH] [--input-mode tracked|worktree]
Exit codes:
0 audit ran; with --check no new/increased violations
2 invalid CLI input or invalid read-set configuration
3 --check requested but baseline is missing/unreadable
4 --check found new or increased violations
"""
from __future__ import annotations
import argparse
import ast as py_ast
import io
import json
import os
import re
import subprocess
import sys
import tokenize
from dataclasses import dataclass, field
from typing import Any
# ---------- constants ----------
# Languages we instrument.
TRACKED_EXTENSIONS: dict[str, str] = {
".go": "go",
".dart": "dart",
".sh": "shell",
".py": "python",
".kt": "kotlin",
".swift": "swift",
".md": "markdown",
}
# Directories to exclude from audit (prefix match against repo-relative paths).
EXCLUSION_PATTERNS: tuple[str, ...] = (
"proto/gen/",
"apps/client/lib/gen/",
"build/vendor/",
"build/data/",
"agent-ops/rules/common/",
"agent-ops/skills/common/",
".git/",
)
# ---------- policy / threshold model ----------
# file-level policy by file kind
FILE_POLICY: dict[str, dict[str, int]] = {
"production": {"warning": 500, "split_review": 800, "exception": 1000},
"test": {"warning": 800, "split_review": 1000},
"skill_entrypoint": {"warning": 300, "split_review": 500},
}
# function-level policy
FUNCTION_POLICY: dict[str, int] = {
"warning": 80,
"split_review": 120,
}
# Severity ordering shared by file and function levels.
SEVERITY_RANK: dict[str, int] = {"warning": 1, "split_review": 2, "exception": 3}
FILE_LEVELS: tuple[str, ...] = ("warning", "split_review", "exception")
FUNCTION_LEVELS: tuple[str, ...] = ("warning", "split_review")
# Test-file naming conventions per language extension.
TEST_BASENAME_PATTERNS: dict[str, tuple[str, ...]] = {
".go": (r".*_test\.go",),
".dart": (r".*_test\.dart",),
".py": (r".*_test\.py", r"test_.*\.py"),
".kt": (r".*Tests?\.kt", r".*_test\.kt"),
".swift": (r".*Tests?\.swift", r".*_test\.swift"),
".sh": (r".*_test\.sh", r"test_.*\.sh"),
}
# Directory components that mark a test tree in any language.
TEST_PATH_COMPONENTS: frozenset[str] = frozenset({"test", "tests"})
# Repository test entrypoints that do not follow a language naming convention.
# `scripts/e2e-*.sh` is the testing-domain E2E/smoke surface (see the domain map
# in agent-ops/rules/project/rules.md), so it carries the test LOC policy.
TEST_PATH_PATTERNS: tuple[str, ...] = (
r"scripts/e2e-[\w.-]+\.sh",
)
def classify_file(path: str) -> str:
"""Return file kind: 'production', 'test', or 'skill_entrypoint'."""
if path.startswith("agent-ops/skills/") and path.endswith("/SKILL.md"):
return "skill_entrypoint"
for pattern in TEST_PATH_PATTERNS:
if re.fullmatch(pattern, path):
return "test"
segments = path.split("/")
if any(segment in TEST_PATH_COMPONENTS for segment in segments[:-1]):
return "test"
basename = segments[-1]
_, ext = os.path.splitext(basename)
for pattern in TEST_BASENAME_PATTERNS.get(ext, ()):
if re.fullmatch(pattern, basename):
return "test"
return "production"
# ---------- lexical helpers ----------
# Languages whose bodies are delimited by braces and share C-style comments.
BRACE_LANGUAGES: tuple[str, ...] = ("go", "dart", "kotlin", "swift")
# Languages whose block comments nest (`/* /* */ */`). Go's do not.
NESTED_COMMENT_LANGUAGES: frozenset[str] = frozenset({"dart", "kotlin", "swift"})
# Multiline (raw) string delimiters that must not terminate at a newline.
MULTILINE_QUOTES: dict[str, tuple[str, ...]] = {
"dart": ("'''", '"""'),
"kotlin": ('"""',),
"swift": ('"""',),
}
HEREDOC_RE = re.compile(
r"<<(?P<dash>-?)\s*(?P<quote>['\"]?)(?P<delim>[A-Za-z_][A-Za-z0-9_]*)(?P=quote)"
)
# Characters that end a shell word: an unquoted `#` is a comment only when the
# previous significant character is one of these (or start of input). A `#`
# inside a word (`foo#bar`) is an ordinary token.
SHELL_WORD_DELIMITERS: frozenset[str] = frozenset(" \t\r\n;&|()<>")
def _strip_comments_and_strings(source: str, lang: str) -> str:
"""Return source with string literals and comments replaced by spaces.
Newlines are always preserved so that line numbers and LOC stay aligned
with the original source. Markdown has no functions to extract.
"""
if lang == "markdown":
return source
if lang == "python":
return _blank_python_strings_and_comments(source)
return _blank_lex_strings_and_comments(source, lang)
def _blank_python_strings_and_comments(source: str) -> str:
"""Blank Python string literals and comments via tokenize, keeping newlines."""
out = list(source)
offsets: list[int] = []
acc = 0
for line in source.splitlines(keepends=True):
offsets.append(acc)
acc += len(line)
def offset(row: int, col: int) -> int:
if row - 1 >= len(offsets):
return len(source)
return offsets[row - 1] + col
try:
tokens = list(tokenize.generate_tokens(io.StringIO(source).readline))
except (tokenize.TokenError, IndentationError, SyntaxError):
return source
for token in tokens:
if token.type not in (tokenize.STRING, tokenize.COMMENT):
continue
start = offset(*token.start)
end = offset(*token.end)
for k in range(start, min(end, len(out))):
if out[k] != "\n":
out[k] = " "
return "".join(out)
def _blank_lex_strings_and_comments(source: str, lang: str) -> str:
"""Blank strings/raw literals/comments for Go/Dart/Shell/Swift/Kotlin."""
if lang == "shell":
return _blank_shell_strings_and_comments(source)
out = list(source)
n = len(source)
def blank(start: int, end: int) -> None:
for k in range(max(start, 0), min(end, n)):
if out[k] != "\n":
out[k] = " "
i = 0
while i < n:
ch = source[i]
if lang in BRACE_LANGUAGES and source.startswith("//", i):
i = _blank_to_line_end(source, i, blank, n)
continue
if lang in BRACE_LANGUAGES and source.startswith("/*", i):
i = _blank_block_comment(source, i, blank, n, lang in NESTED_COMMENT_LANGUAGES)
continue
if lang == "go" and ch == "`":
end = source.find("`", i + 1)
end = n if end < 0 else end + 1
blank(i, end)
i = end
continue
triple = _match_triple_quote(source, i, lang)
if triple:
end = source.find(triple, i + len(triple))
end = n if end < 0 else end + len(triple)
blank(i, end)
i = end
continue
if ch in ('"', "'"):
i = _blank_single_line_string(source, i, ch, blank, n)
continue
i += 1
return "".join(out)
def _blank_shell_strings_and_comments(source: str) -> str:
"""Blank shell strings, heredoc bodies, `${...}` expansions, and comments.
A `#` is a comment token only at the start of a word -- at input start or
after unquoted whitespace/an operator -- and never inside `${...}`, so
`foo#bar` and `${value#x}` keep everything after their `#` and a same-line
function close survives. Escaped braces, quoted strings, and heredoc
bodies blank exactly as in the shared lexer before the split.
"""
out = list(source)
n = len(source)
def blank(start: int, end: int) -> None:
for k in range(max(start, 0), min(end, n)):
if out[k] != "\n":
out[k] = " "
in_word = False
i = 0
while i < n:
ch = source[i]
if ch == "#" and not in_word:
i = _blank_to_line_end(source, i, blank, n)
continue
if source.startswith("${", i):
i = _blank_shell_expansion(source, i, blank, n)
in_word = True
continue
if source.startswith("<<", i):
match = HEREDOC_RE.match(source, i)
if match:
i = _blank_heredoc_body(source, match, blank, n)
in_word = False
continue
# An escaped character outside a string never opens a brace (\\}); an
# escaped newline joins lines, so the word state carries over.
if ch == "\\" and i + 1 < n:
if source[i + 1] in "{}":
blank(i, i + 2)
if source[i + 1] != "\n":
in_word = True
i += 2
continue
if ch in ('"', "'"):
i = _blank_shell_string(source, i, ch, blank, n)
in_word = True
continue
in_word = ch not in SHELL_WORD_DELIMITERS
i += 1
return "".join(out)
def _blank_to_line_end(source: str, start: int, blank: Any, n: int) -> int:
end = source.find("\n", start)
end = n if end < 0 else end
blank(start, end)
return end
def _blank_block_comment(source: str, start: int, blank: Any, n: int, nested: bool) -> int:
"""Blank a /* */ comment, tracking depth for languages whose comments nest."""
depth = 1
j = start + 2
while j < n:
if nested and source.startswith("/*", j):
depth += 1
j += 2
continue
if source.startswith("*/", j):
depth -= 1
j += 2
if depth == 0:
break
continue
j += 1
end = min(j, n)
blank(start, end)
return end
def _match_triple_quote(source: str, i: int, lang: str) -> str | None:
for quote in MULTILINE_QUOTES.get(lang, ()):
if source.startswith(quote, i):
return quote
return None
def _blank_single_line_string(source: str, start: int, quote: str, blank: Any, n: int) -> int:
j = start + 1
while j < n:
ch = source[j]
if ch == "\\" and j + 1 < n:
j += 2
continue
if ch == quote:
j += 1
break
if ch == "\n":
break
j += 1
blank(start, j)
return j
def _blank_shell_string(source: str, start: int, quote: str, blank: Any, n: int) -> int:
"""Blank a shell quoted string, which stays open across newlines.
Only double quotes honour backslash escapes; inside single quotes a
backslash is literal. Newlines inside the literal are preserved by blank(),
so awk bodies and other quoted braces never reach the brace scanner.
"""
j = start + 1
while j < n:
ch = source[j]
if quote == '"' and ch == "\\" and j + 1 < n:
j += 2
continue
if ch == quote:
j += 1
break
j += 1
blank(start, j)
return j
def _blank_shell_expansion(source: str, start: int, blank: Any, n: int) -> int:
"""Blank an unquoted `${...}` parameter expansion, including its braces.
Expansion-internal braces, `#`/`%` operators, and quoted segments belong
to the expansion, not to the surrounding function body, so the whole
region is blanked: its braces never reach the brace scanner and its `#`
never reads as a comment. Nested `${...}` stays balanced via brace depth;
an unterminated expansion blanks to end of input like other unterminated
literals.
"""
depth = 1
j = start + 2
while j < n and depth > 0:
ch = source[j]
if ch == "\\" and j + 1 < n:
j += 2
continue
if ch == "'":
close = source.find("'", j + 1)
j = n if close < 0 else close + 1
continue
if ch == '"':
j += 1
while j < n:
if source[j] == "\\" and j + 1 < n:
j += 2
continue
if source[j] == '"':
j += 1
break
j += 1
continue
if ch == "{":
depth += 1
elif ch == "}":
depth -= 1
j += 1
blank(start, j)
return j
def _blank_heredoc_body(source: str, match: re.Match[str], blank: Any, n: int) -> int:
"""Blank a shell here-doc body up to (not including) its delimiter line."""
delim = match.group("delim")
allow_indent = match.group("dash") == "-"
body_start = source.find("\n", match.end())
if body_start < 0:
return n
body_start += 1
pos = body_start
while pos < n:
line_end = source.find("\n", pos)
line_end = n if line_end < 0 else line_end
line = source[pos:line_end].rstrip("\r")
if (line.lstrip("\t") if allow_indent else line) == delim:
blank(body_start, pos)
return line_end
pos = line_end + 1
blank(body_start, n)
return n
# ---------- function extraction ----------
# Declarations that open a named lexical scope (used to qualify identities).
TYPE_DECL_RE: dict[str, re.Pattern[str]] = {
"dart": re.compile(
r"^\s*(?:(?:abstract|final|base|interface|sealed)\s+)*"
r"(?:class|mixin|extension|enum)\s+(?P<name>\w+)"
),
"kotlin": re.compile(
r"^\s*(?:(?:public|private|internal|protected|open|sealed|abstract|final|data"
r"|inner|value|annotation|enum|companion)\s+)*"
r"(?:class|object|interface)(?:\s+(?P<name>\w+))?"
),
"swift": re.compile(
r"^\s*(?:(?:public|private|internal|fileprivate|open|final)\s+)*"
r"(?:class|struct|enum|protocol|extension|actor)\s+(?P<name>\w+)"
),
}
# Actual function declarations only: Kotlin/Swift properties are not functions.
FUNC_DECL_RE: dict[str, re.Pattern[str]] = {
"go": re.compile(r"^\s*func\s+(?:\((?P<recv>[^)]*)\)\s*)?(?P<name>\w+)\s*[\(\[]"),
"dart": re.compile(
r"^\s*(?:void|int|bool|double|num|String|dynamic|[A-Z]\w*(?:<[^>]+>)?)"
r"\s+(?P<name>\w+)\s*\("
),
"kotlin": re.compile(
r"^\s*(?:(?:public|private|internal|protected|open|override|suspend|inline"
r"|operator|infix|tailrec|external|final|abstract|expect|actual)\s+)*"
r"fun\s+(?:<[^>]*>\s*)?(?:(?P<recv>[\w.]+)\.)?(?P<name>\w+)\s*\("
),
"swift": re.compile(
r"^\s*(?:(?:public|private|internal|fileprivate|open|static|class|final"
r"|override|mutating|nonmutating|convenience|required)\s+)*"
r"func\s+(?P<name>\w+)\s*[<\(]"
),
# The lookahead keeps the match ending on the opening paren, so `match.end()
# - 1` is the declaration's own parameter list like every other language.
"shell": re.compile(r"^\s*(?:function\s+)?(?P<name>[\w:.-]+)\s*\((?=\s*\)\s*\{?\s*$)"),
}
# A declaration header longer than this is not treated as a function body opener.
MAX_DECL_HEADER_LINES = 40
# Names that are a type keyword rather than a callable identity. A Dart
# `void Function(...) cb` field/parameter matches the return-type form but
# declares no callable.
NON_CALLABLE_NAMES: dict[str, frozenset[str]] = {
"dart": frozenset({"Function"}),
}
# Token that introduces an expression body once the header parens are closed.
EXPRESSION_BODY_TOKENS: dict[str, str] = {
"dart": "=>",
"kotlin": "=",
}
# Kotlin joins a depth-0 expression across lines only in the shapes its
# grammar allows: a line ending in an operator token keeps the expression
# open, and the next nonblank line may begin with an operator the grammar
# writes as `NL* operator`. Dart has no such rule (its expression bodies
# terminate with `;`), so these stay Kotlin-only.
# Trailing operator tokens by Kotlin grammar category. Matching any token
# keeps the expression open; `..`/`..<` keep their own range category even
# where a shorter token overlaps, so each grammar rule stays visible.
KOTLIN_SUFFIX_TOKENS_BY_CATEGORY: dict[str, tuple[str, ...]] = {
"equality": ("===", "!==", "==", "!="),
"comparison": ("<=", ">="),
"logical": ("&&", "||"),
"elvis": ("?:",),
"range": ("..<", ".."),
"additive": ("+", "-"),
"multiplicative": ("*", "/", "%"),
"member_access": ("?.", "."),
"arrow": ("->",),
}
KOTLIN_SUFFIX_TOKENS: tuple[str, ...] = tuple(
token
for tokens in KOTLIN_SUFFIX_TOKENS_BY_CATEGORY.values()
for token in tokens
)
# A bare `<`/`>` ending a depth-0 expression line is a comparison operator
# awaiting its RHS regardless of lexical spacing (`1 <` and `1<` parse
# alike) -- except when the `>` closes the balanced generic argument list
# of a cast/type-test RHS. `as`/`is` take a type, not an expression, so
# `value as List<Int>` is a complete expression at its closing `>` while
# `value>` still awaits its comparison RHS.
KOTLIN_BARE_COMPARISON_RE: re.Pattern[str] = re.compile(r"[<>]$")
# The cast/type-test operator that puts a trailing generic argument list in
# type position. `as?` has its own branch because its punctuation token can
# touch the type name, while `as` and `is` retain whitespace/word boundaries
# so identifier tails such as `asList` and `isReady` never become operators.
KOTLIN_TYPE_OPERAND_RE: re.Pattern[str] = re.compile(
r"(?:"
r"(?<![\w`])(?:as|is)\s+"
r"|(?<![\w`])as\?\s*"
r"|!is\s+"
r")"
)
# A Kotlin annotation user type and a regular type head both permit qualified
# or escaped identifiers. Annotation constructor arguments need a bounded
# scanner because their nesting is independent from an outer type projection.
KOTLIN_USER_TYPE_PART_RE: re.Pattern[str] = re.compile(
r"(?:`[^`\n]+`|[A-Za-z_]\w*)"
)
KOTLIN_ANNOTATION_USER_TYPE_RE: re.Pattern[str] = re.compile(
r"@(?:`[^`\n]+`|[A-Za-z_]\w*)"
r"(?:\.(?:`[^`\n]+`|[A-Za-z_]\w*))*\s*$"
)
def _consume_kotlin_user_type(text: str, start: int) -> int | None:
"""Return the end of a qualified Kotlin user type, if one starts here."""
match = KOTLIN_USER_TYPE_PART_RE.match(text, start)
if match is None:
return None
index = match.end()
while index < len(text) and text[index] == ".":
part = KOTLIN_USER_TYPE_PART_RE.match(text, index + 1)
if part is None:
return None
index = part.end()
return index
def _consume_kotlin_annotation_arguments(text: str, start: int) -> int | None:
"""Return the end of balanced annotation constructor arguments."""
if start >= len(text) or text[start] != "(":
return None
depth = 0
for index in range(start, len(text)):
if text[index] == "(":
depth += 1
elif text[index] == ")":
depth -= 1
if depth == 0:
return index + 1
return None
def _consume_kotlin_type_annotations(text: str, start: int) -> tuple[int, bool] | None:
"""Return the end of type annotations and whether their type head is pending."""
index = start
while index < len(text) and text[index] == "@":
annotation_end = _consume_kotlin_user_type(text, index + 1)
if annotation_end is None:
return None
index = annotation_end
separator_start = index
while index < len(text) and text[index].isspace():
index += 1
if index < len(text) and text[index] == "(":
argument_end = _consume_kotlin_annotation_arguments(text, index)
if argument_end is None:
return None
index = argument_end
separator_start = index
while index < len(text) and text[index].isspace():
index += 1
if index == separator_start:
return (index, True) if index == len(text) else None
return index, index == len(text)
def _kotlin_type_operand_before_generic(prefix: str) -> bool:
"""Report whether ``prefix`` ends in a cast/type-test type head.
The bounded parser admits qualified annotation user types and constructor
invocations before the actual type head. It deliberately does not parse a
complete Kotlin type: it only establishes the type context needed when an
immediately following ``<`` might otherwise be a comparison operator.
"""
for match in reversed(list(KOTLIN_TYPE_OPERAND_RE.finditer(prefix))):
index = match.end()
while index < len(prefix) and prefix[index].isspace():
index += 1
annotations = _consume_kotlin_type_annotations(prefix, index)
if annotations is None:
continue
index, pending_type_head = annotations
if pending_type_head:
continue
type_end = _consume_kotlin_user_type(prefix, index)
if type_end is not None and not prefix[type_end:].strip():
return True
return False
def _kotlin_type_annotation_awaits_head(prefix: str) -> bool:
"""Report whether a cast/type-test line ends after type annotations."""
for match in reversed(list(KOTLIN_TYPE_OPERAND_RE.finditer(prefix))):
index = match.end()
while index < len(prefix) and prefix[index].isspace():
index += 1
if index == len(prefix) or prefix[index] != "@":
continue
annotations = _consume_kotlin_type_annotations(prefix, index)
if annotations is not None and annotations[1]:
return True
return False
@dataclass
class _KotlinTypeArgumentState:
"""Track a cast/type-test generic argument list across source lines.
This is intentionally a bounded expression extractor, not a Kotlin
parser. Once a type operator leads into `<...>`, every nested angle pair
is counted until the outer closer. The `>` in a function-type `->` is not
an angle closer. Keeping the consumed source also lets a multiline
`as\nList<...>` test its type operand at the opener rather than treating
each line as an isolated comparison.
"""
prefix: str = ""
angle_depth: int = 0
trailing_closer: bool = False
annotation_paren_depth: int = 0
annotation_bracket_depth: int = 0
@property
def awaits_annotated_type_head(self) -> bool:
"""Keep a multiline cast/type test open after its final annotation."""
return _kotlin_type_annotation_awaits_head(self.prefix)
def consume(self, text: str) -> None:
"""Advance state through one expression slice and retain its context."""
self.trailing_closer = False
for ch in text:
if self.annotation_paren_depth > 0:
if ch == "(":
self.annotation_paren_depth += 1
elif ch == ")":
self.annotation_paren_depth -= 1
elif ch == "[":
self.annotation_bracket_depth += 1
elif ch == "]" and self.annotation_bracket_depth > 0:
self.annotation_bracket_depth -= 1
elif ch == "(" and KOTLIN_ANNOTATION_USER_TYPE_RE.search(self.prefix):
self.annotation_paren_depth = 1
self.annotation_bracket_depth = 0
elif self.angle_depth > 0:
if ch == ">" and not self.prefix.endswith("-"):
self.angle_depth -= 1
if self.angle_depth == 0:
self.trailing_closer = True
elif ch == "<":
self.angle_depth += 1
elif self.trailing_closer and not ch.isspace():
self.trailing_closer = False
elif ch == "<" and _kotlin_type_operand_before_generic(self.prefix):
self.angle_depth = 1
elif self.trailing_closer and not ch.isspace():
self.trailing_closer = False
self.prefix += ch
# Containment (`in`/`!in`), type test (`is`/`!is`), and cast (`as`/`as?`)
# take their RHS after an optional newline. The left guard keeps identifier
# tails (`login`, `Main`) from reading as word operators.
KOTLIN_WORD_OPERATOR_RE: re.Pattern[str] = re.compile(
r"(?:^|[^\w])(?:!in|!is|as\?|as|in|is)$"
)
# A trailing simple identifier is an infix function call awaiting its RHS
# (`1 join`) only when a left operand precedes it. Keywords are never infix
# identifiers, and a keyword-led tail (`else x`, `as String`) or a lone
# value has no left operand, so those end the expression.
KOTLIN_TRAILING_IDENTIFIER_RE: re.Pattern[str] = re.compile(r"[A-Za-z_]\w*$")
KOTLIN_OPERAND_END_RE: re.Pattern[str] = re.compile(r"[\w)\]]$")
KOTLIN_NON_INFIX_WORDS: frozenset[str] = frozenset({
"as", "break", "continue", "do", "else", "false", "fun", "if", "in",
"is", "null", "object", "return", "super", "this", "throw", "true",
"try", "val", "var", "when", "while",
})
KOTLIN_NON_OPERAND_WORDS: frozenset[str] = frozenset({
"as", "break", "continue", "do", "else", "fun", "if", "import", "in",
"is", "object", "package", "return", "throw", "try", "typealias",
"val", "var", "when", "while",
})
# Leading operator tokens the grammar writes as `NL* operator`: member
# access (`.`/`?.`), elvis (`?:`), logical (`&&`/`||`), and cast (`as`/
# `as?`). A following declaration, annotation, or type close never matches,
# so it ends the expression instead of continuing it.
KOTLIN_EXPRESSION_PREFIX_RE: re.Pattern[str] = re.compile(
r"^(?:\?\.|\?:|\.|&&|\|\||as\b)"
)
def _detect_language(filepath: str) -> str:
_, ext = os.path.splitext(filepath)
return TRACKED_EXTENSIONS.get(ext, "")
def _go_receiver_type(receiver: str) -> str:
"""Extract the receiver type name from a Go receiver clause."""
cleaned = receiver.strip()
if not cleaned:
return ""
named = re.match(r"^\w+\s+\*?(\w+)", cleaned)
if named:
return named.group(1)
bare = re.match(r"^\*?(\w+)", cleaned)
return bare.group(1) if bare else ""
def _qualified_function_name(lang: str, match: re.Match[str], scopes: list[tuple[str, int]]) -> str:
"""Build a scope-qualified identity so same-name callables stay distinct."""
groups = match.groupdict()
name = groups.get("name") or "unknown"
if lang == "go":
receiver = _go_receiver_type(groups.get("recv") or "")
return f"{receiver}.{name}" if receiver else name
parts = [scope for scope, _ in scopes]
receiver = groups.get("recv")
if receiver:
parts.append(receiver)
parts.append(name)
return ".".join(parts)
# Go keywords whose `{` opens a type literal rather than a function body.
GO_TYPE_LITERAL_KEYWORDS: frozenset[str] = frozenset({"struct", "interface"})
GO_TRAILING_WORD_RE = re.compile(r"(\w+)$")
def _is_go_type_literal_brace(line: str, index: int) -> bool:
"""Report whether the `{` at `index` belongs to a struct/interface literal.
gofmt keeps the keyword on the same line as its brace, so the preceding
token is read line-locally.
"""
word = GO_TRAILING_WORD_RE.search(line[:index].rstrip())
return bool(word) and word.group(1) in GO_TYPE_LITERAL_KEYWORDS
def _tail_body_brace(lines: list[str], j: int, tail_col: int, lang: str) -> tuple[int, int]:
"""Locate the brace opening the executable body at/after `(j, tail_col)`.
Paren/bracket depth is balanced so a parenthesized result type contributes
no brace, and a Go `struct`/`interface` result type is balanced away as a
type literal. Scanning stops at the end of the first line that leaves every
depth balanced: a completed header can only be followed by the next line's
leading brace, which the caller handles. Returns (-1, -1) when no body
brace opens here.
"""
paren = 0
bracket = 0
type_depth = 0
limit = min(len(lines), j + MAX_DECL_HEADER_LINES)
for row in range(j, limit):
line = lines[row]
for index in range(tail_col if row == j else 0, len(line)):
ch = line[index]
if ch == "(":
paren += 1
elif ch == ")":
paren -= 1
elif ch == "[":
bracket += 1
elif ch == "]":
bracket -= 1
elif ch == "{":
if type_depth > 0:
type_depth += 1
elif paren <= 0 and bracket <= 0:
if lang == "go" and _is_go_type_literal_brace(line, index):
type_depth = 1
else:
return (row, index)
elif ch == "}" and type_depth > 0:
type_depth -= 1
if paren <= 0 and bracket <= 0 and type_depth <= 0:
return (-1, -1)
return (-1, -1)
def _classify_decl_tail(lines: list[str], j: int, tail_col: int, lang: str) -> tuple[str, int, int]:
"""Classify what follows a completed declaration header at `(j, tail_col)`.
Returns (kind, line_idx, col) where col is the body brace itself or the
offset just past the expression token, so the caller's end scan never
re-reads the declaration.
"""
row, col = _tail_body_brace(lines, j, tail_col, lang)
if row >= 0:
return ("brace", row, col)
token = EXPRESSION_BODY_TOKENS.get(lang)
tail = lines[j][tail_col:]
if token and token in tail:
return ("expression", j, tail_col + tail.index(token) + len(token))
following = lines[j + 1] if j + 1 < len(lines) else ""
stripped = following.strip()
if stripped.startswith("{"):
return ("brace", j + 1, following.index("{"))
if token and stripped.startswith(token):
return ("expression", j + 1, following.index(token) + len(token))
return ("none", -1, -1)
def _find_decl_body(lines: list[str], start: int, col: int, lang: str) -> tuple[str, int, int]:
"""Classify the body that the declaration at `start` opens.
Scanning begins at `col`, the declaration's own parameter list, so a Go
receiver clause never reads as the header. Returns ('brace', idx, col) for
a brace body, ('expression', idx, col) for a `=>`/`=` body, and
('none', -1, -1) when the match has no body at all. A body-less match is a
parameter, an abstract/protocol declaration, or a call expression -- never
a callable we can measure, and the header may span lines before its body
opener.
Go generic type parameters (square brackets) and anonymous result types
(parenthesized or struct/interface literals) are tracked separately so
their braces never read as the function body opener.
"""
paren = 0
bracket = 0
opened = False
limit = min(len(lines), start + MAX_DECL_HEADER_LINES)
for j in range(start, limit):
base = col if j == start else 0
line = lines[j]
for index in range(base, len(line)):
ch = line[index]
if ch == "[":
bracket += 1
elif ch == "]":
bracket -= 1
elif ch == "(":
paren += 1
opened = True
elif ch == ")":
paren -= 1
if opened and paren <= 0:
return _classify_decl_tail(lines, j, index + 1, lang)
elif ch == "{" and paren <= 0 and bracket <= 0 and opened:
return ("brace", j, index)
if not opened:
return _classify_decl_tail(lines, j, len(line), lang)
return ("none", -1, -1)
def _find_type_body(lines: list[str], start: int, col: int) -> int:
"""Return the line index opening a type body, or -1 when the type has none.
A type header may span lines (supertype clauses, multiline constructor
parameters), so the scan runs to the real body opener. It stops at `;`
(a body-less type alias/declaration) and at a blank line, which no type
header crosses -- otherwise a body-less type would adopt the next
declaration's brace as its own.
"""
paren = 0
limit = min(len(lines), start + MAX_DECL_HEADER_LINES)
for j in range(start, limit):
line = lines[j][col:] if j == start else lines[j]
if j > start and not line.strip():
return -1
for ch in line:
if ch in "([":
paren += 1
elif ch in ")]":
paren -= 1
elif ch == "{" and paren <= 0:
return j
elif ch == ";" and paren <= 0:
return -1
return -1
def _kotlin_trailing_infix_identifier(text: str, raw_text: str) -> bool:
"""Report whether `text` ends with an infix-call identifier awaiting RHS.
The grammar keeps an infix function call open across a newline between
its identifier and RHS, so `1 join` continues while a lone value, a
chained member (`.size`), or a keyword-led tail (`as String`) does not.
Identifier and operator tokens read from the cleaned `text`; `raw_text`
is the length-aligned raw slice of the same line, consulted only to
recognize a string/character literal left operand (`"x" then`) that
cleaning blanked away.
"""
match = KOTLIN_TRAILING_IDENTIFIER_RE.search(text)
if match is None or match.group() in KOTLIN_NON_INFIX_WORDS:
return False
if raw_text[: match.start()].rstrip().endswith(('"', "'")):
return True
operand = text[: match.start()].rstrip()
if not operand or not KOTLIN_OPERAND_END_RE.search(operand):
return False
word = KOTLIN_TRAILING_IDENTIFIER_RE.search(operand)
return word is None or word.group() not in KOTLIN_NON_OPERAND_WORDS
def _kotlin_trailing_generic_type_closer(text: str) -> bool:
"""Report whether a trailing `>` closes a cast/type-test generic type.
A generic type may contain function, parenthesized, annotated,
definitely-non-null, or escaped-identifier projections, so a character
allowlist is not enough. The token-aware state tracks balanced angle
pairs and ignores the arrow in function types. An extra `>` after the
outer closer remains a comparison continuation.
"""
if not text.endswith(">"):
return False
state = _KotlinTypeArgumentState()
state.consume(text)
return state.angle_depth == 0 and state.trailing_closer
def _kotlin_trailing_continuation(content: str, raw_content: str,
trailing_type_closer: bool = False) -> bool:
"""Report whether a depth-0 Kotlin line ends inside an open expression.
Each check is one grammar category family: symbolic operator tokens,
spacing-independent bare comparison, word operators (containment/type
test/cast), and custom infix-call identifiers. `raw_content` is the
length-aligned raw slice of the same line for literal operand evidence.
"""
text = content.rstrip()
if not text:
return False
if text.endswith(KOTLIN_SUFFIX_TOKENS):
return True
if (KOTLIN_BARE_COMPARISON_RE.search(text)
and not (trailing_type_closer
or _kotlin_trailing_generic_type_closer(text))):
return True
if KOTLIN_WORD_OPERATOR_RE.search(text):
return True
return _kotlin_trailing_infix_identifier(text, raw_content)
def _kotlin_expression_continues(lines: list[str], j: int, content: str,
raw_content: str,
trailing_type_closer: bool = False) -> bool:
"""Report whether a depth-0 Kotlin expression line continues past line j.
A trailing operator or infix identifier keeps the expression open, and a
next nonblank line beginning with a newline-before operator token
(member access, elvis, logical, cast) belongs to the same expression.
Anything else -- the next declaration or a type's closing brace -- ends
it, so sibling members are never swallowed. A pending type annotation is
handled by the caller because its following type head is a continuation.
`raw_content` aligns with `content` for literal operand evidence.
"""
if _kotlin_trailing_continuation(
content, raw_content, trailing_type_closer):
return True
for row in range(j + 1, len(lines)):
following = lines[row].strip()
if not following:
continue
return bool(KOTLIN_EXPRESSION_PREFIX_RE.match(following))
return False
def _find_expression_end(lines: list[str], body_line_idx: int, body_col: int = 0,
lang: str = "", raw_lines: list[str] | None = None) -> int:
"""Find the line index closing an expression body opened at body_line_idx.
Scanning begins at `body_col`, just past the expression token, so the
declaration's own parameter list never reads as a balanced expression and
collapses the body to loc=1. Depth follows bracket/paren/brace through the
expression; a `;` or an unmatched closing delimiter at depth 0 ends it. An
expression token with nothing after it on its own line continues onto the
following lines, which is how multiline Kotlin/Dart bodies keep their LOC.
A depth-0 line break ends the expression unless the language joins it: in
Kotlin a trailing operator-grammar token (symbolic, word, or infix
identifier) or a newline-before operator leading the next nonblank line
keeps a depth-0 expression open. Dart keeps its `;` termination contract
untouched.
`raw_lines`, when given, must align with `lines` char-for-char (cleaning
blanks in place); it supplies the raw slice Kotlin needs to see a string
literal operand that cleaning erased. Without it the cleaned lines stand
in for both.
"""
raw = lines if raw_lines is None else raw_lines
depth = 0
kotlin_type_arguments = _KotlinTypeArgumentState() if lang == "kotlin" else None
for j in range(body_line_idx, len(lines)):
line = lines[j]
begin = body_col if j == body_line_idx else 0
for index in range(begin, len(line)):
ch = line[index]
if ch in "([{":
depth += 1
elif ch in ")]}":
depth -= 1
if depth < 0:
return j
elif ch == ";" and depth <= 0:
return j
if j == body_line_idx and not line[begin:].strip():
continue
if kotlin_type_arguments is not None:
content = line[begin:]
if j != body_line_idx:
content = "\n" + content
kotlin_type_arguments.consume(content)
if depth <= 0:
if kotlin_type_arguments is not None and kotlin_type_arguments.angle_depth > 0:
continue
if lang == "kotlin" and kotlin_type_arguments is not None and (
kotlin_type_arguments.awaits_annotated_type_head
or _kotlin_expression_continues(
lines, j, line[begin:], raw[j][begin:],
kotlin_type_arguments.trailing_closer)):
continue
return j
return len(lines) - 1
def _find_func_end_brace(cleaned_source: str, body_line_idx: int, body_col: int = 0) -> int:
"""Find the line index closing the body opened at (body_line_idx, body_col).
cleaned_source must already have strings/comments blanked, so a brace here
is always a real brace. Scanning starts at `body_col`, the body opener
itself, so a type-literal brace earlier on the declaration line never
closes the function.
"""
lines = cleaned_source.splitlines()
depth = 0
started = False
for j in range(body_line_idx, len(lines)):
line = lines[j]
for index in range(body_col if j == body_line_idx else 0, len(line)):
ch = line[index]
if ch == "{":
depth += 1
started = True
elif ch == "}":
depth -= 1
if started and depth <= 0:
return j
return len(lines) - 1
def _normalize_signature(parts: list[str]) -> str:
return "(" + re.sub(r"\s+", " ", "".join(parts)).strip() + ")"
def _declaration_signature(lines: list[str], start: int, col: int) -> str:
"""Return the normalized parameter list of the declaration at `start`.
Scanning starts at the declaration's own parameter list, so a Go receiver
clause or a generic parameter list never leaks into the signature.
"""
limit = min(len(lines), start + MAX_DECL_HEADER_LINES)
depth = 0
opened = False
parts: list[str] = []
for j in range(start, limit):
line = lines[j][col:] if j == start else lines[j]
for ch in line:
if ch == "(":
depth += 1
opened = True
if depth == 1:
continue
elif ch == ")":
depth -= 1
if opened and depth == 0:
return _normalize_signature(parts)
if opened and depth >= 1:
parts.append(ch)
if opened:
parts.append(" ")
return _normalize_signature(parts)
def _group_by_name(functions: list[dict[str, Any]]) -> dict[str, list[dict[str, Any]]]:
groups: dict[str, list[dict[str, Any]]] = {}
for fn in functions:
groups.setdefault(fn["name"], []).append(fn)
return groups
def _finalize_functions(functions: list[dict[str, Any]]) -> list[dict[str, Any]]:
"""Resolve identity collisions deterministically, then drop internal keys.
A unique callable keeps its plain scope.name identity. Only overload groups
that actually collide take a normalized signature, and only signatures that
still tie take a positional ordinal, so every reported identity is unique
without renaming callables that were already distinct.
"""
for name, group in _group_by_name(functions).items():
if len(group) > 1:
for fn in group:
fn["name"] = f"{name}{fn['signature']}"
for name, group in _group_by_name(functions).items():
if len(group) > 1:
ordered = sorted(group, key=lambda f: (f["start"], f["end"]))
for ordinal, fn in enumerate(ordered, start=1):
fn["name"] = f"{name}#{ordinal}"
for fn in functions:
fn.pop("signature", None)
return sorted(functions, key=lambda f: (-f["loc"], f["name"]))
def _extract_functions_brace(source: str, lang: str) -> list[dict[str, Any]]:
"""Extract functions for brace-delimited languages from cleaned source."""
cleaned = _strip_comments_and_strings(source, lang)
lines = cleaned.splitlines()
raw_lines = source.splitlines()
func_re = FUNC_DECL_RE[lang]
type_re = TYPE_DECL_RE.get(lang)
reserved = NON_CALLABLE_NAMES.get(lang, frozenset())
out: list[dict[str, Any]] = []
scopes: list[tuple[str, int]] = []
scope_openers: dict[int, str] = {}
depth = 0
i = 0
while i < len(lines):
line = lines[i]
opener = scope_openers.pop(i, None)
if opener is not None:
scopes.append((opener, depth))
match = func_re.match(line)
if match and (match.groupdict().get("name") or "") not in reserved:
kind, body, body_col = _find_decl_body(lines, i, match.end() - 1, lang)
if kind != "none":
end = (_find_func_end_brace(cleaned, body, body_col) if kind == "brace"
else _find_expression_end(lines, body, body_col, lang, raw_lines))
out.append({
"name": _qualified_function_name(lang, match, scopes),
"loc": end - i + 1,
"start": i + 1,
"end": end + 1,
"signature": _declaration_signature(lines, i, match.end() - 1),
})
i = end + 1
continue
if type_re is not None:
type_match = type_re.match(line)
if type_match:
body = _find_type_body(lines, i, type_match.end())
if body >= 0:
name = type_match.group("name") or "companion"
if body == i:
scopes.append((name, depth))
else:
scope_openers[body] = name
depth += line.count("{") - line.count("}")
while scopes and depth <= scopes[-1][1]:
scopes.pop()
i += 1
return _finalize_functions(out)
def _python_signature(node: py_ast.AST) -> str:
"""Render a Python parameter list, used only to split colliding identities."""
args = getattr(node, "args", None)
if args is None:
return "()"
names = [arg.arg for arg in (*args.posonlyargs, *args.args)]
if args.vararg:
names.append(f"*{args.vararg.arg}")
names.extend(arg.arg for arg in args.kwonlyargs)
if args.kwarg:
names.append(f"**{args.kwarg.arg}")
return f"({', '.join(names)})"
def _extract_functions_python(source: str, filepath: str) -> list[dict[str, Any]]:
"""Extract Python callables with scope-qualified identity (no class bodies)."""
try:
tree = py_ast.parse(source)
except SyntaxError:
return []
out: list[dict[str, Any]] = []
def visit(node: py_ast.AST, scope: list[str]) -> None:
for child in py_ast.iter_child_nodes(node):
if isinstance(child, (py_ast.FunctionDef, py_ast.AsyncFunctionDef)):
start = child.lineno
end = getattr(child, "end_lineno", child.lineno)
out.append({
"name": ".".join(scope + [child.name]),
"loc": end - start + 1,
"start": start,
"end": end,
"signature": _python_signature(child),
})
visit(child, scope + [child.name])
elif isinstance(child, py_ast.ClassDef):
visit(child, scope + [child.name])
else:
visit(child, scope)
visit(tree, [])
return _finalize_functions(out)
def _extract_functions_go(source: str, filepath: str) -> list[dict[str, Any]]:
return _extract_functions_brace(source, "go")
def _extract_functions_dart(source: str, filepath: str) -> list[dict[str, Any]]:
return _extract_functions_brace(source, "dart")
def _extract_functions_shell(source: str, filepath: str) -> list[dict[str, Any]]:
return _extract_functions_brace(source, "shell")
def _extract_functions_kotlin(source: str, filepath: str) -> list[dict[str, Any]]:
return _extract_functions_brace(source, "kotlin")
def _extract_functions_swift(source: str, filepath: str) -> list[dict[str, Any]]:
return _extract_functions_brace(source, "swift")
EXTRACTORS: dict[str, Any] = {
"python": _extract_functions_python,
"go": _extract_functions_go,
"dart": _extract_functions_dart,
"shell": _extract_functions_shell,
"kotlin": _extract_functions_kotlin,
"swift": _extract_functions_swift,
}
def _extract_functions(source: str, lang: str, filepath: str) -> list[dict[str, Any]]:
if lang == "markdown":
return []
extractor = EXTRACTORS.get(lang)
if extractor is None:
return []
return extractor(source, filepath)
# ---------- read-set extraction ----------
def _extract_imports_python(source: str) -> list[str]:
try:
tree = py_ast.parse(source)
except SyntaxError:
return []
refs: list[str] = []
for node in py_ast.walk(tree):
if isinstance(node, py_ast.Import):
for alias in node.names:
refs.append(alias.name)
elif isinstance(node, py_ast.ImportFrom):
if node.module:
refs.append(node.module)
return sorted(set(refs))
def _extract_imports_go(source: str) -> list[str]:
refs: list[str] = []
block_match = re.findall(r"import\s+\((.*?)\)", source, re.DOTALL)
for block in block_match:
for line in block.splitlines():
m = re.match(r'\s*["\']([^"\']+)["\']\s+', line.strip())
if m:
refs.append(m.group(1))
single_match = re.findall(r'import\s+"([^"]+)"', source)
refs.extend(single_match)
return sorted(set(refs))
def _extract_imports_dart(source: str) -> list[str]:
refs: list[str] = []
refs.extend(re.findall(r"import\s+['\"]([^'\"]+)['\"]", source))
refs.extend(re.findall(r"export\s+['\"]([^'\"]+)['\"]", source))
return sorted(set(refs))
def _extract_imports_shell(source: str) -> list[str]:
refs: list[str] = []
refs.extend(re.findall(r"^\.?\s*\.?\s*\.\s*([^#]+?)(?:\s|$)", source, re.MULTILINE))
refs.extend(re.findall(r"source\s+['\"]([^'\"]+)['\"]", source, re.MULTILINE))
refs.extend(re.findall(r"^source\s+\.?\s*(\S+)", source, re.MULTILINE))
return sorted(set(refs))
def _extract_imports_kotlin(source: str) -> list[str]:
refs: list[str] = []
refs.extend(re.findall(r"^import\s+([\w.]+)\s*$", source, re.MULTILINE))
return sorted(set(refs))
def _extract_imports_swift(source: str) -> list[str]:
refs: list[str] = []
refs.extend(re.findall(r"^import\s+(\w+)", source, re.MULTILINE))
return sorted(set(refs))
IMPORT_EXTRACTORS: dict[str, Any] = {
"python": _extract_imports_python,
"go": _extract_imports_go,
"dart": _extract_imports_dart,
"shell": _extract_imports_shell,
"kotlin": _extract_imports_kotlin,
"swift": _extract_imports_swift,
}
def _extract_read_set(source: str, lang: str) -> list[str]:
extractor = IMPORT_EXTRACTORS.get(lang)
if extractor is None:
return []
return extractor(source)
# ---------- git integration ----------
def _run_git_z(args: list[str]) -> list[str]:
"""Run a NUL-separated git command and return its paths."""
try:
result = subprocess.run(
["git", *args],
capture_output=True,
text=False,
check=True,
)
except subprocess.CalledProcessError as exc:
stderr = exc.stderr.decode("utf-8", errors="replace") if exc.stderr else ""
print(f"git {' '.join(args)} failed: {stderr}", file=sys.stderr)
sys.exit(2)
output = result.stdout.decode("utf-8", errors="replace")
output = output.rstrip("\x00")
if not output:
return []
return output.split("\x00")
def _run_git_ls_files_z() -> list[str]:
"""Tracked paths in the git index."""
return _run_git_z(["ls-files", "-z"])
def _run_git_staged_deletions_z() -> list[str]:
"""Paths deleted in the index relative to HEAD (git rm / git rm --cached)."""
return _run_git_z(["diff-index", "--cached", "--name-only", "--diff-filter=D", "-z", "HEAD"])
def _run_git_ls_files_untracked_z() -> list[str]:
"""Untracked, non-ignored paths."""
return _run_git_z(["ls-files", "-o", "-z", "--exclude-standard"])
def _should_exclude(path: str) -> bool:
for pattern in EXCLUSION_PATTERNS:
if path.startswith(pattern):
return True
return False
def _is_tracked_language(path: str) -> bool:
return bool(_detect_language(path))
# ---------- input mode ----------
def collect_files(input_mode: str) -> list[str]:
"""Collect file paths based on input mode.
- tracked: paths in the git index
- worktree: post-commit projection of the working tree, i.e. every tracked
path that still exists on disk (modified files included) plus
non-ignored untracked paths that are not staged for removal.
"""
if input_mode == "tracked":
candidates = _run_git_ls_files_z()
elif input_mode == "worktree":
staged_deletions = set(_run_git_staged_deletions_z())
candidates = [path for path in _run_git_ls_files_z() if os.path.isfile(path)]
candidates.extend(
path
for path in _run_git_ls_files_untracked_z()
if path not in staged_deletions and os.path.isfile(path)
)
else:
print(f"ERROR: unknown input-mode '{input_mode}'", file=sys.stderr)
sys.exit(2)
selected = {
path
for path in candidates
if path and not _should_exclude(path) and _is_tracked_language(path)
}
return sorted(selected)
# ---------- measurement ----------
@dataclass
class FileReport:
path: str
language: str
kind: str
loc: int
functions: list[dict[str, Any]] = field(default_factory=list)
read_set: list[str] = field(default_factory=list)
largest_function: dict[str, Any] | None = None
def measure_file(path: str, source: str) -> FileReport:
lang = _detect_language(path)
if not lang:
return FileReport(path=path, language="", kind="production", loc=0)
kind = classify_file(path)
lines = source.splitlines()
loc = len(lines)
functions = _extract_functions(source, lang, path)
largest = functions[0] if functions else None
read_set = _extract_read_set(source, lang)
return FileReport(
path=path,
language=lang,
kind=kind,
loc=loc,
functions=functions,
read_set=read_set,
largest_function=largest,
)
# ---------- threshold helpers ----------
def _file_threshold_for_kind(kind: str) -> dict[str, int]:
return FILE_POLICY.get(kind, FILE_POLICY["production"])
def _function_threshold() -> dict[str, int]:
return FUNCTION_POLICY
def _file_level_for_loc(kind: str, loc: int) -> str | None:
"""Return the threshold level ('warning', 'split_review', 'exception') or None."""
policy = _file_threshold_for_kind(kind)
if "exception" in policy and loc > policy["exception"]:
return "exception"
if "split_review" in policy and loc > policy["split_review"]:
return "split_review"
if "warning" in policy and loc > policy["warning"]:
return "warning"
return None
def _function_level_for_loc(loc: int) -> str | None:
"""Return the threshold level ('warning', 'split_review') or None."""
if loc > FUNCTION_POLICY["split_review"]:
return "split_review"
if loc > FUNCTION_POLICY["warning"]:
return "warning"
return None
# ---------- audit ----------
def load_json_file(path: str) -> dict[str, Any] | None:
if not os.path.isfile(path):
return None
try:
with open(path, "r", encoding="utf-8") as f:
return json.load(f)
except (json.JSONDecodeError, OSError):
return None
def load_required_json_file(path: str) -> tuple[dict[str, Any] | None, str | None]:
"""Load a required JSON object config as (config, error).
A missing, unreadable, malformed, or non-object config is a configuration
failure with its own diagnostic. It never degrades to a silent None, which
would skip every check that depends on the config being present.
"""
if not os.path.isfile(path):
return None, f"required config file not found at '{path}'"
try:
with open(path, "r", encoding="utf-8") as f:
data = json.load(f)
except json.JSONDecodeError as exc:
return None, f"malformed JSON in '{path}': {exc}"
except OSError as exc:
return None, f"cannot read '{path}': {exc.strerror}"
if not isinstance(data, dict):
return None, f"root of '{path}' is not a JSON object"
return data, None
def audit(files: list[str]) -> dict[str, Any]:
"""Run readability audit on given files. Returns full audit dict."""
reports: list[FileReport] = []
for f in sorted(files):
full = os.path.join(os.getcwd(), f)
try:
with open(full, "r", encoding="utf-8", errors="replace") as fh:
source = fh.read()
except OSError:
continue
reports.append(measure_file(f, source))
reports.sort(key=lambda r: r.path)
file_entries: list[dict[str, Any]] = []
violations: list[dict[str, Any]] = []
total_loc = 0
total_functions = 0
for r in reports:
if r.language == "":
continue
if r.language == "markdown" and r.kind != "skill_entrypoint":
continue
total_loc += r.loc
total_functions += len(r.functions)
entry: dict[str, Any] = {
"path": r.path,
"language": r.language,
"kind": r.kind,
"lines_of_code": r.loc,
}
if r.functions:
entry["functions"] = r.functions
if r.largest_function:
entry["largest_function"] = r.largest_function
if r.read_set:
entry["read_set"] = r.read_set
file_entries.append(entry)
# file-level threshold
level = _file_level_for_loc(r.kind, r.loc)
if level:
violations.append({
"path": r.path,
"metric": "file_loc",
"level": level,
"value": r.loc,
"reason": f"file {r.kind} exceeds {level} threshold ({r.loc} > {_file_threshold_for_kind(r.kind).get(level, 0)})",
})
# function-level threshold: report each oversized function individually
for fn in r.functions:
fn_level = _function_level_for_loc(fn["loc"])
if fn_level:
violations.append({
"path": r.path,
"metric": "function_loc",
"level": fn_level,
"value": fn["loc"],
"function": fn["name"],
"reason": f"function {fn['name']} exceeds {fn_level} threshold ({fn['loc']} > {_function_threshold().get(fn_level, 0)})",
})
output: dict[str, Any] = {
"audit_version": "2.0",
"input_mode": "tracked", # will be overridden by caller
"files": file_entries,
"violations": sorted(violations, key=lambda v: (v["path"], v["metric"], v.get("function", ""), v.get("level", ""))),
"summary": {
"total_files": len(file_entries),
"total_lines_of_code": total_loc,
"total_functions": total_functions,
"violations_count": len(violations),
},
"policy": {
"file": FILE_POLICY,
"function": FUNCTION_POLICY,
},
}
return output
# ---------- ratchet / baseline ----------
FILE_ENTRY_KEYS: frozenset[str] = frozenset({"path", "metric", "level", "value", "reason"})
FUNCTION_ENTRY_KEYS: frozenset[str] = frozenset({"path", "metric", "level", "value", "function", "reason"})
TASK_TOTAL_ENTRY_KEYS: frozenset[str] = frozenset({"task_id", "metric", "value", "reason"})
def _is_loc_value(value: Any) -> bool:
return isinstance(value, int) and not isinstance(value, bool) and value >= 0
def _is_filled_str(value: Any) -> bool:
return isinstance(value, str) and bool(value.strip())
def _validate_threshold_section(
baseline: dict[str, Any],
section: str,
metric: str,
levels: tuple[str, ...],
allowed_keys: frozenset[str],
errors: list[str],
) -> None:
entries = baseline.get(section, [])
if not isinstance(entries, list):
errors.append(f"{section} is not a list")
return
seen: set[tuple[str, str, str]] = set()
for index, entry in enumerate(entries):
label = f"{section}[{index}]"
if not isinstance(entry, dict):
errors.append(f"{label} is not an object")
continue
keys = set(entry)
missing = sorted(allowed_keys - keys)
unknown = sorted(keys - allowed_keys)
if missing:
errors.append(f"{label} missing keys {missing}")
if unknown:
errors.append(f"{label} unknown keys {unknown}")
if not _is_filled_str(entry.get("path")):
errors.append(f"{label} has invalid 'path'")
if entry.get("metric") != metric:
errors.append(f"{label} has invalid 'metric' (expected '{metric}')")
if entry.get("level") not in levels:
errors.append(f"{label} has invalid 'level' {entry.get('level')!r}")
if not _is_loc_value(entry.get("value")):
errors.append(f"{label} has invalid 'value' {entry.get('value')!r}")
if not _is_filled_str(entry.get("reason")):
errors.append(f"{label} has empty/missing reason")
if metric == "function_loc" and not _is_filled_str(entry.get("function")):
errors.append(f"{label} has invalid 'function'")
identity = (str(entry.get("path")), metric, str(entry.get("function", "")))
if identity in seen:
errors.append(f"{label} duplicate identity {identity}")
seen.add(identity)
def _validate_task_total_section(baseline: dict[str, Any], errors: list[str]) -> None:
entries = baseline.get("task_read_set_totals", [])
if not isinstance(entries, list):
errors.append("task_read_set_totals is not a list")
return
seen: set[str] = set()
for index, entry in enumerate(entries):
label = f"task_read_set_totals[{index}]"
if not isinstance(entry, dict):
errors.append(f"{label} is not an object")
continue
keys = set(entry)
missing = sorted(TASK_TOTAL_ENTRY_KEYS - keys)
unknown = sorted(keys - TASK_TOTAL_ENTRY_KEYS)
if missing:
errors.append(f"{label} missing keys {missing}")
if unknown:
errors.append(f"{label} unknown keys {unknown}")
if not _is_filled_str(entry.get("task_id")):
errors.append(f"{label} has invalid 'task_id'")
if entry.get("metric") != "read_set_total":
errors.append(f"{label} has invalid 'metric' (expected 'read_set_total')")
if not _is_loc_value(entry.get("value")):
errors.append(f"{label} has invalid 'value' {entry.get('value')!r}")
if not _is_filled_str(entry.get("reason")):
errors.append(f"{label} has empty/missing reason")
task_id = str(entry.get("task_id"))
if task_id in seen:
errors.append(f"{label} duplicate task_id {task_id!r}")
seen.add(task_id)
def _validate_baseline(baseline: dict[str, Any]) -> list[str]:
"""Validate baseline schema strictly. Returns list of error messages."""
if not isinstance(baseline, dict):
return ["baseline is not an object"]
errors: list[str] = []
_validate_threshold_section(
baseline, "file_thresholds", "file_loc", FILE_LEVELS, FILE_ENTRY_KEYS, errors)
_validate_threshold_section(
baseline, "function_thresholds", "function_loc", FUNCTION_LEVELS, FUNCTION_ENTRY_KEYS, errors)
_validate_task_total_section(baseline, errors)
return errors
def _violation_identity(violation: dict[str, Any]) -> tuple[str, str, str]:
metric = violation["metric"]
function = violation.get("function", "") if metric == "function_loc" else ""
return (violation["path"], metric, function or "")
def _build_baseline_lookup(baseline: dict[str, Any]) -> dict[tuple[str, str, str], dict[str, Any]]:
"""Build lookup from stable ratchet identity to baseline entry.
The identity deliberately excludes 'level' so that a level drop reads as an
improvement of a known violation rather than a new one.
"""
lookup: dict[tuple[str, str, str], dict[str, Any]] = {}
for entry in baseline.get("file_thresholds", []):
lookup[(entry["path"], "file_loc", "")] = entry
for entry in baseline.get("function_thresholds", []):
lookup[(entry["path"], "function_loc", entry.get("function", ""))] = entry
return lookup
def _regression_reason(violation: dict[str, Any], baseline_entry: dict[str, Any] | None) -> str | None:
"""Return why this violation regresses against baseline, or None if allowed."""
if baseline_entry is None:
return "new violation not in baseline"
level = violation.get("level", "")
baseline_level = baseline_entry.get("level", "")
rank = SEVERITY_RANK.get(level, 0)
baseline_rank = SEVERITY_RANK.get(baseline_level, 0)
if rank > baseline_rank:
return f"level increased from {baseline_level} to {level}"
if rank < baseline_rank:
return None
if violation["value"] > baseline_entry.get("value", 0):
return f"value increased from {baseline_entry.get('value', 0)}"
return None
def _issue(path: str, metric: str, value: int, reason: str,
level: str = "", function: str | None = None) -> dict[str, Any]:
"""Build a fully-rendered issue: every key the CLI prints is always defined."""
return {
"path": path,
"metric": metric,
"level": level,
"value": value,
"function": function,
"reason": reason,
}
def ratchet_read_set_totals(task_read_sets: list[dict[str, Any]],
baseline: dict[str, Any]) -> list[dict[str, Any]]:
"""Fail over-budget task totals that are new or larger than the baseline."""
lookup = {
entry["task_id"]: entry
for entry in baseline.get("task_read_set_totals", [])
if isinstance(entry, dict) and "task_id" in entry
}
issues: list[dict[str, Any]] = []
for task in task_read_sets:
total = task["total_loc"]
max_total = task.get("budget", {}).get("max_total_loc")
if max_total is None or total <= max_total:
continue
baseline_entry = lookup.get(task["task_id"])
if baseline_entry is None:
reason = f"over-budget task total {total} > budget {max_total} is not allowlisted in baseline"
elif total > baseline_entry.get("value", 0):
reason = f"task total increased from {baseline_entry.get('value', 0)} to {total}"
else:
continue
issues.append(_issue(
path=f"<read-set:{task['task_id']}>",
metric="read_set_total",
value=total,
reason=reason,
))
return issues
def ratchet_check(current: dict[str, Any], baseline: dict[str, Any],
read_sets: dict[str, Any] | None) -> list[dict[str, Any]]:
"""Compare current audit against baseline. Returns new/increased violations.
When read_sets is not None the task read-set totals in current are checked
against the same baseline, so one call covers the whole ratchet contract.
"""
errors = _validate_baseline(baseline)
if errors:
return [_issue("<baseline>", "schema", 0,
f"baseline validation errors: {'; '.join(errors)}")]
lookup = _build_baseline_lookup(baseline)
regressions: list[dict[str, Any]] = []
for violation in current.get("violations", []):
baseline_entry = lookup.get(_violation_identity(violation))
reason = _regression_reason(violation, baseline_entry)
if reason is None:
continue
function = violation.get("function", "")
regressions.append(_issue(
path=violation["path"],
metric=violation["metric"],
value=violation["value"],
reason=reason,
level=violation.get("level", ""),
function=function if function else None,
))
if read_sets is not None:
regressions.extend(
ratchet_read_set_totals(current.get("task_read_sets", []), baseline))
seen: set[tuple] = set()
deduped: list[dict[str, Any]] = []
for issue in regressions:
key = (issue["path"], issue["metric"], issue.get("level", ""), issue.get("function") or "")
if key not in seen:
seen.add(key)
deduped.append(issue)
return sorted(deduped, key=lambda v: (v["path"], v["metric"], v.get("level", ""), v.get("function") or ""))
# ---------- read-set audit ----------
READ_SET_ROOT_KEYS: frozenset[str] = frozenset({"version", "generated_at", "tasks"})
SUPPORTED_READ_SET_VERSIONS: frozenset[str] = frozenset({"2.0"})
TASK_KEYS: frozenset[str] = frozenset({"task_id", "description", "files", "budget"})
BUDGET_KEYS: frozenset[str] = frozenset({"max_total_loc", "reason"})
def _config_error(task_id: str, reason: str) -> dict[str, Any]:
issue = _issue(f"<read-set:{task_id or '?'}>", "read_set_config", 0, reason)
issue["task_id"] = task_id
return issue
def _budget_error(task_id: str, value: int, reason: str) -> dict[str, Any]:
issue = _issue(f"<read-set:{task_id or '?'}>", "read_set_budget", value, reason)
issue["task_id"] = task_id
return issue
def _validate_task_budget(task_id: str, budget: Any, errors: list[dict[str, Any]]) -> None:
if not isinstance(budget, dict):
errors.append(_config_error(task_id, "budget is missing or not an object"))
return
unknown = sorted(set(budget) - BUDGET_KEYS)
if unknown:
errors.append(_config_error(task_id, f"budget has unknown keys {unknown}"))
max_total = budget.get("max_total_loc")
if not _is_loc_value(max_total) or max_total == 0:
errors.append(_config_error(task_id, f"budget has invalid 'max_total_loc' {max_total!r}"))
if not isinstance(budget.get("reason", ""), str):
errors.append(_config_error(task_id, "budget 'reason' is not a string"))
def _measure_task_files(task_id: str, files: Any, errors: list[dict[str, Any]]) -> list[dict[str, Any]]:
"""Measure LOC per read-set path in configured order; report config failures."""
if not isinstance(files, list) or not files:
errors.append(_config_error(task_id, "files is missing or empty"))
return []
details: list[dict[str, Any]] = []
seen: set[str] = set()
for path in files:
if not _is_filled_str(path):
errors.append(_config_error(task_id, f"invalid path {path!r} in read_set"))
continue
if path in seen:
errors.append(_config_error(task_id, f"duplicate path '{path}' in read_set"))
continue
seen.add(path)
full = os.path.join(os.getcwd(), path)
try:
with open(full, "r", encoding="utf-8", errors="replace") as fh:
loc = len(fh.read().splitlines())
except OSError:
errors.append(_config_error(task_id, f"missing read_set path '{path}'"))
continue
details.append({"path": path, "loc": loc})
return details
def _validate_read_set_root(read_sets_config: Any) -> dict[str, Any] | None:
"""Validate the read-set root against the exact schema.
Returns the first configuration error, or None when the root carries
exactly the required keys with a supported version and typed metadata. A
root error is fatal: task totals are meaningless under an unknown schema.
"""
if read_sets_config is None:
return _config_error("", "read-set config is required but was not loaded")
if not isinstance(read_sets_config, dict):
return _config_error("", "read-set config root is not an object")
unknown_root = sorted(set(read_sets_config) - READ_SET_ROOT_KEYS)
if unknown_root:
return _config_error("", f"read-set config has unknown root keys {unknown_root}")
for required_key in ("version", "generated_at"):
if required_key not in read_sets_config:
return _config_error(
"", f"read-set config is missing required root key '{required_key}'")
value = read_sets_config[required_key]
if not _is_filled_str(value):
return _config_error(
"", f"read-set config root '{required_key}' must be a non-empty string")
if read_sets_config["version"] not in SUPPORTED_READ_SET_VERSIONS:
return _config_error(
"", f"read-set config unsupported version {read_sets_config['version']!r}; "
f"supported: {sorted(SUPPORTED_READ_SET_VERSIONS)}")
tasks = read_sets_config.get("tasks")
if not isinstance(tasks, list) or not tasks:
return _config_error("", "read-set config has no 'tasks' list")
return None
def audit_read_sets(current: dict[str, Any] | None,
read_sets_config: dict[str, Any] | None) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
"""Validate read-set configuration and compute ordered totals.
Returns (issues, task_read_sets). Issues carry metric 'read_set_config' for
configuration failures and 'read_set_budget' for reasonless over-budget
totals; both render with a defined value/level.
"""
root_error = _validate_read_set_root(read_sets_config)
if root_error is not None:
return [root_error], []
errors: list[dict[str, Any]] = []
task_read_sets: list[dict[str, Any]] = []
tasks = read_sets_config["tasks"]
seen_tasks: set[str] = set()
for index, task in enumerate(tasks):
if not isinstance(task, dict):
errors.append(_config_error("", f"tasks[{index}] is not an object"))
continue
task_id = task.get("task_id", "")
if not _is_filled_str(task_id):
errors.append(_config_error("", f"tasks[{index}] has invalid 'task_id'"))
continue
if task_id in seen_tasks:
errors.append(_config_error(task_id, "duplicate task_id in read-set config"))
continue
seen_tasks.add(task_id)
unknown = sorted(set(task) - TASK_KEYS)
if unknown:
errors.append(_config_error(task_id, f"task has unknown keys {unknown}"))
if not _is_filled_str(task.get("description")):
errors.append(_config_error(
task_id, "task is missing required non-empty 'description'"))
budget = task.get("budget")
_validate_task_budget(task_id, budget, errors)
file_details = _measure_task_files(task_id, task.get("files"), errors)
total_loc = sum(detail["loc"] for detail in file_details)
budget_out = budget if isinstance(budget, dict) else {}
task_read_sets.append({
"task_id": task_id,
"files": file_details,
"total_loc": total_loc,
"budget": budget_out,
"reason": budget_out.get("reason", ""),
})
max_total = budget_out.get("max_total_loc")
if _is_loc_value(max_total) and total_loc > max_total and not _is_filled_str(budget_out.get("reason")):
errors.append(_budget_error(
task_id, total_loc,
f"task total {total_loc} exceeds budget {max_total} with empty reason"))
return errors, task_read_sets
# ---------- CLI ----------
def _render_issue(issue: dict[str, Any]) -> str:
function = f" func={issue['function']}" if issue.get("function") else ""
level = issue.get("level") or "-"
return (f" {issue['path']}: {issue['metric']}{function}={issue.get('value', 0)} "
f"level={level} ({issue['reason']})")
def _write_report(output_path: str, result: dict[str, Any]) -> None:
if output_path == "/dev/null":
return
os.makedirs(os.path.dirname(output_path) or ".", exist_ok=True)
with open(output_path, "w", encoding="utf-8") as f:
json.dump(result, f, indent=2, ensure_ascii=False, sort_keys=False)
f.write("\n")
def main() -> None:
parser = argparse.ArgumentParser(description="readability audit for IOP repo")
parser.add_argument("--check", action="store_true",
help="ratchet: fail if new/increased violations")
parser.add_argument("--output", type=str, default="/dev/null",
help="write JSON to this file (default: /dev/null)")
parser.add_argument("--input-mode", type=str, default="tracked",
choices=["tracked", "worktree"],
help="input mode: tracked (index only) or worktree (post-commit projection)")
args = parser.parse_args()
files = collect_files(args.input_mode)
result = audit(files)
result["input_mode"] = args.input_mode
repo_root = os.getcwd()
read_sets_path = os.path.join(repo_root, "scripts", "readability_read_sets.json")
read_sets, read_sets_error = load_required_json_file(read_sets_path)
if read_sets_error is None:
rs_issues, task_read_sets = audit_read_sets(result, read_sets)
else:
rs_issues, task_read_sets = [_config_error("", read_sets_error)], []
if task_read_sets:
result["task_read_sets"] = task_read_sets
_write_report(args.output, result)
summary = (f"readability-audit: {result['summary']['total_files']} files, "
f"{result['summary']['total_lines_of_code']} LOC, "
f"{result['summary']['total_functions']} functions, "
f"{result['summary']['violations_count']} violations")
config_issues = [i for i in rs_issues if i["metric"] == "read_set_config"]
if config_issues:
print("READ-SET CONFIGURATION FAIL:", file=sys.stderr)
for issue in config_issues:
print(_render_issue(issue), file=sys.stderr)
print(summary)
sys.exit(2)
if args.check:
baseline_path = os.path.join(repo_root, "scripts", "readability_baseline.json")
baseline = load_json_file(baseline_path)
if baseline is None:
print(f"ERROR: baseline not found at {baseline_path}", file=sys.stderr)
sys.exit(3)
regressions = ratchet_check(result, baseline, read_sets)
regressions.extend(i for i in rs_issues if i["metric"] == "read_set_budget")
if regressions:
print("RATCHET FAIL: new or increased violations:", file=sys.stderr)
for issue in regressions:
print(_render_issue(issue), file=sys.stderr)
print(summary)
sys.exit(4)
print("RATCHET OK: no new or increased violations.", file=sys.stderr)
print(summary)
if __name__ == "__main__":
main()
Reference in a new issue View git blame Copy permalink