"""The standard paper: contract types, the clause library, templates and the settings rows (CT-01 … CT-04, CT-02).
Every write here changes CONFIGURATION, so it bumps the config token (ct_config_version) in the same transaction:
the next request on any replica rebuilds its snapshot once; every other request reuses it.
"""
from typing import Annotated, Literal, Optional
from pydantic import BaseModel, Field
from erp_platform.bricks import Refused
from erp_platform.types import Limit, Name, text
from brick_contracts.application import support as S
from brick_contracts.application.use_cases import CT
from brick_contracts.domain import packs
from brick_contracts.domain.contract_type import ContractType
from brick_contracts.domain.errors import Invalid, RuleBroken
from brick_contracts.domain.library import Clause, Template, VariableDef
from brick_contracts.domain.policies import (ApprovalPolicy, cadence_from_row, consumer_from_row, retention_from_row,
rule_from_row)
class SlotIn(BaseModel):
slot_no: int = Field(ge=1, le=500)
family: S.Code
kind: Literal["required", "optional", "alternative_of"] = "required"
clause_id: S.Code
clause_version: int = Field(ge=1, le=100000)
alternative_of_slot: Optional[int] = Field(default=None, ge=1, le=500)
condition: Optional[dict] = None
class VariableIn(BaseModel):
name: S.Code
type: Literal["text", "date", "money", "rate", "int", "enum", "bool", "party"] = "text"
source: Literal["COUNTERPARTY", "ENGAGEMENT", "PAY", "DATES", "QUOTE", "CALLER", "TEMPLATE"] = "CALLER"
required: bool = True
label: Optional[Name] = None
validation: dict = Field(default_factory=dict)
enum_values: list[S.Code] = Field(default_factory=list, max_length=100)
aliases: dict[str, str] = Field(default_factory=dict)
condition: Optional[dict] = None
default_value: Optional[text(4000)] = None
currency_var: S.Code = "currency"
SettingsKind = Literal["approval_policy", "reminder_cadence", "consent_text", "retention_rule", "setting", "consumer"]
def _clause_json(c: Clause, usage: Optional[dict] = None) -> dict:
return {"clause_id": c.clause_id, "family": c.family, "position": c.position, "deviation_level": c.deviation_level,
"category": c.category, "status": c.status, "current_version": c.current_version,
"versions": [{"version": v.version, "title": v.title, "body": v.body, "created_by": v.created_by,
"created_at": S.iso(v.created_at), "on_contracts": (usage or {}).get(v.version, 0)}
for v in sorted(c.versions.values(), key=lambda v: v.version)]}
def _template_json(t: Template) -> dict:
return {"template_id": t.template_id, "type_code": t.type_code, "name": t.name,
"current_version": (t.current_approved().version if t.current_approved() else None),
"versions": [{"version": v.version, "status": v.status, "authored_by": v.authored_by,
"approved_by": v.approved_by, "approved_at": S.iso(v.approved_at), "note": v.note,
"slots": [s.as_json() for s in v.ordered_slots()],
"variables": [d.as_json() for d in sorted(v.variables.values(), key=lambda d: d.name)]}
for v in sorted(t.versions.values(), key=lambda v: v.version)]}
# ── contract types ─────────────────────────────────────────────────────────────────────────
@CT.use_case(kind="command", route="POST /types", permission=S.TYPE_WRITE, idempotent=True)
def keep_contract_type(caller: dict, type_code: S.Code, label: Name, allowed_bp_roles: list[S.Code],
retention_rule: S.Code, approval_policy: S.Code, execution: Literal["SIMPLE", "DEED"] = "SIMPLE",
esign_allowed: bool = True, sensitivity: Literal["commercial", "employment"] = "commercial",
pack: Optional[S.Code] = None, pack_checks: Optional[list[S.Code]] = None,
allow_roleless_person: bool = False, status: Literal["active", "retired"] = "active") -> dict:
"""Keep a contract type (CT-02): who it may be with (business-partner roles), whether it may be signed online,
whether its body is employment-sensitive, how long it is kept and who approves its deviations. A DEED is saved
with esign_allowed = false and can never be sent for online signing. The retention rule and the approval policy
(with its fallback group) must exist."""
at = S.now()
with S.rules(), uow() as u:
cfg = u.config.load()
ct = ContractType.define({"type_code": type_code, "label": label, "allowed_bp_roles": list(allowed_bp_roles),
"execution": execution, "esign_allowed": esign_allowed, "sensitivity": sensitivity,
"retention_rule": retention_rule, "approval_policy": approval_policy, "pack": pack,
"pack_checks": list(pack_checks or []), "allow_roleless_person": allow_roleless_person,
"status": status}, cfg)
u.types.save(ct, by=S.actor(caller), now=at)
u.config.bump(at)
return ct.as_json()
# ── clauses ────────────────────────────────────────────────────────────────────────────────
@CT.use_case(kind="command", route="POST /clauses", permission=S.CLAUSE_WRITE, idempotent=True)
def keep_clause(caller: dict, clause_id: S.Code, family: S.Code, position: Literal["PREFERRED", "ALTERNATE", "FALLBACK"],
title: S.Short, body: S.Words, deviation_level: S.Level = 0, category: S.Code = "",
version: Optional[S.VersionNo] = None) -> dict:
"""Keep a clause of the library (CT-01): its family (liability cap, IP), its position (PREFERRED is the standard,
ALTERNATE an agreed other wording, FALLBACK acceptable with approval) and its words. A new clause is version 1.
Changing the words of a version a contract holds, or an approved template pins, is refused — amend_clause cuts a
new version instead; old versions stay readable."""
at = S.now()
with S.rules(), uow() as u:
c = u.clauses.get(clause_id.lower())
if c is None:
c = Clause.new(clause_id, family, position, deviation_level, title, body, category=category,
by=S.actor(caller), at=at)
else:
if c.family != family.lower():
raise RuleBroken(f"{c.clause_id} is in the {c.family} family; a clause does not change family")
c.set_place(position, deviation_level)
c.category = category or c.category
n = int(version or c.current_version)
cur = c.versions.get(n)
if cur is None:
raise Invalid(f"{c.clause_id} has no version {n}")
if (title.strip(), body.strip()) != (cur.title, cur.body):
pins = [p for t in u.templates.using_clause(c.clause_id) for p in t.pins(c.clause_id, n)]
c.edit_version(n, title, body, contracts_holding=u.contracts.clause_usage(c.clause_id, n),
templates_pinning=pins)
u.clauses.save(c, now=at)
u.config.bump(at)
return _clause_json(c, u.contracts.clause_usage_by_version(c.clause_id))
@CT.use_case(kind="command", route="POST /clauses/{clause_id}/versions", permission=S.CLAUSE_WRITE, idempotent=True)
def amend_clause(caller: dict, clause_id: S.Code, body: S.Words, title: Optional[S.Short] = None) -> dict:
"""Amend a clause's words into a NEW version (CT-01). Contracts already holding older versions are untouched —
each holds its own copy — and templates keep pinning the version they were approved with. The same words twice
is one version."""
at = S.now()
with S.rules(), uow() as u:
c = u.clauses.get(clause_id.lower())
if c is None:
raise S.not_found(f"clause {clause_id}")
c.amend(title, body, by=S.actor(caller), at=at)
u.clauses.save(c, now=at)
u.config.bump(at)
return _clause_json(c, u.contracts.clause_usage_by_version(c.clause_id))
@CT.use_case(kind="command", route="POST /clauses/{clause_id}/retire", permission=S.CLAUSE_WRITE, idempotent=True)
def retire_clause(caller: dict, clause_id: S.Code) -> dict:
"""Retire a clause: no new template may use it. Refused while a template version still REQUIRES it."""
at = S.now()
with S.rules(), uow() as u:
c = u.clauses.get(clause_id.lower())
if c is None:
raise S.not_found(f"clause {clause_id}")
c.retire([p for t in u.templates.using_clause(c.clause_id) for p in t.requires(c.clause_id)])
u.clauses.save(c, now=at)
u.config.bump(at)
return _clause_json(c)
@CT.use_case(kind="query", route="GET /clauses")
def clause_library(caller: dict, family: Optional[S.Code] = None) -> dict:
"""The clause library by family: each clause's position, deviation level and every version with how many
contracts hold it (Usage), and the templates. Standard wording, readable by everyone signed in."""
with uow() as u:
clauses = [c for c in u.clauses.all() if family is None or c.family == family.lower()]
families: dict = {}
for c in clauses:
families.setdefault(c.family, []).append(_clause_json(c, u.contracts.clause_usage_by_version(c.clause_id)))
templates = [{"template_id": t.template_id, "type_code": t.type_code, "name": t.name,
"current_version": t.current_approved().version if t.current_approved() else None}
for t in u.templates.all()]
return {"families": [{"family": f, "clauses": cs} for f, cs in sorted(families.items())],
"templates": templates}
# ── templates ──────────────────────────────────────────────────────────────────────────────
@CT.use_case(kind="command", route="POST /templates", permission=S.TEMPLATE_WRITE, idempotent=True)
def keep_template(caller: dict, template_id: S.Code, type_code: S.Code, name: Name,
slots: Annotated[list[SlotIn], Field(max_length=200)],
variables: Optional[Annotated[list[VariableIn], Field(max_length=300)]] = None,
note: Optional[S.Reason] = None) -> dict:
"""Keep a template's DRAFT version (CT-03): its ordered slots (required, optional, alternative-of another slot),
each pinning clause_id@version, and its typed variables (type, source, required, bounds, the condition that
chooses a clause, the template's standard value). Writes the open draft version, or a new version N+1 when every
version is approved — an approved version is never changed. It drafts nothing until approve_template."""
at = S.now()
with S.rules(), uow() as u:
cfg = u.config.load()
if type_code.lower() not in cfg.types:
raise Invalid(f"there is no contract type {type_code!r}")
t = u.templates.get(template_id.lower())
if t is None:
t = Template.new(template_id, type_code, name)
elif t.type_code != type_code.lower():
raise RuleBroken(f"{t.template_id} is the {t.type_code} template; a template does not change type")
t.name = name.strip() or t.name
variables = variables or []
for v in variables:
VariableDef.from_input(v.model_dump())
n = t.keep_draft([s.model_dump() for s in slots], [v.model_dump() for v in variables], by=S.actor(caller),
at=at, note=note or "")
u.templates.save(t, now=at)
u.config.bump(at)
return {"template_id": t.template_id, "version": n, "status": t.versions[n].status} | {"template": _template_json(t)}
@CT.use_case(kind="command", route="POST /templates/{template_id}/versions/{version}/approve",
permission=S.TEMPLATE_APPROVE, idempotent=True)
def approve_template(caller: dict, template_id: S.Code, version: S.VersionNo) -> dict:
"""Approve a template version (CT-04): it becomes immutable and new drafts of its type use it; contracts on an
earlier version keep theirs. Every slot must pin an existing, active clause version of its family, and the
author may not approve their own version unless the setting author_may_approve allows it."""
at = S.now()
with S.rules(), uow() as u:
cfg = u.config.load()
t = u.templates.get(template_id.lower())
if t is None or int(version) not in t.versions:
raise S.not_found(f"template {template_id} version {version}")
if t.versions[int(version)].status != "APPROVED":
t.approve(version, by=S.actor(caller), at=at, clause_lookup=u.clauses.get,
author_may_approve=cfg.bool_setting("author_may_approve"))
u.templates.save(t, now=at)
u.config.bump(at)
return _template_json(t)
@CT.use_case(kind="command", route="POST /templates/{template_id}/versions/{version}/retire",
permission=S.TEMPLATE_WRITE, idempotent=True)
def retire_template(caller: dict, template_id: S.Code, version: S.VersionNo) -> dict:
"""Retire an approved template version: no new drafts on it; contracts drafted from it are unaffected."""
at = S.now()
with S.rules(), uow() as u:
t = u.templates.get(template_id.lower())
if t is None or int(version) not in t.versions:
raise S.not_found(f"template {template_id} version {version}")
t.retire(version, at=at)
u.templates.save(t, now=at)
u.config.bump(at)
return _template_json(t)
@CT.use_case(kind="query", route="GET /templates/{template_id}")
def show_template(caller: dict, template_id: S.Code) -> dict:
"""One template: every version with its status, slots (clause_id@version) and variable definitions."""
with uow() as u:
t = u.templates.get(template_id.lower())
if t is None:
raise S.not_found(f"template {template_id}")
return _template_json(t)
# ── settings ───────────────────────────────────────────────────────────────────────────────
def _check_rows(kind: str, rows: list, u) -> list:
out = []
if kind == "approval_policy":
by: dict = {}
for r in rows:
r = {"policy_code": str(r.get("policy_code") or "").strip().lower(), "rule_no": int(r.get("rule_no") or 0),
"applies_to": str(r.get("applies_to") or "*"), "approver_group": str(r.get("approver_group") or "").lower(),
"fallback_group": str(r.get("fallback_group") or "").lower(),
"timeout_working_days": int(r.get("timeout_working_days") or 2)}
if not r["policy_code"] or r["rule_no"] < 1:
raise Invalid("an approval rule has a policy_code and a rule_no ≥ 1")
by.setdefault(r["policy_code"], []).append(rule_from_row(r))
out.append(r)
for code, rs in by.items():
ApprovalPolicy(code, tuple(rs))
elif kind == "reminder_cadence":
import json
for r in rows:
c = cadence_from_row(r)
out.append({"cadence_code": c.code, "kind": c.kind, "every_days": c.every_days, "max_count": c.max_count,
"first_after_days": c.first_after_days, "unit": c.unit,
"days_before": json.dumps(list(c.days_before))})
elif kind == "consent_text":
stored = {r["version"]: r["text"] for r in u.config.rows("consent_text")}
for r in rows:
v, t = str(r.get("version") or "").strip(), str(r.get("text") or "").strip()
if not v or len(t) < 20:
raise Invalid("a consent text has a version and the words a signer agrees to")
if v in stored and stored[v] != t:
raise RuleBroken(f"consent text {v} is what signers agreed to; write a new version instead")
out.append({"version": v, "text": t})
elif kind == "retention_rule":
for r in rows:
x = retention_from_row(r)
out.append({"rule_code": x.code, "years_after_end": x.years_after_end, "dead_draft_days": x.dead_draft_days,
"basis": x.basis})
elif kind == "consumer":
import json
for r in rows:
c = consumer_from_row(r)
out.append({"namespace": c.namespace, "principals": json.dumps(list(c.principals)),
"owner_group": c.owner_group, "label": c.label})
else:
for r in rows:
k = str(r.get("key") or r.get("setting_key") or "").strip()
if not k or len(k) > 100:
raise Invalid("a setting has a key")
v = r.get("value")
out.append({"setting_key": k, "value": None if v is None else str(v)[:4000]})
return out
@CT.use_case(kind="command", route="POST /settings/{kind}", permission=S.SETTINGS, idempotent=True)
def keep_settings(caller: dict, kind: SettingsKind, rows: S.Rows) -> dict:
"""Keep settings rows: approval policies (which deviation needs whose approval, a MANDATORY fallback group and
the timeout; each policy has a catch-all rule '*'), reminder cadences (signer chasing; key-date reminders such as
120/90/60/30 days), consent texts (versioned; a version signers agreed to never changes), retention rules (term
+ N years; how long dead drafts linger), consumers (which service identities speak for a source_ref namespace —
hr:, crm:) and settings (link lifetime, signed-copy link lifetime, our legal name, the mail sender, default
signatories, author_may_approve)."""
at = S.now()
with S.rules(), uow() as u:
clean = _check_rows(kind, rows, u)
u.config.save_rows(kind, clean, now=at)
u.config.bump(at)
return {"kind": kind, "rows": len(clean)}
@CT.use_case(kind="command", route="POST /localisation-packs/{pack}", permission=S.SETTINGS, idempotent=True)
def install_contract_pack(caller: dict, pack: S.Code) -> dict:
"""Install a localisation pack's rows (gb: the UK contract types — employment with the ERA 1996 s.1 checks,
freelance with the IR35 check, a deed type that is never signed online — retention of term + 6 years, deeds 12,
the consent text, cadences, the approval policy, the hr and crm consumers). Only rows that are not there yet are
added: your own configuration is never overwritten. Wording (clauses, templates) is yours to keep."""
at = S.now()
module = packs.PACKS.get(pack.lower())
if module is None:
raise Refused(404, f"no localisation pack {pack!r} (have: {', '.join(sorted(packs.PACKS))})")
added: dict = {}
with S.rules(), uow() as u:
seed = module.seed()
for kind in ("retention_rule", "approval_policy", "reminder_cadence", "consent_text", "setting", "consumer"):
key = {"approval_policy": ("policy_code", "rule_no"), "reminder_cadence": ("cadence_code",),
"consent_text": ("version",), "retention_rule": ("rule_code",), "setting": ("setting_key",),
"consumer": ("namespace",)}[kind]
have = {tuple(r[k] for k in key) for r in u.config.rows(kind)}
rows = [r if kind != "setting" else {"setting_key": r["key"], "value": r["value"]} for r in seed[kind]]
fresh = [r for r in rows if tuple(r[k] for k in key) not in have]
if kind == "approval_policy":
taken = {r[0] for r in have}
fresh = [r for r in fresh if r["policy_code"] not in taken]
clean = _check_rows(kind, fresh, u) if fresh else []
u.config.save_rows(kind, clean, now=at)
added[kind] = len(clean)
u.config.bump(at)
with S.rules(), uow() as u: # the types read the policies and rules just written
cfg = u.config.load()
have = set(cfg.types)
n = 0
for row in seed["contract_type"]:
if row["type_code"] not in have:
u.types.save(ContractType.define(row, cfg), by=S.actor(caller), now=at)
n += 1
added["contract_type"] = n
u.config.bump(at)
return {"pack": pack.lower(), "rows": {k: len(v) for k, v in seed.items()},
"note": "rows already there were kept as they are"}
@CT.use_case(kind="query", route="GET /settings", permission=S.SETTINGS)
def show_settings(caller: dict, limit: Limit = 500) -> dict:
"""Every configuration row: types, approval policies, cadences, consent texts, retention rules, consumers and
settings — what the Settings screen edits."""
with uow() as u:
cfg = u.config.load()
return {"types": [t.as_json() for t in sorted(cfg.types.values(), key=lambda t: t.type_code)][:limit],
"approval_policies": [{"policy_code": p.code, "rules": [r.__dict__ for r in p.rules]}
for p in cfg.policies.values()],
"cadences": [c.__dict__ | {"days_before": list(c.days_before)} for c in cfg.cadences.values()],
"consent_texts": [{"version": c.version, "text": c.text} for c in cfg.consents.values()],
"retention_rules": [r.__dict__ for r in cfg.retention.values()],
"consumers": [c.__dict__ | {"principals": list(c.principals)} for c in cfg.consumers.values()],
"settings": dict(sorted(cfg.settings.items())),
"templates": {k: [list(p) for p in v] for k, v in cfg.templates.items()}}
def uow():
from brick_contracts.application import use_cases
return use_cases.repository()
__all__ = ["keep_contract_type", "keep_clause", "amend_clause", "retire_clause", "clause_library", "keep_template",
"approve_template", "retire_template", "show_template", "keep_settings", "install_contract_pack",
"show_settings", "uow"]