orchestrata_ultimate.factory.assumptions.models

Reference › orchestrata_ultimate › factory › assumptions › models

models defines the data model of the assumption ledger and arbiter cascade (SPEC-033 row B13). It has no internal imports and is the base of the assumptions package: the ledger, cascade, adjudicator, prefilter and sources modules use it, as do the anchors, amendment and detectors packages.

Every dataclass is frozen; a change is made by building a replacement with dataclasses.replace and persisting it. The enums are PinningKind (AC, ANCHOR, ADR, ADJUDICATION), AssumptionStatus (PENDING, PINNED, UNPINNED, NEEDS_HUMAN) and Ruling (MATCHED, UNPINNED, NEEDS_HUMAN). The records are Pinning (kind, reference and excerpt), Assumption (text, design node, author identity, optional requirement id, status, resolved stage and open questions), Adjudication (a ruling promoted to the Project KB under ADR-0014, with a superseded_by link), InterpretationDisclosure (a chosen reading and its alternatives, AC-033-C1-03), Candidate (a pinning with its stage-1 score) and CascadeOutcome.

normalize lowercases, strips punctuation and collapses whitespace; fingerprint returns the sha256 hex digest of the normalised text. Assumption.new builds a PENDING assumption whose ASM- id derives deterministically from the fingerprint and the design node.

At a glance

Class diagram

Dependency diagram

API

Classes

PinningKind (enum)

Bases: StrEnum · Line 64

Kinds of pinning artifact an assumption can be matched to.

Pinning (dataclass)

Decorators: dataclass(frozen=True) · Line 74

A concrete artifact an assumption can be pinned to.

AssumptionStatus (enum)

Bases: StrEnum · Line 82

Lifecycle status of an :class:Assumption in the ledger.

Ruling (enum)

Bases: StrEnum · Line 91

The three possible arbiter rulings (KB hits and stage-2 verdicts).

Assumption (dataclass)

Decorators: dataclass(frozen=True) · Line 100

An assumption enumerated by an executing model.

Adjudication (dataclass)

Decorators: dataclass(frozen=True) · Line 153

A promoted ruling in the Project KB (ADR-0014).

InterpretationDisclosure (dataclass)

Decorators: dataclass(frozen=True) · Line 168

Fable's disclosed interpretation of a design node (AC-033-C1-03).

Candidate (dataclass)

Decorators: dataclass(frozen=True) · Line 179

A stage-1 candidate pinning with its lexical match score.

CascadeOutcome (dataclass)

Decorators: dataclass(frozen=True) · Line 187

The result of running :meth:ArbiterCascade.resolve on an assumption.

Functions

Source

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

"""Data model for the SPEC-033 assumption ledger + arbiter cascade (B13).

Every dataclass here is frozen (immutable records) — a "mutation" is done by
building a replacement instance (``dataclasses.replace``) and persisting it
through :class:`~orchestrata_ultimate.factory.assumptions.ledger.AssumptionLedger`
or :class:`~orchestrata_ultimate.factory.assumptions.ledger.FileProjectKB`,
matching the JSONL append/rewrite storage model. Enum values are lowercase
strings with no weasel words (e.g. ``"unpinned"``, never ``"issue"``).
"""

from __future__ import annotations

import hashlib
import re
from dataclasses import dataclass, field
from datetime import UTC, datetime
from enum import StrEnum

_PUNCTUATION_RE = re.compile(r"[^\w\s]", re.UNICODE)
_WHITESPACE_RE = re.compile(r"\s+")


def normalize(text: str) -> str:
    """Lowercase, strip punctuation, collapse whitespace.

    Shared normalization for :func:`fingerprint` and the KB fuzzy-match
    token-set comparison (``ledger.FileProjectKB.lookup``) — the two must
    agree on what counts as "the same text" up to normalization.
    """
    lowered = text.lower()
    no_punct = _PUNCTUATION_RE.sub(" ", lowered)
    return _WHITESPACE_RE.sub(" ", no_punct).strip()


def fingerprint(text: str) -> str:
    """Normalize ``text`` and return the full sha256 hex digest.

    Normalization: lowercase, strip punctuation, collapse whitespace. The
    fingerprint is the full hex digest (64 chars) — unlike ids derived from
    it, fingerprints are never truncated, since they are the durable identity
    used for KB exact-match lookups.
    """
    normalized = normalize(text)
    return hashlib.sha256(normalized.encode("utf-8")).hexdigest()


