File size: 2,413 Bytes
d787a09 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 | """Schema for the process state machine (SPEC §2).
A flow is a namespaced set of Step nodes. Each Step is curated, cited data — not
generated — which is what makes the roadmap hallucination-proof. Steps may be
ordinary process steps or conditional sub-flows (``is_subflow: true``).
"""
from __future__ import annotations
from typing import List, Optional
from pydantic import BaseModel, Field
class RequiredForm(BaseModel):
form_id: str = Field(..., description="Catalog code/ID, or 'verify-against-source'.")
name: str
must_contain: List[str] = Field(default_factory=list)
source: str
verify: bool = Field(
False, description="True if the ID/details still need human verification."
)
class Deadline(BaseModel):
description: str
basis: str = ""
citation: str = ""
verify: bool = False
class Transition(BaseModel):
to: str
when: Optional[str] = Field(
None, description="Condition label; None = default/next transition."
)
label: Optional[str] = None
class HelpTrigger(BaseModel):
when: str
to: str
class Step(BaseModel):
id: str
title: str
summary: str
is_subflow: bool = False
trigger: Optional[str] = None # for sub-flows: the condition that opens it
preconditions: List[str] = Field(default_factory=list)
required_actions: List[str] = Field(default_factory=list)
required_forms: List[RequiredForm] = Field(default_factory=list)
deadlines: List[Deadline] = Field(default_factory=list)
who_to_contact: Optional[str] = None
citations: List[str] = Field(default_factory=list)
transitions: List[Transition] = Field(default_factory=list)
help_triggers: List[HelpTrigger] = Field(default_factory=list)
verify: bool = Field(
False, description="Step-level flag: some specifics need verification."
)
class Flow(BaseModel):
id: str
namespace: str
title: str
description: str
jurisdiction: str
start: str
citations: List[str] = Field(default_factory=list)
steps: List[Step]
def step_map(self) -> dict:
return {s.id: s for s in self.steps}
def main_line(self) -> List[Step]:
"""The non-subflow steps, in declared order (the linear spine)."""
return [s for s in self.steps if not s.is_subflow]
def subflows(self) -> List[Step]:
return [s for s in self.steps if s.is_subflow]
|