feat: 任务深链对话(metadata.taskId)全局共享、不区分用户

任务工作区(rwfx/3qfx/xdfx)的对话改为按 taskId(+agent_id) 全局共享、跨用户可见可续聊，
不再按创建者隔离；普通个人对话仍按 owner 隔离不变。

- ThreadMetaStore.search/count：metadata 含 taskId 时跳过 owner 过滤(SQL+Memory)
- check_access / _check_ownership / update_metadata / delete / _get_owned_record：
  行 metadata 含 taskId 时视为共享，任何登录用户可读/写/续聊/改标题状态/删除
- 覆盖 require_permission(owner_check) + runs 续聊 owner 校验 + 列表
- 测试 tests/test_thread_meta_task_sharing.py(SQL 跨用户可见/隔离/check_access)

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

offline-backend-20260512/backend/packages/harness/deerflow/persistence/thread_meta/memory.py

@@ -34,6 +34,9 @@ class MemoryThreadMetaStore(ThreadMetaStore):
         if item is None:
             return None
         record = dict(item.value)
+        # 任务深链对话（metadata.taskId）全局共享：跳过 owner 校验。
+        if (record.get("metadata") or {}).get("taskId"):
+            return record
         if resolved is not None and record.get("user_id") != resolved:
             return None
         return record
@@ -78,6 +81,9 @@ class MemoryThreadMetaStore(ThreadMetaStore):
         query: str | None = None,
     ) -> list[dict]:
         resolved_user_id = resolve_user_id(user_id, method_name="MemoryThreadMetaStore.search")
+        # 任务深链对话（按 metadata.taskId 过滤）全局共享、不区分用户：跳过 owner 过滤。
+        if metadata and metadata.get("taskId"):
+            resolved_user_id = None
         filter_dict: dict[str, Any] = {}
         if metadata:
             filter_dict.update(metadata)
@@ -137,6 +143,9 @@ class MemoryThreadMetaStore(ThreadMetaStore):
     ) -> int:
         """Count threads matching the same filters as :meth:`search`."""
         resolved_user_id = resolve_user_id(user_id, method_name="MemoryThreadMetaStore.count")
+        # 任务深链对话（按 metadata.taskId 过滤）全局共享、不区分用户：跳过 owner 过滤。
+        if metadata and metadata.get("taskId"):
+            resolved_user_id = None
         filter_dict: dict[str, Any] = {}
         if metadata:
             filter_dict.update(metadata)
@@ -174,6 +183,9 @@ class MemoryThreadMetaStore(ThreadMetaStore):
         item = await self._store.aget(THREADS_NS, thread_id)
         if item is None:
             return not require_existing
+        # 任务深链对话（metadata.taskId）全局共享、不区分用户：任何登录用户都可读写/续聊。
+        if (item.value.get("metadata") or {}).get("taskId"):
+            return True
         record_user_id = item.value.get("user_id")
         if record_user_id is None:
             return True

offline-backend-20260512/backend/packages/harness/deerflow/persistence/thread_meta/sql.py

@@ -97,6 +97,9 @@ class ThreadMetaRepository(ThreadMetaStore):
             row = await session.get(ThreadMetaRow, thread_id)
             if row is None:
                 return not require_existing
+            # 任务深链对话（metadata.taskId）全局共享、不区分用户：任何登录用户都可读写/续聊。
+            if (row.metadata_json or {}).get("taskId"):
+                return True
             if row.user_id is None:
                 return True
             return row.user_id == user_id
@@ -118,6 +121,9 @@ class ThreadMetaRepository(ThreadMetaStore):
         context. Pass ``user_id=None`` to bypass (migration/CLI).
         """
         resolved_user_id = resolve_user_id(user_id, method_name="ThreadMetaRepository.search")
+        # 任务深链对话（按 metadata.taskId 过滤）全局共享、不区分用户：跳过 owner 过滤。
+        if metadata and metadata.get("taskId"):
+            resolved_user_id = None
         # thread_id tie-breaker keeps OFFSET pagination stable when many rows
         # share the same updated_at (bulk imports, scheduler bursts).
         stmt = select(ThreadMetaRow).order_by(ThreadMetaRow.updated_at.desc(), ThreadMetaRow.thread_id.desc())
@@ -178,6 +184,9 @@ class ThreadMetaRepository(ThreadMetaStore):
     ) -> int:
         """Count threads matching the same filters as :meth:`search`."""
         resolved_user_id = resolve_user_id(user_id, method_name="ThreadMetaRepository.count")
+        # 任务深链对话（按 metadata.taskId 过滤）全局共享、不区分用户：跳过 owner 过滤。
+        if metadata and metadata.get("taskId"):
+            resolved_user_id = None
         conditions = []
         if resolved_user_id is not None:
             conditions.append(ThreadMetaRow.user_id == resolved_user_id)
@@ -224,7 +233,12 @@ class ThreadMetaRepository(ThreadMetaStore):
         if resolved_user_id is None:
             return True  # explicit bypass
         row = await session.get(ThreadMetaRow, thread_id)
-        return row is not None and row.user_id == resolved_user_id
+        if row is None:
+            return False
+        # 任务深链对话（metadata.taskId）全局共享：任何用户都可更新标题/状态/updated_at（续聊置顶）。
+        if (row.metadata_json or {}).get("taskId"):
+            return True
+        return row.user_id == resolved_user_id
 
     async def update_display_name(
         self,
@@ -287,7 +301,9 @@ class ThreadMetaRepository(ThreadMetaStore):
             row = await session.get(ThreadMetaRow, thread_id)
             if row is None:
                 return
-            if resolved_user_id is not None and row.user_id != resolved_user_id:
+            # 任务深链对话（metadata.taskId）全局共享：跳过 owner 校验。
+            task_shared = bool((row.metadata_json or {}).get("taskId"))
+            if resolved_user_id is not None and not task_shared and row.user_id != resolved_user_id:
                 return
             merged = dict(row.metadata_json or {})
             merged.update(metadata)
@@ -306,7 +322,9 @@ class ThreadMetaRepository(ThreadMetaStore):
             row = await session.get(ThreadMetaRow, thread_id)
             if row is None:
                 return
-            if resolved_user_id is not None and row.user_id != resolved_user_id:
+            # 任务深链对话（metadata.taskId）全局共享：任何用户都可删除。
+            task_shared = bool((row.metadata_json or {}).get("taskId"))
+            if resolved_user_id is not None and not task_shared and row.user_id != resolved_user_id:
                 return
             await session.delete(row)
             await session.commit()