scripts.check_graph_write_swallow

Reference › scripts › check_graph_write_swallow

The graph-write-swallow gate: a graph write failure must reach the caller. It fails a graph write inside a try body whose except handler cannot inform the caller of the failure, and it works as a ratchet against a baseline so existing findings are tolerated but new ones fail. tests/scripts/test_check_graph_write_swallow.py imports it.

Invocation: main takes one of three mutually exclusive modes: --check (the default; exit 1 on findings that are new against the baseline), --list (print every finding, including waived and baselined ones, and exit 0) and --update-baseline (rewrite the baseline to the current, non-waived findings). Further options are --repo-root, --baseline (default reports/graph-write-swallow-baseline.json under the repo root) and --roots (directories to scan).

scan parses each file and a _ScopeTracker visits every try statement. _try_has_graph_write and _call_is_graph_write decide whether the body writes to the graph, by resolving the Cypher text of a call (_resolve_cypher_text) and recognizing session-like receivers. _handler_informs_caller accepts a handler that re-raises or otherwise outlives the failure to the caller (_name_outlives_handler). A line may waive a finding (_is_waived). Each remaining Finding has a stable id for the baseline file (load_baseline, write_baseline).

At a glance

Class diagram

Dependency diagram

No internal imports and no importers.

API

Classes

Finding (dataclass)

Decorators: dataclass · Line 119

One flagged (or waived) try/except over a graph write.

_ScopeTracker

Bases: ast.NodeVisitor · Line 324

Walks a module tracking the enclosing function/class qualname, the enclosing function AST node (for the outlives/Cypher-variable resolution helpers), and a per-scope ordinal counter, collecting…

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/scripts/check_graph_write_swallow.py

#!/usr/bin/env python3
"""Graph-write-swallow gate — a graph write failure must reach the caller.

WHY: a failed graph write caught and reduced to a log line makes contention
on the shared Memgraph indistinguishable from a bug. The caller gets no
signal, a read-back returns empty, and the failing test asserts on nothing
("assert record is not None"). Two confirmed instances motivated this gate:
``knowledge_graph/synchronizer.py`` (fixed) and
``knowledge_graph/cxp_auditor.py`` (``log_packet``/``log_routing``, which
swallow to a bare ``print()``).

DETECTION RULE (ast-based, not regex). Flag an ``ast.Try``/``ast.TryStar``
when BOTH hold:

  (a) its ``body`` contains a graph WRITE:
        - an unconditional write call — ``execute_write``/``unit_of_work``
          (the method name is the contract; content never exempts them);
        - a ``session.run``/``execute_autocommit`` call whose Cypher
          argument, when statically resolvable (a literal, an f-string, an
          implicit/explicit concatenation of literals, or a single local
          variable assigned one of those forms earlier in scope), contains
          a mutating clause — CREATE / MERGE / DELETE / SET / REMOVE / DROP
          / DETACH (case-insensitive, word-boundary). When the argument
          canNOT be resolved statically (a call result, an unexplained
          parameter, ...), the call stays conservatively classified as a
          write — the gate must not go blind;
        - OR a string constant anywhere in the body matching MERGE /
          CREATE ( / DETACH DELETE / SET (case-insensitive, word-boundary)
          — independent of the two call-based checks above, e.g. a Cypher
          literal passed to a helper the gate doesn't otherwise recognize;
  (b) it has a handler that CANNOT inform the caller: no ``raise``, no
      ``return``, and no assignment/mutation that outlives the handler —
      an attribute assignment, a subscript assignment, ``global``/
      ``nonlocal``, a call to a mutating collection method such as
      ``.append``/``.add``/``.update``, OR an ``ast.Assign``/
      ``ast.AugAssign`` to a bare NAME that is not local to the handler
      (the name is bound earlier in the same function scope — a parameter,
      a prior assignment, a loop variable — before the ``try``, or the name
      is read inside a ``return`` anywhere in the function, so the
      handler's write to it can reach the caller). A handler consisting
      only of logging/``print`` calls and/or ``pass`` is the textbook
      failing case, but this is the general rule: anything that neither
      raises, returns, nor mutates caller-visible state is a swallow.

A handler that re-raises, returns a falsy/None-vs-value signal, sets a
flag/attribute, increments a counter the function later returns, or appends
to an errors collection is COMPLIANT — the point is that the caller can
OBSERVE the failure, not that it must raise.

EXEMPTION: the repo's existing ``# gate-ok: <reason>`` waiver convention
(see ``scripts/gate_parallel_safety.py``) — a comment on the flagged
``try``/``try*`` line, or the line directly above it, waives that finding. A
reason is required; a bare ``# gate-ok:`` does not waive.

BASELINE / RATCHET (mirrors ``scripts/pyright_ratchet.py``'s "fail only on
NEW findings" contract (now in codetools:
src/codetools/refactor/pyright/ratchet.py), adapted to identifier-keyed findings the way
``scripts/check_ai_attribution.py`` keys its baseline on commit SHAs rather
than line numbers): findings are stable-identified by
``<relpath>:<enclosing-scope-qualname>:<ordinal-within-scope>`` — never a
raw line number, since those shift on unrelated edits. The baseline lives at
``reports/graph-write-swallow-baseline.json``. ``--check`` (default) exits 1
only for findings whose id is absent from the baseline; baselined findings
are reported but do not fail. ``--update-baseline`` rewrites the baseline to
the current (non-waived) finding set. ``--list`` prints every finding,
including waived and baselined ones, and always exits 0.

Roots scanned: ``src/`` and ``scripts/``.
"""

