feat: analysis manager

pyproject.toml

@@ -33,6 +33,9 @@ testpaths = ["tests"]
 line-length = 100
 target-version = "py311"
 
+[tool.ruff.lint.isort]
+known-first-party = ["coretrace_python"]
+
 [tool.mypy]
 python_version = "3.11"
 strict = true

src/coretrace_python/analysis/__init__.py

@@ -0,0 +1,29 @@
+"""Analysis infrastructure: typed providers managed as a lazy, cached dependency DAG."""
+
+from coretrace_python.analysis.manager import (
+    AnalysisError,
+    AnalysisManager,
+    CyclicDependencyError,
+    UndeclaredDependencyError,
+    UnregisteredAnalysisError,
+)
+from coretrace_python.analysis.provider import (
+    Analysis,
+    AnalysisContext,
+    AnyAnalysis,
+    FunctionAnalysis,
+    TransformationPass,
+)
+
+__all__ = [
+    "Analysis",
+    "AnalysisContext",
+    "AnalysisError",
+    "AnalysisManager",
+    "AnyAnalysis",
+    "CyclicDependencyError",
+    "FunctionAnalysis",
+    "TransformationPass",
+    "UndeclaredDependencyError",
+    "UnregisteredAnalysisError",
+]

src/coretrace_python/analysis/manager.py

@@ -0,0 +1,145 @@
+"""The Analysis Manager: registry, dependency DAG, lazy evaluation, cache, invalidation."""
+
+from __future__ import annotations
+
+from typing import Any, overload
+
+from coretrace_python.analysis.provider import (
+    Analysis,
+    AnyAnalysis,
+    FunctionAnalysis,
+    R,
+    TransformationPass,
+)
+from coretrace_python.hir import nodes
+from coretrace_python.source import SourceSpan
+
+
+class AnalysisError(Exception):
+    """Misuse of the analysis registry or dependency declarations."""
+
+
+class UnregisteredAnalysisError(AnalysisError):
+    pass
+
+
+class UndeclaredDependencyError(AnalysisError):
+    pass
+
+
+class CyclicDependencyError(AnalysisError):
+    pass
+
+
+_CacheKey = tuple[AnyAnalysis, SourceSpan | None]
+
+
+class AnalysisManager:
+    """Compute registered analyses lazily, once, and share their results."""
+
+    def __init__(self, module: nodes.Module) -> None:
+        self._module = module
+        self._registry: set[AnyAnalysis] = set()
+        self._cache: dict[_CacheKey, Any] = {}
+        self._computing: list[AnyAnalysis] = []
+
+    @property
+    def module(self) -> nodes.Module:
+        return self._module
+
+    # ------------------------------------------------------------------ registry
+
+    def register(self, *analyses: AnyAnalysis) -> None:
+        for analysis in analyses:
+            self._registry.add(analysis)
+            self._check_acyclic(analysis)
+
+    def dependencies(self, analysis: AnyAnalysis) -> frozenset[AnyAnalysis]:
+        """Transitive closure of ``analysis.requires``."""
+
+        found: set[AnyAnalysis] = set()
+        pending = list(analysis.requires)
+        while pending:
+            dependency = pending.pop()
+            if dependency not in found:
+                found.add(dependency)
+                pending.extend(dependency.requires)
+        return frozenset(found)
+
+    def _check_acyclic(self, root: AnyAnalysis) -> None:
+        path: list[AnyAnalysis] = []
+
+        def visit(analysis: AnyAnalysis) -> None:
+            if analysis in path:
+                cycle = [*path[path.index(analysis) :], analysis]
+                raise CyclicDependencyError(
+                    "dependency cycle: " + " -> ".join(a.name for a in cycle)
+                )
+            path.append(analysis)
+            for dependency in analysis.requires:
+                visit(dependency)
+            path.pop()
+
+        visit(root)
+
+    # ------------------------------------------------------------------ evaluation
+
+    @overload
+    def get(self, analysis: type[Analysis[R]], function: None = None) -> R: ...
+
+    @overload
+    def get(self, analysis: type[FunctionAnalysis[R]], function: nodes.Function) -> R: ...
+
+    def get(self, analysis: AnyAnalysis, function: nodes.Function | None = None) -> Any:
+        self._check_target(analysis, function)
+        if analysis not in self._registry:
+            raise UnregisteredAnalysisError(f"analysis {analysis.name!r} is not registered")
+        if self._computing and analysis not in self._computing[-1].requires:
+            current = self._computing[-1]
+            raise UndeclaredDependencyError(
+                f"{current.name} requests {analysis.name} without declaring it in requires"
+            )
+
+        key = self._key(analysis, function)
+        if key in self._cache:
+            return self._cache[key]
+
+        self._computing.append(analysis)
+        try:
+            if function is None:
+                assert issubclass(analysis, Analysis)
+                result = analysis.compute(self)
+            else:
+                assert issubclass(analysis, FunctionAnalysis)
+                result = analysis.compute(self, function)
+        finally:
+            self._computing.pop()
+        self._cache[key] = result
+        return result
+
+    def is_cached(self, analysis: AnyAnalysis, function: nodes.Function | None = None) -> bool:
+        self._check_target(analysis, function)
+        return self._key(analysis, function) in self._cache
+
+    # ------------------------------------------------------------------ invalidation
+
+    def run(self, transformation: type[TransformationPass]) -> None:
+        """Run a transformation, then drop every cached result it does not preserve."""
+
+        transformation.run(self)
+        self._cache = {
+            key: result for key, result in self._cache.items() if key[0] in transformation.preserves
+        }
+
+    # ------------------------------------------------------------------ helpers
+
+    @staticmethod
+    def _check_target(analysis: AnyAnalysis, function: nodes.Function | None) -> None:
+        if issubclass(analysis, FunctionAnalysis) and function is None:
+            raise TypeError(f"{analysis.name} is a function analysis and needs a function")
+        if issubclass(analysis, Analysis) and function is not None:
+            raise TypeError(f"{analysis.name} is a module analysis and takes no function")
+
+    @staticmethod
+    def _key(analysis: AnyAnalysis, function: nodes.Function | None) -> _CacheKey:
+        return (analysis, None if function is None else function.span)

