orchestrata_ultimate.factory.dispatch.conformance_gate

Reference › orchestrata_ultimate › factory › dispatch › conformance_gate

conformance_gate is the pre-merge gate for delegated build work (SPEC-033 Phase 6, ADR-0033). It replaces a placeholder re-gate in the pipeline script that only checked that a namespace had a fact to promote. Two checks now run. scope_gate compares the files a worktree changed against its base commit (_changed_paths) with the task's declared paths, reusing the path-overlap rule from gates, and raises ScopeViolation naming every offending path. review_gate runs the project's code-review.py and then test-review.py as real subprocesses on the changed paths with a timeout; a non-zero exit raises ReviewGateFailure carrying the captured output. conformance_regate composes both into the callable that the merge queue already invokes between DAG levels and returns a RegateVerdict, so a failure takes the existing rejection route with no human step. The scripts.factory_pipeline CLI imports it; tests are in tests/factory/test_conformance_gate.py.

At a glance

Class diagram

Dependency diagram

API

Classes

ScopeViolation

Bases: Exception · Line 67

Raised by :func:scope_gate when the worktree changed a file outside the task's declared scope.

ReviewGateFailure

Bases: Exception · Line 82

Raised by :func:review_gate when code-review.py or test-review.py exits non-zero.

Functions

Source

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

"""Pre-merge conformance gate for delegated build-factory work (SPEC-033
Phase 6 / ADR-0033; see ``plans/workflow-improvements-plan.md`` "Phase 6
audit findings (2026-07-26)").

Closes the two gaps the Phase 6 audit named in the delegation pipeline's
integration re-gate. Before this module, ``scripts/factory_pipeline.py``'s
``_default_regate`` was an intentionally MINIMAL placeholder ("Swap
``_default_regate`` for a real gate once that lane lands" -- its own
docstring) that only checked "does the namespace have at least one fact to
promote". It did NOT check whether the returned work stayed inside its
assigned file scope, and did NOT run it through the project's own review
tooling before promotion:

1. **Scope check** (:func:`scope_gate`) -- a task's worktree may only
   change files inside its ``TaskSpec.declared_paths``. Reuses
   :func:`orchestrata_ultimate.factory.dispatch.gates._path_overlaps`, the
   same file-vs-directory-prefix rule :func:`~.gates.staleness_gate`
   already applies to TRUNK's paths -- here applied to the WORKER's own
   changes instead. An out-of-scope edit is refused, naming every
   offending path.
2. **Review gate** (:func:`review_gate`) -- every changed ``.py`` path is
   run through the project's own ``scripts/code-review.py`` and
   ``scripts/test-review.py`` (real subprocesses, the SAME tools a human
   reviewer would run -- e.g. ``test-review.py``'s T1 rule is what catches
   a mock introduced in a test file). Either exiting non-zero refuses the
   entry, carrying the tool's captured output.

:func:`conformance_regate` composes both into a single
``Callable[[MergeEntry], RegateVerdict]`` -- exactly the seam
:class:`~orchestrata_ultimate.factory.scheduler.queue.MergeQueue` already
calls between DAG levels (``queue.drain()`` -> ``self._regate(entry)``), so
a failing check routes back through the EXISTING rejection mechanism
(``RegateVerdict(passed=False, ...)`` -> ``MergeOutcome(promoted=False)``
-> ``TaskReport(outcome="regate_rejected", ...)``) with zero human step --
never a parallel one. ``RegateVerdict.violations`` (new field on the
existing dataclass, backward compatible -- see :mod:`.scheduler.queue`)
carries the offending file names / failing tool identity as a typed tuple,
alongside the existing human-readable ``reason`` string.

Neither check is a correctness claim (compare
:mod:`orchestrata_ultimate.factory.conformance.engine`'s ADR-0008
boundary): passing both proves the change stayed in its declared lane and
is lint/type/no-mock-clean, nothing about semantic intent. The full
AST≡design structural conformance engine (B2) and any "does trunk-plus-
this-change still satisfy the design" judgement remain separate, not-yet-
wired-here lanes.
"""

from __future__ import annotations

import os
import subprocess
import sys
from collections.abc import Sequence
from pathlib import Path
from uuid import uuid4

from orchestrata_ultimate.factory.dispatch.gates import _path_overlaps
from orchestrata_ultimate.factory.scheduler.queue import MergeEntry, RegateVerdict

# Captured tool output is evidence a check ran and why it failed, not a log
# store -- truncate like factory_pipeline.py's own stdout snippet cap.
_OUTPUT_SNIPPET_LIMIT = 4000
_REVIEW_TIMEOUT_S = 300.0


class ScopeViolation(Exception):
    """Raised by :func:`scope_gate` when the worktree changed a file outside
    the task's declared scope. Names every offending path (never truncated,
    per the project's no-truncated-answer-set discipline -- see
    :mod:`orchestrata_ultimate.factory.enforcement.enforce`'s docstring)."""

    def __init__(self, offending: Sequence[str], declared_paths: Sequence[str]) -> None:
        self.offending = sorted(offending)
        self.declared_paths = list(declared_paths)
        super().__init__(
            f"scope violation refused: file(s) outside declared scope "
            f"{self.offending} (declared: {self.declared_paths})"
        )


class ReviewGateFailure(Exception):
    """Raised by :func:`review_gate` when ``code-review.py`` or
    ``test-review.py`` exits non-zero. Carries the failing tool's name,
    exit code, and captured stdout+stderr (truncated)."""

    def __init__(self, tool: str, returncode: int, output: str) -> None:
        self.tool = tool
        self.returncode = returncode
        self.output = output[:_OUTPUT_SNIPPET_LIMIT]
        super().__init__(
            f"review gate refused: {tool} exited {returncode}\n{self.output}"
        )


