[BREAKING] Python: Bind AG-UI workflow interrupts to the resolved thread id

python/packages/ag-ui/AGENTS.md

@@ -108,6 +108,13 @@ AG-UI protocol integration for building agent UIs with the AG-UI standard.
 - AG-UI Thread and Run ids are client-owned protocol correlation ids. Service-session mode stores provider conversation
   or response ids privately in the thread snapshot. Set `service_session_id_from_thread_id=True` only for compatibility
   when the application intentionally uses a provider continuation id as its AG-UI Thread id.
+- A workflow run resolves its Thread id once. A generated id for an omitted `threadId` is the id reported in
+  `RUN_STARTED`, keys the `workflow_factory` cache and Thread Snapshot, and owns the run's live `request_info`
+  interrupts, so a continuation must send it back. A `workflow_factory` request without `threadId` that carries a
+  `resume` entry or a tool or approval-response message (as `_extract_responses_from_messages` defines them) is
+  rejected with `WORKFLOW_RESUME_NOT_FOUND` before a workflow instance is built. Checkpoint ownership binds only
+  client-supplied Thread ids; a checkpoint of a run without one also records the generated id as an alias, so both
+  the reported id and legacy id-less restores resume it.
 - A configured Snapshot Scope and the client-owned Thread id are combined into a deterministic internal
   `AgentSession.session_id`. Keep the raw Thread id for protocol events and snapshot addressing so equal Thread ids
   in different trusted scopes cannot collide in context-provider state. The deprecated

python/packages/ag-ui/README.md

