"""The customer's process against the standard: playbook_deviations (how a playbook differs from its pack's
reference playbook, and what each difference means), scenario_paths (the scenarios that walk it, as step keys),
and change_playbook (one change at a time — what a chat or a diagram edit asks for — saved as a new version
through every invariant keep_playbook has, plus the kept foundation: domain/deviations.py).
"""
import functools
import importlib.resources
from typing import Literal, Optional
import yaml
from erp_platform.bricks import Refused
from erp_platform.repository import as_conflict
from erp_platform.types import Name, Quantity
from brick_hr.application import support as S
from brick_hr.application import use_cases as uc
from brick_hr.domain import deviations as dv
from brick_hr.domain import playbook as pbk
from brick_hr.domain.configuration import SETTINGS_DEFAULTS, Config
from brick_hr.domain.events import direction_of
from brick_hr.domain.packs import PACKS
from brick_hr.domain.playbook import Playbook
HR = uc.HR
Change = Literal["add_step", "remove_step", "set_owner", "set_timing", "remove_approval"]
# ── the reference ──────────────────────────────────────────────────────────────────────────
def direction(cfg: Config, pb: Playbook) -> Optional[str]:
kinds = [(cfg.event_type(et) or {}).get("kind") for et in pb.event_types]
ds = [d for d in (direction_of(k, (pb.reasons or [None])[0]) for k in kinds if k) if d]
return ds[0] if ds else None
@functools.lru_cache(maxsize=None)
def _pack_playbooks(pack: str) -> tuple:
from brick_hr.application.organisation import _playbook_from
return tuple(_playbook_from(b, 0, f"pack:{pack}", None) for b in PACKS[pack].starter_playbooks())
def reference_for(cfg: Config, pb: Playbook) -> Optional[tuple]:
"""(pack, the reference playbook) a customer's playbook is weighed against: the pack's playbook of the same id,
else the one for the same direction whose event and worker types overlap it most. None without a pack."""
pack = cfg.setting("pack") or ""
if pack not in PACKS:
return None
refs = _pack_playbooks(pack)
same = next((r for r in refs if r.playbook_id == pb.playbook_id), None)
if same is not None:
return pack, same
mine = direction(cfg, pb)
def overlap(r: Playbook) -> int: # an empty list of worker types is every worker type
if not pb.worker_types:
return len(r.worker_types) if r.worker_types else 99
return len(set(r.worker_types) & set(pb.worker_types)) if r.worker_types else len(pb.worker_types)
fits = [r for r in refs if direction(cfg, r) == mine and set(r.event_types) & set(pb.event_types)]
if not fits:
return None
return pack, max(fits, key=lambda r: (overlap(r), -refs.index(r)))
def deviations_of(cfg: Config, pb: Playbook) -> list:
found = reference_for(cfg, pb)
if found is None:
return []
pack, ref = found
mod = PACKS[pack]
values = {k: cfg.setting(k) for k in dv.WATCHED_SETTINGS}
return (dv.compare(pb, ref, gates=getattr(mod, "GATES", {}), safeguards=getattr(mod, "SAFEGUARDS", {}))
+ dv.settings(values, SETTINGS_DEFAULTS, pb))
def _load(uow, playbook_id: str, version: Optional[int] = None) -> Playbook:
v = version or uow.playbooks.latest_version(playbook_id)
pb = uow.playbooks.get(playbook_id, v) if v else None
if pb is None:
raise Refused(404, f"no playbook {playbook_id}" + (f" version {version}" if version else ""))
return pb
def report(cfg: Config, pb: Playbook) -> dict:
devs = deviations_of(cfg, pb)
found = reference_for(cfg, pb)
return {"playbook_id": pb.playbook_id, "version": pb.version,
"reference": {"pack": found[0], "playbook_id": found[1].playbook_id} if found else None,
"summary": dv.summary(devs), "deviations": [d.as_row() for d in devs]}
@HR.use_case(kind="query", route="GET /playbooks/{playbook_id}/deviations", permission=S.CONFIG_WRITE)
def playbook_deviations(caller: dict, playbook_id: Name, version: Optional[Quantity] = None) -> dict:
"""How a playbook (its latest version, or the version given) differs from the reference playbook of its
localisation pack (the same id, else the same direction and worker types), compared by step KEY: each step
added, removed or changed (owner group, timing, condition, an approval no longer waited for…), and each
watched setting (the contract signer, the chase cadence). Every deviation is classified: variation (inside
the defaults), warning (removes a best-practice safeguard — allowed, and flagged to sign knowingly) or
foundation (breaks what is kept — never stored: refused at save). Read-only."""
with S.rules():
with uc.repository() as uow:
pb = _load(uow, playbook_id, version)
return report(uow.config.load(), pb)
# ── scenarios as paths ─────────────────────────────────────────────────────────────────────
@functools.lru_cache(maxsize=None)
def declared_paths() -> tuple:
raw = yaml.safe_load(importlib.resources.files("brick_hr").joinpath("scenario_paths.yaml").read_text())
return tuple(raw["paths"])
@HR.use_case(kind="query", route="GET /playbooks/{playbook_id}/scenario-paths", permission=S.CONFIG_WRITE)
def scenario_paths(caller: dict, playbook_id: Name) -> dict:
"""The scenarios that walk this playbook (through its reference playbook), each as the ordered step keys it
traverses — what a viewer highlights on the BPMN (element Task_<key>) — and whether it still walks on the
latest version: each step there, each waiting for the one before (a removed or re-wired step breaks the
path, and the problem says how). Read-only."""
with S.rules():
with uc.repository() as uow:
pb = _load(uow, playbook_id)
found = reference_for(uow.config.load(), pb)
ref_id = found[1].playbook_id if found else pb.playbook_id
out = []
for p in declared_paths():
if p["playbook"] != ref_id:
continue
problem = pbk.walk_problem(pb, list(p["path"]))
out.append({"scenario_id": p["scenario"], "title": p["title"], "path": list(p["path"]),
"elements": [f"Task_{k}" for k in p["path"]], "walks": problem is None, "problem": problem})
return {"playbook_id": pb.playbook_id, "version": pb.version, "reference": ref_id, "paths": out}
# ── one change ─────────────────────────────────────────────────────────────────────────────
@HR.use_case(kind="command", route="POST /playbooks/{playbook_id}/changes", permission=S.CONFIG_WRITE, idempotent=True)
def change_playbook(caller: dict, playbook_id: Name, change: Change, key: Optional[Name] = None,
step: Optional[dict] = None, after: Optional[list[Name]] = None,
before: Optional[list[Name]] = None, responsible_group: Optional[Name] = None,
timing: Optional[dict] = None) -> dict:
"""One change to a playbook, by step KEY, saved as its next version (open runs keep theirs) — what "add a
director approval before the contract", "IT does the laptop, not the manager", "suspend at 18:00" or "drop
the rate approval" becomes. change: add_step (step = kind, title, responsible_group…; after = the keys it
waits for; before = the keys that then wait for it; it gets the next number, nothing is renumbered),
remove_step (what waited for it then waits for what it waited for), remove_approval, set_owner
(responsible_group), set_timing (timing = anchor, offset, unit, time — anchor may be step:<key>.done).
Refused in words like keep_playbook, and when it would break the kept foundation (an approval before a
destructive step, the transfer before a delete, a pack's legal gate). The answer carries the new version's
deviations from the reference."""
from brick_hr.application.organisation import body_of, save_playbook
at = S.now()
with S.rules():
try:
with uc.repository() as uow:
pb = _load(uow, playbook_id)
steps = pbk.edited(pb, change, key=key, step=step, after=after, before=before,
responsible_group=responsible_group, timing=timing)
new, _created = save_playbook(uow, {**body_of(pb), "steps": steps}, S.actor(caller), at)
answer = report(uow.config.load(), new)
except Exception as exc: # noqa: BLE001
from brick_hr.infrastructure.repositories import is_duplicate
if is_duplicate(exc) and as_conflict(exc) is None:
raise Refused(409, "someone saved a new version of this playbook first — reload it") from None
raise
if change == "add_step": # the new step (or the same one, asked twice)
st = step or {}
key = st.get("key") or next((s.key for s in sorted(new.steps, key=lambda s: -s.step_no)
if (s.title, s.kind) == (st.get("title"), st.get("kind"))), None)
return {**answer, "change": change, "key": key} # the state asked for: a replay answers the same
# ── the picture's words for the settings ───────────────────────────────────────────────────
def chase_words(cfg: Config) -> dict:
"""step kind → how a waiting step of that kind is chased, in words (engine.chase)."""
unit = "working day" if cfg.setting("chase_unit") == "working_days" else "day"
out = {}
for kind, key, then in (("task", "task_chase", "task overdue"), ("approval", "approval_chase", "approval overdue"),
("intake", "intake_chase", "starter details still missing")):
every, most = cfg.int_setting(f"{key}_every_days"), cfg.int_setting(f"{key}_max")
if most <= 0:
out[kind] = "Never chased."
else:
out[kind] = (f"Chased every {every} {unit}{'' if every == 1 else 's'}, up to {most} "
f"time{'' if most == 1 else 's'}, then a task for HR: {then}.")
return out
__all__ = ["playbook_deviations", "scenario_paths", "change_playbook", "deviations_of", "reference_for",
"declared_paths", "chase_words", "report"]