parser.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/registry/roadmap/parser.py
 
# OCT-LINT: disable=tests_exist reason="Covered by oc_chat/tests/registry/roadmap/test_parser.py"
"""
Markdown parser for ROADMAP registry ingest.
 
Purpose
-------
Convert the four ROADMAP markdown artifact families into deterministic
records for the registry v0.3 roadmap tables.
 
Responsibilities
----------------
- Classify roadmap artifact file kinds and versions.
- Extract phases, items, dependencies, tracks, and render-section hints.
- Produce stable semantic hashes for ingest idempotence checks.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - parser lifecycle
    L3 - extracted roadmap records
    L4 - per-row markdown parsing details
 
Contracts
---------
- Parsing is read-only and never mutates registry or filesystem state.
- Output records are deterministic for identical markdown bytes.
 
Dependencies
------------
- oc_chat.registry.canonical (stable JSON hashing)
- oc_diagnostics (structured debug logging via _dbg)
"""
from __future__ import annotations
 
from dataclasses import dataclass
import hashlib
import re
from pathlib import Path
 
from oc_diagnostics import _dbg
 
from oc_chat.registry.canonical import canonical_json
 
_dbg("OC_CHAT.MAIN", "module", "loaded", 4)
 
 
class RoadmapParseError(ValueError):
    """Raised when a roadmap file is not parseable enough for ingest."""
 
 
@dataclass(frozen=True)
class ParsedPhase:
    phase_code: str
    label: str
    sort_order: int
    parent_phase_code: str | None = None
    priority_label: str | None = None
    parallel_track: bool = False
    target_version: str | None = None
 
 
@dataclass(frozen=True)
class ParsedTrack:
    track_code: str
    label: str
    description: str | None
    sort_order: int
 
 
@dataclass(frozen=True)
class ParsedItem:
    item_id: str
    title: str
    track_kind: str
    sort_order: int
    status: str
    priority: str | None = None
    phase_code: str | None = None
    parent_item_id: str | None = None
    target_version: str | None = None
    prose: str | None = None
 
    def natural_key(self) -> str:
        return self.item_id
 
    def content_hash(self) -> str:
        return canonical_record_hash(
            "roadmap_items_v3",
            self.item_id,
            {
                "title": self.title,
                "track_kind": self.track_kind,
                "status": self.status,
                "priority": self.priority,
                "phase_code": self.phase_code,
                "parent_item_id": self.parent_item_id,
                "target_version": self.target_version,
                "prose": self.prose,
                "sort_order": self.sort_order,
            },
        )
 
 
@dataclass(frozen=True)
class ParsedDependency:
    from_item_id: str
    to_item_id: str
    kind: str = "gates"
 
 
@dataclass(frozen=True)
class ParsedRenderSection:
    file_kind: str
    section_path: str
    sort_order: int
    query_kind: str
    query_params_json: str | None = None
 
 
@dataclass(frozen=True)
class ParsedRoadmap:
    path: Path
    file_kind: str
    version_id: str
    title: str
    content_sha256: str
    era_id: str
    created_at: str
    last_updated_at: str
    phases: tuple[ParsedPhase, ...]
    tracks: tuple[ParsedTrack, ...]
    items: tuple[ParsedItem, ...]
    dependencies: tuple[ParsedDependency, ...]
    render_sections: tuple[ParsedRenderSection, ...]
 
 
