From 7fcd5ca31354028b69234558cc5f50ec5f200bdd Mon Sep 17 00:00:00 2001 From: Hugo Date: Thu, 3 Sep 2026 06:38:05 +0200 Subject: [PATCH 1/2] test: define control-flow graph behavior Acceptance tests for per-function CFGs over PyHIR: explicit Branch, Jump, Return, Raise and ForEach terminators, predecessors and successors in block order, reachability, back edges, integrity checks, loop control errors, and the CFG as a function-level analysis. Adds the PyHIR control-flow node tests. Co-Authored-By: Claude Fable 5.1 --- tests/test_cfg.py | 385 ++++++++++++++++++++++++++++++++++++++ tests/test_hir_builder.py | 76 ++++++++ 2 files changed, 461 insertions(+) create mode 100644 tests/test_cfg.py diff --git a/tests/test_cfg.py b/tests/test_cfg.py new file mode 100644 index 0000000..09a9ed1 --- /dev/null +++ b/tests/test_cfg.py @@ -0,0 +1,385 @@ +"""Acceptance tests for control-flow graphs (``docs/architecture.md`` §5). + +Each function becomes a graph of basic blocks. A block holds straight-line PyHIR +statements and ends in exactly one explicit terminator: ``Branch``, ``Jump``, +``Return``, ``Raise`` or ``ForEach`` (Python's iteration protocol kept explicit rather +than encoded as a generic switch). The graph exposes predecessors, successors, +reachability and back edges, and validates its own integrity. + +Expected to remain red until ``coretrace_python.cfg`` exists and PyHIR represents +``if``, ``while``, ``for``, ``break``, ``continue`` and ``raise``. +""" + +from __future__ import annotations + +import dataclasses + +import pytest + +from coretrace_python.analysis import AnalysisManager +from coretrace_python.frontend import build_hir +from coretrace_python.hir import nodes +from coretrace_python.source import SourceManager + +try: + from coretrace_python.cfg import ( + CFG, + BasicBlock, + BlockId, + Branch, + CFGAnalysis, + CFGError, + ForEach, + Jump, + Raise, + Return, + build_cfg, + ) +except ImportError as error: # pragma: no cover - red until the CFG lands + MISSING = error +else: + MISSING = None + + +@pytest.fixture(autouse=True) +def require_cfg() -> None: + if MISSING is not None: + pytest.fail(f"control-flow graphs are not implemented yet: {MISSING}") + + +def function(source_text: str) -> nodes.Function: + module = build_hir(SourceManager().add_source("flow.py", source_text)) + for statement in module.body: + if isinstance(statement, nodes.Function): + return statement + raise AssertionError("no function in source") + + +def cfg_for(source_text: str) -> CFG: + return build_cfg(function(source_text)) + + +def only(cfg: CFG, block_id: BlockId) -> BlockId: + successors = cfg.successors(block_id) + assert len(successors) == 1, successors + return successors[0] + + +def assigned_names(block: BasicBlock) -> list[str]: + return [s.target.identifier for s in block.statements if isinstance(s, nodes.Assign)] + + +# --------------------------------------------------------------------------- straight line + + +def test_straight_line_function_is_one_block() -> None: + cfg = cfg_for("def f(a):\n x = a\n y = x\n return y\n") + entry = cfg.block(cfg.entry) + + assert len(cfg.blocks) == 1 + assert assigned_names(entry) == ["x", "y"] + assert isinstance(entry.terminator, Return) + assert isinstance(entry.terminator.value, nodes.Name) + assert cfg.successors(cfg.entry) == () + assert cfg.predecessors(cfg.entry) == () + + +def test_falling_off_the_end_returns_none() -> None: + cfg = cfg_for("def f():\n x = 1\n") + terminator = cfg.block(cfg.entry).terminator + + assert isinstance(terminator, Return) + assert terminator.value is None + + +def test_terminator_carries_the_span_of_its_statement() -> None: + cfg = cfg_for("def f(a):\n x = a\n return x\n") + assert cfg.block(cfg.entry).terminator.span.start_line == 3 + + +# --------------------------------------------------------------------------- branches + + +def test_if_else_joins_at_a_merge_block() -> None: + cfg = cfg_for( + "def f(safe, x):\n" + " if safe:\n" + " x = sanitize(x)\n" + " else:\n" + " x = 0\n" + " sink(x)\n" + ) + entry = cfg.block(cfg.entry) + assert isinstance(entry.terminator, Branch) + assert isinstance(entry.terminator.condition, nodes.Name) + then_id, else_id = entry.terminator.then_block, entry.terminator.else_block + assert cfg.successors(cfg.entry) == (then_id, else_id) + + then_block, else_block = cfg.block(then_id), cfg.block(else_id) + assert assigned_names(then_block) == ["x"] + assert assigned_names(else_block) == ["x"] + assert isinstance(then_block.terminator, Jump) + assert isinstance(else_block.terminator, Jump) + merge_id = then_block.terminator.target + assert else_block.terminator.target == merge_id + assert cfg.predecessors(merge_id) == (then_id, else_id) + + merge = cfg.block(merge_id) + assert len(merge.statements) == 1 + assert isinstance(merge.terminator, Return) + assert len(cfg.blocks) == 4 + + +def test_if_without_else_branches_straight_to_the_merge() -> None: + # The example of §5: ``if safe: x = sanitize(x)`` then ``sink(x)``. + cfg = cfg_for("def f(safe, x):\n if safe:\n x = sanitize(x)\n sink(x)\n") + entry = cfg.block(cfg.entry) + assert isinstance(entry.terminator, Branch) + + then_block = cfg.block(entry.terminator.then_block) + assert isinstance(then_block.terminator, Jump) + assert then_block.terminator.target == entry.terminator.else_block + # Predecessors follow block order: the entry block is finished before the then block. + assert cfg.predecessors(entry.terminator.else_block) == ( + cfg.entry, + entry.terminator.then_block, + ) + assert len(cfg.blocks) == 3 + + +def test_elif_chains_nest_branches() -> None: + cfg = cfg_for( + "def f(a):\n" + " if a == 1:\n" + " return 1\n" + " elif a == 2:\n" + " return 2\n" + " return 0\n" + ) + first = cfg.block(cfg.entry).terminator + assert isinstance(first, Branch) + second = cfg.block(first.else_block).terminator + assert isinstance(second, Branch) + assert isinstance(cfg.block(first.then_block).terminator, Return) + assert isinstance(cfg.block(second.then_block).terminator, Return) + assert isinstance(cfg.block(second.else_block).terminator, Return) + + +def test_early_return_leaves_one_predecessor_at_the_merge() -> None: + cfg = cfg_for( + "def f(value):\n" + " if not value.isdigit():\n" + " return None\n" + " sink(value)\n" + " return value\n" + ) + branch = cfg.block(cfg.entry).terminator + assert isinstance(branch, Branch) + assert isinstance(cfg.block(branch.then_block).terminator, Return) + assert cfg.successors(branch.then_block) == () + assert cfg.predecessors(branch.else_block) == (cfg.entry,) + + +# --------------------------------------------------------------------------- loops + + +def test_while_loop_has_a_header_with_a_back_edge() -> None: + cfg = cfg_for("def f(n):\n while n > 0:\n n = n - 1\n return n\n") + entry = cfg.block(cfg.entry) + assert isinstance(entry.terminator, Jump) + header_id = entry.terminator.target + + header = cfg.block(header_id) + assert header.statements == () + assert isinstance(header.terminator, Branch) + body_id, exit_id = header.terminator.then_block, header.terminator.else_block + + body = cfg.block(body_id) + assert assigned_names(body) == ["n"] + assert isinstance(body.terminator, Jump) + assert body.terminator.target == header_id + assert cfg.predecessors(header_id) == (cfg.entry, body_id) + assert cfg.back_edges() == frozenset({(body_id, header_id)}) + assert isinstance(cfg.block(exit_id).terminator, Return) + + +def test_for_loop_keeps_the_iteration_protocol_explicit() -> None: + cfg = cfg_for("def f(items):\n total = 0\n for item in items:\n total = total + item\n return total\n") + entry = cfg.block(cfg.entry) + assert assigned_names(entry) == ["total"] + assert isinstance(entry.terminator, Jump) + header = cfg.block(entry.terminator.target) + + assert isinstance(header.terminator, ForEach) + assert header.terminator.target.identifier == "item" + assert isinstance(header.terminator.iterable, nodes.Name) + body = cfg.block(header.terminator.body) + assert assigned_names(body) == ["total"] + assert isinstance(body.terminator, Jump) + assert body.terminator.target == entry.terminator.target + assert cfg.back_edges() == frozenset({(header.terminator.body, entry.terminator.target)}) + assert isinstance(cfg.block(header.terminator.exit).terminator, Return) + + +def test_break_jumps_to_the_loop_exit() -> None: + cfg = cfg_for( + "def f(items):\n" + " for item in items:\n" + " if item:\n" + " break\n" + " return 1\n" + ) + header = cfg.block(only(cfg, cfg.entry)) + assert isinstance(header.terminator, ForEach) + body = cfg.block(header.terminator.body) + assert isinstance(body.terminator, Branch) + break_block = cfg.block(body.terminator.then_block) + + assert break_block.statements == () + assert isinstance(break_block.terminator, Jump) + assert break_block.terminator.target == header.terminator.exit + assert set(cfg.predecessors(header.terminator.exit)) == { + only(cfg, cfg.entry), + body.terminator.then_block, + } + + +def test_continue_jumps_to_the_loop_header() -> None: + cfg = cfg_for( + "def f(n):\n" + " while n:\n" + " if n == 3:\n" + " continue\n" + " n = n - 1\n" + " return n\n" + ) + header_id = only(cfg, cfg.entry) + body = cfg.block(cfg.block(header_id).terminator.then_block) # type: ignore[union-attr] + assert isinstance(body.terminator, Branch) + continue_block = cfg.block(body.terminator.then_block) + + assert isinstance(continue_block.terminator, Jump) + assert continue_block.terminator.target == header_id + assert (body.terminator.then_block, header_id) in cfg.back_edges() + + +@pytest.mark.parametrize( + "statement, location", + [("break", "flow.py:2:5"), ("continue", "flow.py:2:5")], +) +def test_loop_control_outside_a_loop_is_an_error(statement: str, location: str) -> None: + with pytest.raises(CFGError, match=rf"{location}: .*{statement}.*outside"): + cfg_for(f"def f():\n {statement}\n") + + +# --------------------------------------------------------------------------- raise and reachability + + +def test_raise_terminates_a_block_and_makes_the_rest_unreachable() -> None: + cfg = cfg_for( + "def f(a):\n" + " if a:\n" + " raise ValueError(a)\n" + " x = 1\n" + " return a\n" + ) + branch = cfg.block(cfg.entry).terminator + assert isinstance(branch, Branch) + raising = cfg.block(branch.then_block) + assert isinstance(raising.terminator, Raise) + assert isinstance(raising.terminator.exception, nodes.Call) + assert cfg.successors(branch.then_block) == () + + unreachable = [b for b in cfg.blocks.values() if assigned_names(b) == ["x"]] + assert len(unreachable) == 1 + assert unreachable[0].id not in cfg.reachable() + assert cfg.entry in cfg.reachable() + assert branch.else_block in cfg.reachable() + + +def test_bare_raise_has_no_exception_expression() -> None: + cfg = cfg_for("def f():\n raise\n") + terminator = cfg.block(cfg.entry).terminator + assert isinstance(terminator, Raise) + assert terminator.exception is None + + +def test_code_after_return_is_kept_but_unreachable() -> None: + cfg = cfg_for("def f():\n return 1\n x = 2\n") + assert len(cfg.blocks) == 2 + assert cfg.reachable() == frozenset({cfg.entry}) + + +# --------------------------------------------------------------------------- graph shape + + +def test_predecessors_and_successors_mirror_each_other() -> None: + cfg = cfg_for( + "def f(items, flag):\n" + " for item in items:\n" + " if flag:\n" + " continue\n" + " while item:\n" + " item = item - 1\n" + " return 0\n" + ) + for block_id in cfg.blocks: + for successor in cfg.successors(block_id): + assert block_id in cfg.predecessors(successor) + for predecessor in cfg.predecessors(block_id): + assert block_id in cfg.successors(predecessor) + + +def test_block_order_starts_at_entry_and_is_deterministic() -> None: + source_text = "def f(a):\n if a:\n a = 1\n while a:\n a = a - 1\n return a\n" + first, second = cfg_for(source_text), cfg_for(source_text) + + assert next(iter(first.blocks)) == first.entry + assert first == second + assert list(first.blocks) == list(second.blocks) + + +def test_nested_definitions_stay_inside_their_block() -> None: + cfg = cfg_for("def f():\n def g():\n return 1\n class C:\n pass\n return g\n") + entry = cfg.block(cfg.entry) + + assert len(cfg.blocks) == 1 + assert [type(s).__name__ for s in entry.statements] == ["Function", "Class"] + + +def test_graphs_are_immutable() -> None: + cfg = cfg_for("def f():\n return 1\n") + with pytest.raises(dataclasses.FrozenInstanceError): + cfg.entry = BlockId("other") # type: ignore[misc] + with pytest.raises(TypeError): + cfg.blocks[BlockId("other")] = cfg.block(cfg.entry) # type: ignore[index] + + +def test_terminators_must_target_existing_blocks() -> None: + entry = BlockId("entry") + span = function("def f():\n pass\n").span + block = BasicBlock(entry, (), Jump(BlockId("missing"), span)) + + with pytest.raises(CFGError, match="missing"): + CFG(entry, {entry: block}) + + +# --------------------------------------------------------------------------- analysis + + +def test_cfg_is_a_function_analysis() -> None: + module = build_hir( + SourceManager().add_source( + "flow.py", "def a(x):\n if x:\n return 1\n return 0\n\ndef b():\n pass\n" + ) + ) + manager = AnalysisManager(module) + manager.register(CFGAnalysis) + a, b = (s for s in module.body if isinstance(s, nodes.Function)) + + cfg = manager.get(CFGAnalysis, a) + + assert CFGAnalysis.name == "cfg.function" + assert isinstance(cfg, CFG) + assert len(cfg.blocks) == 3 + assert manager.is_cached(CFGAnalysis, b) is False diff --git a/tests/test_hir_builder.py b/tests/test_hir_builder.py index fbf3481..e5f901c 100644 --- a/tests/test_hir_builder.py +++ b/tests/test_hir_builder.py @@ -144,3 +144,79 @@ def test_represents_list_comprehensions() -> None: def test_module_carries_its_dotted_name() -> None: source = SourceManager().add_source("db.py", "value = 1\n", module_name="app.services.db") assert build_hir(source).name == "app.services.db" + + +# --------------------------------------------------------------------------- control flow +# PyHIR nodes required by the Phase 3 CFG builder (docs/architecture.md §3.2, §5). +# Expected to remain red until ``If``, ``While``, ``For``, ``Break``, ``Continue`` and +# ``Raise`` exist. + + +def test_represents_if_with_else_and_elif() -> None: + module = build("def f(a):\n if a == 1:\n return 1\n elif a == 2:\n return 2\n else:\n return 0\n") + function = module.body[0] + assert isinstance(function, nodes.Function) + outer = function.body[0] + assert isinstance(outer, nodes.If) + assert isinstance(outer.condition, nodes.Compare) + assert isinstance(outer.body[0], nodes.Return) + inner = outer.orelse[0] + assert isinstance(inner, nodes.If) + assert isinstance(inner.orelse[0], nodes.Return) + assert outer.span.start_line == 2 + assert inner.span.start_line == 4 + + +def test_represents_while_loops_with_break_and_continue() -> None: + module = build("def f(n):\n while n:\n if n:\n break\n continue\n") + function = module.body[0] + assert isinstance(function, nodes.Function) + loop = function.body[0] + assert isinstance(loop, nodes.While) + assert isinstance(loop.condition, nodes.Name) + conditional = loop.body[0] + assert isinstance(conditional, nodes.If) + assert isinstance(conditional.body[0], nodes.Break) + assert isinstance(loop.body[1], nodes.Continue) + assert loop.body[1].span.start_line == 5 + + +def test_represents_for_loops_over_a_name_target() -> None: + module = build("def f(items):\n for item in items:\n pass\n") + function = module.body[0] + assert isinstance(function, nodes.Function) + loop = function.body[0] + assert isinstance(loop, nodes.For) + assert isinstance(loop.target, nodes.Name) + assert loop.target.identifier == "item" + assert isinstance(loop.iterable, nodes.Name) + assert isinstance(loop.body[0], nodes.Pass) + assert loop.is_async is False + + +def test_represents_raise_with_and_without_exception() -> None: + module = build("def f(a):\n raise ValueError(a)\n\ndef g():\n raise\n") + first, second = module.body + assert isinstance(first, nodes.Function) and isinstance(second, nodes.Function) + raising = first.body[0] + assert isinstance(raising, nodes.Raise) + assert isinstance(raising.exception, nodes.Call) + bare = second.body[0] + assert isinstance(bare, nodes.Raise) + assert bare.exception is None + + +@pytest.mark.parametrize( + "source_text, message", + [ + ("while x:\n pass\nelse:\n pass\n", "else"), + ("for a, b in items:\n pass\n", "single name"), + ("raise Error from cause\n", "from"), + ], + ids=["loop-else", "tuple-target", "raise-from"], +) +def test_unsupported_control_flow_forms_are_reported(source_text: str, message: str) -> None: + from coretrace_python.frontend import HIRBuildError + + with pytest.raises(HIRBuildError, match=message): + build(source_text) From 3e139f6d8f6db0f833be7eacc5001102014e524d Mon Sep 17 00:00:00 2001 From: Hugo Date: Thu, 3 Sep 2026 06:39:15 +0200 Subject: [PATCH 2/2] feat: add control-flow graphs PyHIR represents if, while, for, break, continue and raise; scope and import analyses walk the new bodies. The cfg package builds per-function graphs of basic blocks with explicit Branch, Jump, Return, Raise and ForEach terminators, exposes predecessors, successors, reachability and back edges, validates its integrity, and is registered as the cfg.function analysis. Co-Authored-By: Claude Fable 5.1 --- src/coretrace_python/cfg/__init__.py | 32 ++++ src/coretrace_python/cfg/builder.py | 153 ++++++++++++++++++ src/coretrace_python/cfg/model.py | 158 +++++++++++++++++++ src/coretrace_python/frontend/ast_adapter.py | 33 ++++ src/coretrace_python/hir/nodes.py | 46 ++++++ src/coretrace_python/semantic/imports.py | 5 + src/coretrace_python/semantic/scopes.py | 16 +- 7 files changed, 442 insertions(+), 1 deletion(-) create mode 100644 src/coretrace_python/cfg/__init__.py create mode 100644 src/coretrace_python/cfg/builder.py create mode 100644 src/coretrace_python/cfg/model.py diff --git a/src/coretrace_python/cfg/__init__.py b/src/coretrace_python/cfg/__init__.py new file mode 100644 index 0000000..ea91118 --- /dev/null +++ b/src/coretrace_python/cfg/__init__.py @@ -0,0 +1,32 @@ +"""Per-function control-flow graphs over PyHIR (architecture §5).""" + +from coretrace_python.cfg.builder import CFGAnalysis, build_cfg +from coretrace_python.cfg.model import ( + CFG, + BasicBlock, + BlockId, + Branch, + CFGError, + ForEach, + Jump, + Raise, + Return, + Terminator, + targets, +) + +__all__ = [ + "CFG", + "BasicBlock", + "BlockId", + "Branch", + "CFGAnalysis", + "CFGError", + "ForEach", + "Jump", + "Raise", + "Return", + "Terminator", + "build_cfg", + "targets", +] diff --git a/src/coretrace_python/cfg/builder.py b/src/coretrace_python/cfg/builder.py new file mode 100644 index 0000000..9df6d68 --- /dev/null +++ b/src/coretrace_python/cfg/builder.py @@ -0,0 +1,153 @@ +"""Build a control-flow graph from a PyHIR function.""" + +from __future__ import annotations + +from typing import ClassVar + +from coretrace_python.analysis import AnalysisContext, FunctionAnalysis +from coretrace_python.cfg.model import ( + CFG, + BasicBlock, + BlockId, + Branch, + CFGError, + ForEach, + Jump, + Raise, + Return, + Terminator, +) +from coretrace_python.hir import nodes +from coretrace_python.source import SourceSpan + + +class _Open: + """A block that is still collecting statements and has no terminator yet.""" + + def __init__(self, block_id: BlockId) -> None: + self.id = block_id + self.statements: list[nodes.Statement] = [] + + +class _Builder: + def __init__(self, function: nodes.Function) -> None: + self.function = function + self.blocks: dict[BlockId, BasicBlock] = {} + self.counters: dict[str, int] = {} + self.loops: list[tuple[BlockId, BlockId]] = [] + + def build(self) -> CFG: + entry = BlockId("entry") + end = self.sequence(self.function.body, _Open(entry), None, self.function.span) + if end is not None: + self.finish(end, Return(None, self.function.span)) + return CFG(entry, self.blocks) + + # ------------------------------------------------------------------ blocks + + def new_id(self, kind: str) -> BlockId: + self.counters[kind] = self.counters.get(kind, 0) + 1 + return BlockId(f"{kind}_{self.counters[kind]}") + + def finish(self, block: _Open, terminator: Terminator) -> None: + self.blocks[block.id] = BasicBlock(block.id, tuple(block.statements), terminator) + + def header(self, block_id: BlockId, terminator: Terminator) -> None: + self.blocks[block_id] = BasicBlock(block_id, (), terminator) + + # ------------------------------------------------------------------ statements + + def sequence( + self, + statements: tuple[nodes.Statement, ...], + block: _Open, + continuation: BlockId | None, + join_span: SourceSpan, + ) -> _Open | None: + """Lay ``statements`` out from ``block``. + + Returns the block left open at the end, or ``None`` when control left the + sequence. With a ``continuation``, an open end jumps there instead. + """ + + current: _Open | None = block + last_index = len(statements) - 1 + for index, statement in enumerate(statements): + assert current is not None + is_last = index == last_index + current = self.statement(statement, current, continuation if is_last else None) + if current is None and not is_last: + current = _Open(self.new_id("dead")) + if current is not None and continuation is not None: + self.finish(current, Jump(continuation, join_span)) + return None + return current + + def statement( + self, node: nodes.Statement, block: _Open, continuation: BlockId | None + ) -> _Open | None: + if isinstance(node, nodes.Return): + self.finish(block, Return(node.value, node.span)) + return None + if isinstance(node, nodes.Raise): + self.finish(block, Raise(node.exception, node.span)) + return None + if isinstance(node, nodes.Break): + self.finish(block, Jump(self.loop("break", node.span)[1], node.span)) + return None + if isinstance(node, nodes.Continue): + self.finish(block, Jump(self.loop("continue", node.span)[0], node.span)) + return None + if isinstance(node, nodes.If): + return self.conditional(node, block, continuation) + if isinstance(node, nodes.While | nodes.For): + return self.loop_statement(node, block, continuation) + block.statements.append(node) + return block + + def conditional( + self, node: nodes.If, block: _Open, continuation: BlockId | None + ) -> _Open | None: + merge = continuation if continuation is not None else self.new_id("merge") + then_id = self.new_id("then") + else_id = self.new_id("else") if node.orelse else merge + self.finish(block, Branch(node.condition, then_id, else_id, node.span)) + self.sequence(node.body, _Open(then_id), merge, node.span) + if node.orelse: + self.sequence(node.orelse, _Open(else_id), merge, node.span) + return None if continuation is not None else _Open(merge) + + def loop_statement( + self, node: nodes.While | nodes.For, block: _Open, continuation: BlockId | None + ) -> _Open | None: + header_id = self.new_id("loop") + body_id = self.new_id("body") + exit_id = continuation if continuation is not None else self.new_id("exit") + self.finish(block, Jump(header_id, node.span)) + if isinstance(node, nodes.While): + self.header(header_id, Branch(node.condition, body_id, exit_id, node.span)) + else: + self.header(header_id, ForEach(node.target, node.iterable, body_id, exit_id, node.span)) + self.loops.append((header_id, exit_id)) + try: + self.sequence(node.body, _Open(body_id), header_id, node.span) + finally: + self.loops.pop() + return None if continuation is not None else _Open(exit_id) + + def loop(self, keyword: str, span: SourceSpan) -> tuple[BlockId, BlockId]: + if not self.loops: + raise CFGError(f"{span.display()}: '{keyword}' outside loop") + return self.loops[-1] + + +def build_cfg(function: nodes.Function) -> CFG: + return _Builder(function).build() + + +class CFGAnalysis(FunctionAnalysis[CFG]): + name: ClassVar[str] = "cfg.function" + + @classmethod + def compute(cls, ctx: AnalysisContext, function: nodes.Function) -> CFG: + return build_cfg(function) diff --git a/src/coretrace_python/cfg/model.py b/src/coretrace_python/cfg/model.py new file mode 100644 index 0000000..6296f81 --- /dev/null +++ b/src/coretrace_python/cfg/model.py @@ -0,0 +1,158 @@ +"""Control-flow graph model (architecture §5). + +A block holds straight-line PyHIR statements and ends in exactly one terminator. +Python's iteration protocol stays explicit as ``ForEach`` instead of being encoded as +a generic switch; the CFG must remain semantically faithful to Python. +""" + +from __future__ import annotations + +from collections.abc import Mapping +from dataclasses import dataclass, field +from types import MappingProxyType +from typing import TypeAlias + +from coretrace_python.hir import nodes +from coretrace_python.source import SourceSpan + + +class CFGError(Exception): + """A malformed graph or a control statement used outside its construct.""" + + +@dataclass(frozen=True, order=True) +class BlockId: + value: str + + def __str__(self) -> str: + return self.value + + +@dataclass(frozen=True) +class Branch: + condition: nodes.Expression + then_block: BlockId + else_block: BlockId + span: SourceSpan + + +@dataclass(frozen=True) +class Jump: + target: BlockId + span: SourceSpan + + +@dataclass(frozen=True) +class Return: + value: nodes.Expression | None + span: SourceSpan + + +@dataclass(frozen=True) +class Raise: + exception: nodes.Expression | None + span: SourceSpan + + +@dataclass(frozen=True) +class ForEach: + """Bind the next item of ``iterable`` to ``target`` and enter ``body``, or ``exit``.""" + + target: nodes.Name + iterable: nodes.Expression + body: BlockId + exit: BlockId + span: SourceSpan + + +Terminator: TypeAlias = Branch | Jump | Return | Raise | ForEach + + +def targets(terminator: Terminator) -> tuple[BlockId, ...]: + if isinstance(terminator, Branch): + return (terminator.then_block, terminator.else_block) + if isinstance(terminator, Jump): + return (terminator.target,) + if isinstance(terminator, ForEach): + return (terminator.body, terminator.exit) + return () + + +@dataclass(frozen=True) +class BasicBlock: + id: BlockId + statements: tuple[nodes.Statement, ...] + terminator: Terminator + + +@dataclass(frozen=True) +class CFG: + entry: BlockId + blocks: Mapping[BlockId, BasicBlock] + _successors: Mapping[BlockId, tuple[BlockId, ...]] = field( + init=False, repr=False, compare=False + ) + _predecessors: Mapping[BlockId, tuple[BlockId, ...]] = field( + init=False, repr=False, compare=False + ) + + def __post_init__(self) -> None: + blocks = MappingProxyType(dict(self.blocks)) + if self.entry not in blocks: + raise CFGError(f"entry block {self.entry} does not exist") + predecessors: dict[BlockId, list[BlockId]] = {block_id: [] for block_id in blocks} + successors: dict[BlockId, tuple[BlockId, ...]] = {} + for block_id, block in blocks.items(): + if block.id != block_id: + raise CFGError(f"block {block.id} is stored under {block_id}") + successors[block_id] = targets(block.terminator) + for target in successors[block_id]: + if target not in blocks: + raise CFGError(f"block {block_id} targets unknown block {target}") + predecessors[target].append(block_id) + object.__setattr__(self, "blocks", blocks) + object.__setattr__(self, "_successors", MappingProxyType(successors)) + object.__setattr__( + self, + "_predecessors", + MappingProxyType({k: tuple(v) for k, v in predecessors.items()}), + ) + + def block(self, block_id: BlockId) -> BasicBlock: + return self.blocks[block_id] + + def successors(self, block_id: BlockId) -> tuple[BlockId, ...]: + return self._successors[block_id] + + def predecessors(self, block_id: BlockId) -> tuple[BlockId, ...]: + return self._predecessors[block_id] + + def reachable(self) -> frozenset[BlockId]: + seen: set[BlockId] = set() + pending = [self.entry] + while pending: + block_id = pending.pop() + if block_id not in seen: + seen.add(block_id) + pending.extend(self.successors(block_id)) + return frozenset(seen) + + def back_edges(self) -> frozenset[tuple[BlockId, BlockId]]: + """Edges whose target is on the depth-first path from the entry to their source.""" + + found: set[tuple[BlockId, BlockId]] = set() + on_path: set[BlockId] = set() + finished: set[BlockId] = set() + + def visit(block_id: BlockId) -> None: + on_path.add(block_id) + for successor in self.successors(block_id): + if successor in on_path: + found.add((block_id, successor)) + elif successor not in finished: + visit(successor) + on_path.remove(block_id) + finished.add(block_id) + + visit(self.entry) + return frozenset(found) diff --git a/src/coretrace_python/frontend/ast_adapter.py b/src/coretrace_python/frontend/ast_adapter.py index 350155f..88201de 100644 --- a/src/coretrace_python/frontend/ast_adapter.py +++ b/src/coretrace_python/frontend/ast_adapter.py @@ -154,6 +154,36 @@ def statement(self, node: ast.stmt) -> nodes.Statement: nodes.ImportAlias(alias.name, alias.asname, self.span(alias)) for alias in node.names ) return nodes.ImportFrom(node.module, aliases, node.level, span) + if isinstance(node, ast.If): + return nodes.If( + self.expression(node.test), self.block(node.body), self.block(node.orelse), span + ) + if isinstance(node, ast.While): + if node.orelse: + self.fail(node.orelse[0], "loop else clauses are not supported yet") + return nodes.While(self.expression(node.test), self.block(node.body), span) + if isinstance(node, (ast.For, ast.AsyncFor)): + if node.orelse: + self.fail(node.orelse[0], "loop else clauses are not supported yet") + if not isinstance(node.target, ast.Name): + self.fail(node.target, "only a single name is supported as a loop target") + target = nodes.Name(node.target.id, self.span(node.target)) + return nodes.For( + target, + self.expression(node.iter), + self.block(node.body), + isinstance(node, ast.AsyncFor), + span, + ) + if isinstance(node, ast.Break): + return nodes.Break(span) + if isinstance(node, ast.Continue): + return nodes.Continue(span) + if isinstance(node, ast.Raise): + if node.cause is not None: + self.fail(node.cause, "raise from is not supported yet") + exception = self.expression(node.exc) if node.exc is not None else None + return nodes.Raise(exception, span) if isinstance(node, ast.Global): return nodes.Global(tuple(node.names), span) if isinstance(node, ast.Nonlocal): @@ -164,6 +194,9 @@ def statement(self, node: ast.stmt) -> nodes.Statement: return self.class_definition(node) self.fail(node) + def block(self, statements: list[ast.stmt]) -> tuple[nodes.Statement, ...]: + return tuple(self.statement(statement) for statement in statements) + def class_definition(self, node: ast.ClassDef) -> nodes.Class: if node.decorator_list: self.fail(node, "decorated classes are not supported yet") diff --git a/src/coretrace_python/hir/nodes.py b/src/coretrace_python/hir/nodes.py index 949f0cd..93151c5 100644 --- a/src/coretrace_python/hir/nodes.py +++ b/src/coretrace_python/hir/nodes.py @@ -156,6 +156,46 @@ class ImportFrom: span: SourceSpan +@dataclass(frozen=True, slots=True) +class If: + condition: Expression + body: tuple[Statement, ...] + orelse: tuple[Statement, ...] + span: SourceSpan + + +@dataclass(frozen=True, slots=True) +class While: + condition: Expression + body: tuple[Statement, ...] + span: SourceSpan + + +@dataclass(frozen=True, slots=True) +class For: + target: Name + iterable: Expression + body: tuple[Statement, ...] + is_async: bool + span: SourceSpan + + +@dataclass(frozen=True, slots=True) +class Break: + span: SourceSpan + + +@dataclass(frozen=True, slots=True) +class Continue: + span: SourceSpan + + +@dataclass(frozen=True, slots=True) +class Raise: + exception: Expression | None + span: SourceSpan + + @dataclass(frozen=True, slots=True) class Global: names: tuple[str, ...] @@ -190,6 +230,12 @@ class Class: | Return | ExpressionStatement | Pass + | If + | While + | For + | Break + | Continue + | Raise | Import | ImportFrom | Global diff --git a/src/coretrace_python/semantic/imports.py b/src/coretrace_python/semantic/imports.py index ee4a5b2..c99c6b5 100644 --- a/src/coretrace_python/semantic/imports.py +++ b/src/coretrace_python/semantic/imports.py @@ -72,6 +72,11 @@ def body(self, statements: tuple[nodes.Statement, ...], scope_id: ScopeId) -> No self.bind(scope_id, alias.as_name or alias.name, f"{base}.{alias.name}") elif isinstance(statement, (nodes.Function, nodes.Class)): self.body(statement.body, self.scopes.scope_for(statement).id) + elif isinstance(statement, nodes.If): + self.body(statement.body, scope_id) + self.body(statement.orelse, scope_id) + elif isinstance(statement, (nodes.While, nodes.For)): + self.body(statement.body, scope_id) def base_module(self, statement: nodes.ImportFrom) -> str: if not statement.level: diff --git a/src/coretrace_python/semantic/scopes.py b/src/coretrace_python/semantic/scopes.py index 83bbc26..ca7ec0c 100644 --- a/src/coretrace_python/semantic/scopes.py +++ b/src/coretrace_python/semantic/scopes.py @@ -264,7 +264,21 @@ def statement(self, node: nodes.Statement, scope: _ScopeBuilder) -> None: self.expression(base, scope) scope.bind(node.name, BindingKind.CLASS, node.span) self.body(node.body, self.open(scope, ScopeKind.CLASS, node.name, node.span)) - elif not isinstance(node, nodes.Pass): + elif isinstance(node, nodes.If): + self.expression(node.condition, scope) + self.body(node.body, scope) + self.body(node.orelse, scope) + elif isinstance(node, nodes.While): + self.expression(node.condition, scope) + self.body(node.body, scope) + elif isinstance(node, nodes.For): + self.expression(node.iterable, scope) + scope.bind(node.target.identifier, BindingKind.LOCAL, node.target.span) + self.body(node.body, scope) + elif isinstance(node, nodes.Raise): + if node.exception is not None: + self.expression(node.exception, scope) + elif not isinstance(node, nodes.Pass | nodes.Break | nodes.Continue): raise TypeError(f"unknown statement: {node!r}") def expression(self, node: nodes.Expression, scope: _ScopeBuilder) -> None: