"""compose — bricks into one product, from a compose.yaml.
A brick is one bounded piece of business software; what a customer runs is a COMPOSITION of them — a CRM is
quote + business-partner + pipeline + activity, not a repo of its own. The composition is data:
# compose.yaml
name: crm
version: 0.1.0
tier: 1
bricks:
- { name: business-partner, brick: brick_bp.application.use_cases:BP, version: 0.1.0 }
- { name: quote, brick: brick_quote.application.use_cases:QUOTE, version: 0.1.0 }
app = compose.app("compose.yaml") # or: erp-compose serve compose.yaml
`erp-compose init <module>` writes this file (and clones, installs and checks the bricks) from the method's
reference.repos and brick_entries — see erp_platform.compose_init.
TIER 1 (implemented): one app, one bucket, one sign-in pool. Each brick's routes are served under
`/<brick>/`, and there is ONE `/mcp` carrying every brick's tools — an agent connects once to the whole
product. Every brick is registered in process, so their seams reach each other through the routes (the
permission checks still run) with no network. Tiers 2–3 deploy the SAME bricks as separate apps and wire
their seams with `<NAME>_BASE_URL`; `erp-compose plan` prints that wiring.
THE CHECK, run before anything is served, refuses a composition whose bricks would collide:
* two bricks claiming one TABLE — in one bucket they would write each other's rows (the CRM's
`account_edges` and the ledger's tree merged silently this way; modules/shared.yaml);
* two bricks offering one TOOL NAME — an agent would call one when it meant the other;
* a declared version that is not the version installed.
It also reports every seam: which neighbour each brick asks, and whether the composition has it. A missing
neighbour is allowed — seams are optional by design — but it is said out loud.
CAPABILITIES (capabilities.yaml, erp_platform.capabilities). A brick REQUIRES capabilities (Brick(requires=…)
or a CapabilitySeam in its module) and PROVIDES them (@BRICK.capability). An optional map chooses:
capabilities:
identity.provision_user: gws # or "manual": nobody, even if a brick could
mail.send: manual
The check refuses a capability name the contract does not have, two providers of one capability with no
map entry choosing between them, a map entry naming a brick that does not provide it, and a provider whose
schema.sql lacks its claims and log tables. A required capability with NO provider is allowed and reported
as "manual" — the consumer makes those steps manual tasks (rule 8). compose.app registers the choice so a
CapabilitySeam reaches its provider in process, and hides the capability of a provider not chosen.
"""
import importlib
import importlib.resources
import pathlib
import pkgutil
import re
import sys
from dataclasses import dataclass, field
from typing import Any, Optional
import yaml
from fastapi import Depends, FastAPI
from fastapi.responses import JSONResponse
from erp_platform import auth, capabilities, public
from erp_platform.bricks import Brick, serve
from erp_platform.seam import Seam, register_local
RESERVED_PREFIXES = ("erp_", "platform_meta") # the platform's own tables (platform.sql, migrate's stamp)
# 0.8: every brick shows its process as BPMN (bricks.PROCESS_VIEWS). A brick built for this platform or later
# that has none is REFUSED; one pinned to an older platform is WARNED — today's compositions keep building, and
# the rule bites as each brick moves to 0.8. A pure master-data brick is waived in compose.yaml:
# - { name: business-partner, brick: …, version: 0.3.0, process_view: none } # master data only: no lifecycle to sign
PROCESS_VIEW_SINCE = (0, 8, 0)
class CompositionRefused(Exception):
"""The composition would collide. The message lists every problem, not just the first."""
@dataclass
class Member:
name: str
target: str # "package.module:OBJECT"
version: str
brick: Brick
module: Any
process_view_waiver: Optional[str] = None # the reason, when compose.yaml says process_view: none
@property
def package(self) -> str:
return self.target.split(".", 1)[0]
def schema(self) -> pathlib.Path:
"""The brick's schema.sql, from inside the installed package."""
return pathlib.Path(str(importlib.resources.files(self.package).joinpath("schema.sql")))
def modules(self) -> list:
"""Every module of the brick's PACKAGE, not just the one holding the Brick.
A brick may declare its seams beside the code that uses them (brick-lead keeps three in
application/neighbours.py, behind an anti-corruption layer). Read from the Brick's module alone, the
composition reported none of them and a tier-2 plan gave the lead app no BP_/ACTIVITY_/PIPELINE_BASE_URL.
So: what is already imported, plus every submodule that imports cleanly. `app` modules are skipped (they
build an app), and a module that cannot import without a database (one reading BUCKET at import) is
skipped, not fatal — checking a composition must not need a bucket."""
found = {self.module.__name__: self.module}
prefix = self.package + "."
for name, mod in list(sys.modules.items()):
if mod is not None and (name == self.package or name.startswith(prefix)):
found.setdefault(name, mod)
pkg = sys.modules.get(self.package)
for info in pkgutil.walk_packages(getattr(pkg, "__path__", []), prefix):
if info.name in found or info.name.rsplit(".", 1)[-1] in ("app", "__main__"):
continue
try:
found[info.name] = importlib.import_module(info.name)
except Exception: # noqa: BLE001 — see the docstring
continue
return list(found.values())
def _instances(self, kind: type) -> list:
seen, out = set(), []
for mod in self.modules():
for v in list(vars(mod).values()):
if isinstance(v, kind) and id(v) not in seen:
seen.add(id(v))
out.append(v)
return out
def seams(self) -> list:
"""One Seam per neighbour the brick asks, wherever in its package it is declared (a re-import of the same
object, as brick-lead's use_cases does, counts once)."""
by_name: dict = {}
for s in self._instances(Seam):
by_name.setdefault(s.name, s)
return list(by_name.values())
def requires(self) -> list:
"""Declared on the Brick, plus every CapabilitySeam anywhere in the brick's package."""
found = [v.name for v in self._instances(capabilities.CapabilitySeam)]
return list(dict.fromkeys([*self.brick.requires, *found]))
@dataclass
class Composition:
name: str
version: str
tier: int
members: list = field(default_factory=list)
description: str = ""
capabilities: dict = field(default_factory=dict) # capability → provider brick name | "manual"
def member(self, name: str):
return next((m for m in self.members if m.name == name), None)
def _waivers(text: str) -> dict:
"""brick name → the reason after `process_view: none` — the YAML comment on its line (`# master data only`),
or the string form `process_view: "none: master data only"`. A comment is invisible to yaml.safe_load, so the
raw text is read; the brick is the `name:` on that line, else the nearest one above it."""
out, name = {}, None
for line in text.splitlines():
m = re.search(r"\bname:\s*['\"]?([A-Za-z0-9_-]+)", line)
if m:
name = m.group(1)
w = re.search(r"\bprocess_view:\s*['\"]?none\b(?::\s*([^'\"#}]*))?['\"]?[^#]*(?:#\s*(.*))?$", line)
if w and name:
out[name] = ((w.group(1) or "").strip() or (w.group(2) or "").strip())
return out
def load(path) -> Composition:
text = pathlib.Path(path).read_text()
spec = yaml.safe_load(text) or {}
waivers = _waivers(text)
missing = [k for k in ("name", "version", "bricks") if k not in spec]
if missing:
raise CompositionRefused(f"{path}: compose.yaml needs {', '.join(missing)}")
comp = Composition(spec["name"], str(spec["version"]), int(spec.get("tier", 1)),
description=spec.get("description", ""),
capabilities={str(k): str(v) for k, v in (spec.get("capabilities") or {}).items()})
problems = []
for entry in spec["bricks"]:
target = entry.get("brick", "")
module_name, _, attr = target.partition(":")
try:
module = importlib.import_module(module_name)
brick = getattr(module, attr)
except (ImportError, AttributeError) as e:
problems.append(f"{entry.get('name')}: cannot import {target!r} ({e}) — is the brick installed?")
continue
if not isinstance(brick, Brick):
problems.append(f"{entry.get('name')}: {target} is not a Brick")
continue
pv = entry.get("process_view")
if pv is not None and not str(pv).lower().startswith("none"):
problems.append(f"{entry.get('name')}: process_view is `none # <reason>` (a waiver) or absent, "
f"not {pv!r}")
continue
waiver = waivers.get(entry.get("name"), "") if pv is not None else None
comp.members.append(Member(entry.get("name", brick.name), target, str(entry.get("version", "")),
brick, module, waiver))
if problems:
raise CompositionRefused("\n".join(problems))
return comp
def _tables(schema: pathlib.Path) -> set:
"""The tables a schema creates. Parsed here rather than via migrate, which needs a BUCKET to import —
checking a composition must not need a database."""
return {n.lower() for n in re.findall(r"CREATE TABLE(?: IF NOT EXISTS)? ([A-Za-z0-9_]+)", schema.read_text(), re.I)}
def check(comp: Composition) -> dict:
"""What the composition is, and what is wrong with it. `problems` empty means it may be served."""
problems, tables, tools, seams = [], {}, {}, []
names = [m.name for m in comp.members]
for n in {n for n in names if names.count(n) > 1}:
problems.append(f"brick {n!r} is listed twice")
for m in comp.members:
if m.name != m.brick.name:
problems.append(f"{m.target} is the brick {m.brick.name!r}, listed as {m.name!r}")
if m.version and m.version != m.brick.version:
problems.append(f"{m.name}: compose.yaml says {m.version}, the installed brick is {m.brick.version}")
try:
owned = _tables(m.schema())
except FileNotFoundError:
problems.append(f"{m.name}: no schema.sql inside the package {m.package}")
owned = set()
for t in owned:
if t.startswith(RESERVED_PREFIXES):
problems.append(f"{m.name}: table {t!r} uses a name reserved for erp-platform")
if t in tables:
problems.append(f"table {t!r} is claimed by both {tables[t]} and {m.name}")
tables.setdefault(t, m.name)
for s in m.seams():
seams.append({"from": m.name, "to": s.name, "present": comp.member(s.name) is not None})
wiring, hidden = capability_wiring(comp, problems)
# Tool names: a capability's tool is the same name in every provider, so only the CHOSEN provider's counts.
for m in comp.members:
for uc in m.brick.use_cases:
if (m.name, uc.name) in hidden or not uc.tool: # a seam-only route is no agent's tool
continue
if uc.name in tools and tools[uc.name] != m.name:
problems.append(f"tool {uc.name!r} is offered by both {tools[uc.name]} and {m.name}")
tools.setdefault(uc.name, m.name)
if comp.tier not in (1, 2, 3):
problems.append(f"tier is 1, 2 or 3, not {comp.tier}")
warnings, views = [], []
for m in comp.members:
view = m.brick.process_view
if view:
views.append({"brick": m.name, "view": view})
continue
if m.process_view_waiver is not None:
if not m.process_view_waiver:
problems.append(f"{m.name}: `process_view: none` needs its reason as a comment "
"(`process_view: none # master data only: no lifecycle`)")
else:
views.append({"brick": m.name, "view": None, "waived": m.process_view_waiver})
continue
built_for = platform_floor(m)
say = (f"{m.name}: declares no process view — nothing can show the customer its process as BPMN. Add "
f"`{m.target.rpartition(':')[2]}.declare_process(\"process.yaml\")` (BRICKS.md, \"The process view\"), or "
f"waive it in compose.yaml for a pure master-data brick: `process_view: none # <reason>`")
if built_for is not None and built_for >= PROCESS_VIEW_SINCE:
problems.append(say)
else:
pinned = ".".join(map(str, built_for)) if built_for else "an unknown version"
warnings.append(f"{say} (warning only: it is built for erp-platform {pinned}; refused from "
f"{'.'.join(map(str, PROCESS_VIEW_SINCE))})")
views.append({"brick": m.name, "view": None})
handlers = [{"brick": m.name, "grant": h.grant, "subject_kind": h.subject_kind, "name": h.name,
"acts": h.action is not None} for m in comp.members for h in m.brick.public_handlers]
for m in comp.members:
for h in m.brick.public_handlers:
if h.brick != m.name:
problems.append(f"{m.name}: a public handler is registered for brick {h.brick!r}")
return {"name": comp.name, "version": comp.version, "tier": comp.tier,
"bricks": [{"name": m.name, "version": m.brick.version, "tools": sum(1 for uc in m.brick.use_cases if uc.tool),
"tables": sorted(t for t, owner in tables.items() if owner == m.name)}
for m in comp.members],
"tools": len(tools), "seams": seams, "capabilities": wiring, "public": handlers,
"process_views": views, "warnings": warnings, "problems": problems}
def platform_floor(m: "Member") -> Optional[tuple]:
"""The lowest erp-platform the brick's own distribution asks for (`erp-platform>=0.5.2,<0.6` → (0, 5, 2)),
or None when that cannot be read (a brick on sys.path, not installed)."""
from importlib import metadata
try:
dists = metadata.packages_distributions().get(m.package) or []
reqs = metadata.requires(dists[0]) if dists else None
except Exception: # noqa: BLE001 — unknown is a warning, never a crash
return None
for r in reqs or []:
if re.match(r"^erp[-_]platform\b", r, re.I) and "extra ==" not in r:
v = re.search(r">=\s*(\d+)\.(\d+)(?:\.(\d+))?", r) or re.search(r"@v?(\d+)\.(\d+)(?:\.(\d+))?", r)
return tuple(int(x or 0) for x in v.groups()) if v else None
return None
def capability_wiring(comp: Composition, problems: list) -> tuple:
"""Who provides each capability the composition mentions: ([{capability, provider, ...}], hidden).
`hidden` is the (brick, tool) pairs of providers NOT chosen — left out of both doors. Problems are
appended to `problems`."""
contract = capabilities.contract()
providers: dict = {}
for m in comp.members:
for c in m.brick.provides:
providers.setdefault(c, []).append(m.name)
if m.brick.provides:
try:
owned = _tables(m.schema())
except FileNotFoundError:
owned = set()
t = capabilities.tables(m.brick.name)
missing = sorted({t.claims.name, t.log.name} - owned)
if missing:
problems.append(f"{m.name} provides capabilities but its schema.sql has no {', '.join(missing)} "
f"(capabilities.ddl({m.brick.name!r}) prints them)")
required: dict = {}
for m in comp.members:
for c in m.requires():
required.setdefault(c, []).append(m.name)
for c, who in required.items():
if c not in contract:
problems.append(f"{', '.join(who)} require {c!r}, which is not a capability in capabilities.yaml "
f"{contract.version}")
for c in comp.capabilities:
if c not in contract:
problems.append(f"compose.yaml maps {c!r}, which is not a capability in capabilities.yaml "
f"{contract.version}")
names = {m.name for m in comp.members}
wiring, hidden = [], set()
for c in sorted(set(providers) | set(required) | set(comp.capabilities)):
if c not in contract:
continue
offered, chosen = providers.get(c, []), comp.capabilities.get(c)
provider = None
if chosen == capabilities.MANUAL:
provider = None
elif chosen:
if chosen not in names:
problems.append(f"compose.yaml maps {c} to {chosen!r}, which is not a brick in this composition")
elif chosen not in offered:
problems.append(f"compose.yaml maps {c} to {chosen}, which does not provide it"
+ (f" ({', '.join(offered)} does)" if offered else ""))
else:
provider = chosen
elif len(offered) > 1:
problems.append(f"{c} is provided by both {' and '.join(offered)}; choose one under capabilities: "
"in compose.yaml")
elif offered:
provider = offered[0]
tool = contract.get(c).tool_name
hidden |= {(b, tool) for b in offered if b != provider}
wiring.append({"capability": c, "provider": provider or capabilities.MANUAL,
"required_by": required.get(c, []), "offered_by": offered,
"destructive": contract.get(c).destructive})
return wiring, hidden
def app(path_or_comp) -> FastAPI:
"""The composition as ONE app (tier 1): every brick's routes under /<brick>, one /mcp for every tool."""
comp = path_or_comp if isinstance(path_or_comp, Composition) else load(path_or_comp)
report = check(comp)
if report["problems"]:
raise CompositionRefused("\n".join(report["problems"]))
if comp.tier != 1:
raise CompositionRefused(f"tier {comp.tier} runs each brick as its own app; `erp-compose plan` "
f"prints how to wire them. compose.app() serves tier 1.")
for m in comp.members:
register_local(m.name, m.brick.routes_app()) # seams reach neighbours in process
_, hidden = capability_wiring(comp, [])
capabilities.reset_providers()
for w in report["capabilities"]: # a CapabilitySeam asks this registry first
capabilities.register_provider(w["capability"], None if w["provider"] == capabilities.MANUAL
else w["provider"])
instructions = (comp.description + "\n\n" if comp.description else "") + "\n\n".join(
f"[{m.name}] {m.brick.instructions}" for m in comp.members if m.brick.instructions)
composed = serve([m.brick for m in comp.members] + [_process_brick(comp)], name=comp.name, version=comp.version,
instructions=instructions, prefix_routes=True, hide=frozenset(hidden))
@composed.get("/composition", include_in_schema=False)
def manifest(_caller: dict = Depends(auth.current_user)):
"""What is running: bricks, versions, and every seam with whether its neighbour is here. Signed-in
callers only: deployed with optional login (so an agent can discover where to sign in), an
anonymous request would otherwise get a map of the system."""
return JSONResponse(report)
return composed
def _process_brick(comp: Composition) -> Brick:
"""ONE agent tool for every brick's process view — `show_process_bpmn(brick, process_id?)` — since each brick
serves its own as a route (`GET /<brick>/process/bpmn`) and one tool name cannot be offered twice."""
from erp_platform.bricks import Refused
from typing import Optional as Opt
b = Brick("composition", version=comp.version)
def show_process_bpmn(caller: dict, brick: str, process_id: Opt[str] = None) -> dict:
"""A brick's process as BPMN 2.0 — the diagram the customer sees and signs before the build (erp-compose
signoff record). `brick` is the brick's name in this composition; `process_id` picks one process (an
entity's workflow, or brick-hr's playbook id). Read-only."""
m = comp.member(brick)
if m is None:
raise Refused(404, f"no brick {brick} here (there are: {', '.join(x.name for x in comp.members)})")
uc = next((u for u in m.brick.use_cases if u.name == m.brick.process_view), None)
if uc is None:
raise Refused(404, f"{brick} has no process view" + (f" (waived: {m.process_view_waiver})"
if m.process_view_waiver else ""))
if uc.permission and not auth.has_permission(caller, uc.permission):
raise Refused(403, f"showing {brick}'s process needs {uc.permission}")
if uc.name == "show_playbook_bpmn":
if not process_id:
raise Refused(422, f"{brick} shows a playbook: give process_id = the playbook id")
return uc.fn(caller, playbook_id=process_id)
return uc.fn(caller, process_id=process_id)
b.use_case(kind="query", route="GET /process/{brick}")(show_process_bpmn)
return b
def public_app(path_or_comp) -> FastAPI:
"""The composition's PUBLIC app: the public-link door with every brick's public handlers, /healthz and
/readyz — and nothing else. No brick route, no /mcp, no /composition, no OpenAPI; it never migrates.
Deployed as its own Ignite app: no user pool (public_invoke), its own service account, access log off."""
comp = path_or_comp if isinstance(path_or_comp, Composition) else load(path_or_comp)
report = check(comp)
if report["problems"]:
raise CompositionRefused("\n".join(report["problems"]))
if comp.tier != 1:
raise CompositionRefused(f"tier {comp.tier}: compose.public_app() serves tier 1's one bucket")
return public.app([h for m in comp.members for h in m.brick.public_handlers],
name=f"{comp.name}-public", version=comp.version)
# What the public app's deploy carries, and what it must not.
PUBLIC_ENV = ("BUCKET", "DODIL_ENV", "DODIL_SERVICE_ACCOUNT_ID", "DODIL_SERVICE_ACCOUNT_SECRET",
"PUBLIC_LINK_BASE_URL", "PUBLIC_REDIRECT_HOSTS")
PUBLIC_NEVER = ("DODIL_USER_POOL", "AUTH_ISSUER", "SEAM_CLIENT_SECRET", "CAPABILITY_*", "model or connector keys")
# ── the command line ───────────────────────────────────────────────────────────────────────
def brick_paths(root) -> None:
"""Make the folder's bricks importable for an erp-compose run from OUTSIDE its .venv (uvx at a newer
erp-platform than the bricks pin): owned/<repo>/src and bricks/<repo>/src first, then the folder's .venv
site-packages LAST — so erp_platform itself still resolves to the running version."""
root = pathlib.Path(root).resolve()
for d in ("owned", "bricks"):
for src in sorted((root / d).glob("*/src")):
if str(src) not in sys.path:
sys.path.insert(0, str(src))
for site in sorted((root / ".venv" / "lib").glob("python*/site-packages")):
if str(site) not in sys.path:
sys.path.append(str(site))
def _print_report(r: dict) -> None:
print(f"{r['name']} {r['version']} — tier {r['tier']}, {len(r['bricks'])} bricks, {r['tools']} tools")
for b in r["bricks"]:
print(f" {b['name']:<10} {b['version']:<8} {b['tools']:>2} tools {', '.join(b['tables'])}")
for s in r["seams"]:
print(f" seam {s['from']} → {s['to']}: {'present' if s['present'] else 'ABSENT (optional — it says so when asked)'}")
for c in r.get("capabilities", []):
who = f" (required by {', '.join(c['required_by'])})" if c["required_by"] else " (required by none)"
how = (f"→ {c['provider']}" if c["provider"] != "manual"
else "→ manual (no provider: the consumer makes these steps manual tasks)")
print(f" capability {c['capability']} {how}{who}")
for h in r.get("public", []):
print(f" public {h['brick']}: {h['grant']} {h['subject_kind']} → {h['name']}"
+ (" (+ act)" if h["acts"] else ""))
for v in r.get("process_views", []):
if v.get("view"):
print(f" process {v['brick']}: {v['view']}")
elif v.get("waived"):
print(f" process {v['brick']}: WAIVED — {v['waived']}")
for line in r.get("role_lines", []):
print(line)
for w in r.get("warnings", []):
print(f" WARNING {w}", file=sys.stderr)
for p in r["problems"]:
print(f" PROBLEM {p}", file=sys.stderr)
def _permissions(comp: Composition, report: dict, path, roles_file: Optional[str], strict: bool) -> None:
"""0.9.2: each role's effective permissions per brick, wildcard warnings, and role-name decisions in brick
source (a warning; a problem under --strict). See erp_platform.permissions_check."""
from erp_platform import permissions_check as pc
for use in pc.role_name_uses(comp):
say = (f"a brick decides by permission, not a role name: {use} — use a permission (owner_scope(see_all=…)), "
"or exempt the line with `# role-check: <reason>`")
(report["problems"] if strict else report["warnings"]).append(say)
lines = report.setdefault("role_lines", [])
src = pc.find_roles(path, roles_file)
if src is None:
lines.append(" roles: no role catalogue (no roles.yaml beside compose.yaml or in tests/scenarios/; pass --roles "
"FILE — e.g. `dodil appid roles get <pool> --org <org> -o json`)")
return
try:
roles, services = pc.load_roles(src)
except (OSError, yaml.YAMLError) as e:
report["problems"].append(f"--roles {src}: {e}")
return
r = pc.report(comp, roles, services)
pc.print_report(r, src, lines.append)
report["warnings"].extend(r["warnings"])
report["roles"] = {"source": str(src), "effective": r["effective"]}
_TAIL = " [--strict] [--roles FILE] [--unsigned \"<reason>\"] <compose.yaml>"
VERB_USAGE = {
"check": "usage: erp-compose check [--strict] [--roles FILE] <compose.yaml>\n"
" Load the composition and check its wiring, capabilities and permissions. Touches no database.",
"plan": "usage: erp-compose plan" + _TAIL + "\n"
" The tier 2/3 wiring: one app per brick and its env, and the public app. Refused for an unsigned "
"design.",
"migrate": "usage: erp-compose migrate" + _TAIL + "\n"
" erp-compose migrate --local [DIR] [--unsigned \"<reason>\"] <compose.yaml>\n"
" Apply each brick's schema.sql to the DataK3 bucket (BUCKET, the pg wire, a service-account token),\n"
" then insert the rows a new bucket starts with (each brick's seed; re-run safe).\n"
" Refused for a design the customer has not signed (--unsigned \"<reason>\" records why it goes on).\n"
" --local [DIR]: the DEV tier — SQLite files in DIR (default .erp-local/ beside compose.yaml), "
"created\n from the bricks' own tables; the sign-off gate warns. Refused when BUCKET is set.",
"serve": "usage: erp-compose serve [--public]" + _TAIL + "\n"
" erp-compose serve --local [DIR] [--clock +1d|INSTANT] [--ticks-every 5|--no-ticks] <compose.yaml>\n"
" Serve the composition (PORT, default 8000): the staff app, or with --public the public app.\n"
" --local [DIR]: the DEV tier — the SQLite files `migrate --local` made, on 127.0.0.1, with the demo\n"
" clock mounted; --clock starts the business clock there (settings take effect from tomorrow);\n"
" the bricks' scheduled passes (ar post_pending, raise_due…) run every 5 s unless --no-ticks.",
}
USAGE = ("usage: erp-compose init <module> [<module> …] [--starter NAME …] [--dir DIR] [--method-ref REF] [--no-install]\n"
" erp-compose signoff record|verify --module M --bpmn FILE … | signoff record --all --approver A --reason R\n"
" erp-compose bpmn --from module.yaml|ERP_<MODULE>.md [--workflow ENTITY] [--out process.bpmn]\n"
" erp-compose process --module M --playbook P [--calls config/M.calls.yaml] [--out processes] [compose.yaml]\n"
" erp-compose scaffold <module.yaml> --out DIR [--entity E] [--dist NAME] [--force]\n"
" erp-compose own <brick> --reason \"…\"\n"
" erp-compose change propose|withdraw --module M (--bpmn FILE | --from ERP_M.md)\n"
" erp-compose check|plan|migrate|serve [--public] [--strict] [--roles FILE] [--unsigned \"<reason>\"] <compose.yaml>\n"
" erp-compose migrate|serve --local [DIR] [--clock +1d] <compose.yaml> (the dev tier, SQLite)\n"
" erp-compose seed [--local [DIR]] <compose.yaml> <CALLS.yaml> (configuration calls, re-runnable)\n"
" erp-compose deploy compose.yaml --org ORG --name APP [--repo R] [--warm] [--demo-clock]\n"
" erp-compose teardown compose.yaml --org ORG [--keep-repo]\n"
" erp-compose <command> --help: one command's usage")
def main(argv=None) -> int:
"""erp-compose init <module>… | signoff record|verify … | process --module M --playbook P … | check|plan|migrate|serve [--public] <compose.yaml>"""
argv = list(sys.argv[1:] if argv is None else argv)
if argv and argv[0] == "init":
from erp_platform import compose_init
return compose_init.main(argv[1:])
if argv and argv[0] == "signoff":
from erp_platform import signoff
return signoff.main(argv[1:])
if argv and argv[0] == "change":
from erp_platform import change
return change.main(argv[1:])
if argv and argv[0] == "own":
from erp_platform import compose_init
return compose_init.own_main(argv[1:])
if argv and argv[0] == "scaffold":
from erp_platform import scaffold
return scaffold.main(argv[1:])
if argv and argv[0] == "bpmn":
from erp_platform import bpmn
return bpmn.main(argv[1:])
if argv and argv[0] == "process":
from erp_platform import process_export
return process_export.main(argv[1:])
if argv and argv[0] in ("deploy", "teardown"):
from erp_platform import deploy # 0.9.12: git push → CI → registry → CD, and back
return deploy.main(argv)
if argv and argv[0] == "seed":
from erp_platform import seed # 0.9.10: configuration calls, replayed and re-runnable
return seed.main(argv[1:])
if not argv or argv[0] in ("-h", "--help", "help"):
print(USAGE)
return 0
if argv[0] in VERB_USAGE and any(a in ("-h", "--help") for a in argv[1:]):
# 0.9.9: `migrate --help` read "--help" as the compose.yaml path and crashed (the small-ERP dry run)
print(VERB_USAGE[argv[0]])
return 0
local, local_dir = False, None
for i, a in enumerate(argv):
if a == "--local" or a.startswith("--local="):
local = True
if "=" in a:
local_dir, n = a.split("=", 1)[1] or None, 1
elif i + 1 < len(argv) and not argv[i + 1].startswith("-") and \
not argv[i + 1].endswith((".yaml", ".yml")) and len(argv) - (i + 2) >= 1:
local_dir, n = argv[i + 1], 2
else:
n = 1
del argv[i:i + n]
break
ticks_every, no_ticks = None, "--no-ticks" in argv # 0.9.11: serve --local runs the bricks' passes
argv = [a for a in argv if a != "--no-ticks"]
if "--ticks-every" in argv:
i = argv.index("--ticks-every")
try:
ticks_every = float(argv[i + 1])
assert ticks_every > 0
except (IndexError, ValueError, AssertionError):
print("--ticks-every needs a number of seconds > 0 (default 5)", file=sys.stderr)
return 2
del argv[i:i + 2]
clock = None
if "--clock" in argv:
i = argv.index("--clock")
if i + 1 >= len(argv):
print("--clock needs an offset (+1d) or an instant (2026-11-20T09:00Z)", file=sys.stderr)
return 2
clock = argv[i + 1]
del argv[i:i + 2]
as_public = "--public" in argv
argv = [a for a in argv if a != "--public"]
strict = "--strict" in argv
argv = [a for a in argv if a != "--strict"]
roles_file = None
if "--roles" in argv:
i = argv.index("--roles")
if i + 1 >= len(argv):
print("--roles needs a file: a roles.yaml, or `dodil appid roles get <pool> -o json` saved", file=sys.stderr)
return 2
roles_file = argv[i + 1]
del argv[i:i + 2]
unsigned = None
if "--unsigned" in argv:
i = argv.index("--unsigned")
if i + 1 >= len(argv):
print("--unsigned needs the reason, in words: --unsigned \"<why the customer's sign-off is skipped>\"",
file=sys.stderr)
return 2
unsigned = argv[i + 1]
del argv[i:i + 2]
if len(argv) < 2 or argv[0] not in VERB_USAGE:
print(USAGE, file=sys.stderr)
return 2
verb, path = argv[0], argv[1]
if as_public and verb != "serve":
print("--public goes with serve: the public app never migrates", file=sys.stderr)
return 2
if local and verb not in ("migrate", "serve"):
print("--local goes with migrate or serve (check and plan touch no database)", file=sys.stderr)
return 2
if clock is not None and not (local and verb == "serve"):
print("--clock goes with serve --local: a deployment moves its clock through the demo-clock route",
file=sys.stderr)
return 2
if (no_ticks or ticks_every is not None) and not (local and verb == "serve"):
print("--ticks-every / --no-ticks go with serve --local: a deployment runs its passes from its clock worker",
file=sys.stderr)
return 2
if local and as_public:
print("--public is not served by the local tier: the staff app on 127.0.0.1 is", file=sys.stderr)
return 2
if local:
from erp_platform import local as local_tier
try:
local_tier.guard()
except local_tier.LocalRefused as e:
print(e, file=sys.stderr)
return 1
root = pathlib.Path(path).resolve().parent
brick_paths(root)
try:
comp = load(path)
except CompositionRefused as e:
print(e, file=sys.stderr)
return 1
report = check(comp)
_permissions(comp, report, path, roles_file, strict)
_print_report(report)
if report["problems"]:
return 1
if verb == "check":
return 0
from erp_platform import gate
# 0.8: plan and migrate are refused for a design the customer has not signed as a diagram; serve warns.
if not gate.enforce(f"erp-compose {verb}" + (" --local" if local else ""), root, compose_path=path,
unsigned=unsigned, refuse=verb in ("plan", "migrate") and not local,
say=lambda line: print(line, file=sys.stderr)):
return 1
if local:
# 0.9.9: the dev tier — SQLite files on this machine, the bricks' own create path (erp_platform.local)
data = local_tier.data_dir(root, local_dir)
local_tier.banner(print, verb, data)
try:
if verb == "migrate":
return local_tier.migrate(comp, data)
staged_app, st = local_tier.prepare_serve(comp, data, clock=clock)
local_tier.start_ticks(comp, staged_app, st,
every=None if no_ticks else (ticks_every or local_tier.TICK_EVERY))
except local_tier.LocalRefused as e:
print(e, file=sys.stderr)
return 1
import uvicorn
port = int(__import__("os").environ.get("PORT", "8000"))
with st:
uvicorn.run(staged_app, host=local_tier.HOST, port=port)
return 0
if verb == "plan":
print("\ntier 2/3 wiring — one app per brick, each with:")
wired = {c["capability"]: c["provider"] for c in report["capabilities"]}
for m in comp.members:
env = sorted({s.base_env for s in m.seams() if comp.member(s.name)})
for c in m.requires():
p = wired.get(c, capabilities.MANUAL)
cap = capabilities.contract().get(c)
env.append(f"{cap.env}={p}")
if p != capabilities.MANUAL:
env.append(capabilities.base_env(p))
env = sorted(set(env))
print(f" {m.name}: app {m.package}.app:app" + (f" env {', '.join(env)}" if env else ""))
handlers = report.get("public", [])
print(f"\npublic app ({len(handlers)} handler{'s' if len(handlers) != 1 else ''}) — "
"`erp-compose serve --public compose.yaml`, deployed as its OWN app:")
print(" no user pool (public_invoke true) · access log off · never migrates · only /p /o /c /healthz /readyz")
print(f" env {', '.join(PUBLIC_ENV)}")
print(" its OWN service account, narrower than the staff app's: k3.editor scoped to this one bucket and "
"nothing else (no ignite.model-user, no connector credential). DataK3 roles are per bucket, not per "
"table — the narrowing is what this image holds and is granted, not a table filter.")
print(f" never set {', '.join(PUBLIC_NEVER)}")
print(" the staff app sets PUBLIC_LINK_BASE_URL too (to issue links); rate-limit /p /o /c at the gateway")
return 0
if verb == "migrate":
from erp_platform import migrate
code = migrate.migrate_platform()
if code:
return code
for m in comp.members:
print(f"\n── {m.name} {m.brick.version} ──")
code = migrate.migrate(m.schema(), {m.name: m.brick.version}) or migrate.seed(m) # 0.9.11: + its seed
if code:
return code
return 0
import uvicorn
port = int(__import__("os").environ.get("PORT", "8000"))
if as_public:
# access_log=False: uvicorn's access line is the RAW path, and a public path carries the token.
uvicorn.run(public_app(comp), host="0.0.0.0", port=port, access_log=False)
return 0
uvicorn.run(app(comp), host="0.0.0.0", port=port)
return 0
if __name__ == "__main__":
raise SystemExit(main())