scripts.factory_pipeline

Reference › scripts › factory_pipeline

The command-line interface for the SPEC-033 section 9 DAG scheduler and merge-queue pipeline (B9). It connects through redis_factory, uses the conformance gate in conformance_gate and keeps its state in a PromotionStore; the factory narrative describes the subsystem. No module or test imports it.

Invocation: main has one subcommand, run (_cmd_run, with its parser in _add_run_subparser). It loads a file of tasks (_load_tasks), schedules them as a directed acyclic graph, builds each by running a shell command under a timeout (_make_shell_build, _build) and feeds the results through a merge queue, with a re-gate step before each merge (_default_regate). The outcome is a PipelineReport, printed as JSON (_report_to_dict). The default ledger location comes from _default_ledger. Any exception is printed as a JSON error on standard error and the exit code is 1.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

API

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/scripts/factory_pipeline.py

#!/usr/bin/env python3
"""CLI for the SPEC-033 §9 DAG scheduler + merge queue pipeline (B9).

Subcommands
-----------
run   Load a task DAG from JSON, run it through
      :func:`orchestrata_ultimate.factory.scheduler.scheduler.run_pipeline`
      (level-parallel dispatch, serial integration re-gate + promotion),
      and print the resulting ``PipelineReport`` as JSON.

Tasks-file shape
----------------
::

    {
      "tasks": [
        {
          "task_id": "T1",
          "requirement_id": "REQ-1",
          "declared_paths": ["a.py"],
          "depends_on": [],
          "build_command": "echo building",
          "timeout_s": 300
        }
      ]
    }

Each task's ``build_command`` is a shell command run via ``subprocess``
with ``cwd`` set to the task's leased worktree (``shell=True``, so pipes/
redirection/&& work as in a normal shell build step) -- the real builder
driver this ``build`` callable stands in for is future work (see
``factory/scheduler/dag.py``'s ``TaskSpec`` docstring). A non-zero exit or
a timeout is recorded as ``build_failed``. On success, ONE
``FactoryBuildResult`` fact is written to the task's own namespace
(command + exit code + a truncated stdout snippet) via
``PromotionStore.write_fact`` -- ``PromotionStore.promote`` refuses to
promote an empty namespace (fails loudly, by design), so this is the
minimal real fact every promoted task needs; it is not a placeholder for
the real AST/trace fact-writing a landed builder integration would do.

The integration re-gate this CLI wires in (``_default_regate``) requires
the task's namespace to have written at least one fact
(``PromotionStore.promote`` refuses an empty namespace -- fails loudly, by
design), then runs the SPEC-033 Phase 6 pre-merge conformance gate
(``factory.dispatch.conformance_gate.conformance_regate``, see its
docstring): the worktree's own ``git diff`` since ``--base`` must stay
inside the task's ``declared_paths``, and every changed ``.py`` file must
pass ``scripts/code-review.py`` + ``scripts/test-review.py``. Either check
failing is a typed, auto-routed rejection (``regate_rejected``) -- no human
step. What this does NOT do is the full AST≡design structural conformance
check (``factory/conformance/engine.py``, B2) or any semantic "does trunk-
plus-this-change still satisfy the design" judgement -- those remain
separate lanes (ADR-0008: structure/scope, never intent).

Exit code: 0 iff every task's outcome is ``"promoted"``; 1 otherwise (or on
a CLI-level error).

Examples
--------
    python scripts/factory_pipeline.py run --tasks-file tasks.json \\
        --base HEAD --repo-root . --max-parallel 4
"""

from __future__ import annotations

import argparse
import dataclasses
import json
import subprocess
import sys
from pathlib import Path
from typing import Any

# Ensure repo src is on path when run directly (mirrors factory_enforce.py /
# factory_promote.py / factory_dispatch.py).
_REPO_ROOT = Path(__file__).resolve().parent.parent
_REPO_SRC = _REPO_ROOT / "src"
if str(_REPO_SRC) not in sys.path:
    sys.path.insert(0, str(_REPO_SRC))

