Python: [BREAKING] Bind AG-UI workflow agent executors to one authorization scope

python/packages/ag-ui/AGENTS.md

@@ -131,6 +131,47 @@ AG-UI protocol integration for building agent UIs with the AG-UI standard.
   `single_user=True` assigns one opaque authorization scope for the lifetime of that endpoint registration and must not
   be used for callers with distinct authorization. Use an explicit resolver when snapshots, checkpoints, or provider
   state must retain the same scope across process restarts.
+- An `AgentExecutor` keeps one `AgentSession` (history, pending approvals, standing approval rules) between runs, and
+  that session is not tied to a caller. `_session_ownership.claim_agent_executors` binds every `AgentExecutor` of a
+  workflow, and the `AgentSession` each one holds when the run starts, to the first Snapshot Scope that runs it. It
+  looks into nested `WorkflowExecutor` workflows and into the workflow behind a `WorkflowAgent` that an
+  `AgentExecutor` wraps. A run without a scope is an owner of its own, so a workflow that ran without a scope refuses
+  a later scope and the reverse. The registry (`IdentityOwners`) keys objects by `id()` with a weak reference and
+  never calls their `__hash__` or `__eq__`, so unhashable subclasses work and equal objects stay apart. The claim is
+  one process-wide locked check-and-bind. `AgentFrameworkWorkflow.run` and the `WorkflowAgent` branch of
+  `run_agent_stream` (through `claim_workflow_agent`, which also looks through A2UI runners) call it after a snapshot
+  replay has returned and before any workflow runs, so a replay never takes ownership. Another scope receives
+  `RUN_STARTED` and `RUN_ERROR` with `WORKFLOW_SCOPE_CONFLICT` and never reaches the model. It covers a shared
+  `Workflow` or `WorkflowAgent`, and a `workflow_factory` that returns shared workflows, executors, or an injected
+  `AgentSession`. It follows live objects only and does not cover:
+  - state that a factory loads from storage into a new object (`AgentSession.from_dict()`, `SessionStore.get()`, a
+    history provider that keys data by `session_id`);
+  - sessions that alias state outside the session object (a shared `state` mapping or `service_session_id`);
+  - executors that are not `AgentExecutor` but keep a session or cache (custom executors, the group chat
+    orchestrator);
+  - workflows without an `AgentExecutor`;
+  - a session that an executor creates during a checkpoint restore (it stays reachable only through the bound
+    executor, and the next run of the same scope binds it).
+
+  The rule does not partition the threads of one scope: `single_user=True` or a constant resolver shares one
+  conversation state across the threads of a shared workflow. Never reset sessions or clone executors to resolve a
+  conflict; the factory builds new executors and sessions. The factory receives only the thread id, so it must not
+  restore session state that is keyed by the thread id alone; use `checkpoint_storage`, which AG-UI loads only for the
+  scope and thread that saved it.
+- AG-UI does not support a `WorkflowAgent` that an agent exposes as a tool (`WorkflowAgent.as_tool()`): every caller
+  of the agent would reach one workflow and its executor sessions, and the ownership claim cannot see that path.
+  `BaseAgent.as_tool()` keeps the agent it wraps in the private `_wrapped_agent` attribute of the tool, and
+  `_session_ownership.workflow_agent_tools` follows those links through the tools of an agent, the sub-agents they
+  delegate to, and the agents of every `AgentExecutor`, also through A2UI runners and nested workflows, with a cycle
+  guard. `add_agent_framework_fastapi_endpoint` raises `ValueError` (`_reject_workflow_agent_tools`) for what it can
+  see at registration. `AgentFrameworkWorkflow.run` and `run_agent_stream` end the run with `RUN_ERROR`
+  `WORKFLOW_AGENT_TOOL_UNSUPPORTED` before the ownership claim, which covers workflow factories and tools added after
+  registration. The rejection applies with `single_user=True` as well. A tool function that calls a workflow from
+  custom code, and a tool that a context provider adds while a run is prepared, are not detected. They are unsupported
+  all the same: exposing a `WorkflowAgent` as a tool is not supported in any form, and AG-UI gives no isolation to its
+  callers. The search needs a core whose `as_tool()` records `_wrapped_agent`; with an older core it finds nothing, so
+  raise this package's `agent-framework-core` floor to the first release that has it when you prepare a release.
+  Remove this check together with `WorkflowAgent.as_tool()`.
 - Request `state` is client-owned Shared State and is merged into `AgentSession.state`, minus a protected set:
   tool-approval state, history/context-provider namespaces, message-injection state, standalone FIDES security
   state, and provider-owned keys.

