"""Savepoint-isolation guard.

Hard CI invariant preventing the next recurrence of the bug class behind the
2026-08-06 Zoom-seat incident (PR #2571).

The pattern: a "best-effort" step wrapped in ``except Exception`` so it
"can't break the caller" — but the body writes to the DB. A failed flush
deactivates the *whole* transaction, so the swallow succeeds while the very
next statement (almost always an ``AuditService.log``, which is a bare
``add()`` + ``flush()`` and does not self-isolate) raises
``PendingRollbackError`` / ``InvalidRequestError``. The caller dies anyway,
and the promise in the comment above the ``try`` is a lie.

An audit after that incident found ~30 instances of this shape across
offboarding, access-certification, HR sync and the scheduled jobs. This
guard stops a NEW one from landing.

What it rejects
---------------
A ``try`` whose handler is a broad ``except`` (bare, ``Exception`` or
``BaseException``) where the ``try`` body performs DB work — and which is
NOT isolated by a savepoint. "DB work" is either of:

* **literal**: an ``await`` on ``.flush()`` / ``.commit()`` / ``.execute()``
  / ``.add()`` / ``.log()`` / ``.delete()`` / ``.merge()`` / ``.refresh()``
  / ``.scalar()`` / ``.scalars()`` / ``.get()`` on a session-like receiver,
  or a call whose name matches a known DB-writing helper (``enqueue_*``,
  ``upsert_*``, …);
* **interprocedural** (added 2026-09-02): ``await some_fn(db, ...)`` — an
  awaited call handed the caller's session — where ``some_fn`` itself
  mutates that session. This is the shape that let the github_org
  reconcile bug ship (PR #3394): the loops called a plain domain function,
  so the literal rule saw nothing. Only MUTATIONS count; a callee that
  merely reads cannot poison a transaction. See ``WriteIndex``.

Accepted isolations (any one is enough)
---------------------------------------
* ``async with best_effort(db, label=...)`` / ``savepoint(db, label=...)``
  (``app.core.txn`` — preferred)
* a literal ``db.begin_nested()`` / ``self.db.begin_nested()`` inside the try
* the handler re-raises unconditionally (``raise`` with no new flow after it)
* the handler calls ``is_transaction_aborting_db_error(...)`` — the
  hand-rolled two-tier form, still correct
* the handler rolls back (``await db.rollback()``) before continuing
* the try body opens its OWN session (``async_session_factory()``) — the
  caller's transaction is not at risk

Escape hatch
------------
Add ``# savepoint-exempt: <reason>`` on the ``try:`` line or either of the
two lines above it. The marker forces a human to state why the block is
safe rather than silently regrowing the surface.

Runs without a DB connection — pure AST analysis. Exits 0 on success, 1
with a human-readable diagnostic on failure.
"""

from __future__ import annotations

import argparse
import ast
import json
import sys
from pathlib import Path

_SERVER = Path(__file__).resolve().parents[1]
_BASELINE = _SERVER / "savepoint-baseline.json"

_EXEMPT_MARKER = "savepoint-exempt:"

# Method calls that put the session in a state a failure can poison — but
# ONLY when invoked on a session-like receiver (see ``_SESSION_RECEIVERS``).
# Matching on the bare method name flags ``set.add`` / ``logger.log`` /
# ``subprocess.execute`` and drowns the signal, so both halves are required.
_DB_METHOD_NAMES = frozenset({
    "flush",
    "commit",
    "execute",
    "add",
    "add_all",
    "delete",
    "merge",
    "refresh",
    "scalar",
    "scalars",
    "log",           # AuditService.log — bare add()+flush(), does not isolate
})

# Receiver names that denote "this call runs on the caller's transaction".
# Matched against the *textual* receiver so ``self.db``, ``self._db``,
# ``db``, ``session``, ``self.audit`` and ``AuditService(db)`` all hit.
_SESSION_RECEIVERS = ("db", "session", "audit")

# Helper calls that are known to write through the caller's session. Prefix
# match keeps this resistant to renames; the exemption marker is the valve.
_DB_CALL_PREFIXES = ("enqueue_", "upsert_", "create_or_upsert")

