scripts.factory_detectors

Reference › scripts › factory_detectors

The command-line interface for the SPEC-033 C3 lexical and structural detector set (B22; ADR-0029). It uses the assumption ledger from ledger and the C1 bridge in the detectors package; the factory narrative describes the subsystem. No module or test imports it.

Invocation: main parses the arguments and calls the handler of the chosen subcommand. check scans files or directories (_collect_files) against the rule table and prints the findings, as text or as JSON (_scan_files, _print_check_report), and _apply_tier1_gate turns the most serious findings into a failing exit code. dismiss records a finding as dismissed in a blacklist. rules lists the rule table. feed-c1 passes findings to the C1 bridge, which feeds the assumption ledger.

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

#!/usr/bin/env python3
"""CLI for the SPEC-033 C3 lexical/structural detector set (B22, ADR-0029).

    python scripts/factory_detectors.py check \\
        specs/033-autonomous-build-factory/spec.md
    python scripts/factory_detectors.py check specs/ --tier1-gate
    python scripts/factory_detectors.py rules
    python scripts/factory_detectors.py dismiss \\
        --blacklist .detectors_blacklist.jsonl --fingerprint <fp>
    python scripts/factory_detectors.py feed-c1 specs/033-.../spec.md \\
        --ledger .assumptions.jsonl --design-node "design:foo"
"""

from __future__ import annotations

import argparse
import json
import sys
from pathlib import Path
from typing import TYPE_CHECKING

_REPO_ROOT = Path(__file__).resolve().parent.parent
if str(_REPO_ROOT) not in sys.path:
    sys.path.insert(0, str(_REPO_ROOT))

if TYPE_CHECKING:
    from orchestrata_ultimate.factory.detectors.engine import Blacklist
    from orchestrata_ultimate.factory.detectors.models import Finding
    from orchestrata_ultimate.factory.detectors.rules import RuleFn, RuleMeta

    RuleTable = dict[str, tuple[RuleMeta, RuleFn]]


def _collect_files(paths: list[str]) -> list[Path]:
    """Expand ``paths`` -- directories scan ``*.md`` recursively, files pass through."""
    files: list[Path] = []
    for raw in paths:
        p = Path(raw)
        if p.is_dir():
            files.extend(sorted(p.rglob("*.md")))
        elif p.is_file():
            files.append(p)
    return files


def _finding_to_dict(finding: Finding) -> dict:
    return {
        "rule_id": finding.rule_id,
        "tier": finding.tier.value,
        "line": finding.line,
        "span": list(finding.span),
        "matched_text": finding.matched_text,
        "message": finding.message,
        "weight": finding.weight,
        "fingerprint": finding.fingerprint,
        "question": finding.question,
    }


def _print_findings(findings: list[Finding]) -> None:
    from orchestrata_ultimate.factory.detectors.models import Tier

    if not findings:
        print("no findings")
        return
    for tier in (Tier.BLOCKING, Tier.ADVISORY, Tier.QUESTION):
        tier_findings = [f for f in findings if f.tier == tier]
        if not tier_findings:
            continue
        print(f"\n== {tier.value.upper()} ({len(tier_findings)}) ==")
        for f in tier_findings:
            extra = f" -- {f.question}" if f.question else ""
            print(
                f"  [{f.rule_id}] line {f.line}: '{f.matched_text}' "
                f"(weight={f.weight}){extra}"
            )


def _scan_files(
    files: list[Path], rules: RuleTable, blacklist: Blacklist | None
) -> tuple[list[Finding], list[Finding]]:
    from orchestrata_ultimate.factory.detectors.engine import run_detectors

    all_findings: list[Finding] = []
    all_dismissed: list[Finding] = []
    for path in files:
        text = path.read_text(encoding="utf-8")
        report = run_detectors(text, str(path), rules=rules, blacklist=blacklist)
        all_findings.extend(report.findings)
        all_dismissed.extend(report.dismissed)
    return all_findings, all_dismissed


def _print_check_report(
    files: list[Path],
    findings: list[Finding],
    dismissed: list[Finding],
    *,
    as_json: bool,
) -> None:
    if as_json:
        payload = {
            "findings": [_finding_to_dict(f) for f in findings],
            "dismissed": [_finding_to_dict(f) for f in dismissed],
            "files_scanned": [str(p) for p in files],
        }
        print(json.dumps(payload, indent=2))
        return
    print(f"scanned {len(files)} file(s)")
    _print_findings(findings)
    if dismissed:
        print(f"\n({len(dismissed)} dismissed finding(s) not shown above)")


def _apply_tier1_gate(
    files: list[Path], findings: list[Finding], dismissed: list[Finding]
) -> int:
    from orchestrata_ultimate.factory.detectors.engine import tier1_gate
    from orchestrata_ultimate.factory.detectors.models import DetectorReport

    combined = DetectorReport(
        findings=findings, dismissed=dismissed, files_scanned=[str(p) for p in files]
    )
    blocking = tier1_gate(combined)
    if not blocking:
        return 0
    print(
        f"\ntier1-gate: {len(blocking)} undismissed BLOCKING finding(s)",
        file=sys.stderr,
    )
    return 1


def _cmd_check(args: argparse.Namespace) -> int:
    """Run detectors over files/dirs, print a report, optionally gate on Tier 1."""
    from orchestrata_ultimate.factory.detectors.engine import Blacklist
    from orchestrata_ultimate.factory.detectors.rules import ALL_RULES

    blacklist = Blacklist(Path(args.blacklist)) if args.blacklist else None
    disabled = set(args.disable or [])
    rules = {k: v for k, v in ALL_RULES.items() if k not in disabled}

    files = _collect_files(args.paths)
    findings, dismissed = _scan_files(files, rules, blacklist)
    _print_check_report(files, findings, dismissed, as_json=args.json)

    if args.tier1_gate:
        return _apply_tier1_gate(files, findings, dismissed)
    return 0


def _cmd_dismiss(args: argparse.Namespace) -> int:
    from orchestrata_ultimate.factory.detectors.engine import Blacklist

    blacklist = Blacklist(Path(args.blacklist))
    for fp in args.fingerprint:
        blacklist.dismiss(fp)
    print(f"dismissed {len(args.fingerprint)} fingerprint(s) into {args.blacklist}")
    return 0


def _cmd_rules(_args: argparse.Namespace) -> int:
    from orchestrata_ultimate.factory.detectors.rules import ALL_RULES

    for rule_id, (meta, _fn) in sorted(ALL_RULES.items()):
        precision = (
            f"{meta.published_precision:.2f}"
            if meta.published_precision is not None
            else "unmeasured"
        )
        print(f"{rule_id}\t{meta.tier.value}\t{precision}\t{meta.provenance}")
    return 0


def _cmd_feed_c1(args: argparse.Namespace) -> int:
    from orchestrata_ultimate.factory.assumptions.ledger import AssumptionLedger
    from orchestrata_ultimate.factory.detectors.c1_bridge import (
        questions_to_assumptions,
    )
    from orchestrata_ultimate.factory.detectors.engine import Blacklist, run_detectors

    text = Path(args.path).read_text(encoding="utf-8")
    blacklist = Blacklist(Path(args.blacklist)) if args.blacklist else None
    report = run_detectors(text, args.path, blacklist=blacklist)
    ledger = AssumptionLedger(Path(args.ledger))
    added = questions_to_assumptions(report, args.design_node, ledger)
    print(f"added {len(added)} assumption(s) to {args.ledger}")
    for assumption in added:
        print(f"  {assumption.id}: {assumption.text}")
    return 0


def _add_check_subparser(sub: argparse._SubParsersAction) -> None:
    check_p = sub.add_parser("check", help="Run detectors over files/dirs")
    check_p.add_argument(
        "paths", nargs="+", help="Files or directories (dirs scan *.md)"
    )
    check_p.add_argument("--json", action="store_true", help="Print findings as JSON")
    check_p.add_argument(
        "--disable", action="append", default=[], help="Rule id to disable"
    )
    check_p.add_argument(
        "--blacklist", default="", help="Dismissed-fingerprint JSONL store"
    )
    check_p.add_argument(
        "--tier1-gate",
        action="store_true",
        help="Exit 1 on any undismissed BLOCKING finding",
    )
    check_p.set_defaults(func=_cmd_check)


def _add_dismiss_subparser(sub: argparse._SubParsersAction) -> None:
    dismiss_p = sub.add_parser("dismiss", help="Dismiss a finding by fingerprint")
    dismiss_p.add_argument(
        "--blacklist", required=True, help="Dismissed-fingerprint JSONL store"
    )
    dismiss_p.add_argument(
        "--fingerprint",
        action="append",
        required=True,
        help="Fingerprint to dismiss (repeatable)",
    )
    dismiss_p.set_defaults(func=_cmd_dismiss)


def _add_rules_subparser(sub: argparse._SubParsersAction) -> None:
    rules_p = sub.add_parser(
        "rules", help="List every rule: id, tier, precision, provenance"
    )
    rules_p.set_defaults(func=_cmd_rules)


def _add_feed_c1_subparser(sub: argparse._SubParsersAction) -> None:
    feed_p = sub.add_parser(
        "feed-c1", help="Bridge Tier-3 questions into the C1 assumption ledger"
    )
    feed_p.add_argument("path", help="File to scan")
    feed_p.add_argument("--ledger", required=True, help="Assumption ledger JSONL path")
    feed_p.add_argument(
        "--design-node", required=True, help="Design node the questions apply to"
    )
    feed_p.add_argument(
        "--blacklist", default="", help="Dismissed-fingerprint JSONL store"
    )
    feed_p.set_defaults(func=_cmd_feed_c1)


def _build_parser() -> argparse.ArgumentParser:
    ap = argparse.ArgumentParser(
        description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter
    )
    sub = ap.add_subparsers(dest="command", required=True)
    _add_check_subparser(sub)
    _add_dismiss_subparser(sub)
    _add_rules_subparser(sub)
    _add_feed_c1_subparser(sub)
    return ap


def main() -> int:
    """Parse argv and dispatch to the selected subcommand's handler."""
    args = _build_parser().parse_args()
    result: int = args.func(args)
    return result


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