PD Framework — Fase 4 Plano Técnico (Auto-fix observabilidade)¶

#!/usr/bin/env python3
"""runbook-executor — worker cron Squad Infra.

Roda a cada 5min na VPS Master via harness.sh. Lê issues Linear com label
'auto-fix' criadas pelo webhook v2, executa runbook correspondente se houver
match + allowlist OK, fecha issue ou escala.

Determinístico. NUNCA usa LLM, NUNCA shell arbitrário. Toda ação vem de
runbook .sh pré-aprovado.

Fluxo:
  1. lista issues Linear label='auto-fix' state='Triage|Backlog|Todo'
  2. pra cada issue:
     - extrai alertname (de labels) e target (de description)
     - resolve runbook: squads/infra/runbooks/<categoria>/<rule>.sh
     - se runbook não existe → label='escalado' + chama endpoint approve
     - se existe: valida ALLOWLIST → executa → captura exit code
       - 0 → comment + fecha issue + state='Done' + atualiza STATE.md
       - 2 → comment "auto-resolveu" + fecha
       - 4/≥10 → comment com stderr + label='escalado'
"""

import os, sys, subprocess, re
from pathlib import Path

# import _shared/
sys.path.insert(0, str(Path(__file__).resolve().parents[3] / "_shared"))
from linear_client import graphql
from stevo_client import StevoClient
sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "_core"))
from state_updater import StateUpdater  # noqa

RUNBOOKS_DIR = Path(__file__).resolve().parent.parent / "runbooks"
ALLOWLIST_PATH = RUNBOOKS_DIR / "ALLOWLIST.md"

ALERTNAME_TO_RUNBOOK = {
    # mapeamento explícito (não inferir). Match contra Grafana alertname.
    "vps_load_average_alto":  "load/average-investigar.sh",
    "traefik_ram_alta":       "traefik/ram-restart.sh",
    "tcp_close_wait_excessivo": "tcp/close-wait-investigar.sh",
    "disco_raiz_baixo":       "disco/raiz-log-rotate.sh",
    "ram_disponivel_baixa":   "ram/disponivel-restart-app.sh",
    "ssh_brute_force":        "security/ssh-brute-force-block-ip.sh",
    "vercel_deploy_falhado":  "vercel/deploy-falhado-retry.sh",
}

def list_autofix_issues() -> list[dict]:
    # GraphQL query
    ...

def resolve_runbook(alertname: str) -> Path | None:
    rel = ALERTNAME_TO_RUNBOOK.get(alertname.lower())
    if not rel:
        return None
    p = RUNBOOKS_DIR / rel
    return p if p.exists() else None

def validate_allowlist(target: str) -> bool:
    text = ALLOWLIST_PATH.read_text(encoding="utf-8")
    # busca target na seção "Containers PROIBIDOS" → False
    # busca target na seção "Containers permitidos" → True
    # default → False (deny por padrão)
    ...

def execute_runbook(runbook: Path, alert_id: str, target: str) -> tuple[int, str, str]:
    r = subprocess.run(
        ["bash", str(runbook), alert_id, target],
        capture_output=True, text=True, timeout=120
    )
    return r.returncode, r.stdout, r.stderr

def main():
    issues = list_autofix_issues()
    state = StateUpdater("infra")
    for issue in issues:
        alertname = extract_alertname(issue)
        target = extract_target(issue)
        runbook = resolve_runbook(alertname)
        if not runbook:
            escalate(issue, reason="sem runbook match")
            continue
        if not validate_allowlist(target):
            escalate(issue, reason=f"allowlist viola: {target}")
            continue
        code, out, err = execute_runbook(runbook, issue["id"], target)
        if code == 0:
            close_issue(issue, comment=f"runbook OK\n```\n{out}\n```")
            state.add_l2(f"Resolvido auto: {alertname} ({target})")
        elif code == 2:
            close_issue(issue, comment="alerta auto-resolveu (runbook detectou)")
        else:
            escalate(issue, reason=f"runbook falhou (exit {code})\nstderr:\n{err[:500]}")
    state.save()

if __name__ == "__main__":
    main()

@app.post("/approve/<alert_id>")
def approve(alert_id):
    # 1. valida assinatura Stevo (X-Stevo-Signature header)
    # 2. parseia callback: {from: "5511914912127", text: "1"}
    # 3. lê cache .pending_approvals[alert_id] (criado quando alerta sem runbook chegou)
    # 4. mapeia resposta:
    #    "1"/"2" → executa runbook X/Y da lista oferecida
    #    "3" → manual: deixa issue aberta, manda "OK, vai investigar manualmente"
    #    "4" → ignore: fecha issue com label `ignorado` + razão
    # 5. atualiza STATE.md infra

# pseudo
runbook_candidatos = sugerir_runbooks(alertname)  # heurística top-3
if not runbook_candidatos:
    cria issue normal + manda WhatsApp "alerta sem runbook conhecido, abrir Linear"
else:
    cria issue (sem auto-fix)
    cache.pending_approvals[alert_id] = runbook_candidatos
    stevo.send_text(
        FELIPE_PHONE,
        f"⚠️ {alertname}\n"
        f"Sugestões:\n"
        f"[1] {runbook_candidatos[0]}\n"
        f"[2] {runbook_candidatos[1]}\n"
        f"[3] investigar manual\n"
        f"[4] ignorar"
    )

# em handler do alert
def update_infra_state(alert: dict):
    state_path = REPO_ROOT / "squads/infra/memory/STATE.md"
    # operação atômica: file lock + rewrite
    with filelock.FileLock(f"{state_path}.lock"):
        s = StateUpdater("infra")
        if alert["status"] == "firing":
            s.add_l2(f"[ALERTA] {alert['labels']['alertname']} firing desde {alert['startsAt']}")
        else:  # resolved
            # remove L2 entry correspondente
            s._l2 = [x for x in s._l2 if alert['labels']['alertname'] not in x]
        s.save()
        # git add + commit + push (segue convenção do harness)
        subprocess.run(["git", "-C", str(REPO_ROOT), "add", str(state_path)])
        subprocess.run(["git", "-C", str(REPO_ROOT), "commit", "-m",
                       f"chore(infra): state update — alerta {alert['labels']['alertname']}"])
        subprocess.run(["git", "-C", str(REPO_ROOT), "push"])