feat(backend): isolate expert memory

autogpt_platform/backend/backend/api/features/chat/routes.py

@@ -446,7 +446,7 @@ async def list_sessions(
     ctx: Annotated[auth.RequestContext, Security(auth.get_request_context)],
     limit: int = Query(default=50, ge=1, le=100),
     offset: int = Query(default=0, ge=0),
-    expert_id: str | None = Query(default=None, max_length=128),
+    expert_id: str | None = Query(default=None, min_length=1, max_length=128),
     pinned_first: bool = Query(
         default=True,
         description=(

autogpt_platform/backend/backend/api/features/chat/routes_test.py

@@ -2324,6 +2324,22 @@ def test_list_sessions_filters_by_expert_id(
     assert mock_get.call_args.kwargs["expert_id"] == "expert-1"
 
 
+def test_list_sessions_rejects_empty_expert_id(
+    mocker: pytest_mock.MockerFixture,
+) -> None:
+    """GET /sessions?expert_id= must 422 at validation instead of reaching
+    the db layer, whose ValueError on "" would surface as a 4xx/5xx."""
+    mock_get = mocker.patch(
+        "backend.api.features.chat.routes.get_user_sessions",
+        new_callable=AsyncMock,
+    )
+
+    response = client.get("/sessions?expert_id=")
+
+    assert response.status_code == 422
+    mock_get.assert_not_awaited()
+
+
 def test_list_sessions_can_request_strict_recency(
     mocker: pytest_mock.MockerFixture,
 ) -> None:

autogpt_platform/backend/backend/api/features/experts/experts_db.py

@@ -138,22 +138,34 @@ async def list_experts(user_id: str) -> list[Expert]:
 
 
 async def get_expert(
-    user_id: str, expert_id: str, *, include_workflows: bool = True
+    user_id: str,
+    expert_id: str,
+    *,
+    include_workflows: bool = True,
+    include_archived: bool = False,
 ) -> Expert | None:
     """Fetch a hired expert owned by *user_id*.
 
     Set ``include_workflows=False`` to skip the ExpertWorkflow → LibraryAgent
     + StoreListingVersion joins when the caller only needs the expert's own
     columns. The returned model then always carries an empty ``workflows``
     list — never use that flag to decide whether workflows are installed.
+
+    Archived experts are hidden by default so product surfaces treat them as
+    gone. Set ``include_archived=True`` when the caller must distinguish
+    "archived" (reversible — re-hire revives) from "deleted": the scheduler's
+    scope gate uses this to skip firings without destroying schedules that
+    an un-archive should bring back.
     """
+    where: prisma.types.ExpertWhereInput = {
+        "id": expert_id,
+        "ownerUserId": user_id,
+        "isTemplate": False,
+    }
+    if not include_archived:
+        where["isArchived"] = False
     row = await prisma.models.Expert.prisma().find_first(
-        where={
-            "id": expert_id,
-            "ownerUserId": user_id,
-            "isTemplate": False,
-            "isArchived": False,
-        },
+        where=where,
         include=_WORKFLOW_INCLUDE if include_workflows else None,
     )
     if row is None:

autogpt_platform/backend/backend/copilot/baseline/service.py

@@ -46,6 +46,8 @@
 from backend.copilot.context import get_workspace_manager, set_execution_context
 from backend.copilot.expert_context import build_expert_identity_suffix
 from backend.copilot.graphiti.config import is_enabled_for_user
+from backend.copilot.graphiti.context import fetch_warm_context
+from backend.copilot.graphiti.ingest import enqueue_conversation_turn
 from backend.copilot.local_context_probe import (
     compaction_target_for_window,
     probe_local_context_window,
@@ -1559,6 +1561,34 @@ async def _upload_final_transcript(
         logger.error("[Baseline] Transcript upload failed: %s", upload_err)
 
 
+async def _fetch_graphiti_context(
+    user_id: str,
+    session: ChatSession,
+    message: str | None,
+) -> str | None:
+    return await fetch_warm_context(
+        user_id,
+        message or "",
+        expert_id=session.expert_id,
+    )
+
+
+async def _enqueue_graphiti_turn(
+    user_id: str,
+    session: ChatSession,
+    session_id: str,
+    message: str,
+    assistant_msg: str,
+) -> None:
+    await enqueue_conversation_turn(
+        user_id,
+        session_id,
+        message,
+        assistant_msg=assistant_msg,
+        expert_id=session.expert_id,
+    )
+
+
 async def stream_chat_completion_baseline(
     session_id: str,
     message: str | None = None,
@@ -1591,6 +1621,10 @@ async def stream_chat_completion_baseline(
             f"Session {session_id} not found. Please create a new session first."
         )
 
+    expert_session_suffix = await build_expert_identity_suffix(
+        session.user_id, session.expert_id
+    )
+
     # The session row is the tenancy anchor; the turn entry's org/team only
     # backfills sessions created before org tagging (pre-migration rows).
     if session.organization_id is None and organization_id:
@@ -1781,9 +1815,6 @@ async def stream_chat_completion_baseline(
     # the ~20KB guide warm for the whole session.  Empty string for
     # non-builder sessions keeps the cross-user cache hot.
     builder_session_suffix = await build_builder_system_prompt_suffix(session)
-    expert_session_suffix = await build_expert_identity_suffix(
-        session.user_id, session.expert_id
-    )
     system_prompt = (
         base_system_prompt
         + SHARED_TOOL_NOTES
@@ -1799,9 +1830,7 @@ async def stream_chat_completion_baseline(
     # after openai_messages is built — keeps system prompt static for caching.
     warm_ctx: str | None = None
     if graphiti_enabled and user_id and _pre_drain_msg_count <= 1:
-        from backend.copilot.graphiti.context import fetch_warm_context
-
-        warm_ctx = await fetch_warm_context(user_id, message or "")
+        warm_ctx = await _fetch_graphiti_context(user_id, session, message)
 
     # Context path: transcript content (compacted, isCompactSummary preserved) +
     # gap (DB messages after watermark) + current user turn.
@@ -2524,17 +2553,16 @@ def _trim_openai_on_rollback(_session_anchor: int) -> None:
 
         # --- Graphiti: ingest conversation turn for temporal memory ---
         if graphiti_enabled and user_id and message and is_user_message:
-            from backend.copilot.graphiti.ingest import enqueue_conversation_turn
-
             # Pass only the final assistant reply (after stripping tool-loop
             # chatter) so derived-finding distillation sees the substantive
             # response, not intermediate tool-planning text.
             _ingest_task = asyncio.create_task(
-                enqueue_conversation_turn(
+                _enqueue_graphiti_turn(
                     user_id,
+                    session,
                     session_id,
                     message,
-                    assistant_msg=final_text if state else "",
+                    final_text if state else "",
                 )
             )
             _background_tasks.add(_ingest_task)

autogpt_platform/backend/backend/copilot/baseline/service_unit_test.py

@@ -21,6 +21,8 @@
     _build_cached_system_message,
     _build_natural_finish_empty_fallback_events,
     _compress_session_messages,
+    _enqueue_graphiti_turn,
+    _fetch_graphiti_context,
     _fresh_anthropic_caching_headers,
     _fresh_ephemeral_cache_control,
     _is_anthropic_model,
@@ -29,8 +31,10 @@
     _natural_finish_empty_notice_text,
     _split_user_message_after_skills_block,
     _supports_prompt_cache_markers,
+    stream_chat_completion_baseline,
 )
-from backend.copilot.model import ChatMessage
+from backend.copilot.expert_context import ExpertSessionUnavailableError
+from backend.copilot.model import ChatMessage, ChatSession
 from backend.copilot.response_model import (
     StreamReasoningDelta,
     StreamReasoningEnd,
@@ -45,6 +49,91 @@
 from backend.util.tool_call_loop import LLMLoopResponse, LLMToolCall, ToolCallResult
 
 
+@pytest.mark.asyncio
+async def test_expert_identity_failure_precedes_baseline_turn_mutation() -> None:
+    session = ChatSession.new("user-1", dry_run=False, expert_id="expert-1")
+    identity_mock = AsyncMock(
+        side_effect=ExpertSessionUnavailableError(
+            "The expert for this session no longer exists or is archived."
+        )
+    )
+
+    with (
+        patch(
+            "backend.copilot.baseline.service.build_expert_identity_suffix",
+            new=identity_mock,
+        ),
+        pytest.raises(ExpertSessionUnavailableError),
+    ):
+        async for _ in stream_chat_completion_baseline(
+            session_id=session.session_id,
+            message="private prompt",
+            user_id="user-1",
+            session=session,
+        ):
+            pass
+
+    identity_mock.assert_awaited_once_with("user-1", "expert-1")
+    assert session.messages == []
+
+
+@pytest.mark.asyncio
+async def test_fetch_graphiti_context_uses_expert_session_scope() -> None:
+    session = ChatSession.new(
+        "user-1",
+        dry_run=False,
+        expert_id="expert-1",
+    )
+    fetch_mock = AsyncMock(return_value="expert context")
+
+    with patch(
+        "backend.copilot.baseline.service.fetch_warm_context",
+        new=fetch_mock,
+    ):
+        context = await _fetch_graphiti_context(
+            "user-1",
+            session,
+            "first prompt",
+        )
+
+    assert context == "expert context"
+    fetch_mock.assert_awaited_once_with(
+        "user-1",
+        "first prompt",
+        expert_id="expert-1",
+    )
+
+
+@pytest.mark.asyncio
+async def test_enqueue_graphiti_turn_uses_expert_session_scope() -> None:
+    session = ChatSession.new(
+        "user-1",
+        dry_run=False,
+        expert_id="expert-1",
+    )
+    enqueue_mock = AsyncMock()
+
+    with patch(
+        "backend.copilot.baseline.service.enqueue_conversation_turn",
+        new=enqueue_mock,
+    ):
+        await _enqueue_graphiti_turn(
+            "user-1",
+            session,
+            "session-1",
+            "private prompt",
+            "private response",
+        )
+
+    enqueue_mock.assert_awaited_once_with(
+        "user-1",
+        "session-1",
+        "private prompt",
+        assistant_msg="private response",
+        expert_id="expert-1",
+    )
+
+
 class TestBaselineStreamState:
     def test_defaults(self):
         state = _BaselineStreamState()

autogpt_platform/backend/backend/copilot/db.py

@@ -696,6 +696,7 @@ async def get_user_chat_sessions(
     organization_id: str | None = None,
     title_contains: str | None = None,
     expert_id: str | None = None,
+    autopilot_only: bool = False,
     pinned_first: bool = True,
 ) -> list[ChatSessionInfo]:
     """Get chat sessions for a user, ordered by most recent.
@@ -711,9 +712,18 @@ async def get_user_chat_sessions(
     without waiting on async embedding.
 
     ``expert_id`` restricts the listing to sessions scoped to that expert.
+    ``autopilot_only`` restricts it to sessions whose ``expertId`` is NULL.
+    The explicit flag is necessary because ``expert_id=None`` retains the
+    existing meaning of "all expert scopes" for user-facing session lists.
+
     ``pinned_first=False`` provides strict recency ordering for internal
     adoption flows; the user-facing sidebar keeps pinned sessions first.
     """
+    if expert_id == "":
+        raise ValueError("expert_id must be non-empty")
+    if expert_id is not None and autopilot_only:
+        raise ValueError("expert_id and autopilot_only are mutually exclusive")
+
     params: list[Any] = [user_id]
     conditions = ['"userId" = $1', _EXCLUDE_DREAM_SESSIONS_SQL]
     if organization_id is not None:
@@ -724,9 +734,11 @@ async def get_user_chat_sessions(
     if title_contains:
         params.append(f"%{_escape_like(title_contains)}%")
         conditions.append(f'"title" ILIKE ${len(params)}')
-    if expert_id:
+    if expert_id is not None:
         params.append(expert_id)
         conditions.append(f'"expertId" = ${len(params)}')
+    elif autopilot_only:
+        conditions.append('"expertId" IS NULL')
     params.extend((limit, offset))
     ordering = (
         '"isPinned" DESC, "updatedAt" DESC' if pinned_first else '"updatedAt" DESC'
@@ -745,23 +757,31 @@ async def get_user_session_count(
     user_id: str,
     organization_id: str | None = None,
     expert_id: str | None = None,
+    autopilot_only: bool = False,
 ) -> int:
     """Get the total number of chat sessions for a user.
 
     Applies the same dream-session exclusion, org scoping, and expert
     filter as :func:`get_user_chat_sessions` so pagination totals always
     match the visible list.
     """
+    if expert_id == "":
+        raise ValueError("expert_id must be non-empty")
+    if expert_id is not None and autopilot_only:
+        raise ValueError("expert_id and autopilot_only are mutually exclusive")
+
     params: list[Any] = [user_id]
     conditions = ['"userId" = $1', _EXCLUDE_DREAM_SESSIONS_SQL]
     if organization_id is not None:
         params.append(organization_id)
         conditions.append(
             f'("organizationId" = ${len(params)} OR "organizationId" IS NULL)'
         )
-    if expert_id:
+    if expert_id is not None:
         params.append(expert_id)
         conditions.append(f'"expertId" = ${len(params)}')
+    elif autopilot_only:
+        conditions.append('"expertId" IS NULL')
     rows = await db.query_raw_with_schema(
         'SELECT COUNT(*)::int AS "count" FROM {schema_prefix}"ChatSession" WHERE '
         + " AND ".join(conditions),