"""The Playbook aggregate: the customer's checklist for one kind of run, versioned (hr_playbook playbook_id|version
with its hr_playbook_step rows).
Invariants, refused on save (409, in words):
* an active version is immutable once a run has used it — a change is always a NEW version;
* a destructive capability step depends (directly or through others) on an approval step;
* identity.delete_user depends on a drive.transfer_ownership step in the same playbook (the engine then waits on
drive.transfer_status = completed);
* every step has a KEY — a slug (send_contract), unique in the playbook and never renumbered: what a BPMN id, a
scenario path, a deviation and a signed change item join on (step numbers are positions; keys are names);
* depends_on forms no cycle; step:<n>.done anchors name an existing step;
* a conversion playbook removes no access (a conversion keeps it);
* playbook selection — worker type × event type × reason × org unit — has one answer: the most specific match
wins, and two playbooks that would tie are refused on save.
Step conditions are named specifications from a fixed set in code, chosen by a row:
if: worker_type_is(types) · via_company · rate_outside_band · client_size_is(sizes) · reason_is(reasons)
→ else skipped (reason_is, 0.6.0: the event's reason — the outplacement offer on a redundancy)
requires: pack_requires(record) [if …] → the step is BLOCKED until that compliance record exists
"""
import re
from dataclasses import dataclass, field
from datetime import datetime
from typing import Callable, Optional
from erp_platform import work
from brick_hr.domain.errors import Incomplete, NotFound, RuleBroken
STEP_KINDS = ("task", "action", "document", "draft", "intake", "approval")
BASE_ANCHORS = ("start_date", "last_day", "event_date", "run.created")
SKIP_SPECS = ("worker_type_is", "via_company", "rate_outside_band", "client_size_is", "reason_is")
OUTCOME_KINDS = ("task", "action") # a step that records an outcome (accepted / declined) is done by someone
GATE_SPECS = ("pack_requires",)
FAN_OUTS = ("access:group", "access:chat_space", "access:drive_folder")
# Capabilities that take access away. A conversion keeps access, so its playbook may contain none of them.
ACCESS_REMOVAL = ("identity.suspend_user", "identity.delete_user", "groups.remove_member", "chat.remove_from_space",
"drive.transfer_ownership", "mail.set_autoreply")
ADDS = ("groups.add_member", "chat.add_to_space", "drive.create_folder")
REMOVES = ("groups.remove_member", "chat.remove_from_space")
KEY = re.compile(r"^[a-z][a-z0-9]*(_[a-z0-9]+)*$") # a slug; no double underscore (BPMN ids use it as a separator)
KEY_MAX = 60
@dataclass
class PlaybookStep:
step_no: int
kind: str
title: str
responsible_group: str
timing: dict = field(default_factory=lambda: {"anchor": "run.created", "offset": 0, "unit": "days", "time": None})
capability: Optional[str] = None
input_template: Optional[dict] = None
depends_on: list = field(default_factory=list)
condition: Optional[dict] = None
approval_group: Optional[str] = None
draft_instruction: Optional[str] = None
template_kind: Optional[str] = None
fan_out: Optional[str] = None
key: str = ""
outcomes: list = field(default_factory=list) # 0.6.0: the answers completing it must give (accepted, declined)
def rule(self, timezone: str) -> work.TimingRule:
t = self.timing or {}
return work.TimingRule(t.get("anchor") or "run.created", int(t.get("offset") or 0), t.get("time"),
timezone, t.get("unit") or "days")
@dataclass
class Playbook:
playbook_id: str
version: int
name: str
event_types: list
worker_types: list = field(default_factory=list)
reasons: list = field(default_factory=list)
org_unit_id: Optional[str] = None
status: str = "active"
created_by: str = ""
created_at: Optional[datetime] = None
steps: list = field(default_factory=list)
def step(self, n: int) -> Optional[PlaybookStep]:
return next((s for s in self.steps if s.step_no == n), None)
def by_key(self, key: str) -> Optional[PlaybookStep]:
return next((s for s in self.steps if s.key == key), None)
def keys_of(self, nos) -> list:
"""Step numbers → their keys (sorted by key), for comparing two playbooks by name, not position."""
by = {s.step_no: s.key for s in self.steps}
return sorted(by[n] for n in nos if n in by)
@property
def specificity(self) -> tuple:
return (bool(self.worker_types), bool(self.reasons), self.org_unit_id is not None)
def matches(self, event_type: str, worker_type: str, reason: Optional[str], units: list) -> bool:
return (event_type in self.event_types
and (not self.worker_types or worker_type in self.worker_types)
and (not self.reasons or (reason or "") in self.reasons)
and (self.org_unit_id is None or self.org_unit_id in units))
def content(self) -> tuple:
"""What makes two versions the same checklist (a save of the same content is no new version)."""
return (self.name, sorted(self.event_types), sorted(self.worker_types), sorted(self.reasons), self.org_unit_id,
[(s.step_no, s.key, s.kind, s.title, s.responsible_group, s.capability, _canon(s.input_template),
_canon(s.timing), sorted(s.depends_on), _canon(s.condition), s.approval_group, s.draft_instruction,
s.template_kind, s.fan_out, sorted(s.outcomes or [])) for s in sorted(self.steps, key=lambda s: s.step_no)])
# ── invariants ──────────────────────────────────────────────────────────────────────────
def validate(self, *, known: Callable[[str], bool], destructive: Callable[[str], bool], kind_of_event: Callable,
groups: Callable[[str], bool]) -> None:
if not self.playbook_id or not self.name:
raise Incomplete("a playbook needs an id and a name")
if not self.event_types:
raise Incomplete("a playbook names the event types it is for")
if not self.steps:
raise Incomplete("a playbook has at least one step")
nos = [s.step_no for s in self.steps]
if len(set(nos)) != len(nos) or any(n < 1 for n in nos):
raise Incomplete("step numbers are unique whole numbers from 1")
keys = [s.key for s in self.steps]
for s in self.steps:
if not s.key or len(s.key) > KEY_MAX or not KEY.match(s.key):
raise Incomplete(f"step {s.step_no}: its key {s.key!r} is a slug of lower-case letters, digits and "
f"single underscores, starting with a letter (at most {KEY_MAX})")
if len(set(keys)) != len(keys):
twice = sorted({k for k in keys if keys.count(k) > 1})
raise RuleBroken(f"two steps have the key {', '.join(twice)}: a key names ONE step of the playbook")
by = {s.step_no: s for s in self.steps}
kinds = {kind_of_event(et) for et in self.event_types}
for s in self.steps:
where = f"step {s.step_no} ({s.title or s.kind})"
if s.kind not in STEP_KINDS:
raise Incomplete(f"{where}: a step is one of {', '.join(STEP_KINDS)}")
if not s.title:
raise Incomplete(f"{where}: a step needs a title")
if not s.responsible_group or not groups(s.responsible_group):
raise RuleBroken(f"{where}: there is no responsible group {s.responsible_group!r}")
if s.kind in ("action", "document"):
if not s.capability or not known(s.capability):
raise RuleBroken(f"{where}: {s.capability!r} is not a capability in capabilities.yaml")
elif s.capability:
raise Incomplete(f"{where}: only an action or document step names a capability")
if s.kind == "document" and not s.capability.startswith("contracts."):
raise RuleBroken(f"{where}: a document step asks contracts (contracts.draft)")
if s.kind == "document" and not s.template_kind:
raise Incomplete(f"{where}: a document step names the contract type (template_kind)")
if s.kind == "draft" and not s.draft_instruction:
raise Incomplete(f"{where}: a draft step names its draft instruction")
if s.kind == "approval" and not (s.approval_group or s.responsible_group):
raise Incomplete(f"{where}: an approval names who approves")
if s.outcomes:
if s.kind not in OUTCOME_KINDS:
raise Incomplete(f"{where}: only a task or an action records an outcome")
if (not isinstance(s.outcomes, list) or len(s.outcomes) < 2 or len(set(s.outcomes)) != len(s.outcomes)
or not all(isinstance(o, str) and KEY.match(o) for o in s.outcomes)):
raise Incomplete(f"{where}: outcomes are two or more different slugs (accepted, declined)")
if s.fan_out and s.fan_out not in FAN_OUTS:
raise Incomplete(f"{where}: fan_out is one of {', '.join(FAN_OUTS)}")
if s.fan_out and s.kind != "action":
raise Incomplete(f"{where}: only an action step fans out per access row")
t = s.timing or {}
anchor = t.get("anchor") or "run.created"
if anchor.startswith("step:"):
try:
ref = int(anchor[5:].split(".")[0])
except ValueError:
raise Incomplete(f"{where}: anchor {anchor!r} is step:<n>.done") from None
if ref not in by or not anchor.endswith(".done"):
raise RuleBroken(f"{where}: anchor {anchor} names no step of this playbook")
elif anchor not in BASE_ANCHORS:
raise Incomplete(f"{where}: an anchor is one of {', '.join(BASE_ANCHORS)} or step:<n>.done")
try:
s.rule("UTC")
except ValueError as e:
raise Incomplete(f"{where}: {e}") from None
for d in s.depends_on:
if d not in by:
raise RuleBroken(f"{where}: depends on step {d}, which this playbook does not have")
_check_condition(where, s.condition)
if s.kind == "action" and s.capability in ACCESS_REMOVAL and "Converted" in kinds:
raise RuleBroken(f"{where}: a conversion keeps access — {s.capability} has no place in its playbook")
self._check_cycles(by)
for s in self.steps:
if s.kind in ("action", "document") and destructive(s.capability):
if not any(by[a].kind == "approval" for a in self.ancestors(s.step_no)):
raise RuleBroken(f"step {s.step_no} ({s.capability}) is destructive: it must depend on an "
"approval step — an account is never deleted by a timer")
if s.capability == "identity.delete_user":
if not any(by[a].capability == "drive.transfer_ownership" for a in self.ancestors(s.step_no)):
raise RuleBroken(f"step {s.step_no} deletes the account: it must depend on a "
"drive.transfer_ownership step, so nothing they own is lost")
def ancestors(self, n: int) -> set:
by = {s.step_no: s for s in self.steps}
out, frontier = set(), list(by[n].depends_on)
while frontier:
x = frontier.pop()
if x in out or x not in by:
continue
out.add(x)
frontier += by[x].depends_on
return out
def _check_cycles(self, by: dict) -> None:
state: dict = {}
def visit(n, path):
if state.get(n) == "done":
return
if state.get(n) == "on":
raise RuleBroken(f"the steps depend on each other in a circle: {' → '.join(map(str, path + [n]))}")
state[n] = "on"
for d in by[n].depends_on:
visit(d, path + [n])
anchor = (by[n].timing or {}).get("anchor") or ""
if anchor.startswith("step:"):
ref = int(anchor[5:].split(".")[0])
visit(ref, path + [n])
state[n] = "done"
for n in by:
visit(n, [])
def _check_condition(where: str, cond: Optional[dict]) -> None:
if not cond:
return
if not isinstance(cond, dict) or set(cond) - {"if", "requires"}:
raise Incomplete(f"{where}: a condition is {{if: [...], requires: [...]}}")
for spec in cond.get("if") or []:
if not isinstance(spec, dict) or spec.get("spec") not in SKIP_SPECS:
raise Incomplete(f"{where}: a condition is one of {', '.join(SKIP_SPECS)}")
for gate in cond.get("requires") or []:
if not isinstance(gate, dict) or gate.get("spec") not in GATE_SPECS or not gate.get("record"):
raise Incomplete(f"{where}: a requirement is pack_requires with the record it needs")
for spec in gate.get("if") or []:
if not isinstance(spec, dict) or spec.get("spec") not in SKIP_SPECS:
raise Incomplete(f"{where}: a requirement's condition is one of {', '.join(SKIP_SPECS)}")
def slug(title: str) -> str:
"""A key from a title: 'Director approves the contract' → director_approves_the_contract."""
out = re.sub(r"[^a-z0-9]+", "_", (title or "").lower()).strip("_")[:KEY_MAX].strip("_")
return out if out and out[0].isalpha() else ("step_" + out).strip("_")
def assign_keys(steps: list, previous: Optional[Playbook] = None) -> None:
"""Give every step without a key one, IN PLACE, so a key is never renumbered: a step keeps the key it had in the
previous version (matched by title, else by number, kind and capability), and a new step gets a slug of its
title, made unique — never reusing a key the previous version gave another step (a renamed concept gets a new
key; the old one is not reused). A key given explicitly is kept as given (validate checks it)."""
prev = list(previous.steps) if previous else []
taken = {s.key for s in steps if s.key}
retired = {p.key for p in prev if p.key}
for s in steps:
if s.key:
continue
match = next((p for p in prev if p.key and p.key not in taken and p.title == s.title), None) or \
next((p for p in prev if p.key and p.key not in taken and p.step_no == s.step_no and p.kind == s.kind
and p.capability == s.capability), None)
if match is not None:
s.key = match.key
else:
base, i = slug(s.title or s.kind), 1
s.key = base
while s.key in taken or s.key in retired:
i += 1
s.key = f"{base[:KEY_MAX - 1 - len(str(i))]}_{i}"
taken.add(s.key)
# ── one change at a time (what a chat or a diagram edit asks for), by KEY ─────────────────────────────────────────
CHANGES = ("add_step", "remove_step", "set_owner", "set_timing", "remove_approval")
def edited(pb: Playbook, change: str, *, key: Optional[str] = None, step: Optional[dict] = None,
after: Optional[list] = None, before: Optional[list] = None, responsible_group: Optional[str] = None,
timing: Optional[dict] = None) -> list:
"""The steps of `pb` with ONE change applied, as step dicts (keep_playbook's shape) — the playbook itself is
not touched; saving them makes the next version, through the same invariants as any save.
add_step `step` (kind, title, responsible_group, … and an optional key), depending on the steps
named in `after`; the steps named in `before` then wait for it. It gets the next free number:
no step is renumbered.
remove_step the step `key`; whatever waited for it now waits for what it waited for (the chain holds).
remove_approval remove_step for an approval (refused for any other kind).
set_owner the step `key` is done by `responsible_group` (an approval: approved by it).
set_timing the step `key`'s timing rule, merged: anchor (start_date | last_day | event_date |
run.created | step:<key>.done), offset, unit (days | working_days), time (HH:MM).
"""
if change not in CHANGES:
raise Incomplete(f"a change is one of {', '.join(CHANGES)}")
steps = [dict(s.__dict__, timing=dict(s.timing or {}), depends_on=list(s.depends_on)) for s in pb.steps]
by = {s["key"]: s for s in steps}
def number(k: str) -> int:
if k not in by:
raise NotFound(f"no step {k!r} in {pb.playbook_id} (its keys: {', '.join(sorted(by))})")
return by[k]["step_no"]
if change == "add_step":
if not isinstance(step, dict):
raise Incomplete("add_step needs the step: kind, title, responsible_group (and timing, capability…)")
if "step_no" in step or "depends_on" in step:
raise Incomplete("add_step numbers the step itself; say where it goes with after and before (step keys)")
new = dict(step, step_no=max(s["step_no"] for s in steps) + 1, depends_on=[number(k) for k in after or []])
if new.get("timing"):
new["timing"] = _timing(dict(new["timing"]), number)
same = next((s for s in steps if (s["key"] == new.get("key") if new.get("key") else
(s["title"], s["kind"]) == (new.get("title"), new.get("kind")))), None)
if same is not None: # asked twice: the step is there — the same steps, no new version
if sorted(same["depends_on"]) == sorted(new["depends_on"]) and same["responsible_group"] == \
new.get("responsible_group"):
return steps
raise RuleBroken(f"there is already a step {same['key']!r} ({same['title']}); a key names one step — "
"change it with set_owner or set_timing, or give the new one its own key")
for k in before or []:
number(k)
by[k]["depends_on"] = sorted(set(by[k]["depends_on"]) | {new["step_no"]})
return steps + [new]
if not key:
raise Incomplete(f"{change} names the step by its key")
n = number(key)
target = by[key]
if change in ("remove_step", "remove_approval"):
if change == "remove_approval" and target["kind"] != "approval":
raise RuleBroken(f"{key!r} is a {target['kind']}, not an approval")
anchored = [s["key"] for s in steps if (s.get("timing") or {}).get("anchor", "").startswith(f"step:{n}.")]
if anchored:
raise RuleBroken(f"{', '.join(anchored)} is timed from {key!r}; re-time it first (set_timing)")
rest = [s for s in steps if s["key"] != key]
for s in rest:
if n in s["depends_on"]:
s["depends_on"] = sorted((set(s["depends_on"]) - {n}) | set(target["depends_on"]))
return rest
if change == "set_owner":
if not responsible_group:
raise Incomplete("set_owner names the responsible_group")
target["responsible_group"] = responsible_group
if target["kind"] == "approval":
target["approval_group"] = responsible_group
return steps
if not isinstance(timing, dict) or not timing:
raise Incomplete("set_timing gives the timing: anchor, offset, unit, time")
extra = sorted(set(timing) - {"anchor", "offset", "unit", "time"})
if extra:
raise Incomplete(f"a timing rule has anchor, offset, unit and time; not {', '.join(extra)}")
target["timing"] = _timing({**target["timing"], **timing}, number)
return steps
def _timing(t: dict, number: Callable) -> dict:
"""A step:<key>.done anchor, as the rows store it (step:<n>.done)."""
anchor = t.get("anchor") or "run.created"
if anchor.startswith("step:") and anchor.endswith(".done"):
ref = anchor[5:-5]
if not ref.isdigit():
t["anchor"] = f"step:{number(ref)}.done"
return t
def walk_problem(pb: Playbook, path: list) -> Optional[str]:
"""Is `path` (step keys, in order) a walk of the playbook's graph — every step there, each reachable from the
one before it by dependencies? None when it is; else what is wrong, in words."""
if not path:
return "the path is empty"
nos = {s.key: s.step_no for s in pb.steps}
missing = [k for k in path if k not in nos]
if missing:
return f"no step {', '.join(map(repr, missing))} in {pb.playbook_id} v{pb.version}"
for a, b in zip(path, path[1:]):
if nos[a] not in pb.ancestors(nos[b]):
return f"{b!r} does not wait for {a!r}: there is no flow from one to the other"
return None
def ties(a: Playbook, b: Playbook) -> bool:
"""Would some event match both at the same specificity? Then selection has no single answer."""
if a.playbook_id == b.playbook_id or a.status != "active" or b.status != "active":
return False
if a.specificity != b.specificity:
return False
def meet(x: list, y: list) -> bool:
return (not x and not y) or bool(set(x) & set(y))
return (bool(set(a.event_types) & set(b.event_types)) and meet(a.worker_types, b.worker_types)
and meet(a.reasons, b.reasons) and a.org_unit_id == b.org_unit_id)
def select(playbooks: list, event_type: str, worker_type: str, reason: Optional[str], units: list) -> Optional[Playbook]:
"""The most specific active playbook (latest version of each) matching the event, or None."""
matching = [p for p in playbooks if p.status == "active" and p.matches(event_type, worker_type, reason, units)]
if not matching:
return None
depth = {u: i for i, u in enumerate(units)}
return sorted(matching, key=lambda p: (sum(p.specificity), -depth.get(p.org_unit_id, len(units))),
reverse=True)[0]
def _canon(v):
import json
return json.dumps(v, sort_keys=True, default=str) if v is not None else None
__all__ = ["Playbook", "PlaybookStep", "assign_keys", "slug", "KEY", "CHANGES", "edited", "walk_problem", "STEP_KINDS", "ACCESS_REMOVAL", "ADDS", "REMOVES", "ties", "select"]