"""Domain services — rules that span an aggregate and its inputs, still plain Python.
Resolver (template version, counterparty record, supplied values) → variables + missing[]
DeviationCheck (draft, template version, approval policy) → deviations[] with levels and who must approve
Composer (frozen sections + variables) → the contract's words (what the PDF and the signing page show)
ChainVerifier (events) → valid | broken at seq n
CertificateBuilder(signed version, events, signers) → the certificate page, saying only "signed electronically"
next_chase when a signer whose turn it is is chased next (erp_platform.work's cadence)
"""
import re
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from decimal import Decimal
from typing import Optional
from brick_contracts.domain.errors import Invalid
from brick_contracts.domain.library import TemplateVersion, VariableDef, holds
from brick_contracts.domain.policies import ApprovalPolicy, Deviation
from brick_contracts.domain.values import EventHash, VariableValue, utc_iso
# The ONLY claim this module makes about a signature (modules/contracts in_code_not_rows). Never "advanced",
# "qualified", "legally binding" or anything "equivalent to a wet signature".
CLAIM = ("Signed electronically. A record of who signed, when, from where, and the fingerprint of the document is "
"attached.")
FORBIDDEN_CLAIMS = ("advanced", "qualified", "legally binding", "wet signature", "wet-ink")
PLACEHOLDER = re.compile(r"\{\{\s*([a-z][a-z0-9_]*)\s*\}\}")
# What a COUNTERPARTY variable reads from the business partner when its definition names no bp_field.
BP_FIELDS = {"counterparty_name": "legal_name", "employee_name": "legal_name", "supplier_name": "legal_name",
"counterparty_country": "country", "counterparty_tax_id": "tax_id", "counterparty_email": "work_email"}
def label(code: str) -> str:
return (code or "").replace("_", " ").replace("-", " ")
@dataclass
class Resolution:
variables: dict # name → VariableValue
missing: list
ignored: list # supplied names the template does not define
class Resolver:
"""Fill a template version's variables: COUNTERPARTY ones from the partner (never re-typed: a supplied value for
them is ignored), the rest from what the caller supplied, then `default_from` another variable, then the
template's standard (default_value). A required definition with no value is MISSING; a definition whose
condition does not hold does not apply."""
def __init__(self, version: TemplateVersion, partner: dict, supplied: dict, *, at: datetime,
extra: Optional[dict] = None):
self.version, self.partner, self.at = version, partner or {}, at
self.supplied = {str(k).strip().lower(): v for k, v in (supplied or {}).items()}
self.extra = extra or {}
def _raw(self, d: VariableDef, resolved: dict):
if d.source == "COUNTERPARTY":
f = d.validation.get("bp_field") or BP_FIELDS.get(d.name) or d.name
v = self.partner.get(f)
return (v, "COUNTERPARTY") if v not in (None, "") else (None, None)
v = self.supplied.get(d.name)
if v not in (None, ""):
return v, d.source
src = d.validation.get("default_from")
if src and src in resolved:
return resolved[src].value, "TEMPLATE"
if d.name in self.extra and self.extra[d.name] not in (None, ""):
return self.extra[d.name], "TEMPLATE"
if d.default_value is not None:
return d.default_value, "TEMPLATE"
return None, None
def resolve(self) -> Resolution:
resolved: dict = {}
when = utc_iso(self.at)
pending = sorted(self.version.variables.values(), key=lambda d: d.name)
# Repeat until nothing more resolves: a definition can take its default from another (default_from), or apply
# only when a condition over what is resolved holds — so the order rows come back in never matters.
progress = True
while progress:
progress = False
for d in list(pending):
values = {k: v.value for k, v in resolved.items()} | {k: str(v) for k, v in self.supplied.items()
if v is not None}
if d.condition is not None and not holds(d.condition, values):
continue
raw, source = self._raw(d, resolved)
if raw is None:
continue
cur = None
if d.type in ("money", "rate"):
cur = (self.supplied.get(d.currency_var) or (resolved[d.currency_var].value
if d.currency_var in resolved else None)
or d.validation.get("currency"))
if cur is None and d.currency_var in self.version.variables and d.currency_var not in resolved:
continue # its currency is still to come
resolved[d.name] = VariableValue.parse(d.name, d.type, raw, source, enum_values=d.enum_values,
aliases=d.aliases, currency=cur, resolved_at=when)
pending.remove(d)
progress = True
values = {k: v.value for k, v in resolved.items()} | {k: str(v) for k, v in self.supplied.items() if v is not None}
missing = [d.name for d in pending if d.required and (d.condition is None or holds(d.condition, values))]
for d in pending:
if d.type in ("money", "rate") and d.name in self.supplied and d.name not in resolved:
VariableValue.parse(d.name, d.type, self.supplied[d.name], d.source) # say why: no currency
defined = set(self.version.variables)
ignored = sorted(k for k in self.supplied if k not in defined and k not in self.extra)
return Resolution(resolved, sorted(set(missing)), ignored)
def placeholders(text: str) -> set:
return set(PLACEHOLDER.findall(text or ""))
def unfilled(sections: list, variables: dict) -> list:
"""Variables the chosen words use and nothing has given a value: they are missing too."""
used = set()
for s in sections:
used |= placeholders(s.body) | placeholders(s.title)
return sorted(n for n in used if n not in variables)
class DeviationCheck:
"""CT-06 / CT-07: a Fallback (or an Alternate above level 0), an override, or a variable outside its bounds is a
deviation; the type's policy says which group approves it and the mandatory fallback group."""
def __init__(self, policy: ApprovalPolicy, version: Optional[TemplateVersion]):
self.policy, self.version = policy, version
def _route(self, kind: str, subject: str, position: str = "") -> tuple:
r = self.policy.rule_for(kind, subject, position)
return (r.approver_group, r.fallback_group, r.timeout_working_days) if r else (None, None, 2)
def check(self, sections: list, variables: dict) -> list:
out = []
for s in sorted(sections, key=lambda s: s.position):
if s.deviation_level <= 0:
continue
kind = "override" if s.overridden else "position"
needs, fb, timeout = self._route(kind, s.family, s.position_kind)
what = (f"{label(s.family)} is rewritten for this contract" if s.overridden
else f"{label(s.family)} is a{'n' if s.position_kind == 'ALTERNATE' else ''} "
f"{s.position_kind.title()}")
out.append(Deviation(kind, s.family, what, s.deviation_level, needs, fb, timeout))
defs = self.version.variables if self.version else {}
for name, v in sorted(variables.items()):
d = defs.get(name)
if d is None or not ("min" in d.validation or "max" in d.validation):
continue
amt = v.amount
if amt is None:
continue
band_cur = d.validation.get("currency")
if band_cur and v.currency and band_cur.upper() != v.currency:
needs, fb, timeout = self._route("bound", name)
out.append(Deviation("bound", name, f"{d.label or label(name)} is in {v.currency}, its band is in "
f"{band_cur.upper()}", 1, needs, fb, timeout))
continue
lo = Decimal(str(d.validation["min"])) if "min" in d.validation else None
hi = Decimal(str(d.validation["max"])) if "max" in d.validation else None
band = f"{lo if lo is not None else '…'}–{hi if hi is not None else '…'}{' ' + band_cur.upper() if band_cur else ''}"
if hi is not None and amt > hi:
needs, fb, timeout = self._route("bound", name)
out.append(Deviation("bound", name, f"{d.label or label(name)} {v.value} is above its band ({band})",
1, needs, fb, timeout))
elif lo is not None and amt < lo:
needs, fb, timeout = self._route("bound", name)
out.append(Deviation("bound", name, f"{d.label or label(name)} {v.value} is below its band ({band})",
1, needs, fb, timeout))
return out
class Composer:
"""The contract's words from its frozen parts: the same text is rendered to the PDF and shown on the signing
page, so the signer reads exactly what is hashed."""
@staticmethod
def fill(text: str, variables: dict) -> str:
return PLACEHOLDER.sub(lambda m: variables[m.group(1)].display() if m.group(1) in variables
else f"[{m.group(1)}]", text or "")
@classmethod
def compose(cls, *, title: str, our_name: str, counterparty_name: str, version: int, contract_id: str,
sections: list, variables: dict, signers: list) -> str:
lines = [title, "", f"Between {our_name} (\"we\") and {counterparty_name} (\"you\").",
f"Contract {contract_id}, version {version}.", ""]
for s in sorted(sections, key=lambda s: s.position):
lines.append(f"{s.position}. {cls.fill(s.title, variables)}")
lines.append(cls.fill(s.body, variables))
lines.append("")
lines.append("Signatures")
for s in sorted(signers, key=lambda s: (s.signing_order, s.signer_no)):
side = "for us" if s.side == "US" else "for you"
role = f", {s.title}" if s.title else ""
lines.append(f" {s.full_name}{role} — {side} (signing order {s.signing_order})")
return "\n".join(lines).strip() + "\n"
class ChainVerifier:
"""CT-32: recompute every row_hash; name the first seq that does not match (an edited IP, a removed row)."""
@staticmethod
def verify(contract_id: str, events: list) -> tuple:
prev = EventHash.genesis(contract_id)
for i, e in enumerate(sorted(events, key=lambda e: e.seq), start=1):
if e.seq != i:
return False, i # a row removed (or inserted) before this one
if e.prev_hash != prev:
return False, e.seq
want = EventHash.chain(prev, e.canonical(contract_id)).row_hash
if want != e.row_hash:
return False, e.seq
prev = e.row_hash
return True, None
class CertificateBuilder:
"""CT-13: every signer and every event with time, IP and user agent, the version's sha256, whether the chain
verifies — and the one claim. The executed file's own sha256 cannot be printed on itself: it is in the row."""
@staticmethod
def lines(*, contract_id: str, title: str, version, signers: list, events: list, chain: tuple,
retention_note: str = "") -> list:
ok, broken = chain
out = ["Certificate of signature", "", CLAIM, "",
f"Contract: {title} ({contract_id})",
f"Signed version: {version.version} sha256 {version.sha256}",
f"Evidence chain: {'verifies' if ok else f'BROKEN at event {broken}'}", "", "Signers"]
for s in sorted(signers, key=lambda s: (s.signing_order, s.signer_no)):
out.append(f" {s.signer_no}. {s.full_name} <{s.email}> — side {s.side}, {s.kind.lower()}"
+ (f", {s.title}" if s.title else "") + (f", authority: {s.authority_basis}" if s.authority_basis else ""))
out += ["", "Events (UTC)"]
for e in sorted(events, key=lambda e: e.seq):
who = f"signer {e.signer_no}" if e.signer_no else "—"
extra = []
if e.auth_method:
extra.append(e.auth_method)
if e.pool_sub:
extra.append(f"pool sub {e.pool_sub}")
if e.ip:
extra.append(f"ip {e.ip}")
if e.user_agent:
extra.append(f"agent {e.user_agent[:80]}")
if e.typed_name:
extra.append(f"typed name \"{e.typed_name}\"")
if e.consent_text_version:
extra.append(f"consent text {e.consent_text_version}")
if e.version:
extra.append(f"version {e.version}")
out.append(f" {e.seq:>3} {utc_iso(e.occurred_at)} {e.kind:<17} {who} " + "; ".join(extra))
if retention_note:
out += ["", retention_note]
return out
def next_chase(cadence, turn_started_at: datetime, chase_count: int, last_chased_at: Optional[datetime], calendar=None):
"""When the signer whose turn it is should next be chased (None: chased enough — tell the owner instead)."""
from erp_platform import work
wc = work.ChaseCadence(cadence.every_days, cadence.max_count, cadence.first_after_days, "UTC",
cadence.unit if cadence.unit in work.UNITS else "days")
history = [last_chased_at or turn_started_at] * int(chase_count)
return work.next_chase(wc, turn_started_at, history, calendar or work.WorkCalendar())
def retention_statement(years: int, deed: bool) -> str:
return (f"The signed copy and this record are kept for {years} years after the contract ends "
f"({'a deed' if deed else 'a simple contract'}), then deleted.")
__all__ = ["CLAIM", "FORBIDDEN_CLAIMS", "Resolver", "Resolution", "DeviationCheck", "Composer", "ChainVerifier",
"CertificateBuilder", "next_chase", "unfilled", "placeholders", "label", "retention_statement"]