orchestrata_ultimate.factory.detectors.rules

Reference › orchestrata_ultimate › factory › detectors › rules

rules is the largest module of the C3 detector set (SPEC-033 row B22, ADR-0029): one function per rule, plus the registry the engine iterates. Each rule takes a whole-document RuleContext (text, source name, lines, located SegmentInfo records and repository root, built once by build_context using clause) and returns Finding records. ALL_RULES is built by _build_registry from three tiers: _tier1_registry (blocking; the dictionary-family rules with measured precision, built by _tier1_dictionary_registry, plus structural rules from _tier1_structural_registry), _tier2_registry (advisory) and _tier3_registry (questions).

Two global rules apply to every rule. Conditional clauses are exempt from _comparative_no_baseline and _negative_statement, and _finding_from_segment halves the weight of any finding inside a rationale clause. _lexicon_rule builds the many rules that report one finding per lexicon hit. Hand-written rules include _unresolved_placeholder, _missing_unit, _exact_duplicate, _template_violation, _incomplete_reference (checks REQ, AC, ADR and SPEC ids against the repository, reusing the pinning sources in assumptions.sources), _non_atomic_criterion, _coordination_ambiguity, _vague_pronoun, _quantifier_scope, _operator_placement, _boundary_word and _underspecified_head_noun. No rule sets a completeness verdict (ADR-0011), and a passive-voice rule is deliberately absent. The docstring lists rules deferred until a design contract and glossary exist. The decision phase and the scripts.factory_detectors CLI import it; tests are in tests/factory/test_detector_rules.py.

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

SegmentInfo (dataclass)

Decorators: dataclass(frozen=True) · Line 67

One clause segment located at an absolute line + column offset.

RuleContext (dataclass)

Decorators: dataclass(frozen=True) · Line 83

Everything a rule function needs, precomputed once per document.

Functions

Source

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

"""One detector per rule for the SPEC-033 C3 set (B22, ADR-0029).

Each rule function takes the whole-document :class:`RuleContext` and returns
every :class:`~.models.Finding` it produces; :data:`ALL_RULES` is the
registry the engine iterates. Two global rules apply across rule functions
(not per-rule logic duplicated everywhere):

1. **Conditional exemption** (ADR-0029 rule 1) -- ``comparative-no-baseline``
   and ``negative-statement`` skip any segment whose ``kind == "conditional"``
   entirely, regardless of what else is true of it.
2. **Rationale down-weighting** (ADR-0029 rule 3) -- every segment-based
   finding's ``weight`` is :data:`~.models.RATIONALE_WEIGHT` when
   ``segment.kind == "rationale"``, else ``1.0``. Applied uniformly by
   :func:`_finding_from_segment`.

No detector here may ever set a completeness verdict (ADR-0011); every
function below returns *findings*, and a passive-voice rule is deliberately
absent (ADR-0029 rejects it: not an ISO 29148 §5.2.7 smell, not in Femmer's
catalogue, and INCOSE classifies it as not tool-automatable).

Deferred (documented, not built here; needs the B8 design contract /
glossary): undefined-term-vs-glossary, unmapped-requirement, orphan-design-
node, node-with-no-AC/anchor, dangling-interface-reference.
"""

from __future__ import annotations

import re
from collections.abc import Callable
from dataclasses import dataclass
from pathlib import Path

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.paths import get_project_root
from orchestrata_ultimate.factory.assumptions.models import normalize
from orchestrata_ultimate.factory.assumptions.sources import AcSource, AdrSource
from orchestrata_ultimate.factory.detectors.clause import Segment, segment, sentences
from orchestrata_ultimate.factory.detectors.lexicons import (
    AMBIGUOUS_ADVERBS_ADJECTIVES,
    BOUNDARY_WORDS,
    COMPARATIVES,
    DEICTIC,
    LOOPHOLES,
    NEGATION_MARKERS,
    OPEN_ENDED,
    PLACEHOLDERS,
    SOLUTION_LANGUAGE_NOUNS,
    SUBJECTIVE,
    SUPERLATIVES,
    TEMPORAL_NONSPECIFIC,
    UNDERSPECIFIED_HEAD_NOUNS,
    VAGUE_PRONOUNS,
    WEAK_MODALS,
    find_lexicon_matches,
)
from orchestrata_ultimate.factory.detectors.models import (
    RATIONALE_WEIGHT,
    Finding,
    RuleMeta,
    Tier,
)

RuleFn = Callable[["RuleContext"], list[Finding]]


