"""Drafting and approval: a contract from its type's approved template (CT-05), positions and overrides (CT-06),
values and signers, and the approval of deviations bound to the draft's fingerprint (CT-07).
Every edit re-runs the same review: the fingerprint is recomputed, approvals bound to another fingerprint are void,
an approval opens per group the deviations need, and the state is DRAFT, IN_APPROVAL or APPROVED accordingly.
"""
import json
from datetime import datetime
from typing import Annotated, Literal, Optional, Union
from pydantic import BaseModel, Field
from erp_platform.bricks import Refused
from erp_platform.types import BigId, Email, Name, text
from brick_contracts.application import support as S
from brick_contracts.application.use_cases import CT
from brick_contracts.domain.contract import Contract, Section
from brick_contracts.domain.errors import Invalid, RuleBroken
from brick_contracts.domain.factories import ContractFactory, request_hash
from brick_contracts.domain.library import slots_for
from brick_contracts.domain.services import DeviationCheck, Resolver, unfilled
from brick_contracts.domain.values import THEM, US, VariableValue, contract_id_for
from brick_contracts.infrastructure import tables as T
Scalar = Optional[Union[text(4000), int, bool]]
Values = Annotated[dict[str, Scalar], Field(max_length=300)]
class SignerIn(BaseModel):
side: Literal["US", "THEM"]
full_name: Name
email: Email = ""
title: Optional[Name] = None
authority_basis: Optional[Name] = None
signing_order: S.Order = 1
pool_sub: Optional[Name] = None
def uow():
from brick_contracts.application import use_cases
return use_cases.repository()
# ── the one drafting path (people and the contracts.draft capability) ──────────────────────
def draft(caller: dict, *, type_code: str, bp_id: int, values: dict, signers: Optional[list], source_ref: str,
context_ref: Optional[str], owner: str, owner_group: str, partial_ok: bool,
template_id: Optional[str] = None) -> tuple:
"""(Contract, replayed). contract_id derives from source_ref|type_code: the same request again is the same
contract (a replay); the same key with different input is 409; with partial_ok=False a draft missing required
values is refused 422 with missing[] and NO partial contract is written."""
at = S.now()
cid = contract_id_for(source_ref, type_code)
rh = request_hash(type_code.lower(), int(bp_id), context_ref, values, signers)
with uow() as u:
head = u.contracts.head(cid)
existing = u.contracts.get(cid) if head is not None and head.request_hash == rh else head
if existing is not None:
if existing.request_hash != rh:
raise Refused(409, f"{source_ref} already drafted {cid} as a {existing.type_code.replace('_', ' ')} "
"with different input — use a new source_ref, or edit that draft")
return existing, True
partner = S.partner(int(bp_id), S.actor(caller)) # read through the bp seam, outside our transaction
with S.rules(), uow() as u:
cfg = u.config.load()
ctx = S.drafting(u, cfg, type_code, partner, template_id)
c = ContractFactory.draft(ctx, source_ref=source_ref, values=values, signers=signers, owner=owner,
owner_group=owner_group, context_ref=context_ref, now=at, by=S.actor(caller),
contract_id=cid)
if c.missing and not partial_ok:
raise Refused(422, "missing: " + ", ".join(c.missing) + " — nothing was drafted; send them and ask again")
S.settle(c, cfg, at)
u.contracts.save(c, now=at)
return c, False
def rebuild(c: Contract, ctx, supplied: dict, *, at: datetime, by: str) -> None:
"""After a value changes: resolve again (COUNTERPARTY values stay as held), re-choose the slots a condition
decides — keeping the owner's position choices and overrides on slots still chosen — and review."""
values = {n: v.value for n, v in c.variables.items() if v.source != "COUNTERPARTY"}
for n, v in c.variables.items():
if v.currency:
values.setdefault(ctx.version.variables[n].currency_var if n in ctx.version.variables else "currency",
v.currency)
values.update({str(k).lower(): v for k, v in supplied.items()})
values = {k: v for k, v in values.items() if v is not None and v != ""}
res = Resolver(ctx.version, ctx.partner, values, at=at,
extra={"our_name": ctx.config.setting("our_legal_name", "")}).resolve()
variables = dict(res.variables)
for n, v in c.variables.items():
if v.source == "COUNTERPARTY" or (n == "our_name" and n not in variables):
variables[n] = v
elif n in variables and variables[n].canonical_row() == v.canonical_row() and variables[n].source == v.source:
variables[n] = v # unchanged: keeps when it was resolved
cond = {k: v.value for k, v in variables.items()} | {k: str(v) for k, v in values.items()}
held = {s.slot_no: s for s in c.sections}
sections = []
for i, slot in enumerate(slots_for(ctx.version, cond), start=1):
if slot.slot_no in held and held[slot.slot_no].family == slot.family and (
held[slot.slot_no].clause_id == slot.clause_id or held[slot.slot_no].position_kind != "PREFERRED"
or held[slot.slot_no].overridden):
s = held[slot.slot_no]
s.position = i
sections.append(s)
continue
words = ctx.clause_words(slot.clause_id, slot.clause_version)
title, body, position, level = words
sections.append(Section(i, slot.family, slot.clause_id, slot.clause_version, position, int(level), title, body,
slot.slot_no, slot.kind == "required"))
c.variables, c.sections = variables, sections
from brick_contracts.domain.factories import term_of
c.starts_on, c.ends_on, c.notice_days = term_of(ctx.version, variables)
review(c, ctx, at=at, by=by, extra_missing=res.missing)
def review(c: Contract, ctx, *, at: datetime, by: str, extra_missing: Optional[list] = None) -> None:
missing = set(extra_missing if extra_missing is not None else [m for m in c.missing
if m not in c.variables])
missing |= set(unfilled(c.sections, c.variables))
if ctx.pack is not None:
missing |= set(ctx.pack.draft_missing(ctx.ctype, c.variables, ctx.config))
missing = {m for m in missing if m not in c.variables}
c.review(sorted(missing), DeviationCheck(ctx.policy, ctx.version).check(c.sections, c.variables), now=at, by=by)
if ctx.pack is not None:
ctx.pack.draft_key_dates(ctx.ctype, c, ctx.config)
def _edit(caller: dict, contract_id: str, expected_version: Optional[int], change) -> dict:
"""Load (404 unless the caller manages it), change, review, save — one unit of work."""
at = S.now()
with S.rules(), uow() as u:
view = S.View(u, caller, at)
c = view.contract(contract_id, manage=True)
S.at_version(c, expected_version)
c.require_editable()
ctx = S.drafting_for(u, view.config, c)
before = _rows(c)
change(c, ctx, at, view)
S.settle(c, view.config, at)
if _rows(c) != before or c.new_events:
u.contracts.save(c, now=at) # nothing changed: nothing written (a replay is a no-op)
return S.full(c, view)
def _rows(c) -> dict:
from brick_contracts.infrastructure.repositories import _norm, contract_rows
return {k: _norm(v) for k, v in contract_rows(c).items()}
# ── people ─────────────────────────────────────────────────────────────────────────────────
@CT.use_case(kind="command", route="POST /contracts", permission=S.WRITE, idempotent=True)
def draft_contract(caller: dict, type_code: S.Code, bp_id: BigId, form_key: S.Code, values: Optional[Values] = None,
signers: Optional[Annotated[list[SignerIn], Field(max_length=20)]] = None,
context_ref: Optional[S.Short] = None, template_id: Optional[S.Code] = None) -> dict:
"""Draft a contract for a business partner by choosing its type (CT-05): the type's approved template version is
copied into sections, the variables are filled from the partner (read through business-partner) and from what you
typed, and the readiness panel lists missing[] and deviations[]. form_key is generated when the form opened, so a
double click is ONE draft (source_ref manual:<form_key>). Without signers, our default signatory for the type and
the partner (when it is a person with an address) are named."""
sig = [s.model_dump() for s in signers] if signers else None
c, _ = draft(caller, type_code=type_code, bp_id=int(bp_id), values=dict(values or {}), signers=sig,
source_ref=f"manual:{form_key}", context_ref=context_ref, owner=S.actor(caller), owner_group="",
partial_ok=True, template_id=template_id)
with uow() as u:
view = S.View(u, caller)
return S.full(c, view) if view.body(c) else S.summary(c, view.today)
@CT.use_case(kind="command", route="POST /contracts/{contract_id}/edit", permission=S.WRITE, idempotent=True,
if_match="expected_version")
@S.retry_unless_versioned
def edit_draft(caller: dict, contract_id: S.ContractId, title: Optional[S.Short] = None,
reference: Optional[S.Short] = None, expected_version: Optional[int] = None) -> dict:
"""Change a draft's title (part of what is signed, so it changes the fingerprint) or its internal reference.
Refused once it is out for signature ("recall it to edit") or signed (amend it)."""
def change(c, ctx, at, view):
if title is not None and title.strip():
c.title = title.strip()
if reference is not None:
c.reference = reference.strip()
review(c, ctx, at=at, by=S.actor(caller))
return _edit(caller, contract_id, expected_version, change)
@CT.use_case(kind="command", route="POST /contracts/{contract_id}/variables", permission=S.WRITE, idempotent=True,
if_match="expected_version")
@S.retry_unless_versioned
def set_variables(caller: dict, contract_id: S.ContractId, values: Values, expected_version: Optional[int] = None) -> dict:
"""Set a draft's values (typed: dates, exact amounts with their currency, enums). A condition may choose another
clause (engagement_via = PSC → the PSC warranty). Any change changes the fingerprint, so an approval already given
is void and the draft needs approval again if it still deviates (SC-CT-02). Counterparty values come from the
business partner and are not typed here."""
def change(c, ctx, at, view):
for k in values:
d = ctx.version.variables.get(str(k).lower())
if d is not None and d.source == "COUNTERPARTY":
raise RuleBroken(f"{k} comes from the business partner; change it there, it is read again at send")
rebuild(c, ctx, dict(values), at=at, by=S.actor(caller))
return _edit(caller, contract_id, expected_version, change)
@CT.use_case(kind="command", route="POST /contracts/{contract_id}/sections/{position}/position", permission=S.WRITE,
idempotent=True, if_match="expected_version")
@S.retry_unless_versioned
def choose_position(caller: dict, contract_id: S.ContractId, position: S.SeqNo, clause_id: S.Code,
clause_version: Optional[S.VersionNo] = None, expected_version: Optional[int] = None) -> dict:
"""Switch one section to another clause of the same family — the Preferred, an Alternate or the Fallback (CT-06).
The section takes a copy of that clause's words; the library is untouched. A Fallback (or a deviating Alternate)
makes the draft need approval, and the send says so."""
def change(c, ctx, at, view):
cl = view.uow.clauses.get(clause_id.lower())
if cl is None:
raise Refused(404, f"no clause {clause_id}")
c.choose_position(position, cl, clause_version)
review(c, ctx, at=at, by=S.actor(caller))
return _edit(caller, contract_id, expected_version, change)
@CT.use_case(kind="command", route="POST /contracts/{contract_id}/sections/{position}/override", permission=S.WRITE,
idempotent=True, if_match="expected_version")
@S.retry_unless_versioned
def override_section(caller: dict, contract_id: S.ContractId, position: S.SeqNo, body: S.Words,
title: Optional[S.Short] = None, expected_version: Optional[int] = None) -> dict:
"""Rewrite one section's words for this contract only (CT-06): it is marked overridden at deviation level 2
(non-standard) and needs approval under the type's policy. The library is untouched."""
def change(c, ctx, at, view):
c.override_section(position, title, body)
review(c, ctx, at=at, by=S.actor(caller))
return _edit(caller, contract_id, expected_version, change)
@CT.use_case(kind="command", route="POST /contracts/{contract_id}/signers", permission=S.WRITE, idempotent=True,
if_match="expected_version")
@S.retry_unless_versioned
def add_signer(caller: dict, contract_id: S.ContractId, side: Literal["US", "THEM"], full_name: Name, email: Email = "",
title: Optional[Name] = None, authority_basis: Optional[Name] = None, signing_order: S.Order = 1,
pool_sub: Optional[Name] = None, expected_version: Optional[int] = None) -> dict:
"""Name a signer: side US is our signatory (INTERNAL — signs inside the app as their own sign-in, named by that
email), side THEM the counterparty's (EXTERNAL — signs through a public link sent to this email). Signing order 1
signs first. Both sides must be named before it can be sent. Adding the same email twice is one signer."""
def change(c, ctx, at, view):
if any(s.email == (email or "").strip().lower() and email for s in c.signers):
return
c.add_signer(kind="INTERNAL" if side == US else "EXTERNAL", side=side, signing_order=signing_order,
email=email, full_name=full_name, title=title or "", authority_basis=authority_basis or "",
bp_id=c.bp_id if side == THEM else None, pool_sub=pool_sub)
review(c, ctx, at=at, by=S.actor(caller))
return _edit(caller, contract_id, expected_version, change)
@CT.use_case(kind="command", route="POST /contracts/{contract_id}/signers/{signer_no}", permission=S.WRITE,
idempotent=True, if_match="expected_version")
@S.retry_unless_versioned
def edit_signer(caller: dict, contract_id: S.ContractId, signer_no: S.SignerNo, full_name: Optional[Name] = None,
email: Optional[Email] = None, title: Optional[Name] = None, authority_basis: Optional[Name] = None,
signing_order: Optional[S.Order] = None, remove: bool = False,
expected_version: Optional[int] = None) -> dict:
"""Correct a signer (a mistyped address is fixed without a new key: signers are numbered, not keyed by email),
change the signing order, or remove them (remove = true)."""
def change(c, ctx, at, view):
if remove:
c.remove_signer(signer_no)
else:
c.edit_signer(signer_no, full_name=full_name, email=email, title=title, authority_basis=authority_basis,
signing_order=signing_order)
review(c, ctx, at=at, by=S.actor(caller))
return _edit(caller, contract_id, expected_version, change)
@CT.use_case(kind="command", route="POST /contracts/{contract_id}/approvals", permission=S.WRITE, idempotent=True)
@S.retry_unless_versioned
def request_approval(caller: dict, contract_id: S.ContractId, expected_version: Optional[int] = None) -> dict:
"""Ask for approval again after a rejection, for the draft as it is now (the same fingerprint). Approvals open by
themselves whenever an edit leaves the draft off the standard."""
def change(c, ctx, at, view):
if not c.open_approvals() and not c.approval_granted():
c.request_approval(now=at, by=S.actor(caller))
return _edit(caller, contract_id, expected_version, change)
@CT.use_case(kind="command", route="POST /contracts/{contract_id}/approvals/{seq}/decide", permission=S.DECIDE)
def decide_approval(caller: dict, contract_id: S.ContractId, seq: S.SeqNo, decision: Literal["approve", "reject"],
reason: Optional[S.Reason] = None) -> dict:
"""Approve or reject a deviating draft from your queue (CT-07). The decision binds to the fingerprint the approval
was asked for: a later edit voids it. A rejection says why and returns the draft to DRAFT. You cannot approve a
draft you own; after its timeout an approval belongs to the policy's fallback group."""
from erp_platform.repository import retry_on_conflict
def go():
at = S.now()
with S.rules(), uow() as u:
view = S.View(u, caller, at)
c = u.contracts.get(contract_id)
if c is None or not (contract_id in view.approving or view.sees(c)):
raise S.not_found(f"contract {contract_id}")
a = c.decide(seq, decision == "approve", by=S.actor(caller), groups=view.groups, now=at, reason=reason or "")
S.settle(c, view.config, at)
u.contracts.save(c, now=at)
return {"contract_id": c.contract_id, "seq": a.seq, "state": a.state, "contract_state": c.state,
"draft_fingerprint": a.draft_fingerprint, "decided_by": a.decided_by, "reason": a.reason}
return retry_on_conflict(go)
@CT.use_case(kind="query", route="GET /approvals", permission=S.DECIDE)
def approvals_queue(caller: dict) -> dict:
"""The drafts waiting on your groups (an escalated one on the fallback group): each with the deviations against
the standard and the fingerprint the approval binds to. No contract body is shown here."""
with uow() as u:
view = S.View(u, caller)
rows = u.contracts.approvals_for(view.groups)
ids = sorted({r["contract_id"] for r in rows})
heads = {h["contract_id"]: h for h in u.contracts.headers(where=T.contract.c.contract_id.in_(ids))} if ids else {}
items = []
for r in sorted(rows, key=lambda r: (S.iso(r["due_at"]) or "", r["contract_id"], r["seq"])):
h = heads.get(r["contract_id"])
if h is None or h["state"] not in ("IN_APPROVAL",):
continue
items.append({"contract_id": r["contract_id"], "seq": r["seq"], "title": h["title"], "type_code": h["type_code"],
"owner": h["owner"], "state": r["state"], "group": r["fallback_group"] if r["state"] == "escalated"
else r["approver_group"], "due_at": S.iso(r["due_at"]),
"draft_fingerprint": r["draft_fingerprint"],
"deviations": json.loads(r["deviations_json"] or "[]")})
return {"approvals": items}
__all__ = ["draft", "rebuild", "review", "draft_contract", "edit_draft", "set_variables", "choose_position",
"override_section", "add_signer", "edit_signer", "request_approval", "decide_approval", "approvals_queue"]