scripts.factory_assumptions

Reference › scripts › factory_assumptions

The command-line interface for the SPEC-033 assumption ledger and arbiter cascade (B13, Phase 0). It reads anchors through the sources in source and store; the factory narrative describes the subsystem. No module or test imports it.

Invocation: main takes a subcommand and dispatches to its handler: add (record an assumption), disclose and disclosures (record and list disclosures), resolve (run the pinning cascade over the available sources, built by _build_sources), adjudicate, dor and list. The ledger is a knowledge-base directory: _resolve_kb_dir takes it from the command line or, failing that, from the environment. Each handler returns the process exit code.

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

#!/usr/bin/env python3
"""CLI for the SPEC-033 assumption ledger + arbiter cascade (B13, Phase 0).

Subcommands
-----------
add           Enumerate a new assumption into the ledger.
disclose      Record a Fable interpretation disclosure (AC-033-C1-03).
disclosures   Print interpretation disclosures, optionally filtered by node.
resolve       Run the arbiter cascade over every PENDING assumption.
adjudicate    Stage 3 — human adjudication + KB promotion.
dor           AC-033-C1-02 — Definition-of-Ready check over the ledger.
list          Print every assumption in the ledger.

KB directory resolution (``resolve``/``adjudicate``): ``--kb`` flag >
``FACTORY_KB_DIR`` env var > ``<repo-root>/factory_kb``.

Examples
--------
    python scripts/factory_assumptions.py add --ledger ledger.jsonl \\
        --text "the queue is FIFO" --design-node "N3" --author "sonnet-5:builder"
    python scripts/factory_assumptions.py resolve --ledger ledger.jsonl
    python scripts/factory_assumptions.py dor --ledger ledger.jsonl
"""

from __future__ import annotations

import argparse
import asyncio
import os
import sys
from collections.abc import Mapping
from pathlib import Path
from typing import TYPE_CHECKING

_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))

if TYPE_CHECKING:
    from orchestrata_ultimate.factory.assumptions.ledger import KBProtocol
    from orchestrata_ultimate.factory.assumptions.sources import PinningSource


def _resolve_kb_dir(
    cli_value: str | None, env: Mapping[str, str] | None = None
) -> Path:
    """Resolve the KB directory: ``--kb`` > ``FACTORY_KB_DIR`` > repo default.

    ``env`` defaults to ``os.environ`` (real process env); tests pass an
    explicit mapping instead of mutating the process environment.
    """
    if cli_value:
        return Path(cli_value)
    env = env if env is not None else os.environ
    env_value = env.get("FACTORY_KB_DIR")
    if env_value:
        return Path(env_value)
    return _REPO_ROOT / "factory_kb"


def _cmd_add(args: argparse.Namespace) -> int:
    from orchestrata_ultimate.factory.assumptions.ledger import AssumptionLedger
    from orchestrata_ultimate.factory.assumptions.models import Assumption

    ledger = AssumptionLedger(Path(args.ledger))
    assumption = Assumption.new(
        text=args.text,
        design_node=args.design_node,
        author_identity=args.author,
        requirement_id=args.requirement,
    )
    ledger.add(assumption)
    print(assumption.id)
    return 0


def _cmd_disclose(args: argparse.Namespace) -> int:
    from datetime import UTC, datetime

    from orchestrata_ultimate.factory.assumptions.ledger import AssumptionLedger
    from orchestrata_ultimate.factory.assumptions.models import (
        InterpretationDisclosure,
    )

    ledger = AssumptionLedger(Path(args.ledger))
    disclosure = InterpretationDisclosure(
        design_node=args.design_node,
        chosen_reading=args.chosen,
        alternatives=list(args.alt or []),
        author_identity=args.author,
        created_at=datetime.now(UTC).isoformat(),
    )
    ledger.add_disclosure(disclosure)
    print(f"disclosed: {disclosure.design_node}")
    return 0


