microdao-daarion/services/sofiia-console/app/runbook_parser.py

"""
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