"""The in-process runner: the WHOLE composition in this process, in seconds.
compose.app(...) builds the staff app exactly as `erp-compose serve` does — every brick's routes under
/<brick>, ONE /mcp, seams wired in process — and compose.public_app(...) the public app. Each brick gets its
OWN SQLite file (the tier-1 bucket is shared, but a brick never reads another's tables, so separate files
also prove that); the platform's link tables get one more. Requests go over real HTTP semantics through an
ASGI client, MCP steps through a real MCP ClientSession, on one event loop that runs the app's lifespan.
What only this mode can do: move work.now() directly, and inject failures (s.fault). Each scenario's faults
and clock are undone when it ends, so no scenario leans on another.
ONE STAGED COMPOSITION PER PROCESS AT A TIME: the platform keeps the in-process seam registry, the capability
providers, public.configure and PUBLIC_LINK_BASE_URL process-wide, as a deployed app does. A runner restores
them when it stops; start the next one after.
"""
import importlib
import os
import pathlib
import tempfile
import threading
import time
from contextlib import ExitStack, contextmanager
from typing import Any, Callable, Optional
from urllib.parse import urlparse
import anyio
import anyio.from_thread
import httpx
from erp_platform import capabilities, compose, idempotency, public, retry, seam, work
from erp_platform.scenarios.core import Response, Run, Runner, Scenario, http_response, mcp_response
PUBLIC_BASE = "https://public.scenario.test"
# ── storage: each brick on its own SQLite ──────────────────────────────────────────────────
def _sqlite(path: pathlib.Path):
from sqlalchemy import create_engine, event
engine = create_engine(f"sqlite:///{path}", connect_args={"check_same_thread": False, "timeout": 30})
@event.listens_for(engine, "connect")
def _wal(conn, _rec):
conn.execute("PRAGMA journal_mode=WAL")
return engine
def root_repository(name: str, tables):
"""The repository a brick's use cases get, by the convention below: the class named exactly `Repository` in
`<package>.infrastructure.tables`, else its ONE `<Name>Repository`. Also what a bucket's migrate seeds with."""
repos = [v for k, v in vars(tables).items()
if isinstance(v, type) and k.endswith("Repository") and v.__module__ == tables.__name__]
root = [r for r in repos if r.__name__ == "Repository"]
if root:
repos = root
if len(repos) != 1:
raise RuntimeError(f"{name}: {tables.__name__} must define a `Repository` class or exactly one "
f"*Repository class (found {[r.__name__ for r in repos]}) — or pass storage= for "
"this brick")
return repos[0]
def wire_storage(member, engine, *, create: bool = True) -> Callable[[], None]:
"""Point a brick's storage at `engine`, by the bricks' convention, and return how to undo it.
The convention every brick in this repo family follows (templates/brick-example): the use cases open a unit of
work with `<package>.application.use_cases.repository()`, and `<package>.infrastructure.tables` holds the
MetaData `md` and the repository the use cases get: the class named exactly `Repository` (0.9.1: what
`erp-compose scaffold` writes — one facade over one `<Root>Repository` per aggregate), else its ONE
`<Name>Repository` class. A brick that differs passes its own wire function in `Catalogue(storage={name: fn})`.
When the tables are created and the repository has `seed()`, it is called once (the settings rows). A
capability provider's claim tables go on the same engine."""
pkg = member.package
undo = []
try:
uc = importlib.import_module(f"{pkg}.application.use_cases")
tables = importlib.import_module(f"{pkg}.infrastructure.tables")
except ImportError:
uc = tables = None
if uc is not None and hasattr(uc, "repository") and hasattr(tables, "md"):
factory = root_repository(member.name, tables)
if create:
tables.md.create_all(engine)
if callable(getattr(factory, "seed", None)):
with engine.begin() as conn:
factory(conn).seed()
before = uc.repository
@contextmanager
def repository():
with engine.begin() as conn:
yield factory(conn)
uc.repository = repository
undo.append(lambda: setattr(uc, "repository", before))
elif not member.brick.provides:
raise RuntimeError(f"{member.name}: no {pkg}.application.use_cases.repository / infrastructure.tables.md "
"to wire — pass Catalogue(storage={name: wire(engine) -> undo}) for this brick")
if member.brick.provides:
if create:
capabilities.tables(member.brick.name).metadata.create_all(engine)
before_conn = member.brick.capability_connection
member.brick.capability_connection = engine.begin
undo.append(lambda: setattr(member.brick, "capability_connection", before_conn))
return lambda: [u() for u in reversed(undo)]
# ── the clock ──────────────────────────────────────────────────────────────────────────────
class InProcessClock:
def unavailable(self) -> Optional[str]:
return None
def set(self, at):
return work.set_clock(at=at, freeze=True)
def now(self):
return work.now()
def reset(self) -> None:
work.reset_clock()
# ── failure injection ──────────────────────────────────────────────────────────────────────
class InjectedDatabaseError(Exception):
"""What a driver raises for a transient failure: carries the SQLSTATE the retry loop classifies by."""
def __init__(self, code: str):
super().__init__(f"injected database failure (SQLSTATE {code})")
self.sqlstate = code
class InjectedCrash(RuntimeError):
"""The process died here — between two steps of a use case."""
class _Delayed:
def __init__(self, app, seconds: float):
self.app, self.seconds = app, seconds
async def __call__(self, scope, receive, send):
await anyio.sleep(self.seconds)
await self.app(scope, receive, send)
class _Failing:
def __init__(self, status: int):
self.status = status
async def __call__(self, scope, receive, send):
if scope["type"] != "http":
return
body = b'{"detail": "injected failure"}'
await send({"type": "http.response.start", "status": self.status,
"headers": [(b"content-type", b"application/json")]})
await send({"type": "http.response.body", "body": body})
class Faults:
"""Failure injection for one scenario; everything is undone when the scenario ends.
s.fault.seam_down("stock") the neighbour is unreachable (connection refused)
s.fault.seam_slow("stock", 2.0, timeout=0.5) it answers late — past the seam's timeout
s.fault.seam_failing("stock", 503) it answers an error
s.fault.capability("mail.send", "absent") no provider: the consumer makes a manual task
s.fault.capability("mail.send", "failing") the provider fails, retryable (Transient each try)
s.fault.capability("mail.send", "failed") …fails for good (Failed)
s.fault.capability("mail.send", "connection_required")
s.fault.db_error("40001", use_case="place_order", times=1) before the use case runs
s.fault.db_error("08006", use_case="place_order", after=True) after it committed: the ack is lost
s.fault.crash_after("pkg.application.use_cases:_store", times=1) a crash between two steps
"""
def __init__(self, runner: "InProcessRunner", stack: ExitStack, run: Run):
self.runner, self.stack, self.run = runner, stack, run
@contextmanager
def scope(self):
"""Faults that last only for a `with` block — the neighbour comes back when it ends:
with s.fault.scope() as f:
f.seam_down("stock")
... # stock is down here
... # and up again here
"""
with ExitStack() as inner:
yield Faults(self.runner, inner, self.run)
self.run._step("FAULTS lifted")
def _seams(self, name: str) -> list:
out = []
for m in self.runner.comp.members:
out += [s for s in m._instances(seam.Seam) if s.name == name]
return out
def _env(self, key: str, value: Optional[str]) -> None:
before = os.environ.get(key)
if value is None:
os.environ.pop(key, None)
else:
os.environ[key] = value
self.stack.callback(lambda: os.environ.__setitem__(key, before) if before is not None else os.environ.pop(key, None))
def _local(self, name: str, app: Any) -> None:
before = seam._LOCAL.get(name)
if app is None:
seam.unregister_local(name)
else:
seam.register_local(name, app)
self.stack.callback(lambda: seam.register_local(name, before) if before is not None else seam.unregister_local(name))
def seam_down(self, name: str) -> None:
"""Every seam to `name` (a Seam, or a capability provider reached by its base URL) now meets a closed
port: an unreachable neighbour, not an absent one."""
self.run._step(f"FAULT {name} is down")
dead = "http://127.0.0.1:9"
self._local(name, None)
self._env(capabilities.base_env(name), dead)
for s in self._seams(name):
before = s.base
s.base = dead
self.stack.callback(lambda s=s, b=before: setattr(s, "base", b))
def seam_slow(self, name: str, seconds: float, timeout: Optional[float] = None) -> None:
self.run._step(f"FAULT {name} answers after {seconds:g}s" + (f" (seam timeout {timeout:g}s)" if timeout else ""))
app = seam._LOCAL.get(name)
if app is None:
raise ValueError(f"{name} is not in this composition")
self._local(name, _Delayed(app, seconds))
if timeout is not None:
for s in self._seams(name):
before = s.timeout
s.timeout = timeout
self.stack.callback(lambda s=s, b=before: setattr(s, "timeout", b))
def seam_failing(self, name: str, status: int = 503) -> None:
self.run._step(f"FAULT {name} answers {status}")
self._local(name, _Failing(status))
def capability(self, name: str, mode: str, message: str = "") -> None:
modes = ("absent", "failing", "failed", "connection_required", "down")
if mode not in modes:
raise ValueError(f"mode is one of {modes}")
self.run._step(f"FAULT capability {name}: {mode}")
before = dict(capabilities._PROVIDERS)
self.stack.callback(lambda: (capabilities._PROVIDERS.clear(), capabilities._PROVIDERS.update(before)))
provider = capabilities._PROVIDERS.get(name)
if mode == "absent":
capabilities.register_provider(name, None)
return
if provider is None:
raise ValueError(f"no provider of {name} in this composition to make {mode}")
if mode == "down":
return self.seam_down(provider)
provided = [p for m in self.runner.comp.members if m.name == provider
for p in m._instances(capabilities.Provided) if p.capability.name == name]
if not provided:
raise ValueError(f"{provider} has no Provided object for {name}")
exc = {"failing": capabilities.Transient, "failed": capabilities.Failed,
"connection_required": capabilities.ConnectionRequired}[mode]
text = message or {"failing": "the vendor is not answering",
"failed": "the vendor refused the request",
"connection_required": f"no connection to the tool behind {provider}"}[mode]
def broken(call):
raise exc(text)
for p in provided:
before_h = p.handler
p.handler = broken
self.stack.callback(lambda p=p, h=before_h: setattr(p, "handler", h))
sleep_before = capabilities._sleep
capabilities._sleep = lambda _s: None # the backoff between vendor tries, not waited out here
self.stack.callback(lambda: setattr(capabilities, "_sleep", sleep_before))
def db_error(self, sqlstate: str, *, use_case: Optional[str] = None, times: int = 1, after: bool = False) -> dict:
"""The next `times` runs of `use_case` (any, if None) fail with this SQLSTATE — before the use case
runs, or `after` it ran (its commit landed and the acknowledgement was lost: 08006). Injected where the
doors run a use case (bricks._execute → retry.run), so the retry loop decides as it would for real.
Returns a dict whose "injected" counts the failures that were raised."""
self.run._step(f"FAULT {sqlstate} ×{times} on {use_case or 'any use case'}" + (" after commit" if after else ""))
from erp_platform import requestlog
real = retry.run
state = {"left": times, "injected": 0, "runs": 0}
lock = threading.Lock()
def run(fn, **kw):
current = requestlog._CURRENT.get()
mine = use_case is None or (current is not None and current.use_case == use_case)
if not mine:
return real(fn, **kw)
def attempt():
with lock:
state["runs"] += 1
fire = state["left"] > 0
if fire:
state["left"] -= 1
state["injected"] += 1
if fire and not after:
raise InjectedDatabaseError(sqlstate)
out = fn()
if fire:
raise InjectedDatabaseError(sqlstate)
return out
return real(attempt, **kw)
retry.run = run
backoff = retry.BACKOFF
retry.BACKOFF = (0.01, 0.01, 0.01)
self.stack.callback(lambda: (setattr(retry, "run", real), setattr(retry, "BACKOFF", backoff)))
return state
def crash_after(self, target: str, times: int = 1) -> None:
"""`module:function` runs, then the process "dies" — the next step of the use case never happens.
The request answers 500, as a crashed replica's would."""
self.run._step(f"FAULT crash after {target} ×{times}")
module_name, _, attr = target.partition(":")
mod = importlib.import_module(module_name)
real = getattr(mod, attr)
state = {"left": times}
lock = threading.Lock()
def wrapped(*a, **kw):
out = real(*a, **kw)
with lock:
fire = state["left"] > 0
state["left"] -= fire
if fire:
raise InjectedCrash(f"crashed after {attr}")
return out
setattr(mod, attr, wrapped)
self.stack.callback(lambda: setattr(mod, attr, real))
# ── standing the composition up ────────────────────────────────────────────────────────────
class Staged:
def __init__(self, comp, app, public_app, public_base):
self.comp, self.app, self.public_app, self.public_base = comp, app, public_app, public_base
def stage(catalogue, st: ExitStack, engine_for: Callable[[str], Any], *, create: bool = True) -> Staged:
"""The catalogue's composition, wired and built — what both the in-process runner and a laptop deployment
(scripts/live_scenario_check.py: the same apps on a real bucket, served by uvicorn) stand on.
`engine_for(name)` gives the engine for each brick and for "_platform" (the link tables): one SQLite file
each in process, one DataK3 engine for all of them on a bucket. `create=False` when the schema is already
there (a bucket is migrated from schema.sql, never by create_all). Everything is undone by `st`."""
cat = catalogue
comp = cat.compose if isinstance(cat.compose, compose.Composition) else compose.load(cat.compose)
key = "PUBLIC_LINK_BASE_URL"
before = os.environ.get(key)
os.environ[key] = before or PUBLIC_BASE
st.callback(lambda: os.environ.__setitem__(key, before) if before is not None else os.environ.pop(key, None))
for m in comp.members:
wire = cat.storage.get(m.name) or (lambda e, m=m: wire_storage(m, e, create=create))
undo = wire(engine_for(m.name))
if undo:
st.callback(undo)
platform_engine = engine_for("_platform")
if create:
public.md.create_all(platform_engine)
idempotency.md.create_all(platform_engine)
# 0.9.10: an Idempotency-Key's claims live with the link tables. Unset, the store fell back to erp_platform.db,
# which needs BUCKET — so a keyed command on the local tier (or in a scenario) answered 500.
for m in comp.members:
before_conn = m.brick.idempotency_connection
m.brick.idempotency_connection = platform_engine.begin
st.callback(lambda b=m.brick, c=before_conn: setattr(b, "idempotency_connection", c))
before_public = public.CONNECT
public.configure(platform_engine.begin)
st.callback(lambda: public.configure(before_public))
providers_before = dict(capabilities._PROVIDERS)
st.callback(lambda: (capabilities._PROVIDERS.clear(), capabilities._PROVIDERS.update(providers_before)))
local_before = dict(seam._LOCAL)
st.callback(lambda: (seam._LOCAL.clear(), seam._LOCAL.update(local_before)))
app = compose.app(comp)
st.callback(work.reset_clock)
return Staged(comp, app, compose.public_app(comp), os.environ[key].rstrip("/"))
# ── the runner ─────────────────────────────────────────────────────────────────────────────
class InProcessRunner(Runner):
mode = "inprocess"
max_concurrency = 32
def __init__(self, catalogue, *, workdir: Optional[str] = None):
super().__init__(catalogue)
self.workdir = workdir
self.clock = InProcessClock()
self._stack = ExitStack()
self._scenario_stack: Optional[ExitStack] = None
def start(self) -> None:
st = self._stack
try:
self._start(st)
except BaseException:
st.close()
raise
def _start(self, st: ExitStack) -> None:
if self.workdir is None:
self.workdir = st.enter_context(tempfile.TemporaryDirectory(prefix="erp-scenarios-"))
root = pathlib.Path(self.workdir)
engines = {}
def engine_for(name: str):
engines[name] = _sqlite(root / f"{name}.db")
return engines[name]
st.callback(lambda: [e.dispose() for e in engines.values()])
staged = stage(self.catalogue, st, engine_for)
self.comp, self.app, self.public_app, self.public_base = staged.comp, staged.app, staged.public_app, staged.public_base
self.portal = st.enter_context(anyio.from_thread.start_blocking_portal())
st.enter_context(self.portal.wrap_async_context_manager(self.app.router.lifespan_context(self.app)))
base = (os.environ.get("PUBLIC_BASE_URL") or "http://staff.scenario.test").rstrip("/")
self.base = base
# raise_app_exceptions=False: a crash is the 500 a real server answers, not an exception in the scenario.
self.client = httpx.AsyncClient(transport=httpx.ASGITransport(app=self.app, raise_app_exceptions=False),
base_url=base, timeout=60)
self.public_client = httpx.AsyncClient(transport=httpx.ASGITransport(app=self.public_app, raise_app_exceptions=False),
base_url=self.public_base, timeout=60, follow_redirects=False)
st.callback(lambda: self.portal.call(self.public_client.aclose))
st.callback(lambda: self.portal.call(self.client.aclose))
st.callback(work.reset_clock)
def stop(self) -> None:
self._stack.close()
# per scenario: faults and the clock are undone
def faults(self, run: Run) -> Faults:
return Faults(self, self._scenario_stack, run)
def before(self, sc: Scenario) -> Optional[str]:
self._scenario_stack = ExitStack()
return None
def after(self, sc: Scenario) -> None:
try:
if self._scenario_stack is not None:
self._scenario_stack.close()
finally:
self._scenario_stack = None
work.reset_clock()
# identity
def headers(self, persona: str, request_id: str = "") -> dict:
claims = self.catalogue.roles.claims(persona)
h = seam.gateway_headers(claims) if claims else {}
if request_id:
h["X-Request-Id"] = request_id
return h
# the doors
def http(self, persona, method, path, body, params, request_id) -> Response:
headers = self.headers(persona, request_id)
t0 = time.monotonic()
r = self.portal.call(lambda: self.client.request(method, path, json=body if method != "GET" else None,
params=params, headers=headers))
return http_response("http", persona, method, path, {"params": params, "body": body, "headers": headers},
r, int((time.monotonic() - t0) * 1000))
def raw(self, method, path, headers, body=None) -> Response:
t0 = time.monotonic()
r = self.portal.call(lambda: self.client.request(method, path, headers=headers, content=body))
return http_response("http", "raw", method, path, {"headers": headers}, r, int((time.monotonic() - t0) * 1000))
def mcp(self, persona, name, args, request_id) -> Response:
import httpx2
from mcp import ClientSession
from mcp.client.streamable_http import streamable_http_client
headers = self.headers(persona, request_id)
async def go():
async with httpx2.AsyncClient(transport=httpx2.ASGITransport(app=self.app), headers=headers,
base_url=self.base, timeout=60) as http:
async with streamable_http_client(f"{self.base}/mcp", http_client=http) as (r, w):
async with ClientSession(r, w) as s:
await s.initialize()
return await s.call_tool(name, args)
t0 = time.monotonic()
if persona == "anonymous":
# No identity: the agent door answers HTTP 401 before any session — that IS the answer.
r = self.raw("POST", "/mcp", {"content-type": "application/json", "accept": "application/json"},
b'{"jsonrpc":"2.0","id":1,"method":"initialize","params":{}}')
r.exchange.door, r.exchange.persona, r.exchange.method, r.exchange.target = "mcp", persona, "TOOL", name
r.door = "mcp"
return r
result = self.portal.call(go)
return mcp_response(persona, name, args, result, int((time.monotonic() - t0) * 1000))
def public(self, method, url, form, holder, files=None) -> Response:
u = urlparse(url)
if f"{u.scheme}://{u.netloc}" != self.public_base:
raise ValueError(f"{url!r} is not a link on the public app ({self.public_base})")
t0 = time.monotonic()
r = self.portal.call(lambda: self.public_client.request(method, u.path, data=form if method == "POST" else None, files=files or None,
headers={"user-agent": "erp-scenarios"}))
return http_response("public", holder, method, u.path, {"form": form}, r, int((time.monotonic() - t0) * 1000))