Workflow State Store

Server-side workflow state storage implementations

WorkflowStateStore Protocol

The WorkflowStateStore protocol defines the interface for persisting workflow state. This allows patterns like StepFlow to work with different storage backends (in-memory, file-based, database) without being coupled to any specific implementation.

Implementations receive the sess (session) object to extract a session identifier for scoping state to individual users.

---

WorkflowStateStore

def WorkflowStateStore(
    args:VAR_POSITIONAL, kwargs:VAR_KEYWORD
):

Protocol for workflow state storage backends.

Session ID Extraction

The state store needs a way to identify individual users/sessions. We store a small UUID in the session object itself (the only thing we store there) and use it as a key for server-side storage.

---

set_session_id

def set_session_id(
    sess:Any, # FastHTML session object
    session_id:str, # Workflow session ID to set as active
    key:str='_workflow_session_id', # Session key for storing the ID
)->None:

Set the active workflow session ID, switching all subsequent state reads/writes to that session.

---

get_session_id

def get_session_id(
    sess:Any, # FastHTML session object
    key:str='_workflow_session_id', # Session key for storing the ID
)->str: # Stable session identifier

Get or create a stable session identifier.

InMemoryWorkflowStateStore

A simple in-memory implementation of WorkflowStateStore suitable for development and testing. State is lost when the server restarts.

For production use, this can be replaced with a file-based or database-backed implementation that follows the same protocol.

---

InMemoryWorkflowStateStore

def InMemoryWorkflowStateStore(
    
):

In-memory workflow state storage for development and testing.

Usage Example

# Create a store instance
store = InMemoryWorkflowStateStore()

# Mock session object (in real usage, this comes from FastHTML)
mock_sess = {}

# Store and retrieve state
store.set_current_step("my_workflow", mock_sess, "step_1")
print(f"Current step: {store.get_current_step('my_workflow', mock_sess)}")

store.update_state("my_workflow", mock_sess, {"name": "Alice", "email": "alice@example.com"})
print(f"State: {store.get_state('my_workflow', mock_sess)}")

# Session only contains the session ID
print(f"Session contents: {mock_sess}")

Current step: step_1
State: {'name': 'Alice', 'email': 'alice@example.com'}
Session contents: {'_workflow_session_id': 'c51f6c6f-f753-403c-a92a-3ae4e4275ba1'}

# Verify it satisfies the protocol
assert isinstance(store, WorkflowStateStore), "InMemoryWorkflowStateStore must implement WorkflowStateStore protocol"
print("Protocol check passed!")

Protocol check passed!

# set_session_id: switches the active workflow session key in the HTTP session dict.
sess = {}

# First call mints a new UUID.
first = get_session_id(sess)
assert first == sess["_workflow_session_id"]

# Switch to an explicit session ID.
set_session_id(sess, "explicit-session-42")
assert get_session_id(sess) == "explicit-session-42"
assert sess["_workflow_session_id"] == "explicit-session-42"

# Switch again — should overwrite, not append.
set_session_id(sess, "another-session")
assert get_session_id(sess) == "another-session"

# State-store reads now resolve against the new session ID.
other_store = InMemoryWorkflowStateStore()
set_session_id(sess, "session-A")
other_store.update_state("flow1", sess, {"marker": "A"})
set_session_id(sess, "session-B")
other_store.update_state("flow1", sess, {"marker": "B"})

set_session_id(sess, "session-A")
assert other_store.get_state("flow1", sess) == {"marker": "A"}
set_session_id(sess, "session-B")
assert other_store.get_state("flow1", sess) == {"marker": "B"}

# Custom key argument is respected.
alt = {}
set_session_id(alt, "custom", key="_other_key")
assert alt["_other_key"] == "custom"
assert "_workflow_session_id" not in alt

print("set_session_id tests passed!")