from orchestrata_ultimate.factory.dispatch.conformance_gate import (  # noqa: E402
    conformance_regate,
)
from orchestrata_ultimate.factory.promotion.store import PromotionStore  # noqa: E402
from orchestrata_ultimate.factory.scheduler.dag import (  # noqa: E402
    BuildContext,
    BuildResult,
    TaskSpec,
)
from orchestrata_ultimate.factory.scheduler.queue import (  # noqa: E402
    MergeEntry,
    RegateVerdict,
)
from orchestrata_ultimate.factory.scheduler.scheduler import (  # noqa: E402
    PipelineReport,
    run_pipeline,
)

# Cap on the stdout snippet recorded in the FactoryBuildResult fact -- the
# fact is evidence a build ran, not a log store.
_STDOUT_SNIPPET_LIMIT = 2000

# The project's own review tools (Phase 6 conformance gate) -- always the
# real installation's copies, resolved from this script's own location, not
# from inside a task's throwaway worktree.
_CODE_REVIEW_SCRIPT = _REPO_ROOT / "scripts" / "code-review.py"
_TEST_REVIEW_SCRIPT = _REPO_ROOT / "scripts" / "test-review.py"


def _default_ledger(repo_root: Path) -> Path:
    return repo_root / "factory_ledger.jsonl"


def _make_shell_build(store: PromotionStore, command: str, timeout_s: float) -> Any:
    """Build a real ``TaskSpec.build`` callable that runs ``command`` via shell.

    On a zero exit, writes ONE ``FactoryBuildResult`` fact to the task's own
    namespace -- see the module docstring for why this is required, not
    decorative.
    """

    def _build(ctx: BuildContext) -> BuildResult:
        try:
            proc = subprocess.run(
                command,
                shell=True,
                cwd=ctx.lease.path,
                capture_output=True,
                text=True,
                timeout=timeout_s,
            )
        except subprocess.TimeoutExpired as exc:
            return BuildResult(success=False, detail=f"timeout after {exc.timeout}s")
        if proc.returncode != 0:
            return BuildResult(
                success=False,
                detail=(proc.stderr or proc.stdout or f"exit code {proc.returncode}"),
            )
        store.write_fact(
            ctx.namespace,
            "FactoryBuildResult",
            {
                "task_id": ctx.task_id,
                "command": command,
                "exit_code": proc.returncode,
                "stdout": proc.stdout[:_STDOUT_SNIPPET_LIMIT],
            },
        )
        return BuildResult(success=True, detail=proc.stdout)

    return _build


def _default_regate(store: PromotionStore, entry: MergeEntry) -> RegateVerdict:
    """The real pre-merge conformance gate -- see module docstring and
    ``factory.dispatch.conformance_gate`` (Phase 6 gap closure)."""
    if not store.facts_in(entry.lease.namespace):
        return RegateVerdict(passed=False, reason="no facts to promote")
    return conformance_regate(
        entry,
        code_review_script=_CODE_REVIEW_SCRIPT,
        test_review_script=_TEST_REVIEW_SCRIPT,
    )


def _load_tasks(
    store: PromotionStore, tasks_file: Path, build_timeout_s: float
) -> list[TaskSpec]:
    payload = json.loads(tasks_file.read_text())
    specs: list[TaskSpec] = []
    for raw in payload["tasks"]:
        timeout_s = float(raw.get("timeout_s", build_timeout_s))
        specs.append(
            TaskSpec(
                task_id=raw["task_id"],
                requirement_id=raw["requirement_id"],
                declared_paths=tuple(raw.get("declared_paths", [])),
                depends_on=tuple(raw.get("depends_on", [])),
                build=_make_shell_build(store, raw["build_command"], timeout_s),
            )
        )
    return specs


def _report_to_dict(report: PipelineReport) -> dict[str, Any]:
    return dataclasses.asdict(report)


