"""Runs and their steps: My tasks (HR-03), the run board (HR-04), one run, the preview (HR-10), completing,
approving (HR-07), approving drafts (HR-05), run parties, retries, cancelling (HR-23), the action log (HR-09), and
the tick (HR-30).
An assignee sees and completes only steps of the groups they belong to (a pool role, the derived manager, a run
party, or themselves); any other step is not found.
"""
import base64
import json
from datetime import date, timedelta
from typing import Optional
from pydantic import BaseModel, ConfigDict
from erp_platform import work
from erp_platform.bricks import Refused
from erp_platform.types import BigId, Email, Limit, Name, Note
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 run as R
from brick_hr.domain import services
from brick_hr.domain.errors import RuleBroken
from brick_hr.domain.person import HireSpec, Person
from brick_hr.domain.values import PersonKey
from brick_hr.infrastructure import capability_acl as acl
HR = uc.HR
class Attachment(BaseModel):
"""A file a person attaches when completing a step (the signed contract). Streamed in, never a link."""
model_config = ConfigDict(extra="forbid")
filename: str
content_type: str = "application/pdf"
content_base64: str
class HireForm(BaseModel):
"""A hire about to be recorded, for a preview (the record_hire form)."""
model_config = ConfigDict(extra="forbid")
legal_given_name: str
legal_family_name: str
worker_type: str
job_code: str
org_unit_id: str
start_date: date
personal_email: Optional[str] = None
staff_no: Optional[str] = None
preferred_name: Optional[str] = None
org_id: Optional[str] = None
position_id: Optional[str] = None
location_code: Optional[str] = None
schedule_code: Optional[str] = None
planned_end: Optional[date] = None
rate_amount: Optional[str] = None
rate_basis: Optional[str] = None
currency: Optional[str] = None
event_type: str = "hire"
def _step_view(run: R.Run, s: R.RunStep, *, inputs: bool = True) -> dict:
out = {"run_id": run.run_id, "step_no": s.step_no, "kind": s.kind, "title": s.title,
"responsible_group": s.responsible_group, "state": s.state, "due_at": S.iso(s.due_at),
"depends_on": s.depends_on, "manual": s.manual, "capability": s.capability,
"destructive": s.destructive, "attempts": s.attempts, "last_status": s.last_status,
"provider_ref": s.provider_ref, "message": s.message, "done_by": s.done_by, "done_at": S.iso(s.done_at),
"approved": s.approved, "target": s.target, "idempotency_key": run.key(s) if s.capability else None}
if s.manual or s.kind in ("action", "document"):
out["plan"] = s.plan
out["manual_reason"] = s.manual_reason
if inputs and s.capability:
out["input"] = s.input
if s.kind == "draft":
out["draft_text"] = s.draft_text
if s.kind == "document":
out["contract_id"] = s.contract_id
out["phase"] = (s.result or {}).get("phase")
if s.outcomes: # who sees the step sees its outcome (customer_varies: the default)
out["outcomes"] = list(s.outcomes)
out["outcome"] = (s.result or {}).get("outcome")
return out
def _run_or_404(view: S.View, run_id: str) -> R.Run:
run = view.uow.runs.get(run_id)
if run is None or not view.sees_person(run.person_id):
raise Refused(404, f"no run {run_id}")
return run
def _mine(view: S.View, run: R.Run, s: R.RunStep) -> bool:
if view.companies is not None and run.person_id != view.me and not view.person_in_companies(run.person_id):
return False # hr v4: a group stops at its companies
groups = {s.responsible_group} | ({s.approval_group} if s.approval_group else set())
parties = {g: pid for g, (pid, _by) in run.parties.items()}
return any(view.in_group(g, run.person_id, parties) for g in groups)
def _step_or_404(view: S.View, run_id: str, step_no: str) -> tuple:
"""The step, if the caller belongs to its group (HR admins too) — else 404, never 403."""
run = view.uow.runs.get(run_id)
s = run.steps.get(str(step_no)) if run else None
if s is None or not (_mine(view, run, s) or (view.sees_all and S.has(view.caller, S.PERSON_WRITE)
and view.person_in_companies(run.person_id))):
raise Refused(404, f"no step {step_no} in run {run_id}")
return run, s
# ── reading ────────────────────────────────────────────────────────────────────────────────
@HR.use_case(kind="query", route="GET /my-tasks", permission=S.TASK_READ)
def my_hr_tasks(caller: dict) -> dict:
"""My tasks (HR-03): the open steps of the groups I belong to, as overdue, today and this week (in the org's
time zone), derived on read. A manual task (an action with no provider) shows the capability, the exact input
and the plan in words."""
with S.rules():
with uc.repository() as uow:
uow.prefetch(open_runs=True, seats=True) # the open runs and the directory: one round trip
view = S.View(uow, caller)
ids = sorted({rid for rid, _p, _g, _parties in view.open_steps})
items = []
runs = uow.runs.load_many(ids)
names = uow.persons.names({r.person_id for r in runs}, view.today)
for run in runs:
for s in run.ordered():
if s.finished or not run.deps_done(s) or not _mine(view, run, s):
continue
if run.cancelled_at is not None and not s.step_no.startswith("R"):
continue
v = _step_view(run, s)
v["person"] = {"person_id": run.person_id, "name": names.get(run.person_id, {}).get("name"),
"start_date": run.anchors.get("start_date").isoformat()
if hasattr(run.anchors.get("start_date"), "isoformat") else None}
v["_due"] = s.due_at
items.append(v)
b = work.board(items, now=view.now, timezone=view.tz, due=lambda i: i["_due"])
clean = lambda xs: [{k: v for k, v in x.items() if k != "_due"} for x in xs] # noqa: E731
return {"counts": {k: len(v) for k, v in b.items()}, "overdue": clean(b["overdue"]),
"today": clean(b["today"]), "this_week": clean(b["this_week"]),
"later": clean([i for i in items if i not in b["overdue"] + b["today"] + b["this_week"]])}
@HR.use_case(kind="query", route="GET /runs", permission=S.RUN_READ)
def run_board(caller: dict, direction: Optional[Name] = None, org_unit_id: Optional[Name] = None,
worker_type: Optional[Name] = None, include_done: bool = False) -> dict:
"""The run board (HR-04): every run you may see — person, direction, start or last day, steps done of total,
the next due step, 'waiting on <group>' and what is stuck (overdue, failed, waiting for approval or
signature), all derived on read. Filter by direction, org unit, worker type."""
with S.rules():
with uc.repository() as uow:
view = S.View(uow, caller)
runs = [r for r in uow.runs.load_many() if view.sees_person(r.person_id)]
people = {p.person_id: p for p in uow.persons.load_many({r.person_id for r in runs})}
out = []
for r in runs:
if direction and r.direction != direction:
continue
if not include_done and r.state != "open":
continue
p = people.get(r.person_id)
e = p.engagements.get(r.n) if p else None
a = p.assignments[r.n].at(min(max(view.today, e.start_date), e.end)) if e else None
if org_unit_id and (a is None or a.data.get("org_unit_id") != org_unit_id):
continue
if worker_type and (e is None or e.worker_type != worker_type):
continue
row = R.summary(r, view.now)
row.update(person=p.name(view.today).strip() if p else None,
start_date=e.start_date.isoformat() if e else None,
last_day=e.actual_end.isoformat() if e and e.actual_end else None,
worker_type=e.worker_type if e else None,
org_unit_id=a.data.get("org_unit_id") if a else None)
out.append(row)
out.sort(key=lambda x: (x["next_due"] or {}).get("due_at") or "9999")
return {"runs": out}
@HR.use_case(kind="query", route="GET /runs/{run_id}", permission=S.RUN_READ)
def show_run(caller: dict, run_id: Name) -> dict:
"""One run: its frozen playbook version, every step with its due time, group, state, input and plan, its
parties, and its log (reschedules, approvals, completions, chases, cancellations)."""
with S.rules():
with uc.repository() as uow:
view = S.View(uow, caller)
run = _run_or_404(view, run_id)
out = R.summary(run, view.now)
out["steps"] = [_step_view(run, s) for s in run.ordered()]
out["parties"] = {g: pid for g, (pid, _by) in run.parties.items()}
out["anchors"] = {k: (v.isoformat() if hasattr(v, "isoformat") else v) for k, v in run.anchors.items()}
out["log"] = [{"seq": l.seq, "what": l.what, "from": l.from_value, "to": l.to_value, "actor": l.actor,
"at": S.iso(l.event_at)} for l in run.log]
out["version"] = run.version
return out
def _preview_steps(view: S.View, run: R.Run, p: Person) -> list:
out = []
for s in run.ordered():
v = _step_view(run, s, inputs=False)
if s.kind == "action" and s.capability and not s.finished:
inp = engine.render_input(view, run, s, p)
o = acl.call(s.capability, run.key(s), inp, dry_run=True, actor=S.actor(view.caller))
v["input"] = inp
if o.status == "manual":
v["plan"] = f"manual: no provider — {o.plan}" if "no provider" in o.manual_reason else \
f"manual: {o.manual_reason} — {o.plan}"
else:
v["plan"] = o.plan or acl.describe(s.capability, inp)
v["provider"] = o.provider
out.append(v)
return out
@HR.use_case(kind="command", route="POST /runs/preview", permission=S.RUN_PREVIEW, read_only=True)
def preview_run(caller: dict, run_id: Optional[Name] = None, hire: Optional[HireForm] = None) -> dict:
"""Preview a run before it fires (HR-10, a dry run): for an existing run, or for a hire about to be recorded,
every step with its computed due time, responsible group, and for each action the exact capability input and
the provider's plan (or 'manual: no provider' and the plan in words). Nothing is written, claimed or sent."""
if (run_id is None) == (hire is None):
raise Refused(422, "preview an existing run (run_id) or a hire about to be recorded (hire), one of them")
at = S.now()
with S.rules():
with uc.repository() as uow:
view = S.View(uow, caller, at)
if run_id is not None:
run = _run_or_404(view, run_id)
p = uow.persons.get(run.person_id)
return {"run_id": run_id, "dry_run": True, "steps": _preview_steps(view, run, p)}
from brick_hr.application.people import _assignment, _rate
cfg = view.config
wt = cfg.worker_type(hire.worker_type)
if wt is None:
raise Refused(422, f"there is no worker type {hire.worker_type!r}")
a = _assignment(view, org_id=hire.org_id, org_unit_id=hire.org_unit_id, job_code=hire.job_code,
position_id=hire.position_id, location_code=hire.location_code,
schedule_code=hire.schedule_code, employment_pct=None, cost_centre=None,
d=hire.start_date)
key = PersonKey.of(hire.staff_no, hire.personal_email)
spec = HireSpec(hire.worker_type, hire.start_date, cfg.setting("default_legal_entity") or
view.orgs[a["org_id"]].legal_entity_ref, a, hire.planned_end,
_rate(hire.rate_amount, hire.rate_basis, hire.currency, view), None, None,
(1 if wt.get("via_company") else None), hire.event_type, None)
p = Person.from_hire(key, {"legal_given_name": hire.legal_given_name,
"legal_family_name": hire.legal_family_name,
"preferred_name": hire.preferred_name or hire.legal_given_name,
"personal_email": hire.personal_email}, spec, wt, today=view.today, now=at,
by=S.actor(caller))
view.directory.seats.extend(services.seats_of(p)) # in memory only: nothing is written
run = engine.build_run(view, p, p.events[0], at=at)
return {"run_id": run.run_id, "dry_run": True, "person_id": p.person_id,
"steps": _preview_steps(view, run, p)}
# ── acting ─────────────────────────────────────────────────────────────────────────────────
def _save_run(run: R.Run) -> None:
with uc.repository() as uow:
uow.runs.save(run)
@HR.use_case(kind="command", route="POST /runs/{run_id}/steps/{step_no}/complete", permission=S.TASK_COMPLETE,
idempotent=True)
def complete_step(caller: dict, run_id: Name, step_no: Name, provider_ref: Optional[Name] = None,
output: Optional[dict] = None, note: Optional[Note] = None,
attachment: Optional[Attachment] = None) -> dict:
"""Complete a step of one of my groups (HR-03): records who and when, and the steps that depend on it become
due. For a manual task (an action with no provider) give the provider_ref of what you made (optional) and its
output — e.g. the created primary_email, which is written back to the person; for a contract done by hand,
attach the signed copy. Completing a done step changes nothing."""
at = S.now()
follow = []
with S.rules():
with uc.repository() as uow:
view = S.View(uow, caller, at)
run, s = _step_or_404(view, run_id, step_no)
p = uow.persons.get(run.person_id)
if s.state == R.DONE:
return _step_view(run, s)
cal = engine._calendar(view, p, run.n)
if s.kind == "intake":
raise Refused(409, "the starter's details complete this step when they submit them")
if s.kind == "document":
if attachment is None:
raise Refused(422, "attach the signed copy to complete the contract step")
follow.append(("attach", attachment))
if s.kind == "action" and s.capability:
if not s.manual:
raise Refused(409, f"step {s.step_no} is done by the provider ({acl.provider(s.capability)}); "
"retry it if it is stuck")
if not run.deps_done(s):
raise Refused(409, f"step {s.step_no} waits on its dependencies first", gate="dependencies_done",
fix="complete the steps it depends on")
if s.destructive and not run.approval_granted(s):
raise Refused(409, f"step {s.step_no} is destructive and its approval has not been granted",
gate="approval_granted", fix="the approval before it must be granted first")
outcome = s.outcome_of(output)
o = acl.complete_manual(s.capability, run.key(s), s.input, s.plan, s.manual_reason,
provider_ref=provider_ref, output=output)
o = R.Outcome("done", provider=f"manual:{S.actor(caller)}", provider_ref=o.provider_ref,
output=o.output, message=o.message)
R.apply_outcome(run, s, o, now=at, max_attempts=99, backoff=timedelta(minutes=1), calendar=cal,
timezone=view.tz)
s.done_by = S.actor(caller)
if outcome:
s.result = {**(s.result or {}), "outcome": outcome}
run.add_log("completed", S.actor(caller), at, s.step_no, note or provider_ref or "by hand")
follow += engine.after_done(s)
elif s.kind != "document":
run.complete(s.step_no, by=S.actor(caller), at=at, provider_ref=provider_ref, output=output,
note=note or "", calendar=cal, timezone=view.tz)
if s.kind != "document":
uow.runs.save(run)
if not follow:
return _step_view(run, s) # as committed: nothing after it changed the run
for f in follow:
if f[0] == "attach":
a = f[1]
content = base64.b64decode(a.content_base64)
engine.file_contract(run_id, {"step_no": str(step_no), "contract_id": f"manual-{run_id}#{step_no}",
"file": {"content_base64": a.content_base64, "filename": a.filename,
"content_type": a.content_type}}, at)
del content
else:
engine.follow_up(run_id, str(step_no), f, at)
with uc.repository() as uow:
run = uow.runs.get(run_id)
return _step_view(run, run.steps[str(step_no)])
@HR.use_case(kind="command", route="POST /runs/{run_id}/steps/{step_no}/approve", permission=S.STEP_APPROVE,
idempotent=True)
def approve_step(caller: dict, run_id: Name, step_no: Name, granted: bool, reason: Optional[Note] = None) -> dict:
"""Approve or refuse a step explicitly (HR-07): approved, the destructive step behind it runs when due (a
delete only after the Drive transfer completed); refused, it is cancelled with your reason. Only someone of
the approval group with hr:step:approve can; the decision is in the run log with who and when. The same
decision twice is one approval."""
at = S.now()
with S.rules():
with uc.repository() as uow:
view = S.View(uow, caller, at)
run, s = _step_or_404(view, run_id, step_no)
if s.kind != "approval":
raise Refused(409, f"step {step_no} is not an approval")
if run.approve(s.step_no, granted, reason=reason or "", by=S.actor(caller), at=at):
uow.runs.save(run)
return _step_view(run, s)
@HR.use_case(kind="command", route="POST /runs/{run_id}/steps/{step_no}/approve-draft", permission=S.DRAFT_APPROVE,
idempotent=True)
def approve_draft(caller: dict, run_id: Name, step_no: Name, text: Optional[Note] = None) -> dict:
"""Edit and approve a draft (HR-05): the approved text becomes the input of the dependent step (the welcome
email is sent from your work mailbox to the starter's personal email — or, with no mail provider, a manual
task carrying the exact message). A handover list becomes one task per line for the takeover. A draft that
could not be made is written here by hand. Approving the same text twice changes nothing."""
at = S.now()
with S.rules():
with uc.repository() as uow:
view = S.View(uow, caller, at)
run, s = _step_or_404(view, run_id, step_no)
if s.kind != "draft":
raise Refused(409, f"step {step_no} is not a draft")
final = (text if text is not None else s.draft_text) or ""
if s.state == R.DONE:
if final.strip() == (s.draft_text or "").strip():
return _step_view(run, s)
raise Refused(409, f"step {step_no} was approved by {s.done_by}; it is no longer a draft")
if not final.strip():
raise Refused(422, "there is no draft yet — write the text and approve it")
if not run.deps_done(s):
raise Refused(409, f"step {step_no} waits on its dependencies first", gate="dependencies_done",
fix="complete the steps it depends on")
s.draft_text = final.strip()
s.state, s.done_by, s.done_at, s.last_status = R.DONE, S.actor(caller), at, "approved"
s.next_check_at = None
run.add_log("draft approved", S.actor(caller), at, s.step_no, f"{len(s.draft_text)} characters")
if s.draft_instruction == "handover":
for i, line in enumerate([x.strip(" -•\t") for x in s.draft_text.splitlines() if x.strip()], 1):
no = f"{s.step_no}:item:{i}"
if no not in run.steps:
run.steps[no] = R.RunStep(no, s.sort_no + i, "task", f"Take over: {line[:200]}", "takeover",
state=R.WAITING, due_at=at, depends_on=[s.step_no])
p = uow.persons.get(run.person_id)
run.on_done(s, engine._calendar(view, p, run.n), view.tz)
uow.runs.save(run)
return _step_view(run, s)
@HR.use_case(kind="command", route="POST /runs/{run_id}/parties", permission=S.RUN_RESCHEDULE, idempotent=True)
def assign_run_party(caller: dict, run_id: Name, group_code: Name, person_id: BigId) -> dict:
"""Name who a run-scoped group is for this run — the buddy, the takeover, the signing director. Their steps
then appear on that person's My tasks."""
at = S.now()
with S.rules():
with uc.repository() as uow:
view = S.View(uow, caller, at)
run = _run_or_404(view, run_id)
g = view.config.group(group_code)
if g is None or g.get("resolution") != "run_party":
raise Refused(422, f"{group_code!r} is not a run-party group")
if uow.persons.get(person_id) is None:
raise Refused(422, f"there is no person {person_id}")
if run.parties.get(group_code, (None,))[0] != int(person_id):
run.parties[group_code] = (int(person_id), S.actor(caller))
run.add_log("party", S.actor(caller), at, group_code, person_id)
uow.runs.save(run)
return {"run_id": run_id, "group_code": group_code, "person_id": int(person_id)}
@HR.use_case(kind="command", route="POST /runs/{run_id}/steps/{step_no}/retry", permission=S.ACTION_RETRY)
def retry_step(caller: dict, run_id: Name, step_no: Name) -> dict:
"""Retry an action step that failed or became a manual task (a provider is now connected, a value is fixed):
it is tried again now, through the provider if one is wired. A done step never acts again."""
at = S.now()
with S.rules():
with uc.repository() as uow:
view = S.View(uow, caller, at)
run, s = _step_or_404(view, run_id, step_no)
if s.kind not in ("action", "draft", "document"):
raise Refused(409, f"step {step_no} is a {s.kind}; only an action, a draft or a contract is retried")
if s.finished:
raise Refused(409, f"step {step_no} is {s.state}; it never acts again")
s.state, s.attempts, s.manual, s.manual_reason, s.message = R.PENDING, 0, False, "", "retry asked"
if s.kind == "draft":
s.draft_text = None
s.next_check_at = at
engine.mark_manual(run)
run.add_log("retry", S.actor(caller), at, s.step_no, "")
uow.runs.save(run)
engine.advance(run_id, at)
with uc.repository() as uow:
run = uow.runs.get(run_id)
return _step_view(run, run.steps[str(step_no)])
@HR.use_case(kind="command", route="POST /runs/{run_id}/cancel", permission=S.RUN_CANCEL, destructive=True,
idempotent=True)
def cancel_run(caller: dict, run_id: Name, reason: Note) -> dict:
"""Cancel a run when a starter withdraws (HR-23): every not-done step is cancelled; for each DONE action a
reversal task is created (suspend the account, remove from the group), named after the capability. Nothing is
reversed silently. The cancellation is in the run log. Cancelling twice changes nothing."""
at = S.now()
with S.rules():
with uc.repository() as uow:
view = S.View(uow, caller, at)
run = _run_or_404(view, run_id)
reversals = run.cancel(reason, by=S.actor(caller), at=at)
if reversals or run.log[-1].what == "cancelled" and run.log[-1].event_at == at:
uow.runs.save(run)
return {"run_id": run_id, "state": run.state, "cancel_reason": run.cancel_reason,
"reversals": [_step_view(run, r) for r in run.ordered() if r.step_no.startswith("R")]}
@HR.use_case(kind="query", route="GET /action-log", permission=S.ACTION_LOG)
def read_action_log(caller: dict, person_id: Optional[int] = None, run_id: Optional[Name] = None) -> dict:
"""What the system did (HR-09): each action attempt with the capability, the input, the result status, the
provider_ref, the message, who or what acted and when — automatic attempts and manual tasks alike. For a
person or a run; another person's log is not found to someone with no step there."""
if person_id is None and run_id is None:
raise Refused(422, "the action log of a person (person_id) or of a run (run_id)")
with S.rules():
with uc.repository() as uow:
view = S.View(uow, caller)
if run_id is not None:
runs = [_run_or_404(view, run_id)]
else:
if not view.sees_person(int(person_id)):
raise Refused(404, f"no person {person_id}")
runs = uow.runs.load_many(uow.runs.ids(int(person_id)))
entries = []
for run in runs:
for s in run.ordered():
if not s.capability:
continue
attempts = [l for l in run.log if l.from_value == s.step_no and l.what in (
"attempt failed", "action done", "manual task", "completed", "retry")]
for l in attempts:
entries.append({"run_id": run.run_id, "step_no": s.step_no, "capability": s.capability,
"input": s.input, "what": l.what, "detail": l.to_value, "actor": l.actor,
"at": S.iso(l.event_at)})
entries.append({"run_id": run.run_id, "step_no": s.step_no, "capability": s.capability,
"input": s.input, "what": "state", "status": s.last_status or s.state,
"state": s.state, "manual": s.manual, "provider_ref": s.provider_ref,
"message": s.message or s.manual_reason, "actor": s.done_by,
"at": S.iso(s.done_at), "idempotency_key": run.key(s)})
return {"entries": entries}
@HR.use_case(kind="command", route="POST /run-due-steps", permission=S.RUN_TICK)
def run_due_steps(caller: dict) -> dict:
"""The tick (HR-30), called by the clock every minute or so: expands runs for new events, makes due
future-dated changes reach the business partner, acts on every step that is due and whose dependencies are
done (a task shows as due; an action calls its capability with idempotency key <run_id>|<step_no>; a draft
asks Models; an intake issues the link; an approval waits for a person), chases what is late, warns of
expiries, and once a day applies the retention schedule. Two ticks at once act once per step."""
with S.rules():
return engine.run_due_steps(S.now())
__all__ = ["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", "json", "ev"]