#!/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-?)\s*(?P['\"]?)(?P[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\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\w+))?" ), "swift": re.compile( r"^\s*(?:(?:public|private|internal|fileprivate|open|final)\s+)*" r"(?:class|struct|enum|protocol|extension|actor)\s+(?P\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[^)]*)\)\s*)?(?P\w+)\s*[\(\[]"), "dart": re.compile( r"^\s*(?:void|int|bool|double|num|String|dynamic|[A-Z]\w*(?:<[^>]+>)?)" r"\s+(?P\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[\w.]+)\.)?(?P\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\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[\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` 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"(? 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"", 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("", "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_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_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()