# ── Interprocedural layer (2026-09-02) ───────────────────────────────────
# The rules above only see DB work written LITERALLY in the try body. The
# github_org reconcile loops (PR #3394) called a plain domain function ---
# ``await reconcile_org_invitations(db, org, ...)`` --- which writes, so the
# guard saw nothing and the bug shipped. Both loops promised "one org
# failing does not stop the rest"; that promise only held for pure-Python
# exceptions, and the real failure was a failed flush.
#
# So: an awaited call whose first positional argument (or a ``db=`` /
# ``session=`` keyword) is the caller's session runs ON the caller's
# transaction, and is DB work if the callee itself mutates.
#
# "Mutates", NOT "touches". A SELECT cannot poison a transaction, and the
# naive form of this rule --- flag every ``await f(db, ...)`` --- produced
# 148 new findings, dominated by adapter reads like
# ``await AwsAuditorAdapter.list_acm_certs(session)`` that take the session
# only to load integration credentials. Requiring a real mutation in the
# callee cuts that to 34 (measured 2026-09-02).
_MUTATING_METHODS = frozenset({
    "flush", "commit", "add", "add_all", "delete", "merge",
    "log",           # AuditService.log --- bare add()+flush()
})

# ``.execute()`` counts only when its argument PROVABLY roots at a DML
# construct. ``db.execute(select(...))`` is a read; ``db.execute(stmt)``
# where ``stmt`` is opaque is treated as a read too --- being conservative
# here is what keeps the auditor adapter reads out of the findings.
_DML_ROOTS = frozenset({"insert", "update", "delete"})

# Parameter names that denote "this is the caller's session".  Matched
# EXACTLY, unlike ``_SESSION_RECEIVERS`` above, whose substring test would
# make ``feedback`` a session ("fee-DB-ack").
_SESSION_ARG_NAMES = frozenset({
    "db", "_db", "session", "_session", "sess", "async_session", "db_session",
})

# How far to chase a callee's own callees. Measured on the 2026-09-02 tree:
# depth 0 -> 26 findings, depth 1 -> 34, depth 2 -> 34 (saturated). Depth 1
# is what reaches ``apply_enabled_rules`` -> ``attach(db, ...)``, a live gap
# in github_org that PR #3394 did not fix. Going deeper buys nothing and
# starts classifying half the codebase as "writes".
_WRITE_CHASE_DEPTH = 1

# Isolation signatures — presence of any of these inside the try body (or the
# handler, for the rollback / two-tier forms) clears the finding.
_ISOLATION_IN_BODY = ("begin_nested", "best_effort", "savepoint", "async_session_factory")
_ISOLATION_IN_HANDLER = ("is_transaction_aborting_db_error", "rollback")

# Files that legitimately define or test the primitives themselves.
_ALLOWED_FILES = {
    "app/core/txn.py",
}


def _app_dir() -> Path:
    return Path(__file__).resolve().parent.parent / "app"


def _rel(path: Path, root: Path) -> str:
    return str(path.relative_to(root.parent)).replace("\\", "/")


def _names_in(node: ast.AST) -> set[str]:
    """Every attribute / function / imported name appearing under ``node``.

    Import names matter: a block that opens its own session very often does
    it through an aliased local import
    (``from app.core.database import async_session_factory as _asf``), and
    the call site then reads ``_asf()``. Without collecting the *imported*
    name the isolation signature is invisible and the block is a false
    positive — ``app/main.py`` does exactly this.
    """
    found: set[str] = set()
    for sub in ast.walk(node):
        if isinstance(sub, ast.Attribute):
            found.add(sub.attr)
        elif isinstance(sub, ast.Name):
            found.add(sub.id)
        elif isinstance(sub, ast.ImportFrom | ast.Import):
            for alias in sub.names:
                found.add(alias.name.rsplit(".", 1)[-1])
                if alias.asname:
                    found.add(alias.asname)
    return found


_BROAD_EXC_NAMES = frozenset({"Exception", "BaseException"})


def _is_broad_handler(handler: ast.ExceptHandler) -> bool:
    """True for bare ``except:``, ``except Exception:`` and the TUPLE form.

    The tuple form is not a curiosity — ``app/worker.py`` already carries
    ``except (asyncio.CancelledError, Exception):``. Missing it let a broad
    handler hide behind one extra pair of brackets (review finding,
    2026-08-06).
    """
    if handler.type is None:
        return True
    if isinstance(handler.type, ast.Name):
        return handler.type.id in _BROAD_EXC_NAMES
    if isinstance(handler.type, ast.Tuple):
        return any(
            isinstance(el, ast.Name) and el.id in _BROAD_EXC_NAMES
            for el in handler.type.elts
        )
    return False


