"""The Item aggregate: one committed thing somebody delivers, and every rule about moving it.
No database, no HTTP, no identity: a rule here is true whoever calls it and however it is stored. Time is
handed in (`now`, the business clock from `work.now()`), and so are the customer's settings and calendar,
so every rule is a plain function of what it is given.
"""
from dataclasses import dataclass, field
from datetime import date, datetime
from typing import Optional
from urllib.parse import urlsplit
from erp_platform.work import WorkCalendar
from brick_projects.domain.errors import RuleBroken
from brick_projects.domain.events import Moved
from brick_projects.domain.values import (
OPEN_ITEM_STATES,
ItemApprovedHow,
ItemBillingStatus,
ItemDeliveryStatus,
ItemSource,
ItemState,
ProjectState,
Settings,
)
NO_EVIDENCE = "Attach the evidence first: a link, a file or a sign-off"
SUBMITTER = "You submitted this work, so someone else records its acceptance"
WORK_REF_HINT = "work_ref is an https link, e.g. https://git.example/team/project/-/issues/12"
def check_work_ref(value: str) -> str:
"""Where the team does the work (an issue or milestone URL): https only, optional ("" clears it). It is not
evidence: it never satisfies the acceptance gate."""
ref = (value or "").strip()
if not ref:
return ""
try:
parts = urlsplit(ref)
port = parts.port # raises on an invalid port
except ValueError:
raise RuleBroken(WORK_REF_HINT) from None
plain = all(c.isascii() and c.isprintable() and not c.isspace() for c in ref) # no controls, spaces, format or bidi characters
if (parts.scheme != "https" or not parts.hostname or len(ref) > 500 or not plain or "\\" in ref
or "@" in parts.netloc or (port is not None and not 0 < port < 65536)):
raise RuleBroken(WORK_REF_HINT)
return ref
def calendar_of(settings: Settings) -> WorkCalendar:
return WorkCalendar(working_weekdays=frozenset(settings.working_weekdays),
holidays=frozenset(date.fromisoformat(d) for d in settings.holidays))
def working_days_between(cal: WorkCalendar, start: date, end: date) -> int:
"""Working days after `start` up to and including `end` (0 when end <= start)."""
n, day = 0, start
while day < end:
day = cal.add_working_days(day, 1)
if day <= end:
n += 1
return n
def _day(v) -> Optional[date]:
if v is None:
return None
return v.date() if isinstance(v, datetime) else v
@dataclass
class Item:
"""one committed thing somebody delivers"""
item_id: str
project_id: str = ""
bp_id: int = 0
source_kind: str = ""
source_id: str = ""
line_id: str = ""
product_id: str = ""
billing_kind: str = ""
billing_plan: str = ""
source: str = ItemSource.SOLD.value
title: str = ""
owner: str = ""
due_on: Optional[date] = None
blocked_reason: str = ""
blocked_since: Optional[date] = None
blocked_days: int = 0
submitted_at: Optional[datetime] = None
submitted_by: str = "" # who did the work (email, else sub): they record its acceptance only
submitted_sub: str = "" # ... and their sub: either identity matching is the same person
revision: int = 0 # how many times the client asked for changes (PJ-09)
warned_at: Optional[datetime] = None
approved_at: Optional[datetime] = None
approved_by: str = ""
approved_how: str = ""
delivered_at: Optional[datetime] = None
evidence: str = ""
evidence_kind: str = ""
makegood_for: str = ""
changes_reason: str = "" # what the client last asked to change (PJ-09)
work_ref: str = "" # where the team does the work (https link); not evidence
billing_status: str = ItemBillingStatus.NOT_BILLED.value
delivery_status: str = ItemDeliveryStatus.NOT_REPORTED.value
state: str = ItemState.PLANNED.value
events: list = field(default_factory=list, repr=False, compare=False)
def _moved(self, kind: str, frm: str, by: str, detail: str = "") -> Moved:
e = Moved(self.item_id, kind, frm, self.state, by, detail)
self.events.append(e)
return e
@property
def is_open(self) -> bool:
return self.state in OPEN_ITEM_STATES
def may_move_in(self, project_state: str) -> None:
"""Work moves only while its project is open. A cancelled project's items are frozen; a closed one's sold
items stay as reconciliation left them (a short item is made good by a NEW item, never by finishing it —
or it would be delivered and billed twice); only its makegoods are still worked."""
if project_state == ProjectState.CANCELLED:
raise RuleBroken("the project is cancelled: its work no longer moves")
if project_state == ProjectState.CLOSED and self.source != ItemSource.MAKEGOOD.value:
raise RuleBroken("the project is closed and reconciled: work left short is made good with a makegood")
# ── how an item comes to exist ─────────────────────────────────────────────────────────────
@classmethod
def planned(cls, item_id: str, *, by: str = "", **attrs) -> "Item":
"""plan_from_contract: none → planned. Made by the project's planning (project.plan_sale), which
decides which lines make items; the item's key is <project>|<line>|<n>, so planning again finds it."""
it = cls(item_id=item_id, **attrs)
it.state = ItemState.PLANNED.value
it._moved("plan_from_contract", "none", by, f"from {it.source_kind} {it.source_id}, line {it.line_id}")
return it
@classmethod
def makegood(cls, short: "Item", *, project_closed: bool, by: str = "", title: str = "",
due_on: Optional[date] = None, owner: str = "") -> "Item":
"""raise_makegood: none → planned (PJ-07). Only for an item the closed project left short; the new
item carries no sold line and is never billed; the short item stays short. Its key is
<project>|mg|<short item>, so a second makegood for the same item is the same key — refused."""
if not project_closed:
raise RuleBroken("a makegood is raised once the project has been reconciled short")
if short.source != ItemSource.SOLD.value or not short.is_open:
raise RuleBroken(f"item {short.item_id} is not short: only sold work left open when the window "
"closed can be made good")
it = cls(item_id=f"{short.project_id}|mg|{short.item_id}", project_id=short.project_id,
bp_id=short.bp_id, source_kind=short.source_kind, source_id=short.source_id, line_id="",
product_id=short.product_id, source=ItemSource.MAKEGOOD.value,
title=title or f"Makegood: {short.title}", owner=owner or short.owner, due_on=due_on,
makegood_for=short.item_id)
it._moved("raise_makegood", "none", by, f"makes good {short.item_id}; no revenue")
return it
# ── doing the work ─────────────────────────────────────────────────────────────────────────
def assign(self, *, owner: Optional[str] = None, due_on: Optional[date] = None, title: Optional[str] = None,
work_ref: Optional[str] = None, by: str = "") -> Optional[Moved]:
"""The lead assigns, re-dates or renames open work, or points it at where it is worked (work_ref; "" clears
it). An accepted AI suggestion lands here too."""
if not self.is_open:
raise RuleBroken(f"a {self.state} item cannot be re-planned")
if work_ref is not None:
work_ref = check_work_ref(work_ref)
changed = {k: v for k, v in (("owner", owner), ("due_on", due_on), ("title", title), ("work_ref", work_ref))
if v is not None and getattr(self, k) != v}
if not changed:
return None
for k, v in changed.items():
setattr(self, k, v)
return self._moved("assign", self.state, by, ", ".join(f"{k} {v}" for k, v in changed.items()))
def start(self, *, by: str = "") -> Optional[Moved]:
"""start: planned → in_progress."""
if self.state == ItemState.IN_PROGRESS:
return None
if self.state != ItemState.PLANNED:
raise RuleBroken(f"a {self.state} item cannot be started")
self.state = ItemState.IN_PROGRESS.value
return self._moved("start", ItemState.PLANNED.value, by)
def block(self, reason: str, *, now: datetime, by: str = "") -> Optional[Moved]:
"""block: in_progress → blocked (PJ-02). Says what we wait for and since when."""
if self.state == ItemState.BLOCKED:
return None
if self.state != ItemState.IN_PROGRESS:
raise RuleBroken(f"a {self.state} item cannot be blocked: only work in progress waits on the client")
if not (reason or "").strip():
raise RuleBroken("say what you are waiting for")
self.state = ItemState.BLOCKED.value
self.blocked_reason, self.blocked_since = reason.strip(), _day(now)
return self._moved("block", ItemState.IN_PROGRESS.value, by, self.blocked_reason)
def unblock(self, *, now: datetime, by: str = "") -> Optional[Moved]:
"""unblock: blocked → in_progress. The days blocked are added to the item; slippage leaves them out."""
if self.state == ItemState.IN_PROGRESS:
return None
if self.state != ItemState.BLOCKED:
raise RuleBroken(f"a {self.state} item is not blocked")
days = max(0, (_day(now) - self.blocked_since).days) if self.blocked_since else 0
self.blocked_days += days
self.state = ItemState.IN_PROGRESS.value
detail = f"blocked {days} day(s) on: {self.blocked_reason}"
self.blocked_reason, self.blocked_since = "", None
return self._moved("unblock", ItemState.BLOCKED.value, by, detail)
def attach_evidence(self, evidence: str, kind: str, settings: Settings, *, by: str = "") -> Optional[Moved]:
"""The proof of delivery: a link, a file reference or a sign-off, of a kind the settings offer."""
if not self.is_open:
raise RuleBroken(f"a {self.state} item's evidence is final")
evidence, kind = (evidence or "").strip(), (kind or "").strip()
if not evidence:
raise RuleBroken(NO_EVIDENCE)
if kind not in settings.evidence_kinds:
raise RuleBroken(f"evidence is one of: {', '.join(settings.evidence_kinds)}")
if (self.evidence, self.evidence_kind) == (evidence, kind):
return None
self.evidence, self.evidence_kind = evidence, kind
return self._moved("attach_evidence", self.state, by, f"{kind}: {evidence}")
def submit(self, *, now: datetime, by: str = "", sub: str = "") -> Optional[Moved]:
"""submit: in_progress → in_review. Starts the review clock."""
if self.state == ItemState.IN_REVIEW:
return None
if self.state != ItemState.IN_PROGRESS:
raise RuleBroken(f"a {self.state} item cannot be submitted for review")
self.state = ItemState.IN_REVIEW.value
self.submitted_at, self.submitted_by, self.warned_at = now, (by or "").strip().lower(), None
self.submitted_sub = (sub or "").strip().lower()
return self._moved("submit", ItemState.IN_PROGRESS.value, by)
# ── the client's answer, recorded by staff ─────────────────────────────────────────────────
def window_ends_on(self, cal: WorkCalendar, settings: Settings) -> Optional[date]:
return cal.add_working_days(_day(self.submitted_at), settings.review_window_days) \
if self.submitted_at else None
def deemed_possible_on(self, cal: WorkCalendar, settings: Settings) -> Optional[date]:
return cal.add_working_days(_day(self.warned_at), settings.warning_notice_days) \
if self.warned_at else None
def review_clock(self, now: datetime, cal: WorkCalendar, settings: Settings) -> Optional[dict]:
"""PJ-11, derived on read and never stored: how long the client has left, whether the warning is due,
and from when deemed acceptance may be recorded. Nothing here sends anything."""
if self.state != ItemState.IN_REVIEW or not self.submitted_at:
return None
today, ends = _day(now), self.window_ends_on(cal, settings)
out = {"submitted_on": _day(self.submitted_at).isoformat(), "window_ends_on": ends.isoformat(),
"working_days_left": working_days_between(cal, today, ends),
"warning_due": self.warned_at is None and today > ends,
"warned_on": _day(self.warned_at).isoformat() if self.warned_at else None,
"deemed_possible_on": None, "deemed_acceptance": settings.deemed_acceptance}
if self.warned_at and settings.deemed_acceptance:
out["deemed_possible_on"] = self.deemed_possible_on(cal, settings).isoformat()
return out
def send_warning(self, *, now: datetime, cal: WorkCalendar, settings: Settings,
by: str = "") -> Optional[Moved]:
"""send_warning: a person has warned the client that silence will be taken as acceptance. Only once
the review window has passed with no answer; the system records it, it never sends it."""
if self.state != ItemState.IN_REVIEW:
raise RuleBroken(f"a {self.state} item is not waiting for the client's review")
if self.warned_at is not None:
return None
ends = self.window_ends_on(cal, settings)
if _day(now) <= ends:
raise RuleBroken(f"the client has until {ends.isoformat()} to review; warn them after that")
self.warned_at = now
return self._moved("send_warning", self.state, by,
f"deemed acceptance possible from {self.deemed_possible_on(cal, settings).isoformat()}")
def request_changes(self, reason: str, *, by: str = "") -> Optional[Moved]:
"""request_changes: in_review → in_progress (PJ-09). The same item, one more revision; the review
clock stops and starts again at the next submission."""
if self.state == ItemState.IN_PROGRESS and self.revision and self.submitted_at is None:
if (reason or "").strip() == self.changes_reason:
return None # the same change request again: a repeat changes nothing
raise RuleBroken(f"changes were already requested ({self.changes_reason!r}); submit the work again "
"before recording another request")
if self.state != ItemState.IN_REVIEW:
raise RuleBroken(f"a {self.state} item is not in review")
if not (reason or "").strip():
raise RuleBroken("say what the client asked to change")
self.state = ItemState.IN_PROGRESS.value
self.revision += 1
self.changes_reason = reason.strip()
self.submitted_at = self.warned_at = None
return self._moved("request_changes", ItemState.IN_REVIEW.value, by, reason.strip())
def submitted_by_me(self, by: str, identities: tuple = ()) -> bool:
"""The caller is the submitter if ANY of their identities (email, sub) is any of the submitter's."""
mine = {x.strip().lower() for x in (by, *identities) if x and x.strip()}
theirs = {x.strip().lower() for x in (self.submitted_by, self.submitted_sub) if x and x.strip()}
return bool(mine & theirs)
def _done(self, how: str, kind: str, now: datetime, settings: Settings, by: str, detail: str = "",
may_accept_own: bool = False, identities: tuple = ()) -> Moved:
if self.submitted_by_me(by, identities) and not may_accept_own:
raise RuleBroken(SUBMITTER)
if not self.evidence:
raise RuleBroken(NO_EVIDENCE)
self.state = ItemState.DONE.value
self.approved_how, self.approved_at, self.approved_by, self.delivered_at = how, now, by, now
if self.source == ItemSource.SOLD.value and settings.bills_on_acceptance(self.billing_kind, self.billing_plan):
self.billing_status = ItemBillingStatus.PENDING.value
if self.source == ItemSource.SOLD.value and self.source_kind == "sales_order" and self.line_id:
self.delivery_status = ItemDeliveryStatus.PENDING.value
return self._moved(kind, ItemState.IN_REVIEW.value, by, detail)
def accept(self, *, now: datetime, settings: Settings, by: str = "",
may_accept_own: bool = False, identities: tuple = ()) -> Optional[Moved]:
"""accept: in_review → done (PJ-04). Refused without evidence; records who, when and how (recorded).
A line billed on acceptance is left billing pending: the use case asks invoicing (PJ-22). Whoever submitted
the work records its acceptance only with `may_accept_own` (pj:item:accept_own; the use case decides)."""
if self.state == ItemState.DONE:
return None
if self.state != ItemState.IN_REVIEW:
raise RuleBroken(f"a {self.state} item is not in review: submit it first")
return self._done(ItemApprovedHow.RECORDED.value, "accept", now, settings, by, may_accept_own=may_accept_own,
identities=identities)
def accept_deemed(self, *, now: datetime, cal: WorkCalendar, settings: Settings,
by: str = "", may_accept_own: bool = False,
identities: tuple = ()) -> Optional[Moved]:
"""accept_deemed: in_review → done (PJ-03). Only where deemed acceptance exists, after the warning was
sent AND its notice ended, and with evidence; recorded as deemed, never as the client's yes."""
if self.state == ItemState.DONE:
return None
if self.state != ItemState.IN_REVIEW:
raise RuleBroken(f"a {self.state} item is not in review")
if not settings.deemed_acceptance:
raise RuleBroken("deemed acceptance is switched off here: record the client's answer")
if self.warned_at is None:
raise RuleBroken("warn the client first: acceptance is deemed only after a warning")
possible = self.deemed_possible_on(cal, settings)
if _day(now) < possible:
raise RuleBroken(f"the warning's notice runs until {possible.isoformat()}")
return self._done(ItemApprovedHow.DEEMED.value, "accept_deemed", now, settings, by,
f"warned {_day(self.warned_at).isoformat()}", may_accept_own=may_accept_own,
identities=identities)
def descope(self, reason: str, *, by: str = "") -> Optional[Moved]:
"""descope: open → cancelled (PJ-10). Keeps its history; reconciliation counts it as descoped."""
if self.state == ItemState.CANCELLED:
return None
if self.state == ItemState.DONE:
raise RuleBroken("a done item cannot be descoped")
if not (reason or "").strip():
raise RuleBroken("say why it is descoped")
frm = self.state
self.state = ItemState.CANCELLED.value
return self._moved("descope", frm, by, reason.strip())
# ── billing on acceptance (PJ-22): the use case asks; the item only remembers the answer ─────
def billing_requested(self, *, by: str = "") -> Optional[Moved]:
if self.state != ItemState.DONE or self.billing_status != ItemBillingStatus.PENDING.value:
return None
self.billing_status = ItemBillingStatus.REQUESTED.value
return self._moved("billing_requested", self.state, by, f"acc:{self.item_id}")
def delivery_reported(self, *, by: str = "") -> Optional[Moved]:
if self.state != ItemState.DONE or self.delivery_status != ItemDeliveryStatus.PENDING.value:
return None
self.delivery_status = ItemDeliveryStatus.REPORTED.value
return self._moved("delivery_reported", self.state, by, f"{self.source_id} line {self.line_id}")
# ── derived on read ────────────────────────────────────────────────────────────────────────
def slippage_days(self, now: datetime) -> int:
"""Days late, not counting the days it waited on the client (PJ-02). 0 if on time or undated."""
if self.due_on is None:
return 0
end = _day(self.delivered_at) if self.delivered_at else _day(now)
waiting = self.blocked_days + (max(0, (_day(now) - self.blocked_since).days)
if self.state == ItemState.BLOCKED and self.blocked_since else 0)
return max(0, (end - self.due_on).days - waiting)