# Modeling Sponsored Programs approval workflows in Python

An internal Sponsored Programs approval workflow is deceptively simple to describe and dangerous to build informally: a proposal needs the principal investigator (PI), the department chair, the Sponsored Programs Office (SPO), and the Authorized Organization Representative (AOR) to sign off, in that order, before the institution will release it to Grants.gov. The danger is that “in that order” is a correctness property, and a workflow encoded in ticket statuses or a shared inbox has no way to enforce it — a chair approves a draft the SPO later rejects, an approval lands after the internal deadline, and the record of who authorized the release is unreconstructable. This guide builds that workflow as a concrete, testable finite state machine (FSM) in Python, the deepest layer of institutional approval routing. The output is a machine whose every accepted move appends a timestamped entry — the raw material that building tamper-evident audit logs for grant submissions turns into an immutable chain, and whose terminal state hands a fully-approved package to submission portal sync.

# Phase 1 — Enumerate states, actors, and transitions

Before any code, name the three sets that define the machine. Guessing them later is how workflows accrete illegal shortcuts.

1. States. The milestones a proposal passes through: DRAFT, PI_REVIEW, CHAIR_REVIEW, SPO_REVIEW, AOR_QUEUE, and the terminals SUBMITTED and REJECTED. A state names a place in the route, not an action.
2. Actors and their authority. Each non-terminal state is owned by exactly one role that may act in it — pi, chair, spo, aor — and only the AOR role may cause the SUBMITTED transition. That mapping is the legal core of the workflow: the AOR is the only party the sponsor recognizes as authorized to submit.
3. Transitions. The directed edges between states, written as a table so the route is auditable at a glance. Any edge not in the table is illegal by definition. The rework edge (SPO_REVIEW back to DRAFT) and the reject edges are as much a part of the design as the forward path.

Set up the environment with Pydantic v2 for the typed records:

python -m venv .venv
source .venv/bin/activate
pip install "pydantic>=2.6"

With the three sets fixed, write them down as enumerations and a transition map. This differs from a status field precisely because the map, not the caller, decides what is reachable:

from enum import Enum

class WFState(str, Enum):
    DRAFT = "draft"
    PI_REVIEW = "pi_review"
    CHAIR_REVIEW = "chair_review"
    SPO_REVIEW = "spo_review"
    AOR_QUEUE = "aor_queue"
    SUBMITTED = "submitted"    # terminal
    REJECTED = "rejected"      # terminal

# forward edges of the route; REJECTED is reachable from every active state,
# added programmatically below so the table stays readable.
EDGES: dict[WFState, set[WFState]] = {
    WFState.DRAFT:        {WFState.PI_REVIEW},
    WFState.PI_REVIEW:    {WFState.CHAIR_REVIEW},
    WFState.CHAIR_REVIEW: {WFState.SPO_REVIEW},
    WFState.SPO_REVIEW:   {WFState.AOR_QUEUE, WFState.DRAFT},  # clear or rework
    WFState.AOR_QUEUE:    {WFState.SUBMITTED},
}
for src in (WFState.DRAFT, WFState.PI_REVIEW, WFState.CHAIR_REVIEW,
            WFState.SPO_REVIEW, WFState.AOR_QUEUE):
    EDGES[src].add(WFState.REJECTED)  # any active state may be rejected

# Phase 2 — Implement the machine with typed guards and timestamps

The machine advances only through a single apply method, and that method is where role, ordering, and deadline guards all live. Centralizing the guards means there is exactly one path that can mutate state, so there is exactly one place that appends to the audit trail. Each accepted move produces an immutable Transition record carrying the actor, the timestamp, and an optional note.

from datetime import datetime, timezone
from pydantic import BaseModel, Field, field_validator

class WorkflowError(Exception): ...
class IllegalMove(WorkflowError): ...
class UnauthorizedActor(WorkflowError): ...
class DeadlinePassed(WorkflowError): ...

ROLE_FOR: dict[WFState, str] = {
    WFState.PI_REVIEW: "pi", WFState.CHAIR_REVIEW: "chair",
    WFState.SPO_REVIEW: "spo", WFState.AOR_QUEUE: "spo",
    WFState.SUBMITTED: "aor",
}

class Transition(BaseModel):
    src: WFState
    dst: WFState
    actor: str
    role: str
    at: datetime
    on_behalf_of: str | None = None   # set when a delegate acts
    note: str | None = None

    @field_validator("at")
    @classmethod
    def _aware(cls, v: datetime) -> datetime:
        if v.tzinfo is None:                    # naive time is unauditable
            raise ValueError("timestamp must be timezone-aware")
        return v

class Workflow(BaseModel):
    proposal_id: str
    state: WFState = WFState.DRAFT
    internal_deadline: datetime
    log: list[Transition] = Field(default_factory=list)

    def apply(self, dst: WFState, actor: str, role: str, *,
              on_behalf_of: str | None = None, note: str | None = None) -> None:
        if dst not in EDGES.get(self.state, set()):        # ordering guard
            raise IllegalMove(f"{self.state.value} -> {dst.value} not permitted")
        need = ROLE_FOR.get(dst)
        if need and role != need:                          # role guard
            raise UnauthorizedActor(f"{dst.value} needs {need!r}, got {role!r}")
        now = datetime.now(timezone.utc)
        forward = dst not in (WFState.REJECTED, WFState.DRAFT)
        if forward and now > self.internal_deadline:       # deadline guard
            raise DeadlinePassed(f"internal deadline elapsed for {self.proposal_id}")
        self.log.append(Transition(src=self.state, dst=dst, actor=actor,
                                   role=role, at=now,
                                   on_behalf_of=on_behalf_of, note=note))
        self.state = dst                                   # commit last

