"""After drafting: signed elsewhere (CT-12), cancel (CT-16), amend / renew / terminate (CT-14), administrative changes,
key dates (CT-15), the clock (CT-30) and retention (CT-34).
"""
import base64
import binascii
from datetime import timedelta
from typing import Literal, Optional
from erp_platform import work
from erp_platform.bricks import Refused
from erp_platform.repository import Conflict, retry_on_conflict
from erp_platform.types import Email, Name
from brick_contracts.application import support as S
from brick_contracts.application import workflow as W
from brick_contracts.application.drafting import Values, draft
from brick_contracts.application.use_cases import CT
from brick_contracts.domain.contract import AmendmentLink, ExecutedCopy
from brick_contracts.domain.errors import Invalid, RuleBroken
from brick_contracts.domain.factories import term_of
from brick_contracts.domain.services import next_chase
from brick_contracts.domain.values import (DEAD, DRAFT, EXECUTED, EXTERNAL, INTERNAL, OUT_FOR_SIGNATURE, S_PENDING,
sha256_hex)
from brick_contracts.infrastructure import mail, public_door
MAX_UPLOAD = 3 * 1024 * 1024
def uow():
from brick_contracts.application import use_cases
return use_cases.repository()
# ── signed elsewhere ───────────────────────────────────────────────────────────────────────
@CT.use_case(kind="command", route="POST /contracts/{contract_id}/signed-elsewhere", permission=S.UPLOAD, idempotent=True)
def upload_signed_elsewhere(caller: dict, contract_id: S.ContractId, filename: Name, content_base64: S.B64,
signed_where: S.Short, content_type: S.Code = "application/pdf") -> dict:
"""File a copy signed elsewhere (CT-12) — a deed signed on paper with its witness, or the other side's paper. The
file is kept with its sha256 as the executed copy (provenance UPLOADED); the contract is SIGNED_ELSEWHERE and its
signers SIGNED_ELSEWHERE, never SIGNED, because no event of ours backs a signature; key dates are created as for
any signed contract. The same file twice is one filing."""
def go():
at = S.now()
with S.rules(), uow() as u:
view = S.View(u, caller, at)
c = view.contract(contract_id, manage=True) # 404 first: a file for a row you may not see
try:
content = base64.b64decode(content_base64, validate=True)
except (binascii.Error, ValueError):
raise Refused(422, "content_base64 is not base64") from None
if not content or len(content) > MAX_UPLOAD:
raise Refused(422, f"the file is empty or larger than {MAX_UPLOAD // (1024 * 1024)} MB")
digest = sha256_hex(content)
if c.state == "SIGNED_ELSEWHERE" and c.executed and c.executed.sha256 == digest:
return S.full(c, view)
ext = (filename.rsplit(".", 1)[-1] if "." in filename else "pdf").lower()[:8]
key = f"contracts/{c.contract_id}/executed-upload.{ext}"
c.file_signed_elsewhere(ExecutedCopy(key, digest, len(content), "UPLOADED", None, S.actor(caller), at,
filename, content_type, signed_where), now=at, by=S.actor(caller))
u.objects.put(key, content, content_type)
c.plan_key_dates(view.config.key_date_cadence, c.owner_group or c.owner)
S.settle(c, view.config, at)
u.contracts.save(c, now=at)
return S.full(c, view) | {"_amends": (c.amends_contract_id, c.amendment_kind)}
out = retry_on_conflict(go)
amends = out.pop("_amends", (None, None))
if amends[0]:
W.supersede(amends[0], contract_id, amends[1] or "AMENDMENT")
return out
@CT.use_case(kind="command", route="POST /contracts/{contract_id}/cancel", permission=S.WRITE, idempotent=True)
def cancel(caller: dict, contract_id: S.ContractId, reason: S.Reason) -> dict:
"""Cancel a contract before it is signed (CT-16): CANCELLED, every link revoked (REVOKED), open approvals void.
A signed contract cannot be cancelled — terminate it instead. Cancelling twice is once."""
def go():
at = S.now()
with S.rules(), uow() as u:
view = S.View(u, caller, at)
c = view.contract(contract_id, manage=True)
if c.state == "CANCELLED":
return S.full(c, view)
for lid in c.cancel(reason, now=at):
public_door.revoke(lid, actor=S.actor(caller), reason="cancelled", conn=u.conn)
S.settle(c, view.config, at)
u.contracts.save(c, now=at)
return S.full(c, view)
return retry_on_conflict(go)
# ── amendments ─────────────────────────────────────────────────────────────────────────────
def _amend(caller: dict, contract_id: str, kind: str, changes: dict, note: str) -> dict:
with uow() as u:
view = S.View(u, caller)
parent = view.contract(contract_id, manage=True)
if parent.state not in EXECUTED:
raise Refused(409, "only a signed contract is amended, renewed or terminated")
seq = len([a for a in parent.amendments if a.kind == kind]) + 1
values = {n: v.value for n, v in parent.variables.items() if v.source != "COUNTERPARTY"}
for n, v in parent.variables.items():
if v.currency:
values.setdefault("currency", v.currency)
signers = [{"side": s.side, "full_name": s.full_name, "email": s.email, "title": s.title,
"authority_basis": s.authority_basis, "signing_order": s.signing_order, "pool_sub": s.pool_sub}
for s in parent.signers]
values.update({str(k).lower(): v for k, v in changes.items()})
child, replayed = draft(caller, type_code=parent.type_code, bp_id=parent.bp_id, values=values, signers=signers,
source_ref=f"manual:{kind.lower()}/{parent.contract_id}/{seq}",
context_ref=parent.context_ref, owner=parent.owner or S.actor(caller),
owner_group=parent.owner_group, partial_ok=True)
def mark_child():
at = S.now()
with uow() as u:
c = u.contracts.get(child.contract_id)
if c.amends_contract_id != parent.contract_id:
c.amends_contract_id, c.amendment_kind = parent.contract_id, kind
c.title = f"{kind.title()} of {parent.title}"
u.contracts.save(c, now=at)
def link_parent():
at = S.now()
with uow() as u:
p = u.contracts.get(parent.contract_id)
if not any(a.child_contract_id == child.contract_id for a in p.amendments):
p.amendments.append(AmendmentLink(max((a.seq for a in p.amendments), default=0) + 1, kind,
child.contract_id, note or "", S.actor(caller), at))
u.contracts.save(p, now=at)
retry_on_conflict(mark_child)
retry_on_conflict(link_parent)
with uow() as u:
view = S.View(u, caller)
return S.full(view.contract(child.contract_id), view)
def _end_var(caller, contract_id) -> str:
with uow() as u:
view = S.View(u, caller)
c = view.contract(contract_id, manage=True)
ctx = S.drafting_for(u, view.config, c)
marked = [d.name for d in ctx.version.variables.values() if d.validation.get("term") == "end"]
for n in marked + ["end_date", "ends_on", "last_day"]:
if n in ctx.version.variables:
return n
raise Refused(409, "this contract's template has no end date to move")
@CT.use_case(kind="command", route="POST /contracts/{contract_id}/amendments", permission=S.AMEND)
def amend_contract(caller: dict, contract_id: S.ContractId, changes: Values, note: Optional[S.Reason] = None) -> dict:
"""Amend a signed contract (CT-14): a NEW contract (amends = the original, kind AMENDMENT) drafted from the
original's frozen values with your changes; it goes through approval and signing like any contract, and when it is
signed the original becomes SUPERSEDED. The original's evidence is never touched."""
return _amend(caller, contract_id, "AMENDMENT", dict(changes), note or "")
@CT.use_case(kind="command", route="POST /contracts/{contract_id}/renewal", permission=S.AMEND)
def renew_contract(caller: dict, contract_id: S.ContractId, ends_on: S.Day, changes: Optional[Values] = None,
note: Optional[S.Reason] = None) -> dict:
"""Renew a signed contract to a new end date (CT-14): a RENEWAL contract drafted from the original's frozen values;
when it is signed the original is SUPERSEDED and its END and NOTICE_DEADLINE are replaced by the renewal's."""
return _amend(caller, contract_id, "RENEWAL", {**dict(changes or {}), _end_var(caller, contract_id): ends_on.isoformat()},
note or "")
@CT.use_case(kind="command", route="POST /contracts/{contract_id}/termination", permission=S.AMEND)
def terminate_contract(caller: dict, contract_id: S.ContractId, terminates_on: S.Day,
note: Optional[S.Reason] = None) -> dict:
"""Terminate a signed contract by agreement (CT-14): a TERMINATION contract ending it on `terminates_on`, signed
like any contract; when it is signed the original is TERMINATED and its open key dates close."""
return _amend(caller, contract_id, "TERMINATION", {_end_var(caller, contract_id): terminates_on.isoformat()}, note or "")
@CT.use_case(kind="command", route="POST /contracts/{contract_id}/administrative-changes", permission=S.AMEND)
def record_administrative_change(caller: dict, contract_id: S.ContractId, note: S.Reason,
reference: Optional[S.Short] = None) -> dict:
"""Record an ADMINISTRATIVE change (CT-14) — a typo in an internal reference — without signing: the signed words
are not touched; the change and who made it are kept."""
def go():
at = S.now()
with S.rules(), uow() as u:
view = S.View(u, caller, at)
c = view.contract(contract_id, manage=True)
if reference is not None:
c.reference = reference.strip()
c.amendments.append(AmendmentLink(max((a.seq for a in c.amendments), default=0) + 1, "ADMINISTRATIVE",
None, note.strip(), S.actor(caller), at))
u.contracts.save(c, now=at)
return S.full(c, view)
return retry_on_conflict(go)
@CT.use_case(kind="command", route="POST /contracts/{contract_id}/key-dates/{kind}/{seq}/complete", permission=S.WRITE,
idempotent=True)
def complete_key_date(caller: dict, contract_id: S.ContractId, kind: S.Code, seq: S.SeqNo,
note: Optional[S.Reason] = None) -> dict:
"""Complete a key date (CT-15) — notice given, the statement issued: who and when are recorded and its reminders
stop. Twice is once."""
def go():
at = S.now()
with S.rules(), uow() as u:
view = S.View(u, caller, at)
c = view.contract(contract_id, body=False)
if not (view.manages(c) or view._mine(c.owner, c.owner_group)):
raise S.not_found(f"contract {contract_id}")
k = next((x for x in c.key_dates if x.kind == kind.upper() and x.seq == int(seq)), None)
if k is None:
raise S.not_found(f"key date {kind} {seq}")
if k.done_at is None:
k.done_at, k.done_by, k.note = at, S.actor(caller), (note or "").strip()
u.contracts.save(c, now=at)
return S.summary(c, view.today)
return retry_on_conflict(go)
# ── the clock ──────────────────────────────────────────────────────────────────────────────
def _due_one(c, cfg, at) -> tuple:
"""Decide what the clock owes one contract. Returns (what, urls to deliver, finalize?)."""
what, urls = [], {}
if c.escalate_due(at):
what.append("escalated")
for a in c.approvals:
if a.state == "escalated" and a.escalated_at == at:
W.new_delivery(c, 0, f"NOTICE:escalated:{a.seq}", a.fallback_group, f"Approval needed: {c.title}",
f"Approval {a.seq} for {a.approver_group} passed its timeout and is now for "
f"{a.fallback_group}.", at)
links = c.expire_if_stale(at)
if links is not None:
what.append("expired")
W.new_delivery(c, 0, "NOTICE:expired", c.owner or c.owner_group, f"Not signed in time: {c.title}",
c.last_note, at)
return what, urls, links, False
if c.state == OUT_FOR_SIGNATURE and not c.all_signed:
cadence = cfg.chase_cadence()
for s in c.whose_turn():
started = max([x.signed_at for x in c.signers if x.signing_order < s.signing_order and x.signed_at]
or [c.sent_at or at])
st = next_chase(cadence, started, s.chase_count, s.last_chased_at)
if st.exhausted:
kind = f"NOTICE:chased:{s.signer_no}"
if not any(d.kind == kind for d in c.deliveries):
W.new_delivery(c, 0, kind, c.owner or c.owner_group, f"Still unsigned: {c.title}",
f"{s.full_name} has been chased {s.chase_count} times; chase them yourself.", at)
what.append("exhausted")
continue
if st.due(at):
s.chase_count, s.last_chased_at = s.chase_count + 1, at
subj, body = W.chase(c, s, cfg, None)
d = W.new_delivery(c, s.signer_no, f"CHASE:{s.chase_count}", s.email, subj, body, at)
if s.kind == EXTERNAL and d.channel == "provider":
urls[(s.signer_no, d.kind, d.seq)] = "reissue"
what.append("chased")
for d in c.deliveries:
if d.status == "pending" and d.channel == "provider" and d.retry_at and d.retry_at <= at:
urls[(d.signer_no, d.kind, d.seq)] = "reissue" if d.kind in ("INVITE", "REISSUE") or d.kind.startswith("CHASE") else None
d.retry_at = None
return what, urls, [], c.all_signed and c.executed is None
def tick(at=None) -> dict:
at = at or S.now()
out = {"at": S.iso(at), "contracts": 0, "escalated": 0, "expired": 0, "chased": 0, "executed": 0, "reminders": 0,
"purged": 0}
with uow() as u:
cfg = u.config.load()
due = u.contracts.load_many(u.contracts.due(at))
day = S.today(at, cfg.setting("timezone", "Europe/London"))
kd_ids = u.contracts.key_dates_due(day)
out["contracts"] = len(due)
for c in due:
what, urls, revoke, fin = _due_one(c, cfg, at)
if not (what or urls or revoke or fin or c.next_due_at):
continue
try:
with uow() as u:
for lid in revoke:
public_door.revoke(lid, actor=W.SYSTEM, reason="expired unsigned", conn=u.conn)
for (signer_no, kind, seq), how in list(urls.items()):
if how == "reissue":
s = c.signer(signer_no)
if s.link_id:
issued = public_door.reissue(s.link_id, actor=W.SYSTEM, conn=u.conn)
c.append_event("REISSUED", at, s, version=c.current_version, token_ref=issued.link_id,
detail=f"chase: link replaced {s.link_id}")
s.link_id, urls[(signer_no, kind, seq)] = issued.link_id, issued.url
else:
urls[(signer_no, kind, seq)] = None
S.settle(c, cfg, at)
u.contracts.save(c, now=at)
except Conflict:
continue # someone saved it meanwhile: the next tick decides
for w in what:
out[w] = out.get(w, 0) + 1
W.deliver(c.contract_id, [(k[0], k[1], k[2], v) for k, v in urls.items() if v])
if fin:
r = W.finalize(c.contract_id)
out["executed"] += 1 if r and r.get("executed") else 0
out["reminders"] = remind(kd_ids, day, at)
hour = work.aware(at).astimezone(work.zone(cfg.setting("timezone", "Europe/London"))).hour
if hour >= 2:
with uow() as u:
fresh = u.claims.claim("retention", day.isoformat())
if fresh:
out["purged"] = len(sweep(day))
return out
def remind(ids: list, day, at) -> int:
"""CT-15 / CT-30: one reminder per key date per slot that has come (120/90/60/30 days before), at most once — a
second tick on the same day sends nothing more. When several slots have passed at once, the nearest is sent and
the earlier ones are recorded as passed."""
sent = 0
if not ids:
return 0
with uow() as u:
contracts = u.contracts.load_many(ids)
cfg = u.config.load()
for c in contracts:
new = []
for k in c.key_dates:
if k.done_at is not None or k.next_remind_on is None or k.next_remind_on > day:
continue
come = [(d, when) for d, when in k.pending_slots() if when <= day]
if not come:
continue
nearest = min(d for d, _ in come)
k.reminded = sorted(set(k.reminded) | {d for d, _ in come})
subject = f"{k.kind.replace('_', ' ').title()} in {nearest} days: {c.title}" if nearest else \
f"{k.kind.replace('_', ' ').title()} today: {c.title}"
body = (f"{c.title} ({c.contract_id}): {k.kind.replace('_', ' ').lower()} on {k.due_on.isoformat()} — "
f"{nearest} days from {day.isoformat()}.")
d = W.new_delivery(c, 0, f"REMIND:{k.kind}:{k.seq}:{nearest}", k.owner_group or c.owner, subject, body, at)
new.append((0, d.kind, d.seq, None))
if not new:
continue
try:
with uow() as u:
S.settle(c, cfg, at)
u.contracts.save(c, now=at)
except Conflict:
continue
sent += len(new)
W.deliver(c.contract_id, new)
return sent
@CT.use_case(kind="command", route="POST /run-due", permission=S.TICK, idempotent=True)
def run_due(caller: dict) -> dict:
"""The clock (CT-30), called by the composition's one clock worker: chases the signer whose turn it is, makes a
contract past its link lifetime EXPIRED_UNSIGNED (its links revoked), escalates approvals past their timeout to
the fallback group, produces an executed copy a failed request still owes, reminds owners of key dates (120 / 90 /
60 / 30 days before a notice deadline), and once a day after 02:00 applies retention. Only what is due is read;
each reminder is sent once per slot, so a second tick sends nothing more."""
return tick()
# ── retention ──────────────────────────────────────────────────────────────────────────────
def sweep(day, erase_email: Optional[str] = None) -> list:
"""CT-34: purge what is past its retention date (term + N years; dead drafts after the short window). Every purge
is logged; running it twice changes nothing. An erasure request purges a signer's dead drafts and keeps a live or
executed contract's evidence under the legal-claims exception (UK GDPR art. 17(3)(e)), with the reason recorded."""
done = []
with uow() as u:
ids = u.contracts.retention_due(day)
erase_ids = u.contracts.with_signer_email(erase_email) if erase_email else []
for cid in sorted(set(ids) | set(erase_ids)):
with uow() as u:
c = u.contracts.get(cid)
if c is None:
continue
due = c.retention_until is not None and c.retention_until < day
if not due and cid in erase_ids:
if c.state in DEAD or (c.state == DRAFT and not c.versions):
category = "erasure"
else:
if u.retention.record(day, cid, "erasure_kept",
f"erasure request for {erase_email}: kept — the contract is "
f"{c.state.lower()} and its evidence is needed for legal claims "
"(UK GDPR art. 17(3)(e)) until its retention date"):
done.append({"contract_id": cid, "kept": True})
continue
elif due:
category = "dead_draft" if c.state in DEAD else "executed"
else:
continue
if u.retention.done(day, cid, category):
continue
n = u.objects.delete_prefix(f"contracts/{cid}/")
u.contracts.remove(cid)
u.retention.record(day, cid, category, f"{c.state.lower()} {c.type_code} purged: retention until "
f"{S.iso(c.retention_until)}; {n} file(s); its evidence went with it")
done.append({"contract_id": cid, "category": category})
return done
@CT.use_case(kind="command", route="POST /retention/purge", permission=S.SETTINGS, idempotent=True, destructive=True)
def purge_by_retention(caller: dict, erase_signer_email: Optional[Email] = None) -> dict:
"""Apply retention now (CT-34) — the clock does it daily after 02:00: dead drafts after their short window,
executed contracts and their evidence at term + 6 years (12 for a deed). With erase_signer_email (an erasure
request), that person's dead drafts are purged and live or executed contracts are kept under the legal-claims
exception, the reason recorded. Every purge is logged; twice changes nothing."""
day = S.today()
return {"purged": sweep(day, (erase_signer_email or "").strip().lower() or None), "run_date": day.isoformat()}
__all__ = ["upload_signed_elsewhere", "cancel", "amend_contract", "renew_contract", "terminate_contract",
"record_administrative_change", "complete_key_date", "run_due", "purge_by_retention", "tick", "sweep",
"remind"]