orchestrata_ultimate.factory.scheduler.queue

Reference › orchestrata_ultimate › factory › scheduler › queue

queue is the strictly serial integration re-gate and promotion drain (SPEC-033 row B21, ADR-0004, ADR-0033). ADR-0004 says a merge is fact promotion performed after the integrated-trunk re-gate passes. Builds run level-parallel, but promotion order is a correctness property: a task in level N+1 must see every level-N promotion already in trunk, and two promotions racing on the trunk namespace would make that nondeterministic.

MergeQueue takes a PromotionStore and an injected regate callable. enqueue appends a MergeEntry (lease, task and author identity) and is thread-safe, since parallel build workers call it. drain, called by the scheduler between levels and never alongside build dispatch, processes entries first in, first out: for each it calls regate, and on a passing RegateVerdict calls PromotionStore.promote, returning a MergeOutcome. The queue supplies the promoter identity merge-queue itself, so it is always distinct from the authoring agent. A gate token is never accepted from a caller; it comes only from a passing verdict, and MergeEntry has no token field, so a caller cannot fabricate promotion evidence. conformance_gate provides the production regate. Tests are in tests/factory/test_merge_queue.py.

At a glance

Class diagram

Dependency diagram

API

Classes

RegateVerdict (dataclass)

Decorators: dataclass(frozen=True) · Line 39

The integration re-gate's verdict on one queued :class:MergeEntry.

MergeEntry (dataclass)

Decorators: dataclass(frozen=True) · Line 65

One task queued for the integration re-gate + promotion.

MergeOutcome (dataclass)

Decorators: dataclass(frozen=True) · Line 79

The result of draining one :class:MergeEntry through the re-gate.

MergeQueue

Line 88

Strictly-serial integration re-gate + promotion drain (ADR-0004, B21).

Source

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

"""MergeQueue -- the strictly-serial integration re-gate + promotion drain.

ADR-0004: "merge is fact promotion, performed _after_ the integrated-trunk
re-gate passes". Build dispatch is level-parallel (ThreadPoolExecutor,
:mod:`.scheduler`), but promotion order is a CORRECTNESS property, not a
throughput knob: a level N+1 task's build must observe every level-N
promotion already in trunk, and two promotions racing on the same trunk
namespace would make that observation nondeterministic. This queue is the
single point where that serialization becomes code: entries are appended
(``enqueue``, thread-safe -- called from parallel build workers) and later
drained strictly one-at-a-time (``drain``, called by the scheduler
BETWEEN levels, never concurrently with build dispatch).

ADR-0033 (B21): the queue itself supplies ``promoter_identity="merge-queue"``
to every :meth:`~orchestrata_ultimate.factory.promotion.store.PromotionStore.promote`
call -- it IS the distinct promoting identity the ADR requires, never the
authoring agent. ``gate_token`` is never accepted from a caller: it comes
only from a passing :class:`RegateVerdict`, which this queue itself obtains
by calling the injected ``regate`` callable. A caller cannot fabricate
promotion evidence by constructing a ``MergeEntry`` with a token of its own
choosing, because ``MergeEntry`` carries no token field at all.
"""

from __future__ import annotations

import threading
from collections.abc import Callable
from dataclasses import dataclass

from orchestrata_ultimate.factory.promotion.models import PromotionRecord
from orchestrata_ultimate.factory.promotion.store import PromotionStore
from orchestrata_ultimate.factory.scheduler.dag import TaskSpec
from orchestrata_ultimate.factory.scheduler.worktrees import WorktreeLease

_PROMOTER_IDENTITY = "merge-queue"


@dataclass(frozen=True)
class RegateVerdict:
    """The integration re-gate's verdict on one queued :class:`MergeEntry`.

    ``token`` is only meaningful when ``passed`` is ``True`` -- it becomes
    the ``gate_token`` threaded into
    :meth:`~orchestrata_ultimate.factory.promotion.store.PromotionStore.promote`,
    i.e. it IS the promotion evidence (issued by the gate, never asserted by
    the author, per ADR-0021/B21). ``reason`` is populated on a failing
    verdict for the rejected-entry report (and flows straight through to
    ``TaskReport.detail`` -- see ``scheduler.py``). ``violations`` (Phase 6,
    ``factory.dispatch.conformance_gate``) is the same failure as a typed
    tuple -- offending file names for a scope refusal, or
    ``"<tool>:exit=<code>"`` for a review-gate refusal -- for a caller that
    wants machine-readable detail rather than parsing ``reason``'s prose.
    Defaults to ``()`` so existing ``regate`` callables that only ever set
    ``reason`` (e.g. ``factory_pipeline.py``'s original minimal gate, still
    exercised in ``tests/factory/test_merge_queue.py``) are unaffected.
    """

    passed: bool
    token: str = ""
    reason: str = ""
    violations: tuple[str, ...] = ()


@dataclass(frozen=True)
class MergeEntry:
    """One task queued for the integration re-gate + promotion.

    Deliberately carries no ``gate_token`` field: promotion evidence is
    issued by the ``regate`` callable's verdict, never asserted by the
    caller that builds this entry (ADR-0033/B21).
    """

    lease: WorktreeLease
    task: TaskSpec
    author_identity: str


@dataclass(frozen=True)
class MergeOutcome:
    """The result of draining one :class:`MergeEntry` through the re-gate."""

    entry: MergeEntry
    promoted: bool
    verdict: RegateVerdict
    promotion_record: PromotionRecord | None = None


class MergeQueue:
    """Strictly-serial integration re-gate + promotion drain (ADR-0004, B21).

    Implemented as a plain FIFO list drained sequentially by :meth:`drain`
    -- no background worker thread. A worker thread would still need to be
    the ONLY thing calling :meth:`~PromotionStore.promote` (never two at
    once), and a plain sequential drain called between DAG levels already
    guarantees that with less machinery: :meth:`enqueue` (called
    concurrently from parallel build workers) only ever appends under a
    lock, and :meth:`drain` (called once per level, from the scheduler's
    single thread) processes entries one at a time in enqueue (FIFO) order.
    """

    def __init__(
        self,
        store: PromotionStore,
        regate: Callable[[MergeEntry], RegateVerdict],
    ) -> None:
        self._store = store
        self._regate = regate
        self._entries: list[MergeEntry] = []
        self._lock = threading.Lock()

    def enqueue(self, entry: MergeEntry) -> None:
        """Append ``entry`` to the queue. Thread-safe (parallel build workers)."""
        with self._lock:
            self._entries.append(entry)

    def drain(self) -> list[MergeOutcome]:
        """Regate + promote every queued entry, strictly serially, FIFO order.

        Empties the queue first (under the lock) so a concurrent
        :meth:`enqueue` from a still-running build never observes a
        partially-drained queue. Each entry then runs, one at a time, in
        the calling thread: ``regate(entry)`` -> on a passing verdict,
        ``store.promote(entry.lease.namespace, gate_token=verdict.token,
        author_identity=entry.author_identity, promoter_identity="merge-queue")``;
        on a failing verdict, the entry is recorded rejected and its
        namespace is left untouched (candidate for the caller's eventual
        GC pass -- this queue never GCs).
        """
        with self._lock:
            pending, self._entries = self._entries, []

        outcomes: list[MergeOutcome] = []
        for entry in pending:
            verdict = self._regate(entry)
            if not verdict.passed:
                outcomes.append(
                    MergeOutcome(entry=entry, promoted=False, verdict=verdict)
                )
                continue

            record = self._store.promote(
                entry.lease.namespace,
                gate_token=verdict.token,
                author_identity=entry.author_identity,
                promoter_identity=_PROMOTER_IDENTITY,
            )
            outcomes.append(
                MergeOutcome(
                    entry=entry,
                    promoted=True,
                    verdict=verdict,
                    promotion_record=record,
                )
            )

        return outcomes