@dataclass(frozen=True)
class SegmentInfo:
    """One clause segment located at an absolute line + column offset."""

    line_no: int
    line_text: str
    sentence: str
    sentence_col: int
    segment: Segment

    def abs_span(self, local_start: int, local_end: int) -> tuple[int, int]:
        """Map a (start, end) offset local to ``segment.text`` to line columns."""
        base = self.sentence_col + self.segment.start
        return (base + local_start, base + local_end)


@dataclass(frozen=True)
class RuleContext:
    """Everything a rule function needs, precomputed once per document."""

    text: str
    source_name: str
    lines: list[str]
    segments: list[SegmentInfo]
    repo_root: Path


def build_context(
    text: str, source_name: str, *, repo_root: Path | None = None
) -> RuleContext:
    """Build a :class:`RuleContext` for ``text``: sentences -> segments."""
    root = repo_root if repo_root is not None else get_project_root()
    lines = text.splitlines()
    seg_infos: list[SegmentInfo] = []
    cursor_by_line: dict[int, int] = {}
    for line_no, sent in sentences(text):
        line_text = lines[line_no - 1] if 0 <= line_no - 1 < len(lines) else sent
        start_search = cursor_by_line.get(line_no, 0)
        col = line_text.find(sent, start_search)
        if col == -1:
            col = line_text.find(sent)
        if col == -1:
            col = 0
        cursor_by_line[line_no] = col + len(sent)
        seg_infos.extend(
            SegmentInfo(
                line_no=line_no,
                line_text=line_text,
                sentence=sent,
                sentence_col=col,
                segment=seg,
            )
            for seg in segment(sent)
        )
    return RuleContext(
        text=text,
        source_name=source_name,
        lines=lines,
        segments=seg_infos,
        repo_root=root,
    )


def _finding_from_segment(
    rule_id: str,
    tier: Tier,
    seg_info: SegmentInfo,
    local_start: int,
    local_end: int,
    matched_text: str,
    message: str,
    *,
    question: str | None = None,
) -> Finding:
    """Build a :class:`Finding` from a match local to ``seg_info.segment.text``.

    Applies global rule 3 (rationale down-weighting) uniformly.
    """
    span = seg_info.abs_span(local_start, local_end)
    weight = RATIONALE_WEIGHT if seg_info.segment.kind == "rationale" else 1.0
    return Finding.new(
        rule_id=rule_id,
        tier=tier,
        line=seg_info.line_no,
        span=span,
        matched_text=matched_text,
        message=message,
        context_line=seg_info.line_text,
        weight=weight,
        question=question,
    )


def _lexicon_rule(
    rule_id: str,
    tier: Tier,
    lexicon: frozenset[str],
    message_template: str,
    *,
    question_template: str | None = None,
    word_boundary: bool = True,
) -> RuleFn:
    """Build a rule function: one finding per lexicon hit, in every segment."""

    def _rule(ctx: RuleContext) -> list[Finding]:
        findings: list[Finding] = []
        for seg_info in ctx.segments:
            for m in find_lexicon_matches(
                seg_info.segment.text, lexicon, word_boundary=word_boundary
            ):
                matched = m.group(0)
                question = (
                    question_template.format(matched=matched)
                    if question_template
                    else None
                )
                findings.append(
                    _finding_from_segment(
                        rule_id,
                        tier,
                        seg_info,
                        m.start(),
                        m.end(),
                        matched,
                        message_template.format(matched=matched),
                        question=question,
                    )
                )
        return findings

    return _rule


# ---------------------------------------------------------------------------
# Tier 1 -- BLOCKING
# ---------------------------------------------------------------------------

_subjective_language = _lexicon_rule(
    "subjective-language", Tier.BLOCKING, SUBJECTIVE, "subjective language: '{matched}'"
)
_ambiguous_adverb_adjective = _lexicon_rule(
    "ambiguous-adverb-adjective",
    Tier.BLOCKING,
    AMBIGUOUS_ADVERBS_ADJECTIVES,
    "ambiguous adverb/adjective: '{matched}'",
)
_loophole = _lexicon_rule(
    "loophole", Tier.BLOCKING, LOOPHOLES, "loophole / escape clause: '{matched}'"
)
_open_ended_term = _lexicon_rule(
    "open-ended-term",
    Tier.BLOCKING,
    OPEN_ENDED,
    "open-ended, non-verifiable term: '{matched}'",
)
_weak_modal = _lexicon_rule(
    "weak-modal", Tier.BLOCKING, WEAK_MODALS, "weak modal: '{matched}'"
)


