orchestrata_ultimate.factory.anchors.gate

Reference › orchestrata_ultimate › factory › anchors › gate

gate implements the anchor control of the fast path (SPEC-033 row B15b, ADR-0023). A change qualifies for the fast path only if the anchors of every touched requirement stay satisfied and unchanged; any change that edits an anchor goes to the full decision phase, because editing an anchor needs human ratification (ADR-0013). fast_path_gate is the single function that decides this. It gathers the RATIFIED anchors for the given requirement ids from the AnchorRegistry (_ratified_anchors_for, deduplicated by id), evaluates them with evaluate_all from evaluation, and returns an AnchorGateVerdict with fast_path_eligible, the per-anchor results, a blocking list and the edited_anchor_ids. An INCONCLUSIVE anchor blocks exactly like an UNSATISFIED one, and an anchor named in changed_anchor_ids blocks too. The function only reads the registry; it never edits or ratifies. The fast-path classifier and the scripts.factory_anchors and scripts.factory_fastpath CLIs import it; tests are in tests/factory/test_anchor_gate.py.

At a glance

Class diagram

Dependency diagram

API

Classes

AnchorGateVerdict (dataclass)

Decorators: dataclass(frozen=True) · Line 36

The fast-path decision for one gate evaluation.

Functions

Source

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

"""The ADR-0023 fast-path control (B15b).

ADR-0023, verbatim: "A change qualifies for the fast path only if, in
addition to the structural within-contract proof: the anchors of every
touched acceptance criterion remain **satisfied**, and those anchors remain
**unchanged**. Any change that requires editing an anchor is by definition
not fast-path — it goes to the full decision phase, and (per ADR-0013) the
anchor edit itself requires human ratification."

:func:`fast_path_gate` is the single seam that decides this. It NEVER edits
anchors and NEVER ratifies — it only reads the committed
:class:`~orchestrata_ultimate.factory.anchors.store.AnchorRegistry` and
compares already-observed evidence (:class:`~.evaluation.AnchorEvidence`)
against it. An INCONCLUSIVE anchor blocks the fast path exactly like an
UNSATISFIED one — a fast path may not proceed on an unproven pin.
"""

from __future__ import annotations

from collections.abc import Sequence
from dataclasses import dataclass

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.factory.anchors.evaluation import (
    AnchorEvaluator,
    AnchorEvidence,
    AnchorResult,
    Satisfaction,
    evaluate_all,
)
from orchestrata_ultimate.factory.anchors.models import Anchor, AnchorStatus
from orchestrata_ultimate.factory.anchors.store import AnchorRegistry


@dataclass(frozen=True)
class AnchorGateVerdict:
    """The fast-path decision for one gate evaluation.

    ``blocking`` lists the ids of every anchor that evaluated UNSATISFIED
    or INCONCLUSIVE (exactly why the gate is blocked, if it is).
    ``edited_anchor_ids`` is the (sorted) ``changed_anchor_ids`` the caller
    passed in — any non-empty set disqualifies the fast path regardless of
    ``blocking``, per ADR-0023.
    """

    fast_path_eligible: bool
    results: list[AnchorResult]
    blocking: list[str]
    edited_anchor_ids: list[str]


def _ratified_anchors_for(
    requirement_ids: list[str], registry: AnchorRegistry
) -> list[Anchor]:
    """The RATIFIED anchors touched by ``requirement_ids``, deduplicated by id."""
    anchors: list[Anchor] = []
    seen_ids: set[str] = set()
    for requirement_id in requirement_ids:
        for anchor in registry.anchors(
            requirement_id=requirement_id, status=AnchorStatus.RATIFIED
        ):
            if anchor.anchor_id not in seen_ids:
                seen_ids.add(anchor.anchor_id)
                anchors.append(anchor)
    return anchors


@traces(spec="033", reqs=["REQ-033-C1"])
def fast_path_gate(
    requirement_ids: list[str],
    registry: AnchorRegistry,
    evidence: AnchorEvidence,
    evaluators: Sequence[AnchorEvaluator],
    *,
    changed_anchor_ids: set[str],
) -> AnchorGateVerdict:
    """Decide fast-path eligibility per ADR-0023.

    ADR-0023, verbatim: "A change qualifies for the fast path only if, in
    addition to the structural within-contract proof: the anchors of every
    touched acceptance criterion remain satisfied, and those anchors remain
    unchanged. Any change that requires editing an anchor is by definition
    not fast-path."

    Eligible iff BOTH:

    - every RATIFIED anchor for ``requirement_ids`` evaluates SATISFIED
      (an INCONCLUSIVE anchor blocks the fast path exactly like an
      UNSATISFIED one — a fast path may not proceed on an unproven pin), and
    - ``changed_anchor_ids`` is empty (any anchor edit means the edit
      itself is a loosening per ADR-0013, which requires human
      ratification and the full decision phase — never the fast path).

    This function reads the registry only — it never proposes, ratifies,
    or supersedes an anchor.
    """
    anchors = _ratified_anchors_for(requirement_ids, registry)
    results = evaluate_all(anchors, evidence, evaluators)
    blocking = [
        result.anchor_id
        for result in results
        if result.satisfaction != Satisfaction.SATISFIED
    ]
    edited_anchor_ids = sorted(changed_anchor_ids)
    fast_path_eligible = not blocking and not changed_anchor_ids

    return AnchorGateVerdict(
        fast_path_eligible=fast_path_eligible,
        results=results,
        blocking=blocking,
        edited_anchor_ids=edited_anchor_ids,
    )