orchestrata_ultimate.factory.dispatch.gates

Reference › orchestrata_ultimate › factory › dispatch › gates

gates holds the admission gates of the factory (SPEC-033 row B21, ADR-0033, ADR-0032). They refuse loudly, before dispatch or promotion, in three cases.

• Self-merge. no_self_merge refuses a promotion whose author and promoter identities are equal and raises SelfMergeRefused. An empty or None identity raises IdentityUnknown, following ADR-0032's rule that undetermined is not allowed.
• Duplication. InFlightRegistry is a Redis-backed registry (ADR-0033 section 2): claim takes a task key for a holder with a time-to-live, release frees it only for the current holder (a compare-and-delete script), and holder_of reads the holder. task_key builds the canonical key from a requirement id and unit paths. duplication_gate raises DuplicateDispatch if the key is already held.
• Staleness. staleness_gate raises StaleDispatch if trunk has changed a declared path since the task's base commit; _path_overlaps is the file-or-directory prefix rule.

The docstring states a limit: the OS-level three-user separation of ADR-0018 is not built, so identity here is only the actor string recorded in the ledger. The gate is a convention check, not a privilege boundary. conformance_gate, the promotion store, the scheduler and the dispatch and pipeline CLIs import it; tests are in tests/factory/test_dispatch_gates.py.

At a glance

Class diagram

Dependency diagram

API

Classes

SelfMergeRefused

Bases: Exception · Line 56

Raised by :func:no_self_merge when author and promoter are the same identity (ADR-0033 §1: no agent merges its own work).

IdentityUnknown

Bases: Exception · Line 70

Raised by :func:no_self_merge when either identity is empty/None (ADR-0032: undetermined is refused, never allowed).

DuplicateDispatch

Bases: Exception · Line 87

Raised by :func:duplication_gate when task_key is already claimed by another holder.

StaleDispatch

Bases: Exception · Line 100

