orchestrata_ultimate.factory.enforcement.enforce

Reference › orchestrata_ultimate › factory › enforcement › enforce

enforce is the state-diff enforcement core of the factory (SPEC-033 section 3, row B1a, Phase 1), implementing ADR-0001's boundary. At commit time, the tree diff between base_commit and head_commit is matched content-exactly against the EventLedger and gated by the WriterRegistry (ADR-0017). There is no allow-by-default branch: every path in the delta must be accounted for, an empty ledger against a non-empty delta makes every path a violation, and no violating path is truncated from the report (ADR-0026).

enforce builds the baseline and head trees and the delta with bootstrap_baseline and classify_delta from bootstrap, reads the ledger records for the worktree and head commit, and runs _check_all_paths. _check_added_or_modified and _check_deleted each return a DeltaViolation or None. The four violation kinds are the constants UNREGISTERED_WRITER, HASH_MISMATCH, UNREGISTERED_NEW_FILE and UNDECLARED_DELETION; a deletion counts as declared only when a record's product carries the deletion marker for the path. _writer_almost_matched_detail adds explanatory text when a record matched on path but not on writer. The result is a Verdict with passed and the full list of violations. The scripts.factory_enforce CLI imports it; tests are in tests/factory/test_enforcement_enforce.py.

At a glance

Class diagram

Dependency diagram

API

Classes

DeltaViolation (dataclass)

Decorators: dataclass(frozen=True) · Line 37

One path that failed enforcement.

Verdict (dataclass)

Decorators: dataclass(frozen=True) · Line 46

The result of :func:enforce: pass/fail plus every violation.

Functions

Source

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

"""The state-diff enforcement core (SPEC-033 §3, B1a, Phase 1).

``enforce`` implements ADR-0001's boundary: at commit, the full-filesystem
tree diff between ``base_commit`` and ``head_commit`` is matched
content-exactly against the event ledger, gated by the writer registry
(ADR-0017). There is **no allow-by-default branch** — every path in the
delta is accounted for explicitly, and an empty ledger against a nonempty
delta means every path violates. No violating path is ever truncated out of
the report (ADR-0026 discipline: a governance predicate may not return a
truncated answer set).
"""

from __future__ import annotations

from dataclasses import dataclass
from pathlib import Path

from orchestrata_ultimate.factory.enforcement.bootstrap import (
    DeltaClassification,
    bootstrap_baseline,
    classify_delta,
)
from orchestrata_ultimate.factory.enforcement.ledger import EventLedger, WriterRegistry
from orchestrata_ultimate.factory.enforcement.models import (
    DELETION_MARKER,
    GovernedWriteRecord,
)

# The four violation kinds this module produces.
UNREGISTERED_WRITER = "unregistered_writer"
HASH_MISMATCH = "hash_mismatch"
UNREGISTERED_NEW_FILE = "unregistered_new_file"
UNDECLARED_DELETION = "undeclared_deletion"


@dataclass(frozen=True)
class DeltaViolation:
    """One path that failed enforcement."""

    path: str
    kind: str
    detail: str


@dataclass(frozen=True)
class Verdict:
    """The result of :func:`enforce`: pass/fail plus every violation."""

    passed: bool
    violations: list[DeltaViolation]


def _writer_almost_matched_detail(record: GovernedWriteRecord) -> str:
    w = record.writer
    return (
        f"writer name={w.name!r} binary_hash={w.binary_hash!r} "
        f"config_hash={w.config_hash!r} produced a matching product hash "
        "but is not a registered writer"
    )


