"""seam.py — one module calling another, the same way every time.
Part of the platform layer: copy it verbatim, like `db.py` and `sa_token.py`. It is here
because the first two seams in this catalog hand-rolled the same four things, and got two of
them wrong in both copies.
A SEAM is a cross-module call: invoicing asking the ledger to post a journal, sales asking
invoicing to raise one. The rule it exists to keep is simple — a module calls the owning
module's ROUTE and never writes its tables — because the owner's guards (a closed period, a
balance check, a permission) only run on that path.
Four things every seam needs, and what each one costs when it is re-invented:
* **Identity the receiver will actually read.** Apps take `permissions` from the pool JWT and
nothing else — a permission list in the convenience header is unsigned, so honouring it
would let any caller name its own permissions. Both hand-rolled seams sent plain JSON in
the JWT header, which silently parsed to no claims at all; authorisation then fell back to
whatever the ROLE name happened to grant. It worked, for the wrong reason, in both copies.
* **A derived id.** The same source document must always name the same journal, invoice or
item, or a retry bills the customer twice. Derive it from the source; never mint one.
* **Optional by default.** A module the customer has not bought is not an error. With the
base URL unset every call is a no-op that says so, and the business still works — a shop
that keeps its books in Xero still gets invoices.
* **A neighbour that is DOWN is a reply, not an exception.** This is the one both copies got
wrong: they handled an error STATUS and let a connection failure propagate, so the ledger
being unreachable made issuing an invoice raise a 500. An invoice that is issued and not
yet posted is a recoverable state a person can see and retry; an invoice that refuses to
issue because a different module is restarting is not.
IDENTITY, and why there are two kinds. Against a local uvicorn there is no gateway, so the
seam presents gateway-SHAPED headers and the app trusts them — which is the whole reason a
test suite cannot prove authorisation. Deployed, the neighbour is behind the gateway, which
STRIPS inbound identity headers and mints its own: the fabricated header gets a 401 there.
So a deployed seam carries a real pool token from a dodil-appid SERVICE CLIENT — its own
credential, its own role, revocable without touching any person's account.
KNOWN LOSS, stated rather than hidden: the token says the INVOICING SERVICE posted the
journal, not the person who issued the invoice. The receiving module's audit trail names a
machine. Carrying the human as well needs an on-behalf-of claim the platform does not have
yet; until it does, the human is in the source document the journal points at.
TIERS, and why the same code must run at all three. A composition deploys its bricks as one app
(tier 1), as one app each on a shared sign-in pool (tier 2), or as one app and one bucket each
(tier 3). Which tier is a field in compose.yaml, never a code change — so a seam must behave the
same whether its neighbour is in this process or across the network.
Tier 1 reaches the neighbour IN PROCESS, and deliberately through its ROUTES rather than by
calling its use case as a function. A direct call would skip the permission check that lives in
the neighbour's api/ layer, and the same bricks would enforce authorisation at tier 2 and not at
tier 1. Through the routes, a tier-1 call meets every guard a tier-2 call meets.
What still differs, on purpose: in process there is no gateway, so no machine token is fetched and
the acting human travels in `sub`. A tier-1 neighbour's audit trail therefore names the person,
where tier 2 names the service (the KNOWN LOSS above). Tier 1 is strictly more informative; faking
a machine identity to match a known deficiency would make it worse, not consistent.
"""
import asyncio
import base64
import hashlib
import json
import os
import sys
import threading
import time
from typing import Any, Optional
import httpx
from erp_platform import requestlog
TIMEOUT = 20.0
REASON_CHARS = 300 # enough to carry the other module's refusal, not its stack trace
# The app's OWN machine identity, for calling a neighbour that sits behind the gateway.
# A dodil-appid SERVICE CLIENT: its own credential, its own pool role, revocable on its own.
CLIENT_ID = os.environ.get("SEAM_CLIENT_ID", "")
CLIENT_SECRET = os.environ.get("SEAM_CLIENT_SECRET", "")
TOKEN_URL = os.environ.get("SEAM_TOKEN_URL", "")
_token: dict = {"value": "", "expires": 0.0}
def machine_token() -> str:
"""A pool token for this app, cached until shortly before it expires.
Deployed, a neighbour is reached through the GATEWAY, and the gateway strips inbound
identity headers — it mints them, it does not accept them. So the fabricated X-Dodil-User
that works against a local uvicorn gets a 401 in production, and the seam that passed every
test fails the moment it is real. This is that missing half: client_credentials against the
pool's token endpoint, presented as a bearer token the gateway actually verifies.
Returns "" when no service client is configured, which is the local case — tests call
uvicorn directly, with no gateway in front to satisfy.
"""
if not (CLIENT_ID and CLIENT_SECRET and TOKEN_URL):
return ""
if _token["value"] and time.time() < _token["expires"]:
return _token["value"]
r = httpx.post(TOKEN_URL, data={"grant_type": "client_credentials",
"client_id": CLIENT_ID, "client_secret": CLIENT_SECRET},
timeout=TIMEOUT)
r.raise_for_status()
body = r.json()
_token["value"] = body["access_token"]
# 60s of slack: a token that expires mid-flight fails a posting that was otherwise fine.
_token["expires"] = time.time() + max(int(body.get("expires_in", 300)) - 60, 30)
return _token["value"]
def derived_id(*parts: Any) -> int:
"""A stable 56-bit id for a source document — the same inputs always name the same row.
56 bits sits inside a signed BIGINT with room to spare and makes a collision improbable at
any plausible document count. An integer because the graph walks take integer node ids, and
derived because this platform has no sequences: the id must be computed by the caller, and
a computed id is what makes a retry safe.
"""
seed = ":".join(str(p) for p in parts)
return int.from_bytes(hashlib.blake2b(seed.encode(), digest_size=7).digest(), "big")
def claims(actor: str, roles: list, permissions: list) -> dict:
"""The identity this app presents to its neighbour, carrying the acting human in `sub`."""
return {"sub": actor or "service", "email": actor or "", "connection": "service",
"app_roles": list(roles), "permissions": list(permissions)}
ACTING_FOR_HEADER = "X-Dodil-Acting-For"
SERVICE_SOURCE_HEADER, SERVICE_SOURCE = "X-Dodil-Auth-Source", "service" # 0.9.7: a brick's own call, in process
def _owner_of(module: str) -> str:
"""The brick a module belongs to: brick_inventory.x -> inventory, brick_sales_order.y -> sales_order."""
top = (module or "").split(".")[0]
return top[len("brick_"):] if top.startswith("brick_") else top
ACT_FOR = "erp:seam:act_for" # the permission that lets a SERVICE say whose eyes a read is for (auth.py)
def caller_claims(caller: dict) -> dict:
"""The identity a seam presents IN PROCESS when it acts as the caller: exactly the person — their sub,
email, roles and permissions, nothing of the seam's own."""
return {"sub": caller.get("sub") or caller.get("email") or "", "email": caller.get("email") or "",
"connection": caller.get("connection") or "seam", "app_roles": list(caller.get("roles") or []),
"permissions": list(caller.get("permissions") or [])}
def acting_for(caller: dict) -> str:
"""The X-Dodil-Acting-For value: who a service is reading for, across apps (tiers 2–3). Roles only —
never permissions: the receiving brick scopes rows by them, and a permission is not the service's to
pass on."""
return json.dumps({"sub": caller.get("sub") or "", "email": caller.get("email") or "",
"app_roles": list(caller.get("roles") or [])}, separators=(",", ":"))
def gateway_headers(c: dict) -> dict:
"""The two headers the Ignite gateway injects, in the shape apps actually parse.
The JWT is decode-only at the app — the edge verified the signature and strips inbound
copies — but it must still be a THREE-PART token or the claims are dropped on the floor.
"""
payload = base64.urlsafe_b64encode(json.dumps(c).encode()).decode().rstrip("=")
header = base64.urlsafe_b64encode(b'{"alg":"none","typ":"JWT"}').decode().rstrip("=")
return {"X-Dodil-User": json.dumps(c), "X-Dodil-User-JWT": f"{header}.{payload}.unsigned"}
# Tier 1 — neighbours running in THIS process, by seam name. Filled by register_local().
_LOCAL: dict = {}
def register_local(name: str, app: Any) -> None:
"""Tier 1: this neighbour runs in the same process — call its routes without the network.
The composition's assembler calls this once for each brick it mounts. A registration wins over
a base URL: when the neighbour is in this process, leaving through the network to reach it is a
slower way of calling the same object, and one more thing that can be down.
"""
_LOCAL[name] = app
def unregister_local(name: str) -> None:
_LOCAL.pop(name, None)
class _InProcessTransport(httpx.BaseTransport):
"""Delivers a seam's request to a neighbour's ASGI app in this process. No socket is opened.
Built on httpx's own ASGITransport rather than Starlette's TestClient. TestClient is deprecated
for this (it warns to move to httpx2), and BOTH its defaults are wrong for a seam: it re-raises
a neighbour's crash into the caller and it follows redirects, where the network path does
neither.
"""
def __init__(self, app: Any, tree: Any = None):
# raise_app_exceptions=False is why this class exists. With httpx's default (True) a
# neighbour's crash is re-raised; caught by the thread below, it would be reported as
# "<neighbour> unreachable" with no status — a FALSE diagnosis that sends whoever reads it
# looking for a network fault in a service that is up and crashing. False turns the crash
# into the 500 a real neighbour returns, so the reply says what actually happened.
# (Verified by mutation: flipping it makes test_a_crash_is_a_reply_not_an_exception fail.)
self._asgi = httpx.ASGITransport(app=app, raise_app_exceptions=False)
# 0.9.13: the caller's request tree (erp_platform.repository's identity map), carried onto the delivering
# thread so the neighbour's request reads through it — keyed by the neighbour's own caller.
self._tree = tree
def handle_request(self, request: httpx.Request) -> httpx.Response:
timeout = (request.extensions.get("timeout") or {}).get("read") or TIMEOUT
out: dict = {}
async def deliver() -> None:
response = await self._asgi.handle_async_request(request)
out["body"] = await response.aread()
out["status"], out["headers"] = response.status_code, response.headers
def run() -> None:
# A fresh thread with its own event loop, on every call. A shared loop deadlocks the
# first time an async handler running on it makes a nested seam call: it blocks the very
# loop its neighbour needs. A thread per call cannot deadlock that way.
try:
from erp_platform import repository
repository.adopt_tree(self._tree)
asyncio.run(deliver())
except BaseException as exc: # noqa: BLE001 — surfaced to the caller below
out["error"] = exc
worker = threading.Thread(target=run, name="seam-in-process", daemon=True)
worker.start()
worker.join(timeout)
if worker.is_alive():
# The type the network path raises, so _call reports it identically.
raise httpx.ReadTimeout(f"no reply within {timeout:.0f}s", request=request)
if "error" in out:
raise httpx.TransportError(str(out["error"])[:REASON_CHARS], request=request)
return httpx.Response(out["status"], headers=out["headers"], content=out["body"],
request=request)
class Reply:
"""What came back, in a shape a caller can return to a person without unpacking HTTP."""
def __init__(self, ok: bool, status: int = 0, reason: str = "", data: Optional[dict] = None):
self.ok, self.status, self.reason, self.data = ok, status, reason, data or {}
def __bool__(self) -> bool:
return self.ok
def problem(self) -> dict:
"""The refusal, ready to merge into a result: `{"posted": False, **r.problem()}`.
The other module's reason travels. "Not posted" without a why is a support ticket.
"""
out: dict = {"reason": self.reason}
if self.status:
out["status"] = self.status
return out
class Seam:
"""One neighbouring module, reachable or not.
LEDGER = Seam("ledger", "GL_BASE_URL", roles=["bookkeeper"],
permissions=["gl:journal:post", "gl:subledger:write"])
if not LEDGER.installed:
return {"posted": False, "reason": "no ledger configured"}
r = LEDGER.post("/journal-entry/journals", body, actor)
if not r:
return {"posted": False, **r.problem()}
"""
def __init__(self, name: str, base_env: str, roles: list, permissions: list,
timeout: float = TIMEOUT, owner: Optional[str] = None):
self.name, self.base_env = name, base_env
# 0.9.7: the brick making the call — named on a service-client call in process (identity["service"]).
# By default the package that declares the seam: brick_inventory.infrastructure.seams -> "inventory".
self.owner = owner if owner is not None else _owner_of(sys._getframe(1).f_globals.get("__name__", ""))
self.base = os.environ.get(base_env, "").rstrip("/")
self.roles, self.permissions = roles, permissions
self.timeout = timeout
@property
def local(self) -> Any:
"""The neighbour's app when it runs in this process (tier 1), else None."""
return _LOCAL.get(self.name)
@property
def installed(self) -> bool:
"""Configured nowhere = the customer does not have this module. Not an error.
Configured means in this process (tier 1) or at a base URL (tiers 2 and 3).
"""
return self.local is not None or bool(self.base)
def _headers(self, actor: str, in_process: bool = False, caller: Optional[dict] = None) -> dict:
"""A real bearer token where there is a gateway; gateway-shaped headers where there is not.
The token is the authorisation. The convenience header rides along so the receiving
module can still name the acting human in its audit trail — the gateway replaces it
with what the token says, so it is a courtesy in local runs, not a claim to authority.
AS THE CALLER (`caller=`, 0.5): "may THIS person see deal O-1?" has to be answered by the brick that
owns the deal, with the person's own scope — asked as the seam's service identity, a quote could be
attached to a deal its rep cannot see (the CRM audit, 2026-10-17). In process the neighbour gets
exactly the caller's identity. Across apps it gets the machine token (the authority) plus
X-Dodil-Acting-For (whose eyes), which a receiving brick honours only from a caller holding
erp:seam:act_for — a service client's permission, never a person's (auth.identity_from_headers).
"""
if caller is not None and in_process:
# 0.9.7: the person's own identity (their rows), plus THIS brick's service grant for the door alone
# (`service_permissions`, auth.may_run): an onward read runs on the calling brick's own machine grant,
# never on permissions the person happens to hold, and never wider than the person's rows.
headers = gateway_headers({**caller_claims(caller), "service_permissions": list(self.permissions),
"acting_service": self.owner})
elif in_process:
# 0.9.7: the brick as its OWN service client — a service, not a person, and which one. The gateway mints
# (and strips any inbound) X-Dodil-Auth-Source and the JWT, so only a seam in this process says so.
headers = gateway_headers({**claims(actor, self.roles, self.permissions), "service": self.owner})
headers[SERVICE_SOURCE_HEADER] = SERVICE_SOURCE
else:
headers = gateway_headers(claims(actor or (caller or {}).get("email", ""), self.roles, self.permissions))
if caller is not None:
headers[ACTING_FOR_HEADER] = acting_for(caller)
# The request this call is part of, so the neighbour's log lines carry the same id. Read here,
# on the caller's thread: the in-process transport delivers on a fresh thread, which would not.
rid = requestlog.request_id()
if rid:
headers[requestlog.HEADER] = rid
if in_process:
# No gateway between two bricks in one process, so there is nothing for a token to
# satisfy — and fetching one would put a network call back into a path built to have
# none. The edge already authenticated the outer request; the process is the boundary.
return headers
token = machine_token()
if token:
headers["Authorization"] = f"Bearer {token}"
return headers
def _call(self, method: str, path: str, actor: str, caller: Optional[dict] = None, **kw) -> Reply:
if not self.installed:
return Reply(False, reason=f"{self.name} is not installed")
from erp_platform import repository
app = self.local
try:
if app is not None:
# Tier 1. Same request, same guards, same reply — just no socket.
with httpx.Client(transport=_InProcessTransport(app, repository.current_tree()),
base_url="http://in-process") as client:
r = client.request(method, path, headers=self._headers(actor, True, caller),
timeout=self.timeout, **kw)
else:
r = httpx.request(method, f"{self.base}{path}", headers=self._headers(actor, False, caller),
timeout=self.timeout, **kw)
except httpx.HTTPError as exc:
# Unreachable, refused, timed out, DNS gone. The caller's own work already
# succeeded; this is a fact to report, not an exception to raise through it.
return Reply(False, reason=f"{self.name} unreachable: {str(exc)[:REASON_CHARS]}")
finally:
if app is None and method.upper() != "GET":
# A neighbour in another process may have called back into this brick, in a request this one's
# identity map cannot see: forget what this request read (in process, its writes drop entries).
repository.forget_all()
if r.status_code >= 300:
return Reply(False, status=r.status_code, reason=r.text[:REASON_CHARS])
try:
data = r.json()
except ValueError:
data = {}
return Reply(True, status=r.status_code, data=data if isinstance(data, dict) else {"body": data})
def post(self, path: str, body: Optional[dict] = None, actor: str = "",
params: Optional[dict] = None, caller: Optional[dict] = None) -> Reply:
"""`caller`: ask AS that person (their row scope), not as this seam's service identity."""
return self._call("POST", path, actor, caller=caller, json=body or {}, params=params)
def get(self, path: str, params: Optional[dict] = None, actor: str = "",
caller: Optional[dict] = None) -> Reply:
"""`caller`: ask AS that person (their row scope), not as this seam's service identity."""
return self._call("GET", path, actor, caller=caller, params=params)