Python: [Feature]: Expose get_last_checkpoint_id() and resolve_pause_checkpoint_id() on FunctionalWorkflow #9603

Problem

Graph-based Workflow exposes get_last_checkpoint_id() and await resolve_pause_checkpoint_id(...) (added in #8163 for #8150) so callers and integrations can reliably identify the checkpoint persisted by a specific workflow run or human-in-the-loop (request_info) pause without relying on storage.get_latest(workflow_name=...) or storage.list_checkpoints(workflow_name=...), which are susceptible to cross-run race conditions when concurrent workflow instances share a CheckpointStorage backend.

However, FunctionalWorkflow (python/packages/core/agent_framework/_workflows/_functional.py) does not expose get_last_checkpoint_id() or resolve_pause_checkpoint_id(). Inside FunctionalWorkflow._run_core(), the checkpoint ID returned by await self._save_checkpoint(ctx, storage, ckpt_chain[0]) on completion (L2199) and on WorkflowInterrupted pause (L2225) is discarded, leaving callers with no instance-level way to retrieve the checkpoint ID persisted by the run.

Use Case and Evidence

1. Resuming paused @workflow runs safely in multi-session applications:
   When a FunctionalWorkflow pauses on await ctx.request_info(...) with checkpoint_storage configured, callers need the checkpoint ID produced by that specific run to persist alongside their session/turn metadata and later pass to workflow.run(checkpoint_id=..., responses=...). Today, even the test suite (tests/workflow/test_functional_workflow.py) has to query (await storage.list_checkpoints(workflow_name="wf"))[-1].checkpoint_id, which can pick up another concurrent caller's checkpoint when instances share storage.

2. First-party AG-UI integration (agent-framework-ag-ui):
   agent_framework_ag_ui/_workflow_run.py dynamically probes getattr(workflow, "get_last_checkpoint_id", None) (L149) and getattr(workflow, "resolve_pause_checkpoint_id", None) (L213) to populate metadata.agent_framework.checkpoint_id on interrupt payloads when a workflow pauses on request_info. Because FunctionalWorkflow lacks both methods, paused @workflow instances running behind AG-UI never emit checkpoint_id in their interrupt metadata.

Relevant locations on main (c3ce250695f71e56c9c3f9bff85fef78934520aa):

• Workflow.get_last_checkpoint_id and Workflow.resolve_pause_checkpoint_id:

  agent-framework/python/packages/core/agent_framework/_workflows/_workflow.py

  Lines 1416 to 1501 in c3ce250

  	def get_last_checkpoint_id(self) -> str | None:
  	"""Return the checkpoint id last persisted or restored by this workflow runner."""
  	checkpoint_id = self._runner._previous_checkpoint_id # pyright: ignore[reportPrivateUsage]
  	return str(checkpoint_id) if checkpoint_id is not None else None
  	async def resolve_pause_checkpoint_id(
  	self,
  	request_ids: Collection[str],
  	*,
  	checkpoint_storage: CheckpointStorage | None = None,
  	known_checkpoint_id: str | None = None,
  	baseline_checkpoint_id: str | object | None = _MISSING,
  	) -> str | None:
  	"""Resolve the persisted pause checkpoint that covers ``request_ids``.
  	Prefers this runner's last-saved id when it advanced past the run baseline
  	(captured automatically at ``run()`` start), over shared ``get_latest(workflow_name=...)``.
  	Storage precedence is the run argument, else the runner's effective
  	(runtime / builder) storage. When storage is available, candidates are accepted
  	only if their ``pending_request_info_events`` cover ``request_ids``.
  	An incoming restored checkpoint id is excluded so a failed pause save after
  	resume cannot re-advertise pre-response state.
  	Args:
  	request_ids: Pending request_info ids that must be present on the checkpoint.
  	checkpoint_storage: Optional storage override for this lookup.
  	known_checkpoint_id: Fallback id (for example a cold-resume short-circuit).
  	baseline_checkpoint_id: Optional override for the pre-run runner id. When omitted,
  	uses the baseline captured by the most recent ``run()``. Pass ``None``
  	explicitly to treat any current runner id as newly advanced.
  	Returns:
  	A checkpoint id suitable for durable resume, or ``None`` when none is safe.
  	"""
  	ids = {str(request_id) for request_id in request_ids if request_id}
  	if not ids:
  	return None
  	if baseline_checkpoint_id is _MISSING:
  	baseline_checkpoint_id = self._run_baseline_checkpoint_id
  	# Prefer explicit storage; otherwise load via public RunnerContext APIs
  	# (Protocol has no private `_get_effective_checkpoint_storage`).
  	storage = checkpoint_storage
  	use_context_storage = storage is None and self._runner.context.has_checkpointing()
  	current = self.get_last_checkpoint_id()
  	excluded: set[str] = set()
  	if self._restored_checkpoint_id is not None:
  	excluded.add(self._restored_checkpoint_id)
  	# Run-scoped: do not advertise a pre-run leftover when this run did not persist.
  	runner_candidate: str | None = None
  	if current is not None and current != baseline_checkpoint_id and current not in excluded:
  	runner_candidate = current
  	candidates: list[str] = []
  	if runner_candidate is not None:
  	candidates.append(runner_candidate)
  	if (
  	known_checkpoint_id is not None
  	and known_checkpoint_id not in candidates
  	and known_checkpoint_id not in excluded
  	):
  	candidates.append(str(known_checkpoint_id))
  	if storage is None and not use_context_storage:
  	# Without storage we cannot prove coverage; only advertise a run-scoped runner id.
  	return runner_candidate
  	for candidate in candidates:
  	try:
  	if storage is not None:
  	checkpoint = await storage.load(candidate)
  	else:
  	checkpoint = await self._runner.context.load_checkpoint(candidate)
  	except Exception: # pragma: no cover - storage/type drift
  	logger.debug("Could not load pause checkpoint candidate %s", candidate, exc_info=True)
  	continue
  	if checkpoint is None:
  	continue
  	pending = checkpoint.pending_request_info_events or {}
  	if ids.issubset({str(key) for key in dict(pending)}):
  	return candidate
  	return None

• FunctionalWorkflow._run_core discarding the return value of self._save_checkpoint(...) on completion and WorkflowInterrupted pause:

  agent-framework/python/packages/core/agent_framework/_workflows/_functional.py

  Lines 2198 to 2225 in c3ce250

  	if storage is not None:
  	await self._save_checkpoint(ctx, storage, ckpt_chain[0])
  	# Final status
  	if saw_request:
  	yield _framework_event(WorkflowEvent.status, WorkflowRunState.IDLE_WITH_PENDING_REQUESTS)
  	else:
  	# Clean completion — drop cross-run replay state.
  	self._clear_replay_state()
  	yield _framework_event(WorkflowEvent.status, WorkflowRunState.IDLE)
  	span.add_event(OtelAttr.WORKFLOW_COMPLETED)
  	except WorkflowInterrupted:
  	# Persist step cache for response-only replay
  	self._capture_replay_state(ctx, message)
  	# HITL interruption — yield events collected so far
  	for event in ctx._get_events():
  	if event.type == "request_info":
  	saw_request = True
  	yield event
  	if event.type == "request_info":
  	yield _framework_event(WorkflowEvent.status, WorkflowRunState.IN_PROGRESS_PENDING_REQUESTS)
  	# Save checkpoint
  	if storage is not None:
  	await self._save_checkpoint(ctx, storage, ckpt_chain[0])

• agent_framework_ag_ui/_workflow_run.py checking get_last_checkpoint_id and resolve_pause_checkpoint_id:

  agent-framework/python/packages/ag-ui/agent_framework_ag_ui/_workflow_run.py

  Lines 149 to 227 in c3ce250

  	"value": value,
  	}.items()
  	if value is not None
  	}
  	return {"agent_framework": agent_framework_metadata}
  	def _attach_checkpoint_id_to_interrupts(
  	interrupts: list[dict[str, Any]],
  	checkpoint_id: str | None,
  	) -> list[dict[str, Any]]:
  	"""Attach ``checkpoint_id`` to each interrupt's ``metadata.agent_framework``.
  	Multi-worker hosts need the pause checkpoint on the wire so the next resume can pass
  	``forwardedProps.checkpoint_id`` without a side-channel lookup. No-op when checkpointing
  	is inactive or the id is already present.
  	"""
  	if not checkpoint_id or not interrupts:
  	return interrupts
  	attached: list[dict[str, Any]] = []
  	for interrupt in interrupts:
  	entry = dict(interrupt)
  	metadata = entry.get("metadata")
  	if isinstance(metadata, dict):
  	metadata = dict(metadata)
  	else:
  	metadata = {}
  	agent_framework = metadata.get("agent_framework")
  	if isinstance(agent_framework, dict):
  	agent_framework = dict(agent_framework)
  	else:
  	agent_framework = {}
  	agent_framework.setdefault("checkpoint_id", checkpoint_id)
  	metadata["agent_framework"] = agent_framework
  	entry["metadata"] = metadata
  	attached.append(entry)
  	return attached
  	def _interrupt_request_ids(interrupts: list[dict[str, Any]]) -> set[str]:
  	return {str(item["id"]) for item in interrupts if item.get("id") is not None}
  	async def _pause_checkpoint_id_for_interrupts(
  	*,
  	workflow: Workflow,
  	checkpoint_storage: CheckpointStorage | None,
  	interrupts: list[dict[str, Any]],
  	known_checkpoint_id: str | None = None,
  	baseline_checkpoint_id: Any = _BASELINE_OMITTED,
  	) -> str | None:
  	"""Resolve the pause checkpoint for *this* run's interrupts via core.
  	When ``baseline_checkpoint_id`` is omitted, core uses the baseline captured at
  	``workflow.run()`` start. Pass ``None`` explicitly for short-circuit paths that
  	did not call ``run()`` and should advertise the current runner id when present.
  	"""
  	if not interrupts:
  	return None
  	resolve = getattr(workflow, "resolve_pause_checkpoint_id", None)
  	if not callable(resolve):
  	return None
  	# getattr returns a plain object to the type checker; cast to an awaitable callable.
  	resolve_fn = cast(
  	Callable[..., Awaitable[str | None]],
  	resolve,
  	)
  	kwargs: dict[str, Any] = {
  	"checkpoint_storage": checkpoint_storage,
  	"known_checkpoint_id": known_checkpoint_id,
  	}
  	if baseline_checkpoint_id is not _BASELINE_OMITTED:
  	kwargs["baseline_checkpoint_id"] = baseline_checkpoint_id
  	return await resolve_fn(_interrupt_request_ids(interrupts), **kwargs)

