orchestrata_ultimate.factory.assumptions.sources

Reference › orchestrata_ultimate › factory › assumptions › sources

sources supplies the pool of candidate pinning artifacts that the stage-1 prefilter matches assumptions against (SPEC-033 row B13). PinningSource is a protocol with one method, pinnings, returning a list of Pinning records. Three sources implement it:

• AcSource scans specs/*/spec.md under a given directory for acceptance-criterion lines (AC-NNN-...) and REQ-NNN-... headings, joining wrapped continuation lines of an AC bullet.
• AdrSource scans NNNN-*.md files in an ADR directory and yields one pinning per ADR, built from the title (with the redundant ADR-NNNN: prefix removed) and an excerpt of the Decision section found by _extract_decision_section.
• KbSource turns promoted, non-superseded MATCHED adjudications from a KBProtocol (see ledger) into candidates, so a ruling also matches text that is only a fuzzy neighbour.

The module docstring says the anchor source is not built here; the anchors package supplies it later (task B15a). Importers are cascade, prefilter, contracts.models, the rules detector and the scripts.factory_assumptions CLI.

At a glance

Class diagram

Dependency diagram

API

Classes

PinningSource (protocol)

Bases: Protocol · Line 44

Anything that can yield a pool of candidate pinning artifacts.

AcSource

Line 52

Scans <specs_dir>/*/spec.md for AC lines and REQ-* headings.

AdrSource

Line 134

Scans <adr_dir>/NNNN-*.md for ADR title + Decision excerpts.

KbSource

Line 170

Promoted MATCHED adjudications as stage-1 candidate pinnings.

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/factory/assumptions/sources.py

"""Pinning-artifact sources for the SPEC-033 arbiter cascade (B13).

A :class:`PinningSource` yields the pool of candidate pinning artifacts the
stage-1 prefilter matches assumption text against. Three sources are built
here:

- :class:`AcSource` — acceptance criteria (and ``REQ-NNN-*`` section
  headings) scanned out of ``specs/*/spec.md``.
- :class:`AdrSource` — ADR title + Decision-section excerpts scanned out of
  ``docs/adr/NNNN-*.md``.
- :class:`KbSource` — previously-promoted ``MATCHED`` adjudications, so a
  KB-promoted ruling also becomes a stage-1 candidate for anything that only
  fuzzy-matches it (the exact-fingerprint / Jaccard>=0.8 case is already
  handled directly by stage 0; this is the wider net).

Anchors (SPEC-033 §2) are **not implemented here** — that source arrives with
task B15a. ``PinningSource`` is the seam it will plug into.
"""

from __future__ import annotations

import re
from pathlib import Path
from typing import Protocol

from orchestrata_ultimate.factory.assumptions.ledger import KBProtocol
from orchestrata_ultimate.factory.assumptions.models import Pinning, PinningKind, Ruling

_AC_LINE_RE = re.compile(
    r"^-\s*\*\*(?P<ref>AC-\d{3}-[A-Za-z0-9]+-\d{2})\*\*:\s*(?P<text>.+)$"
)
_REQ_HEADING_RE = re.compile(
    r"^#{1,6}\s*(?P<ref>REQ-\d{3}-\w+(?:-\w+)*)\s*(?:[—-]+\s*(?P<text>.+))?$"
)
_H1_RE = re.compile(r"^#\s+(?P<text>.+)$")
_H2_RE = re.compile(r"^##\s+(?P<text>.+)$")
_ADR_FILENAME_RE = re.compile(r"^(?P<num>\d{4})-")
# ADR H1s are conventionally "# ADR-NNNN: <title>" -- strip the redundant
# "ADR-NNNN:" prefix (the pinning's ``ref`` already carries it) so the
# excerpt is just the descriptive title.
_ADR_TITLE_PREFIX_RE = re.compile(r"^ADR-\d{4}:\s*", re.IGNORECASE)


class PinningSource(Protocol):
    """Anything that can yield a pool of candidate pinning artifacts."""

    def pinnings(self) -> list[Pinning]:
        """Return the pool of candidate pinning artifacts this source offers."""
        ...