def _unresolved_placeholder(ctx: RuleContext) -> list[Finding]:
    """Flag TBD/TBR/TODO/FIXME/???/<placeholder>/XXX anywhere in a line."""
    findings: list[Finding] = [
        Finding.new(
            rule_id="unresolved-placeholder",
            tier=Tier.BLOCKING,
            line=line_no,
            span=(m.start(), m.end()),
            matched_text=m.group(0),
            message=f"unresolved placeholder: '{m.group(0)}'",
            context_line=line,
        )
        for line_no, line in enumerate(ctx.lines, start=1)
        for m in find_lexicon_matches(line, PLACEHOLDERS, word_boundary=True)
    ]
    return findings


_TOKEN_RE = re.compile(r"\S+")
_BARE_NUMBER_RE = re.compile(r"^\d+(\.\d+)?$")
_UNITS = frozenset(
    {
        "ms",
        "s",
        "sec",
        "secs",
        "second",
        "seconds",
        "minute",
        "minutes",
        "hour",
        "hours",
        "%",
        "mb",
        "gb",
        "kb",
        "byte",
        "bytes",
        "px",
        "item",
        "items",
        "retry",
        "retries",
        "attempt",
        "attempts",
        "day",
        "days",
        "req/s",
        "qps",
    }
)
# Preceding tokens that make a following bare number an ordinal/id/version
# reference rather than an unqualified magnitude (missing-unit's exemption).
_ORDINAL_CONTEXT_WORDS = frozenset(
    {
        "step",
        "item",
        "section",
        "phase",
        "stage",
        "version",
        "v",
        "figure",
        "fig",
        "table",
        "page",
        "#",
        # A number after "port"/"ports" is an identifier, not a magnitude
        # ("listens on port 8080" carries no missing unit).
        "port",
        "ports",
    }
)


def _strip_punct(token: str) -> str:
    return token.strip(",.;:()")


def _missing_unit(ctx: RuleContext) -> list[Finding]:
    """A bare number with no unit token within 2 following tokens.

    Skips: the first token on a line ending in "." (a markdown numbered-list
    marker, e.g. "1. Do the thing"), numbers preceded by an ordinal/id/
    version context word (:data:`_ORDINAL_CONTEXT_WORDS`), and any token that
    isn't a *bare* number in the first place -- ``REQ-033``, ``AC-001``,
    ``50ms`` and ``3.14t`` are single fused tokens that never match
    :data:`_BARE_NUMBER_RE` at all.
    """
    findings: list[Finding] = []
    for line_no, line in enumerate(ctx.lines, start=1):
        # Markdown headings are titles, not acceptance criteria — a bare
        # number there ("## 6 Detectors") states no magnitude (wave-B
        # review: Tier-1 findings on ordinary section headings).
        if line.lstrip().startswith("#"):
            continue
        tokens = list(_TOKEN_RE.finditer(line))
        for i, tok_match in enumerate(tokens):
            raw = tok_match.group(0)
            if i == 0 and raw.endswith("."):
                continue
            token = _strip_punct(raw)
            if not _BARE_NUMBER_RE.match(token):
                continue
            # Walk back through conjunction chains so the context word
            # covers every number it governs: in "ports 8080 and 9090",
            # 9090's governing context is still "ports" (likewise
            # "steps 3 and 4").
            j = i - 1
            while j >= 0:
                prev_token = _strip_punct(tokens[j].group(0)).lower()
                if prev_token in ("and", "or") or _BARE_NUMBER_RE.match(prev_token):
                    j -= 1
                    continue
                break
            else:
                prev_token = ""
            if prev_token in _ORDINAL_CONTEXT_WORDS:
                continue
            unit_found = any(
                _strip_punct(tokens[j].group(0)).lower() in _UNITS
                for j in range(i + 1, min(i + 3, len(tokens)))
            )
            if unit_found:
                continue
            findings.append(
                Finding.new(
                    rule_id="missing-unit",
                    tier=Tier.BLOCKING,
                    line=line_no,
                    span=(tok_match.start(), tok_match.end()),
                    matched_text=raw,
                    message=f"bare number '{raw}' with no unit within 2 tokens",
                    context_line=line,
                )
            )
    return findings


_AC_LINE_RE = re.compile(
    r"^-\s*\*\*(?P<ref>AC-\d{3}-[A-Za-z0-9]+-\d{2})\*\*:\s*(?P<text>.+)$"
)
_AC_ATTEMPT_RE = re.compile(r"^-\s*\*\*AC")