def _changed_paths(worktree: Path, base_commit: str) -> list[str]:
    """Every path ``worktree`` changed relative to ``base_commit``.

    Covers both tracked modifications (committed inside the worktree,
    staged, or still unstaged -- ``git diff --name-only <base_commit>``
    diffs the working tree directly against that commit, so all three read
    as one) AND brand-new untracked files (``git ls-files --others
    --exclude-standard``, which a plain ``git diff`` never reports since an
    untracked path was never part of git's tracked state).
    """
    tracked = subprocess.run(
        ["git", "diff", "--name-only", base_commit],
        cwd=worktree,
        capture_output=True,
        text=True,
        check=True,
    )
    untracked = subprocess.run(
        ["git", "ls-files", "--others", "--exclude-standard"],
        cwd=worktree,
        capture_output=True,
        text=True,
        check=True,
    )
    changed = {line for line in tracked.stdout.splitlines() if line}
    changed |= {line for line in untracked.stdout.splitlines() if line}
    return sorted(changed)


def scope_gate(
    worktree: Path, base_commit: str, declared_paths: Sequence[str]
) -> list[str]:
    """Refuse ``worktree`` if it changed any file outside ``declared_paths``.

    Returns the full list of changed paths (whether or not any violates) so
    :func:`conformance_regate` can reuse it for :func:`review_gate` without
    a second git round-trip. Raises :class:`ScopeViolation` naming every
    offending path -- an empty ``declared_paths`` sequence refuses every
    changed path (fail closed: an undeclared scope grants no free pass).
    """
    changed = _changed_paths(worktree, base_commit)
    offending = [
        path
        for path in changed
        if not any(_path_overlaps(path, declared) for declared in declared_paths)
    ]
    if offending:
        raise ScopeViolation(offending, declared_paths)
    return changed


def review_gate(
    paths: Sequence[str],
    *,
    worktree: Path,
    code_review_script: Path,
    test_review_script: Path,
    timeout_s: float = _REVIEW_TIMEOUT_S,
) -> None:
    """Run ``code-review.py`` then ``test-review.py`` against ``paths``.

    ``paths`` are worktree-relative (as returned by :func:`scope_gate`);
    only ``.py`` paths that still exist are handed to the tools (a declared
    -and-deleted path has nothing left to lint). Nothing to review is a
    silent pass -- neither tool is invoked. Real subprocesses, ``cwd=
    worktree`` (the SAME tools + config a human reviewer would run against
    this exact checkout), ``sys.executable`` so the tools run under the
    same interpreter/venv as the calling process. Raises
    :class:`ReviewGateFailure` naming the first tool that exits non-zero.
    ``test-review.py`` runs BEFORE ``code-review.py`` -- it is the more
    specific reviewer for anything named like a test file (T1-T16, e.g. no
    mocking), so a test-file violation is attributed to it even though
    ``code-review.py``'s own ``ChecklistChecker`` would independently catch
    the same ``test_*.py``-only mock-import pattern. Repo-scoped gates
    inside ``code-review.py`` (banned-dep, prolog-boundary) are rooted at
    ``worktree`` via ``CODE_REVIEW_REPO_ROOT`` -- the reviewed checkout's
    own repo, not the host tool repo, so a foreign worktree never pays (or
    is judged by) host-repo-wide gates.
    """
    py_paths = [p for p in paths if p.endswith(".py") and (worktree / p).is_file()]
    if not py_paths:
        return
    for tool_name, script in (
        ("test-review.py", test_review_script),
        ("code-review.py", code_review_script),
    ):
        result = subprocess.run(
            [sys.executable, str(script), *py_paths],
            cwd=worktree,
            capture_output=True,
            text=True,
            timeout=timeout_s,
            env={**os.environ, "CODE_REVIEW_REPO_ROOT": str(worktree)},
        )
        if result.returncode != 0:
            raise ReviewGateFailure(
                tool_name, result.returncode, (result.stdout + result.stderr).strip()
            )


def conformance_regate(
    entry: MergeEntry,
    *,
    code_review_script: Path,
    test_review_script: Path,
    review_timeout_s: float = _REVIEW_TIMEOUT_S,
) -> RegateVerdict:
    """The real pre-merge conformance gate: scope, then review.

    A ``Callable[[MergeEntry], RegateVerdict]`` -- pass this directly (bind
    the two script paths with ``functools.partial``/a lambda) as
    :func:`~orchestrata_ultimate.factory.scheduler.scheduler.run_pipeline`'s
    ``regate`` argument. ``scripts/factory_pipeline.py``'s
    ``_default_regate`` wires this in for the CLI (Phase 6 gap closure); a
    landed builder integration should do the same rather than re-deriving
    these checks.
    """
    try:
        changed = scope_gate(
            entry.lease.path, entry.lease.commit, entry.task.declared_paths
        )
    except ScopeViolation as exc:
        return RegateVerdict(
            passed=False, reason=str(exc), violations=tuple(exc.offending)
        )

    try:
        review_gate(
            changed,
            worktree=entry.lease.path,
            code_review_script=code_review_script,
            test_review_script=test_review_script,
            timeout_s=review_timeout_s,
        )
    except ReviewGateFailure as exc:
        return RegateVerdict(
            passed=False,
            reason=str(exc),
            violations=(f"{exc.tool}:exit={exc.returncode}",),
        )

    return RegateVerdict(passed=True, token=f"conformance-{uuid4().hex}")


__all__ = [
    "ReviewGateFailure",
    "ScopeViolation",
    "conformance_regate",
    "review_gate",
    "scope_gate",
]