Minimal reproduction showing the parity gap:

import asyncio
from agent_framework import InMemoryCheckpointStorage, RunContext, workflow


@workflow
async def approval_flow(prompt: str, ctx: RunContext) -> str:
    approved = await ctx.request_info({"prompt": prompt}, response_type=bool)
    return "approved" if approved else "rejected"


async def main() -> None:
    storage = InMemoryCheckpointStorage()
    wf = approval_flow.build(checkpoint_storage=storage)

    result = await wf.run("deploy v1")
    request_ids = [ev.request_id for ev in result.get_request_info_events() if ev.request_id]
    print("pending request_ids:", request_ids)
    print("has get_last_checkpoint_id:", hasattr(wf, "get_last_checkpoint_id"))
    print("has resolve_pause_checkpoint_id:", hasattr(wf, "resolve_pause_checkpoint_id"))


asyncio.run(main())

Desired Outcome

FunctionalWorkflow exposes the same checkpoint identification API as graph-based Workflow:

• wf.get_last_checkpoint_id() -> str | None: returns the checkpoint ID last persisted or restored by this FunctionalWorkflow instance.
• await wf.resolve_pause_checkpoint_id(request_ids, *, checkpoint_storage=None, known_checkpoint_id=None, baseline_checkpoint_id=...) -> str | None: resolves the run-scoped pause checkpoint whose pending_request_info_events cover request_ids, excluding any incoming restored checkpoint ID that did not advance during the run.

