Turning Incidents into Guardrails

# pytest regression from post-mortem INC-8842
def test_migration_uses_concurrently_for_large_tables():
    sql = Path("alembic/versions/20260612_add_invoice_idx.py").read_text()
    assert "CONCURRENTLY" in sql or "op.create_index" not in sql

# Alert: Celery queue depth leading indicator
- alert: BillingQueueDepthHigh
  expr: celery_queue_length{queue="billing"} > 10000
  for: 5m

"""guardrails_example.py - incident-driven tests and runtime checks."""
from __future__ import annotations
 
from dataclasses import dataclass
from pathlib import Path
 
 
@dataclass(frozen=True)
class Guardrail:
    incident_id: str
    kind: str  # test | alert | lint | runbook
    description: str
    verify: str
 
 
GUARDRAILS: list[Guardrail] = [
    Guardrail(
        incident_id="INC-8842",
        kind="test",
        description="Large-table indexes use CONCURRENTLY",
        verify="pytest tests/test_migrations.py -k concurrently",
    ),
    Guardrail(
        incident_id="INC-9011",
        kind="alert",
        description="Checkout p95 burn rate",
        verify="Grafana: CheckoutSLOBurn",
    ),
    Guardrail(
        incident_id="INC-9011",
        kind="lint",
        description="No sync sleep in async routes",
        verify="scripts/check_async_blocking.py in CI",
    ),
]
 
 
def audit_open_guardrails(registry: list[Guardrail]) -> None:
    for g in registry:
        print(f"[{g.incident_id}] {g.kind}: {g.description} -> {g.verify}")
 
 
if __name__ == "__main__":
    audit_open_guardrails(GUARDRAILS)

# tests/test_incident_9011_loop_block.py
import ast
from pathlib import Path
 
def test_no_time_sleep_in_async_functions():
    src = Path("src/api/routes/checkout.py").read_text()
    tree = ast.parse(src)
    for node in ast.walk(tree):
        if isinstance(node, ast.AsyncFunctionDef):
            for child in ast.walk(node):
                if isinstance(child, ast.Call):
                    func = child.func
                    if isinstance(func, ast.Attribute) and func.attr == "sleep":
                        raise AssertionError(f"time.sleep in async {node.name}")

# pyproject.toml - pip-audit in CI
[tool.uv]
dev-dependencies = ["pip-audit>=2.7"]
 
# scripts/ci.sh
# uv run pip-audit -r requirements.lock