fix(crew-ai): isolate Crew(memory=True) per AG-UI threadId

integrations/crew-ai/python/README.md

@@ -78,6 +78,56 @@ opens, and a `RAW` first event makes the reference client reject the whole strea
 those are held and released once the run has opened. Both RAW buffers are bounded; a
 saturated buffer degrades mirroring (logged) rather than the run.
 
+### Memory is isolated per `threadId` (default ON)
+
+A crew served with `Crew(memory=True)` keeps its memories in one on-disk store,
+namespaced by the *crew name*. Nothing in that namespace derives from the AG-UI
+`threadId`, so without help every chat served by an endpoint reads and writes the
+same namespace and one user's remembered facts surface in another user's chat.
+(Setting `inputs["id"] = thread_id` does not help: that scopes crewai's flow-state
+persistence, a different subsystem.) `Agent(memory=True)` has the same shape one
+level down: the agent builds its *own* memory, which crewai prefers over the
+crew's.
+
+The bridge closes that by giving each request a `MemoryScope` view of the crew's
+memory — and of each agent's own memory — rooted at a path derived from the
+request's `threadId`. Threads are mutually invisible; each still sees its own
+history across sequential runs. One physical store, no directory-per-thread
+sprawl.
+
+Because crewai picks the executing agent off `task.agent` (or `manager_agent`
+under the hierarchical process) and reaches the crew's memory through
+`agent.crew`, the request gets shallow *views* of the crew, its agents and its
+tasks, wired to each other. Nothing shared between concurrent requests is
+mutated, and everything below the views (tools, LLMs, knowledge, the store
+itself) stays shared.
+
+```bash
+# Opt out: restore the pre-fix behaviour of one memory namespace per crew,
+# shared by every chat. Useful when the crew is a durable knowledge base
+# rather than a per-conversation memory.
+AGUI_CREWAI_THREAD_SCOPED_MEMORY=false
+```
+
+Limitations, in order of how likely you are to hit them:
+
+- **Only crews and agents the bridge can reach are scoped.** That means the crew
+  you passed to `add_crewai_crew_fastapi_endpoint`, plus any crew or standalone
+  agent your `Flow` holds as an attribute (a crew's own agents and tasks come
+  with it). A crew or agent *constructed inside* a flow method is created after
+  this point and is not scoped; construct it as a flow attribute, or pass it a
+  `Memory` you scope yourself.
+- **Per-request views are shallow.** Each request runs against copies of the
+  crew, its agents and its tasks, so `crew.tasks[0].output` on the object you
+  built is not filled in by a bridge-served run; read the run's result off the
+  AG-UI event stream instead.
+- **Isolation is logical, not physical.** All threads share one store and one
+  embedder; a scope keeps reads and writes inside a namespace, it is not a
+  security boundary against code that queries the store directly.
+- **Older crewai degrades rather than crashing.** The bridge probes for crewai's
+  unified memory view API at runtime. On a build without it, isolation is not
+  active and the bridge logs one warning saying exactly that.
+
 ### `get_capabilities()`
 
 Returns a capability declaration (the CrewAI counterpart of

integrations/crew-ai/python/ag_ui_crewai/_capabilities.py

@@ -276,6 +276,14 @@ def flow_supports_stream_frames(flow: Any) -> bool:
 _Flow = getattr(_CREWAI_MODULE, "Flow", None) if _CREWAI_MODULE else None
 _Crew = getattr(_CREWAI_MODULE, "Crew", None) if _CREWAI_MODULE else None
 
+# ``BaseAgent`` is the base every crewai agent derives from, including a user's
+# own subclass, so it is the wider net for "this attribute is an agent".
+# ``crewai.Agent`` is the fallback for a build that does not expose it.
+_BASE_AGENT_MODULE, _ = _first_module(["crewai.agents.agent_builder.base_agent"])
+_Agent = (
+    getattr(_BASE_AGENT_MODULE, "BaseAgent", None) if _BASE_AGENT_MODULE else None
+) or (getattr(_CREWAI_MODULE, "Agent", None) if _CREWAI_MODULE else None)
+
 
 def _kwarg_in_signature(func: Any, name: str) -> bool:
     """True when ``func`` declares a parameter ``name`` (or accepts ``**kwargs``).
@@ -388,6 +396,46 @@ def supported_checkpoint_kwargs(method: Any, kwargs: dict[str, Any]) -> dict[str
     return {k: v for k, v in kwargs.items() if k in params}
 
 
+# --------------------------------------------------------------------------
+# Memory-isolation resolution
+# --------------------------------------------------------------------------
+# crewai 1.x replaced the 0.x short-term / entity / long-term stores with ONE
+# ``Memory`` object over ONE store, namespaced by a ``root_scope`` string that
+# ``Crew.create_crew_memory`` derives from the CREW NAME. Nothing in that path
+# derives from the AG-UI ``threadId``, so two chats served by the same endpoint
+# read and write the same namespace.
+#
+# The isolation primitive is ``Memory.scope(path)``, which returns a
+# ``MemoryScope`` view whose reads and writes are confined to ``path`` and
+# below. ``Crew._memory`` is typed ``Memory | MemoryScope | MemorySlice``, so a
+# view is a first-class thing to hand a crew, not a hack.
+#
+# Resolved (never version-gated) so a build without the unified memory API
+# degrades to "no isolation, one warning" rather than crashing. The warning is
+# emitted at the CALL SITE (``_memory``) rather than from ``warn_on_gaps``: an
+# operator who never sets ``memory=True`` has no gap to hear about, and an
+# import-time warning for them would be pure noise.
+_MEMORY_MODULE, _MEMORY_MODULE_NAME = _first_module(["crewai.memory.unified_memory"])
+Memory = getattr(_MEMORY_MODULE, "Memory", None) if _MEMORY_MODULE is not None else None
+
+# crewai's own scope-name sanitizer (``crewai.memory.utils``). Used so a
+# bridge-built scope segment is normalised exactly the way crewai normalises the
+# crew-name segment sitting above it. ``_memory`` carries an equivalent fallback
+# for builds that do not expose it.
+_MEMORY_UTILS_MODULE, _ = _first_module(["crewai.memory.utils"])
+sanitize_scope_name = (
+    getattr(_MEMORY_UTILS_MODULE, "sanitize_scope_name", None)
+    if _MEMORY_UTILS_MODULE is not None
+    else None
+)
+
+# Both are required: the type (to recognise a crew's memory) and the view
+# factory (to derive a per-thread namespace from it).
+_memory_scope_available = Memory is not None and callable(
+    getattr(Memory, "scope", None)
+)
+
+
 @dataclass(frozen=True)
 class _Capabilities:
     """Cached, immutable snapshot of the detected crewai capabilities.
@@ -415,6 +463,12 @@ class _Capabilities:
     checkpoint_fork_supported: bool = False
     checkpoint_events_available: bool = False
     checkpoint_state_module: str | None = None
+    # Per-thread memory isolation: informational. ``_memory`` keys off the
+    # resolved symbols and warns once at the call site, so this is NOT listed in
+    # ``missing`` (which drives import-time warnings that would fire for every
+    # operator, including the majority who never enable crew memory).
+    memory_scope_available: bool = False
+    memory_module: str | None = None
     missing: tuple[str, ...] = field(default_factory=tuple)
 
     def warn_on_gaps(self) -> None:
@@ -488,6 +542,8 @@ def _detect() -> _Capabilities:
         checkpoint_fork_supported=_checkpoint_fork_supported,
         checkpoint_events_available=_checkpoint_events_available,
         checkpoint_state_module=_CKPT_STATE_MODULE_NAME,
+        memory_scope_available=_memory_scope_available,
+        memory_module=_MEMORY_MODULE_NAME,
         missing=tuple(missing),
     )
     caps.warn_on_gaps()

