orchestrata_ultimate.knowledge_graph.adr_parser

Reference › orchestrata_ultimate › knowledge_graph › adr_parser

A pure text-to-dataclass parser for the ADR corpus (ontology phase 2). parse_adrs reads the ## ADR-NNN: Title sections of a project's decisions.md wiki page and returns a list of frozen ParsedAdr records: number, title, status, date, context, decision, alternatives, consequences, plus the derived register and padded_id. parse_adr_file parses one standalone ADR document of the form docs/adr/NNNN-kebab-title.md into a single record. Only the recognised body bullets (status, date, context, decision, alternatives and consequences) are read; other bullets are ignored. Helpers normalise the status to its canonical leading token among Accepted, Proposed, Rejected, Deprecated and Superseded (_normalize_status), fold raw markdown blocks into paragraph text (_fold_block) and classify the heading digits into an ADR register (_classify_register, ADR-0047). The module does no I/O. Its consumers are the scripts sync_architecture_to_graph, which feeds the records to architecture_writer, and check_adr_graph_sync.

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

ParsedAdr (dataclass)

Decorators: dataclass(frozen=True) · Line 167

One parsed ## ADR-NNN: Title section.

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/knowledge_graph/adr_parser.py

"""Pure text -> dataclass parser for the ADR corpus (wave ontology-phase2).

Parses the ``## ADR-NNN: Title`` sections of the knowledge-wiki's per-project
``decisions.md`` page (e.g.
``knowledge-wiki/projects/orchestrata-ultimate/decisions.md``) into
:class:`ParsedAdr` records that ``scripts/sync_architecture_to_graph.py``
feeds into :meth:`ArchitectureWriter.upsert_architecture_decision`.

Each section is shaped::

    ## ADR-001: Memgraph as Semantic Graph Backend

    - **Status:** Accepted
    - **Date:** 2026-07-07
    - **Context:** ...
    - **Decision:** ...
    - **Alternatives:** ...   (optional -- not every ADR records one)
    - **Consequences:** ...

Only the five body bullets (Status/Date/Context/Decision/Alternatives/
Consequences) are recognized; every other bullet is a Nygard-template
optional field this module has no need for, and is silently ignored. Any
prose before, between, or after ``## ADR-*`` sections (front matter, the page
banner, non-ADR ``##`` headings) is ignored entirely -- only real ADR
sections are extracted.

A bullet's value may wrap across several physical source lines (the writer
hand-wraps long Context/Decision/Consequences prose at the editor's line
length, e.g. ``decisions.md`` ADR-019/ADR-020) -- those indented continuation
lines are folded back into the bullet's single logical value, joined with a
single space, exactly the way Markdown itself treats a "loose" list item's
wrapped lines as one paragraph. This is line-*unwrapping*, not reflow: word
order and content are preserved verbatim, nothing is re-wrapped to a new
width. A blank line (or the next bullet/heading) ends the continuation.

The corpus contains a SECOND section shape (ADR-009..ADR-019): a single
inline meta line ``**Date:** 2026-07-08 · **Status:** Accepted`` followed by
free-prose paragraphs instead of the field bullets. For those sections the
meta line supplies ``date``/``status`` and the whole prose body is captured
into ``decision`` (it IS the decision record), with intra-paragraph wrapping
folded exactly like bullet continuations and paragraph breaks preserved as
``\n\n``. ``context``/``alternatives``/``consequences`` stay ``""``.

``status`` is NORMALIZED to its leading canonical token (``Accepted`` /
``Proposed`` / ``Rejected`` / ``Superseded`` / ``Deprecated``): the corpus
annotates statuses ("Accepted, live", "Accepted (session 226)."), and the
ruling-1 immutability guard in ``ArchitectureWriter`` keys on the EXACT
string ``"Accepted"`` -- an annotated variant would silently dodge the
guard. Unrecognized status text is preserved verbatim.

This module does no file I/O -- callers read the markdown text themselves
(``scripts/sync_architecture_to_graph.py`` resolves the ``decisions.md`` path
at call time, per project doctrine, and passes its contents in here).

``parse_adr_file`` (below) parses a DIFFERENT, later corpus shape: the
one-ADR-per-file records under ``docs/adr/NNNN-kebab-title.md`` (format
declared in ``docs/adr/README.md``: ``Status · Date · Context · Decision ·
Alternatives rejected · Consequences · Provenance``). That corpus uses a
level-1 ``# ADR-NNNN: Title`` (or ``# ADR-NNNN — Title``) heading, a
metadata bullet block shaped ``- **Field**: value`` (colon OUTSIDE the bold,
the inverse of this file's ``## ADR-NNN`` bullet shape), and free-prose
``## Context`` / ``## Decision`` / ``## Alternatives rejected`` /
``## Consequences`` sections rather than inline bullets. It is a sibling
parser, not a generalization of ``parse_adrs`` -- the two shapes differ in
heading level, number width, and bullet-colon position, so unifying them
into one regex set would risk silently breaking the working ``decisions.md``
path. See ``parse_adr_file``'s own docstring for its folding policy.
"""

