orchestrata_ultimate.factory.enforcement.models

Reference › orchestrata_ultimate › factory › enforcement › models

models is the data model of the governed-event ledger (SPEC-033 section 3, row B1c, Phase 1). Both dataclasses are frozen. WriterIdentity identifies a governed writer by a name, a binary hash and a config hash. GovernedWriteRecord is one governed-write event: a record id, the writer, materials and products (path to hash maps), the worktree, the commit, optional byproducts and a recording time; GovernedWriteRecord.new builds one with a fresh id. Its field names deliberately mirror in-toto link metadata (ADR-0030), so attestation can map it to a real link. DELETION_MARKER is the sentinel value a record puts in products[path] to declare that the path was deleted. The module has no internal imports; the ledger, enforce, bootstrap and attestation modules import it.

At a glance

Class diagram

Dependency diagram

API

Classes

WriterIdentity (dataclass)

Decorators: dataclass(frozen=True) · Line 29

Identity of a governed writer.

GovernedWriteRecord (dataclass)

Decorators: dataclass(frozen=True) · Line 51

One governed-write event, in-toto-link-SHAPED (ADR-0030).

Functions

Source

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

"""Data model for the governed-event ledger (SPEC-033 §3, B1c, Phase 1).

Every dataclass here is frozen (immutable records) — a governed-write record,
once appended to the ledger, is never mutated in place (mirrors the pattern
in :mod:`orchestrata_ultimate.factory.assumptions.models`).

``GovernedWriteRecord`` is deliberately in-toto-link-SHAPED (ADR-0030): field
names (``materials``/``products``/``byproducts``, an executing identity)
mirror in-toto link metadata so a future signing upgrade (B20) is a format
migration, not a redesign. No signing happens in Phase 1.
"""

from __future__ import annotations

import uuid
from dataclasses import dataclass, field
from datetime import UTC, datetime


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


def _new_record_id() -> str:
    return f"GWR-{uuid.uuid4().hex[:16]}"


@dataclass(frozen=True)
class WriterIdentity:
    """Identity of a governed writer.

    ADR-0017: identity is the **(binary-hash, config-hash) pair, never the
    name alone** — a writer's output is a function of its configuration
    (e.g. ``.pre-commit-config.yaml``), so a config change must change the
    identity rather than silently inheriting trust from the old name. ``name``
    is a human-readable label only; it plays no role in identity comparison
    (see ``WriterRegistry.registered`` in :mod:`.ledger`).
    """

    name: str
    binary_hash: str
    config_hash: str


# The deletion marker: a GovernedWriteRecord's `products[path]` set to this
# sentinel declares that `path` was DELETED by that record, per B1a rule 2.
DELETION_MARKER = "deleted"


@dataclass(frozen=True)
class GovernedWriteRecord:
    """One governed-write event, in-toto-link-SHAPED (ADR-0030).

    ``materials``/``products`` are path -> sha256 maps: materials are the
    inputs the writer consumed, products are the outputs it produced (or, for
    a declared deletion, the path mapped to :data:`DELETION_MARKER`).
    ``worktree``/``commit`` correlate the record to the commit diff it
    governs (ADR-0001). ``byproducts`` is free metadata (e.g. exit code).

    # B20: in-toto signing + key mgmt — no signature exists in Phase 1; the
    # field shape is chosen so adding one is additive, not a redesign.
    """

    record_id: str
    writer: WriterIdentity
    materials: dict[str, str]
    products: dict[str, str]
    worktree: str
    commit: str
    byproducts: dict[str, str] = field(default_factory=dict)
    recorded_at: str = field(default_factory=_utc_now_iso)

    @classmethod
    def new(
        cls,
        writer: WriterIdentity,
        materials: dict[str, str],
        products: dict[str, str],
        worktree: str,
        commit: str,
        byproducts: dict[str, str] | None = None,
    ) -> GovernedWriteRecord:
        """Build a fresh record with a generated id and current timestamp."""
        return cls(
            record_id=_new_record_id(),
            writer=writer,
            materials=dict(materials),
            products=dict(products),
            worktree=worktree,
            commit=commit,
            byproducts=dict(byproducts) if byproducts else {},
        )