integrations/crew-ai/python/ag_ui_crewai/_config.py

@@ -26,6 +26,15 @@
 
 EMIT_RAW_EVENTS_ENV_VAR = "AGUI_CREWAI_EMIT_RAW_EVENTS"
 
+# Per-thread memory isolation (crew and agent) ships ON: sharing one namespace across
+# every AG-UI ``threadId`` leaks one chat's remembered facts into another, which
+# is a privacy bug rather than a feature. The opt-out exists because a
+# deployment may WANT one durable knowledge base behind every chat; turning it
+# off restores the pre-fix "one namespace per crew name" behaviour exactly.
+DEFAULT_THREAD_SCOPED_MEMORY = True
+
+THREAD_SCOPED_MEMORY_ENV_VAR = "AGUI_CREWAI_THREAD_SCOPED_MEMORY"
+
 # Vocabulary ``_parse_env_bool`` accepts, so the "was this value used?" check stays
 # in step with the parser instead of duplicating its token list.
 _BOOL_TOKENS = _TRUE_VALUES | _FALSE_VALUES
@@ -76,3 +85,28 @@ def resolve_emit_raw_events(emit_raw_events: bool | None) -> bool:
     used = raw is not None and raw.strip().casefold() in _BOOL_TOKENS
     _warn_if_env_value_ignored(EMIT_RAW_EVENTS_ENV_VAR, raw, used)
     return resolved
+
+
+def resolve_thread_scoped_memory() -> bool:
+    """Resolve per-thread crew-memory isolation: env var > shipped default (on).
+
+    Env-only, and re-read per request rather than resolved once at registration:
+    unlike ``emit_raw_events`` there is no endpoint-factory argument to conflict
+    with, and an operator flipping the variable should not have to know which
+    call it was frozen at.
+
+    Deliberately NOT ``_parse_env_bool``: that parser treats anything outside its
+    true-set as false, which is the right fail-safe for an option that ships OFF
+    but the wrong one here: a typo would silently DISABLE isolation and restore
+    the cross-thread leak. Only a recognised false token turns it off; anything
+    else keeps the shipped default and warns once.
+    """
+    raw = os.environ.get(THREAD_SCOPED_MEMORY_ENV_VAR)
+    if raw is None:
+        return DEFAULT_THREAD_SCOPED_MEMORY
+    token = raw.strip().casefold()
+    used = token in _BOOL_TOKENS
+    _warn_if_env_value_ignored(THREAD_SCOPED_MEMORY_ENV_VAR, raw, used)
+    if not used:
+        return DEFAULT_THREAD_SCOPED_MEMORY
+    return token in _TRUE_VALUES