def _cmd_disclosures(args: argparse.Namespace) -> int:
    from orchestrata_ultimate.factory.assumptions.ledger import AssumptionLedger

    ledger = AssumptionLedger(Path(args.ledger))
    for d in ledger.disclosures(design_node=args.design_node):
        print(
            f"{d.design_node}: chosen={d.chosen_reading!r} "
            f"alternatives={d.alternatives}"
        )
        print(f"  author={d.author_identity} at={d.created_at}")
    return 0


def _build_sources(kb: KBProtocol, anchors_path: Path) -> list[PinningSource]:
    """Assemble the arbiter cascade's stage-1 pinning sources.

    AC/ADR/KB (B13) always run. ``AnchorSource`` (B15a) is added ONLY when
    ``anchors_path`` exists -- an anchors file is optional (not every project
    has authored/ratified any anchors yet), and a PROPOSED-only or absent
    registry would just yield an empty candidate pool anyway. Ratified
    anchors then become stage-1 candidates alongside AC/ADR/KB pinnings.
    """
    from orchestrata_ultimate.factory.assumptions.sources import (
        AcSource,
        AdrSource,
        KbSource,
    )

    sources: list[PinningSource] = [
        AcSource(_REPO_ROOT / "specs"),
        AdrSource(_REPO_ROOT / "docs" / "adr"),
        KbSource(kb),
    ]
    if anchors_path.exists():
        from orchestrata_ultimate.factory.anchors.source import AnchorSource
        from orchestrata_ultimate.factory.anchors.store import AnchorRegistry

        sources.append(AnchorSource(AnchorRegistry(anchors_path)))
    return sources


def _cmd_resolve(args: argparse.Namespace) -> int:
    from orchestrata_ultimate.factory.assumptions.adjudicator import LlmAdjudicator
    from orchestrata_ultimate.factory.assumptions.cascade import ArbiterCascade
    from orchestrata_ultimate.factory.assumptions.ledger import (
        AssumptionLedger,
        FileProjectKB,
    )

    ledger = AssumptionLedger(Path(args.ledger))
    kb_dir = _resolve_kb_dir(args.kb)
    kb = FileProjectKB(kb_dir)
    sources = _build_sources(kb, kb_dir / "anchors.json")
    adjudicator = (
        LlmAdjudicator(args.adjudicator_profile) if args.adjudicator_profile else None
    )
    cascade = ArbiterCascade(sources, kb, adjudicator)
    outcomes = asyncio.run(cascade.resolve_all(ledger))
    for outcome in outcomes:
        if outcome.status.value == "pinned" and outcome.pinning is not None:
            detail = f"pinning={outcome.pinning.kind.value}:{outcome.pinning.ref}"
        elif outcome.questions:
            detail = f"questions={outcome.questions}"
        else:
            detail = "questions=[]"
        print(
            f"{outcome.assumption_id} {outcome.status.value} "
            f"stage={outcome.stage} {detail}"
        )
    return 0


def _cmd_adjudicate(args: argparse.Namespace) -> int:
    from orchestrata_ultimate.factory.assumptions.cascade import human_adjudicate
    from orchestrata_ultimate.factory.assumptions.ledger import (
        AssumptionLedger,
        FileProjectKB,
    )
    from orchestrata_ultimate.factory.assumptions.models import (
        Pinning,
        PinningKind,
        Ruling,
    )

    ledger = AssumptionLedger(Path(args.ledger))
    kb = FileProjectKB(_resolve_kb_dir(args.kb))
    pinning = None
    if args.pinning_kind and args.pinning_ref:
        pinning = Pinning(
            kind=PinningKind(args.pinning_kind.lower()),
            ref=args.pinning_ref,
            excerpt="",
        )
    adjudication = human_adjudicate(
        ledger=ledger,
        kb=kb,
        assumption_id=args.id,
        ruling=Ruling(args.ruling.lower()),
        pinning=pinning,
        by=args.by,
    )
    print(adjudication.id)
    return 0


