"""local — the DEV tier of `erp-compose migrate|serve`: the same composition on SQLite files on a laptop.
0.9.9. Without it, migrate and serve only reach a DataK3 bucket (BUCKET + the pg wire + a service-account token), so
a builder could not run what they composed on their own machine; the 2026-10-25 small-ERP dry run did it by hand
with erp_platform.scenarios.inprocess.stage. This is that, folded in:
erp-compose migrate --local [DIR] compose.yaml create every brick's tables (+ the platform's link tables)
erp-compose serve --local [DIR] compose.yaml serve compose.app on 127.0.0.1, demo clock mounted
DIR defaults to `.erp-local/` beside compose.yaml: one SQLite file per brick plus `_platform.db`. The tables come
from each brick's own MetaData (`create_all`, then the repository's `seed()`) — the path the in-process scenario
runner stands on — NOT from schema.sql: a bucket is migrated from schema.sql, a laptop is not, so a schema.sql
drift is caught by `erp-compose migrate` on a bucket, never here.
It is explicit and never the default: `--local` is the only way in, and it is REFUSED when BUCKET is set (a shell
that names a bucket meant the bucket) and under DODIL_ENV=prod|live. The sign-off gate WARNS here instead of
refusing — a laptop is where a design is tried before the customer signs it; migrating a bucket still refuses.
The demo clock is mounted on the first brick (`GET|POST /<brick>/demo-clock`, permission `<brick>:clock:set`).
Bricks take a setting, a price list or a map in force from a date AFTER today (a correction is a new dated row,
never a rewrite of the past), so a fresh build cannot sell on the day it was set up; `--clock +1d` (or an
instant) starts the business clock there, and the clock route moves it while serving. Production date semantics
are unchanged: the clock is refused under DODIL_ENV=prod|live, by erp_platform.work itself.
0.9.10 — THE DEV OWNER (the 2026-10-25 timed lab run exported DEV_ALLOW_ANON, DEV_USER_* and a hand-collected
DEV_USER_PERMISSIONS='*:*:*,tax:code:seed,ar:invoice:post,gl:journal:post,…', restarting the server each time a
machine permission refused a call): `serve --local` makes a header-less request the dev owner (`owner`,
owner@localhost, role manager), holding EVERY permission the composed bricks declare — each use case's and each
Brick(machine_permissions=…), machine ones included (a wildcard never reaches those) — and prints what it granted.
An env var that is SET wins (DEV_USER_PERMISSIONS narrows the owner to exactly those, DEV_ALLOW_ANON=0 turns the
stub off). The local tier only: a bucket's serve is unchanged (the gateway's pool JWT says who may do what), and
the platform's own erp:seam:act_for is never granted (no person acts for another).
0.9.11 — THE SCHEDULED WORK (the 2026-10-25 lab run: an invoice never reached the ledger until the agent called
post_to_ledger by hand, because on the local tier nothing ran ar's post_pending). A deployment runs the bricks'
passes from the composition's clock worker (module.yaml `jobs:`); `serve --local` now runs them itself, every
`--ticks-every` seconds (default 5; `--no-ticks` turns it off), through the routes as the dev owner (sub
erp-compose.clock), and logs one line per tick that did something — an unchanged answer is not logged again.
A brick names its passes with `Brick(..., ticks=("post_pending", …))`; a brick that names none gets the commands
that look like a pass: machine permission, idempotent, no required input, named *tick* / *pending* / *due* /
*retry* / *finish* (ar's raise_due, finish_schedules, post_pending; gl's post_pending_reversal; payments'
collections_tick…). Never a bucket: a deployment's clock worker is its own app.
"""
import json
import logging
import os
import pathlib
import re
import threading
from contextlib import ExitStack
from datetime import datetime, timezone
from typing import Optional
from erp_platform import work
DEFAULT_DIR = ".erp-local"
MARKER = "erp-local.json"
HOST = "127.0.0.1"
class LocalRefused(RuntimeError):
pass
def banner(say, verb: str, data: pathlib.Path) -> None:
say(f"LOCAL DEV TIER — erp-compose {verb} --local: SQLite files in {data}/, NOT a DataK3 bucket.")
say(" Nothing here is a deployment. Drop --local (and set BUCKET) to migrate or serve the bucket.")
def guard() -> None:
if os.environ.get("BUCKET", "").strip():
raise LocalRefused(f"--local refused: BUCKET={os.environ['BUCKET']} is set, and a shell that names a bucket "
"meant the bucket. Unset BUCKET to use the local tier, or drop --local to use the bucket.")
if work._live_env():
raise LocalRefused(f"--local refused: DODIL_ENV={work._live_env()} — the local tier is a dev tier")
def data_dir(root: pathlib.Path, given: Optional[str]) -> pathlib.Path:
p = pathlib.Path(given) if given else root / DEFAULT_DIR
return (p if p.is_absolute() else pathlib.Path.cwd() / p).resolve()
def _stage(comp, st: ExitStack, data: pathlib.Path, *, create: bool):
from erp_platform.scenarios.core import Catalogue
from erp_platform.scenarios.inprocess import _sqlite, stage
engines = {}
def engine_for(name: str):
if name not in engines:
engines[name] = _sqlite(data / f"{name}.db")
return engines[name]
st.callback(lambda: [e.dispose() for e in engines.values()])
if not create: # 0.9.10: a folder migrated by 0.9.9 has no idempotency table yet
from erp_platform import idempotency
idempotency.md.create_all(engine_for("_platform"))
return stage(Catalogue(comp.name, compose=comp, roles=None), st, engine_for, create=create)
def migrate(comp, data: pathlib.Path, say=print) -> int:
guard()
data.mkdir(parents=True, exist_ok=True)
ignore = data / ".gitignore" # dev data never lands in the customer's repo; their .gitignore is theirs
if not ignore.exists():
ignore.write_text("# erp-compose migrate --local: SQLite dev data, never committed\n*\n")
with ExitStack() as st:
_stage(comp, st, data, create=True)
versions = {m.name: m.brick.version for m in comp.members}
(data / MARKER).write_text(json.dumps({
"composition": comp.name, "bricks": versions,
"migrated_at": datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z"),
"tier": "local (dev)"}, indent=1) + "\n")
for m in comp.members:
say(f" {m.name} {m.brick.version}: tables created in {data / (m.name + '.db')}")
say(f"local tier migrated: {len(versions)} brick(s) + the platform's link tables in {data}/")
return 0
def clock_brick(comp):
"""The brick the local tier mounts the demo clock on: one that already mounts it, else the first member."""
for m in comp.members:
if any(u.name == f"{m.brick.name}_demo_clock" for u in m.brick.use_cases):
return m.brick, False
return comp.members[0].brick, True
def start_clock(text: Optional[str]) -> Optional[datetime]:
"""--clock: an offset ("+1d") or an instant ("2026-11-20T09:00Z"); the clock ticks on from there."""
if not text:
return None
from erp_platform import democlock
t = text.strip()
if t[:1] in "+-":
return work.set_clock(offset=democlock._offset(t))
return work.set_clock(at=work.parse_instant(t))
def owner_permissions(comp) -> tuple:
"""(person, machine): every concrete permission the composed bricks declare, split by kind. Machine ones need an
exact grant (erp_platform.auth), so the owner holds them by name; the platform's act_for is not a brick's."""
from erp_platform import auth
declared = {u.permission for m in comp.members for u in m.brick.use_cases if u.permission}
declared |= {p for m in comp.members for p in getattr(m.brick, "machine_permissions", ())}
declared = {p for p in declared if auth.is_concrete(p)} - set(auth.MACHINE_PERMISSIONS)
machine = sorted(p for p in declared if auth.is_machine(p))
return sorted(declared - set(machine)), machine
def owner_identity(comp) -> dict:
"""The dev owner as a gateway would mint it: the identity `erp-compose seed` acts as, and serve's header-less one."""
person, machine = owner_permissions(comp)
return {"sub": DEV_OWNER["sub"], "email": DEV_OWNER["email"], "app_roles": list(DEV_OWNER["roles"]),
"permissions": person + machine}
DEV_OWNER = {"sub": "owner", "email": "owner@localhost", "roles": ("manager",)}
def dev_owner(comp, say=print):
"""Make header-less requests the dev owner, holding every declared permission; returns the undo. A set env wins."""
from erp_platform import auth
before = {k: getattr(auth, k) for k in ("DEV_ALLOW_ANON", "DEV_USER_SUB", "DEV_USER_EMAIL", "DEV_USER_ROLES",
"DEV_USER_PERMISSIONS")}
env = os.environ
if "DEV_ALLOW_ANON" not in env:
auth.DEV_ALLOW_ANON = True
if not auth.DEV_ALLOW_ANON:
say(" identity: DEV_ALLOW_ANON is off — every request must carry X-Dodil-User (no dev owner)")
return lambda: None
if "DEV_USER_SUB" not in env:
auth.DEV_USER_SUB = DEV_OWNER["sub"]
if "DEV_USER_EMAIL" not in env:
auth.DEV_USER_EMAIL = DEV_OWNER["email"]
if "DEV_USER_ROLES" not in env:
auth.DEV_USER_ROLES = list(DEV_OWNER["roles"])
person, machine = owner_permissions(comp)
if "DEV_USER_PERMISSIONS" in env:
say(f" identity: header-less requests act as {auth.DEV_USER_SUB} with DEV_USER_PERMISSIONS as set "
f"({len(auth.DEV_USER_PERMISSIONS)} grant(s)) — unset it to hold every permission the bricks declare")
else:
auth.DEV_USER_PERMISSIONS = person + machine
say(f" identity: header-less requests act as the dev owner {auth.DEV_USER_SUB} ({auth.DEV_USER_EMAIL}, roles "
f"{','.join(auth.DEV_USER_ROLES) or '-'}), granted all {len(person) + len(machine)} permissions the "
f"{len(comp.members)} brick(s) declare:")
by = {}
for p in person:
by.setdefault(p.split(":", 1)[0], []).append(p)
say(" person: " + ", ".join(f"{ns} {len(ps)}" for ns, ps in sorted(by.items())))
say(" machine: " + (", ".join(machine) or "(none)") + " (exact grants; a wildcard never reaches them)")
def undo():
for k, v in before.items():
setattr(auth, k, v)
return undo
def mark_seeded(data: pathlib.Path, calls: str) -> None:
"""0.9.10: `seed --local` notes the business day it ran on; serve --local reads it (seeded_clock)."""
marker = data / MARKER
doc = json.loads(marker.read_text()) if marker.exists() else {}
doc["seeded"] = {"business_day": work.now().date().isoformat(), "calls": calls}
marker.write_text(json.dumps(doc, indent=1) + "\n")
def seeded_clock(data: pathlib.Path) -> Optional[str]:
""""+1d" when this folder was seeded today (business day): rows a starter dated $tomorrow are then in force."""
try:
day = json.loads((data / MARKER).read_text()).get("seeded", {}).get("business_day")
except (OSError, ValueError, AttributeError):
return None
return "+1d" if day and day >= work.now().date().isoformat() else None
def prepare_serve(comp, data: pathlib.Path, *, clock: Optional[str] = None, say=print):
"""Everything serve --local does before uvicorn: returns (app, stack). The caller closes the stack."""
guard()
if not (data / MARKER).exists():
raise LocalRefused(f"{data}/ is not migrated: run `erp-compose migrate --local` first "
"(the same order as a bucket: migrate, then serve)")
from erp_platform import democlock
brick, mount = clock_brick(comp)
if mount:
democlock.mount(brick)
st = ExitStack()
try:
st.callback(dev_owner(comp, say)) # 0.9.10: before the apps are built (the MCP door reads it)
staged = _stage(comp, st, data, create=False)
if clock is None and seeded_clock(data):
clock = seeded_clock(data)
say(" seeded today: rows dated from tomorrow (a brick rule) are in force from the start — "
"the business clock starts at +1d (--clock overrides)")
try:
at = start_clock(clock)
except Exception as e: # Refused (a bad offset) or ValueError (a bad instant)
raise LocalRefused(f"--clock {clock!r}: {getattr(e, 'detail', None) or e}") from None
except BaseException:
st.close()
raise
perm = next((u.permission for u in brick.use_cases if u.name == f"set_{brick.name}_demo_clock"), None)
say(f" demo clock: GET|POST /{brick.name}/demo-clock (permission {perm}) — "
+ (f"business clock starts at {at.isoformat()}" if at else "business clock = real time"))
if not at:
say(" Settings, price lists and maps take effect from a date AFTER today (a brick rule, not a local one): "
"to sell on day one, set up today and serve with --clock +1d, or POST {\"advance\": \"+1d\"} to the clock.")
say(f" identity: an X-Dodil-User header is trusted as-is (no gateway here) — so it listens on {HOST} only")
return staged.app, st
# ── 0.9.11: the scheduled work ─────────────────────────────────────────────────────────────
TICK_EVERY = 5.0
PASS_NAME = re.compile(r"(^|_)(tick|pending|due|retry|finish)(_|$)")
CLOCK_SUB = "erp-compose.clock"
def tick_use_cases(comp) -> list:
"""[(member, use case)]: the passes a clock worker runs — what each brick declares as `ticks`, else the commands
that look like one (a machine permission, idempotent, every input defaulted, a pass's name)."""
import inspect
from erp_platform import auth
out = []
for m in comp.members:
named = getattr(m.brick, "ticks", None)
for u in m.brick.use_cases:
if named is not None:
if u.name in named:
out.append((m, u))
continue
if (u.kind == "command" and u.idempotent and u.permission and auth.is_machine(u.permission)
and not u.path_params and PASS_NAME.search(u.name)
and all(d is not inspect.Parameter.empty for _n, _a, d in u.inputs)):
out.append((m, u))
return out
def did(body) -> str:
"""What a pass's answer says it did, in a few words ("" for nothing): counts > 0 and non-empty lists."""
if not isinstance(body, dict):
return ""
parts = []
for k, v in body.items():
if isinstance(v, bool) or k in ("as_of",):
continue
if isinstance(v, (int, float)) and v:
parts.append(f"{k}={v}")
elif isinstance(v, list):
n = sum(1 for x in v if not (isinstance(x, dict) and x.get("already"))) # "already posted" is no work
if n:
parts.append(f"{k}={n}")
elif isinstance(v, dict) and v:
parts.append(f"{k}={len(v)}")
return " ".join(parts)
THREAD = "erp-local-ticks"
QUIET = ("erp_platform.request", "erp_platform.work") # a request line and a demo-clock line per pass, every 5 s
class _Quiet:
"""Drops the ticker thread's own request and demo-clock lines below ERROR: the tick line says what happened."""
def filter(self, record) -> bool:
return record.threadName != THREAD or record.levelno >= 40
class Ticker:
"""A daemon thread running every pass in turn, then sleeping `every` seconds, until stop()."""
def __init__(self, comp, app, every: float = TICK_EVERY, say=print):
self.comp, self.app, self.every, self.say = comp, app, float(every), say
self.passes = tick_use_cases(comp)
self.me = {**owner_identity(comp), "sub": CLOCK_SUB, "email": "clock@localhost"}
self.last: dict = {}
self._stop = threading.Event()
self._thread: Optional[threading.Thread] = None
def once(self) -> list:
"""One round: every pass once. Returns the lines it logged."""
from erp_platform.process_export import reason, send
lines = []
for m, u in self.passes:
if self._stop.is_set():
break
try:
r = send(self.app, {u.name: (m, u)}, self.me, u.name, {})
body = r.json() if r.content else None
what = did(body) if r.status_code < 300 else f"refused {r.status_code}: {reason(r)[:200]}"
except Exception as e: # noqa: BLE001 — a pass that crashes is logged, the loop goes on
what = f"failed {type(e).__name__}: {str(e)[:200]}"
key = f"{m.name}.{u.name}"
if what and what != self.last.get(key):
line = f" tick {work.now().strftime('%H:%M:%S')} {m.name} {u.name}: {what}"
self.say(line)
lines.append(line)
self.last[key] = what
return lines
def _run(self):
while not self._stop.wait(self.every):
self.once()
def start(self) -> "Ticker":
if self.passes:
self._quiet = _Quiet()
for name in QUIET:
logging.getLogger(name).addFilter(self._quiet)
self._thread = threading.Thread(target=self._run, daemon=True, name=THREAD)
self._thread.start()
return self
def stop(self) -> None:
self._stop.set()
if self._thread is not None:
self._thread.join(timeout=self.every + 30)
for name in QUIET:
logging.getLogger(name).removeFilter(self._quiet)
def start_ticks(comp, app, st: ExitStack, *, every: Optional[float], say=print) -> Optional[Ticker]:
"""serve --local: start the ticker (every=None: --no-ticks) and close it with the stack; says what runs."""
if every is None:
say(" scheduled work: off (--no-ticks) — call the passes yourself (e.g. ar post_pending)")
return None
t = Ticker(comp, app, every, say)
if not t.passes:
say(" scheduled work: none — no brick declares a pass")
return None
say(f" scheduled work: every {every:g}s as {CLOCK_SUB} — " + ", ".join(f"{m.name} {u.name}" for m, u in t.passes)
+ " (--ticks-every N, --no-ticks; a line per pass that did something)")
st.callback(t.stop)
return t.start()