"""What every use case shares: the permission names, the refusal mapping, the business clock, who the caller is
(their person), and ROW VISIBILITY — decided here, from `caller`, for both doors alike.
Visibility (modules/hr permissions.row_visibility):
* hr_admin and finance see every person (they hold hr:person:write or hr:sensitive:read);
* a manager sees the people whose DERIVED manager they are (ManagerOf today or on a future slice of the
person's own assignment), and their runs, pay and personnel/pay documents — never sensitive data;
* IT (anyone) sees people with an open step of a group they belong to;
* an assignee sees only steps of the groups they belong to (a pool role, the derived manager, a run party);
* an employee sees only themselves.
A row outside visibility is 404, never 403.
"""
import os
import re
import unicodedata
from contextlib import contextmanager
from datetime import date, datetime, timedelta
from functools import cached_property
from typing import Annotated, Optional
from pydantic import Field
from erp_platform import auth, work
from erp_platform.types import text
from erp_platform.bricks import Refused
from erp_platform.money import InvalidMoney
from brick_hr.domain import services
from brick_hr.domain.errors import Incomplete, NotFound, RuleBroken
from brick_hr.domain.person import Person
from brick_hr.domain.values import OPEN
# ── bounded inputs (erp_platform.types, plus the two it does not have) ─────────────────────
# A date a person types: 1900 … 2199 (the platform's IsoDate is a string; hr's domain works in dates).
Day = Annotated[date, Field(ge=date(1900, 1, 1), le=date(2199, 12, 31))]
Code = text(64) # a catalogue code: a worker type, a job, an org unit, an event type
# A RATE: up to 4 places (NUMERIC(19,4)), which the platform's Amount (≤ 3 places) refuses; money.Rate checks it.
RateText = text(32)
# ── permissions (namespace hr) ─────────────────────────────────────────────────────────────
PERSON_READ, PERSON_WRITE = "hr:person:read", "hr:person:write"
HIRE, LEAVE, EVENT = "hr:hire:record", "hr:leave:record", "hr:event:record"
PAY_READ, PAY_WRITE = "hr:pay:read", "hr:pay:write"
SENSITIVE_READ, SENSITIVE_WRITE = "hr:sensitive:read", "hr:sensitive:write"
DOC_READ, DOC_WRITE = "hr:document:read", "hr:document:write"
ORG_READ, ORG_WRITE = "hr:org:read", "hr:org:write"
RUN_READ, RUN_PREVIEW, RUN_RESCHEDULE, RUN_CANCEL, RUN_TICK = (
"hr:run:read", "hr:run:preview", "hr:run:reschedule", "hr:run:cancel", "hr:run:tick")
TASK_READ, TASK_COMPLETE = "hr:task:read", "hr:task:complete"
STEP_APPROVE, DRAFT_APPROVE, ACTION_RETRY, ACTION_LOG = (
"hr:step:approve", "hr:draft:approve", "hr:action:retry", "hr:action_log:read")
RUN_MONITOR = "hr:run:monitor" # HR-29: the onboarding programme's numbers, over the runs you may see
CONFIG_WRITE, REPORT_READ, PRIVACY, POOL_INVITE, SELF_READ, CLOCK_SET = (
"hr:config:write", "hr:report:read", "hr:privacy:manage", "hr:pool:invite", "hr:self:read", "hr:clock:set")
OFFBOARDING_POLICY = "hr:offboarding:policy" # HR-55: the owner's — when a leaver's ERP sign-in goes
# v4 (HR-63, HR-64): machine permissions — only a service client's exact grant reaches them (Brick machine_permissions)
ASSIGNMENT_READ, ROSTER_READ = "hr:assignment:read", "hr:roster:read"
def has(caller: dict, perm: str) -> bool:
return auth.has_permission(caller, perm)
def actor(caller: dict) -> str:
return caller.get("email") or caller.get("sub") or "unknown"
@contextmanager
def rules():
"""A domain refusal in its own words: RuleBroken 409, Incomplete 422, NotFound 404."""
try:
yield
except Incomplete as e:
raise Refused(422, str(e)) from None
except RuleBroken as e:
raise Refused(409, str(e)) from None
except NotFound as e:
raise Refused(404, str(e)) from None
except InvalidMoney as e:
raise Refused(422, str(e)) from None
except ValueError as e:
if type(e) is ValueError and ("time zone" in str(e) or "time_of_day" in str(e) or "anchor" in str(e)):
raise Refused(422, str(e)) from None
raise
def not_found(what: str) -> Refused:
return Refused(404, f"no {what}")
# ── the business clock ─────────────────────────────────────────────────────────────────────
def now() -> datetime:
"""Business time: work.now() (a demo may move it; never in prod)."""
return work.now()
def local_date(at: datetime, tz: str) -> date:
return work.aware(at).astimezone(work.zone(tz)).date()
def iso(t) -> Optional[str]:
if t is None:
return None
if isinstance(t, datetime):
return work.aware(t).astimezone(work.UTC).isoformat().replace("+00:00", "Z")
return t.isoformat()
def due_label(d: Optional[date]) -> Optional[str]:
return None if d is None or d == OPEN else d.isoformat()
# ── the caller's person, and what they may see ─────────────────────────────────────────────
class View:
"""Who the caller is in hr, and which rows they may see — computed once per unit of work."""
def __init__(self, uow, caller: dict, at: Optional[datetime] = None):
self.uow, self.caller = uow, caller
self.now = at or now()
self.config = uow.config.load()
self.tz = self.config.timezone
self.today = local_date(self.now, self.tz)
@cached_property
def sees_all(self) -> bool:
return has(self.caller, PERSON_WRITE) or has(self.caller, SENSITIVE_READ)
@cached_property
def me(self) -> Optional[int]:
sub = self.caller.get("sub")
email = self.caller.get("email") if self.caller.get("auth_source") != "public" else None
return self.uow.persons.find_me(sub, email) # by sub, else by work email: asked at once
@cached_property
def roles(self) -> set:
# role-check: a responsible group's resolution row (customer configuration) names a pool role ("pool_role:it"); every use case is still gated by its permission
return {str(r).strip().lower() for r in self.caller.get("roles") or []}
@cached_property
def directory(self) -> services.Directory:
return services.Directory(self.uow.persons.seats())
@cached_property
def orgs(self) -> dict:
return self.uow.orgs.all()
@cached_property
def line(self) -> set:
if self.me is None:
return set()
people = {s.person_id for s in self.directory.seats}
return services.line_of(self.me, self.today, self.directory, self.orgs, people)
def manager_of(self, person_id: int, d: Optional[date] = None) -> Optional[int]:
return services.manager_of(person_id, d or self.today, self.directory, self.orgs)
@cached_property
def open_steps(self) -> list:
"""(run_id, person_id, step) for every unfinished step — for 'people with an open step of my group'."""
return self.uow.runs.open_steps()
@cached_property
def step_people(self) -> set:
out = set()
for run_id, person_id, group, parties in self.open_steps:
if self.in_group(group, person_id, parties, run_id=run_id):
out.add(person_id)
return out
def in_group(self, group: str, person_id: int, parties: dict, *, run_id: str = "", on: Optional[date] = None) -> bool:
"""Does the caller belong to this responsible group for this person's run? Resolved now, from the row."""
row = self.config.group(group)
res = (row or {}).get("resolution") or ""
if res.startswith("pool_role:"):
return res.split(":", 1)[1].strip().lower() in self.roles
if self.me is None:
return False
if res == "manager":
return self.me == self.relevant_manager(person_id, on)
if res == "run_party":
return (parties or {}).get(group) == self.me
if res == "person":
return person_id == self.me
return False
def relevant_manager(self, person_id: int, on: Optional[date] = None) -> Optional[int]:
"""The derived manager for a person's run: on the day asked, else on the day their assignment is live
(a future starter's manager is their manager before the start; a leaver's, after the last day)."""
d = on or self.today
m = self.manager_of(person_id, d)
if m is not None:
return m
seats = sorted((s for s in self.directory.seats if s.person_id == person_id), key=lambda s: s.valid_from)
if not seats:
return None
future = [s for s in seats if s.valid_from > d]
probe = future[0].valid_from if future else seats[-1].valid_to
return self.manager_of(person_id, probe)
# ── the companies the caller's grant covers (hr v4: every hr grant is scoped per company) ────────────
@cached_property
def companies(self) -> Optional[frozenset]:
"""The company codes the caller's grant covers (hr_company_grant rows naming their sub or email), or None
for every company — a principal no row names, as the spec's "none named = every company"."""
email = self.caller.get("email") if self.caller.get("auth_source") != "public" else None
return self.config.companies_for(self.caller.get("sub"), email)
def in_companies(self, company_code: Optional[str]) -> bool:
return self.companies is None or (company_code or "") in self.companies
def company_of(self, person_id: int, d: Optional[date] = None) -> Optional[str]:
"""The person's company on `d` (today): the engagement of the seat on that day, else their next one, else
their last — as relevant_manager reads the seat (a future starter belongs to their company before day 1)."""
d = d or self.today
seats = sorted((s for s in self.directory.seats if s.person_id == int(person_id)), key=lambda s: s.valid_from)
if not seats:
return None
on = next((s for s in seats if s.covers(d)), None)
future = [s for s in seats if s.valid_from > d]
return (on or (future[0] if future else seats[-1])).company_code
def person_in_companies(self, person_id: int, d: Optional[date] = None) -> bool:
return self.companies is None or self.in_companies(self.company_of(person_id, d))
def people_in_companies(self) -> set:
return {s.person_id for s in self.directory.seats if self.person_in_companies(s.person_id)}
def sees_person(self, person_id: int) -> bool:
# hr v4: every grant stops at its companies (you always see yourself). Unscoped, nothing more is read.
if self.companies is not None and not (self.me is not None and person_id == self.me) and \
not self.person_in_companies(person_id):
return False
if self.sees_all:
return True
if self.me is not None and person_id == self.me:
return True
return person_id in self.line or person_id in self.step_people
def require_company(self, company_code: Optional[str], what: str) -> None:
"""404 for a company outside the caller's grant — nothing confirms it exists."""
if not self.in_companies(company_code):
raise not_found(what)
def require_person_in_companies(self, person_id: int, what: str) -> None:
if not self.person_in_companies(int(person_id)):
raise not_found(what)
def person(self, person_id: int) -> Person:
"""The person, or 404 — also when it exists and the caller may not see it."""
if not self.sees_person(int(person_id)):
raise not_found(f"person {person_id}")
p = self.uow.persons.get(int(person_id))
if p is None:
raise not_found(f"person {person_id}")
return p
def check_company_grant_change(view: View, rows: list) -> None:
"""hr v4: who may keep hr_company_grant. Only someone whose own grant covers every company (no row names
them) — a grant scoped to UK01 cannot widen itself or anyone else to UK02 — and never a row naming the caller
(no self-approval: by email AND sub), since removing your own last row would widen you to every company."""
if view.companies is not None:
raise Refused(409, "company grants are kept by someone whose grant covers every company")
me = {str(x).strip().lower() for x in (view.caller.get("sub"), view.caller.get("email")) if x}
for r in rows:
if isinstance(r, dict) and str(r.get("principal") or "").strip().lower() in me:
raise Refused(409, "nobody changes their own company grant: ask another administrator")
def manager_line_check(view: View, org_unit_id: str, org_id: str, d: date) -> None:
"""A manager may hire or transfer only into a unit in their line: one led (as chief) by a position they hold
on that date, or below one. Operations (hr:person:write) may anywhere."""
if view.sees_all and has(view.caller, PERSON_WRITE):
return
me = view.me
org = view.orgs.get(org_id)
if me is None or org is None:
raise Refused(409, f"{org_unit_id} is outside your line — ask Operations, or hold the unit's chief position")
held = view.directory.positions_held(me, d) | view.directory.positions_held(me, view.today)
chain = org.ancestors(org_unit_id, d)
if not any(u.chief_position_id in held for u in chain if u.chief_position_id):
raise Refused(409, f"{org_unit_id} is outside your line on {d} — a transfer or hire there needs its chief "
"position (or Operations)")
# ── template values for a run's steps ──────────────────────────────────────────────────────
def _ascii(s: str) -> str:
return re.sub(r"[^a-z0-9]+", "", unicodedata.normalize("NFKD", s or "").encode("ascii", "ignore").decode().lower())
def values_for(view: View, p: Person, n: int, *, anchors: Optional[dict] = None, parties: Optional[dict] = None,
drafts: Optional[dict] = None) -> dict:
"""What a step's input template may say: the person's names and contacts (NOT their sensitive data, which is
not even loaded here), the assignment, dates, the derived manager and the run parties."""
config, today = view.config, view.today
e = p.engagements[n]
rel = min(max(today, e.start_date), e.end)
a = p.assignments[n].at(rel) or p.assignments[n].last()
ad = a.data if a else {}
d = p.detail_at(today)
dd = d.data if d else {}
org = view.orgs.get(ad.get("org_id")) if ad.get("org_id") else None
unit = org.unit_at(ad.get("org_unit_id"), rel) if org else None
path = "/" + "/".join(u.name for u in reversed(org.ancestors(ad.get("org_unit_id"), rel))) if org else ""
job = config.job(ad.get("job_code")) or {}
loc = config.location(ad.get("location_code")) or {}
wt = config.worker_type(e.worker_type) or {}
manager_id = view.relevant_manager(p.person_id, rel)
party_ids = {g: pid for g, (pid, _by) in (parties or {}).items() if pid}
names = view.uow.persons.names([manager_id] + list(party_ids.values()), today)
domain = config.setting("email_domain")
given, family = dd.get("legal_given_name") or "", dd.get("legal_family_name") or ""
proposed = f"{_ascii(dd.get('preferred_name') or given)}.{_ascii(family)}@{domain}" if domain else ""
account = dd.get("work_email") or proposed
manager = names.get(manager_id, {}) if manager_id else {}
buddy = names.get(party_ids.get("buddy"), {}) if party_ids.get("buddy") else {}
start_plus_1 = config.calendar(ad.get("location_code"), ad.get("schedule_code")).add_working_days(e.start_date, 1)
attendees = [x for x in (account, manager.get("work_email"), buddy.get("work_email")) if x]
out = {"person_id": str(p.person_id), "given_name": given, "family_name": family,
"preferred_name": dd.get("preferred_name") or given, "legal_name": f"{given} {family}".strip(),
"personal_email": dd.get("personal_email") or "", "personal_email_list": [x for x in [dd.get("personal_email")] if x],
"work_email": dd.get("work_email") or "", "proposed_work_email": proposed, "account_email": account,
"org_unit_id": ad.get("org_unit_id") or "", "org_unit_name": unit.name if unit else "", "ou_path": path,
"job_code": ad.get("job_code") or "", "job_title": job.get("title") or "", "location_code": ad.get("location_code") or "",
"location_name": loc.get("name") or "", "timezone": config.timezone,
"start_date": e.start_date.isoformat(), "start_plus_1": start_plus_1.isoformat(),
"last_day": e.actual_end.isoformat() if e.actual_end else (e.planned_end.isoformat() if e.planned_end else ""),
"worker_type": e.worker_type, "worker_type_label": wt.get("label") or e.worker_type,
"legal_entity_ref": e.legal_entity_ref, "manager_email": manager.get("work_email") or "",
"manager_name": manager.get("name") or "", "buddy_name": buddy.get("name") or "",
"buddy_email": buddy.get("work_email") or "", "first_week_attendees": attendees,
"client_size": config.client_size(e.legal_entity_ref),
"user_pool": config.setting("user_pool") or os.environ.get("DODIL_USER_POOL", ""),
"pool_user_sub": p.pool_user_sub or None}
for code, text in (drafts or {}).items():
out[f"draft.{code}"] = text
for k, v in (anchors or {}).items():
if k == "event_date" and hasattr(v, "isoformat"):
out["event_date"] = v.isoformat()
return out
def facts_for(view: View, p: Person, n: int, rate_outside_band: bool = False) -> dict:
e = p.engagements[n]
wt = view.config.worker_type(e.worker_type) or {}
prev = p.engagements.get(n - 1)
return {"worker_type": e.worker_type, "via_company": bool(wt.get("via_company")),
"rate_outside_band": rate_outside_band, "client_size": view.config.client_size(e.legal_entity_ref),
"compliance": sorted(k for (m, k), t in p.compliance.items() if m == n and t.active()),
"from_worker_type": prev.worker_type if prev else None}
def rate_outside_band(view: View, p: Person, n: int) -> bool:
"""'rate outside band': the engagement's first rate against its pay grade's (or job's) band."""
from brick_hr.domain.person import rate_of
t = p.pay.get(n)
s = t.first() if t else None
r = rate_of(s)
if r is None:
return False
grade = view.config.pay_grade(s.data.get("pay_grade")) if s.data.get("pay_grade") else None
if grade is None:
a = p.assignments[n].first()
job = view.config.job(a.data.get("job_code")) if a else None
grade = view.config.pay_grade(job.get("default_pay_grade")) if job and job.get("default_pay_grade") else None
if not grade or grade.get("rate_basis") != r.basis or (grade.get("currency") or r.currency) != r.currency:
return False
from decimal import Decimal
lo = Decimal(grade["min_amount"]) if grade.get("min_amount") else None
hi = Decimal(grade["max_amount"]) if grade.get("max_amount") else None
return (lo is not None and r.amount < lo) or (hi is not None and r.amount > hi)
def retry_unless_versioned(fn):
"""A caller who sent NO version asked for "this change, on whatever is there": a lost version race is re-run
on the latest (erp_platform.repository.retry_on_conflict) so both changes land. One who sent the version they
loaded gets the 409 "changed since you opened it" — they decided on what they saw.
A versioned call refused for a RULE is re-run once before the refusal stands. The load reads the root first,
so the children can be newer than the version it read: a change committed between those reads shows its rows
under the old version, the version check passes, and a rule then fires on rows the caller never saw (found
2026-10-18, SC-HR-08: two pay slices on one date, the loser got "a slice already starts on 2026-11-01"
instead of the 409). A refused command wrote nothing, so the re-run is safe; it reads after that change
committed, so a moved version now gives the 409, and an unchanged one gives the same refusal again."""
import functools
from erp_platform.repository import CONFLICT_REASON, retry_on_conflict
@functools.wraps(fn)
def run(caller, *a, **kw):
if kw.get("expected_version") is not None:
try:
return fn(caller, *a, **kw)
except Refused as e:
if e.status == 409 and e.reason == CONFLICT_REASON:
raise
return fn(caller, *a, **kw)
return retry_on_conflict(lambda: fn(caller, *a, **kw))
return run
def days(n: int) -> timedelta:
return timedelta(days=n)
__all__ = ["View", "rules", "has", "actor", "now", "local_date", "iso", "values_for", "facts_for",
"manager_line_check", "not_found", "rate_outside_band", "check_company_grant_change"]