fix(migrations): address greptile P1/P2 review on linter

- prefer-robust-stmts now also inspects raw SQL inside op.execute(...):
  flags op.execute("CREATE INDEX ...") without CONCURRENTLY and
  op.execute("ALTER TABLE ... ADD CONSTRAINT ... FOREIGN KEY ...") without
  NOT VALID. Closes the documented escape hatch where copying SQL from a
  DBA runbook into op.execute bypassed every helper-level check (P1).
- in-band-backfill no longer false-positives on
  INSERT ... ON CONFLICT DO UPDATE SET upserts (legitimate schema-init
  shape). Tightened _DATA_BACKFILL with a (?!SET\b) negative lookahead (P2).
- no-timeout-overrides skips matches that fall after a # on the same line
  so a Python comment quoting an old SET command is no longer flagged (P2).
- Module docstring's Rules section now lists in-band-backfill alongside
  the other five (developer searching the docstring for a CI failure
  rule will find it).

Adds five unit tests: raw CREATE INDEX in op.execute, raw CREATE INDEX
CONCURRENTLY passes, raw ADD CONSTRAINT FK in op.execute, raw FK with
NOT VALID passes, upsert clause not flagged, and SET in Python comment
not flagged.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: declan-scale

agentex/scripts/ci_tools/migration_lint.py

@@ -22,7 +22,10 @@
 - ``prefer-robust-stmts``: ``op.create_index`` without
   ``postgresql_concurrently=True`` and ``op.create_foreign_key`` without
   ``postgresql_not_valid=True`` on populated tables block writers for the
-  duration of the operation.
+  duration of the operation. Also catches the raw-SQL escape hatches
+  ``op.execute("CREATE INDEX ...")`` (without ``CONCURRENTLY``) and
+  ``op.execute("ALTER TABLE ... ADD CONSTRAINT ... FOREIGN KEY ...")``
+  (without ``NOT VALID``).
 - ``disallowed-unique-constraint``: ``op.create_unique_constraint`` builds the
   supporting index while blocking writes; create the index concurrently first
   and attach the constraint with ``USING INDEX``.
@@ -37,6 +40,10 @@
   ``SET statement_timeout``, ``SET idle_in_transaction_session_timeout``, and
   ``RESET`` of those, so a migration cannot quietly disable the runtime
   guardrails.
+- ``in-band-backfill``: ``op.execute("UPDATE ...")`` or ``DELETE FROM`` inside
+  a migration holds row locks for the entire transaction and prevents
+  autovacuum from cleaning up. Move data backfills to an out-of-band
+  operator runbook and keep migrations schema-only.
 
 Escape hatch
 ------------
@@ -127,9 +134,20 @@ def format(self) -> str:
     re.IGNORECASE,
 )
 _DATA_BACKFILL = re.compile(
-    r"\b(?:UPDATE\s+\w|DELETE\s+FROM\b)",
+    # ``UPDATE\s+(?!SET\b)\w`` skips the ``UPDATE SET`` clause that appears
+    # inside ``INSERT ... ON CONFLICT DO UPDATE SET`` upserts (a legitimate
+    # schema-init pattern that should not flag in-band-backfill).
+    r"\b(?:UPDATE\s+(?!SET\b)\w|DELETE\s+FROM\b)",
     re.IGNORECASE,
 )
+_RAW_BLOCKING_INDEX = re.compile(
+    r"\bCREATE\s+(?:UNIQUE\s+)?INDEX\b(?!\s+CONCURRENTLY)",
+    re.IGNORECASE,
+)
+_RAW_VALIDATING_FK = re.compile(
+    r"\bADD\s+CONSTRAINT\b(?:[^;]*?\bFOREIGN\s+KEY\b)(?![^;]*?\bNOT\s+VALID\b)",
+    re.IGNORECASE | re.DOTALL,
+)
 _AUTOCOMMIT_OPENER = re.compile(r"\bwith\b[^#\n]*\bautocommit_block\s*\(")
 
 
@@ -164,6 +182,19 @@ def _has_noqa(source: str, line: int) -> bool:
     return False
 
 