from __future__ import annotations

import re
from dataclasses import dataclass

_HEADING_RE = re.compile(r"^##\s+ADR-([^:\s][^:]*):\s*(.*)$")
_BULLET_RE = re.compile(r"^-\s+\*\*([A-Za-z][A-Za-z ]*):\*\*\s?(.*)$")
# The ADR-009..019 inline shape: **Date:** 2026-07-08 · **Status:** Accepted
_META_LINE_RE = re.compile(
    r"^\*\*Date:\*\*\s*(?P<date>.*?)\s*·\s*\*\*Status:\*\*\s*(?P<status>.*)$"
)

# The only five body bullets this parser recognizes -- everything else
# (e.g. a stray "- **Deciders:**" some future ADR might add) is ignored.
_BULLET_FIELDS: frozenset[str] = frozenset(
    {"status", "date", "context", "decision", "alternatives", "consequences"}
)

# Canonical lifecycle tokens: an annotated corpus status ("Accepted, live",
# "Accepted (session 226).") normalizes to its leading token so the ruling-1
# guard (exact-match on "Accepted") always engages. Longest-first so a
# future "AcceptedX" never half-matches.
_STATUS_TOKENS: tuple[str, ...] = (
    "Superseded",
    "Deprecated",
    "Accepted",
    "Proposed",
    "Rejected",
)


def _classify_register(digits: str) -> str:
    """Classify a heading's digit string into its ADR register (ADR-0047).

    ``"factory"`` for a 4-digit heading id (``docs/adr/NNNN-*.md`` and its
    wiki mirror); ``"project"`` for a 1-3 digit one (the wiki's own,
    pre-factory register). The heading's raw digit string IS the identity
    (``padded_id``) -- this only classifies its width.
    """
    return "factory" if len(digits) == 4 else "project"


def _normalize_status(raw: str) -> str:
    """Reduce an annotated status to its canonical leading token.

    Unrecognized text is returned verbatim (never silently coerced).
    """
    text = raw.strip()
    for token in _STATUS_TOKENS:
        if text == token or text.startswith((f"{token} ", f"{token},", f"{token}.")):
            return token
    return text


# ---------------------------------------------------------------------------
# docs/adr/*.md -- the one-ADR-per-file corpus (see module docstring).
# ---------------------------------------------------------------------------

# "# ADR-0001: Title" or "# ADR-0037 — Title" (em dash variant, ADR-0037).
# The number group is permissive ([^:—] rather than \d+), matching
# _HEADING_RE's design above: a non-numeric "number" (e.g. "ADR-abcd") must
# still MATCH the heading shape and reach the int() conversion below, so the
# "no parseable number" error path stays reachable instead of being dead
# code a stricter \d+ would make unhittable.
_FILE_HEADING_RE = re.compile(r"^#\s+ADR-([^:\s—][^:—]*)\s*(?::|—)\s*(.*)$")
# "- **Field**: value" -- colon OUTSIDE the bold (inverse of _BULLET_RE).
_FILE_LIST_BULLET_RE = re.compile(r"^-\s+\*\*([A-Za-z][A-Za-z ]*)\*\*:\s*(.*)$")
# "**Field:** value" -- bare bold-colon line with no leading "- " (ADR-0037's
# Status/Date lines).
_FILE_BARE_META_RE = re.compile(r"^\*\*([A-Za-z][A-Za-z ]*):\*\*\s*(.*)$")
# "Status: **Accepted** (annotation) · Date: 2026-10-06" -- the plain-text
# one-line form (ADR-0073 and later): no bullet, no bold label, the Date
# riding the same line after a middle dot.
_FILE_STATUS_LINE_RE = re.compile(r"^Status:\s*(.*?)(?:\s+·\s+Date:\s*(.*))?$")
_FILE_SECTION_RE = re.compile(r"^##\s+(.*)$")


