Finite State Machine (FSM)

../_images/job_fsm.svg

Many of our Model classes have a FSM over top of their state. Our FSM allows you to annotate any transition with arbitrary Python function(s) to implement any additional side effect.

In addition, it provides timestamps and any user-comment alongside the transition to provide detailed timeline information. We can also report metrics: how many records in what state at what time, along with overall counts.

class lbn.aidoc.fsm.BaseMachine

Bases: StateMachine

a state machine supporting history and comments

__init__(model, session=None)
active() bool

Lifecycle Check: Returns True if the entity is currently in a live operational state (not terminal/final).

property completed: datetime | None

Discovery: Returns the timestamp if we are currently in a ‘Final’ state.

property created: datetime | None

Discovery: Timestamp of the first heartbeat in history.

property current_state_value

Surgical Binding: The Enum IS the string. Since StyledEnum.__new__ sets the string value to ‘clean_value’, we just return the string representation.

duration(start_state=None, end_state=None) timedelta

Calculates time spent. If final state: difference between first and last history records. Otherwise: difference between now and start.

id = 'basemachine'

Lowercase version of name (e.g. "trafficlightmachine").

property journal: list

The ‘Narrative’: All comments for this entity.

log(note: str)

Notarises an event in the model’s history without changing state. Ensures the audit trail captures the ‘Why’ behind a refresh.

classmethod metrics(session, unit='day', states_filter=None)

Generic FSM Engine: Computes real-time snapshot AND time-series historical run-rates using exact state casing definitions.

model_attribute = 'state'
classmethod model_name() str

Surgical: Derives the target model name from the machine’s name.

name = 'BaseMachine'

The class name of the state machine (e.g. "TrafficLightMachine").

on_comment(content, category='note')

Generic Narrative: Logs to the Comment table.

on_transition(transition, source, target, timestamp=None)

Standardised Audit: Logs every move to the History table.

prepare = <statemachine.callbacks.SpecListGrouper object>
previous_state() State

Looks backward through the model’s history to find the state prior to stalled.

start(note: str = 'Initial intake', timestamp=None)

Surgically anchors the model’s current state as the lifecycle Big Bang.

states = []

Collection of top-level State objects declared on this class.

states_map = {}

Mapping from each state’s value to the corresponding State instance. Includes states at all nesting levels (compound children, parallel regions, etc.).

property timeline: list

The ‘Logbook’: Merged and sorted state history.

transitions() list[str]

The Harvester: Manually crawls every state to aggregate unique trigger identifiers (verbs).

undo(note: str = 'FSM Undo: Reverted last transition') str | None

Surgical Rollback: Retrieves the most recent transition log from the History table, restores the model’s previous state, and wipes the audit record.

class lbn.aidoc.fsm.CompanyMachine

Bases: BaseMachine

activate
active
dead
final_states = [State('Dead', id='dead', value=<EntityStatus.DEAD: 'dead'>, initial=False, final=True, parallel=False)]

List of top-level State objects marked as final.

id = 'companymachine'

Lowercase version of name (e.g. "trafficlightmachine").

inactivate
inactive
kill
name = 'CompanyMachine'

The class name of the state machine (e.g. "TrafficLightMachine").

prepare = <statemachine.callbacks.SpecListGrouper object>
states_map = {EntityStatus.ACTIVE: State('Active', id='active', value=<EntityStatus.ACTIVE: 'active'>, initial=True, final=False, parallel=False), EntityStatus.DEAD: State('Dead', id='dead', value=<EntityStatus.DEAD: 'dead'>, initial=False, final=True, parallel=False), EntityStatus.INACTIVE: State('Inactive', id='inactive', value=<EntityStatus.INACTIVE: 'inactive'>, initial=False, final=False, parallel=False)}

Mapping from each state’s value to the corresponding State instance. Includes states at all nesting levels (compound children, parallel regions, etc.).

class lbn.aidoc.fsm.InterviewMachine

Bases: BaseMachine

__init__(model: Interview, session=None)
cancel
cancelled
completed
confirm
final_states = [State('Completed', id='completed', value=<MeetingStatus.COMPLETED: 'completed'>, initial=False, final=True, parallel=False), State('Cancelled', id='cancelled', value=<MeetingStatus.CANCELLED: 'cancelled'>, initial=False, final=True, parallel=False)]

List of top-level State objects marked as final.

finish
id = 'interviewmachine'

Lowercase version of name (e.g. "trafficlightmachine").

miss
missed
name = 'InterviewMachine'

The class name of the state machine (e.g. "TrafficLightMachine").

