#!/usr/bin/env python3 """Enforce the repository's first-party Go package naming policy.""" from __future__ import annotations import re import subprocess import sys from collections import defaultdict from pathlib import Path from typing import Iterable REPO = Path(__file__).resolve().parents[2] PACKAGE_DECLARATION = re.compile(r"^\s*package\s+([A-Za-z_][A-Za-z0-9_]*)", re.MULTILINE) PACKAGE_DOCUMENTATION = re.compile(r"^// Package\s+([A-Za-z_][A-Za-z0-9_]*)\b", re.MULTILINE) PACKAGE_NAME = re.compile(r"^[a-z][a-z0-9]*$") MODULE_PREFIX = "github.com/tidbcloud/AgentTeam" EXCLUDED_PREFIXES = ( "daemon/web/node_modules/", "kits/agentteam-insight/testdata/smoke/fixture-repo/", "x/", ) RETIRED_PACKAGE_PATHS = ( "internal/api/meshx", "internal/api/personal_access_tokens", "internal/api/publicrepocache", "internal/api/repository_access_tokens", "internal/api/runtimeidentity", "internal/api/workflowbinding", "internal/api/workflowinvocation", "internal/api/workflowplan", "internal/api/workflowrun", "internal/contentstore", "internal/core/appcapabilities", "internal/core/runtimecatalog", "internal/extension/providers/feishucustombot", "internal/extension/providers/mail/api", "internal/extension/providers/mail/connectionconfig", "internal/extension/providers/mail/runtime", "internal/extension/providers/mail/store", "internal/extension/runtime", "internal/knowledgeqa", "internal/maas/hostedavailability", "internal/runtime", "internal/service/agent", "internal/service/agentmanagement", "internal/service/appinstallation", "internal/service/auth", "internal/service/channelmanagement", "internal/service/collaboratorrequest", "internal/service/contentmigration", "internal/service/contentstore", "internal/service/extension", "internal/service/knowledgeqa/knowledgeqaruntime", "internal/service/personaltokens", "internal/service/publicrepocache", "internal/service/pullrequestwatch", "internal/service/resourceaccess", "internal/service/resourceactivity", "internal/service/resourcerepos", "internal/service/roleagents", "internal/service/rolemanagement", "internal/service/teamroleproposal", "internal/store/automations", "internal/store/channelmanagement", "internal/store/contentmigration", "internal/store/contentstore", "internal/store/personalaccesstokens", "internal/store/publicrepocache", "internal/store/repositoryaccesstokens", "internal/store/resourceactivity", "internal/store/resourcememory", "internal/store/resourcerepos", "internal/store/resourceroles", "internal/store/resourceskills", "internal/store/rolemanagement", "internal/store/unreadruntime", "internal/store/workflowbinding", "internal/store/workflowcalendar", "internal/store/workflowinvocation", "internal/store/workflowplan", "internal/store/workflowrun", "internal/trustedidentity", "internal/unreadruntime", "internal/workflow/binding/manage", "internal/workflow/invocation/manage", "internal/workflow/plan/manage", "internal/workflow/run/manage", "internal/workflow/runruntime", "internal/workflowerrorcontract", "internal/workspaceowner", "internal/workspacerouter", ) def tracked_go_files(repo: Path = REPO) -> list[Path]: result = subprocess.run( ["git", "ls-files", "--", "*.go"], cwd=repo, check=True, capture_output=True, text=True, ) return [repo / line for line in result.stdout.splitlines() if line] def tracked_files(repo: Path = REPO) -> list[Path]: result = subprocess.run( ["git", "ls-files", "-co", "--exclude-standard"], cwd=repo, check=True, capture_output=True, text=True, ) return [repo / line for line in result.stdout.splitlines() if line] def relative(repo: Path, path: Path) -> str: return path.resolve().relative_to(repo.resolve()).as_posix() def is_excluded(repo: Path, path: Path) -> bool: path_rel = relative(repo, path) return any(path_rel.startswith(prefix) for prefix in EXCLUDED_PREFIXES) def declared_package(path: Path) -> str | None: match = PACKAGE_DECLARATION.search(path.read_text(errors="ignore")) return match.group(1) if match else None def scan(repo: Path, files: Iterable[Path]) -> list[str]: by_directory: dict[Path, list[tuple[Path, str, bool]]] = defaultdict(list) failures: list[str] = [] for path in files: path = path if path.is_absolute() else repo / path if is_excluded(repo, path): continue source = path.read_text(errors="ignore") declaration = PACKAGE_DECLARATION.search(source) if declaration is None: failures.append(f"{relative(repo, path)}: missing package declaration") continue package = declaration.group(1) for documentation in PACKAGE_DOCUMENTATION.finditer(source[:declaration.start()]): documented = documentation.group(1) if documented != package: failures.append( f"{relative(repo, path)}: package comment names {documented!r}, " f"expected {package!r}" ) by_directory[path.parent].append((path, package, path.name.endswith("_test.go"))) for directory, declarations in sorted(by_directory.items(), key=lambda item: str(item[0])): directory_rel = relative(repo, directory) production = {package for _, package, is_test in declarations if not is_test} if len(production) > 1: failures.append( f"{directory_rel}: multiple production packages: {', '.join(sorted(production))}" ) continue if production: package = next(iter(production)) allowed_tests = {package, f"{package}_test"} invalid_tests = sorted( { declared for _, declared, is_test in declarations if is_test and declared not in allowed_tests } ) if invalid_tests: failures.append( f"{directory_rel}: test package must be {package} or {package}_test, " f"found {', '.join(invalid_tests)}" ) else: test_packages = { declared.removesuffix("_test") for _, declared, _ in declarations } if len(test_packages) != 1: failures.append( f"{directory_rel}: multiple test-only packages: " f"{', '.join(sorted(test_packages))}" ) continue package = next(iter(test_packages)) if not PACKAGE_NAME.fullmatch(package): failures.append( f"{directory_rel}: package {package!r} must be lowercase ASCII " "letters/digits with no underscores" ) continue if package != "main" and package != directory.name: failures.append( f"{directory_rel}: package {package!r} must match directory " f"basename {directory.name!r}" ) return sorted(failures) def scan_retired_import_strings(repo: Path, files: Iterable[Path]) -> list[str]: failures: list[str] = [] for path in files: path = path if path.is_absolute() else repo / path if path.suffix == ".go" or not path.is_file() or is_excluded(repo, path): continue try: source = path.read_text() except UnicodeDecodeError: continue for retired in RETIRED_PACKAGE_PATHS: needle = f"{MODULE_PREFIX}/{retired}" if re.search(re.escape(needle) + r"(?![A-Za-z0-9_/])", source): failures.append( f"{relative(repo, path)}: contains retired Go import path {needle!r}" ) return sorted(failures) def main() -> int: files = tracked_files(REPO) failures = scan(REPO, (path for path in files if path.suffix == ".go")) failures.extend(scan_retired_import_strings(REPO, files)) failures.sort() if failures: print("Go package naming policy violations:", file=sys.stderr) for failure in failures: print(f" - {failure}", file=sys.stderr) return 1 print("Go package naming policy: OK") return 0 if __name__ == "__main__": raise SystemExit(main())