"""The run engine: everything that happens AFTER a personnel event committed, idempotently, in later transactions.
Cross-aggregate consistency is EVENTUAL (modules/hr domain_model.transactions): a personnel event commits on the
Person; then this engine expands the run (Run), syncs the business partner (bp seam), calls capabilities, drafts,
chases, warns and applies retention — each step idempotent on its own key, each write one aggregate per unit of
work, and anything that reads what an earlier transaction wrote re-reads after that commit.
The tick (run_due_steps) calls run_due_steps(now) below. Every part of it may run twice, or concurrently:
* a run is expanded once (its id is derived; the Run save claims version 1);
* a step is acted on by one tick: the tick claims (run_id, step_no, attempt, minute) with a plain INSERT before
it acts, and a concurrent claim fails 23505 and skips; the capability's own key <run_id>|<step_no> makes a
provider act once even across minutes;
* a done step never acts again; a chase is keyed <run_id>|<step_no>|chase|<slot>.
"""
import contextvars
import json
import logging
import os
from datetime import date, datetime, timedelta
from typing import Optional
from erp_platform import work
from erp_platform.repository import as_conflict
from brick_hr.domain import configuration as cfgm
from brick_hr.domain import playbook as pbk
from brick_hr.domain import run as R
from brick_hr.domain import services
from brick_hr.domain.configuration import DRAFTABLE
from brick_hr.domain.errors import RuleBroken
from brick_hr.domain.intake import Intake, intake_id
from brick_hr.domain.person import Document
from brick_hr.domain.privacy import add_months, retention_due
from brick_hr.domain.sensitive import CATEGORY_OF
from brick_hr.domain.values import DAY, run_id as make_run_id
from brick_hr.infrastructure import business_partner as bp
from brick_hr.infrastructure import capability_acl as acl
from brick_hr.infrastructure import contracts_acl as contracts
from brick_hr.infrastructure import models_adapter
from brick_hr.infrastructure.repositories import is_duplicate
log = logging.getLogger("brick_hr.engine")
SYSTEM = {"sub": "hr-system", "email": "hr-system", "roles": ["system"], "auth_source": "system",
"permissions": ["hr:person:write", "hr:sensitive:read", "hr:person:read", "hr:run:read"]}
ACTOR = "hr-system"
def _uc():
from brick_hr.application import use_cases
return use_cases
def _view(uow, at=None):
from brick_hr.application.support import View
return View(uow, SYSTEM, at)
def _committed(fn):
"""Run fn() in a unit of work that is committed here; a lost optimistic version (someone else saved this
aggregate first, at the statement or at COMMIT) is answered None — the caller tries again on the next tick."""
try:
return fn()
except Exception as exc: # noqa: BLE001 — anything but a lost version is re-raised
if as_conflict(exc) is not None:
return None
raise
# ── expansion ──────────────────────────────────────────────────────────────────────────────
def _units_of(view, p, n, d) -> list:
a = p.assignments[n].at(d) or p.assignments[n].first()
if a is None:
return []
org = view.orgs.get(a.data.get("org_id"))
return [u.org_unit_id for u in org.ancestors(a.data.get("org_unit_id"), d)] if org else [a.data.get("org_unit_id")]
def _access(view, p, n, direction, effective: date) -> tuple:
cfg = view.config
t = p.assignments[n]
if direction == "offboarding":
last = p.engagements[n].actual_end or effective
a = t.at(last) or t.last()
return [], [dict(r) for r in cfg.access(a.data.get("org_unit_id"))] if a else []
if direction == "change":
new = t.at(effective)
old = t.at(effective - DAY)
new_rows = cfg.access(new.data.get("org_unit_id")) if new else []
old_rows = cfg.access(old.data.get("org_unit_id")) if old else []
key = lambda r: (r["kind"], r["target"]) # noqa: E731
nk, ok = {key(r) for r in new_rows}, {key(r) for r in old_rows}
return [dict(r) for r in new_rows if key(r) not in ok], [dict(r) for r in old_rows if key(r) not in nk]
a = t.at(effective) or t.first()
return [dict(r) for r in cfg.access(a.data.get("org_unit_id"))] if a else [], []
def context_for(view, p, ev_kind: str, direction: str, n: int, effective: date, payload: dict,
parties: Optional[dict] = None, reason: Optional[str] = None) -> R.ExpansionContext:
from brick_hr.application.support import facts_for, rate_outside_band, values_for
e = p.engagements[n]
anchors = {"start_date": e.start_date, "event_date": effective, "run.created": view.now}
if e.actual_end or e.planned_end:
anchors["last_day"] = e.actual_end or e.planned_end
a = p.assignments[n].at(min(max(view.today, e.start_date), e.end)) or p.assignments[n].first()
cal = view.config.calendar(a.data.get("location_code") if a else None, a.data.get("schedule_code") if a else None)
values = values_for(view, p, n, anchors=anchors, parties=parties)
values["first_project"] = (payload or {}).get("first_project") or ""
add, remove = _access(view, p, n, direction, effective)
facts = {**facts_for(view, p, n, rate_outside_band(view, p, n)), "reason": reason}
timings = {}
if direction == "offboarding":
# HR-55: when the ERP sign-in goes — the dated setting in force on the business date the leaver is recorded,
# for this leave reason or the default. A later change to the setting does not move a run already made.
policy = view.config.dated_setting(cfgm.REVOKE_ACCESS_AT, view.today, reason or "")
t = cfgm.revoke_access_timing(policy)
if t:
timings[cfgm.REVOKES_ACCESS] = t
return R.ExpansionContext(anchors, view.tz, cal, values, add, remove, facts, acl.is_destructive, timings)
def build_run(view, p, ev, *, at: datetime) -> R.Run:
"""RunFactory: the frozen playbook version expanded for this person and event — every step timed, and every
action with no provider marked manual from the start, with the exact input it would send."""
direction = ev.direction
e = p.engagements[ev.n]
ctx = context_for(view, p, ev.kind, direction, ev.n, ev.effective_date, ev.payload, reason=ev.reason)
units = _units_of(view, p, ev.n, ev.effective_date)
pb = pbk.select(view.uow.playbooks.latest(), ev.event_type, e.worker_type, ev.reason, units)
if pb is None:
run = R.Run(ev.run_id, direction, p.person_id, ev.n, str(ev.seq), None, None, at, anchors=dict(ctx.anchors))
run.steps["0"] = R.RunStep("0", 0, "task", f"No playbook matches a {ev.event_type} of a {e.worker_type}: "
"choose or add one in Playbooks, and do this run's steps by hand", "hr",
state=R.WAITING, due_at=at)
return run
run = R.expand(ev.run_id, direction, p.person_id, ev.n, ev.seq, pb, ctx, now=at)
mark_manual(run)
run.add_log("created", ACTOR, at, None, f"{pb.playbook_id} v{pb.version}")
return run
def mark_manual(run: R.Run) -> None:
for s in run.steps.values():
if s.kind == "action" and s.capability and not s.finished:
if acl.provider(s.capability) is None:
s.manual = True
s.manual_reason = f"no provider for {s.capability} in this composition"
s.plan = acl.describe(s.capability, s.input)
else:
s.manual, s.manual_reason = False, ""
s.plan = acl.describe(s.capability, s.input)
def expand(person_id: int, n: int, seq: int, at: datetime, person=None) -> Optional[str]:
"""`person`: the Person as its own unit of work just committed it (record_hire) — used instead of reading
the same rows back; anything else reads it."""
uc = _uc()
def go():
with uc.repository() as uow:
p = person if person is not None and person.person_id == int(person_id) else uow.persons.get(person_id)
ev = p.event(n, seq) if p else None
if ev is None or not ev.run_id or uow.runs.exists(ev.run_id):
return None
view = _view(uow, at)
run = build_run(view, p, ev, at=at)
uow.runs.save(run)
return run.run_id
return _committed(go)
def expand_pending(at: datetime, known: Optional[tuple] = None) -> list:
"""`known` = (events_pending, run ids) already read in this tick (run_due_steps reads them with the due list,
in one round trip)."""
uc = _uc()
if known is None:
with uc.repository() as uow:
pending = uow.persons.events_pending()
existing = set(uow.runs.ids())
else:
pending, existing = known[0], set(known[1])
out = []
for e in sorted(pending, key=lambda e: (e["person_id"], e["n"], e["seq"])):
if e.get("run_id") and e["run_id"] not in existing:
rid = expand(int(e["person_id"]), int(e["n"]), int(e["seq"]), at)
if rid:
out.append(rid)
return out
# ── the business partner (HR-38, HR-50) ────────────────────────────────────────────────────
def sync_partner(person_id: int, at: datetime) -> str:
"""Make the person's partner what hr says today (name, work contact, roles, works_for), then mark every event
that has taken effect as synced. Keyed by the events: a second tick changes nothing."""
uc = _uc()
if not bp.installed():
return "absent" # no business-partner brick in this composition: nothing to load or keep
with uc.repository() as uow:
view = _view(uow, at)
p = uow.persons.get(person_id)
if p is None:
return "gone"
loc_country = ""
a = p.assignment_at(view.today) or ((p.relevant_engagement(view.today).n,
p.assignments[p.relevant_engagement(view.today).n].first())
if p.engagements else None)
if a and a[1] is not None:
loc = view.config.location(a[1].data.get("location_code")) or {}
loc_country = loc.get("country") or ""
card = services.person_card(p, view.today, view.config, loc_country)
companies = [ev.payload.get("company") for ev in p.events if ev.payload.get("company")]
if not bp.installed():
return "absent"
for company in companies:
bp.promote_company(company, ACTOR)
r = bp.sync(card, ACTOR)
if not r.ok:
log.warning(json.dumps({"event": "bp_sync_failed", "person_id": person_id, "status": r.status}))
return r.status
def mark():
with uc.repository() as uow:
p2 = uow.persons.get(person_id)
view2 = _view(uow, at)
changed = False
if p2.bp_id != r.bp_id:
p2.bp_id, changed = r.bp_id, True
for ev in p2.events:
if ev.bp_synced_at is None and ev.takes_effect_on <= view2.today:
ev.bp_synced_at, changed = at, True
if changed:
uow.persons.save(p2)
return "synced"
return _committed(mark) or "retry"
def sync_due(at: datetime) -> list:
"""HR-38: events whose effective date has arrived and whose partner sync has not happened."""
uc = _uc()
if not bp.installed():
return [] # no business-partner brick in this composition: nothing to keep right
with uc.repository() as uow:
view = _view(uow, at)
rows = uow.persons.events_pending()
due = set()
for e in rows:
if e.get("bp_synced_at") is not None:
continue
eff = e["effective_date"]
eff = eff if isinstance(eff, date) else date.fromisoformat(str(eff)[:10])
takes = eff + DAY if e["kind"] == "Left" else eff
if takes <= view.today:
due.add(int(e["person_id"]))
return [(pid, sync_partner(pid, at)) for pid in sorted(due)]
# ── acting on steps ────────────────────────────────────────────────────────────────────────
def claim(run_id: str, step_no: str, attempt: int, at: datetime) -> bool:
"""Plain INSERT of (run_id, step_no, attempt, minute) in its own transaction: one tick wins."""
uc = _uc()
slot = f"{attempt}@{work.aware(at):%Y%m%dT%H%M}"
try:
with uc.repository() as uow:
uow.claims.claim(run_id, f"{step_no}#{slot}", attempt)
return True
except Exception as exc: # noqa: BLE001
if is_duplicate(exc):
return False
raise
def _cadence(view, kind: str) -> work.ChaseCadence:
c = view.config
key = {"intake": "intake_chase", "approval": "approval_chase"}.get(kind, "task_chase")
return work.ChaseCadence(c.int_setting(f"{key}_every_days"), c.int_setting(f"{key}_max"), timezone=view.tz,
unit=c.setting("chase_unit"))
def _chases(run: R.Run, step_no: str) -> list:
return [l.event_at for l in run.log if l.what == "chase" and l.from_value == step_no]
def _calendar(view, p, n):
a = p.assignments[n].first() if n in p.assignments else None
return view.config.calendar(a.data.get("location_code") if a else None, a.data.get("schedule_code") if a else None)
def arm_chase(view, run: R.Run, s: R.RunStep, p) -> None:
"""When the first (next) reminder for a waiting step falls due."""
cad = _cadence(view, "intake" if s.kind == "intake" else ("approval" if s.kind == "approval" else "task"))
base = s.due_at or view.now
st = work.next_chase(cad, base, _chases(run, s.step_no), _calendar(view, p, run.n) if cad.unit != "days" else None)
s.next_check_at = st.next_at
def plan_actions(view, run: R.Run, p, at: datetime) -> list:
"""What the tick should do to this run now: (step_no, what). Pure reading."""
out = []
for s in run.due(at):
out.append((s.step_no, "due"))
for s in run.ordered():
if s.state == R.WAITING and s.next_check_at is not None and work.aware(s.next_check_at) <= work.aware(at):
out.append((s.step_no, "check"))
return out
def _drafts(run: R.Run) -> dict:
return {s.draft_instruction: s.draft_text for s in run.steps.values()
if s.kind == "draft" and s.state == R.DONE and s.draft_instruction and s.draft_text}
def _values(view, run, p):
from brick_hr.application.support import values_for
ev = p.event(run.n, int(run.event_seq)) if str(run.event_seq).isdigit() else None
v = values_for(view, p, run.n, anchors=run.anchors, parties={g: x for g, x in run.parties.items()},
drafts=_drafts(run))
v["first_project"] = (ev.payload.get("first_project") if ev else "") or ""
return v
def _draft_fields(view, run, p, instruction: dict, values: dict) -> dict:
"""ONLY the instruction's allowed fields, from the non-sensitive values (the draft is never given sensitive data)."""
fields = {}
for f in instruction.get("fields") or []:
if f not in DRAFTABLE:
continue
if f in ("open_steps", "line", "units_led", "unit_access"):
fields[f] = _leaver_facts(view, run, p).get(f)
elif f == "legal_given_name":
fields[f] = values.get("given_name")
elif f == "reason_label":
ev = p.event(run.n, int(run.event_seq)) if str(run.event_seq).isdigit() else None
r = view.config.reason(ev.event_type, ev.reason) if ev and ev.reason else None
fields[f] = (r or {}).get("label") or (ev.reason if ev else "")
else:
fields[f] = values.get(f)
return fields
def _leaver_facts(view, run, p) -> dict:
"""What hr knows a leaver looks after (HR-06): their open steps, the people in their line, units they lead,
their unit's groups and spaces. Names only."""
me = p.person_id
d = min(view.today, p.engagements[run.n].end)
people = {s.person_id for s in view.directory.seats}
line = services.line_of(me, d, view.directory, view.orgs, people)
names = view.uow.persons.names(line, view.today)
held = view.directory.positions_held(me, d)
units = []
for org in view.orgs.values():
units += [u.name for u in org.units_on(d) if u.chief_position_id in held]
a = p.assignments[run.n].at(d)
access = [f"{r['kind']} {r['target']}" for r in view.config.access(a.data.get("org_unit_id"))] if a else []
open_steps = []
for r in view.uow.runs.load_many([rid for rid, _pid, _g, parties in view.open_steps
if me in (parties or {}).values()]):
open_steps += [f"{s.title} ({r.run_id})" for s in r.ordered() if not s.finished
and r.parties.get(s.responsible_group, (None,))[0] == me]
return {"open_steps": sorted(set(open_steps)), "line": sorted(v["name"] for v in names.values()),
"units_led": sorted(units), "unit_access": access}
def act(view, run: R.Run, s: R.RunStep, p, what: str, at: datetime) -> Optional[tuple]:
"""Decide and DO one step's work (outside calls happen here, after the claim). Returns (kind, payload) for
apply(), or None when nothing is to be recorded."""
if s.finished:
return None
facts = None
if what == "check":
if s.kind == "document" and (s.result or {}).get("phase") == "sent":
return _contract_poll(run, s, p)
if s.kind in ("task", "approval", "intake") or s.manual:
return ("chase", None)
return None
# what == "due"
if s.kind in ("task", "approval"):
return ("open", None)
if s.kind == "intake":
return _intake(view, run, s, p, at)
from brick_hr.application.support import facts_for
facts = facts_for(view, p, run.n)
gates = R.blocking_requirements(s.condition, facts)
if gates:
return ("blocked", f"blocked: record the {gates[0].replace('_', ' ')} first")
if s.kind == "draft":
return _draft(view, run, s, p, at)
if s.kind == "document":
return _contract_start(view, run, s, p, at)
if s.kind != "action":
return None
if s.destructive and not run.approval_granted(s):
return ("blocked", "blocked: waiting for its approval — a destructive step never fires without one")
if s.capability == "identity.delete_user":
waiting = _transfer_pending(run, s, at)
if waiting is not None:
return waiting
inp = render_input(view, run, s, p)
if not claim(run.run_id, s.step_no, s.attempts + 1, at):
return None
if s.manual and acl.provider(s.capability) is None:
o = acl.call(s.capability, run.key(s), inp, actor=ACTOR) # validated, then the ManualTask
else:
o = acl.call(s.capability, run.key(s), inp, actor=ACTOR)
return ("outcome", (o, inp))
def render_input(view, run, s, p) -> dict:
"""A step's input as it is NOW (a work email made since expansion is in it), for its own target."""
values = _values(view, run, p)
if s.target:
values.update(target=s.target, target_role=(s.input or {}).get("role") or "member")
return R.render(s.template or {}, values)
def _transfer_pending(run, s, at) -> Optional[tuple]:
"""Before identity.delete_user: the Drive transfer it depends on must have COMPLETED (asked on each tick)."""
for a in run.ordered():
if a.capability != "drive.transfer_ownership" or a.step_no not in _ancestors(run, s):
continue
if a.state != R.DONE:
return ("blocked", "blocked: the Drive transfer is not done yet")
if a.manual or (a.result or {}).get("output", {}).get("state") == "completed":
return None
ref = a.provider_ref or (a.result or {}).get("output", {}).get("transfer_ref")
if not ref:
return None
o = acl.call("drive.transfer_status", f"{run.run_id}|{s.step_no}|transfer", {"transfer_ref": ref}, actor=ACTOR)
if o.status == "manual":
return ("outcome", (R.Outcome("manual", manual_input={"transfer_ref": ref},
plan="Confirm the Drive transfer to the manager completed, then delete "
"the account and free the licence",
manual_reason="no provider can confirm the transfer"), {}))
state = (o.output or {}).get("state")
if o.status == "done" and state == "completed":
return None
if o.status == "done" and state == "failed":
return ("outcome", (R.Outcome("failed", retryable=False, message="the Drive transfer failed"), {}))
return ("outcome", (R.Outcome("in_progress", message=f"the Drive transfer is {state or 'still running'}"), {}))
return None
def _ancestors(run, s) -> set:
out, frontier = set(), list(s.depends_on)
while frontier:
x = frontier.pop()
if x in out or x not in run.steps:
continue
out.add(x)
frontier += run.steps[x].depends_on
return out
def _draft(view, run, s, p, at) -> Optional[tuple]:
if s.draft_text:
return None # the first draft is kept
if not claim(run.run_id, s.step_no, s.attempts + 1, at):
return None
instruction = view.config.draft_instruction(s.draft_instruction) or {}
values = _values(view, run, p)
fields = _draft_fields(view, run, p, instruction, values)
drafter = models_adapter.DRAFTER
try:
if drafter is None or not instruction:
raise models_adapter.DraftFailed("no drafter or no draft instruction")
text = drafter.draft(instruction.get("instructions") or "", fields, view.config.setting("ai_model"))
return ("drafted", text)
except models_adapter.DraftFailed as e:
return ("draft_failed", str(e))
except Exception as e: # noqa: BLE001 — any failure is a manual task
return ("draft_failed", type(e).__name__)
def _intake(view, run, s, p, at) -> Optional[tuple]:
"""HR-02/53: the starter's link. With a mail provider it is issued and mailed; without, HR copies a fresh one
(reissue_intake_link answers the url ONCE — it is never stored or logged)."""
e = p.engagements[run.n]
iid = intake_id(p.person_id, run.n)
ensure_intake(p.person_id, run.n, e.worker_type, view.config.setting("pack"), at)
if not _personal(p, view):
return ("intake_manual", "The starter has no personal email: add it to their detail, then use 'Copy a fresh "
"link' (reissue_intake_link) and give it to them")
if acl.provider("mail.send") is None:
return ("intake_manual", f"Send {_personal(p, view)} their details link: open the run and use "
"'Copy a fresh link' (reissue_intake_link) — the link is shown once and never stored")
if not claim(run.run_id, s.step_no, s.attempts + 1, at):
return None
link = issue_link(iid, _personal(p, view), view.config)
body = {"from_email": view.config.setting("hr_mailbox") or _values(view, run, p).get("manager_email") or "",
"to": [_personal(p, view)], "cc": [], "subject": "Your details for your first day",
"body_text": f"Please fill in your details here: {link.url}\nThe link works for "
f"{view.config.int_setting('intake_validity_days')} days.", "body_html": None}
o = acl.call("mail.send", run.key(s), body, actor=ACTOR)
return ("intake_sent", o)
def _personal(p, view) -> str:
d = p.detail_at(view.today)
return (d.data.get("personal_email") or "") if d else ""
def ensure_intake(person_id: int, n: int, worker_type: str, pack: str, at: datetime) -> None:
uc = _uc()
def go():
with uc.repository() as uow:
iid = intake_id(person_id, n)
if uow.intakes.get(iid) is None:
uow.intakes.save(Intake(iid, person_id, n, worker_type, pack))
_committed(go)
def issue_link(iid: str, recipient: str, config):
uc = _uc()
days = config.int_setting("intake_validity_days") or 7
link = uc.HR.links.issue("intake", iid, recipient, "SUBMIT", expires_in=timedelta(days=days), actor=ACTOR)
def go():
with uc.repository() as uow:
i = uow.intakes.get(iid)
if i is not None:
now = datetime.now(work.UTC)
i.issue(link.link_id, now, now + timedelta(days=days))
uow.intakes.save(i)
_committed(go)
return link
def _contract_counterparty(p, n) -> int:
e = p.engagements[n]
return int(e.supplier_bp_id) if e.supplier_bp_id else (p.bp_id or bp.person_bp_id(p.person_id))
# What contracts.status can answer, in hr's terms. Signed: the executed copy is fetched and filed (SIGNED_ELSEWHERE:
# signed on paper and uploaded in contracts — the copy is as good). Over without a signature: a manual task with the
# reason, never a silent failure. Waiting: asked again on the next tick.
SIGNED_STATES = ("SIGNED", "SIGNED_ELSEWHERE", "EXECUTED", "ACTIVE")
UNSIGNED_END = {"DECLINED": "declined", "RECALLED": "recalled", "EXPIRED_UNSIGNED": "left unsigned until it expired",
"EXPIRED": "expired", "CANCELLED": "cancelled"}
def _contract_ended(s, cid: str, state: str) -> tuple:
words = UNSIGNED_END[state]
return ("outcome", (R.Outcome(
"manual", manual_input={"contract_id": cid},
manual_reason=f"the contract was {words} in contracts ({state})",
plan=f"agree the next step with the starter — draft it again in contracts, or have it signed by hand"), {}))
def _missing_from(o) -> list:
"""contracts.draft refuses 422 'missing: a, b — nothing was drafted' (nothing stored): the field names."""
text = o.message or ""
if o.status != "failed" or "missing:" not in text:
return []
names = text.split("missing:", 1)[1].split("—", 1)[0]
return [n.strip() for n in names.split(",") if n.strip()]
def _fill_task(run, s) -> Optional[R.RunStep]:
tasks = [x for x in run.steps.values() if x.step_no.startswith(f"{s.step_no}:fill")]
return max(tasks, key=lambda x: x.sort_no) if tasks else None
def _contract_start(view, run, s, p, at) -> Optional[tuple]:
res = s.result or {}
if res.get("phase") == "drafted":
return _contract_resume(view, run, s, p, at)
if res.get("phase"):
return None
data = contract_input(view, run, s, p)
missing = res.get("missing") or []
if missing:
# asked again only when it can answer otherwise: every field now has a value, or HR says it is filled in
task = _fill_task(run, s)
if not (all(data["variables"].get(f) not in (None, "") for f in missing) or (task and task.state == R.DONE)):
return ("blocked", f"blocked: fill in {', '.join(missing)} for the contract (task {task.step_no if task else ''})")
if not claim(run.run_id, s.step_no, s.attempts + 1, at):
return None
o = contracts.draft(data, actor=ACTOR) # the same key every time: <source_ref>|<type_code>
fields = _missing_from(o)
if fields:
return ("missing", fields)
if o.status != "done":
return ("outcome", (o, data))
cid = (o.output or {}).get("contract_id") or o.provider_ref
if (o.output or {}).get("missing"):
return ("outcome", (R.Outcome("failed", retryable=False,
message=f"the contract needs {', '.join(o.output['missing'])}"), data))
if (o.output or {}).get("approval_required"):
return ("contract", {"phase": "drafted", "contract_id": cid, "note": "waiting for an approval in contracts"})
return _contract_send(run, s, cid)
def _contract_send(run, s, cid: str) -> tuple:
sent = contracts.send(cid, f"{run.run_id}|{s.step_no}|send", actor=ACTOR)
if sent.status == "done":
return ("contract", {"phase": "sent", "contract_id": cid})
if sent.status == "failed" and not sent.retryable:
# drafted, but contracts will not send it as it stands (e.g. nobody named to sign): a person, with why
return ("unsent", (cid, R.Outcome("manual", manual_input={"contract_id": cid},
manual_reason=f"contracts would not send it: {sent.message}",
plan="fix it in contracts and send it from there")))
return ("contract", {"phase": "drafted", "contract_id": cid, "note": sent.message})
def _contract_resume(view, run, s, p, at) -> Optional[tuple]:
"""A drafted contract that was not sent (it needed an approval, or the send did not go through): ask where it
stands; once nothing is waiting for an approval, send it (the same key as the first send), then poll."""
cid = (s.result or {}).get("contract_id")
if not cid:
return None
st = contracts.status(cid, actor=ACTOR)
if st.status != "done":
return ("contract", {"phase": "drafted", "contract_id": cid, "note": st.message})
state = str((st.output or {}).get("state") or "").upper()
if state == "IN_APPROVAL":
return ("contract", {"phase": "drafted", "contract_id": cid, "note": "waiting for an approval in contracts"})
if state in SIGNED_STATES or state in UNSIGNED_END or state == "OUT_FOR_SIGNATURE":
return _contract_poll(run, s, p, known=st) # moved on in contracts meanwhile (sent or signed there)
if not claim(run.run_id, s.step_no, s.attempts + 1, at):
return None
return _contract_send(run, s, cid)
def contract_input(view, run, s, p) -> dict:
values = _values(view, run, p)
e = p.engagements[run.n]
pay = p.pay[run.n].first()
variables = {k: values.get(k) for k in ("legal_name", "preferred_name", "job_title", "org_unit_name",
"location_name", "start_date", "last_day", "worker_type_label")}
variables["client_size_regime"] = view.config.client_size(e.legal_entity_ref) or None
if pay is not None:
variables.update(rate_amount=str(pay.data.get("rate_amount")), rate_basis=pay.data.get("rate_basis"),
currency=pay.data.get("currency"))
signers = None
# who signs for the company: the run's own director party, else the signer the settings name (0.4.0)
director = run.party("director") or int(view.config.setting("contract_signer_person_id") or 0) or None
if director:
dn = view.uow.persons.names([director], view.today).get(director, {})
signers = [{"side": "US", "full_name": dn.get("name") or "", "email": dn.get("work_email") or "",
"title": "Director", "authority_basis": None, "signing_order": 1},
{"side": "THEM", "full_name": values.get("legal_name") or "", "email": values.get("personal_email") or "",
"title": None, "authority_basis": None, "signing_order": 2}]
return contracts.draft_input(s.template_kind or "", _contract_counterparty(p, run.n), f"{p.person_id}|{run.n}",
contracts.source_ref(run.run_id, s.step_no),
{k: v for k, v in variables.items() if v is not None and v != ""}, signers)
def _contract_poll(run, s, p, known=None) -> Optional[tuple]:
cid = (s.result or {}).get("contract_id")
if not cid:
return None
st = known or contracts.status(cid, actor=ACTOR)
if st.status != "done":
return ("contract", {"phase": "sent", "contract_id": cid, "note": st.message})
state = str((st.output or {}).get("state") or "").upper()
if state in SIGNED_STATES:
doc = contracts.fetch_executed(cid, actor=ACTOR)
if doc.status != "done":
return ("contract", {"phase": "sent", "contract_id": cid, "note": "signed; the copy is not ready yet"})
return ("signed", {"contract_id": cid, "file": doc.output})
if state in UNSIGNED_END:
return _contract_ended(s, cid, state)
return ("contract", {"phase": "sent", "contract_id": cid})
def apply(view, run: R.Run, s: R.RunStep, p, result: tuple, at: datetime) -> list:
"""Record what acting came to on the (fresh) run. Returns follow-ups to run after the run is saved."""
kind, payload = result
follow = []
cfg = view.config
cal = _calendar(view, p, run.n)
if kind == "open":
if s.state == R.PENDING:
s.state = R.WAITING
arm_chase(view, run, s, p)
elif kind == "blocked":
s.message, s.next_check_at = payload, at + R.POLL
elif kind == "chase":
follow += chase(view, run, s, p, at)
elif kind == "drafted":
s.draft_text, s.state, s.last_status, s.next_check_at = payload, R.WAITING, "drafted", None
s.message = "drafted: waiting for approval"
run.add_log("drafted", ACTOR, at, s.step_no, cfg.setting("ai_model"))
elif kind == "draft_failed":
s.state, s.manual, s.last_status = R.WAITING, True, "manual"
s.manual_reason = f"write it by hand — the draft could not be made ({payload})"
s.plan = f"Write the {s.title.lower()} by hand, then approve it"
run.add_log("manual task", ACTOR, at, s.step_no, s.manual_reason)
elif kind == "intake_manual":
s.state, s.manual, s.plan, s.last_status = R.WAITING, True, payload, "manual"
s.manual_reason = "no provider for mail.send in this composition"
arm_chase(view, run, s, p)
elif kind == "intake_sent":
o = payload
if o.status == "done":
s.state, s.last_status, s.provider_ref, s.message = R.WAITING, "sent", o.provider_ref, "link sent"
arm_chase(view, run, s, p)
else:
R.apply_outcome(run, s, o, now=at, max_attempts=cfg.int_setting("max_action_attempts"),
backoff=timedelta(minutes=cfg.int_setting("action_backoff_minutes")), calendar=cal,
timezone=view.tz)
elif kind == "missing":
# contracts stored nothing: a task for HR to fill the fields in, and the draft asked again (same key)
fields = list(payload)
s.result = {**(s.result or {}), "missing": fields}
s.message, s.last_status, s.next_check_at = f"blocked: fill in {', '.join(fields)} for the contract", \
"missing", at + R.POLL
task = _fill_task(run, s)
if task is None or task.state == R.DONE:
k = 1 + len([x for x in run.steps if x.startswith(f"{s.step_no}:fill")])
no = f"{s.step_no}:fill" if k == 1 else f"{s.step_no}:fill:{k}"
run.steps[no] = R.RunStep(no, s.sort_no + k, "task", f"Fill in {', '.join(fields)} for the "
f"{(s.template_kind or 'contract').replace('_', ' ')}", "hr", state=R.WAITING,
due_at=at, message="contracts needs them to draft it; then mark this done",
plan=f"Set {', '.join(fields)} where hr keeps them (the person, the job, the "
"location), then mark this task done: the contract is drafted again")
run.add_log("contract needs fields", ACTOR, at, s.step_no, ", ".join(fields))
elif kind == "contract":
s.result = {k: v for k, v in {**(s.result or {}), **payload}.items() if k != "missing"}
s.contract_id = payload.get("contract_id")
s.state = R.WAITING if payload.get("phase") == "sent" else R.PENDING
s.next_check_at = at + R.POLL
s.message = {"sent": "out for signature", "drafted": "drafted"}.get(payload.get("phase"), "")
if payload.get("note") and payload.get("phase") == "drafted":
s.message = f"drafted: {payload['note']}" # e.g. waiting for an approval in contracts
elif kind == "signed":
follow.append(("file_contract", {"step_no": s.step_no, **payload}))
elif kind == "unsent":
cid, o = payload
s.result = {**(s.result or {}), "phase": "drafted", "contract_id": cid}
s.contract_id = cid
follow += apply(view, run, s, p, ("outcome", (o, {})), at)
elif kind == "outcome":
o, inp = payload
if inp:
s.input = inp
if s.kind == "document" and o.status == "manual":
o = R.Outcome("manual", manual_input=o.manual_input, manual_reason=o.manual_reason,
plan=f"Send the {s.template_kind.replace('_', ' ')} for signature and attach the signed "
f"copy when it is back — {o.plan}")
happened = R.apply_outcome(run, s, o, now=at, max_attempts=cfg.int_setting("max_action_attempts"),
backoff=timedelta(minutes=cfg.int_setting("action_backoff_minutes")), calendar=cal,
timezone=view.tz)
if s.manual and s.state == R.WAITING:
arm_chase(view, run, s, p)
if happened == "done":
follow += after_done(s)
return follow
def after_done(s: R.RunStep) -> list:
"""What a done action means for the Person (written after the run is saved, in its own transaction)."""
out = (s.result or {}).get("output") or {}
if s.capability == "identity.provision_user" and (out.get("primary_email") or s.input.get("primary_email")):
return [("work_email", out.get("primary_email") or s.input.get("primary_email"))]
if s.capability == "identity.invite_app_user" and out.get("sub"):
return [("pool_user_sub", out["sub"])]
return []
def chase(view, run, s, p, at) -> list:
"""HR-37: a reminder at most once per cadence slot, up to the maximum; then one escalation to HR."""
kind = "intake" if s.kind == "intake" else ("approval" if s.kind == "approval" else "task")
cad = _cadence(view, kind)
sent = _chases(run, s.step_no)
if len(sent) >= cad.max_count:
esc = f"{s.step_no}:escalate"
if esc not in run.steps:
words = {"approval": "approval overdue", "intake": "starter details still missing"}.get(kind, "task overdue")
run.steps[esc] = R.RunStep(esc, s.sort_no + 999, "task", f"{words}: {s.title}", "hr", state=R.WAITING,
due_at=at, message=f"chased {len(sent)} times")
run.add_log("escalated", ACTOR, at, s.step_no, words)
s.next_check_at = None
return []
slot = len(sent) + 1
to = _recipients(view, run, s, p)
how = "reminder on My tasks"
if to and acl.provider("mail.send") is not None:
body = {"from_email": view.config.setting("hr_mailbox") or "", "to": to, "cc": [],
"subject": f"Reminder: {s.title}",
"body_text": f"A reminder: '{s.title}' is waiting for you.", "body_html": None}
o = acl.call("mail.send", f"{run.run_id}|{s.step_no}|chase|{slot}", body, actor=ACTOR)
how = "reminder sent" if o.status == "done" else f"reminder not sent: {o.message or o.status}"
elif kind == "intake":
no = f"{s.step_no}:chase:{slot}"
run.steps[no] = R.RunStep(no, s.sort_no + slot, "task", f"Remind {', '.join(to) or 'the starter'} to finish "
"their details", "hr", state=R.WAITING, due_at=at, manual=True,
manual_reason="no provider for mail.send", plan=f"Send a reminder to "
f"{', '.join(to) or 'the starter'}: their details are not complete yet")
how = "reminder task for HR"
run.add_log("chase", ACTOR, at, s.step_no, f"slot {slot}: {how}")
st = work.next_chase(cad, s.due_at or at, sent + [at], _calendar(view, p, run.n) if cad.unit != "days" else None)
s.next_check_at = st.next_at
if st.exhausted:
s.next_check_at = at
return []
def _recipients(view, run, s, p) -> list:
if s.kind == "intake":
e = _personal(p, view)
return [e] if e else []
row = view.config.group(s.approval_group if s.kind == "approval" else s.responsible_group) or {}
res = row.get("resolution") or ""
pid = None
if res == "manager":
pid = view.relevant_manager(p.person_id)
elif res == "run_party":
pid = run.party(s.responsible_group)
elif res == "person":
pid = p.person_id
if pid:
e = view.uow.persons.names([pid], view.today).get(pid, {}).get("work_email")
return [e] if e else []
return []
def advance(run_id: str, at: datetime) -> list:
"""Move one run on: read, act (claims and outside calls), then record on a fresh copy and save."""
uc = _uc()
with uc.repository() as uow:
pid = _person_of(run_id)
uow.prefetch(run_id=run_id, person=int(pid) if str(pid).isdigit() else None) # one round trip
run = uow.runs.get(run_id)
if run is None or run.cancelled_at is not None:
return []
p = uow.persons.get(run.person_id)
if p is None:
return []
view = _view(uow, at)
todo = plan_actions(view, run, p, at)
results = []
for step_no, what in todo:
r = act(view, run, run.steps[step_no], p, what, at)
if r is not None:
results.append((step_no, run.steps[step_no].state, r))
if not results:
return []
follow: list = []
done: list = []
def record():
with uc.repository() as uow:
fresh = uow.runs.get(run_id)
# the run is read again (another tick or a person may have moved it); the person is only READ by
# apply (calendars, names, recipients) and was read by this same advance a moment ago
p2 = p if p.person_id == fresh.person_id else uow.persons.get(fresh.person_id)
view2 = _view(uow, at)
follow.clear()
done.clear()
for step_no, state_seen, r in results:
s = fresh.steps.get(step_no)
if s is None or s.state != state_seen:
continue # someone moved it meanwhile: their move stands
follow.extend((step_no, f) for f in apply(view2, fresh, s, p2, r, at))
done.append((step_no, r[0]))
uow.runs.save(fresh)
return True
# Another tick (or a person) saved this run meanwhile: record on the newer copy — the outside calls were made,
# and a lost record would leave a done action looking pending until its claim slot passed.
from erp_platform.repository import retry_on_conflict
ok = _committed(lambda: retry_on_conflict(record))
if not ok:
return []
for step_no, f in follow:
follow_up(run_id, step_no, f, at)
return [f"{run_id}|{n}:{k}" for n, k in done]
def follow_up(run_id: str, step_no: str, f: tuple, at: datetime) -> None:
kind, value = f
uc = _uc()
with uc.repository() as uow:
run = uow.runs.get(run_id)
pid = run.person_id
if kind == "work_email":
def go():
with uc.repository() as uow:
p = uow.persons.get(pid)
view = _view(uow, at)
if p.write_back_work_email(value, view.today, now=at, by=ACTOR):
uow.persons.save(p)
return True
return False
if _committed(go):
sync_partner(pid, at)
elif kind == "pool_user_sub":
def go():
with uc.repository() as uow:
other = uow.persons.find_by_sub(value)
if other is not None and other != pid: # one pool user is one person
log.warning(json.dumps({"event": "sub_in_use", "person_id": pid}))
return
p = uow.persons.get(pid)
if p.pool_user_sub != value:
p.pool_user_sub = value
uow.persons.save(p)
_committed(go)
elif kind == "file_contract":
file_contract(run_id, value, at)
def file_contract(run_id: str, info: dict, at: datetime) -> None:
"""The signed copy lands on the person as a pay-bearing document; then the step is done."""
import base64
uc = _uc()
f = info.get("file") or {}
content = base64.b64decode(f.get("content_base64") or "")
with uc.repository() as uow:
run = uow.runs.get(run_id)
pid, step = run.person_id, run.steps[info["step_no"]]
doc_type = step.template_kind or "employment_contract"
file_document_bytes(pid, f"contract:{info['contract_id']}", doc_type, f.get("filename") or "contract.pdf",
f.get("content_type") or "application/pdf", content, ACTOR, at,
contract_id=info["contract_id"])
def done():
with uc.repository() as uow:
run = uow.runs.get(run_id)
s = run.steps[info["step_no"]]
if s.state == R.DONE:
return
s.result = {**(s.result or {}), "phase": "signed", "document": f"contract:{info['contract_id']}"}
s.state, s.done_at, s.done_by, s.last_status, s.next_check_at = R.DONE, at, "contracts", "done", None
run.add_log("contract signed", "contracts", at, s.step_no, info["contract_id"])
view = _view(uow, at)
p = uow.persons.get(run.person_id)
run.on_done(s, _calendar(view, p, run.n), view.tz)
uow.runs.save(run)
_committed(done)
def file_document_bytes(person_id: int, doc_id: str, document_type: str, filename: str, content_type: str,
content: bytes, by: str, at: datetime, *, contract_id: Optional[str] = None,
expires_on: Optional[date] = None) -> Document:
uc = _uc()
key = f"hr/{int(person_id)}/documents/{doc_id.replace(':', '-').replace('|', '-')}"
with uc.repository() as uow:
cfg = uow.config.load()
dt = cfg.document_type(document_type) or {"access_class": "pay" if contract_id else "personnel"}
sha, size = uow.objects.put(key, content, content_type)
p = uow.persons.get(person_id)
doc = Document(doc_id, document_type, dt["access_class"], filename, content_type, key, sha, size, by, at,
expires_on, contract_id)
if p.file_document(doc):
uow.persons.save(p)
return doc
# ── warnings (HR-41) ───────────────────────────────────────────────────────────────────────
def warnings(at: datetime) -> list:
"""Set-based: a query finds who MAY be due (dates in the window); only they are loaded and decided on."""
uc = _uc()
made = []
with uc.repository() as uow:
view = _view(uow, at)
days_end = view.config.int_setting("warning_days_contract_end")
days_rtw = view.config.int_setting("warning_days_right_to_work")
candidates = uow.persons.warning_candidates(view.today, days_end, days_rtw)
if not candidates:
return made
people = uow.persons.load_many(candidates)
existing = set(uow.runs.ids())
for p in people:
for e in p.engagements.values():
if e.planned_end and e.actual_end is None and view.today <= e.planned_end <= view.today + timedelta(days=days_end):
rid = make_run_id("warning", p.person_id, e.n, f"end-{e.planned_end}")
if rid not in existing:
made += _warning_run(rid, p, e.n, "manager", f"{p.name(view.today)}'s engagement ends "
f"{e.planned_end}: extend it or record a leaver", at)
rtw = p.compliance_on(e.n, "right_to_work_check")
fu = rtw.data.get("follow_up_on") if rtw else None
if fu and e.end >= view.today and view.today <= fu <= view.today + timedelta(days=days_rtw):
rid = make_run_id("warning", p.person_id, e.n, f"rtw-{fu}")
if rid not in existing:
made += _warning_run(rid, p, e.n, "hr", f"Follow-up right-to-work check for "
f"{p.name(view.today)} before {fu}", at)
return made
def _warning_run(rid, p, n, group, title, at) -> list:
uc = _uc()
def go():
with uc.repository() as uow:
if uow.runs.exists(rid):
return []
run = R.Run(rid, "warning", p.person_id, n, rid.split(":")[-1], None, None, at)
run.steps["1"] = R.RunStep("1", 1, "task", title, group, state=R.WAITING, due_at=at)
run.add_log("warning", ACTOR, at, None, title)
uow.runs.save(run)
return [rid]
return _committed(go) or []
# ── retention (HR-40) ──────────────────────────────────────────────────────────────────────
SENSITIVE_BY_CATEGORY = {}
for _k, _c in CATEGORY_OF.items():
SENSITIVE_BY_CATEGORY.setdefault(_c, set()).add(_k)
def retention_sweep(at: datetime, force: bool = False) -> list:
"""Once a day, in the first tick after the configured hour (org time): each category past its period is
purged or anonymised per its rule, one retention-log row per action. Twice in a day changes nothing."""
uc = _uc()
with uc.repository() as uow:
view = _view(uow, at)
hour = view.config.int_setting("retention_sweep_hour")
local = work.aware(at).astimezone(work.zone(view.tz))
if local.hour < hour and not force:
return []
rules = {r["category"]: r for r in view.config.all("retention_rule")}
# set-based: only people with an engagement that ended before today can be due
candidates = uow.persons.ended_before(view.today) if rules else set()
people = uow.persons.load_many(candidates) if candidates else []
# No global "ran today" marker: each (person, category) is acted on once EVER (apply_categories reads the
# retention log first), so a second sweep on the same day — or a concurrent one — writes nothing.
out = []
for p in people:
if not p.engagements or any(e.actual_end is None or e.actual_end >= view.today for e in p.engagements.values()):
continue
ended = max(e.actual_end for e in p.engagements.values())
due = {c: r for c, r in rules.items() if retention_due(r, ended, view.today)}
if due:
out += apply_categories(p.person_id, {c: (r["action"], r["basis"]) for c, r in due.items()}, at,
"retention sweep")
return out
def apply_categories(person_id: int, plan: dict, at: datetime, by: str) -> list:
"""Purge / anonymise / keep each category for one person; one retention-log row per category, once ever."""
uc = _uc()
with uc.repository() as uow:
done = uow.retention.done(person_id)
today = _view(uow, at).today
todo = {c: v for c, v in plan.items()
if not any(r["action"] in ("purged", "anonymised", "kept") for r in done.get(c, []))
or (v[0] == "kept" and not done.get(c))}
todo = {c: v for c, v in todo.items() if not (v[0] != "kept" and any(
r["action"] in ("purged", "anonymised") for r in done.get(c, [])))}
if not todo:
return []
kinds = set()
for c, (action, _b) in todo.items():
if action != "kept":
kinds |= SENSITIVE_BY_CATEGORY.get(c, set())
keys_to_delete = []
def sensitive():
with uc.repository() as uow:
s = uow.sensitive.get(person_id)
if s is not None and kinds & set(s.kinds):
keys_to_delete.extend(s.purge(kinds & set(s.kinds), now=at))
uow.sensitive.save(s)
_committed(sensitive)
anonymised = False
def person():
nonlocal anonymised
with uc.repository() as uow:
p = uow.persons.get(person_id)
if p is None:
return
changed = False
for c, (action, _b) in todo.items():
if action == "kept":
continue
if c in ("documents_pay", "documents_personnel", "right_to_work"):
cls = {"documents_pay": "pay", "documents_personnel": "personnel", "right_to_work": "sensitive"}[c]
for doc_id in [d for d, doc in p.documents.items() if doc.access_class == cls]:
keys_to_delete.append(p.documents.pop(doc_id).object_key)
changed = True
if c == "person_detail":
for sl in p.detail.slices:
sl.data.update(legal_given_name="Erased", legal_family_name="person", preferred_name="",
personal_email=None, personal_phone=None, work_email=None, work_phone=None)
changed = anonymised = True
if changed:
uow.persons.save(p)
_committed(person)
def objects():
with uc.repository() as uow:
for k in keys_to_delete:
if k:
uow.objects.delete(k)
_committed(objects)
if "runs" in todo and todo["runs"][0] != "kept":
_scrub_runs(person_id, at)
if anonymised:
bp.anonymise(person_id, ACTOR)
out = []
for c, (action, basis) in sorted(todo.items()):
word = {"purge": "purged", "anonymise": "anonymised", "kept": "kept"}.get(action, action)
try:
with uc.repository() as uow:
uow.retention.add(today, person_id, c, word, basis, by)
out.append({"person_id": person_id, "category": c, "action": word, "basis": basis})
except Exception as exc: # noqa: BLE001
if not is_duplicate(exc):
raise
return out
def _scrub_runs(person_id: int, at: datetime) -> None:
uc = _uc()
with uc.repository() as uow:
ids = uow.runs.ids(person_id)
for rid in ids:
def go(rid=rid):
with uc.repository() as uow:
r = uow.runs.get(rid)
for s in r.steps.values():
s.input, s.plan, s.draft_text = {}, "", None
s.result = {k: v for k, v in (s.result or {}).items() if k in ("phase",)}
uow.runs.save(r)
_committed(go)
# ── the tick ───────────────────────────────────────────────────────────────────────────────
TICK_WORKERS = int(os.environ.get("HR_TICK_WORKERS", "4"))
def _advance_all(due: list, at: datetime, parallel: bool) -> tuple:
"""Advance every due run. One run that cannot move must not stop every other run: it is logged (by type),
answered, and tried again on the next tick.
On the pg wire the runs of DIFFERENT people move at once (HR_TICK_WORKERS, default 4: each worker holds at
most two connections — the run's and a claim's — so 4 stays under erp_platform.db's pool of 8 and the
tables reader's 16-read cap). A person's runs move one after another, in order: two runs of one person both
writing back to the Person would race for its version. Every part of advance is already safe to run
concurrently (the claims; the version retry) — two ticks at once rely on it. Answers keep the order of `due`."""
def one(rid):
try:
return advance(rid, at), None
except Exception as e: # noqa: BLE001
log.warning(json.dumps({"event": "run_not_advanced", "run_id": rid, "error": type(e).__name__}))
return [], rid
results: dict = {}
if parallel and TICK_WORKERS > 1 and len(due) > 1:
from concurrent.futures import ThreadPoolExecutor
groups: dict = {}
for rid in due:
groups.setdefault(_person_of(rid), []).append(rid)
def group(rids):
return [(rid, one(rid)) for rid in rids]
with ThreadPoolExecutor(min(TICK_WORKERS, len(groups))) as pool:
# each worker runs in a copy of this request's context (one configuration read for the tick)
futures = [pool.submit(contextvars.copy_context().run, group, rids) for rids in groups.values()]
for f in futures:
results.update(f.result())
else:
for rid in due:
results[rid] = one(rid)
acted, failed = [], []
for rid in due:
a, f = results[rid]
acted += a
if f:
failed.append(f)
return acted, failed
def _person_of(run_id: str) -> str:
"""<direction>:<person_id>:<n>:<seq> → person_id (a run id that does not parse is its own group)."""
parts = str(run_id).split(":")
return parts[1] if len(parts) > 2 else run_id
def run_due_steps(at: Optional[datetime] = None) -> dict:
at = work.aware(at or work.now())
uc = _uc()
# One read for what the tick starts from: the events still to expand, the runs there are, and what is due.
with uc.repository() as uow:
pending, run_ids, due = uow.runs.tick_start(at)
parallel = getattr(getattr(uow.conn, "dialect", None), "name", "") == "postgresql"
expanded = expand_pending(at, (pending, run_ids))
synced = sync_due(at)
if expanded: # a run expanded just now may be due at once
with uc.repository() as uow:
due = uow.runs.needing_attention(at)
acted, failed = _advance_all(due, at, parallel)
warned = warnings(at)
swept = retention_sweep(at)
return {"at": at.isoformat().replace("+00:00", "Z"), "expanded": expanded,
"synced": [f"{pid}:{s}" for pid, s in synced], "acted": acted, "not_advanced": failed, "warnings": warned,
"retention": swept}
__all__ = ["run_due_steps", "expand", "build_run", "sync_partner", "advance", "file_document_bytes", "issue_link",
"ensure_intake", "apply_categories", "mark_manual", "context_for", "add_months"]