def _handler_always_reraises(handler: ast.ExceptHandler) -> bool:
    """True if the handler ends in a bare/unconditional ``raise``.

    A handler that re-raises cannot leave the caller running on a poisoned
    session, so it is never a finding regardless of what the body did.
    """
    if not handler.body:
        return False
    last = handler.body[-1]
    return isinstance(last, ast.Raise)


def _receiver_text(node: ast.AST) -> str:
    """Flattened dotted/called receiver text, lowercased.

    ``self.db`` → "self.db"; ``AuditService(self.db)`` → "auditservice self.db".
    Used to decide whether a method call runs on the caller's session.
    """
    parts: list[str] = []
    for sub in ast.walk(node):
        if isinstance(sub, ast.Attribute):
            parts.append(sub.attr)
        elif isinstance(sub, ast.Name):
            parts.append(sub.id)
    return " ".join(parts).lower()


def _is_session_receiver(node: ast.AST) -> bool:
    text = _receiver_text(node)
    return any(token in text for token in _SESSION_RECEIVERS)


class WriteIndex:
    """Which functions mutate the session they are handed.

    Built once per run over ``app/**``. Name-keyed, not fully qualified:
    two functions sharing a name collapse, and the index answers "writes"
    if ANY of them does. That over-approximates, but the alternative --
    real import resolution -- is a static analyser, and the exemption
    marker is the valve for the handful of collisions.
    """

    def __init__(self, funcs: dict[str, list[ast.AST]]) -> None:
        self._funcs = funcs
        self._memo: dict[str, bool] = {}

    @classmethod
    def from_sources(cls, sources: list[str]) -> WriteIndex:
        funcs: dict[str, list[ast.AST]] = {}
        for src in sources:
            try:
                tree = ast.parse(src)
            except SyntaxError:
                continue
            for node in ast.walk(tree):
                if isinstance(node, ast.FunctionDef | ast.AsyncFunctionDef):
                    funcs.setdefault(node.name, []).append(node)
        return cls(funcs)

    def writes(self, name: str, _depth: int = 0, _seen: frozenset[str] = frozenset()) -> bool:
        if _depth == 0 and name in self._memo:
            return self._memo[name]
        if _depth > _WRITE_CHASE_DEPTH or name in _seen or name not in self._funcs:
            return False
        seen = _seen | {name}
        result = False
        for fn in self._funcs[name]:
            if _mutates_session(fn):
                result = True
                break
            for sub in ast.walk(fn):
                if not isinstance(sub, ast.Call) or not _takes_session(sub):
                    continue
                callee = _callee_name(sub)
                if callee and callee != name and self.writes(callee, _depth + 1, seen):
                    result = True
                    break
            if result:
                break
        if _depth == 0:
            self._memo[name] = result
        return result


_REPO_INDEX: WriteIndex | None = None


def _repo_index() -> WriteIndex:
    global _REPO_INDEX
    if _REPO_INDEX is None:
        _REPO_INDEX = WriteIndex.from_sources([
            p.read_text(encoding="utf-8") for p in sorted(_app_dir().rglob("*.py"))
        ])
    return _REPO_INDEX


def _root_name(node: ast.AST) -> str | None:
    """Leftmost ``Name`` of a call/attribute chain: ``update(X).where(Y)`` -> "update"."""
    while True:
        if isinstance(node, ast.Call):
            node = node.func
        elif isinstance(node, ast.Attribute):
            node = node.value
        else:
            break
    return node.id if isinstance(node, ast.Name) else None


def _execute_is_dml(call: ast.Call, assigns: dict[str, ast.AST]) -> bool:
    """``db.execute(...)`` -- write only if the argument provably roots at DML.

    Follows one local rebinding chain so the extremely common
    ``stmt = update(T).where(...)`` / ``await db.execute(stmt)`` split is
    still seen. Anything unresolvable is treated as a READ: a false
    negative here costs one missed finding, a false positive costs ~50
    adapter reads in the auditor domain alone.
    """
    if not call.args:
        return False
    root = _root_name(call.args[0])
    seen: set[str] = set()
    while root is not None and root not in seen:
        if root in _DML_ROOTS:
            return True
        seen.add(root)
        nxt = assigns.get(root)
        if nxt is None:
            return False
        root = _root_name(nxt)
    return False


