diff --git a/.github/workflows/typedpython.yml b/.github/workflows/typedpython.yml new file mode 100644 index 000000000..e93916ab7 --- /dev/null +++ b/.github/workflows/typedpython.yml @@ -0,0 +1,40 @@ +name: TypedPython CI + +# The TypedPython gate and compiler are Python (python-multiplatform-ksp/src/{main,test}/python). +# Gradle does not run them; this workflow does. +on: + push: + branches: [ "develop", "main" ] + paths: [ "python-multiplatform-ksp/**", ".github/workflows/typedpython.yml" ] + pull_request: + branches: [ "develop", "main" ] + paths: [ "python-multiplatform-ksp/**", ".github/workflows/typedpython.yml" ] + +jobs: + test: + name: TypedPython tests (${{ matrix.os }}) + runs-on: ${{ matrix.os }} + strategy: + fail-fast: false + matrix: + os: [ubuntu-latest, macos-latest] + defaults: + run: + working-directory: python-multiplatform-ksp + steps: + - uses: actions/checkout@v4 + + - uses: astral-sh/setup-uv@v5 + + - name: Create the environment + # pyrefly is pinned exactly: the gate reads its versioned type-report format. + run: | + uv venv .venv --python 3.13 + uv pip install --python .venv/bin/python "pyrefly==1.3.2" pytest "cython>=3.1" setuptools + + - name: Behaviour tests + run: .venv/bin/python -m pytest -q -m "not measurement" + + - name: Measurements (recorded, sanity bounds only) + # A shared runner is not an idle machine (AGENTS.md §8): read ratios, not absolute times. + run: .venv/bin/python -m pytest -q -s -m measurement diff --git a/docs/INTENT.md b/docs/INTENT.md index 97f365c16..b103c1a70 100644 --- a/docs/INTENT.md +++ b/docs/INTENT.md @@ -26,6 +26,23 @@ CPython interpreter inside a Kotlin Multiplatform application so that: written (`Greeter('Kotlin').greet(2)` in `sample/`). 3. **The same code runs on every target** a Kotlin Multiplatform app ships to: desktop JVM, Android, iOS, Android native, and the web (wasm, experimental). +4. **The project's own Python is statically typed by default** (TypedPython). The maintainer's goal: + + > "파이썬을 컴파일 언어 및 타입 안정화, 정적 타입 검사, 컴파일 타임 타입검사를 지원할 수 있도록" + > — *Support Python as a compiled language with type stability, static type checking and + > compile-time type checking*, so that Python stays usable as AI-friendly languages such as Rust + > and TypeScript gain ground. + + > "TypedPython을 별도 레포가 아니라 PythonMultiplatform의 기본 동작이 될거야" + > — *TypedPython is not a separate repository; it will be PythonMultiplatform's default behaviour.* + + The maintainer chose the scope "검사만 기본, 컴파일은 opt-in" (2026-10-02): every user Python source + is **checked** by default; **compiling** it to native code is opt-in. The compiled code still runs on + the embedded CPython, as an extension module (`docs/design/typedpython.md`). + + > The maintainer also asked to *explore* GPU kernels written in Python without a separate language + > ("GPU 관련 최적화까지도 별도 언어 없이 지원할 방안을 모색해보자"). That is an exploration, not yet + > a commitment; it is recorded in the design note only. The CPython version moved on from 3.13 (the build pins 3.14.x today and offers opt-in free-threaded 3.14t builds on desktop only); the intent — embed **CPython itself**, not a re-implementation — is unchanged. @@ -88,7 +105,7 @@ no per-platform Python → Kotlin binding code to maintain. | A replacement for the `pythonx` packages | §2.3 — they are real packages in their own repository | | Runtime-reflection or name-lookup based binding | §2.1 — impossible under Kotlin/Native and native image | | A hand-written wrapper per Kotlin function or class | Generic adaptation plus generated tables; per-declaration wrappers were tried in 2024 and rotted (`docs/design/pythonx-adapter-design.md`) | -| A Python re-implementation, transpiler, or JVM-hosted Python | It embeds CPython itself, so C extensions keep working | +| A Python re-implementation, transpiler, or JVM-hosted Python | It embeds CPython itself, so C extensions keep working. TypedPython (§1.4) does not change this: checked code is ordinary Python, and compiled code is a CPython extension module, not a translation to another runtime | | A build tool, dependency resolver or packager | That is pypackpack (the work) and toolchain (the Gradle vocabulary) | | A sub-interpreter parallelism model | Rejected: extensions that do not declare multi-interpreter support fail to import; parallelism comes from free-threaded CPython | | A vendored interpreter | The build downloads and verifies CPython per platform | diff --git a/docs/SPEC.md b/docs/SPEC.md index 539a84a03..961e13cd9 100644 --- a/docs/SPEC.md +++ b/docs/SPEC.md @@ -239,6 +239,32 @@ All in `PM/commonTest`, so they run wherever the interpreter loads. - **N-3** Binding Kotlin/Native klib declarations at run time (B-2). `Status: planned`. - **N-4** Linux and Windows desktop runs in CI (P-2). `Status: planned`. - **N-5** Native-image upcall verification as an automated test (U-6). `Status: planned`. +- **N-6** TypedPython soundness gate (INTENT §1.4; `docs/design/typedpython.md` §4.2): checking a set of + user Python files reports every Pyrefly type error, the Any-leak kinds (`implicit-any-parameter`, + `unannotated-return`, `unknown-variable-type`, `no-any-return-implicit`, `no-any-return-explicit`, + `explicit-any`) and any user-code expression of type `Any` outside `cast()` (`any-flow`) as errors; + the forbidden list (`eval`/`exec`, computed-name `getattr`/`setattr`/`delattr`, patching an imported + module or a module-level class, rebinding an unannotated name to a different type, and a list/set/dict + display whose inferred element type is a union) as warnings in `checked` mode and errors in + `compiled` mode. Nothing + inside dependencies is reported. `python -m typedpython check` exits 0 / 1 / 2 for clean / errors / + tool failure. `Status: partial` — the gate is asserted by + `python-multiplatform-ksp/src/test/python/` (run with that module's `.venv`); it is **not yet run by the + build** (design §6 step 1b). Rebinding and mixed containers are found by a second Pyrefly pass over a + probed copy (design §6.1), which runs only when a file has a name assigned twice or a container + display; `for` targets and comprehension variables are not probed. +- **N-7** TypedPython compilation (INTENT §1.4; design §4.3 stage 1, #23): `typedpython.compiled` is the + identity in CPython; functions it marks, or every module-level function under a first-line + `# typedpython: compiled`, compile through Cython to a CPython extension **with CPython's results**: + `int` keeps arbitrary precision (C `long long` only behind an entry range check, and in pure functions + with overflow checking that redoes the call interpreted), an argument of a different runtime type + takes the interpreted path, `list[float]` elements are unboxed only when the function cannot make them + non-floats, and `math.sqrt` of a negative still raises `ValueError`. A module that fails the gate in + `compiled` mode is not compiled. `python -m typedpython demo [args]` builds a module, runs its + `main()` interpreted and compiled, and fails unless the output is identical. `Status: partial` — + asserted by `python-multiplatform-ksp/src/test/python/test_compile.py` and `test_cli.py` (CI: + `.github/workflows/typedpython.yml`); desktop only, module-level functions with plain signatures only + (methods, defaults, `*args` compile as untyped Python), no M2 benchmark targets yet. --- diff --git a/docs/design/typedpython.md b/docs/design/typedpython.md index fb126c89e..9edebcc76 100644 --- a/docs/design/typedpython.md +++ b/docs/design/typedpython.md @@ -207,6 +207,37 @@ TypedPython 은 그래서 자체 텐서 런타임을 만들지 않고 **torch op 각 단계는 AGENTS.md §5·§14 의 TDD 와 측정 테스트 규정을 따른다. 특히 3 은 경계 비용 감소를, 4 는 upstream 대비 수치 오차를 기준선으로 박아 둔다. +### 6.2 단계 2 (컴파일) 의 현재 상태 (2026-10-03, #23) + +- SPEC N-7. 의미 보존 규칙은 `compiler.py` 모듈 독스트링에 있다: `int` 는 진입 범위 검사 뒤에만 C + `long long` 이고, 순수 함수에서는 overflow 검사 후 인터프리터로 다시 실행(승격)한다. 비순수 함수의 + `int` 지역 변수는 경계가 C 정수인 `range` 인덱스만 C 정수다. `list[float]` 원소는 함수가 그것을 + float 아닌 값으로 바꿀 수 없을 때만 언박싱하고, 진입 시 모든 원소가 정확히 `float` 인지 본다. +- 첫 확인(측정 아님, load 197): 벤치 `nbody.py` 에 모듈 표식을 붙여 `demo` 로 돌리면 출력이 같고 약 + 4배. 스칼라 루프 단위 테스트는 약 136배(load 109). 리스트 원소가 여전히 PyFloat 이라 저장마다 + 박싱이 일어난다 — 고정 길이 배열을 C 배열로 내리는 것이 다음 단계다. + +### 6.1 단계 1 의 현재 상태 (2026-10-03, #16) + +- **위치:** `python-multiplatform-ksp/src/main/python/typedpython/`, 테스트는 + `python-multiplatform-ksp/src/test/python/`. 사용자 지시("ksp랑 통합")에 따라 KSP 모듈 안에 둔다. + KSP 프로세서는 Kotlin 심볼만 보고 파이썬 파일 변경으로는 다시 돌지 않으므로, **실행**은 1b 에서 + Gradle 태스크가 맡는다 — KSP 모듈은 소스와 배포(jar 리소스)의 자리다. +- **구현된 규칙:** SPEC N-6. Pyrefly 의 `Any` 유출 kind 6개를 에러로 올리고, Pyrefly 의 + per-expression 타입 리포트(`--report-pysa`)로 사용자 코드 안의 `Any` 식을 찾는다(`any-flow`). + `cast()` 인자, 버려지는 식 문장의 값, 함수 시그니처 자체는 제외한다. +- **재할당·혼합 컨테이너 (#22):** Pyrefly 1.3.2 는 둘 다 보고하지 않고, 타입 리포트에 대입 대상의 + 타입이 없다. 대신 이름을 *읽는* 위치의 타입은 기록한다. 그래서 대입문 바로 뒤에 같은 줄로 + `; x` 를 붙인 사본을 만들어(줄 번호 불변) 한 번 더 돌리고, 그 위치의 타입으로 판정한다. 두 번 + 이상 대입되는 이름이나 컨테이너 디스플레이가 있는 파일에서만 돈다. `for` 대상과 컴프리헨션 + 변수는 아직 탐침하지 않는다. +- **측정:** 1800줄 합성 모듈에서 plain `pyrefly check` 160–235 ms, 게이트 1.3–1.9 s — 두 번 잰 비율이 + 모두 약 8배다. 탐침 패스가 붙은 뒤로는 약 15배(232 ms → 3.6 s, load 41)다 — 합성 모듈의 + `total += ...` 가 탐침 대상이라 두 번째 실행이 돈다. 단독 측정이 아니다(두 번째는 load average 18.6). 절대값보다 비율을 기준으로 본다. 차이의 + 대부분은 타입 리포트가 typeshed 사본까지 디스크에 쓰는 비용으로 보인다(추정, 미분해). 1b 에서 + 빌드마다 돌리기 전에 줄일 대상이다. +- **고정:** Pyrefly `==1.3.2`. 타입 리포트는 Pyrefly 가 버전을 매기는 포맷이지 공개 API 가 아니다. + ## 7. 위험 - **Pyrefly 의 외부 API 안정성.** 라이브러리 임베드는 확인 전까지 미룬다 (§4.1). diff --git a/python-multiplatform-ksp/.gitignore b/python-multiplatform-ksp/.gitignore new file mode 100644 index 000000000..4dc423b60 --- /dev/null +++ b/python-multiplatform-ksp/.gitignore @@ -0,0 +1,5 @@ +.venv/ +__pycache__/ +.pytest_cache/ +*.egg-info/ +.pytest_tmp/ diff --git a/python-multiplatform-ksp/pyproject.toml b/python-multiplatform-ksp/pyproject.toml new file mode 100644 index 000000000..e2b7853fa --- /dev/null +++ b/python-multiplatform-ksp/pyproject.toml @@ -0,0 +1,31 @@ +# TypedPython soundness gate (docs/design/typedpython.md §4.2). +# The Kotlin processor in this module is built by Gradle; this file only describes the Python +# package that lives next to it in src/main/python. +[project] +name = "typedpython" +version = "0.1.0" +requires-python = ">=3.13" +# Pinned exactly: the gate reads Pyrefly's per-expression type report (--report-pysa), whose +# format is versioned by Pyrefly, not by a public API. +dependencies = ["pyrefly==1.3.2"] + +[project.scripts] +typedpython = "typedpython.__main__:cli" + +[build-system] +requires = ["hatchling"] +build-backend = "hatchling.build" + +[tool.hatch.build.targets.wheel] +packages = ["src/main/python/typedpython"] + +[project.optional-dependencies] +test = ["pytest"] +# `typedpython demo` / compiler.compile_module: Cython and a C compiler for the host. +compile = ["cython>=3.1", "setuptools"] + +[tool.pytest.ini_options] +testpaths = ["src/test/python"] +pythonpath = ["src/main/python"] +addopts = "--basetemp=.pytest_tmp" +markers = ["measurement: records cost; asserts only sanity bounds"] diff --git a/python-multiplatform-ksp/src/main/python/typedpython/__init__.py b/python-multiplatform-ksp/src/main/python/typedpython/__init__.py new file mode 100644 index 000000000..47886a2a1 --- /dev/null +++ b/python-multiplatform-ksp/src/main/python/typedpython/__init__.py @@ -0,0 +1,9 @@ +"""TypedPython — static checking for the Python code of a PythonMultiplatform project. + +See docs/design/typedpython.md. This package is the soundness gate (§4.2). +""" + + +def compiled[T](target: T) -> T: + """Mark a function or class for native compilation. In CPython it does nothing.""" + return target diff --git a/python-multiplatform-ksp/src/main/python/typedpython/__main__.py b/python-multiplatform-ksp/src/main/python/typedpython/__main__.py new file mode 100644 index 000000000..ec0f0ddfa --- /dev/null +++ b/python-multiplatform-ksp/src/main/python/typedpython/__main__.py @@ -0,0 +1,101 @@ +"""`python -m typedpython check [--mode checked|compiled] [--format text|json] PATHS...` + +Exit status: 0 no errors (warnings allowed), 1 errors, 2 the gate itself could not run. +""" +import argparse +import json +import sys +from pathlib import Path + +from typedpython import gate +from typedpython.pyrefly import PyreflyError + +EXIT_CLEAN, EXIT_ERRORS, EXIT_FAILURE = 0, 1, 2 + + +def main(argv: list[str]) -> int: + parser = argparse.ArgumentParser(prog="typedpython") + commands = parser.add_subparsers(dest="command", required=True) + check = commands.add_parser("check", help="type-check user Python code") + check.add_argument("paths", nargs="+", type=Path) + check.add_argument("--mode", choices=["checked", "compiled"], default="checked") + check.add_argument("--format", choices=["text", "json"], default="text") + check.add_argument("--search-path", action="append", type=Path, default=[], + help="directory to resolve imports from (repeatable)") + demo = commands.add_parser( + "demo", help="compile a module, run its main() interpreted and compiled, and compare", + ) + demo.add_argument("module", type=Path) + demo.add_argument("argv", nargs="*", help="sys.argv[1:] for the module's main()") + demo.add_argument("--out", type=Path, default=Path("build/typedpython"), + help="where the extension is built (default: build/typedpython)") + args = parser.parse_args(argv) + + if args.command == "demo": + return _demo(args.module, args.argv, args.out) + + try: + found = gate.check(args.paths, mode=args.mode, search_paths=args.search_path) + except PyreflyError as e: + print(f"typedpython: {e}", file=sys.stderr) + return EXIT_FAILURE + + if args.format == "json": + print(json.dumps([d.to_json() for d in found], indent=2)) + else: + for d in found: + print(f"{d.severity.upper()} {d.path}:{d.line}:{d.column}: {d.message} [{d.rule}]") + + return EXIT_ERRORS if any(d.severity == "error" for d in found) else EXIT_CLEAN + + +def _demo(module: Path, module_argv: list[str], out: Path) -> int: + import contextlib + import io + import time + + from typedpython import compiler + + try: + built = compiler.compile_module(module, out / module.stem) + except (compiler.CompileError, PyreflyError) as e: + print(f"typedpython: {e}", file=sys.stderr) + return EXIT_ERRORS if isinstance(e, compiler.CompileError) else EXIT_FAILURE + + def run(load: object) -> tuple[str, float]: + mod = load() # type: ignore[operator] + saved = sys.argv + sys.argv = [str(module), *module_argv] + out_text = io.StringIO() + try: + with contextlib.redirect_stdout(out_text): + start = time.perf_counter() + mod.main() + elapsed = time.perf_counter() - start + finally: + sys.argv = saved + return out_text.getvalue(), elapsed + + interpreted, t_i = run(built.load_interpreted) + compiled, t_c = run(built.load) + print(f"compiled {module.name} -> {built.extension.name}") + print(f"typed functions: {', '.join(built.typed_functions) or '(none)'}") + print(f"interpreted: {t_i * 1000:.1f} ms compiled: {t_c * 1000:.1f} ms " + f"speedup {t_i / t_c:.1f}x") + if interpreted != compiled: + print("output differs:") + print(f" interpreted: {interpreted!r}") + print(f" compiled: {compiled!r}") + return EXIT_ERRORS + print("identical output:") + print(compiled, end="") + return EXIT_CLEAN + + +def cli() -> None: + """Console-script entry point (`typedpython check ...`).""" + sys.exit(main(sys.argv[1:])) + + +if __name__ == "__main__": + cli() diff --git a/python-multiplatform-ksp/src/main/python/typedpython/compiler.py b/python-multiplatform-ksp/src/main/python/typedpython/compiler.py new file mode 100644 index 000000000..c3569da9a --- /dev/null +++ b/python-multiplatform-ksp/src/main/python/typedpython/compiler.py @@ -0,0 +1,541 @@ +"""`@typedpython.compiled` → Cython pure-Python mode → a CPython extension (design §4.3, stage 1). + +The contract is **same results as CPython** (INTENT §1.4). C types are used only where that +holds: + +- `float` locals and parameters become C `double` — the same IEEE double CPython uses. +- An `int` parameter becomes a C `long long` only after the wrapper has checked, *before the body + runs*, that the argument fits; otherwise the call goes to the interpreted function. No effect has + happened yet, so taking that path is always safe. +- `int` locals of a **pure** function (no stores outside its own locals, no calls except to pure + builtins, `math`, and other pure compiled functions) are C `long long` with overflow checking. + On overflow the whole call is redone by the interpreted function: since nothing outside the + function was touched, redoing it is unobservable. That is how `int` keeps arbitrary precision. +- `int` locals of an impure function stay Python objects, except `range` loop indices whose bounds + are themselves C-typed: an index never exceeds its bound, so it cannot overflow. +- An argument whose runtime type is not exactly the declared one (an `int` passed to a `float` + parameter, a `bool` to an `int` one) also takes the interpreted path, since CPython would + compute with the value it was given. + +Everything else in the module is compiled by Cython as ordinary Python, which it supports +unchanged. +""" +import ast +import importlib.machinery +import importlib.util +import json +import subprocess +import sys +import tempfile +import textwrap +from dataclasses import dataclass, field +from pathlib import Path +from types import ModuleType + +from typedpython import gate, pyrefly, rebinding + +MARKER = "# typedpython: compiled" +PURE_BUILTINS = {"range", "len", "abs", "min", "max", "float", "int", "bool", "round", "divmod"} +C_TYPE = {"int": "cython.longlong", "float": "cython.double", "bool": "cython.bint", "list": "list"} +INTERPRETED_SUFFIX = "_typedpython_interpreted" + + +class CompileError(RuntimeError): + pass + + +@dataclass +class Built: + source: Path + extension: Path + typed_functions: list[str] = field(default_factory=list) + + def load(self) -> ModuleType: + name = self.source.stem + loader = importlib.machinery.ExtensionFileLoader(name, str(self.extension)) + spec = importlib.util.spec_from_file_location(name, str(self.extension), loader=loader) + assert spec is not None + module = importlib.util.module_from_spec(spec) + loader.exec_module(module) + return module + + def load_interpreted(self) -> ModuleType: + return _load_source(self.source.stem + INTERPRETED_SUFFIX, self.source) + + +def compile_module(path: Path, out_dir: Path) -> Built: + path = Path(path).resolve() + out_dir = Path(out_dir).resolve() + errors = [d for d in gate.check([path], mode="compiled") if d.severity == "error"] + if errors: + listing = "\n".join(f"{d.path}:{d.line}:{d.column}: {d.message} [{d.rule}]" for d in errors) + raise CompileError(f"{path.name} does not pass the gate in compiled mode:\n{listing}") + + source = path.read_text() + tree = ast.parse(source, filename=str(path)) + targets = _targets(tree, source) + local_types = _local_types(path, source, tree, targets) + + plans = [_plan(fn, local_types.get(fn.name, {}), {t.name for t in targets}) for fn in targets] + pure = _pure_fixpoint(plans) + generated = _emit(source, tree, plans, pure) + + out_dir.mkdir(parents=True, exist_ok=True) + interpreted = out_dir / (path.stem + INTERPRETED_SUFFIX + ".py") + interpreted.write_text(source) + extension = _build(path.stem, generated, out_dir) + return Built(path, extension, [p.name for p in plans]) + + +# --- what to compile ----------------------------------------------------------------------------- + +def _targets(tree: ast.Module, source: str) -> list[ast.FunctionDef]: + first = next((line.strip() for line in source.splitlines() if line.strip()), "") + whole_module = first == MARKER + return [ + node for node in tree.body + if isinstance(node, ast.FunctionDef) + and (whole_module or any(_is_compiled_decorator(d) for d in node.decorator_list)) + ] + + +def _is_compiled_decorator(node: ast.expr) -> bool: + return (isinstance(node, ast.Name) and node.id == "compiled") or ( + isinstance(node, ast.Attribute) and node.attr == "compiled" + and isinstance(node.value, ast.Name) and node.value.id == "typedpython" + ) + + +def _local_types( + path: Path, source: str, tree: ast.Module, targets: list[ast.FunctionDef], +) -> dict[str, dict[str, set[str]]]: + """Inferred types of each target function's assigned locals, from Pyrefly (via probes).""" + scopes = rebinding._scopes(tree) + wanted_scopes = {scopes.index(fn) for fn in targets} + wanted = [a for a in rebinding.assignments(tree) if a.scope in wanted_scopes] + if not wanted: + return {} + probed, probes = rebinding.probe(source, wanted) + with tempfile.TemporaryDirectory(prefix="typedpython-compile-") as work: + copies = gate._mirror(Path(work), {str(path): probed}) + roots = pyrefly.import_roots([str(path)]) + [copy] = copies + report = pyrefly.run([Path(copy)], [Path(work) / "0", Path(copy).parent, *roots]) + types = report.expression_types.get(copy, {}) + + found: dict[str, dict[str, set[str]]] = {} + for p in probes: + spelled = types.get((p.line, p.column)) + if spelled is None or p.assignment is None: + continue + function = scopes[p.assignment.scope] + assert isinstance(function, ast.FunctionDef) + found.setdefault(function.name, {}).setdefault(p.name, set()).add(rebinding._base(spelled)) + return found + + +# --- planning ------------------------------------------------------------------------------------ + +@dataclass +class Plan: + name: str + node: ast.FunctionDef + params: dict[str, str] # parameter -> "int" | "float" | "bool" + locals: dict[str, str] # local -> "int" | "float" | "bool" + safe_indices: set[str] # int locals that are bounded range indices + returns: str | None + impure_reason: str | None + calls: set[str] # names of other module functions it calls + float_lists: set[str] = field(default_factory=set) # list[float] params unboxed inside + + +def _plan(fn: ast.FunctionDef, inferred: dict[str, set[str]], module_functions: set[str]) -> Plan: + params = { + a.arg: _scalar(a.annotation) for a in fn.args.args + if _scalar(a.annotation) is not None + } + params = {k: v for k, v in params.items() if v is not None} + annotated = { + s.target.id: _scalar(s.annotation) for s in ast.walk(fn) + if isinstance(s, ast.AnnAssign) and isinstance(s.target, ast.Name) + } + locals_: dict[str, str] = {k: v for k, v in annotated.items() if v is not None} + for name, kinds in inferred.items(): + if name in params or name in locals_ or len(kinds) != 1: + continue + kind = next(iter(kinds)) + if kind in C_TYPE: + locals_[name] = kind + + loop_targets: set[str] = set() + for node in ast.walk(fn): + if isinstance(node, ast.For) and isinstance(node.target, ast.Name) and _is_range(node.iter): + loop_targets.add(node.target.id) + locals_.setdefault(node.target.id, "int") + # A loop target assigned anywhere else is not a plain index. + reassigned = { + t.id for node in ast.walk(fn) if isinstance(node, (ast.Assign, ast.AugAssign)) + for t in ast.walk(node) if isinstance(t, ast.Name) and isinstance(t.ctx, ast.Store) + } + c_ints = {k for k, v in {**params, **locals_}.items() if v == "int"} + safe = { + node.target.id for node in ast.walk(fn) + if isinstance(node, ast.For) and isinstance(node.target, ast.Name) + and node.target.id in loop_targets and node.target.id not in reassigned + and _is_range(node.iter) and all(_bounded(arg, c_ints) for arg in node.iter.args) # type: ignore[attr-defined] + } + + impure, calls = _impurity(fn, module_functions) + plan = Plan(fn.name, fn, params, locals_, safe, _scalar(fn.returns), impure, calls) + plan.float_lists = _float_lists(fn, plan) + return plan + + +def _float_lists(fn: ast.FunctionDef, plan: "Plan") -> set[str]: + """`list[float]` parameters whose elements stay exact floats for the whole call. + + The wrapper checks at entry that every element is exactly a `float`. Inside, that stays true if + the function hands the lists to nothing (no calls except pure builtins and `math`) and every + value it stores into them is statically a float. Then each element load can be a C double. + """ + candidates = { + a.arg for a in fn.args.args + if isinstance(a.annotation, ast.Subscript) and _is_list(a.annotation) + and isinstance(a.annotation.slice, ast.Name) and a.annotation.slice.id == "float" + } + if not candidates or plan.calls: + return set() + if plan.impure_reason not in (None, "stores into an attribute or element"): + return set() + doubles = {k for k, v in {**plan.params, **plan.locals}.items() if v == "float"} + for node in ast.walk(fn): + if isinstance(node, (ast.Assign, ast.AugAssign)): + targets = node.targets if isinstance(node, ast.Assign) else [node.target] + for t in targets: + if isinstance(t, ast.Subscript) and isinstance(t.value, ast.Name): + if t.value.id not in candidates: + return set() + value_is_float = _is_float(node.value, doubles, candidates) or ( + isinstance(node, ast.AugAssign) and isinstance(node.op, (ast.Add, ast.Sub, ast.Mult, ast.Div)) + ) + if not value_is_float: + return set() + elif isinstance(t, (ast.Subscript, ast.Attribute)): + return set() + if isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute) \ + and isinstance(node.func.value, ast.Name) and node.func.value.id in candidates: + return set() # a method call (append, sort, ...) on the list + # Every use of a candidate is `name[...]` or `len(name)`: it is never passed on whole. + subscripted = { + id(n.value) for n in ast.walk(fn) + if isinstance(n, ast.Subscript) and isinstance(n.value, ast.Name) + } | { # len() reads the list without touching its elements + id(n.args[0]) for n in ast.walk(fn) + if isinstance(n, ast.Call) and isinstance(n.func, ast.Name) and n.func.id == "len" + and len(n.args) == 1 and isinstance(n.args[0], ast.Name) + } + for n in ast.walk(fn): + if isinstance(n, ast.Name) and n.id in candidates and id(n) not in subscripted: + return set() + return candidates + + +FLOAT_FUNCTIONS = {"sqrt", "exp", "log", "sin", "cos", "tan", "atan2", "hypot", "fabs", "pow"} + + +def _is_float(node: ast.expr, doubles: set[str], float_lists: set[str]) -> bool: + """Statically a Python float (never an int or complex), by Python's arithmetic rules.""" + if isinstance(node, ast.Constant): + return type(node.value) is float + if isinstance(node, ast.Name): + return node.id in doubles + if isinstance(node, ast.Subscript) and isinstance(node.value, ast.Name): + return node.value.id in float_lists + if isinstance(node, ast.UnaryOp) and isinstance(node.op, (ast.USub, ast.UAdd)): + return _is_float(node.operand, doubles, float_lists) + if isinstance(node, ast.BinOp) and isinstance(node.op, (ast.Add, ast.Sub, ast.Mult, ast.Div)): + return _is_float(node.left, doubles, float_lists) or _is_float(node.right, doubles, float_lists) + if isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute) \ + and isinstance(node.func.value, ast.Name) and node.func.value.id == "math": + return node.func.attr in FLOAT_FUNCTIONS + return False + + +def _scalar(annotation: ast.expr | None) -> str | None: + if isinstance(annotation, ast.Name) and annotation.id in C_TYPE: + return annotation.id + if _is_list(annotation): + return "list" + return None + + +def _is_list(annotation: ast.expr | None) -> bool: + """`list` or `list[...]`: typed as Cython's builtin list, so indexing is a direct load.""" + if isinstance(annotation, ast.Name): + return annotation.id == "list" + return isinstance(annotation, ast.Subscript) and isinstance(annotation.value, ast.Name) \ + and annotation.value.id == "list" + + +def _is_range(node: ast.expr) -> bool: + return isinstance(node, ast.Call) and isinstance(node.func, ast.Name) and node.func.id == "range" \ + and not node.keywords and 1 <= len(node.args) <= 3 + + +def _bounded(node: ast.expr, c_ints: set[str]) -> bool: + """A range bound that is itself a checked C integer, a literal, or one of those plus/minus a literal.""" + if isinstance(node, ast.Constant) and type(node.value) is int: + return -2**62 < node.value < 2**62 + if isinstance(node, ast.Name): + return node.id in c_ints + if isinstance(node, ast.BinOp) and isinstance(node.op, (ast.Add, ast.Sub)): + return _bounded(node.left, c_ints) and isinstance(node.right, ast.Constant) \ + and type(node.right.value) is int and abs(node.right.value) < 2**30 + if isinstance(node, ast.Call) and isinstance(node.func, ast.Name) and node.func.id == "len": + return True + return False + + +def _impurity(fn: ast.FunctionDef, module_functions: set[str]) -> tuple[str | None, set[str]]: + calls: set[str] = set() + for node in ast.walk(fn): + if isinstance(node, (ast.Global, ast.Nonlocal)): + return f"declares {type(node).__name__.lower()}", calls + if isinstance(node, (ast.Yield, ast.YieldFrom, ast.Await)): + return "is a generator or coroutine", calls + if isinstance(node, (ast.Assign, ast.AugAssign, ast.AnnAssign, ast.Delete)): + targets = node.targets if isinstance(node, (ast.Assign, ast.Delete)) else [node.target] + for t in targets: + if any(isinstance(n, (ast.Attribute, ast.Subscript)) for n in ast.walk(t)): + return "stores into an attribute or element", calls + if isinstance(node, ast.Call): + f = node.func + if isinstance(f, ast.Name) and f.id in PURE_BUILTINS: + continue + if isinstance(f, ast.Attribute) and isinstance(f.value, ast.Name) and f.value.id == "math": + continue + if isinstance(f, ast.Name) and f.id in module_functions: + calls.add(f.id) + continue + return f"calls `{ast.unparse(f)}`", calls + return None, calls + + +def _pure_fixpoint(plans: list[Plan]) -> set[str]: + pure = {p.name for p in plans if p.impure_reason is None} + changed = True + while changed: + changed = False + for p in plans: + if p.name in pure and not p.calls <= pure: + pure.discard(p.name) + changed = True + return pure + + +# --- emitting ------------------------------------------------------------------------------------ + +def _emit(source: str, tree: ast.Module, plans: list[Plan], pure: set[str]) -> str: + lines = source.splitlines(keepends=True) + stdlib_math = any( + isinstance(n, ast.Import) and any(a.name == "math" and a.asname is None for a in n.names) + for n in tree.body + ) and not any( + isinstance(n, ast.Name) and n.id == "math" and isinstance(n.ctx, ast.Store) for n in ast.walk(tree) + ) + # Replace each target from the bottom up so earlier line numbers stay valid. + for plan in sorted(plans, key=lambda p: p.node.lineno, reverse=True): + start = (plan.node.decorator_list[0].lineno if plan.node.decorator_list else plan.node.lineno) - 1 + end = plan.node.end_lineno + assert end is not None + lines[start:end] = [_function( + plan, plan.name in pure, "".join(lines[plan.node.lineno - 1:end]), stdlib_math, + )] + + header = textwrap.dedent(f'''\ + import cython + import importlib.util as _tp_importlib_util + import os as _tp_os + + __typedpython_deopts__ = 0 + _tp_interpreted_module = None + + + def _tp_interpreted(): + global _tp_interpreted_module + if _tp_interpreted_module is None: + here = _tp_os.path.dirname(__file__) + stem = __name__.rsplit(".", 1)[-1] + path = _tp_os.path.join(here, stem + "{INTERPRETED_SUFFIX}.py") + spec = _tp_importlib_util.spec_from_file_location(stem + "{INTERPRETED_SUFFIX}", path) + module = _tp_importlib_util.module_from_spec(spec) + spec.loader.exec_module(module) + _tp_interpreted_module = module + return _tp_interpreted_module + + + from cython.cimports.libc.math import sqrt as _tp_libc_sqrt + + + @cython.cfunc + @cython.inline + @cython.locals(x=cython.double) + @cython.returns(cython.double) + @cython.exceptval(-1.0, check=True) + def _tp_sqrt(x): + # math.sqrt raises for negatives (NaN and inf pass through); C sqrt would return NaN. + if x < 0.0: + raise ValueError("math domain error") + return _tp_libc_sqrt(x) + + + def _tp_fallback(name, args): + global __typedpython_deopts__ + __typedpython_deopts__ += 1 + return getattr(_tp_interpreted(), name)(*args) + + ''') + return header + "".join(lines) + + +def _function(plan: Plan, is_pure: bool, original: str, stdlib_math: bool) -> str: + """The C-typed body as `_tp_c_` plus a Python-visible wrapper with the original name.""" + fn = plan.node + names = [a.arg for a in fn.args.args] + simple = not (fn.args.vararg or fn.args.kwarg or fn.args.kwonlyargs or fn.args.posonlyargs + or fn.args.defaults) + if not simple: + return original # compiled by Cython as plain Python; typed handling needs a plain signature + + c_locals = dict(plan.params) + for name, kind in plan.locals.items(): + if name in c_locals: + continue + if kind == "int" and not is_pure and name not in plan.safe_indices: + continue # stays a Python int: an impure function cannot be redone on overflow + c_locals[name] = kind + + decorators = ["@cython.ccall"] + if c_locals: + decorators.append("@cython.locals(" + ", ".join(f"{k}={C_TYPE[v]}" for k, v in c_locals.items()) + ")") + if plan.returns in C_TYPE: + decorators.append(f"@cython.returns({C_TYPE[plan.returns]})") + if is_pure: + decorators.append("@cython.overflowcheck(True)") + + body = ast.parse(textwrap.dedent(original)).body[0] + assert isinstance(body, ast.FunctionDef) + body = _Unbox(plan.float_lists, stdlib_math).visit(body) + ast.fix_missing_locations(body) + body.name = f"_tp_c_{plan.name}" + body.decorator_list = [] + for arg in body.args.args: + arg.annotation = None + body.returns = None + c_function = "\n".join(decorators) + "\n" + ast.unparse(body) + "\n" + + guards = [] + for name, kind in plan.params.items(): + guards.append(f"type({name}) is not {kind}") + if name in plan.float_lists: + guards.append(f"not all(type(_tp_e) is float for _tp_e in {name})") + if kind == "int": + guards.append(f"not (-9223372036854775808 <= {name} <= 9223372036854775807)") + args = ", ".join(names) + call_args = f"({args},)" if len(names) == 1 else f"({args})" + guard = " or ".join(guards) or "False" + wrapper = [f"def {plan.name}({args}):"] + wrapper.append(f" if {guard}:") + wrapper.append(f" return _tp_fallback({plan.name!r}, {call_args})") + if is_pure: + wrapper.append(" try:") + wrapper.append(f" return _tp_c_{plan.name}({args})") + wrapper.append(" except OverflowError:") + wrapper.append(f" return _tp_fallback({plan.name!r}, {call_args})") + else: + wrapper.append(f" return _tp_c_{plan.name}({args})") + return c_function + "\n\n" + "\n".join(wrapper) + "\n" + + +class _Unbox(ast.NodeTransformer): + """Element loads of float-element lists become C doubles; `math.sqrt` becomes C `sqrt`.""" + + def __init__(self, float_lists: set[str], stdlib_math: bool) -> None: + self.float_lists = float_lists + self.stdlib_math = stdlib_math + + def visit_AugAssign(self, node: ast.AugAssign) -> ast.stmt: + node = self.generic_visit(node) # type: ignore[assignment] + t = node.target + if isinstance(t, ast.Subscript) and isinstance(t.value, ast.Name) and t.value.id in self.float_lists: + load = ast.Subscript(ast.Name(t.value.id, ast.Load()), t.slice, ast.Load()) + # floats are immutable, so `a[i] -= e` is exactly `a[i] = a[i] - e` + return ast.Assign([t], ast.BinOp(self._double(load), node.op, node.value)) + return node + + def visit_Subscript(self, node: ast.Subscript) -> ast.expr: + node = self.generic_visit(node) # type: ignore[assignment] + if isinstance(node.ctx, ast.Load) and isinstance(node.value, ast.Name) \ + and node.value.id in self.float_lists: + return self._double(node) + return node + + def visit_Call(self, node: ast.Call) -> ast.expr: + node = self.generic_visit(node) # type: ignore[assignment] + f = node.func + if self.stdlib_math and isinstance(f, ast.Attribute) and isinstance(f.value, ast.Name) \ + and f.value.id == "math" and f.attr == "sqrt" and len(node.args) == 1 and not node.keywords: + return ast.Call(ast.Name("_tp_sqrt", ast.Load()), node.args, []) + return node + + @staticmethod + def _double(node: ast.expr) -> ast.expr: + return ast.Call( + ast.Attribute(ast.Name("cython", ast.Load()), "cast", ast.Load()), + [ast.Attribute(ast.Name("cython", ast.Load()), "double", ast.Load()), node], [], + ) + + +# --- building ------------------------------------------------------------------------------------ + +BUILD_SCRIPT = """ +import json, sys +from setuptools import Distribution, Extension +from Cython.Build import cythonize + +cfg = json.loads(sys.argv[1]) +extensions = cythonize( + [Extension(cfg["name"], [cfg["source"]], extra_compile_args=["-O3"])], + build_dir=cfg["build"], quiet=True, language_level=3, + compiler_directives={"annotation_typing": False, "binding": False}, +) +dist = Distribution({"ext_modules": extensions}) +cmd = dist.get_command_obj("build_ext") +cmd.build_lib = cfg["out"] +cmd.build_temp = cfg["build"] + "/tmp" +cmd.ensure_finalized() +cmd.run() +""" + + +def _build(name: str, generated: str, out_dir: Path) -> Path: + build = out_dir / "build" + build.mkdir(parents=True, exist_ok=True) + source = build / f"{name}.py" + source.write_text(generated) + cfg = json.dumps({"name": name, "source": str(source), "build": str(build), "out": str(out_dir)}) + result = subprocess.run([sys.executable, "-c", BUILD_SCRIPT, cfg], capture_output=True, text=True) + if result.returncode != 0: + raise CompileError(f"Cython build of {name} failed:\n{result.stdout}\n{result.stderr}") + built = sorted(out_dir.glob(f"{name}.*.so")) + sorted(out_dir.glob(f"{name}.*.pyd")) + if not built: + raise CompileError(f"Cython build of {name} produced no extension:\n{result.stdout}") + return built[0] + + +def _load_source(name: str, path: Path) -> ModuleType: + spec = importlib.util.spec_from_file_location(name, str(path)) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module diff --git a/python-multiplatform-ksp/src/main/python/typedpython/diagnostic.py b/python-multiplatform-ksp/src/main/python/typedpython/diagnostic.py new file mode 100644 index 000000000..2ec322edc --- /dev/null +++ b/python-multiplatform-ksp/src/main/python/typedpython/diagnostic.py @@ -0,0 +1,17 @@ +from dataclasses import dataclass, asdict +from typing import Literal + +Severity = Literal["error", "warning"] + + +@dataclass(frozen=True, order=True) +class Diagnostic: + path: str + line: int + column: int + rule: str + severity: Severity + message: str + + def to_json(self) -> dict[str, object]: + return asdict(self) diff --git a/python-multiplatform-ksp/src/main/python/typedpython/forbidden.py b/python-multiplatform-ksp/src/main/python/typedpython/forbidden.py new file mode 100644 index 000000000..cccc1651f --- /dev/null +++ b/python-multiplatform-ksp/src/main/python/typedpython/forbidden.py @@ -0,0 +1,110 @@ +"""The forbidden list (docs/design/typedpython.md §3.1), found by walking the AST. + +These are the dynamic features that make a static type unreliable at compile time. The gate +reports them as warnings for `checked` code and as errors for `compiled` code; this module only +finds them. + +Rebinding a name to a different type and containers that mix element types need inferred types +for assignment targets; they live in typedpython.rebinding. +""" +import ast +from dataclasses import dataclass + +DYNAMIC_ATTRIBUTE_CALLS = {"getattr", "setattr", "delattr"} +# Index of the attribute-name argument in each of the calls above. +NAME_ARGUMENT = 1 + + +@dataclass(frozen=True) +class Finding: + rule: str + line: int + column: int + message: str + + +def scan(source: str, path: str) -> list[Finding]: + tree = ast.parse(source, filename=path) + patchable = _module_level_modules_and_classes(tree) + findings: list[Finding] = [] + + for node in ast.walk(tree): + if isinstance(node, ast.Call) and isinstance(node.func, ast.Name): + findings += _check_call(node, node.func.id, patchable) + elif isinstance(node, (ast.Assign, ast.AugAssign, ast.AnnAssign, ast.Delete)): + for target in _targets(node): + if _is_patch_target(target, patchable): + assert isinstance(target, ast.Attribute) and isinstance(target.value, ast.Name) + findings.append(_finding( + "forbidden/monkeypatch", target, + f"`{target.value.id}.{target.attr}` changes a module or class from outside it", + )) + + return sorted(findings, key=lambda f: (f.line, f.column, f.rule)) + + +def _check_call(call: ast.Call, name: str, patchable: set[str]) -> list[Finding]: + if name in ("eval", "exec"): + return [_finding(f"forbidden/{name}", call, f"`{name}` runs code the checker cannot see")] + if name not in DYNAMIC_ATTRIBUTE_CALLS or len(call.args) <= NAME_ARGUMENT: + return [] + + attribute = call.args[NAME_ARGUMENT] + if not (isinstance(attribute, ast.Constant) and isinstance(attribute.value, str)): + return [_finding( + "forbidden/dynamic-attr", call, + f"`{name}` with a computed attribute name hides the attribute from the checker", + )] + + # A literal name is static, but setattr/delattr on a module or class is still a patch. + receiver = call.args[0] + if name != "getattr" and isinstance(receiver, ast.Name) and receiver.id in patchable: + return [_finding( + "forbidden/monkeypatch", call, + f"`{name}` on `{receiver.id}` changes a module or class from outside it", + )] + return [] + + +def _module_level_modules_and_classes(tree: ast.Module) -> set[str]: + """Names bound at module scope to an imported module/name or to a class definition.""" + names: set[str] = set() + for node in tree.body: + if isinstance(node, ast.Import): + names |= {(alias.asname or alias.name).split(".")[0] for alias in node.names} + elif isinstance(node, ast.ImportFrom): + names |= {alias.asname or alias.name for alias in node.names if alias.name != "*"} + elif isinstance(node, ast.ClassDef): + names.add(node.name) + return names + + +def _targets(node: ast.stmt) -> list[ast.expr]: + if isinstance(node, ast.Assign): + return [t for target in node.targets for t in _flatten(target)] + if isinstance(node, ast.Delete): + return [t for target in node.targets for t in _flatten(target)] + if isinstance(node, (ast.AugAssign, ast.AnnAssign)): + return [node.target] + return [] + + +def _flatten(target: ast.expr) -> list[ast.expr]: + if isinstance(target, (ast.Tuple, ast.List)): + return [t for element in target.elts for t in _flatten(element)] + if isinstance(target, ast.Starred): + return _flatten(target.value) + return [target] + + +def _is_patch_target(target: ast.expr, patchable: set[str]) -> bool: + return ( + isinstance(target, ast.Attribute) + and isinstance(target.value, ast.Name) + and target.value.id in patchable + ) + + +def _finding(rule: str, node: ast.expr, message: str) -> Finding: + # ast columns are 0-based; Pyrefly's (and every editor's) are 1-based. + return Finding(rule, node.lineno, node.col_offset + 1, message) diff --git a/python-multiplatform-ksp/src/main/python/typedpython/gate.py b/python-multiplatform-ksp/src/main/python/typedpython/gate.py new file mode 100644 index 000000000..303b5c19f --- /dev/null +++ b/python-multiplatform-ksp/src/main/python/typedpython/gate.py @@ -0,0 +1,182 @@ +"""The soundness gate (docs/design/typedpython.md §4.2). + +A type checker may fall back to `Any` when it is unsure; a compiler cannot. The gate closes that +gap for the files it is given: + + Pyrefly errors, Any-leak kinds included -> error + an expression of type Any in user code -> error (`any-flow`), unless inside cast() + the §3.1 forbidden list -> warning (`checked`) / error (`compiled`) + +Only the given files are user code. Their dependencies are type-checked by Pyrefly only as far +as their stubs or sources say; nothing inside them is reported. +""" +import ast +import shutil +import tempfile +from collections.abc import Sequence +from pathlib import Path +from typing import Literal + +from typedpython import forbidden, pyrefly, rebinding +from typedpython.diagnostic import Diagnostic, Severity + +Mode = Literal["checked", "compiled"] + +Position = tuple[int, int] + + +def check( + paths: Sequence[Path], + mode: Mode = "checked", + search_paths: Sequence[Path] = (), +) -> list[Diagnostic]: + report = pyrefly.run(paths, search_paths) + diagnostics = [ + Diagnostic(e.path, e.line, e.column, f"pyrefly/{e.kind}", "error", e.message) + for e in report.errors + ] + reported = {(d.path, d.line, d.column) for d in diagnostics} + + forbidden_severity: Severity = "error" if mode == "compiled" else "warning" + for path in (str(Path(p).resolve()) for p in paths): + source = Path(path).read_text() + try: + tree = ast.parse(source, filename=path) + except SyntaxError: + continue # Pyrefly has already reported the parse error + + cast_spans = _cast_argument_spans(tree) + discarded = _discarded_values(tree) + for expr in report.any_expressions: + if expr.path != path or (expr.path, expr.line, expr.column) in reported: + continue + if _is_signature(expr.type): + continue # a callable whose signature mentions Any; its result is judged on its own + if ((expr.line, expr.column), (expr.stop_line, expr.stop_column)) in discarded: + continue # an expression statement's value goes nowhere + if any(_inside((expr.line, expr.column), span) for span in cast_spans): + continue + diagnostics.append(Diagnostic( + path, expr.line, expr.column, "any-flow", "error", + f"this expression has type `{expr.type}`; give it a type, or cast() it " + "where it leaves untyped code", + )) + reported.add((expr.path, expr.line, expr.column)) + + for finding in forbidden.scan(source, path): + diagnostics.append(Diagnostic( + path, finding.line, finding.column, finding.rule, forbidden_severity, + finding.message, + )) + + for d in _rebinding_findings(paths, search_paths): + diagnostics.append(Diagnostic( + d.path, d.line, d.column, d.rule, forbidden_severity, d.message, + )) + + return sorted(diagnostics) + + +def _rebinding_findings( + paths: Sequence[Path], search_paths: Sequence[Path], +) -> list[Diagnostic]: + """Second Pyrefly pass over probed copies; skipped when no file has anything to probe.""" + plans: dict[str, tuple[str, list[rebinding.Probe]]] = {} + for path in (str(Path(p).resolve()) for p in paths): + try: + tree = ast.parse(Path(path).read_text(), filename=path) + except SyntaxError: + continue + wanted = rebinding.candidates(rebinding.assignments(tree)) + if wanted: + plans[path] = rebinding.probe(Path(path).read_text(), wanted) + if not plans: + return [] + + found: list[Diagnostic] = [] + with tempfile.TemporaryDirectory(prefix="typedpython-probe-") as work: + copies = _mirror(Path(work), {o: probed for o, (probed, _) in plans.items()}) + roots = pyrefly.import_roots(list(plans)) + mirrored = sorted({str(Path(c).parent) for c in copies}) # flat files: their copy's dir + report = pyrefly.run( + [Path(c) for c in copies], + [*[Path(work) / str(i) for i in range(len(roots))], *map(Path, mirrored), + *search_paths, *roots], + ) + for copy, original in copies.items(): + types = report.expression_types.get(copy, {}) + for f in rebinding.judge(plans[original][1], types): + found.append(Diagnostic(original, f.line, f.column, f.rule, "warning", f.message)) + return found + + +def _mirror(work: Path, probed: dict[str, str]) -> dict[str, str]: + """Copy each file's package tree under `work`, with the probed text in place of the file. + + The copy keeps the file's path relative to its import root, so relative imports and imports + of sibling modules inside the package resolve to the copied tree. Returns copy -> original. + """ + roots = pyrefly.import_roots(list(probed)) + copies: dict[str, str] = {} + copied: set[Path] = set() + for original, text in probed.items(): + root = pyrefly.import_roots([original])[0] + base = work / str(roots.index(root)) + relative = Path(original).relative_to(root) + if len(relative.parts) > 1: + package = root / relative.parts[0] + if package not in copied: + shutil.copytree( + package, base / relative.parts[0], dirs_exist_ok=True, + ignore=shutil.ignore_patterns("__pycache__", ".*"), + ) + copied.add(package) + target = base / relative + target.parent.mkdir(parents=True, exist_ok=True) + target.write_text(text) + copies[str(target.resolve())] = original + return copies + + +def _cast_argument_spans(tree: ast.Module) -> list[tuple[Position, Position]]: + """Source spans of the value argument of every `cast(T, value)` / `typing.cast(T, value)`.""" + spans = [] + for node in ast.walk(tree): + if not isinstance(node, ast.Call) or len(node.args) != 2: + continue + func = node.func + is_cast = (isinstance(func, ast.Name) and func.id == "cast") or ( + isinstance(func, ast.Attribute) and func.attr == "cast" + and isinstance(func.value, ast.Name) and func.value.id == "typing" + ) + value = node.args[1] + if is_cast and value.end_lineno is not None and value.end_col_offset is not None: + spans.append(( + (value.lineno, value.col_offset + 1), + (value.end_lineno, value.end_col_offset + 1), + )) + return spans + + +def _discarded_values(tree: ast.Module) -> set[tuple[Position, Position]]: + """Spans of values computed and thrown away by an expression statement (`eval(s)` alone).""" + spans = set() + for node in ast.walk(tree): + if isinstance(node, ast.Expr): + value = node.value + if value.end_lineno is not None and value.end_col_offset is not None: + spans.add(( + (value.lineno, value.col_offset + 1), + (value.end_lineno, value.end_col_offset + 1), + )) + return spans + + +def _is_signature(spelled: str) -> bool: + # Pyrefly spells callables as `(x: int) -> str` and overload sets as `Overload[...]`. + return "->" in spelled or spelled.startswith("Overload[") + + +def _inside(position: Position, span: tuple[Position, Position]) -> bool: + start, stop = span + return start <= position <= stop diff --git a/python-multiplatform-ksp/src/main/python/typedpython/py.typed b/python-multiplatform-ksp/src/main/python/typedpython/py.typed new file mode 100644 index 000000000..e69de29bb diff --git a/python-multiplatform-ksp/src/main/python/typedpython/pyrefly.py b/python-multiplatform-ksp/src/main/python/typedpython/pyrefly.py new file mode 100644 index 000000000..d987b06c8 --- /dev/null +++ b/python-multiplatform-ksp/src/main/python/typedpython/pyrefly.py @@ -0,0 +1,181 @@ +"""Running Pyrefly and reading what it reports (docs/design/typedpython.md §4.1). + +Pyrefly is used as a CLI, not embedded. Two outputs are read: + +- its JSON error list, with the Any-leak error kinds raised to `error` by a generated config; +- its per-expression type report (`--report-pysa`), which says which expressions in user code + have type `Any`. That report is a Pyrefly-versioned format, which is why pyproject.toml pins + Pyrefly exactly. +""" +import json +import os +import re +import subprocess +import sys +import tempfile +from collections.abc import Sequence +from dataclasses import dataclass +from pathlib import Path + +# Pyrefly error kinds that mean "an Any got in". Pyrefly leaves several of them off by default. +ANY_LEAK_KINDS = ( + "implicit-any-parameter", + "unannotated-return", + "unknown-variable-type", + "no-any-return-implicit", + "no-any-return-explicit", + "explicit-any", +) + +# How Pyrefly spells an Any in the type report: `typing.Any` for an explicit one and `Unknown` +# for an implicit one, alone or inside a larger type such as `list[typing.Any]`. +ANY_IN_TYPE = re.compile(r"(? (line, column) -> type, for every expression the type report records. + expression_types: dict[str, dict[tuple[int, int], str]] + + +def command() -> list[str]: + """The Pyrefly command line: `TYPEDPYTHON_PYREFLY`, or the copy installed next to this Python.""" + override = os.environ.get("TYPEDPYTHON_PYREFLY") + return [override] if override else [sys.executable, "-m", "pyrefly"] + + +def run(paths: Sequence[Path], search_paths: Sequence[Path] = ()) -> Report: + files = [str(Path(p).resolve()) for p in paths] + with tempfile.TemporaryDirectory(prefix="typedpython-") as work: + config = Path(work) / "pyrefly.toml" + config.write_text(_config([*search_paths, *import_roots(files), OWN_ROOT])) + report_dir = Path(work) / "pysa" + argv = [ + *command(), "check", *files, + "--config", str(config), + "--output-format", "json", + "--summary=none", + "--report-pysa", str(report_dir), + "--report-pysa-format", "json", + ] + try: + result = subprocess.run(argv, capture_output=True, text=True, check=False) + except OSError as e: + raise PyreflyError(f"cannot run {argv[0]}: {e}") from e + + # Exit code 1 only means "type errors found"; anything else is a failure of the tool. + if result.returncode not in (0, 1): + raise PyreflyError(f"pyrefly exited {result.returncode}: {result.stderr.strip()}") + try: + raw_errors = json.loads(result.stdout)["errors"] + except (json.JSONDecodeError, KeyError) as e: + raise PyreflyError(f"unreadable pyrefly output: {result.stdout[:200]!r}") from e + + errors = [ + Error(_resolve(e["path"]), e["line"], e["column"], e["name"], e["description"]) + for e in raw_errors + ] + types = _expression_types(report_dir, set(files)) + return Report(errors, _any_expressions(types), _types_by_start(types)) + + +def import_roots(files: Sequence[str]) -> list[Path]: + """The directory each file is imported from: above its outermost package, else its own. + + The config lives in a temporary directory, and Pyrefly would otherwise take that as the import + root, so a package's own imports (relative or absolute) would not resolve. + """ + roots: list[Path] = [] + for file in files: + root = Path(file).parent + while (root / "__init__.py").exists(): + root = root.parent + if root not in roots: + roots.append(root) + return roots + + +def _config(search_paths: Sequence[Path]) -> str: + lines = [] + if search_paths: + quoted = ", ".join(json.dumps(str(Path(p).resolve())) for p in search_paths) + lines.append(f"search-path = [{quoted}]") + lines.append("[errors]") + lines += [f'{kind} = "error"' for kind in ANY_LEAK_KINDS] + return "\n".join(lines) + "\n" + + +def _expression_types(report_dir: Path, files: set[str]) -> list[AnyExpression]: + """Every typed expression the report records in the given files (typeshed and deps skipped).""" + try: + index = json.loads((report_dir / "pyrefly.pysa.json").read_text()) + except (OSError, json.JSONDecodeError) as e: + raise PyreflyError(f"pyrefly wrote no readable type report: {e}") from e + + found: list[AnyExpression] = [] + for module in index["modules"].values(): + source = module.get("source_path", {}).get("FileSystem") + if source is None or _resolve(source) not in files: + continue # typeshed and dependencies: not user code + table_file = report_dir / "type_of_expressions" / module["info_filename"] + if not table_file.exists(): + continue + table = json.loads(table_file.read_text()) + for function in table["functions"].values(): + types = function["type_table"] + for location, type_index in function["locations"].items(): + found.append(_expression(_resolve(source), location, types[type_index]["string"])) + return found + + +def _any_expressions(expressions: list[AnyExpression]) -> list[AnyExpression]: + return [e for e in expressions if ANY_IN_TYPE.search(e.type)] + + +def _types_by_start(expressions: list[AnyExpression]) -> dict[str, dict[tuple[int, int], str]]: + table: dict[str, dict[tuple[int, int], str]] = {} + for e in expressions: + table.setdefault(e.path, {})[(e.line, e.column)] = e.type + return table + + +def _expression(path: str, location: str, spelled: str) -> AnyExpression: + # "6:9-6:15" -> start line/column, stop line/column (1-based, as in Pyrefly's errors). + start, stop = location.split("-") + line, column = map(int, start.split(":")) + stop_line, stop_column = map(int, stop.split(":")) + return AnyExpression(path, line, column, stop_line, stop_column, spelled) + + +def _resolve(path: str) -> str: + return str(Path(path).resolve()) diff --git a/python-multiplatform-ksp/src/main/python/typedpython/rebinding.py b/python-multiplatform-ksp/src/main/python/typedpython/rebinding.py new file mode 100644 index 000000000..8f26cef8e --- /dev/null +++ b/python-multiplatform-ksp/src/main/python/typedpython/rebinding.py @@ -0,0 +1,258 @@ +"""Rebinding a name to a different type, and mixed-element containers (design §3.1). + +Pyrefly 1.3.2 reports neither, and its per-expression type report has no entries for assignment +targets. It does record every *load* of a name, with the name's type at that point. So the gate +writes a probed copy of the file in which each assignment of interest is followed, on the same +line, by a bare load of each target (`x = 1` becomes `x = 1; x`), runs Pyrefly on the copy, and +reads the target types back at the probe positions. Line numbers do not change, and every column +before the insertion point stays where it was. +""" +import ast +import re +from collections.abc import Iterable +from dataclasses import dataclass, field + +# Container types whose element type should be one type, not a union. Tuples are excluded: a +# fixed-length tuple is typed per slot, so `tuple[int, str]` is not "mixed". +MIXABLE = re.compile(r"^(list|set|frozenset|dict)\[(.*)\]$", re.DOTALL) +LITERAL = re.compile(r"^Literal\[(.*)\]$", re.DOTALL) + + +@dataclass(frozen=True) +class Assignment: + """One assignment to a plain name, in the original source.""" + + scope: int + name: str + line: int + column: int + end_line: int + end_byte: int # ast end_col_offset of the whole statement (UTF-8 bytes) + augmented: bool + display: bool # the value is a list/set/dict display with more than one element + + +@dataclass(frozen=True) +class Probe: + name: str + line: int + column: int + assignment: Assignment | None = field(default=None, compare=False) + + +@dataclass(frozen=True) +class Finding: + rule: str + line: int + column: int + message: str + + +def assignments(tree: ast.Module) -> list[Assignment]: + """Assignments to unannotated names, scope by scope (module, class bodies, functions).""" + found: list[Assignment] = [] + for scope_id, scope in enumerate(_scopes(tree)): + declared = _declared_names(scope) + for stmt in _own_statements(scope): + found += [a for a in _assignments_in(stmt, scope_id) if a.name not in declared] + return found + + +def candidates(found: Iterable[Assignment]) -> list[Assignment]: + """The assignments worth probing: names assigned more than once, and container displays.""" + found = list(found) + counts: dict[tuple[int, str], int] = {} + for a in found: + counts[(a.scope, a.name)] = counts.get((a.scope, a.name), 0) + 1 + return [a for a in found if counts[(a.scope, a.name)] > 1 or a.display] + + +def probe(source: str, wanted: Iterable[Assignment] | None = None) -> tuple[str, list[Probe]]: + """Insert a load of each target after its statement; `wanted=None` probes every assignment.""" + if wanted is None: + wanted = assignments(ast.parse(source)) + lines = source.splitlines(keepends=True) + # Insert right-to-left within a line so earlier offsets stay valid. + by_position: dict[tuple[int, int], list[Assignment]] = {} + for a in wanted: + by_position.setdefault((a.end_line, a.end_byte), []).append(a) + + probes: list[Probe] = [] + for (line_no, end_byte) in sorted(by_position, reverse=True): + encoded = lines[line_no - 1].encode() + head, tail = encoded[:end_byte].decode(), encoded[end_byte:].decode() + inserted = "" + added: list[Probe] = [] + for a in by_position[(line_no, end_byte)]: + inserted += "; " + added.append(Probe(a.name, line_no, len(head) + len(inserted) + 1, a)) + inserted += a.name + lines[line_no - 1] = head + inserted + tail + # Probes already placed further right on this line moved by what was just inserted. + probes = [ + Probe(q.name, q.line, q.column + len(inserted), q.assignment) + if q.line == line_no and q.column > len(head) else q + for q in probes + ] + probes += added + + return "".join(lines), probes + + +def judge(probes: list[Probe], types: dict[tuple[int, int], str]) -> list[Finding]: + """Turn probed types into findings.""" + findings: list[Finding] = [] + history: dict[tuple[int, str], tuple[str, Assignment]] = {} + + for p in sorted(probes, key=lambda p: (p.line, p.column)): + a = p.assignment + spelled = types.get((p.line, p.column)) + if a is None or spelled is None or _is_unknown(spelled): + continue + + if a.display and (mixed := _mixed_element(spelled)): + findings.append(Finding( + "forbidden/mixed-container", a.line, a.column, + f"`{a.name}` is `{spelled}`: its elements mix `{mixed}`; annotate the union " + "on purpose or keep one element type", + )) + + key = (a.scope, a.name) + current = _base(spelled) + if key in history and history[key][0] != current: + before = history[key][0] + findings.append(Finding( + "forbidden/rebinding", a.line, a.column, + f"`{a.name}` was `{before}` and is rebound to `{current}`; annotate it " + f"(`{a.name}: {before} | {current}`) or use another name", + )) + history[key] = (current, a) + + return findings + + +# --- AST helpers ------------------------------------------------------------------------------- + +Scope = ast.Module | ast.ClassDef | ast.FunctionDef | ast.AsyncFunctionDef + + +def _scopes(tree: ast.Module) -> list[Scope]: + return [tree] + [ + n for n in ast.walk(tree) + if isinstance(n, (ast.ClassDef, ast.FunctionDef, ast.AsyncFunctionDef)) + ] + + +def _own_statements(scope: Scope) -> Iterable[ast.stmt]: + """Statements of this scope, descending into blocks but not into nested scopes.""" + stack = list(reversed(scope.body)) + while stack: + stmt = stack.pop() + yield stmt + if isinstance(stmt, (ast.ClassDef, ast.FunctionDef, ast.AsyncFunctionDef)): + continue + for name in ("body", "orelse", "finalbody"): + stack.extend(reversed(getattr(stmt, name, []))) + for handler in getattr(stmt, "handlers", []): + stack.extend(reversed(handler.body)) + for case in getattr(stmt, "cases", []): + stack.extend(reversed(case.body)) + + +def _declared_names(scope: Scope) -> set[str]: + """Names whose type is declared (annotated, parameters) or owned elsewhere (global).""" + names: set[str] = set() + if isinstance(scope, (ast.FunctionDef, ast.AsyncFunctionDef)): + args = scope.args + for arg in [*args.posonlyargs, *args.args, *args.kwonlyargs]: + names.add(arg.arg) + names |= {a.arg for a in (args.vararg, args.kwarg) if a is not None} + for stmt in _own_statements(scope): + if isinstance(stmt, ast.AnnAssign) and isinstance(stmt.target, ast.Name): + names.add(stmt.target.id) + elif isinstance(stmt, (ast.Global, ast.Nonlocal)): + names |= set(stmt.names) + return names + + +def _assignments_in(stmt: ast.stmt, scope_id: int) -> list[Assignment]: + if isinstance(stmt, ast.Assign): + targets = [t for target in stmt.targets for t in _names(target)] + display = len(stmt.targets) == 1 and isinstance(stmt.targets[0], ast.Name) \ + and _is_display(stmt.value) + augmented = False + elif isinstance(stmt, ast.AugAssign) and isinstance(stmt.target, ast.Name): + targets, display, augmented = [stmt.target], False, True + else: + return [] + assert stmt.end_lineno is not None and stmt.end_col_offset is not None + return [ + Assignment(scope_id, t.id, t.lineno, t.col_offset + 1, + stmt.end_lineno, stmt.end_col_offset, augmented, display) + for t in targets + ] + + +def _names(target: ast.expr) -> list[ast.Name]: + if isinstance(target, ast.Name): + return [target] + if isinstance(target, (ast.Tuple, ast.List)): + return [n for element in target.elts for n in _names(element)] + if isinstance(target, ast.Starred): + return _names(target.value) + return [] + + +def _is_display(value: ast.expr) -> bool: + if isinstance(value, (ast.List, ast.Set)): + return len(value.elts) > 1 + if isinstance(value, ast.Dict): + return len(value.keys) > 1 + return False + + +# --- type spelling ----------------------------------------------------------------------------- + +def _is_unknown(spelled: str) -> bool: + return spelled in ("Unknown", "typing.Any", "Any") + + +def _base(spelled: str) -> str: + """`Literal[1]` -> `int`, `Literal['a']` -> `str`, `Literal[True]` -> `bool`.""" + m = LITERAL.match(spelled) + if not m: + return spelled + value = m.group(1).strip() + if value in ("True", "False"): + return "bool" + if value[:1] in ("'", '"'): + return "str" + if value[:1] in ("b",) and value[1:2] in ("'", '"'): + return "bytes" + return "int" + + +def _mixed_element(spelled: str) -> str | None: + m = MIXABLE.match(spelled) + if not m: + return None + for argument in _split_top_level(m.group(2)): + if len(_split_top_level(argument, "|")) > 1: + return argument.strip() + return None + + +def _split_top_level(text: str, separator: str = ",") -> list[str]: + parts, depth, current = [], 0, "" + for ch in text: + if ch in "[(": + depth += 1 + elif ch in "])": + depth -= 1 + if ch == separator and depth == 0: + parts.append(current) + current = "" + else: + current += ch + parts.append(current) + return parts diff --git a/python-multiplatform-ksp/src/test/python/conftest.py b/python-multiplatform-ksp/src/test/python/conftest.py new file mode 100644 index 000000000..65b1e5f0a --- /dev/null +++ b/python-multiplatform-ksp/src/test/python/conftest.py @@ -0,0 +1,27 @@ +from pathlib import Path +import textwrap + +import pytest + +FIXTURES = Path(__file__).parent / "fixtures" + + +@pytest.fixture +def write(tmp_path): + """Write a dedented source file under tmp_path and return its path.""" + + def _write(name: str, source: str) -> Path: + path = tmp_path / name + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(textwrap.dedent(source)) + return path + + return _write + + +def rules(diagnostics): + return sorted({d.rule for d in diagnostics}) + + +def errors(diagnostics): + return [d for d in diagnostics if d.severity == "error"] diff --git a/python-multiplatform-ksp/src/test/python/fixtures/deps/untyped_dep.py b/python-multiplatform-ksp/src/test/python/fixtures/deps/untyped_dep.py new file mode 100644 index 000000000..af73da304 --- /dev/null +++ b/python-multiplatform-ksp/src/test/python/fixtures/deps/untyped_dep.py @@ -0,0 +1,7 @@ +# Stands in for a third-party package without stubs. It is deliberately unannotated: the gate +# must not report anything inside it, only where its untyped results flow into user code. +import json + + +def load(text): + return json.loads(text) diff --git a/python-multiplatform-ksp/src/test/python/test_any_flow.py b/python-multiplatform-ksp/src/test/python/test_any_flow.py new file mode 100644 index 000000000..fae430c44 --- /dev/null +++ b/python-multiplatform-ksp/src/test/python/test_any_flow.py @@ -0,0 +1,43 @@ +"""Any reaching user code from an untyped dependency (design §2.3, §4.2).""" +from conftest import FIXTURES, errors +from typedpython import gate + +DEPS = FIXTURES / "deps" + + +def test_untyped_result_flowing_into_user_code_is_an_error(write): + path = write("user.py", """ + from untyped_dep import load + + def total(text: str) -> None: + value = load(text) + print(value) + """) + found = errors(gate.check([path], search_paths=[DEPS])) + flow = [d for d in found if d.rule == "any-flow"] + assert flow, found + assert all(d.path == str(path) for d in flow) + + +def test_cast_at_the_boundary_resolves_it(write): + path = write("user.py", """ + from typing import cast + from untyped_dep import load + + def total(text: str) -> int: + value = cast(int, load(text)) + return value + 1 + """) + assert gate.check([path], search_paths=[DEPS]) == [] + + +def test_the_dependency_itself_is_not_checked(write): + path = write("user.py", """ + from typing import cast + from untyped_dep import load + + def total(text: str) -> int: + return cast(int, load(text)) + """) + found = gate.check([path], search_paths=[DEPS]) + assert not [d for d in found if d.path.endswith("untyped_dep.py")], found diff --git a/python-multiplatform-ksp/src/test/python/test_cli.py b/python-multiplatform-ksp/src/test/python/test_cli.py new file mode 100644 index 000000000..1ac321883 --- /dev/null +++ b/python-multiplatform-ksp/src/test/python/test_cli.py @@ -0,0 +1,90 @@ +"""`python -m typedpython check` exit codes: 0 clean, 1 errors, 2 tool failure.""" +import json +import os +import subprocess +import sys +from pathlib import Path + +SRC = Path(__file__).parents[2] / "main" / "python" + + +def run(*args, env_extra=None): + return run_command("check", *args, env_extra=env_extra) + + +def run_command(command, *args, env_extra=None): + env = dict(os.environ, PYTHONPATH=str(SRC), **(env_extra or {})) + return subprocess.run( + [sys.executable, "-m", "typedpython", command, *map(str, args)], + capture_output=True, text=True, env=env, + ) + + +def test_clean_exits_0(write): + path = write("user.py", "def f(x: int) -> int:\n return x\n") + result = run(path) + assert result.returncode == 0, result.stdout + result.stderr + + +def test_error_exits_1(write): + path = write("user.py", "def f(x):\n return x\n") + result = run(path) + assert result.returncode == 1, result.stdout + result.stderr + assert "implicit-any-parameter" in result.stdout + + +def test_warnings_alone_exit_0(write): + path = write("user.py", "def f(s: str) -> None:\n eval(s)\n") + assert run(path).returncode == 0 + assert run("--mode", "compiled", path).returncode == 1 + + +def test_json_output(write): + path = write("user.py", "def f(x):\n return x\n") + result = run("--format", "json", path) + data = json.loads(result.stdout) + assert {"path", "line", "column", "rule", "severity", "message"} <= set(data[0]) + + +def test_missing_pyrefly_exits_2(write): + path = write("user.py", "def f(x: int) -> int:\n return x\n") + result = run(path, env_extra={"TYPEDPYTHON_PYREFLY": "/nonexistent/pyrefly"}) + assert result.returncode == 2, result.stdout + result.stderr + + +DEMO = """\ +# typedpython: compiled +import sys + + +def total(n: int) -> float: + acc: float = 0.0 + for k in range(n): + acc += k * 0.5 + return acc + + +def main() -> None: + print(f"{total(int(sys.argv[1])):.3f}") +""" + + +def test_demo_runs_both_and_reports_identical_output(write, tmp_path): + import pytest + pytest.importorskip("Cython") + path = write("demo_mod.py", DEMO) + result = run_command("demo", "--out", tmp_path / "build", path, "1000") + assert result.returncode == 0, result.stdout + result.stderr + assert "identical output" in result.stdout + assert "249750.000" in result.stdout + assert "speedup" in result.stdout + + +def test_demo_reports_a_mismatch_as_failure(write, tmp_path, monkeypatch): + import pytest + pytest.importorskip("Cython") + # A module whose output differs run to run cannot be shown identical. + path = write("demo_bad.py", "# typedpython: compiled\nimport time\n\n\ndef main() -> None:\n print(time.perf_counter_ns())\n") + result = run_command("demo", "--out", tmp_path / "build", path) + assert result.returncode == 1, result.stdout + result.stderr + assert "differs" in result.stdout diff --git a/python-multiplatform-ksp/src/test/python/test_compile.py b/python-multiplatform-ksp/src/test/python/test_compile.py new file mode 100644 index 000000000..3ff57d278 --- /dev/null +++ b/python-multiplatform-ksp/src/test/python/test_compile.py @@ -0,0 +1,150 @@ +"""`@typedpython.compiled` → Cython → a desktop CPython extension (design §4.3 stage 1, #23). + +Semantics must not change (INTENT §1.4): every compiled function returns what the interpreted +one returns, including Python's arbitrary-precision `int`. +""" +import importlib +import sys + +import pytest + +import typedpython +from typedpython import compiler + +cython = pytest.importorskip("Cython", reason="compiling needs the [compile] extra") + + +def build(write, tmp_path, name, source): + path = write(f"{name}.py", source) + return compiler.compile_module(path, tmp_path / "out") + + +def test_decorator_is_the_identity_in_cpython(): + def f(x: int) -> int: + return x + + class C: + pass + + assert typedpython.compiled(f) is f + assert typedpython.compiled(C) is C + + +SCALAR = """ + import typedpython + + @typedpython.compiled + def total(n: int) -> float: + acc: float = 0.0 + for k in range(n): + acc += k * 0.5 + return acc +""" + + +def test_compiled_module_is_an_extension_with_the_same_result(write, tmp_path): + built = build(write, tmp_path, "scalar", SCALAR) + assert built.extension.suffix in (".so", ".pyd") + module = built.load() + assert module.__file__ == str(built.extension) + interpreted = built.load_interpreted() + assert module.total(1000) == interpreted.total(1000) + + +FACT = """ + import typedpython + + @typedpython.compiled + def fact(n: int) -> int: + r = 1 + for k in range(2, n + 1): + r *= k + return r +""" + + +def test_int_overflow_promotes_to_a_big_integer(write, tmp_path): + module = build(write, tmp_path, "fact", FACT).load() + assert module.fact(10) == 3628800 + assert module.__typedpython_deopts__ == 0 + assert module.fact(30) == 265252859812191058636308480000000 # past 2**63 + assert module.__typedpython_deopts__ == 1 + + +STORE = """ + import typedpython + + @typedpython.compiled + def store(xs: list[int], n: int) -> None: + xs.append(n) +""" + + +def test_a_big_int_argument_takes_the_interpreted_path_before_any_effect(write, tmp_path): + module = build(write, tmp_path, "store", STORE).load() + xs: list[int] = [] + module.store(xs, 2**70) + assert xs == [2**70] # appended exactly once + + +def test_gate_errors_block_compilation(write, tmp_path): + path = write("bad.py", "import typedpython\n\n@typedpython.compiled\ndef f(x):\n return x\n") + with pytest.raises(compiler.CompileError) as raised: + compiler.compile_module(path, tmp_path / "out") + assert "implicit-any-parameter" in str(raised.value) + + +def test_module_marker_compiles_every_function(write, tmp_path): + built = build(write, tmp_path, "marked", """ + # typedpython: compiled + + def half(n: int) -> float: + return n / 2 + """) + assert "half" in built.typed_functions + + +def test_undecorated_functions_still_work(write, tmp_path): + built = build(write, tmp_path, "mixed", SCALAR + """ + + def plain(n: int) -> int: + return n + 1 + """) + assert built.typed_functions == ["total"] + assert built.load().plain(1) == 2 + + +FLOATS = """ + import math + import typedpython + + @typedpython.compiled + def scale(xs: list[float], k: float) -> None: + for i in range(len(xs)): + xs[i] *= k + xs[i] = math.sqrt(xs[i]) +""" + + +def test_float_lists_keep_python_semantics(write, tmp_path): + built = build(write, tmp_path, "floats", FLOATS) + module, interpreted = built.load(), built.load_interpreted() + a, b = [1.0, 4.0, 9.0], [1.0, 4.0, 9.0] + module.scale(a, 4.0) + interpreted.scale(b, 4.0) + assert a == b and all(type(v) is float for v in a) + + +def test_an_int_element_takes_the_interpreted_path(write, tmp_path): + module = build(write, tmp_path, "floats_int", FLOATS).load() + xs = [1.0, 4] # list[float] admits an int; CPython keeps computing with it + module.scale(xs, 1.0) + assert xs == [1.0, 2.0] + assert module.__typedpython_deopts__ == 1 + + +def test_sqrt_of_a_negative_raises_like_cpython(write, tmp_path): + import pytest + module = build(write, tmp_path, "floats_neg", FLOATS).load() + with pytest.raises(ValueError, match="math domain error"): + module.scale([-1.0], 1.0) diff --git a/python-multiplatform-ksp/src/test/python/test_compile_measurement.py b/python-multiplatform-ksp/src/test/python/test_compile_measurement.py new file mode 100644 index 000000000..c484bbd40 --- /dev/null +++ b/python-multiplatform-ksp/src/test/python/test_compile_measurement.py @@ -0,0 +1,29 @@ +"""What compiling buys on a scalar loop (AGENTS.md §14). Printed, with a sanity bound only.""" +import time + +import pytest + +from test_compile import SCALAR, build + +pytest.importorskip("Cython") + + +@pytest.mark.measurement +def test_scalar_loop_speedup(write, tmp_path): + built = build(write, tmp_path, "scalar", SCALAR) + compiled, interpreted = built.load(), built.load_interpreted() + n = 2_000_000 + + def best(fn): + times = [] + for _ in range(3): + start = time.perf_counter() + fn(n) + times.append(time.perf_counter() - start) + return min(times) + + t_c, t_i = best(compiled.total), best(interpreted.total) + print(f"\n[measure] scalar loop n={n}: interpreted {t_i * 1000:.1f} ms, " + f"compiled {t_c * 1000:.1f} ms, speedup {t_i / t_c:.1f}x") + assert compiled.total(n) == interpreted.total(n) + assert t_c < t_i diff --git a/python-multiplatform-ksp/src/test/python/test_forbidden.py b/python-multiplatform-ksp/src/test/python/test_forbidden.py new file mode 100644 index 000000000..b672a22f8 --- /dev/null +++ b/python-multiplatform-ksp/src/test/python/test_forbidden.py @@ -0,0 +1,54 @@ +"""The §3.1 forbidden list: warning under `checked`, error under `compiled`.""" +import pytest + +from conftest import rules +from typedpython import forbidden, gate + +FLAGGED = { + "forbidden/eval": "def f(s: str) -> None:\n eval(s)\n", + "forbidden/exec": "def f(s: str) -> None:\n exec(s)\n", + "forbidden/dynamic-attr": "def f(o: object, n: str) -> None:\n getattr(o, n)\n", + "forbidden/monkeypatch": "import json\n\ndef f() -> None:\n json.dumps = print\n", +} + + +@pytest.mark.parametrize("rule", sorted(FLAGGED)) +def test_rule_is_found(rule): + assert rule in {f.rule for f in forbidden.scan(FLAGGED[rule], "user.py")} + + +@pytest.mark.parametrize("call", ["setattr(o, n, 1)", "delattr(o, n)"]) +def test_other_dynamic_attribute_calls(call): + source = f"def f(o: object, n: str) -> None:\n {call}\n" + assert {f.rule for f in forbidden.scan(source, "user.py")} == {"forbidden/dynamic-attr"} + + +def test_monkeypatching_a_module_level_class(): + source = ( + "class Greeter:\n pass\n\n" + "def patch() -> None:\n Greeter.hello = 1\n" + ) + assert {f.rule for f in forbidden.scan(source, "user.py")} == {"forbidden/monkeypatch"} + + +@pytest.mark.parametrize("source", [ + "def f(o: object) -> None:\n getattr(o, 'name')\n", + "class C:\n def __init__(self) -> None:\n self.x = 1\n", + "def f() -> None:\n class Local:\n pass\n obj = Local()\n obj.y = 2\n", +]) +def test_static_forms_are_not_flagged(source): + assert forbidden.scan(source, "user.py") == [] + + +def test_findings_carry_a_location(): + [finding] = forbidden.scan("x = 1\neval('x')\n", "user.py") + assert (finding.line, finding.column) == (2, 1) + + +def test_severity_follows_the_mode(write): + path = write("user.py", "def f(s: str) -> None:\n eval(s)\n") + checked = gate.check([path], mode="checked") + compiled = gate.check([path], mode="compiled") + assert [d.severity for d in checked if d.rule == "forbidden/eval"] == ["warning"] + assert [d.severity for d in compiled if d.rule == "forbidden/eval"] == ["error"] + assert rules(checked) == rules(compiled) diff --git a/python-multiplatform-ksp/src/test/python/test_imports.py b/python-multiplatform-ksp/src/test/python/test_imports.py new file mode 100644 index 000000000..4a9a9a876 --- /dev/null +++ b/python-multiplatform-ksp/src/test/python/test_imports.py @@ -0,0 +1,31 @@ +"""User code inside a package must resolve its own imports (defect found 2026-10-03). + +The gate writes its Pyrefly config into a temporary directory. Pyrefly infers the import root +from the config's location, so without help a package's relative and absolute self-imports did +not resolve: every value from them became `Unknown` and failed the gate as `any-flow`. +""" +from conftest import errors +from typedpython import gate + + +def package(write): + write("src/pkg/__init__.py", "") + write("src/pkg/b.py", "def make() -> int:\n return 1\n") + + +def test_relative_import_inside_a_package(write): + package(write) + path = write("src/pkg/a.py", "from .b import make\n\ndef f() -> int:\n x = make()\n return x\n") + assert errors(gate.check([path])) == [] + + +def test_absolute_import_of_the_own_package(write): + package(write) + path = write("src/pkg/c.py", "from pkg.b import make\n\ndef f() -> int:\n return make()\n") + assert errors(gate.check([path])) == [] + + +def test_sibling_module_without_a_package(write): + write("flat/helper.py", "def make() -> int:\n return 1\n") + path = write("flat/main.py", "from helper import make\n\ndef f() -> int:\n return make()\n") + assert errors(gate.check([path])) == [] diff --git a/python-multiplatform-ksp/src/test/python/test_measurement.py b/python-multiplatform-ksp/src/test/python/test_measurement.py new file mode 100644 index 000000000..45e5a15b2 --- /dev/null +++ b/python-multiplatform-ksp/src/test/python/test_measurement.py @@ -0,0 +1,50 @@ +"""What the gate costs on top of a plain `pyrefly check` (design §7, AGENTS.md §14). + +The extra work is the per-expression type report (`--report-pysa`) and the AST pass. The numbers +are printed so a run records them; only a loose sanity bound is asserted. +""" +import subprocess +import sys +import time + +import pytest + +from typedpython import gate + +FUNCTIONS = 300 + + +def synthetic_module() -> str: + lines = [] + for i in range(FUNCTIONS): + lines += [ + f"def f{i}(a: int, b: int) -> int:", + " total = 0", + " for k in range(a):", + " total += k * b", + " return total", + "", + ] + return "\n".join(lines) + + +@pytest.mark.measurement +def test_gate_cost_against_plain_pyrefly(write): + path = write("big.py", synthetic_module()) + lines = synthetic_module().count("\n") + 1 + + start = time.perf_counter() + subprocess.run( + [sys.executable, "-m", "pyrefly", "check", str(path), "--summary=none"], + capture_output=True, check=False, + ) + plain = time.perf_counter() - start + + start = time.perf_counter() + found = gate.check([path]) + gated = time.perf_counter() - start + + print(f"\n[measure] {lines} lines: plain pyrefly {plain * 1000:.0f} ms, " + f"gate {gated * 1000:.0f} ms, ratio {gated / plain:.2f}x") + assert found == [] + assert gated < 60 diff --git a/python-multiplatform-ksp/src/test/python/test_pyrefly_kinds.py b/python-multiplatform-ksp/src/test/python/test_pyrefly_kinds.py new file mode 100644 index 000000000..f7e864dfb --- /dev/null +++ b/python-multiplatform-ksp/src/test/python/test_pyrefly_kinds.py @@ -0,0 +1,69 @@ +"""Each Pyrefly Any-leak kind the gate raises to error (design §4.2).""" +import pytest + +from conftest import errors, rules +from typedpython import gate + +CASES = { + "implicit-any-parameter": """ + def f(x, y: int) -> int: + return y + """, + "unannotated-return": """ + def f(y: int): + return y + """, + "unknown-variable-type": """ + import json + def f() -> None: + d = json.loads("1") + z = d["k"] + """, + "no-any-return-implicit": """ + import json + def f() -> int: + d = json.loads("1") + return d["k"] + """, + "no-any-return-explicit": """ + from typing import Any + def f(v: Any) -> int: + return v + """, + "explicit-any": """ + from typing import Any + def f(v: Any) -> None: + pass + """, +} + + +@pytest.mark.parametrize("kind", sorted(CASES)) +def test_kind_is_an_error(write, kind): + path = write("user.py", CASES[kind]) + found = gate.check([path]) + assert f"pyrefly/{kind}" in rules(errors(found)), found + + +def test_ordinary_type_error_is_an_error(write): + path = write("user.py", """ + def f(y: int) -> int: + return y + "a" + """) + assert errors(gate.check([path])), "a plain type error must still fail the gate" + + +def test_fully_typed_code_is_clean(write): + path = write("user.py", """ + def area(w: int, h: int) -> int: + scale = 2 + return w * h * scale + + class Box: + def __init__(self, w: int) -> None: + self.w = w + + def double(self) -> int: + return self.w * 2 + """) + assert gate.check([path]) == [] diff --git a/python-multiplatform-ksp/src/test/python/test_rebinding.py b/python-multiplatform-ksp/src/test/python/test_rebinding.py new file mode 100644 index 000000000..3d50e272e --- /dev/null +++ b/python-multiplatform-ksp/src/test/python/test_rebinding.py @@ -0,0 +1,140 @@ +"""Rebinding a name to a different type, and mixed-element containers (design §3.1, #22). + +Both are forbidden-list items: warnings under `checked`, errors under `compiled`. They need the +inferred type of an assignment target, which Pyrefly's type report does not give directly; the +gate reads each target back through a probe load (see typedpython.rebinding). +""" +import pytest + +from typedpython import gate, rebinding + +REBIND = "forbidden/rebinding" +MIXED = "forbidden/mixed-container" + + +def found(write, source, mode="checked"): + return gate.check([write("user.py", source)], mode=mode) + + +def by_rule(diagnostics, rule): + return [d for d in diagnostics if d.rule == rule] + + +def test_rebinding_to_another_type_is_reported(write): + result = found(write, """ + def f() -> None: + x = 1 + x = "a" + print(x) + """) + [d] = by_rule(result, REBIND) + assert (d.line, d.severity) == (4, "warning") + assert "int" in d.message and "str" in d.message + + +def test_compiled_mode_makes_it_an_error(write): + result = found(write, """ + def f() -> None: + x = 1 + x = "a" + print(x) + """, mode="compiled") + assert [d.severity for d in by_rule(result, REBIND)] == ["error"] + + +def test_augmented_assignment_that_changes_the_type(write): + result = found(write, """ + def f() -> float: + total = 0 + total += 0.5 + return total + """) + assert by_rule(result, REBIND), result + + +def test_module_level_rebinding(write): + result = found(write, """ + limit = 10 + limit = "ten" + print(limit) + """) + assert by_rule(result, REBIND), result + + +@pytest.mark.parametrize("source", [ + # same type again + "def f() -> None:\n x = 1\n x = 2\n print(x)\n", + # a declared union governs; Pyrefly checks each assignment against it + "def f() -> None:\n x: int | str = 1\n x = 'a'\n print(x)\n", + # an annotated parameter reassigned within its type + "def f(n: int) -> int:\n n = n + 1\n return n\n", +]) +def test_not_rebinding(write, source): + assert by_rule(gate.check([write("user.py", source)]), REBIND) == [] + + +def test_mixed_list_is_reported(write): + result = found(write, """ + def f() -> None: + mixed = [1, "a"] + print(mixed) + """) + [d] = by_rule(result, MIXED) + assert d.line == 3 and "int | str" in d.message + + +def test_mixed_dict_values_are_reported(write): + result = found(write, """ + def f(a: int) -> None: + d = {"k": a, "j": "s"} + print(d) + """) + assert by_rule(result, MIXED), result + + +@pytest.mark.parametrize("source", [ + "def f() -> None:\n nums = [1, 2]\n print(nums)\n", + "def f() -> None:\n pair = (1, 'a')\n print(pair)\n", # tuples are typed per slot + "def f() -> None:\n xs: list[int | str] = [1, 'a']\n print(xs)\n", # declared on purpose +]) +def test_not_mixed(write, source): + assert by_rule(gate.check([write("user.py", source)]), MIXED) == [] + + +def test_probe_keeps_every_line_number(): + source = "def f() -> None:\n x = 1 # one\n x = 'a'\n y, z = 1, 2\n" + probed, probes = rebinding.probe(source) + assert probed.count("\n") == source.count("\n") + assert {p.name for p in probes} >= {"x", "y", "z"} + for p in probes: + line = probed.splitlines()[p.line - 1] + assert line[p.column - 1:].startswith(p.name), (p, line) + + +def test_no_candidates_means_no_second_pyrefly_run(write, monkeypatch): + calls = [] + real = gate.pyrefly.run + + def counting(*args, **kwargs): + calls.append(args) + return real(*args, **kwargs) + + monkeypatch.setattr(gate.pyrefly, "run", counting) + gate.check([write("user.py", "def f(a: int) -> int:\n b = a + 1\n return b\n")]) + assert len(calls) == 1 + + +def test_rebinding_inside_a_package_with_relative_imports(write): + write("pkg/__init__.py", "") + write("pkg/b.py", "def make() -> int:\n return 1\n") + path = write("pkg/a.py", """ + from .b import make + + def f() -> None: + x = make() + x = "s" + print(x) + """) + result = gate.check([path]) + assert by_rule(result, REBIND), result + assert not [d for d in result if d.severity == "error"], result