@@ -495,6 +495,13 @@ AG-UI events and snapshot operations. When using `AgentFrameworkWorkflow(workflo
 also scopes the in-memory workflow cache even without a snapshot store; provide it in multi-user deployments so two
 users who submit the same `threadId` do not share a live `Workflow` instance.
 
+A workflow request that omits `threadId` is assigned a generated Thread id. It is reported in `RUN_STARTED`, keys the
+`workflow_factory` instance and Thread Snapshot, and owns that run's pending `request_info` interrupts. Send it back as
+`threadId` to continue the run; a continuation that omits it receives `WORKFLOW_RESUME_NOT_FOUND`. With
+`workflow_factory`, a request without `threadId` that carries a `resume` entry, a tool result, or an approval response
+is rejected the same way instead of starting a new run, so send `threadId` with every such request. A checkpoint
+created by such a run resumes with `checkpointId` and either the reported `threadId` or, as before, no `threadId`.
+
 Authenticate and authorize the current scope on every request, including hydration and approval resume, rather than
 trusting a scope supplied through Shared State or forwarded properties. A configured resolver must return a non-empty
 string. Returning `None`, an empty string, or another type fails the request with a generic HTTP 500 configuration

python/packages/ag-ui/agent_framework_ag_ui/_workflow.py

@@ -35,7 +35,11 @@
 from agent_framework._telemetry import mark_feature_used
 
 from ._feature_usage import FeatureIndex
-from ._message_adapters import agent_framework_messages_to_agui, agui_messages_to_snapshot_format
+from ._message_adapters import (
+    agent_framework_messages_to_agui,
+    agui_messages_to_snapshot_format,
+    normalize_agui_input_messages,
+)
 from ._run_common import (
     _cancelled_resume_interrupt_ids,
     _extract_resume_payload,
@@ -61,7 +65,11 @@
     generate_event_id,
     make_json_safe,
 )
-from ._workflow_run import _pending_request_events, run_workflow_stream  # pyright: ignore[reportPrivateUsage]
+from ._workflow_run import (  # pyright: ignore[reportPrivateUsage]
+    _extract_responses_from_messages,
+    _pending_request_events,
+    run_workflow_stream,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -232,6 +240,12 @@ def _checkpoint_id_from_input(input_data: dict[str, Any]) -> str | None:
     return str(checkpoint_id)
 
 
+def _message_response_ids(raw_messages: list[dict[str, Any]]) -> set[str]:
+    """Return the request ids that tool or approval-response messages in a request answer."""
+    messages, _ = normalize_agui_input_messages(raw_messages, sanitize_tool_history=False)
+    return set(_extract_responses_from_messages(messages))
+
+
 def _checkpoint_request_owner(metadata: dict[str, Any]) -> WorkflowRequestOwner | None:
     """Read AG-UI workflow request ownership from checkpoint metadata."""
     raw_owner = metadata.get(_CHECKPOINT_REQUEST_OWNER_KEY)
@@ -246,6 +260,15 @@ def _checkpoint_request_owner(metadata: dict[str, Any]) -> WorkflowRequestOwner
     return snapshot_scope, thread_id
 
 
+def _checkpoint_generated_thread_id(metadata: dict[str, Any]) -> str | None:
+    """Read the generated thread id recorded for a checkpoint whose run omitted ``threadId``."""
+    raw_owner = metadata.get(_CHECKPOINT_REQUEST_OWNER_KEY)
+    if not isinstance(raw_owner, dict):
+        return None
+    generated_thread_id = raw_owner.get("generated_thread_id")
+    return generated_thread_id if isinstance(generated_thread_id, str) else None
+
+
 def _workflow_checkpoint_storage(workflow: Workflow) -> CheckpointStorage | None:
     """Return the workflow's build-time checkpoint storage."""
     runner = getattr(workflow, "_runner", None)
@@ -258,21 +281,33 @@ def _workflow_checkpoint_storage(workflow: Workflow) -> CheckpointStorage | None
 class _OwnedWorkflowCheckpointStorage:
     """Attach one AG-UI request owner to checkpoints in their original save."""
 
-    def __init__(self, storage: CheckpointStorage, owner: WorkflowRequestOwner) -> None:
+    def __init__(
+        self, storage: CheckpointStorage, owner: WorkflowRequestOwner, generated_thread_id: str | None = None
+    ) -> None:
         self._storage = storage
         self._owner = owner
+        self._generated_thread_id = generated_thread_id
 
     def _owns(self, checkpoint: WorkflowCheckpoint) -> bool:
         checkpoint_owner = _checkpoint_request_owner(checkpoint.metadata)
-        return checkpoint_owner == self._owner or (checkpoint_owner is None and self._owner == (None, None))
+        if checkpoint_owner == self._owner or (checkpoint_owner is None and self._owner == (None, None)):
+            return True
+        # A run that omitted threadId also records the id it reported, so a continuation can echo that id.
+        generated_thread_id = _checkpoint_generated_thread_id(checkpoint.metadata)
+        return (
+            checkpoint_owner is not None
+            and checkpoint_owner[1] is None
+            and generated_thread_id is not None
+            and self._owner == (checkpoint_owner[0], generated_thread_id)
+        )
 
     async def save(self, checkpoint: WorkflowCheckpoint) -> CheckpointID:
         """Save a checkpoint with ownership for any pending request occurrences."""
         checkpoint.metadata = dict(checkpoint.metadata)
-        checkpoint.metadata[_CHECKPOINT_REQUEST_OWNER_KEY] = {
-            "snapshot_scope": self._owner[0],
-            "thread_id": self._owner[1],
-        }
+        owner_metadata: dict[str, str | None] = {"snapshot_scope": self._owner[0], "thread_id": self._owner[1]}
+        if self._generated_thread_id is not None:
+            owner_metadata["generated_thread_id"] = self._generated_thread_id
+        checkpoint.metadata[_CHECKPOINT_REQUEST_OWNER_KEY] = owner_metadata
         return await self._storage.save(checkpoint)
 
     async def load(self, checkpoint_id: CheckpointID) -> WorkflowCheckpoint:
@@ -663,13 +698,16 @@ def __init__(
         self.snapshot_store = snapshot_store
         self.checkpoint_storage = checkpoint_storage
 
+    @staticmethod
+    def _supplied_thread_id_from_input(input_data: dict[str, Any]) -> str | None:
+        """Return the client-supplied thread id, treating an empty value as omitted."""
+        thread_id = input_data.get("thread_id") or input_data.get("threadId")
+        return str(thread_id) if thread_id else None
+
     @staticmethod
     def _thread_id_from_input(input_data: dict[str, Any]) -> str:
         """Resolve a stable thread id from AG-UI input payload."""
-        thread_id = input_data.get("thread_id") or input_data.get("threadId")
-        if thread_id is not None:
-            return str(thread_id)
-        return str(uuid.uuid4())
+        return AgentFrameworkWorkflow._supplied_thread_id_from_input(input_data) or str(uuid.uuid4())
 
     def _resolve_workflow(self, thread_id: str, snapshot_scope: str | None = None) -> Workflow:
         """Get the workflow instance for the current run."""
@@ -742,17 +780,39 @@ async def run(self, input_data: dict[str, Any]) -> AsyncGenerator[BaseEvent]:
 
         checkpoint_id = _checkpoint_id_from_input(input_data)
 
-        supplied_thread_id = input_data.get("thread_id") or input_data.get("threadId")
-        request_owner = (snapshot_scope, str(supplied_thread_id) if supplied_thread_id is not None else None)
+        supplied_thread_id = self._supplied_thread_id_from_input(input_data)
+        request_owner = (snapshot_scope, supplied_thread_id)
+        # Live interrupts bind to the resolved thread id, including a generated one that RUN_STARTED reports, so a
+        # continuation must echo it. Checkpoint ownership keeps binding only client-supplied ids and records a
+        # generated id as an alias, which keeps legacy id-less checkpoint restores working.
+        live_owner = (snapshot_scope, thread_id)
         resume_interrupt_ids = {
             str(interrupt["id"])
             for interrupt in _normalize_resume_interrupts(resume_payload)
             if interrupt.get("id") is not None
         }
+        if (
+            self.workflow is None
+            and checkpoint_id is None
+            and supplied_thread_id is None
+            and (resume_interrupt_ids or _message_response_ids(raw_messages))
+        ):
+            # A factory run that omits threadId reports a new generated id, so nothing can be pending on it and a
+            # resume entry, tool result, or approval response cannot continue anything.
+            yield RunStartedEvent(run_id=run_id, thread_id=thread_id)
+            yield RunErrorEvent(
+                message="No pending interrupt found for this AG-UI thread.",
+                code="WORKFLOW_RESUME_NOT_FOUND",
+            )
+            return
         workflow = self._resolve_workflow(thread_id, snapshot_scope)
         checkpoint_storage = self.checkpoint_storage or _workflow_checkpoint_storage(workflow)
         owned_checkpoint_storage = (
-            _OwnedWorkflowCheckpointStorage(checkpoint_storage, request_owner)
+            _OwnedWorkflowCheckpointStorage(
+                checkpoint_storage,
+                request_owner,
+                generated_thread_id=thread_id if supplied_thread_id is None else None,
+            )
             if checkpoint_storage is not None
             else None
         )
@@ -766,7 +826,7 @@ async def run(self, input_data: dict[str, Any]) -> AsyncGenerator[BaseEvent]:
         if self.workflow is not None and checkpoint_id is None:
             for request_event in live_pending_events.values():
                 owner = getattr(request_event, _REQUEST_OWNER_ATTRIBUTE, None)
-                if owner != request_owner and (owner is not None or request_owner != (None, None)):
+                if owner != live_owner and (owner is not None or request_owner != (None, None)):
                     yield RunStartedEvent(run_id=run_id, thread_id=thread_id)
                     yield RunErrorEvent(
                         message="No pending interrupt found for this AG-UI thread.",
@@ -801,7 +861,7 @@ async def run(self, input_data: dict[str, Any]) -> AsyncGenerator[BaseEvent]:
             for interrupt_id in resume_interrupt_ids:
                 request_event = live_pending_events.get(interrupt_id)
                 owner = getattr(request_event, _REQUEST_OWNER_ATTRIBUTE, None)
-                if owner != request_owner and (owner is not None or request_owner != (None, None)):
+                if owner != live_owner and (owner is not None or request_owner != (None, None)):
                     yield RunStartedEvent(run_id=run_id, thread_id=thread_id)
                     yield RunErrorEvent(
                         message=f"No pending interrupt found for resume interruptId '{interrupt_id}'.",
@@ -880,6 +940,7 @@ def append_conversational_responses(values: list[Any]) -> None:
             checkpoint_id=checkpoint_id,
             pending_request_events=effective_pending_events,
             on_conversational_responses=append_conversational_responses,
+            resolved_thread_id=thread_id,
         ):
             if snapshot_builder is not None:
                 snapshot_builder.observe(event)
@@ -891,7 +952,7 @@ def append_conversational_responses(values: list[Any]) -> None:
                 interrupt_id = str(event.tool_call_id)
                 pending_event = (await _pending_request_events(workflow)).get(interrupt_id)
                 if pending_event is not None and getattr(pending_event, _REQUEST_OWNER_ATTRIBUTE, None) is None:
-                    setattr(pending_event, _REQUEST_OWNER_ATTRIBUTE, request_owner)
+                    setattr(pending_event, _REQUEST_OWNER_ATTRIBUTE, live_owner)
             if isinstance(event, RunErrorEvent):
                 run_error_emitted = True
                 if getattr(event, "code", None) == "WORKFLOW_RESUME_CANCELLED":

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

@@ -1149,6 +1149,7 @@ async def run_workflow_stream(
     checkpoint_id: str | None = None,
     pending_request_events: dict[str, Any] | None = None,
     on_conversational_responses: Callable[[list[Any]], None] | None = None,
+    resolved_thread_id: str | None = None,
 ) -> AsyncGenerator[BaseEvent]:
     """Run a Workflow and emit AG-UI protocol events.
 
@@ -1168,9 +1169,11 @@ async def run_workflow_stream(
             start-executor message, so resume stays consistent with the core API.
         pending_request_events: Optional effective pending requests already resolved by the wrapper.
         on_conversational_responses: Optional callback for validated non-control responses.
+        resolved_thread_id: Optional thread id already resolved by the caller, reported in protocol events instead of
+            generating a new one when the request omits it.
     """
     supplied_thread_id = input_data.get("thread_id") or input_data.get("threadId")
-    thread_id = supplied_thread_id or str(uuid.uuid4())
+    thread_id = resolved_thread_id or supplied_thread_id or str(uuid.uuid4())
     run_id = input_data.get("run_id") or input_data.get("runId") or str(uuid.uuid4())
     available_interrupts = input_data.get("available_interrupts") or input_data.get("availableInterrupts")
     if available_interrupts: