"""change — the signed BPMN drives every change; the code follows, never the reverse (seemo, 2026-10-18).
erp-compose change propose --module invoicing (--bpmn proposed.bpmn | --from ERP_INVOICING.md)
[--deviations new.json] [--dir .]
erp-compose change withdraw --module invoicing [--dir .]
The person who signs may not read code, so a change is shown in the DIAGRAM first. `propose` compares the proposal
with the signed process (signoff/<module>/<process>.bpmn) by stable id and writes:
signoff/<module>/<process>.change.bpmn the CHANGE VIEW: the proposal with every added / changed element marked
dodil:change="added|changed" and every removed one drawn back in, marked
dodil:change="removed" (a viewer colours them)
signoff/<module>/<process>.pending.bpmn the proposal itself, as it will be signed
signoff/<module>/<process>.pending.json the summary in plain words: what changes, which signed scenarios no longer walk,
which deviations appear or disappear, and any foundation refusal
Then the order is strict: show the change view, ask "are you confirming these changes?" with the consequences,
`erp-compose signoff record` the proposal (which supersedes the old sign-off and keeps it under signoff/history/),
and only then change the code. While a proposal is pending, plan / migrate / `signoff verify --all` refuse.
Why, in one line: agents forget follow-ups, and the end user can't read code.
"""
import argparse
import copy
import json
import pathlib
import sys
import xml.etree.ElementTree as ET
from erp_platform import signoff
from erp_platform.signoff import BPMN, SignoffError, content_hash
DODIL = "https://git.dodil.io/dodil/erp-platform/bpmn/1"
DI = "http://www.omg.org/spec/BPMN/20100524/DI"
OPS = {"added": "Add", "removed": "Remove", "changed": "Change"}
def _register(xml: bytes) -> None:
import io
for _ev, (pfx, uri) in ET.iterparse(io.BytesIO(xml), events=("start-ns",)):
if pfx:
ET.register_namespace(pfx, uri)
def change_view(signed: bytes, proposed: bytes, rows: list) -> bytes:
"""The proposal, every added/changed element marked, every removed element (and its shape) drawn back in."""
_register(signed)
_register(proposed)
ET.register_namespace("dodil", DODIL)
root = ET.fromstring(proposed)
old = ET.fromstring(signed)
by_id = {el.get("id"): el for el in root.iter() if el.get("id")}
old_by_id = {signoff._norm(el.get("id")): (el, parent) for parent in old.iter() for el in parent if el.get("id")}
norm_new = {signoff._norm(i): el for i, el in by_id.items()}
procs = {signoff._norm(p.get("id")): p for p in root.iter(f"{{{BPMN}}}process")}
plane = next(root.iter(f"{{{DI}}}BPMNPlane"), None)
old_shapes = {s.get("bpmnElement"): s for s in old.iter() if s.get("bpmnElement")}
for r in rows:
if r["kind"] in ("process", "collaboration", "lane") and r["op"] == "changed":
continue # their text follows from the steps; mark the steps
if r["op"] in ("added", "changed") and r["id"] in norm_new:
norm_new[r["id"]].set(f"{{{DODIL}}}change", r["op"])
elif r["op"] == "removed" and r["id"] in old_by_id:
el, parent = old_by_id[r["id"]]
home = procs.get(signoff._norm(parent.get("id") or ""))
if home is None or signoff._local(el.tag) in ("lane", "laneSet"):
continue
ghost = copy.deepcopy(el)
for c in [c for c in ghost if signoff._local(c.tag) in ("incoming", "outgoing")]:
ghost.remove(c)
ghost.set(f"{{{DODIL}}}change", "removed")
home.append(ghost)
shape = old_shapes.get(el.get("id"))
if plane is not None and shape is not None:
plane.append(copy.deepcopy(shape))
return ET.tostring(root, encoding="utf-8", xml_declaration=True)
def _name(el: dict) -> str:
return el["fields"].get("name") or el["fields"].get("dodil:step") or ""
def summary(signed: bytes, proposed: bytes, rec: dict, new_devs=None) -> dict:
rows = signoff.diff(signed, proposed)
a, b = signoff.elements(signed), signoff.elements(proposed)
words = []
for r in rows:
if r["kind"] in ("process", "collaboration") or (r["kind"] == "flow" and r["op"] == "changed"):
continue
el = (b if r["op"] != "removed" else a)[r["id"]]
what = {"task": "step", "gateway": "decision", "timer": "wait", "event": "start/end", "flow": "path",
"lane": "role", "participant": "outside party"}.get(r["kind"], r["kind"])
label = _name(el)
if r["kind"] == "flow":
label = f"{el['fields'].get('sourceRef')} → {el['fields'].get('targetRef')}"
line = f"{OPS[r['op']]} {what} \"{label}\" ({r['id']})" if label else f"{OPS[r['op']]} {what} {r['id']}"
if r["op"] == "changed":
line += ": " + "; ".join(f"{k} {signoff._short(x, 40)} → {signoff._short(y, 40)}"
for k, (x, y) in r["changes"].items() if k not in ("incoming", "outgoing"))
words.append(line)
steps = {e["fields"].get("dodil:step") for e in b.values() if e["fields"].get("dodil:step")}
broken = []
for sid in rec.get("scenario_ids") or []:
path = next((s.get("path") for s in rec.get("scenarios") or [] if s.get("id") == sid), None)
missing = [k for k in (path or []) if k not in steps]
if missing:
broken.append({"id": sid, "missing_steps": missing})
old_devs = signoff.deviation_ids(rec.get("deviations") or [])
appear, disappear, refused = [], [], []
if new_devs is not None:
rows_new = new_devs.get("deviations", []) if isinstance(new_devs, dict) else new_devs
now = signoff.deviation_ids(rows_new)
appear = [r for r in rows_new if (r.get("key"), r.get("change"), r.get("field"),
r.get("class") or r.get("cls")) not in old_devs]
disappear = [r for r in rec.get("deviations") or [] if (r.get("key"), r.get("change"), r.get("field"),
r.get("class") or r.get("cls")) not in now]
refused = [r for r in rows_new if (r.get("class") or r.get("cls")) in signoff.REFUSED_CLASSES]
return {"changes": words, "rows": rows, "scenarios_no_longer_walking": broken, "deviations_appear": appear,
"deviations_disappear": disappear, "foundation_refused": refused}
def propose(module: str, proposed: bytes, *, directory=".", new_devs=None, process=None) -> dict:
"""0.8.1: a change is to ONE signed process of the module — `process`, else the one the proposal draws, else the
module's only signed process."""
if not signoff.records(directory, module):
raise SignoffError(f"{pathlib.Path(directory) / 'signoff' / module}/: nothing signed yet for {module} — a first "
"design is signed, not proposed (erp-compose signoff record)")
key, rec = signoff.find(directory, module, process, proposed)
d = signoff.module_dir(directory, module)
signed = (pathlib.Path(directory) / rec.get("bpmn_file", f"signoff/{module}/{key}.bpmn")).read_bytes()
h = content_hash(proposed)
if signoff.matches(rec["bpmn_sha256"], proposed):
raise SignoffError(f"the proposal is the signed process ({h[:8]}): nothing to propose")
s = summary(signed, proposed, rec, new_devs)
(d / f"{key}.pending.bpmn").write_bytes(proposed)
(d / f"{key}.change.bpmn").write_bytes(change_view(signed, proposed, s["rows"]))
pending = {"module": module, "process_id": key, "proposed_sha256": h, "against_sha256": rec["bpmn_sha256"],
"bpmn_file": f"signoff/{module}/{key}.pending.bpmn",
"change_view": f"signoff/{module}/{key}.change.bpmn",
**{k: v for k, v in s.items() if k != "rows"}}
(d / f"{key}.pending.json").write_text(json.dumps(pending, indent=2, ensure_ascii=False) + "\n")
return pending
def report(p: dict) -> str:
lines = [f"change proposed for {p['module']}: {p['against_sha256'][:8]} → {p['proposed_sha256'][:8]}, "
f"{len(p['changes'])} change(s). Show the customer {p['change_view']} (changes marked dodil:change), then ask:",
" \"Are you confirming these changes?\""]
lines += [f" - {c}" for c in p["changes"]]
for s in p["scenarios_no_longer_walking"]:
lines.append(f" ! scenario {s['id']} no longer walks: step(s) {', '.join(s['missing_steps'])} are gone")
for r in p["deviations_appear"]:
lines.append(f" + deviation {r.get('key') or r.get('field')} ({r.get('class')}): {r.get('reason')}")
for r in p["deviations_disappear"]:
lines.append(f" - deviation no longer needed: {r.get('key') or r.get('field')}")
for r in p["foundation_refused"]:
lines.append(f" REFUSED: {r.get('key') or r.get('field')} breaks what the brick keeps (foundation): "
f"{r.get('reason')} — this cannot be signed; change the proposal")
lines.append(f" Once confirmed: erp-compose signoff record --module {p['module']} --process {p.get('process_id')} "
f"--bpmn {p['bpmn_file']} --reason \"<the change, in words>\" "
"--approver \"<name>\" (or --from the design), then change the code. Until then plan and migrate refuse.")
return "\n".join(lines)
def main(argv=None) -> int:
ap = argparse.ArgumentParser(prog="erp-compose change", description=__doc__.split("\n\n")[0])
sub = ap.add_subparsers(dest="verb", required=True)
p = sub.add_parser("propose", help="the change view against the signed process, recorded as pending")
p.add_argument("--module", required=True)
p.add_argument("--process", help="the signed process it changes (default: the one the proposal draws)")
p.add_argument("--bpmn", help="the proposed process")
p.add_argument("--from", dest="source", help="or the design it is drawn from (ERP_<MODULE>.md / module.yaml)")
p.add_argument("--workflow")
p.add_argument("--deviations", help="the proposal's deviations (playbook_deviations' answer)")
p.add_argument("--dir", default=".")
w = sub.add_parser("withdraw", help="drop a pending proposal the customer did not confirm")
w.add_argument("--module", required=True)
w.add_argument("--process", help="one process (default: every pending change of the module)")
w.add_argument("--dir", default=".")
a = ap.parse_args(argv)
d = pathlib.Path(a.dir)
try:
if a.verb == "withdraw":
signoff.records(d, a.module) # the legacy layout moved first
k = signoff.safe_key(a.process) if a.process else "*"
base = signoff.module_dir(d, a.module)
gone = list(base.glob(f"{k}.pending.*")) + list(base.glob(f"{k}.change.bpmn"))
for f in gone:
f.unlink()
print(f"change withdrawn for {a.module}: {len(gone)} file(s) removed" if gone
else f"no pending change for {a.module}")
return 0
if a.source:
from erp_platform import bpmn
xml = bpmn.generate(d / a.source, a.workflow).encode("utf-8")
elif a.bpmn:
xml = pathlib.Path(a.bpmn).read_bytes()
else:
raise SignoffError("give --bpmn (the proposed process) or --from (the design it is drawn from)")
devs = json.loads(pathlib.Path(a.deviations).read_text()) if a.deviations else None
print(report(propose(a.module, xml, directory=d, new_devs=devs, process=a.process)))
return 0
except (SignoffError, FileNotFoundError) as e:
print(f"change {a.verb}: {e}", file=sys.stderr)
return 1
except Exception as e: # noqa: BLE001 — a design that does not draw
if type(e).__name__ == "WorkflowError":
print(f"change {a.verb}: {e}", file=sys.stderr)
return 1
raise
if __name__ == "__main__":
raise SystemExit(main())