def _assumption_id(fp: str, design_node: str) -> str:
    """Derive an ``ASM-`` id from a fingerprint + design node.

    ``ASM-`` + the first 12 hex chars of sha256(``"{fingerprint}|{design_node}"``).
    Two assumptions with the same fingerprint and design node collide on this
    id by construction — that collision IS the duplicate-assumption signal
    :meth:`AssumptionLedger.add` enforces.
    """
    combined = f"{fp}|{design_node}"
    digest = hashlib.sha256(combined.encode("utf-8")).hexdigest()
    return f"ASM-{digest[:12]}"


def _utc_now_iso() -> str:
    return datetime.now(UTC).isoformat()


class PinningKind(StrEnum):
    """Kinds of pinning artifact an assumption can be matched to."""

    AC = "ac"
    ANCHOR = "anchor"
    ADR = "adr"
    ADJUDICATION = "adjudication"


@dataclass(frozen=True)
class Pinning:
    """A concrete artifact an assumption can be pinned to."""

    kind: PinningKind
    ref: str
    excerpt: str


class AssumptionStatus(StrEnum):
    """Lifecycle status of an :class:`Assumption` in the ledger."""

    PENDING = "pending"
    PINNED = "pinned"
    UNPINNED = "unpinned"
    NEEDS_HUMAN = "needs_human"


class Ruling(StrEnum):
    """The three possible arbiter rulings (KB hits and stage-2 verdicts)."""

    MATCHED = "matched"
    UNPINNED = "unpinned"
    NEEDS_HUMAN = "needs_human"


@dataclass(frozen=True)
class Assumption:
    """An assumption enumerated by an executing model.

    ``id`` is deterministic (:func:`_assumption_id`) so the same assumption
    text recurring at the same design node always maps to the same record —
    the mechanism behind AC-033-C1-04 (a KB hit resolves it without a second
    pass).
    """

    id: str
    text: str
    design_node: str
    author_identity: str
    requirement_id: str | None = None
    status: AssumptionStatus = AssumptionStatus.PENDING
    pinning: Pinning | None = None
    stage_resolved: int | None = None
    questions: list[str] = field(default_factory=list)
    created_at: str = field(default_factory=_utc_now_iso)

    @classmethod
    def new(
        cls,
        text: str,
        design_node: str,
        author_identity: str,
        requirement_id: str | None = None,
    ) -> Assumption:
        """Build a fresh ``PENDING`` assumption with a deterministic id.

        Args:
            text: The assumption text as enumerated by the executing model.
            design_node: The design node the assumption was made against.
            author_identity: Identity of the enumerating model/agent (e.g.
                ``"sonnet-5:builder"``) — never empty.
            requirement_id: Optional requirement this assumption relates to.

        Raises:
            ValueError: If ``author_identity`` is empty.
        """
        if not author_identity:
            raise ValueError("author_identity must not be empty")
        fp = fingerprint(text)
        return cls(
            id=_assumption_id(fp, design_node),
            text=text,
            design_node=design_node,
            author_identity=author_identity,
            requirement_id=requirement_id,
        )


@dataclass(frozen=True)
class Adjudication:
    """A promoted ruling in the Project KB (ADR-0014)."""

    id: str
    fingerprint: str
    assumption_text: str
    ruling: Ruling
    pinning: Pinning | None
    adjudicator: str
    requirement_id: str | None
    adjudicated_at: str
    superseded_by: str | None = None


@dataclass(frozen=True)
class InterpretationDisclosure:
    """Fable's disclosed interpretation of a design node (AC-033-C1-03)."""

    design_node: str
    chosen_reading: str
    alternatives: list[str]
    author_identity: str
    created_at: str = field(default_factory=_utc_now_iso)


@dataclass(frozen=True)
class Candidate:
    """A stage-1 candidate pinning with its lexical match score."""

    pinning: Pinning
    score: float


@dataclass(frozen=True)
class CascadeOutcome:
    """The result of running :meth:`ArbiterCascade.resolve` on an assumption."""

    assumption_id: str
    status: AssumptionStatus
    pinning: Pinning | None
    stage: int | None
    questions: list[str]
    candidates: list[Candidate]