prepare = <statemachine.callbacks.SpecListGrouper object>
proposed
reschedule
rescheduling
scheduled
states_map = {MeetingStatus.CANCELLED: State('Cancelled', id='cancelled', value=<MeetingStatus.CANCELLED: 'cancelled'>, initial=False, final=True, parallel=False), MeetingStatus.COMPLETED: State('Completed', id='completed', value=<MeetingStatus.COMPLETED: 'completed'>, initial=False, final=True, parallel=False), MeetingStatus.MISSED: State('Missed', id='missed', value=<MeetingStatus.MISSED: 'missed'>, initial=False, final=False, parallel=False), MeetingStatus.PROPOSED: State('Proposed', id='proposed', value=<MeetingStatus.PROPOSED: 'proposed'>, initial=True, final=False, parallel=False), MeetingStatus.RESCHEDULING: State('Rescheduling', id='rescheduling', value=<MeetingStatus.RESCHEDULING: 'rescheduling'>, initial=False, final=False, parallel=False), MeetingStatus.SCHEDULED: State('Scheduled', id='scheduled', value=<MeetingStatus.SCHEDULED: 'scheduled'>, initial=False, final=False, parallel=False)}

Mapping from each state’s value to the corresponding State instance. Includes states at all nesting levels (compound children, parallel regions, etc.).

class lbn.aidoc.fsm.PersonMachine

Bases: BaseMachine

activate
active
age
dead
engage
final_states = [State('Dead', id='dead', value=<EntityStatus.DEAD: 'dead'>, initial=False, final=True, parallel=False)]

List of top-level State objects marked as final.

ghost
ghosted
id = 'personmachine'

Lowercase version of name (e.g. "trafficlightmachine").

inactivate
inactive
kill
lose
lost
name = 'PersonMachine'

The class name of the state machine (e.g. "TrafficLightMachine").

prepare = <statemachine.callbacks.SpecListGrouper object>
responsive
stale
states_map = {EntityStatus.ACTIVE: State('Active', id='active', value=<EntityStatus.ACTIVE: 'active'>, initial=True, final=False, parallel=False), EntityStatus.DEAD: State('Dead', id='dead', value=<EntityStatus.DEAD: 'dead'>, initial=False, final=True, parallel=False), EntityStatus.GHOSTED: State('Ghosted', id='ghosted', value=<EntityStatus.GHOSTED: 'ghosted'>, initial=False, final=False, parallel=False), EntityStatus.INACTIVE: State('Inactive', id='inactive', value=<EntityStatus.INACTIVE: 'inactive'>, initial=False, final=False, parallel=False), EntityStatus.LOST: State('Lost', id='lost', value=<EntityStatus.LOST: 'lost'>, initial=False, final=False, parallel=False), EntityStatus.RESPONSIVE: State('Responsive', id='responsive', value=<EntityStatus.RESPONSIVE: 'responsive'>, initial=False, final=False, parallel=False), EntityStatus.STALE: State('Stale', id='stale', value=<EntityStatus.STALE: 'stale'>, initial=False, final=False, parallel=False)}

Mapping from each state’s value to the corresponding State instance. Includes states at all nesting levels (compound children, parallel regions, etc.).

unengage
class lbn.aidoc.fsm.TaskMachine

Bases: BaseMachine

The Operational Heart: Unified FSM for the Task/Todo lifecycle.

__init__(model: Task, session=None)
active
block
blocked
complete
done
final_states = [State('Done', id='done', value=<TaskStatus.DONE: 'done'>, initial=False, final=True, parallel=False), State('Ignored', id='ignored', value=<TaskStatus.IGNORED: 'ignored'>, initial=False, final=True, parallel=False)]

List of top-level State objects marked as final.

id = 'taskmachine'

Lowercase version of name (e.g. "trafficlightmachine").

ignore
ignored
name = 'TaskMachine'

The class name of the state machine (e.g. "TrafficLightMachine").

new
prepare = <statemachine.callbacks.SpecListGrouper object>
resume
start
states_map = {TaskStatus.ACTIVE: State('Active', id='active', value=<TaskStatus.ACTIVE: 'active'>, initial=False, final=False, parallel=False), TaskStatus.BLOCKED: State('Blocked', id='blocked', value=<TaskStatus.BLOCKED: 'blocked'>, initial=False, final=False, parallel=False), TaskStatus.DONE: State('Done', id='done', value=<TaskStatus.DONE: 'done'>, initial=False, final=True, parallel=False), TaskStatus.IGNORED: State('Ignored', id='ignored', value=<TaskStatus.IGNORED: 'ignored'>, initial=False, final=True, parallel=False), TaskStatus.NEW: State('New', id='new', value=<TaskStatus.NEW: 'new'>, initial=True, final=False, parallel=False)}

Mapping from each state’s value to the corresponding State instance. Includes states at all nesting levels (compound children, parallel regions, etc.).

lbn.aidoc.fsm.view(machines)