"""The use cases — what Operations, managers, IT, finance, assignees, employees, starters and the clock can do,
each declared ONCE with @HR.use_case and served as an API route AND an MCP tool.
`caller` is the first parameter of every one; row visibility is decided inside the use case from it, so an agent
sees exactly what its person sees. Refusals are Refused(status, reason): a broken domain rule is 409, invalid input
422, a row the caller may not see 404 — never 403, which would confirm it exists.
The modules, by area (each imports HR from here and registers its use cases):
people hire, leave, transfer, pay, extension, rehire, conversion, corrections, date moves, invite,
show_person, show_my_record
sensitive_data show_sensitive, record_sensitive (hr:sensitive:* — and 404 for everyone else)
organisation keep_org, keep_configuration, keep_playbook, install_localisation_pack, org_chart, list_playbooks,
show_playbook_bpmn (a playbook version as BPMN 2.0: domain/bpmn.py)
process playbook_deviations (against the pack's reference: variation / warning / foundation),
scenario_paths (the scenarios that walk a playbook, as step keys), change_playbook (one change)
runs my_hr_tasks, run_board, show_run, preview_run, complete_step, approve_step, approve_draft,
assign_run_party, retry_step, cancel_run, read_action_log, run_due_steps (the tick)
intake the starter's form through the public-link door, reissue_intake_link
documents file_document, list_documents, download_document (streamed; never a link)
compliance record_compliance, payroll extract, privacy requests, retention
reports headcount, starters_and_leavers
"""
import os
from contextlib import contextmanager
from erp_platform.bricks import Brick, Refused
from erp_platform.repository import CONFLICT_REASON
from sqlalchemy.exc import IntegrityError
from brick_hr.infrastructure.capability_acl import REQUIRED
VERSION = "0.6.0"
# hr v4: the seams other modules read through (HR-63, HR-64) are machine permissions — an exact grant only
MACHINE = ("hr:assignment:read", "hr:roster:read")
HR = Brick("hr", version=VERSION, requires=REQUIRED, ids_as_strings=True, machine_permissions=MACHINE, instructions=(
"People, starters and leavers. Who works here, on what terms and since when (effective-dated: nothing is "
"overwritten — a change delimits, a correction supersedes), the org tree and the derived manager, and the "
"onboarding / offboarding / change / conversion RUNS that do the admin: steps due relative to a start date or "
"last day in working days, tasks for IT, finance, the buddy and the manager, actions through capabilities "
"(a manual task with the exact input when no provider is wired), approvals before anything destructive. "
"person_id is the key for a person; run ids are <direction>:<person_id>:<n>:<seq>. Sensitive data (bank, NI, "
"address, right to work) is separate and HR/finance only."))
@contextmanager
def _repository_default():
from erp_platform import db
with _on_engine(db.engine) as uow:
yield uow
def _lost_a_unique_race(exc: IntegrityError) -> bool:
text_ = f"{getattr(exc, 'orig', '')} {exc}"
return "23505" in text_ or "duplicate key" in text_ or "UNIQUE constraint" in text_
@contextmanager
def _on_engine(engine):
"""One transaction. On DataK3 a write conflict on a claim row (a version claim, a step claim, an idempotency
key) can surface at COMMIT — the loser's COMMIT raises 23505 after the use case has returned — so it is caught
here, around the commit, and answered as the same retry-safe 409 a conflict seen mid-use-case gets, never a 500."""
from brick_hr.infrastructure.tables import HrRepository
try:
with engine.begin() as conn:
yield HrRepository(conn)
except IntegrityError as exc:
if not _lost_a_unique_race(exc):
raise
raise Refused(409, CONFLICT_REASON) from exc # the cause stays: is_duplicate still reads it
# The unit of work: ONE connection, one transaction. Tests and the scenario harness swap it for SQLite.
repository = _repository_default
# The demo clock: the route exists only when HR_DEMO_CLOCK=enabled, and the app refuses to start with it set on a
# live deployment (DODIL_ENV=live|prod). erp_platform.democlock also answers 404 there, and work.set_clock raises.
class DemoClockOnLive(RuntimeError):
pass
if os.environ.get("HR_DEMO_CLOCK", "").strip().lower() == "enabled":
if os.environ.get("DODIL_ENV", "").strip().lower() in ("live", "prod"):
raise DemoClockOnLive("HR_DEMO_CLOCK=enabled on a live deployment (DODIL_ENV=live|prod): refusing to start")
from erp_platform import democlock
democlock.mount(HR)
from brick_hr.application import (assignment, compliance, documents, intake, organisation, # noqa: E402,F401
people, process, reports, runs, sensitive_data)
def _one_request(fn):
"""A use case run through a door is one request: one configuration-version read for all of its units of
work (infrastructure/repositories.configuration_snapshot). The function's attributes come with it."""
import functools
from brick_hr.infrastructure.repositories import request_scope
@functools.wraps(fn)
def run(*a, **kw):
with request_scope():
return fn(*a, **kw)
return run
def _scope_every_use_case() -> None:
import dataclasses
HR.use_cases[:] = [dataclasses.replace(u, fn=_one_request(u.fn)) for u in HR.use_cases]
_scope_every_use_case()
__all__ = ["HR", "repository", "VERSION"]