src/coretrace_python/analysis/provider.py

@@ -0,0 +1,73 @@
+"""Typed analysis providers and the context they compute in (architecture §8, §12).
+
+An analysis is a class, not an instance: its ``name`` and ``version`` identify cached
+results, its ``requires`` declares the dependency DAG, and ``compute`` builds the
+result from an ``AnalysisContext``. Read-only analyses never modify shared IR; a
+``TransformationPass`` may, and must say what it preserves.
+"""
+
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from typing import Any, ClassVar, Generic, Protocol, TypeVar, overload
+
+from coretrace_python.hir import nodes
+
+R = TypeVar("R")
+
+
+class Analysis(ABC, Generic[R]):
+    """A module-level analysis producing one immutable result per module."""
+
+    name: ClassVar[str]
+    version: ClassVar[int] = 1
+    requires: ClassVar[frozenset[AnyAnalysis]] = frozenset()
+
+    @classmethod
+    @abstractmethod
+    def compute(cls, ctx: AnalysisContext) -> R:
+        raise NotImplementedError
+
+
+class FunctionAnalysis(ABC, Generic[R]):
+    """A function-level analysis producing one result per function, on demand."""
+
+    name: ClassVar[str]
+    version: ClassVar[int] = 1
+    requires: ClassVar[frozenset[AnyAnalysis]] = frozenset()
+
+    @classmethod
+    @abstractmethod
+    def compute(cls, ctx: AnalysisContext, function: nodes.Function) -> R:
+        raise NotImplementedError
+
+
+AnyAnalysis = type[Analysis[Any]] | type[FunctionAnalysis[Any]]
+
+
+class TransformationPass(ABC):
+    """A pass that may change shared state and therefore invalidates cached results.
+
+    Every cached analysis not listed in ``preserves`` is dropped after the pass runs.
+    """
+
+    name: ClassVar[str]
+    preserves: ClassVar[frozenset[AnyAnalysis]] = frozenset()
+
+    @classmethod
+    @abstractmethod
+    def run(cls, ctx: AnalysisContext) -> None:
+        raise NotImplementedError
+
+
+class AnalysisContext(Protocol):
+    """What an analysis sees while computing: the module and its declared dependencies."""
+
+    @property
+    def module(self) -> nodes.Module: ...
+
+    @overload
+    def get(self, analysis: type[Analysis[R]], function: None = None) -> R: ...
+
+    @overload
+    def get(self, analysis: type[FunctionAnalysis[R]], function: nodes.Function) -> R: ...

src/coretrace_python/ir/lowering.py

@@ -3,8 +3,15 @@
 from __future__ import annotations
 
 from dataclasses import dataclass, field
-from typing import NoReturn
-
+from typing import ClassVar, NoReturn
+
+from coretrace_python.analysis import (
+    Analysis,
+    AnalysisContext,
+    AnalysisManager,
+    AnyAnalysis,
+    FunctionAnalysis,
+)
 from coretrace_python.hir import nodes
 from coretrace_python.ir.model import (
     BasicBlock,
@@ -25,15 +32,15 @@
     Value,
     ValueInstruction,
 )