VERSION_RE = re.compile(r"\b(?:v)?(?P<version>\d+\.\d+)\b")
H1_RE = re.compile(r"^#\s+(?P<title>.+?)\s*$", re.MULTILINE)
# OCT-LINT: disable=no_hardcoded_secrets reason="Roadmap regex patterns are parser constants, not secrets"
HEADER_RE = re.compile(r"^\*\*(?P<name>Version|Created|Last updated):\*\*\s*(?P<value>.+?)\s*$")
PHASE_HEADING_RE = re.compile(
    # OCT-LINT: disable=no_hardcoded_secrets reason="Roadmap regex patterns are parser constants, not secrets"
    r"^#\s+PHASE\s+(?P<code>M\d+(?:\.\d+[a-z]?|[a-z])?)\s*(?:[-\u2014]\s*(?P<label>.+?))?\s*$",
    re.IGNORECASE,
)
ITEM_ID_RE = re.compile(
    # OCT-LINT: disable=no_hardcoded_secrets reason="Roadmap regex patterns are parser constants, not secrets"
    r"^(?:"
    r"M\d+(?:\.\d+(?:\.[a-z0-9]+)?[a-z]?|[a-z])?|"
    r"M11-GATE|"
    r"FS-\d+|V\d+|L\d+|U\d+|X\d+|QUEUE-\d+|TUI-R\d+|"
    r"[A-Z](?:\.\d+[a-z]?| V2|\s+\‍([A-Z]\‍))?|"
    r"[A-Z]\.\d+[a-z]?(?:-\w+)?|"
    r"P\.\d+[a-z]?(?:-\w+)?|O\d[a-z]?"
    r")$"
)
PHASE_CODE_RE = re.compile(r"^M\d+(?:\.\d+[a-z]?|[a-z])?$|^M11-GATE$")
FRACTION_RE = re.compile(r"\b\d+\s*/\s*\d+\b")
PLAN_ID_RE = re.compile(r"\b\d{3}-[A-Za-z0-9][A-Za-z0-9_.-]*\b")
 
DONE_MARKERS = ("\u2705", "[x]", "done", "complete", "completed", "shipped", "closed")
PROGRESS_MARKERS = ("\U0001f535", "[~]", "in progress", "partial", "started")
BLOCKED_MARKERS = ("\U0001f512", "[!]", "blocked", "gated")
TODO_MARKERS = ("\u2b1c", "[ ]", "todo", "not started")
 
PRIORITY_MARKERS = (
    ("critical", ("\U0001f534", "critical", "urgent")),
    ("high", ("\U0001f7e0", "high")),
    ("medium", ("\U0001f7e1", "medium")),
    ("lower", ("\U0001f7e2", "lower", "low", "nice-to-have")),
)
 
DEFAULT_TRACKS = (
    ParsedTrack("m_phase", "M-phase delivery", "Active migration-era sequence", 10),
    ParsedTrack("registry", "Orchestration Registry", "M11 registry track", 20),
    ParsedTrack("vision", "Vision-track", "AI Software Factory backlog", 30),
    ParsedTrack("memory", "Memory / context / learning", "Memory backlog", 40),
    ParsedTrack("undo", "Undo / rollback", "Rollback backlog", 50),
    ParsedTrack("tooling", "Tooling extensions", "Tooling backlog", 60),
    ParsedTrack("git_workflow", "Git workflow", "Repository workflow backlog", 70),
    ParsedTrack("pre_m0_era", "Pre-M0 era", "Historical carried-forward work", 80),
)
 
 
def parse_roadmap_file(path: str | Path) -> ParsedRoadmap:
    """Parse one ROADMAP markdown artifact into normalized records."""
 
    source = Path(path)
    text = source.read_text(encoding="utf-8")
    if not text.strip():
        raise RoadmapParseError(f"roadmap file is empty: {source}")
    file_kind = _file_kind(source)
    version_id = _version_id(source, text)
    if version_id is None:
        raise RoadmapParseError(f"roadmap version not found: {source}")
    header = _header_values(text)
    title = _title(text)
    created_at = header.get("created") or _fallback_timestamp(version_id)
    last_updated_at = header.get("last updated") or created_at
    phases = _parse_phases(text, version_id)
    items = _parse_items(text, file_kind, version_id, phases)
    dependencies = _parse_dependencies(text, {item.item_id for item in items})
    render_sections = _parse_render_sections(text, file_kind)
    return ParsedRoadmap(
        path=source,
        file_kind=file_kind,
        version_id=version_id,
        title=title,
        content_sha256=hashlib.sha256(text.encode("utf-8")).hexdigest(),
        era_id=_era_id(version_id),
        created_at=created_at,
        last_updated_at=last_updated_at,
        phases=phases,
        tracks=DEFAULT_TRACKS,
        items=items,
        dependencies=dependencies,
        render_sections=render_sections,
    )
 
 