def _fold_block(text: str) -> str:
    """Fold a raw markdown text block into paragraph text.

    Blank-line-separated chunks become paragraphs (joined ``\n\n``); within
    a chunk, physical lines join with a single space -- the same
    hand-wrap-unwrapping contract ``parse_adrs`` documents above, restated
    as a block operation instead of an incremental line-scan.
    """
    stripped = text.strip("\n")
    if not stripped:
        return ""
    chunks = re.split(r"\n\s*\n", stripped)
    paragraphs = [
        " ".join(line.strip() for line in chunk.splitlines() if line.strip())
        for chunk in chunks
    ]
    return "\n\n".join(p for p in paragraphs if p)


@dataclass(frozen=True)
class ParsedAdr:
    """One parsed ``## ADR-NNN: Title`` section.

    ``alternatives`` defaults to ``""`` -- not every recorded ADR names one.

    ADR-0047: this project has TWO independent ADR registers that both
    number their entries starting at 1 -- the factory register
    (``docs/adr/NNNN-*.md``, always 4-digit) and the project register (the
    wiki's own ``ADR-NNN`` sections, 1-3 digit). ``number`` alone cannot
    disambiguate them (ADR-001 and ADR-0001 both have ``number == 1``), so
    every ``ParsedAdr`` also carries ``register`` (``"factory"`` for a
    4-digit heading id, ``"project"`` for a 1-3 digit one) and ``padded_id``
    (the heading's digit string verbatim, e.g. ``"0001"`` / ``"001"``) --
    together, ``(register, padded_id)`` is the real identity. ``number``
    stays for callers that only need the integer.
    """

    number: int
    title: str
    status: str
    date: str
    context: str
    decision: str
    alternatives: str
    consequences: str
    register: str
    padded_id: str


def parse_adrs(markdown: str) -> list[ParsedAdr]:
    """Extract every ``## ADR-NNN: Title`` section from ``markdown``.

    Deterministic and byte-stable: calling this twice on the same text
    yields equal results. Non-ADR content (front matter, page banners, other
    ``##`` headings) is skipped over, never raising.

    Raises:
        ValueError: a line matches the ``## ADR-...:`` heading shape but the
            ``NNN`` portion is not a parseable integer -- names the offending
            heading line.
    """
    lines = markdown.splitlines()
    adrs: list[ParsedAdr] = []
    i = 0
    n = len(lines)
    while i < n:
        heading = _HEADING_RE.match(lines[i])
        if heading is None:
            i += 1
            continue

        number_text, title = heading.group(1).strip(), heading.group(2).strip()
        try:
            number = int(number_text)
        except ValueError as exc:
            raise ValueError(
                f"Malformed ADR heading (no parseable number): {lines[i]!r}"
            ) from exc
        register = _classify_register(number_text)
        i += 1

        fields: dict[str, list[str]] = {name: [] for name in _BULLET_FIELDS}
        current: str | None = None
        meta_date: str | None = None
        meta_status: str | None = None
        # Prose paragraphs (the ADR-009..019 shape): list of folded
        # paragraphs, each itself a list of physical lines.
        prose: list[list[str]] = []
        while i < n:
            raw = lines[i]
            if _HEADING_RE.match(raw) or raw.lstrip().startswith("## "):
                break  # next ADR section, or an unrelated ## heading
            stripped = raw.strip()
            # Newer bulleted entries write Date and Status on ONE dashed
            # line ("- **Date:** … · **Status:** …"). The bullet branch would
            # swallow it whole into the `date` field and, worse, leave
            # `current="date"` so every following dashed prose line is folded
            # into `date` too. Strip the dash and let the meta regex parse it
            # (it expects exactly the undashed shape).
            if stripped.startswith("- "):
                meta = _META_LINE_RE.match(stripped[2:])
                if meta is not None:
                    meta_date = meta.group("date").strip()
                    meta_status = meta.group("status").strip()
                    current = None
                    i += 1
                    continue
            meta = _META_LINE_RE.match(stripped)
            if meta is not None:
                meta_date = meta.group("date").strip()
                meta_status = meta.group("status").strip()
                current = None
                i += 1
                continue
            bullet = _BULLET_RE.match(raw)
            if bullet is not None:
                label = bullet.group(1).strip().lower()
                if label in _BULLET_FIELDS:
                    current = label
                    fields[current].append(bullet.group(2).strip())
                    i += 1
                    continue
                # Unknown-label bullet (e.g. "- **Full record:** …"): keep
                # its text as body prose so pointer detection and the register
                # summary stay intact; do not let it poison `current`.
                current = None
                if prose and i > 0 and lines[i - 1].strip() != "":
                    prose[-1].append(stripped)
                else:
                    prose.append([stripped])
                i += 1
                continue
            if stripped == "":
                current = None
                i += 1
                continue
            if current is not None:
                # Continuation line of the current bullet's wrapped prose.
                fields[current].append(stripped)
            else:
                # Free-prose body line (meta-line sections): start a new
                # paragraph after a blank line, else fold into the current one.
                if prose and i > 0 and lines[i - 1].strip() != "":
                    prose[-1].append(stripped)
                else:
                    prose.append([stripped])
            i += 1

        status_raw = " ".join(fields["status"]) or (meta_status or "")
        date = " ".join(fields["date"]) or (meta_date or "")
        prose_text = "\n\n".join(" ".join(par) for par in prose) if prose else ""
        decision = " ".join(fields["decision"])
        context = " ".join(fields["context"])
        if not decision and prose_text:
            # Unstructured section: the prose IS the decision record.
            decision = prose_text
        elif prose_text:
            # Mixed shape (bullet fields + free prose, e.g. "- **Decision:**"
            # with "- **Measured:**" / "Full record:" narrative): the prose is
            # context — never drop it, or pointer detection loses the
            # "Full record:" line and the register summary vanishes.
            context = f"{context} {prose_text}".strip() if context else prose_text
        adrs.append(
            ParsedAdr(
                number=number,
                title=title,
                status=_normalize_status(status_raw),
                date=date,
                context=context,
                decision=decision,
                alternatives=" ".join(fields["alternatives"]),
                consequences=" ".join(fields["consequences"]),
                register=register,
                padded_id=number_text,
            )
        )

    return adrs


