orchestrata_ultimate.factory.assumptions.adjudicator

Reference › orchestrata_ultimate › factory › assumptions › adjudicator

adjudicator implements stage 2 of the arbiter cascade (SPEC-033 row B13, ADR-0024): an independent tier that reads an assumption and proposes a ruling. It is imported by cascade and the scripts.factory_assumptions CLI, and tested in tests/factory/test_adjudicator.py.

AdjudicatorProtocol defines the seam: an identity property and an async adjudicate method that returns an AdjudicatorVerdict (a Ruling, an optional pinning_ref, a list of questions and the raw reply). LlmAdjudicator is the concrete tier. It is configured with a profile name and calls call_llm from the llm client. _build_prompt shows the model only the assumption text, the requirement id and the stage-1 candidates, never the author's rationale. The system prompt demands strict JSON and says the tier produces questions and candidate pinnings for a human, not verdicts.

_parse_verdict decodes the reply and fails closed: any unparseable output becomes NEEDS_HUMAN. IndependenceViolation is declared here, but the check that the adjudicator is not the author runs in cascade.

At a glance

Class diagram

Dependency diagram

API

Classes

IndependenceViolation

Bases: Exception · Line 38

Raised when the configured adjudicator is also the assumption's author.

AdjudicatorVerdict (dataclass)

Decorators: dataclass(frozen=True) · Line 43

The independent tier's output: a ruling plus candidate pinning ref.

AdjudicatorProtocol (protocol)

Bases: Protocol · Line 52

A stage-2 independent adjudication tier.

LlmAdjudicator

Line 126

An AdjudicatorProtocol backed by llm.client.call_llm.

Functions

Source

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

"""Stage 2 — independent-tier adjudication for the arbiter cascade (B13).

ADR-0024's independence guarantee: the tier that adjudicates an assumption
must not be the tier that authored it (enforced in ``cascade.py``, not
here), and it is shown **only** the assumption text, its requirement id, and
the stage-1 candidate pinnings — never the author's rationale. Its job is to
produce *questions and candidate pinnings*, never conclusions: output is
strict JSON, and any unparseable response fails closed to ``needs_human``.
"""

from __future__ import annotations

import json
import re
from dataclasses import dataclass
from typing import Protocol

from orchestrata_ultimate.factory.assumptions.models import Candidate, Ruling
from orchestrata_ultimate.llm.client import call_llm

_SYSTEM_PROMPT = (
    "You are an independent adjudication tier in a governed build factory. "
    "You did not author the assumption you are given, and you are shown "
    "only its text, its requirement id, and candidate pinning artifacts — "
    "never the author's rationale. You do not issue verdicts: you produce "
    "questions and candidate pinnings for a human to judge. "
    "Respond with strict JSON ONLY, no prose, no markdown fence, matching "
    'exactly: {"ruling": "matched"|"unpinned"|"needs_human", '
    '"pinning_ref": <string or null>, "questions": [<string>, ...]}. '
    'Use "matched" only when a candidate pinning genuinely and specifically '
    'covers the assumption (set "pinning_ref" to its ref). Use "unpinned" '
    'when no candidate covers it. Use "needs_human" when it is unclear.'
)

_FENCE_RE = re.compile(r"```(?:json)?\s*(?P<body>.*?)\s*```", re.DOTALL)


class IndependenceViolation(Exception):
    """Raised when the configured adjudicator is also the assumption's author."""


@dataclass(frozen=True)
class AdjudicatorVerdict:
    """The independent tier's output: a ruling plus candidate pinning ref."""

    ruling: Ruling
    pinning_ref: str | None
    questions: list[str]
    raw: str


class AdjudicatorProtocol(Protocol):
    """A stage-2 independent adjudication tier."""

    @property
    def identity(self) -> str:
        """Stable identity string (e.g. ``"tier:deepseek-coder"``) for the
        independence guard: it must never equal an assumption's author."""
        ...

    async def adjudicate(
        self,
        text: str,
        requirement_id: str | None,
        candidates: list[Candidate],
    ) -> AdjudicatorVerdict:
        """Adjudicate ``text`` vs ``candidates`` -- never the author's rationale."""
        ...


def _build_prompt(
    text: str, requirement_id: str | None, candidates: list[Candidate]
) -> str:
    """Pure prompt-construction — presents only text/requirement/candidates."""
    lines = [
        "Assumption:",
        text,
        "",
        f"Requirement: {requirement_id or '(none)'}",
        "",
        "Candidate pinnings (from a deterministic lexical pre-filter; "
        "provisional evidence only):",
    ]
    if not candidates:
        lines.append("  (none)")
    else:
        for candidate in candidates:
            lines.append(
                f"  - {candidate.pinning.ref} (score={candidate.score:.2f}): "
                f"{candidate.pinning.excerpt}"
            )
    lines.append("")
    lines.append(
        "Decide: does a candidate pinning genuinely and specifically cover "
        "this assumption (matched), is it clearly uncovered (unpinned), or "
        "is it unclear (needs_human)? Respond with the required JSON only."
    )
    return "\n".join(lines)


def _parse_verdict(raw: str) -> AdjudicatorVerdict:
    """Parse the adjudicator's raw output. Unparseable -> NEEDS_HUMAN (fail-closed)."""
    body = raw.strip()
    fence_match = _FENCE_RE.search(body)
    if fence_match:
        body = fence_match.group("body").strip()
    try:
        data = json.loads(body)
        ruling = Ruling(str(data["ruling"]).strip().lower())
        pinning_ref = data.get("pinning_ref")
        if pinning_ref is not None:
            pinning_ref = str(pinning_ref)
        raw_questions = data.get("questions") or []
        if not isinstance(raw_questions, list):
            raw_questions = [raw_questions]
        questions = [str(q) for q in raw_questions]
    except Exception:
        return AdjudicatorVerdict(
            ruling=Ruling.NEEDS_HUMAN, pinning_ref=None, questions=[], raw=raw
        )
    return AdjudicatorVerdict(
        ruling=ruling, pinning_ref=pinning_ref, questions=questions, raw=raw
    )


class LlmAdjudicator:
    """An ``AdjudicatorProtocol`` backed by ``llm.client.call_llm``."""

    def __init__(self, profile: str):
        self.profile = profile

    @property
    def identity(self) -> str:
        """This tier's identity string, e.g. ``"tier:deepseek-coder"``."""
        return f"tier:{self.profile}"

    async def adjudicate(
        self,
        text: str,
        requirement_id: str | None,
        candidates: list[Candidate],
    ) -> AdjudicatorVerdict:
        """Call the LLM profile and parse its strict-JSON verdict."""
        prompt = _build_prompt(text, requirement_id, candidates)
        raw = await call_llm(
            _SYSTEM_PROMPT, prompt, model=self.profile, temperature=0.0
        )
        return _parse_verdict(raw)