ad8bddf595
adds runbook_runs/runbook_steps state machine parses markdown runbooks into guided steps supports allowlisted http_check (health/metrics/audit) integrates runbook execution with audit trail exposes authenticated runbook runs API Made-with: Cursor
128 lines
4.0 KiB
Python
128 lines
4.0 KiB
Python
"""
|
||
Runbook parser — markdown → list of steps (manual / http_check).
|
||
PR2: sections by H2 → manual steps; for rehearsal-v1 prepend 3 http_check steps.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import re
|
||
from dataclasses import dataclass, field
|
||
from typing import Any, Dict, List
|
||
|
||
|
||
@dataclass
|
||
class RunbookStep:
|
||
step_index: int
|
||
title: str
|
||
section: str | None
|
||
action_type: str # manual | http_check
|
||
action_json: Dict[str, Any]
|
||
status: str = "pending"
|
||
|
||
|
||
def _rehearsal_http_check_steps(base_url: str) -> List[RunbookStep]:
|
||
"""Three built-in http_check steps for rehearsal v1."""
|
||
base = (base_url or "http://127.0.0.1:8002").rstrip("/")
|
||
return [
|
||
RunbookStep(
|
||
step_index=0,
|
||
title="Health check",
|
||
section="0–5 min — Preflight",
|
||
action_type="http_check",
|
||
action_json={
|
||
"method": "GET",
|
||
"url_path": "/api/health",
|
||
"expect": {"status": [200]},
|
||
"headers": {},
|
||
},
|
||
),
|
||
RunbookStep(
|
||
step_index=1,
|
||
title="Metrics check",
|
||
section="0–5 min — Preflight",
|
||
action_type="http_check",
|
||
action_json={
|
||
"method": "GET",
|
||
"url_path": "/metrics",
|
||
"expect": {"status": [200]},
|
||
"headers": {},
|
||
},
|
||
),
|
||
RunbookStep(
|
||
step_index=2,
|
||
title="Audit auth (expect 401 without key)",
|
||
section="10–15 min — Smoke",
|
||
action_type="http_check",
|
||
action_json={
|
||
"method": "GET",
|
||
"url_path": "/api/audit",
|
||
"expect": {"status": [401]},
|
||
"headers": {},
|
||
},
|
||
),
|
||
]
|
||
|
||
|
||
def _parse_sections(markdown: str) -> List[tuple]:
|
||
"""Split by ## or ###, return [(section_title, content), ...]."""
|
||
sections: List[tuple] = []
|
||
current_title = ""
|
||
current_lines: List[str] = []
|
||
for line in markdown.splitlines():
|
||
if re.match(r"^##\s+", line):
|
||
if current_lines:
|
||
sections.append((current_title, "\n".join(current_lines).strip()))
|
||
current_title = line.lstrip("#").strip()[:300]
|
||
current_lines = []
|
||
elif re.match(r"^###\s+", line):
|
||
if current_lines:
|
||
sections.append((current_title, "\n".join(current_lines).strip()))
|
||
current_title = line.lstrip("#").strip()[:300]
|
||
current_lines = []
|
||
else:
|
||
current_lines.append(line)
|
||
if current_lines:
|
||
sections.append((current_title, "\n".join(current_lines).strip()))
|
||
return sections
|
||
|
||
|
||
def parse_runbook(
|
||
runbook_path: str,
|
||
markdown: str,
|
||
sofiia_url: str = "http://127.0.0.1:8002",
|
||
) -> List[RunbookStep]:
|
||
"""
|
||
Parse markdown into steps. For rehearsal-v1 prepend 3 http_check steps;
|
||
rest are manual (one per H2 section with instructions_md).
|
||
"""
|
||
path_lower = runbook_path.lower()
|
||
steps: List[RunbookStep] = []
|
||
offset = 0
|
||
|
||
if "rehearsal" in path_lower and "30min" in path_lower:
|
||
builtin = _rehearsal_http_check_steps(sofiia_url)
|
||
steps.extend(builtin)
|
||
offset = len(builtin)
|
||
|
||
sections = _parse_sections(markdown)
|
||
for i, (title, content) in enumerate(sections):
|
||
if not title and not content.strip():
|
||
continue
|
||
title_clean = title or f"Step {i + 1}"
|
||
# First section might be title-only; still add as manual
|
||
instructions = content.strip() or title_clean
|
||
if len(instructions) > 8000:
|
||
instructions = instructions[:8000] + "\n..."
|
||
steps.append(
|
||
RunbookStep(
|
||
step_index=offset + i,
|
||
title=title_clean[:200],
|
||
section=title_clean[:200] if title else None,
|
||
action_type="manual",
|
||
action_json={"instructions_md": instructions, "accept_input": True},
|
||
status="pending",
|
||
)
|
||
)
|
||
|
||
return steps
|