"""People: the personnel events (hire, leave, transfer, pay, extension, rehire, conversion), corrections, date
moves, the sign-in invite, and the reads of a person (show_person, list_people, show_my_record).
Each command writes ONE aggregate — the Person — and commits; the run engine then expands the run and syncs the
business partner in later transactions (eventual consistency, each step idempotent on its own key).
"""
from datetime import date, datetime, timedelta
from decimal import Decimal
from typing import Literal, Optional
from pydantic import BaseModel, ConfigDict
from erp_platform import after_response
from erp_platform.bricks import Refused
from erp_platform.types import BigId, Email, Limit, Name, Note
from erp_platform.repository import CONFLICT_REASON, as_conflict
from brick_hr.application import engine
from brick_hr.application import support as S
from brick_hr.application import use_cases as uc
from brick_hr.domain import events as ev
from brick_hr.domain import services
from brick_hr.domain.errors import Incomplete, RuleBroken
from brick_hr.domain.intake import intake_id, status as intake_status
from brick_hr.domain.person import HireSpec, Person, rate_of
from brick_hr.domain.run import RunStep, WAITING
from brick_hr.domain.values import OPEN, PersonKey, rate as make_rate, run_id as make_run_id
from brick_hr.infrastructure import business_partner as bp
HR = uc.HR
class Company(BaseModel):
"""The company a freelancer works through: finance pays the company (its own business partner)."""
model_config = ConfigDict(extra="forbid")
legal_name: str
company_number: Optional[str] = None
country: str = "GB"
vat_number: Optional[str] = None
def _check_version(p: Person, expected: Optional[int]) -> None:
if expected is not None and int(expected) != p.version:
raise Refused(409, CONFLICT_REASON)
def _org_for(view: S.View, org_id: Optional[str], org_unit_id: str, d: date) -> str:
orgs = view.orgs
if org_id:
if org_id not in orgs:
raise Refused(422, f"no org {org_id!r}")
return org_id
default = view.config.setting("default_org_id")
if default and default in orgs:
return default
owning = [o.org_id for o in orgs.values() if o.unit_at(org_unit_id, d) is not None]
if len(owning) == 1:
return owning[0]
raise Refused(422, f"which org tree is {org_unit_id!r} in? give org_id" if owning else
f"no org unit {org_unit_id!r} on {d}")
def _assignment(view: S.View, *, org_id, org_unit_id, job_code, position_id, location_code, schedule_code,
employment_pct, cost_centre, d: date, manager_override_person_id=None) -> dict:
cfg = view.config
oid = _org_for(view, org_id, org_unit_id, d)
org = view.orgs[oid]
unit = org.unit_at(org_unit_id, d)
if unit is None:
raise Refused(422, f"there is no org unit {org_unit_id!r} on {d}")
if cfg.job(job_code) is None:
raise Refused(422, f"there is no job {job_code!r} in the catalogue")
if position_id and org.position_at(position_id, d) is None:
raise Refused(422, f"there is no position {position_id!r} on {d}")
if location_code and cfg.location(location_code) is None:
raise Refused(422, f"there is no location {location_code!r}")
if schedule_code and cfg.schedule(schedule_code) is None:
raise Refused(422, f"there is no work schedule {schedule_code!r}")
pos = org.position_at(position_id, d) if position_id else None
out = {"org_id": oid, "org_unit_id": org_unit_id, "job_code": job_code, "position_id": position_id,
"location_code": location_code or unit.location_code, "schedule_code": schedule_code,
"employment_pct": employment_pct, "cost_centre": cost_centre or (pos.cost_centre if pos else None)
or unit.cost_centre}
if manager_override_person_id:
out["manager_override_person_id"] = int(manager_override_person_id)
return out
def _override_same_org(view: S.View, assignment: dict, d: date) -> None:
m = assignment.get("manager_override_person_id")
if not m:
return
seat = view.directory.seat(int(m), d)
if seat is None or seat.org_id != assignment.get("org_id"):
raise Refused(409, "a manager override names a person assigned in the SAME org tree on that date; across "
"org trees it is refused in v1")
def _rate(rate_amount, rate_basis, currency, view: S.View):
if rate_amount in (None, ""):
return None
r = make_rate(rate_amount, rate_basis, currency or "GBP")
rb = {row["rate_basis"] for row in view.config.all("rate_basis", active_only=True)}
if rb and r.basis not in rb:
raise Refused(422, f"{r.basis} is not one of your rate bases ({', '.join(sorted(rb))})")
return r
def _event_type(view: S.View, code: str, kind: str, reason: Optional[str]) -> None:
et = view.config.event_type(code)
if et is None or not et.get("active", True):
raise Refused(422, f"there is no event type {code!r}")
if et["kind"] != kind:
raise Refused(422, f"event type {code!r} is a {et['kind']}, not a {kind}")
if reason:
r = view.config.reason(code, reason)
if r is None or not r.get("active", True):
raise Refused(422, f"{reason!r} is not a reason for {code!r}")
def _after(person_id: int, events: list, at: datetime, person: Optional[Person] = None) -> None:
"""After the Person committed: expand each new event's run, sync the partner. Each idempotent. `person` is
the Person exactly as this request just committed it (its version included), so expansion need not read it
back — reading it again would return the same rows."""
for e in events:
if e.run_id:
engine.expand(person_id, e.n, e.seq, at, person=person)
# The partner is another brick: after the answer (erp_platform 0.9.6), never inside it. Safe to lose: the event
# keeps bp_synced_at empty until it lands, and the tick's sync_due settles it.
after_response(engine.sync_partner, person_id, at)
def _save_new(view, uow, p: Person) -> None:
uow.persons.save(p)
def _hire_answer(p: Person, n: int, today: date) -> dict:
e = p.engagements[n]
hired = next((x for x in p.events if x.n == n and x.seq == 1), None)
return {"person_id": p.person_id, "person_key": p.person_key, "n": n, "worker_type": e.worker_type,
"start_date": e.start_date.isoformat(), "bp_id": p.bp_id or bp.person_bp_id(p.person_id),
"supplier_bp_id": e.supplier_bp_id, "run_id": hired.run_id if hired else None,
"intake_id": intake_id(p.person_id, n)}
# ── hire ───────────────────────────────────────────────────────────────────────────────────
@HR.use_case(kind="command", route="POST /hires", permission=S.HIRE, idempotent=True)
def record_hire(caller: dict, legal_given_name: Name, legal_family_name: Name, worker_type: Name, job_code: Name,
org_unit_id: Name, start_date: S.Day, personal_email: Optional[Email] = None,
staff_no: Optional[Name] = None, preferred_name: Optional[Name] = None,
personal_phone: Optional[Name] = None, org_id: Optional[Name] = None, position_id: Optional[Name] = None,
location_code: Optional[Name] = None, schedule_code: Optional[Name] = None,
employment_pct: Optional[Name] = None, cost_centre: Optional[Name] = None,
planned_end: Optional[S.Day] = None, legal_entity_ref: Optional[Name] = None,
rate_amount: Optional[S.RateText] = None, rate_basis: Optional[Name] = None, currency: Optional[Name] = None,
pay_grade: Optional[Name] = None, company: Optional[Company] = None, event_type: Name = "hire",
reason: Optional[Note] = None, first_project: Optional[Name] = None) -> dict:
"""Ask for a starter with one form (HR-01, a Hired event): legal name, preferred name, personal email (or a
staff number), worker type, job, org unit, start date and, for a freelancer, a day rate or fixed fee (and the
company when via a company). Creates ONE person (person_id derived from the staff number or the normalised
personal email), its business partner (name and work contact only), engagement 1, the first assignment slice
and pay slice, and one onboarding run from the matching playbook. Submitting twice gives the same person,
engagement and run. A start more than the configured days back is refused (a correction); a second open
engagement for the same person is refused, naming it; a company freelancer with no company is refused."""
at = S.now()
def attempt():
with S.rules():
with uc.repository() as uow:
return _hire_once(uow, caller, at, locals_)
locals_ = dict(legal_given_name=legal_given_name, legal_family_name=legal_family_name, worker_type=worker_type,
job_code=job_code, org_unit_id=org_unit_id, start_date=start_date, personal_email=personal_email,
staff_no=staff_no, preferred_name=preferred_name, personal_phone=personal_phone, org_id=org_id,
position_id=position_id, location_code=location_code, schedule_code=schedule_code,
employment_pct=employment_pct, cost_centre=cost_centre, planned_end=planned_end,
legal_entity_ref=legal_entity_ref, rate_amount=rate_amount, rate_basis=rate_basis,
currency=currency, pay_grade=pay_grade, company=company, event_type=event_type, reason=reason,
first_project=first_project)
try:
p, answer = attempt()
except Exception as exc: # noqa: BLE001
# Two submits of one hire at the same moment: the version claim lets one commit; the other reloads and
# finds the same hire, so both answer the same (SC-HR-21).
if as_conflict(exc) is None:
raise
p, answer = attempt()
if p is not None:
_after(p.person_id, p.events, at, person=p)
else:
_after(answer["person_id"], [], at)
engine.expand(answer["person_id"], answer["n"], 1, at)
return answer
def _hire_once(uow, caller: dict, at: datetime, f: dict) -> tuple:
view = S.View(uow, caller, at)
cfg = view.config
worker_type, start_date, company = f["worker_type"], f["start_date"], f["company"]
wt = cfg.worker_type(worker_type)
if wt is None or not wt.get("active", True):
raise Refused(422, f"there is no worker type {worker_type!r}")
_event_type(view, f["event_type"], ev.HIRED, f["reason"])
back = cfg.int_setting("max_backdate_days")
if (view.today - start_date).days > back:
raise Refused(409, f"start dates more than {back} days back are corrections, ask HR")
assignment = _assignment(view, org_id=f["org_id"], org_unit_id=f["org_unit_id"], job_code=f["job_code"],
position_id=f["position_id"], location_code=f["location_code"],
schedule_code=f["schedule_code"], employment_pct=f["employment_pct"],
cost_centre=f["cost_centre"], d=start_date)
if not S.has(caller, S.PERSON_WRITE):
S.manager_line_check(view, f["org_unit_id"], assignment["org_id"], start_date)
key = PersonKey.of(f["staff_no"], f["personal_email"])
supplier = None
if company is not None:
if not wt.get("via_company"):
raise Refused(409, f"a {wt['label']} engagement is not via a company; leave the company out")
supplier = bp.bp_id_for(bp.company_key(company.model_dump()))
entity = f["legal_entity_ref"] or cfg.setting("default_legal_entity") or \
view.orgs[assignment["org_id"]].legal_entity_ref
view.require_company(entity, f"legal entity {entity}") # hr v4: a grant hires into its companies only
spec = HireSpec(worker_type, start_date, entity, assignment, f["planned_end"],
_rate(f["rate_amount"], f["rate_basis"], f["currency"], view), None, f["pay_grade"], supplier,
f["event_type"], f["reason"])
existing = uow.persons.get(key.person_id, likely_new=True)
if existing is not None:
same = existing.same_hire(spec)
if same is not None:
return None, _hire_answer(existing, same.n, view.today)
live = existing.open_engagement(view.today)
if live is not None:
raise Refused(409, f"{existing.name(view.today).strip()} already has engagement {live.n} "
f"(since {live.start_date}) open — record a leave or a conversion")
raise Refused(409, f"this person worked here before (engagement {existing.latest().n} ended "
f"{existing.latest().actual_end}); record a rehire")
detail = {"legal_given_name": f["legal_given_name"], "legal_family_name": f["legal_family_name"],
"preferred_name": f["preferred_name"] or f["legal_given_name"], "personal_email": f["personal_email"],
"personal_phone": f["personal_phone"]}
p = Person.from_hire(key, detail, spec, wt, today=view.today, now=at, by=S.actor(caller), staff_no=f["staff_no"])
p.bp_id = bp.person_bp_id(p.person_id)
p.events[0].payload.update({k: v for k, v in (("first_project", f["first_project"]),
("company", company.model_dump() if company else None)) if v})
uow.persons.save(p)
return p, _hire_answer(p, 1, view.today)
# ── leave, transfer, pay, extension, rehire, conversion ────────────────────────────────────
def _load_visible(uow, caller, person_id: int, at: datetime) -> tuple:
view = S.View(uow, caller, at)
return view, view.person(person_id)
@HR.use_case(kind="command", route="POST /people/{person_id}/leave", permission=S.LEAVE, if_match="expected_version")
@S.retry_unless_versioned
def record_leave(caller: dict, person_id: BigId, last_day: S.Day, reason: Note, event_type: Name = "leave",
takeover_person_id: Optional[BigId] = None, expected_version: Optional[int] = None) -> dict:
"""Record a leaver with their last day and a reason from the list (HR-06, a Left event): actual_end is set,
assignment and pay are delimited at last day + 1, and one offboarding run starts from the playbook for that
worker type and reason — unless the reason is 'converted'. A handover draft lists what they look after; the
takeover person (a run party) gets each item. With the UK pack, a last day giving less than the statutory
minimum notice is WARNED, never refused."""
at = S.now()
with S.rules():
with uc.repository() as uow:
view, p = _load_visible(uow, caller, person_id, at)
_check_version(p, expected_version)
_event_type(view, event_type, ev.LEFT, reason)
e = p.open_engagement(last_day) or p.engagement_on(last_day)
warnings = []
if e is not None:
wt = view.config.worker_type(e.worker_type) or {}
if wt.get("category") == "employee" and view.config.setting("pack") == "gb":
w = services.notice_warning(e.start_date, last_day, view.today)
if w:
warnings.append(w)
out = p.leave(last_day, reason, event_type, now=at, by=S.actor(caller))
uow.persons.save(p)
_after(person_id, [out], at, person=p) # as committed: the run is expanded without reading it back
if out.run_id and takeover_person_id:
_set_party(out.run_id, "takeover", int(takeover_person_id), S.actor(caller))
return {"person_id": person_id, "n": out.n, "last_day": last_day.isoformat(), "run_id": out.run_id,
"warnings": warnings}
@HR.use_case(kind="command", route="POST /people/{person_id}/transfer", permission=S.EVENT, if_match="expected_version")
@S.retry_unless_versioned
def record_transfer(caller: dict, person_id: BigId, effective_date: S.Day, org_unit_id: Optional[Name] = None,
job_code: Optional[Name] = None, position_id: Optional[Name] = None,
location_code: Optional[Name] = None, schedule_code: Optional[Name] = None,
employment_pct: Optional[Name] = None, cost_centre: Optional[Name] = None,
manager_override_person_id: Optional[BigId] = None, reason: Optional[Note] = None,
event_type: Name = "transfer", expected_version: Optional[int] = None) -> dict:
"""Record a transfer to another team, job, position, location or schedule from a date (HR-17, Transferred):
a new assignment slice delimits the old one; a change run carries the access delta (the new unit's groups and
spaces added, the old ones removed) timed on the event date. A transfer into a unit outside your line is
refused unless you hold the destination's chief position too."""
at = S.now()
with S.rules():
with uc.repository() as uow:
view, p = _load_visible(uow, caller, person_id, at)
_check_version(p, expected_version)
_event_type(view, event_type, ev.TRANSFERRED, reason)
cur = p.assignment_at(effective_date)
if cur is None:
raise Refused(409, f"no engagement is active on {effective_date} to transfer")
base = cur[1].data
values = {"org_unit_id": org_unit_id, "job_code": job_code, "position_id": position_id,
"location_code": location_code, "schedule_code": schedule_code,
"employment_pct": employment_pct, "cost_centre": cost_centre,
"manager_override_person_id": manager_override_person_id}
values = {k: v for k, v in values.items() if v is not None}
if org_unit_id or job_code or position_id:
full = _assignment(view, org_id=base.get("org_id"), org_unit_id=org_unit_id or base.get("org_unit_id"),
job_code=job_code or base.get("job_code"),
position_id=position_id if position_id is not None else None,
location_code=location_code, schedule_code=schedule_code or base.get("schedule_code"),
employment_pct=employment_pct, cost_centre=cost_centre, d=effective_date,
manager_override_person_id=manager_override_person_id)
values.update({k: full[k] for k in ("org_id", "org_unit_id", "job_code", "location_code",
"cost_centre")})
if org_unit_id and not position_id:
values["position_id"] = None
if not S.has(caller, S.PERSON_WRITE) and org_unit_id:
S.manager_line_check(view, org_unit_id, full["org_id"], effective_date)
_override_same_org(view, {**base, **values}, effective_date)
out = p.transfer(effective_date, values, event_type, reason, now=at, by=S.actor(caller))
uow.persons.save(p)
_after(person_id, [out], at, person=p) # as committed: the run is expanded without reading it back
return {"person_id": person_id, "n": out.n, "effective_date": effective_date.isoformat(), "run_id": out.run_id}
@HR.use_case(kind="command", route="POST /people/{person_id}/pay", permission=S.PAY_WRITE, if_match="expected_version")
@S.retry_unless_versioned
def record_pay_change(caller: dict, person_id: BigId, effective_date: S.Day, reason: Note, rate_amount: S.RateText,
rate_basis: Name, currency: Name = "GBP", component: Optional[Name] = None,
frequency: Optional[Name] = None, pay_grade: Optional[Name] = None,
event_type: Name = "pay_change", expected_version: Optional[int] = None) -> dict:
"""Record pay, or a pay component, with an effective date (HR-14, PayChanged): a slice from that date
delimits the one before; the reason is required; a rate is exact to 4 places with its basis (hour, day,
month, year, fixed) from the rows, never converted. Pass expected_version from show_person: if someone saved
since, it is refused 409 'changed since you opened it' and nothing is written."""
at = S.now()
with S.rules():
with uc.repository() as uow:
view, p = _load_visible(uow, caller, person_id, at)
_check_version(p, expected_version)
_event_type(view, event_type, ev.PAY_CHANGED, None)
if not (reason or "").strip():
raise Refused(422, "a pay change needs a reason")
r = _rate(rate_amount, rate_basis, currency, view)
if component:
ct = view.config.component(component)
e = p.engagement_on(effective_date)
if ct is None:
raise Refused(422, f"there is no pay component {component!r}")
if e is not None and ct.get("worker_types") and e.worker_type not in ct["worker_types"]:
raise Refused(409, f"{component} is not paid to a {e.worker_type}")
out = p.change_pay(effective_date, reason, event_type, now=at, by=S.actor(caller), rate=r,
frequency=frequency, pay_grade=pay_grade, component=component)
uow.persons.save(p)
version = p.version
return {"person_id": person_id, "n": out.n, "effective_date": effective_date.isoformat(), "version": version}
@HR.use_case(kind="command", route="POST /people/{person_id}/extension", permission=S.EVENT, if_match="expected_version")
@S.retry_unless_versioned
def record_contract_extension(caller: dict, person_id: BigId, new_end: S.Day, reason: Optional[Note] = None,
event_type: Name = "extension", expected_version: Optional[int] = None) -> dict:
"""Extend a fixed-term engagement (ContractExtended): planned_end moves. If a leave is already recorded, the
last day moves too and the open offboarding run's undone steps are re-timed; done steps stay as they were."""
at = S.now()
with S.rules():
with uc.repository() as uow:
view, p = _load_visible(uow, caller, person_id, at)
_check_version(p, expected_version)
_event_type(view, event_type, ev.CONTRACT_EXTENDED, reason)
out, moved_from = p.extend(new_end, event_type, reason, now=at, by=S.actor(caller))
uow.persons.save(p)
if moved_from:
_retime(person_id, out.n, "last_day", new_end, moved_from, S.actor(caller), at)
engine.sync_partner(person_id, at)
return {"person_id": person_id, "n": out.n, "planned_end": new_end.isoformat(),
"last_day_moved_from": moved_from.isoformat() if moved_from else None}
def _new_engagement_spec(view, *, worker_type, start_date, job_code, org_unit_id, org_id, position_id,
location_code, schedule_code, employment_pct, cost_centre, planned_end, legal_entity_ref,
rate_amount, rate_basis, currency, pay_grade, company, event_type, reason, base=None):
cfg = view.config
wt = cfg.worker_type(worker_type)
if wt is None:
raise Refused(422, f"there is no worker type {worker_type!r}")
base = base or {}
a = _assignment(view, org_id=org_id or base.get("org_id"), org_unit_id=org_unit_id or base.get("org_unit_id"),
job_code=job_code or base.get("job_code"), position_id=position_id,
location_code=location_code or base.get("location_code"),
schedule_code=schedule_code or base.get("schedule_code"), employment_pct=employment_pct,
cost_centre=cost_centre, d=start_date)
supplier = None
if company is not None:
if not wt.get("via_company"):
raise Refused(409, f"a {wt['label']} engagement is not via a company; leave the company out")
supplier = bp.bp_id_for(bp.company_key(company.model_dump()))
entity = legal_entity_ref or cfg.setting("default_legal_entity") or view.orgs[a["org_id"]].legal_entity_ref
view.require_company(entity, f"legal entity {entity}")
return wt, HireSpec(worker_type, start_date, entity, a, planned_end, _rate(rate_amount, rate_basis, currency, view),
None, pay_grade, supplier, event_type, reason)
@HR.use_case(kind="command", route="POST /people/{person_id}/rehire", permission=S.HIRE, idempotent=True)
def record_rehire(caller: dict, person_id: BigId, start_date: S.Day, worker_type: Name, job_code: Name, org_unit_id: Name,
org_id: Optional[Name] = None, position_id: Optional[Name] = None,
location_code: Optional[Name] = None, schedule_code: Optional[Name] = None,
employment_pct: Optional[Name] = None, cost_centre: Optional[Name] = None,
planned_end: Optional[S.Day] = None, legal_entity_ref: Optional[Name] = None,
rate_amount: Optional[S.RateText] = None, rate_basis: Optional[Name] = None, currency: Optional[Name] = None,
pay_grade: Optional[Name] = None, company: Optional[Company] = None, event_type: Name = "rehire",
reason: Optional[Note] = None) -> dict:
"""Rehire someone who worked here before (HR-15, Rehired): the same person gets engagement n+1; one onboarding
run from the rehire playbook starts; NO offboarding run is created by anything in a rehire; the old
engagement, documents and events stay readable. A start while the previous engagement is still active is
refused, saying the first day the new one can start. The same rehire twice is one."""
at = S.now()
with S.rules():
with uc.repository() as uow:
view, p = _load_visible(uow, caller, person_id, at)
_event_type(view, event_type, ev.REHIRED, reason)
wt, spec = _new_engagement_spec(view, worker_type=worker_type, start_date=start_date, job_code=job_code,
org_unit_id=org_unit_id, org_id=org_id, position_id=position_id,
location_code=location_code, schedule_code=schedule_code,
employment_pct=employment_pct, cost_centre=cost_centre,
planned_end=planned_end, legal_entity_ref=legal_entity_ref,
rate_amount=rate_amount, rate_basis=rate_basis, currency=currency,
pay_grade=pay_grade, company=company, event_type=event_type, reason=reason)
same = next((e for e in p.engagements.values() if e.start_date == start_date and e.n > 1
and e.worker_type == worker_type), None)
if same is not None:
out = next(x for x in p.events if x.n == same.n and x.kind == ev.REHIRED)
fresh = False
else:
out = p.rehire(spec, wt, now=at, by=S.actor(caller))
if company is not None:
out.payload["company"] = company.model_dump()
uow.persons.save(p)
fresh = True
if fresh:
_after(person_id, [out], at, person=p)
return {"person_id": person_id, "n": out.n, "start_date": start_date.isoformat(), "run_id": out.run_id}
@HR.use_case(kind="command", route="POST /people/{person_id}/conversion", permission=S.EVENT, idempotent=True)
def record_conversion(caller: dict, person_id: BigId, effective_date: S.Day, worker_type: Name,
job_code: Optional[Name] = None, org_unit_id: Optional[Name] = None, org_id: Optional[Name] = None,
position_id: Optional[Name] = None, location_code: Optional[Name] = None,
schedule_code: Optional[Name] = None, employment_pct: Optional[Name] = None,
cost_centre: Optional[Name] = None, legal_entity_ref: Optional[Name] = None,
rate_amount: Optional[S.RateText] = None, rate_basis: Optional[Name] = None,
currency: Optional[Name] = None, pay_grade: Optional[Name] = None,
event_type: Name = "conversion", reason: Optional[Note] = None) -> dict:
"""Convert a freelancer to an employee in one step (HR-16, Converted): engagement n ends the day before
(reason converted) and n+1 starts on the date, as ONE event in one transaction; ONE conversion run starts
(contract, payroll set-up, partner roles worker → employee, supplier removed for a sole trader); no
offboarding or access-removal step runs. The same conversion twice is one."""
at = S.now()
with S.rules():
with uc.repository() as uow:
view, p = _load_visible(uow, caller, person_id, at)
_event_type(view, event_type, ev.CONVERTED, reason)
done = next((x for x in p.events if x.kind == ev.CONVERTED and x.effective_date == effective_date), None)
if done is not None:
out, fresh = done, False
else:
cur = p.assignment_at(effective_date - timedelta(days=1))
base = cur[1].data if cur else {}
wt, spec = _new_engagement_spec(view, worker_type=worker_type, start_date=effective_date,
job_code=job_code, org_unit_id=org_unit_id, org_id=org_id,
position_id=position_id, location_code=location_code,
schedule_code=schedule_code, employment_pct=employment_pct,
cost_centre=cost_centre, planned_end=None,
legal_entity_ref=legal_entity_ref, rate_amount=rate_amount,
rate_basis=rate_basis, currency=currency, pay_grade=pay_grade,
company=None, event_type=event_type, reason=reason, base=base)
out = p.convert(spec, wt, now=at, by=S.actor(caller))
uow.persons.save(p)
fresh = True
if fresh:
_after(person_id, [out], at, person=p)
return {"person_id": person_id, "n": out.n, "ended_engagement": out.payload.get("ended_engagement"),
"effective_date": effective_date.isoformat(), "run_id": out.run_id}
@HR.use_case(kind="command", route="POST /people/{person_id}/corrections", permission=S.PERSON_WRITE, if_match="expected_version")
@S.retry_unless_versioned
def correct_slice(caller: dict, person_id: BigId, record: Literal["detail", "assignment", "pay", "component"],
valid_from: S.Day, values: dict, n: Optional[int] = None, component: Optional[Name] = None,
reason: Optional[Note] = None, expected_version: Optional[int] = None) -> dict:
"""Correct a slice (a correction, not a change): a new slice with the SAME valid_from supersedes the old one,
which stays readable (superseded_by), so 'what did we believe on date X' is answerable. Pay needs
hr:pay:write as well."""
at = S.now()
with S.rules():
if record in ("pay", "component") and not S.has(caller, S.PAY_WRITE):
raise Refused(403, f"missing permission: {S.PAY_WRITE}")
with uc.repository() as uow:
view, p = _load_visible(uow, caller, person_id, at)
_check_version(p, expected_version)
if record != "detail" and n is None:
e = p.relevant_engagement(valid_from)
n = e.n if e else None
s = p.correct(record, valid_from, values, now=at, by=S.actor(caller), reason=reason or "", n=n,
component=component)
uow.persons.save(p)
version = p.version
if record == "detail":
engine.sync_partner(person_id, at)
return {"person_id": person_id, "record": record, "valid_from": valid_from.isoformat(), "rev": s.rev,
"version": version}
def _retime(person_id: int, n: int, anchor: str, new: date, old: date, by: str, at: datetime) -> list:
"""Re-time every open run of the engagement from a moved anchor; only steps not done move (Retimer)."""
moved = []
with uc.repository() as uow:
ids = [r for r in uow.runs.ids(person_id)]
for rid in ids:
def go(rid=rid):
with uc.repository() as uow:
run = uow.runs.get(rid)
if run is None or run.n != n or run.cancelled_at is not None or run.done:
return None
if anchor not in run.anchors and anchor != "start_date":
return None
view = S.View(uow, engine.SYSTEM, at)
p = uow.persons.get(person_id)
run.anchors[anchor] = new
changed = run.retime(run.anchors, engine._calendar(view, p, n), view.tz)
run.add_log(f"{anchor} moved", by, at, old, new)
uow.runs.save(run)
return (rid, len(changed))
r = engine._committed(go)
if r:
moved.append(r)
return moved
@HR.use_case(kind="command", route="POST /people/{person_id}/move-date", permission=S.RUN_RESCHEDULE, if_match="expected_version")
@S.retry_unless_versioned
def move_event_date(caller: dict, person_id: BigId, which: Literal["start_date", "last_day"], new_date: S.Day,
n: Optional[int] = None, expected_version: Optional[int] = None) -> dict:
"""Move a start date or a last day (HR-08): the engagement and its slices move with it, and every open run's
NOT-done steps get a new due time from their timing rules (working days and the org's timezone respected);
done steps keep their due time and result. The run log says who moved it, from what to what, and when."""
at = S.now()
with S.rules():
with uc.repository() as uow:
view, p = _load_visible(uow, caller, person_id, at)
_check_version(p, expected_version)
e = p.engagement(n) if n is not None else p.open_engagement(view.today) or p.latest()
if e is None:
raise Refused(409, "no engagement to move")
if which == "start_date":
old = p.move_start(e.n, new_date, now=at, by=S.actor(caller))
else:
old = p.move_last_day(e.n, new_date, now=at, by=S.actor(caller))
uow.persons.save(p)
moved = _retime(person_id, e.n, which, new_date, old, S.actor(caller), at) if old != new_date else []
engine.sync_partner(person_id, at)
return {"person_id": person_id, "n": e.n, "which": which, "from": old.isoformat(), "to": new_date.isoformat(),
"runs_retimed": [{"run_id": r, "steps": c} for r, c in moved]}
# ── sign-in invite (HR-25) ─────────────────────────────────────────────────────────────────
@HR.use_case(kind="command", route="POST /people/{person_id}/invite", permission=S.POOL_INVITE, idempotent=True)
def invite_to_sign_in(caller: dict, person_id: BigId) -> dict:
"""Invite a person to sign in (HR-25): an action step asks identity.invite_app_user with their work email and
the pool (idempotency key invite:<person_id>, so inviting twice reuses the same user); the sub it returns is
stored on the person. With no provider it is a manual task: an admin creates the pool user with the exact
input shown and records the sub when completing it."""
import os
at = S.now()
rid = make_run_id("invite", person_id, 0, 0)
with S.rules():
with uc.repository() as uow:
view, p = _load_visible(uow, caller, person_id, at)
email = p.work_email(view.today)
if not email:
raise Refused(409, "they have no work email yet — the account step writes it; invite them after")
if p.pool_user_sub:
return {"person_id": person_id, "run_id": rid, "pool_user_sub": p.pool_user_sub, "state": "done"}
exists = uow.runs.exists(rid)
if not exists:
d = p.detail_at(view.today).data
pool = view.config.setting("user_pool") or os.environ.get("DODIL_USER_POOL", "")
from brick_hr.domain.run import Run
run = Run(rid, "invite", person_id, (p.relevant_engagement(view.today).n if p.engagements else 0), "0",
None, None, at)
inp = {"email": email, "pool": pool, "given_name": d.get("legal_given_name"),
"family_name": d.get("legal_family_name"), "app_roles": ["employee"]}
run.steps["1"] = RunStep("1", 1, "action", f"Invite {p.name(view.today).strip()} to sign in", "hr",
capability="identity.invite_app_user", template=inp, input=inp,
due_at=at, timing={"anchor": "run.created", "offset": 0, "unit": "days",
"time": None})
engine.mark_manual(run)
run.add_log("created", S.actor(caller), at, None, "invite")
uow.runs.save(run)
engine.advance(rid, at)
with uc.repository() as uow:
run = uow.runs.get(rid)
s = run.steps["1"]
return {"person_id": person_id, "run_id": rid, "state": s.state, "manual": s.manual}
# ── reading ────────────────────────────────────────────────────────────────────────────────
def _slices(t, money: bool = False) -> list:
out = []
for s in sorted(t.slices, key=lambda s: (s.valid_from, s.rev)):
data = {k: (str(v) if isinstance(v, Decimal) else (v.isoformat() if isinstance(v, date) else v))
for k, v in s.data.items() if v is not None}
out.append({"valid_from": s.valid_from.isoformat(), "valid_to": S.due_label(s.valid_to) or "open",
"rev": s.rev, "active": s.active, "superseded_by": s.superseded_by,
"recorded_at": S.iso(s.recorded_at), "superseded_at": S.iso(s.superseded_at),
"recorded_by": s.recorded_by, "reason": s.reason, **data})
return out
def person_view(view: S.View, p: Person) -> dict:
caller, today = view.caller, view.today
d = p.detail_at(today)
e = p.relevant_engagement(today)
seat = p.assignment_at(today)
manager = view.relevant_manager(p.person_id)
names = view.uow.persons.names([manager], today) if manager else {}
engagements = []
for x in sorted(p.engagements.values(), key=lambda x: x.n):
engagements.append({"n": x.n, "worker_type": x.worker_type, "legal_entity_ref": x.legal_entity_ref,
"start_date": x.start_date.isoformat(), "planned_end": S.due_label(x.planned_end),
"actual_end": S.due_label(x.actual_end), "leave_reason": x.leave_reason,
"supplier_bp_id": x.supplier_bp_id, "status": x.status(today),
"assignments": _slices(p.assignments[x.n])})
out = {"person_id": p.person_id, "person_key": p.person_key, "staff_no": p.staff_no, "version": p.version,
"bp_id": p.bp_id, "signed_in": bool(p.pool_user_sub),
"name": p.name(today).strip(), "detail": dict((d.data if d else {})),
"detail_history": _slices(p.detail), "engagements": engagements,
"current": ({"n": seat[0], **{k: (str(v) if isinstance(v, Decimal) else v) for k, v in seat[1].data.items()}}
if seat else None),
"manager": ({"person_id": manager, "name": names.get(manager, {}).get("name")} if manager else None),
"events": [{"n": x.n, "seq": x.seq, "event_type": x.event_type, "kind": x.kind, "reason": x.reason,
"effective_date": x.effective_date.isoformat(), "actor": x.actor,
"recorded_at": S.iso(x.recorded_at), "run_id": x.run_id,
"bp_synced": x.bp_synced_at is not None} for x in p.events]}
if S.has(caller, S.PAY_READ):
out["pay"] = {str(n): _slices(t) for n, t in p.pay.items()}
out["pay_components"] = [{"n": n, "component": c, "slices": _slices(t)} for (n, c), t in p.components.items()]
docs = []
for doc in sorted(p.documents.values(), key=lambda x: x.doc_id):
if doc.access_class == "pay" and not S.has(caller, S.PAY_READ):
continue
if doc.access_class == "personnel" and not (S.has(caller, S.DOC_READ) or S.has(caller, S.PAY_READ)):
continue
if doc.access_class == "sensitive":
continue
docs.append({"doc_id": doc.doc_id, "document_type": doc.document_type, "access_class": doc.access_class,
"filename": doc.filename, "sha256": doc.sha256, "size_bytes": doc.size_bytes,
"uploaded_by": doc.uploaded_by, "uploaded_at": S.iso(doc.uploaded_at),
"expires_on": S.due_label(doc.expires_on)})
out["documents"] = docs
out["compliance"] = [{"n": n, "kind": k, "records": _slices(t)} for (n, k), t in sorted(p.compliance.items())]
if e is not None:
out["intake"] = intake_summary(view, p, e.n)
out["runs"] = [{"run_id": r} for r in view.uow.runs.ids(p.person_id)]
return out
def intake_summary(view: S.View, p: Person, n: int) -> Optional[dict]:
""""details: 5 of 6 done" — field NAMES and counts, never a value (the Sensitive aggregate is read for presence
only, here, by the system; nothing of it leaves)."""
iid = intake_id(p.person_id, n)
i = view.uow.intakes.get(iid)
if i is None:
return None
fields = view.config.intake_fields(i.pack, i.worker_type)
s = view.uow.sensitive.get(p.person_id)
held = s.current(view.today) if s else {}
st = intake_status(fields, held)
return {"intake_id": iid, "issued": i.link_ref is not None, "submitted": i.submitted_at is not None,
"done": st["done"], "total": st["total"], "label": st["label"], "complete": st["complete"],
"missing": st["missing"]}
@HR.use_case(kind="query", route="GET /people/{person_id}", permission=S.PERSON_READ, etag="version")
def show_person(caller: dict, person_id: BigId) -> dict:
"""One person as of today: their names, engagements and status (derived), assignment history with the
derived manager, events, documents of the classes you may read, compliance records, their intake progress
(counts and field names, never values), pay if you hold hr:pay:read — never sensitive data. version is what
you pass back as expected_version. A person outside your view is not found."""
with S.rules():
with uc.repository() as uow:
# everything person_view asks for, in one round trip (the same reads it would make one by one)
uow.prefetch(person=person_id, sensitive=person_id, intakes_of=person_id, run_ids_of=person_id,
seats=True)
view = S.View(uow, caller)
return person_view(view, view.person(person_id))
@HR.use_case(kind="query", route="GET /people", permission=S.PERSON_READ)
def list_people(caller: dict, name: Optional[Name] = None, status: Optional[Name] = None, limit: Limit = 100) -> dict:
"""The people you may see: id, name, current unit and job, worker type, engagement status (future, active,
ending, ended). Filter by part of a name or by status."""
with S.rules():
with uc.repository() as uow:
view = S.View(uow, caller)
people = uow.persons.load_many()
out = []
for p in people:
if not view.sees_person(p.person_id):
continue
e = p.relevant_engagement(view.today)
st = e.status(view.today) if e else None
nm = p.name(view.today).strip()
if name and name.strip().lower() not in nm.lower():
continue
if status and st != status:
continue
a = p.assignments[e.n].at(min(max(view.today, e.start_date), e.end)) if e else None
out.append({"person_id": p.person_id, "name": nm, "worker_type": e.worker_type if e else None,
"status": st, "org_unit_id": a.data.get("org_unit_id") if a else None,
"job_code": a.data.get("job_code") if a else None,
"start_date": e.start_date.isoformat() if e else None})
out.sort(key=lambda r: (r["name"].lower(), r["person_id"]))
return {"people": out[:max(1, min(limit, 1000))]}
@HR.use_case(kind="query", route="GET /me", permission=S.SELF_READ)
def show_my_record(caller: dict) -> dict:
"""My record (HR-24): my details, engagement, assignment, my documents of the personnel class, and my first-week
plan. Only my own record; anybody else's is not found. Not my sensitive data (that goes through HR)."""
with S.rules():
with uc.repository() as uow:
view = S.View(uow, caller)
pid = uow.persons.find_by_sub(caller.get("sub") or "")
if pid is None:
raise Refused(404, "no record for you yet — HR invites you once your record exists")
p = uow.persons.get(pid)
e = p.relevant_engagement(view.today)
seat = p.assignments[e.n].at(min(max(view.today, e.start_date), e.end)) if e else None
first_week = []
for rid in uow.runs.ids(pid):
if not rid.startswith("onboarding:"):
continue
run = uow.runs.get(rid)
for s in run.ordered():
if s.state == "skipped" or s.due_at is None:
continue
if s.responsible_group == "starter" or s.capability == "calendar.create_event" or \
s.responsible_group == "buddy" or (s.timing or {}).get("anchor") == "start_date" and \
int((s.timing or {}).get("offset") or 0) >= 0:
first_week.append({"title": s.title, "due_at": S.iso(s.due_at), "state": s.state})
docs = [{"doc_id": d.doc_id, "document_type": d.document_type, "filename": d.filename}
for d in p.documents.values() if d.access_class == "personnel"]
return {"person_id": pid, "name": p.name(view.today).strip(),
"detail": dict(p.detail_at(view.today).data) if p.detail_at(view.today) else {},
"engagement": ({"n": e.n, "worker_type": e.worker_type, "start_date": e.start_date.isoformat(),
"status": e.status(view.today)} if e else None),
"assignment": ({k: (str(v) if isinstance(v, Decimal) else v) for k, v in seat.data.items()}
if seat else None),
"documents": docs, "first_week": first_week}
def _set_party(run_id: str, group: str, person_id: int, by: str) -> None:
def go():
with uc.repository() as uow:
run = uow.runs.get(run_id)
if run is None:
return
if run.parties.get(group, (None,))[0] != person_id:
run.parties[group] = (person_id, by)
run.add_log("party", by, S.now(), group, person_id)
uow.runs.save(run)
engine._committed(go)
__all__ = ["record_hire", "record_leave", "record_transfer", "record_pay_change", "record_contract_extension",
"record_rehire", "record_conversion", "correct_slice", "move_event_date", "invite_to_sign_in",
"show_person", "list_people", "show_my_record", "person_view", "Company", "Incomplete", "RuleBroken",
"WAITING", "OPEN", "rate_of"]