Alternatives Considered

• Querying await storage.get_latest(workflow_name=...) or await storage.list_checkpoints(workflow_name=...) in caller code after wf.run(...). This requires every caller to hold a direct reference to storage, is verbose, and is unsafe when multiple FunctionalWorkflow instances with the same workflow_name share a CheckpointStorage backend concurrently.

Proposed Direction

Bring FunctionalWorkflow to parity with Workflow:

1. Track _last_checkpoint_id, _restored_checkpoint_id, and _run_baseline_checkpoint_id on FunctionalWorkflow, updating _last_checkpoint_id whenever _save_checkpoint persists a step, pause, or completion checkpoint (and when restoring from checkpoint_id).
2. Implement get_last_checkpoint_id(self) -> str | None and async def resolve_pause_checkpoint_id(...) -> str | None on FunctionalWorkflow with the same signature and resolution semantics as Workflow (since FunctionalWorkflow._save_checkpoint already populates WorkflowCheckpoint.pending_request_info_events from ctx._pending_requests).

Scope and Non-Goals

• In scope: FunctionalWorkflow in python/packages/core/agent_framework/_workflows/_functional.py and deterministic unit tests in python/packages/core/tests/workflow/test_functional_workflow.py.
• Non-goals: Changing WorkflowCheckpoint schema or altering graph Workflow checkpoint behavior.

Language

Python

Contribution Intent

I am willing to implement this

Acknowledgements

• I searched existing issues and did not find a duplicate.
• I will wait for explicit maintainer agreement before starting implementation of a non-trivial change.