dodil/erp-brick-expenses
DODIL ERP brick: expenses
src/brick_expenses/application/use_cases.py
"""The use cases - what a person (or their agent) can do with an expense claim.
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
(a PERMISSION, never a role's name) and WHICH ROWS (an employee sees their own claims; a manager the claims waiting for them;
finance and the owner every claim). A row the caller may not see is a 404, never a 403.
The rules that shape every use case (module.yaml in_code_not_rows; tests/test_stories.py has one test per story):
- 'today' is the date in business_timezone (SettingsRepository.timezone), never the UTC date
- nobody approves their own claim, under either login (email OR sub), at any level; the chain is fixed at submission
- one claim key per money or person operation (exp_claim): the loser of a race is a 409, never a second effect
- a neighbour is called only AFTER the local commit, from a pending exp_outbox row written in the transaction that made the
state change; a crash after commit is sent on the next drain, a rolled-back approval sends nothing
- state only moves forward; a returned claim continues as revision n+1 and the approvals of revision n never count for it
- no read-your-writes inside an open transaction (DataK3): every use case reads committed rows, writes, and reports from what
it holds; the approval commits the level, THEN a second transaction completes the chain, THEN a third posts
"""
import base64
import json
import re
from contextlib import contextmanager
from datetime import date, datetime, timedelta
from decimal import Decimal
from typing import Annotated, Literal, Optional
from zoneinfo import ZoneInfo
from erp_platform import auth, work
from erp_platform.bricks import Brick, Refused
from erp_platform.types import BigId, text
from pydantic import Field
from sqlalchemy.exc import IntegrityError
from brick_expenses.application.inputs import (
AmountText,
Attendees,
City,
Code,
Country,
Currency,
Day,
Fields,
Key,
MealProvided,
Merchant,
Miles,
Moment,
Money,
Money0,
Nights,
Note,
Percentage,
RateAmount,
Reason,
ReceiptContents,
Ref,
Title,
YearStart,
day,
parse_money,
)
from brick_expenses.domain import ai, vat
from brick_expenses.domain import cards as cd
from brick_expenses.domain import policy as pol
from brick_expenses.domain import rates as rt
from brick_expenses.domain import rebill as rb
from brick_expenses.domain import receipts as rc
from brick_expenses.domain import reminders as rem
from brick_expenses.domain import report as rp
from brick_expenses.domain.errors import RuleBroken
from brick_expenses.domain.money import ZERO, money
from brick_expenses.domain.values import CONTROL_SETTINGS, SETTING_DEFS, read_setting
from brick_expenses.infrastructure import objects, readers, seams
from brick_expenses.infrastructure.tables import AlreadyDone
# Every decision is a PERMISSION (module.yaml permissions.catalog); no role's name is ever compared.
READ_OWN, WRITE_OWN, SUBMIT = "exp:report:read_own", "exp:report:write_own", "exp:report:submit"
READ_TEAM, READ_ALL = "exp:report:read_team", "exp:report:read_all"
APPROVE, FINANCE_APPROVE, POST = "exp:report:approve", "exp:report:finance_approve", "exp:report:post"
POLICY_WRITE, DELEGATION_WRITE, CONTROLS = "exp:policy:write", "exp:delegation:write", "exp:settings:controls"
RATE_WRITE, CATEGORY_WRITE = "exp:rate:write", "exp:category:write"
CARD_WRITE, CARD_FEED_IMPORT, CARD_FEED_INGEST = "exp:card:write", "exp:card_feed:import", "exp:card_feed:ingest"
CARD_MATCH_OWN, CARD_RESOLVE = "exp:card_txn:match_own", "exp:card_txn:resolve"
# machine: held only by a service client, and by an EXACT grant - no wildcard reaches them
MARK_PAID, LEAVER_INTAKE, POST_AUTO = "exp:report:mark_paid", "exp:leaver:intake", "exp:journal:post_auto"
MACHINE = (MARK_PAID, LEAVER_INTAKE, POST_AUTO, "exp:ai:read_receipt", "exp:report:derive", "exp:payout:request",
"exp:card_feed:ingest", "exp:number_series:claim", "exp:migration:load", "exp:migration:reconcile")
EXP = Brick("expenses", version="0.2.0", ids_as_strings=True, machine_permissions=MACHINE, requires=("mail.send",), instructions=(
"Expenses: an employee claims what they spent for work. Start a claim (create_report) and add each cost as a line: a receipt (add_line, or add_receipt "
"with the photo), mileage (add_mileage) or a daily allowance for a trip (add_per_diem); a receipt in another currency is converted at gl's rate. Company-card "
"transactions wait in list_card_txns: match each to a line (match_card_txn) or mark it personal (mark_personal). submit_report names every gate not met - "
"unconfirmed suggested fields, a missing receipt, a policy block, an unresolved card transaction, a stale line - and fixes the approval chain read from hr on "
"the submission date: the claimant's manager (or whoever they delegated to), then finance above the threshold; nobody approves their own claim under either "
"login. A manager or finance approves one level at a time (approve_level), or sends it back with a question (return_report, which continues the claim as a new "
"revision) or rejects it with a reason. The last approval posts ONE journal (cost, reclaimable VAT where tax says the receipt is a valid VAT invoice, what the "
"employee is owed, what the card paid) and asks payments to pay the employee, less personal card spend netted; the claim is paid when the bank says so. "
"Rates, categories, policy rules and settings are dated rows finance and the owner add. Amounts are exact strings in the claim's currency."))
# The process view (BRICKS.md): the workflows as BPMN, GET /process/bpmn.
EXP.declare_process("process.yaml")
FINAL_STATES = rp.TERMINAL
@contextmanager
def _repository_default():
from erp_platform import db # imported late: it needs BUCKET
with _on_engine(db.engine) as repo:
yield repo
def _lost_a_unique_race(exc: IntegrityError) -> bool:
text = f"{getattr(exc, 'orig', '')} {exc}"
return "23505" in text or "duplicate key" in text or "UNIQUE constraint" in text
@contextmanager
def _on_engine(engine):
"""One transaction. On DataK3 a racer's duplicate key (two approvals of one level, two claims of one number) can surface at
COMMIT, after the use case has returned - so it is caught here, around the commit, and answered as the retry-safe 409 a
conflict seen mid-use-case gets, never a 500."""
from brick_expenses.infrastructure.tables import Repository
try:
with engine.begin() as conn:
yield Repository(conn)
except IntegrityError as exc:
if not _lost_a_unique_race(exc):
raise
raise Refused(409, "Someone else did the same a moment ago - load the claim again and retry") from None
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
except seams.Unavailable as down:
raise Refused(down.http, down.reason) from None
except objects.StoreDown as down:
raise Refused(down.http, down.reason) from None
except IntegrityError:
raise Refused(409, "Someone else did the same a moment ago - load the claim again and retry") from None
def _who(caller: dict) -> str:
return (caller.get("email") or caller.get("sub") or "").strip().lower()
def _me(caller: dict) -> rp.Identity:
"""Both of the caller's names: the same person under either is the same person."""
return rp.Identity((caller.get("email") or "").strip().lower(), (caller.get("sub") or "").strip())
def _now() -> datetime:
return work.now()
def _utc_naive(dt: datetime) -> datetime:
from datetime import timezone
return dt.astimezone(timezone.utc).replace(tzinfo=None) if dt.tzinfo else dt
def _today(repo) -> date:
"""Business today: the date NOW (work.now()) in business_timezone - read from ONE place - never the UTC date."""
return _now().astimezone(ZoneInfo(repo.settings.timezone())).date()
def _s(v) -> Optional[str]:
return None if v is None else f"{Decimal(v):.2f}"
def _need(caller: dict, perm: str) -> None:
if not auth.has_permission(caller, perm):
raise Refused(403, f"missing permission: {perm}")
def _bounds(reason: str, fix: str = "check the value and send it again") -> Refused:
return Refused(422, reason, gate="bounds", fix=fix)
def _cur(v: str) -> str:
"""An ISO 4217 alphabetic code (GBP, EUR), upper-cased."""
v = (v or "").strip().upper()
if len(v) != 3 or not v.isalpha() or not v.isascii():
raise Refused(422, f"{v or 'a blank currency'!r} is not a currency code: use three letters such as GBP")
return v
def _date(v: Optional[str], what: str) -> date:
try:
d = day(v)
except ValueError:
raise _bounds(f"{what} is a date like 2026-11-01") from None
if not date(2000, 1, 1) <= d <= date(2100, 1, 1):
raise _bounds(f"{what} {d} is outside 2000-01-01 .. 2100-01-01")
return d
# ── visibility: which rows ────────────────────────────────────────────────────────────────────
def _can_see(caller: dict, r: rp.Report, delegators: tuple = ()) -> bool:
me = _me(caller)
if auth.has_permission(caller, READ_ALL):
return True
if auth.has_permission(caller, READ_OWN) and me.same_as(r.claimant):
return True
for lv in r.chain: # the people the claim is routed to see it, and whoever covers for them (EXP-09)
if lv.kind == "manager" and auth.has_permission(caller, READ_TEAM):
who = rp.Identity(lv.email, lv.sub)
if me.same_as(who) or any(who.same_as(d) for d in delegators):
return True
if lv.kind == "finance" and auth.has_permission(caller, FINANCE_APPROVE):
return True
return False
def _delegators(repo, caller: dict) -> tuple:
"""Whose manager level the caller covers today: the identities of every delegator with a delegation naming the caller, in force."""
if not auth.has_permission(caller, READ_TEAM):
return ()
today = _today(repo)
if repo.settings.on("delegation_allowed", today).strip().lower() not in ("true", "1", "yes"):
return ()
rows = repo.delegations.active_for(_me(caller), today)
return tuple(rp.Identity(r["delegator_email"] or "", r["delegator_sub"] or "") for r in rows)
def _visible(repo, caller: dict, report_id: str) -> rp.Report:
r = repo.reports.get(report_id)
if r is not None and not _can_see(caller, r) and any(lv.kind == "manager" for lv in r.chain):
if _can_see(caller, r, _delegators(repo, caller)):
return r
if r is None or not _can_see(caller, r):
raise Refused(404, f"no expense report {report_id}")
return r
def _own(repo, caller: dict, report_id: str) -> rp.Report:
"""A claim of the caller's own (and only theirs): another person's, even finance's, is a 404."""
r = repo.reports.get(report_id)
if r is None or not _me(caller).same_as(r.claimant):
raise Refused(404, f"no expense report {report_id}")
return r
# ── views ───────────────────────────────────────────────────────────────────────────────────
def _line_view(ln: rp.Line, reveal: bool = False) -> dict:
return {"line_no": ln.line_no, "kind": ln.kind, "category_code": ln.category_code, "expense_date": ln.expense_date.isoformat(),
"merchant": ln.merchant, "amount": _s(ln.amount), "vat_amount": _s(ln.vat_amount), "paid_with": ln.paid_with,
"cost_centre": ln.cost_centre or None, "receipt_key": ln.receipt_key or None, "description": ln.description or None,
"city": ln.city or None, "nights": ln.nights, "attendees": ln.attendees, "justification": ln.justification or None,
"policy": ln.policy_results, "receipt_amount": _s(ln.receipt_amount), "receipt_currency": ln.receipt_currency or None,
"fx_rate": None if ln.fx_rate is None else f"{ln.fx_rate:.6f}", "treatment_code": ln.treatment_code or None,
"tax": {"tax_code": ln.tax_code or None, "vat_reclaimable": _s(ln.vat_reclaimable), "vat_non_reclaimable": _s(ln.vat_non_reclaimable),
"checked": bool(ln.tax.get("checked")), "gates": ln.tax.get("gates", [])} if ln.tax else None,
"ai": {"model": ln.ai.get("model"), "filled": ln.ai.get("filled", []), "changed": ln.ai.get("changed", []), "suggested": ln.ai.get("suggested", {})}
if ln.ai else None, "possible_duplicate": bool(ln.possible_duplicate_of),
"possible_duplicate_of": (ln.possible_duplicate_of or None) if reveal else None, "unconfirmed_fields": ln.unconfirmed, "stale": ln.stale, "stale_reason": ln.stale_reason or None,
"card_txn_id": ln.card_txn_id or None,
"miles_or_km": None if ln.miles is None else f"{ln.miles:.4f}", "trip_from": ln.trip_from.isoformat(timespec="minutes") if ln.trip_from else None,
"trip_to": ln.trip_to.isoformat(timespec="minutes") if ln.trip_to else None, "meals_provided": ln.meals or None, "calc": ln.calc or None,
"recomputed": ln.recomputed or None,
"rebill": {"bp_id": str(ln.rebill_bp_id), "markup_pct": f"{ln.rebill_markup_pct:.4f}", "tax_category": ln.rebill_tax_category or "standard"} if ln.rebill_bp_id else None}
def _policy_of(repo, r: rp.Report, lines: list) -> dict:
"""line_no -> the results of the rules in force on that line's date (ONE query for the company's rules, however many lines)."""
if not lines:
return {}
rules = repo.policy.rows(r.company_code, max(ln.expense_date for ln in lines))
return {ln.line_no: pol.evaluate(ln, pol.in_force(rules, ln.expense_date), r.currency) for ln in lines}
def _unresolved_cards(repo, r: rp.Report, today: date) -> list:
"""What stops card_resolved: the claimant's card transactions of the claim's period (first receipt date .. today) that are neither matched
nor marked personal (ONE query), and a card line with no transaction matched."""
lo = min((ln.expense_date for ln in r.lines), default=today)
waiting = [t["card_txn_id"] for t in repo.cards.of_holder(r.person_id, status="imported", since=lo, until=today)]
return waiting + [f"line {ln.line_no} (paid with the card, no transaction matched)" for ln in r.lines if ln.paid_with == "card" and not ln.card_txn_id]
def _gates(repo, r: rp.Report, today: date, leaver_until: Optional[date], employed: bool) -> list:
codes = {(ln.category_code, ln.expense_date) for ln in r.lines}
cats = repo.categories.in_force(r.company_code, codes)
required = lambda code: any(c["receipt_required"] for (cc, _), c in cats.items() if cc == code) # noqa: E731
return rp.readiness(r, Decimal(repo.settings.on("receipt_required_above", today)), required, rp.may_claim(employed, leaver_until, today),
_unresolved_cards(repo, r, today))
def _view(repo, r: rp.Report, readiness: Optional[list] = None, reveal: bool = False) -> dict:
"""The claim whole. `reveal`: the viewer is not the claimant, so a line's possible duplicate names the other claim (the claimant is only told
there is one: another person's claim is not theirs to read)."""
out = {"report_id": r.report_id, "person_id": str(r.person_id), "company_code": r.company_code, "revision": r.revision,
"revisions": [{"revision": n, "ended": d} for n, d in repo.reports.revisions(r.report_id) if d in ("returned", "rejected")],
"status": r.status, "title": r.title, "currency": r.currency, "total_amount": _s(r.total()),
"reimbursable_amount": _s(r.reimbursable()), "netted_amount": _s(r.netted_amount) if r.netted_amount else None,
"payout_amount": _s(r.payout_amount()) if r.status in ("posted", "paid") else None, "lines": [_line_view(ln, reveal) for ln in r.lines],
"chain": [{"level": lv.level, "kind": lv.kind, "approver_person_id": None if lv.person_id is None else str(lv.person_id),
"rerouted_from": lv.rerouted_from} for lv in r.chain],
"waiting_level": r.waiting_level, "decisions": [{"level": s.level, "kind": s.kind, "decision": s.decision, "decided_by": s.decided_by,
"reason": s.reason or None, "delegate_for": s.delegate_for or None} for s in r.steps],
"submitted_on": r.submitted_on.isoformat() if r.submitted_on else None,
"approved_on": r.approved_on.isoformat() if r.approved_on else None,
"journal_id": None if r.journal_id is None else str(r.journal_id), "payout_ref": r.payout_ref,
"paid_on": r.paid_on.isoformat() if r.paid_on else None, "fail_reason": r.fail_reason}
if readiness is not None:
out["readiness"] = readiness
box = repo.outbox.of_report(r.report_id) if r.status in ("approved", "post_failed", "posted") else []
if box:
out["outbox"] = [{"target": b["target"], "ref": b["ref"], "status": b["status"], "attempts": b["attempts"], "last_error": b["last_error"]} for b in box]
return out
def _summary(row: dict) -> dict:
return {"report_id": row["report_id"], "status": row["status"], "title": row["title"], "revision": row["revision"], "currency": row["currency"],
"company_code": row["company_code"], "total_amount": _s(row["total_amount"]), "waiting_level": row["waiting_level"]}
# ── reading ─────────────────────────────────────────────────────────────────────────────────
@EXP.use_case(kind="query", route="GET /reports")
def list_reports(caller: dict, scope: Literal["mine", "waiting", "all"] = "mine") -> dict:
"""The claims the caller may see. scope=mine: their own claims. scope=waiting: the claims waiting for THEM to approve (a manager's
team, finance's queue). scope=all: every claim (finance and the owner). Summaries only; get_report gives the lines."""
with repository() as repo:
me = _me(caller)
if scope == "all":
_need(caller, READ_ALL)
rows = repo.reports.summaries(everything=True)
elif scope == "waiting":
if not (auth.has_permission(caller, READ_TEAM) or auth.has_permission(caller, FINANCE_APPROVE)):
raise Refused(403, f"missing permission: {READ_TEAM} or {FINANCE_APPROVE}")
rows = repo.reports.summaries(waiting_for=[me, *_delegators(repo, caller)] if auth.has_permission(caller, READ_TEAM) else None,
finance_queue=auth.has_permission(caller, FINANCE_APPROVE))
else:
_need(caller, READ_OWN)
rows = repo.reports.summaries(mine=me)
return {"reports": [_summary(r) for r in rows]}
@EXP.use_case(kind="query", route="GET /reports/{report_id}")
def get_report(caller: dict, report_id: Ref) -> dict:
"""One claim, whole: its lines, the approval chain and who decided what, the journal and payout it led to, and (while it is a
draft) which gates submit_report would refuse on. A claim the caller may not see is a 404."""
with repository() as repo:
r = _visible(repo, caller, report_id)
gates = None
if r.status == "draft" and _me(caller).same_as(r.claimant):
today = _today(repo)
gates = _gates(repo, r, today, repo.leavers.claims_until(r.person_id), True)
return _view(repo, r, gates, reveal=not _me(caller).same_as(r.claimant))
# ── claiming ────────────────────────────────────────────────────────────────────────────────
@EXP.use_case(kind="command", route="POST /reports", permission=WRITE_OWN, status=201)
def create_report(caller: dict, currency: Currency, title: Optional[Title] = None) -> dict:
"""Start a claim for the caller (expense_report: none -> draft). The caller's person, company and payee are read from hr by their
sign-in; only a person employed today (or inside the leaver window) can claim. The claim gets the next number of its company's
year, UK01-EXP-2026-00042, and currency is the company's currency: every line is in it."""
cur = _cur(currency)
with repository() as repo, _decide():
today = _today(repo)
me = _me(caller)
person = seams.person_by_sub(me.sub, _who(caller)) if me.sub else None
if person is None:
raise Refused(409, "hr holds no person with your sign-in, so you cannot claim", gate="claimant_may_claim",
fix="ask HR to link your login to your person record")
pid = int(person["person_id"])
emp = seams.employment(pid, today, _who(caller))
ok, why = rp.may_claim(bool(emp.get("employed")), repo.leavers.claims_until(pid), today)
if not ok:
raise Refused(409, f"You cannot start a claim: {why}", gate="claimant_may_claim")
company = emp.get("company_code") or person.get("company_code")
if not company:
raise Refused(409, "hr gives no company for you on this date, so the claim cannot be numbered", gate="claimant_may_claim")
number = repo.claims.take("series", f"{company}-{today.year}", _now(), _who(caller))
r = rp.Report(f"{company}-EXP-{today.year}-{number:05d}", pid, company, emp.get("bp_id") or person.get("bp_id"), me, cur, title or "")
repo.reports.create(r, _now())
return _view(repo, r)
@EXP.use_case(kind="command", route="POST /reports/{report_id}/lines", permission=WRITE_OWN, status=201)
def add_line(caller: dict, report_id: Ref, category_code: Code, expense_date: Day, merchant: Merchant, amount: Money,
vat_amount: Money0 = Decimal("0.00"), currency: Optional[Currency] = None, receipt_key: Optional[Key] = None,
description: Optional[Note] = None, paid_with: Literal["own", "card"] = "own", city: Optional[City] = None,
nights: Optional[Nights] = None, attendees: Optional[Attendees] = None, justification: Optional[Note] = None,
receipt: Optional[ReceiptContents] = None, treatment_code: Optional[Code] = None) -> dict:
"""Add one cost to a draft claim: what it was for (a category), the date on the receipt, the merchant and the total as the receipt shows it
(in `currency`, the claim's own when omitted; another currency is converted at gl's spot rate on the receipt date, rounded once, and the rate
is kept on the line - a missing rate refuses the line naming the currency and date). `vat_amount` is the VAT the receipt shows, in the same
currency. `receipt` (supplier, address, VAT number, invoice number) lets tax check the receipt is a valid VAT invoice, which VAT reclaim
needs; `treatment_code` says how tax treats the merchant (uk_business unless the receipt shows otherwise). receipt_key is the DataK3 object key of the receipt image (raw/expenses/<report_id>/...); a line above
receipt_required_above needs one before the claim is submitted. The cost centre is the claimant's from hr. city, nights and attendees (the
people a meal was for) are what the expense policy reads; the policy's answer for the line comes back inline (line.policy) and a block rule
stops submission, a warn rule asks for a justification the approver sees."""
when = _date(expense_date, "expense_date")
with repository() as repo, _decide():
r = _own(repo, caller, report_id)
r.ensure_draft()
today = _today(repo)
cat = repo.categories.in_force(r.company_code, {(category_code, when)}).get((category_code, when))
if cat is None:
raise Refused(409, f"No expense category {category_code} is in force for {r.company_code} on {when}", gate="category_in_force",
fix="pick one of the company's categories")
if len(r.lines) >= rp.LINES_MAX:
raise Refused(409, f"A claim holds at most {rp.LINES_MAX} lines: submit this one and start another")
if receipt_key and not receipt_key.startswith(f"raw/expenses/{report_id}/"):
raise _bounds(f"receipt_key must be the receipt's object key, raw/expenses/{report_id}/<line>/<sha256>.<ext>")
cur = _cur(currency) if currency else r.currency
line = rp.Line(max((ln.line_no for ln in r.lines), default=0) + 1, category_code, when, merchant.strip(), amount, vat_amount, paid_with,
cost_centre=(seams.employment(r.person_id, today, _who(caller)).get("cost_centre") or ""), receipt_key=receipt_key or "",
description=(description or "").strip(), receipt_currency=cur, receipt_amount=amount, city=(city or "").strip(), nights=nights,
attendees=[a.strip() for a in (attendees or []) if a.strip()], justification=(justification or "").strip(),
treatment_code=(treatment_code or "").strip(), contents=receipt.model_dump(exclude_none=True) if receipt else {}, receipt_vat=vat_amount)
_price(r, line, _who(caller), refetch=True)
rp.check_line(line, today=today, currency=cur, report_currency=r.currency)
line.policy_results = _policy_of(repo, r, [line])[line.line_no]
repo.reports.add_lines(r.report_id, r.revision, [line], _now())
r.lines.append(line)
return _view(repo, r)
@EXP.use_case(kind="command", route="DELETE /reports/{report_id}/lines/{line_no}", permission=WRITE_OWN, destructive=True)
def remove_line(caller: dict, report_id: Ref, line_no: BigId) -> dict:
"""Remove one line from a DRAFT claim. A submitted claim's lines are frozen: a returned claim is edited as its next revision."""
with repository() as repo, _decide():
r = _own(repo, caller, report_id)
r.ensure_draft()
if all(ln.line_no != line_no for ln in r.lines):
raise Refused(404, f"report {report_id} has no line {line_no}")
matched = next(ln.card_txn_id for ln in r.lines if ln.line_no == line_no)
if matched:
raise Refused(409, f"line {line_no} is matched to card transaction {matched}: a match is final (the line is the card's receipt)",
fix="submit the claim with it")
repo.reports.remove_line(r.report_id, r.revision, int(line_no))
r.lines = [ln for ln in r.lines if ln.line_no != line_no]
return _view(repo, r)
def _price(r: rp.Report, ln: rp.Line, actor: str, *, refetch: bool) -> None:
"""The line's amount in the claim's currency from the receipt's: the same currency is the receipt amount; another is converted ONCE, half up,
at gl's spot rate on the receipt date (re-read when `refetch`, else the frozen rate stays). The VAT read off the receipt converts the same way."""
cur = ln.receipt_currency or r.currency
if cur == r.currency:
ln.fx_rate, ln.amount, ln.vat_amount = None, ln.receipt_amount, ln.receipt_vat or ZERO
return
if refetch or ln.fx_rate is None:
got = seams.fx_rate(cur, ln.expense_date, r.currency, actor)
if got is None:
raise Refused(409, f"gl holds no {cur} to {r.currency} rate on or before {ln.expense_date}, so a {cur} receipt cannot be converted",
gate="fx_rate", fix=f"ask finance to enter the {cur} rate in gl, then add the line")
ln.fx_rate = vat.rate(got.get("rate"))
ln.amount = vat.convert(ln.receipt_amount, ln.fx_rate)
ln.vat_amount = vat.convert(ln.receipt_vat or ZERO, ln.fx_rate)
if not (ZERO < ln.amount < rp.LINE_MAX):
raise _bounds(f"{ln.receipt_amount:.2f} {cur} at {ln.fx_rate} is {ln.amount:.2f} {r.currency}: a line is more than zero and below 10,000,000,000,000")
def _settle_tax(repo, r: rp.Report, today: date, actor: str, *, only_missing: bool = False) -> list:
"""Ask tax about each receipt line (and whether the receipt is a valid VAT invoice where VAT could be reclaimed), keep the answer as our copy,
and return the lines that changed. A line tax was already asked about with the same inputs is not asked again (a returned claim's unchanged
line); with no tax module nothing is reclaimed; a tax that fails is a retry (502/503), never a guess."""
cats = repo.categories.in_force(r.company_code, {(ln.category_code, ln.expense_date) for ln in r.lines})
changed = []
for ln in r.lines:
if ln.kind != "receipt":
continue
cat = cats[(ln.category_code, ln.expense_date)]
fx = ln.fx_rate or Decimal(1)
parts = {"ref": f"exp:{r.report_id}:{ln.line_no}", "on": ln.expense_date, "cur": ln.receipt_currency or r.currency, "amt": ln.receipt_amount,
"cat": cat["tax_category"], "treat": ln.treatment_code, "fx": fx, "contents": ln.contents, "vat": ln.receipt_vat, "tax": seams.tax_installed()}
d = vat.digest(parts)
if ln.tax.get("digest") == d or (only_missing and ln.tax.get("digest")):
continue
answer, check, gates = None, None, []
if seams.tax_installed():
ref = parts["ref"]
out = seams.compute_tax(vat.compute_body(ref, tax_point=ln.expense_date, currency=parts["cur"], treatment=ln.treatment_code,
description=ln.merchant, tax_category=cat["tax_category"], supply=vat.supply_of(cat["gl_purpose"]),
amount=ln.receipt_amount, fx=fx, company_currency=r.currency), actor)
if out["ok"]:
answer = (out["data"].get("lines") or [None])[0]
elif out["retry"]:
raise Refused(502, f"tax {out['reason']}: try again")
else:
gates = [{"gate": out["gate"] or "tax_refused", "met": False, "reason": out["reason"], "fix": out["fix"]}]
if answer is not None and Decimal(str(answer.get("vat_reclaimable") or 0)) > ZERO and ln.contents:
chk = seams.check_contents(ref, vat.contents_body(ln.contents, ln.expense_date), actor)
if chk["ok"]:
check = chk["data"]
elif chk["retry"]:
raise Refused(502, f"tax {chk['reason']}: try again")
else:
check = {"ready": False, "gates": [{"gate": chk["gate"] or "check_refused", "met": False, "reason": chk["reason"], "fix": chk["fix"]}]}
res = vat.settle(answer, check, fx, vat.convert(ln.receipt_vat or ZERO, fx), ln.amount)
ln.tax_code, ln.vat_amount = res["tax_code"] or "", res["vat"]
ln.vat_reclaimable, ln.vat_non_reclaimable = res["reclaimable"], res["non_reclaimable"]
ln.tax = {"digest": d, "gates": [*gates, *res["gates"]], "checked": check is not None}
changed.append(ln)
return changed
LINE_FIELDS = ("category_code", "expense_date", "merchant", "amount", "vat_amount", "description", "city", "nights", "attendees", "justification")
@EXP.use_case(kind="command", route="POST /reports/{report_id}/lines/{line_no}", permission=WRITE_OWN)
def edit_line(caller: dict, report_id: Ref, line_no: BigId, category_code: Optional[Code] = None, expense_date: Optional[Day] = None,
merchant: Optional[Merchant] = None, amount: Optional[Money] = None, vat_amount: Optional[Money0] = None,
description: Optional[Note] = None, city: Optional[City] = None, nights: Optional[Nights] = None,
attendees: Optional[Attendees] = None, justification: Optional[Note] = None, confirm_fields: Optional[Fields] = None,
receipt: Optional[ReceiptContents] = None, treatment_code: Optional[Code] = None) -> dict:
"""Correct or confirm fields of one line of a DRAFT claim: only what you send changes. amount and vat_amount are as the receipt shows them (in
the line's receipt currency; a foreign line is converted again). A field a receipt reader suggested stays marked
(line.unconfirmed_fields) until you correct it here or name it in confirm_fields, and submit_report refuses while any is marked
(gate lines_ready). The policy's answer for the line is evaluated again. A line matched to a card transaction keeps the card's billed
amount."""
when = _date(expense_date, "expense_date") if expense_date else None
with repository() as repo, _decide():
r = _own(repo, caller, report_id)
r.ensure_draft()
ln = next((x for x in r.lines if x.line_no == int(line_no)), None)
if ln is None:
raise Refused(404, f"report {report_id} has no line {line_no}")
if amount is not None and ln.card_txn_id:
raise Refused(409, f"line {ln.line_no} is matched to card transaction {ln.card_txn_id}: its amount is the card's billed amount")
if ln.kind != "receipt" and any(v is not None for v in (category_code, expense_date, merchant, amount, vat_amount)):
raise Refused(409, f"line {ln.line_no} is a {ln.kind.replace('_', ' ')} computed from the country's rates: remove it and add it again", gate="computed_line")
given = {"category_code": category_code, "expense_date": when, "merchant": merchant.strip() if merchant else None, "amount": amount,
"vat_amount": vat_amount, "treatment_code": treatment_code.strip() if treatment_code else None, "description": description.strip() if description is not None else None,
"city": city.strip() if city is not None else None, "nights": nights,
"attendees": [a.strip() for a in attendees if a.strip()] if attendees is not None else None,
"justification": justification.strip() if justification is not None else None}
before = list(ln.unconfirmed)
unknown = [f for f in (confirm_fields or []) if f not in ln.unconfirmed]
if unknown:
raise _bounds(f"line {ln.line_no} has no unconfirmed field {', '.join(unknown)}", f"unconfirmed: {', '.join(ln.unconfirmed) or 'none'}")
for name, value in given.items():
if value is not None and name not in ("amount", "vat_amount"):
setattr(ln, name, value)
if value is not None and name in ln.unconfirmed:
ln.unconfirmed = [f for f in ln.unconfirmed if f != name] # a corrected field is a confirmed one
ln.unconfirmed = [f for f in ln.unconfirmed if f not in (confirm_fields or [])]
if receipt is not None:
ln.contents = receipt.model_dump(exclude_none=True)
ln.unconfirmed = [f for f in ln.unconfirmed if f != "vat_number"] # the supplier's particulars were typed: that field is confirmed
if amount is not None:
ln.receipt_amount = amount
if vat_amount is not None:
ln.receipt_vat = vat_amount
today = _today(repo)
if ln.card_txn_id: # the card's billed amount stays; only the VAT is converted again
ln.vat_amount = vat.convert(ln.receipt_vat or ZERO, ln.fx_rate or Decimal(1))
else:
_price(r, ln, _who(caller), refetch=when is not None)
if category_code is not None or when is not None:
cat = repo.categories.in_force(r.company_code, {(ln.category_code, ln.expense_date)}).get((ln.category_code, ln.expense_date))
if cat is None:
raise Refused(409, f"No expense category {ln.category_code} is in force for {r.company_code} on {ln.expense_date}",
gate="category_in_force", fix="pick one of the company's categories")
rp.check_line(ln, today=today, currency=ln.receipt_currency or r.currency, report_currency=r.currency)
ln.policy_results = _policy_of(repo, r, [ln])[ln.line_no]
_keep_findings(repo, r, ln, [f for f in before if f not in ln.unconfirmed], _who(caller))
repo.reports.save_lines(r.report_id, r.revision, [ln], _now())
return _view(repo, r)
def _keep_findings(repo, r: rp.Report, ln: rp.Line, decided: list, who: str) -> None:
"""What the person did with the fields the model filled (EXP-01): each field decided now is kept as confirmed, with its final value and whether
the person changed what the model suggested; the line remembers which fields changed so the approver sees them."""
if not decided or not ln.ai.get("finding_no"):
return
final = {"category_code": ln.category_code, "expense_date": ln.expense_date.isoformat(), "merchant": ln.merchant,
"amount": f"{(ln.receipt_amount or ln.amount):.2f}", "vat_amount": f"{(ln.receipt_vat or ZERO):.2f}", "vat_number": ln.contents.get("supplier_vat_number", "")}
got = repo.ai.finding(r.report_id, ln.line_no, int(ln.ai["finding_no"]))
confirmed = json.loads(got["confirmed_json"] or "{}") if got else {}
for f in decided:
changed = final.get(f, "") != ln.ai.get("suggested", {}).get(f, "")
confirmed[f] = {"value": final.get(f, ""), "changed": changed}
if changed and f not in ln.ai.setdefault("changed", []):
ln.ai["changed"].append(f)
if got:
repo.ai.save_finding({**got, "confirmed_json": json.dumps(confirmed, sort_keys=True), "decided_by": who, "decided_at": _utc_naive(_now())}, _now())
# ── the receipt store (EXP-01, EXP-16) ────────────────────────────────────────────────────────────
@EXP.use_case(kind="command", route="POST /reports/{report_id}/receipts", permission=WRITE_OWN, status=201)
def add_receipt(caller: dict, report_id: Ref, content_type: Literal["image/jpeg", "image/png", "application/pdf"],
content_base64: text(4_300_000, min_length=1), line_no: Optional[BigId] = None) -> dict:
"""Store a receipt image (JPEG, PNG or PDF, at most 3 MiB, sent as base64) as a DataK3 object under raw/expenses/<report_id>/<line_no>/<sha256>.<ext>
and keep the evidence for the retention period. With line_no it is attached to that line of your draft; without, it is stored for the NEXT line
(add_line with this receipt_key) - or, if you agreed to the receipt reader (set_ai_consent) and the step is on, the model reads it and the line is
created with its suggestions: date, merchant, total, currency, VAT, the supplier's VAT number and the category, each MARKED until you confirm or
correct it (edit_line), and submit_report refuses while any is marked. Without your consent, with the step off, or when the model cannot answer, no
image goes to a model and you type the fields. The same image is the same object, stored once. If the same image was already attached to another
line - yours or anyone's - the line is marked possible_duplicate: nothing is refused by the mark alone, your approver sees it and decides."""
try:
body = rc.decode(content_type, content_base64)
except RuleBroken as bad:
raise Refused(422, str(bad), gate="bounds") from None
digest = rc.sha256(body)
who = _who(caller)
# 1. what the reader needs, read in a short transaction of its own (no model call while a transaction is open)
plan, why = None, ""
if line_no is None:
with repository() as repo, _decide():
r = _own(repo, caller, report_id)
r.ensure_draft()
today = _today(repo)
if readers.ports.reader is None:
why = "no receipt reader is configured: type the fields"
elif repo.settings.on("ai_read_receipt_mode", today) != "suggest":
why = "the receipt reader is off: type the fields"
elif not ai.consented(repo.ai.current_consent(_my_person(caller), ai.STEP)):
why = "you have not agreed to the receipt reader, or you withdrew: type the fields"
else:
plan = [{"category_code": c["category_code"], "name": c["name"]} for c in repo.categories.all_in_force(r.company_code, today) if c["kind"] == "receipt"]
# 2. the model is asked (an image leaves this brick only now, only with consent), and its answer is checked like anything a person typed
suggestion, model, raw = None, "", {}
if plan is not None:
try:
out = readers.ports.reader.read(body, content_type, plan)
raw = {"fields": out.get("fields"), "confidence": out.get("confidence")}
model = str(out.get("model") or "")
suggestion = ai.fields(out)
except readers.ReaderDown as down:
why = f"the receipt reader could not answer ({down.reason}): type the fields"
except RuleBroken as bad:
why = f"{bad}: type the fields"
# 3. the image is stored, the line created or the receipt attached, and the model's footprint kept - in one transaction
with repository() as repo, _decide():
r = _own(repo, caller, report_id)
r.ensure_draft()
line = next((x for x in r.lines if x.line_no == int(line_no)), None) if line_no is not None else None
if line_no is not None and line is None:
raise Refused(404, f"report {report_id} has no line {line_no}")
n = line.line_no if line else max((x.line_no for x in r.lines), default=0) + 1
today = _today(repo)
key = rc.object_key(r.report_id, n, digest, content_type)
objects.ports.receipts().put(key, body, content_type) # content-addressed and idempotent: an orphan after a rollback is the same bytes again
have = repo.receipts.get(key)
others = [x for x in repo.receipts.of_sha(digest) if x["object_key"] != key]
dup = f"{others[0]['report_id']}:{others[0]['line_no']}" if others else ""
created = None
if suggestion is not None and line is None:
created = _line_from_suggestion(repo, r, n, suggestion, model, key, dup, who, today)
if isinstance(created, str):
why, created = created, None
if have is None:
base = (created or line).expense_date if (created or line) else today
years = int(repo.settings.on("receipt_retention_years", today))
repo.receipts.add({"object_key": key, "report_id": r.report_id, "line_no": n, "sha256": digest, "content_type": content_type,
"size_bytes": len(body), "uploaded_by": who, "uploaded_at": _utc_naive(_now()),
"retain_until": rc.retain_until(base, years), "possible_duplicate_of": dup or None}, _now())
if line:
line.receipt_key, line.possible_duplicate_of = key, dup
repo.reports.save_lines(r.report_id, r.revision, [line], _now())
elif created is not None:
repo.reports.add_lines(r.report_id, r.revision, [created], _now())
repo.ai.add_finding({"report_id": r.report_id, "line_no": n, "finding_no": 1, "step_key": ai.STEP, "model": model, "inputs_digest": digest,
"fields_json": json.dumps(raw, sort_keys=True, default=str),
"confirmed_json": "{}", "decided_by": None, "decided_at": None}, _now())
out = {"receipt_key": key, "line_no": n, "sha256": digest, "size_bytes": len(body), "attached": line is not None or created is not None,
"possible_duplicate": bool(dup), "stored_again": have is not None,
"suggested": created is not None, "suggestion_note": None if created is not None else (why or None)}
if created is not None:
r.lines.append(created)
out["line"] = _line_view(created)
return out
def _line_from_suggestion(repo, r: rp.Report, n: int, sug: dict, model: str, key: str, dup: str, actor: str, today: date):
"""A new line from what the model read, every suggested field marked unconfirmed; or the reason there is no line (a string): a suggestion that a person
could not have typed - an unknown category, a future date, a currency gl has no rate for - is no suggestion."""
when = sug["expense_date"]
cat = repo.categories.in_force(r.company_code, {(sug["category_code"], when)}).get((sug["category_code"], when))
if cat is None or cat["kind"] != "receipt":
return f"the model suggested a category ({sug['category_code']}) the company has not: type the fields"
ln = rp.Line(n, sug["category_code"], when, sug["merchant"], sug["amount"], sug["vat_amount"], "own",
cost_centre=seams.employment(r.person_id, today, actor).get("cost_centre") or "", receipt_key=key, receipt_currency=sug["currency"],
receipt_amount=sug["amount"], receipt_vat=sug["vat_amount"], possible_duplicate_of=dup,
contents={"supplier_name": sug["merchant"], **({"supplier_vat_number": sug["vat_number"]} if sug["vat_number"] else {})})
try:
_price(r, ln, actor, refetch=True)
rp.check_line(ln, today=today, currency=sug["currency"], report_currency=r.currency)
except (Refused, RuleBroken) as bad:
return f"{getattr(bad, 'reason', None) or bad}: type the fields"
ln.policy_results = _policy_of(repo, r, [ln])[ln.line_no]
filled = [f for f in ai.SUGGESTED if f != "vat_number" or sug["vat_number"]]
ln.unconfirmed = list(filled)
ln.ai = {"finding_no": 1, "model": model, "filled": filled, "changed": [],
"suggested": {"category_code": sug["category_code"], "expense_date": when.isoformat(), "merchant": sug["merchant"], "amount": f"{sug['amount']:.2f}",
"vat_amount": f"{sug['vat_amount']:.2f}", "vat_number": sug["vat_number"], "currency": sug["currency"],
"category_reason": sug["category_reason"]}}
return ln
@EXP.use_case(kind="query", route="GET /reports/{report_id}/lines/{line_no}/receipt")
def get_receipt(caller: dict, report_id: Ref, line_no: BigId) -> dict:
"""The receipt image of a line, as base64 with its type, size, digest and the day it may be thrown away: what an approver looks at beside
the line. A claim the caller may not see is a 404."""
with repository() as repo, _decide():
r = _visible(repo, caller, report_id)
line = next((x for x in r.lines if x.line_no == int(line_no)), None)
if line is None or not line.receipt_key:
raise Refused(404, f"line {line_no} of {report_id} has no receipt")
row = repo.receipts.get(line.receipt_key)
if row is None:
raise Refused(404, f"{line.receipt_key} was not stored through add_receipt: there is no image here to show")
body = objects.ports.receipts().get(line.receipt_key)
if body is None:
raise Refused(404, f"the object {line.receipt_key} is not in the store", gate="object_missing")
return {"receipt_key": line.receipt_key, "content_type": row["content_type"], "size_bytes": row["size_bytes"], "sha256": row["sha256"],
"retain_until": row["retain_until"].isoformat() if row["retain_until"] else None,
"content_base64": base64.b64encode(body).decode()}
@EXP.use_case(kind="command", route="POST /reports/{report_id}/cancel", permission=WRITE_OWN)
def cancel_report(caller: dict, report_id: Ref) -> dict:
"""Cancel a draft claim (draft -> cancelled, terminal). A submitted claim is not cancelled: its approver returns or rejects it."""
with repository() as repo, _decide():
r = _own(repo, caller, report_id)
held = [ln.card_txn_id for ln in r.lines if ln.card_txn_id]
if held and r.status == "draft":
raise Refused(409, f"The claim holds card transactions ({', '.join(held)}): a match is final, so it cannot be cancelled", fix="submit it")
r.move("cancel_report")
repo.claims.once("close", f"{r.report_id}:r{r.revision}", _now(), _who(caller))
repo.reports.save(r, _now())
return _view(repo, r)
# ── submitting: the gates, then the chain from hr, fixed ─────────────────────────────────────────
@EXP.use_case(kind="command", route="POST /reports/{report_id}/submit", permission=SUBMIT)
def submit_report(caller: dict, report_id: Ref) -> dict:
"""Submit the caller's draft claim (draft -> submitted). Refused with every unmet gate named: no lines, a suggested field not confirmed
(lines_ready), a line above the receipt threshold with no receipt, a block policy rule broken (the rule's row is named), a card
transaction of the period neither matched nor marked personal, a stale line (recompute it), not employed. The approval chain is read from hr on TODAY's business date and fixed: the claimant's
manager (never the claimant: that level goes one up), then finance when the total is above finance_approval_threshold. The lines
are frozen from here."""
with repository() as repo, _decide():
r = _own(repo, caller, report_id)
r.move("submit_report")
today = _today(repo)
actor = _who(caller)
emp = seams.employment(r.person_id, today, actor)
fresh = _policy_of(repo, r, r.lines) # the rules in force TODAY decide, not the ones the line was added under
moved = []
for ln in r.lines:
if pol.changed(ln.policy_results, fresh[ln.line_no]):
ln.policy_results = fresh[ln.line_no]
moved.append(ln)
gates = _gates(repo, r, today, repo.leavers.claims_until(r.person_id), bool(emp.get("employed")))
unmet = [g for g in gates if not g["met"]]
if unmet:
raise Refused(409, "Cannot submit: " + "; ".join(f"{g['gate']}: {g['reason']}" for g in unmet), gate=unmet[0]["gate"],
fix=unmet[0]["fix"], fields={"gates": gates})
taxed = _settle_tax(repo, r, today, actor) # tax's answer for each receipt, kept as our copy (EXP-18)
total = r.total()
threshold = Decimal(repo.settings.on("finance_approval_threshold", today))
r.chain = rp.derive_chain(r.claimant, total, threshold, lambda n: seams.manager(r.person_id, today, n, actor))
repo.claims.once("submit", f"{r.report_id}:r{r.revision}", _now(), actor)
r.total_amount, r.reimbursable_amount = total, r.reimbursable()
r.submitted_on, r.submitted_at, r.waiting_level, r.waiting_since = today, _now(), r.chain[0].level, today
repo.reports.save_lines(r.report_id, r.revision, list({ln.line_no: ln for ln in [*moved, *taxed]}.values()), _now())
repo.reports.save(r, _now())
return _view(repo, r)
# ── the policy, delegates and the recompute (EXP-07, EXP-09, EXP-20, EXP-22, EXP-23) ───────────────────────────
def _mark_stale(repo, company_code: str, category_codes: list, since: date, reason: str, kinds: Optional[list] = None,
person_id: Optional[int] = None) -> int:
"""Mark every not-yet-approved line a changed rule can touch: dated on or after `since`, of those categories (or kinds). ONE read, ONE write."""
marked = []
for rid, rev, ln in repo.reports.open_lines(company_code=company_code, category_codes=category_codes or None, kinds=kinds, person_id=person_id):
if ln.expense_date >= since:
ln.stale, ln.stale_reason = True, reason
marked.append((rid, rev, ln))
repo.reports.save_marked(marked, _now())
return len(marked)
def _mark_mileage_stale(repo, r: rp.Report) -> None:
"""EXP-20: this claim's last level is approved, so its miles now count towards the tax year's bands: every OTHER not-yet-approved mileage line of
the same person, country, vehicle and tax year is marked stale (the approver recomputes it before deciding). Approved or posted lines never move."""
windows = {}
for ln in r.lines:
if ln.kind == "mileage" and ln.calc.get("tax_year"):
windows[(ln.rate_country, ln.rate_variant)] = tuple(day(x) for x in ln.calc["tax_year"])
if not windows:
return
marked = []
for rid, rev, other in repo.reports.open_lines(person_id=r.person_id, kinds=["mileage"], exclude_report=r.report_id):
span = windows.get((other.rate_country, other.rate_variant))
if span and span[0] <= other.expense_date <= span[1]:
other.stale, other.stale_reason = True, f"earlier mileage approved ({r.report_id})"
marked.append((rid, rev, other))
repo.reports.save_marked(marked, _now())
def _rule_view(r: pol.Rule) -> dict:
return {"company_code": r.company_code, "rule_code": r.rule_code, "valid_from": r.valid_from.isoformat(), "category_code": r.category_code,
"scope": r.scope, "city_or_country": r.city_or_country or None, "limit_amount": _s(r.limit_amount), "currency": r.currency or None,
"mode": r.mode, "requires": r.requires or None, "message": r.message or None}
@EXP.use_case(kind="command", route="POST /policy-rules", status=201)
def put_policy_rule(caller: dict, company_code: Code, rule_code: Code, category_code: Code, scope: Literal["per item", "per night", "per person", "per day"],
limit_amount: Money0, mode: Literal["warn", "block"], valid_from: Day, city_or_country: Optional[City] = None,
currency: Optional[Currency] = None, requires: Optional[Literal["attendees", "justification"]] = None,
message: Optional[Note] = None) -> dict:
"""Add a dated policy row (a limit or a ban on a category; limit 0.00 is a ban). A row is dated: the first row of a rule_code may start
today, a change to an existing one starts AFTER today, so a claim keeps the rule it was written under. Making a rule weaker - a block
turned into a warn, or a higher limit on a block - needs exp:settings:controls (the owner's), never finance alone. Not-yet-approved lines
the new row touches are marked stale (recompute_report)."""
if not (auth.has_permission(caller, POLICY_WRITE) or auth.has_permission(caller, CONTROLS)):
raise Refused(403, f"missing permission: {POLICY_WRITE}")
when = _date(valid_from, "valid_from")
with repository() as repo, _decide():
today = _today(repo)
cur = _cur(currency) if currency else ""
rule = pol.Rule(company_code.strip(), rule_code.strip(), when, category_code.strip(), scope, (city_or_country or "").strip(), limit_amount, cur, mode,
requires or "", (message or "").strip())
before = repo.policy.of_key(rule.company_code, rule.rule_code)
_dated(before, when, today, rule.rule_code)
old = pol.in_force(before, today)[:1]
_need(caller, CONTROLS if pol.loosens(old[0] if old else None, rule) else POLICY_WRITE)
if not repo.categories.in_force(rule.company_code, {(rule.category_code, max(when, today))}):
raise Refused(409, f"No expense category {rule.category_code} is in force for {rule.company_code}", gate="category_in_force")
repo.policy.put(rule, _who(caller), _now())
stale = _mark_stale(repo, rule.company_code, [rule.category_code], when, f"policy {rule.rule_code} from {when}") if when <= today else 0
return {**_rule_view(rule), "lines_marked_stale": stale}
@EXP.use_case(kind="query", route="GET /policy-rules")
def list_policy_rules(caller: dict, company_code: Code, on: Optional[Day] = None) -> dict:
"""The policy rows in force for a company on a date (default today): what a line is checked against, each with its row key."""
_need(caller, READ_OWN)
when = _date(on, "on") if on else None
with repository() as repo:
rows = pol.in_force(repo.policy.rows(company_code, when or _today(repo)), when or _today(repo))
return {"rules": [_rule_view(r) for r in rows]}
@EXP.use_case(kind="command", route="POST /delegations", permission=DELEGATION_WRITE, status=201)
def delegate_approval(caller: dict, delegate_email: text(254), valid_from: Day, valid_to: Day, delegate_sub: Optional[Ref] = None) -> dict:
"""Cover your approvals while you are away: the delegate may decide the manager level of the claims routed to you between two dates (at
most a year). A delegate is never the claimant of a claim they decide, and never decides two levels of one claim. The owner can switch
delegation off (delegation_allowed)."""
start, end = _date(valid_from, "valid_from"), _date(valid_to, "valid_to")
email = delegate_email.strip().lower()
me = _me(caller)
with repository() as repo, _decide():
today = _today(repo)
if repo.settings.on("delegation_allowed", today).strip().lower() not in ("true", "1", "yes"):
raise Refused(409, "Delegation is switched off for this company", gate="delegation_allowed", fix="the owner switches it on")
if start < today or end < start or (end - start).days > 366:
raise _bounds("a delegation starts today or later and lasts at most a year")
if rp.Identity(email, (delegate_sub or "").strip()).same_as(me):
raise Refused(409, "You cannot delegate to yourself")
row = {"delegator": _who(caller), "delegate": email, "valid_from": start, "delegator_email": me.email or None, "delegator_sub": me.sub or None,
"delegate_email": email, "delegate_sub": (delegate_sub or "").strip() or None, "valid_to": end, "set_by": _who(caller)}
repo.delegations.add(row, _now())
return {"delegator": row["delegator"], "delegate": email, "valid_from": start.isoformat(), "valid_to": end.isoformat()}
@EXP.use_case(kind="command", route="POST /delegations/end", permission=DELEGATION_WRITE, idempotent=True)
def end_delegation(caller: dict, delegate_email: text(254), valid_from: Day) -> dict:
"""End one of your delegations now: it stops covering you from yesterday, so nothing already decided changes."""
start = _date(valid_from, "valid_from")
with repository() as repo, _decide():
mine = [d for d in repo.delegations.of_delegator(_me(caller)) if d["delegate"] == delegate_email.strip().lower() and d["valid_from"] == start]
if not mine:
raise Refused(404, "no such delegation of yours")
row = {**mine[0], "valid_to": min(mine[0]["valid_to"] or date.max, _today(repo) - timedelta(days=1))}
repo.delegations.end(row, _now())
return {"delegate": row["delegate"], "valid_from": start.isoformat(), "valid_to": row["valid_to"].isoformat()}
@EXP.use_case(kind="query", route="GET /delegations")
def list_delegations(caller: dict) -> dict:
"""The delegations you gave and the ones that cover for you."""
_need(caller, DELEGATION_WRITE)
with repository() as repo:
me = _me(caller)
view = lambda d: {"delegator": d["delegator"], "delegate": d["delegate"], "valid_from": d["valid_from"].isoformat(), # noqa: E731
"valid_to": d["valid_to"].isoformat() if d["valid_to"] else None}
return {"given": [view(d) for d in repo.delegations.of_delegator(me)],
"covering_me": [view(d) for d in repo.delegations.active_for(me, _today(repo))]}
def _recompute(repo, r: rp.Report, today: date, actor: str = "") -> list:
"""Evaluate every not-yet-approved line again from the rows in force and clear its stale mark: [{line_no, field, was, now, cause}]."""
fresh = _policy_of(repo, r, r.lines)
changes, touched = [], []
for ln in r.lines:
cause = ln.stale_reason or "recomputed"
if r.status == "draft" and ln.fx_rate is not None and not ln.card_txn_id: # a draft's foreign line takes gl's rate as it is now
was = ln.amount
_price(r, ln, actor, refetch=True)
if ln.amount != was:
changes.append({"line_no": ln.line_no, "field": "amount", "was": _s(was), "now": _s(ln.amount), "cause": f"the {ln.receipt_currency} rate changed"})
touched.append(ln)
if pol.changed(ln.policy_results, fresh[ln.line_no]):
changes.append({"line_no": ln.line_no, "field": "policy", "was": [x["rule_code"] for x in ln.policy_results],
"now": [x["rule_code"] for x in fresh[ln.line_no]], "cause": cause})
ln.policy_results = fresh[ln.line_no]
if ln.kind in ("mileage", "per_diem") and r.status in ("draft", "submitted"):
moved = _recompute_computed(repo, r, ln, today)
if moved:
moved["cause"] = cause
changes.append(moved)
touched.append(ln)
for c in (c for c in changes if c["line_no"] == ln.line_no):
ln.recomputed = [*ln.recomputed, {**c, "on": today.isoformat()}] if c not in ln.recomputed else ln.recomputed
if ln.stale or any(c["line_no"] == ln.line_no for c in changes):
ln.stale, ln.stale_reason = False, ""
touched.append(ln)
repo.reports.save_lines(r.report_id, r.revision, touched, _now())
return changes
@EXP.use_case(kind="command", route="POST /reports/{report_id}/recompute", idempotent=True)
def recompute_report(caller: dict, report_id: Ref) -> dict:
"""The one click that clears stale lines: every line of a claim not yet approved is evaluated again from the rules in force, and the answer
lists each change with its cause. The claimant does it on their draft; on a submitted claim, the approver whose level waits does it before
deciding. An approved or posted line is never recomputed."""
with repository() as repo, _decide():
r = repo.reports.get(report_id)
if r is not None and r.status == "draft" and _me(caller).same_as(r.claimant):
_need(caller, WRITE_OWN)
else:
r, _, _ = _decider(repo, caller, report_id, "recompute")
changes = _recompute(repo, r, _today(repo), _who(caller))
if r.status == "submitted":
r.total_amount, r.reimbursable_amount = r.total(), r.reimbursable()
repo.reports.save(r, _now())
out = _view(repo, r)
out["changes"] = changes
return out
# ── deciding: one level at a time ────────────────────────────────────────────────────────────
def _decider(repo, caller: dict, report_id: str, what: str) -> tuple:
"""The report, the level the caller decides now, and whom they decide it FOR (a delegate; "" otherwise). Order: the row is theirs to see
(404), the permission of the level's kind (403), that they ARE the level's approver or hold their cover (403), then the rules (nobody
decides their own claim - not as a delegate either; nobody decides two levels) as 409."""
r = _visible(repo, caller, report_id)
lv = r.next_level() if r.status == "submitted" else None
cover = ""
if lv is not None:
_need(caller, APPROVE if lv.kind == "manager" else FINANCE_APPROVE)
if lv.kind == "manager":
me = _me(caller)
mine = (me.sub == lv.sub) if lv.sub else (me.email == lv.email.lower())
if not mine:
who = rp.Identity(lv.email, lv.sub)
if any(who.same_as(d) for d in _delegators(repo, caller)):
cover = lv.email or lv.sub
else:
raise Refused(403, f"You are not the approver of level {lv.level} of this claim", gate="approver",
fix="it waits for the claimant's manager, or whoever they delegated to")
rp.check_decider(r, _me(caller), lambda p: auth.has_permission(caller, p), what=what)
return r, lv, cover
@EXP.use_case(kind="command", route="POST /reports/{report_id}/approve")
def approve_level(caller: dict, report_id: Ref) -> dict:
"""Approve the level of the claim that waits for the caller: the manager's level needs exp:report:approve and is for the claimant's
manager alone, finance's needs exp:report:finance_approve. Never the claimant's own claim (matched on email OR sub), and finance
never decides a claim whose earlier level it decided. The level is recorded (committed), then the last level completes the claim and
posts ONE journal and asks payments to pay the employee - both after their own commits, from pending rows."""
with repository() as repo, _decide():
r, lv, cover = _decider(repo, caller, report_id, "approve")
me = _me(caller)
stale = [ln for ln in r.lines if ln.stale]
if stale:
raise Refused(409, "Cannot approve: lines computed from rules that changed: " + "; ".join(
f"{ln.line_no} ({ln.stale_reason or 'a rule changed'})" for ln in stale), gate="not_stale", fix="recompute the claim (recompute_report), then decide")
repo.claims.once("approval", f"{r.report_id}:r{r.revision}:{lv.level}", _now(), _who(caller))
step = rp.Step(lv.level, lv.kind, "approved", me.email, me.sub, _now(), approver_person_id=lv.person_id, delegate_for=cover)
repo.reports.add_step(r.report_id, r.revision, step, _now())
r.steps.append(step)
nxt = r.next_level()
r.waiting_level = nxt.level if nxt else None
r.waiting_since = _today(repo) if nxt else None # the next level starts waiting today (EXP-17 counts business days from it)
repo.reports.save(r, _now())
last = nxt is None
if last:
_complete(report_id, _who(caller))
_drain(report_id)
with repository() as repo:
return _view(repo, _visible(repo, caller, report_id))
@EXP.use_case(kind="command", route="POST /reports/{report_id}/return")
def return_report(caller: dict, report_id: Ref, reason: Reason) -> dict:
"""Send the claim back with a question (a reason is required). Revision n ends as returned; the claim continues as a DRAFT of
revision n+1 with the lines copied, and resubmitting starts the chain again from level 1 - the approvals of revision n never
count for n+1. Decided by the level that waits (same permissions and rules as approve_level)."""
if not reason.strip():
raise Refused(422, "a reason is required to return a claim", gate="bounds")
with repository() as repo, _decide():
r, lv, cover = _decider(repo, caller, report_id, "return")
me = _me(caller)
repo.claims.once("close", f"{r.report_id}:r{r.revision}", _now(), _who(caller))
repo.reports.add_step(r.report_id, r.revision, rp.Step(lv.level, lv.kind, "returned", me.email, me.sub, _now(), reason.strip(),
lv.person_id, cover), _now())
r.move("return_report") # terminal for ITS revision ...
old = r.revision
kept = rp.copy_lines(r.lines)
r.revision, r.status, r.chain, r.steps, r.waiting_level = old + 1, "draft", [], [], None # ... the claim goes on as n+1
r.submitted_on = r.submitted_at = r.total_amount = r.reimbursable_amount = None
repo.reports.add_lines(r.report_id, r.revision, kept, _now())
repo.reports.save(r, _now())
return _view(repo, r)
@EXP.use_case(kind="command", route="POST /reports/{report_id}/reject", destructive=True)
def reject_report(caller: dict, report_id: Ref, reason: Reason) -> dict:
"""Reject the claim with a reason (submitted -> rejected, terminal: nothing is posted or paid). Decided by the level that waits
(same permissions and rules as approve_level)."""
if not reason.strip():
raise Refused(422, "a reason is required to reject a claim", gate="bounds")
with repository() as repo, _decide():
r, lv, cover = _decider(repo, caller, report_id, "reject")
me = _me(caller)
repo.claims.once("close", f"{r.report_id}:r{r.revision}", _now(), _who(caller))
step = rp.Step(lv.level, lv.kind, "rejected", me.email, me.sub, _now(), reason.strip(), lv.person_id, cover)
repo.reports.add_step(r.report_id, r.revision, step, _now())
r.steps.append(step)
r.move("reject_report")
r.waiting_level = None
repo.reports.save(r, _now())
return _view(repo, r)
# ── completing, posting, paying: side effects only after a commit, from pending rows ──────────────
def _approval_journal(repo, r: rp.Report, on: date) -> tuple:
"""(source document, gl body) of the ONE journal of an approved revision: its id derived from the report and revision, so posting twice is gl's replay."""
cats = repo.categories.in_force(r.company_code, {(ln.category_code, ln.expense_date) for ln in r.lines})
purposes = {ln.category_code: cats[(ln.category_code, ln.expense_date)]["gl_purpose"] for ln in r.lines}
ref = f"exp:{r.report_id}:r{r.revision}"
return ref, {"journal_id": seams.journal_id(ref, "approve"), "source_module": "expenses", "source_document": ref, "event": "approve",
"company_code": r.company_code, "journal_date": on.isoformat(), "currency": r.currency, "memo": f"Expense claim {r.report_id}"[:200],
"lines": rp.journal_lines(r, purposes, "")}
def _complete(report_id: str, who: str) -> None:
"""Second transaction: every level decided -> approved, dated TODAY (business date), and the journal written as a PENDING row in
this same transaction. Idempotent: a claim already past submitted, or with a level still open, is left alone."""
with repository() as repo, _decide():
r = repo.reports.get(report_id)
if r is None or r.status != "submitted" or r.next_level() is not None or not r.chain:
return
today = _today(repo)
repo.reports.save_lines(r.report_id, r.revision, _settle_tax(repo, r, today, who, only_missing=True), _now()) # a line tax was never asked about
ref, body = _approval_journal(repo, r, today)
repo.claims.once("complete", f"{r.report_id}:r{r.revision}", _now(), who)
_mark_mileage_stale(repo, r)
r.move("complete_approval")
r.approved_on, r.journal_id = today, int(body["journal_id"])
repo.outbox.add("gl", f"{ref}:approve", r.report_id, body, _now())
repo.reports.save(r, _now())
def _payout_body(r: rp.Report, due_on: date) -> dict:
return {"payout_ref": r.payout_ref, "payee_bp_id": str(r.bp_id), "person_id": str(r.person_id), "company_code": r.company_code,
"currency": r.currency, "amount": f"{r.payout_amount():.2f}", "due_on": due_on.isoformat(), "reference": r.report_id,
"source_module": "expenses", "source_document": r.report_id, "clearing_purpose": "employee_payable"}
def _who_system(who: str) -> str:
return who or "expenses"
def _net_personal_spend(repo, r: rp.Report, who: str, today: date) -> None:
"""reimburse_with=nett (EXP-13): the claimant's personal card spend still owed (status personal) is netted off this payout, oldest first, as
many as fit inside what the claim reimburses. Each netted transaction takes its `recover` claim row IN THIS transaction; a transaction a
repayment already took is skipped, never netted twice."""
if repo.settings.on("reimburse_with", today).strip().lower() != "nett" or r.reimbursable() <= ZERO:
return
total = ZERO
ref = f"exp:{r.report_id}:r{r.revision}:pay"
for t in cd.choose_netting(repo.cards.of_holder(r.person_id, status="personal"), r.reimbursable()):
try:
repo.cards.claim(t["card_txn_id"], "recover", _now(), who, report_id=r.report_id, payout_ref=ref)
except AlreadyDone:
continue
total += money(t["billed_amount"])
r.netted_amount = money(total)
def _recover_netted(repo, report_id: str) -> None:
"""The payout is paid (or nothing was left to pay): every transaction netted from it is recovered - personal -> recovered."""
mine = repo.cards.claims_of_report(report_id, "recover")
got = repo.cards.txns([c["card_txn_id"] for c in mine])
for c in mine:
t = got.get(c["card_txn_id"])
if t is not None and t["status"] == "personal":
repo.cards.save_txn({**t, "status": cd.move(t, "recover_personal")}, _now())
def _deliver_gl(row: dict, who: str) -> dict:
"""Send one journal (after its commit). gl's replay answers a repeat. A refusal (the period is closed) parks the claim as
post_failed with gl's gate and reason; no answer keeps it approved for the next pass. Posted -> the payout row is written in the SAME
transaction as the state change."""
body = json.loads(row["payload_json"])
out = seams.post_journal(body, who)
with repository() as repo, _decide():
r = repo.reports.get(row["report_id"])
fresh = repo.outbox.get(row["target"], row["ref"])
if r is None or fresh is None or fresh["status"] != "pending":
return {"ref": row["ref"], "status": "already sent"}
if out["ok"]:
repo.outbox.mark(fresh, "sent", "", _now())
if r.status in ("approved", "post_failed"):
r.move("post_report")
r.fail_reason = None
today = _today(repo)
_net_personal_spend(repo, r, _who_system(who), today)
for ln in r.lines: # a line marked rebillable before posting is asked of ar now, once (its key is the line's)
if ln.rebill_bp_id:
repo.outbox.add("ar", rb.source_id(r.report_id, ln.line_no), r.report_id, rb.invoice_body(r.report_id, ln, r.currency), _now())
if r.payout_amount() > ZERO and r.bp_id:
r.payout_ref = f"exp:{r.report_id}:r{r.revision}:pay"
repo.outbox.add("payments", r.payout_ref, r.report_id, _payout_body(r, today), _now())
elif r.netted_amount and r.payout_amount() == ZERO:
r.move("confirm_paid") # the personal spend netted the whole claim: nothing to pay, the claim is settled
r.paid_on, r.payment_id = today, f"netted:{r.report_id}:r{r.revision}"
_recover_netted(repo, r.report_id)
repo.reports.save(r, _now())
return {"ref": row["ref"], "status": "posted"}
why = f"{out['gate'] + ': ' if out['gate'] else ''}{out['reason']}"
repo.outbox.mark(fresh, "pending", why, _now())
if not out["retry"] and r.status == "approved":
r.move("fail_report_post")
r.fail_reason = why
repo.reports.save(r, _now())
return {"ref": row["ref"], "status": "post_failed" if not out["retry"] else "retry", "reason": why}
def _deliver_payout(row: dict, who: str) -> dict:
"""Ask payments to pay (after the commit that wrote the row). payments replays the same payout_ref; whatever it refuses stays on the
pending row with its reason and the claim stays posted."""
body = json.loads(row["payload_json"])
if body.get("op") == "cancel":
return _deliver_cancel(row, body, who)
out = seams.request_payout(body, who)
with repository() as repo, _decide():
fresh = repo.outbox.get(row["target"], row["ref"])
if fresh is None or fresh["status"] != "pending":
return {"ref": row["ref"], "status": "already sent"}
if out["ok"]:
repo.outbox.mark(fresh, "sent", "", _now())
return {"ref": row["ref"], "status": "requested"}
why = f"{out['gate'] + ': ' if out['gate'] else ''}{out['reason']}"
repo.outbox.mark(fresh, "pending", why, _now())
return {"ref": row["ref"], "status": "refused" if not out["retry"] else "retry", "reason": why}
def _deliver_card_gl(row: dict, who: str) -> dict:
"""Send a card transaction's recovery journal (after the commit that marked it personal). gl replays a repeat; a refusal (a closed period) or
no answer leaves the row pending with the reason for the next pass - there is no claim state to move."""
out = seams.post_journal(json.loads(row["payload_json"]), who)
with repository() as repo, _decide():
fresh = repo.outbox.get(row["target"], row["ref"])
if fresh is None or fresh["status"] != "pending":
return {"ref": row["ref"], "status": "already sent"}
if out["ok"]:
repo.outbox.mark(fresh, "sent", "", _now())
return {"ref": row["ref"], "status": "posted"}
why = f"{out['gate'] + ': ' if out['gate'] else ''}{out['reason']}"
repo.outbox.mark(fresh, "pending", why, _now())
return {"ref": row["ref"], "status": "refused" if not out["retry"] else "retry", "reason": why}
def _deliver_cancel(row: dict, body: dict, who: str) -> dict:
"""Ask payments to cancel a payout (after the commit that wrote the row). Cancelled: the claim has no payout until finance reissues one.
{gate: not_exported} is payments' refusal - the payment stands - kept on the row as refused, never retried. No answer stays pending."""
out = seams.cancel_payout(body["payout_ref"], body["reason"], who)
with repository() as repo, _decide():
fresh = repo.outbox.get(row["target"], row["ref"])
if fresh is None or fresh["status"] != "pending":
return {"ref": row["ref"], "status": "already sent"}
if out["ok"]:
repo.outbox.mark(fresh, "sent", "", _now())
r = repo.reports.get(row["report_id"])
if r is not None and r.status == "posted" and r.payout_ref == body["payout_ref"]:
r.payout_ref = None
repo.reports.save(r, _now())
return {"ref": row["ref"], "status": "cancelled"}
why = f"{out['gate'] + ': ' if out['gate'] else ''}{out['reason']}"
if out["retry"]:
repo.outbox.mark(fresh, "pending", why, _now())
return {"ref": row["ref"], "status": "retry", "reason": why}
repo.outbox.mark(fresh, "refused", why, _now())
return {"ref": row["ref"], "status": "refused", "reason": why, "gate": out["gate"]}
def _drain(report_id: Optional[str] = None, who: str = "expenses", refs: Optional[set] = None) -> list:
"""Send what is pending, oldest first: after a commit, for one claim, or for everything (the worker's tick). One row failing never stops
the others."""
done, seen = [], set()
while True: # a journal that posts writes the payout row: take the next pass, never the same row twice
with repository() as repo:
rows = [r for r in repo.outbox.pending(report_id) if (r["target"], r["ref"]) not in seen and (refs is None or (r["target"], r["ref"]) in refs)]
if not rows:
return done
for row in rows:
seen.add((row["target"], row["ref"]))
try:
if row["target"] == "gl":
done.append(_deliver_gl(row, who) if row["report_id"] else _deliver_card_gl(row, who))
elif row["target"] == "notify":
done.append(_deliver_notify(row, who))
elif row["target"] == "ar":
done.append(_deliver_ar(row, who))
else:
done.append(_deliver_payout(row, who))
except Refused as lost:
done.append({"ref": row["ref"], "status": "lost a race", "reason": lost.reason})
@EXP.use_case(kind="command", route="POST /reports/{report_id}/post", permission=POST, idempotent=True)
def post_report(caller: dict, report_id: Ref) -> dict:
"""Post an approved claim again (approved or post_failed -> posted): finance does this after the ledger refused it (a closed period,
now open). The journal is the one stored at approval - same id, same date - so gl replays it and nothing is posted twice."""
with repository() as repo, _decide():
r = _visible(repo, caller, report_id)
if r.status not in rp.MOVES["post_report"][0]:
if r.status in ("posted", "paid"):
return _view(repo, r)
raise RuleBroken(f"Report {r.report_id} is {r.status}: only an approved claim is posted")
_drain(report_id, _who(caller))
with repository() as repo:
return _view(repo, _visible(repo, caller, report_id))
@EXP.use_case(kind="command", route="POST /outbox/drain", permission=POST_AUTO, idempotent=True, tool=False)
def drain_outbox(caller: dict) -> dict:
"""The worker's tick (no server-side scheduler): complete any claim whose last level was approved but never completed, then send every
pending journal and payout. Safe to run at any time and from two workers: every send is idempotent at the neighbour."""
with repository() as repo:
stuck = repo.reports.stuck_completions()
for rid in stuck:
try:
_complete(rid, _who(caller))
except Refused:
pass
return {"completed": len(stuck), "sent": _drain(None, _who(caller))}
@EXP.use_case(kind="command", route="POST /reports/{report_id}/paid", permission=MARK_PAID, idempotent=True, tool=False)
def confirm_paid(caller: dict, report_id: Ref, payout_ref: Ref, payment_id: Ref, paid_on: Day, amount: AmountText) -> dict:
"""payments tells us the bank paid the payout (posted -> paid, terminal). Idempotent on payout_ref; another payout_ref or another amount
is a 409 {gate: amount_matches}, and a claim that is not posted yet is a 409. Held by payments' service client alone."""
when = _date(paid_on, "paid_on")
try:
paid = parse_money(amount)
except ValueError as e:
raise _bounds(str(e)) from None
with repository() as repo, _decide():
r = repo.reports.get(report_id)
if r is None:
raise Refused(404, f"no expense report {report_id}")
if r.payout_ref != payout_ref:
raise Refused(409, f"{payout_ref} is not the payout of {report_id}", gate="amount_matches", fix="payments confirms the payout we requested")
if money(paid) != r.payout_amount():
raise Refused(409, f"the bank paid {paid:.2f} but the payout is {r.payout_amount():.2f}", gate="amount_matches",
fix="payments and expenses must agree on the amount before it is paid")
if r.status == "paid":
return _view(repo, r) # the same payout_ref and amount again: the same answer
r.move("confirm_paid")
repo.claims.once("paid", payout_ref, _now(), _who(caller), payment_id)
r.paid_on, r.payment_id = when, payment_id
_recover_netted(repo, r.report_id) # the netted personal spend is recovered with the payout
repo.reports.save(r, _now())
return _view(repo, r)
# ── rates, categories and settings: dated rows, never code (EXP-22, EXP-23) ────────────────────────────
def _dated(before: list, when: date, today: date, what: str) -> None:
"""The dating rule of every row finance or the owner adds: the first row of a key may start today, a change starts AFTER today (so a claim keeps
what it was written under), and nothing starts in the past."""
if when < today:
raise _bounds(f"valid_from {when} is in the past: a row starts today or later")
if before and when <= today:
raise Refused(409, f"{what} already has a row: a change starts after today ({today})", gate="valid_from", fix="date the new row tomorrow or later")
def _rate_view(r: rt.Rate) -> dict:
return {"kind": r.kind, "country": r.country, "variant": r.variant, "band": r.band, "valid_from": r.valid_from.isoformat(),
"rate_amount": f"{r.rate_amount:.4f}", "currency": r.currency, "band_upto": None if r.band_upto is None else f"{r.band_upto:.4f}",
"unit": r.unit or None, "partial_day_amount": None if r.partial_day_amount is None else f"{r.partial_day_amount:.4f}",
"breakfast_pct": f"{r.breakfast_pct:.4f}", "lunch_pct": f"{r.lunch_pct:.4f}", "dinner_pct": f"{r.dinner_pct:.4f}", "year_start": r.year_start,
"source": r.source}
@EXP.use_case(kind="command", route="POST /rates", permission=RATE_WRITE, status=201)
def put_rate(caller: dict, kind: Literal["mileage", "per_diem"], country: Country, variant: Code, band: Annotated[int, Field(ge=1, le=20)], valid_from: Day,
rate_amount: RateAmount, currency: Currency, source: text(300, min_length=1), unit: Optional[Literal["mile", "km"]] = None,
band_upto: Optional[RateAmount] = None, partial_day_amount: Optional[RateAmount] = None, breakfast_pct: Percentage = Decimal(0),
lunch_pct: Percentage = Decimal(0), dinner_pct: Percentage = Decimal(0), year_start: YearStart = "01-01") -> dict:
"""Add a dated mileage or daily-allowance rate row for a country: mileage per mile or km by vehicle (variant car, motorcycle, bicycle) in bands
that end at band_upto cumulative miles of the tax year (no end on the last), per_diem per 24 hours by destination (variant domestic or a country
code) with the partial-day amount and the share each provided meal takes off. Every row names its source (the statute or authority) and starts today
(the first row of a key) or after today. A claim uses the row in force on each line's date; unapproved lines of that kind are marked stale."""
when = _date(valid_from, "valid_from")
if kind == "mileage" and unit is None:
raise _bounds("a mileage rate names its unit: mile or km")
if kind == "per_diem" and (partial_day_amount is None or partial_day_amount > rate_amount):
raise _bounds("a daily allowance names the amount for a partial day, and it is not more than the full day")
if band_upto is not None and band_upto <= 0:
raise _bounds("band_upto is more than zero (or left out for the last band)")
with repository() as repo, _decide():
today = _today(repo)
cc, c = _cur(currency), country.upper()
_dated(repo.rates.of_key(kind, c, variant, band), when, today, f"{kind} {c} {variant} band {band}")
row = rt.Rate(kind, c, variant, band, when, rate_amount, cc, band_upto, unit or "", partial_day_amount, breakfast_pct, lunch_pct, dinner_pct, year_start,
source.strip())
repo.rates.put(row, _who(caller), _now())
stale = _mark_stale(repo, "*", [], when, f"{kind} rate {c} {variant} from {when}", kinds=[kind]) if when <= today else 0
return {**_rate_view(row), "lines_marked_stale": stale}
@EXP.use_case(kind="query", route="GET /rates")
def list_rates(caller: dict, kind: Optional[Literal["mileage", "per_diem"]] = None, country: Optional[Country] = None) -> dict:
"""Every rate row (with its source and the date it starts), for a kind and country if given: what finance edits and a claim is computed from."""
_need(caller, READ_OWN)
with repository() as repo:
return {"rates": [_rate_view(r) for r in repo.rates.listed(kind, country.upper() if country else None)]}
def _category_view(c: dict) -> dict:
return {"company_code": c["company_code"], "category_code": c["category_code"], "valid_from": c["valid_from"].isoformat(), "name": c["name"],
"gl_purpose": c["gl_purpose"], "tax_category": c["tax_category"], "receipt_required": bool(c["receipt_required"]), "kind": c["kind"]}
@EXP.use_case(kind="command", route="POST /categories", permission=CATEGORY_WRITE, status=201)
def put_category(caller: dict, company_code: Code, category_code: Annotated[str, Field(pattern=r"^[A-Z][A-Z0-9_]{0,63}$")], name: text(100, min_length=1),
gl_purpose: Annotated[str, Field(pattern=r"^[a-z][a-z0-9_]{0,63}$")], tax_category: Code, valid_from: Day, receipt_required: bool = True,
kind: Literal["receipt", "mileage", "per_diem"] = "receipt") -> dict:
"""Add a dated expense category for a company ('*' = every company): what employees choose, the gl PURPOSE it posts to (lower case, such as
training - gl maps a purpose to an account, so an account number typed here is refused) and tax's category for its VAT. The first row of a code may
start today, a change starts after today; a line uses the row in force on its date. A purpose gl does not map yet is refused by gl at posting, naming it."""
when = _date(valid_from, "valid_from")
with repository() as repo, _decide():
today = _today(repo)
_dated(repo.categories.of_key(company_code.strip(), category_code), when, today, f"{category_code} of {company_code}")
repo.categories.put(company_code.strip(), category_code, name.strip(), gl_purpose, tax_category.strip(), when, _now(), receipt_required, kind)
return _category_view({"company_code": company_code, "category_code": category_code, "valid_from": when, "name": name, "gl_purpose": gl_purpose,
"tax_category": tax_category, "receipt_required": receipt_required, "kind": kind})
@EXP.use_case(kind="query", route="GET /categories")
def list_categories(caller: dict, company_code: Code, on: Optional[Day] = None) -> dict:
"""The expense categories in force for a company on a date (default today): code, name, the gl purpose it posts to, tax's category, whether a
receipt is needed."""
_need(caller, READ_OWN)
with repository() as repo:
when = _date(on, "on") if on else _today(repo)
rows = repo.categories.all_in_force(company_code, when)
return {"categories": [_category_view(c) for c in rows]}
@EXP.use_case(kind="command", route="POST /settings", status=201)
def set_setting(caller: dict, key: Code, value: text(100, min_length=1), valid_from: Day) -> dict:
"""Change a setting from a date: dated rows, so a claim submitted before the date keeps what was in force. The controls (the threshold above which
finance looks, the receipt threshold, delegation, the zone 'today' is counted in, how personal card spend is repaid, the AI steps' modes) are the
owner's alone (exp:settings:controls); the others - the chase days, the leaver window, the retention years - are finance's (exp:policy:write). The
first row of a key may start today; a change starts after today."""
if key not in SETTING_DEFS:
raise _bounds(f"{key} is not a setting", f"settings: {', '.join(sorted(SETTING_DEFS))}")
_need(caller, CONTROLS if key in CONTROL_SETTINGS else POLICY_WRITE)
when = _date(valid_from, "valid_from")
try:
stored = read_setting(key, value)
except ValueError as bad:
raise _bounds(str(bad)) from None
with repository() as repo, _decide():
today = _today(repo)
_dated([r for r in repo.settings.rows() if r["setting_key"] == key], when, today, key)
repo.settings.put(key, stored, when, _who(caller), _now())
return {"key": key, "value": stored, "valid_from": when.isoformat()}
@EXP.use_case(kind="query", route="GET /settings")
def list_settings(caller: dict) -> dict:
"""Every setting with the value in force today, its default, what it changes, and the rows dated after today."""
_need(caller, READ_OWN)
with repository() as repo:
today = _today(repo)
rows = repo.settings.rows()
return {"settings": [{"key": k, "value": repo.settings.on(k, today), "default": d, "changes": why, "owner_only": k in CONTROL_SETTINGS,
"upcoming": [{"valid_from": r["valid_from"].isoformat(), "value": r["value"]} for r in rows
if r["setting_key"] == k and r["valid_from"] > today]} for k, (d, why) in sorted(SETTING_DEFS.items())]}
# ── mileage and daily allowance (EXP-03, EXP-04, EXP-20) ────────────────────────────────────────────
def _bands(repo, country: str, variant: str, when: date, currency: str) -> list:
bands = rt.in_force(repo.rates.rows("mileage", country, variant, when), "mileage", country, variant, when)
if not bands:
raise Refused(409, f"No mileage rate for {country} {variant} is in force on {when}", gate="rate_in_force", fix="finance adds the rate row (put_rate)")
if bands[0].currency != currency:
raise Refused(409, f"The {country} {variant} mileage rate is in {bands[0].currency}; this claim is in {currency}", gate="rate_currency")
return bands
def _mileage(repo, r: rp.Report, ln: rp.Line, bands: list, when: date) -> tuple:
"""(amount, calc) of a mileage line from the APPROVED miles of the person's tax year: derived each time, never stored."""
start, end = rt.tax_year(when, bands[0].year_start)
before = repo.reports.approved_miles(r.person_id, start, end, ln.rate_country, ln.rate_variant, exclude_report=r.report_id)
amount, split = rt.mileage(ln.miles, before, bands)
return amount, {"unit": bands[0].unit, "country": ln.rate_country, "variant": ln.rate_variant, "tax_year": [start.isoformat(), end.isoformat()],
"cumulative_before": f"{before:.4f}", "bands": split, "rate_from": max(b.valid_from for b in bands).isoformat(), "source": bands[0].source}
def _computed_line(repo, r: rp.Report, caller: dict, **fields) -> rp.Line:
cost_centre = seams.employment(r.person_id, _today(repo), _who(caller)).get("cost_centre") or ""
return rp.Line(max((x.line_no for x in r.lines), default=0) + 1, cost_centre=cost_centre, receipt_currency=r.currency, **fields)
@EXP.use_case(kind="command", route="POST /reports/{report_id}/mileage", permission=WRITE_OWN, status=201)
def add_mileage(caller: dict, report_id: Ref, expense_date: Day, miles: Miles, route: Merchant, variant: Code = "car", country: Optional[Country] = None,
description: Optional[Note] = None) -> dict:
"""Claim mileage in your own vehicle by the country's rates: the miles (or km) driven, the route, the vehicle (car, motorcycle, bicycle) and the
country (the company's home country when left out). The rate bands are read from the rows in force on the date: the miles of the tax year already
APPROVED decide which band these start in (0.45 a mile up to 10,000 in a UK tax year, 0.25 after), and the split is shown on the line. A claim approved
before this one is approved can move the band: the line is then marked stale and recomputed (recompute_report) before it can be approved. No receipt
is needed."""
when = _date(expense_date, "expense_date")
with repository() as repo, _decide():
r = _own(repo, caller, report_id)
r.ensure_draft()
today = _today(repo)
if len(r.lines) >= rp.LINES_MAX:
raise Refused(409, f"A claim holds at most {rp.LINES_MAX} lines: submit this one and start another")
if repo.categories.in_force(r.company_code, {("MILEAGE", when)}).get(("MILEAGE", when)) is None:
raise Refused(409, f"No expense category MILEAGE is in force for {r.company_code} on {when}", gate="category_in_force")
c = (country or repo.settings.on("home_country", today)).upper()
bands = _bands(repo, c, variant, when, r.currency)
ln = _computed_line(repo, r, caller, category_code="MILEAGE", expense_date=when, merchant=route.strip(), amount=ZERO, kind="mileage", miles=miles,
rate_country=c, rate_variant=variant, description=(description or "").strip())
ln.amount, ln.calc = _mileage(repo, r, ln, bands, when)
if ln.amount <= ZERO:
raise _bounds("those miles come to nothing at the rate in force")
ln.receipt_amount = ln.amount
rp.check_line(ln, today=today, currency=r.currency, report_currency=r.currency)
ln.policy_results = _policy_of(repo, r, [ln])[ln.line_no]
repo.reports.add_lines(r.report_id, r.revision, [ln], _now())
r.lines.append(ln)
return _view(repo, r)
@EXP.use_case(kind="command", route="POST /reports/{report_id}/per-diem", permission=WRITE_OWN, status=201)
def add_per_diem(caller: dict, report_id: Ref, trip_from: Moment, trip_to: Moment, destination: Code = "domestic", country: Optional[Country] = None,
meals_provided: Optional[Annotated[list[MealProvided], Field(max_length=100)]] = None, description: Optional[Note] = None) -> dict:
"""Claim the daily allowance for a business trip by the country's rules: start and end (the traveller's own clock, like 2026-11-10T07:00), the
destination (domestic or a country code - the rate row decides) and the meals the employer provided. An arrival day and a departure day take the
partial-day amount, a full day (24 hours away) the full rate, a trip inside one day the partial amount when it lasts more than 8 hours; each provided
meal takes its share of the full-day rate off its day and a day never goes below 0.00. Trips of one person never overlap. No receipt is needed."""
try:
start, end = datetime.fromisoformat(trip_from.strip()).replace(tzinfo=None), datetime.fromisoformat(trip_to.strip()).replace(tzinfo=None)
except ValueError:
raise _bounds("a trip's start and end are date-times like 2026-11-10T07:00") from None
for moment in (start, end):
if not date(2000, 1, 1) <= moment.date() <= date(2100, 1, 1):
raise _bounds(f"{moment.date()} is outside 2000-01-01 .. 2100-01-01")
provided = [(day(m.date), m.meal) for m in (meals_provided or [])]
with repository() as repo, _decide():
r = _own(repo, caller, report_id)
r.ensure_draft()
today = _today(repo)
if start.date() > today:
raise Refused(409, f"A trip starting {start.date()} is in the future (today is {today})")
if len(r.lines) >= rp.LINES_MAX:
raise Refused(409, f"A claim holds at most {rp.LINES_MAX} lines: submit this one and start another")
if repo.categories.in_force(r.company_code, {("PER_DIEM", start.date())}).get(("PER_DIEM", start.date())) is None:
raise Refused(409, f"No expense category PER_DIEM is in force for {r.company_code} on {start.date()}", gate="category_in_force")
c = (country or repo.settings.on("home_country", today)).upper()
bands = rt.in_force(repo.rates.rows("per_diem", c, destination, start.date()), "per_diem", c, destination, start.date())
if not bands:
raise Refused(409, f"No daily allowance for {c} {destination} is in force on {start.date()}", gate="rate_in_force", fix="finance adds the rate row (put_rate)")
if bands[0].currency != r.currency:
raise Refused(409, f"The {c} {destination} allowance is in {bands[0].currency}; this claim is in {r.currency}", gate="rate_currency")
for _, _, other in repo.reports.open_lines(person_id=r.person_id, kinds=["per_diem"],
statuses=("draft", "submitted", "approved", "posted", "paid")):
if other.trip_from and other.trip_to and rt.overlaps(start, end, other.trip_from, other.trip_to):
raise Refused(409, f"This trip overlaps another of yours ({other.trip_from:%Y-%m-%d %H:%M} to {other.trip_to:%Y-%m-%d %H:%M}): "
"one trip, one allowance", gate="overlapping_trip")
amount, days = rt.per_diem(start, end, provided, bands[0])
if amount <= ZERO:
raise Refused(409, "That trip earns no allowance (a trip inside one day needs more than 8 hours away)", gate="no_allowance")
ln = _computed_line(repo, r, caller, category_code="PER_DIEM", expense_date=start.date(), merchant=destination, amount=amount, kind="per_diem",
rate_country=c, rate_variant=destination, trip_from=start, trip_to=end,
meals=[{"date": d.isoformat(), "meal": m} for d, m in provided], description=(description or "").strip())
ln.receipt_amount = amount
ln.calc = {"days": days, "country": c, "destination": destination, "rate_from": bands[0].valid_from.isoformat(), "full_day": f"{bands[0].rate_amount:.2f}",
"partial_day": f"{bands[0].partial_day_amount:.2f}", "source": bands[0].source}
rp.check_line(ln, today=today, currency=r.currency, report_currency=r.currency)
ln.policy_results = _policy_of(repo, r, [ln])[ln.line_no]
repo.reports.add_lines(r.report_id, r.revision, [ln], _now())
r.lines.append(ln)
return _view(repo, r)
def _recompute_computed(repo, r: rp.Report, ln: rp.Line, today: date) -> Optional[dict]:
"""A mileage or per diem line again from the rows in force and the approved miles NOW: a change record, or None when nothing moved."""
if ln.kind == "mileage":
bands = _bands(repo, ln.rate_country, ln.rate_variant, ln.expense_date, r.currency)
amount, calc = _mileage(repo, r, ln, bands, ln.expense_date)
elif ln.kind == "per_diem":
bands = rt.in_force(repo.rates.rows("per_diem", ln.rate_country, ln.rate_variant, ln.expense_date), "per_diem", ln.rate_country, ln.rate_variant, ln.expense_date)
if not bands:
raise Refused(409, f"No daily allowance for {ln.rate_country} {ln.rate_variant} is in force on {ln.expense_date}", gate="rate_in_force")
amount, days = rt.per_diem(ln.trip_from, ln.trip_to, [(day(m["date"]), m["meal"]) for m in ln.meals], bands[0])
calc = {**ln.calc, "days": days, "rate_from": bands[0].valid_from.isoformat(), "full_day": f"{bands[0].rate_amount:.2f}",
"partial_day": f"{bands[0].partial_day_amount:.2f}"}
else:
return None
if amount <= ZERO:
raise Refused(409, f"line {ln.line_no} would come to nothing at the rates now in force: remove it", gate="no_allowance")
was, ln.amount, ln.receipt_amount, ln.calc = ln.amount, amount, amount, calc
return None if amount == was else {"line_no": ln.line_no, "field": "amount", "was": _s(was), "now": _s(amount), "cause": ln.stale_reason or "recomputed"}
# ── the daily chase (EXP-17) ─────────────────────────────────────────────────────────────────────────
def _deliver_ar(row: dict, who: str) -> dict:
"""Ask ar to raise the draft invoice line of a re-billed cost (after the commit that wrote the row). ar replays the same source_id; its refusal (an
unknown customer, a tax category it has no rule for) stays on the row as refused until finance marks the line again; no answer stays pending."""
body = json.loads(row["payload_json"])
out = seams.raise_invoice(body, who)
with repository() as repo, _decide():
fresh = repo.outbox.get(row["target"], row["ref"])
if fresh is None or fre