dodil/erp-platform

src/erp_platform/gate.py

"""gate — nothing is built from a design the customer has not seen and signed as a diagram.

The method always said "the customer sees and signs the BPMN before the build". Before 0.8 nothing enforced it:
on 2026-10-18 an agent designed two modules, asked for a signature on the markdown, and never showed a diagram.
So the commands that turn a design into running software check the sign-off first:

  erp-compose plan / migrate      REFUSE a module with no signed process (signoff/<module>/<process>.json), or a
                                  signed process whose BPMN does not
                                  match the BPMN regenerated NOW from what it was signed from (the ERP_<MODULE>.md /
                                  module.yaml the record names, or brick-hr's playbook after the config calls)
  erp-compose init <module>       the same, when ERP_<MODULE>.md for that module is in the folder (it is designed);
                                  in a fresh folder it only says the next step
  erp-compose serve               WARNS (a local look at the software is how a design is shown)
  erp-compose signoff verify --all   the same check for CI: non-zero on any unsigned or drifted module

The modules of a folder are compose.yaml's optional `modules: [invoicing, projects]` plus every ERP_<MODULE>.md
beside it (ERP_INVOICING.md → invoicing), and every module already signed in signoff/. 0.8.1: a module has one
row PER SIGNED PROCESS (hr/employee-onboarding, hr/freelancer-onboarding); each is checked on its own.

THE ESCAPE is `--unsigned "<reason>"`: the command goes on under a loud banner, and the reason is written to
signoff/<module>.unsigned.json (the bench reads it), so skipping the customer is a recorded decision, never a
silent one.
"""
import datetime as dt
import json
import pathlib
import re

BAR = "!" * 78


def modules(root, compose_path=None) -> list:
    root = pathlib.Path(root)
    found = []
    cp = pathlib.Path(compose_path) if compose_path else root / "compose.yaml"
    if cp.exists():
        import yaml
        try:
            spec = yaml.safe_load(cp.read_text()) or {}
        except yaml.YAMLError:
            spec = {}
        # 0.9.12: a brick declared `process_view: "none: <why>"` (master data, no lifecycle) has nothing to sign; init
        # (0.9.11, erp-method v0.1.35) lists it in `modules:`, so `signoff record --all` died on "no workflow has
        # transitions" and the gate held it UNSIGNED forever (the 0.9.12 deploy proof).
        nothing = {str(b.get("name")) for b in spec.get("bricks") or [] if isinstance(b, dict)
                   and str(b.get("process_view") or "").strip().lower().startswith("none")}
        found += [str(m) for m in spec.get("modules") or [] if str(m) not in nothing]
    for f in sorted(root.glob("ERP_*.md")):
        m = re.match(r"^ERP_([A-Za-z0-9_-]+)\.md$", f.name)
        if m:
            found.append(m.group(1).lower())
    from erp_platform import signoff
    found += signoff.signed_modules(root)
    return list(dict.fromkeys(found))


def label(row: dict) -> str:
    p = row.get("process")
    return row["module"] if not p or p == row["module"] else f"{row['module']}/{p}"


def regenerate(root, rec: dict, compose_path=None) -> str:
    """The BPMN the signed source gives NOW. Raises LookupError when the record names no source (signed < 0.8)."""
    root = pathlib.Path(root)
    src = rec.get("source") or {}
    if src.get("kind") == "bpmn":
        from erp_platform import bpmn
        from erp_platform import signoff                 # 0.9.11: processes/<module>/ first, then the folder
        return bpmn.generate(signoff.spec_path(root, rec.get("module") or "", src["from"]), src.get("workflow"))
    if src.get("kind") == "process":
        from erp_platform import process_export
        from erp_platform.compose import brick_paths     # 0.8.2: run from uvx, the bricks are in the folder's .venv
        brick_paths(pathlib.Path(compose_path).resolve().parent if compose_path else root)
        try:                                              # the request log is for a server, not a CI check's output
            from erp_platform.scenarios.cli import quiet
            quiet()
        except Exception:                                 # noqa: BLE001
            pass
        calls = root / src["calls"] if src.get("calls") else None
        return process_export.export(compose_path or root / "compose.yaml", src.get("brick") or rec["module"],
                                     src["playbook"], calls)["bpmn"]
    raise LookupError("the sign-off names no source to regenerate from (signed before 0.8): record it again with "
                      "`erp-compose signoff record --from ERP_<MODULE>.md` (or --calls for a playbook)")


def process_status(root, module: str, key: str, rec: dict, compose_path=None, can_boot=True) -> dict:
    """{module, process, state, detail} for ONE signed process: signed | drifted | not_implemented | pending |
    unverifiable | present (a playbook, not regenerated here)."""
    from erp_platform import signoff
    root = pathlib.Path(root)
    base = signoff.module_dir(root, module)
    row = {"module": module, "process": key}
    pend = base / f"{key}.pending.json"
    if pend.exists():
        p = json.loads(pend.read_text())
        if p.get("proposed_sha256") != rec.get("bpmn_sha256"):
            return {**row, "state": "pending", "detail": (
                f"a proposed change ({p.get('against_sha256', '')[:8]} → {p.get('proposed_sha256', '')[:8]}, "
                f"{len(p.get('changes') or [])} change(s)) is not signed: show the customer {p.get('change_view')}, "
                f"then `erp-compose signoff record` it — or `erp-compose change withdraw --module {module} "
                f"--process {key}`")}
    if (rec.get("source") or {}).get("kind") == "process" and not can_boot:
        return {**row, "state": "present", "detail": f"signed {rec['bpmn_sha256'][:8]} by "
                f"{rec.get('approver')} (a playbook: verified once the bricks are installed)"}
    try:
        xml = regenerate(root, rec, compose_path)
    except LookupError as e:
        return {**row, "state": "unverifiable", "detail": str(e)}
    except Exception as e:                               # noqa: BLE001 — a source that no longer draws
        return {**row, "state": "unverifiable", "detail": f"regenerating it failed: {e}"}
    res = signoff.verify(module, xml, directory=root, process=key)
    v = rec.get("signed_version", 1)
    if res["ok"]:
        std = signoff.baseline(root, module, key, xml)          # 0.9.11: reference only, never a signature
        return {**row, "state": "signed", "detail": f"signed v{v} {res['signed'][:8]} by "
                f"{rec.get('approver')}, {str(rec.get('signed_at'))[:10]}"
                + (f" — identical to the standard v{std.get('signed_version')}" if std else "")}
    # which way round? the build drawing an OLDER signed version = a signed change not built yet
    hist = base / "history"
    for n in signoff.history_versions(hist, key):
        old = json.loads((hist / f"{key}.v{n}.json").read_text())
        if signoff.matches(old.get("bpmn_sha256"), xml) and old.get("signed_version") != v:
            return {**row, "state": "not_implemented", "detail": (
                f"NOT IMPLEMENTED: the customer signed v{v} ({rec.get('reason') or 'no reason recorded'}; "
                f"{str(rec.get('signed_at'))[:10]}), but the build still draws v{old.get('signed_version')}. "
                "Build the signed change, then verify again.\n" + signoff.report(module, res))}
    return {**row, "state": "drifted", "detail": (
        f"DRIFT: the process changed without a new sign-off (signed v{v}). Take the diagram to the customer "
        "first — `erp-compose change propose` — never the code first.\n" + signoff.report(module, res))}


def module_rows(root, module: str, compose_path=None, can_boot=True) -> list:
    """One row per signed process of the module; one `unsigned` row when it has none."""
    from erp_platform import signoff
    root = pathlib.Path(root)
    recs = signoff.records(root, module)
    if not recs:
        return [{"module": module, "process": None, "state": "unsigned",
                 "detail": f"no signoff/{module}/<process>.json: the customer has not signed this module's BPMN"}]
    return [process_status(root, module, k, r, compose_path, can_boot) for k, r in recs.items()]