def _cmd_run(args: argparse.Namespace) -> int:
    from orchestrata_ultimate.core.redis_factory import create_sync_redis
    from orchestrata_ultimate.factory.dispatch.gates import InFlightRegistry
    from orchestrata_ultimate.factory.enforcement.ledger import EventLedger
    from orchestrata_ultimate.factory.promotion.models import RetentionPolicy

    repo_root = Path(args.repo_root).resolve()
    tasks_file = Path(args.tasks_file)
    ledger_path = Path(args.ledger) if args.ledger else _default_ledger(repo_root)

    redis_client = create_sync_redis(
        host=args.redis_host or None,
        port=args.redis_port or None,
        db=args.redis_db or None,
    )
    ledger = EventLedger(ledger_path)
    store = PromotionStore(
        ledger,
        uri=args.uri or None,
        username=args.user or None,
        password=args.password or None,
    )
    registry = InFlightRegistry(redis_client)

    try:
        tasks = _load_tasks(store, tasks_file, args.build_timeout_s)
        report = run_pipeline(
            repo_root,
            tasks,
            registry=registry,
            store=store,
            regate=lambda entry: _default_regate(store, entry),
            base_commit=args.base,
            max_parallel=args.max_parallel,
            retention_policy=RetentionPolicy(
                max_age_seconds=args.retention_max_age_seconds,
                keep_last_n=args.retention_keep_last_n,
            ),
            dispatcher_identity=args.dispatcher_identity,
        )
    finally:
        store.close()
        redis_client.close()

    print(json.dumps(_report_to_dict(report), indent=2))

    all_promoted = all(t.outcome == "promoted" for t in report.task_reports)
    return 0 if all_promoted else 1


def _add_run_subparser(sub: argparse._SubParsersAction) -> None:
    run_p = sub.add_parser(
        "run", help="Run a task DAG through the scheduler + merge queue."
    )
    run_p.add_argument("--tasks-file", required=True)
    run_p.add_argument("--base", required=True, help="Base commit SHA for all leases.")
    run_p.add_argument("--repo-root", default=".")
    run_p.add_argument("--max-parallel", type=int, default=4)
    run_p.add_argument("--build-timeout-s", type=float, default=300.0)
    run_p.add_argument("--ledger", default=None)
    run_p.add_argument("--dispatcher-identity", default="factory-dispatch-cli")
    run_p.add_argument("--retention-max-age-seconds", type=int, default=86400)
    run_p.add_argument("--retention-keep-last-n", type=int, default=5)
    run_p.add_argument(
        "--uri", default=None, help="Graph URI (default: $KG_ROUTING_URI/$KG_URI)."
    )
    run_p.add_argument(
        "--user", default=None, help="Graph username (default: empty — no auth)."
    )
    run_p.add_argument(
        "--password", default=None, help="Graph password (default: empty — no auth)."
    )
    run_p.add_argument(
        "--redis-host", default=None, help="Redis host (default: $REDIS_HOST)."
    )
    run_p.add_argument(
        "--redis-port",
        type=int,
        default=None,
        help="Redis port (default: $REDIS_PORT).",
    )
    run_p.add_argument(
        "--redis-db", type=int, default=None, help="Redis DB (default: $REDIS_DB)."
    )


def _build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="factory_pipeline",
        description="SPEC-033 §9 DAG scheduler + merge queue CLI (B9).",
    )
    sub = parser.add_subparsers(dest="command", required=True)
    _add_run_subparser(sub)
    return parser


def main(argv: list[str] | None = None) -> int:
    """Parse ``argv`` and dispatch to the matching subcommand handler."""
    parser = _build_parser()
    args = parser.parse_args(argv)

    dispatch = {"run": _cmd_run}
    handler = dispatch[args.command]
    try:
        return handler(args)
    except Exception as exc:  # noqa: BLE001 -- CLI boundary: report, never crash
        print(json.dumps({"error": str(exc)}), file=sys.stderr)
        return 1


if __name__ == "__main__":
    sys.exit(main())