+def _is_in_python_comment(source: str, pos: int) -> bool:
+    """Return True if ``pos`` falls after a ``#`` on the same source line.
+
+    Naive — does not account for ``#`` appearing inside a string literal — but
+    Alembic migrations rarely embed ``#`` in SQL, and any false negative here
+    just leaves the previous behavior unchanged. Used to keep regex-level
+    rules from flagging text inside Python comments (e.g. a comment referencing
+    a forbidden ``SET lock_timeout`` for documentation purposes).
+    """
+    line_start = source.rfind("\n", 0, pos) + 1
+    return "#" in source[line_start:pos]
+
+
 def _tables_created(source: str) -> set[str]:
     """Return the set of table names created via ``op.create_table`` in this
     migration.
@@ -242,6 +273,39 @@ def _check_prefer_robust_stmts(path: Path, source: str) -> Iterable[Finding]:
             ),
         )
 
+    # Catch the raw-SQL escape hatches: developers copying ``CREATE INDEX`` /
+    # ``ADD CONSTRAINT FOREIGN KEY`` from a DBA runbook into ``op.execute(...)``
+    # would otherwise bypass every helper-level check above.
+    for match in _OP_EXECUTE.finditer(source):
+        line = _line_of(source, match.start())
+        if _has_noqa(source, line):
+            continue
+        call = _slice_call(source, match.start())
+        if _RAW_BLOCKING_INDEX.search(call):
+            yield Finding(
+                path=path,
+                line=line,
+                rule="prefer-robust-stmts",
+                message=(
+                    'op.execute("CREATE INDEX ...") without CONCURRENTLY '
+                    "takes ShareLock for the whole build and blocks writes. "
+                    "Use CREATE INDEX CONCURRENTLY inside an "
+                    "op.get_context().autocommit_block()."
+                ),
+            )
+        if _RAW_VALIDATING_FK.search(call):
+            yield Finding(
+                path=path,
+                line=line,
+                rule="prefer-robust-stmts",
+                message=(
+                    'op.execute("ALTER TABLE ... ADD CONSTRAINT ... FOREIGN KEY ...") '
+                    "without NOT VALID takes ShareRowExclusiveLock and "
+                    "full-scans the child table to validate. Add NOT VALID, "
+                    "then VALIDATE CONSTRAINT in a follow-up migration."
+                ),
+            )
+
 
 def _check_disallowed_unique_constraint(path: Path, source: str) -> Iterable[Finding]:
     fresh_tables = _tables_created(source)
@@ -353,6 +417,8 @@ def _check_no_timeout_overrides(path: Path, source: str) -> Iterable[Finding]:
         line = _line_of(source, match.start())
         if _has_noqa(source, line):
             continue
+        if _is_in_python_comment(source, match.start()):
+            continue
         yield Finding(
             path=path,
             line=line,
@@ -369,6 +435,8 @@ def _check_no_timeout_overrides(path: Path, source: str) -> Iterable[Finding]:
         line = _line_of(source, match.start())
         if _has_noqa(source, line):
             continue
+        if _is_in_python_comment(source, match.start()):
+            continue
         yield Finding(
             path=path,
             line=line,

agentex/tests/unit/scripts/test_migration_lint.py

@@ -272,6 +272,101 @@ def upgrade():
     assert any(f.rule == "in-band-backfill" for f in findings)
 
 
+def test_in_band_backfill_upsert_not_flagged(tmp_path: Path) -> None:
+    """ON CONFLICT DO UPDATE SET upserts are legitimate schema-init shape."""
+    path = _write(
+        tmp_path,
+        """
+        from alembic import op
+        def upgrade():
+            op.execute(
+                "INSERT INTO foo (id, name) VALUES (1, 'x') "
+                "ON CONFLICT (id) DO UPDATE SET name = EXCLUDED.name"
+            )
+        """,
+    )
+    findings = migration_lint.lint_file(path)
+    assert all(f.rule != "in-band-backfill" for f in findings)
+
+
+def test_raw_sql_create_index_in_op_execute_flagged(tmp_path: Path) -> None:
+    path = _write(
+        tmp_path,
+        """
+        from alembic import op
+        def upgrade():
+            op.execute("CREATE INDEX idx_foo_bar ON foo (bar)")
+        """,
+    )
+    findings = migration_lint.lint_file(path)
+    assert any(f.rule == "prefer-robust-stmts" for f in findings)
+
+
+def test_raw_sql_create_index_concurrently_in_op_execute_passes(tmp_path: Path) -> None:
+    path = _write(
+        tmp_path,
+        """
+        from alembic import op
+        def upgrade():
+            with op.get_context().autocommit_block():
+                op.execute("CREATE INDEX CONCURRENTLY idx_foo_bar ON foo (bar)")
+        """,
+    )
+    findings = [
+        f for f in migration_lint.lint_file(path) if f.rule == "prefer-robust-stmts"
+    ]
+    assert findings == []
+
+
+def test_raw_sql_add_fk_without_not_valid_flagged(tmp_path: Path) -> None:
+    path = _write(
+        tmp_path,
+        """
+        from alembic import op
+        def upgrade():
+            op.execute(
+                "ALTER TABLE foo ADD CONSTRAINT fk_foo_bar "
+                "FOREIGN KEY (bar_id) REFERENCES bar (id)"
+            )
+        """,
+    )
+    findings = migration_lint.lint_file(path)
+    assert any(f.rule == "prefer-robust-stmts" for f in findings)
+
+
+def test_raw_sql_add_fk_with_not_valid_passes(tmp_path: Path) -> None:
+    path = _write(
+        tmp_path,
+        """
+        from alembic import op
+        def upgrade():
+            op.execute(
+                "ALTER TABLE foo ADD CONSTRAINT fk_foo_bar "
+                "FOREIGN KEY (bar_id) REFERENCES bar (id) NOT VALID"
+            )
+        """,
+    )
+    findings = [
+        f for f in migration_lint.lint_file(path) if f.rule == "prefer-robust-stmts"
+    ]
+    assert findings == []
+
+
+def test_set_timeout_in_python_comment_not_flagged(tmp_path: Path) -> None:
+    """A Python comment mentioning a forbidden SET shouldn't flag."""
+    path = _write(
+        tmp_path,
+        """
+        from alembic import op
+        def upgrade():
+            # Previously we used: SET lock_timeout = '5s'
+            op.execute("SELECT 1")
+        """,
+    )
+    findings = migration_lint.lint_file(path)
+    assert all(f.rule != "no-timeout-overrides" for f in findings)
+
+
 def test_in_band_backfill_select_not_flagged(tmp_path: Path) -> None:
     path = _write(
         tmp_path,