python/packages/ag-ui/README.md

@@ -65,9 +65,15 @@ app = FastAPI()
 add_agent_framework_fastapi_endpoint(app, workflow, "/", single_user=True)
 ```
 
+One workflow instance serves every request, so its executors keep their state across callers. A shared workflow that
+holds an `AgentExecutor` (its `AgentSession` keeps history, pending approvals, and standing approval rules) serves only
+the first authorization scope that runs it. Use `workflow_factory`, as shown next, to serve several scopes.
+
 ### Server (Thread-Scoped WorkflowBuilder)
 
-Use `workflow_factory` when your workflow keeps runtime state (for example pending `request_info` interrupts) and must be isolated per AG-UI thread:
+Use `workflow_factory` when your workflow keeps runtime state (for example pending `request_info` interrupts or
+`AgentExecutor` sessions) and must be isolated per authorization scope and AG-UI thread. Build the workflow and its
+executors inside the factory:
 
 ```python
 from fastapi import FastAPI
@@ -76,7 +82,7 @@ from agent_framework.ag_ui import AgentFrameworkWorkflow, add_agent_framework_fa
 
 
 def build_workflow_for_thread(thread_id: str) -> Workflow:
-    # Build a fresh workflow instance for each thread id.
+    # Build a fresh workflow instance, including new AgentExecutor instances, for each thread id.
     return WorkflowBuilder(start_executor=...).build()
 
 
@@ -350,6 +356,40 @@ Authentication in the separate `dependencies` list can allow or reject a request
 server-side state. Do not combine authentication-only dependencies with `single_user=True` when different authenticated
 callers can reach the endpoint.
 
+### Workflows That Hold Agent Sessions
+
+An `AgentExecutor` keeps one `AgentSession` between runs. Conversation history, pending approval requests, and
+standing "always approve" rules live there, and the session is not tied to a caller. An endpoint therefore binds each
+`AgentExecutor`, and the session it holds, to the first authorization scope that runs it. A shared `Workflow` or
+`WorkflowAgent` that holds one serves only that scope. A run in another scope ends with a `WORKFLOW_SCOPE_CONFLICT`
+`RUN_ERROR` before it reaches the model. A run without a scope owns what it runs in the same way, so a workflow that ran
+locally is not handed to a scope later. A snapshot replay starts no workflow and does not take ownership. Pass
+`AgentFrameworkWorkflow(workflow_factory=...)` with a factory that builds a new workflow, including new `AgentExecutor`
+and `AgentSession` instances, for every call to serve several scopes. A factory that returns a workflow, an executor, or
+a session it shares with another scope is refused the same way. Workflows without an `AgentExecutor` are not bound to a
+scope; their pending interrupts stay fenced to the scope and thread that raised them.
+
+With `single_user=True`, or with a resolver that returns one constant scope, a shared workflow keeps one conversation
+state for every thread. Use `workflow_factory` when threads must stay independent.
+
+The endpoint follows live `AgentExecutor` and `AgentSession` objects. It cannot see state that a factory loads from
+storage into a new object. For example, `AgentSession.from_dict()` and `SessionStore.get()` return new objects with the
+same session id and state, so two scopes that load one stored session share its history. The factory receives only the
+thread id, and it runs once for each scope and thread. AG-UI does not pass it the authorization scope. A factory must
+therefore start every session with fresh, independent state. Do not restore session state that is keyed by the thread
+id alone. To continue a conversation, use `checkpoint_storage`: AG-UI restores a checkpoint only for the scope and
+thread that saved it. The rule does not cover executors that are not `AgentExecutor`, or state that sessions share
+outside the session object, for example a `state` dictionary or a `service_session_id`.
+
+AG-UI does not support an agent that exposes a `WorkflowAgent` as a tool (`WorkflowAgent.as_tool()`), however the tool
+gets into the agent, because every caller of that agent would reach the same workflow and its executor sessions. AG-UI
+rejects what it can see. `add_agent_framework_fastapi_endpoint` raises `ValueError` for an agent or workflow that lists
+one, with `single_user=True` as well. When a workflow factory builds one, the run ends with a
+`WORKFLOW_AGENT_TOOL_UNSUPPORTED` `RUN_ERROR` before it reaches the model. AG-UI cannot see a workflow tool that a
+context provider adds while a run is prepared, or a tool function of your own that calls a workflow. Those stay
+unsupported, and AG-UI does not isolate their callers. Serve the workflow with
+`AgentFrameworkWorkflow(workflow_factory=...)`, or build the workflow inside the tool function for every call.
+
 ### Stable Single-User Scope
 
 `single_user=True` creates an opaque scope for one endpoint registration, which is appropriate for local and

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