_WORST = ("unsigned", "pending", "drifted", "not_implemented", "unverifiable", "present", "signed")


def status(root, module: str, compose_path=None, can_boot=True) -> dict:
    """{module, state, detail, processes}: the module as a whole — its worst process decides the state."""
    rows = module_rows(root, module, compose_path, can_boot)
    if len(rows) == 1:
        return {**rows[0], "processes": rows}
    worst = min(rows, key=lambda r: _WORST.index(r["state"]) if r["state"] in _WORST else 0)
    return {"module": module, "process": None, "state": worst["state"], "processes": rows,
            "detail": "; ".join(f"{label(r)}: {r['state']}" for r in rows)}


def check(root, compose_path=None, mods=None, can_boot=True) -> list:
    mods = modules(root, compose_path) if mods is None else mods
    return [r for m in mods for r in module_rows(root, m, compose_path, can_boot)]


def enforce(command: str, root, *, compose_path=None, mods=None, unsigned=None, refuse=True, can_boot=True,
            say=print) -> bool:
    """True when the command may go on. Prints why not; with `unsigned` (a reason) records it and goes on."""
    root = pathlib.Path(root)
    mods = modules(root, compose_path) if mods is None else mods
    rows = check(root, compose_path, mods, can_boot)
    bad = [r for r in rows if r["state"] not in ("signed", "present")]
    if not mods:
        bad = [{"module": "(none)", "state": "unsigned",
                "detail": "no ERP_<MODULE>.md here and no `modules:` in compose.yaml: nothing has been designed "
                          "and signed"}]
    for r in rows:
        if r["state"] in ("signed", "present"):
            say(f"  sign-off {label(r)}: {r['detail']}")
    if not bad:
        return True
    if not refuse:
        say(BAR)
        say(f"  WARNING: {command} of an UNSIGNED or CHANGED design — show it to the customer and sign it:")
        for r in bad:
            say(f"  {label(r)}: {r['state'].upper()} — {r['detail']}")
        say(BAR)
        return True
    if unsigned is not None:
        reason = str(unsigned).strip()
        if not reason:
            say(f"{command}: --unsigned needs the reason, in words (it is recorded)")
            return False
        say(BAR)
        say(f"  UNSIGNED BUILD — {command} goes on WITHOUT the customer's signed BPMN.")
        say(f"  Reason given: {reason}")
        for r in bad:
            say(f"  {label(r)}: {r['state'].upper()} — {r['detail'].splitlines()[0]}")
            if r["module"] != "(none)":
                out = root / "signoff" / f"{r['module']}.unsigned.json"
                out.parent.mkdir(parents=True, exist_ok=True)
                prev = json.loads(out.read_text()) if out.exists() else {"module": r["module"], "events": []}
                prev["events"].append({"command": command, "reason": reason, "state": r["state"],
                                       "at": dt.datetime.now(dt.timezone.utc).replace(microsecond=0).isoformat()
                                       .replace("+00:00", "Z")})
                prev.update({"reason": reason, "state": r["state"]})
                out.write_text(json.dumps(prev, indent=2, ensure_ascii=False) + "\n")
                say(f"  recorded in {out.relative_to(root)}")
        say(BAR)
        return True
    say(f"{command}: REFUSED — a module is built only from the BPMN its customer saw and signed.")
    for r in bad:
        say(f"  {label(r)}: {r['state'].upper()} — {r['detail']}")
    say(NEXT)
    return False


NEXT = ("  Next: show the customer the diagram — `erp-compose bpmn --from ERP_<MODULE>.md --out processes/<module>.bpmn` "
        "(a brick's playbook: `erp-compose process`) — and record their signature:\n"
        "  `erp-compose signoff record --module <module> --from ERP_<MODULE>.md --approver \"<name, role>\"` "
        "(one record per process: a second process of the module is signed beside the first, never over it).\n"
        "  Building without it is `--unsigned \"<reason>\"`, recorded in signoff/<module>.unsigned.json.")