"""What a commit causes, each in its own unit of work and each idempotent on its own key:
deliver a pending delivery through mail.send (key contract_id|signer_no|kind|seq), or — with no provider —
a manual task for the owner with the exact recipient and message (the link is never in it: "copy the
signing link" reissues it and shows it once)
finalize the executed copy the moment the last signer signs (CT-31): the chain verified, the signed version's
bytes checked against their sha256, the certificate page appended, SIGNED, key dates, a READ link per
external signer for their copy; a broken chain refuses it and tells the owner; a retry makes it once
supersede a signed AMENDMENT / RENEWAL / TERMINATION supersedes or terminates its original (CT-14)
"""
import logging
from datetime import timedelta
from typing import Optional
from erp_platform.repository import retry_on_conflict
from brick_contracts.application import support as S
from brick_contracts.domain.contract import Delivery, ExecutedCopy
from brick_contracts.domain.errors import RuleBroken
from brick_contracts.domain.services import CertificateBuilder, ChainVerifier
from brick_contracts.domain.values import EXTERNAL, INTERNAL, SIGNED, SUPERSEDED, TERMINATED, sha256_hex
from brick_contracts.infrastructure import mail, public_door, renderer
log = logging.getLogger("brick_contracts.workflow")
SYSTEM = "contracts"
def uow():
from brick_contracts.application import use_cases
return use_cases.repository()
# ── messages ───────────────────────────────────────────────────────────────────────────────
LINK_PLACEHOLDER = "<the signing link — use 'copy the signing link' in Contracts>"
COPY_PLACEHOLDER = "<the signed-copy link — use 'copy the signed-copy link' in Contracts>"
def our_name(cfg) -> str:
return cfg.setting("our_legal_name", "") or "We"
def invite(c, s, cfg, url: Optional[str]) -> tuple:
subject = f"Please sign: {c.title}"
body = (f"Hello {s.full_name},\n\n{our_name(cfg)} has sent you \"{c.title}\" (version {c.current_version}) to sign "
f"electronically.\n\nOpen it here: {url or LINK_PLACEHOLDER}\n\nThe link works until "
f"{c.link_expires_at.date().isoformat() if c.link_expires_at else 'it is used'} and only for you. You can "
"read the document, sign it, decline it or ask for changes. This is a simple electronic signature: a record "
"of who signed, when, from where, and the fingerprint of the document is kept with it.\n")
return subject, body
def chase(c, s, cfg, url: Optional[str]) -> tuple:
subject = f"Reminder: {c.title} is waiting for your signature"
body = (f"Hello {s.full_name},\n\n\"{c.title}\" is still waiting for your signature.\n\nOpen it here: "
f"{url or LINK_PLACEHOLDER}\n")
return subject, body
def copy_message(c, s, cfg, url: Optional[str]) -> tuple:
subject = f"Signed: {c.title}"
if s.kind == INTERNAL:
return subject, (f"Hello {s.full_name},\n\n\"{c.title}\" is signed by everyone. The executed copy (sha256 "
f"{c.executed.sha256}) is on the record in Contracts.\n")
return subject, (f"Hello {s.full_name},\n\n\"{c.title}\" is signed by everyone. Download your signed copy here: "
f"{url or COPY_PLACEHOLDER}\n\nIts sha256 fingerprint is {c.executed.sha256}. "
f"{S.retention_note(cfg, c)}\n")
def new_delivery(c, signer_no: int, kind: str, recipient: str, subject: str, message: str, at) -> Delivery:
channel = "provider" if mail.provider() else "manual"
d = Delivery(signer_no, kind, 0, channel, recipient, subject, message, at,
reason="" if channel == "provider" else "no provider for mail.send in this composition")
c.add_delivery(d)
d.idempotency_key = f"{c.contract_id}|{signer_no}|{kind}|{d.seq}"
return d
# ── deliver ────────────────────────────────────────────────────────────────────────────────
def deliver(contract_id: str, items: list) -> list:
"""Send each (signer_no, kind, seq, url) that is a provider delivery; record what happened. Outside any open
transaction: the provider call holds none."""
out = []
if not items or mail.provider() is None: # no provider: every delivery is already a manual task
return out
with uow() as u:
c = u.contracts.get(contract_id)
cfg = u.config.load()
if c is None:
return out
results = []
for signer_no, kind, seq, url in items:
d = next((x for x in c.deliveries if (x.signer_no, x.kind, x.seq) == (signer_no, kind, seq)), None)
if d is None or d.channel != "provider" or d.status != "pending":
continue
body = d.message.replace(LINK_PLACEHOLDER, url or LINK_PLACEHOLDER).replace(COPY_PLACEHOLDER,
url or COPY_PLACEHOLDER)
r = mail.send(d.idempotency_key, from_email=cfg.setting("mail_from", ""), to=d.recipient, subject=d.subject,
body=body, actor=SYSTEM)
results.append((signer_no, kind, seq, r))
out.append({"signer_no": signer_no, "kind": kind, "status": r.status})
def record():
at = S.now()
with uow() as u:
c2 = u.contracts.get(contract_id)
cfg2 = u.config.load()
for signer_no, kind, seq, r in results:
d = next((x for x in c2.deliveries if (x.signer_no, x.kind, x.seq) == (signer_no, kind, seq)), None)
if d is None or d.status != "pending":
continue
if r.status == "sent":
d.status, d.provider_ref, d.done_at, d.done_by = "sent", r.provider_ref, at, "mail.send"
if signer_no and kind in ("INVITE", "REISSUE") or kind.startswith("CHASE"):
c2.mark_delivered(signer_no, now=at, by=SYSTEM, channel="mail.send", provider_ref=r.provider_ref)
elif r.status == "manual":
d.channel, d.reason, d.plan = "manual", r.message[:500], r.plan[:1000]
elif r.status == "retry":
d.retry_at = at + timedelta(minutes=15)
else:
d.channel, d.reason = "manual", f"mail.send failed: {r.message[:300]} — send it by hand"
S.settle(c2, cfg2, at)
u.contracts.save(c2, now=at)
if results:
retry_on_conflict(record)
return out
# ── finalize: the executed copy ────────────────────────────────────────────────────────────
def finalize(contract_id: str) -> Optional[dict]:
"""CT-31. Returns what happened (or None when there is nothing to do). Idempotent: one executed copy."""
urls: dict = {}
def go():
urls.clear()
at = S.now()
with uow() as u:
c = u.contracts.get(contract_id)
if c is None or c.executed is not None or not c.all_signed:
return None
cfg = u.config.load()
ok, broken = ChainVerifier.verify(c.contract_id, c.events)
ver = c.current
stored = u.objects.get(ver.object_key)
bytes_ok = stored is not None and sha256_hex(stored[0]) == ver.sha256
if not ok or not bytes_ok:
why = (f"the evidence chain is broken at event {broken}" if not ok else
f"version {ver.version}'s stored bytes no longer match its sha256")
if not c.last_note.startswith("REFUSED"):
c.last_note = f"REFUSED: {why} — the executed copy is not produced"
new_delivery(c, 0, "NOTICE:tamper", c.owner or c.owner_group, f"Evidence problem: {c.title}",
f"{why}. The executed copy was not produced. Check the contract's evidence.", at)
c.next_due_at = None
u.contracts.save(c, now=at)
return {"executed": False, "reason": why}
lines = CertificateBuilder.lines(contract_id=c.contract_id, title=c.title, version=ver, signers=c.signers,
events=c.events, chain=(ok, broken), retention_note=S.retention_note(cfg, c))
executed = renderer.append(stored[0], "\n".join(lines), footer=f"{c.contract_id} — certificate")
key = f"contracts/{c.contract_id}/executed.pdf"
sha, size = u.objects.put(key, executed, "application/pdf")
c.execute(ExecutedCopy(key, sha, size, "GENERATED", ver.version, SYSTEM, at,
f"{c.contract_id}-signed.pdf", "application/pdf"), now=at, chain_ok=ok)
ct = cfg.types.get(c.type_code)
c.plan_key_dates(cfg.key_date_cadence, c.owner_group or c.owner)
for s in c.signers:
url = None
if s.kind == EXTERNAL:
issued = public_door.issue(c.contract_id, s.signer_no, s.email, "READ", days=S.copy_link_days(cfg),
actor=SYSTEM, conn=u.conn)
s.copy_link_id, url = issued.link_id, issued.url
subj, body = copy_message(c, s, cfg, None)
d = new_delivery(c, s.signer_no, "COPY", s.email, subj, body, at)
urls[(s.signer_no, "COPY", d.seq)] = url
S.settle(c, cfg, at)
u.contracts.save(c, now=at)
return {"executed": True, "sha256": sha, "amends": c.amends_contract_id, "kind": c.amendment_kind,
"type": ct.type_code if ct else c.type_code}
try:
out = retry_on_conflict(go)
except RuleBroken as e:
log.warning('{"event": "finalize_refused", "contract_id": "%s", "reason": "%s"}', contract_id, str(e)[:200])
return {"executed": False, "reason": str(e)}
if out and out.get("executed"):
deliver(contract_id, [(k[0], k[1], k[2], v) for k, v in urls.items()])
if out.get("amends"):
supersede(out["amends"], contract_id, out.get("kind") or "AMENDMENT")
return out
def supersede(parent_id: str, child_id: str, kind: str) -> None:
"""CT-14: when the amendment, renewal or termination is signed, the original is SUPERSEDED (or TERMINATED) and its
open key dates close — the new contract's dates replace them. Its evidence is not touched."""
def go():
at = S.now()
with uow() as u:
p = u.contracts.get(parent_id)
if p is None or p.state in (SUPERSEDED, TERMINATED):
return
cfg = u.config.load()
p.state = TERMINATED if kind == "TERMINATION" else SUPERSEDED
p.close_key_dates(now=at, by=SYSTEM, note=f"{kind.lower()} {child_id} signed")
if not any(a.child_contract_id == child_id for a in p.amendments):
from brick_contracts.domain.contract import AmendmentLink
p.amendments.append(AmendmentLink(max((a.seq for a in p.amendments), default=0) + 1, kind, child_id,
"signed", SYSTEM, at))
p.last_note = f"{'terminated' if kind == 'TERMINATION' else 'superseded'} by {child_id}"
S.settle(p, cfg, at)
u.contracts.save(p, now=at)
retry_on_conflict(go)
__all__ = ["deliver", "finalize", "supersede", "invite", "chase", "copy_message", "new_delivery", "LINK_PLACEHOLDER",
"COPY_PLACEHOLDER", "SYSTEM"]