from __future__ import annotations

import argparse
import ast
import json
import re
from dataclasses import dataclass
from pathlib import Path

REPO_ROOT = Path(__file__).resolve().parent.parent
DEFAULT_ROOTS = ("src", "scripts")

# ``# gate-ok: <reason>`` on the flagged line (or the line directly above it)
# waives that one finding. The reason is mandatory — ``\S`` refuses bare tags.
# (same convention + regex as scripts/gate_parallel_safety.py)
_WAIVER_RE = re.compile(r"#\s*gate-ok:\s*\S")

# Unconditional write calls — the method name IS the contract; Cypher
# content never exempts them (a SELECT run inside execute_write is still a
# managed-transaction write call as far as the caller's failure-handling
# contract is concerned).
_UNCONDITIONAL_WRITE_CALL_ATTRS = {"execute_write", "unit_of_work"}
# ``session.run``/``execute_autocommit`` are classified by CONTENT (see
# _call_is_graph_write) since both are also used for reads.
_CONTENT_CLASSIFIED_CALL_ATTRS = {"execute_autocommit"}
_CYPHER_WRITE_RE = re.compile(
    r"\bMERGE\b|\bCREATE\s*\(|\bDETACH\s+DELETE\b|\bSET\b", re.IGNORECASE
)
# Content classifier for a resolved session.run/execute_autocommit Cypher
# string — broader than _CYPHER_WRITE_RE (adds REMOVE/DROP) since this is
# the sole signal once a call is known to target a session/autocommit path.
_MUTATING_CLAUSE_RE = re.compile(
    r"\b(?:CREATE|MERGE|DELETE|SET|REMOVE|DROP|DETACH)\b", re.IGNORECASE
)
# Calls on a mutating collection (``errors.append(exc)``) let a handler's
# side effect outlive it, same as an attribute/subscript assignment does.
_OUTLIVING_CALL_ATTRS = {
    "append",
    "add",
    "update",
    "extend",
    "put",
    "setdefault",
    "insert",
}


@dataclass
class Finding:
    """One flagged (or waived) try/except over a graph write."""

    path: str  # relpath, forward slashes
    scope: str  # enclosing qualname, "<module>" at top level
    ordinal: int  # 1-based occurrence within (path, scope) — NOT a line number
    lineno: int  # for human display only; never part of the identifier
    detail: str
    waived: bool = False

    @property
    def finding_id(self) -> str:
        return f"{self.path}:{self.scope}:{self.ordinal}"

    def as_dict(self) -> dict:
        return {
            "id": self.finding_id,
            "path": self.path,
            "scope": self.scope,
            "line_hint": self.lineno,
            "detail": self.detail,
        }


def _is_waived(lines: list[str], lineno: int) -> bool:
    """True if the 1-based *lineno* (or the line above) carries a gate-ok waiver."""
    for idx in (lineno - 1, lineno - 2):
        if 0 <= idx < len(lines) and _WAIVER_RE.search(lines[idx]):
            return True
    return False


def _is_session_like(value: ast.AST) -> bool:
    """True if *value* looks like a Neo4j/Memgraph driver session object —
    a ``Name`` or attribute access whose identifier contains "session"."""
    if isinstance(value, ast.Name):
        return "session" in value.id.lower()
    if isinstance(value, ast.Attribute):
        return "session" in value.attr.lower()
    return False


