"""capabilities — what a brick may ask another brick to DO in the outside world, by name, not by vendor.
hr must "provision the starter's account" without knowing Google. capabilities.yaml (shipped beside this
file) is the contract: hr REQUIRES `identity.provision_user`, a provider brick (google-workspace) PROVIDES
it, compose.yaml maps one to the other. Its eight rules are what make a call safe to retry from a ticking
run; written once here, no provider re-implements rules 1–7 and no consumer re-implements rule 8.
PROVIDER SIDE
GWS = Brick("gws")
@GWS.capability("identity.provision_user")
def provision_user(call: Call):
'''look the account up; create it if it is missing'''
...
return account_id, {"account_id": ..., "primary_email": ..., "created": True}, "Account created."
@provision_user.plan # optional: the dry run's plan, in the vendor's words
def _(call: Call) -> list: ...
declares that the brick PROVIDES the capability and generates `POST /capabilities/identity.provision_user`
(body {idempotency_key, dry_run, input}) and the MCP tool `identity_provision_user`, with the contract's
destructive hint and the permission `gws:capability:invoke`. The handler gets VALIDATED input and returns
(provider_ref, output, message), or raises Transient / Failed / ConnectionRequired. Everything else —
claim, lease, replay, release, 409, dry run, action log — is done here.
RULE 1, THE CLAIM, and why it is a plain INSERT. A capability call comes from a run that ticks: the same
step is asked again after a timeout, a pod restart, a second tick. The provider must act once. So it
claims the key with a plain INSERT on a primary key; a concurrent duplicate fails with 23505 (measured on
DataK3 0.2.1, 2026-10-17: 20 concurrent INSERTs of one key, 1 acknowledged, 19 rejected) and the loser
re-reads the winner. An upsert would let every caller "win" and act — tests/test_capabilities.py proves
it by putting one back.
The key of the claims table is (capability, idempotency_key, GENERATION). The first claim is generation 1,
which is exactly the plain INSERT on (capability, key) the contract describes. The generation exists for
the LEASE: a claim whose owner died (pod gone between claim and result) expires after ~120s, and the next
caller takes over by inserting generation 2 — again a plain INSERT, so two callers racing to take over
also have exactly one winner. Taking over by UPDATE or by DELETE + INSERT would need the engine to detect
the race; a fresh primary key needs nothing but the property that was measured.
* only a TERMINAL result (done; failed with retryable=false) is stored and replayed;
* a non-terminal one (failed retryable, connection_required) RELEASES the claim, so a retry acts;
* an unexpired claim with no result answers in_progress; an expired one is re-attempted — which is why
every handler must ALSO be idempotent at the vendor (look up before create);
* the same key with DIFFERENT input is 409, compared by a SHA-256 of the canonical JSON of the
validated input.
RULE 2: a dry run validates and returns the plan, and touches no claim. RULE 5: every call — dry runs,
replays and in_progress answers included — writes one row to the provider's action log, with the fields
the provider marks secret replaced by "[redacted]". RULE 7: the handler raises Transient for a vendor
blip; it is retried here with backoff inside the call (attempts=3), then answered failed retryable=true.
THE TABLES a provider ships, prefixed with its brick name like every table it owns. `ddl(<brick>)` prints
them for its schema.sql (erp-compose check refuses a provider whose schema.sql lacks them); `tables()` gives
the SQLAlchemy objects. Lease times use the REAL clock, never work.now(): they are about pods, not business.
CONSUMER SIDE
PROVISION = CapabilitySeam("identity.provision_user")
r = PROVISION.call(f"{run_id}|{step_no}", {...}, actor=caller["email"])
if isinstance(r, ManualTask): store r.as_row() as a task a person completes (rule 8)
elif r.status == "done": ...
The provider is resolved per call: tier 1, the composition registered it (compose.app); tiers 2/3,
`CAPABILITY_IDENTITY_PROVISION_USER=gws` names the provider brick and `GWS_BASE_URL` — the same variable an
ordinary seam to that brick uses — says where it is; otherwise there is NO provider, and the call returns a
ManualTask carrying the exact validated input and the plan in words. A provider answering
connection_required becomes a ManualTask too. Wiring a provider later makes the same call automatic.
Input is validated against the contract BEFORE anything is sent: a consumer's bug fails at its own line.
"""
import functools
import hashlib
import importlib.resources
import json
import logging
import os
import re
import time
import typing
import uuid
from dataclasses import asdict, dataclass
from datetime import datetime, timedelta, timezone
from typing import Any, Callable, Literal, Optional, Union
import yaml
from pydantic import AfterValidator, BaseModel, ConfigDict, StringConstraints, ValidationError, create_model
from sqlalchemy import (Boolean, Column, DateTime, Integer, MetaData, String, Table, and_, delete, insert,
select)
from erp_platform import requestlog
log = logging.getLogger("erp_platform.capabilities")
STATUSES = ("done", "planned", "in_progress", "connection_required", "failed")
LEASE_SECONDS = 120.0
ATTEMPTS = 3
BACKOFF = (0.5, 2.0) # seconds before vendor attempts 2 and 3 — well inside the lease
REDACTED = "[redacted]"
MANUAL = "manual"
class UnknownCapability(KeyError):
"""A capability name that is not in capabilities.yaml."""
def __str__(self) -> str:
return str(self.args[0]) if self.args else "unknown capability"
class InvalidCapabilityInput(ValueError):
"""Input (or a provider's output) that does not match the contract. Lists every problem."""
class Transient(Exception):
"""A provider handler's vendor failure that may succeed later: timeout, 429, 5xx (rule 7)."""
class Failed(Exception):
"""A provider handler's vendor failure that will not succeed by retrying (failed, retryable=false)."""
class ConnectionRequired(Exception):
"""No usable connection to the customer's tool (rule 3). The message names what is missing."""
class CapabilityRefused(Exception):
"""The provider answered with an HTTP error: the consumer got the call wrong (rule 4)."""
def __init__(self, status: int, reason: str):
super().__init__(f"{status}: {reason}")
self.status, self.reason = status, reason
# ── the contract ───────────────────────────────────────────────────────────────────────────
_SCALARS = {"str": str, "int": int, "bool": bool, "boolean": bool}
def _iso(value: str) -> str:
try:
datetime.fromisoformat(value.replace("Z", "+00:00"))
except ValueError:
raise ValueError(f"{value!r} is not an ISO 8601 date-time") from None
return value
def _top_level(text: str, sep: str) -> list:
"""Split on `sep` outside braces: "list of {side: US|THEM}|null" is two parts, not three."""
parts, depth, cur = [], 0, ""
for ch in text:
depth += (ch == "{") - (ch == "}")
if ch == sep and depth == 0:
parts.append(cur)
cur = ""
else:
cur += ch
return [p.strip() for p in parts + [cur]]
_CHOICE = re.compile(r"[a-z_]+|[A-Z][A-Z0-9_]*") # lower-case words, or UPPER_CASE codes (US, IN_APPROVAL)
@dataclass(frozen=True)
class FieldSpec:
"""One input or output field, parsed from the contract's type string ("str|null", "a | b", ...)."""
name: str
type: str # as written in the contract
kind: str # str | int | bool | datetime | enum | list | objects | object | map
nullable: bool = False
choices: tuple = ()
items: tuple = () # kinds "objects" and "object": the FieldSpecs of each element / of the object
@classmethod
def _fields(cls, inner: str) -> tuple:
items = []
for piece in _top_level(inner, ","):
n, _, t = piece.partition(":")
items.append(cls.parse(n.strip(), t.strip() or "str"))
return tuple(items)
@classmethod
def parse(cls, name: str, raw: Any) -> "FieldSpec":
text = str(raw).strip()
parts = _top_level(text, "|")
nullable = "null" in parts
core = " | ".join(p for p in parts if p != "null") if nullable else text
if core.startswith("list of {") and core.endswith("}"):
return cls(name, text, "objects", nullable, items=cls._fields(core[len("list of {"):-1]))
if core.startswith("{") and core.endswith("}"):
return cls(name, text, "object", nullable, items=cls._fields(core[1:-1]))
if core == "list of str":
return cls(name, text, "list", nullable)
if core in ("map of str", "map of any"):
return cls(name, text, "map", nullable, choices=(core.rsplit(" ", 1)[1],))
if core == "iso datetime":
return cls(name, text, "datetime", nullable)
if core in _SCALARS:
return cls(name, text, "bool" if core == "boolean" else core, nullable)
choices = tuple(p.strip() for p in core.split("|"))
if all(_CHOICE.fullmatch(c) for c in choices) and (len(choices) > 1 or choices[0].isupper()):
# Two or more words; or ONE upper-case code, a literal (contracts.send's state: OUT_FOR_SIGNATURE).
return cls(name, text, "enum", nullable, choices=choices)
raise ValueError(f"field {name}: the contract type {text!r} is not one this module understands")
def annotation(self, model_name: str) -> Any:
if self.kind == "str":
t: Any = str if self.nullable else typing.Annotated[str, StringConstraints(min_length=1)]
elif self.kind == "int":
t = int
elif self.kind == "bool":
t = bool
elif self.kind == "datetime":
t = typing.Annotated[str, AfterValidator(_iso)]
elif self.kind == "enum":
t = Literal[self.choices] # type: ignore[valid-type]
elif self.kind == "list":
t = list[str]
elif self.kind == "map":
# "map of any": values the provider interprets (a contract's variables: text, numbers, dates as ISO
# strings, booleans) — JSON scalars only, never a nested structure.
t = dict[str, str] if self.choices == ("str",) else dict[str, Union[str, int, float, bool, None]]
elif self.kind == "object":
t = _model(f"{model_name}_{self.name}", self.items)
else:
t = list[_model(f"{model_name}_{self.name}", self.items)]
return Optional[t] if self.nullable else t
_STRICT = ConfigDict(strict=True, extra="forbid")
def _model(name: str, fields: tuple) -> type:
"""A strict pydantic model for a field list: "1" is not an int, an unknown key is an error."""
spec = {f.name: (f.annotation(name), None if f.nullable else ...) for f in fields}
return create_model(name, __config__=_STRICT, **spec)
@dataclass(frozen=True)
class Capability:
name: str
summary: str
input: tuple # FieldSpec, in contract order
output: tuple
destructive: bool
read: bool = False # answered without a claim (drive.transfer_status)
notes: str = ""
@property
def tool_name(self) -> str:
"""The MCP tool: dots become underscores (identity_provision_user)."""
return self.name.replace(".", "_")
@property
def route(self) -> str:
"""The provider's route keeps the dots."""
return f"/capabilities/{self.name}"
@property
def env(self) -> str:
"""The variable naming this capability's provider brick at tiers 2–3."""
return "CAPABILITY_" + re.sub(r"[^A-Z0-9]", "_", self.name.upper())
@functools.cached_property
def input_model(self) -> type:
return _model(self.tool_name + "_input", self.input)
@functools.cached_property
def output_model(self) -> type:
return _model(self.tool_name + "_output", self.output)
def _validate(self, model: type, data: Any, what: str) -> dict:
if isinstance(data, BaseModel):
data = data.model_dump()
try:
return model.model_validate(data).model_dump()
except ValidationError as e:
problems = "; ".join(f"{'.'.join(str(p) for p in err['loc']) or what}: {err['msg']}"
for err in e.errors())
raise InvalidCapabilityInput(f"{self.name} {what}: {problems}") from None
def validate_input(self, data: Any) -> dict:
"""The input, checked against the contract and normalised (absent nullable fields → None)."""
return self._validate(self.input_model, data, "input")
def validate_output(self, data: Any) -> dict:
return self._validate(self.output_model, data, "output")
def input_hash(self, data: dict) -> str:
"""SHA-256 of the canonical JSON of VALIDATED input: the same request always hashes the same."""
canonical = json.dumps(data, sort_keys=True, separators=(",", ":"), ensure_ascii=False, default=str)
return hashlib.sha256(canonical.encode()).hexdigest()
@dataclass(frozen=True)
class Contract:
version: str
result: dict # RESULT field → its description in the contract
capabilities: dict # name → Capability
def get(self, name: str) -> Capability:
try:
return self.capabilities[name]
except KeyError:
raise UnknownCapability(f"{name!r} is not a capability in capabilities.yaml "
f"{self.version}") from None
def __contains__(self, name: str) -> bool:
return name in self.capabilities
def load(path=None) -> Contract:
"""Parse a contract file. Every field type is parsed here, so a typo in the YAML fails at load."""
text = (open(path).read() if path else
importlib.resources.files("erp_platform").joinpath("capabilities.yaml").read_text())
raw = yaml.safe_load(text)
caps = {}
for name, c in (raw.get("capabilities") or {}).items():
caps[name] = Capability(
name=name, summary=str(c.get("summary", "")).strip(),
input=tuple(FieldSpec.parse(k, v) for k, v in (c.get("input") or {}).items()),
output=tuple(FieldSpec.parse(k, v) for k, v in (c.get("output") or {}).items()),
destructive=bool(c.get("destructive", False)), read=bool(c.get("read", False)),
notes=" ".join(str(c.get("notes", "")).split()))
return Contract(str(raw["version"]), dict(raw.get("result") or {}), caps)
@functools.lru_cache(maxsize=1)
def contract() -> Contract:
"""The contract this platform ships (src/erp_platform/capabilities.yaml), loaded once."""
return load()
# ── the RESULT, and the manual task ─────────────────────────────────────────────────────────
@dataclass(frozen=True)
class Result:
"""The answer to every capability call — the contract's `result:` shape, field for field."""
status: str
retryable: bool = False
provider: str = ""
provider_ref: Optional[str] = None
output: Optional[dict] = None
plan: Optional[list] = None
message: str = ""
idempotent_replay: bool = False
def __post_init__(self) -> None:
if self.status not in STATUSES:
raise ValueError(f"status must be one of {STATUSES}, got {self.status!r}")
@property
def terminal(self) -> bool:
"""Stored and replayed: done, or failed for good (rule 1)."""
return self.status == "done" or (self.status == "failed" and not self.retryable)
def as_dict(self) -> dict:
return asdict(self)
@classmethod
def from_dict(cls, d: dict) -> "Result":
names = {f for f in cls.__dataclass_fields__}
return cls(**{k: v for k, v in d.items() if k in names})
def describe(cap: Capability, data: dict) -> str:
"""The plan in words, for a person doing the step by hand."""
def show(v: Any) -> str:
return json.dumps(v, ensure_ascii=False) if isinstance(v, (list, dict)) else str(v)
given = "; ".join(f"{k}: {show(v)}" for k, v in data.items() if v not in (None, [], ""))
return f"{cap.summary.rstrip('.')} — {given}." if given else cap.summary
@dataclass(frozen=True)
class ManualTask:
"""Rule 8: there is no provider (or it has no connection), so a PERSON does this step.
It carries what the provider would have been sent — the capability and the exact validated input —
and the plan in words. The consumer stores it as one of its own tasks (`as_row()`), shows `plan` to
the person, and when they mark it done calls `complete(provider_ref=...)`, which returns the same
`done` Result a provider would have: the code after it does not know the difference.
"""
capability: str
idempotency_key: str
input: dict
plan: str
reason: str
destructive: bool
summary: str
status: str = MANUAL
def as_row(self) -> dict:
"""Flat, JSON-safe columns for the consumer's task table (input as canonical JSON)."""
return {"capability": self.capability, "idempotency_key": self.idempotency_key,
"input_json": json.dumps(self.input, sort_keys=True, ensure_ascii=False),
"plan": self.plan, "reason": self.reason, "destructive": self.destructive,
"summary": self.summary}
@classmethod
def from_row(cls, row: dict) -> "ManualTask":
return cls(row["capability"], row["idempotency_key"], json.loads(row["input_json"]), row["plan"],
row["reason"], bool(row["destructive"]), row["summary"])
def complete(self, provider_ref: Optional[str] = None, output: Optional[dict] = None,
message: str = "") -> Result:
"""The person did it. `provider_ref` is what they made (the account id), if they recorded it."""
return Result("done", provider=MANUAL, provider_ref=provider_ref, output=output,
message=message or f"done by hand: {self.summary}")
# ── provider side: the claim store ─────────────────────────────────────────────────────────
_TABLES: dict = {}
@dataclass(frozen=True)
class CapabilityTables:
metadata: MetaData
claims: Table
log: Table
def tables(brick_name: str) -> CapabilityTables:
"""The provider's two tables as SQLAlchemy objects (`<brick>_capability_claims`, `_capability_log`)."""
if brick_name not in _TABLES:
prefix = re.sub(r"[^a-z0-9]", "_", brick_name.lower())
md = MetaData()
claims = Table(
f"{prefix}_capability_claims", md,
Column("capability", String, primary_key=True),
Column("idempotency_key", String, primary_key=True),
Column("generation", Integer, primary_key=True, autoincrement=False),
Column("input_hash", String, nullable=False),
Column("attempt_id", String, nullable=False),
Column("claimed_at", DateTime, nullable=False),
Column("lease_until", DateTime, nullable=False),
Column("status", String),
Column("result", String),
Column("finished_at", DateTime))
log_t = Table(
f"{prefix}_capability_log", md,
Column("capability", String, primary_key=True),
Column("idempotency_key", String, primary_key=True),
Column("attempt_id", String, primary_key=True),
Column("event_at", DateTime, nullable=False),
Column("actor", String),
Column("dry_run", Boolean, nullable=False),
Column("input", String, nullable=False),
Column("status", String, nullable=False),
Column("retryable", Boolean),
Column("provider_ref", String),
Column("message", String),
Column("replay", Boolean, nullable=False),
Column("tries", Integer, nullable=False),
Column("error", String),
Column("request_id", String))
_TABLES[brick_name] = CapabilityTables(md, claims, log_t)
return _TABLES[brick_name]
def ddl(brick_name: str) -> str:
"""The CREATE TABLEs a provider brick puts in its schema.sql (erp-compose check looks for them)."""
p = re.sub(r"[^a-z0-9]", "_", brick_name.lower())
return f"""\
-- Capability claims (erp_platform.capabilities, rule 1). A claim is a PLAIN INSERT: a duplicate key fails
-- with 23505 and the caller re-reads the winner. generation 1 is the first claim, and a takeover after the
-- lease expired inserts generation n+1, so a takeover race has one winner too. Only terminal results
-- (done, or failed with retryable=false) are kept in `result`, and a non-terminal outcome deletes its claim.
CREATE TABLE IF NOT EXISTS {p}_capability_claims (
capability VARCHAR NOT NULL,
idempotency_key VARCHAR NOT NULL,
generation INTEGER NOT NULL,
input_hash VARCHAR NOT NULL, -- SHA-256 of the canonical validated input: same key, other input = 409
attempt_id VARCHAR NOT NULL,
claimed_at TIMESTAMP NOT NULL,
lease_until TIMESTAMP NOT NULL, -- UTC; an unfinished claim past this is re-attempted
status VARCHAR, -- NULL while in progress; done | failed once terminal
result VARCHAR, -- the terminal RESULT, as JSON
finished_at TIMESTAMP,
PRIMARY KEY (capability, idempotency_key, generation)
);
-- One row per call (rule 5) — dry runs, replays and in_progress answers included. Secret inputs redacted.
CREATE TABLE IF NOT EXISTS {p}_capability_log (
capability VARCHAR NOT NULL,
idempotency_key VARCHAR NOT NULL,
attempt_id VARCHAR NOT NULL,
event_at TIMESTAMP NOT NULL,
actor VARCHAR,
dry_run BOOLEAN NOT NULL,
input VARCHAR NOT NULL, -- JSON, secret fields replaced by "[redacted]"
status VARCHAR NOT NULL, -- a RESULT status, or "error" when the handler crashed
retryable BOOLEAN,
provider_ref VARCHAR,
message VARCHAR,
replay BOOLEAN NOT NULL,
tries INTEGER NOT NULL, -- vendor attempts inside this call (rule 7)
error VARCHAR, -- an exception's TYPE, never its message
request_id VARCHAR,
PRIMARY KEY (capability, idempotency_key, attempt_id)
);
"""
def _utcnow() -> datetime:
"""Real time, naive UTC — what a TIMESTAMP column holds and hands back on DataK3 and SQLite."""
return datetime.now(timezone.utc).replace(tzinfo=None)
def _is_duplicate(exc: BaseException) -> bool:
from sqlalchemy.exc import IntegrityError
text = str(exc).lower()
return isinstance(exc, IntegrityError) or "23505" in text or "duplicate key" in text
def _default_connect():
from erp_platform import db # imported late: it needs BUCKET
return db.engine.begin()
class _Store:
"""The claims and the log, each operation its own short transaction (the vendor call holds none)."""
def __init__(self, brick_name: str, connect: Callable[[], Any]):
self.t = tables(brick_name)
self.connect = connect
def rows(self, cap: str, key: str) -> list:
c = self.t.claims.c
with self.connect() as conn:
result = conn.execute(select(self.t.claims).where(and_(c.capability == cap, c.idempotency_key == key))
.order_by(c.generation))
return [dict(r._mapping) for r in result]
def insert_claim(self, cap: str, key: str, gen: int, input_hash: str, attempt: str, lease: float) -> bool:
"""The claim: a PLAIN INSERT. True if this call won it; False on a duplicate key."""
now = _utcnow()
try:
with self.connect() as conn:
conn.execute(insert(self.t.claims).values(
capability=cap, idempotency_key=key, generation=gen, input_hash=input_hash,
attempt_id=attempt, claimed_at=now, lease_until=now + timedelta(seconds=lease)))
return True
except Exception as exc: # noqa: BLE001 — re-raised unless a duplicate
if _is_duplicate(exc):
return False
raise
def finish(self, row: dict, result: Result) -> None:
"""Store a terminal result on this call's own claim. Through repository.upsert: if a takeover's
owner deleted the row meanwhile, it is written back — the action DID complete, so replaying it is
right."""
from erp_platform.repository import upsert
full = {**row, "status": result.status, "result": json.dumps(result.as_dict(), default=str),
"finished_at": _utcnow()}
with self.connect() as conn:
upsert(conn, self.t.claims, [full], ("capability", "idempotency_key", "generation"), insert_only=())
def release(self, cap: str, key: str, gen: int) -> None:
"""A non-terminal outcome: drop this call's claim (only its own generation) so a retry acts."""
c = self.t.claims.c
with self.connect() as conn:
conn.execute(delete(self.t.claims).where(and_(c.capability == cap, c.idempotency_key == key,
c.generation == gen, c.status.is_(None))))
def write_log(self, **row: Any) -> None:
with self.connect() as conn:
conn.execute(insert(self.t.log).values(**row))
# ── provider side: the decorator ───────────────────────────────────────────────────────────
@dataclass(frozen=True)
class Call:
"""What a provider's handler receives: validated input, and the key to derive vendor ids from."""
capability: Capability
idempotency_key: str
input: dict
caller: dict
dry_run: bool = False
@property
def actor(self) -> str:
return self.caller.get("email") or self.caller.get("sub") or ""
_sleep = time.sleep
class Provided:
"""One capability a brick provides: its handler, its planner, and everything rules 1–7 need."""
def __init__(self, brick: Any, cap: Capability, handler: Callable, *, secrets: tuple, lease_seconds: float,
attempts: int):
unknown = set(secrets) - {f.name for f in cap.input}
if unknown:
raise ValueError(f"{cap.name}: secrets {sorted(unknown)} are not input fields of the capability")
if attempts < 1 or lease_seconds <= 0:
raise ValueError("attempts ≥ 1 and a positive lease, please")
self.brick, self.capability, self.handler = brick, cap, handler
self.secrets, self.lease_seconds, self.attempts = tuple(secrets), float(lease_seconds), attempts
self.planner: Optional[Callable] = None
functools.update_wrapper(self, handler)
def plan(self, fn: Callable) -> Callable:
"""Register the dry run's planner: fn(call) -> [{action, target, payload}, ...]."""
self.planner = fn
return fn
def __call__(self, call: Call):
return self.handler(call)
@property
def store(self) -> _Store:
return _Store(self.brick.name, self.brick.capability_connection or _default_connect)
def redact(self, data: dict) -> dict:
return {k: (REDACTED if k in self.secrets and v is not None else v) for k, v in data.items()}
def _plan(self, call: Call) -> list:
if self.planner is None:
return [{"action": self.capability.summary, "target": self.capability.name,
"payload": self.redact(call.input)}]
plan = self.planner(call)
if not isinstance(plan, list) or not all(isinstance(p, dict) and "action" in p for p in plan):
raise TypeError(f"{self.capability.name}: a plan is a list of {{action, target, payload}}")
return [{**p, "payload": self.redact(p["payload"])} if isinstance(p.get("payload"), dict) else p
for p in plan]
def _act(self, call: Call) -> tuple:
"""Run the handler, retrying a Transient with backoff (rule 7). Returns (Result, tries)."""
provider = self.brick.name
for attempt in range(1, self.attempts + 1):
try:
answer = self.handler(call)
except Transient as t:
if attempt == self.attempts:
return Result("failed", retryable=True, provider=provider, message=str(t)), attempt
_sleep(BACKOFF[min(attempt - 1, len(BACKOFF) - 1)])
continue
except Failed as f:
return Result("failed", retryable=False, provider=provider, message=str(f)), attempt
except ConnectionRequired as c:
return Result("connection_required", provider=provider, message=str(c)), attempt
if not (isinstance(answer, tuple) and len(answer) == 3):
raise TypeError(f"{self.capability.name}: a handler returns (provider_ref, output, message)")
ref, output, message = answer
return Result("done", provider=provider, provider_ref=None if ref is None else str(ref),
output=self.capability.validate_output(output), message=message or ""), attempt
raise AssertionError("unreachable")
def invoke(self, caller: dict, idempotency_key: str, input: Any, dry_run: bool = False) -> Result:
"""Rules 1, 2, 5 and 7 around the handler. Raises Refused for 409 / 422."""
from erp_platform.bricks import Refused
cap = self.capability
if not isinstance(idempotency_key, str) or not idempotency_key.strip():
raise Refused(422, "idempotency_key is required")
try:
data = cap.validate_input(input)
except InvalidCapabilityInput as e:
raise Refused(422, str(e)) from None
call = Call(cap, idempotency_key, data, caller, dry_run)
store = self.store
attempt = uuid.uuid4().hex
def record(result: Optional[Result], *, tries: int = 0, error: Optional[BaseException] = None,
status: str = "error") -> None:
store.write_log(
capability=cap.name, idempotency_key=idempotency_key, attempt_id=attempt, event_at=_utcnow(),
actor=call.actor or None, dry_run=dry_run, input=json.dumps(self.redact(data), sort_keys=True,
default=str),
status=result.status if result else status, retryable=result.retryable if result else None,
provider_ref=result.provider_ref if result else None, message=result.message if result else None,
replay=bool(result and result.idempotent_replay), tries=tries,
error=type(error).__name__ if error else None, request_id=requestlog.request_id() or None)
if dry_run: # rule 2: plan only, no claim
plan = self._plan(call)
result = Result("planned", provider=self.brick.name, plan=plan,
message=f"dry run: {len(plan)} action(s) would be taken; nothing was done")
record(result)
return result
if cap.read: # a read: no claim, nothing stored
result, tries = self._act(call)
record(result, tries=tries)
return result
digest = cap.input_hash(data)
gen = 1
for _ in range(8):
if store.insert_claim(cap.name, idempotency_key, gen, digest, attempt, self.lease_seconds):
break
rows = store.rows(cap.name, idempotency_key) # lost the INSERT: re-read the winner
if not rows:
gen = 1 # released in between: claim afresh
continue
if rows[-1]["input_hash"] != digest:
record(None, status="conflict")
raise Refused(409, f"idempotency_key {idempotency_key!r} was already used for {cap.name} "
"with different input")
finished = [r for r in rows if r["status"]]
if finished:
stored = json.loads(finished[-1]["result"])
result = Result.from_dict({**stored, "idempotent_replay": True})
record(result)
return result
top = rows[-1]
if top["lease_until"] > _utcnow():
result = Result("in_progress", provider=self.brick.name,
message=f"an earlier call with this key is still running (claimed until "
f"{top['lease_until'].isoformat()}Z); try again later")
record(result)
return result
gen = top["generation"] + 1 # expired: take over with the next generation
else:
result = Result("in_progress", provider=self.brick.name, message="contended claim; try again later")
record(result)
return result
mine = {"capability": cap.name, "idempotency_key": idempotency_key, "generation": gen,
"input_hash": digest, "attempt_id": attempt}
tries = 0
try:
result, tries = self._act(call)
except Exception as exc:
# A crash (a provider bug, a Refused): nothing terminal to keep — release so a retry acts.
store.release(cap.name, idempotency_key, gen)
record(None, tries=max(tries, 1), error=exc)
raise
if result.terminal:
claimed = next((r for r in store.rows(cap.name, idempotency_key) if r["generation"] == gen), None)
store.finish(claimed or {**mine, "claimed_at": _utcnow(), "lease_until": _utcnow()}, result)
else:
store.release(cap.name, idempotency_key, gen)
record(result, tries=tries)
return result
def provide(brick: Any, name: str, *, secrets: tuple = (), lease_seconds: float = LEASE_SECONDS,
attempts: int = ATTEMPTS) -> Callable:
"""`@BRICK.capability(name)` — see the module docstring. Returns the Provided object."""
cap = contract().get(name)
if any(getattr(uc.fn, "__capability__", None) == name for uc in brick.use_cases):
raise ValueError(f"{brick.name} already provides {name}")
def register(handler: Callable) -> Provided:
provided = Provided(brick, cap, handler, secrets=tuple(secrets), lease_seconds=lease_seconds,
attempts=attempts)
Input = cap.input_model
Key = typing.Annotated[str, StringConstraints(min_length=1, max_length=500)]
def endpoint(caller: dict, idempotency_key: str, input: Any, dry_run: bool = False) -> dict:
return provided.invoke(caller, idempotency_key, input, dry_run).as_dict()
endpoint.__name__ = endpoint.__qualname__ = cap.tool_name
endpoint.__annotations__ = {"caller": dict, "idempotency_key": Key, "input": Input, "dry_run": bool,
"return": dict}
extra = (f"\n\n{cap.notes}" if cap.notes else "") + (f"\n\n{(handler.__doc__ or '').strip()}"
if handler.__doc__ else "")
endpoint.__doc__ = (
f"{cap.summary} (capability {cap.name}, provided by {brick.name}).{extra}\n\n"
"Body: idempotency_key (derive it from the work, e.g. '<run_id>|<step_no>'; the same key returns "
"the stored result and acts no further), dry_run (true = return the plan, change nothing), input. "
"Answers status done | planned | in_progress | connection_required | failed.")
endpoint.__capability__ = name
_NAMESPACES[brick.name] = getattr(brick, "namespace", brick.name)
brick.use_case(kind="command", route=f"POST {cap.route}", permission=invoke_permission(brick.name),
idempotent=True, destructive=cap.destructive, read_only=cap.read)(endpoint)
provided.endpoint = endpoint
return provided
return register
# ── consumer side ──────────────────────────────────────────────────────────────────────────
_PROVIDERS: dict = {} # tier 1: capability → provider brick name, or None for "manual"
_NAMESPACES: dict = {} # 0.9.8: provider brick name → its permission namespace (Brick(namespace=…))
def invoke_permission(brick_name: str) -> str:
"""The permission a provider's capability door asks for: its spec namespace, e.g. contract:capability:invoke
for the brick named contracts (0.9.8; it was the brick's name)."""
return f"{_NAMESPACES.get(brick_name, brick_name)}:capability:invoke"
def register_provider(capability: str, brick_name: Optional[str]) -> None:
"""Tier 1: the composition says who provides this capability here (None = nobody: manual)."""
contract().get(capability)
_PROVIDERS[capability] = brick_name
def reset_providers() -> None:
_PROVIDERS.clear()
def base_env(brick_name: str) -> str:
"""The base-URL variable of a brick — the same one an ordinary seam to it reads (GWS_BASE_URL)."""
return re.sub(r"[^A-Z0-9]", "_", brick_name.upper()) + "_BASE_URL"
class CapabilitySeam:
"""A consumer's handle on a capability; the provider is whoever the composition wired, or nobody."""
def __init__(self, capability: str, *, timeout: Optional[float] = None):
self.capability = contract().get(capability) # an unknown name fails at import
self.timeout = timeout
@property
def name(self) -> str:
return self.capability.name
def provider(self) -> Optional[str]:
"""Tier 1 registration, then the CAPABILITY_<NAME> variable, else None (no provider: manual)."""
if self.name in _PROVIDERS:
return _PROVIDERS[self.name]
named = os.environ.get(self.capability.env, "").strip()
return None if named in ("", MANUAL) else named
def manual(self, idempotency_key: str, data: dict, reason: str) -> ManualTask:
cap = self.capability
return ManualTask(cap.name, idempotency_key, data, describe(cap, data), reason, cap.destructive,
cap.summary)
def call(self, idempotency_key: str, input: dict, dry_run: bool = False,
actor: str = "") -> Union[Result, ManualTask]:
"""Ask the provider — or, with none, return the ManualTask for a person (rule 8).
Raises InvalidCapabilityInput before anything is sent, and CapabilityRefused when the provider
answers 4xx (the consumer got the call wrong). A provider that is down or answers 5xx is a
`failed`, retryable Result, never an exception: the consumer's own work goes on.
"""
from erp_platform.seam import TIMEOUT, Seam
if not isinstance(idempotency_key, str) or not idempotency_key.strip():
raise InvalidCapabilityInput("idempotency_key is required")
data = self.capability.validate_input(input)
provider = self.provider()
if provider is None:
return self.manual(idempotency_key, data, f"no provider for {self.name} in this composition")
seam = Seam(provider, base_env(provider), roles=["service"],
permissions=[invoke_permission(provider)], timeout=self.timeout or TIMEOUT)
if not seam.installed:
return Result("failed", retryable=True, provider=provider,
message=f"{provider} provides {self.name} here, but {base_env(provider)} is not set "
"and it is not in this process")
r = seam.post(self.capability.route, {"idempotency_key": idempotency_key, "dry_run": dry_run,
"input": data}, actor)
if not r:
if r.status and r.status < 500:
raise CapabilityRefused(r.status, r.reason)
return Result("failed", retryable=True, provider=provider, message=r.reason or f"{provider} failed")
result = Result.from_dict(r.data)
if result.status == "connection_required":
return self.manual(idempotency_key, data, f"{provider}: {result.message or 'connection required'}")
return result