scripts.factory_dispatch

Reference › scripts › factory_dispatch

The command-line interface for the SPEC-033 B21 dispatch gates (ADR-0033, ADR-0018 and ADR-0032). It wraps the in-flight registry in gates, which stores its claims in Redis through redis_factory; the factory narrative describes the subsystem. No module or test imports it.

Invocation: main has three subcommands, all of which accept the Redis connection options added by _add_redis_args. claim registers a unit of work as in flight (_cmd_claim) and release removes the claim (_cmd_release). check-stale looks for claims that have gone stale (_cmd_check_stale). _make_registry builds the InFlightRegistry from the options. When a gate refuses, _refusal_payload turns the exception into a structured payload that is printed.

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_dispatch.py

#!/usr/bin/env python3
"""CLI for the SPEC-033 B21 dispatch gates (ADR-0033/0018/0032).

Subcommands
-----------
claim        Attempt to claim an in-flight task key (SET NX EX; the claim
             IS the registration). Non-zero exit + a ``DuplicateDispatch``
             payload if the key is already held.
release      Release a held claim (compare-and-delete; releasing as a
             non-holder is a silent no-op, exit 0 either way).
check-stale  Check whether trunk has moved on a declared path since a base
             commit (``git diff --name-only``). Non-zero exit + a
             ``StaleDispatch`` payload if stale or the base commit is
             unresolvable.

Redis connection defaults to ``REDIS_HOST``/``REDIS_PORT``/``REDIS_DB``
environment variables (see ``core.redis_factory.create_sync_redis``);
``--redis-host``/``--redis-port``/``--redis-db`` override them.

Examples
--------
    python scripts/factory_dispatch.py claim --requirement R1 \\
        --paths a.py,b/ --holder H1 --ttl 3600
    python scripts/factory_dispatch.py release --key "R1:abcdef0123456789" \\
        --holder H1
    python scripts/factory_dispatch.py check-stale --base <sha> \\
        --paths a.py,b/ --repo-root .
"""

from __future__ import annotations

import argparse
import json
import sys
from pathlib import Path

# Ensure repo src is on path when run directly (mirrors factory_enforce.py /
# factory_promote.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.gates import (  # noqa: E402
    DuplicateDispatch,
    InFlightRegistry,
    StaleDispatch,
    duplication_gate,
    staleness_gate,
    task_key,
)


def _make_registry(args: argparse.Namespace) -> InFlightRegistry:
    from orchestrata_ultimate.core.redis_factory import create_sync_redis

    client = create_sync_redis(
        host=getattr(args, "redis_host", None) or None,
        port=getattr(args, "redis_port", None) or None,
        db=getattr(args, "redis_db", None) or None,
    )
    return InFlightRegistry(client)


def _refusal_payload(exc: Exception) -> dict:
    return {"error": str(exc), "exception": type(exc).__name__}


def _cmd_claim(args: argparse.Namespace) -> int:
    registry = _make_registry(args)
    key = task_key(args.requirement, args.paths.split(","))
    try:
        duplication_gate(registry, key, args.holder, ttl_s=args.ttl)
    except DuplicateDispatch as exc:
        payload = _refusal_payload(exc)
        payload["task_key"] = key
        print(json.dumps(payload), file=sys.stderr)
        return 1
    print(json.dumps({"task_key": key, "holder": args.holder, "claimed": True}))
    return 0


def _cmd_release(args: argparse.Namespace) -> int:
    registry = _make_registry(args)
    registry.release(args.key, args.holder)
    print(json.dumps({"task_key": args.key, "released_by": args.holder}))
    return 0


def _cmd_check_stale(args: argparse.Namespace) -> int:
    repo_root = Path(args.repo_root)
    declared_paths = args.paths.split(",")
    try:
        staleness_gate(repo_root, args.base, declared_paths)
    except StaleDispatch as exc:
        print(json.dumps(_refusal_payload(exc)), file=sys.stderr)
        return 1
    print(json.dumps({"stale": False, "base": args.base}))
    return 0


def _add_redis_args(parser: argparse.ArgumentParser) -> None:
    parser.add_argument(
        "--redis-host", default=None, help="Redis host (default: $REDIS_HOST)."
    )
    parser.add_argument(
        "--redis-port",
        type=int,
        default=None,
        help="Redis port (default: $REDIS_PORT).",
    )
    parser.add_argument(
        "--redis-db", type=int, default=None, help="Redis DB (default: $REDIS_DB)."
    )


def _add_claim_subparser(sub: argparse._SubParsersAction) -> None:
    claim_p = sub.add_parser("claim", help="Claim an in-flight task key.")
    claim_p.add_argument("--requirement", required=True)
    claim_p.add_argument("--paths", required=True, help="Comma-separated unit paths.")
    claim_p.add_argument("--holder", required=True)
    claim_p.add_argument("--ttl", type=int, default=3600)
    _add_redis_args(claim_p)


def _add_release_subparser(sub: argparse._SubParsersAction) -> None:
    release_p = sub.add_parser("release", help="Release a held claim.")
    release_p.add_argument("--key", required=True)
    release_p.add_argument("--holder", required=True)
    _add_redis_args(release_p)


def _add_check_stale_subparser(sub: argparse._SubParsersAction) -> None:
    stale_p = sub.add_parser(
        "check-stale", help="Check dispatch staleness against trunk."
    )
    stale_p.add_argument("--base", required=True)
    stale_p.add_argument(
        "--paths", required=True, help="Comma-separated declared paths."
    )
    stale_p.add_argument("--repo-root", default=".")


def _build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="factory_dispatch",
        description="SPEC-033 B21 dispatch gates CLI (ADR-0033/0018/0032).",
    )
    sub = parser.add_subparsers(dest="command", required=True)
    _add_claim_subparser(sub)
    _add_release_subparser(sub)
    _add_check_stale_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 = {
        "claim": _cmd_claim,
        "release": _cmd_release,
        "check-stale": _cmd_check_stale,
    }
    handler = dispatch[args.command]
    return handler(args)


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