Merge pull request #101 from codellm-devkit/fix/issue-99-deterministic-output

fix(determinism): byte-identical analysis output for identical runs

Author: rahlk

CHANGELOG.md

@@ -7,6 +7,27 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Fixed
+- **Deterministic output** (#99): the same flags on the same input now emit byte-identical
+  `analysis.json`. Four independent nondeterminism sources were closed:
+  - the CLI re-execs once with `PYTHONHASHSEED=0` (export the variable to opt out or pin another
+    seed) — PyCG's capped fixpoint iterates hash-ordered sets, so an unpinned per-interpreter
+    seed shifted its frontier arbitrarily (observed 1,527 vs 4,712 edges on the same input);
+    Ray workers get the pinned seed via the runtime env;
+  - the PyCG mini-project root is content-derived instead of a random `mkdtemp` (PyCG state keys
+    on absolute module paths), with an exclusive sidecar lock so concurrent analyses of the same
+    project serialize instead of deleting each other's tree mid-run;
+  - entry points are sorted (was filesystem order) and the emitted `call_graph` is canonically
+    ordered by `(src, dst)`;
+  - Jedi inference candidates are tie-broken deterministically (sorted, not set order) — a
+    union-typed receiver (e.g. `IOBase | BufferedRandom | TextIOWrapper`) previously resolved to
+    a different member per run.
+  Regression gates: `test_l2_runs_are_byte_identical` plus the #87 audit gate below.
+- **L2 audit gate** (#87 acceptance): ≥95% of resolved non-constructor callsites must bind to the
+  invoked attribute name, and no callsite may fall back to a class id when the class declares the
+  exact method (`test_l2_audit_gate_callee_name_equality`). The underlying anchoring fix shipped
+  in 1.0.0 via the v0.3.1 merge.
+
 ## [1.0.0] - 2026-07-14
 
 ### Added

codeanalyzer/__main__.py

@@ -1,9 +1,40 @@
+import os
+import sys
 from importlib.metadata import version as _pkg_version, PackageNotFoundError
 from pathlib import Path
 from typing import Optional, Annotated
 
 import typer
 
+
+def _pin_hash_seed() -> None:
+    """Re-exec once with ``PYTHONHASHSEED=0`` unless the caller pinned one.
+
+    PyCG's capped fixpoint (``--pycg-max-iter``) iterates hash-ordered sets
+    keyed on module/access-path strings, so an unpinned per-interpreter hash
+    seed makes the emitted L2+ call graph vary run to run (issue #99). The
+    seed cannot be set after interpreter start, hence the exec. Export
+    PYTHONHASHSEED (any value) to opt out or pin a different seed.
+
+    Only fires when this process really is the CLI (canpy / python -m
+    codeanalyzer): in-process invocations — e.g. Typer's CliRunner in the
+    test suite, or a host app calling the callback — must never have their
+    own process exec'd out from under them."""
+    if os.environ.get("PYTHONHASHSEED") is not None:
+        return
+    argv0 = os.path.basename(sys.argv[0]) if sys.argv else ""
+    is_cli = argv0 in ("canpy", "codeanalyzer") or sys.argv[0].endswith(
+        os.path.join("codeanalyzer", "__main__.py")
+    )
+    if not is_cli:
+        return
+    env = dict(os.environ, PYTHONHASHSEED="0")
+    os.execvpe(
+        sys.executable,
+        [sys.executable, "-m", "codeanalyzer", *sys.argv[1:]],
+        env,
+    )
+
 from codeanalyzer.core import Codeanalyzer
 from codeanalyzer.utils import _set_log_level, logger
 from codeanalyzer.config import OutputFormat
@@ -264,6 +295,10 @@ def main(
         ),
     ] = 50,
 ):
+    # Determinism: pin the interpreter hash seed before any analysis (no-op
+    # when PYTHONHASHSEED is already set; --version exits before this).
+    _pin_hash_seed()
+
     # Flag validation (strict: unrecognized values error out, never fall back).
     selected_graphs = [g.strip() for g in graphs.split(",") if g.strip()]
     from codeanalyzer.dataflow.builder import VALID_GRAPHS

