""" 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 _rehearsal_script_steps(offset: int) -> List[RunbookStep]: """PR3: 3 allowlisted script steps for rehearsal v1 (after http_checks).""" return [ RunbookStep( step_index=offset, title="Preflight check (STRICT=1)", section="0–5 min — Preflight", action_type="script", action_json={ "script": "ops/preflight_sofiia_console.sh", "env": {"STRICT": "1"}, "timeout_s": 120, }, ), RunbookStep( step_index=offset + 1, title="Redis idempotency smoke test", section="10–15 min — Smoke", action_type="script", action_json={ "script": "ops/redis_idempotency_smoke.sh", "env": {}, "timeout_s": 60, }, ), RunbookStep( step_index=offset + 2, title="Generate release evidence", section="25–30 min — Evidence", action_type="script", action_json={ "script": "ops/generate_release_evidence.sh", "env": {}, "timeout_s": 60, }, ), ] 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 + 3 script steps; rest are manual. """ path_lower = runbook_path.lower() steps: List[RunbookStep] = [] offset = 0 if "rehearsal" in path_lower and "30min" in path_lower: http_steps = _rehearsal_http_check_steps(sofiia_url) steps.extend(http_steps) offset = len(http_steps) script_steps = _rehearsal_script_steps(offset) steps.extend(script_steps) offset += len(script_steps) 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