"""Value objects and the vocabulary of states and kinds. Immutable, compared by value, no database.
Fingerprint sha256 hex of a canonical serialisation (a draft, a PDF, an executed file)
VariableValue a typed value + its source + when it was resolved; money as an exact decimal with its currency
KeyDate kind + due_on + owner group + reminder cadence (days before)
ClausePosition PREFERRED | ALTERNATE | FALLBACK with its deviation level, or OVERRIDE (non-standard, level 2)
SignerRef contract_id|signer_no — the subject of a signing link
EventHash row_hash = sha256(prev_hash ‖ canonical row); the first prev_hash = sha256(contract_id)
ConsentText the exact words shown before signing, with their version
contract_id derived: 'ct-' + hex(blake2b(source_ref|type_code, 8)), so a retried caller makes ONE contract
"""
import hashlib
import json
import re
from dataclasses import dataclass, field
from datetime import date, datetime, timezone
from decimal import Decimal, InvalidOperation
from typing import Any, Optional
from brick_contracts.domain.errors import Invalid
# ── contract states (modules/contracts workflows) ───────────────────────────────────────────
DRAFT, IN_APPROVAL, APPROVED, OUT_FOR_SIGNATURE = "DRAFT", "IN_APPROVAL", "APPROVED", "OUT_FOR_SIGNATURE"
SIGNED, SIGNED_ELSEWHERE, TERMINATED, SUPERSEDED = "SIGNED", "SIGNED_ELSEWHERE", "TERMINATED", "SUPERSEDED"
DECLINED, CHANGES_REQUESTED, EXPIRED_UNSIGNED, CANCELLED = "DECLINED", "CHANGES_REQUESTED", "EXPIRED_UNSIGNED", "CANCELLED"
ACTIVE, EXPIRED = "ACTIVE", "EXPIRED" # DERIVED from the term on a signed contract; never stored
EDITABLE = (DRAFT, IN_APPROVAL, APPROVED)
EXECUTED = (SIGNED, SIGNED_ELSEWHERE)
DEAD = (DECLINED, EXPIRED_UNSIGNED, CANCELLED)
STATES = (DRAFT, IN_APPROVAL, APPROVED, OUT_FOR_SIGNATURE, SIGNED, SIGNED_ELSEWHERE, ACTIVE, EXPIRED, TERMINATED,
SUPERSEDED, DECLINED, CHANGES_REQUESTED, EXPIRED_UNSIGNED, CANCELLED)
# ── signers and evidence ────────────────────────────────────────────────────────────────────
US, THEM = "US", "THEM"
INTERNAL, EXTERNAL = "INTERNAL", "EXTERNAL"
EMAIL_LINK, POOL_SESSION = "EMAIL_LINK", "POOL_SESSION"
S_PENDING, S_SIGNED, S_DECLINED, S_CHANGES, S_ELSEWHERE = "PENDING", "SIGNED", "DECLINED", "CHANGES_REQUESTED", "SIGNED_ELSEWHERE"
EVENT_KINDS = ("INVITED", "DELIVERED", "OPENED", "CONSENTED", "SIGNED", "DECLINED", "CHANGES_REQUESTED", "REVOKED",
"REISSUED", "SIGNED_ELSEWHERE", "RECALLED", "EXECUTED")
# ── the library ─────────────────────────────────────────────────────────────────────────────
PREFERRED, ALTERNATE, FALLBACK, OVERRIDE = "PREFERRED", "ALTERNATE", "FALLBACK", "OVERRIDE"
POSITIONS = (PREFERRED, ALTERNATE, FALLBACK)
T_DRAFT, T_APPROVED, T_RETIRED = "DRAFT", "APPROVED", "RETIRED"
SLOT_KINDS = ("required", "optional", "alternative_of")
VAR_TYPES = ("text", "date", "money", "rate", "int", "enum", "bool", "party")
SOURCES = ("COUNTERPARTY", "ENGAGEMENT", "PAY", "DATES", "QUOTE", "CALLER", "TEMPLATE")
SIMPLE, DEED = "SIMPLE", "DEED"
COMMERCIAL, EMPLOYMENT = "commercial", "employment"
AMENDMENT_KINDS = ("AMENDMENT", "RENEWAL", "TERMINATION", "ADMINISTRATIVE")
KEY_DATE_KINDS = ("START", "END", "NOTICE_DEADLINE", "RENEWAL", "PROBATION_END", "PRINCIPAL_STATEMENT_DUE",
"SDS_REVIEW", "PAYMENT_MILESTONE")
def canonical(value: Any) -> str:
"""The one serialisation hashes are taken over: sorted keys, no spaces, exact decimals and ISO dates as text."""
return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False, default=_plain)
def _plain(v: Any) -> Any:
if isinstance(v, Decimal):
return str(v)
if isinstance(v, datetime):
return utc_iso(v)
if isinstance(v, date):
return v.isoformat()
return str(v)
def utc_iso(t: Optional[datetime]) -> Optional[str]:
if t is None:
return None
if t.tzinfo is None:
t = t.replace(tzinfo=timezone.utc)
return t.astimezone(timezone.utc).isoformat(timespec="microseconds").replace("+00:00", "Z")
def sha256_hex(data: bytes) -> str:
return hashlib.sha256(data).hexdigest()
@dataclass(frozen=True)
class Fingerprint:
"""sha256 hex of a canonical serialisation. Compared by value; shown to a signer and an approver."""
hex: str
@classmethod
def of(cls, value: Any) -> "Fingerprint":
return cls(sha256_hex(canonical(value).encode()))
@classmethod
def of_bytes(cls, data: bytes) -> "Fingerprint":
return cls(sha256_hex(data))
def __str__(self) -> str:
return self.hex
@property
def short(self) -> str:
return self.hex[:12]
def contract_id_for(source_ref: str, type_code: str) -> str:
"""'ct-' + 16 hex of blake2b(source_ref|type_code): the same request names the same contract."""
seed = f"{source_ref.strip()}|{type_code.strip().lower()}".encode()
return "ct-" + hashlib.blake2b(seed, digest_size=8).hexdigest()
def namespace_of(source_ref: str) -> str:
"""'hr:onboarding:…#3' → 'hr'. The originating consumer of a contract."""
head, sep, _ = (source_ref or "").partition(":")
return head.strip().lower() if sep else ""
@dataclass(frozen=True)
class SignerRef:
contract_id: str
signer_no: int
def __str__(self) -> str:
return f"{self.contract_id}|{self.signer_no}"
@classmethod
def parse(cls, text: str) -> "SignerRef":
cid, sep, n = (text or "").rpartition("|")
if not sep or not n.isdigit() or not cid:
raise Invalid(f"{text!r} is not a signer reference (contract_id|signer_no)")
return cls(cid, int(n))
@dataclass(frozen=True)
class EventHash:
"""row_hash = sha256(prev_hash ‖ canonical row). The chain's first prev_hash is sha256(contract_id)."""
prev_hash: str
row_hash: str
@staticmethod
def genesis(contract_id: str) -> str:
return sha256_hex(contract_id.encode())
@classmethod
def chain(cls, prev_hash: str, row: dict) -> "EventHash":
return cls(prev_hash, sha256_hex((prev_hash + canonical(row)).encode()))
@dataclass(frozen=True)
class ConsentText:
version: str
text: str
@dataclass(frozen=True)
class ClausePosition:
"""Where a section's words stand against the standard: a library position with its level, or OVERRIDE (2)."""
kind: str
deviation_level: int
def __post_init__(self) -> None:
if self.kind not in POSITIONS + (OVERRIDE,):
raise Invalid(f"a position is one of {', '.join(POSITIONS)} (or OVERRIDE), not {self.kind!r}")
if self.deviation_level not in (0, 1, 2):
raise Invalid("a deviation level is 0 (standard), 1 (needs approval) or 2 (non-standard)")
if self.kind == PREFERRED and self.deviation_level != 0:
raise Invalid("a Preferred clause is the standard: deviation level 0")
if self.kind == FALLBACK and self.deviation_level < 1:
raise Invalid("a Fallback clause is acceptable with approval: deviation level 1 or more")
if self.kind == OVERRIDE and self.deviation_level != 2:
raise Invalid("an override written for one contract is non-standard: deviation level 2")
@classmethod
def override(cls) -> "ClausePosition":
return cls(OVERRIDE, 2)
@property
def deviates(self) -> bool:
return self.deviation_level > 0
@dataclass(frozen=True)
class KeyDate:
"""A dated obligation: NOTICE_DEADLINE = end − notice days is computed and kept as its own date."""
kind: str
seq: int
due_on: date
owner_group: str
days_before: tuple = ()
def __post_init__(self) -> None:
if self.kind not in KEY_DATE_KINDS:
raise Invalid(f"a key date is one of {', '.join(KEY_DATE_KINDS)}, not {self.kind!r}")
def reminder_slots(self) -> list:
"""(days_before, the date it falls on), soonest first."""
from datetime import timedelta
return sorted(((d, self.due_on - timedelta(days=d)) for d in self.days_before), key=lambda x: x[1])
# ── typed variable values ───────────────────────────────────────────────────────────────────
_DECIMAL = re.compile(r"-?\d{1,16}(\.\d{1,4})?")
_CURRENCY = re.compile(r"[A-Z]{3}")
TEXT_LIMIT = 4000
@dataclass(frozen=True)
class VariableValue:
"""A resolved variable: `value` is its canonical text (ISO date, exact decimal, upper-case code, 'true'),
`currency` set for money and rate. Built only through `parse`, so an invalid value never exists."""
name: str
type: str
value: str
source: str
currency: Optional[str] = None
resolved_at: Optional[str] = None
@classmethod
def parse(cls, name: str, type_: str, raw: Any, source: str, *, enum_values: tuple = (), aliases: Optional[dict] = None,
currency: Optional[str] = None, resolved_at: Optional[str] = None) -> "VariableValue":
if type_ not in VAR_TYPES:
raise Invalid(f"{name}: a variable type is one of {', '.join(VAR_TYPES)}, not {type_!r}")
if raw is None or (isinstance(raw, str) and not raw.strip()):
raise Invalid(f"{name}: no value")
text = str(raw).strip() if not isinstance(raw, bool) else ("true" if raw else "false")
if type_ in ("text", "party"):
if len(text) > TEXT_LIMIT:
raise Invalid(f"{name}: at most {TEXT_LIMIT} characters")
out = text
elif type_ == "date":
try:
d = date.fromisoformat(text[:10])
except ValueError:
raise Invalid(f"{name}: {text!r} is not a date (YYYY-MM-DD)") from None
if not date(1900, 1, 1) <= d <= date(2199, 12, 31):
raise Invalid(f"{name}: {d} is outside 1900 … 2199")
out = d.isoformat()
elif type_ in ("money", "rate"):
if isinstance(raw, float):
raise Invalid(f"{name}: send an amount as an exact decimal string (\"750.00\"), not a float")
if not _DECIMAL.fullmatch(text):
raise Invalid(f"{name}: {text!r} is not an exact amount (up to 16 digits and 4 places)")
q = Decimal("0.01") if type_ == "money" else Decimal("0.0001")
amount = Decimal(text)
if type_ == "money" and amount != amount.quantize(q):
raise Invalid(f"{name}: money has at most 2 decimal places, not {text}")
out = str(amount.quantize(q))
cur = (currency or "").strip().upper()
if not _CURRENCY.fullmatch(cur):
raise Invalid(f"{name}: an amount needs its currency (ISO 4217, e.g. GBP)")
currency = cur
elif type_ == "int":
if isinstance(raw, float) or not re.fullmatch(r"-?\d{1,9}", text):
raise Invalid(f"{name}: {text!r} is not a whole number")
out = str(int(text))
elif type_ == "bool":
low = text.lower()
if low not in ("true", "false", "yes", "no", "1", "0"):
raise Invalid(f"{name}: {text!r} is not true or false")
out = "true" if low in ("true", "yes", "1") else "false"
else: # enum
key = text.strip()
mapped = (aliases or {}).get(key) or (aliases or {}).get(key.lower()) or key.upper()
if enum_values and mapped not in enum_values:
raise Invalid(f"{name}: {text!r} is not one of {', '.join(enum_values)}")
out = mapped
return cls(name, type_, out, source, currency if type_ in ("money", "rate") else None, resolved_at)
@property
def amount(self) -> Optional[Decimal]:
if self.type in ("money", "rate", "int"):
try:
return Decimal(self.value)
except InvalidOperation:
return None
return None
@property
def as_date(self) -> Optional[date]:
return date.fromisoformat(self.value) if self.type == "date" else None
def display(self) -> str:
"""How it reads in the contract's words."""
if self.type == "date":
d = self.as_date
return f"{d.day} {d.strftime('%B')} {d.year}"
if self.type in ("money", "rate"):
whole, _, frac = self.value.partition(".")
sign = "-" if whole.startswith("-") else ""
digits = whole.lstrip("-")
grouped = f"{int(digits):,}" if digits else "0"
return f"{self.currency} {sign}{grouped}" + (f".{frac}" if frac else "")
if self.type == "bool":
return "Yes" if self.value == "true" else "No"
return self.value
def as_json(self) -> dict:
out = {"name": self.name, "type": self.type, "value": self.value, "source": self.source}
if self.currency:
out["currency"] = self.currency
return out
def canonical_row(self) -> dict:
return {"name": self.name, "type": self.type, "value": self.value, "currency": self.currency}