orchestrata_ultimate.factory.scheduler.dag

Reference › orchestrata_ultimate › factory › scheduler › dag

dag defines the build DAG and its levelling (SPEC-033 section 9, row B9). TaskSpec is one node: a task id, a requirement id, the declared_paths it may change, its depends_on ids and an injected build callable. The callable receives a BuildContext (task id, WorktreeLease and fact namespace) and returns a BuildResult with a success flag and detail. build_levels applies Kahn's topological sort and returns a list of levels; tasks in the same level are independent and safe to dispatch concurrently, because each has its own (worktree, commit) namespace (ADR-0004). A dependency cycle raises CyclicDependency naming the members. The module makes no git or graph calls. The scheduler, the merge queue and the scripts.factory_pipeline CLI import it; tests are in tests/factory/test_scheduler_dag.py.

At a glance

Class diagram

Dependency diagram

API

Classes

BuildResult (dataclass)

Decorators: dataclass(frozen=True) · Line 19

Outcome of one task's injected build callable.

BuildContext (dataclass)

Decorators: dataclass(frozen=True) · Line 27

What a build callable runs against: the lease + its fact namespace.

TaskSpec (dataclass)

Decorators: dataclass(frozen=True) · Line 42

One node of the build DAG.

CyclicDependency

Bases: Exception · Line 60

Raised by :func:build_levels when the DAG contains a cycle.

Functions

Source

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

"""TaskSpec + Kahn topological leveling for the build DAG (SPEC-033 §9, B9).

Independent tasks within one level are safe to dispatch **concurrently**
(ADR-0004: each gets its own `(worktree, commit)` fact namespace) -- this
module only computes which tasks those are; it makes no git/Neo4j calls of
its own.
"""

from __future__ import annotations

from collections.abc import Callable, Sequence
from dataclasses import dataclass

from orchestrata_ultimate.factory.promotion.models import FactNamespace
from orchestrata_ultimate.factory.scheduler.worktrees import WorktreeLease


@dataclass(frozen=True)
class BuildResult:
    """Outcome of one task's injected ``build`` callable."""

    success: bool
    detail: str = ""


@dataclass(frozen=True)
class BuildContext:
    """What a ``build`` callable runs against: the lease + its fact namespace.

    ``namespace`` is always ``lease.namespace`` -- carried as its own field
    (rather than making callers reach through ``lease``) because it is the
    thing a build callable actually needs to tag its fact writes with, and
    spelling it out here matches how the brief names this seam.
    """

    task_id: str
    lease: WorktreeLease
    namespace: FactNamespace


@dataclass(frozen=True)
class TaskSpec:
    """One node of the build DAG.

    ``build`` is an INJECTED callable -- the real builder integration (an
    actual code-generation driver) is future work; this scheduler only
    needs *some* function from ``BuildContext`` to ``BuildResult`` to prove
    the level-parallel + serial-promotion machinery is sound. Tests pass
    real small functions (closures over whatever state they need), never a
    mock/stub/double.
    """

    task_id: str
    requirement_id: str
    declared_paths: tuple[str, ...]
    depends_on: tuple[str, ...]
    build: Callable[[BuildContext], BuildResult]


class CyclicDependency(Exception):
    """Raised by :func:`build_levels` when the DAG contains a cycle.

    ``members`` names every task that never reached zero in-degree during
    Kahn's algorithm -- i.e. every task that is itself on a cycle, or
    depends (transitively) on one. Sorted for deterministic messages.
    """

    def __init__(self, members: Sequence[str]) -> None:
        self.members = sorted(members)
        super().__init__(f"cyclic dependency among tasks: {self.members}")


def build_levels(tasks: Sequence[TaskSpec]) -> list[list[TaskSpec]]:
    """Kahn topological levels: each level's tasks may dispatch concurrently.

    Order within a level is deterministic (sorted by ``task_id``) so
    dispatch order never depends on input order or dict/set iteration.

    Raises:
        ValueError: A task's ``depends_on`` names an unknown ``task_id``.
        CyclicDependency: The DAG contains a cycle (:attr:`CyclicDependency.members`
            names every task that never reached zero in-degree).
    """
    by_id = {t.task_id: t for t in tasks}
    indegree = {t.task_id: 0 for t in tasks}
    dependents: dict[str, list[str]] = {t.task_id: [] for t in tasks}
    for task in tasks:
        for dep in task.depends_on:
            if dep not in by_id:
                raise ValueError(
                    f"task {task.task_id!r} depends on unknown task {dep!r}"
                )
            indegree[task.task_id] += 1
            dependents[dep].append(task.task_id)

    levels: list[list[TaskSpec]] = []
    remaining = set(by_id)
    while remaining:
        ready = sorted(tid for tid in remaining if indegree[tid] == 0)
        if not ready:
            raise CyclicDependency(sorted(remaining))
        levels.append([by_id[tid] for tid in ready])
        for tid in ready:
            remaining.discard(tid)
            for dependent in dependents[tid]:
                indegree[dependent] -= 1

    return levels