def _exact_duplicate(ctx: RuleContext) -> list[Finding]:
    """Identical normalized AC text appearing twice (second+ occurrence flagged)."""
    findings: list[Finding] = []
    seen: dict[str, int] = {}
    for line_no, line in enumerate(ctx.lines, start=1):
        m = _AC_LINE_RE.match(line.strip())
        if not m:
            continue
        norm = normalize(m.group("text"))
        if not norm:
            continue
        if norm in seen:
            findings.append(
                Finding.new(
                    rule_id="exact-duplicate",
                    tier=Tier.BLOCKING,
                    line=line_no,
                    span=(0, len(line)),
                    matched_text=m.group("text"),
                    message=(
                        f"duplicate of line {seen[norm]}: identical normalized AC text"
                    ),
                    context_line=line,
                )
            )
        else:
            seen[norm] = line_no
    return findings


def _template_violation(ctx: RuleContext) -> list[Finding]:
    """A line clearly attempting the AC bullet shape but not matching it."""
    findings: list[Finding] = []
    for line_no, line in enumerate(ctx.lines, start=1):
        stripped = line.strip()
        if _AC_ATTEMPT_RE.match(stripped) and not _AC_LINE_RE.match(stripped):
            findings.append(
                Finding.new(
                    rule_id="template-violation",
                    tier=Tier.BLOCKING,
                    line=line_no,
                    span=(0, len(line)),
                    matched_text=stripped,
                    message=(
                        "AC line does not match the '- **AC-NNN-...-NN**: text' shape"
                    ),
                    context_line=line,
                )
            )
    return findings


_REF_TOKEN_RE = re.compile(
    r"\b(?:REQ-\d{3}-[A-Za-z0-9]+(?:-[A-Za-z0-9]+)*|AC-\d{3}-[A-Za-z0-9]+-\d{2}"
    r"|ADR-\d{4}|SPEC-\d{3})\b"
)


def _known_ac_req_and_adr_refs(repo_root: Path) -> set[str]:
    """Known ``AC-``/``REQ-``/``ADR-`` refs, reusing the B13 pinning sources."""
    refs: set[str] = set()
    specs_dir = repo_root / "specs"
    adr_dir = repo_root / "docs" / "adr"
    refs.update(p.ref for p in AcSource(specs_dir).pinnings())
    refs.update(p.ref for p in AdrSource(adr_dir).pinnings())
    return refs


def _known_spec_ids(repo_root: Path) -> set[str]:
    """Known ``SPEC-NNN`` ids -- one per ``specs/NNN-*/`` directory."""
    specs_dir = repo_root / "specs"
    if not specs_dir.exists():
        return set()
    ids: set[str] = set()
    for child in sorted(specs_dir.iterdir()):
        if child.is_dir():
            num_match = re.match(r"^(\d{3})-", child.name)
            if num_match:
                ids.add(f"SPEC-{num_match.group(1)}")
    return ids


def _incomplete_reference(ctx: RuleContext) -> list[Finding]:
    """A REQ-/AC-/ADR-/SPEC- token that does not resolve against the repo.

    Reuses :class:`~..assumptions.sources.AcSource` /
    :class:`~..assumptions.sources.AdrSource` (the B13 id-extraction
    approach) to build the known-refs pool -- never copies their regexes.
    """
    known = _known_ac_req_and_adr_refs(ctx.repo_root) | _known_spec_ids(ctx.repo_root)
    findings: list[Finding] = []
    for line_no, line in enumerate(ctx.lines, start=1):
        for m in _REF_TOKEN_RE.finditer(line):
            token = m.group(0)
            if token not in known:
                findings.append(
                    Finding.new(
                        rule_id="incomplete-reference",
                        tier=Tier.BLOCKING,
                        line=line_no,
                        span=(m.start(), m.end()),
                        matched_text=token,
                        message=(
                            f"reference '{token}' does not resolve against the repo"
                        ),
                        context_line=line,
                    )
                )
    return findings


# ---------------------------------------------------------------------------
# Tier 2 -- ADVISORY
# ---------------------------------------------------------------------------

_superlative = _lexicon_rule(
    "superlative", Tier.ADVISORY, SUPERLATIVES, "superlative: '{matched}'"
)

# A bare number immediately following the comparative match (its own direct
# object, e.g. "more than 1 second") is not an independent baseline -- it is
# the comparative's own operand. A genuine baseline is either "than <a named
# artifact>" or a number appearing BEFORE the comparative in the segment
# (e.g. "a 20% increase"). This is what makes ADR-0029's own worked example
# testable: "the system takes more than 1 second to respond" has no baseline
# by this definition and DOES fire outside a conditional; wrapped in "if ...",
# global rule 1 exempts it regardless of baseline.
_NAMED_ARTIFACT_AFTER_THAN_RE = re.compile(r"\bthan\s+(?:the\s+)?[A-Z][\w.\-]*")