def _check_added_or_modified(
    path: str,
    head_hash: str,
    is_added: bool,
    records: list[GovernedWriteRecord],
    registry: WriterRegistry,
) -> DeltaViolation | None:
    """Check one added/modified path; return a violation or ``None``."""
    candidates = [r for r in records if path in r.products]

    if not candidates:
        kind = UNREGISTERED_NEW_FILE if is_added else HASH_MISMATCH
        detail = (
            f"no ledger record declares products[{path!r}]; "
            f"head commit has hash={head_hash!r}"
        )
        return DeltaViolation(path=path, kind=kind, detail=detail)

    exact_hash_matches = [r for r in candidates if r.products[path] == head_hash]
    if any(registry.registered(r.writer) for r in exact_hash_matches):
        return None  # governed: registered writer, content-exact hash match

    if exact_hash_matches:
        # A record matches the hash exactly, but no registered writer
        # produced it -- name the writer that ALMOST matched.
        return DeltaViolation(
            path=path,
            kind=UNREGISTERED_WRITER,
            detail=_writer_almost_matched_detail(exact_hash_matches[0]),
        )

    # Ledger declares SOME hash for this path, but none equals head's.
    best = candidates[0]
    detail = (
        f"ledger declares hash={best.products[path]!r}, "
        f"head commit has hash={head_hash!r}"
    )
    return DeltaViolation(path=path, kind=HASH_MISMATCH, detail=detail)


def _check_deleted(
    path: str,
    records: list[GovernedWriteRecord],
    registry: WriterRegistry,
) -> DeltaViolation | None:
    """Check one deleted path; return a violation or ``None``."""
    declared = [r for r in records if r.products.get(path) == DELETION_MARKER]
    if any(registry.registered(r.writer) for r in declared):
        return None  # governed: registered writer declared this deletion

    if declared:
        return DeltaViolation(
            path=path,
            kind=UNREGISTERED_WRITER,
            detail=_writer_almost_matched_detail(declared[0]),
        )

    return DeltaViolation(
        path=path,
        kind=UNDECLARED_DELETION,
        detail=f"{path!r} is absent from head commit with no declaring ledger record",
    )


def _check_all_paths(
    delta: DeltaClassification,
    head_tree: dict[str, str],
    records: list[GovernedWriteRecord],
    registry: WriterRegistry,
) -> list[DeltaViolation]:
    """Check every added/modified/deleted path; return every violation.

    Fail closed, no allow-by-default branch: every path in ``delta`` is
    checked explicitly and unaccounted-for paths violate.
    """
    violations: list[DeltaViolation] = []

    for path in delta.added:
        v = _check_added_or_modified(path, head_tree[path], True, records, registry)
        if v is not None:
            violations.append(v)

    for path in delta.modified:
        v = _check_added_or_modified(path, head_tree[path], False, records, registry)
        if v is not None:
            violations.append(v)

    for path in delta.deleted:
        v = _check_deleted(path, records, registry)
        if v is not None:
            violations.append(v)

    return violations


def enforce(
    base_commit: str,
    head_commit: str,
    repo_root: Path,
    ledger: EventLedger,
    registry: WriterRegistry,
    worktree: str,
) -> Verdict:
    """Enforce the committed-state boundary between two commits.

    1. Build both trees (path -> sha256) and classify the delta.
    2. Every ``added``/``modified`` path must be the product of a ledger
       record whose writer is registered AND whose recorded product hash
       content-exactly equals the file's hash at ``head_commit``.
    3. Every ``deleted`` path must appear as a product with the
       :data:`~orchestrata_ultimate.factory.enforcement.models.DELETION_MARKER`
       hash in some record from a registered writer (a declared deletion).
    4. Fail closed: any path not accounted for is a violation. There is no
       allow-by-default branch — an empty ledger against a nonempty delta
       means every path in the delta violates.
    """
    base_tree = bootstrap_baseline(repo_root, base_commit)
    head_tree = bootstrap_baseline(repo_root, head_commit)
    delta = classify_delta(base_tree, head_tree)
    records = ledger.records_for(worktree, head_commit)

    violations = _check_all_paths(delta, head_tree, records, registry)
    violations.sort(key=lambda v: (v.path, v.kind))
    return Verdict(passed=not violations, violations=violations)