"""erp-compose process: the customer's process, from the composition and its change list, before anything is built.
uvx --from git+https://git.dodil.io/dodil/erp-platform.git@v0.7.3 erp-compose process \
--module hr --playbook employee-onboarding [--calls config/hr.calls.yaml] [--out processes] [compose.yaml]
Run it from the project folder `erp-compose init` wrote: the bricks are loaded from `bricks/*/src` (--bricks), not
installed, so it works although the bricks pin an older erp-platform for their own runtime.
Boots the composition in process (one SQLite file per brick in a temporary folder, the same `stage` the scenario
runner stands on), replays the configuration calls in order through the routes (a YAML/JSON list of
`{use_case: <name>, args: {...}}`, the file the build and the bench replay too), then writes what the sign-off
card asks for:
<out>/<playbook>.bpmn show_playbook_bpmn the diagram the customer sees and signs
<out>/scenarios.json scenario_paths each scenario as the step keys it walks, and whether it still walks
<out>/deviations.json playbook_deviations every difference from the reference, with its class and reason
and prints the short hash (signoff.content_hash), the deviations by class and the scenarios that no longer walk.
A call the brick refuses (for example a change that breaks the kept foundation) stops the export with exit 1 and
the brick's own status and reason: tell the customer that reason, and take the call out of the list.
Why it exists (lab, 2026-10-18): the method said to show the customer their BPMN and sign it before the build, but
nothing let a builder boot the brick and apply the change list before `init`'s build steps; a builder then
described the process in prose and signed a hash of its own notes.
"""
import argparse
import contextlib
import json
import pathlib
import re
import sys
import tempfile
def _calls(path: pathlib.Path) -> list:
import yaml
rows = yaml.safe_load(path.read_text()) or []
rows = rows.get("calls", []) if isinstance(rows, dict) else rows
if not isinstance(rows, list) or not all(isinstance(r, dict) and r.get("use_case") for r in rows):
raise SystemExit(f"{path}: expected a list of {{use_case: <name>, args: {{...}}}}")
return rows
# ── the replay, shared with `erp-compose seed` (0.9.10) ──────────────────────────────────────────
# A call may name its answer (`as: acct_1200`) and a later call's argument may read it back: a string that is exactly
# "$acct_1200.account_id" becomes that field of that answer (any depth: "$so.lines.0.line_no"). That is how a starter
# maps a posting purpose to the ledger's account_id without knowing how the ledger mints it. "$today", "$tomorrow" and
# "$year" are the business day (erp_platform.work), anywhere in a string ("FY$year"), for a row that must be dated.
_REF = re.compile(r"^\$([A-Za-z_][A-Za-z0-9_-]*)((?:\.[A-Za-z0-9_-]+)*)$")
_DAY = re.compile(r"\$(today|tomorrow|year)\b")
def resolve(value, saved: dict):
"""`value` with every "$name.path" string replaced by the saved answer it names, and $today / $tomorrow / $year
(the business day, YYYY-MM-DD, and its year) replaced wherever they appear ("FY$year"). KeyError names what is missing."""
if isinstance(value, dict):
return {k: resolve(v, saved) for k, v in value.items()}
if isinstance(value, list):
return [resolve(v, saved) for v in value]
if not isinstance(value, str) or "$" not in value:
return value
m = _REF.match(value)
if m and m.group(1) not in ("today", "tomorrow", "year"):
name, path = m.group(1), [p for p in m.group(2).split(".") if p]
if name not in saved:
raise KeyError(f"{value}: no earlier call saved its answer `as: {name}`")
out = saved[name]
for p in path:
try:
out = out[int(p)] if isinstance(out, list) else out[p]
except (KeyError, IndexError, ValueError, TypeError):
raise KeyError(f"{value}: the answer saved as {name} has no {p!r}") from None
return out
from datetime import timedelta
from erp_platform import work
today = work.now().date()
days = {"today": today.isoformat(), "tomorrow": (today + timedelta(days=1)).isoformat(), "year": str(today.year)}
return _DAY.sub(lambda d: days[d.group(1)], value)
def use_cases(comp) -> dict:
"""{use case name: (member, use case)} — the first brick that serves a name wins, as in the composition's MCP."""
ucs = {}
for m in comp.members:
for u in m.brick.use_cases:
ucs.setdefault(u.name, (m, u))
return ucs
def send(app, ucs: dict, me: dict, name: str, args: dict, key=None):
"""One use case through the composition's routes, as the gateway would present `me`. Returns the httpx response."""
import anyio
import httpx
from erp_platform import seam
if name not in ucs:
raise LookupError(f"no use case {name!r} in the composition")
m, u = ucs[name]
args, path = dict(args), u.path
for p in u.path_params:
if p not in args:
raise LookupError(f"{name}: missing path argument {p!r}")
path = path.replace("{" + p + "}", str(args.pop(p)))
h = {**seam.gateway_headers(me), "Accept": "application/json"}
if key and u.takes_idempotency_key:
h["Idempotency-Key"] = key
async def go():
async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://local") as c:
if u.method == "GET":
return await c.get(f"/{m.name}{path}", params={k: v for k, v in args.items() if v is not None}, headers=h)
return await c.request(u.method, f"/{m.name}{path}", json=args, headers=h)
return anyio.run(go)
def reason(r) -> str:
try:
why = r.json().get("detail") or r.text
except Exception:
why = r.text
return str(why)[:400]
def export(compose_path, module: str, playbook: str, calls_path=None, out=None, drop=()) -> dict:
"""The three answers after the calls are replayed. Raises RuntimeError naming a refused call."""
from sqlalchemy import create_engine
from erp_platform import compose
from erp_platform.scenarios import Catalogue
from erp_platform.scenarios.inprocess import stage
from erp_platform.signoff import content_hash
comp = compose.load(str(compose_path))
problems = compose.check(comp)["problems"]
if problems:
raise RuntimeError("composition refused: " + "; ".join(map(str, problems)))
if (comp.member(module) or next((m for m in comp.members if m.name.startswith(module)), None)) is None:
raise RuntimeError(f"no brick named {module} in {compose_path}")
ucs = use_cases(comp)
perms = sorted({u.permission for m in comp.members for u in m.brick.use_cases if u.permission})
me = {"sub": "erp-compose.process", "email": "process@local.test", "app_roles": ["admin"], "permissions": perms}
tmp = pathlib.Path(tempfile.mkdtemp(prefix="erp-process-"))
engines = {}
def engine_for(name):
if name not in engines:
engines[name] = create_engine(f"sqlite:///{tmp / (name + '.db')}", connect_args={"check_same_thread": False})
return engines[name]
with contextlib.ExitStack() as st:
app = stage(Catalogue("process", compose=comp, roles=None), st, engine_for).app
def call(name, **args):
try:
r = send(app, ucs, me, name, args)
except LookupError as e:
raise RuntimeError(str(e)) from None
if not 200 <= r.status_code < 300:
raise RuntimeError(f"{name}: refused {r.status_code}: {reason(r)}")
return r.json()
applied, saved = 0, {}
for n, row in enumerate(_calls(pathlib.Path(calls_path)) if calls_path else [], 1):
try:
ans = call(row["use_case"], **resolve(dict(row.get("args") or {}), saved))
except (RuntimeError, KeyError) as e:
raise RuntimeError(f"{calls_path} call {n}: {e.args[0] if isinstance(e, KeyError) else e}") from None
if row.get("as"):
saved[str(row["as"])] = ans
applied = n
xml = call("show_playbook_bpmn", playbook_id=playbook)["bpmn"]
paths = call("scenario_paths", playbook_id=playbook) if "scenario_paths" in ucs else {"paths": []}
devs = call("playbook_deviations", playbook_id=playbook) if "playbook_deviations" in ucs else {"deviations": []}
if drop:
unknown = [d for d in drop if not any(p.get("scenario_id") == d and p.get("walks") is False for p in paths.get("paths", []))]
if unknown:
raise RuntimeError(f"--accepted-broken {','.join(unknown)}: not a scenario that no longer walks here")
paths = {**paths, "paths": [p for p in paths.get("paths", []) if p.get("scenario_id") not in drop],
"accepted_broken": list(drop)}
res = {"playbook": playbook, "calls": applied, "hash": content_hash(xml), "bpmn": xml, "scenarios": paths, "deviations": devs}
if out:
o = pathlib.Path(out); o.mkdir(parents=True, exist_ok=True)
(o / f"{playbook}.bpmn").write_text(xml)
(o / "scenarios.json").write_text(json.dumps(paths, indent=1))
(o / "deviations.json").write_text(json.dumps(devs, indent=1))
res["out"] = str(o)
return res
def main(argv=None) -> int:
ap = argparse.ArgumentParser(prog="erp-compose process", description=__doc__.split("\n\n")[0])
ap.add_argument("compose", nargs="?", default="compose.yaml")
ap.add_argument("--module", required=True)
ap.add_argument("--playbook", required=True)
ap.add_argument("--calls", help="the configuration calls to replay first (config/<module>.calls.yaml)")
ap.add_argument("--out", default="processes")
ap.add_argument("--accepted-broken", default="", help="scenario ids that no longer walk AND the customer accepted by "
"name: left out of scenarios.json so `signoff record` takes it (it refuses a scenario that does not walk)")
ap.add_argument("--bricks", default="bricks", help="the folder holding the brick clones (init writes bricks/)")
a = ap.parse_args(argv)
for src in sorted(pathlib.Path(a.bricks).glob("*/src")):
sys.path.insert(0, str(src.resolve()))
from erp_platform.compose import brick_paths # 0.8: pinned bricks live in .venv, owned ones in owned/
brick_paths(pathlib.Path(a.compose).resolve().parent)
try:
from erp_platform.scenarios.cli import quiet
quiet()
except Exception:
pass
try:
r = export(a.compose, a.module, a.playbook, a.calls, a.out, tuple(filter(None, a.accepted_broken.split(","))))
except RuntimeError as e:
print(f"refused: {e}", file=sys.stderr)
return 1
devs = r["deviations"].get("deviations", [])
by = {}
for d in devs:
by.setdefault(d.get("class"), []).append(d)
print(f"{r['playbook']}: {r['calls']} calls replayed → {r['out']}/{r['playbook']}.bpmn, scenarios.json, deviations.json")
print(f" short hash {r['hash'][:8]} (what the customer approves; `erp-compose signoff record` checks it)")
for cls in ("variation", "warning", "foundation"):
for d in by.get(cls, []):
print(f" {cls:10} {d.get('key') or d.get('field')}: {d.get('reason')}")
paths = r["scenarios"].get("paths", [])
broken = [p for p in paths if p.get("walks") is False]
print(f" scenarios: {len(paths) - len(broken)} of {len(paths)} walk")
for p in broken:
print(f" NO LONGER WALKS {p.get('scenario_id')}: {p.get('problem')}")
for d in r["scenarios"].get("accepted_broken", []):
print(f" accepted, left out of scenarios.json: {d}")
if broken:
print(" Show the customer these by id. Once they accept them by name, re-run with --accepted-broken "
+ ",".join(p.get("scenario_id") for p in broken) + " before `signoff record`.")
return 0
if __name__ == "__main__":
sys.exit(main())