def _comparative_has_baseline(segment_text: str, match_start: int) -> bool:
    if _NAMED_ARTIFACT_AFTER_THAN_RE.search(segment_text):
        return True
    before = segment_text[:match_start]
    return bool(re.search(r"\d", before))


def _comparative_no_baseline(ctx: RuleContext) -> list[Finding]:
    """Comparative lexicon hit with no baseline; global rule 1 exempts conditionals."""
    findings: list[Finding] = []
    for seg_info in ctx.segments:
        if seg_info.segment.kind == "conditional":
            continue
        text = seg_info.segment.text
        for m in find_lexicon_matches(text, COMPARATIVES):
            if _comparative_has_baseline(text, m.start()):
                continue
            findings.append(
                _finding_from_segment(
                    "comparative-no-baseline",
                    Tier.ADVISORY,
                    seg_info,
                    m.start(),
                    m.end(),
                    m.group(0),
                    f"comparative '{m.group(0)}' with no explicit baseline "
                    "(than X / number / named artifact)",
                )
            )
    return findings


def _negative_statement(ctx: RuleContext) -> list[Finding]:
    """Negation-marker hit; global rule 1 exempts conditional segments."""
    findings: list[Finding] = []
    for seg_info in ctx.segments:
        if seg_info.segment.kind == "conditional":
            continue
        findings.extend(
            _finding_from_segment(
                "negative-statement",
                Tier.ADVISORY,
                seg_info,
                m.start(),
                m.end(),
                m.group(0),
                f"negative statement: '{m.group(0)}'",
            )
            for m in find_lexicon_matches(seg_info.segment.text, NEGATION_MARKERS)
        )
    return findings


# Small curated verb/modal set for the non-atomic-criterion approximation --
# not exhaustive by design (a starter set, per the module's own convention).
_VERB_TOKENS = frozenset(
    {
        "shall",
        "must",
        "will",
        "should",
        "may",
        "can",
        "could",
        "would",
        "validate",
        "validates",
        "store",
        "stores",
        "log",
        "logs",
        "return",
        "returns",
        "display",
        "displays",
        "calculate",
        "calculates",
        "create",
        "creates",
        "update",
        "updates",
        "delete",
        "deletes",
        "send",
        "sends",
        "receive",
        "receives",
        "process",
        "processes",
        "generate",
        "generates",
        "notify",
        "notifies",
        "escalate",
        "escalates",
        "retry",
        "retries",
        "publish",
        "publishes",
        "persist",
        "persists",
        "execute",
        "executes",
        "register",
        "registers",
        "dispatch",
        "dispatches",
        "enqueue",
        "enqueues",
        "format",
        "formats",
        "reject",
        "rejects",
        "accept",
        "accepts",
        "raise",
        "raises",
        "emit",
        "emits",
    }
)
_AND_SPLIT_RE = re.compile(r"\band\b", re.IGNORECASE)


def _has_verb_token(text: str) -> bool:
    tokens = re.findall(r"[A-Za-z']+", text.lower())
    return any(t in _VERB_TOKENS for t in tokens)


def _non_atomic_criterion(ctx: RuleContext) -> list[Finding]:
    """A main segment joining >=2 distinct verb phrases with 'and'."""
    findings: list[Finding] = []
    for seg_info in ctx.segments:
        if seg_info.segment.kind != "main":
            continue
        text = seg_info.segment.text
        if " and " not in f" {text.lower()} ":
            continue
        parts = _AND_SPLIT_RE.split(text)
        verb_bearing = [p for p in parts if _has_verb_token(p)]
        if len(verb_bearing) < 2:
            continue
        m = _AND_SPLIT_RE.search(text)
        if not m:
            continue
        findings.append(
            _finding_from_segment(
                "non-atomic-criterion",
                Tier.ADVISORY,
                seg_info,
                m.start(),
                m.end(),
                m.group(0),
                "criterion joins two verb phrases with 'and' -- consider splitting",
            )
        )
    return findings


_OR_SPLIT_RE = re.compile(r"\bor\b", re.IGNORECASE)