def canonical_record_hash(table: str, natural_key: str, fields: dict[str, object]) -> str:
    """Return the stable ingest hash for one semantic record."""
 
    payload = {"table": table, "natural_key": natural_key, "fields": fields}
    return hashlib.sha256(canonical_json(payload).encode("utf-8")).hexdigest()
 
 
def _file_kind(path: Path) -> str:
    name = path.name.upper()
    if name.startswith("ROADMAP-EXEC-ORDER-"):
        return "roadmap_exec_order"
    if name.startswith("ROADMAP-CHANGELOG-"):
        return "roadmap_changelog"
    if name.startswith("ROADMAP-TODO-"):
        return "roadmap_todo"
    if name.startswith("ROADMAP-"):
        return "roadmap"
    raise RoadmapParseError(f"not a recognized roadmap artifact: {path.name}")
 
 
def _version_id(path: Path, text: str) -> str | None:
    match = VERSION_RE.search(path.stem)
    if match is None:
        header = _header_values(text)
        match = VERSION_RE.search(header.get("version", ""))
    if match is None:
        return None
    return f"v{match.group('version')}"
 
 
def _era_id(version_id: str) -> str:
    major = version_id.removeprefix("v").split(".", 1)[0]
    return f"v{major}"
 
 
def _header_values(text: str) -> dict[str, str]:
    values: dict[str, str] = {}
    for line in text.splitlines():
        match = HEADER_RE.match(line.strip())
        if match is not None:
            values[match.group("name").lower()] = match.group("value").strip()
    return values
 
 
def _title(text: str) -> str:
    match = H1_RE.search(text)
    if match is None:
        raise RoadmapParseError("roadmap H1 title not found")
    return match.group("title").strip()
 
 
def _fallback_timestamp(version_id: str) -> str:
    return f"{version_id}-source"
 
 
def _parse_phases(text: str, version_id: str) -> tuple[ParsedPhase, ...]:
    phases: dict[str, ParsedPhase] = {}
    order = 0
    for line in text.splitlines():
        match = PHASE_HEADING_RE.match(line.strip())
        if match is None:
            continue
        order += 10
        code = match.group("code")
        label = _strip_markdown(match.group("label") or code)
        phases[code] = ParsedPhase(
            phase_code=code,
            label=label,
            sort_order=order,
            parent_phase_code=_parent_phase_code(code),
            parallel_track=code == "M11",
            target_version=_version_target(version_id),
        )
 
    for row in _iter_markdown_rows(text):
        code = _phase_code_from_cells(row)
        if code is None or code in phases:
            continue
        order += 10
        label = _strip_markdown(_cell_after_phase_code(row, code) or code)
        phases[code] = ParsedPhase(
            phase_code=code,
            label=label,
            sort_order=order,
            parent_phase_code=_parent_phase_code(code),
            priority_label=_priority_from_cells(row),
            parallel_track=code == "M11",
            target_version=_version_target(version_id),
        )
    return tuple(sorted(phases.values(), key=lambda item: item.sort_order))
 
 