def parse_adr_file(markdown: str) -> ParsedAdr:
    """Parse ONE ``docs/adr/NNNN-kebab-title.md`` document into a
    :class:`ParsedAdr`.

    Shape (``docs/adr/README.md``)::

        # ADR-0001: Title

        - **Status**: Accepted
        - **Date**: 2026-07-23
        - **Spec**: SPEC-033 §3          (ignored -- not a ParsedAdr field)
        - **Supersedes**: —              (ignored here; the caller reads
                                           the ADR's own PROSE for a real
                                           supersession, per project doctrine
                                           -- this bullet is not trusted alone)

        ## Context
        ...
        ## Decision
        ...
        ## Alternatives rejected
        ...
        ## Consequences
        ...
        ## Provenance
        ...

    ``status`` is normalized via :func:`_normalize_status`, exactly like
    ``parse_adrs`` -- an annotated status (e.g. ADR-0038's ``"Accepted (de
    facto, by commit ...)"``) must still key the ruling-1 immutability guard
    on the exact string ``"Accepted"``.

    ``date`` is kept VERBATIM (not normalized) -- this is intentional and is
    where a reconstructed record's provenance nuance survives: ADR-0038's
    date bullet reads ``"2026-07-06 (the decision, as executed) · this
    record written 2026-08-19"``, and that full annotation is preserved in
    the ``date`` field rather than collapsed to a single ISO date.

    Any text between the metadata bullets and the first ``## `` heading
    (e.g. ADR-0038's "This is a reconstructed ADR" callout) is prepended to
    ``context``. A ``## `` heading that is not Context/Decision/
    Alternatives*/Consequences (e.g. ``## Provenance``, or a one-off custom
    heading like ADR-0038's ``## What is recorded, and what is not``) has no
    slot in :class:`ParsedAdr` -- rather than dropping it, this parser folds
    it into ``consequences``, prefixed with ``### <Heading>`` so the origin
    stays visible in the stored prose. This is a parser-level folding
    choice, not a graph-schema change -- no new field is invented.

    Raises:
        ValueError: the first non-blank line is not a parseable
            ``# ADR-NNNN`` heading.
    """
    lines = markdown.splitlines()
    n = len(lines)
    i = 0
    while i < n and lines[i].strip() == "":
        i += 1
    if i >= n:
        raise ValueError("Empty ADR document")
    heading = _FILE_HEADING_RE.match(lines[i])
    if heading is None:
        raise ValueError(
            f"Malformed ADR file (no '# ADR-NNNN: Title' heading): {lines[i]!r}"
        )
    try:
        number = int(heading.group(1))
    except ValueError as exc:
        raise ValueError(
            f"Malformed ADR heading (no parseable number): {lines[i]!r}"
        ) from exc
    padded_id = heading.group(1).strip()
    title = heading.group(2).strip()
    i += 1
    # A blank line separates the heading from the metadata block in every
    # observed document -- skip it before Phase A starts, else the first
    # blank-line check below would immediately (and wrongly) end the block.
    while i < n and lines[i].strip() == "":
        i += 1

    # --- Phase A: metadata bullet block (Status/Date/... until blank/##) ---
    status_parts: list[str] = []
    date_parts: list[str] = []
    current_label: str | None = None
    while i < n:
        raw = lines[i]
        stripped = raw.strip()
        if stripped == "":
            i += 1
            break
        if _FILE_SECTION_RE.match(stripped):
            break  # no blank line before the first ## section
        status_line = _FILE_STATUS_LINE_RE.match(stripped)
        if status_line is not None:
            current_label = "status"
            status_parts.append(status_line.group(1).replace("**", "").strip())
            if status_line.group(2):
                date_parts.append(status_line.group(2).strip())
            i += 1
            continue
        bullet = _FILE_LIST_BULLET_RE.match(raw) or _FILE_BARE_META_RE.match(stripped)
        if bullet is not None:
            label = bullet.group(1).strip().lower()
            value = bullet.group(2).strip()
            current_label = label
            if label == "status":
                status_parts.append(value)
            elif label == "date":
                date_parts.append(value)
            i += 1
            continue
        # Continuation line of the currently-open bullet.
        if current_label == "status":
            status_parts.append(stripped)
        elif current_label == "date":
            date_parts.append(stripped)
        # else: continuation of an unrecognized bullet (Spec/Supersedes/
        # Amends/Relates to) -- discarded, matching parse_adrs' policy of
        # silently ignoring non-body bullets.
        i += 1

    # --- Phase B: preamble prose before the first ## heading ---
    preamble_start = i
    while i < n and _FILE_SECTION_RE.match(lines[i].strip()) is None:
        i += 1
    preamble = _fold_block("\n".join(lines[preamble_start:i]))

    # --- Phase C: ## sections ---
    blocks: dict[str, list[str]] = {
        "context": [],
        "decision": [],
        "alternatives": [],
        "consequences": [],
    }
    while i < n:
        # Phase B only stops here when lines[i] matches _FILE_SECTION_RE (or
        # i == n, which the loop guard above already excludes), and every
        # iteration below advances i to the same stopping condition before
        # looping again -- lines[i] is guaranteed to be a "## " heading.
        section = _FILE_SECTION_RE.match(lines[i].strip())
        assert section is not None
        heading_text = section.group(1).strip()
        i += 1
        body_start = i
        while i < n and _FILE_SECTION_RE.match(lines[i].strip()) is None:
            i += 1
        body = "\n".join(lines[body_start:i])

        lower = heading_text.lower()
        if lower.startswith("context"):
            target = "context"
        elif lower.startswith("decision"):
            target = "decision"
        elif lower.startswith("alternatives"):
            target = "alternatives"
        elif lower.startswith("consequences"):
            target = "consequences"
        else:
            # No dedicated slot (e.g. Provenance) -- fold into consequences,
            # labeled so the origin is traceable.
            target = "consequences"
            blocks[target].append(f"### {heading_text}")
        folded = _fold_block(body)
        if folded:
            blocks[target].append(folded)

    context = "\n\n".join(p for p in [preamble] + blocks["context"] if p)

    return ParsedAdr(
        number=number,
        title=title,
        status=_normalize_status(" ".join(status_parts)),
        date=" ".join(date_parts),
        context=context,
        decision="\n\n".join(blocks["decision"]),
        alternatives="\n\n".join(blocks["alternatives"]),
        consequences="\n\n".join(blocks["consequences"]),
        register="factory",
        padded_id=padded_id,
    )