Make channel threads visible to connection owners

backend/app/gateway/services.py

@@ -12,6 +12,7 @@ import json
 import logging
 import re
 from collections.abc import Mapping
+from types import SimpleNamespace
 from typing import Any
 
 from fastapi import HTTPException, Request
@@ -19,7 +20,7 @@ from langchain_core.messages import BaseMessage
 from langchain_core.messages.utils import convert_to_messages
 
 from app.gateway.deps import get_run_context, get_run_manager, get_stream_bridge
-from app.gateway.internal_auth import INTERNAL_SYSTEM_ROLE
+from app.gateway.internal_auth import INTERNAL_SYSTEM_ROLE, get_trusted_internal_owner_user_id
 from app.gateway.utils import sanitize_log_param
 from deerflow.config.app_config import get_app_config
 from deerflow.runtime import (
@@ -35,6 +36,7 @@ from deerflow.runtime import (
     run_agent,
 )
 from deerflow.runtime.runs.naming import resolve_root_run_name
+from deerflow.runtime.user_context import reset_current_user, set_current_user
 
 logger = logging.getLogger(__name__)
 
@@ -315,72 +317,85 @@ async def start_run(
                 detail=f"Model {model_name!r} is not in the configured model allowlist",
             )
 
+    owner_user_id = get_trusted_internal_owner_user_id(request)
+    owner_context_token = set_current_user(SimpleNamespace(id=owner_user_id)) if owner_user_id else None
     try:
-        record = await run_mgr.create_or_reject(
-            thread_id,
-            body.assistant_id,
-            on_disconnect=disconnect,
-            metadata=body.metadata or {},
-            kwargs={"input": body.input, "config": body.config},
-            multitask_strategy=body.multitask_strategy,
-            model_name=model_name,
-        )
-    except ConflictError as exc:
-        raise HTTPException(status_code=409, detail=str(exc)) from exc
-    except UnsupportedStrategyError as exc:
-        raise HTTPException(status_code=501, detail=str(exc)) from exc
-
-    # Upsert thread metadata so the thread appears in /threads/search,
-    # even for threads that were never explicitly created via POST /threads
-    # (e.g. stateless runs).
-    try:
-        existing = await run_ctx.thread_store.get(thread_id)
-        if existing is None:
-            await run_ctx.thread_store.create(
+        try:
+            record = await run_mgr.create_or_reject(
                 thread_id,
-                assistant_id=body.assistant_id,
-                metadata=body.metadata,
+                body.assistant_id,
+                on_disconnect=disconnect,
+                metadata=body.metadata or {},
+                kwargs={"input": body.input, "config": body.config},
+                multitask_strategy=body.multitask_strategy,
+                model_name=model_name,
+                user_id=owner_user_id,
             )
-        else:
-            await run_ctx.thread_store.update_status(thread_id, "running")
-    except Exception:
-        logger.warning("Failed to upsert thread_meta for %s (non-fatal)", sanitize_log_param(thread_id))
+        except ConflictError as exc:
+            raise HTTPException(status_code=409, detail=str(exc)) from exc
+        except UnsupportedStrategyError as exc:
+            raise HTTPException(status_code=501, detail=str(exc)) from exc
 
-    agent_factory = resolve_agent_factory(body.assistant_id)
-    graph_input = normalize_input(body.input)
-    config = build_run_config(thread_id, body.config, body.metadata, assistant_id=body.assistant_id)
+        # Upsert thread metadata so the thread appears in /threads/search,
+        # even for threads that were never explicitly created via POST /threads
+        # (e.g. stateless runs).
+        try:
+            existing = await run_ctx.thread_store.get(thread_id)
+            if existing is None and owner_user_id:
+                unscoped_existing = await run_ctx.thread_store.get(thread_id, user_id=None)
+                if unscoped_existing is not None:
+                    if unscoped_existing.get("user_id") != owner_user_id:
+                        await run_ctx.thread_store.update_owner(thread_id, owner_user_id, user_id=None)
+                    existing = await run_ctx.thread_store.get(thread_id)
+            if existing is None:
+                await run_ctx.thread_store.create(
+                    thread_id,
+                    assistant_id=body.assistant_id,
+                    metadata=body.metadata,
+                )
+            else:
+                await run_ctx.thread_store.update_status(thread_id, "running")
+        except Exception:
+            logger.warning("Failed to upsert thread_meta for %s (non-fatal)", sanitize_log_param(thread_id))
 
-    # Merge DeerFlow-specific context overrides into both ``configurable`` and ``context``.
-    # The ``context`` field is a custom extension for the langgraph-compat layer
-    # that carries agent configuration (model_name, thinking_enabled, etc.).
-    # Only agent-relevant keys are forwarded; unknown keys (e.g. thread_id) are ignored.
-    merge_run_context_overrides(config, getattr(body, "context", None))
-    inject_authenticated_user_context(config, request)
+        agent_factory = resolve_agent_factory(body.assistant_id)
+        graph_input = normalize_input(body.input)
+        config = build_run_config(thread_id, body.config, body.metadata, assistant_id=body.assistant_id)
 
-    stream_modes = normalize_stream_modes(body.stream_mode)
+        # Merge DeerFlow-specific context overrides into both ``configurable`` and ``context``.
+        # The ``context`` field is a custom extension for the langgraph-compat layer
+        # that carries agent configuration (model_name, thinking_enabled, etc.).
+        # Only agent-relevant keys are forwarded; unknown keys (e.g. thread_id) are ignored.
+        merge_run_context_overrides(config, getattr(body, "context", None))
+        inject_authenticated_user_context(config, request)
 
-    task = asyncio.create_task(
-        run_agent(
-            bridge,
-            run_mgr,
-            record,
-            ctx=run_ctx,
-            agent_factory=agent_factory,
-            graph_input=graph_input,
-            config=config,
-            stream_modes=stream_modes,
-            stream_subgraphs=body.stream_subgraphs,
-            interrupt_before=body.interrupt_before,
-            interrupt_after=body.interrupt_after,
+        stream_modes = normalize_stream_modes(body.stream_mode)
+
+        task = asyncio.create_task(
+            run_agent(
+                bridge,
+                run_mgr,
+                record,
+                ctx=run_ctx,
+                agent_factory=agent_factory,
+                graph_input=graph_input,
+                config=config,
+                stream_modes=stream_modes,
+                stream_subgraphs=body.stream_subgraphs,
+                interrupt_before=body.interrupt_before,
+                interrupt_after=body.interrupt_after,
+            )
         )
-    )
-    record.task = task
+        record.task = task
 
-    # Title sync is handled by worker.py's finally block which reads the
-    # title from the checkpoint and calls thread_store.update_display_name
-    # after the run completes.
+        # Title sync is handled by worker.py's finally block which reads the
+        # title from the checkpoint and calls thread_store.update_display_name
+        # after the run completes.
 
-    return record
+        return record
+    finally:
+        if owner_context_token is not None:
+            reset_current_user(owner_context_token)
 
 
 async def sse_consumer(