def _coordination_ambiguity(ctx: RuleContext) -> list[Finding]:
    """A sentence containing both 'and' and 'or' with no explicit grouping."""
    findings: list[Finding] = []
    for line_no, sent in sentences(ctx.text):
        and_m = _AND_SPLIT_RE.search(sent)
        or_m = _OR_SPLIT_RE.search(sent)
        if not (and_m and or_m):
            continue
        line_text = (
            ctx.lines[line_no - 1] if 0 <= line_no - 1 < len(ctx.lines) else sent
        )
        winner = or_m if or_m.start() > and_m.start() else and_m
        findings.append(
            Finding.new(
                rule_id="coordination-ambiguity",
                tier=Tier.ADVISORY,
                line=line_no,
                span=(winner.start(), winner.end()),
                matched_text=winner.group(0),
                message="sentence combines 'and' and 'or' with no explicit grouping",
                context_line=line_text,
            )
        )
    return findings


_solution_language = _lexicon_rule(
    "solution-language",
    Tier.ADVISORY,
    SOLUTION_LANGUAGE_NOUNS,
    "solution/implementation language: '{matched}'",
)

# ---------------------------------------------------------------------------
# Tier 3 -- QUESTION-generating only
# ---------------------------------------------------------------------------

_AUX_AFTER_PRONOUN = frozenset(
    {
        "is",
        "are",
        "was",
        "were",
        "has",
        "have",
        "had",
        "will",
        "would",
        "should",
        "must",
        "can",
        "may",
        "might",
        "does",
        "do",
        "shall",
        "refers",
        "means",
    }
)
_WORD_RE = re.compile(r"[A-Za-z']+")


def _vague_pronoun(ctx: RuleContext) -> list[Finding]:
    """A vague pronoun in subject position (first token of a segment)."""
    findings: list[Finding] = []
    for seg_info in ctx.segments:
        text = seg_info.segment.text
        tokens = _WORD_RE.findall(text)
        if len(tokens) < 2:
            continue
        first, second = tokens[0].lower(), tokens[1].lower()
        if first not in VAGUE_PRONOUNS or second not in _AUX_AFTER_PRONOUN:
            continue
        m = _WORD_RE.search(text)
        if not m:
            continue
        findings.append(
            _finding_from_segment(
                "vague-pronoun",
                Tier.QUESTION,
                seg_info,
                m.start(),
                m.end(),
                m.group(0),
                f"vague pronoun '{m.group(0)}' in subject position",
                question=f"what does '{m.group(0)}' refer to here?",
            )
        )
    return findings


_QUANTIFIERS = frozenset({"all", "each", "every", "any"})


def _quantifier_scope(ctx: RuleContext) -> list[Finding]:
    """A quantifier and a negation marker co-occurring in one segment."""
    findings: list[Finding] = []
    for seg_info in ctx.segments:
        text = seg_info.segment.text
        quant_matches = find_lexicon_matches(text, _QUANTIFIERS)
        if not quant_matches:
            continue
        neg_matches = find_lexicon_matches(text, NEGATION_MARKERS)
        if not neg_matches:
            continue
        q, n = quant_matches[0], neg_matches[0]
        start, end = min(q.start(), n.start()), max(q.end(), n.end())
        findings.append(
            _finding_from_segment(
                "quantifier-scope",
                Tier.QUESTION,
                seg_info,
                start,
                end,
                text[start:end],
                f"quantifier '{q.group(0)}' + negation '{n.group(0)}' "
                "-- scope is ambiguous",
                question=(
                    f"does '{q.group(0)} ... {n.group(0)}' mean "
                    "'none' or 'not every one'?"
                ),
            )
        )
    return findings


_OPERATOR_WORDS = frozenset({"only", "also", "even"})
_COMMON_VERBS_FOR_OPERATOR = frozenset(
    {
        "log",
        "validate",
        "allow",
        "block",
        "return",
        "process",
        "execute",
        "display",
        "store",
        "delete",
        "send",
        "receive",
        "update",
        "check",
        "use",
        "support",
        "enable",
        "create",
        "notify",
        "escalate",
        "retry",
        "publish",
        "persist",
        "register",
        "dispatch",
        "enqueue",
        "format",
        "accept",
        "reject",
        "raise",
        "emit",
    }
)
_NEXT_TOKEN_RE = re.compile(r"[A-Za-z0-9']+")


def _operator_placement(ctx: RuleContext) -> list[Finding]:
    """only/also/even not immediately preceding a verb or a number."""
    findings: list[Finding] = []
    for seg_info in ctx.segments:
        text = seg_info.segment.text
        for m in find_lexicon_matches(text, _OPERATOR_WORDS):
            after = text[m.end() :].lstrip()
            next_m = _NEXT_TOKEN_RE.match(after)
            next_tok = next_m.group(0).lower() if next_m else ""
            if next_tok in _COMMON_VERBS_FOR_OPERATOR or next_tok.isdigit():
                continue
            findings.append(
                _finding_from_segment(
                    "operator-placement",
                    Tier.QUESTION,
                    seg_info,
                    m.start(),
                    m.end(),
                    m.group(0),
                    f"operator '{m.group(0)}' not immediately before a verb/number",
                    question=(
                        f"does '{m.group(0)}' modify the following word "
                        "or the whole clause?"
                    ),
                )
            )
    return findings


def _nearby_number(
    text: str, match_start: int, match_end: int, window_tokens: int = 3
) -> str | None:
    """A digit-bearing token within ``window_tokens`` of the match, if any."""
    tokens = list(_TOKEN_RE.finditer(text))
    idx = None
    for i, tok in enumerate(tokens):
        if (
            tok.start() <= match_start < tok.end()
            or match_start <= tok.start() < match_end
        ):
            idx = i
            break
    if idx is None:
        return None
    lo, hi = max(0, idx - window_tokens), min(len(tokens), idx + window_tokens + 1)
    for i in range(lo, hi):
        digit_match = re.search(r"\d+", tokens[i].group(0))
        if digit_match:
            return digit_match.group(0)
    return None


def _boundary_word(ctx: RuleContext) -> list[Finding]:
    """A boundary word within a few tokens of a number -- interval ambiguity.

    Scoped to number-adjacent hits (the classic "up to 12" case ADR-0029
    illustrates); a standalone deadline-word ambiguity with no accompanying
    number (e.g. "by Friday" vs. "until Friday") is a real, separate category
    left for a future extension -- documented here, not built, to keep this
    starter rule's false-positive rate low.
    """
    findings: list[Finding] = []
    for seg_info in ctx.segments:
        text = seg_info.segment.text
        for m in find_lexicon_matches(text, BOUNDARY_WORDS):
            number = _nearby_number(text, m.start(), m.end())
            if number is None:
                continue
            findings.append(
                _finding_from_segment(
                    "boundary-word",
                    Tier.QUESTION,
                    seg_info,
                    m.start(),
                    m.end(),
                    m.group(0),
                    f"boundary word '{m.group(0)}' near a number -- open or closed?",
                    question=f"does '{m.group(0)} {number}' include {number}?",
                )
            )
    return findings


_temporal_nonspecific = _lexicon_rule(
    "temporal-nonspecific",
    Tier.QUESTION,
    TEMPORAL_NONSPECIFIC,
    "non-specific temporal word: '{matched}'",
    question_template="what does '{matched}' mean, numerically?",
)

_deictic_reference = _lexicon_rule(
    "deictic-reference",
    Tier.QUESTION,
    DEICTIC,
    "deictic reference: '{matched}'",
    question_template="what does '{matched}' refer to?",
)

_SPECIFIER_FOLLOWERS_RE = re.compile(
    r"^(returned by|specified in|described in|provided by|captured in|defined in"
    r"|associated with|produced by|from the|of the|that|which)\b",
    re.IGNORECASE,
)


def _underspecified_head_noun(ctx: RuleContext) -> list[Finding]:
    """'the data'/'the information'/'the content'/'the system' with no defining
    qualifier immediately following (heuristic: a following relative/prepositional
    clause like "returned by X" or "that Y" resolves it; a bare occurrence does not).
    """
    findings: list[Finding] = []
    for seg_info in ctx.segments:
        text = seg_info.segment.text
        for m in find_lexicon_matches(text, UNDERSPECIFIED_HEAD_NOUNS):
            after = text[m.end() :].lstrip()
            if _SPECIFIER_FOLLOWERS_RE.match(after):
                continue
            findings.append(
                _finding_from_segment(
                    "underspecified-head-noun",
                    Tier.QUESTION,
                    seg_info,
                    m.start(),
                    m.end(),
                    m.group(0),
                    f"under-specified head noun '{m.group(0)}' "
                    "with no earlier definition",
                    question=(
                        f"what does '{m.group(0)}' refer to -- is it defined earlier?"
                    ),
                )
            )
    return findings


# Shared provenance strings for the registry builders below.
_FEMMER = "ISO/IEC/IEEE 29148 §5.2.7 / Femmer et al. JSS 2017"
_BKK = "Berry, Kamsties & Krieger (2003) §5.2-5.7"
_ISO = "ISO/IEC/IEEE 29148 §5.2.7"
_STRUCTURAL = "SPEC-033 REQ-033-C3 / ADR-0029 (structural)"