def _parse_items(
    text: str,
    file_kind: str,
    version_id: str,
    phases: tuple[ParsedPhase, ...],
) -> tuple[ParsedItem, ...]:
    phase_codes = {phase.phase_code for phase in phases}
    seen: dict[str, ParsedItem] = {}
    order = 0
    for row in _iter_markdown_rows(text):
        item_id = _item_id_from_cells(row)
        if item_id is None:
            continue
        order += 10
        title = _item_title(row, item_id)
        if not title:
            title = item_id
        phase_code = _phase_for_item(item_id, phase_codes)
        track_kind = _track_kind(item_id, file_kind, phase_code)
        status = _status_from_cells(row)
        priority = _priority_from_cells(row)
        prose = _row_prose(row, item_id, title)
        candidate = ParsedItem(
            item_id=item_id,
            title=title,
            track_kind=track_kind,
            sort_order=order,
            status=status,
            priority=priority,
            phase_code=phase_code,
            parent_item_id=_parent_item_id(item_id),
            target_version=_version_target(version_id),
            prose=prose,
        )
        previous = seen.get(item_id)
        if previous is None or _item_score(candidate) > _item_score(previous):
            seen[item_id] = candidate
    return tuple(sorted(seen.values(), key=lambda item: (item.sort_order, item.item_id)))
 
 
def _parse_dependencies(
    text: str,
    item_ids: set[str],
) -> tuple[ParsedDependency, ...]:
    dependencies: set[ParsedDependency] = set()
    for row in _iter_markdown_rows(text):
        item_id = _item_id_from_cells(row)
        if item_id is None:
            continue
        for cell in row:
            if "depend" not in cell.lower() and "gate" not in cell.lower():
                continue
            for target in item_ids:
                if target == item_id:
                    continue
                if re.search(rf"(?<![A-Za-z0-9_.-]){re.escape(target)}(?![A-Za-z0-9_.-])", cell):
                    dependencies.add(ParsedDependency(item_id, target))
    return tuple(sorted(dependencies, key=lambda item: (item.from_item_id, item.to_item_id, item.kind)))
 
 
def _parse_render_sections(
    text: str,
    file_kind: str,
) -> tuple[ParsedRenderSection, ...]:
    sections: list[ParsedRenderSection] = []
    order = 0
    for line in text.splitlines():
        stripped = line.strip()
        if not stripped.startswith("## "):
            continue
        order += 10
        path = stripped
        query_kind = "prose_only"
        if "execution order" in stripped.lower():
            query_kind = "m_phase_table"
        elif "backlog" in stripped.lower():
            query_kind = "backlog_track_table"
        elif "completed" in stripped.lower() or "recent" in stripped.lower():
            query_kind = "recently_completed_window"
        sections.append(
            ParsedRenderSection(
                file_kind=file_kind,
                section_path=path,
                sort_order=order,
                query_kind=query_kind,
            )
        )
    if sections:
        return tuple(sections)
    return (
        ParsedRenderSection(
            file_kind=file_kind,
            section_path="## Roadmap",
            sort_order=10,
            query_kind="prose_only",
        ),
    )
 
 
def _iter_markdown_rows(text: str) -> list[list[str]]:
    rows: list[list[str]] = []
    for line in text.splitlines():
        stripped = line.strip()
        if not (stripped.startswith("|") and stripped.endswith("|")):
            continue
        cells = [cell.strip() for cell in stripped.strip("|").split("|")]
        if not cells:
            continue
        if all(_is_separator_cell(cell) for cell in cells):
            continue
        rows.append(cells)
    return rows
 
 
def _is_separator_cell(cell: str) -> bool:
    text = cell.replace(":", "").replace("-", "").strip()
    return text == ""
 
 
def _item_id_from_cells(cells: list[str]) -> str | None:
    for cell in cells:
        cleaned = _strip_markdown(cell).strip()
        if ITEM_ID_RE.match(cleaned):
            return cleaned
    return None
 
 
def _phase_code_from_cells(cells: list[str]) -> str | None:
    for cell in cells:
        cleaned = _strip_markdown(cell).strip()
        if PHASE_CODE_RE.match(cleaned):
            return cleaned
    return None
 
 
def _cell_after_phase_code(cells: list[str], code: str) -> str | None:
    for index, cell in enumerate(cells):
        if _strip_markdown(cell).strip() != code:
            continue
        if index + 1 < len(cells):
            return cells[index + 1]
    return None
 
 