Raised by :func:staleness_gate when trunk has moved on a path the task declared, or when base_commit cannot be resolved at all (an unreachable commit can't be proven fresh, so it is refused…

InFlightRegistry

Line 161

Redis-backed dispatch in-flight registry (ADR-0033 §2).

Functions

Source

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

"""No-self-merge + dispatch duplication/staleness gates (SPEC-033 §1/§2/§7, B21).

ADR-0033 ("no agent merges its own work; dispatch gates on duplication and
staleness") names three pre-dispatch/pre-promotion failures that must refuse
loudly instead of being adjudicated later:

- **Self-merge** (:func:`no_self_merge`): the identity that authored a
  change may not also be the identity that promotes it (ADR-0033 §1). ADR-
  0032's admission-control contract applies directly here: "undetermined !=
  allowed" -- an empty/``None`` identity is REFUSED (:class:`IdentityUnknown`),
  never treated as a permissive default.
- **Duplication** (:class:`InFlightRegistry` + :func:`duplication_gate`): a
  task already in flight must not be handed to a second agent.
- **Staleness** (:func:`staleness_gate`): a task dispatched against a base
  commit whose declared paths trunk has since touched must be refused, not
  validated against facts that no longer describe trunk.

**Identity limitation (read before relying on this module for isolation).**
ADR-0018 specifies OS-level three-uid separation (agent / sandboxed executor
/ trusted gate) as the structural enforcement of "no agent merges its own
work". That separation is **not built** -- it is B1d, a distinct, not-yet-
landed lane. Until B1d lands and carries its own verified acceptance test
(ADR-0018's own requirement: "verified, never assumed"), "identity" in this
module means only the **ledger-recorded actor string** passed in by the
caller (see ``author_identity``/``promoter_identity`` threaded through
:meth:`orchestrata_ultimate.factory.promotion.store.PromotionStore.promote`).
This module cannot stop a caller from fabricating either string -- it only
refuses to promote when the strings it was given collide or are missing.
Do not read a passing :func:`no_self_merge` call as proof of OS-level
identity separation; it is a convention gate, not a privilege boundary.
"""

from __future__ import annotations

import hashlib
import subprocess
from collections.abc import Sequence
from pathlib import Path
from typing import TYPE_CHECKING, cast

if TYPE_CHECKING:
    from redis import Redis

# Lua compare-and-delete: only the current holder may release -- an
# unguarded DEL would let a non-holder (or a holder whose claim already
# expired and was re-claimed by someone else) blow away a live claim.
_RELEASE_SCRIPT = """
if redis.call('get', KEYS[1]) == ARGV[1] then
    return redis.call('del', KEYS[1])
else
    return 0
end
"""


class SelfMergeRefused(Exception):
    """Raised by :func:`no_self_merge` when author and promoter are the same
    identity (ADR-0033 §1: no agent merges its own work). Carries both
    identities verbatim."""

    def __init__(self, author_identity: str, promoter_identity: str) -> None:
        self.author_identity = author_identity
        self.promoter_identity = promoter_identity
        super().__init__(
            f"self-merge refused: author_identity={author_identity!r} "
            f"== promoter_identity={promoter_identity!r}"
        )


class IdentityUnknown(Exception):
    """Raised by :func:`no_self_merge` when either identity is empty/``None``
    (ADR-0032: undetermined is refused, never allowed). Carries both
    identities verbatim (as given -- one may be legitimately populated)."""

    def __init__(
        self, author_identity: str | None, promoter_identity: str | None
    ) -> None:
        self.author_identity = author_identity
        self.promoter_identity = promoter_identity
        super().__init__(
            f"identity unknown -- refused per ADR-0032: "
            f"author_identity={author_identity!r} "
            f"promoter_identity={promoter_identity!r}"
        )


class DuplicateDispatch(Exception):
    """Raised by :func:`duplication_gate` when ``task_key`` is already
    claimed by another holder. Names the current holder."""

    def __init__(self, task_key: str, holder: str) -> None:
        self.task_key = task_key
        self.holder = holder
        super().__init__(
            f"duplicate dispatch refused: task_key={task_key!r} already "
            f"held by holder={holder!r}"
        )


class StaleDispatch(Exception):
    """Raised by :func:`staleness_gate` when trunk has moved on a path the
    task declared, or when ``base_commit`` cannot be resolved at all (an
    unreachable commit can't be proven fresh, so it is refused the same
    way -- fail closed, per ADR-0002/0032)."""

    def __init__(
        self,
        base_commit: str,
        overlap: Sequence[str] = (),
        *,
        reason: str | None = None,
    ) -> None:
        self.base_commit = base_commit
        self.overlap = list(overlap)
        self.reason = reason
        if reason is not None:
            msg = (
                f"stale dispatch refused: base_commit={base_commit!r} could "
                f"not be resolved ({reason}) -- can't prove fresh, not fresh"
            )
        else:
            msg = (
                f"stale dispatch refused: base_commit={base_commit!r} -- "
                f"trunk has moved on declared path(s) {self.overlap}"
            )
        super().__init__(msg)


def no_self_merge(author_identity: str, promoter_identity: str) -> None:
    """Refuse a promotion whose author and promoter are the same identity.

    ADR-0033 §1: merge requires an identity distinct from the authoring
    identity. Per ADR-0032's admission-control contract ("undetermined !=
    allowed"), an empty/``None`` identity is refused with
    :class:`IdentityUnknown` -- it is NOT treated as "no constraint" or
    silently passed through. Only once both identities are known non-empty
    strings is the equality check meaningful, and equal identities raise
    :class:`SelfMergeRefused`. Comparison is case-sensitive exact match --
    no normalization, no aliasing.
    """
    if not author_identity or not promoter_identity:
        raise IdentityUnknown(author_identity, promoter_identity)
    if author_identity == promoter_identity:
        raise SelfMergeRefused(author_identity, promoter_identity)


def task_key(requirement_id: str, unit_paths: Sequence[str]) -> str:
    """The canonical dispatch task-identity key (ADR-0033 §2).

    ``"{requirement_id}:{sha256(sorted unit_paths)[:16]}"`` -- the paths are
    sorted (so caller-supplied ordering never changes the key) and joined
    with ``"\\n"`` before hashing, which is deterministic and collision-safe
    against path components containing arbitrary characters other than a
    literal newline.
    """
    joined = "\n".join(sorted(unit_paths))
    digest = hashlib.sha256(joined.encode("utf-8")).hexdigest()[:16]
    return f"{requirement_id}:{digest}"


class InFlightRegistry:
    """Redis-backed dispatch in-flight registry (ADR-0033 §2).

    ``claim`` is a single ``SET NX EX`` round-trip -- the claim IS the
    registration, so there is no separate check-then-claim window for two
    dispatchers to race through. ``release`` is a Lua compare-and-delete:
    only the current holder's release actually deletes the key: a
    non-holder's (or an already-expired-and-reclaimed key's) release is a
    silent no-op, never an unguarded ``DEL``.
    """

    def __init__(self, redis_client: Redis) -> None:
        self._client = redis_client
        self._release_script = redis_client.register_script(_RELEASE_SCRIPT)

    def claim(self, task_key: str, holder: str, ttl_s: int) -> bool:
        """Attempt to claim ``task_key`` for ``holder``.

        Returns ``True`` iff this call created the key (i.e. it was not
        already held); ``False`` if another holder already holds it.
        """
        return bool(self._client.set(task_key, holder, nx=True, ex=ttl_s))

    def release(self, task_key: str, holder: str) -> None:
        """Release ``task_key`` iff ``holder`` currently holds it.

        A no-op (no exception, no effect) when ``holder`` does not
        currently hold ``task_key`` -- compare-and-delete via a Lua script,
        never an unguarded ``DEL``.
        """
        self._release_script(keys=[task_key], args=[holder])

    def holder_of(self, task_key: str) -> str | None:
        """Return the current holder of ``task_key``, or ``None`` if unclaimed."""
        return cast("str | None", self._client.get(task_key))


def duplication_gate(
    registry: InFlightRegistry,
    task_key: str,
    holder: str,
    ttl_s: int = 3600,
) -> None:
    """Refuse dispatch if ``task_key`` is already in flight.

    Delegates directly to :meth:`InFlightRegistry.claim` -- the claim
    attempt IS the duplication check (single round-trip, no check-then-claim
    race). On success the claim is now registered under ``holder``; on
    failure, raises :class:`DuplicateDispatch` naming the current holder.
    ``ttl_s`` defaults to 3600s, matching ``scripts/factory_dispatch.py``'s
    ``--ttl`` default.
    """
    if not registry.claim(task_key, holder, ttl_s):
        current_holder = registry.holder_of(task_key)
        raise DuplicateDispatch(task_key, current_holder or "<unknown>")


def _path_overlaps(changed_path: str, declared_path: str) -> bool:
    """Whether ``changed_path`` falls under ``declared_path`` (file or dir)."""
    declared = declared_path.rstrip("/")
    return changed_path == declared or changed_path.startswith(declared + "/")


def staleness_gate(
    repo_root: Path, base_commit: str, declared_paths: Sequence[str]
) -> None:
    """Refuse dispatch if trunk has moved on a path the task declared.

    Runs ``git diff --name-only base_commit..HEAD`` (subprocess, cwd
    ``repo_root``). Any changed path intersecting ``declared_paths``
    (prefix match for directories -- a declared path ending in ``/`` or a
    declared path that is itself a directory containing the changed file)
    raises :class:`StaleDispatch` listing the overlap. Trunk moving on
    disjoint paths is the fast, cheap pass-through (ADR-0033: "trunk moved
    but not in our impact set" stays cheap). An unknown/unreachable
    ``base_commit`` raises :class:`StaleDispatch` too -- fail closed,
    because an unresolvable base can't be proven fresh.
    """
    try:
        result = subprocess.run(
            ["git", "diff", "--name-only", f"{base_commit}..HEAD"],
            cwd=repo_root,
            capture_output=True,
            text=True,
            check=True,
        )
    except subprocess.CalledProcessError as exc:
        reason = (exc.stderr or str(exc)).strip()
        raise StaleDispatch(base_commit, reason=reason) from exc

    changed_paths = [line for line in result.stdout.splitlines() if line]
    overlap = [
        changed
        for changed in changed_paths
        if any(_path_overlaps(changed, declared) for declared in declared_paths)
    ]
    if overlap:
        raise StaleDispatch(base_commit, overlap)