"""work — tracked work: steps that fall due relative to dates, a board of what is due, chasing.
hr runs a starter's onboarding as steps ("order the laptop 10 days before the start date, 09:00 London
time"); project delivery will run milestones the same way. They are separate bricks with separate tables,
but the SHAPE is one — a step is due at a rule relative to an anchor, the anchor moves, open steps move
with it, a person sees what is overdue — so the shape lives here, as pure domain code. No database, no
HTTP, no identity: a brick's domain layer imports it, its infrastructure stores the results.
WHY TIMES ARE COMPUTED IN THE PERSON'S ZONE and stored in UTC. "09:00 on the start date" is a wall-clock
time where the person is: 08:00Z in a London summer and 09:00Z in a London winter. Adding 24 hours to a
UTC instant gets that wrong twice a year, on exactly the day a clock changes — so a rule is evaluated as
a local date and a local time in its IANA zone (zoneinfo), and only then turned into an aware UTC
datetime. Which zone is the consumer's row (the person's, or the office's); this module only obeys it.
On a clock-change day, a local time that does not exist (01:30 on the spring-forward night) lands at the
same distance after the change (02:30 BST, which is 01:30Z), and one that happens twice (01:30 on the
fall-back night) is the FIRST of the two. Both are zoneinfo's `fold=0`, stated here so nobody re-derives it.
WORKING DAYS ("3 working days before the start date", "chase every 2 working days") are counted in a
WorkCalendar the consumer builds from its own work-schedule and holiday rows; this module holds no holiday
data. The count runs on local dates, so the clock time still lands in the zone, DST included.
NAIVE DATETIMES ARE UTC. DataK3 hands a TIMESTAMP back without a zone, and the platform writes UTC into
them (repository stamps datetime.now(timezone.utc)), so an anchor without tzinfo is read as UTC — the
same reading brick-activity's `_aware` makes.
THE DEMO CLOCK. A demo of a 30-day onboarding cannot wait 30 days, so `now()` can be moved forward with
ERP_DEMO_CLOCK_OFFSET ("+3d", "-2h30m", or an ISO time the clock should read). It is a business clock
only — claim leases, tokens and log stamps use the real one — and it is IMPOSSIBLE where it matters:
with DODIL_ENV=prod or live, `now()` raises instead of applying it, and every applied read logs a
WARNING, so a moved clock cannot hide in a log either.
"""
import json
import logging
import os
import re
from dataclasses import dataclass
from datetime import date, datetime, time, timedelta, timezone
from typing import Any, Callable, Iterable, Optional, Union
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
log = logging.getLogger("erp_platform.work")
UTC = timezone.utc
Anchor = Union[date, datetime, None]
_ANCHOR = re.compile(r"[A-Za-z0-9_.:-]{1,128}")
_TIME = re.compile(r"([01]\d|2[0-3]):([0-5]\d)")
def zone(name: str) -> ZoneInfo:
"""An IANA zone by name, or a ValueError that names it. (`tzdata` ships the database, so a slim
image with no /usr/share/zoneinfo still resolves Europe/London.)"""
try:
return ZoneInfo(name)
except (ZoneInfoNotFoundError, ValueError, TypeError):
raise ValueError(f"{name!r} is not an IANA time zone (e.g. Europe/London)") from None
def aware(value: datetime) -> datetime:
"""A datetime with a zone: a naive one is UTC (see the module docstring)."""
return value if value.tzinfo else value.replace(tzinfo=UTC)
UNITS = ("days", "working_days")
@dataclass(frozen=True)
class WorkCalendar:
"""Which days are working days — built by the CONSUMER from its own rows (hr: the work schedule and
the holiday list for the person's country or office). The platform holds no holiday data: which days
a bank holiday falls on is a customer's fact, and a list baked in here would be wrong somewhere.
`working_weekdays` uses Python's numbering, Monday = 0 … Sunday = 6.
"""
working_weekdays: frozenset = frozenset({0, 1, 2, 3, 4})
holidays: frozenset = frozenset()
def __post_init__(self) -> None:
days = frozenset(self.working_weekdays)
if not days or not days <= set(range(7)):
raise ValueError(f"working_weekdays are 0 (Monday) … 6 (Sunday), at least one; got {sorted(days)}")
if any(not isinstance(h, date) or isinstance(h, datetime) for h in self.holidays):
raise ValueError("holidays are dates")
object.__setattr__(self, "working_weekdays", days)
object.__setattr__(self, "holidays", frozenset(self.holidays))
def is_working(self, day: date) -> bool:
return day.weekday() in self.working_weekdays and day not in self.holidays
def add_working_days(self, day: date, n: int) -> date:
"""`n` working days after `day` (before it, when negative), not counting `day` itself.
ZERO IS THE ANCHOR ITSELF, working day or not: "0 working days after a start date that is a
Saturday" is that Saturday. Moving it would change a rule nobody asked to change; a consumer that
wants "the first working day on or after" says +1 from the day before, or checks is_working().
"""
step = 1 if n > 0 else -1
left = abs(n)
while left:
day += timedelta(days=step)
if self.is_working(day):
left -= 1
return day
def _shift(day: date, n: int, unit: str, calendar: Optional[WorkCalendar]) -> date:
if unit == "days":
return day + timedelta(days=n)
if calendar is None:
raise ValueError("an offset in working_days needs the consumer's WorkCalendar")
return calendar.add_working_days(day, n)
def step_anchor(step_no: Any) -> str:
"""The anchor name for "when step <n> was done": `step:3.done`."""
return f"step:{step_no}.done"
# ── when a step is due ─────────────────────────────────────────────────────────────────────
@dataclass(frozen=True)
class TimingRule:
"""Due `offset_days` after (or, negative, before) an anchor, at `time_of_day` in `timezone`.
TimingRule("start_date", -10, "09:00", "Europe/London") # ten days before they start, 9am
TimingRule("run.created", 0, None, "UTC") # at once: the anchor's own moment
`time_of_day=None` keeps the anchor's own local time (midnight for a date anchor).
`offset_unit="working_days"` counts offset_days in the consumer's WorkCalendar, passed to due_at —
"3 working days before the start date" skips the weekend and the bank holiday in between.
"""
anchor: str
offset_days: int = 0
time_of_day: Optional[str] = "09:00"
timezone: str = "UTC"
offset_unit: str = "days"
def __post_init__(self) -> None:
if not isinstance(self.anchor, str) or not _ANCHOR.fullmatch(self.anchor):
raise ValueError(f"anchor must be a name like start_date or step:3.done, got {self.anchor!r}")
if isinstance(self.offset_days, bool) or not isinstance(self.offset_days, int):
raise ValueError(f"offset_days is a whole number of days, got {self.offset_days!r}")
if self.time_of_day is not None and not (isinstance(self.time_of_day, str)
and _TIME.fullmatch(self.time_of_day)):
raise ValueError(f"time_of_day is HH:MM (24h), got {self.time_of_day!r}")
zone(self.timezone)
if self.offset_unit not in UNITS:
raise ValueError(f"offset_unit is one of {UNITS}, got {self.offset_unit!r}")
def due_at(self, anchors: dict, calendar: Optional[WorkCalendar] = None) -> Optional[datetime]:
"""The aware UTC moment this rule falls due, or None while its anchor is not known yet.
`calendar` is required when the offset is in working days (and ignored otherwise)."""
value = anchors.get(self.anchor)
if value is None:
return None
tz = zone(self.timezone)
if isinstance(value, datetime):
local = aware(value).astimezone(tz)
day, clock = local.date(), local.time().replace(tzinfo=None)
elif isinstance(value, date):
day, clock = value, time(0, 0)
else:
raise ValueError(f"anchor {self.anchor} is {value!r}; a date or a datetime was expected")
if self.time_of_day is not None:
hh, mm = map(int, self.time_of_day.split(":"))
clock = time(hh, mm)
# Wall-clock arithmetic in the zone, THEN to UTC: "+1 day at 09:00" is 09:00 tomorrow there,
# whether or not the clocks change tonight.
local_due = datetime.combine(_shift(day, self.offset_days, self.offset_unit, calendar), clock, tzinfo=tz)
return local_due.astimezone(UTC)
@dataclass(frozen=True)
class TrackedStep:
"""A step as recompute() needs it: its key, its rule, and whether it is done."""
key: Any
rule: TimingRule
done: bool = False
def recompute(steps: Iterable[TrackedStep], anchors: dict, calendar: Optional[WorkCalendar] = None) -> dict:
"""The new due time of every step that is NOT done, after anchors changed: {key: due_at or None}.
A done step is left out on purpose — when it was due is history (it may have been late, and a report
says so), and moving it after the fact would rewrite that. `anchors` is every anchor as it now is,
the changed ones included; a step whose anchor is still unknown comes back as None.
"""
return {s.key: s.rule.due_at(anchors, calendar) for s in steps if not s.done}
# ── what is due: the board ───────────────────────────────────────────────────────────────────
def board(items: Iterable, *, now: datetime, timezone: str = "UTC",
due: Callable[[Any], Union[date, datetime, None]] = lambda i: i.due_at,
is_open: Callable[[Any], bool] = lambda i: True, week_days: int = 7) -> dict:
"""A person's open items as overdue, today and this_week — brick-activity's task board, generalised.
"Today" is the person's today: `now` is read in their zone. An item due at a TIME is overdue once that
time has passed (a 09:00 step at 10:00 is late), otherwise it is today's if it falls on their local
today; one due on a DATE is overdue from the day after. this_week is the next `week_days` local days;
later items and items with no due time yet are not on the board. Each list is ordered by due time.
"""
tz = zone(timezone)
now = aware(now)
today = now.astimezone(tz).date()
week_end = today + timedelta(days=week_days)
out: dict = {"overdue": [], "today": [], "this_week": []}
ranked = []
for item in items:
when = due(item)
if when is None or not is_open(item):
continue
if isinstance(when, datetime):
instant = aware(when)
day = instant.astimezone(tz).date()
late = instant < now
else:
day, late = when, when < today
instant = datetime.combine(when, time(0, 0), tzinfo=tz)
if late:
bucket = "overdue"
elif day == today:
bucket = "today"
elif day <= week_end:
bucket = "this_week"
else:
continue
ranked.append((instant, bucket, item))
for _instant, bucket, item in sorted(ranked, key=lambda r: r[0]):
out[bucket].append(item)
return out
# ── chasing ──────────────────────────────────────────────────────────────────────────────────
@dataclass(frozen=True)
class ChaseCadence:
"""Chase every `every_days` days, at most `max_count` times — the consumer's config row.
`first_after_days` (default: every_days) is the wait before the FIRST chase, counted from when the
request went out. Days are wall-clock days in `timezone`, so a 09:00 chase stays a 09:00 chase across
a clock change; `unit="working_days"` counts them in the consumer's WorkCalendar ("chase every 2
working days"), passed to next_chase.
"""
every_days: int
max_count: int
first_after_days: Optional[int] = None
timezone: str = "UTC"
unit: str = "days"
def __post_init__(self) -> None:
for name in ("every_days", "max_count"):
v = getattr(self, name)
if isinstance(v, bool) or not isinstance(v, int) or v < (1 if name == "every_days" else 0):
raise ValueError(f"{name} must be a whole number ≥ {1 if name == 'every_days' else 0}, got {v!r}")
f = self.first_after_days
if f is not None and (isinstance(f, bool) or not isinstance(f, int) or f < 0):
raise ValueError(f"first_after_days must be a whole number ≥ 0, got {f!r}")
zone(self.timezone)
if self.unit not in UNITS:
raise ValueError(f"unit is one of {UNITS}, got {self.unit!r}")
@dataclass(frozen=True)
class ChaseState:
count: int # chases already sent
next_at: Optional[datetime] # aware UTC; None when exhausted
exhausted: bool # max_count reached: stop chasing, escalate to a person
def due(self, now: datetime) -> bool:
"""Is a chase due at `now`?"""
return self.next_at is not None and aware(now) >= self.next_at
def next_chase(cadence: ChaseCadence, sent_at: datetime, history: Iterable[datetime] = (),
calendar: Optional[WorkCalendar] = None) -> ChaseState:
"""When the next chase is due, given when the request went out and the chases sent since.
`calendar` is required for a cadence in working days."""
chases = sorted(aware(h) for h in history)
if len(chases) >= cadence.max_count:
return ChaseState(len(chases), None, True)
tz = zone(cadence.timezone)
if chases:
base, wait = chases[-1], cadence.every_days
else:
base = aware(sent_at)
wait = cadence.every_days if cadence.first_after_days is None else cadence.first_after_days
local = base.astimezone(tz)
nxt = datetime.combine(_shift(local.date(), wait, cadence.unit, calendar),
local.time().replace(tzinfo=None), tzinfo=tz)
return ChaseState(len(chases), nxt.astimezone(UTC), False)
# ── the clock ────────────────────────────────────────────────────────────────────────────────
OFFSET_ENV = "ERP_DEMO_CLOCK_OFFSET"
LIVE_ENVS = frozenset({"prod", "live"})
_OFFSET = re.compile(r"([+-])(?:(\d+)d)?(?:(\d+)h)?(?:(\d+)m)?")
_target_offsets: dict = {} # ISO target → the offset fixed the first time it was read
class DemoClockRefused(RuntimeError):
"""ERP_DEMO_CLOCK_OFFSET is set where the business clock must be the real one."""
def _real_now() -> datetime:
return datetime.now(UTC)
def _offset(raw: str, real: datetime) -> timedelta:
m = _OFFSET.fullmatch(raw)
if m and any(m.groups()[1:]):
sign = -1 if m.group(1) == "-" else 1
d, h, mins = (int(g or 0) for g in m.groups()[1:])
return sign * timedelta(days=d, hours=h, minutes=mins)
# An ISO target: the clock reads that time the first time it is asked, and ticks on from there —
# a frozen clock would stop every "is it due yet" in the demo at one instant.
if raw not in _target_offsets:
try:
target = datetime.fromisoformat(raw.replace("Z", "+00:00"))
except ValueError:
raise ValueError(f"{OFFSET_ENV}={raw!r} is neither an offset (+3d, -2h30m) nor an ISO time") from None
_target_offsets[raw] = aware(target) - real
return _target_offsets[raw]
# The clock set IN THIS PROCESS — by a demo's clock route (erp_platform.democlock) or the scenario harness
# (erp_platform.scenarios). Empty = not set; {"offset": timedelta} ticks on from a moved instant;
# {"frozen": datetime} stands still until it is moved again. It wins over ERP_DEMO_CLOCK_OFFSET.
_SET: dict = {}
def _live_env() -> str:
env = os.environ.get("DODIL_ENV", "").strip().lower()
return env if env in LIVE_ENVS else ""
def parse_instant(text: str) -> datetime:
"""An instant as a person writes it, as aware UTC: an ISO time with an offset or Z
("2026-10-23T08:00Z"), or a local wall-clock time and an IANA zone ("2026-03-29T00:30 Europe/London").
A local time a clock change skips or repeats follows the module's fold=0 rule. Naive ISO is UTC."""
raw = (text or "").strip()
stamp, _, name = raw.partition(" ")
try:
value = datetime.fromisoformat(stamp.replace("Z", "+00:00"))
except ValueError:
raise ValueError(f"{text!r} is not a time: write 2026-10-23T08:00Z or 2026-10-23T09:00 Europe/London") from None
if name:
if value.tzinfo is not None:
raise ValueError(f"{text!r} gives both an offset and a zone — one of them, please")
# fold=0: a skipped time (spring forward) lands the same distance after the change; a repeated one
# (fall back) is the first of the two.
return value.replace(tzinfo=zone(name.strip()), fold=0).astimezone(UTC)
return aware(value).astimezone(UTC)
def set_clock(*, at: Optional[datetime] = None, offset: Optional[timedelta] = None, freeze: bool = False) -> datetime:
"""Move the business clock in this process: to `at`, or by `offset` from the real time; `freeze` stops it
there (a scenario wants every read to agree), otherwise it ticks on. Refused under DODIL_ENV=prod|live,
exactly like ERP_DEMO_CLOCK_OFFSET. Returns the time the clock now reads."""
if _live_env():
raise DemoClockRefused(f"DODIL_ENV={_live_env()}: the business clock is never moved on a live deployment")
if (at is None) == (offset is None):
raise ValueError("set the clock to a time (at=) or by an offset (offset=), not both")
real = _real_now()
target = aware(at) if at is not None else real + offset
_SET.clear()
_SET.update({"frozen": target} if freeze else {"offset": target - real})
return now()
def advance_clock(delta: timedelta) -> datetime:
"""Move the set clock on by `delta` (frozen stays frozen). With nothing set, from the real time."""
if not _SET:
return set_clock(offset=delta)
if _live_env():
raise DemoClockRefused(f"DODIL_ENV={_live_env()}: the business clock is never moved on a live deployment")
if "frozen" in _SET:
_SET["frozen"] = _SET["frozen"] + delta
else:
_SET["offset"] = _SET["offset"] + delta
return now()
def reset_clock() -> None:
"""Back to the real clock (and ERP_DEMO_CLOCK_OFFSET, if it is set)."""
_SET.clear()
def clock_state() -> dict:
"""What the business clock reads, and whether and how it was moved — for a demo's clock route."""
current = now()
real = _real_now()
return {"now": current.isoformat().replace("+00:00", "Z"), "moved": bool(_SET) or bool(os.environ.get(OFFSET_ENV)),
"frozen": "frozen" in _SET, "offset_seconds": round((current - real).total_seconds(), 3)}
def now() -> datetime:
"""The business clock: aware UTC — real, unless a demo moved it (see the module docstring)."""
real = _real_now()
raw = os.environ.get(OFFSET_ENV, "").strip()
if not raw and not _SET:
return real
env = os.environ.get("DODIL_ENV", "").strip().lower()
if env in LIVE_ENVS:
raise DemoClockRefused(f"{OFFSET_ENV if raw else 'the demo clock'} is set and DODIL_ENV={env}: a demo "
"clock is never applied to a live deployment — unset it")
if "frozen" in _SET:
current = _SET["frozen"]
else:
current = real + (_SET["offset"] if _SET else _offset(raw, real))
log.warning(json.dumps({"event": "demo_clock", "offset_seconds": (current - real).total_seconds(),
"dodil_env": env or None}))
return current