Code clean up: Checkpoint and WorkflowBuilder (#1557)

Co-authored-by: Evan Mattson <35585003+moonbox3@users.noreply.github.com>

python/samples/getting_started/workflows/checkpoint/checkpoint_with_human_in_the_loop.py

@@ -140,8 +140,8 @@ class ReviewGateway(Executor):
         # persist iterations. The `RequestInfoExecutor` relies on this state to
         # rehydrate when checkpoints are restored.
         draft = response.agent_run_response.text or ""
-        iteration = int((await ctx.get_state() or {}).get("iteration", 0)) + 1
-        await ctx.set_state({"iteration": iteration, "last_draft": draft})
+        iteration = int((await ctx.get_executor_state() or {}).get("iteration", 0)) + 1
+        await ctx.set_executor_state({"iteration": iteration, "last_draft": draft})
         # Emit a human approval request. Because this flows through
         # RequestInfoExecutor it will pause the workflow until an answer is
         # supplied either interactively or via pre-supplied responses.
@@ -163,7 +163,7 @@ class ReviewGateway(Executor):
         # The RequestResponse wrapper gives us both the human data and the
         # original request message, even when resuming from checkpoints.
         reply = (feedback.data or "").strip()
-        state = await ctx.get_state() or {}
+        state = await ctx.get_executor_state() or {}
         draft = state.get("last_draft") or (feedback.original_request.draft if feedback.original_request else "")
 
         if reply.lower() == "approve":
@@ -175,7 +175,7 @@ class ReviewGateway(Executor):
         # Any other response loops us back to the writer with fresh guidance.
         guidance = reply or "Tighten the copy and emphasise customer benefit."
         iteration = int(state.get("iteration", 1)) + 1
-        await ctx.set_state({"iteration": iteration, "last_draft": draft})
+        await ctx.set_executor_state({"iteration": iteration, "last_draft": draft})
         prompt = (
             "Revise the launch note. Respond with the new copy only.\n\n"
             f"Previous draft:\n{draft}\n\n"
@@ -193,7 +193,7 @@ class FinaliseExecutor(Executor):
     @handler
     async def publish(self, text: str, ctx: WorkflowContext[Any, str]) -> None:
         # Store the output so diagnostics or a UI could fetch the final copy.
-        await ctx.set_state({"published_text": text})
+        await ctx.set_executor_state({"published_text": text})
         # Yield the final output so the workflow completes cleanly.
         await ctx.yield_output(text)
 

python/samples/getting_started/workflows/checkpoint/checkpoint_with_resume.py

@@ -42,7 +42,7 @@ Pipeline:
 5) FinalizeFromAgent yields the final result.
 
 What you learn:
-- How to persist executor state using ctx.get_state and ctx.set_state.
+- How to persist executor state using ctx.get_executor_state and ctx.set_executor_state.
 - How to persist shared workflow state using ctx.set_shared_state for cross-executor visibility.
 - How to configure FileCheckpointStorage and call with_checkpointing on WorkflowBuilder.
 - How to list and inspect checkpoints programmatically.
@@ -73,9 +73,9 @@ class UpperCaseExecutor(Executor):
 
         # Persist executor-local state so it is captured in checkpoints
         # and available after resume for observability or logic.
-        prev = await ctx.get_state() or {}
+        prev = await ctx.get_executor_state() or {}
         count = int(prev.get("count", 0)) + 1
-        await ctx.set_state({
+        await ctx.set_executor_state({
             "count": count,
             "last_input": text,
             "last_output": result,
@@ -122,9 +122,9 @@ class FinalizeFromAgent(Executor):
         result = response.agent_run_response.text or ""
 
         # Persist executor-local state for auditability when inspecting checkpoints.
-        prev = await ctx.get_state() or {}
+        prev = await ctx.get_executor_state() or {}
         count = int(prev.get("count", 0)) + 1
-        await ctx.set_state({
+        await ctx.set_executor_state({
             "count": count,
             "last_output": result,
             "final": True,
@@ -143,9 +143,9 @@ class ReverseTextExecutor(Executor):
         print(f"ReverseTextExecutor: '{text}' -> '{result}'")
 
         # Persist executor-local state so checkpoint inspection can reveal progress.
-        prev = await ctx.get_state() or {}
+        prev = await ctx.get_executor_state() or {}
         count = int(prev.get("count", 0)) + 1
-        await ctx.set_state({
+        await ctx.set_executor_state({
             "count": count,
             "last_input": text,
             "last_output": result,

python/samples/semantic-kernel-migration/processes/fan_out_fan_in_process.py

@@ -144,7 +144,7 @@ async def run_semantic_kernel_process_example() -> None:
         kernel=kernel,
         initial_event=KernelProcessEvent(id=CommonEvents.START_PROCESS.value, data="Initial"),
     ) as process_context:
-        process_state = await process_context.get_state()
+        process_state = await process_context.get_executor_state()
         c_step_state: KernelProcessStepState[CStepState] | None = next(
             (s.state for s in process_state.steps if s.state.name == "CStep"),
             None,

python/samples/semantic-kernel-migration/processes/nested_process.py

@@ -136,7 +136,7 @@ async def run_semantic_kernel_nested_process() -> None:
         initial_event=ProcessEvents.START_PROCESS.value,
         data="Test",
     )
-    process_info = await process_handle.get_state()
+    process_info = await process_handle.get_executor_state()
 
     inner_process: KernelProcess | None = next(
         (s for s in process_info.steps if s.state.name == "Inner"),