def _local_assignments(fn: ast.AST) -> dict[str, ast.AST]:
    out: dict[str, ast.AST] = {}
    for sub in ast.walk(fn):
        if isinstance(sub, ast.Assign) and len(sub.targets) == 1 and isinstance(
            sub.targets[0], ast.Name,
        ):
            out[sub.targets[0].id] = sub.value
        elif isinstance(sub, ast.AnnAssign) and isinstance(
            sub.target, ast.Name,
        ) and sub.value is not None:
            out[sub.target.id] = sub.value
    return out


def _mutates_session(fn: ast.AST) -> bool:
    """True if ``fn``'s own body MUTATES a session it can see."""
    assigns = _local_assignments(fn)
    for sub in ast.walk(fn):
        if not isinstance(sub, ast.Call) or not isinstance(sub.func, ast.Attribute):
            continue
        func = sub.func
        if not _is_session_receiver(func.value):
            continue
        if func.attr in _MUTATING_METHODS:
            return True
        if func.attr == "execute" and _execute_is_dml(sub, assigns):
            return True
    return False


def _is_session_arg(node: ast.AST) -> bool:
    if isinstance(node, ast.Name):
        return node.id in _SESSION_ARG_NAMES
    if isinstance(node, ast.Attribute):
        return node.attr in _SESSION_ARG_NAMES
    return False


def _takes_session(call: ast.Call) -> bool:
    """``f(db, ...)`` / ``f(self.db, ...)`` / ``f(..., session=session)``."""
    if call.args and _is_session_arg(call.args[0]):
        return True
    return any(kw.arg in _SESSION_ARG_NAMES for kw in call.keywords)


# ── Constructor-injected sessions (2026-09-06) ───────────────────────────
# :func:`_takes_session` requires the session to be the callee's FIRST
# POSITIONAL ARGUMENT. That misses the shape this codebase uses most:
#
#     await ProvisioningService(db).retry_task(task_id, ...)
#     svc = NotificationService(self.db); await svc.dispatch(env, ...)
#
# The session goes into the CONSTRUCTOR, so the outer call's first argument
# is a task id and the interprocedural rule saw nothing. Three per-row
# handlers in ``automation_exception/replay.py`` and the permission expiry
# sweep shipped through this hole on 2026-09-06 and were caught by a
# post-merge review rather than by CI — the promise "one row must not stop
# the batch" was untrue on the commonest DB failure, and nothing said so.
#
# A method invoked on an object constructed with the caller's session runs
# on the caller's transaction exactly as ``f(db, ...)`` does, so it is DB
# work under the same condition: the callee mutates.


def _receiver_holds_session(
    node: ast.AST, ctor_bindings: dict[str, ast.Call],
) -> bool:
    """True when ``node`` is a receiver constructed with the caller's session.

    Two forms, both live in this repo:

    * ``Service(db)`` written inline as the receiver;
    * a name bound to one — ``svc = Service(self.db)`` — resolved through
      ``ctor_bindings``, which is collected per module the same
      over-approximating way :class:`WriteIndex` collects functions.

    Deliberately NOT ``self.<attr>``: an attribute assigned in ``__init__``
    would need real class resolution, and every such site in this tree also
    reaches the session directly (``self.db``), which the literal rule
    already covers.
    """
    if isinstance(node, ast.Call):
        return _takes_session(node)
    if isinstance(node, ast.Name):
        bound = ctor_bindings.get(node.id)
        return bound is not None and _takes_session(bound)
    return False


def _ctor_bindings(tree: ast.AST) -> dict[str, ast.Call]:
    """``{local name: the call it was bound to}`` for one module.

    Module-wide rather than per-scope, matching :class:`WriteIndex`'s own
    over-approximation: a name bound to a session-taking constructor
    anywhere in the file is treated as one everywhere. The exemption marker
    is the valve for the rare collision.
    """
    out: dict[str, ast.Call] = {}
    for node in ast.walk(tree):
        if (
            isinstance(node, ast.Assign)
            and len(node.targets) == 1
            and isinstance(node.targets[0], ast.Name)
            and isinstance(node.value, ast.Call)
        ):
            out[node.targets[0].id] = node.value
    return out


