"""signoff — the customer signs their diagram; the build is checked against what they signed.
erp-compose signoff record --module hr --bpmn process.bpmn --scenarios scenarios.json \\
--deviations deviations.json --approver "Jane Doe, MD" [--file ERP_HR.md] [--dir .]
erp-compose signoff verify --module hr --bpmn current.bpmn [--dir .] [--json]
WHAT IS SIGNED is the process, not the drawing. `content_hash` is the sha256 of a CANONICAL form of the BPMN:
1. the diagram interchange is removed (every element in the BPMNDI, DC and DI namespaces: shapes, edges,
bounds, waypoints, colours) — moving a shape or re-routing an edge changes no signature;
2. the exporter's own stamp (`exporter`, `exporterVersion` on <definitions>) is removed — upgrading the tool
that wrote the file is not a process change;
3. the children of definitions, collaboration, process, laneSet and lane are sorted (by id; a lane's
flowNodeRefs by their text) — a modeler that re-orders the file changes nothing;
4. the rest goes through W3C Canonical XML 2.0 (`xml.etree.ElementTree.canonicalize`) with surrounding
whitespace stripped and namespace prefixes rewritten (`bpmn:` vs `bpmn2:` is not a change; an `xsi:type`
value is resolved to its namespace first);
5. (0.9.1, canonical v2) the generator's provenance phrase in a documentation line — "generated from
module.yaml" vs "generated from process.yaml" — is rewritten to "generated from <source>": WHICH file a
diagram was drawn from is not the process. `record` writes v2 (`"canonical": "v2"`); `verify` accepts a
match under v2 or the legacy (pre-0.9.1) form and reports which matched, so old signatures keep verifying.
Any process change — a task, its lane, a flow, a gateway, a timer's text, a condition, a dodil:step or
dodil:deviation attribute, a documentation line — changes the hash. `verify` then prints the change by STABLE
ID (brick-hr: Task_<key>, Flow_<source id>__<target id>, Gateway_<key>__if|__fork|__join, Timer_<key>), so a
mismatch names the step, not a line number.
`record` writes signoff/<module>/<process_id>.json and a copy of the signed document (signoff/<module>/<process_id>.bpmn,
which `verify` diffs against), and the Processes / Scenarios / Deviations sections of ERP_<MODULE>.md.
0.8.1 — A MODULE HAS MANY PROCESSES. Before, a module had one sign-off file, so signing freelancer onboarding after
employee onboarding overwrote the first (the lab dry run of 2026-10-18). Now each process of a module is signed on
its own: `record` adds or supersedes ONE process and never touches the others; its numbered history is per process
(signoff/<module>/history/<process_id>.v<n>.json); `verify --all` and the gate cover every signed process. The
process id is `--process`, else the playbook id, else the id of the single process in the BPMN (Process_<id>_v<n>
→ <id>), else the module (a BPMN drawing several processes at once). The single-file layout of 0.7/0.8.0
(signoff/<module>.json, signoff/history/<module>.v<n>.*) is moved into the new one the first time it is read. It refuses a deviation
of class `foundation` (what a brick keeps is not the customer's to sign away) and a scenario that does not walk.
Standard library only: the bench loads this file on its own, next to whatever erp-platform a build pinned.
"""
import argparse
import copy
import datetime as dt
import hashlib
import json
import pathlib
import re
import sys
import xml.etree.ElementTree as ET
BPMN = "http://www.omg.org/spec/BPMN/20100524/MODEL"
LAYOUT_NS = ("http://www.omg.org/spec/BPMN/20100524/DI", "http://www.omg.org/spec/DD/20100524/DC",
"http://www.omg.org/spec/DD/20100524/DI")
XSI_TYPE = "{http://www.w3.org/2001/XMLSchema-instance}type"
STAMP = ("exporter", "exporterVersion")
SORTED = ("definitions", "collaboration", "process", "laneSet", "lane", "subProcess")
REFUSED_CLASSES = ("foundation",)
_VERSIONED = re.compile(r"^((?:Definitions|Collaboration|Process|BPMNDiagram|BPMNPlane)_.+)_v\d+$")
class SignoffError(Exception):
pass
# ── the canonical form ─────────────────────────────────────────────────────────────────────
def _ns(tag: str) -> str:
return tag[1:].split("}")[0] if tag.startswith("{") else ""
def _local(tag: str) -> str:
return tag.split("}")[-1]
def _parse(xml) -> ET.Element:
if isinstance(xml, (bytes, bytearray)):
xml = xml.decode("utf-8")
try:
return ET.fromstring(xml.encode("utf-8") if xml.lstrip().startswith("<?xml") else xml)
except ET.ParseError as e:
raise SignoffError(f"not well-formed XML: {e}") from None
def _qnames(xml, root: ET.Element) -> None:
"""xsi:type="bpmn:tFormalExpression" → "{BPMN uri}tFormalExpression": the prefix is the file's choice, the
namespace is the meaning (and ElementTree renames prefixes on output, which would orphan the value)."""
if isinstance(xml, str):
xml = xml.encode("utf-8")
import io
prefixes = {}
for _ev, (pfx, uri) in ET.iterparse(io.BytesIO(xml), events=("start-ns",)):
prefixes.setdefault(pfx, uri)
for el in root.iter():
v = el.get(XSI_TYPE)
if v and not v.startswith("{"):
pfx, _, local = v.rpartition(":")
if pfx in prefixes or (not pfx and "" in prefixes):
el.set(XSI_TYPE, f"{{{prefixes[pfx]}}}{local}")
def _strip(el: ET.Element) -> None:
for child in list(el):
if not isinstance(child.tag, str) or _ns(child.tag) in LAYOUT_NS:
el.remove(child)
else:
_strip(child)
for a in [a for a in el.attrib if _ns(a) in LAYOUT_NS]:
del el.attrib[a]
def _sort(el: ET.Element) -> None:
for child in el:
_sort(child)
if _local(el.tag) in SORTED and len(el):
kids = list(el)
for k in kids:
k.tail = None
kids.sort(key=lambda k: (k.get("id") is None, k.get("id") or "", _local(k.tag), (k.text or "").strip()))
el[:] = kids
# 0.9.1 — canonical v2: the generator's PROVENANCE line ("…generated from module.yaml…" vs "…from process.yaml…")
# names the file the diagram was drawn from, not the process; a brick serving its view from a different source file
# than the one the customer signed never hash-matched. v2 rewrites ONLY that phrase inside <documentation> to a
# fixed token; every other word of the process still counts. Signatures recorded before 0.9.1 (the legacy form)
# keep verifying: `matches` accepts either form and says which one matched.
CANONICAL_FORMS = ("v2", "legacy")
_PROVENANCE = re.compile(r"\bgenerated from [\w.\-/]+?\.(?:ya?ml|md)\b")
def _provenance(el: ET.Element) -> None:
for d in el.iter():
if isinstance(d.tag, str) and _local(d.tag) == "documentation" and d.text:
d.text = _PROVENANCE.sub("generated from <source>", d.text)
def canonical(xml, form: str = "v2") -> bytes:
"""The bytes that are hashed: DI and the exporter stamp removed, sorted, C14N 2.0 (v2: and the generator's
"generated from <file>" provenance neutralised)."""
if form not in CANONICAL_FORMS:
raise ValueError(f"unknown canonical form {form!r}")
root = copy.deepcopy(_parse(xml))
_qnames(xml, root)
if _local(root.tag) != "definitions" or _ns(root.tag) != BPMN:
raise SignoffError("not a BPMN 2.0 document (the root is not bpmn:definitions)")
_strip(root)
for a in STAMP:
root.attrib.pop(a, None)
if form == "v2":
_provenance(root)
_sort(root)
return ET.canonicalize(ET.tostring(root, encoding="unicode"), strip_text=True,
rewrite_prefixes=True).encode("utf-8")
def content_hash(xml, form: str = "v2") -> str:
return hashlib.sha256(canonical(xml, form)).hexdigest()
def matches(signed_sha256, xml, drawn_from: str | None = None) -> str | None:
"""Which canonical form of `xml` hashes to the signed hash: "v2", "legacy" (a signature recorded before
0.9.1), or None. `drawn_from` — the file the signed record says it was generated from (module.yaml) — lets a
legacy signature match a view drawn from another file (a brick's process.yaml): the provenance phrase is put
back to what was signed, then hashed the legacy way. Nothing else of the process is touched."""
if not signed_sha256:
return None
for form in CANONICAL_FORMS:
if content_hash(xml, form) == signed_sha256:
return form
if drawn_from:
text = xml.decode("utf-8") if isinstance(xml, (bytes, bytearray)) else xml
name = pathlib.PurePath(drawn_from).name
if content_hash(_PROVENANCE.sub(f"generated from {name}", text), "legacy") == signed_sha256:
return "legacy"
return None
def _drawn_from(rec: dict) -> str | None:
src = rec.get("source") or {}
return src.get("from") if isinstance(src, dict) and src.get("kind") == "bpmn" else None
# ── elements by stable id, and the diff ────────────────────────────────────────────────────
def _kind(el: ET.Element) -> str:
t = _local(el.tag)
if t in ("sequenceFlow", "messageFlow"):
return "flow"
if t.endswith("Gateway"):
return "gateway"
if t.endswith("Event"):
return "timer" if any(_local(c.tag) == "timerEventDefinition" for c in el) else "event"
if t == "task" or t.endswith("Task") or t in ("subProcess", "callActivity"):
return "task"
if t in ("dataObject", "dataObjectReference", "dataStoreReference"):
return "data"
return {"lane": "lane", "participant": "participant", "process": "process",
"collaboration": "collaboration", "laneSet": "lane"}.get(t, "other")
def _norm(v: str) -> str:
m = _VERSIONED.match(v or "")
return m.group(1) if m else v
def _attr_name(a: str) -> str:
ns = _ns(a)
return (("dodil:" if "dodil" in ns else ns + ":") + _local(a)) if ns else a
def _text(el) -> str:
return " ".join("".join(el.itertext()).split())
def elements(xml) -> dict:
"""stable id → {kind, tag, fields}: every element that matters to the process, layout ignored. Version
suffixes of the containers (Process_<p>_v3) are dropped so two versions of one process line up."""
root = _parse(xml)
_strip(root)
lane_of = {}
for lane in root.iter(f"{{{BPMN}}}lane"):
for ref in lane.findall(f"{{{BPMN}}}flowNodeRef"):
lane_of[(ref.text or "").strip()] = lane.get("id")
out = {}
tops = [root]
for parent in root.iter():
if _local(parent.tag) in ("definitions", "collaboration", "process", "laneSet", "subProcess"):
tops.extend(c for c in parent if c.get("id"))
containers = ("definitions", "collaboration", "process", "laneSet", "lane", "subProcess")
for el in tops:
f = {_attr_name(a): _norm(v) for a, v in el.attrib.items() if a != "id" and a not in STAMP}
rest = []
for c in el:
t = _local(c.tag)
if t == "documentation":
f["documentation"] = _text(c)
elif t == "timerEventDefinition":
f["timer"] = "; ".join(f"{_local(x.tag)}: {_text(x)}" for x in c)
elif t == "conditionExpression":
f["condition"] = _text(c)
elif t in ("incoming", "outgoing") or _local(el.tag) in containers:
continue # flows and nodes are elements of their own
else: # ioSpecification, data associations, loop characteristics …
rest.append(ET.canonicalize(ET.tostring(c, encoding="unicode"), strip_text=True,
rewrite_prefixes=True))
if rest:
f["inputs/outputs"] = hashlib.sha256("".join(sorted(rest)).encode()).hexdigest()[:12]
if el.get("id") in lane_of:
f["lane"] = lane_of[el.get("id")]
if _local(el.tag) == "definitions":
f = {k: v for k, v in f.items() if not k.startswith("xmlns")}
out[_norm(el.get("id") or _local(el.tag))] = {"kind": _kind(el), "tag": _local(el.tag), "fields": f}
return out
ORDER = ("task", "gateway", "timer", "event", "flow", "lane", "data", "participant", "process", "collaboration",
"other")
def diff(signed_xml, current_xml) -> list:
"""[{op: added|removed|changed, kind, id, step, changes: {field: [before, after]}}], tasks first."""
a, b = elements(signed_xml), elements(current_xml)
out = []
for i in sorted(set(a) | set(b)):
x, y = a.get(i), b.get(i)
e = y or x
row = {"id": i, "kind": e["kind"], "tag": e["tag"], "step": e["fields"].get("dodil:step")}
if x is None:
out.append({**row, "op": "added"})
elif y is None:
out.append({**row, "op": "removed"})
else:
ch = {k: [x["fields"].get(k), y["fields"].get(k)] for k in sorted(set(x["fields"]) | set(y["fields"]))
if x["fields"].get(k) != y["fields"].get(k)}
if x["tag"] != y["tag"]:
ch["type"] = [x["tag"], y["tag"]]
if ch:
out.append({**row, "op": "changed", "changes": ch})
rank = {k: n for n, k in enumerate(ORDER)}
return sorted(out, key=lambda r: (rank.get(r["kind"], 99), r["id"]))
def _short(v, n=70) -> str:
s = "∅" if v is None else str(v)
return s if len(s) <= n else s[:n - 1] + "…"
def _from_difference(x, y, n=70) -> tuple:
"""Two long texts that share a start are shown from where they differ."""
if not (isinstance(x, str) and isinstance(y, str)) or max(len(x), len(y)) <= n:
return x, y
i = next((i for i, (p, q) in enumerate(zip(x, y)) if p != q), min(len(x), len(y)))
i = max(0, i - 15)
return (("…" if i else "") + x[i:], ("…" if i else "") + y[i:])
def format_diff(rows: list) -> str:
sign = {"added": "+", "removed": "-", "changed": "~"}
lines = []
for r in rows:
head = f" {sign[r['op']]} {r['kind']:<11} {r['id']}" + (f" [step {r['step']}]" if r.get("step") else "")
if r["op"] != "changed":
lines.append(f"{head} {r['op']}" + (f" ({r['tag']})" if r["kind"] == "other" else ""))
continue
lines.append(head)
for k, (x, y) in r["changes"].items():
x, y = _from_difference(x, y)
lines.append(f" {k}: {_short(x)} → {_short(y)}")
return "\n".join(lines)
# ── record ─────────────────────────────────────────────────────────────────────────────────
def _load_json(path) -> object:
try:
return json.loads(pathlib.Path(path).read_text())
except FileNotFoundError:
raise SignoffError(f"{path}: no such file") from None
except json.JSONDecodeError as e:
raise SignoffError(f"{path}: not JSON ({e})") from None
def scenario_list(raw) -> list:
"""scenario_paths' answer ({paths: [...]}) or a list of ids / {scenario_id|id} rows → [{id, …}]."""
rows = raw.get("paths", raw.get("scenarios", raw.get("scenario_ids"))) if isinstance(raw, dict) else raw
if not isinstance(rows, list):
raise SignoffError("scenarios: expected scenario_paths' answer, or a list of ids or rows")
out = []
for r in rows:
r = {"id": r} if isinstance(r, str) else dict(r)
sid = r.get("scenario_id") or r.get("id")
if not sid:
raise SignoffError(f"scenarios: a row without an id: {r}")
if r.get("walks") is False:
raise SignoffError(f"scenario {sid} does not walk on this process ({r.get('problem')}): "
"fix the process or drop the scenario before signing")
out.append({"id": sid, "title": r.get("title"), "path": r.get("path")})
return out
def deviation_list(raw) -> list:
"""playbook_deviations' answer ({deviations: [...]}) or a list of rows → the accepted rows. Foundation refused."""
rows = raw.get("deviations", []) if isinstance(raw, dict) else raw
if not isinstance(rows, list):
raise SignoffError("deviations: expected playbook_deviations' answer, or a list of rows")
refused = [r for r in rows if (r.get("class") or r.get("cls")) in REFUSED_CLASSES]
if refused:
names = ", ".join(f"{r.get('key') or r.get('field')} ({r.get('change')}: {r.get('reason') or 'kept'})"
for r in refused)
raise SignoffError(f"refused: {len(refused)} foundation deviation{'s' if len(refused) != 1 else ''} — "
f"{names}. What a brick keeps cannot be signed away; change the process back first.")
return [{"key": r.get("key"), "change": r.get("change"), "field": r.get("field"),
"class": r.get("class") or r.get("cls"), "reason": r.get("reason"), "accepted": True} for r in rows]
def deviation_ids(rows: list) -> set:
return {(r.get("key"), r.get("change"), r.get("field"), r.get("class") or r.get("cls")) for r in rows}
def _process_ref(xml) -> tuple:
root = _parse(xml)
p = root.find(f"{{{BPMN}}}process")
m = re.match(r"^Process_(.+)_v(\d+)$", (p.get("id") if p is not None else "") or "")
return (m.group(1), int(m.group(2))) if m else (p.get("id") if p is not None else None, None)
def history_versions(hist: pathlib.Path, name: str) -> list:
"""The signed version numbers in a history folder (<name>.v<n>.json; a kept proposal is not one). `name` is the
process id under signoff/<module>/history/ (0.8.1), the module under the legacy signoff/history/."""
pat = re.compile(rf"^{re.escape(name)}\.v(\d+)\.json$")
return sorted(int(m.group(1)) for m in (pat.match(f.name) for f in hist.glob("*.json")) if m) if hist.exists() \
else []
# ── where a process's sign-off lives (0.8.1: one module, many processes) ───────────────────
_KEY = re.compile(r"[^A-Za-z0-9_-]+")
def safe_key(v: str) -> str:
k = _KEY.sub("-", str(v or "")).strip("-")
if not k:
raise SignoffError(f"not a usable process id: {v!r}")
return k
def single_process(xml):
"""The id of the ONE process a BPMN draws (Process_<id>_v<n> → <id>), or None when it draws several."""
try:
ps = _parse(xml).findall(f"{{{BPMN}}}process")
except SignoffError:
return None
if len(ps) != 1:
return None
m = re.match(r"^Process_(.+?)(?:_v\d+)?$", ps[0].get("id") or "")
return m.group(1) if m else (ps[0].get("id") or None)
def module_dir(directory, module: str) -> pathlib.Path:
return pathlib.Path(directory) / "signoff" / module
# 0.9.11 — A COMPOSITION OF MANY MODULES. One module.yaml at the root holds one module; a composition of eight keeps
# each module's spec in its own folder, processes/<module>/module.yaml, beside signoff/<module>/ — the layout
# `erp-compose init` writes. A receipt names its source RELATIVE TO THE MODULE'S FOLDER ("module.yaml"), exactly as
# erp-method's baseline receipts do (modules/<m>/module.yaml + modules/<m>/signoff/<m>/), so those receipts are
# reused unedited. A source is looked for in processes/<module>/ first, then at the root (a one-module folder, and
# receipts written before 0.9.11 that name "processes/<m>/module.yaml" from the root).
SPEC_DIR = "processes"
BASELINE = "baseline" # processes/<module>/baseline/: erp-method's receipts, reference only, never counted
def baseline(directory, module: str, key: str, xml) -> dict | None:
"""The method's standard receipt (processes/<module>/baseline/<key>.json) the process `xml` is identical to."""
f = spec_dir(directory, module) / BASELINE / f"{key}.json"
try:
rec = json.loads(f.read_text())
except (OSError, ValueError):
return None
return rec if matches(rec.get("bpmn_sha256"), xml, _drawn_from(rec)) else None
def spec_dir(directory, module: str) -> pathlib.Path:
return pathlib.Path(directory) / SPEC_DIR / module
def spec_path(directory, module: str, frm: str) -> pathlib.Path:
"""The file a receipt's source `from` names, for this module of this folder."""
per = spec_dir(directory, module) / frm
return per if per.is_file() else pathlib.Path(directory) / frm
def source_ref(directory, module: str, path) -> str:
"""What a receipt records as its source: relative to processes/<module>/ when the file is in it, else to the
folder."""
d = pathlib.Path(directory).resolve()
p = pathlib.Path(path)
p = (p if p.is_absolute() else d / p).resolve()
for base in (spec_dir(d, module), d):
if p.is_relative_to(base):
return str(p.relative_to(base))
return str(path)
def _legacy_key(directory, rec: dict, module: str) -> str:
if rec.get("process_id"):
return safe_key(rec["process_id"])
if (rec.get("source") or {}).get("playbook"):
return safe_key(rec["source"]["playbook"])
f = pathlib.Path(directory) / rec.get("bpmn_file", f"signoff/{module}.bpmn")
if f.exists():
one = single_process(f.read_bytes())
return safe_key(one or module)
return safe_key(rec.get("playbook_id") or module)
def migrate_legacy(directory, module: str) -> str | None:
"""0.7/0.8.0 kept ONE sign-off per module: signoff/<module>.json + .bpmn, signoff/history/<module>.v<n>.*, and
signoff/<module>.pending.* / .change.bpmn. Move them under signoff/<module>/ as that record's process. Idempotent;
returns the process id it moved, or None."""
d = pathlib.Path(directory) / "signoff"
old = d / f"{module}.json"
if not old.is_file():
return None
rec = json.loads(old.read_text())
pid = _legacy_key(directory, rec, module)
base = module_dir(directory, module)
(base / "history").mkdir(parents=True, exist_ok=True)
def move(src: pathlib.Path, dst: pathlib.Path):
if src.exists() and not dst.exists():
src.rename(dst)
elif src.exists():
src.unlink()
def fix(path: pathlib.Path):
r = json.loads(path.read_text())
r["process_id"] = pid
r["bpmn_file"] = f"signoff/{module}/{pid}.bpmn"
path.write_text(json.dumps(r, indent=2, ensure_ascii=False) + "\n")
move(d / f"{module}.bpmn", base / f"{pid}.bpmn")
for suffix in ("pending.json", "pending.bpmn", "change.bpmn"):
move(d / f"{module}.{suffix}", base / f"{pid}.{suffix}")
pend = base / f"{pid}.pending.json"
if pend.exists():
p = json.loads(pend.read_text())
p.update(process_id=pid, bpmn_file=f"signoff/{module}/{pid}.pending.bpmn",
change_view=f"signoff/{module}/{pid}.change.bpmn")
pend.write_text(json.dumps(p, indent=2, ensure_ascii=False) + "\n")
hist = d / "history"
if hist.exists():
for f in sorted(hist.glob(f"{module}.v*")):
m = re.match(rf"^{re.escape(module)}\.(v\d+\..+)$", f.name)
if m:
dst = base / "history" / f"{pid}.{m.group(1)}"
move(f, dst)
if dst.suffix == ".json" and dst.exists():
fix(dst)
move(old, base / f"{pid}.json")
fix(base / f"{pid}.json")
return pid
def records(directory, module: str) -> dict:
"""{process_id: record} of every signed process of a module (the legacy layout migrated first)."""
migrate_legacy(directory, module)
base = module_dir(directory, module)
if not base.is_dir():
return {}
out = {}
for f in sorted(base.glob("*.json")):
if "." in f.stem: # <pid>.pending.json is a proposal, not a signature
continue
out[f.stem] = json.loads(f.read_text())
return out
def signed_modules(directory) -> list:
"""Every module with a sign-off in the folder: signoff/<module>/ folders and legacy signoff/<module>.json."""
d = pathlib.Path(directory) / "signoff"
if not d.is_dir():
return []
found = [f.name for f in sorted(d.iterdir()) if f.is_dir() and f.name != "history" and any(
g for g in f.glob("*.json") if "." not in g.stem)]
found += [f.stem for f in sorted(d.glob("*.json")) if "." not in f.stem]
return list(dict.fromkeys(found))
def find(directory, module: str, process=None, xml=None) -> tuple:
"""(process_id, record): the signed process `process` names, else the one `xml` draws, else the only one."""
recs = records(directory, module)
where = pathlib.Path(directory) / "signoff" / module
if not recs:
raise SignoffError(f"{where}/: no sign-off recorded for {module}")
if process:
k = safe_key(process)
if k not in recs:
raise SignoffError(f"{module}: no signed process {k!r} (signed: {', '.join(recs)})")
return k, recs[k]
if xml is not None:
one = single_process(xml)
for cand in (one, module):
if cand and safe_key(cand) in recs:
return safe_key(cand), recs[safe_key(cand)]
if len(recs) == 1:
return next(iter(recs.items()))
raise SignoffError(f"{module} has {len(recs)} signed processes ({', '.join(recs)}): give --process")
def record(module: str, bpmn_path, scenarios_path, deviations_path, approver: str, *, directory=".",
md_file=None, playbook_id=None, version=None, signed_at=None, source=None, calls=None,
workflow=None, reason=None, implemented_by=(), process=None) -> dict:
"""0.8: `source` (an ERP_<MODULE>.md or module.yaml, relative to `directory`) or `calls` (a playbook's config
calls, with `playbook_id`) is recorded so the gate and `verify --all` can REGENERATE the process and compare.
With a source and no `bpmn_path`, the BPMN is generated here; with both, they must be the same process."""
if not approver or not approver.strip():
raise SignoffError("--approver is the person who signed, by name")
directory = pathlib.Path(directory)
origin = None
if source:
from erp_platform import bpmn
rel = pathlib.Path(source_ref(directory, module, source)) # 0.9.11: per-module folders
try:
generated = bpmn.generate(spec_path(directory, module, str(rel)), workflow).encode("utf-8")
except bpmn.WorkflowError as e:
raise SignoffError(str(e)) from None
origin = {"kind": "bpmn", "from": str(rel), "workflow": workflow}
if md_file is None and spec_path(directory, module, str(rel)).parent == spec_dir(directory, module):
md_file = spec_dir(directory, module) / f"ERP_{module.upper()}.md"
if bpmn_path and content_hash(pathlib.Path(bpmn_path).read_bytes()) != content_hash(generated):
raise SignoffError(f"{bpmn_path} is not the process {rel} generates now: show the customer the current "
f"diagram (erp-compose bpmn --from {rel}) and sign that")
if not bpmn_path:
out = directory / "processes" / f"{module}.bpmn"
out.parent.mkdir(parents=True, exist_ok=True)
out.write_bytes(generated)
bpmn_path = out
elif calls is not None:
origin = {"kind": "process", "brick": module, "playbook": playbook_id, "calls": str(calls) if calls else None}
if not playbook_id:
raise SignoffError("--calls records a playbook's process: give --playbook too")
if not bpmn_path:
raise SignoffError("give --bpmn (the diagram the customer signed) or --from (the design it is drawn from)")
xml = pathlib.Path(bpmn_path).read_bytes()
h = content_hash(xml)
# 0.8: a module with no brick (its BPMN from `erp-compose bpmn --from ERP_<MODULE>.md`) has no scenario_paths
# or playbook_deviations to export: none given is an empty list, said in the record
scen_raw = _load_json(scenarios_path) if scenarios_path else []
dev_raw = _load_json(deviations_path) if deviations_path else []
scen, devs = scenario_list(scen_raw), deviation_list(dev_raw)
pid, ver = _process_ref(xml)
for src in (dev_raw, scen_raw):
if isinstance(src, dict):
pid = src.get("playbook_id") or pid
ver = src.get("version") or ver
answered = next((src.get("playbook_id") for src in (dev_raw, scen_raw) if isinstance(src, dict)
and src.get("playbook_id")), None)
key = safe_key(process or playbook_id or answered or single_process(xml) or module)
migrate_legacy(directory, module)
rec = {"module": module, "process_id": key, "playbook_id": playbook_id or pid, "version": int(version or ver or 0) or None,
"bpmn_sha256": h, "canonical": "v2", "file_sha256": hashlib.sha256(xml).hexdigest(),
"bpmn_file": f"signoff/{module}/{key}.bpmn", "scenario_ids": [s["id"] for s in scen],
"scenarios": scen,
"deviations": devs, "approver": approver.strip(), "source": origin,
"signed_at": signed_at or dt.datetime.now(dt.timezone.utc).replace(microsecond=0).isoformat().replace(
"+00:00", "Z")}
out = module_dir(directory, module)
out.mkdir(parents=True, exist_ok=True)
prev = out / f"{key}.json"
# 0.8: the signed process is VERSION CONTROL. Every signature is a numbered version (v1, v2, …), kept for good
# in signoff/<module>/history/<process>.v<n>.json + .bpmn — append-only, forward only: a revert is a NEW version
# equal to an old one, with its reason. Signing the current process again changes nothing but the record's text.
# 0.8.1: per PROCESS — signing one process of a module never touches another.
hist = out / "history"
versions = history_versions(hist, key)
n = (versions[-1] if versions else 0)
old = json.loads(prev.read_text()) if prev.exists() else None
if old and not versions: # signed before 0.8: its record becomes v1
hist.mkdir(parents=True, exist_ok=True)
old.setdefault("signed_version", 1)
(hist / f"{key}.v1.json").write_text(json.dumps(old, indent=2, ensure_ascii=False) + "\n")
oldx = directory / old.get("bpmn_file", f"signoff/{module}/{key}.bpmn")
if oldx.exists():
(hist / f"{key}.v1.bpmn").write_bytes(oldx.read_bytes())
n = 1
same = old is not None and matches(old.get("bpmn_sha256"), xml) is not None
if old and not same:
rec["supersedes"] = {"signed_version": old.get("signed_version", n), "bpmn_sha256": old.get("bpmn_sha256"),
"signed_at": old.get("signed_at"), "approver": old.get("approver")}
if not (reason and reason.strip()):
raise SignoffError(f"{module}/{key} is signed as v{old.get('signed_version', n)}: a new version needs "
"--reason (the change the customer confirmed, in words)")
elif same and old.get("supersedes"):
rec["supersedes"] = old["supersedes"]
rec["signed_version"] = n if same else n + 1
earlier = next((v for v in versions if matches(json.loads((hist / f"{key}.v{v}.json").read_text())
.get("bpmn_sha256"), xml)), None) if not same else None
if earlier:
rec["reverts_to"] = earlier # forward only: v<n+1> equals v<earlier>
rec["reason"] = (reason or "").strip() or (old.get("reason") if same and old else None) or \
("the first signed version" if rec["signed_version"] == 1 else None)
# the data migrations that implement this version (NNNNN_<name>/{up,down}.sql) — traced, enforced later
rec["implemented_by"] = list(implemented_by or (old.get("implemented_by", []) if same and old else []))
(out / f"{key}.bpmn").write_bytes(xml)
prev.write_text(json.dumps(rec, indent=2, ensure_ascii=False) + "\n")
hist.mkdir(parents=True, exist_ok=True)
(hist / f"{key}.v{rec['signed_version']}.json").write_text(json.dumps(rec, indent=2, ensure_ascii=False) + "\n")
(hist / f"{key}.v{rec['signed_version']}.bpmn").write_bytes(xml)
pend = out / f"{key}.pending.json"
if pend.exists() and matches(json.loads(pend.read_text()).get("proposed_sha256"), xml):
for f in (pend, out / f"{key}.pending.bpmn", out / f"{key}.change.bpmn"):
if f.exists():
f.rename(hist / f"{key}.v{rec['signed_version']}.{f.name.split('.', 1)[1]}")
md = pathlib.Path(md_file) if md_file else directory / f"ERP_{module.upper()}.md"
write_md(md, module, records(directory, module))
return rec
def record_all(approver: str, reason, *, directory=".", compose_path=None, say=print) -> list:
"""0.9.11: the customer's ONE approval of a composition — every workflow of every composed module's spec
(processes/<module>/module.yaml) recorded as its own process receipt, the same as `record --from` one at a time.
A process already signed with the same content keeps its version and takes this approver and reason; a changed
one becomes a new version (the reason says what the customer confirmed). Returns the records."""
from erp_platform import bpmn, gate
directory = pathlib.Path(directory)
mods = gate.modules(directory, compose_path)
if not mods:
raise SignoffError("no modules here: compose.yaml `modules:` names none, and there is no ERP_<MODULE>.md")
missing = [m for m in mods if not (spec_dir(directory, m) / "module.yaml").is_file()]
if missing:
raise SignoffError("no spec for " + ", ".join(missing) + f": --all records each module from "
f"{SPEC_DIR}/<module>/module.yaml (what `erp-compose init` writes); sign the others one at a "
"time with `signoff record --module M --from …`")
out = []
for m in mods:
spec = spec_dir(directory, m) / "module.yaml"
try:
flows = [w.entity for w in bpmn.load(spec).workflows]
except bpmn.WorkflowError as e:
raise SignoffError(str(e)) from None
for wf in flows:
xml = bpmn.generate(spec, wf).encode("utf-8")
key = safe_key(single_process(xml) or f"{m}_{wf}")
drawn = spec_dir(directory, m) / f"{key}.bpmn"
drawn.write_bytes(xml)
before = records(directory, m).get(key)
rec = record(m, drawn, None, None, approver, directory=directory, source=str(spec), workflow=wf,
reason=reason, process=key)
was = "re-signed, unchanged" if before and before.get("signed_version") == rec["signed_version"] else \
f"signed v{rec['signed_version']}"
say(f" {m}/{key}: {was} — {rec['bpmn_sha256'][:8]} by {rec['approver']}")
out.append(rec)
return out
# ── ERP_<MODULE>.md ────────────────────────────────────────────────────────────────────────
def _block(module: str, section: str, body: str) -> str:
return (f"<!-- erp-compose signoff {module} {section}: written by `erp-compose signoff record`; "
f"edit the process and sign again, not this block -->\n{body}\n<!-- /erp-compose signoff {module} -->")
def _put(text: str, section: str, module: str, body: str) -> str:
block = _block(module, section.lower(), body)
pat = re.compile(rf"<!-- erp-compose signoff {re.escape(module)} {section.lower()}:.*?"
rf"<!-- /erp-compose signoff {re.escape(module)} -->", re.S)
if pat.search(text):
return pat.sub(lambda _: block, text, count=1)
head = re.search(rf"^## {section}[^\n]*\n", text, re.M)
if head is None:
return text.rstrip("\n") + f"\n\n## {section}\n{block}\n"
nxt = re.search(r"^## ", text[head.end():], re.M)
at = head.end() + (nxt.start() if nxt else len(text) - head.end())
return text[:at].rstrip("\n") + "\n\n" + block + "\n\n" + text[at:].lstrip("\n")
def write_md(path: pathlib.Path, m: str, recs: dict) -> None:
"""The Processes / Scenarios / Deviations sections: one row per signed process (0.8.1), its scenarios and its
accepted deviations — every process of the module, not just the one signed last."""
procs = ("| Process | Signed version | File | Content sha256 (layout ignored) | Signed by, date |\n"
"|---|---|---|---|---|\n")
scen_rows, dev_rows, n = [], [], 0
for key, rec in recs.items():
day = str(rec.get("signed_at"))[:10]
sv = rec.get("signed_version")
procs += (f"| {key} | v{sv or rec.get('version')} | {rec['bpmn_file']} | `{rec['bpmn_sha256']}` "
f"(short `{rec['bpmn_sha256'][:8]}`) | {rec['approver']}, {day} |\n")
scen = rec.get("scenarios") or [{"id": i} for i in rec.get("scenario_ids") or []]
scen_rows += [f"| {s['id']} | {key} | {s.get('title') or ''} | {' → '.join(s.get('path') or []) or '—'} "
"| signed |" for s in scen]
for d in rec.get("deviations") or []:
n += 1
dev_rows.append(
f"| D-{n} | {key} | {d.get('key') or d.get('field') or '—'} | {d.get('change')}"
f"{(' ' + d['field']) if d.get('field') and d.get('key') else ''} | {d.get('class')} | "
f"{(d.get('reason') or '').replace('|', '/')} | {rec['approver']}, {day} |")
procs += (f"\nCheck a build against it: `erp-compose signoff verify --all` (every signed process), or one: "
f"`erp-compose signoff verify --module {m} --process <process>`.")
scens = ("| ID | Process | Story | Path (step keys) | Status |\n|---|---|---|---|---|\n"
+ ("\n".join(scen_rows) or "| — | | no scenarios signed | | |"))
devs = ("| # | Process | Step / setting | Change | Class | Warning (source) | Accepted by, date |\n"
"|---|---|---|---|---|---|---|\n" + ("\n".join(dev_rows) or "| — | | none: the reference as it stands | | | | |"))
text = path.read_text() if path.exists() else f"# ERP_{m.upper()}\n"
for section, body in (("Processes", procs), ("Scenarios", scens), ("Deviations", devs)):
text = _put(text, section, m, body)
path.write_text(text if text.endswith("\n") else text + "\n")
# ── verify ─────────────────────────────────────────────────────────────────────────────────
def verify(module: str, current_xml, *, directory=".", process=None) -> dict:
"""{ok, process, signed, current, diff}: the current process against its signed record (0.8.1: the process
`process` names, else the one the BPMN draws, else the module's only signed process)."""
key, rec = find(directory, module, process, current_xml)
form = matches(rec["bpmn_sha256"], current_xml, _drawn_from(rec))
cur = content_hash(current_xml, form or "v2")
out = {"ok": form is not None, "process": key, "signed": rec["bpmn_sha256"], "current": cur,
"canonical": form, "diff": []}
if not out["ok"]:
signed_file = pathlib.Path(directory) / rec.get("bpmn_file", f"signoff/{module}/{key}.bpmn")
if signed_file.exists():
if matches(rec["bpmn_sha256"], signed_file.read_bytes()) is None:
out["note"] = f"{signed_file} no longer matches the signed hash: the diff is against an edited copy"
out["diff"] = diff(signed_file.read_bytes(), current_xml)
else:
out["note"] = f"{signed_file} is missing: only the hashes can be compared"
return out
def report(module: str, res: dict) -> str:
if res.get("process") and res["process"] != module:
module = f"{module}/{res['process']}"
if res["ok"]:
return (f"signoff verify {module}: conforms — content sha256 {res['current'][:8]} is the signed one"
f" (canonical {res.get('canonical') or 'v2'})")
rows = res["diff"]
n = {op: sum(r["op"] == op for r in rows) for op in ("added", "removed", "changed")}
lines = [f"signoff verify {module}: MISMATCH — signed {res['signed'][:8]}, current {res['current'][:8]} "
f"({n['added']} added, {n['removed']} removed, {n['changed']} changed)"]
if res.get("note"):
lines.append(f" note: {res['note']}")
if rows:
lines.append(format_diff(rows))
return "\n".join(lines)
# ── the command ────────────────────────────────────────────────────────────────────────────
def main(argv=None) -> int:
ap = argparse.ArgumentParser(prog="erp-compose signoff",
description="Record a customer's sign-off of their BPMN; verify a build against it.")
sub = ap.add_subparsers(dest="verb", required=True)
r = sub.add_parser("record", help="write signoff/<module>.json and the ERP_<MODULE>.md sections")
r.add_argument("--module", help="the module (or --all)")
r.add_argument("--all", action="store_true", help="0.9.11: every workflow of every composed module, from "
"processes/<module>/module.yaml — one receipt per process, for the customer's one approval")
r.add_argument("--compose", default=None, help="with --all: the compose.yaml (default <dir>/compose.yaml)")
r.add_argument("--process", help="the process id (default: the playbook id, else the single process the BPMN "
"draws, else the module); each process of a module is signed on its own")
r.add_argument("--bpmn", help="the BPMN the customer signed (default with --from: generated from it)")
r.add_argument("--from", dest="source", help="the design it is drawn from (ERP_<MODULE>.md or module.yaml): "
"recorded, so the gate and verify --all regenerate and compare")
r.add_argument("--workflow", help="with --from: the one workflow signed (its entity)")
r.add_argument("--reason", help="why this version: required from v2 on (the change the customer confirmed)")
r.add_argument("--implemented-by", default="", help="the data migrations that implement it, comma-separated "
"(NNNNN_<name>, the versioned-migrations layout)")
r.add_argument("--calls", help="a playbook's config calls (config/<module>.calls.yaml), recorded with --playbook "
"so the gate regenerates through `erp-compose process`")
r.add_argument("--scenarios", help="scenario_paths' answer, or a JSON list of scenario ids (default: none)")
r.add_argument("--deviations", help="playbook_deviations' answer, or a JSON list of rows (default: none)")
r.add_argument("--approver", required=True, help="who signed: their name (and role)")
r.add_argument("--file", help="the design file (default <dir>/ERP_<MODULE>.md)")
r.add_argument("--dir", default=".", help="the project folder (signoff/ is written here)")
r.add_argument("--playbook", help="the playbook id (default: from the answers, else the process id)")
r.add_argument("--version", type=int, help="the playbook version (default: from the answers)")
v = sub.add_parser("verify", help="compare a regenerated BPMN with the signed one")
v.add_argument("--module", help="one module (default with --all: every module of the folder)")
v.add_argument("--process", help="one signed process of the module (default: the one the BPMN draws)")
v.add_argument("--bpmn", help="the process as built (default: regenerated from the source the record names)")
v.add_argument("--all", action="store_true", help="CI: every module (compose.yaml `modules:` and each "
"ERP_<MODULE>.md), non-zero on any unsigned, drifted or unverifiable one")
v.add_argument("--compose", default=None, help="the compose.yaml (default <dir>/compose.yaml)")
v.add_argument("--dir", default=".")
v.add_argument("--json", action="store_true", help="print the result as JSON")
a = ap.parse_args(argv)
try:
if a.verb == "record" and a.all:
if a.module or a.bpmn or a.source or a.calls:
raise SignoffError("--all records every composed module from its spec: no --module/--bpmn/--from/--calls")
recs = record_all(a.approver, a.reason, directory=a.dir, compose_path=a.compose)
print(f"signed {len(recs)} process(es) of {len({r['module'] for r in recs})} module(s) for {a.approver.strip()}"
f" → {pathlib.Path(a.dir) / 'signoff'}/")
return 0
if a.verb == "record" and not a.module:
raise SignoffError("give --module (or --all for every composed module)")
if a.verb == "record":
rec = record(a.module, a.bpmn, a.scenarios, a.deviations, a.approver, directory=a.dir, md_file=a.file,
playbook_id=a.playbook, version=a.version, source=a.source, calls=a.calls,
workflow=a.workflow, reason=a.reason,
implemented_by=[x.strip() for x in a.implemented_by.split(",") if x.strip()],
process=a.process)
others = [k for k in records(a.dir, a.module) if k != rec["process_id"]]
print(f"signed {rec['module']}/{rec['process_id']} as version {rec['signed_version']}"
+ (f" (= v{rec['reverts_to']}, a revert)" if rec.get("reverts_to") else "") + ": "
f"{rec['playbook_id']} v{rec['version']}: content sha256 "
f"{rec['bpmn_sha256'][:8]} ({len(rec['scenario_ids'])} scenarios, {len(rec['deviations'])} "
f"deviations accepted) by {rec['approver']} → "
f"{pathlib.Path(a.dir) / 'signoff' / a.module / (rec['process_id'] + '.json')}"
+ (f"; also signed, untouched: {', '.join(others)}" if others else ""))
return 0
if a.all or not a.bpmn:
from erp_platform import gate
mods = [a.module] if a.module else None
if not a.all and not a.module:
raise SignoffError("give --module (with --bpmn, or to regenerate it), or --all")
rows = gate.check(a.dir, a.compose, mods)
if a.process:
rows = [r for r in rows if r.get("process") == a.process]
if not rows:
rows = [{"module": "(none)", "state": "unsigned", "detail": "no ERP_<MODULE>.md and no `modules:` "
"in compose.yaml: nothing to verify, which CI treats as a failure"}]
if a.json:
print(json.dumps(rows, indent=1))
for r_ in rows if not a.json else []:
print(f"signoff {gate.label(r_)}: {r_['state'].upper()} — {r_['detail']}")
return 0 if all(r_["state"] == "signed" for r_ in rows) else 1
res = verify(a.module, pathlib.Path(a.bpmn).read_bytes(), directory=a.dir, process=a.process)
print(json.dumps(res, indent=1) if a.json else report(a.module, res))
return 0 if res["ok"] else 1
except (SignoffError, FileNotFoundError) as e:
print(f"signoff {a.verb}: {e}", file=sys.stderr)
return 1
if __name__ == "__main__":
raise SystemExit(main())