The ordering matters: guards run first and raise before any mutation, so a rejected attempt never leaves a half-applied state or a spurious log entry. State is committed only after the Transition is appended, which guarantees the log and the live state can never disagree. The diagram below shows the full machine with each edge annotated by the guard that gates it.

# Phase 3 — Edge cases and agency-specific overrides

A machine that only handles the happy path is a demo. Real Sponsored Programs routing has to absorb five recurring complications, and each maps to a specific extension of the model above.

• Recall after approval. A PI discovers a budget error after the chair has already approved. Because the chain is serial, recall is a transition back to DRAFT that carries a reason and preserves the prior approvals in the log — the history shows the approval and its recall, never a silent rewind. Only states before AOR_QUEUE may be recalled; once the AOR has released, the submission is frozen.
• Delegation. An unavailable chair delegates authority. The delegate calls apply with their own actor id, the required role, and on_behalf_of set to the delegating official. The Transition record then names both parties, so the audit trail distinguishes a delegated approval from an impersonated one — the two must never look alike.
• Deadline auto-escalation. When the internal deadline passes with the proposal still in flight, the deadline guard blocks a forward move. Rather than dead-ending, catch DeadlinePassed and route the proposal to an escalation handler that notifies the SPO director; any override they grant is itself an apply call and therefore a logged step.
• Multi-PI co-review. A proposal with two co-PIs needs both certifications before it may leave PI_REVIEW. Model this by requiring a quorum of certification steps for that state rather than a single one — hold the transition until every required signer has acted, which the snippet below expresses as a guard over the log.
• Agency-specific AOR rules. The AOR credential differs across the National Institutes of Health (NIH), the National Science Foundation (NSF), and the Department of Defense (DoD), and one of them adds a gate the others lack.

The Department of Defense (DoD) override is handled by inserting an EXPORT_REVIEW state between SPO_REVIEW and AOR_QUEUE for Broad Agency Announcement (BAA) proposals — loaded from the agency profile, not hard-coded — so the International Traffic in Arms Regulations (ITAR) and Export Administration Regulations (EAR) determination becomes a required, logged transition. A quorum guard for the multi-PI case reads the log directly:

def pi_quorum_met(wf: Workflow, required_pis: set[str]) -> bool:
    """True once every required co-PI has recorded a certification step."""
    certified = {t.actor for t in wf.log
                 if t.dst is WFState.CHAIR_REVIEW and t.role == "pi"}
    return required_pis <= certified   # subset: all required PIs have acted

# Phase 4 — Validation and verification

The machine is only trustworthy if its guarantees are tested: no illegal transition survives, a full run emits a complete audit trail, and the deadline guard actually fires. Drive the tests with constructed workflows so they never depend on a live institutional system.

import pytest
from datetime import datetime, timedelta, timezone

def _wf(state: WFState = WFState.DRAFT, days: int = 3) -> Workflow:
    return Workflow(proposal_id="NSF-2026-CAREER-7",
                    state=state,
                    internal_deadline=datetime.now(timezone.utc) + timedelta(days=days))

def test_no_illegal_transition() -> None:
    wf = _wf(WFState.PI_REVIEW)
    with pytest.raises(IllegalMove):
        wf.apply(WFState.AOR_QUEUE, "aor-1", "aor")   # skips chair + SPO
    assert wf.state is WFState.PI_REVIEW and wf.log == []

def test_full_run_emits_ordered_audit_trail() -> None:
    wf = _wf()
    wf.apply(WFState.PI_REVIEW, "pi-1", "pi")
    wf.apply(WFState.CHAIR_REVIEW, "chair-2", "chair")
    wf.apply(WFState.SPO_REVIEW, "spo-3", "spo")
    wf.apply(WFState.AOR_QUEUE, "spo-3", "spo")
    wf.apply(WFState.SUBMITTED, "aor-9", "aor")
    assert wf.state is WFState.SUBMITTED
    assert [t.dst for t in wf.log] == [
        WFState.PI_REVIEW, WFState.CHAIR_REVIEW, WFState.SPO_REVIEW,
        WFState.AOR_QUEUE, WFState.SUBMITTED]
    assert all(t.at.tzinfo is not None for t in wf.log)  # audit-safe stamps

def test_deadline_guard_fires_on_forward_move() -> None:
    wf = _wf(WFState.PI_REVIEW, days=-1)               # deadline already past
    with pytest.raises(DeadlinePassed):
        wf.apply(WFState.CHAIR_REVIEW, "chair-2", "chair")

def test_delegation_records_both_identities() -> None:
    wf = _wf(WFState.PI_REVIEW)
    wf.apply(WFState.CHAIR_REVIEW, "vice-chair-5", "chair",
             on_behalf_of="chair-2")
    step = wf.log[-1]
    assert step.actor == "vice-chair-5" and step.on_behalf_of == "chair-2"

Pair the suite with a review checklist: confirm the recall path preserves prior approvals, that a delegated step is visually distinct from a direct one in the exported log, that the DoD export gate is present for BAA profiles, and that the terminal SUBMITTED state is unreachable without every upstream approval. Only then hand the workflow’s history to the audit layer and its approved package to the portal, closing the same loop the parent section on institutional approval routing describes.

# Frequently asked questions

# Related

• Building tamper-evident audit logs for grant submissions — turns this workflow’s append-only log into a verifiable chain.
• Audit logging and provenance — the section that consumes the routing history.
• Submission portal sync — receives the fully-approved package the terminal state releases.
• Agency form population — builds the SF424 package that enters the workflow as a draft.

Up one level: Institutional Approval Routing