orchestrata_ultimate.factory.assumptions.cascade

Reference › orchestrata_ultimate › factory › assumptions › cascade

cascade is the decision core of the assumption subsystem of the factory (SPEC-033 row B13, requirement REQ-033-C1, ADR-0024). It decides whether an assumption that an executing model has enumerated is backed by an existing artifact. The command-line script scripts.factory_assumptions is its importer, and tests/factory/test_cascade.py exercises it.

ArbiterCascade is built from a list of PinningSource objects, a knowledge-base seam (KBProtocol) and an optional adjudicator. resolve takes one Assumption through three stages and returns a CascadeOutcome. Stage 0 fingerprints the text and asks the knowledge base for a prior adjudication: a MATCHED hit returns PINNED immediately, an UNPINNED hit returns UNPINNED. Stage 1 calls prefilter to collect lexical candidates; these are evidence only and never set a status on their own. Stage 2 hands the text, the requirement id and the candidates to the independent adjudicator from adjudicator.

The design is fail-closed. If the adjudicator's identity equals the assumption's author_identity, resolve raises IndependenceViolation. With no adjudicator configured, the outcome is NEEDS_HUMAN. A MATCHED verdict is accepted only when its pinning_ref is one of the stage-1 candidates, and an UNPINNED verdict is accepted only when no candidate scored at or above STRONG_CANDIDATE_SCORE; any disagreement between the stages falls to NEEDS_HUMAN through _needs_human. resolve never reaches stage 3. resolve_all resolves every PENDING assumption in an AssumptionLedger and persists each result. human_adjudicate is stage 3: it records a human Ruling on the ledger and promotes it to the knowledge base as an Adjudication.

At a glance

Class diagram

Dependency diagram

API

Classes

ArbiterCascade

Decorators: traces(spec='033', reqs=['REQ-033-C1']) · Line 79

Runs the three-stage arbiter cascade over ledger assumptions.

Functions

Source

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

"""``ArbiterCascade`` — the SPEC-033 three-stage arbiter (B13, ADR-0024).

Deterministic stage-combination rules (this exact table):

1. **Stage 0 — KB.** ``kb.lookup(fingerprint, text)`` hits with ruling
   ``MATCHED`` -> ``PINNED``, pinning = the adjudication's pinning if set
   else ``Pinning(ADJUDICATION, adj.id, ...)``, stage=0. Hits with ruling
   ``UNPINNED`` -> ``UNPINNED``, stage=0 (a known-unpinned assumption is a
   known blocking question — no re-adjudication). This is AC-033-C1-04.
2. **Stage 1.** Compute candidates (informational; attached to the outcome).
3. **Independence guard.** If an adjudicator is configured and
   ``adjudicator.identity == assumption.author_identity``, raise
   ``IndependenceViolation`` — the enumerator may not adjudicate its own
   enumeration.
4. **No adjudicator configured** -> ``NEEDS_HUMAN``, stage=None (fail-closed;
   never auto-pin from stage 1 alone).
5. **Stage 2 verdict combination:**
   - ``MATCHED`` and ``pinning_ref`` is among the stage-1 candidate refs ->
     ``PINNED``, stage=2, pinning = that candidate's.
   - ``MATCHED`` but the ref is not among the candidates (or resolves to no
     known pinning) -> ``NEEDS_HUMAN`` (stages disagree).
   - ``UNPINNED`` and no stage-1 candidate scores >= ``STRONG_CANDIDATE_SCORE``
     -> ``UNPINNED`` (blocking question; carries the verdict's questions).
   - ``UNPINNED`` but a strong stage-1 candidate exists -> ``NEEDS_HUMAN``
     (stages disagree).
   - ``NEEDS_HUMAN`` -> ``NEEDS_HUMAN``.
6. **Stage 3** (:func:`human_adjudicate`, separate + synchronous) — updates
   the assumption (``PINNED`` on ``MATCHED`` / ``UNPINNED`` otherwise,
   stage=3) and **promotes** the adjudication into the KB with
   ``adjudicator=f"human:{by}"`` and provenance.

Every ``NEEDS_HUMAN`` outcome carries ``stage=None`` — no stage resolved it;
it is still pending stage 3. Only ``PINNED``/``UNPINNED`` outcomes carry a
concrete stage (0, 2, or — after :func:`human_adjudicate` — 3).
"""