def _item_title(cells: list[str], item_id: str) -> str:
    for index, cell in enumerate(cells):
        if _strip_markdown(cell).strip() != item_id:
            continue
        for candidate in cells[index + 1:]:
            text = _strip_markdown(candidate)
            if text and not _looks_status_or_priority(text):
                return text
    for cell in cells:
        text = _strip_markdown(cell)
        if text and text != item_id and not _looks_status_or_priority(text):
            return text
    return ""
 
 
def _row_prose(cells: list[str], item_id: str, title: str) -> str | None:
    values = [
        _strip_markdown(cell)
        for cell in cells
        if _strip_markdown(cell) not in {item_id, title}
        and not _looks_status_or_priority(_strip_markdown(cell))
    ]
    text = " | ".join(value for value in values if value)
    return text or None
 
 
def _status_from_cells(cells: list[str]) -> str:
    haystack = " ".join(cells).lower()
    if any(marker.lower() in haystack for marker in BLOCKED_MARKERS):
        return "blocked"
    if any(marker.lower() in haystack for marker in DONE_MARKERS):
        return "done"
    if any(marker.lower() in haystack for marker in PROGRESS_MARKERS):
        return "in_progress"
    if any(marker.lower() in haystack for marker in TODO_MARKERS):
        return "todo"
    if "deferred" in haystack:
        return "deferred"
    return "todo"
 
 
def _priority_from_cells(cells: list[str]) -> str | None:
    haystack = " ".join(cells).lower()
    for priority, markers in PRIORITY_MARKERS:
        if any(marker.lower() in haystack for marker in markers):
            return priority
    return None
 
 
def _looks_status_or_priority(text: str) -> bool:
    lowered = text.lower()
    if FRACTION_RE.search(lowered):
        return True
    marker_groups = DONE_MARKERS + PROGRESS_MARKERS + BLOCKED_MARKERS + TODO_MARKERS
    if any(marker.lower() in lowered for marker in marker_groups):
        return True
    return any(
        marker.lower() in lowered
        for _, markers in PRIORITY_MARKERS
        for marker in markers
    )
 
 
def _phase_for_item(item_id: str, phase_codes: set[str]) -> str | None:
    if item_id in phase_codes:
        return item_id
    if not item_id.startswith("M"):
        return None
    parts = item_id.split(".")
    candidates = [parts[0]]
    if len(parts) > 1:
        candidates.insert(0, ".".join(parts[:2]))
    for candidate in candidates:
        if candidate in phase_codes:
            return candidate
    return parts[0] if parts[0].startswith("M") else None
 
 
def _track_kind(item_id: str, file_kind: str, phase_code: str | None) -> str:
    if item_id.startswith("V"):
        return "vision"
    if file_kind == "roadmap_exec_order" and phase_code is None:
        return "pre_m0_era"
    if phase_code is not None:
        return "m_phase"
    if item_id.startswith(("FS-", "P.", "G.", "B.", "O", "L", "U", "X")):
        return "backlog"
    return "pre_m0_era"
 
 
def _parent_phase_code(code: str) -> str | None:
    if "." in code:
        return code.split(".", 1)[0]
    return None
 
 
def _parent_item_id(item_id: str) -> str | None:
    if "." not in item_id or not item_id.startswith("M"):
        return None
    return item_id.rsplit(".", 1)[0]
 
 
def _version_target(version_id: str) -> str:
    return version_id
 
 
def _item_score(item: ParsedItem) -> tuple[int, int]:
    return (
        1 if item.prose else 0,
        len(item.title) + len(item.prose or ""),
    )
 
 
def _strip_markdown(value: str | None) -> str:
    if value is None:
        return ""
    text = value.strip()
    text = re.sub(r"`([^`]+)`", r"\1", text)
    text = re.sub(r"\‍[([^\‍]]+)\‍]\‍([^)]+\‍)", r"\1", text)
    text = text.replace("**", "").replace("__", "").strip()
    return text