dodil/erp-brick-payroll
DODIL ERP brick: payroll
src/brick_payroll/application/filing.py
"""What goes out after an approval, as files handed to a capability - never submitted from here, never inside the approval's transaction:
the FPS (PRL-13) one per approved run, key fps:<run>:<n>; accepted -> the run is `submitted` (record_fps_accepted); rejected stays approved with
HMRC's errors for the admin, and a corrected FPS is n+1 (resend_fps); with no provider a manual task with the file and its deadline
the pension file one per scheme per run, key pension:<scheme>:<run>; the legal due date is the 22nd of the next month (PRL-15)
the EPS (PRL-14) one per tax month per PAYE reference, key eps:<paye ref>:<tax year>:M<nn>:<n>: SMP recovered to date, or no payment for the month;
its journal (event eps) debits paye_ni_payable and credits statutory_pay_recovery
Each is built from COMMITTED rows into a deterministic document (infrastructure/rti.py), claimed by a plain INSERT on its submission id, and handed over
from a pending outbox row by the drain. A submission already made is never made again: the drain finds its row and carries on from its status."""
import json
from datetime import date, timedelta
from decimal import Decimal
from typing import Annotated, Optional
from erp_platform.bricks import Refused
from erp_platform.seam import derived_id
from pydantic import Field
from brick_payroll.application import use_cases as hub
from brick_payroll.application.inputs import Code, Ref
from brick_payroll.domain import pay_run as pr
from brick_payroll.domain import paye
from brick_payroll.domain import statutory as st
from brick_payroll.domain.errors import RuleBroken
from brick_payroll.domain.money import ZERO, money
from brick_payroll.infrastructure import filing, rti, seams
PRL = hub.PRL
RTI_READ, RTI_SUBMIT, FILE_WRITE = "prl:rti:read", "prl:rti:submit", hub.RUN_CALC
TERMINAL = ("accepted", "rejected", "manual")
def _view(s: dict) -> dict:
return {"submission_id": s["submission_id"], "kind": s["kind"], "pay_run_id": s["pay_run_id"], "paye_ref": s["paye_ref"], "tax_year": s["tax_year"],
"tax_period": s["tax_period"], "scheme_code": s["scheme_code"], "n": s["n"], "object_key": s["object_key"], "sha256": s["sha256"],
"status": s["status"], "correlation_id": s["correlation_id"] or None, "message": s["message"] or None,
"errors": json.loads(s["errors_json"] or "[]"), "attempts": s["attempts"], "due_on": s["due_on"].isoformat() if s["due_on"] else None,
"sent_at": s["sent_at"].isoformat() if s["sent_at"] else None, **({"totals": json.loads(s["totals_json"])} if s["totals_json"] else {})}
def _outbox(repo, row: dict, status: str, why: str) -> None:
fresh = repo.outbox.get(row["target"], row["ref"])
if fresh is not None and fresh["status"] == "pending":
repo.outbox.mark(fresh, status, why, hub._now())
def _apply(repo, sub: dict, answer: dict, due: Optional[date], today: date) -> dict:
"""Record what the capability answered on the submission row (status moves forward only: pending/failed -> accepted | rejected | manual)."""
status = answer["status"]
row = {**sub, "attempts": (sub["attempts"] or 0) + 1}
if status == "accepted":
late = due is not None and today > due
row.update(status="accepted", correlation_id=answer.get("correlation_id") or "", sent_at=hub._now().replace(tzinfo=None),
message=("late: accepted after its deadline " + due.isoformat() + "; HMRC needs a late reporting reason") if late else "")
elif status == "rejected":
row.update(status="rejected", errors_json=json.dumps(answer.get("errors") or [], sort_keys=True), message=answer.get("message") or "rejected")
elif status == "manual":
row.update(status="manual", message=answer.get("message") or "send this file by hand")
else:
row.update(status="pending", message=answer.get("message") or "the provider did not answer")
repo.rti.save(row, hub._now())
return row
def _filings_state(repo, run: pr.PayRun) -> tuple:
"""(fps accepted?, the pension side finished?) from the run's submissions and its pension outbox row."""
subs = repo.rti.of_run(run.pay_run_id)
fps_ok = any(s["kind"] == "fps" and s["status"] == "accepted" for s in subs)
box = repo.outbox.get("pension", f"pension:{run.pay_run_id}")
pension_ok = (box is None or box["status"] == "sent") and all(s["status"] == "accepted" for s in subs if s["kind"] == "pension")
return fps_ok, pension_ok
def advance_run(repo, run: pr.PayRun) -> list:
"""After anything changes the world around a run (FPS answered, a payout settled, the journal posted, the pension file answered): take the arrows
the run now qualifies for, in the same transaction as the fact. Payslips must be loaded (lines are not needed)."""
if run.status not in ("approved", "submitted", "paid"):
return []
fps_ok, pension_ok = _filings_state(repo, run)
taken = pr.advance(run, fps_ok, pension_ok)
if taken:
repo.runs.save_root(run, hub._now())
return taken
# ── the FPS ─────────────────────────────────────────────────────────────────────────────────
def _fps_build(run_id: str, n: int, who: str) -> None:
"""Build and claim the FPS n of an approved run from committed rows (a second worker's claim is simply carried on from). hr's identities are read first."""
with hub.repository() as repo:
run = repo.runs.get(run_id, lines=True)
emps = repo.employees.many([p.payroll_id for p in run.payslips])
scheme = repo.config.paye_scheme(run.company_code, run.paye_ref)
idents = {p.payroll_id: seams.identity(p.person_id, run.pay_date, who) for p in run.payslips}
content = rti.fps_xml(run, n, scheme["accounts_office_ref"] if scheme else "", run.payslips, idents, emps)
doc = rti.document(rti.fps_key(run, n), "fps", content, "application/xml", run.pay_run_id, f"{run.pay_run_id}-{n}")
sid = f"fps:{run.pay_run_id}:{n}"
row = {"submission_id": sid, "kind": "fps", "pay_run_id": run.pay_run_id, "company_code": run.company_code, "paye_ref": run.paye_ref,
"tax_year": run.tax_year, "tax_period": run.tax_period, "n": n, "object_key": doc["object_key"], "sha256": doc["sha256"], "status": "pending",
"due_on": run.pay_date}
try:
with hub.repository() as repo:
if repo.rti.get(sid) is None:
repo.documents.add([doc], hub._now())
repo.rti.claim(row, hub._now())
except Refused:
pass # a second worker claimed it first: carry on from its row
def _hand_over(sub: dict, payment_date: str, accounts_office_ref: str, kind: str) -> dict:
data = {"kind": kind, "paye_ref": sub["paye_ref"], "accounts_office_ref": accounts_office_ref, "tax_year": sub["tax_year"], "tax_period": sub["tax_period"],
"payment_date": payment_date, "file_object_key": sub["object_key"], "sha256": sub["sha256"]}
return filing.submit("hmrc.rti_submit", sub["submission_id"], data, sub["due_on"])
def deliver_fps(row: dict, who: str) -> dict:
sid = row["ref"]
run_id, n = row["pay_run_id"], int(sid.rsplit(":", 1)[1])
with hub.repository() as repo:
sub = repo.rti.get(sid)
if sub is None:
_fps_build(run_id, n, who)
with hub.repository() as repo:
sub = repo.rti.get(sid)
run = repo.runs.root(run_id)
scheme = repo.config.paye_scheme(run.company_code, run.paye_ref)
aor = scheme["accounts_office_ref"] if scheme else ""
if sub["status"] in ("pending",):
answer = _hand_over(sub, run.pay_date.isoformat(), aor, "fps")
else:
answer = {"status": sub["status"]}
with hub.repository() as repo, hub._decide():
fresh = repo.rti.get(sid)
run = repo.runs.get(run_id, lines=False)
today = hub._today(repo)
if fresh["status"] == "pending":
fresh = _apply(repo, fresh, answer, run.pay_date, today)
done = {"accepted": ("sent", ""), "manual": ("sent", fresh["message"]), "rejected": ("refused", "rejected by HMRC: see the submission's errors"),
"pending": ("pending", fresh["message"] or "waiting")}[fresh["status"]]
_outbox(repo, row, *done)
if fresh["status"] == "accepted":
advance_run(repo, run)
return {"ref": sid, "status": {"accepted": "accepted", "manual": "manual task", "rejected": "rejected", "pending": "retry"}[fresh["status"]],
**({"reason": fresh["message"]} if fresh["message"] else {})}
# ── the pension contribution file ───────────────────────────────────────────────────────────
def deliver_pension(row: dict, who: str) -> dict:
run_id = row["pay_run_id"]
with hub.repository() as repo:
run = repo.runs.get(run_id, lines=False)
ids = [p.payroll_id for p in run.payslips]
facts = repo.employees.facts(ids, run.pay_date)
have = {s["submission_id"]: s for s in repo.rti.of_run(run_id) if s["kind"] == "pension"}
by_scheme: dict = {}
for p in run.payslips:
mem = (facts.get(p.payroll_id) or {}).get("membership")
if mem:
by_scheme.setdefault(mem["scheme_code"], []).append((p, mem))
results = []
for scheme_code in sorted(by_scheme):
sid = f"pension:{scheme_code}:{run_id}"
members = by_scheme[scheme_code]
if sid not in have:
idents = {p.payroll_id: seams.identity(p.person_id, run.pay_date, who) for p, _ in members}
rows = []
for p, mem in members:
w = idents[p.payroll_id]
qe = (p.ytd.get("pension") or {}).get("qualifying_earnings", "0.00")
rows.append({"payroll_id": p.payroll_id, "ni_number": w.get("ni_number") or "", "surname": w.get("legal_last_name", ""),
"forenames": w.get("legal_first_names", ""), "date_of_birth": w.get("date_of_birth", ""), "pensionable_earnings": qe,
"employee_contribution": f"{p.ee_pension:.2f}", "employer_contribution": f"{p.er_pension:.2f}", "enrolment_status": mem["status"],
"enrolment_date": mem["enrolled_on"].isoformat(), "assessed_as": mem["assessed_as"] or ""})
content = rti.pension_csv(run, rows)
doc = rti.document(rti.pension_key(run, scheme_code), "pension", content, "text/csv", run_id, sid)
totals = {"employee": f"{sum((p.ee_pension for p, _ in members), ZERO):.2f}", "employer": f"{sum((p.er_pension for p, _ in members), ZERO):.2f}"}
sub = {"submission_id": sid, "kind": "pension", "pay_run_id": run_id, "company_code": run.company_code, "paye_ref": run.paye_ref, "tax_year": run.tax_year,
"tax_period": run.tax_period, "scheme_code": scheme_code, "n": 1, "object_key": doc["object_key"], "sha256": doc["sha256"], "status": "pending",
"totals_json": json.dumps(totals, sort_keys=True), "due_on": rti.pension_due(run.pay_date)}
try:
with hub.repository() as repo:
if repo.rti.get(sid) is None:
repo.documents.add([doc], hub._now())
repo.rti.claim(sub, hub._now())
except Refused:
pass
with hub.repository() as repo:
sub = repo.rti.get(sid)
if sub["status"] == "pending":
data = {"scheme_code": scheme_code, "pay_run_id": run_id, "pay_date": run.pay_date.isoformat(), "format": "papdis", "file_object_key": sub["object_key"],
"sha256": sub["sha256"], "totals": json.loads(sub["totals_json"])}
answer = filing.submit("pension.submit_contributions", sid, data, sub["due_on"])
with hub.repository() as repo, hub._decide():
fresh = repo.rti.get(sid)
if fresh["status"] == "pending":
fresh = _apply(repo, fresh, answer, None, hub._today(repo))
results.append(fresh["status"])
else:
results.append(sub["status"])
retry = [r for r in results if r == "pending"]
with hub.repository() as repo, hub._decide():
run = repo.runs.get(run_id, lines=False)
bad = [r for r in results if r == "rejected"]
_outbox(repo, row, "pending" if retry else ("refused" if bad else "sent"), "the scheme or the provider did not answer" if retry else
("a contribution file was rejected: see the submission's errors" if bad else ""))
advance_run(repo, run)
return {"ref": row["ref"], "status": "retry" if retry else ("rejected" if bad else "sent"), "files": len(results)}
# ── the EPS ─────────────────────────────────────────────────────────────────────────────────
def tax_month_dates(tax_year: str, month: int) -> tuple:
"""(first, last day) of tax month 1..12 of a tax year like 2026-27: tax month 1 is 6 April to 5 May."""
y = int(tax_year[:4])
def start(m: int) -> date:
k = 3 + m - 1
return date(y + k // 12, k % 12 + 1, 6)
return start(month), start(month + 1) - timedelta(days=1)
def _year(v: str) -> str:
try:
y = int(v[:4])
ok = len(v) == 7 and v[4] == "-" and int(v[5:]) == (y + 1) % 100 and 2000 <= y <= 2099
except ValueError:
ok = False
if not ok:
raise hub._bounds(f"{v!r} is a tax year like 2026-27")
return v
@PRL.use_case(kind="command", route="POST /rti/eps", permission=RTI_SUBMIT, idempotent=True, tool=False)
def run_eps(caller: dict, paye_ref: Code, tax_year: Code, tax_month: Annotated[int, Field(ge=1, le=12)]) -> dict:
"""The employer payment summary of a tax month (PRL-14), after the month has ended and before the 19th: statutory maternity pay recovered to date
(92% of SMP, 109% (2026/27) when small employers' relief is set), or the no-payment EPS for a month without a pay date, handed to hmrc.rti_submit from a pending
row, with its journal (debit paye_ni_payable, credit statutory_pay_recovery) for what this EPS adds to the last. The same facts a second time send
nothing; changed facts send EPS n+1. The worker's tick: machine permission, not a tool."""
year, ref = _year(tax_year.strip()), paye_ref.strip().upper()
first, last = tax_month_dates(year, tax_month)
who = hub._who(caller)
with hub.repository() as repo, hub._decide():
today = hub._today(repo)
if today <= last:
raise RuleBroken(f"tax month {tax_month} of {year} runs to {last}: its EPS is sent after it ends", gate="month_open")
runs = repo.runs.frozen_of_scheme(ref, year)
smp = repo.runs.element_by_run([r["pay_run_id"] for r in runs], "SMP")
scheme = repo.config.paye_scheme_of(ref)
if scheme is None:
raise RuleBroken(f"no PAYE scheme {ref}", gate="paye_scheme", fix="record it first (put_paye_scheme)")
rates = paye.Rates(repo.config.rates(last))
small = repo.settings.on("small_employer_relief", last) == "true"
allowance = repo.settings.on("employment_allowance_claim", last) == "true"
eps = repo.rti.eps_of(ref, year)
in_year = [r for r in runs if paye.tax_month(r["pay_date"]) <= tax_month]
in_month = [r for r in runs if paye.tax_month(r["pay_date"]) == tax_month]
companies = sorted({r["company_code"] for r in in_year if smp.get(r["pay_run_id"])})
if len(companies) > 1:
raise RuleBroken(f"{', '.join(companies)} share PAYE reference {ref}: one EPS journal cannot be split between companies yet", gate="company_scope")
ytd = sum((st.recovery(smp.get(r["pay_run_id"], ZERO), small, rates) for r in in_year), ZERO)
company = companies[0] if companies else scheme["company_code"]
this_month = [e for e in eps if e["tax_period"] == tax_month]
n = 1
if this_month:
prev = this_month[-1]
same = rti.eps_xml(ref, scheme["accounts_office_ref"], year, tax_month, prev["n"], ytd, not in_month, allowance)
if rti.document("", "eps", same, "application/xml")["sha256"] == prev["sha256"]:
return {"submission": _view(prev), "sent": False, "reason": "the same facts were already sent: nothing to send"}
n = prev["n"] + 1
before = Decimal(json.loads(eps[-1]["totals_json"] or "{}").get("ytd_smp_recovered", "0")) if eps else ZERO
delta = money(ytd - before)
sid = f"eps:{ref}:{year}:M{tax_month:02d}:{n}"
content = rti.eps_xml(ref, scheme["accounts_office_ref"], year, tax_month, n, ytd, not in_month, allowance)
doc = rti.document(rti.eps_key(company, ref, year, tax_month, n), "eps", content, "application/xml", "", sid)
due = date(last.year, last.month, 19)
sub = {"submission_id": sid, "kind": "eps", "pay_run_id": None, "company_code": company, "paye_ref": ref, "tax_year": year, "tax_period": tax_month, "n": n,
"object_key": doc["object_key"], "sha256": doc["sha256"], "status": "pending", "due_on": due,
"totals_json": json.dumps({"ytd_smp_recovered": f"{ytd:.2f}", "added_by_this": f"{delta:.2f}", "no_payment": not in_month}, sort_keys=True)}
repo.documents.add([doc], hub._now())
repo.rti.claim(sub, hub._now())
rows = [("eps", sid, {"submission_id": sid})]
if delta:
journal = {"journal_id": str(derived_id(f"prl:{sid}", "eps")), "source_module": "payroll", "source_document": f"prl:{sid}", "event": "eps",
"company_code": company, "journal_date": last.isoformat(), "currency": "GBP", "memo": f"EPS {sid}",
"lines": [{"purpose": "paye_ni_payable", ("debit" if delta > 0 else "credit"): f"{abs(delta):.2f}"},
{"purpose": "statutory_pay_recovery", ("credit" if delta > 0 else "debit"): f"{abs(delta):.2f}"}]}
rows.append(("gl_eps", sid, journal))
repo.outbox.add_many(rows, sid, hub._now())
hub._drain(sid, who)
with hub.repository() as repo:
return {"submission": _view(repo.rti.get(sid)), "sent": True, "journal": bool(delta)}
def deliver_eps(row: dict, who: str) -> dict:
sid = row["ref"]
with hub.repository() as repo:
sub = repo.rti.get(sid)
scheme = repo.config.paye_scheme_of(sub["paye_ref"])
last = tax_month_dates(sub["tax_year"], sub["tax_period"])[1]
if sub["status"] == "pending":
answer = _hand_over(sub, last.isoformat(), scheme["accounts_office_ref"] if scheme else "", "eps")
else:
answer = {"status": sub["status"]}
with hub.repository() as repo, hub._decide():
fresh = repo.rti.get(sid)
if fresh["status"] == "pending":
fresh = _apply(repo, fresh, answer, sub["due_on"], hub._today(repo))
_outbox(repo, row, {"accepted": "sent", "manual": "sent", "rejected": "refused", "pending": "pending"}[fresh["status"]], fresh["message"] or "")
return {"ref": sid, "status": {"accepted": "accepted", "manual": "manual task", "rejected": "rejected", "pending": "retry"}[fresh["status"]]}
def deliver_eps_journal(row: dict, who: str) -> dict:
out = seams.post_journal(json.loads(row["payload_json"]), who)
with hub.repository() as repo:
why = "" if out["ok"] else f"{out['gate'] + ': ' if out['gate'] else ''}{out['reason']}"
_outbox(repo, row, "sent" if out["ok"] else ("refused" if not out["retry"] else "pending"), why)
return {"ref": row["ref"], "status": "posted" if out["ok"] else ("refused" if not out["retry"] else "retry"), **({"reason": why} if why else {})}
# ── the admin's doors ───────────────────────────────────────────────────────────────────────
@PRL.use_case(kind="command", route="POST /pay-runs/{pay_run_id}/fps/resend", permission=FILE_WRITE)
def resend_fps(caller: dict, pay_run_id: Ref) -> dict:
"""HMRC rejected the run's latest FPS: send the corrected one, n+1 (the run stays approved until one is accepted). Built again from the committed run,
handed over from a pending row after this commit. Refused while the latest FPS is accepted, waiting, or a manual task, or before approval."""
who = hub._who(caller)
with hub.repository() as repo, hub._decide():
run = hub._run(repo, pay_run_id, lines=False)
if run.status not in pr.FROZEN:
raise RuleBroken(f"{pay_run_id} is {run.status}: an FPS exists only after approval", gate="state")
fps = sorted((s for s in repo.rti.of_run(pay_run_id) if s["kind"] == "fps"), key=lambda s: s["n"])
if not fps or fps[-1]["status"] != "rejected":
raise RuleBroken("only a rejected FPS is corrected: the latest is " + (fps[-1]["status"] if fps else "not built yet"), gate="state")
n = fps[-1]["n"] + 1
repo.claims.once("fps_resend", f"{pay_run_id}:{n}", hub._now(), who)
repo.outbox.add("fps", f"fps:{pay_run_id}:{n}", pay_run_id, {"n": n}, hub._now())
hub._drain(pay_run_id, who)
with hub.repository() as repo:
return {"submissions": [_view(s) for s in repo.rti.of_run(pay_run_id)]}
@PRL.use_case(kind="command", route="POST /rti-submissions/{submission_id}/done", permission=FILE_WRITE)
def complete_manual_submission(caller: dict, submission_id: Ref, reference: Ref) -> dict:
"""A manual task done: with no provider the file was sent by hand (to HMRC, or uploaded to the scheme) and `reference` is what they gave back. The
submission becomes accepted; an FPS moves the run to submitted (record_fps_accepted), and the run takes any later arrow it now qualifies for."""
with hub.repository() as repo, hub._decide():
sub = repo.rti.get(submission_id)
if sub is None:
raise Refused(404, f"no submission {submission_id}")
if sub["status"] != "manual":
raise RuleBroken(f"{submission_id} is {sub['status']}: only a manual task is completed by hand", gate="state")
repo.claims.once("manual_done", submission_id, hub._now(), hub._who(caller), reference)
sub = repo.rti.get(submission_id)
repo.rti.save({**sub, "status": "accepted", "correlation_id": reference, "message": f"sent by hand ({hub._who(caller)}): {reference}"}, hub._now())
if sub["pay_run_id"]:
run = repo.runs.get(sub["pay_run_id"], lines=False)
advance_run(repo, run)
done = repo.rti.get(submission_id)
return _view(done)
@PRL.use_case(kind="query", route="GET /rti-submissions", permission=RTI_READ)
def list_rti_submissions(caller: dict, pay_run_id: Optional[Ref] = None) -> dict:
"""The FPS, EPS and pension files of a run (or the latest of every kind): each with its status, HMRC's correlation id or errors, the file's object key and
sha256, and its deadline - including the manual tasks still waiting for a person."""
with hub.repository() as repo:
if pay_run_id:
rows = repo.rti.of_run(pay_run_id)
else:
rows = repo.rti.waiting()
return {"submissions": [_view(s) for s in rows]}