Python: Bug: a process step declared as KernelProcessStep[State] is treated as stateless, so initial_state is rejected and its state is not saved #14734

Description

A process step that names its state type only in the base class, like class Counter(KernelProcessStep[CounterState]), is treated as a step without state.

• initial_state=CounterState(...) raises ProcessInvalidConfigurationException: ... the step does not accept a state.
• The step gets no default state.
• State restored from KernelProcessStateMetadata stays a plain dict.
• The state is left out of the saved process state, so it is lost on persistence.

If the step also declares state: CounterState = Field(default_factory=CounterState), all of this works. Most samples declare it, but UserInputStep in step01_processes.py does not, and it works around the missing default in activate.

Reproduction

I ran this on current main (df1cb33), Python 3.13.

import asyncio

from pydantic import Field

from semantic_kernel import Kernel
from semantic_kernel.functions import kernel_function
from semantic_kernel.kernel_pydantic import KernelBaseModel
from semantic_kernel.processes.kernel_process.kernel_process_step import KernelProcessStep
from semantic_kernel.processes.kernel_process.kernel_process_step_state import KernelProcessStepState
from semantic_kernel.processes.local_runtime.local_event import KernelProcessEvent
from semantic_kernel.processes.local_runtime.local_kernel_process import start
from semantic_kernel.processes.process_builder import ProcessBuilder
from semantic_kernel.processes.process_state_metadata_utils import kernel_process_to_process_state_metadata
from semantic_kernel.processes.process_types import get_generic_state_type


class CounterState(KernelBaseModel):
    count: int = 0


class GenericCounter(KernelProcessStep[CounterState]):
    async def activate(self, state: KernelProcessStepState[CounterState]):
        state.state = state.state or CounterState()
        self.state = state.state

    @kernel_function
    async def run(self, value: str) -> str:
        self.state.count += 1
        return value


class DeclaredCounter(KernelProcessStep[CounterState]):
    state: CounterState = Field(default_factory=CounterState)

    async def activate(self, state: KernelProcessStepState[CounterState]):
        state.state = state.state or CounterState()
        self.state = state.state

    @kernel_function
    async def run(self, value: str) -> str:
        self.state.count += 1
        return value


async def main(step_type):
    print(step_type.__name__)
    print("  get_generic_state_type:", get_generic_state_type(step_type))

    builder = ProcessBuilder(name="p")
    try:
        builder.add_step(step_type, initial_state=CounterState(count=5))
        builder.build()
        print("  initial_state=CounterState(count=5): accepted")
    except Exception as ex:
        print("  initial_state=CounterState(count=5): raised", type(ex).__name__ + ":", ex)

    builder = ProcessBuilder(name="p")
    step = builder.add_step(step_type)
    builder.on_input_event("go").send_event_to(step, function_name="run", parameter_name="value")
    context = await start(process=builder.build(), kernel=Kernel(), initial_event=KernelProcessEvent(id="go", data="x"))
    metadata = kernel_process_to_process_state_metadata(await context.get_state())
    print("  state saved after one run:", metadata.steps_state[step_type.__name__].state)


for step_type in (DeclaredCounter, GenericCounter):
    asyncio.run(main(step_type))

Output on main:

DeclaredCounter
  get_generic_state_type: <class '__main__.CounterState'>
  initial_state=CounterState(count=5): accepted
  state saved after one run: count=1
GenericCounter
  get_generic_state_type: None
  initial_state=CounterState(count=5): raised ProcessInvalidConfigurationException: An initial state was provided for step GenericCounter, but the step does not accept a state.
  state saved after one run: None

The two steps have the same code, except that DeclaredCounter repeats the state field. GenericCounter increments its counter in the run, but the saved state is None.

I also built a process with state_metadata that holds {"count": 7} for a step. The declared step gets CounterState(count=7). The generic step gets the dict {'count': 7}.

Cause

get_generic_state_type in python/semantic_kernel/processes/process_types.py reads the type hint of the state attribute. For a step declared as KernelProcessStep[CounterState], the hint is still Optional[TState]. The function turns the TypeVar into None and returns at once. It never looks at the pydantic field annotation, which pydantic has already resolved to Optional[CounterState].

Because the result is None, ProcessStepBuilder.build_step, LocalStep.initialize_step, StepActor and step_info_to_process_state_metadata all treat the step as stateless.

Expected

KernelProcessStep[CounterState] is enough to give the step a state type. A step should behave the same with or without the repeated state field.

Possible fix

I have a branch ready: https://github.com/VANDRANKI/semantic-kernel/tree/fix/process-step-generic-state-type

When the type hint is a TypeVar, get_generic_state_type now reads model_fields["state"].annotation of the class. Plain KernelProcessStep steps still give None.

Once the type is found, build_step also has to handle state types that are not pydantic models. It creates the default state with t_state(). Before, a type without __fields__ put the class itself in the state. A dataclass state failed validation. It also restores a dataclass state from the saved dict. I tested a pydantic state and a dataclass state.

It adds 10 tests: 3 in test_process_types.py and 7 in a new file, tests/unit/processes/test_process_step_generic_state.py. 9 of them fail on main. The 10th is a control: an unparametrized step still gives None. All 249 process unit tests pass on the branch.

I can't open a pull request from this account right now, so the compare link is main...VANDRANKI:semantic-kernel:fix/process-step-generic-state-type