def _resolve_cypher_text(
    expr: ast.expr | None, container: ast.AST, before_lineno: int, _depth: int = 0
) -> str | None:
    """Best-effort static resolution of a Cypher argument to its literal
    text — enough to classify read vs. write by keyword content, never a
    real interpreter. Handles: a literal string, an f-string (only its
    literal segments — an interpolated ``{...}`` expression contributes
    nothing towards the match, since a mutating clause is never itself
    interpolated in this codebase), an implicit/explicit ``+``
    concatenation of resolvable parts, or a single local variable assigned
    one of those forms earlier (by line number) in *container*. Anything
    else — a call result, an unresolved parameter, an ambiguous multi-
    assignment — returns ``None`` so the caller stays conservative.
    """
    if expr is None or _depth > 8:
        return None
    if isinstance(expr, ast.Constant) and isinstance(expr.value, str):
        return expr.value
    if isinstance(expr, ast.JoinedStr):
        return "".join(
            v.value
            for v in expr.values
            if isinstance(v, ast.Constant) and isinstance(v.value, str)
        )
    if isinstance(expr, ast.BinOp) and isinstance(expr.op, ast.Add):
        left = _resolve_cypher_text(expr.left, container, before_lineno, _depth + 1)
        right = _resolve_cypher_text(expr.right, container, before_lineno, _depth + 1)
        if left is None or right is None:
            return None
        return left + right
    if isinstance(expr, ast.Name):
        candidate: ast.Assign | None = None
        for node in ast.walk(container):
            if not isinstance(node, ast.Assign) or len(node.targets) != 1:
                continue
            target = node.targets[0]
            if not (isinstance(target, ast.Name) and target.id == expr.id):
                continue
            if node.lineno > before_lineno:
                continue
            if candidate is None or node.lineno > candidate.lineno:
                candidate = node
        if candidate is None:
            return None
        return _resolve_cypher_text(
            candidate.value, container, before_lineno, _depth + 1
        )
    return None


def _call_is_graph_write(call: ast.Call, container: ast.AST) -> bool:
    """True if *call* is a graph write — the ``execute_write``/
    ``unit_of_work`` methods unconditionally, ``session.run``/
    ``execute_autocommit`` by resolved Cypher content (conservatively a
    write when the content can't be resolved)."""
    if not isinstance(call.func, ast.Attribute):
        return False
    attr = call.func.attr
    if attr in _UNCONDITIONAL_WRITE_CALL_ATTRS:
        return True
    content_classified = attr in _CONTENT_CLASSIFIED_CALL_ATTRS or (
        attr == "run" and _is_session_like(call.func.value)
    )
    if not content_classified:
        return False
    cypher_arg = call.args[0] if call.args else None
    text = _resolve_cypher_text(cypher_arg, container, call.lineno)
    if text is None:
        return True  # unresolvable Cypher argument — stay conservative
    return bool(_MUTATING_CLAUSE_RE.search(text))


def _try_has_graph_write(body: list[ast.stmt], container: ast.AST) -> bool:
    for stmt in body:
        for sub in ast.walk(stmt):
            if isinstance(sub, ast.Call) and _call_is_graph_write(sub, container):
                return True
            if (
                isinstance(sub, ast.Constant)
                and isinstance(sub.value, str)
                and _CYPHER_WRITE_RE.search(sub.value)
            ):
                return True
    return False


def _assign_targets(node: ast.Assign | ast.AugAssign | ast.AnnAssign) -> list[ast.expr]:
    if isinstance(node, ast.Assign):
        return list(node.targets)
    return [node.target]


def _name_outlives_handler(name: str, container: ast.AST, before_lineno: int) -> bool:
    """True if *name* is bound elsewhere in *container* (a function body, or
    the module body for module-level code) in a way that lets a handler's
    assignment/aug-assignment to that bare NAME reach the caller:

      - *name* is bound before *before_lineno* (the enclosing ``try``'s line)
        by a prior assignment, a function parameter, or a ``for``/``async
        for`` loop variable — the handler is updating an EXISTING binding
        the rest of the function (or a caller closing over it) can observe;
      - OR *name* is read inside any ``return`` statement's value anywhere
        in *container* — the handler's write can reach the caller through
        that return, regardless of where the binding started.

    A conservative ast-level approximation ("bound-before-try OR read-in-a-
    return"), not full dataflow analysis — sufficient to distinguish the
    real motivating case (a counter incremented in a loop's except clause,
    returned in a report dataclass once the loop ends) from a throwaway
    local that dies with the handler.
    """
    for node in ast.walk(container):
        if isinstance(node, ast.Return) and node.value is not None:
            for sub in ast.walk(node.value):
                if isinstance(sub, ast.Name) and sub.id == name:
                    return True
    for node in ast.walk(container):
        lineno = getattr(node, "lineno", None)
        if lineno is None or lineno >= before_lineno:
            continue
        if isinstance(node, (ast.Assign, ast.AugAssign, ast.AnnAssign)):
            for target in _assign_targets(node):
                if isinstance(target, ast.Name) and target.id == name:
                    return True
        elif isinstance(node, (ast.For, ast.AsyncFor)) and isinstance(
            node.target, ast.Name
        ):
            if node.target.id == name:
                return True
        elif isinstance(node, ast.arg) and node.arg == name:
            return True
    return False