from __future__ import annotations

import dataclasses
from datetime import UTC, datetime

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.factory.assumptions.adjudicator import (
    AdjudicatorProtocol,
    IndependenceViolation,
)
from orchestrata_ultimate.factory.assumptions.ledger import AssumptionLedger, KBProtocol
from orchestrata_ultimate.factory.assumptions.models import (
    Adjudication,
    Assumption,
    AssumptionStatus,
    CascadeOutcome,
    Pinning,
    PinningKind,
    Ruling,
    fingerprint,
)
from orchestrata_ultimate.factory.assumptions.prefilter import (
    STRONG_CANDIDATE_SCORE,
    prefilter,
)
from orchestrata_ultimate.factory.assumptions.sources import PinningSource


def _needs_human(
    assumption_id: str, questions: list[str], candidates: list
) -> CascadeOutcome:
    return CascadeOutcome(
        assumption_id=assumption_id,
        status=AssumptionStatus.NEEDS_HUMAN,
        pinning=None,
        stage=None,
        questions=questions,
        candidates=candidates,
    )


@traces(spec="033", reqs=["REQ-033-C1"])
class ArbiterCascade:
    """Runs the three-stage arbiter cascade over ledger assumptions."""

    def __init__(
        self,
        sources: list[PinningSource],
        kb: KBProtocol,
        adjudicator: AdjudicatorProtocol | None,
    ):
        self.sources = sources
        self.kb = kb
        self.adjudicator = adjudicator

    async def resolve(self, assumption: Assumption) -> CascadeOutcome:
        """Resolve one assumption through stages 0-2 (never stage 3)."""
        fp = fingerprint(assumption.text)

        # Stage 0 — KB.
        kb_hit = self.kb.lookup(fp, assumption.text)
        if kb_hit is not None and kb_hit.ruling == Ruling.MATCHED:
            pinning = kb_hit.pinning or Pinning(
                kind=PinningKind.ADJUDICATION,
                ref=kb_hit.id,
                excerpt=kb_hit.assumption_text,
            )
            return CascadeOutcome(
                assumption_id=assumption.id,
                status=AssumptionStatus.PINNED,
                pinning=pinning,
                stage=0,
                questions=[],
                candidates=[],
            )
        if kb_hit is not None and kb_hit.ruling == Ruling.UNPINNED:
            return CascadeOutcome(
                assumption_id=assumption.id,
                status=AssumptionStatus.UNPINNED,
                pinning=None,
                stage=0,
                questions=[],
                candidates=[],
            )

        # Stage 1 — deterministic prefilter (informational only).
        candidates = prefilter(assumption.text, self.sources)

        # Independence guard.
        if (
            self.adjudicator is not None
            and self.adjudicator.identity == assumption.author_identity
        ):
            raise IndependenceViolation(
                f"adjudicator {self.adjudicator.identity!r} may not adjudicate "
                f"its own enumeration (assumption {assumption.id!r})"
            )

        # No adjudicator configured -> fail-closed, never auto-pin from
        # stage 1 alone.
        if self.adjudicator is None:
            return _needs_human(assumption.id, [], candidates)

        # Stage 2 — independent-tier adjudication.
        verdict = await self.adjudicator.adjudicate(
            assumption.text, assumption.requirement_id, candidates
        )
        candidate_pinnings = {c.pinning.ref: c.pinning for c in candidates}
        strong_candidate_exists = any(
            c.score >= STRONG_CANDIDATE_SCORE for c in candidates
        )

        if verdict.ruling == Ruling.MATCHED:
            if (
                verdict.pinning_ref is not None
                and verdict.pinning_ref in candidate_pinnings
            ):
                return CascadeOutcome(
                    assumption_id=assumption.id,
                    status=AssumptionStatus.PINNED,
                    pinning=candidate_pinnings[verdict.pinning_ref],
                    stage=2,
                    questions=[],
                    candidates=candidates,
                )
            return _needs_human(assumption.id, verdict.questions, candidates)

        if verdict.ruling == Ruling.UNPINNED:
            if not strong_candidate_exists:
                return CascadeOutcome(
                    assumption_id=assumption.id,
                    status=AssumptionStatus.UNPINNED,
                    pinning=None,
                    stage=2,
                    questions=verdict.questions,
                    candidates=candidates,
                )
            return _needs_human(assumption.id, verdict.questions, candidates)

        # verdict.ruling == Ruling.NEEDS_HUMAN
        return _needs_human(assumption.id, verdict.questions, candidates)

    async def resolve_all(self, ledger: AssumptionLedger) -> list[CascadeOutcome]:
        """Resolve every ``PENDING`` assumption in ``ledger`` and persist it."""
        outcomes: list[CascadeOutcome] = []
        for assumption in ledger.assumptions():
            if assumption.status != AssumptionStatus.PENDING:
                continue
            outcome = await self.resolve(assumption)
            updated = dataclasses.replace(
                assumption,
                status=outcome.status,
                pinning=outcome.pinning,
                stage_resolved=outcome.stage,
                questions=outcome.questions,
            )
            ledger.update(updated)
            outcomes.append(outcome)
        return outcomes