def _ctor_name(node: ast.AST, ctor_bindings: dict[str, ast.Call]) -> str | None:
    """The class name a session-holding receiver was constructed from."""
    call = node if isinstance(node, ast.Call) else None
    if call is None and isinstance(node, ast.Name):
        call = ctor_bindings.get(node.id)
    if call is None:
        return None
    return call.func.id if isinstance(call.func, ast.Name) else None


#: ``(constructor, method)`` pairs that mutate but wrap EVERY mutation in
#: their own savepoint, so a caller wrapping them again buys nothing.
#:
#: Keyed on the pair, not on the method name, because ``dispatch`` alone is
#: four different functions in this tree (two ASGI middlewares and the
#: security-awareness campaign sender, which commits between phases and
#: must never be wrapped for a different reason).
#:
#: Without this the widened rule above produces **73 findings that are all
#: one fact**: `NotificationService.dispatch` is the single most-called
#: best-effort step in the codebase, and it has isolated its own writes
#: since 2026-05-26. Demanding a redundant SAVEPOINT at 73 call sites
#: would be a guard training its readers to add ceremony.
#:
#: Listing a pair here is a CLAIM, and
#: ``test_listed_self_isolating_calls_really_isolate`` checks it against
#: the AST rather than trusting the list — add a pair and the test tells
#: you whether you were right.
_SELF_ISOLATING_CALLS: frozenset[tuple[str, str]] = frozenset({
    ("NotificationService", "dispatch"),
    ("NotificationService", "enqueue"),
    ("NotificationService", "push_inbox_and_feishu"),
})


def _callee_name(call: ast.Call) -> str | None:
    func = call.func
    if isinstance(func, ast.Name):
        return func.id
    if isinstance(func, ast.Attribute):
        return func.attr
    return None


def _body_touches_db(
    body: list[ast.stmt],
    index: WriteIndex | None = None,
    ctor_bindings: dict[str, ast.Call] | None = None,
) -> bool:
    bindings = ctor_bindings or {}
    for stmt in body:
        for sub in ast.walk(stmt):
            if not isinstance(sub, ast.Call):
                continue
            func = sub.func
            if isinstance(func, ast.Attribute):
                # A poisoning call must be BOTH a session method AND invoked
                # on a session-like receiver.
                if func.attr in _DB_METHOD_NAMES and _is_session_receiver(func.value):
                    return True
                if func.attr.startswith(_DB_CALL_PREFIXES):
                    return True
            elif isinstance(func, ast.Name) and func.id.startswith(_DB_CALL_PREFIXES):
                return True
        if index is None:
            continue
        # Interprocedural: ``await domain_fn(db, ...)`` where the callee
        # mutates. This is the shape that let PR #3394's bug ship.
        for sub in ast.walk(stmt):
            if not isinstance(sub, ast.Await):
                continue
            call = sub.value
            if not isinstance(call, ast.Call):
                continue
            callee = _callee_name(call)
            if not callee:
                continue
            if _takes_session(call) and index.writes(callee):
                return True
            # Constructor-injected session: ``Service(db).method(...)``.
            if not isinstance(call.func, ast.Attribute):
                continue
            receiver = call.func.value
            if not _receiver_holds_session(receiver, bindings):
                continue
            ctor = _ctor_name(receiver, bindings)
            if ctor is not None and (ctor, callee) in _SELF_ISOLATING_CALLS:
                continue
            if index.writes(callee):
                return True
    return False