def _handler_informs_caller(
    handler: ast.ExceptHandler, container: ast.AST, try_lineno: int
) -> bool:
    """True if *handler* gives the caller some way to observe the failure:
    a re-raise, a return, an outliving assignment/mutation to an attribute/
    subscript/non-local NAME, a global/nonlocal declaration, or a call to a
    mutating collection method."""
    for stmt in handler.body:
        for sub in ast.walk(stmt):
            if isinstance(sub, ast.Raise):
                return True
            if isinstance(sub, ast.Return):
                return True
            if isinstance(sub, (ast.Global, ast.Nonlocal)):
                return True
            if isinstance(sub, (ast.Assign, ast.AugAssign, ast.AnnAssign)):
                for target in _assign_targets(sub):
                    if isinstance(target, (ast.Attribute, ast.Subscript)):
                        return True
                    if isinstance(target, ast.Name) and _name_outlives_handler(
                        target.id, container, try_lineno
                    ):
                        return True
            if isinstance(sub, ast.Call) and isinstance(sub.func, ast.Attribute):
                if sub.func.attr in _OUTLIVING_CALL_ATTRS:
                    return True
    return False


class _ScopeTracker(ast.NodeVisitor):
    """Walks a module tracking the enclosing function/class qualname, the
    enclosing function AST node (for the outlives/Cypher-variable
    resolution helpers), and a per-scope ordinal counter, collecting
    graph-write-swallow findings."""

    def __init__(self, relpath: str, lines: list[str], module: ast.Module) -> None:
        self.relpath = relpath
        self.lines = lines
        self._module = module
        self._stack: list[str] = []
        self._func_stack: list[ast.FunctionDef | ast.AsyncFunctionDef] = []
        self._counters: dict[str, int] = {}
        self.findings: list[Finding] = []

    def _scope_name(self) -> str:
        return ".".join(self._stack) if self._stack else "<module>"

    def _container(self) -> ast.AST:
        """The nearest enclosing function, or the module for top-level code
        — the scope ``_resolve_cypher_text``/``_name_outlives_handler``
        search for prior bindings/returns."""
        return self._func_stack[-1] if self._func_stack else self._module

    def _push(self, name: str, node: ast.AST) -> None:
        self._stack.append(name)
        self.generic_visit(node)
        self._stack.pop()

    def visit_FunctionDef(self, node: ast.FunctionDef) -> None:
        self._func_stack.append(node)
        self._push(node.name, node)
        self._func_stack.pop()

    def visit_AsyncFunctionDef(self, node: ast.AsyncFunctionDef) -> None:
        self._func_stack.append(node)
        self._push(node.name, node)
        self._func_stack.pop()

    def visit_ClassDef(self, node: ast.ClassDef) -> None:
        self._push(node.name, node)

    def _check_try(self, node: ast.Try | ast.TryStar) -> None:
        container = self._container()
        if _try_has_graph_write(node.body, container):
            noncompliant = [
                h
                for h in node.handlers
                if not _handler_informs_caller(h, container, node.lineno)
            ]
            if noncompliant:
                scope = self._scope_name()
                self._counters[scope] = self._counters.get(scope, 0) + 1
                ordinal = self._counters[scope]
                waived = _is_waived(self.lines, node.lineno)
                self.findings.append(
                    Finding(
                        path=self.relpath,
                        scope=scope,
                        ordinal=ordinal,
                        lineno=node.lineno,
                        detail=(
                            "graph write in try body swallowed by a handler "
                            "that cannot inform the caller"
                        ),
                        waived=waived,
                    )
                )
        self.generic_visit(node)

    def visit_Try(self, node: ast.Try) -> None:
        self._check_try(node)

    def visit_TryStar(self, node: ast.TryStar) -> None:
        self._check_try(node)


