Merge branch 'master' of http://192.168.0.200/knowledge-base-bj/deerflow-server into dhk-1

offline-backend-20260512/backend/CLAUDE.md

@@ -237,6 +237,7 @@ FastAPI application on port 8001 with health check at `GET /health`. Set `GATEWA
 | **Tags** (`/api/tags`) | `GET /` - list tags (optional `tag_type` filter; open to all); `POST /` - create tag (admin); `PUT /{id}` / `DELETE /{id}` - update/delete tag (admin); `GET /assignments/{target_type}/{target_id}` - tags on a resource; `PUT /assignments/{target_type}/{target_id}` - replace a resource's tags (admin). Tags are typed `agent`/`skill`/`scheduled_task` and may only tag resources of their own type. |
 | **Uploads** (`/api/threads/{id}/uploads`) | `POST /` - upload files (auto-converts PDF/PPT/Excel/Word); `GET /list` - list; `DELETE /{filename}` - delete |
 | **Threads** (`/api/threads/{id}`) | `DELETE /` - remove DeerFlow-managed local thread data after LangGraph thread deletion; unexpected failures are logged server-side and return a generic 500 detail |
+| **Threads search** (`POST /threads/search`, LangGraph-compat) | Conversation listing. **System threads are excluded by default**: rows whose `metadata.system` is truthy (scheduler runs, roundtable workers) never crowd the user's list; a caller whose `metadata` filter explicitly references `thread_type`/`system` (e.g. studio notebooks) opts out of the exclusion. New `query` field = case-insensitive title (`display_name`) substring search, used by the chats page's server-side pagination. `POST /threads/count` takes the same request body (limit/offset ignored) and returns `{total}` under identical filtering/exclusion semantics — powers the chats page's 总数/总页数 display without fetching rows (`ThreadMetaStore.count`). `ThreadMetaStore.search` gained `exclude_system`/`query` params; JSON-blob filters batch-scan until the requested `limit`/`offset` window is full (so pagination over a heavily-filtered set never under-fills), with a `thread_id` tie-breaker on the `updated_at DESC` ordering for stable OFFSET paging. Tests: `tests/test_thread_meta_search.py` |
 | **Artifacts** (`/api/threads/{id}/artifacts`) | `GET /{path}` - serve artifacts; active content types (`text/html`, `application/xhtml+xml`, `image/svg+xml`) are always forced as download attachments to reduce XSS risk; `?download=true` still forces download for other file types |
 | **Suggestions** (`/api/threads/{id}/suggestions`) | `POST /` - generate follow-up questions; rich list/block model content is normalized before JSON parsing |
 | **Thread Runs** (`/api/threads/{id}/runs`) | `POST /` - create background run; `POST /stream` - create + SSE stream; `POST /wait` - create + block; `GET /` - list runs; `GET /{rid}` - run details; `POST /{rid}/cancel` - cancel; `GET /{rid}/join` - join SSE; `GET /{rid}/messages` - paginated messages `{data, has_more}`; `GET /{rid}/events` - full event stream; `GET /../messages` - thread messages with feedback; `GET /../token-usage` - aggregate tokens |

offline-backend-20260512/backend/app/gateway/routers/threads.py

@@ -692,6 +692,32 @@ async def search_threads(body: ThreadSearchRequest, request: Request) -> list[Th
     ]
 
 
+class ThreadCountResponse(BaseModel):
+    """Response model for the thread count endpoint."""
+
+    total: int = Field(description="Number of threads matching the filters")
+
+
+@router.post("/count", response_model=ThreadCountResponse)
+async def count_threads(body: ThreadSearchRequest, request: Request) -> ThreadCountResponse:
+    """Count threads matching the same filters as ``/threads/search``.
+
+    ``limit``/``offset`` in the body are ignored. Backs the chats page
+    pagination total ("共 N 条 / X 页") without fetching any rows.
+    """
+    from app.gateway.deps import get_thread_store
+
+    repo = get_thread_store(request)
+    explicit_system_filter = bool(body.metadata) and ("thread_type" in body.metadata or "system" in body.metadata)
+    total = await repo.count(
+        metadata=body.metadata or None,
+        status=body.status,
+        exclude_system=not explicit_system_filter,
+        query=body.query,
+    )
+    return ThreadCountResponse(total=total)
+
+
 @router.patch("/{thread_id}", response_model=ThreadResponse)
 @require_permission("threads", "write", owner_check=True, require_existing=True)
 async def patch_thread(thread_id: str, body: ThreadPatchRequest, request: Request) -> ThreadResponse:

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

@@ -59,6 +59,22 @@ class ThreadMetaStore(abc.ABC):
         """
         pass
 
+    @abc.abstractmethod
+    async def count(
+        self,
+        *,
+        metadata: dict | None = None,
+        status: str | None = None,
+        user_id: str | None | _AutoSentinel = AUTO,
+        exclude_system: bool = False,
+        query: str | None = None,
+    ) -> int:
+        """Count threads matching the same filters as :meth:`search`.
+
+        Backs pagination UIs that need a total alongside per-page results.
+        """
+        pass
+
     @abc.abstractmethod
     async def update_display_name(self, thread_id: str, display_name: str, *, user_id: str | None | _AutoSentinel = AUTO) -> None:
         pass

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

@@ -126,6 +126,50 @@ class MemoryThreadMetaStore(ThreadMetaStore):
             scan_offset += len(items)
         return matched[offset:needed]
 
+    async def count(
+        self,
+        *,
+        metadata: dict | None = None,
+        status: str | None = None,
+        user_id: str | None | _AutoSentinel = AUTO,
+        exclude_system: bool = False,
+        query: str | None = None,
+    ) -> int:
+        """Count threads matching the same filters as :meth:`search`."""
+        resolved_user_id = resolve_user_id(user_id, method_name="MemoryThreadMetaStore.count")
+        filter_dict: dict[str, Any] = {}
+        if metadata:
+            filter_dict.update(metadata)
+        if status:
+            filter_dict["status"] = status
+        if resolved_user_id is not None:
+            filter_dict["user_id"] = resolved_user_id
+
+        normalized_query = (query or "").strip().lower()
+        total = 0
+        batch_size = 500
+        scan_offset = 0
+        while True:
+            items = await self._store.asearch(
+                THREADS_NS,
+                filter=filter_dict or None,
+                limit=batch_size,
+                offset=scan_offset,
+            )
+            if not items:
+                break
+            for item in items:
+                record = self._item_to_dict(item)
+                if exclude_system and (record.get("metadata") or {}).get("system"):
+                    continue
+                if normalized_query and normalized_query not in (record.get("display_name") or "").lower():
+                    continue
+                total += 1
+            if len(items) < batch_size:
+                break
+            scan_offset += len(items)
+        return total
+
     async def check_access(self, thread_id: str, user_id: str, *, require_existing: bool = False) -> bool:
         item = await self._store.aget(THREADS_NS, thread_id)
         if item is None:

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

@@ -167,6 +167,58 @@ class ThreadMetaRepository(ThreadMetaStore):
                 scan_offset += len(rows)
         return matched[offset:needed]
 
+    async def count(
+        self,
+        *,
+        metadata: dict | None = None,
+        status: str | None = None,
+        user_id: str | None | _AutoSentinel = AUTO,
+        exclude_system: bool = False,
+        query: str | None = None,
+    ) -> int:
+        """Count threads matching the same filters as :meth:`search`."""
+        resolved_user_id = resolve_user_id(user_id, method_name="ThreadMetaRepository.count")
+        conditions = []
+        if resolved_user_id is not None:
+            conditions.append(ThreadMetaRow.user_id == resolved_user_id)
+        if status:
+            conditions.append(ThreadMetaRow.status == status)
+        if query and query.strip():
+            conditions.append(func.lower(ThreadMetaRow.display_name).contains(query.strip().lower()))
+
+        if not metadata and not exclude_system:
+            stmt = select(func.count()).select_from(ThreadMetaRow)
+            for condition in conditions:
+                stmt = stmt.where(condition)
+            async with self._sf() as session:
+                return int((await session.execute(stmt)).scalar_one())
+
+        # JSON filters can't run in portable SQL — scan just the metadata
+        # column in batches and count in Python (mirrors search()).
+        stmt = select(ThreadMetaRow.metadata_json).order_by(ThreadMetaRow.thread_id)
+        for condition in conditions:
+            stmt = stmt.where(condition)
+        total = 0
+        batch_size = 1000
+        scan_offset = 0
+        async with self._sf() as session:
+            while True:
+                result = await session.execute(stmt.limit(batch_size).offset(scan_offset))
+                rows = result.scalars().all()
+                if not rows:
+                    break
+                for metadata_json in rows:
+                    md = metadata_json or {}
+                    if metadata and not all(md.get(k) == v for k, v in metadata.items()):
+                        continue
+                    if exclude_system and md.get("system"):
+                        continue
+                    total += 1
+                if len(rows) < batch_size:
+                    break
+                scan_offset += len(rows)
+        return total
+
     async def _check_ownership(self, session: AsyncSession, thread_id: str, resolved_user_id: str | None) -> bool:
         """Return True if the row exists and is owned (or filter bypassed)."""
         if resolved_user_id is None: