"""The use cases — what a person (or their agent) can do with projects.
Each is declared ONCE and served as an API route and as an MCP tool. Two things are decided here and nowhere
else: WHO MAY (the `permission` on the declaration) and WHICH ROWS (`_sees_*`, from the caller: a row the caller
may not see is a 404, never a 403). Each command is one unit of work: load, let the domain decide, save whole
at the version it was loaded (`If-Match` for a screen).
Row visibility (module.yaml): a lead sees their own projects, a manager and finance every project; a contributor
sees the items assigned to them and their project's headline; a client sees one project through its link.
Calls to other modules (invoicing on acceptance, the order on delivery) happen AFTER the work is saved, through
infrastructure/seams.py: a neighbour that is down leaves the item billing pending; it never undoes acceptance.
"""
import dataclasses
from contextlib import contextmanager
from datetime import date, datetime, timedelta, timezone
from typing import Any, Optional
from erp_platform import auth, public, work
from erp_platform.bricks import Brick, Refused
from erp_platform.repository import retry_on_conflict
from erp_platform.types import BigId, Email, IsoDate, Name, Note, text
from brick_projects.domain.errors import RuleBroken
from brick_projects.domain.item import Item, calendar_of
from brick_projects.domain.project import SOURCE_KINDS, Project, SoldLine, plan_sale, project_key
from brick_projects.domain.readiness import ITEM_TRANSITIONS, item_gates, project_gates, stage_of
from brick_projects.domain.values import ItemBillingStatus, ItemDeliveryStatus, check_setting, setting_warning
from brick_projects.infrastructure import ai_steps, seams
PJ = Brick("pj", version="0.1.4", ids_as_strings=True, machine_permissions=("pj:item:accept_own", "pj:project:reconcile"), instructions=(
"Projects & service delivery: what was sold becomes a delivery plan of items; each item is worked, may wait "
"on the client (blocked, with a reason), is submitted with evidence, and is accepted by the client (recorded "
"by staff) or, after a warning and its notice, deemed accepted. When a project's window closes it is "
"reconciled: sold against delivered per sold line, and a shortfall can be made good with an item that is "
"never billed. Nothing here ever writes to a client: the system shows the clock and a person sends."))
# The process view (BRICKS.md): the signed workflows as BPMN, GET /process/bpmn.
PJ.declare_process("process.yaml")
# Every decision is a PERMISSION (module.yaml permissions.catalog); no role's name is ever compared. Which people hold
# which is the composition's role catalogue — roles.yaml beside this package is the recommended default.
READ_ALL = "pj:project:read_all" # see every project and its items (finance, a manager, the system)
MANAGE = "pj:project:manage" # act as the lead of every project (a manager)
RECONCILE = "pj:project:reconcile" # close a project's window: the scheduler's machine permission
ACCEPT_OWN = "pj:item:accept_own" # also record the acceptance of work one submitted oneself (a firm of one)
Key = text(300)
Reason = text(500)
@contextmanager
def _repository_default():
from erp_platform import db # imported late: it needs BUCKET
from brick_projects.infrastructure.tables import Repository
with db.engine.begin() as conn:
yield Repository(conn)
repository = _repository_default # tests and a composition swap in their own
@contextmanager
def _decide():
"""A broken business rule is a 409 with the rule's own words, through both doors."""
try:
yield
except RuleBroken as broken:
raise Refused(409, str(broken)) from None
def _who(caller: dict) -> str:
"""The caller as a stored owner is: their email, lower-case (issue #3), else their sub."""
return (caller.get("email") or caller.get("sub") or "").strip().lower()
def _ids(caller: dict) -> tuple:
"""Every identity the caller has (email and sub, lower-case): the same person under either is the same person."""
return tuple(dict.fromkeys(x.strip().lower() for x in (caller.get("email"), caller.get("sub")) if x and x.strip()))
def _now() -> datetime:
"""The business clock (work.now(): a demo may move it), as naive UTC like every stored time."""
return work.now().astimezone(timezone.utc).replace(tzinfo=None)
def _date(v: Optional[str], what: str) -> Optional[date]:
if v in (None, ""):
return None
try:
return date.fromisoformat(str(v)[:10])
except ValueError:
raise Refused(422, f"{what} is a date like 2026-11-30") from None
def _same_version(what: str, current: int, asked: Optional[int]) -> None:
if asked is not None and int(asked) != current:
raise Refused(409, f"{what} changed since you opened it (you have version {asked}, it is at {current}) "
"— load it again")
# ── who sees what ─────────────────────────────────────────────────────────────────────────────
def _sees_all(caller: dict) -> bool:
return auth.has_permission(caller, READ_ALL)
def _sees_project(caller: dict, project: Project, items: list) -> bool:
if public.is_public(caller):
return (public.subject_of(caller) or ())[:2] == ("project", project.project_id)
me = _who(caller)
return _sees_all(caller) or project.owner == me or any(i.owner == me for i, _ in items)
def _sees_item(caller: dict, item: Item, project: Optional[Project]) -> bool:
if public.is_public(caller):
return False
me = _who(caller)
return _sees_all(caller) or item.owner == me or (project is not None and project.owner == me)
def _manages(caller: dict, project: Optional[Project]) -> bool:
"""Who may act on a project as a whole, record a client's answer or re-plan its work: its lead (the owner), or
whoever holds pj:project:manage. Seeing a project (a contributor on one item, finance) is not enough."""
if public.is_public(caller):
return False
return auth.has_permission(caller, MANAGE) or (project is not None and project.owner == _who(caller))
def _must_manage(caller: dict, project: Optional[Project], what: str) -> None:
if not _manages(caller, project):
raise Refused(403, f"only the project's lead or a manager can {what}")
def _must_close(caller: dict) -> None:
"""Reconciling closes the window for everyone: the signed process gives it to the system (the scheduler, when
the window ends), which holds pj:project:reconcile. A lead does not close their own project early."""
if not auth.has_permission(caller, RECONCILE):
raise Refused(403, "a project is reconciled by the scheduler when its window ends")
def _project(repo, caller: dict, project_id: str) -> tuple:
found = repo.projects.get(project_id)
if found is None or not _sees_project(caller, found[0], repo.items.of_project(project_id)):
raise Refused(404, f"no project {project_id}")
return found
def _item(repo, caller: dict, item_id: str) -> tuple:
found = repo.items.get(item_id)
if found is None:
raise Refused(404, f"no item {item_id}")
project = repo.projects.get(found[0].project_id)
if not _sees_item(caller, found[0], project[0] if project else None):
raise Refused(404, f"no item {item_id}")
return found
# ── how things read ───────────────────────────────────────────────────────────────────────────
def _fields(agg) -> dict:
return {f.name: getattr(agg, f.name) for f in dataclasses.fields(agg) if f.name != "events"}
def _item_view(item: Item, version: int, settings, now: datetime, project: Optional[Project] = None,
names: Optional[dict] = None) -> dict:
out = _fields(item)
out["version"] = version
out["state_name"] = (names or {}).get(item.state, item.state)
out["review_clock"] = item.review_clock(now, calendar_of(settings), settings)
out["slippage_days"] = item.slippage_days(now)
out["short"] = bool(project and project.is_short(item))
return out
def _project_view(repo, caller: dict, project: Project, version: int) -> dict:
settings, now = repo.settings.current(), _now()
names = {s.state_id: s.name for s in repo.states.all()}
out = _fields(project)
out.pop("reconciliation_json")
out["version"] = version
out["reconciliation"] = project.reconciliation
seen = [(i, v) for i, v in repo.items.of_project(project.project_id)
if public.is_public(caller) or _sees_item(caller, i, project)]
out["stage"] = stage_of(project, [i for i, _ in seen], now.date()) # from what THIS caller sees
out["items"] = [_item_view(i, v, settings, now, project, names) for i, v in seen]
return out
# ── reading ───────────────────────────────────────────────────────────────────────────────────
@PJ.use_case(kind="query", route="GET /projects", permission="pj:project:read")
def list_projects(caller: dict, mine: bool = False, state: Optional[Name] = None) -> dict:
"""The projects the caller may see, newest sale first is not implied: ordered by id. `mine` narrows to the
projects the caller leads; `state` to one state (planned, active, closed, cancelled)."""
with repository() as repo:
items = repo.items.all()
out = []
for p, v in repo.projects.all():
its = [(i, iv) for i, iv in items if i.project_id == p.project_id]
if not _sees_project(caller, p, its) or (mine and p.owner != _who(caller)) or (state and p.state != state):
continue
out.append({"project_id": p.project_id, "bp_id": p.bp_id, "source_kind": p.source_kind,
"source_id": p.source_id, "owner": p.owner, "state": p.state, "starts_on": p.starts_on,
"ends_on": p.ends_on, "version": v, "items": len(its),
"done": sum(1 for i, _ in its if i.state == "done")})
return {"projects": out}
@PJ.use_case(kind="query", route="GET /projects/{project_id}", permission="pj:project:read", etag="version")
def get_project(caller: dict, project_id: Key) -> dict:
"""One project with the items the caller may see (each with its review clock and slippage) and, once
closed, its stored reconciliation of sold against delivered per sold line (PJ-06)."""
with repository() as repo:
return _project_view(repo, caller, *_project(repo, caller, project_id))
@PJ.use_case(kind="query", route="GET /items/{item_id}", permission="pj:item:read", etag="version")
def get_item(caller: dict, item_id: Key) -> dict:
"""One item, with its review clock (PJ-11: days left, whether the warning is due, when acceptance may be
deemed — derived, nothing is sent), its slippage net of days blocked on the client, and its history."""
with repository() as repo:
item, v = _item(repo, caller, item_id)
project = repo.projects.get(item.project_id)
names = {s.state_id: s.name for s in repo.states.all()}
out = _item_view(item, v, repo.settings.current(), _now(), project[0] if project else None, names)
out["history"] = repo.events.of(item_id)
return out
def _can_take(caller: dict, key: str, item: Item, project: Optional[Project]) -> bool:
"""Could THIS caller take the item's transition: do they hold its permission, and are they the person it is for
(the item's owner for the work steps; the project's lead or a manager for the rest)?"""
perm = _ITEM_STEP_PERMISSION[key]
if public.is_public(caller) or not auth.has_permission(caller, perm):
return False
return (perm == "pj:item:work" and item.owner == _who(caller)) or _manages(caller, project)
_ITEM_STEP_PERMISSION = {"start": "pj:item:work", "block": "pj:item:work", "unblock": "pj:item:work",
"submit": "pj:item:work", "send_warning": "pj:item:accept",
"request_changes": "pj:item:accept", "accept": "pj:item:accept",
"accept_deemed": "pj:item:accept", "descope": "pj:item:write"}
def _ready(key: str, name: str, gates: list) -> dict:
return {"transition": key, "name": name, "ready": all(g["met"] for g in gates), "gates": gates}
@PJ.use_case(kind="query", route="GET /items/{item_id}/readiness", permission="pj:item:read")
def item_readiness(caller: dict, item_id: Key) -> dict:
"""For a screen: each next transition of this item that the caller could take, with its gates as {gate, met,
reason, fix} in plain words (for example accept: evidence attached; someone else than the submitter records
it). A transition the caller's permissions never allow is left out. A record out of scope is a 404."""
with repository() as repo:
item, _ = _item(repo, caller, item_id)
found = repo.projects.get(item.project_id)
project = found[0] if found else None
gates = item_gates(item, now=_now(), settings=repo.settings.current(),
project_state=project.state if project else None, me=_who(caller), identities=_ids(caller),
may_accept_own=_may_accept_own(caller))
return {"item_id": item_id, "state": item.state,
"transitions": [_ready(k, ITEM_TRANSITIONS[k][0], g) for k, g in gates.items()
if _can_take(caller, k, item, project)]}
@PJ.use_case(kind="query", route="GET /projects/{project_id}/readiness", permission="pj:project:read")
def project_readiness(caller: dict, project_id: Key) -> dict:
"""For a screen: each next transition of this project that the caller could take (reconcile, cancel), with its
gates as {gate, met, reason, fix} in plain words, and the project's stage. A transition the caller's
permissions never allow is left out. A project out of scope is a 404."""
with repository() as repo:
project, _ = _project(repo, caller, project_id)
items = [i for i, _ in repo.items.of_project(project_id)
if public.is_public(caller) or _sees_item(caller, i, project)]
today = _now().date()
allowed = {} if public.is_public(caller) else {
"reconcile": auth.has_permission(caller, RECONCILE),
"cancel_project": auth.has_permission(caller, "pj:project:write") and _manages(caller, project)}
names = {"reconcile": "Close the window and reconcile", "cancel_project": "Cancel the project"}
return {"project_id": project_id, "state": project.state, "stage": stage_of(project, items, today),
"transitions": [_ready(k, names[k], g) for k, g in project_gates(project, today).items()
if allowed.get(k)]}
@PJ.use_case(kind="query", route="GET /board", permission="pj:item:read")
def board(caller: dict, mine: bool = False, timezone_name: Optional[text(64)] = None) -> dict:
"""What is due across every client the caller may see (PJ-08): overdue first, then today, then this week,
in the caller's time zone (or the business's). Blocked items say what they wait on and since when."""
with repository() as repo:
settings, now = repo.settings.current(), _now()
projects = {p.project_id: p for p, _ in repo.projects.all()}
names = {s.state_id: s.name for s in repo.states.all()}
mine_only = mine or not _sees_all(caller)
visible = [(i, v) for i, v in repo.items.all()
if i.is_open and _sees_item(caller, i, projects.get(i.project_id))
and (not mine or i.owner == _who(caller))]
tz = timezone_name or settings.timezone
try:
buckets = work.board(visible, now=now.replace(tzinfo=timezone.utc), timezone=tz,
due=lambda pair: pair[0].due_on)
except Exception: # noqa: BLE001 — an unknown zone is the caller's mistake
raise Refused(422, f"{tz!r} is not a time zone (e.g. Europe/Stockholm)") from None
def row(pair):
i, v = pair
return {"item_id": i.item_id, "project_id": i.project_id, "title": i.title, "owner": i.owner,
"due_on": i.due_on, "state": i.state, "state_name": names.get(i.state, i.state),
"waiting_on": i.blocked_reason or None, "waiting_since": i.blocked_since,
"slippage_days": i.slippage_days(now), "version": v}
return {"timezone": tz, "scope": "mine" if mine_only else "all",
**{k: [row(p) for p in rows] for k, rows in buckets.items()}}
@PJ.use_case(kind="query", route="GET /acceptance", permission="pj:project:read")
def acceptance(caller: dict, source_kind: Name, source_id: Key) -> dict:
"""Per sold line of a sale, whether its work is accepted, how (recorded or deemed) and when (PJ-23) — what a
contract's phase gate reads instead of keeping its own copy."""
with repository() as repo:
project = repo.projects.get(project_key(source_kind, source_id))
if project is None or not _sees_project(caller, project[0], repo.items.of_project(project[0].project_id)):
raise Refused(404, f"no project for {source_kind} {source_id}")
lines: dict = {}
for i, _ in repo.items.of_project(project[0].project_id):
if not i.line_id:
continue
line = lines.setdefault(i.line_id, {"line_id": i.line_id, "items": 0, "accepted": 0, "descoped": 0,
"items_detail": []})
if i.state == "cancelled":
line["descoped"] += 1 # out of the plan: neither owed nor accepted
else:
line["items"] += 1
line["accepted"] += i.state == "done"
line["items_detail"].append({"item_id": i.item_id, "state": i.state, "approved_how": i.approved_how or None,
"approved_at": i.approved_at})
for line in lines.values():
line["all_accepted"] = line["items"] > 0 and line["accepted"] == line["items"]
return {"project_id": project[0].project_id, "lines": [lines[k] for k in sorted(lines)]}
@PJ.use_case(kind="query", route="GET /states")
def list_states(caller: dict) -> dict:
"""The item states with the names this business uses for them, in board order; which are open, terminal and
count as delivered is fixed by the process."""
with repository() as repo:
return {"states": [{"state_id": s.state_id, "name": s.name, "position": s.position, "is_open": s.is_open,
"is_terminal": s.is_terminal, "counts_as_delivered": s.counts_as_delivered}
for s in repo.states.all()]}
@PJ.use_case(kind="query", route="GET /settings")
def list_settings(caller: dict) -> dict:
"""How this business delivers (module.yaml config_rows): each setting, its value (null = the reference
behaviour) and what it changes."""
with repository() as repo:
return {"settings": repo.settings.all()}
# ── planning from a sale (PJ-01, PJ-20, PJ-21) ────────────────────────────────────────────────
def _lines(raw: list) -> list:
if not isinstance(raw, list) or not 1 <= len(raw) <= 500:
raise Refused(422, "lines is a list of 1 to 500 sold lines")
out = []
for n, line in enumerate(raw, 1):
if not isinstance(line, dict):
raise Refused(422, f"line {n} is an object")
unknown = set(line) - {"line_id", "title", "quantity", "billing_kind", "billing_plan", "product_id", "due_on"}
if unknown:
raise Refused(422, f"line {n}: unknown field(s) {', '.join(sorted(unknown))}")
q = line.get("quantity", 1)
if isinstance(q, bool) or not isinstance(q, int) or not 0 <= q <= 10_000:
raise Refused(422, f"line {n}: quantity is a whole number up to 10000")
strings = {k: str(line.get(k) or "").strip() for k in ("line_id", "title", "billing_kind", "billing_plan",
"product_id")}
if any(len(v) > 200 or public_control(v) for v in strings.values()):
raise Refused(422, f"line {n}: a field is too long or has control characters")
out.append(SoldLine(line_id=strings["line_id"], title=strings["title"] or strings["line_id"], quantity=q,
billing_kind=strings["billing_kind"], billing_plan=strings["billing_plan"],
product_id=strings["product_id"], due_on=_date(line.get("due_on"), f"line {n} due_on")))
return out
def public_control(s: str) -> bool:
return any(ord(c) < 32 for c in s)
@PJ.use_case(kind="command", route="POST /plans", permission="pj:project:plan", idempotent=True)
def plan_from_contract(caller: dict, source_kind: Name, source_id: Key, bp_id: BigId, lines: list,
starts_on: Optional[IsoDate] = None, ends_on: Optional[IsoDate] = None,
owner: Optional[Email] = None, quote_id: Optional[Key] = None) -> dict:
"""Plan the work a sale needs (PJ-01, PJ-20, PJ-21): source_kind is contract, sales_order or manual; each
line is {line_id, title, quantity, billing_kind, billing_plan, product_id?, due_on?}. Lines are expanded by
this business's rule (by default a one-off single-payment line makes one item per unit, another one-off
line one item, and monthly or usage lines none: they elapse and invoicing bills them). The project is keyed
pj:<source_kind>:<source_id>: planning the same sale again creates nothing. A sale with nothing to deliver
opens no project."""
if source_kind not in SOURCE_KINDS:
raise Refused(422, f"source_kind is one of {', '.join(SOURCE_KINDS)}")
sold = _lines(lines)
with repository() as repo, _decide():
settings = repo.settings.current()
project, items = plan_sale(source_kind=source_kind, source_id=source_id, bp_id=int(bp_id),
starts_on=_date(starts_on, "starts_on"), ends_on=_date(ends_on, "ends_on"),
lines=sold, settings=settings, owner=(owner or _who(caller)).strip().lower(),
quote_id=quote_id or "", by=_who(caller))
if project is None:
return {"project": None, "created": 0, "reason": "nothing in this sale is delivered as work"}
found = repo.projects.get(project.project_id)
if found is None:
repo.projects.save(project, 0)
elif found[0].bp_id != project.bp_id:
raise Refused(409, f"{project.project_id} is already planned for another customer")
created = 0
for item in items:
if repo.items.get(item.item_id) is None:
repo.items.save(item, 0)
created += 1
return {"project": _project_view(repo, caller, *repo.projects.get(project.project_id)), "created": created}
# ── working an item ───────────────────────────────────────────────────────────────────────────
WORK, LEAD = "work", "lead"
def _step(item_id: str, caller: dict, version: Optional[int], act, *, who: str, what: str) -> dict:
"""Load the item the caller may see, check they may do THIS to it (work: its owner, the project's lead or a
manager; lead: the project's lead or a manager), that its project still moves, and their version; let the
domain act, save, answer the item."""
with repository() as repo, _decide():
item, current = _item(repo, caller, item_id)
project = repo.projects.get(item.project_id)
if not (who == WORK and item.owner == _who(caller)):
_must_manage(caller, project[0] if project else None, what)
_same_version(f"item {item_id}", current, version)
if project is not None:
item.may_move_in(project[0].state)
settings, now = repo.settings.current(), _now()
if act(item, settings, now, repo, project) is not None:
current = repo.items.save(item, current)
return _item_view(item, current, settings, now, project[0] if project else None)
@PJ.use_case(kind="command", route="POST /items/{item_id}/assign", permission="pj:item:write", idempotent=True,
if_match="version")
def assign_item(caller: dict, item_id: Key, owner: Optional[Email] = None, due_on: Optional[IsoDate] = None,
title: Optional[Name] = None, work_ref: Optional[text(500)] = None,
version: Optional[int] = None) -> dict:
"""Assign open work to a person, give it a due date, rename it, or say where the team does it (work_ref: an
https link such as an issue or milestone; "" clears it; it is not evidence). Fields left out keep their value."""
return _step(item_id, caller, version, lambda i, s, now, repo, p: i.assign(
owner=owner.strip().lower() if owner else None, due_on=_date(due_on, "due_on"), title=title,
work_ref=work_ref, by=_who(caller)), who=LEAD, what='re-plan this work')
@PJ.use_case(kind="command", route="POST /items/{item_id}/start", permission="pj:item:work", idempotent=True,
if_match="version")
def start(caller: dict, item_id: Key, version: Optional[int] = None) -> dict:
"""Start the work (planned → in progress). The project becomes active with its first started item."""
def act(item, settings, now, repo, project):
moved = item.start(by=_who(caller))
if moved is not None and project is not None and project[0].activate(by=_who(caller)) is not None:
repo.projects.save(project[0], project[1])
return moved
return _step(item_id, caller, version, act, who=WORK, what='start this work')
@PJ.use_case(kind="command", route="POST /items/{item_id}/block", permission="pj:item:work", idempotent=True,
if_match="version")
def block(caller: dict, item_id: Key, reason: Reason, version: Optional[int] = None) -> dict:
"""Say the work waits on the client (PJ-02), and for what: it shows blocked since today, and the days it waits
are not counted as our slippage."""
return _step(item_id, caller, version, lambda i, s, now, repo, p: i.block(reason, now=now, by=_who(caller)), who=WORK, what='block this work')
@PJ.use_case(kind="command", route="POST /items/{item_id}/unblock", permission="pj:item:work", idempotent=True,
if_match="version")
def unblock(caller: dict, item_id: Key, version: Optional[int] = None) -> dict:
"""The brief or assets arrived: back to in progress, the days blocked added to the item."""
return _step(item_id, caller, version, lambda i, s, now, repo, p: i.unblock(now=now, by=_who(caller)), who=WORK, what='unblock this work')
@PJ.use_case(kind="command", route="POST /items/{item_id}/evidence", permission="pj:item:work", idempotent=True,
if_match="version")
def attach_evidence(caller: dict, item_id: Key, evidence: Note, kind: Name, version: Optional[int] = None) -> dict:
"""Attach the proof of delivery: a link, a file reference or a sign-off (the kinds this business allows)."""
return _step(item_id, caller, version, lambda i, s, now, repo, p: i.attach_evidence(evidence, kind, s,
by=_who(caller)), who=WORK, what='attach evidence to this work')
@PJ.use_case(kind="command", route="POST /items/{item_id}/submit", permission="pj:item:work", idempotent=True,
if_match="version")
def submit(caller: dict, item_id: Key, evidence: Optional[Note] = None, evidence_kind: Optional[Name] = None,
version: Optional[int] = None) -> dict:
"""Submit the work for the client's review, optionally with its evidence. Starts the review clock."""
def act(item, settings, now, repo, project):
attached = item.attach_evidence(evidence, evidence_kind or "link", settings, by=_who(caller)) \
if evidence else None
return item.submit(now=now, by=_who(caller), sub=(caller.get("sub") or "")) or attached
return _step(item_id, caller, version, act, who=WORK, what='submit this work')
@PJ.use_case(kind="command", route="POST /items/{item_id}/send-warning", permission="pj:item:accept",
idempotent=True, if_match="version")
def send_warning(caller: dict, item_id: Key, version: Optional[int] = None) -> dict:
"""Record that you have warned the client, once their review window passed with no answer, that silence
will be taken as acceptance. The system never sends it: you do, and record it here."""
return _step(item_id, caller, version, lambda i, s, now, repo, p: i.send_warning(
now=now, cal=calendar_of(s), settings=s, by=_who(caller)), who=LEAD, what='record the warning')
@PJ.use_case(kind="command", route="POST /items/{item_id}/request-changes", permission="pj:item:accept",
idempotent=True, if_match="version")
def request_changes(caller: dict, item_id: Key, reason: Reason, version: Optional[int] = None) -> dict:
"""Record the client's change request (PJ-09): the same item goes back to in progress with one more revision,
and its review clock stops."""
return _step(item_id, caller, version, lambda i, s, now, repo, p: i.request_changes(reason, by=_who(caller)), who=LEAD, what="record the client's change request")
@PJ.use_case(kind="command", route="POST /items/{item_id}/descope", permission="pj:item:write", idempotent=True,
if_match="version")
def descope(caller: dict, item_id: Key, reason: Reason, version: Optional[int] = None) -> dict:
"""Take work out of the plan, with the reason (PJ-10): it keeps its history and is counted as descoped, not
short. Done work cannot be descoped."""
return _step(item_id, caller, version, lambda i, s, now, repo, p: i.descope(reason, by=_who(caller)), who=LEAD, what='descope this work')
def _mark(item_id: str, mark) -> None:
"""Record a neighbour's yes on the LATEST item: the acceptance is already committed, so a save that lost a
race is re-run, never reported as a failure (it would invite a second acceptance)."""
def unit():
with repository() as repo:
latest, v = repo.items.get(item_id)
if mark(latest) is not None:
repo.items.save(latest, v)
retry_on_conflict(unit)
def _handover(item_id: str, caller: dict) -> None:
"""Outside the acceptance's transaction, for DONE work only: tell the order one unit was delivered (PJ-21, SO-23)
and ask invoicing for the draft (PJ-22, AR-29). Each is pending until its neighbour says yes; a neighbour that
is down or absent leaves it pending for `retry_handover`, never lost and never done twice."""
with repository() as repo:
found = repo.items.get(item_id)
if found is None or found[0].state != "done":
return
item, me = found[0], _who(caller)
if item.delivery_status == ItemDeliveryStatus.PENDING.value and seams.record_delivery(item, me).done:
_mark(item_id, lambda i: i.delivery_reported(by=me))
if item.billing_status == ItemBillingStatus.PENDING.value and seams.raise_for_acceptance(item, me).done:
_mark(item_id, lambda i: i.billing_requested(by=me))
def _handover_view(item_id: str) -> dict:
with repository() as repo:
item, v = repo.items.get(item_id)
return {"billing_status": item.billing_status, "delivery_status": item.delivery_status, "version": v}
def _may_accept_own(caller: dict) -> bool:
"""Whoever submitted the work records its acceptance only if their role holds pj:item:accept_own: an exact grant
(a machine-style permission: no wildcard reaches it, and no default role has it)."""
return auth.has_permission(caller, ACCEPT_OWN)
def _accepting(item_id: str, caller: dict, version: Optional[int], act) -> dict:
out = _step(item_id, caller, version, act, who=LEAD, what="record the client's acceptance")
_handover(item_id, caller)
out.update(_handover_view(item_id))
return out
@PJ.use_case(kind="command", route="POST /items/{item_id}/accept", permission="pj:item:accept", idempotent=True,
if_match="version")
def accept(caller: dict, item_id: Key, evidence: Optional[Note] = None, evidence_kind: Optional[Name] = None,
version: Optional[int] = None) -> dict:
"""Record the client's acceptance (PJ-04). Refused without evidence: "Attach the evidence first: a link, a
file or a sign-off". Records who, when and how; a line billed on acceptance asks invoicing for its draft,
once (keyed acc:<item_id>)."""
def act(item, settings, now, repo, project):
if evidence:
item.attach_evidence(evidence, evidence_kind or "link", settings, by=_who(caller))
return item.accept(now=now, settings=settings, by=_who(caller), may_accept_own=_may_accept_own(caller),
identities=_ids(caller))
return _accepting(item_id, caller, version, act)
@PJ.use_case(kind="command", route="POST /items/{item_id}/accept-deemed", permission="pj:item:accept",
idempotent=True, if_match="version")
def accept_deemed(caller: dict, item_id: Key, version: Optional[int] = None) -> dict:
"""Record deemed acceptance (PJ-03): only where this business allows it, after the warning was recorded and its
notice has ended, and with evidence. Recorded as deemed, never as the client's yes; billed as accept is."""
return _accepting(item_id, caller, version, lambda i, s, now, repo, p: i.accept_deemed(
now=now, cal=calendar_of(s), settings=s, by=_who(caller), may_accept_own=_may_accept_own(caller),
identities=_ids(caller)))
@PJ.use_case(kind="command", route="POST /items/{item_id}/retry-handover", permission="pj:item:accept",
idempotent=True)
def retry_handover(caller: dict, item_id: Key) -> dict:
"""For accepted work still pending with a neighbour: ask invoicing again for its draft and tell the order
again (they were down or not installed). Each happens once however often it is asked (acc:<item_id>; the
order's delivery is keyed by the item). Nothing to do for work not yet accepted."""
with repository() as repo:
item, _ = _item(repo, caller, item_id)
project = repo.projects.get(item.project_id)
_must_manage(caller, project[0] if project else None, "retry the handover")
_handover(item_id, caller)
return {"item_id": item_id, **_handover_view(item_id)}
# ── the project: makegoods, reconciliation, cancellation ──────────────────────────────────────
@PJ.use_case(kind="command", route="POST /items/{item_id}/makegood", permission="pj:item:write")
def raise_makegood(caller: dict, item_id: Key, due_on: Optional[IsoDate] = None, owner: Optional[Email] = None,
title: Optional[Name] = None) -> dict:
"""Raise a makegood for an item the closed project left short (PJ-07): a new item with no sold line, never
billed; the short item stays short. A second makegood for the same item is refused."""
with repository() as repo, _decide():
short, _ = _item(repo, caller, item_id)
project = repo.projects.get(short.project_id)
_must_manage(caller, project[0] if project else None, "raise a makegood")
mg_id = f"{short.project_id}|mg|{short.item_id}"
if repo.items.get(mg_id) is not None:
raise RuleBroken(f"item {short.item_id} is already being made good ({mg_id})")
made = Item.makegood(short, project_closed=bool(project and project[0].state == "closed"),
by=_who(caller), title=title or "", due_on=_date(due_on, "due_on"),
owner=(owner or "").strip().lower())
v = repo.items.save(made, 0)
return _item_view(made, v, repo.settings.current(), _now(), project[0] if project else None)
def _reconcile(repo, project_id: str, caller: dict) -> tuple:
project, current = repo.projects.get(project_id)
items = [i for i, _ in repo.items.of_project(project_id)]
if project.reconcile(items, now=_now(), by=_who(caller)) is not None:
current = repo.projects.save(project, current)
return project, current
@PJ.use_case(kind="command", route="POST /projects/{project_id}/reconcile", permission=RECONCILE,
idempotent=True, if_match="version")
def reconcile(caller: dict, project_id: Key, version: Optional[int] = None) -> dict:
"""Close an active project and store sold against delivered per sold line (PJ-12): delivered, short, over and
descoped. Once: a closed project is left as it is."""
with repository() as repo, _decide():
project, current = _project(repo, caller, project_id)
_must_close(caller)
_same_version(f"project {project_id}", current, version)
return _project_view(repo, caller, *_reconcile(repo, project_id, caller))
@PJ.use_case(kind="command", route="POST /projects/reconcile-due", permission=RECONCILE, idempotent=True,
tool=False)
def reconcile_due(caller: dict) -> dict:
"""The daily pass (PJ-12): reconcile every active project whose window ended before today. Running it twice
reconciles each once. The scheduler's to run (pj:project:reconcile): it closes everyone's projects."""
_must_close(caller)
today = _now().date()
with repository() as repo:
due = repo.projects.due_for_reconciliation(today)
done = []
for pid in due:
with repository() as repo, _decide():
project, _ = _reconcile(repo, pid, caller)
done.append({"project_id": pid, "short": sum(x["short"] for x in project.reconciliation["lines"])})
return {"reconciled": done}
@PJ.use_case(kind="command", route="POST /projects/{project_id}/cancel", permission="pj:project:write",
idempotent=True, if_match="version")
def cancel_project(caller: dict, project_id: Key, reason: Reason, version: Optional[int] = None) -> dict:
"""Cancel a planned or active project, with the reason. Its items keep their states and history."""
with repository() as repo, _decide():
project, current = _project(repo, caller, project_id)
_must_manage(caller, project, "cancel this project")
_same_version(f"project {project_id}", current, version)
if project.cancel(reason, by=_who(caller)) is not None:
current = repo.projects.save(project, current)
return _project_view(repo, caller, project, current)
# ── AI steps: suggest only, nothing changes until a person acts ───────────────────────────────
@PJ.use_case(kind="query", route="GET /projects/{project_id}/ai/draft-plan", permission="pj:project:write")
def ai_draft_plan(caller: dict, project_id: Key) -> dict:
"""AI suggests dates and owners for the project's open items (suggest only: nothing changes until the lead
assigns them). With no AI provider connected it says so and gives the fallback."""
with repository() as repo:
project, _ = _project(repo, caller, project_id)
_must_manage(caller, project, "plan this project's work")
settings = repo.settings.current()
items = [i.item_id for i, _ in repo.items.of_project(project_id) if i.is_open]
return ai_steps.suggest("ai_draft_plan", settings.ai_mode("ai_draft_plan"),
{"project_id": project_id, "open_items": items})
@PJ.use_case(kind="query", route="GET /items/{item_id}/ai/chase-brief", permission="pj:item:write")
def ai_chase_brief(caller: dict, item_id: Key) -> dict:
"""AI drafts a reminder for a brief the client is late with — a draft for a person to edit and send, only
once the item has waited the chase days. The system never sends it."""
with repository() as repo, _decide():
item, _ = _item(repo, caller, item_id)
settings, today = repo.settings.current(), _now().date()
if item.state != "blocked":
raise RuleBroken(f"a {item.state} item is not waiting on the client")
waited = (today - item.blocked_since).days if item.blocked_since else 0
if waited < settings.chase_days:
raise RuleBroken(f"it has waited {waited} day(s); the reminder is due after {settings.chase_days}")
return ai_steps.suggest("ai_chase_brief", settings.ai_mode("ai_chase_brief"),
{"item_id": item_id, "waiting_on": item.blocked_reason, "days": waited})
@PJ.use_case(kind="query", route="GET /projects/{project_id}/ai/flag-shortfall", permission="pj:project:read")
def ai_flag_shortfall(caller: dict, project_id: Key) -> dict:
"""AI flags the open items at risk of not being done before the project's window closes, while they can still
be delivered. It only warns."""
with repository() as repo, _decide():
project, _ = _project(repo, caller, project_id)
if project.state != "active":
raise RuleBroken(f"a {project.state} project has no window left to protect")
settings = repo.settings.current()
return ai_steps.suggest("ai_flag_shortfall", settings.ai_mode("ai_flag_shortfall"),
{"project_id": project_id, "ends_on": project.ends_on})
# ── the client's view (PJ-05): a read-only link per project and person ────────────────────────
@PJ.use_case(kind="command", route="POST /projects/{project_id}/client-link", permission="pj:project:write")
def issue_client_link(caller: dict, project_id: Key, email: Email) -> dict:
"""A read-only link for one client contact to follow their project without signing in (PJ-05). The url is
returned ONCE — send it yourself; a second link for the same person replaces the first."""
with repository() as repo:
project, _ = _project(repo, caller, project_id)
_must_manage(caller, project, "give a client a link")
if project.state == "cancelled":
raise Refused(409, "the project is cancelled: there is nothing for the client to follow")
issued = PJ.links.issue("project", project_id, email, "READ", expires_in=timedelta(days=180), actor=_who(caller))
return {"link_id": issued.link_id, "url": issued.url, "recipient": email.strip().lower()}
@PJ.use_case(kind="query", route="GET /projects/{project_id}/client-links", permission="pj:project:read")
def list_client_links(caller: dict, project_id: Key) -> dict:
"""The client links to a project, per person: when issued, whether seen, and whether still working. Never the
url itself — it was handed over once."""
with repository() as repo:
project, _ = _project(repo, caller, project_id)
_must_manage(caller, project, "see the client links")
return {"links": [{k: x.get(k) for k in ("link_id", "recipient", "state", "issued_at", "seen")}
for x in PJ.links.status("project", project_id)]}
@PJ.use_case(kind="command", route="POST /projects/{project_id}/client-links/{link_id}/revoke",
permission="pj:project:write", idempotent=True)
def revoke_client_link(caller: dict, project_id: Key, link_id: text(64), reason: Optional[Reason] = None) -> dict:
"""Stop a client link working now (a contact left, a link was forwarded). Its page becomes the dead-link page;
issue a new one for whoever should still follow the project."""
with repository() as repo:
project, _ = _project(repo, caller, project_id)
_must_manage(caller, project, "revoke a client link")
if link_id not in {x["link_id"] for x in PJ.links.status("project", project_id)}:
raise Refused(404, f"no client link {link_id} for this project")
revoked = PJ.links.revoke(link_id, actor=_who(caller), reason=reason or "revoked by staff")
return {"link_id": link_id, "revoked": revoked}
@PJ.public("READ", subject_kind="project")
def client_status(link, req) -> public.Page:
"""The client's page: their project's items, states and evidence. Nothing to press; another project is the
same not-found page as a dead link."""
caller = public.public_actor(link)
with repository() as repo:
try:
view = _project_view(repo, caller, *_project(repo, caller, link.subject_id))
except Refused:
return public.Page("Not found", "<p>This link no longer works.</p>", status=404)
rows = "".join(
f"<tr><td>{public.esc(i['title'])}</td><td>{public.esc(i['state_name'])}</td>"
f"<td>{public.esc(str(i['due_on'] or ''))}</td>"
f"<td>{public.esc(i['evidence']) if i['state'] in ('in_review', 'done') else ''}</td></tr>"
for i in view["items"])
body = (f"<h1>Your project</h1><p>{len(view['items'])} item(s), "
f"{sum(1 for i in view['items'] if i['state'] == 'done')} accepted.</p>"
"<table><tr><th>Item</th><th>Status</th><th>Due</th><th>Evidence</th></tr>"
f"{rows}</table>")
return public.Page("Your project", body)
# ── settings (the manager) ────────────────────────────────────────────────────────────────────
@PJ.use_case(kind="command", route="POST /states/{state_id}", permission="pj:settings:write", idempotent=True)
def rename_state(caller: dict, state_id: Name, name: Optional[Name] = None, position: Optional[int] = None) -> dict:
"""Rename an item state as this business says it ('In review' → 'With client') or move it on the board. The
states and what moves between them are fixed by the signed process."""
with repository() as repo, _decide():
state = repo.states.get(state_id)
if state is None:
raise Refused(404, f"no state {state_id}")
if state.rename(name, position):
repo.states.save(state)
return {"state_id": state.state_id, "name": state.name, "position": state.position}
@PJ.use_case(kind="command", route="POST /settings/{setting_key}", permission="pj:settings:write", idempotent=True)
def set_setting(caller: dict, setting_key: Name, value: Any = None) -> dict:
"""Change one setting (see list_settings for each one's meaning); null puts the reference behaviour back.
The value is checked: a wrong one is refused in words and nothing changes."""
with repository() as repo, _decide():
checked = check_setting(setting_key, value)
stored = list(checked) if isinstance(checked, tuple) else checked
repo.settings.set(setting_key, stored)
return {"setting_key": setting_key, "value": stored, "warning": setting_warning(setting_key, checked)}