-from coretrace_python.semantic.imports import analyze_imports
+from coretrace_python.semantic import SEMANTIC_ANALYSES
 from coretrace_python.semantic.scopes import (
     Resolution,
     ResolutionKind,
     Scope,
     ScopeAnalysis,
-    analyze_scopes,
+    ScopeTable,
 )
-from coretrace_python.semantic.symbols import SymbolAnalysis, SymbolId, analyze_symbols
+from coretrace_python.semantic.symbols import SymbolAnalysis, SymbolId, SymbolTable
 
 
 class LoweringError(Exception):
@@ -42,8 +49,8 @@ class LoweringError(Exception):
 
 @dataclass
 class _FunctionLowerer:
-    symbols: SymbolAnalysis
-    scopes: ScopeAnalysis
+    symbols: SymbolTable
+    scopes: ScopeTable
     scope: Scope
     next_value_id: int = 0
     assigned: set[str] = field(default_factory=set)
@@ -151,26 +158,51 @@ def function(self, node: nodes.Function) -> FunctionIR:
         return FunctionIR(node.name, parameters, (self.block,), node.span)
 
 
+class PyIRAnalysis(FunctionAnalysis[FunctionIR]):
+    """Lower one function to PyIR on demand."""
+
+    name: ClassVar[str] = "ir.pyir"
+    requires: ClassVar[frozenset[AnyAnalysis]] = frozenset({ScopeAnalysis, SymbolAnalysis})
+
+    @classmethod
+    def compute(cls, ctx: AnalysisContext, function: nodes.Function) -> FunctionIR:
+        scopes = ctx.get(ScopeAnalysis)
+        lowerer = _FunctionLowerer(ctx.get(SymbolAnalysis), scopes, scopes.scope_for(function))
+        return lowerer.function(function)
+
+
+class ModuleIRAnalysis(Analysis[ModuleIR]):
+    """Assemble the PyIR of every top-level function of the module."""
+
+    name: ClassVar[str] = "ir.module"
+    requires: ClassVar[frozenset[AnyAnalysis]] = frozenset({PyIRAnalysis})
+
+    @classmethod
+    def compute(cls, ctx: AnalysisContext) -> ModuleIR:
+        functions: list[FunctionIR] = []
+        for statement in ctx.module.body:
+            if isinstance(statement, nodes.Function):
+                functions.append(ctx.get(PyIRAnalysis, statement))
+            elif isinstance(statement, (nodes.Import, nodes.ImportFrom)):
+                continue
+            elif (
+                isinstance(statement, nodes.ExpressionStatement)
+                and isinstance(statement.expression, nodes.Constant)
+                and isinstance(statement.expression.value, str)
+            ):
+                # Permit module docstrings.
+                continue
+            else:
+                raise LoweringError(
+                    f"{statement.span.display()}: "
+                    f"unsupported module syntax: {type(statement).__name__}"
+                )
+        return ModuleIR(tuple(functions))
+
+
 def lower_module(module: nodes.Module) -> ModuleIR:
-    scopes = analyze_scopes(module)
-    symbols = analyze_symbols(scopes, analyze_imports(module, scopes))
-    functions: list[FunctionIR] = []
-    for statement in module.body:
-        if isinstance(statement, nodes.Function):
-            lowerer = _FunctionLowerer(symbols, scopes, scopes.scope_for(statement))
-            functions.append(lowerer.function(statement))
-        elif isinstance(statement, (nodes.Import, nodes.ImportFrom)):
-            continue
-        elif (
-            isinstance(statement, nodes.ExpressionStatement)
-            and isinstance(statement.expression, nodes.Constant)
-            and isinstance(statement.expression.value, str)
-        ):
-            # Permit module docstrings.
-            continue
-        else:
-            raise LoweringError(
-                f"{statement.span.display()}: "
-                f"unsupported module syntax: {type(statement).__name__}"
-            )
-    return ModuleIR(tuple(functions))
+    """Lower a whole module through a fresh analysis manager."""
+
+    manager = AnalysisManager(module)
+    manager.register(*SEMANTIC_ANALYSES, PyIRAnalysis, ModuleIRAnalysis)
+    return manager.get(ModuleIRAnalysis)

src/coretrace_python/semantic/__init__.py

@@ -1 +1,9 @@
-"""Semantic analyses computed from PyHIR: imports, symbols and lexical scopes."""
+"""Semantic analyses computed from PyHIR: scopes, imports and symbols."""
+
+from coretrace_python.semantic.imports import ImportAnalysis
+from coretrace_python.semantic.scopes import ScopeAnalysis
+from coretrace_python.semantic.symbols import SymbolAnalysis
+
+SEMANTIC_ANALYSES = (ScopeAnalysis, ImportAnalysis, SymbolAnalysis)
+
+__all__ = ["SEMANTIC_ANALYSES", "ImportAnalysis", "ScopeAnalysis", "SymbolAnalysis"]