orchestrata_ultimate.factory.anchors.intake

Reference › orchestrata_ultimate › factory › anchors › intake

intake feeds new anchors into the AnchorRegistry. import_dor_candidates reads the JSON export written by anchors_export in the DoR procedure (row B23) and proposes one PROPOSED EXAMPLE anchor per candidate, since steps 2 and 3 of that procedure already produce the example and boundary anchors that ADR-0012 requires. propose_oracle is the amendment-path helper that ADR-0016 mandates: every amendment carries a differential oracle describing prior, observed behaviour, and this function proposes one for a requirement. Both only propose; ratification stays a human act. The scripts.factory_anchors CLI imports the module and tests/factory/test_anchor_intake.py covers it.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

API

Functions

Source

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

"""B23 bridge -- DoR-derived anchor candidates + the amendment oracle helper.

:func:`import_dor_candidates` consumes the JSON that :mod:`orchestrata_
ultimate.factory.dor_procedure.anchors_export` writes (one
:class:`~orchestrata_ultimate.factory.dor_procedure.anchors_export.
AnchorCandidate` dict per step-2 equivalence class) and proposes an EXAMPLE
anchor per candidate -- steps 2-3 of the DoR construction procedure (B23)
double as the example/boundary anchors ADR-0012 requires, so this is a
projection of that work, not new authoring.

:func:`propose_oracle` is the amendment-path helper ADR-0016 mandates: every
amendment must carry a differential oracle (prior, observed behaviour), and
this is how one gets proposed into the registry.
"""

from __future__ import annotations

import json
from pathlib import Path

from orchestrata_ultimate.factory.anchors.models import Anchor, AnchorClass
from orchestrata_ultimate.factory.anchors.store import AnchorRegistry


def import_dor_candidates(export_path: Path, registry: AnchorRegistry) -> list[Anchor]:
    """Propose one PROPOSED EXAMPLE anchor per B23 anchor-candidate export.

    ``export_path`` is the JSON file written by ``factory_dor_procedure.py
    export-anchors`` (a list of :func:`dataclasses.asdict`-serialized
    :class:`~orchestrata_ultimate.factory.dor_procedure.anchors_export.
    AnchorCandidate`). ``statement`` combines the candidate's step-2 ``given``
    and step-3 ``expect`` text; ``boundary`` is carried through unchanged (so
    boundary candidates keep pinning their boundary); ``source`` is
    ``"dor:<requirement_id>/step2-3"``.

    Idempotent: a candidate whose deterministic ``anchor_id`` already exists
    in ``registry`` is skipped, so re-running this after the underlying B23
    artifact gains further answers proposes only what changed.

    Returns:
        The list of newly proposed anchors (already-existing ids omitted).
    """
    with open(Path(export_path), encoding="utf-8") as f:
        raw_candidates: list[dict[str, str]] = json.load(f)

    proposed: list[Anchor] = []
    for item in raw_candidates:
        statement = f"given: {item['given']} | expect: {item['expect']}"
        anchor = Anchor.new(
            requirement_id=item["requirement_id"],
            klass=AnchorClass.EXAMPLE,
            statement=statement,
            boundary=item.get("boundary", ""),
            source=f"dor:{item['requirement_id']}/step2-3",
        )
        if registry.get(anchor.anchor_id) is not None:
            continue
        proposed.append(registry.propose(anchor))
    return proposed


def propose_oracle(
    registry: AnchorRegistry,
    requirement_id: str,
    observed_ref: str,
    statement: str,
) -> Anchor:
    """Propose a DIFFERENTIAL_ORACLE anchor -- the amendment-path helper.

    ADR-0016: every amendment must carry a differential oracle (the prior,
    observed behaviour), since one is always available for a change to
    existing code. ``observed_ref`` identifies that observed behaviour (a
    test run, a recorded transcript, a prior release tag -- never authored
    text) and must be non-empty; it becomes the anchor's ``source`` as
    ``"observed:<observed_ref>"``.

    Raises:
        ValueError: If ``observed_ref`` is empty.
    """
    if not observed_ref:
        raise ValueError("observed_ref must not be empty")
    anchor = Anchor.new(
        requirement_id=requirement_id,
        klass=AnchorClass.DIFFERENTIAL_ORACLE,
        statement=statement,
        source=f"observed:{observed_ref}",
    )
    return registry.propose(anchor)