Coverage for domain/lifecycle.py: 100%
19 statements
« prev ^ index » next coverage.py v7.17.0, created at 2026-11-04 17:24 +0000
« prev ^ index » next coverage.py v7.17.0, created at 2026-11-04 17:24 +0000
1"""Loan lifecycle state machine, ported from ``arive-spec/src/domain/lifecycle.ts``.
3Pure transition rules — no Django, no HTTP. The status × role read-only gating from
4the oracle (``canEditLoan``) depends on RBAC and lands with the auth slice; this module
5covers the stage transition graph.
6"""
8from __future__ import annotations
10from .enums import LoanStage, PriorToCategory
11from .result import AppError, Result, err, ok
13# Allowed forward/alternate transitions (spec §3.1).
14TRANSITIONS: dict[LoanStage, tuple[LoanStage, ...]] = {
15 LoanStage.APPLICATION_INTAKE: (
16 LoanStage.PREAPPROVED, LoanStage.QUALIFICATION, LoanStage.ADVERSE, LoanStage.SUSPENDED,
17 ),
18 LoanStage.PREAPPROVED: (
19 LoanStage.QUALIFICATION, LoanStage.LOAN_SETUP, LoanStage.ADVERSE, LoanStage.SUSPENDED,
20 ),
21 LoanStage.QUALIFICATION: (LoanStage.LOAN_SETUP, LoanStage.ADVERSE, LoanStage.SUSPENDED),
22 LoanStage.LOAN_SETUP: (LoanStage.DISCLOSURE_SENT, LoanStage.ADVERSE, LoanStage.SUSPENDED),
23 LoanStage.DISCLOSURE_SENT: (
24 LoanStage.UNDERWRITING_SUBMITTED, LoanStage.ADVERSE, LoanStage.SUSPENDED,
25 ),
26 LoanStage.UNDERWRITING_SUBMITTED: (
27 LoanStage.APPROVED_WITH_CONDITION, LoanStage.RE_SUBMITTAL, LoanStage.ADVERSE,
28 LoanStage.SUSPENDED,
29 ),
30 LoanStage.APPROVED_WITH_CONDITION: (
31 LoanStage.CLEAR_TO_CLOSE, LoanStage.RE_SUBMITTAL, LoanStage.ADVERSE, LoanStage.SUSPENDED,
32 ),
33 LoanStage.RE_SUBMITTAL: (
34 LoanStage.APPROVED_WITH_CONDITION, LoanStage.ADVERSE, LoanStage.SUSPENDED,
35 ),
36 LoanStage.CLEAR_TO_CLOSE: (LoanStage.DOCS_OUT, LoanStage.ADVERSE, LoanStage.SUSPENDED),
37 LoanStage.DOCS_OUT: (LoanStage.DOCS_SIGNED, LoanStage.ADVERSE, LoanStage.SUSPENDED),
38 LoanStage.DOCS_SIGNED: (LoanStage.LOAN_FUNDED, LoanStage.ADVERSE, LoanStage.SUSPENDED),
39 LoanStage.LOAN_FUNDED: (LoanStage.COMMISSION_PAID, LoanStage.BROKER_CHECK_RECEIVED),
40 LoanStage.COMMISSION_PAID: (LoanStage.BROKER_CHECK_RECEIVED,),
41 LoanStage.BROKER_CHECK_RECEIVED: (),
42 LoanStage.ADVERSE: (LoanStage.APPLICATION_INTAKE,), # reinstate
43 LoanStage.SUSPENDED: (LoanStage.APPLICATION_INTAKE, LoanStage.LOAN_SETUP),
44}
46# Stages with no outgoing transition.
47TERMINAL: frozenset[LoanStage] = frozenset(s for s, nexts in TRANSITIONS.items() if not nexts)
49# Which PriorToCategory's open conditions block a given transition (design.md D3 of
50# add-conditions-and-client-needs — this interpretation of the 4 category names against
51# this stage graph has no other prior art in either codebase).
52CONDITION_GATES: dict[tuple[LoanStage, LoanStage], PriorToCategory] = {
53 (LoanStage.UNDERWRITING_SUBMITTED, LoanStage.APPROVED_WITH_CONDITION): (
54 PriorToCategory.PRIOR_TO_APPROVAL
55 ),
56 (LoanStage.APPROVED_WITH_CONDITION, LoanStage.CLEAR_TO_CLOSE): PriorToCategory.PRIOR_TO_DOCS,
57 (LoanStage.CLEAR_TO_CLOSE, LoanStage.DOCS_OUT): PriorToCategory.PRIOR_TO_CLOSING_DOCS,
58 (LoanStage.DOCS_SIGNED, LoanStage.LOAN_FUNDED): PriorToCategory.PRIOR_TO_FUNDING,
59}
62def can_transition(frm: LoanStage, to: LoanStage) -> bool:
63 return to in TRANSITIONS[frm]
66def valid_next_stages(frm: LoanStage) -> list[LoanStage]:
67 return list(TRANSITIONS[frm])
70def assert_transition(
71 frm: LoanStage,
72 to: LoanStage,
73 open_prior_to_categories: frozenset[str] = frozenset(),
74) -> Result:
75 if frm == to:
76 return ok(to)
77 if not can_transition(frm, to):
78 return err(
79 AppError(
80 code="INVALID_TRANSITION",
81 message=f"Loan cannot move from {frm} to {to}.",
82 )
83 )
84 gate = CONDITION_GATES.get((frm, to))
85 if gate is not None and gate.value in open_prior_to_categories:
86 return err(
87 AppError(
88 code="INVALID_TRANSITION",
89 message=(
90 f"Loan cannot move from {frm} to {to}: "
91 f"an open {gate.value} condition must be cleared or waived first."
92 ),
93 )
94 )
95 return ok(to)