"""Compliance, payroll and privacy: the UK pack's records (right to work HR-26, IR35 status determination HR-27,
the written statement), the payroll extract (HR-19), subject access (HR-21), erasure with legal holds (HR-22).
The retention sweep (HR-40) runs inside the tick (engine.retention_sweep). The module records decisions and
evidence; it never makes a legal determination.
"""
import json
from datetime import date
from typing import Literal, Optional
from erp_platform.bricks import Refused
from erp_platform.types import BigId, Name, Note, text
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 payroll as PR
from brick_hr.domain import privacy as PV
from brick_hr.domain.packs import gb
from brick_hr.infrastructure import business_partner as bp
HR = uc.HR
@HR.use_case(kind="command", route="POST /people/{person_id}/compliance", permission=S.PERSON_WRITE,
idempotent=True)
def record_compliance(caller: dict, person_id: BigId, kind: Literal["right_to_work_check", "status_determination",
"written_statement_given"],
checked_on: S.Day, checked_by: Name, n: Optional[int] = None,
check_type: Optional[Literal["manual_document_check", "identity_service_provider",
"home_office_online_check"]] = None,
outcome: Optional[Literal["inside", "outside", "checked", "given"]] = None,
follow_up_on: Optional[S.Day] = None, evidence_ref: Optional[Name] = None,
reasons: Optional[Note] = None, given_to_worker_on: Optional[S.Day] = None,
given_to_client_on: Optional[S.Day] = None) -> dict:
"""Record a compliance record the UK pack asks for, on the engagement: a right-to-work check (its type —
manual document check, identity service provider, or Home Office online check — the date, who did it, any
follow-up date; a start before the check is flagged), an IR35 status determination (inside or outside, the
reasons, the date, when given to the worker and to whoever we contract with), or the written statement given.
The playbook's gates on these records are then met. It records what was done; it makes no legal judgement."""
at = S.now()
with S.rules():
with uc.repository() as uow:
view = S.View(uow, caller, at)
p = view.person(person_id)
e = p.engagement(n) if n is not None else p.relevant_engagement(view.today)
if e is None:
raise Refused(409, "no engagement to record it on")
flags = []
if kind == "right_to_work_check":
if check_type is None:
raise Refused(422, "a right-to-work check says how it was done (check_type)")
if e.start_date < checked_on:
flags.append(f"the start date ({e.start_date}) is before the check ({checked_on})")
if kind == "status_determination" and outcome not in ("inside", "outside"):
raise Refused(422, "a status determination is inside or outside IR35 (outcome)")
detail = {"check_type": check_type, "given_to_worker_on": str(given_to_worker_on or "") or None,
"given_to_client_on": str(given_to_client_on or "") or None, "flags": flags}
data = {"pack": gb.PACK, "outcome": outcome or "checked", "checked_on": checked_on,
"checked_by": checked_by, "follow_up_on": follow_up_on, "evidence_ref": evidence_ref,
"reasons": reasons, "detail_json": json.dumps(detail, sort_keys=True)}
cur = p.compliance_on(e.n, kind)
if cur is None or cur.valid_from != checked_on or any(
(cur.data.get(k) or None) != (v or None) for k, v in data.items()):
p.record_compliance(e.n, kind, checked_on, data, now=at, by=S.actor(caller))
uow.persons.save(p)
for rid in _open_runs(person_id, e.n):
engine.advance(rid, at)
return {"person_id": person_id, "n": e.n, "kind": kind, "checked_on": checked_on.isoformat(), "flags": flags}
def _open_runs(person_id: int, n: int) -> list:
with uc.repository() as uow:
return [r.run_id for r in uow.runs.load_many(uow.runs.ids(person_id)) if r.n == n and r.state == "open"]
# ── payroll extract (HR-19) ────────────────────────────────────────────────────────────────
def _line_view(ln: PR.Line) -> dict:
return {"person_id": ln.person_id, "n": ln.n, "component": ln.component, "for_period": ln.for_period,
"slice_from": ln.slice_from.isoformat(), "rate_basis": ln.rate_basis,
"rate_amount": str(ln.rate_amount) if ln.rate_amount is not None else None, "currency": ln.currency,
"prorate_from": S.due_label(ln.prorate_from), "prorate_to": S.due_label(ln.prorate_to),
"payable_amount": str(ln.payable_amount) if ln.payable_amount is not None else None,
"change_kind": ln.change_kind, "org_unit_id": ln.org_unit_id, "cost_centre": ln.cost_centre,
"detail": ln.detail}
def _batch_view(b: PR.Batch) -> dict:
return {"batch_id": b.batch_id, "legal_entity_ref": b.legal_entity_ref, "period": b.period, "state": b.state,
"sent_at": S.iso(b.sent_at), "sent_by": b.sent_by, "version": b.version,
"lines": [_line_view(ln) for ln in sorted(b.lines, key=lambda l: (l.person_id, l.n, l.for_period,
l.component, l.slice_from))]}
@HR.use_case(kind="command", route="POST /payroll/extracts", permission=S.PAY_WRITE, idempotent=True)
def produce_payroll_extract(caller: dict, legal_entity_ref: Name, period: text(7)) -> dict:
"""Produce the payroll extract for one legal entity and period (HR-19): every pay and component slice touching
the period with its proration dates, starters and leavers, bank changes (to those who may read them);
freelancer engagements (paid as suppliers) are never in it. A SENT batch is frozen: a later change to a sent
period lands in the next open batch as a retro line, so no row goes out twice. Producing again rebuilds an
unsent batch from the slices."""
at = S.now()
with S.rules():
with uc.repository() as uow:
view = S.View(uow, caller, at)
view.require_company(legal_entity_ref, f"legal entity {legal_entity_ref}")
bid = PR.batch_id(legal_entity_ref, period)
existing = uow.payroll.get(bid)
batches = uow.payroll.batches(legal_entity_ref)
sent = [b for b in batches if b.state == PR.SENT and b.period != period]
payroll_types = {r["worker_type"] for r in view.config.all("worker_type") if r.get("payroll")}
banks = uow.sensitive.bank_slices() if S.has(caller, S.SENSITIVE_READ) else {}
people = uow.persons.load_many()
periods = {period} | {b.period for b in sent if b.period < period}
current = {q: [ln for p in people for ln in PR.lines_for(p, q, legal_entity_ref, payroll_types,
banks.get(p.person_id))] for q in periods}
sent_lines = [ln for b in sorted(sent, key=lambda b: b.period) for ln in b.lines]
b = PR.build(legal_entity_ref, period, current, sent_lines, now=at, existing=existing)
if existing is not None and _same_lines(existing, b):
return _batch_view(existing)
uow.payroll.save(b)
return _batch_view(b)
def _same_lines(a: PR.Batch, b: PR.Batch) -> bool:
key = lambda ln: (ln.key, ln.fingerprint(), ln.change_kind) # noqa: E731
return sorted(map(key, a.lines)) == sorted(map(key, b.lines))
@HR.use_case(kind="command", route="POST /payroll/extracts/{batch_id}/sent", permission=S.PAY_WRITE, idempotent=True)
def mark_extract_sent(caller: dict, batch_id: Name) -> dict:
"""Mark a payroll extract as sent to the payroll provider: it is frozen from now on."""
at = S.now()
with S.rules():
with uc.repository() as uow:
b = uow.payroll.get(batch_id)
if b is None or not S.View(uow, caller, at).in_companies(b.legal_entity_ref):
raise Refused(404, f"no payroll extract {batch_id}")
if b.mark_sent(S.actor(caller), at):
uow.payroll.save(b)
return _batch_view(b)
@HR.use_case(kind="query", route="GET /payroll/extracts/{batch_id}", permission=S.PAY_READ)
def show_payroll_extract(caller: dict, batch_id: Name) -> dict:
"""One payroll extract with its lines."""
with S.rules():
with uc.repository() as uow:
b = uow.payroll.get(batch_id)
if b is None or not S.View(uow, caller).in_companies(b.legal_entity_ref):
raise Refused(404, f"no payroll extract {batch_id}")
v = _batch_view(b)
if not S.has(caller, S.SENSITIVE_READ):
for ln in v["lines"]:
ln["detail"] = {k: x for k, x in (ln["detail"] or {}).items() if k != "bank"}
return v
# ── privacy (HR-21, HR-22) ─────────────────────────────────────────────────────────────────
def _request_view(q: PV.PrivacyRequest) -> dict:
return {"request_id": q.request_id, "person_id": q.person_id, "kind": q.kind,
"received_on": q.received_on.isoformat(), "due_on": q.due_on.isoformat(), "done_at": S.iso(q.done_at),
"done_by": q.done_by, "export_object_key": q.export_object_key, "outcome": q.outcome}
@HR.use_case(kind="command", route="POST /privacy/requests", permission=S.PRIVACY, idempotent=True)
def record_privacy_request(caller: dict, person_id: BigId, kind: Literal["access", "erasure"],
received_on: S.Day) -> dict:
"""Record a subject-access or erasure request with its received date. It is due one month from receipt
(UK GDPR Art. 12(3)). The same request twice is one."""
with S.rules():
with uc.repository() as uow:
if uow.persons.get(person_id) is None or not S.View(uow, caller).person_in_companies(person_id):
raise Refused(404, f"no person {person_id}")
rid = PV.request_id(kind, person_id, received_on)
q = uow.privacy.get(rid)
if q is None:
q = PV.PrivacyRequest(rid, person_id, kind, received_on, PV.due_on(received_on))
uow.privacy.save(q)
return _request_view(q)
def _archive(uow, view, person_id: int) -> dict:
"""Everything we hold on a person: every slice INCLUDING superseded ones, events, documents, sensitive data,
run steps naming them, and their business-partner fields."""
from brick_hr.application.people import _slices
p = uow.persons.get(person_id)
s = uow.sensitive.get(person_id)
runs = uow.runs.load_many(uow.runs.ids(person_id))
named = [r for r in uow.runs.load_many() if person_id in {pid for pid, _ in r.parties.values()}
and r.person_id != person_id]
partner = bp.read(p.bp_id or bp.person_bp_id(person_id), engine.ACTOR) if bp.installed() else None
return {"person": {"person_id": str(p.person_id), "person_key": p.person_key, "staff_no": p.staff_no,
"bp_id": str(p.bp_id) if p.bp_id else None, "pool_user_sub": p.pool_user_sub},
"detail": _slices(p.detail),
"engagements": [{"n": e.n, "worker_type": e.worker_type, "start_date": e.start_date.isoformat(),
"planned_end": S.due_label(e.planned_end), "actual_end": S.due_label(e.actual_end),
"leave_reason": e.leave_reason, "assignments": _slices(p.assignments[e.n]),
"pay": _slices(p.pay[e.n])} for e in sorted(p.engagements.values(), key=lambda e: e.n)],
"pay_components": [{"n": n, "component": c, "slices": _slices(t)} for (n, c), t in p.components.items()],
"events": [{"n": e.n, "seq": e.seq, "kind": e.kind, "event_type": e.event_type, "reason": e.reason,
"effective_date": e.effective_date.isoformat(), "actor": e.actor,
"recorded_at": S.iso(e.recorded_at)} for e in p.events],
"documents": [{"doc_id": d.doc_id, "document_type": d.document_type, "filename": d.filename,
"sha256": d.sha256, "object_key": d.object_key} for d in p.documents.values()],
"compliance": [{"n": n, "kind": k, "records": _slices(t)} for (n, k), t in p.compliance.items()],
"sensitive": {k: [{"valid_from": sl.valid_from.isoformat(), "rev": sl.rev, "active": sl.active,
"value": sl.data.get("value"), "object_key": sl.data.get("object_key")}
for sl in t.slices] for k, t in (s.kinds.items() if s else [])},
"runs": [{"run_id": r.run_id, "direction": r.direction,
"steps": [{"step_no": x.step_no, "title": x.title, "state": x.state, "input": x.input,
"done_by": x.done_by, "done_at": S.iso(x.done_at)} for x in r.ordered()]}
for r in runs],
"named_in_runs": [{"run_id": r.run_id, "as": [g for g, (pid, _b) in r.parties.items() if pid == person_id]}
for r in named],
"business_partner": partner}
@HR.use_case(kind="command", route="POST /privacy/requests/{request_id}/export", permission=S.PRIVACY,
idempotent=True)
def export_subject_access(caller: dict, request_id: Name) -> dict:
"""Answer a subject-access request (HR-21): ONE archive (stored in the bucket) with the person, every slice
including superseded ones, events, documents, sensitive data, run steps naming them, and their
business-partner fields. The request shows its due date; the export is logged with who and when. The archive
is downloaded with download_subject_access. Exporting again keeps the first export."""
at = S.now()
with S.rules():
with uc.repository() as uow:
view = S.View(uow, caller, at)
q = uow.privacy.get(request_id)
if q is None or q.kind != "access" or not view.person_in_companies(q.person_id):
raise Refused(404, f"no subject-access request {request_id}")
if q.export_object_key:
return _request_view(q)
archive = _archive(uow, view, q.person_id)
key = f"hr/{q.person_id}/privacy/{request_id.replace(':', '-')}.json"
uow.objects.put(key, json.dumps(archive, sort_keys=True, default=str).encode(), "application/json")
q.export_object_key, q.done_at, q.done_by = key, at, S.actor(caller)
q.outcome = {"slices": len(archive["detail"]) + sum(len(e["assignments"]) + len(e["pay"])
for e in archive["engagements"]),
"events": len(archive["events"]), "documents": len(archive["documents"]),
"sensitive_kinds": len(archive["sensitive"]), "runs": len(archive["runs"])}
uow.privacy.save(q)
uow.retention.add(view.today, q.person_id, f"export:{request_id}", "exported", "subject access",
S.actor(caller))
return _request_view(q)
@HR.use_case(kind="query", route="GET /privacy/requests/{request_id}/archive", permission=S.PRIVACY)
def download_subject_access(caller: dict, request_id: Name) -> dict:
"""The subject-access archive, streamed (base64 JSON) — never a link."""
import base64
with S.rules():
with uc.repository() as uow:
q = uow.privacy.get(request_id)
if q is not None and not S.View(uow, caller).person_in_companies(q.person_id):
q = None
got = uow.objects.get(q.export_object_key) if q and q.export_object_key else None
if got is None:
raise Refused(404, f"no export for {request_id}")
return {"request_id": request_id, "content_type": got[1], "sha256": got[2],
"content_base64": base64.b64encode(got[0]).decode()}
@HR.use_case(kind="command", route="POST /privacy/requests/{request_id}/erase", permission=S.PRIVACY,
destructive=True, idempotent=True)
def apply_erasure(caller: dict, request_id: Name) -> dict:
"""Apply an erasure request (HR-22): each data category is erased or anonymised UNLESS a retention rule with a
legal-obligation basis still holds it — then it is kept and the basis recorded. The business partner's person
fields are anonymised through the business-partner seam. Everything done is in the retention log. Applying it
twice changes nothing."""
at = S.now()
with S.rules():
with uc.repository() as uow:
view = S.View(uow, caller, at)
q = uow.privacy.get(request_id)
if q is None or q.kind != "erasure" or not view.person_in_companies(q.person_id):
raise Refused(404, f"no erasure request {request_id}")
if q.done_at is not None:
return _request_view(q)
p = uow.persons.get(q.person_id)
if p is not None and any(e.actual_end is None or e.actual_end >= view.today for e in p.engagements.values()):
raise Refused(409, "they still work here: an erasure applies after their last engagement has ended")
ended = max((e.actual_end for e in p.engagements.values()), default=None) if p else None
plan = PV.erasure_plan(view.config, ended, view.today)
done = engine.apply_categories(q.person_id, plan, at, f"erasure {request_id} by {S.actor(caller)}")
with S.rules():
with uc.repository() as uow:
q = uow.privacy.get(request_id)
q.done_at, q.done_by = at, S.actor(caller)
q.outcome = {c: {"action": a, "basis": b} for c, (a, b) in sorted(plan.items())}
q.outcome["applied"] = len(done)
uow.privacy.save(q)
return _request_view(q)
@HR.use_case(kind="query", route="GET /people/{person_id}/retention-log", permission=S.PRIVACY)
def read_retention_log(caller: dict, person_id: BigId) -> dict:
"""What the retention sweep and erasures did to a person's data: date, category, action (purged,
anonymised, kept, exported) and its basis."""
with S.rules():
with uc.repository() as uow:
S.View(uow, caller).require_person_in_companies(person_id, f"person {person_id}")
rows = sorted(uow.retention.rows(person_id), key=lambda r: (str(r["run_date"]), r["category"]))
return {"person_id": person_id, "entries": [
{"run_date": str(r["run_date"]), "category": r["category"], "action": r["action"],
"basis": r["basis"], "actor": r["actor"]} for r in rows]}
__all__ = ["record_compliance", "read_retention_log", "produce_payroll_extract", "mark_extract_sent", "show_payroll_extract",
"record_privacy_request", "export_subject_access", "download_subject_access", "apply_erasure", "date"]