scripts.factory_promote

Reference › scripts › factory_promote

The command-line interface for the SPEC-033 section 4 fact-namespace promotion and garbage-collection store (B4a; ADR-0004 and ADR-0019). It uses the enforcement ledger in ledger and the promotion models in models; the factory narrative describes the subsystem. No module or test imports it.

Invocation: main takes --repo-root, the connection options added by _add_connection_args, and a subcommand. promote promotes facts from one namespace to another (_cmd_promote). gc removes facts that are no longer needed (_cmd_gc). list lists the store's contents (_cmd_list). _make_store builds the PromotionStore, and main always closes it in a finally block. An 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_promote.py

#!/usr/bin/env python3
"""CLI for the SPEC-033 §4 fact-namespace promotion + GC store (B4a, ADR-0004/0019).

Subcommands
-----------
promote  Promote a `(worktree, commit)` namespace's facts into trunk,
         recording a governed event in the B1c ledger. Fails loudly (non-
         zero exit) if the namespace is unknown/empty or `--gate-token` is
         empty.
list     Print every currently-known namespace (worktree, commit, key,
         created_at) as JSON.
gc       Delete namespaces for worktrees no longer live, applying the dual
         age/keep-last-N retention bound (ADR-0004). Trunk is never
         GC-eligible.

Path defaults: ``--ledger`` -> ``<repo-root>/factory_ledger.jsonl`` (mirrors
``scripts/factory_enforce.py``). graph connection defaults to the
``KG_ROUTING_URI``/``KG_URI`` environment variables;
``--uri``/``--user``/``--password`` override them.

Examples
--------
    python scripts/factory_promote.py promote --worktree /repo/wt-a \\
        --commit abc123 --gate-token GT-xyz
    python scripts/factory_promote.py gc --max-age-s 86400 --keep-last 3 \\
        --live-worktrees /repo/wt-a,/repo/wt-b
    python scripts/factory_promote.py list
"""

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_c4.py / kg_schema.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.enforcement.ledger import EventLedger  # noqa: E402
from orchestrata_ultimate.factory.promotion.models import (  # noqa: E402
    FactNamespace,
    RetentionPolicy,
)
from orchestrata_ultimate.factory.promotion.store import PromotionStore  # noqa: E402


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


def _make_store(args: argparse.Namespace, repo_root: Path) -> PromotionStore:
    ledger_path = Path(args.ledger) if args.ledger else _default_ledger(repo_root)
    ledger = EventLedger(ledger_path)
    return PromotionStore(
        ledger,
        uri=getattr(args, "uri", None) or None,
        username=getattr(args, "user", None) or None,
        password=getattr(args, "password", None) or None,
    )


def _cmd_promote(args: argparse.Namespace, store: PromotionStore) -> int:
    ns = FactNamespace(worktree=args.worktree, commit=args.commit)
    record = store.promote(ns, gate_token=args.gate_token)
    print(
        json.dumps(
            {
                "namespace": record.namespace.key,
                "promoted_at": record.promoted_at,
                "commit": record.commit,
                "fact_count": record.fact_count,
                "ledger_event_id": record.ledger_event_id,
                "watermark": record.watermark,
            },
            indent=2,
        )
    )
    return 0


def _cmd_gc(args: argparse.Namespace, store: PromotionStore) -> int:
    policy = RetentionPolicy(max_age_seconds=args.max_age_s, keep_last_n=args.keep_last)
    live_worktrees = [w for w in (args.live_worktrees or "").split(",") if w]
    deleted = store.gc(policy, live_worktrees)
    print(
        json.dumps(
            {
                "deleted": [
                    {"worktree": ns.worktree, "commit": ns.commit} for ns in deleted
                ]
            },
            indent=2,
        )
    )
    return 0


def _cmd_list(_args: argparse.Namespace, store: PromotionStore) -> int:
    print(json.dumps(store.list_namespaces(), indent=2, default=str))
    return 0


def _add_connection_args(parser: argparse.ArgumentParser) -> None:
    parser.add_argument("--repo-root", default=".")
    parser.add_argument("--ledger", default=None)
    parser.add_argument(
        "--uri", default=None, help="Graph URI (default: $KG_ROUTING_URI/$KG_URI)."
    )
    parser.add_argument(
        "--user", default=None, help="Graph username (default: empty — no auth)."
    )
    parser.add_argument(
        "--password", default=None, help="Graph password (default: empty — no auth)."
    )


def _build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="factory_promote",
        description="SPEC-033 §4 fact-namespace promotion + GC CLI (B4a).",
    )
    sub = parser.add_subparsers(dest="command", required=True)

    promote_p = sub.add_parser(
        "promote", help="Promote a (worktree, commit) namespace's facts into trunk."
    )
    promote_p.add_argument("--worktree", required=True)
    promote_p.add_argument("--commit", required=True)
    promote_p.add_argument("--gate-token", required=True)
    _add_connection_args(promote_p)

    gc_p = sub.add_parser(
        "gc", help="Delete abandoned-worktree namespaces per the retention policy."
    )
    gc_p.add_argument("--max-age-s", type=int, required=True)
    gc_p.add_argument("--keep-last", type=int, required=True)
    gc_p.add_argument(
        "--live-worktrees",
        default="",
        help="Comma-separated list of live worktree paths (from `git worktree list`).",
    )
    _add_connection_args(gc_p)

    list_p = sub.add_parser("list", help="Print every currently-known namespace.")
    _add_connection_args(list_p)

    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)

    repo_root = Path(args.repo_root)
    dispatch = {
        "promote": _cmd_promote,
        "gc": _cmd_gc,
        "list": _cmd_list,
    }
    handler = dispatch[args.command]

    store = _make_store(args, repo_root)
    try:
        return handler(args, store)
    except Exception as exc:  # noqa: BLE001
        print(json.dumps({"error": str(exc)}), file=sys.stderr)
        return 1
    finally:
        store.close()


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