"""The Person aggregate: hr_person and everything dated about one human — their names (detail), engagements,
assignments, pay and pay components, personnel events, documents and compliance records.
Every personnel event is applied here, and after every one the invariants are checked (check()):
* TC1: the detail has exactly one slice on every date from the person's first; no gap, no overlap
* TC1 per engagement: the assignment covers the engagement from its start to its end, no gaps, no overlaps
* TC1 per engagement from the FIRST pay slice to the engagement's end (a hire may come before pay is known)
* at most ONE active engagement at a time in v1 (engagement periods never overlap)
* events are append-only, and each one writes at least one slice
* a freelancer engagement via a company names a supplier_bp_id; an employee engagement never does
* a correction keeps the old slice (superseded_by) and is itself a slice
The PersonFactory makes a Person + its first engagement from a Hired event, with person_id derived from the
PersonKey — so a second submit finds the same person and the same engagement, and is a no-op.
"""
from dataclasses import dataclass, field
from datetime import date, datetime
from decimal import Decimal
from typing import Optional
from erp_platform.money import Rate
from brick_hr.domain import events as ev
from brick_hr.domain.errors import Incomplete, RuleBroken
from brick_hr.domain.slices import Slice, Timeline
from brick_hr.domain.values import DAY, OPEN, PersonKey, normalise_email
DETAIL_FIELDS = ("legal_given_name", "legal_family_name", "preferred_name", "personal_email", "personal_phone",
"work_email", "work_phone")
ASSIGNMENT_FIELDS = ("org_id", "org_unit_id", "position_id", "job_code", "location_code", "schedule_code",
"employment_pct", "employee_subgroup", "cost_centre", "manager_override_person_id")
PAY_FIELDS = ("rate_basis", "rate_amount", "currency", "frequency", "pay_grade")
COMPONENT_FIELDS = ("rate_basis", "rate_amount", "currency")
RECORDS = ("detail", "assignment", "pay", "component", "compliance")
@dataclass
class Engagement:
n: int
worker_type: str
legal_entity_ref: str
start_date: date
planned_end: Optional[date] = None
actual_end: Optional[date] = None
leave_reason: Optional[str] = None
supplier_bp_id: Optional[int] = None
created_at: Optional[datetime] = None
@property
def end(self) -> date:
return self.actual_end or OPEN
def active_on(self, d: date) -> bool:
return self.start_date <= d <= self.end
def status(self, today: date) -> str:
"""future / active / ending / ended — DERIVED from the dates, never stored (HR-39)."""
if self.start_date > today:
return "future"
if self.actual_end is None:
return "active"
return "ending" if self.actual_end >= today else "ended"
@dataclass
class Document:
doc_id: str
document_type: str
access_class: str
filename: str
content_type: str
object_key: str
sha256: str
size_bytes: int
uploaded_by: str
uploaded_at: datetime
expires_on: Optional[date] = None
contract_id: Optional[str] = None
@dataclass
class HireSpec:
"""What a Hired / Rehired / Converted event carries for the new engagement."""
worker_type: str
start_date: date
legal_entity_ref: str
assignment: dict
planned_end: Optional[date] = None
rate: Optional[Rate] = None
frequency: Optional[str] = None
pay_grade: Optional[str] = None
supplier_bp_id: Optional[int] = None
event_type: str = ""
reason: Optional[str] = None
def _rate_row(r: Rate, frequency: Optional[str] = None, pay_grade: Optional[str] = None) -> dict:
return {"rate_basis": r.basis, "rate_amount": r.amount, "currency": r.currency,
"frequency": frequency, "pay_grade": pay_grade}
def rate_of(s: Optional[Slice]) -> Optional[Rate]:
if s is None or s.data.get("rate_amount") is None:
return None
return Rate(Decimal(str(s.data["rate_amount"])), s.data["rate_basis"], s.data["currency"])
class Person:
def __init__(self, person_id: int, person_key: str, *, staff_no: Optional[str] = None,
bp_id: Optional[int] = None, pool_user_sub: Optional[str] = None, version: int = 0,
created_at: Optional[datetime] = None):
self.person_id, self.person_key, self.staff_no = int(person_id), person_key, staff_no
self.bp_id, self.pool_user_sub, self.version, self.created_at = bp_id, pool_user_sub, version, created_at
self.detail = Timeline("the person's detail")
self.engagements: dict = {}
self.assignments: dict = {}
self.pay: dict = {}
self.components: dict = {}
self.events: list = []
self.documents: dict = {}
self.compliance: dict = {}
self.new_events: list = []
def __repr__(self) -> str:
return f"<Person {self.person_id} {self.person_key} engagements={sorted(self.engagements)}>"
# ── reading ─────────────────────────────────────────────────────────────────────────────
def engagement(self, n: int) -> Engagement:
e = self.engagements.get(int(n))
if e is None:
raise RuleBroken(f"person {self.person_id} has no engagement {n}")
return e
def engagement_on(self, d: date) -> Optional[Engagement]:
return next((e for e in self.engagements.values() if e.active_on(d)), None)
def open_engagement(self, today: date) -> Optional[Engagement]:
"""The engagement not ended as of `today` (future, active or ending), if any."""
live = [e for e in self.engagements.values() if e.end >= today]
return max(live, key=lambda e: e.n) if live else None
def latest(self) -> Optional[Engagement]:
return self.engagements[max(self.engagements)] if self.engagements else None
def relevant_engagement(self, d: date) -> Optional[Engagement]:
"""The engagement on `d`, else the open one, else the latest — what a view of 'now' shows."""
return self.engagement_on(d) or self.open_engagement(d) or self.latest()
def assignment_at(self, d: date) -> Optional[tuple]:
e = self.engagement_on(d)
if e is None:
return None
s = self.assignments.get(e.n, Timeline("")).at(d)
return (e.n, s) if s else None
def detail_at(self, d: date) -> Optional[Slice]:
return self.detail.at(d) or self.detail.first()
def name(self, d: date) -> str:
s = self.detail_at(d)
if s is None:
return f"person {self.person_id}"
return (s.data.get("preferred_name") or s.data.get("legal_given_name") or "") + " " + \
(s.data.get("legal_family_name") or "")
def work_email(self, d: date) -> str:
s = self.detail_at(d)
return (s.data.get("work_email") or "") if s else ""
def next_seq(self, n: int) -> int:
return max([e.seq for e in self.events if e.n == n] or [0]) + 1
def event(self, n: int, seq: int) -> Optional[ev.Event]:
return next((e for e in self.events if e.n == n and e.seq == seq), None)
def first_day(self) -> Optional[date]:
return min((e.start_date for e in self.engagements.values()), default=None)
# ── the factory ─────────────────────────────────────────────────────────────────────────
@classmethod
def from_hire(cls, key: PersonKey, detail: dict, spec: HireSpec, worker_type: dict, *, today: date,
now: datetime, by: str, staff_no: Optional[str] = None) -> "Person":
"""PersonFactory: a Person, its detail slice, engagement 1, the first assignment (+ pay), from a Hired event."""
p = cls(key.person_id, key.value, staff_no=(staff_no or None), created_at=now)
p.detail.start(min(spec.start_date, today), _clean_detail(detail), now=now, by=by, reason="hired")
p._new_engagement(1, spec, worker_type, now=now, by=by)
p._record(1, ev.HIRED, spec.event_type, spec.reason, spec.start_date, by, now,
p._written(1, spec), {"worker_type": spec.worker_type})
p.check()
return p
def _written(self, n: int, spec: HireSpec) -> list:
out = [f"engagement:{n}", f"assignment:{n}@{spec.start_date}"]
if spec.rate is not None:
out.append(f"pay:{n}@{spec.start_date}")
return out
def _new_engagement(self, n: int, spec: HireSpec, worker_type: dict, *, now: datetime, by: str) -> Engagement:
via_company = bool(worker_type.get("via_company"))
if via_company and not spec.supplier_bp_id:
raise Incomplete("a freelancer working through a company needs the company (finance pays the company)")
if not via_company and spec.supplier_bp_id:
raise RuleBroken(f"a {worker_type.get('label') or spec.worker_type} engagement is not via a company; "
"it names no supplier company")
if spec.planned_end is not None and spec.planned_end < spec.start_date:
raise Incomplete("a planned end before the start date")
for other in self.engagements.values():
if other.end >= spec.start_date and other.start_date <= (spec.planned_end or OPEN):
if other.actual_end is None:
raise RuleBroken(f"engagement {other.n} (since {other.start_date}) is still open; record the "
"leave first — one active engagement at a time")
raise RuleBroken(f"engagement {other.n} is still active on {spec.start_date:%-d %b}; start the new "
f"one on {other.actual_end + DAY:%-d %b}")
e = Engagement(n, spec.worker_type, spec.legal_entity_ref, spec.start_date, spec.planned_end,
supplier_bp_id=spec.supplier_bp_id, created_at=now)
self.engagements[n] = e
a = Timeline(f"engagement {n}'s assignment")
a.start(spec.start_date, _clean_assignment(spec.assignment), now=now, by=by, reason=spec.event_type)
self.assignments[n] = a
pay = Timeline(f"engagement {n}'s pay")
if spec.rate is not None:
pay.start(spec.start_date, _rate_row(spec.rate, spec.frequency, spec.pay_grade), now=now, by=by,
reason=spec.event_type)
self.pay[n] = pay
return e
def _record(self, n: int, kind: str, event_type: str, reason: Optional[str], effective: date, by: str,
now: datetime, written: list, payload: Optional[dict] = None) -> ev.Event:
if not written:
raise RuleBroken("a personnel event writes at least one slice")
seq = self.next_seq(n)
e = ev.Event(n, seq, event_type or kind, kind, reason, effective, by, now, list(written), None, None,
dict(payload or {}))
if e.direction is not None:
from brick_hr.domain.values import run_id
e.run_id = run_id(e.direction, self.person_id, n, seq)
self.events.append(e)
self.new_events.append(e)
return e
# ── personnel events ────────────────────────────────────────────────────────────────────
def same_hire(self, spec: HireSpec) -> Optional[Engagement]:
"""The engagement a repeated hire of the same starter lands on (double submit), or None."""
for e in self.engagements.values():
if e.start_date == spec.start_date and e.worker_type == spec.worker_type:
return e
return None
def rehire(self, spec: HireSpec, worker_type: dict, *, now: datetime, by: str) -> ev.Event:
"""A new engagement n+1 for a person who worked here before. History stays; offboarding never starts."""
if not self.engagements:
raise RuleBroken("this person has never been engaged; record a hire")
n = max(self.engagements) + 1
self._new_engagement(n, spec, worker_type, now=now, by=by)
e = self._record(n, ev.REHIRED, spec.event_type, spec.reason, spec.start_date, by, now,
self._written(n, spec), {"worker_type": spec.worker_type})
self.check()
return e
def convert(self, spec: HireSpec, worker_type: dict, *, now: datetime, by: str) -> ev.Event:
"""Converted: engagement n ends the day before (reason converted) and n+1 starts on the date, as ONE event."""
d = spec.start_date
old = self.engagement_on(d - DAY)
if old is None or old.end < d - DAY:
raise RuleBroken(f"nobody to convert: no engagement is active on {d - DAY}")
if old.actual_end is not None and old.actual_end != d - DAY:
raise RuleBroken(f"engagement {old.n} already ends {old.actual_end}; a conversion ends it on {d - DAY}")
if old.worker_type == spec.worker_type:
raise RuleBroken(f"engagement {old.n} is already {spec.worker_type}; a conversion changes the worker type")
old.actual_end, old.leave_reason = d - DAY, ev.CONVERTED_REASON
for t in self._series(old.n):
t.end_at(d - DAY, now=now)
n = max(self.engagements) + 1
self._new_engagement(n, spec, worker_type, now=now, by=by)
e = self._record(n, ev.CONVERTED, spec.event_type, spec.reason or ev.CONVERTED_REASON, d, by, now,
[f"engagement:{old.n}"] + self._written(n, spec),
{"ended_engagement": old.n, "from_worker_type": old.worker_type,
"worker_type": spec.worker_type})
self.check()
return e
def _series(self, n: int) -> list:
return [self.assignments[n], self.pay[n]] + [t for (m, _c), t in self.components.items() if m == n]
def leave(self, last_day: date, reason: str, event_type: str, *, now: datetime, by: str,
n: Optional[int] = None) -> ev.Event:
if not reason:
raise Incomplete("a leave needs a reason")
e = self.engagement(n) if n is not None else self.open_engagement(last_day) or self.engagement_on(last_day)
if e is None:
raise RuleBroken(f"no engagement is active on {last_day} to end")
if e.actual_end is not None:
raise RuleBroken(f"engagement {e.n} already ends {e.actual_end}; move the last day instead")
if last_day < e.start_date:
raise RuleBroken(f"a last day ({last_day}) before the start ({e.start_date}): cancel the onboarding run "
"instead — the person never started")
e.actual_end, e.leave_reason = last_day, reason
for t in self._series(e.n):
t.end_at(last_day, now=now)
out = self._record(e.n, ev.LEFT, event_type, reason, last_day, by, now, [f"engagement:{e.n}"],
{"last_day": str(last_day)})
self.check()
return out
def transfer(self, effective: date, values: dict, event_type: str, reason: Optional[str], *, now: datetime,
by: str) -> ev.Event:
e = self.engagement_on(effective)
if e is None:
raise RuleBroken(f"no engagement is active on {effective} to transfer")
clean = _clean_assignment(values, partial=True)
if not clean:
raise Incomplete("a transfer changes the org unit, job, position, location or schedule")
self.assignments[e.n].change(effective, clean, now=now, by=by, reason=event_type)
out = self._record(e.n, ev.TRANSFERRED, event_type, reason, effective, by, now,
[f"assignment:{e.n}@{effective}"], {"changed": sorted(clean)})
self.check()
return out
def change_pay(self, effective: date, reason: str, event_type: str, *, now: datetime, by: str,
rate: Optional[Rate] = None, frequency: Optional[str] = None, pay_grade: Optional[str] = None,
component: Optional[str] = None) -> ev.Event:
if not (reason or "").strip():
raise Incomplete("a pay change needs a reason")
if rate is None:
raise Incomplete("a pay change needs the new rate (amount, basis, currency)")
e = self.engagement_on(effective)
if e is None:
raise RuleBroken(f"no engagement is active on {effective} to pay")
if component:
t = self.components.setdefault((e.n, component), Timeline(f"engagement {e.n}'s {component}"))
row = {"rate_basis": rate.basis, "rate_amount": rate.amount, "currency": rate.currency}
written = f"pay_component:{e.n}:{component}@{effective}"
else:
t = self.pay[e.n]
row = _rate_row(rate, frequency, pay_grade)
written = f"pay:{e.n}@{effective}"
t.change(effective, row, now=now, by=by, reason=reason, until=e.end)
out = self._record(e.n, ev.PAY_CHANGED, event_type, reason, effective, by, now, [written],
{"component": component} if component else {})
self.check()
return out
def extend(self, new_end: date, event_type: str, reason: Optional[str], *, now: datetime, by: str,
n: Optional[int] = None) -> tuple:
"""ContractExtended: planned_end moves. If a leave is already recorded, the last day moves with it (the
slices are reopened to the new day). Returns (event, the last day before or None)."""
e = self.engagement(n) if n is not None else self.open_engagement(now.date())
if e is None:
raise RuleBroken("no open engagement to extend")
if new_end < e.start_date:
raise Incomplete("an extension to before the start date")
moved_from = None
if e.actual_end is not None:
if new_end <= e.actual_end:
raise RuleBroken(f"an extension moves the end later than {e.actual_end}")
moved_from = e.actual_end
for t in self._series(e.n):
t.reopen(e.actual_end, new_end)
e.actual_end = new_end
e.planned_end = new_end
out = self._record(e.n, ev.CONTRACT_EXTENDED, event_type, reason, new_end, by, now, [f"engagement:{e.n}"],
{"planned_end": str(new_end), "last_day_moved_from": str(moved_from) if moved_from else None})
self.check()
return out, moved_from
def move_start(self, n: int, new_start: date, *, now: datetime, by: str) -> date:
"""HR-08: the start date moves; the engagement and its first slices move with it. Returns the old date."""
e = self.engagement(n)
old = e.start_date
if new_start == old:
return old
for other in self.engagements.values():
if other.n != n and other.start_date <= new_start <= other.end:
raise RuleBroken(f"engagement {other.n} is still active on {new_start}")
if e.actual_end is not None and new_start > e.actual_end:
raise RuleBroken(f"a start after the last day ({e.actual_end})")
for t in [self.assignments[n], self.pay[n]] + [t for (m, _c), t in self.components.items() if m == n]:
t.move_start(old, new_start, now=now, by=by)
first = self.detail.first()
if first is not None and first.valid_from == old and len(self.detail.active()) == 1 and new_start < old:
self.detail.move_start(old, new_start, now=now, by=by)
e.start_date = new_start
self.check()
return old
def move_last_day(self, n: int, new_last: date, *, now: datetime, by: str) -> date:
e = self.engagement(n)
if e.actual_end is None:
raise RuleBroken(f"engagement {n} has no last day to move; record a leave")
old = e.actual_end
if new_last < e.start_date:
raise RuleBroken("a last day before the start date")
if new_last > old:
for t in self._series(n):
t.reopen(old, new_last)
else:
for t in self._series(n):
t.end_at(new_last, now=now)
e.actual_end = new_last
self.check()
return old
def correct(self, record: str, valid_from: date, values: dict, *, now: datetime, by: str, reason: str = "",
n: Optional[int] = None, component: Optional[str] = None) -> Slice:
"""A correction: a new slice with the same valid_from supersedes the old one, which stays readable."""
if record == "detail":
t, clean = self.detail, _clean_detail(values, partial=True)
elif record == "assignment":
t, clean = self.assignments[self.engagement(n).n], _clean_assignment(values, partial=True)
elif record == "pay":
t, clean = self.pay[self.engagement(n).n], _clean_pay(values)
elif record == "component":
t = self.components.get((self.engagement(n).n, component or ""))
if t is None:
raise RuleBroken(f"engagement {n} has no component {component!r}")
clean = _clean_pay(values, component=True)
else:
raise Incomplete(f"a correction is of {', '.join(r for r in RECORDS if r != 'compliance')}")
if not clean:
raise Incomplete("a correction changes at least one value")
s = t.correct(valid_from, clean, now=now, by=by, reason=reason or "correction")
self.check()
return s
def write_back_work_email(self, email: str, d: date, *, now: datetime, by: str) -> bool:
"""The account step made the work address: a new detail slice from `d` (HR-31). True when it changed."""
email = normalise_email(email, "work email")
cur = self.detail_at(d)
if cur is not None and (cur.data.get("work_email") or "") == email:
return False
if any(s.valid_from == d for s in self.detail.active()):
self.detail.correct(d, {"work_email": email}, now=now, by=by, reason="work email")
elif cur is not None and cur.valid_from > d:
self.detail.correct(cur.valid_from, {"work_email": email}, now=now, by=by, reason="work email")
else:
self.detail.change(d, {"work_email": email}, now=now, by=by, reason="work email")
self.check()
return True
def file_document(self, doc: Document) -> bool:
if doc.doc_id in self.documents:
return False
self.documents[doc.doc_id] = doc
return True
def record_compliance(self, n: int, kind: str, valid_from: date, data: dict, *, now: datetime, by: str) -> Slice:
self.engagement(n)
t = self.compliance.setdefault((n, kind), Timeline(f"engagement {n}'s {kind}"))
if any(s.valid_from == valid_from for s in t.active()):
return t.correct(valid_from, data, now=now, by=by, reason="corrected")
return t.change(valid_from, data, now=now, by=by, reason="recorded", merge=False)
def compliance_on(self, n: int, kind: str) -> Optional[Slice]:
t = self.compliance.get((n, kind))
return t.last() if t else None
# ── invariants ──────────────────────────────────────────────────────────────────────────
def check(self) -> None:
if self.engagements:
self.detail.check_tc1()
first = self.detail.first()
if first is None:
raise RuleBroken("a person needs their detail (names) from their first day")
es = sorted(self.engagements.values(), key=lambda e: e.start_date)
for a, b in zip(es, es[1:]):
if b.start_date <= a.end:
raise RuleBroken(f"engagements {a.n} and {b.n} overlap on {b.start_date}: one active engagement "
"at a time")
for e in es:
self.assignments[e.n].check_tc1(e.start_date, e.end)
pay = self.pay.get(e.n)
if pay is not None and pay.active():
if pay.first().valid_from < e.start_date:
raise RuleBroken(f"engagement {e.n}: pay from {pay.first().valid_from}, before the start")
pay.check_tc1(pay.first().valid_from, e.end)
for (m, c), t in self.components.items():
if m == e.n and t.active():
t.check_tc1()
if t.first().valid_from < e.start_date or t.last().valid_to > e.end:
raise RuleBroken(f"engagement {e.n}: component {c} runs outside the engagement")
for e in self.new_events:
if not e.slices_written:
raise RuleBroken("a personnel event writes at least one slice")
def _clean(values: dict, fields: tuple, what: str, partial: bool) -> dict:
unknown = sorted(set(values) - set(fields))
if unknown:
raise Incomplete(f"{what} has {', '.join(fields)}; not {', '.join(unknown)}")
return {k: (v.strip() if isinstance(v, str) else v) for k, v in values.items() if k in fields
and (not partial or v is not None)}
def _clean_detail(values: dict, partial: bool = False) -> dict:
out = _clean(values, DETAIL_FIELDS, "a person's detail", partial)
for k in ("personal_email", "work_email"):
if out.get(k):
out[k] = normalise_email(out[k], k)
if not partial:
for k in ("legal_given_name", "legal_family_name"):
if not out.get(k):
raise Incomplete(f"{k} is required")
return out
def _clean_assignment(values: dict, partial: bool = False) -> dict:
out = _clean(values, ASSIGNMENT_FIELDS, "an assignment", partial)
if out.get("employment_pct") is not None:
if isinstance(out["employment_pct"], float):
raise Incomplete("employment_pct is exact (\"80\"), never a float")
pct = Decimal(str(out["employment_pct"]))
if not (Decimal(0) < pct <= Decimal(100)):
raise Incomplete("employment_pct is more than 0 and at most 100")
out["employment_pct"] = pct.quantize(Decimal("0.01"))
if not partial:
for k in ("org_unit_id", "job_code"):
if not out.get(k):
raise Incomplete(f"an assignment needs {k}")
return out
def _clean_pay(values: dict, component: bool = False) -> dict:
from brick_hr.domain.values import rate as make_rate
out = _clean(values, COMPONENT_FIELDS if component else PAY_FIELDS, "a pay slice", True)
if any(k in out for k in ("rate_amount", "rate_basis", "currency")):
if not all(k in out for k in ("rate_amount", "rate_basis", "currency")):
raise Incomplete("a corrected rate gives amount, basis and currency together")
r = make_rate(out["rate_amount"], out["rate_basis"], out["currency"])
out.update(rate_amount=r.amount, rate_basis=r.basis, currency=r.currency)
return out
__all__ = ["Person", "Engagement", "Document", "HireSpec", "rate_of", "DETAIL_FIELDS", "ASSIGNMENT_FIELDS",
"PAY_FIELDS", "field"]