@@ -114,6 +114,16 @@
     _resolve_ui_payload,  # type: ignore
     _track_tool_call_segment,  # type: ignore
 )
+from ._session_ownership import (
+    SCOPE_CONFLICT_CODE,
+    SCOPE_CONFLICT_LOG_MESSAGE,
+    SCOPE_CONFLICT_MESSAGE,
+    WORKFLOW_AGENT_TOOL_CODE,
+    WORKFLOW_AGENT_TOOL_MESSAGE,
+    claim_workflow_agent,
+    workflow_agent_tool_error,
+    workflow_agent_tools,
+)
 from ._snapshots import (
     _DEFAULT_STATE_INPUT_KEY,
     _SNAPSHOT_SCOPE_INPUT_KEY,
@@ -3029,6 +3039,19 @@ async def _run_agent_stream(
             yield event
         return
 
+    # A replay above runs no agent, so only a run that reaches the workflow decides who owns its executors.
+    unsupported_tools = workflow_agent_tools(agent)
+    if unsupported_tools:
+        logger.error(workflow_agent_tool_error(agent.name or agent.id, unsupported_tools))
+        yield RunStartedEvent(run_id=run_id, thread_id=thread_id)
+        yield RunErrorEvent(message=WORKFLOW_AGENT_TOOL_MESSAGE, code=WORKFLOW_AGENT_TOOL_CODE)
+        return
+    if not claim_workflow_agent(agent, snapshot_scope):
+        logger.error(SCOPE_CONFLICT_LOG_MESSAGE, agent.name or agent.id)
+        yield RunStartedEvent(run_id=run_id, thread_id=thread_id)
+        yield RunErrorEvent(message=SCOPE_CONFLICT_MESSAGE, code=SCOPE_CONFLICT_CODE)
+        return
+
     snapshot_seed_messages: list[dict[str, Any]] | None = None
 
     if stored_snapshot is not None:

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

@@ -23,6 +23,7 @@
 from ._agent import AgentFrameworkAgent
 from ._approval_state import _APPROVAL_SCOPE_INPUT_KEY
 from ._run_common import _is_snapshot_hydration_request
+from ._session_ownership import workflow_agent_tool_error, workflow_agent_tools
 from ._snapshots import (
     _DEFAULT_STATE_INPUT_KEY,
     _SNAPSHOT_SCOPE_INPUT_KEY,
@@ -120,6 +121,20 @@ def _single_user_scope() -> str:
     return f"{_SINGLE_USER_SCOPE_PREFIX}{uuid.uuid4().hex}"
 
 
+def _reject_workflow_agent_tools(protocol_runner: AgentFrameworkAgent | AgentFrameworkWorkflow) -> None:
+    """Refuse a runner that exposes a ``WorkflowAgent`` as a tool, so that the mistake shows at startup."""
+    # A runner subclass that skips the base constructor has nothing to inspect, and a workflow factory builds its
+    # workflow for each request, so a run checks what the factory built.
+    if isinstance(protocol_runner, AgentFrameworkAgent):
+        subject = getattr(protocol_runner, "agent", None)
+    else:
+        subject = getattr(protocol_runner, "workflow", None)
+    tool_names = workflow_agent_tools(subject)
+    if tool_names:
+        runner_name = getattr(protocol_runner, "name", None) or type(protocol_runner).__name__
+        raise ValueError(workflow_agent_tool_error(runner_name, tool_names))
+
+
 def add_agent_framework_fastapi_endpoint(
     app: FastAPI,
     agent: SupportsAgentRun | AgentFrameworkAgent | Workflow | AgentFrameworkWorkflow,
@@ -145,7 +160,9 @@ def add_agent_framework_fastapi_endpoint(
 
     Args:
         app: The FastAPI application
-        agent: The agent to expose (can be raw SupportsAgentRun or wrapped)
+        agent: The agent to expose (can be raw SupportsAgentRun or wrapped). The endpoint raises ``ValueError`` for an
+            agent, or a workflow, that exposes a ``WorkflowAgent`` as a tool (``WorkflowAgent.as_tool()``), and a run
+            ends with a ``WORKFLOW_AGENT_TOOL_UNSUPPORTED`` ``RUN_ERROR`` when a workflow factory builds one.
         path: The endpoint path
         state_schema: Optional state schema for shared state management; accepts dict or Pydantic model/class
         predict_state_config: Optional predictive state update configuration.
@@ -164,7 +181,8 @@ def add_agent_framework_fastapi_endpoint(
             authorization_scope_resolver for new FastAPI applications because it can consume authenticated dependency
             results. This resolver also scopes internal Agent Sessions, in-memory workflow_factory instances, and
             approval state. It must return a non-empty string derived from authorized request context; invalid results
-            fail the request before accessing state or invoking the runner.
+            fail the request before accessing state or invoking the runner. The shared-workflow rule described for
+            authorization_scope_resolver applies to this resolver as well.
         checkpoint_storage: Optional workflow checkpoint storage, applied when the endpoint exposes a workflow.
             When provided, each run creates a checkpoint at the end of every superstep, and a run may resume from
             a persisted checkpoint by supplying its id in the AG-UI forwarded props
@@ -188,13 +206,19 @@ def add_agent_framework_fastapi_endpoint(
         authorization_scope_resolver: Optional FastAPI dependency callable that returns a non-empty authorization
             scope derived from authenticated request context. The callable may declare normal FastAPI dependencies,
             headers, security schemes, or the current Request. Its result scopes snapshots, internal Agent Sessions,
-            workflow instances and checkpoints, detached-run admission, and server-side approval state.
+            workflow instances and checkpoints, detached-run admission, and server-side approval state. An
+            ``AgentExecutor`` owns one ``AgentSession`` (history, pending approvals, standing approval rules), so a
+            shared ``Workflow`` or ``WorkflowAgent`` that holds one serves only the first scope that runs it. A run in
+            another scope ends with a ``WORKFLOW_SCOPE_CONFLICT`` ``RUN_ERROR`` before it reaches the model. Pass
+            ``AgentFrameworkWorkflow(workflow_factory=...)`` with a factory that builds a new workflow, including new
+            ``AgentExecutor`` and ``AgentSession`` instances, for every call to serve several scopes.
         single_user: Explicitly declare that every caller of this endpoint belongs to one authorization principal.
             The endpoint receives one opaque scope that remains stable for that registration and separates equal route
             paths registered by other applications. Use an explicit resolver when scope continuity must survive process
             restarts or when storage is shared beyond one registration. A single-user application can return its own
             stable application scope from authorization_scope_resolver. Never enable this mode for callers with distinct
-            authorization.
+            authorization. A shared workflow keeps one conversation state for every thread of that principal; use
+            ``workflow_factory`` when threads must stay independent.
     """
     _validate_keepalive_seconds(keepalive_seconds)
     _validate_detached_run_options(max_detached_runs, detached_run_timeout_seconds)
@@ -217,6 +241,8 @@ def add_agent_framework_fastapi_endpoint(
     else:
         raise TypeError("agent must be SupportsAgentRun, Workflow, AgentFrameworkAgent, or AgentFrameworkWorkflow.")
 
+    _reject_workflow_agent_tools(protocol_runner)
+
     _validate_authorization_scope_configuration(
         authorization_scope_resolver=authorization_scope_resolver,
         snapshot_scope_resolver=snapshot_scope_resolver,