@traces(spec="033", reqs=["REQ-033-C1"])
def human_adjudicate(
    ledger: AssumptionLedger,
    kb: KBProtocol,
    assumption_id: str,
    ruling: Ruling,
    pinning: Pinning | None,
    by: str,
    requirement_id: str | None = None,
) -> Adjudication:
    """Stage 3 — human adjudication. Updates the ledger and promotes to the KB.

    Args:
        ledger: The assumption ledger to update.
        kb: The Project KB the ruling is promoted into.
        assumption_id: Id of the assumption being adjudicated.
        ruling: ``MATCHED`` or ``UNPINNED`` — human adjudication is terminal,
            it never produces ``NEEDS_HUMAN``.
        pinning: Required when ``ruling`` is ``MATCHED``; ignored otherwise.
        by: Human identity (e.g. ``"mohamed"``) — must not be empty.
        requirement_id: Requirement to record on the promoted adjudication;
            defaults to the assumption's own ``requirement_id``.

    Raises:
        ValueError: If ``by`` is empty, ``ruling`` is not promotable, the
            assumption is unknown, or ``ruling`` is ``MATCHED`` without a
            ``pinning``.
    """
    if not by:
        raise ValueError("by must not be empty")
    if ruling not in (Ruling.MATCHED, Ruling.UNPINNED):
        raise ValueError(f"ruling must be MATCHED or UNPINNED, got {ruling!r}")

    assumption = ledger.get(assumption_id)
    if assumption is None:
        raise ValueError(f"unknown assumption id {assumption_id!r}")

    if ruling == Ruling.MATCHED:
        if pinning is None:
            raise ValueError("pinning is required when ruling is MATCHED")
        new_status = AssumptionStatus.PINNED
        resolved_pinning: Pinning | None = pinning
    else:
        new_status = AssumptionStatus.UNPINNED
        resolved_pinning = None

    updated = dataclasses.replace(
        assumption,
        status=new_status,
        pinning=resolved_pinning,
        stage_resolved=3,
        questions=[],
    )
    ledger.update(updated)

    adjudication = Adjudication(
        id="",
        fingerprint=fingerprint(assumption.text),
        assumption_text=assumption.text,
        ruling=ruling,
        pinning=resolved_pinning,
        adjudicator=f"human:{by}",
        requirement_id=(
            requirement_id if requirement_id is not None else assumption.requirement_id
        ),
        adjudicated_at=datetime.now(UTC).isoformat(),
        superseded_by=None,
    )
    return kb.promote(adjudication)