class AcSource:
    """Scans ``<specs_dir>/*/spec.md`` for AC lines and ``REQ-*`` headings."""

    def __init__(self, specs_dir: Path):
        self.specs_dir = Path(specs_dir)

    def pinnings(self) -> list[Pinning]:
        """Return one ``Pinning`` per AC line / ``REQ-*`` heading found."""
        if not self.specs_dir.exists():
            return []
        return [
            pinning
            for spec_path in sorted(self.specs_dir.glob("*/spec.md"))
            for pinning in self._scan(spec_path.read_text(encoding="utf-8"))
        ]

    @staticmethod
    def _consume_ac_continuation(lines: list[str], start: int) -> tuple[list[str], int]:
        """Collect wrapped continuation lines following an AC bullet.

        Real specs wrap long AC text onto indented follow-on lines (spec 033
        §6) -- stop at the first blank line or the next bullet/heading.
        """
        parts: list[str] = []
        i = start
        while (
            i < len(lines)
            and lines[i].strip()
            and lines[i][:1].isspace()
            and not lines[i].strip().startswith(("-", "#"))
        ):
            parts.append(lines[i].strip())
            i += 1
        return parts, i

    def _scan(self, text: str) -> list[Pinning]:
        """Scan one spec.md's text for AC lines and REQ-* headings."""
        results: list[Pinning] = []
        lines = text.splitlines()
        i = 0
        while i < len(lines):
            stripped = lines[i].strip()
            ac_match = _AC_LINE_RE.match(stripped)
            if ac_match:
                continuation, i = self._consume_ac_continuation(lines, i + 1)
                excerpt = " ".join([ac_match.group("text").strip(), *continuation])
                results.append(
                    Pinning(
                        kind=PinningKind.AC, ref=ac_match.group("ref"), excerpt=excerpt
                    )
                )
                continue
            req_match = _REQ_HEADING_RE.match(stripped)
            if req_match:
                body = (req_match.group("text") or "").strip()
                results.append(
                    Pinning(
                        kind=PinningKind.AC,
                        ref=req_match.group("ref"),
                        excerpt=body or req_match.group("ref"),
                    )
                )
            i += 1
        return results


def _extract_decision_section(lines: list[str]) -> str:
    """Return the ``## Decision`` section body, or ``""`` if absent."""
    body: list[str] = []
    in_decision = False
    for line in lines:
        h2_match = _H2_RE.match(line.strip())
        if h2_match:
            if in_decision:
                break
            in_decision = h2_match.group("text").strip().lower() == "decision"
            continue
        if in_decision:
            body.append(line)
    return "\n".join(body).strip()


class AdrSource:
    """Scans ``<adr_dir>/NNNN-*.md`` for ADR title + Decision excerpts."""

    def __init__(self, adr_dir: Path):
        self.adr_dir = Path(adr_dir)

    def pinnings(self) -> list[Pinning]:
        """Return one ``Pinning`` per ADR file (title + Decision excerpt)."""
        results: list[Pinning] = []
        if not self.adr_dir.exists():
            return results
        for path in sorted(self.adr_dir.glob("*.md")):
            if path.name.lower() == "readme.md":
                continue
            num_match = _ADR_FILENAME_RE.match(path.name)
            if not num_match:
                continue
            lines = path.read_text(encoding="utf-8").splitlines()
            title = ""
            for line in lines:
                h1_match = _H1_RE.match(line.strip())
                if h1_match:
                    title = _ADR_TITLE_PREFIX_RE.sub("", h1_match.group("text").strip())
                    break
            decision_body = _extract_decision_section(lines)
            excerpt = f"{title}\n\n{decision_body}".strip() if decision_body else title
            results.append(
                Pinning(
                    kind=PinningKind.ADR,
                    ref=f"ADR-{num_match.group('num')}",
                    excerpt=excerpt,
                )
            )
        return results


class KbSource:
    """Promoted ``MATCHED`` adjudications as stage-1 candidate pinnings."""

    def __init__(self, kb: KBProtocol):
        self.kb = kb

    def pinnings(self) -> list[Pinning]:
        """Return one ``Pinning`` per promoted, non-superseded MATCHED adjudication."""
        return [
            Pinning(
                kind=PinningKind.ADJUDICATION,
                ref=adj.id,
                excerpt=adj.assumption_text,
            )
            for adj in self.kb.all()
            if adj.ruling == Ruling.MATCHED and adj.superseded_by is None
        ]