"""The Run aggregate: a playbook expanded for one person and one personnel event (hr_run + hr_run_step,
hr_run_party, hr_run_log). And the domain services that make and move runs: RunExpansion (the RunFactory),
the Retimer, outcome handling, the derived board.
Invariants, enforced here:
* the playbook version is frozen at expansion: later playbook edits never change an open run;
* a destructive step never fires without a GRANTED approval it depends on;
* a done step is never re-timed and never acts again;
* each action/document step carries idempotency_key = <run_id>|<step_no>;
* done, stuck, overdue and "waiting on <group>" are DERIVED on read; nothing stores them.
Step states: pending → waiting → done | skipped | cancelled.
pending not yet due, or due and about to be acted on by the tick
waiting on a person (a task, a manual task, an approval, a draft to approve) or on something outside
(an action in progress, a contract out for signature)
"""
import dataclasses
import json
import re
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from typing import Any, Callable, Optional
from erp_platform import work
from brick_hr.domain.errors import Incomplete, RuleBroken
from brick_hr.domain.playbook import ADDS, REMOVES, Playbook, PlaybookStep
PENDING, WAITING, DONE, SKIPPED, CANCELLED = "pending", "waiting", "done", "skipped", "cancelled"
STATES = (PENDING, WAITING, DONE, SKIPPED, CANCELLED)
FINISHED = (DONE, SKIPPED, CANCELLED)
POLL = timedelta(minutes=1)
# What undoes a done action, when a run is cancelled (HR-23). Nothing is reversed silently: each is a TASK.
REVERSALS = {
"identity.provision_user": ("identity.suspend_user", "suspend the account"),
"groups.add_member": ("groups.remove_member", "remove them from the group"),
"chat.add_to_space": ("chat.remove_from_space", "remove them from the chat space"),
"drive.create_folder": (None, "archive or delete the folder"),
"calendar.create_event": (None, "cancel the calendar event"),
"identity.invite_app_user": (None, "remove the pool user"),
}
def _aware(t: Optional[datetime]) -> Optional[datetime]:
return work.aware(t) if t is not None else None
@dataclass
class RunStep:
step_no: str
sort_no: int
kind: str
title: str
responsible_group: str
playbook_step_no: Optional[int] = None
capability: Optional[str] = None
template: Optional[dict] = None
input: dict = field(default_factory=dict)
plan: str = ""
manual: bool = False
manual_reason: str = ""
timing: dict = field(default_factory=dict)
due_at: Optional[datetime] = None
state: str = PENDING
attempts: int = 0
last_status: Optional[str] = None
provider_ref: Optional[str] = None
result: dict = field(default_factory=dict)
contract_id: Optional[str] = None
draft_text: Optional[str] = None
done_by: Optional[str] = None
done_at: Optional[datetime] = None
depends_on: list = field(default_factory=list)
condition: Optional[dict] = None
approval_group: Optional[str] = None
destructive: bool = False
message: str = ""
next_check_at: Optional[datetime] = None
approved: Optional[bool] = None
target: Optional[str] = None
template_kind: Optional[str] = None
draft_instruction: Optional[str] = None
outcomes: list = field(default_factory=list) # 0.6.0: completing it records one of these as result.outcome
@property
def finished(self) -> bool:
return self.state in FINISHED
def outcome_of(self, output: Optional[dict]) -> Optional[str]:
"""A step with outcomes is completed WITH one of them (output.outcome); any other answer is refused."""
if not self.outcomes:
return None
o = (output or {}).get("outcome")
if o not in self.outcomes:
raise Incomplete(f"step {self.step_no} records its outcome: give output.outcome, one of "
f"{', '.join(self.outcomes)}")
return o
def overdue(self, now: datetime) -> bool:
return not self.finished and self.due_at is not None and _aware(self.due_at) < work.aware(now)
@dataclass
class RunLog:
seq: int
what: str
from_value: Optional[str]
to_value: Optional[str]
actor: str
event_at: datetime
@dataclass
class Run:
run_id: str
direction: str
person_id: int
n: int
event_seq: str
playbook_id: Optional[str]
playbook_version: Optional[int]
created_at: datetime
anchors: dict = field(default_factory=dict)
version: int = 0
cancelled_at: Optional[datetime] = None
cancel_reason: Optional[str] = None
cancelled_by: Optional[str] = None
steps: dict = field(default_factory=dict)
parties: dict = field(default_factory=dict) # group_code -> (person_id, set_by)
log: list = field(default_factory=list)
# ── reading (all derived) ───────────────────────────────────────────────────────────────
def ordered(self) -> list:
return sorted(self.steps.values(), key=lambda s: (s.sort_no, s.step_no))
def step(self, step_no: str) -> RunStep:
s = self.steps.get(str(step_no))
if s is None:
raise RuleBroken(f"run {self.run_id} has no step {step_no}")
return s
@property
def done(self) -> bool:
return self.cancelled_at is None and all(s.finished for s in self.steps.values())
@property
def state(self) -> str:
if self.cancelled_at is not None:
return "cancelled"
return "done" if self.done else "open"
def key(self, step: RunStep) -> str:
return f"{self.run_id}|{step.step_no}"
def deps_done(self, s: RunStep) -> bool:
return all(self.steps[d].state in (DONE, SKIPPED) for d in s.depends_on if d in self.steps)
def approval_granted(self, s: RunStep) -> bool:
"""A destructive step may act only when an approval it depends on (directly or not) was granted."""
seen, frontier = set(), list(s.depends_on)
while frontier:
d = frontier.pop()
if d in seen or d not in self.steps:
continue
seen.add(d)
dep = self.steps[d]
if dep.kind == "approval" and dep.state == DONE and dep.approved is True:
return True
frontier += dep.depends_on
return False
def due(self, now: datetime) -> list:
"""Steps whose time has come and whose dependencies are done — what the tick acts on."""
out = []
for s in self.ordered():
if s.state != PENDING or not self.deps_done(s):
continue
when = s.next_check_at or s.due_at
if when is not None and _aware(when) <= work.aware(now):
out.append(s)
return out
def dependants(self, step_no: str) -> list:
return [s for s in self.ordered() if step_no in s.depends_on]
def add_log(self, what: str, actor: str, at: datetime, from_value: Any = None, to_value: Any = None) -> RunLog:
seq = max([l.seq for l in self.log] or [0]) + 1
entry = RunLog(seq, what, None if from_value is None else str(from_value),
None if to_value is None else str(to_value), actor, at)
self.log.append(entry)
return entry
# ── commands ────────────────────────────────────────────────────────────────────────────
def _open(self) -> None:
if self.cancelled_at is not None:
raise RuleBroken(f"run {self.run_id} was cancelled ({self.cancel_reason})")
def complete(self, step_no: str, *, by: str, at: datetime, provider_ref: Optional[str] = None,
output: Optional[dict] = None, note: str = "", calendar=None, timezone: str = "UTC") -> bool:
"""A person finished a step (or a manual task). False when it was already done: never acts again."""
s = self.step(step_no)
if s.state == DONE:
return False
if s.state in (SKIPPED, CANCELLED):
raise RuleBroken(f"step {s.step_no} is {s.state}; there is nothing to complete")
if s.kind == "approval":
raise RuleBroken("an approval is granted or refused (approve_step), not completed")
if s.kind == "draft" and not s.manual:
raise RuleBroken("a draft is approved (approve_draft), not completed")
if self.cancelled_at is not None and not s.step_no.startswith("R"):
self._open() # after a cancel only the reversal tasks are still to do
if not self.deps_done(s):
waiting = [d for d in s.depends_on if self.steps[d].state not in (DONE, SKIPPED)]
raise RuleBroken(f"step {s.step_no} waits on step(s) {', '.join(waiting)} first")
outcome = s.outcome_of(output)
s.state, s.done_by, s.done_at = DONE, by, at
if outcome:
s.result = {**(s.result or {}), "outcome": outcome}
s.provider_ref = provider_ref or s.provider_ref
if output:
s.result = {**(s.result or {}), "output": output}
s.last_status = "done"
s.next_check_at = None
self.add_log("completed", by, at, s.step_no, note or (provider_ref or ""))
self.on_done(s, calendar, timezone)
return True
def on_done(self, s: RunStep, calendar=None, timezone: str = "UTC") -> None:
"""A step finished: steps timed from it (step:<n>.done) get their due time now."""
if s.playbook_step_no is None or s.done_at is None:
return
anchor = work.step_anchor(s.playbook_step_no)
if all(x.state in (DONE, SKIPPED) for x in self.steps.values() if x.playbook_step_no == s.playbook_step_no):
self.anchors[anchor] = s.done_at
timed = [x for x in self.steps.values() if (x.timing or {}).get("anchor") == anchor and not x.finished]
for x in timed:
rule = PlaybookStep(0, x.kind, x.title, x.responsible_group, timing=x.timing).rule(timezone)
x.due_at = rule.due_at(self.anchors, calendar)
def approve(self, step_no: str, granted: bool, *, reason: str, by: str, at: datetime) -> bool:
s = self.step(step_no)
self._open()
if s.kind != "approval":
raise RuleBroken(f"step {s.step_no} is not an approval")
if s.state == DONE:
if s.approved is bool(granted):
return False # the same decision twice: one approval
raise RuleBroken(f"step {s.step_no} was already {'granted' if s.approved else 'refused'} "
f"by {s.done_by}")
if s.state in (SKIPPED, CANCELLED):
raise RuleBroken(f"step {s.step_no} is {s.state}")
if not granted and not (reason or "").strip():
raise Incomplete("refusing needs a reason")
if not self.deps_done(s):
raise RuleBroken(f"step {s.step_no} is not open for a decision yet")
s.state, s.approved, s.done_by, s.done_at, s.message = DONE, bool(granted), by, at, reason or ""
s.next_check_at = None
self.add_log("approved" if granted else "refused", by, at, s.step_no, reason or "")
if not granted:
for d in self.downstream(s.step_no):
if not d.finished:
d.state, d.message, d.next_check_at = CANCELLED, f"approval refused by {by}: {reason}", None
self.add_log("cancelled", by, at, d.step_no, "approval refused")
return True
def downstream(self, step_no: str) -> list:
out, frontier = [], [step_no]
while frontier:
x = frontier.pop()
for d in self.dependants(x):
if d not in out:
out.append(d)
frontier.append(d.step_no)
return out
def cancel(self, reason: str, *, by: str, at: datetime) -> list:
"""HR-23: every not-done step is cancelled; for each DONE action a reversal TASK is created, named after the
capability. Nothing is reversed silently. Returns the reversal steps."""
if not (reason or "").strip():
raise Incomplete("cancelling a run needs a reason")
if self.cancelled_at is not None:
return []
for s in self.steps.values():
if not s.finished:
s.state, s.message, s.next_check_at = CANCELLED, f"run cancelled: {reason}", None
reversals = []
k = 0
for s in self.ordered():
if s.kind != "action" or s.state != DONE or not s.capability:
continue
undo, words = REVERSALS.get(s.capability, (None, None))
if words is None:
continue
k += 1
inp = {"capability": undo, "of": s.capability, "input": _reversal_input(undo, s.input, reason)}
r = RunStep(f"R{k}", 10_000 + k, "task", f"Reverse {s.capability}: {words}", "it",
input=inp, plan=f"{words[0].upper()}{words[1:]} — {_describe(s.input)}",
manual=True, manual_reason=f"reversal of step {s.step_no} after the run was cancelled",
state=WAITING, due_at=at, target=s.target)
self.steps[r.step_no] = r
reversals.append(r)
self.cancelled_at, self.cancel_reason, self.cancelled_by = at, reason, by
self.add_log("cancelled", by, at, None, reason)
return reversals
def retime(self, anchors: dict, calendar, timezone: str) -> list:
"""Retimer: a new due time for every step NOT finished, from its timing rule; finished steps untouched."""
changed = []
for s in self.steps.values():
if s.finished or not s.timing or s.step_no.startswith("R") or ":chase:" in s.step_no:
continue
rule = PlaybookStep(0, s.kind, s.title, s.responsible_group, timing=s.timing).rule(timezone)
due = rule.due_at(anchors, calendar)
if due != s.due_at:
s.due_at = due
if s.state == PENDING:
s.next_check_at = None
changed.append(s)
return changed
def party(self, group: str) -> Optional[int]:
p = self.parties.get(group)
return p[0] if p else None
def _reversal_input(capability: Optional[str], inp: dict, reason: str) -> dict:
if capability == "identity.suspend_user":
return {"primary_email": inp.get("primary_email"), "reason": f"onboarding cancelled: {reason}"}
if capability in ("groups.remove_member",):
return {"group": inp.get("group"), "member_email": inp.get("member_email")}
if capability in ("chat.remove_from_space",):
return {"space": inp.get("space"), "member_email": inp.get("member_email")}
return dict(inp)
def _describe(inp: dict) -> str:
return "; ".join(f"{k}: {v}" for k, v in (inp or {}).items() if v not in (None, "", []))
# ── outcomes: how the result of acting moves a step (HR-36) ──────────────────────────────────────
@dataclass
class Outcome:
"""What acting on a step came to, in hr's words (the capability ACL translates the platform's Result /
ManualTask into this): done | planned | in_progress | failed | manual."""
status: str
retryable: bool = False
provider: str = ""
provider_ref: Optional[str] = None
output: dict = field(default_factory=dict)
message: str = ""
plan: str = ""
manual_input: Optional[dict] = None
manual_reason: str = ""
def apply_outcome(run: Run, s: RunStep, o: Outcome, *, now: datetime, max_attempts: int, backoff: timedelta,
it_group: str = "it", calendar=None, timezone: str = "UTC") -> str:
"""Move one action step by its outcome. Returns what happened, in a word, for the log and the tick's answer."""
if s.finished:
return "already-finished"
if o.status == "done":
s.state, s.done_at, s.done_by = DONE, now, o.provider or "system"
s.provider_ref, s.last_status, s.message = o.provider_ref, "done", o.message
s.result = {**(s.result or {}), "output": o.output or {}}
s.next_check_at = None
run.add_log("action done", o.provider or "system", now, s.step_no, o.provider_ref or "")
run.on_done(s, calendar, timezone)
return "done"
if o.status == "in_progress":
s.state, s.last_status, s.message = PENDING, "in_progress", o.message
s.next_check_at = now + POLL # asked again next tick; NOT a failure
return "in_progress"
if o.status == "manual":
s.state, s.manual, s.manual_reason = WAITING, True, o.manual_reason
s.input = o.manual_input if o.manual_input is not None else s.input
s.plan, s.last_status, s.next_check_at = o.plan, "manual", None
run.add_log("manual task", "system", now, s.step_no, o.manual_reason)
return "manual"
if o.status == "failed" and o.retryable:
s.attempts += 1
s.last_status, s.message = "failed", o.message
run.add_log("attempt failed", o.provider or "system", now, s.step_no, f"attempt {s.attempts}: {o.message}")
if s.attempts >= max_attempts:
s.state, s.manual, s.responsible_group = WAITING, True, it_group
s.manual_reason = f"the provider failed {s.attempts} times: {o.message}"
s.plan = s.plan or _describe(s.input)
s.next_check_at = None
return "manual-after-failures"
s.state = PENDING
s.next_check_at = now + backoff * (2 ** (s.attempts - 1))
return "retry-later"
if o.status == "failed":
s.attempts += 1
s.state, s.manual, s.responsible_group = WAITING, True, it_group
s.last_status, s.message, s.manual_reason = "failed", o.message, f"the provider refused: {o.message}"
s.plan = s.plan or _describe(s.input)
s.next_check_at = None
run.add_log("attempt failed", o.provider or "system", now, s.step_no, o.message)
return "manual-after-failure"
if o.status == "planned":
return "planned"
raise ValueError(f"unknown outcome {o.status!r}")
# ── RunExpansion: the RunFactory ─────────────────────────────────────────────────────────────
@dataclass
class ExpansionContext:
anchors: dict
timezone: str
calendar: Any
values: dict
access_add: list = field(default_factory=list) # [{kind, target, role}]
access_remove: list = field(default_factory=list)
facts: dict = field(default_factory=dict) # worker_type, via_company, rate_outside_band, client_size
is_destructive: Callable[[str], bool] = lambda c: False
timings: dict = field(default_factory=dict) # capability -> the timing a dated setting gives its step
def holds(spec: dict, facts: dict) -> bool:
name, args = spec.get("spec"), spec.get("args") or []
if name == "worker_type_is": # "of": "previous" — the engagement a conversion ended
r = facts.get("from_worker_type" if spec.get("of") == "previous" else "worker_type") in args
elif name == "via_company":
r = bool(facts.get("via_company")) == (bool(args[0]) if args else True)
elif name == "rate_outside_band":
r = bool(facts.get("rate_outside_band"))
elif name == "client_size_is":
r = (facts.get("client_size") or "") in args
elif name == "reason_is": # the event's reason (a Left event's leave reason)
r = (facts.get("reason") or "") in args
else:
raise Incomplete(f"no condition {name!r}")
return (not r) if spec.get("not") else r
def condition_holds(cond: Optional[dict], facts: dict) -> bool:
return all(holds(s, facts) for s in (cond or {}).get("if") or [])
def blocking_requirements(cond: Optional[dict], facts: dict) -> list:
"""The compliance records a step still needs before it may run (pack_requires)."""
have = set(facts.get("compliance") or [])
out = []
for gate in (cond or {}).get("requires") or []:
if all(holds(s, facts) for s in gate.get("if") or []) and gate["record"] not in have:
out.append(gate["record"])
return out
_PLACE = re.compile(r"\{([a-z_][a-z0-9_.]*)\}")
def render(template: Any, values: dict) -> Any:
"""Fill a step's input template from the run's values: "{work_email}" → the address. A string that is
exactly one placeholder takes the value's own type (a list of attendees stays a list)."""
if isinstance(template, dict):
return {k: render(v, values) for k, v in template.items()}
if isinstance(template, list):
return [render(v, values) for v in template]
if isinstance(template, str):
m = _PLACE.fullmatch(template)
if m:
return values.get(m.group(1))
return _PLACE.sub(lambda m: "" if values.get(m.group(1)) is None else str(values.get(m.group(1))), template)
return template
def expand(run_id: str, direction: str, person_id: int, n: int, event_seq, playbook: Playbook,
ctx: ExpansionContext, *, now: datetime) -> Run:
"""RunExpansion: (personnel event, frozen playbook version, person, org) → a Run, every step timed."""
run = Run(run_id, direction, person_id, n, str(event_seq), playbook.playbook_id, playbook.version, now,
anchors=dict(ctx.anchors))
run.anchors.setdefault("run.created", now)
produced: dict = {}
for ps in sorted(playbook.steps, key=lambda s: s.step_no):
targets = [None]
if ps.fan_out:
kind = ps.fan_out.split(":", 1)[1]
pool = ctx.access_remove if ps.capability in REMOVES else ctx.access_add
targets = [r for r in pool if r["kind"] == kind] or []
if ps.capability and ctx.timings.get(ps.capability):
ps = dataclasses.replace(ps, timing=dict(ctx.timings[ps.capability])) # HR-55: a dated setting's timing
rule = ps.rule(ctx.timezone)
due = rule.due_at(run.anchors, ctx.calendar)
skip = not condition_holds(ps.condition, ctx.facts)
made = []
for i, t in enumerate(targets if targets else ([] if ps.fan_out else [None])):
no = f"{ps.step_no}:{t['target']}" if t else str(ps.step_no)
values = {**ctx.values, **({"target": t["target"], "target_role": t.get("role") or "member"} if t else {})}
s = RunStep(no, ps.step_no * 1000 + i, ps.kind, ps.title, ps.responsible_group, ps.step_no,
capability=ps.capability, template=ps.input_template,
input=render(ps.input_template or {}, values), timing=dict(ps.timing or {}), due_at=due,
condition=ps.condition, approval_group=ps.approval_group or (
ps.responsible_group if ps.kind == "approval" else None),
destructive=bool(ps.capability and ctx.is_destructive(ps.capability)),
target=t["target"] if t else None, template_kind=ps.template_kind,
draft_instruction=ps.draft_instruction, outcomes=list(ps.outcomes or []))
if skip:
s.state, s.message = SKIPPED, "its condition does not hold for this person"
made.append(s)
if ps.fan_out and not made:
made = [RunStep(str(ps.step_no), ps.step_no * 1000, ps.kind, ps.title, ps.responsible_group, ps.step_no,
capability=ps.capability, timing=dict(ps.timing or {}), due_at=due, state=SKIPPED,
message="no access rows for this unit")]
produced[ps.step_no] = [m.step_no for m in made]
for m in made:
run.steps[m.step_no] = m
for ps in playbook.steps:
for no in produced[ps.step_no]:
run.steps[no].depends_on = [x for d in ps.depends_on for x in produced.get(d, [])]
return run
# ── the derived views ───────────────────────────────────────────────────────────────────────
def summary(run: Run, now: datetime) -> dict:
"""What the run board shows: done of total, the next due step, waiting on whom, what is stuck."""
steps = [s for s in run.ordered()]
counted = [s for s in steps if s.state != SKIPPED]
done = [s for s in counted if s.state == DONE]
open_ = [s for s in steps if not s.finished]
nxt = sorted((s for s in open_ if s.due_at is not None), key=lambda s: _aware(s.due_at))
stuck = []
for s in open_:
if s.overdue(now):
stuck.append({"step_no": s.step_no, "why": "overdue"})
if s.manual and s.last_status == "failed":
stuck.append({"step_no": s.step_no, "why": "failed"})
if s.kind == "approval" and s.state == WAITING:
stuck.append({"step_no": s.step_no, "why": "waiting for approval"})
if s.kind == "document" and (s.result or {}).get("phase") == "sent":
stuck.append({"step_no": s.step_no, "why": "waiting for signature"})
if s.message.startswith("blocked:"):
stuck.append({"step_no": s.step_no, "why": s.message})
waiting_on = sorted({s.responsible_group for s in open_ if s.state == WAITING or s.overdue(now)})
return {"run_id": run.run_id, "direction": run.direction, "state": run.state, "person_id": run.person_id,
"n": run.n, "playbook": f"{run.playbook_id} v{run.playbook_version}" if run.playbook_id else None,
"steps_done": len(done), "steps_total": len(counted),
"next_due": ({"step_no": nxt[0].step_no, "title": nxt[0].title,
"due_at": _iso(nxt[0].due_at)} if nxt else None),
"waiting_on": waiting_on, "stuck": stuck}
def _iso(t: Optional[datetime]) -> Optional[str]:
if t is None:
return None
return work.aware(t).astimezone(work.UTC).isoformat().replace("+00:00", "Z")
def dumps(v: Any) -> str:
return json.dumps(v, sort_keys=True, default=str)
__all__ = ["Run", "RunStep", "RunLog", "Outcome", "ExpansionContext", "apply_outcome", "expand", "summary", "render",
"condition_holds", "blocking_requirements", "PENDING", "WAITING", "DONE", "SKIPPED", "CANCELLED",
"FINISHED", "REVERSALS", "ADDS", "REMOVES"]