def _strip_isolated_blocks(body: list[ast.stmt]) -> list[ast.stmt]:
    """``body`` with any statement that is itself savepoint-isolated removed.

    A statement counts as isolated when it is an ``async with`` whose context
    managers mention ``best_effort`` / ``savepoint`` / ``begin_nested`` / a
    session factory — i.e. the writes inside it are already protected and must
    not clear the writes OUTSIDE it.
    """
    kept: list[ast.stmt] = []
    for stmt in body:
        if isinstance(stmt, ast.AsyncWith | ast.With):
            ctx_names: set[str] = set()
            for item in stmt.items:
                ctx_names |= _names_in(item.context_expr)
            if ctx_names & set(_ISOLATION_IN_BODY):
                continue
            kept.append(
                ast.copy_location(
                    ast.If(
                        test=ast.Constant(True),
                        body=_strip_isolated_blocks(stmt.body),
                        orelse=[],
                    ),
                    stmt,
                ),
            )
            continue
        for field in ("body", "orelse", "finalbody"):
            inner = getattr(stmt, field, None)
            if isinstance(inner, list) and inner:
                setattr(stmt, field, _strip_isolated_blocks(inner))
        kept.append(stmt)
    return kept


def _is_exempt(lines: list[str], lineno: int) -> bool:
    """Marker may sit on the ``try:`` line or the two lines above it."""
    for offset in (0, 1, 2):
        idx = lineno - 1 - offset
        if not (0 <= idx < len(lines)):
            continue
        line = lines[idx]
        if _EXEMPT_MARKER not in line:
            continue
        # The marker must be a COMMENT (not a string literal) and must carry
        # an actual reason — "forces a human to justify it" is only true if
        # the reason is enforced (review finding, 2026-08-06).
        before, _, after = line.partition(_EXEMPT_MARKER)
        if "#" not in before:
            continue
        if not after.strip():
            continue
        return True
    return False


def scan_source(
    source: str, lines: list[str], index: WriteIndex | None = None,
) -> list[tuple[int, str]]:
    """Return ``(lineno, snippet)`` for each un-isolated broad-except DB block.

    ``index`` supplies the interprocedural layer (which callees mutate).
    Defaults to the repo-wide index; tests inject
    ``WriteIndex.from_sources([snippet])`` to stay hermetic. Pass an EMPTY
    index to get literal-only detection.
    """
    if index is None:
        index = _repo_index()
    try:
        tree = ast.parse(source)
    except SyntaxError:
        return []
    # ``svc = SomeService(db)`` bindings for this module, so a method call
    # on ``svc`` can be recognised as running on the caller's session.
    bindings = _ctor_bindings(tree)

    # Map each Try to the names visible in its enclosing function. A block
    # that opens its OWN session very often imports the factory ABOVE the
    # ``try`` (``from app.core.database import async_session_factory as
    # _factory`` then ``try: ... async with _factory() as session:``), so
    # looking only inside the try body misses the isolation and reports a
    # false positive. Found 2026-08-06 across the CMDB routers.
    enclosing_names: dict[int, set[str]] = {}
    for fn in ast.walk(tree):
        if not isinstance(fn, ast.FunctionDef | ast.AsyncFunctionDef):
            continue
        fn_names = _names_in(fn)
        for sub in ast.walk(fn):
            if isinstance(sub, ast.Try):
                enclosing_names.setdefault(id(sub), set()).update(fn_names)

    findings: list[tuple[int, str]] = []
    for node in ast.walk(tree):
        if not isinstance(node, ast.Try):
            continue
        broad = [h for h in node.handlers if _is_broad_handler(h)]
        if not broad:
            continue
        if all(_handler_always_reraises(h) for h in broad):
            continue
        if not _body_touches_db(node.body, index, bindings):
            continue
        if _is_exempt(lines, node.lineno):
            continue

        # Isolation must cover the POISONING call, not merely appear
        # somewhere in the body. The old set-intersection cleared the whole
        # try, so the realistic recurrence shape slipped through entirely:
        #
        #     try:
        #         for r in rows:
        #             async with best_effort(db, ...):   # inner, isolated
        #                 await db.flush()
        #         await db.commit()                      # UNISOLATED
        #     except Exception:
        #         pass
        #
        # Re-check the body with every already-isolated sub-block removed: if
        # a poisoning call still remains at the top level, it is unprotected
        # (review finding, 2026-08-06).
        if not _body_touches_db(
            _strip_isolated_blocks(node.body), index, bindings,
        ):
            continue
        # Own-session form: the try body opens a session (`async with X() as
        # y:`) and the enclosing function knows the session factory — the
        # caller's transaction is not at risk even though the import sits
        # outside the try.
        opens_session = any(
            isinstance(sub, ast.AsyncWith)
            and any(isinstance(item.context_expr, ast.Call) for item in sub.items)
            for stmt in node.body
            for sub in ast.walk(stmt)
        )
        if opens_session and "async_session_factory" in enclosing_names.get(
            id(node), set(),
        ):
            continue

        handler_names: set[str] = set()
        for h in broad:
            for stmt in h.body:
                handler_names |= _names_in(stmt)
        if handler_names & set(_ISOLATION_IN_HANDLER):
            continue

        snippet = lines[node.lineno - 1].strip() if node.lineno - 1 < len(lines) else "try:"
        findings.append((node.lineno, snippet))
    return findings


