From d5dded84588e0e897f6807a0c326d731252854f4 Mon Sep 17 00:00:00 2001 From: jaylfc Date: Tue, 28 Jul 2026 00:14:25 +0000 Subject: [PATCH] tsk-mjiicy [OPEN] CI check: detect a PR that silently DELETES merged --- .github/workflows/deleted-symbols-gate.yml | 39 +++ scripts/check_deleted_symbols.py | 249 ++++++++++++++ tests/test_check_deleted_symbols.py | 361 +++++++++++++++++++++ 3 files changed, 649 insertions(+) create mode 100644 .github/workflows/deleted-symbols-gate.yml create mode 100755 scripts/check_deleted_symbols.py create mode 100644 tests/test_check_deleted_symbols.py diff --git a/.github/workflows/deleted-symbols-gate.yml b/.github/workflows/deleted-symbols-gate.yml new file mode 100644 index 000000000..06db6060b --- /dev/null +++ b/.github/workflows/deleted-symbols-gate.yml @@ -0,0 +1,39 @@ +name: Deleted symbols gate + +# Detects PRs that silently delete Python symbols (def/class/test names) that +# landed on the target branch after the PR's merge base. A PR cut before +# hardening commits landed on dev would silently delete them when merged -- +# git reports "Automatic merge went well" with no conflict because the PR +# branch simply wins on files dev touched after the branch point. +# +# This check fails such PRs and names the deleted symbols and the commits that +# added them. A "Removes-Intentionally: " trailer in the PR body +# waives named symbols, making deliberate deletions a conscious, auditable act. +# +# See scripts/check_deleted_symbols.py for the implementation. + +on: + pull_request: + branches: [master, dev] + +jobs: + deleted-symbols-gate: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v7 + with: + fetch-depth: 0 + + - uses: actions/setup-python@v7 + with: + python-version: "3.12" + + - name: Fetch base branch + env: + BASE_REF: ${{ github.base_ref }} + run: git fetch origin "$BASE_REF" + + - name: Check for silently deleted symbols + env: + PR_BODY: ${{ github.event.pull_request.body }} + run: python scripts/check_deleted_symbols.py --base "origin/${{ github.base_ref }}" diff --git a/scripts/check_deleted_symbols.py b/scripts/check_deleted_symbols.py new file mode 100755 index 000000000..9c92d2148 --- /dev/null +++ b/scripts/check_deleted_symbols.py @@ -0,0 +1,249 @@ +#!/usr/bin/env python3 +"""Deleted-symbols guard. + +Detects PRs that silently delete code added to the target branch after the +merge base. This catches the "clean merge, no conflict" failure mode where a +PR branch cut before hardening commits landed on dev would silently delete +them when merged -- git reports "Automatic merge went well" with no conflict +because the PR branch simply wins on files dev touched after the branch point. + +Algorithm: + 1. Compute the merge base of HEAD and the target branch. + 2. Extract all Python def/class symbols at three points: the target branch + head, HEAD (the PR branch), and the merge base. + 3. A symbol is "deleted by the PR" if it exists at the target head but not + at HEAD. It is "added after the merge base" if it does not exist at the + merge base. The intersection is the signal: work that landed on dev after + the branch was cut and is being silently removed. + 4. Fail with an explicit list of the deleted symbols and the commits that + added them. + 5. A "Removes-Intentionally: , ..." trailer in the PR body waives + named symbols, making deliberate deletions a conscious, auditable act. + +Narrow by design: Python def/class names only (test functions are defs). +No semantic analysis -- name-level matching catches every instance hit so far. + +Usage: + python scripts/check_deleted_symbols.py --base origin/dev + python scripts/check_deleted_symbols.py --base origin/dev --pr-body "..." + python scripts/check_deleted_symbols.py --base origin/dev --waived "path/file.py:func" +""" +from __future__ import annotations + +import argparse +import ast +import io +import os +import subprocess +import sys +import tarfile +from dataclasses import dataclass +from pathlib import Path + +REPO_ROOT = Path(__file__).resolve().parent.parent +TRAILER = "Removes-Intentionally:" + + +@dataclass +class Violation: + symbol: str + added_by: str + + +def _run_git(args: list[str], cwd: str | Path | None = None) -> str: + result = subprocess.run( + ["git", *args], cwd=cwd, capture_output=True, text=True, check=True, + ) + return result.stdout + + +def _merge_base(base_ref: str, repo_root: Path = REPO_ROOT) -> str: + return _run_git(["merge-base", "HEAD", base_ref], cwd=repo_root).strip() + + +def _extract_symbols(source: str, file_path: str) -> dict[str, str]: + """Extract symbol identifiers from Python source code. + + Returns a dict mapping "file_path:qualified_name" to kind ("def" or + "class"). Qualified names use dot notation for nested definitions + (e.g. ClassName.method, Outer.Inner). + """ + symbols: dict[str, str] = {} + try: + tree = ast.parse(source) + except SyntaxError: + return symbols + + def visit(node: ast.AST, prefix: str = "") -> None: + for child in ast.iter_child_nodes(node): + if isinstance(child, (ast.FunctionDef, ast.AsyncFunctionDef)): + name = f"{prefix}{child.name}" if prefix else child.name + symbols[f"{file_path}:{name}"] = "def" + visit(child, f"{name}.") + elif isinstance(child, ast.ClassDef): + name = f"{prefix}{child.name}" if prefix else child.name + symbols[f"{file_path}:{name}"] = "class" + visit(child, f"{name}.") + else: + visit(child, prefix) + + visit(tree) + return symbols + + +def _get_symbols_at_ref(ref: str, repo_root: Path = REPO_ROOT) -> dict[str, str]: + """Get all Python symbols at a given git ref using git archive.""" + result = subprocess.run( + ["git", "archive", ref], + cwd=repo_root, capture_output=True, check=True, + ) + symbols: dict[str, str] = {} + with tarfile.open(fileobj=io.BytesIO(result.stdout)) as tar: + for member in tar.getmembers(): + if not member.name.endswith(".py"): + continue + f = tar.extractfile(member) + if f is None: + continue + source = f.read().decode("utf-8", errors="ignore") + symbols.update(_extract_symbols(source, member.name)) + return symbols + + +def _find_adding_commit( + file_path: str, + name: str, + kind: str, + merge_base: str, + target: str, + repo_root: Path = REPO_ROOT, +) -> str: + """Find the commit that added a symbol to the target branch after merge base.""" + leaf = name.split(".")[-1] + search = f"{kind} {leaf}(" + out = _run_git( + ["log", "--oneline", "--reverse", "-S", search, f"{merge_base}..{target}", "--", file_path], + cwd=repo_root, + ) + lines = out.splitlines() + if lines: + return lines[0] + # Fallback: try without the opening paren (for classes without bases, + # e.g. `class Foo:` rather than `class Foo(Bar):`). + search = f"{kind} {leaf}" + out = _run_git( + ["log", "--oneline", "--reverse", "-S", search, f"{merge_base}..{target}", "--", file_path], + cwd=repo_root, + ) + lines = out.splitlines() + if lines: + return lines[0] + return "unknown" + + +def parse_waived_symbols(pr_body: str | None) -> set[str]: + """Parse Removes-Intentionally trailer from PR body text.""" + waived: set[str] = set() + if not pr_body: + return waived + for line in pr_body.splitlines(): + line = line.strip() + if line.startswith(TRAILER): + symbols_str = line[len(TRAILER):].strip() + for sym in symbols_str.split(","): + sym = sym.strip() + if sym: + waived.add(sym) + return waived + + +def find_signal_symbols( + dev_symbols: dict[str, str], + head_symbols: dict[str, str], + mb_symbols: dict[str, str], +) -> dict[str, str]: + """Find symbols at dev that are not at HEAD and not at merge base.""" + dev_keys = set(dev_symbols.keys()) + head_keys = set(head_symbols.keys()) + mb_keys = set(mb_symbols.keys()) + signal_keys = dev_keys - head_keys - mb_keys + return {k: dev_symbols[k] for k in signal_keys} + + +def check_deleted_symbols( + base_ref: str, + repo_root: Path = REPO_ROOT, + pr_body: str | None = None, + waived: set[str] | None = None, +) -> tuple[list[Violation], set[str]]: + """Main check function. + + Returns (violations, waived_symbols) where violations is a list of + Violation objects for non-waived signal symbols, and waived_symbols is + the set of signal symbols that were waived. + """ + merge_base = _merge_base(base_ref, repo_root) + + dev_symbols = _get_symbols_at_ref(base_ref, repo_root) + head_symbols = _get_symbols_at_ref("HEAD", repo_root) + mb_symbols = _get_symbols_at_ref(merge_base, repo_root) + + signal = find_signal_symbols(dev_symbols, head_symbols, mb_symbols) + + waived_set: set[str] = set() + if waived: + waived_set.update(waived) + waived_set.update(parse_waived_symbols(pr_body)) + + violations: list[Violation] = [] + waived_in_signal: set[str] = set() + for symbol, kind in signal.items(): + if symbol in waived_set: + waived_in_signal.add(symbol) + continue + file_path, name = symbol.rsplit(":", 1) + added_by = _find_adding_commit(file_path, name, kind, merge_base, base_ref, repo_root) + violations.append(Violation(symbol=symbol, added_by=added_by)) + + return violations, waived_in_signal + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--base", required=True, help="Target branch ref (e.g. origin/dev)") + parser.add_argument("--pr-body", default=None, help="PR body text (for Removes-Intentionally trailer)") + parser.add_argument("--waived", default=None, help="Comma-separated list of waived symbols") + args = parser.parse_args(argv) + + pr_body = args.pr_body + if pr_body is None: + pr_body = os.environ.get("PR_BODY") + + waived: set[str] = set() + if args.waived: + for sym in args.waived.split(","): + sym = sym.strip() + if sym: + waived.add(sym) + + violations, waived_symbols = check_deleted_symbols(args.base, REPO_ROOT, pr_body, waived) + + if waived_symbols: + for sym in sorted(waived_symbols): + print(f"deleted-symbols-guard: waived via Removes-Intentionally: {sym}") + + if violations: + print( + f"DELETED-SYMBOLS FAIL: this PR deletes {len(violations)} symbol(s) that " + f"landed on dev after your branch was cut:" + ) + for v in violations: + print(f" - {v.symbol} (added by {v.added_by})") + return 1 + + print("deleted-symbols-guard: clean") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/test_check_deleted_symbols.py b/tests/test_check_deleted_symbols.py new file mode 100644 index 000000000..38e031ace --- /dev/null +++ b/tests/test_check_deleted_symbols.py @@ -0,0 +1,361 @@ +"""Tests for the deleted-symbols guard (scripts/check_deleted_symbols.py). + +Each test builds a synthetic git repo in a temp directory with a specific +history shape, then calls check_deleted_symbols() directly and asserts on the +result. This proves the check goes RED (fails), GREEN (passes), and that the +Removes-Intentionally trailer waives a named symbol. +""" +from __future__ import annotations + +import subprocess +import sys +from pathlib import Path + +import pytest + +# scripts/ is not a package; make it importable the same way the other +# scripts/*.py unit tests do (see tests/test_check_schema_migrations.py). +sys.path.insert(0, str(Path(__file__).parent.parent / "scripts")) +import check_deleted_symbols as cds # noqa: E402 + +REPO_ROOT = Path(__file__).resolve().parent.parent + + +def _git(repo: Path, *args: str) -> None: + subprocess.run(["git", *args], cwd=repo, capture_output=True, text=True, check=True) + + +def _init_repo(repo: Path) -> None: + repo.mkdir(parents=True, exist_ok=True) + _git(repo, "init") + _git(repo, "config", "user.name", "Test") + _git(repo, "config", "user.email", "test@test.com") + _git(repo, "config", "commit.gpgsign", "false") + _git(repo, "branch", "-M", "main") + + +def _commit_file(repo: Path, rel_path: str, content: str, message: str) -> None: + full = repo / rel_path + full.parent.mkdir(parents=True, exist_ok=True) + full.write_text(content, encoding="utf-8") + _git(repo, "add", rel_path) + _git(repo, "commit", "-m", message) + + +def _branch(repo: Path, name: str) -> None: + _git(repo, "branch", name) + + +def _checkout(repo: Path, name: str) -> None: + _git(repo, "checkout", "-q", name) + + +# --------------------------------------------------------------------------- +# Core logic unit tests (no git required) +# --------------------------------------------------------------------------- + + +class TestExtractSymbols: + def test_top_level_function_and_class(self): + source = "def foo():\n pass\n\nclass Bar:\n pass\n" + syms = cds._extract_symbols(source, "mod.py") + assert "mod.py:foo" in syms + assert syms["mod.py:foo"] == "def" + assert "mod.py:Bar" in syms + assert syms["mod.py:Bar"] == "class" + + def test_nested_method_qualified_name(self): + source = "class Bar:\n def method(self):\n pass\n" + syms = cds._extract_symbols(source, "mod.py") + assert "mod.py:Bar.method" in syms + assert syms["mod.py:Bar.method"] == "def" + + def test_nested_class(self): + source = "class Outer:\n class Inner:\n pass\n" + syms = cds._extract_symbols(source, "mod.py") + assert "mod.py:Outer.Inner" in syms + assert syms["mod.py:Outer.Inner"] == "class" + + def test_async_function(self): + source = "async def afoo():\n pass\n" + syms = cds._extract_symbols(source, "mod.py") + assert "mod.py:afoo" in syms + assert syms["mod.py:afoo"] == "def" + + def test_syntax_error_returns_empty(self): + syms = cds._extract_symbols("def (:\n", "mod.py") + assert syms == {} + + +class TestParseWaivedSymbols: + def test_parses_comma_separated(self): + body = "Some PR description.\n\nRemoves-Intentionally: mod.py:foo, mod.py:Bar.baz" + waived = cds.parse_waived_symbols(body) + assert waived == {"mod.py:foo", "mod.py:Bar.baz"} + + def test_no_trailer_returns_empty(self): + assert cds.parse_waived_symbols("just a description") == set() + + def test_none_body_returns_empty(self): + assert cds.parse_waived_symbols(None) == set() + + def test_trailer_with_no_symbols(self): + assert cds.parse_waived_symbols("Removes-Intentionally:") == set() + + def test_multiple_trailer_lines(self): + body = "Removes-Intentionally: a.py:foo\n\nRemoves-Intentionally: b.py:Bar" + waived = cds.parse_waived_symbols(body) + assert waived == {"a.py:foo", "b.py:Bar"} + + +class TestFindSignalSymbols: + def test_signal_is_dev_minus_head_minus_mb(self): + dev = {"f.py:a": "def", "f.py:b": "def", "f.py:c": "def"} + head = {"f.py:a": "def", "f.py:b": "def"} + mb = {"f.py:a": "def"} + signal = cds.find_signal_symbols(dev, head, mb) + assert signal == {"f.py:c": "def"} + + def test_no_signal_when_all_at_mb(self): + dev = {"f.py:a": "def"} + head: dict[str, str] = {} + mb = {"f.py:a": "def"} + signal = cds.find_signal_symbols(dev, head, mb) + assert signal == {} + + def test_no_signal_when_present_at_head(self): + dev = {"f.py:a": "def"} + head = {"f.py:a": "def"} + mb = {} + signal = cds.find_signal_symbols(dev, head, mb) + assert signal == {} + + +# --------------------------------------------------------------------------- +# Integration tests with synthetic git repos +# --------------------------------------------------------------------------- + + +class TestCheckDeletedSymbols: + def test_deletes_symbol_added_after_merge_base_fails(self, tmp_path: Path): + """A synthetic PR that deletes a symbol added to dev after its merge + base FAILS the check and names the symbol and the commit that added + it.""" + repo = tmp_path / "repo" + _init_repo(repo) + _commit_file( + repo, "tinyagentos/foo.py", + "def function_a():\n pass\n", + "initial: add function_a", + ) + _branch(repo, "pr-branch") + # On dev (main), add function_b after the branch point. + _commit_file( + repo, "tinyagentos/foo.py", + "def function_a():\n pass\n\ndef function_b():\n pass\n", + "feat: add function_b", + ) + # PR branch was cut before function_b landed; it does not have it. + _checkout(repo, "pr-branch") + + violations, waived = cds.check_deleted_symbols("main", repo) + + assert len(violations) == 1 + v = violations[0] + assert "function_b" in v.symbol + assert v.added_by != "unknown" + assert "function_b" in v.added_by + assert waived == set() + + def test_deletes_own_newly_added_code_passes(self, tmp_path: Path): + """A PR deleting its own newly-added code PASSES -- the symbol was + never on dev, so there is nothing to silently delete.""" + repo = tmp_path / "repo" + _init_repo(repo) + _commit_file( + repo, "tinyagentos/foo.py", + "def function_a():\n pass\n", + "initial: add function_a", + ) + _branch(repo, "pr-branch") + _checkout(repo, "pr-branch") + # On PR branch: add function_x then remove it. + _commit_file( + repo, "tinyagentos/foo.py", + "def function_a():\n pass\n\ndef function_x():\n pass\n", + "feat: add function_x", + ) + _commit_file( + repo, "tinyagentos/foo.py", + "def function_a():\n pass\n", + "refactor: remove function_x", + ) + # dev (main) never had function_x. + + violations, waived = cds.check_deleted_symbols("main", repo) + + assert violations == [] + assert waived == set() + + def test_removes_intentionally_trailer_waives(self, tmp_path: Path): + """The Removes-Intentionally trailer waives a named symbol and the + waiver is logged (returned in the waived set).""" + repo = tmp_path / "repo" + _init_repo(repo) + _commit_file( + repo, "tinyagentos/foo.py", + "def function_a():\n pass\n", + "initial: add function_a", + ) + _branch(repo, "pr-branch") + _commit_file( + repo, "tinyagentos/foo.py", + "def function_a():\n pass\n\ndef function_b():\n pass\n", + "feat: add function_b", + ) + _checkout(repo, "pr-branch") + + pr_body = "Removes-Intentionally: tinyagentos/foo.py:function_b" + violations, waived = cds.check_deleted_symbols("main", repo, pr_body=pr_body) + + assert violations == [] + assert "tinyagentos/foo.py:function_b" in waived + + def test_deleted_class_added_after_merge_base_fails(self, tmp_path: Path): + """Same signal logic for a class, not just a function.""" + repo = tmp_path / "repo" + _init_repo(repo) + _commit_file( + repo, "tinyagentos/foo.py", + "def function_a():\n pass\n", + "initial: add function_a", + ) + _branch(repo, "pr-branch") + _commit_file( + repo, "tinyagentos/foo.py", + "def function_a():\n pass\n\nclass NewClass:\n pass\n", + "feat: add NewClass", + ) + _checkout(repo, "pr-branch") + + violations, waived = cds.check_deleted_symbols("main", repo) + + assert len(violations) == 1 + assert "NewClass" in violations[0].symbol + assert violations[0].added_by != "unknown" + + def test_deleted_method_added_after_merge_base_fails(self, tmp_path: Path): + """A method added to an existing class on dev after the merge base is + caught.""" + repo = tmp_path / "repo" + _init_repo(repo) + _commit_file( + repo, "tinyagentos/foo.py", + "class Foo:\n def existing(self):\n pass\n", + "initial: add Foo.existing", + ) + _branch(repo, "pr-branch") + _commit_file( + repo, "tinyagentos/foo.py", + "class Foo:\n def existing(self):\n pass\n\n def new_method(self):\n pass\n", + "feat: add Foo.new_method", + ) + _checkout(repo, "pr-branch") + + violations, waived = cds.check_deleted_symbols("main", repo) + + assert len(violations) == 1 + assert "Foo.new_method" in violations[0].symbol + + def test_deleted_test_function_added_after_merge_base_fails(self, tmp_path: Path): + """A test function added to dev after the merge base is caught.""" + repo = tmp_path / "repo" + _init_repo(repo) + _commit_file( + repo, "tests/test_foo.py", + "def test_existing():\n pass\n", + "initial: add test_existing", + ) + _branch(repo, "pr-branch") + _commit_file( + repo, "tests/test_foo.py", + "def test_existing():\n pass\n\ndef test_new():\n pass\n", + "test: add test_new", + ) + _checkout(repo, "pr-branch") + + violations, waived = cds.check_deleted_symbols("main", repo) + + assert len(violations) == 1 + assert "test_new" in violations[0].symbol + + def test_no_signal_when_dev_unchanged_since_merge_base(self, tmp_path: Path): + """If dev has not added anything since the merge base, there is no + signal even if the PR deletes code that existed at the merge base.""" + repo = tmp_path / "repo" + _init_repo(repo) + _commit_file( + repo, "tinyagentos/foo.py", + "def function_a():\n pass\n\ndef function_b():\n pass\n", + "initial: add function_a and function_b", + ) + _branch(repo, "pr-branch") + # PR branch deletes function_b (its own code). + _checkout(repo, "pr-branch") + _commit_file( + repo, "tinyagentos/foo.py", + "def function_a():\n pass\n", + "refactor: remove function_b", + ) + + violations, waived = cds.check_deleted_symbols("main", repo) + + assert violations == [] + + def test_waived_via_cli_argument(self, tmp_path: Path): + """The --waived argument also waives symbols (for manual runs).""" + repo = tmp_path / "repo" + _init_repo(repo) + _commit_file( + repo, "tinyagentos/foo.py", + "def function_a():\n pass\n", + "initial: add function_a", + ) + _branch(repo, "pr-branch") + _commit_file( + repo, "tinyagentos/foo.py", + "def function_a():\n pass\n\ndef function_b():\n pass\n", + "feat: add function_b", + ) + _checkout(repo, "pr-branch") + + waived_arg = {"tinyagentos/foo.py:function_b"} + violations, waived = cds.check_deleted_symbols("main", repo, waived=waived_arg) + + assert violations == [] + assert "tinyagentos/foo.py:function_b" in waived + + def test_partial_waiver_still_fails_for_unwaived(self, tmp_path: Path): + """If two symbols are in the signal but only one is waived, the + unwaived one still fails.""" + repo = tmp_path / "repo" + _init_repo(repo) + _commit_file( + repo, "tinyagentos/foo.py", + "def function_a():\n pass\n", + "initial: add function_a", + ) + _branch(repo, "pr-branch") + _commit_file( + repo, "tinyagentos/foo.py", + "def function_a():\n pass\n\ndef function_b():\n pass\n\ndef function_c():\n pass\n", + "feat: add function_b and function_c", + ) + _checkout(repo, "pr-branch") + + pr_body = "Removes-Intentionally: tinyagentos/foo.py:function_b" + violations, waived = cds.check_deleted_symbols("main", repo, pr_body=pr_body) + + assert len(violations) == 1 + assert "function_c" in violations[0].symbol + assert "tinyagentos/foo.py:function_b" in waived