def _scan_file(py_file: Path, repo_root: Path) -> list[Finding]:
    try:
        source = py_file.read_text()
    except OSError, UnicodeDecodeError:
        return []
    try:
        tree = ast.parse(source, filename=str(py_file))
    except SyntaxError:
        return []
    lines = source.splitlines()
    relpath = py_file.relative_to(repo_root).as_posix()
    tracker = _ScopeTracker(relpath, lines, tree)
    tracker.visit(tree)
    return tracker.findings


def scan(roots: list[Path], repo_root: Path = REPO_ROOT) -> list[Finding]:
    """Scan every ``*.py`` file under *roots* (relative to *repo_root* unless
    already absolute) and return the findings, sorted by path."""
    findings: list[Finding] = []
    for root in roots:
        root_path = root if root.is_absolute() else repo_root / root
        if not root_path.exists():
            continue
        for py_file in sorted(root_path.rglob("*.py")):
            findings.extend(_scan_file(py_file, repo_root))
    return findings


def load_baseline(path: Path) -> set[str]:
    if not path.exists():
        return set()
    data = json.loads(path.read_text())
    return {entry["id"] for entry in data.get("findings", [])}


def write_baseline(path: Path, findings: list[Finding]) -> None:
    payload = {
        "generated_by": "scripts/check_graph_write_swallow.py --update-baseline",
        "findings": [f.as_dict() for f in sorted(findings, key=lambda f: f.finding_id)],
    }
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(payload, indent=2) + "\n")


def _print_finding(f: Finding, tag: str) -> None:
    print(f"[{tag}] {f.path}:{f.lineno} ({f.scope}) — {f.detail} [id={f.finding_id}]")


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(
        description=(
            "Gate: fail a graph write in a try body whose except handler "
            "cannot inform the caller of failure (ratchet vs. baseline)."
        )
    )
    mode = parser.add_mutually_exclusive_group()
    mode.add_argument(
        "--check",
        action="store_true",
        help="Ratchet check: exit 1 on findings NEW vs. the baseline (default).",
    )
    mode.add_argument(
        "--list",
        action="store_true",
        help="Print every finding, including waived/baselined ones. Exit 0.",
    )
    mode.add_argument(
        "--update-baseline",
        action="store_true",
        help="Rewrite the baseline to the current (non-waived) finding set.",
    )
    parser.add_argument(
        "--repo-root",
        type=Path,
        default=REPO_ROOT,
        help=f"Repo root to scan/relativize against (default: {REPO_ROOT}).",
    )
    parser.add_argument(
        "--baseline",
        type=Path,
        default=None,
        help=(
            "Baseline JSON path (default: <repo-root>/reports/"
            "graph-write-swallow-baseline.json)."
        ),
    )
    parser.add_argument(
        "--roots",
        nargs="+",
        default=list(DEFAULT_ROOTS),
        help=f"Directories to scan, relative to --repo-root (default: {DEFAULT_ROOTS}).",
    )
    args = parser.parse_args(argv)

    repo_root = args.repo_root.resolve()
    baseline_path = (
        args.baseline
        if args.baseline is not None
        else repo_root / "reports" / "graph-write-swallow-baseline.json"
    )
    roots = [Path(r) for r in args.roots]
    findings = scan(roots, repo_root)
    active = [f for f in findings if not f.waived]

    if args.list:
        if not findings:
            print("[graph-write-swallow] no findings.")
        for f in findings:
            _print_finding(f, "WAIVED" if f.waived else "FLAGGED")
        print(
            f"[graph-write-swallow] {len(findings)} total finding(s), "
            f"{len(active)} active (non-waived)."
        )
        return 0

    if args.update_baseline:
        write_baseline(baseline_path, active)
        print(
            f"[graph-write-swallow] baseline updated: {len(active)} finding(s) "
            f"-> {baseline_path}"
        )
        return 0

    # Default mode: --check (ratchet).
    baseline_ids = load_baseline(baseline_path)
    new_findings = [f for f in active if f.finding_id not in baseline_ids]
    baselined_count = len(active) - len(new_findings)

    if new_findings:
        print(
            f"[graph-write-swallow] FAIL: {len(new_findings)} NEW finding(s) "
            f"not in the baseline ({baselined_count} baselined, not counted):"
        )
        for f in new_findings:
            _print_finding(f, "NEW")
        print(
            "  Fix the swallow (raise/return/record the failure), waive it "
            "with '# gate-ok: <reason>' on the try line or the line above, "
            "or run --update-baseline if this is accepted debt."
        )
        return 1

    print(f"[graph-write-swallow] OK: {baselined_count} baselined finding(s), 0 new.")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())