_FIX_HINT = (
    "Fix by wrapping the fallible step:\n"
    "    from app.core.txn import best_effort\n"
    "    async with best_effort(db, label='...') as step:\n"
    "        ...\n"
    "    if not step.ok:\n"
    "        ...\n\n"
    "If the block is genuinely safe (opens its own session, performs no "
    "writes, or always re-raises), add '# savepoint-exempt: <reason>' on or "
    "just above the `try:`."
)


def collect() -> dict[str, list[str]]:
    """Map ``relative/path.py -> ["lineno: snippet", ...]`` for all findings."""
    app = _app_dir()
    index = _repo_index()
    out: dict[str, list[str]] = {}
    for path in sorted(app.rglob("*.py")):
        rel = _rel(path, app)
        if rel in _ALLOWED_FILES:
            continue
        source = path.read_text(encoding="utf-8")
        lines = source.splitlines()
        hits = [
            f"{lineno}: {snippet}"
            for lineno, snippet in scan_source(source, lines, index)
        ]
        if hits:
            out[rel] = hits
    return out


def _load_baseline() -> dict[str, int]:
    if not _BASELINE.exists():
        return {}
    return dict(json.loads(_BASELINE.read_text(encoding="utf-8")))


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--update", action="store_true",
        help="rewrite the baseline from the current tree",
    )
    parser.add_argument(
        "--list", action="store_true",
        help="print every remaining finding with file:line (backlog view)",
    )
    args = parser.parse_args()

    findings = collect()
    counts = {rel: len(hits) for rel, hits in findings.items()}

    if args.update:
        payload = dict(sorted(counts.items()))
        _BASELINE.write_text(
            json.dumps(payload, indent=2, ensure_ascii=False) + "\n",
            encoding="utf-8",
        )
        print(
            f"[savepoint-guard] baseline written: {len(payload)} file(s), "
            f"{sum(payload.values())} finding(s)"
        )
        return 0

    if args.list:
        for rel in sorted(findings):
            for hit in findings[rel]:
                print(f"{rel}:{hit}")
        print(f"\ntotal: {sum(counts.values())} finding(s) in {len(counts)} file(s)")
        return 0

    baseline = _load_baseline()
    regressions: list[str] = []
    for rel, count in sorted(counts.items()):
        allowed = baseline.get(rel, 0)
        if count > allowed:
            detail = "\n".join(f"      {h}" for h in findings[rel])
            regressions.append(
                f"  {rel}: {count} finding(s), baseline allows {allowed}\n{detail}"
            )

    if regressions:
        print(
            "[savepoint-guard] FAIL — new broad `except` around DB work "
            "without savepoint isolation.\n\n"
            "A failed flush deactivates the WHOLE transaction, so swallowing "
            "the error does NOT keep the session usable — the next statement "
            "(usually an audit write) raises and takes the caller down anyway. "
            "This is the 2026-08-06 Zoom-seat incident's bug class "
            "(PR #2571).\n\n"
            + _FIX_HINT
            + "\n\n"
            + "\n".join(regressions)
        )
        return 1

    total = sum(counts.values())
    baseline_total = sum(baseline.values())
    if total < baseline_total:
        print(
            f"[savepoint-guard] IMPROVED — {baseline_total} → {total} finding(s). "
            "Refresh the ratchet so it cannot rot back:\n"
            "    uv run python scripts/check_savepoint_isolation.py --update"
        )
        return 0

    if total == 0:
        print("[savepoint-guard] OK — every broad-except DB block is isolated.")
    else:
        print(
            f"[savepoint-guard] OK — no new findings "
            f"({total} baselined, being drained; "
            "run with --list to see the backlog)."
        )
    return 0


if __name__ == "__main__":
    sys.exit(main())