"""Monitoring the onboarding programme (HR-29, APQC 7.3.1.3 "Monitor and update employee onboarding program"):
what a run board cannot say in one look — how long runs take, where steps go overdue, how soon a starter can work,
and how far each playbook version strays from the standard. Read-only and DERIVED: computed from runs and their
steps on every read, never stored, so it cannot drift from the runs it describes.
lead time a finished run: the business date (business_timezone) its last step finished on, minus the
business date it was created on, in days. An open run counts its age instead.
overdue a step not finished whose due time has passed (RunStep.overdue) — by owner (the responsible
group) and by team (the person's org unit on the run's date).
first access an onboarding run: the business date its first access step finished (an account made, a group
or chat space joined) minus the start date, in days; negative is before day one, which is the aim.
deviations per playbook version the runs used: its deviations from the pack's reference by class
(variation / warning), and how many runs used it.
A median is the lower middle value (no half days); every date is the business date in business_timezone.
"""
from dataclasses import dataclass
from datetime import date, datetime
from typing import Callable, Iterable, Optional
from erp_platform import work
from brick_hr.domain.run import Run
# The steps that give a starter access: what "first access granted" waits for.
ACCESS_GRANTS = ("identity.provision_user", "groups.add_member", "chat.add_to_space", "drive.create_folder")
@dataclass(frozen=True)
class RunFacts:
"""What the monitor needs about a run's person, read alongside (never the Person aggregate itself)."""
start_date: Optional[date]
org_unit_id: Optional[str]
def business_date(at: datetime, tz: str) -> date:
return work.aware(at).astimezone(work.zone(tz)).date()
def _median(values: list) -> Optional[int]:
if not values:
return None
v = sorted(values)
return v[(len(v) - 1) // 2]
def _stats(values: list) -> dict:
return {"count": len(values), "median_days": _median(values), "max_days": max(values) if values else None}
def finished_on(run: Run) -> Optional[datetime]:
"""When a finished run's last step finished (None while it is open, or when no step records a time)."""
if run.state != "done":
return None
times = [s.done_at for s in run.steps.values() if s.done_at is not None]
return max(times, key=work.aware) if times else None
def first_access(run: Run) -> Optional[datetime]:
times = [s.done_at for s in run.steps.values()
if s.capability in ACCESS_GRANTS and s.state == "done" and s.done_at is not None]
return min(times, key=work.aware) if times else None
def monitor(runs: Iterable[Run], facts: Callable[[Run], RunFacts], now: datetime, tz: str,
deviations: Callable[[str, int], Optional[dict]]) -> dict:
"""The onboarding programme's numbers over `runs` (already filtered to what the caller may see).
`deviations(playbook_id, version)` → {variation: n, warning: n, ...} or None when the version has no reference."""
today = business_date(now, tz)
lead, ages, access, per_run = [], [], [], []
by_owner: dict = {}
by_team: dict = {}
versions: dict = {}
for r in sorted(runs, key=lambda x: x.run_id):
if r.state == "cancelled":
continue
f = facts(r)
created = business_date(r.created_at, tz)
done_at = finished_on(r)
lead_days = (business_date(done_at, tz) - created).days if done_at else None
if lead_days is not None:
lead.append(lead_days)
elif r.state == "open":
ages.append((today - created).days)
acc = first_access(r)
acc_days = (business_date(acc, tz) - f.start_date).days if acc and f.start_date else None
if acc_days is not None:
access.append(acc_days)
overdue = [s for s in r.steps.values() if s.overdue(now)]
for s in overdue:
by_owner[s.responsible_group] = by_owner.get(s.responsible_group, 0) + 1
team = f.org_unit_id or "(none)"
by_team[team] = by_team.get(team, 0) + 1
if r.playbook_id is not None:
k = (r.playbook_id, int(r.playbook_version or 0))
v = versions.setdefault(k, {"playbook_id": k[0], "version": k[1], "runs": 0, "open": 0})
v["runs"] += 1
v["open"] += 1 if r.state == "open" else 0
per_run.append({"run_id": r.run_id, "person_id": r.person_id, "state": r.state,
"playbook_id": r.playbook_id, "playbook_version": r.playbook_version,
"start_date": f.start_date.isoformat() if f.start_date else None,
"org_unit_id": f.org_unit_id, "lead_time_days": lead_days,
"first_access_days": acc_days, "steps_overdue": len(overdue)})
for (pid, ver), v in versions.items():
v["deviations"] = deviations(pid, ver)
return {
"as_of": today.isoformat(),
"lead_time": {**_stats(lead), "open_runs": len(ages), "open_median_age_days": _median(ages)},
"first_access": _stats(access),
"steps_overdue": {"total": sum(by_owner.values()),
"by_owner": [{"responsible_group": g, "overdue": n} for g, n in sorted(by_owner.items())],
"by_team": [{"org_unit_id": t, "overdue": n} for t, n in sorted(by_team.items())]},
"playbook_versions": [versions[k] for k in sorted(versions)],
"runs": per_run,
}
__all__ = ["monitor", "RunFacts", "ACCESS_GRANTS", "first_access", "finished_on", "business_date"]