"""Contract — the aggregate (root ct_contract) with its sections, variables, versions, signers, approvals,
signature events, executed copy, deliveries, key dates and amendment links.
INVARIANTS (each enforced here; the scenario that proves it in brackets):
1. the counterparty is one bp_id with a role the type allows — checked at draft (factories) [SC-CT-19]
2. sections and variables change only while editable (DRAFT / IN_APPROVAL / APPROVED); at send they are frozen
into version N with its sha256; a re-send is version N+1 [SC-CT-22]
3. an approval binds to ONE draft fingerprint; any edit changes the fingerprint and voids it; an owner cannot
approve their own draft; an approval not decided by its timeout goes to the mandatory fallback group, and the
original group can no longer decide it [SC-CT-01, 02, 04]
4. send is refused, in words, while missing[] is non-empty, an approval is open, the type is a deed or
esign_allowed is false, both sides are not named, or a pack check fails [SC-CT-08, 13, 14, 16]
5. every signature event names one version and the sha256 the signer was shown; a SIGN for any version other
than the current frozen one is refused (a known failure mode); signing out of order is refused naming who it waits on;
signing twice is one signature [SC-CT-05, 06, 22]
6. signature events are append-only and hash-chained (prev_hash → row_hash); a broken chain blocks the executed
copy [SC-CT-11]
7. SIGNED needs an executed copy AND a SIGNED event from every signer of the signed version; SIGNED_ELSEWHERE needs
an uploaded executed copy and marks signers SIGNED_ELSEWHERE, never SIGNED (a known failure mode) [SC-CT-08]
8. IP and user agent are structured fields of an event, never free text (a known failure mode)
9. an INTERNAL signer signs only in the app as their pool identity; an EXTERNAL one only through a public link
[SC-CT-25]
10. the executed copy is released to a consumer only if it is the originator (source_ref namespace) [SC-CT-24]
"""
from dataclasses import dataclass, field
from datetime import date, datetime, timedelta, timezone
from typing import Any, Optional
from brick_contracts.domain.errors import Incomplete, Invalid, NotFound, RuleBroken
from brick_contracts.domain.values import (APPROVED, CANCELLED, CHANGES_REQUESTED, DECLINED, DRAFT, EDITABLE,
EMAIL_LINK, EXECUTED, EXPIRED, EXPIRED_UNSIGNED, EXTERNAL, IN_APPROVAL,
INTERNAL, OUT_FOR_SIGNATURE, POOL_SESSION, S_CHANGES, S_DECLINED,
S_ELSEWHERE, S_PENDING, S_SIGNED, SIGNED, SIGNED_ELSEWHERE, SUPERSEDED,
TERMINATED, THEM, US, ACTIVE, ClausePosition, EventHash, Fingerprint,
VariableValue, namespace_of, utc_iso)
UA_CHARS = 300
def _utc(t: Optional[datetime]) -> Optional[datetime]:
if t is None:
return None
return (t if t.tzinfo else t.replace(tzinfo=timezone.utc)).astimezone(timezone.utc)
def evidence_time(t: datetime) -> datetime:
"""Event times are kept to the second: what is hashed must survive every engine's TIMESTAMP round trip."""
return _utc(t).replace(microsecond=0)
@dataclass
class Section:
position: int
family: str
clause_id: str
clause_version: int
position_kind: str
deviation_level: int
title: str
body: str
slot_no: int = 0
required: bool = True
overridden: bool = False
def canonical(self) -> dict:
return {"position": self.position, "family": self.family, "clause_id": self.clause_id,
"clause_version": self.clause_version, "position_kind": self.position_kind,
"deviation_level": self.deviation_level, "overridden": self.overridden, "title": self.title,
"body": self.body}
@dataclass
class Signer:
signer_no: int
kind: str # INTERNAL | EXTERNAL
side: str # US | THEM
signing_order: int
email: str
full_name: str
title: str = ""
authority_basis: str = ""
role: str = "SIGNER"
bp_id: Optional[int] = None
pool_sub: Optional[str] = None
status: str = S_PENDING
link_id: Optional[str] = None # the live SIGN link (EXTERNAL only); the token itself is never held
copy_link_id: Optional[str] = None
signed_at: Optional[datetime] = None
signed_version: Optional[int] = None
first_opened_at: Optional[datetime] = None
last_opened_at: Optional[datetime] = None
delivered_at: Optional[datetime] = None
chase_count: int = 0
last_chased_at: Optional[datetime] = None
def canonical(self) -> dict:
return {"signer_no": self.signer_no, "kind": self.kind, "side": self.side, "order": self.signing_order,
"email": self.email, "full_name": self.full_name, "title": self.title,
"authority_basis": self.authority_basis}
@property
def done(self) -> bool:
return self.status in (S_SIGNED, S_ELSEWHERE)
def is_me(self, caller: dict) -> bool:
"""The signed-in caller IS this internal signer: the pool sub, or the email the gateway verified."""
sub = str(caller.get("sub") or "").lower()
email = str(caller.get("email") or "").lower()
return bool((self.pool_sub and sub and self.pool_sub.lower() == sub) or (email and email == self.email.lower()))
@dataclass
class Version:
version: int
object_key: str
sha256: str
size_bytes: int
frozen_at: datetime
draft_fingerprint: str
text_body: str
frozen_by: str = ""
status: str = "ISSUED" # ISSUED | SUPERSEDED | SIGNED
@dataclass
class Approval:
seq: int
draft_fingerprint: str
deviations: list
approver_group: str
fallback_group: str
due_at: datetime
requested_at: datetime
requested_by: str = ""
state: str = "pending" # pending | approved | rejected | void | escalated
decided_by: Optional[str] = None
decided_at: Optional[datetime] = None
reason: Optional[str] = None
escalated_at: Optional[datetime] = None
original_group: Optional[str] = None
@property
def open(self) -> bool:
return self.state in ("pending", "escalated")
@property
def deciding_group(self) -> str:
return self.fallback_group if self.state == "escalated" else self.approver_group
@dataclass
class SignatureEvent:
seq: int
kind: str
occurred_at: datetime
signer_no: Optional[int] = None
signer_kind: Optional[str] = None
version: Optional[int] = None
version_sha256_shown: Optional[str] = None
email: Optional[str] = None
token_ref: Optional[str] = None
pool_sub: Optional[str] = None
session_id: Optional[str] = None
ip: Optional[str] = None
user_agent: Optional[str] = None
typed_name: Optional[str] = None
consent_text_version: Optional[str] = None
consent_text: Optional[str] = None
auth_method: Optional[str] = None
detail: Optional[str] = None
prev_hash: str = ""
row_hash: str = ""
HASHED = ("seq", "kind", "occurred_at", "signer_no", "signer_kind", "version", "version_sha256_shown", "email",
"token_ref", "pool_sub", "session_id", "ip", "user_agent", "typed_name", "consent_text_version",
"consent_text", "auth_method", "detail")
def canonical(self, contract_id: str) -> dict:
out = {"contract_id": contract_id}
for k in self.HASHED:
v = getattr(self, k)
if k == "occurred_at":
v = utc_iso(evidence_time(v))
elif k in ("seq", "signer_no", "version") and v is not None:
v = int(v)
out[k] = v
return out
@dataclass
class ExecutedCopy:
object_key: str
sha256: str
size_bytes: int
provenance: str # GENERATED | UPLOADED
signed_version: Optional[int]
filed_by: str
filed_at: datetime
filename: str
content_type: str
signed_where: str = ""
@dataclass
class Delivery:
"""Something to get to a person: a signing link, a reminder, the executed copy. channel provider | manual."""
signer_no: int # 0 = the owner (a key-date reminder, a notice)
kind: str # INVITE | CHASE:<n> | COPY | REMIND:<kind>:<seq>:<days> | NOTICE:<what>
seq: int
channel: str
recipient: str
subject: str
message: str
created_at: datetime
status: str = "pending" # pending | sent | done | failed
capability: str = "mail.send"
idempotency_key: str = ""
plan: str = ""
reason: str = ""
provider_ref: Optional[str] = None
done_by: Optional[str] = None
done_at: Optional[datetime] = None
retry_at: Optional[datetime] = None
@dataclass
class KeyDateRow:
kind: str
seq: int
due_on: date
owner_group: str
days_before: tuple = ()
reminded: list = field(default_factory=list) # days-before slots handled (sent or skipped)
done_at: Optional[datetime] = None
done_by: Optional[str] = None
note: str = ""
def pending_slots(self) -> list:
return sorted(((d, self.due_on - timedelta(days=d)) for d in self.days_before if d not in self.reminded),
key=lambda x: x[1])
@property
def next_remind_on(self) -> Optional[date]:
if self.done_at is not None:
return None
slots = self.pending_slots()
return slots[0][1] if slots else None
@dataclass
class AmendmentLink:
seq: int
kind: str
child_contract_id: Optional[str]
note: str
recorded_by: str
recorded_at: datetime
@dataclass
class Contract:
contract_id: str
type_code: str
template_id: str
template_version: int
bp_id: int
source_ref: str
owner: str
created_by: str
created_at: datetime
title: str = ""
owner_group: str = ""
context_ref: Optional[str] = None
state: str = DRAFT
sensitivity: str = "commercial"
execution: str = "SIMPLE"
starts_on: Optional[date] = None
ends_on: Optional[date] = None
notice_days: Optional[int] = None
amends_contract_id: Optional[str] = None
amendment_kind: Optional[str] = None
draft_fingerprint: str = ""
request_hash: str = ""
missing: list = field(default_factory=list)
deviations: list = field(default_factory=list)
reference: str = ""
current_version: int = 0
sent_at: Optional[datetime] = None
link_expires_at: Optional[datetime] = None
signed_at: Optional[datetime] = None
principal_statement_issued_at: Optional[datetime] = None
last_note: str = ""
retention_until: Optional[date] = None
next_due_at: Optional[datetime] = None
version: int = 0 # the optimistic version (erp_platform.repository)
variables: dict = field(default_factory=dict) # name → VariableValue
frozen_in: dict = field(default_factory=dict) # name → version it was frozen in
sections: list = field(default_factory=list)
signers: list = field(default_factory=list)
versions: dict = field(default_factory=dict) # n → Version
approvals: list = field(default_factory=list)
events: list = field(default_factory=list)
new_events: list = field(default_factory=list) # appended in this unit of work (plain INSERTs)
executed: Optional[ExecutedCopy] = None
deliveries: list = field(default_factory=list)
key_dates: list = field(default_factory=list)
amendments: list = field(default_factory=list)
# ── identity and the originator ──────────────────────────────────────────────────────────
@property
def origin(self) -> str:
return namespace_of(self.source_ref)
def originated_by(self, namespaces: list) -> bool:
"""Invariant 10: only the consumer whose call created the contract is given its executed copy."""
return bool(self.origin) and self.origin != "manual" and self.origin in namespaces
# ── state ────────────────────────────────────────────────────────────────────────────────
def derived_state(self, today: date) -> str:
"""ACTIVE and EXPIRED are derived from the term on a signed contract (CT-33); nothing stores them."""
if self.state in EXECUTED:
if self.ends_on and today > self.ends_on:
return EXPIRED
if self.starts_on is None or self.starts_on <= today:
return ACTIVE
return self.state
@property
def editable(self) -> bool:
return self.state in EDITABLE
def require_editable(self, doing: str = "edit it") -> None:
if self.state == CHANGES_REQUESTED:
self.state = DRAFT
return
if self.editable:
return
if self.state == OUT_FOR_SIGNATURE:
raise RuleBroken(f"version {self.current_version} is out for signature; recall it to {doing}")
if self.state in EXECUTED:
raise RuleBroken("a signed contract is not edited — amend, renew or terminate it")
raise RuleBroken(f"the contract is {self.state.lower().replace('_', ' ')}; it can no longer be changed")
# ── the fingerprint and approvals ────────────────────────────────────────────────────────
def fingerprint(self) -> Fingerprint:
return Fingerprint.of({
"title": self.title,
"sections": [s.canonical() for s in sorted(self.sections, key=lambda s: s.position)],
"variables": [v.canonical_row() for _, v in sorted(self.variables.items())],
"signers": [s.canonical() for s in sorted(self.signers, key=lambda s: s.signer_no)]})
def review(self, missing: list, deviations: list, *, now: datetime, by: str, calendar=None) -> None:
"""After any change while editable: recompute the fingerprint; void approvals bound to another one;
open an approval per group the deviations need; set DRAFT / IN_APPROVAL / APPROVED (invariant 3)."""
from erp_platform.work import WorkCalendar
fp = self.fingerprint().hex
self.missing, self.deviations = sorted(set(missing)), [d.as_json() if hasattr(d, "as_json") else d
for d in deviations]
self.draft_fingerprint = fp
for a in self.approvals:
if a.draft_fingerprint != fp and a.state in ("pending", "escalated", "approved"):
a.state, a.reason = "void", "the draft changed after it was asked for"
needed = {}
for d in self.deviations:
if d.get("needs"):
needed.setdefault(d["needs"], {"fallback": d.get("fallback"), "items": [],
"timeout": int(d.get("timeout") or 2)})["items"].append(d)
cal = calendar or WorkCalendar()
for group, info in sorted(needed.items()):
mine = [a for a in self.approvals if a.draft_fingerprint == fp and (a.original_group or a.approver_group) == group]
if any(a.state in ("pending", "escalated", "approved", "rejected") for a in mine):
continue
n = max((a.seq for a in self.approvals), default=0) + 1
due_day = cal.add_working_days(_utc(now).date(), max(1, info["timeout"]))
due = datetime.combine(due_day, _utc(now).time(), tzinfo=timezone.utc)
self.approvals.append(Approval(n, fp, info["items"], group, info["fallback"] or "", due, now, by))
self._settle_approval_state(fp, set(needed))
def _settle_approval_state(self, fp: str, needed: set) -> None:
if not needed:
self.state = DRAFT
return
current = [a for a in self.approvals if a.draft_fingerprint == fp]
groups = {(a.original_group or a.approver_group): a for a in current if a.state != "void"}
if any(a.state == "rejected" for a in groups.values()):
self.state = DRAFT
elif all(g in groups and groups[g].state == "approved" for g in needed):
self.state = APPROVED
else:
self.state = IN_APPROVAL
@property
def approval_required(self) -> bool:
return any(d.get("needs") for d in self.deviations)
def open_approvals(self) -> list:
return [a for a in self.approvals if a.open and a.draft_fingerprint == self.draft_fingerprint]
def approval_granted(self) -> bool:
if not self.approval_required:
return True
needed = {d["needs"] for d in self.deviations if d.get("needs")}
ok = {(a.original_group or a.approver_group) for a in self.approvals
if a.state == "approved" and a.draft_fingerprint == self.draft_fingerprint}
return needed <= ok
def request_approval(self, *, now: datetime, by: str, calendar=None) -> list:
"""Ask again after a rejection (same fingerprint): the rejected approvals are closed and fresh ones open."""
self.require_editable("ask for approval")
if not self.approval_required:
raise RuleBroken("nothing in this draft is off the standard; no approval is needed")
for a in self.approvals:
if a.draft_fingerprint == self.draft_fingerprint and a.state == "rejected":
a.state = "void"
self.review(self.missing, self.deviations, now=now, by=by, calendar=calendar)
return self.open_approvals()
def decide(self, seq: int, approve: bool, *, by: str, groups: set, now: datetime, reason: str = "") -> Approval:
a = next((x for x in self.approvals if x.seq == int(seq)), None)
if a is None:
raise NotFound(f"no approval {seq} on {self.contract_id}")
if a.state not in ("pending", "escalated"):
raise RuleBroken(f"approval {seq} is {a.state}; nothing to decide")
if a.deciding_group not in groups:
if a.state == "escalated" and a.approver_group in groups:
raise RuleBroken(f"approval {seq} passed its timeout and went to {a.fallback_group}; it is theirs now")
raise NotFound(f"no approval {seq} in your queue")
if by and by.lower() in {self.owner.lower(), (self.created_by or "").lower()}:
raise RuleBroken("you own this draft; someone else must approve its deviation")
if a.draft_fingerprint != self.draft_fingerprint:
raise RuleBroken("the draft changed after this approval was asked for; it is void")
if approve:
a.state, a.decided_by, a.decided_at, a.reason = "approved", by, now, reason or None
else:
if not (reason or "").strip():
raise Invalid("a rejection says why")
a.state, a.decided_by, a.decided_at, a.reason = "rejected", by, now, reason.strip()
self.last_note = f"rejected by {by}: {reason.strip()}"
self._settle_approval_state(self.draft_fingerprint, {d["needs"] for d in self.deviations if d.get("needs")})
return a
def escalate_due(self, now: datetime) -> list:
"""SC-CT-04: an approval not decided by its timeout goes to the policy's mandatory fallback group."""
out = []
for a in self.approvals:
if a.state == "pending" and a.draft_fingerprint == self.draft_fingerprint and _utc(a.due_at) <= _utc(now):
a.original_group = a.approver_group
a.state, a.escalated_at = "escalated", now
out.append(a)
return out
# ── editing ──────────────────────────────────────────────────────────────────────────────
def section(self, position: int) -> Section:
s = next((x for x in self.sections if x.position == int(position)), None)
if s is None:
raise NotFound(f"no section {position} in {self.contract_id}")
return s
def override_section(self, position: int, title: Optional[str], body: str) -> Section:
self.require_editable("rewrite a clause")
s = self.section(position)
body = (body or "").strip()
if not body:
raise Invalid("an override carries the words for this contract")
if len(body) > 20000:
raise Invalid("a clause is at most 20 000 characters")
s.title, s.body = (title or s.title).strip(), body
s.position_kind, s.deviation_level, s.overridden = "OVERRIDE", 2, True
return s
def choose_position(self, position: int, clause, version: Optional[int] = None) -> Section:
"""Switch a section to another clause of the same family (Preferred / Alternate / Fallback): a fresh COPY of
that clause's words; the library is untouched (CT-06)."""
self.require_editable("change a clause")
s = self.section(position)
if clause.family != s.family:
raise RuleBroken(f"section {position} is the {s.family} family; {clause.clause_id} is {clause.family}")
if clause.status != "active":
raise RuleBroken(f"{clause.clause_id} is retired")
v = clause.versions[int(version)] if version else clause.current
s.clause_id, s.clause_version, s.title, s.body = clause.clause_id, v.version, v.title, v.body
s.position_kind, s.deviation_level, s.overridden = clause.position, clause.deviation_level, False
return s
def set_variable(self, v: VariableValue) -> None:
self.require_editable("change its values")
self.variables[v.name] = v
def add_signer(self, *, kind: str, side: str, signing_order: int, email: str, full_name: str, title: str = "",
authority_basis: str = "", bp_id: Optional[int] = None, pool_sub: Optional[str] = None) -> Signer:
self.require_editable("change its signers")
side, kind = side.upper(), kind.upper()
if side not in (US, THEM) or kind not in (INTERNAL, EXTERNAL):
raise Invalid("a signer is on side US or THEM, and INTERNAL or EXTERNAL")
if side == US and kind != INTERNAL:
raise RuleBroken("our side signs inside the app (INTERNAL), never through a link")
if side == THEM and kind != EXTERNAL:
raise RuleBroken("the counterparty signs through a public link (EXTERNAL)")
email = (email or "").strip().lower()
if kind == EXTERNAL and "@" not in email:
raise Invalid(f"{full_name or 'the signer'} needs an email address their link can be sent to")
if kind == INTERNAL and not (email or pool_sub):
raise Invalid("our signer is named by their sign-in email (or pool sub)")
if not (full_name or "").strip():
raise Invalid("a signer has a name")
if not 1 <= int(signing_order) <= 20:
raise Invalid("a signing order is 1 … 20")
if any(s.email == email and email for s in self.signers):
raise RuleBroken(f"{email} already signs this contract")
n = max((s.signer_no for s in self.signers), default=0) + 1
s = Signer(n, kind, side, int(signing_order), email, full_name.strip(), (title or "").strip(),
(authority_basis or "").strip(), bp_id=bp_id, pool_sub=pool_sub)
self.signers.append(s)
return s
def signer(self, signer_no: int) -> Signer:
s = next((x for x in self.signers if x.signer_no == int(signer_no)), None)
if s is None:
raise NotFound(f"no signer {signer_no} on {self.contract_id}")
return s
def edit_signer(self, signer_no: int, **changes: Any) -> Signer:
self.require_editable("change its signers")
s = self.signer(signer_no)
for k in ("email", "full_name", "title", "authority_basis", "signing_order", "pool_sub"):
if changes.get(k) is not None:
v = changes[k]
if k == "email":
v = str(v).strip().lower()
if s.kind == EXTERNAL and "@" not in v:
raise Invalid("an external signer needs an email address")
if k == "signing_order":
v = int(v)
if not 1 <= v <= 20:
raise Invalid("a signing order is 1 … 20")
setattr(s, k, v)
return s
def remove_signer(self, signer_no: int) -> None:
self.require_editable("change its signers")
self.signers = [s for s in self.signers if s.signer_no != int(signer_no)]
# ── readiness and the send gate ──────────────────────────────────────────────────────────
@property
def both_sides_named(self) -> bool:
return any(s.side == US for s in self.signers) and any(s.side == THEM for s in self.signers)
def readiness(self, today: date) -> list:
"""What stands between here and signed (CT-33), each item keyed so the screen sends you to the fix."""
turn = self.whose_turn()
return [
{"key": "clauses", "label": "At least one clause", "met": bool(self.sections),
"hint": "a contract with no terms is a letter"},
{"key": "both_sides", "label": "Somebody on each side to sign", "met": self.both_sides_named,
"hint": self._missing_side() or "both sides are named"},
{"key": "term", "label": "A term", "met": self.starts_on is not None,
"hint": "when it starts; the end may be open"},
{"key": "nothing_missing", "label": "Every required value", "met": not self.missing,
"hint": ("missing: " + ", ".join(self.missing)) if self.missing else "complete"},
{"key": "approval", "label": "Deviations approved", "met": self.approval_granted(),
"hint": self._approval_hint()},
{"key": "signed", "label": "Both sides have signed",
"met": self.both_sides_named and all(s.done for s in self.signers),
"hint": f"waiting on {', '.join(s.full_name for s in turn)}" if turn else
(self._missing_side() or "every signer has signed")},
{"key": "executed_copy", "label": "The executed copy is on the record", "met": self.executed is not None,
"hint": "signed elsewhere still means filed here"},
]
def _missing_side(self) -> str:
if not any(s.side == US for s in self.signers):
return "nobody on our side is named, so this cannot be complete"
if not any(s.side == THEM for s in self.signers):
return "nobody on the counterparty's side is named, so this cannot be complete"
return ""
def _approval_hint(self) -> str:
if self.approval_granted():
return "approved" if self.approval_required else "on the standard: no approval needed"
pending = [d for d in self.deviations if d.get("needs")]
return "; ".join(f"approval by {d['needs']} needed: {d['what']}" for d in pending)
def send_refusals(self, ctype, pack_refusals: list) -> list:
"""Invariant 4 — every reason this cannot be sent, in words, in the order a person fixes them."""
out = []
if ctype.is_deed or not ctype.esign_allowed:
out.append(f"a {ctype.label} is {'a deed' if ctype.is_deed else 'not signed online'}: it is not signed "
"online here — sign it on paper and upload the executed copy")
if self.missing:
out.append("missing: " + ", ".join(self.missing))
if not self.sections:
out.append("the contract has no clauses")
if not self.approval_granted():
out.append(self._approval_hint())
side = self._missing_side()
if side:
out.append(side)
for s in self.signers:
if s.kind == EXTERNAL and "@" not in (s.email or ""):
out.append(f"{s.full_name} has no email address for their signing link")
out.extend(pack_refusals)
return out
# ── the chain ────────────────────────────────────────────────────────────────────────────
def append_event(self, kind: str, now: datetime, signer: Optional[Signer] = None, **fields: Any) -> SignatureEvent:
"""Invariant 6: the next event, chained to the last (row_hash = sha256(prev_hash ‖ canonical row))."""
last = self.events[-1] if self.events else None
seq = (last.seq if last else 0) + 1
ua = fields.pop("user_agent", None)
ev = SignatureEvent(seq, kind, evidence_time(now),
signer_no=signer.signer_no if signer else fields.pop("signer_no", None),
signer_kind=signer.kind if signer else None,
email=(signer.email if signer else fields.pop("email", None)),
user_agent=(ua or "")[:UA_CHARS] or None, **fields)
ev.prev_hash = last.row_hash if last else EventHash.genesis(self.contract_id)
ev.row_hash = EventHash.chain(ev.prev_hash, ev.canonical(self.contract_id)).row_hash
self.events.append(ev)
self.new_events.append(ev)
return ev
# ── send ─────────────────────────────────────────────────────────────────────────────────
def freeze(self, *, object_key: str, sha256: str, size_bytes: int, text_body: str, now: datetime, by: str,
link_lifetime: timedelta) -> Version:
"""Invariant 2: version N+1 frozen with its sha256; variables frozen in it; OUT_FOR_SIGNATURE."""
n = max(self.versions, default=0) + 1
for v in self.versions.values():
if v.status == "ISSUED":
v.status = "SUPERSEDED"
ver = Version(n, object_key, sha256, size_bytes, now, self.draft_fingerprint, text_body, by)
self.versions[n] = ver
self.current_version = n
for name in self.variables:
self.frozen_in[name] = n
self.state, self.sent_at, self.link_expires_at = OUT_FOR_SIGNATURE, now, now + link_lifetime
for s in self.signers:
s.status, s.signed_at, s.signed_version, s.chase_count, s.last_chased_at = S_PENDING, None, None, 0, None
self.append_event("INVITED", now, s, version=n, version_sha256_shown=sha256,
auth_method=EMAIL_LINK if s.kind == EXTERNAL else POOL_SESSION)
return ver
@property
def current(self) -> Optional[Version]:
return self.versions.get(self.current_version)
def whose_turn(self) -> list:
"""Everybody at the lowest signing order still to sign (a known failure mode: asking signer 2 first is how the wrong version
gets countersigned)."""
if self.state != OUT_FOR_SIGNATURE:
return []
outstanding = [s for s in self.signers if s.status == S_PENDING]
if not outstanding:
return []
nxt = min(s.signing_order for s in outstanding)
return sorted((s for s in outstanding if s.signing_order == nxt), key=lambda s: s.signer_no)
def _check_can_act(self, s: Signer, version: int, sha_shown: str, *, via: str) -> None:
if self.state != OUT_FOR_SIGNATURE:
raise RuleBroken("this contract is not out for signature any more")
if via == EXTERNAL and s.kind != EXTERNAL:
raise NotFound("no such signer") # invariant 9: a link never signs for us
if via == INTERNAL and s.kind != INTERNAL:
raise NotFound("no such signer")
cur = self.current
if cur is None or int(version) != cur.version or (sha_shown or "") != cur.sha256:
raise RuleBroken("this contract was re-issued: open the new link" if via == EXTERNAL else
f"this contract was re-issued as version {cur.version if cur else '?'}: open it again")
def sign(self, signer_no: int, *, version: int, sha_shown: str, typed_name: str, consented: bool,
consent, now: datetime, via: str, ip: Optional[str] = None, user_agent: Optional[str] = None,
token_ref: Optional[str] = None, pool_sub: Optional[str] = None, session_id: Optional[str] = None) -> bool:
"""Invariant 5. True when this call signed; False when the signer had already signed (one signature)."""
s = self.signer(signer_no)
if s.status == S_SIGNED and s.signed_version == self.current_version:
return False
self._check_can_act(s, version, sha_shown, via=via)
if s.status != S_PENDING:
raise RuleBroken(f"{s.full_name} has already answered ({s.status.lower().replace('_', ' ')})")
turn = self.whose_turn()
if s not in turn:
raise RuleBroken(f"it is not {s.full_name}'s turn yet — this contract signs in order and is waiting on "
f"{', '.join(x.full_name for x in turn)}")
if not consented:
raise Invalid("tick the box to agree to sign electronically")
name = (typed_name or "").strip()
if len(name) < 2 or len(name) > 200:
raise Invalid("type your full name to sign")
method = EMAIL_LINK if via == EXTERNAL else POOL_SESSION
common = dict(version=self.current_version, version_sha256_shown=sha_shown, ip=ip, user_agent=user_agent,
token_ref=token_ref, pool_sub=pool_sub, session_id=session_id, auth_method=method)
self.append_event("CONSENTED", now, s, consent_text_version=consent.version, consent_text=consent.text, **common)
self.append_event("SIGNED", now, s, typed_name=name, consent_text_version=consent.version, **common)
s.status, s.signed_at, s.signed_version = S_SIGNED, now, self.current_version
return True
def record_opened(self, signer_no: int, *, now: datetime, via: str, ip=None, user_agent=None, token_ref=None,
pool_sub=None, session_id=None) -> None:
s = self.signer(signer_no)
cur = self.current
self.append_event("OPENED", now, s, version=cur.version if cur else None,
version_sha256_shown=cur.sha256 if cur else None, ip=ip, user_agent=user_agent,
token_ref=token_ref, pool_sub=pool_sub, session_id=session_id,
auth_method=EMAIL_LINK if via == EXTERNAL else POOL_SESSION)
s.first_opened_at = s.first_opened_at or now
s.last_opened_at = now
if s.side == THEM and self.principal_statement_issued_at is None:
self.principal_statement_issued_at = now # given: they have it in front of them
@property
def all_signed(self) -> bool:
return (self.state == OUT_FOR_SIGNATURE and self.both_sides_named and bool(self.signers)
and all(s.status == S_SIGNED and s.signed_version == self.current_version for s in self.signers))
def decline(self, signer_no: int, reason: str, *, now: datetime, via: str, version: int, sha_shown: str,
ip=None, user_agent=None, token_ref=None, pool_sub=None, session_id=None) -> list:
"""CT-10: DECLINED; every other open link revoked. Returns the link ids to revoke."""
s = self.signer(signer_no)
self._check_can_act(s, version, sha_shown, via=via)
if not (reason or "").strip():
raise Invalid("say why you are declining")
self.append_event("DECLINED", now, s, version=self.current_version, version_sha256_shown=sha_shown, ip=ip,
user_agent=user_agent, token_ref=token_ref, pool_sub=pool_sub, session_id=session_id,
auth_method=EMAIL_LINK if via == EXTERNAL else POOL_SESSION, detail=reason.strip()[:2000])
s.status = S_DECLINED
self.state, self.last_note = DECLINED, f"{s.full_name} declined: {reason.strip()[:500]}"
return self._revoke_all(now, "declined", except_no=s.signer_no)
def request_changes(self, signer_no: int, note: str, *, now: datetime, via: str, version: int, sha_shown: str,
ip=None, user_agent=None, token_ref=None) -> list:
"""CT-10: CHANGES_REQUESTED recorded; every other link revoked; back to DRAFT for a new version."""
s = self.signer(signer_no)
self._check_can_act(s, version, sha_shown, via=via)
if not (note or "").strip():
raise Invalid("say what you would like changed")
self.append_event("CHANGES_REQUESTED", now, s, version=self.current_version, version_sha256_shown=sha_shown,
ip=ip, user_agent=user_agent, token_ref=token_ref,
auth_method=EMAIL_LINK if via == EXTERNAL else POOL_SESSION, detail=note.strip()[:2000])
s.status = S_CHANGES
out = self._revoke_all(now, "changes requested", except_no=s.signer_no)
self._back_to_draft()
self.last_note = f"{s.full_name} asked for changes: {note.strip()[:500]}"
return out
def _back_to_draft(self) -> None:
cur = self.current
if cur is not None and cur.status == "ISSUED":
cur.status = "SUPERSEDED"
for s in self.signers:
s.status = S_PENDING
self.state = DRAFT
self._settle_approval_state(self.draft_fingerprint, {d["needs"] for d in self.deviations if d.get("needs")})
def _revoke_all(self, now: datetime, reason: str, except_no: Optional[int] = None) -> list:
links = []
for s in self.signers:
for attr in ("link_id",):
lid = getattr(s, attr)
if lid:
links.append(lid)
if s.signer_no != except_no:
self.append_event("REVOKED", now, s, version=self.current_version, token_ref=lid,
detail=reason)
setattr(s, attr, None)
return links
def recall(self, reason: str, *, now: datetime) -> list:
"""OUT_FOR_SIGNATURE → DRAFT: every token revoked, the reason recorded; the next send is version N+1."""
if self.state != OUT_FOR_SIGNATURE:
raise RuleBroken("only a contract out for signature is recalled")
if not (reason or "").strip():
raise Invalid("a recall says why")
out = self._revoke_all(now, f"recalled: {reason.strip()[:500]}")
self._back_to_draft()
self.last_note = f"recalled: {reason.strip()[:500]}"
return out
def cancel(self, reason: str, *, now: datetime) -> list:
if self.state in EXECUTED or self.state in (TERMINATED, SUPERSEDED):
raise RuleBroken("a signed contract cannot be cancelled — terminate it instead")
if self.state in (CANCELLED, DECLINED, EXPIRED_UNSIGNED):
raise RuleBroken(f"the contract is already {self.state.lower().replace('_', ' ')}")
if not (reason or "").strip():
raise Invalid("a cancellation says why")
out = self._revoke_all(now, f"cancelled: {reason.strip()[:500]}")
for a in self.approvals:
if a.open:
a.state, a.reason = "void", "cancelled"
self.state, self.last_note = CANCELLED, f"cancelled: {reason.strip()[:500]}"
return out
def expire_if_stale(self, now: datetime) -> Optional[list]:
"""CT-30: past the link lifetime unsigned → EXPIRED_UNSIGNED, its tokens revoked. None: not stale."""
if self.state == OUT_FOR_SIGNATURE and self.link_expires_at and _utc(self.link_expires_at) <= _utc(now) \
and not self.all_signed:
out = self._revoke_all(now, "expired unsigned")
self.state, self.last_note = EXPIRED_UNSIGNED, "the signing links expired before everyone signed"
return out
return None
# ── execution ────────────────────────────────────────────────────────────────────────────
def execute(self, copy: ExecutedCopy, *, now: datetime, chain_ok: bool) -> None:
"""Invariant 7 (online): SIGNED needs a SIGNED event from every signer of the signed version and the copy."""
if self.executed is not None:
return
if not chain_ok:
raise RuleBroken("the evidence chain does not verify; the executed copy is not produced")
if not self.all_signed:
raise RuleBroken("not everyone has signed this version yet")
signed = {e.signer_no for e in self.events if e.kind == "SIGNED" and e.version == self.current_version}
if signed != {s.signer_no for s in self.signers}:
raise RuleBroken("a SIGNED event is missing for a signer of this version")
self.executed = copy
self.current.status = "SIGNED"
self.state, self.signed_at = SIGNED, now
for s in self.signers:
s.link_id = None # used up
def file_signed_elsewhere(self, copy: ExecutedCopy, *, now: datetime, by: str) -> None:
"""Invariant 7 (paper): the uploaded copy; signers SIGNED_ELSEWHERE, never SIGNED (a known failure mode)."""
if self.state not in (DRAFT, APPROVED):
if self.state == IN_APPROVAL:
raise RuleBroken("the draft is waiting for approval; file the signed copy once it is approved")
if self.state == OUT_FOR_SIGNATURE:
raise RuleBroken(f"version {self.current_version} is out for signature online; recall it first")
raise RuleBroken(f"the contract is {self.state.lower().replace('_', ' ')}; nothing to file")
if not self.both_sides_named:
raise RuleBroken(self._missing_side())
self.executed = copy
self.state, self.signed_at = SIGNED_ELSEWHERE, now
for s in self.signers:
s.status = S_ELSEWHERE
self.append_event("SIGNED_ELSEWHERE", now, None, detail=(copy.signed_where or "signed elsewhere")[:500],
version_sha256_shown=copy.sha256)
# ── key dates and retention ──────────────────────────────────────────────────────────────
def plan_key_dates(self, cadence_for, owner_group: str) -> None:
"""START, END and NOTICE_DEADLINE (= end − notice, its own date) for an executed contract (CT-15)."""
wanted = []
if self.starts_on:
wanted.append(("START", self.starts_on))
if self.ends_on:
wanted.append(("END", self.ends_on))
if self.notice_days:
wanted.append(("NOTICE_DEADLINE", self.ends_on - timedelta(days=int(self.notice_days))))
pe = self.variables.get("probation_end")
if pe is not None and pe.type == "date":
wanted.append(("PROBATION_END", pe.as_date))
sds = self.variables.get("sds_issued_on")
if sds is not None and sds.type == "date":
wanted.append(("SDS_REVIEW", sds.as_date + timedelta(days=365)))
for kind, due in wanted:
self.ensure_key_date(kind, due, owner_group, cadence_for(kind))
def ensure_key_date(self, kind: str, due: date, owner_group: str, days_before: tuple) -> KeyDateRow:
cur = next((k for k in self.key_dates if k.kind == kind and k.done_at is None), None)
if cur is not None:
if cur.due_on != due:
cur.due_on, cur.reminded = due, []
return cur
n = max((k.seq for k in self.key_dates if k.kind == kind), default=0) + 1
k = KeyDateRow(kind, n, due, owner_group, tuple(days_before))
self.key_dates.append(k)
return k
def close_key_dates(self, *, now: datetime, by: str, note: str) -> None:
for k in self.key_dates:
if k.done_at is None:
k.done_at, k.done_by, k.note = now, by, note
def retention_from(self, rule, today: date) -> Optional[date]:
"""Term + N years for an executed contract (12 for a deed); a dead draft after the rule's short window."""
if self.state in EXECUTED or self.state in (TERMINATED, SUPERSEDED):
end = self.ends_on or (self.signed_at.date() if self.signed_at else today)
try:
return end.replace(year=end.year + int(rule.years_after_end))
except ValueError: # 29 February
return end.replace(year=end.year + int(rule.years_after_end), day=28)
if self.state in (DECLINED, CANCELLED, EXPIRED_UNSIGNED):
return today + timedelta(days=int(rule.dead_draft_days))
return None
# ── when the clock next has something to do ─────────────────────────────────────────────
def compute_next_due(self, chase_next: Optional[datetime]) -> Optional[datetime]:
cands = []
for a in self.approvals:
if a.state == "pending" and a.draft_fingerprint == self.draft_fingerprint:
cands.append(_utc(a.due_at))
if self.state == OUT_FOR_SIGNATURE:
if self.link_expires_at:
cands.append(_utc(self.link_expires_at))
if chase_next:
cands.append(_utc(chase_next))
if self.all_signed:
cands.append(datetime(2000, 1, 1, tzinfo=timezone.utc)) # finish it: the copy is owed
for d in self.deliveries:
if d.status == "pending" and d.channel == "provider" and d.retry_at:
cands.append(_utc(d.retry_at))
self.next_due_at = min(cands) if cands else None
return self.next_due_at
# ── deliveries ───────────────────────────────────────────────────────────────────────────
def add_delivery(self, d: Delivery) -> Delivery:
same = [x for x in self.deliveries if x.signer_no == d.signer_no and x.kind == d.kind]
d.seq = max((x.seq for x in same), default=0) + 1
self.deliveries.append(d)
return d
def delivery(self, signer_no: int, kind: str, seq: int) -> Delivery:
d = next((x for x in self.deliveries if x.signer_no == int(signer_no) and x.kind == kind and x.seq == int(seq)),
None)
if d is None:
raise NotFound(f"no delivery {signer_no}/{kind}/{seq} on {self.contract_id}")
return d
def mark_delivered(self, signer_no: int, *, now: datetime, by: str, channel: str, provider_ref: Optional[str]) -> None:
s = self.signer(signer_no)
s.delivered_at = s.delivered_at or now
self.append_event("DELIVERED", now, s, version=self.current_version,
detail=f"{channel}{(': ' + provider_ref) if provider_ref else ''}"[:300])
if s.side == THEM and self.principal_statement_issued_at is None:
self.principal_statement_issued_at = now
def missing_error(missing: list) -> Incomplete:
return Incomplete("missing: " + ", ".join(missing), missing)
__all__ = ["Contract", "Section", "Signer", "Version", "Approval", "SignatureEvent", "ExecutedCopy", "Delivery",
"KeyDateRow", "AmendmentLink", "evidence_time", "missing_error"]