def _tier1_dictionary_registry() -> dict[str, tuple[RuleMeta, RuleFn]]:
    """The Femmer-family dictionary rules (measured precision) + ISO placeholders."""
    return {
        "subjective-language": (
            RuleMeta("subjective-language", Tier.BLOCKING, _FEMMER, 0.96),
            _subjective_language,
        ),
        "ambiguous-adverb-adjective": (
            RuleMeta("ambiguous-adverb-adjective", Tier.BLOCKING, _FEMMER, 0.81),
            _ambiguous_adverb_adjective,
        ),
        "loophole": (RuleMeta("loophole", Tier.BLOCKING, _FEMMER, 0.72), _loophole),
        "open-ended-term": (
            RuleMeta("open-ended-term", Tier.BLOCKING, _FEMMER, 0.70),
            _open_ended_term,
        ),
        "weak-modal": (RuleMeta("weak-modal", Tier.BLOCKING, _ISO, None), _weak_modal),
        "unresolved-placeholder": (
            RuleMeta("unresolved-placeholder", Tier.BLOCKING, _ISO, None),
            _unresolved_placeholder,
        ),
    }


def _tier1_structural_registry() -> dict[str, tuple[RuleMeta, RuleFn]]:
    """The structural (non-dictionary-family) BLOCKING rules."""
    return {
        "missing-unit": (
            RuleMeta("missing-unit", Tier.BLOCKING, _STRUCTURAL, None),
            _missing_unit,
        ),
        "exact-duplicate": (
            RuleMeta("exact-duplicate", Tier.BLOCKING, _STRUCTURAL, None),
            _exact_duplicate,
        ),
        "template-violation": (
            RuleMeta("template-violation", Tier.BLOCKING, _STRUCTURAL, None),
            _template_violation,
        ),
        "incomplete-reference": (
            RuleMeta(
                "incomplete-reference",
                Tier.BLOCKING,
                f"{_STRUCTURAL}, reuses B13 sources.py",
                None,
            ),
            _incomplete_reference,
        ),
    }


def _tier1_registry() -> dict[str, tuple[RuleMeta, RuleFn]]:
    return {**_tier1_dictionary_registry(), **_tier1_structural_registry()}


def _tier2_registry() -> dict[str, tuple[RuleMeta, RuleFn]]:
    return {
        "superlative": (
            RuleMeta("superlative", Tier.ADVISORY, _FEMMER, 0.49),
            _superlative,
        ),
        "comparative-no-baseline": (
            RuleMeta("comparative-no-baseline", Tier.ADVISORY, _FEMMER, 0.48),
            _comparative_no_baseline,
        ),
        "negative-statement": (
            RuleMeta("negative-statement", Tier.ADVISORY, _FEMMER, 0.33),
            _negative_statement,
        ),
        "non-atomic-criterion": (
            RuleMeta("non-atomic-criterion", Tier.ADVISORY, _ISO, None),
            _non_atomic_criterion,
        ),
        "coordination-ambiguity": (
            RuleMeta("coordination-ambiguity", Tier.ADVISORY, _BKK, None),
            _coordination_ambiguity,
        ),
        "solution-language": (
            RuleMeta("solution-language", Tier.ADVISORY, _ISO, None),
            _solution_language,
        ),
    }


def _tier3_registry() -> dict[str, tuple[RuleMeta, RuleFn]]:
    return {
        "vague-pronoun": (
            RuleMeta("vague-pronoun", Tier.QUESTION, _FEMMER, 0.26),
            _vague_pronoun,
        ),
        "quantifier-scope": (
            RuleMeta("quantifier-scope", Tier.QUESTION, _BKK, None),
            _quantifier_scope,
        ),
        "operator-placement": (
            RuleMeta("operator-placement", Tier.QUESTION, _BKK, None),
            _operator_placement,
        ),
        "boundary-word": (
            RuleMeta("boundary-word", Tier.QUESTION, _BKK, None),
            _boundary_word,
        ),
        "temporal-nonspecific": (
            RuleMeta("temporal-nonspecific", Tier.QUESTION, _BKK, None),
            _temporal_nonspecific,
        ),
        "underspecified-head-noun": (
            RuleMeta("underspecified-head-noun", Tier.QUESTION, _BKK, None),
            _underspecified_head_noun,
        ),
        "deictic-reference": (
            RuleMeta("deictic-reference", Tier.QUESTION, _BKK, None),
            _deictic_reference,
        ),
    }


@traces(spec="033", reqs=["REQ-033-C3"])
def _build_registry() -> dict[str, tuple[RuleMeta, RuleFn]]:
    """Build the rule registry (see :data:`ALL_RULES`) from its three tiers."""
    return {**_tier1_registry(), **_tier2_registry(), **_tier3_registry()}


ALL_RULES: dict[str, tuple[RuleMeta, RuleFn]] = _build_registry()