codeanalyzer/core.py

@@ -37,6 +37,22 @@
 from codeanalyzer.options import AnalysisOptions
 from codeanalyzer.provenance import analyzer_info, repository_info
 
+def _ensure_ray() -> None:
+    """Initialize Ray with the driver's pinned hash seed in the workers.
+
+    An implicit auto-init would not carry PYTHONHASHSEED into worker
+    interpreters, so PyCG shards (and Jedi inference) run there with random
+    set-iteration order and the emitted edges vary run to run (issue #99)."""
+    if not ray.is_initialized():
+        ray.init(
+            runtime_env={
+                "env_vars": {
+                    "PYTHONHASHSEED": os.environ.get("PYTHONHASHSEED", "0")
+                }
+            },
+        )
+
+
 @ray.remote
 def _process_file_with_ray(py_file: Union[Path, str], project_dir: Union[Path, str], virtualenv: Union[Path, str, None]) -> Dict[str, PyModule]:
     """Processes files in the project directory using Ray for distributed processing.
@@ -438,6 +454,11 @@ def analyze(self) -> Analysis:
             call_graph = merge_edges(call_graph, pycg_edges)
 
         call_graph = filter_external_edges(call_graph, symbol_table)
+        # Canonical edge order: backend iteration order (PyCG dicts, Counter
+        # insertion) is not a contract — sort so identical edge SETS always
+        # serialize identically (issue #99 determinism gate), and so the
+        # external-symbol homing below assigns ids in a stable order.
+        call_graph.sort(key=lambda e: (e.src, e.dst))
 
         # Recreate pyapplication
         app = (
@@ -716,6 +737,7 @@ def _build_symbol_table(self, cached_symbol_table: Optional[Dict[str, PyModule]]
 
             # Process only new/changed files with Ray
             if files_to_process:
+                _ensure_ray()
                 futures = [_process_file_with_ray.remote(py_file, self.project_dir, str(self.virtualenv) if self.virtualenv else None) for py_file in files_to_process]
 
                 with ProgressBar(len(futures), "Building symbol table (parallel)") as progress:

codeanalyzer/semantic_analysis/pycg/pycg_analysis.py

@@ -39,9 +39,12 @@
 # re-enters PyCG's hook before its import graph is ready.  Pre-importing
 # these modules at import time ensures they're already in sys.modules when
 # PyCG's hook is active, preventing the re-entrant ImportManagerError.
+import fcntl
+import hashlib
 import importlib.metadata  # noqa: F401
 import importlib.util  # noqa: F401
 import contextlib
+import os
 import json  # noqa: F401
 import shutil
 import signal
@@ -85,6 +88,15 @@ def _handler(signum: int, frame: object) -> None:
 from codeanalyzer.utils import ProgressBar, logger
 
 
+def _shard_root_path(files: List[str], project_dir: Path) -> Path:
+    """Content-derived mini-project root for a shard: same project + same file
+    set → same path on every run (determinism, issue #99)."""
+    digest = hashlib.sha1(
+        "\0".join([str(project_dir), *sorted(files)]).encode("utf-8")
+    ).hexdigest()[:16]
+    return Path(tempfile.gettempdir()) / f"canpy_pycg_shard_{digest}"
+
+
 def _materialize_shard_root(
     files: List[str],
     project_dir: Path,
@@ -103,10 +115,21 @@ def _materialize_shard_root(
 
     The caller owns the returned *root* and must ``shutil.rmtree`` it.
     """
-    root = Path(tempfile.mkdtemp(prefix="canpy_pycg_shard_"))
+    # Deterministic root: PyCG's capped fixpoint (--pycg-max-iter) is
+    # order-sensitive, and its internal state keys on absolute module paths —
+    # a random mkdtemp suffix changes those strings every run and shifts the
+    # iteration frontier, making the emitted edge set vary run-to-run
+    # (issue #99). Deriving the directory name from the shard's content keeps
+    # the path (and thus the analysis input) identical across runs. Callers
+    # that may run concurrently on the same shard serialize on the sidecar
+    # lock (see _shard_symlink_root).
+    root = _shard_root_path(files, project_dir)
+    if root.exists():
+        shutil.rmtree(root, ignore_errors=True)
+    root.mkdir(parents=True, exist_ok=True)
     entry_points: List[str] = []
     linked_inits: Set[Path] = set()
-    for f in files:
+    for f in sorted(files):
         src = Path(f).resolve()
         try:
             rel = src.relative_to(project_dir)
@@ -142,12 +165,27 @@ def _shard_symlink_root(
     """Context-manager wrapper around :func:`_materialize_shard_root`.
 
     Yields ``(root, entry_points)`` and removes the temp tree on exit.
+
+    The root path is content-derived (determinism, issue #99), so two
+    concurrent analyses of the same shard — e.g. a test suite and a manual
+    run on one project — would collide on it (one rmtree's the tree the
+    other is mid-analysis on). An exclusive flock on a sidecar lockfile
+    serializes them; distinct projects/shards hash to distinct roots and
+    never contend.
     """
-    root, entry_points = _materialize_shard_root(files, project_dir)
+    digest_root = _shard_root_path(files, project_dir)
+    lock_path = digest_root.with_name(digest_root.name + ".lock")
+    lock_fd = os.open(str(lock_path), os.O_CREAT | os.O_RDWR)
     try:
-        yield root, entry_points
+        fcntl.flock(lock_fd, fcntl.LOCK_EX)
+        root, entry_points = _materialize_shard_root(files, project_dir)
+        try:
+            yield root, entry_points
+        finally:
+            shutil.rmtree(root, ignore_errors=True)
     finally:
-        shutil.rmtree(root, ignore_errors=True)
+        fcntl.flock(lock_fd, fcntl.LOCK_UN)
+        os.close(lock_fd)
 
 
 def _pycg_shard_worker(
@@ -469,7 +507,9 @@ def _collect_entry_points(self) -> List[str]:
             ):
                 continue
             paths.append(str(p))
-        return paths
+        # Sorted for run-to-run stability: rglob yields filesystem order, and
+        # PyCG's capped fixpoint is sensitive to entry-point order (issue #99).
+        return sorted(paths)
 
     # ------------------------------------------------------------------
     # Package-root helpers for sharding
@@ -712,18 +752,31 @@ def _run_fileset_shards_ray(
         """
         import os
         import ray
+        from codeanalyzer.core import _ensure_ray
+        _ensure_ray()
 
         os.environ.setdefault("RAY_IGNORE_UNHANDLED_ERRORS", "1")
         remote_fn = ray.remote(_pycg_shard_worker)
 
         roots: List[Path] = []
+        lock_fds: List[int] = []
         futures: List[Any] = []
         meta: Dict[Any, List[str]] = {}  # ObjectRef -> shard file list
         edges_all: List[PyCallEdge] = []
         runaways: List[List[str]] = []
         try:
             with ProgressBar(len(shards), "Building call graph shards (parallel)", item_label="shards") as progress:
                 for files in shards:
+                    # Deterministic roots can collide across concurrent
+                    # analyses of the same project — the driver holds each
+                    # shard's sidecar lock for the whole Ray fan-out (released
+                    # in the finally below with the root cleanup).
+                    lock_fd = os.open(
+                        str(_shard_root_path(files, self.project_dir).with_suffix(".lock")),
+                        os.O_CREAT | os.O_RDWR,
+                    )
+                    fcntl.flock(lock_fd, fcntl.LOCK_EX)
+                    lock_fds.append(lock_fd)
                     root, eps = _materialize_shard_root(files, self.project_dir)
                     roots.append(root)
                     fut = remote_fn.remote(eps, str(root), "", self.max_iter)
@@ -765,6 +818,12 @@ def _run_fileset_shards_ray(
         finally:
             for root in roots:
                 shutil.rmtree(root, ignore_errors=True)
+            for fd in lock_fds:
+                try:
+                    fcntl.flock(fd, fcntl.LOCK_UN)
+                    os.close(fd)
+                except OSError:
+                    pass
         return edges_all, runaways
 
     def _build_sharded(
@@ -870,6 +929,8 @@ def _build_sharded_ray(self, shards: Dict[Path, List[str]]) -> List[PyCallEdge]:
         """
         import os
         import ray
+        from codeanalyzer.core import _ensure_ray
+        _ensure_ray()
 
         # force-cancel kills worker processes; suppress Ray's "worker died
         # unexpectedly" noise since the death is intentional here.

codeanalyzer/syntactic_analysis/symbol_table_builder.py

@@ -57,13 +57,26 @@ def _fallback_signature(self, script_path: Union[Path, str], name: str) -> str:
         relative = Path(script_path).relative_to(self.project_dir)
         return ".".join(relative.with_suffix("").parts) + f".{name}"
 
+    @staticmethod
+    def _first_definition(definitions):
+        """Deterministic pick from Jedi's inference candidates.
+
+        On a union-typed receiver Jedi may return several candidates whose
+        ORDER varies run to run (issue #99 — e.g. ``o.seek`` on
+        ``_IOBase | BufferedRandom | TextIOWrapper``); taking whichever came
+        first made the emitted call graph nondeterministic. Sort on the
+        stable identity (full_name, then name) and take the smallest."""
+        if not definitions:
+            return None
+        return min(definitions, key=lambda d: (d.full_name or "", d.name or ""))
+
     @staticmethod
     def _infer_type(script: Script, line: int, column: int) -> str:
         """Tries to infer the type at a given position using Jedi."""
         try:
-            inference = script.infer(line=line, column=column)
-            if inference:
-                return inference[0].name  # or .full_name
+            d = SymbolTableBuilder._first_definition(script.infer(line=line, column=column))
+            if d is not None:
+                return d.name  # or .full_name
         except Exception:
             pass
         return None
@@ -82,9 +95,9 @@ def _infer_qualified_name(script: Script, line: int, column: int) -> Optional[st
             Optional[str]: The fully qualified name if available, else None.
         """
         try:
-            definitions = script.infer(line=line, column=column)
-            if definitions:
-                return definitions[0].full_name
+            d = SymbolTableBuilder._first_definition(script.infer(line=line, column=column))
+            if d is not None:
+                return d.full_name
         except Exception:
             pass
         return None
@@ -103,10 +116,9 @@ def _infer_callee(
         the call graph.
         """
         try:
-            definitions = script.infer(line=line, column=column)
-            if not definitions:
+            d = SymbolTableBuilder._first_definition(script.infer(line=line, column=column))
+            if d is None:
                 return None, False
-            d = definitions[0]
             is_class = (d.type == "class")
             full = d.full_name
             if is_class and full:
@@ -144,11 +156,17 @@ def _infer_call_return_type(script: Script, line: int, column: int) -> Optional[
         as the callee's own name.
         """
         try:
-            definitions = script.infer(line=line, column=column)
-            if definitions:
-                results = definitions[0].execute()
-                if results:
-                    return results[0].name
+            d = SymbolTableBuilder._first_definition(script.infer(line=line, column=column))
+            if d is not None:
+                # Drop NoneType results before picking: Jedi flaps between
+                # yielding {NoneType} and {} for the None arm of an Optional
+                # return (issue #99), and when a real type is also in the
+                # union it is the informative choice — a bare NoneType says
+                # nothing a missing return_type doesn't.
+                results = [r for r in d.execute() if r.name != "NoneType"]
+                r = SymbolTableBuilder._first_definition(results)
+                if r is not None:
+                    return r.name
         except Exception:
             pass
         return None
@@ -974,9 +992,10 @@ def _symbol_from_name_node(
 
         if script:
             try:
-                definitions = script.infer(line=lineno, column=col_offset)
-                if definitions:
-                    d = definitions[0]
+                d = SymbolTableBuilder._first_definition(
+                    script.infer(line=lineno, column=col_offset)
+                )
+                if d is not None:
                     inferred_type = d.name
                     qname = d.full_name
                     if d.type == "function":