def _cmd_dor(args: argparse.Namespace) -> int:
    from orchestrata_ultimate.factory.assumptions.dor import check_dor
    from orchestrata_ultimate.factory.assumptions.ledger import AssumptionLedger

    ledger = AssumptionLedger(Path(args.ledger))
    result = check_dor(ledger)
    if result.ready:
        print("DoR: ready")
        return 0
    print("DoR: NOT ready")
    for assumption_id, design_node, status in result.blocking:
        print(f"  blocking: {assumption_id} design_node={design_node} status={status}")
    return 1


def _cmd_list(args: argparse.Namespace) -> int:
    from orchestrata_ultimate.factory.assumptions.ledger import AssumptionLedger

    ledger = AssumptionLedger(Path(args.ledger))
    for assumption in ledger.assumptions():
        pinning_desc = (
            f"{assumption.pinning.kind.value}:{assumption.pinning.ref}"
            if assumption.pinning is not None
            else "-"
        )
        print(
            f"{assumption.id} {assumption.status.value} "
            f"design_node={assumption.design_node} pinning={pinning_desc} "
            f"stage={assumption.stage_resolved} text={assumption.text!r}"
        )
    return 0


def _build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="factory_assumptions",
        description="SPEC-033 assumption ledger + arbiter cascade CLI (B13).",
    )
    sub = parser.add_subparsers(dest="command", required=True)

    add_p = sub.add_parser("add", help="Enumerate a new assumption into the ledger.")
    add_p.add_argument("--ledger", required=True)
    add_p.add_argument("--text", required=True)
    add_p.add_argument("--design-node", required=True)
    add_p.add_argument("--author", required=True)
    add_p.add_argument("--requirement", default=None)

    disclose_p = sub.add_parser(
        "disclose", help="Record a Fable interpretation disclosure."
    )
    disclose_p.add_argument("--ledger", required=True)
    disclose_p.add_argument("--design-node", required=True)
    disclose_p.add_argument("--chosen", required=True)
    disclose_p.add_argument("--alt", action="append", default=[])
    disclose_p.add_argument("--author", required=True)

    disclosures_p = sub.add_parser(
        "disclosures", help="Print interpretation disclosures."
    )
    disclosures_p.add_argument("--ledger", required=True)
    disclosures_p.add_argument("--design-node", default=None)

    resolve_p = sub.add_parser(
        "resolve", help="Run the arbiter cascade over every PENDING assumption."
    )
    resolve_p.add_argument("--ledger", required=True)
    resolve_p.add_argument("--kb", default=None)
    resolve_p.add_argument("--adjudicator-profile", default=None)

    adjudicate_p = sub.add_parser(
        "adjudicate", help="Stage 3 -- human adjudication + KB promotion."
    )
    adjudicate_p.add_argument("--ledger", required=True)
    adjudicate_p.add_argument("--id", required=True)
    adjudicate_p.add_argument(
        "--ruling", required=True, choices=["matched", "unpinned"]
    )
    adjudicate_p.add_argument("--pinning-kind", default=None)
    adjudicate_p.add_argument("--pinning-ref", default=None)
    adjudicate_p.add_argument("--by", required=True)
    adjudicate_p.add_argument("--kb", default=None)

    dor_p = sub.add_parser("dor", help="AC-033-C1-02 Definition-of-Ready check.")
    dor_p.add_argument("--ledger", required=True)

    list_p = sub.add_parser("list", help="Print every assumption in the ledger.")
    list_p.add_argument("--ledger", required=True)

    return parser


def main(argv: list[str] | None = None) -> int:
    parser = _build_parser()
    args = parser.parse_args(argv)

    dispatch = {
        "add": _cmd_add,
        "disclose": _cmd_disclose,
        "disclosures": _cmd_disclosures,
        "resolve": _cmd_resolve,
        "adjudicate": _cmd_adjudicate,
        "dor": _cmd_dor,
        "list": _cmd_list,
    }
    handler = dispatch[args.command]
    return handler(args)


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