"""erp-compose deploy / teardown (0.9.12) — ship a signed composition the proper way, and take it all back.
erp-compose deploy compose.yaml --org ORG --name APP [--repo NAME] [--warm] [--demo-clock]
erp-compose teardown compose.yaml --org ORG [--keep-repo]
THE PATH. The composition gets its OWN private Dodil Git repo; `git push` runs the forge's CI (the image is built and
pushed to the org registry by `.github/workflows/build.yml`, the customer's sign-off is checked by
`.github/workflows/ci.yml`), and the forge's CD deploys from `.dodil/deploy.yaml` once `build/build` and
`ci/signoff` are green. `dodil ignite app deploy --code` is never used: it is the throwaway path, capped near 1 MB
(seemo, 2026-10-25).
THE STEPS (each prints what it did; the first refusal stops the run and names the fix; a re-run resumes):
1. gate — `signoff verify --all` must pass: the customer signed every process.
2. provision — bucket, appid pool (roles MERGED into the pool's catalogue, never replaced; sign-up off, the app's
callback added to the redirect URIs, CIMD on), the app's service account (k3.editor scoped to THIS
bucket and nothing else), a machine service client for the MCP checks, the repo's CI secrets.
3. migrate — `erp-compose migrate` on the bucket, as the app's own service account, then the starters' seed.
4. files — Dockerfile, app.py, the two workflows, .dodil/deploy.yaml, .gitignore/.dockerignore (a file without
erp-compose's marker is the customer's and is kept).
5. ship — the private repo, `dodil git ci verify`, commit, push; follow the checks and the CD deploy.
6. verify — GET /healthz, MCP initialize + tools/list with the service client's token.
7. inventory — .dodil/inventory.json: every id and its exact delete command (no secret).
SECRETS never reach argv, stdout or the repo: the service account's and the service client's secrets are read
from the CLI's JSON answer into memory, kept in ~/.dodil-secrets/<app>/ (0600), handed to `dodil git secret set`
on stdin and to `migrate` in its environment. Every argv is checked against them before it runs, and every file
about to be committed is scanned for them.
THE ORG. Every call carries `--org <org>` (`--org-name` for git): the session's active org is never switched.
"""
import argparse
import datetime as _dt
import json
import os
import pathlib
import shutil
import subprocess
import sys
import tempfile
import time
from typing import Callable, Optional
MARKER = "written by erp-compose deploy"
CI_INIT_MARKER = "Built by 'dodil git ci init'" # cli-git c310bf1 (CLI 0.1.86): its two files are ours to refine
PORT = 8000
INVENTORY = ".dodil/inventory.json"
SA_SERVICE, SA_ROLE = "k3-authorization-service", "k3.editor"
MACHINE_ROLE = "erp_machine"
REQUIRED_CHECKS = ("build/build", "ci/signoff")
DEPLOY_CHECK = "dodil-ci/deploy"
DEPLOY_CONFIG_CHECK = "dodil-ci/deploy-config"
TERMINAL = {"passed", "failed", "errored", "cancelled", "skipped"}
class Refused(Exception):
"""A step refused: the message says what and the fix. Nothing after it ran."""
def _say(line=""):
print(line, flush=True)
def default_bin() -> str:
env = os.environ.get("DODIL_BIN")
if env:
return env
found = shutil.which("dodil")
if found:
return found
return str(pathlib.Path.home() / ".local" / "bin" / "dodil")
# ── the dodil CLI, with every argv recorded and checked for secrets ────────────────────────────────────────────────
class Dodil:
def __init__(self, org: str, bin: Optional[str] = None, runner: Optional[Callable] = None):
self.org, self.bin = org, bin or default_bin()
self.runner = runner or subprocess.run
self.secrets: set = set()
self.calls: list = []
def keep_secret(self, value: Optional[str]) -> None:
if value:
self.secrets.add(value)
def guard(self, argv) -> None:
for s in self.secrets:
if any(s in str(a) for a in argv):
raise Refused("refusing to run a dodil command with a secret in its arguments (an erp-compose bug)")
def run(self, *args, stdin: Optional[str] = None, timeout: float = 600) -> tuple:
flag = "--org-name" if args and args[0] == "git" else "--org"
argv = [self.bin, *[str(a) for a in args], flag, self.org, "-o", "json"]
self.guard(argv)
self.calls.append({"argv": argv[1:], "stdin": stdin is not None})
p = self.runner(argv, input=stdin, capture_output=True, text=True, timeout=timeout)
return p.returncode, p.stdout or "", p.stderr or ""
def json(self, *args, stdin=None, what=None, timeout: float = 600):
rc, out, err = self.run(*args, stdin=stdin, timeout=timeout)
if rc != 0:
raise Refused(f"`dodil {' '.join(map(str, args))}` failed: {self._scrub((err or out).strip())[:600]}"
+ (f"\n fix: {what}" if what else ""))
return _loads(out)
def get(self, *args):
"""The resource, or None when the CLI says it does not exist (any other failure refuses)."""
rc, out, err = self.run(*args)
if rc == 0:
return _loads(out)
text = (err + out).lower()
if "not found" in text or "404" in text or "no such" in text or "does not exist" in text:
return None
raise Refused(f"`dodil {' '.join(map(str, args))}` failed: {self._scrub((err or out).strip())[:600]}")
def _scrub(self, text: str) -> str:
for s in self.secrets:
text = text.replace(s, "***")
return text
def _loads(out: str):
out = (out or "").strip()
if not out:
return {}
try:
return json.loads(out)
except ValueError:
i = min([j for j in (out.find("{"), out.find("[")) if j >= 0] or [-1])
if i >= 0:
try:
return json.loads(out[i:])
except ValueError:
pass
return {"text": out}
# ── the local secret store (outside the repo) ──────────────────────────────────────────────────────────────────────
class Secrets:
def __init__(self, directory: pathlib.Path):
self.dir = pathlib.Path(directory)
def read(self, name: str) -> Optional[dict]:
p = self.dir / name
return json.loads(p.read_text()) if p.is_file() else None
def write(self, name: str, data: dict) -> pathlib.Path:
self.dir.mkdir(parents=True, exist_ok=True)
os.chmod(self.dir, 0o700)
p = self.dir / name
fd = os.open(p, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
with os.fdopen(fd, "w") as f:
json.dump(data, f)
return p
def remove(self) -> None:
if self.dir.is_dir():
shutil.rmtree(self.dir)
def _find(d, *keys):
"""The first non-empty value under any of `keys`, searching nested dicts."""
if isinstance(d, dict):
for k in keys:
if d.get(k):
return d[k]
for v in d.values():
r = _find(v, *keys)
if r:
return r
return None
# ── the generated files ────────────────────────────────────────────────────────────────────────────────────────────
def dockerfile() -> str:
return f"""# {MARKER}: edit freely, then delete this line so a re-run keeps your version.
# The composition's image: the bricks and erp-platform come from requirements.txt, PINNED at their tags (public
# git.dodil.io repos, so no credential is in the build). Built by .github/workflows/build.yml on every push.
FROM python:3.12-slim
RUN apt-get update && apt-get install -y --no-install-recommends git ca-certificates \\
&& rm -rf /var/lib/apt/lists/* && useradd --system --uid 10001 --home /app erp
WORKDIR /app
COPY . /app
RUN pip install --no-cache-dir -r requirements.txt "uvicorn>=0.34" \\
&& python -c "import erp_platform; print('erp-platform', erp_platform.__version__)"
USER 10001
ENV PORT={PORT} PYTHONUNBUFFERED=1
EXPOSE {PORT}
CMD ["sh", "-c", "exec uvicorn app:app --host 0.0.0.0 --port ${{PORT}} --proxy-headers --forwarded-allow-ips='*'"]
"""
def app_py(demo_clock_brick: Optional[str]) -> str:
clock = ""
if demo_clock_brick:
clock = (f"\n# --demo-clock: the business clock's read/set tools on {demo_clock_brick} (a DEMO deployment; refused "
"under DODIL_ENV=prod|live).\nfrom erp_platform import democlock\n"
f"democlock.mount(compose.load(_COMPOSE).member({demo_clock_brick!r}).brick)\n")
return f'''"""The composition as one app: every brick's routes under /<module>, one /mcp for every tool. ({MARKER})"""
import pathlib
from erp_platform import compose
_COMPOSE = pathlib.Path(__file__).with_name("compose.yaml")
{clock}app = compose.app(_COMPOSE)
'''
def build_yml(org: str, app: str) -> str:
"""The same workflow `dodil git ci init` writes (cli-git c310bf1): workflow "build", job "build"."""
return f"""# {MARKER} (the workflow `dodil git ci init` writes). Every push builds the image and pushes it to
# the org registry; .dodil/deploy.yaml then deploys it once this check (build/build) is green.
# DODIL_REGISTRY_URL / DODIL_REGISTRY_TOKEN are injected into every job — never create or store them.
name: build
on: [push]
jobs:
build:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- uses: docker/login-action@v3
with:
registry: ${{{{ secrets.DODIL_REGISTRY_URL }}}} # REQUIRED — it defaults to docker.io
username: ci
password: ${{{{ secrets.DODIL_REGISTRY_TOKEN }}}}
- run: |
IMG="${{{{ secrets.DODIL_REGISTRY_URL }}}}/{org}/{app}:${{{{ github.sha }}}}"
docker build -t "$IMG" .
docker push "$IMG"
"""
def ci_yml() -> str:
return f"""# {MARKER}. Check name: ci/signoff — the customer's sign-off, checked on every push; CD waits for it.
# The forge runner's python is externally managed (PEP 668): python 3.12 comes from uv into a venv. One job per
# workflow here: parallel actions/setup-python jobs on one runner collide on the shared toolcache.
name: ci
on:
push: {{ branches: [main] }}
pull_request: {{ branches: [main] }}
workflow_dispatch: {{}}
jobs:
signoff:
runs-on: ubuntu-latest
timeout-minutes: 20
steps:
- uses: actions/checkout@v4
- run: pip install -q --break-system-packages uv && uv venv --python 3.12 -q /tmp/py
- run: uv pip install -q --python /tmp/py/bin/python -r requirements.txt
- run: /tmp/py/bin/erp-compose signoff verify --all
"""
def deploy_yaml(org: str, app: str, pool: str, bucket: str, demo_clock: bool) -> str:
extra = ' ERP_DEMO_CLOCK_OFFSET: "+1d"\n' if demo_clock else ""
return f"""# {MARKER}. The forge's CD (schema v1, dodil-git crates/git-ci-runner/src/deploy_config.rs): after a push to
# main whose required checks passed, it deploys the image the build pushed. `env` is DECLARATIVE: it replaces the
# app's whole environment on every deploy. The two secrets live in the repo's CI secret store
# (`dodil git secret set`), never in git. Reserved (warm) capacity is not a deploy.yaml key: `--warm` sets it on
# the app (`dodil ignite app update {app} --reserved 1`), and a redeploy leaves it alone.
version: 1
create_if_missing: true
apps:
- name: {app}
image: "{{registry}}/{org}/{app}:{{sha}}"
port: {PORT}
health: {{ path: /healthz }}
public_invoke: true
user_pool: {pool}
group: {app}
env:
BUCKET: {bucket}
DODIL_ENV: dev
DODIL_USER_POOL: {org}/{pool}
DODIL_SERVICE_ACCOUNT_ID: "${{{{ secrets.DODIL_SERVICE_ACCOUNT_ID }}}}"
DODIL_SERVICE_ACCOUNT_SECRET: "${{{{ secrets.DODIL_SERVICE_ACCOUNT_SECRET }}}}"
{extra}when:
branches: [main]
require_checks: [{", ".join(REQUIRED_CHECKS)}]
"""
GITIGNORE = f"# {MARKER}\n.venv/\n.erp-local/\n__pycache__/\n*.pyc\n.env\n.dodil/inventory.json\n"
DOCKERIGNORE = f"# {MARKER}\n.git\n.venv\n.erp-local\n__pycache__\n*.pyc\n.env\n.dodil\n.github\ntests\n"
def write_file(root: pathlib.Path, rel: str, text: str, say) -> str:
p = root / rel
if p.exists():
cur = p.read_text()
if cur == text:
say(f" · {rel}: unchanged")
return "unchanged"
if MARKER not in cur and CI_INIT_MARKER not in cur:
say(f" · {rel}: kept (yours: no '{MARKER}' line)")
return "kept"
p.parent.mkdir(parents=True, exist_ok=True)
p.write_text(text)
say(f" ✓ {rel}: written")
return "written"
def seed_files(root: pathlib.Path) -> list:
"""The starters' calls files, a base before what extends it (uk-wholesale before uk-wholesale-sample)."""
files = sorted((root / "config").glob("*.calls.yaml"), key=lambda p: p.name[: -len(".calls.yaml")])
return [str(p.relative_to(root)) for p in files]
# ── deploy ─────────────────────────────────────────────────────────────────────────────────────────────────────────
class Deploy:
def __init__(self, compose_path, org: str, name: str, *, repo: Optional[str] = None, warm: bool = False,
demo_clock: bool = False, dodil: Optional[Dodil] = None, secrets_dir=None, say=_say,
local_run: Optional[Callable] = None, git_run: Optional[Callable] = None,
http=None, sleep=time.sleep, follow_timeout: float = 3600, poll: float = 15):
self.compose_path = pathlib.Path(compose_path).resolve()
self.root = self.compose_path.parent
self.org, self.app = org, name
self.bucket = self.pool = name
self.repo = repo or name
self.sa_name = f"{name}-app"
self.client_id = f"{name}-mcp-check"
self.warm, self.demo_clock = warm, demo_clock
self.d = dodil or Dodil(org)
self.secrets = Secrets(secrets_dir or pathlib.Path.home() / ".dodil-secrets" / name)
self.say = say
self.local_run = local_run or self._local_run
self.git_run = git_run or self._git_run
self.http = http
self.sleep, self.follow_timeout, self.poll = sleep, follow_timeout, poll
self.inv = load_inventory(self.root) or {}
self.inv.update({"org": org, "app": name, "bucket": self.bucket, "pool": self.pool, "repo": self.repo})
self.inv.setdefault("resources", {})
self.inv.setdefault("timings_s", {})
self.fqdn = f"{name}-{org}-{PORT}.ignite.dodil.cloud"
self._sa = None
self._rotated = False
# bookkeeping
def _res(self, key, ident, delete, **extra):
self.inv["resources"][key] = {"id": ident, "delete": delete, **extra}
save_inventory(self.root, self.inv)
def _timed(self, step, fn):
self.say(f"\n── {step} ──")
t = time.time()
out = fn()
self.inv["timings_s"][step] = round(time.time() - t, 1)
save_inventory(self.root, self.inv)
return out
def run(self) -> int:
try:
self._timed("gate", self.gate)
self._timed("provision", self.provision)
self._timed("migrate", self.migrate)
self._timed("files", self.files)
sha = self._timed("push", self.push)
self._timed("ci", lambda: self.follow(sha, REQUIRED_CHECKS))
self._timed("cd", lambda: self.follow(sha, (DEPLOY_CHECK,)))
self._timed("app", self.app_ready)
self._timed("verify", self.verify)
except Refused as e:
self.say(f"\nREFUSED: {e}")
save_inventory(self.root, self.inv)
return 1
self.inv["deployed_at"] = _now()
save_inventory(self.root, self.inv)
self.say(f"\nlive: https://{self.fqdn}\nmcp: https://{self.fqdn}/mcp\ninventory: {self.root / INVENTORY}"
f"\ntimings (s): {json.dumps(self.inv['timings_s'])}")
return 0
# 1
def gate(self):
from erp_platform import gate
rows = gate.check(self.root, str(self.compose_path), None)
bad = [r for r in rows if r["state"] != "signed"]
if not rows or bad:
lines = "\n".join(f" {gate.label(r)}: {r['state'].upper()} — {r['detail']}" for r in bad[:12]) or \
" nothing to verify: no modules"
raise Refused("the customer has not signed every process (`erp-compose signoff verify --all`):\n" + lines +
"\n fix: show the customer the processes, then `erp-compose signoff record --all --approver "
"\"<name>\" --reason \"<what they approved>\"`")
self.say(f" ✓ signoff verify --all: {len(rows)} process(es) signed")
# 2
def provision(self):
d = self.d
b = d.get("data", "bucket", "get", self.bucket)
if b is None:
d.json("data", "bucket", "create", self.bucket, "-d", f"erp-compose deploy: {self.app}")
b = d.json("data", "bucket", "get", self.bucket)
self.say(f" ✓ bucket {self.bucket}: created")
else:
self.say(f" · bucket {self.bucket}: there")
drn = _find(b, "drn") or ""
if f"bucket/{self.bucket}" not in drn:
raise Refused(f"bucket {self.bucket}: no DRN in `dodil data bucket get` ({json.dumps(b)[:200]})")
self._res("bucket", self.bucket, f"dodil data bucket delete {self.bucket} --org {self.org}", drn=drn)
tables_sa = [s for s in _sas(d, f"k3-tables-sa-{self.bucket}-")]
if tables_sa:
s = tables_sa[0]
self._res("bucket_tables_sa", s["uuid"], f"dodil auth service-account delete {s['uuid']} --yes --org "
f"{self.org}", made_by="the platform, with the bucket", name=s["id"])
p = d.get("appid", "pool", "get", self.pool)
if p is None:
d.json("appid", "pool", "create", self.pool, "--with-local", "--display-name", self.app)
p = d.json("appid", "pool", "get", self.pool)
self.say(f" ✓ appid pool {self.pool}: created (email + password on)")
else:
self.say(f" · appid pool {self.pool}: there")
self._res("pool", f"{self.org}/{self.pool}", f"dodil appid pool delete {self.pool} --yes --org {self.org}")
self._roles()
self._settings(p)
self._service_account(drn)
self._service_client()
self._repo()
self._ci_secrets()
def _catalog(self) -> dict:
"""The roles this composition needs: roles.yaml's (or manager = *:*:*), + the machine role."""
from erp_platform import auth, compose, permissions_check
comp = compose.load(str(self.compose_path))
found = permissions_check.find_roles(self.compose_path)
roles = permissions_check.load_roles(found)[0] if found else {"manager": ["*:*:*"]}
machine = sorted(p for p in auth.machine_permissions() if p != auth.ACT_FOR)
roles[MACHINE_ROLE] = sorted(set(roles.get(MACHINE_ROLE, [])) | set(machine))
self._comp = comp
return roles
def _roles(self):
d = self.d
cur = d.json("appid", "roles", "get", self.pool) or {}
cur = {k: list(v) for k, v in cur.items() if isinstance(v, list)}
want = self._catalog()
merged = {k: list(v) for k, v in cur.items()}
for role, perms in want.items():
have = merged.setdefault(role, [])
have.extend(p for p in perms if p not in have)
if merged == cur:
self.say(f" · pool roles: {', '.join(sorted(cur))} (nothing to add)")
return
with tempfile.NamedTemporaryFile("w", suffix=".json", delete=False) as f:
json.dump(merged, f)
try:
d.json("appid", "roles", "set", self.pool, "--file", "@" + f.name)
finally:
os.unlink(f.name)
added = sorted(set(merged) - set(cur))
self.say(f" ✓ pool roles merged (kept {len(cur)}, added {', '.join(added) or 'permissions only'})")
def _settings(self, pool_doc):
raw = _find(pool_doc, "settingsJson")
try:
cur = json.loads(raw) if isinstance(raw, str) and raw.strip() else {}
except ValueError:
cur = {}
cb = f"https://{self.fqdn}/.dodil/auth/callback"
uris = list(cur.get("redirect_uris") or [])
want = dict(cur, allow_signup=False, cimd_enabled=True, redirect_uris=uris + ([cb] if cb not in uris else []))
if want == cur:
self.say(" · pool settings: sign-up off, CIMD on, callback listed")
return
with tempfile.NamedTemporaryFile("w", suffix=".json", delete=False) as f:
json.dump(want, f)
try:
self.d.json("appid", "pool", "update", self.pool, "--settings", "@" + f.name)
finally:
os.unlink(f.name)
self.say(f" ✓ pool settings: sign-up off, CIMD on, redirect URIs + {cb}")
def _service_account(self, drn):
d = self.d
mine = [s for s in _sas(d, self.sa_name) if s["id"].split("@")[0] in (self.sa_name, f"cli-{self.sa_name}")]
stored = self.secrets.read("sa.json")
if not mine:
out = d.json("auth", "service-account", "create", self.sa_name)
uuid, sid = _find(out, "uuid", "serviceAccountUuid"), _find(out, "serviceAccountId")
cid, secret = _find(out, "client_id", "clientId") or sid, _find(out, "secret", "clientSecret")
if not (uuid and cid and secret):
raise Refused("`dodil auth service-account create` answered without uuid/client_id/secret")
stored = {"uuid": uuid, "id": sid, "client_id": cid, "secret": secret}
self.secrets.write("sa.json", stored)
self._rotated = True
self.say(f" ✓ service account {sid}: created (secret → {self.secrets.dir}/sa.json, 0600)")
else:
s = mine[0]
if not stored or stored.get("uuid") != s["uuid"]:
out = d.json("auth", "service-account", "rotate", s["uuid"],
what="the secret is not on this machine: rotating mints a new one")
secret = _find(out, "secret", "clientSecret")
if not secret:
raise Refused("`dodil auth service-account rotate` answered without a secret")
stored = {"uuid": s["uuid"], "id": s["id"], "client_id": _find(out, "client_id", "clientId") or s["id"],
"secret": secret}
self.secrets.write("sa.json", stored)
self._rotated = True
self.say(f" ✓ service account {s['id']}: there, secret not on this machine → rotated")
else:
self.say(f" · service account {s['id']}: there")
d.keep_secret(stored["secret"])
self._sa = stored
self._res("service_account", stored["uuid"], f"dodil auth service-account delete {stored['uuid']} --yes "
f"--org {self.org}", name=stored["id"])
roles = d.json("auth", "service-account", "roles", stored["uuid"]) or {}
text = json.dumps(roles)
if SA_ROLE in text and drn in text:
self.say(f" · {SA_ROLE} on {drn}: granted")
else:
d.json("auth", "service-account", "grant-role", stored["uuid"], SA_SERVICE, SA_ROLE, "--on",
f"resource={drn}")
self.say(f" ✓ {SA_ROLE} granted on bucket/{self.bucket} only")
def _service_client(self):
d = self.d
listed = d.json("appid", "service-client", "list", self.pool) or {}
ids = [c.get("clientId") or c.get("client_id") for c in _items(listed, "serviceClients", "service_clients",
"clients", "items")]
stored = self.secrets.read("sc.json")
if self.client_id in ids and stored:
self.say(f" · service client {self.client_id}: there")
else:
if self.client_id in ids:
d.json("appid", "service-client", "delete", self.pool, self.client_id, "--yes")
self.say(f" · service client {self.client_id}: secret not on this machine → recreated")
roles = [MACHINE_ROLE] + (["manager"] if "manager" in (d.json("appid", "roles", "get", self.pool) or {})
else [])
out = d.json("appid", "service-client", "create", self.pool, "--client-id", self.client_id,
*[x for r in roles for x in ("--role", r)])
secret = _find(out, "clientSecret", "client_secret", "secret")
token_url = _find(out, "tokenEndpoint", "token_endpoint") or \
f"https://appid.dodil.io/{self.org}/{self.pool}/oauth/token"
if not secret:
raise Refused("`dodil appid service-client create` answered without a secret")
stored = {"client_id": self.client_id, "secret": secret, "token_endpoint": token_url, "roles": roles}
self.secrets.write("sc.json", stored)
self.say(f" ✓ service client {self.client_id} ({', '.join(roles)}): created "
f"(secret → {self.secrets.dir}/sc.json, 0600)")
d.keep_secret(stored["secret"])
self._sc = stored
self._res("service_client", self.client_id, f"dodil appid service-client delete {self.pool} {self.client_id} "
f"--yes --org {self.org}")
def _repo(self):
d = self.d
r = d.get("git", "repo", "get", self.repo)
if r is None:
d.json("git", "repo", "create", self.repo, "--visibility", "private", "--default-branch", "main",
"--description", f"ERP composition {self.app} (erp-compose deploy)")
r = d.json("git", "repo", "get", self.repo)
self.say(f" ✓ repo {self.org}/{self.repo}: created (private)")
else:
self.say(f" · repo {self.org}/{self.repo}: there ({r.get('visibility', '?')})")
if r.get("visibility") not in (None, "private"):
raise Refused(f"repo {self.repo} is {r.get('visibility')}: a composition's repo is private. fix: "
f"`dodil git repo visibility {self.repo} private --org-name {self.org}` or --repo <other>")
self._http_url = r.get("http_url") or f"https://git.dodil.io/{self.org}/{self.repo}.git"
self._res("repo", f"{self.org}/{self.repo}", f"dodil git repo delete {self.repo} --yes --org-name {self.org}")
def _ci_secrets(self):
d = self.d
listed = d.json("git", "secret", "list", self.repo) or {}
have = set()
for s in _items(listed, "secrets", "items") or (listed if isinstance(listed, list) else []):
have.add(s.get("name") if isinstance(s, dict) else s)
for name, value in (("DODIL_SERVICE_ACCOUNT_ID", self._sa["client_id"]),
("DODIL_SERVICE_ACCOUNT_SECRET", self._sa["secret"])):
if name in have and not self._rotated:
self.say(f" · CI secret {name}: there")
continue
d.json("git", "secret", "set", self.repo, name, stdin=value)
self.say(f" ✓ CI secret {name}: set (from stdin)")
self._res("ci_secrets", "DODIL_SERVICE_ACCOUNT_ID, DODIL_SERVICE_ACCOUNT_SECRET", "go with the repo")
# 3
def migrate(self):
stamp = _fingerprint(self.root)
if self.inv.get("migrated") == stamp:
self.say(" · migrate + seed: done for this compose.yaml/requirements.txt/config (re-run: delete "
"`migrated` from .dodil/inventory.json)")
return
env = dict(os.environ, BUCKET=self.bucket, DODIL_ENV="dev", DODIL_SERVICE_ACCOUNT_ID=self._sa["client_id"],
DODIL_SERVICE_ACCOUNT_SECRET=self._sa["secret"])
env.pop("DODIL_TOKEN", None)
steps = [("migrate", [str(self.compose_path)])] + \
[("seed", [str(self.compose_path), str(self.root / f)]) for f in seed_files(self.root)]
for verb, args in steps:
self.say(f" erp-compose {verb} {' '.join(pathlib.Path(a).name for a in args)} (as {self._sa['id']})")
t = time.time()
rc, tail = self.local_run([sys.executable, "-m", "erp_platform.compose", verb, *args], env, self.root)
if rc != 0:
raise Refused(f"erp-compose {verb} failed (exit {rc}):\n{_scrub(tail, self.d.secrets)}\n fix: read "
"the lines above; a re-run resumes (migrate and seed skip what is there)")
self.say(f" ✓ {verb}: {time.time() - t:.0f}s")
self.inv["migrated"] = stamp
def _local_run(self, argv, env, cwd):
"""Run it, echoing one line in 20 (a migrate of 8 bricks prints hundreds); returns (exit, the last 25)."""
p = subprocess.Popen(argv, env=env, cwd=cwd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True)
tail = []
for i, line in enumerate(p.stdout):
line = _scrub(line.rstrip(), self.d.secrets)
tail = (tail + [line])[-25:]
if i % 20 == 0:
self.say(f" {line[:160]}")
return p.wait(), "\n".join(tail)
# 4
def files(self):
brick = None
if self.demo_clock:
from erp_platform import compose
comp = getattr(self, "_comp", None) or compose.load(str(self.compose_path))
brick = comp.members[0].name
out = {}
if not (self.root / ".github/workflows/build.yml").exists() and not (self.root / ".dodil/deploy.yaml").exists():
rc, _, _ = self.d.run("git", "ci", "init", "--help")
if rc == 0: # CLI >= 0.1.86: the CLI's scaffold first, then refined below (same build.yml)
rc, o, e = self.d.run("git", "ci", "init", str(self.root), "--app", self.app, "--port", str(PORT),
"--health-path", "/healthz", "--registry-org", self.org.lower())
self.say(" ✓ dodil git ci init" if rc == 0 else f" · dodil git ci init: {(e or o).strip()[:200]}")
for rel, text in (("Dockerfile", dockerfile()), ("app.py", app_py(brick)),
(".github/workflows/build.yml", build_yml(self.org.lower(), self.app)),
(".github/workflows/ci.yml", ci_yml()),
(".dodil/deploy.yaml", deploy_yaml(self.org.lower(), self.app, self.pool, self.bucket,
self.demo_clock)),
(".gitignore", GITIGNORE), (".dockerignore", DOCKERIGNORE)):
out[rel] = write_file(self.root, rel, text, self.say)
return out
# 5
def _git_run(self, args, cwd):
p = subprocess.run(["git", *args], cwd=cwd, capture_output=True, text=True)
return p.returncode, (p.stdout or "") + (p.stderr or "")
def _git(self, *args, ok=True):
rc, out = self.git_run(list(args), self.root)
if rc != 0 and ok:
raise Refused(f"`git {' '.join(args)}` failed: {_scrub(out.strip(), self.d.secrets)[:600]}")
return rc, out
def push(self) -> str:
helper = f"!{self.d.bin} git credential-helper"
if not (self.root / ".git").exists():
self._git("init", "-q", "-b", "main")
self.say(" ✓ git init (branch main)")
rc, url = self._git("remote", "get-url", "dodil", ok=False)
if rc != 0:
self._git("remote", "add", "dodil", self._http_url)
self.say(f" ✓ remote dodil → {self._http_url}")
elif url.strip() != self._http_url:
self._git("remote", "set-url", "dodil", self._http_url)
self.say(f" ✓ remote dodil → {self._http_url} (was {url.strip()})")
self._git("add", "-A")
self._scan_staged()
rc, status = self._git("status", "--porcelain")
if status.strip():
ident = []
if self._git("config", "user.email", ok=False)[0] != 0:
ident = ["-c", "user.name=erp-compose", "-c", "user.email=erp-compose@localhost"]
self._git(*ident, "commit", "-q", "-m", f"{self.app}: the composition, its CI and its deploy "
"(erp-compose deploy)")
self.say(" ✓ commit")
sha = self._git("rev-parse", "HEAD")[1].strip()
v_rc, v_out, v_err = self.d.run("git", "ci", "verify", str(self.root), "--repo", self.repo)
if v_rc != 0:
raise Refused(f"`dodil git ci verify` refused the CI/CD config:\n{(v_out + v_err).strip()[:1500]}")
self.say(" ✓ dodil git ci verify: ok")
rc, remote = self._git("-c", "credential.helper=", "-c", f"credential.helper={helper}", "ls-remote",
"dodil", "refs/heads/main", ok=False)
if rc == 0 and remote.split() and remote.split()[0] == sha:
self.say(f" · push: dodil/main is already {sha[:12]}")
else:
self._git("-c", "credential.helper=", "-c", f"credential.helper={helper}", "push", "-q", "dodil",
"HEAD:refs/heads/main")
self.say(f" ✓ push {sha[:12]} → {self.org}/{self.repo} main")
self.inv["sha"] = sha
return sha
def _scan_staged(self):
rc, files = self._git("ls-files", "-z", "--cached")
for rel in [f for f in files.split("\0") if f]:
p = self.root / rel
try:
text = p.read_text(errors="ignore") if p.is_file() else ""
except OSError:
continue
for s in self.d.secrets:
if s in text:
self._git("reset", "-q")
raise Refused(f"{rel} holds a secret: nothing was committed. fix: remove it from the file "
"(secrets live in the repo's CI secret store and ~/.dodil-secrets)")
def follow(self, sha: str, names) -> None:
"""Wait until every check in `names` passed. `git check list` can lag minutes behind a push: a missing check
is waited for, not concluded on; a failed one prints its log's tail."""
deadline, last, seen = time.time() + self.follow_timeout, {}, False
while True:
rc, out, err = self.d.run("git", "check", "list", self.repo, sha)
checks = {c.get("name"): c for c in (_loads(out).get("checks") or [])} if rc == 0 else {}
if checks:
seen = True
cfg = checks.get(DEPLOY_CONFIG_CHECK)
if cfg and cfg.get("state") in ("failed", "errored"):
raise Refused(f"{DEPLOY_CONFIG_CHECK} {cfg['state']}: {cfg.get('detail', '')[:600]}\n fix: "
f".dodil/deploy.yaml, then `dodil git ci verify` and push")
for n in names:
c = checks.get(n)
state = c.get("state") if c else "not reported yet"
if last.get(n) != state:
self.say(f" {n}: {state}")
last[n] = state
if c and state in TERMINAL and state != "passed":
raise Refused(f"{n} {state} on {sha[:12]}:\n{self._log_tail(c)}\n fix: read the log above, "
"commit the fix and re-run `erp-compose deploy` (it resumes)")
if all((checks.get(n) or {}).get("state") == "passed" for n in names):
return
if time.time() > deadline:
raise Refused(f"timed out after {self.follow_timeout:.0f}s waiting for {', '.join(names)} on "
f"{sha[:12]} ({'no checks reported' if not seen else json.dumps(last)}). fix: `dodil "
f"git check list {self.repo} {sha[:12]} --org-name {self.org}`; re-run to resume")
self.sleep(self.poll)
def _log_tail(self, check) -> str:
cid = check.get("check_id")
rc, out, err = self.d.run("git", "check", "logs", self.repo, cid) if cid else (1, "", "")
text = out if rc == 0 else (check.get("detail") or "")
try:
j = json.loads(text)
text = j.get("logs") or j.get("text") or j.get("detail") or text if isinstance(j, dict) else text
except (ValueError, TypeError):
pass
return "\n".join(f" {l}" for l in str(text).strip().splitlines()[-30:])
# 6
def app_ready(self):
a = self.d.json("ignite", "app", "get", self.app)
data = a.get("data", a)
urls = data.get("public_urls") or []
if urls:
self.fqdn = urls[0].replace("https://", "").rstrip("/")
self._res("app", f"{self.org}/{self.app}", f"dodil ignite app delete {self.app} --yes --org {self.org}",
made_by="the forge's CD (create_if_missing)")
self._res("registry", self.app, f"dodil registry repo delete {self.app} --yes --org {self.org}",
made_by="the build's docker push")
self.say(f" ✓ app {self.app}: {data.get('deployment_state', '?')} v{data.get('active_version', '?')} "
f"at https://{self.fqdn}")
if self.warm:
reserved = (data.get("scaling") or {}).get("reserved_capacity") or 0
if reserved >= 1:
self.say(" · warm: reserved 1")
else:
self.d.json("ignite", "app", "update", self.app, "--reserved", "1")
self.say(" ✓ warm: reserved 1 (Claude.ai allows ~10 s to discover a connector)")
def verify(self):
import httpx
client = self.http or httpx.Client(timeout=300)
base = f"https://{self.fqdn}"
t, deadline = time.time(), time.time() + 900
while True:
try:
r = client.get(base + "/healthz")
if r.status_code == 200:
break
status = r.status_code
except httpx.HTTPError as e: # noqa: PERF203
status = type(e).__name__
if time.time() > deadline:
raise Refused(f"GET {base}/healthz never answered 200 (last: {status}). fix: `dodil ignite app logs "
f"{self.app} --org {self.org}`")
self.sleep(10)
self.say(f" ✓ GET /healthz 200 ({time.time() - t:.1f}s)")
self.inv["timings_s"]["healthz"] = round(time.time() - t, 1)
tok = client.post(self._sc["token_endpoint"], data={"grant_type": "client_credentials",
"client_id": self._sc["client_id"],
"client_secret": self._sc["secret"]})
if tok.status_code != 200:
raise Refused(f"the service client's token was refused ({tok.status_code}): {tok.text[:300]}")
token = tok.json()["access_token"]
t = time.time()
tools = mcp_tools(client, base + "/mcp", token)
self.inv["timings_s"]["first_mcp"] = round(time.time() - t, 1)
self.inv["mcp_tools"] = len(tools)
self.say(f" ✓ MCP initialize + tools/list: {len(tools)} tools ({time.time() - t:.1f}s)")
def mcp_tools(client, url: str, token: str) -> list:
"""initialize → notifications/initialized → tools/list (every page), over streamable HTTP."""
from erp_platform import __version__
h = {"Authorization": f"Bearer {token}", "Accept": "application/json, text/event-stream",
"Content-Type": "application/json"}
def rpc(body, sid=None):
hh = dict(h, **({"Mcp-Session-Id": sid} if sid else {}))
r = client.post(url, json=body, headers=hh)
if r.status_code >= 400:
raise Refused(f"MCP {body.get('method')}: HTTP {r.status_code} {r.text[:300]}")
msg = None
if "event-stream" in r.headers.get("content-type", ""):
for line in r.text.splitlines():
if line.startswith("data:"):
msg = json.loads(line[5:].strip())
elif r.text.strip():
msg = r.json()
if msg and msg.get("error"):
raise Refused(f"MCP {body.get('method')}: {msg['error']}")
return r, (msg or {}).get("result")
r, res = rpc({"jsonrpc": "2.0", "id": 1, "method": "initialize",
"params": {"protocolVersion": "2025-06-18", "capabilities": {},
"clientInfo": {"name": "erp-compose", "version": __version__}}})
sid = r.headers.get("mcp-session-id")
rpc({"jsonrpc": "2.0", "method": "notifications/initialized"}, sid)
tools, cursor, i = [], None, 2
while True:
_, res = rpc({"jsonrpc": "2.0", "id": i, "method": "tools/list", "params": {"cursor": cursor} if cursor
else {}}, sid)
tools += (res or {}).get("tools") or []
cursor = (res or {}).get("nextCursor")
i += 1
if not cursor:
return tools
# ── teardown ───────────────────────────────────────────────────────────────────────────────────────────────────────
class Teardown:
def __init__(self, compose_path, org: str, *, keep_repo=False, dodil: Optional[Dodil] = None, secrets_dir=None,
say=_say, sleep=time.sleep, settle: float = 300):
self.sleep, self.settle = sleep, settle
self.root = pathlib.Path(compose_path).resolve().parent
self.inv = load_inventory(self.root)
if not self.inv:
raise Refused(f"no {INVENTORY} in {self.root}: nothing recorded to tear down")
if self.inv.get("org") != org:
raise Refused(f"{INVENTORY} is for org {self.inv.get('org')!r}, not {org!r}")
self.org, self.keep_repo, self.say = org, keep_repo, say
self.app, self.bucket, self.pool, self.repo = (self.inv[k] for k in ("app", "bucket", "pool", "repo"))
self.d = dodil or Dodil(org)
self.secrets = Secrets(secrets_dir or pathlib.Path.home() / ".dodil-secrets" / self.app)
def _gone(self, what, rc_out_err):
rc, out, err = rc_out_err
text = (out + err).lower()
if rc == 0:
self.say(f" ✓ {what}: deleted")
elif "not found" in text or "404" in text or "does not exist" in text:
self.say(f" · {what}: already gone")
else:
raise Refused(f"deleting {what} failed: {(err or out).strip()[:400]}")
def run(self) -> int:
d = self.d
try:
self._gone(f"ignite app {self.app}", d.run("ignite", "app", "delete", self.app, "--yes"))
for s in _items(d.get("ignite", "secret", "list", "--app", self.app) or {}, "secrets", "data", "items"):
n = s.get("name") if isinstance(s, dict) else s
if n:
self._gone(f"ignite secret {n} (app {self.app})", d.run("ignite", "secret", "delete", n, "--app",
self.app))
for name in sorted(_registry_repos(d, self.app), key=len, reverse=True):
self._gone(f"registry repo {name}", d.run("registry", "repo", "delete", name, "--yes"))
self._gone(f"service client {self.inv['resources'].get('service_client', {}).get('id', '?')}",
d.run("appid", "service-client", "delete", self.pool, f"{self.app}-mcp-check", "--yes"))
self._gone(f"appid pool {self.pool}", d.run("appid", "pool", "delete", self.pool, "--yes"))
sa = (self.inv["resources"].get("service_account") or {}).get("id")
if sa:
self._gone(f"service account {sa}", d.run("auth", "service-account", "delete", sa, "--yes"))
self._gone(f"bucket {self.bucket}", d.run("data", "bucket", "delete", self.bucket))
for s in _sas(d, f"k3-tables-sa-{self.bucket}-"):
self._gone(f"the bucket's tables SA {s['id']}", d.run("auth", "service-account", "delete", s["uuid"],
"--yes"))
if self.keep_repo:
self.say(f" · repo {self.org}/{self.repo}: kept (--keep-repo)")
else:
self._gone(f"repo {self.org}/{self.repo}", d.run("git", "repo", "delete", self.repo, "--yes"))
self.secrets.remove()
self.say(f" ✓ local secrets {self.secrets.dir}: removed")
left, until = self.remaining(), time.time() + self.settle
while self._left(left) and time.time() < until:
# a bucket delete is asynchronous: it lists as BUCKET_STATUS_DELETING for ~40 s (live, 2026-10-25)
self.sleep(10)
left = self.remaining()
except Refused as e:
self.say(f"\nREFUSED: {e}")
return 1
self.say("\nlisted after teardown:")
for kind, names in left.items():
self.say(f" {kind}: {', '.join(names) if names else 'none'}")
if self._left(left):
self.say("REFUSED: resources remain (above)")
return 1
self.inv["torn_down_at"] = _now()
save_inventory(self.root, self.inv)
return 0
def _left(self, left: dict) -> bool:
return any(v for k, v in left.items() if not (k == "git repos" and self.keep_repo))
def remaining(self) -> dict:
d, a = self.d, self.app
buckets = [b.get("name") for b in _items(d.json("data", "bucket", "list"), "buckets")]
apps = [x.get("name") for x in _items(d.json("ignite", "app", "list"), "data", "apps", "items")]
pools = [p.get("name") for p in _items(d.json("appid", "pool", "list"), "pools", "items")]
repos = [r.get("name") for r in _items(d.json("git", "repo", "list"), "repos", "repositories", "items")]
return {
"buckets": [b for b in buckets if b == self.bucket],
"ignite apps": [x for x in apps if x == a],
"appid pools": [p for p in pools if p == self.pool],
"service accounts": [s["id"] for s in _sas(d, a)],
"registry repos": sorted(_registry_repos(d, a)),
"git repos": [r for r in repos if r == self.repo],
}
def _items(doc, *keys) -> list:
if isinstance(doc, list):
return doc
if not isinstance(doc, dict):
return []
for k in keys:
v = doc.get(k)
if isinstance(v, list):
return v
if isinstance(v, dict):
inner = _items(v, *keys)
if inner:
return inner
return []
def _sas(d: Dodil, needle: str) -> list:
out = d.json("auth", "service-account", "list", "--filter", needle) or {}
return [{"id": s.get("serviceAccountId", ""), "uuid": s.get("serviceAccountUuid") or s.get("uuid", "")}
for s in _items(out, "serviceAccounts", "service_accounts") if needle in s.get("serviceAccountId", "")]
def _registry_repos(d: Dodil, app: str) -> list:
out = d.json("registry", "repo", "list", "--page-size", "100") or {}
names = [r.get("name", "") for r in _items(out, "items", "repositories")]
return [n for n in names if n == app or n.startswith(app + "/")]
def _fingerprint(root: pathlib.Path) -> str:
import hashlib
h = hashlib.sha256()
for rel in ["compose.yaml", "requirements.txt", *seed_files(root)]:
p = root / rel
h.update(rel.encode() + b"\0" + (p.read_bytes() if p.is_file() else b""))
return h.hexdigest()[:16]
def _scrub(text: str, secrets) -> str:
for s in secrets:
text = text.replace(s, "***")
return text
def _now() -> str:
return _dt.datetime.now(_dt.timezone.utc).isoformat(timespec="seconds")
def load_inventory(root) -> Optional[dict]:
p = pathlib.Path(root) / INVENTORY
return json.loads(p.read_text()) if p.is_file() else None
def save_inventory(root, inv: dict) -> None:
p = pathlib.Path(root) / INVENTORY
p.parent.mkdir(parents=True, exist_ok=True)
inv = dict(inv, note="ids and delete commands only: no secret is ever written here "
"(they are in ~/.dodil-secrets/<app>/). `erp-compose teardown` deletes in reverse.")
p.write_text(json.dumps(inv, indent=1) + "\n")
# ── CLI ────────────────────────────────────────────────────────────────────────────────────────────────────────────
def main(argv=None) -> int:
argv = list(sys.argv[1:] if argv is None else argv)
verb, rest = (argv[0], argv[1:]) if argv else ("", [])
if verb == "deploy":
ap = argparse.ArgumentParser(prog="erp-compose deploy", description=__doc__.split("\n\n")[0])
ap.add_argument("compose")
ap.add_argument("--org", required=True, help="the org every call acts in (never switched)")
ap.add_argument("--name", required=True, help="the app; also the bucket and the pool")
ap.add_argument("--repo", help="the Dodil Git repo (default: --name)")
ap.add_argument("--warm", action="store_true", help="keep one instance warm (reserved 1)")
ap.add_argument("--demo-clock", action="store_true",
help="DEMO: mount the business clock's tools and start it tomorrow (ERP_DEMO_CLOCK_OFFSET=+1d)")
a = ap.parse_args(rest)
return Deploy(a.compose, a.org, a.name, repo=a.repo, warm=a.warm, demo_clock=a.demo_clock).run()
if verb == "teardown":
ap = argparse.ArgumentParser(prog="erp-compose teardown")
ap.add_argument("compose")
ap.add_argument("--org", required=True)
ap.add_argument("--keep-repo", action="store_true")
a = ap.parse_args(rest)
try:
return Teardown(a.compose, a.org, keep_repo=a.keep_repo).run()
except Refused as e:
print(f"REFUSED: {e}", file=sys.stderr)
return 1
print("usage: erp-compose deploy|teardown compose.yaml --org ORG …", file=sys.stderr)
return 2