"""The capabilities contracts PROVIDES (capabilities.yaml 0.4: contracts.draft / send / status / fetch_executed) — how
hr (and crm's adapter, outside this brick) ask for a contract without ever writing ct_ tables (CT-50, CT-54).
WHO IS CALLING. The platform's seam does not carry the calling brick: in process the caller is the actor string the
consumer passes (hr: 'hr-system'), across apps its service client. So the consumers are ROWS (ct_consumer: namespace →
principals 'sub:…', 'role:…', 'email:…'), and:
* contracts.draft is refused (403) for a source_ref namespace the caller does not speak for — crm cannot draft
under hr:, so it can never become the originator of hr's contract;
* contracts.status answers any registered consumer, with `executed` only for the ORIGINATOR;
* contracts.fetch_executed answers the originator only — anyone else 404 (SC-CT-24).
MISSING VALUES. capabilities.yaml says a draft missing required variables answers "failed, retryable=false, with
missing[] in the output, and NO partial contract", and SC-CT-13 says a second call WITH the values drafts it. A
terminal `failed` is stored under the key and replayed (rule 1), and the same key with other input is 409 — so the
second call could never succeed. This provider therefore REFUSES (422) with missing[] in the words: nothing is stored,
no partial contract exists, and the corrected call drafts (hr's ACL turns the 422 into a failed step naming them).
"""
from erp_platform.bricks import Refused
from erp_platform.capabilities import Failed
from brick_contracts.application import drafting, queries, signing
from brick_contracts.application import support as S
from brick_contracts.application.use_cases import CT
from brick_contracts.domain.values import DRAFT, IN_APPROVAL, namespace_of
def uow():
from brick_contracts.application import use_cases
return use_cases.repository()
def _namespaces(caller: dict) -> list:
with uow() as u:
cfg = u.config.load()
return [c.namespace for c in cfg.consumer_for(caller)], cfg
@CT.capability("contracts.draft")
def contracts_draft(call):
"""Draft from the type's approved template with the caller's VALUES (never clause text). The counterparty must
hold a role the type allows (else 422). Missing required values: refused 422 naming them, nothing drafted."""
inp, caller = call.input, call.caller
src, type_code = inp["source_ref"].strip(), inp["type_code"].strip().lower()
if call.idempotency_key.strip() != f"{src}|{inp['type_code'].strip()}" and call.idempotency_key.strip() != f"{src}|{type_code}":
raise Refused(422, "idempotency_key must be '<source_ref>|<type_code>' — the contract's id is derived from it")
names, cfg = _namespaces(caller)
ns = namespace_of(src)
if not ns or ns == "manual":
raise Refused(422, "source_ref names its consumer first: 'hr:…', 'crm:…'")
if ns not in names:
raise Refused(403, f"this service may not draft for the {ns}: namespace (ct_consumer)")
consumer = cfg.consumers[ns]
try:
c, _ = drafting.draft(caller, type_code=type_code, bp_id=int(inp["counterparty_bp_id"]),
values=dict(inp.get("variables") or {}), signers=inp.get("signers"), source_ref=src,
context_ref=inp.get("context_ref"), owner="", owner_group=consumer.owner_group,
partial_ok=False)
except Refused as r:
if r.status == 409 and "role" in r.reason:
raise Refused(422, r.reason) from None # the contract: a counterparty of the wrong role is 422
raise
state = c.state if c.state in (DRAFT, IN_APPROVAL) else DRAFT
out = {"contract_id": c.contract_id, "state": state, "missing": list(c.missing),
"deviations": [{"what": d.get("what") or "", "level": str(d.get("level")), "needs": d.get("needs")}
for d in c.deviations],
"approval_required": c.state == IN_APPROVAL}
return c.contract_id, out, ("drafted; approval needed: " + "; ".join(d["what"] for d in out["deviations"]
if d["needs"])) if out["approval_required"] else "drafted"
def _originator_check(names: list):
def check(view, c):
if c is None or not c.originated_by(names):
raise Refused(404, f"no contract {c.contract_id if c else ''} of yours")
return check
@CT.capability("contracts.send")
def contracts_send(call):
"""Freeze the draft as version N and send it for signature. Refused (failed, in words) while values are missing,
an approval is open, the type is a deed, only one side is named, or a UK check fails."""
names, _ = _namespaces(call.caller)
try:
c, _, _ = signing.send(call.caller, call.input["contract_id"], check=_originator_check(names))
except signing._NotSent as e:
raise Failed(str(e)) from None
return c.contract_id, signing.send_answer(c), f"version {c.current_version} out for signature"
@contracts_send.plan
def _(call):
with uow() as u:
c = u.contracts.get(call.input["contract_id"])
n = (max(c.versions, default=0) + 1) if c else 1
return [{"action": f"freeze version {n} (PDF + sha256) and issue one signing link per external signer",
"target": call.input["contract_id"], "payload": {"signers": len(c.signers) if c else 0}}]
@CT.capability("contracts.status")
def contracts_status(call):
"""Where a contract stands; `executed` (object key + sha256) only for its originator; never its body."""
names, _ = _namespaces(call.caller)
if not names:
raise Refused(404, "no such contract")
with uow() as u:
view = S.View(u, call.caller)
c = u.contracts.get(call.input["contract_id"])
if c is None:
raise Refused(404, "no such contract")
out = queries.status_of(c, view, executed=c.originated_by(names))
return c.contract_id, out, out["state"].lower()
@CT.capability("contracts.fetch_executed")
def contracts_fetch_executed(call):
"""The executed copy's bytes as base64 with their sha256 — to the originator only (anyone else 404)."""
names, _ = _namespaces(call.caller)
with uow() as u:
c = u.contracts.get(call.input["contract_id"])
if c is None or not c.originated_by(names):
raise Refused(404, "no such contract")
if c.executed is None:
raise Failed("not signed yet: there is no executed copy")
out = queries.executed_bytes(u, c)
return c.contract_id, out, "fetched"
__all__ = ["contracts_draft", "contracts_send", "contracts_status", "contracts_fetch_executed"]